ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/AuditLog/SiteAuditPushFlowTests.cs

using System.Collections.Concurrent;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
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;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;

namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.AuditLog;

/// <summary>
/// End-to-end integration test for the Audit Log (#23) site→central push path
/// introduced by the "real site audit push client" follow-up.
/// </summary>
/// <remarks>
/// <para>
/// Exercises the full production chain in one actor system: the real
/// <see cref="SqliteAuditWriter"/> site SQLite hot-path, the real
/// <see cref="SiteAuditTelemetryActor"/> drain loop, the real
/// <see cref="SiteCommunicationAuditClient"/>, the real
/// <see cref="SiteCommunicationActor"/> forward, the real
/// <see cref="CentralCommunicationActor"/> routing, and the real
/// <c>AuditLogIngestActor</c> ingest — only the cross-cluster gRPC transport itself is
/// substituted by an in-process <see cref="BridgeCentralTransport"/> that Tells the central
/// actor exactly as the real <c>GrpcCentralTransport</c> would (a multi-node cluster is out of
/// scope for an in-process test).
/// </para>
/// <para>
/// The central audit store is an in-memory <see cref="IAuditLogRepository"/> —
/// the production <c>AuditLogRepository</c> emits SQL Server-specific T-SQL and
/// needs an MSSQL container, which this test deliberately avoids. The test
/// asserts both ends of the contract: a central <c>AuditLog</c> row appears AND
/// the site SQLite row flips from <see cref="AuditForwardState.Pending"/> to
/// <see cref="AuditForwardState.Forwarded"/>.
/// </para>
/// </remarks>
public class SiteAuditPushFlowTests : TestKit
{
    /// <summary>
    /// In-process stand-in for the site→central gRPC transport (<c>GrpcCentralTransport</c>):
    /// each site→central send is Tell'd straight to the central actor, preserving the reply-to so
    /// the central reply routes back to the site's Ask. A real transport does exactly this across
    /// the cluster boundary; the in-process bridge keeps the test free of a multi-node/gRPC setup.
    /// </summary>
    private sealed class BridgeCentralTransport : ICentralTransport
    {
        private readonly IActorRef _central;
        private readonly IActorRef _auditIngest;

        /// <param name="central">Stands in for the central control plane's non-audit RPCs.</param>
        /// <param name="auditIngest">
        /// The central audit-ingest singleton. The two ingest RPCs go straight here, mirroring
        /// <c>CentralControlGrpcService</c>, which Asks the proxy directly rather than relaying
        /// through <c>CentralCommunicationActor</c>.
        /// </param>
        public BridgeCentralTransport(IActorRef central, IActorRef auditIngest)
        {
            _central = central;
            _auditIngest = auditIngest;
        }

        public void SubmitNotification(NotificationSubmit message, IActorRef replyTo) => _central.Tell(message, replyTo);
        public void QueryNotificationStatus(NotificationStatusQuery message, IActorRef replyTo) => _central.Tell(message, replyTo);
        public void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo) => _auditIngest.Tell(message, replyTo);
        public void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo) => _auditIngest.Tell(message, replyTo);
        public void ReconcileSite(ReconcileSiteRequest message, IActorRef replyTo) => _central.Tell(message, replyTo);
        public void ReportSiteHealth(SiteHealthReport message, IActorRef replyTo) => _central.Tell(message, replyTo);
        public void SendHeartbeat(HeartbeatMessage message, IActorRef self) => _central.Tell(message, self);
    }

    /// <summary>
    /// Thread-safe in-memory <see cref="IAuditLogRepository"/>. Only
    /// <see cref="InsertIfNotExistsAsync"/> is exercised by the ingest path; the
    /// rest throw because they are not reachable from this test.
    /// </summary>
    private sealed class InMemoryAuditLogRepository : IAuditLogRepository
    {
        private readonly ConcurrentDictionary<Guid, AuditEvent> _rows = new();

        public IReadOnlyCollection<AuditEvent> Rows => _rows.Values.ToList();

        public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
        {
            ArgumentNullException.ThrowIfNull(evt);
            // First-write-wins idempotency, mirroring the production repository.
            _rows.TryAdd(evt.EventId, evt);
            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<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();
    }

    // C3 (Task 2.5): canonical record via the shared factory; ForwardState is a
    // site-storage-only concern (defaulted to Pending by the SQLite writer), not a
    // field on the canonical record.
    private static AuditEvent NewPendingEvent(Guid id) =>
        ScadaBridgeAuditEventFactory.Create(
            channel: AuditChannel.ApiOutbound,
            kind: AuditKind.ApiCall,
            status: AuditStatus.Delivered,
            eventId: id,
            occurredAtUtc: new DateTime(2026, 5, 21, 9, 0, 0, DateTimeKind.Utc),
            target: "ext-system-1",
            sourceSiteId: "site-1");

    [Fact]
    public async Task SiteAuditEvent_DrainsToCentral_AndFlipsSiteRowToForwarded()
    {
        // ── Central side ──────────────────────────────────────────────────
        // Real AuditLogIngestActor over an in-memory repository. Register row 61: the actor's
        // repository-taking test ctor is gone, so the repository is handed out of a stub
        // provider's scope instead of being cached on an actor field.
        var centralRepo = new InMemoryAuditLogRepository();
        var ingestProvider = new ServiceCollection()
            .AddScoped<IAuditLogRepository>(_ => centralRepo)
            .BuildServiceProvider();
        var ingestActor = Sys.ActorOf(Props.Create(() =>
            new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor(
                (IServiceProvider)ingestProvider,
                NullLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor>.Instance)));

        // Real CentralCommunicationActor. It is NOT on the audit path any more (the
        // central-hosted CentralControlGrpcService Asks the ingest singleton directly), but the
        // bridge transport still routes notifications/health/reconcile through it, so it is
        // constructed exactly as production does. Its periodic site-address refresh resolves an
        // ISiteRepository from this provider; an empty result keeps the refresh a clean no-op.
        var siteRepo = Substitute.For<ISiteRepository>();
        siteRepo.GetAllSitesAsync().Returns(Array.Empty<Site>());
        var centralServices = new ServiceCollection();
        centralServices.AddScoped(_ => siteRepo);
        var centralProvider = centralServices.BuildServiceProvider();

        var centralCommActor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(
            centralProvider,
            Substitute.For<ISiteCommandTransport>())));

        // ── Site side ─────────────────────────────────────────────────────
        // Real SqliteAuditWriter on a file-backed SQLite db (the site hot-path
        // + Pending queue). A temp file so it survives across DI scopes.
        var dbPath = Path.Combine(Path.GetTempPath(), $"auditpush-{Guid.NewGuid():N}.db");
        var writerOptions = Options.Create(new SqliteAuditWriterOptions { DatabasePath = dbPath });
        var nodeIdentity = Substitute.For<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.INodeIdentityProvider>();
        nodeIdentity.NodeName.Returns((string?)null);
        await using var writer = new SqliteAuditWriter(
            writerOptions, NullLogger<SqliteAuditWriter>.Instance, nodeIdentity);

        // Real SiteCommunicationActor. Its site→central transport is the in-process bridge that
        // Tells the central actor (standing in for GrpcCentralTransport dialling CentralControlService).
        var siteCommActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor(
            "site-1",
            new CommunicationOptions(),
            CreateTestProbe().Ref, // deployment-manager proxy is unused here
            null,
            null,
            new BridgeCentralTransport(centralCommActor, ingestActor))));

        // The production site audit push client — the unit under integration.
        var auditClient = new SiteCommunicationAuditClient(
            siteCommActor, TimeSpan.FromSeconds(5));

        // Real SiteAuditTelemetryActor drains the writer's Pending queue and
        // pushes via the client. Fast intervals so the test completes quickly.
        var telemetryOptions = Options.Create(new SiteAuditTelemetryOptions
        {
            BatchSize = 256,
            BusyIntervalSeconds = 1,
            IdleIntervalSeconds = 1,
        });
        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
            writer,
            auditClient,
            telemetryOptions,
            NullLogger<SiteAuditTelemetryActor>.Instance)));

        // ── Act ───────────────────────────────────────────────────────────
        // Write an audit event onto the site SQLite hot-path. It lands Pending.
        var eventId = Guid.NewGuid();
        await writer.WriteAsync(NewPendingEvent(eventId));

        // ── Assert ────────────────────────────────────────────────────────
        // Within ~10s the drain loop pushes the event to central AND flips the
        // site row to Forwarded.
        await AwaitAssertAsync(async () =>
        {
            // Central received and persisted the row.
            Assert.Contains(centralRepo.Rows, r => r.EventId == eventId);

            // The site row reached AuditForwardState.Forwarded specifically —
            // not merely "no longer Pending" (a Reconciled row would also leave
            // ReadPendingAsync, so we assert the positive Forwarded state).
            // C3 (Task 2.5): ForwardState is a site-storage-only column, no longer a
            // field on the canonical record. ReadForwardedAsync only returns rows in
            // the Forwarded state, so a single match here proves the row reached it.
            var forwarded = await writer.ReadForwardedAsync(256, CancellationToken.None);
            Assert.Single(forwarded, r => r.EventId == eventId);
        }, TimeSpan.FromSeconds(10), TimeSpan.FromMilliseconds(250));

        // The central-persisted row carries the central-stamped IngestedAtUtc.
        var ingested = centralRepo.Rows.Single(r => r.EventId == eventId);
        Assert.NotNull(ingested.AsRow().IngestedAtUtc);

        // Cleanup the temp SQLite file.
        try { File.Delete(dbPath); } catch { /* best-effort */ }
    }
}