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 InitJoin self-form fallback — whether a node whose peer is down at boot ever becomes
/// operational.
/// </summary>
/// <remarks>
///     <para>
///         <b>Why these tests start real hosts through the production bootstrap.</b> Like
///         <see cref="SplitBrainResolverActivationTests"/>, the question is what a running node
///         actually does, not what an options object holds. Every host here is built with
///         <see cref="ServiceCollectionExtensions.WithOtOpcUaClusterBootstrap"/> — the shipped entry
///         point — so a change that stops arming the fallback (or arms it from a place production
///         does not reach) turns these red. A test that called
///         <see cref="ClusterBootstrapFallback.Arm"/> itself would pin the test's wiring instead.
///     </para>
///     <para>
///         <b>Seed ordering is load-bearing.</b> The node under test lists itself SECOND in its own
///         seed list, behind a dead peer port. Akka only lets the FIRST listed seed self-join, so
///         nothing but the fallback can bring these nodes Up — which is what makes the negative
///         scenarios (disabled window, non-seed node) meaningful rather than vacuous. Each of those
///         carries a positive control (an explicit <c>Cluster.Join(SelfAddress)</c>) proving the node
///         was formable all along and only the fallback was absent.
///     </para>
/// </remarks>
public sealed class SelfFormBootstrapTests
{
    private static int FreePort()
    {
        var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return port;
    }

    private static async Task<IHost> StartNodeAsync(
        string systemName,
        int selfPort,
        int peerPort,
        TimeSpan? selfFormAfter,
        bool selfInSeeds = true)
    {
        var options = new AkkaClusterOptions
        {
            SystemName = systemName,
            Hostname = "127.0.0.1",
            PublicHostname = "127.0.0.1",
            Port = selfPort,
            Roles = new[] { "admin", "driver" },
            SelfFormAfter = selfFormAfter,

            // Self listed SECOND, behind a port nothing is listening on: Akka lets only the first
            // seed self-join, so this node can never form a cluster on its own except via the
            // fallback.
            SeedNodes = selfInSeeds
                ? new[]
                {
                    $"akka.tcp://{systemName}@127.0.0.1:{peerPort}",
                    $"akka.tcp://{systemName}@127.0.0.1:{selfPort}",
                }
                : new[] { $"akka.tcp://{systemName}@127.0.0.1:{peerPort}" },
        };

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

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

    /// <summary>Polls cluster state until it holds at least one Up member, or the deadline passes.</summary>
    private static async Task<bool> WaitForUpMemberAsync(ActorSystem system, TimeSpan timeout)
    {
        var cluster = Akka.Cluster.Cluster.Get(system);
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (cluster.State.Members.Any(m => m.Status == MemberStatus.Up))
            {
                return true;
            }

            await Task.Delay(200);
        }

        return cluster.State.Members.Any(m => m.Status == MemberStatus.Up);
    }

    private static async Task StopAsync(IHost host)
    {
        await host.StopAsync();
        host.Dispose();
    }

    /// <summary>
    /// An unconfigured deployment gets the fallback: a pair node that cold-starts while its peer is
    /// down must come up on its own, not wait for an operator.
    /// </summary>
    [Fact]
    public void SelfFormAfter_defaults_to_ten_seconds()
    {
        new AkkaClusterOptions().SelfFormAfter.ShouldBe(TimeSpan.FromSeconds(10));
    }

    /// <summary>
    /// The core scenario: a node listed as a non-first seed, booting with its peer dead, becomes Up
    /// on its own after the window. Without the fallback it loops on InitJoin forever.
    /// </summary>
    [Fact]
    public async Task Lone_non_first_seed_self_forms_after_the_window()
    {
        var selfPort = FreePort();
        var peerPort = FreePort();
        var host = await StartNodeAsync(
            "otopcua-selfform-1", selfPort, peerPort, TimeSpan.FromSeconds(2));
        try
        {
            var system = host.Services.GetRequiredService<ActorSystem>();

            (await WaitForUpMemberAsync(system, TimeSpan.FromSeconds(20)))
                .ShouldBeTrue("a lone non-first seed must self-form and come Up unattended");
        }
        finally
        {
            await StopAsync(host);
        }
    }

    /// <summary>
    /// The opt-out is real: with the window unset the node keeps waiting on InitJoin. The positive
    /// control proves the node was formable and only the fallback was missing — otherwise this test
    /// would pass just as happily against a node that could never join anything.
    /// </summary>
    [Fact]
    public async Task Disabled_fallback_keeps_waiting()
    {
        var selfPort = FreePort();
        var peerPort = FreePort();
        var host = await StartNodeAsync("otopcua-selfform-2", selfPort, peerPort, selfFormAfter: null);
        try
        {
            var system = host.Services.GetRequiredService<ActorSystem>();

            (await WaitForUpMemberAsync(system, TimeSpan.FromSeconds(6)))
                .ShouldBeFalse("with SelfFormAfter unset the node must keep waiting on InitJoin");

            // Positive control: the node WAS formable all along.
            var cluster = Akka.Cluster.Cluster.Get(system);
            cluster.Join(cluster.SelfAddress);
            (await WaitForUpMemberAsync(system, TimeSpan.FromSeconds(20)))
                .ShouldBeTrue("positive control — an explicit self-join must bring this node Up");
        }
        finally
        {
            await StopAsync(host);
        }
    }

    /// <summary>
    /// THE island guard. A node that is not in its own seed list — the current docker-dev site-node
    /// topology, where site-a/site-b list only central-1 — must never self-form: it would island
    /// itself from the real seeds permanently. Positive control as above.
    /// </summary>
    [Fact]
    public async Task Site_node_seeded_only_by_central_never_self_forms()
    {
        var selfPort = FreePort();
        var peerPort = FreePort();
        var host = await StartNodeAsync(
            "otopcua-selfform-3", selfPort, peerPort, TimeSpan.FromSeconds(1), selfInSeeds: false);
        try
        {
            var system = host.Services.GetRequiredService<ActorSystem>();

            (await WaitForUpMemberAsync(system, TimeSpan.FromSeconds(5)))
                .ShouldBeFalse("a node absent from its own seed list must wait, not island itself");

            var cluster = Akka.Cluster.Cluster.Get(system);
            cluster.Join(cluster.SelfAddress);
            (await WaitForUpMemberAsync(system, TimeSpan.FromSeconds(20)))
                .ShouldBeTrue("positive control — an explicit self-join must bring this node Up");
        }
        finally
        {
            await StopAsync(host);
        }
    }
}