"""step_run 락에 소유자 신원 + 하트비트 + 정지 요청 칸 추가.

`step_run.status='running'` 은 **누가 잡았는지**도 **그가 살아있는지**도
적지 않는 락이었다. 그래서 시스템은 죽음을 경과 시간(기본 3600초)으로
추측했고, 프로세스를 강제 종료하면 한 시간을 기다려야 회수됐다.

추가하는 칸 (전부 nullable — 기존 행 그대로 동작):

- ``owner_host``: 락을 잡은 프로세스의 호스트 이름
- ``owner_pid``: 그 프로세스의 PID
- ``owner_boot_id``: 프로세스 기동 시 1회 생성하는 uuid4. PID 는 재사용되지만
  이것은 아니다 — 같은 host+pid 로 다른 프로세스가 떠도 구별된다.
- ``heartbeat_at``: 일하는 동안 주기적으로 갱신하는 시각.
  다른 호스트라 PID 를 못 물어볼 때의 판정 근거.
- ``cancel_requested_at``: 협조적 정지 요청 시각. 일하는 쪽이 안전 지점에서
  읽고 깨끗이 멈춘다.

★두 시각 칸은 ``TIMESTAMP WITH TIME ZONE`` 이고 **DB 시계**(``CURRENT_TIMESTAMP``)로
 쓴다. 기존 시각 칸(``started_at`` 등)이 text 인데도 형을 맞추지 않은 이유는
 **계약이 다르기** 때문이다 — 이 둘은 「두 시각의 차가 lease 를 넘었나」를
 재는 값이라, 쓰는 쪽과 읽는 쪽의 시계가 다르면 그 차가 거짓이 된다.
 한 시계로 통일하면 시차·파싱·문자열 비교 문제가 통째로 사라진다.
 ``owner_pid`` 는 ``Integer``, 나머지는 ``Text``.

★인덱스 1개: ``(status, owner_host)`` — 기동 시 자기 락 회수 조회용.

Revision ID: 010_step_lock_ownership
Revises: 009_image_asset_disposition
Create Date: 2026-08-26
"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


revision: str = "010_step_lock_ownership"
down_revision: Union[str, None] = "009_image_asset_disposition"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


_TABLE = "step_run"

_NEW_COLUMNS = (
    ("owner_host", lambda: sa.Column("owner_host", sa.Text(), nullable=True)),
    ("owner_pid", lambda: sa.Column("owner_pid", sa.Integer(), nullable=True)),
    ("owner_boot_id", lambda: sa.Column("owner_boot_id", sa.Text(), nullable=True)),
    (
        "heartbeat_at",
        lambda: sa.Column(
            "heartbeat_at", sa.TIMESTAMP(timezone=True), nullable=True
        ),
    ),
    (
        "cancel_requested_at",
        lambda: sa.Column(
            "cancel_requested_at", sa.TIMESTAMP(timezone=True), nullable=True
        ),
    ),
)

_NEW_INDEXES = (
    ("idx_step_run_status_owner_host", ["status", "owner_host"]),
)


# 주행 단위 정지 요청. `step_run.cancel_requested_at` 은 **스텝 하나**만 세우는데,
# `run_steps_batch` 는 `for sid in step_ids:` 로 도는지라 지금 스텝이 멈춰도
# 다음 스텝으로 넘어간다. 주행 전체를 덮는 표가 따로 필요하다.
_CANCEL_TABLE = "run_cancel_request"


def upgrade() -> None:
    bind = op.get_bind()
    inspector = sa.inspect(bind)

    existing_cols = {col["name"] for col in inspector.get_columns(_TABLE)}
    for name, factory in _NEW_COLUMNS:
        if name not in existing_cols:
            op.add_column(_TABLE, factory())

    existing_idx = {ix["name"] for ix in inspector.get_indexes(_TABLE)}
    for idx_name, cols in _NEW_INDEXES:
        if idx_name not in existing_idx:
            op.create_index(idx_name, _TABLE, cols, if_not_exists=True)

    if _CANCEL_TABLE not in set(inspector.get_table_names()):
        op.create_table(
            _CANCEL_TABLE,
            sa.Column("id", sa.Text(), primary_key=True),
            sa.Column("project_id", sa.Text(), nullable=False),
            sa.Column("episode_id", sa.Text(), nullable=False),
            sa.Column(
                "scope", sa.Text(), nullable=False, server_default="episode"
            ),
            sa.Column(
                "requested_at",
                sa.TIMESTAMP(timezone=True),
                nullable=False,
                server_default=sa.text("CURRENT_TIMESTAMP"),
            ),
            sa.Column("requested_by", sa.Text(), nullable=True),
            sa.Column("reason", sa.Text(), nullable=True),
            sa.Column(
                "cleared_at", sa.TIMESTAMP(timezone=True), nullable=True
            ),
            sa.UniqueConstraint(
                "project_id",
                "episode_id",
                "scope",
                name="run_cancel_request_scope_key",
            ),
        )
        op.create_index(
            "idx_run_cancel_request_active",
            _CANCEL_TABLE,
            ["project_id", "episode_id", "cleared_at"],
            if_not_exists=True,
        )


def downgrade() -> None:
    bind = op.get_bind()
    inspector = sa.inspect(bind)

    if _CANCEL_TABLE in set(inspector.get_table_names()):
        op.drop_table(_CANCEL_TABLE)

    existing_idx = {ix["name"] for ix in inspector.get_indexes(_TABLE)}
    for idx_name, _cols in _NEW_INDEXES:
        if idx_name in existing_idx:
            op.drop_index(idx_name, table_name=_TABLE)

    existing_cols = {col["name"] for col in inspector.get_columns(_TABLE)}
    for name, _factory in reversed(_NEW_COLUMNS):
        if name in existing_cols:
            op.drop_column(_TABLE, name)