hyungi_document_server/app/services/study/publish_enqueue.py

"""발행 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 models.study_topic import StudyTopic
from services.study.publish_projection import (
    KIND_EXPLANATION,
    KIND_QUESTION,
    KIND_TOPIC,
    SCHEMA_VERSION,
    payload_hash,
    project_explanation,
    project_question,
    project_topic,
)


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)


async def enqueue_topic_publish(session: AsyncSession, topic: Any) -> None:
    """주제 메타를 outbox 적재(S-1). caller commit. 저작 create/update 결선 + 백필 공용."""
    await enqueue_publish(session, kind=KIND_TOPIC, source_id=topic.id, payload=project_topic(topic))


async def backfill_publish_topics(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> int:
    """active(미삭제) 주제를 id>after_id 부터 bounded 로 outbox 적재(S-1 초기 백필).

    반환 = enqueue 한 주제 수(0 이면 끝). 큰 셋은 마지막 id 로 페이지 반복. caller commit.
    멱등 = 발행 워커의 (payload_hash, deleted) 디둡이 no-op 재투영 흡수(중복 enqueue 무해).
    """
    rows = (
        await session.execute(
            select(StudyTopic)
            .where(StudyTopic.deleted_at.is_(None), StudyTopic.id > after_id)
            .order_by(StudyTopic.id.asc())
            .limit(limit)
        )
    ).scalars().all()
    for t in rows:
        await enqueue_topic_publish(session, t)
    return len(rows)