diff --git a/apps/server/docs/ai-context/README.md b/apps/server/docs/ai-context/README.md index 3f61ae61e..ceed42ac8 100644 --- a/apps/server/docs/ai-context/README.md +++ b/apps/server/docs/ai-context/README.md @@ -31,6 +31,8 @@ - traces / metrics 命名规则，标准 OTel 字段与 `airi.*` 自定义字段边界，SemconvStability 迁移、Counter priming、Dashboard 变量陷阱 - `observability-metrics.md` - 全量 metric 目录（按域分组：HTTP / Auth / Engagement / Revenue / GenAI / Email / Rate limit / Runtime），含名字、类型、Labels、落点 +- `metrics-ownership.md` + - 指标分层规则：什么走 Grafana / 什么走 PostHog / 什么是 Postgres truth；含 7 题判定 Checklist、PostHog 事件命名约定、当前指标归属总表、PostHog 接入路线图 - `auth-and-oidc.md` - 认证与 OIDC Provider 架构、登录流程、trusted clients、踩坑记录 - `email-auth-resend.md` @@ -45,6 +47,10 @@ - 账号注销端到端验证：what's verified（schema/typecheck/units）和 what's pending（live DB + Resend + Stripe trace） - `verifications/admin-flux-grants.md` - Admin 同步发 FLUX 路径：同步 grant / dry-run / adminGuard 拒绝（架构刚从 batch 切换到同步，待重新实测） +- `verifications/flux-unbilled-exploit-fix.md` + - Unpaid-usage exploit 修补（commit `7267b0d6b`）的代码层验证 + 残余 gap（TTS flux-meter 未适配 partial-debit）+ follow-up 清单 +- `verifications/flux-unbilled-reconciliation.md` + - 70.2K 历史漏账的取证 SQL + Loki query 模板、处理决策框架、修补后的监控建议 ## 快速结论 @@ -67,6 +73,7 @@ - 改 Flux 充值价格 / 多币种 / Stripe Product/Price：先看 `stripe-pricing.md` - 改 trace / metric attributes、OTel 命名：先看 `observability-conventions.md` - 加新 metric / 找当前 metric 全量列表：先看 `observability-metrics.md` +- 决定新指标该走 Grafana 还是 PostHog：先看 `metrics-ownership.md` - 改认证、OIDC、登录流程：先看 `auth-and-oidc.md` - 改邮件 service / Better Auth 邮件 callback：先看 `email-auth-resend.md` - 改账号注销 / 业务 service 的 `deleteAllForUser`：先看 `account-deletion.md` diff --git a/apps/server/docs/ai-context/metrics-ownership.md b/apps/server/docs/ai-context/metrics-ownership.md new file mode 100644 index 000000000..9ef5f0424 --- /dev/null +++ b/apps/server/docs/ai-context/metrics-ownership.md @@ -0,0 +1,269 @@ +# Metrics Ownership + +这份文档定义 AIRI 团队的指标分层规则：什么指标该走 Grafana / Prometheus（OTel server-side），什么该走 PostHog（前后端混合 product analytics），同名指标怎么处理。落地这份是为了避免后期"同一个 KPI 三处不同数"的漂移。 + +## 总原则 + +工具职责正交，**互补不互替**： + +| 层 | 工具 | 关键属性 | +|---|---|---| +| **System / API observability** | Grafana Cloud + Prometheus + OTel | 系统健康、延迟、错误率、SRE on-call 告警 | +| **Product analytics** | PostHog Cloud | 用户行为、漏斗、retention、cohort、A/B、feature adoption | +| **Financial truth source** | Postgres (`flux_transaction` / Stripe webhook 持久化) | 收入与扣费 ledger，任何展示都视作近似 | +| **LLM-native observability**（预留） | Langfuse / Helicone（未接入） | token cost、prompt eval、provider trace — 后续按需引入 | + +**业界没有权威的判定 framework**（参见下方"参考来源"），这份文档落实成项目内的可执行规则。 + +## 7 题判定 Checklist + +每条新增指标依次问这 7 个问题： + +| # | 问题 | 偏 Grafana | 偏 PostHog | +|---|------|-----------|------------| +| 1 | 超阈值需要**分钟级 on-call 告警**？ | ✓ | | +| 2 | 主要读者是 **SRE / 后端工程师**，不是 PM？ | ✓ | | +| 3 | 需要跟 **trace / log join**（分布式 debug）？ | ✓ | | +| 4 | 含义依赖**用户身份 / session**（"哪个用户做了什么"）？ | | ✓ | +| 5 | 消费场景是**漏斗 / retention cohort / A/B test**？ | | ✓ | +| 6 | 会被 **CEO / PM 在周会 OKR review** 看？ | | ✓ | +| 7 | 采集点在**前端页面**（pricing page、onboarding）？ | （拿不到） | ✓ | + +**裁决规则**： + +- ≥4 个偏一侧 → 那一侧 +- 平局 → 两边都放，但**指定唯一 truth side**（见下文） +- 如果一个指标 7 题答下来很纠结，多半是**指标定义本身没拆干净**——应该拆成两个不同的指标，分别归到两边，而不是混合归属 + +## Truth Side（重复指标处理） + +业界没有银弹（PostHog 官方在 [issue #43633](https://github.com/posthog/posthog/issues/43633) 也承认 dual-emit 没有统一 pattern）。我们的做法：**接受两边数字差异，dashboard 上标注语义不同**。 + +### Truth side 指定原则 + +| 指标类型 | Truth side | 理由 | +|---|---|---| +| 计费 ledger（每一分钱可审计） | **Postgres** | Grafana / PostHog 都视作近似展示，争议查 SQL | +| HTTP / WS / DB / Stripe webhook **计数** | **Grafana**（OTel counter） | 系统事件，PostHog 看不到 | +| 用户去重 DAU / WAU / retention | **PostHog** | 需要 distinctId 去重，session table 计数不准 | +| 收入展示（MRR / ARR / churn revenue） | **Postgres → 两边展示** | 真相在 Postgres，Grafana 取系统侧切片（panel-30），PostHog 取用户维度切片 | +| LLM token / cost | **Grafana**（短期） | 后续若引入 Langfuse 则迁过去 | +| 用户行为漏斗各步骤 | **PostHog**（必须） | 第一步通常是前端事件，Grafana 拿不到 | + +### Dashboard 标注规则 + +两边都展示的指标，**必须**在 Grafana panel description 和 PostHog insight description 里： + +1. 注明 truth side（"Truth: Postgres `flux_transaction` 表" / "Truth: PostHog 事件去重"） +2. 注明本侧统计的语义差异（如 "Grafana 这里是 session 计数，不去重；PostHog 那边是 user 去重 DAU"） +3. 如果两边数字差异预期 > 10%，写明合理范围 + +## PostHog 事件命名约定 + +格式：`_`，全部 `snake_case`。 + +| 约定 | 示例 | +|---|---| +| 名词在前，动词过去式在后 | `pricing_page_viewed`、`plan_selected`、`payment_completed` | +| 一律 past tense | `signup_completed` 不是 `complete_signup` | +| 不带产品 / 模块前缀 | `chat_session_started` 不是 `airi_chat_session_started` | +| 不带技术细节前缀 | `model_switched` 不是 `frontend_model_switched` | +| properties 用 `snake_case` | `{ plan_id, price_usd, checkout_session_id }` | +| 跟外部系统串联的 ID 用原平台命名 | `stripe_customer_id`、`stripe_subscription_id`、`checkout_session_id` | + +`distinctId` 在登录后必须调 `posthog.identify(userId)`，userId 用 Better Auth 的 user id（跟 server 里的 `c.get('user').id` 一致）。后端 `posthog-node` 上报支付事件时用 fallback 链 `userId` (`session.metadata.userId`) > `email` (`session.customer_email`) > `session.id`——第一项跟前端 `identify` 一致，PostHog person merge 在这里完成。前端 wiring 由 `useSharedAnalyticsStore.initialize()` 自动处理，不需要每个 caller 手动 identify。 + +参考来源：[PostHog: 5 events all teams should track](https://posthog.com/blog/events-you-should-track-with-posthog)。 + +## Grafana 指标命名约定 + +沿用现有 [`observability-conventions.md`](./observability-conventions.md) 不再重复，关键约束： + +- OTel semconv 优先（`http_*` / `db_*` / `gen_ai_*`），匹配不上才放 `airi.*` 命名空间 +- counter 一律 `_total` 后缀，histogram 一律 `_seconds_bucket` / `_bytes_bucket` +- label 基数受控（route pattern 而非 URL，model name 而非 prompt） + +## 当前指标归属总表 + +### Grafana / Prometheus（系统侧） + +来源：`apps/server/src/otel/index.ts` 全量列表见 [`observability-metrics.md`](./observability-metrics.md)。Dashboard 配置在 [`apps/server/otel/grafana/dashboards/build.ts`](../../otel/grafana/dashboards/build.ts)。 + +| 域 | 代表性指标 | Truth | 备注 | +|---|---|---|---| +| HTTP | `http_server_request_duration_seconds_*` | Grafana | OTel 标准 | +| WS | `ws_connections_active` / `ws_messages_*_total` | Grafana | | +| LLM | `gen_ai_client_operation_count_total` / `gen_ai_client_first_token_duration_seconds` | Grafana | | +| Billing | `airi_billing_flux_unbilled_total` | Grafana | **告警必须**：`increase(airi_billing_flux_unbilled_total[5m]) > 0` | +| Auth | `user_active_sessions` | Postgres → Grafana 派生 | 集群级 gauge，用 `avg()` 不要 `sum()` | +| Stripe | `airi_stripe_revenue_minor_unit_total` / `stripe_events_total` | Postgres → 两边展示 | Grafana 是系统侧 webhook 计数 | +| Runtime | `v8js_memory_*` / `nodejs_eventloop_delay_*` | Grafana | per `service_instance_id` | +| Rate-limit | `airi_rate_limit_blocked_total` | Grafana | in-memory per replica | + +### PostHog（前后端混合，产品侧） + +已接入： +- 前端 `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` +- 前端↔后端 identity merge：`useSharedAnalyticsStore.initialize()` watch `authStore.isAuthenticated` 自动调 `posthog.identify(user.id)` / `reset()` + +已埋点： + +| 域 | 事件 | 来源 | 落点 | Truth | +|---|---|---|---|---| +| 付费漏斗 | `pricing_page_viewed` / `plan_selected` / `checkout_started` | 前端 | `packages/stage-pages/src/pages/settings/flux.vue` | PostHog | +| 付费漏斗终点 | `payment_completed` | 后端 webhook | `apps/server/src/routes/stripe/index.ts` | PostHog | +| Activation / Retention | `first_model_selected` / `model_switched` | 前端（consciousness store watcher） | `packages/stage-ui/src/stores/analytics/index.ts` | PostHog | +| Retention | `character_created` | 前端 | `apps/stage-web/src/pages/settings/characters/components/CharacterDialog.vue` | PostHog | +| Retention | `chat_session_started` | 前端 | `packages/stage-ui/src/components/scenarios/chat/components/sessions-drawer.vue` | PostHog | +| Churn | `subscription_cancelled`（带 cancellation_reason） | 后端 webhook | `apps/server/src/routes/stripe/index.ts` | PostHog | +| 老事件 | `provider_card_clicked` / `first_message_sent` | 前端 | `packages/stage-ui/src/composables/use-analytics.ts` | PostHog | + +待埋点（API 已在 `use-analytics.ts` 暴露但调用点未接入）： + +| 域 | 事件 | 状态 | +|---|---|---| +| Activation | `user_signed_up` | 等接到 auth callback 完成事件（Better Auth 的 signUp 成功 hook） | +| Retention | `voice_mode_activated` | 需要先在 hearing store 加显式 `enableVoiceMode` action — 当前 hearing 没有单一"用户主动启用"那一刻的 trigger，被动监听 + 录音 action 不构成 user intent 信号 | +| Feature adoption | `flux_image_generated` | 等图片生成 feature 上线 | + +### 双展示指标（同名两边都有） + +| 指标 | Grafana | PostHog | Truth | 语义差异 | +|---|---|---|---|---| +| 活跃用户数 | `user_active_sessions`（Postgres session 计数） | DAU = 去重 distinctId | **PostHog** | Grafana 是 active **sessions**，PostHog 是 active **users** | +| Checkout 完成数 | `stripe_checkout_completed_total` | `payment_completed` event | **Postgres** | 两边都展示，Grafana 是 webhook 计数，PostHog 是漏斗终点 | +| LLM 请求 | `gen_ai_client_operation_count_total` | `chat_session_started` 等 | **Grafana**（系统计数） | PostHog 是用户维度切片，会少于 Grafana（PostHog 只覆盖 logged-in user） | + +## PostHog 接入路线图 + +落地分两步，**不要一次性埋全部事件**，否则 schema 漂移会很快出现。 + +### 阶段 1（P0 — 付费漏斗 + activation） + +`apps/server`： + +```ts +// services/posthog.ts（新增） +import { PostHog } from 'posthog-node' + +export function createPostHog(env: ServerEnv) { + return new PostHog(env.POSTHOG_KEY, { host: 'https://us.i.posthog.com' }) +} + +// 在 Stripe webhook handler 里 +posthog.capture({ + distinctId: stripeCustomerEmail, + event: 'payment_completed', + properties: { plan_id, amount_usd, stripe_customer_id, stripe_subscription_id } +}) +``` + +`apps/stage-web`： + +```ts +import posthog from 'posthog-js' + +posthog.init(import.meta.env.VITE_POSTHOG_KEY, { + api_host: 'https://us.i.posthog.com', + capture_pageview: false, // 手动 capture 控制语义 +}) +// 登录后 +posthog.identify(user.id) +// 在 pricing.vue +posthog.capture('pricing_page_viewed', { plan_period, source }) +``` + +`apps/stage-tamagotchi`（Electron renderer）： + +```ts +// NOTICE: Electron CSP 下普通 import 会静默失效，必须用 full bundle。 +// 参考：https://posthog.com/tutorials/electron-analytics +import posthog from 'posthog-js/dist/module.full.no-external.js' + +posthog.init(import.meta.env.VITE_POSTHOG_KEY, { + api_host: 'https://us.i.posthog.com', + autocapture: false, // 桌面应用没有传统 URL 路由，手动控制 +}) +``` + +埋点事件清单（P0）： + +- 前端：`pricing_page_viewed`、`plan_selected`、`checkout_started`、`user_signed_up`、`first_message_sent`、`first_model_selected` +- 后端：`payment_completed` + +PostHog UI 配两个 funnel： + +- **付费漏斗** (7d 窗口)：`pricing_page_viewed → plan_selected → checkout_started → payment_completed` +- **激活漏斗** (14d 窗口)：`user_signed_up → first_message_sent → first_model_selected → payment_completed` + +### 阶段 2（P1 — retention / feature adoption / churn） + +埋点事件清单：`character_created`、`voice_mode_activated`、`chat_session_started`、`model_switched`、`flux_image_generated`、`subscription_cancelled`。 + +PostHog UI 配 cohort： + +- **D7 Retention by voice mode**：第一次 session 用了 `voice_mode_activated` 的用户 vs 没用的，看 D7/D30 retention 差异 +- **Churn 14d**：过去 14d 没有 `chat_session_started` 的付费用户，作为召回 cohort + +### Stripe → PostHog 集成路径 + +**两条路径都接**： + +| 路径 | 用途 | +|---|---| +| PostHog Stripe **source connector** | MRR / ARR / churn revenue dashboard（PostHog 原生 Revenue analytics） | +| **手动 capture** `payment_completed`（后端 webhook） | 漏斗终点 event，跟前端 `checkout_started` 串联 | + +不能只用 source connector：它是 data warehouse 层，**不生成 person event，做不了漏斗**。 + +## Grafana Alert SOP + +Alert rules **不放在** `apps/server/otel/grafana/dashboards/build.ts` 里——Grafana Cloud 用 Unified Alerting，rule 在 Grafana UI 或 alerting API 管理，跟 dashboard JSON 解耦。这一节维护我们应该配的 alert rule，新加 rule 时同步更新这里。 + +### P0 — page on-call（PagerDuty / Slack on-call channel） + +| Alert | Query | Threshold | Notes | +|---|---|---|---| +| **Flux Unbilled leak** | `increase(airi_billing_flux_unbilled_total[5m])` | `> 0` for 5m | 收入直接漏；分 `reason` label 看是 `partial_debit_drained`（用户余额耗尽，预期）还是 `debit_failed`（DB / 真异常）。后者更急 | +| **5xx Rate spike** | `100 * sum(rate(http_server_request_duration_seconds_count{http_response_status_code=~"5.."}[5m])) / sum(rate(http_server_request_duration_seconds_count[5m]))` | `> 5%` for 10m | 跟 panel-4 阈值对齐 | +| **Email Failure spike** | `100 * sum(rate(airi_email_failures_total[5m])) / clamp_min(sum(rate(airi_email_send_total[5m])) + sum(rate(airi_email_failures_total[5m])), 1)` | `> 5%` for 10m | Resend / DNS / 黑名单挂了会阻塞注册流程 | + +### P1 — notify only（Slack ops channel，不分页） + +| Alert | Query | Threshold | Notes | +|---|---|---|---| +| **WS Connections cliff** | `sum(ws_connections_active)` | drop to 0 for 5m | 全断说明部署 / LB 异常 | +| **DB Pool exhaustion** | `max by (service_instance_id) (db_client_connection_count)` | `>= DB_POOL_MAX - 1` for 5m | 哪个 instance 满了 | +| **Heap > 85%** | `100 * sum by (service_instance_id) (v8js_memory_heap_used_bytes) / sum by (service_instance_id) (v8js_memory_heap_limit_bytes)` | `> 85%` for 15m | 内存泄漏前兆 | +| **Stripe webhook fail** | `increase(stripe_events_total{event_type="payment_intent.payment_failed"}[1h])` | `> 10` per hour | 支付链路问题 | + +### 配置入口 + +Grafana Cloud → Alerts & IRM → Alert rules → New alert rule。把上面 query 粘进 PromQL editor，threshold 按表设置，labels 加 `severity=p0|p1`，notification policy 按 severity 路由到 PagerDuty 或 Slack。 + +每加一条 alert，**更新这张表**——alert 没在文档里登记 = 不知道为什么 page、不知道 owner、不知道历史阈值改动。 + +## 何时打破规则 + +这份文档定的是**默认值**，不是法律。下列情况可以打破： + +- **系统指标也需要给 PM 看**（如 LLM provider 可用性影响产品决策）→ Grafana truth + 周期性 export 给 PostHog dashboard 展示 +- **产品指标需要分钟级告警**（如付费转化突然归零）→ Grafana alert 监 Stripe webhook 计数，PostHog truth 不变 +- **A/B test 影响系统指标**（如新 LLM router 影响延迟）→ feature flag 同时打到两边，Grafana panel 按 flag value 分线展示 + +打破规则的指标必须在 dashboard description 里说明，**不要静默打破**。 + +## 参考来源 + +业界没有权威 framework，下列来源是这份文档的依据： + +- [PostHog Product Metrics Handbook](https://posthog.com/handbook/product/metrics) — PostHog 自己的内部分层 +- [PostHog issue #43633](https://github.com/posthog/posthog/issues/43633) — dual-emit 问题的工程承认 +- [Honeycomb Observability 2.0](https://www.honeycomb.io/blog/time-to-version-observability-signs-point-to-yes) — "消除工具边界"的少数派立场 +- [Reforge: North Star Metrics](https://www.reforge.com/blog/north-star-metrics) — leading vs lagging 区分 +- [DEV: Metrics for 500 Engineers with Linear + Grafana + PostHog](https://dev.to/johalputt/how-to-set-up-developer-metrics-for-500-engineers-using-linear-20-grafana-110-and-posthog-30-3l73) — 与我们结构最接近的公开案例 +- [PostHog: Stripe payment platform](https://posthog.com/docs/revenue-analytics/payment-platforms/stripe) — Stripe 集成路径官方文档 +- [PostHog: Electron analytics](https://posthog.com/tutorials/electron-analytics) — Electron renderer 接入要点 +- [Google SRE Book: Monitoring Distributed Systems](https://sre.google/sre-book/monitoring-distributed-systems/) — Four Golden Signals +- [Stripe: Essential SaaS Metrics](https://stripe.com/resources/more/essential-saas-metrics) — 收入侧指标定义 diff --git a/apps/server/otel/grafana/dashboards/airi-server-overview-cloud.json b/apps/server/otel/grafana/dashboards/airi-server-overview-cloud.json index fe662457a..a2b91b0a5 100644 --- a/apps/server/otel/grafana/dashboards/airi-server-overview-cloud.json +++ b/apps/server/otel/grafana/dashboards/airi-server-overview-cloud.json @@ -530,97 +530,6 @@ } } }, - "panel-7": { - "kind": "Panel", - "spec": { - "data": { - "kind": "QueryGroup", - "spec": { - "queries": [ - { - "kind": "PanelQuery", - "spec": { - "hidden": false, - "query": { - "datasource": { - "name": "grafanacloud-projairi-prom" - }, - "group": "prometheus", - "kind": "DataQuery", - "spec": { - "expr": "sum by (http_request_method) (increase(http_server_request_duration_seconds_count{service_name=~\"$service\", deployment_environment=~\"$env\", http_request_method!=\"OPTIONS\"}[5m]))", - "legendFormat": "{{http_request_method}}" - }, - "version": "v0" - }, - "refId": "A" - } - } - ], - "queryOptions": {}, - "transformations": [] - } - }, - "description": "Share of inbound HTTP requests by method. Skew toward POST often signals a misbehaving client; surprise PUT/DELETE may indicate stale clients.", - "id": 7, - "links": [], - "title": "HTTP Methods (last 5m)", - "vizConfig": { - "group": "piechart", - "kind": "VizConfig", - "spec": { - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "custom": { - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - } - }, - "unit": "short", - "noValue": "no traffic" - }, - "overrides": [] - }, - "options": { - "displayLabels": [ - "percent" - ], - "legend": { - "calcs": [ - "lastNotNull" - ], - "displayMode": "table", - "placement": "right", - "showLegend": true, - "values": [ - "value", - "percent" - ] - }, - "pieType": "donut", - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false - }, - "tooltip": { - "hideZeros": false, - "mode": "single", - "sort": "none" - } - } - }, - "version": "13.0.0-23630096546" - } - } - }, "panel-8": { "kind": "Panel", "spec": { @@ -730,8 +639,8 @@ "group": "prometheus", "kind": "DataQuery", "spec": { - "expr": "sum by (http_response_status_code) (increase(http_server_request_duration_seconds_count{service_name=~\"$service\", deployment_environment=~\"$env\", http_request_method!=\"OPTIONS\"}[5m]))", - "legendFormat": "{{http_response_status_code}}" + "expr": "topk(10, sum by (http_route) (increase(http_server_request_duration_seconds_count{service_name=~\"$service\", deployment_environment=~\"$env\", http_request_method!=\"OPTIONS\", http_route!=\"\"}[5m])))", + "legendFormat": "{{http_route}}" }, "version": "v0" }, @@ -743,10 +652,10 @@ "transformations": [] } }, - "description": "Distribution of response codes. A healthy server is ~95%+ 2xx — yellow/red slices stand out instantly.", + "description": "Top 10 Hono-matched routes by request count over the last 5 minutes. Answers \"which endpoint is being hit, and how much\" — replaces the previous HTTP status-code donut whose 2xx slice dominated everything else. Cardinality is bounded because `http_route` is the matched route pattern, not the concrete URL.", "id": 9, "links": [], - "title": "HTTP Status Codes (last 5m)", + "title": "Top Routes by Requests (last 5m)", "vizConfig": { "group": "piechart", "kind": "VizConfig", @@ -1170,6 +1079,238 @@ } } }, + "panel-13": { + "kind": "Panel", + "spec": { + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "hidden": false, + "query": { + "datasource": { + "name": "grafanacloud-projairi-prom" + }, + "group": "prometheus", + "kind": "DataQuery", + "spec": { + "expr": "sum(ws_connections_active{service_name=~\"$service\", deployment_environment=~\"$env\"})", + "legendFormat": "connections" + }, + "version": "v0" + }, + "refId": "A" + } + } + ], + "queryOptions": {}, + "transformations": [] + } + }, + "description": "Live WebSocket connection count over time. Row 1's stat shows the current value; this panel lets you correlate connection-count changes with deploys, network blips, or message throughput spikes. Same gauge as Row 1, charted instead of `lastNotNull`.", + "id": 13, + "links": [], + "title": "WS Connections", + "vizConfig": { + "group": "timeseries", + "kind": "VizConfig", + "spec": { + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 20, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "options": { + "annotations": { + "clustering": -1, + "multiLane": false + }, + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + } + }, + "version": "13.0.0-23630096546" + } + } + }, + "panel-14": { + "kind": "Panel", + "spec": { + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "hidden": false, + "query": { + "datasource": { + "name": "grafanacloud-projairi-prom" + }, + "group": "prometheus", + "kind": "DataQuery", + "spec": { + "expr": "topk(10, sum by (http_route) (rate(http_server_request_duration_seconds_count{service_name=~\"$service\", deployment_environment=~\"$env\", http_request_method!=\"OPTIONS\", http_route!=\"\"}[$__rate_interval])))", + "legendFormat": "{{http_route}}" + }, + "version": "v0" + }, + "refId": "A" + } + } + ], + "queryOptions": {}, + "transformations": [] + } + }, + "description": "Per-route request rate, top 10 by current rate. Pair with Row 4 P95 to spot hot endpoints that are also slow. Cardinality is the Hono-matched route pattern (e.g. `/api/v1/openai/v1/chat/completions`), not the concrete URL.", + "id": 14, + "links": [], + "title": "HTTP Request Rate by Route (top 10)", + "vizConfig": { + "group": "timeseries", + "kind": "VizConfig", + "spec": { + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 20, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "reqps" + }, + "overrides": [] + }, + "options": { + "annotations": { + "clustering": -1, + "multiLane": false + }, + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + } + }, + "version": "13.0.0-23630096546" + } + } + }, "panel-20": { "kind": "Panel", "spec": { @@ -2673,19 +2814,6 @@ "kind": "GridLayout", "spec": { "items": [ - { - "kind": "GridLayoutItem", - "spec": { - "element": { - "kind": "ElementReference", - "name": "panel-7" - }, - "height": 7, - "width": 8, - "x": 0, - "y": 0 - } - }, { "kind": "GridLayoutItem", "spec": { @@ -2694,8 +2822,8 @@ "name": "panel-8" }, "height": 7, - "width": 8, - "x": 8, + "width": 12, + "x": 0, "y": 0 } }, @@ -2707,8 +2835,8 @@ "name": "panel-9" }, "height": 7, - "width": 8, - "x": 16, + "width": 12, + "x": 12, "y": 0 } } @@ -2734,7 +2862,7 @@ "name": "panel-10" }, "height": 8, - "width": 8, + "width": 6, "x": 0, "y": 0 } @@ -2747,8 +2875,8 @@ "name": "panel-11" }, "height": 8, - "width": 8, - "x": 8, + "width": 6, + "x": 6, "y": 0 } }, @@ -2760,8 +2888,21 @@ "name": "panel-12" }, "height": 8, - "width": 8, - "x": 16, + "width": 6, + "x": 12, + "y": 0 + } + }, + { + "kind": "GridLayoutItem", + "spec": { + "element": { + "kind": "ElementReference", + "name": "panel-13" + }, + "height": 8, + "width": 6, + "x": 18, "y": 0 } } @@ -2771,6 +2912,33 @@ "title": "Traffic Trends" } }, + { + "kind": "RowsLayoutRow", + "spec": { + "collapse": false, + "layout": { + "kind": "GridLayout", + "spec": { + "items": [ + { + "kind": "GridLayoutItem", + "spec": { + "element": { + "kind": "ElementReference", + "name": "panel-14" + }, + "height": 7, + "width": 24, + "x": 0, + "y": 0 + } + } + ] + } + }, + "title": "Top Endpoints" + } + }, { "kind": "RowsLayoutRow", "spec": { diff --git a/apps/server/otel/grafana/dashboards/build.ts b/apps/server/otel/grafana/dashboards/build.ts index 8c49066d9..278d122e9 100644 --- a/apps/server/otel/grafana/dashboards/build.ts +++ b/apps/server/otel/grafana/dashboards/build.ts @@ -414,17 +414,12 @@ elements['panel-6'] = gaugePanel( // Row 2: Distribution — "what KIND of traffic right now?" // Donuts answer the current breakdown question better than stacked area. // Use `topk(N, ...)` so a long-tail label set doesn't render an unreadable -// 30-slice pie. -elements['panel-7'] = piePanel( - 7, - 'HTTP Methods (last 5m)', - 'Share of inbound HTTP requests by method. Skew toward POST often signals a misbehaving client; surprise PUT/DELETE may indicate stale clients.', - [query( - `sum by (http_request_method) (increase(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS"}[5m]))`, - '{{http_request_method}}', - )], -) - +// 30-slice pie. HTTP method donut was removed because the dimension is so +// low-cardinality (GET/POST/DELETE/HEAD) that the slice ratios barely move — +// the by-method timeseries in Row 3 already conveys the same information +// with time context. Status-code donut was replaced by Top Routes because +// the 2xx slice dominates and the by-status timeseries in Row 5 already +// surfaces 4xx/5xx independently. elements['panel-8'] = piePanel( 8, 'LLM Models (last 5m)', @@ -437,11 +432,11 @@ elements['panel-8'] = piePanel( elements['panel-9'] = piePanel( 9, - 'HTTP Status Codes (last 5m)', - 'Distribution of response codes. A healthy server is ~95%+ 2xx — yellow/red slices stand out instantly.', + 'Top Routes by Requests (last 5m)', + 'Top 10 Hono-matched routes by request count over the last 5 minutes. Answers "which endpoint is being hit, and how much" — replaces the previous HTTP status-code donut whose 2xx slice dominated everything else. Cardinality is bounded because `http_route` is the matched route pattern, not the concrete URL.', [query( - `sum by (http_response_status_code) (increase(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS"}[5m]))`, - '{{http_response_status_code}}', + `topk(10, sum by (http_route) (increase(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS", http_route!=""}[5m])))`, + '{{http_route}}', )], ) @@ -479,6 +474,28 @@ elements['panel-12'] = timeseriesPanel( { unit: 'ops' }, ) +elements['panel-13'] = timeseriesPanel( + 13, + 'WS Connections', + 'Live WebSocket connection count over time. Row 1\'s stat shows the current value; this panel lets you correlate connection-count changes with deploys, network blips, or message throughput spikes. Same gauge as Row 1, charted instead of `lastNotNull`.', + [query(`sum(ws_connections_active{${SERVICE_FILTER}})`, 'connections')], + { unit: 'short' }, +) + +// Row 3.5: Top Endpoints — Row 3 answered "by method/model/WS-channel". +// This row answers "by route" which has higher cardinality and warrants +// a full-width panel + topk so the legend stays readable. +elements['panel-14'] = timeseriesPanel( + 14, + 'HTTP Request Rate by Route (top 10)', + 'Per-route request rate, top 10 by current rate. Pair with Row 4 P95 to spot hot endpoints that are also slow. Cardinality is the Hono-matched route pattern (e.g. `/api/v1/openai/v1/chat/completions`), not the concrete URL.', + [query( + `topk(10, sum by (http_route) (rate(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS", http_route!=""}[$__rate_interval])))`, + '{{http_route}}', + )], + { unit: 'reqps' }, +) + // Row 4: Latency — how slow we are elements['panel-20'] = timeseriesPanel( 20, @@ -648,17 +665,29 @@ const rows = [ item('panel-5', 16, 0, 4, 4), item('panel-6', 20, 0, 4, 4), ]), - // Row 2: 3 donuts × 8 wide × 7 high — current-state distribution + // Row 2: 2 donuts × 12 wide × 7 high — current-state distribution + // Dropped HTTP-methods donut (low-cardinality, redundant with Row 3 by-method + // timeseries) and HTTP-status donut (2xx dominated, 4xx/5xx already broken + // out in Row 5). Replaced status donut with Top Routes which answers a + // higher-information question with the same visual budget. row('Distribution (now)', [ - item('panel-7', 0, 0, 8, 7), - item('panel-8', 8, 0, 8, 7), - item('panel-9', 16, 0, 8, 7), + item('panel-8', 0, 0, 12, 7), + item('panel-9', 12, 0, 12, 7), ]), - // Row 3: 3 timeseries × 8 wide × 8 high — same data as Row 2 but over time + // Row 3: 4 timeseries × 6 wide × 8 high — same data as Row 2 but over time. + // WS Connections joined this row so the live WS state can be correlated + // with HTTP/LLM/WS-message trends on the same time axis. row('Traffic Trends', [ - item('panel-10', 0, 0, 8, 8), - item('panel-11', 8, 0, 8, 8), - item('panel-12', 16, 0, 8, 8), + item('panel-10', 0, 0, 6, 8), + item('panel-11', 6, 0, 6, 8), + item('panel-12', 12, 0, 6, 8), + item('panel-13', 18, 0, 6, 8), + ]), + // Row 3.5: 1 timeseries × 24 wide × 7 high — top routes get the full width + // because route-level cardinality (~5-10 series after topk) needs space + // for the legend table. + row('Top Endpoints', [ + item('panel-14', 0, 0, 24, 7), ]), // Row 4: 2 timeseries × 12 wide × 8 high row('Latency', [ diff --git a/apps/server/package.json b/apps/server/package.json index 197bb1221..3bf6d839a 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -55,6 +55,7 @@ "ioredis": "^5.10.1", "jose": "catalog:", "pg": "^8.20.0", + "posthog-node": "catalog:", "resend": "^6.12.2", "stripe": "^22.0.2", "valibot": "catalog:", diff --git a/apps/server/src/app.test.ts b/apps/server/src/app.test.ts index a35fa2415..deceea820 100644 --- a/apps/server/src/app.test.ts +++ b/apps/server/src/app.test.ts @@ -68,6 +68,7 @@ function createTestDeps() { } as any, otel: null, userDeletionService: {} as any, + posthog: null, } return { diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 0c6ad1412..e06978a82 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -1,4 +1,5 @@ import type Redis from 'ioredis' +import type { PostHog } from 'posthog-node' import type { AuthInstance } from './libs/auth' import type { Database } from './libs/db' @@ -59,6 +60,7 @@ import { createConfigKVService } from './services/config-kv' import { createEmailService } from './services/email' import { createFluxService } from './services/flux' import { createFluxTransactionService } from './services/flux-transaction' +import { createPostHogClient } from './services/posthog' import { createProviderService } from './services/providers' import { createRequestLogService } from './services/request-log' import { createStripeService } from './services/stripe' @@ -85,6 +87,7 @@ interface AppDeps { env: Env otel: OtelInstance | null userDeletionService: UserDeletionService + posthog: PostHog | null } export async function buildApp(deps: AppDeps) { @@ -238,7 +241,7 @@ export async function buildApp(deps: AppDeps) { /** * Stripe routes. */ - .route('/api/v1/stripe', createStripeRoutes(deps.fluxService, deps.stripeService, deps.billingService, deps.configKV, deps.env, deps.redis, deps.otel?.revenue, deps.otel?.rateLimit)) + .route('/api/v1/stripe', createStripeRoutes(deps.fluxService, deps.stripeService, deps.billingService, deps.configKV, deps.env, deps.redis, deps.otel?.revenue, deps.otel?.rateLimit, deps.posthog)) /** * Admin routes — guarded by `ADMIN_EMAILS` allowlist + verified email. @@ -352,6 +355,44 @@ export async function createApp() { }, undefined, dependsOn.otel?.email), }) + // Webhook capture path goes through `captureSafe` → `captureImmediate`, + // which awaits the HTTP send inline, so individual events never sit in + // the background queue. `flush()` + `_shutdown()` on SIGTERM is the belt- + // and-suspenders drain for any future call site that uses the regular + // `capture()` (which only enqueues). + // + // NOTICE: + // We use the underscore-prefixed `_shutdown` despite its "internal" naming + // because the public `shutdown(timeoutMs)` returns void (`types.d.ts:580`) + // — there is no way to await its completion. `_shutdown(timeoutMs)` returns + // `Promise` (`client.d.ts:934`) and is the only way to ensure the + // process doesn't exit while PostHog cleanup is still running. Posthog's + // own examples show `await client._shutdown()` as the recommended pattern. + const posthog = injeca.provide('services:posthog', { + dependsOn: { env: parsedEnv, lifecycle }, + build: ({ dependsOn }) => { + const client = createPostHogClient(dependsOn.env) + if (client) { + dependsOn.lifecycle.appHooks.onStop(async () => { + try { + await client.flush() + } + catch { + // Flush failures on shutdown are non-fatal; we lose at most a + // few queued events. Fall through to shutdown anyway. + } + try { + await client._shutdown(5000) + } + catch { + // Shutdown errors are also non-fatal during process exit. + } + }) + } + return client + }, + }) + const characterService = injeca.provide('services:characters', { dependsOn: { db, otel }, build: ({ dependsOn }) => createCharacterService(dependsOn.db, dependsOn.otel?.engagement), @@ -484,6 +525,7 @@ export async function createApp() { env: parsedEnv, otel, userDeletionService, + posthog, }) // Register the cluster-wide ObservableGauge for active sessions. Each // replica polls the same DB (cached 10s, in-flight coalesced) and the @@ -509,6 +551,7 @@ export async function createApp() { env: resolved.env, otel: resolved.otel, userDeletionService: resolved.userDeletionService, + posthog: resolved.posthog, }) logger.withFields({ hostname: resolved.env.HOST, port: resolved.env.PORT }).log('Server started') diff --git a/apps/server/src/libs/auth.test.ts b/apps/server/src/libs/auth.test.ts index 24d6f613b..47ee266ed 100644 --- a/apps/server/src/libs/auth.test.ts +++ b/apps/server/src/libs/auth.test.ts @@ -85,6 +85,7 @@ describe('seedTrustedClients', () => { expect(pocketClient.tokenEndpointAuthMethod).toBe('none') expect(pocketClient.redirectUris).toEqual([ 'capacitor://localhost/auth/callback', + 'ai.moeru.airi-pocket://links/auth/callback', ]) }) diff --git a/apps/server/src/libs/env.ts b/apps/server/src/libs/env.ts index 2b8714c48..9c833cc2a 100644 --- a/apps/server/src/libs/env.ts +++ b/apps/server/src/libs/env.ts @@ -114,6 +114,11 @@ const EnvSchema = object({ STRIPE_SECRET_KEY: optional(string()), STRIPE_WEBHOOK_SECRET: optional(string()), + // PostHog server-side analytics. Optional — when unset the client is null + // and `captureSafe(...)` is a no-op so webhooks still complete. + POSTHOG_API_KEY: optional(string(), ''), + POSTHOG_HOST: optional(string(), 'https://us.i.posthog.com'), + // LLM gateway (infrastructure config — baked per deployment) GATEWAY_BASE_URL: pipe(string(), nonEmpty('GATEWAY_BASE_URL is required')), DEFAULT_CHAT_MODEL: pipe(string(), nonEmpty('DEFAULT_CHAT_MODEL is required')), diff --git a/apps/server/src/routes/stripe/index.ts b/apps/server/src/routes/stripe/index.ts index 4c2325d20..fe7d26974 100644 --- a/apps/server/src/routes/stripe/index.ts +++ b/apps/server/src/routes/stripe/index.ts @@ -1,4 +1,5 @@ import type Redis from 'ioredis' +import type { PostHog } from 'posthog-node' import type { Env } from '../../libs/env' import type { RateLimitMetrics, RevenueMetrics } from '../../otel' @@ -16,6 +17,7 @@ import { safeParse } from 'valibot' import { authGuard } from '../../middlewares/auth' import { rateLimiter } from '../../middlewares/rate-limit' +import { captureSafe } from '../../services/posthog' import { createBadRequestError, createServiceUnavailableError } from '../../utils/error' import { errorMessageFromUnknown } from '../../utils/error-message' import { resolveTrustedRequestOrigin } from '../../utils/origin' @@ -50,6 +52,7 @@ export function createStripeRoutes( redis: Redis, metrics?: RevenueMetrics | null, rateLimitMetrics?: RateLimitMetrics | null, + posthog?: PostHog | null, ) { const stripe = env.STRIPE_SECRET_KEY ? new Stripe(env.STRIPE_SECRET_KEY) : null @@ -285,7 +288,7 @@ export function createStripeRoutes( switch (event.type) { case 'checkout.session.completed': { - await handleCheckoutSessionCompleted(event.id, event.data.object, fluxService, stripeService, billingService) + const result = await handleCheckoutSessionCompleted(event.id, event.data.object, fluxService, stripeService, billingService) metrics?.stripeCheckoutCompleted.add(1) // Revenue capture in smallest currency unit (e.g. cents). // Cross-currency aggregation is meaningless, so always group by `currency` in queries. @@ -295,6 +298,15 @@ export function createStripeRoutes( source: 'checkout', }) } + // PostHog: funnel terminator. Only fire when the handler actually + // processed the checkout — malformed sessions (missing userId, + // invalid fluxAmount) take the early-return path above and would + // otherwise poison the funnel with phantom conversions. distinctId + // is the Better Auth user id so it merges with the browser's + // `posthog.identify(userId)` and the prior `checkout_started` + // event lines up. See docs/ai-context/metrics-ownership.md. + if (result.processed) + await capturePaymentCompleted(posthog, event.data.object) break } case 'customer.created': @@ -307,6 +319,8 @@ export function createStripeRoutes( case 'customer.subscription.deleted': { await handleSubscriptionEvent(event.data.object, stripeService) metrics?.stripeSubscriptionEvent.add(1, { event_type: event.type.replace('customer.subscription.', '') }) + if (event.type === 'customer.subscription.deleted') + await captureSubscriptionCancelled(posthog, stripeService, event.data.object) break } case 'invoice.created': @@ -339,11 +353,11 @@ async function handleCheckoutSessionCompleted( fluxService: FluxService, stripeService: StripeService, billingService: BillingService, -) { +): Promise<{ processed: boolean }> { const userId = session.metadata?.userId if (!userId) { logger.withFields({ sessionId: session.id }).warn('Checkout session missing userId in metadata') - return + return { processed: false } } logger.withFields({ userId, sessionId: session.id, mode: session.mode, amount: session.amount_total, currency: session.currency }).log('Processing checkout session') @@ -378,13 +392,27 @@ async function handleCheckoutSessionCompleted( }) // Idempotent flux credit: use fluxCredited flag inside a transaction - // to prevent double-crediting on webhook replay - const metadataFlux = session.metadata?.fluxAmount - if (session.mode === 'payment' && session.amount_total != null && metadataFlux) { + // to prevent double-crediting on webhook replay. + // + // For `payment` mode (one-time Flux purchase) `metadata.fluxAmount` is + // required — without it we can't credit anything, and the funnel must + // not see a `payment_completed` event for a checkout that didn't + // actually deliver Flux. Non-`payment` modes (e.g. `setup` for saving + // a card) deliberately skip crediting and still count as processed. + if (session.mode === 'payment') { + if (session.amount_total == null) { + logger.withFields({ userId, sessionId: session.id }).warn('Payment-mode checkout missing amount_total; skipping credit and capture') + return { processed: false } + } + const metadataFlux = session.metadata?.fluxAmount + if (!metadataFlux) { + logger.withFields({ userId, sessionId: session.id }).warn('Payment-mode checkout missing metadata.fluxAmount; skipping credit and capture') + return { processed: false } + } const fluxAmount = Number(metadataFlux) if (!Number.isFinite(fluxAmount) || fluxAmount <= 0) { logger.withFields({ userId, sessionId: session.id, metadataFlux }).warn('Invalid fluxAmount in session metadata, skipping credit') - return + return { processed: false } } const result = await billingService.creditFluxFromStripeCheckout({ @@ -404,6 +432,75 @@ async function handleCheckoutSessionCompleted( balanceAfter: result.balanceAfter, }).log('Processed flux credit for one-time payment') } + + return { processed: true } +} + +async function capturePaymentCompleted( + posthog: PostHog | null | undefined, + session: Stripe.Checkout.Session, +): Promise { + if (!posthog) + return + + const userId = session.metadata?.userId + const email = session.customer_email + || (typeof session.customer_details?.email === 'string' ? session.customer_details.email : null) + || null + + // distinctId fallback chain: userId (Better Auth, matches browser identify()) + // > email (PostHog will merge on identify later) > stripe session id (last + // resort — orphan event but at least we count it). + const distinctId = userId || email || session.id + + const fluxAmount = Number(session.metadata?.fluxAmount) + const stripeCustomerId = typeof session.customer === 'string' ? session.customer : session.customer?.id + + await captureSafe(posthog, { + distinctId, + event: 'payment_completed', + properties: { + amount_total: session.amount_total, + currency: session.currency, + flux_amount: Number.isFinite(fluxAmount) ? fluxAmount : null, + mode: session.mode, + stripe_session_id: session.id, + stripe_customer_id: stripeCustomerId, + stripe_subscription_id: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, + ...(userId ? { user_id: userId } : {}), + // $set populates the PostHog person profile so funnel joins work even + // when a user pays via direct checkout link before ever loading the SPA. + ...(email ? { $set: { email } } : {}), + }, + }) +} + +async function captureSubscriptionCancelled( + posthog: PostHog | null | undefined, + stripeService: StripeService, + subscription: Stripe.Subscription, +): Promise { + if (!posthog) + return + + const stripeCustomerId = typeof subscription.customer === 'string' ? subscription.customer : subscription.customer.id + const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) + const distinctId = customer?.userId || stripeCustomerId + + await captureSafe(posthog, { + distinctId, + event: 'subscription_cancelled', + properties: { + stripe_subscription_id: subscription.id, + stripe_customer_id: stripeCustomerId, + cancel_at_period_end: subscription.cancel_at_period_end, + cancellation_reason: subscription.cancellation_details?.reason ?? null, + cancellation_comment: subscription.cancellation_details?.comment ?? null, + canceled_at: subscription.canceled_at, + ended_at: subscription.ended_at, + ...(customer?.userId ? { user_id: customer.userId } : {}), + }, + }) } async function handleCustomerEvent( diff --git a/apps/server/src/services/billing/flux-meter.ts b/apps/server/src/services/billing/flux-meter.ts index 2ef996c7f..585c0c337 100644 --- a/apps/server/src/services/billing/flux-meter.ts +++ b/apps/server/src/services/billing/flux-meter.ts @@ -6,6 +6,7 @@ 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' const logger = useLogger('flux-meter') @@ -65,9 +66,19 @@ interface AccumulateInput { } interface AccumulateResult { + /** Actual flux charged to the user (== amount we are sure was billed). */ fluxDebited: number + /** Residual debt left in Redis after this call. Includes unbilled units restored on partial drain. */ debtAfter: number + /** User's flux balance after this call. */ balanceAfter: number + /** + * Flux that crossed the meter threshold but couldn't be charged because the + * user's balance was lower than what the request required. > 0 means the + * user received service they only partially paid for. Reflects the gap + * between `requested` and `charged` returned by `billingService.consumeFluxForLLM`. + */ + unbilledFlux: number } /** @@ -134,49 +145,46 @@ export function createFluxMeter( */ async function accumulate(input: AccumulateInput): Promise { if (!Number.isFinite(input.units) || input.units <= 0) - return { fluxDebited: 0, debtAfter: await readDebt(input.userId), balanceAfter: input.currentBalance } + return { fluxDebited: 0, debtAfter: await readDebt(input.userId), balanceAfter: input.currentBalance, unbilledFlux: 0 } const modelLabel = typeof input.metadata?.model === 'string' ? input.metadata.model : 'unknown' metrics?.ttsChars.add(input.units, { meter: config.name, model: modelLabel }) const runtime = await getRuntime() const key = userFluxMeterDebtRedisKey(input.userId, config.name) - const [fluxDebited, debtAfter] = await runScript(key, input.units, runtime) + const [fluxRequested, debtAfterSettlement] = await runScript(key, input.units, runtime) - if (fluxDebited === 0) { + if (fluxRequested === 0) { logger.withFields({ userId: input.userId, meter: config.name, units: input.units, - debtAfter, + debtAfter: debtAfterSettlement, }).debug('Accumulated units below flux threshold') - return { fluxDebited: 0, debtAfter, balanceAfter: input.currentBalance } + return { fluxDebited: 0, debtAfter: debtAfterSettlement, balanceAfter: input.currentBalance, unbilledFlux: 0 } } + let result: Awaited> try { - const { flux } = await billingService.consumeFluxForLLM({ + result = await billingService.consumeFluxForLLM({ userId: input.userId, - amount: fluxDebited, + amount: fluxRequested, requestId: input.requestId, description: `${config.name}_request`, ...(typeof input.metadata?.model === 'string' && { model: input.metadata.model }), }) - - return { fluxDebited, debtAfter, balanceAfter: flux } } catch (error) { - // Restore the already-settled portion back into the debt counter so the - // next successful request picks it up. Without this, a failed debit - // (insufficient balance under concurrency, transient DB error) silently - // under-bills the user for `fluxDebited * unitsPerFlux` units. - const restoreUnits = fluxDebited * runtime.unitsPerFlux + // The billing call threw (balance <= 0 hard floor, transient DB error, + // network blip). The debit did NOT commit, so restore the full + // already-settled portion back into the debt counter for the next + // request to retry. + const restoreUnits = fluxRequested * runtime.unitsPerFlux try { await redis.incrby(key, restoreUnits) await redis.expire(key, runtime.debtTtlSeconds) } catch (rollbackError) { - // Rollback failure is itself a billing leak; log loudly for manual - // reconciliation but do not shadow the original error. logger.withError(rollbackError).withFields({ userId: input.userId, meter: config.name, @@ -186,6 +194,87 @@ export function createFluxMeter( } throw error } + + // Billing-service invariant — checked OUTSIDE the try/catch above so a + // post-debit assertion failure does NOT trigger the "restore full debt" + // rollback path. The DB tx already committed `result.charged`; restoring + // `fluxRequested * unitsPerFlux` would set up a double-charge on the + // next request (LUA re-settles the restored debt, billing re-debits the + // same usage). Surface loud, but don't compensate. + if (!Number.isInteger(result.charged) || result.charged < 0 || result.charged > result.requested) { + logger.withFields({ + userId: input.userId, + meter: config.name, + requestId: input.requestId, + requested: result.requested, + charged: result.charged, + }).error('billing-service returned invalid charged/requested — manual reconciliation needed') + throw new Error(`billing-service returned invalid charged=${result.charged} for requested=${result.requested}`) + } + + // Partial-debit path: balance was insufficient and `debitFlux` drained + // it to zero. We've already DECRBY'd `fluxRequested * unitsPerFlux` from + // the debt counter via the LUA script, but only `result.charged` of + // those flux were actually billed. Restore the gap so the debt counter + // reflects the user's true outstanding obligation, and surface it on + // the same `fluxUnbilled` counter the streaming/non-streaming chat + // paths use (different `reason` label). + // + // REVIEW: Settlement (LUA `runScript`) and the `INCRBY` restore below + // are not atomic. A concurrent `accumulate()` could observe the debt + // counter mid-window (between DECRBY and the restore INCRBY) and + // mis-bill. In practice the window is small (one in-flight DB tx) and + // a re-billing attempt would land in the same partial-debit branch, + // but the right long-term fix is either a short Redis lock keyed by + // `{userId, meter}` around `runScript → consumeFluxForLLM → restore`, + // or moving the unbilled portion into a separate Redis key that the + // LUA script doesn't touch. See codex review thread on PR. + if (result.charged < result.requested) { + const unbilledFlux = result.requested - result.charged + const restoreUnits = unbilledFlux * runtime.unitsPerFlux + + metrics?.fluxUnbilled.add(unbilledFlux, { + source: 'tts_meter', + meter: config.name, + reason: 'partial_debit_drained', + ...(typeof input.metadata?.model === 'string' && { [GEN_AI_ATTR_REQUEST_MODEL]: input.metadata.model }), + }) + + let debtAfterRestore = debtAfterSettlement + try { + debtAfterRestore = await redis.incrby(key, restoreUnits) + await redis.expire(key, runtime.debtTtlSeconds) + } + catch (rollbackError) { + // Log loudly so on-call can reconcile manually; don't shadow the + // partial-debit signal by re-throwing. + logger.withError(rollbackError).withFields({ + userId: input.userId, + meter: config.name, + restoreUnits, + requestId: input.requestId, + }).error('Failed to restore meter debt after partial-debit drain') + } + + logger.withFields({ + userId: input.userId, + meter: config.name, + requestId: input.requestId, + requested: result.requested, + charged: result.charged, + unbilledFlux, + restoreUnits, + }).warn('Partial debit on flux meter — flux drained to zero') + + return { + fluxDebited: result.charged, + debtAfter: debtAfterRestore, + balanceAfter: result.flux, + unbilledFlux, + } + } + + return { fluxDebited: result.charged, debtAfter: debtAfterSettlement, balanceAfter: result.flux, unbilledFlux: 0 } } return { diff --git a/apps/server/src/services/billing/tests/flux-meter.test.ts b/apps/server/src/services/billing/tests/flux-meter.test.ts index 5fd184347..0ceb8dfa2 100644 --- a/apps/server/src/services/billing/tests/flux-meter.test.ts +++ b/apps/server/src/services/billing/tests/flux-meter.test.ts @@ -56,16 +56,31 @@ function createMockRedis() { } } -function createMockBilling(opts: { throwOn?: number } = {}): BillingService { +function createMockBilling(opts: { throwOn?: number, partialChargeOn?: { amount: number, charged: number } } = {}): BillingService { return { consumeFluxForLLM: vi.fn(async ({ userId, amount }: { userId: string, amount: number }) => { if (opts.throwOn != null && amount === opts.throwOn) throw new Error('mock billing failure') - return { userId, flux: 100 - amount } + // Mirror real billing-service partial-debit semantics: drain to zero + // returns `charged < requested`. + if (opts.partialChargeOn != null && amount === opts.partialChargeOn.amount) { + return { userId, flux: 0, charged: opts.partialChargeOn.charged, requested: amount } + } + return { userId, flux: 100 - amount, charged: amount, requested: amount } }), } as unknown as BillingService } +function createMockMetrics() { + const fluxUnbilled = { add: vi.fn() } + const ttsChars = { add: vi.fn() } + const ttsPreflightRejections = { add: vi.fn() } + return { + metrics: { fluxUnbilled, ttsChars, ttsPreflightRejections } as any, + fluxUnbilled, + } +} + function staticRuntime(unitsPerFlux = 1000, debtTtlSeconds = 60) { return vi.fn(async () => ({ unitsPerFlux, debtTtlSeconds })) } @@ -89,7 +104,7 @@ describe('fluxMeter', () => { requestId: 'req-1', }) - expect(result).toEqual({ fluxDebited: 0, debtAfter: 500, balanceAfter: 10 }) + expect(result).toEqual({ fluxDebited: 0, debtAfter: 500, balanceAfter: 10, unbilledFlux: 0 }) expect(billing.consumeFluxForLLM).not.toHaveBeenCalled() }) @@ -193,4 +208,66 @@ describe('fluxMeter', () => { expect(await meter.peekDebt('u1')).toBe(2500) expect(mockRedis.incrby).toHaveBeenCalledWith(expect.stringContaining('u1'), 2000) }) + + // ROOT CAUSE: + // + // Prior to commit 7267b0d6b billing-service.consumeFluxForLLM threw on + // any insufficient-balance, which let flux-meter's catch path restore + // the *entire* settled portion back to the debt counter. After that + // commit billing-service introduced partial-debit semantics: when + // 0 < balance < amount, it drains the balance to zero and returns + // `charged < requested` instead of throwing. flux-meter.accumulate + // continued to read only `{ flux }` from the result, so: + // - the un-charged portion (`requested - charged` flux) was silently + // lost — Redis debt was already DECRBY'd by the LUA script, + // - airi_billing_flux_unbilled_total never fired for tts_meter, so + // the partial-debit revenue leak was invisible in Grafana. + // + // After patch: accumulate destructures `charged / requested`, restores + // `(requested - charged) * unitsPerFlux` back into the debt counter, and + // increments airi_billing_flux_unbilled_total with + // `{ source: 'tts_meter', reason: 'partial_debit_drained' }`. + it('restores partial-drain delta to debt and reports fluxUnbilled (Issue: unpaid-usage-exploit follow-up)', async () => { + // After settlement the meter wants to debit 3 flux, but billing only + // manages to charge 1 (user balance was 1 flux). Expect: + // - fluxDebited == 1 (actual charged), not 3 + // - unbilledFlux == 2 + // - Redis debt restored by 2 * unitsPerFlux = 2000 + // - fluxUnbilled metric incremented by 2 with partial_debit_drained reason + const partialBilling = createMockBilling({ partialChargeOn: { amount: 3, charged: 1 } }) + const { metrics, fluxUnbilled } = createMockMetrics() + const meter = createFluxMeter(mockRedis.redis, partialBilling, { name: 'tts', resolveRuntime: staticRuntime() }, metrics) + + const result = await meter.accumulate({ + userId: 'u1', + units: 3500, + currentBalance: 1, + requestId: 'partial', + metadata: { model: 'eleven_multilingual_v2' }, + }) + + expect(result.fluxDebited).toBe(1) + expect(result.unbilledFlux).toBe(2) + expect(result.balanceAfter).toBe(0) + // Debt = 500 residual (LUA leftover) + 2000 restored from partial drain. + expect(await meter.peekDebt('u1')).toBe(2500) + expect(mockRedis.incrby).toHaveBeenCalledWith(expect.stringContaining('u1'), 2000) + expect(fluxUnbilled.add).toHaveBeenCalledWith(2, expect.objectContaining({ + 'source': 'tts_meter', + 'meter': 'tts', + 'reason': 'partial_debit_drained', + 'gen_ai.request.model': 'eleven_multilingual_v2', + })) + }) + + it('does not report fluxUnbilled when billing fully charges', async () => { + const { metrics, fluxUnbilled } = createMockMetrics() + const meter = createFluxMeter(mockRedis.redis, billing, { name: 'tts', resolveRuntime: staticRuntime() }, metrics) + + const result = await meter.accumulate({ userId: 'u1', units: 1500, currentBalance: 10, requestId: 'full' }) + + expect(result.fluxDebited).toBe(1) + expect(result.unbilledFlux).toBe(0) + expect(fluxUnbilled.add).not.toHaveBeenCalled() + }) }) diff --git a/apps/server/src/services/posthog.ts b/apps/server/src/services/posthog.ts new file mode 100644 index 000000000..5016c62ca --- /dev/null +++ b/apps/server/src/services/posthog.ts @@ -0,0 +1,84 @@ +import type { Env } from '../libs/env' + +import { useLogger } from '@guiiai/logg' +import { PostHog } from 'posthog-node' + +const logger = useLogger('posthog') + +/** + * Server-side PostHog client. Used to capture authoritative business events + * the browser cannot see (Stripe webhooks, subscription state changes, + * admin actions). Pairs with the browser-side PostHog already wired up in + * `packages/stage-ui/src/stores/analytics/posthog.ts`. + * + * Use when: + * - You're inside a server-side handler (Stripe webhook, admin route) and + * need to emit an event that will be analysed in PostHog funnels or + * cohorts (e.g. `payment_completed`, `subscription_cancelled`). + * + * Expects: + * - `POSTHOG_API_KEY` env var. When unset (dev/CI) this returns `null` so + * callers degrade gracefully with `posthog?.capture(...)`. + * - `distinctId` must match the browser's `posthog.identify(userId)` — we + * use the Better Auth `user.id` for that everywhere. Stripe webhooks + * that only have an email use the email as a fallback `distinctId` and + * include `userId` in the event properties so PostHog's merge resolves + * the person. + * + * Returns: + * - A `PostHog` client configured for low-latency immediate sends, or + * `null` when key is unset. Callers must use `captureSafe()` (which + * wraps `captureImmediate`) — the regular `capture()` only enqueues + * and would let webhook responses race ahead of the HTTP send. + */ +export function createPostHogClient(env: Env): PostHog | null { + if (!env.POSTHOG_API_KEY) { + logger.warn('POSTHOG_API_KEY is unset — server-side analytics disabled') + return null + } + + // NOTICE: + // `flushAt: 1` keeps the background-batch threshold low so any stray + // `posthog.capture()` (non-immediate path) flushes promptly. Real send- + // path for webhook events goes through `captureImmediate()` in + // `captureSafe`, which bypasses the queue entirely and resolves only + // after the HTTP round-trip. We also rely on `shutdown(timeoutMs)` from + // app.ts to drain any residual queue on SIGTERM. + return new PostHog(env.POSTHOG_API_KEY, { + host: env.POSTHOG_HOST || 'https://us.i.posthog.com', + flushAt: 1, + flushInterval: 0, + }) +} + +/** + * Safe capture wrapper. PostHog must never block or fail a webhook / + * billing path — any error here is logged and swallowed. + * + * Use when: + * - Inside a server-side handler that has business work to finish even + * if PostHog is down. The handler already wrote to Postgres and + * updated metrics; PostHog is the optional last step. + * + * Expects: + * - Caller awaits the returned promise. We use `captureImmediate` (not + * `capture`) because PostHog Node SDK's regular `capture` only enqueues + * — `flushAt: 1` triggers a *background* flush, which means the webhook + * handler can return before the event reaches PostHog and SIGTERM may + * strand the queued event. `captureImmediate` does the HTTP send inline + * and resolves only after the network round-trip. + */ +export async function captureSafe( + posthog: PostHog | null, + event: { distinctId: string, event: string, properties?: Record }, +): Promise { + if (!posthog) + return + + try { + await posthog.captureImmediate(event) + } + catch (err) { + logger.withError(err).withFields({ event: event.event, distinctId: event.distinctId }).warn('PostHog captureImmediate failed; swallowing to protect caller') + } +} diff --git a/apps/stage-web/src/pages/settings/characters/components/CharacterDialog.vue b/apps/stage-web/src/pages/settings/characters/components/CharacterDialog.vue index 5912fa75e..4f5c1e606 100644 --- a/apps/stage-web/src/pages/settings/characters/components/CharacterDialog.vue +++ b/apps/stage-web/src/pages/settings/characters/components/CharacterDialog.vue @@ -1,6 +1,7 @@