feat(dashboard): in-process session event subscription feeds the viewer registry — SessionDetailsPage drops its /hubs/events hop

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

@@ -1,3 +1,4 @@
+using System.Threading.Channels;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Options;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
@@ -6,11 +7,13 @@ using ZB.MOM.WW.MxGateway.Server.Configuration;
 namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
 
 /// <summary>
-/// Broadcasts MxEvents to <see cref="EventsHub"/> clients subscribed to the
-/// session's group. 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.
+/// 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
@@ -23,7 +26,10 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
 /// </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.
+/// 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>
@@ -31,10 +37,36 @@ public sealed class DashboardEventBroadcaster(
     IHubContext<EventsHub> hubContext,
     EventsHubViewerRegistry viewerRegistry,
     IOptions<GatewayOptions> options,
-    ILogger<DashboardEventBroadcaster> logger) : IDashboardEventBroadcaster
+    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)
     {
@@ -55,6 +87,10 @@ public sealed class DashboardEventBroadcaster(
 
         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.
@@ -85,6 +121,117 @@ public sealed class DashboardEventBroadcaster(
         }
     }
 
+    /// <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
@@ -107,4 +254,75 @@ public sealed class DashboardEventBroadcaster(
 
         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();
+        }
+    }
 }

src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardEventSubscription.cs

@@ -0,0 +1,26 @@
+using System.Threading.Channels;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+
+namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
+
+/// <summary>
+/// A live in-process feed of one session's dashboard-mirrored MxEvents, handed
+/// out by <see cref="IDashboardSessionEventSubscriber.Subscribe"/>. Server-side
+/// Blazor components read it directly instead of looping back through
+/// <see cref="EventsHub"/> over a loopback SignalR connection.
+/// </summary>
+/// <remarks>
+/// Events are delivered exactly as a hub client would see them — the same
+/// redacted clone the group send carries, so <c>MxGateway:Dashboard:ShowTagValues</c>
+/// governs both paths identically. The feed is a bounded, lossy queue: a
+/// consumer that falls behind loses the oldest queued events, matching the
+/// best-effort contract the SignalR mirror already has. Disposing the
+/// subscription deregisters it, which is what lets the broadcaster go back to
+/// skipping all mirror work for a session nobody is watching — so callers must
+/// dispose. Dispose is idempotent.
+/// </remarks>
+public interface IDashboardEventSubscription : IDisposable
+{
+    /// <summary>Gets the reader delivering this session's mirrored events.</summary>
+    ChannelReader<MxEvent> Reader { get; }
+}

src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardSessionEventSubscriber.cs

@@ -0,0 +1,34 @@
+namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
+
+/// <summary>
+/// In-process subscription seam on the dashboard event mirror. Implemented by
+/// <see cref="DashboardEventBroadcaster"/> alongside
+/// <see cref="IDashboardEventBroadcaster"/>.
+/// </summary>
+/// <remarks>
+/// The interactive-server dashboard runs in the same process as the broadcaster,
+/// so a session-details page has no reason to open a loopback SignalR connection
+/// back to <see cref="EventsHub"/> — mint a hub token, negotiate, hold a
+/// WebSocket, and serialize every event — just to read events the broadcaster
+/// already holds. It subscribes here instead. The registry gate stays honest
+/// either way: an in-process subscription registers a synthetic connection id
+/// with <see cref="EventsHubViewerRegistry"/> exactly as the hub registers a real
+/// one, so <see cref="IDashboardEventBroadcaster.Publish"/> keeps skipping the
+/// redaction clone for sessions nobody is watching.
+/// <para>
+/// It is a separate interface rather than a member of
+/// <see cref="IDashboardEventBroadcaster"/> because publishing and consuming are
+/// different roles: the session pipeline only ever publishes, and its test
+/// doubles should not have to implement a subscription feed.
+/// </para>
+/// </remarks>
+public interface IDashboardSessionEventSubscriber
+{
+    /// <summary>Opens an in-process feed of the session's mirrored events.</summary>
+    /// <param name="sessionId">Session id whose events the caller wants.</param>
+    /// <returns>
+    /// The subscription. Dispose it to stop the feed and release the viewer
+    /// registration that keeps the mirror enabled for this session.
+    /// </returns>
+    IDashboardEventSubscription Subscribe(string sessionId);
+}