using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
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>
/// Real-ActorSystem tests for the simultaneous-cold-start split-brain guard (Gitea #33):
/// <see cref="ClusterBootstrapCoordinator"/> + <see cref="ClusterBootstrapGuard"/>, driven exactly as
/// production drives them — the node's ActorSystem is built from the production <c>BuildHocon</c> output
/// with the guard enabled (which emits an EMPTY seed list, so Akka does not auto-join), then the real
/// coordinator issues the reachability-gated <c>JoinSeedNodes</c>. This subsystem has a history of bugs
/// invisible to pure unit tests, so the load-bearing correctness properties are asserted against running
/// nodes, mirroring <see cref="SelfFirstSeedBootstrapTests"/>.
/// </summary>
/// <remarks>
/// Ephemeral loopback ports share host <c>127.0.0.1</c>, so the port breaks the guard's ordinal
/// <c>host:port</c> tie — <c>Min(port)</c> is the founder, <c>Max(port)</c> is the higher (probing) node.
/// </remarks>
public sealed class ClusterBootstrapCoordinatorTests : IAsyncLifetime
{
    private readonly List<ActorSystem> _systems = new();
    private readonly List<ClusterBootstrapCoordinator> _coordinators = new();

    /// <summary>Starts a node with the bootstrap guard ENABLED, wired exactly as the composition roots
    /// do: the ActorSystem is built with an empty HOCON seed list (guard on) and the real coordinator
    /// issues the reachability-gated join. Seeds are listed self-first (as every shipped config is); the
    /// guard, not the order, decides founder vs. joiner.</summary>
    private async Task<ActorSystem> StartGuardedNodeAsync(int selfPort, int peerPort, int probeSeconds = 4)
    {
        var self = $"akka.tcp://scadabridge@127.0.0.1:{selfPort}";
        var peer = $"akka.tcp://scadabridge@127.0.0.1:{peerPort}";
        var nodeOptions = new NodeOptions { Role = "Central", NodeHostname = "127.0.0.1", RemotingPort = selfPort };
        var clusterOptions = new ClusterOptions
        {
            SeedNodes = new List<string> { self, peer },
            BootstrapGuard = new ClusterBootstrapGuardOptions
            {
                Enabled = true,
                PartnerProbeSeconds = probeSeconds,
                PartnerProbeIntervalMs = 250,
                ProbeConnectTimeoutMs = 500,
            },
            StableAfter = TimeSpan.FromSeconds(3),
            HeartbeatInterval = TimeSpan.FromMilliseconds(500),
            FailureDetectionThreshold = TimeSpan.FromSeconds(2),
            MinNrOfMembers = 1,
        };

        var hocon = AkkaHostedService.BuildHocon(
            nodeOptions, clusterOptions, new[] { "Central" },
            TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3));
        var system = ActorSystem.Create("scadabridge", ConfigurationFactory.ParseString(hocon));
        _systems.Add(system);

        var coordinator = new ClusterBootstrapCoordinator(
            () => system, Options.Create(clusterOptions), Options.Create(nodeOptions),
            NullLogger<ClusterBootstrapCoordinator>.Instance);
        _coordinators.Add(coordinator);
        await coordinator.StartAsync(CancellationToken.None);
        return system;
    }

    private static async Task<bool> WaitForUpAsync(ActorSystem system, int expected, TimeSpan timeout)
    {
        var cluster = Akka.Cluster.Cluster.Get(system);
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (cluster.State.Members.Count(m => m.Status == MemberStatus.Up) >= expected) return true;
            await Task.Delay(200);
        }
        return cluster.State.Members.Count(m => m.Status == MemberStatus.Up) >= expected;
    }

    /// <summary>
    /// The FOUNDER (lower address) with the guard on forms a cluster alone when its partner is dead —
    /// exactly as un-guarded self-first does. The guard must not regress the preferred founder.
    /// </summary>
    [Fact]
    public async Task Founder_forms_alone_when_partner_is_dead()
    {
        var low = TwoNodeClusterFixture.GetFreeTcpPort();
        var high = TwoNodeClusterFixture.GetFreeTcpPort();
        var founderPort = Math.Min(low, high);
        var deadPeerPort = Math.Max(low, high); // nothing listening

        var system = await StartGuardedNodeAsync(founderPort, deadPeerPort);

        Assert.True(await WaitForUpAsync(system, 1, TimeSpan.FromSeconds(30)),
            "the founder (lower address) must self-form immediately when its partner is down");
    }

    /// <summary>
    /// THE LOAD-BEARING CASE. The HIGHER node with the guard on must still cold-start ALONE when its
    /// partner is genuinely dead: it probes the (dead) founder, times out, and falls back to self-first.
    /// If the guard's peer-first logic were wrong, this node would hang in JoinSeedNodeProcess forever —
    /// the very regression a naive "always let the lower node found" rule would introduce.
    /// </summary>
    [Fact]
    public async Task Higher_node_forms_alone_when_partner_is_dead_after_probing()
    {
        var low = TwoNodeClusterFixture.GetFreeTcpPort();
        var high = TwoNodeClusterFixture.GetFreeTcpPort();
        var higherPort = Math.Max(low, high);
        var deadFounderPort = Math.Min(low, high); // the founder is dead

        var system = await StartGuardedNodeAsync(higherPort, deadFounderPort, probeSeconds: 3);

        // Must come Up AFTER the probe window (~3s) expires and it falls back to self-first.
        Assert.True(await WaitForUpAsync(system, 1, TimeSpan.FromSeconds(30)),
            "the higher node must fall back to self-first and form alone once the dead partner never answers the probe");
    }

    /// <summary>
    /// FALSIFIABILITY CONTROL for the test above: the higher node does NOT form alone during the probe
    /// window — it is genuinely waiting/probing, not self-forming immediately (which would mean the
    /// guard is inert). If this ever starts seeing a member before the window, the probe is not gating.
    /// </summary>
    [Fact]
    public async Task Higher_node_does_not_form_before_the_probe_window_expires()
    {
        var low = TwoNodeClusterFixture.GetFreeTcpPort();
        var high = TwoNodeClusterFixture.GetFreeTcpPort();
        var higherPort = Math.Max(low, high);
        var deadFounderPort = Math.Min(low, high);

        var system = await StartGuardedNodeAsync(higherPort, deadFounderPort, probeSeconds: 10);

        // Well within the 10s probe window: still no cluster, because it is probing the dead founder.
        Assert.False(await WaitForUpAsync(system, 1, TimeSpan.FromSeconds(3)),
            "the higher node must probe for the founder, not self-form immediately");
    }

    /// <summary>
    /// The higher node JOINS a live founder rather than forming a second cluster: start the founder,
    /// then the higher node — its probe finds the founder reachable, it commits peer-first, and the
    /// pair converges to ONE cluster of two.
    /// </summary>
    [Fact]
    public async Task Higher_node_joins_a_live_founder()
    {
        var low = TwoNodeClusterFixture.GetFreeTcpPort();
        var high = TwoNodeClusterFixture.GetFreeTcpPort();
        var founderPort = Math.Min(low, high);
        var higherPort = Math.Max(low, high);

        var founder = await StartGuardedNodeAsync(founderPort, higherPort);
        Assert.True(await WaitForUpAsync(founder, 1, TimeSpan.FromSeconds(30)),
            "the founder must be Up before the higher node probes it");

        var higher = await StartGuardedNodeAsync(higherPort, founderPort);

        Assert.True(await WaitForUpAsync(higher, 2, TimeSpan.FromSeconds(60)),
            "the higher node must join the founder, forming one cluster of two");
        Assert.True(await WaitForUpAsync(founder, 2, TimeSpan.FromSeconds(60)),
            "the founder must see the higher node as a member of ITS cluster");
    }

    /// <summary>
    /// THE HEADLINE #33 CASE. Both nodes cold-start truly simultaneously with the guard on. Without the
    /// guard each would run FirstSeedNodeProcess, race, and form its own 1-node cluster (split brain).
    /// With it, the founder forms and the higher node probes-then-joins, so the pair converges to
    /// EXACTLY ONE cluster of two — no two oldest, no dual singletons.
    /// </summary>
    [Fact]
    public async Task Both_nodes_cold_starting_together_form_exactly_one_cluster()
    {
        var low = TwoNodeClusterFixture.GetFreeTcpPort();
        var high = TwoNodeClusterFixture.GetFreeTcpPort();
        var founderPort = Math.Min(low, high);
        var higherPort = Math.Max(low, high);

        // Start both without serialization — as a shared power event does.
        var founderTask = StartGuardedNodeAsync(founderPort, higherPort);
        var higherTask = StartGuardedNodeAsync(higherPort, founderPort);
        var founder = await founderTask;
        var higher = await higherTask;

        Assert.True(await WaitForUpAsync(founder, 2, TimeSpan.FromSeconds(60)),
            "the pair must converge to one 2-member cluster, not two 1-member clusters");
        Assert.True(await WaitForUpAsync(higher, 2, TimeSpan.FromSeconds(60)),
            "the higher node must be in the SAME cluster as the founder");
    }

    public Task InitializeAsync() => Task.CompletedTask;

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