diff --git a/apps/backend/package.json b/apps/backend/package.json index e451f6d0..6220a827 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -12,6 +12,7 @@ "finance:rebuild-derived": "tsx scripts/rebuild-finance-derived.ts", "audit:finance": "tsx scripts/audit-finance-derived.ts", "revenue:backfill": "tsx scripts/backfill-revenue-basis.ts", + "voucher:backfill-order-received": "tsx scripts/backfill-voucher-consume-order-received.ts", "cost-exceptions:archive": "tsx scripts/archive-cost-exceptions.ts", "storage:migrate": "tsx scripts/migrate-storage.ts", "storage:migrate:built": "node dist/storage-migrate.js", diff --git a/apps/backend/scripts/backfill-voucher-consume-order-received.ts b/apps/backend/scripts/backfill-voucher-consume-order-received.ts new file mode 100644 index 00000000..eb223cb5 --- /dev/null +++ b/apps/backend/scripts/backfill-voucher-consume-order-received.ts @@ -0,0 +1,261 @@ +#!/usr/bin/env tsx +/** + * 为部署前已支付、已有可核销券但未进入订单级核销队列的订单补投任务。 + * + * 只处理明确可核销的券: + * - 快手电子凭证:UNUSED 且发码回调成功; + * - 抖音码速达:issued/synced 且同步成功。 + * + * 默认 dry-run。执行必须显式确认。 + */ +import process from 'node:process' + +import { closeDb, query } from '../src/db/client.js' +import { loadAppConfigCacheFromDatabase } from '../src/services/config/app-config-store.js' +import { getVoucherConsumePolicy } from '../src/services/fulfillment/voucher-consume-policy-service.js' +import { scheduleVoucherConsumeForOrderReceived } from '../src/services/fulfillment/voucher-consume-queue-service.js' + +const CONFIRMATION = 'ENQUEUE_ORDER_RECEIVED' +const options = parseOptions(process.argv.slice(2)) +const execute = options.execute === 'true' +const confirmation = String(options.confirm || '').trim() +const days = normalizeInteger(options.days, 7, 1, 90) +const batchSize = normalizeInteger(options.batchSize, 500, 1, 5000) +const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 10000) +const dateTo = normalizeDate(options.to) || today() +const dateFrom = normalizeDate(options.from) || shiftDate(dateTo, -(days - 1)) + +const ELIGIBLE_ORDER_SQL = ` + o.created_at >= $1::date + AND o.created_at < ($2::date + INTERVAL '1 day') + AND o.pay_status = 'paid' + AND o.order_status NOT IN ('cancelled', 'closed', 'refunded') + AND ( + ( + $3::boolean + AND (o.provider IN ('91kaquan', 'open_91', '91') OR o.platform = 'kuaishou') + AND EXISTS ( + SELECT 1 + FROM kuaishou_industry_vouchers voucher + WHERE voucher.oid = o.platform_order_id + AND voucher.status = 'UNUSED' + AND voucher.send_callback_status = 'success' + ) + ) + OR + ( + $4::boolean + AND o.provider = 'douyin_mashida' + AND EXISTS ( + SELECT 1 + FROM redeem_certificates certificate + WHERE certificate.provider = 'douyin_mashida' + AND ( + certificate.local_order_id = o.id + OR certificate.order_id = o.platform_order_id + ) + AND certificate.status IN ('issued', 'synced') + AND certificate.sync_status = 'success' + ) + ) + ) + AND NOT EXISTS ( + SELECT 1 + FROM voucher_consume_jobs job + WHERE job.order_id = o.id + AND job.status IN ('pending', 'processing', 'succeeded') + ) +` + +async function main() { + if (dateFrom > dateTo) { + throw new Error(`--from 不能晚于 --to:${dateFrom} > ${dateTo}`) + } + + await loadAppConfigCacheFromDatabase() + const policy = getVoucherConsumePolicy() + const enabledProviders = (['kuaishou_industry', 'douyin_mashida'] as const).filter( + (provider) => policy.platforms[provider].channels.order_received.enabled, + ) + if (enabledProviders.length === 0) { + console.log( + JSON.stringify( + { + mode: execute ? 'execute' : 'dry-run', + dateFrom, + dateTo, + policyDisabled: true, + enabledProviders, + pendingOrders: 0, + }, + null, + 2, + ), + ) + return + } + + const includeKuaishou = enabledProviders.includes('kuaishou_industry') + const includeDouyin = enabledProviders.includes('douyin_mashida') + const pending = await countEligibleOrders(includeKuaishou, includeDouyin) + const startedAt = Date.now() + + if (!execute) { + const samples = await previewEligibleOrders(includeKuaishou, includeDouyin) + console.log( + JSON.stringify( + { + mode: 'dry-run', + dateFrom, + dateTo, + pendingOrders: pending, + batchSize, + estimatedBatches: Math.ceil(pending / batchSize), + samples, + durationMs: Date.now() - startedAt, + hint: `确认后执行:npm run voucher:backfill-order-received -- --from=${dateFrom} --to=${dateTo} --execute --confirm=${CONFIRMATION}`, + }, + null, + 2, + ), + ) + return + } + + if (confirmation !== CONFIRMATION) { + throw new Error(`--confirm 必须是 ${CONFIRMATION}`) + } + + let cursor = 0 + let scanned = 0 + let queued = 0 + let batches = 0 + + while (batches < maxBatches) { + const orderIds = await loadEligibleOrderIds(cursor, batchSize, includeKuaishou, includeDouyin) + if (orderIds.length === 0) break + + const scheduled = await scheduleVoucherConsumeForOrderReceived(orderIds) + scanned += orderIds.length + queued += scheduled.queuedCount + cursor = orderIds[orderIds.length - 1]! + batches += 1 + + if (orderIds.length < batchSize) break + } + + const remaining = await countEligibleOrders(includeKuaishou, includeDouyin) + console.log( + JSON.stringify( + { + mode: 'execute', + dateFrom, + dateTo, + pendingBefore: pending, + batches, + scannedOrders: scanned, + queuedJobs: queued, + remainingOrders: remaining, + durationMs: Date.now() - startedAt, + }, + null, + 2, + ), + ) + + if (remaining > 0) { + console.error(`仍有 ${remaining} 个订单未入队;请增大 --max-batches 后重跑(幂等)。`) + process.exitCode = 2 + } +} + +async function countEligibleOrders(includeKuaishou: boolean, includeDouyin: boolean) { + const result = await query<{ count: number }>( + `SELECT COUNT(*)::int AS count + FROM orders o + WHERE ${ELIGIBLE_ORDER_SQL}`, + [dateFrom, dateTo, includeKuaishou, includeDouyin], + ) + return Number(result.rows[0]?.count || 0) +} + +async function loadEligibleOrderIds( + afterId: number, + limit: number, + includeKuaishou: boolean, + includeDouyin: boolean, +) { + const result = await query<{ id: number }>( + `SELECT o.id + FROM orders o + WHERE o.id > $5 + AND ${ELIGIBLE_ORDER_SQL} + ORDER BY o.id + LIMIT $6`, + [dateFrom, dateTo, includeKuaishou, includeDouyin, afterId, limit], + ) + return result.rows.map((row) => Number(row.id)) +} + +async function previewEligibleOrders(includeKuaishou: boolean, includeDouyin: boolean) { + const result = await query<{ + id: number + provider: string + platform_order_id: string + created_at: string + }>( + `SELECT o.id, o.provider, o.platform_order_id, o.created_at::text + FROM orders o + WHERE ${ELIGIBLE_ORDER_SQL} + ORDER BY o.id + LIMIT 10`, + [dateFrom, dateTo, includeKuaishou, includeDouyin], + ) + return result.rows +} + +function parseOptions(args: string[]) { + const result: Record = {} + for (const arg of args) { + if (!arg.startsWith('--')) continue + const body = arg.slice(2) + const index = body.indexOf('=') + const key = index === -1 ? body : body.slice(0, index) + const value = index === -1 ? 'true' : body.slice(index + 1) + result[key] = value + result[toCamelCase(key)] = value + } + return result +} + +function toCamelCase(value: string) { + return value.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase()) +} + +function normalizeInteger(value: unknown, fallback: number, min: number, max: number) { + const parsed = Number(value) + if (!Number.isInteger(parsed)) return fallback + return Math.min(max, Math.max(min, parsed)) +} + +function normalizeDate(value: unknown): string { + const text = String(value || '').trim() + return /^\d{4}-\d{2}-\d{2}$/.test(text) ? text : '' +} + +function today() { + return new Date().toLocaleDateString('sv-SE', { timeZone: 'Asia/Shanghai' }) +} + +function shiftDate(value: string, days: number) { + const date = new Date(`${value}T12:00:00+08:00`) + date.setUTCDate(date.getUTCDate() + days) + return date.toLocaleDateString('sv-SE', { timeZone: 'Asia/Shanghai' }) +} + +main() + .catch((error) => { + console.error(error instanceof Error ? error.message : error) + process.exitCode = 1 + }) + .finally(() => closeDb()) diff --git a/apps/backend/src/db/migrations/150_voucher_consume_order_received_scope.sql b/apps/backend/src/db/migrations/150_voucher_consume_order_received_scope.sql new file mode 100644 index 00000000..158ba0b5 --- /dev/null +++ b/apps/backend/src/db/migrations/150_voucher_consume_order_received_scope.sql @@ -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 表示与履约渠道无关的订单级收单触发'; diff --git a/apps/backend/src/services/fulfillment/voucher-consume-job-list.test.ts b/apps/backend/src/services/fulfillment/voucher-consume-job-list.test.ts index f5338c18..67ad9c78 100644 --- a/apps/backend/src/services/fulfillment/voucher-consume-job-list.test.ts +++ b/apps/backend/src/services/fulfillment/voucher-consume-job-list.test.ts @@ -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'"), + '数据库必须允许记录订单级核销任务来源', + ) +}) diff --git a/apps/backend/src/services/fulfillment/voucher-consume-order-received.integration.test.ts b/apps/backend/src/services/fulfillment/voucher-consume-order-received.integration.test.ts new file mode 100644 index 00000000..7c31ba9a --- /dev/null +++ b/apps/backend/src/services/fulfillment/voucher-consume-order-received.integration.test.ts @@ -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() + } + }, +) diff --git a/apps/backend/src/services/fulfillment/voucher-consume-policy-service.test.ts b/apps/backend/src/services/fulfillment/voucher-consume-policy-service.test.ts index 893b64fb..7884b7e2 100644 --- a/apps/backend/src/services/fulfillment/voucher-consume-policy-service.test.ts +++ b/apps/backend/src/services/fulfillment/voucher-consume-policy-service.test.ts @@ -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, + }) +}) diff --git a/apps/backend/src/services/fulfillment/voucher-consume-policy-service.ts b/apps/backend/src/services/fulfillment/voucher-consume-policy-service.ts index 569d9f6e..a6e1e50d 100644 --- a/apps/backend/src/services/fulfillment/voucher-consume-policy-service.ts +++ b/apps/backend/src/services/fulfillment/voucher-consume-policy-service.ts @@ -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 : {} diff --git a/apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts b/apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts index 1f3a705d..fbf2a517 100644 --- a/apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts +++ b/apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts @@ -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( + ` + 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' || diff --git a/apps/backend/src/services/order/order-service.ts b/apps/backend/src/services/order/order-service.ts index 04b3ee4d..c717e4de 100644 --- a/apps/backend/src/services/order/order-service.ts +++ b/apps/backend/src/services/order/order-service.ts @@ -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 diff --git a/apps/frontend/src/pages/admin/panels/VoucherConsumePolicyPanel.tsx b/apps/frontend/src/pages/admin/panels/VoucherConsumePolicyPanel.tsx index 3e2bbe01..02175e47 100644 --- a/apps/frontend/src/pages/admin/panels/VoucherConsumePolicyPanel.tsx +++ b/apps/frontend/src/pages/admin/panels/VoucherConsumePolicyPanel.tsx @@ -37,6 +37,7 @@ const PROVIDER_LABEL: Record = { } const CHANNEL_LABEL: Record = { + order_received: '订单级触发', affiliate_dash: '自动发货', work_order: '接单工单', manual: '人工客服', @@ -49,6 +50,10 @@ const MODE_OPTIONS: Record< VoucherConsumeChannel, Array<{ value: VoucherConsumeMode; label: string }> > = { + order_received: [ + { value: 'after_order_received', label: '收到订单后' }, + { value: 'manual', label: '人工核销' }, + ], affiliate_dash: [ { value: 'after_order_received', label: '收到订单后' }, { value: 'after_link_generated', label: '领取链接生成后' }, @@ -103,6 +108,7 @@ const PROVIDER_FILTER_OPTIONS: Array<{ value: VoucherConsumeProvider; label: str ] const CHANNEL_FILTER_OPTIONS: Array<{ value: VoucherConsumeChannel; label: string }> = [ + { value: 'order_received', label: CHANNEL_LABEL.order_received }, { value: 'affiliate_dash', label: CHANNEL_LABEL.affiliate_dash }, { value: 'work_order', label: CHANNEL_LABEL.work_order }, { value: 'manual', label: CHANNEL_LABEL.manual }, @@ -250,7 +256,7 @@ export default function VoucherConsumePolicyPanel() { render: (provider: VoucherConsumeProvider) => PROVIDER_LABEL[provider], }, { - title: '履约渠道', + title: '策略项', dataIndex: 'channel', width: 140, render: (channel: VoucherConsumeChannel) => CHANNEL_LABEL[channel], diff --git a/apps/frontend/src/services/admin/fulfillment-channel.ts b/apps/frontend/src/services/admin/fulfillment-channel.ts index 539aea52..fd6c36e1 100644 --- a/apps/frontend/src/services/admin/fulfillment-channel.ts +++ b/apps/frontend/src/services/admin/fulfillment-channel.ts @@ -162,6 +162,7 @@ export type FinancePipelineStatus = { export type VoucherConsumeProvider = 'kuaishou_industry' | 'douyin_mashida' export type VoucherConsumeChannel = + | 'order_received' | 'affiliate_dash' | 'work_order' | 'manual' diff --git a/apps/frontend/tests/voucher-consume-job-list-ui.test.ts b/apps/frontend/tests/voucher-consume-job-list-ui.test.ts index e19c7c29..d7f008dd 100644 --- a/apps/frontend/tests/voucher-consume-job-list-ui.test.ts +++ b/apps/frontend/tests/voucher-consume-job-list-ui.test.ts @@ -22,7 +22,13 @@ test('最近核销任务使用服务端分页', () => { assert.ok(!panel.includes('pagination={false}\n scroll={{ x: 1200 }}')) }) -test('自动核销策略提供未命中履约渠道的平台兜底配置', () => { +test('自动核销策略提供独立于履约渠道的订单级触发配置', () => { + assert.ok(panel.includes("order_received: '订单级触发'")) + assert.ok(panel.includes("value: 'order_received'")) + assert.ok(panel.includes("title: '策略项'")) +}) + +test('旧未命中渠道兜底配置继续保留并兼容', () => { assert.ok(panel.includes("unmatched: '未命中渠道兜底'")) assert.ok(panel.includes("label: '未命中渠道后'")) assert.ok(panel.includes("value: 'unmatched'")) diff --git a/deploy/post-deploy.sh b/deploy/post-deploy.sh index f3c9771e..a297e8d8 100755 --- a/deploy/post-deploy.sh +++ b/deploy/post-deploy.sh @@ -7,7 +7,7 @@ set -Eeuo pipefail # # 它把以前散落在文档里的脚本步骤串成一条可重复执行的流水线: # ① 存量订单项收入回填 revenue:backfill -# ② 渠道/商品由管道处理(无需回填) +# ② 订单级收单核销补投 voucher:backfill-order-received # ③ 财务历史重建(分窗口) finance:rebuild-derived # ④ 上线验收(只读 SQL) scripts/verify-finance-launch.sql # @@ -40,6 +40,7 @@ WINDOW_DAYS="${WINDOW_DAYS:-30}" BATCH_SIZE="${BATCH_SIZE:-5000}" MAX_BATCHES="${MAX_BATCHES:-1000}" MAX_BACKFILL_ROUNDS="${MAX_BACKFILL_ROUNDS:-5}" +VOUCHER_BACKFILL_DAYS="${VOUCHER_BACKFILL_DAYS:-7}" DRY_RUN=false SKIP_BACKFILL=false @@ -81,6 +82,7 @@ usage() { POST_DEPLOY_COMPOSE_FILE 指定 compose 文件(默认 docker-compose.yml) TOOLS_IMAGE 工具镜像名(默认 order-site/tools:latest) WINDOW_DAYS / BATCH_SIZE / MAX_BATCHES 同上对应选项的默认值 + VOUCHER_BACKFILL_DAYS 订单级收单核销补投回看天数(默认 7) 首次上线顺序(避免商品未归一先同步产生大量异常): ① bash deploy/post-deploy.sh --only=backfill @@ -147,6 +149,7 @@ validate_integer "--window-days" "$WINDOW_DAYS" 1 400 validate_integer "--batch-size" "$BATCH_SIZE" 1 50000 validate_integer "--max-batches" "$MAX_BATCHES" 1 10000 validate_integer "MAX_BACKFILL_ROUNDS" "$MAX_BACKFILL_ROUNDS" 1 100 +validate_integer "VOUCHER_BACKFILL_DAYS" "$VOUCHER_BACKFILL_DAYS" 1 90 should_run() { [[ -z "$ONLY" || "$ONLY" == "$1" ]] @@ -362,7 +365,20 @@ step_backfill() { run_backfill revenue:backfill --batchSize="$BATCH_SIZE" --maxBatches="$MAX_BATCHES" fi - log "② 渠道判定与商品归一出管道负责,无需手工回填" + log "② 补投订单级收单核销任务(最近 ${VOUCHER_BACKFILL_DAYS} 天)" + if [[ "$DRY_RUN" == true ]]; then + if docker image inspect "$TOOLS_IMAGE" >/dev/null 2>&1; then + run_tool voucher:backfill-order-received --days="$VOUCHER_BACKFILL_DAYS" || + warn "订单级核销补投 dry-run 预览失败(不影响后续)" + else + log " [dry-run] 跳过执行(工具镜像未构建)" + fi + else + run_tool voucher:backfill-order-received \ + --days="$VOUCHER_BACKFILL_DAYS" \ + --execute \ + --confirm=ENQUEUE_ORDER_RECEIVED + fi if [[ "$DRY_RUN" != true ]]; then refresh_stats