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
    attribute :key, :string, primary_key?: true
    attribute :plant, :string
    attribute :avg_flow, :float
  end

  actions do
    create :upsert do upsert?(true); upsert_identity(:key); accept([:key, :plant, :avg_flow]) end
  end

  reactive do
    op :fold
    key_rule :all
    # `into` returns the row; the LIBRARY writes it into THIS resource
    # (keyed by :key) and does the coordination Op.put. No `upsert:` needed.
    reduce over: :dmr_rows,
           read: &MyApp.FlowMonth.read/1,
           group_by: &MyApp.FlowMonth.group/1,
           key: &MyApp.FlowMonth.key/1,
           into: &MyApp.FlowMonth.into/2
  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.

The cell key maps to the resource's payload_key attribute (default :key) via the payload_action upsert (default :upsert); set those in the reactive block if they're named otherwise.

Which computation? (reduce / join / aggregate / compute)

Rule of thumb: aggregate when the fold is a datastore aggregate over a relationship (pushdown, no rows in memory); reduce for any other in-BEAM fold; compute when no combinator fits.

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.