refactor(fulfillment): 收敛配置与商品关联写入

2026-11-05 21:01:11 +08:00
parent fcd0c137d5
commit b424333a30
15 changed files with 481 additions and 182 deletions
@@ -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<AffiliateSkuBackfillRow>(
+  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<AffiliateSkuBackfillRow>(
+    `
+      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 }>(
    `
@@ -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<string, unknown> =
  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<string, unknown> =
            $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<string, unknown> =
  await attachCostVersionsForManualConfigs()

  return {
-    updatedCount: result.rowCount || 0,
+    updatedCount: attached.length,
    skuKey,
    dateFrom,
    dateTo,
@@ -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
@@ -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<number | null
  )
  if (current.rows[0]?.status !== 'active') return null

-  await query(
-    'UPDATE order_items SET canonical_product_id = $2, updated_at = NOW() WHERE id = $1 AND canonical_product_id IS NULL',
-    [item.id, productId],
-  )
+  await attachFulfillmentConfigurations({
+    mappings: [{ orderItemId: Number(item.id), canonicalProductId: productId }],
+    onlyWhenCanonicalProductNull: true,
+  })

  await query(
    `
@@ -368,17 +369,15 @@ export async function normalizeOrderItemProducts(
    }

    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)))
    }
  }

@@ -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 }
 }
@@ -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(
    `
@@ -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<string,
  let ruleMatchedCount = 0
  let ruleUpdatedCount = 0
  for (let batch = 0; batch < 40; batch += 1) {
-    const targetRows = await query<{ id: number; canonical_product_id: number }>(
+    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<string,
          ORDER BY wo.order_item_id, wo.id DESC
          LIMIT $4
        )
-        SELECT woi.id, p.id AS canonical_product_id
+        SELECT woi.id, p.id AS canonical_product_id, woi.expected_product_id
        FROM work_order_item woi
        JOIN work_product_rules wpr ON wpr.id = woi.product_rule_id
        JOIN product_catalog p ON p.canonical_sku = 'tpl:' || wpr.rule_key
@@ -95,23 +101,16 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
    lastRuleItemId = lastSelectedId
    ruleMatchedCount += targetRows.rows.length

-    const updated = await query(
-      `
-        UPDATE order_items oi
-        SET canonical_product_id = input.canonical_product_id,
-            updated_at = $1
-        FROM unnest($2::bigint[], $3::bigint[])
-          AS input(id, canonical_product_id)
-        WHERE oi.id = input.id
-          AND oi.canonical_product_id IS DISTINCT FROM input.canonical_product_id
-      `,
-      [
-        now,
-        targetRows.rows.map((row) => 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<string,
    if (lastSelectedId <= lastTailItemId) break
    lastTailItemId = lastSelectedId

-    const updated = await query(
-      `
-        UPDATE order_items oi
-        SET canonical_product_id = input.canonical_product_id,
-            updated_at = $1
-        FROM unnest($2::bigint[], $3::bigint[])
-          AS input(id, canonical_product_id)
-        WHERE oi.id = input.id
-          AND oi.canonical_product_id IS NULL
-      `,
-      [
-        now,
-        tailRows.rows.map((row) => 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
  }

@@ -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(
@@ -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<ConfirmContextRow>(
      `
@@ -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
 }

@@ -10,6 +10,7 @@ import {
  findConflictingCarrier,
  loadActiveCarrierEvidenceForItem,
 } from './carrier-evidence-service.js'
+import { attachFulfillmentConfigurations } from './fulfillment-state-service.js'

 function parseObject(value: unknown): Record<string, unknown> {
  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 {
@@ -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)
@@ -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])
      }
@@ -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 ('))
@@ -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<FulfillmentStateExpected, 'status' | 'channel'>
+    onlyWhenConfigNull?: boolean
+    onlyWhenCanonicalProductNull?: boolean
+    onlyWhenCanonicalProductDifferent?: boolean
+  },
+): Promise<OrderItemRow[]> {
+  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<OrderItemRow>(
+    `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<FulfillmentStateExpected, 'status' | 'channel'>
+  onlyWhenConfigNull?: boolean
+  onlyWhenCanonicalProductNull?: boolean
+  onlyWhenCanonicalProductDifferent?: boolean
+}): Promise<OrderItemRow[]> {
+  return withTransaction((client) =>
+    attachFulfillmentConfigurationsWithExecutor(client as FulfillmentStateExecutor, input),
+  )
+}
+
 export async function listFulfillmentStateEvents(
  orderItemId: number | string,
  options: { limit?: number } = {},
@@ -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)
@@ -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 | 正式履约表 | 建立长期可扩展模型 |