using System.Threading.Channels;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.LocalDb;
namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging;
///
/// Records operational events into the consolidated site database (LocalDb Phase 1).
/// Only the active node generates events, but site_events is a replicated table:
/// when the pair has a peer configured, the standby holds the same history and a failover
/// no longer starts a fresh log. With no peer configured the database is simply local —
/// replication is opt-in via LocalDb:Replication:PeerAddress.
///
///
///
/// A single is owned here and is NOT thread-safe.
/// All access — recording, querying, purging — must be funnelled through
/// , which serialises callers on a shared lock.
///
///
/// Event recording is offloaded to a dedicated background writer thread (fed by a
/// bounded ; capacity ,
/// default 10 000, overflow ).
/// only validates its arguments and enqueues, so callers —
/// typically Akka actor threads on hot paths — never block on disk I/O or on
/// contention for the write lock. The returned completes once the
/// event is durably persisted and faults if the write fails. When a queued
/// event is evicted to make room for a newer one, that event's Task
/// is faulted with and
/// is incremented so the drop is observable.
///
///
public class SiteEventLogger : ISiteEventLogger, IDisposable
{
///
/// Maximum number of queued events committed together in a single transaction
/// per writer-loop drain. Mirrors SqliteAuditWriter's batch size
/// (ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs,
/// ProcessWriteQueueAsync/FlushBatch) — see the comment on
/// for why one transaction per drain
/// replaces the previous one-transaction-per-event shape.
///
internal const int WriteBatchSize = 256;
private readonly SqliteConnection _connection;
private readonly ILocalDb _localDb;
private readonly ILogger _logger;
private readonly object _writeLock = new();
private readonly Channel _writeQueue;
private readonly Task _writerLoop;
private long _failedWriteCount;
private long _committedBatchCount;
private bool _disposed;
///
/// Initializes the event logger over the consolidated site database and starts the
/// background writer loop.
///
/// Site event log configuration (retention settings, queue capacity).
/// Logger for write-failure diagnostics.
///
/// The consolidated site database. The connection it hands out is already open and
/// carries the zb_hlc_next() UDF that site_events' capture triggers
/// call. This logger owns the connection for its lifetime and disposes it.
///
///
/// The former connectionStringOverride seam is deliberately gone rather than
/// preserved. site_events is a replicated table, so a raw connection lacking
/// the UDF would fail closed on every insert — an override would only let a caller
/// build a logger that cannot write. SiteEventLogOptions.DatabasePath is
/// likewise no longer read: the location is LocalDb:Path.
///
public SiteEventLogger(
IOptions options,
ILogger logger,
ILocalDb localDb)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentNullException.ThrowIfNull(localDb);
_logger = logger;
_localDb = localDb;
// Already open — CreateConnection returns a live, pragma-configured,
// UDF-registered connection. Calling Open() on it again would throw.
_connection = localDb.CreateConnection();
InitializeSchema();
// Bounded queue with DropOldest preserves the
// "callers never block" guarantee while putting an
// upper bound on memory under sustained writer slowness. Drops are
// observable — itemDropped faults the evicted Task and increments
// FailedWriteCount.
var capacity = Math.Max(1, options.Value.WriteQueueCapacity);
_writeQueue = Channel.CreateBounded(
new BoundedChannelOptions(capacity)
{
SingleReader = true,
SingleWriter = false,
FullMode = BoundedChannelFullMode.DropOldest,
},
itemDropped: dropped =>
{
Interlocked.Increment(ref _failedWriteCount);
dropped.Completion.TrySetException(
new InvalidOperationException(
$"Event was dropped because the write queue exceeded its bounded capacity ({capacity})."));
});
_writerLoop = Task.Run(ProcessWriteQueueAsync);
}
///
public long FailedWriteCount => Interlocked.Read(ref _failedWriteCount);
///
/// Number of write-batch transactions successfully committed since startup. Test-only
/// observability hook (not part of ) proving the
/// "drain up to queued events into one transaction" shape
/// described on — production code does not read it.
///
internal long CommittedBatchCount => Interlocked.Read(ref _committedBatchCount);
///
/// Runs against the shared connection while holding the
/// write lock, so purge / query / record callers on different threads never use
/// the non-thread-safe concurrently.
/// Returns without invoking the action if the logger has
/// been disposed.
///
/// The action to run against the shared connection.
/// if the action was executed; if the logger is disposed.
internal bool WithConnection(Action action)
{
ArgumentNullException.ThrowIfNull(action);
lock (_writeLock)
{
if (_disposed) return false;
action(_connection);
return true;
}
}
///
/// Runs on a short-lived, dedicated READ connection that is
/// NOT serialised on the write lock. site_events lives in a WAL database, so a
/// reader never blocks the single writer (and vice versa): a slow central event-log
/// query (full-scan keyword search at the storage cap) can no longer stall the writer
/// loop's FlushBatch and overflow the bounded write queue (review 2026-10-21
/// 02/P1). The same pattern OperationTrackingStore.GetStatusAsync uses.
///
/// The return type of the function.
/// The function to run against the read connection.
/// The value returned by .
internal T WithReadConnection(Func func)
{
ArgumentNullException.ThrowIfNull(func);
ObjectDisposedException.ThrowIf(_disposed, this);
using var readConnection = _localDb.CreateConnection();
return func(readConnection);
}
///
/// Runs against the shared connection while holding the
/// write lock. Throws if the logger has
/// been disposed (callers that need a result cannot proceed without the database).
///
/// The return type of the function.
/// The function to run against the shared connection.
/// The value returned by .
internal T WithConnection(Func func)
{
ArgumentNullException.ThrowIfNull(func);
lock (_writeLock)
{
ObjectDisposedException.ThrowIf(_disposed, this);
return func(_connection);
}
}
// Schema lives in SiteEventLogSchema so the Host's AddZbLocalDb onReady callback
// can create this table in the consolidated site database before RegisterReplicated
// installs its capture triggers. Applying it here too keeps a directly-constructed
// logger (tests, tooling) self-sufficient; the DDL is idempotent.
private void InitializeSchema() => SiteEventLogSchema.Apply(_connection);
///
/// Closed set of allowed severities. Case-sensitive to
/// match the SQLite default BINARY collation used by the query filter —
/// a row stored as "error" would be invisible to a query filtering on
/// "Error", so the contract on the way in must match the contract on
/// the way out.
///
private static readonly HashSet AllowedSeverities =
new(StringComparer.Ordinal) { "Info", "Warning", "Error" };
///
public Task LogEventAsync(
string eventType,
string severity,
string? instanceId,
string source,
string message,
string? details = null)
{
ArgumentException.ThrowIfNullOrWhiteSpace(eventType);
ArgumentException.ThrowIfNullOrWhiteSpace(severity);
ArgumentException.ThrowIfNullOrWhiteSpace(source);
ArgumentException.ThrowIfNullOrWhiteSpace(message);
// Reject unknown severities so the query-time filter
// (case-sensitive BINARY collation) and the documented enum stay in sync.
if (!AllowedSeverities.Contains(severity))
{
throw new ArgumentException(
$"Severity '{severity}' is not one of the allowed values: Info, Warning, Error.",
nameof(severity));
}
// The id is minted here, by the application, rather than by SQLite. Under
// LocalDb replication the site pair converges by last-writer-wins on the
// primary key, so a server-minted AUTOINCREMENT would have both nodes
// independently issuing id 1, 2, 3... for unrelated events and silently
// overwriting each other on sync. A GUID makes the event log a pure union.
var pending = new PendingEvent(
Guid.NewGuid().ToString("N"),
DateTimeOffset.UtcNow.ToString("o"),
eventType,
severity,
instanceId,
source,
message,
details);
// Enqueue only — the actual SQLite write happens on the background writer
// thread so the caller (an Akka actor thread on a hot path) never blocks
// on disk I/O or on contention for the write lock.
if (!_writeQueue.Writer.TryWrite(pending))
{
// The channel is bounded with DropOldest, so TryWrite only returns false
// once the channel is completed (logger disposed) — full-buffer pressure
// evicts the oldest pending event instead of failing the write. The event
// cannot be persisted — fault the Task rather than reporting false success,
// so a caller that awaits a critical audit event can tell it was dropped.
pending.Completion.TrySetException(
new ObjectDisposedException(nameof(SiteEventLogger),
"Event could not be recorded: the event logger has been disposed."));
}
return pending.Completion.Task;
}
///
/// Drains up to queued events per iteration and commits
/// them in a single transaction, instead of one transaction per event. Mirrors the
/// reference shape in SqliteAuditWriter.ProcessWriteQueueAsync/FlushBatch
/// (ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs): pull the first
/// item with the blocking ReadAllAsync enumerator, then greedily TryRead
/// whatever else is already queued (non-blocking) up to the batch cap, so a burst of
/// concurrent callers amortises SQLite's per-transaction
/// fsync/journal overhead across the whole burst instead of paying it per row.
///
private async Task ProcessWriteQueueAsync()
{
var batch = new List(WriteBatchSize);
await foreach (var first in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
{
batch.Clear();
batch.Add(first);
while (batch.Count < WriteBatchSize && _writeQueue.Reader.TryRead(out var next))
{
batch.Add(next);
}
FlushBatch(batch);
}
}
///
/// Commits a drained batch of pending events in one transaction, falling back to
/// row-by-row inserts if that transaction fails.
///
/// The batch is the fast path: one transaction amortises SQLite's fsync/journal cost
/// across the whole drain. But a batch is all-or-nothing, so before the fallback existed a
/// SINGLE unwritable row (an oversized details payload, a constraint/trigger reject)
/// faulted all ~ of its innocent neighbours and inflated
/// — the health metric — by the same factor. The rollback
/// is therefore followed by a per-row retry: one poison row costs exactly one row, the
/// rest still land, and the failure count stays truthful. Row-by-row is the degraded
/// path only — it is entered solely after a batch has already failed.
///
/// site_events ids are freshly minted GUIDs per
/// call (no cross-call replay/de-dup concern the way SqliteAuditWriter has), so a
/// re-insert during the fallback cannot duplicate a row the rolled-back batch had already
/// written. site_events is a CDC-replicated table (LocalDb Phase 1); its capture
/// triggers fire per row on INSERT either way, so neither shape changes
/// trigger/replication behavior.
///
private void FlushBatch(IReadOnlyList batch)
{
bool written;
Exception? failure = null;
try
{
written = WithConnection(connection =>
{
using var transaction = connection.BeginTransaction();
try
{
using var cmd = connection.CreateCommand();
cmd.Transaction = transaction;
cmd.CommandText = InsertEventSql;
var pId = cmd.Parameters.Add("$id", SqliteType.Text);
var pTimestamp = cmd.Parameters.Add("$timestamp", SqliteType.Text);
var pEventType = cmd.Parameters.Add("$event_type", SqliteType.Text);
var pSeverity = cmd.Parameters.Add("$severity", SqliteType.Text);
var pInstanceId = cmd.Parameters.Add("$instance_id", SqliteType.Text);
var pSource = cmd.Parameters.Add("$source", SqliteType.Text);
var pMessage = cmd.Parameters.Add("$message", SqliteType.Text);
var pDetails = cmd.Parameters.Add("$details", SqliteType.Text);
foreach (var pending in batch)
{
pId.Value = pending.Id;
pTimestamp.Value = pending.Timestamp;
pEventType.Value = pending.EventType;
pSeverity.Value = pending.Severity;
pInstanceId.Value = (object?)pending.InstanceId ?? DBNull.Value;
pSource.Value = pending.Source;
pMessage.Value = pending.Message;
pDetails.Value = (object?)pending.Details ?? DBNull.Value;
cmd.ExecuteNonQuery();
}
transaction.Commit();
Interlocked.Increment(ref _committedBatchCount);
}
catch
{
transaction.Rollback();
throw;
}
});
}
catch (Exception ex)
{
failure = ex;
written = false;
}
if (failure is not null)
{
if (batch.Count == 1)
{
// Nothing to isolate — retrying the single row would fail identically.
Interlocked.Increment(ref _failedWriteCount);
_logger.LogError(failure, "Failed to record 1 event (sqlite {SqliteError})",
DescribeSqliteError(failure));
batch[0].Completion.TrySetException(failure);
return;
}
// Degraded path: isolate the failure instead of charging the whole batch for it.
_logger.LogWarning(failure,
"Batch insert of {Count} site event(s) failed (sqlite {SqliteError}); retrying row by row",
batch.Count, DescribeSqliteError(failure));
FlushRowByRow(batch);
return;
}
if (written)
{
foreach (var pending in batch)
{
pending.Completion.TrySetResult();
}
}
else
{
// WithConnection returns false only when the logger has been
// disposed mid-drain; none of the batch was persisted. Fault every
// Task instead of reporting false success for a dropped event.
var disposedEx = DisposedMidWrite();
foreach (var pending in batch)
{
pending.Completion.TrySetException(disposedEx);
}
}
}
///
/// Re-inserts a failed batch one row at a time, each in its own implicit transaction, so a
/// single unwritable event costs exactly one event: its own Task faults and it alone is
/// counted in , while every other row in the batch still
/// lands and completes successfully. Only reached from after the
/// batched transaction has already rolled back.
///
/// The rolled-back batch to retry row by row.
private void FlushRowByRow(IReadOnlyList batch)
{
var failed = 0;
foreach (var pending in batch)
{
Exception? rowFailure = null;
bool written;
try
{
written = WithConnection(connection =>
{
using var cmd = connection.CreateCommand();
cmd.CommandText = InsertEventSql;
cmd.Parameters.AddWithValue("$id", pending.Id);
cmd.Parameters.AddWithValue("$timestamp", pending.Timestamp);
cmd.Parameters.AddWithValue("$event_type", pending.EventType);
cmd.Parameters.AddWithValue("$severity", pending.Severity);
cmd.Parameters.AddWithValue("$instance_id", (object?)pending.InstanceId ?? DBNull.Value);
cmd.Parameters.AddWithValue("$source", pending.Source);
cmd.Parameters.AddWithValue("$message", pending.Message);
cmd.Parameters.AddWithValue("$details", (object?)pending.Details ?? DBNull.Value);
cmd.ExecuteNonQuery();
});
}
catch (Exception ex)
{
rowFailure = ex;
written = false;
}
if (written)
{
pending.Completion.TrySetResult();
continue;
}
failed++;
if (rowFailure is null)
{
// Disposed mid-fallback; the remaining rows will take this path too.
pending.Completion.TrySetException(DisposedMidWrite());
continue;
}
_logger.LogError(rowFailure,
"Failed to record event from {Source} on row-by-row retry (sqlite {SqliteError})",
pending.Source, DescribeSqliteError(rowFailure));
pending.Completion.TrySetException(rowFailure);
}
if (failed > 0)
{
// Health Monitoring reads FailedWriteCount: only the rows that could not be
// written are counted, so the metric measures poison rows, not batch size.
Interlocked.Add(ref _failedWriteCount, failed);
}
}
///
/// The exception a pending event's Task is faulted with when the logger was disposed
/// before its write could complete. A fresh instance per call so each faulted Task carries
/// its own stack.
///
/// The exception to fault the pending event with.
private static ObjectDisposedException DisposedMidWrite() =>
new(nameof(SiteEventLogger),
"Event could not be recorded: the event logger was disposed before the write completed.");
///
/// The single INSERT shared by the batched fast path and the row-by-row fallback, so the
/// two shapes can never drift in column list or ordering.
///
private const string InsertEventSql = """
INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details)
VALUES ($id, $timestamp, $event_type, $severity, $instance_id, $source, $message, $details)
""";
///
/// Stops accepting new events, drains the write queue, and disposes the SQLite connection.
///
///
/// _disposed is deliberately NOT set in the first lock block. Completing
/// the channel writer is the whole shutdown signal: observes it
/// (TryWrite returns false) and the writer loop drains what is already buffered
/// before exiting. Flipping _disposed up front would make
/// return false for every remaining , so a graceful shutdown would
/// discard the entire queued backlog — the very events (shutdown diagnostics) most worth
/// keeping. It flips only after the loop has fully drained, where it guards the window
/// between drain and connection close. Mirrors SqliteAuditWriter.DisposeAsync.
///
public void Dispose()
{
Task? writerLoop;
lock (_writeLock)
{
if (_disposed) return;
// Stop accepting new events and let the writer loop drain.
_writeQueue.Writer.TryComplete();
writerLoop = _writerLoop;
}
// Wait for the writer loop to finish outside the lock — the loop itself
// acquires the lock for each write.
try
{
writerLoop?.Wait(TimeSpan.FromSeconds(5));
}
catch (AggregateException)
{
// A faulted writer loop has already been logged per event; nothing more
// to do during disposal.
}
lock (_writeLock)
{
if (_disposed) return;
_disposed = true;
_connection.Dispose();
}
}
///
/// "primary/extended" SQLite result codes (e.g. "10/522") for a
/// anywhere in the chain; "n/a" otherwise.
/// The exception message alone carries only the primary code, which proved
/// too generic to diagnose the 2026-07-20 disk-I/O incident
/// (known-issues/2026-07-20-localdb-disk-io-error-under-load.md).
///
private static string DescribeSqliteError(Exception ex)
{
for (Exception? e = ex; e is not null; e = e.InnerException)
{
if (e is SqliteException se)
{
return $"{se.SqliteErrorCode}/{se.SqliteExtendedErrorCode}";
}
}
return "n/a";
}
/// An event awaiting persistence by the background writer.
private sealed record PendingEvent(
string Id,
string Timestamp,
string EventType,
string Severity,
string? InstanceId,
string Source,
string Message,
string? Details)
{
/// Completes when the event has been durably persisted, or faults on write failure.
public TaskCompletionSource Completion { get; } =
new(TaskCreationOptions.RunContinuationsAsynchronously);
}
}