using System.Text.RegularExpressions;

namespace ZB.MOM.WW.OtOpcUa.Cluster;

/// <summary>
/// Pure decision logic for the simultaneous-cold-start split-brain guard.
/// </summary>
/// <remarks>
///     <para>
///         Both nodes of a 2-node pair are self-first seeds so <b>either</b> can cold-start alone when
///         its partner is dead (see <see cref="AkkaClusterOptions.SeedNodes"/>). The cost is that when
///         BOTH cold-start at the same instant, each runs Akka's <c>FirstSeedNodeProcess</c>, times out
///         waiting for the other, and forms its OWN single-node cluster — a split brain (two Primaries
///         in one pair). Two independent clusters do not auto-merge, so the split persists until an
///         operator intervenes.
///     </para>
///     <para>
///         This guard breaks the symmetry deterministically without giving up cold-start-alone. The
///         node with the lexicographically <b>lower</b> canonical address is the <i>preferred
///         founder</i>: it always uses self-first order and forms immediately. The <b>higher</b> node
///         first probes whether its partner's Akka endpoint is reachable (see
///         <see cref="ClusterBootstrapCoordinator"/>): reachable ⇒ peer-first order so it JOINS the
///         founder rather than racing it; unreachable after the probe window ⇒ the partner is genuinely
///         down, so it falls back to self-first and forms alone. Exactly one node founds when both are
///         present; the lower node always founds when it starts alone.
///     </para>
///     <para>
///         <b>Residual trade-off (accepted).</b> Once the higher node observes the partner reachable it
///         commits to peer-first — Akka's <c>JoinSeedNodeProcess</c> retries <c>InitJoin</c> forever and
///         never self-forms. If the founder dies in the small window between the probe succeeding and
///         the join completing, the higher node hangs unjoined until it is restarted (a restart re-runs
///         the guard, finds the founder unreachable, and self-forms). We deliberately do NOT re-decide
///         mid-handshake — that is exactly the retired <c>SelfFormAfter</c> failure mode. Instead
///         <see cref="ClusterBootstrapCoordinator"/> makes the hang operator-visible with a warning if
///         the node has not come Up within a bounded time after committing peer-first.
///     </para>
///     <para>
///         This decides the join order BEFORE issuing a single <c>JoinSeedNodes</c>, from an explicit
///         reachability signal — unlike the retired <c>SelfFormAfter</c> watchdog, which fired a
///         <c>Join(self)</c> mid-handshake on a bare timeout and could not tell "no seed answered" from
///         "a join is in flight", islanding a node a failover had just bounced.
///     </para>
/// </remarks>
public static class ClusterBootstrapGuard
{
    private static readonly Regex SeedPattern =
        new(@"^akka(?:\.[a-z0-9]+)?://[^@/]+@(?<host>[^:/]+):(?<port>\d+)", RegexOptions.IgnoreCase | RegexOptions.Compiled);

    /// <summary>The analyzed bootstrap role of this node within its seed list.</summary>
    /// <param name="Applies">True only when the guard engages: exactly two seeds, one of them this node.</param>
    /// <param name="IsFounder">True when this node is the preferred founder (lower address) — form immediately, no probe.</param>
    /// <param name="PartnerHost">The partner's advertised host (to probe when this node is the higher one); null when not applicable.</param>
    /// <param name="PartnerPort">The partner's Akka port.</param>
    /// <param name="SelfFirstOrder"><c>[self, partner]</c> — self-first; forms a new cluster when no peer answers.</param>
    /// <param name="PeerFirstOrder"><c>[partner, self]</c> — peer-first; joins the partner's cluster, never self-forms while it is up.</param>
    public sealed record BootstrapRole(
        bool Applies,
        bool IsFounder,
        string? PartnerHost,
        int PartnerPort,
        string[] SelfFirstOrder,
        string[] PeerFirstOrder);

    /// <summary>
    /// Analyzes the configured seed list to decide this node's bootstrap role. The guard engages only
    /// for the Phase-6 pair shape (exactly two seeds, one of them this node); any other shape
    /// (single-seed legacy site node, three+ seeds, self absent) returns <see cref="BootstrapRole.Applies"/>
    /// = false and the caller joins the configured seeds unchanged.
    /// </summary>
    /// <param name="seeds">The configured seed URIs (self-first by convention, but order is not relied on here).</param>
    /// <param name="selfHost">This node's advertised host (<see cref="AkkaClusterOptions.PublicHostname"/>).</param>
    /// <param name="selfPort">This node's Akka port (<see cref="AkkaClusterOptions.Port"/>).</param>
    /// <returns>The analyzed role; never null.</returns>
    public static BootstrapRole Analyze(IReadOnlyList<string> seeds, string selfHost, int selfPort)
    {
        ArgumentNullException.ThrowIfNull(seeds);
        ArgumentException.ThrowIfNullOrWhiteSpace(selfHost);

        var na = new BootstrapRole(false, false, null, 0, Array.Empty<string>(), Array.Empty<string>());
        if (seeds.Count != 2) return na;

        var selfKey = Key(selfHost, selfPort);
        string? self = null, partner = null;
        string? partnerHost = null;
        var partnerPort = 0;

        foreach (var seed in seeds)
        {
            if (!TryParse(seed, out var host, out var port)) return na;
            if (string.Equals(Key(host, port), selfKey, StringComparison.OrdinalIgnoreCase))
                self = seed;
            else
            {
                partner = seed;
                partnerHost = host;
                partnerPort = port;
            }
        }

        // Self must be exactly one of the two, and the other must be a distinct partner.
        if (self is null || partner is null || partnerHost is null) return na;

        var selfFirst = new[] { self, partner };
        var peerFirst = new[] { partner, self };

        // Deterministic tie-break: the lower canonical address is the preferred founder. Both nodes
        // compute the same comparison (same two seeds), so exactly one is the founder. Case-INSENSITIVE
        // to match how self/partner are classified above (and AkkaClusterOptionsValidator.IsSelf):
        // container/DNS hostnames are conventionally case-insensitive, and a casing difference between
        // the two sides' seed config must NOT make both think they are the founder (which would reopen
        // the very split this guard closes).
        var isFounder = string.Compare(
            Key(selfHost, selfPort),
            Key(partnerHost, partnerPort),
            StringComparison.OrdinalIgnoreCase) < 0;

        return new BootstrapRole(true, isFounder, partnerHost, partnerPort, selfFirst, peerFirst);
    }

    /// <summary>
    /// The order the higher (non-founder) node uses once it has learned whether its partner is
    /// reachable: peer-first when the founder is up (join it), self-first when the founder is absent
    /// (form alone — cold-start-alone preserved).
    /// </summary>
    /// <param name="role">The analyzed role (must be applicable and NOT the founder).</param>
    /// <param name="partnerReachable">Whether the partner's Akka endpoint became reachable within the probe window.</param>
    /// <returns>The seed order to pass to <c>Cluster.JoinSeedNodes</c>.</returns>
    public static string[] HigherNodeOrder(BootstrapRole role, bool partnerReachable)
    {
        ArgumentNullException.ThrowIfNull(role);
        return partnerReachable ? role.PeerFirstOrder : role.SelfFirstOrder;
    }

    /// <summary>Parses <c>akka.tcp://system@host:port[/...]</c> into host + port.</summary>
    /// <param name="seed">The seed URI.</param>
    /// <param name="host">The parsed advertised host.</param>
    /// <param name="port">The parsed Akka port.</param>
    /// <returns>True when the seed matched the expected shape.</returns>
    public static bool TryParse(string? seed, out string host, out int port)
    {
        host = string.Empty;
        port = 0;
        if (string.IsNullOrWhiteSpace(seed)) return false;

        var m = SeedPattern.Match(seed.Trim());
        if (!m.Success) return false;
        if (!int.TryParse(m.Groups["port"].Value, out port)) return false;
        host = m.Groups["host"].Value;
        return true;
    }

    private static string Key(string host, int port) => $"{host}:{port}";
}