lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs

using System.Collections.Concurrent;
using System.Threading.Channels;

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

/// <summary>
///     Default <see cref="ITelemetryLocalHub"/>: a process-wide singleton on driver-role nodes.
/// </summary>
/// <remarks>
///     <para>
///         <b>No DPS.</b> This hub is fed only by this node's own in-process producers (a later
///         task taps them). It never subscribes to cluster DistributedPubSub — see the invariant on
///         <see cref="ITelemetryLocalHub"/>.
///     </para>
///     <para>
///         <b>Snapshot-then-attach ordering (no lost delta).</b> <see cref="Emit"/> updates the snapshot
///         cache and fans out under a single gate; <see cref="Subscribe"/> writes the cached snapshots
///         into the new channel and only THEN adds it to the subscriber set — all under that same gate.
///         Because both operations are serialized by the gate, an <see cref="Emit"/> racing a
///         <see cref="Subscribe"/> is resolved one of two ways, both correct: it runs entirely before the
///         subscribe (its value is in the snapshot, and it is not delivered live because the channel is
///         not yet attached), or entirely after (it is delivered live, strictly after the snapshot). No
///         item is dropped or duplicated across the boundary. Fan-out is a non-blocking
///         <see cref="ChannelWriter{T}.TryWrite"/> into bounded/DropOldest channels, so holding the gate
///         never blocks on a slow consumer.
///     </para>
/// </remarks>
public sealed class TelemetryLocalHub : ITelemetryLocalHub
{
    /// <summary>
    ///     How long an eviction tombstone (see <see cref="_evicted"/>) is kept. It only has to outlive the
    ///     removed child's last late write — an in-flight capability call that completes after
    ///     <c>StopChild</c>, bounded by the longest capability timeout (seconds) plus the child's own stop —
    ///     so ten minutes is a wide margin. Without a sweep a driver removed for good kept its id forever.
    /// </summary>
    public static readonly TimeSpan EvictionTombstoneTtl = TimeSpan.FromMinutes(10);

    private readonly object _gate = new();
    private readonly TimeProvider _time;

    /// <summary>Initializes a hub on the system clock.</summary>
    public TelemetryLocalHub() : this(TimeProvider.System) { }

    /// <summary>Initializes a hub on <paramref name="time"/> — the tombstone-sweep test seam.</summary>
    /// <param name="time">The clock the eviction tombstones are stamped and swept by.</param>
    public TelemetryLocalHub(TimeProvider time)
    {
        ArgumentNullException.ThrowIfNull(time);
        _time = time;
    }

    // Plain Dictionary, NOT ConcurrentDictionary: every access is already under _gate, and
    // ConcurrentDictionary.Values would materialize a fresh List (and take its own internal lock) on
    // every Emit — the hot node-actor-thread path. Under the gate a plain Dictionary is safe and
    // allocation-free to iterate.
    private readonly Dictionary<Guid, Channel<TelemetryItem>> _subscribers = new();

    // Snapshot last-value caches. Only ever mutated under _gate (kept ConcurrentDictionary so a
    // Subscribe reading them under the gate is trivially safe even against any future lock-free read).
    // Entries are evicted via Evict when DriverHostActor removes the instance, so a
    // decommissioned driver's last-known state does not replay to new subscribers.
    private readonly ConcurrentDictionary<string, TelemetryItem.Health> _healthCache = new();
    private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new();

    // Instances evicted and not since readmitted, stamped with the eviction time. Their late Health/Resilience
    // items are dropped, so a removed driver's still-stopping child cannot re-prime the cache it was just
    // evicted from. Swept past EvictionTombstoneTtl on the next Evict (never on the Emit hot path). Under _gate.
    private readonly Dictionary<string, DateTimeOffset> _evicted = new(StringComparer.Ordinal);

    /// <inheritdoc />
    public void Emit(TelemetryItem item)
    {
        ArgumentNullException.ThrowIfNull(item);

        lock (_gate)
        {
            switch (item)
            {
                case TelemetryItem.Health h when _evicted.ContainsKey(h.E.DriverInstanceId):
                    return;
                case TelemetryItem.Resilience r when _evicted.ContainsKey(r.E.DriverInstanceId):
                    return;
                case TelemetryItem.Health h:
                    _healthCache[h.E.DriverInstanceId] = h;
                    break;
                case TelemetryItem.Resilience r:
                    _resilienceCache[(r.E.DriverInstanceId, r.E.HostName)] = r;
                    break;
                // Alarm / Script are append-style — never cached.
            }

            foreach (var channel in _subscribers.Values)
                channel.Writer.TryWrite(item); // bounded + DropOldest ⇒ non-blocking, sheds oldest when full
        }
    }

    /// <inheritdoc />
    public void Evict(string driverInstanceId)
    {
        ArgumentNullException.ThrowIfNull(driverInstanceId);

        lock (_gate)
        {
            var now = _time.GetUtcNow();
            SweepExpiredTombstones(now);
            _evicted[driverInstanceId] = now;
            _healthCache.TryRemove(driverInstanceId, out _);
            foreach (var key in _resilienceCache.Keys)
                if (string.Equals(key.InstanceId, driverInstanceId, StringComparison.Ordinal))
                    _resilienceCache.TryRemove(key, out _);
        }
    }

    /// <inheritdoc />
    public void Readmit(string driverInstanceId)
    {
        ArgumentNullException.ThrowIfNull(driverInstanceId);

        lock (_gate)
            _evicted.Remove(driverInstanceId);
    }

    /// <summary>Drops tombstones older than <see cref="EvictionTombstoneTtl"/>. Called under <see cref="_gate"/>
    /// from <see cref="Evict"/> only — the set is bounded by distinct removed ids, so a linear pass per
    /// eviction is cheap, and Emit never pays for it.</summary>
    private void SweepExpiredTombstones(DateTimeOffset now)
    {
        if (_evicted.Count == 0) return;
        List<string>? expired = null;
        foreach (var (id, evictedAt) in _evicted)
        {
            if (now - evictedAt >= EvictionTombstoneTtl)
                (expired ??= new List<string>()).Add(id);
        }

        if (expired is null) return;
        foreach (var id in expired)
            _evicted.Remove(id);
    }

    /// <inheritdoc />
    public ITelemetrySubscription Subscribe(int boundedCapacity)
    {
        if (boundedCapacity < 1)
            throw new ArgumentOutOfRangeException(nameof(boundedCapacity), boundedCapacity, "Capacity must be positive.");

        var channel = Channel.CreateBounded<TelemetryItem>(new BoundedChannelOptions(boundedCapacity)
        {
            FullMode = BoundedChannelFullMode.DropOldest,
            SingleReader = true,
            SingleWriter = false,
        });

        var id = Guid.NewGuid();
        lock (_gate)
        {
            // Snapshot FIRST (Health then Resilience), attach SECOND — both under the gate so a
            // concurrent Emit cannot slip a delta between the snapshot read and the attach.
            foreach (var health in _healthCache.Values)
                channel.Writer.TryWrite(health);
            foreach (var resilience in _resilienceCache.Values)
                channel.Writer.TryWrite(resilience);

            _subscribers[id] = channel;
        }

        return new Subscription(this, id, channel);
    }

    private void Detach(Guid id)
    {
        lock (_gate)
        {
            if (_subscribers.Remove(id, out var channel))
                channel.Writer.TryComplete();
        }
    }

    private sealed class Subscription : ITelemetrySubscription
    {
        private readonly TelemetryLocalHub _hub;
        private readonly Guid _id;
        private int _disposed;

        /// <summary>Creates a subscription over a hub channel.</summary>
        /// <param name="hub">The owning hub, notified on dispose.</param>
        /// <param name="id">The subscription id.</param>
        /// <param name="channel">The channel whose reader is exposed.</param>
        public Subscription(TelemetryLocalHub hub, Guid id, Channel<TelemetryItem> channel)
        {
            _hub = hub;
            _id = id;
            Reader = channel.Reader;
        }

        /// <inheritdoc />
        public ChannelReader<TelemetryItem> Reader { get; }

        /// <summary>Detaches this subscription from the hub; safe to call more than once.</summary>
        public void Dispose()
        {
            if (Interlocked.Exchange(ref _disposed, 1) != 0)
                return; // idempotent
            _hub.Detach(_id);
        }
    }
}