ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditFailurePathTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication;

namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests;

/// <summary>
/// Gitea #73: in-memory coverage of <see cref="SiteCallAuditActor"/>'s error, retry and
/// state-transition branches that the existing suites leave untested — or test only behind
/// the MSSQL fixture, which skips without a database. Covers: a failed page write holding the
/// reconciliation cursor so the next tick re-pulls the window; the pinned latch clearing (and
/// publishing only on transitions); an enumerator fault not wedging the single-flight guard;
/// the per-tick page ceiling; the purge single-flight guard; the read-side and ingest fault
/// mapping; and the Discard / audit-sink branches of the central→site relay.
/// </summary>
public class SiteCallAuditFailurePathTests : TestKit
{
    private static SiteCall Row(string site, DateTime updatedAtUtc, string channel = "ApiOutbound") => new()
    {
        TrackedOperationId = TrackedOperationId.New(),
        Channel = channel,
        Target = "ERP.GetOrder",
        SourceSite = site,
        SourceNode = "node-a",
        Status = "Parked",
        RetryCount = 1,
        CreatedAtUtc = updatedAtUtc,
        UpdatedAtUtc = updatedAtUtc,
        IngestedAtUtc = updatedAtUtc,
    };

    private static SiteCallAuditOptions FastTicks(int batchSize = 500) => new()
    {
        ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100),
        PurgeIntervalOverride = TimeSpan.FromMinutes(10),
        ReconciliationBatchSize = batchSize,
        RelayTimeout = TimeSpan.FromMilliseconds(500),
    };

    /// <summary>Repository whose every member can be scripted; unscripted members are inert.</summary>
    private sealed class ScriptableRepo : ISiteCallAuditRepository
    {
        private int _upsertManyCalls;

        /// <summary>Gets the rows written by successful batched upserts.</summary>
        public List<SiteCall> Written { get; } = new();

        /// <summary>Number of leading UpsertManyAsync calls that throw.</summary>
        public int FailFirstUpsertMany { get; set; }

        /// <summary>Optional override for single-row upserts.</summary>
        public Func<SiteCall, Task>? OnUpsert { get; set; }

        /// <summary>Optional override for GetAsync.</summary>
        public Func<TrackedOperationId, Task<SiteCall?>>? OnGet { get; set; }

        /// <summary>Optional override for QueryAsync.</summary>
        public Func<SiteCallQueryFilter, SiteCallPaging, Task<IReadOnlyList<SiteCall>>>? OnQuery { get; set; }

        /// <summary>Optional override for PurgeTerminalAsync.</summary>
        public Func<DateTime, Task<int>>? OnPurge { get; set; }

        /// <summary>When set, every KPI computation throws it.</summary>
        public Exception? KpiFault { get; set; }

        /// <inheritdoc />
        public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) =>
            OnUpsert?.Invoke(siteCall) ?? Task.CompletedTask;

        /// <inheritdoc />
        public Task UpsertManyAsync(IReadOnlyList<SiteCall> siteCalls, CancellationToken ct = default)
        {
            if (Interlocked.Increment(ref _upsertManyCalls) <= FailFirstUpsertMany)
                throw new InvalidOperationException("simulated page write failure");
            lock (Written) Written.AddRange(siteCalls);
            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
            OnGet?.Invoke(id) ?? Task.FromResult<SiteCall?>(null);

        /// <inheritdoc />
        public Task<IReadOnlyList<SiteCall>> QueryAsync(
            SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
            OnQuery?.Invoke(filter, paging) ?? Task.FromResult<IReadOnlyList<SiteCall>>(Array.Empty<SiteCall>());

        /// <inheritdoc />
        public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
            OnPurge?.Invoke(olderThanUtc) ?? Task.FromResult(0);

        /// <inheritdoc />
        public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
            DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
            KpiFault is null
                ? Task.FromResult(new SiteCallKpiSnapshot(1, 2, 3, 4, null, 5))
                : Task.FromException<SiteCallKpiSnapshot>(KpiFault);

        /// <inheritdoc />
        public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
            DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
            KpiFault is null
                ? Task.FromResult<IReadOnlyList<SiteCallSiteKpiSnapshot>>(Array.Empty<SiteCallSiteKpiSnapshot>())
                : Task.FromException<IReadOnlyList<SiteCallSiteKpiSnapshot>>(KpiFault);

        /// <inheritdoc />
        public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
            DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
            KpiFault is null
                ? Task.FromResult<IReadOnlyList<SiteCallNodeKpiSnapshot>>(Array.Empty<SiteCallNodeKpiSnapshot>())
                : Task.FromException<IReadOnlyList<SiteCallNodeKpiSnapshot>>(KpiFault);
    }

    /// <summary>Enumerator that throws on its first N calls, counting every call.</summary>
    private sealed class FlakyEnumerator(int failFirst, params SiteEntry[] sites) : ISiteEnumerator
    {
        private int _calls;

        /// <summary>Gets the number of enumerations performed so far.</summary>
        public int Calls => Volatile.Read(ref _calls);

        /// <inheritdoc />
        public Task<IReadOnlyList<SiteEntry>> EnumerateAsync(CancellationToken ct = default) =>
            Interlocked.Increment(ref _calls) <= failFirst
                ? throw new InvalidOperationException("simulated config DB outage")
                : Task.FromResult<IReadOnlyList<SiteEntry>>(sites);
    }

    /// <summary>
    /// Keyset-honouring site over a fixed row set that can be switched into "legacy" mode, where
    /// it ignores the cursor and always reports a saturated page.
    /// </summary>
    private sealed class SwitchablePullClient(IEnumerable<SiteCall> rows) : IPullSiteCallsClient
    {
        private readonly List<SiteCall> _rows = rows
            .OrderBy(r => r.UpdatedAtUtc)
            .ThenBy(r => r.TrackedOperationId.ToString(), StringComparer.Ordinal)
            .ToList();

        /// <summary>Gets every pull recorded, with the enumerator tick it ran under.</summary>
        public List<(DateTime Since, string? AfterId)> Calls { get; } = new();

        /// <summary>Gets or sets whether the site ignores <c>after_id</c> (pre-keyset).</summary>
        public volatile bool Legacy;

        /// <inheritdoc />
        public Task<PullSiteCallsResponse> PullAsync(
            string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
        {
            lock (Calls) Calls.Add((sinceUtc, afterId));
            if (Legacy)
                return Task.FromResult(new PullSiteCallsResponse(_rows.Take(batchSize).ToList(), MoreAvailable: true));

            var greater = _rows.Where(r =>
                    r.UpdatedAtUtc > sinceUtc ||
                    (r.UpdatedAtUtc == sinceUtc && string.CompareOrdinal(r.TrackedOperationId.ToString(), afterId) > 0))
                .ToList();
            var page = greater.Take(batchSize).ToList();
            return Task.FromResult(new PullSiteCallsResponse(page, MoreAvailable: greater.Count > page.Count));
        }
    }

    /// <summary>A site with an endless backlog: every pull returns one newer row and MoreAvailable.</summary>
    private sealed class EndlessBacklogPullClient(FlakyEnumerator ticks) : IPullSiteCallsClient
    {
        private static readonly DateTime Start = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
        private int _n;

        /// <summary>Gets every pull with the enumeration (tick) number it belongs to.</summary>
        public List<(int Tick, DateTime Since, DateTime Returned)> Calls { get; } = new();

        /// <inheritdoc />
        public Task<PullSiteCallsResponse> PullAsync(
            string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
        {
            var row = Row(siteId, Start.AddSeconds(Interlocked.Increment(ref _n)));
            lock (Calls) Calls.Add((ticks.Calls, sinceUtc, row.UpdatedAtUtc));
            return Task.FromResult(new PullSiteCallsResponse(new[] { row }, MoreAvailable: true));
        }
    }

    /// <summary>No-op pull client that only arms the reconciliation scheduler.</summary>
    private sealed class NoOpPullClient : IPullSiteCallsClient
    {
        /// <inheritdoc />
        public Task<PullSiteCallsResponse> PullAsync(
            string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) =>
            Task.FromResult(new PullSiteCallsResponse(Array.Empty<SiteCall>(), MoreAvailable: false));
    }

    /// <summary>Central audit sink that throws on every write.</summary>
    private sealed class ThrowingAuditWriter : ICentralAuditWriter
    {
        private int _calls;

        /// <summary>Gets the number of write attempts.</summary>
        public int Calls => Volatile.Read(ref _calls);

        /// <inheritdoc />
        public Task WriteAsync(ZB.MOM.WW.Audit.AuditEvent evt, CancellationToken ct = default)
        {
            Interlocked.Increment(ref _calls);
            throw new InvalidOperationException("audit store down");
        }
    }

    /// <summary>Central audit sink that records every row.</summary>
    private sealed class RecordingAuditWriter : ICentralAuditWriter
    {
        /// <summary>Gets the recorded rows.</summary>
        public List<AuditRowProjection.AuditRowValues> Events { get; } = new();

        /// <inheritdoc />
        public Task WriteAsync(ZB.MOM.WW.Audit.AuditEvent evt, CancellationToken ct = default)
        {
            lock (Events) Events.Add(evt.AsRow());
            return Task.CompletedTask;
        }
    }

    private IActorRef CreateActor(
        ISiteCallAuditRepository repo,
        SiteCallAuditOptions options,
        ISiteEnumerator? sites = null,
        IPullSiteCallsClient? client = null,
        ICentralAuditWriter? auditWriter = null,
        bool serialize = true)
    {
        var provider = TestSupport.StubRepositoryProvider.Build(repo, sites, client, auditWriter, serialize);
        return Sys.ActorOf(Props.Create(() => new SiteCallAuditActor(
            provider, options, NullLogger<SiteCallAuditActor>.Instance)));
    }

    private void Stop(IActorRef actor)
    {
        Watch(actor);
        Sys.Stop(actor);
        ExpectTerminated(actor, TimeSpan.FromSeconds(3));
    }

    // ── Reconciliation: retry, latch and guard transitions ──

    /// <summary>
    /// A page whose batched upsert fails must NOT advance the site's cursor: the next tick
    /// re-pulls the same window from the old cursor, and once the write succeeds the cursor
    /// moves past it. Advancing first would silently drop the failed page from the mirror.
    /// </summary>
    [Fact]
    public void Reconciliation_PageWriteFails_CursorHeld_NextTickRePullsSameWindow()
    {
        var t1 = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
        var rows = new[] { Row("siteA", t1), Row("siteA", t1.AddMinutes(1)) };
        var client = new SwitchablePullClient(rows);
        var repo = new ScriptableRepo { FailFirstUpsertMany = 1 };

        var actor = CreateActor(repo, FastTicks(),
            new FlakyEnumerator(0, new SiteEntry("siteA", "http://a")), client);
        try
        {
            AwaitAssert(() =>
            {
                lock (client.Calls) Assert.True(client.Calls.Count >= 3, $"pulls: {client.Calls.Count}");
            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50));

            List<(DateTime Since, string? AfterId)> calls;
            lock (client.Calls) calls = client.Calls.ToList();
            Assert.Equal(DateTime.MinValue, calls[0].Since); // failed write
            Assert.Equal(DateTime.MinValue, calls[1].Since); // same window re-pulled
            Assert.Equal(t1.AddMinutes(1), calls[2].Since);  // advanced only after the write landed
            lock (repo.Written) Assert.Equal(2, repo.Written.Count);
        }
        finally
        {
            Stop(actor);
        }
    }

    /// <summary>
    /// The pinned latch is transition-only: a site that stays pinned across many ticks publishes
    /// Pinned=true once, and when it starts honouring the keyset again it publishes Pinned=false
    /// exactly once.
    /// </summary>
    [Fact]
    public void Reconciliation_PinnedSiteResumesProgress_PublishesEachTransitionOnce()
    {
        var ts = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
        var rows = new[] { Row("siteP", ts), Row("siteP", ts), Row("siteP", ts) };
        var client = new SwitchablePullClient(rows) { Legacy = true };
        var probe = CreateTestProbe();
        Sys.EventStream.Subscribe(probe.Ref, typeof(SiteCallReconciliationPinnedChanged));

        var actor = CreateActor(new ScriptableRepo(), FastTicks(batchSize: 2),
            new FlakyEnumerator(0, new SiteEntry("siteP", "http://p")), client);
        try
        {
            Assert.True(probe.ExpectMsg<SiteCallReconciliationPinnedChanged>(TimeSpan.FromSeconds(5)).Pinned);

            // Several more pinned ticks: no repeat publication.
            var pullsAtPin = client.Calls.Count;
            AwaitCondition(() => client.Calls.Count >= pullsAtPin + 3, TimeSpan.FromSeconds(5));
            probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));

            client.Legacy = false; // the site is upgraded and honours after_id
            var unpinned = probe.ExpectMsg<SiteCallReconciliationPinnedChanged>(TimeSpan.FromSeconds(5));
            Assert.Equal("siteP", unpinned.SiteId);
            Assert.False(unpinned.Pinned);
            probe.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
        }
        finally
        {
            Sys.EventStream.Unsubscribe(probe.Ref);
            Stop(actor);
        }
    }

    /// <summary>
    /// A site-enumeration fault skips that tick but lowers the single-flight guard, so the next
    /// tick reconciles normally — one config-DB blip must not wedge the self-heal for good.
    /// </summary>
    [Fact]
    public void Reconciliation_EnumeratorThrows_TickSkipped_NextTickReconciles()
    {
        var row = Row("siteA", new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc));
        var sites = new FlakyEnumerator(2, new SiteEntry("siteA", "http://a"));
        var client = new SwitchablePullClient(new[] { row });
        var repo = new ScriptableRepo();

        var actor = CreateActor(repo, FastTicks(), sites, client);
        try
        {
            AwaitAssert(() =>
            {
                lock (repo.Written) Assert.Contains(repo.Written, r => r.TrackedOperationId == row.TrackedOperationId);
            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50));

            Assert.True(sites.Calls >= 3);
            // No pull happened during the two failed enumerations.
            lock (client.Calls) Assert.Equal(DateTime.MinValue, client.Calls[0].Since);
        }
        finally
        {
            Stop(actor);
        }
    }

    /// <summary>
    /// A site that always reports more rows is drained at most 50 pages per tick (bounded, not
    /// pinned — the cursor advances each page), and the next tick resumes from the advanced
    /// cursor rather than starting over.
    /// </summary>
    [Fact]
    public void Reconciliation_EndlessBacklog_BoundedPerTick_NextTickResumesFromAdvancedCursor()
    {
        var sites = new FlakyEnumerator(0, new SiteEntry("siteB", "http://b"));
        var client = new EndlessBacklogPullClient(sites);
        var probe = CreateTestProbe();
        Sys.EventStream.Subscribe(probe.Ref, typeof(SiteCallReconciliationPinnedChanged));

        var actor = CreateActor(new ScriptableRepo(), FastTicks(batchSize: 1), sites, client);
        try
        {
            AwaitAssert(() =>
            {
                lock (client.Calls) Assert.Contains(client.Calls, c => c.Tick >= 2);
            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50));

            List<(int Tick, DateTime Since, DateTime Returned)> calls;
            lock (client.Calls) calls = client.Calls.ToList();
            var firstTick = calls.Where(c => c.Tick == 1).ToList();
            var secondTickFirstPull = calls.First(c => c.Tick == 2);

            Assert.Equal(50, firstTick.Count);
            Assert.Equal(firstTick[^1].Returned, secondTickFirstPull.Since);
            probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); // progressing, so never pinned
        }
        finally
        {
            Sys.EventStream.Unsubscribe(probe.Ref);
            Stop(actor);
        }
    }

    // ── Purge single-flight ──

    /// <summary>
    /// A purge that outlives its interval is not stacked: further ticks are dropped while it is
    /// in flight, and the next purge runs once it completes.
    /// </summary>
    [Fact]
    public void Purge_InFlight_FurtherTicksDropped_UntilItCompletes()
    {
        var release = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
        var calls = 0;
        var repo = new ScriptableRepo
        {
            OnPurge = _ => Interlocked.Increment(ref calls) == 1 ? release.Task : Task.FromResult(0),
        };
        var options = FastTicks();
        options.ReconciliationIntervalOverride = TimeSpan.FromMinutes(10);
        options.PurgeIntervalOverride = TimeSpan.FromMilliseconds(50);

        var actor = CreateActor(repo, options, new FlakyEnumerator(0), new NoOpPullClient(), serialize: false);
        try
        {
            AwaitCondition(() => Volatile.Read(ref calls) == 1, TimeSpan.FromSeconds(3));
            // ~10 purge intervals elapse while the first purge is blocked.
            ExpectNoMsg(TimeSpan.FromMilliseconds(500));
            Assert.Equal(1, Volatile.Read(ref calls));

            release.SetResult(3);
            AwaitCondition(() => Volatile.Read(ref calls) >= 2, TimeSpan.FromSeconds(3));
        }
        finally
        {
            release.TrySetResult(0);
            Stop(actor);
        }
    }

    // ── Ingest and read-side fault mapping ──

    /// <summary>
    /// An ingest write fault replies Accepted=false (the telemetry sender keeps the row for a
    /// later push) and the singleton stays alive to accept the next upsert.
    /// </summary>
    [Fact]
    public void Upsert_RepositoryThrows_RepliesNotAccepted_ThenAcceptsNext()
    {
        var calls = 0;
        var repo = new ScriptableRepo
        {
            OnUpsert = _ => Interlocked.Increment(ref calls) == 1
                ? throw new TimeoutException("SQL timeout")
                : Task.CompletedTask,
        };
        var actor = CreateActor(repo, FastTicks());
        var first = Row("siteA", DateTime.UtcNow);
        var second = Row("siteA", DateTime.UtcNow);

        actor.Tell(new UpsertSiteCallCommand(first));
        var failed = ExpectMsg<UpsertSiteCallReply>();
        Assert.Equal(first.TrackedOperationId, failed.TrackedOperationId);
        Assert.False(failed.Accepted);

        actor.Tell(new UpsertSiteCallCommand(second));
        var ok = ExpectMsg<UpsertSiteCallReply>();
        Assert.Equal(second.TrackedOperationId, ok.TrackedOperationId);
        Assert.True(ok.Accepted);
    }

    /// <summary>
    /// Every read handler maps a repository fault onto Success=false with the root-cause message
    /// and an empty payload — never a dead letter or a crashed singleton — and keeps serving.
    /// </summary>
    [Fact]
    public void ReadHandlers_RepositoryThrows_ReplyFailureWithRootCause_AndStayAlive()
    {
        var fault = new InvalidOperationException("wrapper", new TimeoutException("SQL timeout"));
        var repo = new ScriptableRepo
        {
            KpiFault = fault,
            OnGet = _ => Task.FromException<SiteCall?>(fault),
            OnQuery = (_, _) => Task.FromException<IReadOnlyList<SiteCall>>(fault),
        };
        var actor = CreateActor(repo, FastTicks());

        actor.Tell(new SiteCallDetailRequest("d", Guid.NewGuid()));
        var detail = ExpectMsg<SiteCallDetailResponse>();
        Assert.False(detail.Success);
        Assert.Equal("SQL timeout", detail.ErrorMessage);
        Assert.Null(detail.Detail);

        actor.Tell(new SiteCallKpiRequest("k"));
        var kpi = ExpectMsg<SiteCallKpiResponse>();
        Assert.False(kpi.Success);
        Assert.Equal("SQL timeout", kpi.ErrorMessage);
        Assert.Equal(0, kpi.BufferedCount + kpi.ParkedCount + kpi.StuckCount);

        actor.Tell(new PerSiteSiteCallKpiRequest("s"));
        var perSite = ExpectMsg<PerSiteSiteCallKpiResponse>();
        Assert.False(perSite.Success);
        Assert.Equal("SQL timeout", perSite.ErrorMessage);
        Assert.Empty(perSite.Sites);

        actor.Tell(new PerNodeSiteCallKpiRequest("n"));
        var perNode = ExpectMsg<PerNodeSiteCallKpiResponse>();
        Assert.False(perNode.Success);
        Assert.Equal("SQL timeout", perNode.ErrorMessage);
        Assert.Empty(perNode.Nodes);

        actor.Tell(new SiteCallQueryRequest("q", null, null, null, null, false, null, null, null, null, 50));
        var query = ExpectMsg<SiteCallQueryResponse>();
        Assert.False(query.Success);
        Assert.Equal("SQL timeout", query.ErrorMessage);
        Assert.Empty(query.SiteCalls);
        Assert.Null(query.NextAfterId);

        // Still alive: a healthy KPI read after the faults succeeds.
        repo.KpiFault = null;
        actor.Tell(new SiteCallKpiRequest("k2"));
        Assert.True(ExpectMsg<SiteCallKpiResponse>().Success);
    }

    /// <summary>
    /// The query page size is clamped to [1, 200] before it reaches SQL (an unbounded page from a
    /// crafted request would materialise the whole table), and an empty page carries no cursor.
    /// </summary>
    [Theory]
    [InlineData(0, 1)]
    [InlineData(-5, 1)]
    [InlineData(10_000, 200)]
    [InlineData(37, 37)]
    public void Query_PageSizeClamped_EmptyPageHasNoCursor(int requested, int expected)
    {
        var seen = new List<int>();
        var repo = new ScriptableRepo
        {
            OnQuery = (_, paging) =>
            {
                lock (seen) seen.Add(paging.PageSize);
                return Task.FromResult<IReadOnlyList<SiteCall>>(Array.Empty<SiteCall>());
            },
        };
        var actor = CreateActor(repo, FastTicks());

        actor.Tell(new SiteCallQueryRequest("q", null, null, null, null, false, null, null, null, null, requested));
        var response = ExpectMsg<SiteCallQueryResponse>();

        Assert.True(response.Success);
        lock (seen) Assert.Equal(new[] { expected }, seen);
        Assert.Null(response.NextAfterCreatedAtUtc);
        Assert.Null(response.NextAfterId);
    }

    // ── Relay: Discard branches and the best-effort audit sink ──

    /// <summary>
    /// Discard maps the site's ack the same way Retry does: a clean no-op is NotParked, an ack
    /// with an error is OperationFailed carrying the site's message verbatim — the site was
    /// reached either way — and neither emits an operator audit row.
    /// </summary>
    /// <param name="siteError">The error the site's ack carries, or null for a clean no-op.</param>
    /// <param name="expected">The outcome the relay must report.</param>
    [Theory]
    [InlineData(null, SiteCallRelayOutcome.NotParked)]
    [InlineData("S&F buffer locked", SiteCallRelayOutcome.OperationFailed)]
    public void DiscardSiteCall_SiteDidNotApply_MapsOutcome_NoAuditRow(string? siteError, SiteCallRelayOutcome expected)
    {
        var id = Guid.NewGuid();
        var writer = new RecordingAuditWriter();
        var repo = new ScriptableRepo { OnGet = _ => Task.FromResult<SiteCall?>(Row("site-north", DateTime.UtcNow)) };
        var central = CreateTestProbe();
        var actor = CreateActor(repo, FastTicks(), auditWriter: writer);
        actor.Tell(new RegisterCentralCommunication(central.Ref));

        actor.Tell(new DiscardSiteCallRequest("corr", id, "site-north", RequestedBy: "jdoe"));
        var relay = (DiscardParkedOperation)central.ExpectMsg<SiteEnvelope>().Message;
        central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: false, siteError));

        var response = ExpectMsg<DiscardSiteCallResponse>();
        Assert.Equal(expected, response.Outcome);
        Assert.False(response.Success);
        Assert.True(response.SiteReachable);
        if (siteError is not null)
            Assert.Equal(siteError, response.ErrorMessage);
        else
            Assert.Contains("no longer parked", response.ErrorMessage);

        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        lock (writer.Events) Assert.Empty(writer.Events);
    }

    /// <summary>A Discard before any central→site transport is registered is reported unreachable.</summary>
    [Fact]
    public void DiscardSiteCall_BeforeCentralCommunicationRegistered_ReportsSiteUnreachable()
    {
        var actor = CreateActor(new ScriptableRepo(), FastTicks());

        actor.Tell(new DiscardSiteCallRequest("corr", Guid.NewGuid(), "site-north"));

        var response = ExpectMsg<DiscardSiteCallResponse>();
        Assert.Equal(SiteCallRelayOutcome.SiteUnreachable, response.Outcome);
        Assert.False(response.SiteReachable);
        Assert.False(response.Success);
    }

    /// <summary>
    /// The operator audit row is best-effort: a throwing audit sink must not turn an Applied
    /// relay into a failure — the site already acted, and reporting otherwise would invite a
    /// duplicate retry.
    /// </summary>
    [Fact]
    public void RetrySiteCall_AuditSinkThrows_RelayStillReportsApplied()
    {
        var id = Guid.NewGuid();
        var writer = new ThrowingAuditWriter();
        var repo = new ScriptableRepo { OnGet = _ => Task.FromResult<SiteCall?>(Row("site-north", DateTime.UtcNow)) };
        var central = CreateTestProbe();
        var actor = CreateActor(repo, FastTicks(), auditWriter: writer);
        actor.Tell(new RegisterCentralCommunication(central.Ref));

        actor.Tell(new RetrySiteCallRequest("corr", id, "site-north", RequestedBy: "jdoe"));
        var relay = (RetryParkedOperation)central.ExpectMsg<SiteEnvelope>().Message;
        central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true));

        var response = ExpectMsg<RetrySiteCallResponse>();
        Assert.Equal(SiteCallRelayOutcome.Applied, response.Outcome);
        Assert.True(response.Success);
        Assert.Equal(1, writer.Calls); // the write WAS attempted
    }

    /// <summary>
    /// When the mirror row is missing (telemetry not ingested yet) or carries an unknown channel,
    /// the operator audit row is still written — falling back to ApiOutbound — so who acted is
    /// never lost.
    /// </summary>
    /// <param name="mirrorChannel">The mirror row's channel, or null for no mirror row.</param>
    [Theory]
    [InlineData(null)]
    [InlineData("NotAChannel")]
    public void RetrySiteCall_MirrorRowMissingOrUnknownChannel_AuditFallsBackToApiOutbound(string? mirrorChannel)
    {
        var id = Guid.NewGuid();
        var writer = new RecordingAuditWriter();
        var repo = new ScriptableRepo
        {
            OnGet = _ => Task.FromResult<SiteCall?>(
                mirrorChannel is null ? null : Row("site-north", DateTime.UtcNow, mirrorChannel)),
        };
        var central = CreateTestProbe();
        var actor = CreateActor(repo, FastTicks(), auditWriter: writer);
        actor.Tell(new RegisterCentralCommunication(central.Ref));

        actor.Tell(new RetrySiteCallRequest("corr", id, "site-north", RequestedBy: "jdoe"));
        var relay = (RetryParkedOperation)central.ExpectMsg<SiteEnvelope>().Message;
        central.Reply(new ParkedOperationActionAck(relay.CorrelationId, Applied: true));

        Assert.True(ExpectMsg<RetrySiteCallResponse>().Success);
        AwaitAssert(() =>
        {
            lock (writer.Events)
            {
                var evt = Assert.Single(writer.Events);
                Assert.Equal(AuditChannel.ApiOutbound, evt.Channel);
                Assert.Equal(AuditStatus.Submitted, evt.Status);
                Assert.Equal("jdoe", evt.Actor);
                Assert.Equal(id, evt.CorrelationId);
            }
        });
    }
}