Files
order_site/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts
T

1230 lines
49 KiB
TypeScript

/**
* 履约路由状态机集成测试(需要真实 PostgreSQL)。
*
* 运行:
* FULFILLMENT_ROUTING_TEST_DATABASE_URL="$DATABASE_URL" \
* npx tsx --test src/services/fulfillment/fulfillment-routing.integration.test.ts
*
* 未设置环境变量时跳过,普通 npm test 不依赖数据库。
*/
import assert from 'node:assert/strict'
import test from 'node:test'
const TEST_DATABASE_URL = String(process.env.FULFILLMENT_ROUTING_TEST_DATABASE_URL || '').trim()
test(
'运营指定渠道:先保证载体;支持安全切换;缺载体保持 pending',
{ skip: !TEST_DATABASE_URL ? '未设置 FULFILLMENT_ROUTING_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const {
assignPendingOrderItemChannels,
assignChannelByInternalSku,
deletePendingChannelGroup,
deleteChannelRules,
ensureVoucherRuleForSku,
forceDeleteChannelRules,
listChannelRules,
listChannelRulesPage,
listPendingChannelGroups,
retryPendingCarrierGroup,
switchChannelForOrderItems,
unlockPendingChannelGroup,
upsertChannelRule,
} = await import('./channel-assign-service.js')
const { deleteManualSkuClassification, saveManualFulfillmentConfig } =
await import('../cost-center/manual-fulfillment-config-service.js')
const { runAffiliateFallback } = await import('./affiliate-fallback-service.js')
const { getFulfillmentAutomationCutoff } =
await import('./fulfillment-automation-config-service.js')
const { reconcilePendingCarriers } = await import('./carrier-reconciliation-service.js')
const { retryPendingWorkOrderRouting } = await import('./pending-channel-repair-service.js')
const {
deleteInternalSkuMapping,
listInternalSkuMappingsPage,
resolveInternalSkuMapping,
upsertInternalSkuMapping,
} = await import('./internal-sku-mapping-service.js')
const { resolveOrderItemForFulfillment } = await import('./product-resolution-service.js')
const { updateOrderItemSourceSnapshot } = await import('../../repositories/order-item-repo.js')
const { listTasks } = await import('../../repositories/task-repo.js')
const { listVoucherConsumeJobs } = await import('./voucher-consume-queue-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const shopId = `it-routing-${suffix}`
const manualSku = `集成测试人工-${suffix}`
const missingAffiliateSku = `集成测试自动-${suffix}`
const voucherSku = `集成测试卡券-${suffix}`
const triggerSku = `集成测试规范-${suffix}`
const fallbackSku = `集成测试兜底-${suffix}`
const autoManualSku = `集成测试自动人工-${suffix}`
const retryWorkOrderSku = `集成测试重试接单-${suffix}`
const manualRemoveSku = `集成测试移除人工-${suffix}`
const protectedRuleSku = `集成测试强制删除-${suffix}`
const missingVoucherSku = `集成测试无商品卡券-${suffix}`
const reactivatedVoucherSku = `集成测试归档卡券-${suffix}`
const retryPendingAffiliateSku = `集成测试失败待重试-${suffix}`
const cutoffTestSku = `集成测试自动化边界-${suffix}`
const retryWorkOrderRuleKey = `it-retry-${suffix}`
const orderIds: number[] = []
const orderItemIds: number[] = []
const crossShopSku = `集成测试跨店-${suffix}`
const internalSkus = [
manualSku,
missingAffiliateSku,
voucherSku,
crossShopSku,
triggerSku,
fallbackSku,
autoManualSku,
retryWorkOrderSku,
manualRemoveSku,
protectedRuleSku,
missingVoucherSku,
reactivatedVoucherSku,
retryPendingAffiliateSku,
cutoffTestSku,
]
const createdVoucherProductIds: number[] = []
let autoManualProductId: number | null = null
let retryWorkOrderRuleId: number | null = null
let fallbackProfileId: number | null = null
async function createPendingOrderItem(
sku: string,
options: { provider?: string; platform?: string; shopId?: string; shopName?: string } = {},
) {
const provider = options.provider || '91kaquan'
const platform = options.platform || 'kuaishou-91'
const targetShopId = options.shopId || shopId
const targetShopName = options.shopName || ''
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 (
$1, $2, $3, $4, $5, 'paid', 'paid', 1000, NOW(), NOW(), NOW()
) RETURNING id`,
[
provider,
platform,
targetShopId,
targetShopName,
`IT-ROUTING-${suffix}-${orderItemIds.length + 1}`,
],
)
const orderId = Number(order.rows[0]!.id)
orderIds.push(orderId)
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, $2, $2, 1, $3::jsonb, 1000, 'order_total', NOW(), NOW())
RETURNING id`,
[orderId, sku, JSON.stringify({ skuNick: sku })],
)
const orderItemId = Number(item.rows[0]!.id)
orderItemIds.push(orderItemId)
return orderItemId
}
async function setRoute(input: {
sku: string
channel: string
priority?: number
role?: 'primary' | 'fallback'
configJson?: Record<string, unknown>
source?: string
}) {
const sku = await query<{ id: string }>('SELECT ensure_canonical_internal_sku($1) AS id', [
input.sku,
])
const internalSkuId = Number(sku.rows[0]!.id)
const role = input.role || 'primary'
if (role === 'primary') {
await query(
`UPDATE sku_channel_routes
SET route_role = 'fallback',
priority = GREATEST(priority, 90),
updated_at = NOW()
WHERE internal_sku_id = $1
AND channel <> $2
AND enabled = TRUE
AND route_role = 'primary'`,
[internalSkuId, input.channel],
)
}
const route = await query<{ id: number }>(
`
INSERT INTO sku_channel_routes (
internal_sku_id, channel, route_role, priority, enabled,
carrier_config_json, source
) VALUES ($1, $2, $3, $4, TRUE, $5::jsonb, $6)
ON CONFLICT (internal_sku_id, channel) DO UPDATE SET
route_role = EXCLUDED.route_role,
priority = EXCLUDED.priority,
enabled = TRUE,
carrier_config_json = EXCLUDED.carrier_config_json,
source = EXCLUDED.source,
updated_at = NOW()
RETURNING id
`,
[
internalSkuId,
input.channel,
role,
input.priority ?? (role === 'primary' ? 10 : 90),
JSON.stringify(input.configJson || {}),
input.source || 'integration_test',
],
)
return Number(route.rows[0]!.id)
}
try {
const canonicalResolution = await query<{
internal_sku: string | null
platform_only: string | null
cjk_extension: string | null
case_variant: string | null
}>(
`SELECT
resolve_internal_sku('platform-code', ' 套装-浪漫 天命 ', 'platform-code') AS internal_sku,
resolve_internal_sku('platform-code', 'Platform Title', 'platform-code') AS platform_only,
resolve_internal_sku('', '𠀀', '') AS cjk_extension,
resolve_internal_sku('', '黑色ROCK长袖T恤', '') AS case_variant`,
)
assert.equal(canonicalResolution.rows[0]?.internal_sku, '套装-浪漫天命')
assert.equal(canonicalResolution.rows[0]?.platform_only, null)
assert.equal(canonicalResolution.rows[0]?.cjk_extension, '𠀀')
assert.equal(canonicalResolution.rows[0]?.case_variant, '黑色rock长袖t恤')
assert.ok(Number.isFinite(Date.parse(await getFulfillmentAutomationCutoff())))
const cutoffNewItemId = await createPendingOrderItem(cutoffTestSku)
const cutoffOldItemId = await createPendingOrderItem(cutoffTestSku)
await query(
`UPDATE orders
SET created_at = '2019-01-01T00:00:00+08:00',
updated_at = '2019-01-01T00:00:00+08:00'
WHERE id = (SELECT order_id FROM v_order_item_current WHERE id = $1)`,
[cutoffOldItemId],
)
await setRoute({ sku: cutoffTestSku, channel: 'third_party_voucher' })
const cutoffAssignment = await assignPendingOrderItemChannels({
orderItemIds: [cutoffOldItemId, cutoffNewItemId],
batchSize: 10,
maxBatches: 1,
cutoffCreatedAt: '2020-01-01T00:00:00+08:00',
})
assert.equal(cutoffAssignment.scanned, 1)
assert.equal(cutoffAssignment.assigned, 1)
const cutoffStates = await query<{ id: number; fulfillment_status: string }>(
`SELECT id, fulfillment_status
FROM v_order_item_current
WHERE id = ANY($1::bigint[])
ORDER BY id DESC`,
[[cutoffOldItemId, cutoffNewItemId]],
)
assert.equal(cutoffStates.rows[0]?.fulfillment_status, 'pending')
assert.equal(cutoffStates.rows[1]?.fulfillment_status, 'assigned')
await setRoute({
sku: ` 集成测试 规范-${suffix} `,
channel: 'manual',
role: 'fallback',
})
const normalizedInsert = await query<{ internal_sku: string }>(
`SELECT sku.internal_sku_key AS internal_sku
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.source = 'integration_test'
AND sku.internal_sku_key = canonicalize_internal_sku($1)
ORDER BY route.id DESC
LIMIT 1`,
[triggerSku],
)
assert.equal(normalizedInsert.rows[0]?.internal_sku, triggerSku)
const mappingValue = `MAP-${suffix}`
const mappingSku = `集成映射-${suffix}`
const mappingId = await upsertInternalSkuMapping({
provider: '91kaquan',
shopId,
matchType: 'sku_code',
matchValue: ` ${mappingValue} `,
internalSku: mappingSku,
priority: 10,
source: 'integration_test',
note: '集成测试映射',
})
const resolvedMapping = await resolveInternalSkuMapping({
provider: '91kaquan',
shopId,
candidates: [{ matchType: 'sku_code', value: mappingValue }],
})
assert.equal(resolvedMapping?.internalSku, mappingSku)
const resolvedLongTitleItem = await resolveOrderItemForFulfillment({
provider: '91kaquan',
shopId,
item: {
externalItemId: `mapping-item-${suffix}`,
externalSkuCode: mappingValue,
externalSkuName: '平台长中文商品标题(不应成为内部 SKU)',
skuCode: mappingValue,
skuName: '平台长中文商品标题(不应成为内部 SKU)',
quantity: 1,
},
})
assert.equal(resolvedLongTitleItem.snapshot.internalSku, mappingSku)
assert.equal(resolvedLongTitleItem.skuName, mappingSku)
const mappingPage = await listInternalSkuMappingsPage({
keyword: mappingSku,
page: 1,
pageSize: 10,
})
assert.equal(mappingPage.total, 1)
assert.equal(await deleteInternalSkuMapping(mappingId), true)
const manualItemId = await createPendingOrderItem(manualSku)
await query(
`INSERT INTO manual_fulfillment_configs (
sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at
) VALUES ($1, 'nanhang', 0, 'variable', TRUE, '集成测试', NOW(), NOW())`,
[manualSku],
)
const manualAssigned = await assignChannelByInternalSku({
internalSku: manualSku,
channel: 'manual',
adminUserId: null,
note: '集成测试人工',
})
assert.equal(manualAssigned.updated, 1)
assert.equal(manualAssigned.carrierFailed, 0)
const manualItem = await query<{
fulfillment_status: string
fulfillment_channel: string
fulfillment_config_id: number | null
}>(
`SELECT fulfillment_status, fulfillment_channel, fulfillment_config_id
FROM v_order_item_current WHERE id = $1`,
[manualItemId],
)
assert.equal(manualItem.rows[0]?.fulfillment_status, 'assigned')
assert.equal(manualItem.rows[0]?.fulfillment_channel, 'manual')
assert.ok(Number(manualItem.rows[0]?.fulfillment_config_id) > 0)
const sourceSyncedItem = await updateOrderItemSourceSnapshot(manualItemId, {
skuName: `平台覆盖-${suffix}`,
specJson: JSON.stringify({ skuNick: `平台覆盖-${suffix}` }),
itemSnapshotJson: JSON.stringify({ externalSkuName: `平台覆盖-${suffix}` }),
updatedAt: new Date().toISOString(),
})
assert.equal(sourceSyncedItem?.sku_name, manualSku)
assert.equal(sourceSyncedItem?.spec_json?.skuNick, manualSku)
const switched = await switchChannelForOrderItems({
orderItemIds: [manualItemId],
channel: 'third_party_voucher',
adminUserId: null,
note: '集成测试切换',
})
assert.equal(switched.switched, 1)
assert.equal(switched.failed, 0)
const switchedItem = await query<{
fulfillment_status: string
fulfillment_channel: string
original_channel: string | null
canonical_product_id: number | null
canonical_sku: string | null
item_snapshot_json: Record<string, unknown>
}>(
`SELECT oi.fulfillment_status, oi.fulfillment_channel, oi.original_channel,
oi.canonical_product_id, product.canonical_sku, oi.item_snapshot_json
FROM v_order_item_current oi
LEFT JOIN product_catalog product ON product.id = oi.canonical_product_id
WHERE oi.id = $1`,
[manualItemId],
)
assert.equal(switchedItem.rows[0]?.fulfillment_status, 'assigned')
assert.equal(switchedItem.rows[0]?.fulfillment_channel, 'third_party_voucher')
assert.equal(switchedItem.rows[0]?.original_channel, 'manual')
assert.ok(switchedItem.rows[0]?.canonical_sku?.startsWith('tpv:'))
assert.equal(
switchedItem.rows[0]?.item_snapshot_json.internalSkuChannel,
'third_party_voucher',
)
assert.equal(switchedItem.rows[0]?.item_snapshot_json.isConfigured, false)
const missingAffiliateItemId = await createPendingOrderItem(missingAffiliateSku)
const affiliateFailed = await assignChannelByInternalSku({
internalSku: missingAffiliateSku,
channel: 'affiliate_dash',
adminUserId: null,
note: '集成测试缺映射',
})
assert.equal(affiliateFailed.updated, 0)
assert.equal(affiliateFailed.carrierFailed, 1)
const pendingItem = await query<{
fulfillment_status: string
fulfillment_channel: string | null
fulfillment_pending_reason: string | null
}>(
`SELECT fulfillment_status, fulfillment_channel, fulfillment_pending_reason
FROM v_order_item_current WHERE id = $1`,
[missingAffiliateItemId],
)
assert.equal(pendingItem.rows[0]?.fulfillment_status, 'pending')
assert.equal(pendingItem.rows[0]?.fulfillment_channel, null)
assert.equal(pendingItem.rows[0]?.fulfillment_pending_reason, 'no_carrier')
const autoManualItemId = await createPendingOrderItem(autoManualSku)
const missingManual = await assignChannelByInternalSku({
internalSku: autoManualSku,
channel: 'manual',
adminUserId: null,
note: '集成测试缺人工配置',
})
assert.equal(missingManual.updated, 0)
assert.equal(missingManual.carrierFailed, 1)
const savedManual = await saveManualFulfillmentConfig({
skuKey: autoManualSku,
platformKey: 'fund_cash',
creditQuantity: 69,
note: '集成测试一次配置',
})
autoManualProductId = Number(savedManual.canonicalProductId || 0) || null
assert.ok(autoManualProductId)
assert.ok(savedManual.syncSummary.pendingAssigned >= 1)
const autoManualItem = await query<{
fulfillment_status: string
fulfillment_channel: string
fulfillment_config_id: number | null
canonical_product_id: number | null
}>(
`SELECT fulfillment_status, fulfillment_channel, fulfillment_config_id, canonical_product_id
FROM v_order_item_current WHERE id = $1`,
[autoManualItemId],
)
assert.equal(autoManualItem.rows[0]?.fulfillment_status, 'assigned')
assert.equal(autoManualItem.rows[0]?.fulfillment_channel, 'manual')
assert.ok(Number(autoManualItem.rows[0]?.fulfillment_config_id) > 0)
assert.equal(Number(autoManualItem.rows[0]?.canonical_product_id), autoManualProductId)
const retryWorkOrderItemId = await createPendingOrderItem(retryWorkOrderSku, {
provider: 'douyin_mashida',
platform: 'doudian',
shopId: '305760014',
})
const retryRule = await query<{ id: number }>(
`INSERT INTO work_product_rules (
rule_key, provider, platform, shop_id, sku_code, product_name, match_type,
category_id, enabled, auto_create, reward_amount, requirement_json, sort_order,
created_at, updated_at, pricing_mode
) VALUES (
$1, '', '', '', '', $2, 'exact', NULL, TRUE, TRUE, 1000, '{}'::jsonb, 1,
NOW(), NOW(), 'fixed'
) RETURNING id`,
[retryWorkOrderRuleKey, retryWorkOrderSku],
)
retryWorkOrderRuleId = Number(retryRule.rows[0]?.id || 0)
await query(
`INSERT INTO work_product_rule_mappings (
rule_id, seller_ids_json, rel_item_id, item_title, rel_sku_id, sku_nick,
mapping_type, enabled, created_at, updated_at, sku_match_mode
) VALUES (
$1, '["305760014"]'::jsonb, '', '', '', $2,
'sku_exact', TRUE, NOW(), NOW(), 'exact'
)`,
[retryWorkOrderRuleId, retryWorkOrderSku],
)
await setRoute({
sku: retryWorkOrderSku,
channel: 'work_order',
configJson: { workRuleKey: retryWorkOrderRuleKey },
})
const retriedWorkOrders = await retryPendingWorkOrderRouting({
internalSku: retryWorkOrderSku,
})
assert.equal(retriedWorkOrders.scanned, 1)
assert.equal(retriedWorkOrders.createdWorkOrders, 1)
assert.equal(retriedWorkOrders.assigned, 1)
assert.equal(retriedWorkOrders.remaining, 0)
const retriedItem = await query<{
fulfillment_status: string
fulfillment_channel: string
work_rule_key: string | null
}>(
`SELECT oi.fulfillment_status, oi.fulfillment_channel, wpr.rule_key AS work_rule_key
FROM v_order_item_current oi
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
WHERE oi.id = $1`,
[retryWorkOrderItemId],
)
assert.equal(retriedItem.rows[0]?.fulfillment_status, 'assigned')
assert.equal(retriedItem.rows[0]?.fulfillment_channel, 'work_order')
assert.equal(retriedItem.rows[0]?.work_rule_key, retryWorkOrderRuleKey)
await query(
`UPDATE work_orders SET status = 'cancelled', updated_at = NOW()
WHERE order_item_id = $1`,
[retryWorkOrderItemId],
)
await query(
`UPDATE fulfillments
SET status = 'pending', channel = NULL,
source = 'auto_mapping', pending_reason = 'no_carrier',
pending_detail = $2::jsonb, updated_at = NOW()
WHERE order_item_id = $1`,
[
retryWorkOrderItemId,
JSON.stringify({ carrierIssue: 'carrier_task_closed', channel: 'work_order' }),
],
)
const retriedCarrier = await retryPendingCarrierGroup({
internalSku: retryWorkOrderSku,
retryInactiveCarrier: true,
})
assert.equal(retriedCarrier.scanned, 1)
assert.equal(retriedCarrier.assigned, 1)
assert.equal(retriedCarrier.remaining, 0)
const activeWorkOrders = await query<{ count: number }>(
`SELECT count(*)::int AS count FROM work_orders
WHERE order_item_id = $1 AND status <> 'cancelled'`,
[retryWorkOrderItemId],
)
assert.equal(activeWorkOrders.rows[0]?.count, 1)
const crossShopA = `it-routing-a-${suffix}`
const crossShopB = `it-routing-b-${suffix}`
await createPendingOrderItem(crossShopSku, {
shopId: crossShopA,
shopName: '测试店铺 A',
})
await createPendingOrderItem(crossShopSku, {
provider: 'pdd',
platform: 'pdd',
shopId: crossShopB,
shopName: '测试店铺 B',
})
const pendingGroups = await listPendingChannelGroups({
keyword: crossShopSku,
page: 1,
pageSize: 10,
})
assert.equal(pendingGroups.total, 1)
assert.equal(pendingGroups.items[0]?.internal_sku, crossShopSku)
assert.equal(pendingGroups.items[0]?.item_count, 2)
assert.deepEqual(pendingGroups.items[0]?.providers.sort(), ['91kaquan', 'pdd'])
assert.deepEqual(pendingGroups.items[0]?.shop_ids.sort(), [crossShopA, crossShopB])
assert.deepEqual(
pendingGroups.items[0]?.shops.sort((left, right) => left.id.localeCompare(right.id)),
[
{ id: crossShopA, name: '测试店铺 A' },
{ id: crossShopB, name: '测试店铺 B' },
],
)
const pendingClassification = await listPendingChannelGroups({
keyword: crossShopSku,
issueCategory: 'classification',
page: 1,
pageSize: 10,
})
assert.equal(pendingClassification.total, 1)
assert.equal(
pendingClassification.items[0]?.issue_breakdown[0]?.issue_code,
'missing_channel',
)
const pendingAfterSales = await listPendingChannelGroups({
keyword: crossShopSku,
issueCategory: 'after_sales',
page: 1,
pageSize: 10,
})
assert.equal(pendingAfterSales.total, 0)
const deletedPending = await deletePendingChannelGroup({
internalSku: crossShopSku,
adminUserId: null,
})
assert.equal(deletedPending.archivedItems, 2)
assert.equal(deletedPending.preservedOrders, 2)
assert.equal(
(
await listPendingChannelGroups({
keyword: crossShopSku,
page: 1,
pageSize: 10,
})
).total,
0,
)
const preservedOrders = await query<{ count: number }>(
`SELECT COUNT(*)::int AS count FROM orders WHERE id = ANY($1::bigint[])`,
[orderIds],
)
assert.ok(preservedOrders.rows[0].count >= 2)
const manualRemoveItemId = await createPendingOrderItem(manualRemoveSku, {
shopId: crossShopA,
shopName: '测试店铺 A',
})
await query(
`UPDATE fulfillments
SET status = 'assigned', channel = 'manual',
source = 'carrier_evidence', fulfillment_config_id = NULL,
updated_at = NOW()
WHERE order_item_id = $1`,
[manualRemoveItemId],
)
await query(
`INSERT INTO manual_sku_registry (
sku_key, display_name, shop_count, order_count, item_count, revenue_fen,
first_seen_date, last_seen_date, scan_from_date, scan_to_date, created_at, updated_at
) VALUES ($1, $1, 1, 1, 1, 1000, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, NOW(), NOW())`,
[manualRemoveSku],
)
await setRoute({
sku: manualRemoveSku,
channel: 'manual',
role: 'fallback',
source: 'manual_config',
})
const deletedManual = await deleteManualSkuClassification(manualRemoveSku)
assert.equal(deletedManual.affectedItems, 1)
assert.equal(deletedManual.deletedRegistry, 1)
assert.equal(deletedManual.deletedRules, 1)
const removedManualItem = await query<{
fulfillment_status: string
fulfillment_channel: string | null
}>(`SELECT fulfillment_status, fulfillment_channel FROM v_order_item_current WHERE id = $1`, [
manualRemoveItemId,
])
assert.equal(removedManualItem.rows[0]?.fulfillment_status, 'pending')
assert.equal(removedManualItem.rows[0]?.fulfillment_channel, null)
const protectedRuleId = await setRoute({
sku: protectedRuleSku,
channel: 'manual',
priority: 70,
source: 'manual_config',
})
await assert.rejects(() => deleteChannelRules([protectedRuleId]), /不能在渠道页直接删除/)
const forcedDeleted = await forceDeleteChannelRules([protectedRuleId])
assert.equal(forcedDeleted.deletedCount, 1)
assert.deepEqual(forcedDeleted.deletedIds, [protectedRuleId])
const fallbackItemId = await createPendingOrderItem(fallbackSku)
await setRoute({ sku: fallbackSku, channel: 'affiliate_dash' })
await setRoute({ sku: fallbackSku, channel: 'manual', role: 'fallback' })
const fallbackItem = await query<{ order_id: number }>(
`UPDATE fulfillments
SET status = 'assigned', channel = 'affiliate_dash',
source = 'operator_assigned', locked = FALSE,
updated_at = NOW()
WHERE order_item_id = $1
RETURNING order_id`,
[fallbackItemId],
)
const fallbackOrderId = Number(fallbackItem.rows[0]!.order_id)
const fallbackProfile = await query<{ id: number }>(
`INSERT INTO fulfillment_profiles (
profile_key, name, executor_key, requires_claim, auto_dispatch,
inventory_strategy, config_json, created_at, updated_at
) VALUES ($1, '集成测试自动发货', 'affiliate_dash', TRUE, FALSE,
'static_pool', '{}'::jsonb, NOW(), NOW())
RETURNING id`,
[`it-affiliate-${suffix}`],
)
fallbackProfileId = Number(fallbackProfile.rows[0]!.id)
const fallbackOrder = await query<{
provider: string
platform: string
shop_id: string
shop_name: string
platform_order_id: string
}>(
`SELECT provider, platform, shop_id, shop_name, platform_order_id
FROM orders WHERE id = $1`,
[fallbackOrderId],
)
const orderRow = fallbackOrder.rows[0]!
await query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, created_at, updated_at
) VALUES ($1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash', 'failed', NOW(), NOW())`,
[
fallbackOrderId,
fallbackItemId,
`IT-AFF-FAIL-${suffix}`,
orderRow.provider,
orderRow.platform,
orderRow.shop_id,
orderRow.shop_name,
orderRow.platform_order_id,
fallbackProfileId,
],
)
const fallbackWithoutConfig = await runAffiliateFallback({ limit: 50 })
assert.equal(fallbackWithoutConfig.switched, 0)
assert.equal(
(
await query<{ fulfillment_channel: string }>(
`SELECT fulfillment_channel FROM v_order_item_current WHERE id = $1`,
[fallbackItemId],
)
).rows[0]?.fulfillment_channel,
'affiliate_dash',
)
const manualFallbackConfig = await query<{ id: number }>(
`INSERT INTO manual_fulfillment_configs (
sku_key, platform_key, credit_quantity, cost_mode, enabled, created_at, updated_at
) VALUES ($1, 'nanhang', 0, 'variable', TRUE, NOW(), NOW())
RETURNING id`,
[fallbackSku],
)
const fallbackWithConfig = await runAffiliateFallback({ limit: 50 })
assert.equal(fallbackWithConfig.switched, 1)
const fallbackAssigned = await query<{
fulfillment_channel: string
fulfillment_config_id: number | null
item_snapshot_json: Record<string, unknown>
}>(
`SELECT fulfillment_channel, fulfillment_config_id, item_snapshot_json
FROM v_order_item_current WHERE id = $1`,
[fallbackItemId],
)
assert.equal(fallbackAssigned.rows[0]?.fulfillment_channel, 'manual')
assert.equal(fallbackAssigned.rows[0]?.item_snapshot_json.internalSkuChannel, 'manual')
assert.equal(fallbackAssigned.rows[0]?.item_snapshot_json.isConfigured, true)
assert.equal(
Number(fallbackAssigned.rows[0]?.fulfillment_config_id),
Number(manualFallbackConfig.rows[0]!.id),
)
const retryTaskItemId = await createPendingOrderItem(autoManualSku)
const retryTaskOrder = await query<{
order_id: number
provider: string
platform: string
shop_id: string
shop_name: string
platform_order_id: string
}>(
`SELECT oi.order_id, o.provider, o.platform, o.shop_id, o.shop_name, o.platform_order_id
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = $1`,
[retryTaskItemId],
)
const retryTaskOrderRow = retryTaskOrder.rows[0]!
const insertRetryTask = (
taskNo: string,
status: string,
attemptNo: number,
isCurrent: boolean,
) =>
query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, attempt_no, is_current, created_at, updated_at
) VALUES (
$1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash',
$10, $11, $12, NOW(), NOW()
)`,
[
retryTaskOrderRow.order_id,
retryTaskItemId,
taskNo,
retryTaskOrderRow.provider,
retryTaskOrderRow.platform,
retryTaskOrderRow.shop_id,
retryTaskOrderRow.shop_name,
retryTaskOrderRow.platform_order_id,
fallbackProfileId,
status,
attemptNo,
isCurrent,
],
)
await insertRetryTask(`IT-AFF-CLOSED-${suffix}`, 'closed', 1, false)
await insertRetryTask(`IT-AFF-ACTIVE-${suffix}`, 'paid', 2, true)
await assert.rejects(
() => insertRetryTask(`IT-AFF-DUPLICATE-${suffix}`, 'claimed', 3, false),
/duplicate key value violates unique constraint/,
)
const retryTaskRows = await query<{ task_status: string }>(
`SELECT task_status FROM fulfillment_tasks WHERE order_item_id = $1 ORDER BY id`,
[retryTaskItemId],
)
assert.deepEqual(
retryTaskRows.rows.map((row) => row.task_status),
['closed', 'paid'],
)
const currentTasksOnly = await listTasks({
platformOrderId: retryTaskOrderRow.platform_order_id,
currentOnly: true,
page: 1,
pageSize: 20,
})
assert.equal(currentTasksOnly.total, 1)
assert.equal(currentTasksOnly.items[0]?.task_status, 'paid')
const allTaskAttempts = await listTasks({
platformOrderId: retryTaskOrderRow.platform_order_id,
currentOnly: false,
page: 1,
pageSize: 20,
})
assert.equal(allTaskAttempts.total, 2)
assert.equal(await ensureVoucherRuleForSku(missingVoucherSku), false)
const missingVoucherRules = await query<{ count: number }>(
`SELECT count(*)::int AS count
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE sku.internal_sku_key = $1 AND route.channel = 'third_party_voucher'`,
[missingVoucherSku],
)
assert.equal(missingVoucherRules.rows[0]?.count, 0)
const voucherProduct = await query<{ id: number }>(
`INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
) VALUES ($1, $2, '卡券', '张', 'active', '{}'::jsonb, NOW(), NOW())
RETURNING id`,
[`tpv:${voucherSku}`, `卡券展示名-${suffix}`],
)
createdVoucherProductIds.push(Number(voucherProduct.rows[0]!.id))
assert.equal(await ensureVoucherRuleForSku(voucherSku), true)
const voucherChannel = await query<{
channel: string
source: string
canonical_product_id: string | null
priority: number
is_fallback: boolean
config_json: Record<string, unknown>
}>(
`SELECT route.channel,
route.source,
route.canonical_product_id,
route.priority,
route.route_role = 'fallback' AS is_fallback,
route.carrier_config_json AS config_json
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE sku.internal_sku_key = $1
ORDER BY route.channel`,
[voucherSku],
)
assert.equal(voucherChannel.rows.length, 1)
assert.equal(voucherChannel.rows[0]?.channel, 'third_party_voucher')
assert.equal(voucherChannel.rows[0]?.source, 'voucher')
assert.equal(
Number(voucherChannel.rows[0]?.canonical_product_id),
createdVoucherProductIds[0],
)
assert.equal(voucherChannel.rows[0]?.priority, 10)
assert.equal(voucherChannel.rows[0]?.is_fallback, false)
assert.equal(voucherChannel.rows[0]?.config_json.voucherSku, voucherSku)
const legacyVoucherRules = await query<{ count: number }>(
`SELECT count(*)::int AS count FROM sku_channel_rules
WHERE match_key = $1`,
[voucherSku],
)
assert.equal(legacyVoucherRules.rows[0]?.count, 0)
const savedRule = await upsertChannelRule({
matchKeyType: 'internal_sku',
matchKey: voucherSku,
channel: 'third_party_voucher',
priority: 10,
enabled: true,
note: '集成测试统一渠道',
adminUserId: null,
})
const listedRules = await listChannelRules({ channel: 'third_party_voucher', enabled: true })
assert.ok(
listedRules.some((rule) => rule.id === savedRule.id && rule.match_key === voucherSku),
)
const firstRulePage = await listChannelRulesPage({
channel: 'third_party_voucher',
enabled: true,
page: 1,
pageSize: 2,
})
assert.equal(firstRulePage.page, 1)
assert.equal(firstRulePage.pageSize, 2)
assert.ok(firstRulePage.items.length <= 2)
assert.equal(firstRulePage.total, listedRules.length)
const searchedRules = await listChannelRulesPage({
keyword: voucherSku,
source: 'operator',
enabled: true,
page: 1,
pageSize: 10,
})
assert.ok(searchedRules.items.some((rule) => rule.id === savedRule.id))
assert.ok(searchedRules.items.every((rule) => rule.source === 'operator'))
const deletedRules = await deleteChannelRules([savedRule.id])
assert.equal(deletedRules.deletedCount, 1)
assert.deepEqual(deletedRules.deletedIds, [savedRule.id])
const archivedVoucherProduct = await query<{ id: number }>(
`INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
) VALUES (
$1, $2, '卡券', '张', 'archived',
'{"deletedAt":"2026-01-01T00:00:00.000Z","deletedBy":"integration_test"}'::jsonb,
NOW(), NOW()
)
RETURNING id`,
[`tpv:${reactivatedVoucherSku}`, reactivatedVoucherSku],
)
const archivedVoucherProductId = Number(archivedVoucherProduct.rows[0]!.id)
createdVoucherProductIds.push(archivedVoucherProductId)
const reactivatedRule = await upsertChannelRule({
matchKeyType: 'internal_sku',
matchKey: reactivatedVoucherSku,
channel: 'third_party_voucher',
priority: 80,
enabled: true,
note: '集成测试重新激活归档商品',
adminUserId: null,
})
const reactivated = await query<{
status: string
attributes_json: Record<string, unknown>
canonical_product_id: string | null
priority: number
is_fallback: boolean
}>(
`SELECT product.status, product.attributes_json,
route.canonical_product_id,
route.priority,
route.route_role = 'fallback' AS is_fallback
FROM sku_channel_routes route
JOIN product_catalog product ON product.id = route.canonical_product_id
WHERE route.id = $1`,
[reactivatedRule.id],
)
assert.equal(reactivated.rows[0]?.status, 'active')
assert.equal(Number(reactivated.rows[0]?.canonical_product_id), archivedVoucherProductId)
assert.equal(reactivated.rows[0]?.priority, 10)
assert.equal(reactivated.rows[0]?.is_fallback, false)
assert.equal(reactivated.rows[0]?.attributes_json.deletedAt, undefined)
assert.equal(typeof reactivated.rows[0]?.attributes_json.reactivatedAt, 'string')
const retryPendingItemId = await createPendingOrderItem(retryPendingAffiliateSku)
await setRoute({
sku: retryPendingAffiliateSku,
channel: 'affiliate_dash',
configJson: { affiliateSku: retryPendingAffiliateSku },
})
await query(
`UPDATE fulfillments
SET status = 'assigned',
channel = 'affiliate_dash',
source = 'carrier_evidence',
updated_at = NOW()
WHERE order_item_id = $1`,
[retryPendingItemId],
)
const existingRetryTasks = await query<{ id: number; task_status: string }>(
`SELECT id, task_status
FROM fulfillment_tasks
WHERE order_item_id = $1
ORDER BY id`,
[retryPendingItemId],
)
assert.deepEqual(existingRetryTasks.rows, [], `测试订单项已有任务:${retryPendingItemId}`)
await query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, created_at, updated_at
)
SELECT oi.order_id, oi.id, 1, $2, o.provider, o.platform,
o.shop_id, o.shop_name, o.platform_order_id, profile.id, 'affiliate_dash',
'retry_pending', NOW(), NOW()
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN fulfillment_profiles profile
ON profile.id = $3
AND profile.executor_key = 'affiliate_dash'
WHERE oi.id = $1`,
[retryPendingItemId, `IT-RETRY-PENDING-${suffix}`, fallbackProfileId],
)
const retryPendingTask = await query<{ task_status: string }>(
`SELECT task_status
FROM fulfillment_tasks
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'
ORDER BY id DESC
LIMIT 1`,
[retryPendingItemId],
)
assert.equal(retryPendingTask.rows[0]?.task_status, 'retry_pending')
const blockedRetry = await reconcilePendingCarriers({
orderItemIds: [retryPendingItemId],
includeAssignedMissingCarrier: true,
limit: 1,
})
assert.equal(blockedRetry.scanned, 0)
assert.equal(blockedRetry.affiliateTasksCreated, 0)
const retryPendingState = await query<{
fulfillment_status: string
fulfillment_channel: string | null
fulfillment_pending_reason: string | null
fulfillment_pending_detail: Record<string, unknown>
task_count: number
}>(
`SELECT
oi.fulfillment_status,
oi.fulfillment_channel,
oi.fulfillment_pending_reason,
oi.fulfillment_pending_detail,
(SELECT count(*)::int FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) AS task_count
FROM v_order_item_current oi
WHERE oi.id = $1`,
[retryPendingItemId],
)
assert.equal(retryPendingState.rows[0]?.fulfillment_status, 'assigned')
assert.equal(retryPendingState.rows[0]?.fulfillment_channel, 'affiliate_dash')
assert.equal(retryPendingState.rows[0]?.fulfillment_pending_reason, null)
assert.equal(retryPendingState.rows[0]?.task_count, 1)
await query(
`UPDATE fulfillments
SET status = 'pending',
channel = NULL,
pending_reason = 'no_carrier',
pending_detail = $2::jsonb,
locked = TRUE,
updated_at = NOW()
WHERE order_item_id = $1`,
[
retryPendingItemId,
JSON.stringify({
internalSku: retryPendingAffiliateSku,
channel: 'affiliate_dash',
carrierIssue: 'carrier_task_retry_pending',
}),
],
)
const unlockedPendingGroup = await unlockPendingChannelGroup({
internalSku: retryPendingAffiliateSku,
})
assert.equal(unlockedPendingGroup.updated, 1)
const unlockedPendingItem = await query<{ fulfillment_locked: boolean }>(
`SELECT fulfillment_locked FROM v_order_item_current WHERE id = $1`,
[retryPendingItemId],
)
assert.equal(unlockedPendingItem.rows[0]?.fulfillment_locked, false)
await query(
`UPDATE fulfillment_tasks
SET task_status = 'closed',
result_code = 'USER_APPLY_REFUND',
updated_at = NOW()
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'`,
[retryPendingItemId],
)
await query(
`UPDATE fulfillments
SET status = 'pending',
channel = NULL,
source = 'auto_mapping',
pending_reason = NULL,
pending_detail = NULL,
updated_at = NOW()
WHERE order_item_id = $1`,
[retryPendingItemId],
)
await assert.rejects(
() =>
retryPendingCarrierGroup({
internalSku: retryPendingAffiliateSku,
retryInactiveCarrier: true,
issueCodes: ['user_refund'],
}),
/不允许重试/,
)
const protectedRetryState = await query<{
fulfillment_status: string
fulfillment_pending_detail: Record<string, unknown> | null
task_count: number
}>(
`SELECT
oi.fulfillment_status,
oi.fulfillment_pending_detail,
(SELECT count(*)::int FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) AS task_count
FROM v_order_item_current oi
WHERE oi.id = $1`,
[retryPendingItemId],
)
assert.equal(protectedRetryState.rows[0]?.fulfillment_status, 'pending')
assert.equal(protectedRetryState.rows[0]?.fulfillment_pending_detail, null)
assert.equal(protectedRetryState.rows[0]?.task_count, 1)
const voucherJobPrefix = `IT-VOUCHER-LIST-${suffix}`
const voucherJobs = await query<{ id: number }>(
`INSERT INTO voucher_consume_jobs (
provider, channel, trigger_event, target_key, platform_order_id,
due_at, status, attempt_count, max_attempts, updated_at
) VALUES
('kuaishou_industry', 'affiliate_dash', 'order_received', $1, $2, NOW(), 'succeeded', 1, 8, NOW()),
('kuaishou_industry', 'manual', 'order_received', $3, $4, NOW(), 'failed', 2, 8, NOW()),
('douyin_mashida', 'manual', 'order_received', $5, $6, NOW(), 'failed', 1, 8, NOW())
RETURNING id`,
[
`${voucherJobPrefix}-KS-AFF`,
`${voucherJobPrefix}-KS-AFF-ORDER`,
`${voucherJobPrefix}-KS-MANUAL`,
`${voucherJobPrefix}-KS-MANUAL-ORDER`,
`${voucherJobPrefix}-DY-MANUAL`,
`${voucherJobPrefix}-DY-MANUAL-ORDER`,
],
)
const firstVoucherJobPage = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
page: 1,
pageSize: 2,
})
assert.equal(firstVoucherJobPage.total, 3)
assert.equal(firstVoucherJobPage.items.length, 2)
assert.equal(firstVoucherJobPage.page, 1)
assert.equal(firstVoucherJobPage.pageSize, 2)
const secondVoucherJobPage = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
page: 2,
pageSize: 2,
})
assert.equal(secondVoucherJobPage.total, 3)
assert.equal(secondVoucherJobPage.items.length, 1)
const filteredVoucherJobs = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
provider: 'kuaishou_industry',
channel: 'manual',
status: 'failed',
page: 1,
pageSize: 20,
})
assert.equal(filteredVoucherJobs.total, 1)
assert.equal(filteredVoucherJobs.items[0]?.provider, 'kuaishou_industry')
assert.equal(filteredVoucherJobs.items[0]?.channel, 'manual')
assert.equal(filteredVoucherJobs.items[0]?.status, 'failed')
const searchedVoucherJob = await listVoucherConsumeJobs({
keyword: String(voucherJobs.rows[1]?.id || ''),
status: 'failed',
page: 1,
pageSize: 20,
})
assert.ok(searchedVoucherJob.items.some((job) => job.id === voucherJobs.rows[1]?.id))
} finally {
if (orderItemIds.length > 0) {
await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [
orderItemIds,
])
await query(`DELETE FROM cost_snapshots WHERE order_item_id = ANY($1::bigint[])`, [
orderItemIds,
])
await query(
`DELETE FROM work_order_events WHERE work_order_id IN (
SELECT id FROM work_orders WHERE order_item_id = ANY($1::bigint[])
)`,
[orderItemIds],
)
await query(`DELETE FROM work_orders WHERE order_item_id = ANY($1::bigint[])`, [
orderItemIds,
])
await query(
'ALTER TABLE fulfillment_state_events DISABLE TRIGGER trg_fulfillment_state_events_append_only',
)
try {
await query(`DELETE FROM order_items WHERE id = ANY($1::bigint[])`, [orderItemIds])
} finally {
await query(
'ALTER TABLE fulfillment_state_events ENABLE TRIGGER trg_fulfillment_state_events_append_only',
)
}
}
if (orderIds.length > 0) {
await query(`DELETE FROM orders WHERE id = ANY($1::bigint[])`, [orderIds])
}
await query(`DELETE FROM cost_exceptions WHERE scope_key LIKE $1`, [`%${suffix}%`])
await query(`DELETE FROM voucher_consume_jobs WHERE platform_order_id LIKE $1`, [
`IT-VOUCHER-LIST-${suffix}%`,
])
await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = ANY($1::text[])`, [
internalSkus,
])
await query(
`DELETE FROM internal_sku_mappings WHERE provider = '91kaquan' AND shop_id = $1`,
[shopId],
)
await query(
`DELETE FROM internal_skus
WHERE internal_sku_key = ANY($1::text[])`,
[internalSkus.map((sku) => sku.trim().normalize('NFKC').toLowerCase())],
)
await query(`DELETE FROM manual_sku_registry WHERE sku_key = ANY($1::text[])`, [internalSkus])
await query(
`DELETE FROM sku_channel_rules
WHERE provider = '91kaquan' AND shop_id = $1 AND match_key = ANY($2::text[])`,
[shopId, internalSkus],
)
const generatedProducts = await query<{ id: string }>(
`SELECT id
FROM product_catalog
WHERE canonicalize_internal_sku(substring(canonical_sku from 5)) = ANY($1::text[])`,
[internalSkus],
)
const generatedProductIds = generatedProducts.rows.map((row) => Number(row.id))
if (generatedProductIds.length > 0) {
await query(
`DELETE FROM cost_snapshots
WHERE canonical_product_id = ANY($1::bigint[])
OR cost_version_id IN (
SELECT id FROM cost_versions WHERE canonical_product_id = ANY($1::bigint[])
)`,
[generatedProductIds],
)
await query(
`DELETE FROM sales_daily_facts WHERE canonical_product_id = ANY($1::bigint[])`,
[generatedProductIds],
)
await query(
`DELETE FROM channel_product_mappings WHERE canonical_product_id = ANY($1::bigint[])`,
[generatedProductIds],
)
await query(`DELETE FROM cost_versions WHERE canonical_product_id = ANY($1::bigint[])`, [
generatedProductIds,
])
await query(
`DELETE FROM product_channel_capabilities WHERE product_id = ANY($1::bigint[])`,
[generatedProductIds],
)
await query(`DELETE FROM product_catalog WHERE id = ANY($1::bigint[])`, [
generatedProductIds,
])
}
if (retryWorkOrderRuleId) {
await query(`DELETE FROM work_product_match_logs WHERE rule_id = $1`, [
retryWorkOrderRuleId,
])
await query(`DELETE FROM cost_versions WHERE work_product_rule_id = $1`, [
retryWorkOrderRuleId,
])
await query(`DELETE FROM work_product_rules WHERE id = $1`, [retryWorkOrderRuleId])
}
if (fallbackProfileId) {
await query(`DELETE FROM fulfillment_profiles WHERE id = $1`, [fallbackProfileId])
}
await closeDb()
}
},
)