ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs

using Akka.Actor;
using Akka.Event;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;

namespace ZB.MOM.WW.ScadaBridge.Communication.Actors;

/// <summary>
/// Long-lived (one per active debug session) actor on the central side. Debug sessions
/// are session-based and temporary — this actor holds no persisted state and does not
/// derive from an Akka.Persistence base class; its state does not survive a restart.
/// <para>
/// <b>Stream-first lifecycle.</b> To avoid losing any
/// <see cref="AttributeValueChanged"/>/<see cref="AlarmStateChanged"/> that occurs on
/// the site during the snapshot-build + network-transit window, the gRPC server-streaming
/// subscription is opened FIRST (in <see cref="PreStart"/>), alongside the
/// <c>SubscribeDebugViewRequest</c> sent to the site via CentralCommunicationActor (with
/// THIS actor as the Sender). Live events that arrive before the
/// <see cref="DebugViewSnapshot"/> is delivered are <em>buffered in arrival order</em>.
/// When the snapshot arrives it is delivered to the consumer, then the buffer is flushed
/// in order, <em>deduped</em> against the snapshot (an event whose per-entity timestamp is
/// &lt;= the snapshot's timestamp for the same entity is already reflected → dropped; a
/// strictly-newer event is delivered; an event for an entity absent from the snapshot is
/// delivered). After the flush the actor switches to pass-through: subsequent events go
/// straight to the consumer. A mid-session reconnect (after the snapshot) resumes
/// pass-through — the snapshot is a one-time thing.
/// </para>
/// Stream events are marshalled back to the actor via Self.Tell for thread safety; all
/// state (phase flag + buffer) is mutated only on the actor thread.
/// </summary>
public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
{
    private readonly ILoggingAdapter _log = Context.GetLogger();
    private readonly string _siteIdentifier;
    private readonly string _instanceUniqueName;
    private readonly string _correlationId;
    private readonly IActorRef _centralCommunicationActor;
    private readonly Action<object> _onEvent;
    private readonly Action _onTerminated;
    private readonly SiteStreamGrpcClientFactory _grpcFactory;
    private readonly string _grpcNodeAAddress;
    private readonly string _grpcNodeBAddress;

    private const int MaxRetries = 3;
    private const string ReconnectTimerKey = "grpc-reconnect";
    private const string StabilityTimerKey = "grpc-stability";
    private const string SnapshotTimerKey = "debug-snapshot-deadline";
    private const string ConsumerLivenessTimerKey = "debug-consumer-liveness";
    /// <summary>Delay between gRPC reconnection attempts.</summary>
    internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);

    /// <summary>
    /// Hard deadline on the initial <see cref="DebugViewSnapshot"/> (WP2.3). The site builds
    /// it in milliseconds; if none arrives inside this window the site never answered (the
    /// Ask was lost, the singleton moved mid-request, the instance actor is wedged) and the
    /// session must FAIL rather than sit in the buffering phase accumulating live events
    /// behind a snapshot that is never coming. Settable for tests.
    /// </summary>
    internal static TimeSpan SnapshotTimeout { get; set; } = TimeSpan.FromSeconds(60);

    /// <summary>
    /// How long a freshly-opened gRPC stream must stay up before its retry budget
    /// is considered "recovered" and <see cref="_retryCount"/> is reset to 0.
    /// The retry count must NOT be reset by individual events —
    /// a stream that connects, delivers one event, then fails repeatedly would
    /// otherwise reconnect forever and never trip <see cref="MaxRetries"/>. Resetting
    /// only after a stable interval bounds a flapping stream.
    /// </summary>
    internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60);

    /// <summary>
    /// Orphan window: how long the session may go without ANY sign of life from its CONSUMER
    /// before it self-terminates. Renewed by <see cref="DebugStreamConsumerAlive"/>, which
    /// <c>DebugStreamService</c> Tells on a timer for every session still attached to a
    /// consumer (Blazor debug view or the SignalR hub) — so it measures the consumer, never
    /// the stream. Settable for tests.
    /// </summary>
    internal static TimeSpan ConsumerIdleTimeout { get; set; } = TimeSpan.FromMinutes(5);

    /// <summary>
    /// How often the actor checks the consumer-last-seen stamp against
    /// <see cref="ConsumerIdleTimeout"/>. A self-tick rather than <c>SetReceiveTimeout</c>:
    /// the receive timeout measures the MAILBOX, which conflates site chatter with consumer
    /// liveness — and once stream events were correctly excluded from it (via
    /// <see cref="LiveDebugStreamEvent"/>) nothing recurring reset it at all, so every healthy
    /// session self-terminated one window after its snapshot with a false "Site disconnected".
    /// Settable for tests.
    /// </summary>
    internal static TimeSpan ConsumerLivenessCheckInterval { get; set; } = TimeSpan.FromSeconds(30);

    /// <summary>
    /// When the consumer was last known to be attached (UTC). Seeded in <see cref="PreStart"/>
    /// so a session gets a full window to receive its first keepalive, then refreshed by every
    /// <see cref="DebugStreamConsumerAlive"/>. Actor-thread only.
    /// </summary>
    private DateTime _consumerLastSeenUtc = DateTime.UtcNow;

    private int _retryCount;
    private bool _useNodeA = true;
    private bool _stopped;
    private CancellationTokenSource? _grpcCts;

    /// <summary>
    /// Monotonic stream generation stamped on each opened gRPC stream and echoed back on its
    /// error/completion callbacks: a late callback raced out of a previous (cancelled) stream
    /// carries a stale generation and is ignored, so it can neither burn retry budget nor
    /// open a duplicate stream. Mirrors <c>SiteAlarmAggregatorActor</c>. Actor-thread only.
    /// </summary>
    private int _streamGeneration;

    /// <summary>
    /// Phase flag. <see langword="false"/> until the initial
    /// <see cref="DebugViewSnapshot"/> has been delivered and the pre-snapshot buffer
    /// flushed; <see langword="true"/> thereafter (pass-through). Mutated only on the
    /// actor thread. A reconnect does NOT touch this flag — a mid-session reconnect
    /// (after the snapshot) therefore stays in pass-through, and a reconnect during the
    /// buffering phase (before the snapshot) stays buffering.
    /// </summary>
    private bool _snapshotDelivered;

    /// <summary>
    /// Ordered buffer of live gRPC events (<see cref="AttributeValueChanged"/>/
    /// <see cref="AlarmStateChanged"/>) that arrived before the snapshot was delivered.
    /// Flushed (with per-entity dedup against the snapshot) when the snapshot arrives,
    /// then never used again. Bounded by <see cref="MaxPreSnapshotBuffer"/> with drop-oldest
    /// eviction (WP2.3): a snapshot that never arrives used to buffer without limit on the
    /// central node. Mutated only on the actor thread.
    /// </summary>
    private readonly Queue<object> _preSnapshotBuffer = new();

    /// <summary>
    /// Defensive log threshold: the first warning fires when the pre-snapshot buffer grows
    /// past this many events during a slow snapshot, before the hard cap starts evicting.
    /// </summary>
    private const int BufferWarnThreshold = 10_000;
    private bool _bufferWarned;

    /// <summary>
    /// Hard cap on the pre-snapshot buffer. Beyond it the OLDEST event is evicted — the
    /// snapshot that ends the buffering phase is authoritative for anything that old, so
    /// keeping the newest events is what preserves the post-snapshot delta chain.
    /// </summary>
    private const int MaxPreSnapshotBuffer = 20_000;

    /// <summary>Events evicted from the pre-snapshot buffer in this session. Actor-thread only.</summary>
    private long _preSnapshotDropped;

    /// <summary>
    /// Total pre-snapshot events dropped across all debug sessions on this node — the raw
    /// counter behind <c>scadabridge.central.debug_view.presnapshot_dropped</c>.
    /// </summary>
    internal static long TotalPreSnapshotDropped;

    /// <summary>Timer scheduler for reconnect and stability window timers.</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>
    /// Initializes the debug stream bridge actor and registers message handlers.
    /// </summary>
    /// <param name="siteIdentifier">Site identifier for targeting site-addressed messages and logging.</param>
    /// <param name="instanceUniqueName">Unique name of the instance whose debug stream is being bridged.</param>
    /// <param name="correlationId">Correlation id for the debug session.</param>
    /// <param name="centralCommunicationActor">Actor used to forward site-addressed messages to the site.</param>
    /// <param name="onEvent">Callback invoked on each received debug event.</param>
    /// <param name="onTerminated">Callback invoked when the stream terminates.</param>
    /// <param name="grpcFactory">Factory for creating gRPC streaming clients.</param>
    /// <param name="grpcNodeAAddress">gRPC address of the site's node A.</param>
    /// <param name="grpcNodeBAddress">gRPC address of the site's node B.</param>
    public DebugStreamBridgeActor(
        string siteIdentifier,
        string instanceUniqueName,
        string correlationId,
        IActorRef centralCommunicationActor,
        Action<object> onEvent,
        Action onTerminated,
        SiteStreamGrpcClientFactory grpcFactory,
        string grpcNodeAAddress,
        string grpcNodeBAddress)
    {
        _siteIdentifier = siteIdentifier;
        _instanceUniqueName = instanceUniqueName;
        _correlationId = correlationId;
        _centralCommunicationActor = centralCommunicationActor;
        _onEvent = onEvent;
        _onTerminated = onTerminated;
        _grpcFactory = grpcFactory;
        _grpcNodeAAddress = grpcNodeAAddress;
        _grpcNodeBAddress = grpcNodeBAddress;

        // Initial snapshot response from the site.
        // If the site reports InstanceNotFound=true the instance is not
        // deployed there. Under the stream-first lifecycle the gRPC stream
        // was already opened in PreStart, so the not-found path must tear it down
        // (CleanupGrpc) rather than enter pass-through.  Forward the snapshot (with
        // InstanceNotFound=true) to _onEvent so DebugStreamService's TCS resolves and
        // the caller can inspect the flag; then stop cleanly.
        Receive<DebugViewSnapshot>(snapshot =>
        {
            if (_snapshotDelivered)
            {
                // Defensive: a duplicate / late snapshot after we have already moved to
                // pass-through. The snapshot is a one-time thing — ignore replays so we
                // never re-buffer or double-deliver.
                _log.Debug("Ignoring duplicate DebugViewSnapshot for {0} (already delivered)",
                    _instanceUniqueName);
                return;
            }

            if (snapshot.InstanceNotFound)
            {
                _log.Warning("Instance {0} is not deployed on site; terminating debug stream",
                    _instanceUniqueName);
                // The stream-first subscription opened in PreStart is for a
                // non-deployed instance — cancel it (and any buffered gap events are
                // discarded with the actor). No pass-through.
                // _stopped is set AFTER CleanupGrpc() to match the ordering in the
                // DebugStreamTerminated and consumer-liveness handlers (cosmetic consistency).
                CleanupGrpc();
                _stopped = true;
                _preSnapshotBuffer.Clear();
                _onEvent(snapshot); // resolves the snapshot TCS with InstanceNotFound=true
                // Note: after Context.Stop(Self) below the actor is dead. DebugStreamService
                // inspects InitialSnapshot.InstanceNotFound and calls StopStream, which sends
                // a StopDebugStream message. That Tell arrives after the actor has already
                // stopped, producing a benign Akka dead-letter — expected and harmless.
                Context.Stop(Self);
                return;
            }

            _log.Info("Received initial snapshot for {0} ({1} attrs, {2} alarms); flushing {3} buffered event(s)",
                _instanceUniqueName, snapshot.AttributeValues.Count, snapshot.AlarmStates.Count,
                _preSnapshotBuffer.Count);

            // The snapshot arrived — stand the hard deadline down.
            Timers.Cancel(SnapshotTimerKey);

            // Deliver the snapshot, then flush the gap-window buffer (deduped), then
            // switch to pass-through. Order matters: snapshot first, buffered events next.
            _onEvent(snapshot);
            FlushBuffer(snapshot);
            _snapshotDelivered = true;
        });

        // Hard snapshot deadline (WP2.3). Nothing else ends a session stuck in the
        // buffering phase: the site's reply was lost, so no gRPC error fires, the stream
        // keeps delivering events, and stream traffic does not renew the orphan net (which
        // measures the consumer). Fail the session so the consumer is told and can reopen.
        Receive<DebugSnapshotDeadline>(_ =>
        {
            if (_stopped || _snapshotDelivered) return;
            _log.Error(
                "No debug snapshot for {0} within {1}s ({2} event(s) buffered, {3} dropped); failing the session",
                _instanceUniqueName, SnapshotTimeout.TotalSeconds,
                _preSnapshotBuffer.Count, _preSnapshotDropped);
            CleanupGrpc();
            SendUnsubscribe();
            _stopped = true;
            _preSnapshotBuffer.Clear();
            _onTerminated();
            Context.Stop(Self);
        });

        // Domain events arriving via Self.Tell from the gRPC callback. Stream traffic never
        // proves the CONSUMER is still there, so it deliberately does not touch the orphan
        // net (which now measures the consumer keepalive, not the mailbox). Receiving an
        // event must not reset _retryCount either: a flapping stream that delivers a single
        // event between failures would otherwise never trip MaxRetries. The retry budget is
        // recovered only by GrpcStreamStable (a stream that has stayed up for
        // StabilityWindow). Before the snapshot has been delivered, BUFFER (in arrival order)
        // rather than deliver — these may be gap-window events; after the snapshot has been
        // flushed, pass through directly (phase-dependent behavior).
        Receive<LiveDebugStreamEvent>(wrapped => HandleStreamEvent(wrapped.Event));

        // Unwrapped forms are still accepted (a direct Tell from a test or a future
        // in-process producer) and take the identical path.
        Receive<AttributeValueChanged>(changed => HandleStreamEvent(changed));
        Receive<AlarmStateChanged>(changed => HandleStreamEvent(changed));

        // Stream has been stably connected for StabilityWindow — recover the
        // retry budget so a future transient fault gets a fresh set of retries.
        Receive<GrpcStreamStable>(_ =>
        {
            if (_stopped) return;
            _retryCount = 0;
            _log.Debug("gRPC stream for {0} stable, retry count reset", _instanceUniqueName);
        });

        // gRPC stream error — attempt reconnection
        Receive<GrpcStreamError>(msg =>
        {
            // Ignore a late error raced out of a previous (cancelled) stream: it must not
            // burn retry budget or flip the node a second time.
            if (msg.Generation != _streamGeneration)
            {
                _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})",
                    msg.Generation, _streamGeneration);
                return;
            }
            _log.Warning("gRPC stream error for {0}: {1}", _instanceUniqueName, msg.Exception.Message);
            HandleGrpcError();
        });

        // gRPC stream ended GRACEFULLY (server status OK) — the site's 4h max stream
        // lifetime elapsing or a graceful site shutdown. Not a fault: reopen on the SAME
        // node without spending retry budget. Without this the session went silently deaf.
        Receive<GrpcStreamCompleted>(msg =>
        {
            if (_stopped) return;
            if (msg.Generation != _streamGeneration)
            {
                _log.Debug("Ignoring stale gRPC completion from stream generation {0} (current {1})",
                    msg.Generation, _streamGeneration);
                return;
            }
            HandleGrpcCompleted();
        });

        // Scheduled reconnection
        Receive<ReconnectGrpcStream>(_ => OpenGrpcStream());

        // Consumer requests stop
        Receive<StopDebugStream>(_ =>
        {
            _log.Info("Stopping debug stream for {0}", _instanceUniqueName);
            CleanupGrpc();
            SendUnsubscribe();
            _stopped = true;
            Context.Stop(Self);
        });

        // Site disconnected — CentralCommunicationActor notifies us
        Receive<DebugStreamTerminated>(msg =>
        {
            if (_stopped) return; // Idempotent — gRPC error may arrive simultaneously
            _log.Warning("Debug stream terminated for {0} (site {1} disconnected)", _instanceUniqueName, msg.SiteId);
            CleanupGrpc();
            _stopped = true;
            _onTerminated();
            Context.Stop(Self);
        });

        // Consumer keepalive: DebugStreamService Tells this on a timer for every session it
        // still holds (i.e. still attached to a Blazor debug view / SignalR connection). It
        // is the ONLY thing that renews the orphan window — deliberately, so neither a chatty
        // site nor a silent one can influence it.
        Receive<DebugStreamConsumerAlive>(_ =>
        {
            if (_stopped) return;
            _consumerLastSeenUtc = DateTime.UtcNow;
        });

        // Orphan safety net, CONSUMER-measured (WP2.3 follow-up). A periodic self-tick
        // compares the consumer-last-seen stamp against ConsumerIdleTimeout; the previous
        // SetReceiveTimeout(5 min) measured the mailbox instead, and once stream events were
        // (correctly) marked INotInfluenceReceiveTimeout nothing recurring reset it — the
        // snapshot arrives once and GrpcStreamStable once, so EVERY healthy session died at
        // ~6 minutes and the consumer was told "Site disconnected".
        Receive<ConsumerLivenessTick>(_ =>
        {
            if (_stopped) return;
            var idle = DateTime.UtcNow - _consumerLastSeenUtc;
            if (idle < ConsumerIdleTimeout) return;

            _log.Warning(
                "Debug stream for {0} has had no consumer activity for {1:F0}s (orphaned session), stopping",
                _instanceUniqueName, idle.TotalSeconds);
            Timers.Cancel(ConsumerLivenessTimerKey);
            CleanupGrpc();
            SendUnsubscribe();
            _stopped = true;
            _onTerminated();
            Context.Stop(Self);
        });
    }

    /// <summary>
    /// Handles a live gRPC stream event (<see cref="AttributeValueChanged"/> or
    /// <see cref="AlarmStateChanged"/>). Before the snapshot has been delivered the
    /// event is appended to the ordered pre-snapshot buffer (gap-window capture); after
    /// the snapshot+flush it is passed straight through to the consumer. Always runs on
    /// the actor thread (events are marshalled in via Self.Tell), so the phase flag and
    /// buffer are accessed without locking.
    /// </summary>
    private void HandleStreamEvent(object evt)
    {
        if (_snapshotDelivered)
        {
            _onEvent(evt);
            return;
        }

        if (!_bufferWarned && _preSnapshotBuffer.Count + 1 > BufferWarnThreshold)
        {
            _bufferWarned = true;
            _log.Warning(
                "Pre-snapshot debug-event buffer for {0} exceeded {1} events while awaiting the snapshot " +
                "(hard cap {2}, drop-oldest beyond it).",
                _instanceUniqueName, BufferWarnThreshold, MaxPreSnapshotBuffer);
        }

        while (_preSnapshotBuffer.Count >= MaxPreSnapshotBuffer)
        {
            _preSnapshotBuffer.Dequeue();
            _preSnapshotDropped++;
            Interlocked.Increment(ref TotalPreSnapshotDropped);
            ScadaBridgeTelemetry.RecordDebugPreSnapshotDrop();
            if (_preSnapshotDropped == 1 || _preSnapshotDropped % 500 == 0)
            {
                _log.Warning(
                    "Pre-snapshot debug-event buffer for {0} is at its {1}-event cap; {2} event(s) evicted so far",
                    _instanceUniqueName, MaxPreSnapshotBuffer, _preSnapshotDropped);
            }
        }

        _preSnapshotBuffer.Enqueue(evt);
    }

    /// <summary>
    /// Flushes the pre-snapshot buffer in arrival order, deduping each event against the
    /// just-delivered snapshot.
    /// <para>
    /// <b>Dedup rule.</b> Identity is per-entity:
    /// attributes by (InstanceUniqueName, AttributePath, AttributeName); alarms by
    /// (InstanceUniqueName, AlarmName, SourceReference). For a buffered event whose entity
    /// is present in the snapshot, the comparison is against that entity's snapshot
    /// timestamp: a buffered timestamp &lt;= the snapshot timestamp means the event is
    /// already reflected in the snapshot → DROP; a strictly-newer (&gt;) timestamp means
    /// the event happened after the snapshot was built → DELIVER. The boundary is inclusive
    /// on the snapshot side (equal timestamps are treated as duplicates) — the snapshot is
    /// the authoritative point-in-time value, so an event at the exact same instant carries
    /// no new information. A buffered event whose entity is NOT in the snapshot is a genuine
    /// gap-window event → DELIVER.
    /// </para>
    /// </summary>
    private void FlushBuffer(DebugViewSnapshot snapshot)
    {
        if (_preSnapshotBuffer.Count == 0) return;

        // Build per-entity "as-of" timestamps from the snapshot. If (defensively) the
        // snapshot lists the same entity twice, keep the newest timestamp.
        var attrAsOf = new Dictionary<string, DateTimeOffset>();
        foreach (var a in snapshot.AttributeValues)
        {
            var key = AttributeKey(a);
            if (!attrAsOf.TryGetValue(key, out var existing) || a.Timestamp > existing)
                attrAsOf[key] = a.Timestamp;
        }

        var alarmAsOf = new Dictionary<string, DateTimeOffset>();
        foreach (var al in snapshot.AlarmStates)
        {
            var key = AlarmKey(al);
            if (!alarmAsOf.TryGetValue(key, out var existing) || al.Timestamp > existing)
                alarmAsOf[key] = al.Timestamp;
        }

        var flushed = 0;
        var dropped = 0;
        foreach (var evt in _preSnapshotBuffer)
        {
            if (IsReflectedInSnapshot(evt, attrAsOf, alarmAsOf))
            {
                dropped++;
                continue;
            }

            _onEvent(evt);
            flushed++;
        }

        if (dropped > 0 || flushed > 0)
        {
            _log.Debug("Flushed {0} buffered debug event(s) for {1}, dropped {2} as already-in-snapshot" +
                       " ({3} previously evicted at the buffer cap)",
                flushed, _instanceUniqueName, dropped, _preSnapshotDropped);
        }

        _preSnapshotBuffer.Clear();
    }

    /// <summary>
    /// Returns <see langword="true"/> when a buffered event is already reflected in the
    /// snapshot (same entity, buffered timestamp &lt;= snapshot timestamp) and must be
    /// dropped; otherwise <see langword="false"/> (deliver).
    /// </summary>
    private static bool IsReflectedInSnapshot(
        object evt,
        IReadOnlyDictionary<string, DateTimeOffset> attrAsOf,
        IReadOnlyDictionary<string, DateTimeOffset> alarmAsOf)
    {
        switch (evt)
        {
            case AttributeValueChanged a:
                return attrAsOf.TryGetValue(AttributeKey(a), out var attrTs) && a.Timestamp <= attrTs;
            case AlarmStateChanged al:
                return alarmAsOf.TryGetValue(AlarmKey(al), out var alarmTs) && al.Timestamp <= alarmTs;
            default:
                // Unknown buffered type (should not happen — only attr/alarm are buffered):
                // never treat as a duplicate.
                return false;
        }
    }

    /// <summary>
    /// Delimiter used to join identity components into a single dedup key. A NUL
    /// control character cannot appear in an instance/attribute/alarm name, so
    /// distinct identities never collide on a shared boundary (unlike a space, which
    /// may legitimately occur within a name). Declared as an escaped char so the
    /// source carries no raw NUL byte.
    /// </summary>
    private const char KeyDelimiter = '\u0000';

    /// <summary>
    /// Per-entity dedup key for an attribute change. Each nullable component is guarded
    /// with <c>?? string.Empty</c> so a null can never silently collide with another
    /// key via <see cref="string.Concat"/> (e.g. two entries with null AttributePath
    /// would otherwise share a key with any entry whose AttributePath is the empty string).
    /// </summary>
    private static string AttributeKey(AttributeValueChanged a) =>
        string.Concat(
            a.InstanceUniqueName ?? string.Empty, KeyDelimiter,
            a.AttributePath ?? string.Empty, KeyDelimiter,
            a.AttributeName ?? string.Empty);

    /// <summary>
    /// Per-entity dedup key for an alarm change. Includes <see cref="AlarmStateChanged.SourceReference"/>
    /// so native per-condition alarms (which share an AlarmName but differ by source
    /// reference) are not conflated; empty for computed alarms. Each nullable component is
    /// guarded with <c>?? string.Empty</c> to prevent silent key collisions.
    /// </summary>
    private static string AlarmKey(AlarmStateChanged al) =>
        string.Concat(
            al.InstanceUniqueName ?? string.Empty, KeyDelimiter,
            al.AlarmName ?? string.Empty, KeyDelimiter,
            al.SourceReference ?? string.Empty);

    /// <inheritdoc />
    protected override void PreStart()
    {
        _log.Info("Starting debug stream bridge for {0} on site {1}", _instanceUniqueName, _siteIdentifier);

        // Stream-first: open the gRPC live-event subscription BEFORE (and
        // alongside) requesting the snapshot, so events occurring during the
        // snapshot-build + network-transit window are captured (buffered) and not lost.
        OpenGrpcStream();

        // Send subscribe request via CentralCommunicationActor for the initial snapshot.
        var request = new SubscribeDebugViewRequest(_instanceUniqueName, _correlationId);
        var envelope = new SiteEnvelope(_siteIdentifier, request);
        _centralCommunicationActor.Tell(envelope, Self);

        // Arm the hard snapshot deadline alongside the request.
        if (SnapshotTimeout > TimeSpan.Zero)
            Timers.StartSingleTimer(SnapshotTimerKey, new DebugSnapshotDeadline(), SnapshotTimeout);

        // Arm the consumer-liveness net. The stamp starts now, so the session always gets a
        // full ConsumerIdleTimeout to see its first keepalive from DebugStreamService.
        _consumerLastSeenUtc = DateTime.UtcNow;
        if (ConsumerIdleTimeout > TimeSpan.Zero && ConsumerLivenessCheckInterval > TimeSpan.Zero)
        {
            Timers.StartPeriodicTimer(
                ConsumerLivenessTimerKey, new ConsumerLivenessTick(), ConsumerLivenessCheckInterval);
        }
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = null;
        base.PostStop();
    }

    private void OpenGrpcStream()
    {
        if (_stopped) return;

        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        _log.Info("Opening gRPC stream for {0} to {1}", _instanceUniqueName, endpoint);

        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = new CancellationTokenSource();

        // Arm the stability timer: if the stream stays up for StabilityWindow the
        // retry budget is recovered. Cancelled by HandleGrpcError.
        Timers.StartSingleTimer(StabilityTimerKey, new GrpcStreamStable(), StabilityWindow);

        var generation = ++_streamGeneration;
        var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint);
        var self = Self;
        var ct = _grpcCts.Token;

        // Launch as background task — the callbacks marshal back to the actor via Tell.
        // The task itself is observed below: a fault escaping SubscribeAsync would otherwise
        // leave the session waiting on a stream that does not exist, exception unobserved.
        Task.Run(async () =>
        {
            await client.SubscribeAsync(
                _correlationId,
                _instanceUniqueName,
                // Wrapped so the stream path is explicit (it never renews the orphan net).
                evt => self.Tell(new LiveDebugStreamEvent(evt)),
                ex => self.Tell(new GrpcStreamError(ex, generation)),
                () => self.Tell(new GrpcStreamCompleted(generation)),
                ct);
        }, ct).ContinueWith(t =>
        {
            if (t.IsFaulted)
                self.Tell(new GrpcStreamError(t.Exception!.GetBaseException(), generation));
            else if (t.IsCanceled && !ct.IsCancellationRequested)
                self.Tell(new GrpcStreamCompleted(generation));
            // RanToCompletion: SubscribeAsync already reported its own outcome.
        }, TaskContinuationOptions.ExecuteSynchronously);
    }

    /// <summary>
    /// Handles a graceful end of stream (server status OK). The stream simply expired or the
    /// site shut down cleanly, so the retry budget is left untouched and the endpoint is not
    /// flipped; the reopen is scheduled through the existing reconnect timer, which also
    /// rate-limits a pathological site that keeps closing streams immediately.
    /// </summary>
    private void HandleGrpcCompleted()
    {
        // The stream is gone, so its armed stability timer must not later "recover" a
        // budget that its successor has since spent.
        Timers.Cancel(StabilityTimerKey);

        _log.Info("gRPC stream for {0} completed gracefully (server end of stream); reopening",
            _instanceUniqueName);

        // Release the site-side relay for the finished stream before reopening, so the site
        // is not left with a zombie relay actor for this correlation id.
        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = null;
        _grpcFactory.TryGet(_siteIdentifier, _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress)
            ?.Unsubscribe(_correlationId);

        Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectGrpcStream(), ReconnectDelay);
    }

    private void HandleGrpcError()
    {
        if (_stopped) return;

        // The stream failed before reaching the stability window — its retry
        // budget is NOT recovered.
        Timers.Cancel(StabilityTimerKey);

        _retryCount++;

        if (_retryCount > MaxRetries)
        {
            _log.Error("gRPC stream for {0} exceeded max retries ({1}), terminating", _instanceUniqueName, MaxRetries);
            CleanupGrpc();
            _stopped = true;
            _onTerminated();
            Context.Stop(Self);
            return;
        }

        // Unsubscribe the failed stream on the *previous* endpoint before reconnecting.
        // This cancels the local subscription CTS and -- where the channel is still
        // alive -- propagates gRPC cancellation to the site so its SiteStreamGrpcServer
        // stops the StreamRelayActor for this correlation ID, rather than leaving a
        // zombie relay actor until TCP RST / keepalive eventually detects the loss.
        var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        // TryGet, not GetOrCreate: unsubscribing a failed stream must never open a
        // fresh channel (and, with (site,endpoint) keying, must never touch another
        // session's healthy channel). Absent client => the channel is already gone
        // and the site-side relay will be reaped by keepalive/session-lifetime.
        _grpcFactory.TryGet(_siteIdentifier, previousEndpoint)?.Unsubscribe(_correlationId);

        // Flip to the other node
        _useNodeA = !_useNodeA;

        // First retry is immediate, subsequent retries use a short backoff
        if (_retryCount == 1)
        {
            Self.Tell(new ReconnectGrpcStream());
        }
        else
        {
            Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectGrpcStream(), ReconnectDelay);
        }
    }

    private void CleanupGrpc()
    {
        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = null;

        // TryGet, not GetOrCreate: teardown must never open a fresh channel just to
        // unsubscribe. Absent client => nothing to cancel.
        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId);
    }

    private void SendUnsubscribe()
    {
        var request = new UnsubscribeDebugViewRequest(_instanceUniqueName, _correlationId);
        var envelope = new SiteEnvelope(_siteIdentifier, request);
        _centralCommunicationActor.Tell(envelope, Self);
    }
}

/// <summary>
/// Message sent to a DebugStreamBridgeActor to stop the debug stream session.
/// </summary>
public record StopDebugStream;

/// <summary>
/// Envelope for a live gRPC stream event (<c>AttributeValueChanged</c>/
/// <c>AlarmStateChanged</c>). Kept as a distinct envelope so the high-volume stream path is
/// explicit at the call site; the orphan net no longer keys off the mailbox at all (it
/// measures the consumer keepalive), so a busy site's event flood can neither hold a dead
/// session open nor — as briefly happened — be the only thing keeping a healthy one alive.
/// </summary>
internal record LiveDebugStreamEvent(object Event);

/// <summary>
/// Consumer keepalive: <c>DebugStreamService</c> Tells one of these to every bridge actor
/// whose session is still attached to a consumer, on
/// <see cref="DebugStreamBridgeActor.ConsumerLivenessCheckInterval"/>-scale cadence. Renewing
/// the consumer-last-seen stamp is its ONLY effect.
/// </summary>
public record DebugStreamConsumerAlive;

/// <summary>
/// Internal self-tick that checks the consumer-last-seen stamp against
/// <see cref="DebugStreamBridgeActor.ConsumerIdleTimeout"/>.
/// </summary>
internal record ConsumerLivenessTick;

/// <summary>
/// Internal message: the hard deadline for the initial <c>DebugViewSnapshot</c> expired.
/// </summary>
internal record DebugSnapshotDeadline;

/// <summary>
/// Internal message indicating a gRPC stream error occurred, stamped with the stream
/// generation it came from so a late error out of a cancelled stream can be ignored.
/// </summary>
internal record GrpcStreamError(Exception Exception, int Generation);

/// <summary>
/// Internal message indicating the gRPC stream ended gracefully (server status OK — the
/// site's max stream lifetime elapsed, or the site shut down cleanly), stamped with its
/// stream generation.
/// </summary>
internal record GrpcStreamCompleted(int Generation);

/// <summary>
/// Internal message to trigger gRPC stream reconnection.
/// </summary>
internal record ReconnectGrpcStream;

/// <summary>
/// Internal message indicating the current gRPC stream has been connected long
/// enough (<see cref="DebugStreamBridgeActor.StabilityWindow"/>) to be considered
/// stable, so the reconnect retry budget can be recovered.
/// </summary>
internal record GrpcStreamStable;