using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
namespace ZB.MOM.WW.OtOpcUa.Server.Redundancy;
///
/// Orchestrates Phase 6.3 Stream C: feeds the with the
/// current (topology, peer reachability, apply-in-progress, recovery dwell, self health)
/// inputs and emits the resulting + labelled
/// to subscribers. The OPC UA ServiceLevel variable node consumes this via
/// on every tick.
///
///
/// Pure orchestration — no background timer, no OPC UA stack dep. The caller (a
/// HostedService in a future PR, or a test) drives at
/// whatever cadence is appropriate. Each call reads the inputs + recomputes the ServiceLevel
/// byte; state is fired on the event when the byte differs from
/// the last emitted value (edge-triggered). The event
/// fires whenever the topology's ServerUriArray content changes.
///
public sealed class RedundancyStatePublisher
{
private readonly RedundancyCoordinator _coordinator;
private readonly ApplyLeaseRegistry _leases;
private readonly RecoveryStateManager _recovery;
private readonly PeerReachabilityTracker _peers;
private readonly Func _selfHealthy;
private readonly Func _operatorMaintenance;
private byte _lastByte = 255; // start at Authoritative — harmless before first tick
private IReadOnlyList? _lastServerUriArray;
public RedundancyStatePublisher(
RedundancyCoordinator coordinator,
ApplyLeaseRegistry leases,
RecoveryStateManager recovery,
PeerReachabilityTracker peers,
Func? selfHealthy = null,
Func? operatorMaintenance = null)
{
ArgumentNullException.ThrowIfNull(coordinator);
ArgumentNullException.ThrowIfNull(leases);
ArgumentNullException.ThrowIfNull(recovery);
ArgumentNullException.ThrowIfNull(peers);
_coordinator = coordinator;
_leases = leases;
_recovery = recovery;
_peers = peers;
_selfHealthy = selfHealthy ?? (() => true);
_operatorMaintenance = operatorMaintenance ?? (() => false);
}
///
/// Fires with the current ServiceLevel byte + band on every call to
/// when the byte differs from the previously-emitted one.
///
public event Action? OnStateChanged;
///
/// Fires when the cluster's ServerUriArray (self + peers) content changes — e.g. an
/// operator adds or removes a peer. Consumer is the OPC UA ServerUriArray
/// variable node in Stream C.2.
///
public event Action>? OnServerUriArrayChanged;
/// Snapshot of the last-published ServiceLevel byte — diagnostics + tests.
public byte LastByte => _lastByte;
///
/// Compute the current ServiceLevel + emit change events if anything moved. Caller
/// drives cadence — a 1 s tick in production is reasonable; tests drive it directly.
///
public ServiceLevelSnapshot ComputeAndPublish()
{
var topology = _coordinator.Current;
if (topology is null)
{
// Not yet initialized — surface NoData so clients don't treat us as authoritative.
return Emit((byte)ServiceLevelBand.NoData, null);
}
// Aggregate peer reachability. For 2-node v2.0 clusters there is at most one peer;
// treat "all peers healthy" as the boolean input to the calculator.
var peerReachable = topology.Peers.All(p => _peers.Get(p.NodeId).BothHealthy);
var peerUaHealthy = topology.Peers.All(p => _peers.Get(p.NodeId).UaHealthy);
var peerHttpHealthy = topology.Peers.All(p => _peers.Get(p.NodeId).HttpHealthy);
var role = MapRole(topology.SelfRole);
var value = ServiceLevelCalculator.Compute(
role: role,
selfHealthy: _selfHealthy(),
peerUaHealthy: peerUaHealthy,
peerHttpHealthy: peerHttpHealthy,
applyInProgress: _leases.IsApplyInProgress,
recoveryDwellMet: _recovery.IsDwellMet(),
topologyValid: _coordinator.IsTopologyValid,
operatorMaintenance: _operatorMaintenance());
MaybeFireServerUriArray(topology);
return Emit(value, topology);
}
private static RedundancyRole MapRole(RedundancyRole role) => role switch
{
// Standalone is serving; treat as Primary for the matrix since the calculator
// already special-cases Standalone inside its Compute.
RedundancyRole.Primary => RedundancyRole.Primary,
RedundancyRole.Secondary => RedundancyRole.Secondary,
_ => RedundancyRole.Standalone,
};
private ServiceLevelSnapshot Emit(byte value, RedundancyTopology? topology)
{
var snap = new ServiceLevelSnapshot(
Value: value,
Band: ServiceLevelCalculator.Classify(value),
Topology: topology);
if (value != _lastByte)
{
_lastByte = value;
OnStateChanged?.Invoke(snap);
}
return snap;
}
private void MaybeFireServerUriArray(RedundancyTopology topology)
{
var current = topology.ServerUriArray();
if (_lastServerUriArray is null || !current.SequenceEqual(_lastServerUriArray, StringComparer.Ordinal))
{
_lastServerUriArray = current;
OnServerUriArrayChanged?.Invoke(current);
}
}
}
/// Per-tick output of .
public sealed record ServiceLevelSnapshot(
byte Value,
ServiceLevelBand Band,
RedundancyTopology? Topology);