diff --git a/apps/backend/src/db/migrations/127_repair_affiliate_dash_task_context.sql b/apps/backend/src/db/migrations/127_repair_affiliate_dash_task_context.sql new file mode 100644 index 00000000..06266ee9 --- /dev/null +++ b/apps/backend/src/db/migrations/127_repair_affiliate_dash_task_context.sql @@ -0,0 +1,40 @@ +-- 127_repair_affiliate_dash_task_context.sql +-- +-- 修复电子凭证绑定并发覆盖任务上下文的问题: +-- affiliate_dash 建单事件已经记录了 orderNo/clientOrderNo,但任务 context_json +-- 随后被旧 TaskRow 整段覆盖为空。回调按这两个键查任务时会 404。 +-- 只补缺失键,不覆盖现有 affiliateDash 其它字段。 + +WITH latest_created AS ( + SELECT DISTINCT ON (te.task_id) + te.task_id, + NULLIF(BTRIM(te.payload_json->>'orderNo'), '') AS order_no, + NULLIF(BTRIM(te.payload_json->>'clientOrderNo'), '') AS client_order_no + FROM task_events te + WHERE te.event_type = 'affiliate_dash_order_created' + ORDER BY te.task_id, te.created_at DESC, te.id DESC +) +UPDATE fulfillment_tasks ft +SET + context_json = COALESCE(ft.context_json, '{}'::jsonb) || jsonb_build_object( + 'affiliateDash', + COALESCE(ft.context_json->'affiliateDash', '{}'::jsonb) || jsonb_build_object( + 'orderNo', + latest_created.order_no, + 'clientOrderNo', + COALESCE( + latest_created.client_order_no, + NULLIF(ft.context_json #>> '{affiliateDash,clientOrderNo}', ''), + ft.platform_order_id + ) + ) + ), + updated_at = NOW() +FROM latest_created +WHERE ft.id = latest_created.task_id + AND ft.executor_key = 'affiliate_dash' + AND latest_created.order_no IS NOT NULL + AND ( + COALESCE(ft.context_json #>> '{affiliateDash,orderNo}', '') = '' + OR COALESCE(ft.context_json #>> '{affiliateDash,clientOrderNo}', '') = '' + ); diff --git a/apps/backend/src/repositories/task-repo.ts b/apps/backend/src/repositories/task-repo.ts index 2fece3cc..66bd067c 100644 --- a/apps/backend/src/repositories/task-repo.ts +++ b/apps/backend/src/repositories/task-repo.ts @@ -10,6 +10,7 @@ import type { TaskUpdatePatch, } from '../types/repository/inputs.js' import type { TaskListQueryResult, TaskRow } from '../types/repository/rows.js' +import type { JsonObject } from '../types/json.js' type TaskQueryExecutor = (text: string, params?: unknown[]) => Promise<{ rows: unknown[] }> const AFFILIATE_DASH_EXECUTOR_KEY = 'affiliate_dash' @@ -255,6 +256,31 @@ export async function updateTask( }) } +/** + * 原子合并任务上下文,避免调用方拿旧 TaskRow 整段覆盖并发写入的其它字段。 + * 只合并顶层 key;调用方应传入完整的子对象 patch。 + */ +export async function mergeTaskContext( + taskId: number | string, + patch: JsonObject, + updatedAt?: string | undefined, +): Promise { + return withTransaction(async (client) => { + const result = await client.query<{ id: number }>( + ` + UPDATE fulfillment_tasks + SET context_json = COALESCE(context_json, '{}'::jsonb) || $2::jsonb, + updated_at = COALESCE($3, updated_at) + WHERE id = $1 + RETURNING id + `, + [Number(taskId), JSON.stringify(patch), updatedAt || null], + ) + if (!result.rows[0]) return null + return getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, taskId) + }) +} + export async function updateTaskStatusIfCurrent( taskId: number | string, currentStatus: string, diff --git a/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.test.ts b/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.test.ts new file mode 100644 index 00000000..0a1432b1 --- /dev/null +++ b/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.test.ts @@ -0,0 +1,54 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import type { JsonObject } from '../../../types/json.js' +import type { KuaishouIndustryVoucherRow, TaskRow } from '../../../types/repository/rows.js' +import { attachKuaishouIndustryVoucherToTask } from './voucher-binding-service.js' + +test('电子凭证绑定只原子合并自己的 context key,不覆盖 affiliateDash', async () => { + const task = { + id: 120509, + context_json: { + affiliateDash: { + orderNo: 'FO2026110313080188725b91e99', + clientOrderNo: '2630702009068141', + }, + }, + } as unknown as TaskRow + const voucher = { + oid: '2630702009068141', + voucher_code: 'KSV4YQWP6SBEQFRBCBD', + token: 'token', + status: 'CONSUMED', + unit_index: 1, + send_callback_status: 'success', + send_callback_attempt_count: 1, + } as unknown as KuaishouIndustryVoucherRow + let capturedTaskId = 0 + let capturedPatch: JsonObject | null = null + let capturedUpdatedAt = '' + + const merged = await attachKuaishouIndustryVoucherToTask( + task, + voucher, + { now: '2026-11-03T13:08:01.383Z' }, + { + mergeTaskContext: async (taskId, patch, updatedAt) => { + capturedTaskId = Number(taskId) + capturedPatch = patch + capturedUpdatedAt = String(updatedAt || '') + return task + }, + }, + ) + + assert.equal(merged, task) + assert.equal(capturedTaskId, 120509) + assert.equal(capturedUpdatedAt, '2026-11-03T13:08:01.383Z') + assert.deepEqual(Object.keys(capturedPatch || {}), ['kuaishouIndustryVoucher']) + assert.equal( + (capturedPatch?.kuaishouIndustryVoucher as Record | undefined)?.oid, + '2630702009068141', + ) + assert.equal('affiliateDash' in (capturedPatch || {}), false) +}) diff --git a/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.ts b/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.ts index 83c91e22..80c9c60e 100644 --- a/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.ts +++ b/apps/backend/src/services/platforms/kuaishou-industry/voucher-binding-service.ts @@ -1,11 +1,10 @@ -import { updateTask } from '../../../repositories/task-repo.js' +import { mergeTaskContext } from '../../../repositories/task-repo.js' import type { JsonObject } from '../../../types/json.js' import { listKuaishouIndustryVouchersByOid, updateKuaishouIndustryVoucherByCode, } from '../../../repositories/kuaishou-industry-voucher-repo.js' import { normalizeTimestampIso } from '../../../utils/time.js' -import { parseTaskContext } from '../../../utils/task-json.js' import { isKuaishouIndustryVoucherSendCallbackSuccess, normalizeKuaishouIndustrySendCallbackStatus, @@ -17,6 +16,10 @@ import type { TaskRow, } from '../../../types/repository/rows.js' +type VoucherBindingDeps = { + mergeTaskContext?: typeof mergeTaskContext +} + export async function bindKuaishouIndustryVouchersToOrderTasks( order: OrderRow | null | undefined, tasks: TaskRow[] = [], @@ -62,27 +65,17 @@ export async function attachKuaishouIndustryVoucherToTask( source?: string now?: string } = {}, + deps: VoucherBindingDeps = {}, ) { const now = normalizeTimestampIso(options.now || new Date().toISOString()) - const context = parseTaskContext(task) - const nextVoucherContext = buildKuaishouIndustryVoucherContext( - voucher, - context.kuaishouIndustryVoucher, + const nextVoucherContext = buildKuaishouIndustryVoucherContext(voucher, null, now) + const mergeContext = deps.mergeTaskContext || mergeTaskContext + // 只原子替换本模块负责的顶层 key,禁止用旧 TaskRow 整段覆盖 affiliateDash。 + return mergeContext( + task.id, + { kuaishouIndustryVoucher: nextVoucherContext } satisfies JsonObject, now, ) - const nextContext: JsonObject = { - ...context, - kuaishouIndustryVoucher: nextVoucherContext, - } - - if (JSON.stringify(context) === JSON.stringify(nextContext)) { - return task - } - - return updateTask(task.id, { - context_json: JSON.stringify(nextContext), - updated_at: now, - }) } function resolveTaskForVoucher(voucher: KuaishouIndustryVoucherRow, tasks: TaskRow[]) {