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));
    }

    // Regression for the former product bug where a consumer `INSERT ... ON CONFLICT(id) DO UPDATE`
    // crashed the au capture trigger: SQLite replaces a trigger-body statement's OR-conflict
    // algorithm with the OUTER statement's conflict handling, so the trigger's row_version
    // `INSERT OR REPLACE` degraded to a plain INSERT -> SQLITE_CONSTRAINT_PRIMARYKEY (1555).
    // Fixed by an explicit ON CONFLICT DO UPDATE clause in the capture triggers (not overridden).
    [Fact]
    public async Task Upsert_OnConflictDoUpdate_CapturesCorrectly()
    {
        await using var fx = new ConvergenceFixture();
        const string upsert =
            "INSERT INTO orders (id, sku, qty) VALUES (7, @sku, @qty) " +
            "ON CONFLICT(id) DO UPDATE SET sku = excluded.sku, qty = excluded.qty";

        await fx.A.ExecuteAsync(upsert, new { sku = "FIRST", qty = 1 });  // insert branch -> ai
        await fx.A.ExecuteAsync(upsert, new { sku = "SECOND", qty = 2 }); // DO-UPDATE branch -> au

        // Both branches captured, in order, with full-row payloads.
        var oplog = await fx.A.QueryAsync(
            "SELECT row_json, hlc, is_tombstone FROM __localdb_oplog WHERE table_name='orders' ORDER BY seq",
            static r => (RowJson: r.GetString(0), Hlc: r.GetInt64(1), Tomb: r.GetInt64(2)));
        Assert.Equal(2, oplog.Count);
        Assert.Equal("{\"id\":7,\"sku\":\"FIRST\",\"qty\":1}", oplog[0].RowJson);
        Assert.Equal("{\"id\":7,\"sku\":\"SECOND\",\"qty\":2}", oplog[1].RowJson);
        Assert.All(oplog, o => Assert.Equal(0, o.Tomb));

        // Single row_version entry for the pk, correlated to the LATEST (au-branch) oplog hlc.
        var rv = await fx.A.QueryAsync(
            "SELECT pk_json, hlc, is_tombstone FROM __localdb_row_version WHERE table_name='orders'",
            static r => (PkJson: r.GetString(0), Hlc: r.GetInt64(1), Tomb: r.GetInt64(2)));
        var v = Assert.Single(rv);
        Assert.Equal("{\"id\":7}", v.PkJson);
        Assert.Equal(0, v.Tomb);
        Assert.Equal(oplog[1].Hlc, v.Hlc);

        // And the captured upsert replicates end-to-end.
        await fx.StartAsync();
        await fx.AssertConvergedAsync();
        Assert.Equal((7L, "SECOND", (long?)2), await ReadRow(fx.B, 7));
    }

    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);
    }
}