# Task Survey 答案重複寫入 Bug 修復 — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 消除 `task_survey` 模組的答案重複寫入 bug，把多人協作 stale data 問題一併修掉，並補髒資料清理與 DB unique constraint 護欄。

**Architecture:**
- Phase 1（止血）：service 層改用 `(task_survey_id, question_id)` natural key 做 upsert；GET 加 SQL `DISTINCT ON` 去重；DB 加 `UNIQUE` constraint，先寫 dedup migration 留 `updated_at` 最新一筆。
- Phase 2（協作一致性）：新增 `patch_task_survey_answer` 單題 service + REST `POST /project-survey/answers/<uid>/patch`；socket `on_update` 改走單題 patch、移除 Redis `fill_survey:<room>:updates`；`on_join` snapshot 改讀 DB。
- Phase 3（效能）：`update_task_survey_answer` / `revert_question_answer_from_history` 預取 + dict map，消除 N+1 SELECT。
- Phase 4（收尾）：跑全套整合測試、寫 changelog、issue 移到 resolved/。

**Tech Stack:** Flask + Flask-RESTful + SQLAlchemy 2.x（thread-local session via `jedi_common`）+ Flask-SocketIO + Redis + PostgreSQL（含 RLS）+ pytest（`tests/` 目錄為整合測試）。

**Reference docs:**
- 對應 issue：`docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md`
- Billows 後端 root cause + 修法：`/Users/chouraymond/Projects/Billows/NICS/billows-be/docs/fixes/20260422_survey_checkpoint_performance/02_plan.md` 與 `03_changelog.md`
- Billows 前端 R1/R2 修法：`/Users/chouraymond/Projects/Billows/NICS/billows-fe/docs/fixes/20260417_survey_fill_data_loss/03_changelog.md`

**Conventions（from CLAUDE.md，每個 task 都套用）:**
- 服務層 public method 一律 `@transaction`；不在 helper 內呼叫 `get_session()`。
- Repo 只 `self.session.flush()`，不要 `commit()`；commit 由 `@transaction` 統一收。
- Route 不查 DB、不 import ORM model；權限/前置條件全在 service 層。
- 新 error code：放 `common/code/error_code.py`，命名 `TASK_SURVEY_<HTTP><序號>`。
- SQL migration：開頭 `-- Date: YYYY-MM-DD`；每段 `-- N. 說明 (YYYY-MM-DD)`；新表/新權限段補 `GRANT`（本案改現有表，無新表）。
- audit fields：`created_user` / `updated_user` 是 login_name，response 若有需同時帶 `*_name`；本次只動寫入路徑，response 不改格式。

---

## File Structure

| 檔案 | 動作 | 責任 |
|------|------|------|
| `scripts/sql/2026-04-28-survey-answer-dedup.sql` | 新增 | 髒資料清理 + 加 `UNIQUE (task_survey_id, question_id)` |
| `app/task_survey/service/question_answer_service.py` | 改 | upsert 改 natural key、GET 去重、新增 `patch_task_survey_answer`、預取 + map |
| `app/task_survey/service/question_answer_history_service.py` | 改 | revert 改用預取 + map（消 N+1） |
| `infra/task_survey/repository/question_answer_repo_impl.py` | 改 | 新增 `get_latest_by_task_survey_id`（DISTINCT ON）、`upsert_by_natural_key` |
| `domain/task_survey/repository/question_answer.py` | 改 | 抽象方法簽名跟著加 |
| `domain/task_survey/service/question_answer_domain_service.py` | 改 | wrap 新 repo 方法 |
| `app/task_survey/handler/fill_survey_socketio_handler.py` | 改 | `on_update` 改單題 patch、移除 Redis updates；`on_join` 從 DB 讀 |
| `api/task_survey/routes/question_answer_route.py` | 改 | 新增 `TaskSurveyAnswerPatchRoute`（POST `/project-survey/answers/<uid>/patch`） |
| `api/task_survey/serializers/question_answer.py` | 新增 | `AnswerPatchItemSchema` / `AnswerPatchRequestSchema` |
| `api/task_survey/__init__.py` | 改 | 註冊新 patch route |
| `common/code/error_code.py` | 改 | 新 error code（patch question_uid not found 等） |
| `tests/test_task_survey_answer.py` | 新增 | 整合測試覆蓋 PUT / patch / socket 三條路徑 |
| `docs/changelog/2026-04-28-fix-task-survey-answer-duplicate-phase1.md` | 新增 | Phase 1 changelog |
| `docs/changelog/2026-04-28-fix-task-survey-answer-collab-phase2.md` | 新增 | Phase 2 changelog |
| `docs/changelog/2026-04-28-tweak-task-survey-answer-perf-phase3.md` | 新增 | Phase 3 changelog |
| `docs/issues/resolved/2026-04-28-task-survey-answer-duplicate.md` | rename | Phase 4 收尾 |

---

# Phase 1 — 止血：natural key upsert + DB unique constraint

**Outcome：**
- PUT / socket 任何 payload（帶不帶 uid）都不會再產生重複 row。
- DB 一筆 task_survey_id + question_id 一定唯一（unique constraint 保證）。
- GET 看到的永遠是最新版本（DISTINCT ON）。
- 髒資料清理乾淨。

## Task 1.1：寫整合測試骨架（先 RED）

**Files:**
- Create: `tests/test_task_survey_answer.py`

**前置確認：** `tests/conftest.py` 有現成 `app/client/auth headers` fixture。複製 `tests/test_grc_*.py` 結構即可。

- [ ] **Step 1: 看 conftest 與既有整合測試結構**

```bash
ls tests/conftest.py tests/helpers/ tests/test_grc_*.py | head
```

確認 `auth_headers`、`client`、`session` fixture 名稱。

- [ ] **Step 2: 寫第一個 failing test — 第二次 PUT 不應產生重複 row**

```python
# tests/test_task_survey_answer.py
import pytest

API = "/api/1.0/project-survey/answers"

@pytest.fixture
def seeded_task_survey(client, auth_headers):
    """建立一張 task_survey，回傳 uid 與題目清單。
    依 tests/helpers 內既有 seeding util；若無，直接呼 GRC 啟動專案的 fixture。
    """
    raise NotImplementedError("接 helper")  # 第一輪先讓 collection 就好


def test_put_twice_does_not_duplicate_rows(client, auth_headers, seeded_task_survey, db_session):
    ts_uid = seeded_task_survey["uid"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    payload_1 = {
        "status": 1,
        "answers": {q_uid_1: {"answer": "v1", "score": 0}},
    }
    r1 = client.put(f"{API}/{ts_uid}", json=payload_1, headers=auth_headers)
    assert r1.status_code == 200

    payload_2 = {
        "status": 1,
        "answers": {q_uid_1: {"answer": "v2", "score": 0}},  # 故意不帶 uid
    }
    r2 = client.put(f"{API}/{ts_uid}", json=payload_2, headers=auth_headers)
    assert r2.status_code == 200

    cnt = db_session.execute(
        "SELECT count(*) FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).scalar()
    assert cnt == 1, f"期望 1 row，實際 {cnt}（duplicate bug）"
```

- [ ] **Step 3: 跑測試確認 RED**

```bash
pytest tests/test_task_survey_answer.py::test_put_twice_does_not_duplicate_rows -v
```

預期：`AssertionError: 期望 1 row，實際 2` 或 fixture 未實作。先 commit 失敗的測試骨架。

- [ ] **Step 4: 實作 fixture 補齊**

讀 `tests/test_grc_job.py` / `tests/helpers/` 找到「建立含 task_survey 的 project」最短路徑；複用，不要重新造輪。回填 `seeded_task_survey` 內容（包含 `uid`、`id`、`question_uids` list）。

- [ ] **Step 5: 重跑確認測試現在 fail 在 assertion（不是 fixture error）**

```bash
pytest tests/test_task_survey_answer.py::test_put_twice_does_not_duplicate_rows -v
```

預期：`AssertionError: 期望 1 row，實際 2`（看到 bug 確實重現）

- [ ] **Step 6: Commit RED test**

```bash
git add tests/test_task_survey_answer.py
git commit -m "test(task-survey): add failing test for answer duplicate bug"
```

---

## Task 1.2：service 層改 natural key upsert（GREEN test 1）

**Files:**
- Modify: `app/task_survey/service/question_answer_service.py:49-136`

- [ ] **Step 1: 把 `update_task_survey_answer` 內 line 56-94 換成 natural key upsert**

關鍵差異：
- 不再用 `exist_answers_uid_list`。
- 預先 `existing_by_qid: dict[int, QuestionAnswerEntity] = {a.question_id: a for a in exist_answers}`，重複時保留 `updated_at` 最新（migration 後不該有重複，但 service 為防禦也保留 max 邏輯）。
- 對 `answer_list` 每筆，先 lookup question_id（透過 `survey_question_domain_service.get_questions(survey_id=...)` 一次取，建 `q_uid_to_id`），再去 `existing_by_qid` 找：找到就 update、找不到就 add。

```python
# 替換 question_answer_service.py:54-94 整段
questions = self.survey_question_domain_service.get_questions(survey_id=task_survey.survey_id)
q_uid_to_id = {q.uid: q.id for q in questions}

exist_answers = self.question_answer_domain_service.get_question_answers(
    QuestionAnswerQueryEntity(task_survey_id=task_survey.id)
)
existing_by_qid: dict[int, QuestionAnswerEntity] = {}
for a in exist_answers:
    cur = existing_by_qid.get(a.question_id)
    if cur is None or (a.updated_at and (cur.updated_at is None or a.updated_at > cur.updated_at)):
        existing_by_qid[a.question_id] = a

res_answer_list = []
for question_uid, answer in answers.items():
    question_id = q_uid_to_id.get(question_uid)
    if question_id is None:
        # 整張 survey 沒這題 — 不應該發生，跳過避免抽風
        continue

    payload_answer = answer.get("answer")
    payload_ext_answer = answer.get("ext_answer")
    payload_feedback = answer.get("feedback", "")
    payload_ext_feedback = answer.get("ext_feedback", "")
    payload_score = answer.get("score", 0)

    existing = existing_by_qid.get(question_id)
    if existing is not None:
        existing.score = payload_score
        existing.answer = payload_answer
        existing.ext_answer = payload_ext_answer
        existing.feedback = payload_feedback
        existing.ext_feedback = payload_ext_feedback
        existing.is_delete = 0
        existing.updated_user = user
        self.question_answer_domain_service.update_question_answer(existing)
        res_answer_list.append(existing)
    else:
        new_qa = QuestionAnswerEntity(
            score=payload_score,
            answer=payload_answer,
            ext_answer=payload_ext_answer,
            feedback=payload_feedback,
            ext_feedback=payload_ext_feedback,
            is_delete=0,
            question_id=question_id,
            survey_id=task_survey.survey_id,
            task_survey_id=task_survey.id,
            created_user=user,
            updated_user=user,
        )
        new_qa = self.question_answer_domain_service.add_question_answer(new_qa)
        existing_by_qid[question_id] = new_qa  # 防止同次 payload 同題重複造重複
        res_answer_list.append(new_qa)
```

注意：
- 移除 `answer["question_uid"] = question_uid` 那段、移除 `exist_answers_uid_list` 變數、移除 `new_answers` filter。
- 保留 `is_delete`、`update_user`（注意舊 code 是 `update_user` 而非 `updated_user`，看 entity 欄位名統一）。

- [ ] **Step 2: 重跑 Task 1.1 的測試**

```bash
pytest tests/test_task_survey_answer.py::test_put_twice_does_not_duplicate_rows -v
```

預期：PASS。

- [ ] **Step 3: 加 update value 驗證的測試**

```python
def test_put_twice_updates_existing_value(client, auth_headers, seeded_task_survey, db_session):
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    client.put(f"{API}/{ts_uid}", json={"status": 1, "answers": {q_uid: {"answer": "v1"}}}, headers=auth_headers)
    client.put(f"{API}/{ts_uid}", json={"status": 1, "answers": {q_uid: {"answer": "v2"}}}, headers=auth_headers)

    rows = db_session.execute(
        "SELECT answer FROM survey.question_answers "
        "WHERE task_survey_id = :tid AND question_id IN "
        "  (SELECT id FROM survey.survey_questions WHERE uid = :quid)",
        {"tid": seeded_task_survey["id"], "quid": q_uid},
    ).fetchall()
    assert len(rows) == 1
    assert rows[0][0] == "v2"
```

- [ ] **Step 4: 跑兩個測試確認 GREEN**

```bash
pytest tests/test_task_survey_answer.py -v
```

- [ ] **Step 5: Commit Phase 1 service 修改**

```bash
git add app/task_survey/service/question_answer_service.py tests/test_task_survey_answer.py
git commit -m "fix(task-survey): use natural key upsert in update_task_survey_answer

對應 docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md 根因 §3.1。
不再用 row uid 做 new vs existing 判斷，改用 (task_survey_id, question_id)。"
```

---

## Task 1.3：GET 加 SQL 去重（DISTINCT ON）

**Files:**
- Modify: `infra/task_survey/repository/question_answer_repo_impl.py`
- Modify: `domain/task_survey/repository/question_answer.py`
- Modify: `domain/task_survey/service/question_answer_domain_service.py`
- Modify: `app/task_survey/service/question_answer_service.py:35-46`

- [ ] **Step 1: 寫 failing test：GET 在 DB 有重複時只回最新一筆**

```python
def test_get_dedup_when_dirty_data_exists(client, auth_headers, seeded_task_survey, db_session):
    ts_id = seeded_task_survey["id"]
    q_id = seeded_task_survey["question_id_for"](seeded_task_survey["question_uids"][0])

    # 手插兩筆同 (task_survey_id, question_id)，模擬髒資料（migration 前狀態）
    # 注意：Phase 1.5 加 unique constraint 之後就插不進去；此測試在 1.5 完成後改用 raw SQL with ON CONFLICT DO NOTHING bypass，或拆成「constraint 還沒 enforce 前」的版本
    db_session.execute(
        "INSERT INTO survey.question_answers (task_survey_id, question_id, answer, created_user, updated_user, updated_at) "
        "VALUES (:tid, :qid, '\"old\"', 'seed', 'seed', NOW() - INTERVAL '1 hour')",
        {"tid": ts_id, "qid": q_id},
    )
    db_session.execute(
        "INSERT INTO survey.question_answers (task_survey_id, question_id, answer, created_user, updated_user, updated_at) "
        "VALUES (:tid, :qid, '\"new\"', 'seed', 'seed', NOW())",
        {"tid": ts_id, "qid": q_id},
    )
    db_session.commit()

    r = client.get(f"{API}/{seeded_task_survey['uid']}", headers=auth_headers)
    assert r.status_code == 200
    body = r.get_json()["data"]
    q_uid = seeded_task_survey["question_uids"][0]
    assert body[q_uid]["answer"] == "new"
```

註：此測試與 Task 1.5 add unique constraint 衝突 — 在 Task 1.5 之後改用 staging table 或 marker 來 bypass。先寫失敗版，1.5 時再調整。

- [ ] **Step 2: 跑 RED**

```bash
pytest tests/test_task_survey_answer.py::test_get_dedup_when_dirty_data_exists -v
```

預期：FAIL — 目前 service GET 不去重。

- [ ] **Step 3: 在 repo 加 `get_latest_by_task_survey_id`**

```python
# infra/task_survey/repository/question_answer_repo_impl.py
from sqlalchemy import desc

class QuestionAnswerRepoImpl(...):
    ...
    def get_latest_by_task_survey_id(self, task_survey_id: int) -> list[QuestionAnswerEntity]:
        """以 (question_id) 為 partition、updated_at desc 取每組最新一筆。
        即使 DB 還有重複髒資料也能回對的版本。"""
        from sqlalchemy import select, func, text
        # PostgreSQL DISTINCT ON
        stmt = (
            select(self.model)
            .where(self.model.task_survey_id == task_survey_id)
            .order_by(self.model.question_id, desc(self.model.updated_at), desc(self.model.id))
            .distinct(self.model.question_id)
        )
        rows = self.session.execute(stmt).scalars().all()
        return self.mapper.to_list_entity(rows)
```

- [ ] **Step 4: 在 domain repo interface 加抽象方法**

```python
# domain/task_survey/repository/question_answer.py
@abstractmethod
def get_latest_by_task_survey_id(self, task_survey_id: int) -> list[QuestionAnswerEntity]: ...
```

- [ ] **Step 5: domain service wrap**

```python
# domain/task_survey/service/question_answer_domain_service.py
def get_latest_question_answers(self, task_survey_id: int) -> list[QuestionAnswerEntity]:
    return self.question_answer_repo.get_latest_by_task_survey_id(task_survey_id)
```

- [ ] **Step 6: app service GET 改用新方法**

```python
# app/task_survey/service/question_answer_service.py:40
res = self.question_answer_domain_service.get_latest_question_answers(task_survey.id)
res = QuestionAnswerResponseDTO.from_entity_list(res)
ans = {q.question_uid: q for q in res}
return ans
```

- [ ] **Step 7: 跑全部測試確認 GREEN**

```bash
pytest tests/test_task_survey_answer.py -v
```

- [ ] **Step 8: Commit**

```bash
git add infra/ domain/ app/task_survey/service/question_answer_service.py tests/
git commit -m "fix(task-survey): use DISTINCT ON for GET to ignore duplicate rows"
```

---

## Task 1.4：寫 dedup migration

**Files:**
- Create: `scripts/sql/2026-04-28-survey-answer-dedup.sql`

- [ ] **Step 1: 寫 migration**

```sql
-- Date: 2026-04-28
-- Task Survey 答案表去重 + 加 UNIQUE constraint
-- 對應 issue: docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md
-- 對應 plan : docs/features/FR-021-2604-survey-answer-duplicate-fix/implementation-plan.md (Phase 1.4)

-- 1. 備份重複資料以利稽核（保留 7 天再 drop） (2026-04-28)
CREATE TABLE IF NOT EXISTS survey.question_answers_dedup_backup_20260428 AS
SELECT * FROM survey.question_answers WHERE 1=0;

-- 2. 把要刪除的（同 task_survey_id+question_id 內非最新）寫進備份 (2026-04-28)
WITH ranked AS (
    SELECT id,
           ROW_NUMBER() OVER (
               PARTITION BY task_survey_id, question_id
               ORDER BY updated_at DESC NULLS LAST, id DESC
           ) AS rn
    FROM survey.question_answers
)
INSERT INTO survey.question_answers_dedup_backup_20260428
SELECT qa.* FROM survey.question_answers qa
JOIN ranked r ON r.id = qa.id
WHERE r.rn > 1;

-- 3. 刪除非最新的重複 row (2026-04-28)
WITH ranked AS (
    SELECT id,
           ROW_NUMBER() OVER (
               PARTITION BY task_survey_id, question_id
               ORDER BY updated_at DESC NULLS LAST, id DESC
           ) AS rn
    FROM survey.question_answers
)
DELETE FROM survey.question_answers qa
USING ranked r
WHERE qa.id = r.id AND r.rn > 1;

-- 4. 加 UNIQUE constraint (2026-04-28)
ALTER TABLE survey.question_answers
    ADD CONSTRAINT uq_question_answers_task_survey_question
    UNIQUE (task_survey_id, question_id);

-- 5. 驗證：每組 (task_survey_id, question_id) 必須剛好 1 筆 (2026-04-28)
DO $$
DECLARE
    dup_count integer;
BEGIN
    SELECT COUNT(*) INTO dup_count FROM (
        SELECT 1 FROM survey.question_answers
        GROUP BY task_survey_id, question_id
        HAVING COUNT(*) > 1
    ) g;
    IF dup_count > 0 THEN
        RAISE EXCEPTION 'Dedup failed: % duplicate groups remain', dup_count;
    END IF;
END $$;
```

註：本 migration 改現有表，不新增 table，無需 `GRANT`（原權限延續）。

- [ ] **Step 2: 用 cmmgr 帳號跑 dev migration**

```bash
PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg \
    -f scripts/sql/2026-04-28-survey-answer-dedup.sql
```

預期：所有指令成功，最後 DO block 不拋例外。

- [ ] **Step 3: 驗證 dedup 成功**

```bash
PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg -c "
SELECT
  (SELECT COUNT(*) FROM survey.question_answers) AS total_rows,
  (SELECT COUNT(*) FROM (SELECT 1 FROM survey.question_answers GROUP BY task_survey_id, question_id HAVING COUNT(*) > 1) g) AS duplicate_pairs,
  (SELECT COUNT(*) FROM survey.question_answers_dedup_backup_20260428) AS backup_rows;
"
```

預期：duplicate_pairs = 0；total_rows = 602（distinct 數）；backup_rows = 1736（原 2338 - 602）。

- [ ] **Step 4: 把 Task 1.3 那個 test_get_dedup_when_dirty_data_exists 調整**

加 unique constraint 後，原本手插重複的方式會被擋。改成：

```python
# 改用 ON CONFLICT DO NOTHING + 暫時 disable trigger 是過度。
# 最簡單：把這個 case 改用 raw INSERT bypass constraint 是不可能的。
# 改測試方向：「即使透過 service 多次寫入，GET 也只回最新版本」，這已經被 test_put_twice_updates_existing_value 涵蓋。
# 結論：刪除 test_get_dedup_when_dirty_data_exists，改保留 test_put_twice_updates_existing_value 即可。
```

刪除 `test_get_dedup_when_dirty_data_exists`，理由寫進測試檔註解。

- [ ] **Step 5: 跑全套測試**

```bash
pytest tests/test_task_survey_answer.py -v
```

- [ ] **Step 6: Commit**

```bash
git add scripts/sql/2026-04-28-survey-answer-dedup.sql tests/test_task_survey_answer.py
git commit -m "fix(task-survey): dedup migration + UNIQUE(task_survey_id, question_id)

dev DB 跑後 2338→602 rows，1736 重複 row 進備份表 question_answers_dedup_backup_20260428。"
```

---

## Task 1.5：補 socket 路徑當前的 PUT 行為測試

**Files:**
- Modify: `tests/test_task_survey_answer.py`

socket on_update 目前還是走 `update_task_survey_answer`（Phase 2 才會改成 patch）；先補測試保證 Phase 1 邏輯改完後 socket 也不會生重複。

- [ ] **Step 1: 寫 socket 入口的 service-level 測試**

```python
def test_socket_full_data_path_does_not_duplicate(client, auth_headers, seeded_task_survey, db_session, container):
    """socket on_update 內部呼叫 update_task_survey_answer(room, None, full_data, user)，
    多次呼叫不應產生重複 row（Phase 1 service 修完後）。"""
    qa_service = container.question_answer_container.question_answer_service()
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    full_data_1 = {q_uid: {"answer": "via socket v1"}}
    qa_service.update_task_survey_answer(ts_uid, None, full_data_1, "test-user")

    full_data_2 = {q_uid: {"answer": "via socket v2"}}
    qa_service.update_task_survey_answer(ts_uid, None, full_data_2, "test-user")

    cnt = db_session.execute(
        "SELECT count(*) FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).scalar()
    assert cnt == 1
```

- [ ] **Step 2: 跑測試**

```bash
pytest tests/test_task_survey_answer.py -v
```

- [ ] **Step 3: Commit**

```bash
git add tests/test_task_survey_answer.py
git commit -m "test(task-survey): cover socket on_update path against duplicate"
```

---

## Task 1.6：寫 Phase 1 changelog

**Files:**
- Create: `docs/changelog/2026-04-28-fix-task-survey-answer-duplicate-phase1.md`

- [ ] **Step 1: 寫 changelog**

```markdown
---
type: fix
modules: [task_survey]
issue: docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md
commit: <填 phase1 最後一筆 commit hash>
---

# fix(task-survey): 答案重複寫入 bug Phase 1（natural key upsert + DB 去重）

## 為什麼

`POST /project-survey/answers/<uid>` PUT 與 socket on_update 共用的 `update_task_survey_answer` 用 row uid 判斷新舊，payload 沒帶 uid（或對不上）就 INSERT，造成同 (task_survey_id, question_id) 多筆 row。dev DB 累積 1736 筆重複（71%）。對應 Billows NICS 已修過的 pattern A，跨專案根因相同。

## 變更

- `app/task_survey/service/question_answer_service.py`
  - `update_task_survey_answer` 改用 `(task_survey_id, question_id)` natural key upsert
  - `get_answer_list_by_task_survey_uid` 改用新 repo method `get_latest_question_answers` 走 SQL DISTINCT ON 去重
- `infra/task_survey/repository/question_answer_repo_impl.py` 新增 `get_latest_by_task_survey_id`
- `domain/task_survey/repository/question_answer.py` 新增抽象方法
- `domain/task_survey/service/question_answer_domain_service.py` wrap
- `scripts/sql/2026-04-28-survey-answer-dedup.sql`
  - 把同組非最新 row 寫進備份表 `survey.question_answers_dedup_backup_20260428`
  - 刪除這些 row
  - 加 `UNIQUE (task_survey_id, question_id)` constraint
  - DO block 驗證跑完後沒有重複組

## API 變更

無外部合約變動。GET 回傳的 dict 結構不變、PUT 接收的 payload 結構不變。

## 測試結果

```
$ pytest tests/test_task_survey_answer.py -v
... 3 passed ...
```

dev DB 驗證：
- 跑 migration 前：2338 rows / 602 distinct pairs / 430 重複組
- 跑 migration 後：602 rows / 0 重複組 / backup table 1736 rows

## 參考

- 對應 issue：`docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md`
- 對應 plan：`docs/features/FR-021-2604-survey-answer-duplicate-fix/implementation-plan.md`（Phase 1）
- Billows NICS 對照：`/Users/chouraymond/Projects/Billows/NICS/billows-be/docs/fixes/20260422_survey_checkpoint_performance/`
```

- [ ] **Step 2: Commit Phase 1 完整收尾**

```bash
git add docs/changelog/2026-04-28-fix-task-survey-answer-duplicate-phase1.md
git commit -m "docs(changelog): task-survey answer duplicate fix phase 1"
```

---

# Phase 2 — 多人協作一致性：socket 單題 patch + on_join 從 DB 讀

**Outcome：**
- socket on_update 不再寫整份 full_data，改成只寫被改的那題。
- on_join snapshot 從 DB 讀，移除 Redis `fill_survey:<room>:updates`。
- 新增 REST `POST /project-survey/answers/<uid>/patch`，前端 `fetch keepalive` 可用來在關分頁前 flush dirty 欄位。
- 寫 history 的時機與舊版一致（暫存/提交時，不在每次 socket update 寫）。

## Task 2.1：service 新增 `patch_task_survey_answer`

**Files:**
- Modify: `app/task_survey/service/question_answer_service.py`

- [ ] **Step 1: 寫 failing 測試**

```python
def test_patch_single_question_upserts_one_row(client, auth_headers, seeded_task_survey, db_session, container):
    qa_service = container.question_answer_container.question_answer_service()
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    qa_service.patch_task_survey_answer(ts_uid, q_uid, {"answer": "patched v1"}, "test-user")
    qa_service.patch_task_survey_answer(ts_uid, q_uid, {"answer": "patched v2"}, "test-user")

    rows = db_session.execute(
        "SELECT answer FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).fetchall()
    assert len(rows) == 1
    assert rows[0][0] == "patched v2"
```

- [ ] **Step 2: 跑 RED**

```bash
pytest tests/test_task_survey_answer.py::test_patch_single_question_upserts_one_row -v
```

預期：`AttributeError: 'QuestionAnswerService' object has no attribute 'patch_task_survey_answer'`

- [ ] **Step 3: 實作 service method**

```python
# app/task_survey/service/question_answer_service.py
@transaction
def patch_task_survey_answer(
    self,
    task_survey_uid: str,
    question_uid: str,
    answer_patch: dict,
    user: str,
) -> QuestionAnswerEntity:
    """單題 upsert。供 socket on_update 與 REST patch endpoint 使用。

    與 update_task_survey_answer 不同：
    - 不寫 history（history 只在 PUT 暫存/提交時寫）
    - 不變更 task_survey.status（status 由暫存/提交按鈕掌控）
    - 只動一題，N+1 也不存在
    """
    task_survey = self.task_survey_domain_service.get_task_survey_by_uid(task_survey_uid)
    if not task_survey:
        raise NotFound(ErrorCode.TASK_SURVEY_NOT_FOUND)

    question = next(
        (q for q in self.survey_question_domain_service.get_questions(survey_id=task_survey.survey_id)
         if q.uid == question_uid),
        None,
    )
    if question is None:
        raise NotFound(ErrorCode.TASK_SURVEY_QUESTION_NOT_FOUND)

    existing = self.question_answer_domain_service.get_question_answer(
        QuestionAnswerQueryEntity(task_survey_id=task_survey.id, question_id=question.id)
    )

    if existing is not None:
        existing.answer = answer_patch.get("answer", existing.answer)
        existing.ext_answer = answer_patch.get("ext_answer", existing.ext_answer)
        existing.feedback = answer_patch.get("feedback", existing.feedback)
        existing.ext_feedback = answer_patch.get("ext_feedback", existing.ext_feedback)
        existing.score = answer_patch.get("score", existing.score)
        existing.is_delete = 0
        existing.updated_user = user
        return self.question_answer_domain_service.update_question_answer(existing)

    new_qa = QuestionAnswerEntity(
        score=answer_patch.get("score", 0),
        answer=answer_patch.get("answer"),
        ext_answer=answer_patch.get("ext_answer"),
        feedback=answer_patch.get("feedback", ""),
        ext_feedback=answer_patch.get("ext_feedback", ""),
        is_delete=0,
        question_id=question.id,
        survey_id=task_survey.survey_id,
        task_survey_id=task_survey.id,
        created_user=user,
        updated_user=user,
    )
    return self.question_answer_domain_service.add_question_answer(new_qa)
```

- [ ] **Step 4: 加 error code**

```python
# common/code/error_code.py
TASK_SURVEY_QUESTION_NOT_FOUND = ("問卷題目不存在", "TASK_SURVEY_404006")
```

- [ ] **Step 5: 跑測試**

```bash
pytest tests/test_task_survey_answer.py::test_patch_single_question_upserts_one_row -v
```

- [ ] **Step 6: Commit**

```bash
git add app/task_survey/service/question_answer_service.py common/code/error_code.py tests/test_task_survey_answer.py
git commit -m "feat(task-survey): add patch_task_survey_answer for single-question upsert"
```

---

## Task 2.2：新增 REST `POST /project-survey/answers/<uid>/patch`

**Files:**
- Create: `api/task_survey/serializers/question_answer_patch.py`
- Modify: `api/task_survey/routes/question_answer_route.py`
- Modify: `api/task_survey/__init__.py`

- [ ] **Step 1: 寫 failing API 測試**

```python
def test_post_answers_patch_endpoint(client, auth_headers, seeded_task_survey, db_session):
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    payload = {"patches": [{"question_uid": q_uid, "answer": {"answer": "rest-patch"}}]}
    r = client.post(f"{API}/{ts_uid}/patch", json=payload, headers=auth_headers)
    assert r.status_code == 200

    cnt = db_session.execute(
        "SELECT count(*) FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).scalar()
    assert cnt == 1
```

- [ ] **Step 2: 跑 RED**

預期：404 或 405。

- [ ] **Step 3: 寫 serializer**

```python
# api/task_survey/serializers/question_answer_patch.py
from marshmallow import Schema, fields

class _PatchAnswerInner(Schema):
    answer = fields.Raw(required=False, allow_none=True)
    ext_answer = fields.Raw(required=False, allow_none=True)
    feedback = fields.Raw(required=False, allow_none=True)
    ext_feedback = fields.Raw(required=False, allow_none=True)
    score = fields.Integer(required=False)


class PatchItemSchema(Schema):
    question_uid = fields.String(required=True)
    answer = fields.Nested(_PatchAnswerInner, required=True)


class PatchRequestSchema(Schema):
    patches = fields.List(fields.Nested(PatchItemSchema), required=True)
```

- [ ] **Step 4: 寫 route**

```python
# api/task_survey/routes/question_answer_route.py 新增
class TaskSurveyAnswerPatchRoute(MethodResource):
    @doc(description='批次單題 patch（供 socket / keepalive flush 用）',
         tags=['Main Project Survey Answer'], params=AUTH_PARAMS)
    @jwt_required()
    def post(self, uid,
             question_answer_service: QuestionAnswerService = Provide[
                 Containers.question_answer_container.question_answer_service]):
        user = get_user_context().login_name
        payload = request.get_json(silent=True) or {}
        patches = payload.get("patches", [])
        for patch in patches:
            question_answer_service.patch_task_survey_answer(
                uid, patch["question_uid"], patch.get("answer", {}), user
            )
        return return_response(True, {"applied": len(patches)})
```

- [ ] **Step 5: 註冊 blueprint**

```python
# api/task_survey/__init__.py 在 TaskSurveysAnswersRoute 之後
from api.task_survey.routes.question_answer_route import TaskSurveyAnswerPatchRoute
api.add_resource(TaskSurveyAnswerPatchRoute, '/project-survey/answers/<string:uid>/patch')
```

- [ ] **Step 6: 跑測試**

```bash
pytest tests/test_task_survey_answer.py::test_post_answers_patch_endpoint -v
```

- [ ] **Step 7: Commit**

```bash
git add api/task_survey/
git commit -m "feat(task-survey): add POST /project-survey/answers/<uid>/patch"
```

---

## Task 2.3：socket on_update 改走單題 patch、移除 Redis updates

**Files:**
- Modify: `app/task_survey/handler/fill_survey_socketio_handler.py`

- [ ] **Step 1: 改寫 `on_update`**

```python
def on_update(self, data):
    """問卷填答案更新（單題 patch）。

    舊版接收 full_data 整份覆蓋 + Redis snapshot，會造成 stale data 與重複寫入。
    新版只動被改的那題，避免多人協作覆蓋彼此編輯。
    """
    user = data.get('user')
    room = data.get('room')
    receive_data = data.get('data') or {}
    question_uid = receive_data.get('question_uid')
    answer_patch = receive_data.get('answer') or {}

    if not question_uid:
        emit('error', {'msg': 'question_uid is required'})
        return

    # 寫 DB（單題 upsert）
    self.question_answer_service.patch_task_survey_answer(room, question_uid, answer_patch, user)

    # broadcast（payload 維持原 shape，前端不需動）
    result = {
        'msg': f' {user} 更新資料',
        'user': user,
        'room': room,
        'uid': receive_data.get('uid'),
        'data': receive_data,
    }
    emit('updated', result, room=room)
```

- [ ] **Step 2: 改寫 `on_join` 從 DB 讀 snapshot**

```python
def on_join(self, data):
    user = data.get('user')
    room = data.get('room')
    data["event"] = "join"

    redis_cli = RedisClient()
    redis_cli.ping()
    key = f'fill-survey:{room}:users'
    users = redis_cli.get_all_from_list(key)
    if user not in users:
        redis_cli.lpush(key, user)
    users = redis_cli.get_all_from_list(key)

    result = {
        'msg': f'{user} 加入編輯',
        'users': users,
        'user': user,
        'data': None,
    }

    # 多人時，從 DB 拉最新 answer dict（取代舊 Redis snapshot）
    if len(users) > 1:
        result['data'] = self.question_answer_service.get_answer_list_by_task_survey_uid(room)

    join_room(room)
    emit('joined', result, room=room)
```

- [ ] **Step 3: `on_leave` 移除 Redis updates 清空**

```python
def on_leave(self, data):
    user = data.get('user')
    room = data.get('room')
    data["event"] = "leave"

    redis_cli = RedisClient()
    key = f'fill-survey:{room}:users'
    users = redis_cli.get_all_from_list(key)
    if user in users:
        redis_cli.lrem(key, user)
    users = redis_cli.get_all_from_list(key)

    # 不再需要清 fill_survey:<room>:updates（已不寫入）

    leave_room(room)
    emit('left', {'msg': f'{user} 離開編輯', 'users': users, 'user': user}, room=room)
```

- [ ] **Step 4: 跑全套**

```bash
pytest tests/test_task_survey_answer.py -v
```

- [ ] **Step 5: Manual smoke（兩個瀏覽器分頁）**

跑前端 dev server，兩個分頁開同一張問卷：
1. A 改第 1 題 → B 看到更新；DB 該題只有 1 row。
2. B 改第 2 題 → A 看到更新；A 第 1 題的編輯不被 B 覆蓋。
3. C 後加入 → 看到 A、B 都改完的最新版本。

對照 Billows `04_conversation.md` § 7 的驗證清單。

- [ ] **Step 6: Commit**

```bash
git add app/task_survey/handler/fill_survey_socketio_handler.py
git commit -m "fix(task-survey): socket on_update uses single-question patch, on_join reads from DB

移除 fill_survey:<room>:updates Redis stale snapshot 模式。
對應 Billows R1 / R2 修法。"
```

---

## Task 2.4：清掉 Redis `fill_survey:<room>:updates` 殘留 key（可選 cleanup）

**Files:**
- Modify: `app/task_survey/handler/fill_survey_socketio_handler.py`（移除任何剩下的 rpush/get_last_one_from_list 呼叫）

- [ ] **Step 1: grep 確認沒有殘留**

```bash
grep -rn "fill_survey:.*:updates" app/ infra/ domain/
```

預期：0 行。

- [ ] **Step 2: 在前端發版前先在 dev Redis 清舊 key**

```bash
redis-cli -h 192.168.50.189 KEYS "fill_survey:*:updates" | xargs -I{} redis-cli -h 192.168.50.189 DEL {}
```

註：dev only；staging/prod 等發版時跑同樣指令。

---

## Task 2.5：寫 Phase 2 changelog

**Files:**
- Create: `docs/changelog/2026-04-28-fix-task-survey-answer-collab-phase2.md`

- [ ] **Step 1: 寫 changelog**

內容：socket on_update 改單題 patch、on_join 從 DB、新增 patch endpoint、移除 Redis updates；對應 Billows R1/R2。

- [ ] **Step 2: Commit**

---

# Phase 3 — 效能：消除 N+1 SELECT

**Outcome：**
- `update_task_survey_answer` 100 題 PUT 從目前的 ~200+ SELECT 降到 ~5 SELECT。
- `revert_question_answer_from_history` 100 題 revert 同樣處理。

## Task 3.1：Phase 1 後 update 路徑檢視

Phase 1 的 service 改寫已經建了 `existing_by_qid` map，主要 N+1 已消。但 `question_answer_domain_service.update_question_answer` 內 `verify_question_answer_exist_by_uid` 每筆還跑一次 SELECT，不必要。

- [ ] **Step 1: 把 update path 改成 repo 直接 update（已經拿到 entity，不需要再 verify）**

`question_answer_service.update_task_survey_answer` 內 update branch 直接呼 `self.question_answer_domain_service.question_answer_repo.update(existing)`？— 違反 DDD（不能跨 domain service 直接戳 repo）。

正確做法：在 domain service 加 `update_question_answer_without_verify(entity)`，或修改 `update_question_answer` 改用「entity 已驗證過」的版本。

選比較簡單：

```python
# domain/task_survey/service/question_answer_domain_service.py
def update_question_answer_in_memory(self, entity: QuestionAnswerEntity) -> QuestionAnswerEntity:
    """已 in-memory 確認 entity 來自 DB，不再跑 verify_*。"""
    return self.question_answer_repo.update(entity)
```

service 內 update branch 改用此 method。

- [ ] **Step 2: 加效能測試（100 題 PUT 應 < 3 秒）**

```python
def test_put_100_questions_under_3s(client, auth_headers, seeded_100q_task_survey):
    import time
    ts_uid = seeded_100q_task_survey["uid"]
    answers = {q_uid: {"answer": f"v_{i}"} for i, q_uid in enumerate(seeded_100q_task_survey["question_uids"])}

    t0 = time.time()
    r = client.put(f"{API}/{ts_uid}", json={"status": 1, "answers": answers}, headers=auth_headers)
    elapsed = time.time() - t0

    assert r.status_code == 200
    assert elapsed < 3.0, f"100 題 PUT 跑 {elapsed:.2f}s，超過 3s 上限"
```

- [ ] **Step 3: 跑測試 / commit**

---

## Task 3.2：revert 路徑同樣處理

**Files:**
- Modify: `app/task_survey/service/question_answer_history_service.py`

- [ ] **Step 1: 預取 current_answers**

```python
@transaction
def revert_question_answer_from_history(self, task_survey_uid, history_uid, user) -> bool:
    task_survey = self.task_survey_domain_service.get_task_survey_by_uid(task_survey_uid)
    answer_history = self.question_answer_history_domain_service.get_question_answer_history(history_uid)
    history_answers = self.question_answer_history_detail_domain_service.get_history_details(
        QuestionAnswerHistoryDetailQueryEntity(history_id=answer_history.id)
    )

    # 預取一次當前 answers，建 map
    current_answers = self.question_answer_domain_service.get_question_answers(
        QuestionAnswerQueryEntity(task_survey_id=task_survey.id)
    )
    current_by_qid = {a.question_id: a for a in current_answers}

    history_detail_entity_list = []
    for answer in history_answers:
        question_answer = current_by_qid.get(answer.question_id)
        if question_answer is None:
            continue  # 該題已被刪 — skip

        question_answer.score = answer.score
        question_answer.answer = answer.answer
        question_answer.ext_answer = answer.ext_answer
        question_answer.feedback = answer.feedback
        question_answer.ext_feedback = answer.ext_feedback
        question_answer.updated_user = user
        self.question_answer_domain_service.update_question_answer_in_memory(question_answer)

        history_detail_entity_list.append(
            QuestionAnswerHistoryDetailEntity(
                task_survey_id=task_survey.id,
                question_id=answer.question_id,
                score=answer.score,
                answer=answer.answer,
                ext_answer=answer.ext_answer,
                feedback=answer.feedback,
                ext_feedback=answer.ext_feedback,
                updated_user=user,
                created_user=user,
            )
        )

    history_entity = QuestionAnswerHistoryEntity(
        task_survey_id=task_survey.id,
        type="R",
        updated_user=user,
        created_user=user,
    )
    self.question_answer_history_domain_service.create_history_with_details(
        history_entity, history_detail_entity_list
    )
    return True
```

- [ ] **Step 2: 加 revert 效能測試**

```python
def test_revert_100_questions_under_3s(...):  # 同上 pattern
```

- [ ] **Step 3: 跑 / commit**

---

## Task 3.3：Phase 3 changelog

**Files:**
- Create: `docs/changelog/2026-04-28-tweak-task-survey-answer-perf-phase3.md`

---

# Phase 4 — 收尾

## Task 4.1：跑全套 task_survey 測試

- [ ] **Step 1: 跑**

```bash
pytest tests/test_task_survey_answer.py -v
pytest tests/  # 確認沒打到別處
```

預期全綠。

- [ ] **Step 2: 跑 dev DB 一次完整 e2e check**

```bash
PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg -c "
SELECT COUNT(*) FROM (
    SELECT 1 FROM survey.question_answers
    GROUP BY task_survey_id, question_id HAVING COUNT(*) > 1
) g;
"
```

預期：0。

## Task 4.2：把 issue 從 pending 移到 resolved

- [ ] **Step 1: 在原 issue 檔最前面加 `## ⓪ Resolution` 段**

```markdown
## ⓪ Resolution

| 項目 | 內容 |
|------|------|
| 修復日期 | 2026-04-28 |
| Commits | <Phase1>, <Phase2>, <Phase3> short hashes |
| Changelogs | docs/changelog/2026-04-28-fix-task-survey-answer-duplicate-phase1.md / phase2 / phase3 |
| Follow-up | 前端是否帶 row uid 不是 root cause（後端已防禦），但建議讓 fe 補上 dirty tracking + keepalive flush 對應 Billows fe 修法 |
```

- [ ] **Step 2: `git mv`**

```bash
git mv docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md \
       docs/issues/resolved/2026-04-28-task-survey-answer-duplicate.md
```

- [ ] **Step 3: 把每個 phase 的 changelog frontmatter `issue:` 改指 resolved 路徑**

- [ ] **Step 4: Commit**

```bash
git commit -m "docs(issue): mark task-survey answer duplicate fix resolved"
```

## Task 4.3：staging / prod 上線清單給使用者

寫在最終回覆裡（不進 repo）：

1. Staging / prod 都需要跑 `scripts/sql/2026-04-28-survey-answer-dedup.sql`，跑前先在 staging 確認 backup table 行為。
2. 上線後跑 Redis cleanup：`redis-cli KEYS "fill_survey:*:updates" | xargs -I{} redis-cli DEL {}`
3. 通知前端團隊新增的 `POST /project-survey/answers/<uid>/patch` 可在 unmount/beforeunload/pagehide 用 `fetch({keepalive: true})` flush dirty 欄位（對應 Billows fe 03_changelog § 5）。

---

## 執行順序建議

| Phase | 風險 | 建議獨立 PR / 連 PR |
|-------|------|---------------------|
| 1 | 低 — 純內部邏輯 + DB constraint | 獨立 PR、上 staging 驗 |
| 2 | 中 — 改 socket 行為，多人協作行為變動 | 獨立 PR、上 staging 兩瀏覽器驗 |
| 3 | 低 — 純效能，沒行為變動 | 可併 Phase 2 PR |
| 4 | 低 — 文件 | 併最後一個 PR |

---

## 風險評估

| 面向 | 風險 | 緩解 |
|------|------|------|
| Phase 1 unique constraint | 加 constraint 失敗（dedup 沒清乾淨） | migration 內 DO block 自動驗證、跑前 backup 表保留 7 天 |
| Phase 2 socket 行為變動 | 前端依賴 `full_data` 接收格式 | broadcast 仍 emit `updated`，payload shape 不變；多人測試覆蓋 |
| Phase 2 移除 Redis updates | 既有 prod 環境 Redis 還有舊 key | 上線後跑 cleanup 指令；不影響功能（沒有 reader） |
| Phase 3 update_in_memory | 跳過 verify 可能寫入已被 hard-delete 的 row | 接 entity 來自當下 transaction 內預取，無此 race |
| 前端未配合 | 前端仍送 PUT full payload | 後端 PUT 路徑（natural key upsert）已防禦；前端 patch endpoint 是優化非必要 |