> 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")

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

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

```
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,
      session: true,
    },
  },
},
},
destinations: {
console: {
  code: true, // Built-in code destination
  config: {
    settings: {
      push: "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.                                          |      |
| `consent`       | `WalkerOS.Consent`             | Initial consent state                                                                                       |      |
| `user`          | `any`                          | 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")

```
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,
      session: 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`:

```
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 →**](https://www.walkeros.io/docs/mapping.md)

## 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:

```
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:

```
// 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.

```
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:

```
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:

```
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`):

```
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
```

***

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

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