dachui/order_site
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(fulfillment): 建立履约统一状态视图与分类队列

Browse Source
This commit is contained in:
yml2213
2026-11-05 20:13:34 +08:00
parent e8b76b3b73
commit ed450fac79
12 changed files with 1783 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/services/fulfillment/channel-assign-service.ts
+278 -79
View File
@@ -1016,6 +1016,24 @@ export type PendingGroupRow = {
revenue_fen: number
first_seen_at: string
last_seen_at: string
issue_breakdown: Array<{
issue_code: string
issue_category: string
item_count: number
locked_item_count: number
retryable: boolean
auto_retryable: boolean
owner_role: string
}>
retryable_item_count: number
retryable_unlocked_item_count: number
auto_retryable_item_count: number
auto_retryable_unlocked_item_count: number
config_issue_item_count: number
manual_review_item_count: number
after_sales_item_count: number
system_issue_item_count: number
oldest_age_minutes: number
}
type PendingShopNameRow = {
@@ -1042,21 +1060,64 @@ const PENDING_GROUP_KEY_SQL = `
END
`
const PENDING_STATE_GROUP_KEY_SQL = `
CASE
WHEN COALESCE(BTRIM(v.internal_sku), '') <> ''
THEN 'sku:' || encode(convert_to(v.internal_sku, 'UTF8'), 'hex')
ELSE 'source:' || encode(
convert_to(
CONCAT_WS(CHR(31), v.provider, v.shop_id, v.sku_code),
'UTF8'
),
'hex'
)
END
`
const RETRYABLE_CARRIER_ISSUES = [
'carrier_closed',
'carrier_retry_pending',
'carrier_expired',
'carrier_missing',
'state_inconsistent',
] as const
const AUTO_RETRYABLE_CARRIER_ISSUES = ['carrier_missing', 'state_inconsistent'] as const
const PENDING_ISSUE_CATEGORIES = [
'classification',
'configuration',
'carrier',
'manual_review',
'after_sales',
'system',
] as const
/** 待分类队列：按全局内部 SKU 聚合（跨平台、跨店铺合并为一行）。 */
export async function listPendingChannelGroups(input: {
page?: number
pageSize?: number
keyword?: string
issueCategory?: 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 issueCategoryInput = String(input.issueCategory || '').trim()
const issueCategory = (PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategoryInput)
? issueCategoryInput
: ''
const params: unknown[] = []
const clauses = [`oi.fulfillment_status = 'pending'`]
const clauses = [`v.fulfillment_status = 'pending'`]
if (issueCategory) {
params.push(issueCategory)
clauses.push(`v.issue_category = $${params.length}`)
}
if (keyword) {
params.push(`%${keyword}%`)
clauses.push(
`(${INTERNAL_SKU_SQL} ILIKE $${params.length} OR oi.sku_code ILIKE $${params.length} OR oi.sku_name ILIKE $${params.length} OR oi.manual_sku_key ILIKE $${params.length})`,
`(v.internal_sku ILIKE $${params.length}
OR v.sku_code ILIKE $${params.length}
OR v.sku_name ILIKE $${params.length})`,
)
}
const where = clauses.join(' AND ')
@@ -1065,10 +1126,9 @@ export async function listPendingChannelGroups(input: {
`
WITH pending_items AS (
SELECT
${INTERNAL_SKU_SQL} AS internal_sku,
${PENDING_GROUP_KEY_SQL} AS group_key
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
v.internal_sku,
${PENDING_STATE_GROUP_KEY_SQL} AS group_key
FROM v_fulfillment_item_state v
WHERE ${where}
)
SELECT COUNT(DISTINCT group_key)::int AS count
@@ -1082,57 +1142,124 @@ export async function listPendingChannelGroups(input: {
`
WITH pending_items AS (
SELECT
oi.id,
oi.order_id,
oi.sku_code,
oi.sku_name,
oi.fulfillment_locked,
oi.fulfillment_pending_reason,
oi.fulfillment_pending_detail,
oi.revenue_amount_fen,
oi.created_at,
o.provider,
o.shop_id,
${INTERNAL_SKU_SQL} AS internal_sku,
${PENDING_GROUP_KEY_SQL} AS group_key
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
v.*,
${PENDING_STATE_GROUP_KEY_SQL} AS group_key
FROM v_fulfillment_item_state v
WHERE ${where}
),
grouped AS (
SELECT
group_key,
internal_sku,
MIN(sku_name) AS sku_name,
array_agg(DISTINCT provider) AS providers,
array_agg(DISTINCT shop_id) AS shop_ids,
array_agg(DISTINCT sku_code) AS sku_codes,
array_remove(
array_agg(
DISTINCT CASE
WHEN issue_code = 'missing_internal_sku' THEN 'no_internal_sku'
WHEN issue_code = 'missing_channel' THEN 'no_channel_capability'
WHEN issue_code = '' THEN NULL
ELSE 'no_carrier'
END
),
NULL
) AS pending_reasons,
array_remove(array_agg(DISTINCT issue_code), '') AS pending_subreasons,
COUNT(*) FILTER (WHERE fulfillment_locked)::int AS locked_item_count,
COUNT(*)::int AS item_count,
COALESCE(SUM(revenue_amount_fen), 0)::bigint AS revenue_fen,
MIN(created_at) AS first_seen_at,
MAX(created_at) AS last_seen_at,
COALESCE(MAX(age_minutes), 0)::bigint AS oldest_age_minutes,
COUNT(*) FILTER (WHERE retryable)::int AS retryable_item_count,
COUNT(*) FILTER (WHERE retryable AND NOT fulfillment_locked)::int
AS retryable_unlocked_item_count,
COUNT(*) FILTER (WHERE auto_retryable)::int AS auto_retryable_item_count,
COUNT(*) FILTER (WHERE auto_retryable AND NOT fulfillment_locked)::int
AS auto_retryable_unlocked_item_count,
COUNT(*) FILTER (WHERE issue_category = 'configuration')::int
AS config_issue_item_count,
COUNT(*) FILTER (WHERE issue_category = 'manual_review')::int
AS manual_review_item_count,
COUNT(*) FILTER (WHERE issue_category = 'after_sales')::int
AS after_sales_item_count,
COUNT(*) FILTER (WHERE issue_category = 'system')::int
AS system_issue_item_count
FROM pending_items
GROUP BY group_key, internal_sku
),
issue_counts AS (
SELECT
group_key,
issue_code,
issue_category,
bool_or(retryable) AS retryable,
bool_or(auto_retryable) AS auto_retryable,
MIN(owner_role) AS owner_role,
COUNT(*)::int AS item_count,
COUNT(*) FILTER (WHERE fulfillment_locked)::int AS locked_item_count
FROM pending_items
GROUP BY group_key, issue_code, issue_category
),
issue_summary AS (
SELECT
group_key,
jsonb_agg(
jsonb_build_object(
'issue_code', issue_code,
'issue_category', issue_category,
'item_count', item_count,
'locked_item_count', locked_item_count,
'retryable', retryable,
'auto_retryable', auto_retryable,
'owner_role', owner_role
)
ORDER BY item_count DESC, issue_code
) AS issue_breakdown
FROM issue_counts
GROUP BY group_key
)
SELECT
pi.group_key,
CASE WHEN pi.internal_sku <> '' THEN 'internal_sku' ELSE 'source_sku' END AS scope_type,
pi.internal_sku,
MIN(pi.sku_name) AS sku_name,
array_agg(DISTINCT pi.provider) AS providers,
array_agg(DISTINCT pi.shop_id) AS shop_ids,
array_agg(DISTINCT pi.sku_code) AS sku_codes,
array_remove(array_agg(DISTINCT pi.fulfillment_pending_reason), NULL) AS pending_reasons,
array_remove(
array_agg(
DISTINCT COALESCE(
pi.fulfillment_pending_detail->>'carrierIssue',
pi.fulfillment_pending_detail->>'reason'
)
),
NULL
) AS pending_subreasons,
grouped.group_key,
CASE
WHEN grouped.internal_sku <> '' THEN 'internal_sku'
ELSE 'source_sku'
END AS scope_type,
grouped.internal_sku,
grouped.sku_name,
grouped.providers,
grouped.shop_ids,
grouped.sku_codes,
grouped.pending_reasons,
grouped.pending_subreasons,
COALESCE(cfg.channels, ARRAY[]::text[]) AS configured_channels,
COUNT(*) FILTER (WHERE pi.fulfillment_locked)::int AS locked_item_count,
COUNT(*)::int AS item_count,
COALESCE(SUM(pi.revenue_amount_fen), 0)::bigint AS revenue_fen,
MIN(pi.created_at) AS first_seen_at,
MAX(pi.created_at) AS last_seen_at
FROM pending_items pi
grouped.locked_item_count,
grouped.item_count,
grouped.revenue_fen,
grouped.first_seen_at,
grouped.last_seen_at,
grouped.oldest_age_minutes,
grouped.retryable_item_count,
grouped.retryable_unlocked_item_count,
grouped.auto_retryable_item_count,
grouped.auto_retryable_unlocked_item_count,
grouped.config_issue_item_count,
grouped.manual_review_item_count,
grouped.after_sales_item_count,
grouped.system_issue_item_count,
COALESCE(issue_summary.issue_breakdown, '[]'::jsonb) AS issue_breakdown
FROM grouped
LEFT JOIN issue_summary ON issue_summary.group_key = grouped.group_key
LEFT JOIN LATERAL (
SELECT array_agg(DISTINCT channel.channel ORDER BY channel.channel) AS channels
FROM internal_sku_channels channel
WHERE channel.enabled = TRUE
AND canonicalize_internal_sku(channel.internal_sku) =
pi.internal_sku
grouped.internal_sku
) cfg ON TRUE
GROUP BY pi.group_key, pi.internal_sku, cfg.channels
ORDER BY item_count DESC, internal_sku
ORDER BY grouped.item_count DESC, grouped.internal_sku
LIMIT $${params.length - 1} OFFSET $${params.length}
`,
params,
@@ -1145,16 +1272,13 @@ export async function listPendingChannelGroups(input: {
`
WITH pending_items AS (
SELECT
oi.id,
o.provider,
o.shop_id,
o.shop_name,
oi.sku_code,
${INTERNAL_SKU_SQL} AS internal_sku,
${PENDING_GROUP_KEY_SQL} AS group_key
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'pending'
v.provider,
v.shop_id,
v.shop_name,
${PENDING_STATE_GROUP_KEY_SQL} AS group_key
FROM v_fulfillment_item_state v
WHERE v.fulfillment_status = 'pending'
AND ($2::text = '' OR v.issue_category = $2)
)
SELECT
group_key,
@@ -1179,7 +1303,7 @@ export async function listPendingChannelGroups(input: {
usage_count DESC,
shop_name
`,
[groupKeys],
[groupKeys, issueCategory],
)
const seen = new Set<string>()
@@ -1225,6 +1349,7 @@ export async function retryPendingCarrierGroup(input: {
internalSku: string
retryInactiveCarrier?: boolean
includeLocked?: boolean
issueCodes?: string[]
}): Promise<RetryPendingCarrierGroupResult> {
const internalSku = canonicalizeInternalSku(input.internalSku)
if (!internalSku || !containsCjk(internalSku)) {
@@ -1234,13 +1359,37 @@ export async function retryPendingCarrierGroup(input: {
})
}
const requestedIssueCodes = [
...new Set(
(input.issueCodes || [])
.map((value) => String(value || '').trim())
.filter((value): value is (typeof RETRYABLE_CARRIER_ISSUES)[number] =>
(RETRYABLE_CARRIER_ISSUES as readonly string[]).includes(value),
),
),
]
if ((input.issueCodes || []).length > 0 && requestedIssueCodes.length === 0) {
throw createHttpError('所选问题不允许重试，请刷新后重试', {
statusCode: 409,
errorCode: 'fulfillment_issue_not_retryable',
})
}
const issueCodes =
requestedIssueCodes.length > 0
? requestedIssueCodes
: input.retryInactiveCarrier === true
? [...RETRYABLE_CARRIER_ISSUES]
: [...AUTO_RETRYABLE_CARRIER_ISSUES]
const pending = await query<{ id: number }>(
`SELECT oi.id
FROM order_items oi
`SELECT oi.order_item_id AS id
FROM v_fulfillment_item_state oi
WHERE oi.fulfillment_status = 'pending'
AND ${INTERNAL_SKU_SQL} = $1
ORDER BY oi.id`,
[internalSku],
AND oi.internal_sku = $1
AND oi.issue_code = ANY($2::text[])
AND ($3::boolean OR NOT oi.fulfillment_locked)
ORDER BY oi.order_item_id`,
[internalSku, issueCodes, input.includeLocked === true],
)
const ids = pending.rows.map((row) => Number(row.id))
let workOrdersCreated = 0
@@ -1313,6 +1462,7 @@ export async function retryPendingCarrierGroup(input: {
export async function unlockPendingChannelGroup(input: {
groupKey?: string
internalSku?: string
issueCategory?: string
}): Promise<{ updated: number }> {
const scope = resolvePendingGroupScopeInput(input)
if (!scope) {
@@ -1323,20 +1473,25 @@ export async function unlockPendingChannelGroup(input: {
}
const params: unknown[] = []
const where = ["oi.fulfillment_status = 'pending'", 'oi.fulfillment_locked = TRUE']
const where = ["state.fulfillment_status = 'pending'", 'state.fulfillment_locked = TRUE']
if (scope.type === 'internal_sku') {
params.push(scope.internalSku)
where.push(`${INTERNAL_SKU_SQL} = $${params.length}`)
where.push(`state.internal_sku = $${params.length}`)
} else {
params.push(scope.provider, scope.shopId, scope.skuCode)
where.push(`o.provider = $1 AND o.shop_id = $2 AND oi.sku_code = $3`)
where.push(`state.provider = $1 AND state.shop_id = $2 AND state.sku_code = $3`)
}
const issueCategory = String(input.issueCategory || '').trim()
if ((PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategory)) {
params.push(issueCategory)
where.push(`state.issue_category = $${params.length}`)
}
const result = await query(
`UPDATE order_items oi
`UPDATE order_items item
SET fulfillment_locked = FALSE,
updated_at = NOW()
FROM orders o
WHERE o.id = oi.order_id
FROM v_fulfillment_item_state state
WHERE item.id = state.order_item_id
AND ${where.join(' AND ')}`,
params,
)
@@ -1393,6 +1548,7 @@ function resolvePendingGroupScopeInput(input: {
export async function deletePendingChannelGroup(input: {
groupKey?: string
internalSku?: string
issueCategory?: string
adminUserId: number | null
}): Promise<DeletePendingChannelGroupResult> {
const scope = resolvePendingGroupScopeInput(input)
@@ -1413,6 +1569,16 @@ export async function deletePendingChannelGroup(input: {
itemParams.push(scope.provider, scope.shopId, scope.skuCode)
itemWhere.push(`o.provider = $1 AND o.shop_id = $2 AND oi.sku_code = $3`)
}
const issueCategory = String(input.issueCategory || '').trim()
if ((PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategory)) {
itemParams.push(issueCategory)
itemWhere.push(`EXISTS (
SELECT 1
FROM v_fulfillment_item_state state
WHERE state.order_item_id = oi.id
AND state.issue_category = $${itemParams.length}
)`)
}
const items = await client.query<{ id: number; order_id: number; basis_date: string }>(
`
SELECT
@@ -1520,6 +1686,7 @@ export async function assignPendingChannelGroup(input: {
internalSku?: string
mappedInternalSku?: string
channel: string
issueCategory?: string
adminUserId: number | null
note?: string | undefined
}): Promise<{
@@ -1544,7 +1711,19 @@ export async function assignPendingChannelGroup(input: {
errorCode: 'internal_sku_mapping_required',
})
}
const updated = await query(
const updateParams: unknown[] = [scope.provider, scope.shopId, scope.skuCode, mappedInternalSku]
const issueCategory = String(input.issueCategory || '').trim()
let issueCategoryFilter = ''
if ((PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategory)) {
updateParams.push(issueCategory)
issueCategoryFilter = `AND EXISTS (
SELECT 1
FROM v_fulfillment_item_state state
WHERE state.order_item_id = oi.id
AND state.issue_category = $${updateParams.length}
)`
}
const updated = await query<{ id: number }>(
`
UPDATE order_items oi
SET spec_json = COALESCE(oi.spec_json, '{}'::jsonb)
@@ -1556,8 +1735,10 @@ export async function assignPendingChannelGroup(input: {
AND o.shop_id = $2
AND oi.sku_code = $3
AND oi.fulfillment_status = 'pending'
${issueCategoryFilter}
RETURNING oi.id
`,
[scope.provider, scope.shopId, scope.skuCode, mappedInternalSku],
updateParams,
)
if ((updated.rowCount || 0) === 0) {
throw createHttpError('待分类来源 SKU 已发生变化，请刷新后重试', {
@@ -1568,6 +1749,8 @@ export async function assignPendingChannelGroup(input: {
return assignChannelByInternalSku({
internalSku: mappedInternalSku,
channel: input.channel,
issueCategory: input.issueCategory || '',
orderItemIds: updated.rows.map((row) => Number(row.id)),
adminUserId: input.adminUserId,
note: input.note,
})
@@ -1575,6 +1758,7 @@ export async function assignPendingChannelGroup(input: {
return assignChannelByInternalSku({
internalSku: scope.internalSku,
channel: input.channel,
issueCategory: input.issueCategory || '',
adminUserId: input.adminUserId,
note: input.note,
})
@@ -1584,6 +1768,8 @@ export async function assignPendingChannelGroup(input: {
export async function assignChannelByInternalSku(input: {
internalSku: string
channel: string
issueCategory?: string
orderItemIds?: number[]
adminUserId: number | null
note?: string | undefined
createRule?: boolean | undefined
@@ -1612,19 +1798,32 @@ export async function assignChannelByInternalSku(input: {
})
}
const issueCategory = String(input.issueCategory || '').trim()
const requestedOrderItemIds = [
...new Set(
(input.orderItemIds || []).map(Number).filter((id) => Number.isInteger(id) && id > 0),
),
]
const idParams = [...requestedOrderItemIds]
const idsResult = await query<{ id: number }>(
`
SELECT oi.id
FROM orders o
JOIN order_items oi ON oi.order_id = o.id
WHERE o.id = oi.order_id
AND ${INTERNAL_SKU_SQL} = $1
SELECT state.order_item_id AS id
FROM v_fulfillment_item_state state
WHERE state.internal_sku = $1
AND (
oi.fulfillment_status = 'pending'
OR (oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = $2)
state.fulfillment_status = 'pending'
OR (
state.fulfillment_status = 'assigned'
AND state.stored_channel = $2
)
)
AND ($3::text = '' OR state.issue_category = $3)
AND (
$4::boolean = FALSE
OR state.order_item_id = ANY($5::bigint[])
)
`,
[internalSku, input.channel],
[internalSku, input.channel, issueCategory, requestedOrderItemIds.length > 0, idParams],
)
const draftProductId = await ensureDraftProductForChannel(internalSku, input.channel)
apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts
+29 -10
View File
@@ -430,6 +430,24 @@ test(
{ id: crossShopB, name: '测试店铺 B' },
],
)
const pendingClassification = await listPendingChannelGroups({
keyword: crossShopSku,
issueCategory: 'classification',
page: 1,
pageSize: 10,
})
assert.equal(pendingClassification.total, 1)
assert.equal(
pendingClassification.items[0]?.issue_breakdown[0]?.issue_code,
'missing_channel',
)
const pendingAfterSales = await listPendingChannelGroups({
keyword: crossShopSku,
issueCategory: 'after_sales',
page: 1,
pageSize: 10,
})
assert.equal(pendingAfterSales.total, 0)
const deletedPending = await deletePendingChannelGroup({
internalSku: crossShopSku,
@@ -897,14 +915,18 @@ test(
WHERE id = $1`,
[retryPendingItemId],
)
const protectedRetry = await retryPendingCarrierGroup({
internalSku: retryPendingAffiliateSku,
retryInactiveCarrier: true,
})
assert.equal(protectedRetry.affiliateTasksCreated, 0)
await assert.rejects(
() =>
retryPendingCarrierGroup({
internalSku: retryPendingAffiliateSku,
retryInactiveCarrier: true,
issueCodes: ['user_refund'],
}),
/不允许重试/,
)
const protectedRetryState = await query<{
fulfillment_status: string
fulfillment_pending_detail: Record<string, unknown>
fulfillment_pending_detail: Record<string, unknown> | null
task_count: number
}>(
`SELECT
@@ -916,10 +938,7 @@ test(
[retryPendingItemId],
)
assert.equal(protectedRetryState.rows[0]?.fulfillment_status, 'pending')
assert.equal(
protectedRetryState.rows[0]?.fulfillment_pending_detail.carrierIssue,
'carrier_task_user_refund',
)
assert.equal(protectedRetryState.rows[0]?.fulfillment_pending_detail, null)
assert.equal(protectedRetryState.rows[0]?.task_count, 1)
const voucherJobPrefix = `IT-VOUCHER-LIST-${suffix}`
apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts
+52
View File
@@ -0,0 +1,52 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const migration = readFileSync(
new URL('../../db/migrations/137_fulfillment_item_state_view.sql', import.meta.url),
'utf8',
)
const channelAssign = readFileSync(new URL('./channel-assign-service.ts', import.meta.url), 'utf8')
const auditScript = readFileSync(
new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url),
'utf8',
)
test('履约读模型按订单项拆分渠道、载体、尝试、异常和锁定原因', () => {
assert.ok(migration.includes('CREATE OR REPLACE VIEW v_fulfillment_item_state'))
assert.ok(migration.includes('current_attempt_id'))
assert.ok(migration.includes('carrier_status'))
assert.ok(migration.includes('issue_code'))
assert.ok(migration.includes("'after_sales'"))
assert.ok(migration.includes("'manual_review'"))
assert.ok(migration.includes("fulfillment_status IN ('pending', 'assigned')"))
})
test('待分类查询使用统一读模型并返回异常数量', () => {
assert.ok(channelAssign.includes('FROM v_fulfillment_item_state v'))
assert.ok(channelAssign.includes('issue_breakdown'))
assert.ok(channelAssign.includes('retryable_unlocked_item_count'))
assert.ok(channelAssign.includes('auto_retryable_unlocked_item_count'))
assert.ok(channelAssign.includes('PENDING_ISSUE_CATEGORIES'))
assert.ok(channelAssign.includes('issueCategoryInput'))
})
test('重试必须按可重试 issueCode 限定，不能按整个 SKU 处理售后和复核项', () => {
assert.ok(channelAssign.includes('RETRYABLE_CARRIER_ISSUES'))
assert.ok(channelAssign.includes('issueCodes?: string[]'))
assert.ok(channelAssign.includes('oi.issue_code = ANY($2::text[])'))
})
test('保存渠道、解除锁定、归档和重试都按当前问题分类限定作用域', () => {
assert.ok(channelAssign.includes('issueCategory?: string'))
assert.ok(channelAssign.includes('state.issue_category = $'))
assert.ok(channelAssign.includes('issueCodes?: string[]'))
assert.ok(channelAssign.includes('orderItemIds?: number[]'))
})
test('履约审计脚本覆盖状态不一致、售后队列和重复尝试', () => {
assert.ok(auditScript.includes('已分配但无有效载体'))
assert.ok(auditScript.includes('售后关闭仍进入履约队列'))
assert.ok(auditScript.includes('近 24 小时重复尝试订单项'))
assert.ok(auditScript.includes('--strict'))
})