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

fix(fulfillment): 修复载体重试并完善核销任务列表

Browse Source
This commit is contained in:
yml2213
2026-11-05 18:36:10 +08:00
parent 89a68b5734
commit e8b76b3b73
19 changed files with 945 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/repositories/task-repo.ts
+12
View File
@@ -492,10 +492,22 @@ export function buildTaskListFilters({
roleId = '',
dateFrom = '',
dateTo = '',
currentOnly = false,
}: TaskListQueryInput = {}): { whereClause: string; params: unknown[] } {
const filters: string[] = []
const params: unknown[] = []
if (currentOnly) {
filters.push(`ft.id = (
SELECT latest.id
FROM fulfillment_tasks latest
WHERE latest.order_item_id = ft.order_item_id
AND latest.unit_index = ft.unit_index
ORDER BY latest.id DESC
LIMIT 1
)`)
}
if (keyword) {
const trimmedKeyword = String(keyword).trim()
const isOrderNoLookup = isPlatformOrderIdLookupKeyword(trimmedKeyword)
apps/backend/src/routes/admin/fulfillment-channel.ts
+31
View File
@@ -20,6 +20,7 @@ import {
retryPendingCarrierGroup,
switchChannelForOrderItems,
unlockChannel,
unlockPendingChannelGroup,
upsertChannelRule,
} from '../../services/fulfillment/channel-assign-service.js'
import {
@@ -86,6 +87,9 @@ router.get(
createJsonHandler(
(req) =>
listVoucherConsumeJobs({
keyword: String(req.query.keyword || ''),
provider: String(req.query.provider || ''),
channel: String(req.query.channel || ''),
status: String(req.query.status || ''),
page: Number(req.query.page),
pageSize: Number(req.query.pageSize),
@@ -200,6 +204,8 @@ router.post(
(req) =>
retryPendingCarrierGroup({
internalSku: String(bodyOf(req).internalSku || ''),
retryInactiveCarrier: bodyOf(req).retryInactiveCarrier === true,
includeLocked: bodyOf(req).includeLocked === true,
}),
{
successMessage: '履约载体重试完成',
@@ -215,6 +221,31 @@ router.post(
),
)
router.post(
'/fulfillment/pending/unlock',
requireAdminPermission('finance.config'),
createJsonHandler(
(req) => {
const body = bodyOf(req)
return unlockPendingChannelGroup({
groupKey: String(body.groupKey || ''),
internalSku: String(body.internalSku || ''),
})
},
{
successMessage: '运营锁定已解除',
errorMessage: '解除运营锁定失败',
scope: '[admin/fulfillment/pending/unlock]',
audit: (req, data) => ({
action: 'fulfillment.unlock_pending_group',
targetType: 'pending_group',
targetId: String(bodyOf(req).groupKey || bodyOf(req).internalSku || ''),
data: data as Record<string, unknown>,
}),
},
),
)
router.get(
'/fulfillment/internal-sku-mappings',
requireAdminPermission('finance.view'),
apps/backend/src/services/admin/admin-read-service.ts
+1
View File
@@ -348,6 +348,7 @@ export async function getAdminTasks(
roleId: String(query.roleId || '').trim(),
dateFrom: normalizeDateQuery(query.dateFrom),
dateTo: normalizeDateQuery(query.dateTo, true),
currentOnly: String(query.currentOnly ?? 'true') !== 'false',
})
const viewerContext = createAdminViewerContext(session)
apps/backend/src/services/fulfillment/carrier-creation-service.ts
+57 -17
View File
@@ -54,7 +54,11 @@ async function hasAffiliateTask(orderItemId: number): Promise<boolean> {
return result.rows[0]?.exists === true
}
const BLOCKED_INACTIVE_CARRIER_STATUSES = new Set(['closed', 'cancelled', 'manual_review'])
const BLOCKED_WORK_ORDER_STATUSES = new Set(['closed', 'cancelled', 'manual_review'])
const HARD_BLOCKED_INACTIVE_REASONS = new Set([
'carrier_task_user_refund',
'carrier_task_cancelled',
])
async function findBlockingInactiveCarrier(input: {
orderItemId: number
@@ -64,16 +68,16 @@ async function findBlockingInactiveCarrier(input: {
const result =
input.channel === 'affiliate_dash'
? await query<{ status: string }>(
`SELECT task_status AS status
? await query<{ status: string; result_code: string }>(
`SELECT task_status AS status, result_code
FROM fulfillment_tasks
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'
ORDER BY id DESC
LIMIT 1`,
[input.orderItemId],
)
: await query<{ status: string }>(
`SELECT status
: await query<{ status: string; result_code: string }>(
`SELECT status, '' AS result_code
FROM work_orders
WHERE order_item_id = $1
ORDER BY id DESC
@@ -82,7 +86,14 @@ async function findBlockingInactiveCarrier(input: {
)
const status = String(result.rows[0]?.status || '')
if (!BLOCKED_INACTIVE_CARRIER_STATUSES.has(status)) return null
const resultCode = String(result.rows[0]?.result_code || '')
const shouldBlock =
input.channel === 'affiliate_dash'
? INACTIVE_CARRIER_TASK_STATUSES.includes(
status as (typeof INACTIVE_CARRIER_TASK_STATUSES)[number],
)
: BLOCKED_WORK_ORDER_STATUSES.has(status)
if (!shouldBlock) return null
if (status === 'manual_review') {
return {
@@ -91,6 +102,34 @@ async function findBlockingInactiveCarrier(input: {
message: '已有履约载体进入人工复核，需先处理后再重试',
}
}
if (status === 'retry_pending' || status === 'failed') {
return {
status,
reason: 'carrier_task_retry_pending',
message: '已有履约任务失败待重试，需由运营确认后重新创建',
}
}
if (status === 'expired') {
return {
status,
reason: 'carrier_task_expired',
message: '已有履约任务已过期，需由运营确认后重新创建',
}
}
if (resultCode === 'USER_APPLY_REFUND') {
return {
status,
reason: 'carrier_task_user_refund',
message: '用户已申请退款，不能自动重新发货',
}
}
if (resultCode === 'affiliate_dash_cancelled') {
return {
status,
reason: 'carrier_task_cancelled',
message: '上游履约订单已取消，不能自动重新发货',
}
}
return {
status,
reason: 'carrier_task_closed',
@@ -101,6 +140,14 @@ async function findBlockingInactiveCarrier(input: {
}
}
function shouldBlockInactiveCarrier(
inactiveCarrier: { reason: string } | null,
retryInactiveCarrier: boolean,
): inactiveCarrier is { status: string; reason: string; message: string } {
if (!inactiveCarrier) return false
return !retryInactiveCarrier || HARD_BLOCKED_INACTIVE_REASONS.has(inactiveCarrier.reason)
}
function toResolvedItem(item: OrderItemRow): ResolvedFulfillmentItem {
return {
...item,
@@ -127,15 +174,8 @@ export async function ensureCarrierForOrderItem(input: {
item: OrderItemRow
channel: FulfillmentChannel
retryInactiveCarrier?: boolean
allowInactiveTaskRetry?: boolean
}): Promise<EnsureCarrierResult> {
const {
order,
item,
channel,
retryInactiveCarrier = false,
allowInactiveTaskRetry = false,
} = input
const { order, item, channel, retryInactiveCarrier = false } = input
const existingCarrier = findConflictingCarrier(
await loadActiveCarrierEvidenceForItem(Number(item.id)),
channel,
@@ -194,7 +234,7 @@ export async function ensureCarrierForOrderItem(input: {
orderItemId: Number(item.id),
channel,
})
if (inactiveCarrier && !retryInactiveCarrier) {
if (shouldBlockInactiveCarrier(inactiveCarrier, retryInactiveCarrier)) {
return {
ready: false,
created: false,
@@ -274,7 +314,7 @@ export async function ensureCarrierForOrderItem(input: {
orderItemId: Number(item.id),
channel,
})
if (inactiveCarrier && !retryInactiveCarrier) {
if (shouldBlockInactiveCarrier(inactiveCarrier, retryInactiveCarrier)) {
return {
ready: false,
created: false,
@@ -326,7 +366,7 @@ export async function ensureCarrierForOrderItem(input: {
item_snapshot_json: snapshot,
} as OrderItemRow,
],
{ allowInactiveTaskRetry: retryInactiveCarrier || allowInactiveTaskRetry },
{ allowInactiveTaskRetry: retryInactiveCarrier },
)
}
const ready = await hasAffiliateTask(Number(item.id))
apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts
+47 -2
View File
@@ -76,7 +76,14 @@ async function loadCandidateItems(input: {
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
AND ft.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')})
)`
)
AND COALESCE((
SELECT ft.result_code
FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
ORDER BY ft.id DESC
LIMIT 1
), '') NOT IN ('USER_APPLY_REFUND', 'affiliate_dash_cancelled')`
: `oi.fulfillment_status = 'pending'
${input.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}`
@@ -150,6 +157,38 @@ async function markPendingWithoutCarrier(input: {
)
}
async function markAssignedMissingCarrierPending(input: {
itemId: number
previousChannel: string
internalSku: string
attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }>
}): Promise<void> {
const carrierIssue = input.attempts[0]?.reason || 'carrier_attempt_failed'
await query(
`
UPDATE order_items
SET fulfillment_channel = NULL,
fulfillment_status = 'pending',
fulfillment_source = 'auto_mapping',
fulfillment_locked = FALSE,
fulfillment_pending_reason = 'no_carrier',
fulfillment_pending_detail = $2::jsonb,
updated_at = NOW()
WHERE id = $1 AND fulfillment_status = 'assigned'
`,
[
input.itemId,
JSON.stringify({
internalSku: input.internalSku,
previousChannel: input.previousChannel,
carrierIssue,
carrierMessage: input.attempts[0]?.message || '',
attempts: input.attempts,
}),
],
)
}
async function selectSafeFallback(input: {
orderItemId: number
channel: FulfillmentChannel
@@ -299,7 +338,6 @@ export async function reconcilePendingCarriers(
item,
channel: attempt.channel,
retryInactiveCarrier: input.retryInactiveCarrier === true,
allowInactiveTaskRetry: true,
})
if (!carrier.ready) {
failures.push({
@@ -361,6 +399,13 @@ export async function reconcilePendingCarriers(
internalSku: candidate.internal_sku,
attempts: failures,
})
} else {
await markAssignedMissingCarrierPending({
itemId: Number(item.id),
previousChannel: assignedChannel,
internalSku: candidate.internal_sku,
attempts: failures,
})
}
}
}
apps/backend/src/services/fulfillment/channel-assign-service.ts
+45 -5
View File
@@ -511,7 +511,6 @@ export async function assignOrderItemsAfterCarrierReady(input: {
item,
channel: input.channel,
retryInactiveCarrier: true,
allowInactiveTaskRetry: true,
})
} catch (error) {
carrier = {
@@ -1216,9 +1215,16 @@ export type RetryPendingCarrierGroupResult = {
remaining: number
}
/** 运营确认后重试整组载体：允许重开已关闭的历史尝试，并包含运营锁定的订单项。 */
/**
* 运营确认后重试整组载体。
*
* 默认只处理未锁定且没有历史终态任务的订单；重开旧任务必须显式传入
* retryInactiveCarrier。锁定项需要先解除锁定，避免一次操作误伤整组。
*/
export async function retryPendingCarrierGroup(input: {
internalSku: string
retryInactiveCarrier?: boolean
includeLocked?: boolean
}): Promise<RetryPendingCarrierGroupResult> {
const internalSku = canonicalizeInternalSku(input.internalSku)
if (!internalSku || !containsCjk(internalSku)) {
@@ -1247,8 +1253,8 @@ export async function retryPendingCarrierGroup(input: {
const carrier = await reconcilePendingCarriers({
orderItemIds: batchIds,
limit: batchIds.length,
includeLocked: true,
retryInactiveCarrier: true,
includeLocked: input.includeLocked === true,
retryInactiveCarrier: input.retryInactiveCarrier === true,
})
workOrdersCreated += carrier.workOrdersCreated
affiliateTasksCreated += carrier.affiliateTasksCreated
@@ -1258,7 +1264,7 @@ export async function retryPendingCarrierGroup(input: {
batchSize: batchIds.length,
maxBatches: 1,
afterId: 0,
includeLocked: true,
includeLocked: input.includeLocked === true,
})
assignedItemIds.push(...applied.assignedItemIds)
}
@@ -1303,6 +1309,40 @@ export async function retryPendingCarrierGroup(input: {
}
}
/** 解除一个待分类分组中的运营锁定，让自动路由或后续修复重新接管。 */
export async function unlockPendingChannelGroup(input: {
groupKey?: string
internalSku?: string
}): Promise<{ updated: number }> {
const scope = resolvePendingGroupScopeInput(input)
if (!scope) {
throw createHttpError('缺少待分类队列分组标识', {
statusCode: 400,
errorCode: 'pending_group_key_required',
})
}
const params: unknown[] = []
const where = ["oi.fulfillment_status = 'pending'", 'oi.fulfillment_locked = TRUE']
if (scope.type === 'internal_sku') {
params.push(scope.internalSku)
where.push(`${INTERNAL_SKU_SQL} = $${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`)
}
const result = await query(
`UPDATE order_items oi
SET fulfillment_locked = FALSE,
updated_at = NOW()
FROM orders o
WHERE o.id = oi.order_id
AND ${where.join(' AND ')}`,
params,
)
return { updated: result.rowCount || 0 }
}
export type DeletePendingChannelGroupResult = {
archivedItems: number
preservedOrders: number
apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts
+43
View File
@@ -0,0 +1,43 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const carrierCreation = readFileSync(
new URL('./carrier-creation-service.ts', import.meta.url),
'utf8',
)
const carrierReconciliation = readFileSync(
new URL('./carrier-reconciliation-service.ts', import.meta.url),
'utf8',
)
const channelAssign = readFileSync(new URL('./channel-assign-service.ts', import.meta.url), 'utf8')
const taskRepo = readFileSync(new URL('../../repositories/task-repo.ts', import.meta.url), 'utf8')
test('自动发货历史非活跃任务默认阻止后台自动重开', () => {
assert.ok(carrierCreation.includes('INACTIVE_CARRIER_TASK_STATUSES.includes'))
assert.ok(carrierCreation.includes("reason: 'carrier_task_retry_pending'"))
assert.ok(carrierCreation.includes('HARD_BLOCKED_INACTIVE_REASONS'))
assert.ok(carrierCreation.includes("'carrier_task_user_refund'"))
assert.ok(!carrierReconciliation.includes('allowInactiveTaskRetry: true'))
})
test('已分配但载体失效的订单会退回 pending/no_carrier', () => {
assert.ok(carrierReconciliation.includes('markAssignedMissingCarrierPending'))
assert.ok(carrierReconciliation.includes("fulfillment_status = 'pending'"))
assert.ok(carrierReconciliation.includes("fulfillment_pending_reason = 'no_carrier'"))
assert.ok(carrierReconciliation.includes("'USER_APPLY_REFUND', 'affiliate_dash_cancelled'"))
})
test('整组补载体默认不包含锁定项，也不强制重开历史任务', () => {
assert.ok(channelAssign.includes('retryInactiveCarrier?: boolean'))
assert.ok(channelAssign.includes('includeLocked?: boolean'))
assert.ok(channelAssign.includes('retryInactiveCarrier: input.retryInactiveCarrier === true'))
assert.ok(channelAssign.includes('includeLocked: input.includeLocked === true'))
assert.ok(channelAssign.includes('unlockPendingChannelGroup'))
})
test('任务列表默认只展示每个订单项单元的最新尝试', () => {
assert.ok(taskRepo.includes('currentOnly = false'))
assert.ok(taskRepo.includes('SELECT latest.id'))
assert.ok(taskRepo.includes('latest.unit_index = ft.unit_index'))
})
apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts
+207
View File
@@ -29,11 +29,13 @@ test(
listPendingChannelGroups,
retryPendingCarrierGroup,
switchChannelForOrderItems,
unlockPendingChannelGroup,
upsertChannelRule,
} = await import('./channel-assign-service.js')
const { deleteManualSkuClassification, saveManualFulfillmentConfig } =
await import('../cost-center/manual-fulfillment-config-service.js')
const { runAffiliateFallback } = await import('./affiliate-fallback-service.js')
const { reconcilePendingCarriers } = await import('./carrier-reconciliation-service.js')
const { retryPendingWorkOrderRouting } = await import('./pending-channel-repair-service.js')
const {
deleteInternalSkuMapping,
@@ -43,6 +45,8 @@ test(
} = await import('./internal-sku-mapping-service.js')
const { resolveOrderItemForFulfillment } = await import('./product-resolution-service.js')
const { updateOrderItemSourceSnapshot } = await import('../../repositories/order-item-repo.js')
const { listTasks } = await import('../../repositories/task-repo.js')
const { listVoucherConsumeJobs } = await import('./voucher-consume-queue-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const shopId = `it-routing-${suffix}`
const manualSku = `集成测试人工-${suffix}`
@@ -56,6 +60,7 @@ test(
const protectedRuleSku = `集成测试强制删除-${suffix}`
const missingVoucherSku = `集成测试无商品卡券-${suffix}`
const reactivatedVoucherSku = `集成测试归档卡券-${suffix}`
const retryPendingAffiliateSku = `集成测试失败待重试-${suffix}`
const retryWorkOrderRuleKey = `it-retry-${suffix}`
const orderIds: number[] = []
const orderItemIds: number[] = []
@@ -73,6 +78,7 @@ test(
protectedRuleSku,
missingVoucherSku,
reactivatedVoucherSku,
retryPendingAffiliateSku,
]
const createdVoucherProductIds: number[] = []
let autoManualProductId: number | null = null
@@ -383,6 +389,7 @@ test(
)
const retriedCarrier = await retryPendingCarrierGroup({
internalSku: retryWorkOrderSku,
retryInactiveCarrier: true,
})
assert.equal(retriedCarrier.scanned, 1)
assert.equal(retriedCarrier.assigned, 1)
@@ -645,6 +652,21 @@ test(
retryTaskRows.rows.map((row) => row.task_status),
['closed', 'paid'],
)
const currentTasksOnly = await listTasks({
platformOrderId: retryTaskOrderRow.platform_order_id,
currentOnly: true,
page: 1,
pageSize: 20,
})
assert.equal(currentTasksOnly.total, 1)
assert.equal(currentTasksOnly.items[0]?.task_status, 'paid')
const allTaskAttempts = await listTasks({
platformOrderId: retryTaskOrderRow.platform_order_id,
currentOnly: false,
page: 1,
pageSize: 20,
})
assert.equal(allTaskAttempts.total, 2)
assert.equal(await ensureVoucherRuleForSku(missingVoucherSku), false)
const missingVoucherRules = await query<{ count: number }>(
@@ -772,6 +794,188 @@ test(
assert.equal(reactivated.rows[0]?.is_fallback, false)
assert.equal(reactivated.rows[0]?.attributes_json.deletedAt, undefined)
assert.equal(typeof reactivated.rows[0]?.attributes_json.reactivatedAt, 'string')
const retryPendingItemId = await createPendingOrderItem(retryPendingAffiliateSku)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES (
$1, 'affiliate_dash', 10, FALSE, TRUE,
jsonb_build_object('affiliateSku', $1::text), 'integration_test'
)`,
[retryPendingAffiliateSku],
)
await query(
`UPDATE order_items
SET fulfillment_status = 'assigned',
fulfillment_channel = 'affiliate_dash',
fulfillment_source = 'carrier_evidence',
updated_at = NOW()
WHERE id = $1`,
[retryPendingItemId],
)
await query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, created_at, updated_at
)
SELECT oi.order_id, oi.id, 1, $2, o.provider, o.platform,
o.shop_id, o.shop_name, o.platform_order_id, profile.id, 'affiliate_dash',
'retry_pending', NOW(), NOW()
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
JOIN LATERAL (
SELECT id FROM fulfillment_profiles
WHERE profile_key = 'affiliate_dash' AND executor_key = 'affiliate_dash'
ORDER BY id LIMIT 1
) profile ON TRUE
WHERE oi.id = $1`,
[retryPendingItemId, `IT-RETRY-PENDING-${suffix}`],
)
const blockedRetry = await reconcilePendingCarriers({
orderItemIds: [retryPendingItemId],
includeAssignedMissingCarrier: true,
limit: 1,
})
assert.equal(blockedRetry.affiliateTasksCreated, 0)
const retryPendingState = await query<{
fulfillment_status: string
fulfillment_channel: string | null
fulfillment_pending_reason: string | null
fulfillment_pending_detail: Record<string, unknown>
task_count: number
}>(
`SELECT
oi.fulfillment_status,
oi.fulfillment_channel,
oi.fulfillment_pending_reason,
oi.fulfillment_pending_detail,
(SELECT count(*)::int FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) AS task_count
FROM order_items oi
WHERE oi.id = $1`,
[retryPendingItemId],
)
assert.equal(retryPendingState.rows[0]?.fulfillment_status, 'pending')
assert.equal(retryPendingState.rows[0]?.fulfillment_channel, null)
assert.equal(retryPendingState.rows[0]?.fulfillment_pending_reason, 'no_carrier')
assert.equal(
retryPendingState.rows[0]?.fulfillment_pending_detail.carrierIssue,
'carrier_task_retry_pending',
)
assert.equal(retryPendingState.rows[0]?.task_count, 1)
await query(
`UPDATE order_items SET fulfillment_locked = TRUE, updated_at = NOW() WHERE id = $1`,
[retryPendingItemId],
)
const unlockedPendingGroup = await unlockPendingChannelGroup({
internalSku: retryPendingAffiliateSku,
})
assert.equal(unlockedPendingGroup.updated, 1)
const unlockedPendingItem = await query<{ fulfillment_locked: boolean }>(
`SELECT fulfillment_locked FROM order_items WHERE id = $1`,
[retryPendingItemId],
)
assert.equal(unlockedPendingItem.rows[0]?.fulfillment_locked, false)
await query(
`UPDATE fulfillment_tasks
SET task_status = 'closed',
result_code = 'USER_APPLY_REFUND',
updated_at = NOW()
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'`,
[retryPendingItemId],
)
await query(
`UPDATE order_items
SET fulfillment_status = 'pending',
fulfillment_channel = NULL,
fulfillment_source = 'auto_mapping',
fulfillment_pending_reason = NULL,
fulfillment_pending_detail = NULL,
updated_at = NOW()
WHERE id = $1`,
[retryPendingItemId],
)
const protectedRetry = await retryPendingCarrierGroup({
internalSku: retryPendingAffiliateSku,
retryInactiveCarrier: true,
})
assert.equal(protectedRetry.affiliateTasksCreated, 0)
const protectedRetryState = await query<{
fulfillment_status: string
fulfillment_pending_detail: Record<string, unknown>
task_count: number
}>(
`SELECT
oi.fulfillment_status,
oi.fulfillment_pending_detail,
(SELECT count(*)::int FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) AS task_count
FROM order_items oi
WHERE oi.id = $1`,
[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]?.task_count, 1)
const voucherJobPrefix = `IT-VOUCHER-LIST-${suffix}`
const voucherJobs = await query<{ id: number }>(
`INSERT INTO voucher_consume_jobs (
provider, channel, trigger_event, target_key, platform_order_id,
due_at, status, attempt_count, max_attempts, updated_at
) VALUES
('kuaishou_industry', 'affiliate_dash', 'order_received', $1, $2, NOW(), 'succeeded', 1, 8, NOW()),
('kuaishou_industry', 'manual', 'order_received', $3, $4, NOW(), 'failed', 2, 8, NOW()),
('douyin_mashida', 'manual', 'order_received', $5, $6, NOW(), 'failed', 1, 8, NOW())
RETURNING id`,
[
`${voucherJobPrefix}-KS-AFF`,
`${voucherJobPrefix}-KS-AFF-ORDER`,
`${voucherJobPrefix}-KS-MANUAL`,
`${voucherJobPrefix}-KS-MANUAL-ORDER`,
`${voucherJobPrefix}-DY-MANUAL`,
`${voucherJobPrefix}-DY-MANUAL-ORDER`,
],
)
const firstVoucherJobPage = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
page: 1,
pageSize: 2,
})
assert.equal(firstVoucherJobPage.total, 3)
assert.equal(firstVoucherJobPage.items.length, 2)
assert.equal(firstVoucherJobPage.page, 1)
assert.equal(firstVoucherJobPage.pageSize, 2)
const secondVoucherJobPage = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
page: 2,
pageSize: 2,
})
assert.equal(secondVoucherJobPage.total, 3)
assert.equal(secondVoucherJobPage.items.length, 1)
const filteredVoucherJobs = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
provider: 'kuaishou_industry',
channel: 'manual',
status: 'failed',
page: 1,
pageSize: 20,
})
assert.equal(filteredVoucherJobs.total, 1)
assert.equal(filteredVoucherJobs.items[0]?.provider, 'kuaishou_industry')
assert.equal(filteredVoucherJobs.items[0]?.channel, 'manual')
assert.equal(filteredVoucherJobs.items[0]?.status, 'failed')
const searchedVoucherJob = await listVoucherConsumeJobs({
keyword: String(voucherJobs.rows[1]?.id || ''),
status: 'failed',
page: 1,
pageSize: 20,
})
assert.ok(searchedVoucherJob.items.some((job) => job.id === voucherJobs.rows[1]?.id))
} finally {
if (orderItemIds.length > 0) {
await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [
@@ -795,6 +999,9 @@ test(
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 voucher_consume_jobs WHERE platform_order_id LIKE $1`, [
`IT-VOUCHER-LIST-${suffix}%`,
])
await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = ANY($1::text[])`, [
internalSkus,
])
apps/backend/src/services/fulfillment/voucher-consume-job-list.test.ts
+24
View File
@@ -0,0 +1,24 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const source = readFileSync(new URL('./voucher-consume-queue-service.ts', import.meta.url), 'utf8')
test('核销任务列表支持关键词、平台、渠道和状态组合筛选', () => {
assert.ok(source.includes('keyword?: string'))
assert.ok(source.includes('provider?: string'))
assert.ok(source.includes('channel?: string'))
assert.ok(source.includes('status?: string'))
assert.ok(source.includes('provider = $'))
assert.ok(source.includes('channel = $'))
assert.ok(source.includes('status = $'))
assert.ok(source.includes('platform_order_id ILIKE'))
assert.ok(source.includes('target_key ILIKE'))
})
test('核销任务列表使用服务端总数和分页参数', () => {
assert.ok(source.includes('SELECT COUNT(*) AS total FROM voucher_consume_jobs'))
assert.ok(source.includes('LIMIT $'))
assert.ok(source.includes('OFFSET $'))
assert.ok(source.includes('ORDER BY updated_at DESC, id DESC'))
})
apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts
+46 -1
View File
@@ -22,6 +22,8 @@ import { attachKuaishouIndustryVoucherToTask } from '../platforms/kuaishou-indus
import { verifyCertificate } from '../redeem/certificate-service.js'
import { logIntegration, logWarn } from '../../utils/logger.js'
import {
VOUCHER_CONSUME_CHANNELS,
VOUCHER_CONSUME_PROVIDERS,
getVoucherConsumePolicy,
resolveVoucherConsumePolicy,
type VoucherConsumeChannel,
@@ -99,6 +101,13 @@ type VoucherOrderItemTargetRow = {
const RETRY_DELAYS_SECONDS = [10, 30, 60, 300, 600, 900, 1800]
const PROCESSING_LOCK_SECONDS = 120
const VOUCHER_CONSUME_JOB_STATUSES = [
'pending',
'processing',
'succeeded',
'failed',
'cancelled',
] as const
export async function scheduleVoucherConsumeForTask(
task: TaskRow,
@@ -443,6 +452,9 @@ export async function runVoucherConsumeDispatchJob(
export async function listVoucherConsumeJobs(
input: {
keyword?: string
provider?: string
channel?: string
status?: string
page?: number
pageSize?: number
@@ -450,10 +462,43 @@ export async function listVoucherConsumeJobs(
) {
const page = clampInteger(input.page, 1, 1, Number.MAX_SAFE_INTEGER)
const pageSize = clampInteger(input.pageSize, 20, 1, 100)
const status = String(input.status || '').trim()
const keyword = String(input.keyword || '')
.trim()
.slice(0, 100)
const providerInput = String(input.provider || '').trim()
const provider = (VOUCHER_CONSUME_PROVIDERS as readonly string[]).includes(providerInput)
? providerInput
: ''
const channel = (VOUCHER_CONSUME_CHANNELS as readonly string[]).includes(
String(input.channel || '').trim(),
)
? String(input.channel).trim()
: ''
const status = (VOUCHER_CONSUME_JOB_STATUSES as readonly string[]).includes(
String(input.status || '').trim(),
)
? String(input.status).trim()
: ''
const params: unknown[] = []
const where: string[] = []
if (keyword) {
params.push(`%${keyword}%`)
where.push(
`(CAST(id AS TEXT) ILIKE $${params.length}
OR platform_order_id ILIKE $${params.length}
OR target_key ILIKE $${params.length}
OR CAST(order_id AS TEXT) ILIKE $${params.length})`,
)
}
if (provider) {
params.push(provider)
where.push(`provider = $${params.length}`)
}
if (channel) {
params.push(channel)
where.push(`channel = $${params.length}`)
}
if (status) {
params.push(status)
where.push(`status = $${params.length}`)
apps/backend/src/types/admin/read-inputs.ts
+1
View File
@@ -32,6 +32,7 @@ export type AdminTaskListQueryInput = {
roleId?: string
dateFrom?: string
dateTo?: string
currentOnly?: boolean | string
}
export type AdminEntityIdInput = number | string
apps/backend/src/types/repository/inputs.ts
+1
View File
@@ -62,6 +62,7 @@ export type TaskListQueryInput = {
roleId?: string
dateFrom?: string
dateTo?: string
currentOnly?: boolean
}
export type TaskRuntimeContextPatch = {