feat(server): reimplement message queue, drop outbox

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

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`。
 

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
 

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` 完全重叠，后者更像旧名残留。

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:<userId>`
-
-这说明当前 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` 中增加新的事件处理分支
 - 改聊天同步
   - 先区分“持久化消息”与“广播通知”两层
 - 改部署限流

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";

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": "",

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
     }
   ]

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<typeof createAuth>
-type CharacterService = ReturnType<typeof createCharacterService>
-type ChatService = ReturnType<typeof createChatService>
-type ProviderService = ReturnType<typeof createProviderService>
-type FluxService = ReturnType<typeof createFluxService>
-type ConfigKVService = ReturnType<typeof createConfigKVService>
-type RequestLogService = ReturnType<typeof createRequestLogService>
-type StripeDBService = ReturnType<typeof createStripeService>
-type FluxAuditService = ReturnType<typeof createFluxAuditService>
-type BillingService = ReturnType<typeof createBillingService>
-
 interface AppDeps {
-  auth: AuthService
+  auth: ReturnType<typeof createAuth>
   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<HonoEnv>()
@@ -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,

apps/server/src/bin/__test__/run-billing-events-consumer.test.ts

@@ -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()
-  })
-})

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',
     ]))
   })
 })

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<void> {
+export async function runBillingConsumer(): Promise<void> {
   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<void> {
   )
 
   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<void> {
   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 {

apps/server/src/bin/run-billing-events-consumer.ts

@@ -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<typeof createRedis>) => Promise<void>
-}
-
-export async function runBillingEventsConsumer(options: RunBillingEventsConsumerOptions): Promise<void> {
-  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<typeof createRedis>,
-): Promise<void> {
-  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))
-  }
-}

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<void> {
     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()
 

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'),

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<string, any> = {}): 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<HonoEnv>()
 
   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', {

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,

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'

apps/server/src/schemas/outbox-events.ts

@@ -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<typeof outboxEvents>
-export type NewOutboxEvent = InferInsertModel<typeof outboxEvents>

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<typeof createOutboxService>
+  let billingMq: BillingMqService
   let billingService: ReturnType<typeof createBillingService>
 
   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' }))
     })
   })
 })

apps/server/src/services/__test__/outbox-dispatcher.test.ts

@@ -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()
-  })
-})

apps/server/src/services/__test__/outbox-service.test.ts

@@ -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<typeof createOutboxService>
-
-  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()
-  })
-})

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<void> {
+      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<typeof createBillingConsumerHandler>

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<typeof BillingEventEnvelopeSchema>
 type BalanceChangePayload = InferOutput<typeof BalanceChangePayloadSchema>
 type StripeCheckoutCompletedPayload = InferOutput<typeof StripeCheckoutCompletedPayloadSchema>
 type LlmRequestCompletedPayload = InferOutput<typeof LlmRequestCompletedPayloadSchema>
+type LlmRequestLogPayload = InferOutput<typeof LlmRequestLogPayloadSchema>
 
 export type FluxDebitedEvent = BillingEventEnvelope & {
   eventType: 'flux.debited'
@@ -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<string, string | undefined> {
   event_id: string
@@ -153,5 +170,11 @@ export function parseBillingEvent(fields: Record<string, string | undefined>): 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),
+      }
   }
 }

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<void> {
+    try {
+      await billingMq.publish(event)
+    }
+    catch (error) {
+      logger.withError(error).withFields({
+        eventId: event.eventId,
+        eventType: event.eventType,
+        userId: event.userId,
+      }).error('Failed to publish billing event to stream')
+    }
+  }
+
   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

apps/server/src/services/outbox-dispatcher.ts

@@ -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<number> {
-      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<void> {
-      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<void> {
-  if (signal.aborted) {
-    return
-  }
-
-  await new Promise<void>((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<typeof createOutboxDispatcher>

apps/server/src/services/outbox-service.ts

@@ -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<void> {
-      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<ClaimedOutboxEvent[]> {
-      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<void> {
-      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<void> {
-      await db.update(outboxEvents)
-        .set({
-          claimedBy: null,
-          claimExpiresAt: null,
-          updatedAt: new Date(),
-        })
-        .where(eq(outboxEvents.id, id))
-    },
-  }
-}
-
-export type OutboxService = ReturnType<typeof createOutboxService>