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:

Node shapes (what scaffolding a node needs)

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