ReactiveDag.Node (reactive_dag v0.17.0-rc)

Copy Markdown View Source

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
end

The 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…userows into BEAM?you write
group + avg/sum/count a relationshipaggregatenone (datastore GROUP BY)attribute atoms
fold, with the recompute UNIT declaredrecompute_by + reducethe scoped slicethe unit + the field it comes from + an into: fold
fold one input's rows into per-group summariesreducethe scoped slice of overgroup_by: attrs + an into: fold
reconcile one input's two sides by keyjointhe scoped slice of overside attrs/[key:, where:] + picks
a slot the attributes can't expressthe slot's escapesamequery: (shape the Ash read), fn group/key/into, expand:, status:
arbitrary recompute, kept Ash-nativerun :actionup to the actiona generic action on THIS resource
recompute beyond Ash (LLM, fetch, bespoke)compute Modup to the modulea ReactiveDag.Op

Node shapes (what scaffolding a node needs)

shapedata_layerattributesactionsreactive
payload (materializes typed rows)AshPostgres/Etsthe payload columnsan :upsert actiona combinator, no upsert:
verdict (verdict? true)Ash.DataLayer.Simplenonenonea combinator; rows carry :status
write-elsewhereSimplenonenonea combinator + a custom upsert:
escape hatchSimplenonenonecompute 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 namesdepends_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 ships

tuple_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

cell_id(resource)

@spec cell_id(module()) :: atom()

The cell id for a node resource (explicit id, else the module's snake short-name).

cells(resource, fetch \\ nil)

@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.

gated_id(over, gate, mode \\ :require)

@spec gated_id(String.t(), atom(), :require | :annotate) :: String.t()

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).

graph(resources, opts \\ [])

@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).

reactive(body)

(macro)

to_cell(resource)

@spec to_cell(module()) :: ReactiveDag.Cell.t()

The root cell a NON-generator node resource lowers to.