From 7d64cd5d0b9cb88c0f364f84bd34c0bc52148ce9 Mon Sep 17 00:00:00 2001 From: KIMB0B Date: Tue, 8 Sep 2026 12:27:32 +0900 Subject: [PATCH 1/9] =?UTF-8?q?feat(notification):=20=EC=95=8C=EB=A6=BC=20?= =?UTF-8?q?=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC=EC=9D=B8=20=EB=8D=B0?= =?UTF-8?q?=EC=9D=B4=ED=84=B0=20=EB=AA=A8=EB=8D=B8=20=EB=B0=8F=20=EB=A7=88?= =?UTF-8?q?=EC=9D=B4=EA=B7=B8=EB=A0=88=EC=9D=B4=EC=85=98=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Notification에 event_type 추가 - NotificationRecipient(수신자별 읽음 상태), OutboxEvent(발행 대기함) 모델 추가 - NotificationEventType enum 추가 --- alembic/env.py | 2 + ...34c117_add_notification_pipeline_tables.py | 75 +++++++++++++++++++ app/main.py | 2 + app/modules/notification/models.py | 75 ++++++++++++++++++- app/shared/enums.py | 16 +++- 5 files changed, 168 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py diff --git a/alembic/env.py b/alembic/env.py index 2f64154..a05f4c8 100644 --- a/alembic/env.py +++ b/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 diff --git a/alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py b/alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py new file mode 100644 index 0000000..58ef4a1 --- /dev/null +++ b/alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py @@ -0,0 +1,75 @@ +"""add notification pipeline tables + +Revision ID: c8fc1434c117 +Revises: d8b2114a1c79 +Create Date: 2026-09-08 12:11:23.997757 + +""" +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 = 'c8fc1434c117' +down_revision: Union[str, Sequence[str], None] = 'd8b2114a1c79' +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', sa.Boolean(), nullable=False), + 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)) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + 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 ### diff --git a/app/main.py b/app/main.py index 6f26471..bd34dcc 100644 --- a/app/main.py +++ b/app/main.py @@ -21,6 +21,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.resource.models import Resource from app.modules.work.models import Work diff --git a/app/modules/notification/models.py b/app/modules/notification/models.py index 978ba89..a8f81c4 100644 --- a/app/modules/notification/models.py +++ b/app/modules/notification/models.py @@ -1,8 +1,12 @@ from uuid import UUID, uuid4 -from sqlalchemy import SmallInteger +from pydantic import AwareDatetime +from sqlalchemy import SmallInteger, Column, Enum +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy_utc import UtcDateTime from sqlmodel import Field +from app.shared.enums import NotificationEventType from app.shared.models.base import BaseModel class Notification(BaseModel, table=True): @@ -25,6 +29,18 @@ class Notification(BaseModel, table=True): description="알림 내용" ) + event_type: NotificationEventType = Field( + sa_column=Column( + Enum( + NotificationEventType, + name="notificationeventtype", + values_callable=lambda x: [e.value for e in x], + ), + nullable=False, + ), + description="알림 유형" + ) + target_type: int = Field( sa_type=SmallInteger, nullable=False, @@ -36,4 +52,61 @@ class Notification(BaseModel, table=True): default=None, nullable=True, description="알림 대상 고유키" + ) + +class NotificationRecipient(BaseModel, table=True): + __tablename__ = "notification_recipients" + + notification_id: UUID = Field( + foreign_key="notifications.id", + description="알림 고유키", + primary_key=True, + ) + + member_id: UUID = Field( + foreign_key="members.id", + description="회원 고유키", + primary_key=True, + ) + + read_at: AwareDatetime | None = Field( + default=None, + sa_type=UtcDateTime, + ) + +class OutboxEvent(BaseModel, table=True): + __tablename__ = "outbox_events" + + id: UUID = Field( + default_factory=uuid4, + primary_key=True, + nullable=False, + description="알림 발행 대기함 고유키" + ) + + event_type: NotificationEventType = Field( + sa_column=Column( + Enum( + NotificationEventType, + name="notificationeventtype", + values_callable=lambda x: [e.value for e in x], + ), + nullable=False, + ), + description="알림 유형" + ) + + payload: dict = Field( + sa_column=Column(JSONB, nullable=False), + description="발행할 내용 (받는 대상 / 제목 / 본문 등)" + ) + + dispatched: bool = Field( + default=False, + description="발행 완료 여부" + ) + + attempts: int = Field( + default=0, + description="실패 재시도 횟수 카운트" ) \ No newline at end of file diff --git a/app/shared/enums.py b/app/shared/enums.py index ae03134..45410fe 100644 --- a/app/shared/enums.py +++ b/app/shared/enums.py @@ -25,4 +25,18 @@ class InvitationStatus(str, Enum): class ChatRoomType(str, Enum): GROUP = "group" - DIRECT = "direct" \ No newline at end of file + DIRECT = "direct" + +class NotificationEventType(str, Enum): + # 공지 + NOTICE_CREATED = "notice_created" + NOTICE_UPDATED = "notice_updated" + # 작업 + WORK_ASSIGNED = "work_assigned" + WORK_STATUS_CHANGED = "work_status_changed" + # 프로젝트 + PROJECT_INVITED = "project_invited" + PROJECT_MEMBER_JOINED = "project_member_joined" + PROJECT_INVITATION_DECLINED = "project_invitation_declined" + PROJECT_MEMBER_REMOVED = "project_member_removed" + PROJECT_LEADERSHIP_TRANSFERRED = "project_leadership_transferred" From a32eb1d12cb2d3a7e7a1067e5d75abf4dcfecaac Mon Sep 17 00:00:00 2001 From: KIMB0B Date: Tue, 8 Sep 2026 23:46:57 +0900 Subject: [PATCH 2/9] =?UTF-8?q?feat(notification):=20=EA=B3=B5=EC=A7=80=20?= =?UTF-8?q?=EC=83=9D=EC=84=B1=20=EC=8B=9C=20outbox=20=EC=9D=B4=EB=B2=A4?= =?UTF-8?q?=ED=8A=B8=20=EB=B0=9C=ED=96=89=20=EC=97=B0=EA=B2=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/modules/notice/dependencies.py | 4 ++- app/modules/notice/service.py | 13 +++++++++- app/modules/notification/events.py | 30 ++++++++++++++++++++++ app/modules/notification/models.py | 2 +- app/modules/notification/repository.py | 35 ++++++++++++++++++++++++++ app/modules/project/repository.py | 15 +++++++++++ app/shared/enums.py | 7 ++++++ 7 files changed, 103 insertions(+), 3 deletions(-) create mode 100644 app/modules/notification/events.py create mode 100644 app/modules/notification/repository.py diff --git a/app/modules/notice/dependencies.py b/app/modules/notice/dependencies.py index 3d12f01..494165a 100644 --- a/app/modules/notice/dependencies.py +++ b/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)] diff --git a/app/modules/notice/service.py b/app/modules/notice/service.py index cf88394..f349ac6 100644 --- a/app/modules/notice/service.py +++ b/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,14 @@ 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, + title=saved.title, + detail=saved.detail, + recipient_ids=recipient_ids, + ) await self.session.commit() await self.session.refresh(saved) return saved diff --git a/app/modules/notification/events.py b/app/modules/notification/events.py new file mode 100644 index 0000000..a760bf2 --- /dev/null +++ b/app/modules/notification/events.py @@ -0,0 +1,30 @@ +from uuid import UUID + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.modules.notification.repository import OutboxEventRepository +from app.shared.enums import NotificationEventType, NotificationTargetType + + +class NotificationEvents: + + @staticmethod + def notice_created( + session: AsyncSession, + *, + notice_id: UUID, + title: str, + detail: str | None, + recipient_ids: list[UUID], + ) -> None: + if not recipient_ids: + return + + OutboxEventRepository(session).enqueue( + event_type=NotificationEventType.NOTICE_CREATED, + title=title, + detail=detail, + target_type=NotificationTargetType.NOTICE, + target_id=notice_id, + recipient_ids=recipient_ids, + ) \ No newline at end of file diff --git a/app/modules/notification/models.py b/app/modules/notification/models.py index a8f81c4..e6fdbd5 100644 --- a/app/modules/notification/models.py +++ b/app/modules/notification/models.py @@ -45,7 +45,7 @@ class Notification(BaseModel, table=True): sa_type=SmallInteger, nullable=False, default=0, - description="알림 대상 유형(0: 프로젝트, 1: 작업, 2: 기타)" + description="알림 대상 유형(0: 프로젝트, 1: 작업, 2: 공지, 3: 초대, 4: 기타)" ) target_id: UUID | None = Field( diff --git a/app/modules/notification/repository.py b/app/modules/notification/repository.py new file mode 100644 index 0000000..1dd9049 --- /dev/null +++ b/app/modules/notification/repository.py @@ -0,0 +1,35 @@ +from uuid import UUID + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.modules.notification.models import OutboxEvent +from app.shared.enums import NotificationEventType + + +class OutboxEventRepository: + def __init__(self, session: AsyncSession): + self.session = session + + def enqueue( + self, + *, + event_type: NotificationEventType, + title: str, + detail: str | None, + target_type: int, + target_id: UUID | None, + recipient_ids: list[UUID], + ) -> OutboxEvent: + event = OutboxEvent( + event_type=event_type, + payload={ + "event_type": event_type.value, + "title": title, + "detail": detail, + "target_type": int(target_type), + "target_id": str(target_id) if target_id else None, + "recipient_ids": [str(member_id) for member_id in dict.fromkeys(recipient_ids)], + }, + ) + self.session.add(event) + return event \ No newline at end of file diff --git a/app/modules/project/repository.py b/app/modules/project/repository.py index dd24abb..047d636 100644 --- a/app/modules/project/repository.py +++ b/app/modules/project/repository.py @@ -108,6 +108,21 @@ async def get_members_with_info(self, project_id: UUID) -> "list[tuple[Member, d result = await self.session.execute(stmt) return result.all() + async def get_member_ids(self, project_id: UUID, include_leader: bool = False) -> "list[UUID]": + """ + 프로젝트 멤버의 ID 목록을 조회합니다. + """ + project = await self.get_by_id(project_id) + if not project: + return [] + + stmt = select(ProjectMember.member_id).where(ProjectMember.project_id == project_id) + result = await self.session.execute(stmt) + member_ids = [row[0] for row in result.fetchall()] + if include_leader: + member_ids.append(project.leader_id) + return member_ids + async def delete_member(self, project_id: UUID, member_id: UUID) -> None: """ 프로젝트에서 멤버를 제거합니다. diff --git a/app/shared/enums.py b/app/shared/enums.py index 45410fe..e558fe7 100644 --- a/app/shared/enums.py +++ b/app/shared/enums.py @@ -40,3 +40,10 @@ class NotificationEventType(str, Enum): PROJECT_INVITATION_DECLINED = "project_invitation_declined" PROJECT_MEMBER_REMOVED = "project_member_removed" PROJECT_LEADERSHIP_TRANSFERRED = "project_leadership_transferred" + +class NotificationTargetType(int, Enum): + PROJECT = 0 + WORK = 1 + NOTICE = 2 + INVITATION = 3 + OTHER = 4 From 2de008306771f8e36faddc292c7eae750ae67e5b Mon Sep 17 00:00:00 2001 From: KIMB0B Date: Wed, 9 Sep 2026 02:59:07 +0900 Subject: [PATCH 3/9] =?UTF-8?q?feat(notification):=20outbox=20relay?= =?UTF-8?q?=EB=A1=9C=20=EC=9D=B4=EB=B2=A4=ED=8A=B8=EB=A5=BC=20redis=20stre?= =?UTF-8?q?am=EC=97=90=20=EB=B0=9C=ED=96=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...e7cb5_add_notification_pipeline_tables.py} | 10 ++--- app/main.py | 15 ++++++- app/modules/notification/models.py | 6 +-- app/modules/notification/outbox_relay.py | 41 +++++++++++++++++++ app/modules/notification/repository.py | 22 +++++++++- 5 files changed, 84 insertions(+), 10 deletions(-) rename alembic/versions/{c8fc1434c117_add_notification_pipeline_tables.py => 175711fe7cb5_add_notification_pipeline_tables.py} (94%) create mode 100644 app/modules/notification/outbox_relay.py diff --git a/alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py b/alembic/versions/175711fe7cb5_add_notification_pipeline_tables.py similarity index 94% rename from alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py rename to alembic/versions/175711fe7cb5_add_notification_pipeline_tables.py index 58ef4a1..1ac7177 100644 --- a/alembic/versions/c8fc1434c117_add_notification_pipeline_tables.py +++ b/alembic/versions/175711fe7cb5_add_notification_pipeline_tables.py @@ -1,8 +1,8 @@ """add notification pipeline tables -Revision ID: c8fc1434c117 +Revision ID: 175711fe7cb5 Revises: d8b2114a1c79 -Create Date: 2026-09-08 12:11:23.997757 +Create Date: 2026-09-09 02:23:01.891101 """ from typing import Sequence, Union @@ -13,11 +13,12 @@ from sqlalchemy.dialects import postgresql # revision identifiers, used by Alembic. -revision: str = 'c8fc1434c117' +revision: str = '175711fe7cb5' down_revision: Union[str, Sequence[str], None] = 'd8b2114a1c79' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None + notificationeventtype_enum = postgresql.ENUM( 'notice_created', 'notice_updated', @@ -32,7 +33,6 @@ create_type=False, ) - def upgrade() -> None: """Upgrade schema.""" # ### commands auto generated by Alembic - please adjust! ### @@ -45,7 +45,7 @@ def upgrade() -> None: 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', sa.Boolean(), nullable=False), + sa.Column('dispatched_at', sqlalchemy_utc.sqltypes.UtcDateTime(timezone=True), nullable=True), sa.Column('attempts', sa.Integer(), nullable=False), sa.PrimaryKeyConstraint('id') ) diff --git a/app/main.py b/app/main.py index bd34dcc..490faab 100644 --- a/app/main.py +++ b/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,6 +12,7 @@ from app.core.exceptions import AppError from app.core.logger import setup_logging from app.core.middleware import RequestIdMiddleware +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 @@ -28,6 +32,15 @@ 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()) + + yield + + outbox_relay_task.cancel() + class HealthOut(BaseModel): status: str = Field(example="ok") app: str = Field(example="teampling-api") @@ -41,7 +54,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) diff --git a/app/modules/notification/models.py b/app/modules/notification/models.py index e6fdbd5..c1b0368 100644 --- a/app/modules/notification/models.py +++ b/app/modules/notification/models.py @@ -101,9 +101,9 @@ class OutboxEvent(BaseModel, table=True): description="발행할 내용 (받는 대상 / 제목 / 본문 등)" ) - dispatched: bool = Field( - default=False, - description="발행 완료 여부" + dispatched_at: AwareDatetime | None = Field( + default=None, + sa_type=UtcDateTime, ) attempts: int = Field( diff --git a/app/modules/notification/outbox_relay.py b/app/modules/notification/outbox_relay.py new file mode 100644 index 0000000..75c06b8 --- /dev/null +++ b/app/modules/notification/outbox_relay.py @@ -0,0 +1,41 @@ +import asyncio +import json +import logging + +from app.core.database import AsyncSessionDocker +from app.core.redis import redis_client +from app.modules.notification.repository import OutboxEventRepository + + +logger = logging.getLogger(__name__) + +NOTIFICATION_STREAM_NAME = "notifications:stream" +OUTBOX_POLL_INTERVAL_SECONDS = 1.0 +OUTBOX_BATCH_SIZE = 100 + +async def run_outbox_relay(): + logger.info("Outbox relay 시작됨") + while True: + try: + async with AsyncSessionDocker() as session: + repository = OutboxEventRepository(session) + pending_events = await repository.get_undispatched(limit=OUTBOX_BATCH_SIZE) + + for event in pending_events: + try: + await redis_client.xadd(NOTIFICATION_STREAM_NAME, {"data": json.dumps(event.payload)}) + await repository.mark_dispatched(event) + except Exception as e: + logger.error(f"outbox event {event.id} 발행 실패, 다음 폴링에서 재시도: {e}") + await repository.mark_failed(event) + + await session.commit() + processed_count = len(pending_events) + except asyncio.CancelledError: + logger.info("Outbox relay 정지됨") + raise + except Exception as e: + logger.error(f"Outbox relay 배치 처리 중 오류: {e}") + processed_count = 0 + + await asyncio.sleep(0 if processed_count >= OUTBOX_BATCH_SIZE else OUTBOX_POLL_INTERVAL_SECONDS) \ No newline at end of file diff --git a/app/modules/notification/repository.py b/app/modules/notification/repository.py index 1dd9049..05899d0 100644 --- a/app/modules/notification/repository.py +++ b/app/modules/notification/repository.py @@ -1,6 +1,8 @@ +from datetime import datetime, timezone from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession +from sqlmodel import select from app.modules.notification.models import OutboxEvent from app.shared.enums import NotificationEventType @@ -32,4 +34,22 @@ def enqueue( }, ) self.session.add(event) - return event \ No newline at end of file + return event + + async def get_undispatched(self, limit: int = 100) -> list[OutboxEvent]: + stmt = ( + select(OutboxEvent) + .where(OutboxEvent.dispatched_at.is_(None)) + .limit(limit) + .with_for_update(skip_locked=True) + ) + result = await self.session.execute(stmt) + return list(result.scalars().all()) + + async def mark_dispatched(self, event: OutboxEvent) -> None: + event.dispatched_at = datetime.now(timezone.utc) + self.session.add(event) + + async def mark_failed(self, event: OutboxEvent) -> None: + event.attempts += 1 + self.session.add(event) \ No newline at end of file From 41ae00e193f00df7c364a358ef10c357d6b1b84f Mon Sep 17 00:00:00 2001 From: KIMB0B Date: Wed, 9 Sep 2026 12:24:06 +0900 Subject: [PATCH 4/9] =?UTF-8?q?feat(notification):=20consumer=EB=A1=9C=20s?= =?UTF-8?q?tream=EC=9D=84=20=EB=B9=84=ED=95=B4=20=EC=95=8C=EB=A6=BC=20?= =?UTF-8?q?=EC=83=9D=EC=84=B1=ED=95=98=EB=8F=84=EB=A1=9D=20=EA=B5=AC?= =?UTF-8?q?=ED=98=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/main.py | 4 ++ app/modules/notification/constants.py | 10 +++ app/modules/notification/consumer.py | 89 ++++++++++++++++++++++++ app/modules/notification/manager.py | 13 ++++ app/modules/notification/outbox_relay.py | 5 +- app/modules/notification/repository.py | 35 +++++++++- 6 files changed, 150 insertions(+), 6 deletions(-) create mode 100644 app/modules/notification/constants.py create mode 100644 app/modules/notification/consumer.py create mode 100644 app/modules/notification/manager.py diff --git a/app/main.py b/app/main.py index 490faab..37cf8a4 100644 --- a/app/main.py +++ b/app/main.py @@ -12,6 +12,7 @@ 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 @@ -36,10 +37,13 @@ @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") diff --git a/app/modules/notification/constants.py b/app/modules/notification/constants.py new file mode 100644 index 0000000..b3fd854 --- /dev/null +++ b/app/modules/notification/constants.py @@ -0,0 +1,10 @@ +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) \ No newline at end of file diff --git a/app/modules/notification/consumer.py b/app/modules/notification/consumer.py new file mode 100644 index 0000000..a3948c7 --- /dev/null +++ b/app/modules/notification/consumer.py @@ -0,0 +1,89 @@ +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.manager 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, + recipient_ids=recipient_ids, + ) + + await redis_client.xack(NOTIFICATION_STREAM_NAME, NOTIFICATION_GROUP_NAME, message_id) + + 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, + "created_at": notification.created_at.isoformat(), + } + try: + for member_id in recipient_ids: + await notification_manager.publish(member_id, realtime_payload) + except Exception as e: + logger.warning(f"알림 실시간 발행 실패 후 넘어감(Notification ID: {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) \ No newline at end of file diff --git a/app/modules/notification/manager.py b/app/modules/notification/manager.py new file mode 100644 index 0000000..74c4943 --- /dev/null +++ b/app/modules/notification/manager.py @@ -0,0 +1,13 @@ +import json +from uuid import UUID + +from app.core.redis import redis_client + + +class NotificationConnectionManager: + + @staticmethod + async def publish(member_id: UUID, message: dict): + await redis_client.publish(f"notify:{member_id}", json.dumps(message)) + +notification_manager = NotificationConnectionManager() \ No newline at end of file diff --git a/app/modules/notification/outbox_relay.py b/app/modules/notification/outbox_relay.py index 75c06b8..3bfcdd4 100644 --- a/app/modules/notification/outbox_relay.py +++ b/app/modules/notification/outbox_relay.py @@ -4,15 +4,12 @@ from app.core.database import AsyncSessionDocker from app.core.redis import redis_client +from app.modules.notification.constants import OUTBOX_BATCH_SIZE, NOTIFICATION_STREAM_NAME, OUTBOX_POLL_INTERVAL_SECONDS from app.modules.notification.repository import OutboxEventRepository logger = logging.getLogger(__name__) -NOTIFICATION_STREAM_NAME = "notifications:stream" -OUTBOX_POLL_INTERVAL_SECONDS = 1.0 -OUTBOX_BATCH_SIZE = 100 - async def run_outbox_relay(): logger.info("Outbox relay 시작됨") while True: diff --git a/app/modules/notification/repository.py b/app/modules/notification/repository.py index 05899d0..4d6ef29 100644 --- a/app/modules/notification/repository.py +++ b/app/modules/notification/repository.py @@ -4,10 +4,41 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import select -from app.modules.notification.models import OutboxEvent -from app.shared.enums import NotificationEventType +from app.modules.notification.models import OutboxEvent, Notification, NotificationRecipient +from app.shared.enums import NotificationEventType, NotificationTargetType +class NotificationRepository: + def __init__(self, session: AsyncSession): + self.session = session + + async def create( + self, + *, + event_type: NotificationEventType, + title: str, + detail: str | None, + target_type: NotificationTargetType, + target_id: UUID | None, + recipient_ids: list[UUID], + ) -> Notification: + notification = Notification( + event_type=event_type, + title=title, + detail=detail, + target_type=int(target_type.value), + target_id = target_id + ) + self.session.add(notification) + await self.session.flush() + + for member_id in dict.fromkeys(recipient_ids): + self.session.add(NotificationRecipient(notification_id=notification.id, member_id=member_id)) + + await self.session.commit() + await self.session.refresh(notification) + return notification + class OutboxEventRepository: def __init__(self, session: AsyncSession): self.session = session From c100ca21f0cddfb81dbeb8f43d9a860863513c72 Mon Sep 17 00:00:00 2001 From: KIMB0B Date: Thu, 10 Sep 2026 02:56:00 +0900 Subject: [PATCH 5/9] =?UTF-8?q?feat(notification):=20=EC=95=8C=EB=A6=BC=20?= =?UTF-8?q?=EC=A1=B0=ED=9A=8C/=EC=9D=BD=EC=9D=8C=20=EC=B2=98=EB=A6=AC=20RE?= =?UTF-8?q?ST=20API=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/main.py | 2 + app/modules/notification/consumer.py | 23 +++--- app/modules/notification/dependencies.py | 15 ++++ app/modules/notification/models.py | 4 +- app/modules/notification/repository.py | 78 ++++++++++++++++++- app/modules/notification/router.py | 97 ++++++++++++++++++++++++ app/modules/notification/schemas.py | 62 +++++++++++++++ app/modules/notification/service.py | 66 ++++++++++++++++ 8 files changed, 334 insertions(+), 13 deletions(-) create mode 100644 app/modules/notification/dependencies.py create mode 100644 app/modules/notification/router.py create mode 100644 app/modules/notification/schemas.py create mode 100644 app/modules/notification/service.py diff --git a/app/main.py b/app/main.py index 37cf8a4..3099f17 100644 --- a/app/main.py +++ b/app/main.py @@ -21,6 +21,7 @@ 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 @@ -71,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") diff --git a/app/modules/notification/consumer.py b/app/modules/notification/consumer.py index a3948c7..da01176 100644 --- a/app/modules/notification/consumer.py +++ b/app/modules/notification/consumer.py @@ -38,22 +38,25 @@ async def _handle_message(message_id: str, fields: dict): 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, + "created_at": notification.created_at.isoformat(), + } + + await session.commit() + await redis_client.xack(NOTIFICATION_STREAM_NAME, NOTIFICATION_GROUP_NAME, message_id) - 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, - "created_at": notification.created_at.isoformat(), - } try: for member_id in recipient_ids: await notification_manager.publish(member_id, realtime_payload) except Exception as e: - logger.warning(f"알림 실시간 발행 실패 후 넘어감(Notification ID: {notification.id}): {e}") + logger.warning(f"알림 실시간 발행 실패 후 넘어감(Notification ID: {realtime_payload['notification_id']}): {e}") async def run_notification_consumer(): # XGROUP 생성 시도, 이미 존재하면 무시 diff --git a/app/modules/notification/dependencies.py b/app/modules/notification/dependencies.py new file mode 100644 index 0000000..02f0e6f --- /dev/null +++ b/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)] diff --git a/app/modules/notification/models.py b/app/modules/notification/models.py index c1b0368..423d662 100644 --- a/app/modules/notification/models.py +++ b/app/modules/notification/models.py @@ -4,7 +4,7 @@ from sqlalchemy import SmallInteger, Column, Enum from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy_utc import UtcDateTime -from sqlmodel import Field +from sqlmodel import Field, Relationship from app.shared.enums import NotificationEventType from app.shared.models.base import BaseModel @@ -57,6 +57,8 @@ class Notification(BaseModel, table=True): class NotificationRecipient(BaseModel, table=True): __tablename__ = "notification_recipients" + notification: "Notification" = Relationship() + notification_id: UUID = Field( foreign_key="notifications.id", description="알림 고유키", diff --git a/app/modules/notification/repository.py b/app/modules/notification/repository.py index 4d6ef29..830bb4a 100644 --- a/app/modules/notification/repository.py +++ b/app/modules/notification/repository.py @@ -1,7 +1,9 @@ from datetime import datetime, timezone from uuid import UUID +from sqlalchemy import func, update from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload from sqlmodel import select from app.modules.notification.models import OutboxEvent, Notification, NotificationRecipient @@ -31,14 +33,86 @@ async def create( ) self.session.add(notification) await self.session.flush() + await self.session.refresh(notification) for member_id in dict.fromkeys(recipient_ids): self.session.add(NotificationRecipient(notification_id=notification.id, member_id=member_id)) - await self.session.commit() - await self.session.refresh(notification) return notification + async def list_by_member( + self, + member_id: UUID, + *, + offset: int = 0, + limit: int = 100, + unread_only: bool = False, + ) -> list[NotificationRecipient]: + stmt = ( + select(NotificationRecipient) + .where( + NotificationRecipient.member_id == member_id, + NotificationRecipient.is_deleted == False, + ) + .options(selectinload(NotificationRecipient.notification)) + .order_by(NotificationRecipient.created_at.desc()) + .offset(offset) + .limit(limit) + ) + + if unread_only: + stmt = stmt.where(NotificationRecipient.read_at.is_(None)) + + result = await self.session.execute(stmt) + return list(result.scalars().all()) + + async def count_by_member( + self, + member_id: UUID, + *, + unread_only: bool = False, + ) -> int: + stmt = ( + select(func.count()) + .select_from(NotificationRecipient) + .where( + NotificationRecipient.member_id == member_id, + NotificationRecipient.is_deleted == False, + ) + ) + + if unread_only: + stmt = stmt.where(NotificationRecipient.read_at.is_(None)) + + result = await self.session.execute(stmt) + return result.scalar_one() + + async def get_recipient( + self, + member_id: UUID, + notification_id: UUID, + ) -> NotificationRecipient | None: + stmt = select(NotificationRecipient).where( + NotificationRecipient.member_id == member_id, + NotificationRecipient.notification_id == notification_id, + NotificationRecipient.is_deleted == False, + ) + result = await self.session.execute(stmt) + return result.scalar_one_or_none() + + async def mark_all_read(self, member_id: UUID) -> int: + stmt = ( + update(NotificationRecipient) + .where( + NotificationRecipient.member_id == member_id, + NotificationRecipient.read_at.is_(None), + NotificationRecipient.is_deleted == False, + ) + .values(read_at=datetime.now(timezone.utc)) + ) + result = await self.session.execute(stmt) + return result.rowcount + class OutboxEventRepository: def __init__(self, session: AsyncSession): self.session = session diff --git a/app/modules/notification/router.py b/app/modules/notification/router.py new file mode 100644 index 0000000..611b295 --- /dev/null +++ b/app/modules/notification/router.py @@ -0,0 +1,97 @@ +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Path, Query + +from app.modules.member.dependencies import CurrentMemberDep +from app.modules.notification.dependencies import NotificationServiceDep +from app.modules.notification.schemas import NotificationOut, UnreadCountOut +from app.shared.schemas import ApiResponse, PageOut + + +router = APIRouter(prefix="/notifications", tags=["Notification"]) + +@router.get( + path="", + response_model=ApiResponse[PageOut[NotificationOut]], + summary="내 알림 목록 조회", + description="현재 로그인한 회원에게 온 알림 목록을 최신순으로 조회합니다.", +) +async def list_my_notifications( + current_member: CurrentMemberDep, + service: NotificationServiceDep, + page: Annotated[int, Query(ge=1, description="페이지 번호")] = 1, + size: Annotated[int, Query(ge=1, le=100, description="페이지 크기")] = 50, + unread_only: Annotated[bool, Query(description="안 읽은 알림만 조회")] = False, +): + result = await service.list_by_member( + current_member.id, + page=page, + size=size, + unread_only=unread_only, + ) + return ApiResponse.success( + code="NOTIFICATION_LIST_FETCHED", + message="알림 목록 조회 성공", + data=PageOut[NotificationOut]( + items=[NotificationOut.from_recipient(r) for r in result["items"]], + page=result["page"], + size=result["size"], + total=result["total"], + ), + ) + + +@router.get( + path="/count/unread", + response_model=ApiResponse[UnreadCountOut], + summary="안 읽은 알림 개수 조회", + description="현재 로그인한 회원의 안 읽은 알림 개수를 조회합니다.", +) +async def get_unread_count( + current_member: CurrentMemberDep, + service: NotificationServiceDep, +): + count = await service.count_unread(current_member.id) + return ApiResponse.success( + code="NOTIFICATION_UNREAD_COUNT_FETCHED", + message="안 읽은 알림 개수 조회 성공", + data=UnreadCountOut(count=count), + ) + + +@router.patch( + path="/read/all", + response_model=ApiResponse[None], + summary="모든 알림 읽음 처리", + description="현재 로그인한 회원의 안 읽은 알림을 모두 읽음 처리합니다.", +) +async def mark_all_notifications_read( + current_member: CurrentMemberDep, + service: NotificationServiceDep, +): + count = await service.mark_all_read(current_member.id) + return ApiResponse.success( + code="NOTIFICATION_ALL_READ", + message=f"알림 {count}건 읽음 처리 성공", + data=None, + ) + + +@router.patch( + path="/read/one/{notification_id}", + response_model=ApiResponse[None], + summary="알림 읽음 처리", + description="현재 로그인한 회원의 특정 알림을 읽음 처리합니다.", +) +async def mark_notification_read( + current_member: CurrentMemberDep, + service: NotificationServiceDep, + notification_id: Annotated[UUID, Path(description="읽음 처리할 알림 ID")], +): + await service.mark_read(current_member.id, notification_id) + return ApiResponse.success( + code="NOTIFICATION_READ", + message="알림 읽음 처리 성공", + data=None, + ) diff --git a/app/modules/notification/schemas.py b/app/modules/notification/schemas.py new file mode 100644 index 0000000..7f70458 --- /dev/null +++ b/app/modules/notification/schemas.py @@ -0,0 +1,62 @@ +from datetime import datetime +from uuid import UUID + +from sqlmodel import SQLModel, Field + +from app.modules.notification.models import NotificationRecipient +from app.shared.enums import NotificationEventType + + +class UnreadCountOut(SQLModel): + count: int = Field(description="안 읽은 알림 개수") + + model_config = { + "json_schema_extra": { + "example": { + "count": 3 + } + } + } + + +class NotificationOut(SQLModel): + notification_id: UUID = Field(description="알림 고유키") + event_type: NotificationEventType = Field(description="알림 유형") + title: str = Field(description="알림 제목") + detail: str | None = Field(default=None, description="알림 내용") + target_type: int = Field(description="알림 대상 유형(0: 프로젝트, 1: 작업, 2: 공지, 3: 초대, 4: 기타)") + target_id: UUID | None = Field(default=None, description="알림 대상 고유키") + is_read: bool = Field(description="읽음 여부") + read_at: datetime | None = Field(default=None, description="읽은 시각") + created_at: datetime = Field(description="생성 일시") + + model_config = { + "json_schema_extra": { + "example": { + "notification_id": "5f1672cf-8d99-4b1c-9b5e-9c3ece11b089", + "event_type": "notice_created", + "title": "새 공지가 등록되었습니다.", + "detail": "공지 상세 내용입니다.", + "target_type": 2, + "target_id": "3e1672cf-8d99-4b1c-9b5e-9c3ece11b089", + "is_read": False, + "read_at": None, + "created_at": "2026-09-09T10:00:00Z" + } + } + } + + @classmethod + def from_recipient(cls, recipient: NotificationRecipient) -> "NotificationOut": + notification = recipient.notification + return cls( + notification_id=notification.id, + event_type=notification.event_type, + title=notification.title, + detail=notification.detail, + target_type=notification.target_type, + target_id=notification.target_id, + is_read=recipient.read_at is not None, + read_at=recipient.read_at, + created_at=notification.created_at, + ) diff --git a/app/modules/notification/service.py b/app/modules/notification/service.py new file mode 100644 index 0000000..e7e4c99 --- /dev/null +++ b/app/modules/notification/service.py @@ -0,0 +1,66 @@ +from datetime import datetime, timezone +from typing import Any +from uuid import UUID + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.exceptions import AppError +from app.modules.notification.repository import NotificationRepository + + +class NotificationService: + def __init__(self, session: AsyncSession, repository: NotificationRepository): + self.session = session + self.repository = repository + + async def list_by_member( + self, + member_id: UUID, + *, + page: int = 1, + size: int = 50, + unread_only: bool = False, + ) -> dict[str, Any]: + offset = (page - 1) * size + items = await self.repository.list_by_member( + member_id, + offset=offset, + limit=size, + unread_only=unread_only, + ) + total = await self.repository.count_by_member( + member_id, + unread_only=unread_only, + ) + return { + "items": items, + "page": page, + "size": size, + "total": total, + } + + async def count_unread(self, member_id: UUID) -> int: + return await self.repository.count_by_member(member_id, unread_only=True) + + async def mark_read(self, member_id: UUID, notification_id: UUID) -> None: + recipient = await self.repository.get_recipient(member_id, notification_id) + if recipient is None: + raise AppError.not_found(f"[{notification_id}] 알림") + + if recipient.read_at is None: + recipient.read_at = datetime.now(timezone.utc) + + try: + await self.session.commit() + except Exception: + await self.session.rollback() + raise + + async def mark_all_read(self, member_id: UUID) -> int: + try: + count = await self.repository.mark_all_read(member_id) + await self.session.commit() + return count + except Exception: + await self.session.rollback() + raise From 53e68b4acf245fdfa5364341b8d1667e365de540 Mon Sep 17 00:00:00 2001 From: KIMB0B Date: Fri, 11 Sep 2026 02:22:31 +0900 Subject: [PATCH 6/9] =?UTF-8?q?feat(notification):=20=EC=8B=A4=EC=8B=9C?= =?UTF-8?q?=EA=B0=84=20=EC=95=8C=EB=A6=BC=20Push(WebSocket=20+=20Pub/Sub)?= =?UTF-8?q?=20=EA=B5=AC=ED=98=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/modules/notification/constants.py | 3 +- app/modules/notification/consumer.py | 2 +- app/modules/notification/manager.py | 13 --- app/modules/notification/realtime.py | 63 +++++++++++ app/modules/notification/router.py | 28 ++++- app/static/index.html | 146 ++++++++++++++++++++++++-- 6 files changed, 227 insertions(+), 28 deletions(-) delete mode 100644 app/modules/notification/manager.py create mode 100644 app/modules/notification/realtime.py diff --git a/app/modules/notification/constants.py b/app/modules/notification/constants.py index b3fd854..591c1cf 100644 --- a/app/modules/notification/constants.py +++ b/app/modules/notification/constants.py @@ -7,4 +7,5 @@ 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) \ No newline at end of file +CONSUMER_BLOCK_MS = 5000 # 새 메시지 없을 때 최대 대기 시간(ms) +NOTIFICATION_CHANNEL_PREFIX = "notify:" # 실시간 Pub/Sub 채널 접두사 (publish/subscribe 공유 계약, notify:{member_id}) \ No newline at end of file diff --git a/app/modules/notification/consumer.py b/app/modules/notification/consumer.py index da01176..2fbe351 100644 --- a/app/modules/notification/consumer.py +++ b/app/modules/notification/consumer.py @@ -14,7 +14,7 @@ CONSUMER_READ_COUNT, CONSUMER_BLOCK_MS, ) -from app.modules.notification.manager import notification_manager +from app.modules.notification.realtime import notification_manager from app.modules.notification.repository import NotificationRepository from app.shared.enums import NotificationTargetType, NotificationEventType diff --git a/app/modules/notification/manager.py b/app/modules/notification/manager.py deleted file mode 100644 index 74c4943..0000000 --- a/app/modules/notification/manager.py +++ /dev/null @@ -1,13 +0,0 @@ -import json -from uuid import UUID - -from app.core.redis import redis_client - - -class NotificationConnectionManager: - - @staticmethod - async def publish(member_id: UUID, message: dict): - await redis_client.publish(f"notify:{member_id}", json.dumps(message)) - -notification_manager = NotificationConnectionManager() \ No newline at end of file diff --git a/app/modules/notification/realtime.py b/app/modules/notification/realtime.py new file mode 100644 index 0000000..fb0acfb --- /dev/null +++ b/app/modules/notification/realtime.py @@ -0,0 +1,63 @@ +import asyncio +import json +import logging +from uuid import UUID +from fastapi import WebSocket + +from app.core.redis import redis_client +from app.modules.notification.constants import NOTIFICATION_CHANNEL_PREFIX + + +logger = logging.getLogger(__name__) + +class NotificationConnectionManager: + def __init__(self): + self.active_connections: dict[UUID, set[WebSocket]] = {} + self.sub_tasks: dict[UUID, asyncio.Task] = {} + + async def connect(self, member_id: UUID, websocket: WebSocket): + if member_id not in self.active_connections: + self.active_connections[member_id] = set() + self.sub_tasks[member_id] = asyncio.create_task(self._subscribe(member_id)) + + self.active_connections[member_id].add(websocket) + + def disconnect(self, member_id: UUID, websocket: WebSocket): + if member_id not in self.active_connections: + return + + self.active_connections[member_id].discard(websocket) + + if not self.active_connections[member_id]: + task = self.sub_tasks.pop(member_id, None) + if task: + task.cancel() + del self.active_connections[member_id] + + async def _subscribe(self, member_id: UUID): + pubsub = redis_client.pubsub() + channel = f"{NOTIFICATION_CHANNEL_PREFIX}{member_id}" + try: + await pubsub.subscribe(channel) + async for message in pubsub.listen(): + if message["type"] == "message": + data = json.loads(message["data"]) + await self._local_broadcast(member_id, data) + except asyncio.CancelledError: + await pubsub.unsubscribe(channel) + except Exception as e: + logger.error(f"알림 구독 오류 (member: {member_id}): {e}") + await pubsub.unsubscribe(channel) + + async def _local_broadcast(self, member_id: UUID, message: dict): + for connection in list(self.active_connections.get(member_id, set())): + try: + await connection.send_json(message) + except Exception: + self.active_connections[member_id].discard(connection) + + @staticmethod + async def publish(member_id: UUID, message: dict): + await redis_client.publish(f"{NOTIFICATION_CHANNEL_PREFIX}{member_id}", json.dumps(message)) + +notification_manager = NotificationConnectionManager() \ No newline at end of file diff --git a/app/modules/notification/router.py b/app/modules/notification/router.py index 611b295..0539bdc 100644 --- a/app/modules/notification/router.py +++ b/app/modules/notification/router.py @@ -2,9 +2,12 @@ from uuid import UUID from fastapi import APIRouter, Path, Query +from starlette.websockets import WebSocket, WebSocketDisconnect +from app.core.security import decode_token from app.modules.member.dependencies import CurrentMemberDep from app.modules.notification.dependencies import NotificationServiceDep +from app.modules.notification.realtime import notification_manager from app.modules.notification.schemas import NotificationOut, UnreadCountOut from app.shared.schemas import ApiResponse, PageOut @@ -41,7 +44,6 @@ async def list_my_notifications( ), ) - @router.get( path="/count/unread", response_model=ApiResponse[UnreadCountOut], @@ -59,7 +61,6 @@ async def get_unread_count( data=UnreadCountOut(count=count), ) - @router.patch( path="/read/all", response_model=ApiResponse[None], @@ -77,7 +78,6 @@ async def mark_all_notifications_read( data=None, ) - @router.patch( path="/read/one/{notification_id}", response_model=ApiResponse[None], @@ -95,3 +95,25 @@ async def mark_notification_read( message="알림 읽음 처리 성공", data=None, ) + +@router.websocket("/ws") +async def notification_websocket( + websocket: WebSocket, + token: Annotated[str, Query(description="WebSocket 인증 토큰")], +): + await websocket.accept() + + try: + payload = decode_token(token) + member_id = UUID(payload["sub"]) + except Exception: + await websocket.close(code=1008) # Policy Violation (인증 실패) + return + + await notification_manager.connect(member_id, websocket) + + try: + while True: + await websocket.receive_text() + except WebSocketDisconnect: + notification_manager.disconnect(member_id, websocket) \ No newline at end of file diff --git a/app/static/index.html b/app/static/index.html index 816f681..60d465a 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -5,6 +5,21 @@ 팀플링 - Teampling +
@@ -40,21 +55,13 @@

팀플링에 오신 것을 환영합니다!

localStorage.setItem('user_id', user.id); renderHeader(user.username, user.profile_url); - + initNotifications(); + const contentDiv = document.getElementById('content'); contentDiv.classList.remove('welcome-msg'); contentDiv.innerHTML = `
-
-

새 알림

-
-
-

새로운 알림 기능 준비 중입니다.

-
-
- -

내 프로젝트

@@ -126,6 +133,13 @@

완료

: 'https://cdn-icons-png.flaticon.com/512/149/149071.png'; navLinks.innerHTML = ` +
+ + +