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.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";
    /// <summary>Delay between gRPC reconnection attempts.</summary>
    internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);

    /// <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);

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

    /// <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. Mutated only on the actor thread.
    /// </summary>
    private readonly List<object> _preSnapshotBuffer = new();

    /// <summary>
    /// Defensive log threshold: if the pre-snapshot buffer grows past this many events
    /// during a slow snapshot we log once (events are NOT dropped — the window is short).
    /// </summary>
    private const int BufferWarnThreshold = 10_000;
    private bool _bufferWarned;

    /// <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 ClusterClient 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 ClusterClient 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 (via ClusterClient).
        // 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 ReceiveTimeout 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);

            // 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;
        });

        // Domain events arriving via Self.Tell from gRPC callback.
        // Receiving an event must NOT reset _retryCount — 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 (same handler, phase-dependent behavior).
        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 =>
        {
            _log.Warning("gRPC stream error for {0}: {1}", _instanceUniqueName, msg.Exception.Message);
            HandleGrpcError();
        });

        // 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);
        });

        // Orphan safety net — if nobody stops us within 5 minutes, self-terminate
        Context.SetReceiveTimeout(TimeSpan.FromMinutes(5));
        Receive<ReceiveTimeout>(_ =>
        {
            _log.Warning("Debug stream for {0} timed out (orphaned session), stopping", _instanceUniqueName);
            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;
        }

        _preSnapshotBuffer.Add(evt);
        if (!_bufferWarned && _preSnapshotBuffer.Count > BufferWarnThreshold)
        {
            _bufferWarned = true;
            _log.Warning(
                "Pre-snapshot debug-event buffer for {0} exceeded {1} events while awaiting the snapshot; " +
                "events are still retained (not dropped).",
                _instanceUniqueName, BufferWarnThreshold);
        }
    }

    /// <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",
                flushed, _instanceUniqueName, dropped);
        }

        _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);
    }

    /// <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 client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint);
        var self = Self;
        var ct = _grpcCts.Token;

        // Launch as background task — onEvent and onError marshal back to actor via Tell
        Task.Run(async () =>
        {
            await client.SubscribeAsync(
                _correlationId,
                _instanceUniqueName,
                evt => self.Tell(evt),
                ex => self.Tell(new GrpcStreamError(ex)),
                ct);
        }, ct);
    }

    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>
/// Internal message indicating a gRPC stream error occurred.
/// </summary>
internal record GrpcStreamError(Exception Exception);

/// <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;