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"/>. Fire-and-forget: we
/// hand the send to the hub context and return immediately so the source gRPC
/// stream is never blocked. Errors are logged once and dropped — keeping the
/// SignalR mirror best-effort preserves the gRPC contract that exists today.
/// </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.
/// </remarks>
/// <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 sealed class DashboardEventBroadcaster(
    IHubContext<EventsHub> hubContext,
    EventsHubViewerRegistry viewerRegistry,
    IOptions<GatewayOptions> options,
    ILogger<DashboardEventBroadcaster> logger) : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber
{
    /// <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 already has.
    /// </summary>
    private const int InProcessQueueCapacity = 256;

    private readonly bool _showTagValues = options.Value.Dashboard.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>
    /// 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;

    /// <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) plus a send to an empty SignalR group
        // per event. Bail before both.
        if (!viewerRegistry.HasViewers(sessionId))
        {
            return;
        }

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

        // In-process delivery first: it is synchronous, cannot throw, and must not be
        // skipped by the early return the hub send's guard clause takes.
        DeliverInProcess(sessionId, outbound);

        // Wrap the Task acquisition in a try/catch so a hypothetical synchronous throw
        // from SendAsync (e.g. an implementation that throws before returning the Task)
        // cannot escape Publish. The interface contract is never-throw; fire-and-forget.
        Task send;
        try
        {
            send = hubContext.Clients
                .Group(EventsHub.GroupName(sessionId))
                .SendAsync(EventsHub.EventMessage, outbound);
        }
        catch (Exception ex)
        {
            logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} threw synchronously.", sessionId);
            return;
        }

        if (!send.IsCompletedSuccessfully)
        {
            _ = send.ContinueWith(
                t =>
                {
                    if (t.Exception is { } ex)
                    {
                        logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId);
                    }
                },
                TaskScheduler.Default);
        }
    }

    /// <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 = "inproc-" + 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();
        }
    }
}