feat(typy): ship one-word typing game with tempo and shareable replays

typy/deploy.py

@@ -0,0 +1,160 @@
+"""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))