'''
Real service-tree coverage for the offline replay provider.

'''
from contextlib import AbstractAsyncContextManager
from multiprocessing.shared_memory import SharedMemory
from pathlib import Path
from typing import (
    Any,
    Callable,
)

import msgspec
import numpy as np
import pytest
import tractor
import trio

from piker.brokers import replay
from piker.data import open_feed
from piker.service import Services


FQME: str = 'btcusd.test.replay'


async def _request(
    stream: tractor.MsgStream,
    command: replay.ReplayCommand,
) -> replay.ReplayAck:
    '''
    Send one replay command and receive its correlated reply.

    '''
    await stream.send(command)
    ack: replay.ReplayAck = await stream.receive()
    assert ack.command_id == command.command_id
    return ack


async def _search_replay_symbols(
    portal: tractor.Portal,
) -> dict[str, Any]:
    '''
    Exercise the backend's normal symbol-search context.

    '''
    ctx: tractor.Context
    async with portal.open_context(
        replay.open_symbol_search,
    ) as (ctx, _):
        async with ctx.open_stream() as stream:
            await stream.send('btc')
            return await stream.receive()


async def _run_basic_generation(
    open_test_pikerd: Callable[
        ...,
        AbstractAsyncContextManager,
    ],
    loglevel: str,
) -> tuple[bytes, list[dict[str, Any]], list[float]]:
    '''
    Run one complete replay through real data services.

    '''
    quote_transcript: list[dict[str, Any]] = []
    shm_names: list[str] = []
    services: Services
    async with (
        open_test_pikerd() as (_, _, _, services),
        open_feed(
            [FQME],
            loglevel=loglevel,
        ) as feed,
    ):
        assert 'datad.replay' in services.service_tasks
        assert 'samplerd' in services.service_tasks

        stream: tractor.MsgStream = feed.streams['replay']
        first_msg: dict[str, Any] = await stream.receive()
        quote_transcript.append(first_msg)
        assert first_msg[FQME]['replay_seq'] == 1

        flume = feed.flumes[FQME]
        history_closes: list[float] = (
            flume.hist_shm.array['close'].tolist()
        )
        shm_names.extend([
            flume.hist_shm._shm.name,
            flume.rt_shm._shm.name,
        ])
        assert history_closes == [
            99.2,
            99.4,
            99.6,
            99.8,
            100.2,
            100.6,
        ]

        portal: tractor.Portal = feed.portals[replay]
        matches: dict[str, Any] = await _search_replay_symbols(
            portal
        )
        assert list(matches) == [FQME]

        control_ctx: tractor.Context
        started: replay.ReplaySnapshot
        async with portal.open_context(
            replay.open_replay_control,
        ) as (control_ctx, started):
            assert started.state == 'ready'
            assert started.event_sequence == 1

            async with control_ctx.open_stream() as control:
                with control_ctx.pld_rx.limit_plds(
                    spec=replay.ReplayAck,
                ):
                    await feed.pause()
                    paused: replay.ReplayAck = await _request(
                        control,
                        replay.AwaitState(
                            command_id=1,
                            state='paused',
                        ),
                    )
                    assert paused.ok
                    assert not paused.snapshot.subscriber_active

                    rejected: replay.ReplayAck = await _request(
                        control,
                        replay.Advance(
                            command_id=2,
                            event_sequence=2,
                        ),
                    )
                    assert not rejected.ok
                    assert rejected.snapshot.event_sequence == 1

                    await feed.resume()
                    ready: replay.ReplayAck = await _request(
                        control,
                        replay.AwaitState(
                            command_id=3,
                            state='ready',
                        ),
                    )
                    assert ready.ok
                    assert ready.snapshot.subscriber_active

                    producer_paused = await _request(
                        control,
                        replay.Pause(command_id=4),
                    )
                    assert producer_paused.snapshot.state == 'paused'
                    gate_rejected = await _request(
                        control,
                        replay.Advance(
                            command_id=5,
                            event_sequence=2,
                        ),
                    )
                    assert not gate_rejected.ok
                    producer_ready = await _request(
                        control,
                        replay.Resume(command_id=6),
                    )
                    assert producer_ready.snapshot.state == 'ready'

                    for command_id, event_sequence in (
                        (7, 2),
                        (8, 3),
                        (9, 4),
                    ):
                        advanced: replay.ReplayAck = (
                            await _request(
                                control,
                                replay.Advance(
                                    command_id=command_id,
                                    event_sequence=(
                                        event_sequence
                                    ),
                                ),
                            )
                        )
                        assert advanced.ok
                        quote: dict[str, Any] = (
                            await stream.receive()
                        )
                        quote_transcript.append(quote)
                        assert (
                            quote[FQME]['replay_seq']
                            == event_sequence
                        )

                    snap_ack: replay.ReplayAck = await _request(
                        control,
                        replay.Snapshot(command_id=10),
                    )
                    assert snap_ack.snapshot.state == 'exhausted'
                    transcript: bytes = msgspec.json.encode(
                        snap_ack.snapshot
                    )

        np.testing.assert_allclose(
            flume.rt_shm.array['close'][-1],
            101.0,
        )

    assert not services.service_tasks
    for shm_name in shm_names:
        with pytest.raises(FileNotFoundError):
            SharedMemory(name=shm_name)

    return transcript, quote_transcript, history_closes


@pytest.mark.parametrize(
    'replay_scenario',
    ['basic-v1'],
    indirect=True,
)
def test_real_feed_replay_is_repeatable(
    replay_scenario: Path,
    open_test_pikerd: Callable[
        ...,
        AbstractAsyncContextManager,
    ],
    loglevel: str,
) -> None:
    '''
    Replay identical typed transcripts through fresh actor trees.

    Mocked feed calls can hide backend discovery, datad and samplerd
    startup, history publication, SHM writes, stream subscription
    control, and context teardown failures. Run the same tracked
    scenario through two fresh `open_test_pikerd()` generations.
    Each generation receives the initial quote, pauses the production
    feed, awaits the typed subscriber-state barrier, proves
    advancement is rejected, resumes, repeats that gate check
    through typed producer controls, and explicitly advances every
    remaining event. Exact control bytes, quote dictionaries, and
    history values must match, while service and named-SHM checks
    prove each generation fully tears down before the next one
    starts.

    '''
    assert replay_scenario.name == 'basic-v1.json'

    async def main() -> None:
        first = await _run_basic_generation(
            open_test_pikerd,
            loglevel,
        )
        second = await _run_basic_generation(
            open_test_pikerd,
            loglevel,
        )
        assert first == second

    trio.run(main)


@pytest.mark.parametrize(
    'replay_scenario',
    ['failure-v1'],
    indirect=True,
)
def test_failure_injection_is_acknowledged(
    replay_scenario: Path,
    open_test_pikerd: Callable[
        ...,
        AbstractAsyncContextManager,
    ],
    loglevel: str,
) -> None:
    '''
    Stop at a fixture failure without publishing its quote.

    An injected provider disconnect used to require timing a task
    crash and inferring its position from logs. Start the real feed,
    arm the fixture's third event through typed control, publish
    event two, then request event three. Its correlated negative
    acknowledgement must identify the fixture error, retain event
    sequence two, and expose `failed` state. Reading SHM after the
    acknowledgement proves the rejected event was not hidden in the
    sampler before normal actor and SHM teardown.

    '''
    assert replay_scenario.name == 'failure-v1.json'

    async def main() -> None:
        services: Services
        shm_names: list[str] = []
        async with (
            open_test_pikerd() as (_, _, _, services),
            open_feed(
                [FQME],
                loglevel=loglevel,
            ) as feed,
        ):
            quotes: tractor.MsgStream = feed.streams['replay']
            first: dict[str, Any] = await quotes.receive()
            assert first[FQME]['replay_seq'] == 1

            flume = feed.flumes[FQME]
            shm_names.extend([
                flume.hist_shm._shm.name,
                flume.rt_shm._shm.name,
            ])
            portal: tractor.Portal = feed.portals[replay]
            ctx: tractor.Context
            async with portal.open_context(
                replay.open_replay_control,
            ) as (ctx, _):
                async with ctx.open_stream() as control:
                    with ctx.pld_rx.limit_plds(
                        spec=replay.ReplayAck,
                    ):
                        armed: replay.ReplayAck = await _request(
                            control,
                            replay.FailAt(
                                command_id=1,
                                event_sequence=3,
                            ),
                        )
                        assert armed.ok

                        advanced: replay.ReplayAck = (
                            await _request(
                                control,
                                replay.Advance(
                                    command_id=2,
                                    event_sequence=2,
                                ),
                            )
                        )
                        assert advanced.ok
                        second: dict[str, Any] = (
                            await quotes.receive()
                        )
                        assert second[FQME]['replay_seq'] == 2

                        failed: replay.ReplayAck = await _request(
                            control,
                            replay.Advance(
                                command_id=3,
                                event_sequence=3,
                            ),
                        )
                        assert not failed.ok
                        assert failed.snapshot.state == 'failed'
                        assert failed.snapshot.event_sequence == 2
                        assert failed.error == (
                            'fixture_disconnect: offline provider '
                            'disconnected'
                        )
                        assert (
                            flume.rt_shm.array['close'][-1]
                            == 100.5
                        )

        assert not services.service_tasks
        for shm_name in shm_names:
            with pytest.raises(FileNotFoundError):
                SharedMemory(name=shm_name)

    trio.run(main)