fix(finance): 收敛内部 SKU 履约路由与载体状态

统一内部 SKU 与渠道主/兜底解析,禁止并列主渠道。

运营指定渠道改为先创建真实载体,失败保持明确 pending。

历史缺载体、多载体冲突和自动发货兜底统一由管道修复。

迁移人工配置与旧 sku_code 规则,补充严格预检和集成测试。
This commit is contained in:
yml2213
2026-11-02 15:02:49 +08:00
parent 5fe33b0b46
commit 2f87445ed9
33 changed files with 2199 additions and 446 deletions
@@ -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> 商品稳定定位内部 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;
@@ -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;
@@ -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<string, unknown> = {}) {
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<string, unknown> =
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<T>(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<void> {
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,
],
)
}
@@ -215,21 +215,6 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null
[internalSku],
)
const skuCode = String(item.sku_code || '').trim()
if (skuCode) {
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 ($1, $2, 'sku_code', $3, 'third_party_voucher', 10, 'voucher', TRUE,
'卡券订单自动建商品后生成', NOW(), NOW())
ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING
`,
[item.provider, item.shop_id, skuCode],
)
}
return productId
}
@@ -28,7 +28,7 @@ export async function runAffiliateFallback(
order_item_id: number
provider: string
shop_id: string
manual_sku_key: string | null
internal_sku: string
sku_name: string
config_id: number | null
basis_date: string
@@ -39,23 +39,34 @@ export async function runAffiliateFallback(
oi.id AS order_item_id,
o.provider,
o.shop_id,
oi.manual_sku_key,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku,
oi.sku_name,
cfg.id AS config_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
LEFT JOIN manual_fulfillment_configs cfg
ON cfg.sku_key = oi.manual_sku_key AND cfg.enabled
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 = 'affiliate_dash'
AND oi.fulfillment_locked = FALSE
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 = ANY($1::text[])
SELECT 1 FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND channel.channel = 'manual'
AND channel.is_fallback = TRUE
AND btrim(channel.internal_sku) = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
)
)
AND (
SELECT task.task_status
FROM fulfillment_tasks task
WHERE task.order_item_id = oi.id AND task.executor_key = 'affiliate_dash'
ORDER BY task.unit_index DESC, task.updated_at DESC, task.id DESC
LIMIT 1
) = ANY($1::text[])
AND NOT EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id
@@ -0,0 +1,202 @@
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 { applyIntakeChannelDecision } from './intake-channel-decision-service.js'
import type { ResolvedFulfillmentItem } from './product-resolution-service.js'
function parseObject(value: unknown): Record<string, unknown> {
if (value && typeof value === 'object' && !Array.isArray(value)) {
return value as Record<string, unknown>
}
try {
const parsed = JSON.parse(String(value || '{}'))
return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: {}
} catch {
return {}
}
}
export type EnsureCarrierResult = {
ready: boolean
created: boolean
reason: string
message: string
}
async function hasWorkOrder(orderItemId: number): Promise<boolean> {
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<boolean> {
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<EnsureCarrierResult> {
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 ? '' : '自动发货任务规划失败',
}
}
@@ -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<string, unknown> {
if (value && typeof value === 'object' && !Array.isArray(value)) {
return value as Record<string, unknown>
}
try {
const parsed = JSON.parse(String(value || '{}'))
return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: {}
} catch {
return {}
}
function normalizeChannels(value: unknown): InternalSkuChannelRow[] {
if (!Array.isArray(value)) return []
return value
.filter((row): row is Record<string, unknown> => 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<string, unknown>)
: {},
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<PendingCarrierItem>(
async function loadCandidateItems(input: {
limit: number
orderItemIds?: number[]
assignedMissingCarrier?: boolean
}): Promise<CarrierCandidate[]> {
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<CarrierCandidate>(
`
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<void> {
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<boolean> {
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<number, PendingCarrierItem[]>()
for (const row of rows.rows) {
let fallbacksSelected = 0
let failed = 0
const byOrder = new Map<number, CarrierCandidate[]>()
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<OrderItemRow>(`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),
}
}
@@ -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> = {}): ChannelCandidateItem {
return {
@@ -20,12 +21,26 @@ function buildItem(patch: Partial<ChannelCandidateItem> = {}): 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> = {}): 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> = {}): ChannelRule {
return {
id: 1,
@@ -40,13 +55,13 @@ function buildRule(patch: Partial<ChannelRule> = {}): 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')
@@ -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<OrderItemRow>(`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<typeof order> => 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<ChannelRule[]> {
@@ -735,6 +995,32 @@ async function loadEnabledRules(): Promise<ChannelRule[]> {
return rows.rows
}
async function loadInternalSkuChannelsForKeys(
keys: string[],
): Promise<Map<string, InternalSkuChannelRow[]>> {
const normalized = [...new Set(keys.map(normalizeKey).filter(Boolean))]
const result = new Map<string, InternalSkuChannelRow[]>()
if (normalized.length === 0) return result
const rows = await query<InternalSkuChannelRow>(
`
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
@@ -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] | ''
@@ -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()
}
},
)
@@ -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<string, unknown> {
: {}
}
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<ResolvedFulfillmentItem> {
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<string, unknown> = {
...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 }
}
@@ -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>): 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'],
)
})
@@ -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<string, unknown>
source: string
}
export type InternalSkuChannelPlan = {
primary: InternalSkuChannelRow | null
fallbacks: InternalSkuChannelRow[]
attempts: InternalSkuChannelRow[]
conflicts: InternalSkuChannelRow[]
}
export async function listInternalSkuChannels(
internalSku: string,
): Promise<InternalSkuChannelRow[]> {
@@ -18,18 +27,12 @@ export async function listInternalSkuChannels(
if (!key) return []
const result = await query<InternalSkuChannelRow>(
`
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<FulfillmentChannel | null> {
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<string, unknown> | undefined
source?: string
replacePriority?: boolean
}): Promise<void> {
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<string, unknown> | undefined
}): Promise<void> {
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<number> {
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
}
@@ -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 已有中文 → 不查库;
@@ -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',
)
})
@@ -0,0 +1,61 @@
import type { FulfillmentChannel } from './channel-types.js'
type JsonObject = Record<string, unknown>
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)
}
@@ -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,
}
@@ -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<number> {
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<number> {
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 } })
@@ -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 = {
@@ -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<string, unknown>
item_snapshot_json?: string | Record<string, unknown>
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<string, unknown> | null
created_at?: string
updated_at?: string
}