Create Your Own Source

A source captures events from an environment (browser, server, third-party API) and forwards them to the walkerOS collector.

The source interface

Sources are async functions that receive a context object and return a source instance:

type Source.Init<Types> = (
context: Source.Context<Types>,
) => Source.Instance<Types> | Promise<Source.Instance<Types>>;

The context contains:

interface Source.Context<Types> {
config: Partial<Source.Config<Types>>;  // Settings, mapping, options
env: Types['env'];                       // Environment (push, logger)
logger: Logger;                          // Logging functions
id: string;                              // Source identifier
collector: Collector.Instance;           // Collector reference
}

The returned instance must implement:

interface Source.Instance {
type: string; // Unique identifier
config: Source.Config; // Merged configuration
push: Elb.Fn; // Function to send events (forwards to env.elb)
destroy?(): void | Promise<void>; // Optional cleanup
on?(event, context): void | Promise<void>; // Optional reactive hooks
}

Types bundle

Sources use a Types interface to bundle all TypeScript types:

import type { Source, Elb } from '@walkeros/core';

// 1. Define your settings
interface Settings {
captureClicks?: boolean;
prefix?: string;
}

// 2. Define environment dependencies
interface Env extends Source.BaseEnv {
customAPI?: YourAPIType;
}

// 3. Bundle them together
interface Types extends Source.Types<Settings, never, Elb.Fn, Env> {}

The 4 type parameters:

1. Settings: Source configuration options
2. Mapping: Event mapping (usually never for sources)
3. Push: External push signature (what source.push exposes)
4. Env: Internal dependencies (what source calls via env.elb)

Context destructuring

The context parameter contains everything your source needs:

export const mySource: Source.Init<Types> = async (context) => {
// Destructure what you need from context
const { config = {}, env, logger, id } = context;
const { push: envPush, customAPI } = env;

// customAPI validation only needed if required by your source
if (!customAPI) {
  throw new Error('Source requires customAPI in environment');
}

// Use env.push to send events to collector
await envPush({ name: 'my event', data: { value: 'example' } });
};

The collector provides env.push (formerly env.elb). You provide other dependencies (like window, document, custom APIs) when configuring the source.

Per-scope context (server sources)

A single source factory instance handles many concurrent invocations: an Express server processes overlapping requests, a Lambda may be reused across invocations, a queue consumer processes batches. Each logical unit of work, an HTTP request, a queue message, a websocket frame, is a scope. To keep ingest metadata and response delegates isolated between concurrent scopes, server sources wrap each invocation with context.withScope:

const push = async (req: Request, res: Response): Promise<void> => {
const respond = createRespond((options) => {
  // wire options into res.send / res.json
});

// Each call to withScope builds a fresh Ingest from `req` via
// `config.ingest` mapping, captures `respond`, and runs the body
// with a per-scope env. Concurrent scopes never crosstalk.
await context.withScope(req, respond, async (env) => {
  const eventData = req.body && typeof req.body === 'object' ? req.body : {};
  await env.push(eventData);
  respond({ body: { success: true } });
});
};

Inside the body, env.push carries the scope's ingest and respond all the way to destinations. Transformers and destinations read env.respond to delegate the HTTP response; context.ingest for source-extracted metadata. The scope ends when the body resolves.

Browser sources (and other single-scope sources) skip withScope. A browser tab is a single logical scope for its entire lifetime; calling env.push directly on the factory env is correct. Only server sources handling concurrent inbound work need withScope.

Minimal example

import type { Source, Elb } from '@walkeros/core';

interface Settings {
prefix?: string;
}

interface Env extends Source.BaseEnv {}

interface Types extends Source.Types<Settings, never, Elb.Fn, Env> {}

export const sourceCustom: Source.Init<Types> = async (context) => {
const { config = {}, env } = context;
const { push: envPush } = env;

const settings: Source.Settings<Types> = {
  prefix: 'custom',
  ...config?.settings,
};

const fullConfig: Source.Config<Types> = {
  ...config,
  settings,
};

return {
  type: 'custom',
  config: fullConfig,
  push: envPush,
};
};

Complete example: Event API source

Capturing events from a third-party API with event listeners:

import type { Source, Elb } from '@walkeros/core';

// Your external API interface
interface ExternalAPI {
on(event: string, handler: (data: unknown) => void): void;
off(event: string, handler: (data: unknown) => void): void;
}

interface Settings {
captureInteractions?: boolean;
captureErrors?: boolean;
prefix?: string;
}

interface Env extends Source.BaseEnv {
api?: ExternalAPI;
}

interface Types extends Source.Types<Settings, never, Elb.Fn, Env> {}

export const sourceEventAPI: Source.Init<Types> = async (context) => {
const { config = {}, env } = context;
const { push: envPush, api } = env;

if (!api) throw new Error('Source requires api instance');

const settings: Source.Settings<Types> = {
  captureInteractions: true,
  captureErrors: true,
  prefix: 'app',
  ...config?.settings,
};

const handlers = new Map<string, (data: unknown) => void>();

// Register event handler
const register = (
  event: string,
  transform: (data: unknown) => { name: string; data: unknown },
) => {
  const handler = (data: unknown) => {
    const transformed = transform(data);
    envPush(transformed);
  };
  api.on(event, handler);
  handlers.set(event, handler);
};

// Set up event listeners based on settings
if (settings.captureInteractions) {
  register('userAction', (data) => ({
    name: `${settings.prefix} interaction`,
    data,
  }));
}

if (settings.captureErrors) {
  register('error', (data) => ({
    name: `${settings.prefix} error`,
    data,
  }));
}

return {
  type: 'event-api',
  config: { ...config, settings },
  push: envPush,
  destroy: async () => {
    handlers.forEach((handler, event) => api.off(event, handler));
  },
};
};

Using your source

import { startFlow } from '@walkeros/collector';
import { sourceEventAPI } from './sourceEventAPI';

const externalAPI = getYourAPI();

const { elb } = await startFlow({
sources: {
eventAPI: {
code: sourceEventAPI,
config: {
settings: {
  captureInteractions: true,
  prefix: 'myapp',
},
},
env: {
api: externalAPI, // Collector provides elb automatically
},
},
},
});

Testing your source

Test Utilities

// __tests__/test-utils.ts

export function createMockPush() {
const mock = jest.fn();
mock.mockResolvedValue({
  event: { id: 'test-id' },
  ok: true,
});
return mock;
}

export function createMockAPI() {
return {
  on: jest.fn(),
  off: jest.fn(),
};
}

Test Example

import { sourceEventAPI } from '../index';
import type { Source, Collector } from '@walkeros/core';
import { createMockLogger } from '@walkeros/core';
import { createMockPush, createMockAPI } from './test-utils';

// Helper to create source context for testing
function createSourceContext(
config: Partial<Source.Config<Types>> = {},
env: Partial<Types['env']> = {},
): Source.Context<Types> {
return {
  config,
  env: env as Types['env'],
  logger: env.logger || createMockLogger(),
  id: 'test-event-api',
  collector: {} as Collector.Instance,
};
}

describe('Event API Source', () => {
it('validates required dependencies', async () => {
  const push = createMockPush();

  await expect(
    sourceEventAPI(createSourceContext({}, { push }))
  ).rejects.toThrow('requires api');
});

it('registers event listeners', async () => {
  const api = createMockAPI();
  const push = createMockPush();

  await sourceEventAPI(
    createSourceContext(
      { settings: { captureInteractions: true } },
      { push, api }
    )
  );

  expect(api.on).toHaveBeenCalledWith('userAction', expect.any(Function));
});

it('transforms and forwards events', async () => {
  const api = createMockAPI();
  const push = createMockPush();

  await sourceEventAPI(createSourceContext({}, { push, api }));

  // Simulate event
  const handler = api.on.mock.calls[0][1];
  handler({ action: 'click' });

  expect(push).toHaveBeenCalledWith({
    name: 'app interaction',
    data: { action: 'click' },
  });
});

it('cleans up on destroy', async () => {
  const api = createMockAPI();
  const push = createMockPush();

  const source = await sourceEventAPI(createSourceContext({}, { push, api }));
  await source.destroy?.();

  expect(api.off).toHaveBeenCalled();
});
});

Reading a step back from the collector

When an integration test needs to call a step's raw push directly through the collector (a source's push(req, res), a destination's push(event, ctx), etc.), the bag types (collector.sources, collector.destinations, collector.transformers, collector.stores) erase the per-step generic. The push signature collapses to Elb.Fn on read.

Use the typed accessors from @walkeros/core to recover the narrow type without casts:

import { Source } from '@walkeros/core';
import type { Types as ExpressTypes } from '@walkeros/server-source-express';

const expressSource = Source.getSource<ExpressTypes>(collector, 'express');
await expressSource.push(req, res); // typed!

Symmetric helpers exist for every step kind: Source.getSource, Destination.getDestination, Transformer.getTransformer, Store.getStore. Each throws <Kind> not found: <id> when the id is unknown. The helpers are opt-in and have no runtime cost.

Setup lifecycle (optional)

Sources may implement an optional setup() lifecycle for one-time, operator-time provisioning, for example registering a webhook callback or creating a Pub/Sub subscription. Setup runs only when an operator explicitly invokes walkeros setup source.<name>; the runtime never auto-invokes it. Opt in via config.setup in the flow config (true or an object).

See Setup lifecycle on the destinations page for the full concept, and walkeros setup CLI command for the operator-facing command.

Conditional activation with require

Sources can declare dependencies on collector events. A source with require won't receive lifecycle events (via on()) until all specified events have fired:

await startFlow({
sources: {
  session: {
    code: sessionSource,
    config: { require: ['consent'] },
  },
  dataLayer: {
    code: dataLayerSource,
    config: { require: ['user'] },
  },
},
});

How it works:

• Sources are registered in collector.sources immediately, and the collector runs Instance.init() on each one eagerly after registration. Source.Config.init flips to true once init has run.
• require does not gate code execution. It gates on() delivery. Lifecycle events targeted at a source whose require is unmet are buffered in the source's Instance.queueOn array.
• After each collector event (consent, user, session, run, etc.), the collector decrements the matching entry from each source's require list. When a source's require is empty (and config.init === true), its queued lifecycle events are replayed by calling source.on(type, data) for each entry, then the queue is cleared.
• Chains work naturally: consent clears the session source's require, which fires user, which clears the dataLayer source's require, which then receives any queued events.

Common patterns:

Require	Use case
['consent']	Wait for CMP consent before tracking
['user']	Wait for identity resolution
['session']	Wait for session detection
['consent', 'run']	Wait for both consent and collector run

Common patterns

Polling for API Readiness

When the external API isn't immediately available:

async function waitForAPI<T>(getter: () => T | null): Promise<T> {
return new Promise((resolve) => {
const check = () => {
const instance = getter();
if (instance) resolve(instance);
else setTimeout(check, 100);
};
check();
});
}

// Usage
const api = await waitForAPI(() => window.myAPI || null);

Source as adapter pattern

Sources bridge external systems and the collector:

External System  ←→  Source (Adapter)  ←→  Collector

Two interfaces:

1. External (source.push): Platform-specific signature

   • Browser: push(elem, data, options) → Returns Promise<PushResult>
   • Server: push(req, res) → Returns Promise<void> (writes HTTP response)
   • Your choice: Match your environment's needs

2. Internal (env.elb): Standard collector interface

   • Always Elb.Fn - same across all sources
   • Sources translate external inputs → standard events → env.elb(event)

Example signatures:

// Browser source
export type Push = (elem?: Element, data?: Properties) => Promise<PushResult>;

// HTTP handler source
export type Push = (req: Request, res: Response) => Promise<void>;

// Standard event source
export type Push = Elb.Fn;

The Push type parameter defines what your source exposes externally. Internally, all sources use env.elb to forward to the collector.

Key concepts

• Context pattern: Sources receive a single context object with config, env, logger, id
• Validate custom dependencies: Only check optional env properties your source needs
• Types bundle: Use 4-parameter pattern for full type safety
• Adapter pattern: External push adapts to environment, internal env.push stays standard
• Cleanup: Implement destroy() to remove listeners
• Stateless: Let collector manage state

Package convention

Every walkerOS package includes machine-readable metadata for tooling and discovery.

walkerOS field in package.json

{
"walkerOS": { "type": "source", "platform": "web" },
"keywords": ["walkeros", "walkeros-source"]
}

Field	Required	Description
walkerOS	Yes	Object with type and platform metadata

Build-time generation

Use buildDev() from the shared tsup config to auto-generate walkerOS.json:

import { buildDev } from '@walkeros/config/tsup';

This file contains your package's JSON Schemas and examples, enabling MCP tools and the CLI to validate configurations without installing your package.

Optional: Hints

Packages can export a hints record from src/dev.ts to provide lightweight, actionable context beyond schemas and examples, such as capture timing, event formats, or troubleshooting tips. Hints are serialized into walkerOS.json and surfaced via MCP tools. See the walkeros-create-source skill for details.

Publishing checklist

• walkerOS field in package.json
• Keywords include walkeros and walkeros-source
• buildDev() in tsup.config.ts
• dist/walkerOS.json generated on build
• npm run test passes
• npm run lint passes

Next steps

• Review Browser Source for DOM patterns
• Review DataLayer Source for interception patterns
• Learn about creating destinations