重构:硬切履约路由唯一真源

This commit is contained in:
yml2213
2026-11-08 19:44:27 +08:00
parent f6e193e7f6
commit 6bff1971dd
31 changed files with 1029 additions and 1022 deletions
@@ -771,9 +771,12 @@ export async function deleteManualSkuClassification(
[key],
)
const rules = await client.query(
`DELETE FROM internal_sku_channels
WHERE canonicalize_internal_sku(internal_sku) = $1 AND channel = 'manual'
RETURNING id`,
`DELETE FROM sku_channel_routes route
USING internal_skus sku
WHERE sku.id = route.internal_sku_id
AND sku.internal_sku_key = $1
AND route.channel = 'manual'
RETURNING route.id`,
[key],
)
const products = await client.query(
@@ -1332,9 +1335,7 @@ function requiredRow<T>(row: T | undefined, message: string): T {
return row
}
/**
* 人工配置 → internal_sku_channels(内部 SKU 精确匹配)。
*/
/** 人工配置 → sku_channel_routes(内部 SKU 精确匹配)。 */
async function syncManualChannelRule(
config: {
id: number
@@ -1350,10 +1351,13 @@ async function syncManualChannelRule(
const otherChannelSql = `
SELECT EXISTS (
SELECT 1 FROM internal_sku_channels
WHERE canonicalize_internal_sku(internal_sku) = $1
AND enabled = TRUE
AND channel <> 'manual'
SELECT 1
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE sku.internal_sku_key = $1
AND route.enabled = TRUE
AND route.route_role = 'primary'
AND route.channel <> 'manual'
) AS exists
`
const otherChannel = client
@@ -487,14 +487,16 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
await client.query(
`
UPDATE internal_sku_channels
UPDATE sku_channel_routes route
SET enabled = FALSE,
is_fallback = FALSE,
route_role = 'fallback',
updated_at = $3
WHERE channel = 'third_party_voucher'
FROM internal_skus sku
WHERE sku.id = route.internal_sku_id
AND route.channel = 'third_party_voucher'
AND (
canonical_product_id = $1
OR canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($2)
route.canonical_product_id = $1
OR sku.internal_sku_key = canonicalize_internal_sku($2)
)
`,
[id, internalSku, nowIso()],
@@ -545,11 +547,13 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
)
await client.query(
`
DELETE FROM internal_sku_channels
WHERE channel = 'third_party_voucher'
DELETE FROM sku_channel_routes route
USING internal_skus sku
WHERE sku.id = route.internal_sku_id
AND route.channel = 'third_party_voucher'
AND (
canonical_product_id = $1
OR canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($2)
route.canonical_product_id = $1
OR sku.internal_sku_key = canonicalize_internal_sku($2)
)
`,
[id, internalSku],
@@ -248,7 +248,7 @@ test(
])
await query('DELETE FROM product_channel_capabilities WHERE product_id = $1', [productId])
await query(
'UPDATE internal_sku_channels SET canonical_product_id = NULL WHERE canonical_product_id = $1',
'UPDATE sku_channel_routes SET canonical_product_id = NULL WHERE canonical_product_id = $1',
[productId],
)
await query('DELETE FROM product_catalog WHERE id = $1', [productId])
@@ -67,11 +67,13 @@ export async function runAffiliateFallback(
AND oi.fulfillment_locked = FALSE
AND cfg.id IS NOT NULL
AND EXISTS (
SELECT 1 FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND channel.channel = 'manual'
AND channel.is_fallback = TRUE
AND canonicalize_internal_sku(channel.internal_sku) = resolve_internal_sku(
SELECT 1
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled = TRUE
AND route.channel = 'manual'
AND route.route_role = 'fallback'
AND sku.internal_sku_key = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
)
)
@@ -257,9 +257,12 @@ export async function ensureCarrierForOrderItem(input: {
}),
)
const channelConfig = await query<{ config_json: Record<string, unknown> }>(
`SELECT config_json
FROM internal_sku_channels
WHERE canonicalize_internal_sku(internal_sku) = $1 AND channel = 'work_order' AND enabled = TRUE
`SELECT 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
AND route.channel = 'work_order'
AND route.enabled = TRUE
LIMIT 1`,
[internalSku],
)
@@ -110,25 +110,26 @@ async function loadCandidateItems(input: {
SELECT COALESCE(
jsonb_agg(
jsonb_build_object(
'id', channel.id,
'internal_sku', channel.internal_sku,
'channel', channel.channel,
'priority', channel.priority,
'is_fallback', channel.is_fallback,
'enabled', channel.enabled,
'canonical_product_id', channel.canonical_product_id,
'config_json', channel.config_json,
'source', channel.source
,'note', channel.note
,'created_by_admin_user_id', channel.created_by_admin_user_id
'id', route.id,
'internal_sku', sku.internal_sku_key,
'channel', route.channel,
'priority', route.priority,
'is_fallback', route.route_role = 'fallback',
'enabled', route.enabled,
'canonical_product_id', route.canonical_product_id,
'config_json', route.carrier_config_json,
'source', route.source,
'note', route.note,
'created_by_admin_user_id', route.created_by_admin_user_id
)
ORDER BY channel.priority, channel.channel
ORDER BY route.priority, route.channel
),
'[]'::jsonb
) AS channels
FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND canonicalize_internal_sku(channel.internal_sku) = resolve_internal_sku(
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled = TRUE
AND sku.internal_sku_key = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
)
) capability ON jsonb_array_length(capability.channels) > 0
@@ -246,7 +247,7 @@ async function selectSafeFallback(input: {
}
/**
* 按 internal_sku_channels 的明确优先级补载体。
* 按 sku_channel_routes 的明确优先级补载体。
*
* 主渠道是 affiliate/work_order 但载体暂时创建失败时,只有显式 is_fallback 的
* manual/voucher 才能接管。所有结果都落 pending 明细,不再静默卡住。
@@ -56,7 +56,7 @@ import {
* 判定顺序:
* 1. 运营锁定 → 跳过
* 2. 已有且唯一的真实载体 → 直接按载体落渠道;多载体 → 冲突
* 3. internal_sku_channels:一个主渠道 + 显式兜底;多主渠道 → 冲突
* 3. sku_channel_routes:一个主渠道 + 显式兜底;多主渠道 → 冲突
* 4. 没有统一配置或载体不满足 → 保留 pending 并写明确原因
*/
@@ -1279,11 +1279,11 @@ export async function listPendingChannelGroups(input: {
FROM grouped
LEFT JOIN issue_summary ON issue_summary.group_key = grouped.group_key
LEFT JOIN LATERAL (
SELECT array_agg(DISTINCT channel.channel ORDER BY channel.channel) AS channels
FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND canonicalize_internal_sku(channel.internal_sku) =
grouped.internal_sku
SELECT array_agg(DISTINCT route.channel ORDER BY route.channel) AS channels
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled = TRUE
AND sku.internal_sku_key = grouped.internal_sku
) cfg ON TRUE
ORDER BY grouped.item_count DESC, grouped.internal_sku
LIMIT $${params.length - 1} OFFSET $${params.length}
@@ -2017,11 +2017,18 @@ function buildChannelRuleFilter(input: ChannelRuleFilterInput) {
const CHANNEL_RULE_FROM_SQL = `
FROM (
SELECT id,
SELECT route.id,
'internal_sku'::text AS match_key_type,
canonicalize_internal_sku(internal_sku) AS match_key,
channel, priority, enabled, is_fallback, source, config_json, note
FROM internal_sku_channels
sku.internal_sku_key AS match_key,
route.channel,
route.priority,
route.enabled,
route.route_role = 'fallback' AS is_fallback,
route.source,
route.carrier_config_json AS config_json,
route.note
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
) internal_rules
`
@@ -2331,8 +2338,13 @@ async function assertRulesDeletable(idsInput: number[]): Promise<void> {
const ids = [...new Set(idsInput.map(Number).filter((id) => Number.isInteger(id) && id > 0))]
if (ids.length === 0) return
const rows = await query<{ id: number; source: string; match_key: string; channel: string }>(
`SELECT id, source, canonicalize_internal_sku(internal_sku) AS match_key, channel
FROM internal_sku_channels WHERE id = ANY($1::bigint[])`,
`SELECT route.id,
route.source,
sku.internal_sku_key AS match_key,
route.channel
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.id = ANY($1::bigint[])`,
[ids],
)
const protectedRow = rows.rows.find(
@@ -2355,11 +2367,21 @@ async function loadInternalSkuChannelsForKeys(
const rows = await query<InternalSkuChannelRow>(
`
SELECT id, internal_sku, channel, priority, is_fallback, enabled,
canonical_product_id, config_json, source, note, created_by_admin_user_id
FROM internal_sku_channels
WHERE enabled = TRUE AND canonicalize_internal_sku(internal_sku) = ANY($1::text[])
ORDER BY canonicalize_internal_sku(internal_sku), priority, channel, id
SELECT route.id,
sku.internal_sku_key AS internal_sku,
route.channel,
route.priority,
route.route_role = 'fallback' AS is_fallback,
route.enabled,
route.canonical_product_id,
route.carrier_config_json AS config_json,
route.source,
route.note,
route.created_by_admin_user_id
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled = TRUE AND sku.internal_sku_key = ANY($1::text[])
ORDER BY sku.internal_sku_key, route.priority, route.channel, route.id
`,
[normalized],
)
@@ -128,6 +128,59 @@ test(
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
@@ -156,12 +209,7 @@ test(
WHERE id = (SELECT order_id FROM order_items WHERE id = $1)`,
[cutoffOldItemId],
)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES ($1, 'third_party_voucher', 10, FALSE, TRUE, '{}'::jsonb, 'integration_test')`,
[cutoffTestSku],
)
await setRoute({ sku: cutoffTestSku, channel: 'third_party_voucher' })
const cutoffAssignment = await assignPendingOrderItemChannels({
orderItemIds: [cutoffOldItemId, cutoffNewItemId],
batchSize: 10,
@@ -180,17 +228,18 @@ test(
assert.equal(cutoffStates.rows[0]?.fulfillment_status, 'pending')
assert.equal(cutoffStates.rows[1]?.fulfillment_status, 'assigned')
await query(
`INSERT INTO internal_sku_channels (internal_sku, channel, priority, is_fallback, enabled, source)
VALUES ($1, 'manual', 90, TRUE, TRUE, 'integration_test')`,
[` 集成测试 规范-${suffix} `],
)
await setRoute({
sku: ` 集成测试 规范-${suffix} `,
channel: 'manual',
role: 'fallback',
})
const normalizedInsert = await query<{ internal_sku: string }>(
`SELECT internal_sku
FROM internal_sku_channels
WHERE source = 'integration_test'
AND canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($1)
ORDER BY id DESC
`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],
)
@@ -388,12 +437,11 @@ test(
)`,
[retryWorkOrderRuleId, retryWorkOrderSku],
)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES ($1, 'work_order', 10, FALSE, TRUE, $2::jsonb, 'integration_test')`,
[retryWorkOrderSku, JSON.stringify({ workRuleKey: retryWorkOrderRuleKey })],
)
await setRoute({
sku: retryWorkOrderSku,
channel: 'work_order',
configJson: { workRuleKey: retryWorkOrderRuleKey },
})
const retriedWorkOrders = await retryPendingWorkOrderRouting({
internalSku: retryWorkOrderSku,
})
@@ -535,12 +583,12 @@ test(
) VALUES ($1, $1, 1, 1, 1, 1000, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, NOW(), NOW())`,
[manualRemoveSku],
)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, source, config_json
) VALUES ($1, 'manual', 90, TRUE, TRUE, 'manual_config', '{}'::jsonb)`,
[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)
@@ -554,28 +602,20 @@ test(
assert.equal(removedManualItem.rows[0]?.fulfillment_status, 'pending')
assert.equal(removedManualItem.rows[0]?.fulfillment_channel, null)
const protectedRule = await query<{ id: number }>(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, source, config_json
) VALUES ($1, 'manual', 70, FALSE, TRUE, 'manual_config', '{}'::jsonb)
RETURNING id`,
[protectedRuleSku],
)
const protectedRuleId = Number(protectedRule.rows[0]!.id)
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 query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES
($1, 'affiliate_dash', 10, FALSE, TRUE, '{}'::jsonb, 'integration_test'),
($1, 'manual', 90, TRUE, TRUE, '{}'::jsonb, 'integration_test')`,
[fallbackSku],
)
await setRoute({ sku: fallbackSku, channel: 'affiliate_dash' })
await setRoute({ sku: fallbackSku, channel: 'manual', role: 'fallback' })
const fallbackItem = await query<{ order_id: number }>(
`UPDATE order_items
SET fulfillment_status = 'assigned', fulfillment_channel = 'affiliate_dash',
@@ -743,8 +783,10 @@ test(
assert.equal(await ensureVoucherRuleForSku(missingVoucherSku), false)
const missingVoucherRules = await query<{ count: number }>(
`SELECT count(*)::int AS count FROM internal_sku_channels
WHERE internal_sku = $1 AND channel = 'third_party_voucher'`,
`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)
@@ -767,9 +809,16 @@ test(
is_fallback: boolean
config_json: Record<string, unknown>
}>(
`SELECT channel, source, canonical_product_id, priority, is_fallback, config_json
FROM internal_sku_channels
WHERE internal_sku = $1 ORDER BY channel`,
`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)
@@ -855,10 +904,12 @@ test(
is_fallback: boolean
}>(
`SELECT product.status, product.attributes_json,
channel.canonical_product_id, channel.priority, channel.is_fallback
FROM internal_sku_channels channel
JOIN product_catalog product ON product.id = channel.canonical_product_id
WHERE channel.id = $1`,
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')
@@ -869,15 +920,11 @@ test(
assert.equal(typeof reactivated.rows[0]?.attributes_json.reactivatedAt, 'string')
const retryPendingItemId = await createPendingOrderItem(retryPendingAffiliateSku)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES (
$1, 'affiliate_dash', 10, FALSE, TRUE,
jsonb_build_object('affiliateSku', $1::text), 'integration_test'
)`,
[retryPendingAffiliateSku],
)
await setRoute({
sku: retryPendingAffiliateSku,
channel: 'affiliate_dash',
configJson: { affiliateSku: retryPendingAffiliateSku },
})
await query(
`UPDATE order_items
SET fulfillment_status = 'assigned',
@@ -1067,7 +1114,16 @@ test(
await query(`DELETE FROM work_orders WHERE order_item_id = ANY($1::bigint[])`, [
orderItemIds,
])
await query(`DELETE FROM order_items WHERE 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])
@@ -1083,9 +1139,11 @@ test(
`DELETE FROM internal_sku_mappings WHERE provider = '91kaquan' AND shop_id = $1`,
[shopId],
)
await query(`DELETE FROM internal_sku_channels WHERE internal_sku = ANY($1::text[])`, [
internalSkus,
])
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
@@ -14,6 +14,10 @@ const configSyncMigration = readFileSync(
new URL('../../db/migrations/144_fulfillment_shadow_configuration_sync.sql', import.meta.url),
'utf8',
)
const routeCutoverMigration = readFileSync(
new URL('../../db/migrations/161_move_shadow_sync_to_sku_routes.sql', import.meta.url),
'utf8',
)
test('正式履约影子模型包含履约、尝试和异常三张表', () => {
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillments'))
@@ -33,8 +37,12 @@ test('影子模型从现有真源回填并持续双写', () => {
test('人工配置和统一渠道规则变化会刷新履约影子', () => {
assert.ok(configSyncMigration.includes('trg_manual_fulfillment_config_shadow'))
assert.ok(configSyncMigration.includes('trg_internal_sku_channel_shadow'))
assert.ok(configSyncMigration.includes('sync_fulfillment_shadow(item.id)'))
assert.ok(routeCutoverMigration.includes('trg_sku_route_fulfillment_shadow'))
assert.ok(routeCutoverMigration.includes('sync_fulfillment_shadow_for_sku_route'))
assert.ok(
routeCutoverMigration.includes('DROP TRIGGER IF EXISTS trg_internal_sku_channel_shadow'),
)
})
test('人工履约配置不能关联其他来源商品', () => {
@@ -77,7 +77,7 @@ function withChannelSnapshot(input: {
/**
* 纯函数:把内部 SKU 渠道计划应用到订单项。
* 渠道选择只允许来自 internal_sku_channels,旧路由不能抢占。
* 渠道选择只允许来自 sku_channel_routes,旧表和其他路由不能抢占。
*/
export function applyIntakeChannelPlan(
item: ResolvedFulfillmentItem,
@@ -190,7 +190,7 @@ export function applyIntakeChannelPlan(
})
}
// 接单平台由发单入口读取同一份 internal_sku_channels 创建工单;
// 接单平台由发单入口读取同一份 sku_channel_routes 创建工单;
// 卡券无执行任务。两者都不能留下 affiliate/manual 路由。
return withChannelSnapshot({
item,
@@ -207,7 +207,7 @@ export function applyIntakeChannelPlan(
})
}
/** 下单入口读取唯一渠道源:internal_sku_channels。 */
/** 下单入口读取唯一渠道源:sku_channel_routes。 */
export async function applyIntakeChannelDecision(
item: ResolvedFulfillmentItem,
_input: IntakeChannelDecisionInput,
@@ -1,4 +1,5 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
import {
@@ -53,15 +54,13 @@ test('渠道计划:非兜底渠道即使优先级数值更大也压过兜底
assert.equal(plan.fallbacks[0]?.channel, 'manual')
})
test('渠道计划:仅有一个兜底渠道时仍可作为实际主渠道', () => {
test('渠道计划:只有兜底渠道时保持未路由,必须人工补主渠道', () => {
const plan = planInternalSkuChannels([
channel({ channel: 'manual', priority: 90, is_fallback: true }),
])
assert.equal(plan.primary?.channel, 'manual')
assert.deepEqual(
plan.attempts.map((item) => item.channel),
['manual'],
)
assert.equal(plan.primary, null)
assert.deepEqual(plan.fallbacks, [])
assert.deepEqual(plan.attempts, [])
})
test('渠道计划:历史规范化重复的同一渠道会去重,不误判为冲突', () => {
@@ -73,3 +72,14 @@ test('渠道计划:历史规范化重复的同一渠道会去重,不误判
assert.equal(plan.primary?.id, 2)
assert.deepEqual(plan.conflicts, [])
})
test('运行时渠道服务只读 sku_channel_routes,不再读写旧表', async () => {
const source = await readFile(
new URL('./internal-sku-channel-service.ts', import.meta.url),
'utf8',
)
assert.ok(source.includes('FROM sku_channel_routes'))
assert.ok(source.includes('INSERT INTO sku_channel_routes'))
assert.ok(source.includes('DELETE FROM sku_channel_routes'))
assert.ok(!source.includes('internal_sku_channels'))
})
@@ -32,13 +32,23 @@ export async function listInternalSkuChannels(
if (!key) return []
const result = await query<InternalSkuChannelRow>(
`
SELECT id, internal_sku, channel, priority, is_fallback, enabled, canonical_product_id,
config_json, source, note, created_by_admin_user_id
FROM internal_sku_channels
WHERE canonicalize_internal_sku(internal_sku) = $1 AND enabled = TRUE
SELECT route.id,
sku.internal_sku_key AS internal_sku,
route.channel,
route.priority,
(route.route_role = 'fallback') AS is_fallback,
route.enabled,
route.canonical_product_id,
route.carrier_config_json AS config_json,
route.source,
route.note,
route.created_by_admin_user_id
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE sku.internal_sku_key = $1 AND route.enabled = TRUE
ORDER BY
priority ASC,
channel ASC
route.priority ASC,
route.channel ASC
`,
[key],
)
@@ -55,11 +65,21 @@ export async function loadInternalSkuChannelPlans(
const rows = await query<InternalSkuChannelRow>(
`
SELECT id, internal_sku, channel, priority, is_fallback, enabled,
canonical_product_id, config_json, source, note, created_by_admin_user_id
FROM internal_sku_channels
WHERE enabled = TRUE AND canonicalize_internal_sku(internal_sku) = ANY($1::text[])
ORDER BY canonicalize_internal_sku(internal_sku), priority, channel, id
SELECT route.id,
sku.internal_sku_key AS internal_sku,
route.channel,
route.priority,
(route.route_role = 'fallback') AS is_fallback,
route.enabled,
route.canonical_product_id,
route.carrier_config_json AS config_json,
route.source,
route.note,
route.created_by_admin_user_id
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled = TRUE AND sku.internal_sku_key = ANY($1::text[])
ORDER BY sku.internal_sku_key, route.priority, route.channel, route.id
`,
[keys],
)
@@ -105,7 +125,7 @@ export function planInternalSkuChannels(input: InternalSkuChannelRow[]): Interna
)
const explicitPrimary = channels.filter((row) => !row.is_fallback)
const primaryPool = explicitPrimary.length > 0 ? explicitPrimary : channels
const primaryPool = explicitPrimary
if (primaryPool.length === 0) {
return { primary: null, fallbacks: [], attempts: [], conflicts: [] }
}
@@ -149,46 +169,52 @@ async function upsertInternalSkuChannelWithExecutor(
): Promise<number> {
const internalSku = canonicalizeInternalSku(input.internalSku)
if (!internalSku) return 0
const skuResult = await client.query<{ id: string }>(
'SELECT ensure_canonical_internal_sku($1) AS id',
[internalSku],
)
const internalSkuId = Number(skuResult.rows[0]?.id || 0)
if (!internalSkuId) return 0
if (input.enabled !== false && input.isFallback !== true) {
await client.query(
`UPDATE internal_sku_channels
SET is_fallback = TRUE,
`UPDATE sku_channel_routes
SET route_role = 'fallback',
priority = GREATEST(priority, 90),
updated_at = NOW()
WHERE canonicalize_internal_sku(internal_sku) = $1
WHERE internal_sku_id = $1
AND channel <> $2
AND enabled = TRUE
AND is_fallback = FALSE`,
[internalSku, input.channel],
AND route_role = 'primary'`,
[internalSkuId, input.channel],
)
}
const result = await client.query<{ id: number }>(
`
INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled,
canonical_product_id, config_json, source, note, created_by_admin_user_id,
INSERT INTO sku_channel_routes (
internal_sku_id, channel, route_role, priority, enabled,
canonical_product_id, carrier_config_json, source, note, created_by_admin_user_id,
created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, NOW(), NOW())
ON CONFLICT (internal_sku, channel) DO UPDATE SET
ON CONFLICT (internal_sku_id, channel) DO UPDATE SET
priority = CASE
WHEN $11::boolean THEN EXCLUDED.priority
ELSE LEAST(internal_sku_channels.priority, EXCLUDED.priority)
ELSE LEAST(sku_channel_routes.priority, EXCLUDED.priority)
END,
is_fallback = EXCLUDED.is_fallback,
route_role = EXCLUDED.route_role,
enabled = EXCLUDED.enabled,
canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, sku_channel_routes.canonical_product_id),
carrier_config_json = sku_channel_routes.carrier_config_json || EXCLUDED.carrier_config_json,
source = EXCLUDED.source,
note = CASE WHEN EXCLUDED.note <> '' THEN EXCLUDED.note ELSE internal_sku_channels.note END,
created_by_admin_user_id = COALESCE(EXCLUDED.created_by_admin_user_id, internal_sku_channels.created_by_admin_user_id),
note = CASE WHEN EXCLUDED.note <> '' THEN EXCLUDED.note ELSE sku_channel_routes.note END,
created_by_admin_user_id = COALESCE(EXCLUDED.created_by_admin_user_id, sku_channel_routes.created_by_admin_user_id),
updated_at = NOW()
RETURNING id
`,
[
internalSku,
internalSkuId,
input.channel,
input.isFallback === true ? 'fallback' : 'primary',
Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100,
input.isFallback === true,
input.enabled !== false,
input.canonicalProductId ?? null,
JSON.stringify(input.configJson || {}),
@@ -250,9 +276,12 @@ export async function disableInternalSkuChannel(input: {
}): Promise<number> {
const result = await query(
`
UPDATE internal_sku_channels
UPDATE sku_channel_routes route
SET enabled = FALSE, updated_at = NOW()
WHERE canonicalize_internal_sku(internal_sku) = $1 AND channel = $2
FROM internal_skus sku
WHERE sku.id = route.internal_sku_id
AND sku.internal_sku_key = $1
AND route.channel = $2
`,
[canonicalizeInternalSku(input.internalSku), input.channel],
)
@@ -262,7 +291,7 @@ export async function disableInternalSkuChannel(input: {
export async function deleteInternalSkuChannelById(channelId: number): Promise<boolean> {
const id = Math.trunc(Number(channelId) || 0)
if (id <= 0) return false
const result = await query(`DELETE FROM internal_sku_channels WHERE id = $1`, [id])
const result = await query(`DELETE FROM sku_channel_routes WHERE id = $1`, [id])
return (result.rowCount || 0) > 0
}
@@ -273,7 +302,7 @@ export async function deleteInternalSkuChannelsByIds(idsInput: number[]): Promis
const ids = [...new Set(idsInput.map(Number).filter((id) => Number.isInteger(id) && id > 0))]
if (ids.length === 0) return { deletedIds: [], missingIds: [] }
const result = await query<{ id: number }>(
`DELETE FROM internal_sku_channels WHERE id = ANY($1::bigint[]) RETURNING id`,
`DELETE FROM sku_channel_routes WHERE id = ANY($1::bigint[]) RETURNING id`,
[ids],
)
const deletedIds = result.rows.map((row) => Number(row.id))
@@ -1,5 +1,5 @@
/**
* 内部 SKU 与渠道路由兼容层集成测试(需要真实 PostgreSQL)。
* 内部 SKU 与渠道路由硬切换集成测试(需要真实 PostgreSQL)。
*
* INTERNAL_SKU_ROUTING_TEST_DATABASE_URL=postgres://... \
* npx tsx --test src/services/fulfillment/internal-sku-routing.integration.test.ts
@@ -10,11 +10,12 @@ import test from 'node:test'
const TEST_DATABASE_URL = String(process.env.INTERNAL_SKU_ROUTING_TEST_DATABASE_URL || '').trim()
test(
'旧 internal_sku_channels 写入会同步路由,商品来源与主渠道投影保持分离',
'运行时只写 sku_channel_routes,旧 internal_sku_channels 已停止同步',
{ skip: !TEST_DATABASE_URL ? '未设置 INTERNAL_SKU_ROUTING_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const { upsertInternalSkuChannel } = await import('./internal-sku-channel-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const internalSku = `路由集成测试-${suffix}`
const canonicalSku = `man:${internalSku}`
@@ -22,6 +23,25 @@ test(
let productId = 0
try {
const triggers = await query<{ event_object_table: string; trigger_name: string }>(
`
SELECT DISTINCT event_object_table, trigger_name
FROM information_schema.triggers
WHERE trigger_name IN (
'trg_sync_sku_channel_routes',
'trg_internal_sku_channel_shadow',
'trg_sku_route_fulfillment_shadow'
)
ORDER BY event_object_table, trigger_name
`,
)
assert.deepEqual(triggers.rows, [
{
event_object_table: 'sku_channel_routes',
trigger_name: 'trg_sku_route_fulfillment_shadow',
},
])
const createdSku = await query<{ id: string }>(
'SELECT ensure_canonical_internal_sku($1) AS id',
[internalSku],
@@ -53,14 +73,16 @@ test(
'man: 来源没有主路由时必须显示未路由',
)
await query(
`
INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source
) VALUES ($1, 'manual', 10, FALSE, TRUE, $2, 'integration_test')
`,
[internalSku, productId],
)
await upsertInternalSkuChannel({
internalSku,
channel: 'manual',
priority: 10,
isFallback: false,
canonicalProductId: productId,
configJson: { manualConfigId: 1 },
source: 'integration_test',
replacePriority: true,
})
const primary = await query<{ route_role: string; channel: string }>(
`
@@ -96,28 +118,46 @@ test(
assert.equal(products.pagination.total, overview.productCount)
assert.ok(products.items.every((product) => product.activePrimaryChannel === 'manual'))
await query(
await upsertInternalSkuChannel({
internalSku,
channel: 'work_order',
priority: 10,
isFallback: false,
configJson: { workRuleKey: `rule-${suffix}` },
source: 'integration_test',
replacePriority: true,
})
const switched = await query<{ channel: string; route_role: string }>(
`
UPDATE internal_sku_channels
SET is_fallback = TRUE, updated_at = NOW()
WHERE internal_sku = $1 AND channel = 'manual'
SELECT channel, route_role
FROM sku_channel_routes
WHERE internal_sku_id = $1
ORDER BY channel
`,
[internalSku],
)
const fallback = await query<{ route_role: string }>(
'SELECT route_role FROM sku_channel_routes WHERE internal_sku_id = $1',
[internalSkuId],
)
assert.deepEqual(fallback.rows, [{ route_role: 'fallback' }])
assert.deepEqual(switched.rows, [
{ channel: 'manual', route_role: 'fallback' },
{ channel: 'work_order', route_role: 'primary' },
])
await query(
`
UPDATE internal_sku_channels
SET is_fallback = FALSE, updated_at = NOW()
WHERE internal_sku = $1 AND channel = 'manual'
INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, source
) VALUES ($1, 'affiliate_dash', 1, FALSE, TRUE, 'integration_test')
`,
[internalSku],
)
const legacyOnly = await query<{ count: number }>(
`
SELECT COUNT(*)::int AS count
FROM sku_channel_routes
WHERE internal_sku_id = $1 AND channel = 'affiliate_dash'
`,
[internalSkuId],
)
assert.equal(legacyOnly.rows[0]?.count, 0, '旧表写入不能再生效或同步到新路由')
await assert.rejects(
() =>
@@ -125,7 +165,7 @@ test(
`
INSERT INTO sku_channel_routes (
internal_sku_id, channel, route_role, priority, enabled
) VALUES ($1, 'work_order', 'primary', 1, TRUE)
) VALUES ($1, 'affiliate_dash', 'primary', 1, TRUE)
`,
[internalSkuId],
),
@@ -39,10 +39,11 @@ export async function retryPendingWorkOrderRouting(
WHERE oi.fulfillment_status = 'pending'
AND EXISTS (
SELECT 1
FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND channel.channel = 'work_order'
AND canonicalize_internal_sku(channel.internal_sku) =
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled = TRUE
AND route.channel = 'work_order'
AND sku.internal_sku_key =
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
)
${skuClause}
@@ -56,7 +56,7 @@ export type ResolvedFulfillmentItem = FulfillmentItem & {
* 订单入口只负责把各种平台字段统一归一到中文内部 SKU。
*
* 渠道选择、载体配置和路由优先级都不在这里判断;它们统一读取
* `internal_sku_channels`,避免平台 SKU 或旧路由抢先创建载体。
* `sku_channel_routes`,避免平台 SKU 或旧路由抢先创建载体。
*/
export async function resolveOrderItemForFulfillment(
{ provider = '', platform = '', shopId = '', item = {} }: ResolveOrderItemForFulfillmentInput,