scadaproj/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/QuotedColumnNameTests.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-006 (scadaproj#46): capture escapes both quote styles, so a column called <c>it's</c> was
/// captured fine — but the apply and snapshot SQL embedded the column in a JSON-path string literal
/// that escaped <c>"</c> and not <c>'</c>. The apostrophe ended the literal early, every row of the
/// table was dead-lettered on the receiver (<c>near "s": syntax error</c>), and a quoted PK column
/// faulted the snapshot page query on every reconnect.
/// </summary>
public sealed class QuotedColumnNameTests : IDisposable
{
    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);

    // Both quote characters, in a PK column and in a non-PK column.
    private const string Ddl = "CREATE TABLE q (\"o'key\" TEXT PRIMARY KEY, \"it's\" TEXT, \"say \"\"hi\"\"\" TEXT)";

    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, OplogStore Store, SyncSession Session);

    private async Task<Side> NewSideAsync()
    {
        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
        _paths.Add(path);
        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
        _disposables.Add(db);
        await db.ExecuteAsync(Ddl);
        db.RegisterReplicated("q");

        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, store, 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<string>> Rows(SqliteLocalDb db) =>
        (await db.QueryAsync(
            "SELECT \"o'key\" || '|' || \"it's\" || '|' || \"say \"\"hi\"\"\" FROM q ORDER BY \"o'key\"",
            r => r.GetString(0))).ToList();

    private static async Task<List<string>> DeadLetters(SqliteLocalDb db) =>
        (await db.QueryAsync("SELECT error FROM __localdb_dead_letter", r => r.GetString(0))).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 Deltas_WithQuotesInColumnNames_ReplicateBothWays()
    {
        var a = await NewSideAsync();
        var b = await NewSideAsync();
        await a.Db.ExecuteAsync("INSERT INTO q VALUES ('k1', 'from A', 'a')");
        await b.Db.ExecuteAsync("INSERT INTO q VALUES ('k2', 'from B', 'b')");

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

        string[] expected = ["k1|from A|a", "k2|from B|b"];
        try
        {
            await WaitForAsync(async () =>
                (await Rows(a.Db)).SequenceEqual(expected) && (await Rows(b.Db)).SequenceEqual(expected), RunTimeout);
        }
        catch (TimeoutException)
        {
            Assert.Fail($"Not converged. A dead-letters: [{string.Join("; ", await DeadLetters(a.Db))}], " +
                        $"B dead-letters: [{string.Join("; ", await DeadLetters(b.Db))}]");
        }

        // An update and a delete exercise the PK-keyed json_extract paths too.
        await a.Db.ExecuteAsync("UPDATE q SET \"it's\" = 'changed' WHERE \"o'key\" = 'k2'");
        await b.Db.ExecuteAsync("DELETE FROM q WHERE \"o'key\" = 'k1'");
        await WaitForAsync(async () =>
            (await Rows(a.Db)).SequenceEqual(["k2|changed|b"]) && (await Rows(b.Db)).SequenceEqual(["k2|changed|b"]),
            RunTimeout);

        Assert.Empty(await DeadLetters(a.Db));
        Assert.Empty(await DeadLetters(b.Db));

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

    [Fact]
    public async Task Snapshot_WithAQuotedPrimaryKeyColumn_ReachesAWipedPeer()
    {
        var a = await NewSideAsync();
        await a.Db.ExecuteAsync("INSERT INTO q VALUES ('k1', 'one', 'x')");
        await a.Db.ExecuteAsync("INSERT INTO q VALUES ('k2', 'two', 'y')");
        // Acked and pruned: a wiped peer can only be healed by a snapshot, whose page query joins
        // the ledger to the table through the quoted PK column.
        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());

        var b = await NewSideAsync();
        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 Rows(b.Db)).Count == 2 || runA.IsCompleted, RunTimeout);

        Assert.False(runA.IsFaulted, runA.Exception?.ToString());
        Assert.Equal(new[] { "k1|one|x", "k2|two|y" }, await Rows(b.Db));
        Assert.Empty(await DeadLetters(b.Db));

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