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>, <b>90</b> and <b>97</b> — <b>which
/// configuration source owns a live DCL connection, and how a change reaches it</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>
///
/// <para>
/// <b>Row 97 — the third command.</b> Resolving the winner decides WHAT to push; it took a
/// <see cref="ReconfigureConnectionCommand"/> to make a push against an ALREADY-LIVE connection
/// mean anything at all, because the DCL manager's create handler warns and returns for a name it
/// already holds. So a name the hash cache has never held gets a create, a changed one gets a
/// reconfigure, and an unchanged one gets nothing.
/// </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<ReconfigureConnectionCommand> _reconfigures = 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, _reconfigures, _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();

    private IReadOnlyList<ReconfigureConnectionCommand> ReconfiguresFor(string connectionName) =>
        _reconfigures.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"];

    /// <inheritdoc cref="EndpointOf(CreateConnectionCommand)"/>
    private static string EndpointOf(ReconfigureConnectionCommand 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));

    private async Task AwaitReconfigureCountAsync(int count, string because) =>
        await AwaitAssertAsync(() =>
        {
            Assert.True(ReconfiguresFor(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));
        // Row 97 moved the cost of a flip: a ping-pong now shows up as a stream of RECONFIGURES,
        // each of which drops the link and re-subscribes every tag. Assert on both, or the guard
        // silently stops guarding the thing that would actually happen.
        Assert.Empty(ReconfiguresFor(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> nothing is ever issued for the change at all, and no sweep
    /// tick ever looks at the definitions table.
    /// </para>
    ///
    /// <para>
    /// <b>Register row 97 changed the SHAPE of the second push.</b> The connection already exists
    /// at this point, so the change goes out as a
    /// <see cref="ReconfigureConnectionCommand"/> — a second create would be dropped by the DCL
    /// manager's create-only guard and change nothing.
    /// </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 AwaitReconfigureCountAsync(1,
            "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(ReconfiguresFor(ConnectionName).Last()));

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

    // ── Row 97: a changed configuration reaches the LIVE connection ──

    /// <summary>
    /// <b>THE row-97 reproduction on the sender side.</b> A connection the DCL already holds gets
    /// a new endpoint. What must go out is a <see cref="ReconfigureConnectionCommand"/>, which the
    /// live connection actor applies in place by swapping its adapter and re-issuing its
    /// subscriptions.
    ///
    /// <para>
    /// <b>Fails before the fix:</b> the Deployment Manager re-issued a
    /// <see cref="CreateConnectionCommand"/> and logged a Warning saying it would not take effect
    /// — which was accurate, because <c>DataConnectionManagerActor.HandleCreateConnection</c>
    /// warns and returns for a name it already holds. The hash cache recorded the change as
    /// applied regardless, so a corrected endpoint sat unapplied until a node restart, a singleton
    /// failover, or a remove-then-add.
    /// </para>
    /// </summary>
    [Fact]
    public async Task ChangedConfigurationForALiveConnection_IsSentAsAReconfigure_NotASecondCreate()
    {
        const string instanceName = "ReconfiguredPump";
        CreateBootedDeploymentManager();

        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-rc1", "rev-rc1", true);
        await AwaitCreateCountAsync(1, "the sweep never created the connection");

        // The connection-only artifact deploy that corrects the endpoint.
        await StoreDefinitionAsync(DefinitionEndpoint);

        await AwaitReconfigureCountAsync(1,
            "a changed configuration for an ALREADY-LIVE connection was not sent as a "
            + "ReconfigureConnectionCommand, so the live connection actor keeps serving the old "
            + "endpoint until it is rebuilt from scratch (register row 97)");

        Assert.Equal(DefinitionEndpoint, EndpointOf(ReconfiguresFor(ConnectionName).Single()));

        // A create would have been dropped by the manager's create-only guard — so the change must
        // NOT have been shipped as one.
        Assert.Single(CreatesFor(ConnectionName));
    }

    /// <summary>
    /// The hash guard is unchanged by row 97: an UNCHANGED configuration still sends nothing at
    /// all. This is what keeps a healthy site silent across every sweep tick, and it is the reason
    /// a reconfigure — which drops the link and re-subscribes every tag — is safe to send on the
    /// change edge.
    /// </summary>
    [Fact]
    public async Task UnchangedConfiguration_SendsNeitherACreateNorAReconfigure()
    {
        const string instanceName = "SteadyPump";

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

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

        // Re-write the SAME definition row several sweeps running, and start a second instance
        // naming the same connection: neither is a configuration change.
        await StoreDefinitionAsync(DefinitionEndpoint);
        await _storage.StoreDeployedConfigAsync(
            "SteadyPumpTwo", MakeConfigJson("SteadyPumpTwo"), "dep-st2", "rev-st2", true);
        await Task.Delay(SeveralSweeps);

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

    // ── 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>
    /// Deferred register row 101 — the stale-snapshot race, made deterministic. A sweep's store
    /// read runs OFF the actor thread; one that starts before an artifact apply's store writes
    /// can land AFTER <c>HandleApplyArtifactDataConnectionsToDcl</c> populated the overlay. Its
    /// snapshot then lacks the just-applied rows, and the disappearance detection would tombstone
    /// and STOP the just-created live connections — which nothing re-creates until an instance
    /// redeploy, because the removal also evicts the hash slot the overlay's push is gated on.
    ///
    /// <para>
    /// The crafted <c>InstanceSweepLoaded</c> below IS that race: an empty snapshot stamped with
    /// the pre-apply write generation, delivered after the apply. <b>Fails before the fix</b>:
    /// without the generation guard the actor removes the connection it created a moment ago.
    /// </para>
    /// </summary>
    [Fact]
    public async Task StaleSweepSnapshot_RacingAnArtifactApply_DoesNotRemoveJustAppliedConnections()
    {
        // Sweep effectively disabled so the crafted message is the only sweep result in play.
        var dm = CreateBootedDeploymentManager(sweepSeconds: 3600);

        dm.Tell(new DeployArtifactsCommand(
            "artifact-race-1", null, null, null, null,
            [new DataConnectionArtifact(ConnectionName, "OpcUa", EndpointJson(DefinitionEndpoint), null)],
            null, DateTimeOffset.UtcNow));
        Assert.True(ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(20)).Success);
        await AwaitCreateCountAsync(1, "the artifact apply never created the connection");

        // Generation 0 = captured before the apply bumped it; empty list = the pre-write table.
        dm.Tell(new DeploymentManagerActor.InstanceSweepLoaded(
            [], null, ConnectionDefinitions: [], DefinitionsGenerationAtRead: 0));

        await Task.Delay(TimeSpan.FromSeconds(2));
        Assert.DoesNotContain(_removes, r => r.ConnectionName == ConnectionName);
    }

    /// <summary>
    /// The row-101 guard must not blunt row 90's legitimate disappearance detection: the same
    /// empty snapshot stamped with the CURRENT write generation is a real answer (the store
    /// genuinely holds no rows any more) and still stops the connection.
    /// </summary>
    [Fact]
    public async Task FreshEmptySweepSnapshot_StillRemovesDisappearedConnections()
    {
        var dm = CreateBootedDeploymentManager(sweepSeconds: 3600);

        dm.Tell(new DeployArtifactsCommand(
            "artifact-race-2", null, null, null, null,
            [new DataConnectionArtifact(ConnectionName, "OpcUa", EndpointJson(DefinitionEndpoint), null)],
            null, DateTimeOffset.UtcNow));
        Assert.True(ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(20)).Success);
        await AwaitCreateCountAsync(1, "the artifact apply never created the connection");

        // Exactly one artifact apply has run, so the current write generation is 1.
        dm.Tell(new DeploymentManagerActor.InstanceSweepLoaded(
            [], null, ConnectionDefinitions: [], DefinitionsGenerationAtRead: 1));

        await AwaitAssertAsync(() =>
        {
            Assert.Contains(_removes, r => r.ConnectionName == ConnectionName);
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(10));
    }

    /// <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<ReconfigureConnectionCommand> reconfigures,
            ConcurrentQueue<RemoveConnectionCommand> removes)
        {
            Receive<CreateConnectionCommand>(creates.Enqueue);
            Receive<ReconfigureConnectionCommand>(reconfigures.Enqueue);
            Receive<RemoveConnectionCommand>(removes.Enqueue);
            ReceiveAny(_ => { });
        }
    }
}