diff --git a/registry.py b/registry.py new file mode 100644 index 0000000..eff6a8b --- /dev/null +++ b/registry.py @@ -0,0 +1,157 @@ +"""Durable SQLite registry for Phast app manifests.""" +from __future__ import annotations + +import os +import re +import sqlite3 +from contextlib import contextmanager +from collections.abc import Iterator +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import urlsplit + +SCHEMA_VERSION = 1 +_SLUG = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$") + +class RegistryUnavailable(RuntimeError): + pass + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + +def _validate_slug(value: str) -> str: + value = value.strip() + if not _SLUG.fullmatch(value): + raise ValueError("slug must be 1-63 lowercase letters, numbers, or hyphens") + return value + +def _validate_name(value: str) -> str: + value = value.strip() + if not value or len(value) > 100: + raise ValueError("name must be 1-100 characters") + return value + +def _validate_base_url(value: str) -> str: + value = value.strip().rstrip("/") + parsed = urlsplit(value) + if parsed.scheme != "https" or not parsed.netloc or parsed.path or parsed.query or parsed.fragment: + raise ValueError("base_url must be an HTTPS origin without a path") + return value + +def _validate_route(value: str, field: str) -> str: + value = value.strip() + if not value.startswith("/") or value.startswith("//") or "?" in value or "#" in value: + raise ValueError(f"{field} must be an absolute path without query or fragment") + return value + +def _manifest(row: sqlite3.Row) -> dict: + return { + "schema_version": SCHEMA_VERSION, + "kind": "phast.app", + "slug": row["slug"], + "name": row["name"], + "runtime": "python", + "base_url": row["base_url"], + "human": {"route": row["human_route"]}, + "mcp": {"route": row["mcp_route"]}, + "prephab": {"version": 1}, + "created_at": row["created_at"], + "updated_at": row["updated_at"], + } + +class AppRegistry: + def __init__(self, database: str | Path): + self.database = Path(database) + self.database.parent.mkdir(parents=True, exist_ok=True) + self._initialize() + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.database, timeout=5) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + return connection + + @contextmanager + def _connection(self) -> Iterator[sqlite3.Connection]: + connection = self._connect() + try: + with connection: + yield connection + finally: + connection.close() + + def _initialize(self) -> None: + with self._connection() as connection: + connection.execute("PRAGMA journal_mode = WAL") + connection.execute(""" + CREATE TABLE IF NOT EXISTS apps ( + slug TEXT PRIMARY KEY, + name TEXT NOT NULL, + base_url TEXT NOT NULL UNIQUE, + human_route TEXT NOT NULL, + mcp_route TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """) + + def list(self) -> list[dict]: + with self._connection() as connection: + rows = connection.execute("SELECT * FROM apps ORDER BY slug").fetchall() + return [_manifest(row) for row in rows] + + def get(self, slug: str) -> dict | None: + slug = _validate_slug(slug) + with self._connection() as connection: + row = connection.execute("SELECT * FROM apps WHERE slug = ?", (slug,)).fetchone() + return _manifest(row) if row else None + + def create(self, *, slug: str, name: str, base_url: str, human_route: str = "/", mcp_route: str = "/mcp") -> dict: + values = ( + _validate_slug(slug), _validate_name(name), _validate_base_url(base_url), + _validate_route(human_route, "human_route"), _validate_route(mcp_route, "mcp_route"), + ) + timestamp = _now() + try: + with self._connection() as connection: + connection.execute( + "INSERT INTO apps (slug, name, base_url, human_route, mcp_route, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + (*values, timestamp, timestamp), + ) + except sqlite3.IntegrityError as error: + raise ValueError("an app with this slug or base_url already exists") from error + return self.get(values[0]) + + def update(self, slug: str, *, name: str | None = None, base_url: str | None = None, + human_route: str | None = None, mcp_route: str | None = None) -> dict: + slug = _validate_slug(slug) + changes = { + "name": _validate_name(name) if name is not None else None, + "base_url": _validate_base_url(base_url) if base_url is not None else None, + "human_route": _validate_route(human_route, "human_route") if human_route is not None else None, + "mcp_route": _validate_route(mcp_route, "mcp_route") if mcp_route is not None else None, + } + changes = {key: value for key, value in changes.items() if value is not None} + if not changes: + raise ValueError("at least one field must be updated") + assignments = ", ".join(f"{key} = ?" for key in changes) + try: + with self._connection() as connection: + cursor = connection.execute( + f"UPDATE apps SET {assignments}, updated_at = ? WHERE slug = ?", + (*changes.values(), _now(), slug), + ) + if cursor.rowcount != 1: + raise KeyError(f"unknown app: {slug}") + except sqlite3.IntegrityError as error: + raise ValueError("an app with this base_url already exists") from error + return self.get(slug) + +def registry_from_runtime() -> AppRegistry: + raw = os.environ.get("PHAST_DATA_DIR", "").strip() + if not raw: + raise RegistryUnavailable("PHAST_DATA_DIR is not configured; durable registry is offline") + data_dir = Path(raw) + if not data_dir.is_dir() or not os.path.ismount(data_dir): + raise RegistryUnavailable(f"{data_dir} is not a mounted Fly volume; durable registry is offline") + return AppRegistry(data_dir / "registry.sqlite3")