using System.Text.Json;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
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;

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

/// <summary>
/// Regression tests for deferred register rows <b>78</b> and <b>72</b> — the two ways the site
/// runtime silently diverges from its own durable <c>deployed_configurations</c> rows.
///
/// <para>
/// <b>Row 78 (availability).</b> Instance Actors were created from three paths: the boot-time
/// SQLite load, an explicit deploy/enable command, and (row 67) the startup reconcile's
/// <see cref="EnsureInstanceStartedCommand"/> for configs the reconcile ITSELF fetched. A config
/// that arrives by LocalDb CDC <i>replication</i> hits none of them: the sync applier writes the
/// row straight into the file, AFTER the boot load has read storage and invisibly to the
/// reconcile, which correctly reports no gap because central's view of the site is right. Proven
/// live 2026-08-15 — a fresh site-a-a was seeded from its peer within ~5 s of boot and its
/// enabled instances sat dead, with their configs present, until another restart.
/// </para>
///
/// <para>
/// <b>Row 72 (silent divergence).</b> The reconcile refreshes a STALE row via
/// <c>StoreDeployedConfigIfNewerAsync</c>, but the running Instance Actor keeps serving the OLD
/// compiled config until the next restart or redeploy, and nothing surfaced
/// "stored revision != running revision".
/// </para>
///
/// <para>
/// Both are closed by one periodic sweep owned by the Deployment Manager singleton. Design note:
/// <c>docs/plans/2026-08-15-instance-sweep-self-heal-design.md</c>.
/// </para>
/// </summary>
public class InstanceSweepSelfHealTests : TestKit, IDisposable
{
    private readonly SiteStorageService _storage;
    private readonly ScriptCompilationService _compilationService;
    private readonly SharedScriptLibrary _sharedScriptLibrary;
    private readonly TestLocalDb _localDb;

    public InstanceSweepSelfHealTests()
    {
        _localDb = TestLocalDb.CreateTemp("instance-sweep-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>
    /// Sweep interval used by these tests. One second is the shortest the (whole-second) option
    /// allows; every assertion below therefore budgets several seconds so it is waiting on the
    /// behaviour, not on the clock.
    /// </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 and blocks until its boot-time config load has PROVABLY read
    /// storage (its first instance-count report). Load-bearing for every test here: a row written
    /// before that read is picked up by the boot load itself, which is precisely the path row 78
    /// is about NOT having. Without this gate a test would silently prove the boot load works.
    /// </summary>
    private IActorRef CreateBootedDeploymentManager(
        FakeSiteEventLogger? siteLog = null,
        int sweepSeconds = SweepSeconds)
    {
        var health = new DeploymentManagerStartupProbe();
        var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            new SiteRuntimeOptions { InstanceSweepIntervalSeconds = sweepSeconds },
            NullLogger<DeploymentManagerActor>.Instance,
            dclManager: null,
            healthCollector: health,
            serviceProvider: siteLog == null ? null : new SingleServiceProvider(siteLog),
            loggerFactory: null,
            configFetcher: null)));

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

    private static string MakeConfigJson(string instanceName, string attributeValue = "42") =>
        JsonSerializer.Serialize(new FlattenedConfiguration
        {
            InstanceUniqueName = instanceName,
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "TestAttr", Value = attributeValue, DataType = "Int32"
                }
            ]
        });

    /// <summary>
    /// A config carrying a real script, so its compile gate / compile warm takes long enough for
    /// the deploy pipeline to hold the instance across a sweep tick.
    /// </summary>
    private static string MakeScriptedConfigJson(string instanceName, string scriptCode) =>
        JsonSerializer.Serialize(new FlattenedConfiguration
        {
            InstanceUniqueName = instanceName,
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "TestAttr", Value = "42", DataType = "Int32"
                }
            ],
            Scripts =
            [
                new ResolvedScript
                {
                    CanonicalName = "Worker", Code = scriptCode, TriggerType = "Call"
                }
            ]
        });

    /// <summary>
    /// Asks the Deployment Manager for a debug snapshot — the exact probe that answered
    /// <c>instanceNotFound</c> on the live rig.
    /// </summary>
    private async Task<DebugViewSnapshot> SnapshotAsync(IActorRef dm, string instanceName) =>
        await dm.Ask<DebugViewSnapshot>(
            new SubscribeDebugViewRequest(instanceName, Guid.NewGuid().ToString()),
            TimeSpan.FromSeconds(5));

    private async Task<bool> InstanceIsRunningAsync(IActorRef dm, string instanceName) =>
        !(await SnapshotAsync(dm, instanceName)).InstanceNotFound;

    private async Task<IActorRef> ResolveInstanceChildAsync(IActorRef dm, string instanceName) =>
        await Sys.ActorSelection(dm.Path / instanceName).ResolveOne(TimeSpan.FromSeconds(5));

    /// <summary>
    /// Site events announcing a sweep REPAIR (a start or a stale-revision restart). The terminal
    /// "self-heal failed" event is deliberately excluded — it reports the outcome of a repair
    /// already counted here, so counting both would double-count a single attempt.
    /// </summary>
    private static IReadOnlyList<FakeSiteEventLogger.Entry> SelfHealEvents(
        FakeSiteEventLogger log, string instanceName) =>
        log.OfType("deployment")
            .Where(e => e.InstanceId == instanceName &&
                        e.Message.Contains("self-heal", StringComparison.OrdinalIgnoreCase) &&
                        !e.Message.Contains("self-heal failed", StringComparison.OrdinalIgnoreCase))
            .ToArray();

    /// <summary>
    /// Deploys an instance the ordinary way and blocks until it is running, so a test can start
    /// from a known-good runtime state.
    /// </summary>
    private async Task DeployAndAwaitAsync(
        IActorRef dm, string instanceName, string revision, string configJson)
    {
        dm.Tell(new DeployInstanceCommand(
            $"dep-{revision}", instanceName, revision, configJson, "admin", DateTimeOffset.UtcNow));
        var response = ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(15));
        Assert.Equal(DeploymentStatus.Success, response.Status);
        await AwaitAssertAsync(async () =>
            Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(10));
    }

    // ── Row 78: a config that arrives with no deploy, no enable and no reconcile ──

    /// <summary>
    /// THE row-78 reproduction. The node boots with an empty deployed-config store (so the
    /// boot-time load starts nothing), and the config for an ENABLED instance then appears in
    /// storage on its own — exactly what LocalDb's CDC sync applier does when it seeds a fresh
    /// node from its peer. No <see cref="DeployInstanceCommand"/>, no
    /// <see cref="EnsureInstanceStartedCommand"/>, no reconcile actor: nothing but the row.
    ///
    /// <para>
    /// <b>Fails before the fix</b> at the final assert — the row is in storage and the debug
    /// snapshot still answers <c>InstanceNotFound = true</c>, the live symptom verbatim.
    /// </para>
    /// </summary>
    [Fact]
    public async Task Sweep_StartsEnabledInstance_WhoseConfigArrivedWithoutADeployOrReconcile()
    {
        const string instanceName = "ReplicatedPump";

        var dm = CreateBootedDeploymentManager();

        // Baseline: the node genuinely booted with nothing.
        await AwaitAssertAsync(async () =>
            Assert.False(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(5));

        // Replication seeds the row. This is a bare storage write on purpose — it is the only
        // side effect the CDC applier has, and it is invisible to every other start path.
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-replicated", "rev-replicated", true);

        await AwaitAssertAsync(async () =>
            Assert.True(
                await InstanceIsRunningAsync(dm, instanceName),
                "A replication-seeded config for an enabled instance left no running Instance "
                + "Actor — the instance would stay dead until the next restart (register row 78)"),
            TimeSpan.FromSeconds(20));
    }

    /// <summary>
    /// The heal must be attributable: it emits the same Warning <c>deployment</c> site event
    /// pattern as row 67, with the trigger named so an operator can tell a sweep heal from a
    /// startup-reconcile heal.
    /// </summary>
    [Fact]
    public async Task SweepStartedInstance_EmitsAWarningSiteEvent_NamingTheSweepTrigger()
    {
        const string instanceName = "SurfacedSweepPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

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

        await AwaitAssertAsync(() =>
        {
            Assert.Contains(siteLog.OfType("deployment"), e =>
                e.InstanceId == instanceName &&
                e.Severity == "Warning" &&
                e.Source == "DeploymentManagerActor" &&
                e.Message.Contains("deployment-manager sweep", StringComparison.OrdinalIgnoreCase) &&
                e.Message.Contains("self-heal", StringComparison.OrdinalIgnoreCase));
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));
    }

    // ── Row 78 negative guards ──

    /// <summary>
    /// A DISABLED instance legitimately has a config row and no running actor. The sweep must
    /// never start it — that would silently undo an operator's disable.
    /// </summary>
    [Fact]
    public async Task Sweep_NeverStarts_ADisabledInstance()
    {
        const string instanceName = "DisabledSweepPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

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

        await Task.Delay(SeveralSweeps);

        Assert.False(await InstanceIsRunningAsync(dm, instanceName));
        Assert.Empty(SelfHealEvents(siteLog, instanceName));
    }

    /// <summary>
    /// Idempotence: many sweeps, one start. The sweep runs forever on a live node, so a repair
    /// that re-fired every tick would be a restart loop plus an unbounded stream of Warning site
    /// events.
    /// </summary>
    [Fact]
    public async Task Sweep_StartsTheInstanceExactlyOnce_AcrossManySweeps()
    {
        const string instanceName = "IdempotentSweepPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

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

        await AwaitAssertAsync(async () =>
            Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));

        var afterFirstHeal = await ResolveInstanceChildAsync(dm, instanceName);

        // Let several more sweeps run over the now-healthy state.
        await Task.Delay(SeveralSweeps);

        Assert.Equal(afterFirstHeal, await ResolveInstanceChildAsync(dm, instanceName));
        Assert.Single(SelfHealEvents(siteLog, instanceName));
    }

    /// <summary>
    /// A sweep that finds nothing wrong must be completely silent — no site events at all. This
    /// runs on every tick of every healthy site, forever.
    /// </summary>
    [Fact]
    public async Task Sweep_IsSilent_WhenThereIsNothingToHeal()
    {
        const string instanceName = "HealthySweepPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

        await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));

        var eventsAfterDeploy = siteLog.Entries.Count;
        await Task.Delay(SeveralSweeps);

        Assert.Equal(eventsAfterDeploy, siteLog.Entries.Count);
        Assert.True(await InstanceIsRunningAsync(dm, instanceName));
    }

    /// <summary>
    /// The sweep is opt-out: <c>InstanceSweepIntervalSeconds = 0</c> arms no timer at all, so a
    /// site that wants the pre-fix behaviour keeps it (and the option's disable path is proven,
    /// not merely validated).
    /// </summary>
    [Fact]
    public async Task Sweep_DoesNothing_WhenTheIntervalIsZero()
    {
        const string instanceName = "NoSweepPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 0);

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

        await Task.Delay(SeveralSweeps);

        Assert.False(await InstanceIsRunningAsync(dm, instanceName));
        Assert.Empty(SelfHealEvents(siteLog, instanceName));
    }

    /// <summary>
    /// The sweep must defer to any deploy pipeline that owns the instance. A redeploy passes
    /// through a window where the predecessor has been stopped and removed from routing but its
    /// replacement does not exist yet — the sweep would read that as "enabled config, no actor"
    /// and try to create a child whose actor name is still registered.
    ///
    /// <para>
    /// Sweep ticks are fired continuously across the whole redeploy so the window is genuinely
    /// sampled; the assertion is that the redeploy succeeded untouched and the sweep announced
    /// nothing.
    /// </para>
    /// </summary>
    [Fact]
    public async Task Sweep_DefersToAnInFlightRedeploy()
    {
        const string instanceName = "RedeployingPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

        await DeployAndAwaitAsync(
            dm, instanceName, "rev-1",
            MakeScriptedConfigJson(instanceName, "var a = 1; var b = a + 1; return b;"));

        using var ticking = new CancellationTokenSource();
        var ticker = Task.Run(async () =>
        {
            while (!ticking.IsCancellationRequested)
            {
                dm.Tell(DeploymentManagerActor.InstanceSweepTick.Instance);
                await Task.Delay(5);
            }
        });

        dm.Tell(new DeployInstanceCommand(
            "dep-rev-2", instanceName, "rev-2",
            MakeScriptedConfigJson(instanceName, "var x = 2; var y = x * 3; return y;"),
            "admin", DateTimeOffset.UtcNow));
        var response = ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(20));

        await ticking.CancelAsync();
        await ticker;

        Assert.Equal(DeploymentStatus.Success, response.Status);
        Assert.True(await InstanceIsRunningAsync(dm, instanceName));
        Assert.Empty(SelfHealEvents(siteLog, instanceName));
    }

    // ── Row 72: a running actor serving a revision the store has replaced ──

    /// <summary>
    /// THE row-72 reproduction. An instance is deployed and running at <c>rev-1</c>; the durable
    /// row is then refreshed to <c>rev-2</c> out of band — what
    /// <c>StoreDeployedConfigIfNewerAsync</c> does when the startup reconcile finds a STALE
    /// config, and what a replicated deploy does on the standby.
    ///
    /// <para>
    /// <b>Fails before the fix</b>: the actor keeps serving <c>rev-1</c>'s config indefinitely
    /// (the snapshot still shows the old attribute value) and nothing anywhere says the stored
    /// and running revisions differ.
    /// </para>
    /// </summary>
    [Fact]
    public async Task Sweep_RestartsInstance_ServingAStaleRevision()
    {
        const string instanceName = "StalePump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

        await DeployAndAwaitAsync(
            dm, instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"));

        var beforeHeal = await ResolveInstanceChildAsync(dm, instanceName);

        // The reconcile refreshes the stale row. Storage only; the running actor is untouched.
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName, attributeValue: "222"),
            "dep-rev-2", "rev-2", true);

        // The runtime must converge on the stored config...
        await AwaitAssertAsync(async () =>
        {
            var snapshot = await SnapshotAsync(dm, instanceName);
            Assert.False(snapshot.InstanceNotFound);
            Assert.Contains(snapshot.AttributeValues, a =>
                a.AttributeName == "TestAttr" && a.Value?.ToString() == "222");
        }, TimeSpan.FromSeconds(25));

        // ...through a genuine restart, not an in-place mutation.
        Assert.NotEqual(beforeHeal, await ResolveInstanceChildAsync(dm, instanceName));

        // ...and the divergence must be surfaced, naming BOTH revisions.
        Assert.Contains(SelfHealEvents(siteLog, instanceName), e =>
            e.Severity == "Warning" &&
            e.Message.Contains("rev-1", StringComparison.Ordinal) &&
            e.Message.Contains("rev-2", StringComparison.Ordinal));
    }

    /// <summary>
    /// The stale-revision restart must not fire when the stored and running revisions agree —
    /// otherwise every healthy instance on the site would be restarted once a minute.
    /// </summary>
    [Fact]
    public async Task Sweep_DoesNotRestart_WhenStoredAndRunningRevisionsMatch()
    {
        const string instanceName = "MatchingRevisionPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

        await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));
        var beforeSweeps = await ResolveInstanceChildAsync(dm, instanceName);

        // Rewrite the SAME revision — a same-revision re-store must not read as divergence.
        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName), "dep-rev-1", "rev-1", true);

        await Task.Delay(SeveralSweeps);

        Assert.Equal(beforeSweeps, await ResolveInstanceChildAsync(dm, instanceName));
        Assert.Empty(SelfHealEvents(siteLog, instanceName));
    }

    /// <summary>
    /// A stale-revision restart that the site compile gate REJECTS must be announced once and
    /// then left alone. Retrying it every tick would be an endless stream of Warning + Error
    /// site events for a config that can never apply, and the running instance is unharmed —
    /// the gate returns before anything is stopped, so it keeps serving the old config.
    /// </summary>
    [Fact]
    public async Task Sweep_StaleRestartRejectedByTheCompileGate_IsAnnouncedOnceAndNotRetried()
    {
        const string instanceName = "UncompilablePump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog);

        await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));
        var beforeHeal = await ResolveInstanceChildAsync(dm, instanceName);

        // The refreshed row carries a script that cannot compile.
        await _storage.StoreDeployedConfigAsync(
            instanceName,
            MakeScriptedConfigJson(instanceName, "this is not valid C# at all ((("),
            "dep-rev-bad", "rev-bad", true);

        // The attempt is announced...
        await AwaitAssertAsync(() =>
        {
            Assert.Single(SelfHealEvents(siteLog, instanceName));
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

        // ...its failure is announced...
        await AwaitAssertAsync(() =>
        {
            Assert.Contains(siteLog.OfType("deployment"), e =>
                e.InstanceId == instanceName &&
                e.Severity == "Error" &&
                e.Message.Contains("self-heal failed", StringComparison.OrdinalIgnoreCase));
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

        // ...and several further sweeps add nothing and leave the old actor serving.
        await Task.Delay(SeveralSweeps);

        Assert.Single(SelfHealEvents(siteLog, instanceName));
        Assert.Equal(beforeHeal, await ResolveInstanceChildAsync(dm, instanceName));
        Assert.True(await InstanceIsRunningAsync(dm, instanceName));
    }

    // ── Row 101 (config half): a sweep snapshot older than the actor's own state ──

    /// <summary>
    /// A crafted sweep result standing in for a read that was dispatched earlier than the
    /// generation it is stamped with. <paramref name="generationAtRead"/> <c>0</c> is the value a
    /// read dispatched before this actor had mutated anything would carry; anything the actor has
    /// done since makes it stale. Connection definitions are passed as <see langword="null"/>
    /// (= "the definitions read failed, keep the overlay") so these tests exercise the CONFIG
    /// half of the snapshot in isolation.
    /// </summary>
    private static DeploymentManagerActor.InstanceSweepLoaded SweepSnapshot(
        string instanceName, string revision, string configJson, long generationAtRead,
        bool isEnabled = true) =>
        new(
            [
                new DeployedInstance
                {
                    InstanceUniqueName = instanceName,
                    ConfigJson = configJson,
                    DeploymentId = $"dep-{revision}",
                    RevisionHash = revision,
                    IsEnabled = isEnabled,
                    DeployedAt = DateTimeOffset.UtcNow.ToString("O")
                }
            ],
            null,
            ConnectionDefinitions: null,
            StoreGenerationAtRead: generationAtRead);

    /// <summary>
    /// <b>Deferred register row 101, the row-72 leg.</b> A sweep read dispatched before a deploy
    /// persisted its new row can land after that deploy has completed AND after the deploy
    /// pipeline released the instance — <see cref="DeploymentManagerActor"/>'s
    /// <c>InstanceIsOwnedByADeployPipeline</c> only covers a deploy still in flight, so it is no
    /// defence here. The snapshot then carries the revision the deploy just REPLACED while the
    /// running actor carries the new one, which is bit-for-bit the divergence the row-72 repair
    /// looks for — in the wrong direction.
    ///
    /// <para>
    /// <b>Fails before the fix</b>: the sweep "repairs" the instance BACKWARD, re-deploying the
    /// snapshot's own superseded config JSON through the redeploy path (attribute back to
    /// <c>111</c>) and announcing it as a self-heal. Nothing retries it either — the per-revision
    /// claim marks the OLD revision as attempted, so the instance simply stays reverted.
    /// </para>
    /// </summary>
    [Fact]
    public async Task StaleSweepSnapshot_CarryingASupersededRevision_DoesNotRestartBackward()
    {
        const string instanceName = "BackwardRestartPump";
        var siteLog = new FakeSiteEventLogger();
        // Sweep effectively disabled: the crafted message must be the only sweep result in play.
        var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);

        await DeployAndAwaitAsync(
            dm, instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"));
        await DeployAndAwaitAsync(
            dm, instanceName, "rev-2", MakeConfigJson(instanceName, attributeValue: "222"));

        var afterDeploys = await ResolveInstanceChildAsync(dm, instanceName);

        // Generation 0 = captured before either deploy mutated the store.
        dm.Tell(SweepSnapshot(
            instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"),
            generationAtRead: 0));

        await Task.Delay(SeveralSweeps);

        Assert.Empty(SelfHealEvents(siteLog, instanceName));
        Assert.Equal(afterDeploys, await ResolveInstanceChildAsync(dm, instanceName));
        var snapshot = await SnapshotAsync(dm, instanceName);
        Assert.Contains(snapshot.AttributeValues, a =>
            a.AttributeName == "TestAttr" && a.Value?.ToString() == "222");
    }

    /// <summary>
    /// The row-101 guard must not blunt the row-72 repair it sits in front of: the SAME message
    /// stamped with the CURRENT generation is a genuine answer and must still converge the
    /// runtime on the stored revision.
    ///
    /// <para>
    /// The instance is started by the BOOT LOAD from a pre-written row rather than by a deploy
    /// command, precisely so no store mutation passes through the actor and the current
    /// generation is still <c>0</c> — the same literal the stale test uses, with the opposite
    /// meaning, which is the whole point of the guard.
    /// </para>
    /// </summary>
    [Fact]
    public async Task FreshSweepSnapshot_CarryingANewerRevision_StillRestartsTheInstance()
    {
        const string instanceName = "FreshRestartPump";
        var siteLog = new FakeSiteEventLogger();

        await _storage.StoreDeployedConfigAsync(
            instanceName, MakeConfigJson(instanceName, attributeValue: "111"),
            "dep-rev-1", "rev-1", true);

        var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);
        await AwaitAssertAsync(async () =>
            Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));

        dm.Tell(SweepSnapshot(
            instanceName, "rev-2", MakeConfigJson(instanceName, attributeValue: "222"),
            generationAtRead: 0));

        await AwaitAssertAsync(async () =>
        {
            var snapshot = await SnapshotAsync(dm, instanceName);
            Assert.False(snapshot.InstanceNotFound);
            Assert.Contains(snapshot.AttributeValues, a =>
                a.AttributeName == "TestAttr" && a.Value?.ToString() == "222");
        }, TimeSpan.FromSeconds(25));

        Assert.Contains(SelfHealEvents(siteLog, instanceName), e =>
            e.Severity == "Warning" &&
            e.Message.Contains("rev-1", StringComparison.Ordinal) &&
            e.Message.Contains("rev-2", StringComparison.Ordinal));
    }

    /// <summary>
    /// <b>Deferred register row 101, the row-78 leg.</b> A sweep read dispatched before a delete
    /// committed still shows the instance's row — enabled, with a config — while the actor has
    /// already stopped it and dropped its bookkeeping. The delete specifically clears
    /// <c>_sweepAttemptedRevisions</c>, so the once-per-revision claim is freshly available and
    /// does not stand in the way either.
    ///
    /// <para>
    /// <b>Fails before the fix</b>: the sweep RESURRECTS the instance the operator just deleted,
    /// re-adds it to the deployed-name set, and announces the resurrection as a self-heal.
    /// </para>
    /// </summary>
    [Fact]
    public async Task StaleSweepSnapshot_ShowingADeletedInstance_DoesNotResurrectIt()
    {
        const string instanceName = "ResurrectedPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);

        await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));

        dm.Tell(new DeleteInstanceCommand("cmd-del-1", instanceName, DateTimeOffset.UtcNow));
        Assert.True(ExpectMsg<InstanceLifecycleResponse>(TimeSpan.FromSeconds(15)).Success);
        await AwaitAssertAsync(async () =>
            Assert.False(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(15));

        // Generation 0 = captured before the deploy and the delete mutated the store.
        dm.Tell(SweepSnapshot(
            instanceName, "rev-1", MakeConfigJson(instanceName), generationAtRead: 0));

        await Task.Delay(SeveralSweeps);

        Assert.False(await InstanceIsRunningAsync(dm, instanceName));
        Assert.Empty(SelfHealEvents(siteLog, instanceName));
    }

    /// <summary>
    /// The mirror of the resurrection test: the same enabled-config-with-no-actor snapshot,
    /// stamped with the CURRENT generation, must still perform the row-78 start. Nothing has
    /// mutated the store through this actor, so <c>0</c> is current here.
    /// </summary>
    [Fact]
    public async Task FreshSweepSnapshot_ShowingAStoredEnabledInstance_StillStartsIt()
    {
        const string instanceName = "FreshStartPump";
        var siteLog = new FakeSiteEventLogger();
        var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);

        dm.Tell(SweepSnapshot(
            instanceName, "rev-1", MakeConfigJson(instanceName), generationAtRead: 0));

        await AwaitAssertAsync(async () =>
            Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));
        Assert.Single(SelfHealEvents(siteLog, instanceName));
    }

    /// <summary>
    /// <b>Row 101, boot half.</b> The boot-time load has to COMPLETE — a node whose startup read
    /// is simply dropped stays a silent zero-instance node — so a superseded boot snapshot is
    /// re-read rather than discarded like the sweep's.
    ///
    /// <para>
    /// The race is made deterministic by a slow first read: a lifecycle command that mutates
    /// <c>deployed_configurations</c> is processed while it is in flight, so the snapshot lands
    /// stamped with a generation the actor has already moved past. The observable is the read
    /// itself — the loader must be called a SECOND time — and, just as importantly, exactly twice:
    /// the re-dispatch captures the post-mutation generation, so it converges instead of looping.
    /// </para>
    ///
    /// <para><b>Fails before the fix</b>: one read, and the pre-mutation snapshot is applied.</para>
    /// </summary>
    [Fact]
    public async Task StaleBootSnapshot_IsReReadRatherThanApplied()
    {
        var reads = 0;
        var health = new DeploymentManagerStartupProbe();

        var configLoader = SlowFirstConfigLoad(() => Interlocked.Increment(ref reads));
        var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            new SiteRuntimeOptions { InstanceSweepIntervalSeconds = 3600 },
            NullLogger<DeploymentManagerActor>.Instance,
            dclManager: null,
            healthCollector: health,
            serviceProvider: null,
            loggerFactory: null,
            configFetcher: null,
            startupLoadRetryInterval: null,
            configLoader: configLoader,
            externalSystemCache: null,
            scriptScheduler: null,
            connectionDefinitionLoader: null)));

        // Mutates deployed_configurations (a no-op UPDATE for an unknown instance is still a
        // write dispatched from the actor thread) while the first read is still in flight.
        dm.Tell(new DisableInstanceCommand("cmd-boot-race", "NoSuchInstance", DateTimeOffset.UtcNow));
        ExpectMsg<InstanceLifecycleResponse>(TimeSpan.FromSeconds(15));

        await AwaitAssertAsync(() =>
        {
            Assert.Equal(2, Volatile.Read(ref reads));
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(20));

        // The second read is fresh, so the boot completes and stops re-reading...
        await Task.Delay(TimeSpan.FromSeconds(2));
        Assert.Equal(2, Volatile.Read(ref reads));

        // ...and a LATE superseded snapshot (the S5 retry timer racing a success) is dropped
        // rather than triggering another read, because the boot it would complete is done.
        dm.Tell(new DeploymentManagerActor.StartupConfigsLoaded(
            [], null, ConnectionDefinitions: null, StoreGenerationAtRead: -1));
        await Task.Delay(TimeSpan.FromSeconds(2));
        Assert.Equal(2, Volatile.Read(ref reads));
    }

    /// <summary>
    /// A deployed-config loader whose FIRST call takes long enough for a lifecycle command to be
    /// processed while it is in flight; later calls answer immediately. Always returns an empty
    /// site — these tests assert on how many reads happen, not on what they carry.
    /// </summary>
    private static Func<Task<List<DeployedInstance>>> SlowFirstConfigLoad(Action onRead)
    {
        var calls = 0;
        return async () =>
        {
            onRead();
            if (Interlocked.Increment(ref calls) == 1)
                await Task.Delay(TimeSpan.FromSeconds(2));
            return [];
        };
    }
}