ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs

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;

/// <summary>
/// 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.
/// </summary>
public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
{
    private readonly GrpcChannel? _channel;
    private readonly SiteStreamService.SiteStreamServiceClient? _client;
    private readonly ILogger? _logger;
    private readonly ConcurrentDictionary<string, CancellationTokenSource> _subscriptions = new();

    /// <summary>
    /// The gRPC endpoint (site node address) this client is bound to. The
    /// <see cref="SiteStreamGrpcClientFactory"/> 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.
    /// </summary>
    public virtual string Endpoint { get; } = string.Empty;

    /// <summary>
    /// The HTTP/2 keepalive ping delay actually applied to this client's channel.
    /// Exposed for tests verifying that <see cref="CommunicationOptions"/> is honoured.
    /// </summary>
    internal TimeSpan KeepAlivePingDelay { get; }

    /// <summary>
    /// The HTTP/2 keepalive ping timeout actually applied to this client's channel.
    /// Exposed for tests verifying that <see cref="CommunicationOptions"/> is honoured.
    /// </summary>
    internal TimeSpan KeepAlivePingTimeout { get; }

    /// <summary>
    /// Creates a client with default communication options.
    /// </summary>
    /// <param name="endpoint">The gRPC endpoint address for the site.</param>
    /// <param name="logger">Logger for diagnostics and errors.</param>
    public SiteStreamGrpcClient(string endpoint, ILogger logger)
        : this(endpoint, logger, new CommunicationOptions())
    {
    }

    /// <summary>
    /// Creates a client whose HTTP/2 keepalive is taken from <see cref="CommunicationOptions"/>
    /// rather than hard-coded, satisfying the design doc's "gRPC Connection Keepalive"
    /// section which states these values are configurable.
    /// </summary>
    /// <param name="endpoint">The gRPC endpoint address for the site.</param>
    /// <param name="logger">Logger for diagnostics and errors.</param>
    /// <param name="options">Communication options including keepalive settings.</param>
    public SiteStreamGrpcClient(string endpoint, ILogger logger, CommunicationOptions options)
        : this(endpoint, logger, options, pskProvider: null, siteIdentifier: null)
    {
    }

    /// <summary>
    /// Creates a client that authenticates every call with the site's preshared key.
    /// This is the production shape: <c>SiteStreamService</c> is gated by
    /// <c>ControlPlaneAuthInterceptor</c> on the site node, so a client without credentials
    /// gets <see cref="StatusCode.PermissionDenied"/> on every call.
    /// </summary>
    /// <param name="endpoint">The gRPC endpoint address for the site.</param>
    /// <param name="logger">Logger for diagnostics and errors.</param>
    /// <param name="options">Communication options including keepalive settings.</param>
    /// <param name="pskProvider">Resolves the site's preshared key; null leaves the channel unauthenticated.</param>
    /// <param name="siteIdentifier">Site this channel talks to; null leaves the channel unauthenticated.</param>
    public SiteStreamGrpcClient(
        string endpoint,
        ILogger logger,
        CommunicationOptions options,
        ISitePskProvider? pskProvider,
        string? siteIdentifier)
    {
        ConnectedHeaderTimeout = options.ConnectedHeaderTimeout;
        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
            }
        }.WithMessageLimits().WithSiteCredentials(pskProvider, siteIdentifier, endpoint));
        _client = new SiteStreamService.SiteStreamServiceClient(_channel);
        _logger = logger;
    }

    /// <summary>
    /// Protected constructor for unit testing without a real gRPC channel.
    /// Allows subclassing for mock implementations.
    /// </summary>
    protected SiteStreamGrpcClient()
    {
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="endpoint">The gRPC endpoint address for the site.</param>
    protected SiteStreamGrpcClient(string endpoint)
    {
        Endpoint = endpoint;
    }

    /// <summary>
    /// Creates a test-only instance that has no gRPC channel. Used to test
    /// Unsubscribe and Dispose behavior without needing a real endpoint.
    /// </summary>
    /// <param name="options">Optional communication options; defaults are used when null.</param>
    /// <returns>A <see cref="SiteStreamGrpcClient"/> with no channel or client, for testing only.</returns>
    internal static SiteStreamGrpcClient CreateForTesting(CommunicationOptions? options = null) => new(options ?? new CommunicationOptions());

    private SiteStreamGrpcClient(CommunicationOptions options)
    {
        ConnectedHeaderTimeout = options.ConnectedHeaderTimeout;
    }

    /// <summary>
    /// Number of live subscriptions currently registered on this client. The owning
    /// <see cref="SiteStreamGrpcClientFactory"/> never evicts a client while this is non-zero.
    /// </summary>
    public virtual int ActiveSubscriptionCount => _subscriptions.Count;

    /// <summary>
    /// Registers a CancellationTokenSource for a correlation ID. Test-only.
    /// </summary>
    /// <param name="correlationId">Unique identifier for the subscription.</param>
    /// <param name="cts">CancellationTokenSource for managing the subscription lifecycle.</param>
    internal void AddSubscriptionForTesting(string correlationId, CancellationTokenSource cts)
    {
        _subscriptions[correlationId] = cts;
    }

    /// <summary>
    /// Registers a subscription's CancellationTokenSource for a correlation ID.
    /// If an entry already exists for that correlation ID (a reconnect race where two
    /// <see cref="SubscribeAsync"/> calls briefly share an ID), the prior CTS is
    /// <b>cancelled only</b> — never disposed from here. Internal for testability.
    /// <para>
    /// <b>Cancel-only is the contract</b> (the client half of the
    /// shared-token-source disposal race): the prior handler's <see cref="ConsumeStreamAsync"/> is still running and
    /// still reads <c>cts.Token</c> (opening the call, <c>ReadAllAsync</c>, the
    /// <c>IsCancellationRequested</c> checks), so disposing it from the REPLACING path throws
    /// <see cref="ObjectDisposedException"/> inside the prior stream — and, worse, out of the
    /// <c>prior.Cancel()</c> call itself into the NEW <c>SubscribeAsync</c>, where it reads as a
    /// fault of the current generation and burns a reconnect. Disposal belongs to the OWNING
    /// path: <see cref="SubscribeAsync"/>/<see cref="SubscribeSiteAsync"/> dispose the CTS they
    /// created once their stream loop has fully unwound.
    /// </para>
    /// </summary>
    /// <param name="correlationId">Unique identifier for the subscription.</param>
    /// <param name="cts">CancellationTokenSource for managing the subscription lifecycle.</param>
    internal void RegisterSubscription(string correlationId, CancellationTokenSource cts)
    {
        if (_subscriptions.TryGetValue(correlationId, out var prior) && !ReferenceEquals(prior, cts))
        {
            CancelQuietly(prior);
        }
        _subscriptions[correlationId] = cts;
    }

    /// <summary>
    /// Cancels a subscription token source, swallowing <see cref="ObjectDisposedException"/>.
    /// 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.
    /// </summary>
    /// <param name="cts">The token source to cancel.</param>
    private static void CancelQuietly(CancellationTokenSource cts)
    {
        try
        {
            cts.Cancel();
        }
        catch (ObjectDisposedException)
        {
            // Already disposed by its owning stream loop — nothing left to cancel.
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="correlationId">Unique identifier for the subscription.</param>
    /// <param name="cts">CancellationTokenSource to match before removing.</param>
    internal void RemoveSubscription(string correlationId, CancellationTokenSource cts)
    {
        _subscriptions.TryRemove(new KeyValuePair<string, CancellationTokenSource>(correlationId, cts));
    }

    /// <summary>
    /// 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 <paramref name="onEvent"/> callback delivers domain events,
    /// <paramref name="onError"/> lets the caller handle reconnection after a fault, and
    /// <paramref name="onCompleted"/> reports a graceful end of stream.
    /// </summary>
    /// <param name="correlationId">Unique identifier for this subscription.</param>
    /// <param name="instanceUniqueName">Unique name of the instance to subscribe to.</param>
    /// <param name="onEvent">Callback invoked for each domain event received from the stream.</param>
    /// <param name="onError">Callback invoked when the subscription encounters an error.</param>
    /// <param name="onCompleted">
    /// 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
    /// <paramref name="onError"/>; see <see cref="ConsumeStreamAsync"/>.
    /// </param>
    /// <param name="ct">Cancellation token to stop the subscription.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public virtual async Task SubscribeAsync(
        string correlationId,
        string instanceUniqueName,
        Action<object> onEvent,
        Action<Exception> 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. 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();
        }
    }

    /// <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:
    /// 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 the
    /// 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="onCompleted">
    /// 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
    /// <paramref name="onError"/>; see <see cref="ConsumeStreamAsync"/>.
    /// </param>
    /// <param name="ct">Cancellation token to stop the subscription.</param>
    /// <param name="onConnected">
    /// 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 <paramref name="onError"/>.
    /// </param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public virtual async Task SubscribeSiteAsync(
        string correlationId,
        Action<AlarmStateChanged> onAlarmEvent,
        Action<Exception> 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();
        }
    }

    /// <summary>
    /// Drains a server-streaming call to its end, classifying the outcome into exactly one
    /// of three terminations: a fault (<paramref name="onError"/>), a graceful server-side
    /// end of stream (<paramref name="onCompleted"/>), or our own cancellation (neither).
    /// <para>
    /// The graceful case is the one that matters operationally: the site server caps every
    /// stream at <c>CommunicationOptions.GrpcMaxStreamLifetime</c> (4 h) and, when that
    /// elapses, ends the RPC with status OK. The client-side <c>await foreach</c> then simply
    /// finishes, so before <c>onCompleted</c> existed the consuming actor observed nothing at
    /// all and the stream stayed silently dead until the process restarted.
    /// </para>
    /// Internal so the completion/fault classification can be unit-tested with a fake
    /// <see cref="IAsyncStreamReader{T}"/> and no live channel.
    /// </summary>
    /// <param name="correlationId">Unique identifier for this subscription.</param>
    /// <param name="cts">The subscription's linked token source (owned by the caller).</param>
    /// <param name="openCall">
    /// Opens the server-streaming call. Invoked inside the guarded region so a throw while
    /// opening is reported through <paramref name="onError"/> like any other stream fault.
    /// </param>
    /// <param name="onEvent">Invoked per wire event.</param>
    /// <param name="onError">Invoked once if the stream faulted.</param>
    /// <param name="onCompleted">Invoked once if the server ended the stream with OK.</param>
    /// <param name="onConnected">
    /// Optional; invoked AT MOST ONCE when the site has demonstrably accepted the
    /// subscription — either the server's response headers arrived (bounded by
    /// <see cref="ConnectedHeaderTimeout"/>) 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.
    /// </param>
    /// <returns>A task that completes when the stream has ended and its outcome been reported.</returns>
    internal async Task ConsumeStreamAsync(
        string correlationId,
        CancellationTokenSource cts,
        Func<AsyncServerStreamingCall<SiteStreamEvent>> openCall,
        Action<SiteStreamEvent> onEvent,
        Action<Exception> 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();
    }

    /// <summary>
    /// 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.
    /// </summary>
    internal TimeSpan ConnectedHeaderTimeout { get; } = TimeSpan.FromSeconds(10);

    /// <summary>
    /// Awaits the call's response headers, bounded by <see cref="ConnectedHeaderTimeout"/>.
    /// Returns <see langword="true"/> when the headers arrived (the site accepted the
    /// subscription) and <see langword="false"/> on timeout. A fault propagates (the caller
    /// reports it through <c>onError</c> 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.
    /// <para>
    /// A timeout must NOT be reported as connected: an unreachable/wedged site produces
    /// exactly that shape, and calling <c>onConnected</c> for it made the aggregator clear
    /// <c>_streamDown</c>, 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.
    /// </para>
    /// </summary>
    private async Task<bool> AwaitHeadersAsync(
        AsyncServerStreamingCall<SiteStreamEvent> 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;
        }
    }

    /// <summary>
    /// Cancels an active subscription by correlation ID. Cancel-only: the subscription's own
    /// <see cref="SubscribeAsync"/>/<see cref="SubscribeSiteAsync"/> call disposes the token
    /// source once its stream loop unwinds. Cancelling an entry whose owner already
    /// disposed it — or unsubscribing after the client was disposed — is a silent no-op.
    /// </summary>
    /// <param name="correlationId">Unique identifier of the subscription to cancel.</param>
    public virtual void Unsubscribe(string correlationId)
    {
        if (_subscriptions.TryRemove(correlationId, out var cts))
        {
            CancelQuietly(cts);
        }
    }

    /// <summary>
    /// Unpacks one wire frame into the per-event pipeline (R2 — gRPC event batching).
    /// A plain <c>attribute_changed</c>/<c>alarm_changed</c> frame is delivered as-is; a
    /// <see cref="SiteStreamEventBatch"/> frame is unpacked <b>in order</b> into the same
    /// callback, so every downstream consumer (<c>SiteAlarmAggregatorActor</c>,
    /// <c>DebugStreamBridgeActor</c>, the consumer-keepalive/orphan logic, per-event
    /// <c>Timestamp</c> fidelity) sees no difference between a batched and an unbatched site.
    /// <para>
    /// 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.
    /// </para>
    /// </summary>
    /// <param name="frame">The wire frame received from the site.</param>
    /// <param name="handler">Invoked once per contained event, in arrival order.</param>
    internal static void ForEachEvent(SiteStreamEvent frame, Action<SiteStreamEvent> 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);
            }
        }
    }

    /// <summary>
    /// Converts a proto SiteStreamEvent to the corresponding domain message.
    /// Internal for testability.
    /// </summary>
    /// <param name="evt">The protobuf site stream event to convert.</param>
    /// <returns>The converted domain event, or null if the event type is not recognized.</returns>
    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(),
            // Flow Debugger (design §5.4): fields 7/8 — an older site leaves them at the
            // proto3 defaults, which map to null / false.
            SourceExecutionId: Guid.TryParse(evt.AttributeChanged.SourceExecutionId, out var traceSource) ? traceSource : null,
            Trace: evt.AttributeChanged.Trace),
        SiteStreamEvent.EventOneofCase.AlarmChanged => ConvertAlarmChanged(evt.AlarmChanged),
        _ => null
    };

    /// <summary>
    /// Maps one <see cref="AlarmStateUpdate"/> to the domain record. The wire carries only the raw
    /// <c>priority</c> (its pre-#129 meaning, unchanged), so the unified, higher-is-more-urgent
    /// Severity is derived here from the kind (Gitea #129): banded for MxAccess and computed rows,
    /// verbatim for OPC UA. The raw value stays on <see cref="AlarmConditionState.Priority"/>.
    /// </summary>
    private static AlarmStateChanged ConvertAlarmChanged(AlarmStateUpdate update)
    {
        var kind = ParseAlarmKind(update.Kind);
        return new AlarmStateChanged(
            update.InstanceUniqueName,
            update.AlarmName,
            MapAlarmState(update.State),
            update.Priority,
            update.Timestamp.ToDateTimeOffset())
        {
            Level = MapAlarmLevel(update.Level),
            Message = update.Message ?? string.Empty,
            // Native alarm enrichment (additive — computed alarms carry their default condition).
            Kind = kind,
            Condition = new AlarmConditionState(
                Active: update.Active,
                Acknowledged: update.Acknowledged,
                Confirmed: update.Confirmed,
                Shelve: AlarmShelveStateCodec.Parse(update.ShelveState),
                Suppressed: update.Suppressed,
                Severity: AlarmConditionStateFactory.SeverityFromPriority(kind, update.Priority),
                Priority: update.Priority),
            SourceReference = update.SourceReference ?? string.Empty,
            AlarmTypeName = update.AlarmTypeName ?? string.Empty,
            Category = update.Category ?? string.Empty,
            OperatorUser = update.OperatorUser ?? string.Empty,
            OperatorComment = update.OperatorComment ?? string.Empty,
            OriginalRaiseTime = update.OriginalRaiseTime?.ToDateTimeOffset(),
            CurrentValue = update.CurrentValue ?? string.Empty,
            LimitValue = update.LimitValue ?? string.Empty,
            NativeSourceCanonicalName = update.NativeSourceCanonicalName ?? string.Empty,
            IsConfiguredPlaceholder = update.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 = update.AckTime?.ToDateTimeOffset(),
            // Gitea #60: a pre-#60 site leaves the field at its proto3 default (false).
            IsMesReportable = update.IsMesReportable
        };
    }

    /// <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>
    internal static AlarmKind ParseAlarmKind(string? kind) =>
        System.Enum.TryParse<AlarmKind>(kind, ignoreCase: true, out var k) ? k : AlarmKind.Computed;

    /// <summary>
    /// Maps proto Quality enum to domain string. Internal for testability.
    /// </summary>
    /// <param name="quality">The protobuf quality value to map.</param>
    /// <returns>The mapped quality as a string ("Good", "Uncertain", "Bad", or "Unknown").</returns>
    internal static string MapQuality(Quality quality) => quality switch
    {
        Quality.Good => "Good",
        Quality.Uncertain => "Uncertain",
        Quality.Bad => "Bad",
        _ => "Unknown"
    };

    /// <summary>
    /// Maps proto AlarmStateEnum to domain AlarmState. Internal for testability.
    /// </summary>
    /// <param name="state">The protobuf alarm state to map.</param>
    /// <returns>The mapped domain alarm state.</returns>
    internal static AlarmState MapAlarmState(AlarmStateEnum state) => state switch
    {
        AlarmStateEnum.AlarmStateNormal => AlarmState.Normal,
        AlarmStateEnum.AlarmStateActive => AlarmState.Active,
        _ => AlarmState.Normal
    };

    /// <summary>
    /// Maps proto AlarmLevelEnum to domain AlarmLevel. Internal for testability.
    /// </summary>
    /// <param name="level">The protobuf alarm level to map.</param>
    /// <returns>The mapped domain alarm level.</returns>
    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
    };

    /// <summary>
    /// Cancels every live subscription and releases the underlying gRPC channel. All teardown
    /// here is synchronous (CTS cancellation and <see cref="GrpcChannel.Dispose"/>), so a
    /// synchronous <see cref="Dispose"/> can release everything without sync-over-async blocking.
    /// <para>
    /// Cancel-only, per-entry guarded. The token sources are owned by their
    /// <see cref="SubscribeAsync"/>/<see cref="SubscribeSiteAsync"/> calls, which dispose them
    /// as their loops unwind in response to this cancellation; disposing them here raced those
    /// loops, and one <see cref="ObjectDisposedException"/> aborted the whole foreach — which in
    /// turn aborted <see cref="SiteStreamGrpcClientFactory.DisposeAsync"/>'s client loop and left
    /// every later client (and its channel) undisposed.
    /// </para>
    /// </summary>
    private void ReleaseResources()
    {
        foreach (var cts in _subscriptions.Values)
        {
            CancelQuietly(cts);
        }
        _subscriptions.Clear();

        _channel?.Dispose();
    }

    /// <summary>
    /// Asynchronously disposes of the gRPC client and all subscriptions.
    /// </summary>
    /// <returns>A completed <see cref="ValueTask"/> after all subscriptions and the gRPC channel have been released.</returns>
    public virtual ValueTask DisposeAsync()
    {
        ReleaseResources();
        return ValueTask.CompletedTask;
    }

    /// <summary>
    /// Synchronous disposal. All resources held by this client are released
    /// synchronously, so callers (e.g. <see cref="SiteStreamGrpcClientFactory.Dispose"/>)
    /// need not block on the async disposal path.
    /// </summary>
    public virtual void Dispose()
    {
        ReleaseResources();
    }
}