diff --git a/apps/backend/src/db/migrations/162_fulfillment_queue_buckets.sql b/apps/backend/src/db/migrations/162_fulfillment_queue_buckets.sql new file mode 100644 index 00000000..bacfa63b --- /dev/null +++ b/apps/backend/src/db/migrations/162_fulfillment_queue_buckets.sql @@ -0,0 +1,109 @@ +-- 162_fulfillment_queue_buckets.sql +-- +-- 把履约待办从单一 pending 队列拆成只读派生桶: +-- actionable 正常等待补载体,允许自动扫描 +-- configuration 缺内部 SKU、渠道或渠道配置 +-- operator_review 失败重试、人工复核、已关闭/取消、系统不一致 +-- terminal 已退款、退款申请、平台取消或商业订单已关闭 +-- +-- 本迁移不新增可写状态字段,也不改 fulfillments 的权威状态。 +-- 分类只由订单商业状态、正式履约字段和结构化 issue 派生,供扫描器与后台共用。 + +CREATE INDEX IF NOT EXISTS idx_fulfillments_pending_unlocked_item + ON fulfillments (order_item_id) + WHERE status = 'pending' AND locked = FALSE; + +COMMENT ON INDEX idx_fulfillments_pending_unlocked_item IS + '履约队列扫描:仅覆盖待处理且未锁定的正式履约记录'; + +CREATE OR REPLACE VIEW v_fulfillment_queue AS +WITH classified AS ( + SELECT + state.*, + CASE + WHEN state.pay_status = 'refunded' + OR state.order_status IN ('refunded', 'cancelled', 'closed') + OR state.issue_code IN ('user_refund', 'platform_cancelled') + THEN 'terminal' + WHEN state.issue_category IN ('configuration', 'classification') + OR state.issue_code IN ( + 'missing_internal_sku', + 'missing_channel', + 'manual_config_missing', + 'work_rule_missing', + 'affiliate_mapping_missing' + ) + OR COALESCE(state.fulfillment_pending_detail ->> 'carrierIssue', '') IN ( + 'manual_config_missing', + 'work_rule_missing', + 'affiliate_mapping_missing', + 'missing_internal_sku', + 'missing_channel' + ) + THEN 'configuration' + WHEN state.issue_category = 'manual_review' + OR state.issue_code IN ( + 'carrier_manual_review', + 'carrier_retry_pending', + 'carrier_expired', + 'carrier_closed' + ) + OR COALESCE(state.fulfillment_pending_detail ->> 'carrierIssue', '') IN ( + 'carrier_task_closed', + 'carrier_task_user_refund', + 'carrier_task_retry_pending', + 'carrier_task_cancelled', + 'carrier_task_expired', + 'carrier_task_manual_review' + ) + THEN 'operator_review' + WHEN state.issue_code = 'state_inconsistent' + AND state.issue_category = 'system' + AND COALESCE(state.fulfillment_pending_detail ->> 'carrierIssue', '') + IN ('', 'carrier_not_attempted') + THEN 'actionable' + WHEN state.issue_category = 'system' + THEN 'operator_review' + ELSE 'actionable' + END AS queue_bucket + FROM v_fulfillment_item_state state +), +eligible AS ( + SELECT + classified.*, + ( + classified.fulfillment_status = 'pending' + AND classified.pay_status = 'paid' + AND classified.order_status NOT IN ('refunded', 'cancelled', 'closed') + AND NOT EXISTS ( + SELECT 1 + FROM fulfillment_tasks task + WHERE task.order_item_id = classified.order_item_id + AND task.executor_key = 'affiliate_dash' + AND task.task_status NOT IN ( + 'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled' + ) + ) + AND NOT EXISTS ( + SELECT 1 + FROM work_orders work_order + WHERE work_order.order_item_id = classified.order_item_id + AND work_order.status <> 'cancelled' + ) + ) AS carrier_queue_candidate + FROM classified +) +SELECT + eligible.order_item_id AS id, + eligible.*, + ( + eligible.carrier_queue_candidate + AND eligible.queue_bucket = 'actionable' + AND eligible.fulfillment_locked = FALSE + ) AS actionable_for_carrier, + item.manual_sku_key +FROM eligible +JOIN order_items item ON item.id = eligible.order_item_id; + +COMMENT ON VIEW v_fulfillment_queue IS + '履约待办只读投影:actionable/configuration/operator_review/terminal;actionable_for_carrier 为补载体自动扫描唯一入口'; diff --git a/apps/backend/src/routes/admin/fulfillment-channel.ts b/apps/backend/src/routes/admin/fulfillment-channel.ts index 8384488f..444977a1 100644 --- a/apps/backend/src/routes/admin/fulfillment-channel.ts +++ b/apps/backend/src/routes/admin/fulfillment-channel.ts @@ -140,6 +140,7 @@ router.get( pageSize: Number(req.query.pageSize), keyword: String(req.query.keyword || ''), issueCategory: String(req.query.issueCategory || ''), + queueBucket: String(req.query.queueBucket || ''), }), { successMessage: 'ok', @@ -158,6 +159,7 @@ router.post( groupKey: String(bodyOf(req).groupKey || ''), internalSku: String(bodyOf(req).internalSku || ''), issueCategory: String(bodyOf(req).issueCategory || ''), + queueBucket: String(bodyOf(req).queueBucket || ''), adminUserId: req.adminSession?.userId ?? null, }), { @@ -236,6 +238,7 @@ router.post( groupKey: String(body.groupKey || ''), internalSku: String(body.internalSku || ''), issueCategory: String(body.issueCategory || ''), + queueBucket: String(body.queueBucket || ''), }) }, { @@ -352,6 +355,7 @@ router.post( mappedInternalSku: String(body.mappedInternalSku || ''), channel, issueCategory: String(body.issueCategory || ''), + queueBucket: String(body.queueBucket || ''), adminUserId, note, }) diff --git a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts index db6e78ea..24c0a9b5 100644 --- a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts +++ b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts @@ -659,6 +659,7 @@ async function retryPendingManualItems(skuKey: string): Promise { orderItemIds: ids, limit: ids.length, allowFallback: true, + includeNonActionable: true, }) return result.fallbacksSelected } diff --git a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts index 1e33b2b2..92b43165 100644 --- a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts @@ -55,6 +55,10 @@ async function loadCandidateItems(input: { orderItemIds?: number[] assignedMissingCarrier?: boolean afterId?: number + createdAfter?: string + createdBefore?: string + orderDirection?: 'asc' | 'desc' + includeNonActionable?: boolean includeLocked?: boolean cutoffCreatedAt?: string }): Promise { @@ -76,9 +80,22 @@ async function loadCandidateItems(input: { params.push(cutoffCreatedAt) cutoffFilter = `AND oi.created_at >= $${params.length}::timestamptz` } + const createdAfter = String(input.createdAfter || '').trim() + if (createdAfter && Number.isFinite(Date.parse(createdAfter))) { + params.push(createdAfter) + cutoffFilter += ` AND oi.created_at >= $${params.length}::timestamptz` + } + const createdBefore = String(input.createdBefore || '').trim() + if (createdBefore && Number.isFinite(Date.parse(createdBefore))) { + params.push(createdBefore) + 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') + AND oi.stored_channel IN ('affiliate_dash', 'work_order') + AND oi.queue_bucket = 'actionable' + AND oi.pay_status = 'paid' + AND oi.order_status NOT IN ('refunded', 'cancelled', 'closed') AND NOT EXISTS ( SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled' @@ -96,7 +113,14 @@ async function loadCandidateItems(input: { LIMIT 1 ), '') NOT IN ('USER_APPLY_REFUND', 'affiliate_dash_cancelled')` : `oi.fulfillment_status = 'pending' + AND oi.carrier_queue_candidate = TRUE + ${ + input.includeNonActionable + ? `AND oi.queue_bucket IN ('actionable', 'configuration', 'operator_review')` + : `AND oi.queue_bucket = 'actionable'` + } ${input.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}` + const orderDirection = input.orderDirection === 'desc' ? 'DESC' : 'ASC' const result = await query( ` @@ -105,7 +129,7 @@ async function loadCandidateItems(input: { oi.order_id, resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku, capability.channels - FROM v_order_item_current oi + FROM v_fulfillment_queue oi JOIN LATERAL ( SELECT COALESCE( jsonb_agg( @@ -136,7 +160,7 @@ async function loadCandidateItems(input: { WHERE ${statusFilter} ${idFilter} ${cutoffFilter} - ORDER BY oi.id + ORDER BY oi.id ${orderDirection} LIMIT $1 `, params, @@ -259,6 +283,10 @@ export async function reconcilePendingCarriers( includeAssignedMissingCarrier?: boolean allowFallback?: boolean afterId?: number + createdAfter?: string + createdBefore?: string + orderDirection?: 'asc' | 'desc' + includeNonActionable?: boolean includeLocked?: boolean retryInactiveCarrier?: boolean cutoffCreatedAt?: string @@ -279,6 +307,10 @@ export async function reconcilePendingCarriers( ...(input.afterId != null ? { afterId: input.afterId } : {}), ...(input.includeLocked ? { includeLocked: true } : {}), ...(input.cutoffCreatedAt ? { cutoffCreatedAt: input.cutoffCreatedAt } : {}), + ...(input.createdAfter ? { createdAfter: input.createdAfter } : {}), + ...(input.createdBefore ? { createdBefore: input.createdBefore } : {}), + ...(input.orderDirection ? { orderDirection: input.orderDirection } : {}), + ...(input.includeNonActionable ? { includeNonActionable: true } : {}), ...(input.orderItemIds ? { orderItemIds: input.orderItemIds } : {}), }) @@ -425,18 +457,8 @@ export async function reconcilePendingCarriers( const remainingResult = await query<{ count: number }>( ` SELECT count(*)::int AS count - FROM v_order_item_current oi - WHERE oi.fulfillment_status IN ('pending', 'assigned') - AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order') - AND NOT EXISTS ( - SELECT 1 FROM work_orders wo - WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled' - ) - AND NOT EXISTS ( - 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(', ')}) - ) + FROM v_fulfillment_queue oi + WHERE oi.carrier_queue_candidate = TRUE `, ) @@ -447,10 +469,11 @@ export async function reconcilePendingCarriers( fallbacksSelected, failed, remaining: Number(remainingResult.rows[0]?.count || 0), - nextCursor: input.orderItemIds?.length - ? 0 - : rows.length < limit + nextCursor: + input.orderDirection === 'desc' || input.orderItemIds?.length ? 0 - : Number(rows[rows.length - 1]?.id || 0), + : rows.length < limit + ? 0 + : Number(rows[rows.length - 1]?.id || 0), } } diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index e913a296..c332bd69 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -1060,6 +1060,7 @@ export type PendingGroupRow = { sku_codes: string[] pending_reasons: string[] pending_subreasons: string[] + queue_buckets: FulfillmentQueueBucket[] configured_channels: string[] locked_item_count: number item_count: number @@ -1141,6 +1142,20 @@ const PENDING_ISSUE_CATEGORIES = [ 'after_sales', 'system', ] as const +export const FULFILLMENT_QUEUE_BUCKETS = [ + 'actionable', + 'configuration', + 'operator_review', + 'terminal', +] as const +export type FulfillmentQueueBucket = (typeof FULFILLMENT_QUEUE_BUCKETS)[number] + +function normalizeQueueBucket(value: unknown): FulfillmentQueueBucket | '' { + const bucket = String(value || '').trim() + return (FULFILLMENT_QUEUE_BUCKETS as readonly string[]).includes(bucket) + ? (bucket as FulfillmentQueueBucket) + : '' +} /** 待分类队列:按全局内部 SKU 聚合(跨平台、跨店铺合并为一行)。 */ export async function listPendingChannelGroups(input: { @@ -1148,17 +1163,28 @@ export async function listPendingChannelGroups(input: { pageSize?: number keyword?: string issueCategory?: string -}): Promise<{ items: PendingGroupRow[]; total: number; page: number; pageSize: number }> { + queueBucket?: string +}): Promise<{ + items: PendingGroupRow[] + total: number + page: number + pageSize: number + bucketCounts: Record +}> { 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 queueBucket = normalizeQueueBucket(input.queueBucket) const issueCategoryInput = String(input.issueCategory || '').trim() const issueCategory = (PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategoryInput) ? issueCategoryInput : '' const params: unknown[] = [] const clauses = [`v.fulfillment_status = 'pending'`] - if (issueCategory) { + if (queueBucket) { + params.push(queueBucket) + clauses.push(`v.queue_bucket = $${params.length}`) + } else if (issueCategory) { params.push(issueCategory) clauses.push(`v.issue_category = $${params.length}`) } @@ -1172,13 +1198,49 @@ export async function listPendingChannelGroups(input: { } const where = clauses.join(' AND ') + const bucketCountParams: unknown[] = [] + let bucketCountKeyword = '' + if (keyword) { + bucketCountParams.push(`%${keyword}%`) + bucketCountKeyword = `AND ( + v.internal_sku ILIKE $${bucketCountParams.length} + OR v.sku_code ILIKE $${bucketCountParams.length} + OR v.sku_name ILIKE $${bucketCountParams.length} + )` + } + const bucketCountRows = await query<{ queue_bucket: FulfillmentQueueBucket; count: number }>( + ` + WITH pending_items AS ( + SELECT + v.queue_bucket, + ${PENDING_STATE_GROUP_KEY_SQL} AS group_key + FROM v_fulfillment_queue v + WHERE v.fulfillment_status = 'pending' + ${bucketCountKeyword} + ) + SELECT queue_bucket, COUNT(DISTINCT group_key)::int AS count + FROM pending_items + GROUP BY queue_bucket + `, + bucketCountParams, + ) + const bucketCounts: Record = { + actionable: 0, + configuration: 0, + operator_review: 0, + terminal: 0, + } + for (const row of bucketCountRows.rows) { + bucketCounts[row.queue_bucket] = Number(row.count || 0) + } + const total = await query<{ count: number }>( ` WITH pending_items AS ( SELECT v.internal_sku, ${PENDING_STATE_GROUP_KEY_SQL} AS group_key - FROM v_fulfillment_item_state v + FROM v_fulfillment_queue v WHERE ${where} ) SELECT COUNT(DISTINCT group_key)::int AS count @@ -1194,7 +1256,7 @@ export async function listPendingChannelGroups(input: { SELECT v.*, ${PENDING_STATE_GROUP_KEY_SQL} AS group_key - FROM v_fulfillment_item_state v + FROM v_fulfillment_queue v WHERE ${where} ), grouped AS ( @@ -1217,6 +1279,7 @@ export async function listPendingChannelGroups(input: { NULL ) AS pending_reasons, array_remove(array_agg(DISTINCT issue_code), '') AS pending_subreasons, + array_agg(DISTINCT queue_bucket) AS queue_buckets, 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, @@ -1284,6 +1347,7 @@ export async function listPendingChannelGroups(input: { grouped.sku_codes, grouped.pending_reasons, grouped.pending_subreasons, + grouped.queue_buckets, COALESCE(cfg.channels, ARRAY[]::text[]) AS configured_channels, grouped.locked_item_count, grouped.item_count, @@ -1326,9 +1390,10 @@ export async function listPendingChannelGroups(input: { v.shop_id, v.shop_name, ${PENDING_STATE_GROUP_KEY_SQL} AS group_key - FROM v_fulfillment_item_state v + FROM v_fulfillment_queue v WHERE v.fulfillment_status = 'pending' AND ($2::text = '' OR v.issue_category = $2) + AND ($3::text = '' OR v.queue_bucket = $3) ) SELECT group_key, @@ -1353,7 +1418,7 @@ export async function listPendingChannelGroups(input: { usage_count DESC, shop_name `, - [groupKeys, issueCategory], + [groupKeys, issueCategory, queueBucket], ) const seen = new Set() @@ -1377,6 +1442,7 @@ export async function listPendingChannelGroups(input: { total: total.rows[0]?.count || 0, page, pageSize, + bucketCounts, } } @@ -1453,6 +1519,7 @@ export async function retryPendingCarrierGroup(input: { orderItemIds: batchIds, limit: batchIds.length, includeLocked: input.includeLocked === true, + includeNonActionable: true, retryInactiveCarrier: input.retryInactiveCarrier === true, }) workOrdersCreated += carrier.workOrdersCreated @@ -1532,6 +1599,7 @@ export async function unlockPendingChannelGroup(input: { groupKey?: string internalSku?: string issueCategory?: string + queueBucket?: string }): Promise<{ updated: number }> { const scope = resolvePendingGroupScopeInput(input) if (!scope) { @@ -1551,13 +1619,17 @@ export async function unlockPendingChannelGroup(input: { 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)) { + const queueBucket = normalizeQueueBucket(input.queueBucket) + if (queueBucket) { + params.push(queueBucket) + where.push(`state.queue_bucket = $${params.length}`) + } else if ((PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategory)) { params.push(issueCategory) where.push(`state.issue_category = $${params.length}`) } const matched = await query<{ order_item_id: number }>( `SELECT state.order_item_id - FROM v_fulfillment_item_state state + FROM v_fulfillment_queue state WHERE ${where.join(' AND ')} ORDER BY state.order_item_id`, params, @@ -1623,6 +1695,7 @@ export async function deletePendingChannelGroup(input: { groupKey?: string internalSku?: string issueCategory?: string + queueBucket?: string adminUserId: number | null }): Promise { const scope = resolvePendingGroupScopeInput(input) @@ -1644,7 +1717,16 @@ export async function deletePendingChannelGroup(input: { 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)) { + const queueBucket = normalizeQueueBucket(input.queueBucket) + if (queueBucket) { + itemParams.push(queueBucket) + itemWhere.push(`EXISTS ( + SELECT 1 + FROM v_fulfillment_queue state + WHERE state.order_item_id = oi.id + AND state.queue_bucket = $${itemParams.length} + )`) + } else if ((PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategory)) { itemParams.push(issueCategory) itemWhere.push(`EXISTS ( SELECT 1 @@ -1761,6 +1843,7 @@ export async function assignPendingChannelGroup(input: { mappedInternalSku?: string channel: string issueCategory?: string + queueBucket?: string adminUserId: number | null note?: string | undefined }): Promise<{ @@ -1777,6 +1860,13 @@ export async function assignPendingChannelGroup(input: { errorCode: 'pending_group_key_required', }) } + const queueBucket = normalizeQueueBucket(input.queueBucket) + if (queueBucket === 'terminal') { + throw createHttpError('售后或已终止订单不能重新指定履约渠道', { + statusCode: 409, + errorCode: 'terminal_fulfillment_bucket_not_assignable', + }) + } if (scope.type !== 'internal_sku') { const mappedInternalSku = canonicalizeInternalSku(input.mappedInternalSku) if (!mappedInternalSku || !containsCjk(mappedInternalSku)) { @@ -1788,7 +1878,15 @@ export async function assignPendingChannelGroup(input: { 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)) { + if (queueBucket) { + updateParams.push(queueBucket) + issueCategoryFilter = `AND EXISTS ( + SELECT 1 + FROM v_fulfillment_queue state + WHERE state.order_item_id = oi.id + AND state.queue_bucket = $${updateParams.length} + )` + } else if ((PENDING_ISSUE_CATEGORIES as readonly string[]).includes(issueCategory)) { updateParams.push(issueCategory) issueCategoryFilter = `AND EXISTS ( SELECT 1 @@ -1829,6 +1927,7 @@ export async function assignPendingChannelGroup(input: { internalSku: mappedInternalSku, channel: input.channel, issueCategory: input.issueCategory || '', + queueBucket, orderItemIds: updated.rows.map((row) => Number(row.id)), adminUserId: input.adminUserId, note: input.note, @@ -1838,6 +1937,7 @@ export async function assignPendingChannelGroup(input: { internalSku: scope.internalSku, channel: input.channel, issueCategory: input.issueCategory || '', + queueBucket, adminUserId: input.adminUserId, note: input.note, }) @@ -1848,6 +1948,7 @@ export async function assignChannelByInternalSku(input: { internalSku: string channel: string issueCategory?: string + queueBucket?: string orderItemIds?: number[] adminUserId: number | null note?: string | undefined @@ -1878,6 +1979,13 @@ export async function assignChannelByInternalSku(input: { } const issueCategory = String(input.issueCategory || '').trim() + const queueBucket = normalizeQueueBucket(input.queueBucket) + if (queueBucket === 'terminal') { + throw createHttpError('售后或已终止订单不能重新指定履约渠道', { + statusCode: 409, + errorCode: 'terminal_fulfillment_bucket_not_assignable', + }) + } const requestedOrderItemIds = [ ...new Set( (input.orderItemIds || []).map(Number).filter((id) => Number.isInteger(id) && id > 0), @@ -1887,7 +1995,7 @@ export async function assignChannelByInternalSku(input: { const idsResult = await query<{ id: number }>( ` SELECT state.order_item_id AS id - FROM v_fulfillment_item_state state + FROM v_fulfillment_queue state WHERE state.internal_sku = $1 AND ( state.fulfillment_status = 'pending' @@ -1897,12 +2005,20 @@ export async function assignChannelByInternalSku(input: { ) ) AND ($3::text = '' OR state.issue_category = $3) + AND ($6::text = '' OR state.queue_bucket = $6) AND ( $4::boolean = FALSE OR state.order_item_id = ANY($5::bigint[]) ) `, - [internalSku, input.channel, issueCategory, requestedOrderItemIds.length > 0, idParams], + [ + internalSku, + input.channel, + issueCategory, + requestedOrderItemIds.length > 0, + idParams, + queueBucket, + ], ) const draftProductId = await ensureDraftProductForChannel(internalSku, input.channel) diff --git a/apps/backend/src/services/fulfillment/fulfillment-queue.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-queue.integration.test.ts new file mode 100644 index 00000000..afb8d016 --- /dev/null +++ b/apps/backend/src/services/fulfillment/fulfillment-queue.integration.test.ts @@ -0,0 +1,188 @@ +/** + * 履约待办分桶集成测试(需要真实 PostgreSQL)。 + * + * 运行: + * FULFILLMENT_ROUTING_TEST_DATABASE_URL="$DATABASE_URL" \ + * npx tsx --test src/services/fulfillment/fulfillment-queue.integration.test.ts + * + * 未设置环境变量时跳过,普通 npm test 不依赖数据库。 + */ +import assert from 'node:assert/strict' +import test from 'node:test' + +const TEST_DATABASE_URL = String(process.env.FULFILLMENT_ROUTING_TEST_DATABASE_URL || '').trim() + +test( + '补载体扫描只处理 actionable,配置/人工处理/售后不占用扫描队列', + { skip: !TEST_DATABASE_URL ? '未设置 FULFILLMENT_ROUTING_TEST_DATABASE_URL' : false }, + async () => { + process.env.DATABASE_URL = TEST_DATABASE_URL + const { closeDb, query } = await import('../../db/client.js') + const { listPendingChannelGroups } = await import('./channel-assign-service.js') + const { reconcilePendingCarriers } = await import('./carrier-reconciliation-service.js') + const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}` + const orderIds: number[] = [] + const internalSkuIds: number[] = [] + + async function createPendingItem(input: { + sku: string + channel: 'affiliate_dash' | 'manual' | 'third_party_voucher' + carrierIssue: string + orderStatus?: string + payStatus?: string + configJson?: Record + }) { + const order = await query<{ id: number }>( + `INSERT INTO orders ( + provider, platform, shop_id, shop_name, platform_order_id, + order_status, pay_status, total_amount, paid_at, created_at, updated_at + ) VALUES ( + '91kaquan', 'kuaishou', $1, '履约队列集成测试', $2, + $3, $4, 1000, NOW(), NOW(), NOW() + ) RETURNING id`, + [ + `it-queue-shop-${suffix}`, + `IT-QUEUE-${suffix}-${orderIds.length + 1}`, + input.orderStatus || 'paid', + input.payStatus || 'paid', + ], + ) + const orderId = Number(order.rows[0]!.id) + orderIds.push(orderId) + + const item = await query<{ id: number }>( + `INSERT INTO order_items ( + order_id, sku_code, sku_name, quantity, spec_json, + revenue_amount_fen, revenue_basis, created_at, updated_at + ) VALUES ($1, $2, $2, 1, $3::jsonb, 1000, 'order_total', NOW(), NOW()) + RETURNING id`, + [orderId, input.sku, JSON.stringify({ skuNick: input.sku })], + ) + const orderItemId = Number(item.rows[0]!.id) + + const sku = await query<{ id: string }>('SELECT ensure_canonical_internal_sku($1) AS id', [ + input.sku, + ]) + const internalSkuId = Number(sku.rows[0]!.id) + internalSkuIds.push(internalSkuId) + await query( + `INSERT INTO sku_channel_routes ( + internal_sku_id, channel, route_role, priority, enabled, + carrier_config_json, source + ) VALUES ($1, $2, 'primary', 10, TRUE, $3::jsonb, 'integration_test') + ON CONFLICT (internal_sku_id, channel) DO UPDATE SET + route_role = 'primary', + priority = 10, + enabled = TRUE, + carrier_config_json = EXCLUDED.carrier_config_json, + source = 'integration_test', + updated_at = NOW()`, + [internalSkuId, input.channel, JSON.stringify(input.configJson || {})], + ) + await query( + `UPDATE fulfillments + SET status = 'pending', + channel = NULL, + source = 'auto_mapping', + locked = FALSE, + pending_reason = 'no_carrier', + pending_detail = $2::jsonb, + updated_at = NOW() + WHERE order_item_id = $1`, + [ + orderItemId, + JSON.stringify({ + internalSku: input.sku, + channel: input.channel, + carrierIssue: input.carrierIssue, + }), + ], + ) + return orderItemId + } + + try { + const actionableId = await createPendingItem({ + sku: `集成测试队列自动-${suffix}`, + channel: 'third_party_voucher', + carrierIssue: 'carrier_not_attempted', + }) + const configurationId = await createPendingItem({ + sku: `集成测试队列配置-${suffix}`, + channel: 'manual', + carrierIssue: 'manual_config_missing', + }) + const operatorReviewId = await createPendingItem({ + sku: `集成测试队列人工-${suffix}`, + channel: 'affiliate_dash', + carrierIssue: 'carrier_task_retry_pending', + configJson: { affiliateSku: `it-queue-${suffix}` }, + }) + const terminalId = await createPendingItem({ + sku: `集成测试队列售后-${suffix}`, + channel: 'affiliate_dash', + carrierIssue: 'carrier_not_attempted', + orderStatus: 'refunded', + payStatus: 'refunded', + }) + const ids = [actionableId, configurationId, operatorReviewId, terminalId] + + const states = await query<{ + id: number + queue_bucket: string + actionable_for_carrier: boolean + }>( + `SELECT id, queue_bucket, actionable_for_carrier + FROM v_fulfillment_queue + WHERE id = ANY($1::bigint[]) + ORDER BY id`, + [ids], + ) + assert.deepEqual( + states.rows.map((row) => row.queue_bucket), + ['actionable', 'configuration', 'operator_review', 'terminal'], + ) + assert.deepEqual( + states.rows.map((row) => row.actionable_for_carrier), + [true, false, false, false], + ) + + const actionableGroups = await listPendingChannelGroups({ + keyword: `集成测试队列自动-${suffix}`, + queueBucket: 'actionable', + page: 1, + pageSize: 10, + }) + assert.equal(actionableGroups.total, 1) + assert.equal(actionableGroups.items[0]?.queue_buckets[0], 'actionable') + + const configurationGroups = await listPendingChannelGroups({ + keyword: `集成测试队列配置-${suffix}`, + queueBucket: 'configuration', + page: 1, + pageSize: 10, + }) + assert.equal(configurationGroups.total, 1) + assert.deepEqual(configurationGroups.bucketCounts, { + actionable: 0, + configuration: 1, + operator_review: 0, + terminal: 0, + }) + + const reconciled = await reconcilePendingCarriers({ + orderItemIds: ids, + limit: ids.length, + }) + assert.equal(reconciled.scanned, 1) + } finally { + if (orderIds.length > 0) { + await query('DELETE FROM orders WHERE id = ANY($1::bigint[])', [orderIds]) + } + if (internalSkuIds.length > 0) { + await query('DELETE FROM internal_skus WHERE id = ANY($1::bigint[])', [internalSkuIds]) + } + await closeDb() + } + }, +) 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 123d16b6..a35d7dbc 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts @@ -20,6 +20,10 @@ const currentLatestMigration = readFileSync( new URL('../../db/migrations/145_fulfillment_current_attempt_latest.sql', import.meta.url), 'utf8', ) +const queueBucketMigration = readFileSync( + new URL('../../db/migrations/162_fulfillment_queue_buckets.sql', import.meta.url), + 'utf8', +) test('自动发货历史非活跃任务默认阻止后台自动重开', () => { assert.ok(carrierCreation.includes('INACTIVE_CARRIER_TASK_STATUSES.includes')) @@ -37,6 +41,19 @@ test('已分配但载体失效的订单会退回 pending/no_carrier', () => { assert.ok(carrierReconciliation.includes("'USER_APPLY_REFUND', 'affiliate_dash_cancelled'")) }) +test('补载体自动扫描只读取 actionable 队列', () => { + assert.ok(queueBucketMigration.includes('v_fulfillment_queue')) + assert.ok(queueBucketMigration.includes("'actionable'")) + assert.ok(queueBucketMigration.includes("'configuration'")) + assert.ok(queueBucketMigration.includes("'operator_review'")) + assert.ok(queueBucketMigration.includes("'terminal'")) + assert.ok(queueBucketMigration.includes('actionable_for_carrier')) + assert.ok(carrierReconciliation.includes("oi.queue_bucket = 'actionable'")) + assert.ok(carrierReconciliation.includes('oi.carrier_queue_candidate = TRUE')) + assert.ok(carrierReconciliation.includes('createdBefore')) + assert.ok(carrierReconciliation.includes("orderDirection === 'desc'")) +}) + test('整组补载体默认不包含锁定项,也不强制重开历史任务', () => { assert.ok(channelAssign.includes('retryInactiveCarrier?: boolean')) assert.ok(channelAssign.includes('includeLocked?: boolean')) 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 f5a281db..902d66c5 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts @@ -973,6 +973,7 @@ test( includeAssignedMissingCarrier: true, limit: 1, }) + assert.equal(blockedRetry.scanned, 0) assert.equal(blockedRetry.affiliateTasksCreated, 0) const retryPendingState = await query<{ fulfillment_status: string @@ -991,17 +992,27 @@ test( 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]?.fulfillment_status, 'assigned') + assert.equal(retryPendingState.rows[0]?.fulfillment_channel, 'affiliate_dash') + assert.equal(retryPendingState.rows[0]?.fulfillment_pending_reason, null) assert.equal(retryPendingState.rows[0]?.task_count, 1) await query( - `UPDATE fulfillments SET locked = TRUE, updated_at = NOW() WHERE order_item_id = $1`, - [retryPendingItemId], + `UPDATE fulfillments + SET status = 'pending', + channel = NULL, + pending_reason = 'no_carrier', + pending_detail = $2::jsonb, + locked = TRUE, + updated_at = NOW() + WHERE order_item_id = $1`, + [ + retryPendingItemId, + JSON.stringify({ + internalSku: retryPendingAffiliateSku, + channel: 'affiliate_dash', + carrierIssue: 'carrier_task_retry_pending', + }), + ], ) const unlockedPendingGroup = await unlockPendingChannelGroup({ internalSku: retryPendingAffiliateSku, 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 552f7109..7efab644 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-state-view.test.ts @@ -35,11 +35,13 @@ test('履约读模型按订单项拆分渠道、载体、尝试、异常和锁 }) test('待分类查询使用统一读模型并返回异常数量', () => { - assert.ok(channelAssign.includes('FROM v_fulfillment_item_state v')) + assert.ok(channelAssign.includes('FROM v_fulfillment_queue 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('FULFILLMENT_QUEUE_BUCKETS')) + assert.ok(channelAssign.includes('queueBucket')) assert.ok(channelAssign.includes('issueCategoryInput')) }) diff --git a/apps/backend/src/services/scheduler/config-service.test.ts b/apps/backend/src/services/scheduler/config-service.test.ts index 4ed9dc14..6869e7e1 100644 --- a/apps/backend/src/services/scheduler/config-service.test.ts +++ b/apps/backend/src/services/scheduler/config-service.test.ts @@ -120,6 +120,8 @@ test('normalizeScheduledJobsConfig 收录 v2 六个管道作业并保持各自 const byType = new Map(config.jobs.map((item) => [item.type, item])) assert.equal(byType.get('channel_assign')?.config.batchSize, 800) assert.equal(byType.get('carrier_dispatch')?.enabled, false) + assert.equal(byType.get('carrier_dispatch')?.config.freshWindowMinutes, 15) + assert.equal(byType.get('carrier_dispatch')?.config.freshLimit, 200) assert.equal(byType.get('sales_facts')?.config.days, 7) assert.equal(byType.get('ledger_post')?.config.days, 7) assert.ok(byType.has('product_normalize')) diff --git a/apps/backend/src/services/scheduler/config-service.ts b/apps/backend/src/services/scheduler/config-service.ts index 42963cdc..c6efeb7e 100644 --- a/apps/backend/src/services/scheduler/config-service.ts +++ b/apps/backend/src/services/scheduler/config-service.ts @@ -271,6 +271,8 @@ function normalizeCarrierDispatchJob(rawValue: JsonObject) { intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 60, 30, 3600), config: { limit: normalizeRangeInteger(config.limit, 200, 10, 2000), + freshWindowMinutes: normalizeRangeInteger(config.freshWindowMinutes, 15, 1, 120), + freshLimit: normalizeRangeInteger(config.freshLimit, 200, 10, 2000), }, } } @@ -371,7 +373,7 @@ function createDefaultCarrierDispatchJob() { type: 'carrier_dispatch', enabled: true, intervalSeconds: 60, - config: { limit: 200 }, + config: { limit: 200, freshWindowMinutes: 15, freshLimit: 200 }, } } diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index c49eaa03..c0cc3a6e 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -95,16 +95,35 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { const config = jobConfig(job) const cutoffCreatedAt = await getFulfillmentAutomationCutoff() const limit = clampInt(config.limit, 200, 1, 2000) + const freshWindowMinutes = clampInt(config.freshWindowMinutes, 15, 1, 120) + const freshLimit = clampInt(config.freshLimit, 200, 1, 2000) + const freshCutoff = new Date(Date.now() - freshWindowMinutes * 60_000).toISOString() const reconcileCursor = config.carrierReconcileAfterId != null ? Math.max(0, Math.trunc(Number(config.carrierReconcileAfterId))) : await readWatermarkItemId('carrier_reconcile') - const reconciled = await reconcilePendingCarriers({ - limit: Math.min(limit, 200), - afterId: reconcileCursor, + const freshReconciled = await reconcilePendingCarriers({ + limit: Math.min(freshLimit, 200), + createdAfter: freshCutoff, + orderDirection: 'desc', ...(cutoffCreatedAt ? { cutoffCreatedAt } : {}), }) - await touchWatermark('carrier_reconcile', { lastItemId: reconciled.nextCursor }) + const historicalReconciled = await reconcilePendingCarriers({ + limit: Math.min(limit, 200), + afterId: reconcileCursor, + createdBefore: freshCutoff, + ...(cutoffCreatedAt ? { cutoffCreatedAt } : {}), + }) + const reconciled = { + ...historicalReconciled, + scanned: freshReconciled.scanned + historicalReconciled.scanned, + workOrdersCreated: freshReconciled.workOrdersCreated + historicalReconciled.workOrdersCreated, + affiliateTasksCreated: + freshReconciled.affiliateTasksCreated + historicalReconciled.affiliateTasksCreated, + fallbacksSelected: freshReconciled.fallbacksSelected + historicalReconciled.fallbacksSelected, + failed: freshReconciled.failed + historicalReconciled.failed, + } + await touchWatermark('carrier_reconcile', { lastItemId: historicalReconciled.nextCursor }) const repairedAssigned = await reconcilePendingCarriers({ limit: Math.min(limit, 200), includeAssignedMissingCarrier: true, @@ -386,6 +405,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { `人工载体挂载 ${attached},缺配置 ${missingManual.rows.length},` + `缺履约载体 ${missingCarrier.rows.length},多载体冲突 ${conflictingCarriers.rows.length},异常收口 ${resolved},` + `自动转人工 ${fallback.switched},本轮补载体扫描 ${reconciled.scanned}、创建 ${reconciled.workOrdersCreated + reconciled.affiliateTasksCreated}、失败 ${reconciled.failed},` + + `新单优先 ${freshReconciled.scanned},` + `已分配补载体 ${repairedAssigned.workOrdersCreated + repairedAssigned.affiliateTasksCreated},` + `兜底 ${reconciled.fallbacksSelected + repairedAssigned.fallbacksSelected}`, attached, @@ -395,6 +415,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { resolved, affiliateFallback: fallback.switched, carrierReconciliation: reconciled, + freshCarrierReconciliation: freshReconciled, assignedCarrierRepair: repairedAssigned, } } diff --git a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx index fa652b83..ed6625bf 100644 --- a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx +++ b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx @@ -1,5 +1,5 @@ /** - * 履约渠道页:待分类队列 + 渠道规则 + 问题队列。 + * 履约渠道页:履约待办队列 + 渠道规则 + 问题队列。 * * 设计口径(2026-10-31 v2): * · 渠道由 L1 规则决定,不再由「有没有任务」反推; @@ -60,6 +60,7 @@ import { saveChannelRule, saveInternalSkuMapping, type ChannelRule, + type FulfillmentQueueBucket, type PendingChannelGroup, } from '@/services/admin' import { formatRedeemAmount } from '@/utils/redeem' @@ -102,14 +103,14 @@ const ENABLED_FILTER_OPTIONS = [ const MATCH_KEY_OPTIONS = [{ value: 'internal_sku', label: '内部 SKU' }] -const PENDING_ISSUE_CATEGORY_OPTIONS = [ - { value: 'all', label: '全部问题' }, - { value: 'classification', label: '未分类' }, +const PENDING_QUEUE_BUCKET_OPTIONS: Array<{ + value: FulfillmentQueueBucket + label: string +}> = [ + { value: 'actionable', label: '自动处理中' }, { value: 'configuration', label: '配置待补' }, - { value: 'carrier', label: '载体处理' }, - { value: 'manual_review', label: '人工复核' }, - { value: 'after_sales', label: '售后关闭' }, - { value: 'system', label: '系统异常' }, + { value: 'operator_review', label: '人工处理' }, + { value: 'terminal', label: '售后/终止' }, ] const INTERNAL_SKU_MAPPING_TYPES: Array<{ @@ -130,7 +131,7 @@ export default function AdminFulfillmentChannelPage() { }, + { key: 'pending', label: '履约待办', children: }, { key: 'mappings', label: 'SKU 映射', children: }, { key: 'rules', label: '渠道规则', children: }, { key: 'exceptions', label: '问题队列', children: }, @@ -146,15 +147,15 @@ function PendingPanel() { const queryClient = useQueryClient() const navigate = useNavigate() const [keyword, setKeyword] = useState('') - const [issueCategory, setIssueCategory] = useState('') + const [queueBucket, setQueueBucket] = useState('actionable') const [page, setPage] = useState(1) const [pageSize, setPageSize] = useState(ADMIN_DEFAULT_PAGE_SIZE) const [selections, setSelections] = useState>({}) const [internalSkuDrafts, setInternalSkuDrafts] = useState>({}) const query = useQuery({ - queryKey: ['fulfillment-pending', keyword, issueCategory, page, pageSize], - queryFn: () => fetchPendingChannelGroups({ page, pageSize, keyword, issueCategory }), + queryKey: ['fulfillment-pending', keyword, queueBucket, page, pageSize], + queryFn: () => fetchPendingChannelGroups({ page, pageSize, keyword, queueBucket }), }) const assignMutation = useMutation({ @@ -166,7 +167,7 @@ function PendingPanel() { ? '' : String(internalSkuDrafts[rowKey(row)] || '').trim(), channel: selections[rowKey(row)] || '', - issueCategory, + queueBucket, }), onSuccess: (result) => { const productMessage = result.data.draftProductId ? '和标准商品' : '' @@ -245,7 +246,7 @@ function PendingPanel() { unlockPendingChannelGroup({ groupKey: row.group_key, internalSku: row.internal_sku || '', - issueCategory, + queueBucket, }), onSuccess: (result) => { message.success(`已解除 ${result.data.updated} 个订单项的运营锁定`) @@ -256,7 +257,7 @@ function PendingPanel() { const deleteMutation = useMutation({ mutationFn: (row: PendingChannelGroup) => - deletePendingChannelGroup({ groupKey: row.group_key, issueCategory }), + deletePendingChannelGroup({ groupKey: row.group_key, queueBucket }), onSuccess: (result) => { const data = result.data message.success( @@ -456,7 +457,8 @@ function PendingPanel() { issueBreakdown.length > 0 && issueBreakdown.every((issue) => issue.retryable) const hasOnlyAutoRetryableIssues = issueBreakdown.length > 0 && issueBreakdown.every((issue) => issue.auto_retryable) - const canRunScopedActions = Boolean(issueCategory) || issueBreakdown.length <= 1 + const canRunScopedActions = issueBreakdown.length <= 1 + const canChangeChannel = queueBucket !== 'terminal' const hasUnlockedItems = row.locked_item_count < row.item_count return ( @@ -476,7 +478,8 @@ function PendingPanel() { disabled={ (!row.internal_sku && !String(internalSkuDrafts[rowKey(row)] || '').trim()) || !selections[rowKey(row)] || - !canRunScopedActions + !canRunScopedActions || + !canChangeChannel } loading={assignMutation.isPending} onClick={() => assignMutation.mutate(row)} @@ -607,8 +610,8 @@ function PendingPanel() { return ( option.value === (issueCategory || 'all')) - ?.label || '全部问题' + PENDING_QUEUE_BUCKET_OPTIONS.find((option) => option.value === queueBucket)?.label || + '履约待办' }(${query.data?.data?.total ?? 0} 个 SKU)`} extra={ @@ -645,15 +648,20 @@ function PendingPanel() { type="warning" showIcon style={{ marginBottom: 12 }} - message="指定渠道会先保存渠道和草稿标准商品;载体或成本未配置时会继续留在队列中。忽略归档只影响派生流程,不删除原始订单。" + message="补载体自动扫描只处理“自动处理中”。缺配置、失败重试、人工复核、售后和已终止订单分别留在对应队列,不再占用自动扫描。指定渠道会先保存渠道和草稿标准商品;忽略归档只影响派生流程,不删除原始订单。" /> ({ + value: option.value, + label: `${option.label} ${query.data?.data?.bucketCounts?.[option.value] ?? 0}`, + }))} onChange={(value) => { - setIssueCategory(value === 'all' ? '' : String(value)) + setQueueBucket(String(value) as FulfillmentQueueBucket) setPage(1) + setSelections({}) + setInternalSkuDrafts({}) }} style={{ marginBottom: 12 }} /> diff --git a/apps/frontend/src/services/admin/fulfillment-channel.ts b/apps/frontend/src/services/admin/fulfillment-channel.ts index fd6c36e1..25509b52 100644 --- a/apps/frontend/src/services/admin/fulfillment-channel.ts +++ b/apps/frontend/src/services/admin/fulfillment-channel.ts @@ -1,5 +1,7 @@ import { apiDelete, apiGet, apiPost } from '@/lib/http' +export type FulfillmentQueueBucket = 'actionable' | 'configuration' | 'operator_review' | 'terminal' + /** 待分类队列(按内部 SKU 聚合)。 */ export type PendingChannelGroup = { group_key: string @@ -12,6 +14,7 @@ export type PendingChannelGroup = { sku_codes: string[] pending_reasons: string[] pending_subreasons: string[] + queue_buckets: FulfillmentQueueBucket[] configured_channels: string[] locked_item_count: number item_count: number @@ -43,6 +46,7 @@ export type PendingChannelGroupListResult = { total: number page: number pageSize: number + bucketCounts: Record } /** 渠道规则(L1)。 */ @@ -230,6 +234,7 @@ export function fetchPendingChannelGroups( pageSize?: number keyword?: string issueCategory?: string + queueBucket?: FulfillmentQueueBucket } = {}, ) { return apiGet('/api/v1/admin/fulfillment/pending', params) @@ -302,11 +307,16 @@ export function unlockPendingChannelGroup(payload: { groupKey?: string internalSku?: string issueCategory?: string + queueBucket?: FulfillmentQueueBucket }) { return apiPost<{ updated: number }>('/api/v1/admin/fulfillment/pending/unlock', payload) } -export function deletePendingChannelGroup(payload: { groupKey: string; issueCategory?: string }) { +export function deletePendingChannelGroup(payload: { + groupKey: string + issueCategory?: string + queueBucket?: FulfillmentQueueBucket +}) { return apiPost( '/api/v1/admin/fulfillment/pending/delete', payload, @@ -319,6 +329,7 @@ export function assignChannelByInternalSku(payload: { mappedInternalSku?: string channel: string issueCategory?: string + queueBucket?: FulfillmentQueueBucket note?: string }) { return apiPost<{ diff --git a/apps/frontend/tests/fulfillment-pending-actions.test.ts b/apps/frontend/tests/fulfillment-pending-actions.test.ts index 9425317c..3eb810b6 100644 --- a/apps/frontend/tests/fulfillment-pending-actions.test.ts +++ b/apps/frontend/tests/fulfillment-pending-actions.test.ts @@ -7,7 +7,7 @@ const panel = readFileSync( 'utf8', ) -test('待分类队列按问题类型拆分动作', () => { +test('履约待办按四个工作桶拆分动作', () => { assert.ok(panel.includes('去补配置')) assert.ok(panel.includes('处理人工复核')) assert.ok(panel.includes('确认重试')) @@ -15,9 +15,12 @@ test('待分类队列按问题类型拆分动作', () => { assert.ok(panel.includes('等待补配置')) assert.ok(panel.includes('用户已申请退款')) assert.ok(panel.includes('上游履约已取消')) - assert.ok(panel.includes('PENDING_ISSUE_CATEGORY_OPTIONS')) + assert.ok(panel.includes('PENDING_QUEUE_BUCKET_OPTIONS')) + assert.ok(panel.includes("label: '自动处理中'")) assert.ok(panel.includes("label: '配置待补'")) - assert.ok(panel.includes("label: '售后关闭'")) + assert.ok(panel.includes("label: '人工处理'")) + assert.ok(panel.includes("label: '售后/终止'")) + assert.ok(panel.includes("queueBucket !== 'terminal'")) }) test('重开历史载体必须二次确认且不会处理锁定项', () => { diff --git a/docs/生产升级操作手册.md b/docs/生产升级操作手册.md index fac5797f..19d0b643 100644 --- a/docs/生产升级操作手册.md +++ b/docs/生产升级操作手册.md @@ -23,8 +23,13 @@ updated_at: 2026-11-08 > `160_consignment_auto_match_setting.sql`、`161_semi_auto_fulfillment.sql`。 > 在发布负责人确认目标提交并完成隔离恢复演练前,只作为开发基线, > 不得按旧的 `147–150` 一条路径直接上线。 +> +> 2026-11-14 增量待发布基线:在 `161` 之上新增 +> `162_fulfillment_queue_buckets.sql`,建立履约待办四桶只读投影、 +> 补载体扫描专用索引和新单优先扫描口径。该迁移不改写原始订单、履约尝试或售后事实; +> 仍须完成停写、最终备份、隔离恢复验证、迁移、健康检查和履约审计后才可恢复营业。 -### 1.1 迁移 `151 → 161` 增量门禁 +### 1.1 迁移 `151 → 162` 增量门禁 - 升级前必须确认生产最高迁移仍为 `151`;分支合并后重新核对迁移号,禁止回退或改写 `151` 及以前文件。 - 半自动迁移已收敛为单个 `161_semi_auto_fulfillment.sql`,不得在已应用 main `158–160` @@ -33,8 +38,9 @@ updated_at: 2026-11-08 最终 Connector 来源约束和兼容保留的 `semi_auto.credential.export` 权限记录; 该权限不再参与 CK 导出判定,导出由后台登录、禁止缓存和审计控制; 不迁移或清理订单、退款、资金流水等 A 级事实。 -- 迁移完成后至少验证 `semi_auto_credentials` 表、任务外键和索引存在; - 用测试库完成 `001 → 161` 空库迁移,并在隔离库验证 `151 → 161`。 +- 迁移完成后至少验证 `semi_auto_credentials` 表、任务外键和索引存在,以及 + `v_fulfillment_queue`、补载体部分索引存在; + 用测试库完成 `001 → 162` 空库迁移,并在隔离库验证 `151 → 162`。 - 该增量仍包含迁移,普通停机发布不能使用 `deploy/ubuntu-deploy.sh` 在旧后端运行时直接换库; 必须沿用本手册的停写、最终备份、隔离恢复验证、迁移、新后端健康检查和验收流程。 diff --git a/docs/统一订单履约与财务模型-权威版.md b/docs/统一订单履约与财务模型-权威版.md index 657f6238..f3edc289 100644 --- a/docs/统一订单履约与财务模型-权威版.md +++ b/docs/统一订单履约与财务模型-权威版.md @@ -3,7 +3,7 @@ type: 权威架构与实施计划 description: 订单、内部 SKU、履约路由、履约执行、商品、成本与财务的唯一权威口径及重构计划。 tags: [订单, 内部SKU, 履约, 商品, 成本, 财务, 迁移] status: authoritative -version: 2.17 +version: 2.18 updated_at: 2026-11-14 --- @@ -12,7 +12,7 @@ updated_at: 2026-11-14 > 状态:**唯一权威文档**。 > 最后校准:2026-11-14。 > 生产基线:迁移 `151`。 -> 开发基线:迁移 `161`(仅本地,尚未部署生产)。 +> 开发基线:迁移 `162`(仅本地,尚未部署生产)。 > > 从本文件生效起,订单、内部 SKU、履约渠道、履约载体、标准商品、成本和财务相关实现, > 只能以本文件为业务口径。其他旧分析只作为历史资料,不再具备设计或验收效力。 @@ -269,7 +269,30 @@ fulfillment_issues 同一订单项只能有一个当前尝试。历史尝试只追加,不覆盖。 -### 3.4.1 人工渠道半自动履约 +### 3.4.1 履约待办分桶与自动扫描 + +`v_fulfillment_queue` 是履约待办的唯一只读投影,在正式履约状态和结构化异常之上派生四个 +工作桶,不新增可写业务状态: + +```text +actionable 正常等待补载体,允许自动扫描 +configuration 缺内部 SKU、渠道、模板或人工配置 +operator_review 失败重试、人工复核、载体已关闭/取消或系统不一致 +terminal 已退款、退款申请、平台取消或商业订单已关闭 +``` + +规则: + +- 自动补载体扫描只允许读取 `v_fulfillment_queue.actionable_for_carrier = TRUE`。 +- `configuration`、`operator_review`、`terminal` 不得进入自动补载体 cursor,也不得阻塞新订单。 +- 后台履约待办必须读取同一 `v_fulfillment_queue`,四个页签分别展示对应桶。 +- 售后/终止桶不得重新指定履约渠道;配置桶只补配置,人工处理桶只由运营确认重试或复核。 +- 补载体每日扫描采用“最近窗口新单优先 + 历史 actionable cursor”两段式;新单不能排在 + 配置缺失、退款、关闭工单等非可执行记录之后。 +- 分桶是读模型,不是新的核心状态;判断依据仍为 `orders`、`fulfillments`、 + `fulfillment_attempts`、`fulfillment_issues` 和 `sku_channel_routes`。 + +### 3.4.2 人工渠道半自动履约 半自动履约不是新的履约渠道,而是 `manual` 渠道下的辅助执行域: @@ -980,10 +1003,11 @@ docs/archive/ ## 十三、当前下一步 -截至 2026-11-13,生产基线为 `151`,集成分支开发基线为 `161`。迁移 `152` 保护订单来源快照, +截至 2026-11-14,生产基线为 `151`,集成分支开发基线为 `162`。迁移 `152` 保护订单来源快照, 迁移 `158–160` 收口寄卖登记人同步、寄卖平台与自动匹配设置; 迁移 `161` 一次建立人工渠道半自动任务、内置 Connector 默认值、账号级 Cookie JSON 凭证、 -任务凭证引用和最终约束。迁移中保留的 `semi_auto.credential.export` 权限仅作兼容记录, +任务凭证引用和最终约束;迁移 `162` 建立履约待办四桶只读投影、补载体专用索引和新单优先扫描口径。 +迁移中保留的 `semi_auto.credential.export` 权限仅作兼容记录, 不再作为 CK 导出门禁。 1. 生产升级只按 [生产升级操作手册](生产升级操作手册.md) 执行。