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

fix(analytics): remove TS product events table (#2306)

Browse Source
This commit is contained in:
RainbowBird
2026-08-17 10:16:44 +00:00
committed by GitHub
parent c4778686ff
commit 07b26b530f
28 changed files with 3791 additions and 1098 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stage-ui/src/composables/use-analytics.ts
+3 -5
View File
@@ -351,11 +351,9 @@ export function useAnalytics() {
}
// ─── LLM round events (client-known fields only) ──────────────────────
// Source-of-truth for HTTP status / token usage / billing stage is the
// server, which records them as
// Postgres `product_events` rows — deliberately NOT forwarded to PostHog
// (per-request volume stays in DB/Grafana). These client emits supply the
// user-facing latency picture (TTFT, render time) the server cannot see.
// The server owns HTTP status, token usage, and billing state. It records
// these request-level facts in operational telemetry, not product analytics.
// These client events supply latency data the server cannot observe.
function trackMessageSendStarted(properties: ChatRoundCorrelationProperties & { source: 'text' | 'voice', model?: string }) {
if (!canCapture())
server/apps/api/drizzle/0022_worthless_shriek.sql
+1
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS "product_events";
server/apps/api/drizzle/meta/0022_snapshot.json
+3541
View File
File diff suppressed because it is too large Load Diff
server/apps/api/drizzle/meta/_journal.json
+7
View File
@@ -155,6 +155,13 @@
"when": 1786864179375,
"tag": "0021_chilly_starjammers",
"breakpoints": true
},
{
"idx": 22,
"version": "7",
"when": 1786957404391,
"tag": "0022_worthless_shriek",
"breakpoints": true
}
]
}
server/apps/api/src/app.test.ts
-1
View File
@@ -29,7 +29,6 @@ function createTestDeps() {
productEventService: {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
} as never,
configKV: { getOrThrow: vi.fn() } as never,
redis: redis as never,
server/apps/api/src/app.ts
+5 -6
View File
@@ -175,7 +175,6 @@ 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')
@@ -389,7 +388,7 @@ function parseTtsSource(
}
/**
* Normalizes the client-provided streaming TTS voice bucket for product events.
* Normalizes the client-provided streaming TTS voice bucket for request telemetry.
*/
function parseTtsVoiceType(
value: string | undefined,
@@ -507,8 +506,8 @@ export async function createApp() {
})
const productEventService = injeca.provide('services:productEvents', {
dependsOn: { db, otel, posthogSink },
build: ({ dependsOn }) => createProductEventService(dependsOn.db, dependsOn.otel?.product, dependsOn.posthogSink),
dependsOn: { posthogSink },
build: ({ dependsOn }) => createProductEventService(dependsOn.posthogSink),
})
const characterService = injeca.provide('services:characters', {
@@ -522,8 +521,8 @@ export async function createApp() {
})
const chatService = injeca.provide('services:chats', {
dependsOn: { db, otel, productEventService },
build: ({ dependsOn }) => createChatService(dependsOn.db, dependsOn.otel?.engagement, dependsOn.productEventService),
dependsOn: { db, otel },
build: ({ dependsOn }) => createChatService(dependsOn.db, dependsOn.otel?.engagement),
})
const stripeService = injeca.provide('services:stripe', {
server/apps/api/src/libs/env.ts
+2 -3
View File
@@ -114,11 +114,10 @@ const EnvSchema = object({
DB_POOL_CONNECTION_TIMEOUT_MS: optionalIntegerFromString(5000, 'DB_POOL_CONNECTION_TIMEOUT_MS', 1),
DB_POOL_KEEPALIVE_INITIAL_DELAY_MS: optionalIntegerFromString(10000, 'DB_POOL_KEEPALIVE_INITIAL_DELAY_MS', 1),
// PostHog product-event forwarding (signup / payment / subscription facts).
// PostHog product-event forwarding for server-confirmed funnel facts.
// Defaults to the shared AIRI project key (same browser-safe phc_* key the
// client surfaces embed in posthog.config.ts), so forwarding is on out of
// the box. Set to an empty string to disable; Postgres `product_events`
// stays the source of truth either way.
// the box. Set to an empty string to disable server-side product analytics.
POSTHOG_PROJECT_KEY: optional(string(), 'phc_pzjziJjrVZpa9SqnQqq0QEKvkmuCPH7GDTA6TbRTEf9'), // cspell:disable-line
POSTHOG_API_HOST: optional(string(), 'https://t.airi.build'),
server/apps/api/src/otel/index.ts
+1 -25
View File
@@ -32,7 +32,6 @@ 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,
@@ -256,21 +255,6 @@ 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
@@ -280,7 +264,6 @@ export interface OtelInstance {
email: EmailMetrics
rateLimit: RateLimitMetrics
observability: ObservabilityMetrics
product: ProductMetrics
}
/**
@@ -481,12 +464,6 @@ 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,
@@ -535,11 +512,10 @@ 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, product }
return { auth, engagement, revenue, genAi, gateway, email, rateLimit, observability }
}
const severityMap: Record<string, SeverityNumber> = {
server/apps/api/src/routes/audio-speech-ws/route.test.ts
+1 -31
View File
@@ -184,11 +184,6 @@ function makeFakeDeps(overrides: {
const requestLogService = {
logRequest: vi.fn(async () => undefined),
}
const productEventService = {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
}
const configKV = {
getOptional: vi.fn(async (key: string) => {
if (key === 'UNSPEECH_UPSTREAM') {
@@ -212,7 +207,7 @@ function makeFakeDeps(overrides: {
decryptKey: vi.fn(() => Buffer.from(overrides.decryptedKey ?? 'mock-upstream-token', 'utf8')),
}
return { configKV, envelopeCrypto, fluxService, ttsMeter, requestLogService, productEventService }
return { configKV, envelopeCrypto, fluxService, ttsMeter, requestLogService }
}
/** Drives the WSEvents lifecycle as if a real client had connected. */
@@ -299,17 +294,6 @@ describe('audio-speech-ws route', () => {
status: 200,
fluxConsumed: 1,
})
expect(deps.productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
userId: 'user-123',
feature: 'tts',
action: 'speech_succeeded',
status: 'succeeded',
model: 'volcengine/seed-tts-2.0',
metadata: expect.objectContaining({
voice_id: 'mock',
voice_type: 'official_selected',
}),
}))
})
it('refuses the session with insufficient_flux when the user is broke', async () => {
@@ -335,20 +319,6 @@ describe('audio-speech-ws route', () => {
})
expect(client.closed).toBe(true)
expect(client.closeCode).toBe(1008)
expect(deps.productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
userId: 'user-broke',
feature: 'tts',
action: 'speech_blocked',
status: 'blocked',
source: 'chat_auto_tts',
reason: 'insufficient_balance',
metadata: expect.objectContaining({
trigger: 'auto',
block_reason: 'insufficient_balance',
balance_state: 'insufficient',
flux_balance_bucket: 'zero',
}),
}))
})
it('refuses with streaming_tts_not_configured when UNSPEECH_UPSTREAM.streaming is empty', async () => {
server/apps/api/src/routes/audio-speech-ws/session.ts
+1 -136
View File
@@ -12,7 +12,6 @@ import { useLogger } from '@guiiai/logg'
import { context as otelContext, SpanStatusCode, trace } from '@opentelemetry/api'
import { ofetch } from 'ofetch'
import { fluxBalanceBucket } from '../../services/domain/flux-balance'
import { ApiError } from '../../utils/error'
import { nanoid } from '../../utils/id'
import {
@@ -79,11 +78,10 @@ export interface AudioSpeechSessionAnalytics {
export function createSessionState(
userId: string,
opts: AudioSpeechWsHandlersOptions,
analyticsInput: AudioSpeechSessionAnalytics = {},
_analyticsInput: AudioSpeechSessionAnalytics = {},
): AudioSpeechSessionState {
const requestId = nanoid()
const startedAt = Date.now()
const analytics = normalizeAnalytics(analyticsInput)
const span = tracer.startSpan('llm.gateway.tts.stream', {
attributes: {
[AIRI_ATTR_GEN_AI_OPERATION_KIND]: 'text_to_speech_stream',
@@ -99,9 +97,7 @@ export function createSessionState(
let startValidationStarted = false
let dialStarted = false
let totalInputChars = 0
let preflightFluxBalance: number | undefined
let modelLabel = STREAM_MODEL_LABEL_FALLBACK
let voiceLabel: string | undefined
/**
* Frames the client sent before the upstream finished dialing. Buffered to
* avoid silently dropping the `start` frame; flushed in arrival order once
@@ -118,19 +114,6 @@ export function createSessionState(
return
dialStarted = true
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_requested',
status: 'started',
source: analytics.source,
model: modelLabel,
metadata: {
trigger: analytics.trigger,
...streamingVoiceMetadata(voiceLabel, analytics.voiceType),
},
})
let unspeech: Awaited<ReturnType<AudioSpeechWsHandlersOptions['configKV']['getOptional']>>
try {
unspeech = await opts.configKV.getOptional('UNSPEECH_UPSTREAM')
@@ -151,7 +134,6 @@ export function createSessionState(
// afford the worst-case session.
try {
const flux = await opts.fluxService.getFlux(userId)
preflightFluxBalance = flux.flux
await opts.ttsMeter.assertCanAfford(userId, STREAMING_PREFLIGHT_CHARS_ESTIMATE, flux.flux)
}
catch (err) {
@@ -231,20 +213,6 @@ export function createSessionState(
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: analytics.source,
model: modelLabel,
reason: 'upstream_error',
metadata: {
duration_ms: Date.now() - startedAt,
trigger: analytics.trigger,
...streamingVoiceMetadata(voiceLabel, analytics.voiceType),
},
})
try {
clientWs?.send(JSON.stringify({
event: 'error',
@@ -284,7 +252,6 @@ export function createSessionState(
startValidationStarted = true
modelLabel = startFrame.model
voiceLabel = startFrame.voice
pendingClientFrames.push({ data: payload, isBinary })
void validateStartFrame(startFrame).then((accepted) => {
if (!accepted || closed)
@@ -410,9 +377,6 @@ export function createSessionState(
const model = (parsed as Record<string, unknown>).model
if (typeof model === 'string' && model.length > 0)
modelLabel = model
const voice = (parsed as Record<string, unknown>).voice
if (typeof voice === 'string' && voice.length > 0)
voiceLabel = voice
}
}
catch {
@@ -522,22 +486,6 @@ export function createSessionState(
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: analytics.source,
model: modelLabel,
metadata: {
input_chars: units,
duration_ms: durationMs,
flux_consumed: fluxConsumed,
trigger: analytics.trigger,
...streamingVoiceMetadata(voiceLabel, analytics.voiceType),
},
})
finalize()
}
@@ -560,21 +508,6 @@ export function createSessionState(
if (closed)
return
span.setStatus({ code: SpanStatusCode.ERROR, message: reason })
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_failed',
status: 'failed',
source: analytics.source,
model: modelLabel,
reason,
metadata: {
close_code: code,
duration_ms: Date.now() - startedAt,
trigger: analytics.trigger,
...streamingVoiceMetadata(voiceLabel, analytics.voiceType),
},
})
if (clientWs) {
try {
clientWs.send(JSON.stringify({ event: 'error', code: reason, message: reason }))
@@ -592,25 +525,6 @@ export function createSessionState(
function closeWithBlockedPreflight(code: number, reason: string) {
if (closed)
return
void opts.productEventService.track({
userId,
feature: 'tts',
action: 'speech_blocked',
status: 'blocked',
source: analytics.source,
model: modelLabel,
reason: 'insufficient_balance',
metadata: {
block_reason: 'insufficient_balance',
balance_state: 'insufficient',
flux_balance_bucket: fluxBalanceBucket(preflightFluxBalance),
billing_units: STREAMING_PREFLIGHT_CHARS_ESTIMATE,
close_code: code,
duration_ms: Date.now() - startedAt,
trigger: analytics.trigger,
...streamingVoiceMetadata(voiceLabel, analytics.voiceType),
},
})
if (clientWs) {
try {
clientWs.send(JSON.stringify({ event: 'error', code: reason, message: reason }))
@@ -633,55 +547,6 @@ export function createSessionState(
}
}
function normalizeAnalytics(input: AudioSpeechSessionAnalytics): Required<AudioSpeechSessionAnalytics> {
return {
trigger: normalizeTrigger(input.trigger),
source: normalizeSource(input.source),
voiceType: normalizeVoiceType(input.voiceType),
}
}
function normalizeTrigger(trigger: AudioSpeechSessionAnalytics['trigger']): StreamingTtsTrigger {
return trigger === 'auto' ? 'auto' : 'manual'
}
function normalizeSource(source: AudioSpeechSessionAnalytics['source']): StreamingTtsSource {
switch (source) {
case 'audio.speech.ws':
case 'chat_auto_tts':
case 'manual_preview':
case 'settings_test':
return source
default:
return 'audio.speech.ws'
}
}
/**
* Normalizes streaming TTS voice type into bounded analytics values.
*/
function normalizeVoiceType(voiceType: AudioSpeechSessionAnalytics['voiceType']): StreamingTtsVoiceType {
switch (voiceType) {
case 'official_default':
case 'official_selected':
case 'custom_configured':
case 'voice_pack':
return voiceType
default:
return 'unknown'
}
}
/**
* Builds reusable streaming TTS voice metadata after the start frame is known.
*/
function streamingVoiceMetadata(voiceId: string | undefined, voiceType: StreamingTtsVoiceType): Record<string, unknown> {
return {
...(voiceId ? { voice_id: voiceId } : {}),
voice_type: voiceType,
}
}
function isPaymentRequiredError(err: unknown): boolean {
if (err instanceof ApiError)
return err.statusCode === 402
server/apps/api/src/routes/audio-speech-ws/types.ts
-3
View File
@@ -1,7 +1,6 @@
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'
@@ -19,6 +18,4 @@ 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
}
server/apps/api/src/routes/internal-auth.ts
+2 -2
View File
@@ -12,8 +12,8 @@ const UserDeletionRequestSchema = object({
const AuthEventRequestSchema = object({
userId: pipe(string(), trim(), nonEmpty()),
action: picklist(['user_signed_up', 'session_started']),
source: picklist(['better-auth.user.create', 'better-auth.session.create']),
action: picklist(['user_signed_up']),
source: picklist(['better-auth.user.create']),
})
/**
server/apps/api/src/routes/openai/v1/operations/chat-completions/index.ts
-107
View File
@@ -71,19 +71,6 @@ 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
// its own `airi.gen_ai.gateway.*` attrs on success.
@@ -126,20 +113,6 @@ 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
}
@@ -165,21 +138,6 @@ 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,
},
})
logger.withFields({ requestId, userId: input.userId, model: requestModel, status: response.status, durationMs })
.warn('chat completion delivered with upstream error status')
@@ -433,21 +391,6 @@ 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 {
@@ -534,23 +477,6 @@ 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,
},
})
input.logger.withFields({
requestId: input.requestId,
@@ -604,21 +530,6 @@ 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)
@@ -671,24 +582,6 @@ 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,
},
})
input.logger.withFields({
requestId: input.requestId,
userId: input.userId,
server/apps/api/src/routes/openai/v1/operations/speech-generation/index.ts
-1
View File
@@ -18,7 +18,6 @@ export function speechGeneration(deps: V1RouteDeps): GatewayCallback<'speech.gen
llmRouter: deps.llmRouter,
llmTracing: deps.llmTracing,
providerCatalogService: deps.providerCatalogService,
productEventService: deps.productEventService,
requestLogService: deps.requestLogService,
ttsMeter: deps.ttsMeter,
voicePackService: deps.voicePackService,
server/apps/api/src/routes/openai/v1/route.test.ts
+16 -58
View File
@@ -149,7 +149,6 @@ function createMockProductEventService(): ProductEventService {
return {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
}
}
@@ -1467,13 +1466,14 @@ describe('v1CompletionsRoutes', () => {
* @example
* POST /api/v1/audio/speech { "voice": "alloy" }
*/
it('records TTS voice and Voice Pack metadata in product events', async () => {
globalThis.fetch = vi.fn(async () => new Response(new Uint8Array([1]), {
it('routes TTS requests with Voice Pack metadata', async () => {
const routeTts = vi.fn(async () => new Response(new Uint8Array([1]), {
status: 200,
headers: { 'Content-Type': 'audio/mpeg' },
}))
const productEventService = createMockProductEventService()
const llmRouter = createMockLlmRouter({ routeTts })
const voicePackService = createMockVoicePackService({
findEnabledByVoiceId: vi.fn(async () => ({
id: 'vp-premium',
@@ -1497,13 +1497,13 @@ describe('v1CompletionsRoutes', () => {
undefined,
undefined,
undefined,
undefined,
llmRouter,
createMockLlmTracing(),
productEventService,
voicePackService,
)
await app.fetch(
const response = await app.fetch(
new Request('http://localhost/api/v1/audio/speech', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
@@ -1522,23 +1522,15 @@ describe('v1CompletionsRoutes', () => {
{ user: testUser } as any,
)
expect(productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
action: 'speech_succeeded',
source: 'manual_preview',
metadata: expect.objectContaining({
voice_id: 'alloy',
voice_type: 'voice_pack',
voice_pack_id: 'vp-premium',
expect(response.status).toBe(200)
expect(routeTts).toHaveBeenCalledWith(expect.objectContaining({
modelName: 'tts-1',
input: expect.objectContaining({
text: 'hello',
voice: 'upstream-alloy',
}),
}))
expect(productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
action: 'speech_requested',
metadata: expect.objectContaining({
voice_id: 'alloy',
voice_type: 'voice_pack',
voice_pack_id: 'vp-premium',
}),
}))
}), expect.any(Object))
expect(productEventService.track).not.toHaveBeenCalled()
})
it('should not charge when routeTts upstream returns error', async () => {
@@ -1568,7 +1560,7 @@ describe('v1CompletionsRoutes', () => {
* @example
* routeTts throws ApiError(429, 'TOO_MANY_REQUESTS', 'Too many requests')
*/
it('records routeTts ApiError status and reason in product events', async () => {
it('preserves routeTts ApiError status and reason', async () => {
const productEventService = createMockProductEventService()
const llmRouter = createMockLlmRouter({
routeTts: vi.fn(async () => {
@@ -1596,19 +1588,9 @@ describe('v1CompletionsRoutes', () => {
)
expect(res.status).toBe(429)
expect(productEventService.track).toHaveBeenCalledWith(
expect.objectContaining({
action: 'speech_failed',
reason: 'TOO_MANY_REQUESTS',
metadata: expect.objectContaining({
failure_reason: 'TOO_MANY_REQUESTS',
http_status: 429,
}),
}),
)
})
it('returns 402 and records blocked event for manual TTS when flux is insufficient', async () => {
it('returns 402 for manual TTS when flux is insufficient', async () => {
const productEventService = createMockProductEventService()
const llmRouter = createMockLlmRouter()
const app = createTestApp(
@@ -1632,21 +1614,9 @@ describe('v1CompletionsRoutes', () => {
)
expect(res.status).toBe(402)
expect(llmRouter.routeTts).not.toHaveBeenCalled()
expect(productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
action: 'speech_blocked',
status: 'blocked',
source: 'audio.speech',
reason: 'insufficient_balance',
metadata: expect.objectContaining({
trigger: 'manual',
block_reason: 'insufficient_balance',
balance_state: 'insufficient',
flux_balance_bucket: 'zero',
}),
}))
})
it('returns 204 and records blocked event for auto TTS when flux is insufficient', async () => {
it('returns 204 for auto TTS when flux is insufficient', async () => {
const productEventService = createMockProductEventService()
const llmRouter = createMockLlmRouter()
const app = createTestApp(
@@ -1680,18 +1650,6 @@ describe('v1CompletionsRoutes', () => {
)
expect(res.status).toBe(204)
expect(llmRouter.routeTts).not.toHaveBeenCalled()
expect(productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
action: 'speech_blocked',
status: 'blocked',
source: 'chat_auto_tts',
reason: 'insufficient_balance',
metadata: expect.objectContaining({
trigger: 'auto',
block_reason: 'insufficient_balance',
balance_state: 'insufficient',
flux_balance_bucket: 'zero',
}),
}))
})
it('should not charge when input is empty', async () => {
server/apps/api/src/routes/stripe/operations/checkout.ts
+1
View File
@@ -136,6 +136,7 @@ export function createCheckoutOperation(deps: CheckoutOperationDeps) {
feature: 'billing',
action: 'checkout_started',
status: 'succeeded',
eventId: session.id,
source: 'stripe.checkout',
metadata: {
flux_amount: fluxAmount,
server/apps/api/src/routes/stripe/operations/webhook.ts
+4 -39
View File
@@ -3,7 +3,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 { ProductAction, ProductEventService } from '../../../services/domain/product-events'
import type { ProductEventService } from '../../../services/domain/product-events'
import type { StripeService } from '../../../services/domain/stripe'
import { useLogger } from '@guiiai/logg'
@@ -96,6 +96,7 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
eventId: event.data.object.id,
source: 'stripe.webhook',
metadata: {
amount_total: event.data.object.amount_total,
@@ -119,29 +120,15 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
case 'customer.subscription.created':
case 'customer.subscription.updated':
case 'customer.subscription.deleted': {
const result = await handleSubscriptionEvent(event.data.object, deps.stripeService)
await handleSubscriptionEvent(event.data.object, deps.stripeService)
deps.metrics?.stripeSubscriptionEvent.add(1, { event_type: event.type.replace('customer.subscription.', '') })
const action = subscriptionActionForWebhookEvent(event.type)
if (result && action) {
void deps.productEventService?.track({
userId: result.userId,
feature: 'billing',
action,
status: 'succeeded',
source: 'stripe.webhook',
metadata: {
stripe_price_id: result.stripePriceId ?? null,
stripe_subscription_status: result.subscriptionStatus ?? null,
},
})
}
break
}
case 'invoice.created':
case 'invoice.updated':
case 'invoice.paid':
case 'invoice.payment_failed': {
const result = await handleInvoiceEvent(event.data.object, deps.stripeService)
await handleInvoiceEvent(event.data.object, deps.stripeService)
if (event.type === 'invoice.payment_failed')
deps.metrics?.stripePaymentFailed.add(1)
if (event.type === 'invoice.paid' && event.data.object.amount_paid && event.data.object.currency) {
@@ -150,20 +137,6 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
source: 'invoice',
})
}
if (event.type === 'invoice.paid' && event.data.object.billing_reason === 'subscription_cycle' && result?.stripeSubscriptionId) {
void deps.productEventService?.track({
userId: result.userId,
feature: 'billing',
action: 'subscription_renewed',
status: 'succeeded',
source: 'stripe.webhook',
metadata: {
amount_paid: result.amountPaid ?? null,
currency: result.currency ?? null,
stripe_price_id: result.stripePriceId ?? null,
},
})
}
break
}
}
@@ -172,14 +145,6 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
}
}
function subscriptionActionForWebhookEvent(eventType: Stripe.Event.Type): ProductAction | null {
if (eventType === 'customer.subscription.created')
return 'subscription_started'
if (eventType === 'customer.subscription.deleted')
return 'subscription_cancelled'
return null
}
async function handleCheckoutSessionCompleted(
stripeEventId: string,
session: Stripe.Checkout.Session,
server/apps/api/src/routes/stripe/route.test.ts
+21 -24
View File
@@ -591,6 +591,7 @@ describe('stripeRoutes', () => {
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
eventId: 'cs_1',
source: 'stripe.webhook',
metadata: {
amount_total: 500,
@@ -604,7 +605,7 @@ describe('stripeRoutes', () => {
})
})
it('records subscription lifecycle product events from Stripe webhooks', async () => {
it('processes subscription lifecycle webhooks without product events', async () => {
const subscriptionEvent = {
id: 'evt_sub_created',
type: 'customer.subscription.created',
@@ -627,10 +628,10 @@ describe('stripeRoutes', () => {
},
},
}
const productEventService = { track: vi.fn() }
const stripeService = createMockStripeService({
getCustomerByStripeId: vi.fn(async () => createMockStripeCustomer()),
})
const productEventService = { track: vi.fn(async () => undefined) }
const webhook = createWebhookOperation({
stripe: {
webhooks: {
@@ -646,17 +647,15 @@ describe('stripeRoutes', () => {
await webhook({ signature: 'test_sig', body: '{}' })
expect(productEventService.track).toHaveBeenCalledWith({
expect(stripeService.upsertSubscription).toHaveBeenCalledWith(expect.objectContaining({
userId: 'user-1',
feature: 'billing',
action: 'subscription_started',
status: 'succeeded',
source: 'stripe.webhook',
metadata: {
stripe_price_id: 'price_1',
stripe_subscription_status: 'active',
},
})
stripeSubscriptionId: 'sub_1',
stripeCustomerId: 'cus_1',
stripePriceId: 'price_1',
status: 'active',
cancelAtPeriodEnd: false,
}))
expect(productEventService.track).not.toHaveBeenCalled()
})
it('records subscription renewals only for subscription-cycle paid invoices', async () => {
@@ -688,10 +687,10 @@ describe('stripeRoutes', () => {
},
},
}
const productEventService = { track: vi.fn() }
const stripeService = createMockStripeService({
getCustomerByStripeId: vi.fn(async () => createMockStripeCustomer()),
})
const productEventService = { track: vi.fn(async () => undefined) }
const webhook = createWebhookOperation({
stripe: {
webhooks: {
@@ -707,18 +706,16 @@ describe('stripeRoutes', () => {
await webhook({ signature: 'test_sig', body: '{}' })
expect(productEventService.track).toHaveBeenCalledWith({
expect(stripeService.upsertInvoice).toHaveBeenCalledWith(expect.objectContaining({
userId: 'user-1',
feature: 'billing',
action: 'subscription_renewed',
status: 'succeeded',
source: 'stripe.webhook',
metadata: {
amount_paid: 1200,
currency: 'usd',
stripe_price_id: null,
},
})
stripeInvoiceId: 'inv_1',
stripeCustomerId: 'cus_1',
stripeSubscriptionId: 'sub_1',
status: 'paid',
amountDue: 1_200,
amountPaid: 1_200,
}))
expect(productEventService.track).not.toHaveBeenCalled()
})
})
})
server/apps/api/src/schemas/index.ts
-1
View File
@@ -4,7 +4,6 @@ export * from './config-kv'
export * from './flux'
export * from './flux-transaction'
export * from './llm-request-log'
export * from './product-events'
export * from './provider-catalog'
export * from './providers'
export * from './stripe'
server/apps/api/src/schemas/product-events.ts
-32
View File
@@ -1,32 +0,0 @@
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>
server/apps/api/src/services/adapters/posthog.ts
+6 -3
View File
@@ -11,6 +11,8 @@ export interface PosthogCaptureInput {
distinctId: string
event: string
properties: Record<string, unknown>
/** Stable event UUID used by PostHog ingestion for replay deduplication. */
uuid?: string
}
/**
@@ -37,9 +39,8 @@ export interface PosthogSink {
* generation facts use `captureQueued`, which is buffered by the SDK and
* flushed on shutdown, so chat completion requests don't wait on PostHog.
*
* Capture failures are logged and swallowed — analytics forwarding must
* never fail the Stripe webhook or auth flow that triggered it. The
* Postgres `product_events` row is the source of truth either way.
* Capture failures are logged and swallowed. Analytics forwarding never
* fails the Stripe webhook or auth flow that produced the business fact.
*/
export function createPosthogSink(options: { projectKey: string, host: string }): PosthogSink {
const client = new PostHog(options.projectKey, { host: options.host })
@@ -51,6 +52,7 @@ export function createPosthogSink(options: { projectKey: string, host: string })
distinctId: input.distinctId,
event: input.event,
properties: input.properties,
...(input.uuid && { uuid: input.uuid }),
})
}
catch (err) {
@@ -64,6 +66,7 @@ export function createPosthogSink(options: { projectKey: string, host: string })
distinctId: input.distinctId,
event: input.event,
properties: input.properties,
...(input.uuid && { uuid: input.uuid }),
})
}
catch (err) {
server/apps/api/src/services/domain/chats.ts
+1 -13
View File
@@ -2,7 +2,6 @@ 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'
@@ -50,7 +49,7 @@ export function resolveSenderId(role: string, userId: string): string | null {
// Service factory
// ---------------------------------------------------------------------------
export function createChatService(db: Database, metrics?: EngagementMetrics | null, productEventService?: ProductEventService) {
export function createChatService(db: Database, metrics?: EngagementMetrics | null) {
// ---- internal helpers ---------------------------------------------------
async function verifyMembership(tx: Parameters<Parameters<Database['transaction']>[0]>[0], chatId: string, userId: string) {
@@ -341,17 +340,6 @@ 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)
server/apps/api/src/services/domain/openai-speech/index.ts
+1 -130
View File
@@ -4,7 +4,6 @@ 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 { ProviderCatalogService } from '../provider-catalog'
import type { RequestLogService } from '../request-log'
import type { VoicePackService } from '../voice-packs'
@@ -19,7 +18,6 @@ import {
AIRI_ATTR_GEN_AI_OPERATION_KIND,
GEN_AI_ATTR_REQUEST_MODEL,
} from '../../../utils/observability'
import { fluxBalanceBucket } from '../flux-balance'
const tracer = trace.getTracer('v1-completions')
@@ -51,7 +49,6 @@ export interface OpenAiSpeechServiceDeps {
llmRouter: LlmRouterService
voicePackService: VoicePackService
providerCatalogService: ProviderCatalogService
productEventService: ProductEventService
genAi?: GenAiMetrics | null
llmTracing: {
startTtsGeneration: (input: Parameters<typeof startTtsGeneration>[0]) => TtsGenerationTrace
@@ -66,12 +63,9 @@ export interface OpenAiSpeechRequest {
}
type TtsTrigger = 'auto' | 'manual'
type TtsVoiceType = 'official_default' | 'official_selected' | 'custom_configured' | 'voice_pack' | 'unknown'
interface TtsAnalyticsContext {
trigger: TtsTrigger
source: 'audio.speech' | 'chat_auto_tts' | 'manual_preview' | 'settings_test'
voiceType: TtsVoiceType
}
/**
@@ -112,11 +106,6 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
if (!voicePackRequest.voicePackId && routedVoice)
await deps.providerCatalogService.assertTtsVoiceEnabled(requestModel, routedVoice)
const voiceMetadata = ttsVoiceMetadata({
voice: requestVoice,
voicePackId: voicePackRequest.voicePackId,
voiceType: analytics.voiceType,
})
const billingUnits = Math.ceil(inputText.length * voicePackRequest.costMultiplier)
logger.withFields({
@@ -127,20 +116,6 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
voice: requestVoice,
}).log('tts speech request')
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_requested',
status: 'started',
source: analytics.source,
model: requestModel,
metadata: {
input_chars: inputText.length,
trigger: analytics.trigger,
...voiceMetadata,
},
})
const flux = await deps.fluxService.getFlux(input.userId)
try {
await deps.ttsMeter.assertCanAfford(input.userId, billingUnits, flux.flux)
@@ -149,24 +124,6 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
if (!(err instanceof ApiError) || err.statusCode !== 402)
throw err
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_blocked',
status: 'blocked',
source: analytics.source,
model: requestModel,
reason: 'insufficient_balance',
metadata: {
input_chars: inputText.length,
billing_units: billingUnits,
block_reason: 'insufficient_balance',
balance_state: 'insufficient',
flux_balance_bucket: fluxBalanceBucket(flux.flux),
trigger: analytics.trigger,
...voiceMetadata,
},
})
logger.withError(err).withFields({
requestId,
userId: input.userId,
@@ -227,23 +184,6 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) {
provider: routeCtx.provider,
status: failure.status,
})
void deps.productEventService.track({
userId: input.userId,
feature: 'tts',
action: 'speech_failed',
status: 'failed',
source: analytics.source,
model: requestModel,
provider: routeCtx.provider,
reason: failure.reason,
metadata: {
http_status: failure.status,
duration_ms: Date.now() - startedAt,
failure_reason: failure.reason,
trigger: analytics.trigger,
...voiceMetadata,
},
})
throw err
}
@@ -255,23 +195,6 @@ 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: analytics.source,
model: requestModel,
provider: routeCtx.provider,
reason: 'upstream_error',
metadata: {
http_status: response.status,
duration_ms: durationMs,
failure_reason: 'upstream_error',
trigger: analytics.trigger,
...voiceMetadata,
},
})
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, {
@@ -306,25 +229,6 @@ 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: analytics.source,
model: requestModel,
provider: routeCtx.provider,
metadata: {
http_status: response.status,
input_chars: inputText.length,
billing_units: billingUnits,
cost_multiplier: voicePackRequest.costMultiplier,
duration_ms: durationMs,
flux_consumed: fluxConsumed,
trigger: analytics.trigger,
...voiceMetadata,
},
})
deps.requestLogService.logRequest({
userId: input.userId,
model: requestModel,
@@ -380,40 +284,7 @@ function ttsAnalyticsContext(body: Record<string, unknown>): TtsAnalyticsContext
|| rawSource === 'settings_test'
? rawSource
: 'audio.speech'
const voiceType = normalizeVoiceType(analytics?.voice_type)
return { trigger, source, voiceType }
}
/**
* Normalizes client-provided TTS voice type into bounded analytics values.
*/
function normalizeVoiceType(value: unknown): TtsVoiceType {
switch (value) {
case 'official_default':
case 'official_selected':
case 'custom_configured':
case 'voice_pack':
return value
default:
return 'unknown'
}
}
/**
* Builds reusable low-cardinality voice metadata for every TTS product event.
*/
function ttsVoiceMetadata(input: {
voice?: string
voicePackId?: string
voiceType: TtsVoiceType
}): Record<string, unknown> {
const voiceType = input.voicePackId ? 'voice_pack' : input.voiceType
return {
...(input.voice ? { voice_id: input.voice } : {}),
voice_type: voiceType,
...(input.voicePackId ? { voice_pack_id: input.voicePackId } : {}),
}
return { trigger, source }
}
async function voicePackRequestOptions(
server/apps/api/src/services/domain/product-events.test.ts
+100 -294
View File
@@ -1,148 +1,25 @@
import type { Database } from '../../libs/db'
import type { ProductMetrics } from '../../otel'
import { describe, expect, it, vi } from 'vitest'
import { sql } from 'drizzle-orm'
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,
}])
})
it('writes blocked TTS events for server-side preflight decisions', async () => {
const events = { add: vi.fn() }
const service = createProductEventService(db, { events } as unknown as ProductMetrics)
await service.track({
userId: 'user-1',
feature: 'tts',
action: 'speech_blocked',
status: 'blocked',
source: 'chat_auto_tts',
reason: 'insufficient_balance',
metadata: {
trigger: 'auto',
balance_state: 'insufficient',
flux_balance_bucket: 'zero',
},
})
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(1)
expect(rows[0]).toMatchObject({
feature: 'tts',
action: 'speech_blocked',
status: 'blocked',
source: 'chat_auto_tts',
reason: 'insufficient_balance',
})
expect(events.add).toHaveBeenCalledWith(1, {
feature: 'tts',
action: 'speech_blocked',
status: 'blocked',
source: 'chat_auto_tts',
reason: 'insufficient_balance',
flux_balance_bucket: 'zero',
})
})
it('forwards allowlisted business facts to PostHog keyed by user id, mapping user_signed_up to signup_completed', async () => {
it('captures only the server-side funnel facts shared with the Go service', async () => {
const capture = vi.fn(async () => {})
const sink = { capture, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
await service.track({
userId: 'user-1',
feature: 'auth',
action: 'user_signed_up',
status: 'succeeded',
})
await service.track({
userId: 'user-1',
feature: 'billing',
action: 'checkout_started',
status: 'succeeded',
source: 'stripe.checkout',
})
await service.track({
userId: 'user-1',
feature: 'billing',
@@ -151,14 +28,29 @@ describe('productEventService', () => {
source: 'stripe.webhook',
metadata: { amount_minor_unit: 990, currency: 'usd' },
})
await service.track({
userId: 'user-2',
feature: 'auth',
action: 'user_signed_up',
status: 'succeeded',
})
expect(capture).toHaveBeenNthCalledWith(1, {
distinctId: 'user-1',
event: 'signup_completed',
properties: {
app_surface: 'server',
airi_user_id: 'user-1',
feature: 'auth',
status: 'succeeded',
},
})
expect(capture).toHaveBeenNthCalledWith(2, {
distinctId: 'user-1',
event: 'checkout_created',
properties: {
app_surface: 'server',
airi_user_id: 'user-1',
feature: 'billing',
status: 'succeeded',
source: 'stripe.checkout',
},
})
expect(capture).toHaveBeenNthCalledWith(3, {
distinctId: 'user-1',
event: 'payment_completed',
properties: {
@@ -171,195 +63,113 @@ describe('productEventService', () => {
currency: 'usd',
},
})
expect(capture).toHaveBeenNthCalledWith(2, {
distinctId: 'user-2',
event: 'signup_completed',
properties: {
app_surface: 'server',
airi_user_id: 'user-2',
feature: 'auth',
status: 'succeeded',
},
})
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(2)
})
it('merges Stripe webhook conversions with the browser PostHog person when a distinct id is present', async () => {
it('merges a Stripe conversion with its browser PostHog person', async () => {
const capture = vi.fn(async () => {})
const sink = { capture, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
await service.track({
userId: 'user-1',
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
source: 'stripe.webhook',
eventId: 'cs_123',
metadata: {
posthog_distinct_id: 'anon-browser-1',
posthog_session_id: 'ph-session-1',
stripe_checkout_session_id: 'cs_1',
},
})
expect(capture).toHaveBeenNthCalledWith(1, {
distinctId: 'user-1',
event: '$identify',
uuid: expect.stringMatching(/^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/),
properties: {
$insert_id: 'cs_123',
$anon_distinct_id: 'anon-browser-1',
$session_id: 'ph-session-1',
airi_user_id: 'user-1',
},
})
expect(capture).toHaveBeenNthCalledWith(2, {
expect(capture).toHaveBeenNthCalledWith(2, expect.objectContaining({
distinctId: 'user-1',
event: 'payment_completed',
properties: {
app_surface: 'server',
airi_user_id: 'user-1',
posthog_distinct_id: 'anon-browser-1',
$session_id: 'ph-session-1',
feature: 'billing',
status: 'succeeded',
source: 'stripe.webhook',
posthog_session_id: 'ph-session-1',
stripe_checkout_session_id: 'cs_1',
uuid: expect.stringMatching(/^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/),
properties: expect.objectContaining({ $insert_id: 'cs_123' }),
}))
})
it('uses a stable PostHog UUID for replayed conversion captures', async () => {
const capture = vi.fn(async () => {})
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
const input = {
userId: 'user-1' as const,
feature: 'billing' as const,
action: 'payment_completed' as const,
status: 'succeeded' as const,
eventId: 'cs_replayed',
}
await service.track(input)
await service.track(input)
expect(capture).toHaveBeenCalledTimes(2)
const captures = capture.mock.calls as unknown as Array<[
{
uuid?: string
properties: Record<string, unknown>
},
})
]>
const first = captures[0]![0]
const replay = captures[1]![0]
expect(first.uuid).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/)
expect(replay.uuid).toBe(first.uuid)
expect(first.properties.$insert_id).toBe('cs_replayed')
})
it('does not forward high-volume per-request actions to PostHog', async () => {
it('rejects metadata that can overwrite service-controlled PostHog properties', async () => {
const capture = vi.fn(async () => {})
const sink = { capture, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
await service.track({
userId: 'user-1',
feature: 'gen_ai_chat',
action: 'completion_succeeded',
status: 'succeeded',
})
await service.track({
userId: 'user-1',
feature: 'billing',
action: 'checkout_started',
status: 'started',
})
expect(capture).not.toHaveBeenCalled()
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(2)
})
it('captures an LLM generation as a PostHog AI fact without storing prompts or responses', async () => {
const capture = vi.fn(async () => {})
const captureQueued = vi.fn()
const sink = { capture, captureQueued, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
service.trackGeneration({
userId: 'user-1',
traceId: 'session-1',
generationId: 'round-1',
model: 'openai/gpt-5-mini',
provider: 'openai',
providerType: 'official',
usageSource: 'reported',
inputTokens: 12,
outputTokens: 8,
totalTokens: 20,
conversationId: 'session-1',
conversationIdSource: 'client_header',
roundId: 'round-1',
appSurface: 'electron',
captureSurface: 'server',
})
expect(capture).not.toHaveBeenCalled()
expect(captureQueued).toHaveBeenCalledWith({
distinctId: 'user-1',
event: '$ai_generation',
properties: {
$ai_trace_id: 'session-1',
$ai_session_id: 'session-1',
$ai_span_id: 'round-1',
$ai_model: 'openai/gpt-5-mini',
$ai_provider: 'openai',
$ai_input_tokens: 12,
$ai_output_tokens: 8,
$ai_total_tokens: 20,
$insert_id: 'ai-generation:round-1',
airi_user_id: 'user-1',
provider_type: 'official',
usage_source: 'reported',
token_usage_available: true,
cost_usd_source: 'unavailable',
cost_usd_known: false,
conversation_id: 'session-1',
conversation_id_source: 'client_header',
round_id: 'round-1',
app_surface: 'electron',
capture_surface: 'server',
},
})
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(0)
})
it('does not forward to PostHog when the DB write fails', async () => {
// ROOT CAUSE:
//
// track() swallows DB insert errors to protect the caller, but the
// PostHog forwarding block ran unconditionally afterwards — a Postgres
// outage during a Stripe webhook would mint `payment_completed` in
// PostHog with no `product_events` row backing it, breaking the
// "Postgres is the fact of record" invariant and later reconciliation.
// Found by PR #2038 review.
//
// Fixed by gating forwarding on a `persisted` flag set only after the
// insert resolves.
const capture = vi.fn(async () => {})
const sink = { capture, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
// Simulate a DB outage by renaming the table out from under the insert.
await db.execute(sql`ALTER TABLE product_events RENAME TO product_events_outage`)
try {
await expect(service.track({
userId: 'user-1',
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
metadata: { $insert_id: 'spoofed' },
})).resolves.toBeUndefined()
}
finally {
await db.execute(sql`ALTER TABLE product_events_outage RENAME TO product_events`)
}
expect(capture).not.toHaveBeenCalled()
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(0)
})
it('persists the product event even when a misbehaving sink throws', async () => {
// ROOT CAUSE:
//
// The PosthogSink contract says implementations swallow transport
// errors, but the forwarding call sits on the Stripe webhook path —
// if a sink ever throws, an unguarded `await` would fail the webhook
// after the fact was already persisted, causing Stripe to retry and
// (before the idempotency guard) double-process the payment.
//
// track() therefore wraps forwarding in its own try/catch: the DB row
// must survive and track() must resolve regardless of sink behavior.
it('still captures the funnel event when identity merging fails', async () => {
const capture = vi.fn()
.mockRejectedValueOnce(new Error('identify failed'))
.mockResolvedValueOnce(undefined)
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
await expect(service.track({
userId: 'user-1',
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
eventId: 'cs_456',
metadata: { posthog_distinct_id: 'anon-browser-1' },
})).resolves.toBeUndefined()
expect(capture).toHaveBeenCalledTimes(2)
expect(capture).toHaveBeenNthCalledWith(2, expect.objectContaining({
event: 'payment_completed',
properties: expect.objectContaining({ $insert_id: 'cs_456' }),
}))
})
it('does not fail a business path when capture throws', async () => {
const capture = vi.fn(async () => {
throw new Error('posthog exploded')
})
const sink = { capture, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
await expect(service.track({
userId: 'user-1',
@@ -367,9 +177,5 @@ describe('productEventService', () => {
action: 'payment_completed',
status: 'succeeded',
})).resolves.toBeUndefined()
const rows = await db.select().from(schema.productEvents)
expect(rows).toHaveLength(1)
expect(rows[0]).toMatchObject({ action: 'payment_completed', status: 'succeeded' })
})
})
server/apps/api/src/services/domain/product-events.ts
+59 -156
View File
@@ -1,43 +1,33 @@
import type { Database } from '../../libs/db'
import type { ProductMetrics } from '../../otel'
import type { ProductEventMetadata } from '../../schemas/product-events'
import type { PosthogSink } from '../adapters/posthog'
import { useLogger } from '@guiiai/logg'
import { and, asc, count, gte, lt, sql } from 'drizzle-orm'
import { createHash } from 'node:crypto'
import * as schema from '../../schemas/product-events'
import { useLogger } from '@guiiai/logg'
const logger = useLogger('product-events')
export type ProductFeature = 'auth' | 'chat' | 'gen_ai_chat' | 'tts' | 'billing' | 'voice_pack'
const RESERVED_POSTHOG_METADATA_KEYS = new Set([
'$insert_id',
'$session_id',
'airi_user_id',
'app_surface',
'feature',
'source',
'status',
])
export type ProductEventStatus = 'started' | 'succeeded' | 'failed' | 'blocked'
export type ProductFeature = 'auth' | 'billing'
export type ProductEventStatus = 'succeeded'
export type ProductEventMetadata = Record<string, string | number | boolean | null>
export type ProductAction
= | 'user_signed_up'
| 'session_started'
| 'message_pushed'
| 'completion_requested'
| 'completion_succeeded'
| 'completion_failed'
| 'speech_requested'
| 'speech_succeeded'
| 'speech_failed'
| 'speech_blocked'
| 'voice_pack_created'
| 'voice_pack_updated'
| 'voice_pack_disabled'
| 'checkout_started'
| 'payment_completed'
| 'subscription_started'
| 'subscription_renewed'
| 'subscription_cancelled'
| 'topic_classified'
/**
* Product event fact written to AIRI's own Postgres analytics table.
*/
/** Product funnel fact forwarded to PostHog from the server. */
export interface ProductEventInput {
/** Better Auth user id. Kept in Postgres only; never emitted as a Prometheus label. */
userId: string
@@ -49,31 +39,10 @@ export interface ProductEventInput {
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
/** Stable source event id used by PostHog for replay-safe deduplication. */
eventId?: string
}
/** Product runtime where the user initiated the AI generation. */
@@ -116,10 +85,8 @@ export interface AiGenerationEventInput {
}
/**
* Server-side actions worth a PostHog copy, mapped to the event name the
* client-side funnels expect. Only business facts that terminate or anchor
* a funnel are forwarded — per-request LLM/TTS volume stays in Postgres and
* Grafana where it belongs (see `docs/ai-context/metrics-ownership.md`).
* Server-side actions that anchor a PostHog product funnel. Per-request LLM
* and TTS telemetry stays in operational systems and does not enter this path.
*
* `user_signed_up` maps to `signup_completed` because the identified server
* hook is the canonical registration fact for every signup method. Anonymous
@@ -127,31 +94,8 @@ export interface AiGenerationEventInput {
*/
const POSTHOG_FORWARDED_ACTIONS: Partial<Record<ProductAction, string>> = {
user_signed_up: 'signup_completed',
checkout_started: 'checkout_created',
payment_completed: 'payment_completed',
subscription_started: 'subscription_started',
subscription_renewed: 'subscription_renewed',
subscription_cancelled: 'subscription_cancelled',
}
/**
* Builds bounded Prometheus labels from product event inputs.
*/
function metricLabels(input: ProductEventInput): Record<string, string> {
const attrs: Record<string, string> = {
feature: input.feature,
action: input.action,
status: input.status,
}
if (input.source)
attrs.source = input.source
if (input.reason)
attrs.reason = input.reason
const fluxBalanceBucket = input.metadata?.flux_balance_bucket
if (typeof fluxBalanceBucket === 'string')
attrs.flux_balance_bucket = fluxBalanceBucket
return attrs
}
function stringMetadata(input: ProductEventInput, key: string): string | undefined {
@@ -159,24 +103,32 @@ function stringMetadata(input: ProductEventInput, key: string): string | undefin
return typeof value === 'string' && value.length > 0 ? value : undefined
}
function posthogEventUuid(event: string, eventId: string): string {
const digest = createHash('sha256').update(`airi:posthog:${event}:${eventId}`, 'utf8').digest()
digest[6] = (digest[6] & 0x0F) | 0x50
digest[8] = (digest[8] & 0x3F) | 0x80
const hex = digest.subarray(0, 16).toString('hex')
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`
}
function hasReservedMetadataKey(metadata: ProductEventMetadata | undefined): boolean {
return metadata != null && Object.keys(metadata).some(key => RESERVED_POSTHOG_METADATA_KEYS.has(key))
}
/**
* Creates AIRI's first-party product analytics event writer.
* Creates AIRI's server-side PostHog product analytics 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.
* - A server has an authenticated user id and confirms a funnel fact.
*
* 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.
* - Callers pass only the typed funnel actions in this module.
*
* Returns:
* - Best-effort event writer plus a DB aggregation helper for analytics jobs.
* - A best-effort event writer. Capture errors never change the business flow.
*/
export function createProductEventService(db: Database, metrics?: ProductMetrics | null, posthog?: PosthogSink | null) {
export function createProductEventService(posthog?: PosthogSink | null) {
return {
trackGeneration(input: AiGenerationEventInput): void {
if (!posthog)
@@ -222,60 +174,43 @@ export function createProductEventService(db: Database, metrics?: ProductMetrics
},
async track(input: ProductEventInput): Promise<void> {
// Postgres is the fact of record, so forwarding is gated both ways:
// the DB write comes first (a PostHog outage can't lose the row) and
// forwarding only runs when the row actually landed (a DB outage
// can't mint PostHog events with no DB backing).
let persisted = false
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,
})
persisted = true
metrics?.events.add(1, metricLabels(input))
}
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')
}
// PostHog copy so browser funnels (identified by the same Better Auth
// user id) get their server-side terminator events.
const forwardedEvent = POSTHOG_FORWARDED_ACTIONS[input.action]
if (persisted && posthog && forwardedEvent) {
try {
if (!posthog || !forwardedEvent)
return
if (hasReservedMetadataKey(input.metadata)) {
logger.withFields({ action: input.action }).warn('Rejected reserved PostHog product event metadata')
return
}
const posthogDistinctId = stringMetadata(input, 'posthog_distinct_id')
const posthogSessionId = stringMetadata(input, 'posthog_session_id')
if (posthogDistinctId && posthogDistinctId !== input.userId) {
try {
await posthog.capture({
distinctId: input.userId,
event: '$identify',
properties: {
...(input.eventId && { $insert_id: input.eventId }),
$anon_distinct_id: posthogDistinctId,
airi_user_id: input.userId,
...(posthogSessionId && { $session_id: posthogSessionId }),
},
...(input.eventId && { uuid: posthogEventUuid('$identify', input.eventId) }),
})
}
catch (err) {
logger.withError(err).withFields({ action: input.action }).warn('PostHog anonymous identity capture failed')
}
}
try {
await posthog.capture({
distinctId: input.userId,
event: forwardedEvent,
properties: {
...input.metadata,
...(input.eventId && { $insert_id: input.eventId }),
app_surface: 'server',
airi_user_id: input.userId,
...(posthogDistinctId && { posthog_distinct_id: posthogDistinctId }),
@@ -283,45 +218,13 @@ export function createProductEventService(db: Database, metrics?: ProductMetrics
feature: input.feature,
status: input.status,
...(input.source && { source: input.source }),
...(input.reason && { reason: input.reason }),
...input.metadata,
},
...(input.eventId && { uuid: posthogEventUuid(forwardedEvent, input.eventId) }),
})
}
catch (err) {
// The sink contract already swallows transport errors; this guard
// is the last line so a misbehaving sink can never fail the
// webhook/auth flow that produced the business fact.
logger.withError(err).withFields({ action: input.action }).warn('PostHog forwarding threw; product event already persisted')
logger.withError(err).withFields({ action: input.action }).warn('PostHog product analytics capture failed')
}
}
},
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),
}))
},
}
}
server/apps/api/src/utils/observability.ts
-5
View File
@@ -85,11 +85,6 @@ 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'
server/apps/auth/src/auth.ts
-5
View File
@@ -787,11 +787,6 @@ export function createAuth(
.set({ lastSeenAt: new Date() })
.where(eq(authSchema.user.id, session.userId))
.catch(err => logger.withError(err).withFields({ userId: session.userId }).warn('Failed to update user lastSeenAt; continuing session create'))
void resourceApi?.trackAuthEvent({
userId: session.userId,
action: 'session_started',
source: 'better-auth.session.create',
})
},
},
},
server/apps/auth/src/resource-api.ts
+2 -2
View File
@@ -6,8 +6,8 @@ export type UserDeletionReason = 'user-requested' | 'admin' | 'compliance'
export interface AuthEventInput {
userId: string
action: 'user_signed_up' | 'session_started'
source: 'better-auth.user.create' | 'better-auth.session.create'
action: 'user_signed_up'
source: 'better-auth.user.create'
}
/**