"""Integration tests for /api/acks/stream and /api/ta1-acks/stream.

SP25: both endpoints follow the live-tail wire format used by
/api/claims/stream, /api/remittances/stream, /api/activity/stream:

  * Snapshot: one ``item`` line per existing row (newest first).
  * Closing: one ``snapshot_end`` line.
  * Live: forwarded ``item`` lines for each ``ack_received`` /
    ``ta1_ack_received`` event published on the bus.

These tests follow the same direct-coroutine pattern as
``test_api_stream_live.py`` because ``httpx.ASGITransport`` buffers
the response and never delivers a disconnect message — calling the
endpoint coroutine with a synthetic ``Request`` lets us iterate the
body iterator byte-by-byte and use ``body_iterator.aclose()`` to
simulate a client disconnect.
"""
from __future__ import annotations

import asyncio
import json
from datetime import date

import pytest
from starlette.requests import Request

from cyclone import db
from cyclone.api import app
from cyclone.api_routers.acks import acks_stream
from cyclone.api_routers.ta1_acks import ta1_acks_stream
from cyclone.pubsub import EventBus
from cyclone.store import store


# ---------------------------------------------------------------------------
# Shared helpers — mirror the patterns in test_api_stream_live.py
# ---------------------------------------------------------------------------


def _make_request(path: str = "/api/acks/stream") -> Request:
    """Synthetic Starlette ``Request`` with a no-op receive channel."""

    async def receive():
        # Block forever; ``is_disconnected``'s cancel scope will cancel
        # this await and we'll return None (treated as "no message").
        await asyncio.Event().wait()

    scope = {
        "type": "http",
        "method": "GET",
        "headers": [],
        "query_string": b"",
        "path": path,
        "app": app,
        "scheme": "http",
        "server": ("testserver", 80),
        "client": ("testclient", 50000),
    }
    return Request(scope, receive=receive)


def _resolve_query_defaults(endpoint) -> dict:
    """Resolve FastAPI Query(...) defaults to their inner ``.default``."""
    import inspect
    sig = inspect.signature(endpoint)
    resolved: dict = {}
    for name, param in sig.parameters.items():
        default = param.default
        if hasattr(default, "default"):
            resolved[name] = default.default
        else:
            resolved[name] = default
    return resolved


async def _call_endpoint(endpoint, path: str):
    request = _make_request(path)
    defaults = _resolve_query_defaults(endpoint)
    defaults.pop("request", None)
    return await endpoint(request, **defaults)


async def _read_lines(body_iter, n: int, timeout: float = 5.0) -> list[str]:
    buffer = b""
    lines: list[str] = []
    async with asyncio.timeout(timeout):
        while len(lines) < n:
            try:
                chunk = await body_iter.__anext__()
            except StopAsyncIteration:
                break
            buffer += chunk
            while b"\n" in buffer and len(lines) < n:
                raw, buffer = buffer.split(b"\n", 1)
                if raw:
                    lines.append(raw.decode("utf-8"))
    return lines


async def _read_one_line(body_iter, timeout: float = 5.0) -> str | None:
    buffer = b""
    async with asyncio.timeout(timeout):
        while True:
            try:
                chunk = await body_iter.__anext__()
            except StopAsyncIteration:
                return None
            buffer += chunk
            if b"\n" in buffer:
                raw, _ = buffer.split(b"\n", 1)
                if raw:
                    return raw.decode("utf-8")


async def _read_until_type(
    body_iter, target: str, *, timeout: float = 5.0, max_lines: int = 50
) -> dict | None:
    buffer = b""
    seen = 0
    async with asyncio.timeout(timeout):
        while seen < max_lines:
            try:
                chunk = await body_iter.__anext__()
            except StopAsyncIteration:
                return None
            buffer += chunk
            while b"\n" in buffer:
                raw, buffer = buffer.split(b"\n", 1)
                if not raw:
                    continue
                seen += 1
                obj = json.loads(raw.decode("utf-8"))
                if obj.get("type") == target:
                    return obj
    return None


async def _drain_until_disconnect(body_iter, max_chunks: int = 20) -> None:
    try:
        await body_iter.aclose()
    except (asyncio.CancelledError, GeneratorExit):
        pass


# ---------------------------------------------------------------------------
# Per-test heartbeat patch — 0.2s so idle generators exit promptly.
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def _short_heartbeat(monkeypatch):
    monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2")


@pytest.fixture
def bus() -> EventBus:
    """Fresh EventBus attached to app.state. Tests publish via ``bus.publish``.

    ``add_ack`` / ``add_ta1_ack`` already publish to whichever bus is
    passed in, but the live-event tests want to publish synchronously
    against ``app.state.event_bus`` so the existing-tail subscription
    picks them up.
    """
    new_bus = EventBus(max_queue_size=64)
    app.state.event_bus = new_bus
    return new_bus


# ---------------------------------------------------------------------------
# /api/acks/stream
# ---------------------------------------------------------------------------


async def test_acks_stream_snapshots_existing_rows(bus: EventBus):
    """An existing 999 row surfaces as the first ``item`` line."""
    with db.SessionLocal()() as s:
        store.add_ack(
            source_batch_id="999-SEED-1",
            accepted_count=1,
            rejected_count=0,
            received_count=1,
            ack_code="A",
            raw_json={
                "envelope": {"control_number": "000000001"},
                "set_responses": [{"set_control_number": "PCN-1"}],
            },
            event_bus=bus,
        )

    response = await _call_endpoint(acks_stream, "/api/acks/stream")
    assert response.media_type.startswith("application/x-ndjson")

    lines = await _read_lines(response.body_iterator, n=2)
    items = [json.loads(line) for line in lines]
    assert items[0]["type"] == "item"
    assert items[0]["data"]["source_batch_id"] == "999-SEED-1"
    assert items[0]["data"]["patient_control_number"] == "PCN-1"
    assert items[1] == {"type": "snapshot_end", "data": {"count": 1}}
    await _drain_until_disconnect(response.body_iterator)


async def test_acks_stream_emits_live_event(bus: EventBus):
    """A 999 written while the stream is open shows up as a live item."""
    response = await _call_endpoint(acks_stream, "/api/acks/stream")

    # Drain snapshot (1 snapshot_end line, no prior rows).
    snap = await _read_until_type(
        response.body_iterator, "snapshot_end", timeout=2.0,
    )
    assert snap == {"type": "snapshot_end", "data": {"count": 0}}

    # Advance one step so ``subscribe_raw`` registers. Next line may be
    # a heartbeat — that's fine, we just need the subscription live.
    _ = await _read_one_line(response.body_iterator, timeout=2.0)

    with db.SessionLocal()() as s:
        store.add_ack(
            source_batch_id="999-LIVE-1",
            accepted_count=1,
            rejected_count=0,
            received_count=1,
            ack_code="A",
            raw_json={
                "envelope": {"control_number": "000000002"},
                "set_responses": [],
            },
            event_bus=bus,
        )

    obj = await _read_until_type(
        response.body_iterator, "item", timeout=2.0,
    )
    assert obj is not None, "no item line arrived after publish"
    assert obj["data"]["source_batch_id"] == "999-LIVE-1"
    await _drain_until_disconnect(response.body_iterator)


# ---------------------------------------------------------------------------
# /api/ta1-acks/stream
# ---------------------------------------------------------------------------


async def test_ta1_acks_stream_snapshots_existing_rows(bus: EventBus):
    """An existing TA1 row surfaces as the first ``item`` line."""
    with db.SessionLocal()() as s:
        store.add_ta1_ack(
            source_batch_id="TA1-SEED-1",
            control_number="000000001",
            interchange_date=date(2026, 7, 2),
            interchange_time="1200",
            ack_code="A",
            note_code="000",
            ack_generated_date=None,
            sender_id="S",
            receiver_id="R",
            raw_json={"envelope": {"control_number": "000000001"}},
            event_bus=bus,
        )

    response = await _call_endpoint(ta1_acks_stream, "/api/ta1-acks/stream")
    assert response.media_type.startswith("application/x-ndjson")

    lines = await _read_lines(response.body_iterator, n=2)
    items = [json.loads(line) for line in lines]
    assert items[0]["type"] == "item"
    assert items[0]["data"]["control_number"] == "000000001"
    assert items[0]["data"]["interchange_date"] == "2026-07-02"
    assert items[1] == {"type": "snapshot_end", "data": {"count": 1}}
    await _drain_until_disconnect(response.body_iterator)


async def test_ta1_acks_stream_emits_live_event(bus: EventBus):
    """A TA1 written while the stream is open shows up as a live item."""
    response = await _call_endpoint(ta1_acks_stream, "/api/ta1-acks/stream")

    snap = await _read_until_type(
        response.body_iterator, "snapshot_end", timeout=2.0,
    )
    assert snap == {"type": "snapshot_end", "data": {"count": 0}}
    _ = await _read_one_line(response.body_iterator, timeout=2.0)

    with db.SessionLocal()() as s:
        store.add_ta1_ack(
            source_batch_id="TA1-LIVE-1",
            control_number="000000099",
            interchange_date=date(2026, 7, 2),
            interchange_time="1330",
            ack_code="A",
            note_code="000",
            ack_generated_date=None,
            sender_id="SX",
            receiver_id="RX",
            raw_json={"envelope": {"control_number": "000000099"}},
            event_bus=bus,
        )

    obj = await _read_until_type(
        response.body_iterator, "item", timeout=2.0,
    )
    assert obj is not None, "no item line arrived after publish"
    assert obj["data"]["control_number"] == "000000099"
    assert obj["data"]["sender_id"] == "SX"
    await _drain_until_disconnect(response.body_iterator)