refactor(cost-center): 拆分模板引导与成本解析服务

两个千行级服务按「变化原因」拆开,函数体逐字保留:

template-bootstrap-service.ts 1407 行 →
  - template-bootstrap-service.ts   745  preview/apply、映射计划、模板商品编码
  - affiliate-product-service.ts    199  自动发货商品同步与 SKU 回填
  - work-order-product-service.ts   467  接单平台商品回填与模板发单金额同步

cost-definition-service.ts 1382 行 →
  - cost-definition-service.ts      727  成本项/成本版本/成本异常 CRUD
  - cost-resolution-service.ts      680  预览、解析上下文、快照生成

依赖方向是单向的:CRUD 依赖解析层的 4 个校验/取值函数,解析层不反向依赖;
rebuildFactsForRange 因跨文件复用改为导出。

验证:9 个导出函数全部保留(无丢失),typecheck / lint / 773 测试通过。
This commit is contained in:
yml2213
2026-10-25 11:49:27 +08:00
parent b033746172
commit f10fefc303
7 changed files with 1368 additions and 1339 deletions
@@ -1,6 +1,6 @@
import { Router } from 'express'
import { setHttpCache } from '../../../middleware/http-cache.js'
import { syncAdminWorkOrderTemplateCostVersions } from '../../../services/cost-center/template-bootstrap-service.js'
import { syncAdminWorkOrderTemplateCostVersions } from '../../../services/cost-center/work-order-product-service.js'
import {
deleteAdminWorkCategory,
@@ -0,0 +1,199 @@
/**
* 自动发货(affiliate_dash)商品同步与 SKU 回填。
*/
import { query } from '../../db/client.js'
import { shiftDate } from '../../utils/finance-date-range.js'
import { nowIso } from '../../utils/time.js'
import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js'
export async function syncAdminAffiliateDashProducts() {
const now = nowIso()
const result = await query<{ inserted: boolean }>(
`
WITH auto_task AS (
SELECT
btrim(context_json->>'skuCode') AS sku_code,
btrim(context_json->>'skuName') AS sku_name,
created_at
FROM fulfillment_tasks
WHERE executor_key = 'affiliate_dash'
AND btrim(COALESCE(context_json->>'skuCode', '')) <> ''
AND btrim(COALESCE(context_json->>'skuName', '')) <> ''
),
agg AS (
SELECT sku_code, count(*)::int AS task_count, max(created_at) AS last_fulfilled_at
FROM auto_task
GROUP BY sku_code
),
ranked AS (
SELECT
sku_code,
sku_name,
row_number() OVER (
PARTITION BY sku_code
ORDER BY (sku_name <> sku_code) DESC, count(*) DESC, sku_name
) AS rn
FROM auto_task
GROUP BY sku_code, sku_name
)
INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
)
SELECT
'aff:' || ranked.sku_code,
ranked.sku_name,
'',
CASE WHEN ranked.sku_name LIKE '%个' THEN '个' ELSE '件' END,
'active',
jsonb_build_object(
'source', 'affiliate_dash_task',
'skuCode', ranked.sku_code,
'taskCount', agg.task_count,
'lastFulfilledAt', agg.last_fulfilled_at
),
$1,
$1
FROM ranked
JOIN agg ON agg.sku_code = ranked.sku_code
WHERE ranked.rn = 1
ON CONFLICT (canonical_sku) DO UPDATE
SET canonical_name = EXCLUDED.canonical_name,
unit_name = EXCLUDED.unit_name,
attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json,
updated_at = $1
RETURNING (xmax = 0) AS inserted
`,
[now],
)
const createdCount = result.rows.filter((row) => row.inserted).length
return {
createdCount,
updatedCount: result.rows.length - createdCount,
productCount: result.rows.length,
}
}
type AffiliateSkuBackfillRow = {
affiliate_item_count: number
matched_count: number
updated_count: number
from_date: string | null
to_date: string | null
}
/** 销售事实单次重算最多 400 天,长历史按窗口分批。 */
export async function rebuildFactsForRange(fromDate: string, toDate: string) {
let cursor = fromDate
let deletedCount = 0
let insertedCount = 0
let batchCount = 0
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
batchCount += 1
cursor = shiftDate(chunkEnd, 1)
}
return { fromDate, toDate, deletedCount, insertedCount, batchCount }
}
/**
* 按自动发货履约证据直接回填订单项标准商品。
*
* 自动发货只有 affiliate-dash 一个上游渠道,SKU 固定归一为 `aff:<skuCode>`,
* 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。
*/
export async function backfillAdminAffiliateDashSkus() {
const now = nowIso()
const result = await query<AffiliateSkuBackfillRow>(
`
WITH affiliate_item AS (
SELECT DISTINCT ON (ft.order_item_id)
ft.order_item_id AS id,
btrim(ft.context_json->>'skuCode') AS sku_code
FROM fulfillment_tasks ft
WHERE ft.executor_key = 'affiliate_dash'
AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> ''
ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
),
target AS (
SELECT ai.id, p.id AS canonical_product_id
FROM affiliate_item ai
JOIN product_catalog p
ON p.canonical_sku = 'aff:' || ai.sku_code
),
updated AS (
UPDATE order_items oi
SET canonical_product_id = target.canonical_product_id,
updated_at = $1
FROM target
WHERE oi.id = target.id
AND oi.canonical_product_id IS DISTINCT FROM target.canonical_product_id
RETURNING oi.id
)
SELECT
(
SELECT COUNT(DISTINCT ft.order_item_id)::int
FROM fulfillment_tasks ft
WHERE ft.executor_key = 'affiliate_dash'
) AS affiliate_item_count,
(SELECT COUNT(*)::int FROM target) AS matched_count,
(SELECT COUNT(*)::int FROM updated) AS updated_count,
(
SELECT to_char(
MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
)
FROM affiliate_item
JOIN order_items oi ON oi.id = affiliate_item.id
JOIN orders o ON o.id = oi.order_id
) AS from_date,
(
SELECT to_char(
MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
)
FROM affiliate_item
JOIN order_items oi ON oi.id = affiliate_item.id
JOIN orders o ON o.id = oi.order_id
) AS to_date
`,
[now],
)
const row = result.rows[0]
const affiliateItemCount = Number(row?.affiliate_item_count || 0)
const matchedCount = Number(row?.matched_count || 0)
const updatedCount = Number(row?.updated_count || 0)
const skippedCount = Math.max(0, matchedCount - updatedCount)
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id
AND ft.executor_key = 'affiliate_dash'
)
`,
)
const remainingCount = Number(remainingResult.rows[0]?.total || 0)
let factsRebuild: Awaited<ReturnType<typeof rebuildFactsForRange>> | null = null
if (matchedCount > 0 && row?.from_date && row?.to_date) {
factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date)
}
return {
updatedCount,
skippedCount,
remainingCount,
affiliateItemCount,
matchedCount,
factsRebuild,
}
}
@@ -6,7 +6,7 @@ import {
normalizeProviderKey,
resolveMappingNameKey,
} from './cost-center-shared.js'
import { calculateCostTotal, validateCostVersionScope } from './cost-definition-service.js'
import { calculateCostTotal, validateCostVersionScope } from './cost-resolution-service.js'
import { resolveMappingProductId } from './product-catalog-service.js'
import {
buildTemplateMappingPlan,
@@ -1,20 +1,14 @@
/**
* 成本项 / 成本版本 / 成本异常 / 成本预览 / 成本快照。
* 成本定义 CRUD:成本项、成本版本、成本异常队列。
*
* 口径(2026-10-23 重写):
* 1. 成本项用 cost_model 表达「成本怎么算」,不再有 cost_type + carrier_type;
* 2. 成本模型与履约方式正交:专属模型只对对应履约方式生效,通用模型全局生效;
* 3. 履约方式只从 order_items.fulfillment_mode 读,不允许临时推断;
* 4. 快照与异常表不再存履约载体;
* 5. 匹配不到成本不得默认 0,必须进入 cost_exceptions。
* 金额只在 cost_versions 改;异常队列只做登记与处理,不做成本计算。
* 成本解析与快照生成见 cost-resolution-service.ts。
*/
import { query, withTransaction } from '../../db/client.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import { query } from '../../db/client.js'
import type { JsonObject } from '../../types/json.js'
import { createHttpError } from '../../utils/http.js'
import { nowIso } from '../../utils/time.js'
import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js'
import { resolveSkuNameQuantity } from '../product-match/pricing.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import {
COST_EXCEPTION_STATUSES,
COST_EXCEPTION_SUBJECT_TYPES,
@@ -28,13 +22,10 @@ import {
PRODUCT_STATUSES,
assertKnownProvider,
assertOneOf,
costModelAppliesTo,
costModelScopeSql,
mapCostException,
mapCostItem,
mapCostVersion,
mapMapping,
mapProduct,
normalizeId,
normalizeIsoDate,
normalizeNonNegativeInteger,
@@ -48,19 +39,14 @@ import {
requiredRow,
text,
toNumber,
type CostExceptionRow,
type CostItemRow,
type CostLine,
type CostPreview,
type CostResolution,
type CostVersionRow,
type FulfillmentMode,
type MappingRow,
type MissingCostItem,
type OrderItemCostContext,
type ProductRow,
} from './cost-center-shared.js'
import { getProductRow } from './product-catalog-service.js'
import type { CostExceptionRow, CostItemRow, CostVersionRow } from './cost-center-shared.js'
import {
getCostItemRow,
getCostVersionById,
normalizeCostVersionScopeFields,
validateCostVersionScope,
} from './cost-resolution-service.js'
export async function listAdminCostItems(payload: JsonObject = {}) {
const page = normalizePage(payload.page)
@@ -739,644 +725,3 @@ export async function resolveAdminCostException(
)
return { exception: mapCostException(requiredRow(result.rows[0], '成本异常不存在')) }
}
export async function previewAdminCost(payload: JsonObject = {}) {
const context = await resolveCostContext(payload)
const mapping = await findMappingForContext(context)
const productId = context.canonicalProductId || mapping?.canonical_product_id || null
const product = productId ? mapProduct(await getProductRow(productId)) : null
const asOf =
normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || context.basisAt
const resolution: CostResolution = productId
? await resolveCostLines({
provider: context.provider,
shopId: context.shopId,
fulfillmentMode: context.fulfillmentMode,
canonicalProductId: productId,
workProductRuleId: context.workProductRuleId,
quantity: context.normalizedQuantity,
revenueFen: context.revenueFen,
asOf,
})
: { lines: [], missingCostItems: [], applicableCostItemCount: 0 }
const exceptions: CostPreview['exceptions'] = []
if (!productId) {
exceptions.push({
reason: 'product_mapping_missing',
message: `未找到 ${context.provider} / ${context.shopId || '全局'} / ${context.skuCode} 的标准商品映射`,
detail: {
provider: context.provider,
shopId: context.shopId,
skuCode: context.skuCode,
},
})
} else if (resolution.missingCostItems.length > 0) {
exceptions.push({
reason: 'cost_missing',
message: `以下成本项没有匹配到有效成本版本:${resolution.missingCostItems
.map((item) => `${item.costItemName}(${item.costItemCode})`)
.join('、')}`,
detail: {
fulfillmentMode: context.fulfillmentMode,
costItemIds: resolution.missingCostItems.map((item) => item.costItemId),
},
})
} else if (resolution.applicableCostItemCount === 0) {
exceptions.push({
reason: 'cost_missing',
message: `没有配置适用于「${context.fulfillmentMode}」的成本项`,
detail: { fulfillmentMode: context.fulfillmentMode },
})
}
// 人工履约的成本来自「人工 SKU 配置的消耗积分 × 平台换算率」,版本金额只是占位;
// 这里单独给出提示,并从成本行里剔掉 0 元占位,避免预览与实际扣款对不上。
const manualCost =
context.fulfillmentMode === 'manual' ? await resolveManualCostHint(context.orderItemId) : null
const lines = manualCost
? resolution.lines.filter((line) => line.costItemCode !== 'sys_manual_service')
: resolution.lines
return {
context: {
...context,
basisAt: asOf,
},
mapping: mapping ? mapMapping(mapping) : null,
product,
lines,
missingCostItems: resolution.missingCostItems.filter(
(item) => !(manualCost && item.costItemCode === 'sys_manual_service'),
),
manualCost,
totalCostFen: lines.reduce((sum, line) => sum + line.totalAmountFen, 0),
exceptions,
}
}
/** 人工履约成本提示:直接读订单项绑定的履约配置。 */
async function resolveManualCostHint(orderItemId: number | null) {
if (!orderItemId) return null
const result = await query<{
platform_key: string
credit_quantity: string
cost_mode: string
unit_price_fen: string
}>(
`
SELECT
cfg.platform_key,
cfg.credit_quantity,
cfg.cost_mode,
COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS unit_price_fen
FROM order_items oi
JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
WHERE oi.id = $1
LIMIT 1
`,
[orderItemId],
)
const row = result.rows[0]
if (!row) return null
const creditQuantity = Number(row.credit_quantity || 0)
const unitPriceFen = Number(row.unit_price_fen || 1)
return {
platformKey: row.platform_key,
creditQuantity,
unitPriceFen,
costMode: row.cost_mode === 'fixed' ? ('fixed' as const) : ('variable' as const),
costFen: creditQuantity > 0 ? creditQuantity * unitPriceFen : null,
}
}
export async function createAdminCostSnapshot(payload: JsonObject = {}, _session: AdminSession) {
const preview = await previewAdminCost(payload)
const orderItemId = preview.context.orderItemId
if (!orderItemId) {
throw createHttpError('生成成本快照必须提供订单项 ID', {
statusCode: 400,
errorCode: 'cost_snapshot_order_item_required',
})
}
const now = nowIso()
const stage = payload.snapshotStage === 'actual' ? 'actual' : 'estimated'
const snapshots: Array<Record<string, unknown>> = []
let exceptionCreatedCount = 0
let supersededCount = 0
await withTransaction(async (client) => {
for (const line of preview.lines) {
const currentResult = await client.query<{
id: number
snapshot_stage: string
total_amount_fen: number
}>(
`
SELECT id, snapshot_stage, total_amount_fen
FROM cost_snapshots
WHERE subject_type = 'order_item'
AND subject_id = $1
AND cost_item_id = $2
AND superseded_at IS NULL
FOR UPDATE
`,
[orderItemId, line.costItemId],
)
const current = currentResult.rows[0]
if (current) {
// 预估快照一旦生成就代表支付时点的成本,不因后续改价重写;
// 只有实际成本允许替代当前有效行。
if (stage === 'estimated') continue
if (
current.snapshot_stage === 'actual' &&
Number(current.total_amount_fen) === line.totalAmountFen
) {
continue
}
await client.query(
`UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL`,
[now, current.id],
)
supersededCount += 1
}
const result = await client.query<{ [column: string]: unknown }>(
`
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
) VALUES (
'order_item', $1, $2, $1, $3, $4, $5, $6,
$7, $8, $9, $10, 'resolved', $11::jsonb, $12, $13, $13
)
ON CONFLICT (subject_type, subject_id, cost_item_id)
WHERE superseded_at IS NULL
DO NOTHING
RETURNING *
`,
[
orderItemId,
preview.context.orderId,
preview.context.workOrderId,
line.costItemId,
line.costVersionId,
preview.product?.id || null,
line.quantity,
line.unitAmountFen,
line.totalAmountFen,
stage,
JSON.stringify({
skuCode: preview.context.skuCode,
skuName: preview.context.skuName,
provider: preview.context.provider,
shopId: preview.context.shopId,
normalizedQuantity: preview.context.normalizedQuantity,
fulfillmentMode: preview.context.fulfillmentMode,
basisAt: preview.context.basisAt,
}),
preview.context.basisAt,
now,
],
)
if (result.rows[0]) snapshots.push(result.rows[0])
}
for (const exception of preview.exceptions) {
const result = await client.query(
`
INSERT INTO cost_exceptions (
subject_type, subject_id, order_item_id, work_order_id,
reason, status, detail_json, created_at
) VALUES ('order_item', $1, $1, $2, $3, 'open', $4::jsonb, $5)
ON CONFLICT (subject_type, subject_id, reason) DO NOTHING
`,
[
orderItemId,
preview.context.workOrderId,
exception.reason,
JSON.stringify({
...exception.detail,
message: exception.message,
skuCode: preview.context.skuCode,
}),
now,
],
)
exceptionCreatedCount += result.rowCount || 0
}
})
return {
preview,
snapshots,
createdCount: snapshots.length,
supersededCount,
exceptionCreatedCount,
}
}
async function resolveCostContext(payload: JsonObject): Promise<OrderItemCostContext> {
const orderItemId = normalizeOptionalId(payload.orderItemId ?? payload.order_item_id)
if (orderItemId) {
const result = await query<{
order_item_id: number
order_id: number
provider: string
shop_id: string
sku_code: string
sku_name: string
normalized_sku_name: string | null
normalized_sku_series: string | null
quantity: number
normalized_quantity: number | null
canonical_product_id: number | null
revenue_fen: number
basis_at: string
work_order_id: number | null
work_product_rule_id: number | null
fulfillment_mode: string | null
}>(
`
SELECT
oi.id AS order_item_id,
oi.order_id,
o.provider,
o.shop_id,
oi.sku_code,
oi.sku_name,
oi.normalized_sku_name,
oi.normalized_sku_series,
oi.quantity,
oi.normalized_quantity,
oi.canonical_product_id,
oi.revenue_amount_fen AS revenue_fen,
COALESCE(o.paid_at, o.created_at) AS basis_at,
wo.id AS work_order_id,
wo.product_rule_id AS work_product_rule_id,
oi.fulfillment_mode
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT id, product_rule_id
FROM work_orders
WHERE order_item_id = oi.id
ORDER BY id DESC
LIMIT 1
) wo ON TRUE
WHERE oi.id = $1
LIMIT 1
`,
[orderItemId],
)
const row = result.rows[0]
if (!row) {
throw createHttpError('订单项不存在', {
statusCode: 404,
errorCode: 'cost_order_item_not_found',
})
}
return {
orderItemId: Number(row.order_item_id),
orderId: Number(row.order_id),
provider: row.provider,
shopId: row.shop_id,
skuCode: row.sku_code,
skuName: row.sku_name,
normalizedSkuName:
row.normalized_sku_name || normalizeMatchText(row.sku_name || row.sku_code),
normalizedSkuSeries:
row.normalized_sku_series || normalizeSkuSeriesEquivalence(row.sku_name || row.sku_code),
quantity: Math.max(1, Number(row.quantity) || 1),
normalizedQuantity:
row.normalized_quantity && Number(row.normalized_quantity) > 0
? Number(row.normalized_quantity)
: Math.max(1, resolveSkuNameQuantity(row.sku_name || row.sku_code)),
canonicalProductId: row.canonical_product_id ? Number(row.canonical_product_id) : null,
revenueFen: Math.max(0, Number(row.revenue_fen) || 0),
basisAt: new Date(row.basis_at).toISOString(),
workOrderId: row.work_order_id ? Number(row.work_order_id) : null,
workProductRuleId: row.work_product_rule_id ? Number(row.work_product_rule_id) : null,
fulfillmentMode: resolveFulfillmentMode(row.fulfillment_mode),
}
}
const provider = normalizeRequiredText(payload.provider, '来源', 64)
const skuCode = normalizeRequiredText(payload.skuCode ?? payload.sku_code, '平台 SKU', 120)
const quantity = Math.max(1, Math.trunc(Number(payload.quantity) || 1))
const skuName = normalizeOptionalText(payload.skuName ?? payload.sku_name, 200)
const fulfillmentMode = resolveFulfillmentMode(
payload.fulfillmentMode ?? payload.fulfillment_mode,
)
return {
orderItemId: 0,
orderId: 0,
provider,
shopId: normalizeOptionalText(payload.shopId ?? payload.shop_id, 120),
skuCode,
skuName,
normalizedSkuName: normalizeMatchText(skuName || skuCode),
normalizedSkuSeries: normalizeSkuSeriesEquivalence(skuName || skuCode),
quantity,
normalizedQuantity: quantity,
canonicalProductId: normalizeOptionalId(
payload.canonicalProductId ?? payload.canonical_product_id,
),
revenueFen: Math.max(0, Math.trunc(Number(payload.revenueFen ?? payload.revenue_fen) || 0)),
basisAt:
normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || nowIso(),
workOrderId: null,
workProductRuleId: normalizeOptionalId(
payload.workProductRuleId ?? payload.work_product_rule_id,
),
fulfillmentMode,
}
}
/**
* 履约方式只读 order_items.fulfillment_mode。
* 订单项还没判定时按人工处理,不允许用 provider 或核销码兜底。
*/
function resolveFulfillmentMode(value: unknown): FulfillmentMode {
const normalized = text(value)
if (normalized && FULFILLMENT_MODES.has(normalized)) return normalized as FulfillmentMode
return 'manual'
}
async function findMappingForContext(context: OrderItemCostContext) {
const result = await query<MappingRow>(
`
SELECT m.*, p.canonical_sku, p.canonical_name
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id AND p.status = 'active'
WHERE m.provider = $1
AND m.enabled = TRUE
AND (m.shop_id = '' OR m.shop_id = $3)
AND (
(m.match_type = 'sku_code' AND m.platform_sku_code = $2)
OR (
m.match_type = 'sku_code_name'
AND m.platform_sku_code = $2
AND m.platform_sku_name_key = $4
)
OR (
m.match_type = 'sku_code_series'
AND m.platform_sku_code = $2
AND m.platform_sku_name_key = $5
)
OR (m.match_type = 'sku_name_series' AND m.platform_sku_name_key = $5)
)
ORDER BY
CASE
WHEN m.match_type = 'sku_code' AND m.shop_id = $3 THEN 1
WHEN m.match_type = 'sku_code' AND m.shop_id = '' THEN 2
WHEN m.match_type = 'sku_code_name' AND m.shop_id = $3 THEN 3
WHEN m.match_type = 'sku_code_name' AND m.shop_id = '' THEN 4
WHEN m.match_type = 'sku_code_series' AND m.shop_id = $3 THEN 5
WHEN m.match_type = 'sku_code_series' AND m.shop_id = '' THEN 6
WHEN m.match_type = 'sku_name_series' AND m.shop_id = $3 THEN 7
ELSE 8
END,
m.id DESC
LIMIT 1
`,
[
context.provider,
context.skuCode,
context.shopId,
context.normalizedSkuName,
context.normalizedSkuSeries,
],
)
return result.rows[0] || null
}
async function resolveCostLines(input: {
provider: string
shopId: string
fulfillmentMode: FulfillmentMode
canonicalProductId: number
workProductRuleId: number | null
quantity: number
revenueFen: number
asOf?: string
}): Promise<CostResolution> {
const result = await query<
CostItemRow & {
version_id: number | null
version_scope: string | null
version_amount_fen: number | null
version_rate_bps: number | null
}
>(
`
SELECT
ci.*,
v.id AS version_id,
v.scope_type AS version_scope,
v.amount_fen AS version_amount_fen,
v.rate_bps AS version_rate_bps
FROM cost_items ci
LEFT JOIN LATERAL (
SELECT v.*
FROM cost_versions v
WHERE v.cost_item_id = ci.id
AND v.status = 'active'
AND v.effective_from <= $1::timestamptz
AND (v.effective_to IS NULL OR v.effective_to > $1::timestamptz)
AND (
(v.scope_type = 'rule' AND $5::bigint IS NOT NULL AND v.work_product_rule_id = $5)
OR (v.scope_type = 'product' AND v.canonical_product_id = $4)
OR (v.scope_type = 'shop' AND v.provider = $2 AND v.shop_id = $3)
OR (v.scope_type = 'provider' AND v.provider = $2)
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 ci.status = 'active'
AND ${costModelScopeSql('$6')}
ORDER BY ci.id
`,
[
input.asOf || nowIso(),
input.provider,
input.shopId,
input.canonicalProductId,
input.workProductRuleId,
input.fulfillmentMode,
],
)
const lines: CostLine[] = []
const missingCostItems: MissingCostItem[] = []
for (const row of result.rows) {
if (!row.version_id) {
missingCostItems.push({
costItemId: Number(row.id),
costItemCode: row.code,
costItemName: row.name,
costModel: row.cost_model as MissingCostItem['costModel'],
})
continue
}
const pricingMode = row.pricing_mode
const amountFen = Number(row.version_amount_fen || 0)
const rateBps = Number(row.version_rate_bps || 0)
const quantity = Math.max(0, Number(input.quantity) || 0)
const unitAmountFen = amountFen
const totalAmountFen = calculateCostTotal({
pricingMode,
amountFen,
rateBps,
quantity,
revenueFen: input.revenueFen,
})
lines.push({
costItemId: Number(row.id),
costItemCode: row.code,
costItemName: row.name,
costModel: row.cost_model as CostLine['costModel'],
pricingMode,
costVersionId: Number(row.version_id),
scopeType: String(row.version_scope || ''),
amountFen,
rateBps,
quantity,
unitAmountFen,
totalAmountFen,
})
}
return {
lines,
missingCostItems,
applicableCostItemCount: result.rows.length,
}
}
export function calculateCostTotal(input: {
pricingMode: string
amountFen: number
rateBps: number
quantity: number
revenueFen: number
}) {
const amountFen = Math.max(0, Math.trunc(Number(input.amountFen) || 0))
const rateBps = Math.max(0, Math.min(10000, Math.trunc(Number(input.rateBps) || 0)))
const quantity = Math.max(0, Number(input.quantity) || 0)
const revenueFen = Math.max(0, Math.trunc(Number(input.revenueFen) || 0))
let total = amountFen
if (input.pricingMode === 'per_item') {
total = Math.round(amountFen * quantity)
} else if (input.pricingMode === 'percent_of_revenue') {
total = Math.round((revenueFen * rateBps) / 10000)
} else if (input.pricingMode === 'per_order') {
total = amountFen
}
return Math.max(0, total)
}
export function validateCostVersionScope(
scopeType: string,
input: {
provider: string
shopId: string
canonicalProductId: number | null
workProductRuleId: number | null
},
) {
if (scopeType === 'provider' && !input.provider) {
throw createHttpError('来源范围必须填写来源', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
if (scopeType === 'shop' && (!input.provider || !input.shopId)) {
throw createHttpError('店铺范围必须填写来源和店铺', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
if (scopeType === 'product' && !input.canonicalProductId) {
throw createHttpError('商品范围必须选择标准商品', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
if (scopeType === 'rule' && !input.workProductRuleId) {
throw createHttpError('规则范围必须选择接单规则', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
}
function normalizeCostVersionScopeFields(
scopeType: string,
input: {
provider: string
shopId: string
canonicalProductId: number | null
workProductRuleId: number | null
},
) {
const empty = {
provider: '',
shopId: '',
canonicalProductId: null as number | null,
workProductRuleId: null as number | null,
}
if (scopeType === 'provider') {
return { ...empty, provider: input.provider }
}
if (scopeType === 'shop') {
return { ...empty, provider: input.provider, shopId: input.shopId }
}
if (scopeType === 'product') {
return { ...empty, canonicalProductId: input.canonicalProductId }
}
if (scopeType === 'rule') {
return { ...empty, workProductRuleId: input.workProductRuleId }
}
return empty
}
async function getCostItemRow(id: number) {
const result = await query<CostItemRow>('SELECT * FROM cost_items WHERE id = $1 LIMIT 1', [id])
return requiredRow(result.rows[0], '成本项不存在')
}
async function getCostVersionById(id: number) {
const result = await query<CostVersionRow>(
`
SELECT
v.*,
ci.code AS cost_item_code,
ci.name AS cost_item_name,
ci.cost_model AS cost_item_cost_model,
ci.pricing_mode,
p.canonical_sku,
p.canonical_name
FROM cost_versions v
JOIN cost_items ci ON ci.id = v.cost_item_id
LEFT JOIN product_catalog p ON p.id = v.canonical_product_id
WHERE v.id = $1
LIMIT 1
`,
[id],
)
return mapCostVersion(requiredRow(result.rows[0], '成本版本不存在'))
}
@@ -0,0 +1,680 @@
/**
* 成本解析与快照:预览、解析上下文、按成本版本计算并落快照。
*
* 口径:金额只来自 cost_versions;匹配不到成本进异常队列,不默认 0;
* 历史快照永不回写(superseded_at 只标记被实际成本替代)。
* 成本定义 CRUD 见 cost-definition-service.ts。
*/
import { query, withTransaction } from '../../db/client.js'
import type { JsonObject } from '../../types/json.js'
import { createHttpError } from '../../utils/http.js'
import { getProductRow } from './product-catalog-service.js'
import { nowIso } from '../../utils/time.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js'
import { resolveSkuNameQuantity } from '../product-match/pricing.js'
import {
FULFILLMENT_MODES,
costModelScopeSql,
mapCostVersion,
mapMapping,
mapProduct,
normalizeIsoDate,
normalizeOptionalId,
normalizeOptionalText,
normalizeRequiredText,
requiredRow,
text,
} from './cost-center-shared.js'
import type {
CostItemRow,
CostLine,
CostPreview,
CostResolution,
CostVersionRow,
FulfillmentMode,
MappingRow,
MissingCostItem,
OrderItemCostContext,
} from './cost-center-shared.js'
export async function previewAdminCost(payload: JsonObject = {}) {
const context = await resolveCostContext(payload)
const mapping = await findMappingForContext(context)
const productId = context.canonicalProductId || mapping?.canonical_product_id || null
const product = productId ? mapProduct(await getProductRow(productId)) : null
const asOf =
normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || context.basisAt
const resolution: CostResolution = productId
? await resolveCostLines({
provider: context.provider,
shopId: context.shopId,
fulfillmentMode: context.fulfillmentMode,
canonicalProductId: productId,
workProductRuleId: context.workProductRuleId,
quantity: context.normalizedQuantity,
revenueFen: context.revenueFen,
asOf,
})
: { lines: [], missingCostItems: [], applicableCostItemCount: 0 }
const exceptions: CostPreview['exceptions'] = []
if (!productId) {
exceptions.push({
reason: 'product_mapping_missing',
message: `未找到 ${context.provider} / ${context.shopId || '全局'} / ${context.skuCode} 的标准商品映射`,
detail: {
provider: context.provider,
shopId: context.shopId,
skuCode: context.skuCode,
},
})
} else if (resolution.missingCostItems.length > 0) {
exceptions.push({
reason: 'cost_missing',
message: `以下成本项没有匹配到有效成本版本:${resolution.missingCostItems
.map((item) => `${item.costItemName}(${item.costItemCode})`)
.join('、')}`,
detail: {
fulfillmentMode: context.fulfillmentMode,
costItemIds: resolution.missingCostItems.map((item) => item.costItemId),
},
})
} else if (resolution.applicableCostItemCount === 0) {
exceptions.push({
reason: 'cost_missing',
message: `没有配置适用于「${context.fulfillmentMode}」的成本项`,
detail: { fulfillmentMode: context.fulfillmentMode },
})
}
// 人工履约的成本来自「人工 SKU 配置的消耗积分 × 平台换算率」,版本金额只是占位;
// 这里单独给出提示,并从成本行里剔掉 0 元占位,避免预览与实际扣款对不上。
const manualCost =
context.fulfillmentMode === 'manual' ? await resolveManualCostHint(context.orderItemId) : null
const lines = manualCost
? resolution.lines.filter((line) => line.costItemCode !== 'sys_manual_service')
: resolution.lines
return {
context: {
...context,
basisAt: asOf,
},
mapping: mapping ? mapMapping(mapping) : null,
product,
lines,
missingCostItems: resolution.missingCostItems.filter(
(item) => !(manualCost && item.costItemCode === 'sys_manual_service'),
),
manualCost,
totalCostFen: lines.reduce((sum, line) => sum + line.totalAmountFen, 0),
exceptions,
}
}
/** 人工履约成本提示:直接读订单项绑定的履约配置。 */
async function resolveManualCostHint(orderItemId: number | null) {
if (!orderItemId) return null
const result = await query<{
platform_key: string
credit_quantity: string
cost_mode: string
unit_price_fen: string
}>(
`
SELECT
cfg.platform_key,
cfg.credit_quantity,
cfg.cost_mode,
COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS unit_price_fen
FROM order_items oi
JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
WHERE oi.id = $1
LIMIT 1
`,
[orderItemId],
)
const row = result.rows[0]
if (!row) return null
const creditQuantity = Number(row.credit_quantity || 0)
const unitPriceFen = Number(row.unit_price_fen || 1)
return {
platformKey: row.platform_key,
creditQuantity,
unitPriceFen,
costMode: row.cost_mode === 'fixed' ? ('fixed' as const) : ('variable' as const),
costFen: creditQuantity > 0 ? creditQuantity * unitPriceFen : null,
}
}
export async function createAdminCostSnapshot(payload: JsonObject = {}, _session: AdminSession) {
const preview = await previewAdminCost(payload)
const orderItemId = preview.context.orderItemId
if (!orderItemId) {
throw createHttpError('生成成本快照必须提供订单项 ID', {
statusCode: 400,
errorCode: 'cost_snapshot_order_item_required',
})
}
const now = nowIso()
const stage = payload.snapshotStage === 'actual' ? 'actual' : 'estimated'
const snapshots: Array<Record<string, unknown>> = []
let exceptionCreatedCount = 0
let supersededCount = 0
await withTransaction(async (client) => {
for (const line of preview.lines) {
const currentResult = await client.query<{
id: number
snapshot_stage: string
total_amount_fen: number
}>(
`
SELECT id, snapshot_stage, total_amount_fen
FROM cost_snapshots
WHERE subject_type = 'order_item'
AND subject_id = $1
AND cost_item_id = $2
AND superseded_at IS NULL
FOR UPDATE
`,
[orderItemId, line.costItemId],
)
const current = currentResult.rows[0]
if (current) {
// 预估快照一旦生成就代表支付时点的成本,不因后续改价重写;
// 只有实际成本允许替代当前有效行。
if (stage === 'estimated') continue
if (
current.snapshot_stage === 'actual' &&
Number(current.total_amount_fen) === line.totalAmountFen
) {
continue
}
await client.query(
`UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL`,
[now, current.id],
)
supersededCount += 1
}
const result = await client.query<{ [column: string]: unknown }>(
`
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
) VALUES (
'order_item', $1, $2, $1, $3, $4, $5, $6,
$7, $8, $9, $10, 'resolved', $11::jsonb, $12, $13, $13
)
ON CONFLICT (subject_type, subject_id, cost_item_id)
WHERE superseded_at IS NULL
DO NOTHING
RETURNING *
`,
[
orderItemId,
preview.context.orderId,
preview.context.workOrderId,
line.costItemId,
line.costVersionId,
preview.product?.id || null,
line.quantity,
line.unitAmountFen,
line.totalAmountFen,
stage,
JSON.stringify({
skuCode: preview.context.skuCode,
skuName: preview.context.skuName,
provider: preview.context.provider,
shopId: preview.context.shopId,
normalizedQuantity: preview.context.normalizedQuantity,
fulfillmentMode: preview.context.fulfillmentMode,
basisAt: preview.context.basisAt,
}),
preview.context.basisAt,
now,
],
)
if (result.rows[0]) snapshots.push(result.rows[0])
}
for (const exception of preview.exceptions) {
const result = await client.query(
`
INSERT INTO cost_exceptions (
subject_type, subject_id, order_item_id, work_order_id,
reason, status, detail_json, created_at
) VALUES ('order_item', $1, $1, $2, $3, 'open', $4::jsonb, $5)
ON CONFLICT (subject_type, subject_id, reason) DO NOTHING
`,
[
orderItemId,
preview.context.workOrderId,
exception.reason,
JSON.stringify({
...exception.detail,
message: exception.message,
skuCode: preview.context.skuCode,
}),
now,
],
)
exceptionCreatedCount += result.rowCount || 0
}
})
return {
preview,
snapshots,
createdCount: snapshots.length,
supersededCount,
exceptionCreatedCount,
}
}
async function resolveCostContext(payload: JsonObject): Promise<OrderItemCostContext> {
const orderItemId = normalizeOptionalId(payload.orderItemId ?? payload.order_item_id)
if (orderItemId) {
const result = await query<{
order_item_id: number
order_id: number
provider: string
shop_id: string
sku_code: string
sku_name: string
normalized_sku_name: string | null
normalized_sku_series: string | null
quantity: number
normalized_quantity: number | null
canonical_product_id: number | null
revenue_fen: number
basis_at: string
work_order_id: number | null
work_product_rule_id: number | null
fulfillment_mode: string | null
}>(
`
SELECT
oi.id AS order_item_id,
oi.order_id,
o.provider,
o.shop_id,
oi.sku_code,
oi.sku_name,
oi.normalized_sku_name,
oi.normalized_sku_series,
oi.quantity,
oi.normalized_quantity,
oi.canonical_product_id,
oi.revenue_amount_fen AS revenue_fen,
COALESCE(o.paid_at, o.created_at) AS basis_at,
wo.id AS work_order_id,
wo.product_rule_id AS work_product_rule_id,
oi.fulfillment_mode
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT id, product_rule_id
FROM work_orders
WHERE order_item_id = oi.id
ORDER BY id DESC
LIMIT 1
) wo ON TRUE
WHERE oi.id = $1
LIMIT 1
`,
[orderItemId],
)
const row = result.rows[0]
if (!row) {
throw createHttpError('订单项不存在', {
statusCode: 404,
errorCode: 'cost_order_item_not_found',
})
}
return {
orderItemId: Number(row.order_item_id),
orderId: Number(row.order_id),
provider: row.provider,
shopId: row.shop_id,
skuCode: row.sku_code,
skuName: row.sku_name,
normalizedSkuName:
row.normalized_sku_name || normalizeMatchText(row.sku_name || row.sku_code),
normalizedSkuSeries:
row.normalized_sku_series || normalizeSkuSeriesEquivalence(row.sku_name || row.sku_code),
quantity: Math.max(1, Number(row.quantity) || 1),
normalizedQuantity:
row.normalized_quantity && Number(row.normalized_quantity) > 0
? Number(row.normalized_quantity)
: Math.max(1, resolveSkuNameQuantity(row.sku_name || row.sku_code)),
canonicalProductId: row.canonical_product_id ? Number(row.canonical_product_id) : null,
revenueFen: Math.max(0, Number(row.revenue_fen) || 0),
basisAt: new Date(row.basis_at).toISOString(),
workOrderId: row.work_order_id ? Number(row.work_order_id) : null,
workProductRuleId: row.work_product_rule_id ? Number(row.work_product_rule_id) : null,
fulfillmentMode: resolveFulfillmentMode(row.fulfillment_mode),
}
}
const provider = normalizeRequiredText(payload.provider, '来源', 64)
const skuCode = normalizeRequiredText(payload.skuCode ?? payload.sku_code, '平台 SKU', 120)
const quantity = Math.max(1, Math.trunc(Number(payload.quantity) || 1))
const skuName = normalizeOptionalText(payload.skuName ?? payload.sku_name, 200)
const fulfillmentMode = resolveFulfillmentMode(
payload.fulfillmentMode ?? payload.fulfillment_mode,
)
return {
orderItemId: 0,
orderId: 0,
provider,
shopId: normalizeOptionalText(payload.shopId ?? payload.shop_id, 120),
skuCode,
skuName,
normalizedSkuName: normalizeMatchText(skuName || skuCode),
normalizedSkuSeries: normalizeSkuSeriesEquivalence(skuName || skuCode),
quantity,
normalizedQuantity: quantity,
canonicalProductId: normalizeOptionalId(
payload.canonicalProductId ?? payload.canonical_product_id,
),
revenueFen: Math.max(0, Math.trunc(Number(payload.revenueFen ?? payload.revenue_fen) || 0)),
basisAt:
normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || nowIso(),
workOrderId: null,
workProductRuleId: normalizeOptionalId(
payload.workProductRuleId ?? payload.work_product_rule_id,
),
fulfillmentMode,
}
}
/**
* 履约方式只读 order_items.fulfillment_mode。
* 订单项还没判定时按人工处理,不允许用 provider 或核销码兜底。
*/
function resolveFulfillmentMode(value: unknown): FulfillmentMode {
const normalized = text(value)
if (normalized && FULFILLMENT_MODES.has(normalized)) return normalized as FulfillmentMode
return 'manual'
}
async function findMappingForContext(context: OrderItemCostContext) {
const result = await query<MappingRow>(
`
SELECT m.*, p.canonical_sku, p.canonical_name
FROM channel_product_mappings m
JOIN product_catalog p ON p.id = m.canonical_product_id AND p.status = 'active'
WHERE m.provider = $1
AND m.enabled = TRUE
AND (m.shop_id = '' OR m.shop_id = $3)
AND (
(m.match_type = 'sku_code' AND m.platform_sku_code = $2)
OR (
m.match_type = 'sku_code_name'
AND m.platform_sku_code = $2
AND m.platform_sku_name_key = $4
)
OR (
m.match_type = 'sku_code_series'
AND m.platform_sku_code = $2
AND m.platform_sku_name_key = $5
)
OR (m.match_type = 'sku_name_series' AND m.platform_sku_name_key = $5)
)
ORDER BY
CASE
WHEN m.match_type = 'sku_code' AND m.shop_id = $3 THEN 1
WHEN m.match_type = 'sku_code' AND m.shop_id = '' THEN 2
WHEN m.match_type = 'sku_code_name' AND m.shop_id = $3 THEN 3
WHEN m.match_type = 'sku_code_name' AND m.shop_id = '' THEN 4
WHEN m.match_type = 'sku_code_series' AND m.shop_id = $3 THEN 5
WHEN m.match_type = 'sku_code_series' AND m.shop_id = '' THEN 6
WHEN m.match_type = 'sku_name_series' AND m.shop_id = $3 THEN 7
ELSE 8
END,
m.id DESC
LIMIT 1
`,
[
context.provider,
context.skuCode,
context.shopId,
context.normalizedSkuName,
context.normalizedSkuSeries,
],
)
return result.rows[0] || null
}
async function resolveCostLines(input: {
provider: string
shopId: string
fulfillmentMode: FulfillmentMode
canonicalProductId: number
workProductRuleId: number | null
quantity: number
revenueFen: number
asOf?: string
}): Promise<CostResolution> {
const result = await query<
CostItemRow & {
version_id: number | null
version_scope: string | null
version_amount_fen: number | null
version_rate_bps: number | null
}
>(
`
SELECT
ci.*,
v.id AS version_id,
v.scope_type AS version_scope,
v.amount_fen AS version_amount_fen,
v.rate_bps AS version_rate_bps
FROM cost_items ci
LEFT JOIN LATERAL (
SELECT v.*
FROM cost_versions v
WHERE v.cost_item_id = ci.id
AND v.status = 'active'
AND v.effective_from <= $1::timestamptz
AND (v.effective_to IS NULL OR v.effective_to > $1::timestamptz)
AND (
(v.scope_type = 'rule' AND $5::bigint IS NOT NULL AND v.work_product_rule_id = $5)
OR (v.scope_type = 'product' AND v.canonical_product_id = $4)
OR (v.scope_type = 'shop' AND v.provider = $2 AND v.shop_id = $3)
OR (v.scope_type = 'provider' AND v.provider = $2)
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 ci.status = 'active'
AND ${costModelScopeSql('$6')}
ORDER BY ci.id
`,
[
input.asOf || nowIso(),
input.provider,
input.shopId,
input.canonicalProductId,
input.workProductRuleId,
input.fulfillmentMode,
],
)
const lines: CostLine[] = []
const missingCostItems: MissingCostItem[] = []
for (const row of result.rows) {
if (!row.version_id) {
missingCostItems.push({
costItemId: Number(row.id),
costItemCode: row.code,
costItemName: row.name,
costModel: row.cost_model as MissingCostItem['costModel'],
})
continue
}
const pricingMode = row.pricing_mode
const amountFen = Number(row.version_amount_fen || 0)
const rateBps = Number(row.version_rate_bps || 0)
const quantity = Math.max(0, Number(input.quantity) || 0)
const unitAmountFen = amountFen
const totalAmountFen = calculateCostTotal({
pricingMode,
amountFen,
rateBps,
quantity,
revenueFen: input.revenueFen,
})
lines.push({
costItemId: Number(row.id),
costItemCode: row.code,
costItemName: row.name,
costModel: row.cost_model as CostLine['costModel'],
pricingMode,
costVersionId: Number(row.version_id),
scopeType: String(row.version_scope || ''),
amountFen,
rateBps,
quantity,
unitAmountFen,
totalAmountFen,
})
}
return {
lines,
missingCostItems,
applicableCostItemCount: result.rows.length,
}
}
export function calculateCostTotal(input: {
pricingMode: string
amountFen: number
rateBps: number
quantity: number
revenueFen: number
}) {
const amountFen = Math.max(0, Math.trunc(Number(input.amountFen) || 0))
const rateBps = Math.max(0, Math.min(10000, Math.trunc(Number(input.rateBps) || 0)))
const quantity = Math.max(0, Number(input.quantity) || 0)
const revenueFen = Math.max(0, Math.trunc(Number(input.revenueFen) || 0))
let total = amountFen
if (input.pricingMode === 'per_item') {
total = Math.round(amountFen * quantity)
} else if (input.pricingMode === 'percent_of_revenue') {
total = Math.round((revenueFen * rateBps) / 10000)
} else if (input.pricingMode === 'per_order') {
total = amountFen
}
return Math.max(0, total)
}
export function validateCostVersionScope(
scopeType: string,
input: {
provider: string
shopId: string
canonicalProductId: number | null
workProductRuleId: number | null
},
) {
if (scopeType === 'provider' && !input.provider) {
throw createHttpError('来源范围必须填写来源', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
if (scopeType === 'shop' && (!input.provider || !input.shopId)) {
throw createHttpError('店铺范围必须填写来源和店铺', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
if (scopeType === 'product' && !input.canonicalProductId) {
throw createHttpError('商品范围必须选择标准商品', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
if (scopeType === 'rule' && !input.workProductRuleId) {
throw createHttpError('规则范围必须选择接单规则', {
statusCode: 400,
errorCode: 'cost_version_scope_invalid',
})
}
}
export function normalizeCostVersionScopeFields(
scopeType: string,
input: {
provider: string
shopId: string
canonicalProductId: number | null
workProductRuleId: number | null
},
) {
const empty = {
provider: '',
shopId: '',
canonicalProductId: null as number | null,
workProductRuleId: null as number | null,
}
if (scopeType === 'provider') {
return { ...empty, provider: input.provider }
}
if (scopeType === 'shop') {
return { ...empty, provider: input.provider, shopId: input.shopId }
}
if (scopeType === 'product') {
return { ...empty, canonicalProductId: input.canonicalProductId }
}
if (scopeType === 'rule') {
return { ...empty, workProductRuleId: input.workProductRuleId }
}
return empty
}
export async function getCostItemRow(id: number) {
const result = await query<CostItemRow>('SELECT * FROM cost_items WHERE id = $1 LIMIT 1', [id])
return requiredRow(result.rows[0], '成本项不存在')
}
export async function getCostVersionById(id: number) {
const result = await query<CostVersionRow>(
`
SELECT
v.*,
ci.code AS cost_item_code,
ci.name AS cost_item_name,
ci.cost_model AS cost_item_cost_model,
ci.pricing_mode,
p.canonical_sku,
p.canonical_name
FROM cost_versions v
JOIN cost_items ci ON ci.id = v.cost_item_id
LEFT JOIN product_catalog p ON p.id = v.canonical_product_id
WHERE v.id = $1
LIMIT 1
`,
[id],
)
return mapCostVersion(requiredRow(result.rows[0], '成本版本不存在'))
}
@@ -1,34 +1,23 @@
/**
* 接单模板引导、自动发货商品同步与工单证据回填。
* 接单模板引导:预览与执行、映射计划构建、模板商品编码。
*
* 口径(2026-10-23 重写):
* 1. 商品表不存履约载体,来源只由 canonical_sku 前缀表达:
* aff: 自动发货 / tpl: 接单模板 / wo: 工单证据;
* 2. 自动发货 SKU 固定归一为 aff:<skuCode>,不写渠道映射;
* 3. 匹配方式与名称键都参与映射唯一键,91 同店铺复用 sku_code 才能消歧。
* 关联:
* - affiliate-product-service.ts 自动发货商品同步与 SKU 回填
* - work-order-product-service.ts 接单平台商品回填与模板发单金额同步
*/
import { query, withTransaction } from '../../db/client.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import type { JsonObject } from '../../types/json.js'
import { createHttpError } from '../../utils/http.js'
import { shiftDate } from '../../utils/finance-date-range.js'
import { nowIso } from '../../utils/time.js'
import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js'
import {
MAPPING_MATCH_TYPES,
WORK_ORDER_TEMPLATE_COST_BASELINE,
candidateKey,
mappingKey,
mappingMatchTypeLabel,
normalizeMappingMatchType,
resolveMappingNameKey,
type TemplateMappingMatchType,
} from './cost-center-shared.js'
import { candidateKey, mappingKey, mappingMatchTypeLabel } from './cost-center-shared.js'
import type { TemplateMappingMatchType } from './cost-center-shared.js'
import {
backfillAdminNormalizedQuantities,
backfillAdminProductMappings,
} from './product-catalog-service.js'
import { syncAdminWorkOrderTemplateCostVersions } from './work-order-product-service.js'
export async function previewAdminTemplateProductBootstrap() {
const data = await loadTemplateProductBootstrapData()
@@ -225,657 +214,6 @@ export async function applyAdminTemplateProductBootstrap(
* 平台 SKU 归一,不能作为自动发货依据。SKU 固定为 `aff:<skuCode>`,
* 重复同步只更新名称/证据,不会重复建商品。
*/
export async function syncAdminAffiliateDashProducts() {
const now = nowIso()
const result = await query<{ inserted: boolean }>(
`
WITH auto_task AS (
SELECT
btrim(context_json->>'skuCode') AS sku_code,
btrim(context_json->>'skuName') AS sku_name,
created_at
FROM fulfillment_tasks
WHERE executor_key = 'affiliate_dash'
AND btrim(COALESCE(context_json->>'skuCode', '')) <> ''
AND btrim(COALESCE(context_json->>'skuName', '')) <> ''
),
agg AS (
SELECT sku_code, count(*)::int AS task_count, max(created_at) AS last_fulfilled_at
FROM auto_task
GROUP BY sku_code
),
ranked AS (
SELECT
sku_code,
sku_name,
row_number() OVER (
PARTITION BY sku_code
ORDER BY (sku_name <> sku_code) DESC, count(*) DESC, sku_name
) AS rn
FROM auto_task
GROUP BY sku_code, sku_name
)
INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
)
SELECT
'aff:' || ranked.sku_code,
ranked.sku_name,
'',
CASE WHEN ranked.sku_name LIKE '%个' THEN '个' ELSE '件' END,
'active',
jsonb_build_object(
'source', 'affiliate_dash_task',
'skuCode', ranked.sku_code,
'taskCount', agg.task_count,
'lastFulfilledAt', agg.last_fulfilled_at
),
$1,
$1
FROM ranked
JOIN agg ON agg.sku_code = ranked.sku_code
WHERE ranked.rn = 1
ON CONFLICT (canonical_sku) DO UPDATE
SET canonical_name = EXCLUDED.canonical_name,
unit_name = EXCLUDED.unit_name,
attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json,
updated_at = $1
RETURNING (xmax = 0) AS inserted
`,
[now],
)
const createdCount = result.rows.filter((row) => row.inserted).length
return {
createdCount,
updatedCount: result.rows.length - createdCount,
productCount: result.rows.length,
}
}
type AffiliateSkuBackfillRow = {
affiliate_item_count: number
matched_count: number
updated_count: number
from_date: string | null
to_date: string | null
}
/** 销售事实单次重算最多 400 天,长历史按窗口分批。 */
async function rebuildFactsForRange(fromDate: string, toDate: string) {
let cursor = fromDate
let deletedCount = 0
let insertedCount = 0
let batchCount = 0
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
batchCount += 1
cursor = shiftDate(chunkEnd, 1)
}
return { fromDate, toDate, deletedCount, insertedCount, batchCount }
}
/**
* 按自动发货履约证据直接回填订单项标准商品。
*
* 自动发货只有 affiliate-dash 一个上游渠道,SKU 固定归一为 `aff:<skuCode>`,
* 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。
*/
export async function backfillAdminAffiliateDashSkus() {
const now = nowIso()
const result = await query<AffiliateSkuBackfillRow>(
`
WITH affiliate_item AS (
SELECT DISTINCT ON (ft.order_item_id)
ft.order_item_id AS id,
btrim(ft.context_json->>'skuCode') AS sku_code
FROM fulfillment_tasks ft
WHERE ft.executor_key = 'affiliate_dash'
AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> ''
ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
),
target AS (
SELECT ai.id, p.id AS canonical_product_id
FROM affiliate_item ai
JOIN product_catalog p
ON p.canonical_sku = 'aff:' || ai.sku_code
),
updated AS (
UPDATE order_items oi
SET canonical_product_id = target.canonical_product_id,
updated_at = $1
FROM target
WHERE oi.id = target.id
AND oi.canonical_product_id IS DISTINCT FROM target.canonical_product_id
RETURNING oi.id
)
SELECT
(
SELECT COUNT(DISTINCT ft.order_item_id)::int
FROM fulfillment_tasks ft
WHERE ft.executor_key = 'affiliate_dash'
) AS affiliate_item_count,
(SELECT COUNT(*)::int FROM target) AS matched_count,
(SELECT COUNT(*)::int FROM updated) AS updated_count,
(
SELECT to_char(
MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
)
FROM affiliate_item
JOIN order_items oi ON oi.id = affiliate_item.id
JOIN orders o ON o.id = oi.order_id
) AS from_date,
(
SELECT to_char(
MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
)
FROM affiliate_item
JOIN order_items oi ON oi.id = affiliate_item.id
JOIN orders o ON o.id = oi.order_id
) AS to_date
`,
[now],
)
const row = result.rows[0]
const affiliateItemCount = Number(row?.affiliate_item_count || 0)
const matchedCount = Number(row?.matched_count || 0)
const updatedCount = Number(row?.updated_count || 0)
const skippedCount = Math.max(0, matchedCount - updatedCount)
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id
AND ft.executor_key = 'affiliate_dash'
)
`,
)
const remainingCount = Number(remainingResult.rows[0]?.total || 0)
let factsRebuild: Awaited<ReturnType<typeof rebuildFactsForRange>> | null = null
if (matchedCount > 0 && row?.from_date && row?.to_date) {
factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date)
}
return {
updatedCount,
skippedCount,
remainingCount,
affiliateItemCount,
matchedCount,
factsRebuild,
}
}
type WorkOrderSkuBackfillRow = {
work_order_item_count: number
from_date: string | null
to_date: string | null
}
const WORK_ORDER_BACKFILL_BATCH_SIZE = 20000
/**
* 回填接单平台订单项的标准商品。
*
* 第一步用工单真实使用的 `product_rule_id` 指向 `tpl:<rule_key>`;
* 仍无法归一的长尾项,再按 `provider + shop_id + sku_code + normalized_sku_name`
* 补建 `wo:` 商品和 `sku_code_name` 渠道映射,保证后续同类订单能自动命中。
*/
export async function backfillAdminWorkOrderProductSkus() {
const now = nowIso()
// order_item_id 有唯一索引,这里可以直接 count,避免 DISTINCT ON 被
// 三个子查询反复物化后撞 15 秒语句超时。
const result = await query<WorkOrderSkuBackfillRow>(
`
SELECT
COUNT(*)::int AS work_order_item_count,
to_char(
MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
) AS from_date,
to_char(
MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
) AS to_date
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.order_item_id IS NOT NULL
`,
)
// 单条 UPDATE 覆盖十几万行会撞 15 秒语句超时;按订单项 id 分批,
// 每批只选 2 万条工单证据,再按 id 数组更新。
let lastRuleItemId = 0
let ruleMatchedCount = 0
let ruleUpdatedCount = 0
for (let batch = 0; batch < 40; batch += 1) {
const targetRows = await query<{ id: number; canonical_product_id: number }>(
`
WITH work_order_item AS (
SELECT DISTINCT ON (wo.order_item_id)
wo.order_item_id AS id,
wo.product_rule_id
FROM work_orders wo
WHERE wo.order_item_id IS NOT NULL
AND wo.order_item_id > $1
ORDER BY wo.order_item_id, wo.id DESC
LIMIT $2
)
SELECT woi.id, p.id AS canonical_product_id
FROM work_order_item woi
JOIN work_product_rules wpr ON wpr.id = woi.product_rule_id
JOIN product_catalog p ON p.canonical_sku = 'tpl:' || wpr.rule_key
ORDER BY woi.id
`,
[lastRuleItemId, WORK_ORDER_BACKFILL_BATCH_SIZE],
)
if (targetRows.rows.length === 0) break
const lastSelectedId = Number(targetRows.rows[targetRows.rows.length - 1]?.id || 0)
if (lastSelectedId <= lastRuleItemId) break
lastRuleItemId = lastSelectedId
ruleMatchedCount += targetRows.rows.length
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = $1
FROM unnest($2::bigint[], $3::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS DISTINCT FROM input.canonical_product_id
`,
[
now,
targetRows.rows.map((row) => Number(row.id)),
targetRows.rows.map((row) => Number(row.canonical_product_id)),
],
)
ruleUpdatedCount += updated.rowCount || 0
if (targetRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break
}
const tailProducts = await query<{ inserted: boolean }>(
`
WITH tail AS (
SELECT
oi.normalized_sku_name AS name,
MIN(oi.sku_name) AS sample_name
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
GROUP BY 1
),
missing AS (
SELECT t.name, t.sample_name
FROM tail t
WHERE NOT EXISTS (
SELECT 1 FROM product_catalog p
WHERE p.status = 'active' AND p.canonical_name = t.name
)
)
INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
)
SELECT
'wo:' || substr(md5(m.name), 1, 12),
m.name,
'',
CASE
WHEN m.sample_name LIKE '%颗%' THEN '颗'
WHEN m.sample_name LIKE '%个%' THEN '个'
ELSE '件'
END,
'active',
jsonb_build_object('source', 'work_order_sku_name', 'normalizedName', m.name),
$1,
$1
FROM missing m
ON CONFLICT (canonical_sku) DO UPDATE
SET updated_at = $1
RETURNING (xmax = 0) AS inserted
`,
[now],
)
const tailMappings = await query<{ id: number }>(
`
WITH tail AS (
SELECT
o.provider,
o.shop_id,
oi.sku_code,
MIN(oi.sku_name) AS sku_name,
oi.normalized_sku_name
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
GROUP BY 1, 2, 3, 5
),
resolved AS (
SELECT t.*, p.id AS canonical_product_id
FROM tail t
JOIN LATERAL (
SELECT id
FROM product_catalog
WHERE status = 'active' AND canonical_name = t.normalized_sku_name
ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
LIMIT 1
) p ON TRUE
)
INSERT INTO channel_product_mappings (
provider, shop_id, platform_sku_code, platform_sku_name,
platform_sku_name_key, match_type, canonical_product_id,
quantity_multiplier, confidence, source, enabled, note, created_at, updated_at
)
SELECT
r.provider, r.shop_id, r.sku_code, r.sku_name,
r.normalized_sku_name, 'sku_code_name', r.canonical_product_id,
1, 'exact', 'import', TRUE, '按历史工单 SKU 名称自动补齐', $1, $1
FROM resolved r
ON CONFLICT (
provider, shop_id, platform_sku_code, match_type, platform_sku_name_key
) DO NOTHING
RETURNING id
`,
[now],
)
let lastTailItemId = 0
let tailUpdatedCount = 0
for (let batch = 0; batch < 40; batch += 1) {
const tailRows = await query<{ id: number; canonical_product_id: number }>(
`
WITH tail AS (
SELECT oi.id, oi.normalized_sku_name
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
AND oi.id > $1
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
ORDER BY oi.id
LIMIT $2
)
SELECT t.id, p.id AS canonical_product_id
FROM tail t
JOIN LATERAL (
SELECT id
FROM product_catalog
WHERE status = 'active' AND canonical_name = t.normalized_sku_name
ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
LIMIT 1
) p ON TRUE
ORDER BY t.id
`,
[lastTailItemId, WORK_ORDER_BACKFILL_BATCH_SIZE],
)
if (tailRows.rows.length === 0) break
const lastSelectedId = Number(tailRows.rows[tailRows.rows.length - 1]?.id || 0)
if (lastSelectedId <= lastTailItemId) break
lastTailItemId = lastSelectedId
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = $1
FROM unnest($2::bigint[], $3::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS NULL
`,
[
now,
tailRows.rows.map((row) => Number(row.id)),
tailRows.rows.map((row) => Number(row.canonical_product_id)),
],
)
tailUpdatedCount += updated.rowCount || 0
if (tailRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break
}
const row = result.rows[0]
const workOrderItemCount = Number(row?.work_order_item_count || 0)
const tailProductCount = tailProducts.rows.filter((item) => item.inserted).length
const tailMappingCount = tailMappings.rows.length
const updatedCount = ruleUpdatedCount + tailUpdatedCount
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
`,
)
const remainingCount = Number(remainingResult.rows[0]?.total || 0)
const matchedCount = Math.max(0, workOrderItemCount - remainingCount)
const skippedCount = Math.max(0, matchedCount - updatedCount)
let factsRebuild: Awaited<ReturnType<typeof rebuildFactsForRange>> | null = null
if (matchedCount > 0 && row?.from_date && row?.to_date) {
factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date)
}
return {
updatedCount,
skippedCount,
remainingCount,
workOrderItemCount,
matchedCount,
ruleMatchedCount,
tailUpdatedCount,
tailProductCount,
tailMappingCount,
factsRebuild,
}
}
type WorkOrderTemplateCostSyncRow = {
rule_count: number
unit_price_rule_count: number
active_version_count: number
}
/**
* 把接单模板的发单金额同步成规则级成本版本。
*
* 固定金额模板取 `reward_amount`;按件模板取 `unit_price_fen`,数量由订单 SKU
* 决定。全局 0 元版本只作为“按工单实际结算”的占位,不再代表模板价格。
*/
export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) {
const now = nowIso()
const scopedRuleId = ruleId && Number.isFinite(ruleId) ? Number(ruleId) : null
const retired = await query(
`
WITH target AS (
SELECT
wpr.id AS rule_id,
CASE
WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen
ELSE wpr.reward_amount
END AS amount_fen
FROM work_product_rules wpr
WHERE ($2::bigint IS NULL OR wpr.id = $2)
),
sys AS (
SELECT id FROM cost_items
WHERE code = 'sys_work_order_reward' AND status = 'active'
),
changed AS (
SELECT v.id
FROM cost_versions v
JOIN sys si ON si.id = v.cost_item_id
JOIN target t ON t.rule_id = v.work_product_rule_id
WHERE v.scope_type = 'rule'
AND v.status = 'active'
AND v.amount_fen <> t.amount_fen
)
UPDATE cost_versions v
SET status = 'retired',
effective_to = $1,
change_reason = '模板发单金额变更',
updated_at = $1
FROM changed c
WHERE v.id = c.id
RETURNING v.id
`,
[now, scopedRuleId],
)
const inserted = await query<{ id: number }>(
`
WITH target AS (
SELECT
wpr.id AS rule_id,
CASE
WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen
ELSE wpr.reward_amount
END AS amount_fen
FROM work_product_rules wpr
WHERE ($3::bigint IS NULL OR wpr.id = $3)
),
sys AS (
SELECT id FROM cost_items
WHERE code = 'sys_work_order_reward' AND status = 'active'
),
missing AS (
SELECT
t.rule_id,
t.amount_fen,
si.id AS cost_item_id,
EXISTS (
SELECT 1 FROM cost_versions h
WHERE h.cost_item_id = si.id
AND h.scope_type = 'rule'
AND h.work_product_rule_id = t.rule_id
) AS has_history
FROM target t
CROSS JOIN sys si
WHERE t.amount_fen > 0
AND NOT EXISTS (
SELECT 1 FROM cost_versions v
WHERE v.cost_item_id = si.id
AND v.scope_type = 'rule'
AND v.work_product_rule_id = t.rule_id
AND v.status = 'active'
)
)
INSERT INTO cost_versions (
cost_item_id, scope_type, provider, shop_id, canonical_product_id,
work_product_rule_id, amount_fen, rate_bps, effective_from, effective_to,
source, status, change_reason, created_by_admin_user_id,
approved_by_admin_user_id, created_at, updated_at
)
SELECT
m.cost_item_id,
'rule',
'',
'',
NULL,
m.rule_id,
m.amount_fen,
0,
CASE WHEN m.has_history THEN $1::timestamptz ELSE $2::timestamptz END,
NULL,
'import',
'active',
CASE
WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0
THEN '模板按件发单单价(数量取订单 SKU)'
ELSE '模板固定发单金额'
END,
NULL,
NULL,
$1,
$1
FROM missing m
JOIN work_product_rules wpr ON wpr.id = m.rule_id
ON CONFLICT DO NOTHING
RETURNING id
`,
[now, WORK_ORDER_TEMPLATE_COST_BASELINE, scopedRuleId],
)
const linked = await query(
`
WITH rule_version AS (
SELECT DISTINCT ON (v.work_product_rule_id)
v.work_product_rule_id,
v.id AS cost_version_id,
v.cost_item_id
FROM cost_versions v
JOIN cost_items ci ON ci.id = v.cost_item_id
AND ci.code = 'sys_work_order_reward'
WHERE v.scope_type = 'rule'
AND v.status = 'active'
AND ($1::bigint IS NULL OR v.work_product_rule_id = $1)
ORDER BY v.work_product_rule_id, v.effective_from DESC, v.id DESC
)
UPDATE cost_snapshots cs
SET cost_version_id = rv.cost_version_id
FROM work_orders wo
JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id
WHERE cs.order_item_id = wo.order_item_id
AND cs.cost_item_id = rv.cost_item_id
AND cs.superseded_at IS NULL
AND cs.cost_version_id <> rv.cost_version_id
AND EXISTS (
SELECT 1 FROM cost_versions gv
WHERE gv.id = cs.cost_version_id AND gv.scope_type = 'global'
)
AND ($1::bigint IS NULL OR wo.product_rule_id = $1)
`,
[scopedRuleId],
)
const summary = await query<WorkOrderTemplateCostSyncRow>(
`
SELECT
COUNT(*)::int AS rule_count,
COUNT(*) FILTER (
WHERE wpr.unit_price_fen > 0 AND wpr.reward_amount = 0
)::int AS unit_price_rule_count,
COUNT(v.id) FILTER (
WHERE v.scope_type = 'rule' AND v.status = 'active'
)::int AS active_version_count
FROM work_product_rules wpr
LEFT JOIN cost_versions v
ON v.cost_item_id = (
SELECT id FROM cost_items WHERE code = 'sys_work_order_reward' LIMIT 1
)
AND v.work_product_rule_id = wpr.id
WHERE ($1::bigint IS NULL OR wpr.id = $1)
`,
[scopedRuleId],
)
const row = summary.rows[0]
return {
ruleId: scopedRuleId,
ruleCount: Number(row?.rule_count || 0),
unitPriceRuleCount: Number(row?.unit_price_rule_count || 0),
activeVersionCount: Number(row?.active_version_count || 0),
createdCount: inserted.rows.length,
retiredCount: retired.rowCount || 0,
linkedSnapshotCount: linked.rowCount || 0,
}
}
type TemplateBootstrapTemplateRow = {
id: number
@@ -0,0 +1,467 @@
/**
* 接单平台(work_order)商品回填与模板发单金额同步。
*/
import { query } from '../../db/client.js'
import { nowIso } from '../../utils/time.js'
import { rebuildFactsForRange } from './affiliate-product-service.js'
import { WORK_ORDER_TEMPLATE_COST_BASELINE } from './cost-center-shared.js'
type WorkOrderSkuBackfillRow = {
work_order_item_count: number
from_date: string | null
to_date: string | null
}
const WORK_ORDER_BACKFILL_BATCH_SIZE = 20000
/**
* 回填接单平台订单项的标准商品。
*
* 第一步用工单真实使用的 `product_rule_id` 指向 `tpl:<rule_key>`;
* 仍无法归一的长尾项,再按 `provider + shop_id + sku_code + normalized_sku_name`
* 补建 `wo:` 商品和 `sku_code_name` 渠道映射,保证后续同类订单能自动命中。
*/
export async function backfillAdminWorkOrderProductSkus() {
const now = nowIso()
// order_item_id 有唯一索引,这里可以直接 count,避免 DISTINCT ON 被
// 三个子查询反复物化后撞 15 秒语句超时。
const result = await query<WorkOrderSkuBackfillRow>(
`
SELECT
COUNT(*)::int AS work_order_item_count,
to_char(
MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
) AS from_date,
to_char(
MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
'YYYY-MM-DD'
) AS to_date
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.order_item_id IS NOT NULL
`,
)
// 单条 UPDATE 覆盖十几万行会撞 15 秒语句超时;按订单项 id 分批,
// 每批只选 2 万条工单证据,再按 id 数组更新。
let lastRuleItemId = 0
let ruleMatchedCount = 0
let ruleUpdatedCount = 0
for (let batch = 0; batch < 40; batch += 1) {
const targetRows = await query<{ id: number; canonical_product_id: number }>(
`
WITH work_order_item AS (
SELECT DISTINCT ON (wo.order_item_id)
wo.order_item_id AS id,
wo.product_rule_id
FROM work_orders wo
WHERE wo.order_item_id IS NOT NULL
AND wo.order_item_id > $1
ORDER BY wo.order_item_id, wo.id DESC
LIMIT $2
)
SELECT woi.id, p.id AS canonical_product_id
FROM work_order_item woi
JOIN work_product_rules wpr ON wpr.id = woi.product_rule_id
JOIN product_catalog p ON p.canonical_sku = 'tpl:' || wpr.rule_key
ORDER BY woi.id
`,
[lastRuleItemId, WORK_ORDER_BACKFILL_BATCH_SIZE],
)
if (targetRows.rows.length === 0) break
const lastSelectedId = Number(targetRows.rows[targetRows.rows.length - 1]?.id || 0)
if (lastSelectedId <= lastRuleItemId) break
lastRuleItemId = lastSelectedId
ruleMatchedCount += targetRows.rows.length
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = $1
FROM unnest($2::bigint[], $3::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS DISTINCT FROM input.canonical_product_id
`,
[
now,
targetRows.rows.map((row) => Number(row.id)),
targetRows.rows.map((row) => Number(row.canonical_product_id)),
],
)
ruleUpdatedCount += updated.rowCount || 0
if (targetRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break
}
const tailProducts = await query<{ inserted: boolean }>(
`
WITH tail AS (
SELECT
oi.normalized_sku_name AS name,
MIN(oi.sku_name) AS sample_name
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
GROUP BY 1
),
missing AS (
SELECT t.name, t.sample_name
FROM tail t
WHERE NOT EXISTS (
SELECT 1 FROM product_catalog p
WHERE p.status = 'active' AND p.canonical_name = t.name
)
)
INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
)
SELECT
'wo:' || substr(md5(m.name), 1, 12),
m.name,
'',
CASE
WHEN m.sample_name LIKE '%颗%' THEN '颗'
WHEN m.sample_name LIKE '%个%' THEN '个'
ELSE '件'
END,
'active',
jsonb_build_object('source', 'work_order_sku_name', 'normalizedName', m.name),
$1,
$1
FROM missing m
ON CONFLICT (canonical_sku) DO UPDATE
SET updated_at = $1
RETURNING (xmax = 0) AS inserted
`,
[now],
)
const tailMappings = await query<{ id: number }>(
`
WITH tail AS (
SELECT
o.provider,
o.shop_id,
oi.sku_code,
MIN(oi.sku_name) AS sku_name,
oi.normalized_sku_name
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
GROUP BY 1, 2, 3, 5
),
resolved AS (
SELECT t.*, p.id AS canonical_product_id
FROM tail t
JOIN LATERAL (
SELECT id
FROM product_catalog
WHERE status = 'active' AND canonical_name = t.normalized_sku_name
ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
LIMIT 1
) p ON TRUE
)
INSERT INTO channel_product_mappings (
provider, shop_id, platform_sku_code, platform_sku_name,
platform_sku_name_key, match_type, canonical_product_id,
quantity_multiplier, confidence, source, enabled, note, created_at, updated_at
)
SELECT
r.provider, r.shop_id, r.sku_code, r.sku_name,
r.normalized_sku_name, 'sku_code_name', r.canonical_product_id,
1, 'exact', 'import', TRUE, '按历史工单 SKU 名称自动补齐', $1, $1
FROM resolved r
ON CONFLICT (
provider, shop_id, platform_sku_code, match_type, platform_sku_name_key
) DO NOTHING
RETURNING id
`,
[now],
)
let lastTailItemId = 0
let tailUpdatedCount = 0
for (let batch = 0; batch < 40; batch += 1) {
const tailRows = await query<{ id: number; canonical_product_id: number }>(
`
WITH tail AS (
SELECT oi.id, oi.normalized_sku_name
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
AND oi.id > $1
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
ORDER BY oi.id
LIMIT $2
)
SELECT t.id, p.id AS canonical_product_id
FROM tail t
JOIN LATERAL (
SELECT id
FROM product_catalog
WHERE status = 'active' AND canonical_name = t.normalized_sku_name
ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
LIMIT 1
) p ON TRUE
ORDER BY t.id
`,
[lastTailItemId, WORK_ORDER_BACKFILL_BATCH_SIZE],
)
if (tailRows.rows.length === 0) break
const lastSelectedId = Number(tailRows.rows[tailRows.rows.length - 1]?.id || 0)
if (lastSelectedId <= lastTailItemId) break
lastTailItemId = lastSelectedId
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = $1
FROM unnest($2::bigint[], $3::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS NULL
`,
[
now,
tailRows.rows.map((row) => Number(row.id)),
tailRows.rows.map((row) => Number(row.canonical_product_id)),
],
)
tailUpdatedCount += updated.rowCount || 0
if (tailRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break
}
const row = result.rows[0]
const workOrderItemCount = Number(row?.work_order_item_count || 0)
const tailProductCount = tailProducts.rows.filter((item) => item.inserted).length
const tailMappingCount = tailMappings.rows.length
const updatedCount = ruleUpdatedCount + tailUpdatedCount
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
WHERE oi.canonical_product_id IS NULL
AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
`,
)
const remainingCount = Number(remainingResult.rows[0]?.total || 0)
const matchedCount = Math.max(0, workOrderItemCount - remainingCount)
const skippedCount = Math.max(0, matchedCount - updatedCount)
let factsRebuild: Awaited<ReturnType<typeof rebuildFactsForRange>> | null = null
if (matchedCount > 0 && row?.from_date && row?.to_date) {
factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date)
}
return {
updatedCount,
skippedCount,
remainingCount,
workOrderItemCount,
matchedCount,
ruleMatchedCount,
tailUpdatedCount,
tailProductCount,
tailMappingCount,
factsRebuild,
}
}
type WorkOrderTemplateCostSyncRow = {
rule_count: number
unit_price_rule_count: number
active_version_count: number
}
/**
* 把接单模板的发单金额同步成规则级成本版本。
*
* 固定金额模板取 `reward_amount`;按件模板取 `unit_price_fen`,数量由订单 SKU
* 决定。全局 0 元版本只作为“按工单实际结算”的占位,不再代表模板价格。
*/
export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) {
const now = nowIso()
const scopedRuleId = ruleId && Number.isFinite(ruleId) ? Number(ruleId) : null
const retired = await query(
`
WITH target AS (
SELECT
wpr.id AS rule_id,
CASE
WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen
ELSE wpr.reward_amount
END AS amount_fen
FROM work_product_rules wpr
WHERE ($2::bigint IS NULL OR wpr.id = $2)
),
sys AS (
SELECT id FROM cost_items
WHERE code = 'sys_work_order_reward' AND status = 'active'
),
changed AS (
SELECT v.id
FROM cost_versions v
JOIN sys si ON si.id = v.cost_item_id
JOIN target t ON t.rule_id = v.work_product_rule_id
WHERE v.scope_type = 'rule'
AND v.status = 'active'
AND v.amount_fen <> t.amount_fen
)
UPDATE cost_versions v
SET status = 'retired',
effective_to = $1,
change_reason = '模板发单金额变更',
updated_at = $1
FROM changed c
WHERE v.id = c.id
RETURNING v.id
`,
[now, scopedRuleId],
)
const inserted = await query<{ id: number }>(
`
WITH target AS (
SELECT
wpr.id AS rule_id,
CASE
WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen
ELSE wpr.reward_amount
END AS amount_fen
FROM work_product_rules wpr
WHERE ($3::bigint IS NULL OR wpr.id = $3)
),
sys AS (
SELECT id FROM cost_items
WHERE code = 'sys_work_order_reward' AND status = 'active'
),
missing AS (
SELECT
t.rule_id,
t.amount_fen,
si.id AS cost_item_id,
EXISTS (
SELECT 1 FROM cost_versions h
WHERE h.cost_item_id = si.id
AND h.scope_type = 'rule'
AND h.work_product_rule_id = t.rule_id
) AS has_history
FROM target t
CROSS JOIN sys si
WHERE t.amount_fen > 0
AND NOT EXISTS (
SELECT 1 FROM cost_versions v
WHERE v.cost_item_id = si.id
AND v.scope_type = 'rule'
AND v.work_product_rule_id = t.rule_id
AND v.status = 'active'
)
)
INSERT INTO cost_versions (
cost_item_id, scope_type, provider, shop_id, canonical_product_id,
work_product_rule_id, amount_fen, rate_bps, effective_from, effective_to,
source, status, change_reason, created_by_admin_user_id,
approved_by_admin_user_id, created_at, updated_at
)
SELECT
m.cost_item_id,
'rule',
'',
'',
NULL,
m.rule_id,
m.amount_fen,
0,
CASE WHEN m.has_history THEN $1::timestamptz ELSE $2::timestamptz END,
NULL,
'import',
'active',
CASE
WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0
THEN '模板按件发单单价(数量取订单 SKU)'
ELSE '模板固定发单金额'
END,
NULL,
NULL,
$1,
$1
FROM missing m
JOIN work_product_rules wpr ON wpr.id = m.rule_id
ON CONFLICT DO NOTHING
RETURNING id
`,
[now, WORK_ORDER_TEMPLATE_COST_BASELINE, scopedRuleId],
)
const linked = await query(
`
WITH rule_version AS (
SELECT DISTINCT ON (v.work_product_rule_id)
v.work_product_rule_id,
v.id AS cost_version_id,
v.cost_item_id
FROM cost_versions v
JOIN cost_items ci ON ci.id = v.cost_item_id
AND ci.code = 'sys_work_order_reward'
WHERE v.scope_type = 'rule'
AND v.status = 'active'
AND ($1::bigint IS NULL OR v.work_product_rule_id = $1)
ORDER BY v.work_product_rule_id, v.effective_from DESC, v.id DESC
)
UPDATE cost_snapshots cs
SET cost_version_id = rv.cost_version_id
FROM work_orders wo
JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id
WHERE cs.order_item_id = wo.order_item_id
AND cs.cost_item_id = rv.cost_item_id
AND cs.superseded_at IS NULL
AND cs.cost_version_id <> rv.cost_version_id
AND EXISTS (
SELECT 1 FROM cost_versions gv
WHERE gv.id = cs.cost_version_id AND gv.scope_type = 'global'
)
AND ($1::bigint IS NULL OR wo.product_rule_id = $1)
`,
[scopedRuleId],
)
const summary = await query<WorkOrderTemplateCostSyncRow>(
`
SELECT
COUNT(*)::int AS rule_count,
COUNT(*) FILTER (
WHERE wpr.unit_price_fen > 0 AND wpr.reward_amount = 0
)::int AS unit_price_rule_count,
COUNT(v.id) FILTER (
WHERE v.scope_type = 'rule' AND v.status = 'active'
)::int AS active_version_count
FROM work_product_rules wpr
LEFT JOIN cost_versions v
ON v.cost_item_id = (
SELECT id FROM cost_items WHERE code = 'sys_work_order_reward' LIMIT 1
)
AND v.work_product_rule_id = wpr.id
WHERE ($1::bigint IS NULL OR wpr.id = $1)
`,
[scopedRuleId],
)
const row = summary.rows[0]
return {
ruleId: scopedRuleId,
ruleCount: Number(row?.rule_count || 0),
unitPriceRuleCount: Number(row?.unit_price_rule_count || 0),
activeVersionCount: Number(row?.active_version_count || 0),
createdCount: inserted.rows.length,
retiredCount: retired.rowCount || 0,
linkedSnapshotCount: linked.rowCount || 0,
}
}