#!/usr/bin/env python3
"""Provision ZITADEL end to end: machine key, Terraform, SOPS secrets, users.

One command replaces the manual sequence the Authentik script
(scripts/iam/provision-rbac.py) used to do in a single pass. Phases run in a
fixed order and every phase is safe to re-run:

  keys         extract the iam-admin machine key and PAT from namespace iam
  apply        create project, roles and OIDC apps through the ZITADEL API
  secrets      write generated client ids/secrets back into the SOPS files
  users        import one human user, converting the Django password hash
  verify       check discovery, client count and the groups-claim gate

`all` runs keys, apply, secrets, verify. It never runs users: the password hash
must be supplied by a human, one user at a time.

Nothing here talks to Authentik. Authentik keeps serving until a consumer is
moved deliberately.

Usage:
  scripts/iam/zitadel-provision.py keys
  scripts/iam/zitadel-provision.py apply --org-id <id>
  scripts/iam/zitadel-provision.py secrets --dry-run
  scripts/iam/zitadel-provision.py secrets --apply-config
  scripts/iam/zitadel-provision.py users --django-hash '<pbkdf2_sha256$...>'
  scripts/iam/zitadel-provision.py verify
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import signal
import stat
import subprocess
import sys
import tempfile
from pathlib import Path

import yaml

REPO = Path(__file__).resolve().parents[2]
SOPS_DIR = REPO / "k8s" / "argocd" / "secrets"
SOPS_CONFIG = REPO / ".sops.yaml"
CONF_DIR = Path(os.environ.get("ZITADEL_CONF_DIR", Path.home() / ".config" / "zitadel"))
STATE_FILE = CONF_DIR / "state.json"
NAMESPACE = os.environ.get("ZITADEL_NAMESPACE", "iam")
BASE_URL = os.environ.get("ZITADEL_BASE_URL", "https://zitadel.riotpiao.com")

# ── desired ZITADEL state ──────────────────────────────────────────────────
# Declared here, applied through the API by the `apply` phase. This replaces the
# retired terraform/zitadel root module: same objects, no state file to drift and
# no second writer.
PROJECT_NAME = os.environ.get("ZITADEL_PROJECT_NAME", "homelab")
PROJECT_GROUP = "homelab"

# Role keys EQUAL the Authentik group names, so the `groups` claim keeps working
# for every consumer that maps on it (kube-apiserver RBAC, MinIO policies,
# Grafana roles).
ROLES = {
    "homelab-admins": "Cluster administrators with full access",
    "minio-admins": "MinIO administrators",
    "minio-photos": "Photos bucket (Immich) access",
    "minio-documents": "Documents bucket (Paperless) access",
    "minio-backups": "Backups bucket read-only access",
    "poimen-admins": "Poimen memory administrators",
    "poimen-devs": "Dev and staging projects access",
    "poimen-prod-readonly": "Production projects read-only access",
    "paperless-admins": "Paperless administrators",
    "paperless-finance": "Finance documents",
    "paperless-legal": "Legal documents",
    "paperless-hr": "HR documents",
    "grafana-admins": "Grafana administrators",
    "grafana-editors": "Grafana dashboard editors",
    "grafana-viewers": "Grafana dashboard viewers",
    "sqs-users": "SQS/Temporal queue read access",
    "sqs-writers": "SQS/Temporal queue read/write access",
    "s3-users": "S3 read access",
    "s3-writers": "S3 read/write access",
    # LLM gate for the api-gw. The colon form matches the Authentik-era
    # permission string, so consumer manifests did not change.
    "llm:inference": "LLM inference access (api-gw)",
    "llm-inference": "LLM inference access (dash alias)",
}

_WEB = {"app_type_id": "OIDC_APP_TYPE_WEB", "auth_method_id": "OIDC_AUTH_METHOD_TYPE_BASIC"}
_NATIVE = {"app_type_id": "OIDC_APP_TYPE_NATIVE", "auth_method_id": "OIDC_AUTH_METHOD_TYPE_NONE"}
_PKCE = {"app_type_id": "OIDC_APP_TYPE_USER_AGENT", "auth_method_id": "OIDC_AUTH_METHOD_TYPE_NONE"}

OIDC_APPS = {
    "api-gw": {**_WEB, "redirect_uris": [
        "http://localhost:3000/callback", "https://api.riotpiao.com/callback"]},
    "argocd": {**_WEB, "redirect_uris": [
        "http://localhost:8080/auth/callback",
        "https://argocd.riotpiao.com/auth/callback"]},
    "forgejo": {**_WEB, "redirect_uris": [
        "http://localhost:3000/user/oauth2/authentik/callback",
        "https://forgejo.riotpiao.com/user/oauth2/authentik/callback"]},
    "grafana": {**_WEB, "redirect_uris": [
        "http://localhost:3000/login/generic_oauth",
        "https://grafana.riotpiao.com/login/generic_oauth"]},
    "immich": {**_WEB, "app_type_id": "OIDC_APP_TYPE_NATIVE", "redirect_uris": [
        "app.immich:///oauth-callback", "https://img.riotpiao.com/auth/login",
        "https://img.riotpiao.com/user/oauth2/callback"]},
    "kubernetes": {**_PKCE, "dev_mode": True, "redirect_uris": [
        "http://localhost:8000", "http://localhost:18000"]},
    "minio": {**_WEB, "redirect_uris": [
        "http://localhost:9000/auth/sso/oauth2/code",
        "https://minio.riotpiao.com/auth/sso/oauth2/code"]},
    "paperless": {**_WEB, "redirect_uris": [
        "http://localhost:8000/accounts/oidc/authentik/login/callback/",
        "https://paperless.riotpiao.com/accounts/oidc/authentik/login/callback/"]},
    "poimen": {**_WEB, "redirect_uris": [
        "http://localhost:3000/callback", "https://poimen.riotpiao.com/callback"]},
    "queue-sqs": {**_WEB, "redirect_uris": [
        "http://localhost:8080/callback", "https://queue.riotpiao.com/callback"]},
    "vault": {**_WEB, "redirect_uris": [
        "http://localhost:8200/ui/vault/auth/oidc/oidc/callback",
        "https://vault.riotpiao.com/ui/vault/auth/oidc/oidc/callback"]},
}


_TEMP_FILES: set[str] = set()


def _cleanup_temps(*_a) -> None:
    for path in list(_TEMP_FILES):
        try:
            os.unlink(path)
        except OSError:
            pass
        _TEMP_FILES.discard(path)


for _sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
    signal.signal(_sig, lambda s, f: (_cleanup_temps(), sys.exit(130)))


def die(msg: str) -> None:
    _cleanup_temps()
    print(f"FATAL: {msg}", file=sys.stderr)
    sys.exit(1)


def run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, check=False, text=True, capture_output=True, **kw)


def need(binary: str) -> None:
    if subprocess.run(["sh", "-c", f"command -v {binary}"], capture_output=True).returncode != 0:
        die(f"{binary} is not on PATH")


def write_private(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.fchmod(fd, 0o600)  # open() mode applies at creation only
        with os.fdopen(fd, "wb") as fh:
            fh.write(data)
    finally:
        pass


def secret_value(name: str, key: str) -> str:
    cp = run(["kubectl", "-n", NAMESPACE, "get", "secret", name, "-o", f"jsonpath={{{key}}}"])
    if cp.returncode != 0:
        die(f"cannot read secret {NAMESPACE}/{name}: {cp.stderr.strip()}")
    raw = cp.stdout.strip()
    if not raw:
        die(f"secret {NAMESPACE}/{name} has no key {key}")
    return base64.b64decode(raw).decode()


def load_env_file(path: Path = REPO / ".env") -> None:
    """Load KEY=VALUE lines from .env without overriding the real environment.

    The file is git-ignored and holds the bootstrap token a human pastes in.
    Sourcing it in the shell would work too, but then every phase depends on the
    caller remembering to do it.
    """
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and key not in os.environ:
            os.environ[key] = value


def bootstrap_token() -> str | None:
    """Resolve the iam-admin PAT: environment, then cached file, then cluster.

    Order matters. ZITADEL_BOOTSTRAP_TOKEN in .env is filled by a human after the
    first sync and wins. The cached file under ~/.config/zitadel keeps a run
    working when the API server is unreachable. The cluster Secret is the last
    resort and the only source that always reflects a rotation.
    """
    token = os.environ.get("ZITADEL_BOOTSTRAP_TOKEN", "").strip()
    if token:
        return token
    cached = CONF_DIR / "iam-admin.pat"
    if cached.exists():
        return cached.read_text().strip()
    try:
        return secret_value("iam-admin-pat", ".data.pat").strip()
    except SystemExit:
        return None


def curl_json(method: str, url: str, token: str | None = None, body: dict | None = None,
              timeout: int = 20) -> tuple[int, str]:
    """Call the API without putting a secret in the process argv.

    `curl --config -` reads the options from stdin, so neither the bearer token
    nor the password hash of a user import shows up in `ps`.
    """
    def esc(value: str) -> str:
        return value.replace("\\", "\\\\").replace('"', '\\"')

    cfg = [f'url = "{esc(url)}"', f'request = "{method}"', "silent", "show-error",
           f"max-time = {timeout}", 'write-out = "\\n%{http_code}"',
           'header = "Content-Type: application/json"']
    # Reaching the instance by name needs either public DNS through the tunnel
    # or a pinned address. ZITADEL picks the instance from the Host header, so a
    # plain port-forward fails with "Instance not found". Set this to bypass DNS:
    #   ZITADEL_CURL_RESOLVE=zitadel.riotpiao.com:443:192.168.1.160
    pinned = os.environ.get("ZITADEL_CURL_RESOLVE", "").strip()
    if pinned:
        cfg.append(f'resolve = "{esc(pinned)}"')
    if token:
        cfg.append(f'header = "Authorization: Bearer {esc(token)}"')
    if body is not None:
        cfg.append(f'data = "{esc(json.dumps(body))}"')
    cp = run(["curl", "--config", "-"], input="\n".join(cfg) + "\n")
    if cp.returncode != 0:
        die(f"{method} {url} failed: {cp.stderr.strip()}")
    payload, _, code = cp.stdout.rpartition("\n")
    return int(code.strip() or 0), payload


# ── phase: keys ────────────────────────────────────────────────────────────

def phase_keys(_args: argparse.Namespace) -> int:
    write_private(CONF_DIR / "iam-admin.json", secret_value("iam-admin", ".data.iam-admin\\.json").encode())
    write_private(CONF_DIR / "iam-admin.pat", secret_value("iam-admin-pat", ".data.pat").strip().encode())
    print(f"machine key  -> {CONF_DIR / 'iam-admin.json'} (0600)")
    print(f"PAT          -> {CONF_DIR / 'iam-admin.pat'} (0600)")
    print("both files stay outside the repo; never commit them")
    return 0


# ── phase: apply ───────────────────────────────────────────────────────────

def _api(method: str, path: str, token: str, body: dict | None = None,
         timeout: int = 30) -> tuple[int, dict]:
    code, payload = curl_json(method, f"{BASE_URL}{path}", token=token, body=body, timeout=timeout)
    try:
        return code, (json.loads(payload) if payload.strip() else {})
    except json.JSONDecodeError:
        return code, {"_raw": payload}


def _ensure_project(token: str, name: str) -> str:
    code, d = _api("POST", "/management/v1/projects/_search", token, {})
    project_id = ""
    for row in d.get("result") or []:
        if row.get("name") == name:
            project_id = row["id"]
            break
    if not project_id:
        code, d = _api("POST", "/management/v1/projects", token, {"name": name})
        if code not in (200, 201) or not d.get("id"):
            die(f"create project {name} failed: {code} {d}")
        project_id = d["id"]

    # projectRoleAssertion puts the project roles into the ACCESS token as
    # `urn:zitadel:iam:org:project:roles`. Without it the api-gw sees no
    # llm:inference role and answers 403 for every inference call, even when
    # the caller is correctly authorised in ZITADEL.
    code, d = _api("PUT", f"/management/v1/projects/{project_id}",
                   token, {"name": name, "projectRoleAssertion": True})
    if code not in (200, 201):
        die(f"enable projectRoleAssertion on {name} failed: {code} {d}")
    return project_id


def _ensure_roles(token: str, project_id: str, roles: dict[str, str]) -> None:
    code, d = _api("POST", f"/management/v1/projects/{project_id}/roles/_search", token, {})
    have = {r.get("roleKey") for r in (d.get("result") or [])}
    for key, display in roles.items():
        if key in have:
            continue
        code, d = _api("POST", f"/management/v1/projects/{project_id}/roles", token,
                       {"roleKey": key, "displayName": display, "group": PROJECT_GROUP})
        if code not in (200, 201):
            die(f"create role {key} failed: {code} {d}")


def _ensure_oidc_apps(token: str, project_id: str, apps: dict[str, dict]) -> tuple[dict, dict]:
    code, d = _api("POST", f"/management/v1/projects/{project_id}/apps/_search", token, {})
    existing = {a.get("name"): a for a in (d.get("result") or [])}
    ids: dict[str, str] = {}
    secrets: dict[str, str] = {}
    for name, spec in apps.items():
        row = existing.get(name)
        if row:
            ids[name] = (row.get("oidcConfig") or {}).get("clientId", "")
            app_id = row.get("id")
        else:
            body = {
                "name": name,
                "redirectUris": spec["redirect_uris"],
                "responseTypes": ["OIDC_RESPONSE_TYPE_CODE"],
                "grantTypes": ["OIDC_GRANT_TYPE_AUTHORIZATION_CODE",
                               "OIDC_GRANT_TYPE_REFRESH_TOKEN"],
                "appType": spec.get("app_type_id", "OIDC_APP_TYPE_WEB"),
                "authMethodType": spec.get("auth_method_id", "OIDC_AUTH_METHOD_TYPE_BASIC"),
                "devMode": bool(spec.get("dev_mode")),
                "accessTokenType": "OIDC_TOKEN_TYPE_BEARER",
                "version": "OIDC_VERSION_1_0",
                "idTokenRoleAssertion": True,
            }
            code, d = _api("POST", f"/management/v1/projects/{project_id}/apps/oidc", token, body)
            if code not in (200, 201) or not d.get("appId"):
                die(f"create app {name} failed: {code} {d}")
            app_id = d["appId"]
            ids[name] = d.get("clientId", "")
        if spec.get("auth_method_id") == "OIDC_AUTH_METHOD_TYPE_NONE":
            continue
        code, d = _api("POST",
                       f"/management/v1/projects/{project_id}/apps/{app_id}"
                       "/oidc_config/_generate_client_secret", token, {})
        if code in (200, 201) and d.get("clientSecret"):
            secrets[name] = d["clientSecret"]
    return ids, secrets


def phase_apply(_args: argparse.Namespace) -> int:
    """Declare the ZITADEL objects through the API. No Terraform, no state drift."""
    token = bootstrap_token()
    if not token:
        die("no iam-admin PAT — run the keys phase first")

    project_id = _ensure_project(token, PROJECT_NAME)
    print(f"project   : {PROJECT_NAME} ({project_id})")
    _ensure_roles(token, project_id, ROLES)
    print(f"roles     : {len(ROLES)} ensured")
    ids, secrets = _ensure_oidc_apps(token, project_id, OIDC_APPS)
    print(f"oidc apps : {len(ids)} ensured, {len(secrets)} secrets issued")

    state = {
        "project_id": {"value": project_id},
        "client_ids": {"value": ids},
        "client_secrets": {"value": secrets},
        "machine_keys": {"value": {}},
        "groups_claim_enabled": {"value": False},
    }
    write_private(STATE_FILE, json.dumps(state, indent=2, sort_keys=True).encode())
    print(f"state     -> {STATE_FILE} (0600)")
    return 0


# ── phase: secrets ─────────────────────────────────────────────────────────

# client -> [(sops file, Secret name inside the file, key, kind)]
# Only secrets with a git-tracked SOPS file appear here. Every other client is
# printed as manual: guessing a path writes a credential into the wrong Secret.
SECRET_MAP = {
    "grafana": [("grafana-oidc-secrets.enc.yaml", "grafana-oidc",
                 "GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET", "secret")],
    "minio": [("minio-secrets.enc.yaml", "minio-oidc",
               "MINIO_IDENTITY_OPENID_CLIENT_SECRET", "secret")],
    "paperless": [("paperless-ai-secrets.enc.yaml", "paperless-ai-config",
                   "LLM_AUTH_CLIENT_SECRET", "secret")],
}

# Clients whose client_id lives in a plain (unencrypted) manifest. They need the
# id update as much as the secret: ZITADEL generates a new client_id for every
# application, so a secret-only re-point yields invalid_client.
CLIENT_ID_MAP = {
    "grafana": ("k8s/infra/logging/grafana-values.yaml",
                r"(?m)^(\s*client_id:\s*)\S+\s*$", r"\g<1>{value}"),
    "minio": ("k8s/infra/minio/minio-tenant.yaml",
              r'(?m)^(\s*value:\s*)"(?:minio|\d+)"\s*$', r'\g<1>"{value}"'),
    "paperless": ("k8s/apps/paperless/paperless-ai.yaml",
                  r"client_id=(?:paperless-ai-agent|\d+)", "client_id={value}"),
}

# Secrets that a live workload mounts but no SOPS file in git defines. The phase
# cannot re-point them safely, so it names them.
MANUAL_ONLY: dict[str, str] = {}


def state_outputs() -> dict:
    """Read the object map written by the apply phase."""
    if not STATE_FILE.exists():
        die(f"{STATE_FILE} missing — run the apply phase first")
    return json.loads(STATE_FILE.read_text())


def sops_env() -> dict:
    env = dict(os.environ)
    env.setdefault("SOPS_AGE_KEY_FILE", str(Path.home() / ".sops" / "key.txt"))
    return env


def write_secret(rel_path: str, secret_name: str, key: str, value: str,
                 dry_run: bool) -> None:
    path = SOPS_DIR / rel_path
    if not path.exists():
        die(f"{path} does not exist — the mapping and the repo disagree")
    env = sops_env()
    dec = run(["sops", "--config", str(SOPS_CONFIG), "decrypt", str(path)], env=env)
    if dec.returncode != 0:
        die(f"cannot decrypt {rel_path}: {dec.stderr.strip()}")
    docs = [d for d in yaml.safe_load_all(dec.stdout) if d]
    matching = [d for d in docs if (d.get("metadata") or {}).get("name") == secret_name]
    if len(matching) != 1:
        die(f"{rel_path}: expected exactly one document named {secret_name}, found {len(matching)}")

    current = (matching[0].get("stringData") or {}).get(key)
    if current == value:
        print(f"  unchanged {rel_path}:{secret_name}:{key}")
        return
    if dry_run:
        print(f"  would set {rel_path}:{secret_name}:{key} (len {len(value)})")
        return

    if len(docs) == 1:
        cp = run(["sops", "--config", str(SOPS_CONFIG), "set", "--value-stdin", "--idempotent",
                  str(path), f'["stringData"]["{key}"]'], env=env, input=json.dumps(value))
        if cp.returncode != 0:
            die(f"sops set failed on {rel_path}: {cp.stderr.strip()}")
        print(f"  wrote {rel_path}:{secret_name}:{key}")
        return

    # Multi-document file. `sops set` cannot address one document, so the whole
    # file is rewritten. sops encrypt reads the creation rule from the FILENAME,
    # so the temp path is overridden with the repo-relative path; without that
    # override sops finds no matching rule and refuses.
    matching[0].setdefault("stringData", {})[key] = value
    plain = "---\n".join(yaml.safe_dump(d, sort_keys=False) for d in docs)
    fd, tmp = tempfile.mkstemp(suffix=".yaml", dir=tempfile.gettempdir())
    _TEMP_FILES.add(tmp)
    try:
        with os.fdopen(fd, "w") as fh:
            fh.write(plain)
        enc = run(["sops", "--config", str(SOPS_CONFIG), "encrypt",
                   "--filename-override", str(path), tmp], env=env)
        if enc.returncode != 0:
            die(f"sops encrypt failed for {rel_path}: {enc.stderr.strip()}")
        mode = stat.S_IMODE(path.stat().st_mode)
        fd2, staged = tempfile.mkstemp(suffix=".enc.yaml", dir=str(path.parent))
        try:
            with os.fdopen(fd2, "w") as fh:
                fh.write(enc.stdout)
            os.chmod(staged, mode)
            os.replace(staged, path)  # same directory: atomic, never EXDEV
        finally:
            if os.path.exists(staged):
                os.unlink(staged)
        print(f"  wrote {rel_path}:{secret_name}:{key} (multi-document rewrite)")
    finally:
        if os.path.exists(tmp):
            os.unlink(tmp)
        _TEMP_FILES.discard(tmp)


def update_client_id(client: str, value: str, dry_run: bool, apply_config: bool) -> None:
    spec = CLIENT_ID_MAP.get(client)
    if not spec:
        return
    rel_path, pattern, repl = spec
    path = REPO / rel_path
    if not path.exists():
        die(f"{path} does not exist — the client-id mapping is stale")
    text = path.read_text()
    matches = list(re.finditer(pattern, text))
    if len(matches) != 1:
        die(f"{rel_path}: expected 1 match for the {client} client id, found {len(matches)}")
    new_line = matches[0].expand(repl.format(value=value))
    if matches[0].group(0) == new_line:
        print(f"  unchanged {rel_path}: {client} client id")
        return
    if dry_run or not apply_config:
        print(f"  would change {rel_path}: {matches[0].group(0).strip()} -> {new_line.strip()}")
        if not apply_config:
            print("    re-run with --apply-config to write this plaintext manifest edit")
        return
    path.write_text(text[:matches[0].start()] + new_line + text[matches[0].end():])
    print(f"  wrote {rel_path}: {client} client id")


def phase_secrets(args: argparse.Namespace) -> int:
    need("sops")
    out = state_outputs()
    ids = out.get("client_ids", {}).get("value", {})
    secrets = out.get("client_secrets", {}).get("value", {})
    machine_keys = out.get("machine_keys", {}).get("value", {})

    if not ids and not machine_keys:
        die("no ZITADEL objects recorded yet — run the apply phase first")

    print("[1/3] client secrets into SOPS")
    for client, entries in SECRET_MAP.items():
        for rel_path, secret_name, key, kind in entries:
            value = ids.get(client) if kind == "id" else secrets.get(client)
            if value:
                write_secret(rel_path, secret_name, key, value, args.dry_run)

    print("[2/3] client ids in plain manifests")
    for client in list(CLIENT_ID_MAP) + list(SECRET_MAP):
        if ids.get(client):
            update_client_id(client, ids[client], args.dry_run, args.apply_config)

    print("[3/3] machine keys")
    key_dir = CONF_DIR / "machine-keys"
    for name, key_json in machine_keys.items():
        if args.dry_run:
            print(f"  would write {key_dir / (name + '.json')}")
        else:
            write_private(key_dir / f"{name}.json", key_json.encode())
            print(f"  wrote {key_dir / (name + '.json')} (0600)")

    unhandled = sorted(set(ids) - set(SECRET_MAP) - set(CLIENT_ID_MAP) - set(MANUAL_ONLY))
    if MANUAL_ONLY or unhandled:
        print("\n  MANUAL — no safe write path exists for these clients:")
        for client, why in MANUAL_ONLY.items():
            print(f"    {client}: {why}")
        for client in unhandled:
            print(f"    {client}: set its client_id and secret in its own manifest or Secret file")
    return 0


# ── phase: users ───────────────────────────────────────────────────────────

HASH_RE = re.compile(r"^pbkdf2_sha256\$(?P<rounds>\d+)\$(?P<salt>[^$]+)\$(?P<hash>[^$]+)$")


def convert_hash(django_hash: str) -> str:
    """Django pbkdf2_sha256 -> passwap MCF $pbkdf2-sha256$...

    ZITADEL imports one MCF string via hashedPassword.hash (max 200 chars) and
    pins passwap, whose pbkdf2 format is $pbkdf2-sha256$<rounds>$<salt>$<hash>
    with alt-base64 (standard base64, + -> ., no padding). Rounds, salt and hash
    bytes are unchanged, so the password still verifies.
    """
    m = HASH_RE.match(django_hash.strip())
    if not m:
        die("unsupported hash format — only pbkdf2_sha256 is convertible")
    alt = m.group("hash").rstrip("=").replace("+", ".")
    mcf = f"$pbkdf2-sha256${m.group('rounds')}${m.group('salt')}${alt}"
    if len(mcf) > 200:
        die(f"converted hash is {len(mcf)} chars, ZITADEL accepts 200")
    return mcf


def phase_users(args: argparse.Namespace) -> int:
    token = bootstrap_token()
    if not token:
        die("no bootstrap token — set ZITADEL_BOOTSTRAP_TOKEN in .env, or run the keys phase")
    if not args.django_hash:
        die("pass --django-hash with the pbkdf2_sha256 value exported from Authentik")
    body = {
        "username": args.username,
        "profile": {"givenName": args.given_name, "familyName": args.family_name},
        "email": {"email": args.email, "isVerified": True},
        "hashedPassword": {"hash": convert_hash(args.django_hash)},
        "passwordChangeRequired": False,
    }
    code, payload = curl_json("POST", f"{BASE_URL}/v2/users/human", token=token, body=body)
    if code in (200, 201):
        print(f"  {args.username}: created")
        return 0
    if code == 409:
        print(f"  {args.username}: already exists — the password was NOT updated")
        return 1
    die(f"create {args.username} -> {code} {payload[:300]}")


# ── phase: verify ──────────────────────────────────────────────────────────

def phase_verify(_args: argparse.Namespace) -> int:
    code, _ = curl_json("GET", f"{BASE_URL}/.well-known/openid-configuration", timeout=10)
    print(f"discovery      : {code}")
    if code != 200:
        die("discovery endpoint is not 200 — the instance or the ingress is not ready")

    out = state_outputs()
    ids = out.get("client_ids", {}).get("value", {})
    print(f"client apps    : {len(ids)}")
    enabled = bool(out.get("groups_claim_enabled", {}).get("value"))
    print(f"groups action  : {enabled}")
    if not enabled:
        print("BLOCKED for cutover: the groups claim action is disabled. Every consumer")
        print("that reads `groups` (kube-apiserver, MinIO, Grafana) sees no groups.")
        print("Enable it only after the live gate passes, with the script path set.")
    return 0


PHASES = {"keys": phase_keys, "apply": phase_apply, "secrets": phase_secrets,
          "users": phase_users, "verify": phase_verify}


def main() -> int:
    parser = argparse.ArgumentParser(description="Provision ZITADEL end to end.")
    parser.add_argument("phase", choices=[*PHASES, "all"])
    parser.add_argument("--org-id", help="ZITADEL org id (required for apply)")
    parser.add_argument("--dry-run", action="store_true", help="print actions, write nothing")
    parser.add_argument("--apply-config", action="store_true",
                        help="also rewrite client ids inside plain manifests")
    parser.add_argument("--username", default="rock")
    parser.add_argument("--email", default="rock@riotpiao.com")
    parser.add_argument("--given-name", default="Rock")
    parser.add_argument("--family-name", default="Liang")
    parser.add_argument("--django-hash", help="pbkdf2_sha256 value exported from Authentik")
    args = parser.parse_args()

    load_env_file()
    need("kubectl")
    if args.phase == "all":
        if not args.org_id:
            die("all requires --org-id")
        for name in ("keys", "apply"):
            rc = PHASES[name](args)
            if rc:
                return rc
        rc = phase_secrets(args)
        if rc:
            return rc
        return phase_verify(args)
    return PHASES[args.phase](args)


if __name__ == "__main__":
    sys.exit(main())