feat(node): add request context to OpenTelemetry spans

.changeset/bright-jeans-learn.md

@@ -0,0 +1,5 @@
+---
+'posthog-node': minor
+---
+
+Add an optional OpenTelemetry context processor, and an optional `signal` that ends a context's lifetime.

packages/node/README.md

@@ -8,6 +8,45 @@ SDK usage examples and code snippets live in the official documentation so they
 
 - [Node.js library docs](https://posthog.com/docs/libraries/node)
 
+## OpenTelemetry request context
+
+The optional `posthog-node/otel` entrypoint exports `PostHogContextSpanProcessor`.
+Pass your PostHog client to its constructor and add the processor to the application's OpenTelemetry provider.
+Install the optional `@opentelemetry/api` peer dependency and keep the application's provider, exporter, sampling, and other processors.
+Configure OpenTelemetry before the application accepts requests, and install `setupExpressRequestContext` before Express routes.
+
+The processor adds available `sessionId` and `posthogDistinctId` attributes to recording OpenTelemetry SDK spans.
+Disabled or opted-out clients add no PostHog attributes. Existing span attributes keep their values.
+It also updates an active request span when PostHog reads the request context.
+`enrichCurrentSpan` is also available from the main `posthog-node` entry without loading OpenTelemetry. It is a no-op until a processor is registered.
+Explicit request entry also fills missing IDs on the nearest recording server ancestor, following parent span objects observed by the processor.
+Parent links use weak references, so a retained child does not keep ended ancestors alive.
+If the sampler drops an intermediate span or garbage collection removes it, that parent chain cannot be verified and the earlier server span stays unchanged.
+Ordinary context entry keeps its current-span behavior and does not walk to server ancestors.
+It preserves explicit span attributes and reads values once, so exported spans keep their original context.
+It recognizes `sessionId` and `posthogDistinctId`; `user.id` alone does not override header-derived analytics identity.
+API-only spans without public readable attributes are skipped.
+For a manual request integration, call `enrichCurrentSpan(posthog)` from `posthog-node/otel` after entering the header context.
+This fills missing attributes on an incoming span that already has application context.
+The helper does nothing when no context processor is installed.
+For other Node.js frameworks, pass `{ signal: controller.signal }` to `withContext` or `enterContext` and abort the controller when the request ends.
+This ends the context for detached work, including inherited nested scopes, without cancelling the handler.
+The signal also works without OpenTelemetry. Calls without a signal keep their existing scope behavior.
+`fresh: true` starts an independent context with only the supplied values and its own optional signal.
+A background job can use it with explicit IDs before or after the request ends.
+Without `fresh: true`, nested scopes inherit the parent lifetime as well as their own signal.
+Express preserves an existing application identity and treats incoming headers as fallback analytics context.
+The headers must not be used for authentication or authorization.
+
+Express context stays active through async request work, including promises and timers that continue after the response ends.
+Exceptions keep the context they were captured in while asynchronous error processing finishes.
+Use explicit context for a background job that needs its own IDs.
+The Node.js `PostHogInterceptor` from `posthog-node/nestjs` also enriches request and child spans.
+It keeps its existing fresh header context, with the same lifetime as Express.
+Explicit `sessionId` and `posthogDistinctId` span attributes keep their values.
+This integration does not create IDs, traces, recordings, or log exporters.
+It does not change the application's OpenTelemetry setup or require OpenTelemetry for other SDK use.
+
 ## Questions??
 
 ### [Check out our community page.](https://posthog.com/posts)

packages/node/package.json

@@ -49,16 +49,17 @@
     "@edge-runtime/vm": "^5.0.0",
     "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/context-async-hooks": "^2.2.0",
+    "@opentelemetry/sdk-trace-base": "^2.10.0",
     "@posthog-tooling/tsconfig-base": "workspace:*",
     "@posthog/types": "workspace:^",
     "@rslib/core": "catalog:",
     "@types/express": "^5.0.6",
     "@types/node": "^20.0.0",
     "esbuild": "^0.25.0",
-    "rxjs": "^7.8.0",
     "@vitest/coverage-v8": "catalog:",
-    "vitest": "catalog:",
-    "typescript": "catalog:native"
+    "rxjs": "^7.8.0",
+    "typescript": "catalog:native",
+    "vitest": "catalog:"
   },
   "peerDependencies": {
     "@opentelemetry/api": "^1.2.0",
@@ -120,6 +121,12 @@
       "import": "./dist/experimental.mjs",
       "require": "./dist/experimental.js",
       "default": "./dist/experimental.js"
+    },
+    "./otel": {
+      "types": "./dist/entrypoints/otel.d.ts",
+      "import": "./dist/entrypoints/otel.mjs",
+      "require": "./dist/entrypoints/otel.js",
+      "default": "./dist/entrypoints/otel.js"
     }
   }
 }

packages/node/references/posthog-node-references-latest.json

@@ -2098,6 +2098,11 @@
           "description": "If true, replaces the current context entirely. If false, merges with the existing context (new values override existing ones).   false",
           "type": "boolean",
           "name": "fresh"
+        },
+        {
+          "description": "Ends this context and its inherited scopes when aborted, without cancelling the callback. An already aborted signal starts no active context. A fresh context has an independent lifetime.",
+          "type": "AbortSignal",
+          "name": "signal"
         }
       ],
       "path": "src/extensions/context/types.ts"

packages/node/src/__tests__/context.spec.ts

@@ -46,6 +46,109 @@ describe('PostHog Context', () => {
     await posthog.shutdown()
   })
 
+  describe('context lifetime signals', () => {
+    it.each(['withContext', 'enterContext'] as const)('expires %s without cancelling its callback', async (method) => {
+      const controller = new AbortController()
+      const data = { distinctId: 'request-user', sessionId: 'request-session' }
+      const callback = async (): Promise<string> => {
+        expect(posthog.getContext()).toMatchObject(data)
+        controller.abort()
+        await Promise.resolve()
+        expect(posthog.getContext()).toBeUndefined()
+        return 'handler completed'
+      }
+      const result =
+        method === 'withContext'
+          ? posthog.withContext(data, callback, { signal: controller.signal })
+          : posthog.withContext({}, () => {
+              posthog.enterContext(data, { signal: controller.signal })
+              return callback()
+            })
+      expect(await result).toBe('handler completed')
+      expect(posthog.getContext()).toBeUndefined()
+    })
+
+    it.each([
+      ['parent', false],
+      ['parent', true],
+      ['child', false],
+      ['child', true],
+    ] as const)('composes %s expiry with fresh=%s nested scopes', (expired, fresh) => {
+      const parent = new AbortController()
+      const child = new AbortController()
+      posthog.withContext(
+        { distinctId: 'parent-user', sessionId: 'parent-session' },
+        () => {
+          posthog.withContext(
+            { distinctId: 'child-user' },
+            () => {
+              expect(posthog.getContext()).toMatchObject({ distinctId: 'child-user' })
+              const controller = expired === 'parent' ? parent : child
+              controller.abort()
+              if (expired === 'parent' && fresh) expect(posthog.getContext()).toEqual({ distinctId: 'child-user' })
+              else expect(posthog.getContext()).toBeUndefined()
+            },
+            { fresh, signal: child.signal }
+          )
+          if (expired === 'parent') expect(posthog.getContext()).toBeUndefined()
+          else expect(posthog.getContext()).toMatchObject({ distinctId: 'parent-user', sessionId: 'parent-session' })
+        },
+        { signal: parent.signal }
+      )
+    })
+
+    it.each([{ distinctId: 'new-user' }, { sessionId: 'new-session' }])(
+      'starts explicit scopes after expiry with %j',
+      (data) => {
+        const controller = new AbortController()
+        posthog.withContext(
+          { distinctId: 'ended-user', sessionId: 'ended-session' },
+          () => {
+            controller.abort()
+            posthog.withContext(data, () => expect(posthog.getContext()).toEqual(data), { fresh: true })
+            posthog.withContext(data, () => {
+              expect(posthog.getContext()?.distinctId).toBe(data.distinctId)
+              expect(posthog.getContext()?.sessionId).toBe(data.sessionId)
+            })
+          },
+          { signal: controller.signal }
+        )
+      }
+    )
+
+    it('runs a callback with no active context when its signal is already aborted', () => {
+      const controller = new AbortController()
+      controller.abort()
+      const result = posthog.withContext(
+        { distinctId: 'ended-user' },
+        () => {
+          expect(posthog.getContext()).toBeUndefined()
+          return 'handler completed'
+        },
+        { signal: controller.signal }
+      )
+      expect(result).toBe('handler completed')
+    })
+
+    it('isolates signal expiry across concurrent async contexts', async () => {
+      const controllers = [new AbortController(), new AbortController()]
+      const releases: (() => void)[] = []
+      const results = controllers.map((controller, index) =>
+        posthog.withContext(
+          { sessionId: `session-${index}` },
+          async () => {
+            await new Promise<void>((resolve) => releases.push(resolve))
+            return posthog.getContext()?.sessionId
+          },
+          { signal: controller.signal }
+        )
+      )
+      controllers[0].abort()
+      releases.forEach((release) => release())
+      expect(await Promise.all(results)).toEqual([undefined, 'session-1'])
+    })
+  })
+
   it('should attach context tags to events', async () => {
     posthog.withContext({ properties: { plan: 'premium', region: 'us-east' } }, () => {
       posthog.capture({ distinctId: 'user-1', event: 'test_event' })
@@ -474,6 +577,41 @@ describe('PostHog Context', () => {
     })
   })
 
+  it.each(
+    (['buffered', 'immediate'] as const).flatMap((mode) =>
+      (['context', 'explicit', 'personless'] as const).map((identity) => ({ mode, identity }))
+    )
+  )('keeps $mode exception context after signal expiry with $identity identity', async ({ mode, identity }) => {
+    const controller = new AbortController()
+    const explicit = identity === 'explicit'
+    const contextId = identity === 'personless' ? undefined : 'request-user'
+    const data = { distinctId: contextId, sessionId: 'request-session', properties: { route: '/work' } }
+    let pending: Promise<void> | undefined
+    const capture = (): void => {
+      const error = new Error('operation failed')
+      const properties = explicit ? { $session_id: 'explicit-session', route: '/explicit' } : undefined
+      const distinctId = explicit ? 'explicit-user' : undefined
+      if (mode === 'immediate') pending = posthog.captureExceptionImmediate(error, distinctId, properties)
+      else posthog.captureException(error, distinctId, properties)
+      controller.abort()
+      expect(posthog.getContext()).toBeUndefined()
+    }
+
+    posthog.withContext(data, capture, { signal: controller.signal })
+    await pending
+    await posthog.flush()
+
+    const events = getLastBatchEvents()
+    expect(events).toHaveLength(1)
+    const event = events?.[0]
+    expect(event.event).toBe('$exception')
+    expect(event.properties.$session_id).toBe(explicit ? 'explicit-session' : 'request-session')
+    expect(event.properties.route).toBe(explicit ? '/explicit' : '/work')
+    expect(event.properties.$process_person_profile).toBe(identity === 'personless' ? false : undefined)
+    if (identity === 'personless') expect(event.distinct_id).toEqual(expect.any(String))
+    else expect(event.distinct_id).toBe(explicit ? 'explicit-user' : 'request-user')
+  })
+
   it.each([
     {
       label: 'no distinctId and no context → personless',

packages/node/src/__tests__/extensions/context-listeners.spec.ts

@@ -0,0 +1,29 @@
+import type { PostHogBackendClient } from '@/client'
+
+async function loadTwoModuleCopies() {
+  vi.resetModules()
+  const first = await import('@/extensions/context/listeners')
+  vi.resetModules()
+  const second = await import('@/extensions/context/listeners')
+  return { first, second }
+}
+
+describe('context listeners', () => {
+  it.each([
+    ['the listener', 'second', 'first'],
+    ['the notifier', 'first', 'second'],
+  ] as const)('share one registry when %s comes from another module copy', async (_, adder, notifier) => {
+    const copies = await loadTwoModuleCopies()
+    expect(copies.second).not.toBe(copies.first)
+    const client = {} as PostHogBackendClient
+    const listener = vi.fn()
+
+    const unsubscribe = copies[adder].addContextListener(client, listener)
+    copies[notifier].enrichCurrentSpan(client)
+    unsubscribe()
+    copies[notifier].enrichCurrentSpan(client)
+
+    expect(listener).toHaveBeenCalledTimes(1)
+    expect(listener).toHaveBeenCalledWith(true)
+  })
+})

packages/node/src/__tests__/extensions/nestjs-span-context.spec.ts

@@ -0,0 +1,106 @@
+import { EventEmitter } from 'node:events'
+import { defer, lastValueFrom } from 'rxjs'
+import { context, trace, ROOT_CONTEXT } from '@opentelemetry/api'
+import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks'
+import { BasicTracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'
+import { PostHog } from '@/entrypoints/index.node'
+import { PostHogInterceptor } from '@/extensions/nestjs'
+import { PostHogContextSpanProcessor } from '@/entrypoints/otel'
+
+describe('NestJS span context', () => {
+  const manager = new AsyncLocalStorageContextManager()
+  let client: PostHog
+  let provider: BasicTracerProvider
+  let exporter: InMemorySpanExporter
+
+  beforeAll(() => context.setGlobalContextManager(manager.enable()))
+  afterAll(() => {
+    context.disable()
+    manager.disable()
+  })
+  beforeEach(() => {
+    vi.useRealTimers()
+    client = new PostHog('phc_test')
+    exporter = new InMemorySpanExporter()
+    provider = new BasicTracerProvider({
+      spanProcessors: [new PostHogContextSpanProcessor(client), new SimpleSpanProcessor(exporter)],
+    })
+  })
+  afterEach(async () => {
+    await provider.shutdown()
+    await client.shutdown()
+  })
+
+  function requestContext(response: object): any {
+    return {
+      switchToHttp: () => ({
+        getRequest: () => ({
+          headers: { 'x-posthog-session-id': 'browser-session', 'x-posthog-distinct-id': 'header-user' },
+        }),
+        getResponse: () => response,
+      }),
+    }
+  }
+
+  it.each([
+    ['success', 'finish'],
+    ['error', 'close'],
+  ])('keeps %s context after response %s', async (outcome, event) => {
+    const response = new EventEmitter()
+    const tracer = provider.getTracer('nestjs-test')
+    const incoming = tracer.startSpan('incoming', { attributes: { posthogDistinctId: 'application-user' } })
+    let releasePending!: () => void
+    let releaseLate!: () => void
+    let pending!: Promise<void>
+    let late!: Promise<void>
+    await context.with(trace.setSpan(ROOT_CONTEXT, incoming), () =>
+      client.withContext({ properties: { application: true } }, async () => {
+        const interceptor = new PostHogInterceptor(client)
+        const result = lastValueFrom(
+          interceptor.intercept(requestContext(response), {
+            handle: () =>
+              defer(async () => {
+                const child = tracer.startSpan('child')
+                pending = (async () => {
+                  await new Promise<void>((resolve) => {
+                    releasePending = resolve
+                  })
+                  tracer.startSpan('pending-response').end()
+                })()
+                late = (async () => {
+                  await new Promise<void>((resolve) => {
+                    releaseLate = resolve
+                  })
+                  tracer.startSpan('late').end()
+                  expect(client.getContext()?.sessionId).toBe('browser-session')
+                  child.end()
+                })()
+                if (outcome === 'error') throw new Error('request failure')
+                return 'ok'
+              }),
+          })
+        )
+        if (outcome === 'error') await expect(result).rejects.toThrow('request failure')
+        else expect(await result).toBe('ok')
+        releasePending()
+        await pending
+        response.emit(event)
+        releaseLate()
+        await late
+      })
+    )
+    incoming.end()
+    await provider.forceFlush()
+    const spans = exporter.getFinishedSpans()
+    for (const name of ['child', 'pending-response', 'late']) {
+      expect(spans.find((span) => span.name === name)?.attributes).toEqual({
+        sessionId: 'browser-session',
+        posthogDistinctId: 'header-user',
+      })
+    }
+    expect(spans.find((span) => span.name === 'incoming')?.attributes).toEqual({
+      sessionId: 'browser-session',
+      posthogDistinctId: 'application-user',
+    })
+  })
+})