#!/usr/bin/env python3
"""yakcloud dev — 로컬 kind 클러스터 리컨실 엔진(개발 환경, 콘솔 API 불필요).

관리형 리컨실(yakcloud_deploy.py = 콘솔 API)과 **동일한 계약**을 로컬 kind 에서 재현한다.
그래야 dev(로컬 kind) → val/prod(관리형 RKE2) 가 '직선'이 된다.
  1) requires 소스  → 로컬 컨테이너(Deployment+Service+Secret)로 프로비저닝
  2) workloads      → 이미지 빌드 + kind load + Deployment+Service+Ingress
  3) 바인딩         → 백엔드 _bind_env_for 와 **동일한** <ALIAS>_URL/_HOST/_PORT/… env 주입

dev 는 CLI 로컬 완결 — 콘솔 API·인터넷 불필요(최초 이미지 pull 제외), 오프라인 반복.
kind 생성/삭제(up/down)는 bin/yakcloud 가 담당. 이 엔진은 deploy | status | down(네임스페이스 정리)만.

환경변수:
  YAK_DEV_CONTEXT   kubectl 컨텍스트(기본 kind-yak-dev)
  YAK_DEV_CLUSTER   kind 클러스터 이름(kind load 용, 기본 yak-dev)
  YAK_DEV_RUNTIME   docker|podman (이미지 빌드/로드)
  YAK_DEV_HOST_SUFFIX  인그레스 호스트 접미사(기본 dev.localhost) → <project>.dev.localhost
  TAG               이미지 태그 치환(${TAG}); 없으면 dev
사용: yakcloud_dev.py <deploy|status|down> [manifest.yaml]
"""
from __future__ import annotations

import base64
import json
import os
import secrets
import hashlib
import subprocess
import sys
import time
import urllib.request
from urllib.parse import quote

try:
    import yaml
except ModuleNotFoundError:
    raise SystemExit("PyYAML 필요 — 'pip install pyyaml' 후 다시 실행하세요.")

CTX = os.environ.get("YAK_DEV_CONTEXT", "kind-yak-dev")
KIND_CLUSTER = os.environ.get("YAK_DEV_CLUSTER", "yak-dev")
RUNTIME = os.environ.get("YAK_DEV_RUNTIME", "docker")
HOST_SUFFIX = os.environ.get("YAK_DEV_HOST_SUFFIX", "dev.localhost")
TAG = os.environ.get("TAG", "dev")

# 로컬 데이터 소스 = kind '밖'의 **공유** 도커/포드만 컨테이너(타입별 1개) + 프로젝트별 격리 DB.
#   관리형 shared 모델과 동일 계약 — 이미 있는 공유 서버에 <project>_<src> 격리 DB/유저만 민팅(소스 생성비 최소).
#   클러스터(kind)는 앱(비즈니스)만. 앱은 no-selector Service+Endpoints(host=소스명)로 공유 컨테이너의 격리 DB 접속.
_SVC_PORT = {"POSTGRESQL": 5432, "MYSQL": 3306, "MARIADB": 3306, "MONGODB": 27017, "REDIS": 6379}
_IMG = {"POSTGRESQL": "postgres:16-alpine", "MYSQL": "mysql:8", "MARIADB": "mariadb:11",
        "MONGODB": "mongo:7", "REDIS": "redis:7-alpine"}
_SHARED_CT = {"POSTGRESQL": "yak-dev-postgres", "MYSQL": "yak-dev-mysql", "MARIADB": "yak-dev-mariadb",
              "MONGODB": "yak-dev-mongo", "REDIS": "yak-dev-redis"}
SHARED_NET = os.environ.get("YAK_DEV_NET", "kind")   # kind 가 쓰는 도커 네트워크
SHARED_ADMIN = "yakdevadmin"                          # dev 전용 공유 컨테이너 관리자 비밀번호(로컬 한정)
SUPPORTED = set(_SVC_PORT)
MANAGED = "app.kubernetes.io/managed-by=yakcloud-dev"


def log(m: str) -> None:
    print("\033[36m▸\033[0m " + m, flush=True)


def warn(m: str) -> None:
    print("\033[33m⚠\033[0m " + m, flush=True)


def die(m: str) -> None:
    raise SystemExit("\033[31m✗\033[0m " + m)


def sanitize(s: str) -> str:
    """K8s 이름 규칙(소문자·숫자·하이픈, 시작/끝은 영숫자)."""
    out = "".join(c if (c.isalnum() or c == "-") else "-" for c in (s or "").lower())
    out = out.strip("-") or "app"
    return out[:53]


def sh(cmd: list[str], *, input: str | None = None, check: bool = True,
       capture: bool = True, env: dict | None = None) -> subprocess.CompletedProcess:
    e = dict(os.environ)
    if env:
        e.update(env)
    r = subprocess.run(cmd, input=input, text=True,
                       stdout=subprocess.PIPE if capture else None,
                       stderr=subprocess.PIPE if capture else None, env=e)
    if check and r.returncode != 0:
        msg = (r.stderr or r.stdout or "").strip()
        die(f"명령 실패: {' '.join(cmd)}\n  {msg[:500]}")
    return r


def kubectl(args: list[str], *, input: str | None = None, check: bool = True,
            capture: bool = True) -> subprocess.CompletedProcess:
    return sh(["kubectl", "--context", CTX, *args], input=input, check=check, capture=capture)


def apply(objs: list[dict]) -> None:
    """K8s 오브젝트(dict)들을 kubectl apply -f - 로 반영."""
    doc = "\n---\n".join(json.dumps(o) for o in objs)
    kubectl(["apply", "-f", "-"], input=doc)


def context_ok() -> None:
    r = sh(["kubectl", "config", "get-contexts", "-o", "name"], check=False)
    if CTX not in (r.stdout or "").split():
        die(f"kubectl 컨텍스트 '{CTX}' 없음 — 먼저 'yakcloud dev up' 으로 로컬 클러스터를 만드세요.")


def ensure_ns(ns: str) -> None:
    apply([{"apiVersion": "v1", "kind": "Namespace",
            "metadata": {"name": ns, "labels": {"app.kubernetes.io/managed-by": "yakcloud-dev"}}}])


# ── 소스 자격: 최초 생성 시 랜덤(유저=고정 appuser, db=소스별), 이후 재사용(idempotent) ──────
def get_or_make_creds(ns: str, name: str) -> dict:
    """프로젝트-소스별 비밀번호 — 기존 Secret 이 있으면 재사용(멱등), 없으면 랜덤 생성."""
    sec = f"yak-dev-src-{sanitize(name)}"
    r = kubectl(["get", "secret", sec, "-n", ns, "-o", "json"], check=False)
    if r.returncode == 0:
        pw = (json.loads(r.stdout).get("data", {}) or {}).get("password")
        if pw:
            return {"password": base64.b64decode(pw).decode(), "secret": sec}
    return {"password": secrets.token_hex(16), "secret": sec}


def _rt(args: list[str], **kw) -> subprocess.CompletedProcess:
    return sh([RUNTIME, *args], **kw)


def _dexec(cname: str, shell_cmd: str, check: bool = True) -> subprocess.CompletedProcess:
    return _rt(["exec", cname, "sh", "-c", shell_cmd], check=check)


def _dbname(project: str, source: str) -> str:
    """프로젝트별 격리 단위(DB·유저)명 — DB 안전(소문자·영숫자·언더스코어)."""
    raw = f"{project}_{source}".lower()
    out = "".join(c if (c.isalnum() or c == "_") else "_" for c in raw).strip("_")
    return (out or "app")[:40]


def _container_ip(cname: str) -> str:
    r = _rt(["inspect", "-f",
             '{{(index .NetworkSettings.Networks "' + SHARED_NET + '").IPAddress}}', cname], check=False)
    return (r.stdout or "").strip()


def _ensure_shared(stype: str) -> tuple[str, str]:
    """타입별 '공유' 컨테이너를 kind 네트워크에 보장(idempotent, 관리자 자격) → (이름, IP)."""
    cname = _SHARED_CT[stype]
    if not _rt(["ps", "-q", "-f", f"name=^{cname}$"], check=False).stdout.strip():
        _rt(["rm", "-f", cname], check=False)   # 죽은 동명 컨테이너 정리
        img = _IMG[stype]
        run = ["run", "-d", "--name", cname, "--network", SHARED_NET,
               "--label", "yakcloud.dev/shared=1", "--label", f"yakcloud.dev/type={stype}"]
        if stype == "POSTGRESQL":
            run += ["-e", f"POSTGRES_PASSWORD={SHARED_ADMIN}", img]
            ready = "pg_isready -U postgres >/dev/null 2>&1"
        elif stype == "MYSQL":
            run += ["-e", f"MYSQL_ROOT_PASSWORD={SHARED_ADMIN}", img]
            ready = f'mysql -uroot -p{SHARED_ADMIN} -e "SELECT 1" >/dev/null 2>&1'
        elif stype == "MARIADB":
            run += ["-e", f"MARIADB_ROOT_PASSWORD={SHARED_ADMIN}", img]
            ready = f'mariadb -uroot -p{SHARED_ADMIN} -e "SELECT 1" >/dev/null 2>&1'
        elif stype == "MONGODB":
            run += ["-e", "MONGO_INITDB_ROOT_USERNAME=root", "-e", f"MONGO_INITDB_ROOT_PASSWORD={SHARED_ADMIN}", img]
            ready = (f'mongosh "mongodb://root:{SHARED_ADMIN}@127.0.0.1:27017/admin" --quiet '
                     '--eval "db.runCommand({ping:1}).ok" | grep -q 1')
        else:  # REDIS
            run += [img, "redis-server", "--requirepass", SHARED_ADMIN]
            ready = f'redis-cli -a {SHARED_ADMIN} ping 2>/dev/null | grep -q PONG'
        log(f"공유 소스 컨테이너 '{cname}' ({stype}) 기동 ({RUNTIME}, net={SHARED_NET})")
        _rt(run, capture=False)
        for _ in range(90):
            if _dexec(cname, ready, check=False).returncode == 0:
                break
            time.sleep(2)
        else:
            die(f"공유 컨테이너 '{cname}' 준비 대기 초과 — '{RUNTIME} logs {cname}' 확인")
    ip = _container_ip(cname)
    if not ip:
        die(f"공유 컨테이너 '{cname}' IP 조회 실패 — 도커 네트워크 '{SHARED_NET}' 확인")
    return cname, ip


def _mint(stype: str, cname: str, unit: str, pw: str) -> None:
    """공유 컨테이너에 프로젝트별 격리 DB/유저 생성(idempotent). unit=DB·유저명(영숫자·언더스코어)."""
    if stype == "POSTGRESQL":
        _dexec(cname, f'psql -U postgres -tAc "SELECT 1 FROM pg_roles WHERE rolname=\'{unit}\'" | grep -q 1 '
                      f'|| psql -U postgres -c "CREATE ROLE {unit} LOGIN PASSWORD \'{pw}\'"')
        _dexec(cname, f'psql -U postgres -tAc "SELECT 1 FROM pg_database WHERE datname=\'{unit}\'" | grep -q 1 '
                      f'|| psql -U postgres -c "CREATE DATABASE {unit} OWNER {unit}"')
    elif stype in ("MYSQL", "MARIADB"):
        client = "mysql" if stype == "MYSQL" else "mariadb"
        sql = (f"CREATE DATABASE IF NOT EXISTS {unit}; "
               f"CREATE USER IF NOT EXISTS '{unit}'@'%' IDENTIFIED BY '{pw}'; "
               f"GRANT ALL ON {unit}.* TO '{unit}'@'%'; FLUSH PRIVILEGES;")
        _dexec(cname, f'{client} -uroot -p{SHARED_ADMIN} -e "{sql}"')
    elif stype == "MONGODB":
        js = (f'try{{db.getSiblingDB("{unit}").createUser({{user:"{unit}",pwd:"{pw}",'
              f'roles:[{{role:"readWrite",db:"{unit}"}}]}})}}catch(e){{if(!/already exists/i.test(e.message))throw e}}')
        _dexec(cname, f'mongosh "mongodb://root:{SHARED_ADMIN}@127.0.0.1:27017/admin" --quiet --eval \'{js}\'')


def provision_source(ns: str, project: str, req: dict) -> dict | None:
    """requires 소스 → 공유 도커 컨테이너의 프로젝트별 격리 DB + no-selector Service/Endpoints. 접속정보 반환."""
    name = req["name"]
    stype = req["type"].upper()
    if req.get("mode") == "local":
        warn(f"source '{name}': mode=local(인클러스터)는 로컬 dev 에서 공유 컨테이너로 대체됩니다.")
    if stype not in SUPPORTED:
        die(f"source '{name}' 타입 {stype} — 로컬 dev 미지원(지원: {', '.join(sorted(SUPPORTED))}).")
    svc = sanitize(name)
    unit = _dbname(project, name)
    port = _SVC_PORT[stype]
    pw = get_or_make_creds(ns, name)["password"]   # 프로젝트-소스별 비밀번호(Secret 재사용=멱등)
    cname, ip = _ensure_shared(stype)
    if stype == "REDIS":
        dbnum = str(int(hashlib.sha1(unit.encode()).hexdigest(), 16) % 16)   # DB 번호 격리(0~15)
        conn = {"stype": stype, "host": svc, "port": port, "user": "", "password": SHARED_ADMIN, "db": dbnum}
        log(f"source '{name}' (REDIS) → 공유 '{cname}' · DB#{dbnum} · 서비스 {svc}:{port}")
    else:
        _mint(stype, cname, unit, pw)
        conn = {"stype": stype, "host": svc, "port": port, "user": unit, "password": pw, "db": unit}
        log(f"source '{name}' ({stype}) → 공유 '{cname}' · 격리 DB '{unit}' · 서비스 {svc}:{port}")
    labels = {"app.kubernetes.io/managed-by": "yakcloud-dev", "yakcloud.dev/role": "source", "yakcloud.dev/source": svc}
    # no-selector Service + 수동 Endpoints → kind 밖 공유 컨테이너 IP 로 라우팅.
    apply([
        {"apiVersion": "v1", "kind": "Secret",
         "metadata": {"name": f"yak-dev-src-{svc}", "namespace": ns, "labels": labels},
         "stringData": {"password": pw, "unit": unit}},
        {"apiVersion": "v1", "kind": "Service",
         "metadata": {"name": svc, "namespace": ns, "labels": labels},
         "spec": {"ports": [{"port": port, "targetPort": port}]}},
        {"apiVersion": "v1", "kind": "Endpoints",
         "metadata": {"name": svc, "namespace": ns, "labels": labels},
         "subsets": [{"addresses": [{"ip": ip}], "ports": [{"port": port}]}]},
    ])
    return conn


# ── 바인딩 env: 백엔드 _bind_env_for 와 동일(하이픈 alias → 언더스코어 프리픽스 포함) ────────
def bind_env(alias: str, conn: dict) -> dict:
    p = alias.upper().replace("-", "_")   # 백엔드와 동일 — 하이픈 alias 에서도 env 키 일치
    host, port = conn["host"], str(conn["port"])
    user, pw, db = conn["user"], conn["password"], conn["db"]
    qu, qp = quote(user, safe=""), quote(pw, safe="")
    out = {f"{p}_HOST": host, f"{p}_PORT": port}
    st = conn["stype"]
    if st == "MONGODB":
        out.update({f"{p}_USERNAME": user, f"{p}_PASSWORD": pw, f"{p}_DB": db,
                    f"{p}_URL": f"mongodb://{qu}:{qp}@{host}:{port}/{db}?authSource={db}"})
    elif st == "REDIS":
        out.update({f"{p}_USERNAME": "", f"{p}_PASSWORD": pw, f"{p}_DB": db,
                    f"{p}_URL": f"redis://:{qp}@{host}:{port}/{db}"})
    elif st == "MYSQL":
        out.update({f"{p}_USERNAME": user, f"{p}_PASSWORD": pw, f"{p}_DB": db,
                    f"{p}_URL": f"mysql://{qu}:{qp}@{host}:{port}/{db}"})
    elif st == "MARIADB":
        out.update({f"{p}_USERNAME": user, f"{p}_PASSWORD": pw, f"{p}_DB": db,
                    f"{p}_URL": f"mysql://{qu}:{qp}@{host}:{port}/{db}"})
    else:  # POSTGRESQL
        out.update({f"{p}_USERNAME": user, f"{p}_PASSWORD": pw, f"{p}_DB": db,
                    f"{p}_URL": f"postgresql://{qu}:{qp}@{host}:{port}/{db}"})
    return out


# ── 워크로드: 빌드 + kind load + Deployment/Service/Ingress ─────────────────────
def _image_id(image: str) -> str:
    r = sh([RUNTIME, "image", "inspect", "--format", "{{.Id}}", image], check=False)
    return (r.stdout or "").strip() if r.returncode == 0 else ""


def build_and_load(image: str, ctx_dir: str) -> str:
    log(f"이미지 빌드({RUNTIME}) {image}  ← {ctx_dir}")
    sh([RUNTIME, "build", "-t", image, ctx_dir], capture=False)
    log(f"kind load → 클러스터 {KIND_CLUSTER}")
    env = {"KIND_EXPERIMENTAL_PROVIDER": "podman"} if RUNTIME == "podman" else {}
    sh(["kind", "load", "docker-image", image, "--name", KIND_CLUSTER], capture=False, env=env)
    return _image_id(image)


def deploy_workload(project: str, ns: str, w: dict, src_conns: dict) -> tuple[str, bool]:
    image = w["image"].replace("${TAG}", TAG)
    port = int(w.get("port", 8080))
    ex = w.get("expose", {}) or {}
    res = w.get("resources", {}) or {}
    name = sanitize(w["name"])
    img_id = build_and_load(image, w["build"]) if w.get("build") else ""

    # env = 컨테이너 기본 + 워크로드 선언 env + 바인딩 소스 env(직선 파리티)
    env = [{"name": "PORT", "value": str(port)}, {"name": "APP_NAME", "value": project}]
    wenv = w.get("env") or []
    if isinstance(wenv, dict):
        wenv = [{"key": k, "value": v} for k, v in wenv.items()]
    for e in wenv:
        env.append({"name": e["key"], "value": str(e.get("value", ""))})
    for b in w.get("binds", []) or []:
        conn = src_conns.get(b["source"])
        if not conn:
            die(f"bind '{b['alias']}' → source '{b['source']}' 미해결(requires 에 없음).")
        for k, v in bind_env(b["alias"], conn).items():
            env.append({"name": k, "value": v})

    labels = {"app": name, "app.kubernetes.io/managed-by": "yakcloud-dev", "yakcloud.dev/role": "workload"}
    # 고정 태그(dev) 재빌드 시에도 새 이미지가 반영되도록 image-id 를 파드 템플릿 어노테이션에(spec diff 유발).
    pod_meta = {"labels": labels}
    if img_id:
        pod_meta["annotations"] = {"yakcloud.dev/image-id": img_id}
    container = {"name": name, "image": image, "imagePullPolicy": "IfNotPresent",
                 "ports": [{"containerPort": port}], "env": env,
                 "resources": {"requests": {"cpu": res.get("cpu", "25m"), "memory": res.get("mem", "96Mi")}}}
    if w.get("health"):
        container["readinessProbe"] = {"httpGet": {"path": w["health"], "port": port},
                                       "initialDelaySeconds": 3, "periodSeconds": 5, "failureThreshold": 30}
    dep = {"apiVersion": "apps/v1", "kind": "Deployment",
           "metadata": {"name": name, "namespace": ns, "labels": labels},
           "spec": {"replicas": int(w.get("replicas", 1)),
                    "selector": {"matchLabels": {"app": name}},
                    "template": {"metadata": pod_meta, "spec": {"containers": [container]}}}}
    svc = {"apiVersion": "v1", "kind": "Service",
           "metadata": {"name": name, "namespace": ns, "labels": labels},
           "spec": {"selector": {"app": name}, "ports": [{"port": port, "targetPort": port}]}}

    hosts = ex.get("hosts") or ([ex["host"]] if ex.get("host") else [])
    if not hosts:
        hosts = [f"{project}.{HOST_SUFFIX}"]
    path = ex.get("path", "/") or "/"
    rules = [{"host": h, "http": {"paths": [{"path": path, "pathType": "Prefix",
              "backend": {"service": {"name": name, "port": {"number": port}}}}]}} for h in hosts]
    ing = {"apiVersion": "networking.k8s.io/v1", "kind": "Ingress",
           "metadata": {"name": name, "namespace": ns, "labels": labels,
                        "annotations": {"nginx.ingress.kubernetes.io/rewrite-target": "/"} if ex.get("rewrite") else {}},
           "spec": {"ingressClassName": "nginx", "rules": rules}}

    log(f"deploy '{name}' image={image} port={port} replicas={dep['spec']['replicas']} hosts={hosts}")
    apply([dep, svc, ing])
    r = kubectl(["rollout", "status", f"deployment/{name}", "-n", ns, "--timeout=180s"], check=False)
    ok = r.returncode == 0
    if not ok:
        warn(f"'{name}' 롤아웃 실패/지연 — 'kubectl --context {CTX} -n {ns} get pods' 로 확인")
    return hosts[0], ok


# ── 고아 정리: 매니페스트에 없는(라벨된) 소스/워크로드 리소스 삭제(직선 리컨실) ──────────────
def _prune(ns: str, role: str, label_key: str, desired: set) -> None:
    r = kubectl(["get", "deployment,service,endpoints,secret,configmap,ingress", "-n", ns,
                 "-l", f"{MANAGED},yakcloud.dev/role={role}", "-o", "json"], check=False)
    if r.returncode != 0:
        return
    for it in json.loads(r.stdout or "{}").get("items", []):
        val = (it.get("metadata", {}).get("labels", {}) or {}).get(label_key)
        if val is None or val in desired:
            continue
        kind = it["kind"].lower()
        nm = it["metadata"]["name"]
        log(f"고아 정리({role}): {kind}/{nm}")
        kubectl(["delete", kind, nm, "-n", ns, "--ignore-not-found"], check=False)


def _wl_hosts(project: str, w: dict) -> list:
    ex = w.get("expose", {}) or {}
    hosts = ex.get("hosts") or ([ex["host"]] if ex.get("host") else [])
    return hosts or [f"{project}.{HOST_SUFFIX}"]


def _report_to_console(project: str, reqs: list, wls: list) -> None:
    """dev deploy 결과(소스·워크로드 요약)를 콘솔에 best-effort 리포트 — LOCAL 클러스터 데이터소스·앱 탭 렌더용.
    로그인(URL+TOKEN)+등록명(YAK_REG_NAME) 없으면 조용히 스킵(오프라인 dev 불변식)."""
    url = os.environ.get("YAKCLOUD_URL")
    tok = os.environ.get("YAKCLOUD_TOKEN")
    name = os.environ.get("YAK_REG_NAME")
    if not (url and tok and name):
        return
    report = {
        "project": project,
        "sources": [{"name": r["name"], "type": str(r.get("type", "")).upper()} for r in reqs],
        "workloads": [{"name": sanitize(w["name"]), "image": w["image"].replace("${TAG}", TAG),
                       "hosts": _wl_hosts(project, w), "port": int(w.get("port", 8080))} for w in wls],
    }
    body = json.dumps({"name": name, "displayName": "로컬 개발 (kind)", "report": report}).encode()
    req = urllib.request.Request(url.rstrip("/") + "/api/v1/clusters/local", data=body, method="POST",
                                 headers={"Authorization": "Bearer " + tok, "content-type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=6):
            log("콘솔 로컬 클러스터 목록에 데이터소스·앱 리포트 반영")
    except Exception:  # noqa: BLE001 — best-effort(오프라인/미로그인/네트워크)
        pass


def cmd_deploy(m: dict) -> None:
    context_ok()
    project = sanitize(m.get("project", "app"))
    ns = project
    reqs = m.get("requires", []) or []
    wls = m.get("workloads", []) or []

    # 충돌 가드 — 서로 다른 소스명/워크로드명이 같은 서비스명으로 뭉개지면 자격 오배선 → 거부.
    svc_owner: dict[str, str] = {}
    for r in reqs:
        s = sanitize(r["name"])
        if s in svc_owner and svc_owner[s] != r["name"]:
            die(f"소스명 충돌 — '{svc_owner[s]}' 와 '{r['name']}' 가 같은 서비스명 '{s}' 로 뭉개집니다. 이름을 구분하세요.")
        svc_owner[s] = r["name"]
    for w in wls:
        s = sanitize(w["name"])
        if s in svc_owner:
            die(f"워크로드 '{w['name']}' 가 소스 '{svc_owner[s]}' 와 같은 서비스명 '{s}' 로 충돌합니다.")

    log(f"[dev/local] project '{project}' → kind '{KIND_CLUSTER}' (ns {ns}, tag {TAG})")
    ensure_ns(ns)
    src_conns: dict[str, dict] = {}
    for req in reqs:
        conn = provision_source(ns, project, req)
        if conn:
            src_conns[req["name"]] = conn
    results = [deploy_workload(project, ns, w, src_conns) for w in wls]

    # 매니페스트에서 빠진 소스/워크로드 정리(고아 방지 — add-only 가 아니라 리컨실).
    _prune(ns, "source", "yakcloud.dev/source", {sanitize(r["name"]) for r in reqs})
    _prune(ns, "workload", "app", {sanitize(w["name"]) for w in wls})

    _report_to_console(project, reqs, wls)  # 콘솔 LOCAL 클러스터 탭(데이터소스·앱) 리포트

    failed = [h for h, ok in results if not ok]
    print()
    if failed:
        warn(f"일부 워크로드 롤아웃 실패({len(failed)}/{len(results)}) — 아래 상태 확인 후 재배포.")
    else:
        log("✓ 로컬 배포 완료(직선 dev). 접속:")
    first_host = results[0][0] if results else None
    if first_host:
        print(f"    http://{first_host}/            (kind 인그레스 80 포트)")
        print(f"    curl -s http://{first_host}/ | python3 -m json.tool")
        print(f"    ↑ 응답의 bound_sources 에 바인딩 소스 <ALIAS>_URL 키가 보이면 직선 파리티 성립")
    print(f"    kubectl --context {CTX} -n {ns} get deploy,svc,ing,pods")
    print(f"    승격: yakcloud project deploy --local <tag>  (레지스트리 push) → yakcloud project promote --to val")
    if failed:
        sys.exit(1)


def cmd_status(m: dict) -> None:
    context_ok()
    ns = sanitize(m.get("project", "app"))
    print(f"[dev/local] ns={ns} context={CTX}")
    sh(["kubectl", "--context", CTX, "-n", ns, "get", "deploy,svc,ing,pods"], capture=False, check=False)


def cmd_down(m: dict) -> None:
    """네임스페이스(이 프로젝트의 dev 리소스)만 정리 — kind 클러스터 자체는 'yakcloud dev down' 이 지운다."""
    context_ok()
    ns = sanitize(m.get("project", "app"))
    log(f"네임스페이스 '{ns}' 삭제(로컬 dev 리소스 정리)")
    sh(["kubectl", "--context", CTX, "delete", "namespace", ns, "--ignore-not-found"], capture=False, check=False)


def main() -> None:
    if len(sys.argv) < 2 or sys.argv[1] not in ("deploy", "status", "down"):
        die("사용: yakcloud_dev.py <deploy|status|down> [manifest.yaml]")
    cmd = sys.argv[1]
    path = next((a for a in sys.argv[2:] if not a.startswith("--")), "yakcloud.yaml")
    if not os.path.exists(path):
        die(f"매니페스트 없음: {path} — 'yakcloud project init' 먼저")
    m = yaml.safe_load(open(path)) or {}
    {"deploy": cmd_deploy, "status": cmd_status, "down": cmd_down}[cmd](m)


if __name__ == "__main__":
    main()