ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs

using System.Threading.Channels;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
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.Commons.Types.Enums;
using AuditEvent = ZB.MOM.WW.Audit.AuditEvent;
using AuditOutcome = ZB.MOM.WW.Audit.AuditOutcome;

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

/// <summary>
/// Site-side SQLite hot-path writer for Audit Log events. Mirrors the
/// <see cref="ZB.MOM.WW.ScadaBridge.SiteEventLogging.SiteEventLogger"/> design — a single
/// owned <see cref="SqliteConnection"/> serialised behind a write lock, fed by a
/// bounded <see cref="Channel{T}"/> drained on a dedicated background writer
/// task — so script-thread callers never block on disk I/O.
/// </summary>
/// <remarks>
/// <para>
/// <b>Two-table schema.</b> The site store is now two tables:
/// the append-only canonical <c>audit_event</c> (the 10 canonical
/// <see cref="AuditEvent"/> fields stored directly — NO 24-column decompose) and
/// the mutable operational <c>audit_forward_state</c> sidecar that carries the
/// forwarding lifecycle (<see cref="AuditForwardState"/>), a duplicated
/// <c>OccurredAtUtc</c> for the drain index range-scan, a precomputed
/// <c>IsCachedKind</c> flag that drives the cached/non-cached drain split without
/// re-parsing <c>DetailsJson</c> on the read hot-path, plus attempt bookkeeping.
/// </para>
/// <para>
/// <b>Ephemeral reset.</b> The site SQLite store is ephemeral (≈7-day retention,
/// recreated per deployment), so this schema change is an in-place RESET: the new
/// tables are created and the old single 24-column <c>AuditLog</c> table is
/// DROP-ped if present. No SQLite data migration is performed (and none is
/// needed) — any rows in the old <c>AuditLog</c> table are within the retention
/// window and are discarded by the drop.
/// </para>
/// <para>
/// Site rows always carry <see cref="AuditForwardState.Pending"/> on first
/// insert; the central row-shape's <c>IngestedAtUtc</c> is a DetailsJson field
/// stamped by central on ingest, not a site column.
/// </para>
/// </remarks>
public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLifecycle, IAsyncDisposable, IDisposable
{
    // Microsoft.Data.Sqlite reports a generic SQLITE_CONSTRAINT (error code 19)
    // on a PRIMARY KEY violation; the extended subcode 1555 (SQLITE_CONSTRAINT_PRIMARYKEY)
    // is exposed via SqliteException.SqliteExtendedErrorCode but isn't reliably
    // surfaced across all SQLite builds. We treat any constraint error on insert
    // as a duplicate-eventid race and swallow it (first-write-wins) — the PRIMARY
    // KEY on audit_event.EventId is the constraint that fires first, so this scope
    // is precise (the sidecar insert for the same EventId is in the same
    // transaction and never reached once audit_event's insert throws).
    private const int SqliteErrorConstraint = 19;

    // Max EventIds bound into a single IN (...) UPDATE by MarkForwardedAsync /
    // MarkReconciledAsync. SQLite caps bound parameters per statement (999 on older
    // builds, SQLITE_MAX_VARIABLE_NUMBER); a large drain batch would otherwise blow the
    // ceiling as batch sizes grow. Chunk the id list into ≤500-id slices, all inside one
    // transaction so the batch stays atomic (arch-review 04, P8).
    private const int MarkForwardStateChunkSize = 500;

    private readonly SqliteConnection _connection;
    // Dedicated read-only connection used by GetBacklogStatsAsync,
    // ReadPendingAsync, ReadPendingSinceAsync, and ReadForwardedAsync so a slow
    // backlog scan (COUNT(*) over hundreds of thousands of Pending rows under a
    // central outage) never parks the hot-path writer behind _writeLock.
    // SQLite-with-WAL allows a second connection on the same file to read
    // concurrently with the writer; the writer's WAL pragma is set in
    // InitializeSchema before this connection is opened. The reader connection
    // has its own _readLock because SqliteConnection itself is not thread-safe
    // even in read-only mode — multiple read callers can otherwise interleave
    // commands on the shared connection.
    private readonly SqliteConnection _readConnection;
    private readonly object _readLock = new();
    private readonly SqliteAuditWriterOptions _options;
    private readonly ILogger<SqliteAuditWriter> _logger;
    private readonly INodeIdentityProvider _nodeIdentity;
    private readonly object _writeLock = new();
    private readonly Channel<PendingAuditEvent> _writeQueue;
    private readonly Task _writerLoop;
    private bool _disposed;

    /// <summary>Initializes a new instance of the SqliteAuditWriter class.</summary>
    /// <param name="options">Configuration options for the audit writer.</param>
    /// <param name="logger">Logger instance.</param>
    /// <param name="nodeIdentity">Node identity provider.</param>
    /// <param name="connectionStringOverride">Optional connection string override.</param>
    public SqliteAuditWriter(
        IOptions<SqliteAuditWriterOptions> options,
        ILogger<SqliteAuditWriter> logger,
        INodeIdentityProvider nodeIdentity,
        string? connectionStringOverride = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(logger);
        ArgumentNullException.ThrowIfNull(nodeIdentity);

        _options = options.Value;
        _logger = logger;
        _nodeIdentity = nodeIdentity;

        var connectionString = connectionStringOverride
            ?? $"Data Source={_options.DatabasePath};Cache=Shared";
        _connection = new SqliteConnection(connectionString);
        _connection.Open();

        InitializeSchema();
        LoadMaxServedRowId();

        // Open a second connection for read-only callers
        // (GetBacklogStatsAsync, ReadPendingAsync, ReadPendingSinceAsync,
        // ReadForwardedAsync). InitializeSchema set journal_mode=WAL on the
        // writer connection, which is a database-level setting that persists
        // for the file — subsequent connections to the same file see WAL and
        // can read concurrently with the writer without taking _writeLock.
        // Reuse the same connection string so the read connection sees the
        // same Data Source / Cache settings as the writer.
        _readConnection = new SqliteConnection(connectionString);
        _readConnection.Open();

        // PRAGMA foreign_keys is a per-connection setting. Set it on the read
        // connection as well so that any future read-path change (e.g. a
        // DELETE that may be added later) also benefits from FK enforcement.
        // Pure SELECT queries are unaffected — this is defensive belt-and-
        // suspenders for the read connection.
        using (var pragmaCmd = _readConnection.CreateCommand())
        {
            pragmaCmd.CommandText = "PRAGMA foreign_keys = ON";
            pragmaCmd.ExecuteNonQuery();
        }

        _writeQueue = Channel.CreateBounded<PendingAuditEvent>(
            new BoundedChannelOptions(_options.ChannelCapacity)
            {
                // Wait mode is only what the (never-awaited) async write path would
                // do; WriteAsync uses TryWrite exclusively and fails fast with
                // SqliteAuditWriterSaturatedException when the queue is full, so no
                // caller ever parks on room (review 2026-10-21 06/MEM-1). The
                // FallbackAuditWriter's drop-oldest ring absorbs the overflow.
                FullMode = BoundedChannelFullMode.Wait,
                SingleReader = true,
                SingleWriter = false,
            });
        _writerLoop = Task.Run(ProcessWriteQueueAsync);
    }

    private void InitializeSchema()
    {
        // auto_vacuum must be set before any table is created for it to take
        // effect on a fresh database. INCREMENTAL lets the `PRAGMA
        // incremental_vacuum` run at the end of PurgeExpiredAsync shrink the file
        // after the ~7-day retention purge — see alog.md §10.
        using (var pragmaCmd = _connection.CreateCommand())
        {
            pragmaCmd.CommandText = "PRAGMA auto_vacuum = INCREMENTAL";
            pragmaCmd.ExecuteNonQuery();
        }

        // Enable WAL so a second connection on the same file can
        // serve read-only callers (GetBacklogStatsAsync, ReadPendingAsync,
        // ReadPendingSinceAsync, ReadForwardedAsync) concurrently with the
        // batched writer, decoupling those reads from _writeLock. WAL is a
        // database-level setting persisted in the file header; setting it on
        // the writer connection means every connection opened to the file
        // afterwards inherits WAL behaviour. PRAGMA journal_mode returns the
        // mode actually adopted ("memory" for ":memory:" / shared-cache memory
        // mode, "wal" for file-backed) — we don't error if WAL was rejected
        // because the read connection's correctness does not depend on WAL
        // itself, only its concurrency advantage does.
        using (var pragmaCmd = _connection.CreateCommand())
        {
            pragmaCmd.CommandText = "PRAGMA journal_mode = WAL";
            pragmaCmd.ExecuteNonQuery();
        }

        // synchronous=NORMAL: SQLite's FULL default
        // fsyncs on every transaction commit, which is the dominant cost of the
        // per-event write path this same change coalesces via FlushIntervalMs — FULL
        // would still fsync once per FLUSH even after that fix. NORMAL fsyncs at
        // fewer, WAL-checkpoint-aligned points instead of every commit; the window
        // this opens is a handful of the most recent commits lost on OS crash / power
        // loss (WAL mode itself still guarantees no *corruption*, only a possible
        // short rollback on unclean restart). That trade is acceptable here because
        // audit writes are explicitly best-effort by design (CLAUDE.md: "Audit-write
        // failure NEVER aborts the user-facing action — audit is best-effort, the
        // action's own success/failure path is authoritative") — this is the same
        // durability class as a dropped audit row, not a new risk class.
        using (var pragmaCmd = _connection.CreateCommand())
        {
            pragmaCmd.CommandText = "PRAGMA synchronous = NORMAL";
            pragmaCmd.ExecuteNonQuery();
        }

        // Enable FK enforcement on the WRITE connection. PRAGMA foreign_keys is
        // a per-connection, per-session setting in SQLite — it is NOT persisted
        // in the database file, so every new connection that may INSERT into
        // audit_forward_state must set it for the FK
        //   audit_forward_state.EventId → audit_event.EventId
        // to be a real runtime guard rather than decorative DDL. The write
        // connection owns all INSERTs (and the MarkForwardedAsync /
        // MarkReconciledAsync UPDATEs), so setting it here — after WAL is
        // established, before the CREATE TABLEs — ensures the FK is live for
        // every insert that follows. The existing insert order (audit_event
        // first, then audit_forward_state, inside the same transaction) already
        // satisfies the FK, so no pre-existing rows can violate the constraint.
        using (var pragmaCmd = _connection.CreateCommand())
        {
            pragmaCmd.CommandText = "PRAGMA foreign_keys = ON";
            pragmaCmd.ExecuteNonQuery();
        }

        // In-place reset. The site store is EPHEMERAL (≈7-day
        // retention, recreated per deployment), so we do NOT migrate the old
        // single 24-column AuditLog table to the new two-table shape: any rows
        // it holds are within the retention window and discarded. DROP it if an
        // earlier deployment left it behind, then CREATE the two new tables. This
        // is safe precisely BECAUSE the site store is ephemeral — never do this
        // on a durable store (the central SQL Server side keeps its shim until
        // it is migrated, not reset).
        using (var dropCmd = _connection.CreateCommand())
        {
            dropCmd.CommandText = "DROP TABLE IF EXISTS AuditLog;";
            dropCmd.ExecuteNonQuery();
        }

        using var cmd = _connection.CreateCommand();
        cmd.CommandText = """
            -- Canonical, append-only / write-once: the 10 fields of the canonical
            -- ZB.MOM.WW.Audit.AuditEvent stored directly (DetailsJson carries the
            -- ScadaBridge domain fields). No forwarding state lives here — that is
            -- the audit_forward_state sidecar's concern.
            CREATE TABLE IF NOT EXISTS audit_event (
                EventId        TEXT NOT NULL,
                OccurredAtUtc  TEXT NOT NULL,
                Actor          TEXT NOT NULL,
                Action         TEXT NOT NULL,
                Outcome        TEXT NOT NULL,
                Category       TEXT NULL,
                Target         TEXT NULL,
                SourceNode     TEXT NULL,
                CorrelationId  TEXT NULL,
                DetailsJson    TEXT NULL,
                PRIMARY KEY (EventId)
            );

            -- Operational, mutable: the forwarding lifecycle for each canonical
            -- row. OccurredAtUtc is duplicated here so the drain range-scan stays
            -- on this one table's index; IsCachedKind is precomputed at insert so
            -- the cached/non-cached drain split never re-parses DetailsJson on the
            -- read hot-path.
            CREATE TABLE IF NOT EXISTS audit_forward_state (
                EventId        TEXT    NOT NULL,
                ForwardState   TEXT    NOT NULL,
                OccurredAtUtc  TEXT    NOT NULL,
                IsCachedKind   INTEGER NOT NULL,
                AttemptCount   INTEGER NOT NULL DEFAULT 0,
                LastAttemptUtc TEXT    NULL,
                PRIMARY KEY (EventId),
                FOREIGN KEY (EventId) REFERENCES audit_event(EventId)
            );

            -- Drain index: every read filters on (ForwardState, IsCachedKind) and
            -- range-scans/orders by OccurredAtUtc, so this composite covers the
            -- four reads + the backlog COUNT/MIN.
            CREATE INDEX IF NOT EXISTS IX_fwd
                ON audit_forward_state (ForwardState, IsCachedKind, OccurredAtUtc);

            -- Node-local key/value state that must survive a process restart. Holds the
            -- reconciliation pull's served-rows high-water mark ('max_served_rowid'), so a
            -- restart no longer resets the bound that lets a covering cursor retire rows the
            -- pull already delivered (review 2026-10-21 06/PERF-1 adversarial follow-up).
            -- Additive and idempotent: an older binary simply ignores the table.
            CREATE TABLE IF NOT EXISTS audit_site_state (
                Key   TEXT    NOT NULL PRIMARY KEY,
                Value INTEGER NOT NULL
            );

            -- Flow Debugger (design 2026-09-05 §5.4): PullTraceEvents reads audit_event by
            -- time window + json_extract on ExecutionId/ParentExecutionId; this bounds the scan
            -- to the window. Additive and idempotent.
            CREATE INDEX IF NOT EXISTS IX_audit_event_occurred
                ON audit_event (OccurredAtUtc);

            -- Fresh execution IDs permit clock-independent trace reads. Partial indexes
            -- exclude malformed legacy details before evaluating either JSON expression.
            CREATE INDEX IF NOT EXISTS IX_audit_event_execution
                ON audit_event (json_extract(DetailsJson, '$.executionId'), OccurredAtUtc, EventId)
                WHERE json_valid(DetailsJson);
            CREATE INDEX IF NOT EXISTS IX_audit_event_parent_execution
                ON audit_event (json_extract(DetailsJson, '$.parentExecutionId'), OccurredAtUtc, EventId)
                WHERE json_valid(DetailsJson);
            """;
        cmd.ExecuteNonQuery();
    }

    /// <summary>
    /// Enqueues an audit event for asynchronous batched persistence to SQLite.
    /// Never parks the caller on queue room: a full queue faults the returned task with
    /// <see cref="SqliteAuditWriterSaturatedException"/> immediately.
    /// </summary>
    /// <param name="evt">The audit event to persist.</param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>A task that completes when the event has been persisted.</returns>
    public Task WriteAsync(AuditEvent evt, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(evt);

        // The canonical record carries no ForwardState (a site-storage-only
        // concern). Site rows always start Pending; the sidecar row is written
        // alongside the canonical row in the same transaction.
        var pending = new PendingAuditEvent(evt, AuditForwardState.Pending);

        if (_writeQueue.Writer.TryWrite(pending))
        {
            return pending.Completion.Task;
        }

        // TryWrite fails for exactly two reasons: the writer is completed (disposed)
        // or the queue is at capacity. The second used to fall into an awaited
        // WriteAsync on the channel — one parked waiter, holding its AuditEvent, per
        // fire-and-continue script-thread caller for as long as the background writer
        // stayed stuck (e.g. behind a multi-minute retention purge). That count was
        // unbounded and invisible. Fail fast instead: FallbackAuditWriter routes the
        // event to its bounded drop-oldest ring and bumps SiteAuditWriteFailures, which
        // is the documented best-effort degradation (review 2026-10-21 06/MEM-1).
        if (Volatile.Read(ref _writerCompleted))
        {
            return Task.FromException(new ObjectDisposedException(nameof(SqliteAuditWriter),
                "Event could not be recorded: the audit writer has been disposed."));
        }

        Interlocked.Increment(ref _saturatedRejections);
        return Task.FromException(new SqliteAuditWriterSaturatedException(_options.ChannelCapacity));
    }

    /// <summary>
    /// Number of <see cref="WriteAsync"/> calls refused because the write queue was full
    /// since construction. Observability seam; the health-report count is the
    /// <c>SiteAuditWriteFailures</c> bump <see cref="FallbackAuditWriter"/> makes per refusal.
    /// </summary>
    public long SaturatedRejectionCount => Interlocked.Read(ref _saturatedRejections);
    private long _saturatedRejections;

    // Set (before the channel is completed) once DisposeAsync starts, so a refused
    // TryWrite can tell "disposed" from "full" without racing the channel's state.
    private bool _writerCompleted;

    /// <summary>
    /// Rows moved to <see cref="SiteAuditForwardStates.Rejected"/> since construction (also exported
    /// as the <c>scadabridge.site.audit.rejected_rows</c> counter).
    /// </summary>
    public long RejectedPushRowCount => Interlocked.Read(ref _rejectedPushRows);
    private long _rejectedPushRows;

    private async Task ProcessWriteQueueAsync()
    {
        var batch = new List<PendingAuditEvent>(_options.BatchSize);

        // ReadAllAsync completes when the channel is marked complete (Dispose).
        await foreach (var first in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
        {
            batch.Clear();
            batch.Add(first);
            var deadline = DateTime.UtcNow.AddMilliseconds(_options.FlushIntervalMs);

            // Pull additional ready events up to BatchSize. TryRead is non-
            // blocking and lets us amortise the transaction overhead across a
            // burst of concurrent enqueues. Once the immediately-available
            // burst is drained, keep the transaction open for up to
            // FlushIntervalMs waiting for MORE events to trickle in — this is
            // the soft-flush coalescing
            // this loop provides: without it, a low-rate trickle of writes (one script call every
            // few ms, well under BatchSize) fsyncs once PER EVENT, because the
            // writer loop reaches an empty channel and flushes immediately.
            // FlushIntervalMs bounds the worst-case added latency any single
            // event can see from this coalescing.
            while (batch.Count < _options.BatchSize)
            {
                if (_writeQueue.Reader.TryRead(out var next))
                {
                    batch.Add(next);
                    continue;
                }

                var remaining = deadline - DateTime.UtcNow;
                if (remaining <= TimeSpan.Zero)
                {
                    break;
                }

                var waitToRead = _writeQueue.Reader.WaitToReadAsync().AsTask();
                var completed = await Task.WhenAny(
                    waitToRead, Task.Delay(remaining)).ConfigureAwait(false);

                if (completed != waitToRead)
                {
                    // Deadline elapsed with nothing new — flush what we have.
                    break;
                }

                // WaitToReadAsync resolving false means the channel completed
                // (Dispose) with nothing left to read; resolving true means at
                // least one item is available, so the top of the loop's
                // TryRead will pick it up.
                if (!await waitToRead.ConfigureAwait(false))
                {
                    break;
                }
            }

            FlushBatch(batch);
        }
    }

    /// <summary>
    /// Total number of <see cref="FlushBatch"/> transactions committed since
    /// construction. Test-only observability seam for asserting that FlushIntervalMs coalesces a trickle of
    /// near-simultaneous writes into a single transaction rather than one
    /// commit per event; production code never reads this.
    /// </summary>
    internal int FlushCountForTests => _flushCountForTests;

    /// <summary>
    /// Test-only: the write lock FlushBatch / the flips / the purge serialise on, so a test can
    /// park the background writer deterministically (e.g. to fill the write queue).
    /// </summary>
    internal object WriteLockForTests => _writeLock;
    private int _flushCountForTests;

    private void FlushBatch(IReadOnlyList<PendingAuditEvent> batch)
    {
        lock (_writeLock)
        {
            if (_disposed)
            {
                foreach (var pending in batch)
                {
                    pending.Completion.TrySetException(
                        new ObjectDisposedException(nameof(SqliteAuditWriter),
                            "Event could not be recorded: the audit writer was disposed before the write completed."));
                }
                return;
            }

            using var transaction = _connection.BeginTransaction();
            try
            {
                // INSERT 1: the canonical row, stored DIRECTLY (the 10 canonical
                // fields straight off the AuditEvent — no Decompose; audit_event
                // holds canonical shape, not the legacy 24-column shape).
                using var eventCmd = _connection.CreateCommand();
                eventCmd.Transaction = transaction;
                eventCmd.CommandText = """
                    INSERT INTO audit_event (
                        EventId, OccurredAtUtc, Actor, Action, Outcome,
                        Category, Target, SourceNode, CorrelationId, DetailsJson
                    ) VALUES (
                        $EventId, $OccurredAtUtc, $Actor, $Action, $Outcome,
                        $Category, $Target, $SourceNode, $CorrelationId, $DetailsJson
                    );
                    """;
                var eEventId = eventCmd.Parameters.Add("$EventId", SqliteType.Text);
                var eOccurredAt = eventCmd.Parameters.Add("$OccurredAtUtc", SqliteType.Text);
                var eActor = eventCmd.Parameters.Add("$Actor", SqliteType.Text);
                var eAction = eventCmd.Parameters.Add("$Action", SqliteType.Text);
                var eOutcome = eventCmd.Parameters.Add("$Outcome", SqliteType.Text);
                var eCategory = eventCmd.Parameters.Add("$Category", SqliteType.Text);
                var eTarget = eventCmd.Parameters.Add("$Target", SqliteType.Text);
                var eSourceNode = eventCmd.Parameters.Add("$SourceNode", SqliteType.Text);
                var eCorrelationId = eventCmd.Parameters.Add("$CorrelationId", SqliteType.Text);
                var eDetailsJson = eventCmd.Parameters.Add("$DetailsJson", SqliteType.Text);

                // INSERT 2: the operational sidecar row. ForwardState=Pending,
                // OccurredAtUtc duplicated for the drain index, IsCachedKind
                // precomputed (so the read split never parses DetailsJson),
                // AttemptCount=0, LastAttemptUtc=NULL.
                using var fwdCmd = _connection.CreateCommand();
                fwdCmd.Transaction = transaction;
                fwdCmd.CommandText = """
                    INSERT INTO audit_forward_state (
                        EventId, ForwardState, OccurredAtUtc, IsCachedKind, AttemptCount, LastAttemptUtc
                    ) VALUES (
                        $EventId, $ForwardState, $OccurredAtUtc, $IsCachedKind, 0, NULL
                    );
                    """;
                var fEventId = fwdCmd.Parameters.Add("$EventId", SqliteType.Text);
                var fForwardState = fwdCmd.Parameters.Add("$ForwardState", SqliteType.Text);
                var fOccurredAt = fwdCmd.Parameters.Add("$OccurredAtUtc", SqliteType.Text);
                var fIsCachedKind = fwdCmd.Parameters.Add("$IsCachedKind", SqliteType.Integer);

                foreach (var pending in batch)
                {
                    var evt = pending.Event;
                    // Canonical OccurredAtUtc is UTC by construction; store the
                    // round-trip "o" form so string comparison stays monotonic
                    // (the drain range-scan and ORDER BY rely on it).
                    var occurredText = evt.OccurredAtUtc.UtcDateTime.ToString(
                        "o", System.Globalization.CultureInfo.InvariantCulture);

                    eEventId.Value = evt.EventId.ToString();
                    eOccurredAt.Value = occurredText;
                    // Canonical Actor is a required non-null string.
                    eActor.Value = evt.Actor ?? string.Empty;
                    eAction.Value = evt.Action;
                    eOutcome.Value = evt.Outcome.ToString();
                    eCategory.Value = (object?)evt.Category ?? DBNull.Value;
                    eTarget.Value = (object?)evt.Target ?? DBNull.Value;
                    // SourceNode-stamping: caller-provided value wins (preserves
                    // rows reconciled in from other nodes via the same writer);
                    // otherwise stamp from the local INodeIdentityProvider. The
                    // event record itself is NOT mutated — stamping is at write
                    // time only. If the provider also returns null (unconfigured
                    // node), the column stays NULL — operators see "needs config"
                    // via the schema, not a magic fallback string.
                    var sourceNode = evt.SourceNode ?? _nodeIdentity.NodeName;
                    eSourceNode.Value = (object?)sourceNode ?? DBNull.Value;
                    eCorrelationId.Value = (object?)evt.CorrelationId?.ToString() ?? DBNull.Value;
                    eDetailsJson.Value = (object?)evt.DetailsJson ?? DBNull.Value;

                    fEventId.Value = evt.EventId.ToString();
                    fForwardState.Value = pending.ForwardState.ToString();
                    fOccurredAt.Value = occurredText;
                    fIsCachedKind.Value = IsCachedKind(evt.DetailsJson) ? 1 : 0;

                    try
                    {
                        eventCmd.ExecuteNonQuery();
                        fwdCmd.ExecuteNonQuery();
                        // Do NOT complete pending.Completion here. A row's INSERTs succeeding is not durable until the
                        // transaction actually commits below — completing the caller's Task
                        // this early means a later commit failure (SQLITE_FULL, IOERR, or a
                        // later row in this same batch throwing) reports false success: the
                        // row is rolled back out of the database but the caller — and
                        // FallbackAuditWriter, whose ring-buffer fallback exists precisely
                        // for this outage — already saw it as delivered. Every row in this
                        // batch is completed with TrySetResult() together, AFTER
                        // transaction.Commit() succeeds, mirroring SiteEventLogger.FlushBatch.
                    }
                    catch (SqliteException ex) when (ex.SqliteErrorCode == SqliteErrorConstraint)
                    {
                        // Duplicate EventId — first-write-wins (alog.md §11). The
                        // audit_event PRIMARY KEY throws before the sidecar insert
                        // runs, so neither table gains a second row. Treat as
                        // success: the lifecycle event is durably recorded under
                        // the first writer's payload. Still deferred to after Commit()
                        // below — the surrounding transaction (this row's canonical/sidecar
                        // insert never ran, but earlier rows in the batch did) must still
                        // commit for that to be true.
                        _logger.LogDebug(ex,
                            "Duplicate EventId {EventId} swallowed by SqliteAuditWriter",
                            evt.EventId);
                    }
                }

                transaction.Commit();
                _flushCountForTests++;

                // Only NOW — after the whole batch is durably committed — tell every
                // caller (and FallbackAuditWriter) that their row landed.
                foreach (var pending in batch)
                {
                    pending.Completion.TrySetResult();
                }
            }
            catch (Exception ex)
            {
                transaction.Rollback();
                _logger.LogError(ex, "SqliteAuditWriter batch insert failed; faulting {Count} pending events", batch.Count);
                // Fault EVERY row in the batch, including any whose own INSERTs succeeded
                // above — none of that is true anymore, since the whole transaction (this
                // row's insert included) just rolled back. No TrySetResult has run yet for
                // any row in this batch (see above), so every TrySetException here is a
                // real transition, not a no-op racing an already-completed Task.
                foreach (var pending in batch)
                {
                    pending.Completion.TrySetException(ex);
                }
            }
        }
    }

    // Cached-lifecycle audit kinds that ride the combined-telemetry
    // drain (joined with the operational tracking row + pushed via
    // IngestCachedTelemetryAsync into the central dual-write transaction).
    // This is the SAME set the earlier ReadPendingCachedTelemetryAsync query
    // filtered on (Kind IN (...)); it is now precomputed into the sidecar's
    // IsCachedKind flag at INSERT (see IsCachedKind) so the read split is a cheap
    // integer predicate, not a JSON parse. ReadPendingAsync drains everything
    // with IsCachedKind=0; ReadPendingCachedTelemetryAsync drains IsCachedKind=1.
    private static readonly HashSet<AuditKind> CachedTelemetryKinds = new()
    {
        AuditKind.CachedSubmit,
        AuditKind.ApiCallCached,
        AuditKind.DbWriteCached,
        AuditKind.CachedResolve,
    };

    /// <summary>
    /// Precomputes the sidecar's <c>IsCachedKind</c> flag from a canonical
    /// row's <c>DetailsJson</c>. Parses the <see cref="AuditDetails.Kind"/>
    /// discriminator via <see cref="AuditDetailsCodec"/> and returns <c>true</c>
    /// iff it is one of the cached-lifecycle kinds
    /// (<see cref="AuditKind.CachedSubmit"/>, <see cref="AuditKind.ApiCallCached"/>,
    /// <see cref="AuditKind.DbWriteCached"/>, <see cref="AuditKind.CachedResolve"/>).
    /// Runs once per event at INSERT time so the cached/non-cached drain split is
    /// a cheap integer predicate on read, never a JSON parse on the hot path.
    /// </summary>
    private static bool IsCachedKind(string? detailsJson)
    {
        var details = AuditDetailsCodec.Deserialize(detailsJson);
        var kind = AuditRowProjection.ParseEnum(details.Kind, AuditKind.InboundRequest);
        return CachedTelemetryKinds.Contains(kind);
    }

    /// <summary>
    /// Returns up to <paramref name="limit"/> non-cached pending audit events, oldest first.
    /// Cached-lifecycle kinds are excluded; use <see cref="ReadPendingCachedTelemetryAsync"/> for those.
    /// </summary>
    /// <param name="limit">Maximum number of rows to return.</param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>A task that resolves to a read-only list of pending audit events.</returns>
    public Task<IReadOnlyList<AuditEvent>> ReadPendingAsync(int limit, CancellationToken ct = default)
    {
        if (limit <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
        }

        // Read via the dedicated _readConnection so this scan
        // (which can be expensive when the backlog grows under a central
        // outage) does not block the batched writer on _writeLock. WAL mode
        // gives us a stable snapshot of the tables while writes proceed on the
        // writer connection. _readLock serialises this connection across
        // multiple concurrent read callers since SqliteConnection itself is
        // not thread-safe.
        // JOIN the sidecar and filter on IsCachedKind=0 — the cached-
        // lifecycle kinds (IsCachedKind=1) flow through
        // ReadPendingCachedTelemetryAsync + the combined-telemetry drain. The
        // split is a precomputed integer predicate on the indexed sidecar, not
        // a DetailsJson parse. Ordering is by the sidecar's OccurredAtUtc with
        // EventId as the deterministic tiebreaker.
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            using var cmd = _readConnection.CreateCommand();
            cmd.CommandText = """
                SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
                       ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
                FROM   audit_event ae
                JOIN   audit_forward_state fs ON fs.EventId = ae.EventId
                WHERE  fs.ForwardState = $pending
                  AND  fs.IsCachedKind = 0
                ORDER  BY fs.OccurredAtUtc ASC, ae.EventId ASC
                LIMIT  $limit;
                """;
            cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
            cmd.Parameters.AddWithValue("$limit", limit);

            return Task.FromResult(ReadRows(cmd, limit));
        }
    }

    /// <summary>
    /// Returns up to <paramref name="limit"/> pending cached-lifecycle audit events, oldest first.
    /// Only rows with cached-call kinds (CachedSubmit, ApiCallCached, DbWriteCached, CachedResolve) are included.
    /// </summary>
    /// <param name="limit">Maximum number of rows to return.</param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>A task that resolves to a read-only list of pending cached-telemetry audit events.</returns>
    public Task<IReadOnlyList<AuditEvent>> ReadPendingCachedTelemetryAsync(
        int limit, CancellationToken ct = default)
    {
        if (limit <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
        }

        // Dedicated read surface for the cached-call lifecycle
        // drain — symmetric to ReadPendingAsync but filtered to IsCachedKind=1.
        // Same _readConnection + _readLock pattern so the hot-path writer is not
        // contended.
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            using var cmd = _readConnection.CreateCommand();
            cmd.CommandText = """
                SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
                       ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
                FROM   audit_event ae
                JOIN   audit_forward_state fs ON fs.EventId = ae.EventId
                WHERE  fs.ForwardState = $pending
                  AND  fs.IsCachedKind = 1
                ORDER  BY fs.OccurredAtUtc ASC, ae.EventId ASC
                LIMIT  $limit;
                """;
            cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
            cmd.Parameters.AddWithValue("$limit", limit);

            return Task.FromResult(ReadRows(cmd, limit));
        }
    }

    /// <summary>
    /// Returns up to <paramref name="limit"/> rows in
    /// <see cref="AuditForwardState.Forwarded"/>, oldest
    /// <see cref="AuditEvent.OccurredAtUtc"/> first, with
    /// <see cref="AuditEvent.EventId"/> as the deterministic tiebreaker. The
    /// <see cref="AuditForwardState.Forwarded"/>-specific counterpart of
    /// <see cref="ReadPendingAsync"/>; used by tests to assert a row reached the
    /// <see cref="AuditForwardState.Forwarded"/> state specifically (unlike
    /// <see cref="ReadPendingSinceAsync"/>, which also returns
    /// <see cref="AuditForwardState.Pending"/> rows).
    /// </summary>
    /// <param name="limit">Maximum number of rows to return.</param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>A task that resolves to a read-only list of forwarded audit events.</returns>
    public Task<IReadOnlyList<AuditEvent>> ReadForwardedAsync(int limit, CancellationToken ct = default)
    {
        if (limit <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
        }

        // Mirror ReadPendingAsync — read via _readConnection /
        // _readLock so this query never contends with the batched writer on
        // _writeLock. JOIN the sidecar and filter on ForwardState='Forwarded'
        // (no IsCachedKind split — both cached and non-cached Forwarded rows are
        // returned, as before).
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            using var cmd = _readConnection.CreateCommand();
            cmd.CommandText = """
                SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
                       ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
                FROM   audit_event ae
                JOIN   audit_forward_state fs ON fs.EventId = ae.EventId
                WHERE  fs.ForwardState = $forwarded
                ORDER  BY fs.OccurredAtUtc ASC, ae.EventId ASC
                LIMIT  $limit;
                """;
            cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
            cmd.Parameters.AddWithValue("$limit", limit);

            return Task.FromResult(ReadRows(cmd, limit));
        }
    }

    /// <inheritdoc />
    public Task MarkForwardedAsync(IReadOnlyList<Guid> eventIds, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(eventIds);
        if (eventIds.Count == 0)
        {
            return Task.CompletedTask;
        }

        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            // Flip the sidecar — UPDATE audit_forward_state, not the canonical
            // audit_event (which is append-only / write-once). Bump AttemptCount +
            // stamp LastAttemptUtc so operators can see how many drain passes a row
            // took to forward. Each id is bound as its own parameter, so no string
            // concatenation of user data ever enters the SQL.
            //
            // Defensive state guard: only transition rows that are still Pending or
            // Forwarded (i.e. not yet Reconciled). Without this guard a mis-called
            // batch that includes a Reconciled EventId would silently demote it back
            // to Forwarded — a state regression that would cause duplicate central
            // ingestion. Symmetric with MarkReconciledAsync's
            //   WHERE ForwardState IN ($pending, $forwarded)
            // guard. Current callers only pass Pending IDs, so normal-path behaviour
            // is unchanged; the guard is purely defensive.
            //
            // Stamp one $now for the whole call so every chunk records the same
            // LastAttemptUtc (matching the prior single-statement behaviour).
            var now = DateTime.UtcNow.ToString(
                "o", System.Globalization.CultureInfo.InvariantCulture);
            var forwarded = AuditForwardState.Forwarded.ToString();
            var pending = AuditForwardState.Pending.ToString();

            // Chunk into ≤MarkForwardStateChunkSize-id UPDATEs, all inside ONE
            // transaction so the whole batch commits atomically (arch-review 04, P8).
            using var transaction = _connection.BeginTransaction();
            for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize)
            {
                var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start);
                using var cmd = _connection.CreateCommand();
                cmd.Transaction = transaction;
                var sb = new System.Text.StringBuilder();
                sb.Append("UPDATE audit_forward_state SET ForwardState = $forwarded, ")
                  .Append("AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ")
                  .Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN (");
                for (int i = 0; i < chunk; i++)
                {
                    if (i > 0) sb.Append(',');
                    var p = $"$id{i}";
                    sb.Append(p);
                    cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString());
                }
                sb.Append(");");
                cmd.CommandText = sb.ToString();
                cmd.Parameters.AddWithValue("$forwarded", forwarded);
                cmd.Parameters.AddWithValue("$pending", pending);
                cmd.Parameters.AddWithValue("$now", now);
                cmd.ExecuteNonQuery();
            }
            transaction.Commit();
            return Task.CompletedTask;
        }
    }

    /// <summary>
    /// Returns up to <paramref name="batchSize"/> audit events central has not yet acknowledged
    /// (<see cref="AuditForwardState.Pending"/>, <see cref="SiteAuditForwardStates.PushAbandoned"/>,
    /// <see cref="SiteAuditForwardStates.Rejected"/>) with <see cref="AuditEvent.OccurredAtUtc"/>
    /// &gt;= <paramref name="sinceUtc"/>, oldest first. <see cref="AuditForwardState.Forwarded"/> rows
    /// are NOT served (review 2026-10-21 06/PERF-1). Used by the reconciliation-pull handler.
    /// </summary>
    /// <param name="sinceUtc">Lower bound timestamp (UTC) for event occurrence.</param>
    /// <param name="batchSize">Maximum number of rows to return.</param>
    /// <param name="afterId">
    /// Composite-keyset tiebreak: the EventId of the last row already consumed at
    /// <paramref name="sinceUtc"/>. Non-null switches the predicate from the inclusive
    /// <c>OccurredAtUtc &gt;= $since</c> to the strict composite
    /// <c>(OccurredAtUtc, EventId) &gt; ($since, $afterId)</c>, matching the query's own
    /// ORDER BY so a batch cannot stall on rows sharing one instant.
    /// </param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>A task that resolves to a read-only list of audit events since the given timestamp.</returns>
    public Task<IReadOnlyList<AuditEvent>> ReadPendingSinceAsync(
        DateTime sinceUtc, int batchSize, string? afterId = null, CancellationToken ct = default)
    {
        if (batchSize <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(batchSize), "batchSize must be > 0.");
        }

        // Read via _readConnection / _readLock — same lock-
        // decoupling as ReadPendingAsync. JOIN the sidecar; the range scan
        // is on the sidecar's duplicated OccurredAtUtc so it stays on IX_fwd.
        //
        // ONLY rows central has NOT acknowledged are served: Pending, plus the
        // push-exit states PushAbandoned / Rejected (review 2026-10-21 06/PERF-1).
        // Forwarded rows used to be served too, which made the pull re-ship every
        // row the push had already delivered — central re-inserted each one, and the
        // pull's 256-rows-per-tick budget was spent on acked rows, so any site above
        // ~0.85 audit rows/s fell permanently behind and latched the stalled signal.
        // A Forwarded row is acked by central; the pull has nothing to repair there.
        IReadOnlyList<AuditEvent> rows;
        long batchMaxRowId;
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            using var cmd = _readConnection.CreateCommand();
            var cursorPredicate = afterId is null
                ? "fs.OccurredAtUtc >= $since"
                // Composite keyset, lexicographic on (OccurredAtUtc, EventId) — the same
                // ordering the query applies, so it is a strict "everything after this row".
                : "(fs.OccurredAtUtc > $since OR (fs.OccurredAtUtc = $since AND ae.EventId > $afterId))";

            // fs.rowid rides along as an 11th column purely to maintain
            // _maxServedRowId — the insertion-order high-water mark that bounds
            // MarkReconciledUpToAsync (see that method). It is never mapped onto
            // the returned AuditEvent; rowid stays a storage-layer concept.
            cmd.CommandText = $"""
                SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
                       ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson,
                       fs.rowid
                FROM   audit_event ae
                JOIN   audit_forward_state fs ON fs.EventId = ae.EventId
                WHERE  fs.ForwardState IN ($pending, $pushAbandoned, $rejected)
                  AND  {cursorPredicate}
                ORDER  BY fs.OccurredAtUtc ASC, ae.EventId ASC
                LIMIT  $limit;
                """;
            AddOwedStateParameters(cmd);
            // Normalise to UTC ISO-8601 round-trip format to match how OccurredAtUtc
            // is stored on insert ("o" format) — string comparison is monotonic for
            // that encoding so we can index-scan against it.
            cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
                "o", System.Globalization.CultureInfo.InvariantCulture));
            if (afterId is not null)
            {
                // EventIds are stored as Guid.ToString() ("D"), so compare in that form.
                cmd.Parameters.AddWithValue("$afterId", NormalizeEventId(afterId));
            }
            cmd.Parameters.AddWithValue("$limit", batchSize);

            rows = ReadServedRows(cmd, batchSize, out batchMaxRowId);
        }

        // Advance (and persist) the mark OUTSIDE _readLock so the two locks are never nested in
        // read→write order, and UNDER _writeLock — the lock the purge's rowid clamp runs under.
        // Advancing it under _readLock from a value sampled before the SELECT let a purge clamp
        // that landed in between be overwritten with the stale-high value (review 2026-10-21
        // follow-up L1). Taking max(current, batch max) under _writeLock sees any clamp that
        // already happened; the served rows themselves are owed and so can never be purged.
        // Only when it moved — at most one tiny upsert per pull.
        lock (_writeLock)
        {
            if (!_disposed && batchMaxRowId > Volatile.Read(ref _maxServedRowId))
            {
                Volatile.Write(ref _maxServedRowId, batchMaxRowId);
                try
                {
                    PersistMaxServedRowId(batchMaxRowId);
                }
                catch (Exception ex)
                {
                    _logger.LogWarning(ex,
                        "SqliteAuditWriter could not persist the served high-water mark {Mark}; a restart re-learns it from the next pull.",
                        batchMaxRowId);
                }
            }
        }

        return Task.FromResult(rows);
    }

    /// <summary>
    /// Binds the three "owed to central" states — <c>$pending</c>, <c>$pushAbandoned</c>,
    /// <c>$rejected</c> — that the reconciliation pull serves and retires.
    /// </summary>
    private static void AddOwedStateParameters(SqliteCommand cmd)
    {
        cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
        cmd.Parameters.AddWithValue("$pushAbandoned", SiteAuditForwardStates.PushAbandoned);
        cmd.Parameters.AddWithValue("$rejected", SiteAuditForwardStates.Rejected);
    }

    /// <inheritdoc />
    public Task MarkPushAbandonedAsync(IReadOnlyList<Guid> eventIds, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(eventIds);
        if (eventIds.Count == 0)
        {
            return Task.CompletedTask;
        }

        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            // Pending → PushAbandoned only. Unlike the old "mark Forwarded" this is honest:
            // central never acked these rows, so they stay owed to the reconciliation pull
            // (served, retired by a covering cursor, never age-purged) — review 2026-10-21
            // 06/PERF-1. Attempt bookkeeping is bumped like any drain pass that touched them.
            var now = DateTime.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture);
            using var transaction = _connection.BeginTransaction();
            for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize)
            {
                var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start);
                using var cmd = _connection.CreateCommand();
                cmd.Transaction = transaction;
                var sb = new System.Text.StringBuilder();
                sb.Append("UPDATE audit_forward_state SET ForwardState = $abandoned, ")
                  .Append("AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ")
                  .Append("WHERE ForwardState = $pending AND EventId IN (");
                AppendIdParameters(cmd, sb, eventIds, start, chunk);
                sb.Append(");");
                cmd.CommandText = sb.ToString();
                cmd.Parameters.AddWithValue("$abandoned", SiteAuditForwardStates.PushAbandoned);
                cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
                cmd.Parameters.AddWithValue("$now", now);
                cmd.ExecuteNonQuery();
            }
            transaction.Commit();
            return Task.CompletedTask;
        }
    }

    /// <inheritdoc />
    public Task<int> RecordRejectedPushAttemptsAsync(
        IReadOnlyList<Guid> eventIds, int maxAttempts, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(eventIds);
        if (eventIds.Count == 0)
        {
            return Task.FromResult(0);
        }

        var threshold = Math.Max(1, maxAttempts);
        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            // Review 2026-10-21 06/IMPL-1: AttemptCount/LastAttemptUtc used to move only on
            // the Pending→Forwarded flip, so a row central kept refusing had no history and
            // no give-up — and since the push reads oldest-first, it sat at the head of every
            // batch forever. Bump on every refused attempt, then move rows that reached the
            // threshold out of the push queue into Rejected (still owed to the pull).
            var now = DateTime.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture);
            var flipped = 0;
            using var transaction = _connection.BeginTransaction();
            for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize)
            {
                var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start);

                using (var bump = _connection.CreateCommand())
                {
                    bump.Transaction = transaction;
                    var sb = new System.Text.StringBuilder();
                    sb.Append("UPDATE audit_forward_state SET AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ")
                      .Append("WHERE ForwardState = $pending AND EventId IN (");
                    AppendIdParameters(bump, sb, eventIds, start, chunk);
                    sb.Append(");");
                    bump.CommandText = sb.ToString();
                    bump.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
                    bump.Parameters.AddWithValue("$now", now);
                    bump.ExecuteNonQuery();
                }

                using (var reject = _connection.CreateCommand())
                {
                    reject.Transaction = transaction;
                    var sb = new System.Text.StringBuilder();
                    sb.Append("UPDATE audit_forward_state SET ForwardState = $rejected ")
                      .Append("WHERE ForwardState = $pending AND AttemptCount >= $max AND EventId IN (");
                    AppendIdParameters(reject, sb, eventIds, start, chunk);
                    sb.Append(");");
                    reject.CommandText = sb.ToString();
                    reject.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
                    reject.Parameters.AddWithValue("$rejected", SiteAuditForwardStates.Rejected);
                    reject.Parameters.AddWithValue("$max", threshold);
                    flipped += reject.ExecuteNonQuery();
                }
            }
            transaction.Commit();

            if (flipped > 0)
            {
                _logger.LogWarning(
                    "{Count} site audit row(s) were refused by central on {Max} consecutive pushes and moved out " +
                    "of the push queue as Rejected; the reconciliation pull still serves them (central records " +
                    "a ReconciliationAbandoned row if they can never be inserted).",
                    flipped, threshold);
                // Its OWN counter, not SiteAuditWriteFailures: the local write succeeded, and
                // the refusal is central's verdict on the row — blaming the site's writer for it
                // would be wrong (review 2026-10-21 adversarial follow-up).
                Interlocked.Add(ref _rejectedPushRows, flipped);
                SiteAuditTelemetryMetrics.RecordRejectedRows(flipped);
            }

            return Task.FromResult(flipped);
        }
    }

    private static void AppendIdParameters(
        SqliteCommand cmd, System.Text.StringBuilder sb, IReadOnlyList<Guid> eventIds, int start, int chunk)
    {
        for (int i = 0; i < chunk; i++)
        {
            if (i > 0) sb.Append(',');
            var p = $"$id{i}";
            sb.Append(p);
            cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString());
        }
    }

    /// <summary>
    /// Test/diagnostic read of one row's sidecar state: <c>(ForwardState, AttemptCount)</c>, or
    /// null when the row does not exist.
    /// </summary>
    /// <param name="eventId">The event to look up.</param>
    /// <returns>The row's forward state and attempt count, or null.</returns>
    internal (string State, int AttemptCount)? GetForwardStateForTests(Guid eventId)
    {
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);
            using var cmd = _readConnection.CreateCommand();
            cmd.CommandText = "SELECT ForwardState, AttemptCount FROM audit_forward_state WHERE EventId = $id;";
            cmd.Parameters.AddWithValue("$id", eventId.ToString());
            using var reader = cmd.ExecuteReader();
            return reader.Read() ? (reader.GetString(0), reader.GetInt32(1)) : null;
        }
    }

    /// <summary>Flow Debugger PullTraceEvents: hard cap on ids per read (design §5.4).</summary>
    public const int MaxExecutionIdsPerRead = 64;

    /// <summary>
    /// Reads audit events belonging to any of the given execution ids, newer than <paramref name="sinceUtc"/>, in keyset-paged order.
    /// </summary>
    /// <param name="executionIds">Execution ids to match (capped at <see cref="MaxExecutionIdsPerRead"/>).</param>
    /// <param name="sinceUtc">Lower bound on the event's occurrence time.</param>
    /// <param name="limit">Maximum number of events to return; must be positive.</param>
    /// <param name="ct">Cancellation token.</param>
    /// <param name="afterEventId">Optional keyset cursor; only events after this id are returned.</param>
    /// <returns>The matching audit events.</returns>
    public Task<IReadOnlyList<AuditEvent>> ReadByExecutionIdsAsync(
        IReadOnlyCollection<Guid> executionIds, DateTime sinceUtc, int limit, CancellationToken ct = default, string? afterEventId = null)
    {
        ArgumentNullException.ThrowIfNull(executionIds);
        if (limit <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
        }
        if (executionIds.Count == 0)
        {
            return Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>());
        }
        if (executionIds.Count > MaxExecutionIdsPerRead)
        {
            throw new ArgumentOutOfRangeException(nameof(executionIds),
                $"at most {MaxExecutionIdsPerRead} execution ids per read.");
        }

        // Flow Debugger (design 2026-09-05 §5.4): a POINT read independent of the forwarding
        // sidecar — a traced row is served whether Pending, Forwarded or Reconciled, and
        // serving it moves no cursor. DetailsJson is the codec's camelCase JSON, so the paths
        // are $.executionId / $.parentExecutionId; ids are stored in "D" form.
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            using var cmd = _readConnection.CreateCommand();
            var names = new List<string>(executionIds.Count);
            var i = 0;
            foreach (var id in executionIds)
            {
                var name = $"$e{i++}";
                names.Add(name);
                cmd.Parameters.AddWithValue(name, id.ToString("D"));
            }
            var inList = string.Join(", ", names);

            cmd.CommandText = BuildTraceReadSql(inList);
            cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
                "o", System.Globalization.CultureInfo.InvariantCulture));
            cmd.Parameters.AddWithValue("$after", (object?)afterEventId ?? DBNull.Value);
            cmd.Parameters.AddWithValue("$limit", limit);

            var rows = new List<AuditEvent>(Math.Min(limit, 256));
            using var reader = cmd.ExecuteReader();
            while (reader.Read())
            {
                rows.Add(MapRow(reader));
            }
            return Task.FromResult<IReadOnlyList<AuditEvent>>(rows);
        }
    }

    /// <summary>
    /// Reads matching keys through both expression indexes before loading the bounded payload page.
    /// UNION deduplicates rows matching both IDs. Keeping payloads outside the union bounds sorting
    /// to IDs/timestamps and avoids the optimizer preferring a whole-store timestamp scan.
    /// </summary>
    /// <param name="inList">The pre-built SQL parameter list for the <c>IN (...)</c> clauses.</param>
    /// <returns>The trace-read SQL text.</returns>
    internal static string BuildTraceReadSql(string inList) => $"""
        SELECT e.EventId, e.OccurredAtUtc, e.Actor, e.Action, e.Outcome,
               e.Category, e.Target, e.SourceNode, e.CorrelationId, e.DetailsJson
        FROM audit_event AS e
        JOIN (
            SELECT EventId, OccurredAtUtc
            FROM audit_event INDEXED BY IX_audit_event_execution
            WHERE json_valid(DetailsJson)
              AND json_extract(DetailsJson, '$.executionId') IN ({inList})
              AND (OccurredAtUtc > $since OR (OccurredAtUtc = $since AND ($after IS NULL OR EventId > $after)))
            UNION
            SELECT EventId, OccurredAtUtc
            FROM audit_event INDEXED BY IX_audit_event_parent_execution
            WHERE json_valid(DetailsJson)
              AND json_extract(DetailsJson, '$.parentExecutionId') IN ({inList})
              AND (OccurredAtUtc > $since OR (OccurredAtUtc = $since AND ($after IS NULL OR EventId > $after)))
            ORDER BY OccurredAtUtc, EventId
            LIMIT $limit
        ) AS matched ON matched.EventId = e.EventId
        ORDER BY matched.OccurredAtUtc, matched.EventId;
        """;

    /// <summary>
    /// Highest <c>audit_forward_state.rowid</c> this instance has ever SERVED from
    /// <see cref="ReadPendingSinceAsync"/> — i.e. the insertion-order high-water mark of
    /// rows central could possibly have received through the pull path. Bounds
    /// <see cref="MarkReconciledUpToAsync"/> so a row inserted after that point can never be
    /// retired by a cursor that happens to sit above its timestamp. Monotonic (a later, smaller
    /// batch never lowers it) except for the purge clamp. Every WRITE happens under
    /// <c>_writeLock</c> (the pull's advance and the purge's clamp alike — review 2026-10-21
    /// follow-up L1), so a clamp can never be overwritten with a stale-high value; reads from
    /// other paths go through <see cref="Volatile"/> for visibility. A stale read can only make
    /// the bound smaller, i.e. more conservative, never unsafe.
    /// </summary>
    private long _maxServedRowId;

    private const string MaxServedRowIdKey = "max_served_rowid";

    /// <summary>
    /// Restores <see cref="_maxServedRowId"/> from <c>audit_site_state</c>, clamped to the live
    /// <c>MAX(rowid)</c> so a stale value can never vouch for a reused rowid. Before this the
    /// bound was per-process: after a site-node restart it was 0, and because a healthy site's
    /// pull now often serves nothing, owed rows central had already ingested (e.g. PushAbandoned)
    /// could stay unretired — and counted on the backlog — indefinitely.
    /// </summary>
    private void LoadMaxServedRowId()
    {
        using var cmd = _connection.CreateCommand();
        cmd.CommandText = """
            SELECT MIN(
                IFNULL((SELECT Value FROM audit_site_state WHERE Key = $key), 0),
                IFNULL((SELECT MAX(rowid) FROM audit_forward_state), 0));
            """;
        cmd.Parameters.AddWithValue("$key", MaxServedRowIdKey);
        var value = Convert.ToInt64(cmd.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture);
        Volatile.Write(ref _maxServedRowId, value);
    }

    /// <summary>
    /// Durably records the served high-water mark. Runs on the WRITE connection under
    /// <c>_writeLock</c> (never nested inside <c>_readLock</c>). Best-effort: a failure leaves
    /// the in-memory bound intact and only means a restart re-learns it from the next pull.
    /// </summary>
    private void PersistMaxServedRowId(long value, SqliteTransaction? transaction = null)
    {
        using var cmd = _connection.CreateCommand();
        cmd.Transaction = transaction;
        cmd.CommandText = """
            INSERT INTO audit_site_state (Key, Value) VALUES ($key, $value)
            ON CONFLICT(Key) DO UPDATE SET Value = excluded.Value;
            """;
        cmd.Parameters.AddWithValue("$key", MaxServedRowIdKey);
        cmd.Parameters.AddWithValue("$value", value);
        cmd.ExecuteNonQuery();
    }

    /// <summary>Test seam: the current in-memory served high-water mark.</summary>
    internal long MaxServedRowIdForTests => Volatile.Read(ref _maxServedRowId);

    /// <summary>
    /// Executes a reconciliation-pull read whose projection carries <c>fs.rowid</c> as an
    /// 11th column, materialising the canonical rows and reporting the highest rowid served.
    /// The caller folds that into <see cref="_maxServedRowId"/> under <c>_writeLock</c> before
    /// returning the rows — serving a row is what makes it eligible for cursor-proved
    /// retirement, so the mark must cover a batch before central can see it.
    /// </summary>
    private IReadOnlyList<AuditEvent> ReadServedRows(SqliteCommand cmd, int capacityHint, out long maxRowId)
    {
        var rows = new List<AuditEvent>(Math.Min(capacityHint, 256));
        maxRowId = 0;
        using (var reader = cmd.ExecuteReader())
        {
            while (reader.Read())
            {
                rows.Add(MapRow(reader));
                var rowId = reader.GetInt64(10);
                if (rowId > maxRowId) maxRowId = rowId;
            }
        }

        return rows;
    }

    /// <summary>
    /// Normalises a wire-supplied event id to the exact textual form stored in
    /// <c>audit_event.EventId</c> so the keyset comparison is apples-to-apples. An
    /// unparseable value is passed through verbatim rather than throwing — a malformed
    /// cursor must degrade to "serves a bit too much", never to a failed pull.
    /// </summary>
    private static string NormalizeEventId(string afterId) =>
        Guid.TryParse(afterId, out var g) ? g.ToString() : afterId;

    /// <inheritdoc />
    public Task<int> MarkReconciledUpToAsync(
        DateTime sinceUtc, string? afterId, CancellationToken ct = default)
    {
        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            // Everything at or before central's cursor is proven received. With no
            // afterId the read contract is inclusive (>= $since), so only rows STRICTLY
            // older than the cursor instant are proven — the boundary instant may be
            // half-consumed and must stay servable.
            var predicate = afterId is null
                ? "fs.OccurredAtUtc < $since"
                : "(fs.OccurredAtUtc < $since OR (fs.OccurredAtUtc = $since AND fs.EventId <= $afterId))";

            // INSERTION-ORDER BOUND — "only rows we actually served can retire".
            //
            // The timestamp cursor alone is not enough. OccurredAtUtc is stamped by the
            // caller, so a row can be INSERTED after a batch was served yet carry a
            // timestamp BELOW central's (by then advanced) cursor — a late-stamped insert.
            // The blanket cursor UPDATE retired exactly those rows: never served, never
            // servable again (ReadPendingSinceAsync's keyset has moved past them), and,
            // being Reconciled, purged on age — silent audit loss, a failure mode the old
            // explicit id-set flip could not produce.
            //
            // fs.rowid is insertion order, so bounding the flip at the highest rowid this
            // instance has ever served restores the invariant exactly: rows inserted after
            // that point are out of the flip's reach no matter where the cursor sits.
            //
            // Only the OWED states flip (Pending / PushAbandoned / Rejected — exactly what
            // ReadPendingSinceAsync serves). Forwarded rows are no longer touched: central
            // acked them over the push path, they are already purge-eligible, and flipping
            // them was one UPDATE per pushed row for no benefit (review 2026-10-21 06/PERF-1).
            //
            // The bound is persisted in audit_site_state (loaded at construction, clamped to
            // the live MAX(rowid)), so a site-node restart no longer resets it to 0 and strands
            // rows the pull had already served. Conservative in the safe direction (it can
            // only delay retirement, never lose a row); the liveness note in
            // MarkReconciledUpToAsync's interface doc + Component-AuditLog.md covers what a
            // permanently halted reconciliation means for such rows.
            using var cmd = _connection.CreateCommand();
            cmd.CommandText = $"""
                UPDATE audit_forward_state AS fs
                SET    ForwardState = $reconciled
                WHERE  fs.ForwardState IN ($pending, $pushAbandoned, $rejected)
                  AND  fs.rowid <= $maxServedRowId
                  AND  {predicate};
                """;
            cmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
            AddOwedStateParameters(cmd);
            cmd.Parameters.AddWithValue("$maxServedRowId", Volatile.Read(ref _maxServedRowId));
            cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
                "o", System.Globalization.CultureInfo.InvariantCulture));
            if (afterId is not null)
            {
                cmd.Parameters.AddWithValue("$afterId", NormalizeEventId(afterId));
            }

            return Task.FromResult(cmd.ExecuteNonQuery());
        }
    }

    /// <inheritdoc />
    public Task MarkReconciledAsync(IReadOnlyList<Guid> eventIds, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(eventIds);
        if (eventIds.Count == 0)
        {
            return Task.CompletedTask;
        }

        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            // Flip the sidecar from Pending/Forwarded (or a push-exit state) → Reconciled.
            // Rows already Reconciled are left untouched (idempotent re-call), and the
            // canonical audit_event row is never modified.
            var reconciled = AuditForwardState.Reconciled.ToString();
            var pending = AuditForwardState.Pending.ToString();
            var forwarded = AuditForwardState.Forwarded.ToString();

            // Chunk into ≤MarkForwardStateChunkSize-id UPDATEs, all inside ONE
            // transaction so the whole batch commits atomically (arch-review 04, P8).
            using var transaction = _connection.BeginTransaction();
            for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize)
            {
                var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start);
                using var cmd = _connection.CreateCommand();
                cmd.Transaction = transaction;
                var sb = new System.Text.StringBuilder();
                sb.Append("UPDATE audit_forward_state SET ForwardState = $reconciled ")
                  .Append("WHERE ForwardState IN ($pending, $forwarded, $pushAbandoned, $rejected) AND EventId IN (");
                for (int i = 0; i < chunk; i++)
                {
                    if (i > 0) sb.Append(',');
                    var p = $"$id{i}";
                    sb.Append(p);
                    cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString());
                }
                sb.Append(");");
                cmd.CommandText = sb.ToString();
                cmd.Parameters.AddWithValue("$reconciled", reconciled);
                cmd.Parameters.AddWithValue("$pending", pending);
                cmd.Parameters.AddWithValue("$forwarded", forwarded);
                cmd.Parameters.AddWithValue("$pushAbandoned", SiteAuditForwardStates.PushAbandoned);
                cmd.Parameters.AddWithValue("$rejected", SiteAuditForwardStates.Rejected);
                cmd.ExecuteNonQuery();
            }
            transaction.Commit();
            return Task.CompletedTask;
        }
    }

    /// <inheritdoc />
    public Task<SiteAuditBacklogSnapshot> GetBacklogStatsAsync(CancellationToken ct = default)
    {
        int pendingCount;
        DateTime? oldestPending;

        // Read via the dedicated _readConnection (under
        // _readLock) so this probe — polled every 30 s by SiteAuditBacklogReporter
        // — never blocks the batched hot-path writer on _writeLock. Under a
        // central outage the Pending backlog can grow to hundreds of thousands
        // of rows and the COUNT(*) scan correspondingly stretches; that no
        // longer adds tail latency to user-facing audit writes.
        // Count over the sidecar (audit_forward_state) — the canonical
        // audit_event table carries no ForwardState. The IX_fwd index makes both
        // aggregates cheap (count is a covering scan, min is the first key).
        lock (_readLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            using var cmd = _readConnection.CreateCommand();
            // "Pending" on the health report means OWED to central: the push-exit states
            // (PushAbandoned / Rejected) still await the reconciliation pull and are never
            // purged, so they count toward the backlog and its oldest-row age.
            cmd.CommandText = """
                SELECT COUNT(*), MIN(OccurredAtUtc)
                FROM   audit_forward_state
                WHERE  ForwardState IN ($pending, $pushAbandoned, $rejected);
                """;
            AddOwedStateParameters(cmd);

            using var reader = cmd.ExecuteReader();
            reader.Read();
            pendingCount = reader.GetInt32(0);
            oldestPending = reader.IsDBNull(1)
                ? null
                : DateTime.Parse(reader.GetString(1),
                    System.Globalization.CultureInfo.InvariantCulture,
                    System.Globalization.DateTimeStyles.RoundtripKind);
        }

        // File-size lookup outside the lock — the DatabasePath option is the
        // canonical source. The connection-string-override branch (used by
        // some tests) keeps the same DatabasePath value, so this works
        // uniformly. In-memory / mode=memory paths return 0 because the file
        // doesn't exist on disk.
        long onDiskBytes = 0;
        try
        {
            if (!string.IsNullOrEmpty(_options.DatabasePath) &&
                !_options.DatabasePath.StartsWith(":memory:", StringComparison.Ordinal) &&
                !_options.DatabasePath.Contains("mode=memory", StringComparison.OrdinalIgnoreCase) &&
                File.Exists(_options.DatabasePath))
            {
                onDiskBytes = new FileInfo(_options.DatabasePath).Length;
            }
        }
        catch (Exception ex)
        {
            // File system probe is a best-effort health-metric — never abort
            // a backlog snapshot because stat() failed. Log and report 0.
            _logger.LogDebug(ex,
                "SqliteAuditWriter could not stat DB path {Path} for backlog snapshot.",
                _options.DatabasePath);
        }

        return Task.FromResult(new SiteAuditBacklogSnapshot(
            PendingCount: pendingCount,
            OldestPendingUtc: oldestPending,
            OnDiskBytes: onDiskBytes));
    }

    /// <inheritdoc />
    /// <remarks>
    /// <b>Paged (review 2026-10-21 06/MEM-1).</b> The purge deletes at most
    /// <see cref="SqliteAuditWriterOptions.PurgeBatchSize"/> rows per transaction and runs a
    /// bounded <c>PRAGMA incremental_vacuum(N)</c> after each page, taking <c>_writeLock</c>
    /// per page and releasing it in between, so the batched hot-path writer interleaves with
    /// a multi-million-row purge rather than stalling behind one giant DELETE + an unbounded
    /// vacuum. Every eligible row is still removed by one call; only the lock grain changed.
    /// </remarks>
    public async Task<int> PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default)
    {
        // Cutoff uses the same round-trip "o" encoding the writer stores
        // OccurredAtUtc in, so string comparison stays monotonic (see FlushBatch).
        var cutoff = EnsureUtc(olderThanUtc).ToString(
            "o", System.Globalization.CultureInfo.InvariantCulture);
        var pageSize = Math.Max(1, _options.PurgeBatchSize);

        var total = 0;
        while (true)
        {
            ct.ThrowIfCancellationRequested();
            var purged = PurgePage(cutoff, pageSize);
            total += purged;
            if (purged < pageSize)
            {
                return total;
            }

            // Lock already released by PurgePage; yield so the background writer (and any
            // flip/read waiting on _writeLock) gets the lock before the next page.
            await Task.Yield();
        }
    }

    /// <summary>
    /// Number of purge-page transactions committed since construction. Test-only
    /// observability seam for asserting the purge is paged; production code never reads it.
    /// </summary>
    internal int PurgePageCountForTests => Volatile.Read(ref _purgePageCountForTests);
    private int _purgePageCountForTests;

    /// <summary>
    /// Deletes ONE page (at most <paramref name="pageSize"/> rows) of purge-eligible rows in a
    /// single transaction under <c>_writeLock</c>, then reclaims a bounded number of freed
    /// pages. Returns the number of canonical rows deleted.
    /// </summary>
    private int PurgePage(string cutoff, int pageSize)
    {
        // Runs on the WRITE connection under _writeLock (a DELETE mutates the file),
        // mirroring MarkForwardedAsync / MarkReconciledAsync. Serialising with the
        // batched writer means the purge never races an in-flight FlushBatch.
        lock (_writeLock)
        {
            ObjectDisposedException.ThrowIf(_disposed, this);

            int purged;
            using (var transaction = _connection.BeginTransaction())
            {
                try
                {
                    // Capture this page's eligible EventIds from the sidecar FIRST: the
                    // Forwarded/Reconciled + age criteria live on audit_forward_state, and
                    // it is the FK child of audit_event, so it must be deleted before the
                    // parent. A temp table (rather than an IN (...) list) keeps the page off
                    // the SQLite 999-parameter limit. The hard invariant is encoded here:
                    // only Forwarded/Reconciled rows qualify — Pending (and the pull-owed
                    // PushAbandoned/Rejected states) are never purged on age alone.
                    using (var createCmd = _connection.CreateCommand())
                    {
                        createCmd.Transaction = transaction;
                        createCmd.CommandText = """
                            DROP TABLE IF EXISTS purge_ids;
                            CREATE TEMP TABLE purge_ids AS
                                SELECT EventId FROM audit_forward_state
                                WHERE ForwardState IN ($forwarded, $reconciled)
                                  AND OccurredAtUtc < $cutoff
                                LIMIT $limit;
                            """;
                        createCmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
                        createCmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
                        createCmd.Parameters.AddWithValue("$cutoff", cutoff);
                        createCmd.Parameters.AddWithValue("$limit", pageSize);
                        createCmd.ExecuteNonQuery();
                    }

                    // Delete the sidecar (FK child) first, then the canonical row.
                    using (var delSidecar = _connection.CreateCommand())
                    {
                        delSidecar.Transaction = transaction;
                        delSidecar.CommandText =
                            "DELETE FROM audit_forward_state WHERE EventId IN (SELECT EventId FROM purge_ids);";
                        delSidecar.ExecuteNonQuery();
                    }

                    using (var delEvent = _connection.CreateCommand())
                    {
                        delEvent.Transaction = transaction;
                        delEvent.CommandText =
                            "DELETE FROM audit_event WHERE EventId IN (SELECT EventId FROM purge_ids);";
                        delEvent.ExecuteNonQuery();
                    }

                    // changes() reports rows affected by the most recent DELETE on
                    // this connection (the audit_event delete above); read it before
                    // the DROP so no later statement can perturb the count.
                    using (var changesCmd = _connection.CreateCommand())
                    {
                        changesCmd.Transaction = transaction;
                        changesCmd.CommandText = "SELECT changes();";
                        purged = Convert.ToInt32(changesCmd.ExecuteScalar(),
                            System.Globalization.CultureInfo.InvariantCulture);
                    }

                    using (var dropCmd = _connection.CreateCommand())
                    {
                        dropCmd.Transaction = transaction;
                        dropCmd.CommandText = "DROP TABLE IF EXISTS purge_ids;";
                        dropCmd.ExecuteNonQuery();
                    }

                    // Clamp the served-rows high-water mark to what is still in the table.
                    // SQLite hands a fresh insert MAX(rowid)+1, so a purge that empties the
                    // sidecar (or removes its top rows) makes rowids REUSABLE — a stale-high
                    // _maxServedRowId would then vouch for rows nobody has served. Cheap:
                    // MAX(rowid) is an index-less O(1) lookup.
                    if (purged > 0)
                    {
                        using var maxCmd = _connection.CreateCommand();
                        maxCmd.Transaction = transaction;
                        maxCmd.CommandText = "SELECT IFNULL(MAX(rowid), 0) FROM audit_forward_state;";
                        var liveMax = Convert.ToInt64(maxCmd.ExecuteScalar(),
                            System.Globalization.CultureInfo.InvariantCulture);
                        if (liveMax < Volatile.Read(ref _maxServedRowId))
                        {
                            Volatile.Write(ref _maxServedRowId, liveMax);
                            PersistMaxServedRowId(liveMax, transaction);
                        }
                    }

                    transaction.Commit();
                    Interlocked.Increment(ref _purgePageCountForTests);
                }
                catch
                {
                    transaction.Rollback();
                    throw;
                }
            }

            // Reclaim a BOUNDED number of freed pages back to the OS. A bare
            // `PRAGMA incremental_vacuum;` empties the whole freelist in one step, which
            // after a large purge held _writeLock for as long as the delete itself did.
            // auto_vacuum=INCREMENTAL (set in InitializeSchema) makes this a cheap
            // freelist trim. Runs outside the transaction; a vacuum failure must not fault
            // the purge itself, so it is best-effort.
            if (purged > 0)
            {
                try
                {
                    using var vacuumCmd = _connection.CreateCommand();
                    vacuumCmd.CommandText = string.Create(System.Globalization.CultureInfo.InvariantCulture,
                        $"PRAGMA incremental_vacuum({Math.Max(1, _options.PurgeVacuumPagesPerBatch)});");
                    vacuumCmd.ExecuteNonQuery();
                }
                catch (Exception ex)
                {
                    _logger.LogDebug(ex,
                        "SqliteAuditWriter incremental_vacuum after retention purge failed (non-fatal).");
                }
            }

            return purged;
        }
    }

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

    /// <summary>
    /// Executes <paramref name="cmd"/> (one of the four reads, each already
    /// projecting the 10 <c>audit_event</c> columns in canonical order) and
    /// materialises the rows via <see cref="MapRow"/>.
    /// </summary>
    private static IReadOnlyList<AuditEvent> ReadRows(SqliteCommand cmd, int capacityHint)
    {
        var rows = new List<AuditEvent>(Math.Min(capacityHint, 256));
        using var reader = cmd.ExecuteReader();
        while (reader.Read())
        {
            rows.Add(MapRow(reader));
        }
        return rows;
    }

    /// <summary>
    /// Builds the canonical <see cref="AuditEvent"/> DIRECTLY from the 10
    /// stored <c>audit_event</c> columns — no 24-column <c>Recompose</c>, because
    /// <c>audit_event</c> already holds the canonical fields + <c>DetailsJson</c>.
    /// <c>Outcome</c> is stored as the enum's name; the safe
    /// <see cref="AuditRowProjection.ParseEnum{TEnum}"/> degrades an unknown/renamed
    /// value gracefully rather than throwing.
    /// </summary>
    private static AuditEvent MapRow(SqliteDataReader reader)
    {
        return new AuditEvent
        {
            EventId = Guid.Parse(reader.GetString(0)),
            OccurredAtUtc = new DateTimeOffset(DateTime.SpecifyKind(
                DateTime.Parse(reader.GetString(1),
                    System.Globalization.CultureInfo.InvariantCulture,
                    System.Globalization.DateTimeStyles.RoundtripKind),
                DateTimeKind.Utc)),
            Actor = reader.GetString(2),
            Action = reader.GetString(3),
            Outcome = AuditRowProjection.ParseEnum(reader.GetString(4), AuditOutcome.Success),
            Category = reader.IsDBNull(5) ? null : reader.GetString(5),
            Target = reader.IsDBNull(6) ? null : reader.GetString(6),
            SourceNode = reader.IsDBNull(7) ? null : reader.GetString(7),
            CorrelationId = reader.IsDBNull(8) ? null : Guid.Parse(reader.GetString(8)),
            DetailsJson = reader.IsDBNull(9) ? null : reader.GetString(9),
        };
    }

    /// <summary>
    /// Disposes the audit writer and releases resources.
    /// </summary>
    /// <remarks>
    /// Prefer <see cref="DisposeAsync"/> when possible (DI honours
    /// <see cref="IAsyncDisposable"/> on singletons). The sync path remains for
    /// callers that only know about <see cref="IDisposable"/> (e.g. legacy
    /// composition roots, <c>using</c> statements without <c>await</c>). To
    /// avoid the classic sync-over-async deadlock on a captured
    /// <see cref="SynchronizationContext"/> (ASP.NET request thread, Akka
    /// dispatcher under some configurations), we hop to the thread pool via
    /// <see cref="Task.Run(Func{Task})"/> before blocking on the result — the
    /// async continuation inside <see cref="DisposeAsync"/> then resumes on a
    /// pool thread with no captured context, so <c>GetResult()</c> never waits
    /// on the very thread the continuation needs.
    /// </remarks>
    public void Dispose()
    {
        Task.Run(async () => await DisposeAsync().ConfigureAwait(false))
            .GetAwaiter().GetResult();
    }

    /// <summary>Asynchronously disposes the audit writer and releases resources.</summary>
    /// <returns>A <see cref="ValueTask"/> that completes when all resources have been released.</returns>
    public async ValueTask DisposeAsync()
    {
        Task? writerLoop;
        lock (_writeLock)
        {
            if (_disposed) return;
            Volatile.Write(ref _writerCompleted, true);
            // Stop accepting new events. Completing the channel writer is the
            // shutdown signal: WriteAsync calls observe the completion and
            // fault, and the writer loop drains any already-buffered items
            // before exiting. _disposed is intentionally NOT set here — it
            // flips only after the loop has fully drained (second lock block
            // below), so FlushBatch's existing _disposed check guards the
            // post-drain window when the connection is about to close.
            _writeQueue.Writer.TryComplete();
            writerLoop = _writerLoop;
        }

        // Wait outside the lock — the loop reacquires it for each batch.
        try
        {
            if (writerLoop is not null)
            {
                await writerLoop.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
            }
        }
        catch (TimeoutException)
        {
            _logger.LogWarning("SqliteAuditWriter writer loop did not drain within 5s of dispose.");
        }
        catch (Exception ex)
        {
            // The loop's per-batch try/catch already routed individual failures
            // to pending TCSes; a top-level fault here is unexpected.
            _logger.LogError(ex, "SqliteAuditWriter writer loop faulted during dispose.");
        }

        lock (_writeLock)
        {
            if (_disposed) return;
            _disposed = true;
            _connection.Dispose();
        }

        // Dispose the dedicated read connection after the writer
        // is fully drained and closed. _readLock is taken to fence out any
        // in-flight read caller that grabbed the lock before _disposed flipped
        // — they observe ObjectDisposedException on the next attempt.
        lock (_readLock)
        {
            _readConnection.Dispose();
        }
    }

    /// <summary>An audit event awaiting persistence by the background writer.</summary>
    private sealed class PendingAuditEvent
    {
        /// <summary>Initializes a new instance of the PendingAuditEvent class.</summary>
        /// <param name="evt">The canonical audit event to persist.</param>
        /// <param name="forwardState">Initial site-local forwarding state written to the sidecar row (always Pending for fresh events).</param>
        public PendingAuditEvent(AuditEvent evt, AuditForwardState forwardState)
        {
            Event = evt;
            ForwardState = forwardState;
            Completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        }

        /// <summary>The canonical audit event to persist.</summary>
        public AuditEvent Event { get; }
        /// <summary>Initial forwarding state for this row's sidecar (bound to audit_forward_state.ForwardState).</summary>
        public AuditForwardState ForwardState { get; }
        /// <summary>Task completion source for write completion signaling.</summary>
        public TaskCompletionSource Completion { get; }
    }
}