diff --git a/apps/backend/src/services/cost-center/affiliate-product-service.ts b/apps/backend/src/services/cost-center/affiliate-product-service.ts index 8ebe7d6f..47ffe772 100644 --- a/apps/backend/src/services/cost-center/affiliate-product-service.ts +++ b/apps/backend/src/services/cost-center/affiliate-product-service.ts @@ -7,16 +7,22 @@ import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js import { nowIso } from '../../utils/time.js' import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js' import { AFFILIATE_DASH_PRICES_BASELINE } from './affiliate-dash-prices-baseline.js' +import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js' /** * 清理历史上因为把商品中文名误当成 SKU 插入的非标脏商品,并将误关联的订单项重新归一到标准干净商品。 */ export async function cleanupDirtyAffiliateDashProducts() { - await query(` + const dirtyMappings = await query<{ + order_item_id: number + clean_product_id: number + expected_product_id: number + }>(` WITH dirty_map AS ( SELECT DISTINCT ON (oi.id) oi.id AS order_item_id, - clean.id AS clean_product_id + clean.id AS clean_product_id, + dirty.id AS expected_product_id FROM order_items oi JOIN product_catalog dirty ON dirty.id = oi.canonical_product_id JOIN fulfillment_tasks ft ON ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' @@ -27,11 +33,20 @@ export async function cleanupDirtyAffiliateDashProducts() { WHERE dirty.canonical_sku ~ '^aff:[^a-z0-9_]' ORDER BY oi.id, ft.id DESC ) - UPDATE order_items oi - SET canonical_product_id = dm.clean_product_id - FROM dirty_map dm - WHERE oi.id = dm.order_item_id + SELECT order_item_id, clean_product_id, expected_product_id + FROM dirty_map + ORDER BY order_item_id `) + if (dirtyMappings.rows.length > 0) { + await attachFulfillmentConfigurations({ + mappings: dirtyMappings.rows.map((row) => ({ + orderItemId: Number(row.order_item_id), + canonicalProductId: Number(row.clean_product_id), + expectedCanonicalProductId: Number(row.expected_product_id), + })), + onlyWhenCanonicalProductDifferent: true, + }) + } await query(` DELETE FROM sales_daily_facts @@ -134,8 +149,6 @@ export async function syncAdminAffiliateDashProducts() { type AffiliateSkuBackfillRow = { affiliate_item_count: number - matched_count: number - updated_count: number from_date: string | null to_date: string | null } @@ -165,8 +178,11 @@ export async function rebuildFactsForRange(fromDate: string, toDate: string) { * 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。 */ export async function backfillAdminAffiliateDashSkus() { - const now = nowIso() - const result = await query( + const targets = await query<{ + id: number + canonical_product_id: number + expected_product_id: number | null + }>( ` WITH affiliate_item AS ( SELECT DISTINCT ON (ft.order_item_id) @@ -184,30 +200,37 @@ export async function backfillAdminAffiliateDashSkus() { ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC ), target AS ( - SELECT ai.id, p.id AS canonical_product_id + SELECT + ai.id, + p.id AS canonical_product_id, + oi.canonical_product_id AS expected_product_id FROM affiliate_item ai JOIN product_catalog p ON p.canonical_sku = 'aff:' || ai.sku_code - ), - updated AS ( - UPDATE order_items oi - SET canonical_product_id = target.canonical_product_id, - updated_at = $1 - FROM target - WHERE oi.id = target.id - AND oi.canonical_product_id IS DISTINCT FROM target.canonical_product_id - RETURNING oi.id + JOIN order_items oi ON oi.id = ai.id + ) + SELECT id, canonical_product_id, expected_product_id + FROM target + ORDER BY id + `, + ) + const result = await query( + ` + WITH affiliate_item AS ( + SELECT DISTINCT ON (ft.order_item_id) + ft.order_item_id AS id + FROM fulfillment_tasks ft + WHERE ft.executor_key = 'affiliate_dash' + AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') + AND ft.created_at >= NOW() - INTERVAL '30 days' + AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> '' + ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC ) SELECT ( - SELECT COUNT(DISTINCT ft.order_item_id)::int - FROM fulfillment_tasks ft - WHERE ft.executor_key = 'affiliate_dash' - AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled') - AND ft.created_at >= NOW() - INTERVAL '30 days' + SELECT COUNT(*)::int + FROM affiliate_item ) AS affiliate_item_count, - (SELECT COUNT(*)::int FROM target) AS matched_count, - (SELECT COUNT(*)::int FROM updated) AS updated_count, ( SELECT to_char( GREATEST( @@ -230,13 +253,24 @@ export async function backfillAdminAffiliateDashSkus() { JOIN orders o ON o.id = oi.order_id ) AS to_date `, - [now], ) + const attached = + targets.rows.length > 0 + ? await attachFulfillmentConfigurations({ + mappings: targets.rows.map((row) => ({ + orderItemId: Number(row.id), + canonicalProductId: Number(row.canonical_product_id), + expectedCanonicalProductId: + row.expected_product_id === null ? null : Number(row.expected_product_id), + })), + onlyWhenCanonicalProductDifferent: true, + }) + : [] const row = result.rows[0] const affiliateItemCount = Number(row?.affiliate_item_count || 0) - const matchedCount = Number(row?.matched_count || 0) - const updatedCount = Number(row?.updated_count || 0) + const matchedCount = targets.rows.length + const updatedCount = attached.length const skippedCount = Math.max(0, matchedCount - updatedCount) const remainingResult = await query<{ total: number }>( ` 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 fa20465d..27aa86ed 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 @@ -7,6 +7,7 @@ import { nowIso } from '../../utils/time.js' import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js' import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js' import { + attachFulfillmentConfigurations, transitionFulfillmentStateBatch, transitionFulfillmentStateBatchWithExecutor, } from '../fulfillment/fulfillment-state-service.js' @@ -827,11 +828,14 @@ export async function rematchManualFulfillments(input: Record = const dateFrom = normalizeOptionalDate(input.dateFrom ?? input.date_from, '开始日期') const dateTo = normalizeOptionalDate(input.dateTo ?? input.date_to, '结束日期') - const now = nowIso() - const result = await query<{ id: number }>( + const candidates = await query<{ + order_item_id: number + config_id: number + canonical_product_id: number | null + }>( ` WITH target AS ( - SELECT + SELECT DISTINCT ON (oi.id) oi.id AS order_item_id, cfg.id AS config_id, cfg.canonical_product_id @@ -853,20 +857,27 @@ export async function rematchManualFulfillments(input: Record = $4::date IS NULL OR (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date <= $4::date ) + ORDER BY oi.id, cfg.id ) - UPDATE order_items oi - SET fulfillment_config_id = target.config_id, - canonical_product_id = COALESCE(target.canonical_product_id, oi.canonical_product_id), - updated_at = $5 + SELECT order_item_id, config_id, canonical_product_id FROM target - WHERE oi.id = target.order_item_id - RETURNING oi.id + ORDER BY order_item_id `, - [skuKey, onlyUnmatched, dateFrom, dateTo, now], + [skuKey, onlyUnmatched, dateFrom, dateTo], ) + const attached = await attachFulfillmentConfigurations({ + mappings: candidates.rows.map((row) => ({ + orderItemId: Number(row.order_item_id), + configId: Number(row.config_id), + ...(row.canonical_product_id ? { canonicalProductId: Number(row.canonical_product_id) } : {}), + })), + expected: { status: 'assigned', channel: 'manual' }, + onlyWhenConfigNull: onlyUnmatched, + onlyWhenCanonicalProductNull: true, + }) // 配置补上以后,「人工 SKU 未配置」异常要立即收口,不能等人工去点。 - const updatedIds = result.rows.map((row) => Number(row.id)) + const updatedIds = attached.map((row) => Number(row.id)) if (updatedIds.length > 0) { await resolveExceptionsForItemsByReason(updatedIds, 'manual_config_missing') } @@ -875,7 +886,7 @@ export async function rematchManualFulfillments(input: Record = await attachCostVersionsForManualConfigs() return { - updatedCount: result.rowCount || 0, + updatedCount: attached.length, skuKey, dateFrom, dateTo, diff --git a/apps/backend/src/services/cost-center/product-catalog-service.ts b/apps/backend/src/services/cost-center/product-catalog-service.ts index 617e379d..34aa9b80 100644 --- a/apps/backend/src/services/cost-center/product-catalog-service.ts +++ b/apps/backend/src/services/cost-center/product-catalog-service.ts @@ -19,6 +19,7 @@ import { setAdminAffiliateProductPrice } from './affiliate-product-service.js' import { resolvePriceEffectiveFrom } from './cost-center-normalizers.js' import { ensureVoucherRuleForSku } from '../fulfillment/channel-assign-service.js' import { syncCostSnapshotsForItems } from './cost-snapshot-service.js' +import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js' import { resolveExceptionsForItemsByReason, resolveOrphanCostExceptions, @@ -1375,19 +1376,14 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) { } if (ids.length > 0) { - const updated = await query( - ` - UPDATE order_items oi - SET canonical_product_id = input.canonical_product_id, - updated_at = NOW() - FROM unnest($1::bigint[], $2::bigint[]) - AS input(id, canonical_product_id) - WHERE oi.id = input.id - AND oi.canonical_product_id IS NULL - `, - [ids, productIds], - ) - updatedCount += updated.rowCount || 0 + const updated = await attachFulfillmentConfigurations({ + mappings: ids.map((orderItemId, index) => ({ + orderItemId, + canonicalProductId: productIds[index]!, + })), + onlyWhenCanonicalProductNull: true, + }) + updatedCount += updated.length } batches += 1 if (!all || itemsResult.rows.length < limit) break diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts index 9d0fdf09..d81a6480 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -3,6 +3,7 @@ import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exc import { readWatermarkItemId } from '../watermark-store.js' import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js' import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js' +import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js' /** * 管道 B:商品归一。 @@ -188,10 +189,10 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise 0) { - await query( - ` - UPDATE order_items oi - SET canonical_product_id = input.product_id, updated_at = NOW() - FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id) - WHERE oi.id = input.id - AND oi.canonical_product_id IS NULL - `, - [matched.map((row) => row.id), matched.map((row) => row.productId)], - ) - await resolveExceptionsForItems(matched.map((row) => row.id)) + const attached = await attachFulfillmentConfigurations({ + mappings: matched.map((row) => ({ + orderItemId: row.id, + canonicalProductId: row.productId, + })), + expected: { status: 'assigned' }, + onlyWhenCanonicalProductNull: true, + }) + await resolveExceptionsForItems(attached.map((row) => Number(row.id))) } } @@ -508,17 +507,16 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{ matched.push({ id: item.id, productId }) } if (matched.length > 0) { - await query( - ` - UPDATE order_items oi - SET canonical_product_id = input.product_id, updated_at = NOW() - FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id) - WHERE oi.id = input.id - AND oi.canonical_product_id IS NULL - `, - [matched.map((row) => row.id), matched.map((row) => row.productId)], - ) - await resolveExceptionsForItems(matched.map((row) => row.id)) + const attached = await attachFulfillmentConfigurations({ + mappings: matched.map((row) => ({ + orderItemId: row.id, + canonicalProductId: row.productId, + })), + expected: { status: 'assigned' }, + onlyWhenCanonicalProductNull: true, + }) + await resolveExceptionsForItems(attached.map((row) => Number(row.id))) + return { matched: attached.length, unmatched } } - return { matched: matched.length, unmatched } + return { matched: 0, unmatched } } diff --git a/apps/backend/src/services/cost-center/template-bootstrap-service.ts b/apps/backend/src/services/cost-center/template-bootstrap-service.ts index 7f4c2a7b..5b53afdf 100644 --- a/apps/backend/src/services/cost-center/template-bootstrap-service.ts +++ b/apps/backend/src/services/cost-center/template-bootstrap-service.ts @@ -22,6 +22,7 @@ import { buildTemplateProductSku, inferTemplateUnitName, } from './work-order-template-product-service.js' +import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js' export { buildTemplateProductSku, inferTemplateUnitName } @@ -621,10 +622,10 @@ function compareMappingCandidates(left: TemplateMappingCandidate, right: Templat async function resetImportedTemplateProductBootstrap(range: { fromDate: string; toDate: string }) { let resetOrderItemCount = 0 for (let batch = 0; batch < 20; batch += 1) { - const result = await query<{ id: number }>( + const candidates = await query<{ id: number; expected_product_id: number }>( ` WITH candidates AS ( - SELECT oi.id + SELECT oi.id, oi.canonical_product_id AS expected_product_id FROM order_items oi JOIN orders o ON o.id = oi.order_id JOIN channel_product_mappings m @@ -655,17 +656,22 @@ async function resetImportedTemplateProductBootstrap(range: { fromDate: string; BETWEEN $1::date AND $2::date LIMIT 20000 ) - UPDATE order_items oi - SET canonical_product_id = NULL, - updated_at = NOW() - FROM candidates c - WHERE oi.id = c.id - RETURNING oi.id + SELECT DISTINCT id, expected_product_id + FROM candidates + ORDER BY id `, [range.fromDate, range.toDate], ) - resetOrderItemCount += result.rowCount || 0 - if ((result.rowCount || 0) < 20000) break + if (candidates.rows.length === 0) break + const reset = await attachFulfillmentConfigurations({ + mappings: candidates.rows.map((row) => ({ + orderItemId: Number(row.id), + canonicalProductId: null, + expectedCanonicalProductId: Number(row.expected_product_id), + })), + }) + resetOrderItemCount += reset.length + if (candidates.rows.length < 20000) break } const deleted = await query( ` diff --git a/apps/backend/src/services/cost-center/work-order-product-service.ts b/apps/backend/src/services/cost-center/work-order-product-service.ts index 4317bc33..5ab02b77 100644 --- a/apps/backend/src/services/cost-center/work-order-product-service.ts +++ b/apps/backend/src/services/cost-center/work-order-product-service.ts @@ -6,6 +6,7 @@ import { resolveFinanceDateRange } from '../../utils/finance-date-range.js' import { nowIso } from '../../utils/time.js' import { rebuildFactsForRange } from './affiliate-product-service.js' import { WORK_ORDER_TEMPLATE_COST_BASELINE } from './cost-center-shared.js' +import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js' type WorkOrderSkuBackfillRow = { work_order_item_count: number @@ -64,12 +65,17 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record( + const targetRows = await query<{ + id: number + canonical_product_id: number + expected_product_id: number | null + }>( ` WITH work_order_item AS ( SELECT DISTINCT ON (wo.order_item_id) wo.order_item_id AS id, - wo.product_rule_id + wo.product_rule_id, + oi.canonical_product_id AS expected_product_id FROM work_orders wo JOIN order_items oi ON oi.id = wo.order_item_id JOIN orders o ON o.id = oi.order_id @@ -81,7 +87,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record Number(row.id)), - targetRows.rows.map((row) => Number(row.canonical_product_id)), - ], - ) - ruleUpdatedCount += updated.rowCount || 0 + const updated = await attachFulfillmentConfigurations({ + mappings: targetRows.rows.map((row) => ({ + orderItemId: Number(row.id), + canonicalProductId: Number(row.canonical_product_id), + expectedCanonicalProductId: + row.expected_product_id === null ? null : Number(row.expected_product_id), + })), + onlyWhenCanonicalProductDifferent: true, + }) + ruleUpdatedCount += updated.length if (targetRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break } @@ -249,23 +248,14 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record Number(row.id)), - tailRows.rows.map((row) => Number(row.canonical_product_id)), - ], - ) - tailUpdatedCount += updated.rowCount || 0 + const updated = await attachFulfillmentConfigurations({ + mappings: tailRows.rows.map((row) => ({ + orderItemId: Number(row.id), + canonicalProductId: Number(row.canonical_product_id), + })), + onlyWhenCanonicalProductNull: true, + }) + tailUpdatedCount += updated.length if (tailRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break } diff --git a/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts b/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts index aa428900..cf58fe96 100644 --- a/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts +++ b/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts @@ -105,7 +105,8 @@ test('人工确认前必须自动匹配履约配置,配置保存/定时任务 ), ]) assert.ok( - manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId, now)'), + manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId)') && + manualSource.includes('attachFulfillmentConfigurationsWithExecutor(client'), '确认时必须按 SKU 自动挂渠道', ) assert.ok( diff --git a/apps/backend/src/services/finance/manual-fulfillment-service.ts b/apps/backend/src/services/finance/manual-fulfillment-service.ts index 6b349768..315bbeaa 100644 --- a/apps/backend/src/services/finance/manual-fulfillment-service.ts +++ b/apps/backend/src/services/finance/manual-fulfillment-service.ts @@ -17,7 +17,10 @@ import { import { rebuildChannelSalesDailyFacts } from './sales-facts-channel-service.js' import { syncFinanceLedgerFromFacts } from './finance-ledger-service.js' import { scheduleVoucherConsumeForOrderItems } from '../fulfillment/voucher-consume-queue-service.js' -import { transitionFulfillmentStateWithExecutor } from '../fulfillment/fulfillment-state-service.js' +import { + attachFulfillmentConfigurationsWithExecutor, + transitionFulfillmentStateWithExecutor, +} from '../fulfillment/fulfillment-state-service.js' import { calculateManualCostFen, normalizeManualCreditQuantity, @@ -308,7 +311,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm const result = await withTransaction(async (client) => { // 配置可能晚于订单创建:确认前按 SKU 再匹配一次(与定时 rematch 同口径), // 保证「配置好之后的新单」不需要人工再点一次「重新匹配」。 - await matchManualFulfillmentConfigWithClient(client, orderItemId, now) + await matchManualFulfillmentConfigWithClient(client, orderItemId) const context = await client.query( ` @@ -791,11 +794,7 @@ async function resolveCreditOwnerAdminUserId( * 与 `rematchManualFulfillments` 使用同一套键:规范中文内部 SKU。 * 这样即使配置晚于订单创建,新单在确认时也能自动挂上渠道。 */ -async function matchManualFulfillmentConfigWithClient( - client: PoolClient, - orderItemId: number, - now: string, -) { +async function matchManualFulfillmentConfigWithClient(client: PoolClient, orderItemId: number) { const matched = await client.query<{ id: number canonical_product_id: number | null @@ -826,17 +825,19 @@ async function matchManualFulfillmentConfigWithClient( ) const match = matched.rows[0] if (!match) return false - await client.query( - ` - UPDATE order_items - SET fulfillment_config_id = $2, - canonical_product_id = COALESCE($3, canonical_product_id), - updated_at = $4 - WHERE id = $1 - AND fulfillment_config_id IS NULL - `, - [orderItemId, match.id, match.canonical_product_id, now], - ) + await attachFulfillmentConfigurationsWithExecutor(client, { + mappings: [ + { + orderItemId, + configId: Number(match.id), + ...(match.canonical_product_id + ? { canonicalProductId: Number(match.canonical_product_id) } + : {}), + }, + ], + onlyWhenConfigNull: true, + onlyWhenCanonicalProductNull: true, + }) return true } diff --git a/apps/backend/src/services/fulfillment/carrier-creation-service.ts b/apps/backend/src/services/fulfillment/carrier-creation-service.ts index 0b82679d..5a1e2bf0 100644 --- a/apps/backend/src/services/fulfillment/carrier-creation-service.ts +++ b/apps/backend/src/services/fulfillment/carrier-creation-service.ts @@ -10,6 +10,7 @@ import { findConflictingCarrier, loadActiveCarrierEvidenceForItem, } from './carrier-evidence-service.js' +import { attachFulfillmentConfigurations } from './fulfillment-state-service.js' function parseObject(value: unknown): Record { if (value && typeof value === 'object' && !Array.isArray(value)) { @@ -200,14 +201,18 @@ export async function ensureCarrierForOrderItem(input: { ) const row = config.rows[0] if (row) { - await query( - `UPDATE order_items - SET fulfillment_config_id = $2, - canonical_product_id = COALESCE(canonical_product_id, $3), - updated_at = NOW() - WHERE id = $1`, - [item.id, row.id, row.canonical_product_id], - ) + await attachFulfillmentConfigurations({ + mappings: [ + { + orderItemId: Number(item.id), + configId: Number(row.id), + ...(row.canonical_product_id + ? { canonicalProductId: Number(row.canonical_product_id) } + : {}), + }, + ], + onlyWhenCanonicalProductNull: true, + }) return { ready: true, created: false, reason: 'manual_config_ready', message: '' } } return { diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 6fa2433a..704a549a 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -37,6 +37,7 @@ import { canonicalizeInternalSku, containsCjk } from './internal-sku-resolver.js import { scheduleVoucherConsumeForOrderItems } from './voucher-consume-queue-service.js' import { assignFulfillmentChannelsWithExecutor, + attachFulfillmentConfigurations, transitionFulfillmentState, transitionFulfillmentStateBatch, transitionFulfillmentStateBatchWithExecutor, @@ -1421,20 +1422,39 @@ export async function retryPendingCarrierGroup(input: { assignedItemIds.push(...applied.assignedItemIds) } - await query( - `UPDATE order_items oi - SET fulfillment_config_id = cfg.id, - canonical_product_id = COALESCE(oi.canonical_product_id, cfg.canonical_product_id), - updated_at = NOW() - FROM manual_fulfillment_configs cfg + const manualConfigs = await query<{ + order_item_id: number + config_id: number + canonical_product_id: number | null + }>( + `SELECT DISTINCT ON (oi.id) + oi.id AS order_item_id, + cfg.id AS config_id, + cfg.canonical_product_id + FROM order_items oi + JOIN manual_fulfillment_configs cfg + ON canonicalize_internal_sku(cfg.sku_key) = + resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) + AND cfg.enabled = TRUE WHERE oi.id = ANY($1::bigint[]) AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' - AND cfg.enabled = TRUE - AND canonicalize_internal_sku(cfg.sku_key) = - resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)`, + ORDER BY oi.id, cfg.id`, [ids], ) + if (manualConfigs.rows.length > 0) { + await attachFulfillmentConfigurations({ + mappings: manualConfigs.rows.map((row) => ({ + orderItemId: Number(row.order_item_id), + configId: Number(row.config_id), + ...(row.canonical_product_id + ? { canonicalProductId: Number(row.canonical_product_id) } + : {}), + })), + expected: { status: 'assigned', channel: 'manual' }, + onlyWhenCanonicalProductNull: true, + }) + } if (assignedItemIds.length > 0) { await normalizeProductsForItems(assignedItemIds).catch(() => undefined) const cost = await syncCostSnapshotsForItems(assignedItemIds).catch(() => null) diff --git a/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts index 64ab345a..3707b316 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts @@ -17,6 +17,7 @@ test( const { closeDb, query } = await import('../../db/client.js') const { assignFulfillmentChannelsWithExecutor, + attachFulfillmentConfigurations, listFulfillmentStateEvents, runWithFulfillmentStateTransaction, transitionFulfillmentState, @@ -24,6 +25,8 @@ test( const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}` let orderId = 0 let orderItemId = 0 + let manualConfigId = 0 + let productId = 0 try { const order = await query<{ id: number }>( @@ -143,6 +146,53 @@ test( assert.equal(events[0]?.previous_channel, null) assert.equal(events[0]?.next_channel, 'third_party_voucher') + const product = await query<{ id: number }>( + `INSERT INTO product_catalog ( + canonical_sku, canonical_name, category, unit_name, status, + attributes_json, created_at, updated_at + ) VALUES ($1, $2, '测试', '件', 'active', '{}'::jsonb, NOW(), NOW()) + RETURNING id`, + [`test:ff-state-${suffix}`, `状态事件商品-${suffix}`], + ) + productId = Number(product.rows[0]!.id) + const config = await query<{ id: number }>( + `INSERT INTO manual_fulfillment_configs ( + sku_key, canonical_product_id, platform_key, credit_quantity, + cost_mode, enabled, note, created_at, updated_at + ) VALUES ($1, $2, 'nanhang', 1, 'fixed', TRUE, 'integration_test', NOW(), NOW()) + RETURNING id`, + [`状态事件测试-${suffix}`, productId], + ) + manualConfigId = Number(config.rows[0]!.id) + const staleConfig = await attachFulfillmentConfigurations({ + mappings: [ + { + orderItemId, + configId: manualConfigId, + canonicalProductId: productId, + expectedCanonicalProductId: manualConfigId, + }, + ], + onlyWhenCanonicalProductNull: true, + }) + assert.equal(staleConfig.length, 0, '原商品值已变化时不能覆盖并发结果') + + const configured = await attachFulfillmentConfigurations({ + mappings: [ + { + orderItemId, + configId: manualConfigId, + canonicalProductId: productId, + }, + ], + expected: { status: 'assigned', channel: 'third_party_voucher' }, + onlyWhenConfigNull: true, + onlyWhenCanonicalProductNull: true, + }) + assert.equal(configured.length, 1) + assert.equal(Number(configured[0]?.fulfillment_config_id), manualConfigId) + assert.equal(Number(configured[0]?.canonical_product_id), productId) + await assert.rejects( query(`UPDATE fulfillment_state_events SET reason = 'tampered' WHERE order_item_id = $1`, [ orderItemId, @@ -153,6 +203,12 @@ test( if (orderItemId > 0) { await query(`DELETE FROM order_items WHERE id = $1`, [orderItemId]) } + if (manualConfigId > 0) { + await query(`DELETE FROM manual_fulfillment_configs WHERE id = $1`, [manualConfigId]) + } + if (productId > 0) { + await query(`DELETE FROM product_catalog WHERE id = $1`, [productId]) + } if (orderId > 0) { await query(`DELETE FROM orders WHERE id = $1`, [orderId]) } diff --git a/apps/backend/src/services/fulfillment/fulfillment-state-service.test.ts b/apps/backend/src/services/fulfillment/fulfillment-state-service.test.ts index ae309f9d..fa808a78 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-state-service.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-state-service.test.ts @@ -11,6 +11,15 @@ const stateService = read('./fulfillment-state-service.ts') const channelAssign = read('./channel-assign-service.ts') const carrierReconciliation = read('./carrier-reconciliation-service.ts') const affiliateFallback = read('./affiliate-fallback-service.ts') +const carrierCreation = read('./carrier-creation-service.ts') +const manualFulfillment = read('../finance/manual-fulfillment-service.ts') +const manualConfig = read('../cost-center/manual-fulfillment-config-service.ts') +const productNormalize = read('../cost-center/product-normalize-service.ts') +const productCatalog = read('../cost-center/product-catalog-service.ts') +const affiliateProduct = read('../cost-center/affiliate-product-service.ts') +const workOrderProduct = read('../cost-center/work-order-product-service.ts') +const templateBootstrap = read('../cost-center/template-bootstrap-service.ts') +const schedulerJobs = read('../scheduler/finance-pipeline-jobs.ts') const taskRepo = read('../../repositories/task-repo.ts') test('履约状态事件表记录前后状态并禁止应用修改', () => { @@ -40,6 +49,26 @@ test('高风险履约流程统一通过状态服务写入', () => { } }) +test('履约配置和标准商品关联统一通过状态服务的映射接口写入', () => { + for (const source of [ + carrierCreation, + manualFulfillment, + manualConfig, + productNormalize, + productCatalog, + affiliateProduct, + workOrderProduct, + templateBootstrap, + schedulerJobs, + ]) { + assert.ok(source.includes('fulfillment-state-service')) + assert.doesNotMatch(source, /UPDATE\s+order_items\b[\s\S]{0,300}?\bfulfillment_config_id\s*=/i) + assert.doesNotMatch(source, /UPDATE\s+order_items\b[\s\S]{0,300}?\bcanonical_product_id\s*=/i) + } + assert.ok(stateService.includes('attachFulfillmentConfigurationsWithExecutor')) + assert.ok(stateService.includes('onlyWhenCanonicalProductNull')) +}) + test('任务创建先锁订单项并复用已有活跃尝试', () => { assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE')) assert.ok(taskRepo.includes('task_status NOT IN (')) diff --git a/apps/backend/src/services/fulfillment/fulfillment-state-service.ts b/apps/backend/src/services/fulfillment/fulfillment-state-service.ts index a9fec2e7..2e04a758 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-state-service.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-state-service.ts @@ -86,6 +86,13 @@ export type FulfillmentStateEventRow = { created_at: string } +export type FulfillmentConfigurationMapping = { + orderItemId: number + configId?: number | null + canonicalProductId?: number | null + expectedCanonicalProductId?: number | null +} + type StateColumn = { column: string value: unknown @@ -430,6 +437,132 @@ export async function assignFulfillmentChannelsWithExecutor( } } +export async function attachFulfillmentConfigurationsWithExecutor( + executor: FulfillmentStateExecutor, + input: { + mappings: FulfillmentConfigurationMapping[] + expected?: Pick + onlyWhenConfigNull?: boolean + onlyWhenCanonicalProductNull?: boolean + onlyWhenCanonicalProductDifferent?: boolean + }, +): Promise { + const mappings = (input.mappings || []).map((row) => { + const setConfig = row.configId !== undefined + const setProduct = row.canonicalProductId !== undefined + const requireExpectedProduct = row.expectedCanonicalProductId !== undefined + if (!setConfig && !setProduct) { + throw new TypeError( + 'fulfillment configuration mapping must set configId or canonicalProductId', + ) + } + return { + orderItemId: normalizeId(row.orderItemId), + setConfig, + configId: + row.configId === null || row.configId === undefined ? null : normalizeId(row.configId), + setProduct, + canonicalProductId: + row.canonicalProductId === null || row.canonicalProductId === undefined + ? null + : normalizeId(row.canonicalProductId), + requireExpectedProduct, + expectedCanonicalProductId: + row.expectedCanonicalProductId === null || row.expectedCanonicalProductId === undefined + ? null + : normalizeId(row.expectedCanonicalProductId), + } + }) + if (mappings.length === 0) return [] + if (new Set(mappings.map((row) => row.orderItemId)).size !== mappings.length) { + throw new TypeError('fulfillment configuration mappings must contain unique orderItemId') + } + + const params: unknown[] = [ + mappings.map((row) => row.orderItemId), + mappings.map((row) => row.setConfig), + mappings.map((row) => row.configId), + mappings.map((row) => row.setProduct), + mappings.map((row) => row.canonicalProductId), + mappings.map((row) => row.requireExpectedProduct), + mappings.map((row) => row.expectedCanonicalProductId), + ] + const where = ['item.id = input.order_item_id'] + if (input.expected?.status !== undefined) { + params.push(input.expected.status) + where.push(`item.fulfillment_status IS NOT DISTINCT FROM $${params.length}`) + } + if (input.expected?.channel !== undefined) { + params.push(input.expected.channel) + where.push(`item.fulfillment_channel IS NOT DISTINCT FROM $${params.length}`) + } + if (input.onlyWhenConfigNull) { + where.push('item.fulfillment_config_id IS NULL') + } + if (input.onlyWhenCanonicalProductDifferent) { + where.push( + '(NOT input.set_product OR item.canonical_product_id IS DISTINCT FROM input.canonical_product_id)', + ) + } + where.push( + '(NOT input.require_expected_product ' + + 'OR item.canonical_product_id IS NOT DISTINCT FROM input.expected_canonical_product_id)', + ) + const canonicalProductIfNull = input.onlyWhenCanonicalProductNull === true + params.push(canonicalProductIfNull) + const canonicalProductIfNullParam = params.length + params.push(new Date().toISOString()) + const updatedAtParam = params.length + + const result = await executor.query( + `UPDATE order_items item + SET fulfillment_config_id = CASE + WHEN input.set_config THEN input.config_id + ELSE item.fulfillment_config_id + END, + canonical_product_id = CASE + WHEN NOT input.set_product THEN item.canonical_product_id + WHEN $${canonicalProductIfNullParam}::boolean + THEN COALESCE(item.canonical_product_id, input.canonical_product_id) + ELSE input.canonical_product_id + END, + updated_at = $${updatedAtParam}::timestamptz + FROM unnest( + $1::bigint[], + $2::boolean[], + $3::bigint[], + $4::boolean[], + $5::bigint[], + $6::boolean[], + $7::bigint[] + ) AS input( + order_item_id, + set_config, + config_id, + set_product, + canonical_product_id, + require_expected_product, + expected_canonical_product_id + ) + WHERE ${where.join(' AND ')} + RETURNING item.*`, + params, + ) + return result.rows +} + +export async function attachFulfillmentConfigurations(input: { + mappings: FulfillmentConfigurationMapping[] + expected?: Pick + onlyWhenConfigNull?: boolean + onlyWhenCanonicalProductNull?: boolean + onlyWhenCanonicalProductDifferent?: boolean +}): Promise { + return withTransaction((client) => + attachFulfillmentConfigurationsWithExecutor(client as FulfillmentStateExecutor, input), + ) +} + export async function listFulfillmentStateEvents( orderItemId: number | string, options: { limit?: number } = {}, diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 590a594d..2a0dfd62 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -3,6 +3,7 @@ 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 { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js' import { fulfillmentCreatedAfterSql, getFulfillmentAutomationCutoff, @@ -149,16 +150,16 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { ) let attached = 0 - for (const row of manual.rows) { - const updated = await query( - ` - UPDATE order_items - SET fulfillment_config_id = $2, updated_at = NOW() - WHERE id = $1 AND fulfillment_config_id IS NULL - `, - [row.id, row.config_id], - ) - attached += updated.rowCount || 0 + if (manual.rows.length > 0) { + const configured = await attachFulfillmentConfigurations({ + mappings: manual.rows.map((row) => ({ + orderItemId: Number(row.id), + configId: Number(row.config_id), + })), + expected: { status: 'assigned', channel: 'manual' }, + onlyWhenConfigNull: true, + }) + attached = configured.length } // 人工渠道但没有配置:进「人工 SKU 未配置」队列。 @@ -401,15 +402,20 @@ async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promis // 游标会跳过已扫过的订单项;配置是后来补的,这里兜底把配置挂上。 const attachParams: unknown[] = [] const attachCutoff = fulfillmentCreatedAfterSql(attachParams, cutoffCreatedAt) - await query( + const attachCandidates = await query<{ order_item_id: number; config_id: number }>( ` - UPDATE order_items oi - SET fulfillment_config_id = cfg.id, updated_at = NOW() - FROM manual_fulfillment_configs cfg + SELECT DISTINCT ON (oi.id) + oi.id AS order_item_id, + cfg.id AS config_id + FROM order_items oi + JOIN manual_fulfillment_configs cfg + ON canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku( + oi.manual_sku_key, oi.sku_name, oi.sku_code + ) + AND cfg.enabled WHERE canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku( oi.manual_sku_key, oi.sku_name, oi.sku_code ) - AND cfg.enabled AND oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL @@ -422,9 +428,21 @@ async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promis AND ce.status <> 'resolved' ) ${attachCutoff} + ORDER BY oi.id, cfg.id + LIMIT 500 `, attachParams, ) + if (attachCandidates.rows.length > 0) { + await attachFulfillmentConfigurations({ + mappings: attachCandidates.rows.map((row) => ({ + orderItemId: Number(row.order_item_id), + configId: Number(row.config_id), + })), + expected: { status: 'assigned', channel: 'manual' }, + onlyWhenConfigNull: true, + }) + } const carrierParams: unknown[] = [] const carrierCutoff = fulfillmentCreatedAfterSql(carrierParams, cutoffCreatedAt) diff --git a/docs/履约系统架构复盘与优化方案-2026-11-05.md b/docs/履约系统架构复盘与优化方案-2026-11-05.md index 8e206a0d..3c77f5a1 100644 --- a/docs/履约系统架构复盘与优化方案-2026-11-05.md +++ b/docs/履约系统架构复盘与优化方案-2026-11-05.md @@ -28,9 +28,10 @@ status: draft - 已完成:迁移 `140` 新增只追加的 `fulfillment_state_events`;订单项履约核心字段每次真实变化都由触发器留痕,遗留直写至少记录为 `system` 事件。 - 已完成:新增 `FulfillmentStateService`,履约状态写入先锁订单项、校验期望状态,再统一更新并由事件触发器记录前后值、操作者和原因。 - 已完成:自动判渠、补载体、自动发货兜底、人工确认、人工分类移除、后台锁定/解锁/归档和渠道切换均接入统一状态写入口。 +- 已完成:人工履约配置、标准商品关联、自动发货/接单平台商品回填、模板重置和调度补挂统一通过状态映射接口写入,并校验并发期间的原商品值。 - 已完成:任务创建先锁订单项并复用已有活跃尝试,避免并发调度重复插入同一当前尝试。 - 已完成:新增真实 PostgreSQL 状态事件集成测试;全新库 `001 → 140` 迁移和既有履约路由集成测试通过。 -- 待继续:正式 `fulfillments / attempts / issues` 表、配置关联类直写继续收敛、独立售后队列和生产存量修复。 +- 待继续:正式 `fulfillments / attempts / issues` 表、独立售后队列和生产存量修复。 ### 历史数据边界 @@ -693,7 +694,7 @@ FulfillmentStateService | P0 | 混合组动作收敛 | 防止错误重试和跨范围操作 | | P0 | 退款/取消隔离 | 商业终态不再进入履约队列 | | P0 | 统一读模型 | 让页面和统计使用同一事实 | -| P1 | 履约写入口 | 核心状态已收敛;配置关联字段继续收敛 | +| P1 | 履约写入口 | 核心状态与配置/商品关联均已收敛 | | P1 | 队列拆分 | 让不同角色处理不同类型问题 | | P1 | 当前尝试指针 | 历史尝试不再看起来像当前任务 | | P2 | 正式履约表 | 建立长期可扩展模型 |