`postgres://`¶

PostgreSQL via jackc/pgx/v5 (using pgxpool). Provides a LISTEN/NOTIFY consumer and shares the sql:// component's SQL surface (pg can also use sql://?driver=pgx&dsn=…).

URI shape¶

postgres://[<table>[/<operation>]]?dsn=postgres://user:pass@host:5432/db[&channel=…]

operation ∈ {query, insert, update, delete, execute, copy, notify}.

The copy operation bulk-loads data using the PostgreSQL COPY protocol (COPY … FROM STDIN) — the in-process equivalent of psql \copy. It streams a file or CSV body straight to the server in constant memory, with no per-row SQL, so GB-scale loads run at set-based speed instead of paying the interpreted per-row cost of the insert path (which trips the starlang eval cap at scale). See Bulk COPY loader below.

LISTEN/NOTIFY consumer engages when either: - ?channel=<name> is set (works with any path), OR - the URI shape is postgres://<channel>/notify (path-driven).

The producer's notify operation issues pg_notify(channel, payload) with the In body as the payload.

Parameters¶

Param	Type	Default	Description
`dsn`	string	required	pgx-style connection string (`postgres://user:pass@host:port/db?sslmode=…`).
`channel`	string	—	Consumer: LISTEN channel name. When set, the consumer subscribes via LISTEN/NOTIFY instead of polling. Producer: target channel for the `notify` op.
`pollInterval`	duration	`30s`	Consumer (polling): interval between polls.
`pollSince`	string	—	Consumer (polling): column used as a watermark for change-data-capture polling.
`file`	string	—	Producer (`copy`): stream this local file into `COPY … FROM STDIN` instead of the exchange body. Constant memory — never materialised in Helix.
`columns`	string	—	Producer (`copy`): comma-separated column list. Raw-stream: order must match the CSV; positional row bodies: names the target columns. Omit (raw-stream) to load the table's columns in definition order.
`format`	string	`csv`	Producer (`copy`): `csv` \| `text` \| `binary`.
`header`	bool	`false`	Producer (`copy`): when true, the first CSV line is the header row and is skipped.
`delimiter`	string	`,`	Producer (`copy`): field delimiter (csv/text).
`null`	string	—	Producer (`copy`): the string that represents SQL NULL (csv/text).
`quote`	string	`"`	Producer (`copy`, csv): quote character.
`escape`	string	quote char	Producer (`copy`, csv): escape character.

The SQL producer surface (CRUD ops plus outputType, :#name placeholders, batch, transacted) is shared with the sql:// component. To opt into that surface, either:

Use sql://...?driver=pgx&dsn=… (cross-driver, pgx via database/sql), or
Stay on postgres:// for the pgx-native pool and CRUD ops.

Header conventions¶

LISTEN/NOTIFY consumer writes:

Header	Value
`HelixPgChannel`	The notify channel name.
`HelixPgNotifyPID`	Backend PID that sent the NOTIFY.
`HelixPgPayload`	The notify payload (also set as the body).
`postgres.channel` / `postgres.notifyPid`	Legacy aliases (also written).

postgres://...?notify producer writes:

Header	Value
`postgres.channel`	The target channel.

Body shape¶

LISTEN/NOTIFY consumer: body is the payload string.
notify producer: body must be a string (or anything stringifiable via fmt.Sprintf("%v", body)).
CRUD producer: same as sql:// — map[string]any for insert/update, {query, params} for execute, etc.
copy producer: raw CSV bytes/string body (or a file= path) for raw-stream mode; a row slice ([]map[string]any, [][]any, [][]string) for row mode. Writes postgres.copyRows / postgres.rowsAffected = the number of rows loaded.

Examples¶

LISTEN consumer¶

trigger:
  uri: postgres://?dsn=postgres://app:secret@db:5432/app&channel=alerts
steps:
  - log:
      message: "Got alert: ${body} on ${headers.HelixPgChannel}"

NOTIFY producer¶

- to: postgres://alerts/notify?dsn=postgres://app:secret@db:5432/app
  # body becomes the payload of NOTIFY alerts, 'whatever-this-is'

CRUD insert¶

- to: postgres://users/insert?dsn=postgres://app:secret@db:5432/app
  # body: {"name": "alice", "email": "a@example.com"}

CDC polling consumer (Helix legacy)¶

trigger:
  uri: postgres://events/query?dsn=…&pollInterval=2s&pollSince=created_at
steps:
  - process:
      ref: handle-event

Bulk COPY loader¶

The copy op is the fast path for loading large files — the in-process psql \copy. Point it at a file and it streams straight into the table (≈1 M rows / 75 MB in ~2 s at constant memory; Postgres parses the CSV server-side, so multiline quoted fields are handled natively):

- to: "postgres://staging.invoices/copy?dsn=${PG_DSN}&file=/data/invoices_2026.csv&header=true"

The historical-backfill pattern — COPY into a staging table, then one set-based INSERT … SELECT — keeps the whole job server-side (no starlang, no 30 s eval cap, minutes for GB):

workflows:
  - id: backfill-invoices
    trigger: { uri: "timer://once?repeatCount=1" }
    steps:
      # 1. Stream the raw CSV into a staging table (constant memory).
      - to: "postgres://staging.invoices_raw/copy?dsn=${PG_DSN}&file=/data/hist/invoices_2026.csv&header=true"
      # 2. One set-based transform into the target (client filter, dedup, keys…).
      #    Set the exchange body to {query, params} first, then:
      - to: "postgres://billing_iq_uat.invoice_details/execute?dsn=${PG_DSN}"

You can also feed copy from a streaming csv:// source in row mode — each csv://?stream=true&batch=N batch is a []map[string]any that copy loads via pgx.CopyFrom:

trigger: { uri: "csv:///data/big.csv?stream=true&batch=50000" }
steps:
  - to: "postgres://staging.rows/copy?dsn=${PG_DSN}&columns=invoice_id,client_id,amount"

postgres://¶