diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts new file mode 100644 index 00000000..9d135354 --- /dev/null +++ b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts @@ -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 { + 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 { + 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) +}) diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts new file mode 100644 index 00000000..50b43051 --- /dev/null +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -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 +} + +/** 批量判定:默认只扫 pending 且未锁定的订单项。 */ +export async function assignPendingOrderItemChannels( + options: { batchSize?: number; maxBatches?: number; afterId?: number } = {}, +): Promise { + 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( + ` + 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( + ` + 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 { + 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( + ` + 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 { + const rows = await query( + ` + 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 diff --git a/apps/backend/src/services/fulfillment/channel-types.ts b/apps/backend/src/services/fulfillment/channel-types.ts new file mode 100644 index 00000000..3932b53c --- /dev/null +++ b/apps/backend/src/services/fulfillment/channel-types.ts @@ -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 = { + 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) +}