using System.Runtime.CompilerServices;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Grpc;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;

namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Grpc;

public sealed class EventStreamServiceTests
{
    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);

    /// <summary>Verifies that events from the worker stream maintain their original sequence order.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_YieldsEventsInWorkerOrder()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);
        FakeSessionManager sessionManager = new(session);
        using GatewayMetrics metrics = new();
        EventStreamService service = CreateService(sessionManager, metrics: metrics);
        workerClient.Events.Add(CreateWorkerEvent(sequence: 10, MxEventFamily.OnDataChange));
        workerClient.Events.Add(CreateWorkerEvent(sequence: 11, MxEventFamily.OnWriteComplete));
        workerClient.CompleteAfterConfiguredEvents = true;

        List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);

        Assert.Equal([10UL, 11UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
        Assert.Equal(MxEventFamily.OnDataChange, events[0].Family);
        Assert.Equal(MxEventFamily.OnWriteComplete, events[1].Family);
        Assert.Equal(1, metrics.GetSnapshot().StreamDisconnects);
    }

    /// <summary>
    ///     Owner-scoped attach: the API key that opened a session may attach its event
    ///     stream — the caller key equals the session owner, so streaming proceeds normally.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenCallerKeyMatchesOwner_Streams()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient, ownerKeyId: "key-owner");
        EventStreamService service = CreateService(new FakeSessionManager(session));
        workerClient.Events.Add(CreateWorkerEvent(sequence: 5, MxEventFamily.OnDataChange));
        workerClient.CompleteAfterConfiguredEvents = true;

        List<MxEvent> events = [];
        await foreach (MxEvent mxEvent in service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: "key-owner", CancellationToken.None)
            .WithCancellation(CancellationToken.None))
        {
            events.Add(mxEvent);
        }

        Assert.Equal([5UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
    }

    /// <summary>
    ///     Owner-scoped attach, security control: a caller whose API key differs from
    ///     the key that opened the session is rejected with a <see cref="SessionManagerErrorCode.PermissionDenied"/>
    ///     fault before any events are streamed — closing the reconnect/fan-out trust-boundary hole.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenCallerKeyDiffersFromOwner_ThrowsPermissionDenied()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient, ownerKeyId: "key-owner");
        EventStreamService service = CreateService(new FakeSessionManager(session));

        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () =>
        {
            await foreach (MxEvent _ in service
                .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: "key-intruder", CancellationToken.None)
                .WithCancellation(CancellationToken.None))
            {
                // No event should be yielded — the owner check runs before the first attach.
            }
        });

        Assert.Equal(SessionManagerErrorCode.PermissionDenied, exception.ErrorCode);
        Assert.Equal(0, session.ActiveEventSubscriberCount);
    }

    /// <summary>Verifies that a second event subscriber is rejected when one is already active.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenSecondSubscriberStarts_RejectsClearly()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));
        using CancellationTokenSource firstSubscriberCancellation = new();
        await using IAsyncEnumerator<MxEvent> firstSubscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, firstSubscriberCancellation.Token)
            .GetAsyncEnumerator(firstSubscriberCancellation.Token);
        Task<bool> firstMoveTask = firstSubscriber.MoveNextAsync().AsTask();

        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
        await using IAsyncEnumerator<MxEvent> secondSubscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
            async () => await secondSubscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));

        Assert.Equal(SessionManagerErrorCode.EventSubscriberAlreadyActive, exception.ErrorCode);
        await firstSubscriberCancellation.CancelAsync();
        await Assert.ThrowsAnyAsync<OperationCanceledException>(
            async () => await firstMoveTask.WaitAsync(TestTimeout));
        await firstSubscriber.DisposeAsync();
        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 0);
    }

    /// <summary>Verifies that canceling an event stream detaches the subscriber cleanly.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenCanceled_DetachesSubscriber()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));
        using CancellationTokenSource cancellationTokenSource = new();
        await using IAsyncEnumerator<MxEvent> subscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token)
            .GetAsyncEnumerator(cancellationTokenSource.Token);
        Task<bool> moveTask = subscriber.MoveNextAsync().AsTask();

        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
        await cancellationTokenSource.CancelAsync();
        await Assert.ThrowsAnyAsync<OperationCanceledException>(
            async () => await moveTask.WaitAsync(TestTimeout));
        await subscriber.DisposeAsync();

        await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 0);
    }

    /// <summary>Verifies that disposing an event stream with buffered events resets the queue depth metric.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenDisposedWithBufferedEvents_ResetsStreamQueueDepth()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);
        using GatewayMetrics metrics = new();
        EventStreamService service = CreateService(
            new FakeSessionManager(session),
            metrics,
            queueCapacity: 8);
        workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange));
        workerClient.Events.Add(CreateWorkerEvent(sequence: 2, MxEventFamily.OnDataChange));
        workerClient.Events.Add(CreateWorkerEvent(sequence: 3, MxEventFamily.OnDataChange));
        workerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> subscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        Assert.True(await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
        await WaitUntilAsync(() => metrics.GetSnapshot().GrpcEventStreamQueueDepth > 0);

        await subscriber.DisposeAsync();

        await WaitUntilAsync(() => metrics.GetSnapshot().GrpcEventStreamQueueDepth == 0);
    }

    /// <summary>Verifies that queue depth metrics correctly track concurrent event streams across multiple sessions.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WithConcurrentStreams_TracksAggregateQueueDepth()
    {
        FakeWorkerClient firstWorkerClient = new();
        FakeWorkerClient secondWorkerClient = new();
        GatewaySession firstSession = CreateReadySession(firstWorkerClient, "session-events-1");
        GatewaySession secondSession = CreateReadySession(secondWorkerClient, "session-events-2");
        using GatewayMetrics metrics = new();
        EventStreamService service = CreateService(
            new FakeSessionManager(firstSession, secondSession),
            metrics,
            queueCapacity: 8);
        for (ulong sequence = 1; sequence <= 3; sequence++)
        {
            firstWorkerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
            secondWorkerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        firstWorkerClient.CompleteAfterConfiguredEvents = true;
        secondWorkerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> firstSubscriber = service
            .StreamEventsAsync(CreateRequest(firstSession.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        await using IAsyncEnumerator<MxEvent> secondSubscriber = service
            .StreamEventsAsync(CreateRequest(secondSession.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        Assert.True(await firstSubscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
        Assert.True(await secondSubscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
        await WaitUntilAsync(() => metrics.GetSnapshot().GrpcEventStreamQueueDepth == 4);

        await firstSubscriber.DisposeAsync();

        await WaitUntilAsync(() => metrics.GetSnapshot().GrpcEventStreamQueueDepth == 2);

        await secondSubscriber.DisposeAsync();

        await WaitUntilAsync(() => metrics.GetSnapshot().GrpcEventStreamQueueDepth == 0);
    }

    /// <summary>
    ///     A per-subscriber channel overflow in the session's
    ///     <see cref="SessionEventDistributor"/> faults the whole session under the legacy
    ///     single-subscriber FailFast policy (the default, single-subscriber mode) and records
    ///     the overflow + fault metrics. The distributor completes this subscriber's channel
    ///     with the overflow fault, which surfaces here as the same
    ///     <see cref="SessionManagerErrorCode.EventQueueOverflow"/> the pre-epic per-RPC
    ///     overflow produced.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenStreamQueueOverflows_FaultsSessionAndReportsOverflow()
    {
        FakeWorkerClient workerClient = new();
        using GatewayMetrics metrics = new();
        GatewaySession session = CreateReadySession(
            workerClient,
            queueCapacity: 1,
            metrics: metrics,
            backpressurePolicy: EventBackpressurePolicy.FailFast);
        EventStreamService service = CreateService(
            new FakeSessionManager(session),
            metrics,
            queueCapacity: 1);
        for (ulong sequence = 1; sequence <= 50; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        workerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> subscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        // The pump fans 50 events into a subscriber channel with capacity 1 faster than this
        // single reader drains, so one of the reads observes the terminal overflow fault.
        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
            async () =>
            {
                while (await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout))
                {
                }
            });

        Assert.Equal(SessionManagerErrorCode.EventQueueOverflow, exception.ErrorCode);
        await WaitUntilAsync(() => session.State == SessionState.Faulted);
        Assert.Equal(SessionState.Faulted, session.State);
        GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
        Assert.Equal(1, snapshot.QueueOverflows);
        Assert.Equal(1, snapshot.Faults);
        // The finally block in StreamEventsAsync calls StreamDisconnected("Detached") on the
        // overflow+fault path too; pin it here so a regression removing that call is caught.
        Assert.Equal(1, snapshot.StreamDisconnects);
    }

    /// <summary>
    ///     Under the DisconnectSubscriber policy a per-subscriber channel overflow
    ///     disconnects only that subscriber's stream (terminal
    ///     <see cref="SessionManagerErrorCode.EventQueueOverflow"/>) and records the overflow
    ///     metric, but leaves the session <see cref="SessionState.Ready"/> and records no
    ///     fault. The session, pump, and any other subscribers are unaffected.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenStreamQueueOverflowsWithDisconnectPolicy_LeavesSessionReady()
    {
        FakeWorkerClient workerClient = new();
        using GatewayMetrics metrics = new();
        GatewaySession session = CreateReadySession(
            workerClient,
            queueCapacity: 1,
            metrics: metrics,
            backpressurePolicy: EventBackpressurePolicy.DisconnectSubscriber);
        EventStreamService service = CreateService(
            new FakeSessionManager(session),
            metrics,
            queueCapacity: 1,
            backpressurePolicy: EventBackpressurePolicy.DisconnectSubscriber);
        for (ulong sequence = 1; sequence <= 50; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        workerClient.CompleteAfterConfiguredEvents = true;
        await using IAsyncEnumerator<MxEvent> subscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
            async () =>
            {
                while (await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout))
                {
                }
            });

        Assert.Equal(SessionManagerErrorCode.EventQueueOverflow, exception.ErrorCode);
        Assert.Equal(SessionState.Ready, session.State);
        GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
        Assert.Equal(1, snapshot.QueueOverflows);
        Assert.Equal(0, snapshot.Faults);
        Assert.Equal(1, snapshot.StreamDisconnects);
    }

    /// <summary>Verifies that the event stream does not synthesize OperationComplete events from write completions.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_DoesNotSynthesizeOperationComplete()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));
        workerClient.Events.Add(CreateWorkerEvent(sequence: 10, MxEventFamily.OnWriteComplete));
        workerClient.CompleteAfterConfiguredEvents = true;

        List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);

        MxEvent mxEvent = Assert.Single(events);
        Assert.Equal(MxEventFamily.OnWriteComplete, mxEvent.Family);
        Assert.DoesNotContain(events, candidate => candidate.Family == MxEventFamily.OperationComplete);
    }

    /// <summary>Verifies that a terminal fault from the worker event stream propagates and faults the session.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_WhenWorkerEventStreamFaults_PropagatesTerminalFault()
    {
        FakeWorkerClient workerClient = new()
        {
            TerminalException = new WorkerClientException(
                WorkerClientErrorCode.WorkerFaulted,
                "worker terminal fault"),
        };
        GatewaySession session = CreateReadySession(workerClient);
        using GatewayMetrics metrics = new();
        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
        await using IAsyncEnumerator<MxEvent> subscriber = service
            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
            async () => await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));

        Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
        Assert.Equal(SessionState.Faulted, session.State);
        Assert.Equal(1, metrics.GetSnapshot().Faults);
    }

    /// <summary>
    ///     Resuming with AfterWorkerSequence inside the retained window replays exactly
    ///     the newer retained events (in order, no dup) then live, with NO ReplayGap sentinel.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_ResumeWithinRetainedWindow_ReplaysNewerThenLive_NoSentinel()
    {
        System.Threading.Channels.Channel<WorkerEvent> live =
            System.Threading.Channels.Channel.CreateUnbounded<WorkerEvent>();
        FakeWorkerClient workerClient = new() { LiveEvents = live };
        for (ulong sequence = 1; sequence <= 5; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));

        // Prime: drain the static 1..5 through a first subscriber so the replay ring retains them.
        await PrimeReplayAsync(service, session.SessionId, expectedCount: 5);

        // Resume after sequence 2: retained window [1..5] covers it — replay 3,4,5 then live.
        await using IAsyncEnumerator<MxEvent> resume = service
            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        MxEvent r3 = await ReadNextAsync(resume);
        MxEvent r4 = await ReadNextAsync(resume);
        MxEvent r5 = await ReadNextAsync(resume);
        Assert.Equal(new ulong[] { 3, 4, 5 }, new[] { r3.WorkerSequence, r4.WorkerSequence, r5.WorkerSequence });
        Assert.Null(r3.ReplayGap);

        // No sentinel anywhere; next is a LIVE event.
        live.Writer.TryWrite(CreateWorkerEvent(6, MxEventFamily.OnDataChange));
        MxEvent liveEvent = await ReadNextAsync(resume);
        Assert.Equal(6ul, liveEvent.WorkerSequence);
        Assert.Null(liveEvent.ReplayGap);
    }

    /// <summary>
    ///     Resuming with AfterWorkerSequence older than the oldest retained yields the
    ///     ReplayGap sentinel FIRST (correct requested/oldest), then the retained tail, then live.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_ResumeOlderThanOldestRetained_EmitsSentinelFirst_ThenTailThenLive()
    {
        System.Threading.Channels.Channel<WorkerEvent> live =
            System.Threading.Channels.Channel.CreateUnbounded<WorkerEvent>();
        FakeWorkerClient workerClient = new() { LiveEvents = live };
        for (ulong sequence = 1; sequence <= 5; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        // Replay capacity 3 retains only 3,4,5; 1,2 are evicted.
        GatewaySession session = CreateReadySession(workerClient, replayBufferCapacity: 3);
        EventStreamService service = CreateService(new FakeSessionManager(session));

        await PrimeReplayAsync(service, session.SessionId, expectedCount: 5);

        // Resume after 1: events 1,2 are below the oldest retained (3) and were evicted, so
        // they are unrecoverable => sentinel first, then the retained tail 3,4,5, then live.
        await using IAsyncEnumerator<MxEvent> realResume = service
            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 1), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        MxEvent sentinel = await ReadNextAsync(realResume);
        Assert.NotNull(sentinel.ReplayGap);
        Assert.Equal(1ul, sentinel.ReplayGap.RequestedAfterSequence);
        Assert.Equal(3ul, sentinel.ReplayGap.OldestAvailableSequence);
        Assert.Equal(MxEventFamily.Unspecified, sentinel.Family);
        Assert.Equal(session.SessionId, sentinel.SessionId);

        MxEvent r3 = await ReadNextAsync(realResume);
        MxEvent r4 = await ReadNextAsync(realResume);
        MxEvent r5 = await ReadNextAsync(realResume);
        Assert.Equal(new ulong[] { 3, 4, 5 }, new[] { r3.WorkerSequence, r4.WorkerSequence, r5.WorkerSequence });
        Assert.Null(r3.ReplayGap);

        live.Writer.TryWrite(CreateWorkerEvent(6, MxEventFamily.OnDataChange));
        MxEvent liveEvent = await ReadNextAsync(realResume);
        Assert.Equal(6ul, liveEvent.WorkerSequence);
    }

    /// <summary>
    ///     The replay→live boundary is contiguous — no duplicate and no skip — even
    ///     when events span the handoff.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_ResumeHandoff_IsContiguous_NoDuplicateNoSkip()
    {
        System.Threading.Channels.Channel<WorkerEvent> live =
            System.Threading.Channels.Channel.CreateUnbounded<WorkerEvent>();
        FakeWorkerClient workerClient = new() { LiveEvents = live };
        for (ulong sequence = 1; sequence <= 4; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));

        await PrimeReplayAsync(service, session.SessionId, expectedCount: 4);

        // Resume after 2: replay 3,4 then live 5,6,7. Collect across the boundary and assert
        // the full sequence is contiguous with no duplicate and no skip.
        await using IAsyncEnumerator<MxEvent> resume = service
            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        List<ulong> collected = [];
        collected.Add((await ReadNextAsync(resume)).WorkerSequence); // 3
        collected.Add((await ReadNextAsync(resume)).WorkerSequence); // 4

        for (ulong sequence = 5; sequence <= 7; sequence++)
        {
            live.Writer.TryWrite(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
            collected.Add((await ReadNextAsync(resume)).WorkerSequence);
        }

        Assert.Equal(new ulong[] { 3, 4, 5, 6, 7 }, collected);
    }

    /// <summary>
    ///     The per-item filter applies to REPLAYED events identically to live — a
    ///     replayed event at/below the requested watermark is never delivered.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_ResumeReplay_AppliesPerItemFilter_DropsAtOrBelowWatermark()
    {
        System.Threading.Channels.Channel<WorkerEvent> live =
            System.Threading.Channels.Channel.CreateUnbounded<WorkerEvent>();
        FakeWorkerClient workerClient = new() { LiveEvents = live };
        for (ulong sequence = 1; sequence <= 5; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));

        await PrimeReplayAsync(service, session.SessionId, expectedCount: 5);

        // Resume after 3: only 4,5 may be delivered. Events 1,2,3 — present in the ring but at
        // or below the watermark — must be filtered out of the replay, never seen. The first two
        // reads must be exactly 4 then 5 (no sentinel, no <=3 event); a live tag confirms the
        // stream resumed live strictly after 5.
        await using IAsyncEnumerator<MxEvent> resume = service
            .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 3), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();

        MxEvent first = await ReadNextAsync(resume);
        MxEvent second = await ReadNextAsync(resume);
        Assert.Equal(4ul, first.WorkerSequence);
        Assert.Equal(5ul, second.WorkerSequence);
        Assert.Null(first.ReplayGap);
        Assert.Null(second.ReplayGap);

        // The very next delivered event is the live 6 — proving nothing <=3 slipped in and the
        // handoff resumed strictly after the replay tail.
        live.Writer.TryWrite(CreateWorkerEvent(6, MxEventFamily.OnDataChange));
        MxEvent liveEvent = await ReadNextAsync(resume);
        Assert.Equal(6ul, liveEvent.WorkerSequence);
    }

    /// <summary>
    ///     AfterWorkerSequence == 0 is a fresh stream (not a resume) — no replay, no
    ///     sentinel, just live events as before.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEventsAsync_FreshStreamAfterSequenceZero_NoReplayNoSentinel()
    {
        FakeWorkerClient workerClient = new();
        for (ulong sequence = 1; sequence <= 3; sequence++)
        {
            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
        }

        workerClient.CompleteAfterConfiguredEvents = true;
        GatewaySession session = CreateReadySession(workerClient);
        EventStreamService service = CreateService(new FakeSessionManager(session));

        List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);

        Assert.Equal(new ulong[] { 1, 2, 3 }, events.Select(e => e.WorkerSequence));
        Assert.DoesNotContain(events, e => e.ReplayGap is not null);
    }

    // Drains the first `expectedCount` events through a throwaway subscriber so the session's
    // replay ring retains them, then disposes the subscriber. The pump (started on first
    // attach) keeps running for the session, so subsequent resume attaches see the retained
    // events.
    private static async Task PrimeReplayAsync(
        EventStreamService service,
        string sessionId,
        int expectedCount)
    {
        await using IAsyncEnumerator<MxEvent> primer = service
            .StreamEventsAsync(CreateRequest(sessionId), callerKeyId: null, CancellationToken.None)
            .GetAsyncEnumerator();
        for (int i = 0; i < expectedCount; i++)
        {
            await ReadNextAsync(primer);
        }
    }

    private static async Task<MxEvent> ReadNextAsync(IAsyncEnumerator<MxEvent> enumerator)
    {
        Assert.True(await enumerator.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
        return enumerator.Current;
    }

    private static EventStreamService CreateService(
        FakeSessionManager sessionManager,
        GatewayMetrics? metrics = null,
        int queueCapacity = 8,
        EventBackpressurePolicy backpressurePolicy = EventBackpressurePolicy.FailFast)
    {
        return new EventStreamService(
            sessionManager,
            Options.Create(new GatewayOptions
            {
                Events = new EventOptions
                {
                    QueueCapacity = queueCapacity,
                    BackpressurePolicy = backpressurePolicy,
                },
            }),
            metrics ?? new GatewayMetrics());
    }

    private static async Task<List<MxEvent>> CollectEventsAsync(
        EventStreamService service,
        string sessionId)
    {
        List<MxEvent> events = [];
        await foreach (MxEvent mxEvent in service
            .StreamEventsAsync(CreateRequest(sessionId), callerKeyId: null, CancellationToken.None)
            .WithCancellation(CancellationToken.None))
        {
            events.Add(mxEvent);
        }

        return events;
    }

    private static StreamEventsRequest CreateRequest(string sessionId, ulong afterWorkerSequence = 0)
    {
        return new StreamEventsRequest
        {
            SessionId = sessionId,
            AfterWorkerSequence = afterWorkerSequence,
        };
    }

    private static GatewaySession CreateReadySession(
        FakeWorkerClient workerClient,
        string sessionId = "session-events",
        int queueCapacity = 8,
        GatewayMetrics? metrics = null,
        EventBackpressurePolicy backpressurePolicy = EventBackpressurePolicy.FailFast,
        int replayBufferCapacity = 1024,
        string? ownerKeyId = null)
    {
        // The per-subscriber overflow policy now lives in the session's
        // SessionEventDistributor, so the session must share the same metrics sink and
        // backpressure policy the overflow assertions observe. queueCapacity flows into the
        // distributor's per-subscriber channel bound, which is what overflows.
        GatewaySession session = new(
            sessionId,
            GatewayContractInfo.DefaultBackendName,
            "pipe",
            "nonce",
            "client",
            ownerKeyId: ownerKeyId,
            "client-session",
            "client-correlation",
            TimeSpan.FromSeconds(30),
            TimeSpan.FromSeconds(30),
            TimeSpan.FromSeconds(10),
            TimeSpan.FromMinutes(30),
            DateTimeOffset.UtcNow,
            new SessionEventStreaming(
                new MxAccessGrpcMapper(),
                new EventOptions
                {
                    QueueCapacity = queueCapacity,
                    BackpressurePolicy = backpressurePolicy,
                    ReplayBufferCapacity = replayBufferCapacity,
                    ReplayRetentionSeconds = 0,
                },
                NullLogger<SessionEventDistributor>.Instance,
                TimeProvider.System,
                metrics ?? new GatewayMetrics()));
        session.AttachWorkerClient(workerClient);
        session.MarkReady();

        return session;
    }

    private static WorkerEvent CreateWorkerEvent(
        ulong sequence,
        MxEventFamily family)
    {
        MxEvent mxEvent = new()
        {
            SessionId = "session-events",
            Family = family,
            WorkerSequence = sequence,
        };

        switch (family)
        {
            case MxEventFamily.OnDataChange:
                mxEvent.OnDataChange = new OnDataChangeEvent();
                break;
            case MxEventFamily.OnWriteComplete:
                mxEvent.OnWriteComplete = new OnWriteCompleteEvent();
                break;
            case MxEventFamily.OperationComplete:
                mxEvent.OperationComplete = new OperationCompleteEvent();
                break;
            case MxEventFamily.OnBufferedDataChange:
                mxEvent.OnBufferedDataChange = new OnBufferedDataChangeEvent();
                break;
        }

        return new WorkerEvent
        {
            Event = mxEvent,
        };
    }

    // The real-clock deadline here is load-sensitive on a wide host (windev runs this suite
    // 36-way parallel). Surfacing the unmet condition instead of letting the bare
    // TaskCanceledException escape is what makes such a failure diagnosable rather than a
    // mystery cancellation attributed to "the environment".
    private static async Task WaitUntilAsync(
        Func<bool> predicate,
        [CallerArgumentExpression(nameof(predicate))] string? predicateExpression = null)
    {
        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
        while (!predicate())
        {
            try
            {
                await Task.Delay(TimeSpan.FromMilliseconds(10), cancellationTokenSource.Token);
            }
            catch (OperationCanceledException)
            {
                Assert.Fail(
                    $"Timed out after {TestTimeout} waiting for condition: {predicateExpression}");
            }
        }
    }

    /// <summary>Fake session manager for testing event streams.</summary>
    private sealed class FakeSessionManager : ISessionManager
    {
        private readonly IReadOnlyDictionary<string, GatewaySession> _sessions;

        /// <summary>Initializes a new instance of the FakeSessionManager.</summary>
        /// <param name="sessions">Sessions to manage.</param>
        public FakeSessionManager(params GatewaySession[] sessions)
        {
            _sessions = sessions.ToDictionary(session => session.SessionId, StringComparer.Ordinal);
        }

        /// <inheritdoc />
        public Task<GatewaySession> OpenSessionAsync(
            SessionOpenRequest request,
            string? clientIdentity,
            string? ownerKeyId,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(_sessions.Values.First());
        }

        /// <inheritdoc />
        public bool TryGetSession(
            string sessionId,
            out GatewaySession gatewaySession)
        {
            return _sessions.TryGetValue(sessionId, out gatewaySession!);
        }

        /// <inheritdoc />
        public Task<WorkerCommandReply> InvokeAsync(
            string sessionId,
            WorkerCommand command,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(new WorkerCommandReply());
        }

        /// <inheritdoc />
        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            return _sessions[sessionId].ReadEventsAsync(cancellationToken);
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> KillWorkerAsync(
            string sessionId,
            string reason,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
        }

        /// <inheritdoc />
        public Task<int> CloseExpiredLeasesAsync(
            DateTimeOffset now,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(0);
        }

        /// <inheritdoc />
        public Task ShutdownAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }

    /// <summary>Fake worker client for testing event streams.</summary>
    private sealed class FakeWorkerClient : IWorkerClient
    {
        /// <summary>Gets the list of queued worker events.</summary>
        public List<WorkerEvent> Events { get; } = [];

        /// <summary>Gets or sets whether to complete the event stream after configured events are yielded.</summary>
        public bool CompleteAfterConfiguredEvents { get; set; }

        /// <summary>
        ///     Optional live channel source. When set, the worker drains the static
        ///     <see cref="Events"/> first, then streams from this channel until it completes,
        ///     letting a test feed events on demand (e.g. to exercise replay→live handoff).
        /// </summary>
        public System.Threading.Channels.Channel<WorkerEvent>? LiveEvents { get; init; }

        /// <summary>Gets or sets an optional exception to throw as a terminal event stream fault.</summary>
        public Exception? TerminalException { get; init; }

        /// <inheritdoc />
        public string SessionId { get; } = "session-events";

        /// <inheritdoc />
        public int? ProcessId { get; } = 4321;

        /// <inheritdoc />
        public WorkerClientState State { get; private set; } = WorkerClientState.Ready;

        /// <inheritdoc />
        public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;

        /// <inheritdoc />
        public Task StartAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public Task<WorkerCommandReply> InvokeAsync(
            WorkerCommand command,
            TimeSpan timeout,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(new WorkerCommandReply());
        }

        /// <inheritdoc />
        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            foreach (WorkerEvent workerEvent in Events)
            {
                cancellationToken.ThrowIfCancellationRequested();
                yield return workerEvent;
            }

            if (TerminalException is not null)
            {
                throw TerminalException;
            }

            if (LiveEvents is not null)
            {
                await foreach (WorkerEvent liveEvent in LiveEvents.Reader
                    .ReadAllAsync(cancellationToken)
                    .ConfigureAwait(false))
                {
                    yield return liveEvent;
                }

                yield break;
            }

            if (CompleteAfterConfiguredEvents)
            {
                yield break;
            }

            await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
        }

        /// <inheritdoc />
        public Task ShutdownAsync(
            TimeSpan timeout,
            CancellationToken cancellationToken)
        {
            State = WorkerClientState.Closed;
            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public void Kill(string reason)
        {
            State = WorkerClientState.Faulted;
        }

        /// <summary>No-op disposal; the fake holds no unmanaged resources.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public ValueTask DisposeAsync()
        {
            return ValueTask.CompletedTask;
        }
    }
}