README
¶
nexssp/storage — Flow bundle
This package (nexssp/storage/nexssflow/) is the Flow bundle that
bridges the nexssp/storage adapter registry to .nflow programs.
It is loaded via @require — no Flow checkout changes are needed.
Install
@require github.com/nexssp/storage v0.1.0
@require resolves to github.com/nexssp/storage/nexssflow (the
default nexssflow subdir convention). The bundle's init() calls
core.Register("storage", Bundle) and registers the storage.events
named hook.
Three layers of API
Layer 1 — Direct actions
@storage kv:memory as cache { ttl: "5m" }
noop -> storage.kv.set @{ store: "cache", key: "k", value: "v" }
noop -> storage.kv.get @{ store: "cache", key: "k" }
Eight actions: storage.kv.{get,set,del}, storage.blob.{get,put,del},
storage.lock.{acquire,release}. Each takes a store: argument
naming the @storage declaration. Blob put accepts streamed input and
returns an xfile.Ref; Blob get is a metadata-only lookup returning
{ ref, hit }, never the payload bytes. All Blob operations require an
authenticated caller-and-tenant scope in the request context.
Layer 2 — Middleware wiring (the practical path)
@storage kv:memory as cache { ttl: "5m" }
# Wire Kernel's CacheMiddleware to use the declared store.
# First call misses → handler runs → back-fills.
# Second call hits → handler does not run.
http.request:cached="cache:30s" @{ url: "https://api.example/users/1" }
Three modifiers, all wired by the AtomAdvise hook:
| Modifier | Wires | Value |
|---|---|---|
:cached="store:ttl" |
CacheMiddleware |
<store-name>:<duration> |
:idempotent_store="store" |
IdempotencyMiddleware |
<store-name> |
:dedup_store="store" |
Dedup |
<store-name> |
The store name may contain colons (e.g. kv:redis); the TTL is
parsed from the segment after the last colon.
Layer 3 — Hook observation
@storage kv:memory as cache { ttl: "5m" }
@hook:storage.events
noop -> storage.kv.get @{ store: "cache", key: "k" }
The storage.events hook stamps a *StorageEvents accumulator into
the request context. Every storage.* action records
cache_hit / cache_miss / error events into it. Test fixtures
and observability probes read it via HookProbeKey.
Directive syntax
@storage <kind>:<backend> [as <alias>] { options }
<kind>— one ofkv,blob,lock,fenced_lock,idem,journal,tx,rate_limit. Unknown kinds fail at compile time.<backend>— the adapter name (e.g.memory,redis,s3).as <alias>— optional. Renames the store for this compilation.{ options }— optional. Backend-specific key/value pairs.
Fixtures
nflow run nflows/basic.nflow # KV set + get roundtrip
nflow run nflows/kv_miss.nflow # Missing key returns hit=false
nflow run nflows/blob.nflow # Blob put + get roundtrip
nflow run nflows/lock.nflow # Lock acquire + release
nflow run nflows/alias.nflow # `as` alias
nflow run nflows/read_through.nflow # :cached= read-through pattern
nflow run nflows/idempotent.nflow # :idempotent_store= pattern
nflow run nflows/dedup.nflow # :dedup_store= pattern
Tests
go test -race ./...
The advisor tests (advise_test.go) prove the middleware wiring
works end-to-end: a :cached= modifier causes the second call to be
served from cache with zero handler invocations.
Why this lives in nexssp/storage/nexssflow/, not in flow/extensions/
Three reasons:
-
Adapter packages depend on nexssp/storage, not on Flow. The SQLite, PostgreSQL, and Redis packages in this module import
github.com/nexssp/storagefor theBackendinterface. If the Flow bundle lived insideflow/extensions/storage/, the Redis adapter would have to import Flow to get the bundle — backwards. -
One
@requireloads the built-ins. The user writes one line, not two. The bundle registers memory, SQLite, PostgreSQL, and Redis backends. Additional third-party adapters are integrated by adding their registration to this bundle; no Flow repository change is needed. -
Flow stays focused on language features.
flow/extensions/loop,flow/extensions/on_error,flow/extensions/macrosare language features. Storage is an adapter — it belongs with the adapter contract, not with the grammar.
This matches the pattern documented in
flow/extensions/README.md §External repositories:
"Any repo can become a Flow extension by adding a nexssflow/ package."
License
Apache License 2.0.
Documentation
¶
Index ¶
Constants ¶
const ID = "storage"
Variables ¶
var HookProbeKey = xctx.NewKey[*StorageEvents]("flow.storage.events")
HookProbeKey is the context key set by the storage.events hook when any storage event fires. Test fixtures and runtime probes read it to assert that hooks fired.
The value is a *StorageEvents pointer (never nil when set) so consumers can read the accumulated event log without copying.
Functions ¶
func Bundle ¶
Bundle returns a fresh Bundle for a single @require declaration.
A single bundle instance owns:
- one storage.Registry (the backends are shared, read-only, for every compilation this bundle serves), and
- a list of per-compilation stores it created, so Shutdown can close every opened Instance on teardown.
The three callbacks below cooperate on a single per-compilation store. The order is:
- OnPreprocess creates the store from meta["storage"] once, installs the :cache_store= / :idempotent_store= / :dedup_store= advisor against it, and records it in meta[storeMetaKey].
- Materialize reuses the store from meta (does NOT create a second one) and mounts <alias>.<kind>.<op> actions that close over it.
- WrapPipeline injects the same store into the runtime context so the generic storage.* actions can resolve it via currentStore(ctx).
If a flow declares no @storage directives, all three callbacks are no-ops and the compiled program is not wrapped at all.
Types ¶
type StorageEvent ¶
type StorageEvent struct {
Kind string // "cache_hit", "cache_miss", "error"
Store string // the @storage declaration name
Action string // the action name, e.g. "storage.kv.get"
Key string // the key passed to the action, if any
}
StorageEvent records one storage operation outcome.
type StorageEvents ¶
type StorageEvents struct {
// contains filtered or unexported fields
}
StorageEvents is the per-request accumulator that the storage.events hook stamps into the context on first miss/hit. Concurrent-safe; intended for observability and test assertion.
func (*StorageEvents) Events ¶
func (s *StorageEvents) Events() []StorageEvent
Events returns a snapshot copy of the accumulated events, in chronological order.
func (*StorageEvents) Record ¶
func (s *StorageEvents) Record(e StorageEvent)
Record appends an event. Safe for concurrent use.
Source Files