mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs

using System.Threading.Channels;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;

namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;

/// <summary>
/// Broadcasts MxEvents to the two dashboard audiences for a session: remote
/// <see cref="EventsHub"/> clients subscribed to the session's group, and
/// in-process subscribers opened through
/// <see cref="IDashboardSessionEventSubscriber.Subscribe"/>. <see cref="Publish"/>
/// never blocks: both audiences are fed through bounded drop-oldest queues, so the
/// source gRPC stream is never held up by a slow dashboard viewer. Errors are
/// logged and dropped — keeping the SignalR mirror best-effort preserves the gRPC
/// contract.
/// </summary>
/// <remarks>
/// When <c>MxGateway:Dashboard:ShowTagValues</c> is false (the default), tag
/// values are stripped from a redacted copy of the event before it reaches any
/// dashboard client. The source <see cref="MxEvent"/> is shared with the gRPC
/// event path and the reconnect replay ring, so it is never mutated in place —
/// the redaction is applied to a deep clone. This is the second of two
/// independent layers: <see cref="IDashboardSessionAcl"/> decides at the
/// subscribe seam <em>which</em> sessions a caller may observe at all (see
/// <see cref="EventsHub"/>), while the redaction decides what a permitted
/// subscriber sees — so the value-leak seam stays closed whatever the ACL
/// admits.
/// <para>
/// Hub delivery goes through one <see cref="HubPump"/> per watched session: a
/// <see cref="HubQueueCapacity"/>-deep drop-oldest channel drained by a single task
/// that sends one event at a time, each send time-bounded. An earlier version fired
/// a group send per event and returned, so a slow remote viewer accumulated an
/// unbounded number of pending sends — each holding a cloned event — inside the
/// SignalR lifetime manager. The pump is started by the first event published while
/// a hub connection watches the session and torn down when
/// <see cref="EventsHubViewerRegistry.HubViewersReleased"/> reports the last one gone.
/// </para>
/// </remarks>
public sealed class DashboardEventBroadcaster : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber, IDisposable
{
    /// <summary>
    /// Queue depth per in-process subscriber. The consumer is a Blazor page
    /// rendering the newest handful of events, so a burst it cannot keep up with
    /// is dropped oldest-first rather than allowed to grow — same best-effort
    /// contract the SignalR mirror has.
    /// </summary>
    private const int InProcessQueueCapacity = 256;

    /// <summary>
    /// Queue depth of a session's hub-delivery pump. Same depth and drop policy as
    /// the in-process feed: the remote viewer is a live view, so a burst it cannot
    /// keep up with loses its oldest events rather than buffering without bound.
    /// </summary>
    internal const int HubQueueCapacity = 256;

    private static readonly TimeSpan DefaultHubSendTimeout = TimeSpan.FromSeconds(5);

    private readonly IHubContext<EventsHub> _hubContext;

    private readonly EventsHubViewerRegistry _viewerRegistry;

    private readonly ILogger<DashboardEventBroadcaster> _logger;

    private readonly TimeSpan _hubSendTimeout;

    private readonly bool _showTagValues;

    private readonly object _syncRoot = new();

    /// <summary>
    /// In-process subscribers per session. Values are treated as immutable once
    /// stored: a subscribe or dispose swaps in a new array under
    /// <see cref="_syncRoot"/>, so <see cref="Publish"/> can grab the reference
    /// and write to it after releasing the lock.
    /// </summary>
    private readonly Dictionary<string, InProcessSubscription[]> _inProcessSubscribers =
        new(StringComparer.Ordinal);

    /// <summary>Live hub-delivery pumps per session. Guarded by <see cref="_syncRoot"/>.</summary>
    private readonly Dictionary<string, HubPump> _hubPumps = new(StringComparer.Ordinal);

    /// <summary>
    /// Total live in-process subscribers, read without the lock so the common
    /// case — nobody has a session-details page open — never contends on it.
    /// Written only under <see cref="_syncRoot"/>.
    /// </summary>
    private int _inProcessSubscriberCount;

    private bool _disposed;

    /// <summary>Initializes a new instance of the <see cref="DashboardEventBroadcaster"/> class.</summary>
    /// <param name="hubContext">Hub context used to send to the session's group.</param>
    /// <param name="viewerRegistry">
    /// Live-subscriber registry consulted before any per-event work is done. Both
    /// audiences register here — hub connections by their SignalR connection id,
    /// in-process subscriptions by a synthetic one — so the gate stays a single
    /// source of truth.
    /// </param>
    /// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
    /// <param name="logger">Logger for best-effort mirror failures.</param>
    public DashboardEventBroadcaster(
        IHubContext<EventsHub> hubContext,
        EventsHubViewerRegistry viewerRegistry,
        IOptions<GatewayOptions> options,
        ILogger<DashboardEventBroadcaster> logger)
        : this(hubContext, viewerRegistry, options, logger, DefaultHubSendTimeout)
    {
    }

    /// <summary>Initializes a new instance with a custom hub send timeout (test hook).</summary>
    /// <param name="hubContext">Hub context used to send to the session's group.</param>
    /// <param name="viewerRegistry">Live-subscriber registry.</param>
    /// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
    /// <param name="logger">Logger for best-effort mirror failures.</param>
    /// <param name="hubSendTimeout">Upper bound on one group send before it is abandoned.</param>
    internal DashboardEventBroadcaster(
        IHubContext<EventsHub> hubContext,
        EventsHubViewerRegistry viewerRegistry,
        IOptions<GatewayOptions> options,
        ILogger<DashboardEventBroadcaster> logger,
        TimeSpan hubSendTimeout)
    {
        ArgumentNullException.ThrowIfNull(options);
        _hubContext = hubContext ?? throw new ArgumentNullException(nameof(hubContext));
        _viewerRegistry = viewerRegistry ?? throw new ArgumentNullException(nameof(viewerRegistry));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _hubSendTimeout = hubSendTimeout;
        _showTagValues = options.Value.Dashboard.ShowTagValues;
        _viewerRegistry.HubViewersReleased += OnHubViewersReleased;
    }

    /// <summary>Gets the number of live hub-delivery pumps (one per hub-watched session).</summary>
    internal int ActiveHubPumpCount
    {
        get
        {
            lock (_syncRoot)
            {
                return _hubPumps.Count;
            }
        }
    }

    /// <inheritdoc />
    public void Publish(string sessionId, MxEvent mxEvent)
    {
        if (string.IsNullOrEmpty(sessionId) || mxEvent is null)
        {
            return;
        }

        // Every session's dashboard-mirror subscriber calls Publish for every event,
        // whether or not a browser is on that session's page. Without this gate the
        // steady state — no dashboard viewer at all — still paid a deep protobuf
        // clone (redaction is on by default) per event. Bail before it.
        if (!_viewerRegistry.HasViewers(sessionId))
        {
            return;
        }

        MxEvent outbound = _showTagValues ? mxEvent : RedactValues(mxEvent);

        DeliverInProcess(sessionId, outbound);

        // In-process-only viewers need no hub send at all; the pump exists only
        // while a remote connection is in the session's group.
        if (_viewerRegistry.HasHubViewers(sessionId))
        {
            EnqueueForHub(sessionId, outbound);
        }
    }

    /// <summary>
    /// Stops every hub pump and detaches from the viewer registry. The singleton is
    /// disposed with the host; a pump's in-flight send is bounded by its timeout.
    /// </summary>
    public void Dispose()
    {
        HubPump[] pumps;
        lock (_syncRoot)
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;
            pumps = [.. _hubPumps.Values];
            _hubPumps.Clear();
        }

        _viewerRegistry.HubViewersReleased -= OnHubViewersReleased;
        foreach (HubPump pump in pumps)
        {
            pump.Complete();
        }
    }

    /// <summary>
    /// Queues an event on the session's hub pump, starting the pump on first use.
    /// The write happens under the lock so it cannot land on a pump that a
    /// concurrent teardown has already completed.
    /// </summary>
    /// <param name="sessionId">Session the event belongs to.</param>
    /// <param name="outbound">The event as the dashboard should see it.</param>
    private void EnqueueForHub(string sessionId, MxEvent outbound)
    {
        lock (_syncRoot)
        {
            // Re-check under the lock: the last viewer may have left between the
            // caller's unlocked check and here, and its release handler (which takes
            // this lock) would then never see a pump created after it ran.
            if (_disposed || !_viewerRegistry.HasHubViewers(sessionId))
            {
                return;
            }

            if (!_hubPumps.TryGetValue(sessionId, out HubPump? pump))
            {
                pump = new HubPump(this, sessionId, HubQueueCapacity);
                _hubPumps[sessionId] = pump;
            }

            pump.TryWrite(outbound);
        }
    }

    /// <summary>
    /// Tears down a session's hub pump once its last hub viewer has gone. Re-checks
    /// the registry under the lock: a viewer that re-subscribed in between keeps the
    /// pump, and one that arrives after the teardown gets a fresh pump on the next
    /// event.
    /// </summary>
    /// <param name="sessionId">Session whose last hub viewer left.</param>
    private void OnHubViewersReleased(string sessionId)
    {
        HubPump? pump;
        lock (_syncRoot)
        {
            if (_viewerRegistry.HasHubViewers(sessionId) || !_hubPumps.Remove(sessionId, out pump))
            {
                return;
            }
        }

        pump.Complete();
    }

    /// <summary>
    /// Sends one event to the session's group, bounded by the hub send timeout.
    /// Never throws: a failed or abandoned send is logged and the pump moves on.
    /// </summary>
    /// <param name="sessionId">Session whose group receives the event.</param>
    /// <param name="outbound">The event to send.</param>
    /// <returns>A task that completes when the send finished or was abandoned.</returns>
    private async Task SendToHubAsync(string sessionId, MxEvent outbound)
    {
        try
        {
            bool sent = await BoundedHubSend.SendAsync(
                    _hubContext.Clients.Group(EventsHub.GroupName(sessionId)),
                    EventsHub.EventMessage,
                    outbound,
                    _hubSendTimeout,
                    CancellationToken.None)
                .ConfigureAwait(false);
            if (!sent)
            {
                _logger.LogDebug(
                    "Dashboard event mirror to session {SessionId} timed out after {Timeout}; continuing.",
                    sessionId,
                    _hubSendTimeout);
            }
        }
        catch (Exception ex)
        {
            _logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId);
        }
    }

    /// <inheritdoc />
    public IDashboardEventSubscription Subscribe(string sessionId)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(sessionId);

        // A synthetic connection id keeps the registry's per-connection bookkeeping
        // usable for a subscriber that has no SignalR connection behind it. The
        // "inproc-" prefix cannot collide with a SignalR connection id and makes the
        // origin obvious in a debugger.
        string connectionId = EventsHubViewerRegistry.InProcessConnectionIdPrefix + Guid.NewGuid().ToString("N");
        InProcessSubscription subscription = new(this, sessionId, connectionId, InProcessQueueCapacity);

        // Register before the subscriber becomes a delivery target, exactly as
        // EventsHub.SubscribeSession registers before joining the group: the reverse
        // order would leave a window in which this subscriber is a delivery target but
        // Publish's gate still reports the session unwatched, silently dropping events
        // it should receive. The cost of this order is at worst a redaction clone that
        // reaches nobody for the width of the window.
        _viewerRegistry.AddViewer(connectionId, sessionId);

        lock (_syncRoot)
        {
            _inProcessSubscribers[sessionId] =
                _inProcessSubscribers.TryGetValue(sessionId, out InProcessSubscription[]? existing)
                    ? [.. existing, subscription]
                    : [subscription];

            Volatile.Write(ref _inProcessSubscriberCount, _inProcessSubscriberCount + 1);
        }

        return subscription;
    }

    /// <summary>
    /// Hands the already-redacted event to every in-process subscriber of the
    /// session. Writes are non-blocking and lossy by construction, so this never
    /// stalls the caller's event pipeline.
    /// </summary>
    /// <param name="sessionId">Session the event belongs to.</param>
    /// <param name="outbound">The event as the dashboard should see it.</param>
    private void DeliverInProcess(string sessionId, MxEvent outbound)
    {
        // The gate above admits hub-only viewers too, so check for in-process
        // subscribers before touching the lock at all.
        if (Volatile.Read(ref _inProcessSubscriberCount) == 0)
        {
            return;
        }

        InProcessSubscription[] subscribers;
        lock (_syncRoot)
        {
            if (!_inProcessSubscribers.TryGetValue(sessionId, out InProcessSubscription[]? found))
            {
                return;
            }

            subscribers = found;
        }

        // The array is never mutated in place, so the writes happen outside the lock.
        foreach (InProcessSubscription subscriber in subscribers)
        {
            subscriber.TryWrite(outbound);
        }
    }

    /// <summary>
    /// Removes a disposed subscription from the delivery map and releases its
    /// viewer registration. Called at most once per subscription.
    /// </summary>
    /// <param name="subscription">The subscription being disposed.</param>
    private void Unsubscribe(InProcessSubscription subscription)
    {
        // Drop the delivery target first and deregister after, mirroring
        // EventsHub.UnsubscribeSession: the mirror stays enabled for the brief overlap
        // rather than dropping events still owed to the session's other subscribers.
        lock (_syncRoot)
        {
            if (_inProcessSubscribers.TryGetValue(subscription.SessionId, out InProcessSubscription[]? existing))
            {
                InProcessSubscription[] remaining =
                    [.. existing.Where(candidate => !ReferenceEquals(candidate, subscription))];

                // Equal lengths mean it was never in this bucket, so the counter it
                // would decrement is not its own to release.
                if (remaining.Length != existing.Length)
                {
                    if (remaining.Length == 0)
                    {
                        // Drop the key so the map does not grow one entry per session ever viewed.
                        _inProcessSubscribers.Remove(subscription.SessionId);
                    }
                    else
                    {
                        _inProcessSubscribers[subscription.SessionId] = remaining;
                    }

                    Volatile.Write(ref _inProcessSubscriberCount, _inProcessSubscriberCount - 1);
                }
            }
        }

        _viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId);

        // The synthetic connection id is used once and never reconnects, so nothing
        // else will ever call ReleaseConnection for it; without this the registry
        // would retain an empty per-connection entry per subscription ever opened.
        _viewerRegistry.ReleaseConnection(subscription.ConnectionId);
    }

    /// <summary>
    /// Produces a deep clone of <paramref name="source"/> with every tag-value
    /// field cleared, leaving tag reference, quality, status, and timestamps
    /// intact so the dashboard still renders the event without the value. The
    /// source event is left untouched because it is shared downstream with the
    /// gRPC stream and the replay ring.
    /// </summary>
    /// <param name="source">The source event to redact a copy of.</param>
    /// <returns>A redacted deep clone of the event.</returns>
    private static MxEvent RedactValues(MxEvent source)
    {
        MxEvent redacted = source.Clone();
        redacted.Value = null;

        if (redacted.BodyCase == MxEvent.BodyOneofCase.OnAlarmTransition)
        {
            redacted.OnAlarmTransition.CurrentValue = null;
            redacted.OnAlarmTransition.LimitValue = null;
        }

        return redacted;
    }

    /// <summary>
    /// One in-process subscriber's feed: a bounded, drop-oldest channel plus the
    /// registry bookkeeping that keeps <see cref="Publish"/>'s viewer gate honest
    /// while the feed is live.
    /// </summary>
    private sealed class InProcessSubscription : IDashboardEventSubscription
    {
        private readonly DashboardEventBroadcaster _owner;

        private readonly Channel<MxEvent> _channel;

        private int _disposed;

        /// <summary>Initializes a new instance of the <see cref="InProcessSubscription"/> class.</summary>
        /// <param name="owner">Broadcaster to deregister from on disposal.</param>
        /// <param name="sessionId">Session whose events this subscription carries.</param>
        /// <param name="connectionId">Synthetic connection id registered with the viewer registry.</param>
        /// <param name="capacity">Queue depth before the oldest queued event is dropped.</param>
        internal InProcessSubscription(
            DashboardEventBroadcaster owner,
            string sessionId,
            string connectionId,
            int capacity)
        {
            _owner = owner;
            SessionId = sessionId;
            ConnectionId = connectionId;
            _channel = Channel.CreateBounded<MxEvent>(new BoundedChannelOptions(capacity)
            {
                // DropOldest, not Wait: a write must never block the gRPC event
                // pipeline that calls Publish, and the newest events are the ones a
                // live view wants.
                FullMode = BoundedChannelFullMode.DropOldest,
                SingleReader = true,
                SingleWriter = false,
            });
        }

        /// <inheritdoc />
        public ChannelReader<MxEvent> Reader => _channel.Reader;

        /// <summary>Gets the session this subscription is watching.</summary>
        internal string SessionId { get; }

        /// <summary>Gets the synthetic connection id held in the viewer registry.</summary>
        internal string ConnectionId { get; }

        /// <summary>
        /// Queues an event for the subscriber, dropping the oldest queued event when
        /// the reader has fallen behind. Never blocks and never throws.
        /// </summary>
        /// <param name="mxEvent">The event to queue.</param>
        internal void TryWrite(MxEvent mxEvent) => _channel.Writer.TryWrite(mxEvent);

        /// <summary>
        /// Deregisters the subscription and completes its channel so a reader's
        /// loop ends. Idempotent — a second call does nothing, so it can never
        /// release a viewer count that a sibling subscription owns.
        /// </summary>
        public void Dispose()
        {
            if (Interlocked.Exchange(ref _disposed, 1) != 0)
            {
                return;
            }

            _owner.Unsubscribe(this);
            _channel.Writer.TryComplete();
        }
    }

    /// <summary>
    /// One session's hub-delivery pump: a bounded drop-oldest channel and the single
    /// task that drains it into the session's SignalR group, one bounded send at a
    /// time. The drain is the only reader, so at most one send per session is ever
    /// pending in the lifetime manager.
    /// </summary>
    private sealed class HubPump
    {
        private readonly Channel<MxEvent> _channel;

        /// <summary>Initializes a new instance of the <see cref="HubPump"/> class and starts its drain.</summary>
        /// <param name="owner">Broadcaster whose send path the drain uses.</param>
        /// <param name="sessionId">Session whose group the pump feeds.</param>
        /// <param name="capacity">Queue depth before the oldest queued event is dropped.</param>
        internal HubPump(DashboardEventBroadcaster owner, string sessionId, int capacity)
        {
            _channel = Channel.CreateBounded<MxEvent>(new BoundedChannelOptions(capacity)
            {
                // DropOldest, not Wait: Publish runs on the gRPC event pipeline and
                // must never block, and the newest events are the ones a live view wants.
                FullMode = BoundedChannelFullMode.DropOldest,
                SingleReader = true,
                SingleWriter = false,
            });

            // Task.Run so the drain never starts inline on the publishing thread.
            Completion = Task.Run(() => DrainAsync(owner, sessionId));
        }

        /// <summary>Gets the drain task; it completes once the channel is completed and emptied.</summary>
        internal Task Completion { get; }

        /// <summary>Queues an event, dropping the oldest queued one when full. Never blocks.</summary>
        /// <param name="mxEvent">The event to queue.</param>
        internal void TryWrite(MxEvent mxEvent) => _channel.Writer.TryWrite(mxEvent);

        /// <summary>Completes the channel so the drain exits once it is empty.</summary>
        internal void Complete() => _channel.Writer.TryComplete();

        private async Task DrainAsync(DashboardEventBroadcaster owner, string sessionId)
        {
            ChannelReader<MxEvent> reader = _channel.Reader;
            while (await reader.WaitToReadAsync().ConfigureAwait(false))
            {
                while (reader.TryRead(out MxEvent? next))
                {
                    await owner.SendToHubAsync(sessionId, next).ConfigureAwait(false);
                }
            }
        }
    }
}