> Part of the walkerOS documentation. Project overview and full index: <https://www.walkeros.io/llms.txt>

# Redis Streams

<!-- -->

[Server](#)[ ](https://github.com/elbwalker/walkerOS/tree/main/packages/server/destinations/redis)

<!-- -->

[Source code](https://github.com/elbwalker/walkerOS/tree/main/packages/server/destinations/redis)[ ](https://www.npmjs.com/package/@walkeros/server-destination-redis)

<!-- -->

[Package](https://www.npmjs.com/package/@walkeros/server-destination-redis)

<!-- -->

Beta

Server-side event streaming to [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/) via the [`ioredis`](https://github.com/redis/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[​](#installation "Direct link to Installation")

```
npm install @walkeros/server-destination-redis
```

* Integrated
* Bundled

```
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',
          },
        },
      },
    },
  },
});
```

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](https://www.walkeros.io/docs/getting-started/modes/bundled.md) | [CLI reference](https://www.walkeros.io/docs/apps/cli.md)

## Configuration[​](#configuration "Direct link to Configuration")

This <!-- -->destination<!-- --> uses the standard <!-- -->destination<!-- --> config wrapper (consent, data, env, id, ...). For the shared fields see [destination<!-- --> configuration](https://www.walkeros.io/docs/destinations.md#configuration). Package-specific fields live under `config.settings` and are listed below.

## Settings[​](#settings "Direct link to 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[​](#mapping "Direct link to Mapping")

Per-event rules under `config.mapping`. For the standard rule fields (consent, condition, data, batch, name, policy) see [mapping](https://www.walkeros.io/docs/mapping.md).

| 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": "52f6121f62677cb3",
  "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": "cd8d598791c59997",
  "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": "941f59741bbfeba8",
  "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": "3e6e6112dcd6f101",
  "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": "8cf76f8d39318fe6",
  "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[​](#stream-entry-format "Direct link to 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[​](#serialization-modes "Direct link to Serialization modes")

### JSON (default)[​](#json-default "Direct link to 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[​](#flat "Direct link to 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[​](#authentication "Direct link to Authentication")

### Redis Cloud / Upstash (TLS + password)[​](#redis-cloud--upstash-tls--password "Direct link to 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)[​](#aws-elasticache-in-vpc-no-auth "Direct link to 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)[​](#custom-options-acl-user-non-default-db "Direct link to 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[​](#maxlen-trimming "Direct link to 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).