ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs

using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Tracking;

/// <summary>
/// Audit Log #23 (M3 Bundle A — Task A2) — schema + behaviour tests for the
/// site-local <see cref="OperationTrackingStore"/>.
/// </summary>
/// <remarks>
/// The store now takes <see cref="ILocalDb"/> rather than a connection string, so each
/// test gets a real <c>ILocalDb</c> over a unique temp FILE. The previous
/// <c>mode=memory&amp;cache=shared</c> fixture is gone and cannot come back:
/// <c>LocalDbOptions.Path</c> is a filesystem path, and — more importantly — a raw
/// connection to a shared-cache in-memory database would not carry the
/// <c>zb_hlc_next()</c> UDF the capture triggers call, so testing through one would no
/// longer resemble how the store runs in the host.
/// <para>
/// Verifier connections are plain <see cref="SqliteConnection"/>s over the same file.
/// That is deliberate and safe: they only read, or write columns on an unregistered
/// table (this fixture never calls <c>RegisterReplicated</c>), so no trigger fires and
/// the missing UDF is never reached.
/// </para>
/// </remarks>
public class OperationTrackingStoreTests : IDisposable
{
    // Every ILocalDb and temp file created by this fixture, torn down together. The
    // ILocalDb holds an open master connection anchoring the WAL, so it must be
    // disposed before the file can be deleted.
    private readonly List<ServiceProvider> _providers = [];
    private readonly List<string> _tempFiles = [];

    public void Dispose()
    {
        foreach (var provider in _providers)
        {
            try { provider.Dispose(); } catch { /* best effort */ }
        }

        foreach (var path in _tempFiles)
        {
            foreach (var suffix in new[] { "", "-wal", "-shm" })
            {
                try { File.Delete(path + suffix); } catch { /* best effort */ }
            }
        }

        GC.SuppressFinalize(this);
    }

    /// <summary>Allocates a unique temp database path, registered for cleanup.</summary>
    private string NewDatabasePath(string testName)
    {
        var path = Path.Combine(
            Path.GetTempPath(), $"optracking-{testName}-{Guid.NewGuid():N}.db");
        _tempFiles.Add(path);
        return path;
    }

    /// <summary>
    /// Builds a real <see cref="ILocalDb"/> over <paramref name="databasePath"/>. No
    /// <c>onReady</c> callback and no <c>RegisterReplicated</c> — the store's own
    /// <c>InitializeSchema</c> creates the table, which is what keeps a
    /// directly-constructed store self-sufficient.
    /// </summary>
    private ILocalDb CreateLocalDb(string databasePath)
    {
        var configuration = new ConfigurationBuilder()
            .AddInMemoryCollection(new Dictionary<string, string?>
            {
                ["LocalDb:Path"] = databasePath,
            })
            .Build();

        var provider = new ServiceCollection()
            .AddZbLocalDb(configuration)
            .BuildServiceProvider();
        _providers.Add(provider);

        return provider.GetRequiredService<ILocalDb>();
    }

    private (OperationTrackingStore store, string dataSource) CreateStore(string testName)
    {
        var databasePath = NewDatabasePath(testName);
        var store = new OperationTrackingStore(
            CreateLocalDb(databasePath),
            NullLogger<OperationTrackingStore>.Instance);
        return (store, databasePath);
    }

    private static SqliteConnection OpenVerifierConnection(string databasePath)
    {
        var connection = new SqliteConnection($"Data Source={databasePath}");
        connection.Open();
        return connection;
    }

    [Fact]
    public void Constructor_CreatesOperationTracking_SchemaOnFirstUse()
    {
        var (store, dataSource) = CreateStore(nameof(Constructor_CreatesOperationTracking_SchemaOnFirstUse));
        using (store)
        {
            using var connection = OpenVerifierConnection(dataSource);
            using var cmd = connection.CreateCommand();
            cmd.CommandText = "PRAGMA table_info(OperationTracking);";
            using var reader = cmd.ExecuteReader();

            var columns = new List<(string Name, int Pk, int NotNull)>();
            while (reader.Read())
            {
                columns.Add((reader.GetString(1), reader.GetInt32(5), reader.GetInt32(3)));
            }

            var expected = new[]
            {
                "TrackedOperationId", "Kind", "TargetSummary", "Status",
                "RetryCount", "LastError", "HttpStatus", "CreatedAtUtc",
                "UpdatedAtUtc", "TerminalAtUtc", "SourceInstanceId", "SourceScript",
                "SourceNode",
            };
            Assert.Equal(
                expected.OrderBy(n => n),
                columns.Select(c => c.Name).OrderBy(n => n));

            var pkColumns = columns.Where(c => c.Pk > 0).Select(c => c.Name).ToList();
            Assert.Single(pkColumns);
            Assert.Equal("TrackedOperationId", pkColumns[0]);
        }
    }

    [Fact]
    public void Initialize_creates_OperationTracking_with_SourceNode_column()
    {
        var (store, dataSource) = CreateStore(nameof(Initialize_creates_OperationTracking_with_SourceNode_column));
        using (store)
        {
            using var connection = OpenVerifierConnection(dataSource);
            Assert.True(
                ColumnExists(connection, "SourceNode"),
                "Fresh OperationTracking schema must include the SourceNode column.");
        }
    }

    /// <summary>
    /// The pre-SourceNode <c>OperationTracking</c> schema — the 12-column
    /// CREATE TABLE that has the original source-provenance columns
    /// (<c>SourceInstanceId</c>, <c>SourceScript</c>) but is WITHOUT
    /// <c>SourceNode</c>. A deployment that ran before the SourceNode
    /// stamping work already has an on-disk <c>tracking.db</c> in exactly
    /// this shape, and <c>CREATE TABLE IF NOT EXISTS</c> is a no-op against it.
    /// </summary>
    private const string OldPreSourceNodeSchema = """
        CREATE TABLE IF NOT EXISTS OperationTracking (
            TrackedOperationId TEXT    NOT NULL PRIMARY KEY,
            Kind               TEXT    NOT NULL,
            TargetSummary      TEXT    NULL,
            Status             TEXT    NOT NULL,
            RetryCount         INTEGER NOT NULL DEFAULT 0,
            LastError          TEXT    NULL,
            HttpStatus         INTEGER NULL,
            CreatedAtUtc       TEXT    NOT NULL,
            UpdatedAtUtc       TEXT    NOT NULL,
            TerminalAtUtc      TEXT    NULL,
            SourceInstanceId   TEXT    NULL,
            SourceScript       TEXT    NULL
        );
        CREATE INDEX IF NOT EXISTS IX_OperationTracking_Status_Updated
            ON OperationTracking (Status, UpdatedAtUtc);
        """;

    /// <summary>
    /// Writes the pre-SourceNode schema into <paramref name="databasePath"/> and closes
    /// the connection again. Unlike the old shared-cache in-memory fixture — where the
    /// seeding connection had to stay open or the database evaporated — a file database
    /// persists on its own, so nothing is held across the store's construction.
    /// </summary>
    private static void SeedPreSourceNodeSchemaDatabase(string databasePath)
    {
        using var connection = new SqliteConnection($"Data Source={databasePath}");
        connection.Open();
        using var cmd = connection.CreateCommand();
        cmd.CommandText = OldPreSourceNodeSchema;
        cmd.ExecuteNonQuery();
    }

    private static bool ColumnExists(SqliteConnection connection, string columnName)
    {
        using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT COUNT(*) FROM pragma_table_info('OperationTracking') WHERE name = $name";
        cmd.Parameters.AddWithValue("$name", columnName);
        return Convert.ToInt32(cmd.ExecuteScalar()) > 0;
    }

    private OperationTrackingStore CreateStoreOver(string databasePath)
        => new(CreateLocalDb(databasePath), NullLogger<OperationTrackingStore>.Instance);

    [Fact]
    public async Task Initialize_adds_SourceNode_to_pre_existing_schema()
    {
        var databasePath = NewDatabasePath(nameof(Initialize_adds_SourceNode_to_pre_existing_schema));

        // A pre-SourceNode deployment: the tracking database already exists with
        // the 12-column schema and NO SourceNode column.
        SeedPreSourceNodeSchemaDatabase(databasePath);
        using (var seeded = OpenVerifierConnection(databasePath))
        {
            Assert.True(ColumnExists(seeded, "SourceInstanceId"));
            Assert.True(ColumnExists(seeded, "SourceScript"));
            Assert.False(ColumnExists(seeded, "SourceNode"));
        }

        // Upgrade: a post-branch OperationTrackingStore opens the same database.
        // Its InitializeSchema must ALTER the missing SourceNode column in —
        // the CREATE TABLE IF NOT EXISTS alone is a no-op against the existing
        // table.
        await using (var store = CreateStoreOver(databasePath))
        {
            using (var verifier = OpenVerifierConnection(databasePath))
            {
                Assert.True(
                    ColumnExists(verifier, "SourceNode"),
                    "OperationTrackingStore must ALTER the SourceNode column into a pre-existing OperationTracking table.");
            }

            // A RecordEnqueueAsync binding $sourceNode must now succeed; without
            // the ALTER it would fail with "no such column: SourceNode".
            var id = TrackedOperationId.New();
            await store.RecordEnqueueAsync(
                id,
                kind: "ApiCallCached",
                targetSummary: "ERP.GetOrder",
                sourceInstanceId: "inst-1",
                sourceScript: "ScriptActor:OnTick",
                sourceNode: "node-a");

            var snapshot = await store.GetStatusAsync(id);
            Assert.NotNull(snapshot);
            Assert.Equal("node-a", snapshot!.SourceNode);
        }

        // Idempotency: a second store over the now-upgraded DB must not error
        // (the probe sees SourceNode already present and skips the ALTER).
        await using (var storeAgain = CreateStoreOver(databasePath))
        {
            using var verifier = OpenVerifierConnection(databasePath);
            Assert.True(ColumnExists(verifier, "SourceNode"));
        }
    }

    [Fact]
    public async Task RecordEnqueueAsync_persists_SourceNode()
    {
        var (store, _) = CreateStore(nameof(RecordEnqueueAsync_persists_SourceNode));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(
            id,
            kind: nameof(AuditKind.ApiCallCached),
            targetSummary: "ERP.GetOrder",
            sourceInstanceId: "Plant.Pump42",
            sourceScript: "ScriptActor:OnTick",
            sourceNode: "node-a");

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        Assert.Equal("node-a", snapshot!.SourceNode);
    }

    [Fact]
    public async Task RecordEnqueueAsync_persists_null_SourceNode()
    {
        var (store, _) = CreateStore(nameof(RecordEnqueueAsync_persists_null_SourceNode));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(
            id,
            kind: nameof(AuditKind.ApiCallCached),
            targetSummary: "ERP.GetOrder",
            sourceInstanceId: null,
            sourceScript: null,
            sourceNode: null);

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        Assert.Null(snapshot!.SourceNode);
    }

    [Fact]
    public async Task RecordEnqueueAsync_InsertsSubmittedRow_WithRetryCountZero()
    {
        var (store, dataSource) = CreateStore(nameof(RecordEnqueueAsync_InsertsSubmittedRow_WithRetryCountZero));
        await using var _ = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(
            id,
            kind: nameof(AuditKind.ApiCallCached),
            targetSummary: "ERP.GetOrder",
            sourceInstanceId: "Plant.Pump42",
            sourceScript: "ScriptActor:OnTick",
            sourceNode: null);

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        Assert.Equal(id, snapshot!.Id);
        Assert.Equal(nameof(AuditKind.ApiCallCached), snapshot.Kind);
        Assert.Equal("ERP.GetOrder", snapshot.TargetSummary);
        Assert.Equal(nameof(AuditStatus.Submitted), snapshot.Status);
        Assert.Equal(0, snapshot.RetryCount);
        Assert.Null(snapshot.LastError);
        Assert.Null(snapshot.HttpStatus);
        Assert.Null(snapshot.TerminalAtUtc);
        Assert.Equal("Plant.Pump42", snapshot.SourceInstanceId);
        Assert.Equal("ScriptActor:OnTick", snapshot.SourceScript);
        Assert.Equal(DateTimeKind.Utc, snapshot.CreatedAtUtc.Kind);
        Assert.Equal(DateTimeKind.Utc, snapshot.UpdatedAtUtc.Kind);
    }

    [Fact]
    public async Task RecordEnqueueAsync_Duplicate_IsNoOp_FirstWriteWins()
    {
        var (store, _) = CreateStore(nameof(RecordEnqueueAsync_Duplicate_IsNoOp_FirstWriteWins));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(id, "ApiCallCached", "ERP.GetOrder", "Plant.Pump42", "ScriptActor:OnTick", sourceNode: null);
        await store.RecordEnqueueAsync(id, "ApiCallCached", "OtherTarget", "Other.Instance", "ScriptActor:Other", sourceNode: null);

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        // First-write-wins: the second enqueue is ignored — Target/Source stay first.
        Assert.Equal("ERP.GetOrder", snapshot!.TargetSummary);
        Assert.Equal("Plant.Pump42", snapshot.SourceInstanceId);
        Assert.Equal("ScriptActor:OnTick", snapshot.SourceScript);
        Assert.Equal(nameof(AuditStatus.Submitted), snapshot.Status);
        Assert.Equal(0, snapshot.RetryCount);
    }

    [Fact]
    public async Task RecordAttemptAsync_AdvancesStatusAndRetryCount_OnNonTerminalRow()
    {
        var (store, _) = CreateStore(nameof(RecordAttemptAsync_AdvancesStatusAndRetryCount_OnNonTerminalRow));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(id, "ApiCallCached", "ERP.GetOrder", null, null, sourceNode: null);

        await store.RecordAttemptAsync(
            id,
            status: nameof(AuditStatus.Attempted),
            retryCount: 1,
            lastError: "HTTP 503 from ERP",
            httpStatus: 503);

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        Assert.Equal(nameof(AuditStatus.Attempted), snapshot!.Status);
        Assert.Equal(1, snapshot.RetryCount);
        Assert.Equal("HTTP 503 from ERP", snapshot.LastError);
        Assert.Equal(503, snapshot.HttpStatus);
        Assert.Null(snapshot.TerminalAtUtc);

        // UpdatedAtUtc advances past CreatedAtUtc.
        Assert.True(snapshot.UpdatedAtUtc >= snapshot.CreatedAtUtc);
    }

    [Fact]
    public async Task RecordAttemptAsync_OnTerminalRow_IsNoOp()
    {
        var (store, _) = CreateStore(nameof(RecordAttemptAsync_OnTerminalRow_IsNoOp));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(id, "ApiCallCached", "ERP.GetOrder", null, null, sourceNode: null);
        await store.RecordTerminalAsync(
            id,
            status: nameof(AuditStatus.Delivered),
            lastError: null,
            httpStatus: 200);

        var terminalSnapshot = await store.GetStatusAsync(id);
        Assert.NotNull(terminalSnapshot);
        Assert.NotNull(terminalSnapshot!.TerminalAtUtc);

        // Late attempt telemetry must NOT overwrite the terminal row.
        await store.RecordAttemptAsync(
            id,
            status: nameof(AuditStatus.Attempted),
            retryCount: 5,
            lastError: "late attempt",
            httpStatus: 500);

        var afterLate = await store.GetStatusAsync(id);
        Assert.NotNull(afterLate);
        Assert.Equal(nameof(AuditStatus.Delivered), afterLate!.Status);
        Assert.Equal(0, afterLate.RetryCount);
        Assert.Null(afterLate.LastError);
        Assert.Equal(200, afterLate.HttpStatus);
        Assert.NotNull(afterLate.TerminalAtUtc);
    }

    [Fact]
    public async Task RecordTerminalAsync_FlipsToTerminal_WithTerminalAtUtcSet()
    {
        var (store, _) = CreateStore(nameof(RecordTerminalAsync_FlipsToTerminal_WithTerminalAtUtcSet));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(id, "ApiCallCached", "ERP.GetOrder", null, null, sourceNode: null);

        var beforeTerminal = DateTime.UtcNow;
        await store.RecordTerminalAsync(
            id,
            status: nameof(AuditStatus.Parked),
            lastError: "HTTP 503 (max retries)",
            httpStatus: 503);

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        Assert.Equal(nameof(AuditStatus.Parked), snapshot!.Status);
        Assert.NotNull(snapshot.TerminalAtUtc);
        Assert.Equal(DateTimeKind.Utc, snapshot.TerminalAtUtc!.Value.Kind);
        Assert.True(snapshot.TerminalAtUtc >= beforeTerminal.AddSeconds(-1));
        Assert.Equal("HTTP 503 (max retries)", snapshot.LastError);
        Assert.Equal(503, snapshot.HttpStatus);
    }

    [Fact]
    public async Task GetStatusAsync_Unknown_ReturnsNull()
    {
        var (store, _) = CreateStore(nameof(GetStatusAsync_Unknown_ReturnsNull));
        await using var _store = store;

        var unknown = TrackedOperationId.New();
        var snapshot = await store.GetStatusAsync(unknown);

        Assert.Null(snapshot);
    }

    [Fact]
    public async Task GetStatusAsync_ReturnsLatestSnapshot_AfterMultipleAttempts()
    {
        var (store, _) = CreateStore(nameof(GetStatusAsync_ReturnsLatestSnapshot_AfterMultipleAttempts));
        await using var _store = store;

        var id = TrackedOperationId.New();
        await store.RecordEnqueueAsync(id, "ApiCallCached", "ERP.GetOrder", null, null, sourceNode: null);
        await store.RecordAttemptAsync(id, nameof(AuditStatus.Attempted), 1, "first failure", 503);
        await store.RecordAttemptAsync(id, nameof(AuditStatus.Attempted), 2, "second failure", 503);
        await store.RecordAttemptAsync(id, nameof(AuditStatus.Attempted), 3, "third failure", 504);

        var snapshot = await store.GetStatusAsync(id);
        Assert.NotNull(snapshot);
        Assert.Equal(3, snapshot!.RetryCount);
        Assert.Equal("third failure", snapshot.LastError);
        Assert.Equal(504, snapshot.HttpStatus);
    }

    [Fact]
    public async Task PurgeTerminalAsync_RemovesOldTerminalRows_KeepsRecent_KeepsNonTerminal()
    {
        var (store, dataSource) = CreateStore(nameof(PurgeTerminalAsync_RemovesOldTerminalRows_KeepsRecent_KeepsNonTerminal));
        await using var _store = store;

        // Three rows:
        // (a) terminal, old   → should be purged
        // (b) terminal, fresh → should be kept
        // (c) non-terminal, ancient CreatedAt → should be kept (no TerminalAtUtc)
        var aId = TrackedOperationId.New();
        var bId = TrackedOperationId.New();
        var cId = TrackedOperationId.New();

        await store.RecordEnqueueAsync(aId, "ApiCallCached", "A", null, null, sourceNode: null);
        await store.RecordEnqueueAsync(bId, "ApiCallCached", "B", null, null, sourceNode: null);
        await store.RecordEnqueueAsync(cId, "ApiCallCached", "C", null, null, sourceNode: null);

        await store.RecordTerminalAsync(aId, nameof(AuditStatus.Delivered), null, 200);
        await store.RecordTerminalAsync(bId, nameof(AuditStatus.Delivered), null, 200);

        // Backdate the (a) row's TerminalAtUtc to 30 days ago via a direct UPDATE
        // — RecordTerminalAsync stamps DateTime.UtcNow which we cannot inject.
        // The verifier connection shares the same in-memory store thanks to
        // mode=memory&cache=shared.
        using (var connection = OpenVerifierConnection(dataSource))
        using (var cmd = connection.CreateCommand())
        {
            cmd.CommandText =
                "UPDATE OperationTracking SET TerminalAtUtc = $old WHERE TrackedOperationId = $id;";
            cmd.Parameters.AddWithValue("$old", DateTime.UtcNow.AddDays(-30).ToString("o"));
            cmd.Parameters.AddWithValue("$id", aId.ToString());
            cmd.ExecuteNonQuery();
        }

        // Purge anything terminal older than 7 days.
        var threshold = DateTime.UtcNow.AddDays(-7);
        await store.PurgeTerminalAsync(threshold);

        Assert.Null(await store.GetStatusAsync(aId));      // purged
        Assert.NotNull(await store.GetStatusAsync(bId));   // kept (recent terminal)
        Assert.NotNull(await store.GetStatusAsync(cId));   // kept (non-terminal)
    }

    // ── Site Call Audit #22: ReadChangedSinceAsync (reconciliation pull) ───

    [Fact]
    public async Task ReadChangedSinceAsync_ReturnsRowsAtOrAfterCursor_OldestFirst()
    {
        var (store, dataSource) = CreateStore(nameof(ReadChangedSinceAsync_ReturnsRowsAtOrAfterCursor_OldestFirst));
        await using var _store = store;

        // Three rows with distinct UpdatedAtUtc, written out of chronological
        // order to prove the read sorts by UpdatedAtUtc ascending.
        var older = TrackedOperationId.New();
        var middle = TrackedOperationId.New();
        var newer = TrackedOperationId.New();
        await store.RecordEnqueueAsync(older, nameof(AuditKind.ApiCallCached), "ERP.A", null, null, "node-a");
        await store.RecordEnqueueAsync(middle, nameof(AuditKind.DbWriteCached), "DB.B", null, null, "node-b");
        await store.RecordEnqueueAsync(newer, nameof(AuditKind.ApiCallCached), "ERP.C", null, null, null);

        // Backdate UpdatedAtUtc so the ordering is deterministic and a cursor
        // can be placed cleanly between rows. (Enqueue stamps DateTime.UtcNow;
        // we cannot inject the clock, so set the timestamps directly.)
        var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
        SetUpdatedAt(dataSource, older, t0);
        SetUpdatedAt(dataSource, middle, t0.AddMinutes(10));
        SetUpdatedAt(dataSource, newer, t0.AddMinutes(20));

        // Cursor at the middle row's UpdatedAtUtc: inclusive lower bound, so
        // middle + newer come back, older is excluded.
        var result = await store.ReadChangedSinceAsync(t0.AddMinutes(10), batchSize: 100, ct: CancellationToken.None);

        Assert.Equal(2, result.Count);
        Assert.Equal(middle, result[0].TrackedOperationId);
        Assert.Equal(newer, result[1].TrackedOperationId);
        Assert.True(result[0].UpdatedAtUtc <= result[1].UpdatedAtUtc);
    }

    [Fact]
    public async Task ReadChangedSinceAsync_FromMinValue_ReturnsAllRows()
    {
        var (store, _) = CreateStore(nameof(ReadChangedSinceAsync_FromMinValue_ReturnsAllRows));
        await using var _store = store;

        await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "A", null, null, null);
        await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "B", null, null, null);

        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, ct: CancellationToken.None);

        Assert.Equal(2, result.Count);
    }

    [Fact]
    public async Task ReadChangedSinceAsync_IsBatchCapped()
    {
        var (store, dataSource) = CreateStore(nameof(ReadChangedSinceAsync_IsBatchCapped));
        await using var _store = store;

        var ids = new List<TrackedOperationId>();
        var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
        for (var i = 0; i < 5; i++)
        {
            var id = TrackedOperationId.New();
            ids.Add(id);
            await store.RecordEnqueueAsync(id, nameof(AuditKind.ApiCallCached), $"T{i}", null, null, null);
            SetUpdatedAt(dataSource, id, t0.AddMinutes(i));
        }

        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 3, ct: CancellationToken.None);

        // Capped to 3 — and the cap takes the OLDEST 3 (asc order) so the
        // caller can advance the cursor monotonically across follow-up pulls.
        Assert.Equal(3, result.Count);
        Assert.Equal(ids[0], result[0].TrackedOperationId);
        Assert.Equal(ids[1], result[1].TrackedOperationId);
        Assert.Equal(ids[2], result[2].TrackedOperationId);
    }

    [Fact]
    public async Task ReadChangedSinceAsync_MapsTrackingRowOntoSiteCallOperational()
    {
        var (store, _) = CreateStore(nameof(ReadChangedSinceAsync_MapsTrackingRowOntoSiteCallOperational));
        await using var _store = store;

        var apiId = TrackedOperationId.New();
        var dbId = TrackedOperationId.New();
        await store.RecordEnqueueAsync(apiId, nameof(AuditKind.ApiCallCached), "ERP.GetOrder", "inst-1", "ScriptActor:OnTick", "node-a");
        await store.RecordEnqueueAsync(dbId, nameof(AuditKind.DbWriteCached), "Historian.Write", null, null, "node-b");
        await store.RecordAttemptAsync(apiId, nameof(AuditStatus.Attempted), 2, "HTTP 503", 503);
        await store.RecordTerminalAsync(dbId, nameof(AuditStatus.Parked), "max retries", null);

        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, ct: CancellationToken.None);
        var api = result.Single(r => r.TrackedOperationId == apiId);
        var db = result.Single(r => r.TrackedOperationId == dbId);

        // Kind → Channel projection.
        Assert.Equal("ApiOutbound", api.Channel);
        Assert.Equal("DbOutbound", db.Channel);

        // TargetSummary → Target; SourceNode carried verbatim.
        Assert.Equal("ERP.GetOrder", api.Target);
        Assert.Equal("node-a", api.SourceNode);
        Assert.Equal("node-b", db.SourceNode);

        // Status / RetryCount / LastError / HttpStatus carried through.
        Assert.Equal(nameof(AuditStatus.Attempted), api.Status);
        Assert.Equal(2, api.RetryCount);
        Assert.Equal("HTTP 503", api.LastError);
        Assert.Equal(503, api.HttpStatus);

        // SourceSite is left empty by the store (the site id is not a tracking
        // column); the central client re-stamps it from the dialed siteId.
        Assert.Equal(string.Empty, api.SourceSite);

        // Terminal row carries TerminalAtUtc (UTC kind); active row leaves it null.
        Assert.Null(api.TerminalAtUtc);
        Assert.NotNull(db.TerminalAtUtc);
        Assert.Equal(DateTimeKind.Utc, db.TerminalAtUtc!.Value.Kind);

        // Timestamps round-trip as UTC.
        Assert.Equal(DateTimeKind.Utc, api.CreatedAtUtc.Kind);
        Assert.Equal(DateTimeKind.Utc, api.UpdatedAtUtc.Kind);
    }

    /// <summary>Directly sets a row's UpdatedAtUtc so cursor/ordering tests are deterministic.</summary>
    private static void SetUpdatedAt(string dataSource, TrackedOperationId id, DateTime updatedAtUtc)
    {
        using var connection = OpenVerifierConnection(dataSource);
        using var cmd = connection.CreateCommand();
        cmd.CommandText = "UPDATE OperationTracking SET UpdatedAtUtc = $u WHERE TrackedOperationId = $id;";
        cmd.Parameters.AddWithValue("$u", updatedAtUtc.ToString("o", System.Globalization.CultureInfo.InvariantCulture));
        cmd.Parameters.AddWithValue("$id", id.ToString());
        cmd.ExecuteNonQuery();
    }

    // ── SiteRuntime-024: read/write split + sync-safe Dispose ──────────────

    [Fact]
    public async Task SR024_ConcurrentReads_DoNotBlockOnInFlightWrite()
    {
        // Regression test for SiteRuntime-024 (perf half). Pre-fix, every
        // GetStatusAsync took the same _gate as RecordTerminalAsync, so a single
        // long-running write would queue up every concurrent status query. After
        // the fix, reads open a fresh SqliteConnection per call and don't take
        // the write gate at all — so they should run concurrently with a write.
        //
        // The test seeds a row, then issues many parallel reads while a write is
        // also in flight. We assert the reads return successfully (a regression
        // would either deadlock the test runner or take far longer than the gate
        // would have allowed any single read). The actual timing-comparison
        // assertion would be flaky in CI; this test asserts only correctness +
        // forward progress.
        var (store, _) = CreateStore(nameof(SR024_ConcurrentReads_DoNotBlockOnInFlightWrite));
        await using (store)
        {
            var id = TrackedOperationId.New();
            await store.RecordEnqueueAsync(
                id,
                kind: "ApiCallCached",
                targetSummary: "ERP.GetOrder",
                sourceInstanceId: null,
                sourceScript: null,
                sourceNode: "node-a");

            // Fire 10 concurrent reads + a write in parallel; all must complete.
            var readTasks = Enumerable.Range(0, 10)
                .Select(_ => store.GetStatusAsync(id))
                .ToArray();
            var writeTask = store.RecordAttemptAsync(
                id, status: "Retrying", retryCount: 1, lastError: "transient", httpStatus: 503);

            await Task.WhenAll(readTasks);
            await writeTask;

            foreach (var t in readTasks)
            {
                Assert.NotNull(await t);
            }
        }
    }

    [Fact]
    public async Task SR024_SyncDispose_DoesNotDeadlock_WhenInvokedFromFreshThread()
    {
        // Regression test for SiteRuntime-024 (deadlock half). Pre-fix, Dispose
        // bridged to async via DisposeAsyncCore().AsTask().GetAwaiter().GetResult()
        // — sync-over-async on a SemaphoreSlim can deadlock under a non-reentrant
        // SyncContext (host shutdown continuations). Post-fix, Dispose runs
        // synchronously without acquiring the gate.
        var (store, _) = CreateStore(nameof(SR024_SyncDispose_DoesNotDeadlock_WhenInvokedFromFreshThread));

        // Seed a row so the store has live state when disposed.
        await store.RecordEnqueueAsync(
            TrackedOperationId.New(),
            kind: "ApiCallCached",
            targetSummary: "ERP.GetOrder",
            sourceInstanceId: null,
            sourceScript: null,
            sourceNode: "node-a");

        var disposeReturned = new TaskCompletionSource<bool>();
        var disposeThread = new Thread(() =>
        {
            try
            {
                store.Dispose();
                disposeReturned.SetResult(true);
            }
            catch (Exception ex)
            {
                disposeReturned.SetException(ex);
            }
        }) { IsBackground = true };

        disposeThread.Start();

        // 5s ceiling — if Dispose deadlocks, the test fails with TimeoutException.
        var completed = await Task.WhenAny(
            disposeReturned.Task, Task.Delay(TimeSpan.FromSeconds(5)));
        Assert.Same(disposeReturned.Task, completed);
        Assert.True(await disposeReturned.Task);
    }

    [Fact]
    public async Task SR024_AsyncDispose_DoesNotDeadlock_AndIsIdempotent()
    {
        // The async path must also tolerate Dispose() being called afterwards
        // (host shutdown's standard pattern). The _disposeState exchange should
        // short-circuit the second call.
        var (store, _) = CreateStore(nameof(SR024_AsyncDispose_DoesNotDeadlock_AndIsIdempotent));

        await store.RecordEnqueueAsync(
            TrackedOperationId.New(),
            kind: "ApiCallCached",
            targetSummary: "ERP.GetOrder",
            sourceInstanceId: null,
            sourceScript: null,
            sourceNode: "node-a");

        await store.DisposeAsync();
        // Second call must be a no-op, not throw.
        store.Dispose();
        // 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);
    }

    // ── Task 15: composite (UpdatedAtUtc, TrackedOperationId) keyset ──

    /// <summary>
    /// Seeds <c>batchSize + 2</c> rows that all share ONE <c>UpdatedAtUtc</c> and
    /// pages through them with the composite keyset. Page 1 (no cursor) returns the
    /// first <c>batchSize</c>; page 2 resumes from <c>(sinceUtc, lastId)</c> and
    /// returns the REMAINING rows with no overlap — the stall the plain inclusive
    /// <c>&gt;=</c> resume can never escape.
    /// </summary>
    [Fact]
    public async Task ReadChangedSince_CompositeKeyset_AdvancesPastSharedTimestamp()
    {
        const int batchSize = 3;
        var (store, dataSource) = CreateStore(nameof(ReadChangedSince_CompositeKeyset_AdvancesPastSharedTimestamp));
        await using var _store = store;

        for (var i = 0; i < batchSize + 2; i++)
        {
            await store.RecordEnqueueAsync(TrackedOperationId.New(),
                kind: nameof(AuditKind.ApiCallCached), targetSummary: $"ERP.{i}",
                sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
        }

        var boundary = ForceSharedUpdatedAt(dataSource);

        // Page 1: no cursor → the first batchSize rows, ordered by TrackedOperationId.
        var page1 = await store.ReadChangedSinceAsync(boundary, batchSize);
        Assert.Equal(batchSize, page1.Count);

        // Page 2: resume strictly after the last id of page 1.
        var afterId = page1[^1].TrackedOperationId.ToString();
        var page2 = await store.ReadChangedSinceAsync(boundary, batchSize, afterId);

        // The remaining two rows, none seen on page 1, all strictly greater than the cursor.
        Assert.Equal(2, page2.Count);
        var page1Ids = page1.Select(r => r.TrackedOperationId).ToHashSet();
        Assert.All(page2, r => Assert.DoesNotContain(r.TrackedOperationId, page1Ids));
        Assert.All(page2, r => Assert.True(
            string.CompareOrdinal(r.TrackedOperationId.ToString(), afterId) > 0));

        // Union covers every seeded row exactly once — no gap, no dup.
        Assert.Equal(batchSize + 2, page1Ids.Concat(page2.Select(r => r.TrackedOperationId)).Distinct().Count());
    }

    /// <summary>
    /// Regression guard for the legacy contract: with NO cursor a resume at the
    /// shared timestamp re-reads the same page (the exact stall the keyset fixes) —
    /// an older central that never sends <c>after_id</c> is byte-for-byte unaffected.
    /// </summary>
    [Fact]
    public async Task ReadChangedSince_WithoutKeyset_ReReadsSamePage_LegacyPreserved()
    {
        const int batchSize = 3;
        var (store, dataSource) = CreateStore(nameof(ReadChangedSince_WithoutKeyset_ReReadsSamePage_LegacyPreserved));
        await using var _store = store;

        for (var i = 0; i < batchSize + 2; i++)
        {
            await store.RecordEnqueueAsync(TrackedOperationId.New(),
                kind: nameof(AuditKind.ApiCallCached), targetSummary: $"ERP.{i}",
                sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
        }

        var boundary = ForceSharedUpdatedAt(dataSource);

        var page1 = await store.ReadChangedSinceAsync(boundary, batchSize);
        // Legacy resume: same sinceUtc, no cursor → identical page (the stall).
        var pageAgain = await store.ReadChangedSinceAsync(boundary, batchSize);

        Assert.Equal(
            page1.Select(r => r.TrackedOperationId).ToList(),
            pageAgain.Select(r => r.TrackedOperationId).ToList());
    }

    /// <summary>Forces every OperationTracking row to one exact UpdatedAtUtc instant; returns it.</summary>
    private DateTime ForceSharedUpdatedAt(string dataSource)
    {
        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();
        return boundary;
    }
}