Query trigger

Poll a read-only SQL aggregate over the indexed event tables on an interval and fire when a condition is met ("page me when 24h volume crosses X", "act when cumulative deposits pass a threshold") without an external cron + API.

rflow_version: 1
name: volume-alarm
 
config:
  port: 3940
  db_connection: ${DATABASE_URL}
 
networks:
  - name: ethereum
    chain_id: 1
    rpc: ${ETH_RPC}
 
contracts:
  USDC:
    abi: ./abis/erc20.json
    addresses:
      ethereum: "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
    # rows land in rflow_idx_usdc.transfer without firing any workflow
    index_events: [Transfer]
 
notifications:
  channels:
    ops:
      telegram:
        bot_token: ${TG_BOT_TOKEN}
        chat_id: ${TG_CHAT_ID}
 
workflows:
  volume-alarm:
    trigger:
      query: 
        sql: >-
          SELECT COALESCE(SUM(value::numeric), 0)
          FROM rflow_idx_usdc.transfer
          WHERE block_timestamp > now() - interval '1 hour'
        every: 1m
        condition: "${{ output > wei('10000000', 6) }}"
        mode: threshold          # threshold (default) | changed 
    steps:
      - id: alert
        notify:
          channel: ops
          message: "hourly USDC volume crossed 10M: ${{ format_units(trigger.args.output, 6) }}"

Which tables can the sql: see? Run rflow tables — it lists every queryable table with its columns (name + type) and row counts:

$ rflow tables
table                     columns (name type)                              rows
rflow_idx_usdc.transfer   rindexer_id int4, from bpchar, to bpchar,
                          value varchar, block_timestamp timestamptz, ...  3214
rflow.workflow_runs       id uuid, workflow_name text, ...                   87

rflow validate --preflight additionally warns when a query: statement references a table the database does not have (advisory, never a failure — indexed tables only exist once rflow start has booted the indexer).

Field	Required	Description
`sql`	✅	ONE read-only statement (`SELECT` / `WITH … SELECT`), with the same rules and table naming as the `query:` step
`args`		`$1..$N` values, rendered once at boot: only `constants` / `secrets` / `lists` roots (no trigger context exists yet; validated)
`every`	✅	Poll interval, e.g. `1m`
`condition`	✅	Firing condition over the shaped `output` (alias `result`)
`mode`	`threshold`	`threshold` \| `changed` — identical semantics to the read trigger

The shaped query output (scalar / object / row array, shaped exactly like the query: step) is what condition: sees; NUMERIC values compare as exact numbers, so U256-scale thresholds are precise.

Firing modes — edge-triggered, never spam

threshold (default) — fire on the false→true crossing of condition:. A condition that stays true fires once, not every poll.
changed — fire when the shaped output differs from the previous poll's while the condition holds (the first observation only records a baseline).

The previous poll's state persists per workflow (rflow.read_trigger_state), so a restart mid-"true" does not re-fire a threshold trigger. A transient query failure (database hiccup, table not created yet) logs, keeps the previous state and retries next poll.

The trigger context

Path	Description
`trigger.args.output`	The shaped query output at fire time
`trigger.query.observed_at`	Observation timestamp (ISO)

Exactly-once

Each fire claims query:{workflow}:{poll instant}: one run per polling instant, and the persisted edge state carries the dedupe across restarts.

Honest notes

${{ }} inside sql: is a hard validation error: dynamic values belong in args: (see the injection guard).
The statement runs with the default 5s statement_timeout inside a READ ONLY transaction on rflow's own Postgres. Keep polled aggregates indexed/cheap, or widen every:.
The trigger only sees what the project indexes: event-trigger tables (rflow_wf_<workflow>.<event>) and contracts.<name>.index_events tables (rflow_idx_<contract>.<event>, plus the rflow.* journal).