From 2f87445ed95feb150e0ab5244e19eff4e8850e09 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Mon, 2 Nov 2026 15:01:50 +0800 Subject: [PATCH] =?UTF-8?q?fix(finance):=20=E6=94=B6=E6=95=9B=E5=86=85?= =?UTF-8?q?=E9=83=A8=20SKU=20=E5=B1=A5=E7=BA=A6=E8=B7=AF=E7=94=B1=E4=B8=8E?= =?UTF-8?q?=E8=BD=BD=E4=BD=93=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 统一内部 SKU 与渠道主/兜底解析，禁止并列主渠道。 运营指定渠道改为先创建真实载体，失败保持明确 pending。 历史缺载体、多载体冲突和自动发货兜底统一由管道修复。 迁移人工配置与旧 sku_code 规则，补充严格预检和集成测试。 --- README.md | 2 +- .../audit-fulfillment-finance-alignment.ts | 69 ++- .../scripts/backfill-internal-sku-mappings.ts | 2 +- .../scripts/cleanup-legacy-sku-code-rules.ts | 12 + .../scripts/merge-duplicate-products.ts | 13 + .../scripts/preflight-fulfillment-routing.ts | 123 +++- .../scripts/reclassify-post-cutover.ts | 6 +- .../scripts/seed-local-finance-demo.ts | 40 +- .../120_fulfillment_routing_invariants.sql | 225 +++++++ .../121_internal_sku_single_primary.sql | 25 + .../manual-fulfillment-config-service.ts | 50 +- .../cost-center/product-normalize-service.ts | 15 - .../fulfillment/affiliate-fallback-service.ts | 25 +- .../fulfillment/carrier-creation-service.ts | 202 ++++++ .../carrier-reconciliation-service.ts | 440 ++++++++----- .../channel-assign-service.test.ts | 66 +- .../fulfillment/channel-assign-service.ts | 576 +++++++++++++----- .../src/services/fulfillment/channel-types.ts | 1 + .../fulfillment-routing.integration.test.ts | 136 +++++ .../intake-channel-decision-service.ts | 61 +- .../internal-sku-channel-service.test.ts | 62 ++ .../internal-sku-channel-service.ts | 175 +++++- .../internal-sku-mapping-service.ts | 5 +- .../fulfillment/internal-sku-resolver.test.ts | 50 ++ .../fulfillment/internal-sku-resolver.ts | 61 ++ .../fulfillment/product-resolution-service.ts | 15 +- .../scheduler/finance-pipeline-jobs.ts | 84 ++- .../sync-worker-orders-service.ts | 27 +- apps/backend/src/types/repository/rows.ts | 4 + .../admin/AdminFulfillmentChannelPage.tsx | 9 +- .../src/services/admin/fulfillment-channel.ts | 10 +- docs/履约与财务分流统一方案-2026-11-02.md | 22 +- docs/生产部署核对清单-2026-11-02.md | 32 +- 33 files changed, 2199 insertions(+), 446 deletions(-) create mode 100644 apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql create mode 100644 apps/backend/src/db/migrations/121_internal_sku_single_primary.sql create mode 100644 apps/backend/src/services/fulfillment/carrier-creation-service.ts create mode 100644 apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts create mode 100644 apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts create mode 100644 apps/backend/src/services/fulfillment/internal-sku-resolver.test.ts create mode 100644 apps/backend/src/services/fulfillment/internal-sku-resolver.ts diff --git a/README.md b/README.md index 334a6109..525ed77a 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ SKIP_CHECKS=1 git push | 文件 | 含义 | | --- | --- | | `001_init.sql` | 空库基线，建立完整业务结构 | -| `002_xxx.sql` ~ `080_xxx.sql` | 后续结构变更的增量 SQL（接单平台 004~009，寄卖相关 077~080） | +| `002_xxx.sql` ~ `121_xxx.sql` | 后续结构与数据增量（接单平台、寄卖、财务、履约路由迁移） | 约定： diff --git a/apps/backend/scripts/audit-fulfillment-finance-alignment.ts b/apps/backend/scripts/audit-fulfillment-finance-alignment.ts index 4d536538..e005c2b9 100644 --- a/apps/backend/scripts/audit-fulfillment-finance-alignment.ts +++ b/apps/backend/scripts/audit-fulfillment-finance-alignment.ts @@ -34,6 +34,7 @@ async function main() { work_orders: number affiliate_tasks: number manual_tasks: number + config_missing: number no_carrier: number }>( ` @@ -44,8 +45,19 @@ async function main() { count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'))::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'))::int AS manual_tasks, count(*) FILTER ( - WHERE NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) - AND NOT EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) + WHERE oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL + )::int AS config_missing, + count(*) FILTER ( + WHERE ( + oi.fulfillment_channel = 'affiliate_dash' + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ) + ) OR ( + oi.fulfillment_channel = 'work_order' + AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + ) )::int AS no_carrier FROM order_items oi WHERE oi.fulfillment_status = 'assigned' @@ -63,9 +75,15 @@ async function main() { ` SELECT count(*)::int AS total, - count(*) FILTER (WHERE btrim(COALESCE(manual_sku_key, '')) = '')::int AS null_or_blank, - count(*) FILTER (WHERE btrim(COALESCE(manual_sku_key, '')) ~ '^[0-9]+$')::int AS numeric_only, - count(*) FILTER (WHERE btrim(COALESCE(manual_sku_key, '')) ~ '[一-龥]')::int AS with_cjk + count(*) FILTER (WHERE btrim(COALESCE(resolve_internal_sku( + manual_sku_key, sku_name, sku_code + ), '')) = '')::int AS null_or_blank, + count(*) FILTER (WHERE btrim(COALESCE(resolve_internal_sku( + manual_sku_key, sku_name, sku_code + ), '')) ~ '^[0-9]+$')::int AS numeric_only, + count(*) FILTER (WHERE btrim(COALESCE(resolve_internal_sku( + manual_sku_key, sku_name, sku_code + ), '')) ~ '[一-龥]')::int AS with_cjk FROM order_items WHERE fulfillment_status = 'assigned' `, @@ -104,6 +122,45 @@ async function main() { `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 assignedCarrierConflicts = await query<{ + order_item_id: number + channel: string + carrier_count: number + }>( + `SELECT oi.id AS order_item_id, oi.fulfillment_channel AS channel, + (EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int + + (EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ))::int + + (EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + ))::int AS carrier_count + FROM order_items oi + WHERE oi.fulfillment_status = 'assigned' + AND (EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int + + (EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ))::int + + (EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + ))::int > 1 + ORDER BY oi.id + LIMIT 200`, + ) + const byProvider = await query<{ provider: string total: number @@ -135,6 +192,8 @@ async function main() { rules: rules.rows, duplicateProducts: duplicateProducts.rows[0], capabilities: capabilities.rows, + internalChannelConflicts: internalChannelConflicts.rows, + assignedCarrierConflicts: assignedCarrierConflicts.rows, byProvider: byProvider.rows, }, null, diff --git a/apps/backend/scripts/backfill-internal-sku-mappings.ts b/apps/backend/scripts/backfill-internal-sku-mappings.ts index 044a6ed2..5285ff7e 100644 --- a/apps/backend/scripts/backfill-internal-sku-mappings.ts +++ b/apps/backend/scripts/backfill-internal-sku-mappings.ts @@ -59,7 +59,7 @@ async function main() { note: `从 ${row.total} 条订单项回填`, }) } - written += 1 + if (apply) written += 1 } console.log( diff --git a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts index 918e645e..fc8ebda0 100644 --- a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts +++ b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts @@ -88,6 +88,17 @@ async function main() { `, [internalSku, rule.channel, rule.priority, rule.source || 'operator'], ) + 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 ( @@ -95,6 +106,7 @@ async function main() { ) 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() `, diff --git a/apps/backend/scripts/merge-duplicate-products.ts b/apps/backend/scripts/merge-duplicate-products.ts index ab7ab141..891cb5b6 100644 --- a/apps/backend/scripts/merge-duplicate-products.ts +++ b/apps/backend/scripts/merge-duplicate-products.ts @@ -147,6 +147,18 @@ async function mergeGroup( [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 ( @@ -157,6 +169,7 @@ async function mergeGroup( 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, + enabled = TRUE, canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), updated_at = NOW() `, diff --git a/apps/backend/scripts/preflight-fulfillment-routing.ts b/apps/backend/scripts/preflight-fulfillment-routing.ts index 10ed333a..963ec15e 100644 --- a/apps/backend/scripts/preflight-fulfillment-routing.ts +++ b/apps/backend/scripts/preflight-fulfillment-routing.ts @@ -11,6 +11,7 @@ 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 = ( @@ -54,14 +55,96 @@ async function main() { `SELECT 'pending_no_internal_channel' AS label, count(*)::int AS total FROM order_items oi WHERE oi.fulfillment_status = 'pending' - AND btrim(COALESCE(oi.manual_sku_key, '')) <> '' + 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) = btrim(oi.manual_sku_key) + 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 NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + )`, + [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))::int + + (EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ))::int + + (EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + ))::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.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 @@ -84,8 +167,42 @@ async function main() { const failed: string[] = [] const rename = checks.renameCardChannels as CountRow[] if (rename.some((row) => row.label === 'pending')) failed.push('11月改名卡仍有 pending') - if ((checks.pendingWithoutChannel as CountRow).total > 0) + 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.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 diff --git a/apps/backend/scripts/reclassify-post-cutover.ts b/apps/backend/scripts/reclassify-post-cutover.ts index 2c32f19c..c79323d5 100644 --- a/apps/backend/scripts/reclassify-post-cutover.ts +++ b/apps/backend/scripts/reclassify-post-cutover.ts @@ -36,7 +36,9 @@ async function loadCandidates(limit: number): Promise { FROM internal_sku_channels c WHERE c.enabled AND c.channel IN ('affiliate_dash', 'manual', 'third_party_voucher') - AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name)) + AND btrim(c.internal_sku) = resolve_internal_sku( + oi.manual_sku_key, oi.sku_name, oi.sku_code + ) ORDER BY c.priority ASC, CASE c.channel @@ -68,7 +70,7 @@ async function loadCandidates(limit: number): Promise { ) AND NOT EXISTS ( SELECT 1 FROM work_orders wo - WHERE wo.order_item_id = oi.id AND wo.status = 'accepted' + WHERE wo.order_item_id = oi.id ) AND NOT EXISTS ( SELECT 1 FROM cost_snapshots cs diff --git a/apps/backend/scripts/seed-local-finance-demo.ts b/apps/backend/scripts/seed-local-finance-demo.ts index 288f45a5..a683cc17 100644 --- a/apps/backend/scripts/seed-local-finance-demo.ts +++ b/apps/backend/scripts/seed-local-finance-demo.ts @@ -17,24 +17,41 @@ async function cleanup() { [SHOPS], ) 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], ) 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')`, + `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 = 'aff:SIM_AFF_1')`, ) await query( - `DELETE FROM cost_versions WHERE canonical_product_id IN (SELECT id FROM product_catalog WHERE canonical_sku = 'aff:SIM_AFF_1')`, + `DELETE FROM channel_product_mappings + WHERE canonical_product_id IN ( + SELECT id FROM product_catalog + WHERE canonical_sku IN ('aff:SIM_AFF_1', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B') + )`, + ) + await query( + `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') + )`, ) - await query(`DELETE FROM product_catalog WHERE canonical_sku = 'aff:SIM_AFF_1'`) await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = '模拟人工商品'`) await query( - `DELETE FROM product_catalog WHERE canonical_sku IN ('tpv:模拟三方卡券', 'tpv:模拟三方卡券B')`, + `DELETE FROM product_catalog + WHERE canonical_sku IN ('aff:SIM_AFF_1', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B')`, ) } @@ -97,15 +114,24 @@ async function main() { [affProductId], ) await query( - `INSERT INTO sku_channel_rules (provider, shop_id, match_key_type, match_key, channel, priority, source, enabled, note, created_at, updated_at) - VALUES ('','','sku_code','SIM_AFF_1','affiliate_dash',60,'operator',TRUE,'本地演示',NOW(),NOW())`, + `INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, source, created_at, updated_at + ) VALUES + ('模拟自动发货商品', 'affiliate_dash', 10, FALSE, TRUE, $1, 'demo', NOW(), NOW()), + ('模拟自动发货商品', 'manual', 90, TRUE, TRUE, $1, 'demo', NOW(), NOW())`, + [affProductId], ) - // 人工演示配置 await query( `INSERT INTO manual_fulfillment_configs (sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at) VALUES ('模拟人工商品','nanhang',100,'fixed',TRUE,'本地演示',NOW(),NOW())`, ) + 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())`, + ) // 1. 三方卡券：待分类，指定后自动建商品 const voucherOrder = await createOrder('sim-voucher', 'SIM-VOUCHER-1', 1950) diff --git a/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql b/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql new file mode 100644 index 00000000..a9d24a01 --- /dev/null +++ b/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql @@ -0,0 +1,225 @@ +-- 120_fulfillment_routing_invariants.sql +-- +-- 履约路由唯一口径与数据清理： +-- 1. 订单项内部 SKU 由一个不可变 SQL 函数统一解析； +-- 2. 人工配置自动成为 internal_sku_channels 的 manual 能力； +-- 3. manual 与其他渠道并存时统一降为兜底，避免同优先级被静默抢占； +-- 4. 可确定的旧 sku_code 规则迁移为内部 SKU 渠道并停用旧规则。 + +CREATE OR REPLACE FUNCTION resolve_internal_sku( + p_manual_sku_key TEXT, + p_sku_name TEXT, + p_sku_code TEXT +) RETURNS TEXT +LANGUAGE SQL +IMMUTABLE +PARALLEL SAFE +AS $$ + SELECT COALESCE( + CASE + WHEN BTRIM(COALESCE(p_manual_sku_key, '')) ~ '[一-龥]' + THEN BTRIM(p_manual_sku_key) + END, + CASE + WHEN BTRIM(COALESCE(p_sku_name, '')) ~ '[一-龥]' + THEN BTRIM(p_sku_name) + END, + NULLIF(BTRIM(COALESCE(p_manual_sku_key, '')), ''), + NULLIF(BTRIM(COALESCE(p_sku_name, '')), ''), + NULLIF(BTRIM(COALESCE(p_sku_code, '')), '') + ) +$$; + +COMMENT ON FUNCTION resolve_internal_sku(TEXT, TEXT, TEXT) IS + '订单项内部 SKU 唯一 SQL 口径：中文 manual_sku_key/sku_name 优先，再按历史生成列顺序回退'; + +-- 人工配置本身就是一个 manual 渠道声明。存在其他渠道时默认作为 90 优先级兜底。 +INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, config_json, source, created_at, updated_at +) +SELECT + BTRIM(cfg.sku_key), + 'manual', + CASE WHEN EXISTS ( + SELECT 1 + FROM internal_sku_channels c + WHERE BTRIM(c.internal_sku) = BTRIM(cfg.sku_key) + AND c.enabled = TRUE + AND c.channel <> 'manual' + ) THEN 90 ELSE 10 END, + CASE WHEN EXISTS ( + SELECT 1 + FROM internal_sku_channels c + WHERE BTRIM(c.internal_sku) = BTRIM(cfg.sku_key) + AND c.enabled = TRUE + AND c.channel <> 'manual' + ) THEN TRUE ELSE FALSE END, + cfg.enabled, + cfg.canonical_product_id, + jsonb_build_object( + 'manualConfigId', cfg.id, + 'platformKey', cfg.platform_key, + 'creditQuantity', cfg.credit_quantity, + 'costMode', cfg.cost_mode + ), + 'manual_config', + NOW(), + NOW() +FROM manual_fulfillment_configs cfg +WHERE BTRIM(COALESCE(cfg.sku_key, '')) <> '' +ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = CASE + WHEN internal_sku_channels.source = 'operator' THEN internal_sku_channels.priority + ELSE EXCLUDED.priority + END, + is_fallback = CASE + WHEN internal_sku_channels.source = 'operator' THEN internal_sku_channels.is_fallback + ELSE EXCLUDED.is_fallback + END, + 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, + updated_at = NOW(); + +-- 只要同一内部 SKU 还有自动发货/接单/卡券能力，人工渠道就不能作为并列主渠道。 +UPDATE internal_sku_channels manual_channel +SET priority = GREATEST(manual_channel.priority, 90), + is_fallback = TRUE, + updated_at = NOW() +WHERE manual_channel.channel = 'manual' + AND manual_channel.enabled = TRUE + AND NOT (manual_channel.source = 'operator' AND manual_channel.priority = 10) + AND EXISTS ( + SELECT 1 + FROM internal_sku_channels other_channel + WHERE BTRIM(other_channel.internal_sku) = BTRIM(manual_channel.internal_sku) + AND other_channel.enabled = TRUE + AND other_channel.channel <> 'manual' + ); + +-- 自动发货旧 sku_code 规则能通过 aff: 商品稳定定位内部 SKU，直接迁移。 +INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, config_json, source, created_at, updated_at +) +SELECT + BTRIM(product.canonical_name), + 'affiliate_dash', + MIN(rule.priority), + FALSE, + TRUE, + MIN(product.id), + jsonb_build_object('legacySkuCode', MIN(rule.match_key)), + 'legacy-rule', + NOW(), + NOW() +FROM sku_channel_rules rule +JOIN product_catalog product + ON product.canonical_sku = CONCAT('aff:', rule.match_key) + AND product.status = 'active' +WHERE rule.match_key_type = 'sku_code' + AND rule.channel = 'affiliate_dash' + AND rule.enabled = TRUE + AND BTRIM(COALESCE(product.canonical_name, '')) <> '' +GROUP BY BTRIM(product.canonical_name) +ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + enabled = TRUE, + canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), + config_json = internal_sku_channels.config_json || EXCLUDED.config_json, + updated_at = NOW(); + +UPDATE sku_channel_rules rule +SET enabled = FALSE, + note = CONCAT(COALESCE(rule.note, ''), ' [已迁移 internal_sku]'), + updated_at = NOW() +WHERE rule.match_key_type = 'sku_code' + AND rule.channel = 'affiliate_dash' + AND rule.enabled = TRUE + AND EXISTS ( + SELECT 1 + FROM product_catalog product + WHERE product.canonical_sku = CONCAT('aff:', rule.match_key) + AND product.status = 'active' + AND BTRIM(COALESCE(product.canonical_name, '')) <> '' + ); + +-- 卡券旧规则按历史订单中唯一可解析的内部 SKU 迁移；歧义数据保持原规则不动。 +WITH resolved_legacy AS ( + SELECT + rule.id, + matched.internal_sku, + MIN(rule.priority) AS priority + FROM sku_channel_rules rule + CROSS JOIN LATERAL ( + SELECT + MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku, + COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count + FROM order_items oi + JOIN orders source_order ON source_order.id = oi.order_id + WHERE oi.sku_code = rule.match_key + AND (rule.provider = '' OR source_order.provider = rule.provider) + AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id) + ) matched + WHERE rule.match_key_type = 'sku_code' + AND rule.channel = 'third_party_voucher' + AND rule.enabled = TRUE + AND matched.internal_count = 1 + AND BTRIM(COALESCE(matched.internal_sku, '')) <> '' + GROUP BY rule.id, matched.internal_sku +) +INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + config_json, source, created_at, updated_at +) +SELECT + internal_sku, + 'third_party_voucher', + priority, + FALSE, + TRUE, + '{}'::jsonb, + 'legacy-rule', + NOW(), + NOW() +FROM resolved_legacy +ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + enabled = TRUE, + updated_at = NOW(); + +WITH resolved_legacy AS ( + SELECT rule.id + FROM sku_channel_rules rule + CROSS JOIN LATERAL ( + SELECT + MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku, + COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count + FROM order_items oi + JOIN orders source_order ON source_order.id = oi.order_id + WHERE oi.sku_code = rule.match_key + AND (rule.provider = '' OR source_order.provider = rule.provider) + AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id) + ) matched + WHERE rule.match_key_type = 'sku_code' + AND rule.channel = 'third_party_voucher' + AND rule.enabled = TRUE + AND matched.internal_count = 1 + AND BTRIM(COALESCE(matched.internal_sku, '')) <> '' +) +UPDATE sku_channel_rules rule +SET enabled = FALSE, + note = CONCAT(COALESCE(rule.note, ''), ' [已迁移 internal_sku]'), + updated_at = NOW() +FROM resolved_legacy +WHERE rule.id = resolved_legacy.id; + +ALTER TABLE internal_sku_channels + DROP CONSTRAINT IF EXISTS chk_internal_sku_channels_priority_nonnegative; +ALTER TABLE internal_sku_channels + ADD CONSTRAINT chk_internal_sku_channels_priority_nonnegative CHECK (priority >= 0); + +CREATE INDEX IF NOT EXISTS idx_internal_sku_channels_fallback + ON internal_sku_channels (internal_sku, priority, channel) + WHERE enabled = TRUE AND is_fallback = TRUE; diff --git a/apps/backend/src/db/migrations/121_internal_sku_single_primary.sql b/apps/backend/src/db/migrations/121_internal_sku_single_primary.sql new file mode 100644 index 00000000..4678b033 --- /dev/null +++ b/apps/backend/src/db/migrations/121_internal_sku_single_primary.sql @@ -0,0 +1,25 @@ +-- 121_internal_sku_single_primary.sql +-- +-- 一个内部 SKU 只能有一个主渠道；其他渠道必须显式标记 is_fallback。 +-- 迁移 120 已把人工并列渠道收敛为兜底；这里在任何剩余歧义上显式阻止部署。 + +DO $$ +DECLARE + conflict_sku TEXT; +BEGIN + SELECT internal_sku INTO conflict_sku + FROM internal_sku_channels + WHERE enabled = TRUE AND is_fallback = FALSE + GROUP BY internal_sku + HAVING count(*) > 1 + ORDER BY internal_sku + LIMIT 1; + + IF conflict_sku IS NOT NULL THEN + RAISE EXCEPTION '内部 SKU % 存在多个主渠道，请先将其余渠道标记为 is_fallback', conflict_sku; + END IF; +END $$; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_internal_sku_channels_single_primary + ON internal_sku_channels (internal_sku) + WHERE enabled = TRUE AND is_fallback = FALSE; 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 989775fc..a8807f18 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 @@ -4,6 +4,7 @@ import type { JsonObject } from '../../types/json.js' import { createHttpError } from '../../utils/http.js' import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js' import { nowIso } from '../../utils/time.js' +import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js' /** * 人工履约 SKU 配置服务。 @@ -125,7 +126,7 @@ export async function scanManualSkus(options: ScanOptions) { oi.sku_name, COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date AS stat_date, - oi.manual_sku_key AS sku_key + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key FROM order_items oi JOIN orders o ON o.id = oi.order_id WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' @@ -243,7 +244,7 @@ export async function listManualSkus(input: Record = {}) { JOIN orders o ON o.id = oi.order_id WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND o.provider = $${params.length} - AND oi.manual_sku_key = r.sku_key + AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = r.sku_key )`) } @@ -408,6 +409,15 @@ export async function deleteManualFulfillmentConfig(skuKey: string) { [key], ) await query(`DELETE FROM sku_channel_rules WHERE source = 'manual' AND match_key = $1`, [key]) + await upsertInternalSkuChannel({ + internalSku: key, + channel: 'manual', + priority: 90, + isFallback: true, + enabled: false, + source: 'manual_config', + configJson: { manualConfigId: null }, + }) await syncRegistryConfigId(key, null) return { skuKey: key, deleted: deleted.rowCount || 0 } } @@ -435,7 +445,7 @@ export async function rematchManualFulfillments(input: Record = FROM order_items oi JOIN orders o ON o.id = oi.order_id JOIN manual_fulfillment_configs cfg - ON cfg.sku_key = oi.manual_sku_key + ON cfg.sku_key = resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND cfg.enabled = TRUE AND ($1 = '' OR cfg.sku_key = $1) @@ -719,27 +729,57 @@ function requiredRow(row: T | undefined, message: string): T { async function syncManualChannelRule(config: { id: number sku_key: string + canonical_product_id: number | null enabled: boolean note: string }): Promise { const skuKey = String(config.sku_key || '').trim() if (!skuKey) return + const otherChannel = await query<{ exists: boolean }>( + ` + SELECT EXISTS ( + SELECT 1 FROM internal_sku_channels + WHERE btrim(internal_sku) = $1 AND enabled = TRUE AND channel <> 'manual' + ) AS exists + `, + [skuKey], + ) + const hasOtherChannel = otherChannel.rows[0]?.exists === true + await upsertInternalSkuChannel({ + internalSku: skuKey, + channel: 'manual', + priority: hasOtherChannel ? 90 : 10, + isFallback: hasOtherChannel, + enabled: config.enabled !== false, + canonicalProductId: config.canonical_product_id, + source: 'manual_config', + configJson: { manualConfigId: config.id, note: String(config.note || '').slice(0, 200) }, + replacePriority: true, + }) + await query( ` INSERT INTO sku_channel_rules ( provider, shop_id, match_key_type, match_key, channel, priority, source, source_ref_id, enabled, note, created_at, updated_at ) - VALUES ('', '', 'internal_sku', $1, 'manual', 70, 'manual', $2, $3, $4, NOW(), NOW()) + VALUES ('', '', 'internal_sku', $1, 'manual', $5, 'manual', $2, $3, $4, NOW(), NOW()) ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO UPDATE SET enabled = EXCLUDED.enabled, + priority = EXCLUDED.priority, source = 'manual', source_ref_id = EXCLUDED.source_ref_id, note = EXCLUDED.note, updated_at = NOW() `, - [skuKey, config.id, config.enabled !== false, String(config.note || '').slice(0, 200)], + [ + skuKey, + config.id, + config.enabled !== false, + String(config.note || '').slice(0, 200), + hasOtherChannel ? 90 : 10, + ], ) } diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts index 47f1d870..05523ece 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -215,21 +215,6 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise { + if (value && typeof value === 'object' && !Array.isArray(value)) { + return value as Record + } + try { + const parsed = JSON.parse(String(value || '{}')) + return parsed && typeof parsed === 'object' && !Array.isArray(parsed) + ? (parsed as Record) + : {} + } catch { + return {} + } +} + +export type EnsureCarrierResult = { + ready: boolean + created: boolean + reason: string + message: string +} + +async function hasWorkOrder(orderItemId: number): Promise { + const result = await query<{ exists: boolean }>( + `SELECT EXISTS (SELECT 1 FROM work_orders WHERE order_item_id = $1) AS exists`, + [orderItemId], + ) + return result.rows[0]?.exists === true +} + +async function hasAffiliateTask(orderItemId: number): Promise { + const result = await query<{ exists: boolean }>( + `SELECT EXISTS ( + SELECT 1 FROM fulfillment_tasks + WHERE order_item_id = $1 AND executor_key = 'affiliate_dash' + ) AS exists`, + [orderItemId], + ) + return result.rows[0]?.exists === true +} + +function toResolvedItem(item: OrderItemRow): ResolvedFulfillmentItem { + return { + ...item, + itemId: String(item.id), + externalItemId: String(item.id), + externalSkuCode: item.sku_code, + externalSkuName: item.sku_name, + skuCode: item.sku_code, + skuName: item.sku_name, + spec: parseObject(item.spec_json), + snapshot: parseObject(item.item_snapshot_json), + isConfigured: false, + } as ResolvedFulfillmentItem +} + +/** + * 幂等创建指定渠道的真实载体。 + * + * 自动发货成功标准是存在 affiliate_dash 任务；接单成功标准是存在 work_order。 + * manual/voucher 不需要执行载体，直接视为 ready。 + */ +export async function ensureCarrierForOrderItem(input: { + order: OrderRow + item: OrderItemRow + channel: FulfillmentChannel +}): Promise { + const { order, item, channel } = input + + if (channel === 'manual') { + const config = await query<{ id: number; canonical_product_id: number | null }>( + `SELECT id, canonical_product_id + FROM manual_fulfillment_configs + WHERE sku_key = resolve_internal_sku($1, $2, $3) AND enabled = TRUE + LIMIT 1`, + [item.manual_sku_key || '', item.sku_name, item.sku_code], + ) + const row = config.rows[0] + if (row) { + await query( + `UPDATE order_items + SET fulfillment_config_id = $2, + canonical_product_id = COALESCE(canonical_product_id, $3), + updated_at = NOW() + WHERE id = $1`, + [item.id, row.id, row.canonical_product_id], + ) + return { ready: true, created: false, reason: 'manual_config_ready', message: '' } + } + return { + ready: true, + created: false, + reason: 'manual_config_missing', + message: '人工渠道已选定，但缺少人工履约配置', + } + } + + if (channel === 'third_party_voucher') { + return { ready: true, created: false, reason: 'carrier_not_required', message: '' } + } + + if (channel === 'work_order') { + if (await hasWorkOrder(Number(item.id))) { + return { ready: true, created: false, reason: 'work_order_exists', message: '' } + } + const snapshot = parseObject(item.item_snapshot_json) + const classification = parseObject(snapshot.classification) + const route = parseObject(snapshot.fulfillmentRoute) + const workOrderSnapshot = { + ...snapshot, + classification: { ...classification, channel: 'work_order' }, + fulfillmentRoute: { + ...route, + selectedExecutorKey: '', + reason: '内部 SKU 渠道指定接单平台', + }, + } + const workOrderItem = { + ...item, + item_snapshot_json: workOrderSnapshot, + } as OrderItemRow + const result = await syncWorkerOrdersForSourceOrder(order, [workOrderItem], { + source: 'carrier_creation', + autoOnly: true, + skipChannelAssign: true, + }) + const ready = (await hasWorkOrder(Number(item.id))) || result.created.length > 0 + if (ready) { + await query( + `UPDATE order_items SET item_snapshot_json = $2::jsonb, updated_at = NOW() WHERE id = $1`, + [item.id, JSON.stringify(workOrderSnapshot)], + ) + } + return { + ready, + created: result.created.length > 0, + reason: ready ? 'work_order_created' : result.skipped[0]?.reason || 'work_order_unmatched', + message: ready ? '' : '未匹配到可自动创建的接单模板', + } + } + + if (await hasAffiliateTask(Number(item.id))) { + return { ready: true, created: false, reason: 'affiliate_task_exists', message: '' } + } + + const resolved = await applyIntakeChannelDecision( + toResolvedItem(item), + { + provider: order.provider, + platform: order.platform, + shopId: order.shop_id, + }, + { preferredChannel: 'affiliate_dash' }, + ) + const snapshot = parseObject(resolved.snapshot) + const route = parseObject(snapshot.fulfillmentRoute) + if (route.selectedExecutorKey !== 'affiliate_dash') { + return { + ready: false, + created: false, + reason: 'affiliate_mapping_missing', + message: '自动发货映射缺失', + } + } + + await query( + `UPDATE order_items + SET sku_code = $2, spec_json = $3::jsonb, item_snapshot_json = $4::jsonb, updated_at = NOW() + WHERE id = $1`, + [ + item.id, + String(resolved.skuCode || item.sku_code), + JSON.stringify(resolved.spec || {}), + JSON.stringify(snapshot), + ], + ) + const before = await hasAffiliateTask(Number(item.id)) + if (!before) { + await syncDeliveryTasksForOrder(order, [ + { + ...item, + sku_code: String(resolved.skuCode || item.sku_code), + spec_json: resolved.spec || {}, + item_snapshot_json: snapshot, + } as OrderItemRow, + ]) + } + const ready = await hasAffiliateTask(Number(item.id)) + return { + ready, + created: ready && !before, + reason: ready ? 'affiliate_task_created' : 'affiliate_task_planning_failed', + message: ready ? '' : '自动发货任务规划失败', + } +} diff --git a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts index 794cb6cb..74f4f186 100644 --- a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts @@ -1,178 +1,323 @@ import { query } from '../../db/client.js' import { getOrderById } from '../../repositories/order-repo.js' -import { syncDeliveryTasksForOrder } from '../order/delivery-task-service.js' -import { syncWorkerOrdersForSourceOrder } from '../worker-platform/sync-worker-orders-service.js' -import { applyIntakeChannelDecision } from './intake-channel-decision-service.js' -import type { ResolvedFulfillmentItem } from './product-resolution-service.js' +import type { OrderItemRow } from '../../types/repository/rows.js' import { logWarn } from '../../utils/logger.js' +import { FULFILLMENT_CHANNELS, type FulfillmentChannel } from './channel-types.js' +import { ensureCarrierForOrderItem } from './carrier-creation-service.js' +import { + planInternalSkuChannels, + type InternalSkuChannelRow, +} from './internal-sku-channel-service.js' -type PendingCarrierItem = { +type CarrierCandidate = { id: number order_id: number - sku_code: string - sku_name: string - quantity: number - item_snapshot_json: unknown - spec_json: unknown - channel: 'affiliate_dash' | 'work_order' + internal_sku: string + channels: InternalSkuChannelRow[] } -function parseObject(value: unknown): Record { - if (value && typeof value === 'object' && !Array.isArray(value)) { - return value as Record - } - try { - const parsed = JSON.parse(String(value || '{}')) - return parsed && typeof parsed === 'object' && !Array.isArray(parsed) - ? (parsed as Record) - : {} - } catch { - return {} - } +function normalizeChannels(value: unknown): InternalSkuChannelRow[] { + if (!Array.isArray(value)) return [] + return value + .filter((row): row is Record => Boolean(row) && typeof row === 'object') + .map((row) => ({ + internal_sku: String(row.internal_sku || '').trim(), + channel: String(row.channel || '') as FulfillmentChannel, + priority: Number(row.priority || 100), + is_fallback: row.is_fallback === true, + enabled: row.enabled !== false, + canonical_product_id: + row.canonical_product_id == null ? null : Number(row.canonical_product_id), + config_json: + row.config_json && typeof row.config_json === 'object' + ? (row.config_json as Record) + : {}, + source: String(row.source || ''), + })) + .filter((row) => FULFILLMENT_CHANNELS.includes(row.channel)) } -/** - * 为“内部 SKU 有渠道能力、但还没有载体”的 pending 单补载体。 - * - * - 接单平台：走现有工单匹配器，能匹配模板就建工单； - * - 自动发货：补 fulfillmentRoute/affiliateDash 后建任务； - * - 补不出来的继续保留 no_carrier，等运营补模板/映射。 - * - * 财务渠道判定不在本服务内做；调用方在补完载体后再跑一次 channel_assign。 - */ -export async function reconcilePendingCarriers(input: { limit?: number } = {}): Promise<{ - scanned: number - workOrdersCreated: number - affiliateTasksCreated: number - remaining: number -}> { - const limit = Math.min(500, Math.max(1, Math.trunc(Number(input.limit) || 200))) - const rows = await query( +async function loadCandidateItems(input: { + limit: number + orderItemIds?: number[] + assignedMissingCarrier?: boolean +}): Promise { + const ids = (input.orderItemIds || []) + .map(Number) + .filter((value) => Number.isInteger(value) && value > 0) + const idFilter = ids.length > 0 ? 'AND oi.id = ANY($2::bigint[])' : '' + const statusFilter = input.assignedMissingCarrier + ? `oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order') + AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + )` + : `oi.fulfillment_status = 'pending' AND oi.fulfillment_locked = FALSE` + + const params: unknown[] = [input.limit] + if (ids.length > 0) params.push(ids) + const result = await query( ` - SELECT oi.id, oi.order_id, oi.sku_code, oi.sku_name, oi.quantity, - oi.item_snapshot_json, oi.spec_json, - cap.channel + SELECT + oi.id, + oi.order_id, + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku, + capability.channels FROM order_items oi JOIN LATERAL ( - SELECT c.channel - FROM internal_sku_channels c - WHERE c.enabled = TRUE - AND c.channel IN ('affiliate_dash', 'work_order') - AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name)) - ORDER BY - c.priority ASC, - CASE c.channel - WHEN 'affiliate_dash' THEN 1 - WHEN 'work_order' THEN 2 - ELSE 9 - END, - c.channel ASC - LIMIT 1 - ) cap ON TRUE - WHERE oi.fulfillment_status = 'pending' - AND oi.fulfillment_locked = FALSE - AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) - AND NOT EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' - ) + SELECT COALESCE( + jsonb_agg( + jsonb_build_object( + '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 + ) + ORDER BY channel.priority, channel.channel + ), + '[]'::jsonb + ) AS channels + FROM internal_sku_channels channel + WHERE channel.enabled = TRUE + AND btrim(channel.internal_sku) = resolve_internal_sku( + oi.manual_sku_key, oi.sku_name, oi.sku_code + ) + ) capability ON jsonb_array_length(capability.channels) > 0 + WHERE ${statusFilter} + ${idFilter} ORDER BY oi.id LIMIT $1 `, - [limit], + params, ) + return result.rows.map((row) => ({ ...row, channels: normalizeChannels(row.channels) })) +} + +async function markPendingWithoutCarrier(input: { + itemId: number + internalSku: string + attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }> +}): Promise { + await query( + ` + UPDATE order_items + SET fulfillment_pending_reason = 'no_carrier', + fulfillment_pending_detail = $2::jsonb, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status = 'pending' + `, + [ + input.itemId, + JSON.stringify({ + internalSku: input.internalSku, + attempts: input.attempts, + }), + ], + ) +} + +async function selectSafeFallback(input: { + orderItemId: number + channel: FulfillmentChannel + primaryChannel: FulfillmentChannel + reason: string + includeAssignedMissingCarrier: boolean +}): Promise { + const result = await query( + ` + UPDATE order_items + SET fulfillment_channel = $2, + fulfillment_status = 'assigned', + fulfillment_source = 'auto_mapping', + fulfillment_reason = $3, + channel_switch_reason = $3, + channel_switched_at = NOW(), + fulfillment_pending_reason = NULL, + fulfillment_pending_detail = NULL, + updated_at = NOW() + WHERE id = $1 + AND ( + (fulfillment_status = 'pending' AND fulfillment_locked = FALSE) + OR ( + $4::boolean + AND fulfillment_status = 'assigned' + AND fulfillment_locked = FALSE + AND fulfillment_source NOT IN ('operator_assigned', 'manual_confirm') + ) + ) + `, + [ + input.orderItemId, + input.channel, + `主渠道 ${input.primaryChannel} 无法创建载体：${input.reason}`, + input.includeAssignedMissingCarrier, + ], + ) + return (result.rowCount || 0) > 0 +} + +/** + * 按 internal_sku_channels 的明确优先级补载体。 + * + * 主渠道是 affiliate/work_order 但载体暂时创建失败时，只有显式 is_fallback 的 + * manual/voucher 才能接管。所有结果都落 pending 明细，不再静默卡住。 + */ +export async function reconcilePendingCarriers( + input: { + limit?: number + orderItemIds?: number[] + includeAssignedMissingCarrier?: boolean + allowFallback?: boolean + } = {}, +): Promise<{ + scanned: number + workOrdersCreated: number + affiliateTasksCreated: number + fallbacksSelected: number + failed: number + remaining: number +}> { + const limit = Math.min(500, Math.max(1, Math.trunc(Number(input.limit) || 200))) + const rows = await loadCandidateItems({ + limit, + assignedMissingCarrier: input.includeAssignedMissingCarrier === true, + ...(input.orderItemIds ? { orderItemIds: input.orderItemIds } : {}), + }) let workOrdersCreated = 0 let affiliateTasksCreated = 0 - const byOrder = new Map() - for (const row of rows.rows) { + let fallbacksSelected = 0 + let failed = 0 + const byOrder = new Map() + for (const row of rows) { const list = byOrder.get(Number(row.order_id)) || [] list.push(row) byOrder.set(Number(row.order_id), list) } - for (const [orderId, items] of byOrder) { + for (const [orderId, candidates] of byOrder) { const order = await getOrderById(orderId) if (!order) continue - const itemRows = await query<{ - id: number - order_id: number - sku_code: string - sku_name: string - quantity: number - item_snapshot_json: unknown - spec_json: unknown - }>( - `SELECT id, order_id, sku_code, sku_name, quantity, item_snapshot_json, spec_json - FROM order_items WHERE order_id = $1`, - [orderId], - ) + const itemRows = await query(`SELECT * FROM order_items WHERE order_id = $1`, [ + orderId, + ]) const itemById = new Map(itemRows.rows.map((item) => [Number(item.id), item])) - for (const target of items) { - const item = itemById.get(Number(target.id)) + for (const candidate of candidates) { + const item = itemById.get(Number(candidate.id)) if (!item) continue - try { - if (target.channel === 'work_order') { - const result = await syncWorkerOrdersForSourceOrder(order, [item as never], { - source: 'carrier_reconcile', - autoOnly: true, - }) - workOrdersCreated += result.created.length - continue + const assignedChannel = String(item.fulfillment_channel || '') as FulfillmentChannel + const plan = planInternalSkuChannels(candidate.channels) + const exactOnly = + input.includeAssignedMissingCarrier === true && + (item.fulfillment_locked === true || + item.fulfillment_source === 'operator_assigned' || + item.fulfillment_source === 'manual_confirm') + const attempts = + input.includeAssignedMissingCarrier && exactOnly + ? FULFILLMENT_CHANNELS.includes(assignedChannel) + ? candidate.channels.filter((row) => row.channel === assignedChannel) + : [] + : plan.attempts + if (attempts.length === 0) { + failed += 1 + if (!input.includeAssignedMissingCarrier) { + await query( + `UPDATE order_items + SET fulfillment_pending_reason = 'routing_conflict', + fulfillment_pending_detail = $2::jsonb, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status = 'pending'`, + [ + item.id, + JSON.stringify({ + internalSku: candidate.internal_sku, + channels: plan.conflicts.map((channel) => channel.channel), + }), + ], + ) } + continue + } - const resolved = { - ...item, - itemId: String(item.id), - externalItemId: String(item.id), - externalSkuCode: item.sku_code, - externalSkuName: item.sku_name, - skuCode: item.sku_code, - skuName: item.sku_name, - spec: parseObject(item.spec_json), - snapshot: parseObject(item.item_snapshot_json), - isConfigured: false, - } as unknown as ResolvedFulfillmentItem - const enriched = await applyIntakeChannelDecision(resolved, { - provider: order.provider, - platform: order.platform, - shopId: order.shop_id, - }) - const snapshot = parseObject(enriched.snapshot) - const route = parseObject(snapshot.fulfillmentRoute) - if (route.selectedExecutorKey !== 'affiliate_dash') continue + const failures: Array<{ channel: FulfillmentChannel; reason: string; message: string }> = [] + let resolved = false + for (let index = 0; index < attempts.length; index += 1) { + const attempt = attempts[index]! + try { + const carrier = await ensureCarrierForOrderItem({ + order, + item, + channel: attempt.channel, + }) + if (!carrier.ready) { + failures.push({ + channel: attempt.channel, + reason: carrier.reason, + message: carrier.message, + }) + const canFallback = + input.allowFallback !== false && + !exactOnly && + index === 0 && + (attempt.channel === 'affiliate_dash' || attempt.channel === 'work_order') + if (canFallback) continue + break + } - await query( - `UPDATE order_items - SET sku_code = $2, - spec_json = $3, - item_snapshot_json = $4, - updated_at = NOW() - WHERE id = $1`, - [ - item.id, - String(enriched.skuCode || item.sku_code), - JSON.stringify(enriched.spec || {}), - JSON.stringify(snapshot), - ], - ) - const tasks = await syncDeliveryTasksForOrder(order, [ - { - ...item, - sku_code: String(enriched.skuCode || item.sku_code), - spec_json: enriched.spec || {}, - item_snapshot_json: snapshot, - } as never, - ]) - affiliateTasksCreated += tasks.length - } catch (error) { - logWarn('[carrier-reconciliation]', '补履约载体失败', { - orderItemId: target.id, - channel: target.channel, - error: error instanceof Error ? error.message : String(error), - }) + if (carrier.created && attempt.channel === 'work_order') workOrdersCreated += 1 + if (carrier.created && attempt.channel === 'affiliate_dash') affiliateTasksCreated += 1 + + const isFallbackAttempt = index > 0 + if ( + !exactOnly && + (attempt.channel === 'manual' || + attempt.channel === 'third_party_voucher' || + isFallbackAttempt) + ) { + const primaryChannel = plan.primary?.channel || attempts[0]!.channel + if ( + await selectSafeFallback({ + orderItemId: Number(item.id), + channel: attempt.channel, + primaryChannel, + reason: failures[0]?.reason || 'fallback', + includeAssignedMissingCarrier: input.includeAssignedMissingCarrier === true, + }) + ) { + fallbacksSelected += 1 + } + } + resolved = true + break + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + failures.push({ channel: attempt.channel, reason: 'exception', message }) + logWarn('[carrier-reconciliation]', '补履约载体失败', { + orderItemId: item.id, + channel: attempt.channel, + error: message, + }) + break + } + } + + if (!resolved) { + failed += 1 + if (!input.includeAssignedMissingCarrier) { + await markPendingWithoutCarrier({ + itemId: Number(item.id), + internalSku: candidate.internal_sku, + attempts: failures, + }) + } } } } @@ -181,15 +326,22 @@ export async function reconcilePendingCarriers(input: { limit?: number } = {}): ` SELECT count(*)::int AS count FROM order_items oi - WHERE oi.fulfillment_status = 'pending' - AND oi.fulfillment_pending_reason = 'no_carrier' + WHERE oi.fulfillment_status IN ('pending', 'assigned') + AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order') + AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ) `, ) return { - scanned: rows.rows.length, + scanned: rows.length, workOrdersCreated, affiliateTasksCreated, + fallbacksSelected, + failed, remaining: Number(remainingResult.rows[0]?.count || 0), } } diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts index 98de6e62..56f44eb7 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts @@ -6,6 +6,7 @@ import { type ChannelCandidateItem, type ChannelRule, } from './channel-assign-service.js' +import type { InternalSkuChannelRow } from './internal-sku-channel-service.js' function buildItem(patch: Partial = {}): ChannelCandidateItem { return { @@ -20,12 +21,26 @@ function buildItem(patch: Partial = {}): ChannelCandidateI has_work_order: false, has_affiliate_task: false, has_manual_task: false, - capability_channel: null, + channels: [], basis_date: '2026-10-31', ...patch, } } +function buildChannel(patch: Partial = {}): InternalSkuChannelRow { + return { + internal_sku: '套装-测试', + channel: 'manual', + priority: 10, + is_fallback: false, + enabled: true, + canonical_product_id: null, + config_json: {}, + source: 'test', + ...patch, + } +} + function buildRule(patch: Partial = {}): ChannelRule { return { id: 1, @@ -40,13 +55,13 @@ function buildRule(patch: Partial = {}): ChannelRule { } } -test('渠道判定：精确 SKU 规则优先于载体证据', () => { +test('渠道判定：真实载体优先于冲突的旧规则', () => { const decision = resolveChannelDecision(buildItem({ has_work_order: true }), [ buildRule({ channel: 'third_party_voucher', priority: 10 }), ]) - assert.equal(decision.channel, 'third_party_voucher') - assert.equal(decision.source, 'auto_rule') + assert.equal(decision.channel, 'work_order') + assert.equal(decision.source, 'carrier_evidence') assert.equal(decision.conflict, false) }) @@ -79,19 +94,19 @@ test('渠道判定：同优先级多渠道判为冲突，不静默选择', () => test('渠道判定：不同优先级时取优先级更小的', () => { const decision = resolveChannelDecision(buildItem(), [ - buildRule({ id: 1, channel: 'affiliate_dash', priority: 60 }), + buildRule({ id: 1, channel: 'third_party_voucher', priority: 60 }), buildRule({ id: 2, channel: 'manual', priority: 70 }), ]) - assert.equal(decision.channel, 'affiliate_dash') + assert.equal(decision.channel, 'third_party_voucher') }) test('渠道判定：provider / shop 限定生效，空串表示不限', () => { const scoped = buildRule({ provider: 'pdd', shop_id: 'shop-b' }) assert.equal(resolveChannelDecision(buildItem(), [scoped]).channel, null) - const wide = buildRule({ provider: '', shop_id: '' }) - assert.equal(resolveChannelDecision(buildItem(), [wide]).channel, 'affiliate_dash') + const wide = buildRule({ provider: '', shop_id: '', channel: 'third_party_voucher' }) + assert.equal(resolveChannelDecision(buildItem(), [wide]).channel, 'third_party_voucher') }) test('渠道判定：内部 SKU 与 SKU 名称两种键都能命中', () => { @@ -125,23 +140,50 @@ test('渠道判定：停用的规则不参与', () => { test('渠道判定：卡券/人工能力可兜底，自动发货/接单必须靠载体', () => { const voucher = resolveChannelDecision( - buildItem({ capability_channel: 'third_party_voucher' }), + buildItem({ channels: [buildChannel({ channel: 'third_party_voucher' })] }), [], ) assert.equal(voucher.channel, 'third_party_voucher') assert.equal(voucher.source, 'auto_mapping') - const manual = resolveChannelDecision(buildItem({ capability_channel: 'manual' }), []) + const manual = resolveChannelDecision(buildItem({ channels: [buildChannel()] }), []) assert.equal(manual.channel, 'manual') assert.equal(manual.source, 'auto_mapping') - const affiliate = resolveChannelDecision(buildItem({ capability_channel: 'affiliate_dash' }), []) + const affiliate = resolveChannelDecision( + buildItem({ channels: [buildChannel({ channel: 'affiliate_dash' })] }), + [], + ) assert.equal(affiliate.channel, null, '自动发货能力没有任务时不能只改财务口径') - const workOrder = resolveChannelDecision(buildItem({ capability_channel: 'work_order' }), []) + const workOrder = resolveChannelDecision( + buildItem({ channels: [buildChannel({ channel: 'work_order' })] }), + [], + ) assert.equal(workOrder.channel, null, '接单能力没有工单时不能只改财务口径') }) +test('渠道判定：内部 SKU 配置优先于旧 sku_code 规则', () => { + const decision = resolveChannelDecision( + buildItem({ channels: [buildChannel({ channel: 'third_party_voucher' })] }), + [buildRule({ channel: 'manual' })], + ) + assert.equal(decision.channel, 'third_party_voucher') + assert.equal(decision.source, 'auto_mapping') +}) + +test('渠道判定：内部 SKU 同级多渠道判为冲突', () => { + const decision = resolveChannelDecision( + buildItem({ + channels: [buildChannel({ channel: 'work_order' }), buildChannel({ channel: 'manual' })], + }), + [], + ) + assert.equal(decision.channel, null) + assert.equal(decision.conflict, true) + assert.deepEqual(decision.conflictChannels.sort(), ['manual', 'work_order']) +}) + test('渠道判定：人工任务作为人工载体证据', () => { const decision = resolveChannelDecision(buildItem({ has_manual_task: true }), []) assert.equal(decision.channel, 'manual') diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 4fe3ee6c..d4c05b34 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -1,4 +1,6 @@ import { query } from '../../db/client.js' +import { getOrderById } from '../../repositories/order-repo.js' +import type { OrderItemRow } from '../../types/repository/rows.js' import { createHttpError } from '../../utils/http.js' import { nowIso } from '../../utils/time.js' import { upsertAggregatedException } from '../cost-center/cost-exception-service.js' @@ -18,6 +20,14 @@ import { type MatchKeyType, type RuleSource, } from './channel-types.js' +import { + disableInternalSkuChannel, + planInternalSkuChannels, + setInternalSkuPrimaryChannel, + upsertInternalSkuChannel, + type InternalSkuChannelRow, +} from './internal-sku-channel-service.js' +import { ensureCarrierForOrderItem } from './carrier-creation-service.js' /** * 管道 A：渠道判定。 @@ -27,19 +37,14 @@ import { * * 判定顺序： * 1. 运营锁定 → 跳过 - * 2. 精确规则（sku_channel_rules，优先级升序）；同优先级命中多渠道 → 冲突，保持 pending - * 3. 载体证据：已有工单 → work_order；已有自动发货任务 → affiliate_dash - * 4. 都没有 → 保持 pending（不再兜底成人工） + * 2. 已有且唯一的真实载体 → 直接按载体落渠道；多载体 → 冲突 + * 3. internal_sku_channels：一个主渠道 + 显式兜底；多主渠道 → 冲突 + * 4. 旧 sku_channel_rules 仅作兼容回退，且自动发货/接单仍必须有真实载体 + * 5. 都不满足 → 保留 pending 并写明确原因 */ -/** 内部 SKU：优先中文手动键，其次中文商品名，最后平台 SKU。 */ -const INTERNAL_SKU_SQL = ` - CASE - WHEN btrim(COALESCE(oi.manual_sku_key, '')) ~ '[一-龥]' THEN btrim(oi.manual_sku_key) - WHEN btrim(COALESCE(oi.sku_name, '')) ~ '[一-龥]' THEN btrim(oi.sku_name) - ELSE COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), NULLIF(btrim(oi.sku_name), ''), oi.sku_code) - END -` +/** 数据库函数 `resolve_internal_sku` 是订单项内部 SKU 的唯一 SQL 口径。 */ +const INTERNAL_SKU_SQL = `resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)` export type ChannelRule = { id: number @@ -64,7 +69,7 @@ export type ChannelCandidateItem = { has_work_order: boolean has_affiliate_task: boolean has_manual_task: boolean - capability_channel: FulfillmentChannel | null + channels: InternalSkuChannelRow[] basis_date: string } @@ -74,6 +79,8 @@ export type ChannelDecision = { conflict: boolean conflictChannels: FulfillmentChannel[] matchedRuleIds: number[] + pendingChannel: FulfillmentChannel | null + pendingReason: 'no_internal_sku' | 'no_channel_capability' | 'no_carrier' | '' } function normalizeKey(value: unknown): string { @@ -99,7 +106,14 @@ function resolvePendingReason(item: ChannelCandidateItem): { if (!String(item.internal_sku || '').trim()) { return { reason: 'no_internal_sku', detail: { skuCode: item.sku_code } } } - if (!item.capability_channel) { + const plan = planInternalSkuChannels(item.channels || []) + if (plan.conflicts.length > 0) { + return { + reason: 'routing_conflict', + detail: { channels: plan.conflicts.map((channel) => channel.channel) }, + } + } + if (!plan.primary) { return { reason: 'no_channel_capability', detail: { internalSku: item.internal_sku }, @@ -107,15 +121,90 @@ function resolvePendingReason(item: ChannelCandidateItem): { } return { reason: 'no_carrier', - detail: { internalSku: item.internal_sku, channel: item.capability_channel }, + detail: { + internalSku: item.internal_sku, + channel: plan.primary.channel, + fallbacks: plan.fallbacks.map((channel) => channel.channel), + }, } } +function emptyDecision(pendingReason: ChannelDecision['pendingReason'] = '') { + return { + channel: null, + source: '', + conflict: false, + conflictChannels: [], + matchedRuleIds: [], + pendingChannel: null, + pendingReason, + } satisfies ChannelDecision +} + /** 纯函数：给定规则与订单项，得出渠道决策（便于单测与文档对照）。 */ export function resolveChannelDecision( item: ChannelCandidateItem, rules: ChannelRule[], ): ChannelDecision { + const carrierChannels: FulfillmentChannel[] = [ + ...(item.has_work_order ? ['work_order' as const] : []), + ...(item.has_affiliate_task ? ['affiliate_dash' as const] : []), + ...(item.has_manual_task ? ['manual' as const] : []), + ] + if (carrierChannels.length > 1) { + return { + channel: null, + source: '', + conflict: true, + conflictChannels: carrierChannels, + matchedRuleIds: [], + pendingChannel: carrierChannels[0] || null, + pendingReason: 'no_carrier', + } + } + + const plan = planInternalSkuChannels(item.channels || []) + if (plan.conflicts.length > 0) { + return { + channel: null, + source: '', + conflict: true, + conflictChannels: plan.conflicts.map((channel) => channel.channel), + matchedRuleIds: [], + pendingChannel: null, + pendingReason: 'no_channel_capability', + } + } + + // 真实载体是最终事实；配置只能决定准备创建哪个载体，不能覆盖已经存在的另一类载体。 + const evidence = carrierChannels[0] + if (evidence) { + return { + channel: evidence, + source: 'carrier_evidence', + conflict: false, + conflictChannels: [], + matchedRuleIds: [], + pendingChannel: evidence, + pendingReason: '', + } + } + + if (plan.primary) { + if (plan.primary.channel === 'manual' || plan.primary.channel === 'third_party_voucher') { + return { + channel: plan.primary.channel, + source: 'auto_mapping', + conflict: false, + conflictChannels: [], + matchedRuleIds: [], + pendingChannel: plan.primary.channel, + pendingReason: '', + } + } + return emptyDecision('no_carrier') + } + const hits = rules.filter( (rule) => rule.enabled && @@ -135,58 +224,25 @@ export function resolveChannelDecision( conflict: true, conflictChannels: channels, matchedRuleIds: best.map((rule) => rule.id), + pendingChannel: null, + pendingReason: 'no_channel_capability', } } + const channel = channels[0]! + if (channel === 'affiliate_dash' || channel === 'work_order') { + return emptyDecision('no_carrier') + } return { - channel: channels[0]!, + channel, source: 'auto_rule', conflict: false, conflictChannels: [], matchedRuleIds: best.map((rule) => rule.id), + pendingChannel: channel, + pendingReason: '', } } - - if (item.has_work_order) { - return { - channel: 'work_order', - source: 'carrier_evidence', - conflict: false, - conflictChannels: [], - matchedRuleIds: [], - } - } - if (item.has_affiliate_task) { - return { - channel: 'affiliate_dash', - source: 'carrier_evidence', - conflict: false, - conflictChannels: [], - matchedRuleIds: [], - } - } - if (item.has_manual_task) { - return { - channel: 'manual', - source: 'carrier_evidence', - conflict: false, - conflictChannels: [], - matchedRuleIds: [], - } - } - - // 卡券/人工没有必须创建的履约任务，可直接用商品渠道能力兜底； - // 自动发货/接单平台仍要求真实载体，避免只改口径不派单。 - if (item.capability_channel === 'third_party_voucher' || item.capability_channel === 'manual') { - return { - channel: item.capability_channel, - source: 'auto_mapping', - conflict: false, - conflictChannels: [], - matchedRuleIds: [], - } - } - - return { channel: null, source: '', conflict: false, conflictChannels: [], matchedRuleIds: [] } + return emptyDecision('no_channel_capability') } export type AssignChannelsResult = { @@ -257,27 +313,9 @@ export async function assignPendingOrderItemChannels( EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' - ) AS has_manual_task, - cap.channel AS capability_channel + ) AS has_manual_task FROM order_items oi JOIN orders o ON o.id = oi.order_id - LEFT JOIN LATERAL ( - SELECT c.channel - FROM internal_sku_channels c - WHERE c.enabled = TRUE - AND btrim(c.internal_sku) = ${INTERNAL_SKU_SQL} - ORDER BY - c.priority ASC, - CASE c.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 - LIMIT 1 - ) cap ON TRUE WHERE oi.fulfillment_status = 'pending' AND oi.fulfillment_locked = FALSE AND oi.id > $1 @@ -302,11 +340,33 @@ export async function assignPendingOrderItemChannels( source: FulfillmentSource }> = [] - for (const item of rows.rows) { + const channelsByInternalSku = await loadInternalSkuChannelsForKeys( + rows.rows.map((item) => item.internal_sku), + ) + + for (const row of rows.rows) { + const item: ChannelCandidateItem = { + ...row, + channels: channelsByInternalSku.get(normalizeKey(row.internal_sku)) || [], + } const decision = resolveChannelDecision(item, rules) if (decision.channel) affectedDates.add(item.basis_date) if (decision.conflict) { result.conflicted += 1 + await query( + `UPDATE order_items + SET fulfillment_pending_reason = 'routing_conflict', + fulfillment_pending_detail = $2::jsonb, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status = 'pending'`, + [ + item.id, + JSON.stringify({ + channels: decision.conflictChannels, + ruleIds: decision.matchedRuleIds, + }), + ], + ) await upsertAggregatedException({ reason: 'routing_conflict', scopeKey: `routing_conflict|${item.provider}|${item.shop_id}|${item.sku_code}|-`, @@ -367,13 +427,163 @@ export async function assignPendingOrderItemChannels( return result } -/** 运营指定渠道：写入并锁定，自动判定不再覆盖。 */ +async function assignOrderItemsAfterCarrierReady(input: { + orderItemIds: number[] + channel: FulfillmentChannel + source: FulfillmentSource + lock: boolean + note?: string | undefined +}): Promise<{ updated: number; carrierFailed: number }> { + const ids = [ + ...new Set(input.orderItemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0)), + ] + if (ids.length === 0) return { updated: 0, carrierFailed: 0 } + + const items = ( + await query(`SELECT * FROM order_items WHERE id = ANY($1::bigint[])`, [ids]) + ).rows + const orderIds = [...new Set(items.map((item) => Number(item.order_id)))] + const orders = new Map( + (await Promise.all(orderIds.map((orderId) => getOrderById(orderId)))) + .filter((order): order is NonNullable => Boolean(order)) + .map((order) => [Number(order.id), order]), + ) + const assignedIds: number[] = [] + let carrierFailed = 0 + + for (const item of items) { + if (item.fulfillment_status === 'archived') continue + const order = orders.get(Number(item.order_id)) + if (!order) continue + + if ( + item.fulfillment_status === 'assigned' && + item.fulfillment_channel && + item.fulfillment_channel !== input.channel + ) { + // 旧载体尚未撤销时禁止直接换渠道，否则会同时留下两个履约执行方。 + carrierFailed += 1 + await query( + `UPDATE order_items + SET fulfillment_reason = $2, updated_at = NOW() + WHERE id = $1 AND fulfillment_status = 'assigned'`, + [item.id, '切换渠道前需先撤销现有履约载体'], + ) + continue + } + + let carrier + try { + carrier = await ensureCarrierForOrderItem({ order, item, channel: input.channel }) + } catch (error) { + carrier = { + ready: false, + created: false, + reason: 'exception', + message: error instanceof Error ? error.message : String(error), + } + } + + if (!carrier.ready) { + carrierFailed += 1 + if (item.fulfillment_status === 'assigned') { + // 运营切换渠道失败时恢复原渠道，宁可保留旧状态，也不制造渠道与载体不一致。 + await query( + `UPDATE order_items + SET fulfillment_channel = $2, + fulfillment_status = $3, + fulfillment_source = $4, + fulfillment_locked = $5, + fulfillment_reason = $6, + fulfillment_pending_reason = $7, + fulfillment_pending_detail = $8::jsonb, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status <> 'archived'`, + [ + item.id, + item.fulfillment_channel, + item.fulfillment_status, + item.fulfillment_source, + item.fulfillment_locked, + item.fulfillment_reason, + item.fulfillment_pending_reason, + JSON.stringify(item.fulfillment_pending_detail || null), + ], + ) + } else { + await query( + `UPDATE order_items + SET fulfillment_channel = NULL, + fulfillment_status = 'pending', + fulfillment_source = '', + fulfillment_locked = TRUE, + fulfillment_pending_reason = 'no_carrier', + fulfillment_pending_detail = $2::jsonb, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status <> 'archived'`, + [ + item.id, + JSON.stringify({ + channel: input.channel, + reason: carrier.reason, + message: carrier.message, + }), + ], + ) + } + continue + } + + const updated = await query( + `UPDATE order_items + SET fulfillment_channel = $2, + fulfillment_status = 'assigned', + fulfillment_source = $3, + fulfillment_locked = $4, + fulfillment_reason = COALESCE(NULLIF($5, ''), fulfillment_reason), + fulfillment_pending_reason = NULL, + fulfillment_pending_detail = NULL, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status <> 'archived'`, + [item.id, input.channel, input.source, input.lock, String(input.note || '').slice(0, 200)], + ) + if ((updated.rowCount || 0) > 0) assignedIds.push(Number(item.id)) + } + + if (assignedIds.length > 0) { + await normalizeProductsForItems(assignedIds).catch(() => undefined) + const cost = await syncCostSnapshotsForItems(assignedIds).catch(() => null) + const dates = [ + ...new Set([...(await orderItemBasisDates(assignedIds)), ...(cost?.affectedDates || [])]), + ] + await refreshFactsForDates(dates).catch(() => undefined) + } + const final = await query<{ count: number }>( + `WITH updated AS ( + UPDATE order_items + SET fulfillment_source = $3, + fulfillment_locked = $4, + fulfillment_pending_reason = NULL, + fulfillment_pending_detail = NULL, + updated_at = NOW() + WHERE id = ANY($1::bigint[]) + AND fulfillment_status = 'assigned' + AND fulfillment_channel = $2 + RETURNING id + ) + SELECT count(*)::int AS count FROM updated`, + [ids, input.channel, input.source, input.lock], + ) + return { updated: Number(final.rows[0]?.count || assignedIds.length), carrierFailed } +} + +/** 运营指定渠道：先保证真实载体，成功后才写入并锁定。 */ export async function assignChannelToOrderItems(input: { orderItemIds: number[] channel: string adminUserId: number | null note?: string -}): Promise<{ updated: number }> { +}): Promise<{ updated: number; carrierFailed: number }> { if (!isFulfillmentChannel(input.channel)) { throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) } @@ -384,35 +594,20 @@ export async function assignChannelToOrderItems(input: { throw createHttpError('请选择订单项', { statusCode: 400, errorCode: 'order_item_required' }) } - const result = await query( - ` - UPDATE order_items - SET - fulfillment_channel = $2, - fulfillment_status = 'assigned', - fulfillment_source = 'operator_assigned', - fulfillment_locked = TRUE, - fulfillment_decided_by = $3, - fulfillment_decided_at = NOW(), - fulfillment_reason = COALESCE(NULLIF($4, ''), fulfillment_reason), - updated_at = NOW() - WHERE id = ANY($1::bigint[]) - AND fulfillment_status <> 'archived' - `, - [ids, input.channel, input.adminUserId, String(input.note || '').slice(0, 200)], + const result = await assignOrderItemsAfterCarrierReady({ + orderItemIds: ids, + channel: input.channel, + source: 'operator_assigned', + lock: true, + note: input.note, + }) + await query( + `UPDATE order_items + SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW() + WHERE id = ANY($1::bigint[])`, + [ids, input.adminUserId], ) - - const updated = result.rowCount || 0 - if (updated > 0) { - // 运营指定后先归一商品，再立即算成本并回填事实，避免旧单等冷通道。 - await normalizeProductsForItems(ids).catch(() => undefined) - const cost = await syncCostSnapshotsForItems(ids).catch(() => null) - const dates = [ - ...new Set([...(await orderItemBasisDates(ids)), ...(cost?.affectedDates || [])]), - ] - await refreshFactsForDates(dates).catch(() => undefined) - } - return { updated } + return result } /** 运营解锁：允许自动判定重新接管。 */ @@ -507,7 +702,7 @@ export async function assignChannelByInternalSku(input: { adminUserId: number | null note?: string createRule?: boolean -}): Promise<{ updated: number; ruleCreated: boolean }> { +}): Promise<{ updated: number; ruleCreated: boolean; carrierFailed: number }> { if (!isFulfillmentChannel(input.channel)) { throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) } @@ -521,26 +716,35 @@ export async function assignChannelByInternalSku(input: { const idsResult = await query<{ id: number }>( ` - UPDATE order_items oi - SET - fulfillment_channel = $4, - fulfillment_status = 'assigned', - fulfillment_source = 'operator_assigned', - fulfillment_locked = TRUE, - fulfillment_decided_by = $5, - fulfillment_decided_at = NOW(), - updated_at = NOW() + SELECT oi.id FROM orders o + JOIN order_items oi ON oi.order_id = o.id WHERE o.id = oi.order_id AND o.provider = $1 AND o.shop_id = $2 AND ${INTERNAL_SKU_SQL} = $3 - AND oi.fulfillment_status = 'pending' - RETURNING oi.id + AND ( + oi.fulfillment_status = 'pending' + OR (oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = $4) + ) `, - [input.provider, input.shopId, internalSku, input.channel, input.adminUserId], + [input.provider, input.shopId, internalSku, input.channel], ) + const product = await query<{ id: number | null }>( + `SELECT MIN(id) AS id + FROM product_catalog + WHERE status = 'active' AND btrim(canonical_name) = $1`, + [internalSku], + ) + await setInternalSkuPrimaryChannel({ + internalSku, + channel: input.channel, + canonicalProductId: product.rows[0]?.id == null ? null : Number(product.rows[0].id), + source: 'operator', + configJson: { note: String(input.note || '').slice(0, 200) }, + }) + let ruleCreated = false if (input.createRule !== false) { const rule = await upsertChannelRule({ @@ -559,16 +763,21 @@ export async function assignChannelByInternalSku(input: { } const ids = idsResult.rows.map((row) => Number(row.id)) - if (ids.length > 0) { - await normalizeProductsForItems(ids).catch(() => undefined) - const cost = await syncCostSnapshotsForItems(ids).catch(() => null) - const dates = [ - ...new Set([...(await orderItemBasisDates(ids)), ...(cost?.affectedDates || [])]), - ] - await refreshFactsForDates(dates).catch(() => undefined) - } + const applied = await assignOrderItemsAfterCarrierReady({ + orderItemIds: ids, + channel: input.channel, + source: 'operator_assigned', + lock: true, + note: input.note, + }) + await query( + `UPDATE order_items + SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW() + WHERE id = ANY($1::bigint[])`, + [ids, input.adminUserId], + ) - return { updated: ids.length, ruleCreated } + return { ...applied, ruleCreated } } /** 按 SKU 批量指定（待分类队列的主操作）。 */ @@ -580,31 +789,39 @@ export async function assignChannelBySku(input: { adminUserId: number | null note?: string createRule?: boolean -}): Promise<{ updated: number; ruleCreated: boolean }> { +}): Promise<{ updated: number; ruleCreated: boolean; carrierFailed: number }> { if (!isFulfillmentChannel(input.channel)) { throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) } - const result = await query( + const idsResult = await query<{ id: number; internal_sku: string }>( ` - UPDATE order_items oi - SET - fulfillment_channel = $4, - fulfillment_status = 'assigned', - fulfillment_source = 'operator_assigned', - fulfillment_locked = TRUE, - fulfillment_decided_by = $5, - fulfillment_decided_at = NOW(), - updated_at = NOW() + SELECT oi.id, ${INTERNAL_SKU_SQL} AS internal_sku FROM orders o + JOIN order_items oi ON oi.order_id = o.id WHERE o.id = oi.order_id AND o.provider = $1 AND o.shop_id = $2 AND oi.sku_code = $3 - AND oi.fulfillment_status = 'pending' + AND ( + oi.fulfillment_status = 'pending' + OR (oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = $4) + ) `, - [input.provider, input.shopId, input.skuCode, input.channel, input.adminUserId], + [input.provider, input.shopId, input.skuCode, input.channel], ) + const internalSkus = [ + ...new Set(idsResult.rows.map((row) => String(row.internal_sku || '').trim()).filter(Boolean)), + ] + if (internalSkus.length === 1) { + await setInternalSkuPrimaryChannel({ + internalSku: internalSkus[0]!, + channel: input.channel, + source: 'operator', + configJson: { note: String(input.note || '').slice(0, 200) }, + }) + } + // 默认同时落一条渠道规则：否则同 SKU 的下一单还是会进待分类。 let ruleCreated = false if (input.createRule !== false) { @@ -623,7 +840,21 @@ export async function assignChannelBySku(input: { ruleCreated = rule.id > 0 } - return { updated: result.rowCount || 0, ruleCreated } + const ids = idsResult.rows.map((row) => Number(row.id)) + const applied = await assignOrderItemsAfterCarrierReady({ + orderItemIds: ids, + channel: input.channel, + source: 'operator_assigned', + lock: true, + note: input.note, + }) + await query( + `UPDATE order_items + SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW() + WHERE id = ANY($1::bigint[])`, + [ids, input.adminUserId], + ) + return { ...applied, ruleCreated } } export async function listChannelRules( @@ -712,6 +943,18 @@ export async function upsertChannelRule(input: { ], ) + if (input.matchKeyType === 'internal_sku') { + await upsertInternalSkuChannel({ + internalSku: matchKey, + channel: input.channel, + priority: Math.trunc(Number(input.priority) || 100), + isFallback: Math.trunc(Number(input.priority) || 100) >= 90, + enabled: input.enabled !== false, + source: source === 'operator' ? 'operator' : `rule:${source}`, + replacePriority: true, + }) + } + return { id: result.rows[0]!.id } } @@ -720,8 +963,25 @@ export async function deleteChannelRule(ruleId: number): Promise<{ deleted: bool if (id <= 0) { throw createHttpError('规则 id 无效', { statusCode: 400, errorCode: 'invalid_rule_id' }) } - const result = await query(`DELETE FROM sku_channel_rules WHERE id = $1 RETURNING id`, [id]) - return { deleted: (result.rowCount || 0) > 0 } + const result = await query<{ + id: number + match_key_type: string + match_key: string + channel: string + }>( + `DELETE FROM sku_channel_rules + WHERE id = $1 + RETURNING id, match_key_type, match_key, channel`, + [id], + ) + const deleted = result.rows[0] + if (deleted?.match_key_type === 'internal_sku' && isFulfillmentChannel(deleted.channel)) { + await disableInternalSkuChannel({ + internalSku: deleted.match_key, + channel: deleted.channel, + }) + } + return { deleted: Boolean(deleted) } } async function loadEnabledRules(): Promise { @@ -735,6 +995,32 @@ async function loadEnabledRules(): Promise { return rows.rows } +async function loadInternalSkuChannelsForKeys( + keys: string[], +): Promise> { + const normalized = [...new Set(keys.map(normalizeKey).filter(Boolean))] + const result = new Map() + if (normalized.length === 0) return result + + const rows = await query( + ` + SELECT internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, config_json, source + FROM internal_sku_channels + WHERE enabled = TRUE AND btrim(internal_sku) = ANY($1::text[]) + ORDER BY internal_sku, priority, channel + `, + [normalized], + ) + for (const row of rows.rows) { + const key = normalizeKey(row.internal_sku) + const list = result.get(key) || [] + list.push(row) + result.set(key, list) + } + return result +} + export const CHANNEL_RULE_KEY_TYPES = MATCH_KEY_TYPES export const CHANNEL_RULE_SOURCES = RULE_SOURCES export const CHANNEL_VALUES = FULFILLMENT_CHANNELS diff --git a/apps/backend/src/services/fulfillment/channel-types.ts b/apps/backend/src/services/fulfillment/channel-types.ts index ccaaaa74..26153db2 100644 --- a/apps/backend/src/services/fulfillment/channel-types.ts +++ b/apps/backend/src/services/fulfillment/channel-types.ts @@ -27,6 +27,7 @@ export const FULFILLMENT_SOURCES = [ 'manual_confirm', 'operator_assigned', 'cutover_archive', + 'affiliate_fallback', ] as const export type FulfillmentSource = (typeof FULFILLMENT_SOURCES)[number] | '' diff --git a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts new file mode 100644 index 00000000..3f33d956 --- /dev/null +++ b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts @@ -0,0 +1,136 @@ +/** + * 履约路由状态机集成测试（需要真实 PostgreSQL）。 + * + * 运行： + * FULFILLMENT_ROUTING_TEST_DATABASE_URL="$DATABASE_URL" \ + * npx tsx --test src/services/fulfillment/fulfillment-routing.integration.test.ts + * + * 未设置环境变量时跳过，普通 npm test 不依赖数据库。 + */ +import assert from 'node:assert/strict' +import test from 'node:test' + +const TEST_DATABASE_URL = String(process.env.FULFILLMENT_ROUTING_TEST_DATABASE_URL || '').trim() + +test( + '运营指定渠道：先保证载体，再写 assigned；缺载体保持 pending', + { skip: !TEST_DATABASE_URL ? '未设置 FULFILLMENT_ROUTING_TEST_DATABASE_URL' : false }, + async () => { + process.env.DATABASE_URL = TEST_DATABASE_URL + const { closeDb, query } = await import('../../db/client.js') + const { assignChannelByInternalSku } = await import('./channel-assign-service.js') + const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}` + const shopId = `it-routing-${suffix}` + const manualSku = `集成测试人工-${suffix}` + const missingAffiliateSku = `集成测试自动-${suffix}` + const orderIds: number[] = [] + const orderItemIds: number[] = [] + const internalSkus = [manualSku, missingAffiliateSku] + + async function createPendingOrderItem(sku: string) { + const order = await query<{ id: number }>( + `INSERT INTO orders ( + provider, platform, shop_id, platform_order_id, order_status, pay_status, + total_amount, paid_at, created_at, updated_at + ) VALUES ( + '91kaquan', 'kuaishou-91', $1, $2, 'paid', 'paid', 1000, NOW(), NOW(), NOW() + ) RETURNING id`, + [shopId, `IT-ROUTING-${suffix}-${orderItemIds.length + 1}`], + ) + const orderId = Number(order.rows[0]!.id) + orderIds.push(orderId) + const item = await query<{ id: number }>( + `INSERT INTO order_items ( + order_id, sku_code, sku_name, quantity, spec_json, + revenue_amount_fen, revenue_basis, created_at, updated_at + ) VALUES ($1, $2, $2, 1, $3::jsonb, 1000, 'order_total', NOW(), NOW()) + RETURNING id`, + [orderId, sku, JSON.stringify({ skuNick: sku })], + ) + const orderItemId = Number(item.rows[0]!.id) + orderItemIds.push(orderItemId) + return orderItemId + } + + try { + const manualItemId = await createPendingOrderItem(manualSku) + await query( + `INSERT INTO manual_fulfillment_configs ( + sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at + ) VALUES ($1, 'nanhang', 0, 'variable', TRUE, '集成测试', NOW(), NOW())`, + [manualSku], + ) + + const manualAssigned = await assignChannelByInternalSku({ + provider: '91kaquan', + shopId, + internalSku: manualSku, + channel: 'manual', + adminUserId: null, + note: '集成测试人工', + }) + assert.equal(manualAssigned.updated, 1) + assert.equal(manualAssigned.carrierFailed, 0) + const manualItem = await query<{ + fulfillment_status: string + fulfillment_channel: string + fulfillment_config_id: number | null + }>( + `SELECT fulfillment_status, fulfillment_channel, fulfillment_config_id + FROM order_items WHERE id = $1`, + [manualItemId], + ) + assert.equal(manualItem.rows[0]?.fulfillment_status, 'assigned') + assert.equal(manualItem.rows[0]?.fulfillment_channel, 'manual') + assert.ok(Number(manualItem.rows[0]?.fulfillment_config_id) > 0) + + const missingAffiliateItemId = await createPendingOrderItem(missingAffiliateSku) + const affiliateFailed = await assignChannelByInternalSku({ + provider: '91kaquan', + shopId, + internalSku: missingAffiliateSku, + channel: 'affiliate_dash', + adminUserId: null, + note: '集成测试缺映射', + }) + assert.equal(affiliateFailed.updated, 0) + assert.equal(affiliateFailed.carrierFailed, 1) + const pendingItem = await query<{ + fulfillment_status: string + fulfillment_channel: string | null + fulfillment_pending_reason: string | null + }>( + `SELECT fulfillment_status, fulfillment_channel, fulfillment_pending_reason + FROM order_items WHERE id = $1`, + [missingAffiliateItemId], + ) + assert.equal(pendingItem.rows[0]?.fulfillment_status, 'pending') + assert.equal(pendingItem.rows[0]?.fulfillment_channel, null) + assert.equal(pendingItem.rows[0]?.fulfillment_pending_reason, 'no_carrier') + } finally { + if (orderItemIds.length > 0) { + await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [ + orderItemIds, + ]) + await query(`DELETE FROM cost_snapshots WHERE order_item_id = ANY($1::bigint[])`, [ + orderItemIds, + ]) + await query(`DELETE FROM order_items WHERE id = ANY($1::bigint[])`, [orderItemIds]) + } + if (orderIds.length > 0) { + await query(`DELETE FROM orders WHERE id = ANY($1::bigint[])`, [orderIds]) + } + await query(`DELETE FROM cost_exceptions WHERE scope_key LIKE $1`, [`%${suffix}%`]) + await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = $1`, [manualSku]) + await query(`DELETE FROM internal_sku_channels WHERE internal_sku = ANY($1::text[])`, [ + internalSkus, + ]) + await query( + `DELETE FROM sku_channel_rules + WHERE provider = '91kaquan' AND shop_id = $1 AND match_key = ANY($2::text[])`, + [shopId, internalSkus], + ) + await closeDb() + } + }, +) 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 0210b7c2..635a8d32 100644 --- a/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts +++ b/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts @@ -1,10 +1,9 @@ -import { - listInternalSkuChannels, - type InternalSkuChannelRow, -} from './internal-sku-channel-service.js' +import { listInternalSkuChannels, planInternalSkuChannels } from './internal-sku-channel-service.js' import type { ResolvedFulfillmentItem } from './product-resolution-service.js' import { matchAffiliateDashSkuByInternalSku } from './product-resolution-service.js' import { logWarn } from '../../utils/logger.js' +import type { FulfillmentChannel } from './channel-types.js' +import { resolveInternalSkuKey } from './internal-sku-resolver.js' type IntakeChannelDecisionInput = { provider?: string @@ -18,14 +17,6 @@ function parseObject(value: unknown): Record { : {} } -function resolveInternalSku(item: ResolvedFulfillmentItem): string { - const snapshot = parseObject(item.snapshot) - const spec = parseObject(item.spec) - return String( - snapshot.internalSku || spec.skuNick || spec.outSkuSn || item.skuCode || item.skuName || '', - ).trim() -} - /** * 下单入口读取内部 SKU 渠道决策。 * @@ -36,13 +27,14 @@ function resolveInternalSku(item: ResolvedFulfillmentItem): string { export async function applyIntakeChannelDecision( item: ResolvedFulfillmentItem, _input: IntakeChannelDecisionInput, + options: { preferredChannel?: FulfillmentChannel } = {}, ): Promise { - const internalSku = resolveInternalSku(item) + const internalSku = resolveInternalSkuKey(item) if (!internalSku) return item - let channels: InternalSkuChannelRow[] = [] + let plan try { - channels = await listInternalSkuChannels(internalSku) + plan = planInternalSkuChannels(await listInternalSkuChannels(internalSku)) } catch (error) { // 单测/旧库没有 118 迁移时不影响下单主流程，生产由迁移保证。 logWarn('[intake-channel-decision]', '读取内部 SKU 渠道失败', { @@ -51,14 +43,32 @@ export async function applyIntakeChannelDecision( }) return item } - if (channels.length === 0) return item + if (plan.conflicts.length > 0) { + return { + ...item, + snapshot: { + ...parseObject(item.snapshot), + internalSku, + internalSkuChannelReason: 'routing_conflict', + internalSkuConflictChannels: plan.conflicts.map((channel) => channel.channel), + }, + } + } + + const preferred = options.preferredChannel + ? plan.attempts.find((candidate) => candidate.channel === options.preferredChannel) || + (options.preferredChannel === 'manual' || options.preferredChannel === 'third_party_voucher' + ? null + : undefined) + : undefined + const primary = preferred === null ? null : preferred || plan.primary + if (!primary) return item const snapshot = parseObject(item.snapshot) const route = parseObject(snapshot.fulfillmentRoute) const selected = String(route.selectedExecutorKey || '').trim() // 路由配置未命中时会默认落 manual_dispatch；这不是运营显式规则，不能覆盖内部 SKU 能力。 const isDefaultManualFallback = selected === 'manual_dispatch' && !route.matchedRule - const primary = channels[0]! const nextSnapshot: Record = { ...snapshot, internalSku: String(snapshot.internalSku || internalSku), @@ -69,7 +79,8 @@ export async function applyIntakeChannelDecision( const canOverrideDefaultManual = !selected || isDefaultManualFallback const existingAffiliateSku = String(parseObject(snapshot.affiliateDash).sku || '').trim() - if (primary.channel === 'affiliate_dash' || selected === 'affiliate_dash') { + const desiredChannel = options.preferredChannel || primary.channel + if (desiredChannel === 'affiliate_dash' || selected === 'affiliate_dash') { const dashSku = existingAffiliateSku || matchAffiliateDashSkuByInternalSku(internalSku) if (dashSku) { if (canOverrideDefaultManual || selected === 'affiliate_dash') { @@ -109,17 +120,27 @@ export async function applyIntakeChannelDecision( // 内部能力是接单/卡券，但路由只是默认人工兜底：清掉人工任务，交给工单或财务能力兜底。 if ( - (primary.channel === 'work_order' || primary.channel === 'third_party_voucher') && + (desiredChannel === 'work_order' || desiredChannel === 'third_party_voucher') && isDefaultManualFallback ) { nextSnapshot.fulfillmentRoute = { ...route, selectedExecutorKey: '', - reason: `内部 SKU 渠道能力为 ${primary.channel}，不使用默认人工兜底`, + reason: `内部 SKU 渠道能力为 ${desiredChannel}，不使用默认人工兜底`, matchedRule: null, } return { ...item, isConfigured: false, snapshot: nextSnapshot } } + if (desiredChannel === 'manual' && isDefaultManualFallback) { + nextSnapshot.fulfillmentRoute = { + ...route, + selectedExecutorKey: 'manual_dispatch', + reason: '内部 SKU 渠道能力指定人工客服', + matchedRule: null, + } + return { ...item, isConfigured: true, snapshot: nextSnapshot } + } + return { ...item, snapshot: nextSnapshot } } 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 new file mode 100644 index 00000000..0bd3df58 --- /dev/null +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts @@ -0,0 +1,62 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + planInternalSkuChannels, + type InternalSkuChannelRow, +} from './internal-sku-channel-service.js' + +function channel(patch: Partial): InternalSkuChannelRow { + return { + internal_sku: '内部SKU', + channel: 'manual', + priority: 10, + is_fallback: false, + enabled: true, + canonical_product_id: null, + config_json: {}, + source: 'test', + ...patch, + } +} + +test('渠道计划：主渠道失败后按优先级使用显式兜底', () => { + const plan = planInternalSkuChannels([ + channel({ channel: 'manual', priority: 90, is_fallback: true }), + channel({ channel: 'affiliate_dash', priority: 10 }), + ]) + assert.equal(plan.primary?.channel, 'affiliate_dash') + assert.deepEqual( + plan.attempts.map((item) => item.channel), + ['affiliate_dash', 'manual'], + ) +}) + +test('渠道计划：同级多主渠道是冲突，不能静默按字母选一个', () => { + const plan = planInternalSkuChannels([ + channel({ channel: 'manual', priority: 10 }), + channel({ channel: 'work_order', priority: 10 }), + ]) + assert.equal(plan.primary, null) + assert.deepEqual(plan.conflicts.map((item) => item.channel).sort(), ['manual', 'work_order']) +}) + +test('渠道计划：非兜底渠道即使优先级数值更大也压过兜底渠道', () => { + const plan = planInternalSkuChannels([ + channel({ channel: 'manual', priority: 1, is_fallback: true }), + channel({ channel: 'work_order', priority: 50 }), + ]) + assert.equal(plan.primary?.channel, 'work_order') + assert.equal(plan.fallbacks[0]?.channel, 'manual') +}) + +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'], + ) +}) 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 e55de975..3dfdb563 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts @@ -1,5 +1,6 @@ -import { query } from '../../db/client.js' +import { query, withTransaction } from '../../db/client.js' import type { FulfillmentChannel } from './channel-types.js' +import { compareFulfillmentChannel } from './internal-sku-resolver.js' export type InternalSkuChannelRow = { internal_sku: string @@ -8,9 +9,17 @@ export type InternalSkuChannelRow = { is_fallback: boolean enabled: boolean canonical_product_id: number | null + config_json: Record source: string } +export type InternalSkuChannelPlan = { + primary: InternalSkuChannelRow | null + fallbacks: InternalSkuChannelRow[] + attempts: InternalSkuChannelRow[] + conflicts: InternalSkuChannelRow[] +} + export async function listInternalSkuChannels( internalSku: string, ): Promise { @@ -18,18 +27,12 @@ export async function listInternalSkuChannels( if (!key) return [] const result = await query( ` - SELECT internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source + SELECT internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, + config_json, source FROM internal_sku_channels WHERE btrim(internal_sku) = $1 AND enabled = TRUE ORDER BY priority ASC, - CASE channel - WHEN 'affiliate_dash' THEN 1 - WHEN 'work_order' THEN 2 - WHEN 'third_party_voucher' THEN 3 - WHEN 'manual' THEN 4 - ELSE 9 - END, channel ASC `, [key], @@ -37,15 +40,44 @@ export async function listInternalSkuChannels( return result.rows } -/** 财务侧可直接兜底的渠道：卡券/人工不要求新建履约任务。 */ -export async function resolveInternalSkuFallbackChannel( - internalSku: string, -): Promise { - const channels = await listInternalSkuChannels(internalSku) - const safe = channels.find( - (row) => row.channel === 'third_party_voucher' || row.channel === 'manual', +/** + * 渠道计划：同一时间只允许一个主渠道；同级多渠道视为配置冲突。 + * is_fallback 渠道只在主渠道不可用或失败后按 priority 顺序尝试。 + */ +export function planInternalSkuChannels(input: InternalSkuChannelRow[]): InternalSkuChannelPlan { + const channels = input + .filter((row) => row.enabled) + .slice() + .sort( + (a, b) => + a.priority - b.priority || + Number(a.is_fallback) - Number(b.is_fallback) || + compareFulfillmentChannel(a.channel, b.channel), + ) + + const explicitPrimary = channels.filter((row) => !row.is_fallback) + const primaryPool = explicitPrimary.length > 0 ? explicitPrimary : channels + if (primaryPool.length === 0) { + return { primary: null, fallbacks: [], attempts: [], conflicts: [] } + } + + const topPriority = Math.min(...primaryPool.map((row) => row.priority)) + const top = primaryPool.filter((row) => row.priority === topPriority) + const distinctTop = top.filter( + (row, index) => top.findIndex((candidate) => candidate.channel === row.channel) === index, ) - return safe?.channel || null + if (distinctTop.length > 1) { + return { primary: null, fallbacks: [], attempts: [], conflicts: distinctTop } + } + + const primary = distinctTop[0] || null + const fallbacks = channels.filter((row) => row.channel !== primary?.channel && row.is_fallback) + return { + primary, + fallbacks, + attempts: primary ? [primary, ...fallbacks] : [], + conflicts: [], + } } export async function upsertInternalSkuChannel(input: { @@ -53,31 +85,110 @@ export async function upsertInternalSkuChannel(input: { channel: FulfillmentChannel priority?: number isFallback?: boolean + enabled?: boolean canonicalProductId?: number | null + configJson?: Record | undefined source?: string + replacePriority?: boolean }): Promise { const internalSku = String(input.internalSku || '').trim() if (!internalSku) return - await query( - ` + await withTransaction(async (client) => { + if (input.enabled !== false && input.isFallback !== true) { + await client.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, input.channel], + ) + } + await client.query( + ` INSERT INTO internal_sku_channels ( internal_sku, channel, priority, is_fallback, enabled, - canonical_product_id, source, created_at, updated_at - ) VALUES ($1, $2, $3, $4, TRUE, $5, $6, NOW(), NOW()) + canonical_product_id, config_json, source, created_at, updated_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, NOW(), NOW()) ON CONFLICT (internal_sku, channel) DO UPDATE SET - priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), - is_fallback = internal_sku_channels.is_fallback AND EXCLUDED.is_fallback, + priority = CASE + WHEN $9::boolean THEN EXCLUDED.priority + ELSE LEAST(internal_sku_channels.priority, EXCLUDED.priority) + END, + is_fallback = EXCLUDED.is_fallback, + enabled = EXCLUDED.enabled, canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), - enabled = TRUE, + config_json = internal_sku_channels.config_json || EXCLUDED.config_json, + source = EXCLUDED.source, updated_at = NOW() + `, + [ + internalSku, + input.channel, + Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100, + input.isFallback === true, + input.enabled !== false, + input.canonicalProductId ?? null, + JSON.stringify(input.configJson || {}), + String(input.source || 'auto').trim() || 'auto', + input.replacePriority === true, + ], + ) + }) +} + +/** + * 运营指定一个主渠道。其他渠道保留为显式兜底，避免下一次又被同级优先级抢占。 + */ +export async function setInternalSkuPrimaryChannel(input: { + internalSku: string + channel: FulfillmentChannel + canonicalProductId?: number | null + source?: string + configJson?: Record | undefined +}): Promise { + const internalSku = String(input.internalSku || '').trim() + if (!internalSku) return + + await query( + ` + UPDATE internal_sku_channels + SET priority = GREATEST(priority, 90), + is_fallback = TRUE, + updated_at = NOW() + WHERE btrim(internal_sku) = $1 + AND channel <> $2 + AND enabled = TRUE `, - [ - internalSku, - input.channel, - Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100, - input.isFallback === true, - input.canonicalProductId ?? null, - String(input.source || 'auto').trim() || 'auto', - ], + [internalSku, input.channel], ) + await upsertInternalSkuChannel({ + internalSku, + channel: input.channel, + priority: 10, + isFallback: false, + enabled: true, + canonicalProductId: input.canonicalProductId ?? null, + configJson: input.configJson, + source: input.source || 'operator', + replacePriority: true, + }) +} + +export async function disableInternalSkuChannel(input: { + internalSku: string + channel: FulfillmentChannel +}): Promise { + const result = await query( + ` + UPDATE internal_sku_channels + SET enabled = FALSE, updated_at = NOW() + WHERE btrim(internal_sku) = $1 AND channel = $2 + `, + [String(input.internalSku || '').trim(), input.channel], + ) + return result.rowCount || 0 } diff --git a/apps/backend/src/services/fulfillment/internal-sku-mapping-service.ts b/apps/backend/src/services/fulfillment/internal-sku-mapping-service.ts index fb53621f..19b7245b 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-mapping-service.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-mapping-service.ts @@ -1,4 +1,5 @@ import { query } from '../../db/client.js' +import { containsCjk } from './internal-sku-resolver.js' export type InternalSkuMappingMatchType = | 'sku_code' @@ -114,10 +115,6 @@ export async function upsertInternalSkuMapping(input: { return Number(result.rows[0]?.id || 0) } -function containsCjk(value: string): boolean { - return /[\u4e00-\u9fa5]/.test(value) -} - /** * 订单项级内部 SKU 补全： * - 现有 skuNick/outSkuSn/skuCode/skuName 已有中文 → 不查库； diff --git a/apps/backend/src/services/fulfillment/internal-sku-resolver.test.ts b/apps/backend/src/services/fulfillment/internal-sku-resolver.test.ts new file mode 100644 index 00000000..a77b0a26 --- /dev/null +++ b/apps/backend/src/services/fulfillment/internal-sku-resolver.test.ts @@ -0,0 +1,50 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { resolveInternalSkuKey } from './internal-sku-resolver.js' + +test('内部 SKU：入口显式值最优先', () => { + assert.equal( + resolveInternalSkuKey({ + snapshot: { internalSku: '映射后的内部SKU' }, + manualSkuKey: '平台码123', + skuName: '商品名', + }), + '映射后的内部SKU', + ) +}) + +test('内部 SKU：纯数字平台码不会压过中文商品名', () => { + assert.equal( + resolveInternalSkuKey({ + manualSkuKey: '5003', + skuCode: '5003', + skuName: '幸运币2个', + spec: { outSkuSn: '5003' }, + }), + '幸运币2个', + ) +}) + +test('内部 SKU：中文 skuNick 优先于 outSkuSn 和平台码', () => { + assert.equal( + resolveInternalSkuKey({ + skuCode: 'english-platform-code', + skuName: '展示标题', + spec: { skuNick: '内部中文SKU', outSkuSn: '规格码' }, + }), + '内部中文SKU', + ) +}) + +test('内部 SKU：没有中文时按历史生成列顺序回退', () => { + assert.equal( + resolveInternalSkuKey({ + manualSkuKey: 'manual-key', + skuCode: 'sku-code', + skuName: 'sku-name', + spec: { skuNick: 'nick', outSkuSn: 'out' }, + }), + 'manual-key', + ) +}) diff --git a/apps/backend/src/services/fulfillment/internal-sku-resolver.ts b/apps/backend/src/services/fulfillment/internal-sku-resolver.ts new file mode 100644 index 00000000..56705824 --- /dev/null +++ b/apps/backend/src/services/fulfillment/internal-sku-resolver.ts @@ -0,0 +1,61 @@ +import type { FulfillmentChannel } from './channel-types.js' + +type JsonObject = Record + +export type InternalSkuResolverInput = { + manualSkuKey?: unknown + skuCode?: unknown + skuName?: unknown + spec?: unknown + snapshot?: unknown +} + +function asObject(value: unknown): JsonObject { + return value && typeof value === 'object' && !Array.isArray(value) ? (value as JsonObject) : {} +} + +export function normalizeInternalSku(value: unknown): string { + return String(value ?? '').trim() +} + +export function containsCjk(value: string): boolean { + return /[\u4e00-\u9fa5]/.test(value) +} + +/** + * 内部 SKU 的唯一解析口径。 + * + * 1. 入口显式写入的 snapshot.internalSku 最权威； + * 2. 旧数据优先使用中文键，避免纯数字平台码压过中文 skuNick/sku_name； + * 3. 没有中文时按生成列的历史顺序回退。 + */ +export function resolveInternalSkuKey(input: InternalSkuResolverInput): string { + const snapshot = asObject(input.snapshot) + const spec = asObject(input.spec) + const explicit = normalizeInternalSku(snapshot.internalSku) + if (explicit) return explicit + + const manual = normalizeInternalSku(input.manualSkuKey) + const skuNick = normalizeInternalSku(spec.skuNick) + const outSkuSn = normalizeInternalSku(spec.outSkuSn) + const skuCode = normalizeInternalSku(input.skuCode) + const skuName = normalizeInternalSku(input.skuName) + + return ( + [manual, skuNick, outSkuSn, skuName].find((value) => value && containsCjk(value)) || + [manual, skuNick, outSkuSn, skuCode, skuName].find(Boolean) || + '' + ) +} + +/** 渠道选择里使用的稳定顺序，仅用于同优先级时生成诊断，不用于静默改渠道。 */ +export const INTERNAL_SKU_CHANNEL_ORDER: readonly FulfillmentChannel[] = [ + 'affiliate_dash', + 'work_order', + 'third_party_voucher', + 'manual', +] + +export function compareFulfillmentChannel(a: FulfillmentChannel, b: FulfillmentChannel): number { + return INTERNAL_SKU_CHANNEL_ORDER.indexOf(a) - INTERNAL_SKU_CHANNEL_ORDER.indexOf(b) +} diff --git a/apps/backend/src/services/fulfillment/product-resolution-service.ts b/apps/backend/src/services/fulfillment/product-resolution-service.ts index b87ec2cb..0db636c6 100644 --- a/apps/backend/src/services/fulfillment/product-resolution-service.ts +++ b/apps/backend/src/services/fulfillment/product-resolution-service.ts @@ -11,6 +11,7 @@ import type { AffiliateDashSkuMapping } from '../../types/runtime-config.js' import { FULFILLMENT_EXECUTOR_KEYS } from './executors/types.js' import { normalizeProductMatchName } from '../order/product-match-utils.js' import { resolveInternalSkuForOrderItem } from './internal-sku-mapping-service.js' +import { resolveInternalSkuKey } from './internal-sku-resolver.js' import { logIntegration } from '../../utils/logger.js' export type FulfillmentItem = { @@ -129,9 +130,19 @@ export async function resolveOrderItemForFulfillment({ skuCode: externalSkuCode, skuName: externalSkuName, }) - const nextSpec = internalSkuMapping + const mappedSpec = internalSkuMapping ? { ...(isPlainObject(item.spec) ? item.spec : {}), skuNick: internalSkuMapping.internalSku } : item.spec + const resolvedInternalSku = resolveInternalSkuKey({ + spec: mappedSpec, + snapshot: item.snapshot, + skuCode: externalSkuCode, + skuName: externalSkuName, + }) + // 统一把内部 SKU 固化到 spec.skuNick，使 manual_sku_key 生成列和所有后续管道使用同一键。 + const nextSpec = resolvedInternalSku + ? { ...(isPlainObject(mappedSpec) ? mappedSpec : {}), skuNick: resolvedInternalSku } + : mappedSpec const snapshot = { ...(isPlainObject(item.snapshot) ? item.snapshot : {}), @@ -154,9 +165,9 @@ export async function resolveOrderItemForFulfillment({ } : null, isConfigured: candidate.isConfigured, + ...(resolvedInternalSku ? { internalSku: resolvedInternalSku } : {}), ...(internalSkuMapping ? { - internalSku: internalSkuMapping.internalSku, internalSkuSource: internalSkuMapping.source, internalSkuMappingId: internalSkuMapping.mappingId, } diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index bafc610d..0a889df3 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -86,9 +86,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { const config = jobConfig(job) const limit = clampInt(config.limit, 200, 1, 2000) const reconciled = await reconcilePendingCarriers({ limit: Math.min(limit, 200) }) - if (reconciled.affiliateTasksCreated > 0) { - await assignPendingOrderItemChannels({ batchSize: 200, maxBatches: 1 }).catch(() => undefined) - } + const repairedAssigned = await reconcilePendingCarriers({ + limit: Math.min(limit, 200), + includeAssignedMissingCarrier: true, + }) + await assignPendingOrderItemChannels({ batchSize: 500, maxBatches: 1 }).catch(() => undefined) const cursor = config.afterId != null ? Math.max(0, Math.trunc(Number(config.afterId))) @@ -101,13 +103,17 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { const manual = await query<{ id: number; sku_key: string; config_id: number }>( ` - SELECT oi.id, oi.manual_sku_key AS sku_key, cfg.id AS config_id + SELECT oi.id, + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key, + cfg.id AS config_id FROM order_items oi - JOIN manual_fulfillment_configs cfg ON cfg.sku_key = oi.manual_sku_key AND cfg.enabled + JOIN manual_fulfillment_configs cfg + ON cfg.sku_key = resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) + AND cfg.enabled WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL - AND oi.manual_sku_key = ANY($3::text[]) + AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = ANY($3::text[]) AND oi.id > $2 ORDER BY oi.id LIMIT $1 @@ -131,12 +137,13 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { // 人工渠道但没有配置：进「人工 SKU 未配置」队列。 const missingManual = await query<{ id: number; sku_key: string }>( ` - SELECT oi.id, oi.manual_sku_key AS sku_key + SELECT oi.id, + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key FROM order_items oi WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL - AND NOT (oi.manual_sku_key = ANY($3::text[])) + AND NOT (resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = ANY($3::text[])) AND NOT EXISTS ( SELECT 1 FROM cost_exception_items i @@ -165,7 +172,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { // 已配置人工 SKU、但还没确认履约的单：进入「人工成本未记录」队列。 const pendingManualCost = await query<{ id: number; sku_key: string }>( ` - SELECT oi.id, oi.manual_sku_key AS sku_key + SELECT oi.id, + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key FROM order_items oi WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' @@ -212,7 +220,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { AND NOT EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id - AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher') + AND ft.executor_key = 'affiliate_dash' ) AND oi.id > $2 ORDER BY oi.id @@ -245,6 +253,39 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { detail: { channel: row.channel, skuCode: row.sku_code }, }) } + + // 已分配订单出现两个真实执行方时不能继续算正常渠道，必须进冲突队列人工处置。 + const conflictingCarriers = await query<{ id: number; channel: string; sku_code: string }>( + ` + SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code + FROM order_items oi + WHERE oi.fulfillment_status = 'assigned' + AND ( + EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + )::int + ( + EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ) + )::int + ( + EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + ) + )::int > 1 + ORDER BY oi.id + LIMIT $1 + `, + [limit], + ) + for (const row of conflictingCarriers.rows) { + exceptionEntries.push({ + reason: 'routing_conflict', + scopeKey: `routing_conflict|${row.channel}|${row.sku_code}|carrier`, + orderItemId: Number(row.id), + detail: { channel: row.channel, skuCode: row.sku_code, issue: 'multiple_carriers' }, + }) + } if (exceptionEntries.length > 0) { await upsertAggregatedExceptionsBatch(exceptionEntries) } @@ -254,6 +295,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { ...missingManual.rows, ...pendingManualCost.rows, ...missingCarrier.rows, + ...conflictingCarriers.rows, ]) { maxId = Math.max(maxId, Number(row.id)) } @@ -269,13 +311,17 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { status: 'ok', message: `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}，` + - `缺履约载体 ${missingCarrier.rows.length}，异常收口 ${resolved}，` + - `自动转人工 ${fallback.switched}`, + `缺履约载体 ${missingCarrier.rows.length}，多载体冲突 ${conflictingCarriers.rows.length}，异常收口 ${resolved}，` + + `自动转人工 ${fallback.switched}，链路补载体 ${reconciled.workOrdersCreated + reconciled.affiliateTasksCreated + repairedAssigned.workOrdersCreated + repairedAssigned.affiliateTasksCreated}，` + + `兜底 ${reconciled.fallbacksSelected + repairedAssigned.fallbacksSelected}`, attached, missingManual: missingManual.rows.length, missingCarrier: missingCarrier.rows.length, + conflictingCarriers: conflictingCarriers.rows.length, resolved, affiliateFallback: fallback.switched, + carrierReconciliation: reconciled, + assignedCarrierRepair: repairedAssigned, } } @@ -289,7 +335,7 @@ async function resolveCarrierDispatchExceptions(): Promise { UPDATE order_items oi SET fulfillment_config_id = cfg.id, updated_at = NOW() FROM manual_fulfillment_configs cfg - WHERE cfg.sku_key = oi.manual_sku_key + WHERE cfg.sku_key = resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AND cfg.enabled AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' @@ -315,12 +361,13 @@ async function resolveCarrierDispatchExceptions(): Promise { WHERE ce.reason = 'carrier_missing' AND ce.status <> 'resolved' AND ( - EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) - OR EXISTS ( + (oi.fulfillment_channel = 'work_order' + AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)) + OR (oi.fulfillment_channel = 'affiliate_dash' AND EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id - AND ft.executor_key IN ('affiliate_dash', 'third_party_voucher') - ) + AND ft.executor_key = 'affiliate_dash' + )) ) LIMIT 500 `, @@ -442,8 +489,9 @@ export async function runLedgerPostJob(job: JsonObject = {}) { /** 顺序执行全部管道，供管理端「一键重算」使用。 */ export async function runFinancePipelineOnce(input: { days?: number } = {}) { - const channel = await runChannelAssignJob({ config: { batchSize: 1000, maxBatches: 50 } }) + // 先保证载体，再让财务镜像真实渠道；避免规则先把订单写成 assigned 却永远没有派单。 const carrier = await runCarrierDispatchJob({ config: { limit: 500 } }) + const channel = await runChannelAssignJob({ config: { batchSize: 1000, maxBatches: 50 } }) const product = await runProductNormalizeJob({ config: { batchSize: 1000, maxBatches: 50 } }) const cost = await runCostSnapshotJob({ config: { batchSize: 1000, maxBatches: 50 } }) const facts = await runSalesFactsJob({ config: { days: input.days ?? 3 } }) diff --git a/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts b/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts index c51353fc..2eec3d7b 100644 --- a/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts +++ b/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts @@ -50,7 +50,12 @@ import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-se export async function syncWorkerOrdersForSourceOrder( order: OrderRow, orderItems: OrderItemRow[], - options: { source?: string; autoOnly?: boolean; sourceMetadata?: JsonObject } = {}, + options: { + source?: string + autoOnly?: boolean + sourceMetadata?: JsonObject + skipChannelAssign?: boolean + } = {}, ) { const [rules, mappings] = await Promise.all([ listWorkProductRules({ enabled: true }), @@ -309,15 +314,17 @@ export async function syncWorkerOrdersForSourceOrder( } // 订单处理完成后立即跑一次渠道判定；失败不阻断下单，由定时任务兜底。 - try { - await assignPendingOrderItemChannels({ batchSize: 200, maxBatches: 1 }) - } catch (error) { - logWarn('[worker-order-consumer]', '履约方式归集失败，等待定时任务兜底', { - orderId: order.id, - platformOrderId: order.platform_order_id, - orderItemCount: orderItems.length, - error: error instanceof Error ? error.message : String(error), - }) + if (!options.skipChannelAssign) { + try { + await assignPendingOrderItemChannels({ batchSize: 200, maxBatches: 1 }) + } catch (error) { + logWarn('[worker-order-consumer]', '履约方式归集失败，等待定时任务兜底', { + orderId: order.id, + platformOrderId: order.platform_order_id, + orderItemCount: orderItems.length, + error: error instanceof Error ? error.message : String(error), + }) + } } const summary = { diff --git a/apps/backend/src/types/repository/rows.ts b/apps/backend/src/types/repository/rows.ts index afdefdca..6eef53cd 100644 --- a/apps/backend/src/types/repository/rows.ts +++ b/apps/backend/src/types/repository/rows.ts @@ -33,6 +33,7 @@ export type OrderItemRow = { quantity: number revenue_amount_fen?: number | string | null canonical_product_id?: number | string | null + manual_sku_key?: string | null spec_json: string | Record item_snapshot_json?: string | Record fulfillment_channel?: string | null @@ -41,6 +42,9 @@ export type OrderItemRow = { fulfillment_locked?: boolean | null fulfillment_decided_at?: string | null fulfillment_reason?: string | null + fulfillment_config_id?: number | null + fulfillment_pending_reason?: string | null + fulfillment_pending_detail?: string | Record | null created_at?: string updated_at?: string } diff --git a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx index 430eea91..a4942e86 100644 --- a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx +++ b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx @@ -100,7 +100,13 @@ function PendingPanel() { channel: selections[rowKey(row)] || '', }), onSuccess: (result) => { - message.success(`已指定 ${result.data.updated} 个订单项`) + if (result.data.carrierFailed > 0) { + message.warning( + `已指定 ${result.data.updated} 个订单项，${result.data.carrierFailed} 个因载体创建失败仍待处理`, + ) + } else { + message.success(`已指定 ${result.data.updated} 个订单项`) + } void queryClient.invalidateQueries({ queryKey: ['fulfillment-pending'] }) }, onError: () => message.error('指定渠道失败'), @@ -135,6 +141,7 @@ function PendingPanel() { no_internal_sku: '无内部SKU', no_channel_capability: '无渠道能力', no_carrier: '无履约载体', + routing_conflict: '渠道冲突', } return value?.length ? value.map((reason) => labels[reason] || reason).join(' / ') : '-' }, diff --git a/apps/frontend/src/services/admin/fulfillment-channel.ts b/apps/frontend/src/services/admin/fulfillment-channel.ts index 3ae7be53..579e6f2a 100644 --- a/apps/frontend/src/services/admin/fulfillment-channel.ts +++ b/apps/frontend/src/services/admin/fulfillment-channel.ts @@ -100,7 +100,10 @@ export function assignChannelBySku(payload: { channel: string note?: string }) { - return apiPost<{ updated: number }>('/api/v1/admin/fulfillment/assign', payload) + return apiPost<{ updated: number; carrierFailed: number; ruleCreated: boolean }>( + '/api/v1/admin/fulfillment/assign', + payload, + ) } export function assignChannelToItems(payload: { @@ -108,7 +111,10 @@ export function assignChannelToItems(payload: { channel: string note?: string }) { - return apiPost<{ updated: number }>('/api/v1/admin/fulfillment/assign', payload) + return apiPost<{ updated: number; carrierFailed: number }>( + '/api/v1/admin/fulfillment/assign', + payload, + ) } export function fetchChannelRules(params: { channel?: string } = {}) { diff --git a/docs/履约与财务分流统一方案-2026-11-02.md b/docs/履约与财务分流统一方案-2026-11-02.md index ba8fd651..b91d9e61 100644 --- a/docs/履约与财务分流统一方案-2026-11-02.md +++ b/docs/履约与财务分流统一方案-2026-11-02.md @@ -1,7 +1,7 @@ # 履约与财务分流统一方案 日期：2026-11-02 -状态：设计稿，待实施 +状态：已落地（2026-11-02）；代码基线：迁移 121 + backend/frontend 当前 HEAD 目标：以内部中文 SKU 为唯一主键，统一「下单路由 → 履约载体 → 财务渠道 → 成本」链路。 ## 1. 现状结论 @@ -220,9 +220,19 @@ pending 只保留： - 后端/前端测试、构建、SQL 护栏通过； - 生产迁移/部署验证。 -## 8. 待确认 +## 8. 最终落地口径（2026-11-02） -1. PDD 纯数字内部 SKU 的中文映射由谁维护（商品名/规格/运营表）？ -2. 双渠道商品（自动发货 + 人工）的兜底触发条件：任务失败、超时、还是人工手动切换？ -3. 接单平台是否所有内部 SKU 都必须自动建工单，还是保留“运营指定”模式？ -4. 同名 man/tpl 商品是否合并，还是保留多载体配置、只统一 internal_sku_channels？ +1. **内部 SKU 只有一个解析函数**：`resolve_internal_sku(manual_sku_key, sku_name, sku_code)`；入口 TS 侧使用同一顺序，不再由各脚本自行拼 SQL。 +2. **渠道配置只有一个运行时入口**：`internal_sku_channels`。一个内部 SKU 最多一个主渠道，其余渠道必须 `is_fallback = true`；数据库唯一索引阻止并列主渠道。 +3. **载体先行**：自动发货/接单先创建 `fulfillment_tasks` / `work_orders`，成功后才写 `fulfillment_status = assigned`。创建失败保持 `pending / no_carrier`。 +4. **兜底触发**：主渠道为自动发货/接单且载体无法创建时，按 priority 尝试显式 fallback；人工配置、卡券为无需外部载体的安全渠道。已分配后自动任务失败仍由 `affiliate_fallback` 接管。 +5. **财务只镜像事实**：已有唯一真实载体优先；否则读取统一渠道计划；旧 `sku_channel_rules` 仅保留兼容回退，新增/修改会同步到统一渠道表。 +6. **人工配置即时入表**：保存/停用人工 SKU 配置会同步 `internal_sku_channels`，历史配置由迁移 120 回填。 +7. **旧规则自动迁移**：可确定的 `sku_code` 自动发货/卡券规则在迁移中转为内部 SKU 渠道并停用旧规则；无法唯一解析的规则由严格预检阻断。 +8. **同名 man/tpl 不强制合并**：保留不同载体能力，但同一内部 SKU 只能配置一个主渠道，其余显式兜底。 + +### 数据与回填约定 + +- PDD 纯数字码优先由 `internal_sku_mappings` 维护；未命中时使用中文 `sku_name`，仍无中文则保留原码并进入待分类。 +- 所有历史人工配置进入 `manual` 兜底能力；存在自动/接单/卡券能力时，人工 priority 固定不低于 90。 +- `npm run products:merge-duplicates`、`cleanup-legacy-sku-code-rules`、`reclassify-post-cutover` 保持 dry-run 默认；执行前后都运行严格预检。 diff --git a/docs/生产部署核对清单-2026-11-02.md b/docs/生产部署核对清单-2026-11-02.md index 350dc0f7..9d97dd28 100644 --- a/docs/生产部署核对清单-2026-11-02.md +++ b/docs/生产部署核对清单-2026-11-02.md @@ -1,7 +1,7 @@ # 生产部署核对清单（履约与财务分流统一） 日期：2026-11-02 -适用：生产当前在迁移 115，本次新增 116–119 + backend/frontend。 +适用：生产当前在迁移 115，本次新增 116–121 + backend/frontend。 ## 1. 部署前 @@ -24,6 +24,8 @@ 117_internal_sku_mappings 118_internal_sku_channels 119_fulfillment_pending_reason +120_fulfillment_routing_invariants +121_internal_sku_single_primary ``` 重点检查： @@ -32,10 +34,12 @@ - 118：按商品能力/规则回填 `internal_sku_channels`； - SQL 已做聚合，避免同名商品同渠道 `ON CONFLICT` 二次更新报错； - 119：只加列和索引，风险低。 +- 120：建立统一内部 SKU SQL 函数、回填人工配置、迁移可确定的旧 sku_code 规则。 +- 121：建立唯一主渠道索引；若仍有并列主渠道会主动阻断迁移。 ## 3. 部署顺序 -1. 执行迁移 116–119； +1. 执行迁移 116–121； 2. 部署 backend； 3. 部署 frontend； 4. 启动/恢复定时任务。 @@ -56,6 +60,7 @@ GROUP BY 1; - `no_internal_sku` - `no_channel_capability` - `no_carrier` +- `routing_conflict`（如仍有配置冲突，必须处理） ### 11月改名卡 @@ -110,6 +115,25 @@ WHERE enabled GROUP BY 1; ``` +### 严格预检 + +```bash +# 本地演示数据默认排除 sim-*；生产应直接执行 +npx tsx scripts/preflight-fulfillment-routing.ts --strict + +# 如确实要把演示数据也纳入检查 +npx tsx scripts/preflight-fulfillment-routing.ts --strict --include-demo +``` + +严格预检必须为 0 失败，重点包括： + +- 有内部 SKU 但无渠道能力的 pending； +- pending 未写入明确原因（说明管道尚未跑完）； +- 已分配自动发货/接单但没有真实载体； +- 已分配订单同时存在多个真实履约载体； +- 多个主渠道； +- 仍启用的 `sku_code` 兼容规则。 + ## 5. 数据清理（确认后执行） ```bash @@ -136,6 +160,6 @@ npx tsx scripts/reclassify-post-cutover.ts --apply - 代码：回退到部署前镜像/commit； - 数据库： - - 116–119 均为新增表/列/数据，不删旧表，可保留； + - 116–121 均为新增表/列/函数/数据，不删旧表，可保留； - 若必须回滚，恢复到部署前备份； -- 迁移 118/119 不改旧字段语义，回滚代码后旧逻辑仍可运行。 +- 迁移 118–121 不改旧字段语义，回滚代码后旧逻辑仍可运行；121 的唯一主渠道索引会阻止旧代码写入并列主渠道。