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

feat: add llm usage and stripe identity analytics (#2056)

Browse Source
This commit is contained in:
RainbowBird
2026-07-14 21:56:47 +08:00
committed by GitHub
parent 563a738c88
commit 2e9f9ca435
29 changed files with 1101 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/app.test.ts
+1
View File
@@ -55,6 +55,7 @@ function createTestDeps() {
providerCatalogService: {} as any,
productEventService: {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
},
configKV: {
apps/server/src/routes/admin/voice-packs/route.test.ts
+1
View File
@@ -49,6 +49,7 @@ function createService() {
function createProductEventService(): ProductEventService {
return {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
}
}
apps/server/src/routes/audio-speech-ws/route.test.ts
+1
View File
@@ -186,6 +186,7 @@ function makeFakeDeps(overrides: {
}
const productEventService = {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
}
const configKV = {
apps/server/src/routes/openai/v1/analytics.ts
+14
View File
@@ -0,0 +1,14 @@
import type { AiGenerationAppSurface } from '../../../services/domain/product-events'
export const AIRI_CHAT_SESSION_ID_HEADER = 'x-airi-session-id'
export const AIRI_CHAT_ROUND_ID_HEADER = 'x-airi-round-id'
export const AIRI_CHAT_APP_SURFACE_HEADER = 'x-airi-app-surface'
const CLIENT_CHAT_ANALYTICS_SURFACES = new Set<AiGenerationAppSurface>(['web', 'mobile', 'electron'])
export function resolveChatAnalyticsSurface(value: string | undefined): AiGenerationAppSurface {
if (CLIENT_CHAT_ANALYTICS_SURFACES.has(value as AiGenerationAppSurface))
return value as AiGenerationAppSurface
return 'server'
}
apps/server/src/routes/openai/v1/index.ts
+10 -2
View File
@@ -5,6 +5,12 @@ import type { LlmTracingDeps, V1RouteDeps } from './types'
import { authGuard } from '../../../middlewares/auth'
import { configGuard } from '../../../middlewares/config-guard'
import {
AIRI_CHAT_APP_SURFACE_HEADER,
AIRI_CHAT_ROUND_ID_HEADER,
AIRI_CHAT_SESSION_ID_HEADER,
resolveChatAnalyticsSurface,
} from './analytics'
import { createV1Gateway } from './gateway'
import { chatCompletionsRateLimit } from './middlewares'
import { chatCompletions } from './operations/chat-completions'
@@ -41,7 +47,9 @@ export function createV1Routes(input: CreateV1RoutesDeps) {
return {
userId: user.id,
body,
sessionId: c.req.header('x-airi-session-id'),
sessionId: c.req.header(AIRI_CHAT_SESSION_ID_HEADER),
roundId: c.req.header(AIRI_CHAT_ROUND_ID_HEADER),
appSurface: resolveChatAnalyticsSurface(c.req.header(AIRI_CHAT_APP_SURFACE_HEADER)),
abortSignal: c.req.raw.signal,
}
},
@@ -66,7 +74,7 @@ export function createV1Routes(input: CreateV1RoutesDeps) {
return {
userId: user.id,
body,
sessionId: c.req.header('x-airi-session-id'),
sessionId: c.req.header(AIRI_CHAT_SESSION_ID_HEADER),
abortSignal: c.req.raw.signal,
}
},
apps/server/src/routes/openai/v1/operations/chat-completions/index.ts
+88
View File
@@ -1,5 +1,6 @@
import type { CapabilityAliasRoute } from '../../../../../schemas/provider-catalog'
import type { UsageInfo } from '../../../../../services/domain/billing/billing'
import type { AiGenerationAppSurface } from '../../../../../services/domain/product-events'
import type { GatewayCallback } from '../../gateway'
import type { V1RouteDeps } from '../../types'
@@ -20,9 +21,25 @@ export interface ChatCompletionsOperationRequest {
userId: string
body: Record<string, unknown>
sessionId?: string
roundId?: string
appSurface: AiGenerationAppSurface
abortSignal?: AbortSignal
}
interface GenerationCaptureInput {
deps: V1RouteDeps
userId: string
requestId: string
sessionId?: string
roundId?: string
appSurface: AiGenerationAppSurface
generationModel: string
routeCtxProvider: string
usage: UsageInfo
durationMs: number
stream: boolean
}
export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.completions'> {
const logger = useLogger('v1-completions').useGlobalConfig()
const telemetry = createRouteTelemetry({
@@ -182,7 +199,11 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple
durationMs,
requestId,
userId: input.userId,
sessionId: input.sessionId,
roundId: input.roundId,
appSurface: input.appSurface,
requestModel,
generationModel: langfuseModel,
routeCtxProvider: routeCtx.provider,
billing,
billingPolicy,
@@ -199,7 +220,11 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple
durationMs,
requestId,
userId: input.userId,
sessionId: input.sessionId,
roundId: input.roundId,
appSurface: input.appSurface,
requestModel,
generationModel: langfuseModel,
routeCtxProvider: routeCtx.provider,
billing,
billingPolicy,
@@ -213,6 +238,33 @@ interface ChatModelAliasPlan {
modelIds: string[]
}
function captureGeneration(input: GenerationCaptureInput): void {
const generationId = input.roundId ?? input.requestId
const totalTokens = input.usage.promptTokens != null && input.usage.completionTokens != null
? input.usage.promptTokens + input.usage.completionTokens
: undefined
input.deps.productEventService.trackGeneration({
userId: input.userId,
traceId: input.sessionId ?? input.requestId,
generationId,
model: input.generationModel,
provider: input.routeCtxProvider || 'unknown',
providerType: 'official',
usageSource: input.usage.promptTokens != null || input.usage.completionTokens != null
? 'reported'
: 'unavailable',
inputTokens: input.usage.promptTokens,
outputTokens: input.usage.completionTokens,
totalTokens,
conversationId: input.sessionId,
roundId: generationId,
appSurface: input.appSurface,
latencySeconds: input.durationMs / 1000,
stream: input.stream,
})
}
async function resolveChatModelAliasPlan(deps: V1RouteDeps, aliasId: string): Promise<ChatModelAliasPlan> {
const alias = await deps.providerCatalogService.resolveEnabledAlias('llm', aliasId)
const primaryRoutes = alias.routes.filter(route => route.pool === 'primary')
@@ -302,7 +354,11 @@ function streamChatCompletion(input: {
durationMs: number
requestId: string
userId: string
sessionId?: string
roundId?: string
appSurface: AiGenerationAppSurface
requestModel: string
generationModel: string
routeCtxProvider: string
billing: ChatBilling
billingPolicy: ChatBillingPolicy
@@ -421,6 +477,20 @@ function streamChatCompletion(input: {
})
input.telemetry.recordMetrics({ model: input.requestModel, status: input.response.status, type: 'chat', provider: input.routeCtxProvider, durationMs: input.durationMs, fluxConsumed, ...usage })
captureGeneration({
deps: input.deps,
userId: input.userId,
requestId: input.requestId,
sessionId: input.sessionId,
roundId: input.roundId,
appSurface: input.appSurface,
generationModel: input.generationModel,
routeCtxProvider: input.routeCtxProvider,
usage,
durationMs: input.durationMs,
stream: true,
})
// Debit flux via DB transaction (source of truth)
// NOTICE: streaming response is already sent, so we cannot reject on failure.
// Log at error level so unpaid usage is visible in monitoring/alerts.
@@ -507,7 +577,11 @@ async function completeNonStreamingChat(input: {
durationMs: number
requestId: string
userId: string
sessionId?: string
roundId?: string
appSurface: AiGenerationAppSurface
requestModel: string
generationModel: string
routeCtxProvider: string
billing: ChatBilling
billingPolicy: ChatBillingPolicy
@@ -556,6 +630,20 @@ async function completeNonStreamingChat(input: {
})
input.telemetry.recordMetrics({ model: input.requestModel, status: input.response.status, type: 'chat', provider: input.routeCtxProvider, durationMs: input.durationMs, fluxConsumed, ...usage })
captureGeneration({
deps: input.deps,
userId: input.userId,
requestId: input.requestId,
sessionId: input.sessionId,
roundId: input.roundId,
appSurface: input.appSurface,
generationModel: input.generationModel,
routeCtxProvider: input.routeCtxProvider,
usage,
durationMs: input.durationMs,
stream: false,
})
// Debit flux via DB transaction (source of truth).
// The upstream call has already happened (cost incurred), so partial
// debit + `fluxUnbilled` is the only sane recovery — same shape as the
apps/server/src/routes/openai/v1/route.test.ts
+41 -3
View File
@@ -14,6 +14,11 @@ import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
import { createV1Routes } from '.'
import { ApiError } from '../../../utils/error'
import {
AIRI_CHAT_APP_SURFACE_HEADER,
AIRI_CHAT_ROUND_ID_HEADER,
AIRI_CHAT_SESSION_ID_HEADER,
} from './analytics'
function createMockFluxService(flux = 100): FluxService {
return {
@@ -143,6 +148,7 @@ function createMockLlmRouter(impl?: Partial<LlmRouterService>): LlmRouterService
function createMockProductEventService(): ProductEventService {
return {
track: vi.fn(async () => undefined),
trackGeneration: vi.fn(async () => undefined),
countDistinctUsersByFeature: vi.fn(async () => []),
}
}
@@ -907,7 +913,7 @@ describe('v1CompletionsRoutes', () => {
}
})
it('records Langfuse chat generation with the router-resolved upstream model', async () => {
it('records Langfuse and PostHog generations with authoritative usage and correlation', async () => {
const llmRouter = createMockLlmRouter({
route: vi.fn(async (_req, ctx) => {
if (ctx) {
@@ -924,12 +930,27 @@ describe('v1CompletionsRoutes', () => {
}) as any,
})
const llmTracing = createMockLlmTracing()
const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, llmRouter, llmTracing)
const productEventService = createMockProductEventService()
const app = createTestApp(
createMockFluxService(),
createMockConfigKV(),
undefined,
undefined,
undefined,
llmRouter,
llmTracing,
productEventService,
)
await app.fetch(
new Request('http://localhost/api/v1/openai/chat/completions', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
headers: {
'Content-Type': 'application/json',
[AIRI_CHAT_SESSION_ID_HEADER]: 'conversation-1',
[AIRI_CHAT_ROUND_ID_HEADER]: 'round-1',
[AIRI_CHAT_APP_SURFACE_HEADER]: 'electron',
},
body: JSON.stringify({ model: 'chat-auto', messages: [{ role: 'user', content: 'hi' }] }),
}),
{ user: testUser } as any,
@@ -942,6 +963,23 @@ describe('v1CompletionsRoutes', () => {
userId: 'user-1',
}),
)
expect(productEventService.trackGeneration).toHaveBeenCalledWith({
userId: 'user-1',
traceId: 'conversation-1',
generationId: 'round-1',
model: 'openai/gpt-4o-mini',
provider: 'openrouter',
providerType: 'official',
usageSource: 'reported',
inputTokens: 1,
outputTokens: 2,
totalTokens: 3,
conversationId: 'conversation-1',
roundId: 'round-1',
appSurface: 'electron',
latencySeconds: expect.any(Number),
stream: false,
})
})
it('should not charge flux when upstream returns error', async () => {
apps/server/src/routes/stripe/operations/checkout.ts
+30
View File
@@ -33,6 +33,11 @@ export interface CheckoutOperationInput {
request: Request
}
interface PosthogIdentityHeaders {
distinctId?: string
sessionId?: string
}
/**
* Creates Stripe checkout sessions for Flux packages.
*
@@ -75,6 +80,7 @@ export function createCheckoutOperation(deps: CheckoutOperationDeps) {
const paymentMethods = await deps.configKV.getOptional('STRIPE_PAYMENT_METHODS')
const paymentMethodOptions = await deps.configKV.getOptional('STRIPE_PAYMENT_METHOD_OPTIONS') ?? {}
const posthogIdentity = readPosthogIdentityHeaders(input.request)
const sessionParams: CheckoutSessionCreateParams = {
line_items: [{ price: stripePriceId, quantity: 1 }],
@@ -87,6 +93,8 @@ export function createCheckoutOperation(deps: CheckoutOperationDeps) {
metadata: {
userId: input.user.id,
fluxAmount: String(fluxAmount),
...(posthogIdentity.distinctId && { posthogDistinctId: posthogIdentity.distinctId }),
...(posthogIdentity.sessionId && { posthogSessionId: posthogIdentity.sessionId }),
},
}
@@ -133,9 +141,31 @@ export function createCheckoutOperation(deps: CheckoutOperationDeps) {
flux_amount: fluxAmount,
amount_total: session.amount_total,
currency: session.currency,
...(posthogIdentity.distinctId && { posthog_distinct_id: posthogIdentity.distinctId }),
...(posthogIdentity.sessionId && { posthog_session_id: posthogIdentity.sessionId }),
},
})
return { url: session.url }
}
}
function readPosthogIdentityHeaders(request: Request): PosthogIdentityHeaders {
const distinctId = readStripeMetadataHeader(request, 'x-posthog-distinct-id')
const sessionId = readStripeMetadataHeader(request, 'x-posthog-session-id')
return {
...(distinctId && { distinctId }),
...(sessionId && { sessionId }),
}
}
function readStripeMetadataHeader(request: Request, name: string): string | undefined {
const value = request.headers.get(name)?.trim()
if (!value)
return undefined
// Stripe metadata values are capped and user-controlled headers can be
// oversized. Truncating keeps the checkout request valid without turning
// analytics identity into a payment blocker.
return value.slice(0, 200)
}
apps/server/src/routes/stripe/operations/webhook.ts
+6
View File
@@ -89,6 +89,8 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
const userId = event.data.object.metadata?.userId
if (userId) {
const fluxAmount = Number(event.data.object.metadata?.fluxAmount)
const posthogDistinctId = event.data.object.metadata?.posthogDistinctId
const posthogSessionId = event.data.object.metadata?.posthogSessionId
void deps.productEventService?.track({
userId,
feature: 'billing',
@@ -99,6 +101,10 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
amount_total: event.data.object.amount_total,
currency: event.data.object.currency,
flux_amount: Number.isFinite(fluxAmount) ? fluxAmount : null,
stripe_checkout_session_id: event.data.object.id,
stripe_customer_id: typeof event.data.object.customer === 'string' ? event.data.object.customer : event.data.object.customer?.id ?? null,
...(posthogDistinctId && { posthog_distinct_id: posthogDistinctId }),
...(posthogSessionId && { posthog_session_id: posthogSessionId }),
},
})
}
apps/server/src/routes/stripe/route.test.ts
+141
View File
@@ -10,6 +10,7 @@ import { describe, expect, it, vi } from 'vitest'
import { createStripeRoutes, formatPrice } from '.'
import { ApiError } from '../../utils/error'
import { createCheckoutOperation } from './operations/checkout'
import { createWebhookOperation } from './operations/webhook'
// --- Mock helpers ---
@@ -302,6 +303,77 @@ describe('stripeRoutes', () => {
)
expect(res.status).toBe(503)
})
it('stores browser PostHog identity in Stripe checkout metadata', async () => {
const createSession = vi.fn(async input => ({
id: 'cs_1',
url: 'https://checkout.stripe.com/cs_1',
customer: null,
mode: 'payment',
status: 'open',
payment_status: 'unpaid',
amount_total: 500,
currency: 'usd',
success_url: 'http://localhost/settings/flux?success=true',
cancel_url: 'http://localhost/settings/flux?canceled=true',
payment_intent: null,
subscription: null,
metadata: input.metadata,
expires_at: null,
}))
const productEventService = { track: vi.fn() }
const operation = createCheckoutOperation({
stripe: {
checkout: {
sessions: {
create: createSession,
},
},
} as any,
priceCatalog: {
findActivePrice: vi.fn(async () => ({
id: 'price_test_500',
currency: 'usd',
unitAmount: 500,
currencyOptions: {},
metadata: { fluxAmount: '500' },
})),
getActivePrices: vi.fn(),
} as any,
stripeService: createMockStripeService(),
configKV: createMockConfigKV({ STRIPE_PAYMENT_METHODS: undefined }),
env: testEnv,
productEventService: productEventService as any,
})
await operation({
user: testUser as any,
body: { stripePriceId: 'price_test_500' },
request: new Request('http://localhost/api/v1/stripe/checkout', {
headers: {
'x-posthog-distinct-id': 'anon-browser-1',
'x-posthog-session-id': 'ph-session-1',
},
}),
})
expect(createSession).toHaveBeenCalledWith(expect.objectContaining({
metadata: {
userId: 'user-1',
fluxAmount: '500',
posthogDistinctId: 'anon-browser-1',
posthogSessionId: 'ph-session-1',
},
}))
expect(productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
userId: 'user-1',
action: 'checkout_started',
metadata: expect.objectContaining({
posthog_distinct_id: 'anon-browser-1',
posthog_session_id: 'ph-session-1',
}),
}))
})
})
describe('gET /api/v1/stripe/orders', () => {
@@ -471,6 +543,75 @@ describe('stripeRoutes', () => {
expect(res.status).toBe(503)
})
it('records payment completion with Stripe and PostHog identity from checkout metadata', async () => {
const checkoutEvent = {
id: 'evt_checkout_completed',
type: 'checkout.session.completed',
data: {
object: {
id: 'cs_1',
customer: 'cus_1',
customer_email: 'test@example.com',
mode: 'payment',
status: 'complete',
payment_status: 'paid',
amount_total: 500,
currency: 'usd',
success_url: 'http://localhost/settings/flux?success=true',
cancel_url: 'http://localhost/settings/flux?canceled=true',
payment_intent: 'pi_1',
subscription: null,
metadata: {
userId: 'user-1',
fluxAmount: '500',
posthogDistinctId: 'anon-browser-1',
posthogSessionId: 'ph-session-1',
},
expires_at: null,
},
},
}
const productEventService = { track: vi.fn() }
const billingService = createMockBillingService()
const webhook = createWebhookOperation({
stripe: {
webhooks: {
constructEvent: vi.fn(() => checkoutEvent),
},
} as any,
webhookSecret: 'whsec_test',
fluxService: createMockFluxService(),
stripeService: createMockStripeService(),
billingService,
productEventService: productEventService as any,
})
await webhook({ signature: 'test_sig', body: '{}' })
expect(billingService.creditFluxFromStripeCheckout).toHaveBeenCalledWith(expect.objectContaining({
stripeEventId: 'evt_checkout_completed',
userId: 'user-1',
stripeSessionId: 'cs_1',
fluxAmount: 500,
}))
expect(productEventService.track).toHaveBeenCalledWith({
userId: 'user-1',
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
source: 'stripe.webhook',
metadata: {
amount_total: 500,
currency: 'usd',
flux_amount: 500,
stripe_checkout_session_id: 'cs_1',
stripe_customer_id: 'cus_1',
posthog_distinct_id: 'anon-browser-1',
posthog_session_id: 'ph-session-1',
},
})
})
it('records subscription lifecycle product events from Stripe webhooks', async () => {
const subscriptionEvent = {
id: 'evt_sub_created',
apps/server/src/services/adapters/posthog.ts
+23 -4
View File
@@ -18,6 +18,12 @@ export interface PosthogCaptureInput {
* an interface so tests inject a fake instead of mocking the SDK.
*/
export interface PosthogSink {
/**
* Queue a high-volume analytics event without waiting for a network
* roundtrip. Use on request hot paths where occasional process-exit loss is
* preferable to user-visible latency.
*/
captureQueued?: (input: PosthogCaptureInput) => void
capture: (input: PosthogCaptureInput) => Promise<void>
/** Flush and close the underlying client. Call on server shutdown. */
shutdown: () => Promise<void>
@@ -26,10 +32,10 @@ export interface PosthogSink {
/**
* PostHog sink for server-side product events.
*
* Uses `captureImmediate` (one HTTP roundtrip per event, no background
* queue) on purpose: the forwarded events are low-frequency business facts
* (signup, payment, subscription lifecycle) fired from webhook/auth-hook
* paths where a queued batch could be lost on process exit.
* Low-frequency conversion facts use `captureImmediate` (one HTTP roundtrip
* per event) because they terminate money/auth funnels. High-frequency AI
* 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
@@ -39,6 +45,19 @@ export function createPosthogSink(options: { projectKey: string, host: string })
const client = new PostHog(options.projectKey, { host: options.host })
return {
captureQueued(input: PosthogCaptureInput): void {
try {
client.capture({
distinctId: input.distinctId,
event: input.event,
properties: input.properties,
})
}
catch (err) {
logger.withError(err).withFields({ event: input.event }).warn('Failed to enqueue product event to PostHog')
}
},
async capture(input: PosthogCaptureInput): Promise<void> {
try {
await client.captureImmediate({
apps/server/src/services/domain/product-events.test.ts
+94
View File
@@ -163,6 +163,7 @@ describe('productEventService', () => {
event: 'payment_completed',
properties: {
app_surface: 'server',
airi_user_id: 'user-1',
feature: 'billing',
status: 'succeeded',
source: 'stripe.webhook',
@@ -175,6 +176,7 @@ describe('productEventService', () => {
event: 'signup_completed',
properties: {
app_surface: 'server',
airi_user_id: 'user-2',
feature: 'auth',
status: 'succeeded',
},
@@ -184,6 +186,50 @@ describe('productEventService', () => {
expect(rows).toHaveLength(2)
})
it('merges Stripe webhook conversions with the browser PostHog person when a distinct id is present', async () => {
const capture = vi.fn(async () => {})
const sink = { capture, shutdown: vi.fn(async () => {}) }
const service = createProductEventService(db, null, sink)
await service.track({
userId: 'user-1',
feature: 'billing',
action: 'payment_completed',
status: 'succeeded',
source: 'stripe.webhook',
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',
properties: {
$anon_distinct_id: 'anon-browser-1',
$session_id: 'ph-session-1',
airi_user_id: 'user-1',
},
})
expect(capture).toHaveBeenNthCalledWith(2, {
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',
},
})
})
it('does not forward high-volume per-request actions to PostHog', async () => {
const capture = vi.fn(async () => {})
const sink = { capture, shutdown: vi.fn(async () => {}) }
@@ -207,6 +253,54 @@ describe('productEventService', () => {
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',
roundId: 'round-1',
appSurface: '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',
provider_type: 'official',
usage_source: 'reported',
conversation_id: 'session-1',
round_id: 'round-1',
app_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:
//
apps/server/src/services/domain/product-events.ts
+79
View File
@@ -76,6 +76,27 @@ export interface ProductEventAggregateRow {
distinctUsers: number
}
export type AiGenerationAppSurface = 'server' | 'web' | 'mobile' | 'electron'
/** Content-free PostHog AI generation fact keyed to the authenticated user. */
export interface AiGenerationEventInput {
userId: string
traceId: string
generationId: string
model: string
provider: string
providerType: 'official' | 'custom' | 'unknown'
usageSource: 'reported' | 'estimated' | 'unavailable'
inputTokens?: number
outputTokens?: number
totalTokens?: number
conversationId?: string
roundId?: string
appSurface: AiGenerationAppSurface
latencySeconds?: number
stream?: boolean
}
/**
* Server-side actions worth a PostHog copy, mapped to the event name the
* client-side funnels expect. Only business facts that terminate or anchor
@@ -115,6 +136,11 @@ function metricLabels(input: ProductEventInput): Record<string, string> {
return attrs
}
function stringMetadata(input: ProductEventInput, key: string): string | undefined {
const value = input.metadata?.[key]
return typeof value === 'string' && value.length > 0 ? value : undefined
}
/**
* Creates AIRI's first-party product analytics event writer.
*
@@ -134,6 +160,42 @@ function metricLabels(input: ProductEventInput): Record<string, string> {
*/
export function createProductEventService(db: Database, metrics?: ProductMetrics | null, posthog?: PosthogSink | null) {
return {
trackGeneration(input: AiGenerationEventInput): void {
if (!posthog)
return
const event = {
distinctId: input.userId,
event: '$ai_generation',
properties: {
$ai_trace_id: input.traceId,
...(input.conversationId && { $ai_session_id: input.conversationId }),
$ai_span_id: input.generationId,
$ai_model: input.model,
$ai_provider: input.provider,
...(input.inputTokens != null && { $ai_input_tokens: input.inputTokens }),
...(input.outputTokens != null && { $ai_output_tokens: input.outputTokens }),
...(input.totalTokens != null && { $ai_total_tokens: input.totalTokens }),
...(input.latencySeconds != null && { $ai_latency: input.latencySeconds }),
...(input.stream != null && { $ai_stream: input.stream }),
$insert_id: `ai-generation:${input.generationId}`,
provider_type: input.providerType,
usage_source: input.usageSource,
...(input.conversationId && { conversation_id: input.conversationId }),
...(input.roundId && { round_id: input.roundId }),
app_surface: input.appSurface,
},
}
if (posthog.captureQueued) {
posthog.captureQueued(event)
return
}
void posthog.capture(event)
.catch(err => logger.withError(err).withFields({ generationId: input.generationId }).warn('Failed to capture PostHog AI generation'))
},
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
@@ -171,11 +233,28 @@ export function createProductEventService(db: Database, metrics?: ProductMetrics
const forwardedEvent = POSTHOG_FORWARDED_ACTIONS[input.action]
if (persisted && posthog && forwardedEvent) {
try {
const posthogDistinctId = stringMetadata(input, 'posthog_distinct_id')
const posthogSessionId = stringMetadata(input, 'posthog_session_id')
if (posthogDistinctId && posthogDistinctId !== input.userId) {
await posthog.capture({
distinctId: input.userId,
event: '$identify',
properties: {
$anon_distinct_id: posthogDistinctId,
airi_user_id: input.userId,
...(posthogSessionId && { $session_id: posthogSessionId }),
},
})
}
await posthog.capture({
distinctId: input.userId,
event: forwardedEvent,
properties: {
app_surface: 'server',
airi_user_id: input.userId,
...(posthogDistinctId && { posthog_distinct_id: posthogDistinctId }),
...(posthogSessionId && { $session_id: posthogSessionId }),
feature: input.feature,
status: input.status,
...(input.source && { source: input.source }),