using Akka.Cluster;
using Shouldly;
using Xunit;

namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests;

/// <summary>
/// The load-bearing integration half of arch-review Critical 1 (03/S1) — proves the cluster actually
/// FAILS OVER when the oldest node hard-crashes: the split-brain resolver downs the alone crashed node
/// and the survivor takes over as <c>driver</c>-role leader. The unit guard only asserts the resolver is
/// wired; only a real hard-kill proves failover happens.
///
/// <para>The distinction the graceful <see cref="TwoNodeClusterHarness.StopNodeBAsync"/> path can NOT
/// exercise: a graceful <c>Cluster.Leave</c> removes a member with no downing decision at all, so it
/// passes even with downing disabled. A hard KILL leaves the survivor staring at an <b>Unreachable</b>
/// member — and with no active downing provider that member stays Unreachable forever, never failing
/// over. Only an active resolver downs the alone oldest node and lets the survivor take over.</para>
///
/// <para><b>What the negative control established (2026-07-08):</b> removing the typed
/// <c>ClusterOptions.SplitBrainResolver</c> did NOT break this test — Akka.Cluster.Hosting's
/// <c>WithClustering</c> applies <c>SplitBrainResolverOption.Default</c> when the option is null, which
/// enables the SBR downing provider that then reads the <c>akka.conf</c> <c>keep-oldest</c> block (present
/// on master before Critical 1). So the resolver is active via the framework default + HOCON; the typed
/// option is reinforcing, not the sole activator. This test was verified to genuinely require an active
/// downing provider by a separate run forcing explicit NoDowning
/// (<c>akka.cluster.downing-provider-class = ""</c>), under which it correctly times out with the node
/// stuck Up-but-Unreachable. It therefore guards the failover OUTCOME regardless of activation path.</para>
///
/// <para>Traited <c>Failover</c> — it is deliberately slow (~acceptable-heartbeat-pause 10s +
/// stable-after 15s before the resolver acts).</para>
/// </summary>
[Trait("Category", "Failover")]
public sealed class HardKillFailoverTests
{
    // acceptable-heartbeat-pause (10s) + stable-after (15s) + convergence/takeover margin.
    private static readonly TimeSpan FailoverTimeout = TimeSpan.FromSeconds(60);

    /// <summary>
    /// Hard-killing the oldest node (abrupt ActorSystem terminate, no graceful Leave) must leave the
    /// survivor as the sole <c>driver</c>-role leader — proving the split-brain resolver downed the alone
    /// oldest node and failed over. Without an active resolver the survivor keeps A Unreachable and this
    /// times out.
    /// </summary>
    [Fact]
    public async Task Hard_kill_of_oldest_node_fails_over_to_survivor_via_split_brain_resolver()
    {
        await using var harness = await TwoNodeClusterHarness.StartAsync();

        // Precondition: a converged 2-member cluster from node B's own view.
        var clusterB = Akka.Cluster.Cluster.Get(harness.NodeBSystem);
        clusterB.State.Members.Count(m => m.Status == MemberStatus.Up).ShouldBe(2);

        // Crash the seed / oldest member with no graceful Leave — B now sees it Unreachable.
        await harness.HardKillNodeAAsync();

        // Only an ACTIVE split-brain resolver turns that Unreachable into a downing + takeover. This
        // waits for the settled outcome — B as the sole Up member AND driver-role leader, held stable
        // across consecutive polls — and THROWS if it never happens (the signature of an inactive
        // resolver: A stuck Up-but-Unreachable, no takeover). A clean return IS the SBR proof.
        var elapsed = await harness.WaitForNodeBSoleDriverLeaderAsync(FailoverTimeout);
        elapsed.ShouldBeLessThan(FailoverTimeout);

        TestContext.Current.SendDiagnosticMessage($"Failover completed in {elapsed.TotalSeconds:F1}s");
    }
}