feat(fulfillment): 新增渠道判定管道,规则优先、载体证据兜底

- 只写 channel/status/source,不再由「有没有任务」反推渠道
- 同优先级多渠道判为冲突并保持 pending,进配置冲突队列
- 运营指定渠道即锁定,自动判定不再覆盖
- 待分类队列按 SKU 聚合,支持按 SKU 批量指定
This commit is contained in:
yml2213
2026-10-31 18:12:06 +08:00
parent 4cf700e910
commit 8cdc2423d5
3 changed files with 702 additions and 0 deletions
@@ -0,0 +1,118 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import {
resolveChannelDecision,
type ChannelCandidateItem,
type ChannelRule,
} from './channel-assign-service.js'
function buildItem(patch: Partial<ChannelCandidateItem> = {}): ChannelCandidateItem {
return {
id: 1,
provider: '91kaquan',
shop_id: 'shop-a',
sku_code: 'suit_xxx',
sku_name: '套装-测试',
manual_sku_key: 'suit_xxx',
normalized_sku_name: '套装-测试',
has_work_order: false,
has_affiliate_task: false,
...patch,
}
}
function buildRule(patch: Partial<ChannelRule> = {}): ChannelRule {
return {
id: 1,
provider: '',
shop_id: '',
match_key_type: 'sku_code',
match_key: 'suit_xxx',
channel: 'affiliate_dash',
priority: 60,
enabled: true,
...patch,
}
}
test('渠道判定:精确 SKU 规则优先于载体证据', () => {
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.conflict, false)
})
test('渠道判定:无规则时按载体证据判定', () => {
const workOrder = resolveChannelDecision(buildItem({ has_work_order: true }), [])
assert.equal(workOrder.channel, 'work_order')
assert.equal(workOrder.source, 'carrier_evidence')
const affiliate = resolveChannelDecision(buildItem({ has_affiliate_task: true }), [])
assert.equal(affiliate.channel, 'affiliate_dash')
assert.equal(affiliate.source, 'carrier_evidence')
})
test('渠道判定:既无规则也无载体时保持 pending,不兜底人工', () => {
const decision = resolveChannelDecision(buildItem(), [])
assert.equal(decision.channel, null)
assert.equal(decision.conflict, false)
})
test('渠道判定:同优先级多渠道判为冲突,不静默选择', () => {
const decision = resolveChannelDecision(buildItem(), [
buildRule({ id: 1, channel: 'affiliate_dash', priority: 10 }),
buildRule({ id: 2, channel: 'manual', priority: 10 }),
])
assert.equal(decision.channel, null)
assert.equal(decision.conflict, true)
assert.deepEqual(decision.conflictChannels.sort(), ['affiliate_dash', 'manual'])
})
test('渠道判定:不同优先级时取优先级更小的', () => {
const decision = resolveChannelDecision(buildItem(), [
buildRule({ id: 1, channel: 'affiliate_dash', priority: 60 }),
buildRule({ id: 2, channel: 'manual', priority: 70 }),
])
assert.equal(decision.channel, 'affiliate_dash')
})
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')
})
test('渠道判定:内部 SKU 与 SKU 名称两种键都能命中', () => {
const byInternal = buildRule({
match_key_type: 'internal_sku',
match_key: 'nick-001',
channel: 'manual',
})
assert.equal(
resolveChannelDecision(buildItem({ manual_sku_key: 'nick-001' }), [byInternal]).channel,
'manual',
)
const byName = buildRule({
match_key_type: 'sku_name',
match_key: '套装-测试',
channel: 'third_party_voucher',
})
assert.equal(
resolveChannelDecision(buildItem({ normalized_sku_name: '套装-测试' }), [byName]).channel,
'third_party_voucher',
)
})
test('渠道判定:停用的规则不参与', () => {
const decision = resolveChannelDecision(buildItem(), [buildRule({ enabled: false })])
assert.equal(decision.channel, null)
})
@@ -0,0 +1,512 @@
import { query } from '../../db/client.js'
import { createHttpError } from '../../utils/http.js'
import { nowIso } from '../../utils/time.js'
import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
import {
FULFILLMENT_CHANNELS,
MATCH_KEY_TYPES,
RULE_SOURCES,
isFulfillmentChannel,
isMatchKeyType,
isRuleSource,
type FulfillmentChannel,
type FulfillmentSource,
type MatchKeyType,
type RuleSource,
} from './channel-types.js'
/**
* 管道 A:渠道判定。
*
* 只写 `fulfillment_channel / fulfillment_status / fulfillment_source /
* fulfillment_decided_at`,绝不写商品、成本、事实,也绝不改旧 fulfillment_mode。
*
* 判定顺序:
* 1. 运营锁定 → 跳过
* 2. 精确规则(sku_channel_rules,优先级升序);同优先级命中多渠道 → 冲突,保持 pending
* 3. 载体证据:已有工单 → work_order;已有自动发货任务 → affiliate_dash
* 4. 都没有 → 保持 pending(不再兜底成人工)
*/
export type ChannelRule = {
id: number
provider: string
shop_id: string
match_key_type: MatchKeyType
match_key: string
channel: FulfillmentChannel
priority: number
enabled: boolean
}
export type ChannelCandidateItem = {
id: number
provider: string
shop_id: string
sku_code: string
sku_name: string
manual_sku_key: string | null
normalized_sku_name: string | null
has_work_order: boolean
has_affiliate_task: boolean
}
export type ChannelDecision = {
channel: FulfillmentChannel | null
source: FulfillmentSource
conflict: boolean
conflictChannels: FulfillmentChannel[]
matchedRuleIds: number[]
}
function normalizeKey(value: unknown): string {
return String(value ?? '').trim()
}
function matchesScope(rule: ChannelRule, item: ChannelCandidateItem): boolean {
if (rule.provider && rule.provider !== item.provider) return false
if (rule.shop_id && rule.shop_id !== item.shop_id) return false
return true
}
function itemKey(item: ChannelCandidateItem, keyType: MatchKeyType): string {
if (keyType === 'sku_code') return normalizeKey(item.sku_code)
if (keyType === 'internal_sku') return normalizeKey(item.manual_sku_key)
return normalizeKey(item.normalized_sku_name) || normalizeKey(item.sku_name).toLowerCase()
}
/** 纯函数:给定规则与订单项,得出渠道决策(便于单测与文档对照)。 */
export function resolveChannelDecision(
item: ChannelCandidateItem,
rules: ChannelRule[],
): ChannelDecision {
const hits = rules.filter(
(rule) =>
rule.enabled &&
matchesScope(rule, item) &&
itemKey(item, rule.match_key_type) === normalizeKey(rule.match_key),
)
if (hits.length > 0) {
const minPriority = Math.min(...hits.map((rule) => rule.priority))
const best = hits.filter((rule) => rule.priority === minPriority)
const channels = [...new Set(best.map((rule) => rule.channel))]
if (channels.length > 1) {
return {
channel: null,
source: '',
conflict: true,
conflictChannels: channels,
matchedRuleIds: best.map((rule) => rule.id),
}
}
return {
channel: channels[0]!,
source: 'auto_rule',
conflict: false,
conflictChannels: [],
matchedRuleIds: best.map((rule) => rule.id),
}
}
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: [],
}
}
return { channel: null, source: '', conflict: false, conflictChannels: [], matchedRuleIds: [] }
}
export type AssignChannelsResult = {
scanned: number
assigned: number
conflicted: number
stillPending: number
lastId: number
byChannel: Record<string, number>
}
/** 批量判定:默认只扫 pending 且未锁定的订单项。 */
export async function assignPendingOrderItemChannels(
options: { batchSize?: number; maxBatches?: number; afterId?: number } = {},
): Promise<AssignChannelsResult> {
const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500)))
const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20)))
let cursor = Math.max(0, Math.trunc(options.afterId || 0))
const rules = await loadEnabledRules()
const result: AssignChannelsResult = {
scanned: 0,
assigned: 0,
conflicted: 0,
stillPending: 0,
lastId: cursor,
byChannel: {},
}
for (let batch = 0; batch < maxBatches; batch += 1) {
const rows = await query<ChannelCandidateItem>(
`
SELECT
oi.id,
o.provider,
o.shop_id,
oi.sku_code,
oi.sku_name,
oi.manual_sku_key,
oi.normalized_sku_name,
EXISTS (
SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id
) AS has_work_order,
EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
) AS has_affiliate_task
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'pending'
AND oi.fulfillment_locked = FALSE
AND oi.id > $1
ORDER BY oi.id
LIMIT $2
`,
[cursor, batchSize],
)
if (rows.rows.length === 0) break
result.scanned += rows.rows.length
cursor = Number(rows.rows[rows.rows.length - 1]!.id)
result.lastId = cursor
const assignments: Array<{
id: number
channel: FulfillmentChannel
source: FulfillmentSource
}> = []
for (const item of rows.rows) {
const decision = resolveChannelDecision(item, rules)
if (decision.conflict) {
result.conflicted += 1
await upsertAggregatedException({
reason: 'routing_conflict',
scopeKey: `routing_conflict|${item.provider}|${item.shop_id}|${item.sku_code}|-`,
orderItemId: item.id,
detail: { channels: decision.conflictChannels, ruleIds: decision.matchedRuleIds },
})
continue
}
if (!decision.channel) {
result.stillPending += 1
continue
}
assignments.push({ id: item.id, channel: decision.channel, source: decision.source })
result.byChannel[decision.channel] = (result.byChannel[decision.channel] || 0) + 1
}
if (assignments.length > 0) {
const updated = await query(
`
UPDATE order_items oi
SET
fulfillment_channel = input.channel,
fulfillment_status = 'assigned',
fulfillment_source = input.source,
fulfillment_decided_at = NOW(),
updated_at = NOW()
FROM unnest($1::bigint[], $2::text[], $3::text[])
AS input(id, channel, source)
WHERE oi.id = input.id
AND oi.fulfillment_status = 'pending'
AND oi.fulfillment_locked = FALSE
`,
[
assignments.map((row) => row.id),
assignments.map((row) => row.channel),
assignments.map((row) => row.source),
],
)
result.assigned += updated.rowCount || 0
}
}
return result
}
/** 运营指定渠道:写入并锁定,自动判定不再覆盖。 */
export async function assignChannelToOrderItems(input: {
orderItemIds: number[]
channel: string
adminUserId: number | null
note?: string
}): Promise<{ updated: number }> {
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 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)],
)
return { updated: result.rowCount || 0 }
}
/** 运营解锁:允许自动判定重新接管。 */
export async function unlockChannel(orderItemIds: number[]): Promise<{ updated: number }> {
const ids = [...new Set(orderItemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0))]
if (ids.length === 0) return { updated: 0 }
const result = await query(
`UPDATE order_items SET fulfillment_locked = FALSE, updated_at = NOW() WHERE id = ANY($1::bigint[])`,
[ids],
)
return { updated: result.rowCount || 0 }
}
export type PendingGroupRow = {
provider: string
shop_id: string
sku_code: string
sku_name: string
item_count: number
revenue_fen: number
first_seen_at: string
last_seen_at: string
}
/** 待分类队列:按 SKU 聚合,而不是一单一行。 */
export async function listPendingChannelGroups(input: {
page?: number
pageSize?: number
keyword?: string
}): Promise<{ items: PendingGroupRow[]; total: number; page: number; pageSize: number }> {
const page = Math.max(1, Math.trunc(Number(input.page) || 1))
const pageSize = Math.min(200, Math.max(1, Math.trunc(Number(input.pageSize) || 30)))
const keyword = String(input.keyword || '').trim()
const params: unknown[] = []
const clauses = [`oi.fulfillment_status = 'pending'`]
if (keyword) {
params.push(`%${keyword}%`)
clauses.push(`(oi.sku_code ILIKE $${params.length} OR oi.sku_name ILIKE $${params.length})`)
}
const where = clauses.join(' AND ')
const total = await query<{ count: number }>(
`
SELECT COUNT(*)::int AS count FROM (
SELECT 1
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE ${where}
GROUP BY o.provider, o.shop_id, oi.sku_code, oi.sku_name
) grouped
`,
params,
)
params.push(pageSize, (page - 1) * pageSize)
const rows = await query<PendingGroupRow>(
`
SELECT
o.provider,
o.shop_id,
oi.sku_code,
oi.sku_name,
COUNT(*)::int AS item_count,
COALESCE(SUM(oi.revenue_amount_fen), 0)::bigint AS revenue_fen,
MIN(oi.created_at) AS first_seen_at,
MAX(oi.created_at) AS last_seen_at
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE ${where}
GROUP BY o.provider, o.shop_id, oi.sku_code, oi.sku_name
ORDER BY item_count DESC, oi.sku_code
LIMIT $${params.length - 1} OFFSET $${params.length}
`,
params,
)
return { items: rows.rows, total: total.rows[0]?.count || 0, page, pageSize }
}
/** 按 SKU 批量指定(待分类队列的主操作)。 */
export async function assignChannelBySku(input: {
provider: string
shopId: string
skuCode: string
channel: string
adminUserId: number | null
note?: string
}): Promise<{ updated: number }> {
if (!isFulfillmentChannel(input.channel)) {
throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' })
}
const result = await query(
`
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()
FROM orders o
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'
`,
[input.provider, input.shopId, input.skuCode, input.channel, input.adminUserId],
)
return { updated: result.rowCount || 0 }
}
export async function listChannelRules(
input: {
channel?: string
enabled?: boolean
} = {},
): Promise<ChannelRule[]> {
const params: unknown[] = []
const clauses: string[] = []
if (input.channel) {
params.push(input.channel)
clauses.push(`channel = $${params.length}`)
}
if (typeof input.enabled === 'boolean') {
params.push(input.enabled)
clauses.push(`enabled = $${params.length}`)
}
const where = clauses.length > 0 ? `WHERE ${clauses.join(' AND ')}` : ''
const rows = await query<ChannelRule>(
`
SELECT id, provider, shop_id, match_key_type, match_key, channel, priority, enabled
FROM sku_channel_rules
${where}
ORDER BY priority, channel, match_key
`,
params,
)
return rows.rows
}
export async function upsertChannelRule(input: {
id?: number
provider?: string
shopId?: string
matchKeyType: string
matchKey: string
channel: string
priority?: number
source?: string
enabled?: boolean
note?: string
adminUserId: number | null
}): Promise<{ id: number }> {
if (!isMatchKeyType(input.matchKeyType)) {
throw createHttpError('匹配键类型无效', {
statusCode: 400,
errorCode: 'invalid_match_key_type',
})
}
if (!isFulfillmentChannel(input.channel)) {
throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' })
}
const source: RuleSource = isRuleSource(input.source) ? input.source : 'operator'
const matchKey = String(input.matchKey || '').trim()
if (!matchKey) {
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,
],
)
return { id: result.rows[0]!.id }
}
export async function deleteChannelRule(ruleId: number): Promise<{ deleted: boolean }> {
const id = Math.trunc(Number(ruleId) || 0)
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 }
}
async function loadEnabledRules(): Promise<ChannelRule[]> {
const rows = await query<ChannelRule>(
`
SELECT id, provider, shop_id, match_key_type, match_key, channel, priority, enabled
FROM sku_channel_rules
WHERE enabled = TRUE
`,
)
return rows.rows
}
export const CHANNEL_RULE_KEY_TYPES = MATCH_KEY_TYPES
export const CHANNEL_RULE_SOURCES = RULE_SOURCES
export const CHANNEL_VALUES = FULFILLMENT_CHANNELS
@@ -0,0 +1,72 @@
/**
* 履约渠道 v2:渠道 / 状态 / 决策来源的唯一枚举定义。
*
* 与旧 `fulfillment_mode` 的区别:
* · 渠道只表达「谁来做」,不包含 pending/archived 这类状态;
* · 「待分类」是状态,不是渠道;
* · 「第三方卡券」是渠道,但没有履约载体,只做统计。
*/
export const FULFILLMENT_CHANNELS = [
'work_order',
'affiliate_dash',
'third_party_voucher',
'manual',
] as const
export type FulfillmentChannel = (typeof FULFILLMENT_CHANNELS)[number]
export const FULFILLMENT_STATUSES = ['pending', 'assigned', 'archived'] as const
export type FulfillmentStatus = (typeof FULFILLMENT_STATUSES)[number]
export const FULFILLMENT_SOURCES = [
'auto_rule',
'auto_mapping',
'carrier_evidence',
'manual_confirm',
'operator_assigned',
'cutover_archive',
] as const
export type FulfillmentSource = (typeof FULFILLMENT_SOURCES)[number] | ''
export const CHANNEL_LABELS: Record<FulfillmentChannel, string> = {
work_order: '接单平台',
affiliate_dash: '自动发货',
third_party_voucher: '第三方卡券',
manual: '人工客服',
}
export const MATCH_KEY_TYPES = ['sku_code', 'internal_sku', 'sku_name'] as const
export type MatchKeyType = (typeof MATCH_KEY_TYPES)[number]
export const RULE_SOURCES = ['operator', 'affiliate', 'manual', 'voucher', 'template'] as const
export type RuleSource = (typeof RULE_SOURCES)[number]
/** 渠道是否需要履约载体(卡券渠道不需要)。 */
export function channelNeedsCarrier(channel: FulfillmentChannel | null | undefined): boolean {
if (!channel) return false
return channel !== 'third_party_voucher'
}
/** 渠道是否参与成本核算。 */
export function channelHasCostModel(channel: FulfillmentChannel | null | undefined): boolean {
return (
channel === 'affiliate_dash' || channel === 'work_order' || channel === 'third_party_voucher'
)
}
export function isFulfillmentChannel(value: unknown): value is FulfillmentChannel {
return typeof value === 'string' && (FULFILLMENT_CHANNELS as readonly string[]).includes(value)
}
export function isMatchKeyType(value: unknown): value is MatchKeyType {
return typeof value === 'string' && (MATCH_KEY_TYPES as readonly string[]).includes(value)
}
export function isRuleSource(value: unknown): value is RuleSource {
return typeof value === 'string' && (RULE_SOURCES as readonly string[]).includes(value)
}