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

# Collector

The collector is the central **processing engine** of walkerOS that receives events from sources, **enriches** them with additional data, applies consent rules, and **routes** them to destinations. It acts as the **intelligent middleware** between event capture and event delivery.

#### What it does[​](#what-it-does "Direct link to What it does")

The Collector transforms raw events into enriched, compliant data streams by:

* **Event processing** - Validates, normalizes, and enriches incoming events
* **Consent management** - Applies privacy rules and user consent preferences
* **Data enrichment** - Adds session data, user context, and custom properties
* **Destination routing** - Sends processed events to configured analytics platforms

#### Key features[​](#key-features "Direct link to Key features")

* **Compatibility** - Works in both web browsers and server environments
* **Privacy-first** - Built-in consent management and data protection
* **Event validation** - Ensures data quality and consistency
* **Flexible routing** - Send events to multiple destinations simultaneously
* **Delivery status** - Built-in per-source and per-destination delivery tracking

#### Role in architecture[​](#role-in-architecture "Direct link to Role in architecture")

In the walkerOS data flow, the collector sits between sources and destinations:

Sources capture events and send them to the collector, which processes and routes them to your chosen destinations like Google Analytics, custom APIs, or data warehouses.

## Installation[​](#installation "Direct link to Installation")

```bash
npm install @walkeros/collector
```

## Basic setup[​](#basic-setup "Direct link to Basic setup")

```typescript
import { startFlow } from '@walkeros/collector';
import { sourceBrowser } from '@walkeros/web-source-browser';

const { collector, elb } = await startFlow({
consent: { functional: true },
sources: {
browser: {
  code: sourceBrowser,
  config: {
    settings: {
      pageview: true,
    },
  },
},
},
destinations: {
console: {
  code: {
    type: 'console',
    config: {},
    push: (event) => console.log('Event:', event),
  },
},
},
});
```

## Settings[​](#settings "Direct link to Settings")

| Property        | Type                           | Description                                                                                                                  | More |
| --------------- | ------------------------------ | ---------------------------------------------------------------------------------------------------------------------------- | ---- |
| `run`           | `boolean`                      | Whether to run collector automatically on initialization                                                                     |      |
| `globalsStatic` | `WalkerOS.Properties`          | Static global properties that persist across collector runs                                                                  |      |
| `sessionStatic` | `Collector.SessionStatic`      | Static session data that persists across collector runs                                                                      |      |
| `logger`        | `Logger.Config`                | Logger configuration (level, handler)                                                                                        |      |
| `queueMax`      | `number`                       | Maximum events retained in collector.queue (late-registration replay). FIFO drop on overflow. Default 1000.                  |      |
| `name`          | `string`                       | Flow name; keys this flow's entry in event.source.release and the observer flowId.                                           |      |
| `release`       | `string`                       | Config release id stamped into event.source.release for this flow.                                                           |      |
| `next`          | `Route`                        | Collector chain: runs once per completed event before the destination fan-out. A stop drops the event for every destination. |      |
| `consent`       | `WalkerOS.Consent`             | Initial consent state                                                                                                        |      |
| `user`          | `WalkerOS.User`                | Initial user data                                                                                                            |      |
| `globals`       | `WalkerOS.Properties`          | Initial global properties                                                                                                    |      |
| `sources`       | `Source.InitSources`           | Source configurations                                                                                                        |      |
| `destinations`  | `Destination.InitDestinations` | Destination configurations                                                                                                   |      |
| `transformers`  | `Transformer.Configs`          | Transformer configurations                                                                                                   |      |
| `stores`        | `Store.Configs`                | Store configurations                                                                                                         |      |
| `custom`        | `WalkerOS.Properties`          | Initial custom implementation-specific properties                                                                            |      |
| `hooks`         | `Collector.Hooks`              | Pipeline observation hooks                                                                                                   |      |
| `observe`       | `Observe`                      | Observation session connect config                                                                                           |      |
| `observers`     | `Collector.Observers`          | Caller-supplied advisory observer functions                                                                                  |      |

## Advanced setup[​](#advanced-setup "Direct link to Advanced setup")

```typescript
import { startFlow } from '@walkeros/collector';
import { sourceBrowser } from '@walkeros/web-source-browser';
import { destinationAPI } from '@walkeros/web-destination-api';

const { collector, elb } = await startFlow({
consent: { functional: true },
sources: {
browser: {
  code: sourceBrowser,
  config: {
    settings: {
      pageview: true,
    },
  },
},
},
destinations: {
api: {
  code: destinationAPI,
  config: {
    settings: {
      url: 'https://analytics.example.com/events',
    },
  },
},
},
logger: {
level: 'DEBUG', // Show all logs (ERROR, INFO, DEBUG)
},
});
```

## Event transformation[​](#event-transformation "Direct link to Event transformation")

The collector works with **mapping** to transform events as they flow through the system. Mapping is configured at the destination level and controls how walkerOS events are converted to vendor-specific formats.

For example, transforming a `product add` event into GA4's `add_to_cart`:

```typescript
destinations: {
  ga4: {
    code: destinationGtag,
    config: {
      mapping: {
        product: {
          add: {
            name: 'add_to_cart',
            data: {
              map: {
                currency: { value: 'USD' },
                value: 'data.price',
                items: { loop: ['nested', { map: { item_id: 'data.id' } }] }
              }
            }
          }
        }
      }
    }
  }
}
```

[**Learn more about mapping →**](/preview/pr-749/docs/mapping.md)

## Collector chain (`next`)[​](#collector-chain-next "Direct link to collector-chain-next")

The collector's `next` is a transformer chain that runs once per event, after the collector completed the event and before it hands the event to the destinations. Every destination receives what the chain produces, so it is the place for work all destinations share, such as validation or bot scoring.

```typescript
import { startFlow } from '@walkeros/collector';
import { transformerBot } from '@walkeros/server-transformer-bot';
import { transformerValidate } from '@walkeros/transformer-validate';

await startFlow({
  transformers: {
    bot: { code: transformerBot },
    validate: { code: transformerValidate },
  },
  next: [
    { match: { key: 'event.name', operator: 'eq', value: 'product impression' }, stop: true },
    'bot',
    'validate',
  ],
  destinations: { /* ... */ },
});
```

`next` accepts the same `Route` as every other chain field (see [Chain resolution](/preview/pr-749/docs/getting-started/flow/routing.md#chain-resolution)); in a flow file it is `collector.next`. Its routes read `{ ingest, event }` with the completed event. What the chain does reaches every destination:

* A `stop` (or a transformer returning `false`) drops the event for **all** destinations. To drop an event for one destination only, put the `stop` in that destination's `before`.
* A `many` fork delivers each copy to the destinations on its own, each with its own `event.id`.
* Consent-queued events, late destinations, and replayed pre-run events receive the chain's output; the chain never runs twice for one event.
* `walker config` can replace `next` for later events.

## Status[​](#status "Direct link to Status")

The collector tracks delivery metrics on `collector.status`, giving you real-time visibility into event flow without external monitoring:

```typescript
const { collector } = await startFlow({ /* ... */ });

// After events have been processed
console.log(collector.status);
// {
//   startedAt: 1707580800000,
//   in: 100,       // Total events received
//   out: 95,       // Total events delivered
//   failed: 5,     // Total failures
//   sources: {
//     browser: { count: 100, lastAt: 1707580900000, duration: 1200 }
//   },
//   destinations: {
//     ga4:  { count: 95, failed: 0, lastAt: 1707580900000, duration: 800,
//             queuePushSize: 0, dlqSize: 0 },
//     meta: { count: 0, failed: 5, lastAt: 1707580850000, duration: 400,
//             queuePushSize: 0, dlqSize: 5 }
//   },
//   // Process-wide drop counters keyed by stepId. See `stepId()` in
//   // `@walkeros/core`. Entries are created lazily on first drop.
//   dropped: {
//     // "collector" is the reserved stepId for the collector-level buffer.
//     // "destination.<id>" keys each destination's drop counters.
//     // Each entry exposes optional { queue?, dlq? } counts.
//   }
// }

// Compute average destination push time
const ga4 = collector.status.destinations.ga4;
const avgMs = ga4.duration / (ga4.count + ga4.failed);
```

### What `failed` counts[​](#what-failed-counts "Direct link to what-failed-counts")

`collector.status.failed` is the single counter for any walkerOS-internal pipeline failure. Contributing sites:

* a destination's `push` threw or returned an error,
* an exception escaped the inner pipeline of `collector.push` or `collector.command` (any uncaught error inside the boundary),
* mapping outer-wrap failures (an internal throw inside `processMappingValue`),
* source startup failures (`code()` threw, `init()` threw, queued-on flush threw),
* transformer init failures,
* destination init failures.

Rejected client input does NOT increment `status.failed` either. An event the pipeline declares invalid (a missing or malformed `name`) resolves as `{ ok: false, invalid: true, error }`, logs one warn without a stack, and counts on `collector.status.sources.<id>.rejected`. That keeps `failed` a measure of walkerOS faults rather than of whatever the public internet posts at a collector endpoint.

User-supplied callback throws are visibility-only: they log at error level but do NOT increment `status.failed`. This keeps the counter a clean pipeline-health signal. Sites in this group:

* `on` subscriptions (`destination.on`, `source.on`, consent rules, `ready`, `run`, `session`, generic),
* mapping `condition`, `fn`, and `validate` callbacks.

In both cases the collector logs at error level:

```typescript
// destination-side failure
logger.scope('<destination type>').error('Push failed', { error, event });

// collector-side boundary failure
logger.error('push failed', { event, ingest, error });
logger.error('command failed', { command, data, error });

// pipeline-internal failures (counted)
logger.error('mapping processing failed', { event, error });
logger.scope('source').error('source factory failed', { sourceId, error });
logger.scope('source').error('source init failed', { sourceId, error });
logger.scope('source').error('source on flush failed', { sourceId, type, error });
logger.scope('transformer:<type>').error('transformer init failed', { transformer, error });
logger.scope('<destination type>').error('destination init failed', { error });

// user-callback failures (logged, NOT counted)
logger.error('mapping condition failed', { event, error });
logger.error('mapping fn failed',        { event, error });
logger.error('mapping validate failed',  { event, error });
logger.scope('on').error('on callback failed', { kind, error });
```

A source whose `init()` throws stays with `config.init === false` instead of being marked initialized. Operators reading `source.config.init` see the source visibly stuck, not falsely healthy.

Alarm on the ratio of `status.failed` to `status.in`. The log message text plus structured fields are enough to filter destination vs boundary failures downstream.

**PII note.** Boundary error logs include the full failing event payload so operators can reproduce. If your event payloads carry sensitive data, configure redaction at the logger layer; do not parse and rewrite at every call site.

### Fatal errors[​](#fatal-errors "Direct link to Fatal errors")

Throw `FatalError` (exported from `@walkeros/core`) for invariant violations or operator-initiated aborts that must crash the host process. Standard `Error` is absorbed by the boundary catch, logged, and counted. `FatalError` bypasses the catch and propagates, so a supervisor (CLI runner, Express server, container orchestrator) can terminate cleanly.

```typescript
import { FatalError } from '@walkeros/core';

if (!config.apiKey) {
  throw new FatalError('apiKey missing — refusing to start');
}
```

Queue sizes and DLQ sizes can be read directly from destination instances, or from the status snapshot:

```typescript
const dest = collector.destinations['meta'];
const queueSize = dest.queuePush?.length || 0;
const dlqSize = dest.dlq?.length || 0;

// Or from status (point-in-time snapshots refreshed after each push pass):
const metaStatus = collector.status.destinations['meta'];
metaStatus.queuePushSize;   // events waiting for consent
metaStatus.dlqSize;         // failed pushes retained for triage

// Drop counts live on the process-wide `status.dropped` map, keyed by
// stepId (use `stepId()` from `@walkeros/core`):
import { stepId } from '@walkeros/core';
const metaDrops = collector.status.dropped[stepId('destination', 'meta')];
metaDrops?.queue ?? 0; // monotonic count of evicted consent-queued events
metaDrops?.dlq ?? 0;   // monotonic count of evicted DLQ entries
```

## Buffer bounds[​](#buffer-bounds "Direct link to Buffer bounds")

The collector keeps three internal buffers per process. Each is size-bounded with a configurable cap; on overflow the oldest entries are evicted (FIFO), the corresponding `dropped` counter is incremented, and a warning is logged once per overflow window.

| Buffer                  | Purpose                                            | Default cap | Config                        |
| ----------------------- | -------------------------------------------------- | ----------- | ----------------------------- |
| `collector.queue`       | Replay buffer for late-registered destinations     | 1000        | `Collector.Config.queueMax`   |
| `destination.queuePush` | Per-destination consent-denied buffer              | 1000        | `Destination.Config.queueMax` |
| `destination.dlq`       | Per-destination dead-letter queue of failed pushes | 100         | `Destination.Config.dlqMax`   |

Each step has its own knob; no cascade. Set the collector cap on the collector, and per-destination caps on each destination:

```typescript
const { collector } = await startFlow({
  // Collector replay buffer cap
  queueMax: 5_000,
  destinations: {
    bigquery: {
      code: bigqueryDestination,
      config: {
        // Keep more failed rows for triage on this destination
        dlqMax: 1_000,
      },
    },
  },
});
```

Operators alarm on the `dropped` counters to detect sustained overflow. Counters live on `collector.status.dropped`, keyed by stepId (build the key with `stepId()` from `@walkeros/core`):

```typescript
import { stepId } from '@walkeros/core';

// Collector replay buffer drops: traffic bursts outrunning destination
// registration.
const collectorDrops =
  collector.status.dropped[stepId('collector')]?.queue ?? 0;

// Destination drops by buffer:
//  - queue: consent-denied events evicted from the destination's queuePush
//  - dlq:   failed-push entries evicted from the destination's DLQ
const ga4Drops = collector.status.dropped[stepId('destination', 'ga4')];
ga4Drops?.queue ?? 0;
ga4Drops?.dlq ?? 0;   // sustained non-zero rate signals destination outage
```

***

## In the complete example[​](#in-the-complete-example "Direct link to In the complete example")

The server collector of [`flow-complete.json`](https://github.com/elbwalker/walkerOS/blob/main/packages/cli/examples/flow-complete.json), the tested example flow, sets a logger and runs its chain once per event:

`/flows/server/collector`

```json
{
  "logger": {
    "level": "$var.logLevel"
  },
  "next": [
    "fingerprint",
    "bot",
    {
      "match": {
        "key": "event.trigger",
        "operator": "regex",
        "value": "^(impression|visible)$",
        "not": true
      },
      "next": "enrich"
    }
  ]
}
```

The collector sits between sources and destinations and runs its own chain once per event. [Guide chapter `tour`](https://github.com/elbwalker/walkerOS/blob/main/packages/cli/examples/flow-complete.md#tour-one-file-one-event)

The web collector carries the page defaults. `globals` can be overridden by the page, `globalsStatic` survives a run reset, `custom` holds values that are not event data:

`/flows/web/collector/globals`

```json
{
  "language": "en"
}
```

globals.language is the default the page can override; the contract requires it. [Guide chapter `web-entry`](https://github.com/elbwalker/walkerOS/blob/main/packages/cli/examples/flow-complete.md#web-entry-start-where-the-shop-already-is)

`/flows/web/collector/globalsStatic`

```json
{
  "site": "$env.SITE_ID:shop"
}
```

globalsStatic survives a run reset; the site id belongs there. [Guide chapter `web-entry`](https://github.com/elbwalker/walkerOS/blob/main/packages/cli/examples/flow-complete.md#web-entry-start-where-the-shop-already-is)

`/flows/web/collector/custom`

```json
{
  "environment": "$env.ENVIRONMENT"
}
```

custom carries values for the pipeline that are not event data. [Guide chapter `web-entry`](https://github.com/elbwalker/walkerOS/blob/main/packages/cli/examples/flow-complete.md#web-entry-start-where-the-shop-already-is)

## See also[​](#see-also "Direct link to See also")

* [**Operating modes**](/preview/pr-749/docs/getting-started/modes.md): integrated vs bundled approaches
* [**CLI documentation**](/preview/pr-749/docs/apps/cli.md): configure with JSON and build with CLI
* [**Sources**](/preview/pr-749/docs/sources.md): available event capture sources
* [**Destinations**](/preview/pr-749/docs/destinations.md): available event delivery destinations