climperor/web/cloudflare/_enable_r2.py

"""Create R2 bucket, bind Pages, redeploy Worker using keyzoo-injected CF creds."""

from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[2]
WORKER = Path(__file__).resolve().parent / "player-sync"
ACCT = "510534f7f6284344aadaf2f5a0794d48"
BUCKET = "climperor-player-data"
PAGES = "climperor-relations"
QUEUE = "climperor-player-sync"
D1_ID = "9eeb24ba-acc5-4520-b4e7-754ea776394e"
API = "https://api.cloudflare.com/client/v4"
RESOURCES = Path(__file__).resolve().parent / ".resources.json"


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 CF credentials")
    return email, key


def api(method: str, path: str, body: dict | None = None, retries: int = 6) -> dict:
    email, key = creds()
    data = None if body is None else json.dumps(body).encode("utf-8")
    last = None
    for i in range(retries):
        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-enable-r2",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=60) as resp:
                return json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            raw = e.read().decode("utf-8", errors="replace")
            last = f"{e.code}: {raw[:200]}"
            if e.code in (521, 522, 523, 524, 525, 429, 503) and i + 1 < retries:
                wait = 4 + i * 3
                print(f"retry {i+1}/{retries} after {e.code}, sleep {wait}s", flush=True)
                time.sleep(wait)
                continue
            raise SystemExit(f"CF API {method} {path} -> {last}") from e
        except Exception as e:
            last = str(e)
            if i + 1 < retries:
                wait = 4 + i * 3
                print(f"retry {i+1}/{retries} after {type(e).__name__}, sleep {wait}s", flush=True)
                time.sleep(wait)
                continue
            raise SystemExit(f"CF API {method} {path} -> {last}") from e
    raise SystemExit(f"CF API failed: {last}")


def ensure_r2() -> None:
    try:
        api("GET", f"/accounts/{ACCT}/r2/buckets/{BUCKET}", retries=3)
        print(f"r2 exists: {BUCKET}", flush=True)
        return
    except SystemExit:
        pass
    payload = api("POST", f"/accounts/{ACCT}/r2/buckets", {"name": BUCKET})
    if not payload.get("success"):
        err = str(payload.get("errors") or "")
        if "already exists" in err.lower() or "10004" in err:
            print(f"r2 exists: {BUCKET}", flush=True)
            return
        raise SystemExit(f"r2 create failed: {payload.get('errors')}")
    print(f"r2 created: {BUCKET}", flush=True)


def bind_pages() -> None:
    path = f"/accounts/{ACCT}/pages/projects/{PAGES}"
    project = api("GET", path)["result"]
    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

    def one(base: dict) -> dict:
        out = {
            k: v
            for k, v in base.items()
            if k not in ("d1_databases", "queue_producers", "r2_buckets", "fail_open")
        }
        d1 = dict(base.get("d1_databases") or {})
        d1["DB"] = {"id": D1_ID}
        out["d1_databases"] = d1
        producers = dict(base.get("queue_producers") or {})
        producers["SYNC_QUEUE"] = {"name": QUEUE}
        out["queue_producers"] = producers
        buckets = dict(base.get("r2_buckets") or {})
        buckets["MATCHES"] = {"name": BUCKET}
        out["r2_buckets"] = buckets
        out["fail_open"] = bool(fail_open)
        return out

    api(
        "PATCH",
        path,
        {
            "deployment_configs": {
                "production": one(prod),
                "preview": one(preview),
            }
        },
    )
    print("pages bindings: DB + SYNC_QUEUE + MATCHES", flush=True)


def update_resources() -> None:
    data = {}
    if RESOURCES.is_file():
        data = json.loads(RESOURCES.read_text(encoding="utf-8"))
    data["account_id"] = ACCT
    data["r2"] = {"name": BUCKET, "ready": True}
    data.setdefault("d1", {"name": "climperor-users", "id": D1_ID})
    data.setdefault("queue", {"name": QUEUE})
    data.setdefault("pages_project", PAGES)
    data.setdefault("worker", "climperor-player-sync")
    RESOURCES.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    print(f"updated {RESOURCES}", flush=True)


def deploy_worker() -> None:
    email, key = creds()
    env = os.environ.copy()
    env["CLOUDFLARE_EMAIL"] = email
    env["CLOUDFLARE_API_KEY"] = key
    env["CLOUDFLARE_ACCOUNT_ID"] = ACCT
    print("deploying worker…", flush=True)
    proc = subprocess.run(
        "npx --yes wrangler@3 deploy",
        cwd=str(WORKER),
        env=env,
        shell=True,
    )
    if proc.returncode != 0:
        raise SystemExit(f"worker deploy failed: {proc.returncode}")
    print("worker deployed", flush=True)


def main() -> int:
    ensure_r2()
    bind_pages()
    update_resources()
    deploy_worker()
    return 0


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