ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogIngestActorTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Entities;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations;

namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central;

/// <summary>
/// Bundle D D2 tests for <see cref="AuditLogIngestActor"/>. Uses the same
/// <see cref="MsSqlMigrationFixture"/> as the M1 repository tests so the actor
/// exercises real <see cref="AuditLogRepository.InsertIfNotExistsAsync"/>
/// against a partitioned MSSQL schema (the only way to verify the
/// IngestedAtUtc stamp + duplicate-key idempotency end to end).
/// </summary>
public class AuditLogIngestActorTests : TestKit, IClassFixture<MsSqlMigrationFixture>
{
    private readonly MsSqlMigrationFixture _fixture;

    public AuditLogIngestActorTests(MsSqlMigrationFixture fixture)
    {
        _fixture = fixture;
    }

    private ScadaBridgeDbContext CreateContext()
    {
        var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
            .UseSqlServer(_fixture.ConnectionString)
            .Options;
        return new ScadaBridgeDbContext(options);
    }

    private static string NewSiteId() =>
        "test-bundle-d2-" + Guid.NewGuid().ToString("N").Substring(0, 8);

    private static AuditEvent NewEvent(string siteId, 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: siteId);

    private IActorRef CreateActor(IAuditLogRepository repository)
    {
        // Register row 61: the actor's repository-taking test ctor is gone; the fixture
        // supplies a stub provider that hands this repository out of every scope.
        var provider = TestSupport.StubAuditRepositoryProvider.Build(repository);
        return Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
            provider,
            NullLogger<AuditLogIngestActor>.Instance)));
    }

    [SkippableFact]
    public async Task Receive_BatchOf5_Calls_Repo_5Times_Acks_All_5()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        var siteId = NewSiteId();
        var events = Enumerable.Range(0, 5).Select(_ => NewEvent(siteId)).ToList();

        await using var context = CreateContext();
        var repo = new AuditLogRepository(context);
        var actor = CreateActor(repo);

        actor.Tell(new IngestAuditEventsCommand(events), TestActor);

        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
        Assert.Equal(5, reply.AcceptedEventIds.Count);
        Assert.True(events.Select(e => e.EventId).ToHashSet().SetEquals(reply.AcceptedEventIds.ToHashSet()));

        // Verify rows landed in MSSQL.
        await using var readContext = CreateContext();
        var rows = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .ToListAsync();
        Assert.Equal(5, rows.Count);
    }

    [SkippableFact]
    public async Task Receive_BatchWith_AlreadyExistingEvent_AcksAll_NoDoubleInsert()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        var siteId = NewSiteId();
        var pre = NewEvent(siteId);

        // Pre-insert one event directly via the repo so the actor sees it
        // already present when it processes the batch.
        await using (var seedContext = CreateContext())
        {
            var seedRepo = new AuditLogRepository(seedContext);
            await seedRepo.InsertIfNotExistsAsync(pre);
        }

        // Build the batch including the pre-existing event plus 2 new ones.
        var fresh1 = NewEvent(siteId);
        var fresh2 = NewEvent(siteId);
        var batch = new List<AuditEvent> { pre, fresh1, fresh2 };

        await using var context = CreateContext();
        var repo = new AuditLogRepository(context);
        var actor = CreateActor(repo);

        actor.Tell(new IngestAuditEventsCommand(batch), TestActor);

        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
        // All 3 acked under idempotent first-write-wins.
        Assert.Equal(3, reply.AcceptedEventIds.Count);

        // Verify no double-insert.
        await using var readContext = CreateContext();
        var count = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .CountAsync();
        Assert.Equal(3, count);
    }

    [SkippableFact]
    public async Task Receive_DuplicateEventIds_WithinOnePacket_ProduceOneRow_AndAreAllAcked()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        // WP2.2 writes each packet as ONE set-based statement, whose anti-semi-join
        // only sees already-COMMITTED rows. Two copies of an EventId inside one
        // packet would therefore both pass it and collide on the clustered PK,
        // taking the whole packet down — the repository de-duplicates first.
        // Every id is still acked: the site's contract is "this row is now
        // present at central", which is true for both copies.
        var siteId = NewSiteId();
        var repeated = NewEvent(siteId);
        var other = NewEvent(siteId);
        var batch = new List<AuditEvent> { repeated, other, repeated, repeated };

        await using var context = CreateContext();
        var repo = new AuditLogRepository(context);
        var actor = CreateActor(repo);

        actor.Tell(new IngestAuditEventsCommand(batch), TestActor);

        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
        Assert.Equal(4, reply.AcceptedEventIds.Count);
        Assert.True(
            new[] { repeated.EventId, other.EventId }.ToHashSet()
                .SetEquals(reply.AcceptedEventIds.ToHashSet()));

        await using var readContext = CreateContext();
        var rows = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .ToListAsync();
        Assert.Equal(2, rows.Count);
    }

    [SkippableFact]
    public async Task Receive_SamePacketTwice_IsIdempotent_AcrossPackets()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        // A site whose ack was lost re-delivers the identical packet on the next
        // drain, and the reconciliation pull can re-deliver it a third time. Each
        // replay must ack fully (so the site can finally flip its rows to
        // Forwarded) while writing nothing new.
        var siteId = NewSiteId();
        var events = Enumerable.Range(0, 6).Select(_ => NewEvent(siteId)).ToList();

        await using var context = CreateContext();
        var repo = new AuditLogRepository(context);
        var actor = CreateActor(repo);

        for (var attempt = 0; attempt < 3; attempt++)
        {
            actor.Tell(new IngestAuditEventsCommand(events), TestActor);
            var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
            Assert.Equal(6, reply.AcceptedEventIds.Count);
            Assert.True(
                events.Select(e => e.EventId).ToHashSet()
                    .SetEquals(reply.AcceptedEventIds.ToHashSet()));
        }

        await using var readContext = CreateContext();
        var rows = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .ToListAsync();
        Assert.Equal(6, rows.Count);
    }

    [SkippableFact]
    public async Task Receive_OverlappingPackets_InsertOnlyTheNewRows()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        // Partially-overlapping packets are the normal reconciliation shape: the
        // pull cursor re-serves a tail the push already delivered. The overlap
        // must be a silent no-op and the new rows must land.
        var siteId = NewSiteId();
        var first = Enumerable.Range(0, 4).Select(_ => NewEvent(siteId)).ToList();
        var second = first.Skip(2).Concat(
            Enumerable.Range(0, 3).Select(_ => NewEvent(siteId))).ToList();

        await using var context = CreateContext();
        var repo = new AuditLogRepository(context);
        var actor = CreateActor(repo);

        actor.Tell(new IngestAuditEventsCommand(first), TestActor);
        ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));

        actor.Tell(new IngestAuditEventsCommand(second), TestActor);
        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
        Assert.Equal(5, reply.AcceptedEventIds.Count);

        await using var readContext = CreateContext();
        var rows = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .ToListAsync();
        Assert.Equal(7, rows.Count);
    }

    [SkippableFact]
    public async Task Receive_Sets_IngestedAtUtc_Before_Insert()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        var siteId = NewSiteId();
        var events = Enumerable.Range(0, 3).Select(_ => NewEvent(siteId)).ToList();

        var before = DateTime.UtcNow.AddSeconds(-1);

        await using var context = CreateContext();
        var repo = new AuditLogRepository(context);
        var actor = CreateActor(repo);

        actor.Tell(new IngestAuditEventsCommand(events), TestActor);
        ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));

        var after = DateTime.UtcNow.AddSeconds(1);

        await using var readContext = CreateContext();
        var rows = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .ToListAsync();

        Assert.Equal(3, rows.Count);
        Assert.All(rows, r =>
        {
            Assert.NotNull(r.IngestedAtUtc);
            Assert.InRange(r.IngestedAtUtc!.Value, before, after);
        });
    }

    [SkippableFact]
    public async Task Receive_RepoThrowsForOneEvent_Other4StillPersisted()
    {
        Skip.IfNot(_fixture.Available, _fixture.SkipReason);

        var siteId = NewSiteId();
        var events = Enumerable.Range(0, 5).Select(_ => NewEvent(siteId)).ToList();
        var poisonId = events[2].EventId;

        // Wrapper repo that throws only when the poison EventId is being
        // inserted. The four neighbours must still land in MSSQL.
        await using var context = CreateContext();
        var realRepo = new AuditLogRepository(context);
        var wrappedRepo = new ThrowingRepository(realRepo, poisonId);
        var actor = CreateActor(wrappedRepo);

        actor.Tell(new IngestAuditEventsCommand(events), TestActor);
        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));

        // The actor catches the throw per-row, so 4 ids are accepted and 1 is
        // left out.
        Assert.Equal(4, reply.AcceptedEventIds.Count);
        Assert.DoesNotContain(poisonId, reply.AcceptedEventIds);

        await using var readContext = CreateContext();
        var rows = await readContext.Set<AuditLogRow>()
            .Where(e => e.SourceSiteId == siteId)
            .ToListAsync();
        Assert.Equal(4, rows.Count);
        Assert.DoesNotContain(rows, r => r.EventId == poisonId);
    }

    [Fact]
    public async Task Receive_WhenRepositoryResolutionThrows_ActorSurvives_RepliesEmpty_CountsFailure()
    {
        // AuditLog-017 (covers the AuditLog-014 guard): the production ctor resolves the
        // scoped repository per message. If scope creation / repository resolution throws
        // (transient DI or DbContext-factory fault, pooled-context init, a resolution race
        // during host churn), the outer guard must keep the singleton ALIVE, increment the
        // failure counter, and still reply with whatever was accepted (empty here) so the
        // site keeps its rows Pending and retries — rather than letting the throw restart
        // the singleton and drop the captured reply (the site's Ask would then time out).
        var counter = new CountingFailureCounter();

        // A provider with NO IAuditLogRepository registered → GetRequiredService throws
        // inside the per-message scope; the failure counter IS registered so the guard's
        // catch can surface the fault.
        var services = new ServiceCollection();
        services.AddSingleton<ICentralAuditWriteFailureCounter>(counter);
        await using var provider = services.BuildServiceProvider();

        var actor = Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
            (IServiceProvider)provider, NullLogger<AuditLogIngestActor>.Instance)));

        // First batch: resolution throws → empty reply, one counted failure, no restart.
        actor.Tell(new IngestAuditEventsCommand(
            Enumerable.Range(0, 3).Select(_ => NewEvent(NewSiteId())).ToList()), TestActor);
        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
        Assert.Empty(reply.AcceptedEventIds);
        Assert.Equal(1, counter.Count);

        // Second batch proves the actor was not restarted/wedged: it still processes
        // messages and the guard fires again.
        actor.Tell(new IngestAuditEventsCommand(
            new List<AuditEvent> { NewEvent(NewSiteId()) }), TestActor);
        var reply2 = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
        Assert.Empty(reply2.AcceptedEventIds);
        Assert.Equal(2, counter.Count);
    }

    /// <summary>
    /// The per-row fallback must run on its OWN cancellation token, never the
    /// batch's. Sharing it meant that when the batch failed BECAUSE the 20 s
    /// ingest budget expired, every fallback insert was handed an
    /// already-cancelled token: N instant failures, N counter bumps, zero rows
    /// accepted — the fallback's entire purpose (land the good rows) defeated at
    /// the exact moment it was needed.
    /// </summary>
    /// <remarks>
    /// Pinned by token IDENTITY rather than by expiring a real budget: the budget
    /// is a fixed 20 s and a test that waited for it would cost 20 s of wall clock
    /// to assert something the token comparison establishes outright. Two
    /// <see cref="CancellationToken"/>s are equal iff they come from the same
    /// source, so "not equal" IS "a fresh CTS", and a fresh CTS cannot be
    /// pre-cancelled by the batch.
    /// </remarks>
    [Fact]
    public async Task Receive_WhenBatchFails_PerRowFallbackUsesAFreshToken()
    {
        var repository = new TokenRecordingRepository();
        var actor = CreateActor(repository);

        var events = Enumerable.Range(0, 3).Select(_ => NewEvent(NewSiteId())).ToList();
        actor.Tell(new IngestAuditEventsCommand(events), TestActor);

        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));

        // The fallback landed every row.
        Assert.Equal(3, reply.AcceptedEventIds.Count);

        Assert.NotNull(repository.BatchToken);
        Assert.Equal(3, repository.RowTokens.Count);
        Assert.All(repository.RowTokens, rowToken =>
        {
            Assert.NotEqual(repository.BatchToken!.Value, rowToken);
            Assert.False(rowToken.IsCancellationRequested);
        });

        await Task.CompletedTask;
    }

    /// <summary>Counts how many times the guard's catch surfaced a write failure.</summary>
    private sealed class CountingFailureCounter : ICentralAuditWriteFailureCounter
    {
        public int Count { get; private set; }
        public void Increment() => Count++;
    }

    /// <summary>
    /// Fails the set-based insert (as an expired budget would) and records the
    /// token handed to each write so the fallback's token can be compared with
    /// the batch's.
    /// </summary>
    private sealed class TokenRecordingRepository : IAuditLogRepository
    {
        public CancellationToken? BatchToken { get; private set; }

        public List<CancellationToken> RowTokens { get; } = new();

        public Task<int> InsertManyIfNotExistsAsync(
            IReadOnlyList<AuditEvent> events, TimeSpan? commandTimeout = null, CancellationToken ct = default)
        {
            BatchToken = ct;
            throw new OperationCanceledException("simulated ingest-budget expiry", ct);
        }

        public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
        {
            RowTokens.Add(ct);
            return Task.CompletedTask;
        }

        public Task<IReadOnlyList<AuditEvent>> QueryAsync(
            AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<long> SwitchOutPartitionAsync(
            DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<long> PurgeChannelOlderThanAsync(
            string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<long> BackfillSourceNodeAsync(
            string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<IReadOnlyList<DateTime>> GetPartitionBoundariesOlderThanAsync(
            DateTime threshold, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot> GetKpiSnapshotAsync(
            TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
            Guid executionId, CancellationToken ct = default) =>
            throw new NotSupportedException();

        public Task<IReadOnlyList<string>> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
            throw new NotSupportedException();
    }

    /// <summary>
    /// Tiny test double that delegates to a real repository but throws on a
    /// specified EventId. Used to verify per-row failure isolation: one bad
    /// row must not cause the rest of the batch to be lost.
    /// </summary>
    private sealed class ThrowingRepository : IAuditLogRepository
    {
        private readonly IAuditLogRepository _inner;
        private readonly Guid _poisonId;

        public ThrowingRepository(IAuditLogRepository inner, Guid poisonId)
        {
            _inner = inner;
            _poisonId = poisonId;
        }

        public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
        {
            if (evt.EventId == _poisonId)
            {
                throw new InvalidOperationException("simulated repo failure for poison row");
            }
            return _inner.InsertIfNotExistsAsync(evt, ct);
        }

        public Task<IReadOnlyList<AuditEvent>> QueryAsync(
            AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
            _inner.QueryAsync(filter, paging, ct);

        public Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
            _inner.SwitchOutPartitionAsync(monthBoundary, commandTimeout, ct);

        public Task<long> PurgeChannelOlderThanAsync(
            string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
            _inner.PurgeChannelOlderThanAsync(channel, threshold, batchSize, commandTimeout, ct);

        public Task<long> BackfillSourceNodeAsync(
            string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
            _inner.BackfillSourceNodeAsync(sentinel, before, batchSize, ct);

        public Task<IReadOnlyList<DateTime>> GetPartitionBoundariesOlderThanAsync(
            DateTime threshold, CancellationToken ct = default) =>
            _inner.GetPartitionBoundariesOlderThanAsync(threshold, ct);

        public Task<ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot> GetKpiSnapshotAsync(
            TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) =>
            _inner.GetKpiSnapshotAsync(window, nowUtc, ct);

        public Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
            Guid executionId, CancellationToken ct = default) =>
            _inner.GetExecutionTreeAsync(executionId, ct);

        public Task<IReadOnlyList<string>> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
            _inner.GetDistinctSourceNodesAsync(ct);
    }
}