diff --git a/apps/backend/src/db/migrations/136_reconcile_channel_products_and_work_costs.sql b/apps/backend/src/db/migrations/136_reconcile_channel_products_and_work_costs.sql new file mode 100644 index 00000000..c550830f --- /dev/null +++ b/apps/backend/src/db/migrations/136_reconcile_channel_products_and_work_costs.sql @@ -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 标准商品，历史归档仍保留关联'; diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 4e08ab3c..2fef575f 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -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(`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 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 `, diff --git a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts index 46131fc6..4048286d 100644 --- a/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts +++ b/apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts @@ -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 + }>( + `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 + 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`, [