lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/PeerOpcUaProbeActor.cs

using System.Net.Sockets;
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Commons.Types;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Health;

/// <summary>
/// Periodically pings a peer node's OPC UA endpoint (<c>opc.tcp://peer:4840</c>) and publishes
/// the result on the cluster's redundancy-state input topic so the admin <c>RedundancyStateActor</c>
/// can react.
///
/// The probe is a plain TCP connect to the OPC UA port with a short timeout — enough to detect
/// "is the OPC UA server process up and accepting connections?" A full secure-channel handshake
/// (Hello / Acknowledge) needs the OPC UA Client SDK and a session/PKI setup, which is more than
/// what the redundancy calculator needs. Upgrade to a real Hello probe if a deeper liveness signal
/// is ever required.
/// </summary>
public sealed class PeerOpcUaProbeActor : ReceiveActor, IWithTimers
{
    // Owned by ControlPlane.Redundancy.RedundancyStateActor; duplicated here to avoid a
    // Runtime → ControlPlane project reference. Keep both literals in lock-step.
    public const string RedundancyStateTopic = "redundancy-state";

    public static readonly TimeSpan DefaultProbeInterval = TimeSpan.FromSeconds(10);
    public static readonly TimeSpan DefaultConnectTimeout = TimeSpan.FromSeconds(2);
    public const int DefaultOpcUaPort = 4840;

    public sealed record OpcUaProbeResult(NodeId NodeId, bool Ok);
    public sealed class Tick { public static readonly Tick Instance = new(); private Tick() { } }

    private readonly NodeId _peer;
    private readonly TimeSpan _interval;
    private readonly TimeSpan _connectTimeout;
    private readonly int _opcUaPort;
    private readonly Action<object>? _broadcastOverride;
    private readonly ILoggingAdapter _log = Context.GetLogger();

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

    /// <summary>Creates actor props for the peer OPC UA probe.</summary>
    /// <param name="peer">The node identifier of the peer to probe.</param>
    /// <param name="interval">Optional probe interval; defaults to DefaultProbeInterval.</param>
    /// <param name="connectTimeout">Optional connection timeout; defaults to DefaultConnectTimeout.</param>
    /// <param name="opcUaPort">The OPC UA port to connect to; defaults to DefaultOpcUaPort.</param>
    /// <param name="broadcast">Optional custom broadcast delegate for probe results.</param>
    /// <returns>Props configured to create a PeerOpcUaProbeActor.</returns>
    public static Props Props(
        NodeId peer,
        TimeSpan? interval = null,
        TimeSpan? connectTimeout = null,
        int opcUaPort = DefaultOpcUaPort,
        Action<object>? broadcast = null) =>
        Akka.Actor.Props.Create(() => new PeerOpcUaProbeActor(
            peer,
            interval ?? DefaultProbeInterval,
            connectTimeout ?? DefaultConnectTimeout,
            opcUaPort,
            broadcast));

    /// <summary>Initializes a new instance of the PeerOpcUaProbeActor.</summary>
    /// <param name="peer">The node identifier of the peer to probe.</param>
    /// <param name="interval">The probe interval.</param>
    /// <param name="connectTimeout">The connection timeout.</param>
    /// <param name="opcUaPort">The OPC UA port to connect to.</param>
    /// <param name="broadcastOverride">Optional custom broadcast delegate for probe results.</param>
    public PeerOpcUaProbeActor(
        NodeId peer,
        TimeSpan interval,
        TimeSpan connectTimeout,
        int opcUaPort,
        Action<object>? broadcastOverride)
    {
        _peer = peer;
        _interval = interval;
        _connectTimeout = connectTimeout;
        _opcUaPort = opcUaPort;
        _broadcastOverride = broadcastOverride;

        ReceiveAsync<Tick>(_ => RunProbeAsync());
    }

    /// <inheritdoc />
    protected override void PreStart() =>
        Timers.StartPeriodicTimer("probe", Tick.Instance, _interval);

    private async Task RunProbeAsync()
    {
        var host = ExtractHost(_peer);
        var ok = await TryTcpConnectAsync(host, _opcUaPort, _connectTimeout);
        var msg = new OpcUaProbeResult(_peer, ok);
        if (_broadcastOverride is not null) _broadcastOverride(msg);
        else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(RedundancyStateTopic, msg));
        _log.Debug("PeerOpcUaProbe: pinged {Peer} ({Host}:{Port}) → ok={Ok}",
            _peer, host, _opcUaPort, ok);
    }

    /// <summary>
    /// <see cref="NodeId"/> values are derived as <c>host:port</c> (see <c>ClusterRoleInfo</c>).
    /// Strip the port suffix to get the hostname for the TCP connect.
    /// </summary>
    private static string ExtractHost(NodeId nodeId)
    {
        var s = nodeId.Value;
        var colon = s.LastIndexOf(':');
        return colon > 0 ? s[..colon] : s;
    }

    private static async Task<bool> TryTcpConnectAsync(string host, int port, TimeSpan timeout)
    {
        try
        {
            using var client = new TcpClient();
            using var cts = new CancellationTokenSource(timeout);
            await client.ConnectAsync(host, port, cts.Token);
            return client.Connected;
        }
        catch (Exception)
        {
            return false;
        }
    }
}