/walkeros-understanding-flow

name: walkeros-understanding-flow
description:
  Use when learning walkerOS architecture, understanding data flow, or designing
  composable event pipelines. Covers Source→Collector→Destination pattern and
  separation of concerns.

Understanding walkerOS Flow

Overview

walkerOS follows a **Source → Collector → Destination(s)** architecture for composable, modular event processing.

**Core principle:** Separation of concerns. Each component has one job. Components are composable and replaceable.

The Flow Pattern

[Source.before] → Sources → [Source.next] → Collector → [Collector.next] → [Dest.before] → Destinations → [Dest.next]
(Preprocessing)   (Capture)  (Source chain)  (Processing) (Collector chain) (Dest. chain)   (Delivery)     (Post-push)

Consent-exempt:                                                                                   Post-consent:
- Decode           - Validation     - Event creation  - Validation     - Redaction      - Push     - Audit logging
- Validate format  - Enrichment     - Enrichment      - Bot scoring    - Per-vendor     - Send     - Response parsing
- Authenticate     - Redaction      - Consent check   - Drop for all   - Drop for one   - Store    - Webhooks

`collector.next` runs once per event before the destination fan-out, so its work (and a `stop`) applies to every destination. `destination.before` runs per destination.

Key Concepts

Composability

A Flow combines components. You can:

• Use multiple sources feeding one collector
• Route events to multiple destinations
• Swap components without changing others

The Flow Type

See [packages/core/src/types/flow.ts](../../packages/core/src/types/flow.ts) for the canonical interface.

// Conceptual structure (see source for full type)
interface Flow {
  stores?: Record<string, Store>;
  sources?: Record<string, Source>;
  transformers?: Record<string, Transformer>;
  destinations?: Record<string, Destination>;
  collector?: Collector.InitConfig;
}

Universal Push Interface

**All components communicate via `push` functions:**

| Component | Push Signature | Purpose | | ----------- | ----------------------------- | --------------------- | | Source | `push(input) → events` | Capture external data | | Collector | `push(event) → void` | Process and route | | Destination | `push(event, context) → void` | Transform and deliver |

The collector's `elb()` takes a partial event (`elb({ name: 'page view', data })`) or a `walker` command; `collector.push` takes event objects only. `startFlow` returns the primary source's `push` as `elb`, or the collector's `elb` when no source is configured. The browser source's push takes the same partial event; a server source's push is its request handler.

startFlow Helper

See [packages/collector/src/flow.ts](../../packages/collector/src/flow.ts) for the `startFlow` function.

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

const { collector, elb } = await startFlow({
  stores: {
    /* key-value storage, init first, destroy last */
  },
  sources: {
    /* ... */
  },
  transformers: {
    /* ... */
  },
  destinations: {
    /* ... */
  },
});

Ingest: Mutable Pipeline Context

walkerOS uses a two-layer data model:

• **Event** - strict schema, structured analytics data (name, data, context,

etc.)

• **Ingest** - free-form mutable context that flows alongside events

Any step can read and write arbitrary keys on ingest. The runtime manages `_meta`:

• `_meta.hops` - increments per step (safety valve at 256)
• `_meta.path` - ordered list of step IDs visited (path[0] = source ID)

// In a transformer
push: async (event, context) => {
  context.ingest.botScore = 0.95; // Write freely
  context.ingest.geo = { country: 'DE' };
  console.log(context.ingest._meta.path); // ['express', 'validate', 'enrich']
  return { event };
};

Ingest is cloned per destination to prevent cross-contamination in parallel processing. After a destination push, the response is available at `ingest._response`.

Stores

Three pipeline components (Source / Transformer / Destination) plus Stores as **passive infrastructure**. Stores are not a fourth pipeline stage - they're key-value storage that other components consume via `env`. They have no `push`, no `next`, no `before`; they sit alongside the pipeline rather than inside it.

• Referenced via `$store.storeId` in `env` values (bundled mode) or by passing

the same store definition object used in `stores` (integrated mode)

• **Init first, destroy last** - stores are available before any source,

transformer, or destination starts, and outlive them on shutdown

• **No chains** - stores don't participate in the event pipeline. Components

access them through their `env`.

• Implementations: `@walkeros/server-store-fs` (async, filesystem),

`@walkeros/server-store-s3` (async, S3-compatible), `@walkeros/server-store-gcs`, `@walkeros/server-store-sheets`. The collector ships a built-in in-memory cache tier — enable it on any store via `Flow.Store.cache` instead of declaring a separate memory store.

{
  "stores": {
    "data": { "package": "@walkeros/server-store-fs" }
  },
  "transformers": {
    "fingerprint": {
      "package": "@walkeros/server-transformer-fingerprint",
      "env": { "store": "$store.data" }
    }
  }
}

See [walkeros-understanding-stores](../walkeros-understanding-stores/SKILL.md) for the full store interface and lifecycle.

Separation of Concerns

| Concern | Handled By | NOT Handled By | | ---------------- | -------------- | ----------------------- | | Event capture | Sources | Collector, Destinations | | Event structure | Event model | Components | | Consent checking | Collector | Sources, Destinations | | Transformation | Mapping system | Raw push calls | | Delivery |