refactor(cost-center): 收敛人工 SKU 自动创建流程

2026-11-06 16:37:29 +08:00
parent eef4cc1673
commit 974044a4c6
17 changed files with 601 additions and 731 deletions
@@ -1,7 +1,7 @@
 /**
 * 成本中心 · 商品与映射路由。
 *
- * 覆盖：接单模板初始化、自动发货/接单平台商品同步与回填、人工商品归一、
+ * 覆盖：接单模板初始化、自动发货/接单平台商品同步与回填、人工商品自动投影、
 * 标准商品 CRUD、渠道映射 CRUD、未归一队列、归一数量回填。
 */
 import { Router } from 'express'
@@ -9,8 +9,6 @@ import { Router } from 'express'
 import { ensureVoucherChannelRules } from '../../services/fulfillment/channel-assign-service.js'

 import {
-  archiveHistoricalManualProducts,
-  archiveSupersededManualProducts,
  backfillAdminNormalizedQuantities,
  backfillAdminProductMappings,
  createAdminChannelMapping,
@@ -164,38 +162,6 @@ router.post(
  }),
 )

-router.post(
-  '/cost-center/products/archive-historical-manual',
-  requireAdminPermission('finance.config'),
-  createJsonHandler((req) => archiveHistoricalManualProducts(bodyOf(req)), {
-    successMessage: '历史人工商品已归档',
-    errorMessage: '归档历史人工商品失败',
-    scope: '[admin/cost-center/products/archive-historical-manual]',
-    audit: (_req, data) => ({
-      action: 'cost_historical_manual_products_archived',
-      targetType: 'product_catalog',
-      targetId: '',
-      data: data as Record<string, unknown>,
-    }),
-  }),
-)
-
-router.post(
-  '/cost-center/products/archive-superseded-manual',
-  requireAdminPermission('finance.config'),
-  createJsonHandler((req) => archiveSupersededManualProducts(bodyOf(req)), {
-    successMessage: '重复人工归一商品已归档',
-    errorMessage: '归档重复人工归一商品失败',
-    scope: '[admin/cost-center/products/archive-superseded-manual]',
-    audit: (_req, data) => ({
-      action: 'cost_superseded_manual_products_archived',
-      targetType: 'product_catalog',
-      targetId: '',
-      data: data as Record<string, unknown>,
-    }),
-  }),
-)
-
 router.patch(
  '/cost-center/products/:productId',
  requireAdminPermission('finance.config'),
@@ -34,28 +34,12 @@ export function productSourceOf(canonicalSku: string): string {
 }

 /** 商品来源筛选的 SQL 片段：按 canonical_sku 前缀过滤。 */
-export function productSourceFilterSql(
-  paramIndex: number,
-  column = 'p.canonical_sku',
-  nameColumn = 'p.canonical_name',
-): string {
+export function productSourceFilterSql(paramIndex: number, column = 'p.canonical_sku'): string {
  return `(
    ($${paramIndex} = 'affiliate_dash' AND ${column} LIKE 'aff:%')
    OR ($${paramIndex} = 'work_order' AND (${column} LIKE 'tpl:%' OR ${column} LIKE 'wo:%'))
    OR ($${paramIndex} = 'third_party_voucher' AND ${column} LIKE 'tpv:%')
-    OR (
-      $${paramIndex} = 'manual'
-      AND ${column} LIKE 'man:%'
-      AND NOT EXISTS (
-        SELECT 1
-        FROM internal_sku_channels live_channel
-        WHERE live_channel.enabled = TRUE
-          AND live_channel.is_fallback = FALSE
-          AND live_channel.channel <> 'manual'
-          AND canonicalize_internal_sku(live_channel.internal_sku)
-              = canonicalize_internal_sku(${nameColumn})
-      )
-    )
+    OR ($${paramIndex} = 'manual' AND ${column} LIKE 'man:%')
  )`
 }

@@ -6,6 +6,7 @@
 *   2. 成本项统计按 cost_model 分组，不再读 cost_items.carrier_type。
 */
 import { query } from '../../db/client.js'
+import { productSourceFilterSql } from './cost-center-mappers.js'
 import { PROVIDERS, productSourceOf, toNumber } from './cost-center-shared.js'

 export type CostCenterScope = 'affiliate_dash' | 'work_order' | 'manual'
@@ -37,25 +38,7 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
        SELECT p.id
        FROM product_catalog p
        WHERE p.status = 'active'
-          AND (
-            $1::text IS NULL
-            OR ($1 = 'affiliate_dash' AND p.canonical_sku LIKE 'aff:%')
-            OR ($1 = 'work_order' AND (p.canonical_sku LIKE 'tpl:%' OR p.canonical_sku LIKE 'wo:%'))
-            OR ($1 = 'third_party_voucher' AND p.canonical_sku LIKE 'tpv:%')
-            OR (
-              $1 = 'manual'
-              AND p.canonical_sku LIKE 'man:%'
-              AND NOT EXISTS (
-                SELECT 1
-                FROM internal_sku_channels live_channel
-                WHERE live_channel.enabled = TRUE
-                  AND live_channel.is_fallback = FALSE
-                  AND live_channel.channel <> 'manual'
-                  AND canonicalize_internal_sku(live_channel.internal_sku)
-                      = canonicalize_internal_sku(p.canonical_name)
-              )
-            )
-          )
+          AND ($1::text IS NULL OR ${productSourceFilterSql(1)})
      ),
      scoped_cost_models AS (
        SELECT CASE $1
@@ -8,7 +8,7 @@ import {
  resolveMappingNameKey,
 } from './cost-center-shared.js'
 import { calculateCostTotal, validateCostVersionScope } from './cost-resolution-service.js'
-import { resolveMappingProductId } from './product-catalog-service.js'
+import { createAdminProduct, resolveMappingProductId } from './product-catalog-service.js'
 import { productSourceFilterSql } from './cost-center-mappers.js'
 import {
  buildTemplateMappingPlan,
@@ -310,14 +310,11 @@ test('第三方卡券删除：无历史引用硬删除，有历史引用自动
  assert.ok(routeSource.includes("'/cost-center/voucher-products/:productId'"))
 })

-test('人工客服筛选排除已有其他主渠道的旧 man 商品', () => {
+test('人工客服筛选保留所有可维护的 man 商品和显式兜底配置', () => {
  const sql = productSourceFilterSql(1)
  assert.ok(sql.includes(`$1 = 'manual'`))
-  assert.ok(sql.includes("live_channel.channel <> 'manual'"))
-  assert.ok(
-    sql.includes('canonicalize_internal_sku(live_channel.internal_sku)') &&
-      sql.includes('canonicalize_internal_sku(p.canonical_name)'),
-  )
+  assert.ok(sql.includes("LIKE 'man:%'"))
+  assert.ok(!sql.includes('live_channel'))
 })

 test('成本中心总览异常统计必须走异常表关联，不能全表扫描订单项子查询', async () => {
@@ -328,6 +325,7 @@ test('成本中心总览异常统计必须走异常表关联，不能全表扫
  )
  assert.ok(source.includes('JOIN order_items oi ON oi.id = ce.order_item_id'))
  assert.ok(!source.includes('ce.order_item_id IN ('))
+  assert.ok(source.includes('productSourceFilterSql(1)'))
 })

 test('tpv: 商品来源归为第三方卡券', async () => {
@@ -336,3 +334,16 @@ test('tpv: 商品来源归为第三方卡券', async () => {
  assert.equal(productSourceOf('tpv:模拟卡券'), 'third_party_voucher')
  assert.equal(productSourceOf('man:模拟人工'), 'manual')
 })
+
+test('人工标准商品禁止绕过人工 SKU 配置手工创建', async () => {
+  await assert.rejects(
+    createAdminProduct(
+      {
+        canonicalSku: 'man:破片手榴弹-兔兔白日梦',
+        canonicalName: '破片手榴弹-兔兔白日梦',
+      },
+      {} as never,
+    ),
+    /人工 SKU 配置自动创建/,
+  )
+})
@@ -1,84 +0,0 @@
-/** 旧人工归一商品清理集成测试（真实 PostgreSQL）。 */
-import assert from 'node:assert/strict'
-import test from 'node:test'
-
-const TEST_DATABASE_URL = String(
-  process.env.LEGACY_MANUAL_PRODUCT_CLEANUP_TEST_DATABASE_URL || '',
-).trim()
-
-test(
-  '已被其他主渠道接管的历史 man 商品只归档并修正兜底指向',
-  { skip: !TEST_DATABASE_URL ? '未设置 LEGACY_MANUAL_PRODUCT_CLEANUP_TEST_DATABASE_URL' : false },
-  async () => {
-    process.env.DATABASE_URL = TEST_DATABASE_URL
-    const { closeDb, query } = await import('../../db/client.js')
-    const { archiveSupersededManualProducts } = await import('./product-catalog-service.js')
-    const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
-    const internalSku = `清理测试人工商品-${suffix}`
-    const manSku = `man:${internalSku}`
-    const liveSku = `aff:cleanup-${suffix}`
-    let manProductId = 0
-    let liveProductId = 0
-
-    try {
-      const inserted = await query<{ id: number; canonical_sku: string }>(
-        `
-          INSERT INTO product_catalog (
-            canonical_sku, canonical_name, category, unit_name, status,
-            attributes_json, created_at, updated_at
-          ) VALUES
-            ($1, $3, '人工客服', '件', 'active', '{"source":"manual_normalization"}'::jsonb, NOW(), NOW()),
-            ($2, $3, '自动发货', '件', 'active', '{}'::jsonb, NOW(), NOW())
-          RETURNING id, canonical_sku
-        `,
-        [manSku, liveSku, internalSku],
-      )
-      manProductId = Number(inserted.rows.find((row) => row.canonical_sku === manSku)?.id)
-      liveProductId = Number(inserted.rows.find((row) => row.canonical_sku === liveSku)?.id)
-      assert.ok(manProductId > 0)
-      assert.ok(liveProductId > 0)
-
-      await query(
-        `
-          INSERT INTO internal_sku_channels (
-            internal_sku, channel, priority, is_fallback, enabled,
-            canonical_product_id, config_json, source, created_at, updated_at
-          ) VALUES
-            ($1, 'affiliate_dash', 10, FALSE, TRUE, $3, '{}'::jsonb, 'test', NOW(), NOW()),
-            ($1, 'manual', 90, TRUE, TRUE, $2, '{}'::jsonb, 'test', NOW(), NOW())
-        `,
-        [internalSku, manProductId, liveProductId],
-      )
-
-      const preview = await archiveSupersededManualProducts()
-      assert.equal(preview.candidateCount, 1)
-      assert.ok(preview.sample.some((row) => Number(row.id) === manProductId))
-
-      const result = await archiveSupersededManualProducts({ execute: true })
-      assert.equal(result.archivedCount, 1)
-      assert.equal(result.repointedChannelCount, 1)
-
-      const products = await query<{ canonical_sku: string; status: string }>(
-        'SELECT canonical_sku, status FROM product_catalog WHERE id = ANY($1::bigint[])',
-        [[manProductId, liveProductId]],
-      )
-      assert.equal(products.rows.find((row) => row.canonical_sku === manSku)?.status, 'archived')
-      assert.equal(products.rows.find((row) => row.canonical_sku === liveSku)?.status, 'active')
-      const fallback = await query<{ canonical_product_id: number }>(
-        `SELECT canonical_product_id
-         FROM internal_sku_channels
-         WHERE internal_sku = $1 AND channel = 'manual'`,
-        [internalSku],
-      )
-      assert.equal(Number(fallback.rows[0]?.canonical_product_id), liveProductId)
-    } finally {
-      if (manProductId > 0 || liveProductId > 0) {
-        await query('DELETE FROM internal_sku_channels WHERE internal_sku = $1', [internalSku])
-        await query('DELETE FROM product_catalog WHERE id = ANY($1::bigint[])', [
-          [manProductId, liveProductId],
-        ])
-      }
-      await closeDb()
-    }
-  },
-)
@@ -1,10 +1,15 @@
+import type { PoolClient } from 'pg'
+
 import { query, withTransaction } from '../../db/client.js'
 import { resolveExceptionsForItemsByReason } from './cost-exception-service.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
 import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
 import { nowIso } from '../../utils/time.js'
-import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js'
+import {
+  upsertInternalSkuChannel,
+  upsertInternalSkuChannelWithClient,
+} from '../fulfillment/internal-sku-channel-service.js'
 import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
 import {
  attachFulfillmentConfigurations,
@@ -543,14 +548,14 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
            now,
          ],
        )
-    return requiredRow(result.rows[0], '保存人工履约配置失败')
+    const saved = requiredRow(result.rows[0], '保存人工履约配置失败')
+    await syncRegistryConfigId(saved.sku_key, saved.id, client)
+    await upsertManualRegistryConfig(saved, client)
+    // 标准商品、人工配置和渠道规则必须一起提交，避免保存成功后仍未生效。
+    await syncManualChannelRule(saved, client)
+    return saved
  })

-  await syncRegistryConfigId(config.sku_key, config.id)
-  await upsertManualRegistryConfig(config)
-  // 同步渠道规则：人工配置就是「这个内部 SKU 走人工渠道」的唯一声明，
-  // 少了这一步，新配的 SKU 会一直停在待分类。
-  await syncManualChannelRule(config)
  // 配置补全后，失败的 pending 会因 locked=true 无法被定时任务重试，这里主动解锁并重试。
  const pendingAssigned = await retryPendingManualItems(config.sku_key)
  // 配置保存后立刻回填该 SKU 最近 30 天的订单项，避免必须再点一次「重新匹配」。
@@ -938,26 +943,33 @@ async function attachCostVersionsForManualConfigs() {
 }

 /** 配置保存/删除后同步登记表上的 config_id。 */
-async function syncRegistryConfigId(skuKey: string, configId: number | null) {
-  await query(
-    `UPDATE manual_sku_registry
-               SET config_id = $2, updated_at = $3
-               WHERE canonicalize_internal_sku(sku_key) = $1`,
-    [canonicalizeInternalSku(skuKey), configId, nowIso()],
-  )
+async function syncRegistryConfigId(skuKey: string, configId: number | null, client?: PoolClient) {
+  const sql = `
+    UPDATE manual_sku_registry
+    SET config_id = $2, updated_at = $3
+    WHERE canonicalize_internal_sku(sku_key) = $1
+  `
+  const params = [canonicalizeInternalSku(skuKey), configId, nowIso()]
+  if (client) {
+    await client.query(sql, params)
+    return
+  }
+  await query(sql, params)
 }

 /** 配置保存后确保登记表仍有同一 SKU 的记录，兼容旧扫描清单和运维查询。 */
-async function upsertManualRegistryConfig(config: {
-  id: number
-  sku_key: string
-  canonical_product_id: number | null
-}) {
+async function upsertManualRegistryConfig(
+  config: {
+    id: number
+    sku_key: string
+    canonical_product_id: number | null
+  },
+  client?: PoolClient,
+) {
  const key = canonicalizeInternalSku(config.sku_key)
  if (!key) return
  const now = nowIso()
-  await query(
-    `
+  const sql = `
      INSERT INTO manual_sku_registry (
        sku_key, display_name, config_id, created_at, updated_at
      )
@@ -976,9 +988,13 @@ async function upsertManualRegistryConfig(config: {
        END,
        config_id = EXCLUDED.config_id,
        updated_at = EXCLUDED.updated_at
-    `,
-    [key, config.id, now, config.canonical_product_id],
-  )
+    `
+  const params = [key, config.id, now, config.canonical_product_id]
+  if (client) {
+    await client.query(sql, params)
+    return
+  }
+  await query(sql, params)
 }

 /** 读取单个 SKU 的配置详情（含平台标签与成本折算）。 */
@@ -1299,29 +1315,32 @@ function requiredRow<T>(row: T | undefined, message: string): T {
 /**
 * 人工配置 → internal_sku_channels（内部 SKU 精确匹配）。
 */
-async function syncManualChannelRule(config: {
-  id: number
-  sku_key: string
-  canonical_product_id: number | null
-  enabled: boolean
-  note: string
-}): Promise<void> {
+async function syncManualChannelRule(
+  config: {
+    id: number
+    sku_key: string
+    canonical_product_id: number | null
+    enabled: boolean
+    note: string
+  },
+  client?: PoolClient,
+): Promise<void> {
  const skuKey = String(config.sku_key || '').trim()
  if (!skuKey) return

-  const otherChannel = await query<{ exists: boolean }>(
-    `
+  const otherChannelSql = `
      SELECT EXISTS (
        SELECT 1 FROM internal_sku_channels
        WHERE canonicalize_internal_sku(internal_sku) = $1
          AND enabled = TRUE
          AND channel <> 'manual'
      ) AS exists
-    `,
-    [skuKey],
-  )
+    `
+  const otherChannel = client
+    ? await client.query<{ exists: boolean }>(otherChannelSql, [skuKey])
+    : await query<{ exists: boolean }>(otherChannelSql, [skuKey])
  const hasOtherChannel = otherChannel.rows[0]?.exists === true
-  await upsertInternalSkuChannel({
+  const channelInput = {
    internalSku: skuKey,
    channel: 'manual',
    priority: hasOtherChannel ? 90 : 10,
@@ -1331,5 +1350,10 @@ async function syncManualChannelRule(config: {
    source: 'manual_config',
    configJson: { manualConfigId: config.id, note: String(config.note || '').slice(0, 200) },
    replacePriority: true,
-  })
+  } as const
+  if (client) {
+    await upsertInternalSkuChannelWithClient(client, channelInput)
+    return
+  }
+  await upsertInternalSkuChannel(channelInput)
 }
@@ -319,6 +319,12 @@ export async function listAdminProducts(payload: JsonObject = {}) {

 export async function createAdminProduct(payload: JsonObject, _session: AdminSession) {
  const canonicalSku = normalizeRequiredText(payload.canonicalSku, '标准 SKU', 64)
+  if (canonicalSku.startsWith('man:')) {
+    throw createHttpError('人工商品由人工 SKU 配置自动创建，请从人工 SKU 配置入口新增', {
+      statusCode: 409,
+      errorCode: 'cost_manual_product_managed_by_config',
+    })
+  }
  if (!/^[A-Za-z0-9._:-]+$/.test(canonicalSku)) {
    throw createHttpError('标准 SKU 只能包含字母、数字、点、下划线、冒号和短横线', {
      statusCode: 400,
@@ -586,164 +592,6 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
  })
 }

-/**
- * 批量归档旧人工归一商品。
- *
- * 只处理 attributes_json.source=manual_normalization、当前为 active，
- * 且所有关联订单项都已 archived 的商品。只改 status，不删除任何数据。
- */
-export async function archiveHistoricalManualProducts(payload: JsonObject = {}) {
-  const execute = payload.execute === true || payload.execute === 'true'
-  const candidates = await query<{ id: number; canonical_sku: string; canonical_name: string }>(
-    `
-      SELECT p.id, p.canonical_sku, p.canonical_name
-      FROM product_catalog p
-      WHERE p.status = 'active'
-        AND p.canonical_sku LIKE 'man:%'
-        AND p.attributes_json->>'source' = 'manual_normalization'
-        AND EXISTS (SELECT 1 FROM order_items oi WHERE oi.canonical_product_id = p.id)
-        AND NOT EXISTS (
-          SELECT 1 FROM order_items oi
-          WHERE oi.canonical_product_id = p.id
-            AND oi.fulfillment_status <> 'archived'
-        )
-      ORDER BY p.id
-    `,
-  )
-  const ids = candidates.rows.map((row) => Number(row.id))
-  if (!execute || ids.length === 0) {
-    return {
-      dryRun: !execute,
-      candidateCount: ids.length,
-      archivedCount: 0,
-      sample: candidates.rows.slice(0, 20),
-    }
-  }
-
-  const updated = await query(
-    `UPDATE product_catalog
-     SET status = 'archived', updated_at = NOW()
-     WHERE id = ANY($1::bigint[]) AND status = 'active'`,
-    [ids],
-  )
-  return {
-    dryRun: false,
-    candidateCount: ids.length,
-    archivedCount: updated.rowCount || 0,
-    sample: candidates.rows.slice(0, 20),
-  }
-}
-
-type SupersededManualProductRow = {
-  id: number
-  canonical_sku: string
-  canonical_name: string
-  primary_channel: string
-  target_product_id: number
-  target_product_sku: string
-}
-
-/**
- * 归档已被其他渠道主规则接管的旧人工归一商品。
- *
- * `man:` 只表示旧人工来源，不代表当前主渠道。同名内部 SKU 已有自动发货、
- * 接单平台或卡券主渠道后，旧 man: 商品不再参与新路由，只需从当前目录归档。
- * 历史订单、成本版本和成本快照全部保留；仍指向旧商品的手动兜底改指当前主商品。
- */
-export async function archiveSupersededManualProducts(payload: JsonObject = {}) {
-  const execute = payload.execute === true || payload.execute === 'true'
-  return withTransaction(async (client) => {
-    const candidates = await client.query<SupersededManualProductRow>(
-      `
-        SELECT
-          p.id,
-          p.canonical_sku,
-          p.canonical_name,
-          live.channel AS primary_channel,
-          target.id AS target_product_id,
-          target.canonical_sku AS target_product_sku
-        FROM product_catalog p
-        JOIN LATERAL (
-          SELECT c.channel, c.canonical_product_id
-          FROM internal_sku_channels c
-          WHERE c.enabled = TRUE
-            AND c.is_fallback = FALSE
-            AND c.channel <> 'manual'
-            AND canonicalize_internal_sku(c.internal_sku)
-                = canonicalize_internal_sku(p.canonical_name)
-          ORDER BY c.priority, c.channel
-          LIMIT 1
-        ) live ON TRUE
-        JOIN product_catalog target
-          ON target.id = live.canonical_product_id
-         AND target.status = 'active'
-        WHERE p.status = 'active'
-          AND p.canonical_sku LIKE 'man:%'
-          AND p.attributes_json->>'source' = 'manual_normalization'
-        ORDER BY p.id
-        FOR UPDATE OF p
-      `,
-    )
-    const rows = candidates.rows.map((row) => ({
-      ...row,
-      id: Number(row.id),
-      target_product_id: Number(row.target_product_id),
-    }))
-    if (!execute || rows.length === 0) {
-      return {
-        dryRun: !execute,
-        candidateCount: rows.length,
-        archivedCount: 0,
-        repointedChannelCount: 0,
-        sample: rows.slice(0, 20),
-      }
-    }
-
-    const repointed = await client.query(
-      `
-        UPDATE internal_sku_channels c
-        SET canonical_product_id = input.target_product_id,
-            updated_at = NOW()
-        FROM unnest($1::text[], $2::bigint[], $3::bigint[])
-          AS input(internal_sku, source_product_id, target_product_id)
-        WHERE canonicalize_internal_sku(c.internal_sku)
-            = canonicalize_internal_sku(input.internal_sku)
-          AND c.channel = 'manual'
-          AND c.canonical_product_id = input.source_product_id
-      `,
-      [
-        rows.map((row) => row.canonical_name),
-        rows.map((row) => row.id),
-        rows.map((row) => row.target_product_id),
-      ],
-    )
-    const archived = await client.query(
-      `
-        UPDATE product_catalog
-        SET status = 'archived',
-            attributes_json = COALESCE(attributes_json, '{}'::jsonb) || $1::jsonb,
-            updated_at = NOW()
-        WHERE id = ANY($2::bigint[]) AND status = 'active'
-      `,
-      [
-        JSON.stringify({
-          legacyManualArchived: true,
-          legacyManualArchiveReason: 'superseded_by_active_primary_channel',
-          legacyManualArchivedAt: nowIso(),
-        }),
-        rows.map((row) => row.id),
-      ],
-    )
-    return {
-      dryRun: false,
-      candidateCount: rows.length,
-      archivedCount: archived.rowCount || 0,
-      repointedChannelCount: repointed.rowCount || 0,
-      sample: rows.slice(0, 20),
-    }
-  })
-}
-
 type MappingProductGroupRow = {
  canonical_product_id: number
  canonical_sku: string
@@ -115,6 +115,14 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务
      configSource.includes('dateFrom: shiftDate(today, -29)'),
    '保存人工 SKU 配置后必须立即回填最近 30 天订单项',
  )
+  const saveStart = configSource.indexOf('export async function saveManualFulfillmentConfig')
+  const saveEnd = configSource.indexOf('async function retryPendingManualItems', saveStart)
+  const saveBody = configSource.slice(saveStart, saveEnd)
+  assert.ok(
+    saveBody.indexOf('withTransaction(') <
+      saveBody.indexOf('await syncManualChannelRule(saved, client)'),
+    '标准商品、人工配置和人工渠道规则必须同一事务提交',
+  )
  assert.ok(
    jobSource.includes("fulfillment_channel = 'manual'") &&
      jobSource.includes('fulfillment_config_id'),
@@ -1,3 +1,5 @@
+import type { PoolClient } from 'pg'
+
 import { query, withTransaction } from '../../db/client.js'
 import type { FulfillmentChannel } from './channel-types.js'
 import { canonicalizeInternalSku, compareFulfillmentChannel } from './internal-sku-resolver.js'
@@ -127,7 +129,7 @@ export function planInternalSkuChannels(input: InternalSkuChannelRow[]): Interna
  }
 }

-export async function upsertInternalSkuChannel(input: {
+type InternalSkuChannelUpsertInput = {
  internalSku: string
  channel: FulfillmentChannel
  priority?: number
@@ -139,25 +141,29 @@ export async function upsertInternalSkuChannel(input: {
  note?: string | undefined
  adminUserId?: number | null | undefined
  replacePriority?: boolean
-}): Promise<number> {
+}
+
+async function upsertInternalSkuChannelWithExecutor(
+  client: Pick<PoolClient, 'query'>,
+  input: InternalSkuChannelUpsertInput,
+): Promise<number> {
  const internalSku = canonicalizeInternalSku(input.internalSku)
  if (!internalSku) return 0
-  return withTransaction(async (client) => {
-    if (input.enabled !== false && input.isFallback !== true) {
-      await client.query(
-        `UPDATE internal_sku_channels
-         SET is_fallback = TRUE,
-             priority = GREATEST(priority, 90),
-             updated_at = NOW()
-         WHERE canonicalize_internal_sku(internal_sku) = $1
-           AND channel <> $2
-           AND enabled = TRUE
-           AND is_fallback = FALSE`,
-        [internalSku, input.channel],
-      )
-    }
-    const result = await client.query<{ id: number }>(
-      `
+  if (input.enabled !== false && input.isFallback !== true) {
+    await client.query(
+      `UPDATE internal_sku_channels
+       SET is_fallback = TRUE,
+           priority = GREATEST(priority, 90),
+           updated_at = NOW()
+       WHERE canonicalize_internal_sku(internal_sku) = $1
+         AND channel <> $2
+         AND enabled = TRUE
+         AND is_fallback = FALSE`,
+      [internalSku, input.channel],
+    )
+  }
+  const result = await client.query<{ id: number }>(
+    `
      INSERT INTO internal_sku_channels (
        internal_sku, channel, priority, is_fallback, enabled,
        canonical_product_id, config_json, source, note, created_by_admin_user_id,
@@ -177,23 +183,35 @@ export async function upsertInternalSkuChannel(input: {
        created_by_admin_user_id = COALESCE(EXCLUDED.created_by_admin_user_id, internal_sku_channels.created_by_admin_user_id),
        updated_at = NOW()
      RETURNING id
-      `,
-      [
-        internalSku,
-        input.channel,
-        Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100,
-        input.isFallback === true,
-        input.enabled !== false,
-        input.canonicalProductId ?? null,
-        JSON.stringify(input.configJson || {}),
-        String(input.source || 'auto').trim() || 'auto',
-        String(input.note || '').slice(0, 300),
-        input.adminUserId ?? null,
-        input.replacePriority === true,
-      ],
-    )
-    return Number(result.rows[0]?.id || 0)
-  })
+    `,
+    [
+      internalSku,
+      input.channel,
+      Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100,
+      input.isFallback === true,
+      input.enabled !== false,
+      input.canonicalProductId ?? null,
+      JSON.stringify(input.configJson || {}),
+      String(input.source || 'auto').trim() || 'auto',
+      String(input.note || '').slice(0, 300),
+      input.adminUserId ?? null,
+      input.replacePriority === true,
+    ],
+  )
+  return Number(result.rows[0]?.id || 0)
+}
+
+export async function upsertInternalSkuChannelWithClient(
+  client: PoolClient,
+  input: InternalSkuChannelUpsertInput,
+): Promise<number> {
+  return upsertInternalSkuChannelWithExecutor(client, input)
+}
+
+export async function upsertInternalSkuChannel(
+  input: InternalSkuChannelUpsertInput,
+): Promise<number> {
+  return withTransaction((client) => upsertInternalSkuChannelWithExecutor(client, input))
 }

 /**