"""canary 를 **원본에서 떼어 놓는 자물쇠**. ★원본에 한 글자도 안 쓴다.

Codex 가 못박았다 (2026-08-31) —

> 같은 theroad DB 에 canary project_id 를 넣지 마십시오. **별도 PostgreSQL
> database** 를 같은 cluster 에 run-id 이름으로 만들고 현재 migration 을
> 적용하십시오. SQLite 는 partial index·Postgres 동작이 달라 production
> canary 가 아닙니다. 별도 schema/search_path 도 누락 쿼리가 public 을 볼
> 위험이 있어 차선입니다.

## 문 넷 — ★`app.core.database` 를 **부르기 전에** 선다

    ①`DATABASE_URL` 이 원본과 다르고 DB 이름에 run_id 가 있다
    ②`PROJECTS_DIR` 이 원본과 다르고 그 경로에 run_id 가 있다
    ③산출·장부도 같은 run_id 뿌리 아래다
    ④Opik tag/thread 도 canary 전용이다

★`SessionLocal` 은 **import 시점**에 만들어진다(`app/core/database.py`).
그래서 이 자물쇠는 **그 import 보다 먼저** 걸려야 한다.

## ★지우지 않는다

주행 뒤 DB 를 **그대로 둔다** — Opik·로그·체크포인트와 대조해야 한다.
지우는 것은 **따로 명시한 작업**으로만 한다. 이 파일에 drop 은 없다.
"""
from __future__ import annotations

import os
import re
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import urlparse

#: 저장소 뿌리. ★`artifact/` · `projects/` 는 **여기** 아래다.
REPO = Path(__file__).resolve().parents[3]

#: 원본. ★이것과 **같으면 선다**.
PROD_DB_NAME = "theroad"
#: canary DB 이름의 꼴. ★run_id 가 **이름 안에** 있어야 한다.
DB_PREFIX = "theroad_canary_"
#: run_id 로 받는 꼴 — 짧은 16진수. ★사람이 아무 글자나 못 넣게.
RUN_ID_RE = re.compile(r"^[0-9a-f]{8,32}$")

TRACE_TAG = "op:bundle-canary-fixture"
TRACE_THREAD = "bundle-canary-fixture"


class IsolationRefused(RuntimeError):
    """격리가 안 됐다. ★DB 도 안 열고 파일도 안 만든다."""


def _need(run_id: str) -> str:
    if not RUN_ID_RE.match(str(run_id or "")):
        raise IsolationRefused(
            f"run_id 가 꼴에 안 맞는다: {run_id!r} — 8~32자리 16진수여야 한다")
    return str(run_id)


def db_name(run_id: str) -> str:
    return f"{DB_PREFIX}{_need(run_id)}"


def db_url(run_id: str, *, template: Optional[str] = None) -> str:
    """canary DB 의 접속 주소. ★**접속 정보를 지어내지 않는다**.

    ★앞 판은 기본값에 사용자·비밀번호를 **박아 뒀다**(Codex BLOCK 4). 지금은
    `THEROAD_CANARY_TEMPLATE_URL` 이나 넘겨준 template 이 **없으면 선다**.

    ★조립은 `rpartition('/')` 이 아니라 **SQLAlchemy URL parser** 로 한다 —
    query·escaping 을 보존해야 한다.
    """
    from sqlalchemy.engine import make_url

    # ★본은 `template_url()` 이 정한다 — 없으면 `.env` 에서 파생하고,
    #  그것도 없으면 선다. ★여기에 접속 정보를 박아 두지 않는다.
    # ★★`str(URL)` 은 **비밀번호를 `***` 로 가린다**(SQLAlchemy 기본).
    #  그대로 넘기면 접속이 「인증 실패」로 죽는다 — 실측 2026-08-31.
    #  값을 **찍지는 않되** 넘길 때는 그대로 넘긴다.
    return make_url(template_url(explicit=template)).set(
        database=db_name(run_id)).render_as_string(hide_password=False)


def root_dir(run_id: str, *, base: Optional[Path] = None) -> Path:
    """이 판의 **모든 것**이 사는 자리 — 체크포인트·산출·장부."""
    b = Path(base) if base is not None else Path(
        os.environ.get("THEROAD_CANARY_ROOT") or (REPO / "artifact"))
    return b / f"canary_{_need(run_id)}"


def assert_isolated(run_id: str, *, env: Optional[Dict[str, str]] = None,
                    prod_projects_dir: Optional[str] = None) -> Dict[str, Any]:
    """★**모든 것보다 먼저** 선다. 하나라도 어긋나면 아무것도 안 만든다."""
    e = dict(env if env is not None else os.environ)
    rid = _need(run_id)

    url = e.get("DATABASE_URL") or ""
    if not url:
        raise IsolationRefused("`DATABASE_URL` 이 없다 — 원본으로 갈 뻔했다")
    name = (urlparse(url).path or "").lstrip("/")
    if not name:
        raise IsolationRefused(
            f"`DATABASE_URL` 에 DB 이름이 없다: {masked(url)!r}")
    if name == PROD_DB_NAME:
        raise IsolationRefused(
            f"`DATABASE_URL` 이 **원본 DB**({PROD_DB_NAME}) 를 가리킨다 — 선다")
    if rid not in name:
        raise IsolationRefused(
            f"DB 이름 {name!r} 에 run_id {rid!r} 가 없다 — 어느 판인지 "
            "못 되짚는다")
    if name != db_name(rid):
        raise IsolationRefused(
            f"DB 이름이 약속과 다르다: {name!r} ≠ {db_name(rid)!r}")

    pdir = e.get("PROJECTS_DIR") or ""
    if not pdir:
        raise IsolationRefused("`PROJECTS_DIR` 이 없다 — 원본에 쓸 뻔했다")
    p = Path(pdir).resolve()
    if rid not in str(p):
        raise IsolationRefused(
            f"`PROJECTS_DIR` {p} 에 run_id 가 없다")
    prod = Path(prod_projects_dir
                or os.environ.get("THEROAD_PROD_PROJECTS_DIR")
                or (REPO / "projects")).resolve()
    if p == prod or prod in p.parents:
        raise IsolationRefused(
            f"`PROJECTS_DIR` {p} 이 **원본 자리**({prod}) 안이다 — 선다")

    root = root_dir(rid)
    if root not in p.parents and p != root:
        raise IsolationRefused(
            f"`PROJECTS_DIR` {p} 이 이 판의 뿌리({root}) 밖이다")
    # ★★공개 반환에 **접속 주소 전체를 안 넣는다** (Codex 2026-08-31).
    #  나중에 이 결과를 저장하면 비밀이 샌다. 비밀 아닌 좌표만 남긴다.
    return {"run_id": rid, "db_name": name, "database_url_masked": masked(url),
            "projects_dir": str(p), "root": str(root),
            "trace_tag": TRACE_TAG, "trace_thread": TRACE_THREAD,
            "prod_db_name": PROD_DB_NAME, "prod_projects_dir": str(prod),
            "ok": True}


def assert_database_module_not_loaded() -> None:
    """★`app.core.database` 는 **import 때** `SessionLocal` 을 만든다.

    자물쇠보다 먼저 불려 있으면 **원본으로 이미 이어져 있다**.
    """
    import sys

    if "app.core.database" in sys.modules:
        raise IsolationRefused(
            "`app.core.database` 가 이미 올라와 있다 — `SessionLocal` 이 "
            "원본 URL 로 만들어졌을 수 있다. 자물쇠를 **먼저** 걸어야 한다")


def assert_app_modules_not_loaded() -> Dict[str, Any]:
    """★env 를 박은 **직후** — `app.core.config`·`app.core.database` 가 아직
    안 올라와 있어야 한다. 이미 올라와 있으면 settings/엔진은 **그때 URL** 이다.

    실측 2026-09-02: `canary_run` import 가 배경 술어를 부르며 config 를 올렸고,
    그 뒤 env 를 갈아도 settings 는 원본 URL 이었다. 그 상태로 database 가
    올라와 원본에 붙었다.
    """
    import sys

    loaded = [m for m in ("app.core.config", "app.core.database")
              if m in sys.modules]
    if loaded:
        raise IsolationRefused(
            f"{loaded} 가 env 보다 먼저 올라왔다 — settings/엔진이 그때 URL 로 "
            "굳었다. env 를 박은 뒤 **처음** import 해야 한다")
    return {"checked": True, "loaded_before_env": []}


def _db_identity(url: str) -> Dict[str, str]:
    """비교용 정규화 — host·port·database 만 (비밀번호 마스킹 문자열 비교 금지)."""
    from sqlalchemy.engine import make_url

    u = make_url(str(url))
    return {"host": str(u.host or "").lower(), "port": str(u.port or ""),
            "database": str(u.database or "")}


def assert_settings_follow_env(run_id: str) -> Dict[str, Any]:
    """`app.core.config` 가 이미 올라와 있으면 그 settings 가 **이 판의 env** 를
    보고 있어야 한다. ★env 를 나중에 바꿔도 settings 는 안 따라온다.

    실측 2026-09-02: `scenario()` 가 settings 를 먼저 올려 원본 URL 로 굳었고,
    뒤에 올라온 `app.core.database` 가 원본에 붙었다.
    """
    import sys
    from sqlalchemy.engine import make_url

    want = str(make_url(db_url(run_id)).database or "")
    mod = sys.modules.get("app.core.config")
    if mod is None:
        return {"checked": False, "why": "settings 가 아직 안 올라왔다 — env 가 먼저다",
                "database": want}
    actual = str(make_url(str(mod.settings.database_url)).database or "")
    if actual != want:
        raise IsolationRefused(
            f"settings 가 env 보다 먼저 올라와 {actual!r} 를 본다 — canary "
            f"{want!r} 가 아니다. env 를 **settings 보다 먼저** 박아라")
    return {"checked": True, "database": actual}


def assert_engine_is_canary(run_id: str) -> Dict[str, Any]:
    """★**실제 엔진**이 canary DB 에 붙어 있나 — env 가 아니라 세 곳을 본다:
    settings 의 URL · `engine.url` · `SessionLocal` 의 bind (Codex 조건 3).
    host·port·database 를 **정규화해** 견준다.

    `app.core.database` 를 **여기서** 올린다(env 가 이미 이 판 것이어야 한다).
    이미 올라와 있었다면 그때 URL 로 굳은 것이니 그것을 본다 — 원본이면 선다.
    실측 2026-09-02: 격리 문 뒤의 import 가 원본에 붙은 세션을 만들었고,
    부트스트랩의 거절이 아니었으면 pipeline 이 원본으로 돌 뻔했다.
    """
    want = _db_identity(db_url(run_id))
    from app.core.config import settings
    import app.core.database as dbm

    seen = {
        "settings": _db_identity(settings.database_url),
        "engine": _db_identity(str(dbm.engine.url)),
        "session_bind": _db_identity(str(dbm.SessionLocal.kw["bind"].url)),
    }
    bad = {k: v for k, v in seen.items() if v != want}
    if bad:
        raise IsolationRefused(
            f"canary {want['database']!r}@{want['host']} 가 아닌 곳에 붙어 있다 — "
            f"{ {k: (v['database'], v['host']) for k, v in bad.items()} }. "
            "env 뒤에 처음 import 됐는지 보라")
    return {"checked": True, "database": want["database"],
            "host": want["host"], "compared": sorted(seen)}


def create_database(run_id: str, *, admin_dsn: Optional[str] = None,
                    maintenance_db: str = "postgres") -> Dict[str, Any]:
    """canary DB 를 **만든다**. ★더하기만 한다 — 지우는 길은 여기 없다.

    Args:
        admin_dsn: 관리자 연결. ★**명시해야 한다** — 없으면 선다.
            그리고 그 연결이 **정말 maintenance DB** 인지
            `current_database()` 로 확인한 뒤에만 만든다.

    ★★앞 판은 이 함수가 문을 하나도 안 지나고, 본이 없으면 **원본 theroad 에
    붙은 채** `CREATE DATABASE` 를 냈다 (Codex BLOCK 4). 원본에 붙는 길을
    아예 막는다.

    ★이미 있으면 **그대로 쓴다**. 덮어쓰거나 지우지 않는다.
    """
    import psycopg2
    from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
    from sqlalchemy.engine import make_url

    rid = _need(run_id)
    # ★사람에게 묻지 않는다 — `.env` 의 본에서 database 만 바꿔 파생한다
    dsn = maintenance_dsn(explicit=admin_dsn, maintenance_db=maintenance_db)
    want = str(make_url(dsn).database or "")
    if want == PROD_DB_NAME:
        raise IsolationRefused(
            f"관리자 연결이 **원본 DB**({PROD_DB_NAME}) 를 가리킨다 — 선다")
    if want != maintenance_db:
        raise IsolationRefused(
            f"관리자 연결은 maintenance DB({maintenance_db!r}) 여야 한다 — "
            f"지금 {want!r}")

    con = psycopg2.connect(dsn)
    try:
        con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
        cur = con.cursor()
        # ★★붙고 나서 **정말 그 DB 인지** 다시 본다 — DSN 이 거짓일 수 있다
        cur.execute("SELECT current_database()")
        now = str(cur.fetchone()[0])
        if now != maintenance_db:
            raise IsolationRefused(
                f"붙고 보니 {now!r} 다 — maintenance DB 가 아니면 안 만든다")
        cur.execute("SELECT 1 FROM pg_database WHERE datname = %s",
                    (db_name(rid),))
        existed = cur.fetchone() is not None
        if not existed:
            # ★이름은 `RUN_ID_RE` 로 이미 좁혀 놨다
            cur.execute(f'CREATE DATABASE "{db_name(rid)}"')
        cur.close()
    finally:
        con.close()
    return {"db_name": db_name(rid), "already_existed": existed,
            "admin_connected_to": maintenance_db}


# ─────────────────────────────────────────────────────────────────────
# ★★★Alembic 도 같은 문을 지난다 (Codex ㉢, 2026-08-31)
#
# > fresh subprocess + allowlist env 로 실행하고, import 전에 같은 isolation
# > gate 를 태우십시오. `env.py` 가 실제 DATABASE_URL 을 다시 덮지 않는지
# > 끝점에서 잡고, migration connection 안에서
# > `current_database()==theroad_canary_<run_id>` 를 확인한 뒤에만 upgrade.
# > 완료 후 `alembic_version==head` 확인.
# ─────────────────────────────────────────────────────────────────────

#: ★자식 프로세스에 **넘길 것만** 넘긴다. 원본 URL 이 새면 원본으로 간다.
ENV_ALLOWLIST = ("PATH", "HOME", "LANG", "LC_ALL", "TZ",
                 "VIRTUAL_ENV", "PYTHONPATH", "PYTHONUNBUFFERED",
                 "DATABASE_URL", "PROJECTS_DIR",
                 # ★`alembic/env.py` 가 **이 값이 있을 때만** 문을 연다.
                 #  없으면 기존 migration 동작 그대로다.
                 "THEROAD_CANARY_RUN_ID")

#: ★**절대 안 넘기는 것** — 넘기면 자식이 원본을 볼 수 있다.
ENV_DENY = ("THEROAD_PROD_DATABASE_URL", "THEROAD_PROD_PROJECTS_DIR",
            "THEROAD_CANARY_ADMIN_DSN", "THEROAD_CANARY_TEMPLATE_URL")


def child_env(run_id: str, *, env: Dict[str, str]) -> Dict[str, str]:
    """자식에게 줄 환경. ★허용 목록 **밖은 안 준다**."""
    got = assert_isolated(run_id, env=env)
    out = {k: env[k] for k in ENV_ALLOWLIST if k in env}
    leaked = [k for k in ENV_DENY if k in out]
    if leaked:
        raise IsolationRefused(f"자식에게 새면 안 되는 것이 있다: {leaked}")
    # ★입력 env 끼리 견준다 — 공개 반환에 비밀을 넣지 않기 위해서다
    if out.get("DATABASE_URL") != env.get("DATABASE_URL"):
        raise IsolationRefused("자식의 `DATABASE_URL` 이 문을 지난 것과 다르다")
    return out


def assert_migration_connection(run_id: str, *, current_database: str) -> None:
    """★migration 연결 **안에서** 다시 본다 — `env.py` 가 덮었을 수 있다."""
    want = db_name(run_id)
    if str(current_database) != want:
        raise IsolationRefused(
            f"migration 이 붙은 DB 가 {current_database!r} 다 — {want!r} 여야 "
            "한다. `env.py` 가 `DATABASE_URL` 을 덮었는지 본다. upgrade 안 한다")


def assert_at_head(*, alembic_version: str, head: str) -> None:
    """올린 뒤 **끝까지 갔나**. ★반쯤 올린 판으로 안 간다."""
    if not alembic_version or str(alembic_version) != str(head):
        raise IsolationRefused(
            f"alembic_version 이 {alembic_version!r} 인데 head 는 {head!r} 다 — "
            "반쯤 올린 schema 로는 주행하지 않는다")


def upgrade_command(run_id: str) -> List[str]:
    """★별도 프로세스로 돈다 — 부모의 import 상태를 물려받지 않는다.

    ★`alembic upgrade head` 를 **바로** 부르지 않는다. 그러면 확인 함수들이
    **아무 데서도 안 불린다** (Codex 2026-08-31: 「만들고 안 읽음」).
    `canary_alembic` 이 문을 지나고, 올린 **뒤** canary DB 에서 결과를 본다.
    """
    _need(run_id)
    return [sys.executable,
            str(Path(__file__).resolve().parent / "canary_alembic.py"),
            run_id]


def masked(url: str) -> str:
    """접속 주소에서 **비밀번호를 지운다**. ★사람에게 보이는 자리엔 이것만.

    ★★넘길 때는 그대로 넘기지만(`render_as_string(hide_password=False)`),
    **찍을 때는 반드시 이것**을 쓴다 — 예외 문구도 사람이 보는 자리다.
    """
    try:
        from sqlalchemy.engine import make_url

        return str(make_url(str(url)))      # ★`str()` 이 가려 준다
    except Exception:                       # noqa: BLE001
        import re

        return re.sub(r"(://[^:/@]+:)[^@]*(@)", r"\1***\2", str(url))


def _dotenv_database_url() -> Optional[str]:
    """`backend/.env` 의 `DATABASE_URL`. ★값을 **출력하지 않는다**."""
    p = REPO / "backend" / ".env"
    if not p.is_file():
        return None
    for line in p.read_text(encoding="utf-8", errors="replace").splitlines():
        s = line.strip()
        if s.startswith("DATABASE_URL="):
            return s.split("=", 1)[1].strip().strip('"').strip("'") or None
    return None


def template_url(*, explicit: Optional[str] = None) -> str:
    """접속 정보의 **본**. ★사람에게 묻기 전에 `.env` 에서 파생한다.

    ★값을 로그·산출물에 **찍지 않는다** — 여기서 만들어 바로 쓴다.
    """
    got = (explicit or os.environ.get("THEROAD_CANARY_TEMPLATE_URL")
           or _dotenv_database_url())
    if not got:
        raise IsolationRefused(
            "접속 주소의 본이 없다 — `backend/.env` 의 `DATABASE_URL` 도 "
            "`THEROAD_CANARY_TEMPLATE_URL` 도 못 찾았다")
    return got


def maintenance_dsn(*, explicit: Optional[str] = None,
                    maintenance_db: str = "postgres") -> str:
    """관리자 연결. ★본에서 **database 만** 바꾼다 — 새로 짓지 않는다.

    ★사람에게 묻지 않는다. 실제 연결이 **그때 실패할 때만** 묻는다.
    """
    from sqlalchemy.engine import make_url

    got = (explicit or os.environ.get("THEROAD_CANARY_ADMIN_DSN")
           or make_url(template_url()).set(
               database=maintenance_db).render_as_string(hide_password=False))
    name = str(make_url(got).database or "")
    if name == PROD_DB_NAME:
        raise IsolationRefused(
            f"관리자 연결이 **원본 DB**({PROD_DB_NAME}) 를 가리킨다 — 선다")
    return got


def masked_text(text: str) -> str:
    """글 **안에 섞인** 접속 주소의 비밀번호를 지운다.

    ★alembic 같은 바깥 프로그램의 출력에는 주소가 통째로 섞여 나온다.
    저장하기 전에 반드시 지난다.
    """
    import re

    return re.sub(r"([a-zA-Z0-9+]+://[^:/@\s]+:)[^@\s]*(@)", r"\1***\2",
                  str(text or ""))