ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Integration/SyncCallEmissionEndToEndTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry;
using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Integration.Infrastructure;
using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
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.Integration;

/// <summary>
/// Bundle H — end-to-end test wiring the full Audit Log #23 M2 sync-call pipeline:
/// <see cref="FallbackAuditWriter"/> over a <see cref="SqliteAuditWriter"/> backed by
/// an in-memory SQLite database; the <see cref="SiteAuditTelemetryActor"/> drains
/// Pending rows and pushes them through a stub <see cref="ISiteStreamAuditClient"/>
/// that forwards directly to the central <see cref="AuditLogIngestActor"/> backed
/// by a real <see cref="AuditLogRepository"/> on the <see cref="MsSqlMigrationFixture"/>.
/// </summary>
/// <remarks>
/// <para>
/// This is a <b>component-level</b> integration test, not a full Akka-cluster
/// test (per the M2 brainstorm decision). The stub gRPC client short-circuits
/// the wire so we exercise the real telemetry actor, the real ingest actor, the
/// real SQLite writer, and the real MSSQL repository — without standing up a
/// Kestrel host or two-cluster topology.
/// </para>
/// <para>
/// The site-side telemetry actor's <c>Drain</c> message is private; rather than
/// expose it we drive the drain by setting <c>BusyIntervalSeconds = 1</c> so the
/// initial scheduled tick fires within a second of actor start. Tests then
/// <see cref="TestKitBase.AwaitAssertAsync"/> until the central repository
/// observes the expected rows.
/// </para>
/// <para>
/// Each test uses a unique <c>SourceSiteId</c> (Guid suffix) so concurrent tests
/// and the per-fixture MSSQL database lifetime don't interfere with each other.
/// </para>
/// </remarks>
public class SyncCallEmissionEndToEndTests : TestKit, IClassFixture<MsSqlMigrationFixture>
{
    private readonly MsSqlMigrationFixture _fixture;

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

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

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

    private static AuditEvent NewEvent(string siteId, Guid? id = null) => ScadaBridgeAuditEventFactory.Create(
        eventId: id ?? Guid.NewGuid(),
        occurredAtUtc: DateTime.UtcNow,
        channel: AuditChannel.ApiOutbound,
        kind: AuditKind.ApiCall,
        status: AuditStatus.Delivered,
        sourceSiteId: siteId,
        target: "external-system-a/method");

    private static IOptions<SqliteAuditWriterOptions> InMemorySqliteOptions() =>
        Options.Create(new SqliteAuditWriterOptions
        {
            // Per-test unique database name + Mode=Memory + Cache=Shared keeps
            // the in-memory database alive for the duration of the test even
            // though Microsoft.Data.Sqlite tears the file down with the last
            // connection. The DatabasePath field is unused because we override
            // the connection string below.
            DatabasePath = "ignored",
            BatchSize = 64,
            ChannelCapacity = 1024,
        });

    private static SqliteAuditWriter CreateInMemorySqliteWriter() =>
        // The 4th constructor argument is connectionStringOverride. A unique
        // shared-cache in-memory URI keeps the schema scoped to this writer
        // instance and torn down when the writer is disposed.
        new SqliteAuditWriter(
            InMemorySqliteOptions(),
            NullLogger<SqliteAuditWriter>.Instance,
            new FakeNodeIdentityProvider(),
            connectionStringOverride: $"Data Source=file:auditlog-h-{Guid.NewGuid():N}?mode=memory&cache=shared");

    private static IOptions<SiteAuditTelemetryOptions> FastTelemetryOptions() =>
        Options.Create(new SiteAuditTelemetryOptions
        {
            BatchSize = 256,
            // 1s for both intervals so the initial scheduled tick fires fast
            // and any failure-driven re-tick also fires fast — without
            // requiring a public Drain message to be exposed.
            BusyIntervalSeconds = 1,
            IdleIntervalSeconds = 1,
        });

    private IActorRef CreateIngestActor(IAuditLogRepository repo)
    {
        // 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(repo);
        return Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
            provider,
            NullLogger<AuditLogIngestActor>.Instance)));
    }

    private IActorRef CreateTelemetryActor(
        ISiteAuditQueue queue,
        ISiteStreamAuditClient client) =>
        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
            queue,
            client,
            FastTelemetryOptions(),
            NullLogger<SiteAuditTelemetryActor>.Instance)));

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

        var siteId = NewSiteId();

        // Real central wiring: repo + ingest actor.
        await using var ingestContext = CreateContext();
        var ingestRepo = new AuditLogRepository(ingestContext);
        var ingestActor = CreateIngestActor(ingestRepo);

        // Real site wiring: SQLite (in-memory) + ring + fallback + telemetry.
        await using var sqliteWriter = CreateInMemorySqliteWriter();
        var ring = new RingBufferFallback();
        var fallback = new FallbackAuditWriter(
            sqliteWriter,
            ring,
            new NoOpAuditWriteFailureCounter(),
            NullLogger<FallbackAuditWriter>.Instance);

        var stubClient = new DirectActorSiteStreamAuditClient(ingestActor);
        CreateTelemetryActor(sqliteWriter, stubClient);

        // Act — one fresh event written via the FallbackAuditWriter hot-path.
        var evt = NewEvent(siteId);
        await fallback.WriteAsync(evt);

        // Assert — the central AuditLog row materialises within a window that
        // covers initial tick (1s) + a generous slack for SQLite + the actor
        // round-trip + EF/MSSQL latency.
        await AwaitAssertAsync(async () =>
        {
            await using var readContext = CreateContext();
            var readRepo = new AuditLogRepository(readContext);
            var rows = await readRepo.QueryAsync(
                new AuditLogQueryFilter(SourceSiteIds: new[] { siteId }),
                new AuditLogPaging(PageSize: 10));
            Assert.Single(rows);
            Assert.Equal(evt.EventId, rows[0].EventId);
            // Central stamps IngestedAtUtc; site never sets it.
            Assert.NotNull(rows[0].AsRow().IngestedAtUtc);
        }, TimeSpan.FromSeconds(15));
    }

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

        var siteId = NewSiteId();

        await using var ingestContext = CreateContext();
        var ingestRepo = new AuditLogRepository(ingestContext);
        var ingestActor = CreateIngestActor(ingestRepo);

        await using var sqliteWriter = CreateInMemorySqliteWriter();
        var ring = new RingBufferFallback();
        var fallback = new FallbackAuditWriter(
            sqliteWriter,
            ring,
            new NoOpAuditWriteFailureCounter(),
            NullLogger<FallbackAuditWriter>.Instance);

        // Stub fails the first push; subsequent calls flow through. The
        // telemetry actor's on-failure branch keeps rows in Pending state, so
        // the next tick re-reads them and tries again.
        var stubClient = new DirectActorSiteStreamAuditClient(ingestActor)
        {
            FailNextCallCount = 1,
        };
        CreateTelemetryActor(sqliteWriter, stubClient);

        var evt = NewEvent(siteId);
        await fallback.WriteAsync(evt);

        // Wait long enough for at least one failure-then-success cycle. With
        // both intervals = 1s the actor retries quickly; allow 15s for slow CI.
        await AwaitAssertAsync(async () =>
        {
            await using var readContext = CreateContext();
            var readRepo = new AuditLogRepository(readContext);
            var rows = await readRepo.QueryAsync(
                new AuditLogQueryFilter(SourceSiteIds: new[] { siteId }),
                new AuditLogPaging(PageSize: 10));
            Assert.Single(rows);
            Assert.Equal(evt.EventId, rows[0].EventId);
        }, TimeSpan.FromSeconds(15));

        // Safe as a bare assertion: DirectActorSiteStreamAuditClient increments
        // CallCount as its FIRST statement, before the Ask that writes the
        // central row — so the row existing already implies the second call.
        Assert.True(stubClient.CallCount >= 2,
            $"Expected at least one failed push + one successful push; saw {stubClient.CallCount} total client calls.");

        // The site SQLite row must have flipped to Forwarded after the
        // successful retry. ReadPendingAsync only returns Pending rows; the
        // row should NOT show up there anymore.
        //
        // AwaitAssert, not a bare Assert: the drain marks rows forwarded only
        // AFTER the push returns and its ack is parsed — SiteAuditTelemetryActor
        // pushes via IngestAuditEventsAsync (which is what writes the central
        // row) and only then calls MarkForwardedAsync. Observing the central row
        // above therefore establishes no happens-before edge with the site-side
        // state flip; on a loaded run the post-push continuation can be scheduled
        // after the poll that saw the row, leaving it still Pending. Reproduced
        // deterministically by delaying only that post-push step, which fails
        // exactly this test with "Assert.DoesNotContain() Failure: Filter matched
        // in collection". The bounded wait removes the ordering assumption only —
        // the row must still actually leave Pending or the test fails as before.
        await AwaitAssertAsync(async () =>
        {
            var stillPending = await sqliteWriter.ReadPendingAsync(64);
            Assert.DoesNotContain(stillPending, p => p.EventId == evt.EventId);
        }, TimeSpan.FromSeconds(15));
    }

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

        var siteId = NewSiteId();

        await using var ingestContext = CreateContext();
        var ingestRepo = new AuditLogRepository(ingestContext);
        var ingestActor = CreateIngestActor(ingestRepo);

        await using var sqliteWriter = CreateInMemorySqliteWriter();
        var ring = new RingBufferFallback();
        var fallback = new FallbackAuditWriter(
            sqliteWriter,
            ring,
            new NoOpAuditWriteFailureCounter(),
            NullLogger<FallbackAuditWriter>.Instance);

        var stubClient = new DirectActorSiteStreamAuditClient(ingestActor);
        CreateTelemetryActor(sqliteWriter, stubClient);

        // Both writes carry the SAME EventId. Site SQLite's PRIMARY KEY
        // constraint and the central repo's InsertIfNotExistsAsync both
        // enforce first-write-wins, so only one central row must materialise.
        var sharedId = Guid.NewGuid();
        var evt1 = NewEvent(siteId, sharedId);
        var evt2 = NewEvent(siteId, sharedId);

        await fallback.WriteAsync(evt1);
        await fallback.WriteAsync(evt2);

        await AwaitAssertAsync(async () =>
        {
            await using var readContext = CreateContext();
            var readRepo = new AuditLogRepository(readContext);
            var rows = await readRepo.QueryAsync(
                new AuditLogQueryFilter(SourceSiteIds: new[] { siteId }),
                new AuditLogPaging(PageSize: 10));
            Assert.Single(rows);
            Assert.Equal(sharedId, rows[0].EventId);
        }, TimeSpan.FromSeconds(15));
    }

}