diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundDatabaseHelper.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundDatabaseHelper.cs index 011053b6..806cfb33 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundDatabaseHelper.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundDatabaseHelper.cs @@ -1,6 +1,11 @@ using System.Data.Common; +using System.Diagnostics; using System.Globalization; +using System.Text; +using ZB.MOM.WW.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.InboundAPI; @@ -36,12 +41,35 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI; /// backstop derived from the method timeout — so a slow query is bounded by the method /// timeout instead of running unbounded. /// +/// +/// +/// Audited (Audit Log #23, outbound DB scope; Flow Debugger outbound-calls plan +/// 2026-09-06 D7). Every call emits exactly ONE DbOutbound/DbWrite row +/// through the central using the site's +/// AuditingDbCommand vocabulary: Extra = {"op":"write","rowsAffected":N} for +/// , {"op":"read","rowsReturned":N} for the two query +/// methods; RequestSummary = {"sql":…,"parameters":{"@name":"value"}}. The +/// 8 KB payload cap and per-target SQL-parameter redaction are applied at write time by +/// the writer's IAuditRedactor — the same stage that caps the site's rows — so the +/// helper builds the raw summary only. Emission is best-effort: a build, write or sink +/// fault never reaches the script; the original result (or exception) always does. When +/// a flow-trace sink is present the same instance goes to the +/// sink first, then to the writer, so a Live Flow Debugger run renders the row without +/// a round-trip to the audit store. A helper built without a writer and without a sink +/// (the pre-existing three-argument shape) emits nothing. +/// /// public sealed class InboundDatabaseHelper { private readonly IDatabaseGateway? _gateway; private readonly CancellationToken _ct; private readonly int _commandTimeoutSeconds; + private readonly ICentralAuditWriter? _auditWriter; + private readonly INodeIdentityProvider? _nodeIdentity; + private readonly Guid? _executionId; + private readonly Guid? _parentExecutionId; + private readonly string? _actor; + private readonly Action? _traceSink; /// /// Initializes the helper. @@ -49,13 +77,36 @@ public sealed class InboundDatabaseHelper /// The central database gateway, or null when no gateway is registered (the helper then throws on first use). /// The executing method's deadline token; forwarded to every async DB call so a slow query is bounded by the method timeout. /// The method timeout; used to derive a backstop. (the default) leaves the provider default. - public InboundDatabaseHelper(IDatabaseGateway? gateway, CancellationToken ct, TimeSpan commandTimeout = default) + /// Central-direct-write audit writer for the per-call DbOutbound/DbWrite row; null = no row is written. + /// Stamps SourceNode (central-a/central-b) on the row; null = NULL. + /// The executing inbound request's per-request execution id, stamped as the row's ExecutionId. + /// The spawning execution's id, if any (an inbound request is a root, so callers normally pass null). + /// The executing API method's name; the row's Actor becomes api:{methodName}. Null = NULL actor. + /// Flow Debugger trace sink; receives the same event instance before the writer does. Null = none. + // Trailing optional parameters: the original (gateway, ct, commandTimeout) shape stays + // source-compatible for every existing caller and test. + public InboundDatabaseHelper( + IDatabaseGateway? gateway, + CancellationToken ct, + TimeSpan commandTimeout = default, + ICentralAuditWriter? auditWriter = null, + INodeIdentityProvider? nodeIdentity = null, + Guid? executionId = null, + Guid? parentExecutionId = null, + string? methodName = null, + Action? traceSink = null) { _gateway = gateway; _ct = ct; _commandTimeoutSeconds = commandTimeout > TimeSpan.Zero ? (int)Math.Ceiling(commandTimeout.TotalSeconds) : 0; + _auditWriter = auditWriter; + _nodeIdentity = nodeIdentity; + _executionId = executionId; + _parentExecutionId = parentExecutionId; + _actor = string.IsNullOrEmpty(methodName) ? null : "api:" + methodName; + _traceSink = traceSink; } /// First column of the first row converted to (default if no rows). @@ -66,13 +117,16 @@ public sealed class InboundDatabaseHelper /// The converted scalar, or default when there are no rows / a NULL value. public async Task QuerySingleAsync(string connectionName, string sql, object? parameters = null) { - if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method"); - await using var conn = await _gateway.GetConnectionAsync(connectionName, _ct); - await using var cmd = CreateCommand(conn, sql, parameters); - var result = await cmd.ExecuteScalarAsync(_ct); - if (result is null or DBNull) return default; - if (result is T t) return t; - return (T)Convert.ChangeType(result, typeof(T), CultureInfo.InvariantCulture); + var (result, _) = await RunAudited(connectionName, sql, parameters, op: "read", async () => + { + await using var conn = await _gateway!.GetConnectionAsync(connectionName, _ct); + await using var cmd = CreateCommand(conn, sql, parameters); + var scalar = await cmd.ExecuteScalarAsync(_ct); + if (scalar is null or DBNull) return (default(T), 0); + if (scalar is T t) return (t, 1); + return ((T)Convert.ChangeType(scalar, typeof(T), CultureInfo.InvariantCulture), 1); + }); + return result; } /// All rows as column→value dictionaries (case-insensitive keys). @@ -83,21 +137,24 @@ public sealed class InboundDatabaseHelper public async Task>> QueryAsync( string connectionName, string sql, object? parameters = null) { - if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method"); - await using var conn = await _gateway.GetConnectionAsync(connectionName, _ct); - await using var cmd = CreateCommand(conn, sql, parameters); - await using var reader = await cmd.ExecuteReaderAsync(_ct); - var rows = new List>(); - while (await reader.ReadAsync(_ct)) + var (rows, _) = await RunAudited(connectionName, sql, parameters, op: "read", async () => { - var row = new Dictionary(StringComparer.OrdinalIgnoreCase); - for (var i = 0; i < reader.FieldCount; i++) + await using var conn = await _gateway!.GetConnectionAsync(connectionName, _ct); + await using var cmd = CreateCommand(conn, sql, parameters); + await using var reader = await cmd.ExecuteReaderAsync(_ct); + var list = new List>(); + while (await reader.ReadAsync(_ct)) { - var v = reader.GetValue(i); - row[reader.GetName(i)] = v is DBNull ? null : v; + var row = new Dictionary(StringComparer.OrdinalIgnoreCase); + for (var i = 0; i < reader.FieldCount; i++) + { + var v = reader.GetValue(i); + row[reader.GetName(i)] = v is DBNull ? null : v; + } + list.Add(row); } - rows.Add(row); - } + return ((IReadOnlyList>)list, list.Count); + }); return rows; } @@ -112,10 +169,178 @@ public sealed class InboundDatabaseHelper /// The number of rows affected. public async Task ExecuteAsync(string connectionName, string sql, object? parameters = null) { - if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method"); - await using var conn = await _gateway.GetConnectionAsync(connectionName, _ct); - await using var cmd = CreateCommand(conn, sql, parameters); - return await cmd.ExecuteNonQueryAsync(_ct); + var (affected, _) = await RunAudited(connectionName, sql, parameters, op: "write", async () => + { + await using var conn = await _gateway!.GetConnectionAsync(connectionName, _ct); + await using var cmd = CreateCommand(conn, sql, parameters); + var n = await cmd.ExecuteNonQueryAsync(_ct); + return (n, n); + }); + return affected; + } + + /// + /// Runs and emits the one DbOutbound/DbWrite row for + /// it — Delivered with the body's row count, or Failed with the thrown + /// exception's message, which is then rethrown untouched to the script. The gateway + /// null-check is inside the audited span deliberately so "Database is not available" + /// is itself recorded as a failed outbound DB call. + /// + private async Task<(T Result, int Rows)> RunAudited( + string connectionName, string sql, object? parameters, string op, + Func> body) + { + var startedUtc = DateTime.UtcNow; + var sw = Stopwatch.StartNew(); + (T Result, int Rows) outcome; + try + { + if (_gateway is null) throw new InvalidOperationException("Database is not available for this inbound method"); + outcome = await body(); + } + catch (Exception ex) + { + EmitAudit(connectionName, sql, parameters, op, startedUtc, sw.ElapsedMilliseconds, rows: 0, thrown: ex); + throw; + } + + EmitAudit(connectionName, sql, parameters, op, startedUtc, sw.ElapsedMilliseconds, outcome.Rows, thrown: null); + return outcome; + } + + /// + /// Best-effort emission (mirrors RouteTarget.EmitRouteCall and the site's + /// AuditingDbCommand.EmitAudit): build, sink, write — every layer swallows its + /// own faults so nothing reaches the script. + /// + private void EmitAudit( + string connectionName, string sql, object? parameters, string op, + DateTime startedUtc, long elapsedMs, int rows, Exception? thrown) + { + if (_auditWriter is null && _traceSink is null) return; + + AuditEvent evt; + try + { + // Vocabulary lock with the site's AuditingDbCommand: both reads and writes are + // DbWrite rows on the DbOutbound channel; Extra carries the discriminator. + var extra = op == "write" + ? $"{{\"op\":\"write\",\"rowsAffected\":{rows}}}" + : $"{{\"op\":\"read\",\"rowsReturned\":{rows}}}"; + + evt = ScadaBridgeAuditEventFactory.Create( + channel: AuditChannel.DbOutbound, + kind: AuditKind.DbWrite, + status: thrown is null ? AuditStatus.Delivered : AuditStatus.Failed, + occurredAtUtc: DateTime.SpecifyKind(startedUtc, DateTimeKind.Utc), + actor: _actor, + target: connectionName, + sourceNode: _nodeIdentity?.NodeName, + // A sync one-shot DB call has no operation lifecycle, so CorrelationId stays + // null; ExecutionId ties the row to the inbound request's other rows. + correlationId: null, + executionId: _executionId, + parentExecutionId: _parentExecutionId, + durationMs: (int)Math.Min(elapsedMs, int.MaxValue), + errorMessage: thrown?.Message, + errorDetail: thrown?.ToString(), + requestSummary: BuildRequestSummary(sql, parameters), + responseSummary: null, + payloadTruncated: false, + extra: extra); + } + catch (Exception) + { + return; + } + + // Sink FIRST (task contract): a Live Flow Debugger run reads the row from the + // sink, and the writer's redaction stage must not be the thing gating it. + if (_traceSink is not null) + { + try { _traceSink(evt); } + catch (Exception) { /* a misbehaving sink must not fail the script */ } + } + + if (_auditWriter is null) return; + try + { + var t = _auditWriter.WriteAsync(evt, CancellationToken.None); + if (!t.IsCompletedSuccessfully) + { + _ = t.ContinueWith(static _ => { }, TaskContinuationOptions.OnlyOnFaulted); + } + } + catch (Exception) + { + // Best-effort: an audit failure never fails the DB call. + } + } + + /// + /// Same JSON shape as the site's AuditingDbCommand.BuildRequestSummary: + /// {"sql":"…","parameters":{"@name":"value",…}}, values stringified, NULL as + /// JSON null. Built from the script's arguments rather than a + /// so a failure before the command exists (gateway/open) still carries the statement. + /// Capping and per-target parameter redaction happen at write time in the redactor. + /// + private static string? BuildRequestSummary(string sql, object? parameters) + { + if (string.IsNullOrEmpty(sql)) return null; + + var sb = new StringBuilder(sql.Length + 64); + sb.Append("{\"sql\":"); + AppendJsonString(sb, sql); + + var props = parameters?.GetType().GetProperties(); + if (props is { Length: > 0 }) + { + sb.Append(",\"parameters\":{"); + var first = true; + foreach (var prop in props) + { + if (!first) sb.Append(','); + first = false; + AppendJsonString(sb, "@" + prop.Name); + sb.Append(':'); + var value = prop.GetValue(parameters); + if (value is null || value is DBNull) + { + sb.Append("null"); + } + else + { + AppendJsonString(sb, Convert.ToString(value, CultureInfo.InvariantCulture) ?? string.Empty); + } + } + sb.Append('}'); + } + + sb.Append('}'); + return sb.ToString(); + } + + private static void AppendJsonString(StringBuilder sb, string value) + { + sb.Append('"'); + foreach (var ch in value) + { + switch (ch) + { + case '"': sb.Append("\\\""); break; + case '\\': sb.Append("\\\\"); break; + case '\b': sb.Append("\\b"); break; + case '\f': sb.Append("\\f"); break; + case '\n': sb.Append("\\n"); break; + case '\r': sb.Append("\\r"); break; + case '\t': sb.Append("\\t"); break; + default: + if (ch < 0x20) sb.Append("\\u").Append(((int)ch).ToString("x4")); + else sb.Append(ch); + break; + } + } + sb.Append('"'); } private DbCommand CreateCommand(DbConnection conn, string sql, object? parameters) diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundScriptExecutor.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundScriptExecutor.cs index 4359cef2..bd884279 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundScriptExecutor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundScriptExecutor.cs @@ -381,7 +381,23 @@ public class InboundScriptExecutor // Pass the method timeout so the helper derives a // CommandTimeout backstop and forwards cts.Token to every async DB call — // a slow query is then bounded by the method deadline. - var dbHelper = new InboundDatabaseHelper(gateway, cts.Token, timeout); + // + // Flow Debugger outbound calls (plan 2026-09-06, D7): the helper audits every + // Database.* call as a DbOutbound/DbWrite row. The inbound request's own + // execution id (the parentExecutionId argument — see its doc above) is the + // row's ExecutionId, exactly as the RouteCall row stamps it; an inbound request + // is a root, so the row's ParentExecutionId is null. The writer + node identity + // come from the same scope as the gateway (both null-tolerant: an unregistered + // writer means no row, never a failed script), and the trace sink is the one the + // caller attached to the RouteHelper so a Live run collects both row kinds. + var dbHelper = new InboundDatabaseHelper( + gateway, cts.Token, timeout, + auditWriter: scope?.ServiceProvider.GetService(), + nodeIdentity: scope?.ServiceProvider.GetService(), + executionId: parentExecutionId, + parentExecutionId: null, + methodName: method.Name, + traceSink: route.TraceSink); // Bind the route helper to the method deadline so a // routed Route.To(...).Call(...) inherits the method-level timeout diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs index 8a530e9a..1ca627a1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs @@ -164,6 +164,14 @@ public class RouteHelper new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, _auditWriter, _nodeIdentity, _trace, traceSink); + /// + /// Flow Debugger outbound calls (plan 2026-09-06, D5/D7): the trace sink this helper carries, + /// so can hand the SAME delegate to the script's + /// and a Live run's central Database.* rows land + /// in the same collector as its RouteCall rows. Null when no sink was attached. + /// + public Action? TraceSink => _traceSink; + /// /// Creates a route target for the specified instance. /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundDatabaseHelperAuditTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundDatabaseHelperAuditTests.cs new file mode 100644 index 00000000..0ae68830 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundDatabaseHelperAuditTests.cs @@ -0,0 +1,240 @@ +using System.Data.Common; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests; + +/// +/// Flow Debugger outbound calls (plan 2026-09-06, Task 4 / D7): every +/// call made by a central method script writes ONE +/// DbOutbound.DbWrite row through — best-effort, +/// never into the script — using the site's AuditingDbCommand vocabulary +/// ({"op":"write","rowsAffected":N} / {"op":"read","rowsReturned":N}), and +/// hands the same instance to the flow-trace sink first when one is present. +/// +public class InboundDatabaseHelperAuditTests +{ + // One shared-cache in-memory database per gateway instance. Such a database lives only + // while a connection to it is open, so the gateway ROOTS its keep-alive connection for + // its own lifetime — an unrooted local would be GC-finalized mid-test under load (the + // executor test's Roslyn compile is long enough) and the table would silently vanish. + private sealed class SqliteGateway : IDatabaseGateway + { + private readonly string _cs = $"DataSource=file:dbaudit-{Guid.NewGuid():N}?mode=memory&cache=shared"; + private readonly SqliteConnection _keepAlive; + + public SqliteGateway() + { + _keepAlive = new SqliteConnection(_cs); + _keepAlive.Open(); + using var cmd = _keepAlive.CreateCommand(); + cmd.CommandText = "CREATE TABLE Machine(Code TEXT, SAPID TEXT); INSERT INTO Machine VALUES('Z28061A','131453');"; + cmd.ExecuteNonQuery(); + } + + public async Task GetConnectionAsync(string name, CancellationToken ct = default) + { var c = new SqliteConnection(_cs); await c.OpenAsync(ct); return c; } + public Task CachedWriteAsync(string c, string s, + IReadOnlyDictionary? p = null, string? o = null, CancellationToken ct = default, + TrackedOperationId? t = null, Guid? e = null, string? src = null, Guid? pe = null) + => throw new NotImplementedException(); + } + + private sealed class ThrowingGateway : IDatabaseGateway + { + public Task GetConnectionAsync(string name, CancellationToken ct = default) + => throw new InvalidOperationException("no such connection: " + name); + public Task CachedWriteAsync(string c, string s, + IReadOnlyDictionary? p = null, string? o = null, CancellationToken ct = default, + TrackedOperationId? t = null, Guid? e = null, string? src = null, Guid? pe = null) + => throw new NotImplementedException(); + } + + private static SqliteGateway SeededGateway() => new(); + + private readonly List _written = new(); + private readonly ICentralAuditWriter _writer = Substitute.For(); + private readonly INodeIdentityProvider _node = Substitute.For(); + private readonly Guid _exec = Guid.NewGuid(); + private readonly Guid _parent = Guid.NewGuid(); + + public InboundDatabaseHelperAuditTests() + { + _writer.WriteAsync(Arg.Do(e => _written.Add(e)), Arg.Any()).Returns(Task.CompletedTask); + _node.NodeName.Returns("central-a"); + } + + private InboundDatabaseHelper Helper(IDatabaseGateway? gateway = null, Action? sink = null, + ICentralAuditWriter? writer = null) => + new(gateway ?? SeededGateway(), CancellationToken.None, TimeSpan.Zero, + auditWriter: writer ?? _writer, nodeIdentity: _node, + executionId: _exec, parentExecutionId: _parent, methodName: "movein", traceSink: sink); + + [Fact] + public async Task ExecuteAsync_WritesDbWriteDeliveredRow_WithRowsAffected() + { + var helper = Helper(); + + var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=@c WHERE SAPID=@s", new { c = "Z28061A", s = "131453" }); + + Assert.Equal(1, n); + var evt = Assert.Single(_written); + var row = evt.AsRow(); + Assert.Equal(AuditChannel.DbOutbound, row.Channel); + Assert.Equal(AuditKind.DbWrite, row.Kind); + Assert.Equal(AuditStatus.Delivered, row.Status); + Assert.Equal("MesDb", evt.Target); + Assert.Equal("api:movein", evt.Actor); + Assert.Equal("central-a", evt.SourceNode); + Assert.Equal(_exec, row.ExecutionId); + Assert.Equal(_parent, row.ParentExecutionId); + Assert.Null(row.CorrelationId); + Assert.NotNull(row.DurationMs); + Assert.Null(row.ErrorMessage); + Assert.Equal("{\"op\":\"write\",\"rowsAffected\":1}", row.Extra); + Assert.Contains("\"sql\":\"UPDATE Machine SET Code=@c WHERE SAPID=@s\"", row.RequestSummary); + Assert.Contains("\"@s\":\"131453\"", row.RequestSummary); + Assert.Contains("\"@c\":\"Z28061A\"", row.RequestSummary); + } + + [Fact] + public async Task QueryAsync_WritesReadRow_WithRowsReturned() + { + var helper = Helper(); + + var rows = await helper.QueryAsync("MesDb", "SELECT Code, SAPID FROM Machine WHERE SAPID=@s", new { s = "131453" }); + + Assert.Single(rows); + var row = Assert.Single(_written).AsRow(); + Assert.Equal(AuditKind.DbWrite, row.Kind); + Assert.Equal(AuditStatus.Delivered, row.Status); + Assert.Equal("{\"op\":\"read\",\"rowsReturned\":1}", row.Extra); + } + + [Fact] + public async Task QuerySingleAsync_WritesReadRow_ZeroRowsWhenNoMatch() + { + var helper = Helper(); + + var code = await helper.QuerySingleAsync("MesDb", "SELECT Code FROM Machine WHERE SAPID=@s", new { s = "nope" }); + + Assert.Null(code); + var row = Assert.Single(_written).AsRow(); + Assert.Equal("{\"op\":\"read\",\"rowsReturned\":0}", row.Extra); + } + + [Fact] + public async Task ThrowingGateway_WritesFailedRow_AndRethrowsToScript() + { + var helper = Helper(new ThrowingGateway()); + + var ex = await Assert.ThrowsAsync( + () => helper.ExecuteAsync("Missing", "DELETE FROM Machine", null)); + + Assert.Contains("no such connection", ex.Message); + var evt = Assert.Single(_written); + var row = evt.AsRow(); + Assert.Equal(AuditStatus.Failed, row.Status); + Assert.Equal("Missing", evt.Target); + Assert.Equal(ex.Message, row.ErrorMessage); + Assert.Equal("{\"op\":\"write\",\"rowsAffected\":0}", row.Extra); + Assert.Contains("\"sql\":\"DELETE FROM Machine\"", row.RequestSummary); + } + + [Fact] + public async Task AuditWriterFaulting_DoesNotFailTheCall() + { + var syncThrow = Substitute.For(); + syncThrow.WriteAsync(Arg.Any(), Arg.Any()) + .Returns(_ => throw new InvalidOperationException("audit down")); + var faultedTask = Substitute.For(); + faultedTask.WriteAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromException(new InvalidOperationException("audit down"))); + + var n1 = await Helper(writer: syncThrow).ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null); + var n2 = await Helper(writer: faultedTask).ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null); + + Assert.Equal(1, n1); + Assert.Equal(1, n2); + } + + [Fact] + public async Task TraceSink_ReceivesTheSameInstance_BeforeTheWriter() + { + var order = new List<(string Who, AuditEvent Evt)>(); + _writer.WriteAsync(Arg.Do(e => order.Add(("writer", e))), Arg.Any()).Returns(Task.CompletedTask); + var helper = Helper(sink: e => order.Add(("sink", e))); + + await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null); + + Assert.Equal(new[] { "sink", "writer" }, order.Select(o => o.Who)); + Assert.Same(order[0].Evt, order[1].Evt); + } + + [Fact] + public async Task ThrowingSink_DoesNotFailTheCall_AndWriterStillGetsTheRow() + { + var helper = Helper(sink: _ => throw new InvalidOperationException("bad sink")); + + var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null); + + Assert.Equal(1, n); + Assert.Single(_written); + } + + [Fact] + public async Task NoWriterNoSink_EmitsNothing_AndStillWorks() + { + var helper = new InboundDatabaseHelper(SeededGateway(), CancellationToken.None); + + var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null); + + Assert.Equal(1, n); + Assert.Empty(_written); + } + + [Fact] + public async Task Executor_WiresWriterNodeIdentityExecutionIdAndRouteSink_IntoTheHelper() + { + // End-to-end through InboundScriptExecutor: the DbOutbound row carries the inbound + // request's ExecutionId, "api:{method}" as actor, SourceNode from DI, and the same + // instance reaches the RouteHelper's trace sink (the Flow Debugger's Live path). + var services = new ServiceCollection(); + services.AddSingleton(SeededGateway()); + services.AddSingleton(_writer); + services.AddSingleton(_node); + using var provider = services.BuildServiceProvider(); + var executor = new InboundScriptExecutor(NullLogger.Instance, provider); + + var sunk = new List(); + var route = new RouteHelper(Substitute.For(), Substitute.For()) + .WithTraceSink(sunk.Add); + var method = new ApiMethod( + "movein", + "return await Database.ExecuteAsync(\"MesDb\", \"UPDATE Machine SET Code=@c WHERE SAPID=@s\", new { c = \"Z28061A\", s = \"131453\" });") + { Id = 1, TimeoutSeconds = 10, ReturnDefinition = null }; + var inboundExecution = Guid.NewGuid(); + + var result = await executor.ExecuteAsync( + method, new Dictionary(), route, TimeSpan.FromSeconds(10), + parentExecutionId: inboundExecution, trace: true); + + Assert.True(result.Success, result.ErrorMessage); + var evt = Assert.Single(_written); + var row = evt.AsRow(); + Assert.Equal(AuditChannel.DbOutbound, row.Channel); + Assert.Equal(inboundExecution, row.ExecutionId); + Assert.Null(row.ParentExecutionId); + Assert.Equal("api:movein", evt.Actor); + Assert.Equal("central-a", evt.SourceNode); + Assert.Same(evt, Assert.Single(sunk)); + } +}