Feat: 이벤트 기반 알림 파이프라인 도입

alembic/env.py

@@ -16,6 +16,8 @@
 from app.modules.favorite.models import Favorite
 from app.modules.notice.models import Notice
 from app.modules.notification.models import Notification
+from app.modules.notification.models import NotificationRecipient
+from app.modules.notification.models import OutboxEvent
 from app.modules.project.models import Project
 from app.modules.project.models import ProjectMember
 from app.modules.project.models import ProjectInvitation

alembic/versions/5489d898de06_add_work_assignees.py

@@ -0,0 +1,43 @@
+"""add work_assignees
+
+Revision ID: 5489d898de06
+Revises: e54d101c6638
+Create Date: 2026-09-11 03:19:04.509629
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy_utc
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '5489d898de06'
+down_revision: Union[str, Sequence[str], None] = 'e54d101c6638'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('work_assignees',
+    sa.Column('created_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('updated_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('is_deleted', sa.Boolean(), server_default='false', nullable=False),
+    sa.Column('deleted_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), nullable=True),
+    sa.Column('work_id', sa.Uuid(), nullable=False),
+    sa.Column('member_id', sa.Uuid(), nullable=False),
+    sa.ForeignKeyConstraint(['member_id'], ['members.id'], ),
+    sa.ForeignKeyConstraint(['work_id'], ['works.id'], ),
+    sa.PrimaryKeyConstraint('work_id', 'member_id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('work_assignees')
+    # ### end Alembic commands ###

alembic/versions/e54d101c6638_add_notification_pipeline_tables.py

@@ -0,0 +1,77 @@
+"""add notification pipeline tables
+
+Revision ID: e54d101c6638
+Revises: a930eb13180a
+Create Date: 2026-09-11 02:34:00.261266
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy_utc
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = 'e54d101c6638'
+down_revision: Union[str, Sequence[str], None] = 'a930eb13180a'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+notificationeventtype_enum = postgresql.ENUM(
+    'notice_created',
+    'notice_updated',
+    'work_assigned',
+    'work_status_changed',
+    'project_invited',
+    'project_member_joined',
+    'project_invitation_declined',
+    'project_member_removed',
+    'project_leadership_transferred',
+    name='notificationeventtype',
+    create_type=False,
+)
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    notificationeventtype_enum.create(op.get_bind(), checkfirst=True)
+    op.create_table('outbox_events',
+    sa.Column('created_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('updated_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('is_deleted', sa.Boolean(), server_default='false', nullable=False),
+    sa.Column('deleted_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), nullable=True),
+    sa.Column('id', sa.Uuid(), nullable=False),
+    sa.Column('event_type', notificationeventtype_enum, nullable=False),
+    sa.Column('payload', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('dispatched_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), nullable=True),
+    sa.Column('attempts', sa.Integer(), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('notification_recipients',
+    sa.Column('created_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('updated_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('is_deleted', sa.Boolean(), server_default='false', nullable=False),
+    sa.Column('deleted_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), nullable=True),
+    sa.Column('notification_id', sa.Uuid(), nullable=False),
+    sa.Column('member_id', sa.Uuid(), nullable=False),
+    sa.Column('read_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), nullable=True),
+    sa.ForeignKeyConstraint(['member_id'], ['members.id'], ),
+    sa.ForeignKeyConstraint(['notification_id'], ['notifications.id'], ),
+    sa.PrimaryKeyConstraint('notification_id', 'member_id')
+    )
+    op.add_column('notifications', sa.Column('event_type', notificationeventtype_enum, nullable=False))
+    op.add_column('notifications', sa.Column('project_id', sa.Uuid(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('notifications', 'project_id')
+    op.drop_column('notifications', 'event_type')
+    op.drop_table('notification_recipients')
+    op.drop_table('outbox_events')
+    notificationeventtype_enum.drop(op.get_bind(), checkfirst=True)
+    # ### end Alembic commands ###

app/main.py

@@ -1,3 +1,6 @@
+import asyncio
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, Field, ConfigDict
@@ -9,23 +12,40 @@
 from app.core.exceptions import AppError
 from app.core.logger import setup_logging
 from app.core.middleware import RequestIdMiddleware
+from app.modules.notification.consumer import run_notification_consumer
+from app.modules.notification.outbox_relay import run_outbox_relay
 from app.shared.schemas import ApiResponse
 from app.modules.skill.router import router as skill_router
 from app.modules.member.router import router as member_router
 from app.modules.project.router import router as project_router
 from app.modules.work.router import work_router, project_work_router
 from app.modules.notice.router import router as notice_router
 from app.modules.chat.router import router as chat_router
+from app.modules.notification.router import router as notification_router
 from app.modules.skill.models import Skill
 from app.modules.member.models import Member
 from app.modules.favorite.models import Favorite
 from app.modules.notice.models import Notice
 from app.modules.notification.models import Notification
+from app.modules.notification.models import NotificationRecipient
+from app.modules.notification.models import OutboxEvent
 from app.modules.project.models import Project
 from app.modules.resource.models import Resource
 from app.modules.work.models import Work
 from app.modules.chat.models import ChatRoom, ChatRoomMember, ChatMessage
 
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    outbox_relay_task = asyncio.create_task(run_outbox_relay())
+    consumer_task = asyncio.create_task(run_notification_consumer())
+
+    yield
+
+    outbox_relay_task.cancel()
+    consumer_task.cancel()
+    await asyncio.gather(outbox_relay_task, consumer_task, return_exceptions=True)
+
 class HealthOut(BaseModel):
     status: str = Field(example="ok")
     app: str = Field(example="teampling-api")
@@ -39,7 +59,7 @@ class HealthOut(BaseModel):
 
 def create_app() -> FastAPI:
     setup_logging()
-    app = FastAPI(title=settings.APP_NAME)
+    app = FastAPI(title=settings.APP_NAME, lifespan=lifespan)
 
     # Exception Handler 등록
     register_exception_handlers(app)
@@ -52,6 +72,7 @@ def create_app() -> FastAPI:
     app.include_router(project_work_router)
     app.include_router(notice_router)
     app.include_router(chat_router)
+    app.include_router(notification_router)
 
     # Static Files
     app.mount("/static", StaticFiles(directory="app/static"), name="static")

app/modules/notice/dependencies.py

@@ -5,11 +5,13 @@
 from app.core.database import DbSessionDep
 from app.modules.notice.repository import NoticeRepository
 from app.modules.notice.service import NoticeService
+from app.modules.project.repository import ProjectRepository
 
 
 def get_notice_service(session: DbSessionDep) -> NoticeService:
     repository = NoticeRepository(session)
-    return NoticeService(session, repository)
+    project_repository = ProjectRepository(session)
+    return NoticeService(session, repository, project_repository)
 
 
 NoticeServiceDep = Annotated[NoticeService, Depends(get_notice_service)]

app/modules/notice/service.py

@@ -6,12 +6,15 @@
 from app.modules.notice.models import Notice
 from app.modules.notice.repository import NoticeRepository
 from app.modules.notice.schemas import NoticeCreateIn, NoticeUpdateIn
+from app.modules.notification.events import NotificationEvents
+from app.modules.project.repository import ProjectRepository
 
 
 class NoticeService:
-    def __init__(self, session: AsyncSession, repository: NoticeRepository):
+    def __init__(self, session: AsyncSession, repository: NoticeRepository, project_repository: ProjectRepository):
         self.session = session
         self.repository = repository
+        self.project_repository = project_repository
 
     async def get(self, notice_id: UUID, *, include_deleted: bool = False) -> Notice:
         notice = await self.repository.get_by_id(notice_id, include_deleted=include_deleted)
@@ -57,6 +60,15 @@ async def create(self, data: NoticeCreateIn) -> Notice:
         )
         try:
             saved = await self.repository.save(notice)
+            recipient_ids = await self.project_repository.get_member_ids(saved.project_id, include_leader=True)
+            NotificationEvents.notice_created(
+                self.session,
+                notice_id=saved.id,
+                project_id=saved.project_id,
+                title=saved.title,
+                detail=saved.detail,
+                recipient_ids=recipient_ids,
+            )
             await self.session.commit()
             await self.session.refresh(saved)
             return saved
@@ -74,6 +86,15 @@ async def update(self, target_notice_id: UUID, data: NoticeUpdateIn) -> Notice:
 
         try:
             updated = await self.repository.save(notice)
+            recipient_ids = await self.project_repository.get_member_ids(notice.project_id, include_leader=True)
+            NotificationEvents.notice_updated(
+                self.session,
+                notice_id=notice.id,
+                project_id=notice.project_id,
+                title=notice.title,
+                detail=notice.detail,
+                recipient_ids=recipient_ids,
+            )
             await self.session.commit()
             await self.session.refresh(updated)
             return updated

app/modules/notification/constants.py

@@ -0,0 +1,11 @@
+import socket
+import uuid
+
+NOTIFICATION_STREAM_NAME = "notifications:stream"   # 이벤트가 흐르는 Redis Stream 키 (relay·consumer 공유 계약)
+NOTIFICATION_GROUP_NAME = "notification-workers"    # Consumer Group 이름 (consumer 공유 계약)
+OUTBOX_POLL_INTERVAL_SECONDS = 1.0                  # relay가 outbox를 다시 폴링하기까지 대기(초)
+OUTBOX_BATCH_SIZE = 100                             # relay가 한 번에 처리할 미발행 outbox 행 수
+CONSUMER_NAME = f"{socket.gethostname()}-{uuid.uuid4().hex[:8]}"   # 그룹 내 컨슈머 식별자 (인스턴스마다 유일)
+CONSUMER_READ_COUNT = 10                            # 한 번 XREADGROUP에서 최대 몇 개 읽을지
+CONSUMER_BLOCK_MS = 5000                            # 새 메시지 없을 때 최대 대기 시간(ms)
+NOTIFICATION_CHANNEL_PREFIX = "notify:"             # 실시간 Pub/Sub 채널 접두사 (publish/subscribe 공유 계약, notify:{member_id})

app/modules/notification/consumer.py

@@ -0,0 +1,94 @@
+import asyncio
+import json
+import logging
+from uuid import UUID
+
+import redis
+
+from app.core.database import AsyncSessionDocker
+from app.core.redis import redis_client
+from app.modules.notification.constants import (
+    NOTIFICATION_STREAM_NAME,
+    NOTIFICATION_GROUP_NAME,
+    CONSUMER_NAME,
+    CONSUMER_READ_COUNT,
+    CONSUMER_BLOCK_MS,
+)
+from app.modules.notification.realtime import notification_manager
+from app.modules.notification.repository import NotificationRepository
+from app.shared.enums import NotificationTargetType, NotificationEventType
+
+logger = logging.getLogger(__name__)
+
+async def _handle_message(message_id: str, fields: dict):
+    payload = json.loads(fields["data"])
+    recipient_ids = [UUID(id) for id in payload.get("recipient_ids", [])]
+
+    if not recipient_ids:
+        await redis_client.xack(NOTIFICATION_STREAM_NAME, NOTIFICATION_GROUP_NAME, message_id)
+        return
+
+    async with AsyncSessionDocker() as session:
+        notification = await NotificationRepository(session).create(
+            event_type=NotificationEventType(payload["event_type"]),
+            title=payload["title"],
+            detail=payload["detail"],
+            target_type=NotificationTargetType(payload["target_type"]),
+            target_id=UUID(payload["target_id"]) if payload["target_id"] else None,
+            project_id=UUID(payload["project_id"]) if payload.get("project_id") else None,
+            recipient_ids=recipient_ids,
+        )
+
+        realtime_payload = {
+            "notification_id": str(notification.id),
+            "event_type": notification.event_type,
+            "title": notification.title,
+            "detail": notification.detail,
+            "target_type": notification.target_type,
+            "target_id": str(notification.target_id) if notification.target_id else None,
+            "project_id": str(notification.project_id) if notification.project_id else None,
+            "created_at": notification.created_at.isoformat(),
+        }
+
+        await session.commit()
+
+    await redis_client.xack(NOTIFICATION_STREAM_NAME, NOTIFICATION_GROUP_NAME, message_id)
+
+    try:
+        for member_id in recipient_ids:
+            await notification_manager.publish(member_id, realtime_payload)
+    except Exception as e:
+        logger.warning(f"알림 실시간 발행 실패 후 넘어감(Notification ID: {realtime_payload['notification_id']}): {e}")
+
+async def run_notification_consumer():
+    # XGROUP 생성 시도, 이미 존재하면 무시
+    try:
+        await redis_client.xgroup_create(NOTIFICATION_STREAM_NAME, NOTIFICATION_GROUP_NAME, id="$", mkstream=True)
+    except redis.exceptions.ResponseError as e:
+        if "BUSYGROUP" not in str(e):
+            raise
+    logger.info(f"Notification 컨슈머 '{CONSUMER_NAME}' 시작됨")
+
+    while True:
+        try:
+            response = await redis_client.xreadgroup(
+                NOTIFICATION_GROUP_NAME,
+                CONSUMER_NAME,
+                {NOTIFICATION_STREAM_NAME: ">"},
+                count=CONSUMER_READ_COUNT,
+                block=CONSUMER_BLOCK_MS,
+            )
+            if not response:
+                continue
+            for stream_name, messages in response:
+                for message_id, fields in messages:
+                    try:
+                        await _handle_message(message_id, fields)
+                    except Exception as e:
+                        logger.error(f"Notification 이벤트 {message_id} 처리 실패 (재시도 대상으로 남김): {e}")
+        except asyncio.CancelledError:
+            logger.info("Notification 컨슈머 종료됨")
+            raise
+        except Exception as e:
+            logger.error(f"Notification 컨슈머 루프 오류: {e}")
+            await asyncio.sleep(1)

app/modules/notification/dependencies.py

@@ -0,0 +1,15 @@
+from typing import Annotated
+
+from fastapi import Depends
+
+from app.core.database import DbSessionDep
+from app.modules.notification.repository import NotificationRepository
+from app.modules.notification.service import NotificationService
+
+
+def get_notification_service(session: DbSessionDep) -> NotificationService:
+    repository = NotificationRepository(session)
+    return NotificationService(session, repository)
+
+
+NotificationServiceDep = Annotated[NotificationService, Depends(get_notification_service)]