time/typy/deploy.py

"""Deploy the committed typy Worker, its KV store, quota object, and custom domain.

Uses the repository's existing Cloudflare helper and operator credentials. Run
from a clean checkout after npm run build --prefix typy has been committed.
No Node runtime, package install, workflow change, or D1 permission is needed on
the deployment machine.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import urllib.request

HERE = Path(__file__).resolve().parent
ROOT = HERE.parent
sys.path.insert(0, str(ROOT / ".github" / "time"))
import cloudflare as cf  # noqa: E402

WORKER = "time-typy"
MIGRATION = "typy-v1"
DOMAIN = "typy.yolopreneur.lol"
ASSETS = {
    "/app.js": "public/app.js",
    "/style.css": "public/style.css",
    "/favicon.svg": "public/favicon.svg",
    "/game.js": "src/game.js",
    "/words.js": "src/words.js",
}


def git(*args: str) -> str:
    return subprocess.check_output(["git", "-C", str(ROOT), *args], text=True).strip()


def checked_modules() -> dict[str, bytes]:
    """Never upload stale generated assets beside a newer source file."""
    assets_text = (HERE / "src" / "assets.js").read_text()
    prefix = "export const ASSETS = "
    if not assets_text.startswith(prefix):
        raise RuntimeError("assets.js is missing or not generated by typy/build.mjs")
    assets = json.loads(assets_text[len(prefix):].rstrip(";\n"))
    for url, path in ASSETS.items():
        if assets.get(url, {}).get("body") != (HERE / path).read_text():
            raise RuntimeError(f"Stale {url}: run the build and commit its output first")
    page_first = (HERE / "src" / "page.js").read_text().splitlines()[0]
    template = json.loads(page_first.removeprefix("const TEMPLATE = ").rstrip(";"))
    if template != (HERE / "public" / "index.html").read_text():
        raise RuntimeError("Stale page.js: run the build and commit its output first")
    modules = {path.name: path.read_bytes() for path in sorted((HERE / "src").glob("*.js"))}
    if "worker.js" not in modules:
        raise RuntimeError("src/worker.js is missing")
    if sum(map(len, modules.values())) > 2_000_000:
        raise RuntimeError("Worker source exceeds the bounded two-megabyte deployment")
    return modules


def namespace() -> str:
    for page in range(1, 21):
        rows = cf.cf("GET", f"{cf.account()}/storage/kv/namespaces?per_page=100&page={page}") or []
        for row in rows:
            if row.get("title") == WORKER:
                return row["id"]
        if len(rows) < 100:
            return cf.cf("POST", f"{cf.account()}/storage/kv/namespaces", {"title": WORKER})["id"]
    raise RuntimeError("KV namespace search exceeded its page limit; no namespace was created")


def existing_settings() -> dict:
    try:
        return cf.cf("GET", f"{cf.account()}/workers/scripts/{WORKER}/settings") or {}
    except RuntimeError as error:
        if "-> 404:" in str(error):
            return {}
        raise


def revision(modules: dict[str, bytes], settings: dict[str, str]) -> str:
    digest = hashlib.sha256()
    for name, data in sorted(modules.items()):
        digest.update(name.encode() + b"\0" + data + b"\0")
    digest.update(json.dumps(settings, sort_keys=True).encode())
    return digest.hexdigest()[:16]


def deploy(*, dry_run: bool = False) -> dict:
    modules = checked_modules()
    sha = git("rev-parse", "HEAD")
    if git("status", "--porcelain", "--", "typy"):
        raise RuntimeError("Deploy only from a clean, committed typy checkout")
    settings = {
        "SOURCE_SHA": sha,
        "TOPIC_MODEL": os.environ.get("TOPIC_MODEL", "mistral-small-3-2-24b-instruct"),
        "TOPICS_PER_VISITOR": "8",
        "TOPICS_PER_DAY": "80",
        "SAVES_PER_VISITOR": "30",
        "SAVES_PER_DAY": "2000",
    }
    settings["REV"] = revision(modules, settings)
    summary = {
        "worker": WORKER, "url": "https://" + DOMAIN + "/",
        "source_sha": sha, "rev": settings["REV"],
        "modules": {name: len(data) for name, data in modules.items()},
    }
    if dry_run:
        return {**summary, "dry_run": True}
    for name in ("CLOUDFLARE_API_TOKEN", "CLOUDFLARE_ACCOUNT_ID", "VENICE_API_KEY"):
        if not os.environ.get(name):
            raise RuntimeError(f"{name} is required; no deployment was attempted")
    previous = existing_settings()
    tag = previous.get("migration_tag", "")
    if tag not in ("", MIGRATION):
        raise RuntimeError("Unexpected Durable Object migration tag; review before changing it")
    bindings = [
        {"type": "kv_namespace", "name": "KV", "namespace_id": namespace()},
        {"type": "durable_object_namespace", "name": "QUOTA", "class_name": "QuotaGate"},
        {"type": "secret_text", "name": "VENICE_API_KEY", "text": os.environ["VENICE_API_KEY"]},
        *({"type": "plain_text", "name": key, "text": value} for key, value in settings.items()),
    ]
    metadata = {
        "main_module": "worker.js",
        "compatibility_date": "2026-09-01",
        "bindings": bindings,
        "observability": {"enabled": False},
    }
    has_quota = any(binding.get("type") == "durable_object_namespace" and binding.get("name") == "QUOTA"
                    for binding in previous.get("bindings", []))
    if not tag and not has_quota:
        metadata["migrations"] = {"new_tag": MIGRATION, "new_sqlite_classes": ["QuotaGate"]}
    raw, content_type = cf.multipart([
        ("metadata", "application/json", json.dumps(metadata).encode()),
        *((name, "application/javascript+module", data) for name, data in modules.items()),
    ])
    cf.cf("PUT", f"{cf.account()}/workers/scripts/{WORKER}", data=raw, ctype=content_type)
    cf.attach(WORKER, DOMAIN)
    return {**summary, "migration": MIGRATION}


def health() -> dict:
    request = urllib.request.Request("https://" + DOMAIN + "/api/health", headers={"Cache-Control": "no-cache"})
    with urllib.request.urlopen(request, timeout=15) as response:
        result = json.load(response)
    if result.get("ok") is not True:
        raise RuntimeError("The public health check did not pass")
    return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--health", action="store_true", help="Read the public health endpoint without deploying")
    args = parser.parse_args()
    try:
        print(json.dumps(health() if args.health else deploy(dry_run=args.dry_run), indent=2))
    except Exception as error:
        sys.exit(str(error))