using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure;
using ZB.MOM.WW.ScadaBridge.Host;
using ZB.MOM.WW.ScadaBridge.Host.Actors;

namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster;

/// <summary>
/// Guards the self-first seed-ordering invariant (decision 2026-07-22): every node lists
/// ITSELF as <c>seed-nodes[0]</c> and its partner second.
///
/// <para><b>Why the ordering is the mechanism.</b> Akka runs a different bootstrap process
/// depending on whether <c>seed-nodes[0]</c> is this node's own address. When it is,
/// <c>FirstSeedNodeProcess</c> runs: it InitJoins the OTHER seeds and self-joins only after
/// <c>seed-node-timeout</c> passes with nobody answering. When it is not,
/// <c>JoinSeedNodeProcess</c> runs, which can never form a new cluster — it retries InitJoin
/// forever. That is the "registered outage gap" (docker/README.md): a lone cold-starting
/// non-first seed never came Up. Listing self first closes it using Akka's own protocol.</para>
///
/// <para><b>Why not an external self-form watchdog.</b> A timer that waits N seconds for
/// membership and then calls <c>Cluster.Join(SelfAddress)</c> cannot see Akka's join
/// handshake, so it cannot distinguish "no seed answered" from "a seed answered and the join
/// is in flight". <see cref="Restarting_node_rejoins_its_live_peer_instead_of_self_forming"/>
/// is the case that killed that design: on a routine standby restart the peer is alive but
/// the join is stalled behind removal of this node's own stale incarnation, and a Join(self)
/// issued during <c>TryingToJoin</c> abandons the in-flight join and forms a SECOND cluster at
/// the same address — a permanent split (measured: still split after 90s). Akka's own
/// first-seed process has no such race because it is part of the handshake.</para>
///
/// These build REAL clusters from the production <c>BuildHocon</c> output, like
/// <see cref="TwoNodeClusterFixture"/>, at production failure-detection timings.
/// </summary>
public sealed class SelfFirstSeedBootstrapTests : IAsyncLifetime
{
    private readonly List<ActorSystem> _systems = new();

    /// <summary>Starts a node configured the way every shipped node appsettings is: its own
    /// address first, the partner second. <paramref name="selfFirst"/> = false reproduces the
    /// pre-fix ordering, which is what makes the outage-gap assertions falsifiable.</summary>
    private ActorSystem StartNode(int selfPort, int peerPort, bool selfFirst = true)
    {
        var nodeOptions = new NodeOptions { Role = "Central", NodeHostname = "127.0.0.1", RemotingPort = selfPort };
        var self = $"akka.tcp://scadabridge@127.0.0.1:{selfPort}";
        var peer = $"akka.tcp://scadabridge@127.0.0.1:{peerPort}";
        var clusterOptions = new ClusterOptions
        {
            SeedNodes = selfFirst ? new List<string> { self, peer } : new List<string> { peer, self },
            // Production failure-detection envelope: this is what sets how long a restarting
            // node's join stays stalled behind its own stale incarnation.
            StableAfter = TimeSpan.FromSeconds(15),
            HeartbeatInterval = TimeSpan.FromSeconds(2),
            FailureDetectionThreshold = TimeSpan.FromSeconds(10),
            MinNrOfMembers = 1,
        };
        var hocon = AkkaHostedService.BuildHocon(
            nodeOptions, clusterOptions, new[] { "Central" },
            TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3));
        // Crash simulation: Terminate() must NOT run CoordinatedShutdown (no Leave gossip),
        // matching TwoNodeClusterFixture — otherwise a "restart after crash" degenerates to
        // the graceful path where the peer has already removed the old member.
        var config = ConfigurationFactory
            .ParseString("akka.coordinated-shutdown.run-by-actor-system-terminate = off")
            .WithFallback(ConfigurationFactory.ParseString(hocon));
        var system = ActorSystem.Create("scadabridge", config);
        _systems.Add(system);
        return system;
    }

    [Fact]
    public async Task Lone_cold_start_forms_a_cluster_when_the_peer_is_dead()
    {
        var selfPort = TwoNodeClusterFixture.GetFreeTcpPort();
        var deadPeerPort = TwoNodeClusterFixture.GetFreeTcpPort(); // nothing listening
        var node = StartNode(selfPort, deadPeerPort);

        // Akka's FirstSeedNodeProcess self-joins after seed-node-timeout (~5s) once the dead
        // peer fails to answer InitJoin. This is the unattended cold-start-alone guarantee.
        await TwoNodeClusterFixture.WaitForMembersUp(node, 1, TimeSpan.FromSeconds(30));
        Assert.Equal(MemberStatus.Up, Akka.Cluster.Cluster.Get(node).SelfMember.Status);
    }

    [Fact]
    public async Task Peer_first_ordering_is_the_outage_gap_and_never_forms()
    {
        // FALSIFIABILITY CONTROL for the test above: with the OLD ordering (peer first) the
        // identical scenario never comes Up. If this test ever starts passing quickly, the
        // self-first ordering has stopped being the thing doing the work.
        var selfPort = TwoNodeClusterFixture.GetFreeTcpPort();
        var deadPeerPort = TwoNodeClusterFixture.GetFreeTcpPort();
        var node = StartNode(selfPort, deadPeerPort, selfFirst: false);

        await Task.Delay(TimeSpan.FromSeconds(15)); // 3x the seed-node-timeout
        var cluster = Akka.Cluster.Cluster.Get(node);
        Assert.Empty(cluster.State.Members); // still InitJoin-looping — the registered outage gap

        // Positive control: the node was formable all along; only the ordering blocked it.
        cluster.Join(cluster.SelfAddress);
        await TwoNodeClusterFixture.WaitForMembersUp(node, 1, TimeSpan.FromSeconds(30));
    }

    [Fact]
    public async Task Restarting_node_rejoins_its_live_peer_instead_of_self_forming()
    {
        // The case that rejected the self-form-watchdog design (code review 2026-07-22, C1):
        // a routine standby restart while the peer is alive must rejoin, never island.
        var portA = TwoNodeClusterFixture.GetFreeTcpPort();
        var portB = TwoNodeClusterFixture.GetFreeTcpPort();

        var nodeA = StartNode(portA, portB);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
        var nodeB = StartNode(portB, portA);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(30));

        // Hard-crash B and immediately restart it at the SAME address, as a container
        // recreate / service restart does.
        await nodeB.Terminate().WaitAsync(TimeSpan.FromSeconds(10));
        var nodeB2 = StartNode(portB, portA);

        // Both must converge on ONE 2-member cluster — never two 1-member clusters.
        await TwoNodeClusterFixture.WaitForMembersUp(nodeB2, 2, TimeSpan.FromSeconds(90));
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(90));
    }

    [Fact]
    public async Task Both_nodes_cold_starting_together_converge_on_one_cluster()
    {
        // Self-first on BOTH nodes raises the obvious question: does a simultaneous cold
        // start produce two clusters? While they are mutually reachable it does not — the
        // InitJoin handshake resolves it before either self-joins. (A genuine boot-time
        // PARTITION would still split, which is the same class auto-down already accepts.)
        var portA = TwoNodeClusterFixture.GetFreeTcpPort();
        var portB = TwoNodeClusterFixture.GetFreeTcpPort();

        var nodeA = StartNode(portA, portB);
        var nodeB = StartNode(portB, portA);

        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(60));
        await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(60));
    }

    public Task InitializeAsync() => Task.CompletedTask;

    public async Task DisposeAsync()
    {
        foreach (var s in _systems)
        {
            try { await s.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); } catch { /* teardown */ }
        }
    }
}