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

using System.Collections.Concurrent;
using System.Threading.Channels;
using Akka.Actor;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using GrpcStatus = Grpc.Core.Status;

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

/// <summary>
/// gRPC service that accepts instance stream subscriptions from central nodes.
/// Creates a StreamRelayActor per subscription to bridge Akka domain events
/// through a Channel&lt;T&gt; to the gRPC response stream.
/// </summary>
public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
{
    private readonly ISiteStreamSubscriber _streamSubscriber;
    private ActorSystem? _actorSystem;
    private readonly ILogger<SiteStreamGrpcServer> _logger;
    private readonly ConcurrentDictionary<string, StreamEntry> _activeStreams = new();
    private readonly int _maxConcurrentStreams;
    private readonly TimeSpan _maxStreamLifetime;
    private readonly int _instanceChannelCapacity;
    private readonly int _siteAlarmChannelCapacity;
    private readonly int _streamBatchMaxEvents;
    private readonly TimeSpan _streamBatchWindow;
    private volatile bool _ready;
    // Flipped by CancelAllStreams() when the host enters
    // CoordinatedShutdown so SubscribeInstance refuses new streams with
    // Unavailable before the actor system tears down. Strictly monotonic — once
    // true, never reset (the server is single-lifetime per host).
    private volatile bool _shuttingDown;
    private long _actorCounter;
    // Central-side ingest actor proxy. Set by the host
    // after the cluster singleton starts. When null the
    // IngestAuditEvents RPC replies with an empty IngestAck so sites can
    // safely retry — wiring-incomplete is treated as transient, never fatal.
    private IActorRef? _auditIngestActor;
    // Ask timeout is 30 s. The ingest actor's repo
    // calls are sub-100 ms in steady state; a generous timeout absorbs a slow
    // MSSQL connection without surfacing as a gRPC failure on a healthy site.
    // Shared with CentralCommunicationActor (same assembly) so the two audit-ingest
    // transports use one source of truth for the timeout.
    internal static readonly TimeSpan AuditIngestAskTimeout = TimeSpan.FromSeconds(30);
    // Audit Log: site-local queue handed in by AkkaHostedService on
    // site roles so the central reconciliation puller's PullAuditEvents RPC
    // can read Pending/Forwarded rows. Null when not wired (e.g. central-only
    // host or test composing the server in isolation) — the handler treats
    // the missing queue as "nothing to ship" and returns an empty response so
    // central retries on its next reconciliation cycle.
    private ISiteAuditQueue? _siteAuditQueue;
    // Site Call Audit: site-local operation-tracking store handed in by
    // AkkaHostedService on site roles so the central reconciliation puller's
    // PullSiteCalls RPC can read tracking rows changed since a cursor. Null
    // when not wired (central-only host or test composing the server in
    // isolation) — the handler treats the missing store as "nothing to ship"
    // and returns an empty response so central retries on its next cycle.
    // Mirrors _siteAuditQueue.
    private IOperationTrackingStore? _operationTrackingStore;

    /// <summary>
    /// Test-only constructor — kept <c>internal</c> so the DI container sees a
    /// single public constructor and is not faced with an ambiguous choice.
    /// </summary>
    /// <param name="streamSubscriber">The stream subscriber for managing subscriptions.</param>
    /// <param name="logger">The logger instance.</param>
    /// <param name="maxConcurrentStreams">The maximum concurrent streams (default 100).</param>
    internal SiteStreamGrpcServer(
        ISiteStreamSubscriber streamSubscriber,
        ILogger<SiteStreamGrpcServer> logger,
        int maxConcurrentStreams = 100)
        : this(streamSubscriber, logger, maxConcurrentStreams, TimeSpan.FromHours(4),
            DefaultInstanceChannelCapacity, DefaultSiteAlarmChannelCapacity,
            DefaultStreamBatchMaxEvents, DefaultStreamBatchWindow)
    {
    }

    /// <summary>Fallback Debug View send-channel capacity when no options are bound.</summary>
    internal const int DefaultInstanceChannelCapacity = 1000;

    /// <summary>Fallback site-wide alarm send-channel capacity when no options are bound.</summary>
    internal const int DefaultSiteAlarmChannelCapacity = 20_000;

    /// <summary>Fallback stream-batch size cap when no options are bound (R2).</summary>
    internal const int DefaultStreamBatchMaxEvents = 100;

    /// <summary>Fallback stream-batch coalescing window when no options are bound (R2).</summary>
    internal static readonly TimeSpan DefaultStreamBatchWindow = TimeSpan.FromMilliseconds(25);

    /// <summary>
    /// DI constructor — binds <see cref="CommunicationOptions.GrpcMaxConcurrentStreams"/>
    /// and <see cref="CommunicationOptions.GrpcMaxStreamLifetime"/> so the documented
    /// concurrency limit and the 4-hour zombie-stream session timeout are honoured
    /// rather than hard-coded.
    /// </summary>
    /// <param name="streamSubscriber">The stream subscriber for managing subscriptions.</param>
    /// <param name="logger">The logger instance.</param>
    /// <param name="options">Communication options containing stream limits and timeouts.</param>
    public SiteStreamGrpcServer(
        ISiteStreamSubscriber streamSubscriber,
        ILogger<SiteStreamGrpcServer> logger,
        IOptions<CommunicationOptions> options)
        : this(streamSubscriber, logger,
            options.Value.GrpcMaxConcurrentStreams,
            options.Value.GrpcMaxStreamLifetime,
            options.Value.GrpcInstanceStreamChannelCapacity,
            options.Value.GrpcSiteAlarmStreamChannelCapacity,
            options.Value.GrpcStreamBatchMaxEvents,
            options.Value.GrpcStreamBatchWindow)
    {
    }

    private SiteStreamGrpcServer(
        ISiteStreamSubscriber streamSubscriber,
        ILogger<SiteStreamGrpcServer> logger,
        int maxConcurrentStreams,
        TimeSpan maxStreamLifetime,
        int instanceChannelCapacity,
        int siteAlarmChannelCapacity,
        int streamBatchMaxEvents,
        TimeSpan streamBatchWindow)
    {
        _streamSubscriber = streamSubscriber;
        _logger = logger;
        _maxConcurrentStreams = maxConcurrentStreams;
        _maxStreamLifetime = maxStreamLifetime;
        _instanceChannelCapacity = Math.Max(1, instanceChannelCapacity);
        _siteAlarmChannelCapacity = Math.Max(1, siteAlarmChannelCapacity);
        // Floored/clamped rather than thrown on: CommunicationOptionsValidator already
        // fails the boot on a bad value, and a degenerate one must not blow up deep
        // inside a live RPC on a host composed without validation (tests, embedded use).
        _streamBatchMaxEvents = Math.Max(1, streamBatchMaxEvents);
        _streamBatchWindow = streamBatchWindow < TimeSpan.Zero ? TimeSpan.Zero : streamBatchWindow;
    }

    /// <summary>
    /// Marks the server as ready to accept subscriptions and injects the ActorSystem.
    /// Called after the site runtime actor system is fully initialized.
    /// The ActorSystem is set here rather than via the constructor so that
    /// the gRPC server can be created by DI before the actor system exists.
    /// </summary>
    /// <param name="actorSystem">The initialized Akka actor system.</param>
    public void SetReady(ActorSystem actorSystem)
    {
        _actorSystem = actorSystem;
        _ready = true;
    }

    /// <summary>
    /// Hands the central-side <c>AuditLogIngestActor</c> proxy to the gRPC
    /// server so the <see cref="IngestAuditEvents"/> RPC can route incoming
    /// site batches. Mirrors the way
    /// <c>CommunicationService.SetNotificationOutbox</c> takes the Notification
    /// Outbox singleton proxy. The actor is supplied after the cluster
    /// singleton starts.
    /// </summary>
    /// <param name="proxy">The audit log ingest actor proxy.</param>
    public void SetAuditIngestActor(IActorRef proxy)
    {
        _auditIngestActor = proxy;
    }

    /// <summary>
    /// Hands the site-local <see cref="ISiteAuditQueue"/> (the same
    /// <c>SqliteAuditWriter</c> singleton that backs <see cref="IAuditWriter"/>
    /// on the script thread) to the gRPC server so the
    /// <see cref="PullAuditEvents"/> RPC can serve central's reconciliation
    /// pulls. Mirrors <see cref="SetAuditIngestActor"/>: wired post-construction
    /// because the queue and the gRPC server are both DI singletons brought up
    /// in independent orders on site startup.
    /// </summary>
    /// <param name="queue">The site audit queue for serving reconciliation pulls.</param>
    public void SetSiteAuditQueue(ISiteAuditQueue queue)
    {
        _siteAuditQueue = queue;
    }

    /// <summary>
    /// Hands the site-local <see cref="IOperationTrackingStore"/> (the same
    /// <c>OperationTrackingStore</c> singleton that backs
    /// <c>Tracking.Status(id)</c> on the script thread) to the gRPC server so
    /// the Site Call Audit <see cref="PullSiteCalls"/> RPC can serve
    /// central's reconciliation pulls. Mirrors <see cref="SetSiteAuditQueue"/>:
    /// wired post-construction because the store and the gRPC server are both
    /// DI singletons brought up in independent orders on site startup.
    /// </summary>
    /// <param name="store">The site operation-tracking store for serving reconciliation pulls.</param>
    public void SetOperationTrackingStore(IOperationTrackingStore store)
    {
        _operationTrackingStore = store;
    }

    /// <summary>
    /// Signals the gRPC server to begin its part of the
    /// site shutdown sequence — refuse new <see cref="SubscribeInstance"/>
    /// streams with <see cref="StatusCode.Unavailable"/> and cancel every
    /// active stream so its <c>await foreach</c> observes
    /// <see cref="OperationCanceledException"/> and the response stream
    /// completes with <c>Cancelled</c> on the client. Idempotent — safe to call
    /// more than once. Invoked from the site host's
    /// <c>IHostApplicationLifetime.ApplicationStopping</c> callback BEFORE
    /// Akka's <c>CoordinatedShutdown</c> runs, so in-flight clients get a
    /// clean cancellation they can reconnect on rather than a silent stream
    /// that only times out via gRPC keepalive.
    /// </summary>
    public void CancelAllStreams()
    {
        _shuttingDown = true;
        foreach (var entry in _activeStreams.Values)
        {
            try
            {
                entry.Cts.Cancel();
            }
            catch (ObjectDisposedException)
            {
                // Already cleaned up by its own finally — nothing to do.
            }
        }
    }

    /// <summary>
    /// Exposed for test assertions on the shutdown state.
    /// </summary>
    internal bool IsShuttingDown => _shuttingDown;

    /// <summary>
    /// Number of currently active streaming subscriptions. Exposed for diagnostics.
    /// </summary>
    public int ActiveStreamCount => _activeStreams.Count;

    /// <summary>Effective max concurrent stream limit. Exposed for tests.</summary>
    internal int MaxConcurrentStreams => _maxConcurrentStreams;

    /// <summary>Effective per-stream session lifetime. Exposed for tests.</summary>
    internal TimeSpan MaxStreamLifetime => _maxStreamLifetime;

    /// <summary>Effective Debug View send-channel capacity. Exposed for tests.</summary>
    internal int InstanceChannelCapacity => _instanceChannelCapacity;

    /// <summary>Effective site-wide alarm send-channel capacity. Exposed for tests.</summary>
    internal int SiteAlarmChannelCapacity => _siteAlarmChannelCapacity;

    /// <summary>Effective stream-batch size cap (R2). Exposed for tests.</summary>
    internal int StreamBatchMaxEvents => _streamBatchMaxEvents;

    /// <summary>Effective stream-batch coalescing window (R2). Exposed for tests.</summary>
    internal TimeSpan StreamBatchWindow => _streamBatchWindow;

    /// <summary>
    /// Total events evicted from stream send channels on this node since start (both stream
    /// kinds). Exported as <c>scadabridge.site.stream.events_dropped</c>; exposed here so a
    /// test — and an operator via diagnostics — can read the raw count.
    /// </summary>
    public long DroppedStreamEventCount => Interlocked.Read(ref _droppedStreamEvents);

    private long _droppedStreamEvents;

    /// <inheritdoc />
    public override Task SubscribeInstance(
        InstanceStreamRequest request,
        IServerStreamWriter<SiteStreamEvent> responseStream,
        ServerCallContext context)
        => RunSubscriptionStreamAsync(
            request.CorrelationId,
            responseStream,
            context,
            relay => _streamSubscriber.Subscribe(request.InstanceUniqueName, relay),
            request.InstanceUniqueName,
            _instanceChannelCapacity,
            streamKind: "instance",
            // R2 batch negotiation: proto3 defaults this to false, so a central built
            // before batching existed keeps getting one frame per event.
            batchingSupported: request.BatchingSupported);

    /// <inheritdoc />
    public override Task SubscribeSite(
        SiteStreamRequest request,
        IServerStreamWriter<SiteStreamEvent> responseStream,
        ServerCallContext context)
        => RunSubscriptionStreamAsync(
            request.CorrelationId,
            responseStream,
            context,
            // Site-wide, alarm-only: no per-instance filter. StreamRelayActor
            // already drops IsConfiguredPlaceholder rows and maps only the
            // enriched AlarmStateUpdate, so it is reused unchanged.
            _streamSubscriber.SubscribeSiteAlarms,
            "site-wide alarms",
            // Its OWN, much larger channel (WP2.3): sharing the Debug View's 1000-slot
            // DropOldest meant an alarm burst during a WAN stall silently evicted operator-
            // visible transitions to make room for diagnostics traffic.
            _siteAlarmChannelCapacity,
            streamKind: "site-alarms",
            batchingSupported: request.BatchingSupported);

    /// <summary>
    /// Shared streaming pipeline behind <see cref="SubscribeInstance"/> and
    /// <see cref="SubscribeSite"/>: readiness/shutdown guards, correlation-id
    /// validation, duplicate replacement, the concurrency cap, the bounded
    /// <c>DropOldest</c> channel, the relay actor, connection telemetry, and the
    /// guaranteed cleanup. The only variation between the two RPCs is how the
    /// relay actor is subscribed to the site broadcast hub — supplied by
    /// <paramref name="subscribe"/> — and the log/telemetry description.
    /// </summary>
    /// <param name="correlationId">The client-supplied correlation id (also the actor-name/dedup key).</param>
    /// <param name="responseStream">The gRPC response stream to pump events into.</param>
    /// <param name="context">The server call context (carries the client cancellation token).</param>
    /// <param name="subscribe">Subscribes the relay actor to the hub, returning a subscription id.</param>
    /// <param name="description">Human-readable subscription description for logging.</param>
    /// <param name="channelCapacity">Send-channel capacity for this stream kind.</param>
    /// <param name="streamKind">Telemetry tag for this stream kind (<c>instance</c>/<c>site-alarms</c>).</param>
    /// <param name="batchingSupported">
    /// Whether the SUBSCRIBING CLIENT declared it understands the <c>SiteStreamEventBatch</c>
    /// oneof case (R2). False — the proto3 default an older central necessarily sends —
    /// pins this stream to one event per frame, so a peer that predates batching can never
    /// receive a frame case its generated code drops on the floor.
    /// </param>
    private async Task RunSubscriptionStreamAsync(
        string correlationId,
        IServerStreamWriter<SiteStreamEvent> responseStream,
        ServerCallContext context,
        Func<IActorRef, string> subscribe,
        string description,
        int channelCapacity,
        string streamKind,
        bool batchingSupported)
    {
        if (!_ready)
            throw new RpcException(new GrpcStatus(StatusCode.Unavailable, "Server not ready"));

        // Refuse new subscriptions during shutdown so
        // CoordinatedShutdown can quiesce without racing fresh streams.
        if (_shuttingDown)
            throw new RpcException(new GrpcStatus(StatusCode.Unavailable, "Server shutting down"));

        // correlation_id arrives off the wire on a public gRPC
        // endpoint and is used (below) to compose an Akka actor name. Akka actor names
        // have a restricted character set — a id containing '/', whitespace, or other
        // disallowed characters would make ActorOf throw InvalidActorNameException,
        // escaping as an unhandled RPC fault. Reject unsafe ids cleanly up front.
        if (string.IsNullOrEmpty(correlationId) ||
            !ActorPath.IsValidPathElement(correlationId))
        {
            throw new RpcException(new GrpcStatus(
                StatusCode.InvalidArgument, "correlation_id is missing or not a valid identifier"));
        }

        // Duplicate prevention -- REPLACE the existing stream for this correlationId.
        //
        // The removal is claimed FIRST, but nothing is cancelled until the concurrency cap
        // has had its say, and a REPLACEMENT never consults the cap at all (arch-review
        // 2026-08-19 P3-5). Both halves matter:
        //
        //  * A replacement is net-neutral on the stream count (one entry out, one back in),
        //    so refusing it can only ever destroy capacity, never protect it. The old code
        //    checked `Count >= max` after the removal, which is correct at exactly the cap
        //    but NOT above it — and above it is reachable, because the check below is
        //    deliberately check-then-act (see the note there). In that window a routine
        //    reconnect cancelled a healthy incumbent and was then refused with
        //    ResourceExhausted, leaving the correlation with ZERO streams until its next
        //    retry. Excluding the replacement from the cap removes that outcome entirely.
        //  * A genuinely NEW subscriber that the cap refuses must leave the existing streams
        //    untouched, which is automatic here: TryRemove found nothing, so nothing was
        //    cancelled before the throw.
        //
        // Newest-wins is preserved exactly: a fresh subscribe on a live correlation id
        // always displaces the incumbent, whether that incumbent is healthy or wedged.
        var replacingExistingStream = _activeStreams.TryRemove(correlationId, out var existingEntry);

        // Max concurrent streams, applied only to a genuinely new correlation id.
        // Deliberately check-then-act: two concurrent subscribes at the boundary can
        // both pass, overshooting the cap by at most (concurrent subscribers - 1).
        // Bounded and benign — not worth a lock on this path.
        if (!replacingExistingStream && _activeStreams.Count >= _maxConcurrentStreams)
            throw new RpcException(new GrpcStatus(StatusCode.ResourceExhausted, "Max concurrent streams reached"));

        // CANCEL ONLY, never Dispose. The replaced stream's CTS belongs to its own
        // handler's `using var streamCts`, which is still running and still reads
        // `streamCts.Token` (at the pump call below, and previously at the
        // `ReadAllAsync(streamCts.Token)` it replaced). Disposing it from here raced that
        // read and surfaced as an unhandled ObjectDisposedException escaping the RPC —
        // observed as a full-suite-load-only failure of
        // GrpcStreamIntegrationTests.Pipeline_DuplicateCorrelationId_ReplacesStream, and a
        // pre-existing hazard (the same Dispose + the same first-token-read relationship
        // exist unchanged before R2). Cancellation alone is what replacement needs; the
        // owning handler's `using` still disposes it exactly once on every exit path.
        if (replacingExistingStream)
        {
            try
            {
                existingEntry!.Cts.Cancel();
            }
            catch (ObjectDisposedException)
            {
                // Its owner finished and disposed it between the TryRemove and here —
                // already terminal, nothing to cancel. Mirrors CancelAllStreams().
            }
        }

        using var streamCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
        // Session timeout (design doc "gRPC Connection Keepalive": 4-hour third layer
        // of dead-client detection) — forces a long-lived zombie stream to terminate
        // even if keepalive PINGs never detect the loss.
        if (_maxStreamLifetime > TimeSpan.Zero && _maxStreamLifetime != Timeout.InfiniteTimeSpan)
            streamCts.CancelAfter(_maxStreamLifetime);
        var entry = new StreamEntry(streamCts);
        _activeStreams[correlationId] = entry;

        long dropped = 0;
        var channel = Channel.CreateBounded<SiteStreamEvent>(
            new BoundedChannelOptions(channelCapacity) { FullMode = BoundedChannelFullMode.DropOldest },
            _ =>
            {
                // Lossy-under-backpressure is the debug-view spec; make the real
                // loss visible: first eviction + every 500th thereafter, plus a
                // per-kind counter on the node (WP2.3).
                var n = Interlocked.Increment(ref dropped);
                Interlocked.Increment(ref _droppedStreamEvents);
                ScadaBridgeTelemetry.RecordSiteStreamEventDropped(streamKind);
                if (n == 1 || n % 500 == 0)
                    _logger.LogWarning(
                        "Stream {CorrelationId} ({StreamKind}) backpressure: {Dropped} oldest event(s) evicted so far " +
                        "of a {Capacity}-slot channel",
                        correlationId, streamKind, n, channelCapacity);
            });

        var actorSeq = Interlocked.Increment(ref _actorCounter);
        var relayActor = _actorSystem!.ActorOf(
            Props.Create(typeof(Actors.StreamRelayActor), correlationId, channel.Writer),
            $"stream-relay-{correlationId}-{actorSeq}");

        // The previous code called _streamSubscriber.Subscribe
        // OUTSIDE the try block that owns relay-actor cleanup. If Subscribe threw
        // (stale instance name, index lookup fault, site runtime shutting down),
        // the freshly-created relay actor, the _activeStreams entry, the
        // StreamEntry.Cts, and the Channel<SiteStreamEvent> all leaked because the
        // finally never ran. Wrap Subscribe in its own try so any throw deterministically
        // stops the relay actor, removes the activeStreams entry, and completes the
        // channel before the RpcException escapes to the caller.
        string subscriptionId;
        try
        {
            subscriptionId = subscribe(relayActor);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Subscribe failed for {Description} (correlation {CorrelationId}); cleaning up relay actor.",
                description, correlationId);
            _actorSystem!.Stop(relayActor);
            channel.Writer.TryComplete();
            _activeStreams.TryRemove(
                new KeyValuePair<string, StreamEntry>(correlationId, entry));
            throw;
        }

        _logger.LogInformation(
            "Stream {CorrelationId} started for {Description} (subscription {SubscriptionId})",
            correlationId, description, subscriptionId);

        // Flush response headers NOW, while the relay is attached and before the first event
        // (WP2.3). This is the client's "the site accepted my subscription" signal — the
        // per-site aggregator hangs its once-per-successful-reconnect re-seed off it. Without
        // an explicit flush, ASP.NET Core defers headers to the first written message, so a
        // quiet site would never report connected. Best-effort: a client that vanished between
        // Subscribe and here fails the write, and the read loop below handles the teardown.
        try
        {
            await context.WriteResponseHeadersAsync(Metadata.Empty);
        }
        catch (Exception ex)
        {
            // Deliberately catch-all: an unsent header must never leak the relay actor and
            // the _activeStreams entry, which the finally below owns. Any real transport
            // failure resurfaces on the read/write loop immediately after.
            _logger.LogDebug(ex,
                "Could not flush response headers for stream {CorrelationId}; continuing.", correlationId);
        }

        // Telemetry follow-on: the connection is now fully established (Subscribe
        // succeeded, so no leak via the catch above). Count it up here and balance
        // it in the finally below so the scadabridge.site.stream.subscribers.active
        // gauge is decremented on EVERY exit path — normal completion, client-cancel /
        // duplicate-replacement (OperationCanceledException), server shutdown
        // (CancelAllStreams -> Cts.Cancel), and any other exception — guaranteeing
        // exactly one Closed per Opened and a gauge that never drifts up.
        ScadaBridgeTelemetry.SiteStreamSubscriberOpened();
        try
        {
            // R2 — event batching. The pump replaces the old per-event
            // `await foreach (…) WriteAsync(evt)` loop and is byte-for-byte identical to
            // it when maxBatchEvents == 1, which is exactly what an un-negotiated
            // subscription gets. It sits DOWNSTREAM of the bounded DropOldest channel
            // above, so it changes framing only — never the site's burst ceiling.
            await SiteStreamEventBatcher.PumpAsync(
                channel.Reader,
                correlationId,
                maxBatchEvents: batchingSupported ? _streamBatchMaxEvents : 1,
                maxWindow: _streamBatchWindow,
                (evt, token) => responseStream.WriteAsync(evt, token),
                // Recorded only on a negotiated stream: on an un-negotiated one every
                // frame carries exactly one event, so the histogram would degenerate
                // into a per-event instrument on the product's hottest path.
                batchingSupported
                    ? size => ScadaBridgeTelemetry.RecordSiteStreamBatchSize(streamKind, size)
                    : null,
                streamCts.Token);
        }
        catch (OperationCanceledException)
        {
            // Normal cancellation (client disconnect or duplicate replacement)
        }
        finally
        {
            ScadaBridgeTelemetry.SiteStreamSubscriberClosed();
            _streamSubscriber.RemoveSubscriber(relayActor);
            _actorSystem!.Stop(relayActor);
            channel.Writer.TryComplete();

            // Only remove our own entry -- a replacement stream may have already taken the slot
            _activeStreams.TryRemove(
                new KeyValuePair<string, StreamEntry>(correlationId, entry));

            _logger.LogInformation(
                "Stream {CorrelationId} for {Description} ended",
                correlationId, description);
        }
    }

    /// <inheritdoc />
    public override async Task<IngestAck> IngestAuditEvents(
        AuditEventBatch request,
        ServerCallContext context)
    {
        // Empty batch is a no-op; reply immediately so the client moves on.
        if (request.Events.Count == 0)
        {
            return new IngestAck();
        }

        var actor = _auditIngestActor;
        if (actor is null)
        {
            // Wiring incomplete (host startup race). Sites treat an empty
            // ack as "nothing was acked, leave rows Pending, retry next
            // drain" — exactly the right behaviour during host bring-up.
            _logger.LogWarning(
                "IngestAuditEvents received {Count} events before SetAuditIngestActor was called; returning empty ack.",
                request.Events.Count);
            return new IngestAck();
        }

        var entities = new List<AuditEvent>(request.Events.Count);
        foreach (var dto in request.Events)
        {
            entities.Add(AuditEventDtoMapper.FromDto(dto));
        }

        var cmd = new IngestAuditEventsCommand(entities);
        IngestAuditEventsReply reply;
        try
        {
            reply = await actor.Ask<IngestAuditEventsReply>(
                cmd, AuditIngestAskTimeout, context.CancellationToken);
        }
        catch (Exception ex)
        {
            // Audit ingest is best-effort; failing this RPC at the gRPC layer
            // would surface as a transport error and force the site to retry
            // (which it would do anyway). Logging + an empty ack keeps the
            // semantics consistent with the "wiring incomplete" path above.
            _logger.LogError(ex,
                "AuditLogIngestActor Ask failed for batch of {Count} events; returning empty ack.",
                request.Events.Count);
            return new IngestAck();
        }

        var ack = new IngestAck();
        foreach (var id in reply.AcceptedEventIds)
        {
            ack.AcceptedEventIds.Add(id.ToString());
        }
        return ack;
    }

    /// <inheritdoc />
    public override async Task<IngestAck> IngestCachedTelemetry(
        CachedTelemetryBatch request,
        ServerCallContext context)
    {
        if (request.Packets.Count == 0)
        {
            return new IngestAck();
        }

        var actor = _auditIngestActor;
        if (actor is null)
        {
            _logger.LogWarning(
                "IngestCachedTelemetry received {Count} packets before SetAuditIngestActor was called; returning empty ack.",
                request.Packets.Count);
            return new IngestAck();
        }

        var entries = new List<CachedTelemetryEntry>(request.Packets.Count);
        foreach (var packet in request.Packets)
        {
            var auditEvent = AuditEventDtoMapper.FromDto(packet.AuditEvent);
            var siteCall = SiteCallDtoMapper.FromDto(packet.Operational);
            entries.Add(new CachedTelemetryEntry(auditEvent, siteCall));
        }

        var cmd = new IngestCachedTelemetryCommand(entries);
        IngestCachedTelemetryReply reply;
        try
        {
            reply = await actor.Ask<IngestCachedTelemetryReply>(
                cmd, AuditIngestAskTimeout, context.CancellationToken);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "AuditLogIngestActor Ask failed for combined telemetry batch of {Count} packets; returning empty ack.",
                request.Packets.Count);
            return new IngestAck();
        }

        var ack = new IngestAck();
        foreach (var id in reply.AcceptedEventIds)
        {
            ack.AcceptedEventIds.Add(id.ToString());
        }
        return ack;
    }

    /// <inheritdoc />
    public override async Task<PullAuditEventsResponse> PullAuditEvents(
        PullAuditEventsRequest request,
        ServerCallContext context)
    {
        var queue = _siteAuditQueue;
        if (queue is null)
        {
            _logger.LogWarning(
                "PullAuditEvents invoked before SetSiteAuditQueue was called; returning empty response.");
            return new PullAuditEventsResponse();
        }

        if (request.BatchSize <= 0)
        {
            // Mirrors the SubscribeInstance guard: reject malformed requests
            // cleanly with InvalidArgument so the caller doesn't see a generic
            // RpcException from the underlying SQLite parameter validation.
            throw new RpcException(new GrpcStatus(
                StatusCode.InvalidArgument, "batch_size must be > 0"));
        }

        // sinceUtc defaults to DateTime.MinValue when the wrapper is absent —
        // i.e. "pull from the beginning of recorded history", which is the
        // intended behaviour for the very first reconciliation cycle.
        var since = request.SinceUtc is not null
            ? DateTime.SpecifyKind(request.SinceUtc.ToDateTime(), DateTimeKind.Utc)
            : DateTime.MinValue;

        // Composite-keyset cursor (WP2.3), mirroring PullSiteCalls: proto3
        // defaults an unset string to "", which the queue treats as "no cursor"
        // (legacy inclusive >= behaviour), so an older central is unaffected.
        var afterId = string.IsNullOrEmpty(request.AfterId) ? null : request.AfterId;

        // AT-LEAST-ONCE (WP2.3). The incoming cursor is central's receipt for
        // everything at or before it — that is the ONLY proof the site accepts.
        // Rows are flipped to Reconciled here, at the START of the NEXT pull,
        // instead of right after the previous response was projected: a fault
        // between the response leaving the site and central committing it used
        // to lose the rows outright, because they were already Reconciled and
        // ReadPendingSinceAsync would never serve them again. The flip runs
        // before the read so the rows it retires do not consume this batch's
        // budget. Best-effort — a failure here only costs a re-ship, which
        // central dedups on EventId.
        if (since > DateTime.MinValue)
        {
            try
            {
                await queue.MarkReconciledUpToAsync(since, afterId, context.CancellationToken);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "MarkReconciledUpToAsync failed for cursor since={Since} afterId={AfterId}; rows stay pending for the next pull.",
                    since, afterId);
            }
        }

        IReadOnlyList<AuditEvent> events;
        try
        {
            events = await queue.ReadPendingSinceAsync(
                since, request.BatchSize, afterId, context.CancellationToken);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "ReadPendingSinceAsync failed for since={Since} batch={Batch} afterId={AfterId}; returning empty response.",
                since, request.BatchSize, afterId);
            return new PullAuditEventsResponse();
        }

        var response = new PullAuditEventsResponse
        {
            // batch_size saturated → tell central to issue a follow-up pull
            // with an advanced cursor. The site doesn't compute the cursor —
            // central walks it forward from the last returned OccurredAtUtc
            // (plus, once it sets after_id, that row's EventId).
            MoreAvailable = events.Count >= request.BatchSize,
        };
        foreach (var evt in events)
        {
            response.Events.Add(AuditEventDtoMapper.ToDto(evt));
        }

        return response;
    }

    /// <inheritdoc />
    public override async Task<PullSiteCallsResponse> PullSiteCalls(
        PullSiteCallsRequest request,
        ServerCallContext context)
    {
        var store = _operationTrackingStore;
        if (store is null)
        {
            _logger.LogWarning(
                "PullSiteCalls invoked before SetOperationTrackingStore was called; returning empty response.");
            return new PullSiteCallsResponse();
        }

        if (request.BatchSize <= 0)
        {
            // Mirrors PullAuditEvents: reject malformed requests cleanly with
            // InvalidArgument so the caller doesn't see a generic RpcException
            // from the underlying SQLite parameter validation.
            throw new RpcException(new GrpcStatus(
                StatusCode.InvalidArgument, "batch_size must be > 0"));
        }

        // since_utc defaults to DateTime.MinValue when the wrapper is absent —
        // i.e. "pull from the beginning of recorded history", the intended
        // behaviour for the very first reconciliation cycle.
        var since = request.SinceUtc is not null
            ? DateTime.SpecifyKind(request.SinceUtc.ToDateTime(), DateTimeKind.Utc)
            : DateTime.MinValue;

        // Composite-keyset cursor: proto3 defaults an unset string to
        // "", which the store treats as "no cursor" (legacy inclusive >= behaviour)
        // — so an older central that never sets after_id is unaffected.
        var afterId = string.IsNullOrEmpty(request.AfterId) ? null : request.AfterId;

        IReadOnlyList<SiteCallOperational> operationals;
        try
        {
            operationals = await store.ReadChangedSinceAsync(
                since, request.BatchSize, afterId, context.CancellationToken);
        }
        catch (Exception ex)
        {
            // Best-effort, like PullAuditEvents: a read fault must never abort
            // the reconciliation tick — central retries on its next cycle.
            _logger.LogError(ex,
                "ReadChangedSinceAsync failed for since={Since} batch={Batch} afterId={AfterId}; returning empty response.",
                since, request.BatchSize, afterId);
            return new PullSiteCallsResponse();
        }

        var response = new PullSiteCallsResponse
        {
            // batch_size saturated → tell central to issue a follow-up pull with
            // an advanced cursor. The site doesn't compute the cursor — central
            // walks it forward from the last returned UpdatedAtUtc. Unlike
            // PullAuditEvents there is no MarkReconciled step: the tracking store
            // is the operational source of truth and the central SiteCalls mirror
            // is upsert-on-newer, so re-reading rows is a harmless no-op.
            MoreAvailable = operationals.Count >= request.BatchSize,
        };
        foreach (var op in operationals)
        {
            response.Operationals.Add(SiteCallDtoMapper.ToDto(op));
        }

        return response;
    }

    /// <summary>
    /// Tracks a single active stream so cleanup only removes its own entry.
    /// </summary>
    private sealed record StreamEntry(CancellationTokenSource Cts);
}