using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;

/// <summary>
/// SQLite persistence layer for store-and-forward messages.
/// Uses direct Microsoft.Data.Sqlite (not EF Core) for lightweight site-side storage.
/// No max buffer size per design decision.
/// </summary>
public class StoreAndForwardStorage
{
    private readonly string _connectionString;
    private readonly ILogger<StoreAndForwardStorage> _logger;

    /// <summary>
    /// Initializes a new instance of <see cref="StoreAndForwardStorage"/> with the given SQLite connection string.
    /// </summary>
    /// <param name="connectionString">SQLite connection string for the store-and-forward database.</param>
    /// <param name="logger">Logger for diagnostics.</param>
    public StoreAndForwardStorage(string connectionString, ILogger<StoreAndForwardStorage> logger)
    {
        _connectionString = connectionString;
        _logger = logger;
    }

    /// <summary>
    /// Creates the sf_messages table if it does not exist.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task InitializeAsync()
    {
        EnsureDatabaseDirectoryExists();

        await using var connection = await OpenConnectionAsync();

        // Enable WAL so the concurrent writers this store has by design (script
        // enqueues, the retry sweep's per-target lanes, standby replication applies,
        // central pull queries) read/write without "database is locked". WAL is
        // persistent + file-scoped; in-memory DBs report "memory" instead — harmless,
        // so it is not asserted here.
        await using (var walCmd = connection.CreateCommand())
        {
            walCmd.CommandText = "PRAGMA journal_mode=WAL";
            await walCmd.ExecuteNonQueryAsync();
        }

        await using var command = connection.CreateCommand();
        command.CommandText = @"
            CREATE TABLE IF NOT EXISTS sf_messages (
                id                TEXT PRIMARY KEY,
                category          INTEGER NOT NULL,
                target            TEXT    NOT NULL,
                payload_json      TEXT    NOT NULL,
                retry_count       INTEGER NOT NULL DEFAULT 0,
                max_retries       INTEGER NOT NULL DEFAULT 50,
                retry_interval_ms INTEGER NOT NULL DEFAULT 30000,
                created_at        TEXT    NOT NULL,
                last_attempt_at   TEXT,
                status            INTEGER NOT NULL DEFAULT 0,
                last_error        TEXT,
                origin_instance   TEXT
            );

            CREATE INDEX IF NOT EXISTS idx_sf_messages_status ON sf_messages(status);
            CREATE INDEX IF NOT EXISTS idx_sf_messages_category ON sf_messages(category);
        ";
        await command.ExecuteNonQueryAsync();

        // Additively add the execution_id /
        // source_script columns. CREATE TABLE IF NOT EXISTS above does NOT add
        // columns to a table that already exists from before these fields, so a
        // databases created by an older build needs the columns ALTER-ed in.
        // SQLite has no "ADD COLUMN IF NOT EXISTS"; the column presence is
        // probed first and the ALTER skipped when already there. Both columns
        // are nullable with no default, so any row buffered before this
        // migration reads back ExecutionId/SourceScript = null (back-compat).
        await AddColumnIfMissingAsync(connection, "execution_id", "TEXT");
        await AddColumnIfMissingAsync(connection, "source_script", "TEXT");

        // Additively add the
        // parent_execution_id column the same way — a sibling to execution_id.
        // Nullable with no default, so any row buffered before this migration
        // reads back ParentExecutionId = null (back-compat).
        await AddColumnIfMissingAsync(connection, "parent_execution_id", "TEXT");

        // Additively add the epoch-ms sibling of last_attempt_at (Task 18). The
        // ISO-8601 text column stays authoritative for reads / back-compat; this
        // INTEGER column drives the due predicate so the sweep no longer parses
        // julianday() per row.
        await AddColumnIfMissingAsync(connection, "last_attempt_at_ms", "INTEGER");

        // One-time backfill for rows persisted before the ms column existed: derive
        // epoch-ms from the text timestamp. The "... IS NULL" guard makes this run
        // once per legacy DB and never match again — this is the only remaining
        // julianday() use in the store.
        await using (var backfill = connection.CreateCommand())
        {
            backfill.CommandText = @"
                UPDATE sf_messages
                SET last_attempt_at_ms = CAST((julianday(last_attempt_at) - 2440587.5) * 86400000 AS INTEGER)
                WHERE last_attempt_at IS NOT NULL AND last_attempt_at_ms IS NULL";
            await backfill.ExecuteNonQueryAsync();
        }

        // Covering index for the due query (status filter + ms ordering column).
        // Created after the ALTER above so the column exists.
        await using (var dueIndex = connection.CreateCommand())
        {
            dueIndex.CommandText =
                "CREATE INDEX IF NOT EXISTS idx_sf_messages_status_due ON sf_messages(status, last_attempt_at_ms)";
            await dueIndex.ExecuteNonQueryAsync();
        }

        _logger.LogInformation("Store-and-forward SQLite storage initialized");
    }

    /// <summary>
    /// Adds a column to <c>sf_messages</c>
    /// only when it is not already present. SQLite lacks <c>ADD COLUMN IF NOT
    /// EXISTS</c>, so the schema is probed via <c>PRAGMA table_info</c> first.
    /// Idempotent — safe to run on every <see cref="InitializeAsync"/>.
    /// </summary>
    private static async Task AddColumnIfMissingAsync(
        SqliteConnection connection, string columnName, string columnType)
    {
        await using var probe = connection.CreateCommand();
        probe.CommandText = "SELECT COUNT(*) FROM pragma_table_info('sf_messages') WHERE name = @name";
        probe.Parameters.AddWithValue("@name", columnName);
        var exists = Convert.ToInt32(await probe.ExecuteScalarAsync()) > 0;
        if (exists)
        {
            return;
        }

        await using var alter = connection.CreateCommand();
        // Column name + type are caller-controlled constants, never user input —
        // safe to interpolate (parameters are not permitted in DDL).
        alter.CommandText = $"ALTER TABLE sf_messages ADD COLUMN {columnName} {columnType}";
        await alter.ExecuteNonQueryAsync();
    }

    /// <summary>
    /// Ensures the directory for a file-backed SQLite database exists. SQLite creates
    /// the database file on demand but not its parent directory, so a configured path
    /// such as "./data/store-and-forward.db" fails to open ("unable to open database
    /// file") when the "data" directory does not yet exist. In-memory databases and
    /// bare filenames in the working directory have no directory to create and are
    /// skipped.
    /// </summary>
    private void EnsureDatabaseDirectoryExists()
    {
        var builder = new SqliteConnectionStringBuilder(_connectionString);
        if (builder.Mode == SqliteOpenMode.Memory)
            return;

        var dataSource = builder.DataSource;
        if (string.IsNullOrEmpty(dataSource) || dataSource == ":memory:")
            return;

        var directory = System.IO.Path.GetDirectoryName(System.IO.Path.GetFullPath(dataSource));
        if (!string.IsNullOrEmpty(directory) && !System.IO.Directory.Exists(directory))
        {
            System.IO.Directory.CreateDirectory(directory);
            _logger.LogInformation("Created store-and-forward database directory: {Directory}", directory);
        }
    }

    /// <summary>
    /// Opens a connection with a 5s busy_timeout. Concurrent writers exist by
    /// design (script enqueues, the sweep's lanes, standby replication applies,
    /// central pull queries); with connection-per-operation the pragma must be
    /// set per open (it is per-connection, and Microsoft.Data.Sqlite pooling
    /// makes the extra statement cheap on a pooled physical connection).
    /// </summary>
    private async Task<SqliteConnection> OpenConnectionAsync()
    {
        var connection = new SqliteConnection(_connectionString);
        await connection.OpenAsync();
        await using var pragma = connection.CreateCommand();
        pragma.CommandText = "PRAGMA busy_timeout = 5000";
        await pragma.ExecuteNonQueryAsync();
        return connection;
    }

    /// <summary>
    /// INSERT statement for a full message row. Shared by <see cref="EnqueueAsync"/>
    /// and <see cref="ReplaceAllAsync"/>; bind with <see cref="BindMessageParameters"/>
    /// so the column list and the parameters never drift apart.
    /// </summary>
    private const string InsertMessageSql = @"
            INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries,
                                     retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error,
                                     origin_instance, execution_id, source_script, parent_execution_id)
            VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries,
                    @retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError,
                    @origin, @executionId, @sourceScript, @parentExecutionId)";

    /// <summary>
    /// Binds every column parameter for a full message row (used by
    /// <see cref="InsertMessageSql"/>). GUID ids are stored in canonical "D" string
    /// form and the epoch-ms sibling of last_attempt_at is derived here so a single
    /// place owns the mapping.
    /// </summary>
    private static void BindMessageParameters(SqliteCommand cmd, StoreAndForwardMessage message)
    {
        cmd.Parameters.AddWithValue("@id", message.Id);
        cmd.Parameters.AddWithValue("@category", (int)message.Category);
        cmd.Parameters.AddWithValue("@target", message.Target);
        cmd.Parameters.AddWithValue("@payload", message.PayloadJson);
        cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
        cmd.Parameters.AddWithValue("@maxRetries", message.MaxRetries);
        cmd.Parameters.AddWithValue("@retryIntervalMs", message.RetryIntervalMs);
        cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.ToString("O"));
        cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
            ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
        cmd.Parameters.AddWithValue("@lastAttemptMs",
            (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@status", (int)message.Status);
        cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@origin", (object?)message.OriginInstanceName ?? DBNull.Value);
        // GUID ids stored as canonical "D" strings; null when not threaded.
        cmd.Parameters.AddWithValue("@executionId",
            message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value);
        cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@parentExecutionId",
            message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value);
    }

    /// <summary>
    /// Enqueues a new message with Pending status.
    /// </summary>
    /// <param name="message">The message to enqueue.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task EnqueueAsync(StoreAndForwardMessage message)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = InsertMessageSql;
        BindMessageParameters(cmd, message);

        await cmd.ExecuteNonQueryAsync();
    }

    /// <summary>
    /// Returns every buffered message regardless of status, oldest-first, up to
    /// <paramref name="limit"/> rows, plus a flag indicating the buffer holds more
    /// than that. Anti-entropy groundwork: the active node snapshots its buffer for
    /// a standby resync. One extra row is fetched (<c>limit + 1</c>) purely to
    /// compute the truncation flag without a second COUNT round-trip.
    /// </summary>
    /// <param name="limit">Maximum rows to return; must be positive.</param>
    /// <returns>The oldest-first page (at most <paramref name="limit"/> rows) and whether more rows exist beyond it.</returns>
    public async Task<(List<StoreAndForwardMessage> Messages, bool Truncated)> GetAllMessagesAsync(int limit)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            SELECT id, category, target, payload_json, retry_count, max_retries,
                   retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
                   execution_id, source_script, parent_execution_id
            FROM sf_messages
            ORDER BY created_at ASC
            LIMIT @limitPlusOne";
        cmd.Parameters.AddWithValue("@limitPlusOne", limit + 1);

        var rows = await ReadMessagesAsync(cmd);
        var truncated = rows.Count > limit;
        return (rows.Take(limit).ToList(), truncated);
    }

    /// <summary>
    /// Atomically replaces the entire buffer with <paramref name="messages"/> in a
    /// single transaction (delete-all then insert-all). Standby-side anti-entropy
    /// apply: a peer-join resync overwrites the standby's divergent copy with the
    /// active node's authoritative snapshot. <b>Never call on an active node</b> —
    /// it discards every in-flight row.
    /// </summary>
    /// <param name="messages">The full buffer snapshot to install.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task ReplaceAllAsync(IReadOnlyList<StoreAndForwardMessage> messages)
    {
        await using var connection = await OpenConnectionAsync();
        await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();

        await using (var deleteCmd = connection.CreateCommand())
        {
            deleteCmd.Transaction = transaction;
            deleteCmd.CommandText = "DELETE FROM sf_messages";
            await deleteCmd.ExecuteNonQueryAsync();
        }

        foreach (var message in messages)
        {
            await using var insertCmd = connection.CreateCommand();
            insertCmd.Transaction = transaction;
            insertCmd.CommandText = InsertMessageSql;
            BindMessageParameters(insertCmd, message);
            await insertCmd.ExecuteNonQueryAsync();
        }

        await transaction.CommitAsync();
    }

    /// <summary>
    /// Inserts a message, or updates every mutable column in place if a row with the
    /// same id already exists (<c>ON CONFLICT(id) DO UPDATE</c>). Used by the standby
    /// node when applying a replicated Add/Park/Requeue: a Park/Requeue whose original
    /// Add was lost self-heals (the full message rides in the operation), and a
    /// duplicate Add (e.g. after an anti-entropy resync) applies newest-wins instead of
    /// violating the primary key. The original <c>created_at</c> is preserved — it
    /// orders the retry sweep and must never be overwritten on update.
    /// </summary>
    /// <param name="message">The message to insert or update.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task UpsertMessageAsync(StoreAndForwardMessage message)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries,
                                     retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error,
                                     origin_instance, execution_id, source_script, parent_execution_id)
            VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries,
                    @retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError,
                    @origin, @executionId, @sourceScript, @parentExecutionId)
            ON CONFLICT(id) DO UPDATE SET
                category = excluded.category,
                target = excluded.target,
                payload_json = excluded.payload_json,
                retry_count = excluded.retry_count,
                max_retries = excluded.max_retries,
                retry_interval_ms = excluded.retry_interval_ms,
                last_attempt_at = excluded.last_attempt_at,
                last_attempt_at_ms = excluded.last_attempt_at_ms,
                status = excluded.status,
                last_error = excluded.last_error,
                origin_instance = excluded.origin_instance,
                execution_id = excluded.execution_id,
                source_script = excluded.source_script,
                parent_execution_id = excluded.parent_execution_id";

        cmd.Parameters.AddWithValue("@id", message.Id);
        cmd.Parameters.AddWithValue("@category", (int)message.Category);
        cmd.Parameters.AddWithValue("@target", message.Target);
        cmd.Parameters.AddWithValue("@payload", message.PayloadJson);
        cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
        cmd.Parameters.AddWithValue("@maxRetries", message.MaxRetries);
        cmd.Parameters.AddWithValue("@retryIntervalMs", message.RetryIntervalMs);
        cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.ToString("O"));
        cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
            ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
        cmd.Parameters.AddWithValue("@lastAttemptMs",
            (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@status", (int)message.Status);
        cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@origin", (object?)message.OriginInstanceName ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@executionId",
            message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value);
        cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@parentExecutionId",
            message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value);

        await cmd.ExecuteNonQueryAsync();
    }

    /// <summary>
    /// Gets all messages that are due for retry (Pending status, last attempt older than retry interval).
    /// </summary>
    /// <returns>A task that resolves to the list of messages due for retry, ordered by creation time ascending.</returns>
    public async Task<List<StoreAndForwardMessage>> GetMessagesForRetryAsync(int limit = 0)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            SELECT id, category, target, payload_json, retry_count, max_retries,
                   retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
                   execution_id, source_script, parent_execution_id
            FROM sf_messages
            WHERE status = @pending
              AND (last_attempt_at_ms IS NULL
                   OR retry_interval_ms = 0
                   OR (@nowMs - last_attempt_at_ms) >= retry_interval_ms)
            ORDER BY created_at ASC";

        // Bound the sweep batch (oldest-first). 0 = unlimited (legacy).
        if (limit > 0)
        {
            cmd.CommandText += "\n            LIMIT @limit";
            cmd.Parameters.AddWithValue("@limit", limit);
        }

        cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);
        cmd.Parameters.AddWithValue("@nowMs", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());

        return await ReadMessagesAsync(cmd);
    }

    /// <summary>
    /// Updates a message after a delivery attempt.
    /// </summary>
    /// <param name="message">The message with updated retry count, status, and last error.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task UpdateMessageAsync(StoreAndForwardMessage message)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            UPDATE sf_messages
            SET retry_count = @retryCount,
                last_attempt_at = @lastAttempt,
                last_attempt_at_ms = @lastAttemptMs,
                status = @status,
                last_error = @lastError
            WHERE id = @id";

        cmd.Parameters.AddWithValue("@id", message.Id);
        cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
        cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
            ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
        cmd.Parameters.AddWithValue("@lastAttemptMs",
            (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@status", (int)message.Status);
        cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);

        await cmd.ExecuteNonQueryAsync();
    }

    /// <summary>
    /// Updates a message after a delivery attempt, but only if the row is still
    /// in the expected status. Returns true if the row was updated, false if it had
    /// already been changed (e.g. an operator retried or discarded the message) and so
    /// was skipped.
    /// </summary>
    /// <param name="message">The message with the updated values to persist.</param>
    /// <param name="expectedStatus">The status the row must currently have for the update to proceed.</param>
    /// <returns>A task that resolves to <c>true</c> if the row was updated; <c>false</c> if its status had already changed.</returns>
    public async Task<bool> UpdateMessageIfStatusAsync(
        StoreAndForwardMessage message,
        StoreAndForwardMessageStatus expectedStatus)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            UPDATE sf_messages
            SET retry_count = @retryCount,
                last_attempt_at = @lastAttempt,
                last_attempt_at_ms = @lastAttemptMs,
                status = @status,
                last_error = @lastError
            WHERE id = @id AND status = @expectedStatus";

        cmd.Parameters.AddWithValue("@id", message.Id);
        cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
        cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
            ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
        cmd.Parameters.AddWithValue("@lastAttemptMs",
            (object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@status", (int)message.Status);
        cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@expectedStatus", (int)expectedStatus);

        var rows = await cmd.ExecuteNonQueryAsync();
        return rows > 0;
    }

    /// <summary>
    /// Removes a successfully delivered message.
    /// </summary>
    /// <param name="messageId">The id of the message to remove.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task RemoveMessageAsync(string messageId)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id";
        cmd.Parameters.AddWithValue("@id", messageId);

        await cmd.ExecuteNonQueryAsync();
    }

    /// <summary>
    /// Gets all parked messages, optionally filtered by category, with pagination.
    /// </summary>
    /// <param name="category">Optional category filter; null returns parked messages from all categories.</param>
    /// <param name="pageNumber">1-based page number.</param>
    /// <param name="pageSize">Maximum number of messages to return per page.</param>
    /// <returns>A task that resolves to the page of parked messages and the total count of matching rows.</returns>
    public async Task<(List<StoreAndForwardMessage> Messages, int TotalCount)> GetParkedMessagesAsync(
        StoreAndForwardCategory? category = null,
        int pageNumber = 1,
        int pageSize = 50)
    {
        await using var connection = await OpenConnectionAsync();

        await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();

        // Count
        await using var countCmd = connection.CreateCommand();
        countCmd.Transaction = transaction;
        countCmd.CommandText = category.HasValue
            ? "SELECT COUNT(*) FROM sf_messages WHERE status = @parked AND category = @category"
            : "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
        countCmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
        if (category.HasValue) countCmd.Parameters.AddWithValue("@category", (int)category.Value);
        var totalCount = Convert.ToInt32(await countCmd.ExecuteScalarAsync());

        // Page
        await using var pageCmd = connection.CreateCommand();
        pageCmd.Transaction = transaction;
        var categoryFilter = category.HasValue ? " AND category = @category" : "";
        pageCmd.CommandText = $@"
            SELECT id, category, target, payload_json, retry_count, max_retries,
                   retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
                   execution_id, source_script, parent_execution_id
            FROM sf_messages
            WHERE status = @parked{categoryFilter}
            ORDER BY created_at ASC
            LIMIT @limit OFFSET @offset";

        pageCmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
        if (category.HasValue) pageCmd.Parameters.AddWithValue("@category", (int)category.Value);
        pageCmd.Parameters.AddWithValue("@limit", pageSize);
        pageCmd.Parameters.AddWithValue("@offset", (pageNumber - 1) * pageSize);

        var messages = await ReadMessagesAsync(pageCmd);

        await transaction.CommitAsync();
        return (messages, totalCount);
    }

    /// <summary>
    /// Moves a parked message back to pending for retry.
    /// </summary>
    /// <param name="messageId">The id of the parked message to move back to Pending.</param>
    /// <returns>A task that resolves to <c>true</c> if the message was found and reset to Pending; <c>false</c> if not found or not in Parked status.</returns>
    public async Task<bool> RetryParkedMessageAsync(string messageId)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            UPDATE sf_messages
            SET status = @pending, retry_count = 0, last_error = NULL,
                last_attempt_at = NULL, last_attempt_at_ms = NULL
            WHERE id = @id AND status = @parked";

        cmd.Parameters.AddWithValue("@id", messageId);
        cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);
        cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);

        var rows = await cmd.ExecuteNonQueryAsync();
        return rows > 0;
    }

    /// <summary>
    /// Permanently discards a parked message.
    /// </summary>
    /// <param name="messageId">The id of the parked message to discard.</param>
    /// <returns>A task that resolves to <c>true</c> if the message was found and deleted; <c>false</c> if not found or not in Parked status.</returns>
    public async Task<bool> DiscardParkedMessageAsync(string messageId)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id AND status = @parked";
        cmd.Parameters.AddWithValue("@id", messageId);
        cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);

        var rows = await cmd.ExecuteNonQueryAsync();
        return rows > 0;
    }

    /// <summary>
    /// Gets buffer depth by category (count of pending messages per category).
    /// </summary>
    /// <returns>A task that resolves to a dictionary mapping each category to its pending message count.</returns>
    public async Task<Dictionary<StoreAndForwardCategory, int>> GetBufferDepthByCategoryAsync()
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            SELECT category, COUNT(*) as cnt
            FROM sf_messages
            WHERE status = @pending
            GROUP BY category";
        cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);

        var result = new Dictionary<StoreAndForwardCategory, int>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            var category = (StoreAndForwardCategory)reader.GetInt32(0);
            var count = reader.GetInt32(1);
            result[category] = count;
        }

        return result;
    }

    /// <summary>
    /// Verifies messages are NOT deleted when an instance is deleted.
    /// Returns the count of messages for a given origin instance.
    /// </summary>
    /// <param name="instanceName">The origin instance name to count messages for.</param>
    /// <returns>A task that resolves to the number of messages whose origin instance matches <paramref name="instanceName"/>.</returns>
    public async Task<int> GetMessageCountByOriginInstanceAsync(string instanceName)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            SELECT COUNT(*)
            FROM sf_messages
            WHERE origin_instance = @origin";
        cmd.Parameters.AddWithValue("@origin", instanceName);

        return Convert.ToInt32(await cmd.ExecuteScalarAsync());
    }

    /// <summary>
    /// Gets a message by ID.
    /// </summary>
    /// <param name="messageId">The id of the message to retrieve.</param>
    /// <returns>A task that resolves to the matching message, or <c>null</c> if not found.</returns>
    public async Task<StoreAndForwardMessage?> GetMessageByIdAsync(string messageId)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            SELECT id, category, target, payload_json, retry_count, max_retries,
                   retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
                   execution_id, source_script, parent_execution_id
            FROM sf_messages
            WHERE id = @id";
        cmd.Parameters.AddWithValue("@id", messageId);

        var messages = await ReadMessagesAsync(cmd);
        return messages.FirstOrDefault();
    }

    /// <summary>
    /// Gets the count of parked messages (for health reporting).
    /// </summary>
    /// <returns>A task that resolves to the number of messages currently in Parked status.</returns>
    public async Task<int> GetParkedMessageCountAsync()
    {
        await using var conn = await OpenConnectionAsync();
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
        cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
        var result = await cmd.ExecuteScalarAsync();
        return Convert.ToInt32(result);
    }

    /// <summary>
    /// Gets the <c>created_at</c> of the oldest parked message, or <c>null</c> when
    /// no rows are parked. Backs the parked-retention aging signal on the site health
    /// report: parked rows persist until an operator acts, so their <i>age</i> — not
    /// just their count — is what tells an operator a forgotten site's backlog is
    /// growing stale.
    /// </summary>
    /// <returns>A task that resolves to the oldest parked row's creation time, or <c>null</c> if none are parked.</returns>
    public async Task<DateTimeOffset?> GetOldestParkedCreatedAtAsync()
    {
        await using var conn = await OpenConnectionAsync();
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = "SELECT MIN(created_at) FROM sf_messages WHERE status = @parked";
        cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
        var result = await cmd.ExecuteScalarAsync();
        return result is null or DBNull
            ? null
            : DateTimeOffset.Parse((string)result);
    }

    /// <summary>
    /// Gets total message count by status.
    /// </summary>
    /// <param name="status">The status to filter by.</param>
    /// <returns>A task that resolves to the count of messages with the specified status.</returns>
    public async Task<int> GetMessageCountByStatusAsync(StoreAndForwardMessageStatus status)
    {
        await using var connection = await OpenConnectionAsync();

        await using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @status";
        cmd.Parameters.AddWithValue("@status", (int)status);

        return Convert.ToInt32(await cmd.ExecuteScalarAsync());
    }

    private static async Task<List<StoreAndForwardMessage>> ReadMessagesAsync(SqliteCommand cmd)
    {
        var results = new List<StoreAndForwardMessage>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            results.Add(new StoreAndForwardMessage
            {
                Id = reader.GetString(0),
                Category = (StoreAndForwardCategory)reader.GetInt32(1),
                Target = reader.GetString(2),
                PayloadJson = reader.GetString(3),
                RetryCount = reader.GetInt32(4),
                MaxRetries = reader.GetInt32(5),
                RetryIntervalMs = reader.GetInt64(6),
                CreatedAt = DateTimeOffset.Parse(reader.GetString(7)),
                LastAttemptAt = reader.IsDBNull(8) ? null : DateTimeOffset.Parse(reader.GetString(8)),
                Status = (StoreAndForwardMessageStatus)reader.GetInt32(9),
                LastError = reader.IsDBNull(10) ? null : reader.GetString(10),
                OriginInstanceName = reader.IsDBNull(11) ? null : reader.GetString(11),
                // Rows persisted before the
                // additive migration have no execution_id / source_script value;
                // IsDBNull guards keep those reading back as null (back-compat).
                // Guid.TryParse (not Parse) guards the retry sweep: a corrupt
                // non-null execution_id is treated as "no execution id" rather
                // than throwing FormatException and aborting the whole sweep.
                ExecutionId = ParseGuidColumn(reader, 12),
                SourceScript = reader.IsDBNull(13) ? null : reader.GetString(13),
                // Rows persisted
                // before the additive migration have no parent_execution_id
                // value; the IsDBNull guard inside ParseGuidColumn keeps those
                // reading back as null (back-compat). Guid.TryParse (not Parse)
                // guards the retry sweep against a corrupt non-null value.
                ParentExecutionId = ParseGuidColumn(reader, 14)
            });
        }
        return results;
    }

    /// <summary>
    /// Defensively reads a nullable GUID column (<c>execution_id</c> or
    /// <c>parent_execution_id</c>). A <c>null</c> value (legacy pre-migration
    /// rows) and a malformed non-null value both yield <c>null</c> — a corrupt
    /// id must not throw and abort the retry sweep, which reads many rows.
    /// </summary>
    private static Guid? ParseGuidColumn(System.Data.Common.DbDataReader reader, int ordinal)
    {
        if (reader.IsDBNull(ordinal))
        {
            return null;
        }

        return Guid.TryParse(reader.GetString(ordinal), out var value)
            ? value
            : null;
    }
}