feat(site-call-audit): relay Retry/Discard emit operator-identity audit rows
Additive RequestedBy on Retry/DiscardSiteCallRequest, plumbed from the Central UI. On an Applied relay, SiteCallAuditActor emits one best-effort CachedResolve central direct-write row (Submitted/Discarded) with the operator as Actor. Gated on an injected ICentralAuditWriter (null in existing tests → no mirror read, unchanged); the site remains the source of truth for the state change.
This commit is contained in:
Joseph Doherty
@@ -213,4 +213,159 @@ public class SiteCallRelayTests : TestKit
|
||||
Assert.Equal(SiteCallRelayOutcome.SiteUnreachable, response.Outcome);
|
||||
Assert.False(response.SiteReachable);
|
||||
}
|
||||
|
||||
// ── Task 14: operator-identity audit on a successful relay ──
|
||||
|
||||
/// <summary>Captures the central direct-write audit events emitted by the relay.</summary>
|
||||
private sealed class RecordingCentralAuditWriter : ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter
|
||||
{
|
||||
public List<AuditRowProjection.AuditRowValues> Events { get; } = new();
|
||||
|
||||
public Task WriteAsync(ZB.MOM.WW.Audit.AuditEvent evt, CancellationToken ct = default)
|
||||
{
|
||||
lock (Events)
|
||||
{
|
||||
Events.Add(evt.AsRow());
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A repository whose <see cref="GetAsync"/> returns one fixed row (so the relay
|
||||
/// can read its Channel for the audit event) but still throws on every MUTATING
|
||||
/// call — central never writes the mirror row on the relay path.
|
||||
/// </summary>
|
||||
private sealed class RowReturningRepository : ISiteCallAuditRepository
|
||||
{
|
||||
private readonly SiteCall _row;
|
||||
public RowReturningRepository(SiteCall row) { _row = row; }
|
||||
|
||||
public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
|
||||
Task.FromResult<SiteCall?>(_row.TrackedOperationId == id ? _row : null);
|
||||
|
||||
public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) =>
|
||||
throw new InvalidOperationException("relay must not write the SiteCalls row");
|
||||
|
||||
public Task<IReadOnlyList<SiteCall>> QueryAsync(
|
||||
SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
|
||||
throw new InvalidOperationException("relay must not query the SiteCalls table");
|
||||
|
||||
public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
|
||||
throw new InvalidOperationException("relay must not purge");
|
||||
|
||||
public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
|
||||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||||
throw new InvalidOperationException("relay must not compute KPIs");
|
||||
|
||||
public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
|
||||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||||
throw new InvalidOperationException("relay must not compute per-site KPIs");
|
||||
|
||||
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
|
||||
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
|
||||
throw new InvalidOperationException("relay must not compute per-node KPIs");
|
||||
}
|
||||
|
||||
private static SiteCall MakeRow(Guid id, string channel = "ApiOutbound") => new()
|
||||
{
|
||||
TrackedOperationId = new TrackedOperationId(id),
|
||||
Channel = channel,
|
||||
Target = "ERP.GetOrder",
|
||||
SourceSite = "site-north",
|
||||
SourceNode = "node-a",
|
||||
Status = "Parked",
|
||||
RetryCount = 3,
|
||||
CreatedAtUtc = DateTime.UtcNow,
|
||||
UpdatedAtUtc = DateTime.UtcNow,
|
||||
IngestedAtUtc = DateTime.UtcNow,
|
||||
};
|
||||
|
||||
private (IActorRef Actor, RecordingCentralAuditWriter Writer) CreateActorWithAudit(IActorRef centralComm, SiteCall row)
|
||||
{
|
||||
var writer = new RecordingCentralAuditWriter();
|
||||
var options = new SiteCallAuditOptions { RelayTimeout = TimeSpan.FromMilliseconds(500) };
|
||||
var actor = Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
|
||||
new RowReturningRepository(row),
|
||||
NullLogger<SiteCallAuditActor>.Instance,
|
||||
options,
|
||||
(ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter)writer)));
|
||||
actor.Tell(new RegisterCentralCommunication(centralComm));
|
||||
return (actor, writer);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void RetrySiteCall_WhenApplied_EmitsCachedResolveSubmitted_WithOperator()
|
||||
{
|
||||
var central = CreateTestProbe();
|
||||
var id = Guid.NewGuid();
|
||||
var (actor, writer) = CreateActorWithAudit(central.Ref, MakeRow(id, channel: "ApiOutbound"));
|
||||
|
||||
actor.Tell(new RetrySiteCallRequest("corr-a", id, "site-north", RequestedBy: "jdoe"));
|
||||
|
||||
var envelope = central.ExpectMsg<SiteEnvelope>();
|
||||
var relay = (RetryParkedOperation)envelope.Message;
|
||||
central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true));
|
||||
|
||||
ExpectMsg<RetrySiteCallResponse>(r => r.Success);
|
||||
AwaitAssert(() =>
|
||||
{
|
||||
List<AuditRowProjection.AuditRowValues> events;
|
||||
lock (writer.Events) { events = writer.Events.ToList(); }
|
||||
Assert.Single(events);
|
||||
var evt = events[0];
|
||||
Assert.Equal(Commons.Types.Enums.AuditChannel.ApiOutbound, evt.Channel);
|
||||
Assert.Equal(Commons.Types.Enums.AuditKind.CachedResolve, evt.Kind);
|
||||
Assert.Equal(Commons.Types.Enums.AuditStatus.Submitted, evt.Status);
|
||||
Assert.Equal("jdoe", evt.Actor);
|
||||
Assert.Equal(id, evt.CorrelationId);
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void RetrySiteCall_WhenNotApplied_EmitsNoAuditRow()
|
||||
{
|
||||
var central = CreateTestProbe();
|
||||
var id = Guid.NewGuid();
|
||||
var (actor, writer) = CreateActorWithAudit(central.Ref, MakeRow(id));
|
||||
|
||||
actor.Tell(new RetrySiteCallRequest("corr-b", id, "site-north", RequestedBy: "jdoe"));
|
||||
|
||||
var envelope = central.ExpectMsg<SiteEnvelope>();
|
||||
var relay = (RetryParkedOperation)envelope.Message;
|
||||
// Nothing was parked at the site — no state change, so no operator audit row.
|
||||
central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: false));
|
||||
|
||||
ExpectMsg<RetrySiteCallResponse>(r => r.Outcome == SiteCallRelayOutcome.NotParked);
|
||||
// Give any erroneous emission a chance to land, then assert none did.
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
lock (writer.Events) { Assert.Empty(writer.Events); }
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DiscardSiteCall_WhenApplied_EmitsCachedResolveDiscarded_WithOperator()
|
||||
{
|
||||
var central = CreateTestProbe();
|
||||
var id = Guid.NewGuid();
|
||||
var (actor, writer) = CreateActorWithAudit(central.Ref, MakeRow(id, channel: "DbOutbound"));
|
||||
|
||||
actor.Tell(new DiscardSiteCallRequest("corr-c", id, "site-north", RequestedBy: "jdoe"));
|
||||
|
||||
var envelope = central.ExpectMsg<SiteEnvelope>();
|
||||
var relay = (DiscardParkedOperation)envelope.Message;
|
||||
central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true));
|
||||
|
||||
ExpectMsg<DiscardSiteCallResponse>(r => r.Success);
|
||||
AwaitAssert(() =>
|
||||
{
|
||||
List<AuditRowProjection.AuditRowValues> events;
|
||||
lock (writer.Events) { events = writer.Events.ToList(); }
|
||||
Assert.Single(events);
|
||||
var evt = events[0];
|
||||
Assert.Equal(Commons.Types.Enums.AuditChannel.DbOutbound, evt.Channel);
|
||||
Assert.Equal(Commons.Types.Enums.AuditStatus.Discarded, evt.Status);
|
||||
Assert.Equal("jdoe", evt.Actor);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user