using System.Net;
using System.Net.Sockets;
using Akka.Actor;
using Akka.Cluster;
using Akka.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using Shouldly;
using Xunit;

namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests;

/// <summary>
/// Guards the self-first seed-ordering invariant (decision 2026-07-22): every node that is a seed
/// of its own mesh lists ITSELF as <c>SeedNodes[0]</c> and its partner second.
/// </summary>
/// <remarks>
///     <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 <c>InitJoin</c>s the other seeds and self-joins once
///         <c>seed-node-timeout</c> (5 s) passes with nobody answering. When it is not,
///         <c>JoinSeedNodeProcess</c> runs, and that process can never form a new cluster — it
///         retries <c>InitJoin</c> forever. That is why "both peers are listed in SeedNodes" never
///         meant "either can cold-start alone", and why a node listed second stayed out of the
///         cluster indefinitely while its peer was down.
///     </para>
///     <para>
///         <b>Why not the self-form watchdog this replaces.</b> <c>ClusterBootstrapFallback</c>
///         waited <c>Cluster:SelfFormAfter</c> for membership and then called
///         <c>Cluster.Join(SelfAddress)</c>. A timer outside Akka's join handshake cannot
///         distinguish "no seed answered" from "a seed answered and the join is in flight", and
///         <c>Join(SelfAddress)</c> is <b>not</b> ignored mid-handshake — it wins. The live gate on
///         the docker-dev rig (2026-07-22) caught exactly that: a node bounced by a manual failover
///         restarted, got <c>InitJoinAck</c> from its live peer, but no Welcome inside the window
///         because the peer's ring still held its previous incarnation
///         (<c>Exiting</c> → <c>Down</c> → <c>Removed</c>); the watchdog fired and formed a SECOND
///         cluster. A TCP reachability guard patched that one shape; the race stayed.
///         <see cref="Restarting_node_rejoins_its_live_peer_instead_of_islanding"/> is that scenario,
///         and self-first ordering has no such race because the decision is made inside the
///         handshake by Akka itself.
///     </para>
///     <para>
///         <b>Why these start real hosts through the production bootstrap.</b> Same reason as
///         <see cref="SplitBrainResolverActivationTests"/>: the question is what a running node does.
///         Every node here is built with
///         <see cref="ServiceCollectionExtensions.WithOtOpcUaClusterBootstrap"/> — the shipped entry
///         point — so these run at the production failure-detection envelope (heartbeat 2 s,
///         acceptable pause 10 s, auto-down 15 s, down-removal-margin 15 s) from
///         <c>Resources/akka.conf</c> rather than at timings invented by the test.
///     </para>
/// </remarks>
public sealed class SelfFirstSeedBootstrapTests
{
    private static int FreePort()
    {
        var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return port;
    }

    /// <summary>
    /// Starts a node the way every shipped node config is written: its own address first, the
    /// partner second.
    /// </summary>
    /// <param name="systemName">Actor-system name; also the system part of every seed address.</param>
    /// <param name="selfPort">This node's remoting port.</param>
    /// <param name="peerPort">The partner's remoting port (may have nothing listening on it).</param>
    /// <param name="selfFirst">
    /// <see langword="false"/> reproduces the pre-2026-07-22 ordering (partner first), which is what
    /// makes <see cref="Peer_first_ordering_is_the_outage_gap_and_never_forms"/> a falsifiability
    /// control rather than a restatement of the fix.
    /// </param>
    private static async Task<IHost> StartNodeAsync(
        string systemName,
        int selfPort,
        int peerPort,
        bool selfFirst = true)
    {
        var self = $"akka.tcp://{systemName}@127.0.0.1:{selfPort}";
        var peer = $"akka.tcp://{systemName}@127.0.0.1:{peerPort}";
        var options = new AkkaClusterOptions
        {
            SystemName = systemName,
            Hostname = "127.0.0.1",
            PublicHostname = "127.0.0.1",
            Port = selfPort,
            Roles = new[] { "admin", "driver" },
            SeedNodes = selfFirst ? new[] { self, peer } : new[] { peer, self },
        };

        var builder = Host.CreateDefaultBuilder();
        builder.ConfigureServices(services =>
        {
            services.AddSingleton<IOptions<AkkaClusterOptions>>(Options.Create(options));
            services.AddAkka(systemName, (ab, sp) =>
            {
                ab.WithOtOpcUaClusterBootstrap(sp);

                // Crash simulation, test-only: with this off, ActorSystem.Terminate() skips
                // CoordinatedShutdown and so gossips no Leave — a process death, not a graceful
                // drain. Without it "restart the standby" would degenerate into the easy path where
                // the peer had already removed the old member before the new one dialled in.
                // Prepend is Akka.Hosting's highest-precedence merge mode (see BuildDowningHocon).
                ab.AddHocon(
                    "akka.coordinated-shutdown.run-by-actor-system-terminate = off",
                    HoconAddMode.Prepend);
            });
        });

        var host = builder.Build();
        await host.StartAsync();
        return host;
    }

    /// <summary>Polls until the node sees <paramref name="expected"/> Up members, or the deadline passes.</summary>
    private static async Task<bool> WaitForUpMembersAsync(IHost host, int expected, TimeSpan timeout)
    {
        var cluster = Akka.Cluster.Cluster.Get(host.Services.GetRequiredService<ActorSystem>());
        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;
    }

    private static async Task StopAsync(IHost host)
    {
        try
        {
            await host.StopAsync();
        }
        catch (Exception)
        {
            // Teardown only. A host whose ActorSystem was terminated by a crash-simulation cannot
            // run CoordinatedShutdown again; that must not fail the test that already asserted.
        }

        host.Dispose();
    }

    /// <summary>
    /// The unattended cold-start-alone guarantee: a node whose partner is dead forms the cluster by
    /// itself, through Akka's own <c>FirstSeedNodeProcess</c> — no watchdog, no timer, no operator.
    /// </summary>
    [Fact]
    public async Task Lone_cold_start_forms_a_cluster_when_the_peer_is_dead()
    {
        var selfPort = FreePort();
        var deadPeerPort = FreePort(); // nothing is listening there
        var host = await StartNodeAsync("otopcua-selffirst-1", selfPort, deadPeerPort);
        try
        {
            // seed-node-timeout is Akka's 5 s default; 30 s is slack for a loaded CI box.
            (await WaitForUpMembersAsync(host, 1, TimeSpan.FromSeconds(30)))
                .ShouldBeTrue("a self-first seed must form a cluster alone when no peer answers InitJoin");
        }
        finally
        {
            await StopAsync(host);
        }
    }

    /// <summary>
    /// FALSIFIABILITY CONTROL for the test above. With the OLD ordering — partner first — the
    /// identical scenario never comes Up, because <c>JoinSeedNodeProcess</c> has no self-join path.
    /// If this test ever starts passing quickly, the seed ordering has stopped being the thing doing
    /// the work and the suite above has gone vacuous.
    /// </summary>
    /// <remarks>
    /// This is also the test that fails while the retired <c>ClusterBootstrapFallback</c> watchdog
    /// is still armed: that timer would bring this node Up at <c>SelfFormAfter</c> (10 s) despite the
    /// ordering, which is precisely the "something other than Akka's handshake is deciding" that the
    /// retirement removes.
    /// </remarks>
    [Fact]
    public async Task Peer_first_ordering_is_the_outage_gap_and_never_forms()
    {
        var selfPort = FreePort();
        var deadPeerPort = FreePort();
        var host = await StartNodeAsync("otopcua-selffirst-2", selfPort, deadPeerPort, selfFirst: false);
        try
        {
            // 3x the 5 s seed-node-timeout, and longer than the retired watchdog's 10 s window.
            (await WaitForUpMembersAsync(host, 1, TimeSpan.FromSeconds(15)))
                .ShouldBeFalse("a node listed behind its peer cannot form a cluster — it InitJoin-loops forever");

            var cluster = Akka.Cluster.Cluster.Get(host.Services.GetRequiredService<ActorSystem>());
            cluster.State.Members.ShouldBeEmpty();

            // Positive control: the node was formable all along; only the ordering blocked it.
            cluster.Join(cluster.SelfAddress);
            (await WaitForUpMembersAsync(host, 1, TimeSpan.FromSeconds(30)))
                .ShouldBeTrue("positive control — an explicit self-join must bring this node Up");
        }
        finally
        {
            await StopAsync(host);
        }
    }

    /// <summary>
    /// The case that killed the self-form watchdog (live gate, docker-dev, 2026-07-22): a routine
    /// standby restart while the peer is alive must REJOIN, never island. Under self-first ordering
    /// the restarted node's <c>FirstSeedNodeProcess</c> only self-joins when no seed answers, so the
    /// live peer's <c>InitJoinAck</c> keeps it on the join path however long the peer takes to
    /// retire its previous incarnation.
    /// </summary>
    [Fact]
    public async Task Restarting_node_rejoins_its_live_peer_instead_of_islanding()
    {
        const string systemName = "otopcua-selffirst-3";
        var portA = FreePort();
        var portB = FreePort();

        var nodeA = await StartNodeAsync(systemName, portA, portB);
        IHost? nodeB = null;
        IHost? nodeB2 = null;
        try
        {
            (await WaitForUpMembersAsync(nodeA, 1, TimeSpan.FromSeconds(30))).ShouldBeTrue();

            nodeB = await StartNodeAsync(systemName, portB, portA);
            (await WaitForUpMembersAsync(nodeA, 2, TimeSpan.FromSeconds(60)))
                .ShouldBeTrue("the pair must converge before the restart is meaningful");

            // Hard-crash B and immediately restart it at the SAME address, as a service restart or
            // container recreate does. No Leave gossip: A still holds B's previous incarnation.
            await nodeB.Services.GetRequiredService<ActorSystem>().Terminate()
                .WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);
            nodeB2 = await StartNodeAsync(systemName, portB, portA);

            // ONE cluster of two, never two clusters of one. The budget covers auto-down (15 s) plus
            // down-removal-margin (15 s) before the new incarnation is admitted.
            (await WaitForUpMembersAsync(nodeB2, 2, TimeSpan.FromSeconds(120)))
                .ShouldBeTrue("the restarted node must rejoin its live peer, not form a second cluster");
            (await WaitForUpMembersAsync(nodeA, 2, TimeSpan.FromSeconds(120)))
                .ShouldBeTrue("the surviving node must see the restarted peer as a member of ITS cluster");
        }
        finally
        {
            if (nodeB2 is not null) await StopAsync(nodeB2);
            if (nodeB is not null) await StopAsync(nodeB);
            await StopAsync(nodeA);
        }
    }

    /// <summary>
    /// The obvious objection to self-first on BOTH nodes: does a simultaneous cold start produce two
    /// clusters? While the two are mutually reachable it does not — each runs
    /// <c>FirstSeedNodeProcess</c>, and the <c>InitJoin</c> handshake resolves which one forms before
    /// either self-join deadline expires. (A genuine boot-time PARTITION would still split, the same
    /// dual-active class the <c>auto-down</c> strategy already accepts, with the same recovery.)
    /// </summary>
    [Fact]
    public async Task Both_nodes_cold_starting_together_converge_on_one_cluster()
    {
        const string systemName = "otopcua-selffirst-4";
        var portA = FreePort();
        var portB = FreePort();

        var nodeA = await StartNodeAsync(systemName, portA, portB);
        var nodeB = await StartNodeAsync(systemName, portB, portA);
        try
        {
            (await WaitForUpMembersAsync(nodeA, 2, TimeSpan.FromSeconds(60)))
                .ShouldBeTrue("both nodes cold-starting self-first must converge, not split");
            (await WaitForUpMembersAsync(nodeB, 2, TimeSpan.FromSeconds(60)))
                .ShouldBeTrue("both nodes cold-starting self-first must converge, not split");
        }
        finally
        {
            await StopAsync(nodeB);
            await StopAsync(nodeA);
        }
    }
}