using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using Akka.Remote;
using Akka.Remote.Transport;
using Xunit;
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>
/// ScadaBridge#106: the operational scenarios of the in-repo <see cref="LeaderAutoDowningProvider"/>
/// beyond the hard-crash clones in <see cref="LeaderAutoDownFailoverTests"/> — cold starts
/// (together, and alone with the peer dead: the shape that makes SBR <c>down-if-alone</c>
/// self-down on a cold recreate), a FLAPPING peer that keeps going unreachable for less than
/// the window, a real partition (the accepted dual-active trade), and the opt-in switch being
/// off (the old providers are what Akka loads).
/// <para>
/// Nodes are built from the production <c>BuildHocon</c> output with the shipped self-first
/// seed ordering, exactly like <see cref="SelfFirstSeedBootstrapTests"/>. The flapping and
/// partition cases add Akka's own throttler transport adapter (<c>trttl</c>, the one the
/// MultiNodeTestRunner uses for <c>Blackhole</c>) so the network can be cut and restored
/// without killing either node.
/// </para>
/// </summary>
public sealed class LeaderAutoDownScenarioTests : IAsyncLifetime
{
    private const string Strategy = "leader-auto-down";
    private static readonly TimeSpan Heartbeat = TimeSpan.FromMilliseconds(500);
    private static readonly TimeSpan Threshold = TimeSpan.FromSeconds(2);

    private readonly List<ActorSystem> _systems = new();

    private ActorSystem StartNode(
        int selfPort, int peerPort, string strategy = Strategy, TimeSpan? stableAfter = null, bool throttled = false)
    {
        // With the throttler applied the published address scheme becomes akka.trttl.tcp, so the
        // seeds must carry it too or the nodes never find each other.
        var scheme = throttled ? "akka.trttl.tcp" : "akka.tcp";
        var nodeOptions = new NodeOptions { Role = "Central", NodeHostname = "127.0.0.1", RemotingPort = selfPort };
        var clusterOptions = new ClusterOptions
        {
            SeedNodes = new List<string>
            {
                $"{scheme}://scadabridge@127.0.0.1:{selfPort}",
                $"{scheme}://scadabridge@127.0.0.1:{peerPort}",
            },
            SplitBrainResolverStrategy = strategy,
            StableAfter = stableAfter ?? TimeSpan.FromSeconds(3),
            HeartbeatInterval = Heartbeat,
            FailureDetectionThreshold = Threshold,
            MinNrOfMembers = 1,
            DownIfAlone = true,
        };
        // Transport failure detection far above the cluster window: a blackhole must make the
        // MEMBER unreachable (what the downing provider reacts to), not tear the association down.
        var hocon = AkkaHostedService.BuildHocon(
            nodeOptions, clusterOptions, new[] { "Central" },
            TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(60));
        var overrides = "akka.coordinated-shutdown.run-by-actor-system-terminate = off\n"
            + (throttled ? "akka.remote.dot-netty.tcp.applied-adapters = [trttl]\n" : string.Empty);
        var config = ConfigurationFactory.ParseString(overrides)
            .WithFallback(ConfigurationFactory.ParseString(hocon));
        var system = ActorSystem.Create("scadabridge", config);
        _systems.Add(system);
        return system;
    }

    private static Task<bool> Throttle(ActorSystem from, ActorSystem to, ThrottleMode mode) =>
        ((RemoteActorRefProvider)((ExtendedActorSystem)from).Provider).Transport.ManagementCommand(
            new SetThrottle(Akka.Cluster.Cluster.Get(to).SelfAddress, ThrottleTransportAdapter.Direction.Both, mode));

    private static void AssertTwoUpNoneUnreachable(ActorSystem node, string because)
    {
        var cluster = Akka.Cluster.Cluster.Get(node);
        Assert.False(cluster.IsTerminated, $"{cluster.SelfAddress}: Cluster extension terminated ({because})");
        Assert.Equal(MemberStatus.Up, cluster.SelfMember.Status);
        Assert.Equal(2, cluster.State.Members.Count(m => m.Status == MemberStatus.Up));
        Assert.Empty(cluster.State.Unreachable);
    }

    /// <summary>
    /// Both nodes cold-start together: they converge on ONE two-member cluster and, once
    /// formed, nobody is downed — held for well past failure detection plus the window.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task BothNodesColdStartTogether_FormOneCluster_AndNeitherIsDowned()
    {
        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));
        Assert.IsType<LeaderAutoDowningProvider>(Akka.Cluster.Cluster.Get(nodeA).DowningProvider);

        await Task.Delay(Threshold + TimeSpan.FromSeconds(3) + TimeSpan.FromSeconds(5));
        AssertTwoUpNoneUnreachable(nodeA, "after a joint cold start");
        AssertTwoUpNoneUnreachable(nodeB, "after a joint cold start");
    }

    /// <summary>
    /// A lone cold start with the peer DEAD (the cold-recreate shape where SBR keep-oldest
    /// <c>down-if-alone</c> downs the lone node) forms a one-node cluster that stays Up past the
    /// window; when the peer is later cold-started it joins, and nobody is downed.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task LoneColdStart_PeerDead_StaysUp_ThenPeerColdStartsAndJoins()
    {
        var portA = TwoNodeClusterFixture.GetFreeTcpPort();
        var portB = TwoNodeClusterFixture.GetFreeTcpPort(); // nothing listening yet
        var nodeA = StartNode(portA, portB);

        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
        await Task.Delay(Threshold + TimeSpan.FromSeconds(3) + TimeSpan.FromSeconds(3));
        var clusterA = Akka.Cluster.Cluster.Get(nodeA);
        Assert.False(clusterA.IsTerminated, "lone cold-started node downed itself");
        Assert.Equal(MemberStatus.Up, clusterA.SelfMember.Status);

        var nodeB = StartNode(portB, portA);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(60));
        await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(60));
        await Task.Delay(TimeSpan.FromSeconds(5));
        AssertTwoUpNoneUnreachable(nodeA, "after the peer joined");
        AssertTwoUpNoneUnreachable(nodeB, "after the peer joined");
    }

    /// <summary>
    /// A FLAPPING peer — cut off long enough to be marked unreachable, restored before the
    /// window ends, over and over, for far longer than the window in total — is never downed:
    /// each return to reachable cancels its pending Down and the next cut starts a fresh window.
    /// The test first proves the peer really did go unreachable on (almost) every cut, so the
    /// "never downed" outcome is not vacuous.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task FlappingPeer_UnreachableShorterThanWindow_IsNeverDowned()
    {
        var stableAfter = TimeSpan.FromSeconds(8);
        var portA = TwoNodeClusterFixture.GetFreeTcpPort();
        var portB = TwoNodeClusterFixture.GetFreeTcpPort();
        var nodeA = StartNode(portA, portB, stableAfter: stableAfter, throttled: true);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
        var nodeB = StartNode(portB, portA, stableAfter: stableAfter, throttled: true);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(30));
        await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(30));

        var clusterA = Akka.Cluster.Cluster.Get(nodeA);
        const int flaps = 4;
        var flapsSeenUnreachable = 0;
        var started = DateTime.UtcNow;
        for (var i = 0; i < flaps; i++)
        {
            Assert.True(await Throttle(nodeA, nodeB, Blackhole.Instance), "SetThrottle(Blackhole) rejected");
            var seen = false;
            var cutUntil = DateTime.UtcNow + TimeSpan.FromSeconds(4.5);
            while (DateTime.UtcNow < cutUntil)
            {
                seen |= clusterA.State.Unreachable.Any();
                await Task.Delay(100);
            }

            Assert.True(await Throttle(nodeA, nodeB, Unthrottled.Instance), "SetThrottle(Unthrottled) rejected");
            await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(15));
            await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(15));
            if (seen)
            {
                flapsSeenUnreachable++;
            }
        }

        // Positive control: the cuts really produced unreachability, and the flapping lasted
        // longer than one window — a provider that did not cancel on reachable would have downed.
        Assert.True(flapsSeenUnreachable >= flaps - 1,
            $"only {flapsSeenUnreachable}/{flaps} cuts made the peer unreachable — the flap never happened");
        Assert.True(DateTime.UtcNow - started > stableAfter * 2);

        await Task.Delay(stableAfter + TimeSpan.FromSeconds(2));
        AssertTwoUpNoneUnreachable(nodeA, "flapping peer");
        AssertTwoUpNoneUnreachable(nodeB, "flapping peer");
    }

    /// <summary>
    /// A real partition longer than the window: each side is leader of its own reachable set,
    /// so each downs the other and BOTH stay Up — the dual-active outcome the availability-first
    /// decision accepts (identical to AutoDowning). Also the positive control for the flapping
    /// test: the same cut, held past the window, does down.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task PartitionLongerThanWindow_EachSideDownsTheOther_BothStayUp()
    {
        var stableAfter = TimeSpan.FromSeconds(3);
        var portA = TwoNodeClusterFixture.GetFreeTcpPort();
        var portB = TwoNodeClusterFixture.GetFreeTcpPort();
        var nodeA = StartNode(portA, portB, stableAfter: stableAfter, throttled: true);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
        var nodeB = StartNode(portB, portA, stableAfter: stableAfter, throttled: true);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(30));
        await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(30));
        var addressA = Akka.Cluster.Cluster.Get(nodeA).SelfAddress;
        var addressB = Akka.Cluster.Cluster.Get(nodeB).SelfAddress;

        Assert.True(await Throttle(nodeA, nodeB, Blackhole.Instance));

        await TwoNodeClusterFixture.WaitForMemberRemoved(nodeA, addressB, TimeSpan.FromSeconds(30));
        await TwoNodeClusterFixture.WaitForMemberRemoved(nodeB, addressA, TimeSpan.FromSeconds(30));
        foreach (var node in new[] { nodeA, nodeB })
        {
            var cluster = Akka.Cluster.Cluster.Get(node);
            Assert.False(cluster.IsTerminated, $"{cluster.SelfAddress} downed itself during the partition");
            Assert.Equal(MemberStatus.Up, cluster.SelfMember.Status);
            Assert.Equal(cluster.SelfAddress, cluster.State.Leader);
        }
    }

    /// <summary>
    /// The opt-in switch OFF: the default <c>auto-down</c> still loads Akka's own
    /// <c>AutoDowning</c> and <c>keep-oldest</c> still loads the SBR provider — the in-repo
    /// provider is loaded only for <c>leader-auto-down</c>.
    /// </summary>
    /// <param name="strategy">The configured strategy.</param>
    /// <param name="expectedProvider">The downing provider Akka must have instantiated.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Theory]
    [InlineData("auto-down", typeof(AutoDowning))]
    [InlineData("keep-oldest", typeof(Akka.Cluster.SBR.SplitBrainResolverProvider))]
    [InlineData(Strategy, typeof(LeaderAutoDowningProvider))]
    public async Task SwitchSelectsProvider_OldPathsUnchangedWhenOff(string strategy, Type expectedProvider)
    {
        var portA = TwoNodeClusterFixture.GetFreeTcpPort();
        var portB = TwoNodeClusterFixture.GetFreeTcpPort();
        var nodeA = StartNode(portA, portB, strategy: strategy);
        await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));

        Assert.IsType(expectedProvider, Akka.Cluster.Cluster.Get(nodeA).DowningProvider);
    }

    /// <summary>Prepares the fixture before the tests run.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public Task InitializeAsync() => Task.CompletedTask;

    /// <summary>Releases the resources held by this fixture asynchronously.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task DisposeAsync()
    {
        foreach (var s in _systems)
        {
            try { await s.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); } catch { /* teardown */ }
        }
    }
}