feat(audience): add transport, queue, and context collection to core (#2813)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: nattb8

packages/audience/core/jest.config.ts

@@ -3,10 +3,12 @@ import type { Config } from 'jest';
 const config: Config = {
   roots: ['<rootDir>/src'],
   moduleDirectories: ['node_modules', 'src'],
+  moduleNameMapper: { '^@imtbl/(.*)$': '<rootDir>/../../../node_modules/@imtbl/$1/src' },
   testEnvironment: 'jsdom',
   transform: {
-    '^.+\\.tsx?$': '@swc/jest',
+    '^.+\\.(t|j)sx?$': '@swc/jest',
   },
+  transformIgnorePatterns: [],
 };
 
 export default config;

packages/audience/core/package.json

@@ -5,7 +5,9 @@
   "author": "Immutable",
   "private": true,
   "bugs": "https://github.com/immutable/ts-immutable-sdk/issues",
-  "dependencies": {},
+  "dependencies": {
+    "@imtbl/metrics": "workspace:*"
+  },
   "devDependencies": {
     "@swc/core": "^1.4.2",
     "@swc/jest": "^0.2.37",

packages/audience/core/src/context.test.ts

@@ -0,0 +1,25 @@
+import { collectContext } from './context';
+
+describe('collectContext', () => {
+  it('includes library name and version', () => {
+    const ctx = collectContext();
+    expect(ctx.library).toBe('@imtbl/audience');
+    expect(ctx.libraryVersion).toBeDefined();
+  });
+
+  it('collects browser signals in jsdom', () => {
+    const ctx = collectContext();
+    expect(ctx.userAgent).toBeDefined();
+    expect(ctx.locale).toBeDefined();
+    expect(ctx.timezone).toBeDefined();
+    expect(ctx.screen).toMatch(/^\d+x\d+$/);
+  });
+
+  it('collects page info', () => {
+    const ctx = collectContext();
+    expect(ctx.pageUrl).toBeDefined();
+    expect(ctx.pagePath).toBeDefined();
+    expect(typeof ctx.pageReferrer).toBe('string');
+    expect(typeof ctx.pageTitle).toBe('string');
+  });
+});

packages/audience/core/src/context.ts

@@ -0,0 +1,25 @@
+import type { EventContext } from './types';
+import { isBrowser } from './utils';
+
+// WARNING: DO NOT CHANGE THE STRING BELOW. IT GETS REPLACED AT BUILD TIME.
+const SDK_VERSION = '__SDK_VERSION__';
+
+export function collectContext(): EventContext {
+  const context: EventContext = {
+    library: '@imtbl/audience',
+    libraryVersion: SDK_VERSION,
+  };
+
+  if (!isBrowser()) return context;
+
+  context.userAgent = navigator.userAgent;
+  context.locale = navigator.language;
+  context.timezone = Intl.DateTimeFormat().resolvedOptions().timeZone;
+  context.screen = `${window.screen.width}x${window.screen.height}`;
+  context.pageUrl = window.location.href;
+  context.pagePath = window.location.pathname;
+  context.pageReferrer = document.referrer;
+  context.pageTitle = document.title;
+
+  return context;
+}

packages/audience/core/src/index.ts

@@ -26,3 +26,8 @@ export {
 } from './config';
 
 export { generateId, getTimestamp, isBrowser } from './utils';
+
+export type { Transport } from './transport';
+export { httpTransport, httpSend } from './transport';
+export { MessageQueue } from './queue';
+export { collectContext } from './context';

packages/audience/core/src/queue.test.ts

@@ -0,0 +1,160 @@
+import { MessageQueue } from './queue';
+import type { Transport } from './transport';
+import type { Message } from './types';
+import * as storage from './storage';
+
+function makeMessage(id: string): Message {
+  return {
+    type: 'track',
+    messageId: id,
+    eventTimestamp: '2026-04-01T00:00:00.000Z',
+    anonymousId: 'anon-1',
+    surface: 'web',
+    context: { library: '@imtbl/audience', libraryVersion: '0.0.0' },
+    eventName: 'test',
+  };
+}
+
+function createQueue(
+  transport: Transport,
+  opts: { flushIntervalMs?: number; flushSize?: number } = {},
+) {
+  return new MessageQueue(
+    transport,
+    'https://api.immutable.com/v1/audience/messages',
+    'pk_imx_test',
+    opts.flushIntervalMs ?? 5_000,
+    opts.flushSize ?? 20,
+  );
+}
+
+beforeEach(() => {
+  jest.useFakeTimers();
+  localStorage.clear();
+});
+
+afterEach(() => {
+  jest.useRealTimers();
+});
+
+describe('MessageQueue', () => {
+  it('enqueues messages and flushes them', async () => {
+    const send = jest.fn().mockResolvedValue(true);
+    const queue = createQueue({ send });
+
+    queue.enqueue(makeMessage('1'));
+    queue.enqueue(makeMessage('2'));
+
+    await queue.flush();
+
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(send.mock.calls[0][2].messages).toHaveLength(2);
+    expect(queue.length).toBe(0);
+  });
+
+  it('retains messages on failed flush', async () => {
+    const send = jest.fn().mockResolvedValue(false);
+    const queue = createQueue({ send });
+
+    queue.enqueue(makeMessage('1'));
+    await queue.flush();
+
+    expect(queue.length).toBe(1);
+  });
+
+  it('flushes automatically when batch size is reached', async () => {
+    const send = jest.fn().mockResolvedValue(true);
+    const queue = createQueue({ send }, { flushSize: 2 });
+
+    queue.enqueue(makeMessage('1'));
+    expect(send).not.toHaveBeenCalled();
+
+    queue.enqueue(makeMessage('2'));
+    // flush is async — await the microtask
+    await Promise.resolve();
+    expect(send).toHaveBeenCalledTimes(1);
+  });
+
+  it('flushes on timer interval', async () => {
+    const send = jest.fn().mockResolvedValue(true);
+    const queue = createQueue({ send }, { flushIntervalMs: 1_000 });
+
+    queue.start();
+    queue.enqueue(makeMessage('1'));
+
+    jest.advanceTimersByTime(1_000);
+    // flush is async
+    await Promise.resolve();
+    expect(send).toHaveBeenCalledTimes(1);
+
+    queue.stop();
+  });
+
+  it('persists messages to localStorage', () => {
+    const send = jest.fn().mockResolvedValue(true);
+    const queue = createQueue({ send });
+
+    queue.enqueue(makeMessage('1'));
+
+    const stored = JSON.parse(localStorage.getItem('__imtbl_audience_queue')!);
+    expect(stored).toHaveLength(1);
+    expect(stored[0].messageId).toBe('1');
+  });
+
+  it('restores messages from localStorage on construction', () => {
+    storage.setItem('queue', [makeMessage('restored')]);
+
+    const send = jest.fn().mockResolvedValue(true);
+    const queue = createQueue({ send });
+
+    expect(queue.length).toBe(1);
+  });
+
+  it('does not flush concurrently', async () => {
+    let resolveFirst: () => void;
+    const firstCall = new Promise<boolean>((r) => { resolveFirst = () => r(true); });
+    const send = jest.fn()
+      .mockReturnValueOnce(firstCall)
+      .mockResolvedValue(true);
+
+    const queue = createQueue({ send });
+    queue.enqueue(makeMessage('1'));
+
+    const flush1 = queue.flush();
+    const flush2 = queue.flush(); // should no-op
+
+    resolveFirst!();
+    await flush1;
+    await flush2;
+
+    expect(send).toHaveBeenCalledTimes(1);
+  });
+
+  it('clears all messages and storage', () => {
+    const send = jest.fn().mockResolvedValue(true);
+    const queue = createQueue({ send });
+
+    queue.enqueue(makeMessage('1'));
+    queue.clear();
+
+    expect(queue.length).toBe(0);
+    expect(localStorage.getItem('__imtbl_audience_queue')).toBeNull();
+  });
+
+  it('handles messages enqueued during flush', async () => {
+    let queue: ReturnType<typeof createQueue>;
+    const send = jest.fn().mockImplementation(async () => {
+      // Simulate a message arriving during the network request
+      queue.enqueue(makeMessage('late'));
+      return true;
+    });
+
+    queue = createQueue({ send });
+    queue.enqueue(makeMessage('1'));
+
+    await queue.flush();
+
+    // The original message was sent, but the late one should remain
+    expect(queue.length).toBe(1);
+  });
+});

packages/audience/core/src/queue.ts

@@ -0,0 +1,87 @@
+import type { Message, BatchPayload } from './types';
+import type { Transport } from './transport';
+import * as storage from './storage';
+
+const STORAGE_KEY = 'queue';
+
+/**
+ * Batched message queue with localStorage durability.
+ *
+ * Messages are flushed on a timer OR when the queue reaches `flushSize`,
+ * whichever comes first. On success the sent messages are removed; on
+ * failure they stay queued and retry on the next flush cycle.
+ *
+ * localStorage is used as a write-through cache so messages survive
+ * page navigations. On construction, any previously-persisted messages
+ * are restored into memory.
+ */
+export class MessageQueue {
+  private messages: Message[];
+
+  private timer: ReturnType<typeof setInterval> | null = null;
+
+  private flushing = false;
+
+  constructor(
+    private readonly transport: Transport,
+    private readonly endpointUrl: string,
+    private readonly publishableKey: string,
+    private readonly flushIntervalMs: number,
+    private readonly flushSize: number,
+  ) {
+    this.messages = (storage.getItem(STORAGE_KEY) as Message[] | undefined) ?? [];
+  }
+
+  start(): void {
+    if (this.timer) return;
+    this.timer = setInterval(() => this.flush(), this.flushIntervalMs);
+  }
+
+  stop(): void {
+    if (!this.timer) return;
+    clearInterval(this.timer);
+    this.timer = null;
+  }
+
+  enqueue(message: Message): void {
+    this.messages.push(message);
+    this.persist();
+
+    if (this.messages.length >= this.flushSize) {
+      this.flush();
+    }
+  }
+
+  /** Guard prevents concurrent flushes from racing on the same batch. */
+  async flush(): Promise<void> {
+    if (this.flushing || this.messages.length === 0) return;
+
+    this.flushing = true;
+    try {
+      const batch = [...this.messages];
+      const payload: BatchPayload = { messages: batch };
+
+      const ok = await this.transport.send(this.endpointUrl, this.publishableKey, payload);
+      if (ok) {
+        // Slice rather than clear — new messages may have been enqueued during the request.
+        this.messages = this.messages.slice(batch.length);
+        this.persist();
+      }
+    } finally {
+      this.flushing = false;
+    }
+  }
+
+  get length(): number {
+    return this.messages.length;
+  }
+
+  clear(): void {
+    this.messages = [];
+    storage.removeItem(STORAGE_KEY);
+  }
+
+  private persist(): void {
+    storage.setItem(STORAGE_KEY, this.messages);
+  }
+}

packages/audience/core/src/transport.test.ts

@@ -0,0 +1,56 @@
+import { httpSend } from './transport';
+import type { BatchPayload } from './types';
+
+const payload: BatchPayload = {
+  messages: [
+    {
+      type: 'track',
+      messageId: 'msg-1',
+      eventTimestamp: '2026-04-01T00:00:00.000Z',
+      anonymousId: 'anon-1',
+      surface: 'web',
+      context: { library: '@imtbl/audience', libraryVersion: '0.0.0' },
+      eventName: 'purchase',
+      properties: { value: 9.99 },
+    },
+  ],
+};
+
+describe('httpSend', () => {
+  const originalFetch = global.fetch;
+
+  afterEach(() => {
+    global.fetch = originalFetch;
+  });
+
+  it('sends POST with correct headers and body', async () => {
+    const mockFetch = jest.fn().mockResolvedValue({ ok: true });
+    global.fetch = mockFetch;
+
+    await httpSend('https://api.immutable.com/v1/audience/messages', 'pk_imx_test', payload);
+
+    expect(mockFetch).toHaveBeenCalledWith('https://api.immutable.com/v1/audience/messages', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        'x-immutable-publishable-key': 'pk_imx_test',
+      },
+      body: JSON.stringify(payload),
+    });
+  });
+
+  it('returns true on success', async () => {
+    global.fetch = jest.fn().mockResolvedValue({ ok: true });
+    expect(await httpSend('https://example.com', 'pk', payload)).toBe(true);
+  });
+
+  it('returns false on HTTP error', async () => {
+    global.fetch = jest.fn().mockResolvedValue({ ok: false, status: 500 });
+    expect(await httpSend('https://example.com', 'pk', payload)).toBe(false);
+  });
+
+  it('returns false on network error', async () => {
+    global.fetch = jest.fn().mockRejectedValue(new TypeError('Failed to fetch'));
+    expect(await httpSend('https://example.com', 'pk', payload)).toBe(false);
+  });
+});