scadaproj/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/Convergence/BidirectionalConvergenceTests.cs

namespace ZB.MOM.WW.LocalDb.Tests.Convergence;

/// <summary>
/// End-to-end bidirectional convergence over a REAL gRPC TestHost (see <see cref="ConvergenceFixture"/>):
/// concurrent writes, an update/delete race, offline accumulation, restart-mid-stream, and a pruned-horizon
/// snapshot resync — each driven through the normal <c>ILocalDb</c> API while both sync sessions run.
/// </summary>
public sealed class BidirectionalConvergenceTests
{
    [Fact]
    public async Task ConcurrentWrites_BothSides_Converge()
    {
        await using var fx = new ConvergenceFixture();
        await fx.StartAsync();

        // Interleave writes on both nodes while the sessions run; disjoint key ranges converge to
        // the union.
        for (var i = 1; i <= 25; i++)
        {
            await ConvergenceFixture.UpsertAsync(fx.A, i, "A", i);
            await ConvergenceFixture.UpsertAsync(fx.B, 1000 + i, "B", i);
        }

        // Shared key: A writes it first and lets it converge, THEN B overwrites it. Because B's HLC
        // clock observed A's replicated stamp, B's write is provably the higher HLC -> the LWW winner
        // is deterministically B on both nodes (avoids an HLC-tie decided by random node-id order).
        await ConvergenceFixture.UpsertAsync(fx.A, 100, "A_FIRST", 1);
        await fx.AssertConvergedAsync();
        await ConvergenceFixture.UpsertAsync(fx.B, 100, "B_WINS", 2);

        await fx.AssertConvergedAsync();

        var a = await ReadRow(fx.A, 100);
        var b = await ReadRow(fx.B, 100);
        Assert.Equal("B_WINS", a.Sku);
        Assert.Equal(a, b);
        Assert.Equal(51, await ConvergenceFixture.CountOrdersAsync(fx.A));
        Assert.Equal(51, await ConvergenceFixture.CountOrdersAsync(fx.B));
    }

    [Fact]
    public async Task UpdateDeleteRace_Converges()
    {
        await using var fx = new ConvergenceFixture();

        // Seed the contested key on both nodes and let it converge first.
        await ConvergenceFixture.UpsertAsync(fx.A, 7, "SEED", 1);
        await fx.StartAsync();
        await fx.AssertConvergedAsync();

        // Race: A updates the same pk while B deletes it, inside one window. Whichever HLC wins, both
        // nodes must end identical (either the updated row on both, or absent on both).
        await ConvergenceFixture.UpsertAsync(fx.A, 7, "A_UPDATED", 99);
        await ConvergenceFixture.DeleteAsync(fx.B, 7);

        await fx.AssertConvergedAsync();

        var a = await ConvergenceFixture.CountOrdersAsync(fx.A);
        var b = await ConvergenceFixture.CountOrdersAsync(fx.B);
        Assert.Equal(a, b); // identical outcome; AssertConverged already proved row-level identity
    }

    [Fact]
    public async Task OfflineAccumulation_Reconnect_Converges()
    {
        await using var fx = new ConvergenceFixture();
        await fx.StartAsync();
        await fx.AssertConvergedAsync();

        await fx.KillTransportAsync();

        // Both nodes keep taking local writes while the transport is down.
        for (var i = 1; i <= 20; i++)
        {
            await ConvergenceFixture.UpsertAsync(fx.A, i, "A_OFFLINE", i);
            await ConvergenceFixture.UpsertAsync(fx.B, 500 + i, "B_OFFLINE", i);
        }

        await fx.RestartTransportAsync();
        await fx.AssertConvergedAsync();

        Assert.Equal(40, await ConvergenceFixture.CountOrdersAsync(fx.A));
        Assert.Equal(40, await ConvergenceFixture.CountOrdersAsync(fx.B));
    }

    [Fact]
    public async Task RestartMidStream_NoLoss_NoDuplication()
    {
        const int rows = 400;
        // Small batches so a large transfer is genuinely in-flight when the transport dies mid-stream.
        await using var fx = new ConvergenceFixture(
            tuneInitiator: d => d["LocalDb:Replication:MaxBatchSize"] = "10");

        for (var i = 1; i <= rows; i++)
            await ConvergenceFixture.UpsertAsync(fx.A, i, "R", i);

        await fx.StartAsync();

        // Sever the transport once the transfer is under way (first rows landed on B). In-process
        // gRPC is fast, so the kill may land anywhere in the transfer — the point is that a
        // kill+restart around a large multi-batch transfer neither loses nor duplicates rows.
        await WaitUntilAsync(async () => await ConvergenceFixture.CountOrdersAsync(fx.B) >= 1,
            TimeSpan.FromSeconds(10));
        await fx.KillTransportAsync();

        await fx.RestartTransportAsync();
        await fx.AssertConvergedAsync();

        // Exact count on both sides: no dupes (INSERT OR REPLACE + LWW idempotency) and no loss.
        Assert.Equal(rows, await ConvergenceFixture.CountOrdersAsync(fx.A));
        Assert.Equal(rows, await ConvergenceFixture.CountOrdersAsync(fx.B));
    }

    [Fact]
    public async Task PruneDuringOutage_SnapshotRecovers_E2E()
    {
        // Tiny backlog cap on A so its oplog prunes below B's watermark during the outage, forcing a
        // full-snapshot resync on reconnect rather than incremental deltas.
        await using var fx = new ConvergenceFixture(
            tuneInitiator: d => d["LocalDb:Replication:MaxOplogRows"] = "3");
        await fx.StartAsync();
        await fx.AssertConvergedAsync();

        await fx.KillTransportAsync();

        // Accumulate well past the cap while B is offline.
        for (var i = 1; i <= 20; i++)
            await ConvergenceFixture.UpsertAsync(fx.A, i, "PRUNED", i);
        Assert.True(await fx.EnforceCapsAAsync(), "expected a prune + needs_snapshot flag");

        await fx.RestartTransportAsync();
        // Snapshot delivers all 20 rows; A's snapshot-covered tail oplog is not delta-acked yet, so
        // assert DATA convergence here (not oplog-empty).
        await fx.AssertDataConvergedAsync();
        Assert.Equal(20, await ConvergenceFixture.CountOrdersAsync(fx.B));

        // Deltas resume after the snapshot: a fresh write flows incrementally, its ack prunes the
        // snapshot-covered tail, and the oplog fully drains -> full convergence.
        await ConvergenceFixture.UpsertAsync(fx.A, 21, "POST_SNAP", 21);
        await fx.AssertConvergedAsync();
        Assert.Equal(21, await ConvergenceFixture.CountOrdersAsync(fx.B));
    }

    // TODO(product bug): `INSERT ... ON CONFLICT(id) DO UPDATE` on a replicated table crashes the
    // AFTER-UPDATE capture trigger `__localdb_<table>_au` with
    //   SQLite Error 19 (1555 SQLITE_CONSTRAINT_PRIMARYKEY):
    //   'UNIQUE constraint failed: __localdb_row_version.table_name, __localdb_row_version.pk_json'.
    // Minimal repro: on a fresh replicated `orders` table, upsert the SAME pk twice via ON CONFLICT
    // DO UPDATE (first = insert, second = the DO-UPDATE branch) — no replication, no concurrency.
    // A plain `UPDATE orders SET ...` firing the same au trigger does NOT fail; the defect is
    // specific to the upsert form perturbing last_insert_rowid() inside the au trigger's
    // `INSERT OR REPLACE INTO __localdb_row_version ... WHERE seq = last_insert_rowid()` capture.
    // The convergence tests intentionally avoid the upsert form (see ConvergenceFixture.UpsertAsync).
    // Un-skip once the capture trigger supports UPSERT.
    [Fact(Skip = "Product bug: ON CONFLICT DO UPDATE crashes the au capture trigger (UNIQUE on __localdb_row_version). See TODO above.")]
    public async Task UpsertOnConflictDoUpdate_CrashesCaptureTrigger_ProductBug()
    {
        await using var fx = new ConvergenceFixture();
        await fx.A.ExecuteAsync(
            "INSERT INTO orders (id, sku, qty) VALUES (7, 'FIRST', 1) " +
            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty");
        // Second upsert takes the DO UPDATE branch and throws the UNIQUE-constraint SqliteException.
        await fx.A.ExecuteAsync(
            "INSERT INTO orders (id, sku, qty) VALUES (7, 'SECOND', 2) " +
            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty");
    }

    private static async Task<(long Id, string? Sku, long? Qty)> ReadRow(ILocalDb db, long id)
    {
        var rows = await db.QueryAsync(
            "SELECT id, sku, qty FROM orders WHERE id = @id",
            static r => (r.GetInt64(0), r.IsDBNull(1) ? null : r.GetString(1), (long?)(r.IsDBNull(2) ? null : r.GetInt64(2))),
            new { id });
        return rows.Single();
    }

    private static async Task WaitUntilAsync(Func<Task<bool>> predicate, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (await predicate())
                return;
            await Task.Delay(10);
        }
        throw new TimeoutException("Condition not reached within " + timeout);
    }
}