dbx-main/app/modules/project/db.py

"""project_db PostgreSQL 저장소.

- 드라이버: psycopg 3 (`psycopg[binary,pool]`) — 다른 모듈과 동일 패턴.
- 연결 정보: 환경변수 `PROJECT_DB_URL`
  (예: postgresql://project_app:<pwd>@postgres-db:5432/project_db)
- 스키마는 앱이 만들지 않는다. `scripts/sql/project_db_init.sql` 을 superuser 가
  사전 적용한다. 앱 계정(project_app)은 CRUD 권한만 받는다.
- 연결 풀은 lazy open.

순수 검증/상수는 store.py 에 있고, 여기서는 DB I/O 와 직렬화만 담당한다.
"""

from __future__ import annotations

import logging
from datetime import date, datetime, time
from typing import Any

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool

from app.timezone import KST, now_kst

from . import store

logger = logging.getLogger("project.db")


class ProjectStore:
    def __init__(self, dsn: str, *, min_size: int = 1, max_size: int = 5):
        self._pool = ConnectionPool(
            conninfo=dsn,
            min_size=min_size,
            max_size=max_size,
            kwargs={"row_factory": dict_row, "autocommit": True},
            open=False,
        )
        self._pool.open(wait=False)

    def close(self) -> None:
        self._pool.close()

    # ════════════════════════════════════════════════════════════
    # 프로젝트 / 서브프로젝트
    # ════════════════════════════════════════════════════════════
    def list_projects(
        self, *, include_archived: bool = False, member_email: str | None = None
    ) -> list[dict[str, Any]]:
        """전체 프로젝트(서브 포함) 평면 목록. member_email 지정 시 그 사용자가
        멤버이거나 owner 인 프로젝트만(관리자는 라우터에서 None 으로 전체 조회).
        """
        clauses: list[str] = []
        params: list[Any] = []
        if not include_archived:
            clauses.append("p.status <> 'archived'")
        if member_email:
            clauses.append(
                "(p.owner_email = %s OR EXISTS ("
                "  SELECT 1 FROM project_members m "
                "  WHERE m.project_id = p.id AND m.user_email = %s))"
            )
            params.extend([member_email, member_email])
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"SELECT p.* FROM projects p {where} "
                "ORDER BY p.parent_id NULLS FIRST, p.sort_order ASC, p.id ASC",
                params,
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def get_project(self, *, project_id: int) -> dict[str, Any] | None:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT * FROM projects WHERE id = %s", (project_id,)
            ).fetchone()
        return self._serialize(row) if row else None

    def create_project(
        self,
        *,
        name: str,
        parent_id: int | None = None,
        description: str = "",
        color: str = "",
        owner_email: str = "",
        start_date: str | None = None,
        due_date: str | None = None,
        created_by: str = "",
        seed_stages: bool = True,
    ) -> dict[str, Any]:
        nm = store.validate_project_name(name)
        col = store.validate_color(color)
        sd = store.validate_date(start_date)
        dd = store.validate_date(due_date)
        with self._pool.connection() as conn:
            with conn.transaction():
                if parent_id is not None:
                    parent = conn.execute(
                        "SELECT id FROM projects WHERE id = %s", (parent_id,)
                    ).fetchone()
                    if not parent:
                        raise KeyError(f"상위 프로젝트를 찾을 수 없습니다: {parent_id}")
                row = conn.execute(
                    """
                    INSERT INTO projects
                        (parent_id, name, description, color, owner_email,
                         start_date, due_date, created_by)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
                    RETURNING *
                    """,
                    (
                        parent_id,
                        nm,
                        store.norm_str(description),
                        col,
                        store.norm_str(owner_email, lower=True),
                        sd,
                        dd,
                        store.norm_str(created_by, lower=True),
                    ),
                ).fetchone()
                pid = row["id"]
                if seed_stages:
                    for i, (stage_name, is_done) in enumerate(store.DEFAULT_STAGES):
                        conn.execute(
                            "INSERT INTO project_stages "
                            "(project_id, name, sort_order, is_done_stage) "
                            "VALUES (%s,%s,%s,%s)",
                            (pid, stage_name, i, is_done),
                        )
                self._log(conn, project_id=pid, actor=created_by,
                          action=store.ACTION_CREATED, detail=nm)
        return self._serialize(row)

    def update_project(self, *, project_id: int, fields: dict[str, Any]) -> dict[str, Any]:
        allowed = {
            "name": lambda v: store.validate_project_name(v),
            "description": lambda v: store.norm_str(v),
            "color": lambda v: store.validate_color(v),
            "owner_email": lambda v: store.norm_str(v, lower=True),
            "start_date": lambda v: store.validate_date(v),
            "due_date": lambda v: store.validate_date(v),
            "status": lambda v: (store.norm_str(v, lower=True)
                                 if store.norm_str(v, lower=True) in store.PROJECT_STATUSES
                                 else "active"),
        }
        sets: list[str] = []
        params: list[Any] = []
        for key, fn in allowed.items():
            if key in fields:
                sets.append(f"{key} = %s")
                params.append(fn(fields[key]))
        if not sets:
            existing = self.get_project(project_id=project_id)
            if existing is None:
                raise KeyError(project_id)
            return existing
        params.append(project_id)
        with self._pool.connection() as conn:
            row = conn.execute(
                f"UPDATE projects SET {', '.join(sets)} WHERE id = %s RETURNING *",
                params,
            ).fetchone()
        if not row:
            raise KeyError(project_id)
        return self._serialize(row)

    def delete_project(self, *, project_id: int) -> None:
        """프로젝트(서브프로젝트·단계·업무 CASCADE) 삭제."""
        with self._pool.connection() as conn:
            cur = conn.execute("DELETE FROM projects WHERE id = %s", (project_id,))
            if cur.rowcount == 0:
                raise KeyError(project_id)

    # ════════════════════════════════════════════════════════════
    # 멤버
    # ════════════════════════════════════════════════════════════
    def list_members(self, *, project_id: int) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM project_members WHERE project_id = %s "
                "ORDER BY role DESC, user_email ASC",
                (project_id,),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def is_member(self, *, project_id: int, user_email: str) -> bool:
        email = store.norm_str(user_email, lower=True)
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT 1 FROM projects p "
                "LEFT JOIN project_members m "
                "  ON m.project_id = p.id AND m.user_email = %s "
                "WHERE p.id = %s AND (p.owner_email = %s OR m.id IS NOT NULL) LIMIT 1",
                (email, project_id, email),
            ).fetchone()
        return bool(row)

    def add_member(
        self, *, project_id: int, user_email: str, user_name: str = "", role: str = "member"
    ) -> dict[str, Any]:
        email = store.norm_str(user_email, lower=True)
        if not email or "@" not in email:
            raise ValueError("올바른 이메일이 아닙니다.")
        rl = store.validate_role(role)
        with self._pool.connection() as conn:
            row = conn.execute(
                """
                INSERT INTO project_members (project_id, user_email, user_name, role)
                VALUES (%s,%s,%s,%s)
                ON CONFLICT (project_id, user_email) DO UPDATE
                   SET role = EXCLUDED.role, user_name = EXCLUDED.user_name
                RETURNING *
                """,
                (project_id, email, store.norm_str(user_name), rl),
            ).fetchone()
        return self._serialize(row)

    def remove_member(self, *, project_id: int, user_email: str) -> None:
        email = store.norm_str(user_email, lower=True)
        with self._pool.connection() as conn:
            conn.execute(
                "DELETE FROM project_members WHERE project_id = %s AND user_email = %s",
                (project_id, email),
            )

    # ════════════════════════════════════════════════════════════
    # 단계 (stages)
    # ════════════════════════════════════════════════════════════
    def list_stages(self, *, project_id: int) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM project_stages WHERE project_id = %s "
                "ORDER BY sort_order ASC, id ASC",
                (project_id,),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def add_stage(
        self, *, project_id: int, name: str, is_done_stage: bool = False
    ) -> dict[str, Any]:
        nm = store.norm_str(name)
        if not nm:
            raise ValueError("단계 이름은 필수입니다.")
        with self._pool.connection() as conn:
            nxt = conn.execute(
                "SELECT COALESCE(MAX(sort_order), -1) + 1 AS n "
                "FROM project_stages WHERE project_id = %s",
                (project_id,),
            ).fetchone()
            row = conn.execute(
                "INSERT INTO project_stages (project_id, name, sort_order, is_done_stage) "
                "VALUES (%s,%s,%s,%s) RETURNING *",
                (project_id, nm, int(nxt["n"]), bool(is_done_stage)),
            ).fetchone()
        return self._serialize(row)

    def delete_stage(self, *, stage_id: int) -> None:
        with self._pool.connection() as conn:
            cur = conn.execute("DELETE FROM project_stages WHERE id = %s", (stage_id,))
            if cur.rowcount == 0:
                raise KeyError(stage_id)

    # ════════════════════════════════════════════════════════════
    # 업무 (tasks)
    # ════════════════════════════════════════════════════════════
    def list_tasks(
        self,
        *,
        project_id: int | None = None,
        assignee_email: str | None = None,
        include_subprojects: bool = False,
        limit: int = 1000,
    ) -> list[dict[str, Any]]:
        clauses: list[str] = []
        params: list[Any] = []
        if project_id is not None:
            if include_subprojects:
                clauses.append(
                    "t.project_id IN ("
                    "  WITH RECURSIVE tree AS ("
                    "    SELECT id FROM projects WHERE id = %s "
                    "    UNION ALL "
                    "    SELECT p.id FROM projects p JOIN tree ON p.parent_id = tree.id"
                    "  ) SELECT id FROM tree)"
                )
                params.append(project_id)
            else:
                clauses.append("t.project_id = %s")
                params.append(project_id)
        if assignee_email:
            clauses.append("t.assignee_email = %s")
            params.append(store.norm_str(assignee_email, lower=True))
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        params.append(int(limit))
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"SELECT t.*, p.name AS project_name, p.color AS project_color, "
                f"       s.name AS stage_name, s.is_done_stage "
                f"FROM tasks t "
                f"JOIN projects p ON p.id = t.project_id "
                f"LEFT JOIN project_stages s ON s.id = t.stage_id "
                f"{where} "
                "ORDER BY t.sort_order ASC, t.id ASC LIMIT %s",
                params,
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def get_task(self, *, task_id: int) -> dict[str, Any] | None:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT t.*, p.name AS project_name, p.color AS project_color, "
                "       s.name AS stage_name, s.is_done_stage "
                "FROM tasks t JOIN projects p ON p.id = t.project_id "
                "LEFT JOIN project_stages s ON s.id = t.stage_id "
                "WHERE t.id = %s",
                (task_id,),
            ).fetchone()
        return self._serialize(row) if row else None

    def create_task(
        self,
        *,
        project_id: int,
        title: str,
        stage_id: int | None = None,
        description: str = "",
        assignee_email: str = "",
        assignee_name: str = "",
        priority: str = "normal",
        start_date: str | None = None,
        due_date: str | None = None,
        start_time: str | None = None,
        due_time: str | None = None,
        created_by: str = "",
    ) -> dict[str, Any]:
        ttl = store.validate_task_title(title)
        pr = store.validate_priority(priority)
        sd = store.validate_date(start_date)
        dd = store.validate_date(due_date)
        sti = store.validate_time(start_time)
        dti = store.validate_time(due_time)
        assignee = store.norm_str(assignee_email, lower=True)
        with self._pool.connection() as conn:
            with conn.transaction():
                # stage 미지정 시 프로젝트의 첫 단계로
                if stage_id is None:
                    st = conn.execute(
                        "SELECT id FROM project_stages WHERE project_id = %s "
                        "ORDER BY sort_order ASC, id ASC LIMIT 1",
                        (project_id,),
                    ).fetchone()
                    stage_id = st["id"] if st else None
                nxt = conn.execute(
                    "SELECT COALESCE(MAX(sort_order), -1) + 1 AS n FROM tasks "
                    "WHERE project_id = %s",
                    (project_id,),
                ).fetchone()
                row = conn.execute(
                    """
                    INSERT INTO tasks
                        (project_id, stage_id, title, description, assignee_email,
                         assignee_name, priority, start_date, due_date,
                         start_time, due_time, sort_order, created_by)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                    RETURNING *
                    """,
                    (
                        project_id, stage_id, ttl, store.norm_str(description),
                        assignee, store.norm_str(assignee_name), pr, sd, dd,
                        sti, dti, int(nxt["n"]), store.norm_str(created_by, lower=True),
                    ),
                ).fetchone()
                if assignee:
                    self._log(conn, project_id=project_id, task_id=row["id"],
                              actor=created_by, action=store.ACTION_ASSIGNED,
                              detail=f"{ttl} → {assignee}")
        return self.get_task(task_id=row["id"])

    def update_task(
        self, *, task_id: int, fields: dict[str, Any], actor: str = ""
    ) -> dict[str, Any]:
        """업무 수정. 반환에 알림 힌트를 담는다(라우터가 메일 발송 판단).

        반환 dict 에 추가되는 키:
          _newly_assigned: 새 담당자 이메일 or None
          _newly_completed: 완료로 전환됐으면 True
        """
        prev = self.get_task(task_id=task_id)
        if prev is None:
            raise KeyError(task_id)

        sets: list[str] = []
        params: list[Any] = []
        newly_assigned: str | None = None
        newly_completed = False

        if "title" in fields:
            sets.append("title = %s")
            params.append(store.validate_task_title(fields["title"]))
        if "description" in fields:
            sets.append("description = %s")
            params.append(store.norm_str(fields["description"]))
        if "priority" in fields:
            sets.append("priority = %s")
            params.append(store.validate_priority(fields["priority"]))
        if "start_date" in fields:
            sets.append("start_date = %s")
            params.append(store.validate_date(fields["start_date"]))
        if "due_date" in fields:
            sets.append("due_date = %s")
            params.append(store.validate_date(fields["due_date"]))
        if "start_time" in fields:
            sets.append("start_time = %s")
            params.append(store.validate_time(fields["start_time"]))
        if "due_time" in fields:
            sets.append("due_time = %s")
            params.append(store.validate_time(fields["due_time"]))
        if "assignee_email" in fields:
            new_assignee = store.norm_str(fields["assignee_email"], lower=True)
            sets.append("assignee_email = %s")
            params.append(new_assignee)
            if "assignee_name" in fields:
                sets.append("assignee_name = %s")
                params.append(store.norm_str(fields["assignee_name"]))
            if new_assignee and new_assignee != (prev.get("assignee_email") or ""):
                newly_assigned = new_assignee
        elif "assignee_name" in fields:
            sets.append("assignee_name = %s")
            params.append(store.norm_str(fields["assignee_name"]))

        # 단계 변경 → 완료단계 진입 시 completed_at 자동 세팅
        if "stage_id" in fields:
            new_stage = fields["stage_id"]
            sets.append("stage_id = %s")
            params.append(new_stage)
            done = self._stage_is_done(new_stage)
            if done and not prev.get("completed_at"):
                sets.append("completed_at = %s")
                params.append(now_kst())
                newly_completed = True
            elif not done and prev.get("completed_at"):
                sets.append("completed_at = NULL")

        if not sets:
            return {**prev, "_newly_assigned": None, "_newly_completed": False}

        params.append(task_id)
        with self._pool.connection() as conn:
            with conn.transaction():
                conn.execute(
                    f"UPDATE tasks SET {', '.join(sets)} WHERE id = %s", params
                )
                if newly_assigned:
                    self._log(conn, project_id=prev["project_id"], task_id=task_id,
                              actor=actor, action=store.ACTION_ASSIGNED,
                              detail=f"{prev['title']} → {newly_assigned}")
                if newly_completed:
                    self._log(conn, project_id=prev["project_id"], task_id=task_id,
                              actor=actor, action=store.ACTION_COMPLETED,
                              detail=prev["title"])
        out = self.get_task(task_id=task_id)
        out["_newly_assigned"] = newly_assigned
        out["_newly_completed"] = newly_completed
        return out

    def delete_task(self, *, task_id: int) -> None:
        with self._pool.connection() as conn:
            cur = conn.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
            if cur.rowcount == 0:
                raise KeyError(task_id)

    def _stage_is_done(self, stage_id: Any) -> bool:
        if stage_id is None:
            return False
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT is_done_stage FROM project_stages WHERE id = %s", (stage_id,)
            ).fetchone()
        return bool(row and row["is_done_stage"])

    # ════════════════════════════════════════════════════════════
    # 활동 이력
    # ════════════════════════════════════════════════════════════
    def list_activity(self, *, project_id: int, limit: int = 100) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM project_activity WHERE project_id = %s "
                "ORDER BY created_at DESC, id DESC LIMIT %s",
                (project_id, int(limit)),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    @staticmethod
    def _log(
        conn: Any, *, project_id: int | None = None, task_id: int | None = None,
        actor: str = "", action: str = "", detail: str = "",
    ) -> None:
        conn.execute(
            "INSERT INTO project_activity "
            "(project_id, task_id, actor_email, action, detail) "
            "VALUES (%s,%s,%s,%s,%s)",
            (project_id, task_id, store.norm_str(actor, lower=True), action,
             store.norm_str(detail)),
        )

    # ════════════════════════════════════════════════════════════
    # 댓글 (task_comments)
    # ════════════════════════════════════════════════════════════
    def list_comments(self, *, task_id: int) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM task_comments WHERE task_id = %s "
                "ORDER BY created_at ASC, id ASC",
                (task_id,),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def add_comment(
        self, *, task_id: int, author_email: str, author_name: str, body: str
    ) -> dict[str, Any]:
        text = store.norm_str(body)
        if not text:
            raise ValueError("댓글 내용이 비었습니다.")
        with self._pool.connection() as conn:
            row = conn.execute(
                "INSERT INTO task_comments (task_id, author_email, author_name, body) "
                "VALUES (%s,%s,%s,%s) RETURNING *",
                (task_id, store.norm_str(author_email, lower=True),
                 store.norm_str(author_name), text),
            ).fetchone()
        return self._serialize(row)

    def update_comment(
        self, *, comment_id: int, requester: str, is_admin: bool, body: str
    ) -> dict[str, Any]:
        text = store.norm_str(body)
        if not text:
            raise ValueError("댓글 내용이 비었습니다.")
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT author_email FROM task_comments WHERE id = %s", (comment_id,)
            ).fetchone()
            if not row:
                raise KeyError(comment_id)
            if not is_admin and row["author_email"] != store.norm_str(requester, lower=True):
                raise PermissionError("본인 댓글만 수정할 수 있습니다.")
            out = conn.execute(
                "UPDATE task_comments SET body = %s WHERE id = %s RETURNING *",
                (text, comment_id),
            ).fetchone()
        return self._serialize(out)

    def delete_comment(self, *, comment_id: int, requester: str, is_admin: bool) -> None:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT author_email FROM task_comments WHERE id = %s", (comment_id,)
            ).fetchone()
            if not row:
                raise KeyError(comment_id)
            if not is_admin and row["author_email"] != store.norm_str(requester, lower=True):
                raise PermissionError("본인 댓글만 삭제할 수 있습니다.")
            conn.execute("DELETE FROM task_comments WHERE id = %s", (comment_id,))

    # ════════════════════════════════════════════════════════════
    # 첨부 (task_attachments) — 메타만 저장. 파일은 라우터가 디스크 처리.
    # ════════════════════════════════════════════════════════════
    def list_attachments(self, *, task_id: int) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM task_attachments WHERE task_id = %s "
                "ORDER BY created_at DESC, id DESC",
                (task_id,),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def add_attachment(
        self, *, task_id: int, filename: str, stored_name: str,
        content_type: str, size_bytes: int, uploaded_by: str,
    ) -> dict[str, Any]:
        with self._pool.connection() as conn:
            row = conn.execute(
                "INSERT INTO task_attachments "
                "(task_id, filename, stored_name, content_type, size_bytes, uploaded_by) "
                "VALUES (%s,%s,%s,%s,%s,%s) RETURNING *",
                (task_id, store.norm_str(filename), stored_name,
                 store.norm_str(content_type), int(size_bytes),
                 store.norm_str(uploaded_by, lower=True)),
            ).fetchone()
        return self._serialize(row)

    def get_attachment(self, *, attachment_id: int) -> dict[str, Any] | None:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT * FROM task_attachments WHERE id = %s", (attachment_id,)
            ).fetchone()
        return self._serialize(row) if row else None

    def delete_attachment(self, *, attachment_id: int) -> dict[str, Any]:
        """삭제 후 행(파일 경로 정리용 stored_name 포함)을 반환."""
        with self._pool.connection() as conn:
            row = conn.execute(
                "DELETE FROM task_attachments WHERE id = %s RETURNING *", (attachment_id,)
            ).fetchone()
        if not row:
            raise KeyError(attachment_id)
        return self._serialize(row)

    # ════════════════════════════════════════════════════════════
    # 단계 순서/이름 편집
    # ════════════════════════════════════════════════════════════
    def update_stage(
        self, *, stage_id: int, name: str | None = None,
        is_done_stage: bool | None = None,
    ) -> dict[str, Any]:
        sets: list[str] = []
        params: list[Any] = []
        if name is not None:
            nm = store.norm_str(name)
            if not nm:
                raise ValueError("단계 이름은 필수입니다.")
            sets.append("name = %s")
            params.append(nm)
        if is_done_stage is not None:
            sets.append("is_done_stage = %s")
            params.append(bool(is_done_stage))
        if not sets:
            raise ValueError("변경할 내용이 없습니다.")
        params.append(stage_id)
        with self._pool.connection() as conn:
            row = conn.execute(
                f"UPDATE project_stages SET {', '.join(sets)} WHERE id = %s RETURNING *",
                params,
            ).fetchone()
        if not row:
            raise KeyError(stage_id)
        return self._serialize(row)

    def reorder_stages(self, *, project_id: int, ordered_ids: list[int]) -> None:
        """주어진 순서대로 sort_order 재배정. 해당 프로젝트 단계만 갱신."""
        with self._pool.connection() as conn:
            with conn.transaction():
                for i, sid in enumerate(ordered_ids):
                    conn.execute(
                        "UPDATE project_stages SET sort_order = %s "
                        "WHERE id = %s AND project_id = %s",
                        (i, int(sid), project_id),
                    )

    # ════════════════════════════════════════════════════════════
    # 알림센터 (project_notifications)
    # ════════════════════════════════════════════════════════════
    def add_notification(
        self, *, recipient_email: str, actor_email: str = "",
        project_id: int | None = None, task_id: int | None = None,
        type: str = "", title: str = "", body: str = "",
    ) -> None:
        rcpt = store.norm_str(recipient_email, lower=True)
        if not rcpt:
            return
        # 본인이 본인에게 보내는 알림은 생략(아사나도 자기 행동은 알림 안 함)
        if rcpt == store.norm_str(actor_email, lower=True):
            return
        with self._pool.connection() as conn:
            conn.execute(
                "INSERT INTO project_notifications "
                "(recipient_email, actor_email, project_id, task_id, type, title, body) "
                "VALUES (%s,%s,%s,%s,%s,%s,%s)",
                (rcpt, store.norm_str(actor_email, lower=True), project_id, task_id,
                 type, store.norm_str(title), store.norm_str(body)),
            )

    def list_notifications(
        self, *, recipient_email: str, unread_only: bool = False, limit: int = 100
    ) -> list[dict[str, Any]]:
        rcpt = store.norm_str(recipient_email, lower=True)
        clause = "WHERE recipient_email = %s"
        params: list[Any] = [rcpt]
        if unread_only:
            clause += " AND is_read = FALSE"
        params.append(int(limit))
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"SELECT * FROM project_notifications {clause} "
                "ORDER BY created_at DESC, id DESC LIMIT %s",
                params,
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def count_unread(self, *, recipient_email: str) -> int:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT COUNT(*) AS n FROM project_notifications "
                "WHERE recipient_email = %s AND is_read = FALSE",
                (store.norm_str(recipient_email, lower=True),),
            ).fetchone()
        return int(row["n"] or 0)

    def mark_read(self, *, notification_id: int, recipient_email: str) -> None:
        with self._pool.connection() as conn:
            conn.execute(
                "UPDATE project_notifications SET is_read = TRUE "
                "WHERE id = %s AND recipient_email = %s",
                (notification_id, store.norm_str(recipient_email, lower=True)),
            )

    def mark_all_read(self, *, recipient_email: str) -> int:
        with self._pool.connection() as conn:
            cur = conn.execute(
                "UPDATE project_notifications SET is_read = TRUE "
                "WHERE recipient_email = %s AND is_read = FALSE",
                (store.norm_str(recipient_email, lower=True),),
            )
        return cur.rowcount or 0

    # ════════════════════════════════════════════════════════════
    # 직렬화
    # ════════════════════════════════════════════════════════════
    @staticmethod
    def _serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
        if row is None:
            return None
        out = dict(row)
        for k, v in list(out.items()):
            if isinstance(v, datetime):
                out[k] = v.astimezone(KST).isoformat(timespec="seconds")
            elif isinstance(v, time):
                out[k] = v.strftime("%H:%M")
            elif isinstance(v, date):
                out[k] = v.isoformat()
        for k in ("id", "parent_id", "project_id", "stage_id", "task_id", "sort_order"):
            if k in out and out[k] is not None:
                try:
                    out[k] = int(out[k])
                except (TypeError, ValueError):
                    pass
        for k in ("is_done_stage", "is_read"):
            if k in out and out[k] is not None:
                out[k] = bool(out[k])
        return out