fix(fulfillment): 统一渠道规则与标准商品绑定

2026-11-05 18:17:12 +08:00
parent 8b9b4fc4ad
commit 89a68b5734
3 changed files with 318 additions and 59 deletions
@@ -0,0 +1,139 @@
+-- 136_reconcile_channel_products_and_work_costs.sql
+--
+-- 渠道商品与成本模型归一：
+--   1. 已启用的第三方卡券规则必须有 active tpv: 标准商品；
+--   2. 接单报酬成本版本不得挂在 man: 人工商品上；无快照引用的历史错误版本退休；
+--   3. 保留有快照引用的历史数据，不删除订单、成本版本或成本快照。
+
+-- 卡券规则已有、但标准商品缺失时补建商品。
+INSERT INTO product_catalog (
+  canonical_sku, canonical_name, category, unit_name, status,
+  attributes_json, created_at, updated_at
+)
+SELECT
+  'tpv:' || canonicalize_internal_sku(channel.internal_sku),
+  BTRIM(channel.internal_sku),
+  '卡券',
+  '张',
+  'active',
+  '{"source":"voucher_rule","auto_created":true,"reconciled":true}'::jsonb,
+  NOW(),
+  NOW()
+FROM internal_sku_channels channel
+WHERE channel.channel = 'third_party_voucher'
+  AND channel.enabled = TRUE
+  AND canonicalize_internal_sku(channel.internal_sku) <> ''
+  AND char_length('tpv:' || canonicalize_internal_sku(channel.internal_sku)) <= 64
+  AND NOT EXISTS (
+    SELECT 1
+    FROM product_catalog product
+    WHERE product.status = 'active'
+      AND product.canonical_sku LIKE 'tpv:%'
+      AND canonicalize_internal_sku(substring(product.canonical_sku from 5))
+          = canonicalize_internal_sku(channel.internal_sku)
+  )
+  AND NOT EXISTS (
+    SELECT 1
+    FROM internal_sku_channels other
+    WHERE other.enabled = TRUE
+      AND other.is_fallback = FALSE
+      AND other.channel <> 'third_party_voucher'
+      AND canonicalize_internal_sku(other.internal_sku)
+          = canonicalize_internal_sku(channel.internal_sku)
+  )
+ON CONFLICT (canonical_sku) DO UPDATE SET
+  status = 'active',
+  canonical_name = CASE
+    WHEN BTRIM(product_catalog.canonical_name) = '' THEN EXCLUDED.canonical_name
+    ELSE product_catalog.canonical_name
+  END,
+  updated_at = EXCLUDED.updated_at;
+
+-- 卡券规则统一挂到 active 标准商品，并补齐商品渠道能力。
+UPDATE internal_sku_channels channel
+SET canonical_product_id = product.id,
+    priority = 10,
+    is_fallback = FALSE,
+    config_json = channel.config_json || jsonb_build_object(
+      'voucherSku',
+      canonicalize_internal_sku(channel.internal_sku)
+    ),
+    updated_at = NOW()
+FROM product_catalog product
+WHERE channel.channel = 'third_party_voucher'
+  AND channel.enabled = TRUE
+  AND product.status = 'active'
+  AND product.canonical_sku LIKE 'tpv:%'
+  AND canonicalize_internal_sku(substring(product.canonical_sku from 5))
+      = canonicalize_internal_sku(channel.internal_sku)
+  AND (
+    channel.canonical_product_id IS DISTINCT FROM product.id
+    OR channel.priority <> 10
+    OR channel.is_fallback
+  );
+
+INSERT INTO product_channel_capabilities (
+  product_id, channel, priority, is_fallback, enabled, source, created_at, updated_at
+)
+SELECT
+  channel.canonical_product_id,
+  'third_party_voucher',
+  10,
+  FALSE,
+  TRUE,
+  'voucher',
+  NOW(),
+  NOW()
+FROM internal_sku_channels channel
+JOIN product_catalog product
+  ON product.id = channel.canonical_product_id
+ AND product.status = 'active'
+WHERE channel.channel = 'third_party_voucher'
+  AND channel.enabled = TRUE
+  AND channel.canonical_product_id IS NOT NULL
+ON CONFLICT (product_id, channel) DO UPDATE SET
+  enabled = TRUE,
+  priority = 10,
+  is_fallback = FALSE,
+  updated_at = EXCLUDED.updated_at;
+
+-- 无法补齐标准商品的启用卡券规则必须停用，否则会继续把订单路由到无商品渠道。
+UPDATE internal_sku_channels channel
+SET enabled = FALSE,
+    is_fallback = FALSE,
+    note = CASE
+      WHEN BTRIM(channel.note) = '' THEN '标准商品缺失，迁移 136 自动停用'
+      ELSE channel.note
+    END,
+    updated_at = NOW()
+WHERE channel.channel = 'third_party_voucher'
+  AND channel.enabled = TRUE
+  AND NOT EXISTS (
+    SELECT 1
+    FROM product_catalog product
+    WHERE product.id = channel.canonical_product_id
+      AND product.status = 'active'
+      AND product.canonical_sku LIKE 'tpv:%'
+  );
+
+-- 早期错误地把“接单报酬”成本挂到了人工商品。无快照引用时直接退休。
+UPDATE cost_versions version
+SET status = 'retired',
+    change_reason = CASE
+      WHEN version.change_reason = '' THEN '清理错误挂到人工商品的接单报酬成本版本'
+      ELSE version.change_reason
+    END,
+    updated_at = NOW()
+FROM cost_items item,
+     product_catalog product
+WHERE version.cost_item_id = item.id
+  AND version.canonical_product_id = product.id
+  AND version.status = 'active'
+  AND item.cost_model = 'work_order_reward'
+  AND product.canonical_sku LIKE 'man:%'
+  AND NOT EXISTS (
+    SELECT 1 FROM cost_snapshots snapshot WHERE snapshot.cost_version_id = version.id
+  );
+
+COMMENT ON COLUMN internal_sku_channels.canonical_product_id IS
+  '渠道规则关联的标准商品；启用规则必须指向 active 标准商品，历史归档仍保留关联';
@@ -556,13 +556,13 @@ export async function assignOrderItemsAfterCarrierReady(input: {
      } else {
        await query(
          `UPDATE order_items
-           SET fulfillment_channel = $3,
+           SET fulfillment_channel = NULL,
               fulfillment_status = 'pending',
               fulfillment_source = 'operator_assigned',
               fulfillment_locked = TRUE,
               fulfillment_pending_reason = 'no_carrier',
               fulfillment_pending_detail = $2::jsonb,
-               canonical_product_id = COALESCE($4, canonical_product_id),
+               canonical_product_id = COALESCE($3, canonical_product_id),
               updated_at = NOW()
           WHERE id = $1 AND fulfillment_status <> 'archived'`,
          [
@@ -573,7 +573,6 @@ export async function assignOrderItemsAfterCarrierReady(input: {
              message: carrier.message,
              carrierIssue: carrier.reason,
            }),
-            input.channel,
            input.canonicalProductId || null,
          ],
        )
@@ -761,6 +760,14 @@ export async function switchChannelForOrderItems(input: {
        switchInternalSkus.length === 0 ? 'internal_sku_required' : 'multiple_internal_skus',
    })
  }
+  const targetInternalSku = switchInternalSkus[0]!
+  const draftProductId = await ensureDraftProductForChannel(targetInternalSku, input.channel)
+  if (!draftProductId) {
+    throw createHttpError('渠道标准商品创建或启用失败，请先在商品配置中处理', {
+      statusCode: 409,
+      errorCode: 'channel_standard_product_required',
+    })
+  }

  const items = (
    await query<OrderItemRow>(`SELECT * FROM order_items WHERE id = ANY($1::bigint[])`, [ids])
@@ -909,6 +916,7 @@ export async function switchChannelForOrderItems(input: {
    source: 'operator_assigned',
    lock: true,
    note: input.note,
+    canonicalProductId: draftProductId,
  })
  await query(
    `UPDATE order_items
@@ -917,12 +925,11 @@ export async function switchChannelForOrderItems(input: {
    [readyIds, input.adminUserId],
  )
  if (assigned.updated > 0) {
-    const internalSku = switchInternalSkus[0]!
-    const carrier = await resolveInternalSkuCarrierConfig(internalSku, input.channel)
+    const carrier = await resolveInternalSkuCarrierConfig(targetInternalSku, input.channel)
    await setInternalSkuPrimaryChannel({
-      internalSku,
+      internalSku: targetInternalSku,
      channel: input.channel,
-      canonicalProductId: carrier.canonicalProductId,
+      canonicalProductId: draftProductId,
      configJson: carrier.configJson,
      source: 'operator',
      note: input.note,
@@ -1580,21 +1587,18 @@ export async function assignChannelByInternalSku(input: {
    [internalSku, input.channel],
  )

-  const draftProductId = await ensureDraftProductForChannel(internalSku, input.channel).catch(
-    (error) => {
-      logWarn('[channel-assign]', '自动创建渠道草稿商品失败', {
-        internalSku,
-        channel: input.channel,
-        error: error instanceof Error ? error.message : String(error),
-      })
-      return null
-    },
-  )
+  const draftProductId = await ensureDraftProductForChannel(internalSku, input.channel)
+  if (!draftProductId) {
+    throw createHttpError('渠道标准商品创建或启用失败，请先在商品配置中处理', {
+      statusCode: 409,
+      errorCode: 'channel_standard_product_required',
+    })
+  }
  const carrier = await resolveInternalSkuCarrierConfig(internalSku, input.channel)
  await setInternalSkuPrimaryChannel({
    internalSku,
    channel: input.channel,
-    canonicalProductId: carrier.canonicalProductId,
+    canonicalProductId: draftProductId,
    source: 'operator',
    configJson: {
      ...carrier.configJson,
@@ -1818,7 +1822,18 @@ async function ensureDraftProductForChannel(
        $5::jsonb, $6, $6
      )
      ON CONFLICT (canonical_sku) DO UPDATE
-      SET updated_at = product_catalog.updated_at
+      SET status = 'active',
+          canonical_name = CASE
+            WHEN BTRIM(product_catalog.canonical_name) = '' THEN EXCLUDED.canonical_name
+            ELSE product_catalog.canonical_name
+          END,
+          attributes_json = (
+            product_catalog.attributes_json
+              - 'deletedAt'
+              - 'deletedBy'
+              - 'archivedAt'
+          ) || jsonb_build_object('reactivatedAt', EXCLUDED.updated_at),
+          updated_at = EXCLUDED.updated_at
      RETURNING id, status
    `,
    [
@@ -1891,7 +1906,7 @@ async function resolveInternalSkuCarrierConfig(
     FROM product_catalog
     WHERE status = 'active'
       AND (canonical_sku LIKE $2 OR canonical_sku LIKE $3)
-       AND canonicalize_internal_sku(canonical_name) = $1
+       AND canonicalize_internal_sku(substring(canonical_sku from 5)) = $1
     ORDER BY id`,
    [key, `${prefixes[0]}%`, `${prefixes[1] || prefixes[0]}%`],
  )
@@ -1958,19 +1973,28 @@ export async function upsertChannelRule(input: {
    })
  }

-  const priority = Math.trunc(Number(input.priority) || 100)
+  const draftProductId = await ensureDraftProductForChannel(matchKey, input.channel)
+  if (!draftProductId) {
+    throw createHttpError('渠道标准商品创建或启用失败，请先在商品配置中处理', {
+      statusCode: 409,
+      errorCode: 'channel_standard_product_required',
+    })
+  }
+
+  const priority =
+    input.channel === 'third_party_voucher' ? 10 : Math.trunc(Number(input.priority) || 100)
  const carrier = await resolveInternalSkuCarrierConfig(matchKey, input.channel)
  const id = await upsertInternalSkuChannel({
    internalSku: matchKey,
    channel: input.channel,
    priority,
-    isFallback: priority >= 90,
+    isFallback: input.channel === 'third_party_voucher' ? false : priority >= 90,
    enabled: input.enabled !== false,
    source: source === 'operator' ? 'operator' : `rule:${source}`,
    note: String(input.note || '').slice(0, 300),
    adminUserId: input.adminUserId,
    replacePriority: true,
-    canonicalProductId: carrier.canonicalProductId,
+    canonicalProductId: draftProductId,
    configJson: carrier.configJson,
  })
  return { id }
@@ -2079,15 +2103,18 @@ export async function ensureVoucherRuleForSku(skuCode: string): Promise<boolean>
     LIMIT 1`,
    [`tpv:${key}`],
  )
+  const productId = Number(product.rows[0]?.id || 0)
+  if (!productId) return false
  await upsertInternalSkuChannel({
    internalSku: key,
    channel: 'third_party_voucher',
    priority: 10,
    isFallback: false,
    enabled: true,
-    canonicalProductId: Number(product.rows[0]?.id || 0) || null,
+    canonicalProductId: productId,
    configJson: { voucherSku: key },
    source: 'voucher',
+    replacePriority: true,
  })
  return true
 }
@@ -2110,7 +2137,8 @@ export async function ensureVoucherChannelRules(): Promise<{
     FROM product_catalog product
     WHERE product.status = 'active'
       AND product.canonical_sku LIKE 'tpv:%'
-       AND canonicalize_internal_sku(channel.internal_sku) = canonicalize_internal_sku(product.canonical_name)
+       AND canonicalize_internal_sku(channel.internal_sku) =
+           canonicalize_internal_sku(substring(product.canonical_sku from 5))
       AND channel.channel <> 'third_party_voucher'
       AND channel.enabled = TRUE
       AND channel.is_fallback = FALSE`,
@@ -2122,17 +2150,21 @@ export async function ensureVoucherChannelRules(): Promise<{
        canonical_product_id, config_json, source, created_at, updated_at
      )
      SELECT
-        canonicalize_internal_sku(p.canonical_name), 'third_party_voucher', 10, FALSE, TRUE,
+        canonicalize_internal_sku(substring(p.canonical_sku from 5)),
+        'third_party_voucher', 10, FALSE, TRUE,
        p.id, jsonb_build_object('voucherSku', substring(p.canonical_sku from 5)),
        'voucher', NOW(), NOW()
      FROM product_catalog p
      WHERE p.status = 'active'
        AND p.canonical_sku LIKE 'tpv:%'
-        AND BTRIM(COALESCE(p.canonical_name, '')) <> ''
+        AND canonicalize_internal_sku(substring(p.canonical_sku from 5)) <> ''
      ON CONFLICT (internal_sku, channel) DO UPDATE SET
+        priority = 10,
+        is_fallback = FALSE,
        enabled = TRUE,
-        canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
+        canonical_product_id = EXCLUDED.canonical_product_id,
        config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
+        source = EXCLUDED.source,
        updated_at = NOW()
      RETURNING id
    `,
@@ -54,6 +54,8 @@ test(
    const retryWorkOrderSku = `集成测试重试接单-${suffix}`
    const manualRemoveSku = `集成测试移除人工-${suffix}`
    const protectedRuleSku = `集成测试强制删除-${suffix}`
+    const missingVoucherSku = `集成测试无商品卡券-${suffix}`
+    const reactivatedVoucherSku = `集成测试归档卡券-${suffix}`
    const retryWorkOrderRuleKey = `it-retry-${suffix}`
    const orderIds: number[] = []
    const orderItemIds: number[] = []
@@ -69,8 +71,10 @@ test(
      retryWorkOrderSku,
      manualRemoveSku,
      protectedRuleSku,
+      missingVoucherSku,
+      reactivatedVoucherSku,
    ]
-    let generatedVoucherProductId: number | null = null
+    const createdVoucherProductIds: number[] = []
    let autoManualProductId: number | null = null
    let retryWorkOrderRuleId: number | null = null
    let fallbackProfileId: number | null = null
@@ -251,8 +255,6 @@ test(
        'third_party_voucher',
      )
      assert.equal(switchedItem.rows[0]?.item_snapshot_json.isConfigured, false)
-      generatedVoucherProductId = Number(switchedItem.rows[0]?.canonical_product_id || 0) || null
-
      const missingAffiliateItemId = await createPendingOrderItem(missingAffiliateSku)
      const affiliateFailed = await assignChannelByInternalSku({
        internalSku: missingAffiliateSku,
@@ -644,13 +646,47 @@ test(
        ['closed', 'paid'],
      )

-      await ensureVoucherRuleForSku(voucherSku)
-      const voucherChannel = await query<{ channel: string; source: string }>(
-        `SELECT channel, source FROM internal_sku_channels
+      assert.equal(await ensureVoucherRuleForSku(missingVoucherSku), false)
+      const missingVoucherRules = await query<{ count: number }>(
+        `SELECT count(*)::int AS count FROM internal_sku_channels
+         WHERE internal_sku = $1 AND channel = 'third_party_voucher'`,
+        [missingVoucherSku],
+      )
+      assert.equal(missingVoucherRules.rows[0]?.count, 0)
+
+      const voucherProduct = 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`,
+        [`tpv:${voucherSku}`, `卡券展示名-${suffix}`],
+      )
+      createdVoucherProductIds.push(Number(voucherProduct.rows[0]!.id))
+      assert.equal(await ensureVoucherRuleForSku(voucherSku), true)
+      const voucherChannel = await query<{
+        channel: string
+        source: string
+        canonical_product_id: string | null
+        priority: number
+        is_fallback: boolean
+        config_json: Record<string, unknown>
+      }>(
+        `SELECT channel, source, canonical_product_id, priority, is_fallback, config_json
+         FROM internal_sku_channels
         WHERE internal_sku = $1 ORDER BY channel`,
        [voucherSku],
      )
-      assert.deepEqual(voucherChannel.rows, [{ channel: 'third_party_voucher', source: 'voucher' }])
+      assert.equal(voucherChannel.rows.length, 1)
+      assert.equal(voucherChannel.rows[0]?.channel, 'third_party_voucher')
+      assert.equal(voucherChannel.rows[0]?.source, 'voucher')
+      assert.equal(
+        Number(voucherChannel.rows[0]?.canonical_product_id),
+        createdVoucherProductIds[0],
+      )
+      assert.equal(voucherChannel.rows[0]?.priority, 10)
+      assert.equal(voucherChannel.rows[0]?.is_fallback, false)
+      assert.equal(voucherChannel.rows[0]?.config_json.voucherSku, voucherSku)
      const legacyVoucherRules = await query<{ count: number }>(
        `SELECT count(*)::int AS count FROM sku_channel_rules
         WHERE match_key = $1`,
@@ -692,6 +728,50 @@ test(
      const deletedRules = await deleteChannelRules([savedRule.id])
      assert.equal(deletedRules.deletedCount, 1)
      assert.deepEqual(deletedRules.deletedIds, [savedRule.id])
+
+      const archivedVoucherProduct = 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, '卡券', '张', 'archived',
+           '{"deletedAt":"2026-01-01T00:00:00.000Z","deletedBy":"integration_test"}'::jsonb,
+           NOW(), NOW()
+         )
+         RETURNING id`,
+        [`tpv:${reactivatedVoucherSku}`, reactivatedVoucherSku],
+      )
+      const archivedVoucherProductId = Number(archivedVoucherProduct.rows[0]!.id)
+      createdVoucherProductIds.push(archivedVoucherProductId)
+      const reactivatedRule = await upsertChannelRule({
+        matchKeyType: 'internal_sku',
+        matchKey: reactivatedVoucherSku,
+        channel: 'third_party_voucher',
+        priority: 80,
+        enabled: true,
+        note: '集成测试重新激活归档商品',
+        adminUserId: null,
+      })
+      const reactivated = await query<{
+        status: string
+        attributes_json: Record<string, unknown>
+        canonical_product_id: string | null
+        priority: number
+        is_fallback: boolean
+      }>(
+        `SELECT product.status, product.attributes_json,
+                channel.canonical_product_id, channel.priority, channel.is_fallback
+         FROM internal_sku_channels channel
+         JOIN product_catalog product ON product.id = channel.canonical_product_id
+         WHERE channel.id = $1`,
+        [reactivatedRule.id],
+      )
+      assert.equal(reactivated.rows[0]?.status, 'active')
+      assert.equal(Number(reactivated.rows[0]?.canonical_product_id), archivedVoucherProductId)
+      assert.equal(reactivated.rows[0]?.priority, 10)
+      assert.equal(reactivated.rows[0]?.is_fallback, false)
+      assert.equal(reactivated.rows[0]?.attributes_json.deletedAt, undefined)
+      assert.equal(typeof reactivated.rows[0]?.attributes_json.reactivatedAt, 'string')
    } finally {
      if (orderItemIds.length > 0) {
        await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [
@@ -731,32 +811,40 @@ test(
         WHERE provider = '91kaquan' AND shop_id = $1 AND match_key = ANY($2::text[])`,
        [shopId, internalSkus],
      )
-      if (generatedVoucherProductId) {
-        await query(`DELETE FROM sales_daily_facts WHERE canonical_product_id = $1`, [
-          generatedVoucherProductId,
+      const generatedProducts = await query<{ id: string }>(
+        `SELECT id
+         FROM product_catalog
+         WHERE canonicalize_internal_sku(substring(canonical_sku from 5)) = ANY($1::text[])`,
+        [internalSkus],
+      )
+      const generatedProductIds = generatedProducts.rows.map((row) => Number(row.id))
+      if (generatedProductIds.length > 0) {
+        await query(
+          `DELETE FROM cost_snapshots
+           WHERE canonical_product_id = ANY($1::bigint[])
+              OR cost_version_id IN (
+                SELECT id FROM cost_versions WHERE canonical_product_id = ANY($1::bigint[])
+              )`,
+          [generatedProductIds],
+        )
+        await query(
+          `DELETE FROM sales_daily_facts WHERE canonical_product_id = ANY($1::bigint[])`,
+          [generatedProductIds],
+        )
+        await query(
+          `DELETE FROM channel_product_mappings WHERE canonical_product_id = ANY($1::bigint[])`,
+          [generatedProductIds],
+        )
+        await query(`DELETE FROM cost_versions WHERE canonical_product_id = ANY($1::bigint[])`, [
+          generatedProductIds,
        ])
-        await query(`DELETE FROM cost_snapshots WHERE canonical_product_id = $1`, [
-          generatedVoucherProductId,
+        await query(
+          `DELETE FROM product_channel_capabilities WHERE product_id = ANY($1::bigint[])`,
+          [generatedProductIds],
+        )
+        await query(`DELETE FROM product_catalog WHERE id = ANY($1::bigint[])`, [
+          generatedProductIds,
        ])
-        await query(`DELETE FROM channel_product_mappings WHERE canonical_product_id = $1`, [
-          generatedVoucherProductId,
-        ])
-        await query(`DELETE FROM cost_versions WHERE canonical_product_id = $1`, [
-          generatedVoucherProductId,
-        ])
-        await query(`DELETE FROM product_catalog WHERE id = $1 AND canonical_sku = $2`, [
-          generatedVoucherProductId,
-          `tpv:${manualSku}`,
-        ])
-      }
-      if (autoManualProductId) {
-        await query(`DELETE FROM sales_daily_facts WHERE canonical_product_id = $1`, [
-          autoManualProductId,
-        ])
-        await query(`DELETE FROM cost_versions WHERE canonical_product_id = $1`, [
-          autoManualProductId,
-        ])
-        await query(`DELETE FROM product_catalog WHERE id = $1`, [autoManualProductId])
      }
      if (retryWorkOrderRuleId) {
        await query(`DELETE FROM work_product_match_logs WHERE rule_id = $1`, [