> Part of the walkerOS documentation. Project overview and full index: <https://www.walkeros.io/llms.txt>

# 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[​](#the-source-interface "Direct link to 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[​](#types-bundle "Direct link to 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[​](#context-destructuring "Direct link to 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' } });
};
```

`env` is the author's dependency bag: you provide platform and vendor dependencies (like `window`, `document`, custom APIs) when configuring the source. The collector provides five capabilities into that bag, and they always win: `push`, `elb`, `command`, `logger` and `sources` are runtime-owned, so setting one of them in `env` has no effect.

### Ending the pipeline elsewhere: `terminus`[​](#ending-the-pipeline-elsewhere-terminus "Direct link to ending-the-pipeline-elsewhere-terminus")

A source registration that needs the pipeline to end somewhere other than the collector declares `terminus` next to `code`. A terminus receives the event exactly as the source emitted it, and the entire pipeline is skipped: no `before`/`next` chains, no source `cache`, no `state`, no `mapping` (including `ignore` and `consent`), no minted span id, no per-source ingest, no `respond`, no `status.sources` counting and no observability records. This is a total bypass by design, so there is one rule to know rather than a list of exceptions. It exists for deterministic capture at the source→collector boundary (step examples, dev tooling); production flows leave it unset, and stored flow configs cannot carry it (validation rejects unknown source keys).

## Per-scope context (server sources)[​](#per-scope-context-server-sources "Direct link to 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
  });

  // Normalize the platform request into the shared scope shape first, so
  // `config.ingest` paths resolve the same here as on every other source.
  const scope = buildScopeFromNodeRequest(req, {
    body: req.body,
    defaultProtocol: 'http',
  });

  // Each call to withScope builds a fresh Ingest from the scope via
  // `config.ingest` mapping, captures `respond`, and runs the body
  // with a per-scope env. Concurrent scopes never crosstalk.
  await context.withScope(scope, respond, async (env) => {
    // toEventList decides how many events the body carries.
    for (const event of toEventList(scope.body)) await env.push(event);
    respond({ body: { success: true } });
  });
};
```

A new server source must build a [scope](https://www.walkeros.io/docs/sources/scope.md) and use [`toEventList`](https://www.walkeros.io/docs/sources/envelope.md), so `config.ingest` paths and accepted body forms match every other source. `buildScopeFromNodeRequest`, `normalizeHeaders`, `normalizeQuery`, `normalizeBody`, `toEventList` and `isBatchBody` all come from `@walkeros/core`.

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[​](#minimal-example "Direct link to 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[​](#complete-example-event-api-source "Direct link to 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[​](#using-your-source "Direct link to 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[​](#testing-your-source "Direct link to Testing your source")

### Test Utilities[​](#test-utilities "Direct link to 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[​](#test-example "Direct link to 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[​](#reading-a-step-back-from-the-collector "Direct link to 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)[​](#setup-lifecycle-optional "Direct link to 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](https://www.walkeros.io/docs/destinations/create-your-own.md#setup-lifecycle-optional) on the destinations page for the full concept, and [walkeros setup CLI command](https://www.walkeros.io/docs/apps/cli.md#setup-command) for the operator-facing command.

## Conditional activation with `require`[​](#conditional-activation-with-require "Direct link to 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[​](#common-patterns "Direct link to Common patterns")

### Polling for API Readiness[​](#polling-for-api-readiness "Direct link to 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[​](#source-as-adapter-pattern "Direct link to 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[​](#key-concepts "Direct link to 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[​](#package-convention "Direct link to Package convention")

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

### `walkerOS` field in package.json[​](#walkeros-field-in-packagejson "Direct link to walkeros-field-in-packagejson")

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

| Field      | Required | Description                                |
| ---------- | -------- | ------------------------------------------ |
| `walkerOS` | Yes      | Object with `type` and `platform` metadata |

### Build-time generation[​](#build-time-generation "Direct link to 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[​](#optional-hints "Direct link to 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[​](#publishing-checklist "Direct link to 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[​](#next-steps "Direct link to Next steps")

* Review [Browser Source](https://www.walkeros.io/docs/sources/web/browser.md) for DOM patterns
* Review [DataLayer Source](https://www.walkeros.io/docs/sources/web/dataLayer.md) for interception patterns
* Learn about [creating destinations](https://www.walkeros.io/docs/destinations/create-your-own.md)