refactor(cost-center): 统一人工标准商品与成本配置

apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts

@@ -13,6 +13,7 @@ import {
   normalizeManualCreditQuantity,
   parseManualChannelQuantity,
 } from './manual-credit-quantity.js'
+import { productSourceFilterSql } from './cost-center-mappers.js'
 
 /**
  * 人工履约 SKU 配置服务。
@@ -223,9 +224,7 @@ export async function scanManualSkus(options: ScanOptions) {
   return { scanned: rows.length, registered: rows.length, fromDate, toDate }
 }
 
-/**
- * 列出人工 SKU 清单（登记表 + 配置状态）。
- */
+/** 列出人工标准商品清单（标准商品为唯一主键，配置与订单统计只做关联）。 */
 export async function listManualSkus(input: Record<string, unknown> = {}) {
   const unitPrices = await loadPlatformUnitPriceFen()
   const page = normalizePositiveInteger(input.page, 1, '页码')
@@ -238,13 +237,16 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
   const platformKey = String(input.platformKey || input.platform_key || '').trim()
   const provider = String(input.provider || '').trim()
 
-  const conditions: string[] = ['1 = 1']
-  const params: unknown[] = []
+  const conditions: string[] = [`p.status = 'active'`, `p.canonical_sku LIKE 'man:%'`]
+  const params: unknown[] = ['manual']
+  conditions.push(productSourceFilterSql(params.length))
 
   if (keyword) {
     params.push(`%${escapeLike(keyword)}%`)
     const placeholder = `$${params.length}`
-    conditions.push(`(r.sku_key ILIKE ${placeholder} OR r.display_name ILIKE ${placeholder})`)
+    conditions.push(
+      `(p.canonical_sku ILIKE ${placeholder} OR p.canonical_name ILIKE ${placeholder})`,
+    )
   }
   if (status === 'configured') {
     conditions.push('cfg.id IS NOT NULL')
@@ -266,7 +268,7 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
       WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
         AND o.provider = $${params.length}
         AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
-            = canonicalize_internal_sku(r.sku_key)
+            = canonicalize_internal_sku(substring(p.canonical_sku from 5))
     )`)
   }
 
@@ -274,9 +276,15 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
   const countResult = await query<{ total: number }>(
     `
       SELECT COUNT(*)::int AS total
-      FROM manual_sku_registry r
-      LEFT JOIN manual_fulfillment_configs cfg
-        ON canonicalize_internal_sku(cfg.sku_key) = canonicalize_internal_sku(r.sku_key)
+      FROM product_catalog p
+      LEFT JOIN LATERAL (
+        SELECT *
+        FROM manual_fulfillment_configs cfg
+        WHERE canonicalize_internal_sku(cfg.sku_key)
+            = canonicalize_internal_sku(substring(p.canonical_sku from 5))
+        ORDER BY cfg.id
+        LIMIT 1
+      ) cfg ON TRUE
       WHERE ${where}
     `,
     params,
@@ -284,22 +292,56 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
   const listParams = [...params, pageSize, (page - 1) * pageSize]
   const result = await query<ManualSkuRow>(
     `
+      WITH manual_stats AS (
+        SELECT
+          resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key,
+          COUNT(DISTINCT o.shop_id)::int AS shop_count,
+          COUNT(DISTINCT o.id)::int AS order_count,
+          COUNT(*)::int AS item_count,
+          COALESCE(SUM(GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)), 0)::bigint
+            AS revenue_fen,
+          MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)
+            AS first_seen_date,
+          MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)
+            AS last_seen_date
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        WHERE oi.fulfillment_status = 'assigned'
+          AND oi.fulfillment_channel = 'manual'
+        GROUP BY 1
+      )
       SELECT
-        r.id, r.sku_key, r.display_name, r.shop_count, r.order_count, r.item_count,
-        r.revenue_fen, r.first_seen_date, r.last_seen_date,
-        r.scan_from_date, r.scan_to_date,
+        p.id,
+        substring(p.canonical_sku from 5) AS sku_key,
+        p.canonical_name AS display_name,
+        COALESCE(manual_stats.shop_count, 0)::int AS shop_count,
+        COALESCE(manual_stats.order_count, 0)::int AS order_count,
+        COALESCE(manual_stats.item_count, 0)::int AS item_count,
+        COALESCE(manual_stats.revenue_fen, 0)::bigint AS revenue_fen,
+        manual_stats.first_seen_date,
+        manual_stats.last_seen_date,
+        manual_stats.first_seen_date AS scan_from_date,
+        manual_stats.last_seen_date AS scan_to_date,
         cfg.id AS config_id,
         cfg.platform_key AS config_platform_key,
         cfg.credit_quantity AS config_credit_quantity,
         cfg.cost_mode AS config_cost_mode,
-        cfg.canonical_product_id AS config_canonical_product_id,
+        COALESCE(cfg.canonical_product_id, p.id) AS config_canonical_product_id,
         cfg.enabled AS config_enabled,
         cfg.channels_json AS config_channels_json
-      FROM manual_sku_registry r
-      LEFT JOIN manual_fulfillment_configs cfg
-        ON canonicalize_internal_sku(cfg.sku_key) = canonicalize_internal_sku(r.sku_key)
+      FROM product_catalog p
+      LEFT JOIN LATERAL (
+        SELECT *
+        FROM manual_fulfillment_configs cfg
+        WHERE canonicalize_internal_sku(cfg.sku_key)
+            = canonicalize_internal_sku(substring(p.canonical_sku from 5))
+        ORDER BY cfg.id
+        LIMIT 1
+      ) cfg ON TRUE
+      LEFT JOIN manual_stats
+        ON manual_stats.sku_key = canonicalize_internal_sku(substring(p.canonical_sku from 5))
       WHERE ${where}
-      ORDER BY r.item_count DESC, r.sku_key ASC
+      ORDER BY COALESCE(manual_stats.item_count, 0) DESC, p.canonical_name ASC, p.id ASC
       LIMIT $${listParams.length - 1} OFFSET $${listParams.length}
     `,
     listParams,
@@ -323,20 +365,50 @@ export async function getManualSkuOverview() {
     last_scan_at: string | null
   }>(
     `
+      WITH manual_stats AS (
+        SELECT
+          resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key,
+          COUNT(*)::int AS item_count,
+          MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)
+            AS last_seen_date
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        WHERE oi.fulfillment_status = 'assigned'
+          AND oi.fulfillment_channel = 'manual'
+        GROUP BY 1
+      ),
+      manual_scope AS (
+        SELECT p.id, p.canonical_sku, p.canonical_name, cfg.id AS config_id
+        FROM product_catalog p
+        LEFT JOIN LATERAL (
+          SELECT cfg.id
+          FROM manual_fulfillment_configs cfg
+          WHERE canonicalize_internal_sku(cfg.sku_key)
+              = canonicalize_internal_sku(substring(p.canonical_sku from 5))
+          ORDER BY cfg.id
+          LIMIT 1
+        ) cfg ON TRUE
+        WHERE p.status = 'active'
+          AND p.canonical_sku LIKE 'man:%'
+          AND ${productSourceFilterSql(1)}
+      )
       SELECT
         COUNT(*)::int AS total_sku,
-        COUNT(*) FILTER (WHERE cfg.id IS NOT NULL)::int AS configured_sku,
-        COUNT(*) FILTER (WHERE cfg.id IS NULL)::int AS unconfigured_sku,
-        COALESCE(SUM(r.item_count), 0)::bigint AS total_items,
-        COALESCE(SUM(r.item_count) FILTER (WHERE cfg.id IS NOT NULL), 0)::bigint
+        COUNT(*) FILTER (WHERE config_id IS NOT NULL)::int AS configured_sku,
+        COUNT(*) FILTER (WHERE config_id IS NULL)::int AS unconfigured_sku,
+        COALESCE(SUM(manual_stats.item_count), 0)::bigint AS total_items,
+        COALESCE(SUM(manual_stats.item_count) FILTER (WHERE config_id IS NOT NULL), 0)::bigint
           AS configured_items,
-        COALESCE(SUM(r.item_count) FILTER (WHERE cfg.id IS NULL), 0)::bigint
+        COALESCE(SUM(manual_stats.item_count) FILTER (WHERE config_id IS NULL), 0)::bigint
           AS unconfigured_items,
-        MAX(r.updated_at) AS last_scan_at
-      FROM manual_sku_registry r
-      LEFT JOIN manual_fulfillment_configs cfg
-        ON canonicalize_internal_sku(cfg.sku_key) = canonicalize_internal_sku(r.sku_key)
+        MAX(manual_stats.last_seen_date)::text AS last_scan_at
+      FROM manual_scope
+      LEFT JOIN manual_stats
+        ON manual_stats.sku_key = canonicalize_internal_sku(
+          substring(manual_scope.canonical_sku from 5)
+        )
     `,
+    ['manual'],
   )
   const row = result.rows[0]
   return {
@@ -470,6 +542,7 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
   })
 
   await syncRegistryConfigId(config.sku_key, config.id)
+  await upsertManualRegistryConfig(config)
   // 同步渠道规则：人工配置就是「这个内部 SKU 走人工渠道」的唯一声明，
   // 少了这一步，新配的 SKU 会一直停在待分类。
   await syncManualChannelRule(config)
@@ -556,6 +629,7 @@ export type DeleteManualSkuResult = {
   deletedConfigs: number
   deletedRegistry: number
   deletedRules: number
+  archivedProducts: number
   refreshedDays: number
   skippedClosedDays: number
 }
@@ -671,6 +745,22 @@ export async function deleteManualSkuClassification(
        RETURNING id`,
       [key],
     )
+    const products = await client.query(
+      `
+        UPDATE product_catalog
+        SET status = 'archived',
+            attributes_json = attributes_json || jsonb_build_object(
+              'deletedAt', $2::text,
+              'deletedBy', 'manual_sku_config'
+            ),
+            updated_at = $2::timestamptz
+        WHERE status = 'active'
+          AND canonical_sku LIKE 'man:%'
+          AND canonicalize_internal_sku(substring(canonical_sku from 5)) = $1
+        RETURNING id
+      `,
+      [key, nowIso()],
+    )
 
     if (itemIds.length > 0) {
       await client.query(
@@ -703,6 +793,7 @@ export async function deleteManualSkuClassification(
       deletedConfigs: configs.rowCount || 0,
       deletedRegistry: registry.rowCount || 0,
       deletedRules: rules.rowCount || 0,
+      archivedProducts: products.rowCount || 0,
       dates: [...new Set(items.rows.map((row) => row.basis_date).filter(Boolean))],
     }
   })
@@ -714,6 +805,7 @@ export async function deleteManualSkuClassification(
     deletedConfigs: result.deletedConfigs,
     deletedRegistry: result.deletedRegistry,
     deletedRules: result.deletedRules,
+    archivedProducts: result.archivedProducts,
     refreshedDays: refreshed.days,
     skippedClosedDays: refreshed.skippedClosed,
   }
@@ -840,6 +932,40 @@ async function syncRegistryConfigId(skuKey: string, configId: number | null) {
   )
 }
 
+/** 配置保存后确保登记表仍有同一 SKU 的记录，兼容旧扫描清单和运维查询。 */
+async function upsertManualRegistryConfig(config: {
+  id: number
+  sku_key: string
+  canonical_product_id: number | null
+}) {
+  const key = canonicalizeInternalSku(config.sku_key)
+  if (!key) return
+  const now = nowIso()
+  await query(
+    `
+      INSERT INTO manual_sku_registry (
+        sku_key, display_name, config_id, created_at, updated_at
+      )
+      SELECT
+        $1,
+        COALESCE(NULLIF(product.canonical_name, ''), $1),
+        $2,
+        $3,
+        $3
+      FROM (SELECT 1) seed
+      LEFT JOIN product_catalog product ON product.id = $4
+      ON CONFLICT (sku_key) DO UPDATE SET
+        display_name = CASE
+          WHEN BTRIM(manual_sku_registry.display_name) = '' THEN EXCLUDED.display_name
+          ELSE manual_sku_registry.display_name
+        END,
+        config_id = EXCLUDED.config_id,
+        updated_at = EXCLUDED.updated_at
+    `,
+    [key, config.id, now, config.canonical_product_id],
+  )
+}
+
 /** 读取单个 SKU 的配置详情（含平台标签与成本折算）。 */
 export async function getManualFulfillmentConfig(skuKey: string) {
   const unitPrices = await loadPlatformUnitPriceFen()

apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts

@@ -90,15 +90,20 @@ test('人工确认必须单事务：扣款与快照同事务，且使用 with-cl
 })
 
 test('人工确认前必须自动匹配履约配置，配置保存/定时任务也要回填', async () => {
-  const [manualSource, configSource, jobSource, migrationSource] = await Promise.all([
-    readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'),
-    readFile(
-      new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url),
-      'utf8',
-    ),
-    readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'),
-    readFile(new URL('../../db/migrations/106_manual_sku_key.sql', import.meta.url), 'utf8'),
-  ])
+  const [manualSource, configSource, jobSource, migrationSource, reconcileMigration] =
+    await Promise.all([
+      readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'),
+      readFile(
+        new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url),
+        'utf8',
+      ),
+      readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'),
+      readFile(new URL('../../db/migrations/106_manual_sku_key.sql', import.meta.url), 'utf8'),
+      readFile(
+        new URL('../../db/migrations/135_reconcile_manual_standard_products.sql', import.meta.url),
+        'utf8',
+      ),
+    ])
   assert.ok(
     manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId, now)'),
     '确认时必须按 SKU 自动挂渠道',
@@ -137,6 +142,18 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务
       migrationSource.includes('sku_name'),
     '生成列必须保留 sku_code / sku_name 兜底',
   )
+  const listStart = configSource.indexOf('export async function listManualSkus')
+  const listEnd = configSource.indexOf('export async function getManualSkuOverview')
+  const listBody = configSource.slice(listStart, listEnd)
+  assert.ok(
+    listBody.includes('FROM product_catalog p') && listBody.includes('productSourceFilterSql'),
+    '人工成本配置必须以 man: 标准商品为唯一清单来源',
+  )
+  assert.ok(
+    reconcileMigration.includes('tmp_manual_product_merge') &&
+      reconcileMigration.includes("status = 'archived'"),
+    '存量重复人工 SKU 必须合并到同一标准商品并归档重复项',
+  )
 })
 
 test('人工成本缺失由管道按配置与快照生成并收口', async () => {