Ask your AI

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

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

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

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

npm install @walkeros/collector

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,
    },
  },
},
},
destinations: {
console: {
  code: {
    type: 'console',
    config: {},
    push: (event) => console.log('Event:', event),
  },
},
},
});

Settings

Property	Type	Description
`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

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

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 →

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.

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

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

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

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

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

In the complete example

The server collector of flow-complete.json, the tested example flow, sets a logger and runs its chain once per event:

/flows/server/collector

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

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

{
  "language": "en"
}

globals.language is the default the page can override; the contract requires it. Guide chapter web-entry

/flows/web/collector/globalsStatic

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

globalsStatic survives a run reset; the site id belongs there. Guide chapter web-entry

/flows/web/collector/custom

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

custom carries values for the pipeline that are not event data. Guide chapter web-entry

What it does​

Key features​

Role in architecture​

Installation​

Basic setup​

Settings​

Advanced setup​

Event transformation​

Collector chain (next)​

Status​

What failed counts​

Fatal errors​

Buffer bounds​

In the complete example​

See also​