mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs

using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;

namespace ZB.MOM.WW.MxGateway.Server.Alarms;

/// <summary>
///     The gateway's always-on alarm monitor and broker. It owns one
///     gateway-managed worker session dedicated to alarms, keeps an in-process
///     cache of the active-alarm set fed by that session's transition events
///     (reconciled periodically against the worker's snapshot), and fans the
///     feed out to any number of <see cref="StreamAsync"/> subscribers.
///     The session is re-opened transparently if the worker faults.
/// </summary>
public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmService
{
    private const string MonitorClientName = "gateway-alarm-monitor";
    private const string BackendName = "Galaxy";
    private const int SubscriberQueueCapacity = 2048;
    private static readonly TimeSpan RestartBackoff = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan StartupGrace = TimeSpan.FromSeconds(2);

    private readonly ISessionManager _sessionManager;
    private readonly IAlarmWatchListResolver _watchListResolver;
    private readonly GatewayMetrics _metrics;
    private readonly AlarmsOptions _options;
    private readonly ILogger<GatewayAlarmMonitor> _logger;

    private readonly object _sync = new();
    private readonly Dictionary<string, ActiveAlarmSnapshot> _alarms = new(StringComparer.Ordinal);
    private readonly List<Subscriber> _subscribers = [];

    // Memoized CurrentAlarms projection, guarded by _sync: the cloned, read-only view of _alarms
    // handed to the dashboard and the QueryActiveAlarms RPC. Cloning the whole set per read held
    // _sync — the broadcast lock — for the length of the copy, so a polled dashboard stalled every
    // ApplyTransition/Broadcast behind it. Null means "not built for the current generation":
    // every path that writes _alarms must null this under _sync, or readers keep a stale set.
    private ActiveAlarmSnapshot[]? _currentAlarmsProjection;

    // NEXT-03 dedup tombstones, guarded by _sync: alarm instances whose Clear was synthesized by
    // the most recent reconcile pass, keyed by reference with the instance's original raise
    // timestamp as the identity marker. A buffered live Clear for the same instance is a duplicate
    // of the repair and is suppressed. One generation deep: each reconcile pass replaces the map,
    // so a tombstone lives at least one reconcile interval — far longer than the lease buffer the
    // duplicate would be sitting in — and the map stays bounded by the feed's churn per interval.
    private readonly Dictionary<string, Timestamp> _clearedByReconcile = new(StringComparer.Ordinal);

    // Current provider status (mode + degraded + reason + since), guarded by _sync.
    // Initialized to the alarm-manager, not-degraded baseline so a late joiner sees
    // a sensible status even before any OnAlarmProviderModeChanged event arrives.
    private AlarmProviderMode _providerMode = AlarmProviderMode.Alarmmgr;
    private bool _providerDegraded;
    private string _providerReason = string.Empty;
    private DateTimeOffset _providerSince = DateTimeOffset.UtcNow;

    // Whether the worker's most recent reconcile fetch was capped, guarded by _sync.
    // Written only by ApplyReconcile (and cleared with the cache), so it describes the last full
    // reconcile — not necessarily the current _alarms contents, which live transitions keep moving
    // via ApplyTransition between passes. Read it as "as of the last reconcile, the worker's fetch
    // was capped", which is the right granularity for a completeness caveat.
    private bool _snapshotTruncated;

    private volatile GatewayAlarmMonitorState _state = GatewayAlarmMonitorState.Disabled;
    private volatile string? _lastError;
    private GatewaySession? _session;

    /// <summary>Initializes the gateway alarm monitor.</summary>
    /// <param name="sessionManager">Gateway session manager.</param>
    /// <param name="watchListResolver">Resolver for the subtag-fallback watch-list.</param>
    /// <param name="metrics">Gateway metrics sink.</param>
    /// <param name="options">Gateway options carrying the alarm configuration.</param>
    /// <param name="logger">Diagnostic logger.</param>
    public GatewayAlarmMonitor(
        ISessionManager sessionManager,
        IAlarmWatchListResolver watchListResolver,
        GatewayMetrics metrics,
        IOptions<GatewayOptions> options,
        ILogger<GatewayAlarmMonitor> logger)
    {
        _sessionManager = sessionManager ?? throw new ArgumentNullException(nameof(sessionManager));
        _watchListResolver = watchListResolver ?? throw new ArgumentNullException(nameof(watchListResolver));
        _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
        _options = (options ?? throw new ArgumentNullException(nameof(options))).Value.Alarms;
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    /// <inheritdoc />
    public GatewayAlarmMonitorState State => _state;

    /// <inheritdoc />
    public string? LastError => _lastError;

    /// <inheritdoc />
    public int? WorkerProcessId
    {
        get { lock (_sync) { return _session?.WorkerProcessId; } }
    }

    /// <inheritdoc />
    public IReadOnlyList<ActiveAlarmSnapshot> CurrentAlarms
    {
        get
        {
            lock (_sync)
            {
                // Same clone semantics as an uncached read — callers still get instances no
                // mutation can leak back into the cache — but built once per alarm-set
                // generation instead of once per caller.
                return _currentAlarmsProjection ??= _alarms.Values
                    .Select(alarm => alarm.Clone())
                    .ToArray();
            }
        }
    }

    /// <inheritdoc />
    public bool SnapshotTruncated
    {
        get { lock (_sync) { return _snapshotTruncated; } }
    }

    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        if (!_options.Enabled)
        {
            _state = GatewayAlarmMonitorState.Disabled;
            _logger.LogInformation("Gateway alarm monitor disabled (MxGateway:Alarms:Enabled is false).");
            return;
        }

        string subscription = ResolveSubscription();
        if (string.IsNullOrWhiteSpace(subscription))
        {
            _state = GatewayAlarmMonitorState.Faulted;
            _lastError = "MxGateway:Alarms is enabled but no SubscriptionExpression / DefaultArea is configured.";
            _logger.LogError("{Diagnostic}", _lastError);
            return;
        }

        // Brief grace so worker-process launching and startup orphan cleanup
        // settle before the monitor opens its own session.
        try
        {
            await Task.Delay(StartupGrace, stoppingToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            return;
        }

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await RunMonitorAsync(subscription, stoppingToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception exception)
            {
                _state = GatewayAlarmMonitorState.Faulted;
                _lastError = exception.Message;
                _logger.LogWarning(
                    exception,
                    "Gateway alarm monitor lifecycle faulted; restarting in {Backoff}.",
                    RestartBackoff);
                try
                {
                    await Task.Delay(RestartBackoff, stoppingToken).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    break;
                }
            }
        }

        _state = GatewayAlarmMonitorState.Disabled;
    }

    // One monitoring lifecycle: open a session, subscribe alarms, reconcile,
    // then consume transition events until the session ends or is cancelled.
    private async Task RunMonitorAsync(string subscription, CancellationToken stoppingToken)
    {
        _state = GatewayAlarmMonitorState.Starting;

        // Derive the lifecycle baseline from the configured forced mode so a
        // ForceSubtag / ForceAlarmManager start advertises the correct mode even
        // though no OnAlarmProviderModeChanged event is raised in those modes
        // (only Auto/failover produces that event). ForceSubtag starts degraded.
        AlarmProviderMode initialMode;
        bool initialDegraded;
        string initialReason;
        switch (MapForcedMode(_options.Fallback.Mode))
        {
            case AlarmProviderMode.Subtag:
                initialMode = AlarmProviderMode.Subtag;
                initialDegraded = true;
                initialReason = AlarmProviderReasons.ForcedSubtag;
                break;
            case AlarmProviderMode.Alarmmgr:
                initialMode = AlarmProviderMode.Alarmmgr;
                initialDegraded = false;
                initialReason = string.Empty;
                break;
            default:
                // Unspecified (Auto): the failover consumer starts on the
                // alarm-manager primary and only degrades to subtag on failure.
                initialMode = AlarmProviderMode.Alarmmgr;
                initialDegraded = false;
                initialReason = string.Empty;
                break;
        }

        lock (_sync)
        {
            // Re-baseline the provider status for this lifecycle so a restarted
            // monitor advertises the configured mode until told otherwise.
            _providerMode = initialMode;
            _providerDegraded = initialDegraded;
            _providerReason = initialReason;
            _providerSince = DateTimeOffset.UtcNow;
        }

        // Align the observable gauge with the lifecycle baseline without recording
        // a switch — the gauge was 0 (unknown) from construction until now.
        _metrics.SetAlarmProviderMode(ModeToInt(initialMode));

        GatewaySession session = await _sessionManager.OpenSessionAsync(
                new SessionOpenRequest(BackendName, MonitorClientName, Guid.NewGuid().ToString("N"), CommandTimeout: null),
                MonitorClientName,
                ownerKeyId: null,
                stoppingToken)
            .ConfigureAwait(false);
        lock (_sync) { _session = session; }

        try
        {
            // Attach the internal distributor subscriber BEFORE subscribing (GWC-26). The pump
            // has been running since MarkReady started the dashboard mirror, and the distributor
            // only fans to subscribers registered at the time of the fan-out, so a subscriber
            // taken after SubscribeAlarms + the first reconcile would silently lose every
            // transition raised inside that two-round-trip window — and a missed Acknowledge is
            // never repaired by the presence-only reconcile deltas. Transitions arriving while we
            // subscribe and reconcile simply buffer in this lease's bounded channel; if it ever
            // overflowed, the internal subscriber is disconnected (it never faults the session),
            // the enumeration below ends, and the supervisor loop restarts the lifecycle.
            using IEventSubscriberLease alarmLease = session.AttachInternalEventSubscriber();

            await SubscribeAlarmsAsync(session.SessionId, subscription, stoppingToken).ConfigureAwait(false);
            await ReconcileAsync(session.SessionId, stoppingToken).ConfigureAwait(false);

            _state = GatewayAlarmMonitorState.Monitoring;
            _lastError = null;
            _logger.LogInformation(
                "Gateway alarm monitor active on {Subscription} (session {SessionId}, worker pid {WorkerPid}).",
                subscription,
                session.SessionId,
                session.WorkerProcessId);

            using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
            Task reconcileLoop = ReconcileLoopAsync(session.SessionId, linked.Token);
            try
            {
                // Consume mapped MxEvents through the session's single distributor pump (as an
                // internal, non-counted subscriber) rather than opening a second raw drain of the
                // worker event channel — a second drain would split events with the dashboard
                // mirror pump and silently lose Acknowledge/mode-change transitions. The lease was
                // taken above, before SubscribeAlarms; draining it only now is order-safe because
                // ApplyTransition handles alarms the reconcile already placed in the cache.
                await foreach (MxEvent mxEvent in alarmLease.Reader
                    .ReadAllAsync(linked.Token)
                    .ConfigureAwait(false))
                {
                    if (mxEvent is { BodyCase: MxEvent.BodyOneofCase.OnAlarmTransition }
                        && mxEvent.OnAlarmTransition is not null)
                    {
                        ApplyTransition(mxEvent.OnAlarmTransition);
                    }
                    else if (mxEvent is { BodyCase: MxEvent.BodyOneofCase.OnAlarmProviderModeChanged }
                        && mxEvent.OnAlarmProviderModeChanged is not null)
                    {
                        await ApplyProviderModeChangeAsync(
                                session.SessionId,
                                mxEvent.OnAlarmProviderModeChanged,
                                linked.Token)
                            .ConfigureAwait(false);
                    }
                }
            }
            finally
            {
                await linked.CancelAsync().ConfigureAwait(false);
                try
                {
                    await reconcileLoop.ConfigureAwait(false);
                }
                catch
                {
                    // Reconcile-loop teardown errors are not actionable here.
                }
            }

            // The event stream ended without cancellation — the worker session
            // closed or faulted. Surface it so the supervisor loop restarts.
            throw new InvalidOperationException("Alarm monitor worker event stream ended.");
        }
        finally
        {
            lock (_sync) { _session = null; }
            ClearCache();
            try
            {
                await _sessionManager.CloseSessionAsync(session.SessionId, CancellationToken.None).ConfigureAwait(false);
            }
            catch (Exception exception)
            {
                _logger.LogDebug(exception, "Closing alarm monitor session {SessionId} failed.", session.SessionId);
            }
        }
    }

    private async Task SubscribeAlarmsAsync(string sessionId, string subscription, CancellationToken cancellationToken)
    {
        IReadOnlyList<AlarmSubtagTarget> watchList = await _watchListResolver
            .ResolveAsync(_options, cancellationToken)
            .ConfigureAwait(false);

        AlarmProviderMode forcedMode = MapForcedMode(_options.Fallback.Mode);

        _logger.LogInformation(
            "Alarm subscribe: forcedMode={ForcedMode} configMode={ConfigMode} watchList={WatchListCount}.",
            forcedMode, _options.Fallback.Mode, watchList.Count);

        // When the forced mode is Unspecified (the "Auto" case) and the resolved
        // watch-list is empty — the common alarmmgr-only deployment — the command
        // is identical-in-effect to the historical SubscribeAlarms (wnwrap only):
        // the worker builds the wnwrap consumer and no subtag watch-list.
        SubscribeAlarmsCommand command = new()
        {
            SubscriptionExpression = subscription,
            ForcedMode = forcedMode,
            Failover = new AlarmFailoverConfig
            {
                ConsecutiveFailureThreshold = _options.Fallback.ConsecutiveFailureThreshold,
                FailbackProbeIntervalSeconds = _options.Fallback.FailbackProbeIntervalSeconds,
                FailbackStableProbes = _options.Fallback.FailbackStableProbes,
            },
        };
        command.WatchList.AddRange(watchList);

        WorkerCommandReply reply = await _sessionManager.InvokeAsync(
                sessionId,
                new WorkerCommand
                {
                    Command = new MxCommand
                    {
                        Kind = MxCommandKind.SubscribeAlarms,
                        SubscribeAlarms = command,
                    },
                },
                cancellationToken)
            .ConfigureAwait(false);

        ProtocolStatusCode? code = reply.Reply?.ProtocolStatus?.Code;
        if (code != ProtocolStatusCode.Ok)
        {
            string diagnostic = reply.Reply?.DiagnosticMessage
                ?? reply.Reply?.ProtocolStatus?.Message
                ?? $"status {code}";
            throw new InvalidOperationException($"Worker rejected SubscribeAlarms: {diagnostic}");
        }
    }

    private async Task ReconcileLoopAsync(string sessionId, CancellationToken cancellationToken)
    {
        try
        {
            int seconds = Math.Max(5, _options.ReconcileIntervalSeconds);
            using PeriodicTimer timer = new(TimeSpan.FromSeconds(seconds));
            while (await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false))
            {
                try
                {
                    await ReconcileAsync(sessionId, cancellationToken).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
                catch (Exception exception)
                {
                    _logger.LogDebug(exception, "Alarm reconcile pass failed; keeping the current cache.");
                }
            }
        }
        catch (OperationCanceledException)
        {
        }
    }

    private async Task ReconcileAsync(string sessionId, CancellationToken cancellationToken)
    {
        WorkerCommandReply reply = await _sessionManager.InvokeAsync(
                sessionId,
                new WorkerCommand
                {
                    Command = new MxCommand
                    {
                        Kind = MxCommandKind.QueryActiveAlarms,
                        QueryActiveAlarmsCommand = new QueryActiveAlarmsCommand { AlarmFilterPrefix = string.Empty },
                    },
                },
                cancellationToken)
            .ConfigureAwait(false);

        if (reply.Reply?.ProtocolStatus?.Code != ProtocolStatusCode.Ok)
        {
            return;
        }

        QueryActiveAlarmsReplyPayload? payload = reply.Reply.QueryActiveAlarms;
        if (payload is not null)
        {
            ApplyReconcile(payload.Snapshots, payload.SnapshotTruncated);
        }
    }

    // Applies a live transition to the cache and broadcasts it to subscribers.
    private void ApplyTransition(OnAlarmTransitionEvent transition)
    {
        string reference = transition.AlarmFullReference ?? string.Empty;
        if (reference.Length == 0)
        {
            return;
        }

        lock (_sync)
        {
            if (transition.TransitionKind == AlarmTransitionKind.Clear)
            {
                bool wasKnown = _alarms.Remove(reference);
                if (wasKnown)
                {
                    _currentAlarmsProjection = null;
                }

                if (!wasKnown && IsDuplicateOfReconcileClear(reference, transition))
                {
                    return;
                }
            }
            else
            {
                ActiveAlarmSnapshot snapshot = SnapshotFromTransition(transition);
                bool duplicate = _alarms.TryGetValue(reference, out ActiveAlarmSnapshot? existing)
                    && IsDuplicateOfCachedState(existing, snapshot);
                _alarms[reference] = snapshot;
                _currentAlarmsProjection = null;
                if (duplicate)
                {
                    return;
                }
            }

            Broadcast(new AlarmFeedMessage { Transition = transition }, reference);
        }
    }

    // NEXT-03: best-effort dedup of the reconcile/live race. A reconcile that already synthesized
    // this transition as a feed repair left the cache carrying the worker's transition timestamp
    // and resulting state — both derived from the same worker-side value the live transition
    // carries — so an exact (timestamp, state) match means this live transition's outcome has
    // already been broadcast. Suppress only on a positive match: an unset timestamp on either
    // side keeps today's at-least-once behavior.
    private static bool IsDuplicateOfCachedState(ActiveAlarmSnapshot existing, ActiveAlarmSnapshot incoming)
    {
        return existing.LastTransitionTimestamp is not null
            && incoming.LastTransitionTimestamp is not null
            && existing.LastTransitionTimestamp.Equals(incoming.LastTransitionTimestamp)
            && existing.CurrentState == incoming.CurrentState;
    }

    // NEXT-03, the Clear leg. A reconcile Clear repair removes the cache entry before the buffered
    // live Clear drains, so there is no cached state to compare against; the tombstone recorded by
    // ApplyReconcile identifies the cleared instance by its original raise timestamp instead. The
    // match consumes the tombstone, so a genuinely new raise/clear cycle (which carries a newer
    // original raise timestamp) is never swallowed. Caller holds _sync.
    private bool IsDuplicateOfReconcileClear(string reference, OnAlarmTransitionEvent transition)
    {
        if (transition.OriginalRaiseTimestamp is not null
            && _clearedByReconcile.TryGetValue(reference, out Timestamp? clearedInstance)
            && clearedInstance.Equals(transition.OriginalRaiseTimestamp))
        {
            _clearedByReconcile.Remove(reference);
            return true;
        }

        return false;
    }

    // Handles the worker's provider-mode-change event: updates the stored provider
    // status, broadcasts it to every subscriber (provider status is global, not
    // alarm-scoped), records the switch metric, and forces a cache reconcile so the
    // active-alarm set reflects whatever the new mode reports.
    private async Task ApplyProviderModeChangeAsync(
        string sessionId,
        OnAlarmProviderModeChangedEvent change,
        CancellationToken cancellationToken)
    {
        AlarmProviderMode toMode = change.Mode;
        string reason = change.Reason ?? string.Empty;

        AlarmProviderStatus status;
        int fromModeInt;
        lock (_sync)
        {
            fromModeInt = ModeToInt(_providerMode);
            _providerMode = toMode;
            _providerDegraded = toMode == AlarmProviderMode.Subtag;
            _providerReason = reason;
            _providerSince = DateTimeOffset.UtcNow;
            status = BuildProviderStatus();
            BroadcastToAll(new AlarmFeedMessage { ProviderStatus = status });
        }

        AlarmProviderSwitchReason switchReason = toMode switch
        {
            AlarmProviderMode.Subtag => AlarmProviderSwitchReason.Failover,
            AlarmProviderMode.Alarmmgr => AlarmProviderSwitchReason.Failback,
            _ => AlarmProviderSwitchReason.Unknown,
        };
        _metrics.AlarmProviderSwitched(fromModeInt, ModeToInt(toMode), switchReason);

        _logger.LogInformation(
            "Alarm provider mode changed to {Mode} (degraded={Degraded}): {Reason}",
            toMode,
            status.Degraded,
            reason);

        try
        {
            // Intentionally awaited OUTSIDE _sync: ReconcileAsync acquires _sync itself,
            // so holding it across the await here would deadlock. Subscribers therefore
            // see the ProviderStatus push (above) slightly before the cache is re-seeded
            // by the reconcile — an accepted brief inconsistency.
            await ReconcileAsync(sessionId, cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            throw;
        }
        catch (Exception exception)
        {
            _logger.LogDebug(
                exception,
                "Reconcile after alarm provider mode change failed; keeping the current cache.");
        }
    }

    // Caller holds _sync. Builds an AlarmProviderStatus snapshot of the current state.
    private AlarmProviderStatus BuildProviderStatus()
    {
        return new AlarmProviderStatus
        {
            Mode = _providerMode,
            Degraded = _providerDegraded,
            Reason = _providerReason,
            Since = Timestamp.FromDateTimeOffset(_providerSince),
        };
    }

    // Maps the configured fallback mode string to the forced provider mode the
    // worker honours. Case-insensitive; anything other than the two force values
    // (including the default "Auto") yields Unspecified ("let the worker decide").
    private static AlarmProviderMode MapForcedMode(string? mode)
    {
        if (string.Equals(mode, "ForceAlarmManager", StringComparison.OrdinalIgnoreCase))
        {
            return AlarmProviderMode.Alarmmgr;
        }

        if (string.Equals(mode, "ForceSubtag", StringComparison.OrdinalIgnoreCase))
        {
            return AlarmProviderMode.Subtag;
        }

        return AlarmProviderMode.Unspecified;
    }

    // Maps the provider-mode enum to the integer the metric expects
    // (alarmmgr=1, subtag=2, unknown/unspecified=0).
    private static int ModeToInt(AlarmProviderMode mode) => mode switch
    {
        AlarmProviderMode.Alarmmgr => 1,
        AlarmProviderMode.Subtag => 2,
        _ => 0,
    };

    // Replaces the cache with the worker's authoritative snapshot, broadcasting
    // a synthetic transition for any alarm the live stream missed.
    //
    // Repair scope (GWC-26): presence deltas (Clear/Raise) plus acked-state deltas. These are
    // ALARM FEED transitions (AlarmFeedMessage on the StreamAlarms/dashboard surface), rebuilt
    // from the worker's own authoritative snapshot to repair what the live feed missed. They are
    // not MxEvents and never reach the gRPC StreamEvents path, so this feed-level repair does not
    // breach the "never synthesize events" rule, which governs MxEvent emission.
    //
    // Delivery semantics: feed repair transitions are AT-LEAST-ONCE, not exactly-once. A reconcile
    // reads the worker's current state while the corresponding live transition may still be
    // buffered in the alarm lease's channel; both would then broadcast, and the two are
    // indistinguishable on the feed, since nothing serializes a reconcile against the in-flight
    // live stream. ApplyTransition narrows that window with a best-effort dedup (NEXT-03): a live
    // transition whose worker timestamp and resulting state the cache already carries — or whose
    // Clear matches a tombstone recorded below — was already broadcast as a repair and is
    // suppressed. The dedup fires only on a positive marker match, so the contract stays
    // at-least-once: consumers must still treat alarm state idempotently — apply a transition as
    // "set the alarm to this state", never as an increment or a toggle.
    //
    // Truncation (`snapshotTruncated`) needs no special handling here, and that is worth saying
    // because the obvious worry — a capped fetch reading as a wave of Clears — is answered one
    // level down. The worker merges rather than replaces its retained snapshot on a capped fetch,
    // so the set arriving here still carries the alarms the capped reply had no room to mention.
    // The flag is therefore only recorded, for the operator-facing completeness caveat.
    private void ApplyReconcile(IEnumerable<ActiveAlarmSnapshot> snapshots, bool snapshotTruncated)
    {
        Dictionary<string, ActiveAlarmSnapshot> next = new(StringComparer.Ordinal);
        foreach (ActiveAlarmSnapshot snapshot in snapshots)
        {
            if (!string.IsNullOrEmpty(snapshot.AlarmFullReference))
            {
                next[snapshot.AlarmFullReference] = snapshot;
            }
        }

        lock (_sync)
        {
            // Previous-generation tombstones have outlived the buffered live transitions they
            // guard against (one full reconcile interval); start this pass's generation fresh.
            _clearedByReconcile.Clear();

            foreach (KeyValuePair<string, ActiveAlarmSnapshot> existing in _alarms)
            {
                if (!next.ContainsKey(existing.Key))
                {
                    if (existing.Value.OriginalRaiseTimestamp is not null)
                    {
                        _clearedByReconcile[existing.Key] = existing.Value.OriginalRaiseTimestamp;
                    }

                    Broadcast(
                        new AlarmFeedMessage { Transition = TransitionFromSnapshot(existing.Value, AlarmTransitionKind.Clear) },
                        existing.Key);
                }
            }

            foreach (KeyValuePair<string, ActiveAlarmSnapshot> incoming in next)
            {
                if (!_alarms.TryGetValue(incoming.Key, out ActiveAlarmSnapshot? existing))
                {
                    Broadcast(
                        new AlarmFeedMessage { Transition = TransitionFromSnapshot(incoming.Value, AlarmTransitionKind.Raise) },
                        incoming.Key);
                }
                else if (existing.CurrentState != incoming.Value.CurrentState
                    && incoming.Value.CurrentState == AlarmConditionState.ActiveAcked)
                {
                    // The alarm was already known but the worker now reports it acknowledged: the
                    // live Acknowledge transition never reached the feed. Without this the acked
                    // state is absorbed silently by the snapshot replace below and subscribers show
                    // the alarm unacked until it clears.
                    Broadcast(
                        new AlarmFeedMessage { Transition = TransitionFromSnapshot(incoming.Value, AlarmTransitionKind.Acknowledge) },
                        incoming.Key);
                }
            }

            _alarms.Clear();
            foreach (KeyValuePair<string, ActiveAlarmSnapshot> incoming in next)
            {
                _alarms[incoming.Key] = incoming.Value;
            }

            _snapshotTruncated = snapshotTruncated;
            _currentAlarmsProjection = null;
        }
    }

    // Caller holds _sync. Pushes a feed message to every matching subscriber;
    // a subscriber that has fallen behind is completed with an error and dropped.
    private void Broadcast(AlarmFeedMessage message, string reference)
    {
        for (int index = _subscribers.Count - 1; index >= 0; index--)
        {
            Subscriber subscriber = _subscribers[index];
            if (!subscriber.Matches(reference))
            {
                continue;
            }

            if (!subscriber.Channel.Writer.TryWrite(message))
            {
                subscriber.Channel.Writer.TryComplete(new InvalidOperationException(
                    "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."));
                _subscribers.RemoveAt(index);
            }
        }
    }

    // Caller holds _sync. Pushes a feed message to every subscriber regardless of
    // its alarm-filter prefix. Used for provider-status messages, which are global
    // rather than scoped to a single alarm reference.
    private void BroadcastToAll(AlarmFeedMessage message)
    {
        for (int index = _subscribers.Count - 1; index >= 0; index--)
        {
            Subscriber subscriber = _subscribers[index];
            if (!subscriber.Channel.Writer.TryWrite(message))
            {
                subscriber.Channel.Writer.TryComplete(new InvalidOperationException(
                    "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."));
                _subscribers.RemoveAt(index);
            }
        }
    }

    private void ClearCache()
    {
        lock (_sync)
        {
            _alarms.Clear();
            // The truncation verdict describes the cache generation being discarded, so it goes
            // with it. Carrying it across a monitor restart would caveat an empty set as "may be
            // incomplete" on evidence from a session that no longer exists.
            _snapshotTruncated = false;
            _currentAlarmsProjection = null;
        }
    }

    /// <inheritdoc />
    public async IAsyncEnumerable<AlarmFeedMessage> StreamAsync(
        string? alarmFilterPrefix,
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
    {
        string prefix = alarmFilterPrefix ?? string.Empty;
        Channel<AlarmFeedMessage> channel = Channel.CreateBounded<AlarmFeedMessage>(
            new BoundedChannelOptions(SubscriberQueueCapacity)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleReader = true,
                SingleWriter = false,
            });
        Subscriber subscriber = new(channel, prefix);

        ActiveAlarmSnapshot[] snapshot;
        AlarmProviderStatus providerStatus;
        lock (_sync)
        {
            // Register before snapshotting under the same lock so neither a
            // transition nor a provider-mode change can slip between the snapshot
            // and the live stream.
            _subscribers.Add(subscriber);
            providerStatus = BuildProviderStatus();
            snapshot = _alarms.Values
                .Where(alarm => prefix.Length == 0
                    || alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal))
                .Select(alarm => alarm.Clone())
                .ToArray();
        }

        try
        {
            // Emit the current provider status first so a late joiner immediately
            // learns the mode (and whether the feed is degraded) before any alarms.
            yield return new AlarmFeedMessage { ProviderStatus = providerStatus };

            foreach (ActiveAlarmSnapshot alarm in snapshot)
            {
                yield return new AlarmFeedMessage { ActiveAlarm = alarm };
            }

            yield return new AlarmFeedMessage { SnapshotComplete = true };

            await foreach (AlarmFeedMessage message in channel.Reader
                .ReadAllAsync(cancellationToken)
                .ConfigureAwait(false))
            {
                yield return message;
            }
        }
        finally
        {
            lock (_sync) { _subscribers.Remove(subscriber); }
            channel.Writer.TryComplete();
        }
    }

    /// <inheritdoc />
    public async Task<AcknowledgeAlarmReply> AcknowledgeAsync(
        AcknowledgeAlarmRequest request,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(request);

        string? sessionId;
        lock (_sync) { sessionId = _session?.SessionId; }
        if (sessionId is null || _state != GatewayAlarmMonitorState.Monitoring)
        {
            return new AcknowledgeAlarmReply
            {
                CorrelationId = request.ClientCorrelationId,
                ProtocolStatus = new ProtocolStatus
                {
                    Code = ProtocolStatusCode.WorkerUnavailable,
                    Message = "Gateway alarm monitor is not currently active.",
                },
                DiagnosticMessage = _lastError ?? "Alarm monitor is not running.",
            };
        }

        MxCommand? command = BuildAcknowledgeCommand(request, out string? parseError);
        if (command is null)
        {
            return new AcknowledgeAlarmReply
            {
                CorrelationId = request.ClientCorrelationId,
                ProtocolStatus = new ProtocolStatus
                {
                    Code = ProtocolStatusCode.InvalidRequest,
                    Message = parseError ?? "Invalid acknowledge request.",
                },
                DiagnosticMessage = parseError ?? "Invalid acknowledge request.",
            };
        }

        WorkerCommandReply workerReply = await _sessionManager
            .InvokeAsync(sessionId, new WorkerCommand { Command = command }, cancellationToken)
            .ConfigureAwait(false);

        MxCommandReply mxReply = workerReply.Reply ?? new MxCommandReply
        {
            ProtocolStatus = new ProtocolStatus
            {
                Code = ProtocolStatusCode.ProtocolViolation,
                Message = "Worker reply did not include an MxCommandReply.",
            },
        };

        AcknowledgeAlarmReply reply = new()
        {
            CorrelationId = request.ClientCorrelationId,
            ProtocolStatus = mxReply.ProtocolStatus ?? new ProtocolStatus { Code = ProtocolStatusCode.Ok },
            DiagnosticMessage = mxReply.DiagnosticMessage ?? string.Empty,
        };
        if (mxReply.HasHresult)
        {
            reply.Hresult = mxReply.Hresult;
        }

        return reply;
    }

    private string ResolveSubscription()
    {
        if (!string.IsNullOrWhiteSpace(_options.SubscriptionExpression))
        {
            return _options.SubscriptionExpression;
        }

        if (!string.IsNullOrWhiteSpace(_options.DefaultArea))
        {
            return $@"\\{Environment.MachineName}\Galaxy!{_options.DefaultArea}";
        }

        return string.Empty;
    }

    private static MxCommand? BuildAcknowledgeCommand(AcknowledgeAlarmRequest request, out string? parseError)
    {
        parseError = null;
        if (string.IsNullOrWhiteSpace(request.AlarmFullReference))
        {
            parseError = "alarm_full_reference is required.";
            return null;
        }

        string comment = request.Comment ?? string.Empty;
        string operatorUser = request.OperatorUser ?? string.Empty;

        if (Guid.TryParse(request.AlarmFullReference, out Guid guid))
        {
            return new MxCommand
            {
                Kind = MxCommandKind.AcknowledgeAlarm,
                AcknowledgeAlarmCommand = new AcknowledgeAlarmCommand
                {
                    AlarmGuid = guid.ToString(),
                    Comment = comment,
                    OperatorUser = operatorUser,
                    OperatorNode = string.Empty,
                    OperatorDomain = string.Empty,
                    OperatorFullName = string.Empty,
                },
            };
        }

        if (TryParseAlarmReference(request.AlarmFullReference, out string provider, out string group, out string alarm))
        {
            return new MxCommand
            {
                Kind = MxCommandKind.AcknowledgeAlarmByName,
                AcknowledgeAlarmByNameCommand = new AcknowledgeAlarmByNameCommand
                {
                    AlarmName = alarm,
                    ProviderName = provider,
                    GroupName = group,
                    Comment = comment,
                    OperatorUser = operatorUser,
                    OperatorNode = string.Empty,
                    OperatorDomain = string.Empty,
                    OperatorFullName = string.Empty,
                },
            };
        }

        parseError = "alarm_full_reference must be a canonical GUID or 'Provider!Group.Tag' format.";
        return null;
    }

    /// <summary>
    ///     Parses an alarm reference of the form <c>Provider!Group.Tag</c>: the
    ///     first <c>!</c> splits provider from <c>Group.Tag</c>; the first
    ///     <c>.</c> after the <c>!</c> splits group from tag.
    /// </summary>
    /// <param name="reference">The full alarm reference.</param>
    /// <param name="providerName">The parsed provider.</param>
    /// <param name="groupName">The parsed group/area.</param>
    /// <param name="alarmName">The parsed tag/alarm name.</param>
    /// <returns>true on a well-formed reference; otherwise false.</returns>
    public static bool TryParseAlarmReference(
        string? reference,
        out string providerName,
        out string groupName,
        out string alarmName)
    {
        providerName = string.Empty;
        groupName = string.Empty;
        alarmName = string.Empty;
        if (string.IsNullOrWhiteSpace(reference))
        {
            return false;
        }

        int bang = reference!.IndexOf('!', StringComparison.Ordinal);
        if (bang <= 0 || bang == reference.Length - 1)
        {
            return false;
        }

        string left = reference[..bang];
        string right = reference[(bang + 1)..];
        int dot = right.IndexOf('.', StringComparison.Ordinal);
        if (dot <= 0 || dot == right.Length - 1)
        {
            return false;
        }

        providerName = left;
        groupName = right[..dot];
        alarmName = right[(dot + 1)..];
        return true;
    }

    private static ActiveAlarmSnapshot SnapshotFromTransition(OnAlarmTransitionEvent transition)
    {
        ActiveAlarmSnapshot snapshot = new()
        {
            AlarmFullReference = transition.AlarmFullReference,
            SourceObjectReference = transition.SourceObjectReference,
            AlarmTypeName = transition.AlarmTypeName,
            Severity = transition.Severity,
            CurrentState = transition.TransitionKind == AlarmTransitionKind.Acknowledge
                ? AlarmConditionState.ActiveAcked
                : AlarmConditionState.Active,
            Category = transition.Category,
            Description = transition.Description,
            OperatorUser = transition.OperatorUser,
            OperatorComment = transition.OperatorComment,
            Degraded = transition.Degraded,
            SourceProvider = transition.SourceProvider,
        };
        if (transition.OriginalRaiseTimestamp is not null)
        {
            snapshot.OriginalRaiseTimestamp = transition.OriginalRaiseTimestamp;
        }
        if (transition.TransitionTimestamp is not null)
        {
            snapshot.LastTransitionTimestamp = transition.TransitionTimestamp;
        }
        if (transition.CurrentValue is not null)
        {
            snapshot.CurrentValue = transition.CurrentValue;
        }
        if (transition.LimitValue is not null)
        {
            snapshot.LimitValue = transition.LimitValue;
        }

        return snapshot;
    }

    private static OnAlarmTransitionEvent TransitionFromSnapshot(
        ActiveAlarmSnapshot snapshot,
        AlarmTransitionKind kind)
    {
        OnAlarmTransitionEvent transition = new()
        {
            AlarmFullReference = snapshot.AlarmFullReference,
            SourceObjectReference = snapshot.SourceObjectReference,
            AlarmTypeName = snapshot.AlarmTypeName,
            TransitionKind = kind,
            Severity = snapshot.Severity,
            Category = snapshot.Category,
            Description = snapshot.Description,
            OperatorUser = snapshot.OperatorUser,
            OperatorComment = snapshot.OperatorComment,
            Degraded = snapshot.Degraded,
            SourceProvider = snapshot.SourceProvider,
        };
        if (snapshot.OriginalRaiseTimestamp is not null)
        {
            transition.OriginalRaiseTimestamp = snapshot.OriginalRaiseTimestamp;
        }
        if (snapshot.LastTransitionTimestamp is not null)
        {
            transition.TransitionTimestamp = snapshot.LastTransitionTimestamp;
        }
        if (snapshot.CurrentValue is not null)
        {
            transition.CurrentValue = snapshot.CurrentValue;
        }
        if (snapshot.LimitValue is not null)
        {
            transition.LimitValue = snapshot.LimitValue;
        }

        return transition;
    }

    private sealed class Subscriber(Channel<AlarmFeedMessage> channel, string prefix)
    {
        /// <summary>Gets the channel for publishing alarm messages to this subscriber.</summary>
        public Channel<AlarmFeedMessage> Channel { get; } = channel;

        /// <summary>Determines whether the alarm reference matches this subscriber's filter.</summary>
        /// <param name="reference">The alarm reference to match.</param>
        /// <returns><see langword="true"/> if the reference matches this subscriber's prefix filter.</returns>
        public bool Matches(string reference)
        {
            return prefix.Length == 0 || reference.StartsWith(prefix, StringComparison.Ordinal);
        }
    }
}