Redis Cache Adapter — adapters/redis¶

See also: adapters/redis on pkg.go.dev · Ports — Protocol-Agnostic Wiring · Metrics Observer · Error Handling

Runnable demo: examples/redis-cache — CachePattern declaration, write-through SetAdapter, read-through GetAdapter, Seed warm restart, and a stats.CacheObserver — all against an in-memory fake, no Redis server needed.

adapters/redis makes the cache a typed, codec-validated, declared boundary. Values enter and leave through the port's codex.Codec[T] — every read decodes AND validates, every write validates AND encodes. Keys, TTL, and wire format are declared once on the port; every hit, miss, and write fires an observer event.

Works with any server speaking the Redis protocol (Redis, Valkey, KeyDB, DragonflyDB, …).

Declare the cache on the port — ports.CachePattern¶

Cache keys are shaped like topics and paths — "user:{id}" is as declarable as "sensors/{sensorID}/data":

var UserCache = codex.Must(ports.NewIOPort[UserQuery, User]("user-cache",
    queryCodec, userCodec, ports.PortOptions{}))

var UserCachePattern = ports.CachePattern{Key: "user:{id}", TTL: 15 * time.Minute}

cache, err := UserCache.PluginCachePattern(UserCachePattern) // ports.Cache[User]

Per-key-variable codecs — ports.CacheKeyParam¶

By default, key vars are plain strings — Cache.BuildKey only checks that every {var} has an entry in vars (CacheKeyError on a missing var). Declare a CacheKeyParam to also validate the value, the same way rest.PathParam/events.TopicParam/ports.FilePathParam validate their templated variables:

var UserCache = codex.Must(ports.NewIOPort[UserQuery, User]("user-cache",
    queryCodec, userCodec, ports.PortOptions{}))

_, err := UserCache.PluginCachePattern(ports.CachePattern{
    Key: "user:{id}", TTL: 15 * time.Minute,
    Opts: []ports.CacheOpt{
        ports.CacheKeyParam{Name: "id"}.WithCodec(codex.String().Refine(validate.UUID)),
    },
})
if err != nil {
    panic(err)
}

A non-UUID id now returns ports.CacheKeyParamError{Key, Var, Value, Err} from Cache.BuildKey (and from redis.GetAdapter/SetAdapter, wrapped in their existing CacheError) — before any network call. A var with no matching CacheKeyParam (or one with a nil Codec) substitutes unvalidated, exactly as before this feature. Cache.ValidateKeyVars(vars) runs the same validation without building the key string; Cache.KeySchemas() returns each codec-bearing var's schema.Schema (forward-compatible plumbing for future spec tooling).

Standalone use — building without the pipeline features¶

If your application doesn't use ports/stream pipelines at all, Get and Set are the cache's equivalent of ports.File.Read/.Write or adapters/sql.Validate — plain functions, full codec validation (key vars AND value), no ports.IOAdapter, no stream.Stream, no port anywhere in the call path:

userCache := ports.NewCache("user:{id}", format.JSON(userCodec),
    ports.CacheKeyParam{Name: "id"}.WithCodec(codex.String().Refine(validate.UUID)))
userCache.TTL = 15 * time.Minute

v, ok, err := redis.Get(ctx, client, userCache, map[string]string{"id": userID}, redis.GetOptions{})
// v is fully decoded AND codec-validated; (zero, false, nil) on a miss

err = redis.Set(ctx, client, userCache, map[string]string{"id": user.ID}, user, redis.SetOptions{})
// key-var codec rejection (CacheKeyParamError) happens before any network call

ports.Cache[T] (via NewCache) is the declarative descriptor — the same role RouteHandle/ChannelHandle play via route.ClientHandle()/ channel.ClientHandle(). Get/Set are the concrete redis implementation against it, the same role nethttp.Call/mqtt.Publish play against a route/channel handle. This mirrors every other non-pipeline building block in go-codex — see Design pattern: declarative descriptor + plain function for the full comparison table across file/cache/rest/events/sql.

Seed is a thin wrapper around Get with nil vars — the one case that must run before any stream exists (a LatestPort's warm restart, before the feeding stream is even constructed), so it stays a dedicated zero-vars function rather than asking every caller to pass an empty map. It only applies to a var-free key (e.g. "latest-user", not "user:{id}"):

latestCache := ports.NewCache("latest-user", format.JSON(userCodec))
v, ok, err := redis.Seed(ctx, client, latestCache, redis.SeedOptions{})

Pipeline use — same functions, delegated to¶

GetAdapter/SetAdapter/DrainSetAdapter (the ports.IOAdapter/ SinkAdapter implementations bound via port.Bind) delegate to Get/Set internally per item — calling Get/Set directly and driving a fully-bound port produce identical behavior (same codec validation, same errors, same observer events). Use the adapters when you have an actual pipeline (ports.NewIOPort/SinkPort + Bind); use Get/Set directly everywhere else.

Port-type acceptance (when you DO use a port)¶

Port-type acceptance for CachePattern-declared caches (wrong combinations fail at construction with PatternRegisterError):

The narrow client interface — Commands¶

Constructors accept Commands — a three-method subset — never a concrete client:

type Commands interface {
    Get(ctx context.Context, key string) ([]byte, error) // ErrCacheMiss on missing key
    Set(ctx context.Context, key string, value []byte, ttl time.Duration) error
    Del(ctx context.Context, keys ...string) error
}

Wrap a go-redis client once (NewCommands is the only place go-redis is touched — standalone, cluster, and sentinel clients all satisfy redis.UniversalClient):

client := redis.NewCommands(goredis.NewUniversalClient(
    &goredis.UniversalOptions{Addrs: []string{"localhost:6379"}}))

Unit tests and the example use a hand-written in-memory fake — no test dependencies, no broker in CI.

Adapters¶

Durable LatestPort — the Feed-tee pattern¶

ports.LatestAdapter.Serve is read-only, so persistence composes from existing pieces instead of a new adapter: route the feeding stream through SetAdapter (persist every update), and Seed the first item after a restart:

latest, _ := ports.NewLatestPort[OEE]("oee-latest", oeeCodec, ports.PortOptions{})
cache, err := latest.PluginCachePattern(ports.CachePattern{Key: "latest-oee"})

// Persist every update on the way to Feed:
persisted := redis.SetAdapter[OEE](client, cache,
    func(OEE) map[string]string { return nil }, // var-free key
    redis.SetAdapterOptions{}).Transform(ctx, oeeStream)

// Warm restart: seed the stream with the last known value.
if v, ok, _ := redis.Seed(ctx, client, cache, redis.SeedOptions{}); ok {
    persisted = stream.Merge(ctx, stream.Single(ctx, v), persisted)
}
latest.Feed(ctx, persisted)

Errors¶

var ErrCacheMiss = errors.New("redis: cache miss") // sentinel, survives wrapping

type CacheError struct {
    Key string // expanded key, e.g. "user:42"
    Op  string // "get", "set", "del"
    Err error
}

CacheError implements Error(), Unwrap(), and slog.LogValuer — errors.Is(err, redis.ErrCacheMiss) and errors.As into codex.ValidationErrors both work through the chain. Key-template failures surface as ports.CacheKeyError{Key, Var}.

Error-path ergonomics — handle/log by default¶

Cache is an internal store boundary with no caller to respond to. DrainSetAdapter's OnError func(error) IS the handle action; leaving it nil is the log default (observer-only, via RecordCacheWrite). For a respond-equivalent, compose OnError with a declared events.ErrorChannel — see Error handling guide — store/IO boundaries for the full pattern.

Observer — stats.CacheObserver¶

A new optional extension (type-asserted, like SQLObserver — existing Observer implementations need not change):

type CacheObserver interface {
    RecordCacheHit(key string, duration time.Duration)
    RecordCacheMiss(key string, duration time.Duration)
    RecordCacheWrite(key, op string, success bool, duration time.Duration)
}

• Implemented by NoopObserver, LoggingObserver, and NewFanout.
• Nil Observer resolves from ctx (stats.ObserverFromContext).
• Decode validation failures additionally report per-field via stats.ReportErrors with location "payload".

Scope¶

Phase 1 is cache ops only. Redis pub/sub (SourcePort/SinkPort bridges à la MQTT) is designed in the Phase 2 roadmap — fire-and-forget without persistence (closer to ZeroMQ than MQTT), reusing EventPattern and the existing transport observer hooks. Redis Streams, Lua scripting, and client-side caching remain out of scope.