diff --git a/apps/backend/src/db/migrations/115_product_channel_capabilities.sql b/apps/backend/src/db/migrations/115_product_channel_capabilities.sql new file mode 100644 index 00000000..5562fa34 --- /dev/null +++ b/apps/backend/src/db/migrations/115_product_channel_capabilities.sql @@ -0,0 +1,100 @@ +-- 115_product_channel_capabilities.sql +-- +-- 渠道能力与履约兜底: +-- 1. 商品可同时具备多个渠道能力(自动发货 + 人工兜底); +-- 2. 订单项记录原始渠道与切换原因; +-- 3. 成本快照支持冲销(reversed_at),切换渠道时不双算; +-- 4. fulfillment_source 增加 affiliate_fallback。 + +CREATE TABLE IF NOT EXISTS product_channel_capabilities ( + product_id BIGINT NOT NULL REFERENCES product_catalog(id) ON DELETE CASCADE, + channel TEXT NOT NULL, + priority INTEGER NOT NULL DEFAULT 100, + is_fallback BOOLEAN NOT NULL DEFAULT FALSE, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + source TEXT NOT NULL DEFAULT 'auto', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (product_id, channel) +); + +COMMENT ON TABLE product_channel_capabilities IS + '商品渠道能力:一个商品可同时支持自动发货、人工等;priority 越小越优先'; + +-- 自动发货商品:默认自动优先,人工兜底 +INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback) +SELECT p.id, 'affiliate_dash', 10, FALSE +FROM product_catalog p +WHERE p.canonical_sku LIKE 'aff:%' AND p.status = 'active' +ON CONFLICT (product_id, channel) DO NOTHING; + +INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback) +SELECT p.id, 'manual', 90, TRUE +FROM product_catalog p +WHERE p.canonical_sku LIKE 'aff:%' AND p.status = 'active' +ON CONFLICT (product_id, channel) DO NOTHING; + +-- 人工商品:人工优先 +INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback) +SELECT p.id, 'manual', 10, FALSE +FROM product_catalog p +WHERE p.canonical_sku LIKE 'man:%' AND p.status = 'active' +ON CONFLICT (product_id, channel) DO NOTHING; + +-- 卡券:只统计 +INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback) +SELECT p.id, 'third_party_voucher', 10, FALSE +FROM product_catalog p +WHERE p.canonical_sku LIKE 'tpv:%' AND p.status = 'active' +ON CONFLICT (product_id, channel) DO NOTHING; + +-- 接单模板/工单商品 +INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback) +SELECT p.id, 'work_order', 10, FALSE +FROM product_catalog p +WHERE (p.canonical_sku LIKE 'tpl:%' OR p.canonical_sku LIKE 'wo:%') AND p.status = 'active' +ON CONFLICT (product_id, channel) DO NOTHING; + +ALTER TABLE order_items + ADD COLUMN IF NOT EXISTS original_channel TEXT, + ADD COLUMN IF NOT EXISTS channel_switch_reason TEXT NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS channel_switched_at TIMESTAMPTZ; + +COMMENT ON COLUMN order_items.original_channel IS + '首次分流渠道;fallback/运营改渠道后仍保留,用于审计'; +COMMENT ON COLUMN order_items.channel_switch_reason IS + '渠道切换原因:affiliate_failed / manual_supplement / operator'; + +ALTER TABLE cost_snapshots + ADD COLUMN IF NOT EXISTS reversed_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS reversal_reason TEXT NOT NULL DEFAULT ''; + +DROP INDEX IF EXISTS uq_cost_snapshots_active_subject_item; +CREATE UNIQUE INDEX IF NOT EXISTS uq_cost_snapshots_active_subject_item + ON cost_snapshots (subject_type, subject_id, cost_item_id) + WHERE superseded_at IS NULL AND reversed_at IS NULL; + +-- 人工成本快照同样排除已冲销行,允许冲销后重新确认。 +DROP INDEX IF EXISTS uq_cost_snapshots_manual_order_item; +CREATE UNIQUE INDEX IF NOT EXISTS uq_cost_snapshots_manual_order_item + ON cost_snapshots (order_item_id) + WHERE subject_type = 'manual_case' AND superseded_at IS NULL + AND reversed_at IS NULL AND subject_id > 0; + +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM pg_constraint WHERE conname = 'chk_order_items_fulfillment_source' + ) THEN + ALTER TABLE order_items DROP CONSTRAINT chk_order_items_fulfillment_source; + END IF; + ALTER TABLE order_items + ADD CONSTRAINT chk_order_items_fulfillment_source + CHECK ( + fulfillment_source IN ( + '', 'auto_rule', 'auto_mapping', 'carrier_evidence', + 'manual_confirm', 'operator_assigned', 'cutover_archive', + 'affiliate_fallback' + ) + ); +END $$; diff --git a/apps/backend/src/services/cost-center/cost-center-constants.ts b/apps/backend/src/services/cost-center/cost-center-constants.ts index daca936f..de0aa318 100644 --- a/apps/backend/src/services/cost-center/cost-center-constants.ts +++ b/apps/backend/src/services/cost-center/cost-center-constants.ts @@ -106,11 +106,12 @@ export const WORK_ORDER_TEMPLATE_COST_BASELINE = '2026-07-10T00:00:00+08:00' */ export const PRODUCT_SOURCE_PREFIXES: Array<{ prefix: string - source: 'affiliate_dash' | 'work_order' | 'manual' + source: 'affiliate_dash' | 'work_order' | 'third_party_voucher' | 'manual' }> = [ { prefix: 'aff:', source: 'affiliate_dash' }, { prefix: 'tpl:', source: 'work_order' }, { prefix: 'wo:', source: 'work_order' }, + { prefix: 'tpv:', source: 'third_party_voucher' }, { prefix: 'man:', source: 'manual' }, ] diff --git a/apps/backend/src/services/cost-center/cost-center.test.ts b/apps/backend/src/services/cost-center/cost-center.test.ts index 030ec22f..cef18d66 100644 --- a/apps/backend/src/services/cost-center/cost-center.test.ts +++ b/apps/backend/src/services/cost-center/cost-center.test.ts @@ -298,3 +298,10 @@ test('成本模型与商品来源必须覆盖第三方卡券渠道', () => { assert.ok(productSourceFilterSql(1).includes("'third_party_voucher'")) assert.ok(productSourceFilterSql(1).includes("'tpv:%'")) }) + +test('tpv: 商品来源归为第三方卡券', async () => { + const { productSourceOf } = await import('./cost-center-mappers.js') + const assert = (await import('node:assert/strict')).default + assert.equal(productSourceOf('tpv:模拟卡券'), 'third_party_voucher') + assert.equal(productSourceOf('man:模拟人工'), 'manual') +}) diff --git a/apps/backend/src/services/cost-center/cost-exception-service.ts b/apps/backend/src/services/cost-center/cost-exception-service.ts index 0e13d3ee..3321b47f 100644 --- a/apps/backend/src/services/cost-center/cost-exception-service.ts +++ b/apps/backend/src/services/cost-center/cost-exception-service.ts @@ -295,6 +295,25 @@ export async function resolveExceptionsForItemsByReason( return result.rowCount || 0 } +/** + * 清理“头还在、明细已空”的孤儿异常。 + * 并发收口时可能出现明细已删但头未关闭,按 item 再查也找不到它。 + */ +export async function resolveOrphanCostExceptions(): Promise { + const result = await query<{ id: number }>( + ` + UPDATE cost_exceptions ce + SET status = 'resolved', resolved_at = NOW(), item_count = 0, last_seen_at = NOW() + WHERE ce.status <> 'resolved' + AND NOT EXISTS ( + SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id + ) + RETURNING ce.id + `, + ) + return result.rowCount || 0 +} + export async function resolveExceptionsForItems(orderItemIds: number[]): Promise { const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0) if (ids.length === 0) return 0 diff --git a/apps/backend/src/services/cost-center/cost-resolution-service.ts b/apps/backend/src/services/cost-center/cost-resolution-service.ts index 934dfc39..67bc7fca 100644 --- a/apps/backend/src/services/cost-center/cost-resolution-service.ts +++ b/apps/backend/src/services/cost-center/cost-resolution-service.ts @@ -177,7 +177,7 @@ export async function createAdminCostSnapshot(payload: JsonObject = {}, _session WHERE subject_type = 'order_item' AND subject_id = $1 AND cost_item_id = $2 - AND superseded_at IS NULL + AND superseded_at IS NULL AND reversed_at IS NULL FOR UPDATE `, [orderItemId, line.costItemId], @@ -194,7 +194,7 @@ export async function createAdminCostSnapshot(payload: JsonObject = {}, _session continue } await client.query( - `UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL`, + `UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL AND reversed_at IS NULL`, [now, current.id], ) supersededCount += 1 @@ -212,7 +212,7 @@ export async function createAdminCostSnapshot(payload: JsonObject = {}, _session $7, $8, $9, $10, 'resolved', $11::jsonb, $12, $13, $13 ) ON CONFLICT (subject_type, subject_id, cost_item_id) - WHERE superseded_at IS NULL + WHERE superseded_at IS NULL AND reversed_at IS NULL DO NOTHING RETURNING * `, diff --git a/apps/backend/src/services/cost-center/cost-snapshot-service.ts b/apps/backend/src/services/cost-center/cost-snapshot-service.ts index c7708bba..dcb65109 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -4,6 +4,7 @@ import { nowIso } from '../../utils/time.js' import { GENERIC_COST_MODELS } from './cost-center-constants.js' import { resolveExceptionsForItems, + resolveExceptionsForItemsByReason, upsertAggregatedExceptionsBatch, type AggregatedExceptionEntry, } from './cost-exception-service.js' @@ -371,6 +372,7 @@ async function processCostRows( const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable }> = [] const supersedeIds: number[] = [] const missingEntries: AggregatedExceptionEntry[] = [] + const resolvedMissingIds: number[] = [] for (const item of rows) { const desired: Array> = [] @@ -381,15 +383,25 @@ async function processCostRows( if (channelResolution) { desired.push(channelResolution) } else if (code) { - result.missing += 1 - // 已在队列里的缺失项不再重复写三条 SQL,等价格补齐后自然收口。 - if (!openMissing.has(Number(item.id))) { - missingEntries.push({ - reason: 'cost_missing', - scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`, - orderItemId: Number(item.id), - detail: { channel: item.fulfillment_channel }, - }) + // 历史版本可能已 retired,但订单已有不可变快照:这单不算缺成本, + // 否则“改价后旧订单快照还在、异常却重新冒出来”。 + const channelCostItem = ctx.channelCostItems.get(code) + const hasActiveChannelSnapshot = channelCostItem + ? existingMap.get(Number(item.id))?.has(Number(channelCostItem.id)) + : false + if (hasActiveChannelSnapshot) { + resolvedMissingIds.push(Number(item.id)) + } else { + result.missing += 1 + // 已在队列里的缺失项不再重复写三条 SQL,等价格补齐后自然收口。 + if (!openMissing.has(Number(item.id))) { + missingEntries.push({ + reason: 'cost_missing', + scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`, + orderItemId: Number(item.id), + detail: { channel: item.fulfillment_channel }, + }) + } } } @@ -429,6 +441,11 @@ async function processCostRows( } } + if (resolvedMissingIds.length > 0) { + await resolveExceptionsForItemsByReason(resolvedMissingIds, 'cost_missing').catch( + () => undefined, + ) + } if (missingEntries.length > 0) { await upsertAggregatedExceptionsBatch(missingEntries) } @@ -583,7 +600,7 @@ async function loadActiveSnapshots( ` SELECT id, order_item_id, cost_item_id, cost_version_id, total_amount_fen, snapshot_stage FROM cost_snapshots - WHERE superseded_at IS NULL + WHERE superseded_at IS NULL AND reversed_at IS NULL AND order_item_id = ANY($1::bigint[]) `, [itemIds], @@ -655,7 +672,7 @@ async function insertSnapshots( product_id, quantity, unit_amount_fen, total_amount_fen, stage, channel, basis_at ) - ON CONFLICT (subject_type, subject_id, cost_item_id) WHERE superseded_at IS NULL + ON CONFLICT (subject_type, subject_id, cost_item_id) WHERE superseded_at IS NULL AND reversed_at IS NULL DO NOTHING `, [ diff --git a/apps/backend/src/services/cost-center/product-catalog-service.ts b/apps/backend/src/services/cost-center/product-catalog-service.ts index 17949039..b449accd 100644 --- a/apps/backend/src/services/cost-center/product-catalog-service.ts +++ b/apps/backend/src/services/cost-center/product-catalog-service.ts @@ -18,6 +18,12 @@ import { resolveSkuNameQuantity } from '../product-match/pricing.js' 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 { + resolveExceptionsForItemsByReason, + resolveOrphanCostExceptions, +} from './cost-exception-service.js' +import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js' import { syncAdminWorkProductRulePrice } from '../worker-platform/admin-worker-catalog-service.js' import { AFFILIATE_TASK_AGG_JOIN, @@ -1372,7 +1378,7 @@ export async function setAdminProductPrice( } const now = nowIso() - const effectiveFrom = resolvePriceEffectiveFrom(payload) + let effectiveFrom = resolvePriceEffectiveFrom(payload) const changeReason = String(payload.changeReason || '在商品与成本中心统一调整价格').trim() // 2. 如果是接单模板商品 (tpl:) @@ -1548,6 +1554,35 @@ export async function setAdminProductPrice( ) const existing = existingResult.rows[0] const existingTime = existing ? Date.parse(existing.effective_from) : null + + // 首次配置/补价:如果这个商品还有未算成本的订单,默认从最早缺失订单当天生效, + // 否则当天早些时候的订单会因为“版本生效时间晚于下单时间”永远算不到成本。 + effectiveFrom = resolvePriceEffectiveFrom(payload) + const requestedEffective = payload.effectiveFrom ?? payload.effective_from + if (!requestedEffective) { + const earliest = await query<{ basis_date: string | null }>( + ` + SELECT MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)::text AS basis_date + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + WHERE oi.canonical_product_id = $1 + AND oi.fulfillment_status = 'assigned' + AND NOT EXISTS ( + SELECT 1 FROM cost_snapshots cs + WHERE cs.order_item_id = oi.id + AND cs.cost_item_id = $2 + AND cs.superseded_at IS NULL + AND cs.reversed_at IS NULL + ) + `, + [product.id, costItemId], + ) + const earliestDate = String(earliest.rows[0]?.basis_date || '').trim() + const earliestTime = earliestDate ? Date.parse(`${earliestDate}T00:00:00+08:00`) : NaN + if (Number.isFinite(earliestTime) && (existingTime === null || earliestTime < existingTime)) { + effectiveFrom = `${earliestDate}T00:00:00+08:00` + } + } const targetTime = Date.parse(effectiveFrom) let costVersionId = existing?.id || 0 @@ -1606,6 +1641,59 @@ export async function setAdminProductPrice( costVersionId = inserted.rows[0]?.id || 0 } + // 配置后立刻补算该商品下缺成本的订单,避免用户看到“已配置但异常还在”。 + const missingItems = await query<{ id: number }>( + ` + SELECT oi.id + FROM order_items oi + WHERE oi.canonical_product_id = $1 + AND oi.fulfillment_status = 'assigned' + AND NOT EXISTS ( + SELECT 1 FROM cost_snapshots cs + WHERE cs.order_item_id = oi.id + AND cs.cost_item_id = $2 + AND cs.superseded_at IS NULL + AND cs.reversed_at IS NULL + ) + ORDER BY oi.id + LIMIT 2000 + `, + [product.id, costItemId], + ) + if (missingItems.rows.length > 0) { + const ids = missingItems.rows.map((row) => Number(row.id)) + await syncCostSnapshotsForItems(ids).catch(() => undefined) + const dates = await orderItemBasisDates(ids) + await refreshFactsForDates(dates).catch(() => undefined) + } + + // 已有快照但异常还挂着的(并发/历史遗留),保存成本时一并收口。 + const coveredItems = await query<{ id: number }>( + ` + SELECT oi.id + FROM order_items oi + WHERE oi.canonical_product_id = $1 + AND oi.fulfillment_status = 'assigned' + AND EXISTS ( + SELECT 1 FROM cost_snapshots cs + WHERE cs.order_item_id = oi.id + AND cs.cost_item_id = $2 + AND cs.superseded_at IS NULL + AND cs.reversed_at IS NULL + ) + ORDER BY oi.id + LIMIT 2000 + `, + [product.id, costItemId], + ) + if (coveredItems.rows.length > 0) { + await resolveExceptionsForItemsByReason( + coveredItems.rows.map((row) => Number(row.id)), + 'cost_missing', + ).catch(() => undefined) + } + await resolveOrphanCostExceptions().catch(() => undefined) + return { success: true, product, diff --git a/apps/backend/src/services/cost-center/product-normalize-service.test.ts b/apps/backend/src/services/cost-center/product-normalize-service.test.ts index 0f73d9af..39b7125d 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.test.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.test.ts @@ -1,7 +1,11 @@ import assert from 'node:assert/strict' import test from 'node:test' -import { buildMappingIndex, resolveProductId } from './product-normalize-service.js' +import { + buildMappingIndex, + resolveProductId, + voucherCanonicalSku, +} from './product-normalize-service.js' const mappings = [ { @@ -134,3 +138,24 @@ test('商品归一:第三方卡券按 tpv:SKU 直接命中', () => { 41, ) }) + +test('三方卡券自动建商品时优先使用内部中文 SKU', () => { + assert.equal( + voucherCanonicalSku({ + manual_sku_key: '11月改名卡', + sku_name: '11月改名卡', + sku_code: '192592489694733', + }), + 'tpv:11月改名卡', + ) + assert.equal( + voucherCanonicalSku({ + manual_sku_key: '192592489694733', + sku_name: '11月改名卡', + sku_code: '192592489694733', + }), + 'tpv:11月改名卡', + ) + assert.equal(voucherCanonicalSku({ manual_sku_key: '', sku_name: '', sku_code: 'x' }), 'tpv:x') + assert.equal(voucherCanonicalSku({}), null) +}) diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts index 6ea04101..9403ff05 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -5,7 +5,8 @@ import { readWatermarkItemId } from '../watermark-store.js' /** * 管道 B:商品归一。 * - * 只写 `order_items.canonical_product_id`,不建商品、不改渠道、不碰成本。 + * 只写 `order_items.canonical_product_id`,不改渠道、不碰成本; + * 例外:三方卡券缺商品时按内部中文 SKU 自动建 `tpv:` 商品并补渠道规则。 * * 解析顺序(与旧的四条回填路径统一): * 1. 自动发货渠道:`aff:` 商品直接命中(该渠道的商品就是 SKU 本身); @@ -43,6 +44,7 @@ type ItemRow = { shop_id: string sku_code: string sku_name: string + manual_sku_key: string | null normalized_sku_name: string | null normalized_sku_series: string | null fulfillment_channel: string | null @@ -110,6 +112,94 @@ export function buildMappingIndex( return index } +/** 三方卡券的内部 SKU:优先中文,其次名称/平台码。 */ +export function pickVoucherInternalSku(item: { + manual_sku_key?: string | null + sku_name?: string | null + sku_code?: string | null +}): string { + const candidates = [item.manual_sku_key, item.sku_name, item.sku_code] + .map((value) => String(value || '').trim()) + .filter(Boolean) + return candidates.find((value) => /[\u4e00-\u9fff]/.test(value)) || candidates[0] || '' +} + +export function voucherCanonicalSku(item: { + manual_sku_key?: string | null + sku_name?: string | null + sku_code?: string | null +}): string | null { + const internal = pickVoucherInternalSku(item) + return internal ? `tpv:${internal}` : null +} + +/** + * 三方卡券没有预先建库时,按内部 SKU 自动建 `tpv:` 商品并补渠道规则。 + * 建完后该商品会出现在「商品与成本中心 → 第三方卡券」,运营直接设采购价。 + */ +async function ensureVoucherProductForItem(item: ItemRow): Promise { + const canonicalSku = voucherCanonicalSku(item) + const internalSku = pickVoucherInternalSku(item) + if (!canonicalSku || !internalSku) return null + + const existing = await query<{ id: number; status: string }>( + 'SELECT id, status FROM product_catalog WHERE canonical_sku = $1 LIMIT 1', + [canonicalSku], + ) + let productId = Number(existing.rows[0]?.id || 0) + if (!productId) { + const created = 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', '{"source":"third_party_voucher","auto_created":true}'::jsonb, NOW(), NOW()) + ON CONFLICT (canonical_sku) DO NOTHING + RETURNING id + `, + [canonicalSku, internalSku], + ) + productId = Number(created.rows[0]?.id || 0) + if (!productId) { + const again = await query<{ id: number }>( + 'SELECT id FROM product_catalog WHERE canonical_sku = $1 LIMIT 1', + [canonicalSku], + ) + productId = Number(again.rows[0]?.id || 0) + } + } + if (!productId) return null + + const current = await query<{ status: string }>( + 'SELECT status FROM product_catalog WHERE id = $1', + [productId], + ) + 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], + ) + + const skuCode = String(item.sku_code || '').trim() + if (skuCode) { + await query( + ` + INSERT INTO sku_channel_rules ( + provider, shop_id, match_key_type, match_key, channel, priority, + source, enabled, note, created_at, updated_at + ) + VALUES ($1, $2, 'sku_code', $3, 'third_party_voucher', 10, 'voucher', TRUE, + '卡券订单自动建商品后生成', NOW(), NOW()) + ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING + `, + [item.provider, item.shop_id, skuCode], + ) + } + return productId +} + /** 纯函数:按索引解析商品,便于单测。 */ export function resolveProductId(item: ItemRow, index: MappingIndex): number | null { const provider = String(item.provider || '').trim() @@ -204,6 +294,7 @@ export async function normalizeOrderItemProducts( o.shop_id, oi.sku_code, oi.sku_name, + oi.manual_sku_key, oi.normalized_sku_name, oi.normalized_sku_series, oi.fulfillment_channel, @@ -235,7 +326,10 @@ export async function normalizeOrderItemProducts( const matched: Array<{ id: number; productId: number }> = [] for (const item of rows.rows) { - const productId = resolveProductId(item, index) + let productId = resolveProductId(item, index) + if (!productId && item.fulfillment_channel === 'third_party_voucher') { + productId = await ensureVoucherProductForItem(item) + } if (!productId) { result.unmatched += 1 await upsertAggregatedException({ @@ -338,6 +432,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{ o.shop_id, oi.sku_code, oi.sku_name, + oi.manual_sku_key, oi.normalized_sku_name, oi.normalized_sku_series, oi.fulfillment_channel, @@ -357,7 +452,10 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{ const matched: Array<{ id: number; productId: number }> = [] let unmatched = 0 for (const item of rows.rows) { - const productId = resolveProductId(item, index) + let productId = resolveProductId(item, index) + if (!productId && item.fulfillment_channel === 'third_party_voucher') { + productId = await ensureVoucherProductForItem(item) + } if (!productId) { unmatched += 1 await upsertAggregatedException({ diff --git a/apps/backend/src/services/cost-center/work-order-product-service.ts b/apps/backend/src/services/cost-center/work-order-product-service.ts index 7a70aa1d..8c36dae7 100644 --- a/apps/backend/src/services/cost-center/work-order-product-service.ts +++ b/apps/backend/src/services/cost-center/work-order-product-service.ts @@ -451,7 +451,7 @@ export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) { JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id WHERE cs.order_item_id = wo.order_item_id AND cs.cost_item_id = rv.cost_item_id - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.cost_version_id <> rv.cost_version_id AND EXISTS ( SELECT 1 FROM cost_versions gv diff --git a/apps/backend/src/services/finance/finance-ledger-service.ts b/apps/backend/src/services/finance/finance-ledger-service.ts index daf9ea9e..2e18ccb0 100644 --- a/apps/backend/src/services/finance/finance-ledger-service.ts +++ b/apps/backend/src/services/finance/finance-ledger-service.ts @@ -622,7 +622,7 @@ async function assertPeriodCloseable( SELECT 1 FROM cost_snapshots cs WHERE cs.subject_type = 'manual_case' AND cs.order_item_id = oi.id - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL ) `, [dateFrom, dateTo], @@ -639,7 +639,7 @@ async function assertPeriodCloseable( JOIN order_items oi ON oi.id = cs.order_item_id JOIN orders o ON o.id = oi.order_id WHERE cs.subject_type = 'manual_case' - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date AND cs.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs) @@ -661,7 +661,7 @@ async function assertPeriodCloseable( JOIN order_items oi ON oi.id = cs.order_item_id JOIN orders o ON o.id = oi.order_id WHERE cs.subject_type = 'manual_case' - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date AND cs.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs) diff --git a/apps/backend/src/services/finance/manual-fulfillment-service.ts b/apps/backend/src/services/finance/manual-fulfillment-service.ts index 9586525d..ae12cbd5 100644 --- a/apps/backend/src/services/finance/manual-fulfillment-service.ts +++ b/apps/backend/src/services/finance/manual-fulfillment-service.ts @@ -134,7 +134,7 @@ export async function listManualFulfillmentPending( cs.order_item_id, MAX(cs.created_at) AS fulfilled_at FROM cost_snapshots cs - WHERE cs.subject_type = 'manual_case' AND cs.superseded_at IS NULL + WHERE cs.subject_type = 'manual_case' AND cs.superseded_at IS NULL AND reversed_at IS NULL GROUP BY cs.order_item_id ) ` @@ -294,7 +294,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm // 锁内复查:并发/双击的第二个请求会等锁,醒来后看到快照直接拒绝,不会重复扣款。 const existing = await client.query<{ id: number }>( `SELECT id FROM cost_snapshots - WHERE subject_type = 'manual_case' AND order_item_id = $1 AND superseded_at IS NULL + WHERE subject_type = 'manual_case' AND order_item_id = $1 AND superseded_at IS NULL AND reversed_at IS NULL LIMIT 1`, [orderItemId], ) @@ -874,7 +874,7 @@ export async function getManualFulfillmentCountsByOrderIds( WITH fulfilled AS ( SELECT order_item_id, MAX(created_at) AS fulfilled_at FROM cost_snapshots - WHERE subject_type = 'manual_case' AND superseded_at IS NULL + WHERE subject_type = 'manual_case' AND superseded_at IS NULL AND reversed_at IS NULL GROUP BY order_item_id ) SELECT diff --git a/apps/backend/src/services/finance/sales-facts-channel-service.ts b/apps/backend/src/services/finance/sales-facts-channel-service.ts index 2acfb934..c006ef42 100644 --- a/apps/backend/src/services/finance/sales-facts-channel-service.ts +++ b/apps/backend/src/services/finance/sales-facts-channel-service.ts @@ -105,18 +105,18 @@ export async function rebuildChannelSalesDailyFacts(input: { FROM cost_snapshots cs JOIN cost_items ci ON ci.id = cs.cost_item_id WHERE cs.order_item_id = s.order_item_id - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL AND ci.code = s.channel_cost_code ) WHEN s.channel = 'manual' THEN NOT EXISTS ( SELECT 1 FROM cost_snapshots cs WHERE cs.order_item_id = s.order_item_id - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL AND cs.subject_type = 'manual_case' ) ELSE NOT EXISTS ( SELECT 1 FROM cost_snapshots cs - WHERE cs.order_item_id = s.order_item_id AND cs.superseded_at IS NULL + WHERE cs.order_item_id = s.order_item_id AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL ) END AS cost_missing, COALESCE(c.has_actual, FALSE) AS has_actual @@ -128,7 +128,7 @@ export async function rebuildChannelSalesDailyFacts(input: { SUM(cs.total_amount_fen)::bigint AS cost_fen, BOOL_AND(cs.snapshot_stage = 'actual') AS has_actual FROM cost_snapshots cs - WHERE cs.superseded_at IS NULL + WHERE cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL GROUP BY cs.order_item_id ) c ON c.order_item_id = s.order_item_id ) diff --git a/apps/backend/src/services/fulfillment/affiliate-fallback-service.test.ts b/apps/backend/src/services/fulfillment/affiliate-fallback-service.test.ts new file mode 100644 index 00000000..73604758 --- /dev/null +++ b/apps/backend/src/services/fulfillment/affiliate-fallback-service.test.ts @@ -0,0 +1,12 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { AFFILIATE_FAILURE_STATUSES } from './affiliate-fallback-service.js' + +test('自动发货失败状态全部纳入转人工', () => { + for (const status of ['closed', 'expired', 'manual_review', 'retry_pending', 'cancelled']) { + assert.ok(AFFILIATE_FAILURE_STATUSES.includes(status as never)) + } + assert.ok(!AFFILIATE_FAILURE_STATUSES.includes('redeemed' as never)) + assert.ok(!AFFILIATE_FAILURE_STATUSES.includes('waiting_binding' as never)) +}) diff --git a/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts b/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts new file mode 100644 index 00000000..78d2df25 --- /dev/null +++ b/apps/backend/src/services/fulfillment/affiliate-fallback-service.ts @@ -0,0 +1,122 @@ +import { query, withTransaction } from '../../db/client.js' +import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js' +import { refreshFactsForDates } from '../finance/refresh-facts.js' + +/** + * 自动发货失败 → 人工兜底。 + * + * 触发:该订单项最新一条 affiliate_dash 任务处于失败状态,且没有任何成功任务。 + * 切换后:原渠道成本(sys_affiliate_purchase)冲销,人工成本等确认时再记; + * 自动任务本身不改状态,只记录订单项的切换原因,避免重复触发。 + */ +export const AFFILIATE_FAILURE_STATUSES = [ + 'closed', + 'expired', + 'manual_review', + 'retry_pending', + 'cancelled', +] as const + +const AFFILIATE_SUCCESS_STATUSES = ['redeemed', 'paid'] as const + +export async function runAffiliateFallback( + options: { limit?: number } = {}, +): Promise<{ scanned: number; switched: number }> { + const limit = Math.min(1000, Math.max(1, Math.trunc(options.limit || 200))) + const candidates = await query<{ + id: number + order_item_id: number + provider: string + shop_id: string + manual_sku_key: string | null + sku_name: string + config_id: number | null + basis_date: string + }>( + ` + SELECT + oi.id, + oi.id AS order_item_id, + o.provider, + o.shop_id, + oi.manual_sku_key, + oi.sku_name, + cfg.id AS config_id, + (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + LEFT JOIN manual_fulfillment_configs cfg + ON cfg.sku_key = oi.manual_sku_key AND cfg.enabled + WHERE oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel = 'affiliate_dash' + AND oi.fulfillment_locked = FALSE + AND EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id + AND ft.executor_key = 'affiliate_dash' + AND ft.task_status = ANY($1::text[]) + ) + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id + AND ft.executor_key = 'affiliate_dash' + AND ft.task_status = ANY($2::text[]) + ) + ORDER BY oi.id + LIMIT $3 + `, + [AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit], + ) + + const switchedIds: number[] = [] + const affectedDates = new Set() + + for (const row of candidates.rows) { + const updated = await withTransaction(async (client) => { + const result = await client.query( + ` + UPDATE order_items + SET + original_channel = COALESCE(original_channel, 'affiliate_dash'), + fulfillment_channel = 'manual', + fulfillment_status = 'assigned', + fulfillment_source = 'affiliate_fallback', + channel_switch_reason = 'affiliate_failed', + channel_switched_at = NOW(), + fulfillment_config_id = COALESCE(fulfillment_config_id, $2), + updated_at = NOW() + WHERE id = $1 + AND fulfillment_status = 'assigned' + AND fulfillment_channel = 'affiliate_dash' + AND fulfillment_locked = FALSE + RETURNING id + `, + [row.id, row.config_id], + ) + if ((result.rowCount || 0) === 0) return false + + await client.query( + ` + UPDATE cost_snapshots + SET reversed_at = NOW(), reversal_reason = 'affiliate_failed' + WHERE order_item_id = $1 + AND cost_item_id = (SELECT id FROM cost_items WHERE code = 'sys_affiliate_purchase') + AND superseded_at IS NULL AND reversed_at IS NULL + AND reversed_at IS NULL + `, + [row.id], + ) + return true + }) + if (!updated) continue + switchedIds.push(Number(row.id)) + affectedDates.add(row.basis_date) + } + + for (const id of switchedIds) { + await syncCostSnapshotsForItems([id]).catch(() => undefined) + } + await refreshFactsForDates([...affectedDates]).catch(() => undefined) + + return { scanned: candidates.rows.length, switched: switchedIds.length } +} diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 19c1a8b9..303e3dfd 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -1,8 +1,10 @@ import type { JsonObject } from '../../types/json.js' import { query } from '../../db/client.js' import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js' +import { runAffiliateFallback } from '../fulfillment/affiliate-fallback-service.js' import { resolveExceptionsForItemsByReason, + resolveOrphanCostExceptions, upsertAggregatedExceptionsBatch, type AggregatedExceptionEntry, } from '../cost-center/cost-exception-service.js' @@ -167,7 +169,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { SELECT 1 FROM cost_snapshots cs WHERE cs.subject_type = 'manual_case' AND cs.order_item_id = oi.id - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL ) AND NOT EXISTS ( SELECT 1 @@ -252,17 +254,23 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { } const resolved = await resolveCarrierDispatchExceptions() + const fallback = await runAffiliateFallback({ limit: 200 }).catch(() => ({ + scanned: 0, + switched: 0, + })) const nextCursor = maxId === 0 ? 0 : maxId await touchWatermark('carrier_dispatch', { lastItemId: nextCursor }) return { status: 'ok', message: `人工载体挂载 ${attached},缺配置 ${missingManual.rows.length},` + - `缺履约载体 ${missingCarrier.rows.length},异常收口 ${resolved}`, + `缺履约载体 ${missingCarrier.rows.length},异常收口 ${resolved},` + + `自动转人工 ${fallback.switched}`, attached, missingManual: missingManual.rows.length, missingCarrier: missingCarrier.rows.length, resolved, + affiliateFallback: fallback.switched, } } @@ -335,7 +343,7 @@ async function resolveCarrierDispatchExceptions(): Promise { SELECT 1 FROM cost_snapshots cs WHERE cs.subject_type = 'manual_case' AND cs.order_item_id = i.order_item_id - AND cs.superseded_at IS NULL + AND cs.superseded_at IS NULL AND reversed_at IS NULL ) LIMIT 500 `, @@ -383,6 +391,7 @@ export async function runCostSnapshotJob(job: JsonObject = {}) { await touchWatermark('cost_snapshot', { lastItemId: result.nextCursor }) await touchWatermark('cost_snapshot_hot', { lastItemId: result.hotCursor }) const refreshed = await refreshFactsForDates(result.affectedDates) + await resolveOrphanCostExceptions().catch(() => undefined) return { status: 'ok', message: