Stream trigger (WebSocket)

Start a workflow from messages on an external WebSocket feed — exchange prices, order books, liquidation feeds, oracle vendor feeds, bridge status, internal event streams. Each JSON message is filtered by an expression and, if it matches, becomes an ordinary rflow run.

workflows:
  price-spread-alert:
    trigger:
      stream:
        websocket:
          url: ${BINANCE_WS_URL}          # ws:// or wss:// (a ${VAR} placeholder is fine)
          subscribe:
            method: SUBSCRIBE
            params: ["ethusdt@ticker"]
            id: 1
          message_json_path: "quot;          # extract the payload object (default $)
          where: "${{ message.s == 'ETHUSDT' and wei(message.c, 8) > wei('3500', 8) }}"
          idempotency_key: "${{ message.E }}"
    steps:
      - notify:
          channel: ops
          message: "ETH at ${{ trigger.args.c }}"

The `message` root

Every frame is parsed as JSON (v1 is JSON-only) and narrowed by message_json_path (default $, the whole frame; supports $.a.b and $.a[0]). The result is the message root — the only root the where and idempotency_key expressions can see.

a frame that is not JSON, or where the path selects nothing, is dropped with a warning (never a crash)
where false fires nothing; omit where to fire on every message
wei(message.c, 8) parses a decimal price string to an exact integer for a precise numeric comparison (minijinja has no float coercion)

The run payload is { args: <message>, stream: { url_host, received_at } }, so steps address the message as trigger.args.*.

Exactly-once under reconnect

`idempotency_key`	dedupe key	reconnect behaviour
set (e.g. `${{ message.E }}`)	`stream:{workflow}:{sha256(key)}` via the webhook-idempotency gate	a replayed message inside `idempotency_ttl` (default 7d; 1m–365d) creates no new run
absent	`stream:{workflow}:{uuid}` per message	a reconnect that re-delivers messages can duplicate — documented at-least-once

Set idempotency_key to a stable id the feed provides (an event time, a sequence number, a trade id). A feed without one cannot be exactly-once — drop the key and accept at-least-once, or dedupe downstream.

The key must be a pure function of message. A non-deterministic expression (e.g. one calling now()) would render a different key for every delivery, so the hash never collides and dedupe silently does nothing — rflow validate and boot reject it rather than let you ship a broken exactly-once.

Resilience

field	default	behaviour
connect	15s timeout	a failed/slow connect retries with backoff
`reconnect`	enabled, 1s→60s	exponential backoff on any drop (reset after a healthy session); `enabled: false` exits on the first drop
`heartbeat`	off	send a WebSocket ping every interval and treat it as a liveness probe — if no frame or pong arrives for a few intervals the connection is silently dead (half-open) and is reconnected, instead of the read loop parking on a corpse
`max_message_bytes`	1 MiB	an oversize frame is dropped with a warning (a hard 32 MiB frame closes the connection)
`throttle`	off	same window/cooldown as `trigger.event.throttle` — caps a chatty feed

The read loop never blocks on the journal: matched messages go onto a bounded queue drained by a claim worker, so a slow database sheds load (a warning) rather than stalling reads (which would starve pongs and drop the socket). The stream task is owned by the running stack and shut down cleanly — no leaked task or socket on shutdown or a --watch reload. The closing handshake is itself bounded, so a backpressured or half-open peer cannot wedge teardown.

Redaction

The URL may embed a token (in its query or userinfo). Only the host is ever journaled or logged (stream.url_host) — the full URL, and any token in it, never lands in a payload or a log line.

Rehearse offline

rflow test <workflow> --fixture <file> builds the trigger payload from a fixture that is the extracted message object (bare, or wrapped as { "message": {...} }), so you can exercise the where filter and the steps without touching the internet.