"""외부에서 스텝 상태를 갈아 끼우는 자리들이 **토큰까지 돌리는지** 실물 확인.

2026-08-26 Codex 2차 재리뷰 BLOCK-3.

왜 실물 DB 로 재나:
    이 계약은 전부 SQL 한 문장 안에서 성립한다(`CASE WHEN status='running'`,
    `AND status='running'` CAS, `FOR UPDATE`). MagicMock 으로는 그 문장이
    **실제로 무엇을 바꿨는지** 못 본다 — 넘긴 문자열을 다시 읽을 뿐이다.
    「조립하는 자리를 재고 나가는 것을 쟀다고 말하지 마라」가 여기 그대로다.

무엇을 재나:
    ⓐ 도는 중인 하류를 무효화하면 **토큰이 버려진다** — 그래야 그 worker 가
      다음 안전 지점에서 멈춘다. 안 도는 하류는 토큰이 그대로다.
    ⓑ 토큰을 안 돌린 **외부 status 전이**를 recovery 갱신이 잡아낸다.
    ⓒ 복원은 도는 중인 스텝을 거부한다.
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import text as sql_text

pytestmark = pytest.mark.pg


def _seed_step_run(session, *, pid: str, eid: str, sid: str,
                   status: str, run_id: str) -> None:
    session.execute(sql_text(
        "INSERT INTO step_run (id, project_id, episode_id, step_id, status, "
        "  run_id, created_at, updated_at) "
        "VALUES (:id, :pid, :eid, :sid, :status, :rid, NOW(), NOW())"
    ), {"id": str(uuid.uuid4()), "pid": pid, "eid": eid, "sid": sid,
        "status": status, "rid": run_id})
    session.commit()


def _read(session, *, pid: str, eid: str, sid: str):
    return session.execute(sql_text(
        "SELECT status, run_id FROM step_run "
        "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid"
    ), {"pid": pid, "eid": eid, "sid": sid}).fetchone()


# ── ⓐ 무효화가 토큰을 버리는가 ──────────────────────────────────────


def _한_하류를_고른다(ref: str) -> str:
    from app.core.step_catalog import get_all_downstream_recursive

    downstream = get_all_downstream_recursive(ref)
    assert downstream, f"{ref} 에 하류가 없다 — 이 시험이 아무것도 안 잰다"
    return sorted(downstream)[0]


@pytest.mark.parametrize("prior_status,토큰이_바뀌어야_하나", [
    ("running", True),      # 도는 중 → 버려야 그 worker 가 멈춘다
    ("completed", False),   # 안 도는 것 → 건드릴 이유가 없다
])
def test_샷선택이_도는_하류의_토큰을_버린다(
        pg_session, prior_status, 토큰이_바뀌어야_하나):
    """★프로덕션 함수를 그대로 부른다 — SQL 을 베껴 쓰면 구현이 바뀌어도
     이 시험은 계속 초록이다."""
    from app.services.shot_selection_service import ShotSelectionService

    pid, eid = f"fence-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    sid = _한_하류를_고른다("shot_selection")
    rid = f"run-{uuid.uuid4()}"
    _seed_step_run(pg_session, pid=pid, eid=eid, sid=sid,
                   status=prior_status, run_id=rid)

    ShotSelectionService(pg_session, pid, eid)._mark_downstream_stale()
    pg_session.commit()

    row = _read(pg_session, pid=pid, eid=eid, sid=sid)
    assert row.status == "stale"
    if 토큰이_바뀌어야_하나:
        assert row.run_id != rid and row.run_id.startswith("invalidated-"), (
            "도는 중인 하류의 토큰을 안 버렸다 — 그 worker 는 아무것도 "
            "모른 채 계속 돌며 유료 호출을 이어 간다")
    else:
        assert row.run_id == rid, "안 도는 하류의 토큰을 괜히 버렸다"


@pytest.mark.parametrize("prior_status,토큰이_바뀌어야_하나", [
    ("running", True),
    ("completed", False),
])
def test_중앙_무효화도_도는_하류의_토큰을_버린다(
        pg_session, prior_status, 토큰이_바뀌어야_하나):
    """`StepRunner.invalidate_downstream` — 단일 스텝 실행이 지나는 중앙 경로.

    `shot_selection_service` 만 고치고 여기를 빼 두면 같은 좀비가 남는다.
    """
    from app.services.analysis_dispatch_service import get_step_runner

    ref = "shot_selection"
    pid, eid = f"fence-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    sid = _한_하류를_고른다(ref)
    rid = f"run-{uuid.uuid4()}"
    _seed_step_run(pg_session, pid=pid, eid=eid, sid=sid,
                   status=prior_status, run_id=rid)

    runner = get_step_runner(ref, pid, eid, pg_session, {})
    # 파일은 건드리지 않는다 — 여기서 재는 것은 DB 전이뿐이다.
    runner.invalidate_downstream(delete_checkpoints=False)
    pg_session.commit()

    row = _read(pg_session, pid=pid, eid=eid, sid=sid)
    assert row.status == "stale"
    if 토큰이_바뀌어야_하나:
        assert row.run_id != rid and row.run_id.startswith("invalidated-"), (
            "중앙 무효화가 도는 하류의 토큰을 안 버렸다")
    else:
        assert row.run_id == rid


# ── ⓑ 토큰을 안 돌린 외부 status 전이를 잡는가 ─────────────────────


def test_토큰이_같아도_status_가_바뀌었으면_recovery_갱신이_안_먹는다(pg_session):
    """★`run_id` 만 보면 못 잡는 자리다.

    외부가 status 만 'stale' 로 갈아 끼우고 토큰은 그대로 둔 경우, 도는
    worker 가 자기 카운터를 계속 올릴 수 있으면 그 판단이 전부 유령 위에서
    이뤄진다.
    """
    pid, eid = f"fence-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    sid, rid = "scene_detail", f"run-{uuid.uuid4()}"
    _seed_step_run(pg_session, pid=pid, eid=eid, sid=sid,
                   status="running", run_id=rid)

    # 토큰은 그대로 두고 status 만 밖에서 바꾼다.
    pg_session.execute(sql_text(
        "UPDATE step_run SET status = 'stale' "
        "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid"
    ), {"pid": pid, "eid": eid, "sid": sid})
    pg_session.commit()

    # ★프로덕션 함수를 그대로 부른다 — SQL 을 베껴 쓰면 구현이 바뀌어도
    #  이 시험은 계속 초록이다 (2026-08-26 Codex 3차 재리뷰 지적).
    from app.core.errors import AppError
    from app.services.analysis_dispatch_service import get_step_runner

    runner = get_step_runner(sid, pid, eid, pg_session, {})
    runner.run_id = rid          # 이 주행이 잡았던 토큰

    with pytest.raises(AppError) as ei:
        runner._record_recovery("verify failed")
    assert ei.value.code == "step.owner_lost", (
        "status 가 밖에서 바뀌었는데 카운터가 올랐다 — 이 worker 는 "
        "자기가 아직 주인인 줄 안다")


# ── ⓒ 복원이 도는 중인 스텝을 거부하는가 ──────────────────────────


def test_복원은_도는_중인_스텝을_거부한다(pg_session):
    from app.core.errors import AppError
    from app.services.snapshot_service import SnapshotService

    pid, eid = f"fence-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    _seed_step_run(pg_session, pid=pid, eid=eid, sid="scene_detail",
                   status="running", run_id=f"run-{uuid.uuid4()}")

    svc = SnapshotService(pg_session, pid, eid)
    # 체크포인트 디렉토리를 만들어 준다 — 그게 없으면 다른 이유로 막힌다.
    base = svc._base / "scene_detail"
    base.mkdir(parents=True, exist_ok=True)
    (base / "manifest_20260101_000000.json").write_text(
        '{"status": "completed"}', encoding="utf-8")

    with pytest.raises(AppError) as ei:
        svc.restore("20260101_000000")
    assert ei.value.code == "snapshot.step_running"

    # 파일은 그대로여야 한다 — 하나라도 덮었으면 절반만 복원된 것이다.
    assert not (base / "manifest.json").exists(), (
        "실행 중인데 manifest 를 덮었다")


# ── ⓓ Snapshot 복원의 트랜잭션 범위 (Codex 3차 재리뷰 BLOCK-2) ──────


def _스냅샷_씨앗(svc, sid: str, *, status: str = "completed") -> None:
    d = svc._base / sid
    d.mkdir(parents=True, exist_ok=True)
    (d / f"manifest_20260101_000000.json").write_text(
        '{"status": "%s"}' % status, encoding="utf-8")


def test_복원_중에는_다른_세션의_claim_이_막힌다(
        pg_session, pg_engine, monkeypatch):
    """★핵심은 「미리 running 이면 거부」가 아니라 **복원이 도는 동안 새
     claim 이 기다리는가**다. 두 세션으로 직접 잰다.

     복원이 행을 쥔 채 머물게 파일 복사를 붙잡아 두고, 다른 세션이 짧은
     `lock_timeout` 으로 그 행을 바꾸려다 걸리는지 본다.
    """
    import threading

    from sqlalchemy.orm import sessionmaker

    from app.services import snapshot_service as snap
    from app.services.snapshot_service import SnapshotService

    pid, eid = f"snap-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    sid = "scene_detail"
    _seed_step_run(pg_session, pid=pid, eid=eid, sid=sid,
                   status="completed", run_id=f"run-{uuid.uuid4()}")

    svc = SnapshotService(pg_session, pid, eid)
    _스냅샷_씨앗(svc, sid)

    복사중 = threading.Event()
    놓아준다 = threading.Event()
    진짜복사 = snap.shutil.copy2
    한번만 = {"했나": False}

    def _붙잡고_있는_복사(src, dst):
        r = 진짜복사(src, dst)
        if not 한번만["했나"]:
            한번만["했나"] = True
            복사중.set()
            놓아준다.wait(8)      # 복원이 행을 쥔 채 여기 머문다
        return r

    monkeypatch.setattr(snap.shutil, "copy2", _붙잡고_있는_복사)

    결과 = {}

    def _다른_세션이_자리를_잡으려_한다():
        복사중.wait(8)
        other = sessionmaker(bind=pg_engine)()
        try:
            other.execute(sql_text("SET LOCAL lock_timeout = '400ms'"))
            other.execute(sql_text(
                "UPDATE step_run SET status = 'running' "
                "WHERE project_id = :pid AND episode_id = :eid "
                "  AND step_id = :sid"
            ), {"pid": pid, "eid": eid, "sid": sid})
            other.commit()
            결과["막혔나"] = False       # 안 기다리고 그냥 바꿨다
        except Exception as exc:        # noqa: BLE001
            결과["막혔나"] = "lock" in str(exc).lower()
            결과["사유"] = str(exc)[:200]
            other.rollback()
        finally:
            other.close()
            놓아준다.set()

    t = threading.Thread(target=_다른_세션이_자리를_잡으려_한다)
    t.start()
    svc.restore("20260101_000000", step_id=sid)
    t.join(15)

    assert 결과.get("막혔나") is True, (
        f"복원이 도는 동안 다른 세션이 그 행을 그냥 가져갔다 — FOR UPDATE 가 "
        f"실제로 막고 있지 않다 ({결과})")


def test_실행_기록이_없는_대상은_복원을_거부한다(pg_session):
    """FOR UPDATE 는 있는 행만 잠근다 — 없는 대상은 그 틈이 그대로 열린다."""
    from app.core.errors import AppError
    from app.services.snapshot_service import SnapshotService

    pid, eid = f"snap-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    svc = SnapshotService(pg_session, pid, eid)
    _스냅샷_씨앗(svc, "scene_detail")          # 파일만 있고 step_run 행은 없다

    with pytest.raises(AppError) as ei:
        svc.restore("20260101_000000", step_id="scene_detail")
    assert ei.value.code == "snapshot.step_run_missing"
    assert not (svc._base / "scene_detail" / "manifest.json").exists(), (
        "막아야 할 복원인데 파일을 덮었다")


def test_도는_하류가_있으면_복원_자체를_막는다(pg_session):
    """복원 대상은 안 돌아도 **하류가 돌면** 그 주행이 옛 상류를 보고 있다."""
    from app.core.errors import AppError
    from app.services.snapshot_service import SnapshotService

    ref = "shot_selection"
    pid, eid = f"snap-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    하류 = _한_하류를_고른다(ref)
    _seed_step_run(pg_session, pid=pid, eid=eid, sid=ref,
                   status="completed", run_id=f"run-{uuid.uuid4()}")
    _seed_step_run(pg_session, pid=pid, eid=eid, sid=하류,
                   status="running", run_id=f"run-{uuid.uuid4()}")

    svc = SnapshotService(pg_session, pid, eid)
    _스냅샷_씨앗(svc, ref)

    with pytest.raises(AppError) as ei:
        svc.restore("20260101_000000", step_id=ref)
    assert ei.value.code == "snapshot.step_running"
    assert 하류 in ei.value.message
    assert not (svc._base / ref / "manifest.json").exists(), (
        "하류가 도는데 상류 파일을 덮었다")


def test_파일_교체가_실패하면_파일과_DB_를_같이_되돌린다(
        pg_session, monkeypatch):
    """DB 만 rollback 하면 이미 덮은 manifest 가 남아 파일과 DB 가 갈린다."""
    import shutil as _shutil

    from app.services import snapshot_service as snap
    from app.services.snapshot_service import SnapshotService

    pid, eid = f"snap-{uuid.uuid4()}", f"ep-{uuid.uuid4()}"
    for sid in ("scene_save", "scene_detail"):
        _seed_step_run(pg_session, pid=pid, eid=eid, sid=sid,
                       status="completed", run_id=f"run-{uuid.uuid4()}")

    svc = SnapshotService(pg_session, pid, eid)
    for sid in ("scene_save", "scene_detail"):
        _스냅샷_씨앗(svc, sid)
        (svc._base / sid / "manifest.json").write_text(
            '{"status": "completed", "mark": "원본"}', encoding="utf-8")

    진짜복사 = _shutil.copy2
    횟수 = {"n": 0}

    def _세번째에_터진다(src, dst):
        # ★되돌릴 때는 통과시킨다 — 되돌리기까지 막으면 재는 것이
        #  「되돌리기가 되나」가 아니라 「두 번 연속 실패하면 어떻게 되나」다.
        if 횟수.get("터졌다"):
            return 진짜복사(src, dst)
        횟수["n"] += 1
        if 횟수["n"] >= 3:          # 첫 스텝은 넘기고 두 번째에서 터뜨린다
            횟수["터졌다"] = True
            raise OSError("디스크가 꽉 찼다")
        return 진짜복사(src, dst)

    monkeypatch.setattr(snap.shutil, "copy2", _세번째에_터진다)

    with pytest.raises(OSError):
        svc.restore("20260101_000000")

    for sid in ("scene_save", "scene_detail"):
        본문 = (svc._base / sid / "manifest.json").read_text(encoding="utf-8")
        assert "원본" in 본문, (
            f"{sid} 의 manifest 가 되돌아가지 않았다 — 파일과 DB 가 갈렸다")