diff --git a/apps/backend/src/repositories/worker-platform/work-order-management-repo.ts b/apps/backend/src/repositories/worker-platform/work-order-management-repo.ts index e1ab9f42..d3fa60d1 100644 --- a/apps/backend/src/repositories/worker-platform/work-order-management-repo.ts +++ b/apps/backend/src/repositories/worker-platform/work-order-management-repo.ts @@ -66,69 +66,109 @@ export async function createWorkOrder(input: CreateWorkOrderInput): Promise +> + +/** + * 只更新 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 { - 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) } diff --git a/apps/backend/src/repositories/worker-platform/work-order-update-atomicity.test.ts b/apps/backend/src/repositories/worker-platform/work-order-update-atomicity.test.ts new file mode 100644 index 00000000..9bb21072 --- /dev/null +++ b/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')) +}) diff --git a/apps/backend/src/services/worker-platform/admin-work-order-material-service.ts b/apps/backend/src/services/worker-platform/admin-work-order-material-service.ts index e21565aa..8d130999 100644 --- a/apps/backend/src/services/worker-platform/admin-work-order-material-service.ts +++ b/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 diff --git a/apps/backend/src/services/worker-platform/worker-platform-service.test.ts b/apps/backend/src/services/worker-platform/worker-platform-service.test.ts index 52eb6b92..8943edaf 100644 --- a/apps/backend/src/services/worker-platform/worker-platform-service.test.ts +++ b/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 () => {