using System.Collections.Immutable;
using System.Reflection;
using Akka.Actor;
using Akka.Cluster;
using Akka.TestKit.Xunit2;
using Akka.Util;
using ZB.MOM.WW.ScadaBridge.Host.Actors;

namespace ZB.MOM.WW.ScadaBridge.Host.Tests;

/// <summary>
/// ScadaBridge#106: the decision logic of <see cref="LeaderAutoDownBase"/> must match Akka.NET
/// 1.5.62 <c>AutoDownBase</c> exactly. These cases are the ones Akka's own <c>AutoDownSpec</c>
/// pins — cluster events are fed straight to the actor and the <c>Down</c> seam reports to a
/// probe, so the leader-only and reachability semantics are proven without a running cluster.
/// The two-node real-cluster behaviour is proven separately in IntegrationTests
/// (<c>LeaderAutoDownFailoverTests</c>).
/// </summary>
public class LeaderAutoDownTests : TestKit
{
    private static readonly Address MemberA = new("akka.tcp", "sys", "a", 2552);
    private static readonly Address MemberB = new("akka.tcp", "sys", "b", 2552);
    private static readonly Address MemberC = new("akka.tcp", "sys", "c", 2552);

    private sealed record DownCalled(Address Address);

    /// <summary>Test double: <c>Down</c> reports to a probe instead of calling the cluster.</summary>
    private sealed class TestLeaderAutoDown : LeaderAutoDownBase
    {
        private readonly IActorRef _probe;

        public TestLeaderAutoDown(Address selfAddress, TimeSpan unreachableAfter, IActorRef probe)
            : base(unreachableAfter)
        {
            SelfAddress = selfAddress;
            _probe = probe;
        }

        public override Address SelfAddress { get; }

        public override IScheduler Scheduler => Context.System.Scheduler;

        public override void Down(Address node)
        {
            // Report EVERY call, leader or not: a base that wrongly downs while not leader must
            // show up as an unexpected message, not be swallowed by a throw-and-restart.
            _probe.Tell(new DownCalled(node));
            if (!IsLeader)
            {
                _probe.Tell("down-called-while-not-leader");
            }
        }
    }

    /// <summary>Akka's <c>Member</c> constructors are internal; tests build members via the internal factory.</summary>
    private static Member CreateMember(Address address, MemberStatus status)
    {
        var create = typeof(Member).GetMethod(
            "Create",
            BindingFlags.Static | BindingFlags.NonPublic,
            new[] { typeof(UniqueAddress), typeof(int), typeof(MemberStatus), typeof(ImmutableHashSet<string>), typeof(AppVersion) })
            ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(…) not found — Akka internals changed.");
        return (Member)create.Invoke(null, new object[]
        {
            new UniqueAddress(address, 1), 1, status, ImmutableHashSet<string>.Empty, AppVersion.Zero,
        })!;
    }

    private IActorRef AutoDownActor(TimeSpan unreachableAfter) =>
        Sys.ActorOf(Props.Create(() => new TestLeaderAutoDown(MemberA, unreachableAfter, TestActor)));

    /// <summary>A non-leader never downs an unreachable member.</summary>
    [Fact]
    public void NotLeader_DoesNotDownUnreachable()
    {
        var a = AutoDownActor(TimeSpan.Zero);
        a.Tell(new ClusterEvent.LeaderChanged(MemberB));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberC, MemberStatus.Up)));
        ExpectNoMsg(TimeSpan.FromSeconds(1));
    }

    /// <summary>The leader downs an unreachable member immediately when the window is zero.</summary>
    [Fact]
    public void Leader_DownsUnreachable_WhenWindowIsZero()
    {
        var a = AutoDownActor(TimeSpan.Zero);
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberB, MemberStatus.Up)));
        ExpectMsg(new DownCalled(MemberB));
    }

    /// <summary>The leader downs an unreachable member only after the full window.</summary>
    [Fact]
    public void Leader_DownsUnreachable_AfterWindow()
    {
        var a = AutoDownActor(TimeSpan.FromSeconds(2));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberB, MemberStatus.Up)));
        ExpectNoMsg(TimeSpan.FromSeconds(1));
        ExpectMsg(new DownCalled(MemberB), TimeSpan.FromSeconds(3));
    }

    /// <summary>The initial snapshot drives leadership and schedules members already unreachable.</summary>
    [Fact]
    public void CurrentClusterState_Leader_DownsAlreadyUnreachable()
    {
        var a = AutoDownActor(TimeSpan.Zero);
        var b = CreateMember(MemberB, MemberStatus.Up);
        a.Tell(new ClusterEvent.CurrentClusterState(
            ImmutableSortedSet<Member>.Empty,
            ImmutableHashSet.Create(b),
            ImmutableHashSet<Address>.Empty,
            MemberA,
            ImmutableDictionary<string, Address>.Empty));
        ExpectMsg(new DownCalled(MemberB));
    }

    /// <summary>A pending member is downed as soon as this node becomes leader.</summary>
    [Fact]
    public void BecomingLeader_DownsPendingUnreachable()
    {
        var a = AutoDownActor(TimeSpan.Zero);
        a.Tell(new ClusterEvent.LeaderChanged(MemberB));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberC, MemberStatus.Up)));
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        ExpectMsg(new DownCalled(MemberC));
    }

    /// <summary>A member that timed out while not leader is downed on becoming leader.</summary>
    [Fact]
    public void BecomingLeader_AfterWindow_DownsPendingUnreachable()
    {
        var a = AutoDownActor(TimeSpan.FromSeconds(1));
        a.Tell(new ClusterEvent.LeaderChanged(MemberB));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberC, MemberStatus.Up)));
        ExpectNoMsg(TimeSpan.FromSeconds(2));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        ExpectMsg(new DownCalled(MemberC));
    }

    /// <summary>Losing leadership inside the window means no Down when the timer fires.</summary>
    [Fact]
    public void LosingLeadership_InsideWindow_DoesNotDown()
    {
        var a = AutoDownActor(TimeSpan.FromSeconds(2));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberC, MemberStatus.Up)));
        a.Tell(new ClusterEvent.LeaderChanged(MemberB));
        ExpectNoMsg(TimeSpan.FromSeconds(3));
    }

    /// <summary>A member that becomes reachable again inside the window is not downed.</summary>
    [Fact]
    public void ReachableAgain_InsideWindow_DoesNotDown()
    {
        var a = AutoDownActor(TimeSpan.FromSeconds(2));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        var b = CreateMember(MemberB, MemberStatus.Up);
        a.Tell(new ClusterEvent.UnreachableMember(b));
        a.Tell(new ClusterEvent.ReachableMember(b));
        ExpectNoMsg(TimeSpan.FromSeconds(3));
    }

    /// <summary>A pending member that becomes reachable again is not downed on becoming leader.</summary>
    [Fact]
    public void ReachableAgain_WhilePending_IsForgotten()
    {
        var a = AutoDownActor(TimeSpan.Zero);
        a.Tell(new ClusterEvent.LeaderChanged(MemberB));
        var c = CreateMember(MemberC, MemberStatus.Up);
        a.Tell(new ClusterEvent.UnreachableMember(c));
        a.Tell(new ClusterEvent.ReachableMember(c));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        ExpectNoMsg(TimeSpan.FromSeconds(1));
    }

    /// <summary>A member removed inside the window is not downed.</summary>
    [Fact]
    public void RemovedInsideWindow_DoesNotDown()
    {
        var a = AutoDownActor(TimeSpan.FromSeconds(2));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        var b = CreateMember(MemberB, MemberStatus.Up);
        a.Tell(new ClusterEvent.UnreachableMember(b));
        a.Tell(new ClusterEvent.MemberRemoved(b.Copy(MemberStatus.Removed), MemberStatus.Up));
        ExpectNoMsg(TimeSpan.FromSeconds(3));
    }

    /// <summary>Members already Down or Exiting are skipped (Akka's ConvergenceSkipUnreachableWithMemberStatus).</summary>
    /// <param name="status">The member status.</param>
    [Theory]
    [InlineData(MemberStatus.Down)]
    [InlineData(MemberStatus.Exiting)]
    public void AlreadyDownOrExiting_IsNotDownedAgain(MemberStatus status)
    {
        var a = AutoDownActor(TimeSpan.Zero);
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberB, status)));
        ExpectNoMsg(TimeSpan.FromSeconds(1));
    }

    /// <summary>A repeated UnreachableMember for an already-scheduled member does not schedule a second Down.</summary>
    [Fact]
    public void RepeatedUnreachable_DownsOnce()
    {
        var a = AutoDownActor(TimeSpan.FromSeconds(1));
        a.Tell(new ClusterEvent.LeaderChanged(MemberA));
        var b = CreateMember(MemberB, MemberStatus.Up);
        a.Tell(new ClusterEvent.UnreachableMember(b));
        a.Tell(new ClusterEvent.UnreachableMember(b));
        ExpectMsg(new DownCalled(MemberB), TimeSpan.FromSeconds(3));
        ExpectNoMsg(TimeSpan.FromSeconds(2));
    }
}