import { query } from '../../db/client.js' import { getOrderById } from '../../repositories/order-repo.js' import { getOrderItemById, listOrderItemsByOrderId } from '../../repositories/order-item-repo.js' import { getTaskById, listTasksByOrderId } from '../../repositories/task-repo.js' import { listKuaishouIndustryVouchersByOid } from '../../repositories/kuaishou-industry-voucher-repo.js' import { findRedeemCertificateByOrder, listRedeemCertificatesByLocalOrder, type RedeemCertificateRow, } from '../../repositories/redeem-certificate-repo.js' import type { WorkOrderRow } from '../../repositories/worker-platform/index.js' import { DOUYIN_MASHIDA_PROVIDER, isDouyinMashidaProvider, } from '../platforms/douyin-mashida/order-mapper.js' import { consumeKuaishouIndustryVoucher, isKuaishouIndustryVoucherSendCallbackSuccess, normalizeVoucherStatus, } from '../platforms/kuaishou-industry/voucher-service.js' import { attachKuaishouIndustryVoucherToTask } from '../platforms/kuaishou-industry/voucher-binding-service.js' import { verifyCertificate } from '../redeem/certificate-service.js' import { logIntegration, logWarn } from '../../utils/logger.js' import { getVoucherConsumePolicy, resolveVoucherConsumePolicy, type VoucherConsumeChannel, type VoucherConsumeEvent, type VoucherConsumeProvider, } from './voucher-consume-policy-service.js' import type { KuaishouIndustryVoucherRow, OrderItemRow, OrderRow, TaskRow, } from '../../types/repository/rows.js' import { OPEN_91_PLATFORM, OPEN_91_PROVIDER } from '../open-91/config.js' import { FULFILLMENT_EXECUTOR_KEYS } from './executors/types.js' export type VoucherConsumeScheduleResult = { ok: boolean voucherCount: number consumedCount: number alreadyConsumedCount: number failedCount: number voucherCodes: string[] errorMessage: string queuedCount: number jobIds: number[] dueAt: string status: 'queued' | 'already_scheduled' | 'skipped' | 'failed' } export type VoucherConsumeJobRow = { id: number provider: VoucherConsumeProvider channel: VoucherConsumeChannel trigger_event: VoucherConsumeEvent target_key: string platform_order_id: string order_id: number | null order_item_id: number | null task_id: number | null work_order_id: number | null due_at: string status: 'pending' | 'processing' | 'succeeded' | 'failed' | 'cancelled' attempt_count: number max_attempts: number locked_until: string | null last_error: string result_json: string | Record created_at: string updated_at: string finished_at: string | null } type VoucherConsumeScheduleTarget = { provider: VoucherConsumeProvider channel: VoucherConsumeChannel event: VoucherConsumeEvent targetKey: string platformOrderId: string orderId: number | null orderItemId: number | null taskId: number | null workOrderId: number | null voucherCodes: string[] voucherCount: number } type VoucherOrderItemTargetRow = { id: number order_id: number provider: string platform: string platform_order_id: string fulfillment_channel: string } const RETRY_DELAYS_SECONDS = [10, 30, 60, 300, 600, 900, 1800] const PROCESSING_LOCK_SECONDS = 120 export async function scheduleVoucherConsumeForTask( task: TaskRow, event: VoucherConsumeEvent, ): Promise { const order = await getOrderById(task.order_id) if (!order) { return buildScheduleResult('skipped', { errorMessage: '订单不存在,无法安排核销' }) } const provider = resolveVoucherProvider(order) if (!provider) { return buildScheduleResult('skipped') } const orderItem = await getOrderItemById(task.order_item_id) const channel = resolveTaskVoucherChannel(task, orderItem) if (!channel) { return buildScheduleResult('skipped') } return scheduleVoucherConsumeTarget({ provider, channel, event, targetKey: buildVoucherConsumeTargetKey(task.order_item_id, `task:${task.id}`), platformOrderId: String(order.platform_order_id || task.platform_order_id || '').trim(), orderId: Number(order.id), orderItemId: Number(task.order_item_id || 0) || null, taskId: Number(task.id), workOrderId: null, }) } export async function scheduleVoucherConsumeForWorkOrder( workOrder: WorkOrderRow, event: Extract, ): Promise { const order = workOrder.order_id ? await getOrderById(workOrder.order_id) : null if (!order) { return buildScheduleResult('skipped', { errorMessage: '订单不存在,无法安排核销' }) } const provider = resolveVoucherProvider(order) if (!provider) { return buildScheduleResult('skipped') } return scheduleVoucherConsumeTarget({ provider, channel: 'work_order', event, targetKey: buildVoucherConsumeTargetKey(workOrder.order_item_id, `work_order:${workOrder.id}`), platformOrderId: String(order.platform_order_id || workOrder.platform_order_id || '').trim(), orderId: Number(order.id), orderItemId: Number(workOrder.order_item_id || 0) || null, // work_orders.task_id is vestigial (migration 091); never use it as a join key. taskId: null, workOrderId: Number(workOrder.id), }) } /** * Schedule after an order item has a fulfillment channel. * * The platform voucher may not exist yet when this runs. The queued job resolves * the voucher again at execution time and retries while the incoming voucher is * still being created or synchronized. */ export async function scheduleVoucherConsumeForOrderItems( orderItemIds: number[], event: VoucherConsumeEvent, ): Promise<{ targetCount: number; queuedCount: number; jobIds: number[] }> { const ids = [...new Set(orderItemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0))] if (ids.length === 0) { return { targetCount: 0, queuedCount: 0, jobIds: [] } } const rows = await query( ` SELECT oi.id, oi.order_id, oi.fulfillment_channel, o.provider, o.platform, o.platform_order_id FROM order_items oi JOIN orders o ON o.id = oi.order_id WHERE oi.id = ANY($1::bigint[]) AND oi.fulfillment_status = 'assigned' `, [ids], ) const policy = getVoucherConsumePolicy() const targets: Array<{ provider: VoucherConsumeProvider channel: VoucherConsumeChannel trigger_event: VoucherConsumeEvent target_key: string platform_order_id: string order_id: number order_item_id: number due_at: string }> = [] for (const row of rows.rows) { const provider = resolveVoucherProviderFromFields(row.provider, row.platform) const channel = normalizeVoucherChannel(row.fulfillment_channel) if (!provider || !channel) continue const resolved = resolveVoucherConsumePolicy({ provider, channel, event, policy, }) if (!resolved.enabled) continue targets.push({ provider, channel, trigger_event: event, target_key: buildVoucherConsumeTargetKey(row.id, `order_item:${row.id}`), platform_order_id: String(row.platform_order_id || '').trim(), order_id: Number(row.order_id), order_item_id: Number(row.id), due_at: new Date(Date.now() + resolved.delaySeconds * 1000).toISOString(), }) } if (targets.length === 0) { return { targetCount: 0, queuedCount: 0, jobIds: [] } } const inserted = await query<{ id: number }>( ` INSERT INTO voucher_consume_jobs ( provider, channel, trigger_event, target_key, platform_order_id, order_id, order_item_id, due_at, status, attempt_count, max_attempts, updated_at ) SELECT target.provider, target.channel, target.trigger_event, target.target_key, target.platform_order_id, target.order_id, target.order_item_id, target.due_at, 'pending', 0, 8, NOW() FROM jsonb_to_recordset($1::jsonb) AS target( provider text, channel text, trigger_event text, target_key text, platform_order_id text, order_id bigint, order_item_id bigint, due_at timestamptz ) ON CONFLICT (provider, target_key) WHERE status IN ('pending', 'processing', 'succeeded') DO NOTHING RETURNING id `, [JSON.stringify(targets)], ) logIntegration('[voucher-consume/queue]', '订单渠道确定后批量安排核销', { event, targetCount: targets.length, queuedCount: inserted.rowCount || 0, jobIds: inserted.rows.map((row) => Number(row.id)), }) return { targetCount: targets.length, queuedCount: inserted.rowCount || 0, jobIds: inserted.rows.map((row) => Number(row.id)), } } async function scheduleVoucherConsumeTarget( target: Omit, ): Promise { const policy = resolveVoucherConsumePolicy({ provider: target.provider, channel: target.channel, event: target.event, policy: getVoucherConsumePolicy(), }) if (!policy.enabled) { return buildScheduleResult('skipped') } const candidates = await listVoucherCandidatesForTarget(target) const dueAt = new Date(Date.now() + policy.delaySeconds * 1000).toISOString() const inserted = await query( ` INSERT INTO voucher_consume_jobs ( provider, channel, trigger_event, target_key, platform_order_id, order_id, order_item_id, task_id, work_order_id, due_at, status, attempt_count, max_attempts, updated_at ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'pending', 0, 8, NOW() ) ON CONFLICT (provider, target_key) WHERE status IN ('pending', 'processing', 'succeeded') DO NOTHING RETURNING * `, [ target.provider, target.channel, target.event, target.targetKey, target.platformOrderId, target.orderId, target.orderItemId, target.taskId, target.workOrderId, dueAt, ], ) const job = inserted.rows[0] if (!job) { return buildScheduleResult('already_scheduled', { voucherCount: candidates.voucherCount, voucherCodes: candidates.voucherCodes, }) } logIntegration('[voucher-consume/queue]', '核销任务已安排', { jobId: job.id, provider: target.provider, channel: target.channel, event: target.event, dueAt, orderId: target.orderId, orderItemId: target.orderItemId, workOrderId: target.workOrderId, voucherCount: candidates.voucherCount, }) return buildScheduleResult('queued', { queuedCount: 1, jobIds: [Number(job.id)], dueAt, voucherCount: candidates.voucherCount, voucherCodes: candidates.voucherCodes, }) } export async function runVoucherConsumeDispatchJob( jobConfig: Record = {}, ): Promise> { const batchSize = clampInteger(jobConfig.batchSize, 20, 1, 200) const jobs = await claimDueVoucherConsumeJobs(batchSize) let okCount = 0 let failedCount = 0 let retryCount = 0 let cancelledCount = 0 for (const job of jobs) { try { const policy = resolveVoucherConsumePolicy({ provider: job.provider, channel: job.channel, event: job.trigger_event, policy: getVoucherConsumePolicy(), }) if (!policy.enabled) { await finishVoucherConsumeJob(job.id, 'cancelled', { reason: 'policy_disabled_or_changed', }) cancelledCount += 1 continue } const result = await executeVoucherConsumeJob(job) await finishVoucherConsumeJob(job.id, 'succeeded', result) okCount += 1 } catch (error) { const retry = Number(job.attempt_count || 0) < Number(job.max_attempts || 8) await failVoucherConsumeJob(job, error, retry) if (retry) { retryCount += 1 } else { failedCount += 1 } logWarn('[voucher-consume/dispatch]', '核销任务执行失败', { jobId: job.id, provider: job.provider, channel: job.channel, event: job.trigger_event, attemptCount: job.attempt_count, willRetry: retry, error: error instanceof Error ? error.message : String(error), }) } } return { status: failedCount > 0 ? 'warning' : 'ok', message: `核销任务:领取 ${jobs.length},成功 ${okCount},重试 ${retryCount},` + `失败 ${failedCount},取消 ${cancelledCount}`, checkedCount: jobs.length, okCount, failedCount, retryCount, cancelledCount, } } export async function listVoucherConsumeJobs( input: { status?: string page?: number pageSize?: number } = {}, ) { const page = clampInteger(input.page, 1, 1, Number.MAX_SAFE_INTEGER) const pageSize = clampInteger(input.pageSize, 20, 1, 100) const status = String(input.status || '').trim() const params: unknown[] = [] const where: string[] = [] if (status) { params.push(status) where.push(`status = $${params.length}`) } const whereSql = where.length > 0 ? `WHERE ${where.join(' AND ')}` : '' const countResult = await query<{ total: string | number }>( `SELECT COUNT(*) AS total FROM voucher_consume_jobs ${whereSql}`, params, ) const listParams = [...params, pageSize, (page - 1) * pageSize] const result = await query( ` SELECT * FROM voucher_consume_jobs ${whereSql} ORDER BY updated_at DESC, id DESC LIMIT $${params.length + 1} OFFSET $${params.length + 2} `, listParams, ) return { items: result.rows, total: Number(countResult.rows[0]?.total || 0) || 0, page, pageSize, } } export async function retryVoucherConsumeJob(jobId: unknown): Promise { const id = Number(jobId) if (!Number.isInteger(id) || id <= 0) { return null } const result = await query( ` UPDATE voucher_consume_jobs job SET status = 'pending', attempt_count = 0, locked_until = NULL, last_error = '', result_json = '{}'::jsonb, due_at = NOW(), updated_at = NOW(), finished_at = NULL WHERE job.id = $1 AND job.status IN ('failed', 'cancelled') AND NOT EXISTS ( SELECT 1 FROM voucher_consume_jobs active WHERE active.provider = job.provider AND active.target_key = job.target_key AND active.id <> job.id AND active.status IN ('pending', 'processing', 'succeeded') ) RETURNING job.* `, [id], ) return result.rows[0] || null } async function executeVoucherConsumeJob(job: VoucherConsumeJobRow) { if (job.provider === 'kuaishou_industry') { return executeKuaishouConsumeJob(job) } if (job.provider === DOUYIN_MASHIDA_PROVIDER) { return executeDouyinConsumeJob(job) } throw new Error(`不支持的核销平台:${job.provider}`) } async function executeKuaishouConsumeJob(job: VoucherConsumeJobRow) { const vouchers = await resolveKuaishouVouchersForJob(job) if (vouchers.length === 0) { throw new Error('未找到待核销的快手电子凭证') } const task = job.task_id ? await getTaskById(job.task_id) : null let consumedCount = 0 let alreadyConsumedCount = 0 let skippedCount = 0 for (const voucher of vouchers) { const status = normalizeVoucherStatus(voucher.status) if (status === 'CONSUMED') { alreadyConsumedCount += 1 continue } if (status === 'DESTROYED') { skippedCount += 1 continue } if (!isKuaishouIndustryVoucherSendCallbackSuccess(voucher)) { throw new Error(`电子凭证 ${voucher.voucher_code} 尚未完成发码回调`) } const result = await consumeKuaishouIndustryVoucher(voucher, { source: `delayed_${job.trigger_event}`, consumeTime: Date.now(), ...(task ? { task } : {}), }) if (!result.ok || !result.voucher) { throw new Error(result.errorMessage || `电子凭证 ${voucher.voucher_code} 核销失败`) } consumedCount += 1 if (task) { await attachKuaishouIndustryVoucherToTask(task, result.voucher, { source: `delayed_${job.trigger_event}`, }) } } return { provider: job.provider, voucherCount: vouchers.length, consumedCount, alreadyConsumedCount, skippedCount, } } async function executeDouyinConsumeJob(job: VoucherConsumeJobRow) { const certificates = await resolveDouyinCertificatesForJob(job) if (certificates.length === 0) { throw new Error('未找到待核销的抖音券') } let consumedCount = 0 let alreadyConsumedCount = 0 let skippedCount = 0 for (const certificate of certificates) { if (certificate.status === 'verified') { alreadyConsumedCount += 1 continue } if (certificate.status === 'invalid') { skippedCount += 1 continue } await verifyCertificate({ certNo: certificate.cert_no }) consumedCount += 1 } return { provider: job.provider, voucherCount: certificates.length, consumedCount, alreadyConsumedCount, skippedCount, } } async function resolveKuaishouVouchersForJob( job: VoucherConsumeJobRow, ): Promise { const oid = String(job.platform_order_id || '').trim() if (!oid) return [] const vouchers = await listKuaishouIndustryVouchersByOid(oid) if (vouchers.length === 0) return [] if (job.task_id) { const task = await getTaskById(job.task_id) if (task) { return vouchers.filter( (voucher) => Number(voucher.task_id || 0) === Number(task.id) || Number(voucher.unit_index || 0) === Number(task.unit_index || 0), ) } } const orderId = Number(job.order_id || 0) const orderItemId = Number(job.order_item_id || 0) if (!orderId) return vouchers const orderItems = await listOrderItemsByOrderId(orderId) if (orderItems.length <= 1 || !orderItemId) { return vouchers } const tasks = await listTasksByOrderId(orderId) const itemTaskIds = new Set( tasks .filter((task) => Number(task.order_item_id) === orderItemId) .map((task) => Number(task.id)), ) const byTask = vouchers.filter((voucher) => itemTaskIds.has(Number(voucher.task_id || 0))) if (byTask.length > 0) { return byTask } const range = resolveOrderItemUnitRange(orderItems, orderItemId) if (!range) return [] return vouchers.filter((voucher) => { const unitIndex = Number(voucher.unit_index || 0) return unitIndex >= range.start && unitIndex <= range.end }) } async function resolveDouyinCertificatesForJob( job: VoucherConsumeJobRow, ): Promise { const orderId = Number(job.order_id || 0) const orderItemId = Number(job.order_item_id || 0) const certificates = orderId ? await listRedeemCertificatesByLocalOrder(orderId) : [] const filtered = orderItemId ? certificates.filter( (certificate) => !certificate.order_item_id || Number(certificate.order_item_id) === orderItemId, ) : certificates if (filtered.length > 0) { return filtered } const receiptOrderId = String(job.platform_order_id || '').trim() if (!receiptOrderId) return [] const fallback = await findRedeemCertificateByOrder({ provider: DOUYIN_MASHIDA_PROVIDER, orderId: receiptOrderId, }) return fallback ? [fallback] : [] } async function listVoucherCandidatesForTarget( target: Pick< VoucherConsumeScheduleTarget, 'provider' | 'platformOrderId' | 'orderId' | 'orderItemId' | 'taskId' >, ): Promise<{ voucherCount: number; voucherCodes: string[] }> { if (target.provider === 'kuaishou_industry') { const vouchers = target.platformOrderId ? await listKuaishouIndustryVouchersByOid(target.platformOrderId) : [] return { voucherCount: vouchers.length, voucherCodes: vouchers.map((voucher) => String(voucher.voucher_code || '')).filter(Boolean), } } const certificates = await resolveDouyinCertificatesForJob({ ...buildSyntheticJob(target), }) return { voucherCount: certificates.length, voucherCodes: certificates.map((certificate) => certificate.cert_no), } } function buildSyntheticJob( target: Pick< VoucherConsumeScheduleTarget, 'provider' | 'platformOrderId' | 'orderId' | 'orderItemId' | 'taskId' >, ): VoucherConsumeJobRow { return { id: 0, provider: target.provider, channel: 'affiliate_dash', trigger_event: 'link_generated', target_key: '', platform_order_id: target.platformOrderId, order_id: target.orderId, order_item_id: target.orderItemId, task_id: target.taskId, work_order_id: null, due_at: '', status: 'pending', attempt_count: 0, max_attempts: 8, locked_until: null, last_error: '', result_json: {}, created_at: '', updated_at: '', finished_at: null, } } async function claimDueVoucherConsumeJobs(limit: number): Promise { const result = await query( ` WITH due AS ( SELECT id FROM voucher_consume_jobs WHERE attempt_count < max_attempts AND ( (status = 'pending' AND due_at <= NOW()) OR (status = 'processing' AND locked_until < NOW()) ) ORDER BY due_at ASC, id ASC LIMIT $1 FOR UPDATE SKIP LOCKED ) UPDATE voucher_consume_jobs job SET status = 'processing', attempt_count = job.attempt_count + 1, locked_until = NOW() + ($2 * INTERVAL '1 second'), updated_at = NOW() FROM due WHERE job.id = due.id RETURNING job.* `, [limit, PROCESSING_LOCK_SECONDS], ) return result.rows } async function finishVoucherConsumeJob( jobId: number, status: 'succeeded' | 'cancelled', result: Record, ) { await query( ` UPDATE voucher_consume_jobs SET status = $1, locked_until = NULL, last_error = '', result_json = $2::jsonb, updated_at = NOW(), finished_at = NOW() WHERE id = $3 `, [status, JSON.stringify(result), jobId], ) } async function failVoucherConsumeJob(job: VoucherConsumeJobRow, error: unknown, retry: boolean) { const message = error instanceof Error ? error.message : String(error) if (!retry) { await query( ` UPDATE voucher_consume_jobs SET status = 'failed', locked_until = NULL, last_error = $1, result_json = '{}'::jsonb, updated_at = NOW(), finished_at = NOW() WHERE id = $2 `, [message, job.id], ) return } const attemptIndex = Math.max(0, Number(job.attempt_count || 1) - 1) const retryDelaySeconds = RETRY_DELAYS_SECONDS[Math.min(attemptIndex, RETRY_DELAYS_SECONDS.length - 1)] || 60 await query( ` UPDATE voucher_consume_jobs SET status = 'pending', due_at = NOW() + ($1 * INTERVAL '1 second'), locked_until = NULL, last_error = $2, updated_at = NOW() WHERE id = $3 `, [retryDelaySeconds, message, job.id], ) } function resolveVoucherProvider(order: OrderRow): VoucherConsumeProvider | null { return resolveVoucherProviderFromFields(order.provider, order.platform) } function resolveVoucherProviderFromFields( orderProvider: unknown, orderPlatform: unknown, ): VoucherConsumeProvider | null { if (isDouyinMashidaProvider(orderProvider)) { return 'douyin_mashida' } const provider = String(orderProvider || '').trim() const platform = String(orderPlatform || '').trim() if ( [OPEN_91_PROVIDER, 'open_91', '91'].includes(provider) || [OPEN_91_PLATFORM, 'kuaishou'].includes(platform) ) { return 'kuaishou_industry' } return null } function normalizeVoucherChannel(value: unknown): VoucherConsumeChannel | null { const channel = String(value || '').trim() if ( channel === 'affiliate_dash' || channel === 'work_order' || channel === 'manual' || channel === 'third_party_voucher' || channel === 'on_account' ) { return channel } return null } function resolveTaskVoucherChannel( task: Pick, orderItem: OrderItemRow | null, ): VoucherConsumeChannel | null { const executorKey = String(task.executor_key || '').trim() if (executorKey === FULFILLMENT_EXECUTOR_KEYS.AFFILIATE_DASH) return 'affiliate_dash' if ( executorKey === FULFILLMENT_EXECUTOR_KEYS.MANUAL_DISPATCH || executorKey === FULFILLMENT_EXECUTOR_KEYS.KUAISHOU_CLOUD || executorKey === FULFILLMENT_EXECUTOR_KEYS.KUAISHOU_FEIFEI || executorKey === FULFILLMENT_EXECUTOR_KEYS.KUAISHOU_INDUSTRY ) { return 'manual' } return normalizeVoucherChannel(orderItem?.fulfillment_channel) } function buildVoucherConsumeTargetKey(orderItemId: unknown, fallback: string): string { const normalizedItemId = Number(orderItemId || 0) return normalizedItemId > 0 ? `order_item:${normalizedItemId}` : fallback } function resolveOrderItemUnitRange( orderItems: OrderItemRow[], orderItemId: number, ): { start: number; end: number } | null { let cursor = 0 for (const item of orderItems) { const quantity = Math.max(1, Number(item.quantity || 1)) const start = cursor + 1 const end = cursor + quantity if (Number(item.id) === orderItemId) { return { start, end } } cursor = end } return null } function buildScheduleResult( status: VoucherConsumeScheduleResult['status'], input: Partial = {}, ): VoucherConsumeScheduleResult { return { ok: status !== 'failed', voucherCount: Number(input.voucherCount || 0), consumedCount: 0, alreadyConsumedCount: 0, failedCount: 0, voucherCodes: input.voucherCodes || [], errorMessage: String(input.errorMessage || ''), queuedCount: Number(input.queuedCount || 0), jobIds: input.jobIds || [], dueAt: String(input.dueAt || ''), status, } } function clampInteger(value: unknown, fallback: number, min: number, max: number): number { const parsed = Number(value) if (!Number.isInteger(parsed)) return fallback return Math.min(max, Math.max(min, parsed)) }