outbox.py

Go to the documentation of this file.

"""Durable at-most-once outbox with retry, backoff, and destination caps.
 
Dependencies
------------
- oc_connect.audit (AuditStore, content_hash, timestamp - audits every queue and delivery outcome)
- oc_connect.workspace (comms_root - locates the outbox in the comms bus outbound lane)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (DeliveryReceipt, LaneId, MessageDirection, RiskClass - receipts and lane fields)
- pydantic (BaseModel, ConfigDict - the frozen strict OutboxItem model)
"""
from __future__ import annotations
 
import json
import os
import secrets as _secrets
import time
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Literal
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import (
    DeliveryReceipt,
    LaneId,
    MessageDirection,
    RiskClass,
)
from pydantic import BaseModel, ConfigDict
 
from .audit import AuditRecord, AuditStore, content_hash, timestamp, utc_now
from .workspace import comms_root
 
OutboxState = Literal["queued", "sent", "failed"]
 
 
def default_outbox_path(root: str | Path | None = None) -> Path:
    """Return the durable outbox inside the comms bus outbound lane (§3)."""
    return Path(comms_root(root)) / "outbound" / "outbox.json"
 
 
class OutboxItem(BaseModel):
    """One durable message; bodies live here, never in the audit file.
 
    See :class:`~oc_connect.audit.AuditRecord` for why ``lane`` and
    ``direction`` default to the single plan-219 lane while ``risk_class``
    defaults to ``None``: for a migrated record the lane is a fact and the risk
    class was never recorded.
 
    ``transport_ref`` holds the connector's own message identifier for a
    delivered item. It is what an inbound reply's ``reply_to`` metadata
    resolves against, so correlation needs no second index file (§4).
    """
 
    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
 
    connector: str
    grant_id: str
    destination: str
    content: str
    content_hash: str
    idempotency_key: str
    state: OutboxState
    attempts: int
    dispatch_claimed: bool
    created_at_utc: str
    updated_at_utc: str
    next_attempt_at_utc: str | None = None
    sent_at_utc: str | None = None
    error_code: str | None = None
    lane: LaneId = "outbound-telegram"
    direction: MessageDirection = "outbound"
    risk_class: RiskClass | None = None
    transport_ref: str | None = None
 
 
class OutboxFileError(ValueError):
    """The outbox file is unreadable or malformed."""
 
 
class Outbox:
    """Persist messages and drive bounded delivery through an injected transport."""
 
    def __init__(
        self,
        path: str | Path | None = None,
        *,
        audit: AuditStore | None = None,
        now: Callable[[], datetime] = utc_now,
        sleep: Callable[[float], None] = time.sleep,
        max_attempts: int = 3,
        backoff_seconds: tuple[float, ...] = (0.25, 0.5),
        rate_limit_count: int = 10,
        rate_limit_window_seconds: float = 60.0,
        root: str | Path | None = None,
    ) -> None:
        if max_attempts < 1:
            raise ValueError("max_attempts must be at least one")
        if len(backoff_seconds) < max_attempts - 1:
            raise ValueError("backoff_seconds must cover every retry")
        if rate_limit_count < 1 or rate_limit_window_seconds <= 0:
            raise ValueError("rate limit values must be positive")
        self.path = Path(path) if path is not None else default_outbox_path(root)
        self.audit = audit or AuditStore(root=root)
        self._now = now
        self._sleep = sleep
        self.max_attempts = max_attempts
        self.backoff_seconds = backoff_seconds
        self.rate_limit_count = rate_limit_count
        self.rate_limit_window_seconds = rate_limit_window_seconds
 
    def _load(self) -> dict[str, OutboxItem]:
        if not self.path.is_file():
            return {}
        try:
            payload = json.loads(self.path.read_text(encoding="utf-8"))
            items = {
                item.idempotency_key: item
                for item in (
                    OutboxItem.model_validate(raw)
                    for raw in payload.get("items", [])
                )
            }
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc:
            raise OutboxFileError(
                f"could not parse connector outbox at {self.path} "
                f"({type(exc).__name__})"
            ) from None
        return items
 
    def _write(self, items: dict[str, OutboxItem]) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary = self.path.with_name(
            f".tmp-{self.path.name}-{_secrets.token_hex(8)}"
        )
        payload = {
            "version": 1,
            "items": [items[key].model_dump(mode="json") for key in sorted(items)],
        }
        try:
            with temporary.open("w", encoding="utf-8", newline="\n") as handle:
                json.dump(payload, handle, indent=2, sort_keys=True)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, self.path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
    def enqueue(
        self,
        *,
        connector: str,
        grant_id: str,
        destination: str,
        content: str,
        idempotency_key: str,
        risk_class: RiskClass | None = None,
        lane: LaneId = "outbound-telegram",
        direction: MessageDirection = "outbound",
    ) -> OutboxItem:
        """Persist one queued item, deduplicating by its caller-owned key."""
        fields = (connector, grant_id, destination, content, idempotency_key)
        if not all(value.strip() for value in fields):
            raise ValueError("outbox fields must not be blank")
        items = self._load()
        digest = content_hash(content)
        existing = items.get(idempotency_key)
        if existing is not None:
            if (
                existing.connector == connector
                and existing.grant_id == grant_id
                and existing.destination == destination
                and existing.content_hash == digest
            ):
                return existing
            raise ValueError(
                f"idempotency key {idempotency_key!r} already names different content or scope"
            )
        stamped = timestamp(self._now())
        item = OutboxItem(
            connector=connector,
            grant_id=grant_id,
            destination=destination,
            content=content,
            content_hash=digest,
            idempotency_key=idempotency_key,
            state="queued",
            attempts=0,
            dispatch_claimed=False,
            created_at_utc=stamped,
            updated_at_utc=stamped,
            lane=lane,
            direction=direction,
            risk_class=risk_class,
        )
        items[idempotency_key] = item
        self._write(items)
        self.audit.append(
            connector=connector,
            grant_id=grant_id,
            destination=destination,
            outcome="queued",
            digest=digest,
            idempotency_key=idempotency_key,
            attempts=0,
            lane=lane,
            direction=direction,
            risk_class=risk_class,
        )
        return item
 
    def get(self, idempotency_key: str) -> OutboxItem:
        """Return one item or raise its public key."""
        try:
            return self._load()[idempotency_key]
        except KeyError:
            raise KeyError(f"outbox item {idempotency_key!r} does not exist") from None
 
    def list_items(self) -> tuple[OutboxItem, ...]:
        """Return items in deterministic idempotency-key order."""
        items = self._load()
        return tuple(items[key] for key in sorted(items))
 
    @staticmethod
    def _parse_time(value: str) -> datetime:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
 
    def _rate_limited(self, item: OutboxItem, items: dict[str, OutboxItem]) -> bool:
        cutoff = self._now().astimezone(timezone.utc) - timedelta(
            seconds=self.rate_limit_window_seconds
        )
        sent = sum(
            1
            for other in items.values()
            if other.destination == item.destination
            and other.state == "sent"
            and other.sent_at_utc is not None
            and self._parse_time(other.sent_at_utc) >= cutoff
        )
        return sent >= self.rate_limit_count
 
    @staticmethod
    def _receipt(item: OutboxItem, record: AuditRecord) -> DeliveryReceipt:
        return DeliveryReceipt(
            connector=item.connector,
            destination=item.destination,
            idempotency_key=item.idempotency_key,
            outcome=record.outcome,
            audit_id=record.audit_id,
            attempts=item.attempts,
        )
 
    def _record(
        self,
        item: OutboxItem,
        outcome: Literal["sent", "failed", "deferred"],
        *,
        error_code: str | None = None,
    ) -> DeliveryReceipt:
        record = self.audit.append(
            connector=item.connector,
            grant_id=item.grant_id,
            destination=item.destination,
            outcome=outcome,
            digest=item.content_hash,
            idempotency_key=item.idempotency_key,
            attempts=item.attempts,
            error_code=error_code,
            lane=item.lane,
            direction=item.direction,
            risk_class=item.risk_class,
        )
        return self._receipt(item, record)
 
    def find_by_transport_ref(self, transport_ref: str) -> OutboxItem | None:
        """Return the delivered item a connector message id belongs to.
 
        This is the correlation seam (§4): an inbound reply carries the
        ``message_id`` of the outbound message it answers, and that id was
        recorded here at delivery. Resolution therefore needs no side index
        that could drift from the outbox.
        """
        if not transport_ref.strip():
            return None
        items = self._load()
        for key in sorted(items):
            candidate = items[key]
            if candidate.transport_ref == transport_ref:
                return candidate
        return None
 
    def process(
        self,
        idempotency_key: str,
        send: Callable[[OutboxItem], str | None],
    ) -> DeliveryReceipt:
        """Process one item through a durable at-most-once dispatch claim.
 
        *send* may return the connector's message identifier; when it does, the
        reference is persisted on the item so inbound replies can correlate
        against it.
        """
        items = self._load()
        try:
            item = items[idempotency_key]
        except KeyError:
            raise KeyError(f"outbox item {idempotency_key!r} does not exist") from None
 
        if item.state in ("sent", "failed"):
            existing = self.audit.latest(idempotency_key)
            if existing is None or existing.outcome not in ("sent", "failed"):
                return self._record(item, item.state)
            return self._receipt(item, existing)
 
        if item.dispatch_claimed:
            item = item.model_copy(
                update={
                    "state": "failed",
                    "updated_at_utc": timestamp(self._now()),
                    "error_code": "ambiguous_delivery_after_restart",
                }
            )
            items[idempotency_key] = item
            self._write(items)
            return self._record(
                item,
                "failed",
                error_code="ambiguous_delivery_after_restart",
            )
 
        if self._rate_limited(item, items):
            item = item.model_copy(
                update={
                    "next_attempt_at_utc": timestamp(
                        self._now() + timedelta(seconds=self.rate_limit_window_seconds)
                    ),
                    "updated_at_utc": timestamp(self._now()),
                }
            )
            items[idempotency_key] = item
            self._write(items)
            return self._record(item, "deferred", error_code="destination_rate_cap")
 
        for attempt in range(item.attempts + 1, self.max_attempts + 1):
            item = item.model_copy(
                update={
                    "attempts": attempt,
                    "dispatch_claimed": True,
                    "updated_at_utc": timestamp(self._now()),
                    "next_attempt_at_utc": None,
                    "error_code": None,
                }
            )
            items[idempotency_key] = item
            self._write(items)
            try:
                transport_ref = send(item)
            except Exception as exc:
                error_code = type(exc).__name__
                terminal = attempt >= self.max_attempts
                item = item.model_copy(
                    update={
                        "state": "failed" if terminal else "queued",
                        "dispatch_claimed": False,
                        "updated_at_utc": timestamp(self._now()),
                        "error_code": error_code,
                    }
                )
                items[idempotency_key] = item
                self._write(items)
                _dbg(
                    "OC_CONNECT.OUTBOX",
                    "process",
                    f"transport attempt={attempt} connector={item.connector} "
                    f"destination={item.destination} error={error_code}",
                    2,
                )
                if terminal:
                    return self._record(item, "failed", error_code=error_code)
                self._sleep(self.backoff_seconds[attempt - 1])
                continue
 
            item = item.model_copy(
                update={
                    "state": "sent",
                    "dispatch_claimed": False,
                    "updated_at_utc": timestamp(self._now()),
                    "sent_at_utc": timestamp(self._now()),
                    "error_code": None,
                    "transport_ref": (
                        str(transport_ref) if transport_ref is not None else None
                    ),
                }
            )
            items[idempotency_key] = item
            self._write(items)
            return self._record(item, "sent")
 
        raise RuntimeError("outbox retry loop exhausted without a terminal receipt")
 
 
__all__ = [
    "Outbox",
    "OutboxFileError",
    "OutboxItem",
    "OutboxState",
    "default_outbox_path",
]