feat: add source/redis adapter (Redis Streams)

.github/workflows/source.yml

@@ -61,6 +61,7 @@ jobs:
         module:
           - source/kafka
           - source/jetstream
+          - source/redis
           - source/cloudevents
           - examples/sourcedrive
     runs-on: ${{ matrix.os }}
@@ -108,6 +109,7 @@ jobs:
           - source/statemachine
           - source/kafka
           - source/jetstream
+          - source/redis
           - source/cloudevents
           - examples/sourcedrive
     runs-on: ubuntu-latest

README.md

@@ -61,6 +61,7 @@ stability label.
 | `source`              | Ingress seam: consume streams and drive statecharts; ack on durable transition.  | experimental |
 | `source/kafka`        | Kafka/RedPanda Inlet over franz-go: group consumer, mark-commit-after-process.   | experimental |
 | `source/jetstream`    | NATS JetStream Inlet over nats.go: pull consumer, ack/nak/term, MaxAckPending.    | experimental |
+| `source/redis`        | Redis Streams Inlet over go-redis: consumer group, XACK/pending-claim, DLQ.       | experimental |
 | `source/cloudevents`  | CloudEvents codec with structured and binary content modes.                      | experimental |
 | `source/statemachine` | Bridge: an inbound message drives a transition, ack tied to the durable commit.  | experimental |
 
@@ -81,9 +82,9 @@ snapshots; inspection; and JSON (de)serialization. It is backed by its `analysis
 `evolution`, and `conformance` companion packages. Treat its API as experimental
 until it reaches v1. The `telemetry` interface and its `slog`, `otel`, and
 `datadog` adapters are released. The `sink` egress seam and its destination
-adapters, and the `source` ingress seam with its Kafka and JetStream adapters,
-CloudEvents codec, reliability middleware, and state-machine bridge, are now
-available and documented; the `broker` module is planned.
+adapters, and the `source` ingress seam with its Kafka, JetStream, and Redis
+Streams adapters, CloudEvents codec, reliability middleware, and state-machine
+bridge, are now available and documented; the `broker` module is planned.
 
 ## Roadmap: event-driven seams
 

docs/src/content/docs/source/adapters.md

@@ -45,33 +45,52 @@ deliver-by-start options. A JetStream durable consumer is the grouping analog of
 a Kafka consumer group, but it has no partitions and no assignment callbacks, so
 JetStream does **not** pretend to be a `ConsumerGroups` backend.
 
+## redis (go-redis)
+
+`crucible/source/redis` is a consumer-group reader over go-redis:
+
+```go
+in, _ := redis.New(redis.WithAddr("localhost:6379"), redis.WithGroup("orders-svc"), redis.WithConsumer("worker-1"))
+sub, _ := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}})
+m, _ := sub.Next(ctx)
+_ = sub.Settle(ctx, m, source.Ack())
+```
+
+It reads a Redis Stream with `XREADGROUP`, acks with `XACK`, leaves a naked
+entry in the pending list for a later `XPENDING` plus `XCLAIM` redelivery, and
+routes a terminated entry to a configured dead-letter stream before acking the
+original. Replay is by entry ID through `XRANGE`, and lag comes from
+`XINFO GROUPS` with an `XLEN` fallback. A Redis consumer group is the grouping
+analog of a Kafka consumer group, but it has no partitions and no assignment
+callbacks, so Redis does **not** pretend to be a `ConsumerGroups` backend.
+
 ## Capability table per backend
 
 Capabilities are detected by interface assertion, once, inside the engine. The
 table is honest: an adapter satisfies a capability only when its backend truly
 supports it, and a compile-time `var _ Seekable = ...` assertion in each module
 keeps it accurate.
 
-| Capability | kafka | jetstream | Notes |
-|---|---|---|---|
-| `Seekable` | yes | yes | live `SetOffsets` on Kafka; consumer-recreate on JetStream |
-| `ConsumerGroups` | yes | no | Kafka rebalance hooks; JetStream has no partition assignment |
-| `SharedDurable` | no | yes | the JetStream durable-consumer grouping analog |
-| `PartitionOrdered` | yes | no | Kafka per-partition order |
-| `OrderedDelivery` | no | yes | JetStream `OrderedConsumer`, single-threaded |
-| `Batched` | yes | yes | batched fetch on both |
-| `Transactional` | yes | no | Kafka EOS only; JetStream does not, and does not fake it |
-| `Deduper` | yes | yes | dedup seam (see [reliability](/crucible/source/reliability/)) |
-| `LagReporter` | yes | yes | consumer-lag gauge |
+| Capability | kafka | jetstream | redis | Notes |
+|---|---|---|---|---|
+| `Seekable` | yes | yes | yes | live `SetOffsets` on Kafka; consumer-recreate on JetStream; entry-ID `XRANGE` on Redis |
+| `ConsumerGroups` | yes | no | no | Kafka rebalance hooks; JetStream and Redis have no partition assignment |
+| `SharedDurable` | no | yes | yes | the JetStream durable-consumer and Redis consumer-group grouping analogs |
+| `PartitionOrdered` | yes | no | no | Kafka per-partition order |
+| `OrderedDelivery` | no | yes | no | JetStream `OrderedConsumer`, single-threaded |
+| `Batched` | yes | yes | no | batched fetch on Kafka and JetStream |
+| `Transactional` | yes | no | no | Kafka EOS only; JetStream and Redis do not, and do not fake it |
+| `Deduper` | yes | yes | no | dedup seam (see [reliability](/crucible/source/reliability/)) |
+| `LagReporter` | yes | yes | yes | consumer-lag gauge; Redis reports group lag from `XINFO GROUPS` |
 
 The divergences are documented, never papered over. A `Nak(delay)` is a real
 delayed redelivery on JetStream but is best-effort on Kafka (pause plus reseek),
 and that is called out where it matters.
 
 ## Roadmap and bring your own
 
-`source/redis` (Redis Streams) and `source/cdc` (Debezium/OpenCDC
-change-data-capture) are on the roadmap. The catalog is a convenience: any type
+`source/cdc` (Debezium/OpenCDC change-data-capture) is on the roadmap. The
+catalog is a convenience: any type
 that satisfies the `Inlet` interface is an inlet, and the
 [`memsource`](/crucible/source/reliability/#testing-the-loop) in-memory adapter
 is itself an `Inlet`, so you can drive the whole engine in a unit test with no

magefiles/magefile.go

@@ -56,12 +56,12 @@ var sinkDestinations = []string{
 
 // sourceDestinations are the standalone source modules kept out of the workspace
 // (see modules above): the SDK-backed sources (source/kafka via franz-go,
-// source/jetstream via nats.go, source/cloudevents via the CloudEvents SDK) plus
-// the flagship examples/sourcedrive. Each builds via its own go.mod and replace
-// directives, so the SourceDestinations and Integration targets run with
-// GOWORK=off.
+// source/jetstream via nats.go, source/redis via go-redis, source/cloudevents
+// via the CloudEvents SDK) plus the flagship examples/sourcedrive. Each builds
+// via its own go.mod and replace directives, so the SourceDestinations and
+// Integration targets run with GOWORK=off.
 var sourceDestinations = []string{
-	"source/kafka", "source/jetstream", "source/cloudevents", "examples/sourcedrive",
+	"source/kafka", "source/jetstream", "source/redis", "source/cloudevents", "examples/sourcedrive",
 }
 
 // Pinned tool versions — keep in sync with .github/workflows/ci.yml.

source/redis/README.md

@@ -0,0 +1,79 @@
+# source/redis
+
+A [`crucible/source`](../) ingress adapter that consumes a Redis Stream through
+a consumer group. Runtime dependency:
+[`github.com/redis/go-redis/v9`](https://github.com/redis/go-redis).
+
+```go
+in, err := redis.New(
+    redis.WithAddr("localhost:6379"),
+    redis.WithGroup("orders-svc"),
+    redis.WithConsumer("worker-1"),
+    redis.WithDLQStream("orders.dlq"),
+    redis.WithBlock(5*time.Second),
+    redis.WithCount(64),
+)
+sub, _ := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}})
+
+m, _ := sub.Next(ctx)              // or drive sub with a source.Hopper
+_ = sub.Settle(ctx, m, source.Ack())
+```
+
+Construct an `Inlet` with `New` and functional options (`WithAddr`/`WithClient`,
+`WithGroup`, `WithConsumer`, `WithDLQStream`, `WithBlock`, `WithCount`,
+`WithMinIdle`). The returned `Subscription` reads entries with `Next` (via
+`XREADGROUP`) and applies a handler `source.Result` with `Settle`. Backpressure
+comes from the `Count` batch size plus the block window. Reach the underlying
+`*redis.Client` through `Inlet.As`, or the per-entry `redis.XMessage` through
+`Message.As`; no vendor type appears in the adapter's own signatures.
+
+A stream entry maps onto `source.Message` as follows: the `value` field becomes
+the raw `Value`, the `crucible-key` field (when set) becomes the routing `Key`
+(otherwise the stream name), every field is exposed as a `Header`, `Subject` is
+the stream, and `Cursor` is the entry ID.
+
+## Settle vocabulary
+
+A consumer group reads with `XREADGROUP`; every delivered entry stays in the
+group's Pending Entries List (PEL) until it is settled.
+
+- **Ack** calls `XACK`, removing the entry from the PEL.
+- **Nak** leaves the entry in the PEL. A consumer reclaims and redelivers it by
+  scanning the backlog with `XPENDING` + `XCLAIM` once it has idled past the
+  minimum; call `NakRedeliver` (on a timer or between read cycles) to drive that
+  pass. The cadence is a deployment choice, so the engine does not force it.
+- **Term** (and **Reject**) append the entry's fields plus dead-letter metadata
+  (`crucible-dlq-original-id`, `crucible-dlq-stream`, `crucible-dlq-class`,
+  `crucible-dlq-error`) to the configured `WithDLQStream`, then `XACK` the
+  original. With no dead-letter stream configured, a terminated entry is acked
+  and dropped.
+- **InProgress** is a no-op: Redis has no per-message deadline to extend.
+- **Manual** is a no-op; the handler settled the entry itself through
+  `Message.As` and the client.
+
+## Capabilities
+
+The subscription honestly advertises the Redis-shaped capabilities by type
+assertion: `source.SharedDurable` (the consumer group is the competing-consumer
+analog), `source.Seekable` (replay by entry ID via `XRANGE`, with `SeekToTime`
+translating a timestamp into an entry ID), and `source.LagReporter` (group lag
+from `XINFO GROUPS`, falling back to `XLEN`).
+
+## Redis divergences from the source contract
+
+- **No `ConsumerGroups`.** A Redis consumer group has no partitions and no
+  assignment lifecycle, so the adapter does not implement
+  `source.ConsumerGroups`. The group load-balances across processes instead,
+  surfaced as `source.SharedDurable`. `PartitionKey` is always `""`, so the
+  Hopper shards by `Key`.
+- **No `Transactional`.** Redis Streams offer no consume-side transaction, so
+  the adapter does not implement `source.Transactional`; the capability is
+  absent rather than faked.
+
+## Stability
+
+Experimental (pre-v1). The API may change until the suite locks v1.0.0.
+
+## License
+
+Apache-2.0. See [LICENSE](../../LICENSE) and [NOTICE](../../NOTICE).

source/redis/example_test.go

@@ -0,0 +1,87 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package redis_test
+
+import (
+	"context"
+	"fmt"
+
+	goredis "github.com/redis/go-redis/v9"
+
+	"github.com/stablekernel/crucible/source"
+	credis "github.com/stablekernel/crucible/source/redis"
+)
+
+// exampleClient is a stand-in Client seam that yields one stream entry, so the
+// example runs without a live Redis server. It implements only the methods the
+// consume-and-ack path touches; a real program passes redis.WithAddr or
+// redis.WithClient(realClient) instead.
+type exampleClient struct {
+	credis.Client
+	delivered bool
+}
+
+func (c *exampleClient) XGroupCreateMkStream(ctx context.Context, _, _, _ string) *goredis.StatusCmd {
+	return goredis.NewStatusCmd(ctx)
+}
+
+func (c *exampleClient) XReadGroup(ctx context.Context, a *goredis.XReadGroupArgs) *goredis.XStreamSliceCmd {
+	cmd := goredis.NewXStreamSliceCmd(ctx)
+	if c.delivered {
+		cmd.SetErr(goredis.Nil) // no further entries
+		return cmd
+	}
+	c.delivered = true
+	cmd.SetVal([]goredis.XStream{{
+		Stream: a.Streams[0],
+		Messages: []goredis.XMessage{{
+			ID:     "1526919030474-0",
+			Values: map[string]any{"value": "placed", "crucible-key": "A-1"},
+		}},
+	}})
+	return cmd
+}
+
+func (c *exampleClient) XAck(ctx context.Context, _, _ string, _ ...string) *goredis.IntCmd {
+	cmd := goredis.NewIntCmd(ctx)
+	cmd.SetVal(1)
+	return cmd
+}
+
+// Example shows the consume-and-settle loop: open a consumer-group
+// subscription, take the next entry off the stream, and ack it after handling.
+// In a real program the seam is a live client supplied via WithAddr or
+// WithClient.
+func Example() {
+	in, err := credis.New(
+		credis.WithClient(&exampleClient{}),
+		credis.WithGroup("orders-svc"),
+		credis.WithConsumer("worker-1"),
+	)
+	if err != nil {
+		panic(err)
+	}
+	defer func() { _ = in.Close() }()
+
+	sub, err := in.Subscribe(context.Background(), source.SubscribeConfig{Topics: []string{"orders"}})
+	if err != nil {
+		panic(err)
+	}
+	defer func() { _ = sub.Close() }()
+
+	m, err := sub.Next(context.Background())
+	if err != nil {
+		panic(err)
+	}
+	if err := sub.Settle(context.Background(), m, source.Ack()); err != nil {
+		panic(err)
+	}
+
+	fmt.Println(m.Subject())
+	fmt.Println(string(m.Key()))
+	fmt.Println(string(m.Value()))
+	// Output:
+	// orders
+	// A-1
+	// placed
+}

source/redis/go.mod

@@ -0,0 +1,70 @@
+module github.com/stablekernel/crucible/source/redis
+
+go 1.25.11
+
+replace github.com/stablekernel/crucible/source => ../
+
+replace github.com/stablekernel/crucible/telemetry => ../../telemetry
+
+require (
+	github.com/redis/go-redis/v9 v9.20.0
+	github.com/stablekernel/crucible/source v0.0.0-00010101000000-000000000000
+	github.com/testcontainers/testcontainers-go v0.42.0
+	github.com/testcontainers/testcontainers-go/modules/redis v0.42.0
+)
+
+require (
+	dario.cat/mergo v1.0.2 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/containerd/errdefs v1.0.0 // indirect
+	github.com/containerd/errdefs/pkg v0.3.0 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
+	github.com/cpuguy83/dockercfg v0.3.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/go-connections v0.6.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/ebitengine/purego v0.10.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/klauspost/compress v1.18.5 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/magiconair/properties v1.8.10 // indirect
+	github.com/mdelapenya/tlscert v0.2.0 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/go-archive v0.2.0 // indirect
+	github.com/moby/moby/api v1.54.1 // indirect
+	github.com/moby/moby/client v0.4.0 // indirect
+	github.com/moby/patternmatcher v0.6.1 // indirect
+	github.com/moby/sys/sequential v0.6.0 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
+	github.com/moby/sys/userns v0.1.0 // indirect
+	github.com/moby/term v0.5.2 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
+	github.com/shirou/gopsutil/v4 v4.26.3 // indirect
+	github.com/sirupsen/logrus v1.9.4 // indirect
+	github.com/stablekernel/crucible/telemetry v0.0.0 // indirect
+	github.com/stretchr/testify v1.11.1 // indirect
+	github.com/tklauser/go-sysconf v0.3.16 // indirect
+	github.com/tklauser/numcpus v0.11.0 // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
+	go.opentelemetry.io/otel v1.41.0 // indirect
+	go.opentelemetry.io/otel/metric v1.41.0 // indirect
+	go.opentelemetry.io/otel/trace v1.41.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+	golang.org/x/crypto v0.48.0 // indirect
+	golang.org/x/sys v0.42.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)