From ae94b22d7d75775cbb2d97b18de58be30fbfa1b4 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Wed, 8 Jul 2026 21:58:57 -0400
Subject: [PATCH] test(reconciliation): pin inclusive (>=) cursor reads at the
 site (audit + site-calls); EventId/upsert dedup absorbs boundary duplicates

---
 .../Site/SqliteAuditWriterWriteTests.cs       | 22 +++++++++++
 .../Tracking/OperationTrackingStoreTests.cs   | 39 +++++++++++++++++++
 2 files changed, 61 insertions(+)

diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs
index b5c6c227..c4559c09 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs
@@ -712,4 +712,26 @@ public class SqliteAuditWriterWriteTests
         var row = Assert.Single(rows);
         Assert.Equal(AuditOutcome.Success, row.Outcome);
     }
+
+    // ── Task 24: pin the inclusive (>=) reconciliation cursor boundary ──
+
+    [Fact]
+    public async Task ReadPendingSince_BoundaryTimestamp_IsInclusive_EventIdDedupAbsorbsDuplicates()
+    {
+        // Two rows share one OccurredAtUtc; a pull whose sinceUtc equals that
+        // timestamp MUST return both (inclusive >= read). Central reconciliation
+        // dedups on EventId (insert-if-not-exists), so re-reading the boundary row
+        // is a safe no-op — but an exclusive (>) read would silently and
+        // permanently lose the same-timestamp row. This test pins the >= contract.
+        var (writer, _) = CreateWriter(nameof(ReadPendingSince_BoundaryTimestamp_IsInclusive_EventIdDedupAbsorbsDuplicates));
+        await using var _w = writer;
+
+        var boundary = DateTime.UtcNow;
+        await writer.WriteAsync(NewEvent(id: Guid.NewGuid(), occurredAtUtc: boundary));
+        await writer.WriteAsync(NewEvent(id: Guid.NewGuid(), occurredAtUtc: boundary));
+
+        var rows = await writer.ReadPendingSinceAsync(boundary, batchSize: 10);
+
+        Assert.Equal(2, rows.Count);
+    }
 }
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs
index a89fc398..9f6858d5 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs
@@ -681,4 +681,43 @@ public class OperationTrackingStoreTests
         // And a third async — also a no-op.
         await store.DisposeAsync();
     }
+
+    // ── Task 24: pin the inclusive (>=) reconciliation cursor boundary ──
+
+    [Fact]
+    public async Task ReadChangedSince_BoundaryTimestamp_IsInclusive_UpsertOnNewerAbsorbsDuplicates()
+    {
+        // Two rows share one UpdatedAtUtc; a pull whose sinceUtc equals that instant
+        // MUST return both (inclusive >= read). Central ingest is insert-if-not-exists
+        // + upsert-on-newer, so re-reading the boundary row is a safe no-op — but an
+        // exclusive (>) read would silently and permanently drop the same-timestamp
+        // row. This test pins the >= contract.
+        var (store, dataSource) = CreateStore(
+            nameof(ReadChangedSince_BoundaryTimestamp_IsInclusive_UpsertOnNewerAbsorbsDuplicates));
+        await using var _store = store;
+
+        await store.RecordEnqueueAsync(TrackedOperationId.New(),
+            kind: nameof(AuditKind.ApiCallCached), targetSummary: "ERP.A",
+            sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
+        await store.RecordEnqueueAsync(TrackedOperationId.New(),
+            kind: nameof(AuditKind.ApiCallCached), targetSummary: "ERP.B",
+            sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
+
+        // Force both rows to share one exact UpdatedAtUtc instant, formatted the same
+        // way the read normalises the cursor ('Z'-suffixed round-trip "o").
+        var boundary = DateTime.UtcNow;
+        var boundaryText = DateTime.SpecifyKind(boundary, DateTimeKind.Utc)
+            .ToString("o", System.Globalization.CultureInfo.InvariantCulture);
+        using (var conn = OpenVerifierConnection(dataSource))
+        using (var cmd = conn.CreateCommand())
+        {
+            cmd.CommandText = "UPDATE OperationTracking SET UpdatedAtUtc = $ts";
+            cmd.Parameters.AddWithValue("$ts", boundaryText);
+            cmd.ExecuteNonQuery();
+        }
+
+        var rows = await store.ReadChangedSinceAsync(boundary, batchSize: 10);
+
+        Assert.Equal(2, rows.Count);
+    }
 }