fix(worker-platform): 资料更新不再覆盖并发抢单状态

apps/backend/src/repositories/worker-platform/work-order-management-repo.ts

@@ -66,69 +66,109 @@ export async function createWorkOrder(input: CreateWorkOrderInput): Promise<Work
   return getWorkOrderById(result.rows[0]?.id || 0)
 }
 
+export type WorkOrderUpdatePatch = Partial<
+  Pick<
+    WorkOrderRow,
+    | 'status'
+    | 'assigned_worker_id'
+    | 'material_json'
+    | 'acceptance_json'
+    | 'draft_acceptance_json'
+    | 'problem_note'
+    | 'reward_amount'
+    | 'sharing_enabled'
+    | 'sharing_total_quantity'
+    | 'sharing_unit_reward'
+    | 'published_at'
+    | 'hall_queued_at'
+    | 'pinned_at'
+    | 'assigned_at'
+    | 'submitted_at'
+    | 'accepted_at'
+    | 'updated_at'
+  >
+>
+
+/**
+ * 只更新 patch 中显式传入的列。
+ *
+ * 旧实现先读整行、再把整行写回，并发下会用旧快照覆盖别人刚写入的状态：
+ * 2026-10-31 生产工单 195657 抢单写入 in_progress 后，被资料更新的旧快照写回 open。
+ * 另外可传 expectedStatus 做乐观锁，状态已变化时不落库，由调用方重试或报冲突。
+ */
+export function buildWorkOrderUpdateStatement(
+  workOrderId: number | string,
+  patch: WorkOrderUpdatePatch,
+  options: { expectedStatus?: string } = {},
+): { text: string; values: unknown[] } | null {
+  const assignments: string[] = []
+  const values: unknown[] = []
+  const assign = (column: string, value: unknown, cast = '') => {
+    values.push(value)
+    assignments.push(`${column} = $${values.length}${cast}`)
+  }
+
+  if (patch.status !== undefined) assign('status', patch.status)
+  if (patch.assigned_worker_id !== undefined) {
+    assign('assigned_worker_id', patch.assigned_worker_id || null)
+  }
+  if (patch.material_json !== undefined) {
+    assign('material_json', toJsonString(patch.material_json), '::jsonb')
+    // 派生列与 material_json 同源，只在本次真的改了 material_json 时重算。
+    assign('source_order_amount_fen', resolveSourceOrderAmountFen(patch.material_json))
+  }
+  if (patch.acceptance_json !== undefined) {
+    assign('acceptance_json', toJsonString(patch.acceptance_json), '::jsonb')
+  }
+  if (patch.draft_acceptance_json !== undefined) {
+    assign('draft_acceptance_json', toJsonString(patch.draft_acceptance_json), '::jsonb')
+  }
+  if (patch.problem_note !== undefined) assign('problem_note', patch.problem_note || '')
+  if (patch.reward_amount !== undefined) {
+    assign('reward_amount', toPositiveInteger(patch.reward_amount, 0))
+  }
+  if (patch.sharing_enabled !== undefined) {
+    assign('sharing_enabled', patch.sharing_enabled === true)
+  }
+  if (patch.sharing_total_quantity !== undefined) {
+    assign('sharing_total_quantity', toPositiveInteger(patch.sharing_total_quantity, 1))
+  }
+  if (patch.sharing_unit_reward !== undefined) {
+    assign('sharing_unit_reward', toPositiveInteger(patch.sharing_unit_reward, 0))
+  }
+  if (patch.published_at !== undefined) assign('published_at', patch.published_at || null)
+  if (patch.hall_queued_at !== undefined) assign('hall_queued_at', patch.hall_queued_at || null)
+  if (patch.pinned_at !== undefined) assign('pinned_at', patch.pinned_at || null)
+  if (patch.assigned_at !== undefined) assign('assigned_at', patch.assigned_at || null)
+  if (patch.submitted_at !== undefined) assign('submitted_at', patch.submitted_at || null)
+  if (patch.accepted_at !== undefined) assign('accepted_at', patch.accepted_at || null)
+  if (patch.updated_at !== undefined) assign('updated_at', patch.updated_at)
+
+  if (assignments.length === 0) return null
+
+  values.push(Number(workOrderId))
+  let where = `WHERE id = $${values.length}`
+  if (options.expectedStatus !== undefined) {
+    values.push(options.expectedStatus)
+    where += ` AND status = $${values.length}`
+  }
+
+  return {
+    text: `UPDATE work_orders SET ${assignments.join(', ')} ${where} RETURNING id`,
+    values,
+  }
+}
+
 export async function updateWorkOrder(
   workOrderId: number | string,
-  patch: Partial<
-    Pick<
-      WorkOrderRow,
-      | 'status'
-      | 'assigned_worker_id'
-      | 'material_json'
-      | 'acceptance_json'
-      | 'draft_acceptance_json'
-      | 'problem_note'
-      | 'reward_amount'
-      | 'sharing_enabled'
-      | 'sharing_total_quantity'
-      | 'sharing_unit_reward'
-      | 'published_at'
-      | 'hall_queued_at'
-      | 'pinned_at'
-      | 'assigned_at'
-      | 'submitted_at'
-      | 'accepted_at'
-      | 'updated_at'
-    >
-  >,
+  patch: WorkOrderUpdatePatch,
+  options: { expectedStatus?: string } = {},
 ): Promise<WorkOrderRow | null> {
-  const current = await getWorkOrderById(workOrderId)
-  if (!current) return null
-  const next = { ...current, ...patch }
-  await query(
-    `
-      UPDATE work_orders
-      SET
-        status = $1, assigned_worker_id = $2, material_json = $3::jsonb,
-        acceptance_json = $4::jsonb, draft_acceptance_json = $5::jsonb,
-        problem_note = $6, reward_amount = $7, sharing_enabled = $8,
-        sharing_total_quantity = $9, sharing_unit_reward = $10, published_at = $11,
-        hall_queued_at = $12, pinned_at = $13, assigned_at = $14, submitted_at = $15,
-        accepted_at = $16, source_order_amount_fen = $17, updated_at = $18
-      WHERE id = $19
-    `,
-    [
-      next.status,
-      next.assigned_worker_id || null,
-      toJsonString(next.material_json),
-      toJsonString(next.acceptance_json),
-      toJsonString(next.draft_acceptance_json),
-      next.problem_note || '',
-      toPositiveInteger(next.reward_amount, 0),
-      next.sharing_enabled === true,
-      toPositiveInteger(next.sharing_total_quantity, 1),
-      toPositiveInteger(next.sharing_unit_reward, 0),
-      next.published_at || null,
-      next.hall_queued_at || null,
-      next.pinned_at || null,
-      next.assigned_at || null,
-      next.submitted_at || null,
-      next.accepted_at || null,
-      // material_json 可能被这次 patch 改写（如补资料 / 快手发码回填），派生列跟着重算
-      resolveSourceOrderAmountFen(next.material_json),
-      next.updated_at,
-      Number(workOrderId),
-    ],
-  )
+  const statement = buildWorkOrderUpdateStatement(workOrderId, patch, options)
+  // 空 patch 不产生写操作；返回当前值，保持原有调用语义。
+  if (!statement) return getWorkOrderById(workOrderId)
+  const result = await query<{ id: number }>(statement.text, statement.values)
+  if (!result.rows[0]) return null
   return getWorkOrderById(workOrderId)
 }
 

apps/backend/src/repositories/worker-platform/work-order-update-atomicity.test.ts

@@ -0,0 +1,90 @@
+import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'
+import test from 'node:test'
+
+import { buildWorkOrderUpdateStatement } from './work-order-management-repo.js'
+
+/**
+ * 2026-10-31 生产事故：客服更新资料时把抢单刚写入的 in_progress 覆盖回 open。
+ * 这里锁住两个约束：只更新显式传入的列；状态流转必须带 expectedStatus 乐观锁。
+ */
+test('资料更新不写 status，只更新 material_json 与派生列', () => {
+  const now = '2026-10-31T11:24:00.000Z'
+  const statement = buildWorkOrderUpdateStatement(
+    195657,
+    {
+      material_json: { source: { orderAmountFen: 800 }, collect: { complete: true } },
+      updated_at: now,
+    },
+    { expectedStatus: 'open' },
+  )
+
+  assert.ok(statement)
+  const setClause = statement.text.slice(
+    statement.text.indexOf('SET') + 'SET'.length,
+    statement.text.indexOf('WHERE'),
+  )
+  assert.ok(!setClause.includes('status ='), '资料更新不得重写 status')
+  assert.ok(setClause.includes('material_json = $1::jsonb'))
+  assert.ok(setClause.includes('source_order_amount_fen = $2'))
+  assert.ok(setClause.includes('updated_at = $3'))
+  assert.ok(statement.text.includes('WHERE id = $4 AND status = $5'))
+  assert.deepEqual(statement.values, [
+    JSON.stringify({ source: { orderAmountFen: 800 }, collect: { complete: true } }),
+    800,
+    now,
+    195657,
+    'open',
+  ])
+})
+
+test('真正的状态流转才写 status，并带读取时的状态做乐观锁', () => {
+  const now = '2026-10-31T11:24:00.000Z'
+  const statement = buildWorkOrderUpdateStatement(
+    42,
+    { status: 'unassigned', updated_at: now },
+    { expectedStatus: 'pending_material' },
+  )
+
+  assert.ok(statement)
+  const setClause = statement.text.slice(
+    statement.text.indexOf('SET') + 'SET'.length,
+    statement.text.indexOf('WHERE'),
+  )
+  assert.match(setClause, /status = \$1/)
+  assert.ok(statement.text.includes('WHERE id = $3 AND status = $4'))
+  assert.deepEqual(statement.values, ['unassigned', now, 42, 'pending_material'])
+})
+
+test('未传入的列不会出现在 UPDATE 中', () => {
+  const statement = buildWorkOrderUpdateStatement(
+    7,
+    { pinned_at: null, updated_at: '2026-10-31T11:24:00.000Z' },
+    { expectedStatus: 'open' },
+  )
+
+  assert.ok(statement)
+  const setClause = statement.text.slice(
+    statement.text.indexOf('SET') + 'SET'.length,
+    statement.text.indexOf('WHERE'),
+  )
+  assert.ok(!statement.text.includes('assigned_worker_id ='))
+  assert.ok(!setClause.includes('status ='))
+  assert.ok(statement.text.includes('pinned_at = $1'))
+})
+
+test('空 patch 不产生写语句', () => {
+  assert.equal(buildWorkOrderUpdateStatement(1, {}), null)
+})
+
+test('资料更新服务只在真实流转时写 status，并用读取时的状态重试', async () => {
+  const source = await readFile(
+    new URL('../../services/worker-platform/admin-work-order-material-service.ts', import.meta.url),
+    'utf8',
+  )
+
+  assert.ok(source.includes('expectedStatus: workOrder.status'))
+  assert.ok(source.includes('if (nextStatus !== workOrder.status) patch.status = nextStatus'))
+  assert.equal(source.includes('status: nextStatus,'), false, '不得无条件把读取时的状态写回')
+  assert.ok(source.includes('MATERIAL_UPDATE_MAX_ATTEMPTS'))
+})

apps/backend/src/services/worker-platform/admin-work-order-material-service.ts

@@ -1,5 +1,6 @@
 import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js'
 import { createWorkOrderEvent, updateWorkOrder } from '../../repositories/worker-platform/index.js'
+import type { WorkOrderUpdatePatch } from '../../repositories/worker-platform/index.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
 import { nowIso } from '../../utils/time.js'
@@ -21,59 +22,15 @@ export async function submitAdminWorkOrderMaterial(
   payload: JsonObject = {},
   actorName = '',
 ) {
-  const workOrder = await getRequiredWorkOrder(workOrderId)
-  const materialCapabilities = mapWorkOrderAdmin(workOrder).capabilities
-  if (!materialCapabilities.canUpdateMaterial) {
-    throw createHttpError('当前状态不能更新订单资料', {
-      statusCode: 409,
-      errorCode: 'work_order_material_status_invalid',
-    })
-  }
+  const saved = await persistAdminWorkOrderMaterial(workOrderId, payload)
+  const { workOrder, updated, complete, nextStatus, rewardAmount } = saved
 
-  const fields = resolveRequirementFields(workOrder)
-  const submittedFields = normalizeSubmittedFields(payload.fields || payload.material || payload)
-  const complete = fields.every(
-    (field) => !field.required || String(submittedFields[field.key] || '').trim(),
-  )
-  const now = nowIso()
-  const screenshots = normalizeUploadedFiles(payload.screenshots || payload.files)
-  const hasActiveShares = Number(workOrder.active_share_quantity || 0) > 0
-  const rewardAmount = resolveAdminMaterialRewardAmount({
-    currentRewardAmount: Number(workOrder.reward_amount || 0),
-    requestedRewardAmount: payload.rewardAmount,
-    sharingEnabled: workOrder.sharing_enabled === true,
-    activeShareQuantity: Number(workOrder.active_share_quantity || 0),
-  })
-  const material = {
-    ...safeParseJson(workOrder.material_json),
-    collect: {
-      fields: submittedFields,
-      screenshots,
-      submittedAt: now,
-      complete,
-      source: 'admin',
-    },
-  }
-  const nextStatus =
-    hasActiveShares ||
-    workOrder.status === WORK_ORDER_STATUS.OPEN ||
-    workOrder.status === WORK_ORDER_STATUS.IN_PROGRESS
-      ? workOrder.status
-      : complete
-        ? WORK_ORDER_STATUS.UNASSIGNED
-        : WORK_ORDER_STATUS.PENDING_MATERIAL
-  const updated = await updateWorkOrder(workOrder.id, {
-    status: nextStatus,
-    material_json: material,
-    reward_amount: rewardAmount,
-    updated_at: now,
-  })
   await createWorkOrderEvent({
     workOrderId: workOrder.id,
     actorType: 'admin',
     actorId: actorName,
     eventType:
-      hasActiveShares ||
+      saved.hasActiveShares ||
       workOrder.status === WORK_ORDER_STATUS.OPEN ||
       workOrder.status === WORK_ORDER_STATUS.IN_PROGRESS
         ? 'material_updated'
@@ -81,12 +38,12 @@ export async function submitAdminWorkOrderMaterial(
     fromStatus: workOrder.status,
     toStatus: nextStatus,
     payloadJson: JSON.stringify({
-      fields: submittedFields,
+      fields: saved.submittedFields,
       complete,
       previousRewardAmount: Number(workOrder.reward_amount || 0),
       rewardAmount,
     }),
-    now,
+    now: saved.now,
   })
   if (nextStatus !== WORK_ORDER_STATUS.PENDING_MATERIAL) {
     await resolveAdminNotificationEntity('work_order', Number(workOrder.id), 'material_completed')
@@ -98,6 +55,87 @@ export async function submitAdminWorkOrderMaterial(
   return { order: mapWorkOrderAdmin(updated || workOrder), complete }
 }
 
+const MATERIAL_UPDATE_MAX_ATTEMPTS = 3
+
+/**
+ * 资料更新只写资料相关列；仅当状态确实需要流转时才写 status，并带 expectedStatus 乐观锁。
+ * 并发抢单把 open 改成 in_progress 时，第一次更新会冲突，下一次按新状态重试，
+ * 避免把忙碌中的工单写回大厅。
+ */
+async function persistAdminWorkOrderMaterial(workOrderId: number | string, payload: JsonObject) {
+  for (let attempt = 0; attempt < MATERIAL_UPDATE_MAX_ATTEMPTS; attempt += 1) {
+    const workOrder = await getRequiredWorkOrder(workOrderId)
+    const materialCapabilities = mapWorkOrderAdmin(workOrder).capabilities
+    if (!materialCapabilities.canUpdateMaterial) {
+      throw createHttpError('当前状态不能更新订单资料', {
+        statusCode: 409,
+        errorCode: 'work_order_material_status_invalid',
+      })
+    }
+
+    const fields = resolveRequirementFields(workOrder)
+    const submittedFields = normalizeSubmittedFields(payload.fields || payload.material || payload)
+    const complete = fields.every(
+      (field) => !field.required || String(submittedFields[field.key] || '').trim(),
+    )
+    const now = nowIso()
+    const screenshots = normalizeUploadedFiles(payload.screenshots || payload.files)
+    const hasActiveShares = Number(workOrder.active_share_quantity || 0) > 0
+    const rewardAmount = resolveAdminMaterialRewardAmount({
+      currentRewardAmount: Number(workOrder.reward_amount || 0),
+      requestedRewardAmount: payload.rewardAmount,
+      sharingEnabled: workOrder.sharing_enabled === true,
+      activeShareQuantity: Number(workOrder.active_share_quantity || 0),
+    })
+    const material = {
+      ...safeParseJson(workOrder.material_json),
+      collect: {
+        fields: submittedFields,
+        screenshots,
+        submittedAt: now,
+        complete,
+        source: 'admin',
+      },
+    }
+    const nextStatus =
+      hasActiveShares ||
+      workOrder.status === WORK_ORDER_STATUS.OPEN ||
+      workOrder.status === WORK_ORDER_STATUS.IN_PROGRESS
+        ? workOrder.status
+        : complete
+          ? WORK_ORDER_STATUS.UNASSIGNED
+          : WORK_ORDER_STATUS.PENDING_MATERIAL
+    const patch: WorkOrderUpdatePatch = {
+      material_json: material,
+      reward_amount: rewardAmount,
+      updated_at: now,
+    }
+    // status 与读取时一致时无需重写；只有真正的流转（补资料/进大厅）才写入。
+    if (nextStatus !== workOrder.status) patch.status = nextStatus
+
+    const updated = await updateWorkOrder(workOrder.id, patch, {
+      expectedStatus: workOrder.status,
+    })
+    if (updated) {
+      return {
+        workOrder,
+        updated,
+        complete,
+        nextStatus,
+        rewardAmount,
+        hasActiveShares,
+        submittedFields,
+        now,
+      }
+    }
+  }
+
+  throw createHttpError('订单状态已变化，资料未保存，请刷新后重试', {
+    statusCode: 409,
+    errorCode: 'work_order_material_conflict',
+  })
+}
+
 /** 更新资料时仅普通接单可改价；已有参与者的拼单必须保留份额价格快照。 */
 export function resolveAdminMaterialRewardAmount(input: {
   currentRewardAmount: number

apps/backend/src/services/worker-platform/worker-platform-service.test.ts

@@ -2855,7 +2855,12 @@ test('筛选统计必须优先读派生列 source_order_amount_fen，不再逐
   )
   const writes = managementRepo.split('resolveSourceOrderAmountFen(').length - 1
   assert.equal(writes, 2, '派生列写入点必须正好两处：建单 INSERT 与通用 UPDATE')
-  assert.ok(managementRepo.includes('source_order_amount_fen = $17'))
+  assert.ok(
+    managementRepo.includes(
+      "assign('source_order_amount_fen', resolveSourceOrderAmountFen(patch.material_json))",
+    ),
+    '通用 UPDATE 必须在 material_json 变更时同步维护派生列',
+  )
 })
 
 test('会话 touch 必须同时保留鉴权语义与写入节流', async () => {