"""Tests for the typed ReplayGap signal on Session.stream_events (CLI-15)."""

from __future__ import annotations

from collections.abc import AsyncIterator

import pytest

from zb_mom_ww_mxgateway import ReplayGap, Session
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb


class _FakeClient:
    """Minimal client stub exposing only what Session.stream_events needs."""

    def __init__(self, events: list[pb.MxEvent]) -> None:
        self._events = events
        self.last_request: pb.StreamEventsRequest | None = None

    def stream_events_raw(
        self,
        request: pb.StreamEventsRequest,
    ) -> AsyncIterator[pb.MxEvent]:
        self.last_request = request

        async def _gen() -> AsyncIterator[pb.MxEvent]:
            for event in self._events:
                yield event

        return _gen()


def _gap_sentinel(*, requested: int, oldest: int) -> pb.MxEvent:
    return pb.MxEvent(
        session_id="session-1",
        family=pb.MX_EVENT_FAMILY_UNSPECIFIED,
        replay_gap=pb.ReplayGap(
            requested_after_sequence=requested,
            oldest_available_sequence=oldest,
        ),
    )


def _normal_event(worker_sequence: int) -> pb.MxEvent:
    return pb.MxEvent(
        session_id="session-1",
        worker_sequence=worker_sequence,
        family=pb.MX_EVENT_FAMILY_ON_DATA_CHANGE,
    )


def _session(events: list[pb.MxEvent]) -> tuple[Session, _FakeClient]:
    client = _FakeClient(events)
    session = Session(client=client, session_id="session-1")  # type: ignore[arg-type]
    return session, client


@pytest.mark.asyncio
async def test_replay_gap_sentinel_surfaces_as_typed_signal() -> None:
    session, _client = _session(
        [_gap_sentinel(requested=5, oldest=10), _normal_event(10)],
    )

    items = [item async for item in session.stream_events(after_worker_sequence=5)]

    assert isinstance(items[0], ReplayGap)
    assert items[0].requested_after_sequence == 5
    assert items[0].oldest_available_sequence == 10
    # Resume cursor is oldest_available_sequence - 1 so the next replayed event
    # is the oldest the gateway still retains.
    assert items[0].resume_after_worker_sequence == 9

    # Normal events after the sentinel pass through unchanged as MxEvent.
    assert isinstance(items[1], pb.MxEvent)
    assert not items[1].HasField("replay_gap")
    assert items[1].worker_sequence == 10


@pytest.mark.asyncio
async def test_normal_events_are_unaffected() -> None:
    session, _client = _session([_normal_event(1), _normal_event(2)])

    items = [item async for item in session.stream_events()]

    assert all(isinstance(item, pb.MxEvent) for item in items)
    assert [item.worker_sequence for item in items] == [1, 2]
    assert not any(isinstance(item, ReplayGap) for item in items)


@pytest.mark.asyncio
async def test_stream_events_forwards_resume_cursor() -> None:
    session, client = _session([])

    async for _ in session.stream_events(after_worker_sequence=42):
        pass

    assert client.last_request is not None
    assert client.last_request.after_worker_sequence == 42
    assert client.last_request.session_id == "session-1"


def test_replay_gap_resume_cursor_never_negative() -> None:
    gap = ReplayGap.from_proto(
        pb.ReplayGap(requested_after_sequence=0, oldest_available_sequence=0),
    )
    assert gap.resume_after_worker_sequence == 0