lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Redundancy/RedundancyStateActor.cs

using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.Health.Akka;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using CommonsRedundancyRole = ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy.RedundancyRole;

namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;

/// <summary>
/// Admin-role cluster singleton that aggregates per-node cluster events into a
/// <see cref="RedundancyStateChanged"/> snapshot and publishes it on the <c>redundancy-state</c>
/// DistributedPubSub topic. Subscribers (notably the OPC UA host's ServiceLevel calc) react to
/// topology changes without polling.
///
/// Recomputation is debounced by <see cref="DebounceWindow"/> — a burst of cluster events from
/// a rolling restart should produce one published snapshot, not one per event.
/// </summary>
public sealed class RedundancyStateActor : ReceiveActor, IWithTimers
{
    public const string Topic = "redundancy-state";
    public static readonly TimeSpan DebounceWindow = TimeSpan.FromMilliseconds(250);

    /// <summary>
    /// Default interval for the periodic heartbeat re-publish. DistributedPubSub does not replay
    /// to late subscribers, so a node that subscribes just after a change-driven publish would
    /// otherwise never learn its role until the next membership change. The heartbeat re-publishes
    /// the current snapshot so any missed subscriber converges within this interval.
    /// </summary>
    public static readonly TimeSpan DefaultHeartbeatInterval = TimeSpan.FromSeconds(10);

    private readonly ILoggingAdapter _log = Context.GetLogger();
    private readonly Akka.Cluster.Cluster _cluster;
    private readonly Action<object>? _broadcastOverride;
    private readonly TimeSpan _heartbeatInterval;
    private bool _dirty;

    /// <summary>Gets the timer scheduler for this actor.</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>Creates a Props for creating a RedundancyStateActor.</summary>
    /// <param name="broadcast">Optional broadcast override for testing.</param>
    /// <param name="heartbeatInterval">
    /// Optional periodic heartbeat re-publish interval; defaults to
    /// <see cref="DefaultHeartbeatInterval"/>. Injectable so tests can use a short interval.
    /// </param>
    /// <returns>Props for actor creation.</returns>
    public static Props Props(Action<object>? broadcast = null, TimeSpan? heartbeatInterval = null) =>
        Akka.Actor.Props.Create(() => new RedundancyStateActor(broadcast, heartbeatInterval));

    /// <summary>Initializes a new instance of RedundancyStateActor with no broadcast override.</summary>
    public RedundancyStateActor() : this(null, null) { }

    /// <summary>Initializes a new instance of RedundancyStateActor.</summary>
    /// <param name="broadcast">Optional broadcast override for testing.</param>
    /// <param name="heartbeatInterval">
    /// Optional periodic heartbeat re-publish interval; defaults to
    /// <see cref="DefaultHeartbeatInterval"/>.
    /// </param>
    public RedundancyStateActor(Action<object>? broadcast, TimeSpan? heartbeatInterval = null)
    {
        _cluster = Akka.Cluster.Cluster.Get(Context.System);
        _broadcastOverride = broadcast;
        _heartbeatInterval = heartbeatInterval ?? DefaultHeartbeatInterval;

        Receive<ClusterEvent.IMemberEvent>(_ => MarkDirty());
        Receive<ClusterEvent.LeaderChanged>(_ => MarkDirty());
        Receive<ClusterEvent.RoleLeaderChanged>(_ => MarkDirty());
        Receive<ClusterEvent.CurrentClusterState>(_ => MarkDirty());
        Receive<ClusterEvent.ReachabilityEvent>(_ => MarkDirty());
        Receive<RecomputeNow>(_ => PublishIfDirty());
        Receive<Heartbeat>(_ => { _dirty = true; PublishIfDirty(); });
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        _cluster.Subscribe(
            Self,
            ClusterEvent.InitialStateAsEvents,
            typeof(ClusterEvent.IMemberEvent),
            typeof(ClusterEvent.LeaderChanged),
            typeof(ClusterEvent.RoleLeaderChanged),
            typeof(ClusterEvent.ReachabilityEvent));

        // DistributedPubSub does not replay to late subscribers, so re-publish the current
        // snapshot on a recurring heartbeat — any node that missed the change-driven publish
        // converges within _heartbeatInterval. Idempotent: consumers re-cache the same role.
        Timers.StartPeriodicTimer("heartbeat", Heartbeat.Instance, _heartbeatInterval);
    }

    /// <inheritdoc />
    protected override void PostStop() => _cluster.Unsubscribe(Self);

    private void MarkDirty()
    {
        _dirty = true;
        Timers.StartSingleTimer("debounce", RecomputeNow.Instance, DebounceWindow);
    }

    private void PublishIfDirty()
    {
        if (!_dirty) return;
        _dirty = false;

        var snapshot = BuildSnapshot();
        var msg = new RedundancyStateChanged(snapshot, CorrelationId.NewId());
        if (_broadcastOverride is not null) _broadcastOverride(msg);
        else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(Topic, msg));
        _log.Debug("Published RedundancyStateChanged with {Count} nodes", snapshot.Count);
    }

    /// <summary>The cluster role that marks a node as carrying the driver runtime.</summary>
    public const string DriverRole = RoleParser.Driver;

    /// <summary>
    /// Selects the driver Primary: the <b>oldest</b> Up member carrying the <see cref="DriverRole"/>.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         This deliberately does <b>not</b> use <c>ClusterState.RoleLeader(role)</c>, which it
    ///         previously did. Role leader is the <i>lowest-addressed</i> Up member with the role —
    ///         address order (host, then port), with no relationship to time. Oldest is the member
    ///         with the lowest up-number, and it is what <c>ClusterSingletonManager</c> uses to place
    ///         singletons.
    ///     </para>
    ///     <para>
    ///         On a freshly-formed cluster the two agree, which is why the discrepancy was invisible.
    ///         They diverge after any restart: the restarted node re-joins as the youngest but keeps
    ///         its address, so if it holds the lower address it becomes role leader while the
    ///         singletons — and all the work they own — stay on the other node. Electing it Primary
    ///         would enable the Primary-gated data plane (inbound writes, alarm acks, the fleet-wide
    ///         alerts emit, the alarm-history drain) on the node that is not hosting the work.
    ///     </para>
    ///     <para>
    ///         Only <see cref="MemberStatus.Up"/> members are eligible, matching singleton placement:
    ///         a <c>Leaving</c> node is handing its singletons over and must not be named Primary.
    ///     </para>
    /// </remarks>
    /// <param name="members">The current cluster members, in any order.</param>
    /// <returns>The oldest Up driver member's address, or <c>null</c> when there is none.</returns>
    public static Address? SelectDriverPrimary(IEnumerable<Member> members) =>
        SelectOldestUpMemberOfRole(members, DriverRole);

    /// <summary>
    /// Selects the oldest Up member carrying <paramref name="role"/> — the node that
    /// <c>ClusterSingletonManager</c> would place a role-scoped singleton on.
    /// </summary>
    /// <param name="members">The current cluster members, in any order.</param>
    /// <param name="role">The cluster role to scope the selection to.</param>
    /// <returns>The oldest Up member's address for that role, or <c>null</c> when there is none.</returns>
    /// <remarks>
    ///     <para>
    ///         The age-ordering rationale in <see cref="SelectDriverPrimary"/> applies verbatim to any
    ///         role: oldest, never <c>RoleLeader</c>. The rule itself lives in the shared
    ///         <see cref="ClusterActiveNode"/> (<c>ZB.MOM.WW.Health.Akka</c> 0.3.0) so it has exactly
    ///         one implementation family-wide — this method is the control plane's entry point into it.
    ///     </para>
    ///     <para>
    ///         Delegating rather than re-implementing is what keeps the redundancy snapshot and the
    ///         <c>/health/active</c> tier structurally in agreement. They answer for different
    ///         consumers — the OPC UA <c>ServiceLevel</c> 250/240 split reads this election, while an
    ///         orchestrator routes admin traffic by the health tier — so a divergence would advertise
    ///         one node as authoritative while gating the data plane on another. Both apps in the
    ///         family previously kept private copies of this rule, and both got it wrong in a
    ///         different way.
    ///     </para>
    /// </remarks>
    public static Address? SelectOldestUpMemberOfRole(IEnumerable<Member> members, string role)
    {
        ArgumentNullException.ThrowIfNull(members);
        ArgumentException.ThrowIfNullOrWhiteSpace(role);

        return ClusterActiveNode.OldestUpMember(members, role)?.Address;
    }

    private IReadOnlyList<NodeRedundancyState> BuildSnapshot()
    {
        var driverPrimary = SelectDriverPrimary(_cluster.State.Members);
        var clusterLeader = _cluster.State.Leader;
        var now = DateTime.UtcNow;

        var list = new List<NodeRedundancyState>(_cluster.State.Members.Count);
        foreach (var member in _cluster.State.Members)
        {
            var host = member.Address.Host;
            if (string.IsNullOrWhiteSpace(host)) continue;

            var isPrimary = driverPrimary is not null && driverPrimary == member.Address;
            var role = member.Roles.Contains(DriverRole)
                ? (isPrimary ? CommonsRedundancyRole.Primary : CommonsRedundancyRole.Secondary)
                : CommonsRedundancyRole.Detached;

            list.Add(new NodeRedundancyState(
                ToNodeId(member.Address),
                role,
                IsClusterLeader: clusterLeader == member.Address,
                IsDriverPrimary: isPrimary,
                AsOfUtc: now));
        }
        return list;
    }

    /// <summary>
    ///     Builds the canonical cluster node id (<c>host:port</c>) for an Akka member address — the SAME
    ///     format <c>ClusterRoleInfo.LocalNode</c>/<c>ToNodeId</c> use, so redundancy-state consumers
    ///     (OPC UA ServiceLevel, scripted-alarm emit gate, historian gate) can match their local node.
    /// </summary>
    /// <param name="address">The cluster member's address.</param>
    /// <returns>The canonical <c>host:port</c> node id.</returns>
    public static NodeId ToNodeId(Akka.Actor.Address address) =>
        NodeId.Parse($"{address.Host ?? string.Empty}:{address.Port ?? 0}");

    public sealed class RecomputeNow
    {
        public static readonly RecomputeNow Instance = new();
        private RecomputeNow() { }
    }

    /// <summary>
    /// Marker message fired by the periodic heartbeat timer. Forces a re-publish of the current
    /// snapshot regardless of the change-driven <c>_dirty</c> flag, so late subscribers converge.
    /// </summary>
    public sealed class Heartbeat
    {
        /// <summary>The singleton heartbeat message instance.</summary>
        public static readonly Heartbeat Instance = new();
        private Heartbeat() { }
    }
}