feat(store/redis): namespace keys per tenant with WithKeyPrefix

docs/content/docs/guides/forge-extension.mdx

@@ -47,6 +47,7 @@ Forge will call `ext.Start(ctx)` on application start and `ext.Stop(ctx)` on shu
 | `WithDisableMigrate()` | Skip auto-migration at startup | `false` |
 | `WithGroveDatabase(name)` | Resolve a grove.DB from DI by name | `""` |
 | `WithGroveKV(name)` | Resolve a grove KV store from DI by name | `""` |
+| `WithKVKeyPrefix(prefix)` | Namespace the Redis store's keys per tenant (grove KV path only) | `""` |
 
 ## Auto-migration
 
@@ -114,6 +115,19 @@ ext := extension.New(
 )
 ```
 
+When several dispatch instances share one Redis, give each its own key
+prefix. Every key, stream and pub/sub channel the store touches goes under
+it (`ws_acme:dispatch:job:<id>`), so instances never see each other's
+queues, cron locks or leadership. Leave it empty for a single instance and
+the keys stay `dispatch:*` as before:
+
+```go
+ext := extension.New(
+    extension.WithGroveKV("dispatch-kv"),
+    extension.WithKVKeyPrefix("ws_acme:"),
+)
+```
+
 ### Store resolution order
 
 The extension resolves its store in this order:
@@ -134,6 +148,7 @@ extensions:
     base_path: /api/dispatch
     grove_database: jobs
     grove_kv: dispatch-kv
+    key_prefix: "ws_acme:"
     concurrency: 20
     queues:
       - default
@@ -161,6 +176,7 @@ dispatch:
 | `BasePath` | `base_path` | `string` | `"/api/dispatch"` | URL prefix for all dispatch HTTP routes |
 | `GroveDatabase` | `grove_database` | `string` | `""` | Name of the grove.DB to resolve from DI; empty uses the default DB |
 | `GroveKV` | `grove_kv` | `string` | `""` | Name of the grove KV store to resolve from DI; empty uses the default KV |
+| `KeyPrefix` | `key_prefix` | `string` | `""` | Tenant prefix for every Redis key the grove KV store writes; empty keeps `dispatch:*` |
 | `Concurrency` | `concurrency` | `int` | `10` | Max concurrent jobs |
 | `Queues` | `queues` | `[]string` | `["default"]` | Queues to poll |
 | `DisableRoutes` | `disable_routes` | `bool` | `false` | Skip HTTP route registration |

docs/content/docs/stores/redis.mdx

@@ -27,6 +27,22 @@ if err := s.Ping(ctx); err != nil {
 d, err := dispatch.New(dispatch.WithStore(s))
 ```
 
+## Sharing one Redis between instances
+
+All keys sit under `dispatch:`. To run more than one dispatch instance
+against the same Redis, give each a key prefix and the store puts every
+key, event stream and wake channel under it:
+
+```go
+s := redis.New(kvStore, redis.WithKeyPrefix("ws_acme:"))
+// jobs land at ws_acme:dispatch:job:<id>, the queue at ws_acme:dispatch:queue:<name>
+```
+
+An instance with a prefix never dequeues another instance's jobs, never
+sees its cron names as duplicates, and takes leadership independently. An
+empty prefix keeps the historical unprefixed keys, so existing deployments
+carry on unchanged.
+
 ## Internals
 
 | Aspect | Detail |

extension/config.go

@@ -50,6 +50,12 @@ type Config struct {
 	// (unnamed) kv.Store is used.
 	GroveKV string `json:"grove_kv" mapstructure:"grove_kv" yaml:"grove_kv"`
 
+	// KeyPrefix namespaces every key the Redis-backed store writes, so
+	// several dispatch instances can share one Redis without seeing each
+	// other's jobs, cron locks or leadership. Only the grove KV path reads
+	// it. Empty keeps the historical "dispatch:" keys.
+	KeyPrefix string `json:"key_prefix" mapstructure:"key_prefix" yaml:"key_prefix"`
+
 	// Artifacts configures the artifact plane.
 	Artifacts ArtifactConfig `json:"artifacts" mapstructure:"artifacts" yaml:"artifacts"`
 

extension/extension.go

@@ -165,7 +165,7 @@ func (e *Extension) init(fapp forge.App) error {
 		if err != nil {
 			return fmt.Errorf("dispatch: %w", err)
 		}
-		e.dispatchOpts = append(e.dispatchOpts, dispatch.WithStore(redisstore.New(kvStore)))
+		e.dispatchOpts = append(e.dispatchOpts, dispatch.WithStore(e.buildStoreFromGroveKV(kvStore)))
 	} else if db, err := vessel.Inject[*grove.DB](fapp.Container()); err == nil {
 		// Auto-discover default grove.DB from container (matches authsome/cortex pattern).
 		s, err := e.buildStoreFromGroveDB(db)
@@ -498,6 +498,7 @@ func (e *Extension) loadConfiguration() error {
 		forge.F("base_path", e.config.BasePath),
 		forge.F("grove_database", e.config.GroveDatabase),
 		forge.F("grove_kv", e.config.GroveKV),
+		forge.F("key_prefix", e.config.KeyPrefix),
 	)
 
 	return nil
@@ -624,6 +625,9 @@ func (e *Extension) mergeConfigurations(yamlConfig, programmaticConfig Config) C
 	if yamlConfig.GroveKV == "" && programmaticConfig.GroveKV != "" {
 		yamlConfig.GroveKV = programmaticConfig.GroveKV
 	}
+	if yamlConfig.KeyPrefix == "" && programmaticConfig.KeyPrefix != "" {
+		yamlConfig.KeyPrefix = programmaticConfig.KeyPrefix
+	}
 	if yamlConfig.DWPBasePath == "" && programmaticConfig.DWPBasePath != "" {
 		yamlConfig.DWPBasePath = programmaticConfig.DWPBasePath
 	}
@@ -665,6 +669,12 @@ func (e *Extension) buildStoreFromGroveDB(db *grove.DB) (dispatch.Storer, error)
 	}
 }
 
+// buildStoreFromGroveKV wraps a grove KV store in the Redis backend,
+// namespaced by the configured key prefix.
+func (e *Extension) buildStoreFromGroveKV(kvStore *kv.Store) dispatch.Storer {
+	return redisstore.New(kvStore, redisstore.WithKeyPrefix(e.config.KeyPrefix))
+}
+
 // resolveGroveKV resolves a *kv.Store from the DI container.
 // If GroveKV is set, it looks up the named store; otherwise it uses the default.
 func (e *Extension) resolveGroveKV(fapp forge.App) (*kv.Store, error) {

extension/kvprefix_internal_test.go

@@ -0,0 +1,66 @@
+package extension
+
+import (
+	"testing"
+
+	"github.com/xraph/grove/kv"
+	"github.com/xraph/grove/kv/drivers/redisdriver"
+
+	redisstore "github.com/xraph/dispatch/store/redis"
+)
+
+// unopenedKV returns a Redis-driver KV store that never dials. Building the
+// dispatch store needs only the driver type, so the tests below can pin the
+// wiring without a Redis on the box.
+func unopenedKV(t *testing.T) *kv.Store {
+	t.Helper()
+	s, err := kv.Open(redisdriver.New())
+	if err != nil {
+		t.Fatalf("open kv: %v", err)
+	}
+	return s
+}
+
+func TestBuildStoreFromGroveKV_appliesKeyPrefix(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+	}{
+		{name: "no prefix keeps the historical keys", prefix: ""},
+		{name: "tenant prefix reaches the redis store", prefix: "ws_acme:"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := New(WithKVKeyPrefix(tt.prefix))
+			s, ok := e.buildStoreFromGroveKV(unopenedKV(t)).(*redisstore.Store)
+			if !ok {
+				t.Fatal("grove KV must build a redis store")
+			}
+			if got := s.KeyPrefix(); got != tt.prefix {
+				t.Fatalf("store key prefix = %q, want %q", got, tt.prefix)
+			}
+		})
+	}
+}
+
+func TestMergeConfigurations_keyPrefixYAMLWins(t *testing.T) {
+	e := New()
+	tests := []struct {
+		name         string
+		yaml         string
+		programmatic string
+		want         string
+	}{
+		{name: "yaml set, programmatic empty", yaml: "ws_yaml:", programmatic: "", want: "ws_yaml:"},
+		{name: "yaml empty, programmatic set", yaml: "", programmatic: "ws_go:", want: "ws_go:"},
+		{name: "both set, yaml wins", yaml: "ws_yaml:", programmatic: "ws_go:", want: "ws_yaml:"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := e.mergeConfigurations(Config{KeyPrefix: tt.yaml}, Config{KeyPrefix: tt.programmatic})
+			if got.KeyPrefix != tt.want {
+				t.Fatalf("merged KeyPrefix = %q, want %q", got.KeyPrefix, tt.want)
+			}
+		})
+	}
+}

extension/options.go

@@ -205,6 +205,15 @@ func WithGroveKV(name string) ExtOption {
 	}
 }
 
+// WithKVKeyPrefix namespaces the Redis-backed store's keys (see
+// Config.KeyPrefix). Pair it with WithGroveKV; the grove database path
+// ignores it.
+func WithKVKeyPrefix(prefix string) ExtOption {
+	return func(e *Extension) {
+		e.config.KeyPrefix = prefix
+	}
+}
+
 // WithDWP enables the Dispatch Wire Protocol (DWP) for real-time
 // client communication over WebSocket, SSE, and HTTP RPC.
 // Options configure authentication, codec, and server behaviour.