ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/DclConnectionArbitrationTests.cs

using System.Collections.Concurrent;
using System.Text.Json;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.TestSupport;
using ZB.MOM.WW.ScadaBridge.TestSupport;
using RemoveConnectionCommand = ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors.RemoveConnectionCommand;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;

/// <summary>
/// Regression tests for deferred register rows <b>89</b> and <b>90</b> — <b>which configuration
/// source owns a live DCL connection</b>.
///
/// <para>
/// <b>The arbitration rule under test.</b> The <c>data_connection_definitions</c> store wins
/// wherever it has an opinion; an instance's flattened <c>Connections</c> entry is a deploy-time
/// SNAPSHOT and is authoritative only for names the store has never owned (the deploy-only site,
/// where that table is legitimately empty). Every ensure path resolves the winner BEFORE hashing,
/// so the single per-name hash slot in <c>_createdConnections</c> can only ever hold one source's
/// configuration — which is what makes the two-source recreate-on-every-flip ping-pong
/// unrepresentable rather than merely avoided.
/// </para>
///
/// <para>
/// <b>Verified topology (this is what the register rows really describe).</b> The Deployment
/// Manager is a cluster singleton scoped to <c>site-{SiteId}</c> and is the ONLY sender of
/// <see cref="CreateConnectionCommand"/> / <see cref="RemoveConnectionCommand"/>; the
/// <c>dcl-manager</c> actor exists on every site node but only the singleton's node ever
/// populates one. So the standby's DCL is not "serving the old endpoint" — it is empty, and the
/// divergence surfaces when the singleton's NEXT incarnation rebuilds every connection from the
/// flattened snapshots and never consults the newer definition rows. Design record:
/// <c>docs/plans/2026-08-17-dcl-connection-arbitration-decision.md</c>.
/// </para>
/// </summary>
public class DclConnectionArbitrationTests : TestKit, IDisposable
{
    private const string ConnectionName = "OPC PLC Simulator";

    /// <summary>The endpoint frozen into an instance's flattened config at its deploy time.</summary>
    private const string FlattenedEndpoint = "opc.tcp://old-plc:4840";

    /// <summary>The endpoint a later, connection-only artifact deploy wrote to the store.</summary>
    private const string DefinitionEndpoint = "opc.tcp://new-plc:4840";

    private readonly SiteStorageService _storage;
    private readonly ScriptCompilationService _compilationService;
    private readonly SharedScriptLibrary _sharedScriptLibrary;
    private readonly TestLocalDb _localDb;

    private readonly ConcurrentQueue<CreateConnectionCommand> _creates = new();
    private readonly ConcurrentQueue<RemoveConnectionCommand> _removes = new();

    public DclConnectionArbitrationTests()
    {
        _localDb = TestLocalDb.CreateTemp("dcl-arbitration-test");
        _storage = new SiteStorageService(_localDb.Db, NullLogger<SiteStorageService>.Instance);
        _storage.InitializeAsync().GetAwaiter().GetResult();
        _compilationService = new ScriptCompilationService(
            NullLogger<ScriptCompilationService>.Instance);
        _sharedScriptLibrary = new SharedScriptLibrary(
            _compilationService, NullLogger<SharedScriptLibrary>.Instance);
    }

    void IDisposable.Dispose()
    {
        // Shut the actors down BEFORE the ILocalDb they write through is disposed.
        Shutdown();
        var path = _localDb.Path;
        _localDb.Dispose();
        TestLocalDb.DeleteFiles(path);
    }

    /// <summary>Shortest sweep the (whole-second) option allows.</summary>
    private const int SweepSeconds = 1;

    /// <summary>Long enough for at least three sweep ticks to have come and gone.</summary>
    private static readonly TimeSpan SeveralSweeps = TimeSpan.FromSeconds(4);

    /// <summary>
    /// Starts a Deployment Manager wired to a recording stand-in for the DCL manager, and blocks
    /// until its boot-time store read has PROVABLY completed (see
    /// <see cref="DeploymentManagerStartupProbe"/>). Rows written before that point are consumed
    /// by the boot load, which is a different path from the post-boot sweep — the gate is what
    /// keeps each test proving the path it names.
    /// </summary>
    private IActorRef CreateBootedDeploymentManager(int sweepSeconds = SweepSeconds)
    {
        var dclSpy = ActorOf(Props.Create(() => new DclConnectionSpy(_creates, _removes)));
        var health = new DeploymentManagerStartupProbe();

        var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            new SiteRuntimeOptions { InstanceSweepIntervalSeconds = sweepSeconds },
            NullLogger<DeploymentManagerActor>.Instance,
            dclManager: dclSpy,
            healthCollector: health,
            serviceProvider: null,
            loggerFactory: null,
            configFetcher: null)));

        AwaitCondition(() => health.ReportCount > 0, TimeSpan.FromSeconds(20),
            TimeSpan.FromMilliseconds(25));
        return dm;
    }

    /// <summary>
    /// A flattened config carrying one data connection. The <c>Connections</c> map is the
    /// deploy-time snapshot the arbitration rule demotes whenever a definition row exists.
    /// </summary>
    private static string MakeConfigJson(string instanceName, string endpoint = FlattenedEndpoint) =>
        JsonSerializer.Serialize(new FlattenedConfiguration
        {
            InstanceUniqueName = instanceName,
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "TestAttr", Value = "42", DataType = "Int32"
                }
            ],
            Connections = new Dictionary<string, ConnectionConfig>
            {
                [ConnectionName] = new()
                {
                    Protocol = "OpcUa",
                    ConfigurationJson = EndpointJson(endpoint),
                    FailoverRetryCount = 3
                }
            }
        });

    private static string EndpointJson(string endpoint) => $"{{\"endpointUrl\":\"{endpoint}\"}}";

    /// <summary>Writes the <c>data_connection_definitions</c> row an artifact deploy would.</summary>
    private Task StoreDefinitionAsync(string endpoint) =>
        _storage.StoreDataConnectionDefinitionAsync(
            ConnectionName, "OpcUa", EndpointJson(endpoint), null, 3);

    private IReadOnlyList<CreateConnectionCommand> CreatesFor(string connectionName) =>
        _creates.Where(c => c.ConnectionName == connectionName).ToArray();

    /// <summary>
    /// The endpoint the DCL was actually told to connect to — the whole point of the arbitration.
    /// </summary>
    private static string EndpointOf(CreateConnectionCommand command) =>
        command.PrimaryConnectionDetails["endpoint"];

    private async Task AwaitCreateCountAsync(int count, string because) =>
        await AwaitAssertAsync(() =>
        {
            Assert.True(CreatesFor(ConnectionName).Count >= count, because);
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

    // ── Rule 1: a definition row outranks an older flattened snapshot ──

    /// <summary>
    /// <b>THE row-89 reproduction, in its real (failover / restart) shape.</b> The store holds a
    /// definition row pointing at the new endpoint — put there by a connection-only artifact
    /// deploy and, on a replicated pair, delivered to this node by LocalDb CDC — while the
    /// instance's own flattened config still carries the endpoint frozen in at ITS deploy time.
    /// A singleton incarnation booting here rebuilds the entire DCL from those flattened
    /// snapshots, so this is exactly what the peer node does the moment it takes over.
    ///
    /// <para>
    /// <b>Fails before the fix:</b> the create carries <c>old-plc</c>, because nothing in the
    /// runtime ever read <c>data_connection_definitions</c> — the table had no production reader
    /// at all, so the artifact deploy's change survived only inside the singleton incarnation
    /// that applied it.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DefinitionRow_WinsOverAnOlderFlattenedSnapshot_WhenTheSingletonBoots()
    {
        const string instanceName = "ArbitratedPump";

        await StoreDefinitionAsync(DefinitionEndpoint);
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-a", "rev-a", true);

        CreateBootedDeploymentManager(sweepSeconds: 0);

        await AwaitCreateCountAsync(1, "the boot load never created the connection");

        Assert.Equal(DefinitionEndpoint, EndpointOf(CreatesFor(ConnectionName).Single()));
    }

    /// <summary>
    /// <b>The ping-pong guard.</b> With the two sources disagreeing, the naive form of this fix —
    /// a sweep pass that re-hashes the connection from the definitions table while the deploy
    /// paths re-hash it from the flattened configs — flips the single per-name hash slot on every
    /// tick, and each flip is another <see cref="CreateConnectionCommand"/>, i.e. the live
    /// connection actor recreated and every subscription on it dropped, forever.
    ///
    /// <para>
    /// This passes both before and after the fix (before, because nothing read the definitions at
    /// all) and is the reason the ensure paths resolve the WINNER before hashing rather than
    /// guarding the flip after the fact.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DisagreeingSources_DoNotPingPongTheConnection_AcrossManySweeps()
    {
        const string instanceName = "NoPingPongPump";

        await StoreDefinitionAsync(DefinitionEndpoint);
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-p", "rev-p", true);

        CreateBootedDeploymentManager();

        await AwaitCreateCountAsync(1, "the boot load never created the connection");
        await Task.Delay(SeveralSweeps);

        Assert.Single(CreatesFor(ConnectionName));
        Assert.Empty(_removes);
    }

    // ── Rule 2: no definition row means the flattened config is the source, unchanged ──

    /// <summary>
    /// The deploy-only site: <c>data_connection_definitions</c> is EMPTY because nothing has ever
    /// sent this site a system-wide artifact deploy, and every connection that matters lives
    /// inline in the instance configs. Rule 2 says the flattened config is the source there — not
    /// as a fallback, but as the correct answer — so this must behave exactly as it did before.
    ///
    /// <para>Passes before and after: a deploy-only regression guard.</para>
    /// </summary>
    [Fact]
    public async Task NoDefinitionRow_LeavesTheFlattenedConfigAuthoritative_AtBoot()
    {
        const string instanceName = "DeployOnlyPump";

        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-d", "rev-d", true);

        CreateBootedDeploymentManager(sweepSeconds: 0);

        await AwaitCreateCountAsync(1, "the boot load never created the connection");

        Assert.Equal(FlattenedEndpoint, EndpointOf(CreatesFor(ConnectionName).Single()));
    }

    /// <summary>
    /// The same rule on the POST-boot sweep path (register row 78's start, row 85's connection):
    /// a config that appears in the store on its own starts an instance whose connection, with no
    /// definition row anywhere, still comes from that instance's flattened config.
    ///
    /// <para>Passes before and after: a deploy-only regression guard for the sweep path.</para>
    /// </summary>
    [Fact]
    public async Task NoDefinitionRow_LeavesTheFlattenedConfigAuthoritative_OnTheSweepPath()
    {
        const string instanceName = "DeployOnlySweptPump";
        CreateBootedDeploymentManager();

        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-ds", "rev-ds", true);

        await AwaitCreateCountAsync(1, "the sweep never created the connection");

        Assert.Equal(FlattenedEndpoint, EndpointOf(CreatesFor(ConnectionName).Single()));
    }

    // ── Row 89's live half: a definition arriving after boot converges on the next tick ──

    /// <summary>
    /// <b>A definition row appearing in the store while the node runs</b> — what LocalDb's CDC
    /// applier does on the peer of a replicated pair when a connection-only artifact deploy lands
    /// on the other node. There is no cross-node message for this and none is needed: the sweep
    /// re-reads the table on the next tick, adopts the row into the arbitration overlay, and
    /// pushes the new configuration for the connection the runtime already holds.
    ///
    /// <para>
    /// <b>Fails before the fix:</b> exactly one create is ever issued, carrying <c>old-plc</c>,
    /// and no sweep tick ever looks at the definitions table.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DefinitionArrivingAfterBoot_IsAdoptedByTheNextSweepTick()
    {
        const string instanceName = "CdcDeliveredPump";
        CreateBootedDeploymentManager();

        // The instance starts first, from its own (older) flattened snapshot.
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-c", "rev-c", true);
        await AwaitCreateCountAsync(1, "the sweep never created the connection");
        Assert.Equal(FlattenedEndpoint, EndpointOf(CreatesFor(ConnectionName).First()));

        // Now the definition row lands — a bare storage write, the CDC applier's only side effect.
        await StoreDefinitionAsync(DefinitionEndpoint);

        await AwaitCreateCountAsync(2,
            "a definition row that arrived by replication was never picked up: the live DCL kept "
            + "the endpoint frozen into an instance's flattened config (register row 89)");

        Assert.Equal(DefinitionEndpoint, EndpointOf(CreatesFor(ConnectionName).Last()));

        // And it settles: the overlay now owns the name, so further ticks are silent.
        await Task.Delay(SeveralSweeps);
        Assert.Equal(2, CreatesFor(ConnectionName).Count);
    }

    // ── Row 90: removal, and not being resurrected by a flattened snapshot ──

    /// <summary>
    /// <b>THE row-90 reproduction.</b> A definition row this incarnation had adopted disappears
    /// from the store — a central delete, reconciled by an artifact deploy on the other node and
    /// replicated here as a row deletion. The sweep can PROVE that removal (the store owned the
    /// name a tick ago and does not now), so it stops the live connection actor; and the name is
    /// tombstoned, so the next instance start whose flattened config still lists that connection
    /// does not quietly bring it back.
    ///
    /// <para>
    /// <b>Fails before the fix</b> at the removal assert: nothing ever noticed, and the orphan
    /// connection actor stayed live until the node restarted.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DefinitionRowDisappearing_StopsTheConnection_AndItIsNotResurrected()
    {
        const string firstInstance = "RemovedConnPump";
        const string secondInstance = "LaterPump";

        await StoreDefinitionAsync(DefinitionEndpoint);
        await _storage.StoreDeployedConfigAsync(
            firstInstance, MakeConfigJson(firstInstance), "dep-r1", "rev-r1", true);

        CreateBootedDeploymentManager();
        await AwaitCreateCountAsync(1, "the boot load never created the connection");

        // Central deleted the connection; the reconcile (or CDC) removes the row.
        await _storage.DeleteDataConnectionDefinitionsExceptAsync([]);

        await AwaitAssertAsync(() =>
        {
            Assert.True(
                _removes.Any(r => r.ConnectionName == ConnectionName),
                "the definition row vanished and the live DCL connection actor was left running "
                + "(register row 90)");
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

        var createsAfterRemoval = CreatesFor(ConnectionName).Count;

        // A second instance whose flattened config still names the removed connection is started
        // by the sweep. It must NOT resurrect a connection central has deleted.
        await _storage.StoreDeployedConfigAsync(
            secondInstance, MakeConfigJson(secondInstance), "dep-r2", "rev-r2", true);
        await Task.Delay(SeveralSweeps);

        Assert.Equal(createsAfterRemoval, CreatesFor(ConnectionName).Count);
    }

    /// <summary>
    /// The same tombstone, reached through the path that actually removes connections today: an
    /// artifact deploy whose complete data-connection set no longer contains the name, which the
    /// site reconciles by deleting the row and stopping the actor.
    ///
    /// <para>
    /// <b>Fails before the fix</b> at the final assert: the removal evicted the hash slot but
    /// left no memory of WHY, so the very next instance start re-created the connection from that
    /// instance's flattened config — a centrally-deleted connection resurrected within one sweep
    /// tick.
    /// </para>
    /// </summary>
    [Fact]
    public async Task ArtifactReconcileRemoval_IsNotResurrectedByALaterInstanceStart()
    {
        const string instanceName = "PostReconcilePump";
        var dm = CreateBootedDeploymentManager();

        // Artifact deploy 1 — the connection exists centrally.
        dm.Tell(new DeployArtifactsCommand(
            "artifact-1", null, null, null, null,
            [new DataConnectionArtifact(ConnectionName, "OpcUa", EndpointJson(DefinitionEndpoint), null)],
            null, DateTimeOffset.UtcNow));
        Assert.True(ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(20)).Success);

        // Artifact deploy 2 — central no longer has it, so the site reconciles it away.
        dm.Tell(new DeployArtifactsCommand(
            "artifact-2", null, null, null, null, [], null, DateTimeOffset.UtcNow));
        Assert.True(ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(20)).Success);

        await AwaitAssertAsync(() =>
        {
            Assert.True(
                _removes.Any(r => r.ConnectionName == ConnectionName),
                "the artifact reconcile did not stop the removed connection");
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

        var createsAfterRemoval = CreatesFor(ConnectionName).Count;

        // An instance start naming the removed connection must not undo the removal.
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-pr", "rev-pr", true);
        await Task.Delay(SeveralSweeps);

        Assert.Equal(createsAfterRemoval, CreatesFor(ConnectionName).Count);
    }

    /// <summary>
    /// The tombstone must not be permanent: a definition row IS the store re-asserting ownership
    /// of the name, so a connection central deletes and then re-adds is creatable again.
    ///
    /// <para>
    /// <b>Note what the re-add does and does not do.</b> The overlay is an arbiter, not a creation
    /// source — clearing the tombstone does not by itself resurrect a connection actor, exactly as
    /// a definition row for a name no instance references creates nothing. The connection comes
    /// back with the next instance start, and it comes back at the RE-ADDED endpoint, which is
    /// what proves both halves: the tombstone was cleared, and rule 1 still governs the winner.
    /// This mirrors the removal's own long-standing semantic — a removed connection's attributes
    /// stay bad quality until the referencing instance is redeployed.
    /// </para>
    /// </summary>
    [Fact]
    public async Task ReAddedDefinition_ClearsTheTombstone_SoTheNextInstanceStartRecreatesIt()
    {
        const string firstInstance = "ReAddedPump";
        const string laterInstance = "ReAddedLaterPump";

        await StoreDefinitionAsync(DefinitionEndpoint);
        await _storage.StoreDeployedConfigAsync(
            firstInstance, MakeConfigJson(firstInstance), "dep-ra", "rev-ra", true);

        CreateBootedDeploymentManager();
        await AwaitCreateCountAsync(1, "the boot load never created the connection");

        await _storage.DeleteDataConnectionDefinitionsExceptAsync([]);
        await AwaitAssertAsync(() =>
        {
            Assert.True(_removes.Any(r => r.ConnectionName == ConnectionName),
                "the removal never happened");
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

        // Central re-adds it, at a third endpoint.
        const string reAddedEndpoint = "opc.tcp://re-added-plc:4840";
        await StoreDefinitionAsync(reAddedEndpoint);

        // A later instance start is what actually wants a connection actor again.
        await _storage.StoreDeployedConfigAsync(
            laterInstance, MakeConfigJson(laterInstance), "dep-ral", "rev-ral", true);

        await AwaitCreateCountAsync(2,
            "the tombstone outlived the re-add, so a connection central had restored stayed dead");
        Assert.Equal(reAddedEndpoint, EndpointOf(CreatesFor(ConnectionName).Last()));
    }

    /// <summary>
    /// Stand-in for the DCL manager actor: records the two commands the Deployment Manager is the
    /// sole sender of, and swallows everything else (Instance Actors send their tag subscribes
    /// here too, and those are not what these tests assert on).
    /// </summary>
    private sealed class DclConnectionSpy : ReceiveActor
    {
        public DclConnectionSpy(
            ConcurrentQueue<CreateConnectionCommand> creates,
            ConcurrentQueue<RemoveConnectionCommand> removes)
        {
            Receive<CreateConnectionCommand>(creates.Enqueue);
            Receive<RemoveConnectionCommand>(removes.Enqueue);
            ReceiveAny(_ => { });
        }
    }
}