redistransport

README

Mercure Redis Transport

A Redis transport for the Mercure hub, packaged as a Caddy module.

The open-source Mercure hub ships single-process transports (local, bolt), so a horizontally scaled hub cannot deliver an update published on one instance to subscribers connected to another. This module fixes that. Redis Streams hold the event history, Redis Pub/Sub fans out to the other instances, and a UUID → stream-ID index keeps Last-Event-ID replay O(1).

Requires Redis 5.0 or newer, for Redis Streams. Supported Go, Caddy, and Mercure versions are the ones declared in go.mod. The configuration surface is stable as of 1.0.0 and follows semantic versioning, so breaking changes wait for a major release.

Install

Build Caddy with xcaddy:

xcaddy build \
    --with github.com/dunglas/mercure/caddy \
    --with github.com/desk-ly/mercure-transport-redis

Configure

localhost {
    route {
        mercure {
            transport redis {
                url {$REDIS_URL}
                stream mercure
                size 10000
                history_limit 1000
            }

            publisher_jwt {$MERCURE_PUBLISHER_JWT_KEY}
            subscriber_jwt {$MERCURE_SUBSCRIBER_JWT_KEY}
        }

        respond 404
    }
}

Parameter	Default	Description
url	(required)	Redis URL, e.g. redis://host:6379/0 or rediss:// for TLS. Parsed by go-redis ParseURL, so it also carries credentials and DB index.
stream	mercure	Stream key. Also derives the Pub/Sub channel (<stream>:pubsub) and index hash (<stream>:index).
size	0	Max stream entries, enforced by a background XTRIM every 10s. 0 means no trimming, so the stream grows without bound. Set this.
history_limit	1000	Max entries read per SSE history replay. 0 disables replay, so subscribers fast-forward to live.

Size the stream to cover your longest expected client disconnect at peak event rate. There is no TTL and no eviction fallback, so size is the only bound on Redis memory. Last-Event-ID is client-controlled, so history_limit caps what a single reconnect can cost. Set it to 0 if your clients re-sync through your own API on connect.

Security

The transport trusts its Redis. Anything that can write to <stream>:pubsub can inject updates without a publisher JWT, and anything that can read the stream sees every update in plain text. Subscriber authorization still applies, so an injected update only reaches subscribers whose topic selectors already match it, but its topics and content are attacker-controlled.

Run Redis with AUTH/ACLs and TLS, and do not share the instance with untrusted workloads.

Limitations

• All topics share one stream, and cross-instance replay filters locally rather than at the stream level.
• history_limit bounds the client-facing path only. Internal gap recovery after a Pub/Sub outage may dispatch up to size events at once. The anchor is server-controlled, so this is not client-triggerable.
• The transport cannot signal Mercure that Redis is unreachable. While Pub/Sub is down, local delivery continues but instances diverge until reconnect and replay.
• Pub/Sub delivery is best-effort. Consistency comes from the stream plus replay, not the channel.

Development

make test runs the suite in Docker, so no local Go toolchain is required. Without Docker, go test -race -count=1 ./... does the same. Tests run against miniredis, so no Redis instance is needed.

We built this at desk.ly and share it under AGPL-3.0 so that anyone running the hub has the full source. It is maintained around our own needs rather than as a community project, so we don't take external contributions or offer support. You are free to fork it and take it wherever you need.

License

Copyright © 2026 desk.ly GmbH. Licensed under AGPL-3.0-or-later.

This module links against the Mercure hub, which is AGPL-3.0. A Caddy binary built with it is therefore a combined AGPL-3.0 work. If you run it as a network service, AGPL-3.0 §13 requires you to offer its complete corresponding source to your users.

Documentation

Overview

Caddy module wrapper that registers the Redis transport as http.handlers.mercure.redis. Enables `transport redis { url ...; stream ...; size ... }` in Caddyfile.

Index

• type Redis
  • func (Redis) CaddyModule() caddy.ModuleInfo
  • func (r *Redis) Cleanup() error
  • func (r *Redis) GetTransport() mercure.Transport
  • func (r *Redis) Provision(ctx caddy.Context) error
  • func (r *Redis) UnmarshalCaddyfile(d *caddyfile.Dispenser) error
• type RedisTransport
  • func NewRedisTransport(subscriberList *mercure.SubscriberList, logger *slog.Logger, url string, ...) (*RedisTransport, error)
  • func (t *RedisTransport) AddSubscriber(ctx context.Context, subscriber *mercure.LocalSubscriber) error
  • func (t *RedisTransport) Close(_ context.Context) (err error)
  • func (t *RedisTransport) Dispatch(ctx context.Context, update *mercure.Update) error
  • func (t *RedisTransport) GetSubscribers(_ context.Context) (string, []*mercure.Subscriber, error)
  • func (t *RedisTransport) RemoveSubscriber(_ context.Context, subscriber *mercure.LocalSubscriber) error

Types

type Redis

type Redis struct {
	URL          string `json:"url,omitempty"`
	Stream       string `json:"stream,omitempty"`
	Size         int64  `json:"size,omitempty"`
	HistoryLimit *int64 `json:"history_limit,omitempty"` // nil = default, 0 = replay disabled
	// contains filtered or unexported fields
}

func (Redis) CaddyModule

func (Redis) CaddyModule() caddy.ModuleInfo

func (*Redis) Cleanup

func (r *Redis) Cleanup() error

func (*Redis) GetTransport

func (r *Redis) GetTransport() mercure.Transport

func (*Redis) Provision

func (r *Redis) Provision(ctx caddy.Context) error

func (*Redis) UnmarshalCaddyfile

func (r *Redis) UnmarshalCaddyfile(d *caddyfile.Dispenser) error

type RedisTransport

type RedisTransport struct {
	// contains filtered or unexported fields
}

func NewRedisTransport

func NewRedisTransport(
	subscriberList *mercure.SubscriberList,
	logger *slog.Logger,
	url string,
	stream string,
	size int64,
	historyLimit int64,
) (*RedisTransport, error)

func (*RedisTransport) AddSubscriber

func (t *RedisTransport) AddSubscriber(ctx context.Context, subscriber *mercure.LocalSubscriber) error

func (*RedisTransport) Close

func (t *RedisTransport) Close(_ context.Context) (err error)

func (*RedisTransport) Dispatch

func (t *RedisTransport) Dispatch(ctx context.Context, update *mercure.Update) error

func (*RedisTransport) GetSubscribers

func (t *RedisTransport) GetSubscribers(_ context.Context) (string, []*mercure.Subscriber, error)

func (*RedisTransport) RemoveSubscriber

func (t *RedisTransport) RemoveSubscriber(_ context.Context, subscriber *mercure.LocalSubscriber) error