diff --git a/.changeset/hip-planets-visit.md b/.changeset/hip-planets-visit.md new file mode 100644 index 0000000000..f0072a00a0 --- /dev/null +++ b/.changeset/hip-planets-visit.md @@ -0,0 +1,6 @@ +--- +'@posthog/core': patch +'posthog-node': patch +--- + +Add an internal event-channel mechanism so `$ai_*` events can be routed to a dedicated capture endpoint in their own batch, independent of analytics events. Gated behind the unstable, internal-only `_internal_dedicatedAiEndpoint` option on `posthog-node` — not ready for general use. diff --git a/packages/core/src/posthog-core-stateless.ts b/packages/core/src/posthog-core-stateless.ts index 9705c19016..04ab2e505e 100644 --- a/packages/core/src/posthog-core-stateless.ts +++ b/packages/core/src/posthog-core-stateless.ts @@ -137,6 +137,16 @@ export enum QuotaLimitedFeature { Recordings = 'recordings', } +export type EventChannel = { + queueKey: PostHogPersistedProperty + path: string +} + +export const DEFAULT_EVENT_CHANNEL: EventChannel = { + queueKey: PostHogPersistedProperty.Queue, + path: '/batch/', +} + export abstract class PostHogCoreStateless { // options readonly apiKey: string @@ -959,6 +969,23 @@ export abstract class PostHogCoreStateless { // Default: no-op for sync storage implementations } + /** + * The set of channels this client flushes. Override to add destinations; the + * returned list must include every channel {@link channelForEvent} can return, + * otherwise events routed to a missing channel would never flush. + */ + protected eventChannels(): readonly EventChannel[] { + return [DEFAULT_EVENT_CHANNEL] + } + + /** + * Routes an event to its ingestion channel by name. Default sends everything + * to the main analytics channel; override to partition (e.g. `$ai_*`). + */ + protected channelForEvent(_event: unknown): EventChannel { + return DEFAULT_EVENT_CHANNEL + } + protected enqueue(type: string, _message: any, options?: PostHogCaptureOptions): void { this.wrap(() => { if (this.optedOut) { @@ -974,7 +1001,8 @@ export abstract class PostHogCoreStateless { return } - const queue = this.getPersistedProperty(PostHogPersistedProperty.Queue) || [] + const queueKey = this.channelForEvent(message.event).queueKey + const queue = this.getPersistedProperty(queueKey) || [] if (queue.length >= this.maxQueueSize) { queue.shift() @@ -982,7 +1010,7 @@ export abstract class PostHogCoreStateless { } queue.push({ message }) - this.setPersistedProperty(PostHogPersistedProperty.Queue, queue) + this.setPersistedProperty(queueKey, queue) this._events.emit(type, message) @@ -1032,7 +1060,7 @@ export abstract class PostHogCoreStateless { const payload = JSON.stringify(data) - const url = `${this.host}/batch/` + const url = this.host + this.channelForEvent(message.event).path const gzippedPayload = !this.disableCompression ? await gzipCompress(payload, this.isDebug) : null const fetchOptions: PostHogFetchOptions = { @@ -1170,7 +1198,23 @@ export abstract class PostHogCoreStateless { this.clearFlushTimer() await this._initPromise - let queue = this.getPersistedProperty(PostHogPersistedProperty.Queue) || [] + // Guarantee all flushes run but errors are still thrown at the end + let firstError: unknown + for (const channel of this.eventChannels()) { + try { + await this._flushChannel(channel) + } catch (err) { + firstError ??= err + } + } + + if (firstError !== undefined) { + throw firstError + } + } + + private async _flushChannel(channel: EventChannel): Promise { + let queue = this.getPersistedProperty(channel.queueKey) || [] if (!queue.length) { return @@ -1184,9 +1228,9 @@ export abstract class PostHogCoreStateless { const batchMessages = batchItems.map((item) => item.message) const persistQueueChange = async (): Promise => { - const refreshedQueue = this.getPersistedProperty(PostHogPersistedProperty.Queue) || [] + const refreshedQueue = this.getPersistedProperty(channel.queueKey) || [] const newQueue = refreshedQueue.slice(batchItems.length) - this.setPersistedProperty(PostHogPersistedProperty.Queue, newQueue) + this.setPersistedProperty(channel.queueKey, newQueue) queue = newQueue // Wait for storage to complete to prevent duplicate events on app crash await this.flushStorage() @@ -1204,7 +1248,7 @@ export abstract class PostHogCoreStateless { const payload = JSON.stringify(data) - const url = `${this.host}/batch/` + const url = this.host + channel.path const gzippedPayload = !this.disableCompression ? await gzipCompress(payload, this.isDebug) : null const fetchOptions: PostHogFetchOptions = { @@ -1383,9 +1427,11 @@ export abstract class PostHogCoreStateless { await this.promiseQueue.join() while (true) { - const queue = this.getPersistedProperty(PostHogPersistedProperty.Queue) || [] + const pending = this.eventChannels().some( + (channel) => (this.getPersistedProperty(channel.queueKey) || []).length > 0 + ) - if (queue.length === 0) { + if (!pending) { break } diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index df3eae22b6..c5130ca3b9 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -220,6 +220,9 @@ export enum PostHogPersistedProperty { BootstrapFeatureFlagPayloads = 'bootstrap_feature_flag_payloads', OverrideFeatureFlags = 'override_feature_flags', Queue = 'queue', + // AI events queue. `$ai_*` events are routed here so they flush to a dedicated + // ingestion path independently of the main analytics queue. Only used by posthog-node. + AiQueue = 'ai_queue', // Logs queue. Individual SDKs may route this key to an isolated storage // instance if they want to separate logs write volume from main state. LogsQueue = 'logs_queue', diff --git a/packages/node/src/__tests__/dedicated-ai-endpoint.spec.ts b/packages/node/src/__tests__/dedicated-ai-endpoint.spec.ts new file mode 100644 index 0000000000..6d888023b4 --- /dev/null +++ b/packages/node/src/__tests__/dedicated-ai-endpoint.spec.ts @@ -0,0 +1,79 @@ +import { PostHog, PostHogOptions } from '@/entrypoints/index.node' + +const mockedFetch = jest.spyOn(globalThis, 'fetch').mockImplementation() + +const HOST = 'http://example.com' +const ANALYTICS_URL = `${HOST}/batch/` +const AI_URL = `${HOST}/i/v0/ai/batch/` + +const okResponse = { + status: 200, + text: () => Promise.resolve('ok'), + json: () => Promise.resolve({ status: 'ok' }), +} as any + +// Exact-match — AI_URL contains "/batch/" as a substring, so substring matching would conflate them. +const callsTo = (url: string): any[] => mockedFetch.mock.calls.filter((c) => c[0] === url) + +const batchSentTo = (url: string): any[] | undefined => { + const call = callsTo(url).at(-1) + return call ? JSON.parse((call[1] as any).body).batch : undefined +} + +const newClient = (options: Partial = {}): PostHog => + new PostHog('TEST_API_KEY', { host: HOST, fetchRetryCount: 0, disableCompression: true, ...options }) + +describe('PostHog Node.js — dedicated AI endpoint (_internal_dedicatedAiEndpoint)', () => { + let errorSpy: jest.SpyInstance + + beforeEach(() => { + errorSpy = jest.spyOn(console, 'error').mockImplementation(() => {}) + mockedFetch.mockResolvedValue(okResponse) + }) + + afterEach(() => { + mockedFetch.mockReset() + errorSpy.mockRestore() + }) + + describe('when enabled', () => { + it('routes batched $ai_* events to the dedicated AI path', async () => { + const ph = newClient({ _internal_dedicatedAiEndpoint: true }) + ph.capture({ distinctId: 'u1', event: '$ai_generation', properties: { $ai_model: 'gpt-4' } }) + await ph.shutdown() + + expect(batchSentTo(AI_URL)?.map((e: any) => e.event)).toEqual(['$ai_generation']) + expect(callsTo(ANALYTICS_URL)).toHaveLength(0) + }) + + it('keeps non-AI events on the normal path, in a separate batch from AI events', async () => { + const ph = newClient({ _internal_dedicatedAiEndpoint: true }) + ph.capture({ distinctId: 'u1', event: '$ai_generation', properties: {} }) + ph.capture({ distinctId: 'u1', event: 'button_clicked', properties: {} }) + await ph.shutdown() + + expect(batchSentTo(AI_URL)?.map((e: any) => e.event)).toEqual(['$ai_generation']) + expect(batchSentTo(ANALYTICS_URL)?.map((e: any) => e.event)).toEqual(['button_clicked']) + }) + + it('routes immediate $ai_* captures to the dedicated AI path', async () => { + const ph = newClient({ _internal_dedicatedAiEndpoint: true }) + await ph.captureImmediate({ distinctId: 'u1', event: '$ai_embedding', properties: {} }) + + expect(batchSentTo(AI_URL)?.[0].event).toBe('$ai_embedding') + expect(callsTo(ANALYTICS_URL)).toHaveLength(0) + await ph.shutdown() + }) + }) + + describe('when disabled (default)', () => { + it('routes $ai_* events to the normal batch path', async () => { + const ph = newClient() + ph.capture({ distinctId: 'u1', event: '$ai_generation', properties: {} }) + await ph.shutdown() + + expect(batchSentTo(ANALYTICS_URL)?.map((e: any) => e.event)).toEqual(['$ai_generation']) + expect(callsTo(AI_URL)).toHaveLength(0) + }) + }) +}) diff --git a/packages/node/src/client.ts b/packages/node/src/client.ts index 77174b9e25..b74cc74740 100644 --- a/packages/node/src/client.ts +++ b/packages/node/src/client.ts @@ -1,6 +1,7 @@ import { version } from './version' import { + EventChannel, FeatureFlagValue, isBlockedUA, isPlainObject, @@ -56,6 +57,11 @@ const WAITUNTIL_DEBOUNCE_MS = 50 const WAITUNTIL_MAX_WAIT_MS = 500 const DEFAULT_NODE_HOST = 'https://us.i.posthog.com' +const AI_EVENT_CHANNEL: EventChannel = { + queueKey: PostHogPersistedProperty.AiQueue, + path: '/i/v0/ai/batch/', +} + // Process-wide dedup for deprecation warnings — without this, calling a deprecated // method in a loop would spam logs. Matches Python's `warnings.warn` default-dedup behavior. const _emittedDeprecations = new Set() @@ -134,6 +140,8 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen timer: ReturnType | undefined } + private readonly _useDedicatedAiEndpoint: boolean + /** * Initialize a new PostHog client instance. * @@ -174,6 +182,7 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen super(normalizedApiKey, normalizedOptions) this.options = normalizedOptions + this._useDedicatedAiEndpoint = this.options._internal_dedicatedAiEndpoint === true this.context = this.initializeContext() this.options.featureFlagsPollingInterval = @@ -224,6 +233,17 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen this.maxCacheSize = normalizedOptions.maxCacheSize || MAX_CACHE_SIZE } + protected override eventChannels(): readonly EventChannel[] { + return this._useDedicatedAiEndpoint ? [...super.eventChannels(), AI_EVENT_CHANNEL] : super.eventChannels() + } + + protected override channelForEvent(event: unknown): EventChannel { + if (this._useDedicatedAiEndpoint && typeof event === 'string' && event.startsWith('$ai_')) { + return AI_EVENT_CHANNEL + } + return super.channelForEvent(event) + } + protected override enqueue(type: string, message: any, options?: PostHogCaptureOptions): void { super.enqueue(type, message, options) this.scheduleDebouncedFlush() diff --git a/packages/node/src/types.ts b/packages/node/src/types.ts index d906a0024b..184296714d 100644 --- a/packages/node/src/types.ts +++ b/packages/node/src/types.ts @@ -135,6 +135,13 @@ export type PostHogOptions = Omit & { persistence?: 'memory' personalApiKey?: string privacyMode?: boolean + /** + * Routes `$ai_*` events to a dedicated capture-ai endpoint in their own batch. + * + * @internal Not ready for use — the backend endpoint and ingress routing are still being + * rolled out. Do not enable; behaviour and naming may change or be removed without notice. + */ + _internal_dedicatedAiEndpoint?: boolean enableExceptionAutocapture?: boolean // The interval in milliseconds between polls for refreshing feature flag definitions. Defaults to 30 seconds. featureFlagsPollingInterval?: number