refactor(analytics): replace PostHog with OpenPanel (#2480)
This commit is contained in:
RainbowBird
@@ -28,7 +28,6 @@ function createTestDeps() {
|
||||
providerCatalogService: {} as never,
|
||||
productEventService: {
|
||||
track: vi.fn(async () => undefined),
|
||||
trackGeneration: vi.fn(async () => undefined),
|
||||
} as never,
|
||||
configKV: { getOrThrow: vi.fn() } as never,
|
||||
redis: redis as never,
|
||||
|
||||
+10
-16
@@ -60,7 +60,7 @@ import { createStripeRoutes } from './routes/stripe'
|
||||
import { createVoicePackRoutes } from './routes/voice-packs'
|
||||
import { createConfigKVService } from './services/adapters/config-kv'
|
||||
import { createConfigKVStore } from './services/adapters/config-kv/store'
|
||||
import { createPosthogSink } from './services/adapters/posthog'
|
||||
import { createOpenpanelSink } from './services/adapters/openpanel'
|
||||
import { createBillingService } from './services/domain/billing/billing-service'
|
||||
import { createFluxMeter } from './services/domain/billing/flux-meter'
|
||||
import { createCharacterService } from './services/domain/characters'
|
||||
@@ -529,27 +529,21 @@ export async function createApp() {
|
||||
build: ({ dependsOn }) => createConfigKVService(createConfigKVStore(dependsOn.db, dependsOn.redis)),
|
||||
})
|
||||
|
||||
const posthogSink = injeca.provide('services:posthogSink', {
|
||||
dependsOn: { env: parsedEnv, lifecycle },
|
||||
// POSTHOG_PROJECT_KEY defaults to the shared project key, so the falsy
|
||||
// branch is only reachable via the documented off-switch: setting the
|
||||
// env var to an empty string (valibot defaults don't apply to '').
|
||||
const openpanelSink = injeca.provide('services:openpanelSink', {
|
||||
dependsOn: { env: parsedEnv },
|
||||
build: ({ dependsOn }) => {
|
||||
if (!dependsOn.env.POSTHOG_PROJECT_KEY)
|
||||
const { OPENPANEL_API_URL: apiUrl, OPENPANEL_CLIENT_ID: clientId, OPENPANEL_CLIENT_SECRET: clientSecret } = dependsOn.env
|
||||
if (!apiUrl && !clientId && !clientSecret)
|
||||
return null
|
||||
|
||||
const sink = createPosthogSink({
|
||||
projectKey: dependsOn.env.POSTHOG_PROJECT_KEY,
|
||||
host: dependsOn.env.POSTHOG_API_HOST,
|
||||
})
|
||||
dependsOn.lifecycle.appHooks.onStop(() => sink.shutdown())
|
||||
return sink
|
||||
if (!apiUrl || !clientId || !clientSecret)
|
||||
throw new Error('OpenPanel requires API URL, client id, and client secret')
|
||||
return createOpenpanelSink({ apiUrl, clientId, clientSecret })
|
||||
},
|
||||
})
|
||||
|
||||
const productEventService = injeca.provide('services:productEvents', {
|
||||
dependsOn: { posthogSink },
|
||||
build: ({ dependsOn }) => createProductEventService(dependsOn.posthogSink),
|
||||
dependsOn: { openpanelSink },
|
||||
build: ({ dependsOn }) => createProductEventService(dependsOn.openpanelSink),
|
||||
})
|
||||
|
||||
const characterService = injeca.provide('services:characters', {
|
||||
|
||||
@@ -104,13 +104,9 @@ const EnvSchema = object({
|
||||
|
||||
OTEL_TRACES_SAMPLING_RATIO: optionalNumberFromString(1, 'OTEL_TRACES_SAMPLING_RATIO', 0, 1),
|
||||
PORT: optionalIntegerFromString(3000, 'PORT', 1),
|
||||
POSTHOG_API_HOST: optional(string(), 'https://t.airi.build'),
|
||||
// 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 stage-shared/analytics/posthog), so forwarding is on out of
|
||||
// the box. Set to an empty string to disable server-side product analytics.
|
||||
POSTHOG_PROJECT_KEY: optional(string(), 'phc_pzjziJjrVZpa9SqnQqq0QEKvkmuCPH7GDTA6TbRTEf9'), // cspell:disable-line
|
||||
|
||||
OPENPANEL_API_URL: optional(string()),
|
||||
OPENPANEL_CLIENT_ID: optional(string()),
|
||||
OPENPANEL_CLIENT_SECRET: optional(string()),
|
||||
REDIS_URL: pipe(string(), nonEmpty('REDIS_URL is required')),
|
||||
STRIPE_SECRET_KEY: optional(string()),
|
||||
|
||||
|
||||
@@ -1,21 +1,20 @@
|
||||
import type { AiGenerationAppSurface } from '../../../services/domain/product-events'
|
||||
/** Client runtime declared by the chat request header. */
|
||||
export type ChatAppSurface = 'web' | 'mobile' | 'electron'
|
||||
|
||||
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'])
|
||||
const CLIENT_CHAT_ANALYTICS_SURFACES = new Set<ChatAppSurface>(['web', 'mobile', 'electron'])
|
||||
|
||||
/**
|
||||
* Resolves the product runtime from a trusted client hint.
|
||||
*
|
||||
* Unknown values are not coerced to `server`: `$ai_generation` uses
|
||||
* `capture_surface` for the process that emitted the event, while
|
||||
* `app_surface` stays reserved for the user's actual product runtime.
|
||||
* Unknown values stay absent rather than being attributed to a client runtime.
|
||||
*/
|
||||
export function resolveChatAnalyticsSurface(value: string | undefined): AiGenerationAppSurface | undefined {
|
||||
if (CLIENT_CHAT_ANALYTICS_SURFACES.has(value as AiGenerationAppSurface))
|
||||
return value as AiGenerationAppSurface
|
||||
export function resolveChatAnalyticsSurface(value: string | undefined): ChatAppSurface | undefined {
|
||||
if (CLIENT_CHAT_ANALYTICS_SURFACES.has(value as ChatAppSurface))
|
||||
return value as ChatAppSurface
|
||||
|
||||
return undefined
|
||||
}
|
||||
|
||||
@@ -1,6 +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 { ChatAppSurface } from '../../analytics'
|
||||
import type { GatewayCallback } from '../../gateway'
|
||||
import type { V1RouteDeps } from '../../types'
|
||||
|
||||
@@ -22,24 +22,10 @@ export interface ChatCompletionsOperationRequest {
|
||||
body: Record<string, unknown>
|
||||
sessionId?: string
|
||||
roundId?: string
|
||||
appSurface?: AiGenerationAppSurface
|
||||
appSurface?: ChatAppSurface
|
||||
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({
|
||||
@@ -157,11 +143,7 @@ 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,
|
||||
@@ -178,11 +160,7 @@ 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,
|
||||
@@ -196,37 +174,6 @@ interface ChatModelAliasPlan {
|
||||
modelIds: string[]
|
||||
}
|
||||
|
||||
function captureGeneration(input: GenerationCaptureInput): void {
|
||||
const generationId = input.roundId ?? input.requestId
|
||||
const conversationId = input.sessionId ?? 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: conversationId,
|
||||
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,
|
||||
costUsdSource: 'unavailable',
|
||||
conversationId,
|
||||
conversationIdSource: input.sessionId ? 'client_header' : 'server_request',
|
||||
roundId: generationId,
|
||||
...(input.appSurface && { appSurface: input.appSurface }),
|
||||
captureSurface: 'server',
|
||||
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')
|
||||
@@ -323,11 +270,7 @@ 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
|
||||
@@ -431,20 +374,6 @@ 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.
|
||||
@@ -514,11 +443,7 @@ 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
|
||||
@@ -552,20 +477,6 @@ 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
|
||||
|
||||
@@ -148,7 +148,6 @@ function createMockLlmRouter(impl?: Partial<LlmRouterService>): LlmRouterService
|
||||
function createMockProductEventService(): ProductEventService {
|
||||
return {
|
||||
track: vi.fn(async () => undefined),
|
||||
trackGeneration: vi.fn(async () => undefined),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -912,7 +911,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('records Langfuse and PostHog generations with authoritative usage and correlation', async () => {
|
||||
it('records Langfuse usage without forwarding generations to product analytics', async () => {
|
||||
const llmRouter = createMockLlmRouter({
|
||||
route: vi.fn(async (_req, ctx) => {
|
||||
if (ctx) {
|
||||
@@ -962,88 +961,8 @@ 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,
|
||||
costUsdSource: 'unavailable',
|
||||
conversationId: 'conversation-1',
|
||||
conversationIdSource: 'client_header',
|
||||
roundId: 'round-1',
|
||||
appSurface: 'electron',
|
||||
captureSurface: 'server',
|
||||
latencySeconds: expect.any(Number),
|
||||
stream: false,
|
||||
})
|
||||
})
|
||||
|
||||
it('uses request-level correlation for server-captured generations without chat headers', async () => {
|
||||
const llmRouter = createMockLlmRouter({
|
||||
route: vi.fn(async (_req, ctx) => {
|
||||
if (ctx) {
|
||||
ctx.provider = 'openrouter'
|
||||
ctx.upstreamModel = 'openai/gpt-4o-mini'
|
||||
}
|
||||
return new Response(JSON.stringify({
|
||||
choices: [],
|
||||
usage: { prompt_tokens: 1, completion_tokens: 2, total_tokens: 3 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
})
|
||||
}) as any,
|
||||
})
|
||||
const productEventService = createMockProductEventService()
|
||||
const app = createTestApp(
|
||||
createMockFluxService(),
|
||||
createMockConfigKV(),
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
llmRouter,
|
||||
createMockLlmTracing(),
|
||||
productEventService,
|
||||
)
|
||||
|
||||
await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/chat/completions', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'chat-auto', messages: [{ role: 'user', content: 'hi' }] }),
|
||||
}),
|
||||
{ user: testUser } as any,
|
||||
)
|
||||
|
||||
expect(productEventService.trackGeneration).toHaveBeenCalledWith({
|
||||
userId: 'user-1',
|
||||
traceId: expect.any(String),
|
||||
generationId: expect.any(String),
|
||||
model: 'openai/gpt-4o-mini',
|
||||
provider: 'openrouter',
|
||||
providerType: 'official',
|
||||
usageSource: 'reported',
|
||||
inputTokens: 1,
|
||||
outputTokens: 2,
|
||||
totalTokens: 3,
|
||||
costUsdSource: 'unavailable',
|
||||
conversationId: expect.any(String),
|
||||
conversationIdSource: 'server_request',
|
||||
roundId: expect.any(String),
|
||||
captureSurface: 'server',
|
||||
latencySeconds: expect.any(Number),
|
||||
stream: false,
|
||||
})
|
||||
const generation = vi.mocked(productEventService.trackGeneration).mock.calls[0]?.[0]
|
||||
expect(generation?.traceId).toBe(generation?.conversationId)
|
||||
expect(generation?.roundId).toBe(generation?.generationId)
|
||||
expect(generation).not.toHaveProperty('appSurface')
|
||||
expect(llmTracing.startChatGeneration).toHaveBeenCalledWith(expect.objectContaining({ sessionId: 'conversation-1' }))
|
||||
expect(productEventService.track).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('should not charge flux when upstream returns error', async () => {
|
||||
|
||||
@@ -33,7 +33,7 @@ export interface CheckoutOperationInput {
|
||||
request: Request
|
||||
}
|
||||
|
||||
interface PosthogIdentityHeaders {
|
||||
interface OpenpanelIdentityHeaders {
|
||||
distinctId?: string
|
||||
sessionId?: string
|
||||
}
|
||||
@@ -80,7 +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 openpanelIdentity = readOpenpanelIdentityHeaders(input.request)
|
||||
|
||||
const sessionParams: CheckoutSessionCreateParams = {
|
||||
line_items: [{ price: stripePriceId, quantity: 1 }],
|
||||
@@ -93,8 +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 }),
|
||||
...(openpanelIdentity.distinctId && { openpanelDeviceId: openpanelIdentity.distinctId }),
|
||||
...(openpanelIdentity.sessionId && { openpanelSessionId: openpanelIdentity.sessionId }),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -142,8 +142,8 @@ 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 }),
|
||||
...(openpanelIdentity.distinctId && { openpanel_device_id: openpanelIdentity.distinctId }),
|
||||
...(openpanelIdentity.sessionId && { openpanel_session_id: openpanelIdentity.sessionId }),
|
||||
},
|
||||
})
|
||||
|
||||
@@ -151,9 +151,9 @@ export function createCheckoutOperation(deps: CheckoutOperationDeps) {
|
||||
}
|
||||
}
|
||||
|
||||
function readPosthogIdentityHeaders(request: Request): PosthogIdentityHeaders {
|
||||
const distinctId = readStripeMetadataHeader(request, 'x-posthog-distinct-id')
|
||||
const sessionId = readStripeMetadataHeader(request, 'x-posthog-session-id')
|
||||
function readOpenpanelIdentityHeaders(request: Request): OpenpanelIdentityHeaders {
|
||||
const distinctId = readStripeMetadataHeader(request, 'x-openpanel-device-id')
|
||||
const sessionId = readStripeMetadataHeader(request, 'x-openpanel-session-id')
|
||||
return {
|
||||
...(distinctId && { distinctId }),
|
||||
...(sessionId && { sessionId }),
|
||||
|
||||
@@ -89,8 +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
|
||||
const openpanelDeviceId = event.data.object.metadata?.openpanelDeviceId
|
||||
const openpanelSessionId = event.data.object.metadata?.openpanelSessionId
|
||||
void deps.productEventService?.track({
|
||||
userId,
|
||||
feature: 'billing',
|
||||
@@ -104,8 +104,8 @@ export function createWebhookOperation(deps: WebhookOperationDeps) {
|
||||
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 }),
|
||||
...(openpanelDeviceId && { openpanel_device_id: openpanelDeviceId }),
|
||||
...(openpanelSessionId && { openpanel_session_id: openpanelSessionId }),
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -228,6 +228,9 @@ async function handleCheckoutSessionCompleted(
|
||||
applied: result.applied,
|
||||
balanceAfter: result.balanceAfter,
|
||||
}).log('Processed flux credit for one-time payment')
|
||||
|
||||
// Only the transaction that credits this checkout emits its conversion.
|
||||
return { processed: result.applied }
|
||||
}
|
||||
|
||||
return { processed: true }
|
||||
|
||||
@@ -296,7 +296,7 @@ describe('stripeRoutes', () => {
|
||||
expect(res.status).toBe(503)
|
||||
})
|
||||
|
||||
it('stores browser PostHog identity in Stripe checkout metadata', async () => {
|
||||
it('stores browser OpenPanel identity in Stripe checkout metadata', async () => {
|
||||
const createSession = vi.fn(async input => ({
|
||||
id: 'cs_1',
|
||||
url: 'https://checkout.stripe.com/cs_1',
|
||||
@@ -343,8 +343,8 @@ describe('stripeRoutes', () => {
|
||||
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',
|
||||
'x-openpanel-device-id': 'anon-browser-1',
|
||||
'x-openpanel-session-id': 'ph-session-1',
|
||||
},
|
||||
}),
|
||||
})
|
||||
@@ -353,16 +353,16 @@ describe('stripeRoutes', () => {
|
||||
metadata: {
|
||||
userId: 'user-1',
|
||||
fluxAmount: '500',
|
||||
posthogDistinctId: 'anon-browser-1',
|
||||
posthogSessionId: 'ph-session-1',
|
||||
openpanelDeviceId: 'anon-browser-1',
|
||||
openpanelSessionId: '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',
|
||||
openpanel_device_id: 'anon-browser-1',
|
||||
openpanel_session_id: 'ph-session-1',
|
||||
}),
|
||||
}))
|
||||
})
|
||||
@@ -535,7 +535,7 @@ describe('stripeRoutes', () => {
|
||||
expect(res.status).toBe(503)
|
||||
})
|
||||
|
||||
it('records payment completion with Stripe and PostHog identity from checkout metadata', async () => {
|
||||
it('records payment completion with Stripe and OpenPanel identity from checkout metadata', async () => {
|
||||
const checkoutEvent = {
|
||||
id: 'evt_checkout_completed',
|
||||
type: 'checkout.session.completed',
|
||||
@@ -556,8 +556,8 @@ describe('stripeRoutes', () => {
|
||||
metadata: {
|
||||
userId: 'user-1',
|
||||
fluxAmount: '500',
|
||||
posthogDistinctId: 'anon-browser-1',
|
||||
posthogSessionId: 'ph-session-1',
|
||||
openpanelDeviceId: 'anon-browser-1',
|
||||
openpanelSessionId: 'ph-session-1',
|
||||
},
|
||||
expires_at: null,
|
||||
},
|
||||
@@ -599,10 +599,14 @@ describe('stripeRoutes', () => {
|
||||
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',
|
||||
openpanel_device_id: 'anon-browser-1',
|
||||
openpanel_session_id: 'ph-session-1',
|
||||
},
|
||||
})
|
||||
|
||||
vi.mocked(billingService.creditFluxFromStripeCheckout).mockResolvedValueOnce({ applied: false })
|
||||
await webhook({ signature: 'test_sig', body: '{}' })
|
||||
expect(productEventService.track).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('processes subscription lifecycle webhooks without product events', async () => {
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
import { createServer } from 'node:http'
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { createOpenpanelSink } from './openpanel'
|
||||
|
||||
describe('openPanel sink', () => {
|
||||
it('isolates concurrent user identities and preserves checkout device attribution', async () => {
|
||||
const bodies: string[] = []
|
||||
const server = createServer(async (request, response) => {
|
||||
let body = ''
|
||||
for await (const chunk of request)
|
||||
body += chunk
|
||||
bodies.push(body)
|
||||
response.writeHead(200).end('{}')
|
||||
})
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string')
|
||||
throw new Error('Expected a TCP server address')
|
||||
try {
|
||||
const sink = createOpenpanelSink({ apiUrl: `http://127.0.0.1:${address.port}`, clientId: 'test-client', clientSecret: 'test-secret' })
|
||||
await Promise.all([
|
||||
sink.capture({ userId: 'alice', event: 'payment_completed', deviceId: 'alice-device', properties: { event_id: 'cs_alice' } }),
|
||||
sink.capture({ userId: 'bob', event: 'signup_completed', properties: {} }),
|
||||
])
|
||||
expect(bodies.map(body => JSON.parse(body))).toEqual(expect.arrayContaining([
|
||||
{ type: 'track', payload: { name: 'payment_completed', profileId: 'alice', properties: { event_id: 'cs_alice', __deviceId: 'alice-device' } } },
|
||||
{ type: 'track', payload: { name: 'signup_completed', profileId: 'bob', properties: {} } },
|
||||
]))
|
||||
}
|
||||
finally {
|
||||
server.closeAllConnections()
|
||||
await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()))
|
||||
}
|
||||
})
|
||||
|
||||
it('does not retry a rejected conversion request', async () => {
|
||||
let requests = 0
|
||||
const server = createServer((_request, response) => {
|
||||
requests++
|
||||
response.writeHead(503).end('Unavailable')
|
||||
})
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string')
|
||||
throw new Error('Expected a TCP server address')
|
||||
try {
|
||||
const sink = createOpenpanelSink({ apiUrl: `http://127.0.0.1:${address.port}`, clientId: 'test-client', clientSecret: 'test-secret' })
|
||||
await expect(sink.capture({ userId: 'alice', event: 'payment_completed', properties: {} })).rejects.toThrow('503')
|
||||
expect(requests).toBe(1)
|
||||
}
|
||||
finally {
|
||||
server.closeAllConnections()
|
||||
await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()))
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,53 @@
|
||||
import type { TrackHandlerPayload } from '@openpanel/sdk'
|
||||
|
||||
/** One confirmed product fact, with an optional browser device for attribution. */
|
||||
export interface ProductCaptureInput {
|
||||
userId: string
|
||||
event: string
|
||||
properties: Record<string, unknown>
|
||||
deviceId?: string
|
||||
}
|
||||
|
||||
/** External delivery boundary for confirmed product facts. */
|
||||
export interface ProductAnalyticsSink {
|
||||
capture: (input: ProductCaptureInput) => Promise<void>
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends each fact once. Callers own replay suppression through business state.
|
||||
* Each request supplies its user id, so concurrent users never share SDK identity.
|
||||
*/
|
||||
export function createOpenpanelSink(options: { clientId: string, clientSecret: string, apiUrl: string }): ProductAnalyticsSink {
|
||||
const endpoint = `${options.apiUrl.replace(/\/$/, '')}/track`
|
||||
|
||||
return {
|
||||
async capture(input) {
|
||||
const payload: TrackHandlerPayload = {
|
||||
type: 'track',
|
||||
payload: {
|
||||
name: input.event,
|
||||
profileId: input.userId,
|
||||
properties: {
|
||||
...input.properties,
|
||||
...(input.deviceId && { __deviceId: input.deviceId }),
|
||||
},
|
||||
},
|
||||
}
|
||||
// OpenPanel does not deduplicate events by a caller-supplied UUID. Retrying an
|
||||
// ambiguous response can count a paid conversion twice.
|
||||
const response = await fetch(endpoint, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'content-type': 'application/json',
|
||||
'openpanel-client-id': options.clientId,
|
||||
'openpanel-client-secret': options.clientSecret,
|
||||
'openpanel-sdk-name': 'node',
|
||||
},
|
||||
body: JSON.stringify(payload),
|
||||
signal: AbortSignal.timeout(5000),
|
||||
})
|
||||
if (!response.ok)
|
||||
throw new Error(`OpenPanel rejected the product event (${response.status})`)
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1,81 +0,0 @@
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { PostHog } from 'posthog-node'
|
||||
|
||||
const logger = useLogger('posthog')
|
||||
|
||||
/**
|
||||
* One product event forwarded to PostHog, keyed by the Better Auth user id
|
||||
* so it merges with the browser person identified via `posthog.identify()`.
|
||||
*/
|
||||
export interface PosthogCaptureInput {
|
||||
distinctId: string
|
||||
event: string
|
||||
properties: Record<string, unknown>
|
||||
/** Stable event UUID used by PostHog ingestion for replay deduplication. */
|
||||
uuid?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimal capture boundary the product-events service depends on. Kept as
|
||||
* 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>
|
||||
}
|
||||
|
||||
/**
|
||||
* PostHog sink for server-side product events.
|
||||
*
|
||||
* 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 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 })
|
||||
|
||||
return {
|
||||
captureQueued(input: PosthogCaptureInput): void {
|
||||
try {
|
||||
client.capture({
|
||||
distinctId: input.distinctId,
|
||||
event: input.event,
|
||||
properties: input.properties,
|
||||
...(input.uuid && { uuid: input.uuid }),
|
||||
})
|
||||
}
|
||||
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({
|
||||
distinctId: input.distinctId,
|
||||
event: input.event,
|
||||
properties: input.properties,
|
||||
...(input.uuid && { uuid: input.uuid }),
|
||||
})
|
||||
}
|
||||
catch (err) {
|
||||
logger.withError(err).withFields({ event: input.event }).warn('Failed to forward product event to PostHog')
|
||||
}
|
||||
},
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
await client.shutdown()
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1,181 +1,67 @@
|
||||
import type { ProductCaptureInput } from '../adapters/openpanel'
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createProductEventService } from './product-events'
|
||||
|
||||
describe('productEventService', () => {
|
||||
it('captures only the server-side funnel facts shared with the Go service', async () => {
|
||||
const capture = vi.fn(async () => {})
|
||||
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',
|
||||
action: 'payment_completed',
|
||||
status: 'succeeded',
|
||||
source: 'stripe.webhook',
|
||||
metadata: { amount_minor_unit: 990, currency: 'usd' },
|
||||
})
|
||||
it('routes confirmed product facts to OpenPanel with stable AIRI user ids', async () => {
|
||||
const capture = vi.fn<(input: ProductCaptureInput) => Promise<void>>().mockResolvedValue(undefined)
|
||||
const service = createProductEventService({ capture })
|
||||
await service.track({ userId: 'user-1', feature: 'auth', action: 'user_signed_up', status: 'succeeded' })
|
||||
await service.track({ userId: 'user-2', feature: 'billing', action: 'checkout_started', status: 'succeeded' })
|
||||
|
||||
expect(capture).toHaveBeenNthCalledWith(1, {
|
||||
distinctId: 'user-1',
|
||||
userId: '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: {
|
||||
app_surface: 'server',
|
||||
airi_user_id: 'user-1',
|
||||
feature: 'billing',
|
||||
status: 'succeeded',
|
||||
source: 'stripe.webhook',
|
||||
amount_minor_unit: 990,
|
||||
currency: 'usd',
|
||||
},
|
||||
properties: { app_surface: 'server', airi_user_id: 'user-1', feature: 'auth', status: 'succeeded' },
|
||||
})
|
||||
expect(capture).toHaveBeenNthCalledWith(2, expect.objectContaining({ userId: 'user-2', event: 'checkout_created' }))
|
||||
})
|
||||
|
||||
it('merges a Stripe conversion with its browser PostHog person', async () => {
|
||||
const capture = vi.fn(async () => {})
|
||||
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
|
||||
|
||||
it('passes the checkout device id and source event id without an extra identify event', async () => {
|
||||
const capture = vi.fn<(input: ProductCaptureInput) => Promise<void>>().mockResolvedValue(undefined)
|
||||
const service = createProductEventService({ capture })
|
||||
await service.track({
|
||||
userId: 'user-1',
|
||||
feature: 'billing',
|
||||
action: 'payment_completed',
|
||||
status: 'succeeded',
|
||||
eventId: 'cs_123',
|
||||
metadata: {
|
||||
posthog_distinct_id: 'anon-browser-1',
|
||||
posthog_session_id: 'ph-session-1',
|
||||
},
|
||||
metadata: { openpanel_device_id: 'browser-1', amount_total: 990, currency: 'usd' },
|
||||
})
|
||||
|
||||
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.objectContaining({
|
||||
distinctId: 'user-1',
|
||||
event: 'payment_completed',
|
||||
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('rejects metadata that can overwrite service-controlled PostHog properties', async () => {
|
||||
const capture = vi.fn(async () => {})
|
||||
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
|
||||
|
||||
await expect(service.track({
|
||||
expect(capture).toHaveBeenCalledTimes(1)
|
||||
expect(capture).toHaveBeenCalledWith({
|
||||
userId: 'user-1',
|
||||
feature: 'billing',
|
||||
action: 'payment_completed',
|
||||
status: 'succeeded',
|
||||
metadata: { $insert_id: 'spoofed' },
|
||||
})).resolves.toBeUndefined()
|
||||
deviceId: 'browser-1',
|
||||
event: 'payment_completed',
|
||||
properties: {
|
||||
openpanel_device_id: 'browser-1',
|
||||
amount_total: 990,
|
||||
currency: 'usd',
|
||||
event_id: 'cs_123',
|
||||
app_surface: 'server',
|
||||
airi_user_id: 'user-1',
|
||||
feature: 'billing',
|
||||
status: 'succeeded',
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects metadata that can forge provider identity or the source event id', async () => {
|
||||
const capture = vi.fn(async () => {})
|
||||
const service = createProductEventService({ capture })
|
||||
for (const key of ['event_id', '__deviceId', 'profileId']) {
|
||||
await service.track({ userId: 'user-1', feature: 'billing', action: 'payment_completed', status: 'succeeded', metadata: { [key]: 'forged' } })
|
||||
}
|
||||
expect(capture).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
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 () => {
|
||||
it('does not fail the business path when OpenPanel is unavailable', async () => {
|
||||
const capture = vi.fn(async () => {
|
||||
throw new Error('posthog exploded')
|
||||
throw new Error('unavailable')
|
||||
})
|
||||
const service = createProductEventService({ capture, shutdown: vi.fn(async () => {}) })
|
||||
|
||||
await expect(service.track({
|
||||
userId: 'user-1',
|
||||
feature: 'billing',
|
||||
action: 'payment_completed',
|
||||
status: 'succeeded',
|
||||
})).resolves.toBeUndefined()
|
||||
const service = createProductEventService({ capture })
|
||||
await expect(service.track({ userId: 'user-1', feature: 'billing', action: 'payment_completed', status: 'succeeded' })).resolves.toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
import type { PosthogSink } from '../adapters/posthog'
|
||||
|
||||
import { createHash } from 'node:crypto'
|
||||
import type { ProductAnalyticsSink } from '../adapters/openpanel'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
|
||||
const logger = useLogger('product-events')
|
||||
|
||||
const RESERVED_POSTHOG_METADATA_KEYS = new Set([
|
||||
'$insert_id',
|
||||
const RESERVED_PRODUCT_METADATA_KEYS = new Set([
|
||||
'event_id',
|
||||
'__deviceId',
|
||||
'__identify',
|
||||
'profileId',
|
||||
'$session_id',
|
||||
'airi_user_id',
|
||||
'app_surface',
|
||||
@@ -27,9 +28,9 @@ export type ProductAction
|
||||
| 'checkout_started'
|
||||
| 'payment_completed'
|
||||
|
||||
/** Product funnel fact forwarded to PostHog from the server. */
|
||||
/** Product funnel fact forwarded to OpenPanel from the server. */
|
||||
export interface ProductEventInput {
|
||||
/** Better Auth user id. Kept in Postgres only; never emitted as a Prometheus label. */
|
||||
/** Authenticated user id used for product attribution. Never emitted as a Prometheus label. */
|
||||
userId: string
|
||||
/** Bounded product area used for product dashboards and funnels. */
|
||||
feature: ProductFeature
|
||||
@@ -41,58 +42,19 @@ export interface ProductEventInput {
|
||||
source?: string
|
||||
/** Optional primitive metadata for product analysis. Avoid PII and raw prompts. */
|
||||
metadata?: ProductEventMetadata
|
||||
/** Stable source event id used by PostHog for replay-safe deduplication. */
|
||||
/** Stable source event id for reconciliation. Callers suppress replays before capture. */
|
||||
eventId?: string
|
||||
}
|
||||
|
||||
/** Product runtime where the user initiated the AI generation. */
|
||||
export type AiGenerationAppSurface = 'web' | 'mobile' | 'electron'
|
||||
|
||||
/** Runtime that captured the `$ai_generation` fact. */
|
||||
export type AiGenerationCaptureSurface = 'server' | 'client'
|
||||
|
||||
/** Explains whether `conversation_id` is an app conversation or a server fallback. */
|
||||
export type AiGenerationConversationIdSource = 'client_header' | 'server_request'
|
||||
|
||||
/** Explains whether AIRI supplied a trustworthy USD cost for this generation. */
|
||||
export type AiGenerationCostUsdSource = 'reported' | 'estimated' | 'unavailable'
|
||||
|
||||
/** 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
|
||||
totalCostUsd?: number
|
||||
costUsdSource?: AiGenerationCostUsdSource
|
||||
/** Always present for joins; `conversationIdSource` tells whether it is request-level fallback. */
|
||||
conversationId: string
|
||||
/** Distinguishes real client conversation ids from server-generated request fallbacks. */
|
||||
conversationIdSource: AiGenerationConversationIdSource
|
||||
roundId?: string
|
||||
/** Omitted when the server cannot determine the user's product runtime. */
|
||||
appSurface?: AiGenerationAppSurface
|
||||
/** Defaults to `server` because this service runs in the API process. */
|
||||
captureSurface?: AiGenerationCaptureSurface
|
||||
latencySeconds?: number
|
||||
stream?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Server-side actions that anchor a PostHog product funnel. Per-request LLM
|
||||
* Server-side actions that anchor the 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
|
||||
* auth UI progress uses `signup_form_completed` and never reuses this name.
|
||||
*/
|
||||
const POSTHOG_FORWARDED_ACTIONS: Partial<Record<ProductAction, string>> = {
|
||||
const FORWARDED_ACTIONS: Partial<Record<ProductAction, string>> = {
|
||||
user_signed_up: 'signup_completed',
|
||||
checkout_started: 'checkout_created',
|
||||
payment_completed: 'payment_completed',
|
||||
@@ -103,20 +65,12 @@ 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))
|
||||
return metadata != null && Object.keys(metadata).some(key => RESERVED_PRODUCT_METADATA_KEYS.has(key))
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates AIRI's server-side PostHog product analytics writer.
|
||||
* Sends confirmed product facts to OpenPanel.
|
||||
*
|
||||
* Use when:
|
||||
* - A server has an authenticated user id and confirms a funnel fact.
|
||||
@@ -128,102 +82,38 @@ function hasReservedMetadataKey(metadata: ProductEventMetadata | undefined): boo
|
||||
* Returns:
|
||||
* - A best-effort event writer. Capture errors never change the business flow.
|
||||
*/
|
||||
export function createProductEventService(posthog?: PosthogSink | null) {
|
||||
export function createProductEventService(sink?: ProductAnalyticsSink | null) {
|
||||
return {
|
||||
trackGeneration(input: AiGenerationEventInput): void {
|
||||
if (!posthog)
|
||||
return
|
||||
|
||||
const event = {
|
||||
distinctId: input.userId,
|
||||
event: '$ai_generation',
|
||||
properties: {
|
||||
$ai_trace_id: input.traceId,
|
||||
$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.totalCostUsd != null && { $ai_total_cost_usd: input.totalCostUsd }),
|
||||
...(input.latencySeconds != null && { $ai_latency: input.latencySeconds }),
|
||||
...(input.stream != null && { $ai_stream: input.stream }),
|
||||
$insert_id: `ai-generation:${input.generationId}`,
|
||||
airi_user_id: input.userId,
|
||||
provider_type: input.providerType,
|
||||
usage_source: input.usageSource,
|
||||
token_usage_available: input.usageSource !== 'unavailable',
|
||||
cost_usd_source: input.costUsdSource ?? 'unavailable',
|
||||
cost_usd_known: input.totalCostUsd != null,
|
||||
conversation_id: input.conversationId,
|
||||
conversation_id_source: input.conversationIdSource,
|
||||
...(input.roundId && { round_id: input.roundId }),
|
||||
...(input.appSurface && { app_surface: input.appSurface }),
|
||||
capture_surface: input.captureSurface ?? 'server',
|
||||
},
|
||||
}
|
||||
|
||||
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> {
|
||||
const forwardedEvent = POSTHOG_FORWARDED_ACTIONS[input.action]
|
||||
if (!posthog || !forwardedEvent)
|
||||
const forwardedEvent = FORWARDED_ACTIONS[input.action]
|
||||
if (!sink || !forwardedEvent)
|
||||
return
|
||||
|
||||
if (hasReservedMetadataKey(input.metadata)) {
|
||||
logger.withFields({ action: input.action }).warn('Rejected reserved PostHog product event metadata')
|
||||
logger.withFields({ action: input.action }).warn('Rejected reserved 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')
|
||||
}
|
||||
}
|
||||
const deviceId = stringMetadata(input, 'openpanel_device_id')
|
||||
|
||||
try {
|
||||
await posthog.capture({
|
||||
distinctId: input.userId,
|
||||
await sink.capture({
|
||||
userId: input.userId,
|
||||
...(deviceId && { deviceId }),
|
||||
event: forwardedEvent,
|
||||
properties: {
|
||||
...input.metadata,
|
||||
...(input.eventId && { $insert_id: input.eventId }),
|
||||
...(input.eventId && { event_id: input.eventId }),
|
||||
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 }),
|
||||
},
|
||||
...(input.eventId && { uuid: posthogEventUuid(forwardedEvent, input.eventId) }),
|
||||
})
|
||||
}
|
||||
catch (err) {
|
||||
logger.withError(err).withFields({ action: input.action }).warn('PostHog product analytics capture failed')
|
||||
logger.withError(err).withFields({ action: input.action }).warn('OpenPanel product analytics capture failed')
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user