@transaction
def checkpoint_task_survey_answer(
    self,
    task_survey_uid: str,
    status: int,
    patches: list,        # [{question_uid: str, answer: dict}, ...]
    user: str,
) -> TaskSurveyDTO:
    """暫存/提交 checkpoint，單一 transaction 完成所有寫入。

    流程：
    1. 取 task_survey、survey questions、existing answers map（自然鍵）
    2. 套 patches（in-memory mutation 已存在的 entity，或 ADD 新的）
    3. 補空 row（survey 有題目但 DB 沒 row）— 保留舊 PUT 路徑的「全題目都有 row」語意，便於 revert
    4. 更新 task_survey.status（不可從 status=2/9 回退到較早狀態）
    5. 寫 history + history_details（從 in-memory state，不重 SELECT）
    6. 通知 status change（如有變動）

    與 update_task_survey_answer 差別：
    - 接收 patches list（單題 dict）而非 answers dict（整份覆蓋語意）
    - 走 update_question_answer_in_memory 跳 verify SELECT
    - history details 從 in-memory state，省一輪 SELECT
    """
    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)

    questions = self.survey_question_domain_service.get_questions(survey_id=task_survey.survey_id)
    q_uid_to_obj = {q.uid: q for q in questions}

    exist_answers = self.question_answer_domain_service.get_question_answers(
        QuestionAnswerQueryEntity(task_survey_id=task_survey.id)
    )
    # 重複時保留 updated_at 最新（migration 後不該再有，但服務層仍防禦）
    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 = []

    # 1. 套 patches
    for patch in (patches or []):
        question_uid = patch.get("question_uid")
        answer_patch = patch.get("answer") or {}
        if not question_uid:
            continue
        question = q_uid_to_obj.get(question_uid)
        if question is None:
            continue  # survey 沒這題，跳過

        existing = existing_by_qid.get(question.id)
        if existing is not None:
            if "answer" in answer_patch:
                existing.answer = answer_patch["answer"]
            if "ext_answer" in answer_patch:
                existing.ext_answer = answer_patch["ext_answer"]
            if "feedback" in answer_patch:
                existing.feedback = answer_patch["feedback"]
            if "ext_feedback" in answer_patch:
                existing.ext_feedback = answer_patch["ext_feedback"]
            if "score" in answer_patch:
                existing.score = answer_patch["score"]
            existing.updated_user = user
            self.question_answer_domain_service.update_question_answer_in_memory(existing)
            res_answer_list.append(existing)
        else:
            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", ""),
                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
            res_answer_list.append(new_qa)

    # 2. 補空 row（survey 有題目但 DB 沒 row 且這次沒被 patch）
    new_empty_qas = []
    for question in questions:
        if question.id in existing_by_qid:
            continue
        empty_qa = QuestionAnswerEntity(
            score=0,
            answer=None,
            ext_answer=None,
            feedback="",
            ext_feedback="",
            question_id=question.id,
            survey_id=task_survey.survey_id,
            task_survey_id=task_survey.id,
            created_user=user,
            updated_user=user,
        )
        empty_qa = self.question_answer_domain_service.add_question_answer(empty_qa)
        existing_by_qid[question.id] = empty_qa
        new_empty_qas.append(empty_qa)
        res_answer_list.append(empty_qa)

    # 3. 更新 status（含通知）
    survey_status_change = False
    if status is not None:
        survey_status_change = (status != task_survey.status)
        task_survey.status = status

    task_survey_entity = self.task_survey_domain_service.update_task_survey(task_survey)

    if survey_status_change:
        self.task_survey_service.notify_task_survey_assignee_status_change(task_survey.uid)

    # 4. 寫 history
    history = QuestionAnswerHistoryEntity(
        task_survey_id=task_survey.id,
        updated_user=user,
        created_user=user,
    )
    history_details = []
    for answer in res_answer_list:
        history_details.append(
            QuestionAnswerHistoryDetailEntity(
                task_survey_id=task_survey.id,
                survey_id=task_survey.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,
            )
        )
    self.question_answer_history_domain_service.create_history_with_details(history, history_details)

    return TaskSurveyDTO.from_entity(task_survey_entity)

"""Request schema for POST /project-survey/answers/<uid>/checkpoint."""
from marshmallow import Schema, fields


class _PatchAnswerInner(Schema):
    """Inner answer body — all fields optional (PATCH semantics)."""
    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 CheckpointRequestSchema(Schema):
    status = fields.Integer(required=True)
    patches = fields.List(fields.Nested(PatchItemSchema), required=True)

class TaskSurveyAnswerCheckpointRoute(MethodResource):
    @doc(description='暫存/提交問卷答案（patches-based）',
         tags=['Main Project Survey Answer'], params=AUTH_PARAMS)
    @use_kwargs(CheckpointRequestSchema, location='json', apply=True)
    @jwt_required()
    def post(self, uid, status, patches,
             question_answer_service: QuestionAnswerService = Provide[
                 Containers.question_answer_container.question_answer_service]):
        user = get_user_context().login_name
        data = question_answer_service.checkpoint_task_survey_answer(
            uid, status, patches, user
        )
        return return_response(True, data)

# api/task_survey/__init__.py 內 create_module():
from api.task_survey.routes.question_answer_route import (
    TaskSurveysAnswersRoute,
    TaskSurveyAnswerPatchRoute,
    TaskSurveyAnswerCheckpointRoute,
)
...
api.add_resource(
    TaskSurveyAnswerCheckpointRoute,
    '/project-survey/answers/<string:uid>/checkpoint'
)

result = {
    'msg': f' {user} 更新資料',
    'user': user,
    'room': room,
    'uid': receive_data.get('uid'),
    'data': receive_data,  # 整個 receive_data 含 question_uid + answer
}
emit('updated', result, room=room)

result = {
    'msg': f' {user} 更新資料',
    'user': user,
    'room': room,
    'uid': question_uid,             # ← top-level uid 是 question_uid（語意明確）
    'question_uid': question_uid,    # ← 同上，多一個明確命名讓 FE 不用 fallback
    'data': answer_patch,            # ← 純 answer body，跟 Billows 一致
}
emit('updated', result, room=room)

def test_checkpoint_applies_patches_only(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 套 dirty patches，只動 patches 提到的題目；其他題的舊值不變動。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]
    q_uid_2 = seeded_task_survey["question_uids"][1]

    # 先 round 1：兩題都填 v1
    with app.app_context():
        question_answer_service.update_task_survey_answer(
            ts_uid, 1,
            {q_uid_1: {"answer": "v1-1"}, q_uid_2: {"answer": "v1-2"}},
            "test-user",
        )

    # round 2：checkpoint 只送 q1 的 patch，q2 不送
    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 1,
            [{"question_uid": q_uid_1, "answer": {"answer": "v2-1"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        rows = dict(conn.execute(
            text("""SELECT sq.uid, qa.answer FROM survey.question_answers qa
                    JOIN survey.survey_questions sq ON sq.id = qa.question_id
                    WHERE qa.task_survey_id = :tid"""),
            {"tid": ts_id},
        ).fetchall())
    assert rows[q_uid_1] == "v2-1"  # q1 改了
    assert rows[q_uid_2] == "v1-2"  # q2 沒被覆蓋（重點）


def test_checkpoint_fills_empty_rows_for_missing_questions(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 對 survey 有題目但 DB 沒 row 的題目，補一筆空 row。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    # checkpoint 只 patch q1，q2 從未被填過
    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 1,
            [{"question_uid": q_uid_1, "answer": {"answer": "filled"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        cnt = conn.execute(
            text("SELECT count(*) FROM survey.question_answers WHERE task_survey_id = :tid"),
            {"tid": ts_id},
        ).scalar()
    # seeded_task_survey 種了 2 題，checkpoint 應該補滿到 2 row（q1 + q2 空 row）
    assert cnt == 2


def test_checkpoint_writes_history_once(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 一次呼叫只寫一筆 history（main）。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 1,
            [{"question_uid": q_uid_1, "answer": {"answer": "v1"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        h_cnt = conn.execute(
            text("SELECT count(*) FROM survey.question_answer_histories WHERE task_survey_id = :tid"),
            {"tid": ts_id},
        ).scalar()
    assert h_cnt == 1


def test_checkpoint_updates_status(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 帶 status 應更新 task_survey.status。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 2,  # 提交
            [{"question_uid": q_uid_1, "answer": {"answer": "submitted"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        status = conn.execute(
            text("SELECT status FROM survey.task_surveys WHERE id = :tid"),
            {"tid": ts_id},
        ).scalar()
    assert status == 2


def test_checkpoint_100_questions_select_count_within_budget(
    question_answer_service, seeded_task_survey_100q, app
):
    """100 題 checkpoint 的 SELECT 計數合理（防 N+1 regression）。"""
    from sqlalchemy import event
    ts_uid = seeded_task_survey_100q["uid"]

    # 先填 100 題
    answers_round1 = {
        q_uid: {"answer": f"v_{i}"}
        for i, q_uid in enumerate(seeded_task_survey_100q["question_uids"])
    }
    with app.app_context():
        question_answer_service.update_task_survey_answer(ts_uid, 1, answers_round1, "test-user")

    counts = {"select": 0}
    def _on_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        s = statement.lower()
        if s.startswith("select") and "survey.question_answers" in s:
            counts["select"] += 1

    engine = app.db_engine
    patches = [
        {"question_uid": q_uid, "answer": {"answer": f"v2_{i}"}}
        for i, q_uid in enumerate(seeded_task_survey_100q["question_uids"])
    ]

    counts["select"] = 0
    event.listen(engine, "before_cursor_execute", _on_cursor_execute)
    try:
        with app.app_context():
            question_answer_service.checkpoint_task_survey_answer(
                ts_uid, 1, patches, "test-user"
            )
    finally:
        event.remove(engine, "before_cursor_execute", _on_cursor_execute)

    # 100 題 checkpoint：1 prefetch + 100 in_memory update（各 1 SELECT 在 repo.update 內 SELECT-by-uid）
    # 預期 ≤ 200 SELECT；舊 update_task_survey_answer 是 ≤ 201 範圍
    print(f"[checkpoint select-count] 100 questions: {counts['select']} SELECTs")
    assert counts["select"] <= 220, (
        f"checkpoint 100 questions used {counts['select']} SELECTs against survey.question_answers, "
        "expected ≤ 220 (regression check)"
    )

---
type: tweak
breaking: true
modules: [task_survey]
issue: docs/issues/resolved/2026-04-28-task-survey-answer-duplicate.md
commit: <最後一筆>
---

# tweak(task-survey): 暫存/提交改 patches-based checkpoint endpoint + socket broadcast shape 對齊

## 為什麼

舊 PUT `/project-survey/answers/<uid>` 暫存/提交每次送整份 `answers` dict，
100 題問卷每次儲存都重寫 100 row + 1 history snapshot。多人協作下 DB 寫入塞車。

對應 Billows 已 production 驗證的設計：暫存/提交只送 dirty patches，BE 套 patches +
補空 row（保留全題目都有 row 的 revert 語意）+ 寫一筆 history。

順帶把 socket on_update broadcast 的 shape 跟 Billows 對齊：
舊版 `'data': receive_data`（整 wrapping dict），FE 要 `inner.answer` 再剝一層。
新版 `'data': answer_patch`（純 answer body）+ top-level `'question_uid'`，FE 直接拿。

## 變更

[完整 list 從 plan 的 task 1-5 對應的檔案 / commits]

## API 變更

### 新增
- `POST /api/1.0/project-survey/answers/<uid>/checkpoint`
  body `{status: int, patches: [{question_uid, answer: {answer/ext_answer/feedback/ext_feedback/score}}]}`
  response `{status: true, data: <TaskSurveyDTO>}`

### 廢棄但保留（Excel 批次匯入仍用）
- `PUT /api/1.0/project-survey/answers/<uid>`
  body `{status, answers: {q_uid: {...}}}`

### Socket（breaking — FE 同步改）
- `on 'updated'` broadcast：
  - 舊：`{user, room, uid: q_uid, data: <full receive_data dict>}`
  - 新：`{user, room, uid: q_uid, question_uid: q_uid, data: <answer body>}`

## 測試結果

[16 個 tests pass，5 連跑無 flake，100 題 checkpoint 在 SELECT budget 內]

## 上線清單

1. BE 部署本 commit 之後
2. FE 必須跟著上 Phase 5 commits（前後端綁版本，否則 socket broadcast shape 對不上）
3. 不需要 DB migration

## 提交清單

[列 4.1 ~ 4.5 commits]

// 與 PROJECT_SURVEY_ANSWERS_PATCH 共用 base，後綴自己拼
PROJECT_SURVEY_ANSWERS_CHECKPOINT: getUrl('/project-survey/answers'),
// 用法：`${PROJECT_SURVEY_ANSWERS_CHECKPOINT}/${task_survey_uid}/checkpoint`

function setAnswer(questionId, value) {
    if (!answers.value[questionId]) {
        answers.value[questionId] = { answer: value, ext_answer: '', feedback: '', ext_feedback: '', score: null }
    } else {
        answers.value[questionId].answer = value   // ← mutation 在 socket 覆蓋成 plain object 後不 trigger reactive
    }
}

function setAnswer(questionId, value) {
    const current = answers.value[questionId]
    answers.value[questionId] = current && typeof current === 'object'
        ? { ...current, answer: value }
        : { answer: value, ext_answer: '', feedback: '', ext_feedback: '', score: null }
}

function setExtAnswer(questionId, optId, value) {
    const current = answers.value[questionId] || { answer: null, ext_answer: {}, feedback: '', ext_feedback: '', score: null }
    const newExt = { ...(current.ext_answer || {}), [optId]: value }
    answers.value[questionId] = { ...current, ext_answer: newExt }
}

async function onTempSave() {
    await baseService.put(`${API.PROJECT_SURVEY_ANSWERS}/${taskSurveyUid.value}`, {
        status: 1,
        answers: answers.value,
    })
}
async function onSave(status) {
    ... validate ...
    await baseService.put(`${API.PROJECT_SURVEY_ANSWERS}/${taskSurveyUid.value}`, {
        status,
        answers: payload,
    })
    await loadTaskSurvey(taskSurveyUid.value)
}

async function onTempSave() {
    if (!taskSurveyUid.value) return
    submitted.value = true
    try {
        const patches = buildPatchesFromDirty()
        await baseService.post(
            `${API.PROJECT_SURVEY_ANSWERS_CHECKPOINT}/${taskSurveyUid.value}/checkpoint`,
            { status: 1, patches },
        )
        dirtyAnswers.value.clear()
        toast.add({ severity: 'success', summary: t('lang.survey_v2.preview.save_success'), life: 3000 })
    } catch (e) {
        toast.add({ severity: 'error', summary: t('lang.common.error'), detail: e?.message || '', life: 5000 })
    } finally {
        submitted.value = false
    }
}

async function onSave(status) {
    if (!taskSurveyUid.value) {
        toast.add({ severity: 'success', summary: t('lang.survey_v2.preview.submit_success') + '（預覽模式）', life: 3000 })
        return
    }
    // validate required fields for status 2 / 9 (same logic as before)
    if (status === 2 || status === 9) {
        const missing = allQuestions.value.filter(q => q.isRequired && !isAnswered(q.id))
        if (missing.length > 0) {
            const names = missing.map(q => q.questionNumber || q.question).join('、')
            toast.add({ severity: 'error', summary: t('lang.survey_v2.preview.required_fields_missing'), detail: names, life: 8000 })
            return
        }
    }
    submitted.value = true
    try {
        const patches = buildPatchesFromDirty()
        // score 整數化（沿用既有處理）
        for (const p of patches) {
            if (p.answer && p.answer.score != null) {
                p.answer.score = parseInt(p.answer.score, 10)
            }
        }
        await baseService.post(
            `${API.PROJECT_SURVEY_ANSWERS_CHECKPOINT}/${taskSurveyUid.value}/checkpoint`,
            { status, patches },
        )
        dirtyAnswers.value.clear()
        const msgMap = {
            2: t('lang.survey_v2.preview.submit_success'),
            3: t('lang.survey_v2.preview.review_success'),
            9: t('lang.survey_v2.preview.close_success'),
        }
        toast.add({ severity: 'success', summary: msgMap[status] || t('lang.survey_v2.preview.submit_success'), life: 3000 })
        await loadTaskSurvey(taskSurveyUid.value)
    } catch (e) {
        toast.add({ severity: 'error', summary: t('lang.common.error'), detail: e?.message || '', life: 5000 })
    } finally {
        submitted.value = false
    }
}

export function useSurveySocket({
    taskSurveyUid,
    userLoginName,
    answers,
    onAnswersReload,
    onUserJoined,
    onUserLeft,
    isDirty = () => false,   // ← 新增 optional callback
}) {
    ...
}

socket.on('updated', (data) => {
    if (data.user === userLoginName.value) return
    // BE Phase 4: 'data' 直接是 answer body；'question_uid' top-level 提供
    const questionUid = data.question_uid || data.uid
    const answerBody = data.data
    if (!questionUid || answerBody === undefined) return
    // 使用者本地對該題還在改 → 不覆蓋（避免回灌洗掉本地未送的內容）
    if (isDirty(questionUid)) return
    answers.value[questionUid] = answerBody
})

socket.on('joined', (data) => {
    onlineUsers.value = data.users || []
    if (data.user === userLoginName.value) {
        if (data.data) {
            const merged = { ...answers.value }
            Object.keys(data.data).forEach((qid) => {
                if (!isDirty(qid)) {
                    merged[qid] = data.data[qid]
                }
            })
            answers.value = merged
        }
    } else {
        onUserJoined?.(data)
    }
})

const { ... } = useSurveySocket({
    taskSurveyUid,
    userLoginName: computed(() => userInfo.value?.login_name),
    answers,
    onAnswersReload: () => loadTaskAnswers(taskSurveyUid.value),
    onUserJoined: ...,
    onUserLeft: ...,
    isDirty: (qid) => dirtyAnswers.value.has(qid),
})