ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using NSubstitute.ReceivedExtensions;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;

namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site.Telemetry;

/// <summary>
/// Bundle D D1 tests for <see cref="SiteAuditTelemetryActor"/>. The actor drains
/// the site SQLite queue via <see cref="ISiteAuditQueue"/>, pushes batches via
/// <see cref="ISiteStreamAuditClient"/>, and flips ack'd rows to Forwarded.
/// Both collaborators are NSubstitute mocks so the tests never touch real
/// SQLite or gRPC.
/// </summary>
public class SiteAuditTelemetryActorTests : TestKit
{
    private readonly ISiteAuditQueue _queue = Substitute.For<ISiteAuditQueue>();
    private readonly ISiteStreamAuditClient _client = Substitute.For<ISiteStreamAuditClient>();
    private readonly IOperationTrackingStore _trackingStore = Substitute.For<IOperationTrackingStore>();

    /// <summary>
    /// Fast options so tests don't stall waiting for the scheduler. 1s busy /
    /// 2s idle still exercises the busy-vs-idle branching, but each test
    /// completes in &lt; 5 s wall-clock.
    /// </summary>
    private static IOptions<SiteAuditTelemetryOptions> Opts(
        int batchSize = 256,
        int busySeconds = 1,
        int idleSeconds = 2) =>
        Options.Create(new SiteAuditTelemetryOptions
        {
            BatchSize = batchSize,
            BusyIntervalSeconds = busySeconds,
            IdleIntervalSeconds = idleSeconds,
        });

    private IActorRef CreateActor(IOptions<SiteAuditTelemetryOptions>? options = null) =>
        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
            _queue,
            _client,
            options ?? Opts(),
            NullLogger<SiteAuditTelemetryActor>.Instance,
            (IOperationTrackingStore?)null)));

    /// <summary>
    /// AuditLog-001: builds an actor with the optional
    /// <see cref="IOperationTrackingStore"/> wired in so the cached-drain
    /// scheduler is armed alongside the audit-only drain. Used by the new
    /// cached-drain regression tests below.
    /// </summary>
    private IActorRef CreateActorWithCachedDrain(IOptions<SiteAuditTelemetryOptions>? options = null) =>
        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
            _queue,
            _client,
            options ?? Opts(),
            NullLogger<SiteAuditTelemetryActor>.Instance,
            (IOperationTrackingStore?)_trackingStore)));

    private static AuditEvent NewEvent(Guid? id = null) => ScadaBridgeAuditEventFactory.Create(
        eventId: id ?? Guid.NewGuid(),
        occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
        channel: AuditChannel.ApiOutbound,
        kind: AuditKind.ApiCall,
        status: AuditStatus.Delivered,
        sourceSiteId: "site-1");

    private static IngestAck AckAll(IReadOnlyList<AuditEvent> events)
    {
        var ack = new IngestAck();
        foreach (var e in events)
        {
            ack.AcceptedEventIds.Add(e.EventId.ToString());
        }
        return ack;
    }

    [Fact]
    public async Task Drain_With_50PendingRows_Sends_OneBatch_Of_50_Then_FlipsToForwarded()
    {
        // Arrange — 50 pending rows on the first read, then empty on subsequent
        // reads so the actor settles after one productive drain.
        var pending = Enumerable.Range(0, 50).Select(_ => NewEvent()).ToList();
        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(
                Task.FromResult<IReadOnlyList<AuditEvent>>(pending),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        AuditEventBatch? capturedBatch = null;
        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
            .Returns(call =>
            {
                capturedBatch = call.Arg<AuditEventBatch>();
                return Task.FromResult(AckAll(pending));
            });

        // Act
        CreateActor();

        // Assert — give the scheduler time to fire the initial Drain tick.
        await AwaitAssertAsync(async () =>
        {
            await _client.Received(1).IngestAuditEventsAsync(
                Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>());
            await _queue.Received(1).MarkForwardedAsync(
                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 50), Arg.Any<CancellationToken>());
        }, TimeSpan.FromSeconds(5));

        Assert.NotNull(capturedBatch);
        Assert.Equal(50, capturedBatch!.Events.Count);

        var expected = pending.Select(e => e.EventId).ToHashSet();
        await _queue.Received(1).MarkForwardedAsync(
            Arg.Is<IReadOnlyList<Guid>>(g => g.ToHashSet().SetEquals(expected)),
            Arg.Any<CancellationToken>());
    }

    [Fact]
    public async Task Drain_GrpcThrows_RowsStayPending_NextDrainRetries()
    {
        // Arrange — first read returns 3 rows; the gRPC client throws on the
        // first push, then succeeds on the second. After the second push the
        // queue returns empty so the actor settles.
        var batch = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(
                Task.FromResult<IReadOnlyList<AuditEvent>>(batch),
                Task.FromResult<IReadOnlyList<AuditEvent>>(batch),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        var calls = 0;
        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
            .Returns(_ =>
            {
                calls++;
                if (calls == 1)
                {
                    throw new InvalidOperationException("simulated gRPC failure");
                }
                return Task.FromResult(AckAll(batch));
            });

        // Act
        CreateActor();

        // Assert — eventually MarkForwardedAsync is called exactly once (after
        // the retry succeeded). The first failure must NOT have called
        // MarkForwardedAsync because the rows stay Pending.
        await AwaitAssertAsync(async () =>
        {
            await _queue.Received(1).MarkForwardedAsync(
                Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
        }, TimeSpan.FromSeconds(10));

        Assert.True(calls >= 2, $"Expected at least 2 client calls (1 failure + 1 retry); saw {calls}");
    }

    [Fact]
    public async Task Drain_ZeroPending_SchedulesAtIdleInterval_NoClientCall()
    {
        // Arrange — queue always empty.
        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        // Idle interval = 2 s. Pause 3 s after the first tick (1 s busy on
        // PreStart) and assert the empty-queue branch did NOT push to the
        // client.
        CreateActor(Opts(busySeconds: 1, idleSeconds: 2));

        // Allow the initial tick (~1 s) + a generous window for the idle re-tick.
        await Task.Delay(TimeSpan.FromSeconds(3));

        await _client.DidNotReceiveWithAnyArgs().IngestAuditEventsAsync(default!, default);

        // ReadPendingAsync was called at least once (initial tick), and at
        // most twice within the 3 s window (initial + one idle re-tick).
        var readCalls = _queue.ReceivedCalls()
            .Count(c => c.GetMethodInfo().Name == nameof(ISiteAuditQueue.ReadPendingAsync));
        Assert.InRange(readCalls, 1, 2);
    }

    [Fact]
    public async Task Drain_NonZeroPending_SchedulesAtBusyInterval()
    {
        // Arrange — every read returns 1 row. With busy=1s the actor should
        // re-drain quickly, producing multiple client calls inside a short
        // window.
        var single = new List<AuditEvent> { NewEvent() };
        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(single));

        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
            .Returns(call => Task.FromResult(AckAll(single)));

        CreateActor(Opts(busySeconds: 1, idleSeconds: 10));

        // 3-second window with busy=1s should fit at least 2 drains.
        await Task.Delay(TimeSpan.FromSeconds(3));

        var pushCalls = _client.ReceivedCalls()
            .Count(c => c.GetMethodInfo().Name == nameof(ISiteStreamAuditClient.IngestAuditEventsAsync));
        Assert.True(pushCalls >= 2,
            $"Expected ≥2 pushes within 3s when busy=1s; saw {pushCalls}");
    }

    [Fact]
    public async Task Drain_AcceptedEventIdsSubset_OnlyMarksAccepted()
    {
        // Arrange — 5 rows pushed, but the central ack only lists 3.
        var rows = Enumerable.Range(0, 5).Select(_ => NewEvent()).ToList();
        var ackedIds = rows.Take(3).Select(r => r.EventId).ToList();

        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(
                Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        var partialAck = new IngestAck();
        foreach (var id in ackedIds)
        {
            partialAck.AcceptedEventIds.Add(id.ToString());
        }
        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
            .Returns(Task.FromResult(partialAck));

        // Act
        CreateActor();

        await AwaitAssertAsync(async () =>
        {
            await _queue.Received(1).MarkForwardedAsync(
                Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
        }, TimeSpan.FromSeconds(5));

        // Assert — exactly the 3 ack'd ids made it to MarkForwardedAsync, not
        // the other 2.
        var ackedSet = ackedIds.ToHashSet();
        await _queue.Received(1).MarkForwardedAsync(
            Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3 && g.ToHashSet().SetEquals(ackedSet)),
            Arg.Any<CancellationToken>());
    }

    // ────────────────────────────────────────────────────────────────────────
    // AuditLog-001: combined-telemetry cached-drain regression tests. Verify
    // that the production wiring of the previously-unreachable cached transport
    // routes cached rows through ReadPendingCachedTelemetryAsync +
    // IngestCachedTelemetryAsync (and NOT IngestAuditEventsAsync), and that
    // orphaned audit rows (no tracking snapshot) are logged + skipped rather
    // than crashing the drain.
    // ────────────────────────────────────────────────────────────────────────

    private static AuditEvent NewCachedEvent(
        AuditKind kind = AuditKind.CachedSubmit,
        Guid? eventId = null,
        Guid? correlationId = null,
        string sourceSiteId = "site-1") =>
        // C3 (Task 2.5): canonical record via the shared factory; ForwardState is
        // no longer a record field (the SQLite shim defaults it on INSERT).
        ScadaBridgeAuditEventFactory.Create(
            channel: AuditChannel.ApiOutbound,
            kind: kind,
            status: AuditStatus.Submitted,
            eventId: eventId ?? Guid.NewGuid(),
            occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
            target: "ERP.GetOrder",
            sourceSiteId: sourceSiteId,
            correlationId: correlationId ?? Guid.NewGuid());

    private static TrackingStatusSnapshot NewSnapshot(
        TrackedOperationId id,
        string status = "Submitted",
        int retryCount = 0) => new(
            Id: id,
            Kind: nameof(AuditKind.ApiCallCached),
            TargetSummary: "ERP.GetOrder",
            Status: status,
            RetryCount: retryCount,
            LastError: null,
            HttpStatus: null,
            CreatedAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
            UpdatedAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
            TerminalAtUtc: null,
            SourceInstanceId: "instance-1",
            SourceScript: "script-1",
            SourceNode: "node-a");

    [Fact]
    public async Task CachedDrain_CachedRows_RouteThrough_IngestCachedTelemetry_NotIngestAuditEvents()
    {
        // Arrange — three cached audit rows on the cached queue, each with a
        // matching tracking snapshot. The audit-only queue is empty (those
        // rows are excluded by ReadPendingAsync after AuditLog-001).
        var cachedRows = new[]
        {
            NewCachedEvent(AuditKind.CachedSubmit),
            NewCachedEvent(AuditKind.ApiCallCached),
            NewCachedEvent(AuditKind.CachedResolve),
        };

        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
        _queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(
                Task.FromResult<IReadOnlyList<AuditEvent>>(cachedRows),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        foreach (var row in cachedRows)
        {
            var tid = new TrackedOperationId(row.CorrelationId!.Value);
            _trackingStore.GetStatusAsync(tid, Arg.Any<CancellationToken>())
                .Returns(Task.FromResult<TrackingStatusSnapshot?>(NewSnapshot(tid)));
        }

        CachedTelemetryBatch? capturedBatch = null;
        _client.IngestCachedTelemetryAsync(Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>())
            .Returns(call =>
            {
                capturedBatch = call.Arg<CachedTelemetryBatch>();
                var ack = new IngestAck();
                foreach (var packet in capturedBatch.Packets)
                {
                    ack.AcceptedEventIds.Add(packet.AuditEvent.EventId);
                }
                return Task.FromResult(ack);
            });

        // Act
        CreateActorWithCachedDrain();

        // Assert — exactly one IngestCachedTelemetryAsync push containing all
        // three packets, and zero IngestAuditEventsAsync pushes (the audit-only
        // drain saw an empty queue).
        await AwaitAssertAsync(async () =>
        {
            await _client.Received(1).IngestCachedTelemetryAsync(
                Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>());
            await _queue.Received(1).MarkForwardedAsync(
                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3), Arg.Any<CancellationToken>());
        }, TimeSpan.FromSeconds(5));

        Assert.NotNull(capturedBatch);
        Assert.Equal(3, capturedBatch!.Packets.Count);

        await _client.DidNotReceiveWithAnyArgs().IngestAuditEventsAsync(default!, default);

        var emittedEventIds = capturedBatch.Packets
            .Select(p => Guid.Parse(p.AuditEvent.EventId))
            .ToHashSet();
        var expectedIds = cachedRows.Select(r => r.EventId).ToHashSet();
        Assert.Equal(expectedIds, emittedEventIds);
    }

    [Fact]
    public async Task CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash()
    {
        // Arrange — two cached audit rows: one with a tracking snapshot, one
        // orphaned (the tracking store returns null). The orphaned row must be
        // skipped without aborting the batch — the valid row still flows.
        var orphan = NewCachedEvent(AuditKind.CachedSubmit);
        var valid = NewCachedEvent(AuditKind.CachedResolve);

        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
        _queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(
                Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { orphan, valid }),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        // orphan: tracking returns null
        _trackingStore.GetStatusAsync(
                new TrackedOperationId(orphan.CorrelationId!.Value),
                Arg.Any<CancellationToken>())
            .Returns(Task.FromResult<TrackingStatusSnapshot?>(null));
        // valid: tracking returns a snapshot
        var validTid = new TrackedOperationId(valid.CorrelationId!.Value);
        _trackingStore.GetStatusAsync(validTid, Arg.Any<CancellationToken>())
            .Returns(Task.FromResult<TrackingStatusSnapshot?>(NewSnapshot(validTid, "Delivered")));

        CachedTelemetryBatch? capturedBatch = null;
        _client.IngestCachedTelemetryAsync(Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>())
            .Returns(call =>
            {
                capturedBatch = call.Arg<CachedTelemetryBatch>();
                var ack = new IngestAck();
                foreach (var packet in capturedBatch.Packets)
                {
                    ack.AcceptedEventIds.Add(packet.AuditEvent.EventId);
                }
                return Task.FromResult(ack);
            });

        // Act
        CreateActorWithCachedDrain();

        // Assert — exactly one push containing ONLY the valid row; the orphan
        // is skipped and stays Pending (not in MarkForwardedAsync's id list).
        await AwaitAssertAsync(async () =>
        {
            await _client.Received(1).IngestCachedTelemetryAsync(
                Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>());
        }, TimeSpan.FromSeconds(5));

        Assert.NotNull(capturedBatch);
        Assert.Single(capturedBatch!.Packets);
        Assert.Equal(valid.EventId.ToString(), capturedBatch.Packets[0].AuditEvent.EventId);

        await _queue.Received(1).MarkForwardedAsync(
            Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == valid.EventId),
            Arg.Any<CancellationToken>());
    }

    [Fact]
    public async Task AuditOnlyDrain_StillFlows_When_CachedDrain_IsDisabled()
    {
        // Arrange — ordinary (non-cached) audit rows on the audit-only queue;
        // the actor is constructed WITHOUT a tracking store so the cached
        // scheduler is never armed. Regression guard against the audit-only
        // drain regressing during the AuditLog-001 refactor.
        var rows = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(
                Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));

        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
            .Returns(_ => Task.FromResult(AckAll(rows)));

        // Act — note: CreateActor (no tracking store), not CreateActorWithCachedDrain.
        CreateActor();

        // Assert — audit-only drain flows normally; the cached client is
        // never called and ReadPendingCachedTelemetryAsync is never queried.
        await AwaitAssertAsync(async () =>
        {
            await _client.Received(1).IngestAuditEventsAsync(
                Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>());
            await _queue.Received(1).MarkForwardedAsync(
                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3), Arg.Any<CancellationToken>());
        }, TimeSpan.FromSeconds(5));

        await _client.DidNotReceiveWithAnyArgs().IngestCachedTelemetryAsync(default!, default);
        await _queue.DidNotReceiveWithAnyArgs().ReadPendingCachedTelemetryAsync(default, default);
    }

    /// <summary>
    /// Regression for the 2026-07-20 rig finding (known-issues/2026-07-20-localdb-disk-io-error-under-load.md §8):
    /// both drain handlers await with <c>ConfigureAwait(false)</c>, so when a
    /// dependency call completes off the actor thread (as real SQLite/gRPC
    /// always do) the <c>finally</c>-block re-arm runs on a pool thread with no
    /// active ActorContext. Depending on what that thread's thread-static cell
    /// slot holds, <c>Context</c>/<c>Self</c> either throw
    /// <c>NotSupportedException: There is no active ActorContext</c> (crashing
    /// the actor once per drain — the rig's logged variant) or silently resolve
    /// to a STALE cell of some other actor, misrouting the tick so the drain
    /// loop stops. The two assertions below catch one variant each. Every other
    /// test in this class masks the bug by returning already-completed tasks
    /// from the mocks, which keeps the continuations on the actor thread.
    /// </summary>
    [Fact]
    public async Task Drains_Whose_Awaits_Complete_Off_The_Actor_Thread_Keep_Draining_Without_Crashing()
    {
        // Task.Delay completes on a timer thread; with ConfigureAwait(false)
        // everything after the await — including the finally-block re-arm —
        // stays off the actor context.
        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(async _ =>
            {
                await Task.Delay(25).ConfigureAwait(false);
                return (IReadOnlyList<AuditEvent>)Array.Empty<AuditEvent>();
            });
        _queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(async _ =>
            {
                await Task.Delay(25).ConfigureAwait(false);
                return (IReadOnlyList<AuditEvent>)Array.Empty<AuditEvent>();
            });

        // Variant 1 (throw → restart storm): any NotSupportedException logged
        // during the run fails the filter. Variant 2 (silent misroute → the
        // drain loop stalls after the first tick): the sustained-drain
        // assertion inside fails because reads stop at 1.
        await EventFilter.Exception<NotSupportedException>().ExpectAsync(0, async () =>
        {
            CreateActorWithCachedDrain(Opts(busySeconds: 1, idleSeconds: 1));

            // Three full cycles prove the finally-block re-arm works from an
            // off-context thread: tick → drain → re-arm → tick → …
            await AwaitAssertAsync(async () =>
            {
                await _queue.Received(Quantity.Within(3, int.MaxValue)).ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
                await _queue.Received(Quantity.Within(3, int.MaxValue)).ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
            }, TimeSpan.FromSeconds(10));
        });
    }
}