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

using Akka.Actor;
using Akka.Event;
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>
/// One-per-site aggregator on the active central node backing the operator Alarm
/// Summary live cache (plan #10, Task 4). Holds a transient in-memory
/// <c>Dictionary&lt;AlarmKey, AlarmStateChanged&gt;</c> of the whole site's current
/// alarm state — NO persistence (locked <c>[PERM]</c>: no central alarm store). Created
/// and torn down by <see cref="SiteAlarmLiveCacheService"/> under viewer reference-count.
/// <para>
/// <b>Seed-then-stream ordering</b> (copied from <see cref="DebugStreamBridgeActor"/>):
/// the site-wide, alarm-only gRPC stream (<c>SubscribeSite</c>) is opened FIRST in
/// <see cref="PreStart"/> so live deltas start flowing during the seed's snapshot-build
/// + transit window; deltas arriving while a seed/reconcile fan-out is in flight are
/// <em>buffered in arrival order</em>. When the fan-out completes the cache is rebuilt
/// authoritatively from the fresh snapshot, then the buffer is flushed with per-key
/// dedup against the seed, then the actor passes live deltas straight into the cache.
/// </para>
/// <para>
/// <b>Placeholder reconciliation:</b> the snapshot fan-out carries
/// <see cref="AlarmStateChanged.IsConfiguredPlaceholder"/> rows (a configured native
/// source binding with no active conditions); the live stream drops them server-side.
/// Placeholders are seeded and never expected on the live stream — a live delta for a
/// real alarm has a distinct <c>AlarmKey</c> so it can never wipe a placeholder row and
/// vice-versa. Placeholder-vs-real coherence is refreshed by the periodic reconcile.
/// </para>
/// <para>
/// <b>Failover + drift:</b> a gRPC error flips NodeA↔NodeB with the same retry budget +
/// stability window as <see cref="DebugStreamBridgeActor"/>. A re-seed runs <b>once per
/// successful (re)connect</b> — driven by the stream's connected callback, not by each
/// reconnect ATTEMPT (WP2.3: a site outage used to fan a full snapshot out per retry, all
/// of them against the very site that is unreachable). A periodic reconcile snapshot
/// (<see cref="_reconcileInterval"/>, default 60s, jittered per site so N aggregators do
/// not fan out in lockstep) remains the drift/backstop: it corrects instance-set drift and
/// any missed delta, but it is <em>skipped</em> when a fan-out already completed inside the
/// window, and it publishes to viewers only when the snapshot actually changed the cache.
/// </para>
/// All state is mutated only on the actor thread: gRPC callbacks and fan-out results are
/// marshalled back via <c>Self.Tell</c>, so the cache needs no internal lock. The
/// published snapshot handed to the service is a fresh immutable list (reference swap),
/// so Blazor render threads never observe a partially-mutated cache.
/// </summary>
public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
{
    private readonly ILoggingAdapter _log = Context.GetLogger();
    private readonly string _siteIdentifier;
    private readonly string _correlationId;
    private readonly Func<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> _seedFn;
    private readonly Action<IReadOnlyList<AlarmStateChanged>, bool> _publish;
    private readonly SiteStreamGrpcClientFactory _grpcFactory;
    private readonly string _grpcNodeAAddress;
    private readonly string _grpcNodeBAddress;
    private readonly TimeSpan _reconcileInterval;
    private readonly TimeSpan _publishCoalesce;
    private readonly double _reconcileJitterFraction;

    private const int MaxRetries = 3;
    private const string ReconnectTimerKey = "alarm-grpc-reconnect";
    private const string StabilityTimerKey = "alarm-grpc-stability";
    private const string ReconcileTimerKey = "alarm-reconcile";
    private const string PublishTimerKey = "alarm-publish-coalesce";
    private const string SeedRetryTimerKey = "alarm-seed-retry";

    /// <summary>True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only.</summary>
    private bool _publishPending;

    /// <summary>Delay between gRPC reconnection attempts (ctor-injected; production default 5s).</summary>
    private readonly TimeSpan _reconnectDelay;

    /// <summary>
    /// How long a freshly-opened gRPC stream must stay up before its retry budget is
    /// considered recovered (mirrors <see cref="DebugStreamBridgeActor.StabilityWindow"/>);
    /// ctor-injected (production default 60s).
    /// </summary>
    private readonly TimeSpan _stabilityWindow;

    private int _retryCount;
    private bool _useNodeA = true;
    private bool _stopped;

    /// <summary>
    /// True while there is no live gRPC stream — either it was given up (retry budget
    /// exhausted) or it ended gracefully (server status OK at the site's max stream
    /// lifetime). Reconcile snapshots keep serving in the meantime; the next reconcile tick
    /// self-heals the stream by resetting the retry budget and reopening it, so neither a
    /// sustained site outage nor a routine 4h stream expiry permanently drops the live feed.
    /// Starts <c>true</c>: until the site accepts the first subscription there is no live
    /// feed, and the owning cache must not advertise one. Actor-thread only.
    /// </summary>
    private bool _streamDown = true;

    /// <summary>
    /// Why the stream is down: <c>true</c> = the retry budget was exhausted, so the
    /// self-healing reopen must also reset it; <c>false</c> = it ended gracefully, and the
    /// budget — which a completion neither spends nor refunds — is carried across the reopen
    /// untouched (a stream flapping between faults and clean closes must still trip
    /// MaxRetries). Actor-thread only.
    /// </summary>
    private bool _retryBudgetExhausted;
    private CancellationTokenSource? _grpcCts;
    private CancellationTokenSource? _lifetimeCts;

    /// <summary>Current whole-site alarm state, keyed by <see cref="AlarmKey"/>. Actor-thread only.</summary>
    private readonly Dictionary<string, AlarmStateChanged> _cache = new();

    /// <summary>True once the first seed has completed and been published. Actor-thread only.</summary>
    private bool _seeded;

    /// <summary>True while a seed/reconcile snapshot fan-out is in flight (deltas buffer). Actor-thread only.</summary>
    private bool _fanoutInFlight;

    /// <summary>
    /// Ordered buffer of live deltas that arrived while a fan-out was in flight, capped at
    /// <see cref="MaxBufferedDeltas"/> with drop-oldest eviction (WP2.3 — a stuck fan-out
    /// used to buffer without bound). Dropping the OLDEST is the right eviction here: the
    /// fan-out that follows rebuilds the cache authoritatively, so an evicted delta is
    /// superseded rather than lost, while the newest transitions — the ones the snapshot may
    /// predate — are the ones kept. Actor-thread only.
    /// </summary>
    private readonly Queue<AlarmStateChanged> _buffer = new();

    /// <summary>Total deltas evicted from <see cref="_buffer"/> over this actor's life. Actor-thread only.</summary>
    private long _bufferDropped;

    /// <summary>
    /// Hard cap on the pre-fan-out delta buffer. Beyond this the oldest entry is evicted and
    /// counted; the counter is logged (first drop + every 500th) and exported as
    /// <c>scadabridge.site.alarm_cache.buffer_dropped</c>.
    /// </summary>
    private const int MaxBufferedDeltas = 20_000;

    /// <summary>
    /// True when the next successful (re)connect must run a seed fan-out: set whenever the
    /// stream is lost (fault or graceful completion) and cleared by the connect that consumes
    /// it. This is what makes the re-seed happen once per successful reconnect rather than
    /// once per reconnect ATTEMPT — the old code fanned a full snapshot out on every retry,
    /// against a site that was by definition unreachable. Actor-thread only.
    /// </summary>
    private bool _seedOnConnect;

    /// <summary>
    /// Set when a CONNECT- OR FAILOVER-DRIVEN fan-out finishes (the initial seed, a re-seed
    /// consumed by a successful (re)connect, or a re-seed queued behind one); consumed by the
    /// next reconcile tick, which skips its own fan-out when it finds the flag set. That makes
    /// the connect-driven seed and the periodic backstop mutually exclusive — the pair used to
    /// run BOTH, which is the "full unconditional snapshot every 60s" waste — while bounding
    /// staleness at two intervals (a tick can be skipped at most once in a row).
    /// <para>
    /// A TICK-driven fan-out deliberately does NOT set it. It used to: the tick's own fan-out
    /// completion armed the skip, so steady state ran fan-out → skip → fan-out → skip, i.e.
    /// one reconcile per TWO intervals — halving the not-reporting refresh rate and the alarm
    /// reconcile backstop for no reason (nothing was duplicated to suppress).
    /// </para>
    /// Actor-thread only.
    /// </summary>
    private bool _fanoutSinceLastTick;

    /// <summary>
    /// Whether the in-flight fan-out is connect/failover-driven and therefore makes the next
    /// reconcile tick redundant (see <see cref="_fanoutSinceLastTick"/>). Stable for the
    /// fan-out's lifetime: <see cref="StartFanout"/> never overwrites it while one is in
    /// flight (a colliding request queues instead). Actor-thread only.
    /// </summary>
    private bool _fanoutSuppressesNextTick;

    /// <summary>
    /// Same flag for a fan-out that was queued behind an in-flight one
    /// (<see cref="_reseedQueued"/>) — OR-ed across colliding requests so a queued
    /// connect/failover re-seed keeps its suppression even if a tick collided too.
    /// Actor-thread only.
    /// </summary>
    private bool _queuedFanoutSuppressesNextTick;

    /// <summary>
    /// True between issuing a stream open and its first outcome (connected / error /
    /// completed), so a reconcile tick never stacks a second open on an in-flight one.
    /// Actor-thread only.
    /// </summary>
    private bool _openInFlight;

    /// <summary>Consecutive whole-fan-out failures, driving the seed-retry backoff. Actor-thread only.</summary>
    private int _consecutiveSeedFailures;

    /// <summary>Upper bound on the seed-retry backoff, as a multiple of the reconcile interval.</summary>
    private const int MaxSeedRetryBackoffMultiplier = 8;

    /// <summary>
    /// A failover re-seed was requested while a fan-out was already in flight; it must run
    /// right after the in-flight one completes rather than being silently dropped (N7.1) —
    /// the in-flight snapshot's read-time predates the stream death. Actor-thread only.
    /// </summary>
    private bool _reseedQueued;

    /// <summary>
    /// Monotonic stream generation stamped on each opened gRPC stream and echoed back on its
    /// error callback: a late error raced out of a previous (cancelled) stream carries a stale
    /// generation and is ignored so it never burns retry budget or double-flips (N7.2).
    /// Actor-thread only.
    /// </summary>
    private int _streamGeneration;

    private const int BufferWarnThreshold = 10_000;
    private bool _bufferWarned;

    /// <inheritdoc />
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>
    /// Creates a per-site alarm aggregator.
    /// </summary>
    /// <param name="siteIdentifier">Site identifier (for logging / gRPC client keying).</param>
    /// <param name="correlationId">Correlation id for the site-wide gRPC subscription.</param>
    /// <param name="seedFn">
    /// Snapshot fan-out that returns the whole site's current alarm rows (including
    /// placeholders), best-effort and tolerant of per-instance failure. Re-run on every
    /// seed and reconcile, so it re-enumerates the site's Enabled instances each call.
    /// </param>
    /// <param name="publish">
    /// Publishes a fresh immutable snapshot of the cache to the owning service, which
    /// stores it and raises viewer <c>onChanged</c> callbacks. Invoked on the actor thread.
    /// The second argument is the live-stream liveness flag: <c>false</c> means the site-wide
    /// gRPC stream is currently down (fault-exhausted or gracefully completed), so the cache
    /// is only as fresh as the last reconcile and the page must fall back to polling
    /// (WP2.3 carried residual — <c>IsLive</c> used to stay true across a dead stream).
    /// </param>
    /// <param name="grpcFactory">Factory caching one gRPC client per (site, endpoint).</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>
    /// <param name="reconcileInterval">Periodic reconcile snapshot cadence.</param>
    /// <param name="publishCoalesce">
    /// Publish-coalescing window for live deltas: a positive value batches a delta storm
    /// into one publish per window (review 02 round 2, N6); <see cref="TimeSpan.Zero"/>
    /// restores per-delta publishing (legacy). Seed/reconcile publishes stay immediate.
    /// </param>
    /// <param name="reconnectDelay">Delay between gRPC reconnection attempts (production 5s).</param>
    /// <param name="stabilityWindow">
    /// How long a fresh gRPC stream must stay up before its retry budget recovers (production 60s).
    /// </param>
    /// <param name="reconcileJitterFraction">
    /// Fraction of <paramref name="reconcileInterval"/> added as a per-tick random offset so
    /// N per-site aggregators on one central node do not fan out in lockstep (production 0.2 =
    /// up to +20%). Zero disables jitter, which is what tests want for determinism.
    /// </param>
    public SiteAlarmAggregatorActor(
        string siteIdentifier,
        string correlationId,
        Func<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> seedFn,
        Action<IReadOnlyList<AlarmStateChanged>, bool> publish,
        SiteStreamGrpcClientFactory grpcFactory,
        string grpcNodeAAddress,
        string grpcNodeBAddress,
        TimeSpan reconcileInterval,
        TimeSpan publishCoalesce,
        TimeSpan reconnectDelay,
        TimeSpan stabilityWindow,
        double reconcileJitterFraction = 0.0)
    {
        _siteIdentifier = siteIdentifier;
        _correlationId = correlationId;
        _seedFn = seedFn;
        _publish = publish;
        _grpcFactory = grpcFactory;
        _grpcNodeAAddress = grpcNodeAAddress;
        _grpcNodeBAddress = grpcNodeBAddress;
        _reconcileInterval = reconcileInterval;
        _publishCoalesce = publishCoalesce;
        _reconnectDelay = reconnectDelay;
        _stabilityWindow = stabilityWindow;
        _reconcileJitterFraction = Math.Clamp(reconcileJitterFraction, 0.0, 1.0);

        // Live delta from the site-wide alarm stream (marshalled in via Self.Tell).
        // A received delta must NOT reset the retry budget (a flapping stream that
        // delivers one delta between failures would otherwise never trip MaxRetries).
        Receive<AlarmStateChanged>(HandleLiveDelta);

        // A seed/reconcile fan-out completed.
        Receive<SeedCompleted>(OnSeedCompleted);

        // A seed/reconcile fan-out threw as a whole (individual per-instance faults are
        // swallowed inside seedFn and degrade to fewer rows, not a whole-fan-out failure).
        Receive<SeedFailed>(OnSeedFailed);

        // Periodic reconcile tick (and the re-seed kicked after a reconnect).
        Receive<RunReconcile>(_ => OnReconcileTick());

        // Backoff-scheduled retry of a fan-out that failed as a whole.
        Receive<RetrySeed>(_ =>
        {
            if (_stopped) return;
            // A backoff retry of a failed fan-out is not a connect/failover seed: it must
            // not stand the next reconcile tick down.
            StartFanout(isInitial: false, suppressesNextTick: false);
        });

        // The site-wide stream is confirmed established (response headers received). This —
        // not the reconnect attempt — is the "successful (re)connect" that earns one re-seed.
        Receive<GrpcAlarmStreamConnected>(msg =>
        {
            if (_stopped) return;
            if (msg.Generation != _streamGeneration) return;
            OnStreamConnected();
        });

        // Coalesced-publish tick: one publish for a batch of dirtying deltas (N6).
        Receive<PublishCoalesced>(_ =>
        {
            _publishPending = false;
            if (!_stopped) Publish();
        });

        // Stream stayed up for StabilityWindow — recover the retry budget.
        Receive<GrpcAlarmStreamStable>(_ =>
        {
            if (_stopped) return;
            _retryCount = 0;
            _log.Debug("Site-alarm gRPC stream for {0} stable; retry count reset", _siteIdentifier);
        });

        // gRPC stream error — flip node + reconnect + re-seed.
        Receive<GrpcAlarmStreamError>(msg =>
        {
            // Ignore a late error raced out of a previous (cancelled) stream — the
            // RpcException(Cancelled) filter at SiteStreamGrpcClient.cs covers the normal
            // path, but a genuine socket fault can beat the cancel (N7.2).
            if (msg.Generation != _streamGeneration)
            {
                _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})",
                    msg.Generation, _streamGeneration);
                return;
            }
            _log.Warning("Site-alarm gRPC stream error for {0}: {1}", _siteIdentifier, 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: the stream is marked
        // down so the reconcile tick reopens it (which also re-seeds), but the retry budget
        // is untouched and the node is not flipped. Same generation fence as the error path.
        Receive<GrpcAlarmStreamCompleted>(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();
        });

        Receive<ReconnectAlarmStream>(_ => OpenGrpcStream());

        // Owning service asks us to stop (last viewer left + linger elapsed).
        Receive<StopSiteAlarmAggregator>(_ =>
        {
            _log.Info("Stopping site-alarm aggregator for {0}", _siteIdentifier);
            CleanupGrpc();
            _stopped = true;
            Context.Stop(Self);
        });
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        _log.Info("Starting site-alarm aggregator for site {0}", _siteIdentifier);
        // Telemetry: this aggregator is now a running per-site live cache (gauge +1). Balanced
        // in PostStop, which Akka always runs on termination for any reason.
        ScadaBridgeTelemetry.LiveAlarmAggregatorStarted();
        _lifetimeCts = new CancellationTokenSource();

        // Stream-first: open the site-wide alarm stream BEFORE the first seed so deltas
        // in the seed window are captured (buffered) rather than lost.
        OpenGrpcStream();

        // Kick the initial seed fan-out. The connect callback for this first stream must
        // NOT seed again on top of it, so the flag starts cleared.
        _seedOnConnect = false;
        StartFanout(isInitial: true, suppressesNextTick: true);

        // Periodic reconcile backstop. Single-shot and re-armed with fresh jitter each
        // tick (a periodic timer would lock every site to the same phase forever).
        ArmReconcileTimer();
    }

    /// <summary>
    /// Arms the next reconcile tick at <see cref="_reconcileInterval"/> plus a random
    /// offset of up to <see cref="_reconcileJitterFraction"/> of it, so N per-site
    /// aggregators started together (a central failover restarts them all at once) spread
    /// their fan-outs instead of stampeding the same 60s boundary.
    /// </summary>
    private void ArmReconcileTimer()
    {
        var delay = _reconcileInterval;
        if (_reconcileJitterFraction > 0 && _reconcileInterval > TimeSpan.Zero)
        {
            var jitterTicks = (long)(_reconcileInterval.Ticks * _reconcileJitterFraction * Random.Shared.NextDouble());
            delay += TimeSpan.FromTicks(jitterTicks);
        }
        Timers.StartSingleTimer(ReconcileTimerKey, new RunReconcile(), delay);
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = null;
        _lifetimeCts?.Cancel();
        _lifetimeCts?.Dispose();
        _lifetimeCts = null;
        // Telemetry: this aggregator is no longer running (gauge -1). Balances PreStart.
        ScadaBridgeTelemetry.LiveAlarmAggregatorStopped();
        base.PostStop();
    }

    // ── Reconcile tick ──────────────────────────────────────────────────────────

    /// <summary>
    /// Periodic reconcile backstop: re-run the snapshot fan-out (corrects instance-set drift
    /// + any missed delta) UNLESS a CONNECT- OR FAILOVER-DRIVEN seed already completed inside
    /// this window — that seed makes the tick redundant, and running both was the "full
    /// unconditional snapshot every 60s" waste (WP2.3). A tick's OWN fan-out never arms that
    /// skip: it used to, which made steady state alternate fan-out/skip and halved the
    /// effective reconcile rate. If the live stream was previously given up, self-heal it by
    /// resetting the retry budget and reopening, so a sustained outage never permanently kills
    /// the live feed.
    /// </summary>
    private void OnReconcileTick()
    {
        if (_stopped) return;

        // Re-arm first: every exit path below must leave the backstop running.
        ArmReconcileTimer();

        // A down stream is reopened in preference to fanning out: the reopen's connect
        // callback seeds by itself, so doing both would double the work.
        var reopening = false;
        if (_streamDown && !_openInFlight)
        {
            _log.Info("Site-alarm gRPC stream for {0} was down; reopening on reconcile tick", _siteIdentifier);
            if (_retryBudgetExhausted)
            {
                _retryBudgetExhausted = false;
                _retryCount = 0;
            }
            // Telemetry: a reconcile-driven reopen after the stream was given up is a reconnect.
            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
            _seedOnConnect = true;
            OpenGrpcStream();
            reopening = true;
        }

        if (_fanoutSinceLastTick)
        {
            // A connect-driven seed (or the initial seed) already refreshed the cache inside
            // this window. Clear the flag so the NEXT tick fans out regardless — staleness
            // stays bounded at two intervals even if reconnects keep arriving.
            _fanoutSinceLastTick = false;
            _log.Debug("Site-alarm reconcile for {0} skipped; a fan-out already ran this window",
                _siteIdentifier);
            return;
        }

        if (!reopening)
        {
            // Tick-driven: deliberately does NOT set the skip flag, so the NEXT tick fans
            // out too. Setting it here was the "one reconcile per two intervals" bug.
            StartFanout(isInitial: false, suppressesNextTick: false);
        }
    }

    // ── Seed / reconcile fan-out ────────────────────────────────────────────────

    /// <summary>
    /// Kicks a snapshot fan-out as a background task, marshalling the result back via
    /// <c>Self.Tell</c>. While in flight, live deltas buffer. A reconcile that arrives
    /// while a fan-out is already running is skipped (no stacking).
    /// </summary>
    private void StartFanout(bool isInitial, bool suppressesNextTick)
    {
        if (_stopped) return;
        if (_fanoutInFlight)
        {
            // A failover re-seed requested mid-fan-out must run right after the in-flight
            // one — its snapshot read-time predates the stream death, so skipping it would
            // serve stale up to the next 60s reconcile (N7.1). An initial-seed collision
            // never queues (there is only ever one).
            if (!isInitial)
            {
                _reseedQueued = true;
                _queuedFanoutSuppressesNextTick |= suppressesNextTick;
            }
            return;
        }

        _fanoutInFlight = true;
        _fanoutSuppressesNextTick = suppressesNextTick;
        var self = Self;
        var ct = _lifetimeCts?.Token ?? CancellationToken.None;

        Task.Run(async () =>
        {
            try
            {
                var alarms = await _seedFn(ct);
                self.Tell(new SeedCompleted(alarms, isInitial));
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                // Actor stopping — drop silently.
            }
            catch (Exception ex)
            {
                self.Tell(new SeedFailed(ex, isInitial));
            }
        }, ct);
    }

    private void OnSeedCompleted(SeedCompleted msg)
    {
        if (_stopped) return;

        // Rebuild the cache authoritatively from the fresh snapshot (this is what makes
        // reconcile able to DROP rows for instances/alarms that disappeared — a merge
        // could never remove a stale row since the live stream sends no "removed" event).
        // The previous cache is kept alongside so the reconcile can publish as a DIFF:
        // an unchanged snapshot must not wake every viewer's render path once a minute.
        var previous = _cache.Count == 0
            ? null
            : new Dictionary<string, AlarmStateChanged>(_cache);

        _cache.Clear();
        foreach (var alarm in msg.Alarms)
        {
            var key = AlarmKey(alarm);
            if (!_cache.TryGetValue(key, out var existing) || alarm.Timestamp >= existing.Timestamp)
                _cache[key] = alarm;
        }

        // Flush the deltas buffered during the fan-out, deduped against the seed: a
        // buffered delta whose key is in the fresh snapshot with an equal-or-newer
        // timestamp is already reflected → drop; a strictly-newer (or new-key) delta is
        // applied. Inclusive-on-snapshot boundary matches DebugStreamBridgeActor.
        var flushChanged = FlushBuffer();

        _fanoutInFlight = false;
        // Only a connect/failover-driven seed makes the next tick redundant — a tick's own
        // fan-out must never suppress the tick after it (that halved the reconcile rate).
        if (_fanoutSuppressesNextTick) _fanoutSinceLastTick = true;
        _consecutiveSeedFailures = 0;
        Timers.Cancel(SeedRetryTimerKey);
        var firstSeed = !_seeded;
        _seeded = true;

        _log.Debug("Site-alarm {0} {1} complete: {2} alarm row(s)",
            _siteIdentifier, msg.IsInitial ? "seed" : "reconcile", _cache.Count);

        // The fresh snapshot already carries the buffered deltas; drop any armed coalesce
        // tick so we publish once, immediately — but only when something actually moved.
        Timers.Cancel(PublishTimerKey);
        _publishPending = false;
        if (firstSeed || flushChanged || DiffersFrom(previous))
            Publish();

        // A failover re-seed requested while this fan-out was in flight runs now (N7.1).
        RunQueuedFanoutIfAny();
    }

    /// <summary>
    /// Runs a fan-out that collided with an in-flight one, carrying the suppression flag it
    /// was queued with so a queued connect/failover re-seed still stands the next tick down
    /// (and a queued TICK fan-out still does not).
    /// </summary>
    private void RunQueuedFanoutIfAny()
    {
        if (!_reseedQueued) return;
        _reseedQueued = false;
        var suppresses = _queuedFanoutSuppressesNextTick;
        _queuedFanoutSuppressesNextTick = false;
        StartFanout(isInitial: false, suppressesNextTick: suppresses);
    }

    /// <summary>
    /// True when the just-installed snapshot differs from <paramref name="previous"/> in any
    /// key or any row value — the reconcile's "is this publish worth a viewer fan-out?" test.
    /// A null previous cache (nothing published yet) always counts as different.
    /// </summary>
    private bool DiffersFrom(Dictionary<string, AlarmStateChanged>? previous)
    {
        if (previous is null || previous.Count != _cache.Count) return true;

        foreach (var (key, current) in _cache)
        {
            if (!previous.TryGetValue(key, out var old)) return true;
            // AlarmStateChanged is a record: structural equality covers state, level,
            // timestamp and every native-alarm enrichment field.
            if (!Equals(old, current)) return true;
        }
        return false;
    }

    private void OnSeedFailed(SeedFailed msg)
    {
        if (_stopped) return;

        _consecutiveSeedFailures++;

        _log.Warning(msg.Exception,
            "Site-alarm {0} {1} fan-out failed ({2} consecutive); keeping current cache and retrying with backoff",
            _siteIdentifier, msg.IsInitial ? "seed" : "reconcile", _consecutiveSeedFailures);

        // Don't lose deltas captured during the failed window — apply them pass-through
        // into the (possibly stale/empty) cache. The next reconcile re-seeds authoritatively.
        _fanoutInFlight = false;
        // Same rule as the success path: only a connect/failover-driven fan-out stands the
        // next reconcile tick down.
        if (_fanoutSuppressesNextTick) _fanoutSinceLastTick = true;
        var flushChanged = FlushBuffer(dedupAgainstSeed: false);

        // Only publish if we already had a seed (so IsLive doesn't flip true on a
        // failed initial seed — the page keeps its poll fallback until we truly seed).
        if (_seeded && flushChanged)
        {
            Timers.Cancel(PublishTimerKey);
            _publishPending = false;
            Publish();
        }

        // A failover re-seed requested while this fan-out was in flight runs now (N7.1).
        if (_reseedQueued)
        {
            RunQueuedFanoutIfAny();
            return;
        }

        // Backoff retry (WP2.3): the seed leg used to have none — a site returning errors
        // was re-fanned at full reconcile cadence forever. Retry at reconnectDelay doubling
        // up to MaxSeedRetryBackoffMultiplier × the reconcile interval; the periodic
        // reconcile remains the floor once the backoff saturates.
        ScheduleSeedRetry();
    }

    /// <summary>
    /// Arms the backoff retry for a failed fan-out: <c>reconnectDelay × 2^(failures-1)</c>,
    /// capped at <see cref="MaxSeedRetryBackoffMultiplier"/> × the reconcile interval. Once
    /// the cap is reached the periodic reconcile tick is doing the same work anyway, so no
    /// extra timer is armed.
    /// </summary>
    private void ScheduleSeedRetry()
    {
        var cap = _reconcileInterval > TimeSpan.Zero
            ? TimeSpan.FromTicks(_reconcileInterval.Ticks * MaxSeedRetryBackoffMultiplier)
            : TimeSpan.FromMinutes(8);

        var shift = Math.Min(_consecutiveSeedFailures - 1, 16);
        var delayTicks = _reconnectDelay.Ticks * (1L << shift);
        var delay = TimeSpan.FromTicks(Math.Min(delayTicks, cap.Ticks));
        if (delay <= TimeSpan.Zero) return;

        Timers.StartSingleTimer(SeedRetryTimerKey, new RetrySeed(), delay);
    }

    /// <summary>
    /// Flushes the pre-fan-out buffer in arrival order. When <paramref name="dedupAgainstSeed"/>
    /// is true (normal completion) a buffered delta already reflected in the just-installed
    /// snapshot (same key, timestamp &lt;= cache entry) is dropped; otherwise every buffered
    /// delta is applied pass-through. Returns <c>true</c> when at least one delta changed the
    /// cache (the reconcile's publish-only-on-change test must count these too).
    /// </summary>
    private bool FlushBuffer(bool dedupAgainstSeed = true)
    {
        if (_buffer.Count == 0) return false;

        var changed = false;
        foreach (var delta in _buffer)
        {
            changed |= ApplyDelta(delta, requireStrictlyNewer: dedupAgainstSeed);
        }
        _buffer.Clear();
        return changed;
    }

    // ── Live delta handling ─────────────────────────────────────────────────────

    private void HandleLiveDelta(AlarmStateChanged delta)
    {
        if (_stopped) return;

        if (_fanoutInFlight)
        {
            if (!_bufferWarned && _buffer.Count + 1 > BufferWarnThreshold)
            {
                _bufferWarned = true;
                _log.Warning(
                    "Site-alarm pre-seed buffer for {0} exceeded {1} deltas while a fan-out was in flight " +
                    "(hard cap {2}, drop-oldest beyond it).",
                    _siteIdentifier, BufferWarnThreshold, MaxBufferedDeltas);
            }

            // Hard cap with drop-oldest: an unbounded buffer behind a stuck fan-out was a
            // straight memory leak. The evicted rows are superseded by the snapshot that
            // ends the fan-out, so the drop costs freshness, never correctness.
            while (_buffer.Count >= MaxBufferedDeltas)
            {
                _buffer.Dequeue();
                _bufferDropped++;
                ScadaBridgeTelemetry.RecordLiveAlarmBufferDrop();
                if (_bufferDropped == 1 || _bufferDropped % 500 == 0)
                {
                    _log.Warning(
                        "Site-alarm pre-seed buffer for {0} is at its {1}-delta cap; {2} delta(s) evicted so far",
                        _siteIdentifier, MaxBufferedDeltas, _bufferDropped);
                }
            }

            _buffer.Enqueue(delta);
            return;
        }

        // Pass-through: apply and (coalesced) publish only if the cache actually changed.
        if (ApplyDelta(delta, requireStrictlyNewer: false))
            SchedulePublish();
    }

    /// <summary>
    /// Coalesced publish: with a positive window, the first dirtying delta arms a
    /// single-shot timer and further deltas ride the same tick — one snapshot copy and
    /// one viewer fan-out per window instead of per transition (N6). Zero = legacy
    /// immediate publish. Last write wins, so batching never changes final state.
    /// </summary>
    private void SchedulePublish()
    {
        if (_publishCoalesce <= TimeSpan.Zero) { Publish(); return; }
        if (_publishPending) return;
        _publishPending = true;
        Timers.StartSingleTimer(PublishTimerKey, new PublishCoalesced(), _publishCoalesce);
    }

    /// <summary>
    /// Applies one alarm delta to the cache keyed by <see cref="AlarmKey"/>. Returns
    /// <c>true</c> if the cache changed. A stale (older) delta for an existing key is
    /// ignored. Never carries a placeholder (the live stream drops those), so a live
    /// delta for a real alarm can only add/replace its own key — it never touches a
    /// placeholder row under a different key.
    /// </summary>
    private bool ApplyDelta(AlarmStateChanged delta, bool requireStrictlyNewer)
    {
        var key = AlarmKey(delta);
        if (_cache.TryGetValue(key, out var existing))
        {
            var newer = requireStrictlyNewer
                ? delta.Timestamp > existing.Timestamp
                : delta.Timestamp >= existing.Timestamp;
            if (!newer) return false;
        }
        _cache[key] = delta;
        return true;
    }

    private void Publish()
    {
        // Fresh immutable list — reference swap; readers never see a partial mutation.
        var snapshot = _cache.Values.ToList();
        try
        {
            // Liveness rides every publish so the owning service's IsLive tracks the STREAM,
            // not merely "a snapshot was published once". A down stream means the cache is
            // only as fresh as the last reconcile and the page must keep polling.
            _publish(snapshot, !_streamDown);
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "Site-alarm publish callback threw for {0}; ignoring", _siteIdentifier);
        }
    }

    // ── gRPC stream lifecycle (mirrors DebugStreamBridgeActor) ──────────────────

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

        // The stream is not live again until the server actually accepts it — liveness flips
        // on GrpcAlarmStreamConnected, not on the attempt (WP2.3 carried residual).
        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        _log.Info("Opening site-alarm gRPC stream for {0} to {1}", _siteIdentifier, endpoint);

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

        Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), _stabilityWindow);

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

        // The subscription task itself is observed (below): a fault escaping
        // SubscribeSiteAsync — or a Task.Run that never started because ct was already
        // cancelled — would otherwise leave the actor waiting on a stream that does not
        // exist, with the exception silently unobserved.
        Task.Run(async () =>
        {
            await client.SubscribeSiteAsync(
                _correlationId,
                alarm => self.Tell(alarm),
                ex => self.Tell(new GrpcAlarmStreamError(ex, generation)),
                () => self.Tell(new GrpcAlarmStreamCompleted(generation)),
                ct,
                () => self.Tell(new GrpcAlarmStreamConnected(generation)));
        }, ct).ContinueWith(t =>
        {
            if (t.IsFaulted)
                self.Tell(new GrpcAlarmStreamError(t.Exception!.GetBaseException(), generation));
            else if (t.IsCanceled && !ct.IsCancellationRequested)
                self.Tell(new GrpcAlarmStreamCompleted(generation));
            // RanToCompletion: SubscribeSiteAsync already reported its own outcome.
        }, TaskContinuationOptions.ExecuteSynchronously);
    }

    /// <summary>
    /// The site accepted the subscription — the only point at which the stream is genuinely
    /// live. Consumes the pending re-seed (so exactly ONE fan-out runs per successful
    /// (re)connect, instead of one per reconnect attempt against a site that is by definition
    /// unreachable) and republishes so <c>IsLive</c> recovers.
    /// </summary>
    private void OnStreamConnected()
    {
        _openInFlight = false;

        var wasDown = _streamDown;
        _streamDown = false;

        if (_seedOnConnect)
        {
            _seedOnConnect = false;
            _log.Info("Site-alarm gRPC stream for {0} (re)connected; running one re-seed", _siteIdentifier);
            StartFanout(isInitial: false, suppressesNextTick: true);
        }

        // Liveness recovered — republish so viewers stop falling back to polling even if the
        // seed that follows finds nothing changed.
        if (wasDown && _seeded)
            Publish();
    }

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

        // Stream failed before the stability window — retry budget NOT recovered.
        Timers.Cancel(StabilityTimerKey);

        _openInFlight = false;
        var wasLive = !_streamDown;
        _streamDown = true;

        // The next successful connect owes exactly one re-seed. Setting it here (rather than
        // fanning out now) is the fix for the per-attempt re-fan-out: a site outage used to
        // cost one whole-site snapshot per retry.
        _seedOnConnect = true;

        // Liveness lost — tell viewers immediately so a dead stream stops reading as live.
        if (wasLive && _seeded)
            Publish();

        _retryCount++;

        if (_retryCount > MaxRetries)
        {
            // Give up the stream, but do NOT stop the aggregator: the periodic reconcile
            // still refreshes the cache from site snapshots, so the page keeps a
            // (slower) live-ish view rather than going dark. A later reconcile-triggered
            // reconnect is not attempted here; the stream is simply left down.
            _log.Error("Site-alarm gRPC stream for {0} exceeded max retries ({1}); leaving stream down, " +
                       "reconcile snapshots continue and the next reconcile tick will retry the stream",
                       _siteIdentifier, MaxRetries);
            _retryBudgetExhausted = true;
            CleanupGrpc();
            return;
        }

        // Unsubscribe the failed stream on the previous endpoint (TryGet, never
        // GetOrCreate) so the site releases its relay actor instead of leaving a zombie.
        var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        _grpcFactory.TryGet(_siteIdentifier, previousEndpoint)?.Unsubscribe(_correlationId);

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

        // Telemetry: a NodeA↔NodeB failover flip is a reconnect + re-seed.
        ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();

        // The re-seed is owed to the CONNECT (see _seedOnConnect above), not to this
        // attempt: a snapshot fan-out issued while the site is unreachable degrades to
        // empty rows and would have to be redone on reconnect anyway.
        if (_retryCount == 1)
            Self.Tell(new ReconnectAlarmStream());
        else
            Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), _reconnectDelay);
    }

    /// <summary>
    /// Handles a graceful end of stream. The stream is torn down and left down for the
    /// reconcile tick to reopen — deliberately NOT reopened inline, so a site that keeps
    /// completing streams immediately can never spin this actor into a hot reconnect loop.
    /// The retry budget is neither spent nor reset here (completion is not a fault), and the
    /// endpoint is not flipped: the node that just closed a stream cleanly is healthy.
    /// </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("Site-alarm gRPC stream for {0} completed gracefully (server end of stream); " +
                  "reopening on the next reconcile tick", _siteIdentifier);

        _openInFlight = false;
        var wasLive = !_streamDown;
        _streamDown = true;
        // The reopen owes one re-seed, run when it actually connects.
        _seedOnConnect = true;
        CleanupGrpc();

        // A completed stream is NOT live: without this the cache kept reporting live for up
        // to a full reconcile interval after the site closed the feed (WP2.3 residual).
        if (wasLive && _seeded)
            Publish();
    }

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

        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
        _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId);
    }

    // ── Dedup key (copied from DebugStreamBridgeActor.AlarmKey with attribution) ──

    /// <summary>
    /// NUL delimiter so distinct identities never collide on a shared boundary. Cannot
    /// appear in an instance/alarm name. Mirrors <see cref="DebugStreamBridgeActor"/>.
    /// </summary>
    private const char KeyDelimiter = '\u0000';

    /// <summary>
    /// Per-alarm dedup identity = (InstanceUniqueName, AlarmName, SourceReference) —
    /// identical to <c>DebugStreamBridgeActor.AlarmKey</c> so native per-condition alarms
    /// sharing an AlarmName but differing by source reference are not conflated. Each
    /// nullable component is guarded to prevent silent null/empty key collisions.
    /// </summary>
    private static string AlarmKey(AlarmStateChanged a) =>
        string.Concat(
            a.InstanceUniqueName ?? string.Empty, KeyDelimiter,
            a.AlarmName ?? string.Empty, KeyDelimiter,
            a.SourceReference ?? string.Empty);
}

/// <summary>Message asking a <see cref="SiteAlarmAggregatorActor"/> to stop (last viewer left).</summary>
public sealed record StopSiteAlarmAggregator;

/// <summary>Internal: a seed/reconcile snapshot fan-out completed with the whole-site alarm rows.</summary>
internal sealed record SeedCompleted(IReadOnlyList<AlarmStateChanged> Alarms, bool IsInitial);

/// <summary>Internal: a seed/reconcile snapshot fan-out threw as a whole.</summary>
internal sealed record SeedFailed(Exception Exception, bool IsInitial);

/// <summary>Internal: periodic reconcile tick (and the re-seed kicked after a reconnect).</summary>
internal sealed record RunReconcile;

/// <summary>Internal: backoff-scheduled retry of a fan-out that failed as a whole.</summary>
internal sealed record RetrySeed;

/// <summary>Internal: the site accepted the site-wide alarm subscription (response headers
/// received), stamped with its stream generation so a late connect from a cancelled stream
/// is ignored. This — not the reconnect attempt — is what earns a re-seed.</summary>
internal sealed record GrpcAlarmStreamConnected(int Generation);

/// <summary>Internal: coalesced-publish tick — flush the dirty cache to viewers once (N6).</summary>
internal sealed record PublishCoalesced;

/// <summary>Internal: site-alarm gRPC stream error occurred, stamped with the stream
/// generation it came from so a late error from a cancelled stream can be ignored (N7.2).</summary>
internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation);

/// <summary>Internal: the site-alarm gRPC stream ended gracefully (server status OK — max
/// stream lifetime or site shutdown), stamped with its stream generation so a late completion
/// from a cancelled stream can be ignored.</summary>
internal sealed record GrpcAlarmStreamCompleted(int Generation);

/// <summary>Internal: reconnect the site-alarm gRPC stream (flip node).</summary>
internal sealed record ReconnectAlarmStream;

/// <summary>Internal: the current site-alarm gRPC stream has stayed up long enough to recover the retry budget.</summary>
internal sealed record GrpcAlarmStreamStable;