dachui/order_site
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(finance)!: 强制渠道模型，移除旧履约方式与同步链路

Browse Source 
- 删除 fulfillment_mode 相关的旧同步/回填链路与定时任务
- 删除 cost_finance_sync、成本快照同步、人工商品归一等遗留服务
- 迁移 113 直接删除 order_items.fulfillment_mode 及其索引约束
- 断代脚本自带归档步骤，不再依赖独立回填脚本
- 退款/事实/总账统一走渠道版销售事实
This commit is contained in:
yml2213
2026-10-31 19:52:03 +08:00
parent 808949f3b8
commit 65b7f4f9dc
25 changed files with 156 additions and 2383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/package.json
-2
View File
@@ -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",
apps/backend/scripts/backfill-fulfillment-channel.ts
-241
View File
@@ -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<string, number>()
for (let batch = 0; batch < maxBatches; batch += 1) {
const rows = await query<BatchRow>(
`
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<string, string> = {}
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))
}
apps/backend/scripts/backfill-fulfillment-mode.ts
-88
View File
@@ -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<string, string> = {}
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))
}
apps/backend/scripts/finance-cutover-v2.ts
+53 -17
View File
@@ -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<string, number> = {}
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() {
apps/backend/scripts/sync-cost-finance.ts
-91
View File
@@ -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<string, unknown> = { 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<string, string> = {}
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)
}
apps/backend/src/db/migrations/113_drop_legacy_fulfillment_mode.sql
+40
View File
@@ -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 断代归档';
apps/backend/src/routes/admin/cost-center-catalog.ts
-44
View File
@@ -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<string, unknown>,
}),
}),
)
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<string, unknown>,
}),
},
),
)
router.get(
'/cost-center/products',
requireAdminPermission('finance.view'),
apps/backend/src/routes/admin/cost-center-cost.ts
-50
View File
@@ -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
apps/backend/src/routes/admin/finance.ts
+10 -5
View File
@@ -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<string, unknown>) : {},
),
(req) => {
const body =
req.body && typeof req.body === 'object' ? (req.body as Record<string, unknown>) : {}
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: '重算销售事实失败',
apps/backend/src/services/cost-center/affiliate-product-service.ts
+4 -26
View File
@@ -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,
}
}
apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts
-806
View File
@@ -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<CostSnapshotSyncResult> {
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<ActualCostSnapshotSyncResult> {
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),
}
}
apps/backend/src/services/cost-center/manual-product-normalization-service.ts
-274
View File
@@ -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<NormalizeManualProductsResult> {
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
}
apps/backend/src/services/cost-center/manual-product-normalization.test.ts
-60
View File
@@ -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',
)
})
apps/backend/src/services/finance/finance-controls.test.ts
+1 -1
View File
@@ -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'), '必须返回被跳过的关账期间')
apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts
+15 -17
View File
@@ -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'"),
'确认履约后必须收口人工成本缺失异常',
)
})
apps/backend/src/services/finance/manual-fulfillment.integration.test.ts
-17
View File
@@ -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()
apps/backend/src/services/finance/order-refund-service.ts
+3 -3
View File
@@ -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)
apps/backend/src/services/finance/sales-facts-service.ts
+1 -171
View File
@@ -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<SalesFactsRebuildResult> {
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 })
apps/backend/src/services/fulfillment/fulfillment-mode-service.test.ts
-75
View File
@@ -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 },
)
})
apps/backend/src/services/fulfillment/fulfillment-mode-service.ts
-238
View File
@@ -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<number | string>,
{ 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<ReconcileFulfillmentModesResult> {
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))
}
apps/backend/src/services/scheduler/config-service.test.ts
+24 -1
View File
@@ -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 六个管道作业并保持各自配置', () => {
apps/backend/src/services/scheduler/config-service.ts
+2 -38
View File
@@ -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,
apps/backend/src/services/scheduler/cost-finance-sync-job.ts
-108
View File
@@ -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))
}
apps/backend/src/services/scheduler/scheduler-service.ts
-4
View File
@@ -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)
}
apps/backend/src/services/worker-platform/sync-worker-orders-service.ts
+3 -6
View File
@@ -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,