refactor(server): split services into domain/adapter layers, drop dead code
Why
- src/services/ was an unordered mix of single-file services and module
directories with no shared classification axis, plus several long-dead
admin batch helpers that survived the move to the simpler synchronous
admin-flux-grants flow.
What
- services/ now has two top-level layers:
domain/ — DB state + business rules (billing, characters, chats,
flux, flux-transaction, llm-router, providers, request-log,
stripe, user-deletion, admin/{flux-grants,router-config})
adapters/ — thin wrappers over external SDKs / infra (config-kv, email,
posthog, tts/)
- admin/* moved under domain/admin/ with consistent plural names
(flux-grants, router-config).
- tts-adapters/ collapsed to adapters/tts/ (no redundant -adapters suffix
once nested under adapters/).
- 63 src files + scripts/e2e-llm-router.ts + tests/verifications/_harness.ts
had relative imports rewritten; git mv preserves blame.
- apps/server/CLAUDE.md and docs/ai-context/*.md updated to match new paths.
Dead code removed
- services/admin-flux-grant-batches/ (service + worker + tests, 1090 LOC) —
superseded by admin-flux-grants and never wired into app.ts.
- routes/admin/flux-grant-batches/ — same.
- utils/redis-compressed.ts + test — zero production call sites.
- llm-router/index.ts re-exports trimmed from 26 to 6; only symbols with
external consumers are kept.
Intentionally kept
- schemas/flux-grant-batch.ts and its schemas/index.ts export remain so the
drizzle-kit generate diff stays empty. Removing them is a separate PR
that owns the drop-table migration for flux_grant_batch /
flux_grant_batch_recipient.
Verification
- pnpm -F @proj-airi/server typecheck: passes.
- pnpm exec eslint apps/server: 49 errors, identical to main baseline
(all are pre-existing node/prefer-global/buffer in envelope-crypto and
scripts/e2e-llm-router; untouched by this change).
- Vitest passes per-file; the 6 mockDB hook timeouts under full-parallel
run are the known pushSchema-per-worker infra cost, not a regression.
This commit is contained in:
RainbowBird
@@ -49,7 +49,7 @@ Local observability: `docker compose -f apps/server/docker-compose.otel.yml up -
|
||||
|
||||
- **Flux read/write separation**: `FluxService` reads (Redis cache-aside), `BillingService` writes (single Postgres tx that mutates `user_flux` and writes the matching `flux_transaction` ledger row). Never put write-balance logic in `flux.ts`.
|
||||
- **No async billing pipeline**: debits and credits update balance + ledger in one transaction. The `(user_id, request_id)` partial unique index gives DB-level idempotency for retries; LLM `request log` rows are written best-effort right after the response is delivered.
|
||||
- **In-process LLM/TTS router**: `/api/v1/openai` is dispatched by `services/llm-router` reading `LLM_ROUTER_CONFIG` (per-model upstream chain + envelope-encrypted keys). `chat/completions` walks LLM upstreams with key fallback; `audio/speech` delegates to a TTS adapter (`azure` / `dashscope-cosyvoice` / `volcengine`); `audio/voices` returns the adapter's compiled-in catalog. Server handles auth/billing/logging, not model execution.
|
||||
- **In-process LLM/TTS router**: `/api/v1/openai` is dispatched by `services/domain/llm-router` reading `LLM_ROUTER_CONFIG` (per-model upstream chain + envelope-encrypted keys). `chat/completions` walks LLM upstreams with key fallback; `audio/speech` delegates to a TTS adapter (`azure` / `dashscope-cosyvoice` / `volcengine`); `audio/voices` returns the adapter's compiled-in catalog. Server handles auth/billing/logging, not model execution.
|
||||
- **Redis is cache + pub/sub, not truth**: balance cache, app_settings read cache, WebSocket cross-instance pub/sub. Truth is always Postgres.
|
||||
- **Auth**: Better Auth + OIDC. `sessionMiddleware` fills context but doesn't block; `authGuard` returns 401.
|
||||
- **Multi-instance safe**: all writes go through Postgres transactions; cross-instance messaging uses Redis Pub/Sub. No async work, no in-process singletons — admin flux grants happen synchronously inside the POST that triggered them.
|
||||
|
||||
@@ -62,7 +62,7 @@ auth ──depends on──► userDeletionService ──depends on──► [st
|
||||
auth 和业务 service **互不依赖**，双方都只依赖 `userDeletionService` 这层抽象。这是 DIP 的标准形态。
|
||||
|
||||
```ts
|
||||
// apps/server/src/services/user-deletion/types.ts
|
||||
// apps/server/src/services/domain/user-deletion/types.ts
|
||||
export interface UserDeletionHandler {
|
||||
name: string
|
||||
/** Lower runs first. 10=external side-effects, 20=financial+cache, 30=pure DB */
|
||||
@@ -101,11 +101,11 @@ export interface UserDeletionService {
|
||||
|
||||
**所有读业务表的查询都必须加 `isNull(deletedAt)` 过滤**，否则被删用户的数据还能被列出来 / 关联出来。重点扫描：
|
||||
|
||||
- `apps/server/src/services/flux.ts` — getBalance / readBalance
|
||||
- `apps/server/src/services/characters.ts` — listCharacters
|
||||
- `apps/server/src/services/providers.ts` — listProviderConfigs
|
||||
- `apps/server/src/services/chats.ts` — listChats / listMessages
|
||||
- `apps/server/src/services/billing/billing-service.ts` — invoice / sub 查询
|
||||
- `apps/server/src/services/domain/flux.ts` — getBalance / readBalance
|
||||
- `apps/server/src/services/domain/characters.ts` — listCharacters
|
||||
- `apps/server/src/services/domain/providers.ts` — listProviderConfigs
|
||||
- `apps/server/src/services/domain/chats.ts` — listChats / listMessages
|
||||
- `apps/server/src/services/domain/billing/billing-service.ts` — invoice / sub 查询
|
||||
|
||||
写完后用 `pnpm typecheck` + grep `from(flux|character|chats|providers|stripe)` 兜底。
|
||||
|
||||
@@ -158,8 +158,8 @@ better-auth `internalAdapter.deleteAccounts` 删本地 `account` 表（user 跟
|
||||
- 业务表 schema: `apps/server/src/schemas/{flux,flux-transaction,stripe,characters,user-character,providers,chats}.ts`
|
||||
- Auth schema (不改): `apps/server/src/schemas/accounts.ts`
|
||||
- Auth 配置: `apps/server/src/libs/auth.ts` (extend with `user.deleteUser`)
|
||||
- Email service: `apps/server/src/services/email.ts` (extend interface + Resend impl)
|
||||
- Deletion scheduler: `apps/server/src/services/user-deletion/` (registry only, no domain logic)
|
||||
- Email service: `apps/server/src/services/adapters/email.ts` (extend interface + Resend impl)
|
||||
- Deletion scheduler: `apps/server/src/services/domain/user-deletion/` (registry only, no domain logic)
|
||||
- 各 service 自己的 `deleteAllForUser`: `apps/server/src/services/{characters,chats,flux,providers,stripe}.ts`
|
||||
- UI - settings page: `packages/stage-pages/src/pages/settings/account/account-settings-page.vue` (line ~430 TODO)
|
||||
- UI - confirmation page (新): `apps/ui-server-auth/src/pages/delete-account.vue`
|
||||
|
||||
@@ -87,8 +87,8 @@ SELECT user_id, amount, balance_after, created_at
|
||||
## 6. 实现位置
|
||||
|
||||
- 路由：[`apps/server/src/routes/admin/flux-grants/index.ts`](apps/server/src/routes/admin/flux-grants/index.ts)
|
||||
- Service：[`apps/server/src/services/admin-flux-grants/index.ts`](apps/server/src/services/admin-flux-grants/index.ts)
|
||||
- 单测：[`apps/server/src/services/admin-flux-grants/tests/admin-flux-grants.test.ts`](apps/server/src/services/admin-flux-grants/tests/admin-flux-grants.test.ts)
|
||||
- Service：[`apps/server/src/services/domain/admin/flux-grants/index.ts`](apps/server/src/services/domain/admin/flux-grants/index.ts)
|
||||
- 单测：[`apps/server/src/services/domain/admin/flux-grants/tests/admin-flux-grants.test.ts`](apps/server/src/services/domain/admin/flux-grants/tests/admin-flux-grants.test.ts)
|
||||
- adminGuard：[`apps/server/src/middlewares/admin-guard.ts`](apps/server/src/middlewares/admin-guard.ts)
|
||||
- 数据库：**没有**专门的表；唯一持久化是 `flux_transaction` ledger
|
||||
- 已废弃：`flux_grant_batch` / `flux_grant_batch_recipient`（drizzle migration `0011_superb_lady_deathstrike.sql` 删表）
|
||||
|
||||
@@ -140,10 +140,10 @@ CLI 入口在 `src/bin/run.ts`，只有一种角色：
|
||||
|
||||
### LLM/TTS 路由在进程内，而不是本地 provider 编排
|
||||
|
||||
`/api/v1/openai` 由 `services/llm-router` 读取 `LLM_ROUTER_CONFIG` 后按 upstream 链路 + key rotator 直接调 provider（OpenRouter、Azure Speech、阿里云 DashScope、火山引擎 等），不再依赖外部 knoway sidecar。因此：
|
||||
`/api/v1/openai` 由 `services/domain/llm-router` 读取 `LLM_ROUTER_CONFIG` 后按 upstream 链路 + key rotator 直接调 provider（OpenRouter、Azure Speech、阿里云 DashScope、火山引擎 等），不再依赖外部 knoway sidecar。因此：
|
||||
|
||||
- 服务端关心的是鉴权、限流、计费、日志、观测、上游路由与 key 健康
|
||||
- 具体模型协议翻译由 `services/llm-router` 与 `services/tts-adapters` 的 adapter 完成
|
||||
- 具体模型协议翻译由 `services/domain/llm-router` 与 `services/adapters/tts` 的 adapter 完成
|
||||
|
||||
### Redis 有多种职责，但都不是余额真相源
|
||||
|
||||
@@ -160,4 +160,4 @@ Redis 在这里同时承担：
|
||||
## 当前值得注意的实现信号
|
||||
|
||||
- `/api/v1/openai` 当前开放：`POST /chat/completions`、`POST /chat/completion`、`POST /audio/speech`、`GET /audio/voices`。`handleTranscription` 路由尚未挂载。
|
||||
- `flux_grant_batch` schema / service / route 已被简化版 `admin-flux-grants` 取代，但旧的 `src/schemas/flux-grant-batch.ts`、`src/services/admin-flux-grant-batch/`、`src/routes/admin/flux-grant-batches/` 仍以 dead code 形态残留在仓库里，没有在 `app.ts` 装配。改这块前直接删旧文件，不要继续往里面加东西。
|
||||
- `flux_grant_batch` schema 已被简化版 `admin-flux-grants` 取代。代码层（service / route / worker / tests）已删。但 `src/schemas/flux-grant-batch.ts` 仍在并跟随 `schemas/index.ts` 导出，对应生产 DB 表 `flux_grant_batch` / `flux_grant_batch_recipient` 也仍在。下次清理要删 schema 文件 + 生成 drop table migration，属破坏性 DDL，需单独 PR 处理。
|
||||
|
||||
@@ -53,7 +53,7 @@ TTS 字符、STT 秒等单价 < 1 Flux 的服务通过 `FluxMeter` 累计零头
|
||||
|
||||
## 关键服务
|
||||
|
||||
### BillingService (`services/billing/billing-service.ts`)
|
||||
### BillingService (`services/domain/billing/billing-service.ts`)
|
||||
|
||||
所有余额写操作的唯一入口：
|
||||
|
||||
@@ -62,7 +62,7 @@ TTS 字符、STT 秒等单价 < 1 Flux 的服务通过 `FluxMeter` 累计零头
|
||||
- **`creditFluxFromStripeCheckout()`** — Stripe 一次性支付充值，按 session 幂等
|
||||
- **`creditFluxFromInvoice()`** — Stripe 订阅发票充值，按 invoice 幂等
|
||||
|
||||
### FluxService (`services/flux.ts`)
|
||||
### FluxService (`services/domain/flux.ts`)
|
||||
|
||||
只负责读操作：
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
|
||||
### 单一真相源
|
||||
|
||||
`src/services/config-kv.ts` 中的 `ConfigEntrySchemas` 是以下三件事的单一真相源：
|
||||
`src/services/adapters/config-kv.ts` 中的 `ConfigEntrySchemas` 是以下三件事的单一真相源：
|
||||
|
||||
- 配置值的运行时校验
|
||||
- 配置值的默认值
|
||||
|
||||
@@ -27,7 +27,7 @@ Last updated: 2026-04-27
|
||||
|
||||
In:
|
||||
|
||||
- `apps/server/src/services/email.ts`：统一 `EmailService` 接口（`sendVerification` / `sendPasswordReset` / `sendMagicLink` / `sendChangeEmail`），每个方法对应一个 HTML + plaintext 模板。
|
||||
- `apps/server/src/services/adapters/email.ts`：统一 `EmailService` 接口（`sendVerification` / `sendPasswordReset` / `sendMagicLink` / `sendChangeEmail`），每个方法对应一个 HTML + plaintext 模板。
|
||||
- `apps/server/src/libs/auth.ts`：装上 4 个 callback；启用 `requireEmailVerification: true`；加载 `magicLink` plugin。
|
||||
- `apps/server/src/libs/env.ts`：新增 `RESEND_API_KEY`（必填）、`RESEND_FROM_EMAIL`（必填）、`RESEND_FROM_NAME`（可选）、`AUTH_EMAIL_VERIFY_REDIRECT_URL` / `AUTH_PASSWORD_RESET_REDIRECT_URL`（可选，默认根据 `API_SERVER_URL` 推算 ui-server-auth origin）。
|
||||
- `apps/server/src/app.ts`：把 `EmailService` 通过 `injeca` 装配，注入到 `auth` provider。
|
||||
|
||||
@@ -62,7 +62,7 @@ INCRBY/DECRBY 的组合在 Redis 单线程模型下天然原子；多服务实
|
||||
|
||||
## API
|
||||
|
||||
`packages/server/src/services/billing/flux-meter.ts`
|
||||
`packages/server/src/services/domain/billing/flux-meter.ts`
|
||||
|
||||
- `createFluxMeter(redis, billingService, { name, resolveRuntime })` → meter 实例
|
||||
- `resolveRuntime: () => Promise<{ unitsPerFlux, debtTtlSeconds }>` **每次调用都执行**，不做进程内缓存。多实例部署下任一实例改配置，其它实例下一次请求立即生效。
|
||||
@@ -93,7 +93,7 @@ INCRBY/DECRBY 的组合在 Redis 单线程模型下天然原子；多服务实
|
||||
|
||||
### 接入步骤
|
||||
|
||||
1. 在 `services/config-kv.ts` 加费率/TTL 配置项
|
||||
1. 在 `services/adapters/config-kv.ts` 加费率/TTL 配置项
|
||||
2. 在 `app.ts` 用 `injeca.provide` 注册新 meter，注入对应路由 / 服务
|
||||
3. 在路由中：先 `assertCanAfford`，调上游成功后 `accumulate`
|
||||
4. 加单测覆盖：累计跨阈值、empty input、余额不足
|
||||
|
||||
@@ -13,11 +13,11 @@ remaining findings are deferred — each evaluated through the AGENTS.md
|
||||
"client disconnects but upstream keeps generating + burning paid quota"
|
||||
leak.
|
||||
2. **Failed upstream response bodies not drained** — fixed in
|
||||
`apps/server/src/services/llm-router/router.ts`. Every non-2xx fallback
|
||||
`apps/server/src/services/domain/llm-router/router.ts`. Every non-2xx fallback
|
||||
path now calls `response.body?.cancel()` before continuing. Prevents
|
||||
socket-pool exhaustion under fallback storms.
|
||||
3. **SSML voice attribute injection** — fixed in
|
||||
`apps/server/src/services/tts-adapters/azure.ts`. Voice id is
|
||||
`apps/server/src/services/adapters/tts/azure.ts`. Voice id is
|
||||
regex-validated (`^[a-z0-9-]+$/i`) before SSML interpolation; invalid
|
||||
values throw `BAD_REQUEST`. Prevents attribute-context breakout under
|
||||
the server's Azure credential.
|
||||
|
||||
@@ -112,7 +112,7 @@
|
||||
|
||||
已接入：
|
||||
- 前端 `posthog-js` 通过 `packages/stage-ui/src/stores/analytics/posthog.ts` 初始化，三个 app（web / desktop / pocket）按 `isStageTamagotchi()` 等选 project key
|
||||
- 后端 `posthog-node` 通过 `apps/server/src/services/posthog.ts` + injeca provider `services:posthog`
|
||||
- 后端 `posthog-node` 通过 `apps/server/src/services/adapters/posthog.ts` + injeca provider `services:posthog`
|
||||
- 前端↔后端 identity merge：`useSharedAnalyticsStore.initialize()` watch `authStore.isAuthenticated` 自动调 `posthog.identify(user.id)` / `reset()`
|
||||
|
||||
已埋点：
|
||||
@@ -152,7 +152,7 @@
|
||||
`apps/server`：
|
||||
|
||||
```ts
|
||||
// services/posthog.ts（新增）
|
||||
// services/adapters/posthog.ts（新增）
|
||||
import { PostHog } from 'posthog-node'
|
||||
|
||||
export function createPostHog(env: ServerEnv) {
|
||||
|
||||
@@ -52,13 +52,13 @@ OTel SDK 在导出到 Prometheus 时做两件事：
|
||||
|
||||
| Metric | 类型 | 落点 | Labels |
|
||||
|---|---|---|---|
|
||||
| `chat.messages` | Counter | [services/chats.ts](../../src/services/chats.ts) `pushMessages` | — |
|
||||
| `character.created` | Counter | [services/characters.ts](../../src/services/characters.ts) | — |
|
||||
| `chat.messages` | Counter | [services/domain/chats.ts](../../src/services/domain/chats.ts) `pushMessages` | — |
|
||||
| `character.created` | Counter | [services/domain/characters.ts](../../src/services/domain/characters.ts) | — |
|
||||
| `character.deleted` | Counter | 同上 | — |
|
||||
| `character.engagement` | Counter | 同上（like/bookmark） | `action`（`like` / `unlike` / `bookmark` / `unbookmark`） |
|
||||
| `ws.connections.active` | ObservableGauge | [routes/chat-ws/index.ts](../../src/routes/chat-ws/index.ts) `addCallback` walks `userConnections` Map | — |
|
||||
| `ws.messages.sent` | Counter | 同上 | — |
|
||||
| `ws.messages.received` | Counter | [services/chats.ts](../../src/services/chats.ts) | — |
|
||||
| `ws.messages.received` | Counter | [services/domain/chats.ts](../../src/services/domain/chats.ts) | — |
|
||||
|
||||
## Revenue & Billing
|
||||
|
||||
@@ -80,10 +80,10 @@ OTel SDK 在导出到 Prometheus 时做两件事：
|
||||
| Metric | 类型 | 落点 | Labels |
|
||||
|---|---|---|---|
|
||||
| `airi.billing.flux.consumed` | Counter | [routes/openai/v1/index.ts](../../src/routes/openai/v1/index.ts) `recordMetrics`（chat / tts） | `gen_ai.request.model`、`gen_ai.operation.name`/`airi.gen_ai.operation.kind`、`http.response.status_code` |
|
||||
| `airi.billing.flux.credited` | Counter | [services/billing/billing-service.ts](../../src/services/billing/billing-service.ts) 三条入账路径 | `source`（`stripe.checkout`/`stripe.invoice`/`promo`/`admin_grant`/...）、`type`（`credit`/`promo`） |
|
||||
| `airi.billing.flux.credited` | Counter | [services/domain/billing/billing-service.ts](../../src/services/domain/billing/billing-service.ts) 三条入账路径 | `source`（`stripe.checkout`/`stripe.invoice`/`promo`/`admin_grant`/...）、`type`（`credit`/`promo`） |
|
||||
| `airi.billing.flux.unbilled` | Counter | [routes/openai/v1/index.ts](../../src/routes/openai/v1/index.ts) streaming 路径里 `consumeFluxForLLM` 失败的 catch | `gen_ai.request.model`、`reason`（`debit_failed`）、`stage`（`streaming`） |
|
||||
| `flux.insufficient_balance` | Counter | [services/billing/billing-service.ts](../../src/services/billing/billing-service.ts) `debitFlux` | — |
|
||||
| `airi.billing.tts.chars` | Counter | [services/billing/flux-meter.ts](../../src/services/billing/flux-meter.ts) `accumulate` | `meter`（`tts`）、`model` |
|
||||
| `flux.insufficient_balance` | Counter | [services/domain/billing/billing-service.ts](../../src/services/domain/billing/billing-service.ts) `debitFlux` | — |
|
||||
| `airi.billing.tts.chars` | Counter | [services/domain/billing/flux-meter.ts](../../src/services/domain/billing/flux-meter.ts) `accumulate` | `meter`（`tts`）、`model` |
|
||||
| `airi.billing.tts.preflight_rejections` | Counter | `flux-meter.ts` `assertCanAfford` | `meter`、`reason`（`insufficient_balance`） |
|
||||
|
||||
> **`airi.billing.flux.unbilled` 是 P0 告警金线**：流式响应已经发给用户（HTTP 200，token 已经流出），但 post-stream debit 抛错——response 路径不会因此 5xx，DB latency 也只在 catch 那一瞬间显著。HTTP / DB 告警**覆盖不到**这条静默 revenue leak。推荐 alert：`increase(airi_billing_flux_unbilled_total[5m]) > 0` 持续 > 0 立刻 page。
|
||||
@@ -101,7 +101,7 @@ OTel SDK 在导出到 Prometheus 时做两件事：
|
||||
|
||||
## Email（Resend）
|
||||
|
||||
来源 [services/email.ts](../../src/services/email.ts) 的 `send()` 内部 try/catch。
|
||||
来源 [services/adapters/email.ts](../../src/services/adapters/email.ts) 的 `send()` 内部 try/catch。
|
||||
|
||||
| Metric | 类型 | Labels |
|
||||
|---|---|---|
|
||||
|
||||
@@ -162,9 +162,9 @@ const data = JSON.parse(message) as BroadcastMessage
|
||||
- key helper 集中点
|
||||
- `src/utils/redis-keys.ts`
|
||||
- 余额读 cache-aside
|
||||
- `src/services/flux.ts`
|
||||
- `src/services/domain/flux.ts`
|
||||
- Sub-Flux 计量债务账本
|
||||
- `src/services/billing/flux-meter.ts`
|
||||
- `src/services/domain/billing/flux-meter.ts`
|
||||
- TTS voices 上游响应缓存
|
||||
- `src/routes/openai/v1/index.ts::handleListVoices`
|
||||
- Pub/Sub 聊天广播
|
||||
|
||||
@@ -66,7 +66,7 @@
|
||||
实现位置：
|
||||
|
||||
- route: `src/routes/characters/index.ts`
|
||||
- service: `src/services/characters.ts`
|
||||
- service: `src/services/domain/characters.ts`
|
||||
|
||||
主要能力：
|
||||
|
||||
@@ -91,7 +91,7 @@
|
||||
实现位置：
|
||||
|
||||
- route: `src/routes/providers/index.ts`
|
||||
- service: `src/services/providers.ts`
|
||||
- service: `src/services/domain/providers.ts`
|
||||
|
||||
主要能力：
|
||||
|
||||
@@ -110,7 +110,7 @@
|
||||
实现位置：
|
||||
|
||||
- route: `src/routes/chats/index.ts`
|
||||
- service: `src/services/chats.ts`
|
||||
- service: `src/services/domain/chats.ts`
|
||||
|
||||
主要能力：
|
||||
|
||||
@@ -240,7 +240,7 @@
|
||||
实现位置：
|
||||
|
||||
- route: `src/routes/admin/flux-grants/index.ts`
|
||||
- service: `src/services/admin-flux-grants/index.ts`
|
||||
- service: `src/services/domain/admin/flux-grants/index.ts`
|
||||
- guard: `src/middlewares/admin-guard.ts`
|
||||
|
||||
主要能力：
|
||||
|
||||
@@ -55,7 +55,7 @@ vue-router never registered the route. Re-build → fixes.
|
||||
| Server typecheck | `pnpm -F @proj-airi/server typecheck` exits clean | 2026-04-28 |
|
||||
| Monorepo typecheck | `pnpm typecheck` exits clean across all packages | 2026-04-28 |
|
||||
| Lint | `pnpm lint` reports 0 errors in deletion-service / handler / UI files | 2026-04-28 |
|
||||
| Deletion service unit tests | `pnpm exec vitest run apps/server/src/services/user-deletion` → `2 files / 14 tests pass` (registry priority order, abort-on-error, serial execution, idempotency, per-service `deleteAllForUser` correctness) | 2026-04-28 |
|
||||
| Deletion service unit tests | `pnpm exec vitest run apps/server/src/services/domain/user-deletion` → `2 files / 14 tests pass` (registry priority order, abort-on-error, serial execution, idempotency, per-service `deleteAllForUser` correctness) | 2026-04-28 |
|
||||
| Architecture refactor | Domain knowledge moved out of `*-deletion-handler.ts` files into each business service's own `deleteAllForUser` method. Registry retained as a thin scheduler. `auth → userDeletionService → 5 business services` (auth and services no longer depend on each other). | 2026-04-28 |
|
||||
| Server full test suite | `pnpm -F @proj-airi/server exec vitest run` → 244/245 pass; the single failure (`origin.test.ts`) is pre-existing on `main` and unrelated | 2026-04-28 |
|
||||
| UI typecheck | `pnpm -F @proj-airi/stage-pages typecheck` and `pnpm -F @proj-airi/ui-server-auth typecheck` both clean | 2026-04-28 |
|
||||
|
||||
@@ -21,8 +21,8 @@ import { env, exit } from 'node:process'
|
||||
import Redis from 'ioredis'
|
||||
|
||||
import { parseEnv } from '../src/libs/env'
|
||||
import { createConfigKVService } from '../src/services/config-kv'
|
||||
import { createLlmRouterService } from '../src/services/llm-router'
|
||||
import { createConfigKVService } from '../src/services/adapters/config-kv'
|
||||
import { createLlmRouterService } from '../src/services/domain/llm-router'
|
||||
import { createEnvelopeCrypto } from '../src/utils/envelope-crypto'
|
||||
|
||||
async function main() {
|
||||
|
||||
+30
-30
@@ -5,20 +5,20 @@ import type { AuthInstance } from './libs/auth'
|
||||
import type { Database } from './libs/db'
|
||||
import type { Env } from './libs/env'
|
||||
import type { OtelInstance } from './otel'
|
||||
import type { AdminFluxGrantsService } from './services/admin-flux-grants'
|
||||
import type { AdminRouterConfigService } from './services/admin-router-config'
|
||||
import type { BillingService } from './services/billing/billing-service'
|
||||
import type { FluxMeter } from './services/billing/flux-meter'
|
||||
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 { FluxTransactionService } from './services/flux-transaction'
|
||||
import type { LlmRouterService } from './services/llm-router'
|
||||
import type { ProviderService } from './services/providers'
|
||||
import type { RequestLogService } from './services/request-log'
|
||||
import type { StripeService } from './services/stripe'
|
||||
import type { UserDeletionService } from './services/user-deletion'
|
||||
import type { ConfigKVService } from './services/adapters/config-kv'
|
||||
import type { AdminFluxGrantsService } from './services/domain/admin/flux-grants'
|
||||
import type { AdminRouterConfigService } from './services/domain/admin/router-config'
|
||||
import type { BillingService } from './services/domain/billing/billing-service'
|
||||
import type { FluxMeter } from './services/domain/billing/flux-meter'
|
||||
import type { CharacterService } from './services/domain/characters'
|
||||
import type { ChatService } from './services/domain/chats'
|
||||
import type { FluxService } from './services/domain/flux'
|
||||
import type { FluxTransactionService } from './services/domain/flux-transaction'
|
||||
import type { LlmRouterService } from './services/domain/llm-router'
|
||||
import type { ProviderService } from './services/domain/providers'
|
||||
import type { RequestLogService } from './services/domain/request-log'
|
||||
import type { StripeService } from './services/domain/stripe'
|
||||
import type { UserDeletionService } from './services/domain/user-deletion'
|
||||
import type { HonoEnv } from './types/hono'
|
||||
import type { EnvelopeCrypto } from './utils/envelope-crypto'
|
||||
|
||||
@@ -58,22 +58,22 @@ import { createFluxRoutes } from './routes/flux'
|
||||
import { createV1Routes } from './routes/openai/v1'
|
||||
import { createProviderRoutes } from './routes/providers'
|
||||
import { createStripeRoutes } from './routes/stripe'
|
||||
import { createAdminFluxGrantsService } from './services/admin-flux-grants'
|
||||
import { createAdminRouterConfigService } from './services/admin-router-config'
|
||||
import { createBillingService } from './services/billing/billing-service'
|
||||
import { createFluxMeter } from './services/billing/flux-meter'
|
||||
import { createCharacterService } from './services/characters'
|
||||
import { createChatService } from './services/chats'
|
||||
import { createConfigKVService } from './services/config-kv'
|
||||
import { createEmailService } from './services/email'
|
||||
import { createFluxService } from './services/flux'
|
||||
import { createFluxTransactionService } from './services/flux-transaction'
|
||||
import { createConfigSyncSubscriber, createLlmRouterService } from './services/llm-router'
|
||||
import { createPostHogClient } from './services/posthog'
|
||||
import { createProviderService } from './services/providers'
|
||||
import { createRequestLogService } from './services/request-log'
|
||||
import { createStripeService } from './services/stripe'
|
||||
import { createUserDeletionService } from './services/user-deletion'
|
||||
import { createConfigKVService } from './services/adapters/config-kv'
|
||||
import { createEmailService } from './services/adapters/email'
|
||||
import { createPostHogClient } from './services/adapters/posthog'
|
||||
import { createAdminFluxGrantsService } from './services/domain/admin/flux-grants'
|
||||
import { createAdminRouterConfigService } from './services/domain/admin/router-config'
|
||||
import { createBillingService } from './services/domain/billing/billing-service'
|
||||
import { createFluxMeter } from './services/domain/billing/flux-meter'
|
||||
import { createCharacterService } from './services/domain/characters'
|
||||
import { createChatService } from './services/domain/chats'
|
||||
import { createFluxService } from './services/domain/flux'
|
||||
import { createFluxTransactionService } from './services/domain/flux-transaction'
|
||||
import { createConfigSyncSubscriber, createLlmRouterService } from './services/domain/llm-router'
|
||||
import { createProviderService } from './services/domain/providers'
|
||||
import { createRequestLogService } from './services/domain/request-log'
|
||||
import { createStripeService } from './services/domain/stripe'
|
||||
import { createUserDeletionService } from './services/domain/user-deletion'
|
||||
import { createEnvelopeCrypto } from './utils/envelope-crypto'
|
||||
import { ApiError, createInternalError } from './utils/error'
|
||||
import { nanoid } from './utils/id'
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import type { PostHog } from 'posthog-node'
|
||||
|
||||
import type { AuthMetrics } from '../otel'
|
||||
import type { EmailService } from '../services/email'
|
||||
import type { UserDeletionService } from '../services/user-deletion'
|
||||
import type { EmailService } from '../services/adapters/email'
|
||||
import type { UserDeletionService } from '../services/domain/user-deletion'
|
||||
import type { Database } from './db'
|
||||
import type { Env } from './env'
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import type { MiddlewareHandler } from 'hono'
|
||||
|
||||
import type { ConfigKVService } from '../services/config-kv'
|
||||
import type { ConfigKVService } from '../services/adapters/config-kv'
|
||||
import type { HonoEnv } from '../types/hono'
|
||||
|
||||
import { createServiceUnavailableError } from '../utils/error'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Env } from '../../../../libs/env'
|
||||
import type { AdminRouterConfigService, SliceInput } from '../../../../services/admin-router-config'
|
||||
import type { AdminRouterConfigService, SliceInput } from '../../../../services/domain/admin/router-config'
|
||||
import type { HonoEnv } from '../../../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -1,198 +0,0 @@
|
||||
import type { Env } from '../../../libs/env'
|
||||
import type { FluxGrantBatchService } from '../../../services/admin-flux-grant-batch/flux-grant-batch-service'
|
||||
import type { HonoEnv } from '../../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
import { array, email, integer, maxLength, maxValue, minLength, minValue, nonEmpty, number, object, optional, parse, pipe, safeParse, string, transform } from 'valibot'
|
||||
|
||||
import { adminGuard } from '../../../middlewares/admin-guard'
|
||||
import { authGuard } from '../../../middlewares/auth'
|
||||
import { createBadRequestError, createNotFoundError } from '../../../utils/error'
|
||||
|
||||
/**
|
||||
* Per-batch upper bound on amount per user. Caps a single typo from
|
||||
* issuing absurd amounts. Operator can override later via configKV.
|
||||
*/
|
||||
const MAX_GRANT_AMOUNT_PER_USER = 10_000
|
||||
|
||||
/**
|
||||
* Hard cap on emails per single batch request. Beyond this we'd rather
|
||||
* the operator chunk into multiple batches.
|
||||
*/
|
||||
const MAX_EMAILS_PER_BATCH = 10_000
|
||||
|
||||
const CreateBatchBodySchema = object({
|
||||
name: pipe(string(), nonEmpty('name is required'), maxLength(100)),
|
||||
amount: pipe(
|
||||
number(),
|
||||
integer('amount must be an integer'),
|
||||
minValue(1, 'amount must be at least 1'),
|
||||
maxValue(MAX_GRANT_AMOUNT_PER_USER, `amount must be at most ${MAX_GRANT_AMOUNT_PER_USER}`),
|
||||
),
|
||||
description: optional(pipe(string(), maxLength(500))),
|
||||
emails: pipe(
|
||||
array(pipe(string(), email('emails must be valid email addresses'))),
|
||||
minLength(1, 'emails must not be empty'),
|
||||
maxLength(MAX_EMAILS_PER_BATCH, `emails must be at most ${MAX_EMAILS_PER_BATCH} entries`),
|
||||
),
|
||||
})
|
||||
|
||||
const ListQuerySchema = object({
|
||||
limit: optional(
|
||||
pipe(string(), transform(Number), integer(), minValue(1), maxValue(100)),
|
||||
'20',
|
||||
),
|
||||
cursor: optional(string()),
|
||||
status: optional(string()),
|
||||
})
|
||||
|
||||
/**
|
||||
* Routes for `/api/admin/flux-grant-batches/*`.
|
||||
*
|
||||
* Use when:
|
||||
* - Mounting under `/api/admin/flux-grant-batches` in `app.ts`
|
||||
*
|
||||
* Expects:
|
||||
* - `sessionMiddleware` already attached (so `c.get('user')` is populated)
|
||||
* - `env.ADMIN_EMAILS` configured for the deployment
|
||||
*
|
||||
* Returns:
|
||||
* - A Hono sub-router. The caller mounts it; `app.ts` attaches CORS/body
|
||||
* limit/error handlers globally.
|
||||
*/
|
||||
export function createAdminFluxGrantBatchRoutes(
|
||||
fluxGrantBatchService: FluxGrantBatchService,
|
||||
env: Env,
|
||||
/**
|
||||
* Throttle hint surfaced in dry-run preview to give an estimated duration.
|
||||
* Comes from the same configKV/env value the worker uses.
|
||||
* @default 50
|
||||
*/
|
||||
throttlePerSec: number = 50,
|
||||
) {
|
||||
return new Hono<HonoEnv>()
|
||||
.use('*', authGuard)
|
||||
.use('*', adminGuard(env))
|
||||
.post('/', async (c) => {
|
||||
const user = c.get('user')!
|
||||
const dryRun = c.req.query('dryRun') === 'true'
|
||||
|
||||
const raw = await c.req.json().catch(() => null)
|
||||
if (raw == null)
|
||||
throw createBadRequestError('Request body must be JSON', 'INVALID_BODY')
|
||||
|
||||
const parsed = safeParse(CreateBatchBodySchema, raw)
|
||||
if (!parsed.success) {
|
||||
throw createBadRequestError(
|
||||
'Invalid request body',
|
||||
'INVALID_BODY',
|
||||
parsed.issues.map(i => ({ path: i.path?.map(p => p.key).join('.'), message: i.message })),
|
||||
)
|
||||
}
|
||||
|
||||
const body = parsed.output
|
||||
|
||||
if (dryRun) {
|
||||
const summary = await fluxGrantBatchService.preview({
|
||||
name: body.name,
|
||||
amount: body.amount,
|
||||
description: body.description,
|
||||
emails: body.emails,
|
||||
throttlePerSec,
|
||||
})
|
||||
return c.json({ preview: summary })
|
||||
}
|
||||
|
||||
const { batch, summary } = await fluxGrantBatchService.create({
|
||||
name: body.name,
|
||||
amount: body.amount,
|
||||
description: body.description,
|
||||
emails: body.emails,
|
||||
createdByUserId: user.id,
|
||||
throttlePerSec,
|
||||
})
|
||||
|
||||
return c.json({
|
||||
batch: {
|
||||
id: batch.id,
|
||||
name: batch.name,
|
||||
status: batch.status,
|
||||
createdAt: batch.createdAt.toISOString(),
|
||||
createdByUserId: batch.createdByUserId,
|
||||
},
|
||||
summary: {
|
||||
totalEmails: summary.totalEmails,
|
||||
pending: summary.willGrant,
|
||||
skipped: summary.willSkip.notFound + summary.willSkip.userDeleted + summary.willSkip.duplicateInInput,
|
||||
totalFluxToIssue: summary.totalFluxToIssue,
|
||||
},
|
||||
}, 202)
|
||||
})
|
||||
.get('/', async (c) => {
|
||||
const query = parse(ListQuerySchema, {
|
||||
limit: c.req.query('limit'),
|
||||
cursor: c.req.query('cursor'),
|
||||
status: c.req.query('status'),
|
||||
})
|
||||
|
||||
const { batches, nextCursor } = await fluxGrantBatchService.list({
|
||||
limit: query.limit,
|
||||
cursor: query.cursor,
|
||||
status: query.status,
|
||||
})
|
||||
|
||||
return c.json({
|
||||
batches: batches.map(b => ({
|
||||
id: b.id,
|
||||
name: b.name,
|
||||
type: b.type,
|
||||
amount: b.amount,
|
||||
status: b.status,
|
||||
createdByUserId: b.createdByUserId,
|
||||
createdAt: b.createdAt.toISOString(),
|
||||
startedAt: b.startedAt?.toISOString() ?? null,
|
||||
completedAt: b.completedAt?.toISOString() ?? null,
|
||||
})),
|
||||
nextCursor,
|
||||
})
|
||||
})
|
||||
.get('/:id', async (c) => {
|
||||
const id = c.req.param('id')
|
||||
const result = await fluxGrantBatchService.get(id)
|
||||
if (!result)
|
||||
throw createNotFoundError('Flux grant batch not found', { id })
|
||||
|
||||
return c.json({
|
||||
batch: {
|
||||
id: result.batch.id,
|
||||
name: result.batch.name,
|
||||
type: result.batch.type,
|
||||
amount: result.batch.amount,
|
||||
description: result.batch.description,
|
||||
status: result.batch.status,
|
||||
createdByUserId: result.batch.createdByUserId,
|
||||
createdAt: result.batch.createdAt.toISOString(),
|
||||
startedAt: result.batch.startedAt?.toISOString() ?? null,
|
||||
completedAt: result.batch.completedAt?.toISOString() ?? null,
|
||||
},
|
||||
progress: result.progress,
|
||||
recentFailures: result.recentFailures.map(f => ({
|
||||
id: f.id,
|
||||
inputEmail: f.inputEmail,
|
||||
userId: f.userId,
|
||||
errorReason: f.errorReason,
|
||||
attemptCount: f.attemptCount,
|
||||
lastAttemptedAt: f.lastAttemptedAt?.toISOString() ?? null,
|
||||
})),
|
||||
})
|
||||
})
|
||||
.post('/:id/retry', async (c) => {
|
||||
const id = c.req.param('id')
|
||||
const existing = await fluxGrantBatchService.get(id)
|
||||
if (!existing)
|
||||
throw createNotFoundError('Flux grant batch not found', { id })
|
||||
|
||||
const result = await fluxGrantBatchService.retryFailed(id)
|
||||
return c.json(result)
|
||||
})
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Env } from '../../../libs/env'
|
||||
import type { AdminFluxGrantsService } from '../../../services/admin-flux-grants'
|
||||
import type { AdminFluxGrantsService } from '../../../services/domain/admin/flux-grants'
|
||||
import type { HonoEnv } from '../../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import type { WSContext, WSEvents } from 'hono/ws'
|
||||
|
||||
import type { FluxMeter } from '../../services/billing/flux-meter'
|
||||
import type { ConfigKVService } from '../../services/config-kv'
|
||||
import type { FluxService } from '../../services/flux'
|
||||
import type { RequestLogService } from '../../services/request-log'
|
||||
import type { ConfigKVService } from '../../services/adapters/config-kv'
|
||||
import type { FluxMeter } from '../../services/domain/billing/flux-meter'
|
||||
import type { FluxService } from '../../services/domain/flux'
|
||||
import type { RequestLogService } from '../../services/domain/request-log'
|
||||
import type { EnvelopeCrypto } from '../../utils/envelope-crypto'
|
||||
|
||||
import { Buffer } from 'node:buffer'
|
||||
|
||||
@@ -2,7 +2,7 @@ import type { AuthInstance } from '../../libs/auth'
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { Env } from '../../libs/env'
|
||||
import type { RateLimitMetrics } from '../../otel'
|
||||
import type { ConfigKVService } from '../../services/config-kv'
|
||||
import type { ConfigKVService } from '../../services/adapters/config-kv'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { oauthProviderAuthServerMetadata, oauthProviderOpenIdConfigMetadata } from '@better-auth/oauth-provider'
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { CharacterService } from '../../services/characters'
|
||||
import type { CharacterService } from '../../services/domain/characters'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -6,7 +6,7 @@ import { beforeAll, describe, expect, it } from 'vitest'
|
||||
|
||||
import { createCharacterRoutes } from '.'
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { createCharacterService } from '../../services/characters'
|
||||
import { createCharacterService } from '../../services/domain/characters'
|
||||
import { ApiError } from '../../utils/error'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
@@ -2,7 +2,7 @@ import type Redis from 'ioredis'
|
||||
|
||||
import type { HonoWsInvocableEventContext } from '../../libs/eventa-hono-adapter'
|
||||
import type { EngagementMetrics } from '../../otel'
|
||||
import type { ChatService } from '../../services/chats'
|
||||
import type { ChatService } from '../../services/domain/chats'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { defineInvokeHandler } from '@moeru/eventa'
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { ChatService } from '../../services/chats'
|
||||
import type { ChatService } from '../../services/domain/chats'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { FluxService } from '../../services/flux'
|
||||
import type { FluxTransactionService } from '../../services/flux-transaction'
|
||||
import type { FluxService } from '../../services/domain/flux'
|
||||
import type { FluxTransactionService } from '../../services/domain/flux-transaction'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { FluxService } from '../../services/flux'
|
||||
import type { FluxTransactionService } from '../../services/flux-transaction'
|
||||
import type { FluxService } from '../../services/domain/flux'
|
||||
import type { FluxTransactionService } from '../../services/domain/flux-transaction'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -2,13 +2,13 @@ import type { Context } from 'hono'
|
||||
import type { PostHog } from 'posthog-node'
|
||||
|
||||
import type { GenAiMetrics, RateLimitMetrics, RevenueMetrics } from '../../../otel'
|
||||
import type { UsageInfo } from '../../../services/billing/billing'
|
||||
import type { BillingService } from '../../../services/billing/billing-service'
|
||||
import type { FluxMeter } from '../../../services/billing/flux-meter'
|
||||
import type { ConfigKVService } from '../../../services/config-kv'
|
||||
import type { FluxService } from '../../../services/flux'
|
||||
import type { LlmRouterService } from '../../../services/llm-router'
|
||||
import type { RequestLogService } from '../../../services/request-log'
|
||||
import type { ConfigKVService } from '../../../services/adapters/config-kv'
|
||||
import type { UsageInfo } from '../../../services/domain/billing/billing'
|
||||
import type { BillingService } from '../../../services/domain/billing/billing-service'
|
||||
import type { FluxMeter } from '../../../services/domain/billing/flux-meter'
|
||||
import type { FluxService } from '../../../services/domain/flux'
|
||||
import type { LlmRouterService } from '../../../services/domain/llm-router'
|
||||
import type { RequestLogService } from '../../../services/domain/request-log'
|
||||
import type { HonoEnv } from '../../../types/hono'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
@@ -18,8 +18,8 @@ import { Hono } from 'hono'
|
||||
import { authGuard } from '../../../middlewares/auth'
|
||||
import { configGuard } from '../../../middlewares/config-guard'
|
||||
import { rateLimiter } from '../../../middlewares/rate-limit'
|
||||
import { calculateFluxFromUsage, extractUsageFromBody } from '../../../services/billing/billing'
|
||||
import { captureSafe } from '../../../services/posthog'
|
||||
import { captureSafe } from '../../../services/adapters/posthog'
|
||||
import { calculateFluxFromUsage, extractUsageFromBody } from '../../../services/domain/billing/billing'
|
||||
import { createPaymentRequiredError } from '../../../utils/error'
|
||||
import { nanoid } from '../../../utils/id'
|
||||
import {
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import type { BillingService } from '../../../services/billing/billing-service'
|
||||
import type { ConfigKVService } from '../../../services/config-kv'
|
||||
import type { FluxService } from '../../../services/flux'
|
||||
import type { LlmRouterService } from '../../../services/llm-router'
|
||||
import type { RequestLogService } from '../../../services/request-log'
|
||||
import type { ConfigKVService } from '../../../services/adapters/config-kv'
|
||||
import type { BillingService } from '../../../services/domain/billing/billing-service'
|
||||
import type { FluxService } from '../../../services/domain/flux'
|
||||
import type { LlmRouterService } from '../../../services/domain/llm-router'
|
||||
import type { RequestLogService } from '../../../services/domain/request-log'
|
||||
import type { HonoEnv } from '../../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { ProviderService } from '../../services/providers'
|
||||
import type { ProviderService } from '../../services/domain/providers'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
@@ -6,7 +6,7 @@ import { beforeAll, describe, expect, it } from 'vitest'
|
||||
|
||||
import { createProviderRoutes } from '.'
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { createProviderService } from '../../services/providers'
|
||||
import { createProviderService } from '../../services/domain/providers'
|
||||
import { ApiError } from '../../utils/error'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
@@ -3,10 +3,10 @@ import type { PostHog } from 'posthog-node'
|
||||
|
||||
import type { Env } from '../../libs/env'
|
||||
import type { RateLimitMetrics, RevenueMetrics } from '../../otel'
|
||||
import type { BillingService } from '../../services/billing/billing-service'
|
||||
import type { ConfigKVService } from '../../services/config-kv'
|
||||
import type { FluxService } from '../../services/flux'
|
||||
import type { StripeService } from '../../services/stripe'
|
||||
import type { ConfigKVService } from '../../services/adapters/config-kv'
|
||||
import type { BillingService } from '../../services/domain/billing/billing-service'
|
||||
import type { FluxService } from '../../services/domain/flux'
|
||||
import type { StripeService } from '../../services/domain/stripe'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import Stripe from 'stripe'
|
||||
@@ -17,7 +17,7 @@ import { safeParse } from 'valibot'
|
||||
|
||||
import { authGuard } from '../../middlewares/auth'
|
||||
import { rateLimiter } from '../../middlewares/rate-limit'
|
||||
import { captureSafe } from '../../services/posthog'
|
||||
import { captureSafe } from '../../services/adapters/posthog'
|
||||
import { createBadRequestError, createServiceUnavailableError } from '../../utils/error'
|
||||
import { errorMessageFromUnknown } from '../../utils/error-message'
|
||||
import { resolveTrustedRequestOrigin } from '../../utils/origin'
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import type { StripeCheckoutSession, StripeInvoice } from '../../schemas/stripe'
|
||||
import type { BillingService } from '../../services/billing/billing-service'
|
||||
import type { ConfigKVService } from '../../services/config-kv'
|
||||
import type { FluxService } from '../../services/flux'
|
||||
import type { StripeService } from '../../services/stripe'
|
||||
import type { ConfigKVService } from '../../services/adapters/config-kv'
|
||||
import type { BillingService } from '../../services/domain/billing/billing-service'
|
||||
import type { FluxService } from '../../services/domain/flux'
|
||||
import type { StripeService } from '../../services/domain/stripe'
|
||||
import type { HonoEnv } from '../../types/hono'
|
||||
|
||||
import { Hono } from 'hono'
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { configRedisKey } from '../../utils/redis-keys'
|
||||
import { createConfigKVService } from '../config-kv'
|
||||
import { createConfigKVService } from './config-kv'
|
||||
|
||||
function createMockRedis() {
|
||||
const store = new Map<string, string>()
|
||||
+2
-2
@@ -3,8 +3,8 @@ import type { InferOutput } from 'valibot'
|
||||
|
||||
import { any, array, boolean, check, nonEmpty, number, object, optional, parse, picklist, pipe, record, regex, string } from 'valibot'
|
||||
|
||||
import { createServiceUnavailableError } from '../utils/error'
|
||||
import { configRedisKey } from '../utils/redis-keys'
|
||||
import { createServiceUnavailableError } from '../../utils/error'
|
||||
import { configRedisKey } from '../../utils/redis-keys'
|
||||
|
||||
/**
|
||||
* LLM/TTS router config tree. Single composite entry under configKV holds the
|
||||
@@ -1,12 +1,12 @@
|
||||
import type { Logger } from '@guiiai/logg'
|
||||
|
||||
import type { EmailMetrics } from '../otel'
|
||||
import type { EmailMetrics } from '../../otel'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { errorMessageFrom } from '@moeru/std'
|
||||
import { Resend } from 'resend'
|
||||
|
||||
import { ApiError } from '../utils/error'
|
||||
import { ApiError } from '../../utils/error'
|
||||
|
||||
/**
|
||||
* Outbound email payload accepted by {@link EmailService.send}.
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Env } from '../libs/env'
|
||||
import type { Env } from '../../libs/env'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { PostHog } from 'posthog-node'
|
||||
+1
-1
@@ -6,7 +6,7 @@ import { errorMessageFrom } from '@moeru/std'
|
||||
|
||||
import azureVoices from './voices/azure.json' with { type: 'json' }
|
||||
|
||||
import { createBadRequestError, createInternalError } from '../../utils/error'
|
||||
import { createBadRequestError, createInternalError } from '../../../utils/error'
|
||||
|
||||
// NOTICE:
|
||||
// Voice IDs Azure accepts are stable strings like `en-US-AvaMultilingualNeural`.
|
||||
+1
-1
@@ -8,7 +8,7 @@ import { errorMessageFrom } from '@moeru/std'
|
||||
|
||||
import cosyvoiceVoices from './voices/dashscope-cosyvoice.json' with { type: 'json' }
|
||||
|
||||
import { createInternalError } from '../../utils/error'
|
||||
import { createInternalError } from '../../../utils/error'
|
||||
|
||||
/**
|
||||
* Default DashScope cosyvoice voice id. v2 voice ids carry an explicit `_v2`
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { ApiError } from '../../utils/error'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
import { getAdapter } from './index'
|
||||
|
||||
describe('getAdapter', () => {
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
import type { TtsAdapter, TtsAdapterId } from './types'
|
||||
|
||||
import { createBadRequestError } from '../../utils/error'
|
||||
import { createBadRequestError } from '../../../utils/error'
|
||||
import { azureAdapter } from './azure'
|
||||
import { dashscopeCosyvoiceAdapter } from './dashscope-cosyvoice'
|
||||
import { volcengineAdapter } from './volcengine'
|
||||
+2
-2
@@ -8,8 +8,8 @@ import { errorMessageFrom } from '@moeru/std'
|
||||
|
||||
import volcengineVoices from './voices/volcengine.json' with { type: 'json' }
|
||||
|
||||
import { createInternalError } from '../../utils/error'
|
||||
import { nanoid } from '../../utils/id'
|
||||
import { createInternalError } from '../../../utils/error'
|
||||
import { nanoid } from '../../../utils/id'
|
||||
|
||||
/**
|
||||
* Default Volcengine TTS voice id. `BV001_streaming` is Volcengine's standard
|
||||
@@ -1,460 +0,0 @@
|
||||
import type { Database } from '../../libs/db'
|
||||
|
||||
import { Buffer } from 'node:buffer'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, desc, eq, inArray, isNull, sql } from 'drizzle-orm'
|
||||
|
||||
import * as accountsSchema from '../../schemas/accounts'
|
||||
import * as fluxSchema from '../../schemas/flux'
|
||||
import * as batchSchema from '../../schemas/flux-grant-batch'
|
||||
|
||||
const logger = useLogger('admin-flux-grant-batch-service')
|
||||
|
||||
/**
|
||||
* Per-email outcome of resolving an input list against the user table.
|
||||
* Used by both dry-run preview and persisted recipient rows.
|
||||
*/
|
||||
export interface ResolvedEmail {
|
||||
inputEmail: string
|
||||
userId: string | null
|
||||
status: 'pending' | 'skipped'
|
||||
errorReason: 'not_found' | 'user_deleted' | 'duplicate_in_input' | null
|
||||
}
|
||||
|
||||
export interface ResolveSummary {
|
||||
totalEmails: number
|
||||
willGrant: number
|
||||
willSkip: {
|
||||
notFound: number
|
||||
userDeleted: number
|
||||
duplicateInInput: number
|
||||
}
|
||||
totalFluxToIssue: number
|
||||
estimatedDurationSec: number
|
||||
samples: {
|
||||
willGrant: string[]
|
||||
notFound: string[]
|
||||
userDeleted: string[]
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a list of input emails against the user table.
|
||||
*
|
||||
* Use when:
|
||||
* - Previewing a dry-run before batch creation
|
||||
* - Persisting `flux_grant_batch_recipient` rows with a deterministic resolution snapshot
|
||||
*
|
||||
* Expects:
|
||||
* - `emails` is non-empty; case is preserved in output but compared via LOWER
|
||||
*
|
||||
* Returns:
|
||||
* - One entry per input email (duplicates included with `duplicate_in_input`),
|
||||
* plus a summary used directly in dry-run responses
|
||||
*/
|
||||
async function resolveEmails(
|
||||
db: Database,
|
||||
emails: string[],
|
||||
amountPerUser: number,
|
||||
throttlePerSec: number,
|
||||
): Promise<{ resolved: ResolvedEmail[], summary: ResolveSummary }> {
|
||||
// Lowercase index of inputs → first-seen index. Subsequent occurrences are duplicates.
|
||||
const seenLower = new Map<string, number>()
|
||||
const resolved: ResolvedEmail[] = emails.map((email, idx) => {
|
||||
const lower = email.toLowerCase()
|
||||
if (seenLower.has(lower)) {
|
||||
return { inputEmail: email, userId: null, status: 'skipped', errorReason: 'duplicate_in_input' }
|
||||
}
|
||||
seenLower.set(lower, idx)
|
||||
return { inputEmail: email, userId: null, status: 'pending', errorReason: null }
|
||||
})
|
||||
|
||||
const lowerEmails = Array.from(seenLower.keys())
|
||||
|
||||
// Bulk-fetch matching users (case-insensitive). For 10k inputs this is one
|
||||
// query — Postgres handles `LOWER(email) IN (...)` fine on a unique index
|
||||
// because `email` is already unique; collation just means we filter post-fetch.
|
||||
const users = await db
|
||||
.select({ id: accountsSchema.user.id, email: accountsSchema.user.email })
|
||||
.from(accountsSchema.user)
|
||||
.where(inArray(sql`LOWER(${accountsSchema.user.email})`, lowerEmails))
|
||||
|
||||
const userByLowerEmail = new Map(users.map(u => [u.email.toLowerCase(), u.id]))
|
||||
|
||||
// Bulk-fetch user_flux rows for matched userIds — only need to know which are soft-deleted.
|
||||
const matchedUserIds = users.map(u => u.id)
|
||||
const fluxRows = matchedUserIds.length > 0
|
||||
? await db
|
||||
.select({ userId: fluxSchema.userFlux.userId, deletedAt: fluxSchema.userFlux.deletedAt })
|
||||
.from(fluxSchema.userFlux)
|
||||
.where(inArray(fluxSchema.userFlux.userId, matchedUserIds))
|
||||
: []
|
||||
const deletedUserIds = new Set(
|
||||
fluxRows.filter(r => r.deletedAt != null).map(r => r.userId),
|
||||
)
|
||||
|
||||
// Annotate the resolved list now that we have user lookups + deletion status.
|
||||
for (const entry of resolved) {
|
||||
if (entry.errorReason === 'duplicate_in_input')
|
||||
continue
|
||||
|
||||
const userId = userByLowerEmail.get(entry.inputEmail.toLowerCase())
|
||||
if (!userId) {
|
||||
entry.status = 'skipped'
|
||||
entry.errorReason = 'not_found'
|
||||
continue
|
||||
}
|
||||
|
||||
if (deletedUserIds.has(userId)) {
|
||||
entry.userId = userId
|
||||
entry.status = 'skipped'
|
||||
entry.errorReason = 'user_deleted'
|
||||
continue
|
||||
}
|
||||
|
||||
entry.userId = userId
|
||||
entry.status = 'pending'
|
||||
}
|
||||
|
||||
// Counts and samples for preview output.
|
||||
const willGrant = resolved.filter(r => r.status === 'pending').length
|
||||
const notFound = resolved.filter(r => r.errorReason === 'not_found').length
|
||||
const userDeleted = resolved.filter(r => r.errorReason === 'user_deleted').length
|
||||
const duplicateInInput = resolved.filter(r => r.errorReason === 'duplicate_in_input').length
|
||||
|
||||
const summary: ResolveSummary = {
|
||||
totalEmails: emails.length,
|
||||
willGrant,
|
||||
willSkip: { notFound, userDeleted, duplicateInInput },
|
||||
totalFluxToIssue: willGrant * amountPerUser,
|
||||
estimatedDurationSec: Math.ceil(willGrant / Math.max(1, throttlePerSec)),
|
||||
samples: {
|
||||
willGrant: resolved.filter(r => r.status === 'pending').slice(0, 5).map(r => r.inputEmail),
|
||||
notFound: resolved.filter(r => r.errorReason === 'not_found').slice(0, 5).map(r => r.inputEmail),
|
||||
userDeleted: resolved.filter(r => r.errorReason === 'user_deleted').slice(0, 5).map(r => r.inputEmail),
|
||||
},
|
||||
}
|
||||
|
||||
return { resolved, summary }
|
||||
}
|
||||
|
||||
export interface CreateBatchInput {
|
||||
name: string
|
||||
amount: number
|
||||
description?: string
|
||||
emails: string[]
|
||||
createdByUserId: string
|
||||
throttlePerSec: number
|
||||
}
|
||||
|
||||
export function createFluxGrantBatchService(db: Database) {
|
||||
return {
|
||||
/**
|
||||
* Preview a batch without writing anything. Used by `?dryRun=true`.
|
||||
*/
|
||||
async preview(input: Omit<CreateBatchInput, 'createdByUserId'>) {
|
||||
const { summary } = await resolveEmails(db, input.emails, input.amount, input.throttlePerSec)
|
||||
return summary
|
||||
},
|
||||
|
||||
/**
|
||||
* Create a persistent batch + per-email recipient rows. Worker picks up
|
||||
* `pending` rows asynchronously.
|
||||
*
|
||||
* Resolution happens here (not in the worker) so dry-run preview numbers
|
||||
* match real execution exactly.
|
||||
*/
|
||||
async create(input: CreateBatchInput) {
|
||||
const { resolved, summary } = await resolveEmails(
|
||||
db,
|
||||
input.emails,
|
||||
input.amount,
|
||||
input.throttlePerSec,
|
||||
)
|
||||
|
||||
const batchRow = await db.transaction(async (tx) => {
|
||||
const [created] = await tx.insert(batchSchema.fluxGrantBatch).values({
|
||||
name: input.name,
|
||||
type: 'promo',
|
||||
amount: input.amount,
|
||||
description: input.description,
|
||||
status: 'created',
|
||||
createdByUserId: input.createdByUserId,
|
||||
}).returning()
|
||||
|
||||
if (!created)
|
||||
throw new Error('Failed to insert flux_grant_batch row')
|
||||
|
||||
const recipientRows = resolved.map(r => ({
|
||||
batchId: created.id,
|
||||
inputEmail: r.inputEmail,
|
||||
userId: r.userId,
|
||||
status: r.status,
|
||||
errorReason: r.errorReason,
|
||||
}))
|
||||
|
||||
// Chunk inserts to keep a single statement under Postgres' parameter limit
|
||||
// (≈ 65k params; 5 cols/row = ≈ 13k rows per chunk; we use 1k for headroom).
|
||||
const CHUNK = 1000
|
||||
for (let i = 0; i < recipientRows.length; i += CHUNK)
|
||||
await tx.insert(batchSchema.fluxGrantBatchRecipient).values(recipientRows.slice(i, i + CHUNK))
|
||||
|
||||
return created
|
||||
})
|
||||
|
||||
logger.withFields({
|
||||
batchId: batchRow.id,
|
||||
name: input.name,
|
||||
userCount: input.emails.length,
|
||||
willGrant: summary.willGrant,
|
||||
}).log('Flux grant batch created')
|
||||
|
||||
return { batch: batchRow, summary }
|
||||
},
|
||||
|
||||
/**
|
||||
* Get batch + progress counts + recent failure samples.
|
||||
* Returns null if the batch does not exist.
|
||||
*/
|
||||
async get(batchId: string) {
|
||||
const [row] = await db
|
||||
.select()
|
||||
.from(batchSchema.fluxGrantBatch)
|
||||
.where(eq(batchSchema.fluxGrantBatch.id, batchId))
|
||||
.limit(1)
|
||||
|
||||
if (!row)
|
||||
return null
|
||||
|
||||
const recipientStatusRows = await db
|
||||
.select({ status: batchSchema.fluxGrantBatchRecipient.status, count: sql<number>`count(*)::int` })
|
||||
.from(batchSchema.fluxGrantBatchRecipient)
|
||||
.where(eq(batchSchema.fluxGrantBatchRecipient.batchId, batchId))
|
||||
.groupBy(batchSchema.fluxGrantBatchRecipient.status)
|
||||
|
||||
const progress = { total: 0, pending: 0, granted: 0, skipped: 0, failed: 0 }
|
||||
for (const r of recipientStatusRows) {
|
||||
progress.total += r.count
|
||||
if (r.status === 'pending')
|
||||
progress.pending = r.count
|
||||
else if (r.status === 'granted')
|
||||
progress.granted = r.count
|
||||
else if (r.status === 'skipped')
|
||||
progress.skipped = r.count
|
||||
else if (r.status === 'failed')
|
||||
progress.failed = r.count
|
||||
}
|
||||
|
||||
const recentFailures = await db
|
||||
.select({
|
||||
id: batchSchema.fluxGrantBatchRecipient.id,
|
||||
inputEmail: batchSchema.fluxGrantBatchRecipient.inputEmail,
|
||||
userId: batchSchema.fluxGrantBatchRecipient.userId,
|
||||
errorReason: batchSchema.fluxGrantBatchRecipient.errorReason,
|
||||
attemptCount: batchSchema.fluxGrantBatchRecipient.attemptCount,
|
||||
lastAttemptedAt: batchSchema.fluxGrantBatchRecipient.lastAttemptedAt,
|
||||
})
|
||||
.from(batchSchema.fluxGrantBatchRecipient)
|
||||
.where(and(
|
||||
eq(batchSchema.fluxGrantBatchRecipient.batchId, batchId),
|
||||
eq(batchSchema.fluxGrantBatchRecipient.status, 'failed'),
|
||||
))
|
||||
.orderBy(desc(batchSchema.fluxGrantBatchRecipient.lastAttemptedAt))
|
||||
.limit(20)
|
||||
|
||||
return { batch: row, progress, recentFailures }
|
||||
},
|
||||
|
||||
/**
|
||||
* Paginated list. Cursor is the last seen createdAt+id pair, base64-encoded.
|
||||
*/
|
||||
async list(opts: { limit: number, cursor?: string, status?: string }) {
|
||||
const limit = Math.min(100, Math.max(1, opts.limit))
|
||||
let cursorTime: Date | null = null
|
||||
let cursorId: string | null = null
|
||||
|
||||
if (opts.cursor) {
|
||||
try {
|
||||
const decoded = JSON.parse(Buffer.from(opts.cursor, 'base64').toString('utf-8'))
|
||||
cursorTime = new Date(decoded.t)
|
||||
cursorId = decoded.i
|
||||
}
|
||||
catch {
|
||||
// bad cursor → treat as no cursor
|
||||
}
|
||||
}
|
||||
|
||||
const whereParts = []
|
||||
if (opts.status)
|
||||
whereParts.push(eq(batchSchema.fluxGrantBatch.status, opts.status))
|
||||
if (cursorTime && cursorId) {
|
||||
whereParts.push(sql`(${batchSchema.fluxGrantBatch.createdAt}, ${batchSchema.fluxGrantBatch.id}) < (${cursorTime}, ${cursorId})`)
|
||||
}
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(batchSchema.fluxGrantBatch)
|
||||
.where(whereParts.length > 0 ? and(...whereParts) : undefined)
|
||||
.orderBy(desc(batchSchema.fluxGrantBatch.createdAt), desc(batchSchema.fluxGrantBatch.id))
|
||||
.limit(limit + 1)
|
||||
|
||||
const hasMore = rows.length > limit
|
||||
const items = hasMore ? rows.slice(0, limit) : rows
|
||||
|
||||
let nextCursor: string | null = null
|
||||
if (hasMore) {
|
||||
const last = items[items.length - 1]!
|
||||
nextCursor = Buffer.from(JSON.stringify({ t: last.createdAt.toISOString(), i: last.id })).toString('base64')
|
||||
}
|
||||
|
||||
return { batches: items, nextCursor }
|
||||
},
|
||||
|
||||
/**
|
||||
* Reset all `failed` recipients in a batch back to `pending` so the worker
|
||||
* picks them up on the next poll. Idempotent: returns 0 when nothing is failed.
|
||||
*
|
||||
* Returns:
|
||||
* - { retriedCount } — number of rows transitioned from failed → pending
|
||||
*/
|
||||
async retryFailed(batchId: string) {
|
||||
const updated = await db
|
||||
.update(batchSchema.fluxGrantBatchRecipient)
|
||||
.set({
|
||||
status: 'pending',
|
||||
attemptCount: 0,
|
||||
lastAttemptedAt: null,
|
||||
errorReason: null,
|
||||
})
|
||||
.where(and(
|
||||
eq(batchSchema.fluxGrantBatchRecipient.batchId, batchId),
|
||||
eq(batchSchema.fluxGrantBatchRecipient.status, 'failed'),
|
||||
))
|
||||
.returning({ id: batchSchema.fluxGrantBatchRecipient.id })
|
||||
|
||||
// If a batch was completed and we re-opened pending rows, flip its
|
||||
// status back to running so the worker picks it up.
|
||||
if (updated.length > 0) {
|
||||
await db
|
||||
.update(batchSchema.fluxGrantBatch)
|
||||
.set({ status: 'running', completedAt: null })
|
||||
.where(and(
|
||||
eq(batchSchema.fluxGrantBatch.id, batchId),
|
||||
inArray(batchSchema.fluxGrantBatch.status, ['completed', 'failed_partial']),
|
||||
))
|
||||
}
|
||||
|
||||
logger.withFields({ batchId, retriedCount: updated.length }).log('Retry failed recipients')
|
||||
return { retriedCount: updated.length }
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export type FluxGrantBatchService = ReturnType<typeof createFluxGrantBatchService>
|
||||
|
||||
/**
|
||||
* Exported for test reuse.
|
||||
*/
|
||||
export { resolveEmails }
|
||||
|
||||
/**
|
||||
* Worker-side query: count remaining pending recipients for a batch. Used to
|
||||
* decide when to flip batch status to completed/failed_partial.
|
||||
*
|
||||
* Returns:
|
||||
* - { pending, failed, total } counts
|
||||
*/
|
||||
export async function getBatchTerminalCheck(db: Database, batchId: string) {
|
||||
const rows = await db
|
||||
.select({ status: batchSchema.fluxGrantBatchRecipient.status, count: sql<number>`count(*)::int` })
|
||||
.from(batchSchema.fluxGrantBatchRecipient)
|
||||
.where(eq(batchSchema.fluxGrantBatchRecipient.batchId, batchId))
|
||||
.groupBy(batchSchema.fluxGrantBatchRecipient.status)
|
||||
|
||||
let pending = 0
|
||||
let failed = 0
|
||||
let total = 0
|
||||
for (const r of rows) {
|
||||
total += r.count
|
||||
if (r.status === 'pending')
|
||||
pending = r.count
|
||||
else if (r.status === 'failed')
|
||||
failed = r.count
|
||||
}
|
||||
return { pending, failed, total }
|
||||
}
|
||||
|
||||
/**
|
||||
* Worker-side query: find batches that are still in `created` or `running`
|
||||
* and have at least one pending recipient whose backoff has elapsed. Workers
|
||||
* iterate this set so they don't waste polls on empty/finished batches.
|
||||
*
|
||||
* NOTICE:
|
||||
* Why this isn't `selectDistinct(...).innerJoin(recipient)`:
|
||||
* Postgres rejects `SELECT DISTINCT ... ORDER BY col` when `col` isn't in
|
||||
* the select list (sqlstate 42P10). We previously had `orderBy(createdAt)`
|
||||
* with only `id` + `status` selected → runtime crash on the first worker
|
||||
* tick (caught during local dev 2026-05-08).
|
||||
*
|
||||
* Switched to a plain `SELECT … WHERE EXISTS (recipient row meeting
|
||||
* criteria)`, which is also conceptually right: "show me batches that
|
||||
* have at least one ready-to-process recipient", not "join then dedupe".
|
||||
*/
|
||||
export async function findActiveBatches(db: Database, now: Date) {
|
||||
const rows = await db
|
||||
.select({
|
||||
id: batchSchema.fluxGrantBatch.id,
|
||||
status: batchSchema.fluxGrantBatch.status,
|
||||
})
|
||||
.from(batchSchema.fluxGrantBatch)
|
||||
.where(and(
|
||||
inArray(batchSchema.fluxGrantBatch.status, ['created', 'running']),
|
||||
sql`EXISTS (
|
||||
SELECT 1 FROM ${batchSchema.fluxGrantBatchRecipient}
|
||||
WHERE ${batchSchema.fluxGrantBatchRecipient.batchId} = ${batchSchema.fluxGrantBatch.id}
|
||||
AND ${batchSchema.fluxGrantBatchRecipient.status} = 'pending'
|
||||
AND (${batchSchema.fluxGrantBatchRecipient.lastAttemptedAt} IS NULL
|
||||
OR ${batchSchema.fluxGrantBatchRecipient.lastAttemptedAt} < ${now})
|
||||
)`,
|
||||
))
|
||||
.orderBy(batchSchema.fluxGrantBatch.createdAt)
|
||||
|
||||
return rows
|
||||
}
|
||||
|
||||
/**
|
||||
* Worker-side helper: mark a batch started (status='running', startedAt set)
|
||||
* if it's still 'created'. Idempotent: no-op when already running.
|
||||
*/
|
||||
export async function markBatchStartedIfNeeded(db: Database, batchId: string, now: Date) {
|
||||
await db
|
||||
.update(batchSchema.fluxGrantBatch)
|
||||
.set({ status: 'running', startedAt: now })
|
||||
.where(and(
|
||||
eq(batchSchema.fluxGrantBatch.id, batchId),
|
||||
eq(batchSchema.fluxGrantBatch.status, 'created'),
|
||||
isNull(batchSchema.fluxGrantBatch.startedAt),
|
||||
))
|
||||
}
|
||||
|
||||
/**
|
||||
* Worker-side helper: when no pending rows remain, flip the batch to a
|
||||
* terminal status. `failed_partial` if any failed/skipped, else `completed`.
|
||||
* Idempotent: skips batches already in terminal status.
|
||||
*/
|
||||
export async function finalizeBatchIfDone(db: Database, batchId: string, now: Date) {
|
||||
const counts = await getBatchTerminalCheck(db, batchId)
|
||||
if (counts.pending > 0)
|
||||
return
|
||||
|
||||
const terminal = counts.failed > 0 ? 'failed_partial' : 'completed'
|
||||
|
||||
await db
|
||||
.update(batchSchema.fluxGrantBatch)
|
||||
.set({ status: terminal, completedAt: now })
|
||||
.where(and(
|
||||
eq(batchSchema.fluxGrantBatch.id, batchId),
|
||||
inArray(batchSchema.fluxGrantBatch.status, ['created', 'running']),
|
||||
))
|
||||
}
|
||||
@@ -1,360 +0,0 @@
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { BillingService } from '../billing/billing-service'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { errorMessageFrom } from '@moeru/std'
|
||||
import { and, asc, eq, isNull, or, sql } from 'drizzle-orm'
|
||||
|
||||
import {
|
||||
finalizeBatchIfDone,
|
||||
findActiveBatches,
|
||||
markBatchStartedIfNeeded,
|
||||
} from './flux-grant-batch-service'
|
||||
|
||||
import * as fluxSchema from '../../schemas/flux'
|
||||
import * as batchSchema from '../../schemas/flux-grant-batch'
|
||||
|
||||
const logger = useLogger('admin-flux-grant-batch-worker').useGlobalConfig()
|
||||
|
||||
export interface FluxGrantBatchWorkerOptions {
|
||||
/**
|
||||
* How many pending recipients the worker tries to process per polling tick.
|
||||
* @default 50
|
||||
*/
|
||||
batchSize?: number
|
||||
/**
|
||||
* How many grants to issue per second across this worker instance.
|
||||
* Sleep between grant calls is `1000 / throttlePerSec`.
|
||||
* @default 50
|
||||
*/
|
||||
throttlePerSec?: number
|
||||
/**
|
||||
* Maximum number of attempts before a recipient is marked `failed` permanently.
|
||||
* Operator can re-arm via `POST /api/admin/flux-grant-batches/:id/retry`.
|
||||
* @default 3
|
||||
*/
|
||||
maxAttempts?: number
|
||||
/**
|
||||
* Sleep duration when no work is available, in ms.
|
||||
* @default 1000
|
||||
*/
|
||||
idleSleepMs?: number
|
||||
}
|
||||
|
||||
const DEFAULTS = {
|
||||
batchSize: 50,
|
||||
throttlePerSec: 50,
|
||||
maxAttempts: 3,
|
||||
idleSleepMs: 1000,
|
||||
} as const
|
||||
|
||||
/**
|
||||
* Compute the next-attempt cooldown for a recipient whose previous attempt failed.
|
||||
*
|
||||
* Schedule:
|
||||
* - attempt 0 → 0s (never attempted)
|
||||
* - attempt 1 → 30s
|
||||
* - attempt 2 → 5min
|
||||
* - attempt 3+ → terminal (handled by caller, not this fn)
|
||||
*/
|
||||
export function backoffMs(attempt: number): number {
|
||||
if (attempt <= 0)
|
||||
return 0
|
||||
if (attempt === 1)
|
||||
return 30_000
|
||||
return 5 * 60_000
|
||||
}
|
||||
|
||||
/**
|
||||
* Sleep that wakes up early when the abort signal fires.
|
||||
*/
|
||||
function sleep(ms: number, signal: AbortSignal): Promise<void> {
|
||||
if (ms <= 0)
|
||||
return Promise.resolve()
|
||||
return new Promise((resolve) => {
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
const onAbort = () => {
|
||||
if (timer != null)
|
||||
clearTimeout(timer)
|
||||
resolve()
|
||||
}
|
||||
timer = setTimeout(() => {
|
||||
signal.removeEventListener('abort', onAbort)
|
||||
resolve()
|
||||
}, ms)
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Claim up to `batchSize` pending recipients whose backoff has elapsed.
|
||||
*
|
||||
* Use when:
|
||||
* - Worker tick wants a fresh batch of recipients to process
|
||||
*
|
||||
* Expects:
|
||||
* - Caller is OK with rows being locked for the duration of `processBatch`;
|
||||
* we don't pre-commit a "claimed" status because the row stays locked under
|
||||
* `FOR UPDATE SKIP LOCKED` until the outer transaction completes.
|
||||
*
|
||||
* Returns:
|
||||
* - Up to `batchSize` rows joined with parent batch meta. Empty array means
|
||||
* no work for this batch right now.
|
||||
*/
|
||||
async function selectClaimablePending(
|
||||
db: Database,
|
||||
batchId: string,
|
||||
batchSize: number,
|
||||
now: Date,
|
||||
) {
|
||||
// Backoff cutoffs are encoded as a SQL expression: a row is claimable if
|
||||
// last_attempted_at IS NULL OR (now - last_attempted_at) >= backoff(attempt_count).
|
||||
// Step function: ≤0 attempts → 0s, =1 → 30s, ≥2 → 5min.
|
||||
return db
|
||||
.select({
|
||||
recipientId: batchSchema.fluxGrantBatchRecipient.id,
|
||||
batchId: batchSchema.fluxGrantBatchRecipient.batchId,
|
||||
userId: batchSchema.fluxGrantBatchRecipient.userId,
|
||||
inputEmail: batchSchema.fluxGrantBatchRecipient.inputEmail,
|
||||
attemptCount: batchSchema.fluxGrantBatchRecipient.attemptCount,
|
||||
batchName: batchSchema.fluxGrantBatch.name,
|
||||
batchAmount: batchSchema.fluxGrantBatch.amount,
|
||||
batchDescription: batchSchema.fluxGrantBatch.description,
|
||||
})
|
||||
.from(batchSchema.fluxGrantBatchRecipient)
|
||||
.innerJoin(
|
||||
batchSchema.fluxGrantBatch,
|
||||
eq(batchSchema.fluxGrantBatch.id, batchSchema.fluxGrantBatchRecipient.batchId),
|
||||
)
|
||||
.where(and(
|
||||
eq(batchSchema.fluxGrantBatchRecipient.batchId, batchId),
|
||||
eq(batchSchema.fluxGrantBatchRecipient.status, 'pending'),
|
||||
or(
|
||||
isNull(batchSchema.fluxGrantBatchRecipient.lastAttemptedAt),
|
||||
sql`${batchSchema.fluxGrantBatchRecipient.lastAttemptedAt} +
|
||||
(CASE
|
||||
WHEN ${batchSchema.fluxGrantBatchRecipient.attemptCount} <= 0 THEN INTERVAL '0 seconds'
|
||||
WHEN ${batchSchema.fluxGrantBatchRecipient.attemptCount} = 1 THEN INTERVAL '30 seconds'
|
||||
ELSE INTERVAL '5 minutes'
|
||||
END) <= ${now}`,
|
||||
),
|
||||
))
|
||||
.orderBy(asc(batchSchema.fluxGrantBatchRecipient.createdAt))
|
||||
.limit(batchSize)
|
||||
.for('update', { skipLocked: true })
|
||||
}
|
||||
|
||||
/**
|
||||
* Process one recipient: re-check user soft-delete, call creditFlux, mark outcome.
|
||||
*
|
||||
* Use when:
|
||||
* - Inside the transaction holding the `FOR UPDATE SKIP LOCKED` row
|
||||
*
|
||||
* Expects:
|
||||
* - `userId` is non-null (NULL recipients are pre-skipped at creation time and
|
||||
* never enter the worker queue)
|
||||
*
|
||||
* Returns:
|
||||
* - 'granted' | 'skipped' | 'failed_transient' | 'failed_permanent'
|
||||
*/
|
||||
async function processSingleRecipient(
|
||||
deps: { db: Database, billingService: BillingService },
|
||||
recipient: {
|
||||
recipientId: string
|
||||
batchId: string
|
||||
userId: string
|
||||
inputEmail: string
|
||||
attemptCount: number
|
||||
batchName: string
|
||||
batchAmount: number
|
||||
batchDescription: string | null
|
||||
},
|
||||
maxAttempts: number,
|
||||
now: Date,
|
||||
): Promise<'granted' | 'skipped' | 'failed_transient' | 'failed_permanent'> {
|
||||
// Re-check soft-delete. The user might have signed off between batch
|
||||
// creation and worker pick-up.
|
||||
const [fluxRow] = await deps.db
|
||||
.select({ deletedAt: fluxSchema.userFlux.deletedAt })
|
||||
.from(fluxSchema.userFlux)
|
||||
.where(eq(fluxSchema.userFlux.userId, recipient.userId))
|
||||
.limit(1)
|
||||
|
||||
if (fluxRow && fluxRow.deletedAt != null) {
|
||||
await deps.db
|
||||
.update(batchSchema.fluxGrantBatchRecipient)
|
||||
.set({
|
||||
status: 'skipped',
|
||||
errorReason: 'user_deleted_after_resolution',
|
||||
lastAttemptedAt: now,
|
||||
})
|
||||
.where(eq(batchSchema.fluxGrantBatchRecipient.id, recipient.recipientId))
|
||||
return 'skipped'
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await deps.billingService.creditFlux({
|
||||
userId: recipient.userId,
|
||||
amount: recipient.batchAmount,
|
||||
type: 'promo',
|
||||
requestId: `flux-grant-batch-${recipient.batchId}-${recipient.recipientId}`,
|
||||
description: recipient.batchDescription ?? `Flux grant batch: ${recipient.batchName}`,
|
||||
source: 'admin_promo',
|
||||
auditMetadata: {
|
||||
batchId: recipient.batchId,
|
||||
batchName: recipient.batchName,
|
||||
recipientId: recipient.recipientId,
|
||||
},
|
||||
})
|
||||
|
||||
await deps.db
|
||||
.update(batchSchema.fluxGrantBatchRecipient)
|
||||
.set({
|
||||
status: 'granted',
|
||||
attemptCount: recipient.attemptCount + 1,
|
||||
lastAttemptedAt: now,
|
||||
fluxTransactionId: result.fluxTransactionId,
|
||||
errorReason: null,
|
||||
})
|
||||
.where(eq(batchSchema.fluxGrantBatchRecipient.id, recipient.recipientId))
|
||||
|
||||
return 'granted'
|
||||
}
|
||||
catch (err) {
|
||||
const nextAttempt = recipient.attemptCount + 1
|
||||
const errorMessage = errorMessageFrom(err) ?? 'Unknown error'
|
||||
const isPermanent = nextAttempt >= maxAttempts
|
||||
|
||||
await deps.db
|
||||
.update(batchSchema.fluxGrantBatchRecipient)
|
||||
.set({
|
||||
status: isPermanent ? 'failed' : 'pending',
|
||||
attemptCount: nextAttempt,
|
||||
lastAttemptedAt: now,
|
||||
errorReason: errorMessage.slice(0, 500),
|
||||
})
|
||||
.where(eq(batchSchema.fluxGrantBatchRecipient.id, recipient.recipientId))
|
||||
|
||||
logger.withError(err).withFields({
|
||||
recipientId: recipient.recipientId,
|
||||
batchId: recipient.batchId,
|
||||
userId: recipient.userId,
|
||||
attempt: nextAttempt,
|
||||
isPermanent,
|
||||
}).warn('Recipient grant attempt failed')
|
||||
|
||||
return isPermanent ? 'failed_permanent' : 'failed_transient'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the flux grant batch worker polling loop until `signal` aborts.
|
||||
*
|
||||
* Use when:
|
||||
* - Started alongside the billing-consumer role in `bin/run-billing-consumer.ts`
|
||||
*
|
||||
* Expects:
|
||||
* - `billingService` writes to the same DB / Redis / billing stream as the
|
||||
* API process (multi-instance Railway deployment)
|
||||
*
|
||||
* Call stack:
|
||||
*
|
||||
* runBillingConsumer (../bin/run-billing-consumer)
|
||||
* -> {@link runFluxGrantBatchWorker}
|
||||
* -> {@link findActiveBatches}
|
||||
* -> {@link selectClaimablePending} (FOR UPDATE SKIP LOCKED)
|
||||
* -> {@link processSingleRecipient}
|
||||
* -> billingService.creditFlux (writes ledger + cache + event)
|
||||
* -> {@link finalizeBatchIfDone}
|
||||
*/
|
||||
export async function runFluxGrantBatchWorker(
|
||||
deps: { db: Database, billingService: BillingService },
|
||||
signal: AbortSignal,
|
||||
options: FluxGrantBatchWorkerOptions = {},
|
||||
): Promise<void> {
|
||||
const batchSize = options.batchSize ?? DEFAULTS.batchSize
|
||||
const throttlePerSec = options.throttlePerSec ?? DEFAULTS.throttlePerSec
|
||||
const maxAttempts = options.maxAttempts ?? DEFAULTS.maxAttempts
|
||||
const idleSleepMs = options.idleSleepMs ?? DEFAULTS.idleSleepMs
|
||||
const perGrantSleepMs = Math.max(0, Math.floor(1000 / throttlePerSec))
|
||||
|
||||
logger.withFields({ batchSize, throttlePerSec, maxAttempts, idleSleepMs }).log('Flux grant batch worker started')
|
||||
|
||||
while (!signal.aborted) {
|
||||
try {
|
||||
const now = new Date()
|
||||
const active = await findActiveBatches(deps.db, now)
|
||||
|
||||
if (active.length === 0) {
|
||||
await sleep(idleSleepMs, signal)
|
||||
continue
|
||||
}
|
||||
|
||||
let totalProcessed = 0
|
||||
|
||||
for (const batch of active) {
|
||||
if (signal.aborted)
|
||||
break
|
||||
|
||||
await markBatchStartedIfNeeded(deps.db, batch.id, now)
|
||||
|
||||
// Process the batch in transactional chunks. Each chunk holds
|
||||
// FOR UPDATE SKIP LOCKED on its rows for the duration of the tx.
|
||||
await deps.db.transaction(async (tx) => {
|
||||
const claimed = await selectClaimablePending(tx as unknown as Database, batch.id, batchSize, now)
|
||||
if (claimed.length === 0)
|
||||
return
|
||||
|
||||
for (const recipient of claimed) {
|
||||
if (signal.aborted)
|
||||
break
|
||||
if (recipient.userId == null) {
|
||||
// Defensive: NULL userId rows should never be 'pending' (they're
|
||||
// resolved as 'skipped' at creation), but if one slips in mark it
|
||||
// skipped here too.
|
||||
await (tx as unknown as Database)
|
||||
.update(batchSchema.fluxGrantBatchRecipient)
|
||||
.set({ status: 'skipped', errorReason: 'not_found', lastAttemptedAt: now })
|
||||
.where(eq(batchSchema.fluxGrantBatchRecipient.id, recipient.recipientId))
|
||||
totalProcessed++
|
||||
continue
|
||||
}
|
||||
|
||||
await processSingleRecipient(
|
||||
{ db: tx as unknown as Database, billingService: deps.billingService },
|
||||
{
|
||||
recipientId: recipient.recipientId,
|
||||
batchId: recipient.batchId,
|
||||
userId: recipient.userId,
|
||||
inputEmail: recipient.inputEmail,
|
||||
attemptCount: recipient.attemptCount,
|
||||
batchName: recipient.batchName,
|
||||
batchAmount: recipient.batchAmount,
|
||||
batchDescription: recipient.batchDescription,
|
||||
},
|
||||
maxAttempts,
|
||||
now,
|
||||
)
|
||||
|
||||
totalProcessed++
|
||||
if (perGrantSleepMs > 0)
|
||||
await sleep(perGrantSleepMs, signal)
|
||||
}
|
||||
})
|
||||
|
||||
// After a chunk run, see whether this batch is now done.
|
||||
await finalizeBatchIfDone(deps.db, batch.id, new Date())
|
||||
}
|
||||
|
||||
// No work this tick → idle sleep so we don't hammer the DB.
|
||||
if (totalProcessed === 0)
|
||||
await sleep(idleSleepMs, signal)
|
||||
}
|
||||
catch (err) {
|
||||
logger.withError(err).error('Flux grant batch worker tick failed; sleeping before retry')
|
||||
await sleep(idleSleepMs, signal)
|
||||
}
|
||||
}
|
||||
|
||||
logger.log('Flux grant batch worker stopped')
|
||||
}
|
||||
-251
@@ -1,251 +0,0 @@
|
||||
import type { Database } from '../../../libs/db'
|
||||
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { beforeAll, beforeEach, describe, expect, it } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../../libs/mock-db'
|
||||
import { createFluxGrantBatchService, resolveEmails } from '../flux-grant-batch-service'
|
||||
|
||||
import * as schema from '../../../schemas'
|
||||
|
||||
describe('resolveEmails', () => {
|
||||
let db: Database
|
||||
|
||||
beforeAll(async () => {
|
||||
db = await mockDB(schema)
|
||||
|
||||
// Three users: one normal, one with deleted user_flux, one we never insert
|
||||
// user_flux for at all (so it shows up as "user exists, no flux row" → still
|
||||
// pending since user.id matches; default flux init happens at credit time).
|
||||
await db.insert(schema.user).values([
|
||||
{ id: 'uid_normal', name: 'Normal', email: 'Normal@Example.com' },
|
||||
{ id: 'uid_deleted', name: 'Deleted', email: 'deleted@example.com' },
|
||||
{ id: 'uid_no_flux', name: 'NoFlux', email: 'noflux@example.com' },
|
||||
])
|
||||
|
||||
await db.insert(schema.userFlux).values([
|
||||
{ userId: 'uid_normal', flux: 100 },
|
||||
{ userId: 'uid_deleted', flux: 0, deletedAt: new Date() },
|
||||
])
|
||||
})
|
||||
|
||||
beforeEach(async () => {
|
||||
// No state to reset between tests — resolveEmails is read-only.
|
||||
})
|
||||
|
||||
it('matches users case-insensitively against the user table', async () => {
|
||||
const { resolved, summary } = await resolveEmails(db, ['NORMAL@example.com'], 200, 50)
|
||||
expect(resolved).toHaveLength(1)
|
||||
expect(resolved[0]).toMatchObject({
|
||||
inputEmail: 'NORMAL@example.com',
|
||||
userId: 'uid_normal',
|
||||
status: 'pending',
|
||||
errorReason: null,
|
||||
})
|
||||
expect(summary.willGrant).toBe(1)
|
||||
expect(summary.totalFluxToIssue).toBe(200)
|
||||
})
|
||||
|
||||
it('marks unknown emails as not_found', async () => {
|
||||
const { resolved, summary } = await resolveEmails(db, ['ghost@example.com'], 200, 50)
|
||||
expect(resolved[0]).toMatchObject({
|
||||
inputEmail: 'ghost@example.com',
|
||||
userId: null,
|
||||
status: 'skipped',
|
||||
errorReason: 'not_found',
|
||||
})
|
||||
expect(summary.willGrant).toBe(0)
|
||||
expect(summary.willSkip.notFound).toBe(1)
|
||||
})
|
||||
|
||||
it('marks soft-deleted users as user_deleted (userId still attached for audit)', async () => {
|
||||
const { resolved, summary } = await resolveEmails(db, ['deleted@example.com'], 200, 50)
|
||||
expect(resolved[0]).toMatchObject({
|
||||
inputEmail: 'deleted@example.com',
|
||||
userId: 'uid_deleted',
|
||||
status: 'skipped',
|
||||
errorReason: 'user_deleted',
|
||||
})
|
||||
expect(summary.willSkip.userDeleted).toBe(1)
|
||||
})
|
||||
|
||||
it('keeps the first occurrence and tags subsequent duplicates', async () => {
|
||||
const { resolved, summary } = await resolveEmails(
|
||||
db,
|
||||
['normal@example.com', 'NORMAL@EXAMPLE.COM', 'normal@example.com'],
|
||||
200,
|
||||
50,
|
||||
)
|
||||
expect(resolved).toHaveLength(3)
|
||||
expect(resolved[0].errorReason).toBeNull()
|
||||
expect(resolved[0].status).toBe('pending')
|
||||
expect(resolved[1].errorReason).toBe('duplicate_in_input')
|
||||
expect(resolved[2].errorReason).toBe('duplicate_in_input')
|
||||
expect(summary.willGrant).toBe(1)
|
||||
expect(summary.willSkip.duplicateInInput).toBe(2)
|
||||
})
|
||||
|
||||
it('caps preview samples at 5 entries per category', async () => {
|
||||
const ghosts = Array.from({ length: 12 }, (_, i) => `ghost${i}@example.com`)
|
||||
const { summary } = await resolveEmails(db, ghosts, 200, 50)
|
||||
expect(summary.samples.notFound).toHaveLength(5)
|
||||
expect(summary.willSkip.notFound).toBe(12)
|
||||
})
|
||||
|
||||
it('estimatedDurationSec rounds up based on throttle', async () => {
|
||||
const ten = Array.from({ length: 10 }, (_, i) => `ghost${i + 100}@example.com`)
|
||||
// 10 unknown emails → 0 willGrant → 0s estimate
|
||||
const { summary: zero } = await resolveEmails(db, ten, 200, 50)
|
||||
expect(zero.estimatedDurationSec).toBe(0)
|
||||
|
||||
// 1 grantable + throttle 50 → ceil(1/50) = 1s
|
||||
const { summary: one } = await resolveEmails(db, ['normal@example.com'], 200, 50)
|
||||
expect(one.estimatedDurationSec).toBe(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('createFluxGrantBatchService.create', () => {
|
||||
let db: Database
|
||||
|
||||
beforeAll(async () => {
|
||||
db = await mockDB(schema)
|
||||
|
||||
await db.insert(schema.user).values([
|
||||
{ id: 'uid_grant_a', name: 'A', email: 'a@example.com' },
|
||||
{ id: 'uid_grant_b', name: 'B', email: 'b@example.com' },
|
||||
])
|
||||
await db.insert(schema.userFlux).values([
|
||||
{ userId: 'uid_grant_a', flux: 0 },
|
||||
{ userId: 'uid_grant_b', flux: 0 },
|
||||
])
|
||||
})
|
||||
|
||||
beforeEach(async () => {
|
||||
await db.delete(schema.fluxGrantBatchRecipient)
|
||||
await db.delete(schema.fluxGrantBatch)
|
||||
})
|
||||
|
||||
it('persists batch + per-email recipient rows with resolution status', async () => {
|
||||
const service = createFluxGrantBatchService(db)
|
||||
const { batch, summary } = await service.create({
|
||||
name: 'Test Promo',
|
||||
amount: 200,
|
||||
emails: ['a@example.com', 'unknown@example.com'],
|
||||
createdByUserId: 'uid_admin',
|
||||
throttlePerSec: 50,
|
||||
})
|
||||
|
||||
expect(batch.status).toBe('created')
|
||||
expect(batch.createdByUserId).toBe('uid_admin')
|
||||
expect(summary.willGrant).toBe(1)
|
||||
expect(summary.willSkip.notFound).toBe(1)
|
||||
|
||||
const recipients = await db.select().from(schema.fluxGrantBatchRecipient).where(
|
||||
eq(schema.fluxGrantBatchRecipient.batchId, batch.id),
|
||||
)
|
||||
expect(recipients).toHaveLength(2)
|
||||
|
||||
const aRecipient = recipients.find(r => r.inputEmail === 'a@example.com')!
|
||||
expect(aRecipient.status).toBe('pending')
|
||||
expect(aRecipient.userId).toBe('uid_grant_a')
|
||||
|
||||
const unknownRecipient = recipients.find(r => r.inputEmail === 'unknown@example.com')!
|
||||
expect(unknownRecipient.status).toBe('skipped')
|
||||
expect(unknownRecipient.errorReason).toBe('not_found')
|
||||
expect(unknownRecipient.userId).toBeNull()
|
||||
})
|
||||
|
||||
it('returns a stable preview summary on dry-run without persisting anything', async () => {
|
||||
const service = createFluxGrantBatchService(db)
|
||||
const before = await db.select().from(schema.fluxGrantBatch)
|
||||
const summary = await service.preview({
|
||||
name: 'Preview Test',
|
||||
amount: 50,
|
||||
emails: ['a@example.com', 'b@example.com', 'unknown@example.com'],
|
||||
throttlePerSec: 50,
|
||||
})
|
||||
expect(summary.willGrant).toBe(2)
|
||||
expect(summary.willSkip.notFound).toBe(1)
|
||||
expect(summary.totalFluxToIssue).toBe(100)
|
||||
const after = await db.select().from(schema.fluxGrantBatch)
|
||||
expect(after).toHaveLength(before.length)
|
||||
})
|
||||
})
|
||||
|
||||
describe('createFluxGrantBatchService.retryFailed', () => {
|
||||
let db: Database
|
||||
|
||||
beforeAll(async () => {
|
||||
db = await mockDB(schema)
|
||||
})
|
||||
|
||||
beforeEach(async () => {
|
||||
await db.delete(schema.fluxGrantBatchRecipient)
|
||||
await db.delete(schema.fluxGrantBatch)
|
||||
})
|
||||
|
||||
it('moves failed recipients back to pending and reopens completed batches', async () => {
|
||||
const service = createFluxGrantBatchService(db)
|
||||
|
||||
const [batch] = await db.insert(schema.fluxGrantBatch).values({
|
||||
name: 'Retry Test',
|
||||
type: 'promo',
|
||||
amount: 100,
|
||||
status: 'completed',
|
||||
createdByUserId: 'uid_admin',
|
||||
completedAt: new Date(),
|
||||
}).returning()
|
||||
|
||||
await db.insert(schema.fluxGrantBatchRecipient).values([
|
||||
{
|
||||
batchId: batch!.id,
|
||||
inputEmail: 'failed1@example.com',
|
||||
userId: 'uid_failed1',
|
||||
status: 'failed',
|
||||
attemptCount: 3,
|
||||
errorReason: 'DB timeout',
|
||||
lastAttemptedAt: new Date(),
|
||||
},
|
||||
{
|
||||
batchId: batch!.id,
|
||||
inputEmail: 'granted@example.com',
|
||||
userId: 'uid_granted',
|
||||
status: 'granted',
|
||||
attemptCount: 1,
|
||||
},
|
||||
])
|
||||
|
||||
const result = await service.retryFailed(batch!.id)
|
||||
expect(result.retriedCount).toBe(1)
|
||||
|
||||
const [reopened] = await db.select().from(schema.fluxGrantBatch).where(eq(schema.fluxGrantBatch.id, batch!.id))
|
||||
expect(reopened?.status).toBe('running')
|
||||
expect(reopened?.completedAt).toBeNull()
|
||||
|
||||
const recipients = await db.select().from(schema.fluxGrantBatchRecipient).where(
|
||||
eq(schema.fluxGrantBatchRecipient.batchId, batch!.id),
|
||||
)
|
||||
const failedRecipient = recipients.find(r => r.inputEmail === 'failed1@example.com')!
|
||||
expect(failedRecipient.status).toBe('pending')
|
||||
expect(failedRecipient.attemptCount).toBe(0)
|
||||
expect(failedRecipient.errorReason).toBeNull()
|
||||
expect(failedRecipient.lastAttemptedAt).toBeNull()
|
||||
})
|
||||
|
||||
it('is a no-op when no failed recipients exist (idempotent)', async () => {
|
||||
const service = createFluxGrantBatchService(db)
|
||||
const [batch] = await db.insert(schema.fluxGrantBatch).values({
|
||||
name: 'Idempotent Retry',
|
||||
type: 'promo',
|
||||
amount: 100,
|
||||
status: 'completed',
|
||||
createdByUserId: 'uid_admin',
|
||||
}).returning()
|
||||
|
||||
const result = await service.retryFailed(batch!.id)
|
||||
expect(result.retriedCount).toBe(0)
|
||||
|
||||
const [unchanged] = await db.select().from(schema.fluxGrantBatch).where(eq(schema.fluxGrantBatch.id, batch!.id))
|
||||
expect(unchanged?.status).toBe('completed')
|
||||
})
|
||||
})
|
||||
@@ -1,19 +0,0 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { backoffMs } from '../flux-grant-batch-worker'
|
||||
|
||||
describe('backoffMs', () => {
|
||||
it('returns 0 for unattempted recipients', () => {
|
||||
expect(backoffMs(0)).toBe(0)
|
||||
expect(backoffMs(-1)).toBe(0)
|
||||
})
|
||||
|
||||
it('returns 30s after the first attempt', () => {
|
||||
expect(backoffMs(1)).toBe(30_000)
|
||||
})
|
||||
|
||||
it('returns 5min after the second and subsequent attempts', () => {
|
||||
expect(backoffMs(2)).toBe(300_000)
|
||||
expect(backoffMs(5)).toBe(300_000)
|
||||
})
|
||||
})
|
||||
+4
-4
@@ -1,12 +1,12 @@
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { BillingService } from '../billing/billing-service'
|
||||
import type { Database } from '../../../../libs/db'
|
||||
import type { BillingService } from '../../billing/billing-service'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { errorMessageFrom } from '@moeru/std'
|
||||
import { inArray } from 'drizzle-orm'
|
||||
|
||||
import * as accountsSchema from '../../schemas/accounts'
|
||||
import * as fluxSchema from '../../schemas/flux'
|
||||
import * as accountsSchema from '../../../../schemas/accounts'
|
||||
import * as fluxSchema from '../../../../schemas/flux'
|
||||
|
||||
const logger = useLogger('admin-flux-grants').useGlobalConfig()
|
||||
|
||||
+4
-4
@@ -1,13 +1,13 @@
|
||||
import type { Database } from '../../../libs/db'
|
||||
import type { BillingService } from '../../billing/billing-service'
|
||||
import type { Database } from '../../../../../libs/db'
|
||||
import type { BillingService } from '../../../billing/billing-service'
|
||||
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createAdminFluxGrantsService, resolveEmails } from '..'
|
||||
import { mockDB } from '../../../libs/mock-db'
|
||||
import { mockDB } from '../../../../../libs/mock-db'
|
||||
|
||||
import * as schema from '../../../schemas'
|
||||
import * as schema from '../../../../../schemas'
|
||||
|
||||
describe('resolveEmails', () => {
|
||||
let db: Database
|
||||
+3
-3
@@ -1,12 +1,12 @@
|
||||
import type Redis from 'ioredis'
|
||||
import type { InferOutput } from 'valibot'
|
||||
|
||||
import type { EnvelopeCrypto } from '../../utils/envelope-crypto'
|
||||
import type { ConfigKVService, llmModelSchema, llmRouterConfigSchema, ttsModelSchema, ttsUpstreamSchema } from '../config-kv'
|
||||
import type { EnvelopeCrypto } from '../../../../utils/envelope-crypto'
|
||||
import type { ConfigKVService, llmModelSchema, llmRouterConfigSchema, ttsModelSchema, ttsUpstreamSchema } from '../../../adapters/config-kv'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
|
||||
import { createBadRequestError } from '../../utils/error'
|
||||
import { createBadRequestError } from '../../../../utils/error'
|
||||
|
||||
/**
|
||||
* AAD label used when encrypting/decrypting the streaming TTS upstream key.
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { ConfigKVService } from '../../config-kv'
|
||||
import type { ConfigKVService } from '../../../../adapters/config-kv'
|
||||
|
||||
import { randomBytes } from 'node:crypto'
|
||||
|
||||
@@ -15,7 +15,7 @@ import {
|
||||
createAdminRouterConfigService,
|
||||
redactCiphertext,
|
||||
} from '..'
|
||||
import { createEnvelopeCrypto } from '../../../utils/envelope-crypto'
|
||||
import { createEnvelopeCrypto } from '../../../../../utils/envelope-crypto'
|
||||
|
||||
function freshEnvelope() {
|
||||
return createEnvelopeCrypto({ masterKey: randomBytes(32) })
|
||||
+8
-8
@@ -1,18 +1,18 @@
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { RevenueMetrics } from '../../otel'
|
||||
import type { ConfigKVService } from '../config-kv'
|
||||
import type { Database } from '../../../libs/db'
|
||||
import type { RevenueMetrics } from '../../../otel'
|
||||
import type { ConfigKVService } from '../../adapters/config-kv'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, eq } from 'drizzle-orm'
|
||||
|
||||
import { createPaymentRequiredError } from '../../utils/error'
|
||||
import { userFluxRedisKey } from '../../utils/redis-keys'
|
||||
import { createPaymentRequiredError } from '../../../utils/error'
|
||||
import { userFluxRedisKey } from '../../../utils/redis-keys'
|
||||
|
||||
import * as fluxSchema from '../../schemas/flux'
|
||||
import * as fluxTxSchema from '../../schemas/flux-transaction'
|
||||
import * as stripeSchema from '../../schemas/stripe'
|
||||
import * as fluxSchema from '../../../schemas/flux'
|
||||
import * as fluxTxSchema from '../../../schemas/flux-transaction'
|
||||
import * as stripeSchema from '../../../schemas/stripe'
|
||||
|
||||
const logger = useLogger('billing-service')
|
||||
|
||||
+4
-4
@@ -1,13 +1,13 @@
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { RevenueMetrics } from '../../otel'
|
||||
import type { RevenueMetrics } from '../../../otel'
|
||||
import type { BillingService } from './billing-service'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
|
||||
import { createPaymentRequiredError } from '../../utils/error'
|
||||
import { GEN_AI_ATTR_REQUEST_MODEL } from '../../utils/observability'
|
||||
import { userFluxMeterDebtRedisKey } from '../../utils/redis-keys'
|
||||
import { createPaymentRequiredError } from '../../../utils/error'
|
||||
import { GEN_AI_ATTR_REQUEST_MODEL } from '../../../utils/observability'
|
||||
import { userFluxMeterDebtRedisKey } from '../../../utils/redis-keys'
|
||||
|
||||
const logger = useLogger('flux-meter')
|
||||
|
||||
+5
-5
@@ -1,16 +1,16 @@
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { Database } from '../../../libs/db'
|
||||
import type { createConfigKVService } from '../../config-kv'
|
||||
import type { Database } from '../../../../libs/db'
|
||||
import type { createConfigKVService } from '../../../adapters/config-kv'
|
||||
|
||||
import { and, eq } from 'drizzle-orm'
|
||||
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../../libs/mock-db'
|
||||
import { userFluxRedisKey } from '../../../utils/redis-keys'
|
||||
import { mockDB } from '../../../../libs/mock-db'
|
||||
import { userFluxRedisKey } from '../../../../utils/redis-keys'
|
||||
import { createBillingService } from '../billing-service'
|
||||
|
||||
import * as schema from '../../../schemas'
|
||||
import * as schema from '../../../../schemas'
|
||||
|
||||
function createMockConfigKV(overrides: Record<string, number> = {}): ReturnType<typeof createConfigKVService> {
|
||||
const defaults: Record<string, number> = { INITIAL_USER_FLUX: 100, FLUX_PER_REQUEST: 1, ...overrides }
|
||||
+1
-1
@@ -3,7 +3,7 @@ import type { Database } from '../../libs/db'
|
||||
import { beforeAll, describe, expect, it } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { createCharacterService } from '../characters'
|
||||
import { createCharacterService } from './characters'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
+4
-4
@@ -1,11 +1,11 @@
|
||||
import type { Database } from '../libs/db'
|
||||
import type { EngagementMetrics } from '../otel'
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { EngagementMetrics } from '../../otel'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, eq, isNull, or, sql } from 'drizzle-orm'
|
||||
|
||||
import * as schema from '../schemas/characters'
|
||||
import * as userCharacterSchema from '../schemas/user-character'
|
||||
import * as schema from '../../schemas/characters'
|
||||
import * as userCharacterSchema from '../../schemas/user-character'
|
||||
|
||||
const logger = useLogger('characters')
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { clampLimit, resolveSenderId } from '../chats'
|
||||
import { clampLimit, resolveSenderId } from './chats'
|
||||
|
||||
describe('resolveSenderId', () => {
|
||||
it('returns userId for user role', () => {
|
||||
@@ -1,15 +1,15 @@
|
||||
import type { MessageRole, WireMessage } from '@proj-airi/server-sdk-shared'
|
||||
|
||||
import type { Database } from '../libs/db'
|
||||
import type { EngagementMetrics } from '../otel'
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { EngagementMetrics } from '../../otel'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, eq, gt, inArray, isNull, sql } from 'drizzle-orm'
|
||||
|
||||
import { createForbiddenError, createNotFoundError } from '../utils/error'
|
||||
import { nanoid } from '../utils/id'
|
||||
import { createForbiddenError, createNotFoundError } from '../../utils/error'
|
||||
import { nanoid } from '../../utils/id'
|
||||
|
||||
import * as schema from '../schemas/chats'
|
||||
import * as schema from '../../schemas/chats'
|
||||
|
||||
const logger = useLogger('chats')
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
import { beforeAll, describe, expect, it } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { createFluxTransactionService } from '../flux-transaction'
|
||||
import { createFluxTransactionService } from './flux-transaction'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
+2
-2
@@ -1,9 +1,9 @@
|
||||
import type { Database } from '../libs/db'
|
||||
import type { Database } from '../../libs/db'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, desc, eq, inArray } from 'drizzle-orm'
|
||||
|
||||
import * as schema from '../schemas/flux-transaction'
|
||||
import * as schema from '../../schemas/flux-transaction'
|
||||
|
||||
const logger = useLogger('flux-transaction')
|
||||
|
||||
+2
-2
@@ -1,14 +1,14 @@
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { createConfigKVService } from '../config-kv'
|
||||
import type { createConfigKVService } from '../adapters/config-kv'
|
||||
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { userFluxRedisKey } from '../../utils/redis-keys'
|
||||
import { createFluxService } from '../flux'
|
||||
import { createFluxService } from './flux'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { Database } from '../libs/db'
|
||||
import type { ConfigKVService } from './config-kv'
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { ConfigKVService } from '../adapters/config-kv'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, eq, isNull } from 'drizzle-orm'
|
||||
|
||||
import { userFluxRedisKey } from '../utils/redis-keys'
|
||||
import { userFluxRedisKey } from '../../utils/redis-keys'
|
||||
|
||||
import * as schema from '../schemas/flux'
|
||||
import * as fluxTxSchema from '../schemas/flux-transaction'
|
||||
import * as schema from '../../schemas/flux'
|
||||
import * as fluxTxSchema from '../../schemas/flux-transaction'
|
||||
|
||||
const logger = useLogger('flux-service')
|
||||
|
||||
+2
-2
@@ -1,9 +1,9 @@
|
||||
import type { ConfigKVService } from '../config-kv'
|
||||
import type { ConfigKVService } from '../../adapters/config-kv'
|
||||
import type { RouterConfig } from './types'
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { ApiError } from '../../utils/error'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
import { createConfigLoader } from './config-loader'
|
||||
|
||||
function makeConfig(): RouterConfig {
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
import type { ConfigKVService } from '../config-kv'
|
||||
import type { ConfigKVService } from '../../adapters/config-kv'
|
||||
import type { LlmModel, ModelKind, RouterConfig, TtsModel } from './types'
|
||||
|
||||
import { createBadRequestError, createServiceUnavailableError } from '../../utils/error'
|
||||
import { createBadRequestError, createServiceUnavailableError } from '../../../utils/error'
|
||||
|
||||
/**
|
||||
* Default TTL for the in-memory config cache. Plan KTD-4 fallback path:
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
import type { useLogger } from '@guiiai/logg'
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import type { GatewayMetrics } from '../../otel'
|
||||
import type { GatewayMetrics } from '../../../otel'
|
||||
import type { LlmRouterService } from './router'
|
||||
|
||||
/**
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { ApiError } from '../../utils/error'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
import { mapUpstreamError } from './error-mapping'
|
||||
|
||||
const exampleContext = { triedKeys: 2, triedUpstreams: 1, lastStatusCode: 401 as const }
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
import type { ApiError } from '../../utils/error'
|
||||
import type { ApiError } from '../../../utils/error'
|
||||
|
||||
import { createBadGatewayError, createGatewayTimeoutError, createInternalError, createServiceUnavailableError } from '../../utils/error'
|
||||
import { createBadGatewayError, createGatewayTimeoutError, createInternalError, createServiceUnavailableError } from '../../../utils/error'
|
||||
|
||||
/**
|
||||
* Sanitized context for `mapUpstreamError`.
|
||||
@@ -0,0 +1,6 @@
|
||||
export { createConfigSyncSubscriber } from './config-sync-subscriber'
|
||||
|
||||
export { createLlmRouterService } from './router'
|
||||
export type { LlmRouterService } from './router'
|
||||
|
||||
export type { LlmModel, TtsModel, TtsUpstream } from './types'
|
||||
+3
-3
@@ -2,14 +2,14 @@ import type { Buffer } from 'node:buffer'
|
||||
|
||||
import type { Counter } from '@opentelemetry/api'
|
||||
|
||||
import type { GatewayMetrics } from '../../otel'
|
||||
import type { GatewayMetrics } from '../../../otel'
|
||||
|
||||
import { randomBytes } from 'node:crypto'
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createEnvelopeCrypto } from '../../utils/envelope-crypto'
|
||||
import { ApiError } from '../../utils/error'
|
||||
import { createEnvelopeCrypto } from '../../../utils/envelope-crypto'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
import { createKeyRotator } from './key-rotator'
|
||||
|
||||
function freshMasterKey(): Buffer {
|
||||
+3
-3
@@ -1,11 +1,11 @@
|
||||
import type { Buffer } from 'node:buffer'
|
||||
|
||||
import type { GatewayMetrics } from '../../otel'
|
||||
import type { EnvelopeCrypto } from '../../utils/envelope-crypto'
|
||||
import type { GatewayMetrics } from '../../../otel'
|
||||
import type { EnvelopeCrypto } from '../../../utils/envelope-crypto'
|
||||
|
||||
import { errorMessageFrom } from '@moeru/std'
|
||||
|
||||
import { createServiceUnavailableError } from '../../utils/error'
|
||||
import { createServiceUnavailableError } from '../../../utils/error'
|
||||
|
||||
/**
|
||||
* Minimal shape of one upstream as needed by the rotator. We do not depend
|
||||
+4
-4
@@ -2,16 +2,16 @@ import type { Buffer } from 'node:buffer'
|
||||
|
||||
import type { Counter } from '@opentelemetry/api'
|
||||
|
||||
import type { GatewayMetrics } from '../../otel'
|
||||
import type { ConfigKVService } from '../config-kv'
|
||||
import type { GatewayMetrics } from '../../../otel'
|
||||
import type { ConfigKVService } from '../../adapters/config-kv'
|
||||
import type { RouterConfig } from './types'
|
||||
|
||||
import { randomBytes } from 'node:crypto'
|
||||
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createEnvelopeCrypto } from '../../utils/envelope-crypto'
|
||||
import { ApiError } from '../../utils/error'
|
||||
import { createEnvelopeCrypto } from '../../../utils/envelope-crypto'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
import { createLlmRouterService } from './router'
|
||||
|
||||
function freshMasterKey(): Buffer {
|
||||
+8
-8
@@ -1,9 +1,9 @@
|
||||
import type { Buffer } from 'node:buffer'
|
||||
|
||||
import type { GatewayMetrics } from '../../otel'
|
||||
import type { EnvelopeCrypto } from '../../utils/envelope-crypto'
|
||||
import type { ConfigKVService } from '../config-kv'
|
||||
import type { TtsAdapterId, TtsInput } from '../tts-adapters/types'
|
||||
import type { GatewayMetrics } from '../../../otel'
|
||||
import type { EnvelopeCrypto } from '../../../utils/envelope-crypto'
|
||||
import type { ConfigKVService } from '../../adapters/config-kv'
|
||||
import type { TtsAdapterId, TtsInput } from '../../adapters/tts/types'
|
||||
import type { LlmRouteRequest, LlmUpstream, TtsUpstream } from './types'
|
||||
|
||||
import { Buffer as NodeBuffer } from 'node:buffer'
|
||||
@@ -11,15 +11,15 @@ import { Buffer as NodeBuffer } from 'node:buffer'
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { trace } from '@opentelemetry/api'
|
||||
|
||||
import { ApiError } from '../../utils/error'
|
||||
import { errorMessageFromUnknown } from '../../utils/error-message'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
import { errorMessageFromUnknown } from '../../../utils/error-message'
|
||||
import {
|
||||
AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH,
|
||||
AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID,
|
||||
AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_INDEX,
|
||||
AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL,
|
||||
} from '../../utils/observability'
|
||||
import { getAdapter } from '../tts-adapters'
|
||||
} from '../../../utils/observability'
|
||||
import { getAdapter } from '../../adapters/tts'
|
||||
import { createConfigLoader } from './config-loader'
|
||||
import { mapUpstreamError } from './error-mapping'
|
||||
import { createKeyRotator } from './key-rotator'
|
||||
+1
-1
@@ -15,7 +15,7 @@ import type {
|
||||
llmUpstreamSchema,
|
||||
ttsModelSchema,
|
||||
ttsUpstreamSchema,
|
||||
} from '../config-kv'
|
||||
} from '../../adapters/config-kv'
|
||||
|
||||
/**
|
||||
* Composite router config (the value at `LLM_ROUTER_CONFIG` in configKV).
|
||||
+1
-1
@@ -3,7 +3,7 @@ import type { Database } from '../../libs/db'
|
||||
import { beforeAll, describe, expect, it } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { createProviderService } from '../providers'
|
||||
import { createProviderService } from './providers'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
+2
-2
@@ -1,9 +1,9 @@
|
||||
import type { Database } from '../libs/db'
|
||||
import type { Database } from '../../libs/db'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, eq, isNull, sql } from 'drizzle-orm'
|
||||
|
||||
import * as schema from '../schemas/providers'
|
||||
import * as schema from '../../schemas/providers'
|
||||
|
||||
const logger = useLogger('providers')
|
||||
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
import type { Database } from '../libs/db'
|
||||
import type { Database } from '../../libs/db'
|
||||
|
||||
import * as schema from '../schemas/llm-request-log'
|
||||
import * as schema from '../../schemas/llm-request-log'
|
||||
|
||||
export interface RequestLogEntry {
|
||||
userId: string
|
||||
+1
-1
@@ -4,7 +4,7 @@ import { eq } from 'drizzle-orm'
|
||||
import { beforeAll, beforeEach, describe, expect, it } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { createStripeService } from '../stripe'
|
||||
import { createStripeService } from './stripe'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
import type Stripe from 'stripe'
|
||||
|
||||
import type { Database } from '../libs/db'
|
||||
import type { NewStripeCheckoutSession, NewStripeCustomer, NewStripeInvoice, NewStripeSubscription } from '../schemas/stripe'
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { NewStripeCheckoutSession, NewStripeCustomer, NewStripeInvoice, NewStripeSubscription } from '../../schemas/stripe'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { and, eq, isNull, notInArray } from 'drizzle-orm'
|
||||
|
||||
import * as schema from '../schemas/stripe'
|
||||
import * as schema from '../../schemas/stripe'
|
||||
|
||||
const logger = useLogger('stripe-service')
|
||||
|
||||
+3
-3
@@ -1,15 +1,15 @@
|
||||
import type { Database } from '../../libs/db'
|
||||
import type { Database } from '../../../libs/db'
|
||||
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { beforeAll, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { mockDB } from '../../libs/mock-db'
|
||||
import { mockDB } from '../../../libs/mock-db'
|
||||
import { createCharacterService } from '../characters'
|
||||
import { createChatService } from '../chats'
|
||||
import { createFluxService } from '../flux'
|
||||
import { createProviderService } from '../providers'
|
||||
|
||||
import * as schema from '../../schemas'
|
||||
import * as schema from '../../../schemas'
|
||||
|
||||
function fakeRedis() {
|
||||
const map = new Map<string, string>()
|
||||
@@ -1,28 +0,0 @@
|
||||
export { createConfigLoader } from './config-loader'
|
||||
export type { ConfigLoader, ConfigLoaderOptions, ModelConfigSlice } from './config-loader'
|
||||
|
||||
export { createConfigSyncSubscriber } from './config-sync-subscriber'
|
||||
export type { ConfigSyncSubscriber, ConfigSyncSubscriberOptions } from './config-sync-subscriber'
|
||||
|
||||
export { mapUpstreamError } from './error-mapping'
|
||||
export type { RouterErrorCause, UpstreamAttempt, UpstreamErrorContext } from './error-mapping'
|
||||
|
||||
export { createKeyRotator } from './key-rotator'
|
||||
export type { RotatableUpstream, RotatedKey } from './key-rotator'
|
||||
|
||||
export { createLlmRouterService } from './router'
|
||||
export type { CreateLlmRouterServiceOptions, LlmRouterService } from './router'
|
||||
|
||||
export type {
|
||||
FallbackTriggers,
|
||||
KeyEntry,
|
||||
LlmModel,
|
||||
LlmRouteContext,
|
||||
LlmRouteRequest,
|
||||
LlmUpstream,
|
||||
ModelKind,
|
||||
RouterConfig,
|
||||
RouterDefaults,
|
||||
TtsModel,
|
||||
TtsUpstream,
|
||||
} from './types'
|
||||
@@ -1,60 +0,0 @@
|
||||
import type { Buffer } from 'node:buffer'
|
||||
|
||||
import type Redis from 'ioredis'
|
||||
|
||||
import { promisify } from 'node:util'
|
||||
import { gunzip, gzip } from 'node:zlib'
|
||||
|
||||
const gzipAsync = promisify(gzip)
|
||||
const gunzipAsync = promisify(gunzip)
|
||||
|
||||
// First two bytes of a gzip stream are 0x1F 0x8B. Used to distinguish gzipped
|
||||
// payloads from legacy plain-text entries written before compression was
|
||||
// introduced for a given key.
|
||||
function isGzipped(buf: Buffer): boolean {
|
||||
return buf.length >= 2 && buf[0] === 0x1F && buf[1] === 0x8B
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a Redis value, transparently gunzipping it when it was written by
|
||||
* `setCompressed`. Non-gzipped payloads (e.g. pre-compression legacy entries,
|
||||
* or values set via raw `redis.set`) are returned as plain utf-8 strings so
|
||||
* callers can migrate existing keys without a forced invalidation.
|
||||
*
|
||||
* Use when: value size is large enough (~1KB+) that gzip wins on Redis/network
|
||||
* bandwidth, and you accept the ~10ms CPU trade per request. Caller still
|
||||
* handles parse/validate since we only care about bytes in transit.
|
||||
*
|
||||
* Returns null on cache miss.
|
||||
*/
|
||||
export async function getCompressed(redis: Redis, key: string): Promise<string | null> {
|
||||
const cached = await redis.getBuffer(key)
|
||||
if (cached == null)
|
||||
return null
|
||||
|
||||
return isGzipped(cached)
|
||||
? (await gunzipAsync(cached)).toString('utf8')
|
||||
: cached.toString('utf8')
|
||||
}
|
||||
|
||||
/**
|
||||
* Gzip-compress the value and store under key. Always writes gzipped bytes so
|
||||
* subsequent `getCompressed` reads never hit the legacy passthrough branch.
|
||||
*
|
||||
* Expects: caller has already serialized the value (JSON.stringify, etc.) —
|
||||
* this helper is deliberately type-free about what the string represents.
|
||||
*/
|
||||
export async function setCompressed(
|
||||
redis: Redis,
|
||||
key: string,
|
||||
value: string,
|
||||
ttlSeconds?: number,
|
||||
): Promise<void> {
|
||||
const compressed = await gzipAsync(value)
|
||||
if (ttlSeconds != null) {
|
||||
await redis.set(key, compressed, 'EX', ttlSeconds)
|
||||
}
|
||||
else {
|
||||
await redis.set(key, compressed)
|
||||
}
|
||||
}
|
||||
@@ -1,84 +0,0 @@
|
||||
import { Buffer } from 'node:buffer'
|
||||
import { promisify } from 'node:util'
|
||||
import { gzip } from 'node:zlib'
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { getCompressed, setCompressed } from '../redis-compressed'
|
||||
|
||||
const gzipAsync = promisify(gzip)
|
||||
|
||||
function createMockRedis() {
|
||||
const store = new Map<string, Buffer>()
|
||||
return {
|
||||
getBuffer: vi.fn(async (key: string) => store.get(key) ?? null),
|
||||
set: vi.fn(async (key: string, value: string | Buffer, _ex?: string, _ttl?: number) => {
|
||||
store.set(key, Buffer.isBuffer(value) ? value : Buffer.from(value, 'utf8'))
|
||||
return 'OK'
|
||||
}),
|
||||
_store: store,
|
||||
}
|
||||
}
|
||||
|
||||
describe('redis-compressed', () => {
|
||||
let redis: ReturnType<typeof createMockRedis>
|
||||
|
||||
beforeEach(() => {
|
||||
redis = createMockRedis()
|
||||
})
|
||||
|
||||
describe('setCompressed', () => {
|
||||
it('writes gzipped bytes under the key', async () => {
|
||||
await setCompressed(redis as any, 'k', 'hello world'.repeat(100))
|
||||
|
||||
const stored = redis._store.get('k')
|
||||
expect(stored).toBeDefined()
|
||||
expect(stored![0]).toBe(0x1F)
|
||||
expect(stored![1]).toBe(0x8B)
|
||||
})
|
||||
|
||||
it('sets TTL via EX when ttlSeconds is provided', async () => {
|
||||
await setCompressed(redis as any, 'k', 'payload', 600)
|
||||
|
||||
expect(redis.set).toHaveBeenCalledWith('k', expect.any(Buffer), 'EX', 600)
|
||||
})
|
||||
|
||||
it('omits EX when ttlSeconds is undefined', async () => {
|
||||
await setCompressed(redis as any, 'k', 'payload')
|
||||
|
||||
const [, , ex] = redis.set.mock.calls[0]
|
||||
expect(ex).toBeUndefined()
|
||||
})
|
||||
|
||||
it('round-trips: value written by setCompressed reads back identical via getCompressed', async () => {
|
||||
const original = JSON.stringify({ voices: Array.from({ length: 50 }, (_, i) => ({ id: `v-${i}` })) })
|
||||
await setCompressed(redis as any, 'k', original)
|
||||
|
||||
const read = await getCompressed(redis as any, 'k')
|
||||
expect(read).toBe(original)
|
||||
})
|
||||
})
|
||||
|
||||
describe('getCompressed', () => {
|
||||
it('returns null on cache miss', async () => {
|
||||
const result = await getCompressed(redis as any, 'missing')
|
||||
expect(result).toBeNull()
|
||||
})
|
||||
|
||||
it('gunzips entries written with gzip magic bytes', async () => {
|
||||
const text = 'payload'.repeat(50)
|
||||
const compressed = await gzipAsync(text)
|
||||
redis._store.set('k', compressed)
|
||||
|
||||
const result = await getCompressed(redis as any, 'k')
|
||||
expect(result).toBe(text)
|
||||
})
|
||||
|
||||
it('returns legacy plain-text entries as utf-8 without attempting gunzip', async () => {
|
||||
redis._store.set('k', Buffer.from('{"legacy":"value"}', 'utf8'))
|
||||
|
||||
const result = await getCompressed(redis as any, 'k')
|
||||
expect(result).toBe('{"legacy":"value"}')
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -6,10 +6,10 @@ import { vi } from 'vitest'
|
||||
|
||||
import { buildApp } from '../../src/app'
|
||||
import { mockDB } from '../../src/libs/mock-db'
|
||||
import { createAdminFluxGrantsService } from '../../src/services/admin-flux-grants'
|
||||
import { createBillingService } from '../../src/services/billing/billing-service'
|
||||
import { createFluxService } from '../../src/services/flux'
|
||||
import { createUserDeletionService } from '../../src/services/user-deletion'
|
||||
import { createAdminFluxGrantsService } from '../../src/services/domain/admin/flux-grants'
|
||||
import { createBillingService } from '../../src/services/domain/billing/billing-service'
|
||||
import { createFluxService } from '../../src/services/domain/flux'
|
||||
import { createUserDeletionService } from '../../src/services/domain/user-deletion'
|
||||
import { userFluxRedisKey } from '../../src/utils/redis-keys'
|
||||
|
||||
import * as schema from '../../src/schemas'
|
||||
|
||||
Reference in New Issue
Block a user