ScadaBridge/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardSchema.cs

using Microsoft.Data.Sqlite;

namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;

/// <summary>
/// DDL for the <c>sf_messages</c> table, extracted from
/// <see cref="StoreAndForwardStorage"/> so it can be applied by whoever owns the
/// database file.
/// <para>
/// Deliberately depends only on <c>Microsoft.Data.Sqlite</c>, not on the LocalDb library.
/// The Host applies this DDL to a LocalDb-managed connection before
/// <c>RegisterReplicated</c> installs the capture triggers; nothing about the schema
/// itself is LocalDb-specific, and the store still calls it so a directly-constructed
/// store (tests, tooling) remains self-sufficient.
/// </para>
/// </summary>
public static class StoreAndForwardSchema
{
    /// <summary>
    /// Creates the <c>sf_messages</c> table and its indexes when absent, and additively
    /// upgrades a table created by an older build. Idempotent — safe to run on every
    /// startup.
    /// </summary>
    /// <param name="connection">An open connection to the database that should hold the table.</param>
    public static void Apply(SqliteConnection connection)
    {
        ArgumentNullException.ThrowIfNull(connection);

        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);
        ";
            command.ExecuteNonQuery();
        }

        // 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).
        AddColumnIfMissing(connection, "execution_id", "TEXT");
        AddColumnIfMissing(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).
        AddColumnIfMissing(connection, "parent_execution_id", "TEXT");

        // Additively add the epoch-ms sibling of last_attempt_at. 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.
        AddColumnIfMissing(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.
        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";
            backfill.ExecuteNonQuery();
        }

        // Covering index for the due query (status filter + ms ordering column).
        // Created after the ALTER above so the column exists.
        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)";
            dueIndex.ExecuteNonQuery();
        }
    }

    /// <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="Apply"/>.
    /// </summary>
    private static void AddColumnIfMissing(
        SqliteConnection connection, string columnName, string columnType)
    {
        using (var probe = connection.CreateCommand())
        {
            probe.CommandText = "SELECT COUNT(*) FROM pragma_table_info('sf_messages') WHERE name = @name";
            probe.Parameters.AddWithValue("@name", columnName);
            if (Convert.ToInt32(probe.ExecuteScalar()) > 0)
            {
                return;
            }
        }

        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}";
        alter.ExecuteNonQuery();
    }
}