import logging from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker, DeclarativeBase from app.core.config import settings _logger = logging.getLogger(__name__) class Base(DeclarativeBase): """Single base for ALL models.""" pass def get_engine(): return create_engine(settings.database_url, pool_pre_ping=True, pool_size=10, max_overflow=20, pool_timeout=30) engine = get_engine() SessionLocal = sessionmaker(bind=engine) def register_models() -> None: """SQLAlchemy Base.metadata 에 model 등록만 — lock-free / idempotent. 호출 시점: - dispatch / 임의 script / runtime entry point 가 ORM 사용 전 호출 가능 - init_db() 내부에서도 호출 - DDL/DML 실행하지 않음 → ALTER TABLE / mass UPDATE lock chain 위험 0 FK relation 해소 (예: episode.project_id → project_registry.id) 필수. Codex iter (root cause): init_db() 가 runtime DDL/DML 함수라 dispatch script 가 호출 시 lock chain 발생 → register_models() 로 분리. """ from app.models.catalog import UserAccount, Session, ProjectRegistry, ProjectMember # noqa: F401 from app.models.project import ( # noqa: F401 Episode, EntityCanon, EntityAlias, RelationFact, RelationParticipant, SceneStill, EntityEpisodeLink, ImageAsset, ProjectSettings, WorldGuide, WebbookPackage, GenerationTrace, OperationLog, PipelineProgress, LLMCallLog, CharacterOutlook, ScenePlan, ) from app.logging.models import ActivityLog # noqa: F401 def init_db() -> None: """Startup-only: model 등록 + Base.metadata.create_all + raw SQL _migrations. runtime entry point (dispatch / script) 에서는 호출 금지 — _migrations 의 ALTER TABLE / mass UPDATE 가 ACCESS EXCLUSIVE lock 시도 → backend uvicorn 의 active transaction 과 lock chain. 운영에서는 alembic 으로 schema 이관 진행 중 (legacy _migrations block 은 점진 deprecate). 호출 진입점: backend FastAPI lifespan startup, test conftest, dev/init. runtime script 는 register_models() 만 사용. """ # 1. model 등록 (lock-free) register_models() # 2. table 생성 (idempotent — checkfirst=True 로 already-exists 빠르게 통과) Base.metadata.create_all(engine, checkfirst=True) # 스키마 마이그레이션 — 기존 테이블에 새 컬럼 추가 (없으면) _migrations = [ # step_run 테이블은 SQLAlchemy ORM 외부에서 raw SQL로 다룬다 (model 미정의). # production PG에는 이미 존재하므로 IF NOT EXISTS로 idempotent. test PG DB # (theroad_test)에서는 init_db 시점에 만들어진다. 2026-04-27 hotfix. """CREATE TABLE IF NOT EXISTS step_run ( id text PRIMARY KEY, project_id text NOT NULL, episode_id text NOT NULL, step_id text NOT NULL, status text NOT NULL DEFAULT 'pending', run_id text, resolved_model text, input_hash text, upstream_revision text, prompt_version text, applicable_count integer, completed_count integer DEFAULT 0, failed_count integer DEFAULT 0, error_message text, result_summary text, started_at text, completed_at text, created_at text NOT NULL, updated_at text NOT NULL, sync_status text, sync_error text, synced_at text, recovery_count integer NOT NULL DEFAULT 0, last_recovery_reason text, owner_host text, owner_pid integer, owner_boot_id text, heartbeat_at timestamptz, cancel_requested_at timestamptz, CONSTRAINT step_run_project_id_episode_id_step_id_key UNIQUE(project_id, episode_id, step_id) )""", # 락 소유자 신원 + 하트비트 + 정지 요청 (alembic 010 과 동일, idempotent). # 락이 프로세스의 죽음을 경과 시간으로 추측하던 구조를 신원 확인으로 # 바꾼다 — app/core/step_lock.py 머리말 참조. "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS owner_host TEXT", "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS owner_pid INTEGER", "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS owner_boot_id TEXT", # ★두 시각 칸만 timestamptz 다 (나머지 시각 칸은 text). 「두 시각의 차가 # lease 를 넘었나」를 재는 계약이라 **한 시계**(DB)로 통일해야 한다. "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS heartbeat_at TIMESTAMPTZ", "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS cancel_requested_at TIMESTAMPTZ", "CREATE INDEX IF NOT EXISTS idx_step_run_status_owner_host " "ON step_run (status, owner_host)", # 주행 단위 정지 요청 (alembic 010 과 동일, idempotent). # step_run 의 정지 표식은 **스텝 하나**만 세운다 — run_steps_batch 가 # 다음 sid 로 넘어가는 것을 못 막아 주행 전체를 덮는 표가 따로 필요하다. # app/core/run_control.py 머리말 참조. """CREATE TABLE IF NOT EXISTS run_cancel_request ( id text PRIMARY KEY, project_id text NOT NULL, episode_id text NOT NULL, scope text NOT NULL DEFAULT 'episode', requested_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP, requested_by text, reason text, cleared_at timestamptz, CONSTRAINT run_cancel_request_scope_key UNIQUE(project_id, episode_id, scope) )""", "CREATE INDEX IF NOT EXISTS idx_run_cancel_request_active " "ON run_cancel_request (project_id, episode_id, cleared_at)", "ALTER TABLE project_registry ADD COLUMN IF NOT EXISTS name_en TEXT", "ALTER TABLE entity_canon ADD COLUMN IF NOT EXISTS t2i_prompt TEXT", "ALTER TABLE project_settings ADD COLUMN IF NOT EXISTS style_rules_json TEXT", "ALTER TABLE project_settings ADD COLUMN IF NOT EXISTS world_summary TEXT", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS dependent_scene_id TEXT", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS shot_type_1 TEXT", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS shot_type_2 TEXT", # shot_type 테이블 (촬영 기법) """CREATE TABLE IF NOT EXISTS shot_type ( id SERIAL PRIMARY KEY, name VARCHAR(50) NOT NULL UNIQUE, category VARCHAR(30) NOT NULL, intent VARCHAR(30) NOT NULL, description TEXT NOT NULL, llm_description TEXT NOT NULL, is_active BOOLEAN DEFAULT true, sort_order INTEGER DEFAULT 0 )""", "ALTER TABLE image_asset ADD COLUMN IF NOT EXISTS theme_label TEXT", # Indexes for common query patterns "CREATE INDEX IF NOT EXISTS idx_scene_still_project_episode ON scene_still (project_id, episode_id)", "CREATE INDEX IF NOT EXISTS idx_image_asset_entity_type ON image_asset (entity_id, asset_type)", "CREATE INDEX IF NOT EXISTS idx_image_asset_still ON image_asset (still_id)", "CREATE INDEX IF NOT EXISTS idx_image_asset_project_episode ON image_asset (project_id, episode_id, asset_type)", "CREATE INDEX IF NOT EXISTS idx_entity_episode_link_canon ON entity_episode_link (canon_id, episode_id)", "CREATE INDEX IF NOT EXISTS idx_pipeline_progress_project ON pipeline_progress (project_id, episode_id, operation)", "CREATE INDEX IF NOT EXISTS idx_llm_call_log_project ON llm_call_log (project_id, created_at)", # Phase 4 iter 7 W3+I1 — image step trace_meta (scene/shot/still/entity) # 보존을 위한 free-form JSON column. alembic 006 과 동일 (idempotent). "ALTER TABLE llm_call_log ADD COLUMN IF NOT EXISTS metadata_json TEXT", "CREATE INDEX IF NOT EXISTS idx_activity_log_project ON activity_log (project_id, created_at)", "CREATE INDEX IF NOT EXISTS idx_character_outlook_pair ON character_outlook (character_id, outlook_id)", "CREATE INDEX IF NOT EXISTS idx_image_asset_variant ON image_asset (still_id, asset_type, variant_type)", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS scene_type TEXT DEFAULT 'normal'", # Pipeline v3 — scene_still 확장 "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS scene_summary TEXT", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS audio_entity_ids TEXT DEFAULT '[]'", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS hallucination_entity_ids TEXT DEFAULT '[]'", "ALTER TABLE world_guide ADD COLUMN IF NOT EXISTS source_hash TEXT", # v4: shot 기반 파이프라인 "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS shot_index INTEGER", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS shot_description TEXT", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS based_on_beat INTEGER", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS scene_index INTEGER", "ALTER TABLE image_asset ADD COLUMN IF NOT EXISTS shot_index INTEGER", # shot-more: 선택/비선택 샷 구분 + 이미지 생성 여부 + stale 상태 "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS is_selected BOOLEAN DEFAULT TRUE", "ALTER TABLE scene_still ADD COLUMN IF NOT EXISTS image_generated BOOLEAN DEFAULT FALSE", "UPDATE scene_still SET is_selected = TRUE WHERE is_selected IS NULL", "UPDATE scene_still SET image_generated = EXISTS (SELECT 1 FROM image_asset ia WHERE ia.still_id = scene_still.id) WHERE image_generated IS NULL OR image_generated = FALSE", # short_id 체계 "ALTER TABLE entity_canon ADD COLUMN IF NOT EXISTS short_id TEXT", "CREATE UNIQUE INDEX IF NOT EXISTS uq_entity_canon_short_id ON entity_canon(project_id, short_id) WHERE short_id IS NOT NULL", # 기존 엔티티에 short_id 자동 발급 (mixed-state 안전 — 기존 최대값부터 시작) # ★★★번호는 **끝의 숫자**로 뽑는다 (실측 2026-09-01). 앞에는 # `SUBSTRING(short_id FROM 2)` 로 **한 글자만** 뗐는데, 접두가 두 글자인 # `LP01` 이 오면 `'P01'` 이 되어 정수 변환이 터진다 — 그리고 이 이관은 # startup 에서 **fail-fast** 라 백엔드가 통째로 안 뜬다. # ★`CASE` 에 `location_part` 가 없던 것도 같이 고쳤다. 없으면 그 갈래는 # 조용히 `short_id = NULL` 로 남는다. """ WITH max_nums AS ( SELECT project_id, entity_type, COALESCE(MAX(CAST(SUBSTRING(short_id FROM '[0-9]+$') AS INTEGER)), 0) as max_num FROM entity_canon WHERE short_id IS NOT NULL GROUP BY project_id, entity_type ), ranked AS ( SELECT ec.id, ec.entity_type, ec.project_id, ROW_NUMBER() OVER (PARTITION BY ec.project_id, ec.entity_type ORDER BY ec.created_at) + COALESCE(mn.max_num, 0) as rn FROM entity_canon ec LEFT JOIN max_nums mn ON ec.project_id = mn.project_id AND ec.entity_type = mn.entity_type WHERE ec.short_id IS NULL ) UPDATE entity_canon SET short_id = CASE ranked.entity_type WHEN 'character' THEN 'C' || LPAD(ranked.rn::text, 2, '0') WHEN 'location' THEN 'L' || LPAD(ranked.rn::text, 2, '0') WHEN 'location_part' THEN 'LP' || LPAD(ranked.rn::text, 2, '0') WHEN 'prop' THEN 'P' || LPAD(ranked.rn::text, 2, '0') WHEN 'outlook' THEN 'O' || LPAD(ranked.rn::text, 2, '0') END FROM ranked WHERE entity_canon.id = ranked.id """, # T2I 출현 횟수 "ALTER TABLE entity_episode_link ADD COLUMN IF NOT EXISTS t2i_appearance_count INTEGER DEFAULT 0", # 기획서 (planning document) "ALTER TABLE project_registry ADD COLUMN IF NOT EXISTS planning_doc_text TEXT", # W4 P3-2: sync_status 관측성 — checkpoint 성공 vs DB projection 실패 구별. # nullable이므로 기존 row 영향 없음 (PG 11+ 즉시 반영). "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS sync_status TEXT", "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS sync_error TEXT", "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS synced_at TEXT", # Resume integrity: recovery 추적 (alembic 003과 동일, idempotent) "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS recovery_count INTEGER NOT NULL DEFAULT 0", "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS last_recovery_reason TEXT", # variant_label String → String(255) — chain_bg group_id 등 LLM 생성 라벨 수용 (alembic 004). # 이전 정의(String(32))를 startup 시 강제 축소하면 alembic 004를 무효화하고 35자+ row의 # 기존 데이터를 잘라먹는다. 항상 ≥255로 expand만 하도록 한다 (idempotent). "ALTER TABLE image_asset ALTER COLUMN variant_label TYPE VARCHAR(255)", # image_asset.file_path 절대 경로 차단 CHECK constraint (alembic 005). # ImagePathType 이 ORM bind 시 자동 상대화 하지만, raw SQL 우회 시 절대가 # 들어갈 수 있어 DB 차원 마지막 안전망. duplicate_object 면 skip (idempotent). # 사전 backfill 은 alembic 005 가 처리; startup 은 constraint 보강만. # 운영 절차: production 배포 시 `alembic upgrade head` 가 startup 보다 먼저 # 실행되어야 한다 (절대 row 가 있는 환경에서는 005 backfill → CHECK 추가 # 까지 완결). startup 만 실행되면 check_violation NOTICE 만 띄우고 skip. # 표현식은 alembic 005 CHECK_EXPR + project.py ImageAsset.__table_args__ 와 동일. """DO $$ BEGIN ALTER TABLE image_asset ADD CONSTRAINT ck_image_asset_file_path_relative CHECK (file_path = '' OR file_path NOT LIKE '/%'); EXCEPTION WHEN duplicate_object THEN NULL; WHEN check_violation THEN RAISE NOTICE 'ck_image_asset_file_path_relative skipped: existing rows violate. Run alembic 005 to backfill.'; END $$""", # ── 여러 편 처리 (alembic 014 와 **같은 모양**) ────────────────── # ★alembic 쪽만 고치면 개발·시험 DB 가 갈린다. 두 곳을 같이 본다. # 전부 additive — 기존 행을 안 지우고 안 덮는다. """CREATE TABLE IF NOT EXISTS project_short_id_counter ( project_id TEXT NOT NULL REFERENCES project_registry(id), owner TEXT NOT NULL, high_water INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (project_id, owner) )""", "ALTER TABLE entity_episode_link ADD COLUMN IF NOT EXISTS presence_status TEXT NOT NULL DEFAULT 'active'", "ALTER TABLE entity_episode_link ADD COLUMN IF NOT EXISTS episode_notes_json TEXT", "CREATE INDEX IF NOT EXISTS ix_entity_episode_link_presence ON entity_episode_link(project_id, episode_id, presence_status)", # ★기존 유일성 (character_id, outlook_id) 는 **안 지운다** — legacy 행의 # 중복 방지가 사라진다. episode_id 는 nullable 로 더하기만 한다. "ALTER TABLE character_outlook ADD COLUMN IF NOT EXISTS episode_id TEXT", "CREATE INDEX IF NOT EXISTS ix_character_outlook_episode ON character_outlook(project_id, episode_id)", ] # problems.md #2: 이전 패턴은 except Exception → rollback 후 다음 SQL 로 # silent 진행 → schema drift 가 부팅에서 은폐. 새 패턴은 fail-fast — 첫 실패 # 시 logger.error + RuntimeError 로 부팅 abort. 운영자가 alembic upgrade # 또는 수동 정리 후 재시도. # # 모든 _migrations SQL 은 idempotent 검증 됨: # - CREATE TABLE / INDEX / COLUMN: IF NOT EXISTS 명시 # - ALTER TABLE ALTER COLUMN TYPE: PG 가 동일 type 시 no-op # - UPDATE backfill (image_generated): self-healing — image_asset 매칭이 # 변경되면 매번 재반영, 동일 결과 유지 (idempotent in outcome) # - UPDATE backfill (short_id): WHERE NULL 조건 1회차 후 0 row → no-op # - DO block (file_path CHECK constraint): WHEN duplicate_object skip # 따라서 정상 배포 시퀀스 (alembic upgrade head → startup) 에서 fail 0. # production deploy 시 alembic upgrade 가 startup 보다 먼저 실행되어야 함. # lock_timeout + statement_timeout (Codex iter root cause fix v2): # runtime 또는 동시 startup 시 ALTER TABLE 이 다른 transaction 의 lock 을 # 무한 대기하는 deadlock-like hang 방지. # # iter v1 (SET 단발 후 conn.commit()) 는 production 에서 동작 안 함이 검증됨 # (10분 hang). 가설: SQLAlchemy connection pool 의 connection 재할당 / pool_ # pre_ping reset path 에서 session-level SET 이 풀림. # # iter v2: 매 _migrations SQL 마다 동일 transaction 안에서 SET LOCAL 적용 # (transaction-level — pool / pre_ping reset 무관). lock 30s + statement # 120s — ALTER TABLE 자체 hang 도 statement_timeout 으로 차단. with engine.connect() as conn: for idx, sql in enumerate(_migrations): try: # SET LOCAL — autobegin 으로 시작된 transaction 안에서만 유효. # 다음 conn.commit() 시 transaction 닫히면 자동 reset. # SET LOCAL 자체가 lock 시도 X → SET 으로 hang 위험 0. # non-PG backend (sqlite) 에서는 SET LOCAL 미지원 → 첫 실행 시 # exception. try/except 로 graceful degrade. try: conn.execute(text("SET LOCAL lock_timeout = '30s'")) conn.execute(text("SET LOCAL statement_timeout = '120s'")) except Exception as set_exc: if idx == 0: _logger.warning( "SET LOCAL timeout failed (non-PG backend?): %r — " "migration 진행 (timeout guard 0 — startup hang 위험 잔존)", set_exc, ) conn.execute(text(sql)) conn.commit() except Exception as exc: conn.rollback() # Codex review: 200 자 truncation 으로 short_id CTE / DO block 잘림. # idx + 500 자 확장으로 운영자 진단 가능성 확보. _logger.error( "Startup migration #%d failed (fail-fast). SQL preview: %.500s ... Cause: %r", idx, sql, exc, ) raise RuntimeError( f"DB schema migration #{idx} failed at startup (fail-fast). " "Run `alembic upgrade head` manually before retry. " f"Cause: {exc!r}" ) from exc