fix(finance): 支持生产快照升级并修复成本回填

修复生产 101 到 112 的迁移兼容问题;接单商品回填改为分批执行;修正未验收工单成本快照的 order_id/order_item_id 映射;更新本地生产快照数据基线。
This commit is contained in:
yml2213
2026-10-25 11:10:33 +08:00
parent 790de676c6
commit 7824170d3c
9 changed files with 156 additions and 118 deletions
@@ -246,9 +246,6 @@ CREATE INDEX IF NOT EXISTS idx_cost_exceptions_status_created
ON cost_exceptions (status, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_cost_exceptions_item
ON cost_exceptions (order_item_id, status);
-- 按履约方式筛异常时走半连接,这条索引让 order_items 侧只扫目标履约方式的行。
CREATE INDEX IF NOT EXISTS idx_order_items_fulfillment_id
ON order_items (fulfillment_mode, id);
DO $$
BEGIN
@@ -40,6 +40,10 @@ CREATE INDEX IF NOT EXISTS idx_order_items_fulfillment_pending
ON order_items (order_id)
WHERE fulfillment_mode IS NULL;
-- 按履约方式筛异常时走半连接,这条索引让 order_items 侧只扫目标履约方式的行。
CREATE INDEX IF NOT EXISTS idx_order_items_fulfillment_id
ON order_items (fulfillment_mode, id);
COMMENT ON COLUMN order_items.fulfillment_mode IS
'履约方式:work_order 接单平台 / affiliate_dash 自动发货 / legacy_platform 历史平台(已下线)/ manual 人工;唯一真源,报表与后台只读此字段';
COMMENT ON COLUMN order_items.fulfillment_decided_at IS
@@ -22,13 +22,18 @@
-- 关联备用金支出单(fund_expense_id 为空)的,保留原主体 id 不动,避免误删成本;
-- 有 fund_expense_id 的改挂到备用金支出单。
UPDATE cost_snapshots cs
SET subject_id = mer.fund_expense_id
FROM manual_expense_records mer
WHERE cs.subject_type = 'manual_case'
AND mer.fund_expense_id IS NOT NULL
AND cs.subject_id = mer.id
AND cs.subject_id <> mer.fund_expense_id;
DO $$
BEGIN
IF to_regclass('public.manual_expense_records') IS NOT NULL THEN
UPDATE cost_snapshots cs
SET subject_id = mer.fund_expense_id
FROM manual_expense_records mer
WHERE cs.subject_type = 'manual_case'
AND mer.fund_expense_id IS NOT NULL
AND cs.subject_id = mer.id
AND cs.subject_id <> mer.fund_expense_id;
END IF;
END $$;
-- 若同一备用金支出单已存在有效快照,保留先写的,把重复的作废(superseded_at)。
UPDATE cost_snapshots dup
@@ -14,6 +14,8 @@ WHERE ci.id = v.cost_item_id
AND v.status = 'active';
-- 若人工成本项还没有生效版本,补一条 0 元全局占位,保证快照能挂到版本上。
-- 104 已写入一条 1970 年的 draft 占位;这里必须用新的 effective_from,
-- 否则会与 draft 行撞 uq_cost_versions_scope(唯一键不区分状态)。
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, source,
@@ -21,7 +23,7 @@ INSERT INTO cost_versions (
)
SELECT
ci.id, 'global', '', '', NULL, NULL, 0, 0,
'1970-01-01T00:00:00+08:00', 'manual', 'active',
NOW(), 'manual', 'active',
'系统占位:人工成本按人工 SKU 配置的消耗积分结算', NOW(), NOW()
FROM cost_items ci
WHERE ci.code = 'sys_manual_service'
@@ -536,13 +536,13 @@ async function syncCommittedWorkOrderRewardSnapshots(
ORDER BY wo.order_item_id, wo.id DESC
)
INSERT INTO cost_snapshots (
subject_type, subject_id, order_item_id, work_order_id,
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.work_order_id,
'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',
@@ -419,12 +419,12 @@ export async function backfillAdminAffiliateDashSkus() {
type WorkOrderSkuBackfillRow = {
work_order_item_count: number
matched_count: number
updated_count: number
from_date: string | null
to_date: string | null
}
const WORK_ORDER_BACKFILL_BATCH_SIZE = 20000
/**
* 回填接单平台订单项的标准商品。
*
@@ -434,60 +434,78 @@ type WorkOrderSkuBackfillRow = {
*/
export async function backfillAdminWorkOrderProductSkus() {
const now = nowIso()
// order_item_id 有唯一索引,这里可以直接 count,避免 DISTINCT ON 被
// 三个子查询反复物化后撞 15 秒语句超时。
const result = await query<WorkOrderSkuBackfillRow>(
`
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
ORDER BY wo.order_item_id, wo.id DESC
),
target AS (
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
),
updated AS (
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 = target.canonical_product_id,
SET canonical_product_id = input.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 wo.order_item_id)::int
FROM work_orders wo
WHERE wo.order_item_id IS NOT NULL
) AS work_order_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 work_order_item
JOIN order_items oi ON oi.id = work_order_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 work_order_item
JOIN order_items oi ON oi.id = work_order_item.id
JOIN orders o ON o.id = oi.order_id
) AS to_date
`,
[now],
)
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 }>(
`
@@ -579,16 +597,21 @@ export async function backfillAdminWorkOrderProductSkus() {
[now],
)
const tailUpdated = await query<{ 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 EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
),
resolved AS (
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 (
@@ -598,25 +621,39 @@ export async function backfillAdminWorkOrderProductSkus() {
ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
LIMIT 1
) p ON TRUE
)
UPDATE order_items oi
SET canonical_product_id = r.canonical_product_id,
updated_at = $1
FROM resolved r
WHERE oi.id = r.id
AND oi.canonical_product_id IS NULL
RETURNING oi.id
`,
[now],
)
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 ruleMatchedCount = Number(row?.matched_count || 0)
const ruleUpdatedCount = Number(row?.updated_count || 0)
const tailProductCount = tailProducts.rows.filter((item) => item.inserted).length
const tailMappingCount = tailMappings.rows.length
const tailUpdatedCount = tailUpdated.rows.length
const updatedCount = ruleUpdatedCount + tailUpdatedCount
const remainingResult = await query<{ total: number }>(
`