inbox.py

Go to the documentation of this file.

"""The inbox: one JSON file per message, with states and retention.
 
Proposal V1 §4 (consumption) and §6/§10.1 (retention). Messages move
``unread -> read -> acted``; acting records *which decision* consumed the
message, so an approval can be traced back from the thing it authorised.
 
Retention is the operator's ruling: an acted message prunes one year after it
was acted on, and its audit hash survives the prune. The audit trail is the
durable record; the body is not.
 
Dependencies
------------
- oc_connect.audit (AuditStore, MAX_DECISION_REF, timestamp, utc_now - audits read, acted and pruned transitions)
- oc_connect.comms (message_filename - the bus filename convention for a stored message)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (InboundMessage, InboundState - the stored message model and its states)
"""
from __future__ import annotations
 
import json
import os
import secrets as _secrets
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from pathlib import Path
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import InboundMessage, InboundState
 
from ..audit import MAX_DECISION_REF, AuditStore, timestamp, utc_now
from ..comms import message_filename
 
RETENTION_DAYS = 365
 
 
class InboxStoreError(ValueError):
    """A stored inbound message is unreadable or malformed."""
 
 
def _parse_time(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
 
 
class InboxStore:
    """Durable per-message storage for one inbound lane directory."""
 
    def __init__(
        self,
        directory: str | Path,
        *,
        audit: AuditStore | None = None,
        now: Callable[[], datetime] = utc_now,
        source: str = "telegram",
    ) -> None:
        self.directory = Path(directory)
        self.audit = audit
        self._now = now
        self.source = source
 
    def _write(self, path: Path, message: InboundMessage) -> None:
        self.directory.mkdir(parents=True, exist_ok=True)
        temporary = path.with_name(f".tmp-{path.name}-{_secrets.token_hex(8)}")
        payload = json.dumps(message.model_dump(mode="json"), indent=2, sort_keys=True)
        try:
            with temporary.open("w", encoding="utf-8", newline="\n") as handle:
                handle.write(payload)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
    def paths(self) -> tuple[Path, ...]:
        """Return stored message files in arrival order, ignoring dotfiles."""
        if not self.directory.is_dir():
            return ()
        return tuple(
            sorted(
                path
                for path in self.directory.glob("*.json")
                if path.is_file() and not path.name.startswith(".")
            )
        )
 
    def _read(self, path: Path) -> InboundMessage:
        try:
            return InboundMessage.model_validate_json(
                path.read_text(encoding="utf-8")
            )
        except (OSError, UnicodeDecodeError, ValueError) as exc:
            raise InboxStoreError(
                f"could not parse inbound message at {path} ({type(exc).__name__})"
            ) from None
 
    def store(self, message: InboundMessage) -> Path:
        """Persist one newly-arrived message and return its path.
 
        An id collision is refused rather than overwritten. Two ULIDs collide
        only on a same-millisecond, same-80-bit-entropy draw, so in production
        this never fires - but the failure mode it prevents is an inbox that
        silently loses an operator reply, which must never be the quiet path.
        """
        received = _parse_time(message.received_at_utc)
        path = self.directory / message_filename(
            received, self.source, message.message_id
        )
        if path.exists():
            raise InboxStoreError(
                f"refusing to overwrite stored inbound message "
                f"{message.message_id!r} at {path}"
            )
        self._write(path, message)
        _dbg(
            "OC_CONNECT.INBOX",
            "store",
            f"stored message={message.message_id} lane={message.lane} "
            f"classification={message.classification}",
            2,
        )
        return path
 
    def list_messages(
        self,
        *,
        state: InboundState | None = None,
    ) -> tuple[InboundMessage, ...]:
        """Return stored messages in arrival order, optionally by state."""
        messages = tuple(self._read(path) for path in self.paths())
        if state is None:
            return messages
        return tuple(message for message in messages if message.state == state)
 
    def _locate(self, message_id: str) -> Path:
        suffix = f"_{message_id}.json"
        for path in self.paths():
            if path.name.endswith(suffix):
                return path
        raise KeyError(f"inbound message {message_id!r} is not in the inbox")
 
    def get(self, message_id: str) -> InboundMessage:
        """Return one stored message or fail with its public id."""
        return self._read(self._locate(message_id))
 
    def _audit(self, message: InboundMessage, outcome: str) -> None:
        if self.audit is None:
            return
        decision = message.consumed_by_decision
        self.audit.append(
            connector=message.connector,
            grant_id=None,
            destination=message.chat_id,
            outcome=outcome,  # type: ignore[arg-type]
            digest=message.content_hash,
            idempotency_key=message.message_id,
            attempts=0,
            lane=message.lane,
            direction=message.direction,
            risk_class=message.risk_class,
            decision_ref=decision[:MAX_DECISION_REF] if decision else None,
        )
 
    def mark_read(self, message_id: str) -> InboundMessage:
        """Move one unread message to ``read``; re-reading is idempotent."""
        path = self._locate(message_id)
        message = self._read(path)
        if message.state != "unread":
            return message
        updated = message.model_copy(
            update={"state": "read", "read_at_utc": timestamp(self._now())}
        )
        self._write(path, updated)
        self._audit(updated, "read")
        return updated
 
    def mark_acted(self, message_id: str, decision: str) -> InboundMessage:
        """Record which decision consumed one message (§4 consumption)."""
        if not decision.strip():
            raise ValueError("acting on a message requires the consuming decision")
        path = self._locate(message_id)
        message = self._read(path)
        if message.state == "acted":
            raise ValueError(
                f"inbound message {message_id!r} was already consumed by "
                f"{message.consumed_by_decision!r}"
            )
        stamped = timestamp(self._now())
        updated = message.model_copy(
            update={
                "state": "acted",
                "read_at_utc": message.read_at_utc or stamped,
                "acted_at_utc": stamped,
                "consumed_by_decision": decision.strip(),
            }
        )
        self._write(path, updated)
        self._audit(updated, "acted")
        _dbg(
            "OC_CONNECT.INBOX",
            "mark_acted",
            f"message={message_id} consumed by decision (length="
            f"{len(decision.strip())})",
            2,
        )
        return updated
 
    def count_since(self, cutoff: datetime) -> int:
        """Return how many messages arrived at or after *cutoff*."""
        return sum(
            1
            for message in self.list_messages()
            if _parse_time(message.received_at_utc) >= cutoff
        )
 
    def prune(self, *, retention_days: int = RETENTION_DAYS) -> tuple[str, ...]:
        """Delete acted messages older than the retention window.
 
        Only ``acted`` messages are eligible: an unread or read message is
        still work in progress, however old it is. The audit record and its
        content hash are untouched, so a pruned message stays provable.
        """
        if retention_days < 1:
            raise ValueError("retention_days must be at least one")
        cutoff = self._now().astimezone(timezone.utc) - timedelta(days=retention_days)
        pruned: list[str] = []
        for path in self.paths():
            message = self._read(path)
            if message.state != "acted" or message.acted_at_utc is None:
                continue
            if _parse_time(message.acted_at_utc) > cutoff:
                continue
            self._audit(message, "pruned")
            path.unlink()
            pruned.append(message.message_id)
        if pruned:
            _dbg(
                "OC_CONNECT.INBOX",
                "prune",
                f"pruned {len(pruned)} acted messages past {retention_days} days",
                2,
            )
        return tuple(pruned)
 
 
__all__ = ["RETENTION_DAYS", "InboxStore", "InboxStoreError"]