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