refactor(server): drop redis stream + worker role (#1792)

The Redis Stream `billing-events` + `worker` Railway role +
advisory-lock poller layered together didn't actually buy us reliability
— `debitFlux` swallowed XADD failures, leaving the door open to "balance
updated, ledger row never written". Collapse the whole thing back to:
`creditFlux` and `debitFlux` write `flux_transaction` ledger rows inline
within the same DB transaction that mutates `user_flux`, and `(user_id,
request_id)` remains the partial unique index that keeps retries safe.

Concrete changes:
- Inline ledger inserts in `BillingService.{debitFlux, creditFlux,
creditFluxFromStripeCheckout, creditFluxFromInvoice}`; drop `billingMq`
and `publishEvent` plumbing entirely.
- `routes/openai/v1` writes `llm_request_log` synchronously via the
existing `requestLogService`; the duplicate `llm-request-log.ts` service
module is removed.
- `bin/run-worker.ts`, `libs/mq/*`,
`services/billing/billing-events.ts`,
`services/billing/billing-consumer-handler.ts`, and matching tests are
deleted. CLI now exposes only `api`.
- `BILLING_EVENTS_*` env vars and the `DEFAULT_BILLING_EVENTS_STREAM`
helper are dropped; `docker-compose.yml` no longer ships a worker
service.
- `docs/ai-context/{workers-and-runtime, billing-architecture,
redis-boundaries-and-pubsub, data-model-and-state,
architecture-overview, README}.md`, `CLAUDE.md`, and the existing
verification docs are updated to describe the single-process synchronous
pipeline.

Tests: 29 files / 247 cases pass. Production deployments need to drop
the worker Railway service after this lands.
This commit is contained in:
RainbowBird
2026-05-08 21:14:01 +08:00
parent eba047328a
commit f8d1fa7a64
38 changed files with 1217 additions and 1854 deletions
+9 -7
View File
@@ -9,9 +9,9 @@ Hono-based Node.js backend. Owns auth, billing, chat sync, LLM gateway forwardin
## Deployment Model
- Hosted on **Railway**, multiple instances behind a load balancer.
- Each instance runs one CLI role: `api` or `billing-consumer` (see `src/bin/run.ts`).
- Single CLI role: `api` (see `src/bin/run.ts`). No background polling loops, no fire-and-forget tasks — every write happens inside the request thread.
- Stateless per-instance: no local state that matters across requests.
- Cross-instance coordination via Redis Pub/Sub (WebSocket broadcast) and Redis Streams (billing events).
- Cross-instance coordination via Redis Pub/Sub (WebSocket broadcast). DB-level idempotency (`(userId, requestId)` partial unique index on `flux_transaction`) covers retries.
- Rate limiting is currently **in-memory** (not distributed) — keep this in mind when adding rate-sensitive features.
## Tech Stack
@@ -47,11 +47,12 @@ Local observability: `docker compose -f apps/server/docker-compose.otel.yml up -
## Key Design Decisions
- **Flux read/write separation**: `FluxService` reads (Redis cache-aside), `BillingService` writes (Postgres tx + Redis Stream XADD). Never put write-balance logic in `flux.ts`.
- **Flux read/write separation**: `FluxService` reads (Redis cache-aside), `BillingService` writes (single Postgres tx that mutates `user_flux` and writes the matching `flux_transaction` ledger row). Never put write-balance logic in `flux.ts`.
- **No async billing pipeline**: debits and credits update balance + ledger in one transaction. The `(user_id, request_id)` partial unique index gives DB-level idempotency for retries; LLM `request log` rows are written best-effort right after the response is delivered.
- **LLM gateway proxy**: `/api/v1/openai` forwards to `GATEWAY_BASE_URL`. Server handles auth/billing/logging — not model execution.
- **Redis is cache + messaging, not truth**: balance cache, app_settings read cache, WS cross-instance pub/sub, billing event streams. Truth is always Postgres.
- **Redis is cache + pub/sub, not truth**: balance cache, app_settings read cache, WebSocket cross-instance pub/sub. Truth is always Postgres.
- **Auth**: Better Auth + OIDC. `sessionMiddleware` fills context but doesn't block; `authGuard` returns 401.
- **Multi-instance safe**: all writes go through Postgres transactions; cross-instance messaging uses Redis Pub/Sub and Streams. No in-process singletons that hold mutable state across requests.
- **Multi-instance safe**: all writes go through Postgres transactions; cross-instance messaging uses Redis Pub/Sub. No async work, no in-process singletons — admin flux grants happen synchronously inside the POST that triggered them.
## Detailed Context Docs
@@ -59,9 +60,10 @@ See `docs/ai-context/README.md` for the full index. Key files:
- `architecture-overview.md` — entry, DI, assembly, boundaries
- `transport-and-routes.md` — API surface, route→service mapping
- `data-model-and-state.md` — tables, state ownership, caching
- `billing-architecture.md` — Flux/Stripe/outbox/Streams
- `billing-architecture.md` — Flux/Stripe ledger
- `redis-boundaries-and-pubsub.md` — Redis key/channel boundaries
- `auth-and-oidc.md` — auth flows, OIDC, trusted clients
- `config-and-naming-conventions.md` — configKV, naming rules
- `workers-and-runtime.md`CLI roles, outbox, Streams consumer
- `workers-and-runtime.md`single `api` role, no background loops, no fire-and-forget; everything is synchronous in-request
- `admin-flux-grants.md` — synchronous one-shot flux grant endpoint (no batch tables, no state machine)
- `observability-conventions.md` — OTel naming, custom attributes
-17
View File
@@ -53,23 +53,6 @@ services:
timeout: 5s
retries: 5
billing-consumer:
build:
context: ../..
dockerfile: apps/server/Dockerfile
command: ['pnpm', '-F', '@proj-airi/server', 'run', 'server', 'billing-consumer']
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
env_file:
- path: .env
required: false
- path: .env.local
required: false
restart: unless-stopped
volumes:
db_data:
driver: local
+8 -8
View File
@@ -16,13 +16,13 @@
- `data-model-and-state.md`
- 主要表、状态归属、缓存与事件模型
- `workers-and-runtime.md`
- CLI 角色、outbox dispatcher、Redis Streams consumer、运行时约束
- `api` role、进程内后台 loop、advisory lock 协调、运行时约束
- `redis-boundaries-and-pubsub.md`
- Redis key / channel 收口、Pub/Sub / Streams 边界、运行时校验约束
- Redis key / channel 收口、Pub/Sub 边界、运行时校验约束
- `config-and-naming-conventions.md`
- `configKV` 默认值来源、Redis key 命名、HTTP route 命名、后续收敛 TODO
- `billing-architecture.md`
- 计费链路专项说明,重点看 Flux / Stripe / outbox / Redis Streams
- 计费链路专项说明,重点看 Flux ledger / Stripe 幂等
- `flux-meter.md`
- Sub-Flux 计量服务(TTS/STT 等)的债务账本机制与复用指南
- `observability-conventions.md`
@@ -33,8 +33,8 @@
- Resend 接入、Better Auth 四个邮件 callback、范围 / 决策 / 不做项
- `account-deletion.md`
- 账号注销架构:auth 表 hard delete + 业务表软删,handler 协议、各业务行为、failure 模型
- `admin-flux-grant-batch.md`
- Admin 批量发 FLUX(活动赠送)架构:`flux_grant_batch` + `_recipient` 表、worker 调度、`adminGuard` 邮箱白名单、failure / retry 模型
- `admin-flux-grants.md`
- Admin 批量发 FLUX(活动赠送):单一同步 POST,无 batch 表无后台 loop`adminGuard` 邮箱白名单 + 可选 `idempotencyKey`
- `verifications/email-auth.md`
- 邮箱注册 / 忘记密码 / OIDC 桥接登录 三条用户路径的真实实测证据
- `verifications/account-deletion.md`
@@ -45,7 +45,7 @@
- `apps/server/src/app.ts` 是唯一的 API 应用装配入口。
- 服务端采用 `Hono + injeca + Drizzle + Redis + better-auth`
- 路由层整体较薄,业务逻辑主要在 `src/services/`
- **Postgres 是所有余额与计费状态的唯一真相源**,Redis 只做缓存、KV、Pub/SubStreams。
- **Postgres 是所有余额与计费状态的唯一真相源**,Redis 只做缓存、KV、Pub/Sub。计费链路不再使用 Redis Streams。
- WebSocket 只用于聊天同步,跨实例广播依赖 Redis Pub/Sub。
- 对外 LLM 能力不是本地推理,而是转发到配置里的 gateway,再按 usage / fallback rate 扣 Flux。
@@ -54,8 +54,8 @@
- 改 API 入口或新增依赖:先看 `architecture-overview.md`
- 改某个接口行为:先看 `transport-and-routes.md`
- 改表结构、缓存或幂等:先看 `data-model-and-state.md`
- worker、部署角色、事件处理:先看 `workers-and-runtime.md`
- 改 Redis key、Pub/Sub、Streams 边界:先看 `redis-boundaries-and-pubsub.md`
-后台 loop、部署形态:先看 `workers-and-runtime.md`
- 改 Redis key、Pub/Sub 边界:先看 `redis-boundaries-and-pubsub.md`
- 改配置默认值、Redis key 命名、HTTP route 命名:先看 `config-and-naming-conventions.md`
- 改扣费、充值、Stripe:先看 `billing-architecture.md`
- 改 trace / metric attributes、OTel 命名:先看 `observability-conventions.md`
@@ -0,0 +1,106 @@
# Admin Flux Grants
Admin 一次性给若干用户发 FLUX(Beta 致谢、补偿、运营赠送等)的接口。整个流程**单一同步 HTTP 调用**搞定,没有 batch 表、没有状态机、没有后台 loop。
## 1. 背景
旧设计是 `flux_grant_batch` + `flux_grant_batch_recipient` 两张表 + 状态机 + 异步处理 + retry 端点 + advisory-lock poller~800 行代码。实际产品里 admin 发放频率"几周一次、几十个用户",过度工程。简化为:
- 一个 `POST /api/admin/flux-grants` 接口
- 同步处理:resolve emails → 顺序调 `creditFlux` → 返回每条的 outcome
- 审计走 `flux_transaction` 表(`type='promo'``metadata.description` / `metadata.idempotencyKey`
- 失败处理:admin 看响应里的 `failed[]`,自己再发一次(用 `idempotencyKey` 防止已成功的部分被双发)
## 2. 路由
`POST /api/admin/flux-grants?dryRun=true|false`
Auth`authGuard` + `adminGuard``ADMIN_EMAILS` allowlist + 验证邮箱)。
Body
```ts
{
description: string, // 1..500 chars; 写入 flux_transaction.metadata.description
amount: number, // 1..MAX_GRANT_AMOUNT_PER_USER (10_000), 单人发放数量
emails: string[], // 1..MAX_EMAILS_PER_GRANT (200) 个 email
idempotencyKey?: string, // 可选,最长 100 chars。提供后每个 recipient 的
// requestId = `flux-grant:${idempotencyKey}:${userId}`
// 重发同 (key, recipients) 是 no-op;不提供则每次 grant
// 都会重发。
}
```
dry-run 响应:
```ts
{ preview: { totalEmails, willGrant, willSkip: { notFound, userDeleted, duplicateInInput }, totalFluxToIssue, samples } }
```
实发响应:
```ts
{
summary: { totalEmails, willGrant, willSkip, totalFluxToIssue, samples },
result: {
granted: [{ email, userId, fluxTransactionId, balanceAfter }],
skipped: [{ email, reason: 'duplicate_in_input' | 'not_found' | 'user_deleted' }],
failed: [{ email, userId, error }], // creditFlux 抛错时进这里
},
}
```
## 3. 处理流程(同步)
1. 路由层 valibot 校验 body
2. `service.resolveEmails(emails)`
- 输入小写化后 `IN (...)``user.email`(不能 wrap `LOWER()`,会 break unique index → seq scan
- 命中 user 后再查 `user_flux.deletedAt`
- 重复输入按出现顺序首条留下,后续标 `duplicate_in_input`
3. 对每个 `status='pending'` 的 recipient 顺序调 `BillingService.creditFlux({ userId, amount, type: 'promo', requestId, description, source: 'admin_promo', auditMetadata })`
4. 抛错记到 `result.failed[]`,循环继续;成功记到 `result.granted[]`
5. HTTP 返回完整 `result`
没有 sleep / throttle —— 200 个 recipient × 2050ms 单条 ≈ 410s,安心进 LB 30s 超时窗口。如果以后真的需要更大批量,先评估是否值得拆,再决定加 cap 还是引入异步。
## 4. 失败 / 恢复
| 故障 | 表现 | 恢复 |
|---|---|---|
| 单条 recipient `creditFlux` 抛错(DB blip 等) | 出现在 `result.failed[]` | admin 看响应,自己再发一次相同请求;如果用了 `idempotencyKey`,已 granted 的不会被双发,只重试 failed 的 |
| 整个请求超 LB 超时 | 客户端看到超时,部分 recipient 已扣账 | admin 用同 `idempotencyKey` 重发,已成功的直接幂等跳过 |
| Operator 输错邮箱 / 数量 | 先用 `?dryRun=true` 看 preview | 改完再去掉 dryRun |
## 5. 审计
- 每条成功 grant 在 `flux_transaction` 写一行(`type='promo'``metadata.description``metadata.issuedByUserId`、可选 `metadata.idempotencyKey`
- 没有专门的 admin 报表;用 `/api/v1/flux/history` 或直接 SQL 按 `metadata->>'description'` / `metadata->>'idempotencyKey'`
```sql
-- 看某次 grant 实际发了多少
SELECT user_id, amount, balance_after, created_at
FROM flux_transaction
WHERE metadata->>'idempotencyKey' = 'beta-2026-q2'
ORDER BY created_at;
```
## 6. 实现位置
- 路由:[`apps/server/src/routes/admin/flux-grants/index.ts`](apps/server/src/routes/admin/flux-grants/index.ts)
- Service[`apps/server/src/services/admin-flux-grants/index.ts`](apps/server/src/services/admin-flux-grants/index.ts)
- 单测:[`apps/server/src/services/admin-flux-grants/tests/admin-flux-grants.test.ts`](apps/server/src/services/admin-flux-grants/tests/admin-flux-grants.test.ts)
- adminGuard[`apps/server/src/middlewares/admin-guard.ts`](apps/server/src/middlewares/admin-guard.ts)
- 数据库:**没有**专门的表;唯一持久化是 `flux_transaction` ledger
- 已废弃:`flux_grant_batch` / `flux_grant_batch_recipient`drizzle migration `0011_superb_lady_deathstrike.sql` 删表)
## 7. 不做
- 不做 batch 状态机 / retry endpoint —— 同步响应里已经有 failed 列表,admin 看到失败就自己再发
- 不做异步处理 / 后台 loop —— 200 用户上限完全可以塞进一个 HTTP 请求
- 不做 dashboard 展示 —— 直接查 `flux_transaction` 即可
- 不做高并发 / 大批量 —— 这是 admin 工具不是 bulk import;超 200 就让 admin 拆请求
## 8. 已知不足
- **无 admin-side 失败留痕**`failed[]` 只在 HTTP 响应里返回一次,admin 关掉浏览器就没了。如果将来发现需要"上次失败的那批"持久化,再单独加一张 `admin_grant_attempt_log` 之类的,不要把它做回 batch 表。
- **`emails` 上限 200 是经验估算**:单 `creditFlux` 假设 20–50ms。如果实际生产数据显示更慢,下调上限。
@@ -27,10 +27,9 @@
- 注入 WebSocket
- 绑定 `uncaughtException` / `unhandledRejection`
CLI 入口在 `src/bin/run.ts`支持两种角色:
CLI 入口在 `src/bin/run.ts`只有一种角色:
- `api`
- `billing-consumer`
- `api`HTTP/WS;没有常驻后台 loop,也没有 fire-and-forget 异步任务。admin flux grant 在 POST 请求线程内同步处理完返回;详见 `workers-and-runtime.md`
## 依赖注入结构
@@ -131,7 +130,7 @@ CLI 入口在 `src/bin/run.ts`,支持两种角色:
- 新用户首次读取时初始化余额
- `BillingService`
- 面向写入
- debitFlux:事务内更新余额,事务后 XADD Redis Streamcredit 方法:事务内同步写流水和审计
- debitFlux / credit 方法:事务内同步更新余额并写 `flux_transaction` ledger;事务提交后 best-effort 刷 Redis 余额缓存
这是服务端最重要的边界之一,尽量不要把写余额逻辑重新塞回 `flux.ts`
@@ -1,53 +1,47 @@
# Distributed Billing Plan
# Billing Architecture
## 架构概述
`apps/server` 的计费链采用 **Postgres 作为唯一账本真相源**,Redis 仅作缓存。余额变化路径分两类:`debitFlux` 在 DB 事务内只做 `UPDATE user_flux`transaction/请求日志通过 Redis Stream 异步写入;credit 方法仍在事务内同步写入 transaction log
`apps/server` 的计费链**Postgres 唯一账本真相源,所有余额写操作(debit / credit)和 ledger 行写入都在同一个 DB 事务里完成**。Redis 只承担余额读缓存。不再使用 Redis Stream / 后台 consumer 处理计费副作用
### 数据模型
- **`user_flux`** — 用户余额快照(单行/用户)
- **`flux_transaction`** — append-only 账务流水(type: credit/debit/initial, amount, balanceBefore, balanceAfter, requestId
- partial unique index `(userId, requestId) WHERE requestId IS NOT NULL`DB 层幂等防重
- **`flux_transaction`** — 用户可见的历史记录
- **`flux_transaction`** — append-only 账务流水(type: credit / debit / initial / promo, amount, balanceBefore, balanceAfter, requestId, metadata
- partial unique index `(userId, requestId) WHERE requestId IS NOT NULL`DB 层幂等防重
- **`llm_request_log`** — 每个 LLM/TTS 请求的可观测记录(model / status / duration / fluxConsumed / token 用量)
### debitFlux 链路(已实现)
### debitFlux 链路
DB 事务内仅做
`BillingService.consumeFluxForLLM()` 调用 `debitFlux()`,单个事务内:
1. `SELECT user_flux FOR UPDATE` 锁行
2. 检查余额(不足返回 402
3. 更新 `user_flux.flux`
4. 事务提交后 XADD Redis Stream`billing-events`),携带扣费金额、余额快照、requestId 等
5. 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存
1. 若有 `requestId`,先查 `flux_transaction` 是否已存在同 `(userId, requestId)` 行 → 命中则直接返回历史结果,不再扣费、不写新行(幂等回放)
2. `SELECT user_flux FOR UPDATE` 锁行
3. 检查余额(不足返回 402
4. 更新 `user_flux.flux`
5. `INSERT INTO flux_transaction (...)`,把扣费金额、token 用量、source 写进 metadata
6. 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存(失败仅 warn 日志)
transaction log / audit / llm_request_log 的写入均由 **billing-consumer** 异步完成。
### credit 链路
### credit 方法链路(已实现)
`creditFlux()` / `creditFluxFromStripeCheckout()` / `creditFluxFromInvoice()` 全部在事务内同步:
credit 方法(`creditFlux` / `creditFluxFromStripeCheckout` / `creditFluxFromInvoice`)仍在 DB 事务内同步写入 `flux_transaction``flux_transaction`
- claim 行(Stripe 路径)/ 幂等查 `flux_transaction`admin 路径)
-`user_flux` 行 → 加额 → 更新
-`flux_transaction`
- 事务提交后 `redis.set` 更新缓存
### 异步链路(已实现)
Stripe 路径靠 `stripe_checkout_session.fluxCredited` / `stripe_invoice.fluxCredited` 标志做对象级幂等;admin 路径靠 `(userId, requestId)` 唯一索引做幂等。
- **billing-consumer** — 消费 Redis Stream `billing-events`,将 transaction log、LLM 请求日志异步写入 DB
### LLM 请求日志
### 事件模型
OpenAI route (`routes/openai/v1/index.ts`) 在 `consumeFluxForLLM` 完成后调用 `requestLogService.logRequest(...)` 同步写 `llm_request_log`。失败被记为 warn 日志,不阻断已经返回给用户的响应(流式响应已发出,错误兜不回来;非流式情况下 debit 已扣,request log 丢失也只是观测层面的损失)。
Stream: `billing-events`
| Event Type | 触发场景 |
|---|---|
| `flux.debited` | LLM 请求扣费 |
| `flux.credited` | Stripe 充值、管理员授予 |
| `stripe.checkout.completed` | 一次性支付完成 |
| `llm.request.completed` | LLM 请求结束 |
`llm_request_log` 没有 FK,没有二级索引,单纯追加;写入成本可以忽略。
### 进程角色
通过 `src/bin/run.ts` 分角色启动:
- `api` — HTTP 服务
- `billing-consumer` — 消费 Redis Stream,异步写入 transaction log、LLM 请求日志到 DB
只有 `api` 一个 role`src/bin/run.ts`),且没有任何"常驻后台 loop"或"fire-and-forget 异步任务"。所有写路径(包括 admin flux grant)都在请求线程内完成;多实例安全靠 `(userId, requestId)` 幂等索引。详见 [`workers-and-runtime.md`](workers-and-runtime.md)。
### Stripe 定价
@@ -59,62 +53,63 @@ TTS 字符、STT 秒等单价 < 1 Flux 的服务通过 `FluxMeter` 累计零头
## 关键服务
### BillingService (`services/billing-service.ts`)
### BillingService (`services/billing/billing-service.ts`)
所有余额写操作的唯一入口:
- **`debitFlux()`** — 扣费(LLM 请求),事务内:锁行 → 检余额(402) → 更新余额;事务提交后 XADD `flux.debited` 到 Redis Streamtransaction 由 billing-consumer 异步写入
- **`creditFlux()`** — 通用充值
- **`creditFluxFromStripeCheckout()`** — Stripe 一次性支付充值,幂等(`fluxCredited` 标志)
- **`creditFluxFromInvoice()`** — Stripe 订阅发票充值,幂等
- **`consumeFluxForLLM()`** — LLM 请求扣费包装;事务内 `lock → check → update → insert ledger`,提交后刷 Redis 缓存;带 `requestId` 时支持幂等回放
- **`creditFlux()`** — 通用充值admin promo / 普通 credit);幂等
- **`creditFluxFromStripeCheckout()`** — Stripe 一次性支付充值,按 session 幂等
- **`creditFluxFromInvoice()`** — Stripe 订阅发票充值,按 invoice 幂等
### FluxService (`services/flux.ts`)
只负责读操作:
- **`getFlux()`** — Redis cache-aside 读(miss → DB → 填充 Redis),新用户自动初始化 + 写 transaction log(type=initial)
- **`getFlux()`** — Redis cache-aside 读(miss → DB → 填充 Redis),新用户自动初始化
- **`updateStripeCustomerId()`**
### Redis 职责边界
Redis **不是**余额真相源,仅用于:
- `getFlux()` 读缓存(加速,丢失无影响)
- `getFlux()` 读缓存(丢失无影响)
- 配置 KV
- WebSocket 广播
- Redis Streams 事件总线
不再使用 Redis Streams 做计费链路。
## 实现状态
| Phase | 状态 | 关键点 |
|-------|------|--------|
| 1. DB-first 账本 | ✅ 已完成 | `flux_transaction` 表,`SELECT FOR UPDATE` 原子扣减,Redis 降为缓存 |
| 2. Redis Streams 异步写入 | ✅ 已完成 | debitFlux 事务后 XADDbilling-consumer 异步写 transaction/请求日志 |
| 3. Stripe 幂等 | ✅ 已完成 | checkout + invoice 事务内幂等检查 |
| 4. LLM 计费优化 | ⚠️ 部分 | 已有 `requestId` 和 DB 事务扣费,待加 tiktoken fallback |
| 5. 部署拆分 | ✅ 已完成 | `bin/run.ts` 两角色启动(api / billing-consumer |
| 6. 幂等防重 | ✅ 已完成 | `flux_transaction` partial unique index on `(userId, requestId)` |
| 1. DB-first 账本 | ✅ | `flux_transaction` 表,`SELECT FOR UPDATE` 原子扣减,Redis 降为缓存 |
| 2. 同步事务 ledger 写入 | ✅ | debit / credit 在单一事务内同时改余额和写 ledger,不再有 stream consumer |
| 3. Stripe 幂等 | ✅ | checkout + invoice 事务内幂等检查 |
| 4. LLM 计费优化 | ⚠️ | 已有 `requestId` 和 DB 事务扣费,待加 tiktoken fallback |
| 5. 单进程部署 | ✅ | 只剩 `api` roleadmin flux grant 在 POST 请求线程内同步执行,没有后台 loop |
| 6. 幂等防重 | ✅ | `flux_transaction` partial unique index on `(userId, requestId)` + 事务内回放命中检查 |
### 已删除
- `flux-write-back.ts` — 定时回写补偿机制,不再需要
- `FluxService.consumeFlux()` / `addFlux()` — 写操作已移至 BillingService
- `llm_request_log.settled` — 无消费者,已移除
- `outbox_events` 表及 outbox-dispatcher 进程 — 已移除,统一由 billing-consumer 处理异步写入
- `cache-sync-consumer` 进程角色 — 已合并进 billing-consumer
- `flux-write-back.ts` — 定时回写补偿机制
- `FluxService.consumeFlux()` / `addFlux()` — 写操作集中到 BillingService
- `llm_request_log.settled` — 无消费者
- `outbox_events` 表及 outbox-dispatcher 进程
- `cache-sync-consumer` 进程角色
- **Redis Stream `billing-events` + `worker` role + `billing-consumer-handler`** — 异步副作用全部回收到事务内同步执行;不再有“事务提交了但 XADD 失败 → ledger 丢行”的窗口
- 相关 env`BILLING_EVENTS_STREAM` / `BILLING_EVENTS_CONSUMER_NAME` / `BILLING_EVENTS_BATCH_SIZE` / `BILLING_EVENTS_BLOCK_MS` / `BILLING_EVENTS_MIN_IDLE_MS`
## 剩余 TODO
### Phase 5 完善:LLM 计费精度
### LLM 计费精度
当前 LLM 扣费在 gateway 未返回 token 用量时使用固定 fallback rate,不精确:
- [ ] **tiktoken fallback** — gateway 未返回 usage 时,用 tiktoken 从 request messages + response body 自行计算 token 数
- [x] **消除静默失败** — non-streaming: debit 失败直接抛错阻断响应;streaming: 已发送无法撤回,改为 error 级别日志+记录 requestId 便于追查
- [ ] **tiktoken fallback** — gateway 未返回 usage 时用 tiktoken 从 request messages + response body 自算 token 数
- [x] **消除静默失败** — non-streaming: debit 失败直接抛错阻断响应;streaming: 已发送无法撤回,改为 error 级别日志 + 记录 requestId 便于追查
## 明确不做
- 不引入 Kafka / RabbitMQ
- 不拆成多个独立 repo
- 不做预扣模式(无法准确估算 LLM 响应 token 数)
- 中期如角色扩容策略差异大,再考虑拆为 `server-api` / `server-workers` / `server-webhooks`
- 不再为“异步副作用”单独拉一个 worker 进程;事务内同步搞定就够了。如果以后真有阻塞型耗时副作用,单独评估时再说
@@ -171,8 +171,7 @@
- 所有余额写操作
- DB 事务
- debitFlux:事务内仅更新余额;事务后 XADD Redis Streamtransaction log 由 billing-consumer 异步写入
- credit 方法:事务内同步写 transaction
- debitFlux / credit 方法:事务内 lock → check → update `user_flux` → insert `flux_transaction` ledger
- 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存
这是所有 Flux 写路径应收敛到的中心。
@@ -215,22 +214,19 @@
- channel: `chat:broadcast:<userId>`
### 计费事件流
- stream: 默认 `billing-events`
## 幂等与并发控制
### 余额并发
`billingService` 在事务中:
1. `SELECT user_flux FOR UPDATE`
2. 计算新余额
3. 写余额(debitFlux 事务内仅此一步;credit 方法同步写 transaction
4. 事务提交后 XADD Redis StreamdebitFlux)或直接返回(credit
1. (可选)按 `(userId, requestId)` 命中 ledger → 命中即返回,跳过余下步骤
2. `SELECT user_flux FOR UPDATE`
3. 计算新余额
4. `user_flux` + 写 `flux_transaction` ledger
5. 事务提交后 best-effort `redis.set`
这保证同一用户余额更新是串行化的。
这保证同一用户余额更新是串行化的,并且 ledger 行与余额变更在同一原子提交里
### Stripe 幂等
@@ -242,5 +238,4 @@
## 现有代码中的结构信号
- `request-log.ts``llm-request-log.ts` 完全重叠,后者更像旧名残留
- `accounts.ts``auth.ts` 也是重复 schema,后续如果做整理,应先统一真实使用入口再删副本。
- `accounts.ts``auth.ts` 是重复 schema,后续如果做整理,应先统一真实使用入口再删副本
@@ -29,8 +29,6 @@
- 例如 `config:{key}`
- Pub/Sub
- 例如聊天跨实例广播 `chat:{userId}:broadcast`
- Streams
- 例如 `billing-events`
- 计量债务账本(atomic counter + TTL
- 例如 TTS 累计字符 `user:{userId}:flux-meter:tts:debt`
- 见 [flux-meter.md](flux-meter.md)
@@ -39,7 +37,8 @@
- Postgres 是余额、账本、订单、聊天消息等持久状态的唯一真相源
- Redis Pub/Sub 只负责降低跨实例通知延迟,不提供持久化、回放、补偿
- Redis Streams 用于异步事件消费,但也必须在边界层做输入输出校验
> NOTICE: Redis Streams(曾经的 `billing-events`)已被移除。现在没有任何业务依赖 Stream 抽象,未来如果要再上 Stream,请先重新评估是否真的需要异步副作用,而不是把它作为默认选项。
## Key / Channel 收口规则
@@ -129,21 +128,6 @@ const data = JSON.parse(message) as BroadcastMessage
如果消息不合法,应该记录错误并丢弃,而不是继续广播到本地连接。
## Streams 边界规则
Redis Streams 的参考实现已经在 `src/libs/mq/stream.ts` 里。
这层模式值得复用的点有两个:
- 输入通过 `serialize()` 收口
- 输出通过 `deserialize()` 和运行时检查收口
也就是说:
- 不要在业务代码里裸写 `XADD` / `XREADGROUP`
- 不要相信 Redis 返回值一定符合你期望的 shape
- 边界校验失败时应该尽早抛错,而不是继续传播脏数据
## Chat WS 当前约束
`src/routes/chat-ws.ts` 当前采用:
@@ -0,0 +1,58 @@
# Admin Flux Grants — End-to-End Verification
## 用户路径 1admin 同步发 grant → 余额到账
- **场景**admin 通过 `POST /api/admin/flux-grants` 给一个邮箱发 100 FLUX,HTTP 同步返回 `granted` 数组 → 用户余额上升。
- **命令**(占位,需要重新实测):
```bash
TOKEN=... # admin 用户 access token
curl -s -X POST "http://localhost:3000/api/admin/flux-grants" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"description":"local verify","amount":100,"emails":["rbxin2003@gmail.com"]}'
# 然后查余额
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:3000/api/v1/flux
# 查 ledger
curl -s -H "Authorization: Bearer $TOKEN" 'http://localhost:3000/api/v1/flux/history?limit=3'
```
- **预期**
- HTTP 200body 包含 `result.granted: [{ email, userId, fluxTransactionId, balanceAfter }]``result.failed: []``result.skipped: []`
- `/api/v1/flux` 返回的 `flux` 比之前增加 100
- `/api/v1/flux/history` 顶部一条 `type='promo'`、`description='local verify'`、`metadata.issuedByUserId` = admin 的 userId
- **实际输出**:⏳ 待重新实测(架构刚从 batch 改成同步,旧 verification 已无效)。
- **环境**:本地 `pnpm -F @proj-airi/server dev`commit SHA 待补,`ADMIN_EMAILS` 含 admin 邮箱且 `email_verified=true`。
- **最后验证**:⏳ 待补
## 用户路径 2:dry-run 预览邮箱列表
- **场景**admin 在真发之前用 `?dryRun=true` 看 4 个 emailvalid + 大小写变体重复 + 找不到)的解析结果。
- **命令**
```bash
curl -s -X POST 'http://localhost:3000/api/admin/flux-grants?dryRun=true' \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"description":"smoke","amount":100,"emails":["rbxin2003@gmail.com","RBXIN2003@gmail.com","ghost@nope.example","rbxin2003@gmail.com"]}'
```
- **预期**HTTP 200`{ preview: { willGrant: 1, willSkip: { notFound: 1, userDeleted: 0, duplicateInInput: 2 }, totalFluxToIssue: 100, ... } }`。`flux_transaction` 表无新增行。
- **实际输出**:⏳ 待重新实测
- **环境**:同上
- **最后验证**:⏳ 待补
## 用户路径 3:未登录 / 非 admin / 未验证邮箱被挡住
- **场景**`adminGuard` 三种拒绝路径(401 无 session / 403 不在 allowlist / 403 邮箱未验证)。
- **命令**
```bash
curl -s -w "%{http_code}\n" -X POST http://localhost:3000/api/admin/flux-grants -d '{}'
# 期望 401
```
- **实际**:单元测试 [`admin-guard.test.ts`](apps/server/src/middlewares/tests/admin-guard.test.ts) 覆盖完整三条路径 + case-insensitive 匹配。Live 401/403 端到端验证⏳ 待补。
- **最后验证**:⏳ 待补(unit only
## 已知缺口 / 未验证
- **整套 verification 都需要重跑**:架构从 `flux_grant_batch` 异步处理改成同步 `POST /api/admin/flux-grants` 后,旧的实测输出(含 `mq-stream` 日志、`batch.status: completed` 等)全部失效。Iron Law 要求至少跑一次路径 1 + 2 替换 ⏳ 占位。
- **失败重试 + idempotencyKey**:单测覆盖了"同 key 不同 recipient → 不同 requestId"这一逻辑,但没真跑过"故意打挂 DB 触发部分 failed → 用 `idempotencyKey` 重发只补失败的"端到端。
- **`emails` 上限 200 实际响应时间**:估算 4–10s,没在生产 DB 上跑过。如果 `creditFlux` 单条 > 50ms(远端 Postgres + Redis),需要回头下调上限或加批处理优化。
@@ -2,14 +2,13 @@
## 进程角色
统一入口在 `src/bin/run.ts`
入口:`src/bin/run.ts`
- `api`
- 启动 Hono HTTP + WebSocket 服务
- `billing-consumer`
- 消费 Redis Stream `billing-events`,异步将 ledger、audit log、LLM 请求日志写入 DB
- 没有任何"常驻后台 loop",也没有"POST 触发的 fire-and-forget"。所有写路径都在请求线程里同步完成
这两个角色是当前服务端部署拆分的基本单位
不再有独立的 `worker` / `billing-consumer` 进程。原先的 Redis Stream billing event 链路、advisory-lock poller、admin flux grant batch 异步处理全部移除
## API 角色
@@ -30,84 +29,13 @@
- 启动 HTTP server
- 注入 WebSocket
## Billing Consumer
## Admin flux grant:同步执行
实现位置:
详见 [`admin-flux-grants.md`](admin-flux-grants.md)。简要:admin 调 `POST /api/admin/flux-grants`,路由 handler 在请求线程内顺序对每个 recipient 调 `BillingService.creditFlux`,HTTP 响应里直接返回每条的 outcomegranted / skipped / failed)。失败由 admin 看响应自行重发,可选 `idempotencyKey` 让重发安全。
- 入口:`src/bin/run-billing-consumer.ts`
- worker`src/services/billing-mq-worker.ts`
- stream adapter`src/services/billing-mq.ts`
## 失败 / 崩溃恢复
工作流程:
1. 以 consumer group 模式消费 Redis Stream `billing-events`
2. 根据事件类型分发处理:
- `flux.debited` — 写 `flux_transaction``flux_transaction`
- `llm.request.log` — 写 `llm_request_log`
3. 处理成功后 ACK;handler 抛错时不 ACK,消息保持 pending 等待重试
相关环境变量:
- `BILLING_EVENTS_STREAM`
- `BILLING_EVENTS_CONSUMER_NAME`
- `BILLING_EVENTS_BATCH_SIZE`
- `BILLING_EVENTS_BLOCK_MS`
- `BILLING_EVENTS_MIN_IDLE_MS`
## Redis Streams 语义
`billing-mq.ts` 把 Redis Streams 抽象成:
- `publish()`
- `ensureConsumerGroup()`
- `consume()`
- `claimIdleMessages()`
- `ack()`
这层约束了消息处理语义:
- 使用 consumer group
- 使用 pending reclaim
- handler 抛错时不 ack,消息保持 pending
因此新增新的 stream consumer 时,最安全的方式通常是复用这层,不要自己裸写 `XREADGROUP`
## 聊天 WebSocket 运行时
`src/routes/chat-ws.ts` 还有一套独立于 Redis Streams 的运行时机制:
- 同实例连接保存在进程内 `Map`
- 跨实例 fan-out 通过 Redis Pub/Sub
这意味着:
- WS 广播不具备持久化和重放能力
- 真正补齐消息还是靠 `pullMessages`
- 广播只是为了降低拉取延迟,不代表存在旧式 `sync` 端点
如果要改 Redis key / channel 构造、Pub/Sub payload 或 Streams 边界,先看 `redis-boundaries-and-pubsub.md`
## OpenTelemetry
初始化在 `src/libs/otel.ts`
启用条件:
- `OTEL_EXPORTER_OTLP_ENDPOINT` 存在
覆盖面:
- HTTP
- Auth
- Chat engagement
- Revenue
- LLM
- DB / Redis instrumentation
重要实现细节:
- `sdk.start()` 必须发生在 `metrics.getMeter()` 之前
- `/health` 会被 HTTP instrumentation 忽略
服务端没有需要恢复的"中间状态"。每次 `creditFlux` 自己是一个 DB 事务;要么写进 `flux_transaction` ledger 要么没写,没有第三态。`(user_id, request_id)` partial unique index 保证带 `idempotencyKey` 的重发不会双发。
## 环境变量分层
@@ -131,14 +59,6 @@
- `STRIPE_SECRET_KEY`
- `STRIPE_WEBHOOK_SECRET`
### Billing MQ
- `BILLING_EVENTS_STREAM`
- `BILLING_EVENTS_CONSUMER_NAME`
- `BILLING_EVENTS_BATCH_SIZE`
- `BILLING_EVENTS_BLOCK_MS`
- `BILLING_EVENTS_MIN_IDLE_MS`
### OTel
- `OTEL_SERVICE_NAMESPACE`
@@ -148,15 +68,49 @@
- `OTEL_EXPORTER_OTLP_HEADERS`
- `OTEL_DEBUG`
> NOTICE: `BILLING_EVENTS_*` 已全部移除。
## 聊天 WebSocket 运行时
`src/routes/chat-ws.ts` 是另一种独立运行时:
- 同实例连接保存在进程内 `Map`
- 跨实例 fan-out 通过 Redis Pub/Sub
这意味着:
- WS 广播不具备持久化和重放能力
- 真正补齐消息还是靠 `pullMessages`
- 广播只是为了降低拉取延迟,不代表存在旧式 `sync` 端点
如果要改 Redis key / channel 构造、Pub/Sub payload,先看 `redis-boundaries-and-pubsub.md`
## OpenTelemetry
初始化在 `src/libs/otel.ts`
启用条件:
- `OTEL_EXPORTER_OTLP_ENDPOINT` 存在
覆盖面:
- HTTP
- Auth
- Chat engagement
- Revenue
- LLM
- DB / Redis instrumentation
重要实现细节:
- `sdk.start()` 必须发生在 `metrics.getMeter()` 之前
- `/health` 会被 HTTP instrumentation 忽略
## 运行时修改建议
如果你要改:
- 新增 worker
- 先看 `run.ts` 的角色模型和 `billing-mq-worker.ts`
- 改事件分发
- 先看 billing-consumer handler,在 `billing-mq-worker.ts` 中增加新的事件处理分支
- 改聊天同步
- 先区分“持久化消息”与“广播通知”两层
- 改部署限流
- 注意当前 `rate-limit.ts` 仍是单实例内存模型
- **新增异步工作**:先问三遍"为什么不能在请求线程里同步做完"。绝大多数 admin / webhook / 短 batch 都可以;实在不行也优先 fire-and-forget per-request,而不是引入常驻 loop。
- **真的需要 idle-driven 的活**(清理过期 token、定时聚合等):先评估是否值得。如果是,开 Postgres `pg_cron` 或外部 cron service 调专门的 internal API endpoint,比"进程内常驻 loop"更可观察、更易停。
- **改 Stripe / Flux 写路径**:看 `billing-architecture.md`,所有 ledger 写入都在事务内同步完成
- **改聊天同步**:先区分"持久化消息"与"广播通知"两层
- **改部署限流**:注意当前 `rate-limit.ts` 仍是单实例内存模型
+1 -1
View File
@@ -1,5 +1,5 @@
{
"id": "ca347910-97d0-458e-86d2-5345dc6a3a96",
"id": "859b8c16-2a9a-42d4-ac35-d7d802dac58a",
"prevId": "86b6aa8d-03e2-40f3-8331-582d0bd3c399",
"version": "7",
"dialect": "postgresql",
+3 -3
View File
@@ -75,9 +75,9 @@
{
"idx": 10,
"version": "7",
"when": 1778171601523,
"tag": "0010_aspiring_power_man",
"when": 1778223289077,
"tag": "0010_sudden_bastion",
"breakpoints": true
}
]
}
}
+2 -2
View File
@@ -46,9 +46,9 @@ function createTestDeps() {
fluxTransactionService: {} as any,
stripeService: {} as any,
billingService: {} as any,
fluxGrantBatchService: {} as any,
adminFluxGrantsService: {} as any,
ttsMeter: {} as any,
billingMq: {} as any,
requestLogService: {} as any,
configKV: {
getOrThrow: vi.fn(async (key: string) => {
switch (key) {
+21 -28
View File
@@ -3,10 +3,8 @@ import type Redis from 'ioredis'
import type { AuthInstance } from './libs/auth'
import type { Database } from './libs/db'
import type { Env } from './libs/env'
import type { MqService } from './libs/mq'
import type { OtelInstance } from './libs/otel'
import type { FluxGrantBatchService } from './services/admin-flux-grant-batch/flux-grant-batch-service'
import type { BillingEvent } from './services/billing/billing-events'
import type { AdminFluxGrantsService } from './services/admin-flux-grants'
import type { BillingService } from './services/billing/billing-service'
import type { FluxMeter } from './services/billing/flux-meter'
import type { CharacterService } from './services/characters'
@@ -15,6 +13,7 @@ import type { ConfigKVService } from './services/config-kv'
import type { FluxService } from './services/flux'
import type { FluxTransactionService } from './services/flux-transaction'
import type { ProviderService } from './services/providers'
import type { RequestLogService } from './services/request-log'
import type { StripeService } from './services/stripe'
import type { UserDeletionService } from './services/user-deletion'
import type { HonoEnv } from './types/hono'
@@ -41,7 +40,7 @@ import { createRedis } from './libs/redis'
import { resolveRequestAuth } from './libs/request-auth'
import { sessionMiddleware } from './middlewares/auth'
import { otelMiddleware } from './middlewares/otel'
import { createAdminFluxGrantBatchRoutes } from './routes/admin/flux-grant-batches'
import { createAdminFluxGrantsRoutes } from './routes/admin/flux-grants'
import { createAuthRoutes } from './routes/auth'
import { createCharacterRoutes } from './routes/characters'
import { createChatWsHandlers } from './routes/chat-ws'
@@ -50,8 +49,7 @@ import { createFluxRoutes } from './routes/flux'
import { createV1CompletionsRoutes } from './routes/openai/v1'
import { createProviderRoutes } from './routes/providers'
import { createStripeRoutes } from './routes/stripe'
import { createFluxGrantBatchService } from './services/admin-flux-grant-batch/flux-grant-batch-service'
import { createBillingMq } from './services/billing/billing-events'
import { createAdminFluxGrantsService } from './services/admin-flux-grants'
import { createBillingService } from './services/billing/billing-service'
import { createFluxMeter } from './services/billing/flux-meter'
import { createCharacterService } from './services/characters'
@@ -78,9 +76,9 @@ interface AppDeps {
fluxTransactionService: FluxTransactionService
stripeService: StripeService
billingService: BillingService
fluxGrantBatchService: FluxGrantBatchService
adminFluxGrantsService: AdminFluxGrantsService
ttsMeter: FluxMeter
billingMq: MqService<BillingEvent>
requestLogService: RequestLogService
configKV: ConfigKVService
redis: Redis
env: Env
@@ -213,7 +211,7 @@ export async function buildApp(deps: AppDeps) {
/**
* V1 routes for official provider.
*/
.route('/api/v1/openai', createV1CompletionsRoutes(deps.fluxService, deps.billingService, deps.configKV, deps.billingMq, deps.ttsMeter, deps.redis, deps.env, deps.otel?.genAi))
.route('/api/v1/openai', createV1CompletionsRoutes(deps.fluxService, deps.billingService, deps.configKV, deps.requestLogService, deps.ttsMeter, deps.redis, deps.env, deps.otel?.genAi))
/**
* Flux routes.
@@ -226,10 +224,10 @@ export async function buildApp(deps: AppDeps) {
.route('/api/v1/stripe', createStripeRoutes(deps.fluxService, deps.stripeService, deps.billingService, deps.configKV, deps.env, deps.redis, deps.otel?.revenue))
/**
* Admin routes — guarded by ADMIN_USER_IDS allowlist. v1 only includes
* batch-based promo flux grant operations.
* Admin routes — guarded by `ADMIN_EMAILS` allowlist + verified email.
* v1 only includes synchronous one-shot promo flux grants.
*/
.route('/api/admin/flux-grant-batches', createAdminFluxGrantBatchRoutes(deps.fluxGrantBatchService, deps.env))
.route('/api/admin/flux-grants', createAdminFluxGrantsRoutes(deps.adminFluxGrantsService, deps.env))
/**
* Catch-all 404 in JSON. Replaces hono's default `text/html` "404 Not
@@ -330,13 +328,6 @@ export async function createApp() {
build: ({ dependsOn }) => createConfigKVService(dependsOn.redis),
})
const billingMq = injeca.provide('services:billingMq', {
dependsOn: { redis, env: parsedEnv },
build: ({ dependsOn }) => createBillingMq(dependsOn.redis, {
stream: dependsOn.env.BILLING_EVENTS_STREAM,
}),
})
const emailService = injeca.provide('services:email', {
dependsOn: { env: parsedEnv },
build: ({ dependsOn }) => createEmailService({
@@ -429,13 +420,16 @@ export async function createApp() {
})
const billingService = injeca.provide('services:billing', {
dependsOn: { db, redis, billingMq, configKV, otel },
build: ({ dependsOn }) => createBillingService(dependsOn.db, dependsOn.redis, dependsOn.billingMq, dependsOn.configKV, dependsOn.otel?.revenue),
dependsOn: { db, redis, configKV, otel },
build: ({ dependsOn }) => createBillingService(dependsOn.db, dependsOn.redis, dependsOn.configKV, dependsOn.otel?.revenue),
})
const fluxGrantBatchService = injeca.provide('services:adminFluxGrantBatch', {
dependsOn: { db },
build: ({ dependsOn }) => createFluxGrantBatchService(dependsOn.db),
const adminFluxGrantsService = injeca.provide('services:adminFluxGrants', {
dependsOn: { db, billingService },
build: ({ dependsOn }) => createAdminFluxGrantsService({
db: dependsOn.db,
billingService: dependsOn.billingService,
}),
})
const ttsMeter = injeca.provide('services:ttsMeter', {
@@ -468,9 +462,8 @@ export async function createApp() {
requestLogService,
stripeService,
billingService,
fluxGrantBatchService,
adminFluxGrantsService,
ttsMeter,
billingMq,
configKV,
redis,
env: parsedEnv,
@@ -487,9 +480,9 @@ export async function createApp() {
fluxTransactionService: resolved.fluxTransactionService,
stripeService: resolved.stripeService,
billingService: resolved.billingService,
fluxGrantBatchService: resolved.fluxGrantBatchService,
adminFluxGrantsService: resolved.adminFluxGrantsService,
ttsMeter: resolved.ttsMeter,
billingMq: resolved.billingMq,
requestLogService: resolved.requestLogService,
configKV: resolved.configKV,
redis: resolved.redis,
env: resolved.env,
-106
View File
@@ -1,106 +0,0 @@
import process, { pid } from 'node:process'
import { initLogger, LoggerFormat, LoggerLevel, useLogger } from '@guiiai/logg'
import { createDrizzle, migrateDatabase } from '../libs/db'
import { parseEnv } from '../libs/env'
import { initializeExternalDependency } from '../libs/external-dependency'
import { createMqWorker } from '../libs/mq'
import { createRedis } from '../libs/redis'
import { runFluxGrantBatchWorker } from '../services/admin-flux-grant-batch/flux-grant-batch-worker'
import { createBillingConsumerHandler } from '../services/billing/billing-consumer-handler'
import { createBillingMq } from '../services/billing/billing-events'
import { createBillingService } from '../services/billing/billing-service'
import { createConfigKVService } from '../services/config-kv'
export async function runBillingConsumer(): Promise<void> {
initLogger(LoggerLevel.Debug, LoggerFormat.Pretty)
const env = parseEnv(process.env)
const logger = useLogger('billing-consumer').useGlobalConfig()
const { db, pool } = await initializeExternalDependency(
'Database',
logger,
async (attempt) => {
const connection = createDrizzle(env)
try {
await connection.db.execute('SELECT 1')
logger.log(`Connected to database on attempt ${attempt}`)
await migrateDatabase(connection.db)
logger.log(`Applied schema on attempt ${attempt}`)
return connection
}
catch (error) {
await connection.pool.end()
throw error
}
},
)
const redis = await initializeExternalDependency(
'Redis',
logger,
async (attempt) => {
const instance = createRedis(env.REDIS_URL)
try {
await instance.connect()
logger.log(`Connected to Redis on attempt ${attempt}`)
return instance
}
catch (error) {
instance.disconnect()
throw error
}
},
)
const abortController = new AbortController()
const consumer = env.BILLING_EVENTS_CONSUMER_NAME ?? `billing-consumer-${pid}`
const shutdown = (signalName: string) => {
if (abortController.signal.aborted) {
return
}
logger.withFields({ signalName }).log('Stopping billing consumer')
abortController.abort()
}
process.once('SIGINT', () => shutdown('SIGINT'))
process.once('SIGTERM', () => shutdown('SIGTERM'))
try {
const mq = createBillingMq(redis, {
stream: env.BILLING_EVENTS_STREAM,
})
const handler = createBillingConsumerHandler(db)
const mqWorker = createMqWorker(mq)
// Build a BillingService for the flux grant batch worker. This
// consumer-side instance writes to the same DB / Redis / event stream as
// the API process — multi-instance Railway is the design assumption.
const configKV = createConfigKVService(redis)
const billingService = createBillingService(db, redis, mq, configKV, null)
// Run the Redis Stream consumer and the flux grant batch polling loop in
// parallel. Either rejection aborts both via the shared signal.
await Promise.all([
mqWorker.run({
group: 'billing-consumer',
consumer,
signal: abortController.signal,
batchSize: env.BILLING_EVENTS_BATCH_SIZE,
blockMs: env.BILLING_EVENTS_BLOCK_MS,
minIdleTimeMs: env.BILLING_EVENTS_MIN_IDLE_MS,
onMessage: message => handler.handleMessage(message),
}),
runFluxGrantBatchWorker({ db, billingService }, abortController.signal),
])
}
finally {
await redis.quit()
await pool.end()
}
}
+1 -32
View File
@@ -8,21 +8,6 @@ import { cac } from 'cac'
import { runApiServer } from '../app'
import { errorMessageFromUnknown } from '../utils/error-message'
import { runBillingConsumer } from './run-billing-consumer'
const serverRoles = ['api', 'billing-consumer'] as const
type ServerRole = typeof serverRoles[number]
async function runServerRole(role: ServerRole): Promise<void> {
switch (role) {
case 'api':
await runApiServer()
return
case 'billing-consumer':
await runBillingConsumer()
}
}
export function createServerCli() {
const cli = cac('server')
@@ -30,29 +15,13 @@ export function createServerCli() {
cli
.usage('<role>')
.command('api', 'Start the HTTP/WebSocket API process')
.action(() => runServerRole('api'))
cli
.command('billing-consumer', 'Start the billing events consumer (transactions, audit, request logs)')
.action(() => runServerRole('billing-consumer'))
.action(() => runApiServer())
cli.help()
return cli
}
export function parseServerRole(args: string[]): ServerRole | null {
const cli = createServerCli()
cli.parse(['node', 'server', ...args], { run: false })
const role = cli.matchedCommandName
if (!role) {
return null
}
return serverRoles.includes(role as ServerRole) ? role as ServerRole : null
}
async function main(): Promise<void> {
const cli = createServerCli()
cli.parse(process.argv, { run: false })
-8
View File
@@ -6,8 +6,6 @@ import { useLogger } from '@guiiai/logg'
import { injeca } from 'injeca'
import { integer, maxValue, minValue, nonEmpty, object, optional, parse, pipe, string, transform } from 'valibot'
import { DEFAULT_BILLING_EVENTS_STREAM } from '../utils/redis-keys'
function optionalIntegerFromString(defaultValue: number, envKey: string, minimum: number) {
return optional(
pipe(
@@ -70,12 +68,6 @@ const EnvSchema = object({
DEFAULT_CHAT_MODEL: pipe(string(), nonEmpty('DEFAULT_CHAT_MODEL is required')),
DEFAULT_TTS_MODEL: pipe(string(), nonEmpty('DEFAULT_TTS_MODEL is required')),
BILLING_EVENTS_STREAM: optional(string(), DEFAULT_BILLING_EVENTS_STREAM),
BILLING_EVENTS_CONSUMER_NAME: optional(string()),
BILLING_EVENTS_BATCH_SIZE: optionalIntegerFromString(10, 'BILLING_EVENTS_BATCH_SIZE', 1),
BILLING_EVENTS_BLOCK_MS: optionalIntegerFromString(5000, 'BILLING_EVENTS_BLOCK_MS', 1),
BILLING_EVENTS_MIN_IDLE_MS: optionalIntegerFromString(30000, 'BILLING_EVENTS_MIN_IDLE_MS', 1),
// Database pool
DB_POOL_MAX: optionalIntegerFromString(20, 'DB_POOL_MAX', 1),
DB_POOL_IDLE_TIMEOUT_MS: optionalIntegerFromString(30000, 'DB_POOL_IDLE_TIMEOUT_MS', 1),
-13
View File
@@ -1,13 +0,0 @@
export { createMqService } from './stream'
export type { MqService } from './stream'
export type {
ClaimIdleOptions,
ConsumeOptions,
MqOptions,
RedisCommandClient,
StreamMessage,
WorkerOptions,
} from './types'
export { createMqWorker } from './worker'
export type { MqWorker } from './worker'
-192
View File
@@ -1,192 +0,0 @@
import type {
ClaimIdleOptions,
ConsumeOptions,
MqOptions,
RedisArgument,
RedisCommandClient,
StreamMessage,
} from './types'
import { useLogger } from '@guiiai/logg'
type RedisStreamEntry = [streamMessageId: string, fieldValues: string[]]
type RedisReadGroupResponse = [stream: string, entries: RedisStreamEntry[]][]
type RedisAutoClaimResponse = [nextStartId: string, entries: RedisStreamEntry[], deletedIds?: string[]]
const logger = useLogger('mq-stream').useGlobalConfig()
/**
* Create a typed Redis Stream service.
*
* The caller supplies serialize/deserialize functions so this module
* stays domain-agnostic it only knows how to talk to Redis Streams.
*/
export function createMqService<TEvent>(redis: RedisCommandClient, options: MqOptions<TEvent>) {
const { stream, serialize, deserialize } = options
function parseEntry(entry: unknown): StreamMessage<TEvent> {
if (!Array.isArray(entry) || entry.length !== 2) {
throw new Error('Redis Stream entry has an invalid shape')
}
const [streamMessageId, rawFieldValues] = entry
if (typeof streamMessageId !== 'string') {
throw new TypeError('Redis Stream entry is missing a valid message id')
}
if (!Array.isArray(rawFieldValues)) {
throw new TypeError('Redis Stream entry fields are invalid')
}
return { streamMessageId, event: deserialize(toFieldRecord(rawFieldValues)) }
}
function parseReadGroupResponse(response: unknown): StreamMessage<TEvent>[] {
if (response == null) {
return []
}
if (!Array.isArray(response)) {
throw new TypeError('Redis XREADGROUP returned an invalid response')
}
return response.flatMap((streamResponse) => {
if (!Array.isArray(streamResponse) || streamResponse.length !== 2) {
throw new Error('Redis XREADGROUP returned an invalid stream payload')
}
const [, entries] = streamResponse as RedisReadGroupResponse[number]
return entries.map(parseEntry)
})
}
function parseAutoClaimResponse(response: unknown): StreamMessage<TEvent>[] {
if (response == null) {
return []
}
if (!Array.isArray(response) || response.length < 2) {
throw new Error('Redis XAUTOCLAIM returned an invalid response')
}
const [, entries] = response as RedisAutoClaimResponse
if (!Array.isArray(entries)) {
throw new TypeError('Redis XAUTOCLAIM returned invalid entries')
}
return entries.map(parseEntry)
}
return {
stream,
async publish(event: TEvent): Promise<string> {
const fields = serialize(event)
const xaddArgs: RedisArgument[] = [stream]
if (options.maxLength != null) {
xaddArgs.push('MAXLEN', '~', options.maxLength)
}
xaddArgs.push('*', ...toRedisFieldArguments(fields))
const streamMessageId = await redis.call('XADD', ...xaddArgs)
if (typeof streamMessageId !== 'string') {
throw new TypeError('Redis XADD did not return a stream message id')
}
logger.withFields({ stream, streamMessageId }).log('Published event to Redis Stream')
return streamMessageId
},
async ensureConsumerGroup(group: string, startId = '0'): Promise<boolean> {
try {
await redis.call('XGROUP', 'CREATE', stream, group, startId, 'MKSTREAM')
return true
}
catch (error) {
if (error instanceof Error && error.message.includes('BUSYGROUP')) {
return false
}
throw error
}
},
async consume(consumeOptions: ConsumeOptions): Promise<StreamMessage<TEvent>[]> {
const response = await redis.call(
'XREADGROUP',
'GROUP',
consumeOptions.group,
consumeOptions.consumer,
'COUNT',
consumeOptions.count ?? 10,
'BLOCK',
consumeOptions.blockMs ?? 5_000,
'STREAMS',
stream,
consumeOptions.startId ?? '>',
)
return parseReadGroupResponse(response)
},
async claimIdleMessages(claimOptions: ClaimIdleOptions): Promise<StreamMessage<TEvent>[]> {
const response = await redis.call(
'XAUTOCLAIM',
stream,
claimOptions.group,
claimOptions.consumer,
claimOptions.minIdleTimeMs,
claimOptions.startId ?? '0-0',
'COUNT',
claimOptions.count ?? 10,
)
return parseAutoClaimResponse(response)
},
async ack(group: string, streamMessageIds: string | string[]): Promise<number> {
const ids = Array.isArray(streamMessageIds) ? streamMessageIds : [streamMessageIds]
if (ids.length === 0) {
return 0
}
const acked = await redis.call('XACK', stream, group, ...ids)
if (typeof acked !== 'number') {
throw new TypeError('Redis XACK did not return an acknowledgement count')
}
return acked
},
}
}
function toRedisFieldArguments(fields: Record<string, string | undefined>): RedisArgument[] {
return Object.entries(fields)
.filter(([, value]) => value !== undefined)
.flatMap(([key, value]) => [key, value as string])
}
function toFieldRecord(fieldValues: string[]): Record<string, string> {
if (fieldValues.length % 2 !== 0) {
throw new Error('Redis Stream entry fields must be key/value pairs')
}
const fields: Record<string, string> = {}
for (let index = 0; index < fieldValues.length; index += 2) {
const key = fieldValues[index]
const value = fieldValues[index + 1]
if (typeof key !== 'string' || typeof value !== 'string') {
throw new TypeError('Redis Stream entry contains non-string field data')
}
fields[key] = value
}
return fields
}
export type MqService<TEvent> = ReturnType<typeof createMqService<TEvent>>
@@ -1,119 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import { createMqWorker } from '../worker'
function createMessage() {
return {
streamMessageId: '1740000000000-0',
event: {
eventId: 'evt-1',
eventType: 'flux.debited' as const,
aggregateId: 'user-1',
userId: 'user-1',
requestId: 'req-1',
occurredAt: '2026-03-24T00:00:00.000Z',
schemaVersion: 1,
payload: {
amount: 5,
balanceAfter: 95,
source: 'llm',
},
},
}
}
describe('mqWorker', () => {
it('reclaims pending messages before reading new ones and acks after handling', async () => {
const controller = new AbortController()
const message = createMessage()
const mq = {
ensureConsumerGroup: vi.fn(async () => true),
claimIdleMessages: vi.fn(async () => [message]),
consume: vi.fn(async () => []),
ack: vi.fn(async () => 1),
}
const worker = createMqWorker(mq as any)
const handled: string[] = []
await worker.run({
group: 'billing',
consumer: 'billing-1',
signal: controller.signal,
onMessage: vi.fn(async (incomingMessage) => {
handled.push(incomingMessage.event.eventId)
controller.abort()
}),
})
expect(mq.ensureConsumerGroup).toHaveBeenCalledWith('billing')
expect(mq.claimIdleMessages).toHaveBeenCalledWith({
group: 'billing',
consumer: 'billing-1',
minIdleTimeMs: 30000,
count: 10,
})
expect(mq.consume).not.toHaveBeenCalled()
expect(mq.ack).toHaveBeenCalledWith('billing', '1740000000000-0')
expect(handled).toEqual(['evt-1'])
})
it('reads new messages when there are no idle pending messages', async () => {
const controller = new AbortController()
const message = createMessage()
const mq = {
ensureConsumerGroup: vi.fn(async () => true),
claimIdleMessages: vi.fn(async () => []),
consume: vi.fn(async () => [message]),
ack: vi.fn(async () => 1),
}
const worker = createMqWorker(mq as any)
await worker.run({
group: 'billing',
consumer: 'billing-1',
signal: controller.signal,
batchSize: 5,
blockMs: 250,
minIdleTimeMs: 1000,
onMessage: vi.fn(async () => {
controller.abort()
}),
})
expect(mq.consume).toHaveBeenCalledWith({
group: 'billing',
consumer: 'billing-1',
count: 5,
blockMs: 250,
})
})
it('leaves failed messages pending by not acking them', async () => {
const controller = new AbortController()
const mq = {
ensureConsumerGroup: vi.fn(async () => true),
claimIdleMessages: vi.fn(async () => [createMessage()]),
consume: vi.fn(async () => []),
ack: vi.fn(async () => 1),
}
const worker = createMqWorker(mq as any)
await worker.run({
group: 'billing',
consumer: 'billing-1',
signal: controller.signal,
onMessage: vi.fn(async () => {
controller.abort()
throw new Error('handler failed')
}),
})
expect(mq.ack).not.toHaveBeenCalled()
})
})
-47
View File
@@ -1,47 +0,0 @@
export type RedisArgument = string | number
export interface RedisCommandClient {
call: (command: string, ...args: RedisArgument[]) => Promise<unknown>
}
export interface MqOptions<TEvent> {
/** Redis Stream key name. */
stream: string
/** Approximate max stream length (MAXLEN ~). Unbounded if omitted. */
maxLength?: number
/** Convert a typed event into flat Redis field/value pairs. */
serialize: (event: TEvent) => Record<string, string | undefined>
/** Reconstruct a typed event from flat Redis field/value pairs. */
deserialize: (fields: Record<string, string>) => TEvent
}
export interface StreamMessage<TEvent> {
streamMessageId: string
event: TEvent
}
export interface ConsumeOptions {
group: string
consumer: string
count?: number
blockMs?: number
startId?: string
}
export interface ClaimIdleOptions {
group: string
consumer: string
minIdleTimeMs: number
startId?: string
count?: number
}
export interface WorkerOptions<TEvent> {
group: string
consumer: string
signal: AbortSignal
batchSize?: number
blockMs?: number
minIdleTimeMs?: number
onMessage: (message: StreamMessage<TEvent>) => Promise<void>
}
-59
View File
@@ -1,59 +0,0 @@
import type { MqService } from './stream'
import type { StreamMessage, WorkerOptions } from './types'
import { useLogger } from '@guiiai/logg'
const logger = useLogger('mq-worker').useGlobalConfig()
/**
* Create a consumer worker that processes messages from a Redis Stream.
*
* The loop first reclaims idle (possibly stalled) messages, then falls
* back to consuming new ones. Each message is passed to `onMessage`;
* on success it is acknowledged, on failure it stays pending for retry.
*/
export function createMqWorker<TEvent>(mq: MqService<TEvent>) {
return {
async run(options: WorkerOptions<TEvent>): Promise<void> {
await mq.ensureConsumerGroup(options.group)
while (!options.signal.aborted) {
const reclaimedMessages = await mq.claimIdleMessages({
group: options.group,
consumer: options.consumer,
minIdleTimeMs: options.minIdleTimeMs ?? 30_000,
count: options.batchSize ?? 10,
})
const messages: StreamMessage<TEvent>[] = reclaimedMessages.length > 0
? reclaimedMessages
: await mq.consume({
group: options.group,
consumer: options.consumer,
count: options.batchSize ?? 10,
blockMs: options.blockMs ?? 5_000,
})
if (messages.length === 0) {
continue
}
for (const message of messages) {
try {
await options.onMessage(message)
await mq.ack(options.group, message.streamMessageId)
}
catch (error) {
logger.withError(error).withFields({
group: options.group,
consumer: options.consumer,
streamMessageId: message.streamMessageId,
}).error('MQ handler failed; leaving message pending')
}
}
}
},
}
}
export type MqWorker<TEvent> = ReturnType<typeof createMqWorker<TEvent>>
@@ -0,0 +1,103 @@
import type { Env } from '../../../libs/env'
import type { AdminFluxGrantsService } from '../../../services/admin-flux-grants'
import type { HonoEnv } from '../../../types/hono'
import { Hono } from 'hono'
import { array, email, integer, maxLength, maxValue, minLength, minValue, nonEmpty, number, object, optional, pipe, safeParse, string } from 'valibot'
import { adminGuard } from '../../../middlewares/admin-guard'
import { authGuard } from '../../../middlewares/auth'
import { createBadRequestError } from '../../../utils/error'
/**
* Per-grant cap on amount per user. Caps a single typo from issuing
* absurd amounts. Operator can override later via configKV if ever needed.
*/
const MAX_GRANT_AMOUNT_PER_USER = 10_000
/**
* Hard cap on emails per single grant request.
*
* NOTICE:
* Processing is synchronous inside the HTTP request: at the default
* `creditFlux` cost (single-row update + ledger insert + Redis cache write
* 2050ms per recipient) 200 recipients fit comfortably under typical
* 30s load-balancer timeouts. Higher counts mean splitting into multiple
* calls admin tooling, not bulk import.
*/
const MAX_EMAILS_PER_GRANT = 200
const GrantBodySchema = object({
description: pipe(string(), nonEmpty('description is required'), maxLength(500)),
amount: pipe(
number(),
integer('amount must be an integer'),
minValue(1, 'amount must be at least 1'),
maxValue(MAX_GRANT_AMOUNT_PER_USER, `amount must be at most ${MAX_GRANT_AMOUNT_PER_USER}`),
),
emails: pipe(
array(pipe(string(), email('emails must be valid email addresses'))),
minLength(1, 'emails must not be empty'),
maxLength(MAX_EMAILS_PER_GRANT, `emails must be at most ${MAX_EMAILS_PER_GRANT} entries`),
),
/**
* Optional. When set, recipient `creditFlux` calls become idempotent
* across retries re-firing the same `(idempotencyKey, recipient)`
* combination is a no-op via the `(user_id, request_id)` partial unique
* index. Use when admin wants safe retry semantics; omit for "I want to
* grant again on purpose".
*/
idempotencyKey: optional(pipe(string(), maxLength(100))),
})
/**
* Admin routes for issuing one-shot FLUX grants.
*
* Mounted at `/api/admin/flux-grants`. The whole grant flow lives in a
* single synchronous endpoint no batch table, no state machine, no
* background loop. Audit trail is the per-recipient `flux_transaction`
* ledger row produced by `BillingService.creditFlux`; admin can query it
* via the existing `/api/v1/flux/history` endpoint or directly in DB.
*/
export function createAdminFluxGrantsRoutes(
fluxGrantsService: AdminFluxGrantsService,
env: Env,
) {
return new Hono<HonoEnv>()
.use('*', authGuard)
.use('*', adminGuard(env))
.post('/', async (c) => {
const user = c.get('user')!
const dryRun = c.req.query('dryRun') === 'true'
const raw = await c.req.json().catch(() => null)
if (raw == null)
throw createBadRequestError('Request body must be JSON', 'INVALID_BODY')
const parsed = safeParse(GrantBodySchema, raw)
if (!parsed.success) {
throw createBadRequestError(
'Invalid request body',
'INVALID_BODY',
parsed.issues.map(i => ({ path: i.path?.map(p => p.key).join('.'), message: i.message })),
)
}
const body = parsed.output
if (dryRun) {
const summary = await fluxGrantsService.preview({ amount: body.amount, emails: body.emails })
return c.json({ preview: summary })
}
const { summary, result } = await fluxGrantsService.grant({
amount: body.amount,
description: body.description,
emails: body.emails,
createdByUserId: user.id,
idempotencyKey: body.idempotencyKey,
})
return c.json({ summary, result })
})
}
+10 -23
View File
@@ -2,14 +2,13 @@ import type { Context } from 'hono'
import type Redis from 'ioredis'
import type { Env } from '../../../libs/env'
import type { MqService } from '../../../libs/mq'
import type { GenAiMetrics } from '../../../libs/otel'
import type { UsageInfo } from '../../../services/billing/billing'
import type { BillingEvent } from '../../../services/billing/billing-events'
import type { BillingService } from '../../../services/billing/billing-service'
import type { FluxMeter } from '../../../services/billing/flux-meter'
import type { ConfigKVService } from '../../../services/config-kv'
import type { FluxService } from '../../../services/flux'
import type { RequestLogService } from '../../../services/request-log'
import type { HonoEnv } from '../../../types/hono'
import { useLogger } from '@guiiai/logg'
@@ -82,7 +81,7 @@ function getLlmMetricAttributes(opts: { model: string, type: string, status: num
}
}
export function createV1CompletionsRoutes(fluxService: FluxService, billingService: BillingService, configKV: ConfigKVService, billingMq: MqService<BillingEvent>, ttsMeter: FluxMeter, redis: Redis, env: Env, genAi?: GenAiMetrics | null) {
export function createV1CompletionsRoutes(fluxService: FluxService, billingService: BillingService, configKV: ConfigKVService, requestLogService: RequestLogService, ttsMeter: FluxMeter, redis: Redis, env: Env, genAi?: GenAiMetrics | null) {
const logger = useLogger('v1-completions').useGlobalConfig()
// TODO: Extract this compat route into smaller facades/modules.
// It currently mixes auth, rate limiting, proxying, billing, telemetry, and event publishing in one transport layer entrypoint.
@@ -100,23 +99,11 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi
genAi.tokenUsageOutput.add(opts.completionTokens, attrs)
}
function publishRequestLog(entry: { userId: string, model: string, status: number, durationMs: number, fluxConsumed: number, promptTokens?: number, completionTokens?: number }) {
billingMq.publish({
eventId: nanoid(),
eventType: 'llm.request.log' as const,
aggregateId: entry.userId,
userId: entry.userId,
occurredAt: new Date().toISOString(),
schemaVersion: 1,
payload: {
model: entry.model,
status: entry.status,
durationMs: entry.durationMs,
fluxConsumed: entry.fluxConsumed,
promptTokens: entry.promptTokens,
completionTokens: entry.completionTokens,
},
}).catch(err => logger.withError(err).warn('Failed to publish request log event'))
function recordRequestLog(entry: { userId: string, model: string, status: number, durationMs: number, fluxConsumed: number, promptTokens?: number, completionTokens?: number }) {
// Best-effort: a failed request log must not surface to the user — the
// upstream LLM response has already been delivered (or is mid-stream) by
// the time we get here. Log loss is observability-only.
requestLogService.logRequest(entry).catch(err => logger.withError(err).warn('Failed to write llm_request_log row'))
}
// NOTICE: Billing is best-effort — flux is debited AFTER the LLM response is sent.
@@ -270,7 +257,7 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi
}
catch (err) { logger.withError(err).withFields({ userId: user.id, fluxConsumed, requestId }).error('Failed to debit flux after streaming — unpaid usage') }
publishRequestLog({
recordRequestLog({
userId: user.id,
model: requestModel,
status: response.status,
@@ -315,7 +302,7 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi
completionTokens: usage.completionTokens,
})
publishRequestLog({
recordRequestLog({
userId: user.id,
model: requestModel,
status: response.status,
@@ -397,7 +384,7 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi
span.end()
recordMetrics({ model: requestModel, status: response.status, type: 'tts', durationMs, fluxConsumed })
publishRequestLog({
recordRequestLog({
userId: user.id,
model: requestModel,
status: response.status,
+16 -27
View File
@@ -1,9 +1,8 @@
import type { Env } from '../../../libs/env'
import type { MqService } from '../../../libs/mq'
import type { BillingEvent } from '../../../services/billing/billing-events'
import type { BillingService } from '../../../services/billing/billing-service'
import type { ConfigKVService } from '../../../services/config-kv'
import type { FluxService } from '../../../services/flux'
import type { RequestLogService } from '../../../services/request-log'
import type { HonoEnv } from '../../../types/hono'
import { Buffer } from 'node:buffer'
@@ -13,7 +12,6 @@ import { afterAll, describe, expect, it, vi } from 'vitest'
import { createV1CompletionsRoutes } from '.'
import { ApiError } from '../../../utils/error'
import { DEFAULT_BILLING_EVENTS_STREAM } from '../../../utils/redis-keys'
// --- Mock helpers ---
@@ -65,15 +63,10 @@ function createMockConfigKV(overrides: Record<string, any> = {}): ConfigKVServic
} as any
}
function createMockBillingMq(): MqService<BillingEvent> {
function createMockRequestLogService(): RequestLogService {
return {
stream: DEFAULT_BILLING_EVENTS_STREAM,
publish: vi.fn(async () => '1-0'),
ensureConsumerGroup: vi.fn(async () => true),
consume: vi.fn(async () => []),
claimIdleMessages: vi.fn(async () => []),
ack: vi.fn(async () => 1),
} as any
logRequest: vi.fn(async () => undefined),
}
}
function createMockRedis() {
@@ -111,7 +104,7 @@ function createTestApp(
fluxService: FluxService,
configKV: ConfigKVService,
billingService?: BillingService,
billingMq?: MqService<BillingEvent>,
requestLogService?: RequestLogService,
ttsMeter?: ReturnType<typeof createMockTtsMeter>,
env?: Env,
redis?: ReturnType<typeof createMockRedis>,
@@ -120,7 +113,7 @@ function createTestApp(
fluxService,
billingService ?? createMockBillingService(),
configKV,
billingMq ?? createMockBillingMq(),
requestLogService ?? createMockRequestLogService(),
ttsMeter ?? createMockTtsMeter(),
(redis ?? createMockRedis()) as any,
env ?? createMockEnv(),
@@ -333,14 +326,14 @@ describe('v1CompletionsRoutes', () => {
expect(res.status).toBe(503)
})
it('should publish request log event via billingMq', async () => {
it('writes a synchronous llm_request_log entry after a successful debit', async () => {
globalThis.fetch = vi.fn(async () => new Response('{}', {
status: 200,
headers: { 'Content-Type': 'application/json' },
}))
const billingMq = createMockBillingMq()
const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, billingMq)
const requestLogService = createMockRequestLogService()
const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, requestLogService)
await app.fetch(
new Request('http://localhost/api/v1/openai/chat/completions', {
@@ -351,16 +344,12 @@ describe('v1CompletionsRoutes', () => {
{ user: testUser } as any,
)
expect(billingMq.publish).toHaveBeenCalledWith(
expect(requestLogService.logRequest).toHaveBeenCalledWith(
expect.objectContaining({
eventType: 'llm.request.log',
aggregateId: 'user-1',
userId: 'user-1',
payload: expect.objectContaining({
model: 'gpt-4',
status: 200,
fluxConsumed: 1,
}),
model: 'gpt-4',
status: 200,
fluxConsumed: 1,
}),
)
})
@@ -385,8 +374,8 @@ describe('v1CompletionsRoutes', () => {
}))
const billingService = createMockBillingService(100)
const billingMq = createMockBillingMq()
const app = createTestApp(createMockFluxService(100), createMockConfigKV(), billingService, billingMq)
const requestLogService = createMockRequestLogService()
const app = createTestApp(createMockFluxService(100), createMockConfigKV(), billingService, requestLogService)
const res = await app.fetch(
new Request('http://localhost/api/v1/openai/chat/completions', {
@@ -403,7 +392,7 @@ describe('v1CompletionsRoutes', () => {
await Promise.resolve()
expect(billingService.consumeFluxForLLM).not.toHaveBeenCalled()
expect(billingMq.publish).not.toHaveBeenCalled()
expect(requestLogService.logRequest).not.toHaveBeenCalled()
})
})
@@ -0,0 +1,240 @@
import type { Database } from '../../libs/db'
import type { BillingService } from '../billing/billing-service'
import { useLogger } from '@guiiai/logg'
import { errorMessageFrom } from '@moeru/std'
import { inArray } from 'drizzle-orm'
import * as accountsSchema from '../../schemas/accounts'
import * as fluxSchema from '../../schemas/flux'
const logger = useLogger('admin-flux-grants').useGlobalConfig()
export type SkipReason = 'duplicate_in_input' | 'not_found' | 'user_deleted'
export interface ResolvedRecipient {
inputEmail: string
userId: string | null
status: 'pending' | 'skipped'
errorReason: SkipReason | null
}
export interface PreviewSummary {
totalEmails: number
willGrant: number
willSkip: { notFound: number, userDeleted: number, duplicateInInput: number }
totalFluxToIssue: number
samples: { willGrant: string[], notFound: string[], userDeleted: string[] }
}
export interface GrantResult {
granted: { email: string, userId: string, fluxTransactionId: string, balanceAfter: number }[]
skipped: { email: string, reason: SkipReason }[]
failed: { email: string, userId: string, error: string }[]
}
export interface GrantInput {
amount: number
description: string
emails: string[]
createdByUserId: string
/**
* When provided, recipient `requestId`s are derived as
* `flux-grant:${idempotencyKey}:${userId}` so re-running the same call
* with the same key + recipients is a no-op (handled by the partial
* unique index on `flux_transaction(user_id, request_id)`).
* When omitted, every recipient gets a fresh requestId re-issuing the
* same grant will double-credit, which is the right default for "I made
* a typo and want to send again".
*/
idempotencyKey?: string
}
/**
* Resolve operator-supplied input emails against the user table.
*
* Use when:
* - Either `preview` (dry-run) or the actual grant call needs the same
* per-email outcome shape
*
* Expects:
* - `user.email` is stored lowercase (better-auth normalizes on signup
* for both email/password and OAuth). Wrapping the column in `LOWER()`
* in the query would bypass the unique index on `email` and force a
* sequential scan, so input is lowercased instead.
*
* Returns:
* - One `ResolvedRecipient` per input email (duplicates included with
* `duplicate_in_input` so the caller can audit them)
*/
async function resolveEmails(db: Database, emails: string[]): Promise<ResolvedRecipient[]> {
const seenLower = new Map<string, number>()
const resolved: ResolvedRecipient[] = emails.map((email, idx) => {
const lower = email.toLowerCase()
if (seenLower.has(lower))
return { inputEmail: email, userId: null, status: 'skipped', errorReason: 'duplicate_in_input' }
seenLower.set(lower, idx)
return { inputEmail: email, userId: null, status: 'pending', errorReason: null }
})
const lowerEmails = Array.from(seenLower.keys())
const users = lowerEmails.length === 0
? []
: await db
.select({ id: accountsSchema.user.id, email: accountsSchema.user.email })
.from(accountsSchema.user)
.where(inArray(accountsSchema.user.email, lowerEmails))
const userByLowerEmail = new Map(users.map(u => [u.email.toLowerCase(), u.id]))
const matchedUserIds = users.map(u => u.id)
const fluxRows = matchedUserIds.length === 0
? []
: await db
.select({ userId: fluxSchema.userFlux.userId, deletedAt: fluxSchema.userFlux.deletedAt })
.from(fluxSchema.userFlux)
.where(inArray(fluxSchema.userFlux.userId, matchedUserIds))
const deletedUserIds = new Set(fluxRows.filter(r => r.deletedAt != null).map(r => r.userId))
for (const entry of resolved) {
if (entry.errorReason === 'duplicate_in_input')
continue
const userId = userByLowerEmail.get(entry.inputEmail.toLowerCase())
if (!userId) {
entry.status = 'skipped'
entry.errorReason = 'not_found'
continue
}
if (deletedUserIds.has(userId)) {
entry.userId = userId
entry.status = 'skipped'
entry.errorReason = 'user_deleted'
continue
}
entry.userId = userId
entry.status = 'pending'
}
return resolved
}
function buildPreviewSummary(resolved: ResolvedRecipient[], amountPerUser: number): PreviewSummary {
const willGrant = resolved.filter(r => r.status === 'pending').length
const notFound = resolved.filter(r => r.errorReason === 'not_found').length
const userDeleted = resolved.filter(r => r.errorReason === 'user_deleted').length
const duplicateInInput = resolved.filter(r => r.errorReason === 'duplicate_in_input').length
return {
totalEmails: resolved.length,
willGrant,
willSkip: { notFound, userDeleted, duplicateInInput },
totalFluxToIssue: willGrant * amountPerUser,
samples: {
willGrant: resolved.filter(r => r.status === 'pending').slice(0, 5).map(r => r.inputEmail),
notFound: resolved.filter(r => r.errorReason === 'not_found').slice(0, 5).map(r => r.inputEmail),
userDeleted: resolved.filter(r => r.errorReason === 'user_deleted').slice(0, 5).map(r => r.inputEmail),
},
}
}
export function createAdminFluxGrantsService(deps: { db: Database, billingService: BillingService }) {
const { db, billingService } = deps
return {
/**
* Dry-run preview: returns what would happen without writing anything.
*/
async preview(input: { amount: number, emails: string[] }): Promise<PreviewSummary> {
const resolved = await resolveEmails(db, input.emails)
return buildPreviewSummary(resolved, input.amount)
},
/**
* Issue a grant to every resolvable email, sequentially.
*
* Use when:
* - Admin clicks "send" on a grant. Returns once every recipient has
* either been credited, marked skipped (resolution-time issue), or
* marked failed (`creditFlux` threw).
*
* Expects:
* - Caller has admin authority (route middleware enforces this)
* - Batch size fits inside the load balancer timeout the route
* layer caps `emails.length`
*
* Returns:
* - Per-email outcome buckets. The same `inputEmail` order is preserved
* inside each bucket so the operator can spot recipient-specific
* issues without correlating across responses.
*/
async grant(input: GrantInput): Promise<{ summary: PreviewSummary, result: GrantResult }> {
const resolved = await resolveEmails(db, input.emails)
const summary = buildPreviewSummary(resolved, input.amount)
const result: GrantResult = { granted: [], skipped: [], failed: [] }
for (const entry of resolved) {
if (entry.status === 'skipped') {
result.skipped.push({ email: entry.inputEmail, reason: entry.errorReason ?? 'not_found' })
continue
}
// entry.status === 'pending' implies userId is set
const userId = entry.userId!
const requestId = input.idempotencyKey != null
? `flux-grant:${input.idempotencyKey}:${userId}`
: undefined
try {
const credited = await billingService.creditFlux({
userId,
amount: input.amount,
type: 'promo',
requestId,
description: input.description,
source: 'admin_promo',
auditMetadata: {
description: input.description,
issuedByUserId: input.createdByUserId,
...(input.idempotencyKey != null && { idempotencyKey: input.idempotencyKey }),
},
})
result.granted.push({
email: entry.inputEmail,
userId,
fluxTransactionId: credited.fluxTransactionId,
balanceAfter: credited.balanceAfter,
})
}
catch (err) {
const message = errorMessageFrom(err) ?? 'Unknown error'
result.failed.push({ email: entry.inputEmail, userId, error: message.slice(0, 500) })
logger.withError(err).withFields({ userId, email: entry.inputEmail }).warn('Flux grant failed')
}
}
logger.withFields({
description: input.description,
attempted: summary.willGrant,
granted: result.granted.length,
skipped: result.skipped.length,
failed: result.failed.length,
amount: input.amount,
issuedByUserId: input.createdByUserId,
}).log('Admin flux grant completed')
return { summary, result }
},
}
}
export type AdminFluxGrantsService = ReturnType<typeof createAdminFluxGrantsService>
/**
* Exported for unit tests of resolution edge cases (case folding, duplicate
* handling, soft-delete detection).
*/
export { resolveEmails }
@@ -0,0 +1,328 @@
import type { Database } from '../../../libs/db'
import type { BillingService } from '../../billing/billing-service'
import { eq } from 'drizzle-orm'
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
import { createAdminFluxGrantsService, resolveEmails } from '..'
import { mockDB } from '../../../libs/mock-db'
import * as schema from '../../../schemas'
describe('resolveEmails', () => {
let db: Database
beforeAll(async () => {
db = await mockDB(schema)
// Three users: one normal, one with deleted user_flux, one we never insert
// user_flux for at all (so it shows up as "user exists, no flux row" → still
// pending since user.id matches; default flux init happens at credit time).
//
// NOTICE: All stored emails are lowercase. resolveEmails relies on this
// (Codex review 2026-05-08 flagged that wrapping user.email in LOWER()
// bypasses the unique index and seq-scans). better-auth normalizes emails
// on signup, so this matches production reality.
await db.insert(schema.user).values([
{ id: 'uid_normal', name: 'Normal', email: 'normal@example.com' },
{ id: 'uid_deleted', name: 'Deleted', email: 'deleted@example.com' },
{ id: 'uid_no_flux', name: 'NoFlux', email: 'noflux@example.com' },
{ id: 'uid_mixed_case', name: 'MixedCase', email: 'Mixed@Example.com' },
])
await db.insert(schema.userFlux).values([
{ userId: 'uid_normal', flux: 100 },
{ userId: 'uid_deleted', flux: 0, deletedAt: new Date() },
])
})
it('lowercases input before matching the (lowercase) stored email', async () => {
const resolved = await resolveEmails(db, ['NORMAL@example.com'])
expect(resolved).toHaveLength(1)
expect(resolved[0]).toMatchObject({
inputEmail: 'NORMAL@example.com',
userId: 'uid_normal',
status: 'pending',
errorReason: null,
})
})
it('treats non-lowercase stored emails as not_found (documented limitation)', async () => {
// Stored email is 'Mixed@Example.com' (mixed case); we look up by lowercase
// 'mixed@example.com', which won't match because we don't wrap user.email
// in LOWER() — that would defeat the unique index and seq-scan the table.
const resolved = await resolveEmails(db, ['mixed@example.com'])
expect(resolved[0]).toMatchObject({
inputEmail: 'mixed@example.com',
userId: null,
status: 'skipped',
errorReason: 'not_found',
})
})
it('marks unknown emails as not_found', async () => {
const resolved = await resolveEmails(db, ['ghost@example.com'])
expect(resolved[0]).toMatchObject({
inputEmail: 'ghost@example.com',
userId: null,
status: 'skipped',
errorReason: 'not_found',
})
})
it('marks soft-deleted users as user_deleted (userId still attached for audit)', async () => {
const resolved = await resolveEmails(db, ['deleted@example.com'])
expect(resolved[0]).toMatchObject({
inputEmail: 'deleted@example.com',
userId: 'uid_deleted',
status: 'skipped',
errorReason: 'user_deleted',
})
})
it('keeps the first occurrence and tags subsequent duplicates', async () => {
const resolved = await resolveEmails(db, [
'normal@example.com',
'NORMAL@EXAMPLE.COM',
'normal@example.com',
])
expect(resolved).toHaveLength(3)
expect(resolved[0].errorReason).toBeNull()
expect(resolved[0].status).toBe('pending')
expect(resolved[1].errorReason).toBe('duplicate_in_input')
expect(resolved[2].errorReason).toBe('duplicate_in_input')
})
it('returns empty resolution for empty input without hitting the DB', async () => {
const resolved = await resolveEmails(db, [])
expect(resolved).toEqual([])
})
})
describe('adminFluxGrantsService.preview', () => {
let db: Database
beforeAll(async () => {
db = await mockDB(schema)
await db.insert(schema.user).values([
{ id: 'uid_prev_a', name: 'A', email: 'preva@example.com' },
])
await db.insert(schema.userFlux).values([
{ userId: 'uid_prev_a', flux: 0 },
])
})
it('returns counts and samples without writing anything', async () => {
const billingService = { creditFlux: vi.fn() } as unknown as BillingService
const service = createAdminFluxGrantsService({ db, billingService })
const summary = await service.preview({
amount: 50,
emails: ['preva@example.com', 'ghost@example.com', 'preva@example.com'],
})
expect(summary).toEqual({
totalEmails: 3,
willGrant: 1,
willSkip: { notFound: 1, userDeleted: 0, duplicateInInput: 1 },
totalFluxToIssue: 50,
samples: {
willGrant: ['preva@example.com'],
notFound: ['ghost@example.com'],
userDeleted: [],
},
})
expect(billingService.creditFlux).not.toHaveBeenCalled()
})
it('caps preview samples at 5 entries per category', async () => {
const billingService = { creditFlux: vi.fn() } as unknown as BillingService
const service = createAdminFluxGrantsService({ db, billingService })
const ghosts = Array.from({ length: 12 }, (_, i) => `ghost${i}@example.com`)
const summary = await service.preview({ amount: 50, emails: ghosts })
expect(summary.willSkip.notFound).toBe(12)
expect(summary.samples.notFound).toHaveLength(5)
})
})
describe('adminFluxGrantsService.grant', () => {
let db: Database
beforeAll(async () => {
db = await mockDB(schema)
await db.insert(schema.user).values([
{ id: 'uid_grant_a', name: 'A', email: 'granta@example.com' },
{ id: 'uid_grant_b', name: 'B', email: 'grantb@example.com' },
{ id: 'uid_grant_c', name: 'C', email: 'grantc@example.com' },
])
await db.insert(schema.userFlux).values([
{ userId: 'uid_grant_a', flux: 0 },
{ userId: 'uid_grant_b', flux: 0, deletedAt: new Date() },
{ userId: 'uid_grant_c', flux: 0 },
])
})
beforeEach(async () => {
await db.delete(schema.fluxTransaction)
})
it('credits resolvable recipients and bucket-sorts the per-email outcomes', async () => {
const creditFlux = vi.fn(async ({ userId }: { userId: string }) => ({
balanceBefore: 0,
balanceAfter: 100,
fluxTransactionId: `ftx-${userId}`,
idempotent: false,
}))
const billingService = { creditFlux } as unknown as BillingService
const service = createAdminFluxGrantsService({ db, billingService })
const { summary, result } = await service.grant({
amount: 100,
description: 'Beta thanks',
emails: [
'granta@example.com', // pending → granted
'grantb@example.com', // soft-deleted → skipped(user_deleted)
'GRANTA@example.com', // duplicate
'ghost@example.com', // not_found
],
createdByUserId: 'uid_admin',
})
expect(summary).toMatchObject({
totalEmails: 4,
willGrant: 1,
willSkip: { notFound: 1, userDeleted: 1, duplicateInInput: 1 },
totalFluxToIssue: 100,
})
expect(result.granted).toEqual([{
email: 'granta@example.com',
userId: 'uid_grant_a',
fluxTransactionId: 'ftx-uid_grant_a',
balanceAfter: 100,
}])
expect(result.skipped).toEqual(expect.arrayContaining([
{ email: 'grantb@example.com', reason: 'user_deleted' },
{ email: 'GRANTA@example.com', reason: 'duplicate_in_input' },
{ email: 'ghost@example.com', reason: 'not_found' },
]))
expect(result.failed).toEqual([])
expect(creditFlux).toHaveBeenCalledTimes(1)
expect(creditFlux).toHaveBeenCalledWith(expect.objectContaining({
userId: 'uid_grant_a',
amount: 100,
type: 'promo',
description: 'Beta thanks',
source: 'admin_promo',
requestId: undefined, // no idempotencyKey in this test
auditMetadata: expect.objectContaining({
description: 'Beta thanks',
issuedByUserId: 'uid_admin',
}),
}))
})
it('catches per-recipient creditFlux errors and continues with the rest', async () => {
let calls = 0
const creditFlux = vi.fn(async ({ userId }: { userId: string }) => {
calls += 1
if (calls === 1)
throw new Error('DB timeout')
return {
balanceBefore: 0,
balanceAfter: 100,
fluxTransactionId: `ftx-${userId}`,
idempotent: false,
}
})
const billingService = { creditFlux } as unknown as BillingService
const service = createAdminFluxGrantsService({ db, billingService })
const { result } = await service.grant({
amount: 100,
description: 'Resilience test',
emails: ['granta@example.com', 'grantc@example.com'],
createdByUserId: 'uid_admin',
})
expect(result.granted).toHaveLength(1)
expect(result.granted[0].userId).toBe('uid_grant_c')
expect(result.failed).toHaveLength(1)
expect(result.failed[0]).toMatchObject({ email: 'granta@example.com', userId: 'uid_grant_a', error: 'DB timeout' })
})
it('forwards a deterministic requestId per recipient when idempotencyKey is provided', async () => {
const creditFlux = vi.fn(async () => ({
balanceBefore: 0,
balanceAfter: 100,
fluxTransactionId: 'ftx',
idempotent: false,
}))
const billingService = { creditFlux } as unknown as BillingService
const service = createAdminFluxGrantsService({ db, billingService })
await service.grant({
amount: 100,
description: 'Idempotent thanks',
emails: ['granta@example.com', 'grantc@example.com'],
createdByUserId: 'uid_admin',
idempotencyKey: 'beta-2026-q2',
})
expect(creditFlux).toHaveBeenNthCalledWith(1, expect.objectContaining({
userId: 'uid_grant_a',
requestId: 'flux-grant:beta-2026-q2:uid_grant_a',
}))
expect(creditFlux).toHaveBeenNthCalledWith(2, expect.objectContaining({
userId: 'uid_grant_c',
requestId: 'flux-grant:beta-2026-q2:uid_grant_c',
}))
})
it('end-to-end: actually writes flux_transaction rows for granted recipients via the real BillingService path', async () => {
// Light integration sanity check — we still mock BillingService here, but
// verify the service pipes through the right shape and granted set
// matches what the route would surface.
const inserted: { userId: string, requestId?: string }[] = []
const creditFlux = vi.fn(async ({ userId, requestId, amount }: { userId: string, requestId?: string, amount: number }) => {
const [row] = await db.insert(schema.fluxTransaction).values({
userId,
type: 'promo',
amount,
balanceBefore: 0,
balanceAfter: amount,
requestId: requestId ?? null,
description: 'mocked',
}).returning()
inserted.push({ userId, requestId })
return {
balanceBefore: 0,
balanceAfter: amount,
fluxTransactionId: row!.id,
idempotent: false,
}
})
const billingService = { creditFlux } as unknown as BillingService
const service = createAdminFluxGrantsService({ db, billingService })
const { result } = await service.grant({
amount: 25,
description: 'Integration test',
emails: ['granta@example.com', 'grantc@example.com'],
createdByUserId: 'uid_admin',
idempotencyKey: 'int-1',
})
expect(result.granted).toHaveLength(2)
const ledger = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.requestId, 'flux-grant:int-1:uid_grant_a'))
expect(ledger).toHaveLength(1)
expect(ledger[0]?.amount).toBe(25)
expect(inserted.map(r => r.requestId).sort()).toEqual([
'flux-grant:int-1:uid_grant_a',
'flux-grant:int-1:uid_grant_c',
])
})
})
@@ -1,86 +0,0 @@
import type { Database } from '../../libs/db'
import type { StreamMessage } from '../../libs/mq'
import type { BillingEvent } from './billing-events'
import { useLogger } from '@guiiai/logg'
import * as fluxTxSchema from '../../schemas/flux-transaction'
import * as llmRequestLogSchema from '../../schemas/llm-request-log'
const logger = useLogger('billing-consumer-handler').useGlobalConfig()
export function createBillingConsumerHandler(db: Database) {
return {
async handleMessage(message: StreamMessage<BillingEvent>): Promise<void> {
const { event } = message
switch (event.eventType) {
case 'flux.debited': {
const balanceBefore = event.payload.balanceAfter != null
? event.payload.balanceAfter + event.payload.amount
: 0
// NOTICE: onConflictDoNothing handles redelivery after crash —
// the unique index (userId, requestId) prevents duplicate transaction entries.
await db.insert(fluxTxSchema.fluxTransaction).values({
userId: event.userId,
type: 'debit',
amount: event.payload.amount,
balanceBefore,
balanceAfter: event.payload.balanceAfter ?? balanceBefore - event.payload.amount,
requestId: event.requestId,
description: event.payload.description ?? event.payload.source ?? 'LLM request',
metadata: event.payload.metadata != null || event.payload.source != null
? {
...(event.payload.metadata as Record<string, unknown>),
source: event.payload.source,
}
: undefined,
}).onConflictDoNothing()
logger.withFields({
eventId: event.eventId,
userId: event.userId,
amount: event.payload.amount,
}).log('Wrote debit transaction')
break
}
case 'llm.request.log': {
// NOTICE: Use eventId as PK to make redelivery idempotent.
await db.insert(llmRequestLogSchema.llmRequestLog).values({
id: event.eventId,
userId: event.userId,
model: event.payload.model,
status: event.payload.status,
durationMs: event.payload.durationMs,
fluxConsumed: event.payload.fluxConsumed,
promptTokens: event.payload.promptTokens,
completionTokens: event.payload.completionTokens,
}).onConflictDoNothing()
logger.withFields({
eventId: event.eventId,
userId: event.userId,
model: event.payload.model,
}).log('Wrote LLM request log')
break
}
case 'flux.credited':
case 'stripe.checkout.completed':
case 'llm.request.completed': {
// These events are handled synchronously or not yet consumed.
// Log for observability but no async DB writes needed.
logger.withFields({
eventId: event.eventId,
eventType: event.eventType,
}).log('Acknowledged event (no async action)')
break
}
}
},
}
}
export type BillingConsumerHandler = ReturnType<typeof createBillingConsumerHandler>
@@ -1,197 +0,0 @@
import type { InferOutput } from 'valibot'
import type { RedisCommandClient } from '../../libs/mq'
import {
literal,
nonEmpty,
number,
object,
optional,
parse,
pipe,
string,
union,
unknown,
} from 'valibot'
import { createMqService } from '../../libs/mq'
import { DEFAULT_BILLING_EVENTS_STREAM } from '../../utils/redis-keys'
const BillingEventTypeSchema = union([
literal('flux.debited'),
literal('flux.credited'),
literal('stripe.checkout.completed'),
literal('llm.request.completed'),
literal('llm.request.log'),
])
const BalanceChangePayloadSchema = object({
amount: number(),
balanceAfter: optional(number()),
source: optional(pipe(string(), nonEmpty())),
description: optional(pipe(string(), nonEmpty())),
metadata: optional(unknown()),
})
const StripeCheckoutCompletedPayloadSchema = object({
stripeEventId: pipe(string(), nonEmpty()),
stripeSessionId: pipe(string(), nonEmpty()),
amount: number(),
currency: pipe(string(), nonEmpty()),
})
const LlmRequestCompletedPayloadSchema = object({
model: pipe(string(), nonEmpty()),
status: number(),
fluxConsumed: number(),
promptTokens: optional(number()),
completionTokens: optional(number()),
})
const LlmRequestLogPayloadSchema = object({
model: pipe(string(), nonEmpty()),
status: number(),
durationMs: number(),
fluxConsumed: number(),
promptTokens: optional(number()),
completionTokens: optional(number()),
})
const BillingEventEnvelopeSchema = object({
eventId: pipe(string(), nonEmpty()),
eventType: BillingEventTypeSchema,
aggregateId: pipe(string(), nonEmpty()),
userId: pipe(string(), nonEmpty()),
requestId: optional(pipe(string(), nonEmpty())),
occurredAt: pipe(string(), nonEmpty()),
schemaVersion: number(),
payload: unknown(),
})
export type BillingEventType = InferOutput<typeof BillingEventTypeSchema>
type BillingEventEnvelope = InferOutput<typeof BillingEventEnvelopeSchema>
type BalanceChangePayload = InferOutput<typeof BalanceChangePayloadSchema>
type StripeCheckoutCompletedPayload = InferOutput<typeof StripeCheckoutCompletedPayloadSchema>
type LlmRequestCompletedPayload = InferOutput<typeof LlmRequestCompletedPayloadSchema>
type LlmRequestLogPayload = InferOutput<typeof LlmRequestLogPayloadSchema>
export type FluxDebitedEvent = BillingEventEnvelope & {
eventType: 'flux.debited'
payload: BalanceChangePayload
}
export type FluxCreditedEvent = BillingEventEnvelope & {
eventType: 'flux.credited'
payload: BalanceChangePayload
}
export type StripeCheckoutCompletedEvent = BillingEventEnvelope & {
eventType: 'stripe.checkout.completed'
payload: StripeCheckoutCompletedPayload
}
export type LlmRequestCompletedEvent = BillingEventEnvelope & {
eventType: 'llm.request.completed'
payload: LlmRequestCompletedPayload
}
export type LlmRequestLogEvent = BillingEventEnvelope & {
eventType: 'llm.request.log'
payload: LlmRequestLogPayload
}
export type BillingEvent
= | FluxDebitedEvent
| FluxCreditedEvent
| StripeCheckoutCompletedEvent
| LlmRequestCompletedEvent
| LlmRequestLogEvent
export interface SerializedBillingEventFields extends Record<string, string | undefined> {
event_id: string
event_type: BillingEventType
aggregate_id: string
user_id: string
request_id?: string
occurred_at: string
schema_version: string
payload: string
}
export function serializeBillingEvent(event: BillingEvent): SerializedBillingEventFields {
return {
event_id: event.eventId,
event_type: event.eventType,
aggregate_id: event.aggregateId,
user_id: event.userId,
request_id: event.requestId,
occurred_at: event.occurredAt,
schema_version: String(event.schemaVersion),
payload: JSON.stringify(event.payload),
}
}
/**
* Create a Redis Stream MQ service pre-configured for billing events.
*/
export function createBillingMq(redis: RedisCommandClient, options: { stream?: string, maxLength?: number } = {}) {
return createMqService<BillingEvent>(redis, {
stream: options.stream ?? DEFAULT_BILLING_EVENTS_STREAM,
maxLength: options.maxLength,
serialize: serializeBillingEvent,
deserialize: parseBillingEvent,
})
}
export function parseBillingEvent(fields: Record<string, string | undefined>): BillingEvent {
const payload = fields.payload
if (payload == null) {
throw new TypeError('Billing event payload is required')
}
const parsedEnvelope = parse(BillingEventEnvelopeSchema, {
eventId: fields.event_id,
eventType: fields.event_type,
aggregateId: fields.aggregate_id,
userId: fields.user_id,
requestId: fields.request_id,
occurredAt: fields.occurred_at,
schemaVersion: Number(fields.schema_version),
payload: JSON.parse(payload),
})
switch (parsedEnvelope.eventType) {
case 'flux.debited':
return {
...parsedEnvelope,
eventType: 'flux.debited',
payload: parse(BalanceChangePayloadSchema, parsedEnvelope.payload),
}
case 'flux.credited':
return {
...parsedEnvelope,
eventType: 'flux.credited',
payload: parse(BalanceChangePayloadSchema, parsedEnvelope.payload),
}
case 'stripe.checkout.completed':
return {
...parsedEnvelope,
eventType: 'stripe.checkout.completed',
payload: parse(StripeCheckoutCompletedPayloadSchema, parsedEnvelope.payload),
}
case 'llm.request.completed':
return {
...parsedEnvelope,
eventType: 'llm.request.completed',
payload: parse(LlmRequestCompletedPayloadSchema, parsedEnvelope.payload),
}
case 'llm.request.log':
return {
...parsedEnvelope,
eventType: 'llm.request.log',
payload: parse(LlmRequestLogPayloadSchema, parsedEnvelope.payload),
}
}
}
@@ -1,16 +1,13 @@
import type Redis from 'ioredis'
import type { Database } from '../../libs/db'
import type { MqService } from '../../libs/mq'
import type { RevenueMetrics } from '../../libs/otel'
import type { ConfigKVService } from '../config-kv'
import type { BillingEvent } from './billing-events'
import { useLogger } from '@guiiai/logg'
import { and, eq } from 'drizzle-orm'
import { createPaymentRequiredError } from '../../utils/error'
import { nanoid } from '../../utils/id'
import { userFluxRedisKey } from '../../utils/redis-keys'
import * as fluxSchema from '../../schemas/flux'
@@ -22,7 +19,6 @@ const logger = useLogger('billing-service')
export function createBillingService(
db: Database,
redis: Redis,
billingMq: MqService<BillingEvent>,
_configKV: ConfigKVService,
metrics?: RevenueMetrics | null,
) {
@@ -40,27 +36,12 @@ export function createBillingService(
}
/**
* Publish a billing event to the Redis Stream.
* Best-effort: failures are logged but not re-thrown so callers are not blocked.
*/
async function publishEvent(event: BillingEvent): Promise<void> {
try {
await billingMq.publish(event)
}
catch (error) {
logger.withError(error).withFields({
eventId: event.eventId,
eventType: event.eventType,
userId: event.userId,
}).error('Failed to publish billing event to stream')
}
}
/**
* Debit flux from a user's balance within a DB transaction.
* The transaction ONLY locks the row and updates the balance.
* Transaction entries are written by the billing-mq consumer
* after it processes the flux.debited event published post-commit.
* Debit flux from a user's balance within a single DB transaction.
*
* The transaction locks the user_flux row, validates the balance, updates
* it, and writes the matching `flux_transaction` ledger entry all in one
* commit. The unique partial index `(user_id, request_id) WHERE request_id IS NOT NULL`
* keeps retries idempotent at the DB level.
*
* Private call domain-specific wrappers (e.g. consumeFluxForLLM) instead.
*/
@@ -73,7 +54,27 @@ export function createBillingService(
metadata?: Record<string, unknown>
}): Promise<{ userId: string, flux: number }> {
const result = await db.transaction(async (tx) => {
// 1. Lock the row and read current balance
// Idempotency: a previous successful debit with the same requestId
// returns the prior post-balance and skips the second deduction.
// Mirrors creditFlux's idempotent path so retries (network errors,
// worker restarts) don't double-charge.
if (input.requestId != null) {
const [existing] = await tx
.select({
balanceAfter: fluxTxSchema.fluxTransaction.balanceAfter,
})
.from(fluxTxSchema.fluxTransaction)
.where(and(
eq(fluxTxSchema.fluxTransaction.userId, input.userId),
eq(fluxTxSchema.fluxTransaction.requestId, input.requestId),
))
.limit(1)
if (existing) {
return { userId: input.userId, flux: existing.balanceAfter, idempotent: true as const }
}
}
const [row] = await tx
.select({ flux: fluxSchema.userFlux.flux })
.from(fluxSchema.userFlux)
@@ -92,44 +93,42 @@ export function createBillingService(
const balanceAfter = balanceBefore - input.amount
// 2. Update balance
await tx.update(fluxSchema.userFlux)
.set({ flux: balanceAfter, updatedAt: new Date() })
.where(eq(fluxSchema.userFlux.userId, input.userId))
return { userId: input.userId, flux: balanceAfter, balanceBefore }
})
// 3. Update Redis cache after commit (best-effort)
await updateRedisCache(input.userId, result.flux)
// 4. Publish flux.debited event to stream; transaction + audit written by consumer
await publishEvent({
eventId: nanoid(),
eventType: 'flux.debited',
aggregateId: input.userId,
userId: input.userId,
requestId: input.requestId,
occurredAt: new Date().toISOString(),
schemaVersion: 1,
payload: {
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: 'debit',
amount: input.amount,
balanceAfter: result.flux,
source: input.source,
description: input.description,
metadata: input.metadata,
},
balanceBefore,
balanceAfter,
requestId: input.requestId,
description: input.description ?? input.source,
metadata: input.metadata != null || input.source != null
? {
...input.metadata,
source: input.source,
}
: undefined,
})
return { userId: input.userId, flux: balanceAfter, idempotent: false as const }
})
logger.withFields({ userId: input.userId, amount: input.amount, balance: result.flux }).log('Debited flux')
if (!result.idempotent) {
await updateRedisCache(input.userId, result.flux)
}
logger.withFields({ userId: input.userId, amount: input.amount, balance: result.flux, idempotent: result.idempotent }).log('Debited flux')
return { userId: result.userId, flux: result.flux }
}
return {
/**
* Debit flux for an LLM API request (chat, TTS).
* Passes token usage as opaque metadata carried through the flux.debited event
* so the billing-mq consumer can write it to the transaction log.
* Token usage is persisted in the `flux_transaction.metadata` column so
* the existing transaction-history UI can render per-request token counts.
*/
async consumeFluxForLLM(input: {
userId: string
@@ -157,7 +156,22 @@ export function createBillingService(
/**
* Credit flux to a user's balance within a DB transaction.
* Generic credit method for non-Stripe flows (e.g. admin grants).
* Transaction entries are written inside the transaction for immediate visibility.
*
* Idempotency:
* When `requestId` is provided, the call is idempotent across crash /
* retry boundaries. If a `flux_transaction` row with the same
* `(user_id, request_id)` already exists, this method returns that
* existing row's balance + id without re-crediting the user, without
* touching `user_flux`, and without re-emitting the Redis cache write.
*
* This guards against the worker crash window where:
* 1. `creditFlux` commits the credit
* 2. caller crashes before marking its own state (e.g. recipient row) granted
* 3. on restart, caller sees pending state and calls `creditFlux` again with same requestId
*
* Without idempotency, step 3 would hit the `(user_id, request_id)`
* unique index and throw causing the caller to mark the work failed
* even though the user was already credited.
*/
async creditFlux(input: {
userId: string
@@ -172,15 +186,38 @@ export function createBillingService(
*/
type?: 'credit' | 'promo'
auditMetadata?: Record<string, unknown>
}): Promise<{ balanceBefore: number, balanceAfter: number, fluxTransactionId: string }> {
}): Promise<{ balanceBefore: number, balanceAfter: number, fluxTransactionId: string, idempotent: boolean }> {
const ledgerType = input.type ?? 'credit'
const result = await db.transaction(async (tx) => {
// Ensure user record exists
const txResult = await db.transaction(async (tx) => {
if (input.requestId != null) {
const [existing] = await tx
.select({
id: fluxTxSchema.fluxTransaction.id,
balanceBefore: fluxTxSchema.fluxTransaction.balanceBefore,
balanceAfter: fluxTxSchema.fluxTransaction.balanceAfter,
})
.from(fluxTxSchema.fluxTransaction)
.where(and(
eq(fluxTxSchema.fluxTransaction.userId, input.userId),
eq(fluxTxSchema.fluxTransaction.requestId, input.requestId),
))
.limit(1)
if (existing) {
return {
balanceBefore: existing.balanceBefore,
balanceAfter: existing.balanceAfter,
fluxTransactionId: existing.id,
idempotent: true,
}
}
}
await tx.insert(fluxSchema.userFlux)
.values({ userId: input.userId, flux: 0 })
.onConflictDoNothing({ target: fluxSchema.userFlux.userId })
// Lock and read current balance
const [row] = await tx
.select({ flux: fluxSchema.userFlux.flux })
.from(fluxSchema.userFlux)
@@ -190,12 +227,10 @@ export function createBillingService(
const balanceBefore = row!.flux
const balanceAfter = balanceBefore + input.amount
// Update balance
await tx.update(fluxSchema.userFlux)
.set({ flux: balanceAfter, updatedAt: new Date() })
.where(eq(fluxSchema.userFlux.userId, input.userId))
// Transaction entry
const [insertedTx] = await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: ledgerType,
@@ -207,35 +242,34 @@ export function createBillingService(
metadata: input.auditMetadata,
}).returning({ id: fluxTxSchema.fluxTransaction.id })
return { balanceBefore, balanceAfter, fluxTransactionId: insertedTx!.id }
return {
balanceBefore,
balanceAfter,
fluxTransactionId: insertedTx!.id,
idempotent: false,
}
})
await updateRedisCache(input.userId, result.balanceAfter)
if (txResult.idempotent) {
logger.withFields({
userId: input.userId,
requestId: input.requestId,
fluxTransactionId: txResult.fluxTransactionId,
}).log('Credited flux (idempotent replay — no side effects emitted)')
return txResult
}
// Publish flux.credited event after commit
await publishEvent({
eventId: nanoid(),
eventType: 'flux.credited',
aggregateId: input.userId,
userId: input.userId,
requestId: input.requestId,
occurredAt: new Date().toISOString(),
schemaVersion: 1,
payload: {
amount: input.amount,
balanceAfter: result.balanceAfter,
source: input.source,
},
})
await updateRedisCache(input.userId, txResult.balanceAfter)
logger.withFields({ userId: input.userId, amount: input.amount, balance: result.balanceAfter }).log('Credited flux')
return result
logger.withFields({ userId: input.userId, amount: input.amount, balance: txResult.balanceAfter }).log('Credited flux')
return txResult
},
/**
* Credit flux from a Stripe checkout session (one-time payment).
* Idempotent: checks fluxCredited flag before applying.
* Transaction entries are written inside the transaction for immediate visibility.
* Idempotent: claims the checkout session row by flipping `fluxCredited`
* from false to true; replays of the same Stripe event observe the row
* already claimed and apply nothing.
*/
async creditFluxFromStripeCheckout(input: {
stripeEventId: string
@@ -251,7 +285,6 @@ export function createBillingService(
// checkout session row exactly once via `fluxCredited = false -> true`, which
// covers both Stripe retries of the same event and distinct Event objects that
// still refer to the same checkout session.
// Atomic claim: set fluxCredited = true only if currently false
const [claimed] = await tx.update(stripeSchema.stripeCheckoutSession)
.set({ fluxCredited: true, updatedAt: new Date() })
.where(and(
@@ -264,12 +297,10 @@ export function createBillingService(
return { applied: false }
}
// Ensure user record exists
await tx.insert(fluxSchema.userFlux)
.values({ userId: input.userId, flux: 0 })
.onConflictDoNothing({ target: fluxSchema.userFlux.userId })
// Lock and read balance
const [currentFlux] = await tx
.select({ flux: fluxSchema.userFlux.flux })
.from(fluxSchema.userFlux)
@@ -279,14 +310,12 @@ export function createBillingService(
const balanceBefore = currentFlux!.flux
const balanceAfter = balanceBefore + input.fluxAmount
// Update balance
await tx.update(fluxSchema.userFlux)
.set({ flux: balanceAfter, updatedAt: new Date() })
.where(eq(fluxSchema.userFlux.userId, input.userId))
const description = `Stripe payment ${input.currency?.toUpperCase() ?? 'UNKNOWN'} ${(input.amountTotal / 100).toFixed(2)}`
// Transaction entry
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: 'credit',
@@ -307,39 +336,6 @@ export function createBillingService(
if (txResult.applied && txResult.balanceAfter != null) {
await updateRedisCache(input.userId, txResult.balanceAfter)
// Publish both events after commit
const occurredAt = new Date().toISOString()
await publishEvent({
eventId: nanoid(),
eventType: 'flux.credited',
aggregateId: input.userId,
userId: input.userId,
requestId: input.stripeEventId,
occurredAt,
schemaVersion: 1,
payload: {
amount: input.fluxAmount,
balanceAfter: txResult.balanceAfter,
source: 'stripe.checkout.completed',
},
})
await publishEvent({
eventId: nanoid(),
eventType: 'stripe.checkout.completed',
aggregateId: input.stripeSessionId,
userId: input.userId,
requestId: input.stripeEventId,
occurredAt,
schemaVersion: 1,
payload: {
stripeEventId: input.stripeEventId,
stripeSessionId: input.stripeSessionId,
amount: input.amountTotal,
currency: input.currency ?? 'unknown',
},
})
}
return txResult
@@ -347,8 +343,8 @@ export function createBillingService(
/**
* Credit flux from a Stripe invoice payment (subscription).
* Idempotent: checks fluxCredited flag on the invoice record.
* Transaction entries are written inside the transaction for immediate visibility.
* Idempotent: claims the invoice row by flipping `fluxCredited`
* from false to true; replays observe it already claimed and apply nothing.
*/
async creditFluxFromInvoice(input: {
stripeEventId: string
@@ -363,7 +359,6 @@ export function createBillingService(
// as checkout sessions. We intentionally dedupe on the invoice record instead of
// only on Stripe `event.id`, because Stripe may emit multiple events that map to
// the same paid invoice while the balance must only be credited once.
// Atomic claim: set fluxCredited = true only if currently false
const [claimed] = await tx.update(stripeSchema.stripeInvoice)
.set({ fluxCredited: true, updatedAt: new Date() })
.where(and(
@@ -376,12 +371,10 @@ export function createBillingService(
return { applied: false }
}
// Ensure user record exists
await tx.insert(fluxSchema.userFlux)
.values({ userId: input.userId, flux: 0 })
.onConflictDoNothing({ target: fluxSchema.userFlux.userId })
// Lock and read balance
const [currentFlux] = await tx
.select({ flux: fluxSchema.userFlux.flux })
.from(fluxSchema.userFlux)
@@ -391,14 +384,12 @@ export function createBillingService(
const balanceBefore = currentFlux!.flux
const balanceAfter = balanceBefore + input.fluxAmount
// Update balance
await tx.update(fluxSchema.userFlux)
.set({ flux: balanceAfter, updatedAt: new Date() })
.where(eq(fluxSchema.userFlux.userId, input.userId))
const description = `Subscription invoice ${input.currency.toUpperCase()} ${(input.amountPaid / 100).toFixed(2)}`
// Transaction entry
await tx.insert(fluxTxSchema.fluxTransaction).values({
userId: input.userId,
type: 'credit',
@@ -419,22 +410,6 @@ export function createBillingService(
if (txResult.applied && txResult.balanceAfter != null) {
await updateRedisCache(input.userId, txResult.balanceAfter)
// Publish flux.credited event after commit
await publishEvent({
eventId: nanoid(),
eventType: 'flux.credited',
aggregateId: input.userId,
userId: input.userId,
requestId: input.stripeEventId,
occurredAt: new Date().toISOString(),
schemaVersion: 1,
payload: {
amount: input.fluxAmount,
balanceAfter: txResult.balanceAfter,
source: 'invoice.paid',
},
})
}
return txResult
@@ -1,68 +0,0 @@
import type { Database } from '../../../libs/db'
import { eq } from 'drizzle-orm'
import { beforeAll, beforeEach, describe, expect, it } from 'vitest'
import { mockDB } from '../../../libs/mock-db'
import { createBillingConsumerHandler } from '../billing-consumer-handler'
import * as schema from '../../../schemas'
describe('billingConsumerHandler', () => {
let db: Database
beforeAll(async () => {
db = await mockDB(schema)
await db.insert(schema.user).values({
id: 'user-billing-handler-1',
name: 'Billing Handler User',
email: 'billing-handler@example.com',
})
})
beforeEach(async () => {
await db.delete(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, 'user-billing-handler-1'))
})
it('writes debit transaction metadata so token usage can be shown in the UI', async () => {
const handler = createBillingConsumerHandler(db)
await handler.handleMessage({
streamMessageId: '1-0',
event: {
eventId: 'evt-1',
eventType: 'flux.debited',
aggregateId: 'user-billing-handler-1',
userId: 'user-billing-handler-1',
requestId: 'req-1',
occurredAt: '2026-03-27T00:00:00.000Z',
schemaVersion: 1,
payload: {
amount: 3,
balanceAfter: 97,
source: 'llm.request',
description: 'gpt-5',
metadata: { promptTokens: 111, completionTokens: 222 },
},
},
})
const [txRecord] = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.requestId, 'req-1'))
expect(txRecord).toMatchObject({
userId: 'user-billing-handler-1',
type: 'debit',
amount: 3,
balanceBefore: 100,
balanceAfter: 97,
requestId: 'req-1',
description: 'gpt-5',
metadata: {
promptTokens: 111,
completionTokens: 222,
source: 'llm.request',
},
})
})
})
@@ -1,157 +0,0 @@
import { describe, expect, it } from 'vitest'
import { parseBillingEvent, serializeBillingEvent } from '../billing-events'
describe('billingEvents', () => {
it('serializes and parses a flux debited event', () => {
const event = {
eventId: 'evt-1',
eventType: 'flux.debited' as const,
aggregateId: 'user-1',
userId: 'user-1',
requestId: 'req-1',
occurredAt: '2026-03-24T00:00:00.000Z',
schemaVersion: 1,
payload: {
amount: 12,
balanceAfter: 88,
source: 'llm',
description: 'gpt-5',
metadata: { promptTokens: 100, completionTokens: 200 },
},
}
const serialized = serializeBillingEvent(event)
expect(serialized).toEqual({
event_id: 'evt-1',
event_type: 'flux.debited',
aggregate_id: 'user-1',
user_id: 'user-1',
request_id: 'req-1',
occurred_at: '2026-03-24T00:00:00.000Z',
schema_version: '1',
payload: JSON.stringify({
amount: 12,
balanceAfter: 88,
source: 'llm',
description: 'gpt-5',
metadata: { promptTokens: 100, completionTokens: 200 },
}),
})
expect(parseBillingEvent(serialized)).toEqual(event)
})
it('serializes and parses a flux credited event without request id', () => {
const event = {
eventId: 'evt-2',
eventType: 'flux.credited' as const,
aggregateId: 'user-2',
userId: 'user-2',
occurredAt: '2026-03-24T00:00:00.000Z',
schemaVersion: 1,
payload: {
amount: 20,
balanceAfter: 120,
source: 'stripe',
},
}
expect(parseBillingEvent(serializeBillingEvent(event))).toEqual(event)
})
it('parses stripe checkout completed payloads', () => {
const parsed = parseBillingEvent({
event_id: 'evt-3',
event_type: 'stripe.checkout.completed',
aggregate_id: 'checkout-1',
user_id: 'user-3',
occurred_at: '2026-03-24T00:00:00.000Z',
schema_version: '1',
payload: JSON.stringify({
stripeEventId: 'stripe-evt-1',
stripeSessionId: 'cs_test_123',
amount: 999,
currency: 'usd',
}),
})
expect(parsed).toEqual({
eventId: 'evt-3',
eventType: 'stripe.checkout.completed',
aggregateId: 'checkout-1',
userId: 'user-3',
occurredAt: '2026-03-24T00:00:00.000Z',
schemaVersion: 1,
payload: {
stripeEventId: 'stripe-evt-1',
stripeSessionId: 'cs_test_123',
amount: 999,
currency: 'usd',
},
})
})
it('parses llm request completed payloads', () => {
const parsed = parseBillingEvent({
event_id: 'evt-4',
event_type: 'llm.request.completed',
aggregate_id: 'req-4',
user_id: 'user-4',
request_id: 'req-4',
occurred_at: '2026-03-24T00:00:00.000Z',
schema_version: '1',
payload: JSON.stringify({
model: 'gpt-5',
status: 200,
fluxConsumed: 3,
promptTokens: 100,
completionTokens: 200,
}),
})
expect(parsed).toEqual({
eventId: 'evt-4',
eventType: 'llm.request.completed',
aggregateId: 'req-4',
userId: 'user-4',
requestId: 'req-4',
occurredAt: '2026-03-24T00:00:00.000Z',
schemaVersion: 1,
payload: {
model: 'gpt-5',
status: 200,
fluxConsumed: 3,
promptTokens: 100,
completionTokens: 200,
},
})
})
it('throws when payload json is invalid', () => {
expect(() => parseBillingEvent({
event_id: 'evt-5',
event_type: 'flux.debited',
aggregate_id: 'user-5',
user_id: 'user-5',
occurred_at: '2026-03-24T00:00:00.000Z',
schema_version: '1',
payload: '{',
})).toThrow()
})
it('throws when payload shape does not match event type', () => {
expect(() => parseBillingEvent({
event_id: 'evt-6',
event_type: 'stripe.checkout.completed',
aggregate_id: 'checkout-6',
user_id: 'user-6',
occurred_at: '2026-03-24T00:00:00.000Z',
schema_version: '1',
payload: JSON.stringify({
amount: 100,
currency: 'usd',
}),
})).toThrow()
})
})
@@ -1,256 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import { DEFAULT_BILLING_EVENTS_STREAM } from '../../../utils/redis-keys'
import { createBillingMq } from '../billing-events'
function createEvent() {
return {
eventId: 'evt-1',
eventType: 'flux.debited' as const,
aggregateId: 'user-1',
userId: 'user-1',
requestId: 'req-1',
occurredAt: '2026-03-24T00:00:00.000Z',
schemaVersion: 1,
payload: {
amount: 5,
balanceAfter: 95,
source: 'llm',
description: 'gpt-5',
metadata: { promptTokens: 100, completionTokens: 200 },
},
}
}
describe('billingMqService', () => {
it('publishes an event to the configured stream', async () => {
const redis = {
call: vi.fn(async () => '1740000000000-0'),
}
const mq = createBillingMq(redis, {
stream: 'billing-events-test',
maxLength: 1_000,
})
await expect(mq.publish(createEvent())).resolves.toBe('1740000000000-0')
expect(redis.call).toHaveBeenCalledWith(
'XADD',
'billing-events-test',
'MAXLEN',
'~',
1000,
'*',
'event_id',
'evt-1',
'event_type',
'flux.debited',
'aggregate_id',
'user-1',
'user_id',
'user-1',
'request_id',
'req-1',
'occurred_at',
'2026-03-24T00:00:00.000Z',
'schema_version',
'1',
'payload',
JSON.stringify({
amount: 5,
balanceAfter: 95,
source: 'llm',
description: 'gpt-5',
metadata: { promptTokens: 100, completionTokens: 200 },
}),
)
})
it('throws when publish does not return a stream message id', async () => {
const mq = createBillingMq({
call: vi.fn(async () => 123),
})
await expect(mq.publish(createEvent())).rejects.toThrow('Redis XADD did not return a stream message id')
})
it('creates a consumer group and returns true when the group is new', async () => {
const mq = createBillingMq({
call: vi.fn(async () => 'OK'),
})
await expect(mq.ensureConsumerGroup('billing')).resolves.toBe(true)
})
it('returns false when the consumer group already exists', async () => {
const mq = createBillingMq({
call: vi.fn(async () => {
throw new Error('BUSYGROUP Consumer Group name already exists')
}),
})
await expect(mq.ensureConsumerGroup('billing')).resolves.toBe(false)
})
it('rethrows non-BUSYGROUP errors when creating a consumer group', async () => {
const mq = createBillingMq({
call: vi.fn(async () => {
throw new Error('NOAUTH')
}),
})
await expect(mq.ensureConsumerGroup('billing')).rejects.toThrow('NOAUTH')
})
it('consumes stream entries from a consumer group', async () => {
const mq = createBillingMq({
call: vi.fn(async () => [[
DEFAULT_BILLING_EVENTS_STREAM,
[[
'1740000000000-0',
[
'event_id',
'evt-1',
'event_type',
'flux.debited',
'aggregate_id',
'user-1',
'user_id',
'user-1',
'request_id',
'req-1',
'occurred_at',
'2026-03-24T00:00:00.000Z',
'schema_version',
'1',
'payload',
JSON.stringify({
amount: 5,
balanceAfter: 95,
source: 'llm',
description: 'gpt-5',
metadata: { promptTokens: 100, completionTokens: 200 },
}),
],
]],
]]),
})
await expect(mq.consume({
group: 'billing',
consumer: 'consumer-1',
count: 20,
blockMs: 100,
})).resolves.toEqual([{
streamMessageId: '1740000000000-0',
event: createEvent(),
}])
})
it('returns an empty array when no messages are available', async () => {
const mq = createBillingMq({
call: vi.fn(async () => null),
})
await expect(mq.consume({
group: 'billing',
consumer: 'consumer-1',
})).resolves.toEqual([])
})
it('throws when xreadgroup returns an invalid payload', async () => {
const mq = createBillingMq({
call: vi.fn(async () => ['not-an-array-entry']),
})
await expect(mq.consume({
group: 'billing',
consumer: 'consumer-1',
})).rejects.toThrow('Redis XREADGROUP returned an invalid stream payload')
})
it('claims idle pending messages', async () => {
const mq = createBillingMq({
call: vi.fn(async () => [
'1740000000001-0',
[[
'1740000000000-0',
[
'event_id',
'evt-1',
'event_type',
'flux.debited',
'aggregate_id',
'user-1',
'user_id',
'user-1',
'request_id',
'req-1',
'occurred_at',
'2026-03-24T00:00:00.000Z',
'schema_version',
'1',
'payload',
JSON.stringify({
amount: 5,
balanceAfter: 95,
source: 'llm',
description: 'gpt-5',
metadata: { promptTokens: 100, completionTokens: 200 },
}),
],
]],
[],
]),
})
await expect(mq.claimIdleMessages({
group: 'billing',
consumer: 'consumer-1',
minIdleTimeMs: 30_000,
})).resolves.toEqual([{
streamMessageId: '1740000000000-0',
event: createEvent(),
}])
})
it('throws when xautoclaim returns an invalid payload', async () => {
const mq = createBillingMq({
call: vi.fn(async () => ['1740000000001-0']),
})
await expect(mq.claimIdleMessages({
group: 'billing',
consumer: 'consumer-1',
minIdleTimeMs: 30_000,
})).rejects.toThrow('Redis XAUTOCLAIM returned an invalid response')
})
it('acks one or more stream messages', async () => {
const redis = {
call: vi.fn(async () => 2),
}
const mq = createBillingMq(redis)
await expect(mq.ack('billing', ['1-0', '2-0'])).resolves.toBe(2)
expect(redis.call).toHaveBeenCalledWith('XACK', DEFAULT_BILLING_EVENTS_STREAM, 'billing', '1-0', '2-0')
})
it('returns zero when ack receives an empty message id list', async () => {
const redis = {
call: vi.fn(),
}
const mq = createBillingMq(redis)
await expect(mq.ack('billing', [])).resolves.toBe(0)
expect(redis.call).not.toHaveBeenCalled()
})
it('throws when ack does not return a number', async () => {
const mq = createBillingMq({
call: vi.fn(async () => '2'),
})
await expect(mq.ack('billing', '1-0')).rejects.toThrow('Redis XACK did not return an acknowledgement count')
})
})
@@ -1,15 +1,13 @@
import type Redis from 'ioredis'
import type { Database } from '../../../libs/db'
import type { MqService } from '../../../libs/mq'
import type { createConfigKVService } from '../../config-kv'
import type { BillingEvent } from '../billing-events'
import { eq } from 'drizzle-orm'
import { and, eq } from 'drizzle-orm'
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
import { mockDB } from '../../../libs/mock-db'
import { DEFAULT_BILLING_EVENTS_STREAM, userFluxRedisKey } from '../../../utils/redis-keys'
import { userFluxRedisKey } from '../../../utils/redis-keys'
import { createBillingService } from '../billing-service'
import * as schema from '../../../schemas'
@@ -35,21 +33,9 @@ function createMockRedis(): Redis {
} as unknown as Redis
}
function createMockBillingMq(): MqService<BillingEvent> {
return {
stream: DEFAULT_BILLING_EVENTS_STREAM,
publish: vi.fn(async () => '1-0'),
ensureConsumerGroup: vi.fn(async () => true),
consume: vi.fn(async () => []),
claimIdleMessages: vi.fn(async () => []),
ack: vi.fn(async () => 1),
} as any
}
describe('billingService', () => {
let db: Database
let redis: Redis
let billingMq: MqService<BillingEvent>
let billingService: ReturnType<typeof createBillingService>
beforeAll(async () => {
@@ -64,8 +50,7 @@ describe('billingService', () => {
beforeEach(async () => {
redis = createMockRedis()
billingMq = createMockBillingMq()
billingService = createBillingService(db, redis, billingMq, createMockConfigKV())
billingService = createBillingService(db, redis, createMockConfigKV())
await db.delete(schema.fluxTransaction)
await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
@@ -114,11 +99,6 @@ describe('billingService', () => {
source: 'stripe.checkout.completed',
})
// Verify billing events published to stream
expect(billingMq.publish).toHaveBeenCalledTimes(2)
expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({ eventType: 'flux.credited' }))
expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({ eventType: 'stripe.checkout.completed' }))
// Verify stripe session marked as credited
const [sessionRecord] = await db.select().from(schema.stripeCheckoutSession).where(eq(schema.stripeCheckoutSession.stripeSessionId, 'sess-billing-1'))
expect(sessionRecord?.fluxCredited).toBe(true)
@@ -148,13 +128,14 @@ describe('billingService', () => {
expect(second).toEqual({ applied: false })
// Only 2 publish calls from the first invocation
expect(billingMq.publish).toHaveBeenCalledTimes(2)
// Idempotent replay must not double-write the ledger
const txRecords = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, 'user-billing-1'))
expect(txRecords).toHaveLength(1)
})
})
describe('consumeFluxForLLM', () => {
it('deducts balance, publishes flux.debited event, updates Redis', async () => {
it('deducts balance, writes the ledger row inside the transaction, and refreshes Redis', async () => {
// Setup: give user some flux first
await db.insert(schema.userFlux).values({ userId: 'user-billing-1', flux: 100 })
@@ -173,18 +154,25 @@ 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 (transaction written by consumer)
expect(billingMq.publish).toHaveBeenCalledTimes(1)
expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({
eventType: 'flux.debited',
// Ledger row written inline (no async consumer involved post-refactor)
const [txRecord] = await db.select().from(schema.fluxTransaction).where(and(
eq(schema.fluxTransaction.userId, 'user-billing-1'),
eq(schema.fluxTransaction.requestId, 'req-1'),
))
expect(txRecord).toMatchObject({
userId: 'user-billing-1',
payload: expect.objectContaining({
amount: 30,
balanceAfter: 70,
description: 'gpt-4',
metadata: { promptTokens: 120, completionTokens: 80 },
}),
}))
type: 'debit',
amount: 30,
balanceBefore: 100,
balanceAfter: 70,
requestId: 'req-1',
description: 'gpt-4',
})
expect(txRecord?.metadata).toMatchObject({
promptTokens: 120,
completionTokens: 80,
source: 'llm.request',
})
// Verify Redis cache updated
expect(redis.set).toHaveBeenCalledWith(userFluxRedisKey('user-billing-1'), '70')
@@ -204,14 +192,11 @@ describe('billingService', () => {
const txRecords = await db.select().from(schema.fluxTransaction)
expect(txRecords).toHaveLength(0)
// Verify no event was published
expect(billingMq.publish).not.toHaveBeenCalled()
})
})
describe('creditFlux', () => {
it('credits balance with transaction + outbox', async () => {
it('credits balance and writes the ledger row in one transaction', async () => {
const result = await billingService.creditFlux({
userId: 'user-billing-1',
amount: 50,
@@ -221,6 +206,7 @@ describe('billingService', () => {
expect(result.balanceAfter).toBe(50)
expect(result.balanceBefore).toBe(0)
expect(result.idempotent).toBe(false)
// Verify transaction
const txRecords = await db.select().from(schema.fluxTransaction).where(eq(schema.fluxTransaction.userId, 'user-billing-1'))
@@ -231,10 +217,62 @@ describe('billingService', () => {
balanceBefore: 0,
balanceAfter: 50,
})
})
// Verify billing event published to stream
expect(billingMq.publish).toHaveBeenCalledTimes(1)
expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({ eventType: 'flux.credited' }))
it('is idempotent across retries with the same requestId', async () => {
// ROOT CAUSE:
//
// Worker crash window: creditFlux commits the credit, then the
// grant-batch poller crashes before marking its own state row
// (e.g. flux_grant_batch_recipient) as granted. On restart the poller
// re-claims the same row and calls creditFlux again with the same
// requestId.
//
// Before the fix: second call hit the unique index on
// (user_id, request_id) and threw, the poller's catch block marked
// the recipient as `failed` despite the user already having been credited.
// User got the FLUX but the recipient row was stuck in failed.
//
// After the fix: second call detects the existing flux_transaction row,
// returns it as an idempotent success without touching balance or cache.
// Poller advances to granted normally.
const requestId = 'campaign-replay-test'
const first = await billingService.creditFlux({
userId: 'user-billing-1',
amount: 100,
requestId,
description: 'Replay test',
source: 'admin',
})
expect(first.idempotent).toBe(false)
expect(first.balanceAfter).toBe(100)
// Second call with same requestId — simulates crash-recovery retry.
const second = await billingService.creditFlux({
userId: 'user-billing-1',
amount: 100,
requestId,
description: 'Replay test',
source: 'admin',
})
expect(second.idempotent).toBe(true)
// Same record returned, not a fresh credit
expect(second.fluxTransactionId).toBe(first.fluxTransactionId)
expect(second.balanceAfter).toBe(first.balanceAfter)
// Balance must NOT have doubled
const [fluxRow] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
expect(fluxRow!.flux).toBe(100)
// Only one ledger row exists (unique index would prevent a second anyway,
// but verify the function didn't try to insert and silently swallow)
const txRecords = await db.select().from(schema.fluxTransaction).where(and(
eq(schema.fluxTransaction.userId, 'user-billing-1'),
eq(schema.fluxTransaction.requestId, requestId),
))
expect(txRecords).toHaveLength(1)
})
})
})
@@ -1,23 +0,0 @@
import type { Database } from '../libs/db'
import * as schema from '../schemas/llm-request-log'
export interface RequestLogEntry {
userId: string
model: string
status: number
durationMs: number
fluxConsumed: number
promptTokens?: number
completionTokens?: number
}
export function createLLMRequestLogService(db: Database) {
return {
async logRequest(entry: RequestLogEntry) {
await db.insert(schema.llmRequestLog).values(entry)
},
}
}
export type RequestLogService = ReturnType<typeof createLLMRequestLogService>
-2
View File
@@ -15,8 +15,6 @@ export function createRedisKey(...parts: RedisKeyPart[]): string {
return parts.map(normalizeRedisKeyPart).join(':')
}
export const DEFAULT_BILLING_EVENTS_STREAM = createRedisKey('billing', 'events')
export function configRedisKey(key: string): string {
return createRedisKey('config', key)
}
@@ -3,7 +3,6 @@ import { describe, expect, it } from 'vitest'
import {
configRedisKey,
createRedisKey,
DEFAULT_BILLING_EVENTS_STREAM,
lockRedisKey,
userChatBroadcastRedisKey,
userFluxRedisKey,
@@ -21,7 +20,6 @@ describe('redis key utils', () => {
})
it('exposes stable helpers for config, user, and lock namespaces', () => {
expect(DEFAULT_BILLING_EVENTS_STREAM).toBe('billing:events')
expect(configRedisKey('FLUX_PER_REQUEST')).toBe('config:FLUX_PER_REQUEST')
expect(userFluxRedisKey('user-1')).toBe('user:user-1:flux')
expect(userChatBroadcastRedisKey('user-1')).toBe('user:user-1:chat:broadcast')