lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Fleet/ClusterNodeAddressReconcilerActor.cs

using Akka.Actor;
using Akka.Cluster;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Configuration;

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

/// <summary>
///     Admin-role cluster singleton that runs <see cref="ClusterNodeAddressReconciler"/> against live
///     membership and logs what it finds. A singleton rather than a per-admin-node actor so a
///     two-admin fleet reports each disagreement once.
/// </summary>
/// <remarks>
///     Recomputes on membership change (debounced) and on a slow periodic sweep, and <b>logs only
///     when the finding set changes</b>. A node legitimately down would otherwise reprint the same
///     warning every sweep until someone silences it by ignoring the log, which is the failure mode
///     this check exists to avoid.
/// </remarks>
public sealed class ClusterNodeAddressReconcilerActor : ReceiveActor, IWithTimers
{
    /// <summary>Driver role name — only members carrying it are expected to have a ClusterNode row.</summary>
    public const string DriverRole = RoleParser.Driver;

    /// <summary>Collapses a burst of membership events (a rolling restart) into one reconcile.</summary>
    public static readonly TimeSpan DebounceWindow = TimeSpan.FromSeconds(2);

    /// <summary>Periodic sweep, so a row edited in the AdminUI is checked without a topology change.</summary>
    public static readonly TimeSpan DefaultSweepInterval = TimeSpan.FromMinutes(5);

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly string? _ownClusterId;
    private readonly Akka.Cluster.Cluster _cluster;
    private readonly TimeSpan _sweepInterval;
    private readonly ILoggingAdapter _log = Context.GetLogger();
    private IReadOnlyList<ClusterNodeAddressMismatch>? _lastReported;

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

    /// <summary>Creates Props for the reconciler.</summary>
    /// <param name="dbFactory">Factory for the config database context.</param>
    /// <param name="ownClusterId">
    ///     The admin node's own <c>ClusterId</c> (e.g. <c>MAIN</c>), from
    ///     <see cref="IClusterRoleInfo.ClusterId"/>. Non-null/empty scopes the reconcile to rows in that
    ///     cluster — the only members central's own mesh can see after the Phase 6 split. Null/empty
    ///     (a legacy admin node with no cluster role) reconciles the whole fleet, unchanged.
    /// </param>
    /// <param name="sweepInterval">Periodic sweep interval; defaults to <see cref="DefaultSweepInterval"/>.</param>
    /// <returns>Props for actor creation.</returns>
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        string? ownClusterId = null,
        TimeSpan? sweepInterval = null) =>
        Akka.Actor.Props.Create(() =>
            new ClusterNodeAddressReconcilerActor(dbFactory, ownClusterId, sweepInterval));

    /// <summary>Initializes a new instance of the <see cref="ClusterNodeAddressReconcilerActor"/> class.</summary>
    /// <param name="dbFactory">Factory for the config database context.</param>
    /// <param name="ownClusterId">
    ///     The admin node's own <c>ClusterId</c>; scopes the reconcile to that cluster's rows when
    ///     non-null/empty, else reconciles the whole fleet (legacy behaviour).
    /// </param>
    /// <param name="sweepInterval">Periodic sweep interval; defaults to <see cref="DefaultSweepInterval"/>.</param>
    public ClusterNodeAddressReconcilerActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        string? ownClusterId = null,
        TimeSpan? sweepInterval = null)
    {
        _dbFactory = dbFactory;
        _ownClusterId = string.IsNullOrEmpty(ownClusterId) ? null : ownClusterId;
        _cluster = Akka.Cluster.Cluster.Get(Context.System);
        _sweepInterval = sweepInterval ?? DefaultSweepInterval;

        Receive<ClusterEvent.IMemberEvent>(_ => ScheduleReconcile());
        Receive<ClusterEvent.CurrentClusterState>(_ => ScheduleReconcile());
        Receive<ReconcileNow>(_ => Reconcile());
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        _cluster.Subscribe(Self, ClusterEvent.InitialStateAsEvents, typeof(ClusterEvent.IMemberEvent));
        Timers.StartPeriodicTimer("sweep", ReconcileNow.Instance, _sweepInterval, _sweepInterval);
    }

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

    private void ScheduleReconcile() =>
        Timers.StartSingleTimer("debounce", ReconcileNow.Instance, DebounceWindow);

    private void Reconcile()
    {
        var observed = _cluster.State.Members
            .Where(m => m.Status == MemberStatus.Up && m.Roles.Contains(DriverRole))
            .Select(m => (Host: m.Address.Host ?? string.Empty, Port: m.Address.Port ?? 0))
            .Where(a => !string.IsNullOrWhiteSpace(a.Host) && a.Port > 0)
            .ToList();

        List<ClusterNodeAddress> rows;
        List<string> enabled;
        try
        {
            using var db = _dbFactory.CreateDbContext();
            // Per-cluster mesh Phase 6: scope to the admin node's OWN cluster's rows. After the split,
            // central's mesh sees only its own pair, so a row in another cluster it cannot observe would
            // false-report EnabledRowNotInCluster forever. A legacy admin node with no cluster role
            // (_ownClusterId == null) still sees every member via gossip, so it reconciles the full fleet.
            var scoped = db.ClusterNodes.AsNoTracking();
            if (_ownClusterId is not null)
                scoped = scoped.Where(n => n.ClusterId == _ownClusterId);
            rows = scoped
                .Select(n => new ClusterNodeAddress(n.NodeId, n.Host, n.AkkaPort))
                .ToList();
            // Same predicate the coordinator's expected-ack set uses — a node the deploy path will
            // not wait for must not be reported as missing either, or the maintenance hatch trades a
            // failed deployment for a permanent warning.
            enabled = scoped
                .Where(n => n.Enabled && !n.MaintenanceMode).Select(n => n.NodeId).ToList();
        }
        catch (Exception ex)
        {
            // A consistency check must never take the admin node down with it. Central SQL being
            // briefly unreachable is an operational fact, not a reason to restart this singleton.
            _log.Warning(ex, "ClusterNode address reconcile skipped — config database unreachable");
            return;
        }

        var findings = ClusterNodeAddressReconciler.Reconcile(observed, rows, enabled);

        if (_lastReported is not null && findings.SequenceEqual(_lastReported)) return;
        _lastReported = findings;

        if (findings.Count == 0)
        {
            _log.Info("ClusterNode addresses reconcile with cluster membership ({Count} driver members)",
                observed.Count);
            return;
        }

        foreach (var f in findings)
        {
            // Error for the two shapes that are always a misconfiguration; Warning for a row whose
            // node is merely absent, which is a legitimate state during maintenance.
            if (f.Kind == AddressMismatchKind.EnabledRowNotInCluster)
                _log.Warning("ClusterNode address check [{Kind}] {NodeId}: {Detail}", f.Kind, f.NodeId, f.Detail);
            else
                _log.Error("ClusterNode address check [{Kind}] {NodeId}: {Detail}", f.Kind, f.NodeId, f.Detail);
        }
    }

    /// <summary>Self-message: run a reconcile pass now.</summary>
    public sealed class ReconcileNow
    {
        /// <summary>The singleton instance.</summary>
        public static readonly ReconcileNow Instance = new();
        private ReconcileNow() { }
    }
}