"""Tierguard client: emit per-tenant API-usage events to a Tierguard ingest endpoint.

Xero and QuickBooks Online only report usage for your app as a whole. Embed this
client at the boundary where your code calls those APIs, where you know which of
your tenants triggered each call, and record one event per call.

Install: download ``tierguard_client.py`` from https://tierguard.dev/docs/api and
put it next to your code. It needs Python 3.9+ and ``httpx``, nothing else.

Create ONE client per process, at module level, and reuse it. Do not build a
client per call or per request:

    from tierguard_client import TierguardClient

    tg = TierguardClient("https://tierguard.dev", "YOUR_INGEST_TOKEN")

    resp = xero.get(f"/connections/{org_id}/Invoices")
    tg.record(tenant_ref=org_id, endpoint="/Invoices",
              bytes_egress=len(resp.content), call_count=1)

``record()`` never performs network I/O, never blocks and never raises. Events
are buffered in memory and sent in batches by a background thread every
``flush_interval`` seconds, or as soon as ``batch_size`` events are waiting.
When Tierguard is unreachable the client backs off exponentially and keeps up
to ``max_buffer`` events, dropping the oldest beyond that. Pass a stable
``event_id`` per logical call so retries are deduplicated server-side.

Buffered events are flushed when the interpreter exits and on ``close()``.
In a serverless function the process may be frozen as soon as the handler
returns, so call ``tg.flush()`` before returning.
"""

from __future__ import annotations

import atexit
import collections
import datetime
import functools
import logging
import os
import random
import threading
import time
import uuid
import weakref
from typing import Any, Optional

import httpx

__version__ = "0.1.0"

__all__ = ["TierguardClient", "__version__"]

logger = logging.getLogger("tierguard_client")

_SERVER_MAX_BATCH = 1000
_DEFAULT_MAX_BUFFER = 10_000
_MAX_STR_LEN = 256
_MAX_NUMERIC = 10 ** 15
_RETRYABLE_STATUS = frozenset({408, 429})


def _sender_loop(
    ref: "weakref.ReferenceType[TierguardClient]",
    wake: threading.Event,
    stop: threading.Event,
    interval: float,
) -> None:
    while not stop.is_set():
        wake.wait(interval)
        wake.clear()
        if stop.is_set():
            break
        client = ref()
        if client is None:
            break
        try:
            client._drain_if_due()
        except Exception:
            logger.debug("tierguard sender iteration failed", exc_info=True)
        del client


def _close_at_exit(ref: "weakref.ReferenceType[TierguardClient]") -> None:
    client = ref()
    if client is not None:
        client.close()


class TierguardClient:
    """Buffering, non-blocking client for ``POST /v1/ingest``. See the module docstring."""

    def __init__(
        self,
        base_url: str,
        token: str,
        *,
        batch_size: int = 100,
        timeout: float = 5.0,
        max_buffer: int = _DEFAULT_MAX_BUFFER,
        transport: Optional[httpx.BaseTransport] = None,
        flush_interval: float = 5.0,
        backoff_base: float = 1.0,
        backoff_max: float = 300.0,
        shutdown_timeout: float = 5.0,
    ) -> None:
        self._url = base_url.rstrip("/") + "/v1/ingest"
        self._token = token
        self._batch_size = max(1, min(batch_size, _SERVER_MAX_BATCH))
        self._max_buffer = max(self._batch_size, max_buffer)
        self._timeout = timeout
        self._transport = transport
        self._flush_interval = flush_interval
        self._backoff_base = backoff_base
        self._backoff_max = backoff_max
        self._shutdown_timeout = shutdown_timeout
        self._http = httpx.Client(timeout=timeout, transport=transport)
        self._closed = False
        self._pid = os.getpid()
        self._init_state()

    def _init_state(self) -> None:
        self._lock = threading.Lock()
        self._send_lock = threading.Lock()
        self._wake = threading.Event()
        self._stop = threading.Event()
        self._buffer: "collections.deque[dict[str, Any]]" = collections.deque()
        self._thread: Optional[threading.Thread] = None
        self._exit_hook: Optional[Any] = None
        self._failures = 0
        self._retry_at = 0.0
        self._overflow_warned = False
        self._dropped = 0

    def _check_fork(self) -> None:
        if os.getpid() == self._pid:
            return
        self._pid = os.getpid()
        self._init_state()
        self._http = httpx.Client(timeout=self._timeout, transport=self._transport)

    def record(
        self,
        tenant_ref: str,
        endpoint: str,
        *,
        bytes_egress: Optional[int] = None,
        call_count: Optional[int] = None,
        occurred_at: Optional[datetime.datetime] = None,
        event_id: Optional[str] = None,
    ) -> str:
        resolved_id = event_id if event_id else uuid.uuid4().hex
        try:
            if self._closed:
                logger.debug("tierguard client is closed; event %s not recorded", resolved_id)
                return resolved_id
            self._check_fork()
            event = self._build_event(
                tenant_ref, endpoint, bytes_egress, call_count, occurred_at, resolved_id
            )
        except Exception as exc:
            logger.warning("tierguard dropped an invalid event: %s", exc)
            try:
                with self._lock:
                    self._dropped += 1
            except Exception:
                pass
            return resolved_id
        try:
            self._enqueue(event)
        except Exception:
            logger.warning("tierguard could not buffer event", exc_info=True)
        return resolved_id

    @staticmethod
    def _metric(value: Any, name: str) -> int:
        number = int(value)
        if not 0 <= number <= _MAX_NUMERIC:
            raise ValueError(f"{name} out of range")
        return number

    def _build_event(
        self,
        tenant_ref: Any,
        endpoint: Any,
        bytes_egress: Any,
        call_count: Any,
        occurred_at: Optional[datetime.datetime],
        event_id: Any,
    ) -> dict[str, Any]:
        if bytes_egress is None and call_count is None:
            call_count = 1
        when = occurred_at or datetime.datetime.now(datetime.timezone.utc)
        if not isinstance(event_id, str) or len(event_id) > _MAX_STR_LEN:
            raise ValueError("event_id must be a string of at most 256 characters")
        event: dict[str, Any] = {
            "tenant_ref": str(tenant_ref)[:_MAX_STR_LEN],
            "endpoint": str(endpoint)[:_MAX_STR_LEN],
            "occurred_at": when.isoformat(),
            "event_id": event_id,
        }
        if bytes_egress is not None:
            event["bytes_egress"] = self._metric(bytes_egress, "bytes_egress")
        if call_count is not None:
            event["call_count"] = self._metric(call_count, "call_count")
        return event

    def _enqueue(self, event: dict[str, Any]) -> None:
        with self._lock:
            self._buffer.append(event)
            warn = self._trim_locked()
            ready = len(self._buffer) >= self._batch_size and time.monotonic() >= self._retry_at
        if warn:
            logger.warning("tierguard buffer full; dropping oldest events")
        self._ensure_thread()
        if ready:
            self._wake.set()

    def _trim_locked(self) -> bool:
        overflow = len(self._buffer) - self._max_buffer
        if overflow <= 0:
            return False
        for _ in range(overflow):
            self._buffer.popleft()
        self._dropped += overflow
        first = not self._overflow_warned
        self._overflow_warned = True
        return first

    def _ensure_thread(self) -> None:
        thread = self._thread
        if thread is not None and thread.is_alive():
            return
        with self._lock:
            if self._closed or (self._thread is not None and self._thread.is_alive()):
                return
            thread = threading.Thread(
                target=_sender_loop,
                args=(weakref.ref(self), self._wake, self._stop, self._flush_interval),
                name="tierguard-client",
                daemon=True,
            )
            self._thread = thread
            thread.start()
            if self._exit_hook is None:
                self._exit_hook = functools.partial(_close_at_exit, weakref.ref(self))
                atexit.register(self._exit_hook)

    def record_response(self, tenant_ref: str, response: httpx.Response) -> str:
        try:
            endpoint = response.request.url.path
        except RuntimeError:
            endpoint = ""
        try:
            size: Optional[int] = len(response.content)
        except Exception:
            size = None
        return self.record(tenant_ref, endpoint, bytes_egress=size, call_count=1)

    def _drain_if_due(self) -> None:
        while True:
            if not self._buffer or self._stop.is_set():
                return
            if time.monotonic() < self._retry_at:
                return
            with self._send_lock:
                if not self._buffer or time.monotonic() < self._retry_at:
                    return
                outcome = self._attempt(None)
                if outcome is None:
                    return
            if len(self._buffer) < self._batch_size:
                return

    def _attempt(self, deadline: Optional[float]) -> Optional[int]:
        with self._lock:
            count = min(len(self._buffer), _SERVER_MAX_BATCH)
            batch = [self._buffer.popleft() for _ in range(count)]
        if not batch:
            return 0
        timeout = self._timeout
        if deadline is not None:
            timeout = max(0.01, min(self._timeout, deadline - time.monotonic()))
        try:
            resp = self._http.post(
                self._url,
                json=batch,
                headers={"Authorization": f"Bearer {self._token}"},
                timeout=timeout,
            )
        except Exception:
            logger.warning(
                "tierguard flush failed; re-buffering %d event(s)", len(batch), exc_info=True
            )
            self._requeue(batch)
            return None
        status = resp.status_code
        if 200 <= status < 300:
            try:
                accepted = int(resp.json().get("accepted", 0))
            except Exception:
                accepted = 0
            self._on_success()
            return accepted
        if status in _RETRYABLE_STATUS or status >= 500:
            logger.warning(
                "tierguard flush got HTTP %d; re-buffering %d event(s)", status, len(batch)
            )
            self._requeue(batch)
            return None
        logger.warning(
            "tierguard server rejected a batch of %d event(s) with HTTP %d; dropping it",
            len(batch),
            status,
        )
        with self._lock:
            self._dropped += len(batch)
        self._open_circuit()
        return None

    def _requeue(self, batch: list[dict[str, Any]]) -> None:
        with self._lock:
            self._buffer.extendleft(reversed(batch))
            warn = self._trim_locked()
        if warn:
            logger.warning("tierguard buffer full; dropping oldest events")
        self._open_circuit()

    def _open_circuit(self) -> None:
        self._failures += 1
        exponent = min(self._failures - 1, 60)
        delay = min(self._backoff_max, self._backoff_base * 2 ** exponent)
        self._retry_at = time.monotonic() + delay * random.uniform(0.5, 1.0)

    def _on_success(self) -> None:
        self._failures = 0
        self._retry_at = 0.0
        self._overflow_warned = False

    def _flush_until(self, deadline: Optional[float]) -> int:
        if deadline is None:
            self._send_lock.acquire()
        elif not self._send_lock.acquire(timeout=max(0.0, deadline - time.monotonic())):
            return 0
        total = 0
        try:
            while self._buffer:
                if deadline is not None and time.monotonic() >= deadline:
                    break
                outcome = self._attempt(deadline)
                if outcome is None:
                    break
                total += outcome
        finally:
            self._send_lock.release()
        return total

    def flush(self) -> int:
        try:
            self._check_fork()
            return self._flush_until(None)
        except Exception:
            logger.warning("tierguard flush failed", exc_info=True)
            return 0

    @property
    def pending(self) -> int:
        return len(self._buffer)

    @property
    def dropped(self) -> int:
        return self._dropped

    def close(self, timeout: Optional[float] = None) -> None:
        try:
            if self._closed:
                return
            self._check_fork()
            limit = self._shutdown_timeout if timeout is None else timeout
            deadline = time.monotonic() + limit
            self._closed = True
            self._stop.set()
            self._wake.set()
            if self._exit_hook is not None:
                atexit.unregister(self._exit_hook)
            thread = self._thread
            if thread is not None and thread is not threading.current_thread():
                thread.join(max(0.0, deadline - time.monotonic()))
            drainer = threading.Thread(
                target=self._final_drain, args=(deadline,), name="tierguard-close", daemon=True
            )
            drainer.start()
            drainer.join(max(0.0, deadline - time.monotonic()))
            self._http.close()
        except Exception:
            logger.debug("tierguard close failed", exc_info=True)

    def _final_drain(self, deadline: float) -> None:
        try:
            self._flush_until(deadline)
        except Exception:
            logger.debug("tierguard final drain failed", exc_info=True)

    def __enter__(self) -> "TierguardClient":
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()