scadaproj/ZB.MOM.WW.Overview/src/ZB.MOM.WW.Overview/Polling/LeaderResolver.cs

namespace ZB.MOM.WW.Overview.Polling;

/// <summary>
/// Derives per-group leader state from whatever health check a member publishes its cluster view
/// under.
/// </summary>
/// <remarks>
/// Pure and static: given the same instance snapshots it always produces the same answer, which is
/// what makes the split-brain rule — the one output an operator would act on at 3am — exhaustively
/// testable without a cluster.
/// </remarks>
public static class LeaderResolver
{
    /// <summary>
    /// The data key holding the address of the node actually in charge — the oldest Up member,
    /// published by the active tier's <c>ActiveNodeHealthCheck</c> from ZB.MOM.WW.Health 0.3.0.
    /// </summary>
    public const string ActiveNodeDataKey = "activeNode";

    /// <summary>
    /// The data key holding the Akka cluster leader address, published by
    /// <c>AkkaClusterHealthCheck</c>. Fallback only — see <see cref="LeaderReportedBy"/>.
    /// </summary>
    public const string LeaderDataKey = "leader";

    /// <summary>Resolves leader state for every cluster group present in <paramref name="instances"/>.</summary>
    /// <param name="instances">One application's instances.</param>
    /// <returns>Group state in first-appearance order; empty when the application has no groups.</returns>
    public static IReadOnlyList<GroupSnapshot> Resolve(IEnumerable<InstanceSnapshot> instances)
    {
        ArgumentNullException.ThrowIfNull(instances);

        var all = instances as IReadOnlyList<InstanceSnapshot> ?? instances.ToList();
        var groups = new List<GroupSnapshot>();

        foreach (var name in all
                     .Select(i => i.Group)
                     .Where(g => !string.IsNullOrWhiteSpace(g))
                     .Distinct(StringComparer.Ordinal))
        {
            var members = all.Where(i => string.Equals(i.Group, name, StringComparison.Ordinal)).ToList();

            // A group of instances that have no active/standby concept has no leader to report;
            // rendering a leader chip for one would invent a distinction that does not exist.
            if (!members.Exists(m => m.HasActiveRole))
                continue;

            groups.Add(ResolveGroup(name!, members, all));
        }

        return groups;
    }

    private static GroupSnapshot ResolveGroup(
        string name,
        IReadOnlyList<InstanceSnapshot> members,
        IReadOnlyList<InstanceSnapshot> allInstances)
    {
        // Only members that are actually talking get a vote. A Down or Unreachable node's last
        // known leader is stale by definition, and counting it would manufacture a disagreement
        // every time a node goes away mid-failover — the exact moment the flag must stay honest.
        var votes = members
            .Where(m => m.Status is InstanceStatus.Up or InstanceStatus.Degraded)
            .Select(LeaderReportedBy)
            .Where(address => !string.IsNullOrWhiteSpace(address))
            .Select(address => address!)
            .ToList();

        if (votes.Count == 0)
            return new GroupSnapshot(name, null, null, Disagreement: false, []);

        var distinct = votes.Distinct(StringComparer.Ordinal).OrderBy(a => a, StringComparer.Ordinal).ToList();

        // The most-reported address wins the chip, ties broken ordinally so the display is stable
        // rather than flickering between two equally-supported claims on consecutive sweeps.
        var winner = votes
            .GroupBy(a => a, StringComparer.Ordinal)
            .OrderByDescending(g => g.Count())
            .ThenBy(g => g.Key, StringComparer.Ordinal)
            .First()
            .Key;

        return new GroupSnapshot(
            name,
            ResolveToInstanceName(winner, allInstances),
            winner,
            Disagreement: distinct.Count > 1,
            distinct);
    }

    /// <summary>Finds the address of the in-charge node one instance reports.</summary>
    /// <param name="instance">The instance whose report to search.</param>
    /// <returns>The reported address, or null when this instance publishes none.</returns>
    /// <remarks>
    ///     <para>
    ///         <b><c>activeNode</c> first, <c>leader</c> only as a fallback.</b> They are different
    ///         nodes and the chip must show the first. The Akka cluster <i>leader</i> is the
    ///         lowest-addressed Up member — address order, no relationship to time — while the node
    ///         that actually holds the singletons and answers 200 on <c>/health/active</c> is the
    ///         <i>oldest</i> Up member. They agree only until some node restarts, after which the
    ///         restarted node rejoins youngest but keeps its address; observed live on the rebuilt
    ///         OtOpcUa rig, where the leader was <c>central-1</c> while <c>central-2</c> was the
    ///         active node. Ranking <c>leader</c> first would put the Active chip on the standby.
    ///     </para>
    ///     <para>
    ///         The fallback still matters: it is what an instance running Health &lt; 0.3.0, or one
    ///         whose active tier was not probed, publishes. Showing the leader there is better than
    ///         showing nothing, and it is the pre-0.3.0 behaviour unchanged.
    ///     </para>
    ///     <para>
    ///         Matched on the DATA KEY, not on the check name. The two apps register the same shared
    ///         checks under different names — ScadaBridge calls its cluster check
    ///         <c>akka-cluster</c>, OtOpcUa calls it <c>akka</c>; the active check is
    ///         <c>active-node</c> in one and <c>cluster-primary</c> in the other — and a resolver
    ///         keyed on any of those names silently renders no chip for the other app, which is
    ///         exactly how this was missed until a live rig showed OtOpcUa's groups bare while
    ///         ScadaBridge's were fully populated. The data key is the part of the contract the
    ///         shared checks actually own.
    ///     </para>
    /// </remarks>
    internal static string? LeaderReportedBy(InstanceSnapshot instance)
    {
        if (instance.Report?.Entries is not { } entries)
            return null;

        string? leaderFallback = null;

        foreach (var entry in entries.Values)
        {
            if (entry.DataString(ActiveNodeDataKey) is { Length: > 0 } activeNode)
                return activeNode;

            leaderFallback ??= entry.DataString(LeaderDataKey) is { Length: > 0 } leader ? leader : null;
        }

        return leaderFallback;
    }

    /// <summary>
    /// Maps an Akka address to a registry instance name by host.
    /// </summary>
    /// <param name="akkaAddress">An address of the form <c>akka.tcp://system@host:port</c>.</param>
    /// <param name="instances">Candidate instances.</param>
    /// <returns>The matching instance's name, or null when nothing matches.</returns>
    /// <remarks>
    /// Host only, deliberately: the Akka remoting port and the HTTP port are different by design, so
    /// comparing ports would never match. An unmatched address is not an error — the card falls back
    /// to showing the raw address, which is still the useful thing to see.
    /// </remarks>
    internal static string? ResolveToInstanceName(string akkaAddress, IReadOnlyList<InstanceSnapshot> instances)
    {
        var host = HostOf(akkaAddress);
        if (host is null)
            return null;

        foreach (var instance in instances)
        {
            if (Uri.TryCreate(instance.BaseUrl, UriKind.Absolute, out var uri) &&
                string.Equals(uri.Host, host, StringComparison.OrdinalIgnoreCase))
            {
                return instance.Instance;
            }
        }

        return null;
    }

    /// <summary>Extracts the host from an Akka address.</summary>
    /// <param name="akkaAddress">The address to parse.</param>
    /// <returns>The host, or null when the address is not in the expected form.</returns>
    internal static string? HostOf(string? akkaAddress)
    {
        if (string.IsNullOrWhiteSpace(akkaAddress))
            return null;

        var at = akkaAddress.LastIndexOf('@');
        if (at < 0 || at == akkaAddress.Length - 1)
            return null;

        var hostAndPort = akkaAddress[(at + 1)..];
        var colon = hostAndPort.LastIndexOf(':');
        var host = colon > 0 ? hostAndPort[..colon] : hostAndPort;

        return host.Length == 0 ? null : host;
    }
}