Merge branch 'worktree-agent-a7084b23177344196' into arch-review-remediation

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

@@ -758,10 +758,17 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
     /// </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, CancellationToken ct = default)
+        DateTime sinceUtc, int batchSize, string? afterId = null, CancellationToken ct = default)
     {
         if (batchSize <= 0)
         {
@@ -779,13 +786,19 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
             ObjectDisposedException.ThrowIf(_disposed, this);
 
             using var cmd = _readConnection.CreateCommand();
-            cmd.CommandText = """
+            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))";
+
+            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 IN ($pending, $forwarded)
-                  AND  fs.OccurredAtUtc >= $since
+                  AND  {cursorPredicate}
                 ORDER  BY fs.OccurredAtUtc ASC, ae.EventId ASC
                 LIMIT  $limit;
                 """;
@@ -796,12 +809,63 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
             // 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);
 
             return Task.FromResult(ReadRows(cmd, batchSize));
         }
     }
 
+    /// <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))";
+
+            using var cmd = _connection.CreateCommand();
+            cmd.CommandText = $"""
+                UPDATE audit_forward_state AS fs
+                SET    ForwardState = $reconciled
+                WHERE  fs.ForwardState IN ($pending, $forwarded)
+                  AND  {predicate};
+                """;
+            cmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
+            cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
+            cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
+            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)
     {