from __future__ import annotations import asyncio import ssl from dataclasses import dataclass from datetime import UTC, datetime from enum import StrEnum from time import monotonic from typing import Any from urllib.parse import urljoin, urlsplit import httpx from opentelemetry.trace import SpanKind from app.telemetry import record_crm, record_retry, safe_span class CrmOutcome(StrEnum): SUCCEEDED = "succeeded" RETRY = "retry" UNCERTAIN = "uncertain" PERMANENT = "permanent" RATE_LIMITED = "rate_limited" @dataclass(frozen=True) class CrmResult: outcome: CrmOutcome result: Any = None error_code: str | None = None http_status: int | None = None retry_after: float | None = None class CrmClient: """Host-locked, verified-TLS Bitrix client; redirects are never followed.""" def __init__(self, credential_url: str, approved_host: str, timeout: float) -> None: parsed = urlsplit(credential_url) if parsed.scheme != "https" or parsed.hostname != approved_host: raise ValueError("credential URL is outside approved Bitrix host") self._base_url = credential_url.rstrip("/") + "/" self._host = approved_host self._client = httpx.AsyncClient( timeout=httpx.Timeout(timeout), verify=ssl.create_default_context(), follow_redirects=False, limits=httpx.Limits(max_connections=4, max_keepalive_connections=2), headers={"Accept": "application/json"}, ) async def close(self) -> None: await self._client.aclose() async def call(self, method: str, params: dict[str, Any], *, mutating: bool) -> CrmResult: started = monotonic() outcome = "error" operation = method.rsplit(".", 1)[-1] try: with safe_span( "bitrix_sync.crm.request", kind=SpanKind.CLIENT, attributes={"crm.operation": operation}, ): result = await self._call(method, params, mutating=mutating) outcome = result.outcome.value if result.outcome in { CrmOutcome.RETRY, CrmOutcome.UNCERTAIN, CrmOutcome.RATE_LIMITED, }: record_retry("crm") return result finally: record_crm(method, outcome, monotonic() - started) async def _call(self, method: str, params: dict[str, Any], *, mutating: bool) -> CrmResult: if method not in ALLOWED_METHODS: raise ValueError("unapproved CRM method") url = urljoin(self._base_url, method + ".json") if urlsplit(url).hostname != self._host: raise ValueError("CRM host changed during URL construction") try: response = await self._client.post(url, json=params) except (httpx.ConnectError, httpx.ReadError, httpx.RemoteProtocolError): return CrmResult(CrmOutcome.RETRY, error_code="crm_network") except httpx.TimeoutException: outcome = CrmOutcome.UNCERTAIN if mutating else CrmOutcome.RETRY return CrmResult(outcome, error_code="crm_timeout") if response.is_redirect: return CrmResult( CrmOutcome.PERMANENT, error_code="crm_redirect_rejected", http_status=response.status_code, ) retry_after = _retry_after(response) if response.status_code in (408, 429) or response.status_code >= 500: return CrmResult( CrmOutcome.RETRY, error_code=f"crm_http_{response.status_code}", http_status=response.status_code, retry_after=retry_after, ) try: payload = response.json() except ValueError: return CrmResult( CrmOutcome.PERMANENT, error_code="crm_malformed_response", http_status=response.status_code, ) error = payload.get("error") if error == "QUERY_LIMIT_EXCEEDED": return CrmResult(CrmOutcome.RATE_LIMITED, error_code=error, retry_after=retry_after) if error == "OPERATION_TIME_LIMIT": return CrmResult(CrmOutcome.RETRY, error_code=error, retry_after=retry_after) if error: permanent = error in { "ERROR_METHOD_NOT_FOUND", "ERROR_WRONG_AUTH_TYPE", "INVALID_CREDENTIALS", "ACCESS_DENIED", "ERROR_ARGUMENT", } return CrmResult( CrmOutcome.PERMANENT if permanent else CrmOutcome.RETRY, error_code=str(error)[:64], http_status=response.status_code, ) return CrmResult(CrmOutcome.SUCCEEDED, result=payload.get("result")) async def batch(self, commands: list[tuple[str, dict[str, Any]]]) -> list[CrmResult]: if not commands: return [] if len(commands) > 50: raise ValueError("Bitrix batch limit exceeded") cmd = { str(index): f"{method}?{httpx.QueryParams(params)}" for index, (method, params) in enumerate(commands) if method in ALLOWED_METHODS } batch = await self.call("batch", {"halt": 0, "cmd": cmd}, mutating=True) if batch.outcome != CrmOutcome.SUCCEEDED: return [batch for _ in commands] result = batch.result or {} successes = result.get("result", {}) errors = result.get("result_error", {}) return [ CrmResult(CrmOutcome.SUCCEEDED, result=successes.get(str(i))) if str(i) in successes else CrmResult(CrmOutcome.RETRY, error_code=str(errors.get(str(i), "batch_missing"))) for i in range(len(commands)) ] ALLOWED_METHODS = frozenset( { "batch", "crm.duplicate.findbycomm", "crm.contact.get", "crm.contact.add", "crm.contact.update", "crm.contact.userfield.list", "crm.item.list", "crm.item.get", "crm.item.add", "crm.item.update", } ) def _retry_after(response: httpx.Response) -> float | None: value = response.headers.get("Retry-After") if not value: return None try: return max(0.0, float(value)) except ValueError: try: retry_at = datetime.fromisoformat(value).astimezone(UTC) return max(0.0, (retry_at - datetime.now(UTC)).total_seconds()) except ValueError: return None class TokenBucket: def __init__(self, refill_per_second: float, burst: int, max_in_flight: int) -> None: self.refill_per_second = refill_per_second self.burst = float(burst) self.tokens = float(burst) self.updated_at = asyncio.get_running_loop().time() self._lock = asyncio.Lock() self._slots = asyncio.Semaphore(max_in_flight) async def acquire(self) -> None: await self._slots.acquire() while True: async with self._lock: now = asyncio.get_running_loop().time() self.tokens = min( self.burst, self.tokens + (now - self.updated_at) * self.refill_per_second ) self.updated_at = now if self.tokens >= 1: self.tokens -= 1 return delay = (1 - self.tokens) / self.refill_per_second await asyncio.sleep(delay) def release(self) -> None: self._slots.release()