# retoor <retoor@molodetz.nl>

from __future__ import annotations

import asyncio
import logging
import time
from typing import Awaitable, Callable, Optional

import httpx

logger = logging.getLogger(__name__)


def percentile(sorted_values: list[float], q: float) -> float:
    if not sorted_values:
        return 0.0
    if len(sorted_values) == 1:
        return float(sorted_values[0])
    rank = (len(sorted_values) - 1) * q
    low = int(rank)
    high = min(low + 1, len(sorted_values) - 1)
    frac = rank - low
    return float(sorted_values[low] + (sorted_values[high] - sorted_values[low]) * frac)


class CircuitBreaker:
    def __init__(self, threshold: int, cooldown_seconds: int):
        self.threshold = threshold
        self.cooldown_seconds = cooldown_seconds
        self.failures = 0
        self.opened_at: Optional[float] = None

    def configure(self, threshold: int, cooldown_seconds: int) -> None:
        self.threshold = threshold
        self.cooldown_seconds = cooldown_seconds

    @property
    def is_open(self) -> bool:
        return self.opened_at is not None

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if (time.monotonic() - self.opened_at) >= self.cooldown_seconds:
            self.opened_at = None
            self.failures = 0
            return True
        return False

    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None

    def record_failure(self) -> None:
        self.failures += 1
        if self.threshold > 0 and self.failures >= self.threshold:
            self.opened_at = time.monotonic()


async def _backoff(backoff_ms: int, attempt: int) -> None:
    delay = max(0, backoff_ms) * attempt / 1000.0
    if delay > 0:
        await asyncio.sleep(delay)


async def retry_send(
    do_call: Callable[[], Awaitable[httpx.Response]],
    sem: asyncio.Semaphore,
    max_retries: int,
    backoff_ms: int,
    log: Optional[Callable[[str], None]] = None,
) -> tuple[Optional[httpx.Response], Optional[Exception], int, float]:
    log = log or (lambda message: None)
    attempts = 0
    last_exc: Optional[Exception] = None
    queue_wait_ms = 0.0
    while attempts <= max_retries:
        attempts += 1
        wait_start = time.monotonic()
        async with sem:
            queue_wait_ms += (time.monotonic() - wait_start) * 1000
            try:
                resp = await do_call()
                exc = None
            except httpx.RequestError as e:
                resp = None
                exc = e
        if exc is not None:
            last_exc = exc
            if attempts > max_retries:
                return None, exc, attempts, queue_wait_ms
            log(
                f"upstream connection failed, retrying ({attempts}/{max_retries}): {exc}"
            )
            await _backoff(backoff_ms, attempts)
            continue
        if resp.status_code >= 500 and attempts <= max_retries:
            log(f"upstream {resp.status_code}, retrying ({attempts}/{max_retries})")
            try:
                await resp.aclose()
            except Exception:  # noqa: BLE001 - releasing the connection must never block a retry
                pass
            await _backoff(backoff_ms, attempts)
            continue
        return resp, None, attempts, queue_wait_ms
    return None, last_exc, attempts, queue_wait_ms
