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/5489d898de06_add_work_assignees.py b/alembic/versions/5489d898de06_add_work_assignees.py
new file mode 100644
index 0000000..1201915
--- /dev/null
+++ b/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 ###
diff --git a/alembic/versions/e54d101c6638_add_notification_pipeline_tables.py b/alembic/versions/e54d101c6638_add_notification_pipeline_tables.py
new file mode 100644
index 0000000..2cbc81d
--- /dev/null
+++ b/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 ###
diff --git a/app/main.py b/app/main.py
index 6f26471..3099f17 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,8 @@
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
@@ -16,16 +21,31 @@
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")
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 87c6770..e0c2d6c 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,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
diff --git a/app/modules/notification/constants.py b/app/modules/notification/constants.py
new file mode 100644
index 0000000..591c1cf
--- /dev/null
+++ b/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})
\ 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..ffaac6f
--- /dev/null
+++ b/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)
\ No newline at end of file
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/events.py b/app/modules/notification/events.py
new file mode 100644
index 0000000..49bb4ee
--- /dev/null
+++ b/app/modules/notification/events.py
@@ -0,0 +1,116 @@
+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,
+ project_id: UUID,
+ title: str,
+ detail: str | None,
+ recipient_ids: list[UUID],
+ ) -> None:
+ NotificationEvents._notice_event(
+ session,
+ event_type=NotificationEventType.NOTICE_CREATED,
+ notice_id=notice_id, project_id=project_id,
+ title=title, detail=detail, recipient_ids=recipient_ids,
+ )
+
+ @staticmethod
+ def notice_updated(
+ session: AsyncSession,
+ *,
+ notice_id: UUID,
+ project_id: UUID,
+ title: str,
+ detail: str | None,
+ recipient_ids: list[UUID],
+ ) -> None:
+ NotificationEvents._notice_event(
+ session,
+ event_type=NotificationEventType.NOTICE_UPDATED,
+ notice_id=notice_id, project_id=project_id,
+ title=title, detail=detail, recipient_ids=recipient_ids,
+ )
+
+ @staticmethod
+ def _notice_event(session, *, event_type, notice_id, project_id, title, detail, recipient_ids):
+ if not recipient_ids:
+ return
+ OutboxEventRepository(session).enqueue(
+ event_type=event_type,
+ title=title,
+ detail=detail,
+ target_type=NotificationTargetType.NOTICE,
+ target_id=notice_id,
+ project_id=project_id,
+ recipient_ids=recipient_ids,
+ )
+
+ # ---------- 프로젝트 ----------
+ @staticmethod
+ def project_invited(session, *, project_id, project_name, recipient_ids):
+ NotificationEvents._project_event(session, event_type=NotificationEventType.PROJECT_INVITED, project_id=project_id, project_name=project_name, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def project_member_joined(session, *, project_id, project_name, recipient_ids):
+ NotificationEvents._project_event(session, event_type=NotificationEventType.PROJECT_MEMBER_JOINED, project_id=project_id, project_name=project_name, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def project_invitation_declined(session, *, project_id, project_name, recipient_ids):
+ NotificationEvents._project_event(session, event_type=NotificationEventType.PROJECT_INVITATION_DECLINED, project_id=project_id, project_name=project_name, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def project_member_removed(session, *, project_id, project_name, recipient_ids):
+ NotificationEvents._project_event(session, event_type=NotificationEventType.PROJECT_MEMBER_REMOVED, project_id=project_id, project_name=project_name, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def project_leadership_transferred(session, *, project_id, project_name, recipient_ids):
+ NotificationEvents._project_event(session, event_type=NotificationEventType.PROJECT_LEADERSHIP_TRANSFERRED, project_id=project_id, project_name=project_name, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def _project_event(session, *, event_type, project_id, project_name, recipient_ids):
+ if not recipient_ids:
+ return
+ OutboxEventRepository(session).enqueue(
+ event_type=event_type,
+ title=project_name,
+ detail=None,
+ target_type=NotificationTargetType.PROJECT,
+ target_id=project_id,
+ project_id=project_id,
+ recipient_ids=recipient_ids,
+ )
+
+ # ---------- 작업 ----------
+ @staticmethod
+ def work_assigned(session, *, work_id, project_id, title, recipient_ids):
+ NotificationEvents._work_event(session, event_type=NotificationEventType.WORK_ASSIGNED, work_id=work_id, project_id=project_id, title=title, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def work_status_changed(session, *, work_id, project_id, title, recipient_ids):
+ NotificationEvents._work_event(session, event_type=NotificationEventType.WORK_STATUS_CHANGED, work_id=work_id, project_id=project_id, title=title, recipient_ids=recipient_ids)
+
+ @staticmethod
+ def _work_event(session, *, event_type, work_id, project_id, title, recipient_ids):
+ if not recipient_ids:
+ return
+ OutboxEventRepository(session).enqueue(
+ event_type=event_type,
+ title=title,
+ detail=None,
+ target_type=NotificationTargetType.WORK,
+ target_id=work_id,
+ project_id=project_id,
+ recipient_ids=recipient_ids,
+ )
diff --git a/app/modules/notification/models.py b/app/modules/notification/models.py
index 978ba89..577a614 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 sqlmodel import Field
+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, Relationship
+from app.shared.enums import NotificationEventType
from app.shared.models.base import BaseModel
class Notification(BaseModel, table=True):
@@ -25,15 +29,92 @@ 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,
default=0,
- description="알림 대상 유형(0: 프로젝트, 1: 작업, 2: 기타)"
+ description="알림 대상 유형(0: 프로젝트, 1: 작업, 2: 공지, 3: 초대, 4: 기타)"
)
target_id: UUID | None = Field(
default=None,
nullable=True,
description="알림 대상 고유키"
+ )
+
+ project_id: UUID | None = Field(
+ default=None,
+ nullable=True,
+ description="알림이 속한 프로젝트 고유키"
+ )
+
+class NotificationRecipient(BaseModel, table=True):
+ __tablename__ = "notification_recipients"
+
+ notification: "Notification" = Relationship()
+
+ 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_at: AwareDatetime | None = Field(
+ default=None,
+ sa_type=UtcDateTime,
+ )
+
+ attempts: int = Field(
+ default=0,
+ description="실패 재시도 횟수 카운트"
)
\ No newline at end of file
diff --git a/app/modules/notification/outbox_relay.py b/app/modules/notification/outbox_relay.py
new file mode 100644
index 0000000..b204c5c
--- /dev/null
+++ b/app/modules/notification/outbox_relay.py
@@ -0,0 +1,42 @@
+import asyncio
+import json
+import logging
+
+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__)
+
+async def _dispatch_pending(session) -> int:
+ """미발행 outbox 이벤트를 Redis Stream으로 발행하고 처리 건수를 반환"""
+ 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()
+ return len(pending_events)
+
+async def run_outbox_relay():
+ logger.info("Outbox relay 시작됨")
+ while True:
+ try:
+ async with AsyncSessionDocker() as session:
+ processed_count = await _dispatch_pending(session)
+ 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/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/repository.py b/app/modules/notification/repository.py
new file mode 100644
index 0000000..991564f
--- /dev/null
+++ b/app/modules/notification/repository.py
@@ -0,0 +1,164 @@
+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
+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,
+ project_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,
+ project_id=project_id,
+ )
+ 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))
+
+ 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
+
+ def enqueue(
+ self,
+ *,
+ event_type: NotificationEventType,
+ title: str,
+ detail: str | None,
+ target_type: int,
+ target_id: UUID | None,
+ project_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,
+ "project_id": str(project_id) if project_id else None,
+ "recipient_ids": [str(member_id) for member_id in dict.fromkeys(recipient_ids)],
+ },
+ )
+ self.session.add(event)
+ 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
diff --git a/app/modules/notification/router.py b/app/modules/notification/router.py
new file mode 100644
index 0000000..0539bdc
--- /dev/null
+++ b/app/modules/notification/router.py
@@ -0,0 +1,119 @@
+from typing import Annotated
+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
+
+
+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,
+ )
+
+@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/modules/notification/schemas.py b/app/modules/notification/schemas.py
new file mode 100644
index 0000000..82d70e0
--- /dev/null
+++ b/app/modules/notification/schemas.py
@@ -0,0 +1,64 @@
+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="알림 대상 고유키")
+ project_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,
+ project_id=notification.project_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
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/modules/project/service.py b/app/modules/project/service.py
index 4c7acd3..f22b0ab 100644
--- a/app/modules/project/service.py
+++ b/app/modules/project/service.py
@@ -12,6 +12,7 @@
from app.modules.project.models import Project, ProjectInvitation, ProjectMember
from app.modules.project.repository import ProjectRepository
from app.modules.project.schemas import ProjectCreateIn, ProjectUpdateIn
+from app.modules.notification.events import NotificationEvents
from app.shared.enums import InvitationStatus
from app.shared.utils.email import send_email
@@ -219,6 +220,13 @@ async def invite_member(self, project_id: UUID, member_id: UUID) -> ProjectInvit
body = f"{invitee.username}님, {project.name} 프로젝트에 초대되었습니다.\n\n수락하시려면 아래 링크를 클릭하세요:\n{invite_url}"
await send_email(subject, invitee.email, body)
+
+ NotificationEvents.project_invited(
+ self.session,
+ project_id=project_id,
+ project_name=project.name,
+ recipient_ids=[member_id],
+ )
await self.session.commit()
await self.session.refresh(saved)
return saved
@@ -248,6 +256,14 @@ async def accept_invitation(self, token: str, current_member_id: UUID) -> Projec
self.session.add(project_member)
invitation.status = InvitationStatus.ACCEPTED
+
+ project = await self.get(invitation.project_id)
+ NotificationEvents.project_member_joined(
+ self.session,
+ project_id=invitation.project_id,
+ project_name=project.name,
+ recipient_ids=[project.leader_id],
+ )
await self.session.commit()
await self.session.refresh(invitation)
return invitation
@@ -264,6 +280,14 @@ async def decline_invitation(self, token: str, current_member_id: UUID) -> Proje
raise AppError.forbidden("본인에게 발송된 초대가 아닙니다.")
invitation.status = InvitationStatus.DECLINED
+
+ project = await self.get(invitation.project_id)
+ NotificationEvents.project_invitation_declined(
+ self.session,
+ project_id=invitation.project_id,
+ project_name=project.name,
+ recipient_ids=[project.leader_id],
+ )
await self.session.commit()
await self.session.refresh(invitation)
return invitation
@@ -277,6 +301,13 @@ async def remove_member(self, project_id: UUID, member_id: UUID) -> None:
raise AppError.bad_request("리더는 자신을 퇴출할 수 없습니다.")
await self.repository.delete_member(project_id, member_id)
+
+ NotificationEvents.project_member_removed(
+ self.session,
+ project_id=project_id,
+ project_name=project.name,
+ recipient_ids=[member_id],
+ )
await self.session.commit()
async def leave_project(self, project_id: UUID, member_id: UUID) -> None:
@@ -322,6 +353,12 @@ async def transfer_leader(
updated = await self.repository.save(project)
await self.repository.delete_member(project_id, new_leader_member_id)
self.session.add(ProjectMember(project_id=project_id, member_id=actor_member_id))
+ NotificationEvents.project_leadership_transferred(
+ self.session,
+ project_id=project_id,
+ project_name=project.name,
+ recipient_ids=[new_leader_member_id],
+ )
await self.session.commit()
await self.session.refresh(updated)
return updated
diff --git a/app/modules/work/models.py b/app/modules/work/models.py
index 67c39c0..0b104f5 100644
--- a/app/modules/work/models.py
+++ b/app/modules/work/models.py
@@ -13,11 +13,25 @@
from app.modules.project.models import Project
from app.modules.member.models import Member
+class WorkAssignee(BaseModel, table=True):
+ __tablename__ = "work_assignees"
+
+ work_id: UUID = Field(foreign_key="works.id", primary_key=True)
+ member_id: UUID = Field(foreign_key="members.id", primary_key=True)
+
class Work(BaseModel, table=True):
__tablename__ = "works"
project: "Project" = Relationship(back_populates="works")
author: "Member" = Relationship(back_populates="created_works")
+ assignees: list["Member"] = Relationship(
+ link_model=WorkAssignee,
+ sa_relationship_kwargs={"lazy": "selectin"},
+ )
+
+ @property
+ def assignee_ids(self) -> list[UUID]:
+ return [assignee.id for assignee in self.assignees]
id: UUID = Field(
default_factory=uuid4,
diff --git a/app/modules/work/repository.py b/app/modules/work/repository.py
index 335ff9c..9df8057 100644
--- a/app/modules/work/repository.py
+++ b/app/modules/work/repository.py
@@ -1,11 +1,11 @@
from datetime import datetime, timezone
from uuid import UUID
-from sqlalchemy import func, or_
+from sqlalchemy import func, or_, delete
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
-from app.modules.work.models import Work
+from app.modules.work.models import Work, WorkAssignee
from app.shared.enums import WorkState
@@ -13,6 +13,17 @@ class WorkRepository:
def __init__(self, session: AsyncSession):
self.session = session
+ async def set_assignees(self, work_id: UUID, member_ids: list[UUID]) -> None:
+ """작업 담당자를 전달된 목록으로 교체한다 (기존 전부 삭제 후 삽입)."""
+ await self.session.execute(delete(WorkAssignee).where(WorkAssignee.work_id == work_id))
+ for member_id in dict.fromkeys(member_ids):
+ self.session.add(WorkAssignee(work_id=work_id, member_id=member_id))
+
+ async def get_assignee_ids(self, work_id: UUID) -> list[UUID]:
+ stmt = select(WorkAssignee.member_id).where(WorkAssignee.work_id == work_id)
+ result = await self.session.execute(stmt)
+ return [row[0] for row in result.fetchall()]
+
async def get_by_id(self, work_id: UUID, *, include_deleted: bool = False) -> Work | None:
query = select(Work).where(Work.id == work_id)
if not include_deleted:
diff --git a/app/modules/work/schemas.py b/app/modules/work/schemas.py
index 087e9d1..4e1f205 100644
--- a/app/modules/work/schemas.py
+++ b/app/modules/work/schemas.py
@@ -13,6 +13,7 @@ class WorkCreateIn(SQLModel):
start_date: datetime = Field(description="작업 시작 일자")
end_date: datetime = Field(description="작업 종료 일자")
state: WorkState = Field(default=WorkState.PLANNED, description="작업 상태(planned, doing, done)")
+ assignee_ids: list[UUID] = Field(default_factory=list, description="담당자 회원 ID 목록")
@model_validator(mode="after")
def validate_dates(self) -> "WorkCreateIn":
@@ -39,6 +40,7 @@ class WorkUpdateIn(SQLModel):
start_date: datetime | None = Field(default=None, description="작업 시작 일자")
end_date: datetime | None = Field(default=None, description="작업 종료 일자")
state: WorkState | None = Field(default=None, description="작업 상태(planned, doing, done)")
+ assignee_ids: list[UUID] | None = Field(default=None, description="담당자 회원 ID 목록")
@model_validator(mode="after")
def validate_dates(self) -> "WorkUpdateIn":
@@ -66,6 +68,7 @@ class WorkOut(SQLModel):
start_date: datetime = Field(description="작업 시작 일자")
end_date: datetime = Field(description="작업 종료 일자")
state: WorkState = Field(description="작업 상태(planned, doing, done)")
+ assignee_ids: list[UUID] = Field(default_factory=list, description="담당자 회원 ID 목록")
created_at: datetime = Field(description="생성 일시")
updated_at: datetime | None = Field(default=None, description="수정 일시")
diff --git a/app/modules/work/service.py b/app/modules/work/service.py
index f52651c..c6fad5e 100644
--- a/app/modules/work/service.py
+++ b/app/modules/work/service.py
@@ -8,6 +8,7 @@
from app.modules.work.models import Work
from app.modules.work.repository import WorkRepository
from app.modules.work.schemas import WorkCreateIn, WorkUpdateIn
+from app.modules.notification.events import NotificationEvents
from app.shared.enums import WorkState
@@ -62,16 +63,25 @@ async def get_stats(self, project_id: UUID) -> dict[str, int]:
return await self.repository.get_stats_by_project(project_id)
async def create(self, actor_member_id: UUID, data: WorkCreateIn) -> Work:
- work = Work(
- **data.model_dump(),
- author_id=actor_member_id
- )
+ dump = data.model_dump()
+ assignee_ids = dump.pop("assignee_ids", [])
+ work = Work(**dump, author_id=actor_member_id)
try:
saved = await self.repository.save(work)
+ work_id = saved.id
+ project_id = saved.project_id
+ title = saved.title
+ await self.repository.set_assignees(work_id, assignee_ids)
+ NotificationEvents.work_assigned(
+ self.session,
+ work_id=work_id,
+ project_id=project_id,
+ title=title,
+ recipient_ids=assignee_ids,
+ )
await self.session.commit()
- await self.session.refresh(saved)
- return saved
+ return await self.get(work_id)
except IntegrityError as e:
await self.session.rollback()
@@ -83,18 +93,47 @@ async def update(self, target_work_id: UUID, actor_member_id: UUID, data: WorkUp
if actor_member_id != work.author_id:
raise AppError.forbidden("본인이 만든 작업만 수정할 수 있습니다.")
- patch = data.model_dump(
- exclude_unset=True,
- )
+ patch = data.model_dump(exclude_unset=True)
+ new_assignee_ids = patch.pop("assignee_ids", None)
+ state_provided = "state" in patch
+ prev_state = work.state
for k, v in patch.items():
setattr(work, k, v)
try:
updated = await self.repository.save(work)
+
+ # 담당자 교체 + 새로 추가된 담당자에게 work_assigned
+ if new_assignee_ids is not None:
+ prev_assignees = set(await self.repository.get_assignee_ids(target_work_id))
+ await self.repository.set_assignees(target_work_id, new_assignee_ids)
+ newly_added = [m for m in dict.fromkeys(new_assignee_ids) if m not in prev_assignees]
+ NotificationEvents.work_assigned(
+ self.session,
+ work_id=updated.id,
+ project_id=updated.project_id,
+ title=updated.title,
+ recipient_ids=newly_added,
+ )
+
+ # 상태 변경 시 담당자 + 작성자에게 work_status_changed
+ if state_provided and updated.state != prev_state:
+ if new_assignee_ids is not None:
+ assignee_ids = list(dict.fromkeys(new_assignee_ids))
+ else:
+ assignee_ids = await self.repository.get_assignee_ids(target_work_id)
+ recipients = list(dict.fromkeys([*assignee_ids, updated.author_id]))
+ NotificationEvents.work_status_changed(
+ self.session,
+ work_id=updated.id,
+ project_id=updated.project_id,
+ title=updated.title,
+ recipient_ids=recipients,
+ )
+
await self.session.commit()
- await self.session.refresh(updated)
- return updated
+ return await self.get(target_work_id)
except IntegrityError as e:
await self.session.rollback()
diff --git a/app/shared/enums.py b/app/shared/enums.py
index 16115d3..ce56a9e 100644
--- a/app/shared/enums.py
+++ b/app/shared/enums.py
@@ -27,7 +27,28 @@ class ChatRoomType(str, Enum):
GROUP = "group"
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"
+
+class NotificationTargetType(int, Enum):
+ PROJECT = 0
+ WORK = 1
+ NOTICE = 2
+ INVITATION = 3
+ OTHER = 4
+
class ChatEventType(str, Enum):
MESSAGE = "message"
TYPING = "typing"
- PRESENCE = "presence"
\ No newline at end of file
+ PRESENCE = "presence"
diff --git a/app/static/auth.js b/app/static/auth.js
new file mode 100644
index 0000000..ba6ed2e
--- /dev/null
+++ b/app/static/auth.js
@@ -0,0 +1,63 @@
+// 공용 인증 fetch 래퍼
+// access token 만료(401) 시, refresh token으로 자동 재발급 후 원 요청을 1회 재시도한다.
+// 각 페이지의 기존 fetch 호출을 바꿀 필요 없이, window.fetch를 감싸서 전역 적용한다.
+(function () {
+ const origFetch = window.fetch.bind(window);
+ let reissuePromise = null; // 동시 다발 401에서 재발급을 1번만 수행하기 위한 락
+
+ function urlOf(input) {
+ return typeof input === 'string' ? input : (input && input.url) || '';
+ }
+
+ async function reissueOnce() {
+ const refreshToken = localStorage.getItem('refresh_token');
+ if (!refreshToken) return false;
+ try {
+ const res = await origFetch('/members/reissue', {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ refresh_token: refreshToken }),
+ });
+ if (!res.ok) return false;
+ const tokens = await res.json(); // TokenOut (감싸지 않은 직접 객체)
+ if (!tokens.access_token) return false;
+ localStorage.setItem('access_token', tokens.access_token);
+ if (tokens.refresh_token) localStorage.setItem('refresh_token', tokens.refresh_token);
+ return true;
+ } catch (e) {
+ return false;
+ }
+ }
+
+ window.fetch = async function (input, init = {}) {
+ const res = await origFetch(input, init);
+ const url = urlOf(input);
+
+ // 401이 아니거나, 인증 엔드포인트 자체(재발급/로그인)면 그대로 반환
+ if (res.status !== 401 || url.includes('/members/reissue') || url.includes('/members/login')) {
+ return res;
+ }
+
+ // 401 → 재발급 (여러 요청이 동시에 401이면 재발급은 1번만)
+ if (!reissuePromise) {
+ reissuePromise = reissueOnce().finally(() => { reissuePromise = null; });
+ }
+ const ok = await reissuePromise;
+
+ if (!ok) {
+ // 재발급 실패(리프레시도 만료 등) → 로그아웃 후 로그인 페이지로
+ localStorage.clear();
+ if (!location.pathname.endsWith('/login.html')) {
+ location.href = '/static/login.html';
+ }
+ return res;
+ }
+
+ // 새 access token으로 Authorization 교체 후 원 요청 1회 재시도
+ const retryInit = Object.assign({}, init);
+ retryInit.headers = Object.assign({}, init.headers || {}, {
+ 'Authorization': `Bearer ${localStorage.getItem('access_token')}`,
+ });
+ return origFetch(input, retryInit);
+ };
+})();
diff --git a/app/static/index.html b/app/static/index.html
index 816f681..1e22f7d 100644
--- a/app/static/index.html
+++ b/app/static/index.html
@@ -5,6 +5,22 @@
팀플링 - Teampling
+
+
@@ -40,21 +56,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 = `