query

Read-only SQL over the project's indexed event tables (and the rflow journal) — decisions over history, not just the triggering event. With assert: it becomes a hard gate; without, it's enrichment for later steps.

- id: gate
  query:
    sql: >-
      SELECT COALESCE(SUM(value::numeric), 0)
      FROM rflow_indexer_rflow_idx_usdc.transfer
      WHERE lower("to") = lower($1)
        AND block_timestamp > now() - interval '1 day'
    args: ["${{ constants.vault }}"]
    assert: "${{ output < wei('1000000', 6) }}"   # daily inflow cap

Fields

Where the data lives — the table naming rule

rflow's embedded indexer writes decoded event rows to Postgres with fully deterministic names:

Workflow/contract names are lowercased with hyphens → underscores; event names are snake-cased (PoolCreated → pool_created). Don't memorize it — rflow tables prints the live catalog with columns and row counts:

$ rflow tables
table                                        columns                              rows
rflow_indexer_rflow_idx_usdc.transfer        rindexer_id, contract_address, ...   3214
rflow.workflow_runs                          id, workflow_name, trigger_key, ...    87

Every event table carries the decoded params as columns plus the indexer's injected columns: contract_address, tx_hash, block_number, block_timestamp, block_hash, network, tx_index, log_index.

Column types & U256 exactness

The indexer's column mapping keeps big integers exact:

• ints up to 128 bits → NUMERIC
• larger ints — including uint256 amounts — → VARCHAR(78) holding the exact decimal string. Aggregate or compare through a cast: SUM(value::numeric), value::numeric > $1
• addresses → CHAR(42) lowercase hex — compare with lower("to") = lower($1) and quote the reserved column names ("from", "to")

NUMERIC results decode as exact decimal strings and cross into assert:/condition: expressions as exact numbers — U256 comparisons like output >= wei('5', 18) are precise, never floats. See U256 semantics.

Output shaping

The result set becomes steps.<id>.output:

• 1 row × 1 column → the scalar itself (the common aggregate case)
• 1 row × N columns → an object keyed by column name
• N rows → an array of objects, capped at 1000 rows — over the cap the value becomes { "rows": [...], "truncated": true } and a warning logs. Aggregate in SQL instead of fetching raw rows.

Injection guard

${{ }} inside sql: is a hard validation error. Dynamic values go through args:, which bind as $1..$N parameters — never spliced into the SQL text. On top of the textual read-only gate (single statement, no data-modifying keywords, WITH d AS (DELETE …) rejected), every query runs inside a Postgres READ ONLY transaction, so writes are refused by the database itself.

# ✗ rflow validate error — templating inside sql:
sql: "SELECT sum(value) FROM t WHERE \"to\" = '${{ trigger.args.to }}'"
 
# ✓ bind it
sql: SELECT sum(value::numeric) FROM rflow_indexer_rflow_idx_usdc.transfer WHERE "to" = $1
args: ["${{ trigger.args.to }}"]

assert — the gate

A false assert: fails the step with kind assert_failed — a deliberate stop, never retried; the run follows on_failure. A fired timeout maps to the timeout failure kind; other query failures are data_unavailable.

To fire a workflow FROM an aggregate instead of gating inside one, see the query trigger.

Timing honesty

Rows appear in the event tables when the indexer processes the block — milliseconds behind head in steady state, further behind during backfill. A query: step in a workflow triggered by the very event it aggregates may or may not see that event's own row; anchor thresholds so off-by-one-event does not matter, or aggregate over explicitly closed windows (block_number <= $2 with args: [..., "${{ trigger.block_number - 1 }}"]).