using System.Collections.Concurrent; using Grpc.Core; using Grpc.Net.Client; using Microsoft.Extensions.Logging; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using Google.Protobuf.WellKnownTypes; namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// /// Per-site gRPC client that manages streaming subscriptions to a site's /// SiteStreamGrpcServer. The central-side DebugStreamBridgeActor uses this /// to open server-streaming calls for individual instances. /// public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable { private readonly GrpcChannel? _channel; private readonly SiteStreamService.SiteStreamServiceClient? _client; private readonly ILogger? _logger; private readonly ConcurrentDictionary _subscriptions = new(); /// /// The gRPC endpoint (site node address) this client is bound to. The /// compares this against the requested /// endpoint so a NodeA→NodeB failover flip (or a site address edit) is honoured /// rather than served stale from cache. /// public virtual string Endpoint { get; } = string.Empty; /// /// The HTTP/2 keepalive ping delay actually applied to this client's channel. /// Exposed for tests verifying that is honoured. /// internal TimeSpan KeepAlivePingDelay { get; } /// /// The HTTP/2 keepalive ping timeout actually applied to this client's channel. /// Exposed for tests verifying that is honoured. /// internal TimeSpan KeepAlivePingTimeout { get; } /// /// Creates a client with default communication options. /// /// The gRPC endpoint address for the site. /// Logger for diagnostics and errors. public SiteStreamGrpcClient(string endpoint, ILogger logger) : this(endpoint, logger, new CommunicationOptions()) { } /// /// Creates a client whose HTTP/2 keepalive is taken from /// rather than hard-coded, satisfying the design doc's "gRPC Connection Keepalive" /// section which states these values are configurable. /// /// The gRPC endpoint address for the site. /// Logger for diagnostics and errors. /// Communication options including keepalive settings. public SiteStreamGrpcClient(string endpoint, ILogger logger, CommunicationOptions options) : this(endpoint, logger, options, pskProvider: null, siteIdentifier: null) { } /// /// Creates a client that authenticates every call with the site's preshared key. /// This is the production shape: SiteStreamService is gated by /// ControlPlaneAuthInterceptor on the site node, so a client without credentials /// gets on every call. /// /// The gRPC endpoint address for the site. /// Logger for diagnostics and errors. /// Communication options including keepalive settings. /// Resolves the site's preshared key; null leaves the channel unauthenticated. /// Site this channel talks to; null leaves the channel unauthenticated. public SiteStreamGrpcClient( string endpoint, ILogger logger, CommunicationOptions options, ISitePskProvider? pskProvider, string? siteIdentifier) { Endpoint = endpoint; KeepAlivePingDelay = options.GrpcKeepAlivePingDelay; KeepAlivePingTimeout = options.GrpcKeepAlivePingTimeout; _channel = GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions { HttpHandler = new SocketsHttpHandler { KeepAlivePingDelay = options.GrpcKeepAlivePingDelay, KeepAlivePingTimeout = options.GrpcKeepAlivePingTimeout, KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always } }.WithSiteCredentials(pskProvider, siteIdentifier)); _client = new SiteStreamService.SiteStreamServiceClient(_channel); _logger = logger; } /// /// Protected constructor for unit testing without a real gRPC channel. /// Allows subclassing for mock implementations. /// protected SiteStreamGrpcClient() { } /// /// Protected constructor for unit testing — records the endpoint without /// opening a real gRPC channel, so endpoint-aware factory behaviour can be /// exercised by test doubles. /// /// The gRPC endpoint address for the site. protected SiteStreamGrpcClient(string endpoint) { Endpoint = endpoint; } /// /// Creates a test-only instance that has no gRPC channel. Used to test /// Unsubscribe and Dispose behavior without needing a real endpoint. /// /// A with no channel or client, for testing only. internal static SiteStreamGrpcClient CreateForTesting() => new(); /// /// Registers a CancellationTokenSource for a correlation ID. Test-only. /// /// Unique identifier for the subscription. /// CancellationTokenSource for managing the subscription lifecycle. internal void AddSubscriptionForTesting(string correlationId, CancellationTokenSource cts) { _subscriptions[correlationId] = cts; } /// /// Registers a subscription's CancellationTokenSource for a correlation ID. /// If an entry already exists for that correlation ID (a reconnect race where two /// calls briefly share an ID), the prior CTS is /// cancelled only — never disposed from here. Internal for testability. /// /// Cancel-only is the contract (arch-review 2026-08-19 P3-1, the client half of the /// row-47 class): the prior handler's is still running and /// still reads cts.Token (opening the call, ReadAllAsync, the /// IsCancellationRequested checks), so disposing it from the REPLACING path throws /// inside the prior stream — and, worse, out of the /// prior.Cancel() call itself into the NEW SubscribeAsync, where it reads as a /// fault of the current generation and burns a reconnect. Disposal belongs to the OWNING /// path: / dispose the CTS they /// created once their stream loop has fully unwound. /// /// /// Unique identifier for the subscription. /// CancellationTokenSource for managing the subscription lifecycle. internal void RegisterSubscription(string correlationId, CancellationTokenSource cts) { if (_subscriptions.TryGetValue(correlationId, out var prior) && !ReferenceEquals(prior, cts)) { CancelQuietly(prior); } _subscriptions[correlationId] = cts; } /// /// Cancels a subscription token source, swallowing . /// A CTS whose owning stream loop has already unwound and disposed it is simply a /// subscription that no longer needs cancelling; the caller must never see that race as a /// failure (arch-review 2026-08-19 P3-1). /// /// The token source to cancel. private static void CancelQuietly(CancellationTokenSource cts) { try { cts.Cancel(); } catch (ObjectDisposedException) { // Already disposed by its owning stream loop — nothing left to cancel. } } /// /// Removes the subscription entry for a correlation ID only if the stored CTS is /// exactly the one supplied. A racing replacement stream may already own the slot, /// in which case this is a no-op. Internal for testability. /// /// Unique identifier for the subscription. /// CancellationTokenSource to match before removing. internal void RemoveSubscription(string correlationId, CancellationTokenSource cts) { _subscriptions.TryRemove(new KeyValuePair(correlationId, cts)); } /// /// Opens a server-streaming subscription for a specific instance. /// This is a long-running async method; the caller launches it as a background task. /// The callback delivers domain events, /// lets the caller handle reconnection after a fault, and /// reports a graceful end of stream. /// /// Unique identifier for this subscription. /// Unique name of the instance to subscribe to. /// Callback invoked for each domain event received from the stream. /// Callback invoked when the subscription encounters an error. /// /// Callback invoked when the server ended the RPC with OK — the site's max stream /// lifetime elapsing or a graceful site shutdown. Mutually exclusive with /// ; see . /// /// Cancellation token to stop the subscription. /// A task that represents the asynchronous operation. public virtual async Task SubscribeAsync( string correlationId, string instanceUniqueName, Action onEvent, Action onError, Action onCompleted, 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 InstanceStreamRequest { CorrelationId = correlationId, InstanceUniqueName = instanceUniqueName, // R2 batch negotiation. Declaring support is safe against ANY site: one that // predates batching ignores the unknown field and keeps sending per-event // frames, which ForEachEvent handles as the single-event case. BatchingSupported = true }; void Deliver(SiteStreamEvent single) { var domainEvent = ConvertToDomainEvent(single); if (domainEvent != null) onEvent(domainEvent); } // This call OWNS `cts` — it created it, so it disposes it once the stream loop has // fully unwound (P3-1). No other path may dispose it while this one is still reading // cts.Token; a replacing subscription cancels only. try { await ConsumeStreamAsync( correlationId, cts, () => _client.SubscribeInstance(request, cancellationToken: cts.Token), frame => ForEachEvent(frame, Deliver), onError, onCompleted); } finally { cts.Dispose(); } } /// /// Opens the site-wide, alarm-only server-streaming subscription /// (SubscribeSite) 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 SubscribeSiteAlarms hub drops the /// per-instance filter and carries only events for /// all instances on the site (attributes are deliberately excluded — the /// summary never shows them and they are far higher-volume). /// /// The callback is deliberately typed as /// rather than the generic Action<object> used by /// : 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 ; a non-alarm /// event (which should never appear on this stream) is defensively ignored rather /// than delivered or thrown. /// /// This is a long-running async method; the caller launches it as a background task. /// Error handling, cancellation and cleanup mirror . /// /// Unique identifier for this subscription. /// Callback invoked for each alarm delta received from the site-wide stream. /// Callback invoked when the subscription encounters an error. /// /// Callback invoked when the server ended the RPC with OK — the site's max stream /// lifetime elapsing or a graceful site shutdown. Mutually exclusive with /// ; see . /// /// Cancellation token to stop the subscription. /// /// Optional callback invoked once when the site has ACCEPTED the subscription — response /// headers received (the site writes them as soon as its relay actor is subscribed, so no /// event can be missed after this point), or the first event received if the peer defers /// its headers. A header timeout is NOT treated as connected. The per-site aggregator uses /// this to run exactly one re-seed per successful (re)connect instead of one per reconnect /// attempt. Never invoked more than once per call, and never after . /// /// A task that represents the asynchronous operation. public virtual async Task SubscribeSiteAsync( string correlationId, Action onAlarmEvent, Action onError, Action onCompleted, CancellationToken ct, Action? onConnected = null) { 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, // R2 batch negotiation — see SubscribeAsync. BatchingSupported = true }; void Deliver(SiteStreamEvent single) { // Site-wide stream is alarm-only by contract; defensively ignore anything else. if (ConvertToAlarmEvent(single) is { } alarm) onAlarmEvent(alarm); } // Owns `cts` — see SubscribeAsync. try { await ConsumeStreamAsync( correlationId, cts, () => _client.SubscribeSite(request, cancellationToken: cts.Token), frame => ForEachEvent(frame, Deliver), onError, onCompleted, onConnected); } finally { cts.Dispose(); } } /// /// Drains a server-streaming call to its end, classifying the outcome into exactly one /// of three terminations: a fault (), a graceful server-side /// end of stream (), or our own cancellation (neither). /// /// The graceful case is the one that matters operationally: the site server caps every /// stream at CommunicationOptions.GrpcMaxStreamLifetime (4 h) and, when that /// elapses, ends the RPC with status OK. The client-side await foreach then simply /// finishes, so before onCompleted existed the consuming actor observed nothing at /// all and the stream stayed silently dead until the process restarted. /// /// Internal so the completion/fault classification can be unit-tested with a fake /// and no live channel. /// /// Unique identifier for this subscription. /// The subscription's linked token source (owned by the caller). /// /// Opens the server-streaming call. Invoked inside the guarded region so a throw while /// opening is reported through like any other stream fault. /// /// Invoked per wire event. /// Invoked once if the stream faulted. /// Invoked once if the server ended the stream with OK. /// /// Optional; invoked AT MOST ONCE when the site has demonstrably accepted the /// subscription — either the server's response headers arrived (bounded by /// ) or, for a peer that defers headers until its /// first message, the first event was received. A header timeout alone is never /// reported as connected: that is also exactly what an unreachable site looks like. /// /// A task that completes when the stream has ended and its outcome been reported. internal async Task ConsumeStreamAsync( string correlationId, CancellationTokenSource cts, Func> openCall, Action onEvent, Action onError, Action onCompleted, Action? onConnected = null) { var completedGracefully = false; var connectedReported = false; // Fires onConnected AT MOST ONCE, from whichever proof of a live peer arrives // first: the response headers, or (for a peer that defers headers until its first // message) the first event. A header TIMEOUT is deliberately NOT such a proof — // see AwaitHeadersAsync. void ReportConnected() { if (connectedReported || onConnected is null) return; connectedReported = true; onConnected(); } try { using (var call = openCall()) { if (onConnected is not null && await AwaitHeadersAsync(call, cts.Token).ConfigureAwait(false)) { ReportConnected(); } await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token)) { // Fallback connected signal: an event can only come from a peer that // accepted the subscription, so it proves what the headers would have. // Raised BEFORE the event is delivered so the consumer sees // connected-then-event ordering. ReportConnected(); onEvent(evt); } } completedGracefully = true; } catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled && cts.IsCancellationRequested) { // OUR OWN cancellation (Unsubscribe / reconnect / channel teardown we asked // for) — not an error. The IsCancellationRequested guard matters: a Cancelled // status can also originate at the PEER or from a channel disposed underneath // us, and swallowing THAT fired none of onError/onCompleted/onConnected, so // the consuming actor kept a dead stream marked live forever. Foreign // Cancelled now falls through to the onError path below. } catch (OperationCanceledException) when (cts.IsCancellationRequested) { // Our own Unsubscribe/reconnect cancelled the read — not an error either. } catch (Exception ex) { onError(ex); } finally { // Remove only our own entry -- a racing reconnect may already own the slot. RemoveSubscription(correlationId, cts); } // Raised outside the try so a throwing callback is not misreported as a stream // fault. Suppressed when our own token was cancelled as the stream ended: that is // a teardown, and the caller has already moved on to a newer stream. if (completedGracefully && !cts.IsCancellationRequested) onCompleted(); } /// /// How long to wait for response headers before giving up on them as the connected /// signal and falling back to the first received event. A peer that only flushes headers /// with its first message would otherwise hold the connected signal — and with it the /// aggregator's re-seed — for as long as the site happens to be quiet. /// internal static TimeSpan ConnectedHeaderTimeout { get; set; } = TimeSpan.FromSeconds(10); /// /// Awaits the call's response headers, bounded by . /// Returns when the headers arrived (the site accepted the /// subscription) and on timeout. A fault propagates (the caller /// reports it through onError like any other stream fault); on timeout the /// abandoned headers task is observed so a later fault on it can never surface as an /// unobserved task exception. /// /// A timeout must NOT be reported as connected: an unreachable/wedged site produces /// exactly that shape, and calling onConnected for it made the aggregator clear /// _streamDown, consume its pending re-seed and fan a full snapshot out at a site /// that never answered. The caller instead treats the FIRST RECEIVED EVENT as the /// fallback connected signal — real proof of a live peer, and the quiet-site case the /// timeout was added for is covered by the reconcile backstop. /// /// private static async Task AwaitHeadersAsync( AsyncServerStreamingCall call, CancellationToken ct) { var headers = call.ResponseHeadersAsync; try { await headers.WaitAsync(ConnectedHeaderTimeout, ct).ConfigureAwait(false); return true; } catch (TimeoutException) { _ = headers.ContinueWith( t => _ = t.Exception, CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); return false; } } /// /// Cancels an active subscription by correlation ID. Cancel-only: the subscription's own /// / call disposes the token /// source once its stream loop unwinds (P3-1). Cancelling an entry whose owner already /// disposed it — or unsubscribing after the client was disposed — is a silent no-op. /// /// Unique identifier of the subscription to cancel. public virtual void Unsubscribe(string correlationId) { if (_subscriptions.TryRemove(correlationId, out var cts)) { CancelQuietly(cts); } } /// /// Unpacks one wire frame into the per-event pipeline (R2 — gRPC event batching). /// A plain attribute_changed/alarm_changed frame is delivered as-is; a /// frame is unpacked in order into the same /// callback, so every downstream consumer (SiteAlarmAggregatorActor, /// DebugStreamBridgeActor, the consumer-keepalive/orphan logic, per-event /// Timestamp fidelity) sees no difference between a batched and an unbatched site. /// /// Unpacking is deliberately NON-RECURSIVE: the server never nests a batch inside a /// batch, and a nested or unknown inner case from a malformed peer is skipped rather /// than followed. Internal for testability. /// /// /// The wire frame received from the site. /// Invoked once per contained event, in arrival order. internal static void ForEachEvent(SiteStreamEvent frame, Action handler) { if (frame.EventCase != SiteStreamEvent.EventOneofCase.Batch) { handler(frame); return; } foreach (var inner in frame.Batch.Events) { if (inner.EventCase is SiteStreamEvent.EventOneofCase.AttributeChanged or SiteStreamEvent.EventOneofCase.AlarmChanged) { handler(inner); } } } /// /// Converts a proto SiteStreamEvent to the corresponding domain message. /// Internal for testability. /// /// The protobuf site stream event to convert. /// The converted domain event, or null if the event type is not recognized. internal static object? ConvertToDomainEvent(SiteStreamEvent evt) => evt.EventCase switch { SiteStreamEvent.EventOneofCase.AttributeChanged => new AttributeValueChanged( evt.AttributeChanged.InstanceUniqueName, evt.AttributeChanged.AttributePath, evt.AttributeChanged.AttributeName, evt.AttributeChanged.Value, MapQuality(evt.AttributeChanged.Quality), evt.AttributeChanged.Timestamp.ToDateTimeOffset()), SiteStreamEvent.EventOneofCase.AlarmChanged => new AlarmStateChanged( evt.AlarmChanged.InstanceUniqueName, evt.AlarmChanged.AlarmName, MapAlarmState(evt.AlarmChanged.State), evt.AlarmChanged.Priority, evt.AlarmChanged.Timestamp.ToDateTimeOffset()) { Level = MapAlarmLevel(evt.AlarmChanged.Level), Message = evt.AlarmChanged.Message ?? string.Empty, // Native alarm enrichment (additive — computed alarms carry their default condition). Kind = ParseAlarmKind(evt.AlarmChanged.Kind), Condition = new AlarmConditionState( Active: evt.AlarmChanged.Active, Acknowledged: evt.AlarmChanged.Acknowledged, Confirmed: evt.AlarmChanged.Confirmed, Shelve: AlarmShelveStateCodec.Parse(evt.AlarmChanged.ShelveState), Suppressed: evt.AlarmChanged.Suppressed, Severity: evt.AlarmChanged.Priority), SourceReference = evt.AlarmChanged.SourceReference ?? string.Empty, AlarmTypeName = evt.AlarmChanged.AlarmTypeName ?? string.Empty, Category = evt.AlarmChanged.Category ?? string.Empty, OperatorUser = evt.AlarmChanged.OperatorUser ?? string.Empty, OperatorComment = evt.AlarmChanged.OperatorComment ?? string.Empty, OriginalRaiseTime = evt.AlarmChanged.OriginalRaiseTime?.ToDateTimeOffset(), CurrentValue = evt.AlarmChanged.CurrentValue ?? string.Empty, LimitValue = evt.AlarmChanged.LimitValue ?? string.Empty, NativeSourceCanonicalName = evt.AlarmChanged.NativeSourceCanonicalName ?? string.Empty, IsConfiguredPlaceholder = evt.AlarmChanged.IsConfiguredPlaceholder, // MES alarm-status API §6.4: ack instant; an absent proto Timestamp (unacked // condition, computed alarm, or a pre-AckTime site) stays null. AckTime = evt.AlarmChanged.AckTime?.ToDateTimeOffset() }, _ => null }; /// /// Maps a proto to a domain , /// returning null for any non-alarm event. Used by /// to enforce the alarm-only contract of the site-wide stream without duplicating the /// enrichment mapping in . Internal for testability. /// /// The protobuf site stream event to convert. /// The mapped , or null if the event is not an alarm. internal static AlarmStateChanged? ConvertToAlarmEvent(SiteStreamEvent evt) => ConvertToDomainEvent(evt) as AlarmStateChanged; /// Parses the wire "kind" string back to ; defaults to Computed. /// The wire "kind" string from the gRPC payload; null or unrecognised defaults to . /// The parsed , or when the value is null or unrecognised. internal static AlarmKind ParseAlarmKind(string? kind) => System.Enum.TryParse(kind, ignoreCase: true, out var k) ? k : AlarmKind.Computed; /// /// Maps proto Quality enum to domain string. Internal for testability. /// /// The protobuf quality value to map. /// The mapped quality as a string ("Good", "Uncertain", "Bad", or "Unknown"). internal static string MapQuality(Quality quality) => quality switch { Quality.Good => "Good", Quality.Uncertain => "Uncertain", Quality.Bad => "Bad", _ => "Unknown" }; /// /// Maps proto AlarmStateEnum to domain AlarmState. Internal for testability. /// /// The protobuf alarm state to map. /// The mapped domain alarm state. internal static AlarmState MapAlarmState(AlarmStateEnum state) => state switch { AlarmStateEnum.AlarmStateNormal => AlarmState.Normal, AlarmStateEnum.AlarmStateActive => AlarmState.Active, _ => AlarmState.Normal }; /// /// Maps proto AlarmLevelEnum to domain AlarmLevel. Internal for testability. /// /// The protobuf alarm level to map. /// The mapped domain alarm level. internal static AlarmLevel MapAlarmLevel(AlarmLevelEnum level) => level switch { AlarmLevelEnum.AlarmLevelLow => AlarmLevel.Low, AlarmLevelEnum.AlarmLevelLowLow => AlarmLevel.LowLow, AlarmLevelEnum.AlarmLevelHigh => AlarmLevel.High, AlarmLevelEnum.AlarmLevelHighHigh => AlarmLevel.HighHigh, _ => AlarmLevel.None }; /// /// Cancels every live subscription and releases the underlying gRPC channel. All teardown /// here is synchronous (CTS cancellation and ), so a /// synchronous can release everything without sync-over-async blocking. /// /// Cancel-only, per-entry guarded (P3-1). The token sources are owned by their /// / calls, which dispose them /// as their loops unwind in response to this cancellation; disposing them here raced those /// loops, and one aborted the whole foreach — which in /// turn aborted 's client loop and left /// every later client (and its channel) undisposed. /// /// private void ReleaseResources() { foreach (var cts in _subscriptions.Values) { CancelQuietly(cts); } _subscriptions.Clear(); _channel?.Dispose(); } /// /// Asynchronously disposes of the gRPC client and all subscriptions. /// /// A completed after all subscriptions and the gRPC channel have been released. public virtual ValueTask DisposeAsync() { ReleaseResources(); return ValueTask.CompletedTask; } /// /// Synchronous disposal. All resources held by this client are released /// synchronously, so callers (e.g. ) /// need not block on the async disposal path. /// public virtual void Dispose() { ReleaseResources(); } }