feat(sitestream): add central SubscribeSiteAsync alarm-only client (plan #10 T3)

Add SiteStreamGrpcClient.SubscribeSiteAsync mirroring SubscribeAsync but for the site-wide, alarm-only SubscribeSite RPC: builds a SiteStreamRequest, opens the server stream, and delivers each event via a typed Action<AlarmStateChanged> callback (this stream is alarm-only by contract, so Task 4's per-site cache consumes an alarm delta with no downstream type test). Reuses the shared enrichment mapping via a new internal ConvertToAlarmEvent helper that returns null for any non-alarm event, defensively filtering anything that should never appear on the stream. Factory unchanged - it already caches a client per (site, endpoint). Adds focused unit tests for the alarm-only filter and the test-only-client guard. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
2026-07-10 11:49:49 -04:00
parent 9c631b47e1
commit 6d5ecc92a5
2 changed files with 151 additions and 0 deletions
@@ -200,6 +200,74 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
        }
    }

+    /// <summary>
+    /// Opens the site-wide, <b>alarm-only</b> server-streaming subscription
+    /// (<c>SubscribeSite</c>) for a whole site rather than a single instance. This is
+    /// the central per-site feed that backs the aggregated Alarm Summary live cache
+    /// (plan #10): the site runtime's <c>SubscribeSiteAlarms</c> hub drops the
+    /// per-instance filter and carries only <see cref="AlarmStateChanged"/> events for
+    /// <em>all</em> instances on the site (attributes are deliberately excluded — the
+    /// summary never shows them and they are far higher-volume).
+    /// <para>
+    /// The callback is deliberately <b>typed</b> as <see cref="AlarmStateChanged"/>
+    /// rather than the generic <c>Action&lt;object&gt;</c> used by
+    /// <see cref="SubscribeAsync"/>: this stream is alarm-only by contract, so Task 4's
+    /// per-site cache consumes an alarm delta directly with no downstream type test.
+    /// The mapping is still shared via <see cref="ConvertToDomainEvent"/>; a non-alarm
+    /// event (which should never appear on this stream) is defensively ignored rather
+    /// than delivered or thrown.
+    /// </para>
+    /// This is a long-running async method; the caller launches it as a background task.
+    /// Error handling, cancellation and cleanup mirror <see cref="SubscribeAsync"/>.
+    /// </summary>
+    /// <param name="correlationId">Unique identifier for this subscription.</param>
+    /// <param name="onAlarmEvent">Callback invoked for each alarm delta received from the site-wide stream.</param>
+    /// <param name="onError">Callback invoked when the subscription encounters an error.</param>
+    /// <param name="ct">Cancellation token to stop the subscription.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    public virtual async Task SubscribeSiteAsync(
+        string correlationId,
+        Action<AlarmStateChanged> onAlarmEvent,
+        Action<Exception> onError,
+        CancellationToken ct)
+    {
+        if (_client is null)
+            throw new InvalidOperationException("Cannot subscribe on a test-only client.");
+
+        var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        RegisterSubscription(correlationId, cts);
+
+        var request = new SiteStreamRequest
+        {
+            CorrelationId = correlationId
+        };
+
+        try
+        {
+            using var call = _client.SubscribeSite(request, cancellationToken: cts.Token);
+
+            await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token))
+            {
+                // Site-wide stream is alarm-only by contract; defensively ignore anything else.
+                if (ConvertToAlarmEvent(evt) is { } alarm)
+                    onAlarmEvent(alarm);
+            }
+        }
+        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
+        {
+            // Normal cancellation — not an error
+        }
+        catch (Exception ex)
+        {
+            onError(ex);
+        }
+        finally
+        {
+            // Remove only our own entry -- a racing reconnect may already own the slot.
+            RemoveSubscription(correlationId, cts);
+        }
+    }
+
    /// <summary>
    /// Cancels an active subscription by correlation ID.
    /// </summary>
@@ -260,6 +328,17 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
        _ => null
    };

+    /// <summary>
+    /// Maps a proto <see cref="SiteStreamEvent"/> to a domain <see cref="AlarmStateChanged"/>,
+    /// returning <c>null</c> for any non-alarm event. Used by <see cref="SubscribeSiteAsync"/>
+    /// to enforce the alarm-only contract of the site-wide stream without duplicating the
+    /// enrichment mapping in <see cref="ConvertToDomainEvent"/>. Internal for testability.
+    /// </summary>
+    /// <param name="evt">The protobuf site stream event to convert.</param>
+    /// <returns>The mapped <see cref="AlarmStateChanged"/>, or <c>null</c> if the event is not an alarm.</returns>
+    internal static AlarmStateChanged? ConvertToAlarmEvent(SiteStreamEvent evt) =>
+        ConvertToDomainEvent(evt) as AlarmStateChanged;
+
    /// <summary>Parses the wire "kind" string back to <see cref="AlarmKind"/>; defaults to Computed.</summary>
    /// <param name="kind">The wire "kind" string from the gRPC payload; null or unrecognised defaults to <see cref="AlarmKind.Computed"/>.</param>
    /// <returns>The parsed <see cref="AlarmKind"/>, or <see cref="AlarmKind.Computed"/> when the value is null or unrecognised.</returns>
@@ -208,6 +208,78 @@ public class SiteStreamGrpcClientTests
        Assert.True(cts2.IsCancellationRequested);
    }

+    // --- Site-wide (SubscribeSite) alarm-only stream (plan #10 T3) ---
+
+    [Fact]
+    public void ConvertToAlarmEvent_AlarmChanged_ReturnsMappedAlarm()
+    {
+        // An alarm event on the site-wide stream is delivered to onAlarmEvent with full enrichment.
+        var ts = DateTimeOffset.UtcNow;
+        var evt = new SiteStreamEvent
+        {
+            CorrelationId = "site-corr",
+            AlarmChanged = new AlarmStateUpdate
+            {
+                InstanceUniqueName = "Site1.Motor01",
+                AlarmName = "T01.Hi",
+                State = AlarmStateEnum.AlarmStateActive,
+                Priority = 700,
+                Timestamp = Timestamp.FromDateTimeOffset(ts),
+                Kind = "NativeOpcUa",
+                Active = true,
+                SourceReference = "T01.Hi"
+            }
+        };
+
+        var alarm = SiteStreamGrpcClient.ConvertToAlarmEvent(evt);
+
+        Assert.NotNull(alarm);
+        Assert.Equal("Site1.Motor01", alarm!.InstanceUniqueName);
+        Assert.Equal("T01.Hi", alarm.AlarmName);
+        Assert.Equal(AlarmState.Active, alarm.State);
+        Assert.Equal(AlarmKind.NativeOpcUa, alarm.Kind);
+        Assert.Equal("T01.Hi", alarm.SourceReference);
+    }
+
+    [Fact]
+    public void ConvertToAlarmEvent_AttributeChanged_ReturnsNull()
+    {
+        // Attribute events must never appear on the alarm-only site-wide stream; if one
+        // somehow arrives it is defensively filtered out rather than delivered or thrown.
+        var evt = new SiteStreamEvent
+        {
+            CorrelationId = "site-corr",
+            AttributeChanged = new AttributeValueUpdate
+            {
+                InstanceUniqueName = "Site1.Pump01",
+                AttributePath = "Modules.IO",
+                AttributeName = "Temperature",
+                Value = "42.5",
+                Quality = Quality.Good,
+                Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow)
+            }
+        };
+
+        Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt));
+    }
+
+    [Fact]
+    public void ConvertToAlarmEvent_UnknownEvent_ReturnsNull()
+    {
+        var evt = new SiteStreamEvent { CorrelationId = "site-corr" };
+        Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt));
+    }
+
+    [Fact]
+    public async Task SubscribeSiteAsync_OnTestOnlyClient_Throws()
+    {
+        // Guards against subscribing on a channel-less test double, mirroring SubscribeAsync.
+        var client = SiteStreamGrpcClient.CreateForTesting();
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() =>
+            client.SubscribeSiteAsync("corr", _ => { }, _ => { }, CancellationToken.None));
+    }
+
    // --- Communication-003 regression tests ---

    [Fact]