moeka-project/apps/server/docs/ai-context/workers-and-runtime.md

Workers And Runtime

进程角色

统一入口在 src/bin/run.ts：

• api
  • 启动 Hono HTTP + WebSocket 服务
• billing-consumer
  • 消费 Redis Stream billing-events，异步将 ledger、audit log、LLM 请求日志写入 DB

这两个角色是当前服务端部署拆分的基本单位。

API 角色

启动路径：

• src/bin/run.ts
• runApiServer()
• createApp()

启动时会做的事情：

• 解析 env
• 初始化日志
• 可选初始化 OTel
• 连接 Postgres / Redis
• 跑数据库迁移
• 装配服务
• 启动 HTTP server
• 注入 WebSocket

Billing Consumer

实现位置：

• 入口：src/bin/run-billing-consumer.ts
• worker：src/services/billing-mq-worker.ts
• stream adapter：src/services/billing-mq.ts

工作流程：

1. 以 consumer group 模式消费 Redis Stream billing-events
2. 根据事件类型分发处理：
   • flux.debited — 写 flux_ledger 和 flux_audit_log
   • llm.request.log — 写 llm_request_log
3. 处理成功后 ACK；handler 抛错时不 ACK，消息保持 pending 等待重试

相关环境变量：

• BILLING_EVENTS_STREAM
• BILLING_EVENTS_CONSUMER_NAME
• BILLING_EVENTS_BATCH_SIZE
• BILLING_EVENTS_BLOCK_MS
• BILLING_EVENTS_MIN_IDLE_MS

Redis Streams 语义

billing-mq.ts 把 Redis Streams 抽象成：

• publish()
• ensureConsumerGroup()
• consume()
• claimIdleMessages()
• ack()

这层约束了消息处理语义：

• 使用 consumer group
• 使用 pending reclaim
• handler 抛错时不 ack，消息保持 pending

因此新增新的 stream consumer 时，最安全的方式通常是复用这层，不要自己裸写 XREADGROUP。

聊天 WebSocket 运行时

src/routes/chat-ws.ts 还有一套独立于 Redis Streams 的运行时机制：

• 同实例连接保存在进程内 Map
• 跨实例 fan-out 通过 Redis Pub/Sub

这意味着：

• WS 广播不具备持久化和重放能力
• 真正补齐消息还是靠 pullMessages
• 广播只是为了降低拉取延迟，不代表存在旧式 sync 端点

如果要改 Redis key / channel 构造、Pub/Sub payload 或 Streams 边界，先看 redis-boundaries-and-pubsub.md。

OpenTelemetry

初始化在 src/libs/otel.ts。

启用条件：

• OTEL_EXPORTER_OTLP_ENDPOINT 存在

覆盖面：

• HTTP
• Auth
• Chat engagement
• Revenue
• LLM
• DB / Redis instrumentation

重要实现细节：

• sdk.start() 必须发生在 metrics.getMeter() 之前
• /health 会被 HTTP instrumentation 忽略

环境变量分层

基础运行

• HOST
• PORT
• API_SERVER_URL
• DATABASE_URL
• REDIS_URL

Auth

• AUTH_GOOGLE_CLIENT_ID
• AUTH_GOOGLE_CLIENT_SECRET
• AUTH_GITHUB_CLIENT_ID
• AUTH_GITHUB_CLIENT_SECRET

Stripe

• STRIPE_SECRET_KEY
• STRIPE_WEBHOOK_SECRET

Billing MQ

• BILLING_EVENTS_STREAM
• BILLING_EVENTS_CONSUMER_NAME
• BILLING_EVENTS_BATCH_SIZE
• BILLING_EVENTS_BLOCK_MS
• BILLING_EVENTS_MIN_IDLE_MS

OTel

• OTEL_SERVICE_NAMESPACE
• OTEL_SERVICE_NAME
• OTEL_TRACES_SAMPLING_RATIO
• OTEL_EXPORTER_OTLP_ENDPOINT
• OTEL_EXPORTER_OTLP_HEADERS
• OTEL_DEBUG

运行时修改建议

如果你要改：

• 新增 worker
  • 先看 run.ts 的角色模型和 billing-mq-worker.ts
• 改事件分发
  • 先看 billing-consumer handler，在 billing-mq-worker.ts 中增加新的事件处理分支
• 改聊天同步
  • 先区分“持久化消息”与“广播通知”两层
• 改部署限流
  • 注意当前 rate-limit.ts 仍是单实例内存模型