"""발행 outbox enqueue + 초기 백필 (docsrv-viewer-publish). enqueue_publish: 저작/4-A 트랜잭션이 같은 session(=같은 Postgres tx)에서 호출 → caller commit (P0-1 규율: 콘텐츠 변경과 outbox INSERT 원자성, dual-write 회피). payload/hash 스냅샷. enqueue_question_publish: 문항 + (ready면)해설을 함께 적재. 저작 쓰기/4-A 완료/백필 공용. backfill_publish_questions: 기존 active 문항을 bounded 로 1회 outbox 적재(초기 백필, P2-1 bounded page). 멱등 = 발행 워커의 (payload_hash, deleted) 디둡이 no-op 재투영 흡수(중복 enqueue 무해). ★주의: 저작 엔드포인트(study_questions create/update)·4-A 워커에서의 enqueue 결선은 P0-1b (기존 hot 파일 수정이라 별 increment). 본 모듈은 호출 라이브러리 + 수동/백필 진입점. """ from __future__ import annotations from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.published import PublishOutbox from models.study_question import StudyQuestion from services.study.publish_projection import ( KIND_EXPLANATION, KIND_QUESTION, SCHEMA_VERSION, payload_hash, project_explanation, project_question, ) async def enqueue_publish( session: AsyncSession, *, kind: str, source_id: int, payload: dict[str, Any] | None, deleted: bool = False, ) -> None: """outbox 1행 INSERT. caller 가 commit (저자 tx 동봉). deleted=True 면 tombstone(payload={}).""" body: dict[str, Any] = payload if payload is not None else {} session.add( PublishOutbox( kind=kind, source_id=source_id, payload=body, payload_hash=payload_hash(body), schema_version=SCHEMA_VERSION, deleted=deleted, ) ) async def enqueue_question_publish(session: AsyncSession, q: Any) -> None: """문항 + (ready면)해설을 outbox 적재. caller commit.""" await enqueue_publish(session, kind=KIND_QUESTION, source_id=q.id, payload=project_question(q)) expl = project_explanation(q) if expl is not None: await enqueue_publish(session, kind=KIND_EXPLANATION, source_id=q.id, payload=expl) async def backfill_publish_questions(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> int: """active(미삭제) 문항을 id>after_id 부터 bounded 로 outbox 적재. 반환 = enqueue 한 문항 수(0 이면 끝). 큰 셋은 마지막 id 로 페이지 반복. caller commit. """ rows = ( await session.execute( select(StudyQuestion) .where(StudyQuestion.deleted_at.is_(None), StudyQuestion.id > after_id) .order_by(StudyQuestion.id.asc()) .limit(limit) ) ).scalars().all() for q in rows: await enqueue_question_publish(session, q) return len(rows)