refactor(server/flux): replace UpDownCounter with ObservableGauge for WebSocket connections

- Updated the EngagementMetrics interface to use ObservableGauge for tracking active WebSocket connections. - Added detailed comments explaining the rationale for this change, highlighting the benefits of using a pull-based gauge over a delta-based counter. - Implemented the ObservableGauge in the createChatWsHandlers function, ensuring it accurately reflects the live count of active connections. - Removed the previous UpDownCounter logic to prevent issues with connection drift during process crashes or network interruptions.
2026-05-09 00:23:02 +08:00
parent 917450fdfb
commit fe59f91c84
7 changed files with 4396 additions and 817 deletions
@@ -28,7 +28,9 @@
 - `flux-meter.md`
  - Sub-Flux 计量服务（TTS/STT 等）的债务账本机制与复用指南
 - `observability-conventions.md`
-  - traces / metrics 命名规则，标准 OTel 字段与 `airi.*` 自定义字段边界
+  - traces / metrics 命名规则，标准 OTel 字段与 `airi.*` 自定义字段边界，SemconvStability 迁移、Counter priming、Dashboard 变量陷阱
+- `observability-metrics.md`
+  - 全量 metric 目录（按域分组：HTTP / Auth / Engagement / Revenue / GenAI / Email / Rate limit / Runtime），含名字、类型、Labels、落点
 - `auth-and-oidc.md`
  - 认证与 OIDC Provider 架构、登录流程、trusted clients、踩坑记录
 - `email-auth-resend.md`
@@ -64,6 +66,7 @@
 - 改扣费、充值、Stripe：先看 `billing-architecture.md`
 - 改 Flux 充值价格 / 多币种 / Stripe Product/Price：先看 `stripe-pricing.md`
 - 改 trace / metric attributes、OTel 命名：先看 `observability-conventions.md`
+- 加新 metric / 找当前 metric 全量列表：先看 `observability-metrics.md`
 - 改认证、OIDC、登录流程：先看 `auth-and-oidc.md`
 - 改邮件 service / Better Auth 邮件 callback：先看 `email-auth-resend.md`
 - 改账号注销 / 业务 service 的 `deleteAllForUser`：先看 `account-deletion.md`
@@ -56,7 +56,7 @@ Redis 相关优先复用 instrumentation 自动产生的标准属性，不要重
 - 仅 AIRI 内部存在的流式控制字段
 - 临时调试但仍需要进入可观测系统的业务字段

-当前示例：
+当前 attribute 示例：

 - `AIRI_ATTR_BILLING_FLUX_CONSUMED`
 - `AIRI_ATTR_GEN_AI_STREAM`
@@ -66,6 +66,14 @@ Redis 相关优先复用 instrumentation 自动产生的标准属性，不要重
 - `AIRI_ATTR_GEN_AI_INPUT_TEXT`
 - `AIRI_ATTR_GEN_AI_OUTPUT_TEXT`

+当前 `airi.*` metric 命名空间（Prom 系列名见 [`observability-metrics.md`](./observability-metrics.md)）：
+
+- 计费：`airi.billing.flux.consumed` / `.credited` / `.unbilled` / `.tts.chars` / `.tts.preflight_rejections`
+- 收入：`airi.stripe.revenue`
+- 邮件：`airi.email.send` / `.failures` / `.duration`
+- 限流：`airi.rate_limit.blocked`
+- GenAI：`airi.gen_ai.stream.interrupted`
+
 ## Metric Name 策略

 当前 `apps/server` 仍保留以下 metric name：
@@ -157,3 +165,65 @@ span name 目前允许保留业务可读格式，例如：
 - [apps/server/src/libs/otel.ts](/apps/server/src/libs/otel.ts)
 - [services/telegram-bot/src/llm/actions.ts](/services/telegram-bot/src/llm/actions.ts)
 - [services/telegram-bot/src/bots/telegram/agent/actions/read-message.ts](/services/telegram-bot/src/bots/telegram/agent/actions/read-message.ts)
+
+## SemconvStability 迁移说明
+
+`@opentelemetry/instrumentation-http` 0.215+ 默认 OLD semconv（`http.server.duration` in ms），不是 STABLE 名。AIRI 在 [apps/server/instrumentation.mjs](/apps/server/instrumentation.mjs) 顶部强制 `OTEL_SEMCONV_STABILITY_OPT_IN=http`（仅 STABLE）。
+
+| Semconv 模式 | 发哪些 series | 我们用 |
+|---|---|---|
+| OLD（默认）| `http.server.duration` (ms)、`http.client.duration` (ms)、attr 用 `http.method` / `http.status_code` | ❌ |
+| STABLE（`=http`）| `http.server.request.duration` (s)、`http.client.request.duration` (s)、attr 用 `http.request.method` / `http.response.status_code` | ✅ |
+| 双发（`=http/dup`）| 上面两套都发 | 仅在有外部 OLD-name 消费者待迁移时启用 |
+
+**为什么直接 STABLE-only**：
+
+- grep 整仓库零 OLD-name 引用
+- Dashboard 与服务代码 checked in 在一起，无外部 dashboard
+- 迁移没有自然终点，OLD 系列不显式清理就一直占 storage
+- 双发会让每条 HTTP 请求 cardinality 翻倍
+
+**何时切回 `dup`**：将来如果有别的 service 主动 scrape 本 server 的 OLD-name 系列，临时切几周完成迁移即可。
+
+## Counter priming 注意事项
+
+OTel SDK 的 Counter / UpDownCounter 在第一次 `.add()` 之前**完全不出现在 Prometheus 抓取里**。Histogram 同理（要等第一次 `.record()`）。
+
+后果：低流量 metric 在 dashboard 上看起来像「埋点丢了」，告警里 `absent()` 也无法工作。
+
+[apps/server/src/libs/otel.ts](/apps/server/src/libs/otel.ts) 的 `primeCounter` 在 SDK 启动后给每个 Counter 调一次 `.add(0)`，把 series 注册出来；`0` 不影响 rate / sum 计算。
+
+加新 Counter 时**记得加进 prime 列表**，否则未触发的指标在 Grafana 里就是空的。
+
+验证脚本：[apps/server/src/scripts/otel-smoke.mjs](/apps/server/src/scripts/otel-smoke.mjs)
+
+```sh
+pnpm -F @proj-airi/server exec node --import tsx ./src/scripts/otel-smoke.mjs
+```
+
+打印 SDK 启动后立即可见的所有 instrument 名字。
+
+## Dashboard 变量陷阱
+
+**变量定义里不要引用业务 metric**。早期 [airi-server-overview-cloud.json](/apps/server/otel/grafana/dashboards/airi-server-overview-cloud.json) 的 `$env` / `$service` 都从 `http_server_request_duration_seconds_count` 取 label values —— 升级 instrumentation-http 后这个系列没了，导致：
+
+1. 两个变量解析为空字符串
+2. 所有 panel 的 `{service_name=~"$service", deployment_environment=~"$env"}` 匹配零 series
+3. 整个 dashboard 全 No Data，**包括那些 metric 还活着的 panel**
+
+修法：变量改用 `target_info`。这是 OTel SDK 启动就发的 resource-only series，永远存在，且天然自带 `service_name` / `deployment_environment` / `service_version` 这套 resource attributes。
+
+```promql
+# Good
+label_values(target_info, deployment_environment)
+label_values(target_info{deployment_environment=~"$env"}, service_name)
+
+# Bad — 任何业务 metric 改名/迁移就全盘崩
+label_values(http_server_request_duration_seconds_count, deployment_environment)
+```
+
+后续新增 dashboard 默认沿用 `target_info` 这条惯例。
+
+## 完整 metric 目录
+
+按域分组的全量 metric 清单（名字、类型、单位、labels、落点）见 [`observability-metrics.md`](./observability-metrics.md)。每加一个新 metric 时同步更新该文档。
@@ -0,0 +1,157 @@
+# Metrics Catalog
+
+服务端当前所有 metric 的完整目录。按业务领域分组。
+
+> 命名规则、`airi.*` 边界、attribute 选择请看 [`observability-conventions.md`](./observability-conventions.md)。本文档只做"哪些 metric 存在、怎么查"。
+
+## 名字到 Prometheus 系列的换算
+
+OTel SDK 在导出到 Prometheus 时做两件事：
+
+1. `.` → `_`：`airi.billing.flux.consumed` → `airi_billing_flux_consumed`
+2. Counter 加 `_total` 后缀：`auth.attempts` → `auth_attempts_total`
+3. Histogram 拆三件套：`http.server.request.duration` →
+   - `http_server_request_duration_seconds_bucket`（含 `le` label）
+   - `http_server_request_duration_seconds_count`
+   - `http_server_request_duration_seconds_sum`
+4. UpDownCounter 不加 `_total`：`ws.connections.active` → `ws_connections_active`
+5. 带单位的 instrument 在 SDK 导出时把单位插进名字：`airi.stripe.revenue`（unit `minor_unit`）→ `airi_stripe_revenue_minor_unit_total`
+
+> 查询面板若拼名字时不确定后缀，先用 `{__name__=~"airi_billing_flux.*"}` 之类正则探一下。
+
+## HTTP（来自 instrumentation-http）
+
+| Metric | 类型 | Unit | 来源 | 关键 attributes |
+|---|---|---|---|---|
+| `http.server.request.duration` | Histogram | s | `instrumentation-http`（STABLE semconv） | `http.request.method`、`http.route`、`http.response.status_code` |
+| `http.server.active_requests` | UpDownCounter | — | [middlewares/otel.ts](../../src/middlewares/otel.ts) `otelMiddleware` | `http.request.method`、`http.route` |
+
+> **STABLE-only**：[instrumentation.mjs:25](../../instrumentation.mjs) 把 `OTEL_SEMCONV_STABILITY_OPT_IN=http` 提前注入。OLD 系列（`http.server.duration` in ms）不再发射。详见 [`observability-conventions.md` 的 SemconvStability 章节](./observability-conventions.md#semconvstability-迁移说明)。
+
+## Auth & Users
+
+全部由 [libs/auth.ts](../../src/libs/auth.ts) Better Auth hooks 触发。
+
+| Metric | 类型 | 落点（hook） | Labels |
+|---|---|---|---|
+| `auth.attempts` | Counter | `before` hook，path 含 `/sign-in` 或 `/sign-up` | `auth.method`（path 末段） |
+| `auth.failures` | Counter | `after` hook，`ctx.context.returned` 含 `error` | `auth.method` |
+| `user.registered` | Counter | `databaseHooks.user.create.after` | — |
+| `user.login` | Counter | `databaseHooks.session.create.after` | — |
+| `user.active_sessions` | UpDownCounter | session create / delete | — |
+
+## Engagement
+
+| Metric | 类型 | 落点 | Labels |
+|---|---|---|---|
+| `chat.messages` | Counter | [services/chats.ts](../../src/services/chats.ts) `pushMessages` | — |
+| `character.created` | Counter | [services/characters.ts](../../src/services/characters.ts) | — |
+| `character.deleted` | Counter | 同上 | — |
+| `character.engagement` | Counter | 同上（like/bookmark） | `action`（`like` / `unlike` / `bookmark` / `unbookmark`） |
+| `ws.connections.active` | UpDownCounter | [routes/chat-ws/index.ts](../../src/routes/chat-ws/index.ts) | — |
+| `ws.messages.sent` | Counter | 同上 | — |
+| `ws.messages.received` | Counter | [services/chats.ts](../../src/services/chats.ts) | — |
+
+## Revenue & Billing
+
+### Stripe lifecycle
+
+| Metric | 类型 | 落点 | Labels |
+|---|---|---|---|
+| `stripe.checkout.created` | Counter | [routes/stripe/index.ts](../../src/routes/stripe/index.ts) `/checkout` POST | — |
+| `stripe.checkout.completed` | Counter | webhook `checkout.session.completed` | — |
+| `stripe.payment.failed` | Counter | webhook `invoice.payment_failed` | — |
+| `stripe.subscription.event` | Counter | webhook `customer.subscription.*` | `event_type`（`created`/`updated`/`deleted`） |
+| `stripe.events` | Counter | 任何 webhook | `event_type`（完整 event.type，e.g. `invoice.paid`） |
+| `airi.stripe.revenue` | Counter（`minor_unit`） | webhook `checkout.session.completed` + `invoice.paid` | `currency`、`source`（`checkout`/`invoice`） |
+
+> **金额单位**：`airi.stripe.revenue` 用最小币种单位（cents 等），跨币种 sum 没有意义，**永远 `sum by (currency)`**。要换主单位（dollars 等）做 `/ 100` 即可，前提是该币种没有不同 minor unit 比例。
+
+### Flux ledger
+
+| 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.unbilled` | Counter | `routes/openai/v1/index.ts` 流式 debit 失败 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` |
+| `airi.billing.tts.preflight_rejections` | Counter | `flux-meter.ts` `assertCanAfford` | `meter`、`reason`（`insufficient_balance`） |
+
+> **`airi.billing.flux.unbilled` 是 P0 告警金线**：任何持续 > 0 都意味着真实收入泄漏，应当 page。语义上等于"流式响应已经发给用户但 DB debit 失败的 Flux 量"。
+
+## GenAI
+
+| Metric | 类型 | Unit | 落点 | Labels |
+|---|---|---|---|---|
+| `gen_ai.client.operation.duration` | Histogram | s | `routes/openai/v1/index.ts` `recordMetrics` | `gen_ai.request.model`、`gen_ai.operation.name`/`airi.gen_ai.operation.kind`、`http.response.status_code` |
+| `gen_ai.client.operation.count` | Counter | — | 同上 | 同上 |
+| `gen_ai.client.token.usage.input` | Counter | — | 同上 | 同上 |
+| `gen_ai.client.token.usage.output` | Counter | — | 同上 | 同上 |
+| `gen_ai.client.first_token.duration` | Histogram | s | 流式 reader 第一个非空 chunk 抵达时 | `gen_ai.request.model`、`gen_ai.operation.name` |
+| `airi.gen_ai.stream.interrupted` | Counter | — | 流式 reader catch | `gen_ai.request.model`、`stage`（`before_first_chunk`/`mid_stream`） |
+
+## Email（Resend）
+
+来源 [services/email.ts](../../src/services/email.ts) 的 `send()` 内部 try/catch。
+
+| Metric | 类型 | Labels |
+|---|---|---|
+| `airi.email.send` | Counter | `template`（`verification`/`password_reset`/`magic_link`/`change_email`/`delete_account`/`unknown`） |
+| `airi.email.failures` | Counter | `template`、`error_name`（Resend `error.name` 或 `unhandled`） |
+| `airi.email.duration` | Histogram（s） | `template`、`outcome`（`ok`/`error`） |
+
+## Rate limiting
+
+来源 [middlewares/rate-limit.ts](../../src/middlewares/rate-limit.ts) 的 `handler`。
+
+| Metric | 类型 | Labels |
+|---|---|---|
+| `airi.rate_limit.blocked` | Counter | `route`（callsite 提供，e.g. `auth.api` / `openai.completions` / `stripe.checkout`）、`key_type`（`user`/`ip`）、`limit`（窗口内最大次数） |
+
+> **注意**：`route` 是 callsite 显式提供的稳定 label，不是 raw URL path —— URL path 是高 cardinality，会爆炸。新加 rate limiter 时记得传 `routeLabel`。
+
+## Node.js Runtime
+
+来自 `@opentelemetry/instrumentation-runtime-node`，下面这些是 dashboard 上用到的子集（不全列）：
+
+- `v8js.memory.heap.{used,limit,space.physical_size,space.available_size}` Gauge / bytes
+- `nodejs.eventloop.delay.{p50,p99,mean,...}` Gauge / s
+- `nodejs.eventloop.utilization` Gauge / ratio
+- `v8js.gc.duration` Histogram / s
+
+## 已落地的 dashboard 行映射
+
+[airi-server-overview-cloud.json](../../otel/grafana/dashboards/airi-server-overview-cloud.json)，从上到下：
+
+| Row | 关键 metric |
+|---|---|
+| HTTP Overview | `http.server.request.duration`（rate / P95 / by route / 5xx 率） |
+| Auth & Users | `auth.attempts` / `auth.failures` / `user.{login,registered,active_sessions}` + 失败率 |
+| Engagement | `ws.connections.active` / `ws.messages.{sent,received}` / `chat.messages` / `character.{created,deleted,engagement}` |
+| Business Metrics | `airi.billing.flux.consumed` / `flux.insufficient_balance` / `gen_ai.client.token.usage.*` / `stripe.checkout.completed` / `airi.billing.flux.credited` |
+| Stripe Detail | `stripe.{events,subscription.event,payment.failed}` / checkout funnel / `airi.stripe.revenue` |
+| Node.js Runtime | runtime instrumentation 那一批 |
+| LLM Gateway | `gen_ai.client.{operation.count,operation.duration,token.usage.*,first_token.duration}` / `airi.billing.flux.consumed` / `airi.billing.flux.unbilled` / `airi.gen_ai.stream.interrupted` / `airi.billing.tts.chars` |
+| Reliability | `airi.email.{send,failures}` 失败率 / `airi.rate_limit.blocked` / `airi.billing.tts.preflight_rejections` |
+| Application Logs | Loki，不是 Prometheus |
+
+## 验证 metric 是否已注册
+
+[`src/scripts/otel-smoke.mjs`](../../src/scripts/otel-smoke.mjs) 跑一遍：
+
+```sh
+pnpm -F @proj-airi/server exec node --import tsx ./src/scripts/otel-smoke.mjs
+```
+
+会打印 SDK 启动时立即 export 的所有 instrument 名字。**Counter 通过 `.add(0)` priming**（[libs/otel.ts](../../src/libs/otel.ts) `primeCounter`）后会出现在这里 —— Histogram 不会，要等真实 `.record()` 才出现。
+
+## 加新 metric 时的 checklist
+
+1. 决定命名空间：能映射到 OTel semconv 就用标准名，否则放 `airi.*`（不要造新顶级前缀）
+2. 在 [utils/observability.ts](../../src/utils/observability.ts) 加常量
+3. 在 [libs/otel.ts](../../src/libs/otel.ts) 的对应 metric group 接口（`HttpMetrics`/`AuthMetrics`/...）加字段，并在 `initOtel` 里 `meter.create*` 创建
+4. **如果是 Counter，在 `primeCounter` 调用列表里加一行** —— 否则低流量时 panel 看起来"没数据"
+5. 在 callsite 通过 DI 拿到 metrics 对象后调 `.add()` / `.record()`
+6. 跑 `pnpm -F @proj-airi/server exec node --import tsx ./src/scripts/otel-smoke.mjs` 确认注册
+7. 更新本文档对应章节
@@ -29,9 +29,24 @@ import { PgInstrumentation } from '@opentelemetry/instrumentation-pg'
 // Source: node_modules/.pnpm/@opentelemetry+instrumentation-http@0.215.0/.../build/src/http.js L25-72
 // MUST run before `new HttpInstrumentation(...)` below — its constructor reads
 // the env var once and caches the result.
+//
+// Use a truthy check (not `??=`): `process.env.X` is `''` when the platform
+// (e.g. Railway) registers the var without a value, and `??=` does NOT override
+// empty strings — that would silently fall back to OLD semconv with no signal
+// in logs. Truthy check covers both `undefined` and `''`.
 // Removal condition: ops sets OTEL_SEMCONV_STABILITY_OPT_IN explicitly in the
-// deployment platform (Railway env), then this preload default can be deleted.
-env.OTEL_SEMCONV_STABILITY_OPT_IN ??= 'http'
+// deployment platform with a non-empty value, then this preload default can be
+// deleted.
+if (!env.OTEL_SEMCONV_STABILITY_OPT_IN) {
+  env.OTEL_SEMCONV_STABILITY_OPT_IN = 'http'
+}
+
+// Surface the resolved value in stdout BEFORE any instrumentation constructor
+// runs. Lets ops grep Railway logs for `[otel-preload]` to confirm the preload
+// actually executed and what semconv mode is active. Without this, a misloaded
+// preload (wrong `--import` path, missing flag, build cache) is invisible
+// until you query Prometheus and notice STABLE-name series are missing.
+console.info(`[otel-preload] OTEL_SEMCONV_STABILITY_OPT_IN=${env.OTEL_SEMCONV_STABILITY_OPT_IN}`)

 registerInstrumentations({
  instrumentations: [
@@ -1,4 +1,4 @@
-import type { Counter, Histogram, UpDownCounter } from '@opentelemetry/api'
+import type { Counter, Histogram, ObservableGauge, UpDownCounter } from '@opentelemetry/api'

 import type { Env } from './env'

@@ -77,7 +77,24 @@ export interface EngagementMetrics {
  characterCreated: Counter
  characterDeleted: Counter
  characterEngagement: Counter
-  wsConnectionsActive: UpDownCounter
+  /**
+   * Pull-based gauge for active WebSocket connections.
+   *
+   * Use when:
+   * - Querying current concurrent WS connections in Grafana / alerts.
+   *
+   * Why ObservableGauge instead of UpDownCounter:
+   * - UpDownCounter is delta-based (+1 / -1) and drifts when disconnect
+   *   handlers miss (process crash, SIGKILL, TCP RST, network blackhole).
+   * - ObservableGauge runs a callback at every export interval and reports
+   *   the live registry size, so a missed -1 self-corrects on the next
+   *   scrape instead of leaking forever.
+   *
+   * Expects:
+   * - Caller (`createChatWsHandlers`) registers exactly one callback via
+   *   `addCallback`. Multiple callbacks would double-count.
+   */
+  wsConnectionsActive: ObservableGauge
  wsMessagesSent: Counter
  wsMessagesReceived: Counter
 }
@@ -258,8 +275,15 @@ export function initOtel(env: Env): OtelInstance | undefined {
    characterEngagement: meter.createCounter(METRIC_CHARACTER_ENGAGEMENT, {
      description: 'Number of character engagement actions (like/bookmark)',
    }),
-    wsConnectionsActive: meter.createUpDownCounter(METRIC_WS_CONNECTIONS_ACTIVE, {
-      description: 'Active WebSocket connections',
+    // NOTICE:
+    // ObservableGauge — caller (chat-ws factory) registers a callback that
+    // reads the live connection registry on each export interval. UpDownCounter
+    // was previously used but drifted: missed `-1` on process crash / SIGKILL /
+    // TCP RST left the counter stuck high until Prom staleness expired the
+    // dead instance's series (~5 min). The pull-based gauge self-corrects on
+    // the next scrape because there is no delta state to leak.
+    wsConnectionsActive: meter.createObservableGauge(METRIC_WS_CONNECTIONS_ACTIVE, {
+      description: 'Active WebSocket connections (live registry size, scraped per export interval)',
    }),
    wsMessagesSent: meter.createCounter(METRIC_WS_MESSAGES_SENT, {
      description: 'Messages sent via WebSocket',
@@ -60,6 +60,17 @@ export function createChatWsHandlers(
  // Dedicated subscriber connection (ioredis requires a separate connection for subscribe mode)
  const sub = redis.duplicate()

+  // Pull-based active-connection gauge: walk the local registry on each
+  // export interval and report the actual live count. Registered exactly
+  // once per process here (factory runs once via injeca); duplicate
+  // registration would double-count.
+  metrics?.wsConnectionsActive.addCallback((result) => {
+    let total = 0
+    for (const conns of userConnections.values())
+      total += conns.size
+    result.observe(total)
+  })
+
  sub.on('message', (_channel: string, message: string) => {
    try {
      const data = parseChatBroadcastMessage(message)
@@ -113,13 +124,11 @@ export function createChatWsHandlers(
        addConnection(userId, ctx)
        ensureSubscribed(userId)
        log.withFields({ userId }).log('WS connected')
-        metrics?.wsConnectionsActive.add(1)

        ctx.on(wsDisconnectedEvent, () => {
          removeConnection(userId, ctx)
          maybeUnsubscribe(userId)
          log.withFields({ userId }).log('WS disconnected')
-          metrics?.wsConnectionsActive.add(-1)
        })

        // RPC: send messages