valkey

The valkey step executes cache commands or Pub/Sub subscriptions against a declared connection "valkey" instance.

# Read from cache
valkey "check_cache" {
  connection = "cache"
  op         = "get"
  key        = "users:${ctx.request.path.id}"
}

# Write with TTL
valkey "save_cache" {
  connection = "cache"
  op         = "set"
  key        = "users:${ctx.request.path.id}"
  value      = steps.fetch_user.row
  ttl        = "15m"
}

# Delete key
valkey "purge" {
  connection = "cache"
  op         = "del"
  key        = "users:${ctx.request.path.id}"
}

# Real-time Pub/Sub subscription
valkey "live_alerts" {
  connection = "cache"
  op         = "subscribe"
  channel    = "system.alerts"
}

Attributes

Attribute	Type	Required	Description
`connection`	`string`	yes	Declared `connection "valkey"` identifier
`op`	`string`	yes	`"get"`, `"set"`, `"del"`, or `"subscribe"`
`key`	`string`	for get/set/del	Cache key expression
`value`	`any`	for set	Payload serialized as a JSON string
`channel`	`string`	for sub	Pub/Sub channel name expression
`ttl`	`Duration`	no	Expiration duration like `"10m"` or `"24h"`
`when`	`bool`	no	Conditional expression

Exported outputs

For `op = "get"`

steps.<name>.found: true if the key exists, false on a cache miss.
steps.<name>.value: Unmarshaled JSON data, or null if missing.

A cache miss sets found = false without an error, allowing conditional fallbacks:

sql "fetch_db" {
  when       = !steps.check_cache.found
  connection = "main"
  query      = "SELECT id, email FROM users WHERE id = @id"
  args       = { id = ctx.request.path.id }
}

For `op = "set"` and `op = "del"`

steps.<name>.success: true when the operation succeeds.

For `op = "subscribe"`

steps.<name>.stream: An iter.Seq2[any, error] iterator yielding real-time messages. Pipe this directly into a terminal stream "sse" step.

#valkey

#Attributes

#Exported outputs

#For op = "get"