yovision/Brain/src/yovision_brain/integration/event_export/client.py

"""Synchronous, request-bound HTTPS delivery for Brain event exports."""

from __future__ import annotations

import hashlib
import json
import re
import secrets
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Protocol
from urllib.parse import urlsplit

from yovision_brain.integration.machine_identity import Signer, TransportPolicy

EVENT_PATH = "/v1/events"
_REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$")


class EventDeliveryError(RuntimeError):
    """An event was not accepted; callers must retain or reproduce the fact."""

    def __init__(self, code: str, *, terminal: bool) -> None:
        super().__init__(code)
        self.code = code
        self.terminal = terminal


@dataclass(frozen=True, slots=True)
class DeliveryResult:
    disposition: str
    request_id: str


class _Headers(Protocol):
    def get(self, name: str, default: str | None = None) -> str | None: ...


class _Response(Protocol):
    status: int
    headers: _Headers

    def read(self, amount: int = -1) -> bytes: ...
    def close(self) -> None: ...


class _Opener(Protocol):
    def open(self, request: urllib.request.Request, timeout: float) -> _Response: ...


class HTTPSMachineIdentitySender:
    def __init__(
        self,
        endpoint: str,
        signer: Signer,
        policy: TransportPolicy,
        *,
        opener: _Opener | None = None,
    ) -> None:
        parsed = urlsplit(endpoint)
        if (
            parsed.scheme != "https"
            or not parsed.hostname
            or parsed.username is not None
            or parsed.password is not None
            or parsed.path not in {"", "/"}
            or parsed.query
            or parsed.fragment
        ):
            raise ValueError("event export endpoint must be an HTTPS origin")
        policy.validate()
        self._endpoint = endpoint.rstrip("/")
        self._signer = signer
        self._policy = policy
        self._opener = opener or urllib.request.build_opener(
            urllib.request.HTTPSHandler(context=policy.ssl_context())
        )

    def send(self, body: bytes) -> DeliveryResult:
        if len(body) > self._policy.max_request_bytes:
            raise EventDeliveryError("event_request_too_large", terminal=True)
        request_id = "req-" + secrets.token_urlsafe(16)
        if not _REQUEST_ID.fullmatch(request_id):  # pragma: no cover - defensive invariant
            raise RuntimeError("generated request id is invalid")
        token = self._signer.mint(
            "yovision-sense", ("events:ingest",), "POST", EVENT_PATH, body
        )
        request = urllib.request.Request(
            self._endpoint + EVENT_PATH,
            data=body,
            method="POST",
            headers={
                "Authorization": "Bearer " + token,
                "Content-Type": "application/json",
                "X-Request-ID": request_id,
            },
        )
        try:
            response = self._opener.open(
                request, timeout=self._policy.request_timeout_ms / 1000
            )
        except urllib.error.HTTPError as exc:
            response_body = exc.read(64 * 1024 + 1)
            code = _problem_code(response_body) or "event_delivery_rejected"
            raise EventDeliveryError(
                code,
                terminal=400 <= exc.code < 500 and exc.code != 429,
            ) from None
        except (OSError, TimeoutError, urllib.error.URLError):
            raise EventDeliveryError("event_delivery_unavailable", terminal=False) from None

        try:
            response_body = response.read(64 * 1024 + 1)
            if len(response_body) > 64 * 1024:
                raise EventDeliveryError("event_response_invalid", terminal=False)
            if response.status not in {200, 201, 202}:
                raise EventDeliveryError(
                    "event_delivery_rejected",
                    terminal=400 <= response.status < 500 and response.status != 429,
                )
            response_request_id = response.headers.get("X-Request-ID")
            if response_request_id != request_id:
                raise EventDeliveryError("event_response_invalid", terminal=False)
            disposition = _disposition(response_body, response.status, body)
            return DeliveryResult(disposition=disposition, request_id=request_id)
        finally:
            response.close()


def _problem_code(body: bytes) -> str | None:
    try:
        value = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    code = value.get("code") if isinstance(value, dict) else None
    return code if isinstance(code, str) and re.fullmatch(r"[a-z][a-z0-9_]{0,63}", code) else None


def _disposition(body: bytes, status: int, request_body: bytes) -> str:
    try:
        value = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise EventDeliveryError("event_response_invalid", terminal=False) from None
    disposition = value.get("disposition") if isinstance(value, dict) else None
    allowed = {"accepted", "created", "duplicate"}
    if disposition not in allowed:
        raise EventDeliveryError("event_response_invalid", terminal=False)
    if status == 202 and disposition != "accepted":
        raise EventDeliveryError("event_response_invalid", terminal=False)
    try:
        sent = json.loads(request_body)
        response_identity = (
            value["producer_id"],
            value["source_event_id"],
            value["payload_sha256"],
        )
        expected_identity = (
            sent["producer_id"],
            sent["source_event_id"],
            hashlib.sha256(request_body).hexdigest(),
        )
    except (KeyError, TypeError, UnicodeDecodeError, json.JSONDecodeError):
        raise EventDeliveryError("event_response_invalid", terminal=False) from None
    if response_identity != expected_identity:
        raise EventDeliveryError("event_response_invalid", terminal=False)
    return disposition