lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmSfSchema.cs

using System.Security.Cryptography;
using System.Text;
using Microsoft.Data.Sqlite;

namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;

/// <summary>
///     DDL for the alarm store-and-forward buffer: one row per alarm event awaiting delivery to
///     the historian gateway.
/// </summary>
/// <remarks>
///     <para>
///         Replaces the standalone <c>alarm-historian.db</c> the sink used to own outright. Living
///         in the consolidated LocalDb file is what lets the buffer replicate to the redundant pair
///         peer, so a node that dies with undelivered alarm history no longer takes it to the grave.
///     </para>
///     <para>
///         Deliberately depends on nothing but <see cref="SqliteConnection"/> so it can be applied
///         to any connection — the host's <c>LocalDbSetup.OnReady</c> in production, and a bare
///         connection in a test — without dragging the DI graph along. Mirrors
///         <c>DeploymentCacheSchema</c>.
///     </para>
///     <para>
///         <b>Why <see cref="IdColumn"/> is TEXT and app-minted.</b> The legacy queue keyed on
///         <c>RowId INTEGER PRIMARY KEY AUTOINCREMENT</c>. Convergence is last-writer-wins over the
///         primary key, so two nodes independently allocating rowid 7 for different alarms would
///         silently overwrite one another. The sink mints the id from a hash of the payload, which
///         additionally makes the same event converge to one row when both nodes of a pair enqueue
///         it — as they legitimately do in the window before the first redundancy snapshot arrives.
///     </para>
/// </remarks>
public static class AlarmSfSchema
{
    /// <summary>Table holding queued alarm events awaiting delivery.</summary>
    public const string EventsTable = "alarm_sf_events";

    /// <summary>The single primary-key column.</summary>
    public const string IdColumn = "id";

    /// <summary>
    ///     Derives a row's primary key from its serialized payload.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         Deterministic rather than a fresh GUID, so that the same event arriving twice
    ///         converges to one row under last-writer-wins instead of duplicating. That happens for
    ///         real in two places: <c>HistorianAdapterActor</c> default-writes while its redundancy
    ///         role is unknown, so both nodes of a pair accept the same fanned transition in the
    ///         window before the first snapshot; and both nodes independently migrate their own
    ///         legacy queue file, which for a warm pair holds overlapping history.
    ///     </para>
    ///     <para>
    ///         Two genuinely distinct events cannot collide. <c>AlarmHistorianEvent</c> carries a
    ///         full-precision timestamp alongside the alarm id, transition kind, message and user,
    ///         so an equal hash means an equal event.
    ///     </para>
    /// </remarks>
    /// <param name="payloadJson">The serialized <c>AlarmHistorianEvent</c>.</param>
    /// <returns>The row's primary key.</returns>
    public static string DeriveId(string payloadJson)
    {
        ArgumentNullException.ThrowIfNull(payloadJson);
        return Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payloadJson)));
    }

    /// <summary>
    ///     Creates the buffer table if it does not already exist. Idempotent.
    /// </summary>
    /// <param name="connection">
    ///     An already-open connection. <c>ILocalDb.CreateConnection()</c> hands out open,
    ///     pragma-configured connections — do not call <c>Open()</c> on one.
    /// </param>
    public static void Apply(SqliteConnection connection)
    {
        ArgumentNullException.ThrowIfNull(connection);

        using var cmd = connection.CreateCommand();

        // Columns map 1:1 onto the legacy Queue table so the one-time migrator is a straight copy
        // and the drain keeps its existing semantics: dead_lettered is the same 0/1 flag, and an
        // acknowledged row is DELETEd rather than marked. Deleting keeps the table bounded without
        // a second sweeper, and the replication engine carries the delete as a tombstone, so the
        // peer drops its copy too.
        //
        // The drain index orders by enqueued_at_utc because a hashed TEXT id carries no insertion
        // order the way the legacy AUTOINCREMENT RowId did. Timestamps are round-trip ("O") format,
        // so lexicographic ordering is chronological; id breaks ties so the order is total and the
        // index covers the drain's read.
        cmd.CommandText = """
            CREATE TABLE IF NOT EXISTS alarm_sf_events (
                id               TEXT    NOT NULL PRIMARY KEY,
                alarm_id         TEXT    NOT NULL,
                enqueued_at_utc  TEXT    NOT NULL,
                payload_json     TEXT    NOT NULL,
                attempt_count    INTEGER NOT NULL DEFAULT 0,
                last_attempt_utc TEXT    NULL,
                last_error       TEXT    NULL,
                dead_lettered    INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS ix_alarm_sf_events_drain
                ON alarm_sf_events (dead_lettered, enqueued_at_utc, id);
            """;
        cmd.ExecuteNonQuery();
    }
}