using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Internal;
using ZB.MOM.WW.LocalDb.Registration;

namespace ZB.MOM.WW.LocalDb.Replication.Internal;

/// <summary>
/// Consumes an inbound full-snapshot the receive loop routes to it: SnapshotBegin opens a receive,
/// each SnapshotBatch merges through the SAME last-writer-wins path as deltas (never truncating —
/// the receiver may hold newer local writes), and SnapshotComplete advances the durable watermark
/// so incremental deltas resume above it.
/// </summary>
internal interface ISnapshotApplier
{
    Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct);
    Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct);
    Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct);
}

/// <summary>
/// Which tables a snapshot covers. An EMPTY <see cref="Tables"/> is the wire's own encoding of
/// "all registered tables" (see <c>SnapshotBegin.tables</c>) and is what every pre-0.3.0 peer both
/// sends and expects, so it is deliberately the same value here rather than a separate flag.
/// </summary>
/// <remarks>
/// The distinction is not cosmetic: a FULL snapshot carries every row state as of its
/// <c>as_of_seq</c>, so the receiver may jump its watermark there and the sender may skip the
/// deltas it covered. A PARTIAL one carries only the listed tables, so neither side may move a
/// watermark that also speaks for the tables it left out.
/// </remarks>
internal sealed record SnapshotScope(IReadOnlyList<string> Tables)
{
    /// <summary>Every registered table — the only scope a pre-0.3.0 peer can be sent.</summary>
    public static readonly SnapshotScope Full = new([]);

    public bool IsPartial => Tables.Count > 0;

    public bool Includes(string tableName) =>
        Tables.Count == 0 || Tables.Contains(tableName, StringComparer.Ordinal);
}

/// <summary>
/// Streams a snapshot of the tables in the requested <see cref="SnapshotScope"/> — every registered
/// table for a peer that has fallen behind the pruned oplog horizon, or just the baselined ones for
/// a scoped resync. Reads the authoritative per-row version from <c>__localdb_row_version</c> joined
/// to each live table, pages it into <see cref="SnapshotBatch"/> messages through the gated data
/// lane, and brackets them with SnapshotBegin/SnapshotComplete carrying the max oplog seq as-of.
/// </summary>
/// <remarks>
/// Clearing the snapshot debt is deliberately NOT done here: whether the flags may be cleared at all
/// depends on the peer's capability (a 0.3.0 peer acknowledges the snapshot, a pre-0.3.0 one cannot),
/// which is session state. <see cref="SyncSession"/> owns that decision; this type only streams.
/// </remarks>
internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, ReplicationOptions options, ILogger logger)
{
    /// <summary>
    /// Streams the snapshot for <paramref name="scope"/>; returns the as-of oplog seq it covers, so
    /// the session can skip already-covered tail deltas (a FULL snapshot only — a partial one speaks
    /// for nothing outside its tables).
    /// </summary>
    public async Task<long> SendAsync(SnapshotScope scope, Func<SyncMessage, CancellationToken, Task> send, CancellationToken ct)
    {
        // The as-of is the highest seq this snapshot can stand in for. MAX(seq) alone is wrong once
        // ack-pruning has emptied the oplog: it reads 0, and a receiver whose watermark genuinely IS
        // 0 (a rebuilt node) is then never lifted by the snapshot — its MAX-guarded watermark stays
        // 0, the empty-oplog horizon rule (streamable from last_acked_seq + 1) sees a gap on the
        // next handshake, and the whole database is re-streamed on every reconnect until this node
        // happens to write again. Every delta pruned below last_acked_seq is already folded into the
        // rows this snapshot carries, so last_acked_seq is the floor the as-of must carry.
        var asOfSeq = Math.Max(
            await store.GetMaxSeqAsync(ct),
            (await store.GetPeerStateAsync(ct)).LastAckedSeq);
        var begin = new SnapshotBegin { AsOfSeq = asOfSeq };
        begin.Tables.AddRange(scope.Tables);
        await send(new SyncMessage { SnapshotBegin = begin }, ct);

        var totalRows = 0L;
        foreach (var table in db.ReplicatedTables.Values
                     .Where(t => scope.Includes(t.Name))
                     .OrderBy(t => t.Name, StringComparer.Ordinal))
        {
            var sql = BuildPageSql(table);
            // Keyset cursor (pk_json > last seen), not LIMIT/OFFSET: concurrent inserts/prunes
            // between pages cannot shift the window, so no ledger row is skipped or re-sent.
            var lastPk = "";
            while (true)
            {
                var rows = await db.QueryAsync(
                    sql,
                    // Column 4 is NULL for tombstones AND for live rows whose joined table row is
                    // missing; a proto optional string cannot be assigned null, so carry it as a
                    // nullable and only set row_json when present.
                    static r => (
                        PkJson: r.GetString(0),
                        Hlc: r.GetInt64(1),
                        NodeId: r.GetString(2),
                        IsTombstone: r.GetInt64(3) != 0,
                        RowJson: r.IsDBNull(4) ? null : r.GetString(4)),
                    new { t = table.Name, page = options.MaxBatchSize, lastPk }, ct);
                if (rows.Count == 0)
                    break;
                lastPk = rows[^1].PkJson;

                // One DB page may span several messages: the page is bounded by ROW COUNT, but a
                // message is bounded by BYTES (ReplicationOptions.MaxBatchBytes) — the same budget
                // the delta pump honours, and for the same reason. Wide rows are exactly why a
                // snapshot is the easiest way to exceed the transport's message limit: it ships
                // whole rows, all of them, back to back.
                var batch = new SnapshotBatch { TableName = table.Name };
                var bytes = 0L;
                foreach (var row in rows)
                {
                    if (!row.IsTombstone && row.RowJson is null)
                    {
                        // Ledger says live but the LEFT JOIN found no table row: ledger/table
                        // divergence (e.g. an unreplicated manual delete). Emitting {"col":null,...}
                        // would fabricate a phantom row on the peer, so skip it defensively.
                        logger.LogWarning(
                            "Snapshot skipping live-ledger row {Table} {PkJson}: no matching table row (ledger/table divergence).",
                            batch.TableName, row.PkJson);
                        continue;
                    }
                    var snapshotRow = new SnapshotRow
                    {
                        PkJson = row.PkJson,
                        Hlc = row.Hlc,
                        NodeId = row.NodeId,
                        IsTombstone = row.IsTombstone,
                    };
                    if (row.RowJson is not null)
                        snapshotRow.RowJson = row.RowJson;

                    var size = RowSize(snapshotRow);
                    // Flush before adding, never after: a message keeps at least one row however
                    // large, so an oversized row goes out alone rather than stalling the snapshot.
                    if (batch.Rows.Count > 0 && bytes + size > options.MaxBatchBytes)
                    {
                        await send(new SyncMessage { SnapshotBatch = batch }, ct);
                        totalRows += batch.Rows.Count;
                        batch = new SnapshotBatch { TableName = table.Name };
                        bytes = 0;
                    }
                    batch.Rows.Add(snapshotRow);
                    bytes += size;
                }
                if (batch.Rows.Count > 0)
                {
                    await send(new SyncMessage { SnapshotBatch = batch }, ct);
                    totalRows += batch.Rows.Count;
                }

                if (rows.Count < options.MaxBatchSize)
                    break;
            }
        }

        await send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = asOfSeq } }, ct);
        logger.LogInformation(
            "Snapshot sent (as_of_seq {AsOfSeq}, {Rows} rows, scope {Scope}).",
            asOfSeq, totalRows, scope.IsPartial ? string.Join(",", scope.Tables) : "all");
        return asOfSeq;
    }

    /// <summary>
    /// Serialized bytes one row contributes to its batch message: the row's own encoding plus the
    /// repeated field's tag and length prefix, as protobuf itself computes them.
    /// </summary>
    private static long RowSize(SnapshotRow row) =>
        1 + Google.Protobuf.CodedOutputStream.ComputeMessageSize(row);

    // Pages one table's rows by joining the row-version ledger to the live table on the PK columns
    // extracted from pk_json. Rows present in the live table but absent from __localdb_row_version
    // (data written BEFORE the table was registered for replication) are NOT snapshotted: they were
    // never captured, consistent with CDC semantics.
    //
    // That is a real hole for a node that starts replicating LATE — its whole pre-existing dataset is
    // invisible to the peer, silently and permanently — and the supported way to close it is to
    // register with `ILocalDb.RegisterReplicated(table, baselineExistingRows: true)`, which seeds the
    // ledger at an LWW-safe floor and flags the resync that carries the rows here. Nothing changes on
    // this side: baselined rows are ordinary ledger rows and page like any other.
    private static string BuildPageSql(ReplicatedTable table)
    {
        var name = Ident(table.Name);
        var join = string.Join(" AND ", table.PkColumns.Select(p => $"t.{Ident(p)} = json_extract(rv.pk_json, {JsonPath(p)})"));
        var jsonObject = "json_object(" +
            string.Join(", ", table.Columns.Select(c => $"{StringLiteral(c.Name)}, t.{Ident(c.Name)}")) + ")";
        // A live ledger row whose join found no table row (first PK column NULL — PK columns are
        // never NULL on a real join hit) also yields NULL row json, so the caller can detect the
        // divergence instead of shipping a fabricated all-NULL row.
        var joinMiss = $"t.{Ident(table.PkColumns[0])} IS NULL";
        return
            "SELECT rv.pk_json, rv.hlc, rv.node_id, rv.is_tombstone, " +
            $"CASE WHEN rv.is_tombstone = 1 OR {joinMiss} THEN NULL ELSE {jsonObject} END " +
            $"FROM __localdb_row_version rv LEFT JOIN {name} t ON {join} " +
            "WHERE rv.table_name = @t AND rv.pk_json > @lastPk ORDER BY rv.pk_json LIMIT @page";
    }

    private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";

    private static string StringLiteral(string value) => "'" + value.Replace("'", "''") + "'";

    // A single-quoted JSON path with a quoted key member so a column name with reserved characters
    // still resolves: $."col".
    private static string JsonPath(string column) => "'$.\"" + column.Replace("\"", "\\\"") + "\"'";
}

/// <summary>
/// The inbound-snapshot consumer wired into <see cref="SyncSession"/>. Each batch merges through the
/// shared <see cref="LwwApplier"/> (seq 0 — snapshot rows carry no oplog seq — so the applier's
/// MAX-guarded watermark update is a no-op, and per-row LWW protects any newer local write). On
/// complete, the remote watermark is set to the snapshot's as-of seq so deltas resume above it.
/// </summary>
internal sealed class SnapshotApplier : ISnapshotApplier
{
    private readonly LwwApplier _applier;
    private readonly OplogStore _store;
    private readonly ILogger _logger;
    private long _asOfSeq;
    private long _rows;
    private long _maxHlc;
    private bool _partial;

    // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge
    // reaches the database through the applier and store; kept for construction symmetry in the factory.
    public SnapshotApplier(SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger)
    {
        _ = db;
        _applier = applier;
        _store = store;
        _logger = logger;
    }

    public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct)
    {
        _asOfSeq = begin.AsOfSeq;
        _rows = 0;
        _maxHlc = 0;
        // A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can
        // emit) means every registered table. Read here rather than on complete because
        // SnapshotComplete does not repeat it — the receive loop guarantees a Begin precedes both.
        _partial = begin.Tables.Count > 0;
        _logger.LogInformation(
            "Snapshot receive begun (as_of_seq {AsOfSeq}, scope {Scope}).",
            _asOfSeq, _partial ? string.Join(",", begin.Tables) : "all");
        return Task.CompletedTask;
    }

    public async Task OnBatchAsync(SnapshotBatch batch, CancellationToken ct)
    {
        if (batch.Rows.Count == 0)
            return;

        var entries = new List<OplogEntryRecord>(batch.Rows.Count);
        foreach (var row in batch.Rows)
        {
            if (row.Hlc > _maxHlc) _maxHlc = row.Hlc;
            entries.Add(new OplogEntryRecord(
                0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone));
        }
        await _applier.ApplyBatchAsync(entries, ct);
        _rows += batch.Rows.Count;
    }

    public async Task OnCompleteAsync(SnapshotComplete complete, CancellationToken ct)
    {
        // A PARTIAL snapshot may not move the inbound seq watermark. That watermark is one number
        // for the whole peer oplog, and jumping it to as_of_seq would declare every table applied
        // through that point — including the ones this snapshot deliberately left out, whose deltas
        // below as_of_seq the sender would then never resend. Only the HLC is carried across (seq 0
        // is a no-op against the MAX guard), so the clock still cannot regress below what the peer
        // has stamped. The sender makes the mirror-image concession: it does not advance its own
        // sent-thru watermark past a partial snapshot, so those deltas are still pumped normally
        // and merely re-deliver rows the snapshot already carried, which LWW absorbs.
        var seq = _partial ? 0 : complete.AsOfSeq;
        // SetLastAppliedRemoteSeqAsync is MAX-guarded, so a delta applied mid-snapshot with a seq
        // above the snapshot's as-of (or an HLC above what the snapshot carried) is never regressed.
        await _store.SetLastAppliedRemoteSeqAsync(seq, _maxHlc, ct);
        _logger.LogInformation(
            "Snapshot receive complete (as_of_seq {AsOfSeq}, {Rows} rows, partial {Partial}).",
            complete.AsOfSeq, _rows, _partial);
    }
}