scadaproj/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs

using System.Globalization;
using ZB.MOM.WW.LocalDb.Hlc;

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

/// <summary>A single oplog delta as read for outbound replication. <see cref="RowJson"/> is null for tombstones.</summary>
internal sealed record OplogEntryRecord(
    long Seq, string TableName, string PkJson, string? RowJson, long Hlc, string NodeId, bool IsTombstone);

/// <summary>The persisted peer-sync watermarks and flags from <c>__localdb_peer_state</c>.</summary>
internal sealed record PeerState(
    string? PeerNodeId, long LastAckedSeq, long LastAppliedRemoteSeq, long LastSeenHlc, bool NeedsSnapshot, string? LastSyncUtc);

/// <summary>
/// DB-side operations over the replication bookkeeping tables: batched oplog reads above a
/// watermark, peer ack + prune, backlog/age cap enforcement, tombstone retention, and peer state.
/// </summary>
internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<DateTimeOffset>? utcNow = null)
{
    // Must match the delete trigger's strftime('%Y-%m-%dT%H:%M:%fZ') output so string comparison is chronological.
    internal const string TombstoneUtcFormat = "yyyy-MM-ddTHH:mm:ss.fffZ";

    private readonly Func<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);

    /// <summary>
    /// Reads up to <paramref name="maxBatch"/> oplog entries with <c>seq &gt; afterSeq</c> in ascending
    /// seq order. Caller contract: this read resumes SILENTLY past a cap-pruned gap, so a caller
    /// streaming incrementally must first confirm its position is still below the horizon — see
    /// <see cref="GetMinSeqAsync"/>, which <c>SyncSession</c>'s pump checks on every iteration.
    /// </summary>
    public Task<IReadOnlyList<OplogEntryRecord>> ReadBatchAboveAsync(long afterSeq, int maxBatch, CancellationToken ct = default) =>
        db.QueryAsync(
            "SELECT seq, table_name, pk_json, row_json, hlc, node_id, is_tombstone FROM __localdb_oplog " +
            "WHERE seq > @afterSeq ORDER BY seq LIMIT @maxBatch",
            static r => new OplogEntryRecord(
                r.GetInt64(0), r.GetString(1), r.GetString(2), r.IsDBNull(3) ? null : r.GetString(3),
                r.GetInt64(4), r.GetString(5), r.GetInt64(6) != 0),
            new { afterSeq, maxBatch }, ct);

    /// <summary>
    /// Records the peer's delivery ack and prunes oplog rows at or below the watermark, in one
    /// transaction. The watermark is monotonic (a late out-of-order ack never regresses it) and
    /// clamped to the highest seq this node has ever produced: an ack beyond that is a protocol
    /// violation, so <c>needs_snapshot</c> is set to force a resync rather than trust the peer.
    /// </summary>
    public async Task RecordPeerAckAsync(long ackedSeq, CancellationToken ct = default)
    {
        await using var tx = await db.BeginTransactionAsync(ct);
        // Known-max includes last_acked_seq: acked rows are already pruned, so a duplicate ack of
        // a pruned seq is benign — only an ack beyond anything ever produced is a violation.
        var knownMax = (await tx.QueryAsync(
            "SELECT MAX((SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog), " +
            "(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1))",
            static r => r.GetInt64(0), null, ct))[0];
        // Pruning up to the clamped ack may delete rows the peer never actually received; that is
        // safe only because needs_snapshot = 1 commits in the SAME transaction, forcing a resync.
        if (ackedSeq > knownMax)
            await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, ct);
        await tx.ExecuteAsync(
            "UPDATE __localdb_peer_state SET last_acked_seq = MAX(last_acked_seq, @acked), last_sync_utc = @now WHERE id = 1",
            new { acked = Math.Min(ackedSeq, knownMax), now = FormatUtc(_utcNow()) }, ct);
        await tx.ExecuteAsync(
            "DELETE FROM __localdb_oplog WHERE seq <= (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
            null, ct);
        await tx.CommitAsync(ct);
    }

    /// <summary>Deletes row-version tombstones older than <see cref="ReplicationOptions.TombstoneRetention"/>; live rows and fresh tombstones are kept.</summary>
    public Task PruneTombstonesAsync(CancellationToken ct = default) =>
        db.ExecuteAsync(
            "DELETE FROM __localdb_row_version WHERE is_tombstone = 1 AND tombstone_utc < @cutoff",
            new { cutoff = FormatUtc(_utcNow() - options.TombstoneRetention) }, ct);

    /// <summary>
    /// The unacked backlog — oplog entries above the peer's last-acked watermark — in O(1).
    /// </summary>
    /// <remarks>
    /// <para>
    /// <b>Why not <c>COUNT(*)</c>.</b> The obvious query walks every unacked row, and this value is
    /// read once a minute by maintenance AND on every metrics scrape. A node whose peer has been
    /// down for a day therefore pays a full scan of a million-row backlog, repeatedly, precisely
    /// when it is already under stress.
    /// </para>
    /// <para>
    /// <b>The formula.</b> <c>MAX(seq) - MAX(last_acked_seq, MIN(seq) - 1)</c>, floored at 0. Both
    /// MIN and MAX over <c>seq</c> are single b-tree seeks (it is the rowid, and SQLite's min/max
    /// optimization turns each into one seek), and the peer-state read is a one-row lookup — so the
    /// whole thing is three seeks regardless of backlog size.
    /// </para>
    /// <para>
    /// <b>Why it is exact.</b> <c>seq</c> is AUTOINCREMENT, so retained rows are a contiguous run
    /// wherever nothing has punched a hole in the middle — and no ordinary path punches holes: ack
    /// pruning deletes <c>seq &lt;= last_acked_seq</c> (a prefix), the row cap deletes below
    /// <c>MAX(seq) - keep + 1</c> (a prefix), and the age cap deletes below an HLC cutoff, which is
    /// a prefix because HLCs are stamped in seq order. Clamping the floor to <c>MIN(seq) - 1</c> is
    /// what keeps the answer exact after a prune has moved the horizon above the ack watermark;
    /// without it a cap-pruned oplog would report its whole seq range forever and re-flag the cap on
    /// every tick.
    /// </para>
    /// <para>
    /// <b>The one inexact case</b> is <c>ILocalDb.DeregisterReplicated</c>, which deletes one
    /// table's rows from wherever they sit and so can leave holes. The estimate then over-reports by
    /// at most the number of holes, until the watermark advances past them. Over-reporting is the
    /// safe direction: it can only trigger a snapshot resync earlier, never suppress one.
    /// </para>
    /// </remarks>
    public async Task<long> GetOplogDepthAsync(CancellationToken ct = default)
    {
        var rows = await db.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct);
        return rows[0];
    }

    /// <summary>
    /// The unacked backlog read synchronously on a dedicated short-lived connection, for the
    /// metric-observation callbacks (gauge scrape / status poll); mirrors
    /// <see cref="GetOplogDepthAsync"/>'s query. Deliberate tradeoff: synchronous I/O on the
    /// metrics-collection thread is acceptable here because WAL-mode readers never block (or are
    /// blocked by) writers, the query is three b-tree seeks, and the worst case is bounded by
    /// <c>BusyTimeoutMs</c> — all at scrape frequency, not on any data path. May throw if the db
    /// is disposing — callers on the observation path swallow and drop the sample.
    /// </summary>
    public long GetOplogDepthSync()
    {
        using var conn = db.CreateConnection();
        using var cmd = conn.CreateCommand();
        cmd.CommandText = DepthSql;
        return (long)cmd.ExecuteScalar()!;
    }

    // Both depth accessors run exactly this; keeping it in one place is what makes "mirrors the
    // async query" a fact rather than a comment that can rot.
    internal const string DepthSql =
        "SELECT MAX(0, " +
        "COALESCE((SELECT MAX(seq) FROM __localdb_oplog), 0) - " +
        "MAX((SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1), " +
        "COALESCE((SELECT MIN(seq) FROM __localdb_oplog), 1) - 1))";

    /// <summary>
    /// The highest seq this node has ever produced: the oplog's max, or — once acked rows have been
    /// pruned away — the ack watermark that pruned them. Nothing we emit can exceed it, so a peer
    /// claiming to have applied more than this is not talking about the stream we are producing now.
    /// </summary>
    public async Task<long> GetKnownMaxSeqAsync(CancellationToken ct = default)
    {
        var rows = await db.QueryAsync(
            "SELECT MAX((SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog), " +
            "(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1))",
            static r => r.GetInt64(0), null, ct);
        return rows[0];
    }

    /// <summary>
    /// The oldest oplog seq still present — the delta horizon — or 0 when the oplog is empty.
    /// </summary>
    /// <remarks>
    /// One b-tree seek (<c>seq</c> is the rowid and SQLite optimizes MIN into a seek), which is what
    /// makes it affordable for the pump to check on every flush interval. Compared against a
    /// session's sent watermark it answers the only question the pump has: is everything I have not
    /// yet sent still here? Ack-pruning deletes <c>seq &lt;= last_acked_seq</c> and the peer can only
    /// ack what it received, so in a healthy session the horizon never climbs past
    /// <c>sentThruSeq + 1</c>; a higher one means a cap prune ate unsent deltas.
    /// </remarks>
    public async Task<long> GetMinSeqAsync(CancellationToken ct = default)
    {
        var rows = await db.QueryAsync(
            "SELECT COALESCE(MIN(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
        return rows[0];
    }

    /// <summary>The highest oplog seq present, or 0 when the oplog is empty.</summary>
    public async Task<long> GetMaxSeqAsync(CancellationToken ct = default)
    {
        var rows = await db.QueryAsync(
            "SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog", static r => r.GetInt64(0), null, ct);
        return rows[0];
    }

    /// <summary>
    /// Enforces the backlog caps in one transaction; returns true if a snapshot resync was flagged.
    /// Rows cap: backlog above <see cref="ReplicationOptions.MaxOplogRows"/> prunes to the newest
    /// MaxOplogRows entries. Age cap: an oldest entry whose HLC physical time exceeds
    /// <see cref="ReplicationOptions.MaxOplogAge"/> prunes every entry past the age cutoff — disk
    /// safety over delta continuity. Either trigger sets <c>needs_snapshot</c>.
    /// </summary>
    public async Task<bool> EnforceCapsAsync(CancellationToken ct = default)
    {
        var ageCutoffMs = (_utcNow() - options.MaxOplogAge).ToUnixTimeMilliseconds();

        // Condition reads and prunes share the transaction so the decision cannot go stale (TOCTOU).
        await using var tx = await db.BeginTransactionAsync(ct);

        // Same O(1) expression as GetOplogDepthAsync — the caps check is the per-minute reader that
        // made the old COUNT(*) expensive in the first place.
        var depth = (await tx.QueryAsync(DepthSql, static r => r.GetInt64(0), null, ct))[0];
        var overRows = depth > options.MaxOplogRows;

        var oldest = await tx.QueryAsync(
            "SELECT hlc FROM __localdb_oplog ORDER BY seq LIMIT 1", static r => r.GetInt64(0), null, ct);
        var overAge = oldest.Count > 0 && HybridLogicalClock.PhysicalMs(oldest[0]) < ageCutoffMs;

        if (!overRows && !overAge)
            return false;

        await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, ct);
        if (overRows)
            await tx.ExecuteAsync(
                "DELETE FROM __localdb_oplog WHERE seq < ((SELECT MAX(seq) FROM __localdb_oplog) - @keep + 1)",
                new { keep = options.MaxOplogRows }, ct);
        if (overAge)
            await tx.ExecuteAsync(
                "DELETE FROM __localdb_oplog WHERE (hlc >> 16) < @ageCutoffMs", new { ageCutoffMs }, ct);
        await tx.CommitAsync(ct);
        return true;
    }

    /// <summary>Reads the singleton peer-state row.</summary>
    public async Task<PeerState> GetPeerStateAsync(CancellationToken ct = default)
    {
        var rows = await db.QueryAsync(
            "SELECT peer_node_id, last_acked_seq, last_applied_remote_seq, last_seen_hlc, needs_snapshot, last_sync_utc " +
            "FROM __localdb_peer_state WHERE id = 1",
            static r => new PeerState(
                r.IsDBNull(0) ? null : r.GetString(0), r.GetInt64(1), r.GetInt64(2), r.GetInt64(3),
                r.GetInt64(4) != 0, r.IsDBNull(5) ? null : r.GetString(5)),
            null, ct);
        return rows[0];
    }

    /// <summary>Records the identity of the peer node this database syncs with.</summary>
    public Task SetPeerNodeIdAsync(string peerNodeId, CancellationToken ct = default) =>
        db.ExecuteAsync("UPDATE __localdb_peer_state SET peer_node_id = @v WHERE id = 1", new { v = peerNodeId }, ct);

    /// <summary>
    /// Forgets everything we knew about the previous peer's stream: the inbound seq watermark and
    /// its observed clock. Called when the peer's node id changes.
    /// </summary>
    /// <remarks>
    /// <para>
    /// <c>last_applied_remote_seq</c> is a watermark in the PEER's seq space, and a rebuilt peer is
    /// a new database whose oplog numbers from 1. Carrying the old peer's watermark across the
    /// swap makes the new peer's early writes look already-applied, so they are dropped — silently,
    /// and for as many writes as the old watermark was high.
    /// </para>
    /// <para>
    /// <c>last_acked_seq</c> is deliberately NOT reset: it describes OUR oplog (what has been
    /// pruned), not the peer's, and it is what tells a rebuilt peer's handshake that it needs a
    /// snapshot. Clearing it would make an already-pruned node look like it had never written
    /// anything and turn the back-fill back off.
    /// </para>
    /// </remarks>
    public Task ResetInboundWatermarkAsync(CancellationToken ct = default) =>
        db.ExecuteAsync(
            "UPDATE __localdb_peer_state SET last_applied_remote_seq = 0, last_seen_hlc = 0 WHERE id = 1", null, ct);

    /// <summary>
    /// Sets or clears the DATABASE-WIDE snapshot-resync flag — the one that means "every registered
    /// table". Set by the oplog-horizon paths (cap prune, impossible peer ack), which invalidate
    /// delta continuity for all tables at once. Per-table flags live in
    /// <c>__localdb_snapshot_state</c>; see <see cref="GetTablesNeedingSnapshotAsync"/>.
    /// </summary>
    public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) =>
        db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct);

    /// <summary>
    /// The tables individually flagged as owing the peer a snapshot (schema v3), in ordinal name
    /// order. Independent of the database-wide flag: baselining one table flags only that table.
    /// </summary>
    public Task<IReadOnlyList<string>> GetTablesNeedingSnapshotAsync(CancellationToken ct = default) =>
        db.QueryAsync(
            "SELECT table_name FROM __localdb_snapshot_state WHERE needs_snapshot <> 0 ORDER BY table_name",
            static r => r.GetString(0), null, ct);

    /// <summary>
    /// Clears the snapshot debt a completed snapshot discharged: the named tables when
    /// <paramref name="tables"/> is non-empty, otherwise <b>everything</b> — the database-wide flag
    /// and every per-table row, because a full snapshot subsumes each of them.
    /// </summary>
    /// <remarks>
    /// <para>
    /// Rows are deleted rather than zeroed: absence is the resting state, and keeping a row per
    /// table that has ever been baselined would accumulate bookkeeping that means nothing. Runs in
    /// one transaction so a full snapshot's two clears cannot half-apply and leave the pair
    /// re-snapshotting on every reconnect.
    /// </para>
    /// <para>
    /// A scoped clear deliberately does NOT touch the database-wide flag. If the cap enforcer
    /// flagged a whole-database resync while this snapshot was in flight, that debt is real and
    /// unrelated — the snapshot only carried some of the tables — so it must survive. The full
    /// clear may take it down because a full snapshot carries every row state as of its as-of seq,
    /// which is exactly what the pruned deltas would have delivered.
    /// </para>
    /// </remarks>
    public async Task ClearSnapshotDebtAsync(IReadOnlyList<string> tables, CancellationToken ct = default)
    {
        await using var tx = await db.BeginTransactionAsync(ct);
        if (tables.Count == 0)
        {
            await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 0 WHERE id = 1", null, ct);
            await tx.ExecuteAsync("DELETE FROM __localdb_snapshot_state", null, ct);
        }
        else
        {
            foreach (var table in tables)
                await tx.ExecuteAsync(
                    "DELETE FROM __localdb_snapshot_state WHERE table_name = @t", new { t = table }, ct);
        }
        await tx.CommitAsync(ct);
    }

    /// <summary>
    /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps
    /// <c>last_sync_utc</c>. Both watermarks are MAX-guarded (monotonic): a snapshot completing with a
    /// lower as-of seq than a concurrently-applied delta must never regress the durable watermark.
    /// </summary>
    public Task SetLastAppliedRemoteSeqAsync(long seq, long lastSeenHlc, CancellationToken ct = default) =>
        db.ExecuteAsync(
            "UPDATE __localdb_peer_state SET last_applied_remote_seq = MAX(last_applied_remote_seq, @seq), " +
            "last_seen_hlc = MAX(last_seen_hlc, @hlc), last_sync_utc = @now WHERE id = 1",
            new { seq, hlc = lastSeenHlc, now = FormatUtc(_utcNow()) }, ct);

    private static string FormatUtc(DateTimeOffset value) =>
        value.UtcDateTime.ToString(TombstoneUtcFormat, CultureInfo.InvariantCulture);
}