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.

rflow_version: 1
name: vault-inflow-guard
 
config:
  port: 3940
  db_connection: ${DATABASE_URL}
 
constants:
  vault: "0x1f9090aaE28b8a3dCeaDf281B0F12828e676c326"
 
networks:
  - name: ethereum
    chain_id: 1
    rpc: ${ETH_RPC}
 
contracts:
  # index_events fills rflow_idx_usdc.transfer without firing any workflow
  USDC:
    abi: ./abis/usdc.json
    addresses:
      ethereum: "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
    index_events: [Transfer]
 
workflows:
  vault-inflow-guard:
    trigger:
      interval:
        every: 5m
    steps:

      - id: gate
        query:
          sql: >-
            SELECT COALESCE(SUM(value::numeric), 0)
            FROM 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

Which tables exist? Ask rflow tables

Don't guess table names. rflow tables prints the live catalog: every column with its type, plus row counts:

$ rflow tables
table                          columns (name type)                              rows
rflow_idx_usdc.transfer        rindexer_id int4, contract_address bpchar,
                               from bpchar, to bpchar, value varchar,
                               tx_hash bpchar, block_number numeric, ...        3214
rflow_wf_copy_trade.transfer   (same shape — one table per triggered event)      112
rflow.workflow_runs            id uuid, workflow_name text, trigger_key ...       87

With no database reachable it prints the table names the config implies. rflow validate --preflight checks that every table referenced in query: SQL (steps and triggers) exists, warning (never failing) with the closest existing table when one doesn't.

The naming rule is deterministic:

Workflow/contract names are lowercased with hyphens → underscores; event names are snake-cased (PoolCreated → pool_created).

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_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 }}"]).