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;
///
/// Central-side cluster singleton that ingests batches of
/// rows pushed from sites via the
/// IngestAuditEvents 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
/// ) and inserted idempotently
/// via — duplicates
/// are silently swallowed (first-write-wins), whether they repeat inside one
/// packet or across packets.
///
///
///
/// 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 Forwarded.
///
///
/// 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 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.
///
///
/// Two constructors exist for a deliberate reason: the test ctor injects a
/// concrete against a per-test MSSQL fixture
/// (the only way to verify the ingest-timestamp stamp + duplicate-key
/// idempotency end to end), while the production host wiring registers the
/// actor as a cluster singleton and must therefore resolve the repository —
/// which is a scoped EF Core service — from a fresh DI scope per message.
/// Mirroring the Notification Outbox actor's pattern.
///
///
public class AuditLogIngestActor : ReceiveActor
{
///
/// Overall budget for one ingest message's database work, deliberately
/// SHORTER than the gRPC Ask that wraps it.
///
///
/// The path used to stack three identical 30 s budgets — the site's Ask, the
/// central gRPC handler's Ask (SiteStreamGrpcServer.AuditIngestAskTimeout)
/// 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.
///
/// The full ladder is now strictly monotonic end to end:
/// CommunicationOptions.AuditForwardTimeout (35 s, the site-side forward Ask) >
/// SiteStreamGrpcServer.AuditIngestAskTimeout (30 s, the gRPC deadline AND central's
/// Ask of this singleton) > (20 s) >
/// (15 s).
///
///
internal static readonly TimeSpan IngestBudget = TimeSpan.FromSeconds(20);
///
/// Per-statement SQL timeout for the ingest write, strictly inside
/// 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.
///
internal static readonly TimeSpan IngestSqlCommandTimeout = TimeSpan.FromSeconds(15);
///
/// Budget for the per-row fallback loop, granted as a FRESH
/// rather than reusing the batch's.
///
///
/// 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 ,
/// 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.
///
internal static readonly TimeSpan IngestFallbackBudget = TimeSpan.FromSeconds(5);
private readonly IServiceProvider? _serviceProvider;
private readonly IAuditLogRepository? _injectedRepository;
private readonly ILogger _logger;
///
/// Test-mode constructor — injects a concrete repository instance whose
/// lifetime exceeds the test, so the actor reuses the same instance across
/// every message. Used by the MSSQL-backed TestKit fixture.
///
/// Audit log repository instance shared across all messages.
/// Logger for ingest diagnostics.
public AuditLogIngestActor(
IAuditLogRepository repository,
ILogger logger)
{
ArgumentNullException.ThrowIfNull(repository);
ArgumentNullException.ThrowIfNull(logger);
_injectedRepository = repository;
_logger = logger;
ReceiveAsync(OnIngestAsync);
// The single-repository test ctor cannot service the dual-write —
// it has no SiteCalls repo and no DbContext. The handler still
// registers (so callers don't dead-letter) but replies empty so the
// test surface stays explicit about what this ctor supports.
ReceiveAsync(OnCachedTelemetryWithoutDualWriteAsync);
}
///
/// Production constructor — resolves from
/// a fresh DI scope per message because the repository is a scoped EF Core
/// service registered by AddConfigurationDatabase. The actor itself
/// is a long-lived cluster singleton, so it cannot hold a scope across
/// messages.
///
/// Root service provider used to open a fresh scope per message.
/// Logger for ingest diagnostics.
public AuditLogIngestActor(
IServiceProvider serviceProvider,
ILogger logger)
{
ArgumentNullException.ThrowIfNull(serviceProvider);
ArgumentNullException.ThrowIfNull(logger);
_serviceProvider = serviceProvider;
_logger = logger;
ReceiveAsync(OnIngestAsync);
ReceiveAsync(OnCachedTelemetryAsync);
}
///
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(cmd.Events.Count);
// Resolve the repository for the whole batch — one DbContext per
// message, mirroring NotificationOutboxActor. The injected-repository
// mode (test ctor) skips the scope entirely.
// The IAuditRedactor is also resolved from the per-message scope when
// one is available so the row is truncated + redacted before
// InsertIfNotExistsAsync. The single-repository test ctor has no
// service provider — it falls through with no redactor, which preserves
// the small-payload assumptions baked into the existing fixtures.
// Use CreateAsyncScope + await using so scoped EF Core
// services (IAsyncDisposable DbContexts) dispose asynchronously
// without blocking on sync Dispose() of pending connection cleanup.
if (_injectedRepository is not null)
{
await IngestWithRepositoryAsync(_injectedRepository, redactor: null, failureCounter: null, cmd, nowUtc, accepted)
.ConfigureAwait(false);
}
else
{
// 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();
var redactor = scope.ServiceProvider.GetService();
// 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();
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()?.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 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(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);
}
}
}
}
///
/// Dual-write handler. For every 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.
///
///
/// Per-entry isolation — one entry's failed transaction does NOT abort
/// other entries in the batch (each gets its own
///
/// 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.
///
private async Task OnCachedTelemetryAsync(IngestCachedTelemetryCommand cmd)
{
var replyTo = Sender;
var accepted = new List(cmd.Entries.Count);
try
{
await using var scope = _serviceProvider!.CreateAsyncScope();
var auditRepo = scope.ServiceProvider.GetRequiredService();
var siteCallRepo = scope.ServiceProvider.GetRequiredService();
var dbContext = scope.ServiceProvider.GetRequiredService();
// 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();
// Same best-effort central health counter as
// the OnIngestAsync path — null on test composition roots that
// skip the registration.
var failureCounter = scope.ServiceProvider.GetService();
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));
}
///
/// Attempts the whole cached-telemetry packet as ONE transaction: a single
/// set-based audit insert followed by one monotonic SiteCalls upsert
/// per entry. Returns when it committed (and only then
/// appends to ), when the
/// caller should fall back to the per-entry transaction loop.
///
///
/// 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
/// until the commit returns, so a failed attempt
/// leaves the caller's list untouched and the retry starts from a clean slate.
///
private async Task TryIngestCachedBatchAsync(
Microsoft.EntityFrameworkCore.Storage.IExecutionStrategy strategy,
ScadaBridgeDbContext dbContext,
IAuditLogRepository auditRepo,
ISiteCallAuditRepository siteCallRepo,
IAuditRedactor? redactor,
IngestCachedTelemetryCommand cmd,
List 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(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;
}
///
/// Fallback handler installed on the single-repository test ctor — that
/// ctor has no DbContext and no , so
/// it cannot service the dual-write. Logs a warning and replies with an
/// empty ack so callers fall through to their retry path.
///
private Task OnCachedTelemetryWithoutDualWriteAsync(IngestCachedTelemetryCommand cmd)
{
_logger.LogWarning(
"AuditLogIngestActor received IngestCachedTelemetryCommand on the single-repository ctor; dual-write requires the IServiceProvider ctor. Replying with empty ack ({Count} entries).",
cmd.Entries.Count);
Sender.Tell(new IngestCachedTelemetryReply(Array.Empty()));
return Task.CompletedTask;
}
}