An Ash resource extension that makes a resource a node in a reactive DAG.
The resource IS the node and its own payload table: its reactive block
defines the computation, its attributes are the rows the node materializes.
This is the intended shape — one resource, both roles.
defmodule MyApp.FlowMonth do
use Ash.Resource,
domain: MyApp.Domain,
data_layer: AshPostgres.DataLayer, # the node's OWN payload table
extensions: [ReactiveDag.Node]
attributes do # the payload columns; the row
attribute :plant, :string, primary_key?: true # IS its identity — the
attribute :month, :string, primary_key?: true # cell key "north|2024-01"
attribute :avg_flow, :float # is its serialization
end
actions do
create :upsert do upsert?(true); accept([:plant, :month, :avg_flow]) end
end
reactive do
op :fold
# ASH-FIRST: `recompute_by` names the UNIT a change invalidates (and
# supplies the edge + the claim rule); the library reads :dmr_rows,
# folds each group, and upserts the row by its Ash IDENTITY with the
# coordination Op.put. No `read:`, no `key:`, no key column, no
# `upsert:` — each slot has an escape hatch when the shape outgrows
# attributes.
recompute_by :plant, to: :dmr_rows, from: :plant
reduce group_by: [:plant, :month],
into: [avg: [flow: :avg_flow]]
end
endThe library closes the payload loop: a reduce/join whose into returns a
row, with no upsert:, has that row written into the node's own resource
(ReactiveDag.Node.Payload) with change-detection. Writing into a different
resource is the explicit deviation — supply a custom upsert: for that.
Keys work like Ash keys. A SINGLE-attribute primary key is the payload key
(derived — declare payload_key only for a non-PK key column); a COMPOSITE
primary key means the row IS its identity: no key column at all, the upsert
conflicts on the primary key, and the cell key is the identity's
serialization in primary-key order ("gf|2025"). The payload_action
upsert defaults to :upsert.
Which computation? (the Ash-first ladder)
Start from what Ash expresses declaratively; each step outward trades declarativeness for power:
| you want to… | use | rows into BEAM? | you write |
|---|---|---|---|
group + avg/sum/count a relationship | aggregate | none (datastore GROUP BY) | attribute atoms |
| fold, with the recompute UNIT declared | recompute_by + reduce | the scoped slice | the unit + the field it comes from + an into: fold |
| fold one input's rows into per-group summaries | reduce | the scoped slice of over | group_by: attrs + an into: fold |
| reconcile one input's two sides by key | join | the scoped slice of over | side attrs/[key:, where:] + picks |
| a slot the attributes can't express | the slot's escape | same | query: (shape the Ash read), fn group/key/into, expand:, status: |
| arbitrary recompute, kept Ash-native | run :action | up to the action | a generic action on THIS resource |
| recompute beyond Ash (LLM, fetch, bespoke) | compute Mod | up to the module | a ReactiveDag.Op |
Node shapes (what scaffolding a node needs)
| shape | data_layer | attributes | actions | reactive |
|---|---|---|---|---|
| payload (materializes typed rows) | AshPostgres/Ets | the payload columns | an :upsert action | a combinator, no upsert: |
verdict (verdict? true) | Ash.DataLayer.Simple | none | none | a combinator; rows carry :status |
| write-elsewhere | Simple | none | none | a combinator + a custom upsert: |
| escape hatch | Simple | none | none | compute Mod |
A payload node's into row is written into the resource itself (the payload
loop, ReactiveDag.Node.Payload); the cell key maps to the payload_key
attribute (default :key) via the payload_action upsert (default :upsert).
Cell ids (the vocabulary of every edge)
A node's cell id defaults to the resource module's short name, snake-cased
(MyApp.FlowMonth → :flow_month); set id: to override. This id is what
every edge names — depends_on [:flow_month], ref :flow_month, and the ids
passed to graph/2. If an edge doesn't resolve, it's almost always an id that
doesn't match a node's (defaulted or explicit) id.
Assembling + running
ReactiveDag.Node.graph/2 builds a ReactiveDag.Plan from a list of node
resources; the substrate reads only the reactive block. Then:
plan = ReactiveDag.Node.graph([FlowMonth, FiscalLines, …], for_each: &fetch/1)
ReactiveDag.Drain.run(plan,
recompute: ReactiveDag.Node.Recompute,
key_rule: ReactiveDag.Node.KeyRule)Config
config :reactive_dag,
repo: MyApp.Repo, # REQUIRED (raises if unset)
tuple_table: "my_tuple", # coordination spine table (must match your migration)
dirty_table: "my_dirty", # frontier table (must match your migration)
coordination_writer: MyApp.Writer # optional; a spine-only default shipstuple_table/dirty_table default silently, so a name that doesn't match your
migration yields empty results with no error — set them explicitly.
Summary
Functions
The cell id for a node resource (explicit id, else the module's snake short-name).
The ReactiveDag.Cells a node resource lowers to (no graph math): its root
cell + one per nested compose. Legs are lowered by-name via the shared
ReactiveDag.Lowering.walk — a ref/dep resolves to an existing cell id
(no new cell), a compose recurses into an intermediate cell.
The reserved id of the attested view a gate: interposes over over —
<over>@<gate> for the blocking mode, <over>@<gate>~annotate for the
non-blocking one (two projections → two cells).
Assemble a ReactiveDag.Plan from a list of node resources. Each resource
contributes its root cell PLUS an intermediate cell per nested compose leg
(lowered through ReactiveDag.Lowering.walk). The union is validated and
depth-ordered by ReactiveDag.Graph.build/1.
The root cell a NON-generator node resource lowers to.
Functions
The cell id for a node resource (explicit id, else the module's snake short-name).
@spec cells(module(), (atom() -> [map()]) | nil) :: [ReactiveDag.Cell.t()]
The ReactiveDag.Cells a node resource lowers to (no graph math): its root
cell + one per nested compose. Legs are lowered by-name via the shared
ReactiveDag.Lowering.walk — a ref/dep resolves to an existing cell id
(no new cell), a compose recurses into an intermediate cell.
The reserved id of the attested view a gate: interposes over over —
<over>@<gate> for the blocking mode, <over>@<gate>~annotate for the
non-blocking one (two projections → two cells).
@spec graph( [module()], keyword() ) :: ReactiveDag.Plan.t()
Assemble a ReactiveDag.Plan from a list of node resources. Each resource
contributes its root cell PLUS an intermediate cell per nested compose leg
(lowered through ReactiveDag.Lowering.walk). The union is validated and
depth-ordered by ReactiveDag.Graph.build/1.
Pass :for_each to expand GENERATOR nodes: a (population_atom -> [member])
fun. A node with for_each: :pop builds no template cell — instead, for each
member it builds an instance sub-tree rooted at <id>.<member.id>, with the
member's meta merged onto every instance cell (the host's per-member stamp,
e.g. a probe filter). A member is any map with an :id (+ optional :meta).
Without a fetcher, a generator node is skipped (and logged by the caller).
@spec to_cell(module()) :: ReactiveDag.Cell.t()
The root cell a NON-generator node resource lowers to.