using System.Globalization;
using Microsoft.Data.Sqlite;
using ZB.MOM.WW.LocalDb.Internal;
using ZB.MOM.WW.LocalDb.Registration;

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

/// <summary>Outcome of applying one inbound batch. <see cref="AppliedThruSeq"/> is the batch's highest seq regardless of per-entry disposition, so the watermark advances even over discarded/dead-lettered entries.</summary>
internal sealed record ApplyResult(long AppliedThruSeq, int Applied, int Discarded, int DeadLettered);

/// <summary>
/// Applies a peer's oplog batch under last-writer-wins in one transaction: the applying guard is
/// raised so capture triggers skip these writes, per-entry LWW resolves against the local
/// row-version (hlc, then node_id ordinal tie-break), poison rows are dead-lettered while the
/// batch continues — each entry's row write and ledger write sharing a savepoint so a dead-lettered
/// entry leaves neither — and the remote watermark advances in the SAME transaction as the data so a
/// crash-replay of the batch is a no-op (LWW's strict-greater compare loses on equal versions).
/// This never writes <c>__localdb_oplog</c>: the 2-node topology does not re-forward applied deltas.
/// </summary>
/// <remarks>
/// Every statement is issued through a <see cref="PreparedStatements"/> set held for the life of the
/// batch, so each distinct statement is prepared ONCE and then rebound per row. The previous shape
/// went through the generic <c>ILocalDbTransaction</c> façade, which builds a fresh
/// <see cref="SqliteCommand"/> per call — three SQL parses per applied row (version lookup, the
/// table's delete/upsert, the row-version write), on the path that drains a reconnect backlog.
/// </remarks>
internal sealed class LwwApplier(SqliteLocalDb db, Func<DateTimeOffset>? utcNow = null)
{
    private readonly Func<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);

    /// <summary>
    /// Distinct statements prepared by the most recent <see cref="ApplyBatchAsync"/>. Test-only
    /// observation: it is the direct evidence that statement count tracks the number of distinct
    /// STATEMENTS rather than the number of rows.
    /// </summary>
    internal int LastBatchStatementCount { get; private set; }

    public async Task<ApplyResult> ApplyBatchAsync(IReadOnlyList<OplogEntryRecord> entries, CancellationToken ct)
    {
        if (entries.Count == 0)
            return new ApplyResult(0, 0, 0, 0);

        // Validated before opening the transaction: an unknown table is a schema disagreement
        // between the peers (a protocol violation), not a per-row poison — reject the whole batch
        // so nothing is half-applied.
        foreach (var entry in entries)
            if (!db.ReplicatedTables.ContainsKey(entry.TableName))
                throw new LocalDbSchemaMismatchException(
                    $"Received an oplog entry for table '{entry.TableName}', which is not registered for replication on this node.");

        var utcNow = _utcNow().UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture);
        var maxSeq = 0L;
        var maxHlc = 0L;
        foreach (var entry in entries)
        {
            if (entry.Seq > maxSeq) maxSeq = entry.Seq;
            if (entry.Hlc > maxHlc) maxHlc = entry.Hlc;
        }

        var applied = 0;
        var discarded = 0;
        var deadLettered = 0;

        await using var tx = await db.BeginPreparedTransactionAsync(ct);
        using var sql = new PreparedStatements(tx.Connection, tx.Sqlite);

        await sql.SetApplyingAsync(1, ct);

        foreach (var entry in entries)
        {
            // Without row data the upsert would bind NULL to every column — on an INTEGER PRIMARY
            // KEY table that silently inserts a phantom auto-rowid row — so it is poison, not a win.
            if (!entry.IsTombstone && entry.RowJson is null)
            {
                await sql.DeadLetterAsync(entry, "non-tombstone entry with null row_json", utcNow, ct);
                deadLettered++;
                continue;
            }

            var local = await sql.ReadLocalVersionAsync(entry.TableName, entry.PkJson, ct);

            var remoteWins = local is null
                || entry.Hlc > local.Value.Hlc
                || (entry.Hlc == local.Value.Hlc && string.CompareOrdinal(entry.NodeId, local.Value.NodeId) > 0);

            if (!remoteWins)
            {
                discarded++;
                continue;
            }

            var table = db.ReplicatedTables[entry.TableName];

            // An entry is TWO statements — the row change and its ledger row — and SQLite aborts
            // only the failing statement, so a fault between them used to commit a half-applied
            // entry: a tombstone whose table row is gone while __localdb_row_version still calls it
            // live (which SnapshotStreamer can only warn about and skip, so the peer never learns
            // the row died), or an upsert whose new row carries the OLD version and therefore loses
            // the next LWW comparison it should win. A per-entry savepoint makes the pair
            // all-or-nothing without widening the poison catch to the whole batch.
            //
            // Opened OUTSIDE the try on purpose: a SAVEPOINT that failed opened nothing, so the
            // handler's ROLLBACK TO would name a savepoint that does not exist and fault in turn.
            await sql.EnterEntryAsync(ct);
            try
            {
                if (entry.IsTombstone)
                    await sql.DeleteAsync(table, entry.PkJson, ct);
                else
                    await sql.UpsertAsync(table, entry.RowJson!, ct);

                await sql.WriteRowVersionAsync(entry, utcNow, ct);
                await sql.ReleaseEntryAsync(ct);
                applied++;
            }
            catch (SqliteException ex)
            {
                // Deliberately narrow: only statement-level SQLite errors (e.g. json_extract on
                // malformed row_json) are treated as per-entry poison — SQLite aborts just the
                // statement, not the transaction, so the batch continues after dead-lettering.
                // Any other exception propagates with the savepoint still open, and disposing the
                // outer transaction unwinds the whole batch (fail-closed).
                try
                {
                    await sql.RollbackEntryAsync(ct);
                    await sql.DeadLetterAsync(entry, ex.Message, utcNow, ct);
                }
                catch (Exception recoveryFailure) when (recoveryFailure is not OperationCanceledException)
                {
                    // Recovering from the poison entry itself failed — a wedged connection, an I/O
                    // error, a dead-letter INSERT that faults. The outcome is unchanged and still
                    // fail-closed (nothing commits; disposing the batch transaction unwinds it),
                    // but reporting only the recovery failure would throw away the ORIGINAL cause,
                    // which is the one that says WHY this entry is poison. Both are carried out:
                    // the original as InnerException (AggregateException surfaces the first inner
                    // there), the recovery failure alongside it with its own stack intact.
                    throw new AggregateException(
                        $"Failed to dead-letter a poison oplog entry for table '{entry.TableName}' " +
                        $"(seq {entry.Seq}): {recoveryFailure.Message}. The entry's own fault is the " +
                        "inner exception. The batch is not committed.",
                        ex, recoveryFailure);
                }

                deadLettered++;
            }
        }

        await sql.AdvancePeerStateAsync(maxSeq, maxHlc, utcNow, ct);
        await sql.SetApplyingAsync(0, ct);

        LastBatchStatementCount = sql.StatementCount;

        await tx.CommitAsync(ct);

        db.Clock.Observe(maxHlc);
        return new ApplyResult(maxSeq, applied, discarded, deadLettered);
    }

    /// <summary>
    /// The statement set one apply-batch runs on: each distinct statement is built and prepared on
    /// first use, then rebound and re-executed per row. Statements whose SQL depends on the table
    /// (delete / upsert) are cached per table; the rest are shared across all of them.
    /// </summary>
    /// <remarks>
    /// Reusing the same <see cref="SqliteCommand"/> with unchanged <c>CommandText</c> is what makes
    /// this prepare-once: Microsoft.Data.Sqlite holds the compiled <c>sqlite3_stmt</c> on the command
    /// and only recompiles when the text changes. <see cref="SqliteCommand.Prepare"/> is called
    /// explicitly anyway, so the intent is stated rather than inferred from an implementation detail.
    /// </remarks>
    private sealed class PreparedStatements(SqliteConnection connection, SqliteTransaction transaction) : IDisposable
    {
        private readonly List<SqliteCommand> _owned = [];
        private readonly Dictionary<string, SqliteCommand> _deletes = new(StringComparer.Ordinal);
        private readonly Dictionary<string, SqliteCommand> _upserts = new(StringComparer.Ordinal);

        /// <summary>Name of the per-entry savepoint. Fixed: only one is ever open at a time.</summary>
        private const string EntrySavepoint = "zb_lww_entry";

        private SqliteCommand? _applying;
        private SqliteCommand? _enterEntry;
        private SqliteCommand? _releaseEntry;
        private SqliteCommand? _rollbackEntry;
        private SqliteCommand? _lookup;
        private SqliteCommand? _rowVersion;
        private SqliteCommand? _deadLetter;
        private SqliteCommand? _peerState;

        /// <summary>How many distinct statements this set has prepared.</summary>
        public int StatementCount => _owned.Count;

        public Task SetApplyingAsync(int value, CancellationToken ct)
        {
            var cmd = _applying ??= Create(
                "UPDATE __localdb_applying SET applying = @applying WHERE id = 1", "@applying");
            cmd.Parameters["@applying"].Value = value;
            return cmd.ExecuteNonQueryAsync(ct);
        }

        /// <summary>Opens the per-entry savepoint inside the batch transaction.</summary>
        public Task EnterEntryAsync(CancellationToken ct)
        {
            var cmd = _enterEntry ??= Create($"SAVEPOINT {EntrySavepoint}");
            return cmd.ExecuteNonQueryAsync(ct);
        }

        /// <summary>Commits the entry's work into the batch transaction (pops the savepoint).</summary>
        public Task ReleaseEntryAsync(CancellationToken ct)
        {
            var cmd = _releaseEntry ??= Create($"RELEASE {EntrySavepoint}");
            return cmd.ExecuteNonQueryAsync(ct);
        }

        /// <summary>
        /// Undoes everything the entry did and pops the savepoint, leaving the batch transaction
        /// open and usable. Two statements, because ROLLBACK TO rewinds to the savepoint but leaves
        /// it ON the stack — without the RELEASE, every dead-lettered entry would nest another one.
        /// </summary>
        public async Task RollbackEntryAsync(CancellationToken ct)
        {
            var cmd = _rollbackEntry ??= Create($"ROLLBACK TO {EntrySavepoint}");
            await cmd.ExecuteNonQueryAsync(ct);
            await ReleaseEntryAsync(ct);
        }

        public async Task<(long Hlc, string NodeId)?> ReadLocalVersionAsync(
            string tableName, string pkJson, CancellationToken ct)
        {
            var cmd = _lookup ??= Create(
                "SELECT hlc, node_id FROM __localdb_row_version WHERE table_name = @t AND pk_json = @pk",
                "@t", "@pk");
            cmd.Parameters["@t"].Value = tableName;
            cmd.Parameters["@pk"].Value = pkJson;

            await using var reader = await cmd.ExecuteReaderAsync(ct);
            if (!await reader.ReadAsync(ct))
                return null;
            return (reader.GetInt64(0), reader.GetString(1));
        }

        public Task DeleteAsync(ReplicatedTable table, string pkJson, CancellationToken ct)
        {
            if (!_deletes.TryGetValue(table.Name, out var cmd))
                _deletes[table.Name] = cmd = Create(BuildDelete(table), "@pk");
            cmd.Parameters["@pk"].Value = pkJson;
            return cmd.ExecuteNonQueryAsync(ct);
        }

        public Task UpsertAsync(ReplicatedTable table, string rowJson, CancellationToken ct)
        {
            if (!_upserts.TryGetValue(table.Name, out var cmd))
                _upserts[table.Name] = cmd = Create(BuildUpsert(table), "@row");
            cmd.Parameters["@row"].Value = rowJson;
            return cmd.ExecuteNonQueryAsync(ct);
        }

        public Task WriteRowVersionAsync(OplogEntryRecord entry, string utcNow, CancellationToken ct)
        {
            var cmd = _rowVersion ??= Create(
                "INSERT OR REPLACE INTO __localdb_row_version " +
                "(table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
                "VALUES (@t, @pk, @hlc, @node, @tomb, @tut)",
                "@t", "@pk", "@hlc", "@node", "@tomb", "@tut");
            cmd.Parameters["@t"].Value = entry.TableName;
            cmd.Parameters["@pk"].Value = entry.PkJson;
            cmd.Parameters["@hlc"].Value = entry.Hlc;
            cmd.Parameters["@node"].Value = entry.NodeId;
            cmd.Parameters["@tomb"].Value = entry.IsTombstone ? 1 : 0;
            cmd.Parameters["@tut"].Value = entry.IsTombstone ? utcNow : DBNull.Value;
            return cmd.ExecuteNonQueryAsync(ct);
        }

        public Task DeadLetterAsync(OplogEntryRecord entry, string error, string utcNow, CancellationToken ct)
        {
            var cmd = _deadLetter ??= Create(
                "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)",
                "@utc", "@t", "@pk", "@row", "@hlc", "@node", "@err");
            cmd.Parameters["@utc"].Value = utcNow;
            cmd.Parameters["@t"].Value = entry.TableName;
            cmd.Parameters["@pk"].Value = entry.PkJson;
            cmd.Parameters["@row"].Value = (object?)entry.RowJson ?? DBNull.Value;
            cmd.Parameters["@hlc"].Value = entry.Hlc;
            cmd.Parameters["@node"].Value = entry.NodeId;
            cmd.Parameters["@err"].Value = error;
            return cmd.ExecuteNonQueryAsync(ct);
        }

        public Task AdvancePeerStateAsync(long maxSeq, long maxHlc, string utcNow, CancellationToken ct)
        {
            var cmd = _peerState ??= Create(
                "UPDATE __localdb_peer_state SET " +
                "last_applied_remote_seq = MAX(last_applied_remote_seq, @maxSeq), " +
                "last_seen_hlc = MAX(last_seen_hlc, @maxHlc), last_sync_utc = @utc WHERE id = 1",
                "@maxSeq", "@maxHlc", "@utc");
            cmd.Parameters["@maxSeq"].Value = maxSeq;
            cmd.Parameters["@maxHlc"].Value = maxHlc;
            cmd.Parameters["@utc"].Value = utcNow;
            return cmd.ExecuteNonQueryAsync(ct);
        }

        private SqliteCommand Create(string sql, params string[] parameterNames)
        {
            var cmd = connection.CreateCommand();
            try
            {
                cmd.Transaction = transaction;
                cmd.CommandText = sql;
                foreach (var name in parameterNames)
                    cmd.Parameters.Add(new SqliteParameter(name, DBNull.Value));
                cmd.Prepare();
                _owned.Add(cmd);
                return cmd;
            }
            catch
            {
                cmd.Dispose();
                throw;
            }
        }

        public void Dispose()
        {
            foreach (var cmd in _owned)
                cmd.Dispose();
            _owned.Clear();
        }

        private static string BuildDelete(ReplicatedTable table)
        {
            var pkMatch = string.Join(
                " AND ", table.PkColumns.Select(p => $"{Ident(p)} = json_extract(@pk, {JsonPath(p)})"));
            return $"DELETE FROM {Ident(table.Name)} WHERE {pkMatch}";
        }

        private static string BuildUpsert(ReplicatedTable table)
        {
            var columns = string.Join(", ", table.Columns.Select(c => Ident(c.Name)));
            var values = string.Join(", ", table.Columns.Select(c => $"json_extract(@row, {JsonPath(c.Name)})"));
            return $"INSERT OR REPLACE INTO {Ident(table.Name)} ({columns}) VALUES ({values})";
        }

        // Local copy of TriggerSqlGenerator's convention (that helper lives in the core assembly).
        private static string Ident(string name) => "\"" + name.Replace("\"", "\"\"") + "\"";

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