Redis Streams

Server Source code Package Beta

Server-side event streaming to Redis Streams via the ioredis client. Each event is appended to a configurable stream via XADD, with optional approximate MAXLEN trimming for bounded memory usage, JSON or flat serialization modes, and graceful shutdown via destroy().

Where this fits

Redis Streams is a server destination in the walkerOS flow:

Receives events server-side from the collector, serializes them as JSON (or flat fields), and appends them to a Redis Stream for downstream consumers (stream processors, worker queues, pub/sub fan-out, XREAD consumer groups).

Installation

npm install @walkeros/server-destination-redis

Integrated

import { startFlow } from '@walkeros/collector';
import { destinationRedis } from '@walkeros/server-destination-redis';

await startFlow({
destinations: {
  redis: {
    code: destinationRedis,
    config: {
      settings: {
        redis: {
          streamKey: 'walkeros:events',
          url: 'redis://localhost:6379',
        },
      },
    },
  },
},
});

Bundled

Add to your flow.json destinations:

"destinations": {
"redis": {
  "package": "@walkeros/server-destination-redis",
  "config": {
    "settings": {
      "redis": {
        "streamKey": "walkeros:events",
        "url": "redis://localhost:6379"
      }
    }
  }
}
}

See bundled mode setup | CLI reference

Configuration

This destination uses the standard destination config wrapper (consent, data, env, id, ...). For the shared fields see destination configuration. Package-specific fields live under config.settings and are listed below.

Settings

Property	Type	Description	More
redis*	redis	Redis Streams configuration
streamKey*	string	Redis stream key name (like 'walkeros:events'). All events are appended to this stream via XADD.
url	string	Redis connection URL (like 'redis://localhost:6379' or 'rediss://:password@host:6380'). Supports redis:// and rediss:// (TLS) protocols.
options	Record<string, any>	ioredis connection options. Used when url is not provided. Supports host, port, password, db, tls, and all other ioredis options.
maxLen	integer	Maximum stream length. Enables approximate MAXLEN trimming on every XADD to bound memory usage (like 50000).
exactTrimming	boolean	Use exact MAXLEN instead of approximate (~). Not recommended for production. Default: false.
serialization	'json' | 'flat'	Serialization mode. 'json' stores the full event as a single 'event' field (default). 'flat' stores top-level event fields as separate stream entry fields.

* Required fields

Mapping

Per-event rules under config.mapping. For the standard rule fields (consent, condition, data, batch, name, policy) see mapping.

Property	Type	Description	More
streamKey	string	Override Redis stream key for this rule. Takes precedence over settings.redis.streamKey.

Examples

Default stream

An event is appended to the configured Redis stream via XADD with the full event JSON as a single field.

Event

{
  "name": "page view",
  "data": {
    "domain": "www.example.com",
    "title": "walkerOS documentation",
    "referrer": "https://www.walkeros.io/",
    "search": "?foo=bar",
    "hash": "#hash",
    "id": "/docs/"
  },
  "context": {
    "dev": [
      "test",
      1
    ]
  },
  "globals": {
    "pagegroup": "docs"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "child",
      "data": {
        "is": "subordinated"
      }
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "d035e97f7d26a5c8",
  "trigger": "load",
  "entity": "page",
  "action": "view",
  "timestamp": 1700000100,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}

Out

client.xadd([
  "walkeros:events",
  "*",
  "event",
  "json:event"
])

Order event

An order complete event is appended to the Redis stream alongside other event types for downstream consumers.

Event

{
  "name": "order complete",
  "data": {
    "id": "ORD-400",
    "total": 99.99,
    "currency": "EUR"
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "5b3a0b7654d0249a",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700000101,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}

Out

client.xadd([
  "walkeros:events",
  "*",
  "event",
  "json:event"
])

Stream key override

A mapping rule routes the event to a dedicated Redis stream instead of the destination default.

Event

{
  "name": "order complete",
  "data": {
    "id": "ORD-500",
    "total": 42
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "0149b274a307ef06",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700000104,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}

Mapping

{
  "settings": {
    "streamKey": "walkeros:orders"
  }
}

Out

client.xadd([
  "walkeros:orders",
  "*",
  "event",
  "json:event"
])

Exact trim

XADD uses exact MAXLEN trimming to enforce a precise Redis stream length at the cost of extra work.

Event

{
  "name": "page view",
  "data": {
    "domain": "www.example.com",
    "title": "walkerOS documentation",
    "referrer": "https://www.walkeros.io/",
    "search": "?foo=bar",
    "hash": "#hash",
    "id": "/docs/"
  },
  "context": {
    "dev": [
      "test",
      1
    ]
  },
  "globals": {
    "pagegroup": "docs"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "child",
      "data": {
        "is": "subordinated"
      }
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "ada46cb806b04026",
  "trigger": "load",
  "entity": "page",
  "action": "view",
  "timestamp": 1700000103,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}

Out

client.xadd([
  "walkeros:events",
  "MAXLEN",
  5000,
  "*",
  "event",
  "json:event"
])

MAXLEN trim

XADD uses approximate MAXLEN trimming to cap the Redis stream length, discarding older entries efficiently.

Event

{
  "name": "product view",
  "data": {
    "id": "SKU-123",
    "name": "Widget"
  },
  "context": {
    "shopping": [
      "detail",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [],
  "consent": {
    "functional": true
  },
  "id": "77460366027848ac",
  "trigger": "load",
  "entity": "product",
  "action": "view",
  "timestamp": 1700000102,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}

Out

client.xadd([
  "walkeros:events",
  "MAXLEN",
  "~",
  50000,
  "*",
  "event",
  "json:event"
])

The destination creates a single long-lived ioredis client during init(). On flow hot-swap or server shutdown, destroy() calls client.quit() to flush in-flight commands and close the TCP connection gracefully. User-provided clients (wired in via _client) are left untouched.

Stream entry format

Each event is appended via XADD <streamKey> [MAXLEN [~] <n>] * <field> <value> ...:

• streamKey: from settings.redis.streamKey (or mapping.settings.streamKey override)
• *: auto-generated entry ID (ms-timestamp + sequence)
• fields: in json mode, a single event field with the full event JSON-stringified; in flat mode, top-level event keys as separate fields (nested objects JSON-encoded)
• MAXLEN: when settings.redis.maxLen is set, trims the stream to ~N entries. Approximate (~) by default; set exactTrimming: true to trim exactly (slower)

Use mapping.settings.streamKey to route specific events to dedicated streams (e.g. orders to walkeros:orders, identities to walkeros:identities).

Serialization modes

JSON (default)

Stores the full walkerOS event as a single event field. Easiest to consume, decode with JSON.parse().

XADD walkeros:events * event '{"entity":"page","action":"view",...}'

Flat

Stores top-level event fields as separate stream entry fields. Nested objects are JSON-encoded. Useful when downstream consumers want to filter or project specific fields without parsing the full event.

XADD walkeros:events * entity page action view timestamp 1700000100 data '{"id":"Home"}'

Configure via settings.redis.serialization: 'flat'.

Authentication

Redis Cloud / Upstash (TLS + password)

{
"redis": {
  "streamKey": "walkeros:events",
  "url": "rediss://default:$env.REDIS_PASSWORD@my-endpoint.upstash.io:6380"
}
}

AWS ElastiCache (in-VPC, no auth)

{
"redis": {
  "streamKey": "walkeros:events",
  "url": "redis://my-cluster.cache.amazonaws.com:6379"
}
}

Custom options (ACL user, non-default db)

{
"redis": {
  "streamKey": "walkeros:events",
  "options": {
    "host": "redis.example.com",
    "port": 6379,
    "username": "$env.REDIS_USER",
    "password": "$env.REDIS_PASSWORD",
    "db": 1,
    "tls": {}
  }
}
}

MAXLEN trimming

Redis Streams grow unbounded by default. For high-volume pipelines, cap stream length with maxLen:

{
"redis": {
  "streamKey": "walkeros:events",
  "url": "redis://localhost:6379",
  "maxLen": 50000
}
}

Approximate trimming (~) is used by default. Redis trims to "about N" entries with minimal overhead. For exact trimming, set exactTrimming: true (slower, use only when strict bounds are required).