diff --git a/apps/backend/src/config/app-config-keys.ts b/apps/backend/src/config/app-config-keys.ts index 706b01ae..d6a7c66d 100644 --- a/apps/backend/src/config/app-config-keys.ts +++ b/apps/backend/src/config/app-config-keys.ts @@ -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 diff --git a/apps/backend/src/db/migrations/138_fulfillment_current_attempt.sql b/apps/backend/src/db/migrations/138_fulfillment_current_attempt.sql new file mode 100644 index 00000000..03cc5a64 --- /dev/null +++ b/apps/backend/src/db/migrations/138_fulfillment_current_attempt.sql @@ -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 + '是否为该订单项单元当前尝试；历史尝试保留但不再标记为当前'; diff --git a/apps/backend/src/db/migrations/139_fulfillment_automation_cutoff.sql b/apps/backend/src/db/migrations/139_fulfillment_automation_cutoff.sql new file mode 100644 index 00000000..caa37899 --- /dev/null +++ b/apps/backend/src/db/migrations/139_fulfillment_automation_cutoff.sql @@ -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 是履约自动化生效边界'; diff --git a/apps/backend/src/repositories/task-repo.ts b/apps/backend/src/repositories/task-repo.ts index cd29492c..abd8296b 100644 --- a/apps/backend/src/repositories/task-repo.ts +++ b/apps/backend/src/repositories/task-repo.ts @@ -94,6 +94,18 @@ export async function listTasksByOrderIds(orderIds: number[]): Promise { 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 { 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[] = [] diff --git a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts index 06cf2f44..e40df725 100644 --- a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts @@ -52,6 +52,7 @@ async function loadCandidateItems(input: { assignedMissingCarrier?: boolean afterId?: number includeLocked?: boolean + cutoffCreatedAt?: string }): Promise { 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 } : {}), }) diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index a731ae07..842d8fd1 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -254,6 +254,7 @@ export async function assignPendingOrderItemChannels( afterId?: number orderItemIds?: number[] includeLocked?: boolean + cutoffCreatedAt?: string } = {}, ): Promise { 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 `, diff --git a/apps/backend/src/services/fulfillment/fulfillment-automation-config-service.ts b/apps/backend/src/services/fulfillment/fulfillment-automation-config-service.ts new file mode 100644 index 00000000..83b70b57 --- /dev/null +++ b/apps/backend/src/services/fulfillment/fulfillment-automation-config-service.ts @@ -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 { + 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` +} diff --git a/apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts b/apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts index 599807ec..52b6d831 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts @@ -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')) }) diff --git a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts index c5249f68..42f659c2 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts @@ -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 }>( diff --git a/apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts b/apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts index deee603c..48cae4bc 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts @@ -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')) +}) diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 7ff70079..590a594d 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -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 { +async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promise { // 游标会跳过已扫过的订单项；配置是后来补的，这里兜底把配置挂上。 + const attachParams: unknown[] = [] + const attachCutoff = fulfillmentCreatedAfterSql(attachParams, cutoffCreatedAt) await query( ` UPDATE order_items oi @@ -382,9 +421,17 @@ async function resolveCarrierDispatchExceptions(): Promise { 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 { 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 { 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 { AND cs.order_item_id = i.order_item_id AND cs.superseded_at IS NULL AND reversed_at IS NULL ) + ${manualCostCutoff} LIMIT 500 `, + manualCostParams, ), ]) diff --git a/docs/README.md b/docs/README.md index 2c3212d6..cc152c9f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,6 +1,6 @@ # order_site 文档索引 -> 最后补充：2026-11-05，新增《履约系统架构复盘与优化方案》，代码基线 `e8b76b3b`，生产数据库基线迁移 `136`。 +> 最后补充：2026-11-05，新增《履约系统架构复盘与优化方案》；生产代码基线 `e8b76b3b` / 迁移 `136`，开发分支 `dev_11.5_2` 已推进到迁移 `139`。 > 数据基线：生产 dump `backups/postgres/order_site-20261022-092601.dump`（数据 2026-07-10 ~ 2026-10-22）；历史全量数字仅作基线，**上线验收以 [财务上线初始化与验收-2026-10-26.md](财务上线初始化与验收-2026-10-26.md) 为准**。 > 本文是 `docs/` 的唯一入口。**整体财务流程看 [财务流程总览-2026-11-01.md](财务流程总览-2026-11-01.md)**；数据模型看 [成本中心与履约模型.md](成本中心与履约模型.md)，接口看 [成本中心API契约.md](成本中心API契约.md)。 > 新增文档必须在下表登记状态；过期文档不再直接改写结论，统一移入 `docs/archive/`。 diff --git a/docs/履约系统架构复盘与优化方案-2026-11-05.md b/docs/履约系统架构复盘与优化方案-2026-11-05.md index 34fdbf5c..9f971b08 100644 --- a/docs/履约系统架构复盘与优化方案-2026-11-05.md +++ b/docs/履约系统架构复盘与优化方案-2026-11-05.md @@ -9,8 +9,8 @@ status: draft > 状态：**方案**，尚未实施。 > 最后校准：2026-11-05。 -> 代码基线：`e8b76b3b`。 -> 数据库基线：迁移 `136`。 +> 代码基线：`e8b76b3b` + `dev_11.5_2` 履约架构改造。 +> 数据库基线：生产迁移 `136`；开发分支迁移 `138`。 > 生产核对方式：只读 SQL、后台页面、现行文档与源码交叉检查。 ## 实施进度 @@ -22,8 +22,23 @@ status: draft - 已完成：待分类查询返回每类问题的订单项数量，而不是只返回原因集合。 - 已完成：保存渠道、补载体、确认重试、解除锁定和归档按当前问题分类限定作用域。 - 已完成：新增 `npm run audit:fulfillment` 只读审计脚本。 +- 已完成：迁移 `138` 为履约任务增加 `attempt_no / is_current`，每个订单项单元只允许一个当前尝试。 +- 已完成：创建新任务时自动退休旧当前尝试，任务页“当前任务”改为读取正式指针。 +- 已完成：迁移 `139` 写入 `fulfillment_automation_cutoff`，自动渠道判定、补载体、自动转人工和异常收口只处理生效时间之后创建的订单。 - 待继续：正式 `fulfillments / attempts / issues / events` 表、统一写入口、独立售后队列和生产存量修复。 +### 历史数据边界 + +生产历史履约异常已经由人工处理完成。历史订单只保留查询和审计价值，不参与任何自动尝试： + +- 不自动重新判定渠道。 +- 不自动补建任务或工单。 +- 不自动重开失败、关闭、待重试任务。 +- 不自动转人工或修改历史异常状态。 +- 运营仍可在后台显式查看和处理历史订单，但必须经过人工动作。 + +自动化边界以 `app_config_entries.fulfillment_automation_cutoff` 为唯一真源。 + ## 一、结论 这套系统并不需要推倒重写，但不能继续靠新增字段、JSON 原因和页面按钮修补。