app/registry.py

"""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")