Create your own transformer

Build custom transformers for validation, enrichment, or redaction.

The transformer interface

Transformers are functions that receive a context object and return a transformer instance:

type Transformer.Init<Types> = (
  context: Transformer.Context<Types>,
) => Transformer.Instance<Types>;

The context contains:

interface Transformer.Context<Types> {
  config: Partial<Transformer.Config<Types>>;  // Settings, mapping, next chain
  env: Types['env'];                          // Environment dependencies
  logger: Logger;                             // Logging functions
  id: string;                                 // Transformer identifier
  collector: Collector.Instance;              // Collector reference
  ingest?: Ingest;                            // Request metadata from source
}

The returned instance must implement:

interface Transformer.Instance {
  type: string;           // Unique identifier
  config: Transformer.Config;
  push(event, context): Transformer.Result | false | void;
}

Return values

Return	Behavior
{ event }	Continue chain with modified event
void	Continue chain, event unchanged
false	Stop chain, event dropped
{ event, next }	Redirect to a different chain (fan-out)
{ event, respond }	Continue chain with wrapped respond function

Minimal example

import type { Transformer } from '@walkeros/core';

interface Settings {
  removeFields?: string[];
}

interface Types extends Transformer.Types<Settings> {}

export const transformerRedact: Transformer.Init<Types> = (context) => {
  // Destructure what you need from context
  const { config = {} } = context;
  const settings = config.settings || {};

  return {
    type: 'redact',
    config,

    push(event, pushContext) {
      const { logger } = pushContext;

      for (const field of settings.removeFields || []) {
        if (event.data?.[field]) {
          delete event.data[field];
          logger.debug('Redacted field', { field });
        }
      }

      return { event };
    }
  };
};

Push context

The push function receives an event and a push context:

• config: Transformer configuration
• env: Environment dependencies
• logger: Scoped logger for output
• id: Transformer identifier
• collector: Access to collector instance
• ingest: Request metadata from source (optional)

push(event, context) {
  const { logger, id, ingest } = context;

  logger.debug('Processing event', { transformer: id, event: event.name });

  // Access request metadata if available
  if (ingest?.ip) {
    event.data = { ...event.data, clientIp: ingest.ip };
  }

  return { event };
}

Examples

Enrich: add server-side data

push(event, context) {
  event.data = {
    ...event.data,
    enrichedAt: Date.now(),
    region: 'eu-west-1',
  };

  return { event };
}

Validate: custom check

push(event, context) {
  if (!event.data?.id) {
    context.logger.error('Missing required id');
    return false;  // Stop chain
  }

  return { event };
}

Using your transformer

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

await startFlow({
  transformers: {
    redact: {
      code: transformerRedact,
      config: {
        settings: {
          removeFields: ['email', 'phone']
        }
      }
    }
  }
});

Pass-through steps

A path is the multi-step chain through a flow's transformer section. A pass-through step (short: pass) is a single step within a path that has no code and no package. The runtime synthesizes its push function for you, so the step exists purely to wire chain hops, run a cache check, or apply a declarative mapping.

Three variants exist:

1. Before/next chain only

A named hop in the chain. Useful for reusing a chain prefix across destinations.

"enrichServer": { "before": ["filterBots", "sessionLookup"] }

2. Cache only

A step that runs a cache check, often for dedup or response caching.

"dedup": {
  "cache": {
    "stop": true,
    "rules": [{ "key": ["event.id"], "ttl": 60 }]
  }
}

3. Mapping only

A step that runs a declarative event-to-event mapping using Mapping.Config primitives (policy, per-rule policy, mapping[].name, mapping[].ignore).

"redactPII": {
  "mapping": {
    "policy": { "user.email": { "value": "[redacted]" } }
  }
}

Dual semantic of mapping

The mapping field uses the same Mapping.Config shape on destinations and transformer steps, but the semantic differs by position. On a destination, mapping shapes the vendor payload. On a transformer step, it mutates the event itself. Vendor-payload fields (data, per-rule data, silent) are ignored at the transformer position with a one-time warning at init.

Closed schema

Transformer entries use a closed schema: unknown top-level keys are errors. This catches typos like { "rules": [], "stop": true } at the top of a step (forgot the cache: wrapper).

Testing

import { transformerRedact } from '../index';
import type { Transformer, Collector } from '@walkeros/core';
import { createMockLogger } from '@walkeros/core';

// Helper to create transformer context for testing
function createTransformerContext(
  config: Partial<Transformer.Config<Types>> = {},
): Transformer.Context<Types> {
  return {
    config,
    env: {} as Types['env'],
    logger: createMockLogger(),
    id: 'test-redact',
    collector: {} as Collector.Instance,
  };
}

// Helper to create push context for testing
function createPushContext(): Transformer.Context<Types> {
  return {
    config: {},
    env: {} as Types['env'],
    logger: createMockLogger(),
    id: 'test-redact',
    collector: {} as Collector.Instance,
  };
}

describe('Redact Transformer', () => {
  it('removes specified fields', () => {
    const transformer = transformerRedact(
      createTransformerContext({
        settings: { removeFields: ['email'] }
      })
    );

    const event = { data: { id: '123', email: 'test@example.com' } };
    const pushContext = createPushContext();

    const result = transformer.push(event, pushContext);

    expect(result.event.data.email).toBeUndefined();
    expect(result.event.data.id).toBe('123');
  });
});

Package convention

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

walkerOS field in package.json

{
  "walkerOS": { "type": "transformer" },
  "keywords": ["walkeros", "walkeros-transformer"]
}

Field	Required	Description
walkerOS	Yes	Object with type metadata (and platform for sources/destinations)

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

Packages can export a hints record from src/dev.ts to provide lightweight, actionable context beyond schemas and examples, such as validation behavior, enrichment patterns, or troubleshooting tips. Hints are serialized into walkerOS.json and surfaced via MCP tools. See the walkeros-create-transformer skill for details.

Publishing checklist

• walkerOS field in package.json
• Keywords include walkeros and walkeros-transformer
• buildDev() in tsup.config.ts
• dist/walkerOS.json generated on build
• npm run test passes
• npm run lint passes

Next steps

• Learn about creating sources
• Learn about creating destinations