add a redis streams transport as @node-ts/bus-redis

.changeset/redis-streams-transport.md

@@ -0,0 +1,7 @@
+---
+'@node-ts/bus-redis': minor
+---
+
+Add `RedisTransport`, a Redis Streams transport for Redis 7+ and Valkey 7.2+ (#347). Each service's queue is a stream read by a consumer group, with fan-out through subscription sets, delayed retries in a sorted set, messages taken over from stopped processes after `visibilityTimeoutMs`, a dead letter stream per queue with `deadLetterRetentionMs`, `bus provision` support with a `redis-acl` runtime permission plan, and replies. It needs `redis` (node-redis) 6 as a peer dependency.
+
+**Breaking:** this replaces `@node-ts/bus-redis` 0.x, the Redis lists transport that shared its name: 0.1.8 (`latest`, for bus-core 1.x) and 0.1.1–0.1.7 and 0.1.9 (the `0.x` tag, for bus-core 0.6 with inversify). It's a new transport with a new configuration and key layout, and messages in 0.x queues aren't moved: see "Migrating from @node-ts/bus-redis 0.x" at https://node-ts.github.io/bus/transports/redis.

.circleci/config.yml

@@ -16,6 +16,7 @@ jobs:
           RABBITMQ_URL: amqp://guest:guest@127.0.0.1
           POSTGRES_URL: postgres://postgres:password@127.0.0.1:5432/postgres
           MONGODB_URL: mongodb://127.0.0.1:27017/workflows?directConnection=true
+          REDIS_URL: redis://127.0.0.1:6379
       - image: rabbitmq:3-management
       - image: localstack/localstack:3
         environment:
@@ -26,6 +27,10 @@ jobs:
       # A single-node replica set, since the outbox's transactions need one. "Wait for integration infra" initiates it.
       - image: mongo:7
         command: ['--replSet', 'rs0']
+      - image: redis:8
+      # The Redis transport's tests run against Valkey too, on a port of its own since containers share the network
+      - image: valkey/valkey:8
+        command: ['valkey-server', '--port', '6381']
     resource_class: large
     steps:
       - checkout
@@ -71,6 +76,8 @@ jobs:
             wait_for_port 5672
             wait_for_port 5432
             wait_for_port 27017
+            wait_for_port 6379
+            wait_for_port 6381
             node .circleci/init-mongo-replica-set.mjs
             for i in $(seq 1 60); do
               curl -sf http://127.0.0.1:4566/_localstack/health | grep -q '"sqs": "\(available\|running\)"' && break
@@ -80,6 +87,9 @@ jobs:
       - run:
           name: Integration Test
           command: pnpm run test:integration
+      - run:
+          name: Integration Test (Redis transport on Valkey)
+          command: REDIS_URL=redis://127.0.0.1:6381 pnpm exec dotenv -e test.env -- jest --runInBand --coverage=false packages/bus-redis/src/redis-transport.integration.ts
       # Save workspace for subsequent jobs (i.e. test)
       - persist_to_workspace:
           root: .

CLAUDE.md

@@ -32,7 +32,7 @@ pnpm exec dotenv -e test.env -- jest packages/bus-sqs/src/sqs-transport.spec.ts
 - Jest is configured once at the root (`jest.config.ts`, ts-jest using `tsconfig.test.json`; a `moduleNameMapper` lets relative `.js` imports, as in generated message types, resolve to `.ts` source); `test/setup.ts` drops the default logger's `@node-ts/…` console warnings and errors (set `BUS_TEST_LOGS=true` to see them). Other console output still shows, so don't leave `console.log` in tests.
 - Tests: `*.spec.ts` = unit, `*.integration.ts` = integration. `packages/bus-test/src/in-memory.integration.ts` runs bus-test's own round trip suites over the in-memory queue and persistence.
 - **Build before testing across packages**: every package has `main: ./dist/index.js`, so e.g. bus-sqs tests import the built bus-core and bus-test, not their source. Changes in bus-core need `pnpm build` before they're visible to other packages.
-- Integration tests for adapters need local infra; `docker compose up -d --wait` starts it all from the root `docker-compose.yml`, and waits until it's healthy (Mongo's healthcheck initiates its replica set). Endpoints default to the compose ports and can be overridden with env vars (listed in `test.env`): `LOCALSTACK_ENDPOINT` (default `http://localhost:4566`, dummy AWS creds in `test.env`), `RABBITMQ_URL` (`amqp://guest:guest@0.0.0.0`), `POSTGRES_URL` (`postgres://postgres:password@localhost:6432/postgres`), `MONGODB_URL` (`mongodb://localhost:27017/workflows?directConnection=true`, a single-node replica set, see `packages/bus-mongodb/CLAUDE.md`). CI runs all integration tests against CircleCI secondary containers (`.circleci/config.yml`).
+- Integration tests for adapters need local infra; `docker compose up -d --wait` starts it all from the root `docker-compose.yml`, and waits until it's healthy (Mongo's healthcheck initiates its replica set). Endpoints default to the compose ports and can be overridden with env vars (listed in `test.env`): `LOCALSTACK_ENDPOINT` (default `http://localhost:4566`, dummy AWS creds in `test.env`), `RABBITMQ_URL` (`amqp://guest:guest@0.0.0.0`), `POSTGRES_URL` (`postgres://postgres:password@localhost:6432/postgres`), `MONGODB_URL` (`mongodb://localhost:27017/workflows?directConnection=true`, a single-node replica set, see `packages/bus-mongodb/CLAUDE.md`), `REDIS_URL` (`redis://localhost:6380`; compose also has Valkey on 6381). CI runs all integration tests against CircleCI secondary containers (`.circleci/config.yml`), and bus-redis' again against Valkey.
 - Formatting is automatic: a `.claude/settings.json` hook runs prettier on every file Claude edits, and husky + lint-staged format on commit.
 - Linting: `eslint.config.mjs` is `recommended` from ESLint and typescript-eslint, plus the type-aware `no-floating-promises`, `no-misused-promises` and `await-thenable`, with `eslint-config-prettier` so it never fights prettier. Type information comes from `tsconfig.eslint.json` (specs and `test/` fixtures included), and cross-package types come from each package's `dist`, so build first. lint-staged also runs ESLint on staged TS/JS files. Await or `.catch` every promise; use `void` only for deliberate fire-and-forget, with a comment saying why. Unused parameters are allowed when prefixed with `_`.
 - Dependency updates are grouped by Renovate (`renovate.json`); the maintainer installs the Renovate GitHub app.
@@ -46,7 +46,7 @@ pnpm exec dotenv -e test.env -- jest packages/bus-sqs/src/sqs-transport.spec.ts
 
 - `bus-messages` — `Message`/`Command`/`Event` base types and `MessageAttributes`. Messages identify themselves by `$name` (usually a static `NAME`) and `$version`.
 - `bus-core` — the bus itself, the `Transport`/`Persistence`/`Serializer`/`Receiver` interfaces (`Transport.endpointName` is the service queue's name), and in-memory defaults (`InMemoryQueue`, `InMemoryPersistence`, `JsonSerializer`).
-- Transports: `bus-sqs` (SNS topics fanned out to SQS queues, with policy generation), `bus-rabbitmq`, and `PostgresTransport` in `bus-postgres` (queues in shared Postgres tables, no broker; see `packages/bus-postgres/CLAUDE.md`).
+- Transports: `bus-sqs` (SNS topics fanned out to SQS queues, with policy generation), `bus-rabbitmq`, `PostgresTransport` in `bus-postgres` (queues in shared Postgres tables, no broker; see `packages/bus-postgres/CLAUDE.md`), and `bus-redis` (a Redis Stream per queue, read by a consumer group, on Redis 7+/Valkey 7.2+; see `packages/bus-redis/CLAUDE.md`).
 - Persistence (workflow state and delayed messages): `bus-postgres`, `bus-mongodb`.
 - `bus-sqs-lambda` — a `Receiver` that feeds Lambda SQS events into the bus instead of the bus polling the transport.
 - `bus-opentelemetry` — `openTelemetry()` middleware for spans, W3C trace context (carried in message attributes) and messaging metrics, with only `@opentelemetry/api` as a peer. See `packages/bus-opentelemetry/CLAUDE.md`.

MIGRATING.md

@@ -160,6 +160,12 @@ Two things behave differently:
 - **The AMQP `messageId` property is now the bus' `messageId`**, so it's the same for every message sent with the same id, rather than a new UUID per publish. `sentAt` is carried in a `sentAt` header.
 - `amqplib` is now version 2.2. It ships its own types, so remove `@types/amqplib`. `heartbeat=0` in a connection string now disables heartbeats.
 
+## @node-ts/bus-redis
+
+- **It's a new transport on Redis Streams**, replacing the 0.x Redis lists transport from the `node-ts/bus-redis` repository: 0.1.8 (npm's `latest`, for bus-core 1.x, configured with `new RedisTransport({ queueName, connectionString })`), and 0.1.1–0.1.7 and 0.1.9 (the `0.x` tag, for bus-core 0.6, loaded as the inversify `BusRedisModule`). Pass `new RedisTransport({ queueName, connection: { url } })` to `withTransport()`: `connectionString` is now `connection.url`, and inversify isn't used. It needs Redis 7.0+ or Valkey 7.2+, and `redis` (node-redis) 6 as a peer dependency.
+- **`maxRetries` is removed**: use the bus' recoverability policy, `defaultRecoverability({ maxAttempts })`. `visibilityTimeout` is now `visibilityTimeoutMs`, `subscriptionsKeyPrefix` is now `keyPrefix`, and `withScheduler` is removed.
+- **Messages in 0.x queues aren't moved**, and the two versions can't send each other messages. Drain the old queues, then provision and deploy every service that exchanges messages together. See [Migrating from @node-ts/bus-redis 0.x](https://node-ts.github.io/bus/transports/redis#migrating-from-node-ts-bus-redis-0-x).
+
 ## @node-ts/bus-sqs-lambda
 
 - **The receiver applies the bus' recoverability policy.** A record that's retried has its visibility timeout set to the policy's delay, rather than waiting out the queue's visibility timeout, and a record that's dead-lettered, by the policy or `failMessage()`, is moved to the dead letter queue with its failure metadata and reported to Lambda as handled.

README.md

@@ -21,6 +21,7 @@ Requires Node.js 24 or later.
 - [@node-ts/bus-sqs](https://github.com/node-ts/bus/tree/master/packages/bus-sqs) - An Amazon SQS transport
 - [@node-ts/bus-sqs-lambda](https://github.com/node-ts/bus/tree/master/packages/bus-sqs-lambda) - A receiver that handles SQS messages in AWS Lambda
 - [@node-ts/bus-postgres](https://github.com/node-ts/bus/tree/master/packages/bus-postgres) - A Postgres persistence for workflow state, and a Postgres transport that needs no broker
+- [@node-ts/bus-redis](https://github.com/node-ts/bus/tree/master/packages/bus-redis) - A Redis Streams transport, for Redis or Valkey
 - [@node-ts/bus-mongodb](https://github.com/node-ts/bus/tree/master/packages/bus-mongodb) - A MongoDB persistence for workflow state
 - [@node-ts/bus-opentelemetry](https://github.com/node-ts/bus/tree/master/packages/bus-opentelemetry) - OpenTelemetry tracing and metrics, as bus middleware
 - [@node-ts/bus-nestjs](https://github.com/node-ts/bus/tree/master/packages/bus-nestjs) - A NestJS module: handlers and workflows as providers, and the bus run by Nest's lifecycle

docker-compose.yml

@@ -22,6 +22,28 @@ services:
     ports:
       - '6432:5432'
 
+  redis:
+    image: redis:8
+    ports:
+      - '6380:6379'
+    healthcheck:
+      test: ['CMD', 'redis-cli', 'ping']
+      interval: 2s
+      timeout: 5s
+      retries: 30
+
+  # Optional: run the Redis transport's tests against Valkey with
+  # REDIS_URL=redis://localhost:6381
+  valkey:
+    image: valkey/valkey:8
+    ports:
+      - '6381:6379'
+    healthcheck:
+      test: ['CMD', 'valkey-cli', 'ping']
+      interval: 2s
+      timeout: 5s
+      retries: 30
+
   mongo:
     image: mongo:7
     # A single-node replica set, since the outbox's transactions need one. The

docs/.vitepress/config.mts

@@ -127,6 +127,7 @@ const infrastructure: DefaultTheme.SidebarItem[] = [
       { text: 'Amazon SQS', link: '/transports/amazon-sqs' },
       { text: 'SQS and Lambda', link: '/transports/sqs-lambda' },
       { text: 'Postgres', link: '/transports/postgres' },
+      { text: 'Redis', link: '/transports/redis' },
       { text: 'Custom transports', link: '/transports/custom' }
     ]
   },

docs/guide/provisioning.md

@@ -16,6 +16,7 @@ Everything is worked out from the bus' configuration: its transport, persistence
 | [Amazon SQS](/transports/amazon-sqs#provisioning)       | An SNS topic per message, the service queue and dead letter queue, a subscription per handled message, and the queue's access policy |
 | [RabbitMQ](/transports/rabbitmq#provisioning)           | A fanout exchange per message, the service queue with its retry and dead letter queues, and the bindings                             |
 | [Postgres transport](/transports/postgres#provisioning) | The schema, the messages, queues, subscriptions and dead letters tables, the service's queue, and a subscription per handled message |
+| [Redis](/transports/redis#provisioning)                 | The service queue's stream and its consumer group, and the queue in the subscription set of each handled message                     |
 | [Postgres](/persistence/postgres#provisioning)          | The schema, a table per workflow state with indexes on its lookups, and the outgoing messages table                                  |
 | [MongoDB](/persistence/mongodb#provisioning)            | A collection per workflow state with indexes on its lookups, and the outgoing messages collection                                    |
 
@@ -118,7 +119,7 @@ SqsTransport (7 resources)
 
 - `adapters` has an entry for each transport or persistence that provisions anything, in the order they ran: the persistence first, then the transport.
 - Each resource has a `type`, its `name` (an ARN for AWS resources) and, for some, the `properties` it's created with. The types are listed on each adapter's page.
-- `runtimePermissions` is only included with `--permissions`. Its `format` says what `document` is: `iam-policy` (an IAM policy document), `rabbitmq-permissions` (the `configure`, `write` and `read` expressions for the vhost), `sql` (a list of grant statements) or `mongodb-privileges` (the privileges of a role). The `ProvisionReport` type, exported by `@node-ts/bus-cli`, describes it.
+- `runtimePermissions` is only included with `--permissions`. Its `format` says what `document` is: `iam-policy` (an IAM policy document), `rabbitmq-permissions` (the `configure`, `write` and `read` expressions for the vhost), `sql` (a list of grant statements), `redis-acl` (the ACL rules of a user) or `mongodb-privileges` (the privileges of a role). The `ProvisionReport` type, exported by `@node-ts/bus-cli`, describes it.
 
 ## Checking resources at startup
 
@@ -158,5 +159,5 @@ A [custom transport](/transports/custom) or [persistence](/persistence/custom) i
 
 ## See also
 
-- [Amazon SQS](/transports/amazon-sqs#provisioning), [RabbitMQ](/transports/rabbitmq#provisioning), the [Postgres transport](/transports/postgres#provisioning), [Postgres](/persistence/postgres#provisioning) and [MongoDB](/persistence/mongodb#provisioning), for what each provisions and the permissions it needs
+- [Amazon SQS](/transports/amazon-sqs#provisioning), [RabbitMQ](/transports/rabbitmq#provisioning), the [Postgres transport](/transports/postgres#provisioning), [Redis](/transports/redis#provisioning), [Postgres](/persistence/postgres#provisioning) and [MongoDB](/persistence/mongodb#provisioning), for what each provisions and the permissions it needs
 - [`BusInstance.provision`](/api/bus-core/classes/BusInstance#provision), [`ProvisioningPlan`](/api/bus-core/interfaces/ProvisioningPlan) and [`ResourcesNotProvisioned`](/api/bus-core/classes/ResourcesNotProvisioned) in the API reference

docs/guide/recoverability.md

@@ -83,13 +83,14 @@ A message that was failed with `failMessage()` without anything throwing has a `
 | ------------------------------------ | --------------------------------------------------------------------------------------------------------------------------- |
 | [RabbitMQ](/transports/rabbitmq)     | an AMQP header on the dead-lettered message                                                                                 |
 | [Amazon SQS](/transports/amazon-sqs) | an SQS message attribute on the dead-lettered message. The SNS envelope in its body, with the message's attributes, is kept |
+| [Redis](/transports/redis)           | a `bus-failure` field of the entry in the queue's dead letter stream                                                        |
 | In-memory queue                      | the dead-lettered message's `raw.headers`                                                                                   |
 
 Transports also add it to messages they dead-letter themselves because they can't be parsed. `bus-failure` is reserved, so outgoing middleware can't set it.
 
 ## Replaying dead-lettered messages
 
-Once the cause is fixed, move the messages back to the service queue with the broker's own tools: a [dead letter queue redrive](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-dead-letter-queue-redrive.html) on Amazon SQS, or a [shovel](https://www.rabbitmq.com/docs/shovel) on RabbitMQ. The RabbitMQ transport leaves its attempt count off dead-lettered messages, so a replayed message gets all its attempts again.
+Once the cause is fixed, move the messages back to the service queue with the broker's own tools: a [dead letter queue redrive](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-dead-letter-queue-redrive.html) on Amazon SQS, a [shovel](https://www.rabbitmq.com/docs/shovel) on RabbitMQ, or the [redrive script](/transports/redis#retries-and-dead-letters) on Redis. The RabbitMQ transport leaves its attempt count off dead-lettered messages, so a replayed message gets all its attempts again.
 
 ## Receivers
 

docs/package.json

@@ -28,6 +28,7 @@
     "@node-ts/bus-opentelemetry": "workspace:^",
     "@node-ts/bus-postgres": "workspace:^",
     "@node-ts/bus-rabbitmq": "workspace:^",
+    "@node-ts/bus-redis": "workspace:^",
     "@node-ts/bus-sqs": "workspace:^",
     "@node-ts/bus-sqs-lambda": "workspace:^",
     "@node-ts/bus-test": "workspace:^",

docs/snippets/redis.ts

@@ -0,0 +1,24 @@
+import { Bus } from '@node-ts/bus-core'
+import { RedisTransport, RedisTransportConfiguration } from '@node-ts/bus-redis'
+import { reserveRoomHandler } from './handlers/reserve-room-handler'
+import { messageTypes } from './message-types.generated'
+
+const redisConfiguration: RedisTransportConfiguration = {
+  queueName: 'reservations-service',
+  connection: { url: 'redis://localhost:6379' },
+  // Longer than the slowest handler takes, or a message still being handled is received again
+  visibilityTimeoutMs: 30_000
+}
+const redisTransport = new RedisTransport(redisConfiguration)
+
+const bus = Bus.configure()
+  .withMessageTypes(messageTypes)
+  .withTransport(redisTransport)
+  .withHandler(reserveRoomHandler)
+  // For local development: creates the queue's stream, its consumer group and its subscriptions when the bus
+  // initializes. In production, create them at deploy time with `bus provision` instead.
+  .withAutoProvision()
+  .build()
+
+await bus.initialize()
+await bus.start()

docs/transports.md

@@ -14,6 +14,7 @@ By default the bus uses an in-memory queue, `InMemoryQueue`, which needs nothing
   <Card title="Amazon SQS" tag="@node-ts/bus-sqs" link="/transports/amazon-sqs">AWS' managed queues, with SNS topics for events. `bus provision` creates the topics, queues and subscriptions, or you manage them.</Card>
   <Card title="SQS and Lambda" tag="@node-ts/bus-sqs-lambda" link="/transports/sqs-lambda">Handle SQS messages in AWS Lambda, which reads the queue and passes batches to the bus.</Card>
   <Card title="Postgres" tag="@node-ts/bus-postgres" link="/transports/postgres">Queues in the Postgres database you already have, with no broker to run. Suits one service, or services that share a database.</Card>
+  <Card title="Redis" tag="@node-ts/bus-redis" link="/transports/redis">Queues in Redis Streams, on the Redis or Valkey you may already run. Retries and dead letters are built on top, with no modules.</Card>
   <Card title="Custom transports" link="/transports/custom">Adapt another broker by implementing the Transport interface, and check it with the conformance suite.</Card>
 </FeatureGrid>