dachui/order_site
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复：收到订单后按订单级触发自动核销

Browse Source
This commit is contained in:
yml2213
2026-11-08 21:29:51 +08:00
parent d76e8265f8
commit 996054a5c5
13 changed files with 683 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/db/migrations/150_voucher_consume_order_received_scope.sql
+24
View File
@@ -0,0 +1,24 @@
-- 150_voucher_consume_order_received_scope.sql
--
-- “收到订单后”核销改为订单级触发：
-- 1. 不再依赖履约渠道、订单项状态或待处理原因；
-- 2. 订单级任务以 order:<订单ID> 去重；
-- 3. channel=order_received 仅用于任务来源审计和后台筛选。
ALTER TABLE voucher_consume_jobs
DROP CONSTRAINT IF EXISTS voucher_consume_jobs_channel_check;
ALTER TABLE voucher_consume_jobs
ADD CONSTRAINT voucher_consume_jobs_channel_check
CHECK (channel IN (
'order_received',
'affiliate_dash',
'work_order',
'manual',
'third_party_voucher',
'on_account',
'unmatched'
));
COMMENT ON COLUMN voucher_consume_jobs.channel IS
'触发核销时的策略来源；order_received 表示与履约渠道无关的订单级收单触发';
apps/backend/src/services/fulfillment/voucher-consume-job-list.test.ts
+23
View File
@@ -11,10 +11,18 @@ const fallbackMigration = readFileSync(
new URL('../../db/migrations/146_voucher_consume_fallback_channel.sql', import.meta.url),
'utf8',
)
const orderReceivedMigration = readFileSync(
new URL('../../db/migrations/150_voucher_consume_order_received_scope.sql', import.meta.url),
'utf8',
)
const channelAssignSource = readFileSync(
new URL('./channel-assign-service.ts', import.meta.url),
'utf8',
)
const orderServiceSource = readFileSync(
new URL('../order/order-service.ts', import.meta.url),
'utf8',
)
test('核销任务列表支持关键词、平台、渠道和状态组合筛选', () => {
assert.ok(source.includes('keyword?: string'))
@@ -63,3 +71,18 @@ test('未命中履约渠道时按平台策略安排兜底核销，不修改订
'只允许未命中渠道进入兜底，不能覆盖无载体等其它问题',
)
})
test('收到订单后按订单级策略入队，不依赖订单项渠道或状态', () => {
assert.ok(source.includes('scheduleVoucherConsumeForOrderReceived'))
assert.ok(source.includes("channel: 'order_received'"))
assert.ok(source.includes('target_key: `order:${row.id}`'))
assert.ok(source.includes("AND o.pay_status = 'paid'"))
assert.ok(
orderServiceSource.includes('scheduleVoucherConsumeForOrderReceived([Number(order.id)])'),
'订单入库成功后必须投递订单级收单核销任务',
)
assert.ok(
orderReceivedMigration.includes("'order_received'"),
'数据库必须允许记录订单级核销任务来源',
)
})
apps/backend/src/services/fulfillment/voucher-consume-order-received.integration.test.ts
+111
View File
@@ -0,0 +1,111 @@
/**
* 订单级收单核销集成测试（需要真实 PostgreSQL）。
*
* 运行：
* VOUCHER_CONSUME_ORDER_RECEIVED_TEST_DATABASE_URL="$DATABASE_URL" \
* npx tsx --test src/services/fulfillment/voucher-consume-order-received.integration.test.ts
*/
import assert from 'node:assert/strict'
import test from 'node:test'
import type { VoucherConsumePolicy } from './voucher-consume-policy-service.js'
const TEST_DATABASE_URL = String(
process.env.VOUCHER_CONSUME_ORDER_RECEIVED_TEST_DATABASE_URL || '',
).trim()
test(
'订单级收单核销不依赖履约渠道或订单项状态',
{ skip: !TEST_DATABASE_URL ? '未设置 VOUCHER_CONSUME_ORDER_RECEIVED_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const { loadAppConfigCacheFromDatabase } = await import('../config/app-config-store.js')
const { getVoucherConsumePolicy, saveVoucherConsumePolicy } =
await import('./voucher-consume-policy-service.js')
const { scheduleVoucherConsumeForOrderReceived, listVoucherConsumeJobs } =
await import('./voucher-consume-queue-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const platformOrderId = `IT-ORDER-RECEIVED-${suffix}`
let orderId = 0
let orderItemId = 0
let originalPolicy: VoucherConsumePolicy | null = null
try {
await loadAppConfigCacheFromDatabase()
originalPolicy = getVoucherConsumePolicy()
const enabledPolicy = structuredClone(originalPolicy)
enabledPolicy.enabled = true
enabledPolicy.platforms.kuaishou_industry.enabled = true
enabledPolicy.platforms.kuaishou_industry.channels.order_received = {
enabled: true,
mode: 'after_order_received',
delaySeconds: 7,
}
await saveVoucherConsumePolicy(enabledPolicy)
const order = await query<{ id: number }>(
`INSERT INTO orders (
provider, platform, shop_id, shop_name, platform_order_id, order_status, pay_status,
total_amount, paid_at, created_at, updated_at
) VALUES (
'91kaquan', 'kuaishou', 'it-order-received', '', $1, 'paid', 'paid',
1000, NOW(), NOW(), NOW()
) RETURNING id`,
[platformOrderId],
)
orderId = Number(order.rows[0]!.id)
const item = await query<{ id: number }>(
`INSERT INTO order_items (
order_id, sku_code, sku_name, quantity, spec_json,
revenue_amount_fen, revenue_basis, created_at, updated_at
) VALUES (
$1, 'it-order-received-sku', '订单级核销测试商品', 1, '{}'::jsonb,
1000, 'order_total', NOW(), NOW()
) RETURNING id`,
[orderId],
)
orderItemId = Number(item.rows[0]!.id)
const before = Date.now()
const scheduled = await scheduleVoucherConsumeForOrderReceived([orderId])
assert.equal(scheduled.targetCount, 1)
assert.equal(scheduled.queuedCount, 1)
assert.equal(scheduled.jobIds.length, 1)
const jobs = await listVoucherConsumeJobs({
keyword: platformOrderId,
page: 1,
pageSize: 20,
})
assert.equal(jobs.total, 1)
const job = jobs.items[0]
assert.equal(job?.channel, 'order_received')
assert.equal(job?.trigger_event, 'order_received')
assert.equal(job?.target_key, `order:${orderId}`)
assert.equal(Number(job?.order_id), orderId)
assert.equal(job?.order_item_id, null)
assert.ok(Date.parse(String(job?.due_at)) >= before + 6_000)
const repeated = await scheduleVoucherConsumeForOrderReceived([orderId])
assert.equal(repeated.targetCount, 1)
assert.equal(repeated.queuedCount, 0)
} finally {
if (orderId > 0) {
await query(`DELETE FROM voucher_consume_jobs WHERE order_id = $1`, [orderId])
}
if (orderItemId > 0) {
await query(`DELETE FROM order_items WHERE id = $1`, [orderItemId])
}
if (orderId > 0) {
await query(`DELETE FROM orders WHERE id = $1`, [orderId])
}
if (originalPolicy) {
await saveVoucherConsumePolicy(originalPolicy)
}
await closeDb()
}
},
)
apps/backend/src/services/fulfillment/voucher-consume-policy-service.test.ts
+60
View File
@@ -10,6 +10,11 @@ import {
test('默认核销策略：自动发货 5 秒，工单发布或指派后 10 秒，人工任务 10 秒', () => {
const policy = createDefaultVoucherConsumePolicy()
assert.deepEqual(policy.platforms.kuaishou_industry.channels.order_received, {
enabled: false,
mode: 'after_order_received',
delaySeconds: 5,
})
assert.deepEqual(policy.platforms.kuaishou_industry.channels.affiliate_dash, {
enabled: true,
mode: 'after_link_generated',
@@ -189,3 +194,58 @@ test('旧配置自动补齐未命中渠道兜底，并支持手动开启', () =>
{ enabled: true, delaySeconds: 12, mode: 'after_order_received' },
)
})
test('旧配置开启未命中渠道兜底时，自动继承为订单级收单核销', () => {
const policy = normalizeVoucherConsumePolicy({
version: 2,
enabled: true,
platforms: {
kuaishou_industry: {
enabled: true,
channels: {
unmatched: {
enabled: true,
mode: 'after_order_received',
delaySeconds: 12,
},
},
},
},
})
assert.deepEqual(policy.platforms.kuaishou_industry.channels.order_received, {
enabled: true,
mode: 'after_order_received',
delaySeconds: 12,
})
})
test('显式订单级策略优先于旧未命中渠道兜底配置', () => {
const policy = normalizeVoucherConsumePolicy({
version: 2,
enabled: true,
platforms: {
kuaishou_industry: {
enabled: true,
channels: {
order_received: {
enabled: false,
mode: 'manual',
delaySeconds: 0,
},
unmatched: {
enabled: true,
mode: 'after_order_received',
delaySeconds: 12,
},
},
},
},
})
assert.deepEqual(policy.platforms.kuaishou_industry.channels.order_received, {
enabled: false,
mode: 'manual',
delaySeconds: 0,
})
})
apps/backend/src/services/fulfillment/voucher-consume-policy-service.ts
+23
View File
@@ -6,6 +6,7 @@ export const VOUCHER_CONSUME_PROVIDERS = ['kuaishou_industry', 'douyin_mashida']
export type VoucherConsumeProvider = (typeof VOUCHER_CONSUME_PROVIDERS)[number]
export const VOUCHER_CONSUME_CHANNELS = [
'order_received',
'affiliate_dash',
'work_order',
'manual',
@@ -132,6 +133,11 @@ function createDefaultPlatformPolicy(): VoucherPlatformConsumePolicy {
return {
enabled: true,
channels: {
order_received: {
enabled: false,
mode: 'after_order_received',
delaySeconds: 5,
},
affiliate_dash: {
enabled: true,
mode: 'after_link_generated',
@@ -174,10 +180,16 @@ function normalizePlatformPolicy(
const source = isPlainObject(rawValue) ? rawValue : {}
const rawChannels = isPlainObject(source.channels) ? source.channels : {}
const legacyThirdPartyDisabled = isLegacyThirdPartyDisabled(rawValue, version)
const rawOrderReceived =
rawChannels.order_received ?? inferOrderReceivedPolicyFromUnmatched(rawChannels.unmatched)
return {
enabled: normalizeBoolean(source.enabled, fallback.enabled),
channels: {
order_received: normalizeChannelPolicy(rawOrderReceived, fallback.channels.order_received, [
'manual',
'after_order_received',
]),
affiliate_dash: normalizeChannelPolicy(
rawChannels.affiliate_dash,
fallback.channels.affiliate_dash,
@@ -213,6 +225,17 @@ function normalizePlatformPolicy(
}
}
function inferOrderReceivedPolicyFromUnmatched(rawValue: unknown) {
if (!isPlainObject(rawValue)) return {}
if (rawValue.enabled !== true) return {}
if (String(rawValue.mode || '').trim() !== 'after_order_received') return {}
return {
enabled: true,
mode: 'after_order_received',
delaySeconds: rawValue.delaySeconds,
}
}
function isLegacyThirdPartyDisabled(rawPlatform: unknown, version: number): boolean {
if (version >= 2 || !isPlainObject(rawPlatform)) return false
const rawChannels = isPlainObject(rawPlatform.channels) ? rawPlatform.channels : {}
apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts
+135 -4
View File
@@ -110,10 +110,19 @@ type VoucherConsumeTargetInsert = {
target_key: string
platform_order_id: string
order_id: number
order_item_id: number
order_item_id: number | null
due_at: string
}
type VoucherOrderTargetRow = {
id: number
provider: string
platform: string
platform_order_id: string
pay_status: string
order_status: string
}
const RETRY_DELAYS_SECONDS = [10, 30, 60, 300, 600, 900, 1800]
const PROCESSING_LOCK_SECONDS = 120
const VOUCHER_CONSUME_JOB_STATUSES = [
@@ -224,9 +233,32 @@ export async function scheduleVoucherConsumeForOrderItems(
for (const row of rows.rows) {
const provider = resolveVoucherProviderFromFields(row.provider, row.platform)
const channel = normalizeVoucherChannel(row.fulfillment_channel)
if (!provider || !channel) continue
if (!provider) continue
if (event === 'order_received') {
const orderReceivedPolicy = resolveVoucherConsumePolicy({
provider,
channel: 'order_received',
event,
policy,
})
if (orderReceivedPolicy.enabled) {
targets.push({
provider,
channel: 'order_received',
trigger_event: event,
target_key: `order:${row.order_id}`,
platform_order_id: String(row.platform_order_id || '').trim(),
order_id: Number(row.order_id),
order_item_id: null,
due_at: new Date(Date.now() + orderReceivedPolicy.delaySeconds * 1000).toISOString(),
})
continue
}
}
const channel = normalizeVoucherChannel(row.fulfillment_channel)
if (!channel) continue
const resolved = resolveVoucherConsumePolicy({
provider,
channel,
@@ -250,6 +282,66 @@ export async function scheduleVoucherConsumeForOrderItems(
return queueVoucherConsumeTargets(targets, event)
}
/**
* 订单级“收到订单后”核销。
*
* 该入口只判断平台策略，不读取履约渠道、订单项状态、载体或待处理原因。
* 券可能尚未到达；任务执行时会重新解析券，并按既有重试策略等待发码完成。
*/
export async function scheduleVoucherConsumeForOrderReceived(
orderIds: number[],
): Promise<{ targetCount: number; queuedCount: number; jobIds: number[] }> {
const ids = [...new Set(orderIds.map(Number).filter((id) => Number.isInteger(id) && id > 0))]
if (ids.length === 0) {
return { targetCount: 0, queuedCount: 0, jobIds: [] }
}
const rows = await query<VoucherOrderTargetRow>(
`
SELECT
o.id,
o.provider,
o.platform,
o.platform_order_id,
o.pay_status,
o.order_status
FROM orders o
WHERE o.id = ANY($1::bigint[])
AND o.pay_status = 'paid'
AND o.order_status NOT IN ('cancelled', 'closed', 'refunded')
`,
[ids],
)
const policy = getVoucherConsumePolicy()
const targets: VoucherConsumeTargetInsert[] = []
for (const row of rows.rows) {
const provider = resolveVoucherProviderFromFields(row.provider, row.platform)
if (!provider) continue
const resolved = resolveVoucherConsumePolicy({
provider,
channel: 'order_received',
event: 'order_received',
policy,
})
if (!resolved.enabled) continue
targets.push({
provider,
channel: 'order_received',
trigger_event: 'order_received',
target_key: `order:${row.id}`,
platform_order_id: String(row.platform_order_id || '').trim(),
order_id: Number(row.id),
order_item_id: null,
due_at: new Date(Date.now() + resolved.delaySeconds * 1000).toISOString(),
})
}
return queueVoucherConsumeTargets(targets, 'order_received')
}
/**
* 未命中任何履约渠道时，按平台兜底策略安排核销。
*
@@ -285,11 +377,32 @@ export async function scheduleVoucherConsumeFallbackForOrderItems(
)
const policy = getVoucherConsumePolicy()
const targets: VoucherConsumeTargetInsert[] = []
const orderReceivedTargets: VoucherConsumeTargetInsert[] = []
for (const row of rows.rows) {
const provider = resolveVoucherProviderFromFields(row.provider, row.platform)
if (!provider) continue
const orderReceivedPolicy = resolveVoucherConsumePolicy({
provider,
channel: 'order_received',
event,
policy,
})
if (orderReceivedPolicy.enabled) {
orderReceivedTargets.push({
provider,
channel: 'order_received',
trigger_event: event,
target_key: `order:${row.order_id}`,
platform_order_id: String(row.platform_order_id || '').trim(),
order_id: Number(row.order_id),
order_item_id: null,
due_at: new Date(Date.now() + orderReceivedPolicy.delaySeconds * 1000).toISOString(),
})
continue
}
const resolved = resolveVoucherConsumePolicy({
provider,
channel: 'unmatched',
@@ -310,7 +423,15 @@ export async function scheduleVoucherConsumeFallbackForOrderItems(
})
}
return queueVoucherConsumeTargets(targets, event)
const [orderReceivedResult, fallbackResult] = await Promise.all([
queueVoucherConsumeTargets(orderReceivedTargets, event),
queueVoucherConsumeTargets(targets, event),
])
return {
targetCount: orderReceivedResult.targetCount + fallbackResult.targetCount,
queuedCount: orderReceivedResult.queuedCount + fallbackResult.queuedCount,
jobIds: [...orderReceivedResult.jobIds, ...fallbackResult.jobIds],
}
}
async function queueVoucherConsumeTargets(
@@ -360,6 +481,15 @@ async function queueVoucherConsumeTargets(
order_item_id bigint,
due_at timestamptz
)
WHERE target.channel <> 'order_received'
OR NOT EXISTS (
SELECT 1
FROM voucher_consume_jobs active
WHERE active.provider = target.provider
AND active.order_id = target.order_id
AND active.target_key <> target.target_key
AND active.status IN ('pending', 'processing', 'succeeded')
)
ON CONFLICT (provider, target_key)
WHERE status IN ('pending', 'processing', 'succeeded')
DO NOTHING
@@ -1016,6 +1146,7 @@ function resolveVoucherProviderFromFields(
function normalizeVoucherChannel(value: unknown): VoucherConsumeChannel | null {
const channel = String(value || '').trim()
if (
channel === 'order_received' ||
channel === 'affiliate_dash' ||
channel === 'work_order' ||
channel === 'manual' ||
apps/backend/src/services/order/order-service.ts
+12
View File
@@ -21,6 +21,7 @@ import {
resolveOrderFulfillmentReadiness,
syncOrderFulfillmentAttachments,
} from '../fulfillment/order-fulfillment-readiness-service.js'
import { scheduleVoucherConsumeForOrderReceived } from '../fulfillment/voucher-consume-queue-service.js'
import { syncWorkerOrdersForSourceOrder } from '../worker-platform/index.js'
import {
publishAdminOrderRealtimeChange,
@@ -212,6 +213,17 @@ export async function upsertOrderFromSource(
const { order, orderItems } = persisted
if (String(order.pay_status || '').trim() === 'paid') {
await scheduleVoucherConsumeForOrderReceived([Number(order.id)]).catch((error) => {
logWarn('[order-service]', '收到订单后安排订单级自动核销失败', {
orderId: order.id,
provider: order.provider,
platformOrderId: order.platform_order_id,
error: error instanceof Error ? error.message : String(error),
})
})
}
const configuredOrderItems = orderItems.filter(isOrderItemConfiguredForFulfillment)
const readiness =
configuredOrderItems.length > 0