ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs

using Akka.Actor;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.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.Audit;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;

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

/// <summary>
/// Site-side actor that drains the local SQLite audit queue and pushes Pending
/// rows to central via two parallel transports:
/// <list type="bullet">
///   <item><description><c>IngestAuditEvents</c> for the audit-only path —
///   sync ApiCall/DbWrite, NotifySend, InboundRequest and similar single-row
///   lifecycle events.</description></item>
///   <item><description><c>IngestCachedTelemetry</c> for the combined-telemetry
///   path — cached-call lifecycle rows (<c>CachedSubmit</c>,
///   <c>ApiCallCached</c>/<c>DbWriteCached</c>, <c>CachedResolve</c>) joined
///   with the matching <c>OperationTracking</c> row, written at central as a
///   single dual-write transaction (AuditLog + SiteCalls).</description></item>
/// </list>
/// </summary>
/// <remarks>
/// <para>
/// The drain self-ticks via two private messages — <c>Drain</c> for the
/// audit-only path and <c>CachedDrain</c> for the combined path — each
/// scheduled independently. Cadence is options-driven:
/// <c>BusyIntervalSeconds</c> when the previous drain found rows (or faulted —
/// we want quick recovery), <c>IdleIntervalSeconds</c> when the queue was empty.
/// The two drains share the same cadence configuration but advance their own
/// timers so a stall on one path does not block the other.
/// </para>
/// <para>
/// Collaborators are injected as interfaces (<see cref="ISiteAuditQueue"/>,
/// <see cref="ISiteStreamAuditClient"/>, optional
/// <see cref="IOperationTrackingStore"/>) so unit tests substitute with
/// NSubstitute and never touch real SQLite or gRPC. The
/// <see cref="IOperationTrackingStore"/> is optional — central composition
/// roots and tests that don't exercise the cached path can leave it null, in
/// which case the cached-drain scheduler is never armed.
/// </para>
/// <para>
/// Audit-write paths must be fail-safe — a thrown
/// exception inside the actor MUST NOT crash it. Both Drain handlers wrap
/// their pipelines in a top-level try/catch that logs and re-schedules; the
/// actor's <see cref="SupervisorStrategy"/> defaults to
/// <see cref="Akka.Actor.SupervisorStrategy.DefaultStrategy"/>'s Restart for
/// child actors — but this actor has no children, so the catch is what
/// matters.
/// </para>
/// <para>
/// Wires the previously-unreachable combined-telemetry transport.
/// Prior to this the cached audit rows flowed through the audit-only drain via
/// <c>IngestAuditEventsAsync</c> and the central <c>OnCachedTelemetryAsync</c>
/// dual-write handler was dead production code; the operational <c>SiteCalls</c>
/// half was never sent to central.
/// </para>
/// </remarks>
public class SiteAuditTelemetryActor : ReceiveActor
{
    private readonly ISiteAuditQueue _queue;
    private readonly ISiteStreamAuditClient _client;
    private readonly IOperationTrackingStore? _trackingStore;
    private readonly SiteAuditTelemetryOptions _options;
    private readonly ILogger<SiteAuditTelemetryActor> _logger;
    // Captured at construction (both are thread-safe immutable handles) because
    // ScheduleNext/ScheduleNextCached run from the drains' finally blocks, whose
    // ConfigureAwait(false) continuations complete on pool threads with no
    // active ActorContext — reading Context/Self there either throws
    // NotSupportedException or, worse, silently resolves a STALE cell left in
    // the thread-static slot and re-arms the tick at the wrong actor
    // (known-issues/2026-07-20-localdb-disk-io-error-under-load.md §8).
    private readonly IScheduler _scheduler;
    private readonly IActorRef _self;
    private ICancelable? _pendingTick;
    private ICancelable? _pendingCachedTick;
    // Per-actor lifecycle CTS so an in-flight drain (queue read,
    // gRPC push, mark-forwarded write) is actually cancelled when the actor is
    // stopped — without it, a stuck IngestAuditEventsAsync would hold the
    // continuation through CoordinatedShutdown's actor-system terminate window.
    // Cancelled in PostStop; never reset (the actor is single-lifetime).
    // The same CTS gates the cached-drain pipeline (queue read + tracking
    // lookup + gRPC push) so both paths observe shutdown cooperatively.
    private readonly CancellationTokenSource _lifecycleCts = new();

    /// <summary>Initializes the actor with its drain queue, gRPC client, options, and logger.</summary>
    /// <param name="queue">The site-local SQLite audit queue to drain.</param>
    /// <param name="client">The gRPC client used to push audit events to central.</param>
    /// <param name="options">Telemetry options controlling drain intervals and batch size.</param>
    /// <param name="logger">Logger instance.</param>
    /// <param name="trackingStore">
    /// Optional site-local operation tracking store. When supplied the actor
    /// runs the combined-telemetry cached-drain in parallel with the audit-only
    /// drain; when null (central composition roots, tests that don't exercise
    /// cached calls) the cached scheduler is never armed and only the
    /// audit-only drain runs.
    /// </param>
    public SiteAuditTelemetryActor(
        ISiteAuditQueue queue,
        ISiteStreamAuditClient client,
        IOptions<SiteAuditTelemetryOptions> options,
        ILogger<SiteAuditTelemetryActor> logger,
        IOperationTrackingStore? trackingStore = null)
    {
        ArgumentNullException.ThrowIfNull(queue);
        ArgumentNullException.ThrowIfNull(client);
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(logger);

        _queue = queue;
        _client = client;
        _options = options.Value;
        _logger = logger;
        _trackingStore = trackingStore;
        _scheduler = Context.System.Scheduler;
        _self = Self;

        ReceiveAsync<Drain>(_ => OnDrainAsync());
        ReceiveAsync<CachedDrain>(_ => OnCachedDrainAsync());
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        base.PreStart();
        // Initial ticks fire on the busy interval so both drains start polling
        // soon after host startup. A subsequent empty drain will move to the
        // idle interval naturally.
        ScheduleNext(TimeSpan.FromSeconds(_options.BusyIntervalSeconds));
        if (_trackingStore is not null)
        {
            ScheduleNextCached(TimeSpan.FromSeconds(_options.BusyIntervalSeconds));
        }
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        _pendingTick?.Cancel();
        _pendingCachedTick?.Cancel();
        // Cancel any in-flight drain so a stuck queue read or
        // gRPC push does not hold the continuation past actor stop.
        try
        {
            _lifecycleCts.Cancel();
        }
        catch (ObjectDisposedException)
        {
            // PostStop may run after a prior Dispose path — benign.
        }
        _lifecycleCts.Dispose();
        base.PostStop();
    }

    private async Task OnDrainAsync()
    {
        var nextDelay = TimeSpan.FromSeconds(_options.BusyIntervalSeconds);
        // Route every async dependency call through the
        // per-actor lifecycle token so PostStop cancellation actually
        // propagates into the queue read, the gRPC push, and the
        // mark-forwarded write. OperationCanceledException is swallowed by
        // the catch-all below.
        var ct = _lifecycleCts.Token;
        try
        {
            var pending = await _queue.ReadPendingAsync(_options.BatchSize, ct)
                .ConfigureAwait(false);
            if (pending.Count == 0)
            {
                // No rows — settle into the idle cadence until the next write
                // bumps us back into the busy cadence.
                nextDelay = TimeSpan.FromSeconds(_options.IdleIntervalSeconds);
                return;
            }

            var batch = BuildBatch(pending);

            IngestAck ack;
            try
            {
                ack = await _client.IngestAuditEventsAsync(batch, ct)
                    .ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                // gRPC fault — leave the rows in Pending so the next drain
                // retries — "On gRPC exception (any), log
                // Warning, schedule next Drain in BusyIntervalSeconds."
                _logger.LogWarning(ex,
                    "IngestAuditEvents push failed for {Count} pending events; will retry next drain.",
                    pending.Count);
                return;
            }

            var acceptedIds = ParseAcceptedIds(ack);
            if (acceptedIds.Count > 0)
            {
                await _queue.MarkForwardedAsync(acceptedIds, ct)
                    .ConfigureAwait(false);
            }
        }
        catch (Exception ex)
        {
            // Catch-all so a SQLite hiccup or mapper bug never crashes the
            // actor. The next tick is still scheduled in the finally block.
            _logger.LogError(ex,
                "Unexpected error during audit-log telemetry drain (sqlite {SqliteError}).",
                SqliteErrorCodes.Describe(ex));
        }
        finally
        {
            // If the actor is already shutting down, do not
            // arm another tick — the scheduler would fire after PostStop and
            // the message would land in dead letters.
            if (!_lifecycleCts.IsCancellationRequested)
            {
                ScheduleNext(nextDelay);
            }
        }
    }

    /// <summary>
    /// Combined-telemetry drain. Reads cached-lifecycle audit
    /// rows, joins each with the matching <see cref="IOperationTrackingStore"/>
    /// snapshot, builds a <see cref="CachedTelemetryBatch"/>, and pushes via
    /// <see cref="ISiteStreamAuditClient.IngestCachedTelemetryAsync"/>. Rows
    /// whose tracking snapshot is missing (race with retention purge / late
    /// audit row) are logged + skipped — the operational half will be
    /// re-emitted on the next lifecycle event, and the audit row stays
    /// <see cref="Commons.Types.Enums.AuditForwardState.Pending"/> so a later
    /// drain (or reconciliation pull) can revisit it.
    /// </summary>
    private async Task OnCachedDrainAsync()
    {
        var nextDelay = TimeSpan.FromSeconds(_options.BusyIntervalSeconds);
        var ct = _lifecycleCts.Token;
        try
        {
            // _trackingStore is non-null by construction here — the cached
            // scheduler is only armed when it was supplied (see PreStart).
            // Defensive check kept for clarity and to silence the compiler's
            // null-flow analysis.
            if (_trackingStore is null)
            {
                return;
            }

            var pending = await _queue
                .ReadPendingCachedTelemetryAsync(_options.BatchSize, ct)
                .ConfigureAwait(false);
            if (pending.Count == 0)
            {
                nextDelay = TimeSpan.FromSeconds(_options.IdleIntervalSeconds);
                return;
            }

            var batch = new CachedTelemetryBatch();
            var emittedEventIds = new List<Guid>(pending.Count);
            // Rows whose operational half can never be built. They are marked
            // Forwarded so they LEAVE this queue — see AbandonUnresolvableAsync
            // for why that is safe and why leaving them Pending is not.
            var abandonedEventIds = new List<Guid>();
            // Counted, not logged per row: a whole batch can fail identically
            // (tracking store down, tracking retention elapsed), and one line
            // per row per drain is what turned this into a log flood.
            var deferredNoSnapshot = 0;
            var lookupFailures = 0;
            Exception? firstLookupFailure = null;
            var graceSeconds = Math.Max(0, _options.CachedTrackingGraceSeconds);
            var abandonBefore = DateTime.UtcNow - TimeSpan.FromSeconds(graceSeconds);

            foreach (var auditRow in pending)
            {
                if (auditRow.CorrelationId is null)
                {
                    // CorrelationId carries the TrackedOperationId for cached
                    // rows — see CachedCallLifecycleBridge.BuildPacket. Without
                    // it there is nothing to look up, ever, so this row is
                    // abandoned immediately rather than after the grace period.
                    abandonedEventIds.Add(auditRow.EventId);
                    continue;
                }

                TrackingStatusSnapshot? snapshot;
                try
                {
                    snapshot = await _trackingStore
                        .GetStatusAsync(new TrackedOperationId(auditRow.CorrelationId.Value), ct)
                        .ConfigureAwait(false);
                }
                catch (Exception ex)
                {
                    // A tracking-store throw must NOT abort the rest of the
                    // batch — the audit half is best-effort. A throw is a
                    // STORE fault (locked, corrupt) rather than a verdict about
                    // this row, so the row is never abandoned on this path: it
                    // stays Pending and retries once the store recovers.
                    lookupFailures++;
                    firstLookupFailure ??= ex;
                    continue;
                }

                if (snapshot is null)
                {
                    // No tracking row. Within the grace window this is the
                    // ordinary write race (the audit row landed first), so
                    // retry. Past it the snapshot is gone for good — tracking
                    // retention elapsed, or the two stores were reset
                    // independently — and retrying forever would wedge the
                    // queue behind these rows.
                    if (auditRow.OccurredAtUtc <= abandonBefore)
                    {
                        abandonedEventIds.Add(auditRow.EventId);
                    }
                    else
                    {
                        deferredNoSnapshot++;
                    }

                    continue;
                }

                var packet = BuildCachedPacket(auditRow, snapshot);
                batch.Packets.Add(packet);
                emittedEventIds.Add(auditRow.EventId);
            }

            if (lookupFailures > 0)
            {
                _logger.LogWarning(firstLookupFailure,
                    "Cached-telemetry drain: tracking lookup failed for {Count} of {Total} rows " +
                    "(sqlite {SqliteError}); they stay Pending and retry next drain. First failure attached.",
                    lookupFailures, pending.Count, SqliteErrorCodes.Describe(firstLookupFailure!));
            }

            if (deferredNoSnapshot > 0)
            {
                _logger.LogDebug(
                    "Cached-telemetry drain: {Count} row(s) have no tracking snapshot yet and are inside the " +
                    "{Grace}s grace window; retrying next drain.",
                    deferredNoSnapshot, graceSeconds);
            }

            if (abandonedEventIds.Count > 0)
            {
                await AbandonUnresolvableAsync(abandonedEventIds, graceSeconds, ct)
                    .ConfigureAwait(false);
            }

            if (batch.Packets.Count == 0)
            {
                // Nothing resolvable in this read. Any permanently-unresolvable
                // rows have just been marked Forwarded above, so the next drain
                // sees past them rather than re-reading the same head of queue.
                return;
            }

            IngestAck ack;
            try
            {
                ack = await _client.IngestCachedTelemetryAsync(batch, ct)
                    .ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "IngestCachedTelemetry push failed for {Count} cached events; will retry next drain.",
                    batch.Packets.Count);
                return;
            }

            var acceptedIds = ParseAcceptedIds(ack);
            if (acceptedIds.Count > 0)
            {
                await _queue.MarkForwardedAsync(acceptedIds, ct)
                    .ConfigureAwait(false);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "Unexpected error during cached-telemetry drain (sqlite {SqliteError}).",
                SqliteErrorCodes.Describe(ex));
        }
        finally
        {
            if (!_lifecycleCts.IsCancellationRequested && _trackingStore is not null)
            {
                ScheduleNextCached(nextDelay);
            }
        }
    }

    /// <summary>
    /// Marks cached rows whose operational half can never be built as
    /// Forwarded, so they leave the cached-drain queue.
    /// </summary>
    /// <remarks>
    /// <para>
    /// <b>Why they must leave.</b> Leaving an unresolvable row Pending — the
    /// previous behaviour — is not a harmless skip. The queue is read
    /// oldest-first with a fixed <c>BatchSize</c>, so once a batch's worth of
    /// unresolvable rows collects at the head, every drain re-reads exactly
    /// those rows, fails identically, and never sees the newer rows behind
    /// them. The cached-telemetry path stalls permanently, and each pass logs
    /// once per row (measured at ~2 800 warnings/minute on a rig).
    /// </para>
    /// <para>
    /// <b>Why Forwarded is the honest state.</b> Its role in this state machine
    /// is "no longer owed by the drain, still eligible for reconciliation" —
    /// which is exactly the situation here. The audit half is NOT dropped:
    /// <c>ReadPendingSinceAsync</c> covers Forwarded rows as well as Pending
    /// ones and central dedups on EventId, so the reconciliation pull still
    /// delivers them. What is genuinely lost is the operational
    /// (<c>SiteCalls</c>) half — and that is unrecoverable regardless, because
    /// the tracking row it would have been built from no longer exists.
    /// </para>
    /// <para>
    /// A failure to mark is swallowed: the rows simply stay Pending and the
    /// next drain retries. Escalating here would take down the audit drain over
    /// a best-effort cleanup.
    /// </para>
    /// </remarks>
    private async Task AbandonUnresolvableAsync(
        IReadOnlyList<Guid> eventIds, int graceSeconds, CancellationToken ct)
    {
        try
        {
            await _queue.MarkForwardedAsync(eventIds, ct).ConfigureAwait(false);
            _logger.LogWarning(
                "Cached-telemetry drain: abandoned the operational half of {Count} row(s) with no " +
                "resolvable tracking snapshot after {Grace}s; marked Forwarded so they no longer block " +
                "the queue. The audit half still reaches central via the reconciliation pull. This is " +
                "expected after tracking retention elapses or the tracking store is reset independently " +
                "of the audit store.",
                eventIds.Count, graceSeconds);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Cached-telemetry drain: could not mark {Count} unresolvable row(s) Forwarded " +
                "(sqlite {SqliteError}); they stay Pending and will be retried.",
                eventIds.Count, SqliteErrorCodes.Describe(ex));
        }
    }

    private static AuditEventBatch BuildBatch(IReadOnlyList<AuditEvent> events)
    {
        var batch = new AuditEventBatch();
        foreach (var e in events)
        {
            batch.Events.Add(AuditEventDtoMapper.ToDto(e));
        }
        return batch;
    }

    /// <summary>
    /// Build the combined wire packet from one cached audit row
    /// + its matching operational tracking snapshot. The operational state
    /// reflects the latest tracking row at emission time (not the per-event
    /// status the audit row implies) because central's <c>SiteCalls</c>
    /// upsert is monotonic — it never rolls back. The audit row preserves
    /// per-event lifecycle granularity for the audit trail.
    /// </summary>
    private static CachedTelemetryPacket BuildCachedPacket(
        AuditEvent auditRow, TrackingStatusSnapshot snapshot)
    {
        // SourceSiteId + Channel ride inside the canonical record's
        // DetailsJson — decompose to read them.
        var audit = AuditRowProjection.Decompose(auditRow);
        var sourceSite = audit.SourceSiteId ?? string.Empty;
        // Channel string form mirrors the AuditChannel-to-string convention used
        // by SiteCallOperational + CachedCallLifecycleBridge.BuildPacket.
        var channelString = audit.Channel.ToString();
        var target = auditRow.Target ?? snapshot.TargetSummary ?? string.Empty;

        var operationalDto = new SiteCallOperationalDto
        {
            TrackedOperationId = snapshot.Id.Value.ToString("D"),
            Channel = channelString,
            Target = target,
            SourceSite = sourceSite,
            SourceNode = snapshot.SourceNode ?? string.Empty,
            Status = snapshot.Status,
            RetryCount = snapshot.RetryCount,
            LastError = snapshot.LastError ?? string.Empty,
            CreatedAtUtc = Timestamp.FromDateTime(EnsureUtc(snapshot.CreatedAtUtc)),
            UpdatedAtUtc = Timestamp.FromDateTime(EnsureUtc(snapshot.UpdatedAtUtc)),
        };
        if (snapshot.HttpStatus.HasValue)
        {
            operationalDto.HttpStatus = snapshot.HttpStatus.Value;
        }
        if (snapshot.TerminalAtUtc.HasValue)
        {
            operationalDto.TerminalAtUtc =
                Timestamp.FromDateTime(EnsureUtc(snapshot.TerminalAtUtc.Value));
        }

        return new CachedTelemetryPacket
        {
            AuditEvent = AuditEventDtoMapper.ToDto(auditRow),
            Operational = operationalDto,
        };
    }

    private static DateTime EnsureUtc(DateTime value) =>
        value.Kind == DateTimeKind.Utc
            ? value
            : DateTime.SpecifyKind(value.ToUniversalTime(), DateTimeKind.Utc);

    private static IReadOnlyList<Guid> ParseAcceptedIds(IngestAck ack)
    {
        if (ack.AcceptedEventIds.Count == 0)
        {
            return Array.Empty<Guid>();
        }

        var list = new List<Guid>(ack.AcceptedEventIds.Count);
        foreach (var raw in ack.AcceptedEventIds)
        {
            if (Guid.TryParse(raw, out var id))
            {
                list.Add(id);
            }
            // Malformed ids are ignored — central should never emit them, but
            // we refuse to crash the actor over a bad string.
        }
        return list;
    }

    // Must stay off Context/Self: called from off-context continuations — see
    // the _scheduler/_self field comment.
    private void ScheduleNext(TimeSpan delay)
    {
        _pendingTick?.Cancel();
        _pendingTick = _scheduler.ScheduleTellOnceCancelable(
            delay,
            _self,
            Drain.Instance,
            _self);
    }

    private void ScheduleNextCached(TimeSpan delay)
    {
        _pendingCachedTick?.Cancel();
        _pendingCachedTick = _scheduler.ScheduleTellOnceCancelable(
            delay,
            _self,
            CachedDrain.Instance,
            _self);
    }

    /// <summary>Self-tick message that triggers an audit-only drain cycle.</summary>
    private sealed class Drain
    {
        public static readonly Drain Instance = new();
        private Drain() { }
    }

    /// <summary>
    /// Self-tick message that triggers a combined-telemetry drain cycle.
    /// Introduced alongside the cached-drain to keep the two
    /// paths' cadences independent — a stall on one does not block the other.
    /// </summary>
    private sealed class CachedDrain
    {
        public static readonly CachedDrain Instance = new();
        private CachedDrain() { }
    }
}