diff --git a/apps/server/docker-compose.yml b/apps/server/docker-compose.yml index 884374a2f..0435bd6fa 100644 --- a/apps/server/docker-compose.yml +++ b/apps/server/docker-compose.yml @@ -53,28 +53,11 @@ services: timeout: 5s retries: 5 - outbox-dispatcher: + billing-consumer: build: context: ../.. dockerfile: apps/server/Dockerfile - command: ['pnpm', '-F', '@proj-airi/server', 'run', 'server', 'outbox-dispatcher'] - depends_on: - db: - condition: service_healthy - redis: - condition: service_healthy - env_file: - - path: .env - required: false - - path: .env.local - required: false - restart: unless-stopped - - cache-sync-consumer: - build: - context: ../.. - dockerfile: apps/server/Dockerfile - command: ['pnpm', '-F', '@proj-airi/server', 'run', 'server', 'cache-sync-consumer'] + command: ['pnpm', '-F', '@proj-airi/server', 'run', 'server', 'billing-consumer'] depends_on: db: condition: service_healthy diff --git a/apps/server/docs/ai-context/architecture-overview.md b/apps/server/docs/ai-context/architecture-overview.md index fe8c088b6..1092fc715 100644 --- a/apps/server/docs/ai-context/architecture-overview.md +++ b/apps/server/docs/ai-context/architecture-overview.md @@ -27,11 +27,10 @@ - 注入 WebSocket - 绑定 `uncaughtException` / `unhandledRejection` -CLI 入口在 `src/bin/run.ts`,支持三种角色: +CLI 入口在 `src/bin/run.ts`,支持两种角色: - `api` -- `cache-sync-consumer` -- `outbox-dispatcher` +- `billing-consumer` ## 依赖注入结构 @@ -44,7 +43,6 @@ CLI 入口在 `src/bin/run.ts`,支持三种角色: - `redis` - `configKV` - 服务 - - `outboxService` - `auth` - `characterService` - `providerService` @@ -57,7 +55,7 @@ CLI 入口在 `src/bin/run.ts`,支持三种角色: 这个装配顺序说明了几个事实: -- `billingService` 依赖 `db + redis + outboxService` +- `billingService` 依赖 `db + redis` - `fluxService` 只读余额,不承担余额写入职责 - `auth` 直接绑定数据库 schema,不是外部独立服务 @@ -82,7 +80,6 @@ CLI 入口在 `src/bin/run.ts`,支持三种角色: - `flux.ts` - `billing-service.ts` - `stripe.ts` -- `outbox-service.ts` 这里是主要改动面。大多数业务改动都不应该直接写进 route handler。 @@ -92,7 +89,7 @@ CLI 入口在 `src/bin/run.ts`,支持三种角色: - Drizzle schema 基本覆盖了所有核心表 - 数据迁移由 `@proj-airi/server-schema` 提供 -- `app.ts` 和 `run-outbox-dispatcher.ts` 启动时都会执行迁移 +- `app.ts` 启动时会执行迁移 ## 中间件与通用约束 @@ -134,7 +131,7 @@ CLI 入口在 `src/bin/run.ts`,支持三种角色: - 新用户首次读取时初始化余额 - `BillingService` - 面向写入 - - 事务内更新余额、流水、审计、outbox + - debitFlux:事务内更新余额,事务后 XADD Redis Stream;credit 方法:事务内同步写流水和审计 这是服务端最重要的边界之一,尽量不要把写余额逻辑重新塞回 `flux.ts`。 diff --git a/apps/server/docs/ai-context/billing-architecture.md b/apps/server/docs/ai-context/billing-architecture.md index b9e79095e..e57031b9f 100644 --- a/apps/server/docs/ai-context/billing-architecture.md +++ b/apps/server/docs/ai-context/billing-architecture.md @@ -2,31 +2,34 @@ ## 架构概述 -`apps/server` 的计费链采用 **Postgres 作为唯一账本真相源**,Redis 仅作缓存。所有余额变化在 DB 事务内原子完成,同步写入 `flux_ledger`(流水)和 `outbox_events`(事件),通过 Redis Streams 分发给下游 consumer。 +`apps/server` 的计费链采用 **Postgres 作为唯一账本真相源**,Redis 仅作缓存。余额变化路径分两类:`debitFlux` 在 DB 事务内只做 `UPDATE user_flux`,ledger/audit/请求日志通过 Redis Stream 异步写入;credit 方法仍在事务内同步写入 ledger 和 audit。 ### 数据模型 - **`user_flux`** — 用户余额快照(单行/用户) - **`flux_ledger`** — append-only 账务流水(type: credit/debit/initial, amount, balanceBefore, balanceAfter, requestId) - 含 partial unique index `(userId, requestId) WHERE requestId IS NOT NULL`,DB 层幂等防重 -- **`outbox_events`** — 事件暂存,claim-lease 模式分发 - **`flux_audit_log`** — 用户可见的历史记录 -### 同步链路(已实现) +### debitFlux 链路(已实现) -每次余额变化的 DB 事务内: +DB 事务内仅做: 1. `SELECT user_flux FOR UPDATE` 锁行 -2. 更新 `user_flux.flux` -3. 写 `flux_ledger` -4. 写 `flux_audit_log` -5. 写 `outbox_events` -6. 事务提交后 best-effort 更新 Redis 缓存 +2. 检查余额(不足返回 402) +3. 更新 `user_flux.flux` +4. 事务提交后 XADD Redis Stream(`billing-events`),携带扣费金额、余额快照、requestId 等 +5. 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存 + +ledger / audit / llm_request_log 的写入均由 **billing-consumer** 异步完成。 + +### credit 方法链路(已实现) + +credit 方法(`creditFlux` / `creditFluxFromStripeCheckout` / `creditFluxFromInvoice`)仍在 DB 事务内同步写入 `flux_ledger` 和 `flux_audit_log`。 ### 异步链路(已实现) -- **outbox-dispatcher** — 轮询 `outbox_events`,发布到 Redis Stream `billing-events` -- **cache-sync-consumer** — 消费 Stream 事件,同步 Redis 缓存(处理 `flux.debited` 和 `flux.credited`) +- **billing-consumer** — 消费 Redis Stream `billing-events`,将 ledger、audit log、LLM 请求日志异步写入 DB ### 事件模型 @@ -44,8 +47,7 @@ Stream: `billing-events` 通过 `src/bin/run.ts` 分角色启动: - `api` — HTTP 服务 -- `outbox-dispatcher` — outbox → Redis Stream -- `cache-sync-consumer` — Redis 缓存同步(处理 `flux.debited` + `flux.credited`) +- `billing-consumer` — 消费 Redis Stream,异步写入 ledger、audit log、LLM 请求日志到 DB ## 关键服务 @@ -53,7 +55,7 @@ Stream: `billing-events` 所有余额写操作的唯一入口: -- **`debitFlux()`** — 扣费(LLM 请求),事务内:锁行 → 检余额(402) → 更新余额 → ledger → audit → outbox(`flux.debited`) +- **`debitFlux()`** — 扣费(LLM 请求),事务内:锁行 → 检余额(402) → 更新余额;事务提交后 XADD `flux.debited` 到 Redis Stream,ledger/audit 由 billing-consumer 异步写入 - **`creditFlux()`** — 通用充值 - **`creditFluxFromStripeCheckout()`** — Stripe 一次性支付充值,幂等(`fluxCredited` 标志) - **`creditFluxFromInvoice()`** — Stripe 订阅发票充值,幂等 @@ -79,20 +81,19 @@ Redis **不是**余额真相源,仅用于: | Phase | 状态 | 关键点 | |-------|------|--------| | 1. DB-first 账本 | ✅ 已完成 | `flux_ledger` 表,`SELECT FOR UPDATE` 原子扣减,Redis 降为缓存 | -| 2. Outbox 事件 | ✅ 已完成 | 所有余额变化产生 outbox 事件,debit + credit 均覆盖 | -| 3. Redis Streams | ✅ 已完成 | MQ、dispatcher、worker 全部就位 | -| 4. Stripe 幂等 | ✅ 已完成 | checkout + invoice 事务内幂等检查 | -| 5. LLM 计费优化 | ⚠️ 部分 | 已有 `requestId` 和 DB 事务扣费,待加 tiktoken fallback | -| 6. 部署拆分 | ✅ 已完成 | `bin/run.ts` 三角色启动(api / outbox-dispatcher / cache-sync-consumer) | -| 7. 幂等防重 | ✅ 已完成 | `flux_ledger` partial unique index on `(userId, requestId)` | -| 8. Cache-sync 适配 | ✅ 已完成 | 同时处理 `flux.debited` 和 `flux.credited` 事件 | +| 2. Redis Streams 异步写入 | ✅ 已完成 | debitFlux 事务后 XADD,billing-consumer 异步写 ledger/audit/请求日志 | +| 3. Stripe 幂等 | ✅ 已完成 | checkout + invoice 事务内幂等检查 | +| 4. LLM 计费优化 | ⚠️ 部分 | 已有 `requestId` 和 DB 事务扣费,待加 tiktoken fallback | +| 5. 部署拆分 | ✅ 已完成 | `bin/run.ts` 两角色启动(api / billing-consumer) | +| 6. 幂等防重 | ✅ 已完成 | `flux_ledger` partial unique index on `(userId, requestId)` | ### 已删除 - `flux-write-back.ts` — 定时回写补偿机制,不再需要 - `FluxService.consumeFlux()` / `addFlux()` — 写操作已移至 BillingService - `llm_request_log.settled` — 无消费者,已移除 -- `billing-consumer` 进程角色 — 空壳(仅 log),已移除;需要账务分析时重新添加 +- `outbox_events` 表及 outbox-dispatcher 进程 — 已移除,统一由 billing-consumer 处理异步写入 +- `cache-sync-consumer` 进程角色 — 已合并进 billing-consumer ## 剩余 TODO diff --git a/apps/server/docs/ai-context/data-model-and-state.md b/apps/server/docs/ai-context/data-model-and-state.md index 2c7b858ed..419c747b6 100644 --- a/apps/server/docs/ai-context/data-model-and-state.md +++ b/apps/server/docs/ai-context/data-model-and-state.md @@ -10,7 +10,6 @@ - Flux 余额与账本 - Stripe 业务镜像 - LLM 请求日志 - - outbox 事件 - `Redis` - Flux 余额缓存 - 服务配置 KV @@ -150,19 +149,6 @@ - 只做追加写入 - 明确不加 user 外键,以避免高并发写入的额外约束成本 -### Outbox - -- `outbox_events` - -来源文件: - -- `src/schemas/outbox-events.ts` - -说明: - -- 本质上是 DB 内事件暂存区 -- 通过 `claimedBy + claimExpiresAt + publishedAt` 实现 lease/claim 分发 - ## 服务与状态写入边界 ### `createFluxService()` @@ -177,7 +163,7 @@ - 扣费 - 充值 -- ledger / audit / outbox 写入 +- ledger / audit 写入 ### `createBillingService()` @@ -185,8 +171,9 @@ - 所有余额写操作 - DB 事务 -- ledger / audit / outbox 联动 -- 事务完成后 best-effort 更新 Redis +- debitFlux:事务内仅更新余额;事务后 XADD Redis Stream,ledger/audit 由 billing-consumer 异步写入 +- credit 方法:事务内同步写 ledger / audit +- 事务提交后 best-effort `redis.set` 更新 Flux 余额缓存 这是所有 Flux 写路径应收敛到的中心。 @@ -212,8 +199,7 @@ 写入来源: - `fluxService.getFlux()` cache miss 后回填 -- `billingService` 余额事务成功后 best-effort 更新 -- `cache-sync-consumer` 消费 `flux.debited` / `flux.credited` 后同步 +- `billingService` 余额事务提交后 best-effort `redis.set` 直接更新(API 进程内同步) ### 配置 KV @@ -241,8 +227,8 @@ 1. `SELECT user_flux FOR UPDATE` 2. 计算新余额 -3. 写余额 -4. 写 ledger / audit / outbox +3. 写余额(debitFlux 事务内仅此一步;credit 方法同步写 ledger / audit) +4. 事务提交后 XADD Redis Stream(debitFlux)或直接返回(credit) 这保证同一用户余额更新是串行化的。 @@ -254,15 +240,6 @@ - `stripe_invoice.fluxCredited` - `flux_ledger(userId, requestId)` 唯一约束 -### outbox 并发 - -`outbox-service.ts` 使用: - -- `FOR UPDATE SKIP LOCKED` -- `claimExpiresAt` - -这允许多个 dispatcher 并行拉取待发布事件。 - ## 现有代码中的结构信号 - `request-log.ts` 与 `llm-request-log.ts` 完全重叠,后者更像旧名残留。 diff --git a/apps/server/docs/ai-context/workers-and-runtime.md b/apps/server/docs/ai-context/workers-and-runtime.md index 04c1433dc..cc0b2fbf3 100644 --- a/apps/server/docs/ai-context/workers-and-runtime.md +++ b/apps/server/docs/ai-context/workers-and-runtime.md @@ -6,12 +6,10 @@ - `api` - 启动 Hono HTTP + WebSocket 服务 -- `cache-sync-consumer` - - 消费 Redis Streams 中的计费事件,回写 Flux Redis 缓存 -- `outbox-dispatcher` - - 从 Postgres `outbox_events` 拉取未发布事件,投递到 Redis Streams +- `billing-consumer` + - 消费 Redis Stream `billing-events`,异步将 ledger、audit log、LLM 请求日志写入 DB -这三个角色已经是当前服务端部署拆分的基本单位。 +这两个角色是当前服务端部署拆分的基本单位。 ## API 角色 @@ -32,61 +30,21 @@ - 启动 HTTP server - 注入 WebSocket -## Outbox Dispatcher +## Billing Consumer 实现位置: -- 入口:`src/bin/run-outbox-dispatcher.ts` -- 服务:`src/services/outbox-dispatcher.ts` -- 存储:`src/services/outbox-service.ts` -- MQ:`src/services/billing-mq.ts` - -工作流程: - -1. 从 `outbox_events` claim 一批未发布事件 -2. 逐条发布到 Redis Stream -3. 发布成功后写 `publishedAt` 和 `streamMessageId` -4. 失败则释放 claim,等待下一轮处理 - -关键机制: - -- 支持多实例并发 dispatcher -- claim 通过 TTL 失效,避免 worker 崩掉后永久锁死 - -相关环境变量: - -- `OUTBOX_DISPATCHER_NAME` -- `OUTBOX_DISPATCHER_BATCH_SIZE` -- `OUTBOX_DISPATCHER_CLAIM_TTL_MS` -- `OUTBOX_DISPATCHER_POLL_MS` -- `BILLING_EVENTS_STREAM` - -## Billing Events Consumer - -实现位置: - -- 入口:`src/bin/run-billing-events-consumer.ts` +- 入口:`src/bin/run-billing-consumer.ts` - worker:`src/services/billing-mq-worker.ts` - stream adapter:`src/services/billing-mq.ts` -当前默认 handler: +工作流程: -- `handleCacheSyncMessage()` - -它只处理: - -- `flux.credited` -- `flux.debited` - -并把 `payload.balanceAfter` 写回 Redis: - -- key: `flux:` - -这说明当前 consumer 的目标非常克制: - -- 不是账务真相处理器 -- 不是分析流水处理器 -- 只是缓存一致性补偿器 +1. 以 consumer group 模式消费 Redis Stream `billing-events` +2. 根据事件类型分发处理: + - `flux.debited` — 写 `flux_ledger` 和 `flux_audit_log` + - `llm.request.log` — 写 `llm_request_log` +3. 处理成功后 ACK;handler 抛错时不 ACK,消息保持 pending 等待重试 相关环境变量: @@ -172,17 +130,13 @@ - `STRIPE_SECRET_KEY` - `STRIPE_WEBHOOK_SECRET` -### Billing MQ / Outbox +### Billing MQ - `BILLING_EVENTS_STREAM` - `BILLING_EVENTS_CONSUMER_NAME` - `BILLING_EVENTS_BATCH_SIZE` - `BILLING_EVENTS_BLOCK_MS` - `BILLING_EVENTS_MIN_IDLE_MS` -- `OUTBOX_DISPATCHER_NAME` -- `OUTBOX_DISPATCHER_BATCH_SIZE` -- `OUTBOX_DISPATCHER_CLAIM_TTL_MS` -- `OUTBOX_DISPATCHER_POLL_MS` ### OTel @@ -200,7 +154,7 @@ - 新增 worker - 先看 `run.ts` 的角色模型和 `billing-mq-worker.ts` - 改事件分发 - - 先看 outbox,而不是直接在业务事务里调用 Redis Streams + - 先看 billing-consumer handler,在 `billing-mq-worker.ts` 中增加新的事件处理分支 - 改聊天同步 - 先区分“持久化消息”与“广播通知”两层 - 改部署限流 diff --git a/apps/server/drizzle/0005_tearful_kronos.sql b/apps/server/drizzle/0005_tough_living_tribunal.sql similarity index 61% rename from apps/server/drizzle/0005_tearful_kronos.sql rename to apps/server/drizzle/0005_tough_living_tribunal.sql index 6b0e72155..efe52f297 100644 --- a/apps/server/drizzle/0005_tearful_kronos.sql +++ b/apps/server/drizzle/0005_tough_living_tribunal.sql @@ -10,25 +10,6 @@ CREATE TABLE "flux_ledger" ( "created_at" timestamp DEFAULT now() NOT NULL ); --> statement-breakpoint -CREATE TABLE "outbox_events" ( - "id" text PRIMARY KEY NOT NULL, - "event_id" text NOT NULL, - "event_type" text NOT NULL, - "aggregate_id" text NOT NULL, - "user_id" text NOT NULL, - "request_id" text, - "schema_version" integer NOT NULL, - "payload" text NOT NULL, - "occurred_at" timestamp NOT NULL, - "available_at" timestamp DEFAULT now() NOT NULL, - "claimed_by" text, - "claim_expires_at" timestamp, - "published_at" timestamp, - "stream_message_id" text, - "created_at" timestamp DEFAULT now() NOT NULL, - "updated_at" timestamp DEFAULT now() NOT NULL -); ---> statement-breakpoint ALTER TABLE "llm_request_log" DROP CONSTRAINT "llm_request_log_user_id_user_id_fk"; --> statement-breakpoint ALTER TABLE "messages" ALTER COLUMN "sender_id" DROP NOT NULL;--> statement-breakpoint @@ -40,7 +21,4 @@ ALTER TABLE "flux_ledger" ADD CONSTRAINT "flux_ledger_user_id_user_id_fk" FOREIG CREATE INDEX "flux_ledger_user_id_idx" ON "flux_ledger" USING btree ("user_id");--> statement-breakpoint CREATE INDEX "flux_ledger_created_at_idx" ON "flux_ledger" USING btree ("created_at");--> statement-breakpoint CREATE UNIQUE INDEX "flux_ledger_user_request_uniq" ON "flux_ledger" USING btree ("user_id","request_id") WHERE request_id IS NOT NULL;--> statement-breakpoint -CREATE UNIQUE INDEX "outbox_events_event_id_idx" ON "outbox_events" USING btree ("event_id");--> statement-breakpoint -CREATE INDEX "outbox_events_publish_scan_idx" ON "outbox_events" USING btree ("published_at","available_at","claim_expires_at","created_at");--> statement-breakpoint -CREATE INDEX "outbox_events_claimed_by_idx" ON "outbox_events" USING btree ("claimed_by");--> statement-breakpoint ALTER TABLE "llm_request_log" DROP COLUMN "settled"; \ No newline at end of file diff --git a/apps/server/drizzle/meta/0005_snapshot.json b/apps/server/drizzle/meta/0005_snapshot.json index aa35261e1..0f6dda602 100644 --- a/apps/server/drizzle/meta/0005_snapshot.json +++ b/apps/server/drizzle/meta/0005_snapshot.json @@ -1,5 +1,5 @@ { - "id": "22b43247-dc0b-4bde-8869-955460dbd3e2", + "id": "11d1fa8c-cf77-42ef-9776-3a74733be0ac", "prevId": "45bc0dad-65f5-4695-9115-3d7d376b6440", "version": "7", "dialect": "postgresql", @@ -1529,182 +1529,6 @@ "checkConstraints": {}, "isRLSEnabled": false }, - "public.outbox_events": { - "name": "outbox_events", - "schema": "", - "columns": { - "id": { - "name": "id", - "type": "text", - "primaryKey": true, - "notNull": true - }, - "event_id": { - "name": "event_id", - "type": "text", - "primaryKey": false, - "notNull": true - }, - "event_type": { - "name": "event_type", - "type": "text", - "primaryKey": false, - "notNull": true - }, - "aggregate_id": { - "name": "aggregate_id", - "type": "text", - "primaryKey": false, - "notNull": true - }, - "user_id": { - "name": "user_id", - "type": "text", - "primaryKey": false, - "notNull": true - }, - "request_id": { - "name": "request_id", - "type": "text", - "primaryKey": false, - "notNull": false - }, - "schema_version": { - "name": "schema_version", - "type": "integer", - "primaryKey": false, - "notNull": true - }, - "payload": { - "name": "payload", - "type": "text", - "primaryKey": false, - "notNull": true - }, - "occurred_at": { - "name": "occurred_at", - "type": "timestamp", - "primaryKey": false, - "notNull": true - }, - "available_at": { - "name": "available_at", - "type": "timestamp", - "primaryKey": false, - "notNull": true, - "default": "now()" - }, - "claimed_by": { - "name": "claimed_by", - "type": "text", - "primaryKey": false, - "notNull": false - }, - "claim_expires_at": { - "name": "claim_expires_at", - "type": "timestamp", - "primaryKey": false, - "notNull": false - }, - "published_at": { - "name": "published_at", - "type": "timestamp", - "primaryKey": false, - "notNull": false - }, - "stream_message_id": { - "name": "stream_message_id", - "type": "text", - "primaryKey": false, - "notNull": false - }, - "created_at": { - "name": "created_at", - "type": "timestamp", - "primaryKey": false, - "notNull": true, - "default": "now()" - }, - "updated_at": { - "name": "updated_at", - "type": "timestamp", - "primaryKey": false, - "notNull": true, - "default": "now()" - } - }, - "indexes": { - "outbox_events_event_id_idx": { - "name": "outbox_events_event_id_idx", - "columns": [ - { - "expression": "event_id", - "isExpression": false, - "asc": true, - "nulls": "last" - } - ], - "isUnique": true, - "concurrently": false, - "method": "btree", - "with": {} - }, - "outbox_events_publish_scan_idx": { - "name": "outbox_events_publish_scan_idx", - "columns": [ - { - "expression": "published_at", - "isExpression": false, - "asc": true, - "nulls": "last" - }, - { - "expression": "available_at", - "isExpression": false, - "asc": true, - "nulls": "last" - }, - { - "expression": "claim_expires_at", - "isExpression": false, - "asc": true, - "nulls": "last" - }, - { - "expression": "created_at", - "isExpression": false, - "asc": true, - "nulls": "last" - } - ], - "isUnique": false, - "concurrently": false, - "method": "btree", - "with": {} - }, - "outbox_events_claimed_by_idx": { - "name": "outbox_events_claimed_by_idx", - "columns": [ - { - "expression": "claimed_by", - "isExpression": false, - "asc": true, - "nulls": "last" - } - ], - "isUnique": false, - "concurrently": false, - "method": "btree", - "with": {} - } - }, - "foreignKeys": {}, - "compositePrimaryKeys": {}, - "uniqueConstraints": {}, - "policies": {}, - "checkConstraints": {}, - "isRLSEnabled": false - }, "public.system_provider_configs": { "name": "system_provider_configs", "schema": "", diff --git a/apps/server/drizzle/meta/_journal.json b/apps/server/drizzle/meta/_journal.json index 46c361540..d776cf3f8 100644 --- a/apps/server/drizzle/meta/_journal.json +++ b/apps/server/drizzle/meta/_journal.json @@ -40,8 +40,8 @@ { "idx": 5, "version": "7", - "when": 1774552566896, - "tag": "0005_tearful_kronos", + "when": 1774582846222, + "tag": "0005_tough_living_tribunal", "breakpoints": true } ] diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index f18f3fb7e..2bddd424e 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -2,6 +2,15 @@ import type Redis from 'ioredis' import type { Env } from './libs/env' import type { OtelInstance } from './libs/otel' +import type { BillingMqService } from './services/billing-mq' +import type { BillingService } from './services/billing-service' +import type { CharacterService } from './services/characters' +import type { ChatService } from './services/chats' +import type { ConfigKVService } from './services/config-kv' +import type { FluxService } from './services/flux' +import type { FluxAuditService } from './services/flux-audit' +import type { ProviderService } from './services/providers' +import type { StripeService } from './services/stripe' import type { HonoEnv } from './types/hono' import process from 'node:process' @@ -31,61 +40,36 @@ import { createFluxRoutes } from './routes/flux' import { createProviderRoutes } from './routes/providers' import { createStripeRoutes } from './routes/stripe' import { createV1CompletionsRoutes } from './routes/v1completions' +import { createBillingMqService } from './services/billing-mq' import { createBillingService } from './services/billing-service' import { createCharacterService } from './services/characters' import { createChatService } from './services/chats' import { createConfigKVService } from './services/config-kv' import { createFluxService } from './services/flux' import { createFluxAuditService } from './services/flux-audit' -import { createOutboxService } from './services/outbox-service' import { createProviderService } from './services/providers' import { createRequestLogService } from './services/request-log' import { createStripeService } from './services/stripe' import { ApiError, createInternalError, createUnauthorizedError } from './utils/error' import { getTrustedOrigin } from './utils/origin' -type AuthService = ReturnType -type CharacterService = ReturnType -type ChatService = ReturnType -type ProviderService = ReturnType -type FluxService = ReturnType -type ConfigKVService = ReturnType -type RequestLogService = ReturnType -type StripeDBService = ReturnType -type FluxAuditService = ReturnType -type BillingService = ReturnType - interface AppDeps { - auth: AuthService + auth: ReturnType characterService: CharacterService chatService: ChatService providerService: ProviderService fluxService: FluxService fluxAuditService: FluxAuditService - requestLogService: RequestLogService - stripeService: StripeDBService + stripeService: StripeService billingService: BillingService + billingMqService: BillingMqService configKV: ConfigKVService redis: Redis env: Env otel: OtelInstance | null } -function buildApp({ - auth, - characterService, - chatService, - providerService, - fluxService, - fluxAuditService, - requestLogService, - stripeService, - billingService, - configKV, - redis, - env, - otel, -}: AppDeps) { +function buildApp(deps: AppDeps) { const logger = useLogger('app').useGlobalConfig() const app = new Hono() @@ -98,20 +82,20 @@ function buildApp({ ) .use(honoLogger()) - if (otel) { - app.use('*', otelMiddleware(otel.http)) + if (deps.otel) { + app.use('*', otelMiddleware(deps.otel.http)) } // WebSocket setup — must be registered BEFORE bodyLimit middleware const { injectWebSocket, upgradeWebSocket } = createNodeWebSocket({ app }) - const chatWsSetup = createChatWsHandlers(chatService, redis, otel?.engagement ?? null) + const chatWsSetup = createChatWsHandlers(deps.chatService, deps.redis, deps.otel?.engagement ?? null) app.get('/ws/chat', upgradeWebSocket(async (c) => { const token = c.req.query('token') if (!token) { throw createUnauthorizedError('Missing token') } - const session = await auth.api.getSession({ + const session = await deps.auth.api.getSession({ headers: new Headers({ Authorization: `Bearer ${token}` }), }) if (!session?.user) { @@ -121,7 +105,7 @@ function buildApp({ })) const builtApp = app - .use('*', sessionMiddleware(auth)) + .use('*', sessionMiddleware(deps.auth)) .use('*', bodyLimit({ maxSize: 1024 * 1024 })) .onError((err, c) => { if (err instanceof ApiError) { @@ -157,37 +141,37 @@ function buildApp({ windowSec: 60, keyGenerator: c => c.req.header('x-forwarded-for') ?? c.req.header('x-real-ip') ?? 'unknown', })) - .on(['POST', 'GET'], '/api/auth/*', c => auth.handler(c.req.raw)) + .on(['POST', 'GET'], '/api/auth/*', c => deps.auth.handler(c.req.raw)) /** * Character routes are handled by the character service. */ - .route('/api/characters', createCharacterRoutes(characterService)) + .route('/api/characters', createCharacterRoutes(deps.characterService)) /** * Provider routes are handled by the provider service. */ - .route('/api/providers', createProviderRoutes(providerService)) + .route('/api/providers', createProviderRoutes(deps.providerService)) /** * Chat routes are handled by the chat service. */ - .route('/api/chats', createChatRoutes(chatService)) + .route('/api/chats', createChatRoutes(deps.chatService)) /** * V1 routes for official provider. */ - .route('/api/v1', createV1CompletionsRoutes(fluxService, billingService, configKV, requestLogService, otel?.llm ?? null)) + .route('/api/v1', createV1CompletionsRoutes(deps.fluxService, deps.billingService, deps.configKV, deps.billingMqService, deps.otel?.llm)) /** * Flux routes. */ - .route('/api/flux', createFluxRoutes(fluxService, fluxAuditService)) + .route('/api/flux', createFluxRoutes(deps.fluxService, deps.fluxAuditService)) /** * Stripe routes. */ - .route('/api/stripe', createStripeRoutes(fluxService, stripeService, billingService, configKV, env, otel?.revenue)) + .route('/api/stripe', createStripeRoutes(deps.fluxService, deps.stripeService, deps.billingService, deps.configKV, deps.env, deps.otel?.revenue)) return { app: builtApp, injectWebSocket } } @@ -272,9 +256,11 @@ export async function createApp() { build: ({ dependsOn }) => createConfigKVService(dependsOn.redis), }) - const outboxService = injeca.provide('services:outbox', { - dependsOn: { db }, - build: ({ dependsOn }) => createOutboxService(dependsOn.db), + const billingMqService = injeca.provide('services:billingMq', { + dependsOn: { redis, env: parsedEnv }, + build: ({ dependsOn }) => createBillingMqService(dependsOn.redis, { + stream: dependsOn.env.BILLING_EVENTS_STREAM, + }), }) const auth = injeca.provide('services:auth', { @@ -318,8 +304,8 @@ export async function createApp() { }) const billingService = injeca.provide('services:billing', { - dependsOn: { db, redis, outboxService, configKV, otel }, - build: ({ dependsOn }) => createBillingService(dependsOn.db, dependsOn.redis, dependsOn.outboxService, dependsOn.configKV, dependsOn.otel?.revenue), + dependsOn: { db, redis, billingMqService, configKV, otel }, + build: ({ dependsOn }) => createBillingService(dependsOn.db, dependsOn.redis, dependsOn.billingMqService, dependsOn.configKV, dependsOn.otel?.revenue), }) await injeca.start() @@ -334,6 +320,7 @@ export async function createApp() { requestLogService, stripeService, billingService, + billingMqService, configKV, redis, env: parsedEnv, @@ -346,9 +333,9 @@ export async function createApp() { providerService: resolved.providerService, fluxService: resolved.fluxService, fluxAuditService: resolved.fluxAuditService, - requestLogService: resolved.requestLogService, stripeService: resolved.stripeService, billingService: resolved.billingService, + billingMqService: resolved.billingMqService, configKV: resolved.configKV, redis: resolved.redis, env: resolved.env, diff --git a/apps/server/src/bin/__test__/run-billing-events-consumer.test.ts b/apps/server/src/bin/__test__/run-billing-events-consumer.test.ts deleted file mode 100644 index 60b86d310..000000000 --- a/apps/server/src/bin/__test__/run-billing-events-consumer.test.ts +++ /dev/null @@ -1,58 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' - -import { handleCacheSyncMessage } from '../run-billing-events-consumer' - -describe('handleCacheSyncMessage', () => { - it('updates the flux cache when a balance event includes balanceAfter', async () => { - const redis = { - set: vi.fn(async () => 'OK'), - } - - await handleCacheSyncMessage({ - streamMessageId: '1740000000000-0', - event: { - eventId: 'evt-1', - eventType: 'flux.credited', - aggregateId: 'user-1', - userId: 'user-1', - requestId: 'req-1', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - amount: 10, - balanceAfter: 110, - source: 'stripe.checkout.completed', - }, - }, - }, redis as any) - - expect(redis.set).toHaveBeenCalledWith('flux:user-1', '110') - }) - - it('ignores non-balance events', async () => { - const redis = { - set: vi.fn(async () => 'OK'), - } - - await handleCacheSyncMessage({ - streamMessageId: '1740000000000-1', - event: { - eventId: 'evt-2', - eventType: 'stripe.checkout.completed', - aggregateId: 'sess-1', - userId: 'user-1', - requestId: 'req-2', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - stripeEventId: 'stripe-evt-1', - stripeSessionId: 'sess-1', - amount: 500, - currency: 'usd', - }, - }, - }, redis as any) - - expect(redis.set).not.toHaveBeenCalled() - }) -}) diff --git a/apps/server/src/bin/__test__/run.test.ts b/apps/server/src/bin/__test__/run.test.ts index fce4cbaa7..a62461ad6 100644 --- a/apps/server/src/bin/__test__/run.test.ts +++ b/apps/server/src/bin/__test__/run.test.ts @@ -5,8 +5,7 @@ import { createServerCli, parseServerRole } from '../run' describe('server cli', () => { it('parses supported roles', () => { expect(parseServerRole(['api'])).toBe('api') - expect(parseServerRole(['cache-sync-consumer'])).toBe('cache-sync-consumer') - expect(parseServerRole(['outbox-dispatcher'])).toBe('outbox-dispatcher') + expect(parseServerRole(['billing-consumer'])).toBe('billing-consumer') }) it('returns null for unsupported or missing roles', () => { @@ -19,8 +18,7 @@ describe('server cli', () => { expect(cli.commands.map(command => command.name)).toEqual(expect.arrayContaining([ 'api', - 'cache-sync-consumer', - 'outbox-dispatcher', + 'billing-consumer', ])) }) }) diff --git a/apps/server/src/bin/run-outbox-dispatcher.ts b/apps/server/src/bin/run-billing-consumer.ts similarity index 67% rename from apps/server/src/bin/run-outbox-dispatcher.ts rename to apps/server/src/bin/run-billing-consumer.ts index 8d65ccabc..7dd4d8fd5 100644 --- a/apps/server/src/bin/run-outbox-dispatcher.ts +++ b/apps/server/src/bin/run-billing-consumer.ts @@ -6,9 +6,9 @@ import { createDrizzle, migrateDatabase } from '../libs/db' import { parseEnv } from '../libs/env' import { initializeExternalDependency } from '../libs/external-dependency' import { createRedis } from '../libs/redis' +import { createBillingConsumerHandler } from '../services/billing-consumer-handler' import { createBillingMqService } from '../services/billing-mq' -import { createOutboxDispatcher } from '../services/outbox-dispatcher' -import { createOutboxService } from '../services/outbox-service' +import { createBillingMqWorker } from '../services/billing-mq-worker' function parsePositiveInteger(rawValue: string, envKey: string): number { const parsed = Number(rawValue) @@ -19,11 +19,11 @@ function parsePositiveInteger(rawValue: string, envKey: string): number { return parsed } -export async function runOutboxDispatcher(): Promise { +export async function runBillingConsumer(): Promise { initLogger(LoggerLevel.Debug, LoggerFormat.Pretty) const env = parseEnv(process.env) - const logger = useLogger('outbox-dispatcher').useGlobalConfig() + const logger = useLogger('billing-consumer').useGlobalConfig() const { db, pool } = await initializeExternalDependency( 'Database', logger, @@ -62,14 +62,14 @@ export async function runOutboxDispatcher(): Promise { ) const abortController = new AbortController() - const claimedBy = env.OUTBOX_DISPATCHER_NAME ?? `outbox-dispatcher-${pid}` + const consumer = env.BILLING_EVENTS_CONSUMER_NAME ?? `billing-consumer-${pid}` const shutdown = (signalName: string) => { if (abortController.signal.aborted) { return } - logger.withFields({ signalName }).log('Stopping outbox dispatcher') + logger.withFields({ signalName }).log('Stopping billing consumer') abortController.abort() } @@ -77,18 +77,21 @@ export async function runOutboxDispatcher(): Promise { process.once('SIGTERM', () => shutdown('SIGTERM')) try { - const outboxService = createOutboxService(db) - const billingMqService = createBillingMqService(redis, { + const mq = createBillingMqService(redis, { stream: env.BILLING_EVENTS_STREAM, }) - const dispatcher = createOutboxDispatcher(outboxService, billingMqService) - await dispatcher.run({ - claimedBy, + const handler = createBillingConsumerHandler(db) + const worker = createBillingMqWorker(mq) + + await worker.run({ + group: 'billing-consumer', + consumer, signal: abortController.signal, - batchSize: parsePositiveInteger(env.OUTBOX_DISPATCHER_BATCH_SIZE, 'OUTBOX_DISPATCHER_BATCH_SIZE'), - claimTtlMs: parsePositiveInteger(env.OUTBOX_DISPATCHER_CLAIM_TTL_MS, 'OUTBOX_DISPATCHER_CLAIM_TTL_MS'), - pollIntervalMs: parsePositiveInteger(env.OUTBOX_DISPATCHER_POLL_MS, 'OUTBOX_DISPATCHER_POLL_MS'), + batchSize: parsePositiveInteger(env.BILLING_EVENTS_BATCH_SIZE, 'BILLING_EVENTS_BATCH_SIZE'), + blockMs: parsePositiveInteger(env.BILLING_EVENTS_BLOCK_MS, 'BILLING_EVENTS_BLOCK_MS'), + minIdleTimeMs: parsePositiveInteger(env.BILLING_EVENTS_MIN_IDLE_MS, 'BILLING_EVENTS_MIN_IDLE_MS'), + onMessage: message => handler.handleMessage(message), }) } finally { diff --git a/apps/server/src/bin/run-billing-events-consumer.ts b/apps/server/src/bin/run-billing-events-consumer.ts deleted file mode 100644 index c6e2d0742..000000000 --- a/apps/server/src/bin/run-billing-events-consumer.ts +++ /dev/null @@ -1,112 +0,0 @@ -import type { BillingStreamMessage } from '../services/billing-mq' - -import process, { pid } from 'node:process' - -import { initLogger, LoggerFormat, LoggerLevel, useLogger } from '@guiiai/logg' - -import { parseEnv } from '../libs/env' -import { initializeExternalDependency } from '../libs/external-dependency' -import { createRedis } from '../libs/redis' -import { createBillingMqService } from '../services/billing-mq' -import { createBillingMqWorker } from '../services/billing-mq-worker' -import { fluxRedisKey } from '../services/flux' - -function parsePositiveInteger(rawValue: string, envKey: string): number { - const parsed = Number(rawValue) - if (!Number.isInteger(parsed) || parsed <= 0) { - throw new Error(`${envKey} must be a positive integer`) - } - - return parsed -} - -export interface RunBillingEventsConsumerOptions { - group: string - loggerName: string - handleMessage?: (message: BillingStreamMessage, redis: ReturnType) => Promise -} - -export async function runBillingEventsConsumer(options: RunBillingEventsConsumerOptions): Promise { - initLogger(LoggerLevel.Debug, LoggerFormat.Pretty) - - const env = parseEnv(process.env) - const logger = useLogger(options.loggerName).useGlobalConfig() - 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 ?? `${options.group}-${pid}` - - const shutdown = async (signalName: string) => { - if (abortController.signal.aborted) { - return - } - - logger.withFields({ signalName }).log('Stopping billing MQ consumer') - abortController.abort() - } - - process.once('SIGINT', () => { - void shutdown('SIGINT') - }) - process.once('SIGTERM', () => { - void shutdown('SIGTERM') - }) - - try { - const mq = createBillingMqService(redis, { - stream: env.BILLING_EVENTS_STREAM, - }) - - const worker = createBillingMqWorker(mq) - const handleMessage = options.handleMessage ?? (async (message: BillingStreamMessage) => { - logger.withFields({ - group: options.group, - consumer, - eventId: message.event.eventId, - eventType: message.event.eventType, - aggregateId: message.event.aggregateId, - userId: message.event.userId, - streamMessageId: message.streamMessageId, - }).log('Consumed billing MQ event') - }) - - await worker.run({ - group: options.group, - consumer, - signal: abortController.signal, - batchSize: parsePositiveInteger(env.BILLING_EVENTS_BATCH_SIZE, 'BILLING_EVENTS_BATCH_SIZE'), - blockMs: parsePositiveInteger(env.BILLING_EVENTS_BLOCK_MS, 'BILLING_EVENTS_BLOCK_MS'), - minIdleTimeMs: parsePositiveInteger(env.BILLING_EVENTS_MIN_IDLE_MS, 'BILLING_EVENTS_MIN_IDLE_MS'), - onMessage: message => handleMessage(message, redis), - }) - } - finally { - await redis.quit() - } -} - -export async function handleCacheSyncMessage( - message: BillingStreamMessage, - redis: ReturnType, -): Promise { - if ((message.event.eventType === 'flux.credited' || message.event.eventType === 'flux.debited') - && message.event.payload.balanceAfter != null) { - await redis.set(fluxRedisKey(message.event.userId), String(message.event.payload.balanceAfter)) - } -} diff --git a/apps/server/src/bin/run.ts b/apps/server/src/bin/run.ts index a7006a025..6059122a5 100644 --- a/apps/server/src/bin/run.ts +++ b/apps/server/src/bin/run.ts @@ -8,10 +8,9 @@ import { errorMessageFrom } from '@moeru/std' import { cac } from 'cac' import { runApiServer } from '../app' -import { handleCacheSyncMessage, runBillingEventsConsumer } from './run-billing-events-consumer' -import { runOutboxDispatcher } from './run-outbox-dispatcher' +import { runBillingConsumer } from './run-billing-consumer' -const serverRoles = ['api', 'cache-sync-consumer', 'outbox-dispatcher'] as const +const serverRoles = ['api', 'billing-consumer'] as const type ServerRole = typeof serverRoles[number] @@ -20,15 +19,8 @@ async function runServerRole(role: ServerRole): Promise { case 'api': await runApiServer() return - case 'cache-sync-consumer': - await runBillingEventsConsumer({ - group: 'cache-sync', - loggerName: 'cache-sync-consumer', - handleMessage: handleCacheSyncMessage, - }) - return - case 'outbox-dispatcher': - await runOutboxDispatcher() + case 'billing-consumer': + await runBillingConsumer() } } @@ -41,12 +33,8 @@ export function createServerCli() { .action(() => runServerRole('api')) cli - .command('cache-sync-consumer', 'Start the cache-sync Redis Streams consumer') - .action(() => runServerRole('cache-sync-consumer')) - - cli - .command('outbox-dispatcher', 'Publish DB outbox events to Redis Streams') - .action(() => runServerRole('outbox-dispatcher')) + .command('billing-consumer', 'Start the billing events consumer (ledger, audit, request logs)') + .action(() => runServerRole('billing-consumer')) cli.help() diff --git a/apps/server/src/libs/env.ts b/apps/server/src/libs/env.ts index 63252397d..48327c402 100644 --- a/apps/server/src/libs/env.ts +++ b/apps/server/src/libs/env.ts @@ -30,10 +30,6 @@ const EnvSchema = object({ BILLING_EVENTS_BLOCK_MS: optional(string(), '5000'), BILLING_EVENTS_MIN_IDLE_MS: optional(string(), '30000'), - OUTBOX_DISPATCHER_NAME: optional(string()), - OUTBOX_DISPATCHER_BATCH_SIZE: optional(string(), '10'), - OUTBOX_DISPATCHER_CLAIM_TTL_MS: optional(string(), '30000'), - OUTBOX_DISPATCHER_POLL_MS: optional(string(), '1000'), // OpenTelemetry OTEL_SERVICE_NAMESPACE: optional(string(), 'airi'), OTEL_SERVICE_NAME: optional(string(), 'server'), diff --git a/apps/server/src/routes/__test__/v1completions.test.ts b/apps/server/src/routes/__test__/v1completions.test.ts index 3a873d441..9fbc6bfe5 100644 --- a/apps/server/src/routes/__test__/v1completions.test.ts +++ b/apps/server/src/routes/__test__/v1completions.test.ts @@ -1,7 +1,7 @@ +import type { BillingMqService } from '../../services/billing-mq' import type { BillingService } from '../../services/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 { Hono } from 'hono' @@ -53,19 +53,24 @@ function createMockConfigKV(overrides: Record = {}): ConfigKVServic } as any } -function createMockRequestLogService(): RequestLogService { +function createMockBillingMq(): BillingMqService { return { - logRequest: vi.fn(async () => {}), + stream: 'billing-events', + 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 } function createTestApp( fluxService: FluxService, configKV: ConfigKVService, - requestLogService: RequestLogService, billingService?: BillingService, + billingMq?: BillingMqService, ) { - const routes = createV1CompletionsRoutes(fluxService, billingService ?? createMockBillingService(), configKV, requestLogService, null) + const routes = createV1CompletionsRoutes(fluxService, billingService ?? createMockBillingService(), configKV, billingMq ?? createMockBillingMq(), null) const app = new Hono() app.onError((err, c) => { @@ -108,7 +113,6 @@ describe('v1CompletionsRoutes', () => { const app = createTestApp( createMockFluxService(), createMockConfigKV(), - createMockRequestLogService(), ) const res = await app.request('/api/v1/chat/completions', { @@ -123,7 +127,6 @@ describe('v1CompletionsRoutes', () => { const app = createTestApp( createMockFluxService(0), createMockConfigKV(), - createMockRequestLogService(), ) const res = await app.fetch( @@ -147,8 +150,7 @@ describe('v1CompletionsRoutes', () => { const fluxService = createMockFluxService(100) const billingService = createMockBillingService(100) const configKV = createMockConfigKV({ GATEWAY_BASE_URL: 'http://mock-gateway/' }) - const requestLogService = createMockRequestLogService() - const app = createTestApp(fluxService, configKV, requestLogService, billingService) + const app = createTestApp(fluxService, configKV, billingService) const res = await app.fetch( new Request('http://localhost/api/v1/chat/completions', { @@ -187,7 +189,6 @@ describe('v1CompletionsRoutes', () => { const app = createTestApp( createMockFluxService(), createMockConfigKV({ DEFAULT_CHAT_MODEL: 'anthropic/claude-sonnet' }), - createMockRequestLogService(), ) await app.fetch( @@ -213,7 +214,7 @@ describe('v1CompletionsRoutes', () => { headers: { 'Content-Type': 'application/json' }, })) - const app = createTestApp(createMockFluxService(), createMockConfigKV(), createMockRequestLogService()) + const app = createTestApp(createMockFluxService(), createMockConfigKV()) await app.fetch( new Request('http://localhost/api/v1/chat/completions', { @@ -239,7 +240,7 @@ describe('v1CompletionsRoutes', () => { })) const billingService = createMockBillingService(100) - const app = createTestApp(createMockFluxService(100), createMockConfigKV(), createMockRequestLogService(), billingService) + const app = createTestApp(createMockFluxService(100), createMockConfigKV(), billingService) const res = await app.fetch( new Request('http://localhost/api/v1/chat/completions', { @@ -260,7 +261,7 @@ describe('v1CompletionsRoutes', () => { // Override getOptional to return null for required keys configKV.getOptional = vi.fn(async () => null) - const app = createTestApp(createMockFluxService(), configKV, createMockRequestLogService()) + const app = createTestApp(createMockFluxService(), configKV) const res = await app.fetch( new Request('http://localhost/api/v1/chat/completions', { @@ -273,14 +274,14 @@ describe('v1CompletionsRoutes', () => { expect(res.status).toBe(503) }) - it('should log the request', async () => { + it('should publish request log event via billingMq', async () => { globalThis.fetch = vi.fn(async () => new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' }, })) - const requestLogService = createMockRequestLogService() - const app = createTestApp(createMockFluxService(), createMockConfigKV(), requestLogService) + const billingMq = createMockBillingMq() + const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, billingMq) await app.fetch( new Request('http://localhost/api/v1/chat/completions', { @@ -291,12 +292,16 @@ describe('v1CompletionsRoutes', () => { { user: testUser } as any, ) - expect(requestLogService.logRequest).toHaveBeenCalledWith( + expect(billingMq.publish).toHaveBeenCalledWith( expect.objectContaining({ + eventType: 'llm.request.log', + aggregateId: 'user-1', userId: 'user-1', - model: 'gpt-4', - status: 200, - fluxConsumed: 1, + payload: expect.objectContaining({ + model: 'gpt-4', + status: 200, + fluxConsumed: 1, + }), }), ) }) @@ -310,7 +315,7 @@ describe('v1CompletionsRoutes', () => { headers: { 'Content-Type': 'audio/mpeg' }, })) - const app = createTestApp(createMockFluxService(), createMockConfigKV(), createMockRequestLogService()) + const app = createTestApp(createMockFluxService(), createMockConfigKV()) const res = await app.fetch( new Request('http://localhost/api/v1/audio/speech', { @@ -336,7 +341,7 @@ describe('v1CompletionsRoutes', () => { headers: { 'Content-Type': 'application/json' }, })) - const app = createTestApp(createMockFluxService(), createMockConfigKV(), createMockRequestLogService()) + const app = createTestApp(createMockFluxService(), createMockConfigKV()) const formData = new FormData() formData.append('file', new Blob(['audio']), 'test.wav') @@ -360,7 +365,7 @@ describe('v1CompletionsRoutes', () => { describe('route matching', () => { it('gET /api/v1/chat/completions should return 404', async () => { - const app = createTestApp(createMockFluxService(), createMockConfigKV(), createMockRequestLogService()) + const app = createTestApp(createMockFluxService(), createMockConfigKV()) const res = await app.fetch( new Request('http://localhost/api/v1/chat/completions', { method: 'GET' }), @@ -375,7 +380,7 @@ describe('v1CompletionsRoutes', () => { headers: { 'Content-Type': 'application/json' }, })) - const app = createTestApp(createMockFluxService(), createMockConfigKV(), createMockRequestLogService()) + const app = createTestApp(createMockFluxService(), createMockConfigKV()) const res = await app.fetch( new Request('http://localhost/api/v1/chat/completion', { diff --git a/apps/server/src/routes/v1completions.ts b/apps/server/src/routes/v1completions.ts index 7eb6e3a7c..f29d4a229 100644 --- a/apps/server/src/routes/v1completions.ts +++ b/apps/server/src/routes/v1completions.ts @@ -2,10 +2,10 @@ import type { Context } from 'hono' import type { LlmMetrics } from '../libs/otel' import type { UsageInfo } from '../services/billing' +import type { BillingMqService } from '../services/billing-mq' import type { BillingService } from '../services/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 { useLogger } from '@guiiai/logg' @@ -42,7 +42,7 @@ function normalizeBaseUrl(gatewayBaseUrl: string): string { return gatewayBaseUrl.endsWith('/') ? gatewayBaseUrl : `${gatewayBaseUrl}/` } -export function createV1CompletionsRoutes(fluxService: FluxService, billingService: BillingService, configKV: ConfigKVService, requestLogService: RequestLogService, llm: LlmMetrics | null) { +export function createV1CompletionsRoutes(fluxService: FluxService, billingService: BillingService, configKV: ConfigKVService, billingMq: BillingMqService, llm?: LlmMetrics | null) { const logger = useLogger('v1-completions').useGlobalConfig() function recordMetrics(opts: { model: string, status: number, type: string, durationMs: number, fluxConsumed: number, promptTokens?: number, completionTokens?: number }) { @@ -58,6 +58,25 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi llm.tokensCompletion.add(opts.completionTokens, { model: opts.model }) } + 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')) + } + // NOTICE: Billing is best-effort — flux is debited AFTER the LLM response is sent. // This is a deliberate tradeoff: users get lower latency and uninterrupted streaming, // at the cost of a small revenue leak when debit fails (e.g. DB timeout). @@ -174,7 +193,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') } - requestLogService.logRequest({ + publishRequestLog({ userId: user.id, model: requestModel, status: response.status, @@ -182,7 +201,7 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi fluxConsumed: actualCharged, promptTokens: usage.promptTokens, completionTokens: usage.completionTokens, - }).catch(err => logger.withError(err).warn('Failed to log streaming request')) + }) } })() @@ -215,7 +234,7 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi description: requestModel, }) - requestLogService.logRequest({ + publishRequestLog({ userId: user.id, model: requestModel, status: response.status, @@ -223,7 +242,7 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi fluxConsumed, promptTokens: usage.promptTokens, completionTokens: usage.completionTokens, - }).catch(err => logger.withError(err).warn('Failed to log request')) + }) return c.json(responseBody) } @@ -278,13 +297,13 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi span.end() recordMetrics({ model: requestModel, status: response.status, type: 'tts', durationMs, fluxConsumed: fluxPerRequest }) - requestLogService.logRequest({ + publishRequestLog({ userId: user.id, model: requestModel, status: response.status, durationMs, fluxConsumed: fluxPerRequest, - }).catch(err => logger.withError(err).warn('Failed to log TTS request')) + }) return new Response(response.body, { status: response.status, @@ -343,13 +362,13 @@ export function createV1CompletionsRoutes(fluxService: FluxService, billingServi span.end() recordMetrics({ model: 'auto', status: response.status, type: 'asr', durationMs, fluxConsumed: fluxPerRequest }) - requestLogService.logRequest({ + publishRequestLog({ userId: user.id, model: 'auto', status: response.status, durationMs, fluxConsumed: fluxPerRequest, - }).catch(err => logger.withError(err).warn('Failed to log ASR request')) + }) return new Response(response.body, { status: response.status, diff --git a/apps/server/src/schemas/index.ts b/apps/server/src/schemas/index.ts index 3d1363da6..b836f8282 100644 --- a/apps/server/src/schemas/index.ts +++ b/apps/server/src/schemas/index.ts @@ -5,7 +5,6 @@ export * from './flux' export * from './flux-audit-log' export * from './flux-ledger' export * from './llm-request-log' -export * from './outbox-events' export * from './providers' export * from './stripe' export * from './user-character' diff --git a/apps/server/src/schemas/outbox-events.ts b/apps/server/src/schemas/outbox-events.ts deleted file mode 100644 index dabc7f8c7..000000000 --- a/apps/server/src/schemas/outbox-events.ts +++ /dev/null @@ -1,31 +0,0 @@ -import type { InferInsertModel, InferSelectModel } from 'drizzle-orm' - -import { index, integer, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core' - -import { nanoid } from '../utils/id' - -export const outboxEvents = pgTable('outbox_events', { - id: text('id').primaryKey().$defaultFn(() => nanoid()), - eventId: text('event_id').notNull(), - eventType: text('event_type').notNull(), - aggregateId: text('aggregate_id').notNull(), - userId: text('user_id').notNull(), - requestId: text('request_id'), - schemaVersion: integer('schema_version').notNull(), - payload: text('payload').notNull(), - occurredAt: timestamp('occurred_at').notNull(), - availableAt: timestamp('available_at').defaultNow().notNull(), - claimedBy: text('claimed_by'), - claimExpiresAt: timestamp('claim_expires_at'), - publishedAt: timestamp('published_at'), - streamMessageId: text('stream_message_id'), - createdAt: timestamp('created_at').defaultNow().notNull(), - updatedAt: timestamp('updated_at').defaultNow().notNull(), -}, table => [ - uniqueIndex('outbox_events_event_id_idx').on(table.eventId), - index('outbox_events_publish_scan_idx').on(table.publishedAt, table.availableAt, table.claimExpiresAt, table.createdAt), - index('outbox_events_claimed_by_idx').on(table.claimedBy), -]) - -export type OutboxEvent = InferSelectModel -export type NewOutboxEvent = InferInsertModel diff --git a/apps/server/src/services/__test__/billing-service.test.ts b/apps/server/src/services/__test__/billing-service.test.ts index ecd115b47..163b39715 100644 --- a/apps/server/src/services/__test__/billing-service.test.ts +++ b/apps/server/src/services/__test__/billing-service.test.ts @@ -1,6 +1,7 @@ import type Redis from 'ioredis' import type { Database } from '../../libs/db' +import type { BillingMqService } from '../billing-mq' import type { createConfigKVService } from '../config-kv' import { eq } from 'drizzle-orm' @@ -8,7 +9,6 @@ import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' import { mockDB } from '../../libs/mock-db' import { createBillingService } from '../billing-service' -import { createOutboxService } from '../outbox-service' import * as schema from '../../schemas' @@ -30,15 +30,25 @@ function createMockRedis(): Redis { } as unknown as Redis } +function createMockBillingMq(): BillingMqService { + return { + stream: 'billing-events', + 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 outboxService: ReturnType + let billingMq: BillingMqService let billingService: ReturnType beforeAll(async () => { db = await mockDB(schema) - outboxService = createOutboxService(db) await db.insert(schema.user).values({ id: 'user-billing-1', @@ -49,9 +59,9 @@ describe('billingService', () => { beforeEach(async () => { redis = createMockRedis() - billingService = createBillingService(db, redis, outboxService, createMockConfigKV()) + billingMq = createMockBillingMq() + billingService = createBillingService(db, redis, billingMq, createMockConfigKV()) - await db.delete(schema.outboxEvents) await db.delete(schema.fluxAuditLog) await db.delete(schema.fluxLedger) await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1')) @@ -98,10 +108,10 @@ describe('billingService', () => { expect(auditRecords).toHaveLength(1) expect(auditRecords[0]?.amount).toBe(50) - // Verify outbox events - const outboxRecords = await db.select().from(schema.outboxEvents).orderBy(schema.outboxEvents.createdAt) - expect(outboxRecords).toHaveLength(2) - expect(outboxRecords.map(record => record.eventType)).toEqual(['flux.credited', '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')) @@ -132,13 +142,13 @@ describe('billingService', () => { expect(second).toEqual({ applied: false }) - const outboxRecords = await db.select().from(schema.outboxEvents) - expect(outboxRecords).toHaveLength(2) + // Only 2 publish calls from the first invocation + expect(billingMq.publish).toHaveBeenCalledTimes(2) }) }) describe('debitFlux', () => { - it('deducts balance, writes ledger + audit + outbox, updates Redis', async () => { + it('deducts balance, publishes flux.debited event, updates Redis', async () => { // Setup: give user some flux first await db.insert(schema.userFlux).values({ userId: 'user-billing-1', flux: 100 }) @@ -155,26 +165,16 @@ describe('billingService', () => { const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1')) expect(fluxRecord?.flux).toBe(70) - // Verify ledger - const ledgerRecords = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-billing-1')) - expect(ledgerRecords).toHaveLength(1) - expect(ledgerRecords[0]).toMatchObject({ - type: 'debit', - amount: 30, - balanceBefore: 100, - balanceAfter: 70, - requestId: 'req-1', - }) - - // Verify audit log - const auditRecords = await db.select().from(schema.fluxAuditLog).where(eq(schema.fluxAuditLog.userId, 'user-billing-1')) - expect(auditRecords).toHaveLength(1) - expect(auditRecords[0]?.amount).toBe(-30) - - // Verify outbox event - const outboxRecords = await db.select().from(schema.outboxEvents) - expect(outboxRecords).toHaveLength(1) - expect(outboxRecords[0]?.eventType).toBe('flux.debited') + // Verify flux.debited event published to stream (ledger + audit written by consumer) + expect(billingMq.publish).toHaveBeenCalledTimes(1) + expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({ + eventType: 'flux.debited', + userId: 'user-billing-1', + payload: expect.objectContaining({ + amount: 30, + balanceAfter: 70, + }), + })) // Verify Redis cache updated expect(redis.set).toHaveBeenCalledWith('flux:user-billing-1', '70') @@ -195,8 +195,8 @@ describe('billingService', () => { const ledgerRecords = await db.select().from(schema.fluxLedger) expect(ledgerRecords).toHaveLength(0) - const outboxRecords = await db.select().from(schema.outboxEvents) - expect(outboxRecords).toHaveLength(0) + // Verify no event was published + expect(billingMq.publish).not.toHaveBeenCalled() }) }) @@ -222,10 +222,9 @@ describe('billingService', () => { balanceAfter: 50, }) - // Verify outbox - const outboxRecords = await db.select().from(schema.outboxEvents) - expect(outboxRecords).toHaveLength(1) - expect(outboxRecords[0]?.eventType).toBe('flux.credited') + // Verify billing event published to stream + expect(billingMq.publish).toHaveBeenCalledTimes(1) + expect(billingMq.publish).toHaveBeenCalledWith(expect.objectContaining({ eventType: 'flux.credited' })) }) }) }) diff --git a/apps/server/src/services/__test__/outbox-dispatcher.test.ts b/apps/server/src/services/__test__/outbox-dispatcher.test.ts deleted file mode 100644 index b93964224..000000000 --- a/apps/server/src/services/__test__/outbox-dispatcher.test.ts +++ /dev/null @@ -1,80 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' - -import { createOutboxDispatcher } from '../outbox-dispatcher' - -describe('outboxDispatcher', () => { - it('publishes claimed events and marks them as published', async () => { - const outboxService = { - claimPending: vi.fn(async () => ([ - { - id: 'outbox-1', - event: { - eventId: 'evt-1', - eventType: 'flux.credited' as const, - aggregateId: 'user-1', - userId: 'user-1', - requestId: 'req-1', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - amount: 10, - balanceAfter: 110, - source: 'stripe.checkout.completed', - }, - }, - }, - ])), - markPublished: vi.fn(async () => {}), - releaseClaim: vi.fn(async () => {}), - } - - const billingMqService = { - publish: vi.fn(async () => '1740000000000-0'), - } - - const dispatcher = createOutboxDispatcher(outboxService as any, billingMqService as any) - await expect(dispatcher.dispatchBatch('dispatcher-1', 10, 30_000)).resolves.toBe(1) - - expect(billingMqService.publish).toHaveBeenCalled() - expect(outboxService.markPublished).toHaveBeenCalledWith('outbox-1', '1740000000000-0') - expect(outboxService.releaseClaim).not.toHaveBeenCalled() - }) - - it('releases claims when publish fails', async () => { - const outboxService = { - claimPending: vi.fn(async () => ([ - { - id: 'outbox-2', - event: { - eventId: 'evt-2', - eventType: 'flux.credited' as const, - aggregateId: 'user-2', - userId: 'user-2', - requestId: 'req-2', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - amount: 5, - balanceAfter: 15, - source: 'stripe.checkout.completed', - }, - }, - }, - ])), - markPublished: vi.fn(async () => {}), - releaseClaim: vi.fn(async () => {}), - } - - const billingMqService = { - publish: vi.fn(async () => { - throw new Error('redis down') - }), - } - - const dispatcher = createOutboxDispatcher(outboxService as any, billingMqService as any) - await expect(dispatcher.dispatchBatch('dispatcher-2', 10, 30_000)).resolves.toBe(1) - - expect(outboxService.releaseClaim).toHaveBeenCalledWith('outbox-2') - expect(outboxService.markPublished).not.toHaveBeenCalled() - }) -}) diff --git a/apps/server/src/services/__test__/outbox-service.test.ts b/apps/server/src/services/__test__/outbox-service.test.ts deleted file mode 100644 index c7ede088d..000000000 --- a/apps/server/src/services/__test__/outbox-service.test.ts +++ /dev/null @@ -1,91 +0,0 @@ -import type { Database } from '../../libs/db' - -import { eq } from 'drizzle-orm' -import { beforeAll, describe, expect, it } from 'vitest' - -import { mockDB } from '../../libs/mock-db' -import { createOutboxService } from '../outbox-service' - -import * as schema from '../../schemas' - -describe('outboxService', () => { - let db: Database - let outboxService: ReturnType - - beforeAll(async () => { - db = await mockDB(schema) - outboxService = createOutboxService(db) - }) - - it('enqueues and claims unpublished events', async () => { - await outboxService.enqueue(db, { - eventId: 'evt-1', - eventType: 'flux.credited', - aggregateId: 'user-1', - userId: 'user-1', - requestId: 'req-1', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - amount: 10, - balanceAfter: 110, - source: 'stripe.checkout.completed', - }, - }) - - const claimed = await outboxService.claimPending({ - claimedBy: 'dispatcher-1', - limit: 10, - claimTtlMs: 30_000, - }) - - expect(claimed).toHaveLength(1) - expect(claimed[0]?.event).toEqual({ - eventId: 'evt-1', - eventType: 'flux.credited', - aggregateId: 'user-1', - userId: 'user-1', - requestId: 'req-1', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - amount: 10, - balanceAfter: 110, - source: 'stripe.checkout.completed', - }, - }) - }) - - it('marks events as published and can release claims', async () => { - await outboxService.enqueue(db, { - eventId: 'evt-2', - eventType: 'stripe.checkout.completed', - aggregateId: 'sess-2', - userId: 'user-2', - requestId: 'req-2', - occurredAt: '2026-03-24T00:00:00.000Z', - schemaVersion: 1, - payload: { - stripeEventId: 'stripe-evt-2', - stripeSessionId: 'sess-2', - amount: 500, - currency: 'usd', - }, - }) - - const [claimed] = await outboxService.claimPending({ - claimedBy: 'dispatcher-2', - limit: 10, - claimTtlMs: 30_000, - }) - - expect(claimed).toBeDefined() - await outboxService.releaseClaim(claimed!.id) - await outboxService.markPublished(claimed!.id, '1740000000000-0') - - const [published] = await db.select().from(schema.outboxEvents).where(eq(schema.outboxEvents.id, claimed!.id)) - expect(published?.streamMessageId).toBe('1740000000000-0') - expect(published?.publishedAt).toBeInstanceOf(Date) - expect(published?.claimedBy).toBeNull() - }) -}) diff --git a/apps/server/src/services/billing-consumer-handler.ts b/apps/server/src/services/billing-consumer-handler.ts new file mode 100644 index 000000000..9a5a5169f --- /dev/null +++ b/apps/server/src/services/billing-consumer-handler.ts @@ -0,0 +1,83 @@ +import type { Database } from '../libs/db' +import type { BillingStreamMessage } from './billing-mq' + +import { useLogger } from '@guiiai/logg' + +import * as fluxAuditSchema from '../schemas/flux-audit-log' +import * as fluxLedgerSchema from '../schemas/flux-ledger' +import * as llmRequestLogSchema from '../schemas/llm-request-log' + +const logger = useLogger('billing-consumer-handler').useGlobalConfig() + +export function createBillingConsumerHandler(db: Database) { + return { + async handleMessage(message: BillingStreamMessage): Promise { + const { event } = message + + switch (event.eventType) { + case 'flux.debited': { + const balanceBefore = event.payload.balanceAfter != null + ? event.payload.balanceAfter + event.payload.amount + : 0 + + await db.insert(fluxLedgerSchema.fluxLedger).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.source ?? 'LLM request', + }) + + await db.insert(fluxAuditSchema.fluxAuditLog).values({ + userId: event.userId, + type: 'consumption', + amount: -event.payload.amount, + description: event.payload.source ?? 'LLM request', + }) + + logger.withFields({ + eventId: event.eventId, + userId: event.userId, + amount: event.payload.amount, + }).log('Wrote debit ledger + audit') + break + } + + case 'llm.request.log': { + await db.insert(llmRequestLogSchema.llmRequestLog).values({ + 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, + }) + + 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 diff --git a/apps/server/src/services/billing-events.ts b/apps/server/src/services/billing-events.ts index e21912dd0..d1316895b 100644 --- a/apps/server/src/services/billing-events.ts +++ b/apps/server/src/services/billing-events.ts @@ -20,6 +20,7 @@ const BillingEventTypeSchema = union([ literal('flux.credited'), literal('stripe.checkout.completed'), literal('llm.request.completed'), + literal('llm.request.log'), ]) const BalanceChangePayloadSchema = object({ @@ -43,6 +44,15 @@ const LlmRequestCompletedPayloadSchema = object({ 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, @@ -60,6 +70,7 @@ type BillingEventEnvelope = InferOutput type BalanceChangePayload = InferOutput type StripeCheckoutCompletedPayload = InferOutput type LlmRequestCompletedPayload = InferOutput +type LlmRequestLogPayload = InferOutput export type FluxDebitedEvent = BillingEventEnvelope & { eventType: 'flux.debited' @@ -81,11 +92,17 @@ export type LlmRequestCompletedEvent = BillingEventEnvelope & { 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 { event_id: string @@ -153,5 +170,11 @@ export function parseBillingEvent(fields: Record): B eventType: 'llm.request.completed', payload: parse(LlmRequestCompletedPayloadSchema, parsedEnvelope.payload), } + case 'llm.request.log': + return { + ...parsedEnvelope, + eventType: 'llm.request.log', + payload: parse(LlmRequestLogPayloadSchema, parsedEnvelope.payload), + } } } diff --git a/apps/server/src/services/billing-service.ts b/apps/server/src/services/billing-service.ts index 1b9c94974..59db8f930 100644 --- a/apps/server/src/services/billing-service.ts +++ b/apps/server/src/services/billing-service.ts @@ -2,8 +2,9 @@ import type Redis from 'ioredis' import type { Database } from '../libs/db' import type { RevenueMetrics } from '../libs/otel' +import type { BillingEvent } from './billing-events' +import type { BillingMqService } from './billing-mq' import type { ConfigKVService } from './config-kv' -import type { OutboxService } from './outbox-service' import { useLogger } from '@guiiai/logg' import { eq } from 'drizzle-orm' @@ -22,7 +23,7 @@ const logger = useLogger('billing-service') export function createBillingService( db: Database, redis: Redis, - outboxService: OutboxService, + billingMq: BillingMqService, _configKV: ConfigKVService, metrics?: RevenueMetrics | null, ) { @@ -39,10 +40,29 @@ 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 { + 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') + } + } + return { /** * Debit flux from a user's balance within a DB transaction. - * Writes flux_ledger + flux_audit_log + outbox event atomically. + * The transaction ONLY locks the row and updates the balance. + * Ledger + audit entries are written by the billing-mq consumer + * after it processes the flux.debited event published post-commit. */ async debitFlux(input: { userId: string @@ -75,54 +95,36 @@ export function createBillingService( .set({ flux: balanceAfter, updatedAt: new Date() }) .where(eq(fluxSchema.userFlux.userId, input.userId)) - // 3. Append ledger entry - await tx.insert(fluxLedgerSchema.fluxLedger).values({ - userId: input.userId, - type: 'debit', - amount: input.amount, - balanceBefore, - balanceAfter, - requestId: input.requestId, - description: input.description ?? 'LLM request', - }) - - // 4. Append audit log (user-facing history) - await tx.insert(fluxAuditSchema.fluxAuditLog).values({ - userId: input.userId, - type: 'consumption', - amount: -input.amount, - description: input.description ?? 'LLM request', - }) - - // 5. Enqueue outbox event - await outboxService.enqueue(tx, { - eventId: nanoid(), - eventType: 'flux.debited', - aggregateId: input.userId, - userId: input.userId, - requestId: input.requestId, - occurredAt: new Date().toISOString(), - schemaVersion: 1, - payload: { - amount: input.amount, - balanceAfter, - source: 'llm.request', - }, - }) - - return { userId: input.userId, flux: balanceAfter } + return { userId: input.userId, flux: balanceAfter, balanceBefore } }) - // 6. Update Redis cache after commit (best-effort) + // 3. Update Redis cache after commit (best-effort) await updateRedisCache(input.userId, result.flux) + // 4. Publish flux.debited event to stream; ledger + audit written by consumer + await publishEvent({ + eventId: nanoid(), + eventType: 'flux.debited', + aggregateId: input.userId, + userId: input.userId, + requestId: input.requestId, + occurredAt: new Date().toISOString(), + schemaVersion: 1, + payload: { + amount: input.amount, + balanceAfter: result.flux, + source: 'llm.request', + }, + }) + logger.withFields({ userId: input.userId, amount: input.amount, balance: result.flux }).log('Debited flux') - return result + return { userId: result.userId, flux: result.flux } }, /** * Credit flux to a user's balance within a DB transaction. * Generic credit method for non-Stripe flows (e.g. admin grants). + * Ledger + audit entries are written inside the transaction for immediate visibility. */ async creditFlux(input: { userId: string @@ -173,27 +175,27 @@ export function createBillingService( metadata: input.auditMetadata, }) - // Outbox event - await outboxService.enqueue(tx, { - 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, - source: input.source, - }, - }) - return { balanceBefore, balanceAfter } }) await updateRedisCache(input.userId, result.balanceAfter) + // 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, + }, + }) + logger.withFields({ userId: input.userId, amount: input.amount, balance: result.balanceAfter }).log('Credited flux') return result }, @@ -201,6 +203,7 @@ export function createBillingService( /** * Credit flux from a Stripe checkout session (one-time payment). * Idempotent: checks fluxCredited flag before applying. + * Ledger + audit entries are written inside the transaction for immediate visibility. */ async creditFluxFromStripeCheckout(input: { stripeEventId: string @@ -270,9 +273,15 @@ export function createBillingService( }, }) - // Outbox events + return { applied: true, balanceAfter } + }) + + if (txResult.applied && txResult.balanceAfter != null) { + await updateRedisCache(input.userId, txResult.balanceAfter) + + // Publish both events after commit const occurredAt = new Date().toISOString() - await outboxService.enqueue(tx, { + await publishEvent({ eventId: nanoid(), eventType: 'flux.credited', aggregateId: input.userId, @@ -282,12 +291,12 @@ export function createBillingService( schemaVersion: 1, payload: { amount: input.fluxAmount, - balanceAfter, + balanceAfter: txResult.balanceAfter, source: 'stripe.checkout.completed', }, }) - await outboxService.enqueue(tx, { + await publishEvent({ eventId: nanoid(), eventType: 'stripe.checkout.completed', aggregateId: input.stripeSessionId, @@ -302,12 +311,6 @@ export function createBillingService( currency: input.currency ?? 'unknown', }, }) - - return { applied: true, balanceAfter } - }) - - if (txResult.applied && txResult.balanceAfter != null) { - await updateRedisCache(input.userId, txResult.balanceAfter) } return txResult @@ -316,6 +319,7 @@ export function createBillingService( /** * Credit flux from a Stripe invoice payment (subscription). * Idempotent: checks fluxCredited flag on the invoice record. + * Ledger + audit entries are written inside the transaction for immediate visibility. */ async creditFluxFromInvoice(input: { stripeEventId: string @@ -385,8 +389,14 @@ export function createBillingService( }, }) - // Outbox event - await outboxService.enqueue(tx, { + return { applied: true, balanceAfter } + }) + + 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, @@ -396,16 +406,10 @@ export function createBillingService( schemaVersion: 1, payload: { amount: input.fluxAmount, - balanceAfter, + balanceAfter: txResult.balanceAfter, source: 'invoice.paid', }, }) - - return { applied: true, balanceAfter } - }) - - if (txResult.applied && txResult.balanceAfter != null) { - await updateRedisCache(input.userId, txResult.balanceAfter) } return txResult diff --git a/apps/server/src/services/outbox-dispatcher.ts b/apps/server/src/services/outbox-dispatcher.ts deleted file mode 100644 index 92c5638c9..000000000 --- a/apps/server/src/services/outbox-dispatcher.ts +++ /dev/null @@ -1,83 +0,0 @@ -import type { BillingMqService } from './billing-mq' -import type { OutboxService } from './outbox-service' - -import { useLogger } from '@guiiai/logg' - -export interface RunOutboxDispatcherOptions { - claimedBy: string - signal: AbortSignal - batchSize?: number - claimTtlMs?: number - pollIntervalMs?: number -} - -const logger = useLogger('outbox-dispatcher').useGlobalConfig() - -export function createOutboxDispatcher(outboxService: OutboxService, billingMqService: BillingMqService) { - return { - async dispatchBatch(claimedBy: string, batchSize: number, claimTtlMs: number): Promise { - const claimedEvents = await outboxService.claimPending({ - claimedBy, - limit: batchSize, - claimTtlMs, - }) - - for (const claimedEvent of claimedEvents) { - try { - const streamMessageId = await billingMqService.publish(claimedEvent.event) - await outboxService.markPublished(claimedEvent.id, streamMessageId) - } - catch (error) { - await outboxService.releaseClaim(claimedEvent.id) - logger.withError(error).withFields({ - claimedBy, - outboxId: claimedEvent.id, - eventId: claimedEvent.event.eventId, - eventType: claimedEvent.event.eventType, - }).error('Failed to dispatch outbox event') - } - } - - return claimedEvents.length - }, - - async run(options: RunOutboxDispatcherOptions): Promise { - while (!options.signal.aborted) { - const dispatchedCount = await this.dispatchBatch( - options.claimedBy, - options.batchSize ?? 10, - options.claimTtlMs ?? 30_000, - ) - - if (dispatchedCount > 0) { - continue - } - - await sleep(options.pollIntervalMs ?? 1_000, options.signal) - } - }, - } -} - -async function sleep(timeoutMs: number, signal: AbortSignal): Promise { - if (signal.aborted) { - return - } - - await new Promise((resolve) => { - const timeout = setTimeout(() => { - signal.removeEventListener('abort', onAbort) - resolve() - }, timeoutMs) - - const onAbort = () => { - clearTimeout(timeout) - signal.removeEventListener('abort', onAbort) - resolve() - } - - signal.addEventListener('abort', onAbort, { once: true }) - }) -} - -export type OutboxDispatcher = ReturnType diff --git a/apps/server/src/services/outbox-service.ts b/apps/server/src/services/outbox-service.ts deleted file mode 100644 index 1e12e73c6..000000000 --- a/apps/server/src/services/outbox-service.ts +++ /dev/null @@ -1,123 +0,0 @@ -import type { Database } from '../libs/db' -import type { BillingEvent } from './billing-events' - -import { and, asc, eq, inArray, isNull, lte, or } from 'drizzle-orm' - -import { outboxEvents } from '../schemas/outbox-events' -import { parseBillingEvent, serializeBillingEvent } from './billing-events' - -interface OutboxWriter { - insert: Database['insert'] -} - -export interface ClaimOutboxEventsOptions { - claimedBy: string - limit: number - claimTtlMs: number -} - -export interface ClaimedOutboxEvent { - id: string - event: BillingEvent -} - -export function createOutboxService(db: Database) { - return { - async enqueue(writer: OutboxWriter, event: BillingEvent): Promise { - const serializedEvent = serializeBillingEvent(event) - await writer.insert(outboxEvents).values({ - eventId: serializedEvent.event_id, - eventType: serializedEvent.event_type, - aggregateId: serializedEvent.aggregate_id, - userId: serializedEvent.user_id, - requestId: serializedEvent.request_id, - schemaVersion: Number(serializedEvent.schema_version), - payload: serializedEvent.payload, - occurredAt: new Date(serializedEvent.occurred_at), - }) - }, - - async claimPending(options: ClaimOutboxEventsOptions): Promise { - const now = new Date() - const claimExpiresAt = new Date(now.getTime() + options.claimTtlMs) - - return db.transaction(async (tx) => { - const claimable = tx.$with('claimable').as( - tx - .select({ id: outboxEvents.id }) - .from(outboxEvents) - .where(and( - isNull(outboxEvents.publishedAt), - lte(outboxEvents.availableAt, now), - or( - isNull(outboxEvents.claimExpiresAt), - lte(outboxEvents.claimExpiresAt, now), - ), - )) - .orderBy(asc(outboxEvents.createdAt)) - .limit(options.limit) - .for('update', { skipLocked: true }), - ) - - const claimedRows = await tx - .with(claimable) - .update(outboxEvents) - .set({ - claimedBy: options.claimedBy, - claimExpiresAt, - updatedAt: now, - }) - .where(inArray(outboxEvents.id, tx.select({ id: claimable.id }).from(claimable))) - .returning({ - id: outboxEvents.id, - eventId: outboxEvents.eventId, - eventType: outboxEvents.eventType, - aggregateId: outboxEvents.aggregateId, - userId: outboxEvents.userId, - requestId: outboxEvents.requestId, - schemaVersion: outboxEvents.schemaVersion, - payload: outboxEvents.payload, - occurredAt: outboxEvents.occurredAt, - }) - - return claimedRows.map(row => ({ - id: row.id, - event: parseBillingEvent({ - event_id: row.eventId, - event_type: row.eventType, - aggregate_id: row.aggregateId, - user_id: row.userId, - request_id: row.requestId ?? undefined, - schema_version: String(row.schemaVersion), - payload: row.payload, - occurred_at: row.occurredAt.toISOString(), - }), - })) - }) - }, - - async markPublished(id: string, streamMessageId: string): Promise { - await db.update(outboxEvents) - .set({ - publishedAt: new Date(), - streamMessageId, - claimedBy: null, - claimExpiresAt: null, - updatedAt: new Date(), - }) - .where(eq(outboxEvents.id, id)) - }, - - async releaseClaim(id: string): Promise { - await db.update(outboxEvents) - .set({ - claimedBy: null, - claimExpiresAt: null, - updatedAt: new Date(), - }) - .where(eq(outboxEvents.id, id)) - }, - } -} - -export type OutboxService = ReturnType diff --git a/docs/superpowers/plans/2026-03-27-remove-outbox-simplify-mq.md b/docs/superpowers/plans/2026-03-27-remove-outbox-simplify-mq.md new file mode 100644 index 000000000..2002050fa --- /dev/null +++ b/docs/superpowers/plans/2026-03-27-remove-outbox-simplify-mq.md @@ -0,0 +1,1534 @@ +# Remove Outbox Pattern & Simplify MQ Architecture + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Remove the DB outbox pattern, consolidate 3 server processes into 2 (`api` + `billing-consumer`), and make the MQ actually useful by handling async writes (ledger, audit log, request log) via direct `XADD` from the API. + +**Architecture:** API writes balance to DB synchronously (source of truth), then fires `XADD` to Redis Stream for async side effects. A single `billing-consumer` process reads the stream and batch-writes ledger entries, audit logs, and LLM request logs to DB. The outbox table, outbox-dispatcher process, and cache-sync-consumer are removed entirely. + +**Tech Stack:** Hono, Drizzle ORM, Redis Streams (ioredis), Valibot, Vitest, injeca + +--- + +## File Structure + +### Files to DELETE +- `src/services/outbox-service.ts` — DB outbox enqueue/claim/publish logic +- `src/services/outbox-dispatcher.ts` — DB→Stream polling bridge +- `src/services/__test__/outbox-service.test.ts` — outbox service tests +- `src/services/__test__/outbox-dispatcher.test.ts` — outbox dispatcher tests +- `src/bin/run-outbox-dispatcher.ts` — outbox-dispatcher CLI entrypoint +- `src/bin/run-billing-events-consumer.ts` — cache-sync consumer CLI entrypoint +- `src/schemas/outbox-events.ts` — Drizzle schema for outbox_events table + +### Files to MODIFY +- `src/services/billing-service.ts` — Remove outboxService dep, remove ledger/audit from transaction, add XADD after commit +- `src/services/billing-mq.ts` — Keep as-is (already has publish/consume/ack) +- `src/services/billing-mq-worker.ts` — Keep as-is (generic consumer loop) +- `src/services/billing-events.ts` — Add `llm.request.log` event type for request logging +- `src/app.ts` — Remove outboxService from DI, inject billingMqService into billing-service and v1completions +- `src/routes/v1completions.ts` — Replace fire-and-forget `requestLogService.logRequest()` with XADD +- `src/bin/run.ts` — Replace 3 commands with 2: `api` + `billing-consumer` +- `src/schemas/index.ts` — Remove outbox-events re-export +- `src/libs/env.ts` — Remove OUTBOX_DISPATCHER_* env vars +- `src/services/__test__/billing-service.test.ts` — Remove outbox assertions, add XADD mock assertions + +### Files to CREATE +- `src/bin/run-billing-consumer.ts` — New consumer entrypoint that handles all stream events +- `src/services/billing-consumer-handler.ts` — Message handler: routes events to ledger/audit/request-log DB writes +- `src/services/__test__/billing-consumer-handler.test.ts` — Tests for the new handler + +--- + +## Task 1: Remove outbox schema from exports and create DB migration + +**Files:** +- Modify: `src/schemas/index.ts:8` +- Create: new drizzle migration to drop `outbox_events` table + +- [ ] **Step 1: Remove outbox-events from schema index** + +In `src/schemas/index.ts`, remove line 8: +```ts +export * from './outbox-events' +``` + +- [ ] **Step 2: Generate drizzle migration to drop the outbox_events table** + +Run: +```bash +cd apps/server && pnpm drizzle-kit generate +``` + +Expected: A new migration file in `drizzle/` that drops the `outbox_events` table and its indexes. + +- [ ] **Step 3: Verify migration looks correct** + +Read the generated migration file and confirm it contains `DROP TABLE "outbox_events"` and drops the associated indexes. + +- [ ] **Step 4: Commit** + +```bash +git add apps/server/src/schemas/index.ts apps/server/drizzle/ +git commit -m "chore(server): drop outbox_events table from schema" +``` + +--- + +## Task 2: Delete outbox service, dispatcher, and their tests + +**Files:** +- Delete: `src/services/outbox-service.ts` +- Delete: `src/services/outbox-dispatcher.ts` +- Delete: `src/services/__test__/outbox-service.test.ts` +- Delete: `src/services/__test__/outbox-dispatcher.test.ts` +- Delete: `src/schemas/outbox-events.ts` + +- [ ] **Step 1: Delete the files** + +```bash +cd apps/server +rm src/services/outbox-service.ts +rm src/services/outbox-dispatcher.ts +rm src/services/__test__/outbox-service.test.ts +rm src/services/__test__/outbox-dispatcher.test.ts +rm src/schemas/outbox-events.ts +``` + +- [ ] **Step 2: Verify no remaining imports of deleted files** + +Search for any remaining imports: +```bash +grep -r "outbox-service\|outbox-dispatcher\|outbox-events" apps/server/src/ --include="*.ts" +``` + +Expected: Hits in `billing-service.ts`, `app.ts`, `run.ts`, `billing-service.test.ts` — these will be fixed in later tasks. + +- [ ] **Step 3: Commit** + +```bash +git add -A apps/server/src/services/outbox-service.ts apps/server/src/services/outbox-dispatcher.ts apps/server/src/services/__test__/outbox-service.test.ts apps/server/src/services/__test__/outbox-dispatcher.test.ts apps/server/src/schemas/outbox-events.ts +git commit -m "refactor(server): delete outbox service, dispatcher, and schema" +``` + +--- + +## Task 3: Add `llm.request.log` event type to billing-events + +**Files:** +- Modify: `src/services/billing-events.ts` +- Modify: `src/services/__test__/billing-events.test.ts` + +- [ ] **Step 1: Add the new event type and payload schema** + +In `src/services/billing-events.ts`: + +Add `literal('llm.request.log')` to `BillingEventTypeSchema`: +```ts +const BillingEventTypeSchema = union([ + literal('flux.debited'), + literal('flux.credited'), + literal('stripe.checkout.completed'), + literal('llm.request.completed'), + literal('llm.request.log'), +]) +``` + +Add the payload schema after `LlmRequestCompletedPayloadSchema`: +```ts +const LlmRequestLogPayloadSchema = object({ + model: pipe(string(), nonEmpty()), + status: number(), + durationMs: number(), + fluxConsumed: number(), + promptTokens: optional(number()), + completionTokens: optional(number()), +}) +``` + +Add the event type: +```ts +export type LlmRequestLogEvent = BillingEventEnvelope & { + eventType: 'llm.request.log' + payload: LlmRequestLogPayload +} + +type LlmRequestLogPayload = InferOutput +``` + +Add to the `BillingEvent` union: +```ts +export type BillingEvent + = | FluxDebitedEvent + | FluxCreditedEvent + | StripeCheckoutCompletedEvent + | LlmRequestCompletedEvent + | LlmRequestLogEvent +``` + +Add the case to `parseBillingEvent`: +```ts +case 'llm.request.log': + return { + ...parsedEnvelope, + eventType: 'llm.request.log', + payload: parse(LlmRequestLogPayloadSchema, parsedEnvelope.payload), + } +``` + +- [ ] **Step 2: Run existing billing-events tests** + +```bash +pnpm exec vitest run apps/server/src/services/__test__/billing-events.test.ts +``` + +Expected: PASS (new type doesn't break existing serialization/parsing) + +- [ ] **Step 3: Commit** + +```bash +git add apps/server/src/services/billing-events.ts +git commit -m "feat(server): add llm.request.log event type to billing events" +``` + +--- + +## Task 4: Rewrite billing-service to remove outbox and add XADD + +**Files:** +- Modify: `src/services/billing-service.ts` + +- [ ] **Step 1: Write the failing test for the new billing-service signature** + +In `src/services/__test__/billing-service.test.ts`, replace the entire file. The key changes: +- Remove `outboxService` dependency +- Add `billingMqService` mock (with `publish` method) +- Remove all `outboxEvents` assertions +- Assert `billingMqService.publish` is called with correct event data +- Keep ledger and audit writes IN the transaction (they're still valuable for consistency, and we'll extract them to MQ in a follow-up if needed) + +Actually, per the plan discussion: ledger and audit should be REMOVED from the transaction and moved to the consumer. The transaction should ONLY update `user_flux`. The XADD after commit carries the data for the consumer to write ledger + audit. + +Update `billing-service.test.ts`: + +```ts +import type Redis from 'ioredis' + +import type { Database } from '../../libs/db' +import type { BillingMqService } from '../billing-mq' +import type { createConfigKVService } from '../config-kv' + +import { eq } from 'drizzle-orm' +import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' + +import { mockDB } from '../../libs/mock-db' +import { createBillingService } from '../billing-service' + +import * as schema from '../../schemas' + +function createMockConfigKV(overrides: Record = {}): ReturnType { + const defaults: Record = { INITIAL_USER_FLUX: 100, FLUX_PER_CENT: 1, FLUX_PER_REQUEST: 1, ...overrides } + return { + get: vi.fn(async (key: string) => defaults[key]), + getOrThrow: vi.fn(async (key: string) => defaults[key]), + getOptional: vi.fn(async (key: string) => defaults[key] ?? null), + set: vi.fn(), + } as any +} + +function createMockRedis(): Redis { + const store = new Map() + return { + get: vi.fn(async (key: string) => store.get(key) ?? null), + set: vi.fn(async (key: string, value: string) => { store.set(key, value); return 'OK' }), + } as unknown as Redis +} + +function createMockBillingMq(): BillingMqService { + return { + stream: 'billing-events', + publish: vi.fn(async () => '1-0'), + ensureConsumerGroup: vi.fn(async () => true), + consume: vi.fn(async () => []), + claimIdleMessages: vi.fn(async () => []), + ack: vi.fn(async () => 1), + } +} + +describe('billingService', () => { + let db: Database + let redis: Redis + let billingMq: BillingMqService + let billingService: ReturnType + + beforeAll(async () => { + db = await mockDB(schema) + + await db.insert(schema.user).values({ + id: 'user-billing-1', + name: 'Billing User', + email: 'billing@example.com', + }) + }) + + beforeEach(async () => { + redis = createMockRedis() + billingMq = createMockBillingMq() + billingService = createBillingService(db, redis, billingMq, createMockConfigKV()) + + await db.delete(schema.fluxAuditLog) + await db.delete(schema.fluxLedger) + await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1')) + await db.delete(schema.stripeCheckoutSession).where(eq(schema.stripeCheckoutSession.stripeSessionId, 'sess-billing-1')) + + await db.insert(schema.stripeCheckoutSession).values({ + userId: 'user-billing-1', + stripeSessionId: 'sess-billing-1', + mode: 'payment', + status: 'complete', + paymentStatus: 'paid', + amountTotal: 500, + currency: 'usd', + fluxCredited: false, + }) + }) + + describe('creditFluxFromStripeCheckout', () => { + it('credits flux, records ledger + audit, and publishes events to stream', async () => { + const result = await billingService.creditFluxFromStripeCheckout({ + stripeEventId: 'stripe-evt-1', + userId: 'user-billing-1', + stripeSessionId: 'sess-billing-1', + amountTotal: 500, + currency: 'usd', + fluxAmount: 50, + }) + + expect(result).toEqual({ applied: true, balanceAfter: 50 }) + + const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1')) + expect(fluxRecord?.flux).toBe(50) + + // Verify ledger entry (still written in credit transactions for immediate consistency) + const ledgerRecords = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-billing-1')) + expect(ledgerRecords).toHaveLength(1) + expect(ledgerRecords[0]?.type).toBe('credit') + + // Verify audit log (still written in credit transactions) + const auditRecords = await db.select().from(schema.fluxAuditLog).where(eq(schema.fluxAuditLog.userId, 'user-billing-1')) + expect(auditRecords).toHaveLength(1) + + // 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) + + // Verify events published to stream (not outbox) + expect(billingMq.publish).toHaveBeenCalledTimes(2) + const calls = (billingMq.publish as any).mock.calls + expect(calls[0][0].eventType).toBe('flux.credited') + expect(calls[1][0].eventType).toBe('stripe.checkout.completed') + + // Verify Redis cache updated + expect(redis.set).toHaveBeenCalledWith('flux:user-billing-1', '50') + }) + + it('is idempotent when the checkout session was already credited', async () => { + await billingService.creditFluxFromStripeCheckout({ + stripeEventId: 'stripe-evt-1', + userId: 'user-billing-1', + stripeSessionId: 'sess-billing-1', + amountTotal: 500, + currency: 'usd', + fluxAmount: 50, + }) + + const second = await billingService.creditFluxFromStripeCheckout({ + stripeEventId: 'stripe-evt-1', + userId: 'user-billing-1', + stripeSessionId: 'sess-billing-1', + amountTotal: 500, + currency: 'usd', + fluxAmount: 50, + }) + + expect(second).toEqual({ applied: false }) + }) + }) + + describe('debitFlux', () => { + it('deducts balance via DB, publishes event to stream for async ledger/audit', async () => { + await db.insert(schema.userFlux).values({ userId: 'user-billing-1', flux: 100 }) + + const result = await billingService.debitFlux({ + userId: 'user-billing-1', + amount: 30, + requestId: 'req-1', + description: 'gpt-4', + }) + + expect(result).toEqual({ userId: 'user-billing-1', flux: 70 }) + + // Verify DB balance updated + const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1')) + expect(fluxRecord?.flux).toBe(70) + + // Verify NO ledger/audit written synchronously (moved to consumer) + const ledgerRecords = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-billing-1')) + expect(ledgerRecords).toHaveLength(0) + + const auditRecords = await db.select().from(schema.fluxAuditLog).where(eq(schema.fluxAuditLog.userId, 'user-billing-1')) + expect(auditRecords).toHaveLength(0) + + // Verify event published to stream + expect(billingMq.publish).toHaveBeenCalledTimes(1) + const publishedEvent = (billingMq.publish as any).mock.calls[0][0] + expect(publishedEvent.eventType).toBe('flux.debited') + expect(publishedEvent.payload.amount).toBe(30) + expect(publishedEvent.payload.balanceAfter).toBe(70) + + // Verify Redis cache updated + expect(redis.set).toHaveBeenCalledWith('flux:user-billing-1', '70') + }) + + it('throws 402 when balance is insufficient', async () => { + await db.insert(schema.userFlux).values({ userId: 'user-billing-1', flux: 5 }) + + await expect(billingService.debitFlux({ + userId: 'user-billing-1', + amount: 10, + })).rejects.toThrow('Insufficient flux') + + // Verify no side effects + const [fluxRecord] = await db.select().from(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1')) + expect(fluxRecord?.flux).toBe(5) + + expect(billingMq.publish).not.toHaveBeenCalled() + }) + }) + + describe('creditFlux', () => { + it('credits balance with ledger + audit + stream event', async () => { + const result = await billingService.creditFlux({ + userId: 'user-billing-1', + amount: 50, + description: 'Admin grant', + source: 'admin', + }) + + expect(result.balanceAfter).toBe(50) + expect(result.balanceBefore).toBe(0) + + // Verify stream event published + expect(billingMq.publish).toHaveBeenCalledTimes(1) + const publishedEvent = (billingMq.publish as any).mock.calls[0][0] + expect(publishedEvent.eventType).toBe('flux.credited') + }) + }) +}) +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +pnpm exec vitest run apps/server/src/services/__test__/billing-service.test.ts +``` + +Expected: FAIL — `createBillingService` still expects `outboxService` parameter. + +- [ ] **Step 3: Rewrite billing-service.ts** + +Replace `src/services/billing-service.ts` with: + +```ts +import type Redis from 'ioredis' + +import type { Database } from '../libs/db' +import type { RevenueMetrics } from '../libs/otel' +import type { BillingEvent } from './billing-events' +import type { BillingMqService } from './billing-mq' +import type { ConfigKVService } from './config-kv' + +import { useLogger } from '@guiiai/logg' +import { eq } from 'drizzle-orm' + +import { createPaymentRequiredError } from '../utils/error' +import { nanoid } from '../utils/id' +import { fluxRedisKey } from './flux' + +import * as fluxSchema from '../schemas/flux' +import * as fluxAuditSchema from '../schemas/flux-audit-log' +import * as fluxLedgerSchema from '../schemas/flux-ledger' +import * as stripeSchema from '../schemas/stripe' + +const logger = useLogger('billing-service') + +export function createBillingService( + db: Database, + redis: Redis, + billingMq: BillingMqService, + _configKV: ConfigKVService, + metrics?: RevenueMetrics | null, +) { + /** + * Update Redis cache after a successful DB transaction. + * Best-effort: cache loss is harmless since DB is the source of truth. + */ + async function updateRedisCache(userId: string, balance: number): Promise { + try { + await redis.set(fluxRedisKey(userId), String(balance)) + } + catch { + logger.withFields({ userId }).warn('Failed to update Redis cache after balance change') + } + } + + /** + * Publish a billing event to Redis Stream. + * Best-effort: the event carries data for async side effects (ledger, audit). + * If publish fails, the side effects are lost but the balance change is already committed. + */ + async function publishEvent(event: BillingEvent): Promise { + 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') + } + } + + return { + /** + * Debit flux from a user's balance. + * Only UPDATE user_flux in transaction (minimal lock). + * Ledger + audit are written async by the billing consumer via stream event. + */ + async debitFlux(input: { + userId: string + amount: number + requestId?: string + description?: string + }): Promise<{ userId: string, flux: number }> { + const result = await db.transaction(async (tx) => { + const [row] = await tx + .select({ flux: fluxSchema.userFlux.flux }) + .from(fluxSchema.userFlux) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + .for('update') + + if (!row) { + throw new Error(`No flux record for user ${input.userId}`) + } + + const balanceBefore = row.flux + if (balanceBefore < input.amount) { + metrics?.fluxInsufficientBalance.add(1) + throw createPaymentRequiredError('Insufficient flux') + } + + const balanceAfter = balanceBefore - input.amount + + 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 } + }) + + await updateRedisCache(input.userId, result.flux) + + // Publish event for async ledger + audit writes + await publishEvent({ + eventId: nanoid(), + eventType: 'flux.debited', + aggregateId: input.userId, + userId: input.userId, + requestId: input.requestId, + occurredAt: new Date().toISOString(), + schemaVersion: 1, + payload: { + amount: input.amount, + balanceAfter: result.flux, + source: input.description ?? 'LLM request', + }, + }) + + logger.withFields({ userId: input.userId, amount: input.amount, balance: result.flux }).log('Debited flux') + return { userId: result.userId, flux: result.flux } + }, + + /** + * Credit flux to a user's balance. + * Credits keep ledger + audit in the transaction for immediate consistency + * (low frequency, user expects to see the record right away). + */ + async creditFlux(input: { + userId: string + amount: number + requestId?: string + description: string + source: string + auditMetadata?: Record + }): Promise<{ balanceBefore: number, balanceAfter: number }> { + const result = await db.transaction(async (tx) => { + await tx.insert(fluxSchema.userFlux) + .values({ userId: input.userId, flux: 0 }) + .onConflictDoNothing({ target: fluxSchema.userFlux.userId }) + + const [row] = await tx + .select({ flux: fluxSchema.userFlux.flux }) + .from(fluxSchema.userFlux) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + .for('update') + + const balanceBefore = row!.flux + const balanceAfter = balanceBefore + input.amount + + await tx.update(fluxSchema.userFlux) + .set({ flux: balanceAfter, updatedAt: new Date() }) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + + await tx.insert(fluxLedgerSchema.fluxLedger).values({ + userId: input.userId, + type: 'credit', + amount: input.amount, + balanceBefore, + balanceAfter, + requestId: input.requestId, + description: input.description, + }) + + await tx.insert(fluxAuditSchema.fluxAuditLog).values({ + userId: input.userId, + type: 'addition', + amount: input.amount, + description: input.description, + metadata: input.auditMetadata, + }) + + return { balanceBefore, balanceAfter } + }) + + await updateRedisCache(input.userId, result.balanceAfter) + + 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, + }, + }) + + logger.withFields({ userId: input.userId, amount: input.amount, balance: result.balanceAfter }).log('Credited flux') + return result + }, + + /** + * Credit flux from a Stripe checkout session (one-time payment). + * Idempotent: checks fluxCredited flag before applying. + */ + async creditFluxFromStripeCheckout(input: { + stripeEventId: string + userId: string + stripeSessionId: string + amountTotal: number + currency: string | null + fluxAmount: number + }): Promise<{ applied: boolean, balanceAfter?: number }> { + const txResult = await db.transaction(async (tx) => { + const record = await tx.query.stripeCheckoutSession.findFirst({ + where: (table, { eq }) => eq(table.stripeSessionId, input.stripeSessionId), + }) + + if (!record || record.fluxCredited) { + return { applied: false } + } + + await tx.insert(fluxSchema.userFlux) + .values({ userId: input.userId, flux: 0 }) + .onConflictDoNothing({ target: fluxSchema.userFlux.userId }) + + const [currentFlux] = await tx + .select({ flux: fluxSchema.userFlux.flux }) + .from(fluxSchema.userFlux) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + .for('update') + + const balanceBefore = currentFlux!.flux + const balanceAfter = balanceBefore + input.fluxAmount + + await tx.update(fluxSchema.userFlux) + .set({ flux: balanceAfter, updatedAt: new Date() }) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + + await tx.update(stripeSchema.stripeCheckoutSession) + .set({ fluxCredited: true, updatedAt: new Date() }) + .where(eq(stripeSchema.stripeCheckoutSession.stripeSessionId, input.stripeSessionId)) + + const description = `Stripe payment ${input.currency?.toUpperCase() ?? 'UNKNOWN'} ${(input.amountTotal / 100).toFixed(2)}` + + await tx.insert(fluxLedgerSchema.fluxLedger).values({ + userId: input.userId, + type: 'credit', + amount: input.fluxAmount, + balanceBefore, + balanceAfter, + requestId: input.stripeEventId, + description, + }) + + await tx.insert(fluxAuditSchema.fluxAuditLog).values({ + userId: input.userId, + type: 'addition', + amount: input.fluxAmount, + description, + metadata: { + stripeEventId: input.stripeEventId, + stripeSessionId: input.stripeSessionId, + source: 'stripe.checkout.completed', + }, + }) + + return { applied: true, balanceAfter, balanceBefore } + }) + + if (txResult.applied && txResult.balanceAfter != null) { + await updateRedisCache(input.userId, txResult.balanceAfter) + + 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 { applied: txResult.applied, balanceAfter: txResult.balanceAfter } + }, + + /** + * Credit flux from a Stripe invoice payment (subscription). + * Idempotent: checks fluxCredited flag on the invoice record. + */ + async creditFluxFromInvoice(input: { + stripeEventId: string + userId: string + stripeInvoiceId: string + amountPaid: number + currency: string + fluxAmount: number + }): Promise<{ applied: boolean, balanceAfter?: number }> { + const txResult = await db.transaction(async (tx) => { + const record = await tx.query.stripeInvoice.findFirst({ + where: (table, { eq }) => eq(table.stripeInvoiceId, input.stripeInvoiceId), + }) + + if (!record || record.fluxCredited) { + return { applied: false } + } + + await tx.insert(fluxSchema.userFlux) + .values({ userId: input.userId, flux: 0 }) + .onConflictDoNothing({ target: fluxSchema.userFlux.userId }) + + const [currentFlux] = await tx + .select({ flux: fluxSchema.userFlux.flux }) + .from(fluxSchema.userFlux) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + .for('update') + + const balanceBefore = currentFlux!.flux + const balanceAfter = balanceBefore + input.fluxAmount + + await tx.update(fluxSchema.userFlux) + .set({ flux: balanceAfter, updatedAt: new Date() }) + .where(eq(fluxSchema.userFlux.userId, input.userId)) + + await tx.update(stripeSchema.stripeInvoice) + .set({ fluxCredited: true, updatedAt: new Date() }) + .where(eq(stripeSchema.stripeInvoice.stripeInvoiceId, input.stripeInvoiceId)) + + const description = `Subscription invoice ${input.currency.toUpperCase()} ${(input.amountPaid / 100).toFixed(2)}` + + await tx.insert(fluxLedgerSchema.fluxLedger).values({ + userId: input.userId, + type: 'credit', + amount: input.fluxAmount, + balanceBefore, + balanceAfter, + requestId: input.stripeEventId, + description, + }) + + await tx.insert(fluxAuditSchema.fluxAuditLog).values({ + userId: input.userId, + type: 'addition', + amount: input.fluxAmount, + description, + metadata: { + stripeEventId: input.stripeEventId, + stripeInvoiceId: input.stripeInvoiceId, + source: 'invoice.paid', + }, + }) + + return { applied: true, balanceAfter } + }) + + if (txResult.applied && txResult.balanceAfter != null) { + await updateRedisCache(input.userId, txResult.balanceAfter) + + 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 + }, + } +} + +export type BillingService = ReturnType +``` + +Key changes: +- Parameter 3 changed from `outboxService: OutboxService` to `billingMq: BillingMqService` +- `debitFlux` transaction only does `SELECT FOR UPDATE` + `UPDATE user_flux` (no more ledger/audit/outbox INSERTs) +- After commit, publishes `flux.debited` event to stream via `XADD` +- Credit methods keep ledger+audit in transaction (low frequency, user expects immediate visibility) +- Credits publish events to stream after commit instead of outbox + +- [ ] **Step 4: Run the test** + +```bash +pnpm exec vitest run apps/server/src/services/__test__/billing-service.test.ts +``` + +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add apps/server/src/services/billing-service.ts apps/server/src/services/__test__/billing-service.test.ts +git commit -m "refactor(server): replace outbox with direct XADD in billing-service" +``` + +--- + +## Task 5: Create billing-consumer-handler + +**Files:** +- Create: `src/services/billing-consumer-handler.ts` +- Create: `src/services/__test__/billing-consumer-handler.test.ts` + +- [ ] **Step 1: Write the failing test** + +Create `src/services/__test__/billing-consumer-handler.test.ts`: + +```ts +import type { Database } from '../../libs/db' +import type { BillingStreamMessage } from '../billing-mq' + +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 + let handler: ReturnType + + beforeAll(async () => { + db = await mockDB(schema) + handler = createBillingConsumerHandler(db) + + await db.insert(schema.user).values({ + id: 'user-consumer-1', + name: 'Consumer User', + email: 'consumer@example.com', + }) + await db.insert(schema.userFlux).values({ userId: 'user-consumer-1', flux: 70 }) + }) + + beforeEach(async () => { + await db.delete(schema.fluxLedger) + await db.delete(schema.fluxAuditLog) + await db.delete(schema.llmRequestLog) + }) + + it('writes ledger + audit for flux.debited events', async () => { + const message: BillingStreamMessage = { + streamMessageId: '1-0', + event: { + eventId: 'evt-1', + eventType: 'flux.debited', + aggregateId: 'user-consumer-1', + userId: 'user-consumer-1', + requestId: 'req-1', + occurredAt: new Date().toISOString(), + schemaVersion: 1, + payload: { amount: 30, balanceAfter: 70, source: 'gpt-4' }, + }, + } + + await handler.handleMessage(message) + + const ledger = await db.select().from(schema.fluxLedger).where(eq(schema.fluxLedger.userId, 'user-consumer-1')) + expect(ledger).toHaveLength(1) + expect(ledger[0]).toMatchObject({ + type: 'debit', + amount: 30, + balanceAfter: 70, + }) + + const audit = await db.select().from(schema.fluxAuditLog).where(eq(schema.fluxAuditLog.userId, 'user-consumer-1')) + expect(audit).toHaveLength(1) + expect(audit[0]).toMatchObject({ + type: 'consumption', + amount: -30, + }) + }) + + it('writes request log for llm.request.log events', async () => { + const message: BillingStreamMessage = { + streamMessageId: '2-0', + event: { + eventId: 'evt-2', + eventType: 'llm.request.log', + aggregateId: 'user-consumer-1', + userId: 'user-consumer-1', + occurredAt: new Date().toISOString(), + schemaVersion: 1, + payload: { model: 'gpt-4', status: 200, durationMs: 1500, fluxConsumed: 30, promptTokens: 100, completionTokens: 50 }, + }, + } + + await handler.handleMessage(message) + + const logs = await db.select().from(schema.llmRequestLog) + expect(logs).toHaveLength(1) + expect(logs[0]).toMatchObject({ + userId: 'user-consumer-1', + model: 'gpt-4', + status: 200, + durationMs: 1500, + fluxConsumed: 30, + }) + }) + + it('ignores flux.credited events (already handled synchronously)', async () => { + const message: BillingStreamMessage = { + streamMessageId: '3-0', + event: { + eventId: 'evt-3', + eventType: 'flux.credited', + aggregateId: 'user-consumer-1', + userId: 'user-consumer-1', + occurredAt: new Date().toISOString(), + schemaVersion: 1, + payload: { amount: 50, balanceAfter: 120, source: 'admin' }, + }, + } + + await handler.handleMessage(message) + + // No additional ledger/audit writes (credit already wrote them synchronously) + const ledger = await db.select().from(schema.fluxLedger) + expect(ledger).toHaveLength(0) + }) +}) +``` + +- [ ] **Step 2: Run test to verify it fails** + +```bash +pnpm exec vitest run apps/server/src/services/__test__/billing-consumer-handler.test.ts +``` + +Expected: FAIL — module not found + +- [ ] **Step 3: Implement billing-consumer-handler.ts** + +Create `src/services/billing-consumer-handler.ts`: + +```ts +import type { Database } from '../libs/db' +import type { BillingStreamMessage } from './billing-mq' + +import { useLogger } from '@guiiai/logg' + +import * as fluxAuditSchema from '../schemas/flux-audit-log' +import * as fluxLedgerSchema from '../schemas/flux-ledger' +import * as llmRequestLogSchema from '../schemas/llm-request-log' + +const logger = useLogger('billing-consumer-handler').useGlobalConfig() + +export function createBillingConsumerHandler(db: Database) { + return { + async handleMessage(message: BillingStreamMessage): Promise { + const { event } = message + + switch (event.eventType) { + case 'flux.debited': { + const balanceBefore = event.payload.balanceAfter != null + ? event.payload.balanceAfter + event.payload.amount + : 0 + + await db.insert(fluxLedgerSchema.fluxLedger).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.source ?? 'LLM request', + }) + + await db.insert(fluxAuditSchema.fluxAuditLog).values({ + userId: event.userId, + type: 'consumption', + amount: -event.payload.amount, + description: event.payload.source ?? 'LLM request', + }) + + logger.withFields({ + eventId: event.eventId, + userId: event.userId, + amount: event.payload.amount, + }).log('Wrote debit ledger + audit') + break + } + + case 'llm.request.log': { + await db.insert(llmRequestLogSchema.llmRequestLog).values({ + 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, + }) + + 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 +``` + +- [ ] **Step 4: Run test** + +```bash +pnpm exec vitest run apps/server/src/services/__test__/billing-consumer-handler.test.ts +``` + +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add apps/server/src/services/billing-consumer-handler.ts apps/server/src/services/__test__/billing-consumer-handler.test.ts +git commit -m "feat(server): add billing consumer handler for async ledger/audit/log writes" +``` + +--- + +## Task 6: Replace request log fire-and-forget with XADD in v1completions + +**Files:** +- Modify: `src/routes/v1completions.ts` + +- [ ] **Step 1: Update v1completions to accept billingMqService and publish llm.request.log events** + +In `src/routes/v1completions.ts`: + +1. Add `BillingMqService` to imports and function signature: +```ts +import type { BillingMqService } from '../services/billing-mq' +``` + +Change the function signature to: +```ts +export function createV1CompletionsRoutes( + fluxService: FluxService, + billingService: BillingService, + configKV: ConfigKVService, + requestLogService: RequestLogService, + billingMq: BillingMqService, + llm: LlmMetrics | null, +) +``` + +2. Replace every `requestLogService.logRequest({...}).catch(...)` call (lines 177-185, 218-226, 281-287, 346-352) with a `publishRequestLog` helper: + +Add helper at the top of the function body: +```ts +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')) +} +``` + +Replace each fire-and-forget call. For example, line 177-185 becomes: +```ts +publishRequestLog({ + userId: user.id, + model: requestModel, + status: response.status, + durationMs, + fluxConsumed: actualCharged, + promptTokens: usage.promptTokens, + completionTokens: usage.completionTokens, +}) +``` + +Do the same for all four locations (streaming chat, non-streaming chat, TTS, ASR). + +- [ ] **Step 2: Update app.ts to pass billingMq to v1completions** + +In `src/app.ts`, the `billingMq` service needs to be created and passed. Add to the injeca DI setup: + +```ts +const billingMqService = injeca.provide('services:billingMq', { + dependsOn: { redis, env: parsedEnv }, + build: ({ dependsOn }) => createBillingMqService(dependsOn.redis, { + stream: dependsOn.env.BILLING_EVENTS_STREAM, + }), +}) +``` + +Add `createBillingMqService` import: +```ts +import { createBillingMqService } from './services/billing-mq' +``` + +Update the `billingService` provider to use `billingMqService` instead of `outboxService`: +```ts +const billingService = injeca.provide('services:billing', { + dependsOn: { db, redis, billingMqService, configKV, otel }, + build: ({ dependsOn }) => createBillingService(dependsOn.db, dependsOn.redis, dependsOn.billingMqService, dependsOn.configKV, dependsOn.otel?.revenue), +}) +``` + +Remove the `outboxService` provider entirely. + +Remove the import: +```ts +// DELETE: import { createOutboxService } from './services/outbox-service' +``` + +Add `billingMqService` to the resolve and pass to `buildApp`: +- Add to `AppDeps` interface and `buildApp` params +- Pass to `createV1CompletionsRoutes`: +```ts +.route('/api/v1', createV1CompletionsRoutes(fluxService, billingService, configKV, requestLogService, billingMqService, otel?.llm ?? null)) +``` + +- [ ] **Step 3: Verify typecheck passes** + +```bash +cd apps/server && pnpm typecheck +``` + +Expected: PASS (no type errors) + +- [ ] **Step 4: Commit** + +```bash +git add apps/server/src/routes/v1completions.ts apps/server/src/app.ts +git commit -m "refactor(server): replace fire-and-forget request logging with XADD" +``` + +--- + +## Task 7: Rewrite bin/ entrypoints — consolidate to api + billing-consumer + +**Files:** +- Create: `src/bin/run-billing-consumer.ts` +- Modify: `src/bin/run.ts` +- Delete: `src/bin/run-billing-events-consumer.ts` +- Delete: `src/bin/run-outbox-dispatcher.ts` + +- [ ] **Step 1: Create run-billing-consumer.ts** + +```ts +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 { createRedis } from '../libs/redis' +import { createBillingConsumerHandler } from '../services/billing-consumer-handler' +import { createBillingMqService } from '../services/billing-mq' +import { createBillingMqWorker } from '../services/billing-mq-worker' + +function parsePositiveInteger(rawValue: string, envKey: string): number { + const parsed = Number(rawValue) + if (!Number.isInteger(parsed) || parsed <= 0) { + throw new Error(`${envKey} must be a positive integer`) + } + + return parsed +} + +export async function runBillingConsumer(): Promise { + 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.DATABASE_URL) + + 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 = createBillingMqService(redis, { + stream: env.BILLING_EVENTS_STREAM, + }) + + const handler = createBillingConsumerHandler(db) + const worker = createBillingMqWorker(mq) + + await worker.run({ + group: 'billing-consumer', + consumer, + signal: abortController.signal, + batchSize: parsePositiveInteger(env.BILLING_EVENTS_BATCH_SIZE, 'BILLING_EVENTS_BATCH_SIZE'), + blockMs: parsePositiveInteger(env.BILLING_EVENTS_BLOCK_MS, 'BILLING_EVENTS_BLOCK_MS'), + minIdleTimeMs: parsePositiveInteger(env.BILLING_EVENTS_MIN_IDLE_MS, 'BILLING_EVENTS_MIN_IDLE_MS'), + onMessage: message => handler.handleMessage(message), + }) + } + finally { + await redis.quit() + await pool.end() + } +} +``` + +- [ ] **Step 2: Rewrite run.ts with 2 commands** + +```ts +#!/usr/bin/env node + +import process from 'node:process' + +import { pathToFileURL } from 'node:url' + +import { errorMessageFrom } from '@moeru/std' +import { cac } from 'cac' + +import { runApiServer } from '../app' +import { runBillingConsumer } from './run-billing-consumer' + +const serverRoles = ['api', 'billing-consumer'] as const + +type ServerRole = typeof serverRoles[number] + +async function runServerRole(role: ServerRole): Promise { + switch (role) { + case 'api': + await runApiServer() + return + case 'billing-consumer': + await runBillingConsumer() + } +} + +export function createServerCli() { + const cli = cac('server') + + cli + .usage('') + .command('api', 'Start the HTTP/WebSocket API process') + .action(() => runServerRole('api')) + + cli + .command('billing-consumer', 'Start the billing events consumer (ledger, audit, request logs)') + .action(() => runServerRole('billing-consumer')) + + 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 { + const cli = createServerCli() + cli.parse(process.argv, { run: false }) + + if (!cli.matchedCommand) { + cli.outputHelp() + process.exitCode = 1 + return + } + + await cli.runMatchedCommand() +} + +function isExecutedAsMainModule(): boolean { + const entryFile = process.argv[1] + if (!entryFile) { + return false + } + + return import.meta.url === pathToFileURL(entryFile).href +} + +if (isExecutedAsMainModule()) { + void main().catch((error: unknown) => { + process.stderr.write(`${errorMessageFrom(error) ?? 'Unknown error'}\n`) + process.exit(1) + }) +} +``` + +- [ ] **Step 3: Delete old entrypoints** + +```bash +cd apps/server +rm src/bin/run-billing-events-consumer.ts +rm src/bin/run-outbox-dispatcher.ts +``` + +- [ ] **Step 4: Commit** + +```bash +git add apps/server/src/bin/ +git commit -m "refactor(server): consolidate 3 processes into api + billing-consumer" +``` + +--- + +## Task 8: Clean up env vars and remove OUTBOX_DISPATCHER config + +**Files:** +- Modify: `src/libs/env.ts` + +- [ ] **Step 1: Remove outbox dispatcher env vars from env.ts** + +Remove these lines from the env schema: +```ts +OUTBOX_DISPATCHER_BATCH_SIZE: optional(string(), '10'), +OUTBOX_DISPATCHER_CLAIM_TTL_MS: optional(string(), '30000'), +OUTBOX_DISPATCHER_POLL_MS: optional(string(), '1000'), +OUTBOX_DISPATCHER_NAME: optional(string()), +``` + +- [ ] **Step 2: Verify typecheck** + +```bash +cd apps/server && pnpm typecheck +``` + +Expected: PASS + +- [ ] **Step 3: Commit** + +```bash +git add apps/server/src/libs/env.ts +git commit -m "chore(server): remove outbox dispatcher env vars" +``` + +--- + +## Task 9: Run full test suite and fix any remaining issues + +- [ ] **Step 1: Run all server tests** + +```bash +pnpm exec vitest run apps/server/ +``` + +Expected: All tests PASS. + +- [ ] **Step 2: Run typecheck** + +```bash +cd apps/server && pnpm typecheck +``` + +Expected: PASS + +- [ ] **Step 3: Run lint** + +```bash +pnpm lint:fix +``` + +Expected: PASS with auto-fixes applied. + +- [ ] **Step 4: Final commit if lint made changes** + +```bash +git add -A apps/server/ +git commit -m "chore(server): fix lint after outbox removal refactor" +``` + +--- + +## Task 10: Update Dockerfile and documentation + +**Files:** +- Modify: `apps/server/Dockerfile` +- Modify: `apps/server/production/railway/Dockerfile` +- Modify: `apps/server/README.md` (if exists) + +- [ ] **Step 1: Update Dockerfiles** + +Check if Dockerfiles reference `outbox-dispatcher` or `cache-sync-consumer` commands and update them to only use `api` or `billing-consumer`. + +- [ ] **Step 2: Update any documentation referencing the 3-process architecture** + +Search the repo for references to `outbox-dispatcher`, `cache-sync-consumer`, `outbox` in docs/README files and update them to reflect the new 2-process architecture. + +- [ ] **Step 3: Commit** + +```bash +git add apps/server/Dockerfile apps/server/production/railway/Dockerfile +git commit -m "docs(server): update Dockerfiles and docs for 2-process architecture" +```