如何基于 income_funds 和 expenditure_funds 实现 billing system

结论

更可靠的做法不是从三种方案中任选一种，而是把它们组合起来：

1. 以不可变的资金流水作为账务事实来源；
2. 在 account 中保存当前余额，供日常快速查询；
3. 使用数据库事务，原子地写入流水并更新余额；
4. 使用 period 保存已关闭周期的余额快照，供报表和对账使用；
5. 通过幂等键、调整流水和审计日志追踪错误。

简单来说，account.balance 是可以根据流水重建的缓存，资金流水才是最终依据。period 用于结账和对账，不应代替实时账户余额。

现有方案的问题

直接执行：

sum(confirmed income_funds) - sum(expenditure_funds)

这个计算在逻辑上没有问题，但数据量增加后，实时聚合的成本会越来越高。它适合对账，不适合成为每次查询余额的主要路径。

只增加 account.balance 确实能提高读取速度，但如果流水写入和余额更新不在同一个原子操作中，就可能出现以下问题：

• 流水已写入，余额却没有更新；
• 余额已更新，流水写入却失败；
• 请求重试导致余额被重复修改；
• 修改历史记录后，无法解释余额的来源；
• 并发更新导致余额相互覆盖。

只增加 period 也解决不了问题。未关闭周期内的余额会持续变化。如果把最后一个 period.end_balance 当作当前余额，就需要不断修改这个字段，它实际上会变成另一种 account.balance。已经关闭的周期原则上不应再修改，否则历史报表也会随之变化。

推荐的数据模型

如果业务允许调整表结构，建议把收入和支出合并到一张不可变流水表中。与维护两个 collection 相比，统一模型更容易处理一致性、排序、对账和审计。

accounts
  id
  balance
  version
  updated_at

ledger_entries
  id
  account_id
  type                 // income, expenditure, adjustment
  amount               // 始终为正数
  direction            // credit, debit
  status               // pending, posted, reversed
  effective_at
  created_at
  idempotency_key
  reference_id
  reversal_of
  metadata

periods
  id
  account_id
  start_at
  end_at
  opening_balance
  total_credit
  total_debit
  closing_balance
  status               // open, closed
  closed_at

其中：

posted_delta =
  amount                  when direction = credit
  -amount                 when direction = debit

当前余额应满足：

account.balance
= sum(posted credits)
- sum(posted debits)

如果暂时必须保留 income_funds 和 expenditure_funds，也可以沿用同样的原则，但需要为支出增加明确的状态。否则，尚未完成、已经取消或执行失败的支出是否应当扣款，会变得难以判断。

流水与余额必须在同一个事务中更新

确认一笔收入时，应在同一个数据库事务中完成以下操作：

1. 锁定原流水，或者对它执行条件更新；
2. 将状态从 pending 改为 posted；
3. 增加账户余额；
4. 提交事务。

伪代码如下：

async function confirmIncome(db, entryId) {
  return db.transaction(async (tx) => {
    const entry = await tx.ledger_entries.findOneForUpdate({
      id: entryId
    });

    if (!entry) {
      throw new Error("Entry not found");
    }

    if (entry.status === "posted") {
      return; // 已处理，保证重复调用不会重复入账
    }

    if (entry.status !== "pending") {
      throw new Error("Invalid status transition");
    }

    await tx.ledger_entries.update(
      { id: entry.id, status: "pending" },
      { status: "posted" }
    );

    await tx.accounts.increment(
      { id: entry.account_id },
      {
        balance: entry.amount,
        version: 1
      }
    );
  });
}

新增支出的逻辑相同，只是余额变化为负数：

await tx.accounts.increment(
  { id: entry.account_id },
  {
    balance: -entry.amount,
    version: 1
  }
);

具体使用哪个 API 并不重要，重要的是状态变更和余额增减必须共同成功或共同失败。如果数据库不支持跨 collection 事务，可以考虑：

• 将账户和流水迁移到支持事务的数据模型中；
• 使用原子条件更新；
• 使用 outbox/event log 配合幂等消费者，同时接受最终一致性。

余额是核心财务数据，应优先使用强一致事务。

不要直接修改已经入账的流水

如果一笔 posted 流水有误，不建议直接修改它的 amount、direction，也不要删除记录。更稳妥的做法是先创建冲正流水，再写入正确的流水。

例如，原收入为 100，实际应为 80：

原流水：       credit  100  posted
冲正流水：     debit   100  posted  reversal_of=<原流水ID>
正确流水：     credit   80  posted

这样最终净变化仍为 80，同时保留了完整的修正记录，可以回答：

• 哪条记录有误；
• 何时发现错误；
• 谁进行了修正；
• 修正前后的内容分别是什么；
• 哪个业务请求触发了修正。

refunded 最好也通过一条关联原交易的反向流水来表达，而不是只修改原收入的状态。单纯修改状态会丢失退款时间，也会导致历史报表随着状态变化。

使用幂等键防止重复记账

数据库事务不会自动处理客户端的重复请求。例如，支付平台重复发送同一个回调时，两个事务仍可能分别成功。

可以为每个业务操作保存唯一的 idempotency_key：

idempotency_key = provider + ":" + provider_transaction_id

并建立唯一约束：

CREATE UNIQUE INDEX uq_ledger_idempotency
ON ledger_entries (account_id, idempotency_key);

这样，同一个支付、退款或扣款请求即使被多次提交，也只能生成一次有效记账。

period 应该作为结账快照

period 用于生成稳定报表和缩小对账范围，不负责维护实时余额。

关闭周期时计算：

closing_balance
= opening_balance
+ total_posted_credit
- total_posted_debit

同时验证：

current_period.opening_balance
= previous_period.closing_balance

周期一旦关闭，就应保持不可变。之后如果发现历史错误，应在当前开放周期中创建调整或冲正流水，而不是修改已经关闭的周期。

一个典型的关账过程如下：

async function closePeriod(db, accountId, startAt, endAt) {
  return db.transaction(async (tx) => {
    const account = await tx.accounts.findOneForUpdate({
      id: accountId
    });

    const totals = await tx.ledger_entries.aggregate({
      account_id: accountId,
      status: "posted",
      effective_at: {
        $gte: startAt,
        $lt: endAt
      }
    });

    const closingBalance =
      totals.opening_balance +
      totals.total_credit -
      totals.total_debit;

    await tx.periods.insert({
      account_id: accountId,
      start_at: startAt,
      end_at: endAt,
      opening_balance: totals.opening_balance,
      total_credit: totals.total_credit,
      total_debit: totals.total_debit,
      closing_balance: closingBalance,
      status: "closed"
    });
  });
}

实际实现时，还要避免在关账过程中继续插入属于该周期的流水。常见做法包括锁定账户、记录关账截止序号，或者在周期关闭后禁止写入 effective_at 早于 end_at 的流水。

报表不要依赖当前余额倒推

资金报表应直接查询流水，并明确使用哪个时间字段：

• created_at：系统记录流水的时间；
• effective_at：流水在财务上所属的期间；
• 支付平台时间：外部交易实际发生的时间。

报表通常按 effective_at 统计，审计则按 created_at 查看。

示例：

const report = await ledgerEntries.aggregate([
  {
    $match: {
      account_id: accountId,
      status: "posted",
      effective_at: {
        $gte: periodStart,
        $lt: periodEnd
      }
    }
  },
  {
    $group: {
      _id: "$direction",
      total: { $sum: "$amount" },
      count: { $sum: 1 }
    }
  }
]);

时间范围建议使用左闭右开区间：

[start_at, end_at)

这样相邻周期不会重复包含边界上的交易。

对账仍然有必要

保存 account.balance 不代表以后不再根据流水重新计算余额。两者应分别承担不同职责：

• 日常查询直接读取 account.balance；
• 定期根据流水重新计算余额，用于对账；
• 发现差异时发出告警，不要自动静默覆盖；
• 通过流水 ID、业务引用、幂等键和时间定位问题。

对账公式为：

calculated_balance
= opening_balance
+ sum(posted credits after opening)
- sum(posted debits after opening)

对账不必每次扫描全部历史数据。可以从最近一个可信的已关闭周期快照开始，只计算该周期之后的流水。数据量继续增加时，也可以按日或按月建立汇总快照，但这些汇总仍然是可以重建的数据。

其他注意事项

金额不要使用浮点数。可以使用数据库提供的精确 decimal 类型，也可以按最小货币单位保存为整数：

10.25 CNY -> 1025 fen

如果系统支持多币种，余额和流水都必须包含 currency，不同币种不能直接相加：

account_id + currency

还应为以下字段建立索引或唯一约束：

(account_id, status, effective_at)
(account_id, idempotency_key) UNIQUE
(reference_id)
(reversal_of)

同时记录必要的审计信息，例如：

created_by
reason
request_id
source
created_at

最终可以把三者的职责概括为：不可变流水解释”钱为什么发生变化”，account.balance 快速回答”现在有多少钱”，period 记录”某个结算周期发生了什么”。职责分开后，系统才能兼顾读取效率、数据一致性和错误追溯。

备注：内容仅供参考。