ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs

using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.AuditLog.Redaction;
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.ConfigurationDatabase;

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

/// <summary>
/// Central-side cluster singleton that ingests batches of
/// <see cref="AuditEvent"/> rows pushed from sites via the
/// <c>IngestAuditEvents</c> gRPC RPC. Each row is stamped with the central-side
/// ingest timestamp into DetailsJson (there is no promoted IngestedAtUtc
/// column — the value is a DetailsJson field set via
/// <see cref="AuditRowProjection.WithIngestedAtUtc"/>) and inserted idempotently
/// via <see cref="IAuditLogRepository.InsertManyIfNotExistsAsync"/> — duplicates
/// are silently swallowed (first-write-wins), whether they repeat inside one
/// packet or across packets.
/// </summary>
/// <remarks>
/// <para>
/// Idempotency is the contract: a row that already exists at central counts
/// as "accepted" for the purposes of the reply, because the storage state is
/// consistent and the site is free to flip its local row to <c>Forwarded</c>.
/// </para>
/// <para>
/// Audit-write failures must NEVER abort the user-facing action. Each message
/// is written as ONE set-based statement (or one transaction, for the cached
/// dual-write); if that fails the actor retries the same work row-by-row inside
/// per-row try/catch, so a single bad row still cannot cause the rest of the
/// batch to be lost. It also guards scope/repository
/// resolution so a transient DI fault cannot restart the singleton — those
/// catches are what keep this actor alive across handler throws, not the
/// supervisor strategy. The <see cref="SupervisorStrategy"/> override returns
/// the Akka default decider (Restart for most exceptions) and governs children
/// only; this actor has no children today, so the override is a forward-compat
/// placeholder.
/// </para>
/// <para>
/// There is exactly ONE constructor, taking an <see cref="IServiceProvider"/>:
/// the actor is a cluster singleton and the repository is a scoped EF Core
/// service, so it must be resolved from a fresh DI scope per message and never
/// held on a field. Mirroring the Notification Outbox actor's pattern. Tests
/// that need a concrete repository (e.g. the MSSQL-backed fixture verifying the
/// ingest-timestamp stamp and duplicate-key idempotency) supply a stub provider
/// that hands one out, rather than a second constructor that would cache it —
/// see register row 61.
/// </para>
/// </remarks>
public class AuditLogIngestActor : ReceiveActor
{
    /// <summary>
    /// Overall budget for one ingest message's database work, deliberately
    /// SHORTER than the gRPC Ask that wraps it.
    /// </summary>
    /// <remarks>
    /// The path used to stack three identical 30 s budgets — the site's Ask, the
    /// central gRPC handler's Ask (<c>SiteStreamGrpcServer.AuditIngestAskTimeout</c>)
    /// and the ADO.NET command default — so they all expired at the same instant. The
    /// caller therefore learned nothing except "it took 30 s": no partial ack, no
    /// distinction between a slow database and a wedged singleton. Making the
    /// innermost budget strictly smallest means a slow batch is abandoned by the
    /// actor FIRST, with the accepted-so-far ids still replied, while the outer
    /// Asks are still waiting.
    /// <para>
    /// The full ladder is now strictly monotonic end to end:
    /// <c>CommunicationOptions.AuditForwardTimeout</c> (35 s, the site-side forward Ask) &gt;
    /// <c>SiteStreamGrpcServer.AuditIngestAskTimeout</c> (30 s, the gRPC deadline AND central's
    /// Ask of this singleton) &gt; <see cref="IngestBudget"/> (20 s) &gt;
    /// <see cref="IngestSqlCommandTimeout"/> (15 s).
    /// </para>
    /// </remarks>
    internal static readonly TimeSpan IngestBudget = TimeSpan.FromSeconds(20);

    /// <summary>
    /// Per-statement SQL timeout for the ingest write, strictly inside
    /// <see cref="IngestBudget"/> so a single wedged statement surfaces as a SQL
    /// timeout the per-batch catch can log, rather than consuming the whole
    /// actor-level budget and starving the rest of the batch.
    /// </summary>
    internal static readonly TimeSpan IngestSqlCommandTimeout = TimeSpan.FromSeconds(15);

    /// <summary>
    /// Budget for the per-row fallback loop, granted as a FRESH
    /// <see cref="CancellationTokenSource"/> rather than reusing the batch's.
    /// </summary>
    /// <remarks>
    /// The fallback used to run on the batch's own token. When the batch failed
    /// BECAUSE that token expired, every fallback insert was handed an
    /// already-cancelled token and failed instantly: N logged errors, N counter
    /// bumps, zero rows accepted, and the site retried the whole packet anyway.
    /// A fresh, deliberately SHORT budget (shorter than <see cref="IngestBudget"/>,
    /// so the reply still beats the caller's Ask even when the batch consumed its
    /// whole allowance) gives the poison-row isolation the fallback exists for a
    /// real chance to land the good rows.
    /// </remarks>
    internal static readonly TimeSpan IngestFallbackBudget = TimeSpan.FromSeconds(5);

    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<AuditLogIngestActor> _logger;

    /// <summary>
    /// The only constructor — resolves <see cref="IAuditLogRepository"/> from
    /// a fresh DI scope per message because the repository is a scoped EF Core
    /// service registered by <c>AddConfigurationDatabase</c>. The actor itself
    /// is a long-lived cluster singleton, so it cannot hold a scope across
    /// messages.
    ///
    /// <para>
    /// Register row 61: a second, repository-taking "test-mode" constructor used
    /// to cache a scoped repository on a field for the actor's whole lifetime.
    /// Nothing in production selected it, but it left the captive-dependency bug
    /// one call site away and had to be allowlisted in the Host tests'
    /// <c>CaptiveDependencyGuard</c>. It is gone; test fixtures build a stub
    /// <see cref="IServiceProvider"/> instead, which keeps the actor's own
    /// contract "resolve per scope, hold nothing".
    /// </para>
    /// </summary>
    /// <param name="serviceProvider">Root service provider used to open a fresh scope per message.</param>
    /// <param name="logger">Logger for ingest diagnostics.</param>
    public AuditLogIngestActor(
        IServiceProvider serviceProvider,
        ILogger<AuditLogIngestActor> logger)
    {
        ArgumentNullException.ThrowIfNull(serviceProvider);
        ArgumentNullException.ThrowIfNull(logger);

        _serviceProvider = serviceProvider;
        _logger = logger;

        ReceiveAsync<IngestAuditEventsCommand>(OnIngestAsync);
        ReceiveAsync<IngestCachedTelemetryCommand>(OnCachedTelemetryAsync);
    }

    /// <inheritdoc />
    protected override SupervisorStrategy SupervisorStrategy()
    {
        return new OneForOneStrategy(maxNrOfRetries: 0, withinTimeRange: TimeSpan.Zero, decider:
            Akka.Actor.SupervisorStrategy.DefaultDecider);
    }

    private async Task OnIngestAsync(IngestAuditEventsCommand cmd)
    {
        // Sender is captured before the first await — Akka resets Sender
        // between message dispatches, so a post-await Tell would go to
        // DeadLetters.
        var replyTo = Sender;
        var nowUtc = DateTime.UtcNow;
        var accepted = new List<Guid>(cmd.Events.Count);

        // Resolve the repository for the whole batch — one DbContext per
        // message, mirroring NotificationOutboxActor.
        // The IAuditRedactor is also resolved from the per-message scope when
        // one is available so the row is truncated + redacted before
        // InsertIfNotExistsAsync. A composition root that registers no redactor
        // (test fixtures) falls back to the SafeDefault inside
        // IngestWithRepositoryAsync rather than to pass-through.
        // Use CreateAsyncScope + await using so scoped EF Core
        // services (IAsyncDisposable DbContexts) dispose asynchronously
        // without blocking on sync Dispose() of pending connection cleanup.
        //
        // Guard scope-creation + repository resolution in a
        // try/catch, mirroring OnCachedTelemetryAsync. A transient DI /
        // DbContext-factory fault (pooled-context init, SQL-connection
        // exhaustion, a resolution race during host churn) would otherwise
        // propagate out of the ReceiveAsync handler, trip the parent's
        // supervision, and RESTART this central singleton over a transient
        // fault — dropping the captured reply so the site's Ask times out.
        // Best-effort audit must never wedge the singleton: log, optionally
        // bump the failure counter, and still reply with whatever was
        // accepted (empty on an up-front scope-resolution throw) so the
        // site keeps its rows Pending and retries on the next drain.
        try
        {
            await using var scope = _serviceProvider.CreateAsyncScope();
            var repository = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
            var redactor = scope.ServiceProvider.GetService<IAuditRedactor>();
            // Central health counter is best-effort —
            // unregistered (test composition roots) means the per-row catch
            // simply logs without surfacing on the health dashboard.
            var failureCounter = scope.ServiceProvider.GetService<ICentralAuditWriteFailureCounter>();
            await IngestWithRepositoryAsync(repository, redactor, failureCounter, cmd, nowUtc, accepted)
                .ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            // Scope creation or a required-service resolution threw before
            // (or while) processing the batch. Surface a sustained fault on
            // the dashboard if the counter is registered, but never let the
            // throw escape the handler and restart the singleton.
            try { _serviceProvider.GetService<ICentralAuditWriteFailureCounter>()?.Increment(); }
            catch { /* counter must never throw — defence in depth */ }
            _logger.LogError(
                ex,
                "Audit event batch ingest failed before/while resolving the repository scope; replying with {Accepted} accepted row(s). The site keeps unaccepted rows Pending and retries on the next drain.",
                accepted.Count);
        }

        replyTo.Tell(new IngestAuditEventsReply(accepted));
    }

    private async Task IngestWithRepositoryAsync(
        IAuditLogRepository repository,
        IAuditRedactor? redactor,
        ICentralAuditWriteFailureCounter? failureCounter,
        IngestAuditEventsCommand cmd,
        DateTime nowUtc,
        List<Guid> accepted)
    {
        // Stamp the ingest timestamp here, not at the site. Redact BEFORE the
        // IngestedAtUtc stamp so the redacted copy carries the central-side
        // ingest timestamp. The redactor is contract-bound to never throw; a null
        // redactor (test composition root, no IAuditRedactor registered) falls
        // back to the SafeDefault rather than pass-through, so HTTP header
        // redaction always runs. IngestedAtUtc is a DetailsJson field on the
        // canonical record, so stamp it via the projection helper.
        var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
        var projected = new List<AuditEvent>(cmd.Events.Count);
        foreach (var evt in cmd.Events)
        {
            var filtered = safeRedactor.Apply(evt);
            projected.Add(AuditRowProjection.WithIngestedAtUtc(filtered, nowUtc));
        }

        // ONE set-based statement for the whole packet (WP2.2) instead of one
        // IF NOT EXISTS … INSERT round trip per event. A telemetry packet of 200
        // rows used to cost 200 sequential round trips on the central singleton's
        // dispatcher; it now costs two. Idempotency is unchanged — duplicates
        // within the packet collapse first-write-wins, duplicates against
        // committed rows are eliminated by the anti-semi-join, and a concurrent
        // writer degrades to the repository's per-row fallback.
        using var budget = new CancellationTokenSource(IngestBudget);
        try
        {
            await repository
                .InsertManyIfNotExistsAsync(projected, IngestSqlCommandTimeout, budget.Token)
                .ConfigureAwait(false);

            // A batch that returned without throwing means every row is now
            // present — inserted here or already committed by an earlier delivery.
            // Both count as accepted: the storage state is consistent and the site
            // is free to flip its local rows to Forwarded.
            foreach (var evt in cmd.Events)
            {
                accepted.Add(evt.EventId);
            }
        }
        catch (Exception ex)
        {
            // The batch is a throughput optimisation, NOT a change to the
            // failure grain. The documented invariant — "a single bad row cannot
            // cause the rest of the batch to be lost" — is preserved by falling
            // back to the per-row path here, so a poison row still costs only
            // itself. Re-running rows the failed batch may already have committed
            // is safe: every insert is idempotent on EventId.
            //
            // The fallback gets its OWN token. Reusing the batch's meant that a
            // batch which failed because the budget EXPIRED handed the loop an
            // already-cancelled token, so every row failed instantly and the
            // fallback did nothing but multiply the log and counter noise by the
            // row count.
            var budgetExpired = budget.IsCancellationRequested;
            using var fallbackBudget = new CancellationTokenSource(IngestFallbackBudget);

            _logger.LogWarning(ex,
                "Set-based ingest of {Count} audit event(s) failed{BudgetNote}; falling back to per-row inserts (fresh {FallbackSeconds}s budget) so one bad row does not sink the batch.",
                cmd.Events.Count,
                budgetExpired ? " after exhausting the ingest budget" : string.Empty,
                IngestFallbackBudget.TotalSeconds);

            if (budgetExpired)
            {
                // A blown budget is ONE failure event for the batch, not one per
                // row — bump the health counter once here and suppress the
                // per-row bumps below so the dashboard sees a timeout as a
                // timeout rather than as N independent write failures.
                try { failureCounter?.Increment(); }
                catch { /* counter must never throw — defence in depth */ }
            }

            for (var i = 0; i < projected.Count; i++)
            {
                try
                {
                    await repository.InsertIfNotExistsAsync(projected[i], fallbackBudget.Token).ConfigureAwait(false);
                    accepted.Add(cmd.Events[i].EventId);
                }
                catch (Exception rowEx)
                {
                    // Per-row catch — one bad row never sinks the whole batch.
                    // The row stays Pending at the site; the next drain retries.
                    // Bump the central health counter so a sustained insert-throw
                    // failure surfaces on the dashboard — unless the batch already
                    // counted itself as a single budget-exhaustion failure.
                    if (!budgetExpired)
                    {
                        try { failureCounter?.Increment(); }
                        catch { /* counter must never throw — defence in depth */ }
                    }

                    _logger.LogError(rowEx,
                        "Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
                        cmd.Events[i].EventId);
                }
            }
        }
    }

    /// <summary>
    /// Dual-write handler. For every <see cref="CachedTelemetryEntry"/> the
    /// actor opens a fresh MS SQL transaction, inserts the AuditLog row
    /// idempotently AND upserts the SiteCalls row monotonically. Both succeed
    /// or both roll back, so the audit and operational mirrors never drift
    /// mid-row. The IngestedAtUtc stamp is unified between the two rows so a
    /// downstream join lines up cleanly.
    /// </summary>
    /// <remarks>
    /// Per-entry isolation — one entry's failed transaction does NOT abort
    /// other entries in the batch (each gets its own
    /// <see cref="Microsoft.EntityFrameworkCore.RelationalDatabaseFacadeExtensions.BeginTransactionAsync"/>
    /// scope and a try/catch around it). Audit-write failure NEVER aborts the
    /// user-facing action — the site keeps the row Pending and retries on the
    /// next drain.
    /// </remarks>
    private async Task OnCachedTelemetryAsync(IngestCachedTelemetryCommand cmd)
    {
        var replyTo = Sender;
        var accepted = new List<Guid>(cmd.Entries.Count);

        try
        {
            await using var scope = _serviceProvider.CreateAsyncScope();
            var auditRepo = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
            var siteCallRepo = scope.ServiceProvider.GetRequiredService<ISiteCallAuditRepository>();
            var dbContext = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
            // Resolve the redactor for the whole batch from
            // the scope; null = SafeDefault for test composition roots that
            // skip the redactor registration. The redactor is contract-bound to
            // never throw, so we can apply it inside the per-entry try
            // without risking an unbounded blast radius.
            var redactor = scope.ServiceProvider.GetService<IAuditRedactor>();
            // Same best-effort central health counter as
            // the OnIngestAsync path — null on test composition roots that
            // skip the registration.
            var failureCounter = scope.ServiceProvider.GetService<ICentralAuditWriteFailureCounter>();

            var strategy = dbContext.Database.CreateExecutionStrategy();

            // Fast path (WP2.2): the WHOLE packet in ONE transaction — one
            // set-based audit insert plus one single-statement upsert per entry.
            // The predecessor opened a transaction PER entry and issued three
            // statements inside it (IF NOT EXISTS insert, then the two-statement
            // SiteCalls upsert), so a 50-entry packet cost ~200 round trips and 50
            // commits. If anything faults, the per-entry loop below re-runs the
            // packet with its original per-entry isolation, so the documented
            // "one entry's failure does not abort the others" invariant survives —
            // it is simply no longer paid for on the healthy path.
            using var budget = new CancellationTokenSource(IngestBudget);
            if (await TryIngestCachedBatchAsync(
                    strategy, dbContext, auditRepo, siteCallRepo, redactor, cmd, accepted, budget.Token)
                .ConfigureAwait(false))
            {
                replyTo.Tell(new IngestCachedTelemetryReply(accepted));
                return;
            }

            foreach (var entry in cmd.Entries)
            {
                try
                {
                    await strategy.ExecuteAsync(async () =>
                    {
                        await using var tx = await dbContext.Database
                            .BeginTransactionAsync()
                            .ConfigureAwait(false);

                        // Stamp IngestedAtUtc on both rows from a single
                        // central-side instant so a join on the two tables sees
                        // matching timestamps (debugging convenience, not a
                        // correctness invariant).
                        var ingestedAt = DateTime.UtcNow;
                        // Redact the audit half BEFORE the dual-write — only the
                        // AuditLog row's payload columns are redactable; SiteCalls
                        // carries operational state only (status, retry count) and
                        // is left untouched. Null redactor falls back
                        // to SafeDefault so header redaction always runs.
                        // IngestedAtUtc is a DetailsJson field
                        // on the canonical record, so stamp it via the projection helper.
                        var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
                        var filteredAudit = safeRedactor.Apply(entry.Audit);
                        var auditStamped = AuditRowProjection.WithIngestedAtUtc(filteredAudit, ingestedAt);
                        var siteCallStamped = entry.SiteCall with { IngestedAtUtc = ingestedAt };

                        await auditRepo.InsertIfNotExistsAsync(auditStamped)
                            .ConfigureAwait(false);
                        await siteCallRepo.UpsertAsync(siteCallStamped)
                            .ConfigureAwait(false);

                        await tx.CommitAsync().ConfigureAwait(false);
                    }).ConfigureAwait(false);

                    accepted.Add(entry.Audit.EventId);
                }
                catch (Exception ex)
                {
                    // Both rows rolled back via the disposing transaction. The
                    // EventId is NOT added to `accepted` so the site keeps its
                    // row Pending and retries on the next drain. Other entries
                    // in the batch continue with their own transactions.
                    // Bump the central health counter so a
                    // sustained dual-write failure surfaces on the dashboard.
                    try { failureCounter?.Increment(); }
                    catch { /* counter must never throw — defence in depth */ }
                    _logger.LogError(
                        ex,
                        "Combined telemetry dual-write failed for AuditEvent {EventId} / TrackedOperationId {TrackedOpId}; rolled back.",
                        entry.Audit.EventId,
                        entry.SiteCall.TrackedOperationId);
                }
            }
        }
        catch (Exception ex)
        {
            // Resolving the scope itself threw (e.g. DI mis-wiring). Log and
            // reply with whatever we managed to accept (likely empty) — the
            // central singleton MUST stay alive.
            _logger.LogError(
                ex,
                "Combined telemetry batch ingest failed before per-entry processing.");
        }

        replyTo.Tell(new IngestCachedTelemetryReply(accepted));
    }

    /// <summary>
    /// Attempts the whole cached-telemetry packet as ONE transaction: a single
    /// set-based audit insert followed by one monotonic <c>SiteCalls</c> upsert
    /// per entry. Returns <see langword="true"/> when it committed (and only then
    /// appends to <paramref name="accepted"/>), <see langword="false"/> when the
    /// caller should fall back to the per-entry transaction loop.
    /// </summary>
    /// <remarks>
    /// The batch is all-or-nothing by construction — it is a single transaction —
    /// which is why a failure MUST fall back rather than report partial success:
    /// the per-entry loop is what turns one poison entry into one lost entry
    /// instead of a lost packet. Nothing is appended to
    /// <paramref name="accepted"/> until the commit returns, so a failed attempt
    /// leaves the caller's list untouched and the retry starts from a clean slate.
    /// </remarks>
    private async Task<bool> TryIngestCachedBatchAsync(
        Microsoft.EntityFrameworkCore.Storage.IExecutionStrategy strategy,
        ScadaBridgeDbContext dbContext,
        IAuditLogRepository auditRepo,
        ISiteCallAuditRepository siteCallRepo,
        IAuditRedactor? redactor,
        IngestCachedTelemetryCommand cmd,
        List<Guid> accepted,
        CancellationToken ct)
    {
        try
        {
            await strategy.ExecuteAsync(async () =>
            {
                await using var tx = await dbContext.Database
                    .BeginTransactionAsync(ct)
                    .ConfigureAwait(false);

                // One central-side instant for the whole packet so a join across
                // the two tables sees matching timestamps (debugging convenience,
                // not a correctness invariant).
                var ingestedAt = DateTime.UtcNow;
                var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;

                var auditRows = new List<AuditEvent>(cmd.Entries.Count);
                foreach (var entry in cmd.Entries)
                {
                    // Only the AuditLog row's payload columns are redactable;
                    // SiteCalls carries operational state only (status, retry
                    // count) and is left untouched.
                    var filteredAudit = safeRedactor.Apply(entry.Audit);
                    auditRows.Add(AuditRowProjection.WithIngestedAtUtc(filteredAudit, ingestedAt));
                }

                await auditRepo
                    .InsertManyIfNotExistsAsync(auditRows, IngestSqlCommandTimeout, ct)
                    .ConfigureAwait(false);

                foreach (var entry in cmd.Entries)
                {
                    await siteCallRepo
                        .UpsertAsync(entry.SiteCall with { IngestedAtUtc = ingestedAt }, ct)
                        .ConfigureAwait(false);
                }

                await tx.CommitAsync(ct).ConfigureAwait(false);
            }).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            // No health-counter bump here — the per-entry retry is the authority
            // on whether this packet genuinely failed, and double-counting a
            // batch that the fallback then writes successfully would make the
            // dashboard read as a sustained fault during normal contention.
            _logger.LogWarning(
                ex,
                "Batched cached-telemetry dual-write of {Count} entr(ies) failed; falling back to per-entry transactions.",
                cmd.Entries.Count);
            return false;
        }

        foreach (var entry in cmd.Entries)
        {
            accepted.Add(entry.Audit.EventId);
        }

        return true;
    }

}