fix(fulfillment): 修复自动发货任务上下文并发覆盖

This commit is contained in:
yml2213
2026-11-03 13:14:07 +08:00
parent b3fb330ad9
commit b842d00fc3
4 changed files with 132 additions and 19 deletions
@@ -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}', '') = ''
);
@@ -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<TaskRow | null> {
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,
@@ -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<string, unknown> | undefined)?.oid,
'2630702009068141',
)
assert.equal('affiliateDash' in (capturedPatch || {}), false)
})
@@ -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[]) {