climperor/web/cloudflare/provision.py

"""Create Cloudflare D1 / R2 / Queues for Climperor player data and bind Pages.

Credentials: CLOUDFLARE_EMAIL + CLOUDFLARE_API_KEY (or keyzoo refining/cloudflare).
Writes web/cloudflare/.resources.json with ids (no secrets).
"""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
OUT = Path(__file__).resolve().parent / ".resources.json"
MIGRATION = Path(__file__).resolve().parent / "migrations" / "0001_init.sql"
PAGES_PROJECT = "climperor-relations"
D1_NAME = "climperor-users"
R2_NAME = "climperor-player-data"
QUEUE_NAME = "climperor-player-sync"
DLQ_NAME = "climperor-player-sync-dlq"
WORKER_NAME = "climperor-player-sync"
API = "https://api.cloudflare.com/client/v4"


def creds() -> tuple[str, str]:
    email = os.environ.get("CLOUDFLARE_EMAIL") or os.environ.get(
        "KEYZOO_ASSET_META_USERNAME"
    )
    key = os.environ.get("CLOUDFLARE_API_KEY") or os.environ.get(
        "KEYZOO_ASSET_SECRET_GLOBAL_API_KEY"
    )
    if not email or not key:
        raise SystemExit("missing CLOUDFLARE_EMAIL / CLOUDFLARE_API_KEY")
    return email, key


def api(method: str, path: str, body: dict | None = None) -> dict:
    email, key = creds()
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        API + path,
        data=data,
        method=method,
        headers={
            "X-Auth-Email": email,
            "X-Auth-Key": key,
            "Content-Type": "application/json",
            "User-Agent": "climperor-provision",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            payload = json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        raise SystemExit(f"CF API {method} {path} -> {e.code}: {raw[:400]}") from e
    if not payload.get("success"):
        raise SystemExit(f"CF API failed: {payload.get('errors')}")
    return payload


def account_id() -> str:
    rows = api("GET", "/accounts")["result"]
    if not rows:
        raise SystemExit("no Cloudflare accounts")
    return rows[0]["id"]


def ensure_d1(acct: str) -> str:
    listed = api("GET", f"/accounts/{acct}/d1/database")["result"] or []
    for row in listed:
        if row.get("name") == D1_NAME:
            print(f"d1 exists: {row['uuid']}")
            return row["uuid"]
    created = api(
        "POST",
        f"/accounts/{acct}/d1/database",
        {"name": D1_NAME},
    )["result"]
    print(f"d1 created: {created['uuid']}")
    return created["uuid"]


def run_migration(acct: str, db_id: str) -> None:
    sql = MIGRATION.read_text(encoding="utf-8")
    api(
        "POST",
        f"/accounts/{acct}/d1/database/{db_id}/query",
        {"sql": sql},
    )
    print("d1 migration applied")


def ensure_r2(acct: str) -> bool:
    """Return True if bucket exists/created. False if R2 not enabled on account."""
    try:
        api("GET", f"/accounts/{acct}/r2/buckets/{R2_NAME}")
        print(f"r2 exists: {R2_NAME}")
        return True
    except SystemExit:
        pass
    try:
        api("POST", f"/accounts/{acct}/r2/buckets", {"name": R2_NAME})
        print(f"r2 created: {R2_NAME}")
        return True
    except SystemExit as e:
        msg = str(e)
        if "already exists" in msg.lower() or "10004" in msg:
            print(f"r2 exists: {R2_NAME}")
            return True
        if "10042" in msg or "enable R2" in msg:
            print(
                "r2 skipped: enable R2 in Cloudflare Dashboard "
                "(https://dash.cloudflare.com/?to=/:account/r2), then re-run"
            )
            return False
        raise


def ensure_queue(acct: str, name: str) -> str:
    listed = api("GET", f"/accounts/{acct}/queues")["result"] or []
    # API shape may be {result: [...]} or {result: {queues: [...]}}
    rows = listed if isinstance(listed, list) else (listed.get("queues") or [])
    for row in rows:
        if row.get("queue_name") == name or row.get("name") == name:
            qid = row.get("queue_id") or row.get("id")
            print(f"queue exists: {name} ({qid})")
            return qid
    created = api("POST", f"/accounts/{acct}/queues", {"queue_name": name})["result"]
    qid = created.get("queue_id") or created.get("id")
    print(f"queue created: {name} ({qid})")
    return qid


def patch_wrangler(d1_id: str) -> None:
    path = Path(__file__).resolve().parent / "player-sync" / "wrangler.toml"
    text = path.read_text(encoding="utf-8")
    text = text.replace("REPLACE_D1_ID", d1_id)
    path.write_text(text, encoding="utf-8")
    print(f"updated {path}")


def write_resources(
    acct: str,
    d1_id: str,
    queue_id: str | None,
    dlq_id: str | None,
    *,
    r2_ok: bool,
) -> None:
    payload = {
        "account_id": acct,
        "d1": {"name": D1_NAME, "id": d1_id},
        "r2": {"name": R2_NAME, "ready": r2_ok},
        "queue": {"name": QUEUE_NAME, "id": queue_id},
        "dlq": {"name": DLQ_NAME, "id": dlq_id},
        "worker": WORKER_NAME,
        "pages_project": PAGES_PROJECT,
    }
    OUT.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    print(f"wrote {OUT}")


def ensure_queue_soft(acct: str, name: str) -> str | None:
    try:
        return ensure_queue(acct, name)
    except SystemExit as e:
        print(f"queue skipped ({name}): {e}")
        return None


def _env_bindings(
    base: dict,
    *,
    d1_id: str,
    queue_id: str | None,
    r2_ok: bool,
    fail_open: bool | None,
) -> dict:
    """Build one environment's deployment_config with required bindings."""
    out = {
        k: v
        for k, v in base.items()
        if k
        not in (
            "d1_databases",
            "queue_producers",
            "r2_buckets",
            "fail_open",
        )
    }
    d1_bindings = dict(base.get("d1_databases") or {})
    d1_bindings["DB"] = {"id": d1_id}
    out["d1_databases"] = d1_bindings
    if queue_id:
        producers = dict(base.get("queue_producers") or {})
        producers["SYNC_QUEUE"] = {"name": QUEUE_NAME}
        out["queue_producers"] = producers
    if r2_ok:
        buckets = dict(base.get("r2_buckets") or {})
        buckets["MATCHES"] = {"name": R2_NAME}
        out["r2_buckets"] = buckets
    # Cloudflare requires fail_open equal on production and preview.
    if fail_open is not None:
        out["fail_open"] = bool(fail_open)
    return out


def bind_pages(
    acct: str,
    d1_id: str,
    queue_id: str | None,
    *,
    r2_ok: bool = False,
) -> None:
    """Attach D1 (+ Queue / R2) to Pages production and preview bindings."""
    path = f"/accounts/{acct}/pages/projects/{PAGES_PROJECT}"
    try:
        project = api("GET", path)["result"]
    except SystemExit as e:
        print(f"pages bind skipped (project missing?): {e}")
        return
    dc = project.get("deployment_configs") or {}
    prod = dict(dc.get("production") or {})
    preview = dict(dc.get("preview") or {})
    fail_open = prod.get("fail_open")
    if fail_open is None:
        fail_open = preview.get("fail_open")
    if fail_open is None:
        fail_open = False
    body = {
        "deployment_configs": {
            "production": _env_bindings(
                prod,
                d1_id=d1_id,
                queue_id=queue_id,
                r2_ok=r2_ok,
                fail_open=fail_open,
            ),
            "preview": _env_bindings(
                preview,
                d1_id=d1_id,
                queue_id=queue_id,
                r2_ok=r2_ok,
                fail_open=fail_open,
            ),
        }
    }
    try:
        api("PATCH", path, body)
        print(f"pages bindings updated on {PAGES_PROJECT} (prod+preview)")
    except SystemExit as e:
        print(f"pages bind soft-fail (set manually): {e}")


def main() -> int:
    acct = account_id()
    print(f"account_id: {acct}")
    d1_id = ensure_d1(acct)
    run_migration(acct, d1_id)
    r2_ok = ensure_r2(acct)
    queue_id = ensure_queue_soft(acct, QUEUE_NAME)
    dlq_id = ensure_queue_soft(acct, DLQ_NAME)
    patch_wrangler(d1_id)
    bind_pages(acct, d1_id, queue_id, r2_ok=r2_ok)
    write_resources(acct, d1_id, queue_id, dlq_id, r2_ok=r2_ok)
    print(
        "next: deploy worker with wrangler + bind D1/R2/Queue to Pages project "
        f"{PAGES_PROJECT} (see web/cloudflare/README.md)"
    )
    return 0 if d1_id else 1


if __name__ == "__main__":
    raise SystemExit(main())