fix(fulfillment): 收口渠道商品与统一结算边界

This commit is contained in:
yml2213
2026-11-06 17:12:44 +08:00
parent e6b3835c9f
commit a333c05df4
23 changed files with 1113 additions and 106 deletions
@@ -165,6 +165,15 @@ test('人工确认前必须自动匹配履约配置,配置保存/定时任务
)
})
test('统一结算金额未知时只生成待对账账单,不写 0 元实际成本', async () => {
const source = await readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8')
assert.ok(source.includes("platformKey === 'on_account' && costFen <= 0"))
assert.ok(source.includes('if (!onAccountCostPending)'))
assert.ok(source.includes("reason: 'on_account_cost_pending'"))
assert.ok(source.includes('costPending: onAccountCostPending'))
})
test('人工成本缺失由管道按配置与快照生成并收口', async () => {
const [pipelineSource, manualSource] = await Promise.all([
readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'),
@@ -21,6 +21,7 @@ import {
attachFulfillmentConfigurationsWithExecutor,
transitionFulfillmentStateWithExecutor,
} from '../fulfillment/fulfillment-state-service.js'
import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
import {
calculateManualCostFen,
normalizeManualCreditQuantity,
@@ -411,6 +412,20 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
errorCode: 'manual_fulfillment_platform_required',
})
}
if (selectedChannel.platformKey === 'on_account') {
const existingBill = await client.query<{ id: number }>(
`SELECT id FROM on_account_bills
WHERE order_item_id = $1 AND status IN ('pending', 'reconciled')
LIMIT 1`,
[orderItemId],
)
if (existingBill.rows[0]) {
throw createHttpError('该订单项已生成统一结算账单,不能重复操作', {
statusCode: 409,
errorCode: 'manual_fulfillment_on_account_already_exists',
})
}
}
// 锁内复查:并发/双击的第二个请求会等锁,醒来后看到快照直接拒绝,不会重复扣款。
const existing = await client.query<{ id: number }>(
@@ -447,6 +462,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
providedCostFen,
})
const costFen = costPlan.costFen
const onAccountCostPending = platformKey === 'on_account' && costFen <= 0
const payeeName = optionalText(payload.payeeName ?? payload.payee_name, 120)
const note = optionalText(payload.note, 300)
const fulfillmentReference = buildManualFulfillmentReference({
@@ -455,16 +471,8 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
orderItemId,
})
// 先把成本项/版本解析完再扣钱:版本缺失不能发生在扣款之后。
const costItemId = await resolveManualCostItemIdWithClient(client)
const versionId = await resolveCostVersionIdWithClient(client, costItemId, row, now)
if (!versionId) {
throw createHttpError('该成本项没有可用的成本版本,请先在成本中心配置后重试', {
statusCode: 409,
errorCode: 'manual_fulfillment_cost_version_missing',
})
}
// 扣款前先确定最终成本模式。统一结算未填金额时只生成待对账账单,
// 不写 0 元 actual 成本快照,避免利润被虚增。
let fundExpenseId: number | null = null
let creditEntryId: number | null = null
let balanceAfter: number | null = null
@@ -522,11 +530,22 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
subjectSource = 'platform_credit'
}
// subject_id 固定为 order_item_id:不再与 manual_expenses.id /
// platform_credit_entries.id 混用,避免两张表序列撞车。
// subject_type 仍为 manual_case,资金流水只放在 metadata 里。
const snapshot = await client.query<{ id: number }>(
`
let snapshotId: number | null = null
if (!onAccountCostPending) {
const costItemId = await resolveManualCostItemIdWithClient(client)
const versionId = await resolveCostVersionIdWithClient(client, costItemId, row, now)
if (!versionId) {
throw createHttpError('该成本项没有可用的成本版本,请先在成本中心配置后重试', {
statusCode: 409,
errorCode: 'manual_fulfillment_cost_version_missing',
})
}
// subject_id 固定为 order_item_id:不再与 manual_expenses.id /
// platform_credit_entries.id 混用,资金流水只放在 metadata 里。
// subject_type 仍为 manual_case。
const snapshot = await client.query<{ id: number }>(
`
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,
@@ -540,50 +559,51 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
)
ON CONFLICT DO NOTHING
RETURNING id
`,
[
orderItemId,
row.order_id_resolved,
row.work_order_id,
costItemId,
versionId,
row.canonical_product_id,
costFen,
JSON.stringify({
source: 'manual_fulfillment_confirm',
subjectSource,
configId: Number(row.config_id),
platformKey,
platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
creditOwnerAdminUserId:
platformKey === 'fund_cash'
? null
: Number(payload.creditOwnerAdminUserId || 0) || null,
creditQuantity,
creditUnitPriceFen: unitPriceFen,
costMode: costPlan.costMode,
`,
[
orderItemId,
row.order_id_resolved,
row.work_order_id,
costItemId,
versionId,
row.canonical_product_id,
costFen,
fundExpenseId,
creditEntryId,
balanceAfter,
payeeName,
note,
fulfilledBy: {
userId: Number(session.userId || 0),
displayName: String(session.displayName || ''),
role: String(session.role || ''),
},
}),
now,
],
)
const snapshotId = snapshot.rows[0]?.id ?? null
if (!snapshotId) {
// 订单项唯一索引兜底:任何竞态都会走到这里,抛错回滚本次扣款。
throw createHttpError('该订单项已有人工成本快照,不能重复录入', {
statusCode: 409,
errorCode: 'manual_fulfillment_snapshot_conflict',
})
JSON.stringify({
source: 'manual_fulfillment_confirm',
subjectSource,
configId: Number(row.config_id),
platformKey,
platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
creditOwnerAdminUserId:
platformKey === 'fund_cash'
? null
: Number(payload.creditOwnerAdminUserId || 0) || null,
creditQuantity,
creditUnitPriceFen: unitPriceFen,
costMode: costPlan.costMode,
costFen,
fundExpenseId,
creditEntryId,
balanceAfter,
payeeName,
note,
fulfilledBy: {
userId: Number(session.userId || 0),
displayName: String(session.displayName || ''),
role: String(session.role || ''),
},
}),
now,
],
)
snapshotId = snapshot.rows[0]?.id ?? null
if (!snapshotId) {
// 订单项唯一索引兜底:任何竞态都会走到这里,抛错回滚本次扣款。
throw createHttpError('该订单项已有人工成本快照,不能重复录入', {
statusCode: 409,
errorCode: 'manual_fulfillment_snapshot_conflict',
})
}
}
if (platformKey === 'on_account') {
@@ -641,30 +661,46 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
reason: platformKey,
})
// 人工缺成本异常立即收口,不等下一轮定时任务。
await client.query(
`UPDATE cost_exceptions
SET status = 'resolved', resolved_at = $2
WHERE subject_type = 'order_item'
AND subject_id = $1
AND reason = 'manual_cost_missing'
AND status = 'open'`,
[orderItemId, now],
)
if (onAccountCostPending) {
await upsertAggregatedException({
reason: 'on_account_cost_pending',
scopeKey: `on_account_cost_pending|-|-|${row.internal_sku}|on_account`,
orderItemId,
detail: {
billId: onAccountBillId,
internalSku: row.internal_sku,
platformOrderId: row.platform_order_id,
},
client,
})
} else {
// 人工缺成本异常立即收口,不等下一轮定时任务。
await client.query(
`UPDATE cost_exceptions
SET status = 'resolved', resolved_at = $2
WHERE subject_type = 'order_item'
AND subject_id = $1
AND reason = 'manual_cost_missing'
AND status = 'open'`,
[orderItemId, now],
)
}
// v2 聚合异常:删除本条明细,没人再引用的异常头自动关闭。
await client.query(`DELETE FROM cost_exception_items WHERE order_item_id = $1`, [orderItemId])
await client.query(
`
UPDATE cost_exceptions ce
SET status = 'resolved', resolved_at = $1, item_count = 0, last_seen_at = $1
WHERE ce.status <> 'resolved'
AND NOT EXISTS (
SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
)
`,
[now],
)
if (!onAccountCostPending) {
// v2 聚合异常:成本已经落账后,删除本条明细并自动关闭空异常头。
await client.query(`DELETE FROM cost_exception_items WHERE order_item_id = $1`, [orderItemId])
await client.query(
`
UPDATE cost_exceptions ce
SET status = 'resolved', resolved_at = $1, item_count = 0, last_seen_at = $1
WHERE ce.status <> 'resolved'
AND NOT EXISTS (
SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
)
`,
[now],
)
}
return {
orderItemId,
@@ -678,6 +714,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
creditEntryId,
balanceAfter,
onAccountBillId,
costPending: onAccountCostPending,
}
})
@@ -693,7 +730,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
* 事实任务默认只回看最近几天,确认一笔较早的订单时不会自动覆盖,
* 所以这里按该订单自己的业务日期精确重算。
*/
async function refreshFactsForOrderItem(orderItemId: number): Promise<void> {
export async function refreshFactsForOrderItem(orderItemId: number): Promise<void> {
try {
const basis = await query<{ stat_date: string | null }>(
`
@@ -841,7 +878,7 @@ async function matchManualFulfillmentConfigWithClient(client: PoolClient, orderI
return true
}
async function resolveManualCostItemIdWithClient(client: PoolClient) {
export async function resolveManualCostItemIdWithClient(client: PoolClient) {
const result = await client.query<{ id: number }>(
`SELECT id FROM cost_items WHERE code = 'sys_manual_service' AND status = 'active' LIMIT 1`,
)
@@ -855,7 +892,7 @@ async function resolveManualCostItemIdWithClient(client: PoolClient) {
return id
}
async function resolveCostVersionIdWithClient(
export async function resolveCostVersionIdWithClient(
client: PoolClient,
costItemId: number,
row: {
@@ -0,0 +1,14 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
test('统一结算对账用实际金额替代成本快照并重算事实', async () => {
const source = await readFile(new URL('./on-account-bill-service.ts', import.meta.url), 'utf8')
assert.ok(source.includes('withTransaction('))
assert.ok(source.includes("status = 'reconciled'"))
assert.ok(source.includes('SET superseded_at = $2'))
assert.ok(source.includes("source: 'on_account_reconcile'"))
assert.ok(source.includes("'on_account_cost_pending'"))
assert.ok(source.includes('refreshFactsForOrderItem'))
})
@@ -0,0 +1,306 @@
import { query, withTransaction } from '../../db/client.js'
import type { JsonObject } from '../../types/json.js'
import { createHttpError } from '../../utils/http.js'
import { nowIso } from '../../utils/time.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import { resolveExceptionsForItemsByReason } from '../cost-center/cost-exception-service.js'
import {
refreshFactsForOrderItem,
resolveCostVersionIdWithClient,
resolveManualCostItemIdWithClient,
} from './manual-fulfillment-service.js'
type OnAccountBillRow = {
id: number
bill_no: string
order_id: number
order_item_id: number
sku_key: string
canonical_product_id: number | null
platform_order_id: string
amount_fen: string | number
status: 'pending' | 'reconciled' | 'void'
payee_name: string
note: string
metadata_json: Record<string, unknown>
created_by_admin_user_id: number | null
created_at: string
updated_at: string
reconciled_at: string | null
provider?: string
shop_id?: string
sku_name?: string
revenue_amount_fen?: string | number
}
function mapBill(row: OnAccountBillRow) {
return {
id: Number(row.id),
billNo: row.bill_no,
orderId: Number(row.order_id),
orderItemId: Number(row.order_item_id),
skuKey: row.sku_key,
skuName: row.sku_name || row.sku_key,
canonicalProductId: row.canonical_product_id ? Number(row.canonical_product_id) : null,
platformOrderId: row.platform_order_id,
provider: row.provider || '',
shopId: row.shop_id || '',
revenueFen: Number(row.revenue_amount_fen || 0),
amountFen: Number(row.amount_fen || 0),
status: row.status,
payeeName: row.payee_name || '',
note: row.note || '',
metadata: row.metadata_json || {},
createdByAdminUserId: row.created_by_admin_user_id
? Number(row.created_by_admin_user_id)
: null,
createdAt: row.created_at,
updatedAt: row.updated_at,
reconciledAt: row.reconciled_at,
}
}
const BILL_SELECT = `
SELECT
bill.*,
order_row.provider,
order_row.shop_id,
item.sku_name,
item.revenue_amount_fen
FROM on_account_bills bill
JOIN orders order_row ON order_row.id = bill.order_id
JOIN order_items item ON item.id = bill.order_item_id
`
export async function listOnAccountBills(input: JsonObject = {}) {
const page = Math.max(1, Math.trunc(Number(input.page) || 1))
const pageSize = Math.min(100, Math.max(1, Math.trunc(Number(input.pageSize) || 20)))
const keyword = String(input.keyword || '').trim()
const status = String(input.status || '').trim()
const where: string[] = []
const params: unknown[] = []
if (keyword) {
params.push(`%${keyword}%`)
where.push(`(
bill.bill_no ILIKE $${params.length}
OR bill.platform_order_id ILIKE $${params.length}
OR bill.sku_key ILIKE $${params.length}
OR item.sku_name ILIKE $${params.length}
)`)
}
if (['pending', 'reconciled', 'void'].includes(status)) {
params.push(status)
where.push(`bill.status = $${params.length}`)
}
const whereSql = where.length > 0 ? `WHERE ${where.join(' AND ')}` : ''
const total = await query<{ total: number }>(
`SELECT COUNT(*)::int AS total
FROM on_account_bills bill
JOIN order_items item ON item.id = bill.order_item_id
${whereSql}`,
params,
)
const listParams = [...params, pageSize, (page - 1) * pageSize]
const result = await query<OnAccountBillRow>(
`${BILL_SELECT}
${whereSql}
ORDER BY
CASE bill.status WHEN 'pending' THEN 0 WHEN 'reconciled' THEN 1 ELSE 2 END,
bill.created_at DESC,
bill.id DESC
LIMIT $${params.length + 1} OFFSET $${params.length + 2}`,
listParams,
)
return {
items: result.rows.map(mapBill),
pagination: {
page,
pageSize,
total: Number(total.rows[0]?.total || 0),
},
}
}
export async function reconcileOnAccountBill(
billIdInput: unknown,
payload: JsonObject,
session: AdminSession,
) {
const billId = Number(billIdInput)
if (!Number.isInteger(billId) || billId <= 0) {
throw createHttpError('统一结算账单 ID 无效', { statusCode: 400 })
}
const amountFen = Number(payload.amountFen ?? payload.amount_fen)
if (!Number.isFinite(amountFen) || amountFen <= 0 || !Number.isInteger(amountFen)) {
throw createHttpError('实际结算金额必须是大于 0 的整数分', {
statusCode: 400,
errorCode: 'on_account_amount_invalid',
})
}
const now = nowIso()
const result = await withTransaction(async (client) => {
const billResult = await client.query<OnAccountBillRow>(
`${BILL_SELECT}
WHERE bill.id = $1
LIMIT 1
FOR UPDATE OF bill, item`,
[billId],
)
const bill = billResult.rows[0]
if (!bill) {
throw createHttpError('统一结算账单不存在', { statusCode: 404 })
}
if (bill.status !== 'pending') {
throw createHttpError('只有待对账账单可以确认结算', {
statusCode: 409,
errorCode: 'on_account_bill_not_pending',
})
}
const orderResult = await client.query<{
provider: string
shop_id: string
paid_at: string | null
created_at: string
}>(
`SELECT provider, shop_id, paid_at, created_at
FROM orders WHERE id = $1`,
[bill.order_id],
)
const order = orderResult.rows[0]
if (!order) {
throw createHttpError('统一结算账单关联订单不存在', { statusCode: 409 })
}
const basisAt = order.paid_at || order.created_at
const period = await client.query<{ status: string }>(
`SELECT status FROM finance_periods
WHERE period_key = to_char($1::timestamptz AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM')
LIMIT 1`,
[basisAt],
)
if (period.rows[0]?.status === 'closed') {
throw createHttpError('该订单所属期间已关账,请先反关账后再对账', {
statusCode: 409,
errorCode: 'on_account_period_closed',
})
}
const workOrder = await client.query<{ product_rule_id: number | null }>(
`SELECT product_rule_id FROM work_orders
WHERE order_item_id = $1 AND status <> 'cancelled'
ORDER BY id DESC LIMIT 1`,
[bill.order_item_id],
)
const costItemId = await resolveManualCostItemIdWithClient(client)
const versionId = await resolveCostVersionIdWithClient(
client,
costItemId,
{
order_id_resolved: Number(bill.order_id),
provider: order.provider,
shop_id: order.shop_id,
canonical_product_id: bill.canonical_product_id,
work_product_rule_id: workOrder.rows[0]?.product_rule_id || null,
credit_unit_price_fen: '1',
},
basisAt,
)
if (!versionId) {
throw createHttpError('人工履约成本版本不可用,无法完成对账', {
statusCode: 409,
errorCode: 'on_account_cost_version_missing',
})
}
await client.query(
`UPDATE cost_snapshots
SET superseded_at = $2
WHERE order_item_id = $1
AND subject_type = 'manual_case'
AND superseded_at IS NULL
AND reversed_at IS NULL`,
[bill.order_item_id, now],
)
const snapshot = await client.query<{ id: number }>(
`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 (
'manual_case', $1, $2, $1, NULL,
$3, $4, $5,
1, $6, $6, 'actual', 'manual_input',
$7::jsonb, $8, $9, $9
)
RETURNING id`,
[
bill.order_item_id,
bill.order_id,
costItemId,
versionId,
bill.canonical_product_id,
amountFen,
JSON.stringify({
source: 'on_account_reconcile',
billId: bill.id,
billNo: bill.bill_no,
platformKey: 'on_account',
previousAmountFen: Number(bill.amount_fen || 0),
reconciledBy: {
userId: Number(session.userId || 0),
displayName: String(session.displayName || ''),
},
}),
basisAt,
now,
],
)
const snapshotId = Number(snapshot.rows[0]?.id || 0)
if (!snapshotId) {
throw createHttpError('统一结算成本快照保存失败', { statusCode: 500 })
}
const updated = await client.query<OnAccountBillRow>(
`UPDATE on_account_bills
SET amount_fen = $1,
status = 'reconciled',
payee_name = $2,
note = $3,
metadata_json = metadata_json || $4::jsonb,
reconciled_at = $5,
updated_at = $5
WHERE id = $6
RETURNING *`,
[
amountFen,
String(payload.payeeName ?? payload.payee_name ?? bill.payee_name ?? '').slice(0, 120),
String(payload.note ?? bill.note ?? '').slice(0, 300),
JSON.stringify({
reconciledByAdminUserId: Number(session.userId || 0) || null,
costSnapshotId: snapshotId,
}),
now,
bill.id,
],
)
return {
bill: mapBill({ ...bill, ...updated.rows[0]!, amount_fen: amountFen }),
orderItemId: Number(bill.order_item_id),
snapshotId,
amountFen,
}
})
await Promise.all([
resolveExceptionsForItemsByReason([result.orderItemId], 'on_account_cost_pending'),
resolveExceptionsForItemsByReason([result.orderItemId], 'manual_cost_missing'),
resolveExceptionsForItemsByReason([result.orderItemId], 'cost_missing'),
]).catch(() => undefined)
await refreshFactsForOrderItem(result.orderItemId).catch(() => undefined)
return result
}