using System.Globalization;
using System.Threading.Channels;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Hlc;
using ZB.MOM.WW.LocalDb.Internal;

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

/// <summary>
/// The symmetric 2-node sync protocol as one state machine: both the gRPC service and the gRPC
/// client run this identical logic over a <see cref="SyncDuplex"/>. It performs the fail-closed
/// handshake, then runs a bidirectional steady state until the token is cancelled or the stream
/// ends: an outbound phase (the full snapshot this node owes, if any, then the change pump) and an
/// inbound apply/ack loop, STARTED TOGETHER so the inbox keeps draining even while a snapshot is
/// being sent. All outbound traffic flows through a single writer task fed by two lanes: an
/// unbounded CONTROL lane (handshake messages, acks) and a bounded DATA lane (delta/snapshot
/// batches). Receiving therefore never blocks on the ability to write — acks always fit — while the
/// bounded data lane is the pump's backpressure.
/// Local writes never block on the peer; the first fault tears down all loops and propagates.
/// </summary>
internal sealed class SyncSession
{
    private const uint LibSchemaVersion = 1;

    // Data-lane depth bounds outbound batch buffering; control-lane volume is bounded by the
    // inbound batch rate (the peer's flow control), so its lane can safely be unbounded.
    private const int DataLaneCapacity = 16;

    // How long a faulting session's teardown lets the writer keep flushing already-enqueued
    // messages before the backstop cancel; only ever reached when the transport itself stalls.
    private static readonly TimeSpan FaultFlushGrace = TimeSpan.FromSeconds(2);

    private readonly SqliteLocalDb _db;
    private readonly OplogStore _store;
    private readonly LwwApplier _applier;
    private readonly ReplicationOptions _options;
    private readonly ILogger _logger;
    private readonly Func<DateTimeOffset> _utcNow;
    private readonly LocalDbMetrics? _metrics;
    private readonly SyncStatus? _status;

    // Highest oplog seq handed to the wire this session; the pump reads strictly above it so a batch
    // is never re-sent before its ack (the reliable ordered stream delivers exactly once in-session).
    private long _sentThruSeq;
    private long _peerAckedSeq;
    private string? _peerNodeId;
    private bool _driftWarned;

    // Set true between an inbound SnapshotBegin and its SnapshotComplete. Only ever touched on the
    // single receive-loop task, so no synchronization is needed.
    private bool _snapshotReceiving;

    // Whether the peer speaks the 0.3.0 snapshot extensions: scoped snapshots
    // (SnapshotBegin.tables) and the SnapshotAck that answers a completed one. Written once during
    // the handshake — which completes before either loop starts — then read-only.
    private bool _peerSupportsPartialSnapshot;

    // The scope of the snapshot this node is streaming, held until the peer's SnapshotAck discharges
    // it; null on every session that owes none. Assigned on the pump task strictly BEFORE the
    // snapshot's first message is enqueued and read on the receive loop when the ack lands, so the
    // ordering is causal — volatile only to make the cross-thread publication explicit.
    private SnapshotScope? _sentSnapshotScope;

    public SyncSession(
        SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger,
        Func<DateTimeOffset>? utcNow = null, LocalDbMetrics? metrics = null, SyncStatus? status = null)
    {
        _db = db;
        _store = store;
        _applier = applier;
        _options = options;
        _logger = logger;
        _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
        _metrics = metrics;
        _status = status;
    }

    /// <summary>
    /// Invoked when THIS node owes the peer a snapshot (its own snapshot-required computation, not
    /// the peer's) before deltas resume. Receives the negotiated <see cref="SnapshotScope"/> and a
    /// gated send delegate that enqueues on the bounded data lane (never the raw transport); returns
    /// the as-of oplog seq the snapshot covers so the pump can skip already-covered tail deltas.
    /// Wired to <see cref="SnapshotStreamer.SendAsync"/>.
    /// </summary>
    public Func<SnapshotScope, Func<SyncMessage, CancellationToken, Task>, CancellationToken, Task<long>>? SnapshotSender { get; set; }

    /// <summary>
    /// Consumes an inbound snapshot inline on the receive loop: SnapshotBegin opens the receive,
    /// SnapshotBatch merges through LWW, SnapshotComplete advances the watermark. Wired to
    /// <see cref="Internal.SnapshotApplier"/>.
    /// </summary>
    public ISnapshotApplier? SnapshotApplier { get; set; }

    /// <summary>Highest seq the peer has acknowledged applying this session.</summary>
    public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq);

    /// <summary>Highest oplog seq handed to the wire (or covered by a sent snapshot). Test/metric observation only.</summary>
    internal long SentThruSeq => Interlocked.Read(ref _sentThruSeq);

    /// <summary>The peer node's identity, once the handshake has completed. Volatile: written on the session task, read from test/metric threads.</summary>
    public string? PeerNodeId
    {
        get => Volatile.Read(ref _peerNodeId);
        private set => Volatile.Write(ref _peerNodeId, value);
    }

    public async Task RunAsync(SyncDuplex duplex, CancellationToken ct)
    {
        var control = Channel.CreateUnbounded<SyncMessage>(new UnboundedChannelOptions { SingleReader = true });
        var data = Channel.CreateBounded<SyncMessage>(new BoundedChannelOptions(DataLaneCapacity) { SingleReader = true });
        Task GatedDataSend(SyncMessage message, CancellationToken token) => data.Writer.WriteAsync(message, token).AsTask();

        using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
        var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token);
        try
        {
            var snapshotScope = await HandshakeAsync(duplex, control.Writer, linked.Token);

            // The receive loop starts BEFORE any snapshot this node owes is streamed. That ordering
            // is load-bearing, not incidental: a snapshot sent from inside the handshake left the
            // inbox undrained for its whole duration, so a MUTUAL snapshot (both sides flagged —
            // e.g. both baselined an existing dataset) required an unbounded inbox to avoid
            // deadlock, and therefore buffered a whole peer dataset in memory. With the receive loop
            // running throughout, the peer's snapshot is applied as it arrives, the inbox can be
            // bounded (see GrpcSyncDuplex.InboxCapacity), and mutual snapshots exchange under
            // ordinary flow control. Interleaving is already safe by design: snapshot rows and
            // deltas both merge through LWW, and the snapshot's watermark write is MAX-guarded.
            var receive = ReceiveLoopAsync(duplex, control.Writer, linked.Token);
            var pump = SnapshotThenPumpLoopAsync(snapshotScope, GatedDataSend, data.Writer, linked.Token);
            var loops = new[] { pump, receive, writer };

            // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every
            // one of them under a swallow (so no secondary exception can replace the primary and
            // nothing is left running/unobserved), then surface the primary by awaiting it.
            var finished = await Task.WhenAny(loops);
            linked.Cancel();
            foreach (var loop in loops)
                if (!ReferenceEquals(loop, finished))
                    await DrainSecondaryAsync(loop);
            await finished;
        }
        catch
        {
            // Handshake-phase faults (and the primary rethrow above) land here: the writer may
            // still be running, so stop and observe it before the primary exception surfaces.
            // It must be allowed to DRAIN, not be cancelled outright: the Handshake enqueued at
            // the top of HandshakeAsync can still be sitting in the control lane (the enqueue →
            // receive → validate → fault path has no yields, while the writer's wakeup waits in
            // the thread-pool queue), and that message is the only thing that lets the peer
            // compute its own fail-closed verdict — mismatch detection is symmetric, so this
            // node's locally-thrown verdict reaches the peer only as the Handshake it transmits.
            // Cancelling first killed it almost every time the peer's handshake was already
            // buffered, leaving the healthy peer to misread a schema mismatch as a generic
            // disconnect and reconnect-loop at the fast backoff. Completing the lanes lets the
            // writer send what is already enqueued and exit on its own; the deferred cancel is
            // the backstop that keeps a wedged transport from holding teardown hostage.
            control.Writer.TryComplete();
            data.Writer.TryComplete();
            linked.CancelAfter(FaultFlushGrace);
            await DrainSecondaryAsync(writer);
            throw;
        }
    }

    /// <summary>
    /// Runs the two-phase fail-closed handshake and returns the snapshot THIS node owes the peer —
    /// null for none, otherwise the <see cref="SnapshotScope"/> to stream (its own computation from
    /// its own state; the peer's answer drives the peer's send direction, symmetrically). Sending
    /// that snapshot is deliberately NOT part of the handshake; see <see cref="RunAsync"/>.
    /// </summary>
    private async Task<SnapshotScope?> HandshakeAsync(
        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
    {
        var peerState = await _store.GetPeerStateAsync(ct);
        var localHandshake = BuildHandshake(peerState);

        // Both peers send before reading, so the transport must buffer at least two in-flight
        // messages (Handshake + HandshakeAck); gRPC's initial flow-control window covers that.
        await controlLane.WriteAsync(new SyncMessage { Handshake = localHandshake }, ct);

        var firstInbound = await ReceiveOneAsync(duplex, ct);
        if (firstInbound.MsgCase != SyncMessage.MsgOneofCase.Handshake)
            throw new InvalidOperationException(
                $"Replication protocol error: expected a Handshake as the first message, got {firstInbound.MsgCase}.");
        var peerHandshake = firstInbound.Handshake;

        ValidateHandshake(localHandshake, peerHandshake);

        // Read before anything can act on it. A pre-0.3.0 peer leaves the field at its proto3
        // default, so "false" covers both "explicitly unsupported" and "older build" without a
        // version check — which is the point of advertising a capability rather than a version.
        _peerSupportsPartialSnapshot = peerHandshake.SupportsPartialSnapshot;

        // A different node id on the far end means the peer was replaced, not merely restarted — a
        // rebuilt host, a restored-from-nothing volume. Its oplog numbers from 1 again, so the
        // watermark we hold (in the OLD peer's seq space) would silently swallow its first N
        // writes. Forget it before anything reads it.
        if (peerState.PeerNodeId is not null && peerState.PeerNodeId != peerHandshake.NodeId)
        {
            _logger.LogWarning(
                "Replication peer identity changed ({OldPeer} -> {NewPeer}): the peer was rebuilt, so its seq space restarted. Resetting the inbound watermark.",
                peerState.PeerNodeId, peerHandshake.NodeId);
            await _store.ResetInboundWatermarkAsync(ct);
            peerState = peerState with { LastAppliedRemoteSeq = 0, LastSeenHlc = 0 };
        }

        await _store.SetPeerNodeIdAsync(peerHandshake.NodeId, ct);
        PeerNodeId = peerHandshake.NodeId;
        _status?.SetPeerNodeId(peerHandshake.NodeId);
        // The peer's watermark decides where our pump starts, so an impossible claim silently
        // skips real data. It can only exceed what we have ever produced if the peer is remembering
        // a previous incarnation of THIS node — i.e. WE were rebuilt and our seqs restarted at 1 —
        // in which case its number says nothing about the stream we hold now, and starting from it
        // would skip our entire oplog. Fall back to the beginning; LWW makes re-sending harmless.
        var knownMaxSeq = await _store.GetKnownMaxSeqAsync(ct);
        var peerAppliedThruSeq = peerHandshake.LastAppliedRemoteSeq <= knownMaxSeq
            ? peerHandshake.LastAppliedRemoteSeq
            : 0;
        if (peerAppliedThruSeq != peerHandshake.LastAppliedRemoteSeq)
            _logger.LogWarning(
                "Peer claims to have applied our seq {PeerClaim}, beyond the {KnownMax} we have ever produced — this node was rebuilt. Re-sending our oplog from the start.",
                peerHandshake.LastAppliedRemoteSeq, knownMaxSeq);

        _sentThruSeq = peerAppliedThruSeq;
        Interlocked.Exchange(ref _peerAckedSeq, peerAppliedThruSeq);

        var snapshotScope = await ComputeSnapshotScopeAsync(peerHandshake, peerState, ct);
        await controlLane.WriteAsync(
            new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotScope is not null } }, ct);

        var peerAckMsg = await ReceiveOneAsync(duplex, ct);
        if (peerAckMsg.MsgCase != SyncMessage.MsgOneofCase.HandshakeAck)
            throw new InvalidOperationException(
                $"Replication protocol error: expected a HandshakeAck, got {peerAckMsg.MsgCase}.");

        return snapshotScope;
    }

    /// <summary>
    /// The outbound steady state: stream the full snapshot this node owes the peer (if any), then
    /// run the incremental delta pump. One task, because the ordering between them is a protocol
    /// requirement — the pump must not put deltas on the data lane ahead of the snapshot, and it
    /// starts from the as-of seq the snapshot covered.
    /// </summary>
    /// <remarks>
    /// Split out of the handshake so the receive loop is already draining the inbox while this
    /// streams (see <see cref="RunAsync"/>). Nothing else about the snapshot changes: it goes
    /// through the same gated data lane, so it is still bounded by the writer's flow control rather
    /// than racing the transport.
    /// </remarks>
    private async Task SnapshotThenPumpLoopAsync(
        SnapshotScope? snapshotScope, Func<SyncMessage, CancellationToken, Task> dataSend,
        ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
    {
        if (snapshotScope is not null)
        {
            if (SnapshotSender is null)
                throw new NotSupportedException("snapshot required but no snapshot sender configured");
            // Published BEFORE the first frame is enqueued, so the peer's SnapshotAck can never
            // arrive on the receive loop while this is still null.
            Volatile.Write(ref _sentSnapshotScope, snapshotScope);
            var snapshotAsOfSeq = await SnapshotSender(snapshotScope, dataSend, ct);
            _metrics?.RecordSnapshotSent();
            _status?.SetLastSyncUtc(_utcNow());
            // A FULL snapshot carries every row state up to as-of, so skipping the pump past it
            // avoids re-sending tail deltas the peer would only discard via LWW. A PARTIAL one
            // speaks only for its own tables, so the pump must still deliver every delta below
            // as-of — the peer has not moved its watermark either (see SnapshotApplier.OnComplete).
            if (!snapshotScope.IsPartial && snapshotAsOfSeq > _sentThruSeq)
                _sentThruSeq = snapshotAsOfSeq;

            // A peer that cannot acknowledge the snapshot leaves only the 0.2.x option: clear once
            // the frames are on the data lane. The window that opens (enqueued, not yet delivered —
            // a crash there loses the snapshot with the debt already discharged) is why 0.3.0 peers
            // wait for the ack instead. The reconnect handshake's gap check re-detects it whenever
            // the oplog is non-empty; an EMPTY oplog remains the undetectable corner.
            if (!_peerSupportsPartialSnapshot)
                await ClearSentSnapshotDebtAsync(ct);
        }

        await PumpLoopAsync(dataLane, ct);
    }

    /// <summary>
    /// Discharges the snapshot debt this session's outbound snapshot covered, exactly once.
    /// </summary>
    /// <remarks>
    /// Taking the scope out of the field before clearing makes this idempotent, which matters
    /// because two paths reach it — the pre-0.3.0 clear-on-enqueue above and the peer's SnapshotAck
    /// — and a peer that acknowledges anyway (a newer build than its handshake claimed) must not
    /// clear a second time on top of flags a later baseline may already have re-raised.
    /// </remarks>
    private async Task ClearSentSnapshotDebtAsync(CancellationToken ct)
    {
        var scope = Interlocked.Exchange(ref _sentSnapshotScope, null);
        if (scope is null)
            return;
        await _store.ClearSnapshotDebtAsync(scope.Tables, ct);
    }

    // The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's
    // inbound processing is never starved of its acks behind a backlog of DeltaBatches.
    private static async Task WriterLoopAsync(
        SyncDuplex duplex, ChannelReader<SyncMessage> control, ChannelReader<SyncMessage> data, CancellationToken ct)
    {
        Task<bool>? controlWait = null;
        Task<bool>? dataWait = null;
        var controlDone = false;
        var dataDone = false;
        while (true)
        {
            ct.ThrowIfCancellationRequested();
            if (control.TryRead(out var message) || data.TryRead(out message))
            {
                await duplex.Send(message, ct);
                continue;
            }
            // WaitToReadAsync => false means the lane was COMPLETED and is drained: stop waiting on
            // it (re-creating its waiter would instantly re-complete false — a 100% CPU spin) and
            // exit once both lanes are terminal. Nothing completes the lanes today, but Task 12 or
            // a future refactor must not silently detonate this loop.
            if (controlDone && dataDone)
                return;
            // At most one outstanding waiter per lane (SingleReader); a stale already-completed
            // waiter only costs one extra TryRead pass.
            if (!controlDone) controlWait ??= control.WaitToReadAsync(ct).AsTask();
            if (!dataDone) dataWait ??= data.WaitToReadAsync(ct).AsTask();
            await Task.WhenAny(controlWait ?? dataWait!, dataWait ?? controlWait!);
            if (controlWait is { IsCompleted: true })
            {
                controlDone = !await controlWait;
                controlWait = null;
            }
            if (dataWait is { IsCompleted: true })
            {
                dataDone = !await dataWait;
                dataWait = null;
            }
        }
    }

    /// <summary>
    /// Reads oplog deltas above the sent watermark and ships them, splitting each read into as many
    /// messages as <see cref="ReplicationOptions.MaxBatchBytes"/> requires.
    /// </summary>
    /// <remarks>
    /// Row count alone cannot bound a message: rows are arbitrarily wide, so any row-count cap is
    /// really a guess about the widest row, and guessing wrong produces a message the transport
    /// rejects — replication then stops converging with a size error, not a data error. Budgeting by
    /// summed serialized bytes bounds the thing that is actually limited, and lets the row cap go
    /// back to meaning what it says.
    /// </remarks>
    private async Task PumpLoopAsync(ChannelWriter<SyncMessage> dataLane, CancellationToken ct)
    {
        while (true)
        {
            ct.ThrowIfCancellationRequested();
            var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct);
            if (batch.Count > 0)
            {
                var index = 0;
                while (index < batch.Count)
                {
                    var delta = new DeltaBatch();
                    var bytes = 0L;
                    while (index < batch.Count)
                    {
                        var proto = ToProto(batch[index]);
                        var size = EntrySize(proto);
                        // The first entry of a message is taken however large it is: refusing an
                        // oversized row would stall the stream on it permanently, whereas sending it
                        // alone either succeeds or fails loudly at the transport with its real size.
                        if (delta.Entries.Count > 0 && bytes + size > _options.MaxBatchBytes)
                            break;
                        delta.Entries.Add(proto);
                        bytes += size;
                        index++;
                    }
                    // Blocking here on a full data lane IS the pump's backpressure.
                    await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct);
                    // Advanced per MESSAGE, not per read: a fault mid-split must not re-send what
                    // already went out, and must not skip what did not.
                    _sentThruSeq = batch[index - 1].Seq;
                }
                continue;
            }

            await Task.Delay(_options.FlushInterval, ct);
        }
    }

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

    private async Task ReceiveLoopAsync(
        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, CancellationToken ct)
    {
        // One-message lookahead. Coalescing means reading ahead until a message turns out NOT to be
        // of the kind being coalesced, and a Channel cannot un-read — so that message is parked here
        // and processed first on the next pass. Its presence also keeps the outer loop from waiting
        // on the inbox when there is already work in hand.
        SyncMessage? lookahead = null;

        while (lookahead is not null || await duplex.Inbox.WaitToReadAsync(ct))
        {
            while (true)
            {
                var msg = lookahead;
                lookahead = null;
                if (msg is null)
                {
                    if (!duplex.Inbox.TryRead(out var read))
                        break;
                    msg = read;
                }

                switch (msg.MsgCase)
                {
                    case SyncMessage.MsgOneofCase.DeltaBatch:
                        // Deltas apply normally even mid-snapshot: LWW makes the interleave safe.
                        lookahead = await HandleCoalescedDeltasAsync(duplex, controlLane, msg.DeltaBatch, ct);
                        break;
                    case SyncMessage.MsgOneofCase.DeltaAck:
                        lookahead = await HandleCoalescedAcksAsync(duplex, msg.DeltaAck, ct);
                        break;
                    case SyncMessage.MsgOneofCase.SnapshotBegin:
                        if (SnapshotApplier is null)
                            throw new NotSupportedException("snapshot received but no snapshot applier configured");
                        _snapshotReceiving = true;
                        await SnapshotApplier.OnBeginAsync(msg.SnapshotBegin, ct);
                        break;
                    case SyncMessage.MsgOneofCase.SnapshotBatch:
                        if (!_snapshotReceiving)
                            throw new InvalidOperationException(
                                "Replication protocol error: SnapshotBatch with no active snapshot (missing SnapshotBegin).");
                        await SnapshotApplier!.OnBatchAsync(msg.SnapshotBatch, ct);
                        break;
                    case SyncMessage.MsgOneofCase.SnapshotComplete:
                        if (!_snapshotReceiving)
                            throw new InvalidOperationException(
                                "Replication protocol error: SnapshotComplete with no active snapshot (missing SnapshotBegin).");
                        await SnapshotApplier!.OnCompleteAsync(msg.SnapshotComplete, ct);
                        _snapshotReceiving = false;
                        _metrics?.RecordSnapshotReceived();
                        _status?.SetLastSyncUtc(_utcNow());
                        // Acknowledge only to a peer that asked for it. Sent AFTER the applier's
                        // durable watermark write, so the ack means "I hold these rows", which is
                        // what lets the sender drop its per-table flags. On the unbounded control
                        // lane, like DeltaAck: an ack must never queue behind a data backlog.
                        if (_peerSupportsPartialSnapshot)
                            await controlLane.WriteAsync(
                                new SyncMessage { SnapshotAck = new SnapshotAck { AsOfSeq = msg.SnapshotComplete.AsOfSeq } }, ct);
                        break;
                    case SyncMessage.MsgOneofCase.SnapshotAck:
                        // The peer has durably applied the snapshot THIS node streamed, so the
                        // tables it covered no longer owe one. Nothing to reconcile against the
                        // ack's as-of seq: only one snapshot is ever sent per session, so the scope
                        // held in _sentSnapshotScope is unambiguously the one being acknowledged.
                        await ClearSentSnapshotDebtAsync(ct);
                        break;
                    default:
                        throw new InvalidOperationException(
                            $"Replication protocol error: unexpected {msg.MsgCase} after the handshake.");
                }
            }
        }
    }

    /// <summary>
    /// Applies the given delta batch together with any delta batches ALREADY waiting in the inbox,
    /// as one transaction answered by one ack; returns the non-delta message that ended the run (to
    /// be processed next), or null if the inbox merely ran dry.
    /// </summary>
    /// <remarks>
    /// <para>
    /// Byte-budgeting the sender produces more, smaller messages, and one apply transaction plus one
    /// ack round-trip per message is the cost that would otherwise scale with that. Coalescing pays
    /// it once per burst instead — and a burst is exactly what a reconnect after an outage delivers.
    /// </para>
    /// <para>
    /// Only messages already queued are absorbed (<c>TryRead</c>, never a wait), so this adds no
    /// latency: a trickle of one batch at a time coalesces nothing and behaves exactly as before.
    /// The run stops at the sender's own caps, so the extra memory held is bounded by roughly one
    /// sender batch beyond the budget — the overshoot of finishing whichever batch crosses it.
    /// </para>
    /// </remarks>
    private async Task<SyncMessage?> HandleCoalescedDeltasAsync(
        SyncDuplex duplex, ChannelWriter<SyncMessage> controlLane, DeltaBatch first, CancellationToken ct)
    {
        var entries = new List<OplogEntry>(first.Entries);
        var bytes = 0L;
        foreach (var entry in first.Entries)
            bytes += EntrySize(entry);

        SyncMessage? lookahead = null;
        while (entries.Count < _options.MaxBatchSize && bytes < _options.MaxBatchBytes
               && duplex.Inbox.TryRead(out var next))
        {
            if (next.MsgCase != SyncMessage.MsgOneofCase.DeltaBatch)
            {
                lookahead = next;
                break;
            }
            foreach (var entry in next.DeltaBatch.Entries)
            {
                entries.Add(entry);
                bytes += EntrySize(entry);
            }
        }

        await ApplyDeltaEntriesAsync(controlLane, entries, ct);
        return lookahead;
    }

    /// <summary>
    /// Records the given ack together with any acks ALREADY waiting in the inbox, as a single
    /// watermark write; returns the non-ack message that ended the run, or null if the inbox ran dry.
    /// </summary>
    /// <remarks>
    /// Exactly equivalent to recording each in turn, and strictly cheaper: <c>applied_thru_seq</c> is
    /// monotonic and <c>RecordPeerAckAsync</c> is MAX-guarded, so every superseded ack would have
    /// written a watermark that the final one overwrites and run a DELETE below its own horizon.
    /// Unbounded coalescing is safe here — acks collapse into one number, accumulating nothing.
    /// </remarks>
    private async Task<SyncMessage?> HandleCoalescedAcksAsync(
        SyncDuplex duplex, DeltaAck first, CancellationToken ct)
    {
        var acked = first.AppliedThruSeq;

        SyncMessage? lookahead = null;
        while (duplex.Inbox.TryRead(out var next))
        {
            if (next.MsgCase != SyncMessage.MsgOneofCase.DeltaAck)
            {
                lookahead = next;
                break;
            }
            if (next.DeltaAck.AppliedThruSeq > acked)
                acked = next.DeltaAck.AppliedThruSeq;
        }

        InterlockedMax(ref _peerAckedSeq, acked);
        await _store.RecordPeerAckAsync(acked, ct);
        return lookahead;
    }

    private async Task ApplyDeltaEntriesAsync(
        ChannelWriter<SyncMessage> controlLane, IReadOnlyList<OplogEntry> batchEntries, CancellationToken ct)
    {
        if (batchEntries.Count == 0)
            return;

        var now = _utcNow();
        var nowMs = now.ToUnixTimeMilliseconds();
        var driftLimitMs = nowMs + (long)_options.MaxHlcDriftAhead.TotalMilliseconds;
        var toApply = new List<OplogEntryRecord>(batchEntries.Count);
        var maxSeq = 0L;
        var driftDeadLettered = 0;

        foreach (var proto in batchEntries)
        {
            if (proto.Seq > maxSeq) maxSeq = proto.Seq;
            var entry = FromProto(proto);
            var physicalMs = HybridLogicalClock.PhysicalMs(entry.Hlc);
            if (physicalMs > driftLimitMs)
            {
                if (!_driftWarned)
                {
                    // Rate-limited to one warning per session: drift is usually a sustained clock
                    // condition and would otherwise flood the log at the delta rate.
                    _driftWarned = true;
                    _logger.LogWarning(
                        "Inbound oplog entry HLC physical time {EntryMs} ms is ahead of local UTC {NowMs} ms by more than MaxHlcDriftAhead ({Drift}). Further drift this session is not re-logged.",
                        physicalMs, nowMs, _options.MaxHlcDriftAhead);
                }
                // Dead-letter is the fail-closed disposition: faulting the session instead would
                // livelock on reconnect (the peer resends the same drifted entries forever).
                if (_options.FailClosedOnDrift)
                {
                    await DeadLetterDriftedAsync(entry, physicalMs, nowMs, now, ct);
                    driftDeadLettered++;
                    continue;
                }
            }
            toApply.Add(entry);
        }

        var result = await _applier.ApplyBatchAsync(toApply, ct);
        _metrics?.RecordApplied(result.Applied);
        _metrics?.RecordConflictsDiscarded(result.Discarded);
        _metrics?.RecordDeadLettered(result.DeadLettered + driftDeadLettered);
        _status?.SetLastSyncUtc(now);
        var ackThru = Math.Max(result.AppliedThruSeq, maxSeq);
        if (ackThru > result.AppliedThruSeq)
            // Dead-lettered entries carried the batch's top seqs; advance the durable remote
            // watermark over them (MAX-guarded like the applier's) so a reconnect handshake does
            // not make the peer resend what this node has already discarded.
            await _db.ExecuteAsync(
                "UPDATE __localdb_peer_state SET last_applied_remote_seq = MAX(last_applied_remote_seq, @seq) WHERE id = 1",
                new { seq = ackThru }, ct);
        await controlLane.WriteAsync(new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = ackThru } }, ct);
    }

    // Not transactional with the watermark update: a crash between them redelivers the batch and
    // dead-letters the same entry again (at-least-once — a benign duplicate dead-letter row).
    private Task DeadLetterDriftedAsync(
        OplogEntryRecord entry, long physicalMs, long nowMs, DateTimeOffset now, CancellationToken ct) =>
        _db.ExecuteAsync(
            "INSERT INTO __localdb_dead_letter (received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
            "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
            new
            {
                utc = now.UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture),
                t = entry.TableName,
                pk = entry.PkJson,
                row = entry.RowJson,
                hlc = entry.Hlc,
                node = entry.NodeId,
                err = $"HLC drift ahead: entry physical time {physicalMs} ms (Unix UTC) exceeds local now {nowMs} ms + MaxHlcDriftAhead {_options.MaxHlcDriftAhead} (fail-closed).",
            }, ct);

    private async Task DrainSecondaryAsync(Task task)
    {
        try
        {
            await task;
        }
        catch (Exception ex)
        {
            _logger.LogDebug(ex, "Secondary sync-session task ended during teardown.");
        }
    }

    private Handshake BuildHandshake(PeerState peerState)
    {
        var handshake = new Handshake
        {
            NodeId = _db.NodeId,
            LibSchemaVersion = LibSchemaVersion,
            LastAppliedRemoteSeq = peerState.LastAppliedRemoteSeq,
            // Unconditional: this build always understands a scoped snapshot and always answers a
            // completed one. It is a statement about the code, not about configuration or state.
            SupportsPartialSnapshot = true,
        };
        foreach (var table in _db.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
            handshake.Tables.Add(new TableDigest { TableName = table.Name, Digest = table.Digest });
        return handshake;
    }

    private static void ValidateHandshake(Handshake local, Handshake peer)
    {
        var versionMismatch = local.LibSchemaVersion != peer.LibSchemaVersion;

        var localByName = local.Tables.ToDictionary(t => t.TableName, t => t.Digest, StringComparer.Ordinal);
        var peerByName = peer.Tables.ToDictionary(t => t.TableName, t => t.Digest, StringComparer.Ordinal);
        var mismatched = new SortedSet<string>(StringComparer.Ordinal);
        foreach (var name in localByName.Keys.Union(peerByName.Keys, StringComparer.Ordinal))
            if (!localByName.TryGetValue(name, out var ld) || !peerByName.TryGetValue(name, out var pd) || ld != pd)
                mismatched.Add(name);

        if (!versionMismatch && mismatched.Count == 0)
            return;

        var parts = new List<string>();
        if (versionMismatch)
            parts.Add($"lib_schema_version {local.LibSchemaVersion} != peer {peer.LibSchemaVersion}");
        if (mismatched.Count > 0)
            parts.Add($"table schema mismatch on: {string.Join(", ", mismatched)}");

        throw new LocalDbSchemaMismatchException(
            "Replication handshake rejected (fail-closed): " + string.Join("; ", parts) + ".");
    }

    /// <summary>
    /// Decides what snapshot this node owes the peer: null for none, <see cref="SnapshotScope.Full"/>
    /// for every registered table, or a scoped snapshot naming just the tables that need one.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The two sources of snapshot debt are genuinely different in reach, which is why they are
    /// stored separately. An <b>oplog-horizon</b> gap — the peer's watermark below the oldest delta
    /// still streamable, which is what a cap prune or an impossible peer ack leaves behind — means
    /// deltas have been lost for the oplog as a whole, so nothing short of every table is sound. A
    /// <b>baseline</b> means one table's pre-existing rows were never captured, and says nothing at
    /// all about the others.
    /// </para>
    /// <para>
    /// A scope that happens to name every registered table is returned as Full rather than as a
    /// list: it covers exactly what a full snapshot covers, so it should also get a full snapshot's
    /// watermark advance on both sides, and it needs no promotion for a pre-0.3.0 peer.
    /// </para>
    /// <para>
    /// Flags naming tables that are not registered are ignored (they can only stream nothing).
    /// <c>DeregisterReplicated</c> drops them with the table's ledger rows, so the surviving way to
    /// hold one is to drop the SQL table without deregistering — which the handshake's fail-closed
    /// digest comparison rejects for its own reasons.
    /// </para>
    /// </remarks>
    private async Task<SnapshotScope?> ComputeSnapshotScopeAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
    {
        if (peerState.NeedsSnapshot || await HasHorizonGapAsync(peerHandshake, peerState, ct))
            return SnapshotScope.Full;

        var flagged = await _store.GetTablesNeedingSnapshotAsync(ct);
        if (flagged.Count == 0)
            return null;

        var scoped = flagged.Where(_db.ReplicatedTables.ContainsKey).ToList();
        if (scoped.Count == 0)
            return null;
        if (scoped.Count >= _db.ReplicatedTables.Count)
            return SnapshotScope.Full;

        // A peer that would read SnapshotBegin.tables as "all tables" and then advance its watermark
        // past deltas this snapshot did not carry gets the full snapshot instead. Wider than asked
        // for, but it is the only shape the older protocol can express correctly — and it is what
        // 0.2.x did for every baseline.
        if (!_peerSupportsPartialSnapshot)
        {
            _logger.LogInformation(
                "Peer {PeerNodeId} does not support scoped snapshots; promoting the snapshot of {Tables} to a full one.",
                PeerNodeId, string.Join(",", scoped));
            return SnapshotScope.Full;
        }

        return new SnapshotScope(scoped);
    }

    private async Task<bool> HasHorizonGapAsync(Handshake peerHandshake, PeerState peerState, CancellationToken ct)
    {
        var minRows = await _db.QueryAsync(
            "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
        var minSeq = minRows[0];

        // The oldest seq we could still stream as a delta. With rows in the oplog that is simply the
        // oldest one; with an EMPTY oplog it is the next seq we will ever write — one past
        // last_acked_seq, because ack-pruning is the only thing that empties the oplog without
        // flagging needs_snapshot, and it deletes exactly seq <= last_acked_seq.
        //
        // Treating an empty oplog as "no gap possible" is what let a wiped peer starve: a converged
        // pair prunes to empty on ack, so the healthy node held no delta at all, and a peer that
        // came back with a zero watermark and an empty database was told it needed nothing. The
        // steady state of a healthy pair was precisely the state that could not heal one.
        var streamableFrom = minSeq > 0 ? minSeq : peerState.LastAckedSeq + 1;

        // A gap exists when the peer's applied watermark falls below what we can still stream.
        return peerHandshake.LastAppliedRemoteSeq < streamableFrom - 1;
    }

    private static async Task<SyncMessage> ReceiveOneAsync(SyncDuplex duplex, CancellationToken ct)
    {
        if (await duplex.Inbox.WaitToReadAsync(ct) && duplex.Inbox.TryRead(out var message))
            return message;
        throw new InvalidOperationException(
            "Replication protocol error: the peer closed the stream before completing the handshake.");
    }

    private static OplogEntry ToProto(OplogEntryRecord entry)
    {
        var proto = new OplogEntry
        {
            Seq = entry.Seq,
            TableName = entry.TableName,
            PkJson = entry.PkJson,
            Hlc = entry.Hlc,
            NodeId = entry.NodeId,
            IsTombstone = entry.IsTombstone,
        };
        if (entry.RowJson is not null)
            proto.RowJson = entry.RowJson;
        return proto;
    }

    private static OplogEntryRecord FromProto(OplogEntry entry) =>
        new(entry.Seq, entry.TableName, entry.PkJson, entry.HasRowJson ? entry.RowJson : null,
            entry.Hlc, entry.NodeId, entry.IsTombstone);

    private static void InterlockedMax(ref long target, long value)
    {
        long current;
        do
        {
            current = Interlocked.Read(ref target);
            if (value <= current)
                return;
        }
        while (Interlocked.CompareExchange(ref target, value, current) != current);
    }
}