cf_valkey_state

package module
v0.0.1 Latest Latest
Warning

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

 Go to latest Published: Aug 9, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

README

caerus-framework-valkey-state

CI codecov License

Caerus Framework Valkey State Component. Keyed state services — sessions, cache, and rate limiting — built on top of the caerus-framework-valkey component, so apps (e.g. caerus-auth-api) don't import valkey-go or manage keys/TTLs themselves. Framework-owned lifecycle, configuration (file + env + flags), live tunable reload with last-good semantics, logging through the framework logs component, and observability health/metrics.

The component is a stateless consumer of a CFValkey peer: it holds the peer component pointer (never a client snapshot) and builds every command through the peer's live Client() and prefix-aware Key(), so reconnects and key prefixes stay consistent and the peer's OnConfigReload owns client rotation.

Wiring

Two wiring shapes are supported. Prefer the app-owned shape (demoapp golden path): main declares only the chassis (valkey-state alongside postgres / valkey) and the app class; product machinery that uses sessions, cache, or rate limits lives under the app and resolves the component as a peer at Init. Use the simple main-level shape for one-off binaries.

App-owned consumer (golden — demoapp pattern)

main declares valkey-state as chassis and runs the app class; it never touches the component itself:

fw := cf.New(&cf.FrameworkOptions{
	Logs: &cf.LogsSettings{Format: "json", Level: "info", ConfigSource: "logs"},
	Observability: &cf.ObservabilitySettings{Address: ":9090", ConfigSource: "observability"},
	Components: []cf.CaerusComponent{
		cf_postgres.New(cf_postgres.WithConfigSource("postgresql", "config/postgresql.json")),
		cf_valkey.New(cf_valkey.WithConfigSource("valkey", "config/valkey.json")),
		cf_valkey_state.New(cf_valkey_state.WithConfigSource("state", "config/state.json")),
		app.New(app.Options{}),
	},
})
if err := fw.RunWithSignals(context.Background()); err != nil {
	log.Fatal(err)
}

The app resolves the valkey-state component pointer once at Init (never a client snapshot), declares it in GetDependencies, and calls the accessors per use:

type App struct {
	state *cf_valkey_state.CFState
}

func (a *App) GetDependencies() []string {
	return []string{cf_valkey_state.ComponentName} // + logs, chassis peers
}

func (a *App) Init(ctx context.Context, fw *cf.CaerusFramework) error {
	st, ok := cf.Get[*cf_valkey_state.CFState](fw)
	if !ok {
		return errors.New("app: valkey-state component missing")
	}
	a.state = st
	return nil
}

Multiple valkey-state instances in one process use WithName + GetByName[*cf_valkey_state.CFState](fw, "sessions"); a state component bound to a named valkey peer uses WithValkeyName (its GetDependencies reports the peer's actual name, so framework Validate passes).

Simple main-level wiring

For a one-off binary, register the components directly and use cf.MustGet to reach the component:

fw := cf.New()

logs := cf_logs.New(cf_logs.WithWriter(os.Stdout))
valkey := cf_valkey.New(cf_valkey.WithConfigSource("valkey", "config/valkey.json"))
state := cf_valkey_state.New(cf_valkey_state.WithConfigSource("state", "config/state.json"))
fw.AddComponent(logs)
fw.AddComponent(valkey) // GetDependencies() -> [logs configuration]
fw.AddComponent(state)  // GetDependencies() -> [valkey logs configuration]

In both shapes the component is cf.ConfigSourceRegistrar-self-sufficient: WithConfigSource registers the Source[StateConfig] with the configuration component during argv absorption, so main never touches os.Getenv/ParseFlags. The --state path flag and per-field flags come from the source declaration.

Usage

The app resolves the component once at Init (see Wiring above) and calls the accessors per use. All keys are namespaced (session:*, cache:*, rl:*) and built through the peer's prefix-aware Key(), so the valkey WithKeyPrefix applies on top.

// sessions — opaque JSON values with a TTL
sess := map[string]any{"user_id": "u1", "roles": []string{"admin"}}
_ = st.CreateSession(ctx, "tok-abc", sess, 24*time.Hour) // ttl 0 → configured default
var got map[string]any
found, _ := st.GetSession(ctx, "tok-abc", &got)          // found=false on miss/expiry
st.SessionExists(ctx, "tok-abc")
st.TouchSession(ctx, "tok-abc", 15*time.Minute)          // sliding window
st.RevokeSession(ctx, "tok-abc")

// cache — JSON cache-aside with a singleflight get-or-load
_ = st.CacheSet(ctx, someStruct, time.Minute, "catalog", "sku-1")
var cached someStruct
_ = st.CacheGet(ctx, &cached, "catalog", "sku-1") // no-op miss (dst untouched)

val, shared, _ := st.CacheGetOrLoad(ctx, time.Minute, func(ctx context.Context) ([]byte, error) {
	return json.Marshal(fetchFromDB(ctx))
}, "catalog", "sku-2")
// concurrent callers coalesce onto one load (process-local singleflight)

// rate limiting — atomic fixed-window counter (Lua)
res, _ := st.RateLimit(ctx, "login:"+email, 5, time.Minute)
if !res.Allowed {
	// respond 429; res.ResetIn is the window time remaining (Retry-After)
}
// res.Count is the current count within the window

CacheGetOrLoad's load runs at most once per key per in-flight wave inside this process (same singleflight semantics as the valkey patterns package); other pods still stampede on a cold key — compose with a valkey patterns mutex for cross-pod coalescing. Rate-limit errors are returned to the caller: fail-open/fail-closed policy belongs to the app.

Options

Option Description
WithConfig(StateConfig) static config snapshot; non-zero fields override option-set defaults
WithConfigSource(name, path, …) bind a configuration source for Init + OnConfigReload; the module registers the Source[StateConfig] itself (declares configuration dep)
WithSessionTTL(d) default session TTL when a call leaves ttl at zero (default 24h)
WithCacheTTL(d) default cache TTL when a call leaves ttl at zero (default 5m)
WithValkeyName(name) bind to a named valkey peer (WithName on the valkey side; default "valkey")
WithName(name) custom component name for multiple instances (default "valkey-state")
WithLogger(*slog.Logger) explicit logger override; defaults to the framework logs component's logger (re-delivered on logs Reconfigure), falling back to slog.Default()

Configuration

Load StateConfig through the configuration component. The default EnvPrefix is STATE_ (from the source name); env tags map STATE_SESSION_TTL_SEC, STATE_CACHE_TTL_SEC.

{
  "session_ttl_sec": 86400,
  "cache_ttl_sec": 300
}

The tunables apply live: on reload OnConfigReload re-reads the source and replaces the TTL defaults (last-good on failure). The valkey peer owns connection rotation, so a reload never rebuilds anything here.

Health reports initialized/uninitialized (the peer's client is the liveness source; real connectivity is the valkey component's own Health, aggregated by observability's /readyz). Metrics emits the following while initialized, nil before Init/after Shutdown:

Metric Type Labels
valkey_state_info gauge 1 component
valkey_state_config_reloads_total counter component
valkey_state_sessions_created_total counter component
valkey_state_sessions_revoked_total counter component
valkey_state_sessions_touched_total counter component
valkey_state_cache_hits_total counter component
valkey_state_cache_misses_total counter component
valkey_state_rate_allowed_total counter component
valkey_state_rate_rejected_total counter component

Counters appear at zero until first fired, so the series are always present while initialized. Cache hit ratio is valkey_state_cache_hits_total / (valkey_state_cache_hits_total + valkey_state_cache_misses_total); rate-limit rejection rate is rate(valkey_state_rate_rejected_total[5m]).

Tests

Unit tests cover the component contract, options/config layering, dependency declaration, and Init failure modes with no external service. Integration tests (real session/cache/rate-limit behavior, TTL expiry, sliding-window touch, singleflight coalescing, Lua atomicity) run only when the gate env var is set:

docker run -d --rm -p 6379:6379 --name v valkey/valkey:8
VALKEY_ADDR=127.0.0.1:6379 go test -race ./...

License

Apache License 2.0 — see LICENSE.

Documentation

Index

Constants

View Source 
const (
	// ComponentName is the framework component name for the valkey-state
	// component. It is the identifier other components use in GetDependencies
	// to require it.
	ComponentName = "valkey-state"

	// ComponentStage is the stage data-layer components initialize in. It is
	// not a built-in bootstrap stage; AddComponent registers it automatically
	// the first time a component declares it.
	ComponentStage = cf.Stage("data")
)

Variables

This section is empty.

Functions

This section is empty.

Types

type CFState 

type CFState struct {
	// contains filtered or unexported fields
}

CFState is the caerus-framework-valkey-state component. It is a stateless consumer of a cf_valkey.CFValkey peer: it holds the peer component (never a client snapshot) and builds every command through the peer's live Client() and prefix-aware Key(), so reconnects and key prefixes stay consistent. It provides three keyed state services over that client:

  • Sessions: opaque JSON values with a TTL and sliding-window touch.
  • Cache: JSON cache-aside plus a singleflight get-or-load.
  • Rate limit: atomic fixed-window counters.

func New 

func New(opts ...Option) *CFState

New creates a state component. The valkey peer is resolved at Init, not here.

func (*CFState) CacheGet 

func (c *CFState) CacheGet(ctx context.Context, dst any, parts ...string) error

CacheGet reads a JSON value at cache:<parts...> into dst. A miss returns nil and leaves dst untouched. Cache hit/miss counters are updated.

func (*CFState) CacheGetOrLoad 

func (c *CFState) CacheGetOrLoad(ctx context.Context, ttl time.Duration, load LoadFunc, parts ...string) (val []byte, shared bool, err error)

CacheGetOrLoad returns the bytes cached at cache:<parts...>, or runs load once for concurrent callers in this process (singleflight), then stores the result with ttl. A ttl <= 0 uses the configured default.

WARNING: singleflight is process-local only; other pods still stampede on a cold key. shared=true when this caller waited on another goroutine's load. Failure modes: a load error is returned to waiters and nothing is cached; a SET failure after a successful load returns the loaded bytes along with the error so callers can log without losing the value.

func (*CFState) CacheSet 

func (c *CFState) CacheSet(ctx context.Context, v any, ttl time.Duration, parts ...string) error

CacheSet marshals v as JSON and stores it at cache:<parts...> with the given TTL. A ttl <= 0 uses the configured default.

func (*CFState) Client 

func (c *CFState) Client() valkey.Client

Client returns the peer's live valkey client (nil before Init or after Shutdown). It follows the peer-pointer convention: the peer is re-read per use, so a client swap on the valkey side is picked up immediately.

func (*CFState) CreateSession 

func (c *CFState) CreateSession(ctx context.Context, id string, v any, ttl time.Duration) error

CreateSession stores a JSON session value at session:<id> with the given TTL. A ttl <= 0 uses the configured default. The session id is chosen by the caller (typically an opaque token); it is never derived from the value.

func (*CFState) GetDependencies 

func (c *CFState) GetDependencies() []string

GetDependencies implements cf.Dependencies. The component depends on the valkey component it consumes (the actual peer name when WithValkeyName is set, the default ComponentName otherwise), logs through the framework logs component, and depends on configuration when WithConfigSource is set. Peer names are fixed at construction, so the graph is stable before Init.

func (*CFState) GetInitOrderStage 

func (c *CFState) GetInitOrderStage() cf.Stage

GetInitOrderStage implements cf.CaerusComponent.

func (*CFState) GetSession 

func (c *CFState) GetSession(ctx context.Context, id string, dst any) (found bool, err error)

GetSession reads the session at session:<id> and unmarshals it into dst. found is false when the session does not exist or has expired. The session TTL is not extended here; call TouchSession for a sliding window.

func (*CFState) Health 

func (c *CFState) Health(ctx context.Context) error

Health implements cf.HealthProvider. It reports healthy while the peer's valkey client is initialized; real connectivity is owned by the valkey component's own Health (aggregated by observability's /readyz).

func (*CFState) Init 

func (c *CFState) Init(ctx context.Context, fw *cf.CaerusFramework) error

Init implements cf.CaerusComponent. It resolves the valkey peer component (by name or the default "valkey"), failing fast when it is missing or not yet initialized. No connection is opened here; the peer owns its client.

func (*CFState) Key 

func (c *CFState) Key(parts ...string) string

Key builds a prefix-aware key through the valkey peer. Useful for composing with caerus-framework-valkey/patterns helpers (e.g. patterns.NewMutex) against the same key space.

func (*CFState) Metrics 

func (c *CFState) Metrics() []cf_observability.Metric

Metrics implements cf_observability.MetricsProvider. It reports operation counters while the peer's client is initialized; before Init or after Shutdown it returns nil, so the observability component skips it (lazy pickup). Counters are cumulative for the process lifetime and are emitted (zero until first fired) so the series are always present.

func (*CFState) Name 

func (c *CFState) Name() string

Name implements cf.CaerusComponent. Returns the custom name set via WithName, or the default ComponentName ("valkey-state") if no custom name was set.

func (*CFState) OnConfigReload 

func (c *CFState) OnConfigReload(source string, cfg any)

OnConfigReload implements cf.ConfigReloader. It re-reads the behavior tunables (session/cache TTLs) from the bound configuration source. No connection is rebuilt: the valkey peer owns client rotation, and this component is stateless over it.

func (*CFState) RateLimit 

func (c *CFState) RateLimit(ctx context.Context, key string, limit int64, window time.Duration) (RateLimitResult, error)

RateLimit is an atomic fixed-window rate limiter. Each call increments the counter at rl:<key> and reports whether the resulting count is within limit for the window. The counter and its expiry are set atomically (Lua), so a key never leaks without a TTL and concurrent first calls cannot race the expiry.

The key is caller-chosen (e.g. "login:user@example.com" or "ip:1.2.3.4"). A transport error is returned to the caller: fail-open/fail-closed policy belongs to the app (auth-api keeps its "valkey down → allow request" choice).

func (*CFState) RegisterConfigSources 

func (c *CFState) RegisterConfigSources(conf any) error

RegisterConfigSources implements cf.ConfigSourceRegistrar. The framework calls it during argv absorption; it registers this component's configuration source (name, path, env prefix, format, Owner) with the configuration component. No-op when no source is bound.

func (*CFState) RevokeSession 

func (c *CFState) RevokeSession(ctx context.Context, id string) error

RevokeSession deletes session:<id>. Deleting a missing session is a no-op success.

func (*CFState) SessionExists 

func (c *CFState) SessionExists(ctx context.Context, id string) (bool, error)

SessionExists reports whether session:<id> exists and is unexpired.

func (*CFState) Shutdown 

func (c *CFState) Shutdown(ctx context.Context) error

Shutdown implements cf.CaerusComponent. It unsubscribes the logs subscription and drops the valkey peer. Further use returns an error.

func (*CFState) TouchSession 

func (c *CFState) TouchSession(ctx context.Context, id string, ttl time.Duration) error

TouchSession extends session:<id> with a fresh TTL (sliding window). A missing or expired session is a no-op success. A ttl <= 0 uses the configured default.

type LoadFunc 

type LoadFunc func(ctx context.Context) ([]byte, error)

LoadFunc fetches the canonical value on a cache miss. Called at most once per key per in-flight wave inside this process.

type Option 

type Option func(*options)

Option configures the state component at construction time.

func WithCacheTTL 

func WithCacheTTL(d time.Duration) Option

WithCacheTTL sets the default cache TTL used when a cache call leaves ttl at zero (default 5m).

func WithConfig 

func WithConfig(cfg StateConfig) Option

WithConfig sets a static configuration snapshot. Non-zero fields of cfg override the values set by the convenience options. Prefer WithConfigSource when using caerus-framework-configuration with hot-reload.

func WithConfigSource 

func WithConfigSource(name, path string, opts ...SourceOption) Option

WithConfigSource binds this component to a named configuration source and registers that source with the configuration component (via the framework's ConfigSourceRegistrar pass during argv absorption). The module owns the Source: the config type, the default EnvPrefix and its Owner (Name(), so named instances reload correctly). main only points the instance at where the config lives. 

cf_valkey_state.New(cf_valkey_state.WithConfigSource("state", "config/state.json"))
cf_valkey_state.New(cf_valkey_state.WithConfigSource("sess", "/etc/app/sess.yaml",
    cf_valkey_state.WithSourceFormat(cf_configuration.FormatYAML)))

A path of "" registers an env-only (fileless) source when the EnvPrefix is non-empty. The path CLI override stays --<source-name> (ParseFlags). Declares a dependency on "configuration".

func WithLogger 

func WithLogger(logger *slog.Logger) Option

WithLogger overrides the logger used for component diagnostics. By default the component logs through the framework logs component (declared in GetDependencies); WithLogger is an explicit override for tests and embedded use and wins over the framework logger. slog.Default() remains the fallback only when neither is available.

func WithName 

func WithName(name string) Option

WithName sets a custom component name, allowing multiple state instances in the same process. The default name is "valkey-state" (ComponentName). Retrieve named instances with GetByName[*CFState](fw, "sessions").

func WithSessionTTL 

func WithSessionTTL(d time.Duration) Option

WithSessionTTL sets the default session TTL used when a session call leaves ttl at zero (default 24h).

func WithValkeyName 

func WithValkeyName(name string) Option

WithValkeyName binds the state component to a valkey component with the given name (WithName on the valkey side). The default is the valkey ComponentName ("valkey").

type RateLimitResult 

type RateLimitResult struct {
	// Allowed is true when Count does not exceed the limit for the window.
	Allowed bool
	// Count is the incremented request count within the current window.
	Count int64
	// ResetIn is the time remaining until the window resets (for Retry-After).
	ResetIn time.Duration
}

RateLimitResult reports the outcome of a RateLimit check.

type SourceOption 

type SourceOption func(*sourceOptions)

SourceOption configures the self-registered configuration source created by WithConfigSource.

func WithSourceEnvPrefix 

func WithSourceEnvPrefix(prefix string) SourceOption

WithSourceEnvPrefix sets the environment overlay prefix for the source (default: the uppercase source name with "-" replaced by "_", plus "_"). An empty prefix disables env overlay.

func WithSourceFormat 

func WithSourceFormat(f cf_configuration.Format) SourceOption

WithSourceFormat forces the file format instead of inferring it from the path extension (".yaml"/".yml" → YAML; anything else JSON).

type StateConfig 

type StateConfig struct {
	// SessionTTLSec is the default session TTL in seconds when a call leaves
	// ttl at zero (default 24h).
	SessionTTLSec int64 `json:"session_ttl_sec,omitempty" yaml:"session_ttl_sec,omitempty" env:"SESSION_TTL_SEC"`
	// CacheTTLSec is the default cache TTL in seconds when a call leaves ttl
	// at zero (default 5m).
	CacheTTLSec int64 `json:"cache_ttl_sec,omitempty" yaml:"cache_ttl_sec,omitempty" env:"CACHE_TTL_SEC"`
}

StateConfig is the file/env-drivable behavior configuration. Load it through the configuration component (caerus-framework-configuration) and pass it via WithConfigSource; both JSON and YAML tags are provided.

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.