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:25:48 +08:00
parent ed450fac79
commit ca214854bb
14 changed files with 340 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/config/app-config-keys.ts
+2
View File
@@ -17,4 +17,6 @@ export const APP_CONFIG_KEYS = {
pddSource: 'pdd_source',
/** 抖音码速达（抖店中转开放平台）接入配置：uid/secret/baseUrl/certLink 等 */
douyinMashida: 'douyin_mashida',
/** 履约自动化生效边界：此时间之前的历史订单不再参与自动判定和自动重试 */
fulfillmentAutomationCutoff: 'fulfillment_automation_cutoff',
} as const
apps/backend/src/db/migrations/138_fulfillment_current_attempt.sql
+54
View File
@@ -0,0 +1,54 @@
-- 138_fulfillment_current_attempt.sql
--
-- 履约尝试指针：
-- 1. 每个订单项单元显式记录尝试序号；
-- 2. 同一订单项单元只允许一个当前尝试；
-- 3. 新尝试创建时自动把旧尝试标为非当前，历史仍完整保留。
ALTER TABLE fulfillment_tasks
ADD COLUMN IF NOT EXISTS attempt_no INTEGER NOT NULL DEFAULT 1,
ADD COLUMN IF NOT EXISTS is_current BOOLEAN NOT NULL DEFAULT TRUE;
WITH ranked AS (
SELECT
id,
order_item_id,
unit_index,
ROW_NUMBER() OVER (
PARTITION BY order_item_id, unit_index
ORDER BY id
)::int AS attempt_no,
ROW_NUMBER() OVER (
PARTITION BY order_item_id, unit_index
ORDER BY id DESC
) AS current_rank
FROM fulfillment_tasks
),
duplicate_units AS (
SELECT order_item_id, unit_index
FROM fulfillment_tasks
GROUP BY order_item_id, unit_index
HAVING COUNT(*) > 1
)
UPDATE fulfillment_tasks task
SET attempt_no = ranked.attempt_no,
is_current = ranked.current_rank = 1
FROM ranked
JOIN duplicate_units duplicate
ON duplicate.order_item_id = ranked.order_item_id
AND duplicate.unit_index = ranked.unit_index
WHERE task.id = ranked.id
;
CREATE UNIQUE INDEX IF NOT EXISTS uq_fulfillment_tasks_current_unit
ON fulfillment_tasks (order_item_id, unit_index)
WHERE is_current = TRUE;
CREATE INDEX IF NOT EXISTS idx_fulfillment_tasks_attempt_history
ON fulfillment_tasks (order_item_id, unit_index, attempt_no DESC, id DESC);
COMMENT ON COLUMN fulfillment_tasks.attempt_no IS
'同一订单项单元的履约尝试序号，从 1 开始递增';
COMMENT ON COLUMN fulfillment_tasks.is_current IS
'是否为该订单项单元当前尝试；历史尝试保留但不再标记为当前';
apps/backend/src/db/migrations/139_fulfillment_automation_cutoff.sql
+30
View File
@@ -0,0 +1,30 @@
-- 139_fulfillment_automation_cutoff.sql
--
-- 历史履约数据已完成人工处置，不再参与自动判定、自动补载体或自动重试。
-- 从本迁移执行时间起，只自动处理之后新建的订单。
INSERT INTO app_config_entries (
config_key,
config_json,
source_file_path,
migrated_from_file,
created_at,
updated_at
) VALUES (
'fulfillment_automation_cutoff',
jsonb_build_object(
'cutoff',
to_char(
NOW() AT TIME ZONE 'Asia/Shanghai',
'YYYY-MM-DD"T"HH24:MI:SS.MS+08:00'
)
),
'',
FALSE,
NOW(),
NOW()
)
ON CONFLICT (config_key) DO NOTHING;
COMMENT ON TABLE app_config_entries IS
'运行时配置中心，保存从 JSON 文件迁移来的平台和系统配置；fulfillment_automation_cutoff 是履约自动化生效边界';
apps/backend/src/repositories/task-repo.ts
+18 -10
View File
@@ -94,6 +94,18 @@ export async function listTasksByOrderIds(orderIds: number[]): Promise<TaskRow[]
export async function createTask(input: TaskCreateInput): Promise<TaskRow | null> {
return withTransaction(async (client) => {
const nextAttempt = await client.query<{ attempt_no: number }>(
`SELECT COALESCE(MAX(attempt_no), 0) + 1 AS attempt_no
FROM fulfillment_tasks
WHERE order_item_id = $1 AND unit_index = $2`,
[input.orderItemId, input.unitIndex],
)
await client.query(
`UPDATE fulfillment_tasks
SET is_current = FALSE
WHERE order_item_id = $1 AND unit_index = $2 AND is_current = TRUE`,
[input.orderItemId, input.unitIndex],
)
const taskResult = await client.query<{ id: number }>(
`
INSERT INTO fulfillment_tasks (
@@ -121,11 +133,13 @@ export async function createTask(input: TaskCreateInput): Promise<TaskRow | null
last_error,
context_json,
created_at,
updated_at
updated_at,
attempt_no,
is_current
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
$11, $12, $13, $14, $15, $16, $17, $18, $19,
$20, $21, $22, $23::jsonb, $24, $25
$20, $21, $22, $23::jsonb, $24, $25, $26, TRUE
)
RETURNING id
`,
@@ -155,6 +169,7 @@ export async function createTask(input: TaskCreateInput): Promise<TaskRow | null
input.contextJson || '{}',
input.createdAt,
input.updatedAt,
Number(nextAttempt.rows[0]?.attempt_no || 1),
],
)
@@ -498,14 +513,7 @@ export function buildTaskListFilters({
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
)`)
filters.push('ft.is_current = TRUE')
}
if (keyword) {
apps/backend/src/services/fulfillment/affiliate-fallback-service.ts
+10 -2
View File
@@ -22,9 +22,16 @@ export const AFFILIATE_FAILURE_STATUSES = [
const AFFILIATE_SUCCESS_STATUSES = ['redeemed', 'paid'] as const
export async function runAffiliateFallback(
options: { limit?: number } = {},
options: { limit?: number; cutoffCreatedAt?: string } = {},
): Promise<{ scanned: number; switched: number }> {
const limit = Math.min(1000, Math.max(1, Math.trunc(options.limit || 200)))
const params: unknown[] = [AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit]
const cutoffCreatedAt = String(options.cutoffCreatedAt || '').trim()
let cutoffFilter = ''
if (cutoffCreatedAt && Number.isFinite(Date.parse(cutoffCreatedAt))) {
params.push(cutoffCreatedAt)
cutoffFilter = `AND o.created_at >= $${params.length}::timestamptz`
}
const candidates = await query<{
id: number
order_item_id: number
@@ -80,10 +87,11 @@ export async function runAffiliateFallback(
AND ft.executor_key = 'affiliate_dash'
AND ft.task_status = ANY($2::text[])
)
${cutoffFilter}
ORDER BY oi.id
LIMIT $3
`,
[AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit],
params,
)
const switchedIds: number[] = []
apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts
+10
View File
@@ -52,6 +52,7 @@ async function loadCandidateItems(input: {
assignedMissingCarrier?: boolean
afterId?: number
includeLocked?: boolean
cutoffCreatedAt?: string
}): Promise<CarrierCandidate[]> {
const ids = (input.orderItemIds || [])
.map(Number)
@@ -65,6 +66,12 @@ async function loadCandidateItems(input: {
params.push(Math.max(0, Math.trunc(Number(input.afterId))))
idFilter = `AND oi.id > $${params.length}`
}
const cutoffCreatedAt = String(input.cutoffCreatedAt || '').trim()
let cutoffFilter = ''
if (cutoffCreatedAt && Number.isFinite(Date.parse(cutoffCreatedAt))) {
params.push(cutoffCreatedAt)
cutoffFilter = `AND oi.created_at >= $${params.length}::timestamptz`
}
const statusFilter = input.assignedMissingCarrier
? `oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
@@ -123,6 +130,7 @@ async function loadCandidateItems(input: {
) capability ON jsonb_array_length(capability.channels) > 0
WHERE ${statusFilter}
${idFilter}
${cutoffFilter}
ORDER BY oi.id
LIMIT $1
`,
@@ -252,6 +260,7 @@ export async function reconcilePendingCarriers(
afterId?: number
includeLocked?: boolean
retryInactiveCarrier?: boolean
cutoffCreatedAt?: string
} = {},
): Promise<{
scanned: number
@@ -268,6 +277,7 @@ export async function reconcilePendingCarriers(
assignedMissingCarrier: input.includeAssignedMissingCarrier === true,
...(input.afterId != null ? { afterId: input.afterId } : {}),
...(input.includeLocked ? { includeLocked: true } : {}),
...(input.cutoffCreatedAt ? { cutoffCreatedAt: input.cutoffCreatedAt } : {}),
...(input.orderItemIds ? { orderItemIds: input.orderItemIds } : {}),
})
apps/backend/src/services/fulfillment/channel-assign-service.ts
+8
View File
@@ -254,6 +254,7 @@ export async function assignPendingOrderItemChannels(
afterId?: number
orderItemIds?: number[]
includeLocked?: boolean
cutoffCreatedAt?: string
} = {},
): Promise<AssignChannelsResult> {
const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500)))
@@ -288,6 +289,12 @@ export async function assignPendingOrderItemChannels(
params.push(requestedIds)
requestedFilter = `AND oi.id = ANY($${params.length}::bigint[])`
}
const cutoffCreatedAt = String(options.cutoffCreatedAt || '').trim()
let cutoffFilter = ''
if (cutoffCreatedAt && Number.isFinite(Date.parse(cutoffCreatedAt))) {
params.push(cutoffCreatedAt)
cutoffFilter = `AND o.created_at >= $${params.length}::timestamptz`
}
for (let batch = 0; batch < maxBatches; batch += 1) {
params[0] = cursor
@@ -323,6 +330,7 @@ export async function assignPendingOrderItemChannels(
${options.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}
AND oi.id > $1
${requestedFilter}
${cutoffFilter}
ORDER BY oi.id
LIMIT $2
`,
apps/backend/src/services/fulfillment/fulfillment-automation-config-service.ts
+30
View File
@@ -0,0 +1,30 @@
import { query } from '../../db/client.js'
import { APP_CONFIG_KEYS } from '../../config/app-config-keys.js'
type FulfillmentAutomationConfig = {
cutoff: string
}
export async function getFulfillmentAutomationCutoff(): Promise<string> {
const result = await query<{ config_json: FulfillmentAutomationConfig | null }>(
`SELECT config_json
FROM app_config_entries
WHERE config_key = $1
LIMIT 1`,
[APP_CONFIG_KEYS.fulfillmentAutomationCutoff],
)
const cutoff = String(result.rows[0]?.config_json?.cutoff || '').trim()
if (!cutoff || !Number.isFinite(Date.parse(cutoff))) return ''
return cutoff
}
export function fulfillmentCreatedAfterSql(
params: unknown[],
cutoff: string,
alias = 'oi',
): string {
const normalized = String(cutoff || '').trim()
if (!normalized || !Number.isFinite(Date.parse(normalized))) return ''
params.push(normalized)
return `AND ${alias}.created_at >= $${params.length}::timestamptz`
}
apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts
+14 -2
View File
@@ -12,6 +12,10 @@ const carrierReconciliation = readFileSync(
)
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')
const currentAttemptMigration = readFileSync(
new URL('../../db/migrations/138_fulfillment_current_attempt.sql', import.meta.url),
'utf8',
)
test('自动发货历史非活跃任务默认阻止后台自动重开', () => {
assert.ok(carrierCreation.includes('INACTIVE_CARRIER_TASK_STATUSES.includes'))
@@ -38,6 +42,14 @@ test('整组补载体默认不包含锁定项，也不强制重开历史任务',
test('任务列表默认只展示每个订单项单元的最新尝试', () => {
assert.ok(taskRepo.includes('currentOnly = false'))
assert.ok(taskRepo.includes('SELECT latest.id'))
assert.ok(taskRepo.includes('latest.unit_index = ft.unit_index'))
assert.ok(taskRepo.includes('ft.is_current = TRUE'))
})
test('创建新履约尝试时自动退休旧当前尝试', () => {
assert.ok(currentAttemptMigration.includes('attempt_no INTEGER'))
assert.ok(currentAttemptMigration.includes('is_current BOOLEAN'))
assert.ok(currentAttemptMigration.includes('uq_fulfillment_tasks_current_unit'))
assert.ok(taskRepo.includes('MAX(attempt_no)'))
assert.ok(taskRepo.includes('SET is_current = FALSE'))
assert.ok(taskRepo.includes('$26, TRUE'))
})
apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts
+62 -7
View File
@@ -19,6 +19,7 @@ test(
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const {
assignPendingOrderItemChannels,
assignChannelByInternalSku,
deletePendingChannelGroup,
deleteChannelRules,
@@ -35,6 +36,8 @@ test(
const { deleteManualSkuClassification, saveManualFulfillmentConfig } =
await import('../cost-center/manual-fulfillment-config-service.js')
const { runAffiliateFallback } = await import('./affiliate-fallback-service.js')
const { getFulfillmentAutomationCutoff } =
await import('./fulfillment-automation-config-service.js')
const { reconcilePendingCarriers } = await import('./carrier-reconciliation-service.js')
const { retryPendingWorkOrderRouting } = await import('./pending-channel-repair-service.js')
const {
@@ -61,6 +64,7 @@ test(
const missingVoucherSku = `集成测试无商品卡券-${suffix}`
const reactivatedVoucherSku = `集成测试归档卡券-${suffix}`
const retryPendingAffiliateSku = `集成测试失败待重试-${suffix}`
const cutoffTestSku = `集成测试自动化边界-${suffix}`
const retryWorkOrderRuleKey = `it-retry-${suffix}`
const orderIds: number[] = []
const orderItemIds: number[] = []
@@ -79,6 +83,7 @@ test(
missingVoucherSku,
reactivatedVoucherSku,
retryPendingAffiliateSku,
cutoffTestSku,
]
const createdVoucherProductIds: number[] = []
let autoManualProductId: number | null = null
@@ -140,6 +145,40 @@ test(
assert.equal(canonicalResolution.rows[0]?.platform_only, null)
assert.equal(canonicalResolution.rows[0]?.cjk_extension, '𠀀')
assert.equal(canonicalResolution.rows[0]?.case_variant, '黑色rock长袖t恤')
assert.ok(Number.isFinite(Date.parse(await getFulfillmentAutomationCutoff())))
const cutoffNewItemId = await createPendingOrderItem(cutoffTestSku)
const cutoffOldItemId = await createPendingOrderItem(cutoffTestSku)
await query(
`UPDATE orders
SET created_at = '2019-01-01T00:00:00+08:00',
updated_at = '2019-01-01T00:00:00+08:00'
WHERE id = (SELECT order_id FROM order_items WHERE id = $1)`,
[cutoffOldItemId],
)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES ($1, 'third_party_voucher', 10, FALSE, TRUE, '{}'::jsonb, 'integration_test')`,
[cutoffTestSku],
)
const cutoffAssignment = await assignPendingOrderItemChannels({
orderItemIds: [cutoffOldItemId, cutoffNewItemId],
batchSize: 10,
maxBatches: 1,
cutoffCreatedAt: '2020-01-01T00:00:00+08:00',
})
assert.equal(cutoffAssignment.scanned, 1)
assert.equal(cutoffAssignment.assigned, 1)
const cutoffStates = await query<{ id: number; fulfillment_status: string }>(
`SELECT id, fulfillment_status
FROM order_items
WHERE id = ANY($1::bigint[])
ORDER BY id DESC`,
[[cutoffOldItemId, cutoffNewItemId]],
)
assert.equal(cutoffStates.rows[0]?.fulfillment_status, 'pending')
assert.equal(cutoffStates.rows[1]?.fulfillment_status, 'assigned')
await query(
`INSERT INTO internal_sku_channels (internal_sku, channel, priority, is_fallback, enabled, source)
@@ -147,7 +186,13 @@ test(
[` 集成测试 规范－${suffix} `],
)
const normalizedInsert = await query<{ internal_sku: string }>(
`SELECT internal_sku FROM internal_sku_channels WHERE source = 'integration_test'`,
`SELECT internal_sku
FROM internal_sku_channels
WHERE source = 'integration_test'
AND canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($1)
ORDER BY id DESC
LIMIT 1`,
[triggerSku],
)
assert.equal(normalizedInsert.rows[0]?.internal_sku, triggerSku)
@@ -636,13 +681,21 @@ test(
[retryTaskItemId],
)
const retryTaskOrderRow = retryTaskOrder.rows[0]!
const insertRetryTask = (taskNo: string, status: string) =>
const insertRetryTask = (
taskNo: string,
status: string,
attemptNo: number,
isCurrent: boolean,
) =>
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
) VALUES ($1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash', $10, NOW(), NOW())`,
task_status, attempt_no, is_current, created_at, updated_at
) VALUES (
$1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash',
$10, $11, $12, NOW(), NOW()
)`,
[
retryTaskOrderRow.order_id,
retryTaskItemId,
@@ -654,12 +707,14 @@ test(
retryTaskOrderRow.platform_order_id,
fallbackProfileId,
status,
attemptNo,
isCurrent,
],
)
await insertRetryTask(`IT-AFF-CLOSED-${suffix}`, 'closed')
await insertRetryTask(`IT-AFF-ACTIVE-${suffix}`, 'paid')
await insertRetryTask(`IT-AFF-CLOSED-${suffix}`, 'closed', 1, false)
await insertRetryTask(`IT-AFF-ACTIVE-${suffix}`, 'paid', 2, true)
await assert.rejects(
() => insertRetryTask(`IT-AFF-DUPLICATE-${suffix}`, 'claimed'),
() => insertRetryTask(`IT-AFF-DUPLICATE-${suffix}`, 'claimed', 3, false),
/duplicate key value violates unique constraint/,
)
const retryTaskRows = await query<{ task_status: string }>(
apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts
+21
View File
@@ -11,6 +11,18 @@ const auditScript = readFileSync(
new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url),
'utf8',
)
const automationCutoffMigration = readFileSync(
new URL('../../db/migrations/139_fulfillment_automation_cutoff.sql', import.meta.url),
'utf8',
)
const automationConfig = readFileSync(
new URL('./fulfillment-automation-config-service.ts', import.meta.url),
'utf8',
)
const schedulerJobs = readFileSync(
new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url),
'utf8',
)
test('履约读模型按订单项拆分渠道、载体、尝试、异常和锁定原因', () => {
assert.ok(migration.includes('CREATE OR REPLACE VIEW v_fulfillment_item_state'))
@@ -50,3 +62,12 @@ test('履约审计脚本覆盖状态不一致、售后队列和重复尝试', ()
assert.ok(auditScript.includes('近 24 小时重复尝试订单项'))
assert.ok(auditScript.includes('--strict'))
})
test('历史履约数据不进入自动任务，只处理自动化生效时间之后的新订单', () => {
assert.ok(automationCutoffMigration.includes('fulfillment_automation_cutoff'))
assert.ok(automationConfig.includes('getFulfillmentAutomationCutoff'))
assert.ok(automationConfig.includes('fulfillmentCreatedAfterSql'))
assert.ok(schedulerJobs.includes('cutoffCreatedAt'))
assert.ok(schedulerJobs.includes('getFulfillmentAutomationCutoff'))
assert.ok(schedulerJobs.includes('runAffiliateFallback'))
})
apps/backend/src/services/scheduler/finance-pipeline-jobs.ts
+63 -9
View File
@@ -3,6 +3,10 @@ import { query } from '../../db/client.js'
import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
import { runAffiliateFallback } from '../fulfillment/affiliate-fallback-service.js'
import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
import {
fulfillmentCreatedAfterSql,
getFulfillmentAutomationCutoff,
} from '../fulfillment/fulfillment-automation-config-service.js'
import { processDueOrderServiceAutoPublishJobs } from '../worker-platform/order-service-auto-publish-service.js'
import {
resolveExceptionsForItemsByReason,
@@ -52,12 +56,14 @@ function clampInt(value: unknown, fallback: number, min: number, max: number): n
export async function runChannelAssignJob(job: JsonObject = {}) {
const config = jobConfig(job)
const cutoffCreatedAt = await getFulfillmentAutomationCutoff()
const result = await assignPendingOrderItemChannels({
batchSize: clampInt(config.batchSize, 500, 1, 5000),
maxBatches: clampInt(config.maxBatches, 20, 1, 200),
...(config.afterId != null
? { afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER) }
: {}),
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
})
await touchWatermark('channel_assign', { lastItemId: result.nextCursor })
@@ -85,6 +91,7 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
export async function runCarrierDispatchJob(job: JsonObject = {}) {
const config = jobConfig(job)
const cutoffCreatedAt = await getFulfillmentAutomationCutoff()
const limit = clampInt(config.limit, 200, 1, 2000)
const reconcileCursor =
config.carrierReconcileAfterId != null
@@ -93,13 +100,19 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
const reconciled = await reconcilePendingCarriers({
limit: Math.min(limit, 200),
afterId: reconcileCursor,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
})
await touchWatermark('carrier_reconcile', { lastItemId: reconciled.nextCursor })
const repairedAssigned = await reconcilePendingCarriers({
limit: Math.min(limit, 200),
includeAssignedMissingCarrier: true,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
})
await assignPendingOrderItemChannels({ batchSize: 500, maxBatches: 1 }).catch(() => undefined)
await assignPendingOrderItemChannels({
batchSize: 500,
maxBatches: 1,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
}).catch(() => undefined)
const cursor =
config.afterId != null
? Math.max(0, Math.trunc(Number(config.afterId)))
@@ -109,6 +122,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
'SELECT canonicalize_internal_sku(sku_key) AS sku_key FROM manual_fulfillment_configs WHERE enabled',
)
const configKeys = configKeysResult.rows.map((row) => row.sku_key)
const manualParams: unknown[] = [limit, cursor, configKeys]
const manualCutoff = fulfillmentCreatedAfterSql(manualParams, cutoffCreatedAt)
const manual = await query<{ id: number; sku_key: string; config_id: number }>(
`
@@ -126,10 +141,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND oi.fulfillment_config_id IS NULL
AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = ANY($3::text[])
AND oi.id > $2
${manualCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor, configKeys],
manualParams,
)
let attached = 0
@@ -146,6 +162,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 人工渠道但没有配置：进「人工 SKU 未配置」队列。
const missingManualParams: unknown[] = [limit, cursor, configKeys]
const missingManualCutoff = fulfillmentCreatedAfterSql(missingManualParams, cutoffCreatedAt)
const missingManual = await query<{ id: number; sku_key: string }>(
`
SELECT oi.id,
@@ -164,10 +182,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ce.status <> 'resolved'
)
AND oi.id > $2
${missingManualCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor, configKeys],
missingManualParams,
)
const exceptionEntries: AggregatedExceptionEntry[] = []
@@ -181,6 +200,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 已配置人工 SKU、但还没确认履约的单：进入「人工成本未记录」队列。
const pendingManualCostParams: unknown[] = [limit, cursor]
const pendingManualCostCutoff = fulfillmentCreatedAfterSql(
pendingManualCostParams,
cutoffCreatedAt,
)
const pendingManualCost = await query<{ id: number; sku_key: string }>(
`
SELECT oi.id,
@@ -204,10 +228,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ce.status <> 'resolved'
)
AND oi.id > $2
${pendingManualCostCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor],
pendingManualCostParams,
)
for (const row of pendingManualCost.rows) {
@@ -220,6 +245,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 自动发货 / 接单平台被运营指定后却没有载体：不能静默卡住，进问题队列。
const missingCarrierParams: unknown[] = [limit, cursor]
const missingCarrierCutoff = fulfillmentCreatedAfterSql(missingCarrierParams, cutoffCreatedAt)
const missingCarrierRaw = await query<{
id: number
channel: string
@@ -241,10 +268,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
)
AND oi.id > $2
${missingCarrierCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor],
missingCarrierParams,
)
const queuedCarrier = await query<{ order_item_id: number }>(
@@ -273,6 +301,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 已分配订单出现两个真实执行方时不能继续算正常渠道，必须进冲突队列人工处置。
const conflictingCarrierParams: unknown[] = [limit]
const conflictingCarrierCutoff = fulfillmentCreatedAfterSql(
conflictingCarrierParams,
cutoffCreatedAt,
)
const conflictingCarriers = await query<{
id: number
channel: string
@@ -299,10 +332,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
)
)::int > 1
${conflictingCarrierCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit],
conflictingCarrierParams,
)
for (const row of conflictingCarriers.rows) {
exceptionEntries.push({
@@ -331,8 +365,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
maxId = Math.max(maxId, Number(row.id))
}
const resolved = await resolveCarrierDispatchExceptions()
const fallback = await runAffiliateFallback({ limit: 200 }).catch(() => ({
const resolved = await resolveCarrierDispatchExceptions(cutoffCreatedAt)
const fallback = await runAffiliateFallback({
limit: 200,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
}).catch(() => ({
scanned: 0,
switched: 0,
}))
@@ -360,8 +397,10 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
/**
* 载体/配置补上后，把对应异常自动关掉（每轮最多处理 500 条）。
*/
async function resolveCarrierDispatchExceptions(): Promise<number> {
async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promise<number> {
// 游标会跳过已扫过的订单项；配置是后来补的，这里兜底把配置挂上。
const attachParams: unknown[] = []
const attachCutoff = fulfillmentCreatedAfterSql(attachParams, cutoffCreatedAt)
await query(
`
UPDATE order_items oi
@@ -382,9 +421,17 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
AND ce.reason = 'manual_config_missing'
AND ce.status <> 'resolved'
)
${attachCutoff}
`,
attachParams,
)
const carrierParams: unknown[] = []
const carrierCutoff = fulfillmentCreatedAfterSql(carrierParams, cutoffCreatedAt)
const configParams: unknown[] = []
const configCutoff = fulfillmentCreatedAfterSql(configParams, cutoffCreatedAt)
const manualCostParams: unknown[] = []
const manualCostCutoff = fulfillmentCreatedAfterSql(manualCostParams, cutoffCreatedAt)
const [carrier, config, manualCost] = await Promise.all([
query<{ order_item_id: number }>(
`
@@ -404,8 +451,10 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
))
)
${carrierCutoff}
LIMIT 500
`,
carrierParams,
),
query<{ order_item_id: number }>(
`
@@ -416,14 +465,17 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
WHERE ce.reason = 'manual_config_missing'
AND ce.status <> 'resolved'
AND oi.fulfillment_config_id IS NOT NULL
${configCutoff}
LIMIT 500
`,
configParams,
),
query<{ order_item_id: number }>(
`
SELECT DISTINCT i.order_item_id
FROM cost_exception_items i
JOIN cost_exceptions ce ON ce.id = i.exception_id
JOIN order_items oi ON oi.id = i.order_item_id
WHERE ce.reason = 'manual_cost_missing'
AND ce.status <> 'resolved'
AND EXISTS (
@@ -432,8 +484,10 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
AND cs.order_item_id = i.order_item_id
AND cs.superseded_at IS NULL AND reversed_at IS NULL
)
${manualCostCutoff}
LIMIT 500
`,
manualCostParams,
),
])