feat: add cache.redis module and step.cache_get/set/delete pipeline steps

module/cache_redis.go

@@ -0,0 +1,161 @@
+package module
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/CrisisTextLine/modular"
+	"github.com/redis/go-redis/v9"
+)
+
+// CacheModule defines the interface for cache operations used by pipeline steps.
+type CacheModule interface {
+	Get(ctx context.Context, key string) (string, error)
+	Set(ctx context.Context, key, value string, ttl time.Duration) error
+	Delete(ctx context.Context, key string) error
+}
+
+// RedisClient is the subset of go-redis client methods used by RedisCache.
+// Keeping it as an interface enables mocking in tests.
+type RedisClient interface {
+	Ping(ctx context.Context) *redis.StatusCmd
+	Get(ctx context.Context, key string) *redis.StringCmd
+	Set(ctx context.Context, key string, value any, expiration time.Duration) *redis.StatusCmd
+	Del(ctx context.Context, keys ...string) *redis.IntCmd
+	Close() error
+}
+
+// RedisCacheConfig holds configuration for the cache.redis module.
+type RedisCacheConfig struct {
+	Address    string
+	Password   string
+	DB         int
+	Prefix     string
+	DefaultTTL time.Duration
+}
+
+// RedisCache is a module that connects to a Redis instance and exposes
+// Get/Set/Delete operations for use by pipeline steps.
+type RedisCache struct {
+	name   string
+	cfg    RedisCacheConfig
+	client RedisClient
+	logger modular.Logger
+}
+
+// NewRedisCache creates a new RedisCache module with the given name and config.
+func NewRedisCache(name string, cfg RedisCacheConfig) *RedisCache {
+	return &RedisCache{
+		name:   name,
+		cfg:    cfg,
+		logger: &noopLogger{},
+	}
+}
+
+// NewRedisCacheWithClient creates a RedisCache backed by a pre-built client.
+// This is intended for testing only.
+func NewRedisCacheWithClient(name string, cfg RedisCacheConfig, client RedisClient) *RedisCache {
+	return &RedisCache{
+		name:   name,
+		cfg:    cfg,
+		client: client,
+		logger: &noopLogger{},
+	}
+}
+
+func (r *RedisCache) Name() string { return r.name }
+
+func (r *RedisCache) Init(app modular.Application) error {
+	r.logger = app.Logger()
+	return nil
+}
+
+// Start connects to Redis and verifies the connection with PING.
+func (r *RedisCache) Start(ctx context.Context) error {
+	if r.client != nil {
+		// Already set (e.g. in tests)
+		return nil
+	}
+
+	opts := &redis.Options{
+		Addr: r.cfg.Address,
+		DB:   r.cfg.DB,
+	}
+	if r.cfg.Password != "" {
+		opts.Password = r.cfg.Password
+	}
+
+	r.client = redis.NewClient(opts)
+
+	if err := r.client.Ping(ctx).Err(); err != nil {
+		_ = r.client.Close()
+		r.client = nil
+		return fmt.Errorf("cache.redis %q: ping failed: %w", r.name, err)
+	}
+
+	r.logger.Info("Redis cache started", "name", r.name, "address", r.cfg.Address)
+	return nil
+}
+
+// Stop closes the Redis connection.
+func (r *RedisCache) Stop(_ context.Context) error {
+	if r.client != nil {
+		r.logger.Info("Redis cache stopped", "name", r.name)
+		return r.client.Close()
+	}
+	return nil
+}
+
+// Get retrieves a value from Redis by key (with prefix applied).
+// Returns redis.Nil wrapped in an error when the key does not exist.
+func (r *RedisCache) Get(ctx context.Context, key string) (string, error) {
+	if r.client == nil {
+		return "", fmt.Errorf("cache.redis %q: not started", r.name)
+	}
+	val, err := r.client.Get(ctx, r.prefixed(key)).Result()
+	if err != nil {
+		return "", err
+	}
+	return val, nil
+}
+
+// Set stores a value in Redis with optional TTL.  A zero duration uses the
+// module-level default; if the default is also zero the key never expires.
+func (r *RedisCache) Set(ctx context.Context, key, value string, ttl time.Duration) error {
+	if r.client == nil {
+		return fmt.Errorf("cache.redis %q: not started", r.name)
+	}
+	if ttl == 0 {
+		ttl = r.cfg.DefaultTTL
+	}
+	return r.client.Set(ctx, r.prefixed(key), value, ttl).Err()
+}
+
+// Delete removes a key from Redis (with prefix applied).
+func (r *RedisCache) Delete(ctx context.Context, key string) error {
+	if r.client == nil {
+		return fmt.Errorf("cache.redis %q: not started", r.name)
+	}
+	return r.client.Del(ctx, r.prefixed(key)).Err()
+}
+
+func (r *RedisCache) prefixed(key string) string {
+	return r.cfg.Prefix + key
+}
+
+func (r *RedisCache) ProvidesServices() []modular.ServiceProvider {
+	return []modular.ServiceProvider{
+		{Name: r.name, Description: "Redis cache connection", Instance: r},
+	}
+}
+
+func (r *RedisCache) RequiresServices() []modular.ServiceDependency {
+	return nil
+}
+
+// ExpandEnvString resolves ${VAR} and $VAR environment variable references.
+func ExpandEnvString(s string) string {
+	return os.ExpandEnv(s)
+}

module/cache_redis_test.go

@@ -0,0 +1,173 @@
+package module
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+// newTestRedisCache creates a RedisCache backed by a miniredis server.
+func newTestRedisCache(t *testing.T) (*RedisCache, *miniredis.Miniredis) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { client.Close() })
+
+	cfg := RedisCacheConfig{
+		Address:    mr.Addr(),
+		Prefix:     "test:",
+		DefaultTTL: time.Hour,
+	}
+	cache := NewRedisCacheWithClient("cache", cfg, client)
+	return cache, mr
+}
+
+func TestRedisCacheGetSetDelete(t *testing.T) {
+	ctx := context.Background()
+	cache, _ := newTestRedisCache(t)
+
+	// Set a value
+	if err := cache.Set(ctx, "mykey", "myvalue", 0); err != nil {
+		t.Fatalf("Set failed: %v", err)
+	}
+
+	// Get it back
+	val, err := cache.Get(ctx, "mykey")
+	if err != nil {
+		t.Fatalf("Get failed: %v", err)
+	}
+	if val != "myvalue" {
+		t.Errorf("expected %q, got %q", "myvalue", val)
+	}
+
+	// Delete it
+	if err := cache.Delete(ctx, "mykey"); err != nil {
+		t.Fatalf("Delete failed: %v", err)
+	}
+
+	// Get after delete should return redis.Nil
+	_, err = cache.Get(ctx, "mykey")
+	if err == nil {
+		t.Fatal("expected error after delete, got nil")
+	}
+}
+
+func TestRedisCacheKeyPrefix(t *testing.T) {
+	ctx := context.Background()
+	cache, mr := newTestRedisCache(t)
+
+	if err := cache.Set(ctx, "hello", "world", 0); err != nil {
+		t.Fatalf("Set failed: %v", err)
+	}
+
+	// Verify prefix is stored in miniredis
+	keys := mr.Keys()
+	found := false
+	for _, k := range keys {
+		if k == "test:hello" {
+			found = true
+			break
+		}
+	}
+	if !found {
+		t.Errorf("expected key %q in redis, got keys: %v", "test:hello", keys)
+	}
+}
+
+func TestRedisCacheDefaultTTL(t *testing.T) {
+	ctx := context.Background()
+	cache, mr := newTestRedisCache(t)
+
+	// Set with TTL=0 should use DefaultTTL (1 hour)
+	if err := cache.Set(ctx, "ttlkey", "ttlval", 0); err != nil {
+		t.Fatalf("Set failed: %v", err)
+	}
+
+	ttl := mr.TTL("test:ttlkey")
+	if ttl <= 0 {
+		t.Errorf("expected positive TTL, got %v", ttl)
+	}
+}
+
+func TestRedisCacheExplicitTTL(t *testing.T) {
+	ctx := context.Background()
+	cache, mr := newTestRedisCache(t)
+
+	// Set with explicit TTL=30m
+	if err := cache.Set(ctx, "short", "val", 30*time.Minute); err != nil {
+		t.Fatalf("Set failed: %v", err)
+	}
+
+	ttl := mr.TTL("test:short")
+	// miniredis reports TTL in seconds-level precision; just verify it's set
+	if ttl <= 0 {
+		t.Errorf("expected positive TTL, got %v", ttl)
+	}
+	if ttl > time.Hour {
+		t.Errorf("expected TTL <= 1h, got %v", ttl)
+	}
+}
+
+func TestRedisCacheMiss(t *testing.T) {
+	ctx := context.Background()
+	cache, _ := newTestRedisCache(t)
+
+	_, err := cache.Get(ctx, "nonexistent")
+	if err == nil {
+		t.Fatal("expected error for missing key")
+	}
+}
+
+func TestRedisCacheNotStarted(t *testing.T) {
+	ctx := context.Background()
+	cfg := RedisCacheConfig{Address: "localhost:6379", Prefix: "wf:"}
+	cache := NewRedisCache("cache", cfg)
+
+	if _, err := cache.Get(ctx, "k"); err == nil {
+		t.Error("expected error from Get when not started")
+	}
+	if err := cache.Set(ctx, "k", "v", 0); err == nil {
+		t.Error("expected error from Set when not started")
+	}
+	if err := cache.Delete(ctx, "k"); err == nil {
+		t.Error("expected error from Delete when not started")
+	}
+}
+
+func TestRedisCacheInit(t *testing.T) {
+	cfg := RedisCacheConfig{Address: "localhost:6379", Prefix: "wf:"}
+	cache := NewRedisCache("cache", cfg)
+	app := NewMockApplication()
+
+	if err := cache.Init(app); err != nil {
+		t.Fatalf("Init failed: %v", err)
+	}
+}
+
+func TestRedisCacheStop(t *testing.T) {
+	ctx := context.Background()
+	cache, _ := newTestRedisCache(t)
+
+	if err := cache.Stop(ctx); err != nil {
+		t.Fatalf("Stop failed: %v", err)
+	}
+	// Stop when already nil is a no-op
+	cache2 := NewRedisCache("cache2", RedisCacheConfig{})
+	if err := cache2.Stop(ctx); err != nil {
+		t.Fatalf("Stop on uninitialised cache failed: %v", err)
+	}
+}
+
+func TestRedisCacheProvidesServices(t *testing.T) {
+	cache := NewRedisCache("mycache", RedisCacheConfig{})
+	svcs := cache.ProvidesServices()
+	if len(svcs) != 1 {
+		t.Fatalf("expected 1 service, got %d", len(svcs))
+	}
+	if svcs[0].Name != "mycache" {
+		t.Errorf("expected service name %q, got %q", "mycache", svcs[0].Name)
+	}
+}