scadaproj/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/ApplyUpsertTests.cs

using System.Threading.Channels;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Internal;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.LocalDb.Replication.Internal;

namespace ZB.MOM.WW.LocalDb.Tests;

/// <summary>
/// LocalDb-003 (scadaproj#38): the apply path upserts on the PRIMARY KEY only. <c>INSERT OR REPLACE</c>
/// also resolved a secondary <c>UNIQUE</c> conflict by deleting the OTHER row — a different primary
/// key, with no capture, no ledger change and nothing shipped — so a pair diverged silently.
/// </summary>
public sealed class ApplyUpsertTests : IDisposable
{
    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);

    private readonly List<string> _paths = [];
    private readonly List<IDisposable> _disposables = [];

    public void Dispose()
    {
        foreach (var d in _disposables)
            d.Dispose();
        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
        foreach (var p in _paths)
        {
            if (File.Exists(p)) File.Delete(p);
            if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
            if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
        }
    }

    private sealed record Side(SqliteLocalDb Db, SyncSession Session);

    private async Task<SqliteLocalDb> NewDbAsync(params string[] ddl)
    {
        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
        _paths.Add(path);
        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
        _disposables.Add(db);
        foreach (var statement in ddl)
            await db.ExecuteAsync(statement);
        return db;
    }

    private static Side SessionOver(SqliteLocalDb db)
    {
        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
        var store = new OplogStore(db, options);
        var applier = new LwwApplier(db);
        var session = new SyncSession(db, store, applier, options, NullLogger.Instance);
        session.SnapshotSender = new SnapshotStreamer(db, store, options, NullLogger.Instance).SendAsync;
        session.SnapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance);
        return new Side(db, session);
    }

    private static (SyncDuplex A, SyncDuplex B) DuplexPair()
    {
        var aToB = Channel.CreateUnbounded<SyncMessage>();
        var bToA = Channel.CreateUnbounded<SyncMessage>();
        return (
            new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader },
            new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader });
    }

    private static async Task<List<long>> Ids(SqliteLocalDb db, string table) =>
        (await db.QueryAsync($"SELECT id FROM {table} ORDER BY id", r => r.GetInt64(0))).ToList();

    private static async Task<long> DeadLetterCount(SqliteLocalDb db) =>
        (await db.QueryAsync("SELECT COUNT(*) FROM __localdb_dead_letter", r => r.GetInt64(0)))[0];

    private static async Task<List<string>> LiveLedger(SqliteLocalDb db, string table) =>
        (await db.QueryAsync(
            "SELECT pk_json FROM __localdb_row_version WHERE table_name = @t AND is_tombstone = 0 ORDER BY pk_json",
            r => r.GetString(0), new { t = table })).ToList();

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

    private static async Task SwallowAsync(Task task)
    {
        try { await task; }
        catch (OperationCanceledException) { }
    }

    [Fact]
    public async Task SecondaryUniqueConflict_DoesNotDeleteAnotherRow_AndIsDeadLettered()
    {
        const string ddl = "CREATE TABLE t (id INTEGER PRIMARY KEY, code TEXT UNIQUE)";
        var dbA = await NewDbAsync(ddl);
        var dbB = await NewDbAsync(ddl);
        dbA.RegisterReplicated("t");
        dbB.RegisterReplicated("t");

        // The same natural key created on each node during a partition.
        await dbA.ExecuteAsync("INSERT INTO t (id, code) VALUES (1, 'X')");
        await dbB.ExecuteAsync("INSERT INTO t (id, code) VALUES (2, 'X')");

        var a = SessionOver(dbA);
        var b = SessionOver(dbB);
        var (da, dbx) = DuplexPair();
        using var cts = new CancellationTokenSource(RunTimeout);
        var runA = a.Session.RunAsync(da, cts.Token);
        var runB = b.Session.RunAsync(dbx, cts.Token);

        await WaitForAsync(async () => await DeadLetterCount(dbA) == 1 && await DeadLetterCount(dbB) == 1, RunTimeout);

        // Each node keeps its own row: nothing was deleted behind the ledger's back, and the
        // conflict is visible as a dead-letter instead of a silent loss.
        Assert.Equal(new[] { 1L }, await Ids(dbA, "t"));
        Assert.Equal(new[] { 2L }, await Ids(dbB, "t"));
        Assert.Equal(new[] { "{\"id\":1}" }, await LiveLedger(dbA, "t"));
        Assert.Equal(new[] { "{\"id\":2}" }, await LiveLedger(dbB, "t"));

        cts.Cancel();
        await SwallowAsync(runA);
        await SwallowAsync(runB);
    }

    [Fact]
    public async Task RemoteUpdate_OfAParentWithRestrictChildren_Applies()
    {
        // REPLACE deleted the parent before re-inserting it, which an ON DELETE RESTRICT child
        // refuses outright — so every remote update of such a parent was dead-lettered. An upsert
        // updates in place.
        string[] ddl =
        [
            "CREATE TABLE parent (id INTEGER PRIMARY KEY, name TEXT)",
            "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id) ON DELETE RESTRICT)",
        ];
        var db = await NewDbAsync(ddl);
        db.RegisterReplicated("parent");
        db.RegisterReplicated("child");
        await db.ExecuteAsync("INSERT INTO parent (id, name) VALUES (1, 'old')");
        await db.ExecuteAsync("INSERT INTO child (id, parent_id) VALUES (10, 1)");

        var result = await new LwwApplier(db).ApplyBatchAsync(
            [new OplogEntryRecord(1, "parent", "{\"id\":1}", "{\"id\":1,\"name\":\"new\"}", long.MaxValue / 2, "peer", false)],
            CancellationToken.None);

        Assert.Equal(1, result.Applied);
        Assert.Equal(0, result.DeadLettered);
        var name = (await db.QueryAsync("SELECT name FROM parent WHERE id = 1", r => r.GetString(0)))[0];
        Assert.Equal("new", name);
    }

    [Fact]
    public async Task TableWhosePrimaryKeyIsEveryColumn_StillApplies()
    {
        var db = await NewDbAsync("CREATE TABLE link (a INTEGER, b INTEGER, PRIMARY KEY (a, b))");
        db.RegisterReplicated("link");

        var applier = new LwwApplier(db);
        var entry = new OplogEntryRecord(1, "link", "{\"a\":1,\"b\":2}", "{\"a\":1,\"b\":2}", 1000, "peer", false);
        Assert.Equal(1, (await applier.ApplyBatchAsync([entry], CancellationToken.None)).Applied);
        // Re-applying a newer version of the same key is a conflict on the PK with nothing to update.
        Assert.Equal(1, (await applier.ApplyBatchAsync([entry with { Hlc = 2000 }], CancellationToken.None)).Applied);
        Assert.Equal(1L, (await db.QueryAsync("SELECT COUNT(*) FROM link", r => r.GetInt64(0)))[0]);
    }
}