using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Microsoft.Extensions.Time.Testing;
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.Security.Authentication;
using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;

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

/// <summary>
///     End-to-end reconnect/replay tests through the real gRPC <c>StreamEvents</c> path via the
///     fake worker harness (server half). A single subscriber detaches (cancel + await
///     the stream task so its lease is fully disposed), the session is retained by detach-grace
///     while the distributor pump keeps appending worker events to the replay ring, and a second
///     stream reconnects with <see cref="StreamEventsRequest.AfterWorkerSequence"/>. Covers both
///     the no-gap resume (cursor inside the retained window) and the <c>ReplayGap</c> sentinel
///     (cursor predates the oldest retained event after capacity eviction).
/// </summary>
/// <remarks>
///     These tests run in single-subscriber mode (<c>AllowMultipleEventSubscribers=false</c>).
///     Reconnect works because the first stream is FULLY detached before the reconnect attaches:
///     awaiting the first stream task runs <c>EventStreamService</c>'s finally block, which
///     disposes the subscriber lease and drops the session's active-subscriber count back to
///     zero, so the reconnect's <c>AttachEventSubscriberWithReplay</c> sees an empty slot rather
///     than an "already active" rejection. The distributor (and its replay ring) is created once
///     per session and survives detach, so events emitted while no subscriber is attached are
///     retained for the reconnecting stream.
/// </remarks>
public sealed class GatewayEndToEndReconnectReplayTests
{
    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
    private const int ServerHandle = 3001;
    private const int ItemHandle = 4002;

    /// <summary>
    ///     Reconnecting inside the retained window replays exactly the events newer than the
    ///     resume cursor — the retained tail of the first batch plus the events emitted while
    ///     detached — in strictly ascending order, with no duplicates and no <c>ReplayGap</c>
    ///     sentinel.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEvents_ReconnectInsideRetainedWindow_ReplaysTailNoGap()
    {
        const int firstBatch = 4;
        const int secondBatch = 3;

        GatedEventFakeWorkerProcessLauncher launcher = new();
        // Capacity 16 retains every event emitted here (7 total), so nothing is evicted and a
        // resume from a middle cursor never produces a gap.
        await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16);

        string sessionId = await OpenSessionAsync(fixture, "reconnect-no-gap");

        // ---- first connection: receive the first batch, capture its real sequences ----
        using CancellationTokenSource writer1Cts = new();
        RecordingServerStreamWriter<MxEvent> writer1 = new();
        Task stream1Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId },
                writer1,
                new TestServerCallContext(cancellationToken: writer1Cts.Token)));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
        await WireUpAdviseAsync(fixture, sessionId);

        for (int i = 0; i < firstBatch; i++)
        {
            launcher.AllowNextEvent();
        }

        IReadOnlyList<MxEvent> batch1 = await writer1.WaitForMessageCountAsync(firstBatch, TestTimeout);
        ulong[] batch1Sequences = batch1.Select(e => e.WorkerSequence).ToArray();

        // Resume cursor: a middle event of the first batch. Everything strictly newer than this
        // must be redelivered to the reconnecting stream.
        ulong cursor = batch1Sequences[1];
        int retainedTailFromBatch1 = batch1Sequences.Count(s => s > cursor);
        Assert.Equal(2, retainedTailFromBatch1); // sanity: events at index 2 and 3

        // ---- fully detach writer1 (cancel + await so its lease is disposed) ----
        await DetachAsync(writer1Cts, stream1Task);

        // ---- emit more events while detached; the ring must retain them for the reconnect ----
        for (int i = 0; i < secondBatch; i++)
        {
            launcher.AllowNextEvent();
        }

        // ---- reconnect with the middle cursor ----
        RecordingServerStreamWriter<MxEvent> writer2 = new();
        Task stream2Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = cursor },
                writer2,
                new TestServerCallContext()));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);

        int expected = retainedTailFromBatch1 + secondBatch;
        IReadOnlyList<MxEvent> resumed = await writer2.WaitForMessageCountAsync(expected, TestTimeout);

        // ---- tear down before asserting so a hang can't wedge the run ----
        launcher.StopEmitting();
        await CloseAndDrainAsync(fixture, sessionId, stream2Task, launcher);

        // ---- assertions ----
        Assert.Equal(expected, resumed.Count);

        // No sentinel: every message is a real event, none carries a ReplayGap.
        Assert.All(resumed, e => Assert.Null(e.ReplayGap));

        // Every replayed/live event is strictly newer than the cursor.
        Assert.All(resumed, e => Assert.True(
            e.WorkerSequence > cursor,
            $"Event sequence {e.WorkerSequence} is not newer than cursor {cursor}."));

        // Strictly ascending, no duplicates.
        for (int i = 1; i < resumed.Count; i++)
        {
            Assert.True(
                resumed[i].WorkerSequence > resumed[i - 1].WorkerSequence,
                $"Sequences must be strictly ascending: {resumed[i - 1].WorkerSequence} then {resumed[i].WorkerSequence}.");
        }

        Assert.Equal(resumed.Count, resumed.Select(e => e.WorkerSequence).Distinct().Count());

        // The retained tail of the first batch (the two events newer than the cursor) is
        // replayed first, before the events emitted while detached.
        Assert.Equal(batch1Sequences[2], resumed[0].WorkerSequence);
        Assert.Equal(batch1Sequences[3], resumed[1].WorkerSequence);
    }

    /// <summary>
    ///     Reconnecting with a cursor that predates the oldest retained event (after capacity
    ///     eviction) yields the <c>ReplayGap</c> sentinel FIRST — family unspecified, no body, no
    ///     per-item fields, correct requested/oldest sequences — followed by exactly the retained
    ///     tail, in ascending order.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task StreamEvents_ReconnectWithStaleCursor_EmitsReplayGapSentinelFirst()
    {
        const int capacity = 3;
        const int totalEvents = 6;

        GatedEventFakeWorkerProcessLauncher launcher = new();
        // Small capacity forces eviction: with 6 events emitted and a 3-slot ring, the first 3
        // are evicted and only the newest 3 remain replayable.
        await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity);

        string sessionId = await OpenSessionAsync(fixture, "reconnect-gap");

        using CancellationTokenSource writer1Cts = new();
        RecordingServerStreamWriter<MxEvent> writer1 = new();
        Task stream1Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId },
                writer1,
                new TestServerCallContext(cancellationToken: writer1Cts.Token)));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
        await WireUpAdviseAsync(fixture, sessionId);

        // Emit more events than the ring can hold. Waiting for writer1 to receive all of them
        // proves the pump has appended (and evicted) every event, so the ring is settled to its
        // final newest-`capacity` contents before we reconnect.
        for (int i = 0; i < totalEvents; i++)
        {
            launcher.AllowNextEvent();
        }

        IReadOnlyList<MxEvent> all = await writer1.WaitForMessageCountAsync(totalEvents, TestTimeout);
        ulong[] sequences = all.Select(e => e.WorkerSequence).ToArray();

        // With capacity C and N total events, the newest C are retained; the oldest still
        // available is the (N-C+1)th event — index N-C in emission (ascending) order.
        ulong oldestAvailable = sequences[totalEvents - capacity];

        await DetachAsync(writer1Cts, stream1Task);

        // Resume after sequence 1: the real event sequences are envelope-numbered and start well
        // above 1 (startup + command-reply envelopes consume the low numbers), so cursor 1
        // predates every event and, with the oldest three already evicted, forces a gap.
        const ulong staleCursor = 1;
        RecordingServerStreamWriter<MxEvent> writer2 = new();
        Task stream2Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = staleCursor },
                writer2,
                new TestServerCallContext()));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);

        // sentinel + the retained tail (capacity events, all newer than the stale cursor).
        int expected = 1 + capacity;
        IReadOnlyList<MxEvent> resumed = await writer2.WaitForMessageCountAsync(expected, TestTimeout);

        launcher.StopEmitting();
        await CloseAndDrainAsync(fixture, sessionId, stream2Task, launcher);

        // ---- the first message is the ReplayGap sentinel ----
        Assert.Equal(expected, resumed.Count);

        MxEvent sentinel = resumed[0];
        Assert.NotNull(sentinel.ReplayGap);
        Assert.Equal(MxEventFamily.Unspecified, sentinel.Family);
        Assert.Equal(MxEvent.BodyOneofCase.None, sentinel.BodyCase);
        Assert.Equal(sessionId, sentinel.SessionId);
        Assert.Equal(staleCursor, sentinel.ReplayGap.RequestedAfterSequence);
        Assert.Equal(oldestAvailable, sentinel.ReplayGap.OldestAvailableSequence);

        // ---- the sentinel is followed by the retained tail, ascending, no further sentinels ----
        IReadOnlyList<MxEvent> tail = resumed.Skip(1).ToArray();
        Assert.Equal(capacity, tail.Count);
        Assert.All(tail, e => Assert.Null(e.ReplayGap));
        Assert.All(tail, e => Assert.True(
            e.WorkerSequence >= oldestAvailable,
            $"Retained event {e.WorkerSequence} is older than the oldest available {oldestAvailable}."));

        ulong[] expectedTail = sequences.Skip(totalEvents - capacity).ToArray();
        Assert.Equal(expectedTail, tail.Select(e => e.WorkerSequence).ToArray());
    }

    /// <summary>
    ///     Regression, end to end: when age eviction has emptied the replay ring, the
    ///     <c>ReplayGap</c> sentinel carries the next deliverable sequence (highest observed + 1),
    ///     so a client applying the documented
    ///     <c>after_worker_sequence = oldest_available_sequence - 1</c> formula on its follow-up
    ///     resume receives subsequent live events. Under the pre-fix sentinel value of <c>0</c>
    ///     the formula wrapped to <see cref="ulong.MaxValue"/> and the resumed stream never
    ///     delivered another event.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReconnectAfterFullAgeEvictionResumesWithSentinelFormula()
    {
        const int firstBatch = 4;
        const double retentionSeconds = 30;

        FakeTimeProvider clock = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
        GatedEventFakeWorkerProcessLauncher launcher = new();

        // Capacity is ample; age is the eviction axis under test.
        await using ReconnectReplayGatewayServiceFixture fixture = new(
            launcher,
            replayBufferCapacity: 16,
            replayRetentionSeconds: retentionSeconds,
            timeProvider: clock);

        string sessionId = await OpenSessionAsync(fixture, "reconnect-aged-out");

        using CancellationTokenSource writer1Cts = new();
        RecordingServerStreamWriter<MxEvent> writer1 = new();
        Task stream1Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId },
                writer1,
                new TestServerCallContext(cancellationToken: writer1Cts.Token)));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
        await WireUpAdviseAsync(fixture, sessionId);

        for (int i = 0; i < firstBatch; i++)
        {
            launcher.AllowNextEvent();
        }

        IReadOnlyList<MxEvent> batch1 = await writer1.WaitForMessageCountAsync(firstBatch, TestTimeout);
        ulong[] batch1Sequences = batch1.Select(e => e.WorkerSequence).ToArray();

        // The client's cursor is mid-batch: it detached before consuming the tail, so it is
        // genuinely behind the highest sequence the distributor observed.
        ulong staleCursor = batch1Sequences[1];
        ulong highestSeen = batch1Sequences[firstBatch - 1];

        await DetachAsync(writer1Cts, stream1Task);

        // Age every retained event out of the ring: the next resume finds it empty.
        clock.Advance(TimeSpan.FromSeconds(retentionSeconds * 2));

        // ---- first reconnect: receives only the sentinel, nothing is replayable ----
        using CancellationTokenSource writer2Cts = new();
        RecordingServerStreamWriter<MxEvent> writer2 = new();
        Task stream2Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = staleCursor },
                writer2,
                new TestServerCallContext(cancellationToken: writer2Cts.Token)));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
        IReadOnlyList<MxEvent> gapOnly = await writer2.WaitForMessageCountAsync(1, TestTimeout);

        MxEvent sentinel = gapOnly[0];
        Assert.NotNull(sentinel.ReplayGap);
        Assert.Equal(staleCursor, sentinel.ReplayGap.RequestedAfterSequence);

        // The whole point of the fix: the sentinel anchors the resume at the next sequence that
        // can still be delivered, not at 0.
        Assert.Equal(highestSeen + 1, sentinel.ReplayGap.OldestAvailableSequence);

        await DetachAsync(writer2Cts, stream2Task);

        // ---- second reconnect: the client applies the documented oldest - 1 formula ----
        // Unchecked ulong arithmetic: with the pre-fix sentinel value of 0 this wraps to
        // ulong.MaxValue and the live filter drops every subsequent event.
        ulong resumeCursor = sentinel.ReplayGap.OldestAvailableSequence - 1;

        RecordingServerStreamWriter<MxEvent> writer3 = new();
        Task stream3Task = Task.Run(async () =>
            await fixture.Service.StreamEvents(
                new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = resumeCursor },
                writer3,
                new TestServerCallContext()));

        await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);

        launcher.AllowNextEvent();
        IReadOnlyList<MxEvent> live = await writer3.WaitForMessageCountAsync(1, TestTimeout);

        launcher.StopEmitting();
        await CloseAndDrainAsync(fixture, sessionId, stream3Task, launcher);

        // A live event arrives on the resumed stream, and it is a real event, not another gap.
        MxEvent delivered = live[0];
        Assert.Null(delivered.ReplayGap);
        Assert.Equal(MxEventFamily.OnDataChange, delivered.Family);
        Assert.True(
            delivered.WorkerSequence > highestSeen,
            $"Live event {delivered.WorkerSequence} must be newer than the highest previously seen {highestSeen}.");
    }

    // ---- shared flow helpers ----

    private static async Task<string> OpenSessionAsync(
        ReconnectReplayGatewayServiceFixture fixture,
        string name)
    {
        OpenSessionReply openReply = await fixture.Service.OpenSession(
            new OpenSessionRequest
            {
                ClientSessionName = name,
                ClientCorrelationId = $"open-{name}",
                CommandTimeout = Duration.FromTimeSpan(TestTimeout),
            },
            new TestServerCallContext()).ConfigureAwait(false);

        Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code);
        return openReply.SessionId;
    }

    private static async Task WireUpAdviseAsync(
        ReconnectReplayGatewayServiceFixture fixture,
        string sessionId)
    {
        MxCommandReply registerReply = await fixture.Service.Invoke(
            CreateRegisterRequest(sessionId),
            new TestServerCallContext()).ConfigureAwait(false);
        Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code);

        MxCommandReply addItemReply = await fixture.Service.Invoke(
            CreateAddItemRequest(sessionId, registerReply.Register.ServerHandle),
            new TestServerCallContext()).ConfigureAwait(false);
        Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);

        MxCommandReply adviseReply = await fixture.Service.Invoke(
            CreateAdviseRequest(sessionId, registerReply.Register.ServerHandle, addItemReply.AddItem.ItemHandle),
            new TestServerCallContext()).ConfigureAwait(false);
        Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
    }

    // Cancels the stream's token and awaits the stream task. Awaiting is load-bearing: it lets
    // EventStreamService's finally block dispose the subscriber lease (dropping the session's
    // active-subscriber count to zero) BEFORE the reconnect attaches, so the reconnect never
    // races a still-registered subscriber.
    private static async Task DetachAsync(CancellationTokenSource cts, Task streamTask)
    {
        await cts.CancelAsync().ConfigureAwait(false);
        try
        {
            await streamTask.WaitAsync(TestTimeout).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // Expected: the iterator surfaces the cancellation.
        }
        catch (RpcException rpc) when (rpc.StatusCode == StatusCode.Cancelled)
        {
            // Also acceptable depending on gRPC exception wrapping.
        }
    }

    private static async Task CloseAndDrainAsync(
        ReconnectReplayGatewayServiceFixture fixture,
        string sessionId,
        Task streamTask,
        GatedEventFakeWorkerProcessLauncher launcher)
    {
        await fixture.Service.CloseSession(
            new CloseSessionRequest { SessionId = sessionId, ClientCorrelationId = "close-reconnect" },
            new TestServerCallContext()).ConfigureAwait(false);

        try
        {
            await streamTask.WaitAsync(TestTimeout).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
        }

        await launcher.WorkerTask.WaitAsync(TestTimeout).ConfigureAwait(false);
    }

    // ---- request builders ----

    private static MxCommandRequest CreateRegisterRequest(string sessionId) =>
        new()
        {
            SessionId = sessionId,
            ClientCorrelationId = "register-rr",
            Command = new MxCommand
            {
                Kind = MxCommandKind.Register,
                Register = new RegisterCommand { ClientName = "reconnect-replay-e2e-client" },
            },
        };

    private static MxCommandRequest CreateAddItemRequest(string sessionId, int serverHandle) =>
        new()
        {
            SessionId = sessionId,
            ClientCorrelationId = "add-item-rr",
            Command = new MxCommand
            {
                Kind = MxCommandKind.AddItem,
                AddItem = new AddItemCommand
                {
                    ServerHandle = serverHandle,
                    ItemDefinition = "Galaxy.Tag.Value",
                },
            },
        };

    private static MxCommandRequest CreateAdviseRequest(
        string sessionId,
        int serverHandle,
        int itemHandle) =>
        new()
        {
            SessionId = sessionId,
            ClientCorrelationId = "advise-rr",
            Command = new MxCommand
            {
                Kind = MxCommandKind.Advise,
                Advise = new AdviseCommand { ServerHandle = serverHandle, ItemHandle = itemHandle },
            },
        };

    private static void ConfigureCommandReply(MxCommandReply reply, MxCommandKind kind)
    {
        switch (kind)
        {
            case MxCommandKind.Register:
                reply.Register = new RegisterReply { ServerHandle = ServerHandle };
                break;
            case MxCommandKind.AddItem:
                reply.AddItem = new AddItemReply { ItemHandle = ItemHandle };
                break;
        }
    }

    // ---- fixture ----

    /// <summary>
    ///     Gateway service fixture in single-subscriber mode with a configurable replay ring
    ///     capacity, so each test can retain the whole event history (no gap) or force capacity
    ///     eviction (gap).
    /// </summary>
    private sealed class ReconnectReplayGatewayServiceFixture : IAsyncDisposable
    {
        private readonly GatewayMetrics _metrics = new();
        private readonly SessionRegistry _registry = new();

        /// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
        /// <param name="launcher">Fake worker process launcher backing the session manager.</param>
        /// <param name="replayBufferCapacity">Replay ring capacity for the session's event distributor.</param>
        /// <param name="replayRetentionSeconds">
        ///     Replay retention window. The default keeps age-eviction effectively off for the
        ///     duration of a fast test; the age-eviction test shortens it and drives
        ///     <paramref name="timeProvider"/>.
        /// </param>
        /// <param name="timeProvider">
        ///     Clock handed to the session manager, which flows through to the session's event
        ///     distributor. Pass a fake to make age-eviction deterministic.
        /// </param>
        public ReconnectReplayGatewayServiceFixture(
            IWorkerProcessLauncher launcher,
            int replayBufferCapacity,
            double replayRetentionSeconds = 300,
            TimeProvider? timeProvider = null)
        {
            IOptions<GatewayOptions> options = Options.Create(
                CreateOptions(replayBufferCapacity, replayRetentionSeconds));
            SessionWorkerClientFactory workerClientFactory = new(
                launcher,
                options,
                _metrics,
                NullLoggerFactory.Instance);
            SessionManager sessionManager = new(
                _registry,
                workerClientFactory,
                options,
                _metrics,
                logger: NullLogger<SessionManager>.Instance,
                timeProvider: timeProvider,
                dashboardEventBroadcaster: NullDashboardEventBroadcaster.Instance);
            MxAccessGrpcMapper mapper = new();
            EventStreamService eventStreamService = new(
                sessionManager,
                options,
                _metrics);

            Service = new MxAccessGatewayService(
                sessionManager,
                new GatewayRequestIdentityAccessor(),
                new AllowAllConstraintEnforcer(),
                new MxAccessGrpcRequestValidator(),
                mapper,
                eventStreamService,
                _metrics,
                NullLogger<MxAccessGatewayService>.Instance,
                new FakeGatewayAlarmService());
        }

        /// <summary>Gets the gateway service under test.</summary>
        public MxAccessGatewayService Service { get; }

        /// <summary>
        ///     Polls <see cref="GatewaySession.ActiveEventSubscriberCount"/> for
        ///     <paramref name="sessionId"/> until it reaches <paramref name="n"/>, bounded by
        ///     <paramref name="timeout"/>. Fails the test on timeout. This is the deterministic
        ///     gate that proves the production code has (re)registered a subscriber before the
        ///     test drives events or reconnects.
        /// </summary>
        /// <param name="sessionId">Identifier of the session to poll.</param>
        /// <param name="n">Target subscriber count to wait for.</param>
        /// <param name="timeout">Maximum time to wait before failing the test.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public async Task WaitForSubscriberCountAsync(string sessionId, int n, TimeSpan timeout)
        {
            using CancellationTokenSource deadlineCts = new(timeout);

            while (true)
            {
                if (_registry.TryGet(sessionId, out GatewaySession? session)
                    && session.ActiveEventSubscriberCount >= n)
                {
                    return;
                }

                if (deadlineCts.IsCancellationRequested)
                {
                    int actual = _registry.TryGet(sessionId, out GatewaySession? s)
                        ? s.ActiveEventSubscriberCount
                        : -1;
                    Assert.Fail(
                        $"Timed out waiting for {n} event subscriber(s) on session {sessionId}. "
                        + $"Actual count after {timeout.TotalSeconds:0.#}s: {actual}.");
                }

                await Task.Delay(millisecondsDelay: 5, deadlineCts.Token).ConfigureAwait(false);
            }
        }

        /// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public async ValueTask DisposeAsync()
        {
            foreach (GatewaySession session in _registry.Snapshot())
            {
                await session.DisposeAsync().ConfigureAwait(false);
            }

            _metrics.Dispose();
        }

        private static GatewayOptions CreateOptions(int replayBufferCapacity, double replayRetentionSeconds) =>
            new()
            {
                Worker = new WorkerOptions
                {
                    StartupTimeoutSeconds = 5,
                    ShutdownTimeoutSeconds = 5,
                    HeartbeatIntervalSeconds = 30,
                    HeartbeatGraceSeconds = 30,
                    MaxMessageBytes = WorkerFrameProtocolOptions.DefaultMaxMessageBytes,
                },
                Sessions = new SessionOptions
                {
                    DefaultCommandTimeoutSeconds = 5,
                    MaxSessions = 4,

                    // Single-subscriber mode: a fully-detached subscriber (stream task awaited)
                    // frees the sole slot, so the reconnect attaches cleanly without needing
                    // multi-subscriber fan-out. Detach-grace keeps the session Ready across the
                    // detach and the fixture runs no lease-reaper, so the session survives to be
                    // reconnected.
                    AllowMultipleEventSubscribers = false,
                    MaxEventSubscribersPerSession = 8,
                },
                Events = new EventOptions
                {
                    QueueCapacity = 32,
                    ReplayBufferCapacity = replayBufferCapacity,

                    // Defaults to a window long enough that age-eviction never fires during a
                    // fast test, so capacity is the only eviction axis; the age-eviction test
                    // shortens it and drives a fake clock.
                    ReplayRetentionSeconds = replayRetentionSeconds,
                },
            };
    }

    // ---- fake worker launcher ----

    /// <summary>
    ///     Fake worker that emits events one at a time, gated by <see cref="AllowNextEvent"/>, so
    ///     the test drives event timing deterministically. Modeled on the multi-subscriber E2E
    ///     tests' gated launcher. The worker loop is independent of subscribers, so events emitted
    ///     while no gRPC stream is attached still flow through the distributor pump into the
    ///     session's replay ring — exactly the condition the reconnect/replay tests exercise. Call
    ///     <see cref="StopEmitting"/> before closing the session so the loop exits cleanly and can
    ///     process the shutdown envelope.
    /// </summary>
    private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher
    {
        public const int ProcessId = 7730;

        private readonly FakeWorkerProcess _process = new(ProcessId);

        // Capacity 64 so AllowNextEvent can be called ahead of time without blocking.
        private readonly SemaphoreSlim _emitGate = new(0, 64);
        private volatile bool _stopEmitting;

        /// <summary>Gets the task representing the fake worker's running background loop.</summary>
        public Task WorkerTask { get; private set; } = Task.CompletedTask;

        /// <summary>Releases the gate so the worker emits one event.</summary>
        public void AllowNextEvent() => _emitGate.Release();

        /// <summary>
        ///     Signals the worker to stop waiting for the emit gate and process the shutdown
        ///     envelope. Must be called before <c>CloseSession</c>.
        /// </summary>
        public void StopEmitting()
        {
            _stopEmitting = true;
            _emitGate.Release(); // unblock a pending gate wait if any
        }

        /// <inheritdoc />
        public Task<WorkerProcessHandle> LaunchAsync(
            WorkerProcessLaunchRequest request,
            CancellationToken cancellationToken = default)
        {
            WorkerTask = RunWorkerAsync(request, cancellationToken);

            return Task.FromResult(new WorkerProcessHandle(
                _process,
                new WorkerProcessCommandLine("reconnect-replay-fake-worker.exe", []),
                DateTimeOffset.UtcNow));
        }

        private async Task RunWorkerAsync(
            WorkerProcessLaunchRequest request,
            CancellationToken cancellationToken)
        {
            await using FakeWorkerHarness harness = await FakeWorkerHarness.ConnectToGatewayPipeAsync(
                request.SessionId,
                request.Nonce,
                request.PipeName,
                request.ProtocolVersion,
                cancellationToken: cancellationToken).ConfigureAwait(false);
            await harness.CompleteStartupAsync(ProcessId, cancellationToken: cancellationToken).ConfigureAwait(false);

            int advisedServerHandle = 0;
            int advisedItemHandle = 0;
            int emittedCount = 0;

            while (!cancellationToken.IsCancellationRequested)
            {
                // While subscribed and not stopped, emit gated events using a non-blocking peek
                // at the gate so incoming envelopes (including shutdown) are never starved.
                while (advisedServerHandle != 0
                    && !_stopEmitting
                    && await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false))
                {
                    int index = ++emittedCount;
                    await harness.EmitEventAsync(
                        MxEventFamily.OnDataChange,
                        cancellationToken,
                        mxEvent =>
                        {
                            mxEvent.ServerHandle = advisedServerHandle;
                            mxEvent.ItemHandle = advisedItemHandle;
                            mxEvent.Quality = 192;
                            mxEvent.Value = new MxValue
                            {
                                DataType = MxDataType.String,
                                StringValue = $"reconnect-value-{index}",
                            };
                            mxEvent.OnDataChange = new OnDataChangeEvent();
                        }).ConfigureAwait(false);
                }

                WorkerEnvelope? envelope;
                try
                {
                    using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
                    readCts.CancelAfter(TimeSpan.FromMilliseconds(50));
                    envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false);
                }
                catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
                {
                    // Timed out waiting for an envelope — loop back to check the gate / emit.
                    continue;
                }

                if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
                {
                    await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
                    _process.MarkExited(0);
                    return;
                }

                if (envelope.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerCommand)
                {
                    throw new InvalidOperationException($"Unexpected envelope {envelope.BodyCase}.");
                }

                MxCommand command = envelope.WorkerCommand.Command;
                await harness.ReplyToCommandAsync(
                    envelope,
                    configureReply: reply => ConfigureCommandReply(reply, command.Kind),
                    cancellationToken: cancellationToken).ConfigureAwait(false);

                if (command.Kind == MxCommandKind.Advise)
                {
                    advisedServerHandle = command.Advise.ServerHandle;
                    advisedItemHandle = command.Advise.ItemHandle;
                }
            }
        }
    }
}