lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs

using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Serilog.Core;
using Serilog.Events;
using Shouldly;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;

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

/// <summary>
///     LocalDb Phase 1 (Task 2) — pins the deployment-cache schema and the load-bearing
///     <c>DDL → RegisterReplicated → writes</c> ordering inside <see cref="LocalDbSetup.OnReady"/>.
/// </summary>
/// <remarks>
///     <para>
///         The ordering assertion is the point of this file. <c>RegisterReplicated</c> installs the
///         capture triggers; rows written <b>before</b> it are never captured into the oplog and
///         therefore never reach the peer — silently, and forever. A reordered <c>OnReady</c> passes
///         every schema-shape test while shipping a replication path that quietly moves nothing.
///     </para>
///     <para>
///         <b>BEHAVIOUR CHANGE (2026-08-18): registration is CONDITIONAL on this node actually
///         being configured to replicate.</b> Until now <c>OnReady</c> registered all four tables
///         unconditionally, so every test here could build a database from <c>LocalDb:Path</c>
///         alone and still find them registered. They now pass a replication section, because that
///         is what makes registration happen at all — see
///         <see cref="OnReady_WhenReplicationIsNotConfigured_RegistersNothingAndInstallsNoTriggers"/>
///         for the other branch. The registered-table-set pin is therefore now a pin on
///         <b>configured</b> nodes; the unconfigured node's pin is the deregistration self-heal.
///     </para>
///     <para>
///         Built against a real temp-file <see cref="ILocalDb"/> rather than a hand-written schema:
///         the library has no in-memory mode, and asserting against a schema the test itself wrote
///         would only prove the test agrees with itself. That is doubly true here — the baseline and
///         deregistration behaviours live entirely inside the library, so only a real file can show
///         them.
///     </para>
/// </remarks>
public sealed class LocalDbSetupTests : IDisposable
{
    /// <summary>
    ///     The initiating half of a replicating pair: it dials, so it carries <c>PeerAddress</c>.
    /// </summary>
    private static readonly KeyValuePair<string, string?>[] InitiatorReplication =
    [
        new("LocalDb:Replication:PeerAddress", "http://peer:9001"),
        new("LocalDb:Replication:ApiKey", "test-sync-key"),
    ];

    /// <summary>
    ///     The passive half of the same pair. It never dials, so it has <b>no</b> <c>PeerAddress</c>
    ///     — only the shared key (plus a listen port, which is Program.cs's concern). Keying the
    ///     predicate on <c>PeerAddress</c> alone would strip capture from this node and leave the
    ///     pair converging in one direction only.
    /// </summary>
    private static readonly KeyValuePair<string, string?>[] PassiveReplication =
    [
        new("LocalDb:Replication:ApiKey", "test-sync-key"),
    ];

    private readonly string _dbPath =
        Path.Combine(Path.GetTempPath(), $"otopcua-localdb-setup-{Guid.NewGuid():N}.db");

    private readonly List<ServiceProvider> _providers = [];

    /// <summary>
    ///     Boots <see cref="LocalDbSetup.OnReady"/> over the test's database file.
    /// </summary>
    /// <remarks>
    ///     Successive calls reopen the <b>same</b> file, which is how the self-heal and late-opt-in
    ///     cases are expressed: boot once under one replication configuration, write rows, then boot
    ///     again under another. That is exactly the sequence an operator performs by editing config
    ///     and restarting the node.
    /// </remarks>
    private ILocalDb Boot(
        KeyValuePair<string, string?>[]? replication = null,
        Serilog.ILogger? logger = null)
    {
        // A previous boot's provider still holds the file open; the library has no second-handle
        // mode. Closing it (and the pool) is what makes a reopen behave like a process restart.
        DisposeProviders();

        var settings = new List<KeyValuePair<string, string?>>
        {
            new("LocalDb:Path", _dbPath),
        };
        settings.AddRange(replication ?? []);

        var configuration = new ConfigurationBuilder()
            .AddInMemoryCollection(settings)
            .Build();

        var provider = new ServiceCollection()
            .AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration, logger))
            .BuildServiceProvider();

        _providers.Add(provider);

        return provider.GetRequiredService<ILocalDb>();
    }

    private static async Task<int> ScalarAsync(ILocalDb db, string sql) =>
        (await db.QueryAsync(sql, r => r.GetInt32(0), parameters: null, TestContext.Current.CancellationToken))[0];

    private static Task InsertPointerAsync(ILocalDb db, string clusterId) =>
        db.ExecuteAsync(
            """
            INSERT INTO deployment_pointer
                (cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc)
            VALUES (@ClusterId, @DeploymentId, @RevisionHash, @Sha, @AppliedAtUtc)
            """,
            new
            {
                ClusterId = clusterId,
                DeploymentId = "0123456789abcdef0123456789abcdef",
                RevisionHash = new string('a', 64),
                Sha = new string('b', 64),
                AppliedAtUtc = "2026-07-20T00:00:00.0000000Z",
            },
            TestContext.Current.CancellationToken);

    [Fact]
    public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFourReplicatedTables()
    {
        // Both directions are load-bearing. "No fewer" catches a dropped RegisterReplicated call
        // (that table then never replicates). "No more" catches an accidental registration —
        // every replicated table costs three triggers and oplog volume on every write.
        var db = Boot(InitiatorReplication);

        db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
            .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
    }

    [Fact]
    public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFourTables()
    {
        // The passive half of a pair carries the key and nothing else. If the predicate were
        // "PeerAddress is set", this node would install no capture triggers and its OWN writes
        // would never reach the initiator — a pair that replicates in one direction only, with
        // nothing logged anywhere. This test is the falsifier for that narrower predicate.
        var db = Boot(PassiveReplication);

        db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal)
            .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]);
    }

    [Fact]
    public async Task OnReady_WhenReplicationIsNotConfigured_RegistersNothingAndInstallsNoTriggers()
    {
        // The default-OFF node. It has no peer, so capture buys it nothing and costs it two INSERTs
        // plus a json_object of the whole row inside every write transaction — for an oplog that is
        // never drained. Asserting the trigger count as well as ReplicatedTables matters: the
        // dictionary is in-process bookkeeping, the triggers are what actually run.
        var db = Boot();

        db.ReplicatedTables.ShouldBeEmpty();

        // None of the three application schemas defines a trigger of its own, so every trigger in
        // this file would be a capture trigger.
        (await ScalarAsync(db, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'")).ShouldBe(0);
    }

    [Fact]
    public void DeploymentArtifacts_PkIsDeploymentIdPlusChunkIndex()
    {
        // The composite PK is what makes an artifact chunkable. A single-column PK here would
        // make every chunk of a deployment collide onto one row under LWW.
        var db = Boot(InitiatorReplication);

        db.ReplicatedTables["deployment_artifacts"].PkColumns
            .ShouldBe(["deployment_id", "chunk_index"]);
    }

    [Fact]
    public void DeploymentPointer_PkIsClusterId()
    {
        // One current-deployment pointer per cluster: the pair's two nodes converge on the same
        // row rather than each keeping its own.
        var db = Boot(InitiatorReplication);

        db.ReplicatedTables["deployment_pointer"].PkColumns.ShouldBe(["cluster_id"]);
    }

    [Fact]
    public void AlarmSfEvents_PkIsTheAppMintedId()
    {
        // A single TEXT PK the application mints. The legacy queue this table replaces keyed on
        // `RowId INTEGER PRIMARY KEY AUTOINCREMENT`, which LWW cannot replicate: two nodes would
        // independently allocate rowid 7 for different alarms and silently overwrite each other.
        var db = Boot(InitiatorReplication);

        db.ReplicatedTables["alarm_sf_events"].PkColumns.ShouldBe(["id"]);
    }

    [Fact]
    public void AlarmConditionState_PkIsTheScriptedAlarmId()
    {
        // One condition-state row per alarm identity. The pair's two nodes converge on the same row
        // under LWW rather than each keeping its own, and Save is an unconditional upsert onto it.
        var db = Boot(InitiatorReplication);

        db.ReplicatedTables["alarm_condition_state"].PkColumns.ShouldBe(["scripted_alarm_id"]);
    }

    [Fact]
    public async Task AlarmSfEventRows_EnterTheOplog()
    {
        // Same ordering assertion as the deployment-pointer test below, for the table Phase 2 adds.
        // Worth its own case because the alarm table is registered by a different call site, and a
        // registration added ahead of its DDL would fail loudly while one added after the migrator
        // would fail silently.
        var db = Boot(InitiatorReplication);

        await db.ExecuteAsync(
            """
            INSERT INTO alarm_sf_events
                (id, alarm_id, enqueued_at_utc, payload_json, attempt_count)
            VALUES (@Id, @AlarmId, @EnqueuedAtUtc, @PayloadJson, 0)
            """,
            new
            {
                Id = new string('c', 64),
                AlarmId = "equip/alarm-1",
                EnqueuedAtUtc = "2026-07-21T00:00:00.0000000Z",
                PayloadJson = """{"AlarmId":"equip/alarm-1"}""",
            },
            TestContext.Current.CancellationToken);

        (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_sf_events'"))
            .ShouldBe(1);
    }

    [Fact]
    public async Task RowsWrittenAfterOnReady_EnterTheOplog()
    {
        // THE ordering assertion. If DDL and RegisterReplicated were swapped — or a write were
        // added to OnReady ahead of registration — the triggers would not exist at write time and
        // this count would stay 0 while every other test in this file still passed.
        var db = Boot(InitiatorReplication);

        await InsertPointerAsync(db, "SITE-A");

        (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog")).ShouldBeGreaterThanOrEqualTo(1);
    }

    [Fact]
    public async Task EnablingReplicationLater_BaselinesTheRowsAlreadyInTheFile()
    {
        // The defect this whole change exists for. Capture is change-data-capture and the snapshot
        // streamer pages from __localdb_row_version, so rows written while replication was off are
        // in neither — a late opt-in used to converge on subsequent writes only, silently and
        // forever. RegisterReplicated(table, baselineExistingRows: true) seeds the ledger at the LWW
        // floor and flags a resync so they actually travel.
        //
        // The assertion is on the ledger entry's HLC, not merely on its existence. A row captured by
        // a trigger carries a genuine (large, non-zero) stamp; a BASELINED row is seeded at the LWW
        // floor of HLC 0, which is what makes it lose to any real remote write of the same key and
        // win only where the peer holds no version at all. Asserting existence alone would pass
        // against the old unconditional registration, where the trigger simply captured the write —
        // i.e. it would prove nothing about the flag.
        var unreplicated = Boot();
        await InsertPointerAsync(unreplicated, "SITE-A");

        var replicated = Boot(InitiatorReplication);

        (await ScalarAsync(
                replicated,
                """
                SELECT COUNT(*) FROM __localdb_row_version
                WHERE table_name = 'deployment_pointer' AND hlc = 0
                """))
            .ShouldBe(1);
    }

    [Fact]
    public async Task DisablingReplication_DropsTheCaptureTriggersAnEarlierBootInstalled()
    {
        // The self-heal half. Merely declining to register would leave a file registered by an
        // earlier build capturing forever — on precisely the node that has no peer to capture for.
        // DeregisterReplicated drops the triggers and prunes that table's oplog and row-version
        // rows.
        var replicated = Boot(InitiatorReplication);
        await InsertPointerAsync(replicated, "SITE-A");
        (await ScalarAsync(replicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'"))
            .ShouldBe(12);

        var unreplicated = Boot();

        unreplicated.ReplicatedTables.ShouldBeEmpty();
        (await ScalarAsync(unreplicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'"))
            .ShouldBe(0);
    }

    [Fact]
    public async Task DisablingReplication_LogsOnceWhenItActuallyCleanedSomething()
    {
        // An operator-visible record that a node just stopped capturing, and how much it dropped.
        // Silent trigger removal on a node someone MEANT to replicate is the failure this line
        // exists to make diagnosable.
        var replicated = Boot(InitiatorReplication);
        await InsertPointerAsync(replicated, "SITE-A");

        var sink = new CollectingSink();
        Boot(replication: null, logger: BuildLogger(sink));

        sink.Events.Count(e => e.Level == LogEventLevel.Information).ShouldBe(1);
    }

    [Fact]
    public void DisablingReplication_SaysNothingWhenThereWasNothingToClean()
    {
        // The steady state of every default-OFF node in the fleet: it boots unreplicated every day
        // and has nothing to deregister. A log line here would be a daily false alarm, which is how
        // the real one gets ignored.
        var sink = new CollectingSink();
        Boot(replication: null, logger: BuildLogger(sink));

        sink.Events.ShouldBeEmpty();
    }

    private static Serilog.ILogger BuildLogger(ILogEventSink sink) =>
        new Serilog.LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(sink).CreateLogger();

    /// <summary>Captures whatever <see cref="LocalDbSetup.OnReady"/> logs.</summary>
    private sealed class CollectingSink : ILogEventSink
    {
        public List<LogEvent> Events { get; } = [];

        public void Emit(LogEvent logEvent) => Events.Add(logEvent);
    }

    private void DisposeProviders()
    {
        foreach (var provider in _providers)
            provider.Dispose();

        _providers.Clear();

        // No in-memory mode, so these are real files. Pooled connections keep a handle open past
        // dispose; clearing the pools first is what lets a reopen (or the delete below) succeed.
        SqliteConnection.ClearAllPools();
    }

    public void Dispose()
    {
        DisposeProviders();

        foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" })
        {
            if (File.Exists(path))
                File.Delete(path);
        }
    }
}