lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs

using Microsoft.Extensions.Configuration;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;

namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;

/// <summary>
///     The <c>onReady</c> callback handed to <c>AddZbLocalDb</c>: creates the deployment-cache,
///     alarm store-and-forward, and scripted-alarm condition-state tables and — when this node
///     actually replicates — opts them into change capture.
/// </summary>
/// <remarks>
///     <para>
///         Public rather than internal only so <c>LocalDbSetupTests</c> can drive the production
///         callback directly. Initialising a test database from a hand-written copy of this schema
///         would prove only that the test agrees with itself — the whole value of those tests is
///         that they exercise <b>this</b> method.
///     </para>
/// </remarks>
public static class LocalDbSetup
{
    /// <summary>
    ///     The four tables that participate in replication — registered when this node is configured
    ///     to replicate, and deregistered (stale capture triggers dropped) when it is not.
    /// </summary>
    /// <remarks>
    ///     One list shared by both branches on purpose. A table added to only the registration
    ///     branch would replicate on a configured node and then never be cleaned off an
    ///     unconfigured one; a table added to only the deregistration branch would be cleaned on a
    ///     node that never registered it.
    /// </remarks>
    private static readonly string[] ReplicatedTables =
    [
        DeploymentCacheSchema.ArtifactsTable,
        DeploymentCacheSchema.PointerTable,
        AlarmSfSchema.EventsTable,
        AlarmConditionStateSchema.StateTable,
    ];

    /// <summary>
    ///     Initialises the local database: DDL first, then replication registration (or its removal).
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>THE ORDER IS LOAD-BEARING: DDL → RegisterReplicated → writes.</b>
    ///         <c>RegisterReplicated</c> is what installs the three AFTER triggers that capture
    ///         changes into the oplog. Any row written before that call is never captured, so it
    ///         never reaches the peer — silently, and permanently, because nothing ever revisits
    ///         history. <see cref="AlarmSfLegacyMigrator"/> therefore runs <b>last</b>, after every
    ///         registration — it writes rows, and they must be captured like any other write.
    ///     </para>
    ///     <para>
    ///         <b>Registration is CONDITIONAL on <see cref="ReplicationIsConfigured"/>.</b> It used
    ///         to be unconditional, on every driver node, replicating or not. That was not merely
    ///         wasteful — an unreplicated node paid two INSERTs plus a <c>json_object</c> of the
    ///         whole row inside every write transaction, for an oplog nothing ever drained — it also
    ///         made <c>baselineExistingRows</c> unusable, because passing it on a node with no peer
    ///         would flag a snapshot resync for a peer that does not exist. The ordering rule above
    ///         is unaffected: when registration is skipped there is no capture to be too late for.
    ///     </para>
    ///     <para>
    ///         <b>The unconfigured branch actively cleans up</b> rather than merely abstaining.
    ///         <c>DeregisterReplicated</c> drops any capture triggers an earlier build left in the
    ///         file and prunes that table's oplog and row-version rows. Declining to register alone
    ///         would leave every file written by a pre-2026-08-18 build capturing forever — on
    ///         precisely the node that has no peer to capture for.
    ///     </para>
    ///     <para>
    ///         <b>A failure part-way through either loop fails host start, and that is deliberate.</b>
    ///         Neither branch catches, so a throw on the third of four tables leaves two flipped and
    ///         propagates out of the <c>AddZbLocalDb</c> singleton factory — a node that came up
    ///         anyway would be capturing (or not) on an arbitrary subset, which is silently partial
    ///         replication with nothing in the log to say so. Both library calls are idempotent, so
    ///         the retried boot simply completes the flip rather than compounding it.
    ///     </para>
    ///     <para>
    ///         The alarm buffer's tables are created unconditionally, regardless of whether this
    ///         node has <c>AlarmHistorian:Enabled</c> set, and regardless of replication. Creating
    ///         one lazily when the sink first appears would mean a node that enables the historian
    ///         later writes rows before its triggers exist — which is precisely the silent-loss
    ///         shape above.
    ///     </para>
    /// </remarks>
    /// <param name="db">The freshly constructed local database.</param>
    /// <param name="configuration">
    ///     Application configuration, read for the replication predicate and by the legacy migrator
    ///     for the pre-consolidation queue's path. Required rather than optional: an overload that
    ///     silently skipped the migration would be one wiring mistake away from discarding a node's
    ///     undelivered alarm history.
    /// </param>
    /// <param name="logger">
    ///     Where the deregistration self-heal reports what it cleaned. Defaults to Serilog's static
    ///     logger, which is what production uses: this runs inside the <c>AddZbLocalDb</c> singleton
    ///     factory, where resolving an <c>ILogger</c> would nest a service resolution inside a
    ///     singleton construction. <c>Program.cs</c> assigns <c>Serilog.Log.Logger</c> from the DI
    ///     root logger before anything resolves <c>ILocalDb</c>. The parameter exists so tests can
    ///     observe the line rather than infer it.
    /// </param>
    public static void OnReady(ILocalDb db, IConfiguration configuration, Serilog.ILogger? logger = null)
    {
        ArgumentNullException.ThrowIfNull(db);
        ArgumentNullException.ThrowIfNull(configuration);

        // CreateConnection() hands back an already-open, pragma-configured connection carrying the
        // zb_hlc_next() UDF the capture triggers need. Calling Open() on it would throw.
        using (var connection = db.CreateConnection())
        {
            DeploymentCacheSchema.Apply(connection);
            AlarmSfSchema.Apply(connection);
            AlarmConditionStateSchema.Apply(connection);
        }

        if (ReplicationIsConfigured(configuration))
            RegisterForCapture(db);
        else
            RemoveStaleCapture(db, logger ?? Serilog.Log.Logger);

        // LAST, and only here. This is the one call in OnReady that writes rows.
        AlarmSfLegacyMigrator.Migrate(db, configuration);
    }

    /// <summary>
    ///     Whether this node participates in LocalDb replication, and therefore needs the capture
    ///     triggers at all.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Either key counts, and the OR is not belt-and-braces — it is the only predicate
    ///         that is true on BOTH halves of a replicating pair.</b> Replication is one
    ///         bidirectional stream that exactly one side dials, so only the initiator sets
    ///         <c>LocalDb:Replication:PeerAddress</c>; the passive node carries the shared
    ///         <c>ApiKey</c> and a <c>LocalDb:SyncListenPort</c> and nothing else. (The docker-dev
    ///         rig is the reference: <c>site-a-1</c> has PeerAddress + ApiKey, <c>site-a-2</c> has
    ///         ApiKey + SyncListenPort, and central / site-b carry no <c>Replication</c> section at
    ///         all.) Keying on PeerAddress alone would strip capture from every passive node — its
    ///         local writes would stop reaching the initiator and the pair would converge in one
    ///         direction only, silently.
    ///     </para>
    ///     <para>
    ///         <b><c>LocalDb:SyncListenPort</c> deliberately does NOT join the OR.</b> The passive
    ///         endpoint is fail-closed on the key: with a port but no <c>ApiKey</c>,
    ///         <c>LocalDbSyncAuthInterceptor</c> refuses every inbound stream, so such a node
    ///         replicates nothing and has nothing to capture for. Adding the port to the predicate
    ///         would install capture triggers on a node whose sync endpoint rejects all callers.
    ///         <c>LocalDbReplicationHealthCheck</c> deliberately asks a <i>different</i> question —
    ///         "was this node MEANT to replicate?", which is <c>PeerAddress</c> OR a bound listen
    ///         port — so a port-without-key node correctly reports Degraded there while correctly
    ///         capturing nothing here. The two predicates diverge only on that misconfiguration,
    ///         and each is right about its own question.
    ///     </para>
    ///     <para>
    ///         The keys are read straight from configuration rather than through
    ///         <c>IOptions&lt;ReplicationOptions&gt;</c> because this runs inside the
    ///         <c>AddZbLocalDb</c> factory, where resolving another option would nest a service
    ///         resolution inside a singleton construction. The section path is
    ///         <see cref="LocalDbRegistration.ReplicationSectionPath"/> — the same constant
    ///         <c>AddZbLocalDbReplication</c> binds — so the predicate and the engine cannot
    ///         disagree about which node replicates.
    ///     </para>
    ///     <para>
    ///         <b>Both directions of a change to this predicate are handled at boot, and both
    ///         require the PAIR to be changed together.</b> The sync handshake compares
    ///         registered-table digests fail-closed, so a node that drops a table its peer still
    ///         replicates stops syncing with a schema mismatch. Enable (or disable) replication on
    ///         both nodes and restart both — never one.
    ///     </para>
    /// </remarks>
    /// <param name="configuration">The application configuration.</param>
    /// <returns><c>true</c> when this node is configured to replicate.</returns>
    public static bool ReplicationIsConfigured(IConfiguration configuration)
    {
        // Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture
        // changes and the nodes that stand down for a peer can never disagree.
        return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration);
    }

    /// <summary>Opts every replicated table into change capture, baselining what is already there.</summary>
    /// <remarks>
    ///     <para>
    ///         <c>baselineExistingRows: true</c> is what makes turning replication ON for a node that
    ///         has been running WITHOUT it actually converge. Capture is change-data-capture and the
    ///         snapshot streamer pages from <c>__localdb_row_version</c>, so rows written before the
    ///         first registration are in neither: a late opt-in used to replicate only writes made
    ///         after the restart, silently and forever. Baselining seeds the ledger at the LWW floor
    ///         (HLC 0, this node's id) and flags a snapshot resync, so a baselined row loses to any
    ///         genuine remote write of the same key and wins only where the peer has no version at
    ///         all. It is idempotent, and therefore free on every boot after the first — which is
    ///         why it is unconditional here rather than behind a one-shot flag.
    ///     </para>
    ///     <para>
    ///         ⚠️ <b>On the pinned LocalDb 0.2.1 the snapshot flag is per-DATABASE, not per-table.</b>
    ///         Baselining anything flags a resync that streams <b>every</b> registered table in full,
    ///         and on a pair enabling replication late both nodes do that to each other at once — the
    ///         largest transfer the library performs. Bounded, but expect one slow first sync, and
    ///         trim <c>alarm_sf_events</c> before enabling replication on a node that has been
    ///         buffering alarm history for a long time.
    ///     </para>
    ///     <para>
    ///         The per-table loop is kept rather than collapsed into a hypothetical bulk call: the
    ///         library's contract is per-table (PK validation, digest, and — from LocalDb 0.3.0 — the
    ///         snapshot flag itself), so a bulk API would only be able to give back the coarse
    ///         behaviour.
    ///     </para>
    /// </remarks>
    private static void RegisterForCapture(ILocalDb db)
    {
        foreach (var table in ReplicatedTables)
            db.RegisterReplicated(table, baselineExistingRows: true);
    }

    /// <summary>Drops capture this node must not pay for, and says so once if it found any.</summary>
    /// <remarks>
    ///     <c>DeregisterReplicated</c> is idempotent and deliberately usable on a table this process
    ///     never registered, so a never-replicated file simply reports nothing to clean and stays
    ///     silent. Symmetry is not a concern on this branch: this node has no peer, so there is no
    ///     handshake digest for a one-sided deregistration to fail — and re-enabling later is safe
    ///     despite the ledger prune, because <see cref="RegisterForCapture"/> baselines.
    /// </remarks>
    private static void RemoveStaleCapture(ILocalDb db, Serilog.ILogger logger)
    {
        var cleaned = 0;

        foreach (var table in ReplicatedTables)
        {
            if (db.DeregisterReplicated(table))
                cleaned++;
        }

        // Only when something was actually removed. Every default-OFF node in the fleet takes this
        // branch on every boot; logging unconditionally would be a daily non-event, which is how the
        // one boot that DID silently stop capturing gets overlooked.
        if (cleaned > 0)
        {
            logger.Information(
                "LocalDb: removed stale change-capture from {TableCount} table(s) — this node has no "
                + "replication configured ({PeerAddressKey}/{ApiKeyKey} are both absent — unset or blank), "
                + "so the capture "
                + "triggers an earlier build installed have been dropped and their oplog/row-version "
                + "rows pruned.",
                cleaned,
                $"{LocalDbRegistration.ReplicationSectionPath}:PeerAddress",
                $"{LocalDbRegistration.ReplicationSectionPath}:ApiKey");
        }
    }
}