"""bybit_concurrency — the bounded-concurrency fetch helper + the rate-limit backoff wrapper.

No network, no real sleeps: `fetch_one` is a fake that records concurrency, and the backoff sleeper is a
no-op that records the waited durations. These tests pin the contract the ingest rewires depend on:
  * fetch_symbols_concurrently: every symbol fetched + written EXACTLY once, regardless of completion order;
    in-flight fetches are bounded by `workers`; writes happen AS-COMPLETED (not collected-then-written),
    serialized through the main thread.
  * with_rate_limit_backoff: a rate-limited fetch (Bybit retCode OR an HTTP-429 exception) is retried with
    exponential backoff up to `tries`, then surfaces; a success short-circuits.
"""
from __future__ import annotations

import threading
import time

import pytest

from fxhnt.application.bybit_concurrency import (
    RateLimitError,
    fetch_symbols_concurrently,
    with_rate_limit_backoff,
)


# -- fetch_symbols_concurrently --------------------------------------------------------------------
def test_all_symbols_fetched_and_written_exactly_once():
    symbols = [f"S{i}USDT" for i in range(50)]

    def fetch_one(sym: str) -> str:
        return f"data:{sym}"

    written: dict[str, str] = {}
    fetch_symbols_concurrently(
        symbols, fetch_one, workers=8, on_result=lambda s, r: written.__setitem__(s, r))

    assert set(written) == set(symbols)               # every symbol written
    assert len(written) == len(symbols)               # exactly once (dict would collapse dups anyway)
    assert all(written[s] == f"data:{s}" for s in symbols)  # result routed to the right symbol


def test_results_written_for_all_even_with_out_of_order_completion():
    # Reverse-skew the fetch latency so completion order != submission order, then assert nothing is dropped.
    symbols = [f"S{i}USDT" for i in range(20)]

    def fetch_one(sym: str) -> int:
        idx = int(sym[1:-4])
        time.sleep(0.001 * (len(symbols) - idx))   # earlier symbols finish LAST
        return idx

    written: list[str] = []
    lock = threading.Lock()

    def on_result(sym: str, _r: int) -> None:
        with lock:
            written.append(sym)

    fetch_symbols_concurrently(symbols, fetch_one, workers=8, on_result=on_result)
    assert sorted(written) == sorted(symbols)        # all present, none dropped/duplicated


def test_in_flight_fetches_bounded_by_workers():
    workers = 4
    symbols = [f"S{i}USDT" for i in range(40)]
    inflight = 0
    peak = 0
    lock = threading.Lock()

    def fetch_one(sym: str) -> str:
        nonlocal inflight, peak
        with lock:
            inflight += 1
            peak = max(peak, inflight)
        time.sleep(0.005)                            # hold the slot so concurrency can build up
        with lock:
            inflight -= 1
        return sym

    fetch_symbols_concurrently(symbols, fetch_one, workers=workers, on_result=lambda s, r: None)
    assert peak <= workers, f"peak in-flight {peak} exceeded workers {workers}"
    assert peak > 1, "expected actual concurrency (peak should exceed 1)"


def test_writes_happen_as_completed_not_collected_first():
    # If results were collected-then-written, no on_result would fire until ALL fetches finished. We prove
    # as-completed by blocking the LAST-submitted fetch until an EARLY on_result has already fired.
    symbols = [f"S{i}USDT" for i in range(5)]
    early_written = threading.Event()
    release_last = threading.Event()

    def fetch_one(sym: str) -> str:
        if sym == "S4USDT":
            # the last symbol cannot complete until an earlier write already happened
            assert release_last.wait(timeout=5), "as-completed write never fired before last fetch"
        return sym

    def on_result(sym: str, _r: str) -> None:
        if sym != "S4USDT":
            early_written.set()
            release_last.set()

    fetch_symbols_concurrently(symbols, fetch_one, workers=3, on_result=on_result)
    assert early_written.is_set()


def test_on_result_called_in_single_thread_serialized():
    # Writes funnel through the main thread, so on_result must never overlap (no lock needed by the caller).
    symbols = [f"S{i}USDT" for i in range(30)]
    concurrent_writes = 0
    peak_writes = 0
    lock = threading.Lock()

    def on_result(_sym: str, _r: str) -> None:
        nonlocal concurrent_writes, peak_writes
        with lock:
            concurrent_writes += 1
            peak_writes = max(peak_writes, concurrent_writes)
        time.sleep(0.001)
        with lock:
            concurrent_writes -= 1

    fetch_symbols_concurrently(symbols, lambda s: s, workers=8, on_result=on_result)
    assert peak_writes == 1, f"on_result overlapped ({peak_writes}) — writes are not serialized"


def test_no_on_result_is_allowed():
    # on_result is optional; helper still drains all fetches (exceptions would surface).
    seen: list[str] = []
    fetch_symbols_concurrently(["AUSDT", "BUSDT"], lambda s: seen.append(s) or s, workers=2)
    assert sorted(seen) == ["AUSDT", "BUSDT"]


def test_fetch_exception_propagates():
    def fetch_one(sym: str) -> str:
        if sym == "BADUSDT":
            raise ValueError("boom")
        return sym

    with pytest.raises(ValueError, match="boom"):
        fetch_symbols_concurrently(
            ["AUSDT", "BADUSDT", "CUSDT"], fetch_one, workers=3, on_result=lambda s, r: None)


# -- with_rate_limit_backoff -----------------------------------------------------------------------
def _sleep_recorder():
    waited: list[float] = []
    return waited, waited.append


def test_backoff_retries_on_bybit_retcode_then_succeeds():
    # Bybit signals rate limiting via retCode (10006 / 10018). Two rate-limited responses, then success.
    waited, sleeper = _sleep_recorder()
    calls = {"n": 0}

    def fetch(url: str) -> dict:
        calls["n"] += 1
        if calls["n"] <= 2:
            return {"retCode": 10006, "retMsg": "Too many visits", "result": {}}
        return {"retCode": 0, "result": {"list": [1, 2, 3]}}

    wrapped = with_rate_limit_backoff(fetch, sleep=sleeper, base=0.5, tries=5)
    out = wrapped("https://api.bybit.com/x")
    assert out == {"retCode": 0, "result": {"list": [1, 2, 3]}}
    assert calls["n"] == 3                          # 2 rate-limited + 1 success
    assert waited == [0.5, 1.0]                     # exponential backoff between the 2 retries (no real sleep)


def test_backoff_retries_on_http_429_exception_then_succeeds():
    # An HTTP 429 raised by the fetch (e.g. urllib HTTPError code 429) is treated as rate limiting too.
    waited, sleeper = _sleep_recorder()
    calls = {"n": 0}

    class _HTTP429(Exception):
        code = 429

    def fetch(url: str) -> dict:
        calls["n"] += 1
        if calls["n"] == 1:
            raise _HTTP429("429 Too Many Requests")
        return {"retCode": 0, "result": {"list": []}}

    wrapped = with_rate_limit_backoff(fetch, sleep=sleeper, base=0.5, tries=5)
    out = wrapped("https://api.bybit.com/x")
    assert out == {"retCode": 0, "result": {"list": []}}
    assert calls["n"] == 2
    assert waited == [0.5]


def test_backoff_exhausts_tries_and_raises_ratelimiterror():
    waited, sleeper = _sleep_recorder()

    def fetch(url: str) -> dict:
        return {"retCode": 10018, "retMsg": "exceeded ip rate limit"}

    wrapped = with_rate_limit_backoff(fetch, sleep=sleeper, base=0.5, tries=4)
    with pytest.raises(RateLimitError):
        wrapped("https://api.bybit.com/x")
    assert len(waited) == 3                          # tries-1 backoffs before giving up (no real sleep)
    assert waited == [0.5, 1.0, 2.0]                 # exponential: base * 2**i


def test_backoff_does_not_retry_non_ratelimit_errors():
    # A non-rate-limit error (e.g. a real bug or a different retCode) must surface immediately, no retry.
    waited, sleeper = _sleep_recorder()

    def fetch(url: str) -> dict:
        raise ValueError("genuine bug")

    wrapped = with_rate_limit_backoff(fetch, sleep=sleeper, base=0.5, tries=5)
    with pytest.raises(ValueError, match="genuine bug"):
        wrapped("https://api.bybit.com/x")
    assert waited == []                              # no backoff for non-rate-limit failures


def test_backoff_passes_through_normal_success_no_sleep():
    waited, sleeper = _sleep_recorder()
    wrapped = with_rate_limit_backoff(
        lambda url: {"retCode": 0, "result": {"list": []}}, sleep=sleeper, base=0.5, tries=5)
    assert wrapped("u") == {"retCode": 0, "result": {"list": []}}
    assert waited == []
