yilsem/aiturk-hermes-ide
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 3 Wiki Activity
Files
aiturk/main
aiturk-hermes-ide/gateway/hosted_rooms.py
T

2447 lines
87 KiB
Python
Raw Permalink Blame History

"""Durable state for gateway-hosted Bot Mode rooms.
This module owns only hosted-room identity and its append-only event log. It
does not deliver events, lease relay work, or run agent turns; those concerns
belong to the relay and the future hosted-room driver. Keeping that boundary
explicit lets the room log compose with a durable relay without creating a
second transport queue.
The caller supplies the database path so tests and alternate gateway layouts
can isolate state. Production handlers use the gateway's root ``state.db``.
"""
from __future__ import annotations
import hashlib
import json
import re
import sqlite3
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Iterator, Mapping, NoReturn
PROTOCOL_VERSION = 2
MAX_ROOM_ID_CHARS = 128
MAX_EVENT_ID_CHARS = 128
MAX_ROOM_NAME_CHARS = 200
MAX_EVENT_KIND_CHARS = 64
MAX_ACTOR_ID_CHARS = 128
MAX_ACTOR_LABEL_CHARS = 200
MAX_MEMBERS = 128
MAX_MEMBERS_JSON_BYTES = 128 * 1024
MAX_EVENT_JSON_BYTES = 256 * 1024
MAX_LOG_LIMIT = 500
MAX_LOG_PAGE_BYTES = 2 * 1024 * 1024
MAX_ROOM_LIST_LIMIT = 500
MAX_ACTIVE_ROOMS = 256
MAX_DISBANDED_ROOM_TOMBSTONES = 512
DISBANDED_ROOM_RETENTION_SECONDS = 90 * 24 * 60 * 60
MAX_EVENTS_PER_ROOM = 50_000
MAX_ROOM_EVENT_BYTES = 256 * 1024 * 1024
# Leave substantial headroom below the pre-update state.db snapshot ceiling.
# Event accounting does not include SQLite indexes or repeated room ids, so the
# logical budget must stay well below the physical-file limit.
MAX_GATEWAY_EVENT_BYTES = 16 * 1024 * 1024
CONTROL_EVENT_COUNT_RESERVE = 64
CONTROL_EVENT_BYTE_RESERVE = 1024 * 1024
_JOURNAL_MODE_LOCK_RETRIES = 8
_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$")
_EVENT_KIND_RE = re.compile(r"^[a-z][a-z0-9_.-]*$")
_ROOM_SCHEMA_COLUMNS = frozenset({
"room_id",
"name",
"members_json",
"authority_gateway_id",
"authority_epoch",
"next_seq",
"event_bytes",
"revision",
"created_at",
"updated_at",
"disbanded_at",
})
_EVENT_SCHEMA_COLUMNS = frozenset({
"room_id",
"seq",
"event_id",
"kind",
"actor_json",
"authority_epoch",
"payload_json",
"created_at",
})
_RETIRED_ROOM_SCHEMA_COLUMNS = frozenset({"room_id", "retired_at"})
_LINK_SCHEMA_COLUMNS = frozenset({
"room_id",
"member_id",
"target_url",
"target_profile",
"grant",
"catalog_json",
"cancellation_scope_id",
"trace_id",
"transport_security",
"status",
"updated_at",
})
_REMOTE_RUN_SCHEMA_COLUMNS = frozenset({
"room_id",
"home_install_id",
"authority_gateway_id",
"authority_epoch",
"member_id",
"task_id",
"execution_generation",
"target_install_id",
"target_profile",
"run_id",
"session_id",
"created_at",
"updated_at",
})
_REMOTE_RUN_IDENTITY_COLUMNS = (
"room_id",
"home_install_id",
"authority_gateway_id",
"authority_epoch",
"member_id",
"target_install_id",
"target_profile",
"task_id",
"execution_generation",
)
_REVOKED_GRANT_SCHEMA_COLUMNS = frozenset({
"scope_key",
"expires_at",
"revoked_before",
})
_PEER_RESERVATION_SCHEMA_COLUMNS = frozenset({
"room_id",
"member_id",
"target_profile",
"authority_gateway_id",
"authority_epoch",
"expires_at",
"revoked_at",
"created_at",
"updated_at",
})
_EVENT_KINDS_BY_ACTOR = {
"user": frozenset({"message.user"}),
"member": frozenset({"message.member"}),
"gateway": frozenset({
"member.unavailable",
"room.activity",
"room.stop_requested",
"turn.deferred",
"turn.reassigned",
"turn.cancelled",
"turn.failed",
"turn.settled",
"turn.started",
}),
"system": frozenset({
"authority.claimed",
"authority.lost",
"room.created",
"room.disbanded",
"room.members_changed",
"room.renamed",
}),
}
_ACTOR_FIELDS = frozenset({"kind", "id", "display_name", "profile", "connection_id"})
class HostedRoomError(ValueError):
"""Base class for invalid or conflicting hosted-room operations."""
class RoomNotFoundError(HostedRoomError):
"""Raised when a room does not exist or has been disbanded."""
class RoomHistoryExpiredError(RoomNotFoundError):
"""Raised when a retired room remains reserved after history compaction."""
reason = "room_history_expired"
class RoomConflictError(HostedRoomError):
"""Raised when an idempotency key is reused for different room state."""
class RoomProbeUnavailableError(HostedRoomError):
"""Raised when a non-blocking ownership probe cannot read the room store."""
class EventConflictError(HostedRoomError):
"""Raised when an event id is reused with different immutable content."""
class AuthorityConflictError(HostedRoomError):
"""Raised when a stale room authority attempts to mutate hosted state."""
reason = "authority_conflict"
class AuthoritySupersededError(AuthorityConflictError):
"""Raised when a successful authority claim was later superseded."""
def _primary_key_columns(conn: sqlite3.Connection, table: str) -> tuple[str, ...]:
return tuple(
str(row[1])
for row in sorted(
(row for row in conn.execute(f"PRAGMA table_info({table})") if row[5]),
key=lambda row: int(row[5]),
)
)
def _migrate_remote_run_schema(conn: sqlite3.Connection) -> None:
"""Fence legacy receipts behind a complete authority-lineage key."""
columns = {
str(row[1])
for row in conn.execute("PRAGMA table_info(hosted_room_remote_runs)")
}
if (
_REMOTE_RUN_SCHEMA_COLUMNS.issubset(columns)
and _primary_key_columns(conn, "hosted_room_remote_runs")
== _REMOTE_RUN_IDENTITY_COLUMNS
):
return
conn.execute("DROP TABLE IF EXISTS hosted_room_remote_runs_migrating")
conn.execute(
"""CREATE TABLE hosted_room_remote_runs_migrating (
room_id TEXT NOT NULL,
home_install_id TEXT NOT NULL,
authority_gateway_id TEXT NOT NULL,
authority_epoch INTEGER NOT NULL CHECK (authority_epoch >= 1),
member_id TEXT NOT NULL,
task_id TEXT NOT NULL,
execution_generation INTEGER NOT NULL CHECK (execution_generation >= 1),
target_install_id TEXT NOT NULL,
target_profile TEXT NOT NULL,
run_id TEXT NOT NULL,
session_id TEXT NOT NULL,
created_at REAL NOT NULL,
updated_at REAL NOT NULL,
PRIMARY KEY (
room_id, home_install_id, authority_gateway_id, authority_epoch,
member_id, target_install_id, target_profile, task_id,
execution_generation
)
)"""
)
if columns:
home = "home_install_id" if "home_install_id" in columns else "'legacy'"
gateway = (
"authority_gateway_id"
if "authority_gateway_id" in columns
else "'legacy'"
)
epoch = "authority_epoch" if "authority_epoch" in columns else "1"
conn.execute(
f"""INSERT OR IGNORE INTO hosted_room_remote_runs_migrating(
room_id, home_install_id, authority_gateway_id,
authority_epoch, member_id, task_id,
execution_generation, target_install_id, target_profile,
run_id, session_id, created_at, updated_at
)
SELECT room_id, {home}, {gateway}, {epoch}, member_id, task_id,
execution_generation, target_install_id, target_profile,
run_id, session_id, created_at, updated_at
FROM hosted_room_remote_runs"""
)
conn.execute("DROP TABLE hosted_room_remote_runs")
conn.execute(
"ALTER TABLE hosted_room_remote_runs_migrating "
"RENAME TO hosted_room_remote_runs"
)
def default_db_path() -> Path:
"""Return the gateway-wide state database for the active install."""
from hermes_constants import get_hermes_home
home = get_hermes_home()
root = home.parent.parent if home.parent.name == "profiles" else home
return root / "state.db"
def local_authority_gateway_id() -> str:
"""Return the stable server-owned identity for hosted-room authority."""
from hermes_cli.install_identity import get_install_id
install_id = get_install_id()
if not install_id:
raise HostedRoomError("stable gateway install identity is unavailable")
return _validate_identifier(
f"install:{install_id}",
label="authority_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
def _canonical_json(value: Any, *, label: str, max_bytes: int) -> str:
try:
encoded = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
except (TypeError, ValueError, RecursionError) as exc:
raise HostedRoomError(f"{label} must be JSON-serializable") from exc
if len(encoded.encode("utf-8")) > max_bytes:
raise HostedRoomError(f"{label} is too large")
return encoded
def _validate_identifier(value: Any, *, label: str, max_chars: int) -> str:
if not isinstance(value, str):
raise HostedRoomError(f"{label} must be a string")
value = value.strip()
if not value or len(value) > max_chars or not _IDENTIFIER_RE.fullmatch(value):
raise HostedRoomError(f"invalid {label}")
return value
def user_event_id(client_event_id: Any) -> str:
"""Map a client retry key into the server-owned user-event namespace."""
normalized = _validate_identifier(
client_event_id,
label="event_id",
max_chars=MAX_EVENT_ID_CHARS,
)
digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()
return f"user:{digest}"
def _validate_room_name(value: Any) -> str:
if not isinstance(value, str):
raise HostedRoomError("name must be a string")
value = value.strip()
if not value or len(value) > MAX_ROOM_NAME_CHARS:
raise HostedRoomError("invalid room name")
return value
def _validate_members(value: Any) -> tuple[list[dict[str, Any]], str]:
if not isinstance(value, list):
raise HostedRoomError("members must be a list")
if len(value) > MAX_MEMBERS:
raise HostedRoomError("too many room members")
members: list[dict[str, Any]] = []
for member in value:
if not isinstance(member, dict):
raise HostedRoomError("each room member must be an object")
members.append(dict(member))
encoded = _canonical_json(
members,
label="members",
max_bytes=MAX_MEMBERS_JSON_BYTES,
)
return members, encoded
def _legacy_members_match(
existing_json: str,
proposed: list[dict[str, Any]],
) -> bool:
"""Allow adoption to add routing metadata an older room could not store."""
try:
existing = json.loads(existing_json)
except (TypeError, ValueError):
return False
if not isinstance(existing, list) or len(existing) != len(proposed):
return False
for previous, current in zip(existing, proposed, strict=True):
if not isinstance(previous, dict):
return False
previous = dict(previous)
current = dict(current)
previous_target = previous.pop("target", None)
current_target = current.pop("target", None)
if previous != current:
return False
if previous_target not in (None, {}) and previous_target != current_target:
return False
return True
def _validate_event_kind(value: Any) -> str:
if not isinstance(value, str):
raise HostedRoomError("kind must be a string")
value = value.strip()
if (
not value
or len(value) > MAX_EVENT_KIND_CHARS
or not _EVENT_KIND_RE.fullmatch(value)
):
raise HostedRoomError("invalid event kind")
return value
def _optional_actor_field(actor: dict[str, Any], field: str, max_chars: int) -> str:
value = actor.get(field)
if value is None:
return ""
if not isinstance(value, str):
raise HostedRoomError(f"actor.{field} must be a string")
value = value.strip()
if len(value) > max_chars:
raise HostedRoomError(f"actor.{field} is too long")
return value
def _validate_actor(value: Any, *, kind: str) -> tuple[dict[str, str], str]:
if not isinstance(value, dict):
raise HostedRoomError("actor must be an object")
unknown = set(value) - _ACTOR_FIELDS
if unknown:
raise HostedRoomError(f"unknown actor fields: {', '.join(sorted(unknown))}")
actor_kind = value.get("kind")
if not isinstance(actor_kind, str) or actor_kind not in _EVENT_KINDS_BY_ACTOR:
raise HostedRoomError("invalid actor.kind")
if kind not in _EVENT_KINDS_BY_ACTOR[actor_kind]:
raise HostedRoomError(f"actor kind '{actor_kind}' cannot append '{kind}'")
actor_id = _validate_identifier(
value.get("id"),
label="actor.id",
max_chars=MAX_ACTOR_ID_CHARS,
)
actor = {"kind": actor_kind, "id": actor_id}
for field, max_chars in (
("display_name", MAX_ACTOR_LABEL_CHARS),
("profile", MAX_ACTOR_ID_CHARS),
("connection_id", MAX_ACTOR_ID_CHARS),
):
field_value = _optional_actor_field(value, field, max_chars)
if field_value:
actor[field] = field_value
encoded = _canonical_json(
actor,
label="actor",
max_bytes=4 * 1024,
)
return actor, encoded
def _initialize_schema(conn: sqlite3.Connection) -> None:
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_rooms (
room_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
members_json TEXT NOT NULL,
authority_gateway_id TEXT NOT NULL,
authority_epoch INTEGER NOT NULL DEFAULT 1 CHECK (authority_epoch >= 1),
next_seq INTEGER NOT NULL DEFAULT 1 CHECK (next_seq >= 1),
event_bytes INTEGER NOT NULL DEFAULT 0 CHECK (event_bytes >= 0),
revision INTEGER NOT NULL DEFAULT 1 CHECK (revision >= 1),
created_at REAL NOT NULL,
updated_at REAL NOT NULL,
disbanded_at REAL
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_room_events (
room_id TEXT NOT NULL,
seq INTEGER NOT NULL CHECK (seq >= 1),
event_id TEXT NOT NULL,
kind TEXT NOT NULL,
actor_json TEXT NOT NULL,
authority_epoch INTEGER CHECK (authority_epoch IS NULL OR authority_epoch >= 1),
payload_json TEXT NOT NULL,
created_at REAL NOT NULL,
PRIMARY KEY (room_id, seq),
UNIQUE (room_id, event_id),
FOREIGN KEY (room_id) REFERENCES hosted_rooms(room_id)
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_room_retired_ids (
room_id TEXT PRIMARY KEY,
retired_at REAL NOT NULL
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_room_links (
room_id TEXT NOT NULL,
member_id TEXT NOT NULL,
target_url TEXT NOT NULL,
target_profile TEXT NOT NULL,
grant TEXT NOT NULL,
catalog_json TEXT NOT NULL,
cancellation_scope_id TEXT NOT NULL,
trace_id TEXT NOT NULL,
transport_security TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'ready',
updated_at REAL NOT NULL,
PRIMARY KEY (room_id, member_id)
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_room_remote_runs (
room_id TEXT NOT NULL,
home_install_id TEXT NOT NULL,
authority_gateway_id TEXT NOT NULL,
authority_epoch INTEGER NOT NULL CHECK (authority_epoch >= 1),
member_id TEXT NOT NULL,
task_id TEXT NOT NULL,
execution_generation INTEGER NOT NULL CHECK (execution_generation >= 1),
target_install_id TEXT NOT NULL,
target_profile TEXT NOT NULL,
run_id TEXT NOT NULL,
session_id TEXT NOT NULL,
created_at REAL NOT NULL,
updated_at REAL NOT NULL,
PRIMARY KEY (
room_id, home_install_id, authority_gateway_id, authority_epoch,
member_id, target_install_id, target_profile, task_id,
execution_generation
)
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_room_revoked_grants (
scope_key TEXT PRIMARY KEY,
expires_at REAL NOT NULL,
revoked_before REAL NOT NULL
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS hosted_room_peer_reservations (
room_id TEXT NOT NULL,
member_id TEXT NOT NULL,
target_profile TEXT NOT NULL,
authority_gateway_id TEXT NOT NULL,
authority_epoch INTEGER NOT NULL CHECK (authority_epoch >= 1),
expires_at REAL NOT NULL,
revoked_at REAL,
created_at REAL NOT NULL,
updated_at REAL NOT NULL,
PRIMARY KEY (room_id, member_id, target_profile)
)"""
)
room_columns = {row[1] for row in conn.execute("PRAGMA table_info(hosted_rooms)")}
if "authority_gateway_id" not in room_columns:
conn.execute(
"ALTER TABLE hosted_rooms "
"ADD COLUMN authority_gateway_id TEXT NOT NULL DEFAULT 'legacy'"
)
if "authority_epoch" not in room_columns:
conn.execute(
"ALTER TABLE hosted_rooms "
"ADD COLUMN authority_epoch INTEGER NOT NULL DEFAULT 1"
)
backfill_event_bytes = "event_bytes" not in room_columns
if backfill_event_bytes:
conn.execute(
"ALTER TABLE hosted_rooms ADD COLUMN event_bytes INTEGER NOT NULL DEFAULT 0"
)
event_columns = {
row[1] for row in conn.execute("PRAGMA table_info(hosted_room_events)")
}
if "actor_json" not in event_columns:
# Draft builds before the actor contract carried no identity. Preserve
# their inert replay rows explicitly as legacy system events rather
# than guessing a user or Bot author.
legacy_actor = _canonical_json(
{"kind": "system", "id": "legacy"},
label="actor",
max_bytes=4 * 1024,
)
escaped_actor = legacy_actor.replace("'", "''")
conn.execute(
"ALTER TABLE hosted_room_events "
f"ADD COLUMN actor_json TEXT NOT NULL DEFAULT '{escaped_actor}'"
)
if "authority_epoch" not in event_columns:
conn.execute(
"ALTER TABLE hosted_room_events ADD COLUMN authority_epoch INTEGER"
)
if backfill_event_bytes:
conn.execute(
"""UPDATE hosted_rooms
SET event_bytes=COALESCE((
SELECT SUM(
length(CAST(event_id AS BLOB)) +
length(CAST(kind AS BLOB)) +
length(CAST(actor_json AS BLOB)) +
length(CAST(payload_json AS BLOB))
)
FROM hosted_room_events
WHERE hosted_room_events.room_id=hosted_rooms.room_id
), 0)"""
)
# Old schemas kept the final identity tombstone in hosted_rooms itself.
# Copy those identities before bounded history pruning can remove their
# heavier room/event payloads. This compact registry is intentionally
# permanent: a stale coordinate must never name a different Group Chat.
conn.execute(
"""INSERT OR IGNORE INTO hosted_room_retired_ids (room_id, retired_at)
SELECT room_id, disbanded_at FROM hosted_rooms
WHERE disbanded_at IS NOT NULL"""
)
_migrate_remote_run_schema(conn)
conn.execute(
"""CREATE INDEX IF NOT EXISTS idx_hosted_room_events_cursor
ON hosted_room_events(room_id, seq)"""
)
if not _schema_is_current(conn):
raise HostedRoomError("hosted room schema migration did not complete")
def _schema_is_current(conn: sqlite3.Connection) -> bool:
room_columns = frozenset(
row[1] for row in conn.execute("PRAGMA table_info(hosted_rooms)")
)
event_columns = frozenset(
row[1] for row in conn.execute("PRAGMA table_info(hosted_room_events)")
)
retired_room_columns = frozenset(
row[1] for row in conn.execute("PRAGMA table_info(hosted_room_retired_ids)")
)
link_columns = frozenset(
row[1] for row in conn.execute("PRAGMA table_info(hosted_room_links)")
)
remote_run_columns = frozenset(
row[1] for row in conn.execute("PRAGMA table_info(hosted_room_remote_runs)")
)
revoked_grant_columns = frozenset(
row[1]
for row in conn.execute("PRAGMA table_info(hosted_room_revoked_grants)")
)
peer_reservation_columns = frozenset(
row[1]
for row in conn.execute("PRAGMA table_info(hosted_room_peer_reservations)")
)
if not _ROOM_SCHEMA_COLUMNS.issubset(room_columns):
return False
if not _EVENT_SCHEMA_COLUMNS.issubset(event_columns):
return False
if not _RETIRED_ROOM_SCHEMA_COLUMNS.issubset(retired_room_columns):
return False
if not _LINK_SCHEMA_COLUMNS.issubset(link_columns):
return False
if not _REMOTE_RUN_SCHEMA_COLUMNS.issubset(remote_run_columns):
return False
if (
_primary_key_columns(conn, "hosted_room_remote_runs")
!= _REMOTE_RUN_IDENTITY_COLUMNS
):
return False
if not _REVOKED_GRANT_SCHEMA_COLUMNS.issubset(revoked_grant_columns):
return False
if not _PEER_RESERVATION_SCHEMA_COLUMNS.issubset(peer_reservation_columns):
return False
index = conn.execute(
"""SELECT 1 FROM sqlite_master
WHERE type='index' AND name='idx_hosted_room_events_cursor'"""
).fetchone()
return index is not None
def list_room_link_records(db_path: Path | str) -> list[dict[str, Any]]:
"""Return private RoomLink records without logging or formatting grants."""
with _transaction(db_path) as conn:
rows = conn.execute(
"""SELECT room_id, member_id, target_url, target_profile, grant,
catalog_json, cancellation_scope_id, trace_id,
transport_security, status, updated_at
FROM hosted_room_links
ORDER BY room_id, member_id"""
).fetchall()
return [dict(row) for row in rows]
def upsert_room_link_record(
db_path: Path | str,
*,
record: Mapping[str, Any],
max_links: int,
) -> None:
"""Atomically insert or replace one private RoomLink record."""
with _transaction(db_path, immediate=True) as conn:
existing = conn.execute(
"SELECT 1 FROM hosted_room_links WHERE room_id=? AND member_id=?",
(record["room_id"], record["member_id"]),
).fetchone()
if existing is None:
count = int(
conn.execute("SELECT COUNT(*) FROM hosted_room_links").fetchone()[0]
)
if count >= max_links:
raise HostedRoomError("too many stored room links")
conn.execute(
"""INSERT INTO hosted_room_links(
room_id, member_id, target_url, target_profile, grant,
catalog_json, cancellation_scope_id, trace_id,
transport_security, status, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(room_id, member_id) DO UPDATE SET
target_url=excluded.target_url,
target_profile=excluded.target_profile,
grant=excluded.grant,
catalog_json=excluded.catalog_json,
cancellation_scope_id=excluded.cancellation_scope_id,
trace_id=excluded.trace_id,
transport_security=excluded.transport_security,
status=excluded.status,
updated_at=excluded.updated_at""",
(
record["room_id"],
record["member_id"],
record["target_url"],
record["target_profile"],
record["grant"],
record["catalog_json"],
record["cancellation_scope_id"],
record["trace_id"],
record["transport_security"],
record["status"],
record["updated_at"],
),
)
def update_room_link_status(
db_path: Path | str,
*,
room_id: str,
member_id: str,
status: str,
now: float | None = None,
) -> bool:
"""Persist a non-secret route health classification."""
with _transaction(db_path, immediate=True) as conn:
cursor = conn.execute(
"""UPDATE hosted_room_links SET status=?, updated_at=?
WHERE room_id=? AND member_id=?""",
(
status,
float(now if now is not None else time.time()),
room_id,
member_id,
),
)
return cursor.rowcount == 1
def delete_room_link_records(db_path: Path | str, *, room_id: str) -> int:
"""Delete persisted peer routes after their target grants are revoked."""
with _transaction(db_path, immediate=True) as conn:
cursor = conn.execute(
"DELETE FROM hosted_room_links WHERE room_id=?",
(room_id,),
)
return cursor.rowcount
def _room_grant_scope_key(claims: Mapping[str, Any]) -> str:
"""Return a stable non-secret key for one room/home/target/profile scope."""
import hashlib
fields = {
key: str(claims.get(key) or "")
for key in (
"room_id",
"home_install_id",
"authority_gateway_id",
"authority_epoch",
"member_id",
"target_install_id",
"target_profile",
)
}
if not all(fields.values()):
raise HostedRoomError("room grant scope is incomplete")
return hashlib.sha256(
json.dumps(fields, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
def revoke_room_grant_scope(
db_path: Path | str,
*,
claims: Mapping[str, Any],
expires_at: float,
now: float | None = None,
) -> None:
"""Revoke every grant issued at or before now for one exact room scope."""
scope_key = _room_grant_scope_key(claims)
timestamp = float(now if now is not None else time.time())
expiry = float(expires_at)
if expiry <= timestamp:
return
with _transaction(db_path, immediate=True) as conn:
conn.execute(
"DELETE FROM hosted_room_revoked_grants WHERE expires_at<=?",
(timestamp,),
)
conn.execute(
"""INSERT INTO hosted_room_revoked_grants(
scope_key, expires_at, revoked_before
) VALUES (?, ?, ?)
ON CONFLICT(scope_key) DO UPDATE SET
expires_at=MAX(hosted_room_revoked_grants.expires_at,
excluded.expires_at),
revoked_before=MAX(hosted_room_revoked_grants.revoked_before,
excluded.revoked_before)""",
(scope_key, expiry, timestamp),
)
conn.execute(
"""UPDATE hosted_room_peer_reservations
SET revoked_at=?, updated_at=?
WHERE room_id=? AND member_id=? AND target_profile=?
AND authority_gateway_id=? AND authority_epoch=?""",
(
timestamp,
timestamp,
str(claims.get("room_id") or ""),
str(claims.get("member_id") or ""),
str(claims.get("target_profile") or ""),
str(claims.get("authority_gateway_id") or ""),
int(claims.get("authority_epoch") or 0),
),
)
def reserve_peer_room(
db_path: Path | str,
*,
claims: Mapping[str, Any],
expires_at: float,
now: float | None = None,
) -> None:
"""Fence direct Desktop prompts before the first peer run is admitted."""
timestamp = float(now if now is not None else time.time())
expiry = float(expires_at)
if expiry <= timestamp:
raise HostedRoomError("peer room reservation must expire in the future")
values = (
_validate_identifier(
claims.get("room_id"), label="room_id", max_chars=MAX_ROOM_ID_CHARS
),
_validate_identifier(
claims.get("member_id"), label="member_id", max_chars=MAX_ACTOR_ID_CHARS
),
_validate_identifier(
claims.get("target_profile"),
label="target_profile",
max_chars=MAX_ACTOR_ID_CHARS,
),
_validate_identifier(
claims.get("authority_gateway_id"),
label="authority_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
),
int(claims.get("authority_epoch") or 0),
)
if values[4] < 1:
raise HostedRoomError("authority_epoch must be positive")
with _transaction(db_path, immediate=True) as conn:
conn.execute(
"DELETE FROM hosted_room_peer_reservations WHERE expires_at<=?",
(timestamp,),
)
authority_rows = conn.execute(
"""SELECT authority_gateway_id, authority_epoch
FROM hosted_room_peer_reservations
WHERE room_id=? AND target_profile=?
AND expires_at>? AND revoked_at IS NULL""",
(values[0], values[2], timestamp),
).fetchall()
if any(
int(row["authority_epoch"]) > values[4]
or (
int(row["authority_epoch"]) == values[4]
and str(row["authority_gateway_id"]) != values[3]
)
for row in authority_rows
):
raise AuthorityConflictError("peer room reservation authority changed")
conn.execute(
"""UPDATE hosted_room_peer_reservations
SET revoked_at=?, updated_at=?
WHERE room_id=? AND target_profile=?
AND authority_epoch<? AND revoked_at IS NULL""",
(timestamp, timestamp, values[0], values[2], values[4]),
)
existing = conn.execute(
"""SELECT authority_gateway_id, authority_epoch
FROM hosted_room_peer_reservations
WHERE room_id=? AND member_id=? AND target_profile=?""",
values[:3],
).fetchone()
if existing is not None and (
int(existing["authority_epoch"]) > values[4]
or (
int(existing["authority_epoch"]) == values[4]
and str(existing["authority_gateway_id"]) != values[3]
)
):
raise AuthorityConflictError("peer room reservation authority changed")
conn.execute(
"""INSERT INTO hosted_room_peer_reservations(
room_id, member_id, target_profile, authority_gateway_id,
authority_epoch, expires_at, revoked_at, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?)
ON CONFLICT(room_id, member_id, target_profile) DO UPDATE SET
authority_gateway_id=excluded.authority_gateway_id,
authority_epoch=excluded.authority_epoch,
expires_at=MAX(hosted_room_peer_reservations.expires_at,
excluded.expires_at),
revoked_at=NULL,
updated_at=excluded.updated_at""",
(*values, expiry, timestamp, timestamp),
)
def peer_room_is_reserved(
db_path: Path | str,
*,
room_id: str,
target_profile: str,
now: float | None = None,
) -> bool:
"""Return whether a live target-side RoomLink reservation fences Desktop."""
timestamp = float(now if now is not None else time.time())
with _transaction(db_path) as conn:
row = conn.execute(
"""SELECT 1 FROM hosted_room_peer_reservations
WHERE room_id=? AND target_profile=?
AND expires_at>? AND revoked_at IS NULL
LIMIT 1""",
(
_validate_identifier(
room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS
),
_validate_identifier(
target_profile,
label="target_profile",
max_chars=MAX_ACTOR_ID_CHARS,
),
timestamp,
),
).fetchone()
return row is not None
def peer_room_grant_is_current(
db_path: Path | str,
*,
claims: Mapping[str, Any],
now: float | None = None,
) -> bool:
"""Require a grant to match the target's current live reservation."""
timestamp = float(now if now is not None else time.time())
room_id = _validate_identifier(
claims.get("room_id"), label="room_id", max_chars=MAX_ROOM_ID_CHARS
)
member_id = _validate_identifier(
claims.get("member_id"), label="member_id", max_chars=MAX_ACTOR_ID_CHARS
)
target_profile = _validate_identifier(
claims.get("target_profile"),
label="target_profile",
max_chars=MAX_ACTOR_ID_CHARS,
)
authority_gateway_id = _validate_identifier(
claims.get("authority_gateway_id"),
label="authority_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
authority_epoch = int(claims.get("authority_epoch") or 0)
if authority_epoch < 1:
raise HostedRoomError("authority_epoch must be positive")
with _transaction(db_path) as conn:
row = conn.execute(
"""SELECT 1 FROM hosted_room_peer_reservations
WHERE room_id=? AND member_id=? AND target_profile=?
AND authority_gateway_id=? AND authority_epoch=?
AND expires_at>? AND revoked_at IS NULL
LIMIT 1""",
(
room_id,
member_id,
target_profile,
authority_gateway_id,
authority_epoch,
timestamp,
),
).fetchone()
return row is not None
def room_grant_is_revoked(
db_path: Path | str,
*,
claims: Mapping[str, Any],
now: float | None = None,
) -> bool:
"""Return whether a grant predates its exact scope's revocation fence."""
timestamp = float(now if now is not None else time.time())
scope_key = _room_grant_scope_key(claims)
issued_at = float(claims.get("issued_at") or 0)
with _transaction(db_path) as conn:
row = conn.execute(
"""SELECT revoked_before FROM hosted_room_revoked_grants
WHERE scope_key=? AND expires_at>?""",
(scope_key, timestamp),
).fetchone()
return row is not None and issued_at <= float(row["revoked_before"])
def _remote_run_identity(record: Mapping[str, Any]) -> tuple[Any, ...]:
return tuple(record[column] for column in _REMOTE_RUN_IDENTITY_COLUMNS)
def upsert_remote_run_receipt(
db_path: Path | str,
*,
record: Mapping[str, Any],
now: float | None = None,
) -> None:
"""Durably bind one logical peer task attempt to its remote run handle."""
timestamp = float(now if now is not None else time.time())
identity = _remote_run_identity(record)
with _transaction(db_path, immediate=True) as conn:
existing = conn.execute(
"""SELECT * FROM hosted_room_remote_runs
WHERE room_id=? AND home_install_id=?
AND authority_gateway_id=? AND authority_epoch=?
AND member_id=? AND target_install_id=?
AND target_profile=? AND task_id=?
AND execution_generation=?""",
identity,
).fetchone()
immutable = (*identity, record["run_id"], record["session_id"])
if existing is not None:
stored = (*_remote_run_identity(existing), existing["run_id"], existing["session_id"])
if stored != immutable:
raise HostedRoomError(
"remote run receipt conflicts with its logical task"
)
conn.execute(
"""UPDATE hosted_room_remote_runs SET updated_at=?
WHERE room_id=? AND home_install_id=?
AND authority_gateway_id=? AND authority_epoch=?
AND member_id=? AND target_install_id=?
AND target_profile=? AND task_id=?
AND execution_generation=?""",
(timestamp, *identity),
)
return
conn.execute(
"""INSERT INTO hosted_room_remote_runs(
room_id, home_install_id, authority_gateway_id,
authority_epoch, member_id, target_install_id,
target_profile, task_id, execution_generation, run_id,
session_id, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
*immutable,
timestamp,
timestamp,
),
)
def list_remote_run_receipts(
db_path: Path | str,
*,
room_id: str | None = None,
target_profile: str | None = None,
session_id: str | None = None,
) -> list[dict[str, Any]]:
"""Return remote run handles in durable task order."""
conditions: list[str] = []
values: list[Any] = []
for column, value in (
("room_id", room_id),
("target_profile", target_profile),
("session_id", session_id),
):
if value is not None:
conditions.append(f"{column}=?")
values.append(value)
where = f" WHERE {' AND '.join(conditions)}" if conditions else ""
with _transaction(db_path) as conn:
rows = conn.execute(
"SELECT * FROM hosted_room_remote_runs"
+ where
+ " ORDER BY created_at, task_id, execution_generation",
values,
).fetchall()
return [dict(row) for row in rows]
def remote_run_receipt(
db_path: Path | str,
*,
record: Mapping[str, Any],
) -> dict[str, Any] | None:
"""Return the exact durable remote run handle for one task attempt."""
identity = _remote_run_identity(record)
with _transaction(db_path) as conn:
row = conn.execute(
"""SELECT * FROM hosted_room_remote_runs
WHERE room_id=? AND home_install_id=?
AND authority_gateway_id=? AND authority_epoch=?
AND member_id=? AND target_install_id=?
AND target_profile=? AND task_id=?
AND execution_generation=?""",
identity,
).fetchone()
return dict(row) if row is not None else None
def _connect(db_path: Path | str) -> sqlite3.Connection:
from hermes_state import apply_wal_with_fallback
path = Path(db_path)
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(path, timeout=10)
conn.row_factory = sqlite3.Row
try:
for attempt in range(_JOURNAL_MODE_LOCK_RETRIES):
try:
apply_wal_with_fallback(conn, db_label="state.db (hosted_rooms)")
break
except sqlite3.OperationalError as exc:
if (
str(exc).lower() != "database is locked"
or attempt + 1 == _JOURNAL_MODE_LOCK_RETRIES
):
raise
# SQLite's journal-mode pragma may not honor the connection's
# busy timeout while another first opener initializes the DB,
# especially on Windows. Retry only that transient lock class.
time.sleep(0.01 * (2**attempt))
conn.execute("PRAGMA foreign_keys=ON")
if _schema_is_current(conn):
return conn
# Multiple profile gateways share this root database. Serialize every
# draft-schema transition in SQLite itself so a crash rolls back the
# whole DDL/data migration and another process can safely retry it.
conn.execute("BEGIN IMMEDIATE")
_initialize_schema(conn)
conn.commit()
except Exception:
conn.rollback()
conn.close()
raise
return conn
def _read_connection(db_path: Path | str) -> sqlite3.Connection:
"""Open the room store without steady-state journal or migration writes."""
path = Path(db_path)
if not path.is_file():
initialized = _connect(path)
initialized.close()
conn = sqlite3.connect(path, timeout=10)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
if _schema_is_current(conn):
return conn
conn.close()
migrated = _connect(path)
migrated.close()
conn = sqlite3.connect(path, timeout=10)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
return conn
@contextmanager
def _transaction(
db_path: Path | str, *, immediate: bool = False
) -> Iterator[sqlite3.Connection]:
conn = _connect(db_path)
try:
if immediate:
conn.execute("BEGIN IMMEDIATE")
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def _raise_room_not_found(conn: sqlite3.Connection, room_id: str) -> NoReturn:
retained = conn.execute(
"SELECT 1 FROM hosted_rooms WHERE room_id=?",
(room_id,),
).fetchone()
if retained is not None:
# A retained disband tombstone still has replayable history. The
# caller simply did not opt into reading disbanded rooms.
raise RoomNotFoundError("hosted room not found")
retired = conn.execute(
"SELECT 1 FROM hosted_room_retired_ids WHERE room_id=?",
(room_id,),
).fetchone()
if retired is not None:
raise RoomHistoryExpiredError(
"Group Chat history expired; room_id remains permanently retired"
)
raise RoomNotFoundError("hosted room not found")
def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
return (
conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
(table,),
).fetchone()
is not None
)
def _room_from_row(row: sqlite3.Row, *, idempotent: bool = False) -> dict[str, Any]:
room = {
"room_id": row["room_id"],
"name": row["name"],
"members": json.loads(row["members_json"]),
"authority_gateway_id": row["authority_gateway_id"],
"authority_epoch": int(row["authority_epoch"]),
"revision": int(row["revision"]),
"created_at": float(row["created_at"]),
"updated_at": float(row["updated_at"]),
"idempotent": idempotent,
}
if "disbanded_at" in row.keys() and row["disbanded_at"] is not None:
room["disbanded_at"] = float(row["disbanded_at"])
if "next_seq" in row.keys():
room["latest_seq"] = int(row["next_seq"]) - 1
return room
def _event_storage_bytes(
*, event_id: str, kind: str, actor_json: str, payload_json: str
) -> int:
return len((event_id + kind + actor_json + payload_json).encode("utf-8"))
def _assert_event_capacity(
conn: sqlite3.Connection,
*,
room: sqlite3.Row,
additional_bytes: int,
allow_control: bool = False,
) -> None:
event_limit = MAX_EVENTS_PER_ROOM + (
CONTROL_EVENT_COUNT_RESERVE if allow_control else 0
)
room_byte_limit = MAX_ROOM_EVENT_BYTES + (
CONTROL_EVENT_BYTE_RESERVE if allow_control else 0
)
gateway_byte_limit = MAX_GATEWAY_EVENT_BYTES + (
CONTROL_EVENT_BYTE_RESERVE if allow_control else 0
)
if int(room["next_seq"]) - 1 >= event_limit:
raise HostedRoomError(
"This Group Chat reached its history limit. Start a new Group Chat to continue."
)
room_bytes = int(room["event_bytes"])
if room_bytes + additional_bytes > room_byte_limit:
raise HostedRoomError(
"This Group Chat reached its storage limit. Start a new Group Chat to continue."
)
gateway_bytes = int(
conn.execute(
"SELECT COALESCE(SUM(event_bytes), 0) FROM hosted_rooms"
).fetchone()[0]
)
if gateway_bytes + additional_bytes > gateway_byte_limit:
_prune_disbanded_rooms_locked(
conn,
now=None,
max_gateway_event_bytes=max(0, gateway_byte_limit - additional_bytes),
)
gateway_bytes = int(
conn.execute(
"SELECT COALESCE(SUM(event_bytes), 0) FROM hosted_rooms"
).fetchone()[0]
)
if gateway_bytes + additional_bytes > gateway_byte_limit:
raise HostedRoomError(
"Group Chat storage is full on this host. Delete an old Group Chat and try again."
)
def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
return (
conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
(table,),
).fetchone()
is not None
)
def _prune_disbanded_rooms_locked(
conn: sqlite3.Connection,
*,
now: float | None,
max_gateway_event_bytes: int | None = None,
) -> int:
candidates: set[str] = set()
if now is not None:
cutoff = now - DISBANDED_ROOM_RETENTION_SECONDS
candidates.update(
str(row["room_id"])
for row in conn.execute(
"""SELECT room_id FROM hosted_rooms
WHERE disbanded_at IS NOT NULL AND disbanded_at<=?""",
(cutoff,),
).fetchall()
)
candidates.update(
str(row["room_id"])
for row in conn.execute(
"""SELECT room_id FROM hosted_rooms
WHERE disbanded_at IS NOT NULL
ORDER BY disbanded_at DESC, room_id ASC
LIMIT -1 OFFSET ?""",
(MAX_DISBANDED_ROOM_TOMBSTONES,),
).fetchall()
)
if max_gateway_event_bytes is not None:
retained_bytes = int(
conn.execute(
"SELECT COALESCE(SUM(event_bytes), 0) FROM hosted_rooms"
).fetchone()[0]
)
if retained_bytes > max_gateway_event_bytes:
for row in conn.execute(
"""SELECT room_id, event_bytes FROM hosted_rooms
WHERE disbanded_at IS NOT NULL
ORDER BY disbanded_at ASC, room_id ASC"""
).fetchall():
room_id = str(row["room_id"])
if room_id not in candidates:
candidates.add(room_id)
retained_bytes -= int(row["event_bytes"])
if retained_bytes <= max_gateway_event_bytes:
break
if not candidates:
return 0
placeholders = ",".join("?" for _ in candidates)
room_ids = tuple(sorted(candidates))
conn.execute(
f"""INSERT OR IGNORE INTO hosted_room_retired_ids (room_id, retired_at)
SELECT room_id, disbanded_at FROM hosted_rooms
WHERE room_id IN ({placeholders}) AND disbanded_at IS NOT NULL""",
room_ids,
)
dependent_tables = (
"hosted_room_policy_transcript_state",
"hosted_room_policy_transcript",
"hosted_room_policy_publications",
"hosted_room_policy_watermarks",
"hosted_room_policy_events",
"hosted_room_policy_threads",
"hosted_room_policy_cursors",
"hosted_room_driver_tasks",
"hosted_room_driver_leases",
"hosted_room_remote_runs",
"hosted_room_links",
"hosted_room_peer_reservations",
"hosted_room_events",
)
for table in dependent_tables:
if _table_exists(conn, table):
conn.execute(
f"DELETE FROM {table} WHERE room_id IN ({placeholders})",
room_ids,
)
conn.execute(
f"DELETE FROM hosted_rooms WHERE room_id IN ({placeholders})",
room_ids,
)
return len(room_ids)
def prune_disbanded_rooms(
db_path: Path | str,
*,
now: float | None = None,
) -> int:
"""Purge deleted Group Chat payloads while reserving their identities."""
timestamp = time.time() if now is None else float(now)
with _transaction(db_path, immediate=True) as conn:
return _prune_disbanded_rooms_locked(conn, now=timestamp)
def _event_from_row(row: sqlite3.Row, *, idempotent: bool = False) -> dict[str, Any]:
return {
"room_id": row["room_id"],
"seq": int(row["seq"]),
"event_id": row["event_id"],
"kind": row["kind"],
"actor": json.loads(row["actor_json"]),
"authority_epoch": (
int(row["authority_epoch"]) if row["authority_epoch"] is not None else None
),
"payload": json.loads(row["payload_json"]),
"created_at": float(row["created_at"]),
"idempotent": idempotent,
}
def create_room(
db_path: Path | str,
*,
room_id: Any,
name: Any,
members: Any,
authority_gateway_id: Any,
now: float | None = None,
) -> dict[str, Any]:
"""Create a room, or return the identical existing room idempotently."""
room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
name = _validate_room_name(name)
normalized_members, members_json = _validate_members(members)
authority_gateway_id = _validate_identifier(
authority_gateway_id,
label="authority_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
now = time.time() if now is None else float(now)
with _transaction(db_path, immediate=True) as conn:
if conn.execute(
"SELECT 1 FROM hosted_room_retired_ids WHERE room_id=?",
(room_id,),
).fetchone():
raise RoomConflictError("room_id belongs to a disbanded room")
existing = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, event_bytes, revision,
created_at, updated_at, disbanded_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
if existing is not None:
if existing["disbanded_at"] is not None:
raise RoomConflictError("room_id belongs to a disbanded room")
legacy_adoption = (
existing["authority_gateway_id"] == "legacy"
and authority_gateway_id != "legacy"
)
members_match = existing["members_json"] == members_json or (
legacy_adoption
and _legacy_members_match(existing["members_json"], normalized_members)
)
if existing["name"] != name or not members_match:
raise RoomConflictError("room_id already exists with different state")
if legacy_adoption:
target_epoch = int(existing["authority_epoch"]) + 1
seq = int(existing["next_seq"])
claim_actor_json = _canonical_json(
{"kind": "system", "id": "authority-control"},
label="actor",
max_bytes=4 * 1024,
)
claim_payload_json = _canonical_json(
{
"previous_gateway_id": "legacy",
"authority_gateway_id": authority_gateway_id,
"authority_epoch": target_epoch,
},
label="payload",
max_bytes=MAX_EVENT_JSON_BYTES,
)
claim_bytes = _event_storage_bytes(
event_id="system:authority-adopted",
kind="authority.claimed",
actor_json=claim_actor_json,
payload_json=claim_payload_json,
)
_assert_event_capacity(
conn,
room=existing,
additional_bytes=claim_bytes,
allow_control=True,
)
conn.execute(
"""INSERT INTO hosted_room_events
(room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at)
VALUES (?, ?, 'system:authority-adopted',
'authority.claimed', ?, ?, ?, ?)""",
(
room_id,
seq,
claim_actor_json,
target_epoch,
claim_payload_json,
now,
),
)
adopted = conn.execute(
"""UPDATE hosted_rooms
SET members_json=?, authority_gateway_id=?, authority_epoch=?,
next_seq=next_seq+1, revision=revision+1,
event_bytes=event_bytes+?, updated_at=?
WHERE room_id=? AND authority_gateway_id='legacy'
AND authority_epoch=? AND next_seq=?
AND disbanded_at IS NULL""",
(
members_json,
authority_gateway_id,
target_epoch,
claim_bytes,
now,
room_id,
int(existing["authority_epoch"]),
seq,
),
)
if adopted.rowcount != 1:
raise AuthorityConflictError("legacy room adoption lost its fence")
existing = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, revision, created_at,
updated_at, disbanded_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
if existing is None: # pragma: no cover - row updated above
raise RuntimeError("adopted room could not be reloaded")
result = _room_from_row(existing, idempotent=True)
result["adopted"] = True
claim_event = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM hosted_room_events
WHERE room_id=? AND event_id='system:authority-adopted'""",
(room_id,),
).fetchone()
if claim_event is None: # pragma: no cover - inserted above
raise RuntimeError("legacy adoption event could not be reloaded")
result["claim_event"] = _event_from_row(claim_event)
return result
if existing["authority_gateway_id"] != authority_gateway_id:
raise RoomConflictError(
"room_id already belongs to a different authority"
)
return _room_from_row(existing, idempotent=True)
active_rooms = int(
conn.execute(
"SELECT COUNT(*) FROM hosted_rooms WHERE disbanded_at IS NULL"
).fetchone()[0]
)
if active_rooms >= MAX_ACTIVE_ROOMS:
raise HostedRoomError(
"This host has too many active Group Chats. Delete one and try again."
)
conn.execute(
"""INSERT INTO hosted_rooms
(room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, event_bytes, revision,
created_at, updated_at, disbanded_at)
VALUES (?, ?, ?, ?, 1, 1, 0, 1, ?, ?, NULL)""",
(room_id, name, members_json, authority_gateway_id, now, now),
)
row = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, revision, created_at, updated_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
if row is None: # pragma: no cover - guarded by the insert above
raise RuntimeError("created room could not be reloaded")
result = _room_from_row(row)
result["members"] = normalized_members
return result
def list_rooms(
db_path: Path | str,
*,
include_disbanded: bool = False,
limit: int = MAX_ROOM_LIST_LIMIT,
offset: int = 0,
) -> list[dict[str, Any]]:
"""Return one bounded read-only page ordered by most recent change."""
if (
isinstance(limit, bool)
or not isinstance(limit, int)
or not 1 <= limit <= MAX_ROOM_LIST_LIMIT
):
raise HostedRoomError(f"limit must be between 1 and {MAX_ROOM_LIST_LIMIT}")
if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
raise HostedRoomError("offset must be a non-negative integer")
conn = _read_connection(db_path)
try:
rows = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, revision, created_at, updated_at,
disbanded_at
FROM hosted_rooms
WHERE disbanded_at IS NULL OR ?
ORDER BY updated_at DESC, room_id ASC
LIMIT ? OFFSET ?""",
(int(include_disbanded), limit, offset),
).fetchall()
finally:
conn.close()
return [_room_from_row(row) for row in rows]
def rename_room(
db_path: Path | str,
*,
room_id: Any,
event_id: Any,
name: Any,
now: float | None = None,
) -> dict[str, Any]:
"""Rename a live room and append its replay event atomically."""
room_id = _validate_identifier(
room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS
)
event_id = _validate_identifier(
event_id, label="event_id", max_chars=MAX_EVENT_ID_CHARS
)
name = _validate_room_name(name)
now = time.time() if now is None else float(now)
actor_json = _canonical_json(
{"kind": "system", "id": "room-control"},
label="actor",
max_bytes=4 * 1024,
)
payload_json = _canonical_json(
{"name": name}, label="payload", max_bytes=MAX_EVENT_JSON_BYTES
)
with _transaction(db_path, immediate=True) as conn:
room = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, event_bytes, revision, created_at,
updated_at, disbanded_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
if room is None:
_raise_room_not_found(conn, room_id)
if room["disbanded_at"] is not None:
raise RoomNotFoundError("hosted room not found")
existing = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM hosted_room_events WHERE room_id=? AND event_id=?""",
(room_id, event_id),
).fetchone()
if existing is not None:
if existing["kind"] != "room.renamed" or existing["payload_json"] != payload_json:
raise EventConflictError(
"event_id already exists with different immutable content"
)
result = _room_from_row(room, idempotent=True)
result["event"] = _event_from_row(existing, idempotent=True)
return result
seq = int(room["next_seq"])
epoch = int(room["authority_epoch"])
event_bytes = _event_storage_bytes(
event_id=event_id,
kind="room.renamed",
actor_json=actor_json,
payload_json=payload_json,
)
_assert_event_capacity(conn, room=room, additional_bytes=event_bytes)
conn.execute(
"""UPDATE hosted_rooms
SET name=?, next_seq=?, event_bytes=event_bytes+?,
revision=revision+1, updated_at=?
WHERE room_id=?""",
(name, seq + 1, event_bytes, now, room_id),
)
conn.execute(
"""INSERT INTO hosted_room_events(
room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
) VALUES (?, ?, ?, 'room.renamed', ?, ?, ?, ?)""",
(room_id, seq, event_id, actor_json, epoch, payload_json, now),
)
updated = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, revision, created_at,
updated_at, disbanded_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
event = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM hosted_room_events WHERE room_id=? AND event_id=?""",
(room_id, event_id),
).fetchone()
result = _room_from_row(updated)
result["event"] = _event_from_row(event)
return result
def append_event(
db_path: Path | str,
*,
room_id: Any,
event_id: Any,
kind: Any,
actor: Any,
payload: Any,
authority_gateway_id: Any = None,
authority_epoch: Any = None,
now: float | None = None,
) -> dict[str, Any]:
"""Append one immutable event and allocate its per-room sequence atomically.
Repeating the same ``event_id`` and immutable content returns the original
event. Reusing the id for different content fails closed.
"""
room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
event_id = _validate_identifier(
event_id,
label="event_id",
max_chars=MAX_EVENT_ID_CHARS,
)
kind = _validate_event_kind(kind)
normalized_actor, actor_json = _validate_actor(actor, kind=kind)
authority_scoped = normalized_actor["kind"] in {
"user",
"member",
"gateway",
"system",
}
normalized_authority_gateway_id: str | None = None
normalized_authority_epoch: int | None = None
if authority_scoped:
normalized_authority_gateway_id = _validate_identifier(
authority_gateway_id,
label="authority_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
if (
normalized_actor["kind"] == "gateway"
and normalized_actor["id"] != normalized_authority_gateway_id
):
raise HostedRoomError("gateway actor.id must match authority_gateway_id")
if (
isinstance(authority_epoch, bool)
or not isinstance(authority_epoch, int)
or authority_epoch < 1
):
raise HostedRoomError("authority_epoch must be a positive integer")
normalized_authority_epoch = authority_epoch
elif authority_gateway_id is not None or authority_epoch is not None:
raise HostedRoomError(
"authority fields are only valid for room-scoped events"
)
if not isinstance(payload, dict):
raise HostedRoomError("payload must be an object")
payload_json = _canonical_json(
payload,
label="payload",
max_bytes=MAX_EVENT_JSON_BYTES,
)
now = time.time() if now is None else float(now)
with _transaction(db_path, immediate=True) as conn:
existing = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at
FROM hosted_room_events WHERE room_id=? AND event_id=?""",
(room_id, event_id),
).fetchone()
if existing is not None:
if (
existing["kind"] != kind
or existing["actor_json"] != actor_json
or existing["authority_epoch"] != normalized_authority_epoch
or existing["payload_json"] != payload_json
):
raise EventConflictError(
"event_id already exists with different content"
)
return _event_from_row(existing, idempotent=True)
room = conn.execute(
"""SELECT next_seq, event_bytes, authority_gateway_id, authority_epoch
FROM hosted_rooms
WHERE room_id=? AND disbanded_at IS NULL""",
(room_id,),
).fetchone()
if room is None:
_raise_room_not_found(conn, room_id)
if authority_scoped and (
room["authority_gateway_id"] != normalized_authority_gateway_id
or int(room["authority_epoch"]) != normalized_authority_epoch
):
raise AuthorityConflictError("stale hosted room authority")
seq = int(room["next_seq"])
event_bytes = _event_storage_bytes(
event_id=event_id,
kind=kind,
actor_json=actor_json,
payload_json=payload_json,
)
_assert_event_capacity(
conn,
room=room,
additional_bytes=event_bytes,
allow_control=kind
in {
"authority.claimed",
"authority.lost",
"room.disbanded",
"room.stop_requested",
},
)
conn.execute(
"""INSERT INTO hosted_room_events
(room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
room_id,
seq,
event_id,
kind,
actor_json,
normalized_authority_epoch,
payload_json,
now,
),
)
advanced = conn.execute(
"""UPDATE hosted_rooms
SET next_seq=?, event_bytes=event_bytes+?, updated_at=?
WHERE room_id=? AND next_seq=?""",
(seq + 1, event_bytes, now, room_id, seq),
)
if advanced.rowcount != 1:
raise RuntimeError("hosted room sequence advance lost its write fence")
row = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at
FROM hosted_room_events WHERE room_id=? AND seq=?""",
(room_id, seq),
).fetchone()
if row is None: # pragma: no cover - guarded by the insert above
raise RuntimeError("appended event could not be reloaded")
result = _event_from_row(row)
result["actor"] = normalized_actor
return result
def probe_hosted_room(db_path: Path | str, *, room_id: Any) -> bool:
"""Check room ownership without creating or migrating the shared store.
This runs on the synchronous prompt-admission path for older Desktop
clients, so it fails quickly under contention instead of blocking the
WebSocket reader for SQLite's normal ten-second timeout.
"""
checked_room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
path = Path(db_path)
if not path.is_file():
return False
try:
conn = sqlite3.connect(path, timeout=0.05)
try:
table = conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' "
"AND name='hosted_rooms' LIMIT 1"
).fetchone()
if table is None:
return False
return (
conn.execute(
"SELECT 1 FROM hosted_rooms WHERE room_id=? "
"AND disbanded_at IS NULL LIMIT 1",
(checked_room_id,),
).fetchone()
is not None
)
finally:
conn.close()
except sqlite3.Error as exc:
raise RoomProbeUnavailableError(
"hosted room ownership is temporarily unavailable"
) from exc
def probe_peer_room_reservation(
db_path: Path | str,
*,
room_id: Any,
target_profile: Any,
now: float | None = None,
) -> bool:
"""Check a peer reservation without creating or migrating shared state."""
checked_room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
checked_profile = _validate_identifier(
target_profile,
label="target_profile",
max_chars=MAX_ACTOR_ID_CHARS,
)
path = Path(db_path)
if not path.is_file():
return False
checked_now = float(now if now is not None else time.time())
try:
conn = sqlite3.connect(path, timeout=0.05)
try:
table = conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' "
"AND name='hosted_room_peer_reservations' LIMIT 1"
).fetchone()
if table is None:
return False
return (
conn.execute(
"""SELECT 1 FROM hosted_room_peer_reservations
WHERE room_id=? AND target_profile=?
AND expires_at>? AND revoked_at IS NULL
LIMIT 1""",
(checked_room_id, checked_profile, checked_now),
).fetchone()
is not None
)
finally:
conn.close()
except sqlite3.Error as exc:
raise RoomProbeUnavailableError(
"peer room ownership is temporarily unavailable"
) from exc
def room_state(
db_path: Path | str,
*,
room_id: Any,
include_disbanded: bool = False,
) -> dict[str, Any]:
"""Return durable replay and authority state for one room."""
room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
with _transaction(db_path) as conn:
row = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, revision, created_at, updated_at,
disbanded_at
FROM hosted_rooms
WHERE room_id=? AND (disbanded_at IS NULL OR ?)""",
(room_id, int(include_disbanded)),
).fetchone()
if row is None:
_raise_room_not_found(conn, room_id)
claim_row = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at
FROM hosted_room_events
WHERE room_id=? AND kind='authority.claimed'
AND authority_epoch=?
ORDER BY seq DESC LIMIT 1""",
(room_id, int(row["authority_epoch"])),
).fetchone()
state = _room_from_row(row)
state["latest_seq"] = int(row["next_seq"]) - 1
if claim_row is not None:
state["authority_claim"] = _event_from_row(claim_row)
return state
def request_room_stop(
db_path: Path | str,
*,
room_id: Any,
cancel_id: Any,
expected_gateway_id: Any,
expected_epoch: Any,
) -> dict[str, Any]:
"""Append an idempotent fence that supersedes earlier user turns."""
cancel_id = _validate_identifier(
cancel_id,
label="cancel_id",
max_chars=MAX_EVENT_ID_CHARS,
)
digest = hashlib.sha256(cancel_id.encode()).hexdigest()[:32]
return append_event(
db_path,
room_id=room_id,
event_id=f"room-stop:{digest}",
kind="room.stop_requested",
actor={"kind": "gateway", "id": expected_gateway_id},
payload={"cancel_id": cancel_id},
authority_gateway_id=expected_gateway_id,
authority_epoch=expected_epoch,
)
def claim_authority(
db_path: Path | str,
*,
room_id: Any,
expected_gateway_id: Any,
expected_epoch: Any,
new_gateway_id: Any,
event_id: Any,
now: float | None = None,
) -> dict[str, Any]:
"""Fence a verified authority transfer with a compare-and-swap epoch.
This storage primitive does not decide *when* takeover is safe. A future
replicated driver must call it only after its lease/quorum policy has
established that the previous owner can no longer commit.
"""
room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
expected_gateway_id = _validate_identifier(
expected_gateway_id,
label="expected_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
new_gateway_id = _validate_identifier(
new_gateway_id,
label="new_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
event_id = _validate_identifier(
event_id,
label="event_id",
max_chars=MAX_EVENT_ID_CHARS,
)
if (
isinstance(expected_epoch, bool)
or not isinstance(expected_epoch, int)
or expected_epoch < 1
):
raise HostedRoomError("expected_epoch must be a positive integer")
now = time.time() if now is None else float(now)
target_epoch = expected_epoch + 1
claim_actor = {"kind": "system", "id": "authority-control"}
claim_actor_json = _canonical_json(
claim_actor,
label="actor",
max_bytes=4 * 1024,
)
claim_payload = {
"previous_gateway_id": expected_gateway_id,
"authority_gateway_id": new_gateway_id,
"authority_epoch": target_epoch,
}
claim_payload_json = _canonical_json(
claim_payload,
label="payload",
max_bytes=MAX_EVENT_JSON_BYTES,
)
with _transaction(db_path, immediate=True) as conn:
row = conn.execute(
"""SELECT authority_gateway_id, authority_epoch, next_seq, event_bytes
FROM hosted_rooms
WHERE room_id=? AND disbanded_at IS NULL""",
(room_id,),
).fetchone()
if row is None:
_raise_room_not_found(conn, room_id)
current_gateway = str(row["authority_gateway_id"])
current_epoch = int(row["authority_epoch"])
existing_event = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at
FROM hosted_room_events WHERE room_id=? AND event_id=?""",
(room_id, event_id),
).fetchone()
if existing_event is not None:
if (
existing_event["kind"] != "authority.claimed"
or existing_event["actor_json"] != claim_actor_json
or existing_event["authority_epoch"] != target_epoch
or existing_event["payload_json"] != claim_payload_json
):
raise EventConflictError(
"event_id already exists with different content"
)
if current_gateway != new_gateway_id or current_epoch != target_epoch:
raise AuthoritySupersededError(
"authority claim succeeded but was later superseded"
)
idempotent = True
elif current_gateway != expected_gateway_id or current_epoch != expected_epoch:
raise AuthorityConflictError("hosted room authority changed")
else:
seq = int(row["next_seq"])
claim_bytes = _event_storage_bytes(
event_id=event_id,
kind="authority.claimed",
actor_json=claim_actor_json,
payload_json=claim_payload_json,
)
_assert_event_capacity(
conn,
room=row,
additional_bytes=claim_bytes,
allow_control=True,
)
conn.execute(
"""INSERT INTO hosted_room_events
(room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at)
VALUES (?, ?, ?, 'authority.claimed', ?, ?, ?, ?)""",
(
room_id,
seq,
event_id,
claim_actor_json,
target_epoch,
claim_payload_json,
now,
),
)
updated = conn.execute(
"""UPDATE hosted_rooms
SET authority_gateway_id=?, authority_epoch=authority_epoch+1,
next_seq=next_seq+1, event_bytes=event_bytes+?,
revision=revision+1, updated_at=?
WHERE room_id=? AND disbanded_at IS NULL
AND authority_gateway_id=? AND authority_epoch=?""",
(
new_gateway_id,
claim_bytes,
now,
room_id,
expected_gateway_id,
expected_epoch,
),
)
if updated.rowcount != 1:
raise AuthorityConflictError("hosted room authority changed")
idempotent = False
existing_event = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM hosted_room_events WHERE room_id=? AND event_id=?""",
(room_id, event_id),
).fetchone()
state_row = conn.execute(
"""SELECT room_id, name, members_json, authority_gateway_id,
authority_epoch, next_seq, revision, created_at, updated_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
if state_row is None: # pragma: no cover - room exists in this transaction
raise RuntimeError("claimed room could not be reloaded")
state = _room_from_row(state_row, idempotent=idempotent)
state["latest_seq"] = int(state_row["next_seq"]) - 1
if existing_event is None: # pragma: no cover - both claim paths set it
raise RuntimeError("authority claim event could not be reloaded")
state["claim_event"] = _event_from_row(
existing_event,
idempotent=idempotent,
)
return state
def disband_room(
db_path: Path | str,
*,
room_id: Any,
expected_gateway_id: Any,
expected_epoch: Any,
now: float | None = None,
) -> dict[str, Any]:
"""Tombstone a room id permanently and idempotently."""
room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
expected_gateway_id = _validate_identifier(
expected_gateway_id,
label="expected_gateway_id",
max_chars=MAX_ACTOR_ID_CHARS,
)
if (
isinstance(expected_epoch, bool)
or not isinstance(expected_epoch, int)
or expected_epoch < 1
):
raise HostedRoomError("expected_epoch must be a positive integer")
now = time.time() if now is None else float(now)
with _transaction(db_path, immediate=True) as conn:
room = conn.execute(
"""SELECT authority_gateway_id, authority_epoch, next_seq,
event_bytes, disbanded_at
FROM hosted_rooms WHERE room_id=?""",
(room_id,),
).fetchone()
if room is None:
retired = conn.execute(
"SELECT retired_at FROM hosted_room_retired_ids WHERE room_id=?",
(room_id,),
).fetchone()
if retired is None:
raise RoomNotFoundError("hosted room not found")
return {
"room_id": room_id,
"disbanded_at": float(retired["retired_at"]),
"idempotent": True,
"history_expired": True,
}
if room["disbanded_at"] is not None:
conn.execute(
"""INSERT OR IGNORE INTO hosted_room_retired_ids
(room_id, retired_at) VALUES (?, ?)""",
(room_id, float(room["disbanded_at"])),
)
event = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM hosted_room_events
WHERE room_id=? AND event_id='system:room-disbanded'""",
(room_id,),
).fetchone()
return {
"room_id": room_id,
"disbanded_at": float(room["disbanded_at"]),
"idempotent": True,
**(
{"event": _event_from_row(event, idempotent=True)}
if event is not None
else {}
),
}
if (
str(room["authority_gateway_id"]) != expected_gateway_id
or int(room["authority_epoch"]) != expected_epoch
):
raise AuthorityConflictError("stale hosted room authority")
seq = int(room["next_seq"])
actor_json = _canonical_json(
{"kind": "system", "id": "room-control"},
label="actor",
max_bytes=4 * 1024,
)
payload_json = _canonical_json(
{"room_id": room_id},
label="payload",
max_bytes=MAX_EVENT_JSON_BYTES,
)
disband_bytes = _event_storage_bytes(
event_id="system:room-disbanded",
kind="room.disbanded",
actor_json=actor_json,
payload_json=payload_json,
)
_assert_event_capacity(
conn,
room=room,
additional_bytes=disband_bytes,
allow_control=True,
)
conn.execute(
"""INSERT INTO hosted_room_events
(room_id, seq, event_id, kind, actor_json, authority_epoch,
payload_json, created_at)
VALUES (?, ?, 'system:room-disbanded', 'room.disbanded', ?, ?, ?, ?)""",
(
room_id,
seq,
actor_json,
int(room["authority_epoch"]),
payload_json,
now,
),
)
updated = conn.execute(
"""UPDATE hosted_rooms
SET disbanded_at=?, updated_at=?, revision=revision+1,
next_seq=next_seq+1, event_bytes=event_bytes+?
WHERE room_id=? AND disbanded_at IS NULL
AND authority_gateway_id=? AND authority_epoch=?""",
(
now,
now,
disband_bytes,
room_id,
expected_gateway_id,
expected_epoch,
),
)
if updated.rowcount != 1:
raise RoomConflictError("hosted room disband lost its fence")
conn.execute(
"""INSERT OR IGNORE INTO hosted_room_retired_ids
(room_id, retired_at) VALUES (?, ?)""",
(room_id, now),
)
event = conn.execute(
"""SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM hosted_room_events
WHERE room_id=? AND event_id='system:room-disbanded'""",
(room_id,),
).fetchone()
if event is None: # pragma: no cover - inserted in this transaction
raise RuntimeError("room disband event could not be reloaded")
_prune_disbanded_rooms_locked(
conn,
now=now,
max_gateway_event_bytes=MAX_GATEWAY_EVENT_BYTES,
)
return {
"room_id": room_id,
"disbanded_at": now,
"idempotent": False,
"event": _event_from_row(event),
}
def read_events(
db_path: Path | str,
*,
room_id: Any,
since_seq: Any = 0,
limit: Any = 100,
include_disbanded: bool = False,
) -> dict[str, Any]:
"""Read a monotonic room-log delta after ``since_seq``."""
room_id = _validate_identifier(
room_id,
label="room_id",
max_chars=MAX_ROOM_ID_CHARS,
)
if isinstance(since_seq, bool) or not isinstance(since_seq, int) or since_seq < 0:
raise HostedRoomError("since_seq must be a non-negative integer")
if (
isinstance(limit, bool)
or not isinstance(limit, int)
or not 1 <= limit <= MAX_LOG_LIMIT
):
raise HostedRoomError(f"limit must be between 1 and {MAX_LOG_LIMIT}")
with _transaction(db_path) as conn:
room = conn.execute(
"""SELECT next_seq, authority_gateway_id, authority_epoch
FROM hosted_rooms
WHERE room_id=? AND (disbanded_at IS NULL OR ?)""",
(room_id, int(include_disbanded)),
).fetchone()
if room is None:
_raise_room_not_found(conn, room_id)
latest_seq = int(room["next_seq"]) - 1
authority_gateway = str(room["authority_gateway_id"])
authority_epoch = int(room["authority_epoch"])
if since_seq > latest_seq:
raise HostedRoomError("since_seq is ahead of the hosted room log")
rows = conn.execute(
"""WITH candidates AS (
SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at,
SUM(
LENGTH(CAST(event_id AS BLOB)) +
LENGTH(CAST(kind AS BLOB)) +
LENGTH(CAST(actor_json AS BLOB)) +
LENGTH(CAST(payload_json AS BLOB))
) OVER (ORDER BY seq ASC) AS cumulative_bytes
FROM hosted_room_events
WHERE room_id=? AND seq>?
ORDER BY seq ASC LIMIT ?
)
SELECT room_id, seq, event_id, kind, actor_json,
authority_epoch, payload_json, created_at
FROM candidates
WHERE cumulative_bytes<=?
ORDER BY seq ASC""",
(room_id, since_seq, limit, MAX_LOG_PAGE_BYTES),
).fetchall()
events = [_event_from_row(row) for row in rows]
def build_page(page_events: list[dict[str, Any]]) -> dict[str, Any]:
cursor = page_events[-1]["seq"] if page_events else since_seq
return {
"events": page_events,
"cursor": cursor,
"latest_seq": latest_seq,
"has_more": cursor < latest_seq,
"authority": {
"gateway_id": authority_gateway,
"epoch": authority_epoch,
},
}
def page_bytes(page: dict[str, Any]) -> int:
return len(
json.dumps(
page,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
)
page = build_page(events)
if events and page_bytes(page) > MAX_LOG_PAGE_BYTES:
low, high = 1, len(events)
while low < high:
middle = (low + high + 1) // 2
candidate = build_page(events[:middle])
if page_bytes(candidate) <= MAX_LOG_PAGE_BYTES:
low = middle
else:
high = middle - 1
page = build_page(events[:low])
if page_bytes(page) > MAX_LOG_PAGE_BYTES:
raise HostedRoomError("hosted room event exceeds replay page limit")
return page
Reference in New Issue View Git Blame Copy Permalink