nanakura/moeka-project
1
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat(server): refactor flux auditing to transaction logging

Browse Source
This commit is contained in:
RainbowBird
2026-03-28 02:25:44 +08:00
committed by RainbowBird
parent 28f5c87fb2
commit 2f730ef93a
29 changed files with 2506 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/docs/ai-context/architecture-overview.md
+1 -1
View File
@@ -48,7 +48,7 @@ CLI 入口在 `src/bin/run.ts`，支持两种角色：
- `providerService`
- `chatService`
- `stripeService`
- `fluxAuditService`
- `fluxTransactionService`
- `fluxService`
- `requestLogService`
- `billingService`
apps/server/docs/ai-context/billing-architecture.md
+12 -12
View File
@@ -2,14 +2,14 @@
## 架构概述
`apps/server` 的计费链采用 **Postgres 作为唯一账本真相源**，Redis 仅作缓存。余额变化路径分两类：`debitFlux` 在 DB 事务内只做 `UPDATE user_flux`，ledger/audit/请求日志通过 Redis Stream 异步写入；credit 方法仍在事务内同步写入 ledger 和 audit。
`apps/server` 的计费链采用 **Postgres 作为唯一账本真相源**，Redis 仅作缓存。余额变化路径分两类：`debitFlux` 在 DB 事务内只做 `UPDATE user_flux`，transaction/请求日志通过 Redis Stream 异步写入；credit 方法仍在事务内同步写入 transaction log。
### 数据模型
- **`user_flux`** — 用户余额快照（单行/用户）
- **`flux_ledger`** — append-only 账务流水（type: credit/debit/initial, amount, balanceBefore, balanceAfter, requestId）
- **`flux_transaction`** — append-only 账务流水（type: credit/debit/initial, amount, balanceBefore, balanceAfter, requestId）
- 含 partial unique index `(userId, requestId) WHERE requestId IS NOT NULL`，DB 层幂等防重
- **`flux_audit_log`** — 用户可见的历史记录
- **`flux_transaction`** — 用户可见的历史记录
### debitFlux 链路（已实现）
@@ -21,15 +21,15 @@ DB 事务内仅做：
4. 事务提交后 XADD Redis Stream（`billing-events`），携带扣费金额、余额快照、requestId 等
5. 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存
ledger / audit / llm_request_log 的写入均由 **billing-consumer** 异步完成。
transaction log / audit / llm_request_log 的写入均由 **billing-consumer** 异步完成。
### credit 方法链路（已实现）
credit 方法（`creditFlux` / `creditFluxFromStripeCheckout` / `creditFluxFromInvoice`）仍在 DB 事务内同步写入 `flux_ledger` 和 `flux_audit_log`。
credit 方法（`creditFlux` / `creditFluxFromStripeCheckout` / `creditFluxFromInvoice`）仍在 DB 事务内同步写入 `flux_transaction` 和 `flux_transaction`。
### 异步链路（已实现）
- **billing-consumer** — 消费 Redis Stream `billing-events`，将 ledger、audit log、LLM 请求日志异步写入 DB
- **billing-consumer** — 消费 Redis Stream `billing-events`，将 transaction log、LLM 请求日志异步写入 DB
### 事件模型
@@ -47,7 +47,7 @@ Stream: `billing-events`
通过 `src/bin/run.ts` 分角色启动：
- `api` — HTTP 服务
- `billing-consumer` — 消费 Redis Stream，异步写入 ledger、audit log、LLM 请求日志到 DB
- `billing-consumer` — 消费 Redis Stream，异步写入 transaction log、LLM 请求日志到 DB
## 关键服务
@@ -55,7 +55,7 @@ Stream: `billing-events`
所有余额写操作的唯一入口：
- **`debitFlux()`** — 扣费（LLM 请求），事务内：锁行 → 检余额(402) → 更新余额；事务提交后 XADD `flux.debited` 到 Redis Stream，ledger/audit 由 billing-consumer 异步写入
- **`debitFlux()`** — 扣费（LLM 请求），事务内：锁行 → 检余额(402) → 更新余额；事务提交后 XADD `flux.debited` 到 Redis Stream，transaction 由 billing-consumer 异步写入
- **`creditFlux()`** — 通用充值
- **`creditFluxFromStripeCheckout()`** — Stripe 一次性支付充值，幂等(`fluxCredited` 标志)
- **`creditFluxFromInvoice()`** — Stripe 订阅发票充值，幂等
@@ -64,7 +64,7 @@ Stream: `billing-events`
只负责读操作：
- **`getFlux()`** — Redis cache-aside 读（miss → DB → 填充 Redis），新用户自动初始化 + 写 ledger(type=initial)
- **`getFlux()`** — Redis cache-aside 读（miss → DB → 填充 Redis），新用户自动初始化 + 写 transaction log(type=initial)
- **`updateStripeCustomerId()`**
### Redis 职责边界
@@ -80,12 +80,12 @@ Redis **不是**余额真相源，仅用于：
| Phase | 状态 | 关键点 |
|-------|------|--------|
| 1. DB-first 账本 | ✅ 已完成 | `flux_ledger` 表，`SELECT FOR UPDATE` 原子扣减，Redis 降为缓存 |
| 2. Redis Streams 异步写入 | ✅ 已完成 | debitFlux 事务后 XADD，billing-consumer 异步写 ledger/audit/请求日志 |
| 1. DB-first 账本 | ✅ 已完成 | `flux_transaction` 表，`SELECT FOR UPDATE` 原子扣减，Redis 降为缓存 |
| 2. Redis Streams 异步写入 | ✅ 已完成 | debitFlux 事务后 XADD，billing-consumer 异步写 transaction/请求日志 |
| 3. Stripe 幂等 | ✅ 已完成 | checkout + invoice 事务内幂等检查 |
| 4. LLM 计费优化 | ⚠️ 部分 | 已有 `requestId` 和 DB 事务扣费，待加 tiktoken fallback |
| 5. 部署拆分 | ✅ 已完成 | `bin/run.ts` 两角色启动（api / billing-consumer） |
| 6. 幂等防重 | ✅ 已完成 | `flux_ledger` partial unique index on `(userId, requestId)` |
| 6. 幂等防重 | ✅ 已完成 | `flux_transaction` partial unique index on `(userId, requestId)` |
### 已删除
apps/server/docs/ai-context/data-model-and-state.md
+13 -13
View File
@@ -94,29 +94,29 @@
### Flux / 账本 / 审计
- `user_flux`
- `flux_ledger`
- `flux_audit_log`
- `flux_transaction`
- `flux_transaction`
来源文件：
- `src/schemas/flux.ts`
- `src/schemas/flux-ledger.ts`
- `src/schemas/flux-audit-log.ts`
- `src/schemas/flux-transaction.ts`
- `src/schemas/flux-transaction.ts`
职责边界：
- `user_flux`
- 当前余额快照
- `flux_ledger`
- `flux_transaction`
- append-only 账本流水
- 偏系统真相源
- `flux_audit_log`
- `flux_transaction`
- 用户可见历史
- 偏产品展示
关键约束：
- `flux_ledger` 对 `(userId, requestId)` 有部分唯一索引
- `flux_transaction` 对 `(userId, requestId)` 有部分唯一索引
- 用来做扣费 / 充值幂等
### Stripe 业务镜像
@@ -133,7 +133,7 @@
说明：
- 这些表是 Stripe 状态的本地镜像
- 真正的余额变化仍由 `billingService` 写入 `user_flux + flux_ledger`
- 真正的余额变化仍由 `billingService` 写入 `user_flux + flux_transaction`
- `fluxCredited` 字段用于避免重复入账
### LLM 请求日志
@@ -163,7 +163,7 @@
- 扣费
- 充值
- ledger / audit 写入
- transaction 写入
### `createBillingService()`
@@ -171,8 +171,8 @@
- 所有余额写操作
- DB 事务
- debitFlux：事务内仅更新余额；事务后 XADD Redis Stream，ledger/audit 由 billing-consumer 异步写入
- credit 方法：事务内同步写 ledger / audit
- debitFlux：事务内仅更新余额；事务后 XADD Redis Stream，transaction log 由 billing-consumer 异步写入
- credit 方法：事务内同步写 transaction
- 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存
这是所有 Flux 写路径应收敛到的中心。
@@ -227,7 +227,7 @@
1. `SELECT user_flux FOR UPDATE`
2. 计算新余额
3. 写余额（debitFlux 事务内仅此一步；credit 方法同步写 ledger / audit）
3. 写余额（debitFlux 事务内仅此一步；credit 方法同步写 transaction）
4. 事务提交后 XADD Redis Stream（debitFlux）或直接返回（credit）
这保证同一用户余额更新是串行化的。
@@ -238,7 +238,7 @@
- `stripe_checkout_session.fluxCredited`
- `stripe_invoice.fluxCredited`
- `flux_ledger(userId, requestId)` 唯一约束
- `flux_transaction(userId, requestId)` 唯一约束
## 现有代码中的结构信号
apps/server/docs/ai-context/transport-and-routes.md
+1 -1
View File
@@ -188,7 +188,7 @@
- route: `src/routes/flux/index.ts`
- services:
- `fluxService`
- `fluxAuditService`
- `fluxTransactionService`
主要能力：
apps/server/docs/ai-context/workers-and-runtime.md
+1 -1
View File
@@ -42,7 +42,7 @@
1. 以 consumer group 模式消费 Redis Stream `billing-events`
2. 根据事件类型分发处理：
- `flux.debited` — 写 `flux_ledger` 和 `flux_audit_log`
- `flux.debited` — 写 `flux_transaction` 和 `flux_transaction`
- `llm.request.log` — 写 `llm_request_log`
3. 处理成功后 ACK；handler 抛错时不 ACK，消息保持 pending 等待重试
apps/server/drizzle/0007_red_nicolaos.sql
+21
View File
@@ -0,0 +1,21 @@
CREATE TABLE "flux_transaction" (
"id" text PRIMARY KEY NOT NULL,
"user_id" text NOT NULL,
"type" text NOT NULL,
"amount" bigint NOT NULL,
"balance_before" bigint NOT NULL,
"balance_after" bigint NOT NULL,
"request_id" text,
"description" text NOT NULL,
"metadata" jsonb,
"created_at" timestamp DEFAULT now() NOT NULL
);
--> statement-breakpoint
ALTER TABLE "flux_ledger" DISABLE ROW LEVEL SECURITY;--> statement-breakpoint
DROP TABLE "flux_ledger" CASCADE;--> statement-breakpoint
ALTER TABLE "user_flux" ALTER COLUMN "flux" SET DATA TYPE bigint;--> statement-breakpoint
ALTER TABLE "llm_request_log" ALTER COLUMN "flux_consumed" SET DATA TYPE bigint;--> statement-breakpoint
ALTER TABLE "flux_transaction" ADD CONSTRAINT "flux_transaction_user_id_user_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."user"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
CREATE INDEX "flux_tx_user_id_idx" ON "flux_transaction" USING btree ("user_id");--> statement-breakpoint
CREATE INDEX "flux_tx_created_at_idx" ON "flux_transaction" USING btree ("created_at");--> statement-breakpoint
CREATE UNIQUE INDEX "flux_tx_user_request_uniq" ON "flux_transaction" USING btree ("user_id","request_id") WHERE request_id IS NOT NULL;
apps/server/drizzle/meta/0007_snapshot.json
+2270
View File
File diff suppressed because it is too large Load Diff
apps/server/drizzle/meta/_journal.json
+7
View File
@@ -50,6 +50,13 @@
"when": 1774584037626,
"tag": "0006_overconfident_susan_delgado",
"breakpoints": true
},
{
"idx": 7,
"version": "7",
"when": 1774632446757,
"tag": "0007_red_nicolaos",
"breakpoints": true
}
]
}
apps/server/package.json
+3 -3
View File
@@ -40,9 +40,9 @@
"@proj-airi/drizzle-orm-browser-migrator": "^0.1.6",
"@proj-airi/server-schema": "workspace:*",
"@proj-airi/server-sdk-shared": "workspace:*",
"better-auth": "^1.5.6",
"better-auth": "catalog:",
"cac": "catalog:",
"drizzle-orm": "^0.45.1",
"drizzle-orm": "catalog:",
"drizzle-valibot": "catalog:",
"hono": "catalog:",
"hono-rate-limiter": "catalog:",
@@ -56,6 +56,6 @@
"devDependencies": {
"@better-auth/cli": "^1.4.21",
"@types/pg": "^8.20.0",
"drizzle-kit": "^0.31.10"
"drizzle-kit": "catalog:"
}
}
apps/server/src/app.ts
+13 -13
View File
@@ -9,7 +9,7 @@ import type { CharacterService } from './services/characters'
import type { ChatService } from './services/chats'
import type { ConfigKVService } from './services/config-kv'
import type { FluxService } from './services/flux'
import type { FluxAuditService } from './services/flux-audit'
import type { FluxTransactionService } from './services/flux-transaction'
import type { ProviderService } from './services/providers'
import type { StripeService } from './services/stripe'
import type { HonoEnv } from './types/hono'
@@ -47,7 +47,7 @@ import { createCharacterService } from './services/characters'
import { createChatService } from './services/chats'
import { createConfigKVService } from './services/config-kv'
import { createFluxService } from './services/flux'
import { createFluxAuditService } from './services/flux-audit'
import { createFluxTransactionService } from './services/flux-transaction'
import { createProviderService } from './services/providers'
import { createRequestLogService } from './services/request-log'
import { createStripeService } from './services/stripe'
@@ -60,7 +60,7 @@ interface AppDeps {
chatService: ChatService
providerService: ProviderService
fluxService: FluxService
fluxAuditService: FluxAuditService
fluxTransactionService: FluxTransactionService
stripeService: StripeService
billingService: BillingService
billingMq: MqService<BillingEvent>
@@ -70,7 +70,7 @@ interface AppDeps {
otel: OtelInstance | null
}
function buildApp(deps: AppDeps) {
async function buildApp(deps: AppDeps) {
const logger = useLogger('app').useGlobalConfig()
const app = new Hono<HonoEnv>()
@@ -144,8 +144,8 @@ function buildApp(deps: AppDeps) {
* Rate limited by IP: 20 requests per minute.
*/
.use('/api/auth/*', rateLimiter({
max: 20,
windowSec: 60,
max: await deps.configKV.getOrThrow('AUTH_RATE_LIMIT_MAX'),
windowSec: await deps.configKV.getOrThrow('AUTH_RATE_LIMIT_WINDOW_SEC'),
keyGenerator: c => c.req.header('x-forwarded-for') ?? c.req.header('x-real-ip') ?? 'unknown',
}))
.on(['POST', 'GET'], '/api/auth/*', c => deps.auth.handler(c.req.raw))
@@ -173,7 +173,7 @@ function buildApp(deps: AppDeps) {
/**
* Flux routes.
*/
.route('/api/v1/flux', createFluxRoutes(deps.fluxService, deps.fluxAuditService))
.route('/api/v1/flux', createFluxRoutes(deps.fluxService, deps.fluxTransactionService))
/**
* Stripe routes.
@@ -183,7 +183,7 @@ function buildApp(deps: AppDeps) {
return { app: builtApp, injectWebSocket }
}
export type AppType = ReturnType<typeof buildApp>['app']
export type AppType = Awaited<ReturnType<typeof buildApp>>['app']
export async function createApp() {
initLogger(LoggerLevel.Debug, LoggerFormat.Pretty)
@@ -295,9 +295,9 @@ export async function createApp() {
build: ({ dependsOn }) => createStripeService(dependsOn.db),
})
const fluxAuditService = injeca.provide('services:fluxAudit', {
const fluxTransactionService = injeca.provide('services:fluxTransaction', {
dependsOn: { db },
build: ({ dependsOn }) => createFluxAuditService(dependsOn.db),
build: ({ dependsOn }) => createFluxTransactionService(dependsOn.db),
})
const fluxService = injeca.provide('services:flux', {
@@ -323,7 +323,7 @@ export async function createApp() {
chatService,
providerService,
fluxService,
fluxAuditService,
fluxTransactionService,
requestLogService,
stripeService,
billingService,
@@ -333,13 +333,13 @@ export async function createApp() {
env: parsedEnv,
otel,
})
const { app, injectWebSocket } = buildApp({
const { app, injectWebSocket } = await buildApp({
auth: resolved.auth,
characterService: resolved.characterService,
chatService: resolved.chatService,
providerService: resolved.providerService,
fluxService: resolved.fluxService,
fluxAuditService: resolved.fluxAuditService,
fluxTransactionService: resolved.fluxTransactionService,
stripeService: resolved.stripeService,
billingService: resolved.billingService,
billingMq: resolved.billingMq,
apps/server/src/bin/run.ts
+1 -1
View File
@@ -33,7 +33,7 @@ export function createServerCli() {
.action(() => runServerRole('api'))
cli
.command('billing-consumer', 'Start the billing events consumer (ledger, audit, request logs)')
.command('billing-consumer', 'Start the billing events consumer (transactions, audit, request logs)')
.action(() => runServerRole('billing-consumer'))
cli.help()
apps/server/src/routes/flux/index.ts
+6 -3
View File
@@ -1,5 +1,5 @@
import type { FluxService } from '../../services/flux'
import type { FluxAuditService } from '../../services/flux-audit'
import type { FluxTransactionService } from '../../services/flux-transaction'
import type { HonoEnv } from '../../types/hono'
import { Hono } from 'hono'
@@ -8,7 +8,10 @@ import { parse } from 'valibot'
import { authGuard } from '../../middlewares/auth'
import { LimitOffsetPaginationQuerySchema } from '../../utils/http-query'
export function createFluxRoutes(fluxService: FluxService, fluxAuditService: FluxAuditService) {
export function createFluxRoutes(
fluxService: FluxService,
fluxTransactionService: FluxTransactionService,
) {
return new Hono<HonoEnv>()
.use('*', authGuard)
.get('/', async (c) => {
@@ -23,7 +26,7 @@ export function createFluxRoutes(fluxService: FluxService, fluxAuditService: Flu
offset: c.req.query('offset'),
})
const { records, hasMore } = await fluxAuditService.getHistory(user.id, limit, offset)
const { records, hasMore } = await fluxTransactionService.getHistory(user.id, limit, offset)
return c.json({
records: records.map(r => ({
apps/server/src/routes/flux/route.test.ts
+10 -10
View File
@@ -1,5 +1,5 @@
import type { FluxService } from '../../services/flux'
import type { FluxAuditService } from '../../services/flux-audit'
import type { FluxTransactionService } from '../../services/flux-transaction'
import type { HonoEnv } from '../../types/hono'
import { Hono } from 'hono'
@@ -15,14 +15,14 @@ function createMockFluxService(): FluxService {
} as any
}
function createMockFluxAuditService(): FluxAuditService {
function createMockFluxTransactionService(): FluxTransactionService {
return {
createEntry: vi.fn(),
createEntries: vi.fn(),
getHistory: vi.fn(async (_userId: string, limit: number, offset: number) => ({
records: [
{
id: 'ledger-1',
id: 'tx-1',
type: 'credit',
amount: 5,
description: 'Top up',
@@ -35,8 +35,8 @@ function createMockFluxAuditService(): FluxAuditService {
} as any
}
function createTestApp(fluxService: FluxService, fluxAuditService: FluxAuditService) {
const routes = createFluxRoutes(fluxService, fluxAuditService)
function createTestApp(fluxService: FluxService, fluxTransactionService: FluxTransactionService) {
const routes = createFluxRoutes(fluxService, fluxTransactionService)
const app = new Hono<HonoEnv>()
app.onError((err, c) => {
@@ -68,7 +68,7 @@ const testUser = { id: 'user-1', name: 'Test User', email: 'test@example.com' }
describe('fluxRoutes', () => {
it('get /api/v1/flux should return the current user balance', async () => {
const fluxService = createMockFluxService()
const app = createTestApp(fluxService, createMockFluxAuditService())
const app = createTestApp(fluxService, createMockFluxTransactionService())
const res = await app.fetch(
new Request('http://localhost/api/v1/flux'),
@@ -81,8 +81,8 @@ describe('fluxRoutes', () => {
})
it('get /api/v1/flux/history should clamp pagination query values', async () => {
const fluxAuditService = createMockFluxAuditService()
const app = createTestApp(createMockFluxService(), fluxAuditService)
const fluxTransactionService = createMockFluxTransactionService()
const app = createTestApp(createMockFluxService(), fluxTransactionService)
const res = await app.fetch(
new Request('http://localhost/api/v1/flux/history?limit=999&offset=-12'),
@@ -90,11 +90,11 @@ describe('fluxRoutes', () => {
)
expect(res.status).toBe(200)
expect(fluxAuditService.getHistory).toHaveBeenCalledWith('user-1', 100, 0)
expect(fluxTransactionService.getHistory).toHaveBeenCalledWith('user-1', 100, 0)
expect(await res.json()).toEqual({
records: [
{
id: 'ledger-1',
id: 'tx-1',
type: 'credit',
amount: 5,
description: 'Top up',
apps/server/src/routes/openai/v1/index.ts
+4 -2
View File
@@ -249,7 +249,8 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi
userId: user.id,
amount: fluxConsumed,
requestId,
description: requestModel,
description: 'llm_request',
model: requestModel,
promptTokens: usage.promptTokens,
completionTokens: usage.completionTokens,
})
@@ -296,7 +297,8 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi
userId: user.id,
amount: fluxConsumed,
requestId,
description: requestModel,
description: 'llm_request',
model: requestModel,
promptTokens: usage.promptTokens,
completionTokens: usage.completionTokens,
})
apps/server/src/schemas/flux-ledger.ts → apps/server/src/schemas/flux-transaction.ts
+8 -8
View File
@@ -1,24 +1,24 @@
import { sql } from 'drizzle-orm'
import { index, integer, jsonb, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core'
import { bigint, index, jsonb, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core'
import { nanoid } from '../utils/id'
import { user } from './accounts'
export const fluxLedger = pgTable('flux_ledger', {
export const fluxTransaction = pgTable('flux_transaction', {
id: text('id').primaryKey().$defaultFn(() => nanoid()),
userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }),
type: text('type').notNull(), // 'credit' | 'debit' | 'initial'
amount: integer('amount').notNull(), // always positive
balanceBefore: integer('balance_before').notNull(),
balanceAfter: integer('balance_after').notNull(),
amount: bigint('amount', { mode: 'number' }).notNull(), // always positive
balanceBefore: bigint('balance_before', { mode: 'number' }).notNull(),
balanceAfter: bigint('balance_after', { mode: 'number' }).notNull(),
requestId: text('request_id'), // nullable; used for idempotency on debit/credit
description: text('description').notNull(),
metadata: jsonb('metadata'), // { promptTokens, completionTokens, stripeSessionId, ... }
createdAt: timestamp('created_at').defaultNow().notNull(),
}, table => [
index('flux_ledger_user_id_idx').on(table.userId),
index('flux_ledger_created_at_idx').on(table.createdAt),
uniqueIndex('flux_ledger_user_request_uniq')
index('flux_tx_user_id_idx').on(table.userId),
index('flux_tx_created_at_idx').on(table.createdAt),
uniqueIndex('flux_tx_user_request_uniq')
.on(table.userId, table.requestId)
.where(sql`request_id IS NOT NULL`),
])
apps/server/src/schemas/flux.ts
+2 -2
View File
@@ -1,10 +1,10 @@
import { integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
import { bigint, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
import { user } from './accounts'
export const userFlux = pgTable('user_flux', {
userId: text('user_id').primaryKey().references(() => user.id, { onDelete: 'cascade' }),
flux: integer('flux').notNull().default(0),
flux: bigint('flux', { mode: 'number' }).notNull().default(0),
stripeCustomerId: text('stripe_customer_id'),
updatedAt: timestamp('updated_at').defaultNow().notNull(),
})
apps/server/src/schemas/index.ts
+1 -1
View File
@@ -2,7 +2,7 @@ export * from './accounts'
export * from './characters'
export * from './chats'
export * from './flux'
export * from './flux-ledger'
export * from './flux-transaction'
export * from './llm-request-log'
export * from './providers'
export * from './stripe'
apps/server/src/schemas/llm-request-log.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
import { bigint, integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
import { nanoid } from '../utils/id'
@@ -8,7 +8,7 @@ export const llmRequestLog = pgTable('llm_request_log', {
model: text('model').notNull(),
status: integer('status').notNull(),
durationMs: integer('duration_ms').notNull(),
fluxConsumed: integer('flux_consumed').notNull(),
fluxConsumed: bigint('flux_consumed', { mode: 'number' }).notNull(),
promptTokens: integer('prompt_tokens'),
completionTokens: integer('completion_tokens'),
createdAt: timestamp('created_at').defaultNow().notNull(),
apps/server/src/services/billing/billing-consumer-handler.ts
+4 -4
View File
@@ -4,7 +4,7 @@ import type { BillingEvent } from './billing-events'
import { useLogger } from '@guiiai/logg'
import * as fluxLedgerSchema from '../../schemas/flux-ledger'
import * as fluxTxSchema from '../../schemas/flux-transaction'
import * as llmRequestLogSchema from '../../schemas/llm-request-log'
const logger = useLogger('billing-consumer-handler').useGlobalConfig()
@@ -21,8 +21,8 @@ export function createBillingConsumerHandler(db: Database) {
: 0
// NOTICE: onConflictDoNothing handles redelivery after crash —
// the unique index (userId, requestId) prevents duplicate ledger entries.
await db.insert(fluxLedgerSchema.fluxLedger).values({
// the unique index (userId, requestId) prevents duplicate transaction entries.
await db.insert(fluxTxSchema.fluxTransaction).values({
userId: event.userId,
type: 'debit',
amount: event.payload.amount,
@@ -42,7 +42,7 @@ export function createBillingConsumerHandler(db: Database) {
eventId: event.eventId,
userId: event.userId,
amount: event.payload.amount,
}).log('Wrote debit ledger + audit')
}).log('Wrote debit transaction')
break
}
apps/server/src/services/billing/billing-service.ts
+19 -16
View File
@@ -14,7 +14,7 @@ import { nanoid } from '../../utils/id'
import { userFluxRedisKey } from '../../utils/redis-keys'
import * as fluxSchema from '../../schemas/flux'
import * as fluxLedgerSchema from '../../schemas/flux-ledger'
import * as fluxTxSchema from '../../schemas/flux-transaction'
import * as stripeSchema from '../../schemas/stripe'
const logger = useLogger('billing-service')
@@ -59,7 +59,7 @@ export function createBillingService(
/**
* Debit flux from a user's balance within a DB transaction.
* The transaction ONLY locks the row and updates the balance.
* Ledger + audit entries are written by the billing-mq consumer
* Transaction entries are written by the billing-mq consumer
* after it processes the flux.debited event published post-commit.
*
* Private — call domain-specific wrappers (e.g. consumeFluxForLLM) instead.
@@ -103,7 +103,7 @@ export function createBillingService(
// 3. Update Redis cache after commit (best-effort)
await updateRedisCache(input.userId, result.flux)
// 4. Publish flux.debited event to stream; ledger + audit written by consumer
// 4. Publish flux.debited event to stream; transaction + audit written by consumer
await publishEvent({
eventId: nanoid(),
eventType: 'flux.debited',
@@ -129,13 +129,14 @@ export function createBillingService(
/**
* Debit flux for an LLM API request (chat, TTS, ASR).
* Passes token usage as opaque metadata carried through the flux.debited event
* so the billing-mq consumer can write it to the ledger.
* so the billing-mq consumer can write it to the transaction log.
*/
async consumeFluxForLLM(input: {
userId: string
amount: number
requestId?: string
description?: string
model?: string
promptTokens?: number
completionTokens?: number
}): Promise<{ userId: string, flux: number }> {
@@ -145,16 +146,18 @@ export function createBillingService(
requestId: input.requestId,
description: input.description,
source: 'llm.request',
metadata: input.promptTokens != null || input.completionTokens != null
? { promptTokens: input.promptTokens, completionTokens: input.completionTokens }
: undefined,
metadata: {
...(input.model != null && { model: input.model }),
...(input.promptTokens != null && { promptTokens: input.promptTokens }),
...(input.completionTokens != null && { completionTokens: input.completionTokens }),
},
})
},
/**
* Credit flux to a user's balance within a DB transaction.
* Generic credit method for non-Stripe flows (e.g. admin grants).
* Ledger + audit entries are written inside the transaction for immediate visibility.
* Transaction entries are written inside the transaction for immediate visibility.
*/
async creditFlux(input: {
userId: string
@@ -185,8 +188,8 @@ export function createBillingService(
.set({ flux: balanceAfter, updatedAt: new Date() })
.where(eq(fluxSchema.userFlux.userId, input.userId))
// Ledger entry
await tx.insert(fluxLedgerSchema.fluxLedger).values({
// Transaction entry
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: 'credit',
amount: input.amount,
@@ -225,7 +228,7 @@ export function createBillingService(
/**
* Credit flux from a Stripe checkout session (one-time payment).
* Idempotent: checks fluxCredited flag before applying.
* Ledger + audit entries are written inside the transaction for immediate visibility.
* Transaction entries are written inside the transaction for immediate visibility.
*/
async creditFluxFromStripeCheckout(input: {
stripeEventId: string
@@ -276,8 +279,8 @@ export function createBillingService(
const description = `Stripe payment ${input.currency?.toUpperCase() ?? 'UNKNOWN'} ${(input.amountTotal / 100).toFixed(2)}`
// Ledger entry
await tx.insert(fluxLedgerSchema.fluxLedger).values({
// Transaction entry
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: 'credit',
amount: input.fluxAmount,
@@ -338,7 +341,7 @@ export function createBillingService(
/**
* Credit flux from a Stripe invoice payment (subscription).
* Idempotent: checks fluxCredited flag on the invoice record.
* Ledger + audit entries are written inside the transaction for immediate visibility.
* Transaction entries are written inside the transaction for immediate visibility.
*/
async creditFluxFromInvoice(input: {
stripeEventId: string
@@ -388,8 +391,8 @@ export function createBillingService(
const description = `Subscription invoice ${input.currency.toUpperCase()} ${(input.amountPaid / 100).toFixed(2)}`
// Ledger entry
await tx.insert(fluxLedgerSchema.fluxLedger).values({
// Transaction entry
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: 'credit',
amount: input.fluxAmount,
apps/server/src/services/billing/tests/billing-consumer-handler.test.ts
+4 -4
View File
@@ -22,10 +22,10 @@ describe('billingConsumerHandler', () => {
})
beforeEach(async () => {
await db.delete(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-billing-handler-1'))
await db.delete(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, 'user-billing-handler-1'))
})
it('writes debit ledger metadata so token usage can be shown in the UI', async () => {
it('writes debit transaction metadata so token usage can be shown in the UI', async () => {
const handler = createBillingConsumerHandler(db)
await handler.handleMessage({
@@ -48,9 +48,9 @@ describe('billingConsumerHandler', () => {
},
})
const [ledgerRecord] = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.requestId, 'req-1'))
const [txRecord] = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.requestId, 'req-1'))
expect(ledgerRecord).toMatchObject({
expect(txRecord).toMatchObject({
userId: 'user-billing-handler-1',
type: 'debit',
amount: 3,
apps/server/src/services/billing/tests/billing-service.test.ts
+19 -19
View File
@@ -67,7 +67,7 @@ describe('billingService', () => {
billingMq = createMockBillingMq()
billingService = createBillingService(db, redis, billingMq, createMockConfigKV())
await db.delete(schema.fluxLedger)
await db.delete(schema.fluxTransaction)
await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
await db.delete(schema.stripeCheckoutSession).where(eq(schema.stripeCheckoutSession.stripeSessionId, 'sess-billing-1'))
@@ -84,7 +84,7 @@ describe('billingService', () => {
})
describe('creditFluxFromStripeCheckout', () => {
it('credits flux, records ledger + audit, and enqueues outbox events in one transaction', async () => {
it('credits flux, records transaction, and enqueues outbox events in one transaction', async () => {
const result = await billingService.creditFluxFromStripeCheckout({
stripeEventId: 'stripe-evt-1',
userId: 'user-billing-1',
@@ -99,16 +99,16 @@ describe('billingService', () => {
const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
expect(fluxRecord?.flux).toBe(50)
// Verify ledger entry
const ledgerRecords = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-billing-1'))
expect(ledgerRecords).toHaveLength(1)
expect(ledgerRecords[0]?.type).toBe('credit')
expect(ledgerRecords[0]?.amount).toBe(50)
expect(ledgerRecords[0]?.balanceBefore).toBe(0)
expect(ledgerRecords[0]?.balanceAfter).toBe(50)
// Verify transaction entry
const txRecords = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, 'user-billing-1'))
expect(txRecords).toHaveLength(1)
expect(txRecords[0]?.type).toBe('credit')
expect(txRecords[0]?.amount).toBe(50)
expect(txRecords[0]?.balanceBefore).toBe(0)
expect(txRecords[0]?.balanceAfter).toBe(50)
// Verify metadata on ledger entry
expect(ledgerRecords[0]?.metadata).toMatchObject({
// Verify metadata on transaction entry
expect(txRecords[0]?.metadata).toMatchObject({
stripeEventId: 'stripe-evt-1',
stripeSessionId: 'sess-billing-1',
source: 'stripe.checkout.completed',
@@ -173,7 +173,7 @@ describe('billingService', () => {
const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
expect(fluxRecord?.flux).toBe(70)
// Verify flux.debited event published to stream (ledger + audit written by consumer)
// Verify flux.debited event published to stream (transaction written by consumer)
expect(billingMq.publish).toHaveBeenCalledTimes(1)
expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({
eventType: 'flux.debited',
@@ -202,8 +202,8 @@ describe('billingService', () => {
const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
expect(fluxRecord?.flux).toBe(5)
const ledgerRecords = await db.select().from(schema.fluxLedger)
expect(ledgerRecords).toHaveLength(0)
const txRecords = await db.select().from(schema.fluxTransaction)
expect(txRecords).toHaveLength(0)
// Verify no event was published
expect(billingMq.publish).not.toHaveBeenCalled()
@@ -211,7 +211,7 @@ describe('billingService', () => {
})
describe('creditFlux', () => {
it('credits balance with ledger + audit + outbox', async () => {
it('credits balance with transaction + outbox', async () => {
const result = await billingService.creditFlux({
userId: 'user-billing-1',
amount: 50,
@@ -222,10 +222,10 @@ describe('billingService', () => {
expect(result.balanceAfter).toBe(50)
expect(result.balanceBefore).toBe(0)
// Verify ledger
const ledgerRecords = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-billing-1'))
expect(ledgerRecords).toHaveLength(1)
expect(ledgerRecords[0]).toMatchObject({
// Verify transaction
const txRecords = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, 'user-billing-1'))
expect(txRecords).toHaveLength(1)
expect(txRecords[0]).toMatchObject({
type: 'credit',
amount: 50,
balanceBefore: 0,
apps/server/src/services/config-kv.ts
+2
View File
@@ -34,6 +34,8 @@ const ConfigEntrySchemas = {
MAX_CHECKOUT_AMOUNT_CENTS: optional(number(), 1_000_000),
GATEWAY_BASE_URL: string(),
DEFAULT_CHAT_MODEL: string(),
AUTH_RATE_LIMIT_MAX: optional(number(), 20),
AUTH_RATE_LIMIT_WINDOW_SEC: optional(number(), 60),
} as const
type ConfigDefinitions = {
apps/server/src/services/flux-audit.ts
-52
View File
@@ -1,52 +0,0 @@
import type { Database } from '../libs/db'
import { useLogger } from '@guiiai/logg'
import { desc, eq } from 'drizzle-orm'
import * as schema from '../schemas/flux-ledger'
const logger = useLogger('flux-audit')
export interface AuditEntry {
userId: string
type: 'credit' | 'debit' | 'initial'
amount: number
balanceBefore: number
balanceAfter: number
requestId?: string
description: string
metadata?: Record<string, unknown>
}
export function createFluxAuditService(db: Database) {
return {
async log(entry: AuditEntry) {
await db.insert(schema.fluxLedger).values(entry)
logger.withFields({ userId: entry.userId, type: entry.type, amount: entry.amount }).log('Audit entry recorded')
},
async logBatch(entries: AuditEntry[]) {
if (entries.length === 0)
return
await db.insert(schema.fluxLedger).values(entries)
logger.withFields({ count: entries.length }).log('Audit batch recorded')
},
async getHistory(userId: string, limit: number, offset: number) {
const records = await db.query.fluxLedger.findMany({
where: eq(schema.fluxLedger.userId, userId),
orderBy: [desc(schema.fluxLedger.createdAt)],
limit: limit + 1, // fetch one extra to determine hasMore
offset,
})
const hasMore = records.length > limit
if (hasMore)
records.pop()
return { records, hasMore }
},
}
}
export type FluxAuditService = ReturnType<typeof createFluxAuditService>
apps/server/src/services/flux-transaction.ts
+52
View File
@@ -0,0 +1,52 @@
import type { Database } from '../libs/db'
import { useLogger } from '@guiiai/logg'
import { desc, eq } from 'drizzle-orm'
import * as schema from '../schemas/flux-transaction'
const logger = useLogger('flux-transaction')
export interface TransactionEntry {
userId: string
type: 'credit' | 'debit' | 'initial'
amount: number
balanceBefore: number
balanceAfter: number
requestId?: string
description: string
metadata?: Record<string, unknown>
}
export function createFluxTransactionService(db: Database) {
return {
async log(entry: TransactionEntry) {
await db.insert(schema.fluxTransaction).values(entry)
logger.withFields({ userId: entry.userId, type: entry.type, amount: entry.amount }).log('Transaction recorded')
},
async logBatch(entries: TransactionEntry[]) {
if (entries.length === 0)
return
await db.insert(schema.fluxTransaction).values(entries)
logger.withFields({ count: entries.length }).log('Transaction batch recorded')
},
async getHistory(userId: string, limit: number, offset: number) {
const records = await db.query.fluxTransaction.findMany({
where: eq(schema.fluxTransaction.userId, userId),
orderBy: [desc(schema.fluxTransaction.createdAt)],
limit: limit + 1, // fetch one extra to determine hasMore
offset,
})
const hasMore = records.length > limit
if (hasMore)
records.pop()
return { records, hasMore }
},
}
}
export type FluxTransactionService = ReturnType<typeof createFluxTransactionService>
apps/server/src/services/flux.ts
+4 -4
View File
@@ -9,7 +9,7 @@ import { eq } from 'drizzle-orm'
import { userFluxRedisKey } from '../utils/redis-keys'
import * as schema from '../schemas/flux'
import * as fluxLedgerSchema from '../schemas/flux-ledger'
import * as fluxTxSchema from '../schemas/flux-transaction'
const logger = useLogger('flux-service')
@@ -30,16 +30,16 @@ export function createFluxService(db: Database, redis: Redis, configKV: ConfigKV
if (!record) {
const initialFlux = await configKV.getOrThrow('INITIAL_USER_FLUX')
// Transaction: create user_flux + flux_ledger atomically
// Transaction: create user_flux + flux_transaction atomically
await db.transaction(async (tx) => {
const [inserted] = await tx.insert(schema.userFlux)
.values({ userId, flux: initialFlux })
.onConflictDoNothing({ target: schema.userFlux.userId })
.returning()
// Only write ledger if we actually created the record (not a conflict)
// Only write transaction if we actually created the record (not a conflict)
if (inserted) {
await tx.insert(fluxLedgerSchema.fluxLedger).values({
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId,
type: 'initial',
amount: initialFlux,
apps/server/src/services/tests/flux-audit.test.ts → apps/server/src/services/tests/flux-transaction.test.ts
+18 -18
View File
@@ -1,27 +1,27 @@
import { beforeAll, describe, expect, it } from 'vitest'
import { mockDB } from '../../libs/mock-db'
import { createFluxAuditService } from '../flux-audit'
import { createFluxTransactionService } from '../flux-transaction'
import * as schema from '../../schemas'
describe('fluxAuditService', () => {
describe('fluxTransactionService', () => {
let db: any
let service: ReturnType<typeof createFluxAuditService>
let service: ReturnType<typeof createFluxTransactionService>
beforeAll(async () => {
db = await mockDB(schema)
await db.insert(schema.user).values({
id: 'user-audit',
name: 'Audit User',
email: 'audit@example.com',
id: 'user-tx',
name: 'Transaction User',
email: 'tx@example.com',
})
service = createFluxAuditService(db)
service = createFluxTransactionService(db)
})
it('log should insert a single ledger entry', async () => {
it('log should insert a single transaction entry', async () => {
await service.log({
userId: 'user-audit',
userId: 'user-tx',
type: 'credit',
amount: 500,
balanceBefore: 0,
@@ -30,7 +30,7 @@ describe('fluxAuditService', () => {
metadata: { stripeSessionId: 'sess_123' },
})
const { records } = await service.getHistory('user-audit', 10, 0)
const { records } = await service.getHistory('user-tx', 10, 0)
expect(records).toHaveLength(1)
expect(records[0].type).toBe('credit')
expect(records[0].amount).toBe(500)
@@ -38,39 +38,39 @@ describe('fluxAuditService', () => {
it('logBatch should insert multiple entries', async () => {
await service.logBatch([
{ userId: 'user-audit', type: 'debit', amount: 10, balanceBefore: 500, balanceAfter: 490, description: 'gpt-4o' },
{ userId: 'user-audit', type: 'debit', amount: 5, balanceBefore: 490, balanceAfter: 485, description: 'gpt-4o-mini' },
{ userId: 'user-tx', type: 'debit', amount: 10, balanceBefore: 500, balanceAfter: 490, description: 'gpt-4o' },
{ userId: 'user-tx', type: 'debit', amount: 5, balanceBefore: 490, balanceAfter: 485, description: 'gpt-4o-mini' },
])
const { records } = await service.getHistory('user-audit', 10, 0)
const { records } = await service.getHistory('user-tx', 10, 0)
expect(records).toHaveLength(3) // 1 from previous test + 2 batch
})
it('logBatch with empty array should be a no-op', async () => {
await service.logBatch([])
const { records } = await service.getHistory('user-audit', 10, 0)
const { records } = await service.getHistory('user-tx', 10, 0)
expect(records).toHaveLength(3)
})
it('getHistory should paginate correctly with hasMore', async () => {
const { records, hasMore } = await service.getHistory('user-audit', 2, 0)
const { records, hasMore } = await service.getHistory('user-tx', 2, 0)
expect(records).toHaveLength(2)
expect(hasMore).toBe(true)
})
it('getHistory should return hasMore=false on last page', async () => {
const { records, hasMore } = await service.getHistory('user-audit', 10, 0)
const { records, hasMore } = await service.getHistory('user-tx', 10, 0)
expect(records).toHaveLength(3)
expect(hasMore).toBe(false)
})
it('getHistory should respect offset', async () => {
const { records } = await service.getHistory('user-audit', 10, 2)
const { records } = await service.getHistory('user-tx', 10, 2)
expect(records).toHaveLength(1)
})
it('getHistory should return records ordered by createdAt desc', async () => {
const { records } = await service.getHistory('user-audit', 10, 0)
const { records } = await service.getHistory('user-tx', 10, 0)
for (let i = 1; i < records.length; i++) {
expect(new Date(records[i - 1].createdAt).getTime())
.toBeGreaterThanOrEqual(new Date(records[i].createdAt).getTime())
apps/server/src/services/tests/flux.test.ts
+5 -5
View File
@@ -55,7 +55,7 @@ describe('fluxService (DB-backed)', () => {
service = createFluxService(db, redis, createMockConfigKV())
// Clean up flux-related tables
await db.delete(schema.fluxLedger).where(eq(schema.fluxLedger.userId, testUser.id))
await db.delete(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, testUser.id))
await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, testUser.id))
})
@@ -65,12 +65,12 @@ describe('fluxService (DB-backed)', () => {
expect(redis.set).toHaveBeenCalledWith(userFluxRedisKey(testUser.id), '100')
})
it('getFlux should write a ledger entry on initialization', async () => {
it('getFlux should write a transaction entry on initialization', async () => {
await service.getFlux(testUser.id)
const ledgerRecords = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, testUser.id))
expect(ledgerRecords).toHaveLength(1)
expect(ledgerRecords[0]).toMatchObject({
const txRecords = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, testUser.id))
expect(txRecords).toHaveLength(1)
expect(txRecords[0]).toMatchObject({
type: 'initial',
amount: 100,
balanceBefore: 0,