From 6bff1971ddec639f7d11cae55eae3dcf7e751d24 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 8 Nov 2026 18:44:46 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=EF=BC=9A=E7=A1=AC=E5=88=87?= =?UTF-8?q?=E5=B1=A5=E7=BA=A6=E8=B7=AF=E7=94=B1=E5=94=AF=E4=B8=80=E7=9C=9F?= =?UTF-8?q?=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 3 +- apps/backend/package.json | 1 - .../audit-fulfillment-finance-alignment.ts | 63 +-- .../audit-product-routing-alignment.ts | 26 +- .../scripts/cleanup-legacy-sku-code-rules.ts | 134 ------ apps/backend/scripts/finance-cutover-v2.ts | 2 +- .../scripts/merge-duplicate-products.ts | 60 +-- .../scripts/normalize-internal-sku-configs.ts | 201 -------- .../scripts/preflight-fulfillment-routing.ts | 345 -------------- apps/backend/scripts/rebuild-after-cutover.ts | 6 +- .../scripts/reclassify-post-cutover.ts | 18 +- .../scripts/seed-local-finance-demo.ts | 91 ++-- ...0_switch_fulfillment_routing_to_routes.sql | 433 ++++++++++++++++++ .../161_move_shadow_sync_to_sku_routes.sql | 46 ++ .../162_drop_legacy_routing_read_surfaces.sql | 12 + .../manual-fulfillment-config-service.ts | 24 +- .../cost-center/product-catalog-service.ts | 22 +- ...rk-order-template-cost.integration.test.ts | 2 +- .../fulfillment/affiliate-fallback-service.ts | 12 +- .../fulfillment/carrier-creation-service.ts | 9 +- .../carrier-reconciliation-service.ts | 33 +- .../fulfillment/channel-assign-service.ts | 56 ++- .../fulfillment-routing.integration.test.ts | 190 +++++--- .../fulfillment-shadow-model.test.ts | 10 +- .../intake-channel-decision-service.ts | 6 +- .../internal-sku-channel-service.test.ts | 22 +- .../internal-sku-channel-service.ts | 95 ++-- .../internal-sku-routing.integration.test.ts | 86 +++- .../pending-channel-repair-service.ts | 9 +- .../fulfillment/product-resolution-service.ts | 2 +- docs/统一订单履约与财务模型-权威版.md | 32 +- 31 files changed, 1029 insertions(+), 1022 deletions(-) delete mode 100644 apps/backend/scripts/cleanup-legacy-sku-code-rules.ts delete mode 100644 apps/backend/scripts/normalize-internal-sku-configs.ts delete mode 100644 apps/backend/scripts/preflight-fulfillment-routing.ts create mode 100644 apps/backend/src/db/migrations/160_switch_fulfillment_routing_to_routes.sql create mode 100644 apps/backend/src/db/migrations/161_move_shadow_sync_to_sku_routes.sql create mode 100644 apps/backend/src/db/migrations/162_drop_legacy_routing_read_surfaces.sql diff --git a/AGENTS.md b/AGENTS.md index 3adf284d..65b7039c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -62,7 +62,8 @@ 5. 页面与统计如果显示同一个业务量，必须读取同一投影或同一查询口径。 6. 不把临时修数据脚本当成长期业务流程。 7. 历史清理必须可 dry-run、可审计、可回滚，并且全量扫描存量。 -8. 删除兼容代码前必须完成“回填 → 双写 → 对比 → 切换读路径 → 停止旧写 → 观察”。 +8. 结构性切换采用“备份 → 回填 → 一致性审计 → 原子切换 → 观察”。 + 不再新增双读、双写或旧表运行期兼容；无法确定的数据进入待分类，不做猜测。 ## 文档规则 diff --git a/apps/backend/package.json b/apps/backend/package.json index fd5fd577..faf6a594 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -10,7 +10,6 @@ "db:migrate:create": "tsx scripts/create-migration.ts", "db:migrate:status": "tsx scripts/migration-status.ts", "finance:sync": "tsx scripts/sync-cost-finance.ts", - "fulfillment:normalize-sku-configs": "tsx scripts/normalize-internal-sku-configs.ts", "finance:rebuild-cutover": "tsx scripts/rebuild-after-cutover.ts", "revenue:backfill": "tsx scripts/backfill-revenue-basis.ts", "cost-exceptions:archive": "tsx scripts/archive-cost-exceptions.ts", diff --git a/apps/backend/scripts/audit-fulfillment-finance-alignment.ts b/apps/backend/scripts/audit-fulfillment-finance-alignment.ts index 1cdd3811..2a2a3db2 100644 --- a/apps/backend/scripts/audit-fulfillment-finance-alignment.ts +++ b/apps/backend/scripts/audit-fulfillment-finance-alignment.ts @@ -90,22 +90,6 @@ async function main() { `, ) - const rules = await query<{ - match_key_type: string - channel: string - scope: string - total: number - }>( - ` - SELECT match_key_type, channel, - CASE WHEN provider = '' AND shop_id = '' THEN 'global' ELSE 'scoped' END AS scope, - count(*)::int AS total - FROM sku_channel_rules - GROUP BY 1, 2, 3 - ORDER BY 1, 2, 3 - `, - ) - const duplicateProducts = await query<{ duplicate_names: number; duplicate_rows: number }>( ` SELECT count(*)::int AS duplicate_names, COALESCE(sum(total), 0)::int AS duplicate_rows @@ -119,43 +103,29 @@ async function main() { `, ) - const capabilities = await query<{ channel: string; total: number }>( - `SELECT channel, count(*)::int AS total FROM product_channel_capabilities WHERE enabled GROUP BY 1 ORDER BY 2 DESC`, - ) - - const internalChannelConflicts = await query<{ internal_sku: string; total: number }>( - `SELECT internal_sku, count(*)::int AS total - FROM internal_sku_channels - WHERE enabled = TRUE AND is_fallback = FALSE - GROUP BY internal_sku - HAVING count(*) > 1 - ORDER BY count(*) DESC, internal_sku`, - ) - - const internalChannels = await query<{ + const routes = await query<{ channel: string + route_role: string source: string total: number }>( - `SELECT channel, source, count(*)::int AS total - FROM internal_sku_channels + `SELECT channel, route_role, source, count(*)::int AS total + FROM sku_channel_routes WHERE enabled = TRUE - GROUP BY channel, source - ORDER BY channel, source`, + GROUP BY channel, route_role, source + ORDER BY channel, route_role, source`, ) - const canonicalChannelDuplicates = await query<{ - internal_sku: string - channel: string + const internalRouteConflicts = await query<{ + internal_sku_id: number total: number }>( - `SELECT canonicalize_internal_sku(internal_sku) AS internal_sku, - channel, count(*)::int AS total - FROM internal_sku_channels - WHERE enabled = TRUE - GROUP BY canonicalize_internal_sku(internal_sku), channel + `SELECT internal_sku_id, count(*)::int AS total + FROM sku_channel_routes + WHERE enabled = TRUE AND route_role = 'primary' + GROUP BY internal_sku_id HAVING count(*) > 1 - ORDER BY count(*) DESC, internal_sku, channel`, + ORDER BY count(*) DESC, internal_sku_id`, ) const assignedCarrierConflicts = await query<{ @@ -220,12 +190,9 @@ async function main() { pending: pending.rows[0], channelCarrier: channelCarrier.rows, internalSku: internalSku.rows[0], - rules: rules.rows, duplicateProducts: duplicateProducts.rows[0], - capabilities: capabilities.rows, - internalChannels: internalChannels.rows, - internalChannelConflicts: internalChannelConflicts.rows, - canonicalChannelDuplicates: canonicalChannelDuplicates.rows, + routes: routes.rows, + internalRouteConflicts: internalRouteConflicts.rows, assignedCarrierConflicts: assignedCarrierConflicts.rows, byProvider: byProvider.rows, }, diff --git a/apps/backend/scripts/audit-product-routing-alignment.ts b/apps/backend/scripts/audit-product-routing-alignment.ts index f1a83ca5..4b161e79 100644 --- a/apps/backend/scripts/audit-product-routing-alignment.ts +++ b/apps/backend/scripts/audit-product-routing-alignment.ts @@ -18,6 +18,7 @@ async function main() { orphanManualProducts, multiplePrimaryRoutes, missingInternalSku, + unroutedProducts, invalidSourceKinds, ] = await Promise.all([ query<{ source_kind: string; primary_channel: string | null; total: number }>( @@ -46,7 +47,9 @@ async function main() { SELECT CASE WHEN primary_route.channel = 'manual' THEN 'primary_manual' - WHEN manual_route.route_role = 'fallback' THEN 'manual_fallback' + WHEN primary_route.channel IS NOT NULL + AND manual_route.route_role = 'fallback' THEN 'manual_fallback' + WHEN manual_route.route_role = 'fallback' THEN 'fallback_without_primary' ELSE 'missing_manual_route' END AS route_state, COUNT(*)::int AS total @@ -123,6 +126,19 @@ async function main() { ORDER BY 1 `, ), + query<{ + product_id: number + canonical_sku: string + canonical_name: string + source_kind: string + }>( + ` + SELECT id AS product_id, canonical_sku, canonical_name, source_kind + FROM v_product_current + WHERE status = 'active' AND active_primary_channel IS NULL + ORDER BY source_kind, id + `, + ), query<{ source_kind: string; total: number }>( ` SELECT source_kind, COUNT(*)::int AS total @@ -146,6 +162,11 @@ async function main() { multiplePrimaryRoutes: multiplePrimaryRoutes.rows.length, missingInternalSku: missingInternalSku.rows.reduce((sum, row) => sum + Number(row.total), 0), invalidSourceKinds: invalidSourceKinds.rows.reduce((sum, row) => sum + Number(row.total), 0), + invalidManualConfigRoutes: manualConfigRoutes.rows + .filter( + (row) => row.route_state !== 'primary_manual' && row.route_state !== 'manual_fallback', + ) + .reduce((sum, row) => sum + Number(row.total), 0), } console.info('[product-routing-audit] 商品来源 × 当前渠道') @@ -169,6 +190,9 @@ async function main() { console.info('[product-routing-audit] 非法商品来源') console.table(invalidSourceKinds.rows) + console.info('[product-routing-audit] 未路由活动商品（待人工分类）') + console.table(unroutedProducts.rows) + console.info('[product-routing-audit] 关键异常', critical) if (strict && Object.values(critical).some((count) => count > 0)) { process.exitCode = 1 diff --git a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts deleted file mode 100644 index 19e84961..00000000 --- a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts +++ /dev/null @@ -1,134 +0,0 @@ -/** - * 把旧的 sku_code 渠道规则转换为 internal_sku 全局规则，并停用旧规则。 - * - * 用法： - * npx tsx scripts/cleanup-legacy-sku-code-rules.ts - * npx tsx scripts/cleanup-legacy-sku-code-rules.ts --apply - */ -import { closeDb, query } from '../src/db/client.js' - -type LegacyRule = { - id: number - provider: string - shop_id: string - match_key: string - channel: string - priority: number - source: string -} - -async function resolveInternalSku(rule: LegacyRule): Promise { - if (rule.channel === 'affiliate_dash') { - const product = await query<{ canonical_name: string }>( - `SELECT canonical_name FROM product_catalog - WHERE canonical_sku = $1 AND status = 'active' AND btrim(canonical_name) <> '' - LIMIT 1`, - [`aff:${rule.match_key}`], - ) - if (product.rows[0]?.canonical_name) return String(product.rows[0].canonical_name).trim() - } - - const params: unknown[] = [rule.match_key] - let scope = '' - if (rule.provider) { - params.push(rule.provider) - scope += ` AND o.provider = $${params.length}` - } - if (rule.shop_id) { - params.push(rule.shop_id) - scope += ` AND o.shop_id = $${params.length}` - } - const items = await query<{ internal_sku: string }>( - ` - SELECT DISTINCT btrim(oi.manual_sku_key) AS internal_sku - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.sku_code = $1 - AND btrim(COALESCE(oi.manual_sku_key, '')) ~ '[㐀-䶿一-鿿豈-﫿𠀀-𯨟𰀀-𲎯]' - ${scope} - LIMIT 2 - `, - params, - ) - const values = items.rows.map((row) => String(row.internal_sku || '').trim()).filter(Boolean) - return values.length === 1 ? values[0]! : '' -} - -async function main() { - const apply = process.argv.includes('--apply') - const rules = await query( - ` - SELECT id, provider, shop_id, match_key, channel, priority, source - FROM sku_channel_rules - WHERE match_key_type = 'sku_code' AND enabled = TRUE - ORDER BY id - `, - ) - - const converted: Array<{ id: number; matchKey: string; internalSku: string; channel: string }> = - [] - const unresolved: LegacyRule[] = [] - for (const rule of rules.rows) { - const internalSku = await resolveInternalSku(rule) - if (!internalSku) { - unresolved.push(rule) - continue - } - converted.push({ id: rule.id, matchKey: rule.match_key, internalSku, channel: rule.channel }) - if (!apply) continue - - await query( - `UPDATE internal_sku_channels - SET is_fallback = TRUE, - priority = GREATEST(priority, 90), - updated_at = NOW() - WHERE btrim(internal_sku) = $1 - AND channel <> $2 - AND enabled = TRUE - AND is_fallback = FALSE`, - [internalSku, rule.channel], - ) - await query( - ` - INSERT INTO internal_sku_channels ( - internal_sku, channel, priority, is_fallback, enabled, source - ) VALUES ($1, $2, $3, FALSE, TRUE, 'legacy-rule') - ON CONFLICT (internal_sku, channel) DO UPDATE SET - priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), - is_fallback = FALSE, - enabled = TRUE, - updated_at = NOW() - `, - [internalSku, rule.channel, rule.priority], - ) - await query( - `UPDATE sku_channel_rules - SET enabled = FALSE, - note = CONCAT(COALESCE(note, ''), ' [已迁移 internal_sku:', $2::text, ']'), - updated_at = NOW() - WHERE id = $1`, - [rule.id, internalSku], - ) - } - - console.log( - JSON.stringify( - { - apply, - legacyRules: rules.rows.length, - convertible: converted.length, - converted, - unresolved, - }, - null, - 2, - ), - ) -} - -main() - .catch((error) => { - console.error(error) - process.exitCode = 1 - }) - .finally(() => closeDb()) diff --git a/apps/backend/scripts/finance-cutover-v2.ts b/apps/backend/scripts/finance-cutover-v2.ts index 19a1da5c..5fdaf7dc 100644 --- a/apps/backend/scripts/finance-cutover-v2.ts +++ b/apps/backend/scripts/finance-cutover-v2.ts @@ -13,7 +13,7 @@ import { closeDb, query } from '../src/db/client.js' * 保留：orders / order_items / work_orders / fulfillment_tasks（审计只读） * product_catalog / channel_product_mappings / cost_items / * cost_versions / work_product_rules / manual_fulfillment_configs / - * sku_channel_rules（配置） + * sku_channel_routes（配置） * * 前置： * 1. 迁移 112/113 已执行； diff --git a/apps/backend/scripts/merge-duplicate-products.ts b/apps/backend/scripts/merge-duplicate-products.ts index 92abfbe1..343bfecf 100644 --- a/apps/backend/scripts/merge-duplicate-products.ts +++ b/apps/backend/scripts/merge-duplicate-products.ts @@ -70,6 +70,10 @@ async function mergeGroup( 'UPDATE manual_fulfillment_configs SET canonical_product_id = $2, updated_at = NOW() WHERE canonical_product_id = $1', [duplicateId, group.keeper_id], ) + await client.query( + 'UPDATE sku_channel_routes SET canonical_product_id = $2, updated_at = NOW() WHERE canonical_product_id = $1', + [duplicateId, group.keeper_id], + ) // 映射：先删掉会与主商品冲突的重复映射，再迁移其余。 const conflict = await client.query( @@ -136,44 +140,42 @@ async function mergeGroup( } await query( - ` - INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback, enabled, source) - VALUES ($1, 'affiliate_dash', 10, FALSE, TRUE, 'merged'), ($1, 'manual', 20, TRUE, TRUE, 'merged') - ON CONFLICT (product_id, channel) DO UPDATE - SET priority = LEAST(product_channel_capabilities.priority, EXCLUDED.priority), - is_fallback = product_channel_capabilities.is_fallback OR EXCLUDED.is_fallback, - updated_at = NOW() - `, + `UPDATE sku_channel_routes + SET route_role = 'fallback', + priority = GREATEST(priority, 90), + updated_at = NOW() + WHERE internal_sku_id = ( + SELECT internal_sku_id FROM product_catalog WHERE id = $1 + ) + AND channel <> 'affiliate_dash' + AND enabled = TRUE + AND route_role = 'primary'`, [group.keeper_id], ) - await query( - `UPDATE internal_sku_channels - SET is_fallback = TRUE, - priority = GREATEST(priority, 90), - updated_at = NOW() - WHERE btrim(internal_sku) = $1 - AND channel <> 'affiliate_dash' - AND enabled = TRUE - AND is_fallback = FALSE`, - [group.canonical_name], - ) - await query( ` - INSERT INTO internal_sku_channels ( - internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source + INSERT INTO sku_channel_routes ( + internal_sku_id, channel, route_role, priority, enabled, canonical_product_id, source ) - VALUES ($1, 'affiliate_dash', 10, FALSE, TRUE, $2, 'merged'), - ($1, 'manual', 20, TRUE, TRUE, $2, 'merged') - ON CONFLICT (internal_sku, channel) DO UPDATE SET - priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), - is_fallback = internal_sku_channels.is_fallback OR EXCLUDED.is_fallback, + SELECT product.internal_sku_id, merged.channel, merged.route_role, + merged.priority, TRUE, product.id, 'merged' + FROM product_catalog product + CROSS JOIN ( + VALUES + ('affiliate_dash', 'primary', 10), + ('manual', 'fallback', 20) + ) merged(channel, route_role, priority) + WHERE product.id = $1 + AND product.internal_sku_id IS NOT NULL + ON CONFLICT (internal_sku_id, channel) DO UPDATE SET + priority = LEAST(sku_channel_routes.priority, EXCLUDED.priority), + route_role = EXCLUDED.route_role, enabled = TRUE, - canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), + canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, sku_channel_routes.canonical_product_id), updated_at = NOW() `, - [group.canonical_name, group.keeper_id], + [group.keeper_id], ) // 只删除已经没有引用的重复商品，仍被引用就归档。 diff --git a/apps/backend/scripts/normalize-internal-sku-configs.ts b/apps/backend/scripts/normalize-internal-sku-configs.ts deleted file mode 100644 index 89523e00..00000000 --- a/apps/backend/scripts/normalize-internal-sku-configs.ts +++ /dev/null @@ -1,201 +0,0 @@ -/** - * 合并规范化后重复的内部 SKU 渠道配置。 - * - * 默认只读；确认后加 --apply。 - * 用法：npx tsx scripts/normalize-internal-sku-configs.ts [--apply] - */ -import { closeDb, query, withTransaction } from '../src/db/client.js' -import { canonicalizeInternalSku } from '../src/services/fulfillment/internal-sku-resolver.js' - -type ChannelRow = { - id: number - internal_sku: string - channel: string - priority: number - is_fallback: boolean - enabled: boolean - canonical_product_id: number | null - config_json: Record - source: string - note: string - created_by_admin_user_id: number | null -} - -type DuplicateGroup = { - internalSku: string - channel: string - rows: ChannelRow[] -} - -type MappingRow = { - id: number - provider: string - shop_id: string - match_type: string - match_value: string - internal_sku: string - priority: number - enabled: boolean - source: string - note: string -} - -async function main() { - const apply = process.argv.includes('--apply') - const result = await query( - `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 - ORDER BY canonicalize_internal_sku(internal_sku), channel, is_fallback, priority, id`, - ) - - const groups = new Map() - for (const row of result.rows) { - const internalSku = canonicalizeInternalSku(row.internal_sku) - const key = `${internalSku}|${row.channel}` - const group = groups.get(key) || { internalSku, channel: row.channel, rows: [] } - group.rows.push({ ...row, id: Number(row.id), priority: Number(row.priority) }) - groups.set(key, group) - } - - const duplicates = [...groups.values()].filter((group) => group.rows.length > 1) - const mappingRows = await query( - `SELECT id, provider, shop_id, match_type, match_value, internal_sku, - priority, enabled, source, note - FROM internal_sku_mappings - ORDER BY provider, shop_id, match_type, canonicalize_internal_sku(match_value), enabled DESC, priority, id`, - ) - const duplicateMappings = new Map() - for (const row of mappingRows.rows) { - const key = [ - row.provider, - row.shop_id, - row.match_type, - canonicalizeInternalSku(row.match_value), - ].join('|') - const list = duplicateMappings.get(key) || [] - list.push({ ...row, id: Number(row.id), priority: Number(row.priority) }) - duplicateMappings.set(key, list) - } - const mappingDuplicates = [...duplicateMappings.values()].filter((rows) => rows.length > 1) - if (apply) { - for (const group of duplicates) { - await withTransaction(async (client) => { - const rows = ( - await client.query( - `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 id = ANY($1::bigint[]) - ORDER BY is_fallback, priority, id - FOR UPDATE`, - [group.rows.map((row) => row.id)], - ) - ).rows.map((row) => ({ ...row, id: Number(row.id), priority: Number(row.priority) })) - if (rows.length <= 1) return - - const keeper = rows[0]! - const primaryRows = rows.filter((row) => !row.is_fallback) - const configJson = rows.reduce>( - (merged, row) => ({ ...merged, ...(row.config_json || {}) }), - {}, - ) - const source = rows.find((row) => row.source === 'operator')?.source || keeper.source - const note = [ - ...new Set(rows.map((row) => String(row.note || '').trim()).filter(Boolean)), - ].join(' / ') - await client.query(`DELETE FROM internal_sku_channels WHERE id = ANY($1::bigint[])`, [ - rows.slice(1).map((row) => row.id), - ]) - await client.query( - `UPDATE internal_sku_channels - SET internal_sku = $2, - priority = $3, - is_fallback = $4, - enabled = $5, - canonical_product_id = $6, - config_json = $7::jsonb, - source = $8, - note = $9, - created_by_admin_user_id = $10, - updated_at = NOW() - WHERE id = $1`, - [ - keeper.id, - group.internalSku, - Math.min(...(primaryRows.length > 0 ? primaryRows : rows).map((row) => row.priority)), - primaryRows.length === 0, - rows.some((row) => row.enabled), - rows.find((row) => row.canonical_product_id != null)?.canonical_product_id || null, - JSON.stringify(configJson), - source, - note, - rows.find((row) => row.created_by_admin_user_id != null)?.created_by_admin_user_id || - null, - ], - ) - }) - } - for (const rows of mappingDuplicates) { - await withTransaction(async (client) => { - const keeper = rows[0]! - await client.query(`DELETE FROM internal_sku_mappings WHERE id = ANY($1::bigint[])`, [ - rows.slice(1).map((row) => row.id), - ]) - await client.query( - `UPDATE internal_sku_mappings - SET match_value = $2, internal_sku = $3, priority = $4, - enabled = $5, source = $6, note = $7, updated_at = NOW() - WHERE id = $1`, - [ - keeper.id, - canonicalizeInternalSku(keeper.match_value), - canonicalizeInternalSku(keeper.internal_sku), - Math.min(...rows.map((row) => row.priority)), - rows.some((row) => row.enabled), - keeper.source, - [...new Set(rows.map((row) => String(row.note || '').trim()).filter(Boolean))].join( - ' / ', - ), - ], - ) - }) - } - } - - const duplicateManualConfigs = await query<{ label: string; total: number }>( - `SELECT canonicalize_internal_sku(sku_key) AS label, count(*)::int AS total - FROM manual_fulfillment_configs - GROUP BY canonicalize_internal_sku(sku_key) - HAVING count(*) > 1 - ORDER BY count(*) DESC, label - LIMIT 20`, - ) - - console.log( - JSON.stringify( - { - apply, - duplicateGroups: duplicates.length, - duplicateRows: duplicates.reduce((total, group) => total + group.rows.length, 0), - duplicateMappingGroups: mappingDuplicates.length, - duplicateMappingRows: mappingDuplicates.reduce((total, rows) => total + rows.length, 0), - sample: duplicates.slice(0, 20).map((group) => ({ - internalSku: group.internalSku, - channel: group.channel, - ids: group.rows.map((row) => row.id), - })), - duplicateManualConfigs: duplicateManualConfigs.rows, - }, - null, - 2, - ), - ) -} - -main() - .catch((error) => { - console.error(error) - process.exitCode = 1 - }) - .finally(() => closeDb()) diff --git a/apps/backend/scripts/preflight-fulfillment-routing.ts b/apps/backend/scripts/preflight-fulfillment-routing.ts deleted file mode 100644 index 3a7eaa91..00000000 --- a/apps/backend/scripts/preflight-fulfillment-routing.ts +++ /dev/null @@ -1,345 +0,0 @@ -/** - * 履约/财务统一分流部署后核对（只读）。 - * - * 用法： - * npx tsx scripts/preflight-fulfillment-routing.ts - * npx tsx scripts/preflight-fulfillment-routing.ts --strict - */ -import { closeDb, query } from '../src/db/client.js' - -type CountRow = { label: string; total: number } - -async function main() { - const strict = process.argv.includes('--strict') - const includeDemo = process.argv.includes('--include-demo') - const checks: Record = {} - - checks.migrations = ( - await query<{ name: string }>(`SELECT name FROM schema_migrations ORDER BY id DESC LIMIT 5`) - ).rows.map((row) => row.name) - - checks.pendingReasons = ( - await query( - `SELECT COALESCE(fulfillment_pending_reason, '(null)') AS label, count(*)::int AS total - FROM order_items WHERE fulfillment_status = 'pending' - GROUP BY 1 ORDER BY 2 DESC`, - ) - ).rows - - checks.renameCardChannels = ( - await query( - `SELECT COALESCE(fulfillment_channel, '(pending)') AS label, count(*)::int AS total - FROM order_items - WHERE sku_name = '11月改名卡' - GROUP BY 1 ORDER BY 2 DESC`, - ) - ).rows - - checks.rules = ( - await query( - `SELECT match_key_type || '/' || channel || '/' || - CASE WHEN enabled THEN 'on' ELSE 'off' END AS label, count(*)::int AS total - FROM sku_channel_rules GROUP BY 1 ORDER BY 2 DESC`, - ) - ).rows - - checks.internalChannels = ( - await query( - `SELECT channel AS label, count(*)::int AS total - FROM internal_sku_channels WHERE enabled GROUP BY 1 ORDER BY 2 DESC`, - ) - ).rows - - checks.uppercaseInternalChannels = ( - await query( - `SELECT 'uppercase_internal_sku' AS label, count(*)::int AS total - FROM internal_sku_channels - WHERE internal_sku ~ '[A-Z]'`, - ) - ).rows[0] - - checks.pendingWithoutChannel = ( - await query( - `SELECT 'pending_no_internal_channel' AS label, count(*)::int AS total - FROM order_items oi - WHERE oi.fulfillment_status = 'pending' - AND btrim(COALESCE(resolve_internal_sku( - oi.manual_sku_key, oi.sku_name, oi.sku_code - ), '')) <> '' - AND NOT EXISTS ( - SELECT 1 FROM internal_sku_channels c - WHERE c.enabled AND btrim(c.internal_sku) = resolve_internal_sku( - oi.manual_sku_key, oi.sku_name, oi.sku_code - ) - )`, - ) - ).rows[0] - - checks.assignedWithoutCarrier = ( - await query( - `SELECT 'assigned_without_carrier' AS label, count(*)::int AS total - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_status = 'assigned' - AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order') - AND ($1::boolean OR o.shop_id NOT LIKE 'sim-%') - AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled') - AND NOT EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' - AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') - )`, - [includeDemo], - ) - ).rows[0] - - checks.assignedWithMultipleCarriers = ( - await query( - `SELECT 'assigned_with_multiple_carriers' AS label, count(*)::int AS total - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_status = 'assigned' - AND ($1::boolean OR o.shop_id NOT LIKE 'sim-%') - AND (EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int - + (EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' - AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') - ))::int - + (EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' - AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') - ))::int > 1`, - [includeDemo], - ) - ).rows[0] - - checks.internalChannelConflicts = ( - await query( - `SELECT internal_sku AS label, count(*)::int AS total - FROM internal_sku_channels - WHERE enabled = TRUE AND is_fallback = FALSE - GROUP BY internal_sku - HAVING count(*) > 1 - ORDER BY count(*) DESC, internal_sku`, - ) - ).rows - - checks.canonicalChannelDuplicates = ( - await query( - `SELECT canonicalize_internal_sku(internal_sku) AS label, count(*)::int AS total - FROM internal_sku_channels - WHERE enabled = TRUE - GROUP BY canonicalize_internal_sku(internal_sku), channel - HAVING count(*) > 1 - ORDER BY count(*) DESC, label`, - ) - ).rows - - checks.incompleteCarrierConfigs = ( - await query( - `SELECT channel AS label, count(*)::int AS total - FROM internal_sku_channels - WHERE enabled = TRUE - AND ( - ( - channel = 'work_order' - AND ( - COALESCE(config_json->>'workRuleKey', '') = '' - OR NOT EXISTS ( - SELECT 1 FROM work_product_rules rule - WHERE rule.rule_key = config_json->>'workRuleKey' - AND rule.enabled = TRUE - ) - ) - ) - OR ( - channel = 'manual' - AND NOT EXISTS ( - SELECT 1 FROM manual_fulfillment_configs cfg - WHERE cfg.enabled = TRUE - AND canonicalize_internal_sku(cfg.sku_key) - = canonicalize_internal_sku(internal_sku_channels.internal_sku) - ) - ) - ) - GROUP BY channel ORDER BY count(*) DESC`, - ) - ).rows - - checks.ambiguousCarrierConfigs = ( - await query( - `SELECT canonicalize_internal_sku(canonical_name) || '/' || channel AS label, - count(DISTINCT canonical_sku)::int AS total - FROM ( - SELECT canonical_name, canonical_sku, - CASE - WHEN canonical_sku LIKE 'aff:%' THEN 'affiliate_dash' - WHEN canonical_sku LIKE 'tpl:%' OR canonical_sku LIKE 'wo:%' THEN 'work_order' - WHEN canonical_sku LIKE 'tpv:%' THEN 'third_party_voucher' - END AS channel - FROM product_catalog - WHERE status = 'active' - AND (canonical_sku LIKE 'aff:%' OR canonical_sku LIKE 'tpl:%' - OR canonical_sku LIKE 'wo:%' OR canonical_sku LIKE 'tpv:%') - ) carrier - WHERE channel IS NOT NULL - GROUP BY canonicalize_internal_sku(canonical_name), channel - HAVING count(DISTINCT canonical_sku) > 1 - ORDER BY count(DISTINCT canonical_sku) DESC, label`, - ) - ).rows - - checks.ambiguousManualConfigs = ( - await query( - `SELECT canonicalize_internal_sku(sku_key) AS label, count(*)::int AS total - FROM manual_fulfillment_configs - WHERE enabled = TRUE - GROUP BY canonicalize_internal_sku(sku_key) - HAVING count(*) > 1 - ORDER BY count(*) DESC, label`, - ) - ).rows - - checks.ambiguousInternalSkuMappings = ( - await query( - `SELECT provider || '/' || shop_id || '/' || match_type || '/' || - canonicalize_internal_sku(match_value) AS label, - count(*)::int AS total - FROM internal_sku_mappings - WHERE enabled = TRUE - GROUP BY provider, shop_id, match_type, canonicalize_internal_sku(match_value) - HAVING count(*) > 1 - ORDER BY count(*) DESC, label - LIMIT 200`, - ) - ).rows - - checks.pendingWithoutChineseInternalSku = ( - await query( - `SELECT 'pending_without_chinese_internal_sku' AS label, count(*)::int AS total - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_status = 'pending' - AND ($1::boolean OR o.shop_id NOT LIKE 'sim-%') - AND COALESCE(resolve_internal_sku( - oi.manual_sku_key, oi.sku_name, oi.sku_code - ), '') = ''`, - [includeDemo], - ) - ).rows[0] - - checks.enabledLegacyRules = ( - await query( - `SELECT match_key_type || '/' || channel AS label, count(*)::int AS total - FROM sku_channel_rules - WHERE enabled = TRUE AND match_key_type = 'sku_code' - GROUP BY 1 ORDER BY 2 DESC`, - ) - ).rows - - checks.demoPending = ( - await query( - `SELECT 'demo_pending' AS label, count(*)::int AS total - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_status = 'pending' AND o.shop_id LIKE 'sim-%'`, - ) - ).rows[0] - - checks.pendingWithoutReason = ( - await query( - `SELECT 'pending_without_reason' AS label, count(*)::int AS total - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_status = 'pending' - AND ($1::boolean OR o.shop_id NOT LIKE 'sim-%') - AND oi.fulfillment_pending_reason IS NULL`, - [includeDemo], - ) - ).rows[0] - - checks.channelCarrier = ( - await query<{ - channel: string - total: number - work_orders: number - affiliate_tasks: number - manual_tasks: number - }>( - `SELECT COALESCE(fulfillment_channel, '(pending)') AS channel, - count(*)::int AS total, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int AS work_orders, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS affiliate_tasks, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS manual_tasks - FROM order_items oi - WHERE oi.fulfillment_status = 'assigned' - GROUP BY 1 ORDER BY 2 DESC`, - ) - ).rows - - const failed: string[] = [] - const rename = checks.renameCardChannels as CountRow[] - if (rename.some((row) => row.label === 'pending')) failed.push('11月改名卡仍有 pending') - if (!includeDemo && (checks.demoPending as CountRow).total > 0) { - checks.excludedDemoPending = (checks.demoPending as CountRow).total - } - if ((checks.pendingWithoutChannel as CountRow).total > 0 && !includeDemo) { - // 本地演示店铺故意保留 pending；严格模式默认只校验非演示数据。 - const productionPending = await query( - `SELECT 'pending_no_internal_channel' AS label, count(*)::int AS total - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.fulfillment_status = 'pending' - AND o.shop_id NOT LIKE 'sim-%' - AND btrim(COALESCE(resolve_internal_sku( - oi.manual_sku_key, oi.sku_name, oi.sku_code - ), '')) <> '' - AND NOT EXISTS ( - SELECT 1 FROM internal_sku_channels c - WHERE c.enabled AND btrim(c.internal_sku) = resolve_internal_sku( - oi.manual_sku_key, oi.sku_name, oi.sku_code - ) - )`, - ) - if (productionPending.rows[0]!.total > 0) - failed.push('存在有内部 SKU 但无渠道能力的非演示 pending') - } else if ((checks.pendingWithoutChannel as CountRow).total > 0) { - failed.push('存在有内部 SKU 但无渠道能力的 pending') - } - if ((checks.assignedWithoutCarrier as CountRow).total > 0) - failed.push('存在已分配自动发货/接单渠道但没有真实载体的订单项') - if ((checks.assignedWithMultipleCarriers as CountRow).total > 0) - failed.push('存在同时挂载多个真实履约载体的已分配订单项') - if ((checks.internalChannelConflicts as CountRow[]).length > 0) - failed.push('存在多个非兜底主渠道') - if ((checks.uppercaseInternalChannels as CountRow).total > 0) - failed.push('内部 SKU 渠道配置仍含未规范化的大写英文') - if ((checks.canonicalChannelDuplicates as CountRow[]).length > 0) - failed.push('存在规范化后重复的内部 SKU 渠道配置') - if ((checks.incompleteCarrierConfigs as CountRow[]).length > 0) - failed.push('存在缺少 workRuleKey 或人工履约配置的启用渠道') - if ((checks.ambiguousCarrierConfigs as CountRow[]).length > 0) - failed.push('存在同一中文内部 SKU 的多个载体配置') - if ((checks.ambiguousManualConfigs as CountRow[]).length > 0) - failed.push('存在规范化后重复的人工履约配置') - if ((checks.ambiguousInternalSkuMappings as CountRow[]).length > 0) - failed.push('存在规范化后重复的内部 SKU 映射') - if ((checks.pendingWithoutChineseInternalSku as CountRow).total > 0) - failed.push('存在无法归一到中文内部 SKU 的 pending 订单项') - if ((checks.enabledLegacyRules as CountRow[]).length > 0) - failed.push('仍有启用的 sku_code 兼容规则，未完全收敛到内部 SKU') - if ((checks.pendingWithoutReason as CountRow).total > 0) - failed.push('存在未写入 pending 原因的订单项，请先执行履约/财务管道') - - console.log(JSON.stringify({ strict, failed, ...checks }, null, 2)) - if (strict && failed.length > 0) process.exitCode = 2 -} - -main() - .catch((error) => { - console.error(error) - process.exitCode = 1 - }) - .finally(() => closeDb()) diff --git a/apps/backend/scripts/rebuild-after-cutover.ts b/apps/backend/scripts/rebuild-after-cutover.ts index ffe534e4..f6f35bbf 100644 --- a/apps/backend/scripts/rebuild-after-cutover.ts +++ b/apps/backend/scripts/rebuild-after-cutover.ts @@ -83,10 +83,12 @@ main() async function assertSchema() { const ready = await query<{ ready: boolean }>( - `SELECT to_regclass('public.sku_channel_rules') IS NOT NULL AS ready`, + `SELECT + to_regclass('public.internal_skus') IS NOT NULL + AND to_regclass('public.sku_channel_routes') IS NOT NULL AS ready`, ) if (!ready.rows[0]?.ready) { - throw new Error('缺少 112 迁移产物；请先执行 npm run db:migrate') + throw new Error('缺少统一 SKU 路由迁移产物；请先执行 npm run db:migrate') } } diff --git a/apps/backend/scripts/reclassify-post-cutover.ts b/apps/backend/scripts/reclassify-post-cutover.ts index dcb42ac7..2c98f059 100644 --- a/apps/backend/scripts/reclassify-post-cutover.ts +++ b/apps/backend/scripts/reclassify-post-cutover.ts @@ -32,23 +32,25 @@ async function loadCandidates(limit: number): Promise { FROM order_items oi JOIN orders o ON o.id = oi.order_id JOIN LATERAL ( - SELECT c.channel - FROM internal_sku_channels c - WHERE c.enabled - AND c.channel IN ('affiliate_dash', 'manual', 'third_party_voucher') - AND btrim(c.internal_sku) = resolve_internal_sku( + SELECT route.channel + FROM sku_channel_routes route + JOIN internal_skus sku ON sku.id = route.internal_sku_id + WHERE route.enabled + AND route.route_role = 'primary' + AND route.channel IN ('affiliate_dash', 'work_order', 'manual', 'third_party_voucher') + AND sku.internal_sku_key = resolve_internal_sku( oi.manual_sku_key, oi.sku_name, oi.sku_code ) ORDER BY - c.priority ASC, - CASE c.channel + route.priority ASC, + CASE route.channel WHEN 'affiliate_dash' THEN 1 WHEN 'work_order' THEN 2 WHEN 'third_party_voucher' THEN 3 WHEN 'manual' THEN 4 ELSE 9 END, - c.channel ASC + route.channel ASC LIMIT 1 ) cap ON TRUE WHERE oi.fulfillment_status = 'assigned' diff --git a/apps/backend/scripts/seed-local-finance-demo.ts b/apps/backend/scripts/seed-local-finance-demo.ts index 070d3d1f..429e18e9 100644 --- a/apps/backend/scripts/seed-local-finance-demo.ts +++ b/apps/backend/scripts/seed-local-finance-demo.ts @@ -18,26 +18,19 @@ async function cleanup() { ) await query(`DELETE FROM sales_daily_facts WHERE shop_id = ANY($1)`, [SHOPS]) await query( - `DELETE FROM internal_sku_channels - WHERE internal_sku IN ('模拟自动发货商品', '模拟三方卡券', '模拟人工商品')`, - ) - await query( - `DELETE FROM order_items WHERE order_id IN (SELECT id FROM orders WHERE shop_id = ANY($1))`, - [SHOPS], + 'ALTER TABLE fulfillment_state_events DISABLE TRIGGER trg_fulfillment_state_events_append_only', ) + try { + await query( + `DELETE FROM order_items WHERE order_id IN (SELECT id FROM orders WHERE shop_id = ANY($1))`, + [SHOPS], + ) + } finally { + await query( + 'ALTER TABLE fulfillment_state_events ENABLE TRIGGER trg_fulfillment_state_events_append_only', + ) + } await query(`DELETE FROM orders WHERE shop_id = ANY($1)`, [SHOPS]) - await query( - `DELETE FROM sku_channel_rules - WHERE match_key IN ('SIM_AFF_1', 'SIMV001', 'SIMM001', - '模拟自动发货商品', '模拟三方卡券', '模拟人工商品')`, - ) - await query( - `DELETE FROM product_channel_capabilities - WHERE product_id IN ( - SELECT id FROM product_catalog - WHERE canonical_sku IN ('aff:SIM_AFF_1', 'man:模拟人工商品') - )`, - ) await query( `DELETE FROM channel_product_mappings WHERE canonical_product_id IN ( @@ -51,7 +44,9 @@ async function cleanup() { `DELETE FROM cost_versions WHERE canonical_product_id IN ( SELECT id FROM product_catalog - WHERE canonical_sku IN ('aff:SIM_AFF_1', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B') + WHERE canonical_sku IN ( + 'aff:SIM_AFF_1', 'man:模拟人工商品', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B' + ) )`, ) await query( @@ -64,6 +59,14 @@ async function cleanup() { 'aff:SIM_AFF_1', 'man:模拟人工商品', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B' )`, ) + await query( + `DELETE FROM internal_skus + WHERE internal_sku_key IN ( + canonicalize_internal_sku('模拟自动发货商品'), + canonicalize_internal_sku('模拟三方卡券'), + canonicalize_internal_sku('模拟人工商品') + )`, + ) } async function createOrder(shop: string, orderNo: string, amountFen = 1000) { @@ -119,11 +122,6 @@ async function main() { VALUES ($1,'product',$2,500,0,'2026-10-01T00:00:00+08:00','active','本地演示采购价',NOW(),NOW())`, [affCostItem.rows[0]!.id, affProductId], ) - await query( - `INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback, enabled, source) - VALUES ($1,'affiliate_dash',10,FALSE,TRUE,'demo'), ($1,'manual',20,TRUE,TRUE,'demo')`, - [affProductId], - ) const manualProduct = await query<{ id: number }>( `INSERT INTO product_catalog ( canonical_sku, canonical_name, category, unit_name, status, @@ -134,12 +132,6 @@ async function main() { ) RETURNING id`, ) const manualProductId = Number(manualProduct.rows[0]!.id) - await query( - `INSERT INTO product_channel_capabilities ( - product_id, channel, priority, is_fallback, enabled, source - ) VALUES ($1, 'manual', 10, FALSE, TRUE, 'demo')`, - [manualProductId], - ) const manualCostItem = await query<{ id: number }>( `SELECT id FROM cost_items WHERE code = 'sys_manual_service'`, ) @@ -154,17 +146,6 @@ async function main() { 'active', '本地演示人工成本', NOW(), NOW())`, [manualCostItem.rows[0]!.id, affProductId, manualProductId], ) - await query( - `INSERT INTO internal_sku_channels ( - internal_sku, channel, priority, is_fallback, enabled, - canonical_product_id, config_json, source, created_at, updated_at - ) VALUES - ('模拟自动发货商品', 'affiliate_dash', 10, FALSE, TRUE, $1, - '{"affiliateSku":"SIM_AFF_1"}'::jsonb, 'demo', NOW(), NOW()), - ('模拟自动发货商品', 'manual', 90, TRUE, TRUE, $1, - '{}'::jsonb, 'demo', NOW(), NOW())`, - [affProductId], - ) // 人工演示配置 await query( `INSERT INTO manual_fulfillment_configs ( @@ -176,9 +157,31 @@ async function main() { [affProductId, manualProductId], ) await query( - `INSERT INTO internal_sku_channels ( - internal_sku, channel, priority, is_fallback, enabled, source, created_at, updated_at - ) VALUES ('模拟人工商品', 'manual', 10, FALSE, TRUE, 'demo', NOW(), NOW())`, + ` + INSERT INTO sku_channel_routes ( + internal_sku_id, channel, route_role, priority, enabled, + canonical_product_id, carrier_config_json, source, created_at, updated_at + ) + SELECT sku_id, channel, route_role, priority, TRUE, product_id, config_json, 'demo', NOW(), NOW() + FROM ( + VALUES + (ensure_canonical_internal_sku('模拟自动发货商品'), 'affiliate_dash', 'primary', 10, + $1::bigint, '{"affiliateSku":"SIM_AFF_1"}'::jsonb), + (ensure_canonical_internal_sku('模拟自动发货商品'), 'manual', 'fallback', 90, + $1::bigint, '{}'::jsonb), + (ensure_canonical_internal_sku('模拟人工商品'), 'manual', 'primary', 10, + $2::bigint, '{}'::jsonb) + ) demo(sku_id, channel, route_role, priority, product_id, config_json) + ON CONFLICT (internal_sku_id, channel) DO UPDATE SET + route_role = EXCLUDED.route_role, + priority = EXCLUDED.priority, + enabled = TRUE, + canonical_product_id = EXCLUDED.canonical_product_id, + carrier_config_json = EXCLUDED.carrier_config_json, + source = EXCLUDED.source, + updated_at = NOW() + `, + [affProductId, manualProductId], ) // 1. 三方卡券：待分类，指定后自动建商品 diff --git a/apps/backend/src/db/migrations/160_switch_fulfillment_routing_to_routes.sql b/apps/backend/src/db/migrations/160_switch_fulfillment_routing_to_routes.sql new file mode 100644 index 00000000..0101504b --- /dev/null +++ b/apps/backend/src/db/migrations/160_switch_fulfillment_routing_to_routes.sql @@ -0,0 +1,433 @@ +-- 160_switch_fulfillment_routing_to_routes.sql +-- +-- 硬切换履约路由真源： +-- 1. 所有标准商品补齐 internal_sku_id； +-- 2. 运行时唯一路由真源改为 sku_channel_routes； +-- 3. 商品当前渠道不再回退 source_kind； +-- 4. 履约读模型不再读取 internal_sku_channels； +-- 5. 停止旧表触发同步，不再双写。 + +UPDATE product_catalog product +SET internal_sku_id = ensure_canonical_internal_sku(product.canonical_name), + updated_at = NOW() +WHERE product.internal_sku_id IS NULL + AND canonicalize_internal_sku(product.canonical_name) <> ''; + +CREATE OR REPLACE FUNCTION normalize_product_identity() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + NEW.source_kind := CASE + WHEN NEW.canonical_sku LIKE 'aff:%' THEN 'affiliate_dash' + WHEN NEW.canonical_sku LIKE 'tpl:%' OR NEW.canonical_sku LIKE 'wo:%' THEN 'work_order' + WHEN NEW.canonical_sku LIKE 'tpv:%' THEN 'third_party_voucher' + WHEN NEW.canonical_sku LIKE 'man:%' THEN 'manual' + ELSE COALESCE(NULLIF(NEW.source_kind, ''), 'unclassified') + END; + + IF NEW.internal_sku_id IS NULL THEN + NEW.internal_sku_id := ensure_canonical_internal_sku(NEW.canonical_name); + END IF; + + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS trg_sync_sku_channel_routes ON internal_sku_channels; +DROP FUNCTION IF EXISTS sync_sku_channel_routes_trigger(); +DROP FUNCTION IF EXISTS sync_sku_channel_routes_for_key(TEXT); + +COMMENT ON TABLE internal_sku_channels IS + '冻结的旧路由输入表；迁移 160 后运行时不读不写，仅保留历史数据供追溯'; + +CREATE OR REPLACE VIEW v_product_current AS +SELECT + product.*, + sku.internal_sku_key, + primary_route.channel AS active_primary_channel, + COALESCE(fallback_routes.channels, ARRAY[]::text[]) AS fallback_channels, + manual_config.id AS manual_config_id, + (manual_config.id IS NOT NULL) AS has_active_manual_config +FROM product_catalog product +LEFT JOIN internal_skus sku ON sku.id = product.internal_sku_id +LEFT JOIN LATERAL ( + SELECT route.channel + FROM sku_channel_routes route + WHERE route.internal_sku_id = product.internal_sku_id + AND route.enabled = TRUE + AND route.route_role = 'primary' + ORDER BY route.priority, route.channel + LIMIT 1 +) primary_route ON TRUE +LEFT JOIN LATERAL ( + SELECT ARRAY_AGG(route.channel ORDER BY route.priority, route.channel) AS channels + FROM sku_channel_routes route + WHERE route.internal_sku_id = product.internal_sku_id + AND route.enabled = TRUE + AND route.route_role = 'fallback' +) fallback_routes ON TRUE +LEFT JOIN LATERAL ( + SELECT config.id + FROM manual_fulfillment_configs config + WHERE config.enabled = TRUE + AND ( + config.canonical_product_id = product.id + OR ( + sku.internal_sku_key IS NOT NULL + AND canonicalize_internal_sku(config.sku_key) = sku.internal_sku_key + ) + ) + ORDER BY config.id + LIMIT 1 +) manual_config ON TRUE; + +COMMENT ON VIEW v_product_current IS + '标准商品统一投影：来源与当前渠道分离，当前渠道只来自 sku_channel_routes'; + +CREATE OR REPLACE VIEW v_fulfillment_item_state AS +WITH active_items AS ( + SELECT + oi.*, + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku + FROM order_items oi + WHERE oi.fulfillment_status IN ('pending', 'assigned') +), +base AS ( + SELECT + oi.id AS order_item_id, + oi.order_id, + o.id AS local_order_id, + o.provider, + o.platform, + o.shop_id, + o.shop_name, + o.platform_order_id, + o.pay_status, + o.order_status, + oi.fulfillment_status AS item_fulfillment_status, + oi.fulfillment_channel AS stored_channel, + oi.fulfillment_source, + oi.fulfillment_locked, + oi.fulfillment_pending_reason, + oi.fulfillment_pending_detail, + oi.sku_code, + oi.sku_name, + oi.quantity, + oi.revenue_amount_fen, + oi.created_at, + oi.updated_at, + oi.internal_sku, + channel_plan.channel AS configured_channel, + channel_plan.config_json AS configured_channel_config, + COALESCE(channel_stats.channel_count, 0) AS channel_count, + COALESCE(latest_task.task_id, 0) AS latest_task_id, + latest_task.executor_key AS latest_task_executor_key, + latest_task.task_status AS latest_task_status, + latest_task.result_code AS latest_task_result_code, + latest_task.result_message AS latest_task_result_message, + latest_task.updated_at AS latest_task_updated_at, + COALESCE(latest_work_order.work_order_id, 0) AS latest_work_order_id, + latest_work_order.status AS latest_work_order_status, + latest_work_order.result_code AS latest_work_order_result_code, + latest_work_order.updated_at AS latest_work_order_updated_at, + EXISTS ( + SELECT 1 + FROM fulfillment_tasks active_task + WHERE active_task.order_item_id = oi.id + AND active_task.executor_key IN ('affiliate_dash', 'manual_dispatch') + AND active_task.task_status NOT IN ( + 'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled' + ) + ) AS has_active_task, + EXISTS ( + SELECT 1 + FROM work_orders active_work_order + WHERE active_work_order.order_item_id = oi.id + AND active_work_order.status <> 'cancelled' + ) AS has_active_work_order, + EXISTS ( + SELECT 1 + FROM manual_fulfillment_configs cfg + WHERE cfg.enabled = TRUE + AND canonicalize_internal_sku(cfg.sku_key) = oi.internal_sku + ) AS has_manual_config + FROM active_items oi + JOIN orders o ON o.id = oi.order_id + LEFT JOIN LATERAL ( + SELECT route.channel, route.carrier_config_json AS config_json + 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 = oi.internal_sku + ORDER BY + (route.route_role = 'fallback'), + route.priority, + route.channel, + route.id + LIMIT 1 + ) channel_plan ON TRUE + LEFT JOIN LATERAL ( + SELECT COUNT(*)::int AS channel_count + 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 = oi.internal_sku + ) channel_stats ON TRUE + LEFT JOIN LATERAL ( + SELECT + task.id AS task_id, + task.executor_key, + task.task_status, + task.result_code, + task.result_message, + task.updated_at + FROM fulfillment_tasks task + WHERE task.order_item_id = oi.id + ORDER BY task.id DESC + LIMIT 1 + ) latest_task ON TRUE + LEFT JOIN LATERAL ( + SELECT + work_order.id AS work_order_id, + work_order.status, + ''::text AS result_code, + work_order.updated_at + FROM work_orders work_order + WHERE work_order.order_item_id = oi.id + ORDER BY work_order.id DESC + LIMIT 1 + ) latest_work_order ON TRUE +), +classified AS ( + SELECT + base.*, + COALESCE(base.stored_channel, base.configured_channel) AS intended_channel, + CASE + WHEN base.pay_status = 'refunded' + OR base.order_status IN ('refunded', 'cancelled') + OR base.latest_task_result_code = 'USER_APPLY_REFUND' + THEN TRUE + ELSE FALSE + END AS is_user_refund, + CASE + WHEN base.pay_status = 'cancelled' + OR base.order_status = 'cancelled' + OR base.latest_task_result_code = 'affiliate_dash_cancelled' + THEN TRUE + ELSE FALSE + END AS is_platform_cancelled + FROM base +), +current_attempt AS ( + SELECT + classified.*, + CASE + WHEN classified.latest_task_id > 0 + AND ( + classified.latest_work_order_id = 0 + OR classified.latest_task_updated_at >= classified.latest_work_order_updated_at + ) + THEN 'task' + WHEN classified.latest_work_order_id > 0 THEN 'work_order' + ELSE '' + END AS current_attempt_type, + CASE + WHEN classified.latest_task_id > 0 + AND ( + classified.latest_work_order_id = 0 + OR classified.latest_task_updated_at >= classified.latest_work_order_updated_at + ) + THEN classified.latest_task_id + WHEN classified.latest_work_order_id > 0 THEN classified.latest_work_order_id + ELSE 0 + END AS current_attempt_id, + CASE + WHEN classified.latest_task_id > 0 + AND ( + classified.latest_work_order_id = 0 + OR classified.latest_task_updated_at >= classified.latest_work_order_updated_at + ) + THEN classified.latest_task_status + WHEN classified.latest_work_order_id > 0 THEN classified.latest_work_order_status + ELSE '' + END AS current_attempt_status, + CASE + WHEN classified.latest_task_id > 0 + AND ( + classified.latest_work_order_id = 0 + OR classified.latest_task_updated_at >= classified.latest_work_order_updated_at + ) + THEN classified.latest_task_result_code + WHEN classified.latest_work_order_id > 0 THEN classified.latest_work_order_result_code + ELSE '' + END AS current_attempt_result_code + FROM classified +), +state_classified AS ( + SELECT + current_attempt.*, + CASE + WHEN current_attempt.intended_channel IN ('third_party_voucher', 'on_account') + THEN 'not_required' + WHEN current_attempt.intended_channel = 'manual' + AND current_attempt.has_manual_config + THEN 'ready' + WHEN current_attempt.has_active_task OR current_attempt.has_active_work_order + THEN 'ready' + WHEN current_attempt.latest_task_status IN ( + 'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled' + ) + THEN 'terminal' + WHEN current_attempt.latest_work_order_status = 'cancelled' + THEN 'terminal' + WHEN current_attempt.intended_channel IS NOT NULL + THEN 'missing' + ELSE 'missing' + END AS carrier_status, + CASE + WHEN COALESCE(current_attempt.internal_sku, '') = '' + THEN 'missing_internal_sku' + WHEN current_attempt.channel_count = 0 + THEN 'missing_channel' + ELSE 'configured' + END AS route_status + FROM current_attempt +), +issue_classified AS ( + SELECT + state_classified.*, + CASE + WHEN state_classified.is_user_refund THEN 'user_refund' + WHEN state_classified.is_platform_cancelled THEN 'platform_cancelled' + WHEN state_classified.route_status = 'missing_internal_sku' THEN 'missing_internal_sku' + WHEN state_classified.route_status = 'missing_channel' THEN 'missing_channel' + WHEN state_classified.intended_channel = 'manual' + AND NOT state_classified.has_manual_config + THEN 'manual_config_missing' + WHEN state_classified.intended_channel = 'work_order' + AND BTRIM(COALESCE( + state_classified.configured_channel_config->>'workRuleKey', + '' + )) = '' + THEN 'work_rule_missing' + WHEN state_classified.intended_channel = 'affiliate_dash' + AND BTRIM(COALESCE( + state_classified.configured_channel_config->>'affiliateSku', + '' + )) = '' + THEN 'affiliate_mapping_missing' + WHEN state_classified.carrier_status IN ('ready', 'not_required') THEN '' + WHEN state_classified.latest_task_status = 'manual_review' + THEN 'carrier_manual_review' + WHEN state_classified.latest_task_status IN ('failed', 'retry_pending') + THEN 'carrier_retry_pending' + WHEN state_classified.latest_task_status = 'expired' THEN 'carrier_expired' + WHEN state_classified.latest_task_status = 'closed' + OR state_classified.latest_work_order_status = 'cancelled' + THEN 'carrier_closed' + WHEN state_classified.item_fulfillment_status = 'assigned' + THEN 'state_inconsistent' + WHEN state_classified.intended_channel IN ( + 'affiliate_dash', 'work_order', 'manual' + ) + THEN 'carrier_missing' + ELSE 'state_inconsistent' + END AS issue_code + FROM state_classified +), +categorized AS ( + SELECT + issue_classified.*, + CASE + WHEN issue_code = '' THEN '' + WHEN issue_code IN ('user_refund', 'platform_cancelled') THEN 'after_sales' + WHEN issue_code = 'carrier_manual_review' THEN 'manual_review' + WHEN issue_code IN ('missing_internal_sku', 'missing_channel') THEN 'classification' + WHEN issue_code IN ( + 'manual_config_missing', 'work_rule_missing', 'affiliate_mapping_missing' + ) THEN 'configuration' + WHEN issue_code = 'state_inconsistent' THEN 'system' + ELSE 'carrier' + END AS issue_category + FROM issue_classified +) +SELECT + order_item_id, + order_id, + local_order_id, + provider, + platform, + shop_id, + shop_name, + platform_order_id, + pay_status, + order_status, + item_fulfillment_status AS fulfillment_status, + internal_sku, + intended_channel, + stored_channel, + configured_channel, + route_status, + carrier_status, + current_attempt_type, + current_attempt_id, + current_attempt_status, + current_attempt_result_code, + latest_task_id, + latest_task_executor_key, + latest_task_status, + latest_task_result_code, + latest_work_order_id, + latest_work_order_status, + issue_code, + issue_category, + CASE + WHEN issue_code = '' THEN FALSE + WHEN issue_code IN ( + 'user_refund', 'platform_cancelled', 'carrier_manual_review', + 'manual_config_missing', 'work_rule_missing', 'affiliate_mapping_missing', + 'missing_internal_sku', 'missing_channel' + ) THEN FALSE + ELSE TRUE + END AS retryable, + CASE + WHEN issue_code IN ('carrier_missing', 'state_inconsistent') THEN TRUE + ELSE FALSE + END AS auto_retryable, + CASE + WHEN issue_code IN ('user_refund', 'platform_cancelled', 'carrier_manual_review') + THEN 'support' + WHEN issue_code IN ( + 'missing_internal_sku', 'missing_channel', 'manual_config_missing', + 'work_rule_missing', 'affiliate_mapping_missing' + ) THEN 'operator' + WHEN issue_code = '' THEN '' + ELSE 'system' + END AS owner_role, + fulfillment_locked, + CASE + WHEN NOT fulfillment_locked THEN '' + ELSE COALESCE( + NULLIF(fulfillment_pending_detail->>'carrierIssue', ''), + NULLIF(fulfillment_pending_detail->>'reason', ''), + 'operator_assigned' + ) + END AS lock_reason, + fulfillment_pending_reason, + fulfillment_pending_detail, + fulfillment_source, + sku_code, + sku_name, + quantity, + revenue_amount_fen, + created_at, + updated_at, + GREATEST( + 0, + FLOOR(EXTRACT(EPOCH FROM (NOW() - created_at)) / 60) + )::bigint AS age_minutes +FROM categorized; + +COMMENT ON VIEW v_fulfillment_item_state IS + '履约统一读模型：每个订单项一行，仅读取 sku_channel_routes 的当前路由'; diff --git a/apps/backend/src/db/migrations/161_move_shadow_sync_to_sku_routes.sql b/apps/backend/src/db/migrations/161_move_shadow_sync_to_sku_routes.sql new file mode 100644 index 00000000..3b592402 --- /dev/null +++ b/apps/backend/src/db/migrations/161_move_shadow_sync_to_sku_routes.sql @@ -0,0 +1,46 @@ +-- 161_move_shadow_sync_to_sku_routes.sql +-- +-- 路由唯一真源已经切换到 sku_channel_routes。 +-- 履约影子模型必须跟随新表变化，旧表不再拥有任何同步触发器。 + +DROP TRIGGER IF EXISTS trg_internal_sku_channel_shadow ON internal_sku_channels; +DROP FUNCTION IF EXISTS sync_fulfillment_shadow_for_internal_channel(); + +CREATE OR REPLACE FUNCTION sync_fulfillment_shadow_for_sku_route() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + old_key TEXT; + new_key TEXT; +BEGIN + IF TG_OP <> 'INSERT' THEN + SELECT internal_sku_key INTO old_key + FROM internal_skus + WHERE id = OLD.internal_sku_id; + END IF; + IF TG_OP <> 'DELETE' THEN + SELECT internal_sku_key INTO new_key + FROM internal_skus + WHERE id = NEW.internal_sku_id; + END IF; + + PERFORM sync_fulfillment_shadow(item.id) + FROM order_items item + WHERE item.fulfillment_status IN ('pending', 'assigned') + AND canonicalize_internal_sku( + resolve_internal_sku(item.manual_sku_key, item.sku_name, item.sku_code) + ) IN (old_key, new_key); + + RETURN COALESCE(NEW, OLD); +END; +$$; + +DROP TRIGGER IF EXISTS trg_sku_route_fulfillment_shadow ON sku_channel_routes; +CREATE TRIGGER trg_sku_route_fulfillment_shadow +AFTER INSERT OR UPDATE OR DELETE ON sku_channel_routes +FOR EACH ROW +EXECUTE FUNCTION sync_fulfillment_shadow_for_sku_route(); + +COMMENT ON FUNCTION sync_fulfillment_shadow_for_sku_route() IS + 'sku_channel_routes 变化后刷新同内部 SKU 的履约影子'; diff --git a/apps/backend/src/db/migrations/162_drop_legacy_routing_read_surfaces.sql b/apps/backend/src/db/migrations/162_drop_legacy_routing_read_surfaces.sql new file mode 100644 index 00000000..c090e79d --- /dev/null +++ b/apps/backend/src/db/migrations/162_drop_legacy_routing_read_surfaces.sql @@ -0,0 +1,12 @@ +-- 162_drop_legacy_routing_read_surfaces.sql +-- +-- 160/161 已把业务读模型和影子同步切到 sku_channel_routes。 +-- 这里删除最后一个旧表只读视图，以及旧表最后一个规范化触发器。 + +DROP VIEW IF EXISTS v_fulfillment_item_state_raw; + +DROP TRIGGER IF EXISTS trg_normalize_internal_sku_channel_key ON internal_sku_channels; +DROP FUNCTION IF EXISTS normalize_internal_sku_channel_key(); + +COMMENT ON TABLE internal_sku_channels IS + '冻结的历史路由输入；无触发器、无运行时读取、无运行时写入'; diff --git a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts index dcfab3f0..4a49eee6 100644 --- a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts +++ b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts @@ -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(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 diff --git a/apps/backend/src/services/cost-center/product-catalog-service.ts b/apps/backend/src/services/cost-center/product-catalog-service.ts index 4514d546..1aeef2ac 100644 --- a/apps/backend/src/services/cost-center/product-catalog-service.ts +++ b/apps/backend/src/services/cost-center/product-catalog-service.ts @@ -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], diff --git a/apps/backend/src/services/cost-center/work-order-template-cost.integration.test.ts b/apps/backend/src/services/cost-center/work-order-template-cost.integration.test.ts index 225c1531..fdeb3647 100644 --- a/apps/backend/src/services/cost-center/work-order-template-cost.integration.test.ts +++ b/apps/backend/src/services/cost-center/work-order-template-cost.integration.test.ts @@ -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]) diff --git a/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts b/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts index a9928ac9..b139fe5d 100644 --- a/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts +++ b/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts @@ -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 ) ) diff --git a/apps/backend/src/services/fulfillment/carrier-creation-service.ts b/apps/backend/src/services/fulfillment/carrier-creation-service.ts index 5a1e2bf0..bc1a790e 100644 --- a/apps/backend/src/services/fulfillment/carrier-creation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-creation-service.ts @@ -257,9 +257,12 @@ export async function ensureCarrierForOrderItem(input: { }), ) const channelConfig = await query<{ config_json: Record }>( - `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], ) diff --git a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts index 55d53cb8..2635427d 100644 --- a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts @@ -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 明细，不再静默卡住。 diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index e317faac..ab410257 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -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 { 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( ` - 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], ) diff --git a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts index 42f659c2..1e1f626b 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts @@ -128,6 +128,59 @@ test( return orderItemId } + async function setRoute(input: { + sku: string + channel: string + priority?: number + role?: 'primary' | 'fallback' + configJson?: Record + 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 }>( - `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 diff --git a/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts b/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts index 6c1df388..ee9b0051 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts @@ -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('人工履约配置不能关联其他来源商品', () => { diff --git a/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts b/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts index ee92a616..01532f49 100644 --- a/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts +++ b/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts @@ -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, diff --git a/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts b/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts index 0c949b5f..ac728c0a 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts @@ -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')) +}) diff --git a/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts b/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts index 227cccc4..6d01d188 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts @@ -32,13 +32,23 @@ export async function listInternalSkuChannels( if (!key) return [] const result = await query( ` - 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( ` - 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 { 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 { 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 { 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)) diff --git a/apps/backend/src/services/fulfillment/internal-sku-routing.integration.test.ts b/apps/backend/src/services/fulfillment/internal-sku-routing.integration.test.ts index 5f19a2b3..37f236c2 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-routing.integration.test.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-routing.integration.test.ts @@ -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], ), diff --git a/apps/backend/src/services/fulfillment/pending-channel-repair-service.ts b/apps/backend/src/services/fulfillment/pending-channel-repair-service.ts index 599c0ce0..1a890fb4 100644 --- a/apps/backend/src/services/fulfillment/pending-channel-repair-service.ts +++ b/apps/backend/src/services/fulfillment/pending-channel-repair-service.ts @@ -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} diff --git a/apps/backend/src/services/fulfillment/product-resolution-service.ts b/apps/backend/src/services/fulfillment/product-resolution-service.ts index 8dc36213..59a0c637 100644 --- a/apps/backend/src/services/fulfillment/product-resolution-service.ts +++ b/apps/backend/src/services/fulfillment/product-resolution-service.ts @@ -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, diff --git a/docs/统一订单履约与财务模型-权威版.md b/docs/统一订单履约与财务模型-权威版.md index 62b5f802..bc1b15f3 100644 --- a/docs/统一订单履约与财务模型-权威版.md +++ b/docs/统一订单履约与财务模型-权威版.md @@ -12,7 +12,7 @@ updated_at: 2026-11-08 > 状态：**唯一权威文档**。 > 最后校准：2026-11-08。 > 生产基线：迁移 `146`。 -> 开发基线：迁移 `159`（仅本地，尚未部署生产）。 +> 开发基线：迁移 `162`（仅本地，尚未部署生产）。 > > 从本文件生效起，订单、内部 SKU、履约渠道、履约载体、标准商品、成本和财务相关实现， > 只能以本文件为业务口径。其他旧分析只作为历史资料，不再具备设计或验收效力。 @@ -311,7 +311,7 @@ finance_journal_* -- 总账凭证，可重建 2026-11-08 生产只读核对结果： -- 生产迁移 `146`，本地开发基线迁移 `159`。 +- 生产迁移 `146`，本地开发基线迁移 `162`。 - 启用标准商品 `1127`：`aff:` 30、`tpl:` 903、`tpv:` 46、`man:` 148。 - `man:` 商品中只有 105 个当前主渠道为人工；41 个已走其他主渠道，2 个无主渠道。 - 人工页头部显示 `107`，商品列表显示 `148`，相差的 41 个就是跨渠道旧商品。 @@ -357,7 +357,7 @@ A 级数据只能追加新事件或做字段兼容迁移，不能为了“模型 ### B 级：业务输入，必须迁移而不是重建 - `internal_sku_mappings` -- `internal_sku_channels` 中仍有效的路由意图 +- `sku_channel_routes` 中的有效主渠道与显式兜底 - `work_product_rules` 及模板匹配配置 - `manual_fulfillment_configs` - `product_catalog` 中仍有效的标准商品 @@ -383,6 +383,8 @@ A 级数据只能追加新事件或做字段兼容迁移，不能为了“模型 ### D 级：兼容期结束后删除 - `product_channel_capabilities` +- 已冻结的 `internal_sku_channels` +- `sku_channel_rules` 中由 `internal_sku_mappings` 和 `sku_channel_routes` 取代的规则 - `manual_sku_registry` 中可由配置重建的登记行 - 订单项上的重复履约结果字段，在正式模型切换后删除 - 商品表中的历史来源前缀判断代码 @@ -504,9 +506,9 @@ has_active_cost | --- | --- | --- | | 0 口径冻结与文档收口 | 已完成 | 否 | | 1 只读审计与领域约束 | 本地进行中 | 否 | -| 2 建立目标身份与投影 | 本地完成 | 只新增兼容结构 | -| 3 迁移路由与商品配置 | 本地进行中 | 是，需备份 | -| 4 重建履约模型 | 未开始 | 是，需断代 | +| 2 建立目标身份与投影 | 商品侧本地完成 | 只新增兼容结构 | +| 3 迁移路由与商品配置 | 本地完成，生产未执行 | 是，需备份 | +| 4 重建履约模型 | 本地准备中 | 是，需断代 | | 5 重建财务派生数据 | 未开始 | 是，需备份与对账 | | 6 停止旧写路径并删除兼容层 | 未开始 | 是，需观察期结束 | @@ -575,17 +577,25 @@ assigned 且 carrier_status=missing 必须进入异常队列 - 迁移 `157`：保证只有兜底配置的 SKU 不会被误升级成主渠道。 - 迁移 `158`：禁止 `man:` 来源直接冒充人工主渠道；没有 primary route 时显示“未路由”。 - 迁移 `159`：只归档无主路由、无人工配置、无启用映射的孤儿 `man:` 商品。 +- 迁移 `160`：商品身份全部补齐，运行硬切到 `sku_channel_routes`，删除旧同步触发器； + `v_product_current` 不再回退 `source_kind`，`v_fulfillment_item_state` 不再读取旧表。 +- 迁移 `161`：履约影子同步触发器从旧表迁移到 `sku_channel_routes`。 +- 迁移 `162`：删除旧履约只读视图和旧表最后一个触发器。 - 商品列表、总览、渠道映射和商品下拉统一读取商品当前渠道。 - 运行期代码停止写入 `product_channel_capabilities`，该表仅保留到兼容期结束。 - 只读审计：`npm run audit:product-routing -- --strict`。 - 本地严格审计关键异常归零：孤儿人工商品、多主路由、`man/tpv` 缺内部 SKU、非法来源。 -- 15 个自动/卡券/接单主渠道商品的人工配置均保留为显式 `fallback`，没有被误升为人工主渠道。 +- 自动/卡券/接单主渠道商品的人工配置均保留为显式 `fallback`，没有被误升为人工主渠道。 - 1 个无主路由、无人工配置、无启用映射的 `man:` 商品已归档，未删除历史引用。 +- 旧 `internal_sku_channels` 已冻结：运行时不再读、不再写，也不再有视图或触发器依赖。 +- 空库从迁移 `001` 顺序执行到 `162` 已通过；新库只保留 + `sku_channel_routes` 影子同步触发器，旧表同步触发器不存在。 +- 无主路由的活动商品会进入“未路由”，不自动猜测渠道。 当前本地仍需业务处理的数据： -- 1 个待分类订单项尚未命中内部 SKU 映射； -- 1 个已分配人工订单项没有启用的人工履约配置。 +- 待分类订单项尚未命中内部 SKU 映射的，进入 `missing_channel`； +- 已分配人工订单项缺少启用人工配置的，进入 `manual_config_missing`。 验收： @@ -595,6 +605,8 @@ assigned 且 carrier_status=missing 必须进入异常队列 ### 阶段 3：迁移路由与商品配置 +状态：本地完成，生产未执行。按确认后的硬切换方案，不保留双读、双写和旧表运行期兼容。 + 目标： - 将 `internal_sku_channels` 迁移到 `sku_channel_routes`。 @@ -613,6 +625,8 @@ assigned 且 carrier_status=missing 必须进入异常队列 - 路由迁移前后主渠道结果一致。 - 人工页数量等于当前人工渠道商品数。 - 卡券和统一结算不再出现在人工页。 +- 运行时 `internal_sku_channels` 读写点为 `0`。 +- 旧路由表和影子同步触发器已停用，后续只通过 `sku_channel_routes` 修改路由。 ### 阶段 4：重建履约模型