feat(finance): 完成财务控制、L1 钱包硬化与迁移收敛

- P0/P1:影子总账追加调整、关账/退款/资金/积分幂等、成本版本追溯、对账订单项维度
- 迁移收敛:原 103–114 合并为 103_finance_foundation、104_finance_controls、105_work_order_consignment
- 数据:L1 钱包 BIGINT/CHECK/奖励幂等;重复奖励审计口径修正与人工核对清单
- 报表:销售毛利缺失成本收入提示,旧收支页明确现金流口径
- 文档:同步迁移基线、合并评估、代码复核与上线说明
This commit is contained in:
yml2213
2026-10-27 16:08:16 +08:00
parent a53321ef30
commit 9d5af359fa
47 changed files with 4527 additions and 1410 deletions
@@ -263,6 +263,10 @@ function normalizePayeeType(value: unknown): string {
return normalized
}
function normalizeRequestId(value: unknown): string {
return text(value).slice(0, 80)
}
function normalizeReconciliationStatus(value: unknown): string {
const normalized = text(value)
if (!RECONCILIATION_STATUSES.has(normalized)) {
@@ -1215,34 +1219,46 @@ async function insertEntry(
evidence: JsonObject[]
note: string
createdAt: string
requestId?: string
},
) {
const result = await client.query<FundEntryRow>(
`
INSERT INTO fund_entries (
entry_no, account_id, entry_type, amount_fen, balance_after_fen,
transfer_no, counterparty_account_id, reconciliation_id, occurred_at,
created_by_admin_user_id, evidence_json, note, created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb, $12, $13)
RETURNING *
`,
[
generateNo('FE'),
input.accountId,
input.entryType,
input.amountFen,
input.balanceAfterFen,
input.transferNo || '',
input.counterpartyAccountId || null,
input.reconciliationId || null,
input.occurredAt,
input.actorUserId,
JSON.stringify(input.evidence),
input.note,
input.createdAt,
],
)
return result.rows[0]
try {
const result = await client.query<FundEntryRow>(
`
INSERT INTO fund_entries (
entry_no, account_id, entry_type, amount_fen, balance_after_fen,
transfer_no, counterparty_account_id, reconciliation_id, occurred_at,
created_by_admin_user_id, evidence_json, note, created_at, request_id
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb, $12, $13, $14)
RETURNING *
`,
[
generateNo('FE'),
input.accountId,
input.entryType,
input.amountFen,
input.balanceAfterFen,
input.transferNo || '',
input.counterpartyAccountId || null,
input.reconciliationId || null,
input.occurredAt,
input.actorUserId,
JSON.stringify(input.evidence),
input.note,
input.createdAt,
input.requestId || '',
],
)
return result.rows[0]
} catch (error) {
if ((error as { code?: string }).code === '23505') {
throw createHttpError('该请求已处理,请勿重复提交', {
statusCode: 409,
errorCode: 'fund_request_duplicate',
})
}
throw error
}
}
export async function createAdminFundFunding(payload: JsonObject, session: AdminSession) {
@@ -1259,6 +1275,7 @@ export async function createAdminFundFunding(payload: JsonObject, session: Admin
const evidence = normalizeEvidence(payload.evidence || payload.evidenceFiles)
const note = text(payload.note)
assertEvidenceOrNote(evidence, note)
const requestId = normalizeRequestId(payload.requestId ?? payload.request_id)
const now = nowIso()
return withTransaction(async (client) => {
@@ -1293,6 +1310,7 @@ export async function createAdminFundFunding(payload: JsonObject, session: Admin
evidence,
note,
createdAt: now,
requestId,
})
return {
@@ -1315,6 +1333,7 @@ export async function createAdminFundIssue(payload: JsonObject, session: AdminSe
const evidence = normalizeEvidence(payload.evidence || payload.evidenceFiles)
const note = text(payload.note)
assertEvidenceOrNote(evidence, note)
const requestId = normalizeRequestId(payload.requestId ?? payload.request_id)
const supervisorAccountId = await resolveSupervisorAccountId(payload, session)
const now = nowIso()
@@ -1374,6 +1393,7 @@ export async function createAdminFundIssue(payload: JsonObject, session: AdminSe
evidence,
note,
createdAt: now,
requestId,
})
const agentEntry = await insertEntry(client, {
accountId: agentAccountId,
@@ -1387,6 +1407,7 @@ export async function createAdminFundIssue(payload: JsonObject, session: AdminSe
evidence,
note,
createdAt: now,
requestId,
})
return {
@@ -1435,6 +1456,7 @@ export async function createAdminFundReturn(payload: JsonObject, session: AdminS
const evidence = normalizeEvidence(payload.evidence || payload.evidenceFiles)
const note = text(payload.note)
assertEvidenceOrNote(evidence, note)
const requestId = normalizeRequestId(payload.requestId ?? payload.request_id)
const agent = await selectAccountRow(agentAccountId)
if (!agent || agent.account_type !== 'agent') {
@@ -1513,6 +1535,7 @@ export async function createAdminFundReturn(payload: JsonObject, session: AdminS
evidence,
note,
createdAt: now,
requestId,
})
const supervisorEntry = await insertEntry(client, {
accountId: supervisorAccountId,
@@ -1526,6 +1549,7 @@ export async function createAdminFundReturn(payload: JsonObject, session: AdminS
evidence,
note,
createdAt: now,
requestId,
})
return {
@@ -1588,6 +1612,7 @@ export type FundExpenseWriteInput = {
note: string
actorUserId: number
allowDuplicate?: boolean
requestId?: string
}
/**
@@ -1667,6 +1692,7 @@ export async function writeFundExpense(
evidence: input.evidence,
note: input.note,
createdAt: now,
requestId: input.requestId || '',
})
const entryRow = requiredRow(entry, '写入支出流水失败')
@@ -1767,6 +1793,7 @@ export async function createAdminFundExpense(payload: JsonObject, session: Admin
}
const allowDuplicate = payload.allowDuplicate === true
const requestId = normalizeRequestId(payload.requestId ?? payload.request_id)
return withTransaction(async (client) => {
const written = await writeFundExpense(client, {
@@ -1785,6 +1812,7 @@ export async function createAdminFundExpense(payload: JsonObject, session: Admin
note,
actorUserId: session.userId,
allowDuplicate,
requestId,
})
return { expense: mapManualExpense(written.expense) }
})
@@ -2084,22 +2112,8 @@ export async function confirmAdminFundReconciliation(
})
}
const differenceFen = toNumber(reconciliation.difference_fen)
const reason = text(payload.reason) || reconciliation.reason
if (differenceFen !== 0) {
if (!hasFundPermission(session, 'fund.adjust')) {
throw createHttpError('有差异的对账需要财务或管理员调整', {
statusCode: 403,
errorCode: 'fund_adjust_forbidden',
})
}
if (!reason) {
throw createHttpError('有差异时必须填写差异原因', {
statusCode: 400,
errorCode: 'fund_reconciliation_reason_required',
})
}
}
const actualBalanceFen = toNumber(reconciliation.actual_balance_fen)
const now = nowIso()
return withTransaction(async (client) => {
@@ -2112,8 +2126,28 @@ export async function confirmAdminFundReconciliation(
})
}
// 关键:用“确认时点的当前余额”重算差额,而不是创建对账时的旧差额。
// 否则创建后发生的发放/支出/缴回会让调整后的余额仍然不等于实际盘点。
const currentBalanceFen = toNumber(account.balance_fen)
const differenceFen = actualBalanceFen - currentBalanceFen
if (differenceFen !== 0) {
const nextBalance = toNumber(account.balance_fen) + differenceFen
if (!hasFundPermission(session, 'fund.adjust')) {
throw createHttpError('有差异的对账需要财务或管理员调整', {
statusCode: 403,
errorCode: 'fund_adjust_forbidden',
})
}
if (!reason) {
throw createHttpError('有差异时必须填写差异原因', {
statusCode: 400,
errorCode: 'fund_reconciliation_reason_required',
})
}
}
if (differenceFen !== 0) {
const nextBalance = currentBalanceFen + differenceFen
if (nextBalance < 0) {
throw createHttpError('调整后余额不能为负', {
statusCode: 409,
@@ -2129,7 +2163,7 @@ export async function confirmAdminFundReconciliation(
entryType: 'adjustment',
amountFen: differenceFen,
balanceAfterFen: nextBalance,
reconciliationId: reconciliationId,
reconciliationId,
occurredAt: now,
actorUserId: session.userId,
evidence: normalizeEvidence(payload.evidence || reconciliation.evidence_json),
@@ -2141,11 +2175,16 @@ export async function confirmAdminFundReconciliation(
await client.query(
`
UPDATE fund_reconciliations
SET status = 'confirmed', reason = $1, confirmed_by_admin_user_id = $2,
confirmed_at = $3, updated_at = $3
SET status = 'confirmed',
reason = $1,
system_balance_fen = $5,
difference_fen = $6,
confirmed_by_admin_user_id = $2,
confirmed_at = $3,
updated_at = $3
WHERE id = $4
`,
[reason, session.userId, now, reconciliationId],
[reason, session.userId, now, reconciliationId, currentBalanceFen, differenceFen],
)
const updated = await client.query<ReconciliationRow>(
@@ -6,6 +6,9 @@ export const ADMIN_PERMISSION_KEYS = [
'finance.review',
'finance.config',
'finance.export',
'finance.post',
'finance.close',
'finance.reconcile',
'fund.view',
'fund.issue',
'fund.expense',
@@ -342,18 +342,49 @@ export async function setAdminAffiliateProductPrice(
}
}
// 2. 追溯改价或相同生效时间改价:直接就地更新当前 active 版本,避免破坏 effective_to > effective_from 约束或触发唯一索引冲突
if (existing && existingTime !== null && targetTime <= existingTime) {
// 2. 追溯改价:退休旧版本并新增一个从 targetTime 起生效的版本,旧行保留可追溯。
if (existing && existingTime !== null && targetTime < existingTime) {
await query("UPDATE cost_versions SET status = 'retired', updated_at = $1 WHERE id = $2", [
now,
existing.id,
])
const inserted = await query<{ id: number }>(
`
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
) VALUES (
$1, 'product', '', '', $2,
NULL, $3, 0, $4, NULL,
'manual', 'active', $5, $6, $6, $7, $7
)
RETURNING id
`,
[costItem.id, product.id, amountFen, effectiveFrom, changeReason, session.userId, now],
)
return {
success: true,
product,
costVersionId: Number(inserted.rows[0]?.id || 0),
amountFen,
effectiveFrom,
message: '采购成本已更新(新增追溯版本)',
}
}
// 2.1 相同生效时间的更正:没有可插入的时间区间,只能就地更正并保留变更原因。
if (existing && existingTime !== null && targetTime === existingTime) {
await query(
`UPDATE cost_versions
SET amount_fen = $1,
effective_from = $2,
change_reason = $3,
created_by_admin_user_id = $4,
approved_by_admin_user_id = $4,
updated_at = $5
WHERE id = $6`,
[amountFen, effectiveFrom, changeReason, session.userId, now, existing.id],
change_reason = $2,
created_by_admin_user_id = $3,
approved_by_admin_user_id = $3,
updated_at = $4
WHERE id = $5`,
[amountFen, changeReason, session.userId, now, existing.id],
)
return {
success: true,
@@ -178,6 +178,7 @@ export async function grantPlatformCredit(payload: JsonObject, session: AdminSes
const quantity = normalizePositiveQuantity(payload.quantity)
const platformRef = optionalText(payload.platformRef ?? payload.platform_ref, 100)
const note = optionalText(payload.note, 300)
const requestId = optionalText(payload.requestId ?? payload.request_id, 80)
const unitPrices = await loadUnitPrices()
const unitPriceFen = unitPrices[platformKey]
@@ -194,16 +195,36 @@ export async function grantPlatformCredit(payload: JsonObject, session: AdminSes
WHERE id = $1`,
[account.id, nextBalance, now],
)
const entryResult = await client.query<CreditEntryRow>(
`
INSERT INTO platform_credit_entries (
account_id, entry_type, quantity_delta, balance_after, amount_fen,
platform_ref, note, created_at
) VALUES ($1, 'grant', $2, $3, $4, $5, $6, $7)
RETURNING *
`,
[account.id, quantity, nextBalance, quantity * unitPriceFen, platformRef, note, now],
)
let entryResult
try {
entryResult = await client.query<CreditEntryRow>(
`
INSERT INTO platform_credit_entries (
account_id, entry_type, quantity_delta, balance_after, amount_fen,
platform_ref, note, request_id, created_at
) VALUES ($1, 'grant', $2, $3, $4, $5, $6, $7, $8)
RETURNING *
`,
[
account.id,
quantity,
nextBalance,
quantity * unitPriceFen,
platformRef,
note,
requestId,
now,
],
)
} catch (error) {
if ((error as { code?: string }).code === '23505') {
throw createHttpError('该请求已处理,请勿重复提交', {
statusCode: 409,
errorCode: 'platform_credit_request_duplicate',
})
}
throw error
}
return {
accountId: Number(account.id),
platformKey,
@@ -236,6 +257,7 @@ export async function adjustPlatformCredit(payload: JsonObject, session: AdminSe
errorCode: 'platform_credit_adjust_note_required',
})
}
const requestId = optionalText(payload.requestId ?? payload.request_id, 80)
const unitPrices = await loadUnitPrices()
const unitPriceFen = unitPrices[platformKey]
const now = nowIso()
@@ -256,16 +278,27 @@ export async function adjustPlatformCredit(payload: JsonObject, session: AdminSe
WHERE id = $1`,
[account.id, nextBalance, now],
)
const entryResult = await client.query<CreditEntryRow>(
`
INSERT INTO platform_credit_entries (
account_id, entry_type, quantity_delta, balance_after, amount_fen,
note, created_at
) VALUES ($1, 'adjust', $2, $3, $4, $5, $6)
RETURNING *
`,
[account.id, quantity, nextBalance, quantity * unitPriceFen, note, now],
)
let entryResult
try {
entryResult = await client.query<CreditEntryRow>(
`
INSERT INTO platform_credit_entries (
account_id, entry_type, quantity_delta, balance_after, amount_fen,
note, request_id, created_at
) VALUES ($1, 'adjust', $2, $3, $4, $5, $6, $7)
RETURNING *
`,
[account.id, quantity, nextBalance, quantity * unitPriceFen, note, requestId, now],
)
} catch (error) {
if ((error as { code?: string }).code === '23505') {
throw createHttpError('该请求已处理,请勿重复提交', {
statusCode: 409,
errorCode: 'platform_credit_request_duplicate',
})
}
throw error
}
return {
accountId: Number(account.id),
platformKey,
@@ -1391,30 +1391,49 @@ export async function setAdminProductPrice(
let costVersionId = existing?.id || 0
if (existing && existingTime !== null && targetTime <= existingTime) {
// 追溯改价或相同生效时间改价:就地更新 active 版本
await query(
`UPDATE cost_versions
SET amount_fen = $1,
effective_from = $2,
canonical_product_id = $3,
work_product_rule_id = $4,
change_reason = $5,
created_by_admin_user_id = $6,
approved_by_admin_user_id = $6,
updated_at = $7
WHERE id = $8`,
if (existing && existingTime !== null && targetTime < existingTime) {
// 追溯改价:退休旧版本并新增一个从 targetTime 起生效的版本,旧行保留可追溯。
// 不能就地改金额,否则 cost_snapshots.cost_version_id 指向的版本内容会被篡改。
await query("UPDATE cost_versions SET status = 'retired', updated_at = $1 WHERE id = $2", [
now,
existing.id,
])
const inserted = await query<{ id: number }>(
`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
) VALUES (
$1, 'rule', '', '', $2,
$3, $4, 0, $5, NULL,
'manual', 'active', $6, $7,
$7, $8, $8
) RETURNING id`,
[
amountFen,
effectiveFrom,
costItemId,
product.id,
rule?.id ?? null,
amountFen,
effectiveFrom,
changeReason,
session.userId,
now,
existing.id,
],
)
costVersionId = inserted.rows[0]?.id || 0
} else if (existing && existingTime !== null && targetTime === existingTime) {
// 相同生效时间的更正:没有可插入的时间区间,只能就地更正并保留变更原因。
await query(
`UPDATE cost_versions
SET amount_fen = $1,
change_reason = $2,
created_by_admin_user_id = $3,
approved_by_admin_user_id = $3,
updated_at = $4
WHERE id = $5`,
[amountFen, changeReason, session.userId, now, existing.id],
)
} else {
if (existing) {
await query(
@@ -1510,17 +1529,36 @@ export async function setAdminProductPrice(
const targetTime = Date.parse(effectiveFrom)
let costVersionId = existing?.id || 0
if (existing && existingTime !== null && targetTime <= existingTime) {
if (existing && existingTime !== null && targetTime < existingTime) {
await query("UPDATE cost_versions SET status = 'retired', updated_at = $1 WHERE id = $2", [
now,
existing.id,
])
const inserted = await query<{ id: number }>(
`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
) VALUES (
$1, 'product', '', '', $2,
NULL, $3, 0, $4, NULL,
'manual', 'active', $5, $6,
$6, $7, $7
) RETURNING id`,
[costItemId, product.id, amountFen, effectiveFrom, changeReason, session.userId, now],
)
costVersionId = inserted.rows[0]?.id || 0
} else if (existing && existingTime !== null && targetTime === existingTime) {
await query(
`UPDATE cost_versions
SET amount_fen = $1,
effective_from = $2,
change_reason = $3,
created_by_admin_user_id = $4,
approved_by_admin_user_id = $4,
updated_at = $5
WHERE id = $6`,
[amountFen, effectiveFrom, changeReason, session.userId, now, existing.id],
change_reason = $2,
created_by_admin_user_id = $3,
approved_by_admin_user_id = $3,
updated_at = $4
WHERE id = $5`,
[amountFen, changeReason, session.userId, now, existing.id],
)
} else {
if (existing) {
@@ -0,0 +1,154 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
const read = (path: URL) => readFile(path, 'utf8')
test('销售事实重算必须冻结已关账期间', async () => {
const source = await read(new URL('./sales-facts-service.ts', import.meta.url))
assert.ok(source.includes('finance_periods'), '重算必须读取会计期间')
assert.ok(source.includes("p.status = 'closed'"), '必须识别已关账期间')
assert.ok(source.includes('skippedClosedPeriods'), '必须返回被跳过的关账期间')
})
test('人工确认/退款不得静默改已关账期间', async () => {
const [manual, refund] = await Promise.all([
read(new URL('./manual-fulfillment-service.ts', import.meta.url)),
read(new URL('./order-refund-service.ts', import.meta.url)),
])
assert.ok(manual.includes('manual_fulfillment_period_closed'))
assert.ok(refund.includes('order_refund_period_closed'))
})
test('总账同步必须追加调整凭证,不能删除重建', async () => {
const source = await read(new URL('./finance-ledger-service.ts', import.meta.url))
assert.ok(!source.includes('DELETE FROM finance_journal_entries'), '已过账凭证不得删除重建')
assert.ok(source.includes('sales_revenue_adjustment'), '收入变化必须走调整凭证')
assert.ok(source.includes('sales_cost_adjustment'), '成本变化必须走调整凭证')
assert.ok(source.includes('finance_sales_sync:'), '按日同步必须加 advisory lock')
})
test('关账必须具备门槛、期间锁和重复关账保护', async () => {
const source = await read(new URL('./finance-ledger-service.ts', import.meta.url))
assert.ok(source.includes('assertPeriodCloseable'), '关账必须校验门槛')
assert.ok(source.includes('finance_period_close_blocked'), '关账失败必须给出明确错误码')
assert.ok(source.includes('finance_period_already_closed'), '重复关账必须拒绝')
assert.ok(source.includes('finance_period:'), '凭证写入与关账必须共用期间锁')
assert.ok(
source.includes('manual_cost_missing') || source.includes('未确认成本'),
'必须拦人工缺成本',
)
})
test('平台退款必须先校验可退金额并落退款意图', async () => {
const source = await read(new URL('./order-refund-service.ts', import.meta.url))
const platformStart = source.indexOf('export async function approveAndRecordKuaishouRefund')
const platformEnd = source.indexOf('type PlatformRefundIntentRow', platformStart)
const body = source.slice(platformStart, platformEnd)
assert.ok(
body.indexOf('getRefundableFen') < body.indexOf('approveKuaishouIndustryRefund'),
'可退金额校验必须发生在平台调用之前',
)
assert.ok(body.includes('beginPlatformRefundIntent'), '必须先落退款意图')
assert.ok(source.includes('finance_refund_intents'), '必须有平台退款状态机')
assert.ok(source.includes('order_refund_external_ref_exists'), '外部退款单号必须幂等')
assert.ok(source.includes('nextPayStatus'), '全额退款撤销必须恢复订单支付状态')
assert.ok(source.includes('订单已退款,成本异常自动关闭'), '撤销必须重新打开成本异常')
})
test('备用金对账确认必须按当前余额重算差额', async () => {
const source = await read(new URL('../admin/admin-fund-service.ts', import.meta.url))
const start = source.indexOf('export async function confirmAdminFundReconciliation')
const end = source.indexOf('export async function getAdminFundSupervisorOptions', start)
const body = source.slice(start, end)
assert.ok(body.includes('actualBalanceFen - currentBalanceFen'), '必须用确认时点余额重算')
assert.ok(body.includes('system_balance_fen = $5'), '必须回写确认时点的系统余额')
assert.ok(body.includes('difference_fen = $6'), '必须回写确认时点的差额')
})
test('迁移 104 提供退款/支出幂等与凭证完整性约束', async () => {
const migration = await read(
new URL('../../db/migrations/104_finance_controls.sql', import.meta.url),
)
assert.ok(migration.includes('uq_order_refunds_source_external_ref'))
assert.ok(migration.includes('uq_manual_expenses_order_item'))
assert.ok(migration.includes('finance_refund_intents'))
assert.ok(migration.includes('trg_finance_journal_balanced'))
assert.ok(migration.includes('trg_finance_journal_immutable'))
assert.ok(migration.includes('trg_fund_entries_balance'))
assert.ok(migration.includes('trg_fund_accounts_balance'))
assert.ok(migration.includes("'finance.post'"))
assert.ok(migration.includes("'finance.close'"))
assert.ok(migration.includes("'finance.reconcile'"))
})
test('财务写路由使用拆分后的权限,人工履约包含 finance', async () => {
const [ledgerRoute, financeRoute, orderRoute] = await Promise.all([
read(new URL('../../routes/admin/finance-ledger.ts', import.meta.url)),
read(new URL('../../routes/admin/finance.ts', import.meta.url)),
read(new URL('../../routes/admin/orders.ts', import.meta.url)),
])
assert.ok(ledgerRoute.includes("requireAdminPermission('finance.post')"))
assert.ok(ledgerRoute.includes("requireAdminPermission('finance.close')"))
assert.ok(ledgerRoute.includes("requireAdminPermission('finance.reconcile')"))
assert.ok(
financeRoute.includes("requireAdminRoles(['admin', 'operator', 'support', 'finance'])"),
'文档承诺财务可代录,路由必须允许 finance',
)
assert.ok(orderRoute.includes("requireAdminPermission('finance.post')"))
})
test('P1:资金/积分写命令支持 requestId 幂等', async () => {
const [fund, credit, migration] = await Promise.all([
read(new URL('../admin/admin-fund-service.ts', import.meta.url)),
read(new URL('../cost-center/platform-credit-service.ts', import.meta.url)),
read(new URL('../../db/migrations/104_finance_controls.sql', import.meta.url)),
])
assert.ok(fund.includes('requestId') && fund.includes('fund_request_duplicate'))
assert.ok(credit.includes('platform_credit_request_duplicate'))
assert.ok(migration.includes('uq_fund_entries_account_request'))
assert.ok(migration.includes('uq_platform_credit_entries_account_request'))
})
test('P1:成本版本追溯改价退休旧版本而不是覆盖金额', async () => {
const [product, affiliate] = await Promise.all([
read(new URL('../cost-center/product-catalog-service.ts', import.meta.url)),
read(new URL('../cost-center/affiliate-product-service.ts', import.meta.url)),
])
assert.ok(product.includes('新增一个从 targetTime 起生效的版本'))
assert.ok(product.includes("status = 'retired'"))
assert.ok(affiliate.includes('新增追溯版本'))
assert.ok(affiliate.includes("status = 'retired'"))
})
test('P1:平台对账支持订单项/平台明细维度', async () => {
const [service, migration] = await Promise.all([
read(new URL('./finance-reconciliation-service.ts', import.meta.url)),
read(new URL('../../db/migrations/104_finance_controls.sql', import.meta.url)),
])
assert.ok(service.includes('orderItemId') && service.includes('platformItemId'))
assert.ok(service.includes('order_item_id'))
assert.ok(migration.includes('platform_item_id'))
assert.ok(migration.includes('order_item_id'))
})
test('L1:钱包金额 BIGINT、非负约束与奖励流水幂等', async () => {
const migration = await read(
new URL('../../db/migrations/104_finance_controls.sql', import.meta.url),
)
assert.ok(migration.includes('ALTER COLUMN total_amount TYPE BIGINT'))
assert.ok(migration.includes('ALTER COLUMN available_amount TYPE BIGINT'))
assert.ok(migration.includes('chk_worker_wallets_nonnegative'))
assert.ok(migration.includes('chk_worker_wallet_ledgers_nonnegative_after'))
assert.ok(migration.includes('worker_wallet_ledger_event_key'))
assert.ok(migration.includes('uq_worker_wallet_ledger_event_key'))
assert.ok(migration.includes('v_worker_wallet_duplicate_rewards'))
assert.ok(migration.includes("NOT (NEW.payload_json ? 'inventoryId')"))
assert.ok(migration.includes('GROUP BY worker_id, related_work_order_id'))
})
test('销售毛利必须同时暴露缺失成本收入,避免只看毛利绝对值', async () => {
const source = await read(new URL('./sales-facts-service.ts', import.meta.url))
assert.ok(source.includes('cost_missing_revenue_fen'))
assert.ok(source.includes('costMissingRevenueFen'))
})
@@ -34,6 +34,8 @@ export type PostJournalInput = {
memo?: string
metadata?: JsonObject
lines: JournalLineInput[]
/** 允许命中已有幂等键时直接返回(默认抛 409,避免“以为新建了、实际复用了旧凭证”)。 */
allowExisting?: boolean
}
export type JournalEntry = {
@@ -68,6 +70,7 @@ export async function postJournalEntry(
const occurredDate = normalizeTimestampDate(input.occurredAt)
const occurredAt = occurredDate.toISOString()
const periodKey = formatShanghaiDate(occurredDate).slice(0, 7)
await acquireFinancePeriodLock(client, periodKey)
await assertPeriodOpen(client, periodKey)
const entryNo = `FJ-${periodKey.replace('-', '')}-${randomId('', 8)}`
@@ -100,6 +103,12 @@ export async function postJournalEntry(
LIMIT 1`,
[input.eventType, input.refType, input.refId],
)
if (!input.allowExisting) {
throw createHttpError('该业务事件已生成过凭证,不能重复记账', {
statusCode: 409,
errorCode: 'finance_journal_duplicate_reference',
})
}
return {
created: false,
entryId: Number(existing.rows[0]?.id || 0),
@@ -189,81 +198,179 @@ export async function syncFinanceLedgerFromFacts(input: { dateFrom: string; date
)
let created = 0
let replaced = 0
let adjusted = 0
let skipped = 0
for (const row of rows.rows) {
const occurredAt = `${row.stat_date}T12:00:00+08:00`
const periodKey = formatShanghaiDate(new Date(occurredAt)).slice(0, 7)
const period = await query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
[periodKey],
)
if (period.rows[0]?.status === 'closed') {
// 已关账期间不可改账;即使销售事实后来变化,也保留关账时点的凭证。
skipped += 1
continue
}
// 未关账期间:销售事实可能因退款/收入回填而变化,按日整体重建,
// 避免旧的幂等凭证一直停留在旧金额;删除与重建放在同一事务内。
const revenue = Number(row.gross_revenue_fen || 0)
const refund = Number(row.refund_amount_fen || 0)
const cost = Number(row.cost_amount_fen || 0)
const entries: PostJournalInput[] = []
if (revenue > 0) {
entries.push({
eventType: 'sales_revenue',
refType: 'sales_fact',
refId: row.stat_date,
occurredAt,
memo: `${row.stat_date} 销售收入`,
lines: [
{ accountCode: '1101', debitFen: revenue, memo: '应收平台' },
{ accountCode: '4001', creditFen: revenue, memo: '商品销售收入' },
],
})
}
if (refund > 0) {
entries.push({
eventType: 'sales_refund',
refType: 'sales_fact',
refId: row.stat_date,
occurredAt,
memo: `${row.stat_date} 退款冲减`,
lines: [
{ accountCode: '4101', debitFen: refund, memo: '退款冲减收入' },
{ accountCode: '1101', creditFen: refund, memo: '应收平台冲减' },
],
})
}
if (cost > 0) {
entries.push({
eventType: 'sales_cost',
refType: 'sales_fact',
refId: row.stat_date,
occurredAt,
memo: `${row.stat_date} 履约成本`,
lines: [
{ accountCode: '5001', debitFen: cost, memo: '履约成本' },
{ accountCode: '2001', creditFen: cost, memo: '应付履约款' },
],
})
}
const dayCreated = await withTransaction(async (client) => {
const deleted = await client.query(
`DELETE FROM finance_journal_entries WHERE ref_type = 'sales_fact' AND ref_id = $1`,
// 按日串行化:避免两个同步任务同时算出同一个 delta 并重复补记。
// 锁在事务内,事务提交/回滚自动释放。
const dayResult = await withTransaction(async (client) => {
await client.query(
`SELECT pg_advisory_xact_lock(hashtext('finance_sales_sync:' || $1)::bigint)`,
[row.stat_date],
)
replaced += deleted.rowCount || 0
let count = 0
for (const entry of entries) {
const result = await postJournalEntry(entry, client)
if (result.created) count += 1
const period = await client.query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
[periodKey],
)
if (period.rows[0]?.status === 'closed') {
// 已关账期间不可改账;销售事实层也已在 rebuild 时冻结该期间。
return { created: 0, adjusted: 0, skipped: 1 }
}
return count
const existing = await readSalesFactJournalTotals(client, row.stat_date)
let dayCreated = 0
let dayAdjusted = 0
const postDelta = async (input: {
eventType: string
adjustmentType: string
hasExisting: boolean
delta: number
normalLines: (amount: number) => JournalLineInput[]
reverseLines: (amount: number) => JournalLineInput[]
memo: string
}) => {
if (input.delta === 0) return
const isAdjustment = input.hasExisting
const eventType = isAdjustment ? input.adjustmentType : input.eventType
const refId = isAdjustment
? `${row.stat_date}:${input.adjustmentType}:${randomId('', 6)}`
: row.stat_date
const amount = Math.abs(input.delta)
const result = await postJournalEntry(
{
eventType,
refType: 'sales_fact',
refId,
occurredAt,
memo: input.memo,
metadata: {
statDate: row.stat_date,
deltaFen: input.delta,
adjustment: isAdjustment,
},
lines: input.delta > 0 ? input.normalLines(amount) : input.reverseLines(amount),
},
client,
)
if (result.created) {
if (isAdjustment) dayAdjusted += 1
else dayCreated += 1
}
}
await postDelta({
eventType: 'sales_revenue',
adjustmentType: 'sales_revenue_adjustment',
hasExisting: existing.hasRevenue,
delta: revenue - existing.revenue,
normalLines: (amount) => [
{ accountCode: '1101', debitFen: amount, memo: '应收平台' },
{ accountCode: '4001', creditFen: amount, memo: '商品销售收入' },
],
reverseLines: (amount) => [
{ accountCode: '4001', debitFen: amount, memo: '商品销售收入冲减' },
{ accountCode: '1101', creditFen: amount, memo: '应收平台冲减' },
],
memo: `${row.stat_date} 销售收入`,
})
await postDelta({
eventType: 'sales_refund',
adjustmentType: 'sales_refund_adjustment',
hasExisting: existing.hasRefund,
delta: refund - existing.refund,
normalLines: (amount) => [
{ accountCode: '4101', debitFen: amount, memo: '退款冲减收入' },
{ accountCode: '1101', creditFen: amount, memo: '应收平台冲减' },
],
reverseLines: (amount) => [
{ accountCode: '1101', debitFen: amount, memo: '应收平台恢复' },
{ accountCode: '4101', creditFen: amount, memo: '退款冲减恢复' },
],
memo: `${row.stat_date} 退款冲减`,
})
await postDelta({
eventType: 'sales_cost',
adjustmentType: 'sales_cost_adjustment',
hasExisting: existing.hasCost,
delta: cost - existing.cost,
normalLines: (amount) => [
{ accountCode: '5001', debitFen: amount, memo: '履约成本' },
{ accountCode: '2001', creditFen: amount, memo: '应付履约款' },
],
reverseLines: (amount) => [
{ accountCode: '2001', debitFen: amount, memo: '应付履约款冲减' },
{ accountCode: '5001', creditFen: amount, memo: '履约成本冲减' },
],
memo: `${row.stat_date} 履约成本`,
})
return { created: dayCreated, adjusted: dayAdjusted, skipped: 0 }
})
created += dayCreated
created += dayResult.created
adjusted += dayResult.adjusted
skipped += dayResult.skipped
}
return {
range: { fromDate: dateFrom, toDate: dateTo },
created,
adjusted,
skipped,
}
}
type SalesFactJournalTotalRow = {
event_type: string
account_code: string
debit_fen: string | number
credit_fen: string | number
}
async function readSalesFactJournalTotals(client: PoolClient, statDate: string) {
const result = await client.query<SalesFactJournalTotalRow>(
`
SELECT
e.event_type,
l.account_code,
COALESCE(SUM(l.debit_fen), 0)::bigint AS debit_fen,
COALESCE(SUM(l.credit_fen), 0)::bigint AS credit_fen
FROM finance_journal_entries e
JOIN finance_journal_lines l ON l.entry_id = e.id
WHERE e.ref_type = 'sales_fact'
AND e.ref_id = $1
AND e.status = 'posted'
GROUP BY e.event_type, l.account_code
`,
[statDate],
)
const rows = result.rows
const revenueTypes = new Set(['sales_revenue', 'sales_revenue_adjustment'])
const refundTypes = new Set(['sales_refund', 'sales_refund_adjustment'])
const costTypes = new Set(['sales_cost', 'sales_cost_adjustment'])
const net = (types: Set<string>, accountCode: string) =>
rows
.filter((row) => types.has(row.event_type) && row.account_code === accountCode)
.reduce((sum, row) => sum + Number(row.debit_fen || 0) - Number(row.credit_fen || 0), 0)
return {
revenue: net(revenueTypes, '1101'),
refund: net(refundTypes, '4101'),
cost: net(costTypes, '5001'),
hasRevenue: rows.some((row) => revenueTypes.has(row.event_type)),
hasRefund: rows.some((row) => refundTypes.has(row.event_type)),
hasCost: rows.some((row) => costTypes.has(row.event_type)),
}
return { range: { fromDate: dateFrom, toDate: dateTo }, created, replaced, skipped }
}
export async function listJournalEntries(input: Record<string, unknown> = {}) {
@@ -349,11 +456,13 @@ export async function listJournalEntries(input: Record<string, unknown> = {}) {
}
}
export async function getTrialBalance(input: { dateFrom: string; dateTo: string }) {
export async function getTrialBalance(
input: { dateFrom: string; dateTo: string },
client?: PoolClient,
) {
const dateFrom = normalizeDate(input.dateFrom)
const dateTo = normalizeDate(input.dateTo)
const rows = await query<Record<string, unknown>>(
`
const sql = `
SELECT
a.account_code, a.account_name, a.account_type, a.normal_side,
COALESCE(SUM(l.debit_fen), 0)::bigint AS debit_fen,
@@ -373,10 +482,11 @@ export async function getTrialBalance(input: { dateFrom: string; dateTo: string
AND (e.id IS NOT NULL OR (l.id IS NULL))
GROUP BY a.account_code, a.account_name, a.account_type, a.normal_side
ORDER BY a.account_code
`,
[dateFrom, dateTo],
)
const items = rows.rows.map((row) => ({
`
const result = client
? await client.query<Record<string, unknown>>(sql, [dateFrom, dateTo])
: await query<Record<string, unknown>>(sql, [dateFrom, dateTo])
const items = result.rows.map((row) => ({
accountCode: String(row.account_code || ''),
accountName: String(row.account_name || ''),
accountType: String(row.account_type || ''),
@@ -410,42 +520,254 @@ export async function listFinancePeriods() {
}
}
export async function closeFinancePeriod(periodKeyInput: unknown, session: AdminSession) {
export async function closeFinancePeriod(
periodKeyInput: unknown,
session: AdminSession,
reason = '',
) {
const periodKey = normalizePeriodKey(periodKeyInput)
const existing = await query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
[periodKey],
)
if (existing.rows[0]?.status === 'closed') {
throw createHttpError(`会计期间 ${periodKey} 已关账,不能重复关账`, {
statusCode: 409,
errorCode: 'finance_period_already_closed',
})
}
const { startsAt, endsAt, dateFrom, dateTo } = periodRange(periodKey)
// 先补齐该期间的凭证,避免漏记。
await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
const trial = await getTrialBalance({ dateFrom, dateTo })
const now = nowIso()
await query(
const result = await withTransaction(async (client) => {
// 与 postJournalEntry 使用同一把期间锁:关账检查与凭证写入不能并发穿透。
await acquireFinancePeriodLock(client, periodKey)
const current = await client.query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
[periodKey],
)
if (current.rows[0]?.status === 'closed') {
throw createHttpError(`会计期间 ${periodKey} 已关账,不能重复关账`, {
statusCode: 409,
errorCode: 'finance_period_already_closed',
})
}
const trial = await getTrialBalance({ dateFrom, dateTo }, client)
await assertPeriodCloseable(client, periodKey, dateFrom, dateTo, trial)
const now = nowIso()
await client.query(
`
INSERT INTO finance_periods (
period_key, starts_at, ends_at, status, summary_json,
closed_at, closed_by_admin_user_id, created_at, updated_at
) VALUES ($1, $2, $3, 'closed', $4::jsonb, $5, $6, $5, $5)
ON CONFLICT (period_key) DO UPDATE
SET status = 'closed',
starts_at = EXCLUDED.starts_at,
ends_at = EXCLUDED.ends_at,
summary_json = EXCLUDED.summary_json,
closed_at = EXCLUDED.closed_at,
closed_by_admin_user_id = EXCLUDED.closed_by_admin_user_id,
updated_at = EXCLUDED.updated_at
`,
[
periodKey,
startsAt,
endsAt,
JSON.stringify({
totals: trial.totals,
balanced: trial.balanced,
itemCount: trial.items.length,
closeReason: String(reason || '').slice(0, 500),
}),
now,
Number(session.userId || 0) || null,
],
)
return { trial }
})
return { periodKey, trial: result.trial }
}
async function assertPeriodCloseable(
client: PoolClient,
periodKey: string,
dateFrom: string,
dateTo: string,
trial: { balanced: boolean },
) {
const issues: string[] = []
if (!trial.balanced) {
issues.push('凭证借贷不平衡')
}
// 上线后人工单必须全覆盖:从第一次人工配置起,未确认的人工订单项不允许关账。
const manualMissing = await client.query<{ total: number }>(
`
INSERT INTO finance_periods (
period_key, starts_at, ends_at, status, summary_json,
closed_at, closed_by_admin_user_id, created_at, updated_at
) VALUES ($1, $2, $3, 'closed', $4::jsonb, $5, $6, $5, $5)
ON CONFLICT (period_key) DO UPDATE
SET status = 'closed',
starts_at = EXCLUDED.starts_at,
ends_at = EXCLUDED.ends_at,
summary_json = EXCLUDED.summary_json,
closed_at = EXCLUDED.closed_at,
closed_by_admin_user_id = EXCLUDED.closed_by_admin_user_id,
updated_at = EXCLUDED.updated_at
SELECT COUNT(*)::int AS total
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_mode = 'manual'
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND oi.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs)
AND NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.subject_type = 'manual_case'
AND cs.order_item_id = oi.id
AND cs.superseded_at IS NULL
)
`,
[
periodKey,
startsAt,
endsAt,
JSON.stringify({
totals: trial.totals,
balanced: trial.balanced,
itemCount: trial.items.length,
}),
now,
Number(session.userId || 0) || null,
],
[dateFrom, dateTo],
)
return { periodKey, trial }
if (Number(manualMissing.rows[0]?.total || 0) > 0) {
issues.push(`存在 ${Number(manualMissing.rows[0]?.total)} 个人工订单项未确认成本`)
}
// 人工成本必须能找到资金流水:现金 fundExpenseId 或积分 creditEntryId。
const manualWithoutFund = await client.query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM cost_snapshots cs
JOIN order_items oi ON oi.id = cs.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE cs.subject_type = 'manual_case'
AND cs.superseded_at IS NULL
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND cs.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs)
AND cs.metadata_json->>'fundExpenseId' IS NULL
AND cs.metadata_json->>'creditEntryId' IS NULL
`,
[dateFrom, dateTo],
)
if (Number(manualWithoutFund.rows[0]?.total || 0) > 0) {
issues.push(`存在 ${Number(manualWithoutFund.rows[0]?.total)} 个人工成本快照没有资金流水`)
}
// 现金人工成本金额必须等于备用金支出金额。
const manualAmountMismatch = await client.query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM cost_snapshots cs
JOIN manual_expenses me ON me.order_item_id = cs.order_item_id
JOIN order_items oi ON oi.id = cs.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE cs.subject_type = 'manual_case'
AND cs.superseded_at IS NULL
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND cs.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs)
AND cs.total_amount_fen <> me.amount_fen
`,
[dateFrom, dateTo],
)
if (Number(manualAmountMismatch.rows[0]?.total || 0) > 0) {
issues.push(
`存在 ${Number(manualAmountMismatch.rows[0]?.total)} 个人工成本与备用金支出金额不一致`,
)
}
// 未解决的平台对账差异不允许关账。
const unresolvedReconciliations = await client.query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM finance_reconciliations
WHERE status = 'differences'
AND period_from <= $2::date
AND period_to >= $1::date
`,
[dateFrom, dateTo],
)
if (Number(unresolvedReconciliations.rows[0]?.total || 0) > 0) {
issues.push(`存在 ${Number(unresolvedReconciliations.rows[0]?.total)} 个未解决的对账差异批次`)
}
// 账表一致:销售事实与 sales_fact 凭证必须逐项相等,防止事实层被并发重算后总账未跟上。
const ledgerVsFacts = await client.query<{
facts_revenue_fen: string
facts_refund_fen: string
facts_cost_fen: string
ledger_revenue_fen: string
ledger_refund_fen: string
ledger_cost_fen: string
}>(
`
WITH facts AS (
SELECT
COALESCE(SUM(gross_revenue_fen), 0)::bigint AS revenue_fen,
COALESCE(SUM(refund_amount_fen), 0)::bigint AS refund_fen,
COALESCE(SUM(cost_amount_fen), 0)::bigint AS cost_fen
FROM sales_daily_facts
WHERE stat_date BETWEEN $1::date AND $2::date
),
ledger AS (
SELECT
COALESCE(SUM(l.debit_fen) FILTER (
WHERE l.account_code = '1101'
AND e.event_type IN ('sales_revenue', 'sales_revenue_adjustment')
), 0)::bigint
- COALESCE(SUM(l.credit_fen) FILTER (
WHERE l.account_code = '1101'
AND e.event_type IN ('sales_revenue', 'sales_revenue_adjustment')
), 0)::bigint AS revenue_fen,
COALESCE(SUM(l.debit_fen) FILTER (
WHERE l.account_code = '4101'
AND e.event_type IN ('sales_refund', 'sales_refund_adjustment')
), 0)::bigint
- COALESCE(SUM(l.credit_fen) FILTER (
WHERE l.account_code = '4101'
AND e.event_type IN ('sales_refund', 'sales_refund_adjustment')
), 0)::bigint AS refund_fen,
COALESCE(SUM(l.debit_fen) FILTER (
WHERE l.account_code = '5001'
AND e.event_type IN ('sales_cost', 'sales_cost_adjustment')
), 0)::bigint
- COALESCE(SUM(l.credit_fen) FILTER (
WHERE l.account_code = '5001'
AND e.event_type IN ('sales_cost', 'sales_cost_adjustment')
), 0)::bigint AS cost_fen
FROM finance_journal_entries e
JOIN finance_journal_lines l ON l.entry_id = e.id
WHERE e.ref_type = 'sales_fact'
AND e.status = 'posted'
AND e.occurred_at >= ($1::date::timestamp AT TIME ZONE 'Asia/Shanghai')
AND e.occurred_at < (($2::date + interval '1 day')::timestamp AT TIME ZONE 'Asia/Shanghai')
)
SELECT
facts.revenue_fen AS facts_revenue_fen,
facts.refund_fen AS facts_refund_fen,
facts.cost_fen AS facts_cost_fen,
ledger.revenue_fen AS ledger_revenue_fen,
ledger.refund_fen AS ledger_refund_fen,
ledger.cost_fen AS ledger_cost_fen
FROM facts, ledger
`,
[dateFrom, dateTo],
)
const compare = ledgerVsFacts.rows[0]
if (
compare &&
(Number(compare.facts_revenue_fen) !== Number(compare.ledger_revenue_fen) ||
Number(compare.facts_refund_fen) !== Number(compare.ledger_refund_fen) ||
Number(compare.facts_cost_fen) !== Number(compare.ledger_cost_fen))
) {
issues.push('销售事实与总账凭证不一致')
}
if (issues.length > 0) {
throw createHttpError(`期间 ${periodKey} 不满足关账条件:${issues.join(';')}`, {
statusCode: 409,
errorCode: 'finance_period_close_blocked',
})
}
}
export async function reopenFinancePeriod(periodKeyInput: unknown) {
@@ -454,15 +776,15 @@ export async function reopenFinancePeriod(periodKeyInput: unknown) {
`
UPDATE finance_periods
SET status = 'open', closed_at = NULL, closed_by_admin_user_id = NULL, updated_at = $2
WHERE period_key = $1
WHERE period_key = $1 AND status = 'closed'
RETURNING *
`,
[periodKey, nowIso()],
)
if (!result.rows[0]) {
throw createHttpError('期间不存在或尚未关账', {
statusCode: 404,
errorCode: 'finance_period_not_found',
statusCode: 409,
errorCode: 'finance_period_not_closed',
})
}
return mapPeriod(result.rows[0])
@@ -500,11 +822,13 @@ function normalizeLines(lines: JournalLineInput[]) {
errorCode: 'finance_journal_account_required',
})
}
const debitFen = Math.trunc(Number(line.debitFen || 0))
const creditFen = Math.trunc(Number(line.creditFen || 0))
const debitFen = Number(line.debitFen || 0)
const creditFen = Number(line.creditFen || 0)
if (
!Number.isFinite(debitFen) ||
!Number.isFinite(creditFen) ||
!Number.isInteger(debitFen) ||
!Number.isInteger(creditFen) ||
debitFen < 0 ||
creditFen < 0 ||
(debitFen > 0 && creditFen > 0) ||
@@ -534,6 +858,12 @@ function normalizeLines(lines: JournalLineInput[]) {
return normalized
}
async function acquireFinancePeriodLock(client: PoolClient, periodKey: string) {
await client.query(`SELECT pg_advisory_xact_lock(hashtext('finance_period:' || $1)::bigint)`, [
periodKey,
])
}
async function assertPeriodOpen(client: PoolClient, periodKey: string) {
const result = await client.query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
@@ -16,6 +16,10 @@ import type { AdminSession } from '../admin/admin-auth-service.js'
type ReconciliationRowInput = {
platformOrderId?: unknown
platform_order_id?: unknown
platformItemId?: unknown
platform_item_id?: unknown
orderItemId?: unknown
order_item_id?: unknown
entryType?: unknown
entry_type?: unknown
amountFen?: unknown
@@ -75,13 +79,19 @@ export async function importFinanceReconciliation(payload: JsonObject, session:
for (const [index, row] of rows.entries()) {
const duplicateKey =
row.entryType === 'sale'
? `${row.platformOrderId}|sale`
: `${row.platformOrderId}|refund|${row.externalRef || row.amountFen}`
? `${row.platformOrderId}|sale|${row.platformItemId || row.orderItemId || row.amountFen}`
: `${row.platformOrderId}|refund|${row.externalRef || row.orderItemId || row.amountFen}`
const isDuplicate = seen.has(duplicateKey)
seen.add(duplicateKey)
const match = isDuplicate
? { status: 'duplicate' as const, orderId: null, refundId: null, expected: null }
? {
status: 'duplicate' as const,
orderId: null,
orderItemId: null,
refundId: null,
expected: null,
}
: await matchRow(client, provider, row)
const expected = match.expected
const diff = expected === null ? null : row.amountFen - expected
@@ -93,12 +103,13 @@ export async function importFinanceReconciliation(payload: JsonObject, session:
await client.query(
`
INSERT INTO finance_reconciliation_items (
reconciliation_id, line_no, provider, platform_order_id, external_ref,
entry_type, amount_fen, occurred_at, matched_order_id, matched_refund_id,
expected_fen, diff_fen, match_status, detail_json, created_at
reconciliation_id, line_no, provider, platform_order_id, platform_item_id,
external_ref, entry_type, amount_fen, occurred_at, matched_order_id,
order_item_id, matched_refund_id, expected_fen, diff_fen, match_status,
detail_json, created_at
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8::timestamptz, $9, $10,
$11, $12, $13, $14::jsonb, $15
$1, $2, $3, $4, $5, $6, $7, $8, $9::timestamptz, $10,
$11, $12, $13, $14, $15, $16::jsonb, $17
)
`,
[
@@ -106,11 +117,13 @@ export async function importFinanceReconciliation(payload: JsonObject, session:
index + 1,
provider,
row.platformOrderId,
row.platformItemId,
row.externalRef,
row.entryType,
row.amountFen,
row.occurredAt,
match.orderId,
match.orderItemId,
match.refundId,
expected,
diff,
@@ -227,6 +240,8 @@ export async function getFinanceReconciliation(idInput: unknown) {
id: Number(row.id),
lineNo: Number(row.line_no),
platformOrderId: String(row.platform_order_id || ''),
platformItemId: String(row.platform_item_id || ''),
orderItemId: row.order_item_id ? Number(row.order_item_id) : null,
externalRef: String(row.external_ref || ''),
entryType: String(row.entry_type || ''),
amountFen: Number(row.amount_fen || 0),
@@ -275,10 +290,51 @@ async function matchRow(
): Promise<{
status: 'matched' | 'amount_mismatch' | 'missing_order' | 'missing_refund'
orderId: number | null
orderItemId: number | null
refundId: number | null
expected: number | null
}> {
if (row.entryType === 'sale') {
if (row.orderItemId) {
const item = await client.query<{
id: number
order_id: number
revenue_fen: number
}>(
`
SELECT
oi.id,
oi.order_id,
GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)::bigint AS revenue_fen
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = $3
AND o.provider = $1
AND o.platform_order_id = $2
LIMIT 1
`,
[provider, row.platformOrderId, row.orderItemId],
)
const foundItem = item.rows[0]
if (!foundItem) {
return {
status: 'missing_order',
orderId: null,
orderItemId: null,
refundId: null,
expected: null,
}
}
const expected = Number(foundItem.revenue_fen || 0)
return {
status: expected === row.amountFen ? 'matched' : 'amount_mismatch',
orderId: Number(foundItem.order_id),
orderItemId: Number(foundItem.id),
refundId: null,
expected,
}
}
const order = await client.query<{ id: number; total_amount: number }>(
`SELECT id, total_amount FROM orders
WHERE provider = $1 AND platform_order_id = $2
@@ -287,30 +343,50 @@ async function matchRow(
)
const found = order.rows[0]
if (!found) {
return { status: 'missing_order', orderId: null, refundId: null, expected: null }
return {
status: 'missing_order',
orderId: null,
orderItemId: null,
refundId: null,
expected: null,
}
}
const expected = Number(found.total_amount || 0)
return {
status: expected === row.amountFen ? 'matched' : 'amount_mismatch',
orderId: Number(found.id),
orderItemId: null,
refundId: null,
expected,
}
}
const params: unknown[] = [provider, row.platformOrderId]
let externalClause = ''
const clauses: string[] = []
if (row.externalRef) {
params.push(row.externalRef)
externalClause = ` AND r.external_ref = $${params.length}`
clauses.push(`r.external_ref = $${params.length}`)
} else {
// 没有平台退款单号时,至少要求金额相同,避免把任意一笔退款误配到账单行。
params.push(row.amountFen)
clauses.push(`r.amount_fen = $${params.length}`)
}
const refund = await client.query<{ id: number; amount_fen: number }>(
if (row.orderItemId) {
params.push(row.orderItemId)
clauses.push(`r.order_item_id = $${params.length}`)
}
const refund = await client.query<{
id: number
amount_fen: number
order_item_id: number
}>(
`
SELECT r.id, r.amount_fen
SELECT r.id, r.amount_fen, r.order_item_id
FROM order_refunds r
JOIN orders o ON o.id = r.order_id
WHERE o.provider = $1 AND o.platform_order_id = $2
AND r.status = 'confirmed' ${externalClause}
AND r.status = 'confirmed'
AND ${clauses.join(' AND ')}
ORDER BY r.id DESC
LIMIT 1
`,
@@ -318,12 +394,19 @@ async function matchRow(
)
const found = refund.rows[0]
if (!found) {
return { status: 'missing_refund', orderId: null, refundId: null, expected: null }
return {
status: 'missing_refund',
orderId: null,
orderItemId: null,
refundId: null,
expected: null,
}
}
const expected = Number(found.amount_fen || 0)
return {
status: expected === row.amountFen ? 'matched' : 'amount_mismatch',
orderId: null,
orderItemId: Number(found.order_item_id),
refundId: Number(found.id),
expected,
}
@@ -337,6 +420,23 @@ function normalizeRow(row: ReconciliationRowInput, provider: string, index: numb
errorCode: 'finance_reconciliation_order_id_required',
})
}
const platformItemId = String(row.platformItemId ?? row.platform_item_id ?? '').trim()
const rawOrderItemId = row.orderItemId ?? row.order_item_id
let orderItemId: number | null = null
if (
rawOrderItemId !== undefined &&
rawOrderItemId !== null &&
String(rawOrderItemId).trim() !== ''
) {
const parsed = Number(rawOrderItemId)
if (!Number.isInteger(parsed) || parsed <= 0) {
throw createHttpError(`第 ${index + 1} 行订单项 ID 无效`, {
statusCode: 400,
errorCode: 'finance_reconciliation_order_item_invalid',
})
}
orderItemId = parsed
}
const entryType = String(row.entryType ?? row.entry_type ?? 'sale').trim()
if (entryType !== 'sale' && entryType !== 'refund') {
throw createHttpError(`第 ${index + 1} 行类型只能是 sale/refund`, {
@@ -362,6 +462,8 @@ function normalizeRow(row: ReconciliationRowInput, provider: string, index: numb
return {
provider,
platformOrderId,
platformItemId,
orderItemId,
entryType,
amountFen,
externalRef: String(row.externalRef ?? row.external_ref ?? '').trim(),
@@ -127,9 +127,9 @@ test('人工成本缺失必须进入异常队列,且只监控首次配置之
)
})
test('迁移 108 提供订单项级人工成本唯一约束与 subject 语义约束', async () => {
test('迁移 104 提供订单项级人工成本唯一约束与 subject 语义约束', async () => {
const source = await readFile(
new URL('../../db/migrations/108_finance_cost_integrity.sql', import.meta.url),
new URL('../../db/migrations/104_finance_controls.sql', import.meta.url),
'utf8',
)
assert.ok(source.includes('uq_cost_snapshots_manual_order_item'), '缺少订单项级人工快照唯一索引')
@@ -146,10 +146,7 @@ test('迁移 108 提供订单项级人工成本唯一约束与 subject 语义约
test('总账/对账/关账/平台退款能力已接入', async () => {
const [migration, ledger, reconciliation, refund, manual] = await Promise.all([
readFile(
new URL('../../db/migrations/108_finance_cost_integrity.sql', import.meta.url),
'utf8',
),
readFile(new URL('../../db/migrations/104_finance_controls.sql', import.meta.url), 'utf8'),
readFile(new URL('./finance-ledger-service.ts', import.meta.url), 'utf8'),
readFile(new URL('./finance-reconciliation-service.ts', import.meta.url), 'utf8'),
readFile(new URL('./order-refund-service.ts', import.meta.url), 'utf8'),
@@ -65,6 +65,7 @@ type ConfirmContextRow = {
sku_code: string
sku_name: string
revenue_fen: string
basis_at: string
canonical_product_id: number | null
fulfillment_mode: string
config_id: number | null
@@ -216,6 +217,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
oi.sku_code,
oi.sku_name,
COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
COALESCE(o.paid_at, o.created_at) AS basis_at,
oi.canonical_product_id,
oi.fulfillment_mode,
oi.fulfillment_config_id AS config_id,
@@ -254,6 +256,21 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
errorCode: 'manual_fulfillment_not_manual',
})
}
// 已关账期间不能再改成本事实;需要先反关账,避免“扣了钱但销售事实/总账没记”。
const period = await client.query<{ status: string }>(
`SELECT p.status
FROM finance_periods p
WHERE p.period_key = to_char(($1::timestamptz AT TIME ZONE 'Asia/Shanghai'), 'YYYY-MM')
LIMIT 1`,
[row.basis_at],
)
if (period.rows[0]?.status === 'closed') {
throw createHttpError('该订单所属期间已关账,请先反关账后再确认履约', {
statusCode: 409,
errorCode: 'manual_fulfillment_period_closed',
})
}
if (!row.config_id) {
throw createHttpError('该商品还没有配置履约渠道,请先在成本中心配置', {
statusCode: 409,
@@ -19,7 +19,7 @@ test(
{ skip: !TEST_DATABASE_URL ? '未设置 MANUAL_FULFILLMENT_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { query, closeDb } = await import('../../db/client.js')
const { query, withTransaction, closeDb } = await import('../../db/client.js')
const { confirmManualFulfillment } = await import('./manual-fulfillment-service.js')
const { syncEstimatedOrderItemCostSnapshots } =
await import('../cost-center/cost-snapshot-sync-service.js')
@@ -57,13 +57,23 @@ test(
)
const adminId = Number(admin.rows[0]?.id)
adminIds.push(adminId)
const fundAccount = await query<{ id: number }>(
`INSERT INTO fund_accounts (
account_no, account_type, owner_admin_user_id, balance_fen, status, created_at, updated_at
) VALUES ($1, 'agent', $2, 10000, 'active', NOW(), NOW()) RETURNING id`,
[`FA-IT-${suffix}`, adminId],
)
const fundAccountId = Number(fundAccount.rows[0]?.id)
const fundAccountId = await withTransaction(async (client) => {
const account = await client.query<{ id: number }>(
`INSERT INTO fund_accounts (
account_no, account_type, owner_admin_user_id, balance_fen, status, created_at, updated_at
) VALUES ($1, 'agent', $2, 10000, 'active', NOW(), NOW()) RETURNING id`,
[`FA-IT-${suffix}`, adminId],
)
const id = Number(account.rows[0]?.id)
await client.query(
`INSERT INTO fund_entries (
entry_no, account_id, entry_type, amount_fen, balance_after_fen,
occurred_at, created_by_admin_user_id, evidence_json, note, created_at
) VALUES ($1, $2, 'opening', 10000, 10000, NOW(), NULL, '[]'::jsonb, '集成测试期初', NOW())`,
[`FE-IT-${suffix}`, id],
)
return id
})
fundAccountIds.push(fundAccountId)
const session = {
@@ -420,6 +430,13 @@ test(
)
const periodKey = todayDate.slice(0, 7)
// 故意制造的快照失败单尚未确认,关账门槛必须先拦住。
await assert.rejects(
() => closeFinancePeriod(periodKey, session),
/不满足关账条件.*人工订单项未确认成本/,
)
// 清掉失败单后,关账才允许通过。
await confirmManualFulfillment({ orderItemId: failItem.orderItemId, costFen: 700 }, session)
await closeFinancePeriod(periodKey, session)
await assert.rejects(
() =>
@@ -492,10 +509,15 @@ test(
])
}
if (fundAccountIds.length > 0) {
await query(`DELETE FROM fund_entries WHERE account_id = ANY($1::bigint[])`, [
fundAccountIds,
])
await query(`DELETE FROM fund_accounts WHERE id = ANY($1::bigint[])`, [fundAccountIds])
await withTransaction(async (client) => {
await client.query(`SET LOCAL app.fund_balance_check = 'off'`)
await client.query(`DELETE FROM fund_entries WHERE account_id = ANY($1::bigint[])`, [
fundAccountIds,
])
await client.query(`DELETE FROM fund_accounts WHERE id = ANY($1::bigint[])`, [
fundAccountIds,
])
})
}
if (creditAccountIds.length > 0) {
await query(`DELETE FROM platform_credit_entries WHERE account_id = ANY($1::bigint[])`, [
@@ -7,6 +7,8 @@
*
* 后续接平台退款 API 时,只需写 source='api'/'platform' 的记录,其余逻辑复用。
*/
import type { PoolClient } from 'pg'
import { query, withTransaction } from '../../db/client.js'
import type { AdminSession } from '../admin/admin-auth-service.js'
import { createHttpError } from '../../utils/http.js'
@@ -140,6 +142,7 @@ export async function createManualRefund(
errorCode: 'order_refund_order_not_found',
})
}
await assertBasisPeriodOpen(client, order.basis_at, '退款')
const itemsResult = await client.query<{
id: number
@@ -203,38 +206,63 @@ export async function createManualRefund(
})
}
if (externalRef) {
const duplicate = await client.query<{ id: number }>(
`SELECT id FROM order_refunds
WHERE source = $1 AND external_ref = $2
LIMIT 1`,
[source, externalRef],
)
if (duplicate.rows[0]) {
throw createHttpError('该平台退款单号已登记过,不能重复退款', {
statusCode: 409,
errorCode: 'order_refund_external_ref_exists',
})
}
}
const batchNo = generateRefundNo(new Date(refundedAt))
const insertedIds: number[] = []
for (const [index, row] of rows.entries()) {
const inserted = await client.query<{ id: number }>(
`
INSERT INTO order_refunds (
refund_no, batch_no, order_id, order_item_id, amount_fen,
reason, note, source, status, refunded_at, external_ref,
operator_admin_user_id, created_at, updated_at
) VALUES (
$1, $2, $3, $4, $5,
$6, $7, $8, 'confirmed', $9::timestamptz, $10,
$11, $12::timestamptz, $12::timestamptz
)
RETURNING id
`,
[
`${batchNo}-${String(index + 1).padStart(2, '0')}`,
batchNo,
orderId,
row.itemId,
row.amountFen,
reason,
note,
source,
refundedAt,
externalRef,
session?.userId || null,
now,
],
)
insertedIds.push(Number(inserted.rows[0]?.id || 0))
try {
const inserted = await client.query<{ id: number }>(
`
INSERT INTO order_refunds (
refund_no, batch_no, order_id, order_item_id, amount_fen,
reason, note, source, status, refunded_at, external_ref,
operator_admin_user_id, created_at, updated_at
) VALUES (
$1, $2, $3, $4, $5,
$6, $7, $8, 'confirmed', $9::timestamptz, $10,
$11, $12::timestamptz, $12::timestamptz
)
RETURNING id
`,
[
`${batchNo}-${String(index + 1).padStart(2, '0')}`,
batchNo,
orderId,
row.itemId,
row.amountFen,
reason,
note,
source,
refundedAt,
externalRef,
session?.userId || null,
now,
],
)
insertedIds.push(Number(inserted.rows[0]?.id || 0))
} catch (error) {
if ((error as { code?: string }).code === '23505') {
throw createHttpError('该平台退款单号已登记过,不能重复退款', {
statusCode: 409,
errorCode: 'order_refund_external_ref_exists',
})
}
throw error
}
}
// 订单项都退完 → 订单级 pay_status 同步为 refunded(与平台全额退款口径一致)。
@@ -327,10 +355,11 @@ export async function voidOrderRefund(
id: number
order_id: number
status: string
pay_status: string
basis_at: string
}>(
`
SELECT r.id, r.order_id, r.status, COALESCE(o.paid_at, o.created_at) AS basis_at
SELECT r.id, r.order_id, r.status, o.pay_status, COALESCE(o.paid_at, o.created_at) AS basis_at
FROM order_refunds r
JOIN orders o ON o.id = r.order_id
WHERE r.id = $1
@@ -351,6 +380,7 @@ export async function voidOrderRefund(
errorCode: 'order_refund_already_voided',
})
}
await assertBasisPeriodOpen(client, row.basis_at, '撤销退款')
await client.query(
`
UPDATE order_refunds
@@ -363,6 +393,50 @@ export async function voidOrderRefund(
`,
[now, reason, session?.userId || null, id],
)
// 撤销后按“剩余 confirmed 退款”重算订单支付状态:
// 全额退款撤销必须把订单从 refunded 恢复为 paid,否则销售事实仍会按全额退款冲减。
const totals = await client.query<{ revenue_fen: number; refunded_fen: number }>(
`
SELECT
COALESCE(SUM(GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)), 0)::bigint AS revenue_fen,
COALESCE((
SELECT SUM(r.amount_fen)
FROM order_refunds r
JOIN order_items ri ON ri.id = r.order_item_id
WHERE ri.order_id = $1 AND r.status = 'confirmed'
), 0)::bigint AS refunded_fen
FROM order_items oi
WHERE oi.order_id = $1
`,
[row.order_id],
)
const totalRevenue = Number(totals.rows[0]?.revenue_fen || 0)
const totalRefunded = Number(totals.rows[0]?.refunded_fen || 0)
let nextPayStatus = row.pay_status
if (totalRevenue > 0) {
nextPayStatus = totalRefunded >= totalRevenue ? 'refunded' : 'paid'
}
await client.query(
`UPDATE orders SET pay_status = $1, updated_at = $2::timestamptz WHERE id = $3`,
[nextPayStatus, now, row.order_id],
)
// 退款时自动关闭的成本异常,撤销后重新打开,避免成本缺口被永久忽略。
await client.query(
`
UPDATE cost_exceptions
SET status = 'open',
resolved_at = NULL,
resolved_by_admin_user_id = NULL,
resolution_note = ''
WHERE status = 'ignored'
AND resolution_note = '订单已退款,成本异常自动关闭'
AND order_item_id IN (SELECT id FROM order_items WHERE order_id = $1)
`,
[row.order_id],
)
return { orderId: Number(row.order_id), basisAt: row.basis_at }
})
@@ -404,6 +478,26 @@ export function mapOrderRefund(row: OrderRefundRow): OrderRefund {
}
}
async function assertBasisPeriodOpen(
client: Pick<PoolClient, 'query'>,
basisAt: string,
action: string,
) {
const basisDate = toChinaDate(basisAt)
if (!basisDate) return
const periodKey = basisDate.slice(0, 7)
const period = await client.query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
[periodKey],
)
if (period.rows[0]?.status === 'closed') {
throw createHttpError(`订单所属期间 ${periodKey} 已关账,请先反关账后再${action}`, {
statusCode: 409,
errorCode: 'order_refund_period_closed',
})
}
}
function normalizeId(value: unknown, label: string): number {
const id = Number(value || 0)
if (!Number.isFinite(id) || id <= 0) {
@@ -484,16 +578,6 @@ export async function approveAndRecordKuaishouRefund(
errorCode: 'platform_refund_order_item_required',
})
}
const itemExist = await query<{ id: number }>(
`SELECT id FROM order_items WHERE id = $1 AND order_id = $2 LIMIT 1`,
[orderItemId, orderId],
)
if (!itemExist.rows[0]) {
throw createHttpError('订单项不存在或不属于该订单', {
statusCode: 404,
errorCode: 'platform_refund_order_item_not_found',
})
}
const order = await query<{ provider: string; platform_order_id: string }>(
`SELECT provider, platform_order_id FROM orders WHERE id = $1 LIMIT 1`,
[orderId],
@@ -519,27 +603,228 @@ export async function approveAndRecordKuaishouRefund(
errorCode: 'platform_refund_amount_invalid',
})
}
const approval = await approveKuaishouIndustryRefund({
refundId: platformRefundId,
refundAmount: amountFen,
desc: input.desc || input.note || '',
refundHandingWay: input.refundHandingWay,
...(input.sellerId ? { sellerId: input.sellerId } : {}),
})
const refund = await createManualRefund(
{
orderId,
orderItemId,
amountFen,
reason: input.reason || '平台退款',
note: input.note,
refundedAt: input.refundedAt,
source: 'platform',
externalRef: platformRefundId,
},
session,
const existingItem = await query<{ id: number }>(
`SELECT id FROM order_items WHERE id = $1 AND order_id = $2 LIMIT 1`,
[orderItemId, orderId],
)
return { approval, ...refund }
if (!existingItem.rows[0]) {
throw createHttpError('订单项不存在或不属于该订单', {
statusCode: 404,
errorCode: 'platform_refund_order_item_not_found',
})
}
// 已关账期间不能直接改原期间事实;必须先反关账,否则平台退款后本地无法正确冲减。
const basis = await query<{ basis_at: string }>(
`SELECT COALESCE(paid_at, created_at) AS basis_at FROM orders WHERE id = $1 LIMIT 1`,
[orderId],
)
if (basis.rows[0]?.basis_at) {
const basisDate = toChinaDate(String(basis.rows[0].basis_at))
const periodKey = basisDate.slice(0, 7)
const period = await query<{ status: string }>(
`SELECT status FROM finance_periods WHERE period_key = $1 LIMIT 1`,
[periodKey],
)
if (period.rows[0]?.status === 'closed') {
throw createHttpError(`订单所属期间 ${periodKey} 已关账,请先反关账后再退款`, {
statusCode: 409,
errorCode: 'order_refund_period_closed',
})
}
}
// 已经登记过的平台退款必须幂等返回,不能因为剩余可退为 0 被误判为超额。
const recorded = await findRecordedPlatformRefund(orderId, platformRefundId)
if (recorded) {
return {
approval: { idempotent: true, platformRefundId },
batchNo: recorded.batchNo,
refunds: [recorded],
fullRefunded: false,
closedExceptionCount: 0,
}
}
// 先做可退金额校验,再落退款意图;平台调用之后不再做“才发现金额超了”的校验。
const remainingFen = await getRefundableFen(orderId, orderItemId)
if (amountFen > remainingFen) {
throw createHttpError(`退款金额超出可退金额(剩余 ${formatFen(remainingFen)})`, {
statusCode: 409,
errorCode: 'order_refund_amount_exceeds',
})
}
const intent = await beginPlatformRefundIntent({
provider: '91kaquan',
externalRef: platformRefundId,
orderId,
orderItemId,
amountFen,
actorUserId: session?.userId || null,
})
const needsPlatformCall =
intent.status === 'pending' ||
(intent.status === 'failed' && !String(intent.error_message || '').startsWith('record_failed:'))
let approval: unknown = { idempotent: true, platformRefundId }
if (needsPlatformCall) {
try {
approval = await approveKuaishouIndustryRefund({
refundId: platformRefundId,
refundAmount: amountFen,
desc: input.desc || input.note || '',
refundHandingWay: input.refundHandingWay,
...(input.sellerId ? { sellerId: input.sellerId } : {}),
})
await updatePlatformRefundIntent(intent.id, 'platform_approved')
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
await updatePlatformRefundIntent(intent.id, 'failed', `platform_call_failed:${message}`)
throw error
}
}
try {
const refund = await createManualRefund(
{
orderId,
orderItemId,
amountFen,
reason: input.reason || '平台退款',
note: input.note,
refundedAt: input.refundedAt,
source: 'platform',
externalRef: platformRefundId,
},
session,
)
await updatePlatformRefundIntent(intent.id, 'recorded')
return { approval, ...refund }
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
await updatePlatformRefundIntent(intent.id, 'failed', `record_failed:${message}`)
throw createHttpError(
`平台已同意退款,但本地记账失败:${message}。已保留退款意图,可用同一 platformRefundId 重试`,
{
statusCode: 502,
errorCode: 'platform_refund_record_failed',
},
)
}
}
type PlatformRefundIntentRow = {
id: number
status: string
error_message: string
}
async function beginPlatformRefundIntent(input: {
provider: string
externalRef: string
orderId: number
orderItemId: number
amountFen: number
actorUserId: number | null
}): Promise<PlatformRefundIntentRow> {
return withTransaction(async (client) => {
const existing = await client.query<PlatformRefundIntentRow>(
`SELECT id, status, error_message
FROM finance_refund_intents
WHERE provider = $1 AND external_ref = $2
FOR UPDATE`,
[input.provider, input.externalRef],
)
if (existing.rows[0]) return existing.rows[0]
try {
const inserted = await client.query<PlatformRefundIntentRow>(
`
INSERT INTO finance_refund_intents (
provider, external_ref, order_id, order_item_id, amount_fen,
status, error_message, created_by_admin_user_id, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, 'pending', '', $6, $7, $7)
RETURNING id, status, error_message
`,
[
input.provider,
input.externalRef,
input.orderId,
input.orderItemId,
input.amountFen,
input.actorUserId,
nowIso(),
],
)
return inserted.rows[0] as PlatformRefundIntentRow
} catch (error) {
// 并发请求同时插入时,唯一键冲突后读取已存在的意图,由调用方按状态决定是否重试。
if ((error as { code?: string }).code === '23505') {
const again = await client.query<PlatformRefundIntentRow>(
`SELECT id, status, error_message
FROM finance_refund_intents
WHERE provider = $1 AND external_ref = $2
LIMIT 1`,
[input.provider, input.externalRef],
)
return again.rows[0] as PlatformRefundIntentRow
}
throw error
}
})
}
async function updatePlatformRefundIntent(
id: number,
status: 'platform_approved' | 'recorded' | 'failed',
errorMessage = '',
) {
await query(
`UPDATE finance_refund_intents
SET status = $2, error_message = $3, updated_at = $4
WHERE id = $1`,
[id, status, errorMessage.slice(0, 1000), nowIso()],
)
}
async function getRefundableFen(orderId: number, orderItemId: number): Promise<number> {
const result = await query<{ remaining_fen: string }>(
`
SELECT
(GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)
- COALESCE((
SELECT SUM(r.amount_fen)
FROM order_refunds r
WHERE r.order_item_id = oi.id AND r.status = 'confirmed'
), 0))::bigint AS remaining_fen
FROM order_items oi
WHERE oi.id = $1 AND oi.order_id = $2
LIMIT 1
`,
[orderItemId, orderId],
)
if (!result.rows[0]) {
throw createHttpError('订单项不存在或不属于该订单', {
statusCode: 404,
errorCode: 'platform_refund_order_item_not_found',
})
}
return Number(result.rows[0].remaining_fen || 0)
}
async function findRecordedPlatformRefund(orderId: number, externalRef: string) {
const result = await query<{ id: number }>(
`SELECT id FROM order_refunds
WHERE order_id = $1 AND source = 'platform' AND external_ref = $2 AND status = 'confirmed'
LIMIT 1`,
[orderId, externalRef],
)
if (!result.rows[0]) return null
const refunds = await listOrderRefunds(orderId)
return refunds.find((item) => item.id === Number(result.rows[0]?.id)) || null
}
function normalizeRefundSource(value: unknown): 'manual' | 'api' | 'platform' | 'legacy' {
@@ -14,6 +14,7 @@ export type SalesFactsRebuildResult = {
range: { fromDate: string; toDate: string }
deletedCount: number
insertedCount: number
skippedClosedPeriods: string[]
computedAt: string
}
@@ -50,9 +51,26 @@ export async function rebuildSalesDailyFacts(
const range = resolveFinanceDateRange(input, { defaultDays: 30, maxDays: 400 })
const computedAt = nowIso()
// 已关账期间必须冻结:只重算开放期间,避免“总账已关账、报表事实还在变”。
const closedPeriods = await query<{ period_key: string }>(
`SELECT period_key
FROM finance_periods
WHERE status = 'closed'
AND period_key BETWEEN $1 AND $2
ORDER BY period_key`,
[range.fromDate.slice(0, 7), range.toDate.slice(0, 7)],
)
const skippedClosedPeriods = closedPeriods.rows.map((row) => row.period_key)
return withTransaction(async (client) => {
const deleted = await client.query(
`DELETE FROM sales_daily_facts WHERE stat_date BETWEEN $1::date AND $2::date`,
`DELETE FROM sales_daily_facts f
WHERE f.stat_date BETWEEN $1::date AND $2::date
AND NOT EXISTS (
SELECT 1 FROM finance_periods p
WHERE p.status = 'closed'
AND p.period_key = to_char(f.stat_date, 'YYYY-MM')
)`,
[range.fromDate, range.toDate],
)
@@ -103,6 +121,14 @@ export async function rebuildSalesDailyFacts(
WHERE o.pay_status IN ('paid', 'refunded')
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND NOT EXISTS (
SELECT 1 FROM finance_periods p
WHERE p.status = 'closed'
AND p.period_key = to_char(
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date,
'YYYY-MM'
)
)
)
INSERT INTO sales_daily_facts (
stat_date, provider, shop_id, canonical_product_id, fulfillment_mode,
@@ -146,6 +172,7 @@ export async function rebuildSalesDailyFacts(
range,
deletedCount: deleted.rowCount || 0,
insertedCount: inserted.rowCount || 0,
skippedClosedPeriods,
computedAt,
}
})
@@ -163,6 +190,7 @@ export async function getSalesProfitOverview(input: SalesFactsInput = {}) {
net_revenue_fen: number
cost_amount_fen: number
cost_missing_count: number
cost_missing_revenue_fen: number
gross_profit_fen: number
unmapped_revenue_fen: number
}>(
@@ -174,6 +202,9 @@ export async function getSalesProfitOverview(input: SalesFactsInput = {}) {
COALESCE(SUM(net_revenue_fen), 0)::bigint AS net_revenue_fen,
COALESCE(SUM(cost_amount_fen), 0)::bigint AS cost_amount_fen,
COALESCE(SUM(cost_missing_count), 0)::int AS cost_missing_count,
COALESCE(
SUM(net_revenue_fen) FILTER (WHERE cost_missing_count > 0), 0
)::bigint AS cost_missing_revenue_fen,
COALESCE(SUM(gross_profit_fen), 0)::bigint AS gross_profit_fen,
COALESCE(SUM(gross_revenue_fen) FILTER (WHERE canonical_product_id IS NULL), 0)::bigint
AS unmapped_revenue_fen
@@ -301,6 +332,7 @@ export async function getSalesProfitOverview(input: SalesFactsInput = {}) {
netRevenueFen: Number(summary?.net_revenue_fen || 0),
costAmountFen: Number(summary?.cost_amount_fen || 0),
costMissingCount: Number(summary?.cost_missing_count || 0),
costMissingRevenueFen: Number(summary?.cost_missing_revenue_fen || 0),
grossProfitFen: Number(summary?.gross_profit_fen || 0),
unmappedRevenueFen: unmappedRevenue,
unmappedRevenuePct:
@@ -1701,12 +1701,12 @@ test('拼单份数全部承接后不再显示在抢单大厅', () => {
/**
* 同一打手可以多次拼单(需求),所以「取消后重新加入」不再靠复用原份额行:
* 迁移 109 删掉了「一人一条」唯一键,每次参与都插入新份额,取消的历史份额保留追溯。
* 迁移 105 删掉了「一人一条」唯一键,每次参与都插入新份额,取消的历史份额保留追溯。
*/
test('拼单份额取消后原打手可以重新参与(一人一条唯一键已移除)', async () => {
const { readFile } = await import('node:fs/promises')
const migration = await readFile(
new URL('../../db/migrations/109_work_order_share_participation.sql', import.meta.url),
new URL('../../db/migrations/105_work_order_consignment.sql', import.meta.url),
'utf8',
)
assert.ok(