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 = False,
    ) -> 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, created_by) "
                            "VALUES (%s,%s,%s,%s,%s)",
                            (pid, stage_name, i, is_done, store.norm_str(created_by, lower=True)),
                        )
                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, created_by: str = ""
    ) -> 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, created_by) "
                "VALUES (%s,%s,%s,%s,%s) RETURNING *",
                (project_id, nm, int(nxt["n"]), bool(is_done_stage),
                 store.norm_str(created_by, lower=True)),
            ).fetchone()
        return self._serialize(row)

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

    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)
    # ════════════════════════════════════════════════════════════
    # 업무 목록에 함께 붙이는 하위업무 집계(댓글수와 같은 방식). 여러 쿼리에서
    # 재사용하므로 상수로 뽑아둔다.
    _SUBTASK_COUNTS_SQL = (
        "(SELECT COUNT(*) FROM tasks st WHERE st.parent_task_id = t.id) AS subtask_total, "
        "(SELECT COUNT(*) FROM tasks st WHERE st.parent_task_id = t.id "
        "        AND st.completed_at IS NOT NULL) AS subtask_done"
    )

    def list_tasks(
        self,
        *,
        project_id: int | None = None,
        assignee_email: str | None = None,
        include_subprojects: bool = False,
        parent_task_id: int | None = None,
        top_level_only: bool = True,
        include_links: bool = True,
        limit: int = 1000,
    ) -> list[dict[str, Any]]:
        """업무 목록.

        기본은 최상위 업무만(`parent_task_id IS NULL`) — 하위 업무는 부모 안에서만
        보인다(아사나와 동일, 보드/캘린더/리스트 카드로 따로 뜨지 않는다).
        `parent_task_id` 를 주면 그 부모의 하위 업무만 받는다.

        `project_id` 하나만 지정하고(`include_subprojects=False`, 최상위 조회일
        때) `include_links=True`(기본값)이면, 그 프로젝트에 멀티호밍
        (`task_project_links`)으로 연결된 다른 프로젝트의 업무도 함께 반환한다.
        프로젝트 상세 화면이 '이 프로젝트만' 보여줄 때 쓰는 경로가 이 조합이다.
        다른 조합(홈 전체보기·내 업무 등, `project_id=None`)은 연결 업무를 섞지
        않는다 — 같은 업무가 여러 곳에서 중복으로 집계되는 것을 막기 위함이다.
        """
        link_mode = (
            include_links and project_id is not None
            and not include_subprojects and parent_task_id is None
        )
        if link_mode:
            return self._list_tasks_for_project_with_links(
                project_id=project_id, assignee_email=assignee_email, limit=limit,
            )

        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))
        if parent_task_id is not None:
            clauses.append("t.parent_task_id = %s")
            params.append(parent_task_id)
        elif top_level_only:
            clauses.append("t.parent_task_id IS NULL")
        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"       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, "
                f"       {self._SUBTASK_COUNTS_SQL} "
                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 _list_tasks_for_project_with_links(
        self, *, project_id: int, assignee_email: str | None, limit: int,
    ) -> list[dict[str, Any]]:
        """`list_tasks` 의 "이 프로젝트 + 여기 연결된 업무" 경로.

        `t.*` 뒤에 같은 이름(`stage_id`)을 다시 셀렉트해 덮어쓴다 — psycopg 의
        dict_row 는 중복 컬럼명이 있으면 마지막 값으로 덮어쓰므로, "이 프로젝트
        맥락에서의 세션"(연결이면 link.stage_id, 기본홈이면 t.stage_id)이
        최종값이 된다. 원래 소속 프로젝트 배지(`project_name`/`project_color`)는
        `t.project_id` 기준 그대로 둬 "다른 프로젝트에서 온 업무"임을 알 수 있게
        한다.
        """
        clauses = [
            "t.parent_task_id IS NULL",
            "(t.project_id = %(pid)s OR link.project_id = %(pid)s)",
        ]
        params: dict[str, Any] = {"pid": project_id, "limit": int(limit)}
        if assignee_email:
            clauses.append("t.assignee_email = %(assignee)s")
            params["assignee"] = store.norm_str(assignee_email, lower=True)
        where = "WHERE " + " AND ".join(clauses)
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"""
                SELECT t.*, p.name AS project_name, p.color AS project_color,
                       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count,
                       {self._SUBTASK_COUNTS_SQL},
                       (link.id IS NOT NULL) AS is_linked,
                       COALESCE(es.name, s.name) AS stage_name,
                       COALESCE(es.is_done_stage, s.is_done_stage) AS is_done_stage,
                       COALESCE(link.stage_id, t.stage_id) AS stage_id
                FROM tasks t
                JOIN projects p ON p.id = t.project_id
                LEFT JOIN project_stages s ON s.id = t.stage_id
                LEFT JOIN task_project_links link
                       ON link.task_id = t.id AND link.project_id = %(pid)s
                      AND t.project_id <> %(pid)s
                LEFT JOIN project_stages es ON es.id = link.stage_id
                {where}
                ORDER BY t.sort_order ASC, t.id ASC LIMIT %(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(
                f"SELECT t.*, p.name AS project_name, p.color AS project_color, "
                f"       s.name AS stage_name, s.is_done_stage, "
                f"       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, "
                f"       {self._SUBTASK_COUNTS_SQL} "
                f"FROM tasks t JOIN projects p ON p.id = t.project_id "
                f"LEFT JOIN project_stages s ON s.id = t.stage_id "
                f"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 = "",
        parent_task_id: int | None = None,
    ) -> 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():
                if parent_task_id is not None:
                    # 하위 업무는 부모와 같은 프로젝트에 속하고(담당자 선택지 등이
                    # 맞아떨어지게), 보드에는 안 보이므로 세션을 갖지 않는다.
                    parent = conn.execute(
                        "SELECT project_id FROM tasks WHERE id = %s", (parent_task_id,)
                    ).fetchone()
                    if not parent:
                        raise KeyError(f"상위 업무를 찾을 수 없습니다: {parent_task_id}")
                    project_id = parent["project_id"]
                    stage_id = None
                elif stage_id is None:
                    # stage 미지정 시 프로젝트의 첫 세션으로
                    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, parent_task_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,%s)
                    RETURNING *
                    """,
                    (
                        project_id, stage_id, parent_task_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 자동 세팅. 단, 실제로 단계가
        # 바뀐 경우에만 판정한다 — 업무 편집 모달의 "저장"은 매번 현재 stage_id
        # 를 그대로 다시 보내는데, 여기서 무조건 판정하면 방금 "완료로 표시"로
        # 켠 completed_at 을(단계는 그대로인데) 다시 꺼버리는 버그가 있었다.
        if "stage_id" in fields:
            new_stage = fields["stage_id"]
            sets.append("stage_id = %s")
            params.append(new_stage)
            if new_stage != prev.get("stage_id"):
                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")
        # 직접 완료 토글 — 하위 업무는 세션(칸반 컬럼)이 없어 위 자동 세팅이
        # 적용되지 않는다. "stage_id" 를 같이 보냈으면 그쪽 판정이 우선한다.
        elif "completed" in fields:
            if fields["completed"] and not prev.get("completed_at"):
                sets.append("completed_at = %s")
                params.append(now_kst())
                newly_completed = True
            elif not fields["completed"] 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"])

    # ════════════════════════════════════════════════════════════
    # 멀티호밍 (task_project_links) — 업무 1개가 여러 프로젝트/세션에 동시 소속.
    # 원래 소속(tasks.project_id/stage_id)은 여기서 건드리지 않는다("기본 홈"은
    # 업무 삭제로만 없어진다). 같은 프로젝트에 중복 연결하면 세션만 갱신한다.
    # ════════════════════════════════════════════════════════════
    def add_task_link(
        self, *, task_id: int, project_id: int, stage_id: int | None = None,
        created_by: str = "",
    ) -> dict[str, Any]:
        with self._pool.connection() as conn:
            row = conn.execute(
                """
                INSERT INTO task_project_links (task_id, project_id, stage_id, created_by)
                VALUES (%s,%s,%s,%s)
                ON CONFLICT (task_id, project_id) DO UPDATE
                   SET stage_id = EXCLUDED.stage_id
                RETURNING *
                """,
                (task_id, project_id, stage_id, store.norm_str(created_by, lower=True)),
            ).fetchone()
        return self._serialize(row)

    def remove_task_link(self, *, task_id: int, project_id: int) -> None:
        with self._pool.connection() as conn:
            conn.execute(
                "DELETE FROM task_project_links WHERE task_id = %s AND project_id = %s",
                (task_id, project_id),
            )

    def list_task_links(self, *, task_id: int) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                """
                SELECT l.*, p.name AS project_name, p.color AS project_color,
                       s.name AS stage_name
                FROM task_project_links l
                JOIN projects p ON p.id = l.project_id
                LEFT JOIN project_stages s ON s.id = l.stage_id
                WHERE l.task_id = %s
                ORDER BY l.created_at ASC
                """,
                (task_id,),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    # ════════════════════════════════════════════════════════════
    # 활동 이력
    # ════════════════════════════════════════════════════════════
    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),
                    )

    def reorder_tasks(self, *, stage_id: int, ordered_ids: list[int]) -> None:
        """보드 카드 드래그로 같은 세션 안에서 순서 바꾸기. sort_order 는
        업무 생성 시 프로젝트 전체 기준으로 매겨지지만(다른 세션과 섞여도
        무방 — 화면은 항상 세션으로 먼저 필터링한 뒤 이 값으로 정렬한다),
        여기서는 이 세션 소속 업무만 골라 0부터 다시 매긴다."""
        with self._pool.connection() as conn:
            with conn.transaction():
                for i, tid in enumerate(ordered_ids):
                    conn.execute(
                        "UPDATE tasks SET sort_order = %s "
                        "WHERE id = %s AND stage_id = %s",
                        (i, int(tid), stage_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

    # ════════════════════════════════════════════════════════════
    # 실시간 새로고침 — 프로젝트/업무/댓글/멤버/단계 중 가장 최근 변경 시각
    # ════════════════════════════════════════════════════════════
    def latest_version(self) -> str:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT GREATEST("
                "  COALESCE((SELECT MAX(updated_at) FROM projects), 'epoch'::timestamptz),"
                "  COALESCE((SELECT MAX(updated_at) FROM tasks), 'epoch'::timestamptz),"
                "  COALESCE((SELECT MAX(created_at) FROM task_comments), 'epoch'::timestamptz),"
                "  COALESCE((SELECT MAX(created_at) FROM project_members), 'epoch'::timestamptz),"
                "  COALESCE((SELECT MAX(created_at) FROM project_stages), 'epoch'::timestamptz)"
                ") AS ts"
            ).fetchone()
        ts = row["ts"] if row else None
        return ts.isoformat() if ts else ""

    # ════════════════════════════════════════════════════════════
    # 직렬화
    # ════════════════════════════════════════════════════════════
    @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",
                  "parent_task_id"):
            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