diff --git a/apps/backend/package.json b/apps/backend/package.json index cd67a411..46a33a59 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -11,8 +11,6 @@ "db:migrate:status": "tsx scripts/migration-status.ts", "finance:sync": "tsx scripts/sync-cost-finance.ts", "finance:rebuild-cutover": "tsx scripts/rebuild-after-cutover.ts", - "fulfillment:backfill": "tsx scripts/backfill-fulfillment-mode.ts", - "fulfillment:backfill-channel": "tsx scripts/backfill-fulfillment-channel.ts", "revenue:backfill": "tsx scripts/backfill-revenue-basis.ts", "cost-exceptions:archive": "tsx scripts/archive-cost-exceptions.ts", "storage:migrate": "tsx scripts/migrate-storage.ts", diff --git a/apps/backend/scripts/backfill-fulfillment-channel.ts b/apps/backend/scripts/backfill-fulfillment-channel.ts deleted file mode 100644 index 440f8e2f..00000000 --- a/apps/backend/scripts/backfill-fulfillment-channel.ts +++ /dev/null @@ -1,241 +0,0 @@ -import process from 'node:process' - -import { closeDb, query } from '../src/db/client.js' - -/** - * 112 之后的渠道回填：把旧 `fulfillment_mode` 转成 - * `fulfillment_channel` + `fulfillment_status` + `fulfillment_source`。 - * - * 规则（2026-10-31 断代）： - * 1. 业务日期 < 2026-10-31 → status=archived，source=cutover_archive； - * channel 按旧 mode 映射（legacy_platform 没有新渠道，置空）。 - * 2. 业务日期 >= 2026-10-31： - * · work_order / affiliate_dash → status=assigned，source=carrier_evidence； - * · manual → status=pending（旧 manual 是兜底值，必须由运营重新确认）； - * · 其余 / NULL → status=pending。 - * - * 幂等：每批按 id 重算，只有值变化才写；运营锁定行（fulfillment_locked）跳过。 - * 分批：keyset 扫描，单批默认 5000，避免 438k 行整表 UPDATE 撞语句超时。 - * - * 用法： - * npm run fulfillment:backfill-channel -- --dry-run - * npm run fulfillment:backfill-channel -- --batch-size=5000 - * npm run fulfillment:backfill-channel -- --max-batches=100 --after-id=123456 - */ - -/** 财务断代日：这一天之前的数据只保留审计价值。 */ -const CUTOVER_DATE = '2026-10-31' - -const options = parseOptions(process.argv.slice(2)) -const batchSize = normalizeInteger(options.batchSize, 5000, 1, 50_000) -const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 100_000) -const afterId = normalizeInteger(options.afterId, 0, 0, Number.MAX_SAFE_INTEGER) -const dryRun = options.dryRun === 'true' || options['dry-run'] === 'true' - -const CHANNEL_SQL = ` - CASE oi.fulfillment_mode - WHEN 'work_order' THEN 'work_order' - WHEN 'affiliate_dash' THEN 'affiliate_dash' - WHEN 'manual' THEN 'manual' - ELSE NULL - END -` - -const DECISION_SQL = ` - CASE - WHEN (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - < DATE '${CUTOVER_DATE}' THEN 'archived' - WHEN oi.fulfillment_mode = 'manual' THEN 'pending' - WHEN oi.fulfillment_mode IN ('work_order', 'affiliate_dash') THEN 'assigned' - ELSE 'pending' - END -` - -const SOURCE_SQL = ` - CASE - WHEN (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - < DATE '${CUTOVER_DATE}' THEN 'cutover_archive' - WHEN oi.fulfillment_mode IN ('work_order', 'affiliate_dash') THEN 'carrier_evidence' - ELSE '' - END -` - -type BatchRow = { - id: number - channel: string | null - status: string - source: string - changed: boolean -} - -async function main() { - await assertSchema() - - const startedAt = Date.now() - let cursor = afterId - let scanned = 0 - let updated = 0 - const summary = new Map() - - for (let batch = 0; batch < maxBatches; batch += 1) { - const rows = await query( - ` - SELECT - oi.id, - ${CHANNEL_SQL} AS channel, - ${DECISION_SQL} AS status, - ${SOURCE_SQL} AS source, - ( - oi.fulfillment_channel IS DISTINCT FROM (${CHANNEL_SQL}) - OR oi.fulfillment_status IS DISTINCT FROM (${DECISION_SQL}) - OR oi.fulfillment_source IS DISTINCT FROM (${SOURCE_SQL}) - ) AS changed - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE oi.id > $1 - ORDER BY oi.id - LIMIT $2 - `, - [cursor, batchSize], - ) - - if (rows.rows.length === 0) break - - scanned += rows.rows.length - cursor = rows.rows[rows.rows.length - 1]!.id - - for (const row of rows.rows) { - const key = `${row.status}/${row.channel ?? '-'}` - summary.set(key, (summary.get(key) || 0) + 1) - } - - const changedRows = rows.rows.filter((row) => row.changed) - if (changedRows.length === 0) { - process.stderr.write(`批 ${batch + 1}: 扫描 ${rows.rows.length}，无变化\n`) - continue - } - - if (dryRun) { - updated += changedRows.length - process.stderr.write( - `批 ${batch + 1}: 扫描 ${rows.rows.length}，将更新 ${changedRows.length}\n`, - ) - continue - } - - const ids = changedRows.map((row) => row.id) - const channels = changedRows.map((row) => row.channel) - const statuses = changedRows.map((row) => row.status) - const sources = changedRows.map((row) => row.source) - - const result = await query( - ` - UPDATE order_items oi - SET - fulfillment_channel = input.channel, - fulfillment_status = input.status, - fulfillment_source = input.source, - fulfillment_decided_at = NOW(), - updated_at = NOW() - FROM unnest($1::bigint[], $2::text[], $3::text[], $4::text[]) - AS input(id, channel, status, source) - WHERE oi.id = input.id - AND oi.fulfillment_locked = FALSE - `, - [ids, channels, statuses, sources], - ) - - updated += result.rowCount || 0 - process.stderr.write( - `批 ${batch + 1}: 扫描 ${rows.rows.length}，更新 ${result.rowCount || 0}\n`, - ) - } - - const remaining = await query<{ count: number }>( - `SELECT COUNT(*)::int AS count FROM order_items WHERE id > $1`, - [cursor], - ) - - console.log( - JSON.stringify( - { - cutoverDate: CUTOVER_DATE, - dryRun, - batchSize, - afterId, - scanned, - updated, - lastId: cursor, - remaining: remaining.rows[0]?.count || 0, - summary: Object.fromEntries([...summary.entries()].sort()), - durationMs: Date.now() - startedAt, - }, - null, - 2, - ), - ) - - if ((remaining.rows[0]?.count || 0) > 0) { - console.error( - `仍有 ${remaining.rows[0]?.count} 个订单项未扫描；请增大 --max-batches 或用 --after-id=${cursor} 续跑。`, - ) - process.exitCode = 2 - } -} - -main() - .catch((error) => { - console.error(error instanceof Error ? error.message : error) - process.exitCode = 1 - }) - .finally(async () => { - await closeDb() - }) - -async function assertSchema() { - const columns = await query<{ column_name: string }>( - ` - SELECT column_name - FROM information_schema.columns - WHERE table_schema = 'public' - AND table_name = 'order_items' - AND column_name IN ( - 'fulfillment_mode', 'fulfillment_channel', - 'fulfillment_status', 'fulfillment_source', 'fulfillment_locked' - ) - `, - ) - const found = new Set(columns.rows.map((row) => row.column_name)) - const missing = [ - 'fulfillment_mode', - 'fulfillment_channel', - 'fulfillment_status', - 'fulfillment_source', - 'fulfillment_locked', - ].filter((name) => !found.has(name)) - - if (missing.length > 0) { - throw new Error(`order_items 缺少列：${missing.join(', ')}；请先执行迁移 112`) - } -} - -function parseOptions(args: string[]) { - const result: Record = {} - for (const arg of args) { - if (!arg.startsWith('--')) continue - const body = arg.slice(2) - const index = body.indexOf('=') - const rawKey = index === -1 ? body : body.slice(0, index) - const value = index === -1 ? 'true' : body.slice(index + 1) - result[rawKey] = value - // 同时支持 --batch-size 与 --batchSize 两种写法。 - result[rawKey.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] = value - } - return result -} - -function normalizeInteger(value: unknown, fallback: number, min: number, max: number) { - const parsed = Number(value) - if (!Number.isInteger(parsed)) return fallback - return Math.min(max, Math.max(min, parsed)) -} diff --git a/apps/backend/scripts/backfill-fulfillment-mode.ts b/apps/backend/scripts/backfill-fulfillment-mode.ts deleted file mode 100644 index 47f1d891..00000000 --- a/apps/backend/scripts/backfill-fulfillment-mode.ts +++ /dev/null @@ -1,88 +0,0 @@ -import process from 'node:process' - -import { closeDb, query } from '../src/db/client.js' -import { - getFulfillmentModeCounts, - reconcilePendingFulfillmentModes, -} from '../src/services/fulfillment/fulfillment-mode-service.js' - -const options = parseOptions(process.argv.slice(2)) -const batchSize = normalizeInteger(options.batchSize, 5000, 1, 50_000) -const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 10_000) -const afterId = normalizeInteger(options.afterId, 0, 0, Number.MAX_SAFE_INTEGER) - -async function main() { - const column = await query<{ exists: boolean }>( - ` - SELECT EXISTS ( - SELECT 1 - FROM information_schema.columns - WHERE table_schema = 'public' - AND table_name = 'order_items' - AND column_name = 'fulfillment_mode' - ) AS exists - `, - ) - - if (!column.rows[0]?.exists) { - throw new Error('order_items.fulfillment_mode 不存在，请先执行 npm run db:migrate（迁移 105）') - } - - const startedAt = Date.now() - const result = await reconcilePendingFulfillmentModes({ - batchSize, - maxBatches, - reason: 'backfill_2026_10_23', - afterId, - }) - const counts = await getFulfillmentModeCounts() - - console.log( - JSON.stringify( - { - ...result, - batchSize, - maxBatches, - counts, - durationMs: Date.now() - startedAt, - }, - null, - 2, - ), - ) - - if (result.remainingCount > 0) { - console.error(`仍有 ${result.remainingCount} 个订单项未回填；请增大 --max-batches 或检查数据。`) - process.exitCode = 2 - } -} - -main() - .catch((error) => { - console.error(error instanceof Error ? error.message : error) - process.exitCode = 1 - }) - .finally(async () => { - await closeDb() - }) - -function parseOptions(args: string[]) { - const result: Record = {} - for (const arg of args) { - if (!arg.startsWith('--')) continue - const body = arg.slice(2) - const index = body.indexOf('=') - if (index === -1) { - result[body] = 'true' - } else { - result[body.slice(0, index)] = body.slice(index + 1) - } - } - return result -} - -function normalizeInteger(value: unknown, fallback: number, min: number, max: number) { - const parsed = Number(value) - if (!Number.isInteger(parsed)) return fallback - return Math.min(max, Math.max(min, parsed)) -} diff --git a/apps/backend/scripts/finance-cutover-v2.ts b/apps/backend/scripts/finance-cutover-v2.ts index acee8a5e..19a1da5c 100644 --- a/apps/backend/scripts/finance-cutover-v2.ts +++ b/apps/backend/scripts/finance-cutover-v2.ts @@ -16,9 +16,10 @@ import { closeDb, query } from '../src/db/client.js' * sku_channel_rules（配置） * * 前置： - * 1. 迁移 112 已执行； - * 2. `npm run fulfillment:backfill-channel` 已完成（不允许有 assigned 但无渠道的行）； - * 3. 已做数据库备份。 + * 1. 迁移 112/113 已执行； + * 2. 已做数据库备份。 + * + * 断代本身包含两步：先把断代前的订单项批量置为 archived，再清理派生数据。 * * 默认 dry-run，只打印将删除的行数；加 `--execute --confirm=2026-10-31` 才真正执行。 * 不可重复执行：成功后会写入 finance_cutovers，第二次会被拦下。 @@ -77,6 +78,8 @@ async function main() { await assertNotExecuted() + const archived = await archiveHistoricalItems() + const deleted: Record = {} for (const item of [...DERIVED_TABLES].sort((a, b) => a.order - b.order)) { const result = await query( @@ -110,7 +113,13 @@ async function main() { console.log( JSON.stringify( - { mode: 'execute', cutoverDate: CUTOVER_DATE, deleted, durationMs: Date.now() - startedAt }, + { + mode: 'execute', + cutoverDate: CUTOVER_DATE, + archivedItems: archived.updated, + deleted, + durationMs: Date.now() - startedAt, + }, null, 2, ), @@ -181,21 +190,48 @@ async function assertPreconditions() { `有 ${bad} 行 status=assigned 但 channel 为空；请先执行 fulfillment:backfill-channel`, ) } +} - const pendingBackfill = await query<{ count: number }>( - ` - SELECT COUNT(*)::int AS count - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date < $1::date - AND oi.fulfillment_status <> 'archived' - `, - [CUTOVER_DATE], - ) - const notArchived = pendingBackfill.rows[0]?.count || 0 - if (notArchived > 0) { - throw new Error(`有 ${notArchived} 行断代前订单项未归档；请先执行 fulfillment:backfill-channel`) +/** + * 断代前的订单项统一置为 archived（分批，避免 42 万行整表更新撞超时）。 + */ +async function archiveHistoricalItems(): Promise<{ updated: number }> { + let cursor = 0 + let updated = 0 + for (let batch = 0; batch < 500; batch += 1) { + const rows = await query<{ id: string }>( + ` + SELECT oi.id::text + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + WHERE (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date < $1::date + AND oi.fulfillment_status <> 'archived' + AND oi.id > $2 + ORDER BY oi.id + LIMIT 20000 + `, + [CUTOVER_DATE, cursor], + ) + if (rows.rows.length === 0) break + cursor = Number(rows.rows[rows.rows.length - 1]!.id) + + const result = await query( + ` + UPDATE order_items + SET fulfillment_status = 'archived', + fulfillment_channel = NULL, + fulfillment_source = 'cutover_archive', + fulfillment_decided_at = NOW(), + updated_at = NOW() + WHERE id = ANY($1::bigint[]) + AND fulfillment_status <> 'archived' + `, + [rows.rows.map((row) => Number(row.id))], + ) + updated += result.rowCount || 0 + process.stderr.write(`归档 ${result.rowCount || 0} 行（游标 ${cursor}）\n`) } + return { updated } } async function assertNotExecuted() { diff --git a/apps/backend/scripts/sync-cost-finance.ts b/apps/backend/scripts/sync-cost-finance.ts deleted file mode 100644 index 3f6d1ba8..00000000 --- a/apps/backend/scripts/sync-cost-finance.ts +++ /dev/null @@ -1,91 +0,0 @@ -import process from 'node:process' - -import { closeDb } from '../src/db/client.js' -import { - syncActualWorkOrderCostSnapshots, - syncEstimatedOrderItemCostSnapshots, -} from '../src/services/cost-center/cost-snapshot-sync-service.js' -import { rebuildSalesDailyFacts } from '../src/services/finance/sales-facts-service.js' -import { resolveFinanceDateRange, shiftDate } from '../src/utils/finance-date-range.js' -import { nowIso } from '../src/utils/time.js' - -type Mode = 'estimated' | 'actual' | 'facts' | 'all' - -const options = parseOptions(process.argv.slice(2)) -const mode = normalizeMode(options.mode) -const days = normalizeDays(options.days) -const fallbackTo = options.to || formatShanghaiDate(new Date()) -const fallbackFrom = options.from || shiftDate(fallbackTo, -(days - 1)) - -async function main() { - const input = { dateFrom: fallbackFrom, dateTo: fallbackTo } - const range = resolveFinanceDateRange(input, { defaultDays: days, maxDays: 400 }) - const startedAt = Date.now() - const result: Record = { mode, range } - - if (mode === 'estimated' || mode === 'all') { - result.estimated = await syncEstimatedOrderItemCostSnapshots(input) - } - if (mode === 'actual' || mode === 'all') { - result.actual = await syncActualWorkOrderCostSnapshots(input) - } - if (mode === 'facts' || mode === 'all') { - result.facts = await rebuildSalesDailyFacts(input) - } - - result.durationMs = Date.now() - startedAt - result.finishedAt = nowIso() - console.log(JSON.stringify(result, null, 2)) -} - -main() - .catch((error) => { - console.error(error instanceof Error ? error.message : error) - process.exitCode = 1 - }) - .finally(async () => { - await closeDb() - }) - -function parseOptions(args: string[]) { - const result: Record = {} - for (const arg of args) { - if (!arg.startsWith('--')) continue - const body = arg.slice(2) - const index = body.indexOf('=') - if (index === -1) { - result[body] = 'true' - } else { - result[body.slice(0, index)] = body.slice(index + 1) - } - } - return result -} - -function normalizeMode(value: unknown): Mode { - const mode = String(value || 'all').trim() - if (mode === 'estimated' || mode === 'actual' || mode === 'facts' || mode === 'all') { - return mode - } - throw new Error('--mode 只能是 estimated / actual / facts / all') -} - -function normalizeDays(value: unknown) { - const parsed = Number(value || 7) - if (!Number.isInteger(parsed) || parsed < 1 || parsed > 400) { - throw new Error('--days 必须是 1-400 的整数') - } - return parsed -} - -/** 模块级只构造一次：`new Intl.DateTimeFormat()` 每次约 0.5ms，逐条调用会线性放大。 */ -const SHANGHAI_DATE_FORMATTER = new Intl.DateTimeFormat('en-CA', { - timeZone: 'Asia/Shanghai', - year: 'numeric', - month: '2-digit', - day: '2-digit', -}) - -function formatShanghaiDate(value: Date) { - return SHANGHAI_DATE_FORMATTER.format(value) -} diff --git a/apps/backend/src/db/migrations/113_drop_legacy_fulfillment_mode.sql b/apps/backend/src/db/migrations/113_drop_legacy_fulfillment_mode.sql new file mode 100644 index 00000000..bc084b91 --- /dev/null +++ b/apps/backend/src/db/migrations/113_drop_legacy_fulfillment_mode.sql @@ -0,0 +1,40 @@ +-- 113_drop_legacy_fulfillment_mode.sql +-- +-- 强制新模型：彻底删除旧 `order_items.fulfillment_mode` 及其索引/约束。 +-- +-- 背景： +-- · 112 已加入 fulfillment_channel + fulfillment_status + fulfillment_source； +-- · 代码侧已全部改为读写新列，`fulfillment_mode` 不再有任何读写方； +-- · 保留旧列只会让「新旧并存」继续存在，因此这一版直接删除。 +-- +-- 注意： +-- 1. 本迁移只做 DDL，不搬运数据； +-- 2. 升级时应在维护窗口内先跑迁移，紧接着执行 `npm run finance:cutover-v2` +-- 完成断代归档与派生数据清理，避免中间态被业务读到； +-- 3. 已被 112 标记为 archived 的历史行不受影响。 + +-- 上一版 113 曾加过「旧异常写入口」兼容触发器，这里一并清理，不再保留兼容层。 +DROP TRIGGER IF EXISTS trg_cost_exceptions_fill_scope_key ON cost_exceptions; +DROP FUNCTION IF EXISTS cost_exceptions_fill_scope_key(); + +ALTER TABLE order_items + DROP CONSTRAINT IF EXISTS chk_order_items_fulfillment_mode; + +-- 依赖旧列的部分索引必须先删，否则 DROP COLUMN 会被拒绝。 +DROP INDEX IF EXISTS idx_order_items_fulfillment_mode; +DROP INDEX IF EXISTS idx_order_items_fulfillment_pending; +DROP INDEX IF EXISTS idx_order_items_fulfillment_id; +DROP INDEX IF EXISTS idx_order_items_fulfillment_pending_id; +DROP INDEX IF EXISTS idx_order_items_fulfillment_product; +DROP INDEX IF EXISTS idx_order_items_manual_pending_cover; +DROP INDEX IF EXISTS idx_order_items_manual_sku_key_stored; +DROP INDEX IF EXISTS idx_order_items_manual_internal_sku; +DROP INDEX IF EXISTS idx_order_items_fulfillment_config; + +ALTER TABLE order_items + DROP COLUMN IF EXISTS fulfillment_mode; + +COMMENT ON COLUMN order_items.fulfillment_channel IS + '履约渠道（唯一真源）：work_order / affiliate_dash / third_party_voucher / manual；pending 时为空'; +COMMENT ON COLUMN order_items.fulfillment_status IS + '渠道状态：pending 待分类 / assigned 已确定 / archived 断代归档'; diff --git a/apps/backend/src/routes/admin/cost-center-catalog.ts b/apps/backend/src/routes/admin/cost-center-catalog.ts index 90f94783..25872e4c 100644 --- a/apps/backend/src/routes/admin/cost-center-catalog.ts +++ b/apps/backend/src/routes/admin/cost-center-catalog.ts @@ -6,7 +6,6 @@ */ import { Router } from 'express' -import { normalizeManualProducts } from '../../services/cost-center/manual-product-normalization-service.js' import { backfillAdminNormalizedQuantities, backfillAdminProductMappings, @@ -27,7 +26,6 @@ import { backfillAdminAffiliateDashSkus, importAdminAffiliateDashPricesBaseline, setAdminAffiliateProductPrice, - syncAdminAffiliateDashCostSnapshots, syncAdminAffiliateDashProducts, } from '../../services/cost-center/affiliate-product-service.js' import { @@ -153,22 +151,6 @@ router.post( ), ) -router.post( - '/cost-center/affiliate-products/sync-snapshots', - requireAdminPermission('finance.config'), - createJsonHandler(() => withBatchTimeout(() => syncAdminAffiliateDashCostSnapshots()), { - successMessage: '已重算自动发货预估成本快照与销售事实', - errorMessage: '重算自动发货成本快照失败', - scope: '[admin/cost-center/affiliate-products/sync-snapshots]', - audit: (_req, data) => ({ - action: 'cost_affiliate_snapshots_synced', - targetType: 'cost_snapshots', - targetId: '', - data: data as Record, - }), - }), -) - router.post( '/cost-center/work-order-products/backfill', requireAdminPermission('finance.config'), @@ -204,32 +186,6 @@ router.post( }), ) -router.post( - '/cost-center/manual-products/normalize', - requireAdminPermission('finance.config'), - createJsonHandler( - (req) => - withBatchTimeout(() => - normalizeManualProducts({ - dateFrom: bodyOf(req).dateFrom as string | undefined, - dateTo: bodyOf(req).dateTo as string | undefined, - dryRun: bodyOf(req).dryRun === true, - }), - ), - { - successMessage: '人工商品归一完成', - errorMessage: '归一人工商品失败', - scope: '[admin/cost-center/manual-products/normalize]', - audit: (_req, data) => ({ - action: 'cost_manual_products_normalized', - targetType: 'product_catalog', - targetId: '', - data: data as Record, - }), - }, - ), -) - router.get( '/cost-center/products', requireAdminPermission('finance.view'), diff --git a/apps/backend/src/routes/admin/cost-center-cost.ts b/apps/backend/src/routes/admin/cost-center-cost.ts index 24e50604..6501387c 100644 --- a/apps/backend/src/routes/admin/cost-center-cost.ts +++ b/apps/backend/src/routes/admin/cost-center-cost.ts @@ -21,10 +21,6 @@ import { createAdminCostSnapshot, previewAdminCost, } from '../../services/cost-center/cost-resolution-service.js' -import { - syncActualWorkOrderCostSnapshots, - syncEstimatedOrderItemCostSnapshots, -} from '../../services/cost-center/cost-snapshot-sync-service.js' import type { JsonObject } from '../../types/json.js' import { bodyOf, idOf, sessionOf } from './request-helpers.js' import { createJsonHandler, requireAdminPermission } from './session.js' @@ -244,50 +240,4 @@ router.post( }), ) -router.post( - '/cost-center/snapshots/sync', - requireAdminPermission('finance.config'), - createJsonHandler( - async (req) => { - const body = bodyOf(req) - const mode = String(body.mode || 'both').trim() - const estimated = mode === 'actual' ? null : await syncEstimatedOrderItemCostSnapshots(body) - const actual = mode === 'estimated' ? null : await syncActualWorkOrderCostSnapshots(body) - return { mode, estimated, actual } - }, - { - successMessage: '成本快照回填完成', - errorMessage: '回填成本快照失败', - scope: '[admin/cost-center/snapshots/sync]', - audit: (req, data) => { - const result = data as { - mode?: string - estimated?: { createdCount?: number; exceptionCreatedCount?: number } | null - actual?: { - createdCount?: number - directRewardCount?: number - upgradedCount?: number - estimatedRewardCount?: number - } | null - } - return { - action: 'cost_snapshot_synced', - targetType: 'cost_snapshot', - targetId: '', - data: { - mode: result.mode || '', - estimatedCreated: result.estimated?.createdCount || 0, - actualCreated: - (result.actual?.directRewardCount || 0) + (result.actual?.upgradedCount || 0), - estimatedRewardCreated: result.actual?.estimatedRewardCount || 0, - exceptions: result.estimated?.exceptionCreatedCount || 0, - dateFrom: req.body?.dateFrom || '', - dateTo: req.body?.dateTo || '', - }, - } - }, - }, - ), -) - export default router diff --git a/apps/backend/src/routes/admin/finance.ts b/apps/backend/src/routes/admin/finance.ts index c6ef3abf..31e9bf58 100644 --- a/apps/backend/src/routes/admin/finance.ts +++ b/apps/backend/src/routes/admin/finance.ts @@ -7,8 +7,8 @@ import { import { getSalesProfitOverview, listSalesDailyFacts, - rebuildSalesDailyFacts, } from '../../services/finance/sales-facts-service.js' +import { rebuildChannelSalesDailyFacts } from '../../services/finance/sales-facts-channel-service.js' import { confirmManualFulfillment, listManualFulfillmentPending, @@ -70,10 +70,15 @@ router.post( requireAdminRoles(['admin', 'operator', 'finance']), requireAdminPermission('finance.post'), createJsonHandler( - (req) => - rebuildSalesDailyFacts( - req.body && typeof req.body === 'object' ? (req.body as Record) : {}, - ), + (req) => { + const body = + req.body && typeof req.body === 'object' ? (req.body as Record) : {} + const today = new Date().toLocaleDateString('en-CA', { timeZone: 'Asia/Shanghai' }) + return rebuildChannelSalesDailyFacts({ + dateFrom: String(body.dateFrom || body.date_from || today).slice(0, 10), + dateTo: String(body.dateTo || body.date_to || today).slice(0, 10), + }) + }, { successMessage: '销售事实已重算', errorMessage: '重算销售事实失败', diff --git a/apps/backend/src/services/cost-center/affiliate-product-service.ts b/apps/backend/src/services/cost-center/affiliate-product-service.ts index 0b3be2ff..3ea198d9 100644 --- a/apps/backend/src/services/cost-center/affiliate-product-service.ts +++ b/apps/backend/src/services/cost-center/affiliate-product-service.ts @@ -4,8 +4,7 @@ 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 { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js' -import { syncEstimatedOrderItemCostSnapshots } from './cost-snapshot-sync-service.js' +import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js' import { AFFILIATE_DASH_PRICES_BASELINE } from './affiliate-dash-prices-baseline.js' /** @@ -149,9 +148,9 @@ export async function rebuildFactsForRange(fromDate: string, toDate: string) { while (cursor <= toDate) { const windowEnd = shiftDate(cursor, 399) const chunkEnd = windowEnd > toDate ? toDate : windowEnd - const rebuilt = await rebuildSalesDailyFacts({ dateFrom: cursor, dateTo: chunkEnd }) - deletedCount += rebuilt.deletedCount - insertedCount += rebuilt.insertedCount + const rebuilt = await rebuildChannelSalesDailyFacts({ dateFrom: cursor, dateTo: chunkEnd }) + deletedCount += rebuilt.deleted + insertedCount += rebuilt.inserted batchCount += 1 cursor = shiftDate(chunkEnd, 1) } @@ -548,24 +547,3 @@ export async function importAdminAffiliateDashPricesBaseline(session: AdminSessi skippedCount, } } - -/** - * 一键重算自动发货预估成本快照并重算销售事实。 - * 默认统计最新的一个月（30 天）数据，避免扫描混乱的历史遗留数据。 - */ -export async function syncAdminAffiliateDashCostSnapshots(options: { days?: number } = {}) { - const days = options.days && options.days > 0 ? options.days : 30 - const today = formatShanghaiDate(new Date()) - const fromDate = shiftDate(today, -(days - 1)) - - const snapshotResult = await syncEstimatedOrderItemCostSnapshots({ - dateFrom: fromDate, - dateTo: today, - }) - const factsResult = await rebuildFactsForRange(fromDate, today) - - return { - snapshotResult, - factsResult, - } -} diff --git a/apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts b/apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts deleted file mode 100644 index cdfda4c5..00000000 --- a/apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts +++ /dev/null @@ -1,806 +0,0 @@ -import type { PoolClient } from 'pg' - -import { withTransaction } from '../../db/client.js' -import { resolveFinanceDateRange } from '../../utils/finance-date-range.js' -import { costModelScopeSql } from './cost-center-shared.js' - -export type CostSnapshotSyncInput = { - dateFrom?: unknown - dateTo?: unknown - from?: unknown - to?: unknown -} - -export type CostSnapshotSyncResult = { - range: { fromDate: string; toDate: string } - createdCount: number - exceptionCreatedCount: number - resolvedExceptionCount: number -} - -export type ActualCostSnapshotSyncResult = { - range: { fromDate: string; toDate: string } - createdCount: number - upgradedCount: number - directRewardCount: number - estimatedRewardCount: number -} - -/** - * 按订单项批量生成预估成本快照。 - * - * 只处理已有 canonical_product_id 的订单项；未归一订单项进入 - * product_mapping_missing 异常，不在这里猜测商品。 - */ -export async function syncEstimatedOrderItemCostSnapshots( - input: CostSnapshotSyncInput = {}, -): Promise { - const range = resolveFinanceDateRange(input, { defaultDays: 7, maxDays: 400 }) - - return withTransaction(async (client) => { - const inserted = await client.query( - ` - WITH item_scope AS ( - SELECT - oi.id AS order_item_id, - oi.order_id, - o.provider, - o.shop_id, - oi.sku_code, - oi.sku_name, - oi.canonical_product_id, - COALESCE(o.paid_at, o.created_at) AS basis_at, - GREATEST(COALESCE(oi.normalized_quantity, oi.quantity, 1), 1)::numeric AS quantity, - COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, - wo.id AS work_order_id, - wo.product_rule_id AS work_product_rule_id, - COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - -- order_item_id 上有唯一部分索引，一个订单项最多一张工单； - -- 原来的 ORDER BY id DESC LIMIT 1 逐行探测会做 3.7 万次索引查找。 - LEFT JOIN work_orders wo ON wo.order_item_id = oi.id - WHERE o.pay_status IN ('paid', 'refunded') - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - ), - resolved AS ( - SELECT - s.*, - ci.id AS cost_item_id, - ci.pricing_mode, - v.id AS cost_version_id, - v.amount_fen, - v.rate_bps - FROM item_scope s - JOIN cost_items ci ON ci.status = 'active' - AND ${costModelScopeSql('s.fulfillment_mode')} - AND ci.pricing_mode IN ('per_item', 'per_order', 'percent_of_revenue') - AND ci.code <> 'sys_work_order_reward' - -- 人工成本只能在「确认履约」时按实际支出写 actual 快照， - -- 这里的版本金额是 0 元占位，不能拿来生成预估快照。 - AND ci.cost_model <> 'manual_actual' - JOIN LATERAL ( - SELECT v.id, v.amount_fen, v.rate_bps - FROM cost_versions v - WHERE v.cost_item_id = ci.id - AND v.status = 'active' - AND v.effective_from <= s.basis_at - AND (v.effective_to IS NULL OR v.effective_to > s.basis_at) - AND ( - (v.scope_type = 'rule' AND s.work_product_rule_id IS NOT NULL - AND v.work_product_rule_id = s.work_product_rule_id) - OR (v.scope_type = 'product' AND s.canonical_product_id IS NOT NULL - AND v.canonical_product_id = s.canonical_product_id) - OR (v.scope_type = 'shop' AND v.provider = s.provider AND v.shop_id = s.shop_id) - OR (v.scope_type = 'provider' AND v.provider = s.provider) - OR (v.scope_type = 'global') - ) - ORDER BY - CASE v.scope_type - WHEN 'rule' THEN 1 - WHEN 'product' THEN 2 - WHEN 'shop' THEN 3 - WHEN 'provider' THEN 4 - ELSE 5 - END, - v.effective_from DESC, - v.id DESC - LIMIT 1 - ) v ON TRUE - WHERE s.canonical_product_id IS NOT NULL - ), - calculated AS ( - SELECT - r.*, - CASE r.pricing_mode - WHEN 'per_item' THEN ROUND(r.amount_fen::numeric * r.quantity)::bigint - WHEN 'per_order' THEN r.amount_fen - WHEN 'percent_of_revenue' - THEN ROUND((r.revenue_fen::numeric * r.rate_bps::numeric) / 10000.0)::bigint - ELSE r.amount_fen - END AS total_amount_fen - FROM resolved r - ) - INSERT INTO cost_snapshots ( - subject_type, subject_id, order_id, order_item_id, work_order_id, - cost_item_id, cost_version_id, canonical_product_id, - quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source, - metadata_json, cost_basis_at, resolved_at, created_at - ) - SELECT - 'order_item', c.order_item_id, c.order_id, c.order_item_id, c.work_order_id, - c.cost_item_id, c.cost_version_id, c.canonical_product_id, - CASE WHEN c.pricing_mode = 'per_item' THEN c.quantity ELSE 1 END, - CASE WHEN c.pricing_mode = 'percent_of_revenue' THEN c.total_amount_fen ELSE c.amount_fen END, - c.total_amount_fen, - 'estimated', 'resolved', - jsonb_build_object( - 'provider', c.provider, - 'shopId', c.shop_id, - 'skuCode', c.sku_code, - 'skuName', c.sku_name, - 'pricingMode', c.pricing_mode, - 'revenueFen', c.revenue_fen, - 'normalizedQuantity', c.quantity, - 'fulfillmentMode', c.fulfillment_mode, - 'basisAt', c.basis_at - ), - c.basis_at, NOW(), NOW() - FROM calculated c - ON CONFLICT (subject_type, subject_id, cost_item_id) - WHERE superseded_at IS NULL - DO NOTHING - `, - [range.fromDate, range.toDate], - ) - - const productExceptions = await insertAndResolveProductMappingExceptions( - client, - range.fromDate, - range.toDate, - ) - const costExceptions = await insertAndResolveCostMissingExceptions( - client, - range.fromDate, - range.toDate, - ) - - return { - range, - createdCount: inserted.rowCount || 0, - exceptionCreatedCount: productExceptions.created + costExceptions.created, - resolvedExceptionCount: productExceptions.resolved + costExceptions.resolved, - } - }) -} - -/** - * 工单验收后，把订单项当前预估快照升级为实际快照。 - * - * 如果配置了 work_order_reward 成本项，则用工单实际 reward_amount 覆盖金额； - * 否则仍冻结当前版本金额，但把阶段标为 actual。旧快照保留 superseded_at。 - */ -export async function syncActualWorkOrderCostSnapshots( - input: CostSnapshotSyncInput = {}, -): Promise { - const range = resolveFinanceDateRange(input, { defaultDays: 30, maxDays: 400 }) - - return withTransaction(async (client) => { - const directReward = await syncDirectWorkOrderRewardSnapshots( - client, - range.fromDate, - range.toDate, - ) - const estimatedReward = await syncCommittedWorkOrderRewardSnapshots( - client, - range.fromDate, - range.toDate, - ) - const upgraded = await client.query( - ` - WITH accepted AS ( - SELECT DISTINCT ON (wo.order_item_id) - wo.id AS work_order_id, - wo.order_item_id, - GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount, - COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at, NOW()) AS accepted_at - FROM work_orders wo - JOIN order_items oi ON oi.id = wo.order_item_id - JOIN orders o ON o.id = oi.order_id - WHERE wo.status = 'accepted' - AND wo.order_item_id IS NOT NULL - AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at) - AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date - ORDER BY wo.order_item_id, wo.id DESC - ), - active AS ( - SELECT - cs.*, - a.work_order_id AS actual_work_order_id, - a.reward_amount, - a.accepted_at, - ci.cost_model - FROM cost_snapshots cs - JOIN accepted a ON cs.subject_type = 'order_item' AND cs.subject_id = a.order_item_id - JOIN cost_items ci ON ci.id = cs.cost_item_id - WHERE cs.superseded_at IS NULL - AND cs.snapshot_stage IN ('estimated', 'actual') - AND ci.code <> 'sys_work_order_reward' - ), - target AS ( - SELECT * - FROM active -WHERE snapshot_stage = 'estimated' - OR total_amount_fen <> CASE - WHEN cost_model = 'work_order_reward' AND reward_amount > 0 THEN reward_amount - ELSE total_amount_fen - END - ), - superseded AS ( - UPDATE cost_snapshots cs - SET superseded_at = NOW() - FROM target t - WHERE cs.id = t.id - AND cs.superseded_at IS NULL - RETURNING - cs.id, cs.subject_id, cs.order_id, cs.order_item_id, - cs.cost_item_id, cs.cost_version_id, cs.canonical_product_id, - cs.quantity, cs.unit_amount_fen, cs.total_amount_fen, cs.metadata_json - ) - INSERT INTO cost_snapshots ( - subject_type, subject_id, order_id, order_item_id, work_order_id, - cost_item_id, cost_version_id, canonical_product_id, - quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source, - metadata_json, cost_basis_at, resolved_at, created_at - ) - SELECT - 'order_item', s.subject_id, s.order_id, s.order_item_id, a.work_order_id, - s.cost_item_id, s.cost_version_id, s.canonical_product_id, - CASE WHEN ci.cost_model = 'work_order_reward' AND a.reward_amount > 0 THEN 1 - ELSE s.quantity END, - CASE WHEN ci.cost_model = 'work_order_reward' AND a.reward_amount > 0 - THEN a.reward_amount ELSE s.unit_amount_fen END, - CASE WHEN ci.cost_model = 'work_order_reward' AND a.reward_amount > 0 - THEN a.reward_amount ELSE s.total_amount_fen END, - 'actual', 'resolved', - s.metadata_json || jsonb_build_object( - 'actualWorkOrderId', a.work_order_id, - 'actualRewardFen', a.reward_amount, - 'basisAt', a.accepted_at - ), - a.accepted_at, NOW(), NOW() - FROM superseded s - JOIN accepted a ON a.order_item_id = s.subject_id - JOIN cost_items ci ON ci.id = s.cost_item_id - ON CONFLICT (subject_type, subject_id, cost_item_id) - WHERE superseded_at IS NULL - DO NOTHING - `, - [range.fromDate, range.toDate], - ) - - return { - range, - createdCount: directReward.inserted + estimatedReward.inserted + (upgraded.rowCount || 0), - upgradedCount: upgraded.rowCount || 0, - directRewardCount: directReward.inserted, - estimatedRewardCount: estimatedReward.inserted, - } - }) -} - -/** - * 接单平台的实际成本不依赖预估快照：工单验收后直接按 reward_amount 写 actual。 - * 系统成本项 sys_work_order_reward 的版本金额只是占位，不会进入预估阶段。 - */ -async function syncDirectWorkOrderRewardSnapshots( - client: PoolClient, - fromDate: string, - toDate: string, -) { - const superseded = await client.query( - ` - WITH system_item AS ( - SELECT ci.id AS cost_item_id, v.id AS cost_version_id - FROM cost_items ci - JOIN LATERAL ( - SELECT id - FROM cost_versions - WHERE cost_item_id = ci.id - AND status = 'active' - AND effective_from <= NOW() - ORDER BY effective_from DESC, id DESC - LIMIT 1 - ) v ON TRUE - WHERE ci.code = 'sys_work_order_reward' - AND ci.status = 'active' - ), - accepted AS ( - SELECT DISTINCT ON (wo.order_item_id) - wo.id AS work_order_id, - wo.order_item_id, - GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount, - COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at, NOW()) AS accepted_at - FROM work_orders wo - JOIN order_items oi ON oi.id = wo.order_item_id - JOIN orders o ON o.id = oi.order_id - WHERE wo.status = 'accepted' - AND wo.order_item_id IS NOT NULL - AND COALESCE(wo.reward_amount, 0) >= 0 - AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at) - AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date - ORDER BY wo.order_item_id, wo.id DESC - ) - UPDATE cost_snapshots cs - SET superseded_at = NOW() - FROM accepted a - JOIN system_item si ON TRUE - JOIN LATERAL ( - SELECT id, snapshot_stage, total_amount_fen - FROM cost_snapshots - WHERE subject_type = 'order_item' - AND subject_id = a.order_item_id - AND cost_item_id = si.cost_item_id - AND superseded_at IS NULL - ORDER BY id DESC - LIMIT 1 - FOR UPDATE - ) current ON TRUE - WHERE cs.id = current.id - AND cs.superseded_at IS NULL - AND (current.snapshot_stage <> 'actual' OR current.total_amount_fen <> a.reward_amount) - `, - [fromDate, toDate], - ) - - const inserted = await client.query( - ` - WITH system_item AS ( - SELECT ci.id AS cost_item_id, v.id AS cost_version_id - FROM cost_items ci - JOIN LATERAL ( - SELECT id - FROM cost_versions - WHERE cost_item_id = ci.id - AND status = 'active' - AND effective_from <= NOW() - ORDER BY effective_from DESC, id DESC - LIMIT 1 - ) v ON TRUE - WHERE ci.code = 'sys_work_order_reward' - AND ci.status = 'active' - ), - accepted AS ( - SELECT DISTINCT ON (wo.order_item_id) - wo.id AS work_order_id, - wo.order_item_id, - wo.product_rule_id, - GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount, - COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at, NOW()) AS accepted_at - FROM work_orders wo - JOIN order_items oi ON oi.id = wo.order_item_id - JOIN orders o ON o.id = oi.order_id - WHERE wo.status = 'accepted' - AND wo.order_item_id IS NOT NULL - AND COALESCE(wo.reward_amount, 0) >= 0 - AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at) - AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date - ORDER BY wo.order_item_id, wo.id DESC - ) - INSERT INTO cost_snapshots ( - subject_type, subject_id, order_id, order_item_id, work_order_id, - cost_item_id, cost_version_id, canonical_product_id, - quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source, - metadata_json, cost_basis_at, resolved_at, created_at - ) - SELECT - 'order_item', a.order_item_id, oi.order_id, a.order_item_id, a.work_order_id, - si.cost_item_id, COALESCE(rv.id, si.cost_version_id), - oi.canonical_product_id, - 1, a.reward_amount, a.reward_amount, 'actual', 'resolved', - jsonb_build_object( - 'source', 'system_work_order_reward', - 'actualWorkOrderId', a.work_order_id, - 'actualRewardFen', a.reward_amount, - 'basisAt', a.accepted_at - ), - a.accepted_at, NOW(), NOW() - FROM accepted a - CROSS JOIN system_item si - JOIN order_items oi ON oi.id = a.order_item_id - LEFT JOIN LATERAL ( - SELECT v.id - FROM cost_versions v - WHERE v.cost_item_id = si.cost_item_id - AND v.scope_type = 'rule' - AND v.work_product_rule_id = a.product_rule_id - AND v.status IN ('active', 'retired') - AND v.effective_from <= a.accepted_at - AND (v.effective_to IS NULL OR v.effective_to > a.accepted_at) - ORDER BY v.effective_from DESC, v.id DESC - LIMIT 1 - ) rv ON TRUE - WHERE NOT EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.subject_type = 'order_item' - AND cs.subject_id = a.order_item_id - AND cs.cost_item_id = si.cost_item_id - AND cs.superseded_at IS NULL - AND cs.snapshot_stage = 'actual' - AND cs.total_amount_fen = a.reward_amount - ) - ON CONFLICT (subject_type, subject_id, cost_item_id) - WHERE superseded_at IS NULL - DO NOTHING - `, - [fromDate, toDate], - ) - - return { - inserted: inserted.rowCount || 0, - superseded: superseded.rowCount || 0, - } -} - -/** - * 未验收工单的报酬在创建时已确定，先按 reward_amount 写 estimated 快照。 - * 验收后由 syncDirectWorkOrderRewardSnapshots 升级为 actual；取消工单不计成本。 - */ -async function syncCommittedWorkOrderRewardSnapshots( - client: PoolClient, - fromDate: string, - toDate: string, -) { - const superseded = await client.query( - ` - WITH system_item AS ( - SELECT ci.id AS cost_item_id - FROM cost_items ci - WHERE ci.code = 'sys_work_order_reward' - AND ci.status = 'active' - ), - latest AS ( - SELECT DISTINCT ON (wo.order_item_id) - wo.id AS work_order_id, - wo.order_item_id, - GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount - FROM work_orders wo - JOIN orders o ON o.id = wo.order_id - WHERE wo.order_item_id IS NOT NULL - AND wo.status IN ( - 'pending_material', 'in_progress', 'unassigned', 'open', - 'pending_acceptance', 'problem' - ) - AND COALESCE(wo.reward_amount, 0) > 0 - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - ORDER BY wo.order_item_id, wo.id DESC - ) - UPDATE cost_snapshots cs - SET superseded_at = NOW() - FROM latest l - JOIN system_item si ON TRUE - WHERE cs.subject_type = 'order_item' - AND cs.subject_id = l.order_item_id - AND cs.cost_item_id = si.cost_item_id - AND cs.superseded_at IS NULL - AND cs.snapshot_stage = 'estimated' - AND ( - cs.work_order_id IS DISTINCT FROM l.work_order_id - OR cs.total_amount_fen <> l.reward_amount - ) - `, - [fromDate, toDate], - ) - - const inserted = await client.query( - ` - WITH system_item AS ( - SELECT ci.id AS cost_item_id, v.id AS cost_version_id - FROM cost_items ci - JOIN LATERAL ( - SELECT id - FROM cost_versions - WHERE cost_item_id = ci.id - AND status = 'active' - AND effective_from <= NOW() - ORDER BY effective_from DESC, id DESC - LIMIT 1 - ) v ON TRUE - WHERE ci.code = 'sys_work_order_reward' - AND ci.status = 'active' - ), - latest AS ( - SELECT DISTINCT ON (wo.order_item_id) - wo.id AS work_order_id, - wo.order_item_id, - wo.product_rule_id, - wo.status, - GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount, - COALESCE(wo.created_at, o.paid_at, o.created_at, NOW()) AS basis_at - FROM work_orders wo - JOIN orders o ON o.id = wo.order_id - WHERE wo.order_item_id IS NOT NULL - AND wo.status IN ( - 'pending_material', 'in_progress', 'unassigned', 'open', - 'pending_acceptance', 'problem' - ) - AND COALESCE(wo.reward_amount, 0) > 0 - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - ORDER BY wo.order_item_id, wo.id DESC - ) - INSERT INTO cost_snapshots ( - subject_type, subject_id, order_id, order_item_id, work_order_id, - cost_item_id, cost_version_id, canonical_product_id, - quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source, - metadata_json, cost_basis_at, resolved_at, created_at - ) - SELECT - 'order_item', l.order_item_id, oi.order_id, l.order_item_id, l.work_order_id, - si.cost_item_id, COALESCE(rv.id, si.cost_version_id), - oi.canonical_product_id, - 1, l.reward_amount, l.reward_amount, 'estimated', 'resolved', - jsonb_build_object( - 'source', 'work_order_reward_committed', - 'workOrderStatus', l.status, - 'estimatedRewardFen', l.reward_amount, - 'basisAt', l.basis_at - ), - l.basis_at, NOW(), NOW() - FROM latest l - CROSS JOIN system_item si - JOIN order_items oi ON oi.id = l.order_item_id - LEFT JOIN LATERAL ( - SELECT v.id - FROM cost_versions v - WHERE v.cost_item_id = si.cost_item_id - AND v.scope_type = 'rule' - AND v.work_product_rule_id = l.product_rule_id - AND v.status IN ('active', 'retired') - AND v.effective_from <= l.basis_at - AND (v.effective_to IS NULL OR v.effective_to > l.basis_at) - ORDER BY v.effective_from DESC, v.id DESC - LIMIT 1 - ) rv ON TRUE - WHERE NOT EXISTS ( - SELECT 1 FROM work_orders accepted - WHERE accepted.order_item_id = l.order_item_id - AND accepted.status = 'accepted' - ) - AND NOT EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.subject_type = 'order_item' - AND cs.subject_id = l.order_item_id - AND cs.cost_item_id = si.cost_item_id - AND cs.superseded_at IS NULL - ) - ON CONFLICT (subject_type, subject_id, cost_item_id) - WHERE superseded_at IS NULL - DO NOTHING - `, - [fromDate, toDate], - ) - - return { - inserted: inserted.rowCount || 0, - superseded: superseded.rowCount || 0, - } -} - -async function insertAndResolveProductMappingExceptions( - client: PoolClient, - fromDate: string, - toDate: string, -) { - const inserted = await client.query( - ` - WITH item_scope AS ( - SELECT - oi.id AS order_item_id, - o.provider, - o.shop_id, - oi.sku_code, - oi.sku_name, - oi.canonical_product_id, - COALESCE(o.paid_at, o.created_at) AS basis_at, - wo.id AS work_order_id, - COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - LEFT JOIN work_orders wo ON wo.order_item_id = oi.id - WHERE o.pay_status IN ('paid', 'refunded') - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - ) - INSERT INTO cost_exceptions ( - subject_type, subject_id, order_item_id, work_order_id, - reason, status, detail_json, created_at - ) - SELECT - 'order_item', s.order_item_id, s.order_item_id, s.work_order_id, - 'product_mapping_missing', 'open', - jsonb_build_object( - 'provider', s.provider, - 'shopId', s.shop_id, - 'skuCode', s.sku_code, - 'skuName', s.sku_name, - 'fulfillmentMode', s.fulfillment_mode, - 'basisAt', s.basis_at, - 'message', '订单项没有归一到标准商品，销售事实会进入未归一桶' - ), - NOW() - FROM item_scope s - WHERE s.canonical_product_id IS NULL - ON CONFLICT (subject_type, subject_id, reason) DO NOTHING - `, - [fromDate, toDate], - ) - - const resolved = await client.query( - ` - UPDATE cost_exceptions ce - SET status = 'resolved', resolved_at = NOW() - WHERE ce.status = 'open' - AND ce.subject_type = 'order_item' - AND ce.reason = 'product_mapping_missing' - -- 只收口最近创建、仍在同步视野内的异常：全表收口会随着历史异常累积 - -- 变成每次同步都扫几十万行（41 万单测试库实测单次 7.4 秒）。 - AND ce.created_at >= ($1::date - INTERVAL '7 days') - AND EXISTS ( - SELECT 1 FROM order_items oi - WHERE oi.id = ce.subject_id AND oi.canonical_product_id IS NOT NULL - ) - `, - [fromDate], - ) - - return { created: inserted.rowCount || 0, resolved: resolved.rowCount || 0 } -} - -async function insertAndResolveCostMissingExceptions( - client: PoolClient, - fromDate: string, - toDate: string, -) { - const inserted = await client.query( - ` - WITH item_scope AS ( - SELECT - oi.id AS order_item_id, - oi.canonical_product_id, - oi.fulfillment_config_id, - COALESCE(o.paid_at, o.created_at) AS basis_at, - wo.id AS work_order_id, - COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - LEFT JOIN work_orders wo ON wo.order_item_id = oi.id - WHERE o.pay_status IN ('paid', 'refunded') - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - ) - INSERT INTO cost_exceptions ( - subject_type, subject_id, order_item_id, work_order_id, - reason, status, detail_json, created_at - ) - SELECT - 'order_item', s.order_item_id, s.order_item_id, s.work_order_id, - 'cost_missing', 'open', - jsonb_build_object( - 'canonicalProductId', s.canonical_product_id, - 'fulfillmentMode', s.fulfillment_mode, - 'basisAt', s.basis_at, - 'message', '订单项没有匹配到任何成本版本' - ), - NOW() - FROM item_scope s - WHERE s.canonical_product_id IS NOT NULL - -- 人工履约不靠成本版本结算：金额由「人工 SKU 配置 × 换算率」决定， - -- 并在客服确认履约时写 manual_case 实际快照，因此不参与成本缺失判定。 - AND s.fulfillment_mode <> 'manual' - AND NOT EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.subject_type = 'order_item' - AND cs.subject_id = s.order_item_id - AND cs.superseded_at IS NULL - ) - ON CONFLICT (subject_type, subject_id, reason) DO NOTHING - `, - [fromDate, toDate], - ) - - const manualInserted = await client.query( - ` - WITH cutoff AS ( - SELECT MIN(created_at) AS enforce_from - FROM manual_fulfillment_configs - ), - item_scope AS ( - SELECT - oi.id AS order_item_id, - oi.fulfillment_config_id, - COALESCE(o.paid_at, o.created_at) AS basis_at - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - WHERE o.pay_status IN ('paid', 'refunded') - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - AND oi.fulfillment_mode = 'manual' - ) - INSERT INTO cost_exceptions ( - subject_type, subject_id, order_item_id, work_order_id, - reason, status, detail_json, created_at - ) - SELECT - 'order_item', s.order_item_id, s.order_item_id, NULL, - 'manual_cost_missing', 'open', - jsonb_build_object( - 'fulfillmentMode', 'manual', - 'basisAt', s.basis_at, - 'configMissing', s.fulfillment_config_id IS NULL, - 'message', CASE - WHEN s.fulfillment_config_id IS NULL - THEN '人工订单项未配置履约渠道，请先在成本中心配置人工 SKU' - ELSE '人工订单项尚未确认履约并记录成本' - END - ), - NOW() - FROM item_scope s - CROSS JOIN cutoff c - WHERE c.enforce_from IS NOT NULL - -- 只监控「第一次配置人工履约之后」的订单项：历史人工单按既定口径不补录。 - AND s.basis_at >= c.enforce_from - AND NOT EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.subject_type = 'manual_case' - AND cs.order_item_id = s.order_item_id - AND cs.superseded_at IS NULL - ) - ON CONFLICT (subject_type, subject_id, reason) DO NOTHING - `, - [fromDate, toDate], - ) - - const resolved = await client.query( - ` - UPDATE cost_exceptions ce - SET status = 'resolved', resolved_at = NOW() - WHERE ce.status = 'open' - AND ce.subject_type = 'order_item' - AND ce.reason = 'cost_missing' - AND ce.created_at >= ($1::date - INTERVAL '7 days') - AND EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.subject_type = 'order_item' - AND cs.subject_id = ce.subject_id - AND cs.superseded_at IS NULL - ) - `, - [fromDate], - ) - - const manualResolved = await client.query( - ` - UPDATE cost_exceptions ce - SET status = 'resolved', resolved_at = NOW() - WHERE ce.status = 'open' - AND ce.subject_type = 'order_item' - AND ce.reason = 'manual_cost_missing' - AND ce.created_at >= ($1::date - INTERVAL '7 days') - AND EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.subject_type = 'manual_case' - AND cs.order_item_id = ce.subject_id - AND cs.superseded_at IS NULL - ) - `, - [fromDate], - ) - - return { - created: (inserted.rowCount || 0) + (manualInserted.rowCount || 0), - resolved: (resolved.rowCount || 0) + (manualResolved.rowCount || 0), - } -} diff --git a/apps/backend/src/services/cost-center/manual-product-normalization-service.ts b/apps/backend/src/services/cost-center/manual-product-normalization-service.ts deleted file mode 100644 index 7212ef01..00000000 --- a/apps/backend/src/services/cost-center/manual-product-normalization-service.ts +++ /dev/null @@ -1,274 +0,0 @@ -/** - * P1 人工商品归一：把人工履约订单项归到 `man:<内部SKU>` 标准商品。 - * - * 为什么不用渠道映射：人工单的「内部 SKU」是快手 spec_json.skuNick / 拼多多 - * spec_json.outSkuSn，与平台 sku_code 不是同一个值（91kaquan 实测 0 条相同）， - * 所以按 provider + shop + sku_code 建映射抓不到这些单。这里直接按内部 SKU 归一。 - * - * 幂等：商品按 canonical_sku 去重；订单项只更新 canonical_product_id IS NULL 的行， - * 并按 id 分批（每批 2 万）避免撞 DATABASE_STATEMENT_TIMEOUT_MS。 - */ -import { query } from '../../db/client.js' -import { nowIso } from '../../utils/time.js' - -const MANUAL_PRODUCT_CATEGORY = '人工客服' -const MANUAL_PRODUCT_UNIT = '件' -const NORMALIZE_BATCH_SIZE = 20000 - -export type NormalizeManualProductsInput = { - dateFrom?: string | undefined - dateTo?: string | undefined - dryRun?: boolean - resolveExceptions?: boolean -} - -export type NormalizeManualProductsResult = { - dryRun: boolean - range: { fromDate: string; toDate: string } | null - distinctSkuCount: number - pendingItemCount: number - createdProductCount: number - updatedItemCount: number - updatedSnapshotCount: number - resolvedExceptionCount: number - remainingItemCount: number -} - -/** - * 内部 SKU：由 order_items.manual_sku_key 生成列统一提供 - * （skuNick → outSkuSn → sku_code → sku_name，均去空格）。 - */ -const MANUAL_SKU_KEY_SQL = `oi.manual_sku_key` - -export async function normalizeManualProducts( - input: NormalizeManualProductsInput = {}, -): Promise { - const fromDate = normalizeDate(input.dateFrom) - const toDate = normalizeDate(input.dateTo) - const hasRange = Boolean(fromDate || toDate) - const dryRun = input.dryRun === true - const now = nowIso() - const joinSql = hasRange ? 'JOIN orders o ON o.id = oi.order_id' : '' - - // 日期条件按传入的起始占位符编号生成，便于不同语句复用。 - function buildRangeClause(startIndex: number) { - const params: unknown[] = [] - const parts: string[] = [] - if (fromDate) { - parts.push( - `(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date >= $${startIndex + params.length}::date`, - ) - params.push(fromDate) - } - if (toDate) { - parts.push( - `(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date <= $${startIndex + params.length}::date`, - ) - params.push(toDate) - } - return { sql: parts.length ? `AND ${parts.join(' AND ')}` : '', params } - } - - const candidateRange = buildRangeClause(1) - const pendingWhere = ` - oi.fulfillment_mode = 'manual' - AND oi.canonical_product_id IS NULL - AND ${MANUAL_SKU_KEY_SQL} IS NOT NULL - ${candidateRange.sql} - ` - - const candidatesResult = await query<{ - sku_key: string - canonical_name: string - item_count: number - }>( - ` - WITH item_keys AS ( - SELECT - ${MANUAL_SKU_KEY_SQL} AS sku_key, - NULLIF(TRIM(oi.sku_name), '') AS sku_name - FROM order_items oi - ${joinSql} - WHERE ${pendingWhere} - ) - SELECT - k.sku_key, - LEFT( - COALESCE(NULLIF(TRIM(r.display_name), ''), MIN(k.sku_name), k.sku_key), - 120 - ) AS canonical_name, - COUNT(*)::int AS item_count - FROM item_keys k - LEFT JOIN manual_sku_registry r ON r.sku_key = k.sku_key - GROUP BY k.sku_key, r.display_name - ORDER BY COUNT(*) DESC, k.sku_key ASC - `, - candidateRange.params, - ) - - const candidates = candidatesResult.rows.map((row) => ({ - skuKey: row.sku_key, - canonicalSku: `man:${row.sku_key}`, - canonicalName: row.canonical_name || row.sku_key, - itemCount: Number(row.item_count || 0), - })) - const pendingItemCount = candidates.reduce((sum, item) => sum + item.itemCount, 0) - - if (dryRun) { - return { - dryRun: true, - range: hasRange ? { fromDate: fromDate || '', toDate: toDate || '' } : null, - distinctSkuCount: candidates.length, - pendingItemCount, - createdProductCount: 0, - updatedItemCount: 0, - updatedSnapshotCount: 0, - resolvedExceptionCount: 0, - remainingItemCount: pendingItemCount, - } - } - - let createdProductCount = 0 - if (candidates.length > 0) { - const inserted = await query( - ` - INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, - attributes_json, created_at, updated_at - ) - SELECT - input.canonical_sku, - input.canonical_name, - $2, - $3, - 'active', - jsonb_build_object('source', 'manual_normalization', 'internalSku', input.sku_key), - $4::timestamptz, - $4::timestamptz - FROM unnest($1::text[], $5::text[], $6::text[]) - AS input(canonical_sku, canonical_name, sku_key) - ON CONFLICT (canonical_sku) DO NOTHING - RETURNING id - `, - [ - candidates.map((item) => item.canonicalSku), - MANUAL_PRODUCT_CATEGORY, - MANUAL_PRODUCT_UNIT, - now, - candidates.map((item) => item.canonicalName), - candidates.map((item) => item.skuKey), - ], - ) - createdProductCount = inserted.rowCount || 0 - } - - // 回填订单项：按 id 分批，每批重新取日期参数，避免单条语句超过 15s。 - let updatedItemCount = 0 - let updatedSnapshotCount = 0 - let lastId = 0 - for (let batch = 0; batch < 40; batch += 1) { - const range = buildRangeClause(3) - const updated = await query( - ` - WITH target AS ( - SELECT oi.id, p.id AS product_id - FROM order_items oi - ${joinSql} - JOIN product_catalog p - ON p.canonical_sku = 'man:' || ${MANUAL_SKU_KEY_SQL} - WHERE oi.fulfillment_mode = 'manual' - AND oi.canonical_product_id IS NULL - AND oi.id > $1 - ${range.sql} - ORDER BY oi.id - LIMIT $2 - ) - UPDATE order_items oi - SET canonical_product_id = t.product_id, - updated_at = NOW() - FROM target t - WHERE oi.id = t.id - AND oi.canonical_product_id IS NULL - RETURNING oi.id - `, - [lastId, NORMALIZE_BATCH_SIZE, ...range.params], - ) - const rowCount = updated.rowCount || 0 - if (rowCount === 0) break - updatedItemCount += rowCount - const ids = updated.rows as Array<{ id: number }> - lastId = Number(ids[ids.length - 1]?.id || lastId) - // 已有成本快照补商品维度：**只处理本批刚归一过的订单项**。 - // 原来是无条件全表条件更新，要扫 900MB 的 order_items（堆 + 两个位图索引回表）， - // 生产实测每次 15~16 秒、必然撞 statement_timeout，而且失败会把后半段 - // （销售事实重算、影子总账同步）整条链跳过。收口到本批 id 后走 idx_cost_snapshots_item， - // 毫秒级完成，且结果完全一样（唯一会变的行就是这些刚补上商品的订单项）。 - if (ids.length > 0) { - const updatedSnapshots = await query( - ` - UPDATE cost_snapshots cs - SET canonical_product_id = oi.canonical_product_id - FROM order_items oi - WHERE oi.id = ANY($1::bigint[]) - AND cs.order_item_id = oi.id - AND cs.canonical_product_id IS NULL - AND oi.canonical_product_id IS NOT NULL - `, - [ids.map((row) => Number(row.id))], - ) - updatedSnapshotCount += updatedSnapshots.rowCount || 0 - } - if (rowCount < NORMALIZE_BATCH_SIZE) break - } - - let resolvedExceptionCount = 0 - if (input.resolveExceptions !== false && updatedItemCount > 0) { - const resolved = await query( - ` - UPDATE cost_exceptions ce - SET status = 'resolved', resolved_at = $1::timestamptz - WHERE ce.status = 'open' - AND ce.subject_type = 'order_item' - AND ce.reason = 'product_mapping_missing' - AND EXISTS ( - SELECT 1 FROM order_items oi - WHERE oi.id = ce.subject_id AND oi.canonical_product_id IS NOT NULL - ) - `, - [now], - ) - resolvedExceptionCount = resolved.rowCount || 0 - } - - const remainingRange = buildRangeClause(1) - const remaining = await query<{ total: number }>( - ` - SELECT COUNT(*)::int AS total - FROM order_items oi - ${joinSql} - WHERE oi.fulfillment_mode = 'manual' - AND oi.canonical_product_id IS NULL - ${remainingRange.sql} - `, - remainingRange.params, - ) - - return { - dryRun: false, - range: hasRange ? { fromDate: fromDate || '', toDate: toDate || '' } : null, - distinctSkuCount: candidates.length, - pendingItemCount, - createdProductCount, - updatedItemCount, - updatedSnapshotCount, - resolvedExceptionCount, - remainingItemCount: Number(remaining.rows[0]?.total || 0), - } -} - -function normalizeDate(value: unknown): string { - const text = String(value || '').trim() - if (!text) return '' - if (!/^\d{4}-\d{2}-\d{2}$/.test(text)) return '' - return text -} diff --git a/apps/backend/src/services/cost-center/manual-product-normalization.test.ts b/apps/backend/src/services/cost-center/manual-product-normalization.test.ts deleted file mode 100644 index d1c54dd7..00000000 --- a/apps/backend/src/services/cost-center/manual-product-normalization.test.ts +++ /dev/null @@ -1,60 +0,0 @@ -import assert from 'node:assert/strict' -import test from 'node:test' - -/** - * 2026-10-30 生产事故复盘：cost-finance-sync 每 15 分钟的「成本快照补商品维度」 - * 原先是一条无条件全表更新，要扫 899MB 的 order_items，实测 15.2s / 16.5s 撞 - * statement_timeout，并且失败会把后面的「销售事实重算 + 影子总账同步」整条跳过。 - * - * 修复口径：只更新本批刚归一过的 order_item_id（走 idx_cost_snapshots_item）， - * 结果完全一样但毫秒级完成。这里用源码断言把它钉住。 - */ -test('成本快照补商品维度必须按本批 order_item_id 收口，不能再全表更新', async () => { - const { readFile } = await import('node:fs/promises') - const src = await readFile( - new URL('./manual-product-normalization-service.ts', import.meta.url), - 'utf8', - ) - - assert.ok( - src.includes('oi.id = ANY($1::bigint[])'), - '成本快照更新缺少本批 id 收口：会退化成扫 900MB 的 order_items（实测 15 秒超时）', - ) - assert.ok( - src.includes('AND cs.order_item_id = oi.id'), - '收口后仍要按 order_item_id 关联，不能改成按 subject_id 猜', - ) - - // 更新必须发生在分批循环内部（本批 ids 还在作用域里），不能在循环之后 - const loopStart = src.indexOf('for (let batch = 0; batch < 40; batch += 1) {') - const loopEnd = src.indexOf('let resolvedExceptionCount = 0') - const snapshotUpdateAt = src.indexOf('UPDATE cost_snapshots cs') - assert.ok(loopStart > 0 && loopEnd > loopStart, '找不到归一的分批循环') - assert.ok( - snapshotUpdateAt > loopStart && snapshotUpdateAt < loopEnd, - '成本快照更新必须留在分批循环内：循环外拿不到本批 ids，就只能退回全表条件', - ) - - // 兜底：不允许再出现「无 id 限制 + fulfillment_mode 条件」的老写法 - const legacyPattern = /UPDATE cost_snapshots cs[\s\S]{0,400}?oi\.fulfillment_mode = 'manual'/ - assert.equal( - legacyPattern.test(src), - false, - '不能再按 fulfillment_mode 全表更新（会整表 bitmap 回表）', - ) -}) - -test('迁移 110 提供 order_items(fulfillment_mode, canonical_product_id) 覆盖索引', async () => { - const { readFile } = await import('node:fs/promises') - const migration = await readFile( - new URL('../../db/migrations/110_order_items_fulfillment_product_index.sql', import.meta.url), - 'utf8', - ) - assert.ok(migration.includes('idx_order_items_fulfillment_product')) - assert.ok(migration.includes('(fulfillment_mode, canonical_product_id)')) - assert.ok(migration.includes('INCLUDE (id)')) - assert.ok( - !/^\s*CREATE INDEX CONCURRENTLY/m.test(migration), - '迁移在事务内执行，不能使用 CONCURRENTLY', - ) -}) diff --git a/apps/backend/src/services/finance/finance-controls.test.ts b/apps/backend/src/services/finance/finance-controls.test.ts index 6968f992..717a2815 100644 --- a/apps/backend/src/services/finance/finance-controls.test.ts +++ b/apps/backend/src/services/finance/finance-controls.test.ts @@ -5,7 +5,7 @@ import test from 'node:test' const read = (path: URL) => readFile(path, 'utf8') test('销售事实重算必须冻结已关账期间', async () => { - const source = await read(new URL('./sales-facts-service.ts', import.meta.url)) + const source = await read(new URL('./sales-facts-channel-service.ts', import.meta.url)) assert.ok(source.includes('finance_periods'), '重算必须读取会计期间') assert.ok(source.includes("p.status = 'closed'"), '必须识别已关账期间') assert.ok(source.includes('skippedClosedPeriods'), '必须返回被跳过的关账期间') diff --git a/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts b/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts index 80904211..fb2a4973 100644 --- a/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts +++ b/apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts @@ -85,7 +85,7 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务 new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url), 'utf8', ), - readFile(new URL('../scheduler/cost-finance-sync-job.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'), ]) assert.ok( @@ -99,10 +99,9 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务 '保存人工 SKU 配置后必须立即回填最近 30 天订单项', ) assert.ok( - jobSource.includes('rematchManualFulfillments({') && - jobSource.includes('onlyUnmatched: true') && - jobSource.includes('dateFrom:'), - '定时同步必须自动回填未匹配的人工订单项，并限制在最近窗口', + jobSource.includes("fulfillment_channel = 'manual'") && + jobSource.includes('fulfillment_config_id'), + '人工载体管道必须挂载人工配置', ) assert.ok( configSource.includes('manual_sku_key'), @@ -117,21 +116,20 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务 ) }) -test('人工成本缺失必须进入异常队列，且只监控首次配置之后的订单项', async () => { - const source = await readFile( - new URL('../cost-center/cost-snapshot-sync-service.ts', import.meta.url), - 'utf8', - ) - assert.ok(source.includes("'manual_cost_missing'"), '缺少人工成本缺失异常') +test('人工成本缺失由管道按配置与快照生成并收口', async () => { + const [pipelineSource, manualSource] = await Promise.all([ + readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'), + readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'), + ]) + assert.ok(pipelineSource.includes("'manual_cost_missing'"), '管道必须产出人工成本缺失异常') assert.ok( - source.includes('FROM manual_fulfillment_configs') && - source.includes('s.basis_at >= c.enforce_from'), - '人工异常必须以首次配置时间为起点，避免历史人工单刷屏', + pipelineSource.includes("cs.subject_type = 'manual_case'") && + pipelineSource.includes('cs.order_item_id = oi.id'), + '人工成本缺失必须以人工快照是否存在为判定', ) assert.ok( - source.includes("cs.subject_type = 'manual_case'") && - source.includes('cs.order_item_id = s.order_item_id'), - '人工异常回填/收口必须按订单项关联人工快照', + manualSource.includes("reason = 'manual_cost_missing'"), + '确认履约后必须收口人工成本缺失异常', ) }) diff --git a/apps/backend/src/services/finance/manual-fulfillment.integration.test.ts b/apps/backend/src/services/finance/manual-fulfillment.integration.test.ts index 862de978..9223ea8a 100644 --- a/apps/backend/src/services/finance/manual-fulfillment.integration.test.ts +++ b/apps/backend/src/services/finance/manual-fulfillment.integration.test.ts @@ -21,8 +21,6 @@ test( process.env.DATABASE_URL = TEST_DATABASE_URL const { query, withTransaction, closeDb } = await import('../../db/client.js') const { confirmManualFulfillment } = await import('./manual-fulfillment-service.js') - const { syncEstimatedOrderItemCostSnapshots } = - await import('../cost-center/cost-snapshot-sync-service.js') const { rematchManualFulfillments } = await import('../cost-center/manual-fulfillment-config-service.js') const { @@ -364,14 +362,6 @@ test( orderIds.push(pendingItem.orderId) orderItemIds.push(pendingItem.orderItemId) - await syncEstimatedOrderItemCostSnapshots({ dateFrom: today, dateTo: today }) - const openManualException = await query<{ count: number }>( - `SELECT COUNT(*)::int AS count FROM cost_exceptions - WHERE reason = 'manual_cost_missing' AND status = 'open' AND subject_id = $1`, - [pendingItem.orderItemId], - ) - assert.equal(Number(openManualException.rows[0]?.count), 1, '人工缺成本必须进异常队列') - await confirmManualFulfillment( { orderItemId: pendingItem.orderItemId, costFen: 100 }, session, @@ -382,13 +372,6 @@ test( [pendingItem.orderItemId], ) assert.equal(Number(openAfterConfirm.rows[0]?.count), 0, '确认后应立即收口人工异常') - await syncEstimatedOrderItemCostSnapshots({ dateFrom: today, dateTo: today }) - const resolvedManualException = await query<{ count: number }>( - `SELECT COUNT(*)::int AS count FROM cost_exceptions - WHERE reason = 'manual_cost_missing' AND status = 'resolved' AND subject_id = $1`, - [pendingItem.orderItemId], - ) - assert.equal(Number(resolvedManualException.rows[0]?.count), 1, '确认后异常必须自动收口') // 7. 总账：销售事实同步、试算平衡、关账阻断、平台对账匹配 const todayDate = shanghaiDate() diff --git a/apps/backend/src/services/finance/order-refund-service.ts b/apps/backend/src/services/finance/order-refund-service.ts index 16427be8..8e68ddaf 100644 --- a/apps/backend/src/services/finance/order-refund-service.ts +++ b/apps/backend/src/services/finance/order-refund-service.ts @@ -14,7 +14,7 @@ import type { AdminSession } from '../admin/admin-auth-service.js' import { createHttpError } from '../../utils/http.js' import { generateRefundNo } from '../../utils/random.js' import { nowIso } from '../../utils/time.js' -import { rebuildSalesDailyFacts } from './sales-facts-service.js' +import { rebuildChannelSalesDailyFacts } from './sales-facts-channel-service.js' import { approveKuaishouIndustryRefund } from '../platforms/kuaishou-industry/refund-service.js' import { syncFinanceLedgerFromFacts } from './finance-ledger-service.js' @@ -328,7 +328,7 @@ export async function createManualRefund( const basisDate = toChinaDate(written.basisAt) if (basisDate) { - await rebuildSalesDailyFacts({ dateFrom: basisDate, dateTo: basisDate }) + await rebuildChannelSalesDailyFacts({ dateFrom: basisDate, dateTo: basisDate }) await syncFinanceLedgerFromFacts({ dateFrom: basisDate, dateTo: basisDate }) } @@ -442,7 +442,7 @@ export async function voidOrderRefund( const rebuiltDate = toChinaDate(context.basisAt) if (rebuiltDate) { - await rebuildSalesDailyFacts({ dateFrom: rebuiltDate, dateTo: rebuiltDate }) + await rebuildChannelSalesDailyFacts({ dateFrom: rebuiltDate, dateTo: rebuiltDate }) await syncFinanceLedgerFromFacts({ dateFrom: rebuiltDate, dateTo: rebuiltDate }) } const refunds = await listOrderRefunds(context.orderId) diff --git a/apps/backend/src/services/finance/sales-facts-service.ts b/apps/backend/src/services/finance/sales-facts-service.ts index f0954035..6d93dbbf 100644 --- a/apps/backend/src/services/finance/sales-facts-service.ts +++ b/apps/backend/src/services/finance/sales-facts-service.ts @@ -1,7 +1,6 @@ -import { query, withTransaction } from '../../db/client.js' +import { query } from '../../db/client.js' import { createHttpError } from '../../utils/http.js' import { resolveFinanceDateRange } from '../../utils/finance-date-range.js' -import { nowIso } from '../../utils/time.js' export type SalesFactsInput = { dateFrom?: unknown @@ -10,14 +9,6 @@ export type SalesFactsInput = { to?: unknown } -export type SalesFactsRebuildResult = { - range: { fromDate: string; toDate: string } - deletedCount: number - insertedCount: number - skippedClosedPeriods: string[] - computedAt: string -} - type SalesFactRow = { [column: string]: unknown id: number @@ -45,167 +36,6 @@ type SalesFactRow = { * 重算指定日期范围的销售事实。日期统一按 Asia/Shanghai 归属。 * 未归一商品允许 canonical_product_id 为空，汇总时形成“未归一”桶。 */ -export async function rebuildSalesDailyFacts( - input: SalesFactsInput = {}, -): Promise { - const range = resolveFinanceDateRange(input, { defaultDays: 30, maxDays: 400 }) - const computedAt = nowIso() - - // 已关账期间必须冻结：只重算开放期间，避免“总账已关账、报表事实还在变”。 - const closedPeriods = await query<{ period_key: string }>( - `SELECT period_key - FROM finance_periods - WHERE status = 'closed' - AND period_key BETWEEN $1 AND $2 - ORDER BY period_key`, - [range.fromDate.slice(0, 7), range.toDate.slice(0, 7)], - ) - const skippedClosedPeriods = closedPeriods.rows.map((row) => row.period_key) - - return withTransaction(async (client) => { - const deleted = await client.query( - `DELETE FROM sales_daily_facts f - WHERE f.stat_date BETWEEN $1::date AND $2::date - AND NOT EXISTS ( - SELECT 1 FROM finance_periods p - WHERE p.status = 'closed' - AND p.period_key = to_char(f.stat_date, 'YYYY-MM') - )`, - [range.fromDate, range.toDate], - ) - - const inserted = await client.query( - ` - WITH scoped_orders AS MATERIALIZED ( - -- 先物化 7 天窗口内的订单集合，避免 planner 对每个订单项做一次 - -- orders 随机回表；后续全部走 hash/merge join。 - SELECT - o.id, - o.provider, - o.shop_id, - o.pay_status, - COALESCE(o.paid_at, o.created_at) AS basis_at - FROM orders o - WHERE o.pay_status IN ('paid', 'refunded') - AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date - BETWEEN $1::date AND $2::date - AND NOT EXISTS ( - SELECT 1 FROM finance_periods p - WHERE p.status = 'closed' - AND p.period_key = to_char( - (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date, - 'YYYY-MM' - ) - ) - ), - scoped_items AS MATERIALIZED ( - SELECT - oi.id, - oi.order_id, - oi.canonical_product_id, - oi.fulfillment_mode, - oi.revenue_amount_fen, - oi.normalized_quantity, - oi.quantity - FROM order_items oi - WHERE oi.order_id IN (SELECT id FROM scoped_orders) - ), - cost_by_item AS ( - -- 一次全量聚合后再 hash join，替代逐项 LATERAL 查 cost_snapshots。 - SELECT - order_item_id, - SUM(total_amount_fen)::bigint AS cost_fen, - COUNT(*)::int AS snapshot_count, - COUNT(*) FILTER (WHERE snapshot_stage = 'actual')::int AS actual_count - FROM cost_snapshots - WHERE superseded_at IS NULL - GROUP BY order_item_id - ), - refunds_by_item AS ( - SELECT - order_item_id, - SUM(amount_fen)::bigint AS refund_fen, - COUNT(*)::int AS refund_count - FROM order_refunds - WHERE status = 'confirmed' - GROUP BY order_item_id - ), - item_base AS ( - SELECT - (o.basis_at AT TIME ZONE 'Asia/Shanghai')::date AS stat_date, - o.provider, - o.shop_id, - oi.canonical_product_id, - COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode, - oi.order_id, - COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, - CASE - WHEN COALESCE(refunds.refund_count, 0) > 0 - THEN LEAST( - COALESCE(refunds.refund_fen, 0), - GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0) - )::bigint - WHEN o.pay_status = 'refunded' - THEN GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)::bigint - ELSE 0::bigint - END AS refund_fen, - GREATEST(COALESCE(oi.normalized_quantity, oi.quantity, 1), 1)::numeric AS quantity, - COALESCE(cost.cost_fen, 0)::bigint AS cost_fen, - COALESCE(cost.snapshot_count, 0)::int AS snapshot_count, - COALESCE(cost.actual_count, 0)::int AS actual_count - FROM scoped_items oi - JOIN scoped_orders o ON o.id = oi.order_id - LEFT JOIN cost_by_item cost ON cost.order_item_id = oi.id - LEFT JOIN refunds_by_item refunds ON refunds.order_item_id = oi.id - ) - INSERT INTO sales_daily_facts ( - stat_date, provider, shop_id, canonical_product_id, fulfillment_mode, - order_count, item_count, quantity, - gross_revenue_fen, refund_amount_fen, net_revenue_fen, - cost_amount_fen, cost_missing_count, cost_stage, gross_profit_fen, - computed_at, created_at, updated_at - ) - SELECT - stat_date, - provider, - shop_id, - canonical_product_id, - fulfillment_mode, - COUNT(DISTINCT order_id)::int, - COUNT(*)::int, - SUM(quantity), - SUM(revenue_fen), - SUM(refund_fen), - SUM(revenue_fen - refund_fen), - SUM(cost_fen), - COUNT(*) FILTER (WHERE snapshot_count = 0)::int, - CASE - WHEN COUNT(*) FILTER (WHERE snapshot_count = 0) = COUNT(*) THEN 'missing' - WHEN COUNT(*) FILTER (WHERE snapshot_count = 0) > 0 THEN 'partial' - WHEN COUNT(*) FILTER (WHERE snapshot_count > 0 AND actual_count < snapshot_count) > 0 - THEN 'estimated' - ELSE 'actual' - END, - SUM(revenue_fen - refund_fen) - SUM(cost_fen), - $3::timestamptz, - $3::timestamptz, - $3::timestamptz - FROM item_base - GROUP BY stat_date, provider, shop_id, canonical_product_id, fulfillment_mode - `, - [range.fromDate, range.toDate, computedAt], - ) - - return { - range, - deletedCount: deleted.rowCount || 0, - insertedCount: inserted.rowCount || 0, - skippedClosedPeriods, - computedAt, - } - }) -} - export async function getSalesProfitOverview(input: SalesFactsInput = {}) { const range = resolveFinanceDateRange(input, { defaultDays: 30, maxDays: 400 }) diff --git a/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts b/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts deleted file mode 100644 index 155fb8ba..00000000 --- a/apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts +++ /dev/null @@ -1,75 +0,0 @@ -import test from 'node:test' -import assert from 'node:assert/strict' - -import { classifyFulfillmentMode, resolveReconcileCursor } from './fulfillment-mode-service.js' - -test('履约判定优先级：工单 > 自动发货 > 历史平台 > 人工', () => { - assert.equal( - classifyFulfillmentMode({ - hasWorkOrder: true, - hasAffiliateDashTask: true, - hasLegacyPlatformTask: true, - }), - 'work_order', - ) - assert.equal( - classifyFulfillmentMode({ - hasWorkOrder: false, - hasAffiliateDashTask: true, - hasLegacyPlatformTask: true, - }), - 'affiliate_dash', - ) - assert.equal( - classifyFulfillmentMode({ - hasWorkOrder: false, - hasAffiliateDashTask: false, - hasLegacyPlatformTask: true, - }), - 'legacy_platform', - ) - assert.equal( - classifyFulfillmentMode({ - hasWorkOrder: false, - hasAffiliateDashTask: false, - hasLegacyPlatformTask: false, - }), - 'manual', - ) -}) - -test('keyset 游标：整批取满继续前进', () => { - assert.deepEqual( - resolveReconcileCursor({ - cursor: 0, - selectedCount: 2000, - lastScannedId: 2000, - batchSize: 2000, - }), - { nextCursor: 2000, done: false }, - ) -}) - -test('keyset 游标：不足一批表示已经扫到底', () => { - assert.deepEqual( - resolveReconcileCursor({ - cursor: 1000, - selectedCount: 999, - lastScannedId: 1999, - batchSize: 2000, - }), - { nextCursor: 1999, done: true }, - ) -}) - -test('keyset 游标：游标没有前进时强制结束，避免死循环', () => { - assert.deepEqual( - resolveReconcileCursor({ - cursor: 500, - selectedCount: 2000, - lastScannedId: 500, - batchSize: 2000, - }), - { nextCursor: 500, done: true }, - ) -}) diff --git a/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts b/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts deleted file mode 100644 index f1d7e784..00000000 --- a/apps/backend/src/services/fulfillment/fulfillment-mode-service.ts +++ /dev/null @@ -1,238 +0,0 @@ -import { query } from '../../db/client.js' -import { nowIso } from '../../utils/time.js' - -export type OrderFulfillmentMode = 'work_order' | 'affiliate_dash' | 'legacy_platform' | 'manual' - -export type FulfillmentModeDecisionInput = { - hasWorkOrder: boolean - hasAffiliateDashTask: boolean - hasLegacyPlatformTask: boolean -} - -export type ReconcileFulfillmentModesOptions = { - batchSize?: number - maxBatches?: number - reason?: string - /** 从该 order_item id 之后继续扫（keyset 分页；默认 0 表示从头开始）。 */ - afterId?: number -} - -export type ReconcileFulfillmentModesResult = { - updatedCount: number - remainingCount: number - batches: number - reason: string - finishedAt: string - /** 本次扫到的最大 order_item id；可作为下一次的 afterId 续跑。 */ - lastScannedId: number -} - -/** - * 统一履约判定（纯函数，供单测与文档对照）。 - * - * 优先级固定：工单 → 自动发货任务 → 历史平台任务 → 人工。 - * 核销码、provider、商品来源都不参与判定。 - */ -export function classifyFulfillmentMode({ - hasWorkOrder, - hasAffiliateDashTask, - hasLegacyPlatformTask, -}: FulfillmentModeDecisionInput): OrderFulfillmentMode { - if (hasWorkOrder) return 'work_order' - if (hasAffiliateDashTask) return 'affiliate_dash' - if (hasLegacyPlatformTask) return 'legacy_platform' - return 'manual' -} - -const FULFILLMENT_MODE_CASE_SQL = ` - CASE - WHEN EXISTS ( - SELECT 1 FROM work_orders wo - WHERE wo.order_item_id = oi.id - ) THEN 'work_order' - WHEN EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id - AND ft.executor_key = 'affiliate_dash' - ) THEN 'affiliate_dash' - WHEN EXISTS ( - SELECT 1 FROM fulfillment_tasks ft - WHERE ft.order_item_id = oi.id - AND ft.executor_key IN ('kuaishou_ct_assisted', 'kuaishou_feifei') - ) THEN 'legacy_platform' - ELSE 'manual' - END -` - -/** - * 重新判定指定订单项的履约方式。 - * - * 只在模式发生变化时更新 fulfillment_decided_at / fulfillment_reason； - * 同一订单项重复调用是幂等的。 - */ -export async function decideOrderItemsFulfillmentModes( - orderItemIds: Array, - { reason = 'order_sync' }: { reason?: string } = {}, -): Promise<{ updatedCount: number }> { - const ids = Array.from( - new Set( - orderItemIds - .map((value) => Number(value)) - .filter((value) => Number.isInteger(value) && value > 0), - ), - ) - if (ids.length === 0) return { updatedCount: 0 } - - const result = await query<{ fulfillment_mode: OrderFulfillmentMode }>( - ` - WITH decided AS ( - SELECT oi.id, ${FULFILLMENT_MODE_CASE_SQL} AS mode - FROM order_items oi - WHERE oi.id = ANY($1::bigint[]) - ) - UPDATE order_items oi - SET - fulfillment_mode = d.mode, - fulfillment_decided_at = NOW(), - fulfillment_reason = $2 - FROM decided d - WHERE oi.id = d.id - AND ( - oi.fulfillment_mode IS NULL - OR oi.fulfillment_mode IS DISTINCT FROM d.mode - ) - RETURNING oi.fulfillment_mode - `, - [ids, String(reason || 'order_sync')], - ) - - return { updatedCount: result.rowCount || 0 } -} - -/** - * 批处理游标推进。 - * - * keyset 分页必须保证单调前进，否则同一批数据会被反复取到、陷入死循环。 - * selectedCount < batchSize 表示已经扫到底；lastScannedId 没有前进同样视为结束。 - */ -export function resolveReconcileCursor(input: { - cursor: number - selectedCount: number - lastScannedId: number - batchSize: number -}): { nextCursor: number; done: boolean } { - const nextCursor = Math.max(input.cursor, input.lastScannedId) - const done = input.selectedCount < input.batchSize || nextCursor <= input.cursor - return { nextCursor, done } -} - -/** - * 兜底归集：分批处理 fulfillment_mode 为 NULL 的订单项。 - * - * 定时任务只需处理新单；历史回填用同一个函数，把 maxBatches 调大即可。 - * 走 keyset（id > 上一批最大 id），每批只扫索引里剩下的未判定行， - * 不会像 OFFSET 那样反复从头扫已回填的数据。 - */ -export async function reconcilePendingFulfillmentModes({ - batchSize = 2000, - maxBatches = 10, - reason = 'reconcile', - afterId = 0, -}: ReconcileFulfillmentModesOptions = {}): Promise { - const normalizedBatchSize = clampInteger(batchSize, 2000, 1, 50_000) - const normalizedMaxBatches = clampInteger(maxBatches, 10, 1, 10_000) - const normalizedReason = String(reason || 'reconcile').trim() || 'reconcile' - let cursor = Math.max(0, Math.trunc(Number(afterId) || 0)) - let updatedCount = 0 - let batches = 0 - - for (let index = 0; index < normalizedMaxBatches; index += 1) { - // keyset：每批从上一批最大 id 之后继续，避免每次从头扫已回填过的行。 - // 返回 pending 的数量与最大 id，用于判断是否扫到底、以及下一批的起点。 - const result = await query<{ - updated_count: number - selected_count: number - last_id: number | string | null - }>( - ` - WITH pending AS ( - SELECT oi.id - FROM order_items oi - WHERE oi.fulfillment_mode IS NULL - AND oi.id > $3 - ORDER BY oi.id - LIMIT $1 - ), - decided AS ( - SELECT oi.id, ${FULFILLMENT_MODE_CASE_SQL} AS mode - FROM order_items oi - JOIN pending p ON p.id = oi.id - ), - updated AS ( - UPDATE order_items oi - SET - fulfillment_mode = d.mode, - fulfillment_decided_at = NOW(), - fulfillment_reason = $2 - FROM decided d - WHERE oi.id = d.id - AND oi.fulfillment_mode IS NULL - RETURNING oi.id - ) - SELECT - (SELECT COUNT(*)::int FROM updated) AS updated_count, - (SELECT COUNT(*)::int FROM pending) AS selected_count, - (SELECT COALESCE(MAX(id), 0)::bigint FROM pending) AS last_id - `, - [normalizedBatchSize, normalizedReason, cursor], - ) - - const row = result.rows[0] - const step = resolveReconcileCursor({ - cursor, - selectedCount: Number(row?.selected_count || 0), - lastScannedId: Number(row?.last_id || 0), - batchSize: normalizedBatchSize, - }) - updatedCount += Number(row?.updated_count || 0) - batches += 1 - cursor = step.nextCursor - if (step.done) break - } - - const remaining = await query<{ count: number }>( - `SELECT COUNT(*)::int AS count FROM order_items WHERE fulfillment_mode IS NULL`, - ) - - return { - updatedCount, - remainingCount: Number(remaining.rows[0]?.count || 0), - batches, - reason: normalizedReason, - finishedAt: nowIso(), - lastScannedId: cursor, - } -} - -export async function getFulfillmentModeCounts(): Promise< - Array<{ fulfillmentMode: OrderFulfillmentMode | null; count: number }> -> { - const result = await query<{ fulfillment_mode: OrderFulfillmentMode | null; count: number }>( - ` - SELECT fulfillment_mode, COUNT(*)::int AS count - FROM order_items - GROUP BY fulfillment_mode - ORDER BY COUNT(*) DESC - `, - ) - return result.rows.map((row) => ({ - fulfillmentMode: row.fulfillment_mode || null, - count: Number(row.count || 0), - })) -} - -function clampInteger(value: unknown, fallback: number, min: number, max: number) { - const parsed = Number(value) - if (!Number.isInteger(parsed)) return fallback - return Math.min(max, Math.max(min, parsed)) -} diff --git a/apps/backend/src/services/scheduler/config-service.test.ts b/apps/backend/src/services/scheduler/config-service.test.ts index dacfc066..1f2322b9 100644 --- a/apps/backend/src/services/scheduler/config-service.test.ts +++ b/apps/backend/src/services/scheduler/config-service.test.ts @@ -41,7 +41,30 @@ test('normalizeScheduledJobsConfig ensures default jobs exist', () => { assert.ok(ids.includes('deposit-unfreeze')) assert.ok(ids.includes('data-retention')) assert.ok(ids.includes('pdd_order_sync')) - assert.ok(ids.includes('cost-finance-sync')) + // 旧财务闭环作业已下线，不再作为默认任务注入。 + assert.equal(ids.includes('cost-finance-sync'), false) + assert.ok(ids.includes('channel-assign')) + assert.ok(ids.includes('ledger-post')) +}) + +test('旧的 cost_finance_sync 任务已彻底移除，不再出现在配置里', () => { + const config = normalizeScheduledJobsConfig({ + enabled: true, + jobs: [ + { + id: 'cost-finance-sync', + type: 'cost_finance_sync', + enabled: true, + intervalSeconds: 900, + config: { days: 7 }, + }, + ], + }) + + assert.equal( + config.jobs.some((item) => item.type === 'cost_finance_sync'), + false, + ) }) test('normalizeScheduledJobsConfig 收录 v2 六个管道作业并保持各自配置', () => { diff --git a/apps/backend/src/services/scheduler/config-service.ts b/apps/backend/src/services/scheduler/config-service.ts index cf5860ea..923f9078 100644 --- a/apps/backend/src/services/scheduler/config-service.ts +++ b/apps/backend/src/services/scheduler/config-service.ts @@ -10,7 +10,6 @@ const WORK_ORDER_TIMEOUT_JOB_ID = 'work-order-timeout' const DEPOSIT_UNFREEZE_JOB_ID = 'deposit-unfreeze' const DATA_RETENTION_JOB_ID = 'data-retention' const PDD_ORDER_SYNC_JOB_ID = 'pdd_order_sync' -const COST_FINANCE_SYNC_JOB_ID = 'cost-finance-sync' /** v2 管道作业：各自只负责一层，可单独启停。 */ const CHANNEL_ASSIGN_JOB_ID = 'channel-assign' @@ -84,7 +83,6 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) { const hasDepositUnfreeze = jobs.some((item) => item.id === DEPOSIT_UNFREEZE_JOB_ID) const hasDataRetention = jobs.some((item) => item.id === DATA_RETENTION_JOB_ID) const hasPddOrderSync = jobs.some((item) => item.id === PDD_ORDER_SYNC_JOB_ID) - const hasCostFinanceSync = jobs.some((item) => item.id === COST_FINANCE_SYNC_JOB_ID) const hasChannelAssign = jobs.some((item) => item.id === CHANNEL_ASSIGN_JOB_ID) const hasCarrierDispatch = jobs.some((item) => item.id === CARRIER_DISPATCH_JOB_ID) const hasProductNormalize = jobs.some((item) => item.id === PRODUCT_NORMALIZE_JOB_ID) @@ -104,9 +102,8 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) { if (!hasPddOrderSync) { jobs.push(createDefaultPddOrderSyncJob()) } - if (!hasCostFinanceSync) { - jobs.push(createDefaultCostFinanceSyncJob()) - } + // 旧作业 cost_finance_sync 已下线：v2 管道接管，且它写旧字段会覆盖新口径。 + // 保留配置项仅用于兼容读取，运行期强制停用。 // v2 管道：老配置里没有这些任务时自动补齐，保证升级后无需手工添加。 if (!hasChannelAssign) jobs.push(createDefaultChannelAssignJob()) if (!hasCarrierDispatch) jobs.push(createDefaultCarrierDispatchJob()) @@ -143,9 +140,6 @@ function normalizeScheduledJob(rawValue: unknown) { if (type === 'pdd_order_sync') { return normalizePddOrderSyncJob(rawValue) } - if (type === 'cost_finance_sync') { - return normalizeCostFinanceSyncJob(rawValue) - } if (type === 'channel_assign') { return normalizeChannelAssignJob(rawValue) } @@ -182,20 +176,6 @@ function normalizePddOrderSyncJob(rawValue: JsonObject) { } } -function normalizeCostFinanceSyncJob(rawValue: JsonObject) { - const config = isPlainObject(rawValue.config) ? rawValue.config : {} - - return { - id: COST_FINANCE_SYNC_JOB_ID, - type: 'cost_finance_sync', - enabled: rawValue.enabled !== false, - intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 3600), - config: { - days: normalizeRangeInteger(config.days, 7, 1, 90), - }, - } -} - function normalizeDepositUnfreezeJob(rawValue: JsonObject) { const config = isPlainObject(rawValue.config) ? rawValue.config : {} @@ -242,7 +222,6 @@ function createDefaultScheduledJobsConfig() { createDefaultDepositUnfreezeJob(), createDefaultDataRetentionJob(), createDefaultPddOrderSyncJob(), - createDefaultCostFinanceSyncJob(), createDefaultChannelAssignJob(), createDefaultCarrierDispatchJob(), createDefaultProductNormalizeJob(), @@ -406,21 +385,6 @@ function createDefaultPddOrderSyncJob() { } } -function createDefaultCostFinanceSyncJob() { - return { - id: COST_FINANCE_SYNC_JOB_ID, - type: 'cost_finance_sync', - enabled: true, - // 一次同步要跑 6 条全窗口聚合（41 万单实测 30–60 秒）。5 分钟一次会让 - // 数据库长期处在 10–20% 的同步负载下，后台页面首屏被拖慢；财务事实 - // 15 分钟刷新已足够，需要更快可在后台定时任务里改回 300。 - intervalSeconds: 900, - config: { - days: 7, - }, - } -} - function createDefaultDataRetentionJob() { return { id: DATA_RETENTION_JOB_ID, diff --git a/apps/backend/src/services/scheduler/cost-finance-sync-job.ts b/apps/backend/src/services/scheduler/cost-finance-sync-job.ts deleted file mode 100644 index 0a2cf65e..00000000 --- a/apps/backend/src/services/scheduler/cost-finance-sync-job.ts +++ /dev/null @@ -1,108 +0,0 @@ -import type { JsonObject } from '../../types/json.js' -import { - syncActualWorkOrderCostSnapshots, - syncEstimatedOrderItemCostSnapshots, -} from '../cost-center/cost-snapshot-sync-service.js' -import { normalizeManualProducts } from '../cost-center/manual-product-normalization-service.js' -import { rematchManualFulfillments } from '../cost-center/manual-fulfillment-config-service.js' -import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js' -import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-service.js' -import { reconcilePendingFulfillmentModes } from '../fulfillment/fulfillment-mode-service.js' - -/** - * 财务闭环同步：履约方式兜底 -> 人工商品归一 -> 预估成本快照 -> 工单实际快照 -> 销售事实重算。 - * 只处理最近 N 天，避免定时任务扫全量历史。 - */ -export async function runCostFinanceSyncJob(job: JsonObject = {}) { - const config = job.config && typeof job.config === 'object' ? (job.config as JsonObject) : {} - const days = clampInteger(config.days, 30, 1, 90) - - const fulfillment = await reconcilePendingFulfillmentModes({ - batchSize: 2000, - maxBatches: 10, - reason: 'cost_finance_sync', - }) - // 人工单按内部 SKU 归一；异常收口交给预估同步统一处理，这里只补商品维度。 - const manualProducts = await normalizeManualProducts({ - dateFrom: shiftDays(-(days - 1)), - dateTo: today(), - resolveExceptions: false, - }) - // 人工 SKU 配置可能在订单之后才补：每次同步都按配置回填未匹配订单项， - // 保证「配置好人工之后的新单」自动挂上渠道，不依赖管理员手动点「重新匹配」。 - const manualMatches = await rematchManualFulfillments({ - onlyUnmatched: true, - dateFrom: shiftDays(-(days - 1)), - dateTo: today(), - }) - - const estimated = await syncEstimatedOrderItemCostSnapshots({ - dateFrom: shiftDays(-(days - 1)), - dateTo: today(), - }) - const actual = await syncActualWorkOrderCostSnapshots({ - dateFrom: shiftDays(-(days - 1)), - dateTo: today(), - }) - const facts = await rebuildSalesDailyFacts({ - dateFrom: shiftDays(-(days - 1)), - dateTo: today(), - }) - // 影子总账按日同步销售事实（幂等；已关账期间会跳过）。 - const ledger = await syncFinanceLedgerFromFacts({ - dateFrom: shiftDays(-(days - 1)), - dateTo: today(), - }) - - return { - status: 'ok', - message: - `履约方式 ${fulfillment.updatedCount} 项 / ` + - `人工商品 ${manualProducts.updatedItemCount} 项 / ` + - `人工渠道匹配 ${manualMatches.updatedCount} 项 / ` + - `成本快照 ${estimated.createdCount} 商品预估 / ` + - `${actual.directRewardCount + actual.upgradedCount} 工单实际 / ` + - `${actual.estimatedRewardCount} 工单在途，销售事实 ${facts.insertedCount} 行，` + - `总账凭证 ${ledger.created} 条`, - days, - fulfillmentModeUpdated: fulfillment.updatedCount, - fulfillmentModeRemaining: fulfillment.remainingCount, - manualItemsNormalized: manualProducts.updatedItemCount, - manualProductsCreated: manualProducts.createdProductCount, - manualItemsMatched: manualMatches.updatedCount, - estimatedCreated: estimated.createdCount, - actualCreated: actual.directRewardCount + actual.upgradedCount, - estimatedRewardCreated: actual.estimatedRewardCount, - exceptionsCreated: estimated.exceptionCreatedCount, - factsInserted: facts.insertedCount, - ledgerEntriesCreated: ledger.created, - ledgerSkippedClosedPeriods: ledger.skipped, - dateFrom: facts.range.fromDate, - dateTo: facts.range.toDate, - } -} - -/** 模块级只构造一次：`new Intl.DateTimeFormat()` 每次约 0.5ms，逐条调用会线性放大。 */ -const SHANGHAI_DATE_FORMATTER = new Intl.DateTimeFormat('en-CA', { - timeZone: 'Asia/Shanghai', - year: 'numeric', - month: '2-digit', - day: '2-digit', -}) - -function today() { - return SHANGHAI_DATE_FORMATTER.format(new Date()) -} - -function shiftDays(days: number) { - const [year, month, day] = today().split('-').map(Number) - const date = new Date(Date.UTC(year || 1970, (month || 1) - 1, day || 1)) - date.setUTCDate(date.getUTCDate() + days) - return date.toISOString().slice(0, 10) -} - -function clampInteger(value: unknown, fallback: number, min: number, max: number) { - const parsed = Number(value) - if (!Number.isInteger(parsed)) return fallback - return Math.min(max, Math.max(min, parsed)) -} diff --git a/apps/backend/src/services/scheduler/scheduler-service.ts b/apps/backend/src/services/scheduler/scheduler-service.ts index ccafbc0f..8a407e25 100644 --- a/apps/backend/src/services/scheduler/scheduler-service.ts +++ b/apps/backend/src/services/scheduler/scheduler-service.ts @@ -6,7 +6,6 @@ import { runDepositUnfreezeJob } from './deposit-unfreeze-job.js' import { runWorkOrderTimeoutJob } from './work-order-timeout-job.js' import { runDataRetentionJob } from './data-retention-job.js' import { runPddOrderSyncJob } from './pdd-order-sync-job.js' -import { runCostFinanceSyncJob } from './cost-finance-sync-job.js' import { runCarrierDispatchJob, runChannelAssignJob, @@ -213,9 +212,6 @@ function dispatchJob(job: JsonObject) { if (job.type === 'pdd_order_sync') { return runPddOrderSyncJob(job) } - if (job.type === 'cost_finance_sync') { - return runCostFinanceSyncJob(job) - } if (job.type === 'channel_assign') { return runChannelAssignJob(job) } diff --git a/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts b/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts index 88911900..c51353fc 100644 --- a/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts +++ b/apps/backend/src/services/worker-platform/sync-worker-orders-service.ts @@ -45,7 +45,7 @@ import { } from './mappers.js' import { getWorkerProductMatchConfig } from './worker-product-match-config-service.js' import { registerWorkOrderShopName } from './work-order-shop-options-cache.js' -import { decideOrderItemsFulfillmentModes } from '../fulfillment/fulfillment-mode-service.js' +import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js' export async function syncWorkerOrdersForSourceOrder( order: OrderRow, @@ -308,12 +308,9 @@ export async function syncWorkerOrdersForSourceOrder( } } - // 订单处理完成后统一落履约事实；失败不阻断下单，由财务定时任务兜底归集。 + // 订单处理完成后立即跑一次渠道判定；失败不阻断下单，由定时任务兜底。 try { - await decideOrderItemsFulfillmentModes( - orderItems.map((item) => Number(item.id)), - { reason: options.source || 'source_order' }, - ) + await assignPendingOrderItemChannels({ batchSize: 200, maxBatches: 1 }) } catch (error) { logWarn('[worker-order-consumer]', '履约方式归集失败，等待定时任务兜底', { orderId: order.id,