diff --git a/README.md b/README.md index 525ed77a..a93b6fd9 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ SKIP_CHECKS=1 git push | 文件 | 含义 | | --- | --- | | `001_init.sql` | 空库基线，建立完整业务结构 | -| `002_xxx.sql` ~ `121_xxx.sql` | 后续结构与数据增量（接单平台、寄卖、财务、履约路由迁移） | +| `002_xxx.sql` ~ `121_xxx.sql` | 后续结构与数据增量；当前生产发布基线为 119，120–121 为本次履约路由升级 | 约定： diff --git a/apps/backend/scripts/audit-fulfillment-finance-alignment.ts b/apps/backend/scripts/audit-fulfillment-finance-alignment.ts index e005c2b9..b054b1db 100644 --- a/apps/backend/scripts/audit-fulfillment-finance-alignment.ts +++ b/apps/backend/scripts/audit-fulfillment-finance-alignment.ts @@ -16,11 +16,11 @@ async function main() { ` SELECT count(*)::int AS total, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int AS with_work_order, - 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 with_affiliate_task, - 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 with_manual_task, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int AS with_work_order, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS with_affiliate_task, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS with_manual_task, count(*) FILTER ( - WHERE NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + WHERE NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled') AND NOT EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) )::int AS no_carrier FROM order_items oi @@ -41,9 +41,9 @@ async function main() { SELECT COALESCE(NULLIF(oi.fulfillment_channel, ''), '(pending)') AS channel, count(*)::int AS total, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int AS work_orders, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'))::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 EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int AS work_orders, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS affiliate_tasks, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS manual_tasks, count(*) FILTER ( WHERE oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL )::int AS config_missing, @@ -53,10 +53,11 @@ async function main() { AND NOT EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ) ) OR ( oi.fulfillment_channel = 'work_order' - AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled') ) )::int AS no_carrier FROM order_items oi @@ -137,25 +138,29 @@ async function main() { 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 work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int + (EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ))::int + (EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ))::int 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 + AND (EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int + (EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ))::int + (EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ))::int > 1 ORDER BY oi.id LIMIT 200`, @@ -172,7 +177,7 @@ async function main() { count(*)::int AS total, count(*) FILTER (WHERE oi.fulfillment_status = 'pending')::int AS pending, count(*) FILTER ( - WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled') OR EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) )::int AS with_carrier FROM order_items oi diff --git a/apps/backend/scripts/check-sql-guard.ts b/apps/backend/scripts/check-sql-guard.ts index d8a64158..0df4576f 100644 --- a/apps/backend/scripts/check-sql-guard.ts +++ b/apps/backend/scripts/check-sql-guard.ts @@ -92,6 +92,11 @@ const failures: string[] = [] // 用「已发布基线」而不是 HEAD 做比较，是因为以 HEAD 为基线时，本次提交新增的 // 迁移也会被算成历史迁移，合并提交后必然自相矛盾。 const MIGRATION_BASELINE_REF = resolveMigrationBaselineRef() +// 生产当前已发布到 119；120+ 仍允许在部署前重写。部署 120 后应将此值提升到 120。 +const PUBLISHED_MIGRATION_MAX_VERSION = Math.max( + 0, + Math.trunc(Number(process.env.PUBLISHED_MIGRATION_MAX_VERSION) || 119), +) function resolveMigrationBaselineRef(): string | null { const candidates = [process.env.MIGRATION_BASELINE_REF, 'main', 'origin/main'].filter( @@ -132,7 +137,11 @@ if (!MIGRATION_BASELINE_REF) { const changedHistoricalMigrations = diffOutput .split('\n') .map((line) => line.trim().split(/\s+/)) - .filter(([status, file]) => status && file?.endsWith('.sql') && status !== 'A') + .filter(([status, file]) => { + if (!status || !file?.endsWith('.sql') || status === 'A') return false + const version = Number(path.basename(file).slice(0, 3)) + return Number.isFinite(version) && version <= PUBLISHED_MIGRATION_MAX_VERSION + }) .map(([status, file]) => `${status} ${file}`) if (changedHistoricalMigrations.length > 0) { failures.push( diff --git a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts index fc8ebda0..e3c3b6d7 100644 --- a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts +++ b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts @@ -77,17 +77,6 @@ async function main() { converted.push({ id: rule.id, matchKey: rule.match_key, internalSku, channel: rule.channel }) if (!apply) continue - 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 ('', '', 'internal_sku', $1, $2, $3, $4, TRUE, - '由旧 sku_code 规则迁移', NOW(), NOW()) - ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING - `, - [internalSku, rule.channel, rule.priority, rule.source || 'operator'], - ) await query( `UPDATE internal_sku_channels SET is_fallback = TRUE, diff --git a/apps/backend/scripts/merge-duplicate-products.ts b/apps/backend/scripts/merge-duplicate-products.ts index 891cb5b6..92abfbe1 100644 --- a/apps/backend/scripts/merge-duplicate-products.ts +++ b/apps/backend/scripts/merge-duplicate-products.ts @@ -176,19 +176,6 @@ async function mergeGroup( [group.canonical_name, group.keeper_id], ) - 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 ('', '', 'internal_sku', $1, 'affiliate_dash', 10, 'affiliate', TRUE, - '合并重复商品后按内部SKU生成', NOW(), NOW()) - ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING - `, - [group.canonical_name], - ) - // 只删除已经没有引用的重复商品，仍被引用就归档。 for (const duplicateId of group.duplicate_ids) { const refs = await query<{ total: number }>( diff --git a/apps/backend/scripts/preflight-fulfillment-routing.ts b/apps/backend/scripts/preflight-fulfillment-routing.ts index 963ec15e..1e8d5bfa 100644 --- a/apps/backend/scripts/preflight-fulfillment-routing.ts +++ b/apps/backend/scripts/preflight-fulfillment-routing.ts @@ -75,10 +75,11 @@ async function main() { 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 work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled') AND NOT EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') )`, [includeDemo], ) @@ -91,14 +92,16 @@ async function main() { 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 + AND (EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int + (EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ))::int + (EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ))::int > 1`, [includeDemo], ) @@ -155,9 +158,9 @@ async function main() { }>( `SELECT COALESCE(fulfillment_channel, '(pending)') AS channel, count(*)::int AS total, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int AS work_orders, - count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'))::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 EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int AS work_orders, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS affiliate_tasks, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS manual_tasks FROM order_items oi WHERE oi.fulfillment_status = 'assigned' GROUP BY 1 ORDER BY 2 DESC`, diff --git a/apps/backend/scripts/reclassify-post-cutover.ts b/apps/backend/scripts/reclassify-post-cutover.ts index c79323d5..dcb42ac7 100644 --- a/apps/backend/scripts/reclassify-post-cutover.ts +++ b/apps/backend/scripts/reclassify-post-cutover.ts @@ -60,6 +60,7 @@ async function loadCandidates(limit: number): Promise { OR EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') ) ) AND NOT EXISTS ( @@ -70,7 +71,7 @@ async function loadCandidates(limit: number): Promise { ) AND NOT EXISTS ( SELECT 1 FROM work_orders wo - WHERE wo.order_item_id = oi.id + WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled' ) 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 a683cc17..070d3d1f 100644 --- a/apps/backend/scripts/seed-local-finance-demo.ts +++ b/apps/backend/scripts/seed-local-finance-demo.ts @@ -32,13 +32,19 @@ async function cleanup() { '模拟自动发货商品', '模拟三方卡券', '模拟人工商品')`, ) await query( - `DELETE FROM product_channel_capabilities WHERE product_id IN (SELECT id FROM product_catalog WHERE canonical_sku = 'aff:SIM_AFF_1')`, + `DELETE FROM product_channel_capabilities + WHERE product_id IN ( + SELECT id FROM product_catalog + WHERE canonical_sku IN ('aff:SIM_AFF_1', 'man:模拟人工商品') + )`, ) await query( `DELETE FROM channel_product_mappings WHERE canonical_product_id IN ( SELECT id FROM product_catalog - WHERE canonical_sku IN ('aff:SIM_AFF_1', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B') + WHERE canonical_sku IN ( + 'aff:SIM_AFF_1', 'man:模拟人工商品', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B' + ) )`, ) await query( @@ -48,10 +54,15 @@ async function cleanup() { WHERE canonical_sku IN ('aff:SIM_AFF_1', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B') )`, ) - await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = '模拟人工商品'`) + await query( + `DELETE FROM manual_fulfillment_configs + WHERE sku_key IN ('模拟人工商品', '模拟自动发货商品')`, + ) await query( `DELETE FROM product_catalog - WHERE canonical_sku IN ('aff:SIM_AFF_1', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B')`, + WHERE canonical_sku IN ( + 'aff:SIM_AFF_1', 'man:模拟人工商品', 'tpv:模拟三方卡券', 'tpv:模拟三方卡券B' + )`, ) } @@ -113,19 +124,56 @@ async function main() { VALUES ($1,'affiliate_dash',10,FALSE,TRUE,'demo'), ($1,'manual',20,TRUE,TRUE,'demo')`, [affProductId], ) + const manualProduct = await query<{ id: number }>( + `INSERT INTO product_catalog ( + canonical_sku, canonical_name, category, unit_name, status, + attributes_json, created_at, updated_at + ) VALUES ( + 'man:模拟人工商品', '模拟人工商品', '人工', '件', 'active', + '{}'::jsonb, NOW(), NOW() + ) RETURNING id`, + ) + const manualProductId = Number(manualProduct.rows[0]!.id) + await query( + `INSERT INTO product_channel_capabilities ( + product_id, channel, priority, is_fallback, enabled, source + ) VALUES ($1, 'manual', 10, FALSE, TRUE, 'demo')`, + [manualProductId], + ) + const manualCostItem = await query<{ id: number }>( + `SELECT id FROM cost_items WHERE code = 'sys_manual_service'`, + ) + await query( + `INSERT INTO cost_versions ( + cost_item_id, scope_type, canonical_product_id, amount_fen, rate_bps, + effective_from, status, change_reason, created_at, updated_at + ) VALUES + ($1, 'product', $2, 100, 0, '2026-10-01T00:00:00+08:00', + 'active', '自动发货失败转人工成本', NOW(), NOW()), + ($1, 'product', $3, 100, 0, '2026-10-01T00:00:00+08:00', + 'active', '本地演示人工成本', NOW(), NOW())`, + [manualCostItem.rows[0]!.id, affProductId, manualProductId], + ) await query( `INSERT INTO internal_sku_channels ( internal_sku, channel, priority, is_fallback, enabled, - canonical_product_id, source, created_at, updated_at + canonical_product_id, config_json, source, created_at, updated_at ) VALUES - ('模拟自动发货商品', 'affiliate_dash', 10, FALSE, TRUE, $1, 'demo', NOW(), NOW()), - ('模拟自动发货商品', 'manual', 90, TRUE, TRUE, $1, 'demo', NOW(), NOW())`, + ('模拟自动发货商品', 'affiliate_dash', 10, FALSE, TRUE, $1, + '{"affiliateSku":"SIM_AFF_1"}'::jsonb, 'demo', NOW(), NOW()), + ('模拟自动发货商品', 'manual', 90, TRUE, TRUE, $1, + '{}'::jsonb, 'demo', NOW(), NOW())`, [affProductId], ) // 人工演示配置 await query( - `INSERT INTO manual_fulfillment_configs (sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at) - VALUES ('模拟人工商品','nanhang',100,'fixed',TRUE,'本地演示',NOW(),NOW())`, + `INSERT INTO manual_fulfillment_configs ( + sku_key, canonical_product_id, platform_key, credit_quantity, + cost_mode, enabled, note, created_at, updated_at + ) VALUES + ('模拟自动发货商品', $1, 'nanhang', 100, 'fixed', TRUE, '自动发货兜底', NOW(), NOW()), + ('模拟人工商品', $2, 'nanhang', 100, 'fixed', TRUE, '本地演示', NOW(), NOW())`, + [affProductId, manualProductId], ) await query( `INSERT INTO internal_sku_channels ( diff --git a/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql b/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql index a9d24a01..dcb4e86e 100644 --- a/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql +++ b/apps/backend/src/db/migrations/120_fulfillment_routing_invariants.sql @@ -1,10 +1,22 @@ -- 120_fulfillment_routing_invariants.sql -- --- 履约路由唯一口径与数据清理： --- 1. 订单项内部 SKU 由一个不可变 SQL 函数统一解析； --- 2. 人工配置自动成为 internal_sku_channels 的 manual 能力； --- 3. manual 与其他渠道并存时统一降为兜底，避免同优先级被静默抢占； --- 4. 可确定的旧 sku_code 规则迁移为内部 SKU 渠道并停用旧规则。 +-- 生产迁移 119 之后的履约路由统一迁移。 +-- +-- 目标： +-- 1. 订单项内部 SKU 只有一个 SQL 解析口径； +-- 2. internal_sku_channels 成为唯一渠道配置源； +-- 3. 人工配置、旧 sku_code 规则和渠道适配参数全部回填到统一表； +-- 4. 原有 sku_code 规则只做一次性迁移，迁移后可确定的规则全部停用。 + +-- 稳定 id 让后台渠道配置可以通过统一 API 增删改，而不必继续暴露旧规则表。 +ALTER TABLE internal_sku_channels + ADD COLUMN IF NOT EXISTS id BIGSERIAL, + ADD COLUMN IF NOT EXISTS note TEXT NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS created_by_admin_user_id BIGINT + REFERENCES admin_users(id) ON DELETE SET NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_internal_sku_channels_id + ON internal_sku_channels (id); CREATE OR REPLACE FUNCTION resolve_internal_sku( p_manual_sku_key TEXT, @@ -33,7 +45,7 @@ $$; COMMENT ON FUNCTION resolve_internal_sku(TEXT, TEXT, TEXT) IS '订单项内部 SKU 唯一 SQL 口径：中文 manual_sku_key/sku_name 优先，再按历史生成列顺序回退'; --- 人工配置本身就是一个 manual 渠道声明。存在其他渠道时默认作为 90 优先级兜底。 +-- 人工配置就是 manual 渠道声明。存在其他渠道时先按 90 优先级兜底。 INSERT INTO internal_sku_channels ( internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, config_json, source, created_at, updated_at @@ -42,15 +54,13 @@ SELECT BTRIM(cfg.sku_key), 'manual', CASE WHEN EXISTS ( - SELECT 1 - FROM internal_sku_channels c + 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 + SELECT 1 FROM internal_sku_channels c WHERE BTRIM(c.internal_sku) = BTRIM(cfg.sku_key) AND c.enabled = TRUE AND c.channel <> 'manual' @@ -69,20 +79,241 @@ SELECT 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, + 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 还有自动发货/接单/卡券能力，人工渠道就不能作为并列主渠道。 +-- 旧 sku_code 规则能稳定解析出的内部 SKU，统一补到入口映射表。 +WITH legacy_candidates AS ( + SELECT + rule.provider, + rule.shop_id, + rule.match_key AS sku_code, + BTRIM(product.canonical_name) AS internal_sku, + 10 AS priority + 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 BTRIM(COALESCE(product.canonical_name, '')) <> '' + + UNION ALL + + SELECT + rule.provider, + rule.shop_id, + rule.match_key AS sku_code, + matched.internal_sku, + 20 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 matched.internal_count = 1 + AND BTRIM(COALESCE(matched.internal_sku, '')) <> '' +), +ranked_legacy_candidates AS ( + SELECT *, + ROW_NUMBER() OVER ( + PARTITION BY provider, shop_id, sku_code + ORDER BY priority, internal_sku + ) AS candidate_rank + FROM legacy_candidates +) +INSERT INTO internal_sku_mappings ( + provider, shop_id, match_type, match_value, internal_sku, + priority, enabled, source, note, created_at, updated_at +) +SELECT + provider, shop_id, 'sku_code', sku_code, internal_sku, + priority, TRUE, 'legacy-rule', '由旧 sku_code 渠道规则迁移', NOW(), NOW() +FROM ranked_legacy_candidates +WHERE candidate_rank = 1 +ON CONFLICT (provider, shop_id, match_type, match_value) DO UPDATE SET + internal_sku = EXCLUDED.internal_sku, + priority = LEAST(internal_sku_mappings.priority, EXCLUDED.priority), + enabled = TRUE, + source = EXCLUDED.source, + note = EXCLUDED.note, + updated_at = NOW(); + +-- 可确定的旧 sku_code 渠道转换为 internal_sku_channels。 +WITH resolved_rules AS ( + SELECT + rule.id, + rule.provider, + rule.shop_id, + rule.match_key, + rule.channel, + rule.priority, + COALESCE( + ( + SELECT BTRIM(product.canonical_name) + FROM product_catalog product + WHERE product.canonical_sku = CONCAT('aff:', rule.match_key) + AND product.status = 'active' + ORDER BY product.id + LIMIT 1 + ), + matched.internal_sku + ) AS internal_sku, + matched.canonical_product_id + FROM sku_channel_rules rule + LEFT 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, + MIN(oi.canonical_product_id) AS canonical_product_id + 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 ON matched.internal_count = 1 + WHERE rule.match_key_type = 'sku_code' + AND rule.enabled = TRUE +) +INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, config_json, source, created_at, updated_at +) +SELECT + BTRIM(internal_sku), + channel, + MIN(priority), + FALSE, + TRUE, + MIN(canonical_product_id), + jsonb_build_object( + 'legacySkuCode', MIN(match_key), + 'legacyProvider', MIN(provider), + 'legacyShopId', MIN(shop_id) + ), + 'legacy-rule', + NOW(), + NOW() +FROM resolved_rules +WHERE BTRIM(COALESCE(internal_sku, '')) <> '' +GROUP BY BTRIM(internal_sku), channel +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(); + +-- 已转成内部 SKU 的旧规则停用；真正无法解析的规则保留 enabled，交给严格预检阻断。 +WITH resolved_rule_ids AS ( + SELECT rule.id + FROM sku_channel_rules rule + LEFT 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 ON matched.internal_count = 1 + WHERE rule.match_key_type = 'sku_code' + 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' + ) + OR 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_rule_ids +WHERE rule.id = resolved_rule_ids.id; + +-- 旧 internal_sku 规则已完整复制到统一表，不再参与运行时判定。 +UPDATE sku_channel_rules +SET enabled = FALSE, + note = CONCAT(COALESCE(note, ''), ' [internal_sku_channels 已成为真源]'), + updated_at = NOW() +WHERE match_key_type = 'internal_sku' AND enabled = TRUE; + +-- 回填适配参数，避免统一表只有“渠道名”而实际还要猜 affiliate SKU / 工单模板。 +WITH affiliate_config AS ( + SELECT + BTRIM(product.canonical_name) AS internal_sku, + MIN(SUBSTRING(product.canonical_sku FROM 5)) AS affiliate_sku, + MIN(product.id) AS canonical_product_id + FROM product_catalog product + WHERE product.status = 'active' AND product.canonical_sku LIKE 'aff:%' + GROUP BY BTRIM(product.canonical_name) +) +UPDATE internal_sku_channels channel +SET config_json = channel.config_json || jsonb_build_object( + 'affiliateSku', affiliate_config.affiliate_sku + ), + canonical_product_id = COALESCE(channel.canonical_product_id, affiliate_config.canonical_product_id), + updated_at = NOW() +FROM affiliate_config +WHERE channel.channel = 'affiliate_dash' + AND BTRIM(channel.internal_sku) = affiliate_config.internal_sku; + +WITH work_order_config AS ( + SELECT + BTRIM(product.canonical_name) AS internal_sku, + MIN(SUBSTRING(product.canonical_sku FROM 5)) AS work_rule_key, + MIN(product.id) AS canonical_product_id + FROM product_catalog product + WHERE product.status = 'active' AND product.canonical_sku LIKE 'tpl:%' + GROUP BY BTRIM(product.canonical_name) +) +UPDATE internal_sku_channels channel +SET config_json = channel.config_json || jsonb_build_object( + 'workRuleKey', work_order_config.work_rule_key + ), + canonical_product_id = COALESCE(channel.canonical_product_id, work_order_config.canonical_product_id), + updated_at = NOW() +FROM work_order_config +WHERE channel.channel = 'work_order' + AND BTRIM(channel.internal_sku) = work_order_config.internal_sku; + +WITH voucher_config AS ( + SELECT + BTRIM(product.canonical_name) AS internal_sku, + MIN(SUBSTRING(product.canonical_sku FROM 5)) AS voucher_sku, + MIN(product.id) AS canonical_product_id + FROM product_catalog product + WHERE product.status = 'active' AND product.canonical_sku LIKE 'tpv:%' + GROUP BY BTRIM(product.canonical_name) +) +UPDATE internal_sku_channels channel +SET config_json = channel.config_json || jsonb_build_object( + 'voucherSku', voucher_config.voucher_sku + ), + canonical_product_id = COALESCE(channel.canonical_product_id, voucher_config.canonical_product_id), + updated_at = NOW() +FROM voucher_config +WHERE channel.channel = 'third_party_voucher' + AND BTRIM(channel.internal_sku) = voucher_config.internal_sku; + +-- 只要同一内部 SKU 还有非人工渠道，人工统一降为显式兜底。 UPDATE internal_sku_channels manual_channel SET priority = GREATEST(manual_channel.priority, 90), is_fallback = TRUE, @@ -98,128 +329,17 @@ WHERE manual_channel.channel = 'manual' 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); +ALTER TABLE internal_sku_channels + DROP CONSTRAINT IF EXISTS chk_internal_sku_channels_config_object; +ALTER TABLE internal_sku_channels + ADD CONSTRAINT chk_internal_sku_channels_config_object + CHECK (jsonb_typeof(config_json) = 'object'); + 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 index 4678b033..cb17f55f 100644 --- a/apps/backend/src/db/migrations/121_internal_sku_single_primary.sql +++ b/apps/backend/src/db/migrations/121_internal_sku_single_primary.sql @@ -23,3 +23,9 @@ 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; + +ALTER TABLE internal_sku_channels + DROP CONSTRAINT IF EXISTS chk_internal_sku_channels_nonempty_trimmed; +ALTER TABLE internal_sku_channels + ADD CONSTRAINT chk_internal_sku_channels_nonempty_trimmed + CHECK (internal_sku <> '' AND internal_sku = BTRIM(internal_sku)); diff --git a/apps/backend/src/routes/admin/cost-center-catalog.ts b/apps/backend/src/routes/admin/cost-center-catalog.ts index e79e4671..b49e3631 100644 --- a/apps/backend/src/routes/admin/cost-center-catalog.ts +++ b/apps/backend/src/routes/admin/cost-center-catalog.ts @@ -374,8 +374,8 @@ router.post( errorMessage: '补齐卡券渠道规则失败', scope: '[admin/cost-center/voucher-products/ensure-rules]', audit: (_req, data) => ({ - action: 'voucher_channel_rules_ensured', - targetType: 'sku_channel_rules', + action: 'voucher_internal_channels_ensured', + targetType: 'internal_sku_channels', targetId: '', data: data as Record, }), diff --git a/apps/backend/src/routes/admin/fulfillment-channel.ts b/apps/backend/src/routes/admin/fulfillment-channel.ts index 58125a9a..778803e3 100644 --- a/apps/backend/src/routes/admin/fulfillment-channel.ts +++ b/apps/backend/src/routes/admin/fulfillment-channel.ts @@ -14,6 +14,7 @@ import { deleteChannelRule, listChannelRules, listPendingChannelGroups, + switchChannelForOrderItems, unlockChannel, upsertChannelRule, } from '../../services/fulfillment/channel-assign-service.js' @@ -143,6 +144,35 @@ router.post( ), ) +router.post( + '/fulfillment/switch', + requireAdminPermission('finance.config'), + createJsonHandler( + (req) => { + const body = bodyOf(req) + return switchChannelForOrderItems({ + orderItemIds: Array.isArray(body.orderItemIds) + ? body.orderItemIds.map((value: unknown) => Number(value)) + : [Number(body.orderItemId)], + channel: String(body.channel || ''), + adminUserId: req.adminSession?.userId ?? null, + note: String(body.note || ''), + }) + }, + { + successMessage: '履约渠道已切换', + errorMessage: '切换履约渠道失败', + scope: '[admin/fulfillment/switch]', + audit: (req, data) => ({ + action: 'fulfillment.switch_channel', + targetType: 'order_item', + targetId: String(bodyOf(req).orderItemId || ''), + data: { payload: bodyOf(req), result: data as Record }, + }), + }, + ), +) + router.get( '/fulfillment/rules', requireAdminPermission('finance.view'), @@ -170,9 +200,7 @@ router.post( const body = bodyOf(req) return upsertChannelRule({ ...(body.id ? { id: Number(body.id) } : {}), - provider: String(body.provider || ''), - shopId: String(body.shopId || ''), - matchKeyType: String(body.matchKeyType || 'sku_code'), + matchKeyType: String(body.matchKeyType || 'internal_sku'), matchKey: String(body.matchKey || ''), channel: String(body.channel || ''), priority: Number(body.priority), @@ -188,7 +216,7 @@ router.post( scope: '[admin/fulfillment/rules]', audit: (req, data) => ({ action: 'fulfillment.upsert_rule', - targetType: 'sku_channel_rule', + targetType: 'internal_sku_channel', targetId: String((data as { id?: number })?.id || ''), data: { payload: bodyOf(req) }, }), diff --git a/apps/backend/src/services/cost-center/affiliate-product-service.ts b/apps/backend/src/services/cost-center/affiliate-product-service.ts index 3413b0a7..8ebe7d6f 100644 --- a/apps/backend/src/services/cost-center/affiliate-product-service.ts +++ b/apps/backend/src/services/cost-center/affiliate-product-service.ts @@ -178,6 +178,7 @@ export async function backfillAdminAffiliateDashSkus() { ) AS sku_code FROM fulfillment_tasks ft WHERE ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') AND ft.created_at >= NOW() - INTERVAL '30 days' AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> '' ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC @@ -202,6 +203,7 @@ export async function backfillAdminAffiliateDashSkus() { SELECT COUNT(DISTINCT ft.order_item_id)::int FROM fulfillment_tasks ft WHERE ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') AND ft.created_at >= NOW() - INTERVAL '30 days' ) AS affiliate_item_count, (SELECT COUNT(*)::int FROM target) AS matched_count, @@ -245,6 +247,7 @@ export async function backfillAdminAffiliateDashSkus() { SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') AND ft.created_at >= NOW() - INTERVAL '30 days' ) `, 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 a8807f18..f7a62b4f 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 @@ -408,7 +408,6 @@ export async function deleteManualFulfillmentConfig(skuKey: string) { `DELETE FROM manual_fulfillment_configs WHERE sku_key = $1 RETURNING id`, [key], ) - await query(`DELETE FROM sku_channel_rules WHERE source = 'manual' AND match_key = $1`, [key]) await upsertInternalSkuChannel({ internalSku: key, channel: 'manual', @@ -722,9 +721,7 @@ function requiredRow(row: T | undefined, message: string): T { } /** - * 人工配置 → 渠道规则（内部 SKU 精确匹配）。 - * - * 幂等：按唯一键 (provider, shop, match_key_type, match_key, channel) upsert。 + * 人工配置 → internal_sku_channels（内部 SKU 精确匹配）。 */ async function syncManualChannelRule(config: { id: number @@ -757,29 +754,4 @@ async function syncManualChannelRule(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', $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), - 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 05523ece..693474e1 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -198,23 +198,10 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise undefined) - // 同一内部中文 SKU 跨平台码统一分流到卡券渠道。 - 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 ('', '', 'internal_sku', $1, 'third_party_voucher', 10, 'voucher', TRUE, - '卡券商品自动建商品后生成', NOW(), NOW()) - ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING - `, - [internalSku], - ) - return productId } diff --git a/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts b/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts index 6e109615..ad72624b 100644 --- a/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts +++ b/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts @@ -31,6 +31,7 @@ export async function runAffiliateFallback( internal_sku: string sku_name: string config_id: number | null + config_product_id: number | null basis_date: string }>( ` @@ -42,6 +43,7 @@ export async function runAffiliateFallback( resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku, oi.sku_name, cfg.id AS config_id, + cfg.canonical_product_id AS config_product_id, (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date FROM order_items oi JOIN orders o ON o.id = oi.order_id @@ -95,6 +97,7 @@ export async function runAffiliateFallback( channel_switch_reason = 'affiliate_failed', channel_switched_at = NOW(), fulfillment_config_id = COALESCE(fulfillment_config_id, $2), + canonical_product_id = $3, updated_at = NOW() WHERE id = $1 AND fulfillment_status = 'assigned' @@ -102,7 +105,7 @@ export async function runAffiliateFallback( AND fulfillment_locked = FALSE RETURNING id `, - [row.id, row.config_id], + [row.id, row.config_id, row.config_product_id], ) if ((result.rowCount || 0) === 0) return false diff --git a/apps/backend/src/services/fulfillment/carrier-creation-service.ts b/apps/backend/src/services/fulfillment/carrier-creation-service.ts index a3977393..179ccdb7 100644 --- a/apps/backend/src/services/fulfillment/carrier-creation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-creation-service.ts @@ -2,9 +2,10 @@ import { query } from '../../db/client.js' import type { OrderItemRow, OrderRow } from '../../types/repository/rows.js' import { syncDeliveryTasksForOrder } from '../order/delivery-task-service.js' import { syncWorkerOrdersForSourceOrder } from '../worker-platform/sync-worker-orders-service.js' -import type { FulfillmentChannel } from './channel-types.js' +import { INACTIVE_CARRIER_TASK_STATUSES, type FulfillmentChannel } from './channel-types.js' import { applyIntakeChannelDecision } from './intake-channel-decision-service.js' import type { ResolvedFulfillmentItem } from './product-resolution-service.js' +import { resolveInternalSkuKey } from './internal-sku-resolver.js' function parseObject(value: unknown): Record { if (value && typeof value === 'object' && !Array.isArray(value)) { @@ -29,7 +30,9 @@ export type EnsureCarrierResult = { 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`, + `SELECT EXISTS ( + SELECT 1 FROM work_orders WHERE order_item_id = $1 AND status <> 'cancelled' + ) AS exists`, [orderItemId], ) return result.rows[0]?.exists === true @@ -40,8 +43,9 @@ async function hasAffiliateTask(orderItemId: number): Promise { `SELECT EXISTS ( SELECT 1 FROM fulfillment_tasks WHERE order_item_id = $1 AND executor_key = 'affiliate_dash' + AND task_status <> ALL($2::text[]) ) AS exists`, - [orderItemId], + [orderItemId, INACTIVE_CARRIER_TASK_STATUSES], ) return result.rows[0]?.exists === true } @@ -110,6 +114,23 @@ export async function ensureCarrierForOrderItem(input: { if (await hasWorkOrder(Number(item.id))) { return { ready: true, created: false, reason: 'work_order_exists', message: '' } } + const internalSku = resolveInternalSkuKey({ + manualSkuKey: item.manual_sku_key, + skuCode: item.sku_code, + skuName: item.sku_name, + spec: parseObject(item.spec_json), + snapshot: parseObject(item.item_snapshot_json), + }) + const channelConfig = await query<{ config_json: Record }>( + `SELECT config_json + FROM internal_sku_channels + WHERE btrim(internal_sku) = $1 AND channel = 'work_order' AND enabled = TRUE + LIMIT 1`, + [internalSku], + ) + const preferredWorkRuleKey = String( + channelConfig.rows[0]?.config_json?.workRuleKey || '', + ).trim() const snapshot = parseObject(item.item_snapshot_json) const classification = parseObject(snapshot.classification) const route = parseObject(snapshot.fulfillmentRoute) @@ -130,6 +151,7 @@ export async function ensureCarrierForOrderItem(input: { source: 'carrier_creation', autoOnly: true, skipChannelAssign: true, + ...(preferredWorkRuleKey ? { preferredWorkRuleKey } : {}), }) const ready = (await hasWorkOrder(Number(item.id))) || result.created.length > 0 if (ready) { diff --git a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts index 74f4f186..8b10d04b 100644 --- a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts @@ -2,7 +2,11 @@ import { query } from '../../db/client.js' import { getOrderById } from '../../repositories/order-repo.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 { + FULFILLMENT_CHANNELS, + INACTIVE_CARRIER_TASK_STATUSES, + type FulfillmentChannel, +} from './channel-types.js' import { ensureCarrierForOrderItem } from './carrier-creation-service.js' import { planInternalSkuChannels, @@ -21,6 +25,7 @@ function normalizeChannels(value: unknown): InternalSkuChannelRow[] { return value .filter((row): row is Record => Boolean(row) && typeof row === 'object') .map((row) => ({ + id: Number(row.id || 0), internal_sku: String(row.internal_sku || '').trim(), channel: String(row.channel || '') as FulfillmentChannel, priority: Number(row.priority || 100), @@ -33,6 +38,9 @@ function normalizeChannels(value: unknown): InternalSkuChannelRow[] { ? (row.config_json as Record) : {}, source: String(row.source || ''), + note: String(row.note || ''), + created_by_admin_user_id: + row.created_by_admin_user_id == null ? null : Number(row.created_by_admin_user_id), })) .filter((row) => FULFILLMENT_CHANNELS.includes(row.channel)) } @@ -49,10 +57,14 @@ async function loadCandidateItems(input: { 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 work_orders wo + WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled' + ) AND NOT EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')}) )` : `oi.fulfillment_status = 'pending' AND oi.fulfillment_locked = FALSE` @@ -70,6 +82,7 @@ async function loadCandidateItems(input: { SELECT COALESCE( jsonb_agg( jsonb_build_object( + 'id', channel.id, 'internal_sku', channel.internal_sku, 'channel', channel.channel, 'priority', channel.priority, @@ -78,6 +91,8 @@ async function loadCandidateItems(input: { 'canonical_product_id', channel.canonical_product_id, 'config_json', channel.config_json, 'source', channel.source + ,'note', channel.note + ,'created_by_admin_user_id', channel.created_by_admin_user_id ) ORDER BY channel.priority, channel.channel ), @@ -138,6 +153,8 @@ async function selectSafeFallback(input: { fulfillment_reason = $3, channel_switch_reason = $3, channel_switched_at = NOW(), + canonical_product_id = NULL, + fulfillment_config_id = NULL, fulfillment_pending_reason = NULL, fulfillment_pending_detail = NULL, updated_at = NOW() @@ -328,10 +345,14 @@ export async function reconcilePendingCarriers( FROM order_items oi 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 work_orders wo + WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled' + ) AND NOT EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')}) ) `, ) 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 56f44eb7..8e9420ab 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts @@ -29,6 +29,7 @@ function buildItem(patch: Partial = {}): ChannelCandidateI function buildChannel(patch: Partial = {}): InternalSkuChannelRow { return { + id: 1, internal_sku: '套装-测试', channel: 'manual', priority: 10, @@ -37,6 +38,8 @@ function buildChannel(patch: Partial = {}): InternalSkuCh canonical_product_id: null, config_json: {}, source: 'test', + note: '', + created_by_admin_user_id: null, ...patch, } } diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index d4c05b34..b13ec83f 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -1,5 +1,8 @@ -import { query } from '../../db/client.js' +import { query, withTransaction } from '../../db/client.js' +import { closeClaimTokensByTaskId } from '../../repositories/claim-token-repo.js' import { getOrderById } from '../../repositories/order-repo.js' +import { createTaskEvent } from '../../repositories/task-event-repo.js' +import { createWorkOrderEvent } from '../../repositories/worker-platform/work-order-event-repo.js' import type { OrderItemRow } from '../../types/repository/rows.js' import { createHttpError } from '../../utils/http.js' import { nowIso } from '../../utils/time.js' @@ -10,10 +13,10 @@ import { normalizeProductsForItems } from '../cost-center/product-normalize-serv import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js' import { FULFILLMENT_CHANNELS, + INACTIVE_CARRIER_TASK_STATUSES, MATCH_KEY_TYPES, RULE_SOURCES, isFulfillmentChannel, - isMatchKeyType, isRuleSource, type FulfillmentChannel, type FulfillmentSource, @@ -21,13 +24,14 @@ import { type RuleSource, } from './channel-types.js' import { - disableInternalSkuChannel, + deleteInternalSkuChannelById, planInternalSkuChannels, setInternalSkuPrimaryChannel, upsertInternalSkuChannel, type InternalSkuChannelRow, } from './internal-sku-channel-service.js' import { ensureCarrierForOrderItem } from './carrier-creation-service.js' +import { TASK_STATUS } from '../../domain/task-status.js' /** * 管道 A：渠道判定。 @@ -45,6 +49,7 @@ import { ensureCarrierForOrderItem } from './carrier-creation-service.js' /** 数据库函数 `resolve_internal_sku` 是订单项内部 SKU 的唯一 SQL 口径。 */ const INTERNAL_SKU_SQL = `resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)` +const ACTIVE_CARRIER_TASK_SQL = `task.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')})` export type ChannelRule = { id: number @@ -304,15 +309,17 @@ export async function assignPendingOrderItemChannels( (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date, EXISTS ( - SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id + SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled' ) AS has_work_order, EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + SELECT 1 FROM fulfillment_tasks task + WHERE task.order_item_id = oi.id AND task.executor_key = 'affiliate_dash' + AND ${ACTIVE_CARRIER_TASK_SQL} ) AS has_affiliate_task, EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + SELECT 1 FROM fulfillment_tasks task + WHERE task.order_item_id = oi.id AND task.executor_key = 'manual_dispatch' + AND ${ACTIVE_CARRIER_TASK_SQL} ) AS has_manual_task FROM order_items oi JOIN orders o ON o.id = oi.order_id @@ -577,6 +584,205 @@ async function assignOrderItemsAfterCarrierReady(input: { return { updated: Number(final.rows[0]?.count || assignedIds.length), carrierFailed } } +const WORK_ORDER_SWITCH_BLOCKED = new Set([ + 'open', + 'in_progress', + 'pending_acceptance', + 'problem', + 'accepted', +]) +const TASK_SWITCH_BLOCKED = new Set([ + TASK_STATUS.REDEEMING, + TASK_STATUS.DISPATCHED_PENDING_RETURN, + TASK_STATUS.COMPLETED, + TASK_STATUS.REDEEMED, +]) + +/** + * 显式切换渠道：先取消尚未实际履约的旧载体，再创建新载体。 + * 已经成功或进入实际交付阶段的订单不允许自动切换。 + */ +export async function switchChannelForOrderItems(input: { + orderItemIds: number[] + channel: string + adminUserId: number | null + note?: string +}): Promise<{ + switched: number + failed: number + errors: Array<{ orderItemId: number; reason: string }> +}> { + if (!isFulfillmentChannel(input.channel)) { + throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) + } + const ids = [ + ...new Set(input.orderItemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0)), + ] + if (ids.length === 0) { + throw createHttpError('请选择订单项', { statusCode: 400, errorCode: 'order_item_required' }) + } + + 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 readyIds: number[] = [] + const errors: Array<{ orderItemId: number; reason: string }> = [] + + for (const item of items) { + const order = orders.get(Number(item.order_id)) + if (!order) { + errors.push({ orderItemId: Number(item.id), reason: 'order_not_found' }) + continue + } + if (item.fulfillment_status === 'archived') { + errors.push({ orderItemId: Number(item.id), reason: 'archived' }) + continue + } + if (item.fulfillment_channel === input.channel) { + readyIds.push(Number(item.id)) + continue + } + + try { + const workOrders = await query<{ id: number; status: string }>( + `SELECT id, status FROM work_orders + WHERE order_item_id = $1 AND status <> 'cancelled'`, + [item.id], + ) + const blockedWorkOrder = workOrders.rows.find((row) => + WORK_ORDER_SWITCH_BLOCKED.has(String(row.status)), + ) + if (blockedWorkOrder) { + throw createHttpError('旧接单工单已进入履约阶段，不能自动切换渠道', { + statusCode: 409, + errorCode: 'channel_switch_work_order_active', + }) + } + const tasks = await query<{ id: number; task_status: string }>( + `SELECT id, task_status FROM fulfillment_tasks WHERE order_item_id = $1`, + [item.id], + ) + const blockedTask = tasks.rows.find((row) => TASK_SWITCH_BLOCKED.has(String(row.task_status))) + if (blockedTask) { + throw createHttpError('旧履约任务已进入实际交付阶段，不能自动切换渠道', { + statusCode: 409, + errorCode: 'channel_switch_task_active', + }) + } + + const now = nowIso() + await withTransaction(async (client) => { + const locked = await client.query( + `SELECT * FROM order_items WHERE id = $1 FOR UPDATE`, + [item.id], + ) + if (!locked.rows[0] || locked.rows[0].fulfillment_status === 'archived') { + throw createHttpError('订单项状态已变化', { + statusCode: 409, + errorCode: 'channel_switch_order_item_changed', + }) + } + for (const workOrder of workOrders.rows) { + await client.query( + `UPDATE work_orders + SET status = 'cancelled', assigned_worker_id = NULL, assigned_at = NULL, + deadline_at = NULL, published_at = NULL, updated_at = $2 + WHERE id = $1 AND status <> 'cancelled'`, + [workOrder.id, now], + ) + } + for (const task of tasks.rows) { + await client.query( + `UPDATE fulfillment_tasks + SET task_status = $2, delivery_status = 'closed', result_code = 'closed_by_channel_switch', + result_message = $3, user_action_status = 'closed', claim_expires_at = $4, + updated_at = $4 + WHERE id = $1`, + [task.id, TASK_STATUS.CLOSED, '履约渠道已切换，原任务关闭', now], + ) + } + await client.query( + `UPDATE cost_snapshots + SET superseded_at = $2 + WHERE order_item_id = $1 AND superseded_at IS NULL AND reversed_at IS NULL`, + [item.id, now], + ) + await client.query( + `UPDATE order_items + SET original_channel = COALESCE(original_channel, fulfillment_channel), + fulfillment_channel = NULL, + fulfillment_status = 'pending', + fulfillment_source = 'operator_assigned', + fulfillment_locked = FALSE, + fulfillment_reason = $2, + channel_switch_reason = 'operator', + channel_switched_at = $3, + canonical_product_id = NULL, + fulfillment_config_id = NULL, + fulfillment_pending_reason = NULL, + fulfillment_pending_detail = NULL, + updated_at = $3 + WHERE id = $1`, + [item.id, String(input.note || '运营切换履约渠道').slice(0, 200), now], + ) + }) + + for (const workOrder of workOrders.rows) { + await createWorkOrderEvent({ + workOrderId: Number(workOrder.id), + actorType: 'admin', + actorId: String(input.adminUserId || ''), + eventType: 'cancelled_by_channel_switch', + fromStatus: workOrder.status, + toStatus: 'cancelled', + payloadJson: JSON.stringify({ channel: input.channel, note: input.note || '' }), + now, + }) + } + for (const task of tasks.rows) { + await closeClaimTokensByTaskId(Number(task.id), now) + await createTaskEvent( + Number(task.id), + 'task_closed_by_channel_switch', + { channel: input.channel, note: input.note || '', previousStatus: task.task_status }, + now, + ) + } + readyIds.push(Number(item.id)) + } catch (error) { + errors.push({ + orderItemId: Number(item.id), + reason: error instanceof Error ? error.message : String(error), + }) + } + } + + const assigned = await assignOrderItemsAfterCarrierReady({ + orderItemIds: readyIds, + 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[])`, + [readyIds, input.adminUserId], + ) + return { + switched: assigned.updated, + failed: errors.length + assigned.carrierFailed, + errors, + } +} + /** 运营指定渠道：先保证真实载体，成功后才写入并锁定。 */ export async function assignChannelToOrderItems(input: { orderItemIds: number[] @@ -700,8 +906,8 @@ export async function assignChannelByInternalSku(input: { internalSku: string channel: string adminUserId: number | null - note?: string - createRule?: boolean + note?: string | undefined + createRule?: boolean | undefined }): Promise<{ updated: number; ruleCreated: boolean; carrierFailed: number }> { if (!isFulfillmentChannel(input.channel)) { throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) @@ -713,6 +919,12 @@ export async function assignChannelByInternalSku(input: { errorCode: 'internal_sku_required', }) } + if (input.createRule === false) { + throw createHttpError('内部 SKU 渠道配置必须持久化，不能只改当前订单', { + statusCode: 400, + errorCode: 'persistent_channel_rule_required', + }) + } const idsResult = await query<{ id: number }>( ` @@ -743,24 +955,11 @@ export async function assignChannelByInternalSku(input: { canonicalProductId: product.rows[0]?.id == null ? null : Number(product.rows[0].id), source: 'operator', configJson: { note: String(input.note || '').slice(0, 200) }, + note: input.note, + adminUserId: input.adminUserId, }) - let ruleCreated = false - if (input.createRule !== false) { - const rule = await upsertChannelRule({ - provider: input.provider, - shopId: input.shopId, - matchKeyType: 'internal_sku', - matchKey: internalSku, - channel: input.channel, - priority: 10, - source: 'operator', - enabled: true, - note: `运营在待分类队列指定：${String(input.note || '').slice(0, 120)}`, - adminUserId: input.adminUserId, - }) - ruleCreated = rule.id > 0 - } + const ruleCreated = true const ids = idsResult.rows.map((row) => Number(row.id)) const applied = await assignOrderItemsAfterCarrierReady({ @@ -787,8 +986,8 @@ export async function assignChannelBySku(input: { skuCode: string channel: string adminUserId: number | null - note?: string - createRule?: boolean + note?: string | undefined + createRule?: boolean | undefined }): Promise<{ updated: number; ruleCreated: boolean; carrierFailed: number }> { if (!isFulfillmentChannel(input.channel)) { throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' }) @@ -813,48 +1012,24 @@ export async function assignChannelBySku(input: { 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) }, + if (internalSkus.length === 0) { + return { updated: 0, ruleCreated: false, carrierFailed: 0 } + } + if (internalSkus.length > 1) { + throw createHttpError('同一平台 SKU 对应多个内部 SKU，请按内部 SKU 指定渠道', { + statusCode: 409, + errorCode: 'sku_code_ambiguous_internal_sku', }) } - - // 默认同时落一条渠道规则：否则同 SKU 的下一单还是会进待分类。 - let ruleCreated = false - if (input.createRule !== false) { - const rule = await upsertChannelRule({ - provider: input.provider, - shopId: input.shopId, - matchKeyType: 'sku_code', - matchKey: input.skuCode, - channel: input.channel, - priority: 10, - source: 'operator', - enabled: true, - note: `运营在待分类队列指定：${String(input.note || '').slice(0, 120)}`, - adminUserId: input.adminUserId, - }) - ruleCreated = rule.id > 0 - } - - const ids = idsResult.rows.map((row) => Number(row.id)) - const applied = await assignOrderItemsAfterCarrierReady({ - orderItemIds: ids, + return assignChannelByInternalSku({ + provider: input.provider, + shopId: input.shopId, + internalSku: internalSkus[0]!, channel: input.channel, - source: 'operator_assigned', - lock: true, + adminUserId: input.adminUserId, note: input.note, + createRule: input.createRule, }) - 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( @@ -877,19 +1052,26 @@ export async function listChannelRules( const rows = await query( ` SELECT id, provider, shop_id, match_key_type, match_key, channel, priority, enabled - FROM sku_channel_rules + FROM ( + SELECT id, ''::text AS provider, ''::text AS shop_id, + 'internal_sku'::text AS match_key_type, internal_sku AS match_key, + channel, priority, enabled + FROM internal_sku_channels + ) internal_rules ${where} ORDER BY priority, channel, match_key `, params, ) - return rows.rows + return rows.rows.map((row) => ({ + ...row, + id: Number(row.id), + priority: Number(row.priority), + })) } export async function upsertChannelRule(input: { id?: number - provider?: string - shopId?: string matchKeyType: string matchKey: string channel: string @@ -899,8 +1081,8 @@ export async function upsertChannelRule(input: { note?: string adminUserId: number | null }): Promise<{ id: number }> { - if (!isMatchKeyType(input.matchKeyType)) { - throw createHttpError('匹配键类型无效', { + if (input.matchKeyType !== 'internal_sku') { + throw createHttpError('渠道配置只接受内部 SKU；旧 sku_code 规则仅用于一次性迁移', { statusCode: 400, errorCode: 'invalid_match_key_type', }) @@ -914,48 +1096,19 @@ export async function upsertChannelRule(input: { throw createHttpError('匹配键不能为空', { statusCode: 400, errorCode: 'match_key_required' }) } - const result = await query<{ id: number }>( - ` - INSERT INTO sku_channel_rules ( - provider, shop_id, match_key_type, match_key, channel, priority, - source, enabled, note, created_by_admin_user_id, created_at, updated_at - ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW(), NOW()) - ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) - DO UPDATE SET - priority = EXCLUDED.priority, - enabled = EXCLUDED.enabled, - note = EXCLUDED.note, - updated_at = NOW() - RETURNING id - `, - [ - String(input.provider || '').trim(), - String(input.shopId || '').trim(), - input.matchKeyType, - matchKey, - input.channel, - Math.trunc(Number(input.priority) || 100), - source, - input.enabled !== false, - String(input.note || '').slice(0, 200), - input.adminUserId, - ], - ) - - 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 } + const priority = Math.trunc(Number(input.priority) || 100) + const id = await upsertInternalSkuChannel({ + internalSku: matchKey, + channel: input.channel, + priority, + isFallback: priority >= 90, + enabled: input.enabled !== false, + source: source === 'operator' ? 'operator' : `rule:${source}`, + note: String(input.note || '').slice(0, 300), + adminUserId: input.adminUserId, + replacePriority: true, + }) + return { id } } export async function deleteChannelRule(ruleId: number): Promise<{ deleted: boolean }> { @@ -963,25 +1116,7 @@ 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<{ - 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) } + return { deleted: await deleteInternalSkuChannelById(id) } } async function loadEnabledRules(): Promise { @@ -992,7 +1127,11 @@ async function loadEnabledRules(): Promise { WHERE enabled = TRUE `, ) - return rows.rows + return rows.rows.map((row) => ({ + ...row, + id: Number(row.id), + priority: Number(row.priority), + })) } async function loadInternalSkuChannelsForKeys( @@ -1004,8 +1143,8 @@ async function loadInternalSkuChannelsForKeys( const rows = await query( ` - SELECT internal_sku, channel, priority, is_fallback, enabled, - canonical_product_id, config_json, source + SELECT id, internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, config_json, source, note, created_by_admin_user_id FROM internal_sku_channels WHERE enabled = TRUE AND btrim(internal_sku) = ANY($1::text[]) ORDER BY internal_sku, priority, channel @@ -1029,19 +1168,23 @@ export const CHANNEL_VALUES = FULFILLMENT_CHANNELS export async function ensureVoucherRuleForSku(skuCode: string): Promise { const key = String(skuCode || '').trim() if (!key) return false - const result = await query<{ id: number }>( - ` - 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', $1, 'third_party_voucher', 70, 'voucher', TRUE, '卡券商品自动生成', NOW(), NOW()) - ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING - RETURNING id - `, - [key], + const product = await query<{ id: number }>( + `SELECT id FROM product_catalog + WHERE canonical_sku = $1 AND status = 'active' + LIMIT 1`, + [`tpv:${key}`], ) - return (result.rowCount || 0) > 0 + await upsertInternalSkuChannel({ + internalSku: key, + channel: 'third_party_voucher', + priority: 10, + isFallback: false, + enabled: true, + canonicalProductId: Number(product.rows[0]?.id || 0) || null, + configJson: { voucherSku: key }, + source: 'voucher', + }) + return true } /** @@ -1054,28 +1197,38 @@ export async function ensureVoucherChannelRules(): Promise<{ created: number productCount: number }> { + await query( + `UPDATE internal_sku_channels channel + SET is_fallback = TRUE, + priority = GREATEST(priority, 90), + updated_at = NOW() + FROM product_catalog product + WHERE product.status = 'active' + AND product.canonical_sku LIKE 'tpv:%' + AND BTRIM(channel.internal_sku) = BTRIM(product.canonical_name) + AND channel.channel <> 'third_party_voucher' + AND channel.enabled = TRUE + AND channel.is_fallback = FALSE`, + ) const created = await query<{ id: number }>( ` - INSERT INTO sku_channel_rules ( - provider, shop_id, match_key_type, match_key, channel, priority, - source, enabled, note, created_at, updated_at + INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, config_json, source, created_at, updated_at ) SELECT - '', '', 'sku_code', substring(p.canonical_sku from 5), 'third_party_voucher', 70, - 'voucher', TRUE, '卡券商品自动生成', NOW(), NOW() + BTRIM(p.canonical_name), 'third_party_voucher', 10, FALSE, TRUE, + p.id, jsonb_build_object('voucherSku', substring(p.canonical_sku from 5)), + 'voucher', NOW(), NOW() FROM product_catalog p WHERE p.status = 'active' AND p.canonical_sku LIKE 'tpv:%' - AND btrim(substring(p.canonical_sku from 5)) <> '' - AND NOT EXISTS ( - SELECT 1 FROM sku_channel_rules r - WHERE r.provider = '' - AND r.shop_id = '' - AND r.match_key_type = 'sku_code' - AND r.match_key = substring(p.canonical_sku from 5) - AND r.channel = 'third_party_voucher' - ) - ON CONFLICT DO NOTHING + AND BTRIM(COALESCE(p.canonical_name, '')) <> '' + ON CONFLICT (internal_sku, channel) DO UPDATE SET + 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() RETURNING id `, ) diff --git a/apps/backend/src/services/fulfillment/channel-types.ts b/apps/backend/src/services/fulfillment/channel-types.ts index 26153db2..205dd6fd 100644 --- a/apps/backend/src/services/fulfillment/channel-types.ts +++ b/apps/backend/src/services/fulfillment/channel-types.ts @@ -20,6 +20,16 @@ export const FULFILLMENT_STATUSES = ['pending', 'assigned', 'archived'] as const export type FulfillmentStatus = (typeof FULFILLMENT_STATUSES)[number] +/** 这些任务状态只保留审计价值，不再代表当前履约载体。 */ +export const INACTIVE_CARRIER_TASK_STATUSES = [ + 'closed', + 'expired', + 'failed', + 'manual_review', + 'retry_pending', + 'cancelled', +] as const + export const FULFILLMENT_SOURCES = [ 'auto_rule', 'auto_mapping', diff --git a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts index 3f33d956..66db5f89 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts @@ -13,19 +13,27 @@ import test from 'node:test' const TEST_DATABASE_URL = String(process.env.FULFILLMENT_ROUTING_TEST_DATABASE_URL || '').trim() test( - '运营指定渠道：先保证载体，再写 assigned；缺载体保持 pending', + '运营指定渠道：先保证载体；支持安全切换；缺载体保持 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 { + assignChannelByInternalSku, + deleteChannelRule, + ensureVoucherRuleForSku, + listChannelRules, + switchChannelForOrderItems, + upsertChannelRule, + } = 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 voucherSku = `集成测试卡券-${suffix}` const orderIds: number[] = [] const orderItemIds: number[] = [] - const internalSkus = [manualSku, missingAffiliateSku] + const internalSkus = [manualSku, missingAffiliateSku, voucherSku] async function createPendingOrderItem(sku: string) { const order = await query<{ id: number }>( @@ -84,6 +92,33 @@ test( assert.equal(manualItem.rows[0]?.fulfillment_channel, 'manual') assert.ok(Number(manualItem.rows[0]?.fulfillment_config_id) > 0) + const switched = await switchChannelForOrderItems({ + orderItemIds: [manualItemId], + channel: 'third_party_voucher', + adminUserId: null, + note: '集成测试切换', + }) + assert.equal(switched.switched, 1) + assert.equal(switched.failed, 0) + const switchedItem = await query<{ + fulfillment_status: string + fulfillment_channel: string + original_channel: string | null + canonical_product_id: number | null + canonical_sku: string | null + }>( + `SELECT oi.fulfillment_status, oi.fulfillment_channel, oi.original_channel, + oi.canonical_product_id, product.canonical_sku + FROM order_items oi + LEFT JOIN product_catalog product ON product.id = oi.canonical_product_id + WHERE oi.id = $1`, + [manualItemId], + ) + assert.equal(switchedItem.rows[0]?.fulfillment_status, 'assigned') + assert.equal(switchedItem.rows[0]?.fulfillment_channel, 'third_party_voucher') + assert.equal(switchedItem.rows[0]?.original_channel, 'manual') + assert.ok(switchedItem.rows[0]?.canonical_sku?.startsWith('tpv:')) + const missingAffiliateItemId = await createPendingOrderItem(missingAffiliateSku) const affiliateFailed = await assignChannelByInternalSku({ provider: '91kaquan', @@ -107,6 +142,34 @@ test( 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') + + await ensureVoucherRuleForSku(voucherSku) + const voucherChannel = await query<{ channel: string; source: string }>( + `SELECT channel, source FROM internal_sku_channels + WHERE internal_sku = $1 ORDER BY channel`, + [voucherSku], + ) + assert.deepEqual(voucherChannel.rows, [{ channel: 'third_party_voucher', source: 'voucher' }]) + const legacyVoucherRules = await query<{ count: number }>( + `SELECT count(*)::int AS count FROM sku_channel_rules + WHERE match_key = $1`, + [voucherSku], + ) + assert.equal(legacyVoucherRules.rows[0]?.count, 0) + const savedRule = await upsertChannelRule({ + matchKeyType: 'internal_sku', + matchKey: voucherSku, + channel: 'third_party_voucher', + priority: 10, + enabled: true, + note: '集成测试统一渠道', + adminUserId: null, + }) + const listedRules = await listChannelRules({ channel: 'third_party_voucher', enabled: true }) + assert.ok( + listedRules.some((rule) => rule.id === savedRule.id && rule.match_key === voucherSku), + ) + assert.equal((await deleteChannelRule(savedRule.id)).deleted, true) } finally { if (orderItemIds.length > 0) { await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [ 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 635a8d32..908d5ae4 100644 --- a/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts +++ b/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts @@ -78,10 +78,14 @@ export async function applyIntakeChannelDecision( const canOverrideDefaultManual = !selected || isDefaultManualFallback const existingAffiliateSku = String(parseObject(snapshot.affiliateDash).sku || '').trim() + const configuredAffiliateSku = String(primary.config_json.affiliateSku || '').trim() const desiredChannel = options.preferredChannel || primary.channel if (desiredChannel === 'affiliate_dash' || selected === 'affiliate_dash') { - const dashSku = existingAffiliateSku || matchAffiliateDashSkuByInternalSku(internalSku) + const dashSku = + existingAffiliateSku || + configuredAffiliateSku || + matchAffiliateDashSkuByInternalSku(internalSku) if (dashSku) { if (canOverrideDefaultManual || selected === 'affiliate_dash') { nextSnapshot.fulfillmentRoute = { diff --git a/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts b/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts index 0bd3df58..2447ad19 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.test.ts @@ -8,6 +8,7 @@ import { function channel(patch: Partial): InternalSkuChannelRow { return { + id: 1, internal_sku: '内部SKU', channel: 'manual', priority: 10, @@ -16,6 +17,8 @@ function channel(patch: Partial): InternalSkuChannelRow { canonical_product_id: null, config_json: {}, source: 'test', + note: '', + created_by_admin_user_id: null, ...patch, } } 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 3dfdb563..a32f4b11 100644 --- a/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts @@ -3,6 +3,7 @@ import type { FulfillmentChannel } from './channel-types.js' import { compareFulfillmentChannel } from './internal-sku-resolver.js' export type InternalSkuChannelRow = { + id: number internal_sku: string channel: FulfillmentChannel priority: number @@ -11,6 +12,8 @@ export type InternalSkuChannelRow = { canonical_product_id: number | null config_json: Record source: string + note: string + created_by_admin_user_id: number | null } export type InternalSkuChannelPlan = { @@ -27,8 +30,8 @@ export async function listInternalSkuChannels( if (!key) return [] const result = await query( ` - SELECT internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, - config_json, source + SELECT id, internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, + config_json, source, note, created_by_admin_user_id FROM internal_sku_channels WHERE btrim(internal_sku) = $1 AND enabled = TRUE ORDER BY @@ -89,11 +92,13 @@ export async function upsertInternalSkuChannel(input: { canonicalProductId?: number | null configJson?: Record | undefined source?: string + note?: string | undefined + adminUserId?: number | null | undefined replacePriority?: boolean -}): Promise { +}): Promise { const internalSku = String(input.internalSku || '').trim() - if (!internalSku) return - await withTransaction(async (client) => { + if (!internalSku) return 0 + return withTransaction(async (client) => { if (input.enabled !== false && input.isFallback !== true) { await client.query( `UPDATE internal_sku_channels @@ -107,15 +112,16 @@ export async function upsertInternalSkuChannel(input: { [internalSku, input.channel], ) } - await client.query( + const result = await client.query<{ id: number }>( ` INSERT INTO internal_sku_channels ( internal_sku, channel, priority, is_fallback, enabled, - canonical_product_id, config_json, source, created_at, updated_at - ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, NOW(), NOW()) + canonical_product_id, config_json, source, note, created_by_admin_user_id, + created_at, updated_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, NOW(), NOW()) ON CONFLICT (internal_sku, channel) DO UPDATE SET priority = CASE - WHEN $9::boolean THEN EXCLUDED.priority + WHEN $11::boolean THEN EXCLUDED.priority ELSE LEAST(internal_sku_channels.priority, EXCLUDED.priority) END, is_fallback = EXCLUDED.is_fallback, @@ -123,7 +129,10 @@ export async function upsertInternalSkuChannel(input: { canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), config_json = internal_sku_channels.config_json || EXCLUDED.config_json, source = EXCLUDED.source, + note = CASE WHEN EXCLUDED.note <> '' THEN EXCLUDED.note ELSE internal_sku_channels.note END, + created_by_admin_user_id = COALESCE(EXCLUDED.created_by_admin_user_id, internal_sku_channels.created_by_admin_user_id), updated_at = NOW() + RETURNING id `, [ internalSku, @@ -134,9 +143,12 @@ export async function upsertInternalSkuChannel(input: { input.canonicalProductId ?? null, JSON.stringify(input.configJson || {}), String(input.source || 'auto').trim() || 'auto', + String(input.note || '').slice(0, 300), + input.adminUserId ?? null, input.replacePriority === true, ], ) + return Number(result.rows[0]?.id || 0) }) } @@ -149,22 +161,12 @@ export async function setInternalSkuPrimaryChannel(input: { canonicalProductId?: number | null source?: string configJson?: Record | undefined + note?: string | undefined + adminUserId?: number | null }): 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], - ) await upsertInternalSkuChannel({ internalSku, channel: input.channel, @@ -174,6 +176,8 @@ export async function setInternalSkuPrimaryChannel(input: { canonicalProductId: input.canonicalProductId ?? null, configJson: input.configJson, source: input.source || 'operator', + note: input.note, + adminUserId: input.adminUserId, replacePriority: true, }) } @@ -192,3 +196,10 @@ export async function disableInternalSkuChannel(input: { ) return result.rowCount || 0 } + +export async function deleteInternalSkuChannelById(channelId: number): Promise { + const id = Math.trunc(Number(channelId) || 0) + if (id <= 0) return false + const result = await query(`DELETE FROM internal_sku_channels WHERE id = $1`, [id]) + return (result.rowCount || 0) > 0 +} diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 0a889df3..475a06a0 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -213,7 +213,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { ` SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code FROM order_items oi - LEFT JOIN work_orders wo ON wo.order_item_id = oi.id + LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled' WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order') AND wo.id IS NULL @@ -221,6 +221,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled') ) AND oi.id > $2 ORDER BY oi.id @@ -261,16 +262,18 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { FROM order_items oi WHERE oi.fulfillment_status = 'assigned' AND ( - EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled') )::int + ( EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled') ) )::int + ( EXISTS ( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled') ) )::int > 1 ORDER BY oi.id @@ -362,11 +365,12 @@ async function resolveCarrierDispatchExceptions(): Promise { AND ce.status <> 'resolved' AND ( (oi.fulfillment_channel = 'work_order' - AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)) + AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled')) 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 = 'affiliate_dash' + AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled') )) ) LIMIT 500 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 2eec3d7b..000716a0 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 @@ -55,6 +55,7 @@ export async function syncWorkerOrdersForSourceOrder( autoOnly?: boolean sourceMetadata?: JsonObject skipChannelAssign?: boolean + preferredWorkRuleKey?: string } = {}, ) { const [rules, mappings] = await Promise.all([ @@ -80,9 +81,36 @@ export async function syncWorkerOrdersForSourceOrder( continue } - const match = resolveMatchingProductRuleDecision(order, item, rules, mappings, { - legacyFallbackEnabled: productMatchConfig.legacyFallbackEnabled, - }) + const preferredRuleKey = String(options.preferredWorkRuleKey || '').trim() + const preferredRule = preferredRuleKey + ? rules.find((rule) => String(rule.rule_key || '').trim() === preferredRuleKey) || null + : null + const match = + preferredRuleKey && !preferredRule + ? { + rule: null, + mappingId: null, + reason: 'unmatched' as const, + score: 0, + candidates: [], + diagnosis: [ + { + dimension: 'internal_sku_channel', + reason: 'configured_work_rule_missing', + }, + ], + } + : preferredRule + ? { + rule: preferredRule, + mappingId: null, + reason: 'matched' as const, + score: 1, + candidates: [{ ruleKey: preferredRuleKey, score: 1, mappingId: null }], + } + : resolveMatchingProductRuleDecision(order, item, rules, mappings, { + legacyFallbackEnabled: productMatchConfig.legacyFallbackEnabled, + }) const matchContext = resolveKuaishouWorkProductMatchContext(item) const snapshot = safeParseJson(item.item_snapshot_json) const sellerId = diff --git a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx index a4942e86..df7de64a 100644 --- a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx +++ b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx @@ -58,11 +58,7 @@ const CHANNEL_COLOR: Record = { manual: 'orange', } -const MATCH_KEY_OPTIONS = [ - { value: 'sku_code', label: '平台 SKU' }, - { value: 'internal_sku', label: '内部 SKU' }, - { value: 'sku_name', label: 'SKU 名称' }, -] +const MATCH_KEY_OPTIONS = [{ value: 'internal_sku', label: '内部 SKU' }] export default function AdminFulfillmentChannelPage() { return ( @@ -214,7 +210,7 @@ function PendingPanel() { type="info" showIcon style={{ marginBottom: 12 }} - message="待分类订单的收入已经计入「未分类」桶，不会丢失；指定渠道后会自动进入商品归一与成本核算。" + message="指定渠道会写入该内部 SKU 的全局配置，并影响所有平台和店铺；系统会先确认载体创建成功，再进入商品归一与成本核算。" /> rowKey(row)} @@ -259,8 +255,7 @@ function RulesPanel() { }) const columns = [ - { title: '来源', dataIndex: 'provider', width: 120, render: (v: string) => v || '不限' }, - { title: '店铺', dataIndex: 'shop_id', width: 140, render: (v: string) => v || '不限' }, + { title: '作用范围', width: 110, render: () => 全局 }, { title: '匹配键', dataIndex: 'match_key_type', @@ -303,7 +298,7 @@ function RulesPanel() { return (