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-002 (scadaproj#37): a snapshot must deliver FK-linked rows whatever order they arrive in.
/// The streamer used to send tables by name and rows by <c>pk_json</c> text, so a child (or a
/// self-referencing row whose parent sorts later) reached the receiver first, failed its immediate
/// FK check, and was dead-lettered — permanently, because the snapshot's watermark and ack covered it.
/// </summary>
public sealed class SnapshotForeignKeyTests : 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, OplogStore Store, SyncSession Session);

    // Parent sorts AFTER child by name ("z_" vs "a_"), and the self-referencing table's child row
    // (pk {"id":10}) sorts BEFORE its parent ({"id":2}) as pk_json text — both orders the old
    // streamer produced.
    private async Task<Side> NewSideAsync(ReplicationOptions? options = null)
    {
        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("CREATE TABLE z_parent (id INTEGER PRIMARY KEY, name TEXT)");
        await db.ExecuteAsync("CREATE TABLE a_child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES z_parent(id))");
        await db.ExecuteAsync("CREATE TABLE node (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES node(id))");
        db.RegisterReplicated("z_parent");
        db.RegisterReplicated("a_child");
        db.RegisterReplicated("node");

        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<long> Count(SqliteLocalDb db, string table) =>
        (await db.QueryAsync($"SELECT COUNT(*) FROM {table}", r => r.GetInt64(0)))[0];

    private static async Task<List<string>> DeadLetters(SqliteLocalDb db) =>
        (await db.QueryAsync("SELECT table_name || ' ' || pk_json || ': ' || 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 FullSnapshot_ToAWipedPeer_DeliversChildrenAndSelfReferences()
    {
        // MaxBatchSize 1 puts every row in its own batch (own transaction), so nothing can hide an
        // ordering problem by landing parent and child in one apply.
        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxBatchSize = 1 };
        var a = await NewSideAsync(options);
        await a.Db.ExecuteAsync("INSERT INTO z_parent (id, name) VALUES (1, 'p')");
        await a.Db.ExecuteAsync("INSERT INTO a_child (id, parent_id) VALUES (10, 1)");
        await a.Db.ExecuteAsync("INSERT INTO node (id, parent_id) VALUES (2, NULL)");
        await a.Db.ExecuteAsync("INSERT INTO node (id, parent_id) VALUES (10, 2)");
        // Everything acked and pruned: a wiped peer can only be healed by a snapshot.
        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());

        var b = await NewSideAsync(options);
        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 b.Db.QueryAsync("SELECT last_applied_remote_seq FROM __localdb_peer_state", r => r.GetInt64(0)))[0] > 0,
            RunTimeout);

        Assert.Empty(await DeadLetters(b.Db));
        Assert.Equal(1, await Count(b.Db, "z_parent"));
        Assert.Equal(1, await Count(b.Db, "a_child"));
        Assert.Equal(2, await Count(b.Db, "node"));

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

    [Fact]
    public async Task Streamer_SendsParentTablesBeforeTheirChildren()
    {
        var a = await NewSideAsync();
        await a.Db.ExecuteAsync("INSERT INTO z_parent (id, name) VALUES (1, 'p')");
        await a.Db.ExecuteAsync("INSERT INTO a_child (id, parent_id) VALUES (10, 1)");
        var streamer = new SnapshotStreamer(a.Db, a.Store, new ReplicationOptions(), NullLogger.Instance);

        var tables = new List<string>();
        await streamer.SendAsync(SnapshotScope.Full, (m, _) =>
        {
            if (m.MsgCase == SyncMessage.MsgOneofCase.SnapshotBatch)
                tables.Add(m.SnapshotBatch.TableName);
            return Task.CompletedTask;
        }, CancellationToken.None);

        Assert.Equal(new[] { "z_parent", "a_child" }, tables);
    }

    // ---- receiver side: an older sender still streams children first --------------------------

    private static SnapshotBatch Batch(string table, params (long Id, string Json)[] rows)
    {
        var batch = new SnapshotBatch { TableName = table };
        foreach (var (id, json) in rows)
            batch.Rows.Add(new SnapshotRow { PkJson = $"{{\"id\":{id}}}", RowJson = json, Hlc = 1000 + id, NodeId = "peer" });
        return batch;
    }

    [Fact]
    public async Task Receiver_AppliesAChildThatArrivesBeforeItsParent()
    {
        var b = await NewSideAsync();
        var store = new OplogStore(b.Db, new ReplicationOptions());
        var applier = new SnapshotApplier(b.Db, new LwwApplier(b.Db), store, NullLogger.Instance);

        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
        await applier.OnBatchAsync(Batch("a_child", (10, "{\"id\":10,\"parent_id\":1}")), default);
        await applier.OnBatchAsync(Batch("node", (10, "{\"id\":10,\"parent_id\":2}")), default);
        await applier.OnBatchAsync(Batch("node", (2, "{\"id\":2,\"parent_id\":null}")), default);
        await applier.OnBatchAsync(Batch("z_parent", (1, "{\"id\":1,\"name\":\"p\"}")), default);
        await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default);

        Assert.Empty(await DeadLetters(b.Db));
        Assert.Equal(1, await Count(b.Db, "a_child"));
        Assert.Equal(2, await Count(b.Db, "node"));
    }

    [Fact]
    public async Task Receiver_DeadLettersATrueOrphan_AtSnapshotComplete()
    {
        var b = await NewSideAsync();
        var store = new OplogStore(b.Db, new ReplicationOptions());
        var applier = new SnapshotApplier(b.Db, new LwwApplier(b.Db), store, NullLogger.Instance);

        await applier.OnBeginAsync(new SnapshotBegin { AsOfSeq = 5 }, default);
        await applier.OnBatchAsync(Batch("a_child", (10, "{\"id\":10,\"parent_id\":99}")), default);
        // Still pending, not yet dead-lettered: its parent could be in a later batch.
        Assert.Empty(await DeadLetters(b.Db));
        await applier.OnCompleteAsync(new SnapshotComplete { AsOfSeq = 5 }, default);

        var dl = Assert.Single(await DeadLetters(b.Db));
        Assert.Contains("FOREIGN KEY", dl, StringComparison.OrdinalIgnoreCase);
        Assert.Equal(0, await Count(b.Db, "a_child"));
    }
}