"""Private SQLite storage for negotiated hosted-room links. Route metadata and its scoped grant share the gateway's private root ``state.db``. SQLite WAL plus ``BEGIN IMMEDIATE`` owns concurrency; grants are never included in reprs, status payloads, or exception messages. """ from __future__ import annotations import contextlib import json import os import time from dataclasses import dataclass, field from functools import partial from pathlib import Path from typing import Any, Mapping from gateway import hosted_rooms from gateway.hosted_room_peer import ( GatewayRoomCatalog, HostedRoomPeerError, TransportSecurity, validate_room_link_url) from gateway.hosted_rooms_common import DbPath, compact_json, exact_fields, identifier MAX_LINKS = 512 MAX_GRANT_CHARS = 16 * 1024 _LEGACY_FIELDS = { "room_id", "member_id", "target_url", "target_profile", "grant", "catalog", "cancellation_scope_id", "trace_id", "updated_at"} _OPTIONAL_FIELDS = {"transport_security", "status"} # SQLite record columns that map 1:1 onto mapping fields, in record order # (``catalog_json`` is the serialized ``catalog``). _RECORD_FIELDS = ( "room_id", "member_id", "target_url", "target_profile", "grant", "cancellation_scope_id", "trace_id", "transport_security", "status", "updated_at") _STATUSES = {"ready", "unavailable", "needs_reauthorization"} @dataclass(frozen=True) class StoredRoomLink: room_id: str member_id: str target_url: str target_profile: str grant: str = field(repr=False) catalog: GatewayRoomCatalog cancellation_scope_id: str trace_id: str transport_security: TransportSecurity status: str updated_at: float @classmethod def from_mapping(cls, value: Mapping[str, Any]) -> "StoredRoomLink": _link_fields(value) room_id = _short_string(value["room_id"], "room_id") member_id = _short_string(value["member_id"], "member_id") target_profile = _short_string(value["target_profile"], "target_profile") target_url, detected_security = validate_room_link_url(value["target_url"]) transport_security = str(value.get("transport_security") or detected_security) if transport_security != detected_security: raise HostedRoomPeerError("transport_security does not match target_url") grant = str(value["grant"] or "") if not grant or len(grant) > MAX_GRANT_CHARS: raise HostedRoomPeerError("room grant is missing or too large") status = str(value.get("status") or "ready") if status not in _STATUSES: raise HostedRoomPeerError("stored room link status is invalid") updated_at = float(value["updated_at"]) if not updated_at > 0: raise HostedRoomPeerError("updated_at must be positive") return cls( room_id=room_id, member_id=member_id, target_url=target_url, target_profile=target_profile, grant=grant, catalog=GatewayRoomCatalog.from_mapping(value["catalog"]), cancellation_scope_id=_short_string(value["cancellation_scope_id"], "cancellation_scope_id"), trace_id=_short_string(value["trace_id"], "trace_id"), transport_security=transport_security, status=status, updated_at=updated_at) @classmethod def from_record(cls, value: Mapping[str, Any]) -> "StoredRoomLink": try: catalog = json.loads(str(value["catalog_json"])) except Exception as exc: raise HostedRoomPeerError("stored room link catalog is unreadable") from exc return cls.from_mapping({**{name: value[name] for name in _RECORD_FIELDS}, "catalog": catalog}) def catalog_mapping(self) -> dict[str, Any]: return self.catalog.as_mapping() def as_record(self) -> dict[str, Any]: record = {name: getattr(self, name) for name in _RECORD_FIELDS} record["catalog_json"] = compact_json(self.catalog_mapping(), ensure_ascii=False) return record _link_fields = partial( exact_fields, label="stored room link", required=_LEGACY_FIELDS, optional=_OPTIONAL_FIELDS, error=HostedRoomPeerError, not_object="stored room link fields are invalid", missing_fmt="stored room link fields are invalid", unknown_fmt="stored room link fields are invalid") def _short_string(value: Any, field: str) -> str: return identifier( str(value or ""), label=field, error=HostedRoomPeerError, max_chars=256, pattern=None, invalid=f"{field} is invalid") def _link_rows(db_path: DbPath) -> list[dict[str, Any]]: rows = hosted_rooms.list_room_link_records(db_path) if len(rows) > MAX_LINKS: raise HostedRoomPeerError("stored room link list is invalid") return rows def load_room_links(db_path: DbPath) -> tuple[StoredRoomLink, ...]: return tuple(StoredRoomLink.from_record(row) for row in _link_rows(db_path)) def load_room_links_tolerant(db_path: DbPath) -> tuple[tuple[StoredRoomLink, ...], tuple[str, ...]]: """Load healthy routes while quarantining malformed rows by identity.""" links, errors = [], [] for row in _link_rows(db_path): try: links.append(StoredRoomLink.from_record(row)) except Exception: errors.append(f"{row.get('room_id') or 'unknown'}:{row.get('member_id') or 'unknown'}:invalid") return tuple(links), tuple(errors) def save_room_link(db_path: DbPath, link: StoredRoomLink) -> None: hosted_rooms.upsert_room_link_record(db_path, record=link.as_record(), max_links=MAX_LINKS) if os.name == "posix": with contextlib.suppress(OSError): Path(db_path).chmod(0o600) def mark_room_link_status(db_path: DbPath, *, room_id: str, member_id: str, status: str) -> bool: if status not in _STATUSES: raise HostedRoomPeerError("stored room link status is invalid") return hosted_rooms.update_room_link_status( db_path, room_id=_short_string(room_id, "room_id"), member_id=_short_string(member_id, "member_id"), status=status) def make_stored_link( *, room_id: str, member_id: str, target_url: str, target_profile: str, grant: str, catalog: GatewayRoomCatalog, cancellation_scope_id: str, trace_id: str) -> StoredRoomLink: target_url, transport_security = validate_room_link_url(target_url) return StoredRoomLink.from_mapping({ "room_id": room_id, "member_id": member_id, "target_url": target_url, "target_profile": target_profile, "grant": grant, "catalog": catalog.as_mapping(), "cancellation_scope_id": cancellation_scope_id, "trace_id": trace_id, "transport_security": transport_security, "status": "ready", "updated_at": time.time()})