redis

package
v0.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 29, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package redis provides a typed cache boundary for go-codex pipelines, backed by any server speaking the Redis protocol.

Typed cache boundary

Values enter and leave the cache through the port's codex.Codec — every read decodes AND validates, every write validates AND encodes. The cache key is declared once as a template on the port (ports.CachePattern) and expanded per item; TTL and wire format (JSON/YAML/TOML) are part of the same declaration. 

var UserCache = codex.Must(ports.NewIOPort[UserQuery, User]("user-cache",
    queryCodec, userCodec, ports.PortOptions{
        Patterns: []ports.Pattern{
            ports.CachePattern{Key: "user:{id}", TTL: 15 * time.Minute},
        },
    }))

Port mapping

Narrow client interface

All constructors accept Commands — a three-method subset of the Redis command surface — never a concrete client. NewCommands adapts a go-redis redis.UniversalClient; unit tests use a hand-written fake. This is the only package file that imports go-redis.

Observer integration

Every lookup and write fires stats.CacheObserver events (hit, miss, write) when the configured Observer implements it — always type-asserted, existing Observer implementations need not change. Decode validation failures additionally report per-field via stats.ReportErrors with location "payload". A nil Observer resolves from ctx (stats.ObserverFromContext).

Errors

All failures are wrapped in CacheError (implements slog.LogValuer and Unwrap). A missing key surfaces as ErrCacheMiss — reachable through errors.Is on the wrapped chain.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var ErrCacheMiss = errors.New("redis: cache miss")

ErrCacheMiss is the sentinel for a missing cache key. Commands.Get implementations return it (wrapped or bare) when the key does not exist; GetAdapter maps it to skip-or-error per GetAdapterOptions.MissIsError. Test with errors.Is — it survives CacheError wrapping via Unwrap.

Functions

func DrainSetAdapter 

func DrainSetAdapter[T any](
	client Commands,
	cache ports.Cache[T],
	keyFn func(T) map[string]string,
	opts SetAdapterOptions,
) ports.SinkAdapter[T]

DrainSetAdapter returns a ports.SinkAdapter that writes every item to the cache — the terminal variant of SetAdapter for pipeline ends. Write errors go to SetAdapterOptions.OnError (dropped when nil); source stream errors are also forwarded to OnError. 

sinkPort.Bind(ctx, redis.DrainSetAdapter(client, cacheHandle,
    func(u User) map[string]string { return map[string]string{"id": u.ID} },
    redis.SetAdapterOptions{}))

keyFn may be nil under the same conditions as SetAdapter — vars are then derived PER-ITEM from each item's own merge-capable key fields.

func Get 

func Get[T any](ctx context.Context, client Commands, cache ports.Cache[T], vars map[string]string, opts GetOptions) (T, bool, error)

Get looks up a single value in the cache — full codec validation (key vars AND value), no ports.IOAdapter, no gstream.Stream involved. This is the plain-function standalone entrypoint for a non-pipeline application, mirroring ports.File.Read/adapters/sql.Validate: ports.Cache is the declarative descriptor (built via ports.NewCache or a ports.CachePattern), and Get is the concrete redis implementation of a read against it — the same relationship ports.File has to its Read/Write methods, or a route handle has to adapters/nethttp.Call. 

v, ok, err := redis.Get(ctx, client, userCache,
    map[string]string{"id": userID}, redis.GetOptions{})

GetAdapter delegates to Get per item — calling Get directly and driving GetAdapter through a bound ports.IOPort produce identical behavior.

Returns (zero, false, nil) on a miss by default (an empty cache is not an error); set MissIsError to get a CacheError wrapping ErrCacheMiss instead. Key-build, transport, and decode failures return CacheError.

func GetAdapter 

func GetAdapter[Req, Resp any](
	client Commands,
	cache ports.Cache[Resp],
	keyFn func(Req) map[string]string,
	opts GetAdapterOptions,
) ports.IOAdapter[Req, Resp]

GetAdapter returns a ports.IOAdapter that looks each Req up in the cache. The key is built from cache.Key (ports.CacheHandle) with vars extracted by keyFn; the stored bytes are decoded and codec-validated through cache.Format. Use with ports.IOPort.Bind: 

cacheHandle, _ := ports.CacheHandle[User](userPort)
userPort.Bind(ctx, redis.GetAdapter(client, cacheHandle,
    func(q UserQuery) map[string]string { return map[string]string{"id": q.ID} },
    redis.GetAdapterOptions{}))

Hit → decoded Resp downstream + RecordCacheHit. Miss → skip (or CacheError wrapping ErrCacheMiss when MissIsError) + RecordCacheMiss. Key-build, transport, and decode failures → CacheError on Stream.Errors.

Looks up via GetMerged — when cache declares merge-capable key params (via ports.NewCacheKeyParam), the vars derived from keyFn(Req) are ADDITIONALLY merged into the decoded Resp (e.g. a key-derived id is populated onto Resp automatically). Identical to a bare Get when cache declares no merge fields.

Example 
package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	adapterredis "github.com/DaniDeer/go-codex/adapters/redis"
	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/ports"
	"github.com/DaniDeer/go-codex/stream"
	"github.com/DaniDeer/go-codex/validate"
)

type user struct {
	ID   string
	Name string
}

var userCodec = codex.Struct[user](
	codex.RequiredField("id", codex.String().Refine(validate.NonEmptyString),
		func(u user) string { return u.ID },
		func(u *user, v string) { u.ID = v },
	),
	codex.RequiredField("name", codex.String().Refine(validate.NonEmptyString),
		func(u user) string { return u.Name },
		func(u *user, v string) { u.Name = v },
	),
)

type userQuery struct{ ID string }

// fakeCommands is the hand-written Commands fake — an in-memory map plus
// recorded calls. No live Redis, no miniredis.
type fakeCommands struct {
	mu     sync.Mutex
	store  map[string][]byte
	ttls   map[string]time.Duration
	getErr error
	setErr error
}

func newFake() *fakeCommands {
	return &fakeCommands{store: map[string][]byte{}, ttls: map[string]time.Duration{}}
}

func (f *fakeCommands) Get(_ context.Context, key string) ([]byte, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.getErr != nil {
		return nil, f.getErr
	}
	b, ok := f.store[key]
	if !ok {
		return nil, adapterredis.ErrCacheMiss
	}
	return b, nil
}

func (f *fakeCommands) Set(_ context.Context, key string, value []byte, ttl time.Duration) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.setErr != nil {
		return f.setErr
	}
	f.store[key] = value
	f.ttls[key] = ttl
	return nil
}

func (f *fakeCommands) Del(_ context.Context, keys ...string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	for _, k := range keys {
		delete(f.store, k)
	}
	return nil
}

func userCache() ports.Cache[user] {
	port, err := ports.NewIOPort[userQuery, user]("user-cache", codex.Struct[userQuery](
		codex.RequiredField("id", codex.String(),
			func(q userQuery) string { return q.ID },
			func(q *userQuery, v string) { q.ID = v },
		),
	), userCodec, ports.PortOptions{})
	if err != nil {
		panic(err)
	}
	c, err := port.PluginCachePattern(ports.CachePattern{Key: "user:{id}", TTL: 15 * time.Minute})
	if err != nil {
		panic(err)
	}
	return c
}

func queryStream(qs ...userQuery) stream.Stream[userQuery] {
	ch := make(chan userQuery, len(qs))
	for _, q := range qs {
		ch <- q
	}
	close(ch)
	errCh := make(chan error)
	close(errCh)
	return stream.Stream[userQuery]{Values: ch, Errors: errCh}
}

func main() {
	ctx := context.Background()
	fake := newFake()
	fake.store["user:42"] = []byte(`{"id":"42","name":"Ada"}`)

	cache := userCache()
	adapter := adapterredis.GetAdapter[userQuery, user](fake, cache,
		func(q userQuery) map[string]string { return map[string]string{"id": q.ID} },
		adapterredis.GetAdapterOptions{})

	out := adapter.Transform(ctx, queryStream(userQuery{ID: "42"}, userQuery{ID: "404"}))
	vals, _ := stream.Collect(ctx, out)
	for _, u := range vals {
		fmt.Println(u.Name)
	}
}

Output:
Ada

func GetMerged 

func GetMerged[T any](ctx context.Context, client Commands, cache ports.Cache[T], vars map[string]string, opts GetOptions) (T, bool, error)

GetMerged is the decode-merge convenience: it looks up exactly like Get, then ADDITIONALLY merges vars into the SAME returned value via codex.DecodeVars, using the merge-capable fields registered via ports.NewCacheKeyParam — mirrors ports.File.ReadMerged/ [events.ChannelHandle.DecodeMerged] for the cache boundary.

Additive — Get is unchanged; GetMerged behaves identically to a bare Get when the cache declares no merge-capable key params (ports.Cache.MergeFields is empty) or on a miss (nothing to merge into).

Example — key template "user:{id}" declared with ports.NewCacheKeyParam, so the extracted id is merged into the returned struct's own field: 

user, ok, err := redis.GetMerged(ctx, client, userCache, map[string]string{"id": id}, redis.GetOptions{})
// ok && user.ID == id, no manual assignment needed.

func Seed 

func Seed[T any](ctx context.Context, client Commands, cache ports.Cache[T], opts SeedOptions) (T, bool, error)

Seed reads the cache handle's var-free key and returns the decoded, codec-validated value — the warm-restart read for a durable ports.LatestPort: persist updates with SetAdapter/DrainSetAdapter on the feeding stream, then Seed the first item after a restart. A thin wrapper around Get with nil vars — the one case that must run before any stream exists, so it stays a dedicated zero-vars function rather than asking every LatestPort warm-restart caller to pass an empty map. 

if v, ok, err := redis.Seed(ctx, client, cacheHandle, redis.SeedOptions{}); ok && err == nil {
    seeded := stream.Single(ctx, v)
    // Merge seeded with the live stream before latestPort.Feed.
}

Returns (zero, false, nil) on a miss — an empty cache is not an error. Decode and transport failures return (zero, false, CacheError).

Example 
package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	adapterredis "github.com/DaniDeer/go-codex/adapters/redis"
	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/ports"
	"github.com/DaniDeer/go-codex/validate"
)

type user struct {
	ID   string
	Name string
}

var userCodec = codex.Struct[user](
	codex.RequiredField("id", codex.String().Refine(validate.NonEmptyString),
		func(u user) string { return u.ID },
		func(u *user, v string) { u.ID = v },
	),
	codex.RequiredField("name", codex.String().Refine(validate.NonEmptyString),
		func(u user) string { return u.Name },
		func(u *user, v string) { u.Name = v },
	),
)

type userQuery struct{ ID string }

// fakeCommands is the hand-written Commands fake — an in-memory map plus
// recorded calls. No live Redis, no miniredis.
type fakeCommands struct {
	mu     sync.Mutex
	store  map[string][]byte
	ttls   map[string]time.Duration
	getErr error
	setErr error
}

func newFake() *fakeCommands {
	return &fakeCommands{store: map[string][]byte{}, ttls: map[string]time.Duration{}}
}

func (f *fakeCommands) Get(_ context.Context, key string) ([]byte, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.getErr != nil {
		return nil, f.getErr
	}
	b, ok := f.store[key]
	if !ok {
		return nil, adapterredis.ErrCacheMiss
	}
	return b, nil
}

func (f *fakeCommands) Set(_ context.Context, key string, value []byte, ttl time.Duration) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.setErr != nil {
		return f.setErr
	}
	f.store[key] = value
	f.ttls[key] = ttl
	return nil
}

func (f *fakeCommands) Del(_ context.Context, keys ...string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	for _, k := range keys {
		delete(f.store, k)
	}
	return nil
}

func userCache() ports.Cache[user] {
	port, err := ports.NewIOPort[userQuery, user]("user-cache", codex.Struct[userQuery](
		codex.RequiredField("id", codex.String(),
			func(q userQuery) string { return q.ID },
			func(q *userQuery, v string) { q.ID = v },
		),
	), userCodec, ports.PortOptions{})
	if err != nil {
		panic(err)
	}
	c, err := port.PluginCachePattern(ports.CachePattern{Key: "user:{id}", TTL: 15 * time.Minute})
	if err != nil {
		panic(err)
	}
	return c
}

func main() {
	ctx := context.Background()
	fake := newFake()
	fake.store["latest-oee"] = []byte(`{"id":"line-1","name":"0.87"}`)

	cache := ports.Cache[user]{Key: "latest-oee", Format: userCache().Format}
	v, ok, err := adapterredis.Seed(ctx, fake, cache, adapterredis.SeedOptions{})
	fmt.Println(ok, err == nil, v.Name)
}

Output:
true true 0.87

func Set 

func Set[T any](ctx context.Context, client Commands, cache ports.Cache[T], vars map[string]string, v T, opts SetOptions) error

Set writes a single value to the cache — full codec validation (key vars AND value), no ports.IOAdapter, no gstream.Stream involved. This is the plain-function standalone entrypoint for a non-pipeline application, mirroring ports.File.Write/adapters/sql.Validate: ports.Cache is the declarative descriptor, and Set is the concrete redis implementation of a write against it. 

err := redis.Set(ctx, client, userCache,
    map[string]string{"id": user.ID}, user, redis.SetOptions{})

SetAdapter and DrainSetAdapter delegate to Set per item — calling Set directly and driving them through a bound port produce identical behavior. Errors are CacheError; unlike the pipeline adapters, Set returns the error directly instead of routing it past a "pass through regardless" step, since there is no downstream to protect.

func SetAdapter 

func SetAdapter[T any](
	client Commands,
	cache ports.Cache[T],
	keyFn func(T) map[string]string,
	opts SetAdapterOptions,
) ports.IOAdapter[T, T]

SetAdapter returns a ports.IOAdapter that writes each item to the cache (write-through) and passes it through unchanged. Encode and write failures go to Stream.Errors as CacheError — the item is STILL passed through: a cache write failure must not drop pipeline data. 

port.Bind(ctx, redis.SetAdapter(client, cacheHandle,
    func(u User) map[string]string { return map[string]string{"id": u.ID} },
    redis.SetAdapterOptions{}))

keyFn may be nil when cache declares merge-capable key params (via ports.NewCacheKeyParam): vars are then derived PER-ITEM from each item's own merge fields automatically via SetHandle — the same "one struct, one call" convenience [mqtt5.PublishHandle] provides. Pass a non-nil keyFn to keep building the map yourself (e.g. no merge fields declared, or vars come from a field the cached type doesn't have).

func SetHandle 

func SetHandle[T any](ctx context.Context, client Commands, cache ports.Cache[T], v T, opts SetOptions) error

SetHandle is the single-call convenience wrapper around Set: it derives the key vars from v automatically via codex.EncodeVars(v, ports.Cache.MergeFields()...) — one struct in, no manual vars map — mirroring [mqtt5.PublishHandle]/ports.WriteHandle's convenience for the cache boundary.

Set remains available as the lower-level escape hatch for callers that build the vars map themselves (e.g. no merge-capable key params declared, or vars come from a non-struct source). 

err := redis.SetHandle(ctx, client, userCache, user, redis.SetOptions{})
// key derived from user's own ID field — no manual vars map.

Types

type CacheError 

type CacheError struct {
	// Key is the expanded cache key (e.g. "user:42"). May be empty when key
	// building itself failed.
	Key string
	// Op is the operation: "get", "set", or "del".
	Op string
	// Err is the underlying error.
	Err error
}

CacheError wraps any cache operation failure: key building, transport, encode/decode, or a miss surfaced as an error.

func (CacheError) Error 

func (e CacheError) Error() string

func (CacheError) LogValue 

func (e CacheError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (CacheError) Unwrap 

func (e CacheError) Unwrap() error

Unwrap allows errors.Is and errors.As to reach the underlying error, including ErrCacheMiss and codex.ValidationErrors.

type Commands 

type Commands interface {
	// Get returns the value stored at key, or an error satisfying
	// errors.Is(err, ErrCacheMiss) when the key does not exist.
	Get(ctx context.Context, key string) ([]byte, error)
	// Set stores value at key. ttl zero means no expiry.
	Set(ctx context.Context, key string, value []byte, ttl time.Duration) error
	// Del removes the given keys. Deleting a missing key is not an error.
	Del(ctx context.Context, keys ...string) error
}

Commands is the narrow Redis command surface the adapters use. Constructors accept this interface — never a concrete client — so unit tests run against a hand-written fake and the adapter stays decoupled from the client library.

NewCommands adapts a go-redis client. Any other implementation works as long as it honours the ErrCacheMiss contract on Get.

func NewCommands 

func NewCommands(c goredis.UniversalClient) Commands

NewCommands wraps a go-redis goredis.UniversalClient as Commands. *redis.Client, *redis.ClusterClient, and sentinel-failover clients all satisfy UniversalClient — one shim covers every deployment shape. This is the only place the adapter touches go-redis.

type GetAdapterOptions 

type GetAdapterOptions struct {
	// MissIsError routes a cache miss to Stream.Errors as a [CacheError]
	// wrapping [ErrCacheMiss]. Default false: a miss emits nothing for the
	// item (the IOAdapter 0..N contract) — the idiomatic read-through shape
	// where downstream handles only cache hits.
	MissIsError bool
	// Buffer sizes the output channels. Default 0 (unbuffered).
	Buffer int
	// Observer receives [stats.CacheObserver] hit/miss events and per-field
	// validation reports. Resolved from ctx when nil.
	Observer stats.Observer
}

GetAdapterOptions configures GetAdapter.

type GetOptions 

type GetOptions struct {
	// MissIsError, when true, returns a [CacheError] wrapping [ErrCacheMiss]
	// on a cache miss instead of (zero, false, nil).
	MissIsError bool
	// Observer receives [stats.CacheObserver] hit/miss events and per-field
	// validation reports. Resolved from ctx when nil.
	Observer stats.Observer
}

GetOptions configures Get.

type SeedOptions 

type SeedOptions struct {
	// Observer receives the [stats.CacheObserver] hit/miss event.
	// Resolved from ctx when nil.
	Observer stats.Observer
}

SeedOptions configures Seed.

type SetAdapterOptions 

type SetAdapterOptions struct {
	// TTL overrides the cache handle's declared TTL when non-zero.
	TTL time.Duration
	// Buffer sizes the output channels ([SetAdapter] only). Default 0.
	Buffer int
	// OnError receives each write error ([DrainSetAdapter] only — SetAdapter
	// routes errors to its output Stream.Errors).
	OnError func(error)
	// Observer receives [stats.CacheObserver] write events. Resolved from
	// ctx when nil.
	Observer stats.Observer
}

SetAdapterOptions configures SetAdapter and DrainSetAdapter.

type SetOptions 

type SetOptions struct {
	// TTL overrides the cache descriptor's declared TTL when non-zero.
	TTL time.Duration
	// Observer receives [stats.CacheObserver] write events. Resolved from
	// ctx when nil.
	Observer stats.Observer
}

SetOptions configures Set.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.