FastAPI 모놀리식에서 Domain Event + Event Bus 적용기

app/shared/events.py

"""
Domain Event 기반 타입

이벤트는 발생한 도메인 사실만 담는 불변 값 객체다. payload에는 식별자와
도메인 사실만 담는다 (API DTO/ORM 인스턴스 금지, datetime은 naive-UTC,
컬렉션은 tuple — frozen 유지).
"""
from dataclasses import dataclass
from enum import Enum


class EventOrigin(str, Enum):
    """이벤트를 촉발한 source 도메인 (왕복 전파 차단용)"""
    TODO = "todo"
    SCHEDULE = "schedule"


@dataclass(frozen=True, kw_only=True)
class DomainEvent:
    """
    모든 도메인 이벤트의 베이스

    :param owner_id: 리소스 소유자 (OIDC sub). 핸들러는 변경 대상의
        owner_id가 이와 일치할 때만 수정한다.
    :param origin: 이벤트를 촉발한 source 도메인
    """
    owner_id: str
    origin: EventOrigin

    def event_key(self) -> tuple:
        """
        중복 dispatch 억제 키 — 타입 + 주요 식별자

        같은 drain 루프 안에서 동일 키의 이벤트는 한 번만 처리된다.
        서브클래스는 주요 식별자를 포함하도록 재정의한다.
        """
        return (type(self).__qualname__, self.owner_id)

app/shared/uow.py

"""
세션 스코프 이벤트 수집 helper

범용 이벤트 버스/outbox가 아니라 동기 트랜잭션 내 도메인 이벤트 투영 큐
도메인 모델이 SQLModel ORM 클래스의 alias라 엔티티에 `.events` 리스트를
직접 붙일 수 없어, 세션 스코프 수집(`session.info`)을 사용한다.
"""
from typing import TYPE_CHECKING, List

from sqlmodel import Session

if TYPE_CHECKING:
    from app.shared.events import DomainEvent

_EVENTS_KEY = "domain_events"


def collect_event(session: Session, event: "DomainEvent") -> None:
    """이벤트를 세션 큐에 적재 (dispatch는 flush_and_dispatch_events에서)"""
    session.info.setdefault(_EVENTS_KEY, []).append(event)


def drain_events(session: Session) -> List["DomainEvent"]:
    """수집된 이벤트를 전부 꺼내 반환 (큐는 비워짐)"""
    events = session.info.get(_EVENTS_KEY, [])
    session.info[_EVENTS_KEY] = []
    return events


def clear_events(session: Session) -> None:
    """이벤트 큐 비우기 — rollback 경계에서 필수 호출"""
    session.info[_EVENTS_KEY] = []


def flush_and_dispatch_events(session: Session) -> None:
    """
    flush → drain/handle 루프 → flush (commit 없음)

    쓰기 use-case가 응답 DTO를 조립하기 전에 호출해야 핸들러가 만든
    자동 생성 리소스가 응답에 즉시 보인다. 핸들러 예외는 전파된다 →
    호출자(트랜잭션 경계)에서 전체 rollback.
    """
    from app.shared import messagebus

    session.flush()
    events = drain_events(session)
    if events:
        messagebus.handle(events, session)
    session.flush()


def commit_with_events(session: Session) -> None:
    """미처리 이벤트 dispatch를 한 번 더 보장(안전망)한 뒤 commit"""
    flush_and_dispatch_events(session)
    session.commit()

app/shared/messagebus.py

"""
In-process Message Bus

큐 방식 drain 루프로 이벤트를 핸들러에 전달한다. 모든 핸들러는 같은
세션/트랜잭션에서 실행되며, 하나라도 실패하면 예외가 전파되어 전체
rollback된다. cascade 무한루프는 event_key 중복 억제와
MAX_DISPATCH_DEPTH 상한으로 차단한다.
"""
import logging
from typing import Callable, Dict, List, Sequence, Type

from sqlmodel import Session

from app.shared.events import DomainEvent
from app.shared.uow import drain_events

logger = logging.getLogger(__name__)

Handler = Callable[[DomainEvent, Session], None]

#: 한 drain 루프에서 처리하는 **총 이벤트 수** 상한 (폭주 cascade 차단).
#: 재귀 깊이가 아니다 — 배치성 대량 발행이 필요해지면 깊이/총량 분리 필요.
MAX_DISPATCH_DEPTH = 10

#: 중앙 핸들러 registry — register()로만 등록
HANDLERS: Dict[Type[DomainEvent], List[Handler]] = {}

#: @register가 기록하는 등록 명세. 모듈 import는 한 번만 데코레이터를
#: 실행하므로, reset_handlers() 이후 register_handlers()가 이 명세를
#: replay해 registry를 복구한다.
_KNOWN_REGISTRATIONS: List[tuple] = []


class MaxDispatchDepthExceededError(RuntimeError):
    """drain 루프가 MAX_DISPATCH_DEPTH를 초과 — 폭주 cascade 의심"""

    def __init__(self, event: DomainEvent):
        super().__init__(
            f"Max event dispatch depth ({MAX_DISPATCH_DEPTH}) exceeded "
            f"while handling {type(event).__qualname__}"
        )
        self.event = event


def register(event_type: Type[DomainEvent]):
    """
    핸들러 등록 데코레이터 — @register(TodoDeadlineSet)

    같은 (event_type, handler) 쌍은 한 번만 등록된다(idempotent) —
    모듈 재-import에 안전.
    """

    def decorator(fn: Handler) -> Handler:
        if (event_type, fn) not in _KNOWN_REGISTRATIONS:
            _KNOWN_REGISTRATIONS.append((event_type, fn))
        _add_handler(event_type, fn)
        return fn

    return decorator


def _add_handler(event_type: Type[DomainEvent], fn: Handler) -> None:
    handlers = HANDLERS.setdefault(event_type, [])
    if fn not in handlers:
        handlers.append(fn)


def handle(events: Sequence[DomainEvent], session: Session) -> None:
    """
    큐 방식 drain 루프

    - event_key 중복은 skip (왕복 전파 종료)
    - 핸들러 실행 후 flush → 후속 핸들러가 앞선 변경을 조회 가능
    - 핸들러가 collect한 후속 이벤트를 재수집해 같은 루프에서 처리
    - 예외는 삼키지 않고 전파 → 호출자에서 전체 rollback
    """
    queue = list(events)
    seen: set = set()
    depth = 0

    while queue:
        event = queue.pop(0)

        key = event.event_key()
        if key in seen:
            logger.debug("Duplicate event suppressed in dispatch chain: %r", event)
            continue
        seen.add(key)

        if depth >= MAX_DISPATCH_DEPTH:
            raise MaxDispatchDepthExceededError(event)
        depth += 1

        for handler in HANDLERS.get(type(event), []):
            handler(event, session)

        session.flush()
        queue.extend(drain_events(session))


def register_handlers() -> None:
    """
    도메인 핸들러 일괄 등록 (idempotent, reset 후 복구 가능)

    앱 startup(lifespan)과 테스트 fixture 양쪽에서 호출되므로 재호출에
    안전해야 한다. import는 첫 호출에서만 데코레이터를 실행하므로,
    기록된 등록 명세를 replay해 reset_handlers() 이후에도 복구한다.
    순환 import 방지를 위해 lazy import.
    """
    import app.domain.schedule.event_handlers  # noqa: F401
    import app.domain.todo.event_handlers  # noqa: F401

    for event_type, fn in _KNOWN_REGISTRATIONS:
        _add_handler(event_type, fn)


def reset_handlers() -> None:
    """registry 초기화 — 테스트 전용. register_handlers()로 복구할 수 있다."""
    HANDLERS.clear()

app/domain/todo/enums.py

"""
Todo Enums
"""
from enum import Enum


class TodoStatus(str, Enum):
    """Todo 상태"""
    UNSCHEDULED = "UNSCHEDULED"
    SCHEDULED = "SCHEDULED"
    DONE = "DONE"
    CANCELLED = "CANCELLED"


class LinkedSchedulePolicy(str, Enum):
    """마감 취소·Todo 삭제 시 연결 Schedule의 수명주기 처리 정책. 서버 기본값은 없다."""
    DELETE = "delete"  # 투영 삭제 + visibility cleanup
    UNLINK = "unlink"  # 연결만 해제 — 일반 일정으로 생존 (데이터 보존)

app/domain/schedule/enums.py (발췌)

class SourceTodoPolicy(str, Enum):
    """
    투영 일정이 마감 표시이기를 그만둘 때(삭제·반복 전환) source Todo 처리 정책.
    서버 기본값은 없다.
    """
    CLEAR_DEADLINE = "clear_deadline"  # Todo의 deadline도 제거
    KEEP_DEADLINE = "keep_deadline"    # deadline 유지 (연동 플래그만 해제)

app/domain/todo/events.py

"""
Todo Domain Events

payload는 식별자만 담고, 수신 핸들러가 같은 트랜잭션에서 현재 상태를
조회해 투영한다. 예외적으로 dispatch 시점에 원본이 사라지는 사실만
스냅샷으로 운반한다.
"""
from dataclasses import dataclass
from typing import Tuple
from uuid import UUID

from app.domain.todo.enums import LinkedSchedulePolicy
from app.shared.events import DomainEvent


@dataclass(frozen=True, kw_only=True)
class TodoEvent(DomainEvent):
    """Todo 이벤트 공통 베이스 — event_key에 todo_id 포함"""
    todo_id: UUID

    def event_key(self) -> tuple:
        return (type(self).__qualname__, str(self.todo_id))


@dataclass(frozen=True, kw_only=True)
class TodoDeadlineSet(TodoEvent):
    """deadline이 새로 설정됨 (생성 시 포함)"""


@dataclass(frozen=True, kw_only=True)
class TodoDeadlineChanged(TodoEvent):
    """deadline 값이 변경됨"""


@dataclass(frozen=True, kw_only=True)
class TodoDeadlineRemoved(TodoEvent):
    """
    deadline이 제거됨

    :param policy: 클라이언트가 명시한 연결 투영 처리 정책
    """
    policy: LinkedSchedulePolicy


@dataclass(frozen=True, kw_only=True)
class TodoContentChanged(TodoEvent):
    """제목·설명이 변경됨"""


@dataclass(frozen=True, kw_only=True)
class TodoTagsChanged(TodoEvent):
    """태그가 변경됨"""


@dataclass(frozen=True, kw_only=True)
class TodoDeleted(TodoEvent):
    """
    Todo가 삭제됨

    :param linked_schedule_ids: 삭제 직전 스냅샷한 연결 Schedule ID.
        `Schedule.source_todo_id`가 `ondelete=SET NULL`이라 삭제 flush
        후에는 조회할 수 없어 이벤트가 사실로서 운반한다.
    :param policy: 클라이언트가 명시한 연결 투영 처리 정책
    """
    linked_schedule_ids: Tuple[UUID, ...]
    policy: LinkedSchedulePolicy

app/domain/schedule/events.py

"""
Schedule Domain Events

역전파(Schedule→Todo)는 `source_todo_id`가 있는 투영 일정에 한정된다.
"""
from dataclasses import dataclass
from typing import Optional
from uuid import UUID

from app.domain.schedule.enums import SourceTodoPolicy
from app.shared.events import DomainEvent


@dataclass(frozen=True, kw_only=True)
class ScheduleEvent(DomainEvent):
    """Schedule 이벤트 공통 베이스 — event_key에 schedule_id 포함"""
    schedule_id: UUID

    def event_key(self) -> tuple:
        return (type(self).__qualname__, str(self.schedule_id))


@dataclass(frozen=True, kw_only=True)
class ScheduleTimeChanged(ScheduleEvent):
    """일정 시간이 변경됨. origin이 TODO면 재전파하지 않는다."""


@dataclass(frozen=True, kw_only=True)
class ScheduleDeleted(ScheduleEvent):
    """
    일정이 삭제됨

    :param source_todo_id: 삭제 직전 스냅샷 (row가 사라져 사후 조회 불가)
    :param policy: source-linked 일정이면 클라이언트가 명시한 source Todo
        처리 정책. unlinked 일정 삭제면 None
    """
    source_todo_id: Optional[UUID]
    policy: Optional[SourceTodoPolicy]


@dataclass(frozen=True, kw_only=True)
class ScheduleUnlinkedFromTodo(ScheduleEvent):
    """
    일정이 마감 투영이기를 그만둠 (반복 전환으로 연결 해제)

    :param source_todo_id: 해제 직전 스냅샷
    :param policy: 클라이언트가 명시한 source Todo 처리 정책
    """
    source_todo_id: UUID
    policy: SourceTodoPolicy


@dataclass(frozen=True, kw_only=True)
class ScheduleCreatedWithTodoOption(ScheduleEvent):
    """
    create_todo_options와 함께 일정이 생성됨

    핸들러가 low-level로 Todo를 생성하므로 TodoDeadlineSet이 재발화하지
    않는다 (A→B→A 순환의 구조적 차단).

    :param tag_group_id: 생성될 Todo가 속할 TagGroup
    """
    tag_group_id: UUID

app/domain/schedule/event_handlers.py

"""
Todo 이벤트를 받아 투영 Schedule을 동기화하는 핸들러.

핸들러는 내부 projection이라 사용자 권한 검증 대신 owner 불변조건
(변경 대상 owner_id == event.owner_id)을 강제하고, CurrentUser 없이
low-level 연산만 사용한다. 자동 생성 리소스에 visibility는 복사하지 않는다.
"""
import logging
from datetime import timedelta
from uuid import UUID

from sqlmodel import Session

from app.crud import schedule as schedule_crud
from app.crud import tag as tag_crud
from app.crud import todo as todo_crud
from app.crud import visibility as visibility_crud
from app.domain.todo.enums import LinkedSchedulePolicy
from app.domain.todo.events import (
    TodoContentChanged,
    TodoDeadlineChanged,
    TodoDeadlineRemoved,
    TodoDeadlineSet,
    TodoDeleted,
    TodoTagsChanged,
)
from app.domain.visibility.enums import ResourceType
from app.models.schedule import Schedule
from app.models.todo import Todo
from app.shared.messagebus import register

logger = logging.getLogger(__name__)

#: 투영 생성 시 초기 길이. 이후 길이는 사용자가 조정할 수 있고
#: deadline 이동 시 보존된다.
INITIAL_PROJECTION_DURATION = timedelta(hours=1)


# ============================================================
# 내부 helper
# ============================================================

def _get_owned_todo(session: Session, event) -> Todo | None:
    """owner 불변조건을 통과한 Todo 반환 (없으면 None)"""
    todo = todo_crud.get_todo(session, event.todo_id, event.owner_id)
    if todo is None:
        logger.warning(
            "Todo %s not found for %s (owner=%s) — skipping projection",
            event.todo_id, type(event).__qualname__, event.owner_id,
        )
    return todo


def _sync_projection_tags(session: Session, schedule_id: UUID, todo: Todo) -> None:
    """투영 태그를 Todo 태그로 전체 교체. 소유자의 태그만 연결한다."""
    tag_crud.delete_all_schedule_tags(session, schedule_id)
    for tag in todo.tags:
        if tag.owner_id == todo.owner_id:
            tag_crud.add_schedule_tag(session, schedule_id, tag.id)


def _create_projection(session: Session, todo: Todo) -> Schedule:
    """투영 Schedule 생성 — 제목·설명·태그는 Todo에서, 길이는 초기값"""
    schedule = Schedule(
        owner_id=todo.owner_id,
        title=todo.title,
        description=todo.description,
        start_time=todo.deadline,
        end_time=todo.deadline + INITIAL_PROJECTION_DURATION,
        source_todo_id=todo.id,
        # state는 모델 기본값 PLANNED, visibility는 복사하지 않음 (기본 PRIVATE)
    )
    session.add(schedule)
    session.flush()

    _sync_projection_tags(session, schedule.id, todo)
    return schedule


def _move_projection(session: Session, schedule: Schedule, todo: Todo) -> None:
    """start만 이동하고 사용자가 조정한 길이는 보존한다."""
    duration = schedule.end_time - schedule.start_time
    schedule.start_time = todo.deadline
    schedule.end_time = todo.deadline + duration
    session.flush()


def _delete_schedule_with_cleanup(session: Session, schedule: Schedule) -> None:
    """Schedule 삭제 + 접근권한 설정 cleanup (orphan visibility 방지)"""
    visibility_crud.delete_visibility_by_resource(
        session, ResourceType.SCHEDULE, schedule.id
    )
    schedule_crud.delete_schedule(session, schedule)


def _unlink_schedule(session: Session, schedule: Schedule) -> None:
    """연결만 해제 — 일반 일정으로 생존 (메모·태그·공유 보존)"""
    schedule.source_todo_id = None
    session.flush()


# ============================================================
# 핸들러
# ============================================================

@register(TodoDeadlineSet)
def create_projection_schedule(event: TodoDeadlineSet, session: Session) -> None:
    """deadline 설정 → 투영 생성. 이미 있으면 이동 (단일 투영 불변조건)"""
    todo = _get_owned_todo(session, event)
    if todo is None or todo.deadline is None:
        return

    schedules = schedule_crud.get_schedules_by_source_todo_id(
        session, event.todo_id, event.owner_id
    )
    if schedules:
        _move_projection(session, schedules[0], todo)
    else:
        _create_projection(session, todo)


@register(TodoDeadlineChanged)
def move_projection_schedule(event: TodoDeadlineChanged, session: Session) -> None:
    """deadline 변경 → 투영 이동. 투영이 없으면 재생성 (조용한 no-op 금지)"""
    todo = _get_owned_todo(session, event)
    if todo is None or todo.deadline is None:
        return

    schedules = schedule_crud.get_schedules_by_source_todo_id(
        session, event.todo_id, event.owner_id
    )
    if schedules:
        _move_projection(session, schedules[0], todo)
    else:
        _create_projection(session, todo)


@register(TodoDeadlineRemoved)
def remove_projection_schedule(event: TodoDeadlineRemoved, session: Session) -> None:
    """deadline 제거 → 클라이언트 정책에 따라 삭제 또는 연결 해제"""
    schedules = schedule_crud.get_schedules_by_source_todo_id(
        session, event.todo_id, event.owner_id
    )
    for schedule in schedules:
        if event.policy == LinkedSchedulePolicy.UNLINK:
            _unlink_schedule(session, schedule)
        else:
            _delete_schedule_with_cleanup(session, schedule)


@register(TodoContentChanged)
def sync_projection_content(event: TodoContentChanged, session: Session) -> None:
    """제목·설명 변경 → 투영에 최신 내용 반영"""
    todo = _get_owned_todo(session, event)
    if todo is None:
        return

    schedules = schedule_crud.get_schedules_by_source_todo_id(
        session, event.todo_id, event.owner_id
    )
    for schedule in schedules:
        schedule.title = todo.title
        schedule.description = todo.description
    if schedules:
        session.flush()


@register(TodoTagsChanged)
def sync_projection_tags(event: TodoTagsChanged, session: Session) -> None:
    """태그 변경 → 투영 태그 전체 교체"""
    todo = _get_owned_todo(session, event)
    if todo is None:
        return

    schedules = schedule_crud.get_schedules_by_source_todo_id(
        session, event.todo_id, event.owner_id
    )
    for schedule in schedules:
        _sync_projection_tags(session, schedule.id, todo)


@register(TodoDeleted)
def remove_projections_of_deleted_todo(event: TodoDeleted, session: Session) -> None:
    """Todo 삭제 → 스냅샷된 연결 투영을 클라이언트 정책에 따라 처리"""
    schedules = schedule_crud.get_schedules_by_ids(
        session, list(event.linked_schedule_ids)
    )
    for schedule in schedules:
        if schedule.owner_id != event.owner_id:  # owner 불변조건
            logger.warning(
                "Schedule %s owner mismatch on TodoDeleted — skipping", schedule.id
            )
            continue
        if event.policy == LinkedSchedulePolicy.UNLINK:
            _unlink_schedule(session, schedule)
        else:
            _delete_schedule_with_cleanup(session, schedule)

app/domain/todo/event_handlers.py

"""
Schedule 이벤트를 받아 source Todo를 동기화하는 핸들러.

역투영은 source_todo_id가 있는 일정에 한정하며, origin이 TODO인 변경은
재전파하지 않는다. status는 직접 쓰기(이벤트 미발행)로 갱신해 drain
재수집을 막는다.
"""
import logging

from sqlmodel import Session

from app.crud import schedule as schedule_crud
from app.crud import tag as tag_crud
from app.crud import todo as todo_crud
from app.domain.schedule.enums import SourceTodoPolicy
from app.domain.schedule.events import (
    ScheduleCreatedWithTodoOption,
    ScheduleDeleted,
    ScheduleTimeChanged,
    ScheduleUnlinkedFromTodo,
)
from app.domain.todo.enums import TodoStatus
from app.models.todo import Todo
from app.shared.events import EventOrigin
from app.shared.messagebus import register

logger = logging.getLogger(__name__)


def _apply_detach_policy(session: Session, todo: Todo, policy: SourceTodoPolicy) -> None:
    """투영이 사라진 Todo에 클라이언트 정책 적용 + 연동 플래그 해제"""
    if policy == SourceTodoPolicy.CLEAR_DEADLINE:
        todo.deadline = None
    # 터미널 상태(DONE/CANCELLED)는 덮지 않음 — 플래그 해제는 SCHEDULED에만
    if todo.status == TodoStatus.SCHEDULED:
        todo.status = TodoStatus.UNSCHEDULED
    session.flush()


@register(ScheduleTimeChanged)
def backpropagate_deadline(event: ScheduleTimeChanged, session: Session) -> None:
    """투영 일정 시간 이동 → Todo.deadline 역투영"""
    if event.origin == EventOrigin.TODO:
        return  # Todo발 변경의 재전파 차단

    schedule = schedule_crud.get_schedule(session, event.schedule_id, event.owner_id)
    if schedule is None or schedule.source_todo_id is None:
        return
    if schedule.recurrence_rule:
        # 반복 일정이 된 투영은 시점 마감의 투영이 아님 — 역투영 제외
        logger.debug(
            "Skipping deadline back-propagation for recurring schedule %s",
            schedule.id,
        )
        return

    todo = todo_crud.get_todo(session, schedule.source_todo_id, event.owner_id)
    if todo is None:
        return

    todo.deadline = schedule.start_time
    session.flush()


@register(ScheduleDeleted)
def apply_policy_on_projection_deleted(event: ScheduleDeleted, session: Session) -> None:
    """투영 삭제 → 클라이언트 정책에 따라 deadline 처리 + 연동 플래그 해제"""
    if event.origin == EventOrigin.TODO:
        return  # Todo발 삭제/정리는 Todo 쪽 흐름이 이미 상태를 관리
    if event.source_todo_id is None:
        return

    todo = todo_crud.get_todo(session, event.source_todo_id, event.owner_id)
    if todo is None:
        return

    # 내부 자동 삭제 경로는 policy가 없을 수 있음 → 데이터 보존 쪽 선택
    _apply_detach_policy(session, todo, event.policy or SourceTodoPolicy.KEEP_DEADLINE)


@register(ScheduleUnlinkedFromTodo)
def apply_policy_on_projection_unlinked(
        event: ScheduleUnlinkedFromTodo, session: Session
) -> None:
    """반복 전환으로 연결 해제 → 클라이언트 정책에 따라 deadline 처리 + 플래그 해제"""
    todo = todo_crud.get_todo(session, event.source_todo_id, event.owner_id)
    if todo is None:
        return

    _apply_detach_policy(session, todo, event.policy)


@register(ScheduleCreatedWithTodoOption)
def create_todo_for_schedule(
        event: ScheduleCreatedWithTodoOption, session: Session
) -> None:
    """
    create_todo_options → 연결 Todo 생성 + 역연결

    low-level 생성이므로 TodoDeadlineSet이 재발화하지 않는다
    (A→B→A 순환의 구조적 차단). 태그는 Schedule에서 복사한다.
    """
    schedule = schedule_crud.get_schedule(session, event.schedule_id, event.owner_id)
    if schedule is None or schedule.source_todo_id is not None:
        return

    todo = Todo(
        owner_id=event.owner_id,
        title=schedule.title,
        description=schedule.description,
        deadline=schedule.start_time,
        tag_group_id=event.tag_group_id,
        parent_id=None,
        status=TodoStatus.SCHEDULED,  # 투영(이 Schedule)이 있으므로 SCHEDULED
    )
    session.add(todo)
    session.flush()

    # 태그 복사 (owner 불변조건: 이벤트 소유자의 태그만)
    for tag in schedule.tags:
        if tag.owner_id == event.owner_id:
            tag_crud.add_todo_tag(session, todo.id, tag.id)

    schedule.source_todo_id = todo.id
    session.flush()

app/db/session.py (발췌)

def get_db() -> Generator[Session, None, None]:
    """
    읽기 전용 세션 (commit 없음)
    - GET 요청 등 읽기 작업에 사용
    - FastAPI가 자동으로 세션 close 처리
    """
    from app.shared.uow import clear_events

    with _session_manager.get_session() as session:
        try:
            yield session
        except Exception:
            session.rollback()
            raise
        finally:
            if session.new or session.dirty or session.deleted:
                logger.warning(
                    "Read-only session rolled back with pending changes. "
                    "new=%d dirty=%d deleted=%d",
                    len(session.new),
                    len(session.dirty),
                    len(session.deleted),
                )
            session.rollback()
            # session.info는 트랜잭션 aware가 아님 → rollback 경계에서 clear
            clear_events(session)


def get_db_transactional() -> Generator[Session, None, None]:
    """
    트랜잭션 자동 관리 세션
    - POST/PUT/DELETE 등 쓰기 작업에 사용
    - 함수 종료 성공: 미처리 도메인 이벤트 dispatch(안전망) 후 commit 자동
    - 예외 발생: rollback 자동 + 이벤트 큐 clear
    """
    from app.shared.uow import clear_events, commit_with_events

    with _session_manager.get_session() as session:
        try:
            yield session
            commit_with_events(session)
        except Exception:
            session.rollback()
            # session.info는 트랜잭션 aware가 아님 → rollback 경계에서 clear
            clear_events(session)
            raise

배선 (발췌)

# app/main.py — lifespan startup
# 1-1. Domain Event 핸들러 등록 (idempotent)
from app.shared.messagebus import register_handlers
register_handlers()
logger.info("✅ Domain event handlers registered")

# tests/conftest.py
@pytest.fixture(autouse=True, scope="session")
def _register_domain_event_handlers():
    """
    도메인 이벤트 핸들러 일괄 등록

    서비스 단위 테스트는 app.main(lifespan)을 import하지 않으므로,
    여기서 등록하지 않으면 투영 동기화가 조용히 누락됩니다.
    register_handlers는 idempotent라 e2e(lifespan 경유)와 중복 호출에 안전합니다.
    """
    from app.shared.messagebus import register_handlers
    register_handlers()