lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs

using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health;

/// <summary>
///     Subscribes the <c>ScanState</c> attribute of every <c>$WinPlatform</c> /
///     <c>$AppEngine</c> object the discoverer surfaced and translates ScanState
///     value-changes into per-host <see cref="HostConnectivityStatus"/> updates.
///     Ports the state machine in
///     <c>Driver.Galaxy.Host/Backend/Stability/GalaxyRuntimeProbeManager.cs</c> onto the
///     gateway subscription path.
/// </summary>
/// <remarks>
///     Address grammar: each platform tag's probe address is
///     <c>{platformTagName}.ScanState</c>. The watcher subscribes that address through
///     <see cref="IGalaxySubscriber"/>; the EventPump (PR 4.4) routes inbound
///     OnDataChange events back via <see cref="OnProbeValueChanged"/>. State decoding:
///     <list type="bullet">
///       <item>Quality &lt; <c>192</c> (Good) → <see cref="HostState.Unknown"/>.</item>
///       <item>Value <c>1</c>, <c>true</c>, or "Running" → <see cref="HostState.Running"/>.</item>
///       <item>Value <c>0</c>, <c>false</c>, or "Stopped" → <see cref="HostState.Stopped"/>.</item>
///       <item>Anything else with Good quality → <see cref="HostState.Faulted"/>.</item>
///     </list>
///     <see cref="SyncPlatformsAsync"/> is idempotent — call it after every
///     Discover / Rediscover. Newly-added platforms are subscribed; removed ones are
///     unsubscribed and dropped from the aggregator.
/// </remarks>
public sealed class PerPlatformProbeWatcher : IDisposable
{
    public const string ProbeSuffix = ".ScanState";

    private readonly IGalaxySubscriber _subscriber;
    private readonly HostStatusAggregator _aggregator;
    private readonly ILogger _logger;
    private readonly int _bufferedUpdateIntervalMs;

    // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe;
    // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call.
    private readonly ConcurrentDictionary<string, int> _itemHandlesByPlatform =
        new(StringComparer.OrdinalIgnoreCase);
    private readonly Lock _syncLock = new();
    private bool _disposed;

    /// <summary>Initializes a new instance of the PerPlatformProbeWatcher class.</summary>
    /// <param name="subscriber">The Galaxy subscriber for managing probe subscriptions.</param>
    /// <param name="aggregator">The host status aggregator for tracking platform connectivity.</param>
    /// <param name="logger">Optional logger for diagnostic messages.</param>
    /// <param name="bufferedUpdateIntervalMs">Buffered update interval in milliseconds; must be >= 0.</param>
    public PerPlatformProbeWatcher(
        IGalaxySubscriber subscriber,
        HostStatusAggregator aggregator,
        ILogger? logger = null,
        int bufferedUpdateIntervalMs = 0)
    {
        _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
        _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator));
        _logger = logger ?? NullLogger.Instance;
        if (bufferedUpdateIntervalMs < 0)
        {
            throw new ArgumentOutOfRangeException(nameof(bufferedUpdateIntervalMs),
                "bufferedUpdateIntervalMs must be >= 0; 0 means use the gw's default cadence.");
        }
        _bufferedUpdateIntervalMs = bufferedUpdateIntervalMs;
    }

    /// <summary>Snapshot of platform tag names currently watched.</summary>
    public IReadOnlyCollection<string> WatchedPlatforms => [.. _itemHandlesByPlatform.Keys];

    /// <summary>
    ///     Reconcile the watched platform set against <paramref name="platformTagNames"/>.
    ///     Subscribes new entries, unsubscribes dropped ones. Calling with the same set is
    ///     a no-op.
    /// </summary>
    /// <param name="platformTagNames">The platform tag names to synchronize.</param>
    /// <param name="cancellationToken">A cancellation token to cancel the operation.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task SyncPlatformsAsync(
        IEnumerable<string> platformTagNames, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(platformTagNames);

        var desired = new HashSet<string>(platformTagNames, StringComparer.OrdinalIgnoreCase);

        // Compute deltas under the lock so concurrent SyncPlatformsAsync calls don't
        // race on the membership view.
        List<string> toAdd;
        List<(string Platform, int ItemHandle)> toRemove;
        lock (_syncLock)
        {
            toAdd = [.. desired.Where(p => !_itemHandlesByPlatform.ContainsKey(p))];
            toRemove = [.. _itemHandlesByPlatform
                .Where(kvp => !desired.Contains(kvp.Key) && kvp.Value > 0)
                .Select(kvp => (kvp.Key, kvp.Value))];

            // Drop removed entries from the membership map up-front so a concurrent
            // OnProbeValueChanged for them is silently ignored. The unsubscribe RPC
            // runs outside the lock.
            foreach (var (platform, _) in toRemove)
            {
                _itemHandlesByPlatform.TryRemove(platform, out _);
                _aggregator.Remove(platform);
            }
        }

        if (toRemove.Count > 0)
        {
            try
            {
                await _subscriber.UnsubscribeBulkAsync(
                    [.. toRemove.Select(t => t.ItemHandle)], cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "PerPlatformProbeWatcher unsubscribe failed for {Count} probe(s); aggregator entries already cleared.",
                    toRemove.Count);
            }
        }

        if (toAdd.Count == 0) return;

        var probeAddresses = toAdd.Select(p => p + ProbeSuffix).ToArray();
        // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
        // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
        // changes are rare so a coarser interval is usually fine; deployments that need
        // tighter health visibility can dial it down through GalaxyDriverOptions.
        var results = await _subscriber.SubscribeBulkAsync(
            probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false);

        for (var i = 0; i < toAdd.Count; i++)
        {
            var platform = toAdd[i];
            var match = results.FirstOrDefault(r => string.Equals(
                r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase));

            var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0;
            _itemHandlesByPlatform[platform] = itemHandle;

            if (itemHandle <= 0)
            {
                _logger.LogWarning(
                    "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}",
                    platform, match?.ErrorMessage ?? "<no result returned>");
            }
        }
    }

    /// <summary>
    ///     Route an OnDataChange for a probe address into the aggregator. The EventPump
    ///     (PR 4.4) calls this; tests can drive it directly to exercise the state machine
    ///     without spinning a real gw. Foreign references (anything not ending in
    ///     <see cref="ProbeSuffix"/>, or a probe for a platform we're not tracking) are
    ///     silently dropped.
    /// </summary>
    /// <param name="fullReference">The full reference path of the probe attribute.</param>
    /// <param name="value">The probe value to decode.</param>
    /// <param name="qualityByte">The quality byte for the value.</param>
    public void OnProbeValueChanged(string fullReference, object? value, byte qualityByte)
    {
        if (_disposed) return;
        ArgumentNullException.ThrowIfNull(fullReference);

        if (!fullReference.EndsWith(ProbeSuffix, StringComparison.OrdinalIgnoreCase)) return;
        var platform = fullReference[..^ProbeSuffix.Length];
        if (!_itemHandlesByPlatform.ContainsKey(platform)) return;

        var state = DecodeState(value, qualityByte);
        _aggregator.Update(new HostConnectivityStatus(platform, state, DateTime.UtcNow));
    }

    /// <summary>
    ///     Decode a ScanState value + raw quality byte to a <see cref="HostState"/>.
    ///     Public for tests that want to pin the decoding table.
    /// </summary>
    /// <param name="value">The probe value to decode.</param>
    /// <param name="qualityByte">The quality byte for the value.</param>
    /// <returns>The decoded host state.</returns>
    public static HostState DecodeState(object? value, byte qualityByte)
    {
        if (qualityByte < 192) return HostState.Unknown;

        return value switch
        {
            bool b => b ? HostState.Running : HostState.Stopped,
            int i => i == 1 ? HostState.Running : i == 0 ? HostState.Stopped : HostState.Faulted,
            short s => s == 1 ? HostState.Running : s == 0 ? HostState.Stopped : HostState.Faulted,
            long l => l == 1 ? HostState.Running : l == 0 ? HostState.Stopped : HostState.Faulted,
            string str when string.Equals(str, "Running", StringComparison.OrdinalIgnoreCase) => HostState.Running,
            string str when string.Equals(str, "Stopped", StringComparison.OrdinalIgnoreCase) => HostState.Stopped,
            _ => HostState.Faulted,
        };
    }

    /// <summary>Disposes the probe watcher and unsubscribes all tracked platforms.</summary>
    public void Dispose()
    {
        if (_disposed) return;
        _disposed = true;

        // Best-effort unsubscribe everything we know about. Run synchronously through
        // GetAwaiter().GetResult() since Dispose is sync; transport errors are swallowed.
        var liveHandles = _itemHandlesByPlatform.Values.Where(h => h > 0).ToArray();
        _itemHandlesByPlatform.Clear();
        if (liveHandles.Length > 0)
        {
            try { _subscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None).GetAwaiter().GetResult(); }
            catch (Exception ex) { _logger.LogWarning(ex, "PerPlatformProbeWatcher dispose unsubscribe failed"); }
        }
    }
}