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

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

/// <summary>
/// Cluster-level proof of the admin "Trigger failover" control: the current active node
/// (the OLDEST Up member — <see cref="ActiveNodeEvaluator"/>'s rule, which is where
/// ClusterSingletonManager places singletons) leaves GRACEFULLY, so its singletons hand over
/// rather than being killed, and the survivor becomes the active node.
///
/// <para>Graceful <c>Leave</c>, never <c>Down</c>: a Down would skip singleton hand-off and
/// leave the pair to the downing strategy. The peer guard matters just as much — failing over
/// a single-node cluster is not a failover, it is an outage, so the service refuses.</para>
/// </summary>
public sealed class ManualFailoverTests : IAsyncDisposable
{
    private readonly List<ActorSystem> _systems = new();

    /// <summary>Starts a one-node cluster (self-first seed list, so it forms alone).</summary>
    private ActorSystem StartSoloNode()
    {
        var port = TwoNodeClusterFixture.GetFreeTcpPort();
        var nodeOptions = new NodeOptions { Role = "Central", NodeHostname = "127.0.0.1", RemotingPort = port };
        var clusterOptions = new ClusterOptions
        {
            SeedNodes = new List<string> { $"akka.tcp://scadabridge@127.0.0.1:{port}" },
            AllowSingleNodeCluster = true,
            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);
        return system;
    }

    [Fact]
    public async Task Failover_makes_the_oldest_leave_and_the_survivor_take_over()
    {
        await using var f = await TwoNodeClusterFixture.StartAsync();
        var oldest = Akka.Cluster.Cluster.Get(f.NodeA); // NodeA started first = oldest
        Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(oldest, "Central"),
            "precondition: NodeA must be the active (oldest Up) node before the failover");

        // Issued from the OTHER node, exactly as the UI does it (the button is served by
        // whichever node Traefik routed the admin to).
        var target = AkkaManualFailoverService.FailOverCore(f.NodeB, "Central");

        Assert.Equal(oldest.SelfAddress, target);

        // Graceful exit path: the left node's own ActorSystem terminates…
        await f.NodeA.WhenTerminated.WaitAsync(TimeSpan.FromSeconds(30));
        // …and the survivor becomes a 1-member cluster and the oldest-Up active node.
        await TwoNodeClusterFixture.WaitForMemberRemoved(f.NodeB, oldest.SelfAddress, TimeSpan.FromSeconds(30));
        Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(Akka.Cluster.Cluster.Get(f.NodeB), "Central"));
    }

    [Fact]
    public async Task Failover_refuses_when_no_peer_exists()
    {
        var solo = StartSoloNode();
        await TwoNodeClusterFixture.WaitForMembersUp(solo, 1, TimeSpan.FromSeconds(30));

        var target = AkkaManualFailoverService.FailOverCore(solo, "Central");

        Assert.Null(target);
        // Positive assert: the refusal left the node running and still active — the guard
        // must not have half-issued a Leave.
        await Task.Delay(TimeSpan.FromSeconds(2));
        Assert.Equal(MemberStatus.Up, Akka.Cluster.Cluster.Get(solo).SelfMember.Status);
        Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(Akka.Cluster.Cluster.Get(solo), "Central"));
    }

    [Fact]
    public async Task Failover_dry_run_names_the_target_without_moving_the_cluster()
    {
        // The service resolves the target first (to audit it BEFORE acting); that probe must
        // not itself perturb the cluster.
        await using var f = await TwoNodeClusterFixture.StartAsync();
        var oldest = Akka.Cluster.Cluster.Get(f.NodeA);

        var probed = AkkaManualFailoverService.FailOverCore(f.NodeB, "Central", dryRun: true);

        Assert.Equal(oldest.SelfAddress, probed);
        await Task.Delay(TimeSpan.FromSeconds(3));
        Assert.Equal(2, Akka.Cluster.Cluster.Get(f.NodeB).State.Members.Count(m => m.Status == MemberStatus.Up));
        Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(oldest, "Central"));
    }

    public async ValueTask DisposeAsync()
    {
        foreach (var s in _systems)
        {
            if (s is { WhenTerminated.IsCompleted: false })
            {
                try { await s.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); } catch { /* teardown */ }
            }
        }
    }
}