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;
///
/// Site-side actor that drains the local SQLite audit queue and pushes Pending
/// rows to central via two parallel transports:
///
/// - IngestAuditEvents for the audit-only path —
/// sync ApiCall/DbWrite, NotifySend, InboundRequest and similar single-row
/// lifecycle events.
/// - IngestCachedTelemetry for the combined-telemetry
/// path — cached-call lifecycle rows (CachedSubmit,
/// ApiCallCached/DbWriteCached, CachedResolve) joined
/// with the matching OperationTracking row, written at central as a
/// single dual-write transaction (AuditLog + SiteCalls).
///
///
///
///
/// The drain self-ticks via two private messages — Drain for the
/// audit-only path and CachedDrain for the combined path — each
/// scheduled independently. Cadence is options-driven:
/// BusyIntervalSeconds when the previous drain found rows (or faulted —
/// we want quick recovery), IdleIntervalSeconds 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.
///
///
/// Collaborators are injected as interfaces (,
/// , optional
/// ) so unit tests substitute with
/// NSubstitute and never touch real SQLite or gRPC. The
/// 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.
///
///
/// 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 defaults to
/// 's Restart for
/// child actors — but this actor has no children, so the catch is what
/// matters.
///
///
/// Wires the previously-unreachable combined-telemetry transport.
/// Prior to this the cached audit rows flowed through the audit-only drain via
/// IngestAuditEventsAsync and the central OnCachedTelemetryAsync
/// dual-write handler was dead production code; the operational SiteCalls
/// half was never sent to central.
///
///
public class SiteAuditTelemetryActor : ReceiveActor
{
private readonly ISiteAuditQueue _queue;
private readonly ISiteStreamAuditClient _client;
private readonly IOperationTrackingStore? _trackingStore;
private readonly SiteAuditTelemetryOptions _options;
private readonly ILogger _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();
/// Initializes the actor with its drain queue, gRPC client, options, and logger.
/// The site-local SQLite audit queue to drain.
/// The gRPC client used to push audit events to central.
/// Telemetry options controlling drain intervals and batch size.
/// Logger instance.
///
/// 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.
///
public SiteAuditTelemetryActor(
ISiteAuditQueue queue,
ISiteStreamAuditClient client,
IOptions options,
ILogger 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(_ => OnDrainAsync());
ReceiveAsync(_ => OnCachedDrainAsync());
}
///
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));
}
}
///
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);
}
}
}
///
/// Combined-telemetry drain. Reads cached-lifecycle audit
/// rows, joins each with the matching
/// snapshot, builds a , and pushes via
/// . 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
/// so a later
/// drain (or reconciliation pull) can revisit it.
///
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(pending.Count);
foreach (var auditRow in pending)
{
if (auditRow.CorrelationId is null)
{
// CorrelationId carries the TrackedOperationId for cached
// rows — see CachedCallLifecycleBridge.BuildPacket. Without
// it we can't look up the tracking row; log + skip so the
// bad row doesn't block the rest of the batch. The audit
// row stays Pending (still not in emittedEventIds) and
// central reconciliation will pick it up.
_logger.LogWarning(
"Cached-telemetry drain: audit row {EventId} ({Action}) has no CorrelationId; skipping.",
auditRow.EventId, auditRow.Action);
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. Log and skip
// this row; it stays Pending for the next drain.
_logger.LogWarning(ex,
"Cached-telemetry drain: tracking lookup threw for {EventId} (TrackedOperationId {Tid}, sqlite {SqliteError}); skipping.",
auditRow.EventId, auditRow.CorrelationId, SqliteErrorCodes.Describe(ex));
continue;
}
if (snapshot is null)
{
// No tracking row — possible if the audit row is older
// than the tracking retention window, or the tracking
// store was reset. The audit half remains valid and will
// be picked up by central reconciliation; skip the
// combined push for this row.
_logger.LogWarning(
"Cached-telemetry drain: no tracking snapshot for {EventId} (TrackedOperationId {Tid}); skipping.",
auditRow.EventId, auditRow.CorrelationId);
continue;
}
var packet = BuildCachedPacket(auditRow, snapshot);
batch.Packets.Add(packet);
emittedEventIds.Add(auditRow.EventId);
}
if (batch.Packets.Count == 0)
{
// Every row in this read was skipped (no CorrelationId / no
// tracking snapshot). Leave them Pending and try again next
// drain — the underlying race normally resolves on its own.
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);
}
}
}
private static AuditEventBatch BuildBatch(IReadOnlyList events)
{
var batch = new AuditEventBatch();
foreach (var e in events)
{
batch.Events.Add(AuditEventDtoMapper.ToDto(e));
}
return batch;
}
///
/// 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 SiteCalls
/// upsert is monotonic — it never rolls back. The audit row preserves
/// per-event lifecycle granularity for the audit trail.
///
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 ParseAcceptedIds(IngestAck ack)
{
if (ack.AcceptedEventIds.Count == 0)
{
return Array.Empty();
}
var list = new List(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);
}
/// Self-tick message that triggers an audit-only drain cycle.
private sealed class Drain
{
public static readonly Drain Instance = new();
private Drain() { }
}
///
/// 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.
///
private sealed class CachedDrain
{
public static readonly CachedDrain Instance = new();
private CachedDrain() { }
}
}