nanakura/moeka-project
1
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat(server): add product analytics events (#1941)

Browse Source
This commit is contained in:
RainbowBird
2026-06-03 23:03:33 +08:00
committed by GitHub
parent 41e8cd7626
commit 7ac69db4ef
24 changed files with 3590 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/drizzle/0014_vengeful_blonde_phantom.sql
+17
View File
@@ -0,0 +1,17 @@
CREATE TABLE "product_events" (
"id" text PRIMARY KEY NOT NULL,
"user_id" text NOT NULL,
"feature" text NOT NULL,
"action" text NOT NULL,
"status" text NOT NULL,
"source" text,
"model" text,
"provider" text,
"reason" text,
"metadata" jsonb,
"created_at" timestamp DEFAULT now() NOT NULL
);
--> statement-breakpoint
CREATE INDEX "product_events_feature_action_created_at_idx" ON "product_events" USING btree ("feature","action","created_at");--> statement-breakpoint
CREATE INDEX "product_events_user_id_created_at_idx" ON "product_events" USING btree ("user_id","created_at");--> statement-breakpoint
CREATE INDEX "product_events_created_at_idx" ON "product_events" USING btree ("created_at");
apps/server/drizzle/meta/0014_snapshot.json
+2936
View File
File diff suppressed because it is too large Load Diff
apps/server/drizzle/meta/_journal.json
+7
View File
@@ -99,6 +99,13 @@
"when": 1779728850192,
"tag": "0013_naive_groot",
"breakpoints": true
},
{
"idx": 14,
"version": "7",
"when": 1780498188307,
"tag": "0014_vengeful_blonde_phantom",
"breakpoints": true
}
]
}
apps/server/src/app.test.ts
+4
View File
@@ -51,6 +51,10 @@ function createTestDeps() {
adminUsersService: {} as any,
ttsMeter: {} as any,
requestLogService: {} as any,
productEventService: {
track: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
},
configKV: {
getOrThrow: vi.fn(async (key: string) => {
switch (key) {
apps/server/src/app.ts
+18 -5
View File
@@ -16,6 +16,7 @@ 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 { ProductEventService } from './services/domain/product-events'
import type { ProviderService } from './services/domain/providers'
import type { RequestLogService } from './services/domain/request-log'
import type { StripeService } from './services/domain/stripe'
@@ -75,6 +76,7 @@ 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 { createProductEventService } from './services/domain/product-events'
import { createProviderService } from './services/domain/providers'
import { createRequestLogService } from './services/domain/request-log'
import { createStripeService } from './services/domain/stripe'
@@ -99,6 +101,7 @@ interface AppDeps {
adminUsersService: AdminUsersService
ttsMeter: FluxMeter
requestLogService: RequestLogService
productEventService: ProductEventService
configKV: ConfigKVService
envelopeCrypto: EnvelopeCrypto
redis: Redis
@@ -186,6 +189,7 @@ export async function buildApp(deps: AppDeps) {
fluxService: deps.fluxService,
ttsMeter: deps.ttsMeter,
requestLogService: deps.requestLogService,
productEventService: deps.productEventService,
})
app.get('/api/v1/audio/speech/ws', upgradeWebSocket(async (c) => {
const token = c.req.query('token')
@@ -221,11 +225,13 @@ export async function buildApp(deps: AppDeps) {
billingService: deps.billingService,
configKV: deps.configKV,
requestLogService: deps.requestLogService,
productEventService: deps.productEventService,
ttsMeter: deps.ttsMeter,
llmRouter: deps.llmRouter,
genAi: deps.otel?.genAi,
revenue: deps.otel?.revenue,
rateLimitMetrics: deps.otel?.rateLimit,
posthog: deps.posthog,
})
const builtApp = app
@@ -351,7 +357,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, deps.posthog))
.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, deps.productEventService))
/**
* Admin routes — guarded by the `adminGuard` role check (`role === 'admin'`,
@@ -519,6 +525,11 @@ export async function createApp() {
},
})
const productEventService = injeca.provide('services:productEvents', {
dependsOn: { db, otel },
build: ({ dependsOn }) => createProductEventService(dependsOn.db, dependsOn.otel?.product),
})
const characterService = injeca.provide('services:characters', {
dependsOn: { db, otel },
build: ({ dependsOn }) => createCharacterService(dependsOn.db, dependsOn.otel?.engagement),
@@ -530,8 +541,8 @@ export async function createApp() {
})
const chatService = injeca.provide('services:chats', {
dependsOn: { db, otel },
build: ({ dependsOn }) => createChatService(dependsOn.db, dependsOn.otel?.engagement),
dependsOn: { db, otel, productEventService },
build: ({ dependsOn }) => createChatService(dependsOn.db, dependsOn.otel?.engagement, dependsOn.productEventService),
})
const stripeService = injeca.provide('services:stripe', {
@@ -579,7 +590,7 @@ export async function createApp() {
})
const auth = injeca.provide('services:auth', {
dependsOn: { db, env: parsedEnv, otel, email: emailService, userDeletionService, posthog },
dependsOn: { db, env: parsedEnv, otel, email: emailService, userDeletionService, posthog, productEventService },
build: async ({ dependsOn }) => {
// Seed trusted OIDC clients into DB so FK constraints on oauth_access_token are satisfied
await seedTrustedClients(dependsOn.db, dependsOn.env)
@@ -592,7 +603,7 @@ export async function createApp() {
redirectUris: client.redirectUris.join(', '),
}).log('OIDC trusted client ready')
}
return createAuth(dependsOn.db, dependsOn.env, dependsOn.email, dependsOn.otel?.auth, dependsOn.userDeletionService, dependsOn.posthog)
return createAuth(dependsOn.db, dependsOn.env, dependsOn.email, dependsOn.otel?.auth, dependsOn.userDeletionService, dependsOn.posthog, dependsOn.productEventService)
},
})
@@ -688,6 +699,7 @@ export async function createApp() {
fluxService,
fluxTransactionService,
requestLogService,
productEventService,
stripeService,
billingService,
adminFluxGrantsService,
@@ -735,6 +747,7 @@ export async function createApp() {
adminUsersService: resolved.adminUsersService,
ttsMeter: resolved.ttsMeter,
requestLogService: resolved.requestLogService,
productEventService: resolved.productEventService,
configKV: resolved.configKV,
envelopeCrypto: resolved.envelopeCrypto,
redis: resolved.redis,
apps/server/src/libs/auth.ts
+16
View File
@@ -2,6 +2,7 @@ import type { PostHog } from 'posthog-node'
import type { AuthMetrics } from '../otel'
import type { EmailService } from '../services/adapters/email'
import type { ProductEventService } from '../services/domain/product-events'
import type { UserDeletionService } from '../services/domain/user-deletion'
import type { Database } from './db'
import type { Env } from './env'
@@ -365,6 +366,7 @@ export function createAuth(
metrics?: AuthMetrics | null,
userDeletionService?: UserDeletionService,
posthog?: PostHog | null,
productEventService?: ProductEventService,
) {
return betterAuth({
secret: env.BETTER_AUTH_SECRET,
@@ -663,6 +665,13 @@ export function createAuth(
create: {
after: async (user) => {
metrics?.userRegistered.add(1)
await productEventService?.track({
userId: user.id,
feature: 'auth',
action: 'user_signed_up',
status: 'succeeded',
source: 'better-auth.user.create',
})
await captureSafe(posthog ?? null, {
event: 'user_signed_up',
distinctId: user.id,
@@ -701,6 +710,13 @@ export function createAuth(
.update(authSchema.user)
.set({ lastSeenAt: new Date() })
.where(eq(authSchema.user.id, session.userId))
await productEventService?.track({
userId: session.userId,
feature: 'auth',
action: 'session_started',
status: 'succeeded',
source: 'better-auth.session.create',
})
await captureSafe(posthog ?? null, {
event: 'session_started',
distinctId: session.userId,
apps/server/src/otel/index.ts
+25 -1
View File
@@ -30,6 +30,7 @@ import {
METRIC_AIRI_GEN_AI_GATEWAY_UPSTREAM_ERRORS,
METRIC_AIRI_GEN_AI_STREAM_INTERRUPTED,
METRIC_AIRI_OBSERVABILITY_READ_ERRORS,
METRIC_AIRI_PRODUCT_EVENTS,
METRIC_AIRI_RATE_LIMIT_BLOCKED,
METRIC_AIRI_STRIPE_REVENUE,
METRIC_AIRI_TTS_CHARS,
@@ -299,6 +300,21 @@ export interface ObservabilityMetrics {
metricReadErrors: Counter
}
export interface ProductMetrics {
/**
* Low-cardinality product event counter.
*
* Use when:
* - Reporting feature/event volume in Prometheus and Grafana.
*
* Expects:
* - Labels stay bounded (`feature`, `action`, `status`, optional
* `source`). Never attach `user_id`, `session_id`, request ids, models
* with unbounded aliases, or free-form error messages here.
*/
events: Counter
}
export interface OtelInstance {
auth: AuthMetrics
engagement: EngagementMetrics
@@ -308,6 +324,7 @@ export interface OtelInstance {
email: EmailMetrics
rateLimit: RateLimitMetrics
observability: ObservabilityMetrics
product: ProductMetrics
}
/**
@@ -511,6 +528,12 @@ export function initOtel(env: Env): OtelInstance | null {
}),
}
const product: ProductMetrics = {
events: meter.createCounter(METRIC_AIRI_PRODUCT_EVENTS, {
description: 'Low-cardinality product event volume. Distinct users live in Postgres product_events, not Prometheus labels.',
}),
}
// NOTICE:
// OTel SDK only emits a Counter time series after .add() runs the first time.
// Without this priming step, low-traffic counters (auth_failures_total,
@@ -560,10 +583,11 @@ export function initOtel(env: Env): OtelInstance | null {
email.failures,
rateLimit.blocked,
observability.metricReadErrors,
product.events,
]
for (const counter of counters) counter.add(0)
return { auth, engagement, revenue, genAi, gateway, email, rateLimit, observability }
return { auth, engagement, revenue, genAi, gateway, email, rateLimit, observability, product }
}
const severityMap: Record<string, SeverityNumber> = {
apps/server/src/routes/audio-speech-ws/route.test.ts
+5 -1
View File
@@ -166,6 +166,10 @@ function makeFakeDeps(overrides: {
const requestLogService = {
logRequest: vi.fn(async () => undefined),
}
const productEventService = {
track: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
}
const configKV = {
getOptional: vi.fn(async (key: string) => {
if (key === 'UNSPEECH_UPSTREAM') {
@@ -185,7 +189,7 @@ function makeFakeDeps(overrides: {
decryptKey: vi.fn(() => Buffer.from(overrides.decryptedKey ?? 'mock-upstream-token', 'utf8')),
}
return { configKV, envelopeCrypto, fluxService, ttsMeter, requestLogService }
return { configKV, envelopeCrypto, fluxService, ttsMeter, requestLogService, productEventService }
}
/** Drives the WSEvents lifecycle as if a real client had connected. */
apps/server/src/routes/audio-speech-ws/session.ts
+48
View File
@@ -91,6 +91,15 @@ export function createSessionState(userId: string, opts: AudioSpeechWsHandlersOp
}
async function dialUpstream() {
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_requested',
status: 'started',
source: 'audio.speech.ws',
model: modelLabel,
})
let unspeech: Awaited<ReturnType<AudioSpeechWsHandlersOptions['configKV']['getOptional']>>
try {
unspeech = await opts.configKV.getOptional('UNSPEECH_UPSTREAM')
@@ -191,6 +200,18 @@ export function createSessionState(userId: string, opts: AudioSpeechWsHandlersOp
log.withError(err).withFields({ userId }).warn('upstream ws error')
span.recordException(err)
span.setStatus({ code: SpanStatusCode.ERROR, message: err.message })
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_failed',
status: 'failed',
source: 'audio.speech.ws',
model: modelLabel,
reason: 'upstream_error',
metadata: {
duration_ms: Date.now() - startedAt,
},
})
try {
clientWs?.send(JSON.stringify({
event: 'error',
@@ -383,6 +404,20 @@ export function createSessionState(userId: string, opts: AudioSpeechWsHandlersOp
log.withError(err).warn('failed to write request log for streaming tts')
}
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
source: 'audio.speech.ws',
model: modelLabel,
metadata: {
input_chars: units,
duration_ms: durationMs,
flux_consumed: fluxConsumed,
},
})
finalize()
}
@@ -405,6 +440,19 @@ export function createSessionState(userId: string, opts: AudioSpeechWsHandlersOp
if (closed)
return
span.setStatus({ code: SpanStatusCode.ERROR, message: reason })
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_failed',
status: 'failed',
source: 'audio.speech.ws',
model: modelLabel,
reason,
metadata: {
close_code: code,
duration_ms: Date.now() - startedAt,
},
})
if (clientWs) {
try {
clientWs.send(JSON.stringify({ event: 'error', code: reason, message: reason }))
apps/server/src/routes/audio-speech-ws/types.ts
+3
View File
@@ -1,6 +1,7 @@
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 { ProductEventService } from '../../services/domain/product-events'
import type { RequestLogService } from '../../services/domain/request-log'
import type { EnvelopeCrypto } from '../../utils/envelope-crypto'
@@ -18,4 +19,6 @@ export interface AudioSpeechWsHandlersOptions {
ttsMeter: FluxMeter
/** Persists request accounting after a stream finishes. */
requestLogService: RequestLogService
/** Writes first-party product analytics for distinct-user aggregation. */
productEventService: ProductEventService
}
apps/server/src/routes/openai/v1/operations/chat-completions/index.ts
+105
View File
@@ -55,6 +55,18 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple
stream,
messageCount: Array.isArray(body.messages) ? body.messages.length : undefined,
}).log('chat completion request')
void deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_requested',
status: 'started',
source: 'openai.chat.completions',
model: requestModel,
metadata: {
stream,
message_count: Array.isArray(body.messages) ? body.messages.length : null,
},
})
// Server-connection attrs come from the router (which knows the actual
// upstream baseURL it dispatched to) — it enriches the active span with
@@ -90,6 +102,20 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple
sessionId: input.sessionId,
}).fail('Router exhausted or unknown model')
telemetry.recordMetrics({ model: requestModel, status: 502, type: 'chat', provider: routeCtx.provider, durationMs: Date.now() - startedAt, fluxConsumed: 0 })
void deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_failed',
status: 'failed',
source: 'openai.chat.completions',
model: requestModel,
provider: routeCtx.provider,
reason: 'router_exhausted',
metadata: {
duration_ms: Date.now() - startedAt,
stream,
},
})
throw err
}
@@ -115,6 +141,21 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple
telemetry.failSpan(span, `Gateway ${response.status}`)
generationTrace.fail(`Gateway ${response.status}`)
telemetry.recordMetrics({ model: requestModel, status: response.status, type: 'chat', provider: routeCtx.provider, durationMs, fluxConsumed: 0 })
void deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_failed',
status: 'failed',
source: 'openai.chat.completions',
model: requestModel,
provider: routeCtx.provider,
reason: 'upstream_error',
metadata: {
http_status: response.status,
duration_ms: durationMs,
stream,
},
})
// Emit server-side so funnels see real HTTP status — the client only
// ever observes "stream closed" and cannot tell 401 / 429 / 5xx apart.
void captureSafe(deps.posthog ?? null, {
@@ -254,6 +295,21 @@ function streamChatCompletion(input: {
input.telemetry.endSpan(input.span)
input.generationTrace.fail('Gateway stream interrupted')
input.telemetry.recordMetrics({ model: input.requestModel, status: input.response.status, type: 'chat', provider: input.routeCtxProvider, durationMs: input.durationMs, fluxConsumed: 0 })
void input.deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_failed',
status: 'failed',
source: 'openai.chat.completions',
model: input.requestModel,
provider: input.routeCtxProvider,
reason: 'stream_interrupted',
metadata: {
http_status: input.response.status,
duration_ms: input.durationMs,
stream: true,
},
})
}
else if (streamCompleted) {
try {
@@ -326,6 +382,23 @@ function streamChatCompletion(input: {
promptTokens: usage.promptTokens,
completionTokens: usage.completionTokens,
})
void input.deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_succeeded',
status: 'succeeded',
source: 'openai.chat.completions',
model: input.requestModel,
provider: input.routeCtxProvider,
metadata: {
http_status: input.response.status,
duration_ms: input.durationMs,
prompt_tokens: usage.promptTokens ?? 0,
completion_tokens: usage.completionTokens ?? 0,
flux_consumed: actualCharged,
stream: true,
},
})
void captureSafe(input.deps.posthog ?? null, {
distinctId: input.userId,
@@ -390,6 +463,21 @@ async function completeNonStreamingChat(input: {
input.telemetry.failSpan(input.span, 'Failed to parse upstream response body')
input.generationTrace.fail('Failed to parse upstream response body')
input.telemetry.recordMetrics({ model: input.requestModel, status: input.response.status, type: 'chat', provider: input.routeCtxProvider, durationMs: input.durationMs, fluxConsumed: 0 })
void input.deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_failed',
status: 'failed',
source: 'openai.chat.completions',
model: input.requestModel,
provider: input.routeCtxProvider,
reason: 'malformed_upstream_response',
metadata: {
http_status: input.response.status,
duration_ms: input.durationMs,
stream: false,
},
})
throw err
}
const usage = extractUsageFromBody(responseBody)
@@ -428,6 +516,23 @@ async function completeNonStreamingChat(input: {
promptTokens: usage.promptTokens,
completionTokens: usage.completionTokens,
})
void input.deps.productEventService.track({
userId: input.userId,
feature: 'gen_ai_chat',
action: 'completion_succeeded',
status: 'succeeded',
source: 'openai.chat.completions',
model: input.requestModel,
provider: input.routeCtxProvider,
metadata: {
http_status: input.response.status,
duration_ms: input.durationMs,
prompt_tokens: usage.promptTokens ?? 0,
completion_tokens: usage.completionTokens ?? 0,
flux_consumed: actualCharged,
stream: false,
},
})
void captureSafe(input.deps.posthog ?? null, {
distinctId: input.userId,
apps/server/src/routes/openai/v1/operations/speech-generation/index.ts
+1
View File
@@ -17,6 +17,7 @@ export function speechGeneration(deps: V1RouteDeps): GatewayCallback<'speech.gen
genAi: deps.genAi,
llmRouter: deps.llmRouter,
llmTracing: deps.llmTracing,
productEventService: deps.productEventService,
requestLogService: deps.requestLogService,
ttsMeter: deps.ttsMeter,
})
apps/server/src/routes/openai/v1/route.test.ts
+4
View File
@@ -142,6 +142,10 @@ function createTestApp(
billingService: billingService ?? createMockBillingService(),
configKV,
requestLogService: requestLogService ?? createMockRequestLogService(),
productEventService: {
track: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
},
ttsMeter: ttsMeter ?? createMockTtsMeter(),
llmRouter: llmRouter ?? createMockLlmRouter(),
genAi: null,
apps/server/src/routes/openai/v1/types.ts
+2
View File
@@ -7,6 +7,7 @@ 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 { ChatGenerationTrace, TtsGenerationTrace } from '../../../services/domain/llm-tracing'
import type { ProductEventService } from '../../../services/domain/product-events'
import type { RequestLogService } from '../../../services/domain/request-log'
import { startChatGeneration, startTtsGeneration } from '../../../services/domain/llm-tracing'
@@ -21,6 +22,7 @@ export interface V1RouteDeps {
billingService: BillingService
configKV: ConfigKVService
requestLogService: RequestLogService
productEventService: ProductEventService
ttsMeter: FluxMeter
llmRouter: LlmRouterService
genAi?: GenAiMetrics | null
apps/server/src/routes/stripe/index.ts
+4 -1
View File
@@ -6,6 +6,7 @@ import type { RateLimitMetrics, RevenueMetrics } from '../../otel'
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 { ProductEventService } from '../../services/domain/product-events'
import type { StripeService } from '../../services/domain/stripe'
import type { HonoEnv } from '../../types/hono'
@@ -47,10 +48,11 @@ export function createStripeRoutes(
metrics?: RevenueMetrics | null,
rateLimitMetrics?: RateLimitMetrics | null,
posthog?: PostHog | null,
productEventService?: ProductEventService,
) {
const stripe = env.STRIPE_SECRET_KEY ? new Stripe(env.STRIPE_SECRET_KEY) : null
const priceCatalog = stripe ? createStripePriceCatalog(stripe, redis) : null
const checkout = createCheckoutOperation({ stripe, priceCatalog, stripeService, configKV, env, metrics })
const checkout = createCheckoutOperation({ stripe, priceCatalog, stripeService, configKV, env, metrics, productEventService })
const webhook = createWebhookOperation({
stripe,
webhookSecret: env.STRIPE_WEBHOOK_SECRET,
@@ -59,6 +61,7 @@ export function createStripeRoutes(
billingService,
metrics,
posthog,
productEventService,
})
return new Hono<HonoEnv>()
apps/server/src/routes/stripe/operations/checkout.ts
+14
View File
@@ -3,6 +3,7 @@ import type Stripe from 'stripe'
import type { Env } from '../../../libs/env'
import type { RevenueMetrics } from '../../../otel'
import type { ConfigKVService } from '../../../services/adapters/config-kv'
import type { ProductEventService } from '../../../services/domain/product-events'
import type { StripeService } from '../../../services/domain/stripe'
import type { HonoEnv } from '../../../types/hono'
import type { StripePriceCatalog } from '../price-catalog'
@@ -23,6 +24,7 @@ export interface CheckoutOperationDeps {
configKV: ConfigKVService
env: Env
metrics?: RevenueMetrics | null
productEventService?: ProductEventService
}
export interface CheckoutOperationInput {
@@ -121,6 +123,18 @@ export function createCheckoutOperation(deps: CheckoutOperationDeps) {
})
deps.metrics?.stripeCheckoutCreated.add(1)
void deps.productEventService?.track({
userId: input.user.id,
feature: 'billing',
action: 'checkout_started',
status: 'succeeded',
source: 'stripe.checkout',
metadata: {
flux_amount: fluxAmount,
amount_total: session.amount_total,
currency: session.currency,
},
})
return { url: session.url }
}
apps/server/src/routes/stripe/operations/webhook.ts
+20 -1
View File
@@ -4,6 +4,7 @@ import type Stripe from 'stripe'
import type { RevenueMetrics } from '../../../otel'
import type { BillingService } from '../../../services/domain/billing/billing-service'
import type { FluxService } from '../../../services/domain/flux'
import type { ProductEventService } from '../../../services/domain/product-events'
import type { StripeService } from '../../../services/domain/stripe'
import { useLogger } from '@guiiai/logg'
@@ -22,6 +23,7 @@ export interface WebhookOperationDeps {
billingService: BillingService
metrics?: RevenueMetrics | null
posthog?: PostHog | null
productEventService?: ProductEventService
}
export interface WebhookOperationInput {
@@ -80,8 +82,25 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
// 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)
if (result.processed) {
const userId = event.data.object.metadata?.userId
if (userId) {
const fluxAmount = Number(event.data.object.metadata?.fluxAmount)
void deps.productEventService?.track({
userId,
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
source: 'stripe.webhook',
metadata: {
amount_total: event.data.object.amount_total,
currency: event.data.object.currency,
flux_amount: Number.isFinite(fluxAmount) ? fluxAmount : null,
},
})
}
await capturePaymentCompleted(deps.posthog, event.data.object)
}
break
}
case 'customer.created':
apps/server/src/schemas/index.ts
+1
View File
@@ -4,6 +4,7 @@ export * from './chats'
export * from './flux'
export * from './flux-transaction'
export * from './llm-request-log'
export * from './product-events'
export * from './providers'
export * from './stripe'
export * from './user-character'
apps/server/src/schemas/product-events.ts
+32
View File
@@ -0,0 +1,32 @@
import type { InferInsertModel, InferSelectModel } from 'drizzle-orm'
import { index, jsonb, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
import { nanoid } from '../utils/id'
export type ProductEventMetadata = Record<string, string | number | boolean | null>
export const productEvents = pgTable(
'product_events',
{
id: text('id').primaryKey().$defaultFn(() => nanoid()),
userId: text('user_id').notNull(),
feature: text('feature').notNull(),
action: text('action').notNull(),
status: text('status').notNull(),
source: text('source'),
model: text('model'),
provider: text('provider'),
reason: text('reason'),
metadata: jsonb('metadata').$type<ProductEventMetadata>(),
createdAt: timestamp('created_at').defaultNow().notNull(),
},
table => [
index('product_events_feature_action_created_at_idx').on(table.feature, table.action, table.createdAt),
index('product_events_user_id_created_at_idx').on(table.userId, table.createdAt),
index('product_events_created_at_idx').on(table.createdAt),
],
)
export type ProductEvent = InferSelectModel<typeof productEvents>
export type NewProductEvent = InferInsertModel<typeof productEvents>
apps/server/src/services/domain/chats.ts
+13 -1
View File
@@ -2,6 +2,7 @@ import type { MessageRole, WireMessage } from '@proj-airi/server-sdk-shared'
import type { Database } from '../../libs/db'
import type { EngagementMetrics } from '../../otel'
import type { ProductEventService } from './product-events'
import { useLogger } from '@guiiai/logg'
import { and, eq, gt, inArray, isNull, sql } from 'drizzle-orm'
@@ -49,7 +50,7 @@ export function resolveSenderId(role: string, userId: string, characterId?: stri
// Service factory
// ---------------------------------------------------------------------------
export function createChatService(db: Database, metrics?: EngagementMetrics | null) {
export function createChatService(db: Database, metrics?: EngagementMetrics | null, productEventService?: ProductEventService) {
// ---- internal helpers ---------------------------------------------------
async function verifyMembership(tx: Parameters<Parameters<Database['transaction']>[0]>[0], chatId: string, userId: string) {
@@ -299,6 +300,17 @@ export function createChatService(db: Database, metrics?: EngagementMetrics | nu
if (result.totalCount > 0) {
metrics?.chatMessages.add(result.totalCount)
void productEventService?.track({
userId,
feature: 'chat',
action: 'message_pushed',
status: 'succeeded',
source: 'chat.ws.push_messages',
metadata: {
message_count: result.totalCount,
new_count: result.newCount,
},
})
}
metrics?.wsMessagesReceived.add(result.totalCount)
apps/server/src/services/domain/openai-speech/index.ts
+56
View File
@@ -4,6 +4,7 @@ import type { FluxMeter } from '../billing/flux-meter'
import type { FluxService } from '../flux'
import type { LlmRouterService } from '../llm-router'
import type { startTtsGeneration, TtsGenerationTrace } from '../llm-tracing'
import type { ProductEventService } from '../product-events'
import type { RequestLogService } from '../request-log'
import { useLogger } from '@guiiai/logg'
@@ -32,6 +33,7 @@ export interface OpenAiSpeechServiceDeps {
requestLogService: RequestLogService
ttsMeter: FluxMeter
llmRouter: LlmRouterService
productEventService: ProductEventService
genAi?: GenAiMetrics | null
llmTracing: {
startTtsGeneration: (input: Parameters<typeof startTtsGeneration>[0]) => TtsGenerationTrace
@@ -78,6 +80,18 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
voice: typeof input.body.voice === 'string' ? input.body.voice : undefined,
}).log('tts speech request')
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_requested',
status: 'started',
source: 'audio.speech',
model: requestModel,
metadata: {
input_chars: inputText.length,
},
})
const flux = await deps.fluxService.getFlux(input.userId)
if (flux.flux <= 0)
throw createPaymentRequiredError('Insufficient flux')
@@ -127,6 +141,19 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
provider: routeCtx.provider,
status: 502,
})
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_failed',
status: 'failed',
source: 'audio.speech',
model: requestModel,
provider: routeCtx.provider,
reason: 'router_exhausted',
metadata: {
duration_ms: Date.now() - startedAt,
},
})
throw err
}
@@ -138,6 +165,20 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
span.end()
generationTrace.fail(`Gateway ${response.status}`)
recordMetrics({ model: requestModel, status: response.status, provider: routeCtx.provider, durationMs, fluxConsumed: 0 })
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_failed',
status: 'failed',
source: 'audio.speech',
model: requestModel,
provider: routeCtx.provider,
reason: 'upstream_error',
metadata: {
http_status: response.status,
duration_ms: durationMs,
},
})
logger.withFields({ requestId, userId: input.userId, model: requestModel, status: response.status, durationMs })
.warn('tts speech delivered with upstream error status')
return new Response(response.body, {
@@ -172,6 +213,21 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
}
recordMetrics({ model: requestModel, status: response.status, provider: routeCtx.provider, durationMs, fluxConsumed })
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
source: 'audio.speech',
model: requestModel,
provider: routeCtx.provider,
metadata: {
http_status: response.status,
input_chars: inputText.length,
duration_ms: durationMs,
flux_consumed: fluxConsumed,
},
})
deps.requestLogService.logRequest({
userId: input.userId,
model: requestModel,
apps/server/src/services/domain/product-events.test.ts
+102
View File
@@ -0,0 +1,102 @@
import type { Database } from '../../libs/db'
import type { ProductMetrics } from '../../otel'
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
import { mockDB } from '../../libs/mock-db'
import { createProductEventService } from './product-events'
import * as schema from '../../schemas'
describe('productEventService', () => {
let db: Database
beforeAll(async () => {
db = await mockDB(schema)
})
beforeEach(async () => {
await db.delete(schema.productEvents)
})
it('writes first-party events and increments only low-cardinality metric labels', async () => {
const events = { add: vi.fn() }
const service = createProductEventService(db, { events } as unknown as ProductMetrics)
await service.track({
userId: 'user-1',
feature: 'gen_ai_chat',
action: 'completion_succeeded',
status: 'succeeded',
source: 'openai.chat.completions',
model: 'openrouter/anthropic/claude-sonnet-4',
provider: 'openrouter',
metadata: {
stream: false,
flux_consumed: 3,
},
})
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(1)
expect(rows[0]).toMatchObject({
userId: 'user-1',
feature: 'gen_ai_chat',
action: 'completion_succeeded',
status: 'succeeded',
source: 'openai.chat.completions',
model: 'openrouter/anthropic/claude-sonnet-4',
provider: 'openrouter',
})
expect(events.add).toHaveBeenCalledWith(1, {
feature: 'gen_ai_chat',
action: 'completion_succeeded',
status: 'succeeded',
source: 'openai.chat.completions',
})
})
it('aggregates event volume and distinct users by feature/action/status', async () => {
const service = createProductEventService(db)
const createdAt = new Date('2026-06-03T00:00:00.000Z')
await service.track({
userId: 'user-1',
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
source: 'audio.speech',
createdAt,
})
await service.track({
userId: 'user-1',
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
source: 'audio.speech.ws',
createdAt,
})
await service.track({
userId: 'user-2',
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
source: 'audio.speech',
createdAt,
})
const rows = await service.countDistinctUsersByFeature({
from: new Date('2026-06-02T00:00:00.000Z'),
to: new Date('2026-06-04T00:00:00.000Z'),
})
expect(rows).toEqual([{
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
eventCount: 3,
distinctUsers: 2,
}])
})
})
apps/server/src/services/domain/product-events.ts
+152
View File
@@ -0,0 +1,152 @@
import type { Database } from '../../libs/db'
import type { ProductMetrics } from '../../otel'
import type { ProductEventMetadata } from '../../schemas/product-events'
import { useLogger } from '@guiiai/logg'
import { and, asc, count, gte, lt, sql } from 'drizzle-orm'
import * as schema from '../../schemas/product-events'
const logger = useLogger('product-events')
export type ProductFeature = 'auth' | 'chat' | 'gen_ai_chat' | 'tts' | 'billing'
export type ProductEventStatus = 'started' | 'succeeded' | 'failed'
export type ProductAction
= | 'user_signed_up'
| 'session_started'
| 'message_pushed'
| 'completion_requested'
| 'completion_succeeded'
| 'completion_failed'
| 'speech_requested'
| 'speech_succeeded'
| 'speech_failed'
| 'checkout_started'
| 'payment_completed'
/**
* Product event fact written to AIRI's own Postgres analytics table.
*/
export interface ProductEventInput {
/** Better Auth user id. Kept in Postgres only; never emitted as a Prometheus label. */
userId: string
/** Bounded product area used for product dashboards and funnels. */
feature: ProductFeature
/** Bounded user/business action within the feature. */
action: ProductAction
/** Lifecycle state for the action. */
status: ProductEventStatus
/** Optional bounded route/surface label such as `openai.chat.completions`. */
source?: string
/** Optional model alias for DB-side drilldown. Do not expose as a Prometheus label. */
model?: string
/** Optional provider name for DB-side drilldown. */
provider?: string
/** Optional bounded failure reason or business outcome. */
reason?: string
/** Optional primitive metadata for product analysis. Avoid PII and raw prompts. */
metadata?: ProductEventMetadata
/** Override for tests/backfills. Defaults to database/server current time. */
createdAt?: Date
}
export interface ProductEventAggregateInput {
/** Inclusive lower time bound. */
from: Date
/** Exclusive upper time bound. Omit for open-ended queries. */
to?: Date
}
export interface ProductEventAggregateRow {
feature: string
action: string
status: string
eventCount: number
distinctUsers: number
}
/**
* Creates AIRI's first-party product analytics event writer.
*
* Use when:
* - Server-side product behavior has a user id and should be queryable by
* distinct users, funnels, or retention windows.
* - Grafana needs low-cardinality event volume while Postgres keeps user-level
* detail.
*
* Expects:
* - Callers pass only bounded `feature` / `action` / `status` values.
* - PII, prompts, request ids, sessions, and user ids are not written into
* Prometheus labels. User id is stored only in the DB row.
*
* Returns:
* - Best-effort event writer plus a DB aggregation helper for analytics jobs.
*/
export function createProductEventService(db: Database, metrics?: ProductMetrics | null) {
return {
async track(input: ProductEventInput): Promise<void> {
try {
await db.insert(schema.productEvents).values({
userId: input.userId,
feature: input.feature,
action: input.action,
status: input.status,
source: input.source,
model: input.model,
provider: input.provider,
reason: input.reason,
metadata: input.metadata,
createdAt: input.createdAt,
})
const attrs: Record<string, string> = {
feature: input.feature,
action: input.action,
status: input.status,
}
if (input.source)
attrs.source = input.source
metrics?.events.add(1, attrs)
}
catch (err) {
logger.withError(err).withFields({
userId: input.userId,
feature: input.feature,
action: input.action,
status: input.status,
}).warn('Failed to write product event; swallowing to protect caller')
}
},
async countDistinctUsersByFeature(input: ProductEventAggregateInput): Promise<ProductEventAggregateRow[]> {
const where = input.to
? and(gte(schema.productEvents.createdAt, input.from), lt(schema.productEvents.createdAt, input.to))
: gte(schema.productEvents.createdAt, input.from)
const rows = await db
.select({
feature: schema.productEvents.feature,
action: schema.productEvents.action,
status: schema.productEvents.status,
eventCount: count(),
distinctUsers: sql<number>`count(distinct ${schema.productEvents.userId})::int`,
})
.from(schema.productEvents)
.where(where)
.groupBy(schema.productEvents.feature, schema.productEvents.action, schema.productEvents.status)
.orderBy(asc(schema.productEvents.feature), asc(schema.productEvents.action), asc(schema.productEvents.status))
return rows.map(row => ({
feature: row.feature,
action: row.action,
status: row.status,
eventCount: Number(row.eventCount),
distinctUsers: Number(row.distinctUsers),
}))
},
}
}
export type ProductEventService = ReturnType<typeof createProductEventService>
apps/server/src/utils/observability.ts
+5
View File
@@ -97,6 +97,11 @@ export const METRIC_AIRI_TTS_PREFLIGHT_REJECTIONS = 'airi.billing.tts.preflight_
// AIRI observability — self-monitoring for the metric pipeline
export const METRIC_AIRI_OBSERVABILITY_READ_ERRORS = 'airi.observability.read_errors'
// Product analytics — low-cardinality event volume only. User-level product
// analytics live in Postgres `product_events`; never add user identifiers to
// this metric's labels.
export const METRIC_AIRI_PRODUCT_EVENTS = 'airi.product.events'
// AIRI revenue — actual money in (smallest currency unit, e.g. cents)
export const METRIC_AIRI_STRIPE_REVENUE = 'airi.stripe.revenue'