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.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);
    }

    private IReadOnlyList<NodeRedundancyState> BuildSnapshot()
    {
        var driverLeader = _cluster.State.RoleLeader("driver");
        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 role = member.Roles.Contains("driver")
                ? (driverLeader == member.Address ? CommonsRedundancyRole.Primary : CommonsRedundancyRole.Secondary)
                : CommonsRedundancyRole.Detached;

            list.Add(new NodeRedundancyState(
                ToNodeId(member.Address),
                role,
                IsClusterLeader: clusterLeader == member.Address,
                IsRoleLeaderForDriver: driverLeader == member.Address,
                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() { }
    }
}