Selkie.git | lib/Selkie/ | Store.rakumod
=begin pod
=head1 NAME
Selkie::Store - Reactive state store with dispatch, effects, and subscriptions
=head1 SYNOPSIS
=begin code :lang<raku>
my $store = $app.store;
# Register a handler that returns effects (never mutate state directly)
$store.register-handler('counter/inc', -> $st, %ev {
my $current = $st.get-in('count') // 0;
(db => { count => $current + 1 },);
});
# Subscribe a widget to a path — marks it dirty when the value changes
$store.subscribe('my-counter', ['count'], $widget);
# Or: subscribe with a callback that does something on change
$store.subscribe-with-callback(
'counter-text',
-> $s { "Count: {$s.get-in('count') // 0}" },
-> $text { $label.set-text($text) },
$label,
);
# Fire an event from anywhere — any thread, even — it's queued and
# processed on the next tick. The queue is lock-protected.
$store.dispatch('counter/inc');
=end code
=head1 DESCRIPTION
Selkie's store is inspired by L<re-frame|https://github.com/day8/re-frame> —
one centralized state atom with a one-way data flow:
user action → dispatch event
event → handlers return effects
effects → mutate state (via registered effect handlers)
state → subscriptions notify widgets
widgets → re-render
Handlers are B<pure functions>. Given a store and event payload, they
return a list of effects. Effects are where side effects live — built-in
ones (C<db>, C<db-replace>, C<db-delete>, C<db-update>, C<dispatch>,
C<async>) cover most needs; you can register your own with
C<register-fx>.
=head2 Why?
Why go through all this ceremony versus just mutating state? Because:
=item State changes are B<auditable> — every mutation is a named event with a payload
=item Handlers are B<testable> — no notcurses, no widgets, just pure functions
=item Time-travel / logging / middleware become possible without changing app code
=item Subscriptions B<derive> UI from state — you don't manually sync widgets with state
You don't have to use the store. For small apps, widget Supplies taped
directly to side-effecting code works fine. The store shines when shared
state grows — multiple widgets reading the same data, actions that
cascade, async workflows with multiple steps.
=head2 Thread safety
C<dispatch> is the B<only> store method that may be called from a thread
other than the main loop. The event queue is lock-protected, so worker
threads — the C<async> effect's on-success / on-failure dispatches, or
app-owned C<start> blocks reporting progress — can enqueue concurrently
with the main loop's drain without losing events. Everything else
(handlers, effects, subscriptions, C<tick>, state reads and writes) runs
on the main loop thread; off-thread dispatches are simply picked up on
the next tick. To get work I<onto> a worker thread, use the C<async>
effect; to get results I<back>, dispatch — never touch state directly
from a worker.
=head1 EXAMPLES
=head2 A counter
Two handlers, one subscription:
=begin code :lang<raku>
$store.register-handler('counter/inc', -> $st, %ev {
(db => { count => ($st.get-in('count') // 0) + 1 },);
});
$store.register-handler('counter/reset', -> $st, %ev {
(db => { count => 0 },);
});
$store.subscribe-with-callback(
'counter-display',
-> $s { "Count: {$s.get-in('count') // 0}" },
-> $text { $display.set-text($text) },
$display,
);
$inc-button.on-press.tap: -> $ { $store.dispatch('counter/inc') };
$reset-button.on-press.tap: -> $ { $store.dispatch('counter/reset') };
=end code
=head2 Chaining events
The C<dispatch> effect lets a handler trigger another event. Use it when
one action implies another:
=begin code :lang<raku>
$store.register-handler('user/logged-in', -> $st, %ev {
(
db => { user => %ev<user> },
dispatch => { event => 'inbox/fetch' },
);
});
=end code
=head2 Replacing or deleting a subtree
C<db> is a deep-merge effect. Use C<db-replace> when you need an exact
replacement, including replacing a populated Hash with an empty Hash:
=begin code :lang<raku>
$store.register-handler('image-gen/clear', -> $st, %ev {
(db-delete => { path => <ui image-gen> },);
});
$store.register-handler('filters/reset', -> $st, %ev {
(db-replace => { path => <ui filters>, value => %() },);
});
=end code
Both effects notify subscriptions watching the target path or overlapping
ancestor/descendant paths.
=head2 Async work
The C<async> effect runs work on a worker thread, then dispatches a
follow-up event with the result (or error). The handler itself returns
immediately — the store doesn't block:
=begin code :lang<raku>
$store.register-handler('inbox/fetch', -> $st, %ev {
(async => {
work => -> { fetch-messages-from-api() },
on-success => 'inbox/loaded',
on-failure => 'inbox/load-failed',
},);
});
$store.register-handler('inbox/loaded', -> $st, %ev {
(db => { inbox => %ev<result> },);
});
$store.register-handler('inbox/load-failed', -> $st, %ev {
(db => { error => %ev<error> },);
});
=end code
=head2 Computed subscriptions
C<subscribe-computed> watches a derived value. Fires only when the
computed result changes, not every time the underlying state does:
=begin code :lang<raku>
$store.subscribe-computed(
'unread-count',
-> $s { $s.get-in('inbox').grep(*<read> == False).elems },
$badge-widget,
);
=end code
=head2 Debugging
Turn on logging during development to see the data flow:
=begin code :lang<raku>
$app.store.enable-debug; # logs to $*ERR
# or to a file:
$app.store.enable-debug(log => open('store.log', :w));
=end code
Output:
[1776073200.123] dispatch task/add text=Buy milk
[1776073200.123] → db: {tasks => [...], next-id => 5}
[1776073200.124] sub[task-list] fired: [...]
=head1 EFFECTS
Handlers return effects rather than mutating state directly. Each effect
is a C<Pair> of C<name => params>, or an C<Associative> with multiple
pairs.
=head2 Built-in effects
=item C<< db => { ... } >> — deep-merge into the state tree. Nested hashes are merged recursively; non-hash values are set directly. This is the workhorse effect — most handlers return one. An empty Hash merged into a populated Hash is a no-op because there are no keys to merge; use C<db-replace> or C<db-delete> when you need to clear a subtree.
=item C<< db-replace => { path => <ui filters>, value => %() } >> — replace the value at C<path> exactly, auto-creating intermediate Hashes like C<assoc-in>. Use this when the new value must replace the old subtree rather than merge into it.
=item C<< db-delete => { path => <ui image-gen> } >> — delete the key at C<path>. Missing paths are ignored. Use this to represent absence rather than storing a sentinel such as C<Nil>.
=item C<< db-update => { fn => &fn } >> — transform the whole state with a pure function: C<&fn> receives the current state and returns the new state. On a typed store (see TYPED STORE below) this is the primary write effect — C<&fn> takes the root schema instance and returns a patched clone, and only the actually-changed paths notify subscribers. On the untyped store C<&fn> takes the state Hash, must return an Associative, and the whole tree is marked dirty (the digest gate still stops no-op subscriber fires).
=item C<< dispatch => { event => 'name', ...payload } >> — enqueue another event. Processed in the same tick.
=item C<< async => { work => &fn, on-success => 'name', on-failure => 'name' } >> — run C<&fn> on a worker thread. On return, dispatch C<on-success> with a C<result> payload. On throw, dispatch C<on-failure> with an C<error> payload.
=head2 Event and effect payload contracts
Event names are C<Str:D>. C<dispatch($event, *%payload)> enqueues the
event name plus exactly the named payload pairs the caller supplied.
Handlers registered with C<register-handler> are called as
C<&handler($store, %payload)> in registration order. Missing event
handlers are a no-op.
Handlers return effects. Each returned effect must be either a C<Pair>
whose key is the effect name and whose value is the parameter payload,
or an C<Associative> whose key/value pairs are effect names and
parameter payloads. Effect parameter payloads must be C<Associative>;
scalar payloads are rejected and routed through C<__effect-error>.
Unknown effect names are ignored.
Built-in effect payloads:
=item C<db>: the payload itself is a Hash-shaped update tree. Nested
Associatives are deep-merged into existing Associatives; any other
value replaces the target key. Empty Associatives merge no keys.
=item C<db-replace>: requires C<path> and C<value>. C<path> may be a
Positional path or a single scalar key and must not be empty. Missing
intermediate paths are created as Hashes. C<value> replaces the target
subtree exactly.
=item C<db-delete>: requires non-empty C<path>, with the same path shape
as C<db-replace>. Missing paths are ignored. Existing target keys are
deleted.
=item C<db-update>: requires C<fn>, a Callable called with the current
state and returning the new state. On a typed store the return must be
an instance of the store's state-class; on the untyped store it must
be C<Associative>. A wrong-shaped return dies and routes through
C<__effect-error> like any other effect failure.
=item C<dispatch>: requires C<event>. All other keys are forwarded as
the payload for that queued event.
=item C<async>: requires C<work>, a Callable run on a worker thread.
C<on-success> is optional; when present it receives a payload with
C<result>. C<on-failure> is optional; when present it receives a
payload with C<error> (a Str), C<exception> (the original exception
object), and C<backtrace> (a Str).
Effect handler exceptions and event handler exceptions do not escape the
dispatch loop. They enqueue C<__effect-error> unless the failing event is
already C<__effect-error>. Its payload is:
=item C<effect-name> — the effect name, or C<event-handler[$event]> for
an event-handler exception
=item C<event> — the event whose handler produced the failing effect
(or whose handler threw), so a validation failure names the dispatch
that caused it; empty string when the effect ran outside an event
context
=item C<error> — the exception message
=item C<exception> — the original exception object
=item C<backtrace> — the full backtrace string when available
=item C<params> — the effect params, or the original event payload for
event-handler exceptions
=head2 Custom effects
Register your own with C<register-fx> for repeated side effects that
you want named:
=begin code :lang<raku>
$store.register-fx('log', -> $store, %params {
my $line = %params<line>;
$log-file.say("[{DateTime.now}] $line");
});
# Then use from any handler:
$store.register-handler('task/deleted', -> $st, %ev {
(
db => { tasks => %ev<remaining> },
log => { line => "deleted task {%ev<id>}" },
);
});
=end code
=head1 SUBSCRIPTIONS
Four kinds, chosen by what you want to do on change:
=item C<subscribe($id, @path, $widget)> — watch a state path. The widget is marked dirty when the value at that path changes. Simplest form.
=item C<subscribe-path-callback($id, @path, &callback, $widget)> — watch
a state path and call C<&callback($new-value)> when it changes.
Push-based like C<subscribe>; use this when the path itself is the
derived value and you need a side effect as well as a dirty mark.
=item C<subscribe-computed($id, &compute, $widget)> — watch a derived value. Runs C<&compute> every tick; marks the widget dirty when the return value changes.
=item C<subscribe-with-callback($id, &compute, &callback, $widget)> — same as computed, but also invokes C<&callback> with the new value. Use when the widget needs to be re-configured (e.g. C<set-items> on a list), not just re-rendered.
All three use C<$id> as a unique key — re-subscribing with the same id
overwrites the previous subscription. C<unsubscribe($id)> removes one
by key; C<unsubscribe-widget($w)> removes every subscription bound to
a widget (used during widget destruction, where it fires once per
widget of the torn-down subtree — it reads a widget-keyed reverse
index, so its cost tracks that widget's own subscriptions rather than
the size of the store).
=head2 Mutation safety
It's safe for a subscription's callback to call C<unsubscribe>,
C<unsubscribe-widget>, or any cascade thereof (closing a modal,
clearing a container, swapping a pane) while a tick is in progress.
C<unsubscribe> calls made from inside C<!check-subscriptions> or
C<!flush-push-subs> are queued into a pending set; the actual hash
deletions are applied after the walk exits, via a C<LEAVE> block
that runs whether the walk returns normally or unwinds via an
exception. During the walk, queued ids are skipped in the dispatch
loop, so a sub unsubscribed by an earlier callback in the same tick
does B<not> fire later in that same tick — matching the observable
behaviour callers depended on before the defer mechanism existed.
Re-subscribing the same id during a walk (C<unsubscribe('foo')>
followed by C<subscribe-with-callback('foo', ...)> in the same
callback) cancels the queued removal and tears down the old entry
cleanly before installing the new one, so the deferred flush never
clobbers the new registration.
Mutations from outside a walk — the common case, a handler running
in C<!process-queue> or app code calling C<unsubscribe> directly —
take effect immediately as before. The defer mechanism is invisible
to that path.
The invariant the walks maintain: C<%!subscriptions>,
C<%!push-subs-by-key>, and the by-widget reverse index are never
mutated while a walk is iterating them — all three are written from
the same C<!do-unsubscribe> chokepoint, so the deferred flush keeps
them consistent with each other rather than only with the queue.
Earlier implementations relied on C<:exists> guards plus
C<.keys.List> snapshots, but neither survived realistic load
(App::Cantina character changes cascade ~120 C<unsubscribe> calls
through one callback firing); the typed eager Array snapshot
(C<my Str:D @ids = %!subscriptions.keys>) plus the queued-mutation
discipline above is what actually holds.
=head2 Equality semantics — content digests
Change detection compares a B<content digest> of the watched value
against the previous one (see C<!value-digest> / C<!sub-changed>). The
digest folds Hashes (key-sorted) and Arrays structurally and keys most
leaves — value type or object — by C<.WHICH>. C<Blob>/C<Buf> values are
also leaves, but include only type, byte count, and C<.WHICH> so a parent
Hash containing image bytes never folds the bytes themselves. Comparing
two digests is then a flat string compare, which replaced a per-field
structural C<eqv> walk: on a chat C<Message> carrying ~1 K tokens of text
that is ~1000× cheaper, because C<eqv> pays reflective per-attribute
dispatch.
C<.WHICH> at a leaf cuts both ways correctly. For a B<value type> it is
content-derived (two equal C<Str>s share it), so a fingerprint C<Str>
recomputed each tick compares by content. For a B<reference> it is
identity: a wholesale-replaced record reads as changed, the same
instance as unchanged.
Object attributes are B<not> folded. The store holds more than plain
data records — e.g. the focused widget, whose attribute graph has
native plane handles (which surface as the NQP null) and
parent/child cycles that would explode a recursive fold. Identity is
the right comparison there, and for Cantina's immutable,
replace-wholesale records it is also exact (a content change is always
a fresh instance). The only cost is a no-op write that re-reads
identical rows into fresh objects firing one cheap re-render.
=head2 Mutable nested data and C<:identity-check-only>
The default ("structural") regime recomputes the digest on every fire.
That sidesteps the in-place-mutation trap: a subscription on path C<a>,
plus a caller doing C<$store.assoc-in('a', 'b', 'c', :value(42))>.
C<assoc-in> mutates a leaf inside the live nested Hash, so the Hash at
C<a> is the I<same> object before and after — a bare C<===> identity
check would match and silently suppress the fire. Re-digesting catches
it, because the mutated leaf's C<.WHICH> moved.
=begin code :lang<raku>
# Watch a mutable nested hash. The default just works — deep
# assoc-in writes under <prefs> are detected even though the prefs
# Hash reference itself is unchanged.
$store.subscribe('prefs-changed', <prefs>, $widget);
$store.assoc-in('prefs', 'theme', :value('dark')); # fires
$store.assoc-in('prefs', 'lang', :value('en')); # fires
=end code
Pass C<:identity-check-only> when the watched value is B<replaced
wholesale> on every real change — never mutated in place. It adds an
C<===> pre-check: an unchanged tick (the same instance / value as last
fire) short-circuits without computing a digest at all — the hot path
during streaming, where a watched Array keeps its identity across many
no-op ticks. On an identity change the digest is still computed, so an
equal-content replacement does not fire. Do B<not> use it on a value
you mutate in place (C<assoc-in> on a deep child) — the C<===> shortcut
would miss the write.
The flag exists on C<subscribe>, C<subscribe-path-callback>,
C<subscribe-computed>, and C<subscribe-with-callback>.
B<Migration from pre-0.8.0 Selkie>: the old C<:deep-equality-check>
flag is removed. Drop the flag — the safe behaviour is now the
default. Pass C<:identity-check-only> at any call site whose value is
replaced wholesale and that fires on a hot path.
=head1 TYPED STORE
The state tree is an untyped Hash by default — and stays that way
forever; the typed store is opt-in per store, not a migration target.
Passing a L<Selkie::Store::Schema> subclass at construction swaps the
Hash for a typed, immutable schema tree:
=begin code :lang<raku>
class MyApp::Db is Selkie::Store::Schema {
has Selkie::Store::Schema::UI $.ui .= new; # required framework slot
has Str $.user-name = '';
has %.prefs;
}
my $store = Selkie::Store.new(state-class => MyApp::Db);
# or, with a pre-built root (tests, restored sessions):
my $store2 = Selkie::Store.new(initial-state => MyApp::Db.new(user-name => 'Kai'));
=end code
What changes, and what doesn't:
=item B<Same effects, now validated.> Handlers keep returning C<db> /
C<db-replace> / C<db-delete> effects with the same shapes; keys and
paths are validated against the schema's slots at dispatch time,
and values are type-checked by the slot's constraint. A typo'd key
or wrong-typed value routes through C<__effect-error> — with the
offending C<event> named in the payload — instead of silently
malforming the tree.
=item B<C<db-update> is the typed workhorse.> C<< db-update => { fn
=> -> MyApp::Db $s { $s.with(user-name => 'Kai') } } >> transforms
the root through the schema's clone-with methods. Because clones
share unchanged branches, the store diffs old vs new root by
identity and notifies exactly the changed paths.
=item B<Reads: C<.state> or C<get-in>.> C<.state> returns the root
schema instance (typed accessors: C<$store.state.user-name>);
C<get-in> keeps working with slot names as path segments and keeps
its "missing step reads as Nil" contract. C<.db> dies on a typed
store — there is no state Hash to hand out.
=item B<Subscriptions are unchanged.> Path subscriptions address
slots the way they addressed Hash keys; computed subscriptions can
use typed accessors in their compute blocks. No signature changed.
Schema instances compare by identity in the digest (see "Equality
semantics") — exact, because no-op writes return the identical
instance.
=item B<Plain-Hash slots stay untyped territory.> A C<%.>-slot keeps
classic Hash semantics (deep-merge, auto-vivify, C<:delete>) inside
that region — handy for incremental migration.
=item B<The C<ui> slot is required.> The framework's focus state
lives at C<ui.focused-widget> / C<ui.focus-action>; the root schema
must carry a defined L<Selkie::Store::Schema::UI> (or subclass) as
C<ui>, and construction dies with the exact line to add if it
doesn't.
When to use which: a 200-line tool with a handful of state keys loses
nothing to the Hash store; define a schema when the state tree is big
enough that a typo'd path or a wrong-shaped write costs real
debugging time — the schema converts those from silent corruption
into loud dispatch-time errors. Full slot/write semantics live in
L<Selkie::Store::Schema>'s Pod.
=head1 SEE ALSO
=item L<Selkie::Widget> — widgets receive subscriptions and dispatch events
=item L<Selkie::App> — registers C<ui/focus>, C<ui/focus-next>, C<ui/focus-prev> handlers by default
=item L<Selkie::Store::Schema> — the typed state tree: slots, immutable writes, diffing
=item L<Selkie::Test::Store> — C<mock-store> accepts C<:state-class> / C<:initial-state>
=end pod
unit class Selkie::Store;
use Selkie::Widget;
use Selkie::Trace;
use Selkie::Store::Schema;
has %!db;
# --- Typed store (opt-in) ------------------------------------------
#
# When constructed with :state-class (a Selkie::Store::Schema
# subclass) or :initial-state (a Schema instance), state lives in
# $!schema-state instead of %!db, and every write path branches to
# the schema's immutable write methods (with / merge / set-at /
# deleted-at). Dirty paths come from the identity diff of old vs new
# root (Schema.changed-paths), feeding the same push-subscription
# machinery as the untyped store. With neither argument, nothing
# below is touched and every code path is identical to the untyped
# store. See "TYPED STORE" in the Pod.
has $.state-class;
has $!schema-state;
# Dispatch is callable from any thread — the async fx fires its
# on-success / on-failure dispatches from the worker, and apps push
# progress events from their own `start {}` pipelines. The queue is
# therefore lock-protected: `dispatch` pushes under $!queue-lock and
# !process-queue swaps the queue out under the same lock. Without it,
# a push landing between the drain's copy and clear is silently lost.
# The lock is held only for the push / swap — never while handlers
# run — so handlers re-dispatching (the `dispatch` fx, __effect-error)
# can't deadlock.
has @!event-queue;
has Lock $!queue-lock .= new;
has %!handlers; # event-name → Array[&handler]
has %!fx-handlers; # fx-name → &handler
has %!subscriptions; # sub-id → Hash{ path|compute, last-digest, last-raw, widget, deep }
has Bool $!subs-primed = False;
#|( Reverse index: C<$widget.WHICH> → SetHash of the sub-ids bound to
that widget. Exists so C<unsubscribe-widget> — which fires once per
widget of a torn-down subtree, i.e. hundreds of times per modal or
screen teardown — is a hash lookup instead of a grep over every
live subscription (measured at 38-40ms per dialog close in a
consumer app running ~850 live subs).
Keyed on C<.WHICH> rather than the widget itself so the index never
pins a destroyed widget alive, and because C<.WHICH> equality is
exactly the C<===> test the grep used — including for the
undefined-widget case, where subscriptions registered with a type
object all share one bucket.
Maintained in lockstep with C<%!subscriptions>: written by
C<!index-widget-sub> from every C<subscribe*> variant and cleared by
C<!do-unsubscribe>, which is the single point where an entry
actually leaves C<%!subscriptions> — including the deferred flush at
the end of a subscription walk. Updating it at the C<unsubscribe>
request site instead would desync the index from the hash for the
remainder of any walk. )
has %!subs-by-widget;
# --- Push-based path subscription infrastructure -------------------
#
# Path subscriptions (`subscribe` + `subscribe-path-callback`) don't
# get walked every tick. Instead, every mutation to the store records
# the affected paths in `@!dirty-paths` (`assoc-in`, `db`, `db-replace`,
# and `db-delete` all use this path). On tick, `!flush-push-subs` drains that
# set and fires the matching subscribers — those whose path is a
# prefix of a written path (ancestor notification: "something in my
# subtree changed") OR whose path has a written path as a prefix
# (descendant notification: "my subtree was replaced / written
# over"). Exact-match is the prefix-equal edge of either direction.
#
# After firing, each matched subscriber's current value is compared
# to its last-known value by content digest (see !sub-changed); the
# callback + widget-dirty only actually fire when the digest differs,
# so a no-op write doesn't spuriously fire subs.
#
# `computed` and `callback` subscriptions still pay the per-tick
# walk because their compute closures can depend on arbitrary state
# we can't cheaply index. The set of push-participating types is
# fixed: 'path', 'path-callback'. See `!check-subscriptions` for
# the pull split.
has @!dirty-paths; # List of path Lists written since last flush
has %!push-subs-by-key; # path-key → Array[sub-id]
# Walks (!check-subscriptions, !flush-push-subs) set this True
# while iterating. Any `unsubscribe` called during that window is
# captured in $!pending-unsubscribes and applied after the walk
# exits via the LEAVE block — never mid-iteration. See "Mutation
# safety" in the Pod above for the contract.
has Bool $!in-subscription-walk = False;
has SetHash $!pending-unsubscribes .= new;
# Ids whose subscribe-path-callback prime is currently on the call
# stack — the re-entrancy latch that turns a callback re-subscribing
# its own id (directly or via a cycle) from unbounded recursion into
# a logged no-op. See subscribe-path-callback.
has SetHash $!priming-subs .= new;
has IO::Handle $!debug-log;
has Bool $!debug-dispatches = False;
has Bool $!debug-effects = False;
has Bool $!debug-subscriptions = False;
# Async-effect tracking: every Promise produced by register-fx('async')
# is registered here so App.shutdown can drain in-flight work before
# tearing down notcurses. Without this the worker thread can complete
# after notcurses_stop and dispatch into handlers whose native deps
# are gone (SIGBUS on plane access, NPE on Nil $!nc).
has @!async-effects;
has Lock $!async-lock .= new;
has Bool $!shutting-down = False;
#|( Enable logging of dispatches, effects, and subscription fires. Output
lines go to C<$log> (defaults to C<$*ERR>). Pass
C<:!dispatches>, C<:!effects>, or C<:!subscriptions> to silence a
specific category. Overhead when disabled is a single Bool check per
hook. )
method enable-debug(
IO::Handle :$log = $*ERR,
Bool :$dispatches = True,
Bool :$effects = True,
Bool :$subscriptions = True,
) {
$!debug-log = $log;
$!debug-dispatches = $dispatches;
$!debug-effects = $effects;
$!debug-subscriptions = $subscriptions;
}
#| Turn off debug logging.
method disable-debug() {
$!debug-log = IO::Handle;
$!debug-dispatches = $!debug-effects = $!debug-subscriptions = False;
}
method !log-line(Str:D $line) {
return without $!debug-log;
my $ts = now.Rat.fmt('%.3f');
$!debug-log.say("[$ts] $line");
}
method !fmt-value($v --> Str) {
return '(undef)' without $v;
if $v ~~ Blob {
return "{$v.^name}(bytes={$v.bytes}, token={$v.WHICH.Str.substr(0, 16)})";
}
if $v ~~ Associative {
my @keys = $v.keys.sort;
my @parts;
for @keys.head(6) -> $k {
@parts.push: "$k=" ~ self!fmt-value($v{$k});
}
@parts.push('...') if @keys.elems > 6;
return '{' ~ @parts.join(', ') ~ '}';
}
if $v ~~ Positional {
my @vals = $v.list.head(6).map({ self!fmt-value($_) }).List;
@vals.push('...') if $v.elems > 6;
return '[' ~ @vals.join(', ') ~ ']';
}
try {
my $s = $v.gist;
$s = $s.substr(0, 80) ~ '…' if $s.chars > 80;
return $s;
}
'<unprintable>';
}
submethod TWEAK(:$initial-state) {
# Typed-store opt-in (see "TYPED STORE" in the Pod). Both spellings
# land in $!schema-state; :initial-state wins on the instance and
# back-fills $.state-class from its type.
if $initial-state.defined {
die "Store :initial-state must be a defined Selkie::Store::Schema instance, got {$initial-state.^name}"
unless $initial-state ~~ Selkie::Store::Schema:D;
if $!state-class ~~ Selkie::Store::Schema {
die "Store :initial-state ({$initial-state.^name}) is not an instance of :state-class ({$!state-class.^name})"
unless $initial-state ~~ $!state-class;
}
$!schema-state = $initial-state;
$!state-class = $initial-state.WHAT;
}
elsif $!state-class ~~ Selkie::Store::Schema {
die "Store :state-class must be the schema CLASS (a type object), not an instance — pass :initial-state for a pre-built root"
if $!state-class.defined;
$!schema-state = $!state-class.new;
}
self!validate-schema-root if $!schema-state.defined;
self.register-fx('db', -> $store, %params {
$store!deep-merge(%params);
});
self.register-fx('db-replace', -> $store, %params {
die "db-replace fx requires :value" unless %params<value>:exists;
$store!replace-in(
$store!effect-path(%params, 'db-replace'),
%params<value>,
);
});
self.register-fx('db-delete', -> $store, %params {
$store!delete-in($store!effect-path(%params, 'db-delete'));
});
self.register-fx('db-update', -> $store, %params {
die "db-update fx requires :fn (a Callable taking the current state and returning the new state)"
unless %params<fn> ~~ Callable;
$store!apply-db-update(%params<fn>);
});
self.register-fx('dispatch', -> $store, %params {
$store.dispatch(%params<event>, |%params.grep(*.key ne 'event').Hash);
});
self.register-fx('async', -> $store, %params {
my &work = %params<work> // die "async fx requires :work";
my $on-success = %params<on-success>;
my $on-failure = %params<on-failure>;
# Only the SCHEDULING decision checks shutting-down: once
# drain-async is in progress, no new workers get spawned.
# In-flight workers keep running so their on-success /
# on-failure dispatches can complete while the store is still
# alive (the test "drain waits, then delivers on-success"
# depends on this). Stale dispatches arriving after the App's
# run loop has fully exited are harmless — the event queue
# just isn't drained again. No `return` from pointy blocks
# — they aren't Routines (memory: raku_return_in_closures).
unless $store.shutting-down {
my $p = start {
CATCH {
default {
my $msg = .message;
my $bt = .?backtrace.?full.?Str // '';
$store!report-async-failure($msg, $bt);
if $on-failure {
$store.dispatch($on-failure,
error => $msg,
exception => $_,
backtrace => $bt,
);
}
}
}
my $result = work();
if $on-success {
$store.dispatch($on-success, result => $result);
}
};
$store!track-async-effect($p);
}
});
# Default handler for handler-exception events: log to the store's
# log so apps don't have to register one to get diagnostics. Apps
# can register additional handlers via `register-handler` to display
# toasts / modals / send telemetry.
self.register-handler('__effect-error', -> $store, %payload {
$store!log-line(
"[effect-error] {%payload<effect-name>}: {%payload<error>}\n"
~ (%payload<backtrace> // '')
);
();
});
}
# --- State access ---
#|( Access the raw state Hash. Read-only in practice — mutate via
dispatch. On a typed store there is no state Hash; this dies with
a pointer to C<state>. )
method db(--> Hash) {
die "This store holds typed schema state ({$!schema-state.^name}) — read it via .state (typed accessors) or get-in, not .db"
if $!schema-state.defined;
%!db
}
#|( The current state: the root schema instance on a typed store, the
state Hash otherwise. The uniform read entry point — C<.state>
works on both store kinds where C<.db> is Hash-only. Read-only in
practice — mutate via dispatch. )
method state() { $!schema-state // %!db }
#|( Deep-read a value at a path. Returns C<Nil> if any step in the path
is missing or not a Hash. Useful in handlers and subscription
computes:
my $name = $store.get-in('app', 'user', 'name'); # may be Nil
)
method get-in(*@path) {
return self!schema-get-in(@path) if $!schema-state.defined;
my $current = %!db;
for @path -> $key {
return Nil unless $current ~~ Associative;
return Nil unless $current{$key}:exists;
$current = $current{$key};
}
$current<>;
}
# get-in against the schema tree: schema nodes resolve segments as
# slots (unknown slot → Nil, keeping get-in's documented Hash-parity
# "missing step reads as Nil" contract — the STRICT read path is the
# schema's own accessors via .state); plain-Hash regions resolve as
# keys exactly like the untyped store.
method !schema-get-in(@path) {
my $current = $!schema-state;
for @path -> $key {
if $current ~~ Selkie::Store::Schema {
return Nil unless $current.has-slot($key.Str);
$current = $current.slot-value($key.Str);
}
elsif $current ~~ Associative {
return Nil unless $current{$key}:exists;
$current = $current{$key};
}
else {
return Nil;
}
}
$current<>;
}
#|( Deep-write a value at a path, auto-creating intermediate Hashes as
needed. B<Prefer dispatch-and-handle over calling this directly from
app code> — direct mutations bypass the auditability that the
dispatch pattern gives you. This is public for legitimate framework
use (e.g. App's internal focus-action flag).
$store.assoc-in('app', 'user', 'name', value => 'Alice');
)
method assoc-in(*@path, :$value!) {
die "Path must not be empty" unless @path;
if $!schema-state.defined {
self!swap-schema-state($!schema-state.set-at(@path.List, $value));
return;
}
my $target = %!db;
for @path[0 ..^ @path.end] -> $key {
$target{$key} = {} unless $target{$key} ~~ Associative;
$target = $target{$key};
}
$target{@path[*-1]} = $value;
self!mark-path-dirty(@path.List);
}
# --- Events ---
#|( Enqueue an event for processing on the next tick. The event name is
routed to every handler registered for that name; their returned
effects are applied in order.
Safe to call from any thread: the queue is lock-protected, so
worker threads (async fx callbacks, app-owned C<start> blocks)
can dispatch concurrently with the main loop's drain without
losing events. Off-thread dispatches are picked up on the next
main-loop tick. )
method dispatch(Str:D $event, *%payload) {
$!queue-lock.protect: {
@!event-queue.push({ :$event, :%payload });
}
}
#|( Register a handler for an event name. Multiple handlers per event
are supported — all are called and their effects merged.
The handler signature is C<sub ($store, %payload --> @effects)>.
Return a single effect Pair, a list of Pairs, or a Hash of effects.
Return an empty list or C<()> to apply no effects. )
method register-handler(Str:D $event, &handler) {
%!handlers{$event} = [] unless %!handlers{$event}:exists;
%!handlers{$event}.push(&handler);
}
# --- Effects ---
#|( Register a custom effect handler. The handler receives the store and
a Hash of params, and performs side effects. See L<EFFECTS> above
for the built-in ones and an example of registering your own. )
method register-fx(Str:D $fx-name, &handler) {
%!fx-handlers{$fx-name} = &handler;
}
# --- Subscriptions ---
#|( Canonical string encoding of a path for use as a Hash key in the
push-subscription reverse index. We join segments with C<\0>
(NUL) because it's guaranteed never to appear in realistic path
keys — all our keys are user-chosen Str. The decode helper is
the inverse. Empty path encodes to the empty string. )
our sub path-key(@path --> Str) {
@path.map(*.Str).join("\0");
}
our sub key-path(Str $key --> List) {
return ().List unless $key.chars;
$key.split("\0").List;
}
#|( Subscribe a widget to a state path. When the value at the path
changes, the widget is marked dirty on the next tick. The C<$id>
identifies the subscription for later C<unsubscribe> — use a unique
name per subscription.
Push-based: the store pushes a notification to this subscription
only when a write touches the path (exact, ancestor, or descendant —
see the push-sub dispatch block at the top of the file). Idle
ticks with no writes do zero work per subscription. Initial prime
(marking the widget dirty so its first render happens) runs
synchronously here.
Change detection defaults to a content digest (see B<Equality
semantics> in this module's Pod) — correct for mutable containers
(Hash / Array) updated in place by C<assoc-in> on a deep child,
where the ancestor's reference is unchanged across the write. The
historical default was identity-only (C<===>), which silently
suppressed those writes; opt back into an C<===> pre-check via
C<:identity-check-only> for paths whose value is replaced wholesale
on each write.
See B<MUTABLE NESTED DATA> in this module's Pod. )
method subscribe(Str:D $id, @path, Selkie::Widget $widget,
Bool :$identity-check-only = False) {
# Re-subscribing an id that's queued for deferred removal: tear
# down the old entry cleanly *now* and drop the queue marker, so
# the end-of-walk flush doesn't clobber the new registration.
if $!pending-unsubscribes{$id} {
self!do-unsubscribe($id);
$!pending-unsubscribes.unset($id);
}
self!index-widget-sub($id, $widget);
%!subscriptions{$id} = {
type => 'path',
path => @path.List,
widget => $widget,
deep => !$identity-check-only,
};
self!index-push-sub($id, @path);
# Prime synchronously: mark the widget dirty so it renders with
# its initial bound value without needing a synthetic event.
$widget.mark-dirty if $widget.defined;
# Prime the change-detection cache (the value's digest) so the first
# actual write only fires if it represents a real change.
self!sub-changed(%!subscriptions{$id}, self.get-in(|@path));
}
#|( Subscribe to a state path with a callback — fires C<&callback($new-value)>
on any real change to the path (exact, ancestor write, or descendant
write that replaced the subtree). No compute function needed: the
path IS the watched expression. Also marks the owning widget dirty
so it re-renders after the callback configures it.
Use when your widget needs reconfiguration on change (e.g. C<set-items>
on a list, C<set-text> on a label) rather than just re-rendering the
same computed output. Equivalent to C<subscribe-with-callback> with
a trivial compute closure, but without the per-tick closure
invocation cost — push-based like C<subscribe>.
Change detection is digest-based, mirroring C<subscribe>. Pass
C<:identity-check-only> when the watched path's value is replaced
wholesale per write. The callback always receives the live current
value; only the change-gate computes the digest.
B<The prime calls your callback synchronously, before this method
returns.> Do not call C<subscribe-path-callback> again from inside
that callback for an id that is currently priming — a callback that
re-arms its own subscription on some condition (a cache miss, an
empty payload) would otherwise recurse without bound. The store
guards this: a prime re-entered for an id still inside its own
prime (directly or through a cycle of subscriptions) registers the
subscription but skips the nested prime, and logs the skip when
debug logging is enabled. )
method subscribe-path-callback(Str:D $id, @path, &callback, Selkie::Widget $widget,
Bool :$identity-check-only = False) {
if $!pending-unsubscribes{$id} {
self!do-unsubscribe($id);
$!pending-unsubscribes.unset($id);
}
self!index-widget-sub($id, $widget);
%!subscriptions{$id} = {
type => 'path-callback',
path => @path.List,
callback => &callback,
widget => $widget,
deep => !$identity-check-only,
};
self!index-push-sub($id, @path);
# Prime: fire the callback with the current value so the widget
# gets configured immediately. Mark dirty too — callback may or
# may not trigger a render indirectly, but the prime render is
# always desired.
#
# Re-entrancy latch: the prime runs the caller's callback
# synchronously, and a callback that re-subscribes its own id (or
# completes a cycle back to one still priming) would recurse
# without bound — observed as a silent app-wide freeze, since the
# recursion never returns to the tick that hosts it. The
# subscription above is fully registered either way; only the
# nested prime is skipped.
if $!priming-subs{$id} {
self!log-line("subscribe-path-callback('$id') re-entered its own "
~ "prime — nested prime skipped; do not re-subscribe from "
~ "inside a prime callback");
return;
}
$!priming-subs.set($id);
# LEAVE fires on EVERY method exit — including the early return
# above, which lexically precedes it. Guard the unset so a
# latch-hit exit can't clear the entry the OUTER prime set; only
# the frame that set the latch releases it.
my Bool $latched = True;
LEAVE $!priming-subs.unset($id) if $latched;
my $current = self.get-in(|@path);
self!sub-changed(%!subscriptions{$id}, $current); # prime change cache
$widget.mark-dirty if $widget.defined;
&callback($current);
}
method !index-push-sub(Str:D $id, @path) {
my $key = path-key(@path);
%!push-subs-by-key{$key} = [] unless %!push-subs-by-key{$key}:exists;
%!push-subs-by-key{$key}.push: $id;
}
method !unindex-push-sub(Str:D $id, @path) {
my $key = path-key(@path);
return unless %!push-subs-by-key{$key}:exists;
%!push-subs-by-key{$key} = %!push-subs-by-key{$key}.grep(* ne $id).Array;
%!push-subs-by-key{$key}:delete unless %!push-subs-by-key{$key}.elems;
}
#|( Subscribe a widget to a computed value. C<&compute> receives the
store each tick and should return the value. The widget is marked
dirty when the return value changes.
Change detection compares a content digest of the result (see
B<Equality semantics> in this module's Pod). This correctly handles
compute functions that return references into mutable nested state —
the common case of C<$store.get-in('chat', 'messages')> after an
C<assoc-in> deep-write, where identity-only comparison sees the same
Hash reference and silently suppresses the fire.
Set C<:identity-check-only> when C<&compute>'s result is replaced
wholesale per change (a fresh Str, a derived count, a freshly-built
Hash) — this adds an C<===> pre-check that skips the digest on
unchanged ticks. )
method subscribe-computed(Str:D $id, &compute, Selkie::Widget $widget,
Bool :$identity-check-only = False) {
if $!pending-unsubscribes{$id} {
self!do-unsubscribe($id);
$!pending-unsubscribes.unset($id);
}
self!index-widget-sub($id, $widget);
%!subscriptions{$id} = {
type => 'computed',
compute => &compute,
widget => $widget,
deep => !$identity-check-only,
};
$!subs-primed = False;
}
#|( Like C<subscribe-computed>, but also invokes C<&callback> with the
new value whenever it changes. Use this when your widget needs to
be re-configured (e.g. C<set-items> on a list, C<set-text> on a
label), not just re-rendered.
Change detection is digest-based; see C<subscribe-computed>. Pass
C<:identity-check-only> when the compute result is replaced
wholesale per change. The callback always receives the live
compute result. )
method subscribe-with-callback(Str:D $id, &compute, &callback, Selkie::Widget $widget,
Bool :$identity-check-only = False) {
if $!pending-unsubscribes{$id} {
self!do-unsubscribe($id);
$!pending-unsubscribes.unset($id);
}
self!index-widget-sub($id, $widget);
%!subscriptions{$id} = {
type => 'callback',
compute => &compute,
callback => &callback,
widget => $widget,
deep => !$identity-check-only,
};
$!subs-primed = False;
}
#|( Remove a subscription by its id. No-op if the id isn't registered.
During a subscription walk (C<!check-subscriptions> /
C<!flush-push-subs>), the actual removal is deferred to walk
exit; the queued id is skipped in the dispatch loop, so an
earlier callback's unsubscribe is observed in the same tick.
Outside a walk, removal is immediate as before. )
method unsubscribe(Str:D $id) {
if $!in-subscription-walk {
$!pending-unsubscribes.set($id);
return;
}
self!do-unsubscribe($id);
}
# Actual hash mutation — caller has already established that this
# is safe (either outside a walk, or via !flush-pending-unsubscribes
# at walk exit).
method !do-unsubscribe(Str:D $id) {
my %sub = %!subscriptions{$id} // return;
if (%sub<type> // '') eq any('path', 'path-callback') {
self!unindex-push-sub($id, %sub<path>);
}
self!unindex-widget-sub($id, %sub<widget>);
%!subscriptions{$id}:delete;
}
#|( Bind C<$id> to C<$widget> in the reverse index, first releasing
whatever widget the id was previously bound to. Re-subscribing an
existing id to a different widget is legal (it overwrites the
C<%!subscriptions> entry), so without the release step the old
widget's bucket would keep a stale id and C<unsubscribe-widget>
would tear down a subscription that no longer belongs to it. )
method !index-widget-sub(Str:D $id, $widget) {
with %!subscriptions{$id} -> %existing {
self!unindex-widget-sub($id, %existing<widget>);
}
my $key = $widget.WHICH;
%!subs-by-widget{$key} = SetHash.new unless %!subs-by-widget{$key}:exists;
%!subs-by-widget{$key}.set($id);
}
method !unindex-widget-sub(Str:D $id, $widget) {
my $key = $widget.WHICH;
return unless %!subs-by-widget{$key}:exists;
%!subs-by-widget{$key}.unset($id);
%!subs-by-widget{$key}:delete unless %!subs-by-widget{$key}.elems;
}
# Drain the pending-unsubscribe set, applying each deferred removal.
# Called from the LEAVE block of every walk wrapper.
method !flush-pending-unsubscribes() {
return unless $!pending-unsubscribes.elems;
my @to-flush = $!pending-unsubscribes.keys.List;
$!pending-unsubscribes = SetHash.new;
self!do-unsubscribe($_) for @to-flush;
}
#|( Remove every subscription bound to a widget. Called automatically by
C<Selkie::Container> when a child is removed — you rarely call this
yourself.
O(subs on this widget), not O(all subs), via the C<%!subs-by-widget>
reverse index; a widget that never subscribed has no bucket and
costs one failed hash lookup. Removal itself goes through
C<unsubscribe>, so calling this from inside a subscription callback
defers exactly like a direct C<unsubscribe> does — the bucket keeps
the ids until the walk's flush actually removes them, which makes a
repeat call in the same walk a harmless no-op rather than a
double-removal. )
method unsubscribe-widget(Selkie::Widget $widget) {
my $key = $widget.WHICH;
return unless %!subs-by-widget{$key}:exists;
# Snapshot: !do-unsubscribe mutates (and can delete) the bucket.
my Str:D @to-remove = %!subs-by-widget{$key}.keys;
self.unsubscribe($_) for @to-remove;
}
#| Total number of active subscriptions. Excludes ids that have been
#| queued for deferred removal but not yet flushed (so counts reflect
#| the post-walk state). Primarily useful for tests asserting that
#| destroy / unsubscribe-widget cleanup ran.
method subscription-count(--> Int) {
%!subscriptions.elems - $!pending-unsubscribes.elems
}
# --- Frame tick ---
#|( Process one tick of the store. Drains the event queue (invoking
handlers, applying effects, possibly enqueuing more events — capped
at 100 iterations to prevent infinite loops), then walks every
subscription and compares its current value against the previous
one.
The subscription walk is skipped on ticks where the event queue was
empty I<and> every subscription has already been primed with its
initial value. With nothing new in the store, no subscription value
can have changed — walking them would be wasted work. Registering
a new subscription flips the prime flag so the next tick initializes
it regardless of event activity.
Called automatically by C<Selkie::App.run> each frame. Call
explicitly only when you need to force state resolution outside
the main loop (e.g. bootstrapping state before C<run> starts). )
method tick(--> Bool) {
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.tick', cat => 'store',
args => %(subscriptions => %!subscriptions.elems))
!! Nil;
my $had-events = self!process-queue;
# Push-based path subs: fire only subscribers whose paths
# overlap with writes that happened during this tick (or any
# writes left over from pre-tick bootstrap). Zero cost on idle
# ticks where the store wasn't written.
my $had-writes = @!dirty-paths.elems > 0;
self!flush-push-subs if $had-writes;
# Pull-based computed/callback subs: still walked every tick
# that had events (or was the first tick after a new sub was
# added and hasn't been primed yet).
if $had-events || !$!subs-primed {
self!check-subscriptions;
$!subs-primed = True;
}
# Signal "activity happened this tick" back to the event loop so
# it can keep the tick rate hot instead of falling to the idle
# ladder. Priming is internal and doesn't count as real activity.
my Bool $activity = ($had-events || $had-writes).Bool;
with $span {
.finish(
had_events => $had-events,
had_writes => $had-writes,
activity => $activity,
);
}
$activity;
}
#|( Drain C<@!dirty-paths> and fire every push subscription whose
bound path overlaps with any written path — where "overlaps"
means either path is a prefix of the other (ancestor / descendant
/ exact). Dedupe: a sub only fires once per flush even if
multiple writes match it. After dispatching, the dirty-path set
is cleared. Firing still respects value-change semantics:
C<!sub-changed> gates whether C<&callback> / C<mark-dirty>
actually runs, so no-op writes don't produce spurious fires. )
method !flush-push-subs() {
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.subscriptions.push-flush', cat => 'store',
args => %(dirty_paths => @!dirty-paths.elems))
!! Nil;
my Int $fired = 0;
# Same defer wrapper as !check-subscriptions — see "Mutation
# safety" in the module Pod. Path-callbacks fired from here can
# call unsubscribe / unsubscribe-widget; those calls queue and
# apply at walk exit.
$!in-subscription-walk = True;
LEAVE {
$!in-subscription-walk = False;
self!flush-pending-unsubscribes;
}
my @writes = @!dirty-paths;
@!dirty-paths = ();
my %to-fire; # sub-id → True (dedupe within a flush)
for @writes -> @write {
# Ancestor side: walk every prefix of @write and look up
# exact-path subscribers at that prefix depth. Includes the
# empty prefix (watch-everything root sub).
for 0 .. @write.elems -> $len {
my $prefix-key = path-key(@write[^$len]);
next unless %!push-subs-by-key{$prefix-key}:exists;
%to-fire{$_} = True for %!push-subs-by-key{$prefix-key}.list;
}
# Descendant side: any sub whose bound path has @write as
# a prefix (and is strictly longer, so we don't double-count
# the exact match captured by the ancestor loop above).
my $write-prefix = path-key(@write);
for %!push-subs-by-key.kv -> $key, @sub-ids {
next if $key eq $write-prefix; # exact covered by ancestor loop
next unless $key.starts-with(
$write-prefix eq '' ?? '' !! $write-prefix ~ "\0"
);
%to-fire{$_} = True for @sub-ids.list;
}
}
for %to-fire.keys -> $sub-id {
next if $!pending-unsubscribes{$sub-id};
next unless %!subscriptions{$sub-id}:exists;
my %sub := %!subscriptions{$sub-id};
my $current = self.get-in(|%sub<path>);
next unless self!sub-changed(%sub, $current);
if $!debug-subscriptions {
self!log-line(" push-sub[$sub-id] fired: " ~ self!fmt-value($current));
}
%sub<widget>.mark-dirty if %sub<widget>.defined;
if (%sub<type> // '') eq 'path-callback' && %sub<callback>.defined {
%sub<callback>($current);
}
$fired++;
}
with $span {
.finish(candidate_subs => %to-fire.elems, fired => $fired);
}
}
method !process-queue(--> Bool) {
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.process-queue', cat => 'store')
!! Nil;
my $max-iterations = 100;
my $iteration = 0;
my Bool $had-events = False;
my Int $event-count = 0;
my Int $handler-count = 0;
while $iteration++ < $max-iterations {
# Swap the queue out under the lock so a concurrent off-thread
# dispatch can't land between the copy and the clear (it would
# be wiped). Handlers run outside the lock — they may dispatch.
my @batch = $!queue-lock.protect: {
my @b = @!event-queue;
@!event-queue = ();
@b
};
last unless @batch;
$had-events = True;
$event-count += @batch.elems;
for @batch -> %entry {
my $event = %entry<event>;
my %payload = %entry<payload>;
if $!debug-dispatches {
my $payload-str = %payload.elems
?? %payload.kv.map(-> $k, $v { "$k=" ~ self!fmt-value($v) }).join(' ')
!! '';
self!log-line("dispatch $event $payload-str");
}
my @handlers = |(%!handlers{$event} // []);
if $!debug-dispatches && @handlers == 0 {
self!log-line(" (no handler)");
}
for @handlers -> &handler {
$handler-count++;
# Event handlers get the same isolation as effect
# handlers — a buggy `register-handler` callback
# shouldn't kill the dispatch loop and crash the TUI.
# Re-entrance guard skips the route for failures
# inside the __effect-error chain itself.
my @effects;
{
my $handler-span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.handler', cat => 'store',
args => %(event => $event, handlers => @handlers.elems))
!! Nil;
CATCH {
default {
self!log-line("[event-handler-error] $event: {.message}")
if $!debug-dispatches;
unless $event eq '__effect-error' {
self.dispatch('__effect-error',
effect-name => "event-handler[$event]",
event => $event,
error => .message,
exception => $_,
backtrace => .backtrace.full.Str,
params => %payload,
);
}
}
}
@effects = handler(self, %payload);
with $handler-span {
.finish(effects => @effects.elems);
}
}
self!apply-effects(@effects, :event($event)) if @effects;
}
}
}
with $span {
.finish(
iterations => $iteration - 1,
events => $event-count,
handlers => $handler-count,
);
}
$had-events;
}
method !apply-effects(@effects, Str :$event) {
for @effects -> $fx {
next unless $fx ~~ Pair | Associative;
if $fx ~~ Pair {
self!run-effect($fx.key, $fx.value, :$event);
} elsif $fx ~~ Associative {
for $fx.kv -> $fx-name, $fx-params {
self!run-effect($fx-name, $fx-params, :$event);
}
}
}
}
method !run-effect(Str:D $fx-name, $fx-params, Str :$event) {
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.effect', cat => 'store',
args => %(name => $fx-name, event => $event // ''))
!! Nil;
if $!debug-effects {
self!log-line(" → $fx-name: " ~ self!fmt-value($fx-params));
}
my $handler = %!fx-handlers{$fx-name};
unless $handler ~~ Callable {
self!log-line(" (unknown effect '$fx-name')") if $!debug-effects;
with $span {
.finish(known => False);
}
return;
}
{
# Isolate handler exceptions AND payload-shape violations: a
# buggy effect handler — or a handler that passes the wrong
# payload shape upstream — shouldn't tear out of the dispatch
# loop and bring down the whole app. The failure is logged
# and routed back into the event queue as `__effect-error`
# so apps can register a handler that surfaces failures
# (toast, modal, telemetry) instead of debugging through
# stderr behind notcurses's alt-screen.
CATCH {
default {
self!log-line("[effect-error] $fx-name: {.message}")
if $!debug-effects;
# Re-entrance guard: if the failing effect IS the
# error-event handler chain, drop the failure to avoid
# infinite recursion. We've already logged it.
unless $fx-name eq '__effect-error' {
self.dispatch('__effect-error',
effect-name => $fx-name,
event => ($event // ''),
error => .message,
exception => $_,
backtrace => .backtrace.full.Str,
params => $fx-params,
);
}
}
}
# Effect payloads must be Associative. The dispatcher used to
# wrap bare scalars as `{ value => $x }` automatically — but
# that silently rewrote the shape so handlers got payloads
# they didn't expect. Now we throw with a clear migration
# message. The throw is caught above and routed to
# __effect-error rather than killing the dispatch loop.
unless $fx-params ~~ Associative {
die "Effect '$fx-name' payload must be Associative, got "
~ "{$fx-params.^name}. Wrap it: \{ value => \$x \}, "
~ "or pass an empty Hash for no payload.";
}
$handler(self, $fx-params);
}
with $span {
.finish(known => True);
}
}
method !check-subscriptions() {
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.subscriptions.pull-check', cat => 'store',
args => %(subscriptions => %!subscriptions.elems))
!! Nil;
my Int $checked = 0;
my Int $fired = 0;
# See "Mutation safety" in the module Pod. The flag tells
# `unsubscribe` to queue removals; the LEAVE block runs whether
# we exit normally or by exception, so we never wedge the store.
$!in-subscription-walk = True;
LEAVE {
$!in-subscription-walk = False;
self!flush-pending-unsubscribes;
}
# Typed eager Array snapshot: assignment to `Str:D @ids` throws
# X::TypeCheck::Assignment loudly if any key isn't a defined Str
# (instead of corrupting an unrelated downstream `:exists` call).
# The Array binding is an explicit container slot that the spesh
# optimiser cannot fuse away — `.keys.List` was insufficient in
# practice for the same purpose.
my Str:D @ids = %!subscriptions.keys;
for @ids -> $id {
next if $!pending-unsubscribes{$id};
next unless %!subscriptions{$id}:exists;
my %sub := %!subscriptions{$id};
# Push-handled types bypass the per-tick walk entirely —
# they fire from C<!flush-push-subs> on actual writes.
next if (%sub<type> // '') eq any('path', 'path-callback');
$checked++;
my $current;
{
my $compute-span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.subscription.compute', cat => 'store',
args => %(id => $id, type => (%sub<type> // '')))
!! Nil;
$current = do given %sub<type> {
when 'computed' | 'callback' { %sub<compute>(self) }
};
with $compute-span {
.finish;
}
}
if self!sub-changed(%sub, $current) {
$fired++;
if $!debug-subscriptions {
self!log-line(" sub[$id] fired: " ~ self!fmt-value($current));
}
%sub<widget>.mark-dirty if %sub<widget>.defined;
if %sub<type> eq 'callback' && %sub<callback>.defined {
%sub<callback>($current);
}
}
}
with $span {
.finish(checked => $checked, fired => $fired);
}
}
# Hard ceiling on digest recursion so a cyclic or pathologically deep
# store value can never hang the change-detection walk.
my constant DIGEST-MAX-DEPTH = 40;
#|( Content digest of a subscription value, used for change detection.
Folds Hashes (canonical, key-sorted) and Arrays structurally, and
keys most leaves — value type OR object — by C<.WHICH>. C<Blob>/C<Buf>
leaves are compact type/byte-count/C<.WHICH> tokens so large image
binaries are never traversed byte-by-byte through a parent structure.
The result is a single Str, so comparing two digests is a flat string
compare instead of C<eqv>'s reflective per-field walk — ~1000× cheaper
on a rich object like a chat C<Message> carrying ~1 K tokens of text.
C<.WHICH> at a leaf does the right thing both ways: for a value type
it is content-derived (two equal C<Str>s share it, so a fingerprint
C<Str> recomputed each tick compares by content), and for a reference
it is identity (a wholesale-replaced record reads as changed; the
same instance reads as unchanged).
We deliberately do NOT fold an object's attributes. The store holds
not only plain data records but references like the focused widget,
whose attribute graph contains native plane handles (which read back
as the NQP null and can't be passed on) and parent<->child cycles
(which would explode the walk). Identity is the correct comparison
for those, and for an immutable, replace-wholesale record it is also
exact — a content change is always a fresh instance. The only cost is
that a no-op write re-reading identical rows into fresh objects fires
one cheap re-render. C<DIGEST-MAX-DEPTH> bounds a pathologically deep
(or self-referential) Hash / Array. )
method !value-digest($v, Int $depth = 0 --> Str) {
return '∞' if $depth > DIGEST-MAX-DEPTH;
# Every undefined value digests to one token. The previous comparison
# treated all undefined as equal (its `!a.defined && !b.defined` short-
# circuit), and `get-in` legitimately returns different undefined
# sentinels for the same logical "no value" — Nil when an intermediate
# is missing, the Any leaf when only the final key is absent — so a
# type-tagged marker would spuriously fire on an unrelated sibling write.
return 'U' unless $v.defined;
if $v ~~ Associative {
return '{' ~ $v.keys.sort.map(-> $k {
$k ~ '=' ~ self!value-digest($v{$k}, $depth + 1)
}).join(';') ~ '}';
}
if $v ~~ Blob {
# Blob/Buf can be megabytes of image data. Treat them as a leaf
# before the Positional branch so subscription digests never walk
# every byte of a parent Hash/Array that happens to contain one.
return 'B(' ~ $v.^name ~ ':' ~ $v.bytes ~ ':' ~ $v.WHICH.Str ~ ')';
}
if $v ~~ Positional {
return '[' ~ $v.list.map({ self!value-digest($_, $depth + 1) }).join(';') ~ ']';
}
# Any other leaf — value type OR object — by its C<.WHICH>:
# content-derived for value types (two equal Strs share it, so a
# fingerprint Str recomputed per tick compares by content), and
# identity for references (a wholesale-replaced record reads as
# changed; the same instance reads as unchanged).
#
# We deliberately do NOT fold an object's attributes. The store holds
# not just plain data records but things like the focused Selkie
# widget, whose attribute graph has native plane handles (which read
# back as the NQP null and can't even be passed on) and parent<->child
# cycles (which would explode the walk). Identity is the right
# comparison for those, and for Cantina's immutable, replace-wholesale
# records identity is also exact (a content change is always a new
# instance). The cost is a no-op reload that re-reads identical rows
# into fresh objects firing one cheap re-render — rare and bounded.
$v.WHICH.Str;
}
#|( Decide whether a subscription's watched value changed since its last
fire, updating the subscription's cached comparison state in place.
Returns True when the caller should fire (mark dirty / run callback).
Two regimes, keyed on C<%sub<deep>>:
=item B<Structural> (the default): the value may be mutated in place
(e.g. C<assoc-in> into a nested Hash), so we always recompute the
content digest and compare. A deep mutation that leaves the
container's identity intact is still caught — the mutated leaf's
C<.WHICH> changes the digest.
=item B<Replace-only> (C<:identity-check-only>): the value is
guaranteed to be replaced wholesale on every real change, never
mutated in place. An C<===> pre-check then short-circuits unchanged
ticks (same instance / same value) without computing a digest at
all — the hot path during streaming, where the watched Array keeps
its identity. On an identity change we still digest, so an
equal-content replacement (a no-op reload re-reading the same rows
into fresh objects) does NOT spuriously fire.
This replaces the old snapshot-and-C<eqv> machinery with the same
observable behaviour but a flat C<.WHICH>-token string compare. )
method !sub-changed(%sub, $current --> Bool) {
my $primed = %sub<last-digest>:exists;
unless %sub<deep> {
# Short-circuit BEFORE any span: this is the per-sub per-tick
# hot path (identity-check-only subs during streaming), and a
# span here recorded nothing but its own overhead — two events
# per sub per event-tick in the 2026-07 capture.
if $primed && $current === %sub<last-raw> {
return False;
}
%sub<last-raw> = $current;
}
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.subscription.digest', cat => 'store',
args => %(type => (%sub<type> // ''), deep => (%sub<deep> // False)))
!! Nil;
my $digest = self!value-digest($current);
my $changed = !$primed || $digest ne %sub<last-digest>;
%sub<last-digest> = $digest;
with $span {
.finish(changed => $changed);
}
$changed;
}
method !deep-merge(%updates) {
if $!schema-state.defined {
self!swap-schema-state($!schema-state.merge(%updates));
return;
}
self!merge-into(%!db, %updates, ());
}
method !effect-path(%params, Str:D $fx-name --> List) {
die "$fx-name fx requires :path" unless %params<path>:exists;
my $path = %params<path>;
my @path = $path ~~ Positional ?? $path.list !! ($path,);
die "$fx-name fx path must not be empty" unless @path.elems;
@path.List;
}
method !replace-in(@path, $value) {
if $!schema-state.defined {
self!swap-schema-state($!schema-state.set-at(@path, $value));
return;
}
my $target = %!db;
for @path[0 ..^ @path.end] -> $key {
$target{$key} = {} unless $target{$key} ~~ Associative;
$target = $target{$key};
}
$target{@path[*-1]} = $value;
self!mark-path-dirty(@path.List);
}
method !delete-in(@path) {
if $!schema-state.defined {
self!swap-schema-state($!schema-state.deleted-at(@path));
return;
}
my $target = %!db;
for @path[0 ..^ @path.end] -> $key {
return unless $target ~~ Associative;
return unless $target{$key}:exists;
return unless $target{$key} ~~ Associative;
$target = $target{$key};
}
return unless $target ~~ Associative;
return unless $target{@path[*-1]}:exists;
$target{@path[*-1]}:delete;
self!mark-path-dirty(@path.List);
}
method !merge-into(%target, %source, @path-so-far) {
for %source.kv -> $key, $value {
my @here = (|@path-so-far, $key);
if $value ~~ Associative && %target{$key} ~~ Associative {
self!merge-into(%target{$key}, $value, @here);
} else {
%target{$key} = $value;
# Mark this exact leaf path as dirty. The push-sub flush
# will also notify any subscriber whose path is a prefix
# (ancestor) via its prefix walk, so marking every
# intermediate level would produce duplicate fires.
self!mark-path-dirty(@here.List);
}
}
}
method !mark-path-dirty(@path) {
@!dirty-paths.push: @path.List;
}
# --- Typed store internals -----------------------------------------
# Root-schema contract check at construction: the framework's focus
# machinery writes ui.focused-widget / ui.focus-action, so a typed
# store without a conforming `ui` slot would fail at first Tab press
# instead of at Store.new. Fail early, with the line to add.
method !validate-schema-root() {
my $root = $!schema-state;
unless $root.has-slot('ui')
&& $root.slot-value('ui') ~~ Selkie::Store::Schema::UI:D {
die "The root schema {$root.^name} must declare a defined 'ui' slot holding a Selkie::Store::Schema::UI (or subclass) instance — the framework's focus state lives there. Add:\n"
~ " has Selkie::Store::Schema::UI \$.ui .= new;";
}
}
# The single write sink for the typed store: every schema write path
# (deep-merge, replace, delete, assoc-in, db-update) produces a new
# root and lands here. Identity-equal root means a no-op write —
# nothing marked, no subscriber ever fires. Otherwise the identity
# diff yields exactly the changed paths for the push-sub flush.
method !swap-schema-state($new) {
my $old = $!schema-state;
return if $new === $old;
$!schema-state = $new;
self!mark-path-dirty($_) for $new.changed-paths($old);
}
method !apply-db-update(&fn) {
if $!schema-state.defined {
my $new = fn($!schema-state);
die "db-update fn must return a {$!state-class.^name} instance, got {$new.^name}"
unless $new ~~ Selkie::Store::Schema:D && $new ~~ $!state-class;
self!swap-schema-state($new);
return;
}
my $new = fn(%!db);
die "db-update fn must return an Associative state, got {$new.^name}"
unless $new ~~ Associative;
%!db = %$new unless $new === %!db;
# Root-path dirty: the push-sub flush's prefix matching notifies
# every path subscription, and the digest gate stops the no-op
# fires. Blunt by design on the untyped store — the schema store
# gets precise diffs from !swap-schema-state instead.
self!mark-path-dirty(().List);
}
# --- Async effect tracking ---
#| True once C<drain-async> has flipped the store into shutdown mode.
#| Callers (typically the `async` fx) can poll this to short-circuit
#| dispatches that would race against a tearing-down App.
method shutting-down(--> Bool) { $!shutting-down }
method !track-async-effect(Promise $p) {
$!async-lock.protect: { @!async-effects.push: $p };
}
#|( Wait for every in-flight async-effect worker to complete (or for the
timeout to elapse), flip the store into shutdown mode so any further
`async` dispatches no-op, and clear the tracking list. Called from
C<Selkie::App.shutdown> before notcurses is torn down so completing
workers can't dispatch into handlers whose native deps are gone.
Idempotent — a second call is a fast no-op (the list is empty and
`$!shutting-down` is already True). )
method drain-async(:$timeout = 5) {
$!shutting-down = True;
my @snapshot;
$!async-lock.protect: {
@snapshot = @!async-effects;
@!async-effects = ();
}
return unless @snapshot;
# Spanned: this await (bounded by :$timeout) is part of app
# shutdown — without the span, a slow quit can't be attributed
# between effect draining and the app's own teardown.
my $span = Selkie::Trace.enabled
?? Selkie::Trace.start('store.drain-async', cat => 'store',
args => %(pending => @snapshot.elems, timeout_s => $timeout.Num))
!! Nil;
await Promise.anyof(Promise.allof(@snapshot), Promise.in($timeout));
$span.finish(
drained => ?(@snapshot.grep({ .status !~~ Planned }).elems == @snapshot.elems),
) with $span;
}
method !report-async-failure(Str $message, Str $backtrace) {
self!log-line("[async-effect] $message\n$backtrace") if $!debug-effects;
}