PgFlow: Elixir vs Supabase (TypeScript/Deno)

Both implementations share the same base PostgreSQL schema and SQL orchestration layer. They can run side-by-side against the same database.

Architecture

Flow Definition

TypeScript — fluent builder with method chaining:

const flow = new Flow<{ orderId: number }>({ slug: 'process_order' })
  .step({ slug: 'validate' }, (input) => ({ valid: true }))
  .step({ slug: 'charge', dependsOn: ['validate'] }, (deps) => ({ charged: true }))
  .map({ slug: 'notify' }, (item) => ({ sent: true }))

Elixir — macro DSL:

defmodule ProcessOrder do
  use PgFlow.Flow
  @flow queue: :process_order, max_attempts: 3

  step :validate do
    fn input, _ctx -> %{valid: true} end
  end

  step :charge, depends_on: [:validate] do
    fn deps, _ctx -> %{charged: true} end
  end

  map :notify do
    fn item, _ctx -> %{sent: true} end
  end
end

Both compile to identical SQL: pgflow.create_flow() + pgflow.add_step().

Step Types

Step/Flow Options

The Elixir DSL always writes its defaults into the flow definition. A flow defined in both languages without explicit options therefore stores different opt_max_attempts, opt_base_delay, and opt_timeout values depending on which runtime compiled it first. Set these options explicitly on flows that both runtimes serve.

Conditional Pattern Matching

Both implementations use PostgreSQL's @> (JSON contains) operator for pattern matching:

• Root steps - Pattern matches against the flow's input map
• Dependent steps - Pattern matches against %{dep_slug => output, ...}, a map built from dependencies' outputs

Example (Elixir):

step :charge, depends_on: [:validate], if: %{"validate" => %{"valid" => true}} do
  fn deps, _ctx -> %{charged: true} end
end

When :validate runs and returns %{"valid" => true, "id" => 123}, the :charge step runs because its input map %{"validate" => %{"valid" => true, "id" => 123}} contains the pattern %{"validate" => %{"valid" => true}}.

Non-cascade-skipped dependencies are omitted from dependent inputs: if :validate is skipped (not cascade), :charge receives %{} instead of %{"validate" => nil}.

Handler Context

Both lazy-load flow input to avoid unnecessary DB queries for non-root steps.

Worker Model

TypeScript: Edge Worker loops

Edge Function instance or Node/Deno process
  → EdgeWorker.start(flow, config)   (once per process)
  → register worker → loop: read_with_poll → start_tasks → handlers → heartbeat
  → stops on shutdown signal, CPU/wall-clock limit, or deprecated_at
  → ensure_workers() on pg_cron pings Edge Functions to start replacements
  → horizontal scaling = more instances/processes

Elixir: Long-Running OTP GenServers

PgFlow.Supervisor (rest_for_one)
├── Task.Supervisor
├── Signal.Notify (when signal_strategy: :notify)
├── WorkerSupervisor (dynamic)
│   └── Worker.Server (one per flow/job)
├── StalledTaskRecovery
└── :flow_starter (one-shot registration)

Worker configuration

Both runtimes configure workers per process, not in the database. Nothing below changes the pgflow schema, persisted flow definitions, queue names, or message format, so TypeScript and Elixir workers can use different values on the same flow.

PgFlow.scale_worker/2 changes the worker on the calling node only and keeps the new limits in memory. A crash or supervisor restart returns to the values from config, worker_options, and start_worker/2. Scale-down never interrupts running handlers. Neither runtime coordinates concurrency through the database: the effective concurrency of a flow is the sum over every Elixir node and every TypeScript process serving it.

PgFlow.start_worker/2 accepts only :repo, :max_concurrency, and :batch_size, and raises ArgumentError on any other key. worker_options keys must also appear in :flows or :jobs. scale_worker/2 and check_worker/2 also take :timeout (default 5000 ms) for the call to the worker.

Polling Protocol

Both use the same two-phase protocol:

1. Reserve messages from pgmq (makes them invisible)
2. pgflow.start_tasks() — atomically claim tasks, return details
3. Execute handler
4. pgflow.complete_task() or pgflow.fail_task()

Difference in phase 1:

TypeScript does the waiting inside PostgreSQL: one read_with_poll call holds a connection and re-checks the queue every pollIntervalMs (100ms for flow workers) for up to maxPollSeconds (2s for flow workers, 5s for queue workers). Elixir workers schedule their own polls with Process.send_after, so pgmq.read() returns at once and the connection goes back to the Ecto pool. Backoff timers are cancellable, so a NOTIFY or a task completion wakes the worker immediately.

Note that read_with_poll is not truly push-based — PostgreSQL runs an internal poll loop at pollIntervalMs (100ms for flow workers, 200ms for queue workers) inside the blocking call:

Deno worker                    PostgreSQL
   |                              |
   |---- read_with_poll() ------->|
   |     (connection held)        |-- SELECT FROM q_* ... nothing
   |                              |-- wait pollIntervalMs (100ms)
   |                              |-- SELECT FROM q_* ... nothing
   |                              |-- wait pollIntervalMs (100ms)
   |                              |-- SELECT FROM q_* ... MESSAGE!
   |<--- [message] ---------------|

The real question is where the poll loop lives and what tradeoffs that creates:

Queue Notifications

Upstream pgflow never uses LISTEN/NOTIFY. It creates no triggers on pgmq queue tables and does not call pgmq.enable_notify_insert. LISTEN/NOTIFY is an Elixir-only signal strategy (signal_strategy: :notify) and needs pgmq 1.8 or newer, the first release with enable_notify_insert.

In :notify mode, PgFlow.FlowStarter enables notifications for each flow queue it serves at boot. pgmq's enable_notify_insert/2 creates the trigger_notify_queue_insert_listeners constraint trigger on pgmq.q_<queue> as DEFERRABLE INITIALLY IMMEDIATE. That trigger updates the queue's row in pgmq.notify_insert_throttle at each insert and holds that row lock until the inserting transaction ends. One long transaction that enqueues, such as a batch import that starts many flows, blocks every other producer on that queue until it commits.

By default (notify_trigger: :deferred) PgFlow recreates the trigger under the same name as DEFERRABLE INITIALLY DEFERRED. This stays within pgmq's own contract: every pgmq release since 1.7.0 creates the trigger DEFERRABLE, so any caller can already defer it with SET CONSTRAINTS, and every pgmq function that manages the trigger finds it by name. With the deferred trigger:

• The throttle check and NOTIFY run at COMMIT. A producer holds the throttle row lock only while its own commit runs.
• pgmq.disable_notify_insert/1 still removes it.
• A boot that finds the deferred trigger already in place only reconciles the queue's throttle row, inserting it when it is missing, with no DDL. Otherwise the install takes a per-queue advisory lock and waits at most notify_lock_timeout_ms (default 2000) for the queue table lock, then retries with backoff.
• mix pgflow.check_schema warns, without failing, about an immediate trigger on a pgflow flow queue, a notify trigger whose queue has no throttle row, and a partitioned flow queue that the installed pgmq cannot notify for.

Set notify_trigger: :pgmq_default to keep pgmq's trigger. Use one value on every node: a node booting with :pgmq_default, or an older PgFlow release in :notify mode, puts pgmq's immediate trigger back, and the next :deferred boot replaces it again.

Effect on mixed deployments:

• TypeScript workers are unaffected. They never LISTEN and read only committed messages with read_with_poll.
• TypeScript and SQL producers on a shared flow queue (start_flow, complete_task through pgmq.send_batch) fire the trigger at their commit. They also stop holding the throttle row lock across their transaction. No function signature or result changes.
• A deferred row trigger queues one event per inserted row until commit. A very large map fan-out in one transaction carries that queue in memory and runs the throttle check once per row at commit.
• Host code that runs SET CONSTRAINTS ALL IMMEDIATE moves the trigger back to insert time for that transaction.
• Dropping and recreating a queue removes its trigger. Upstream does this in delete_flow_and_data and in is_local() recompilation. PgFlow reinstalls the trigger on its next boot; until then the 30s fallback poll picks up messages.

pgmq compatibility notes:

• A transaction that sends to a queue and then runs pgmq.drop_queue, pgmq.purge_queue, or ALTER TABLE on it fails with "pending trigger events" while the trigger is deferred. Run SET CONSTRAINTS pgmq.trigger_notify_queue_insert_listeners IMMEDIATE first (PgFlow.Queries.Pgmq.flush_notify_events/1 does so when the trigger exists). PgFlow.delete_flow/1 does this itself. Upstream's SQL delete_flow_and_data does not, so a TypeScript or SQL caller that sends and deletes a flow in one transaction needs the statement too.
• REPEATABLE READ and SERIALIZABLE producers can get a serialization failure (40001) at COMMIT when another producer notified on the same queue during their transaction. The immediate trigger has the same failure for a shorter window (from insert to commit).
• ALTER EXTENSION pgmq UPDATE to 1.13.1 rebuilds every notify trigger as pgmq's immediate trigger. Workers defer them again on their next boot, and mix pgflow.check_schema warns until then.
• pgmq keeps pgmq.notify_insert_throttle UNLOGGED, so crash recovery and Neon compute restarts empty it, and a queue without its row sends no NOTIFY with either trigger. PgFlow.Signal.Notify restores missing rows for the queues it listens on every 30 seconds, and every boot restores them too.
• Partitioned queues never notify on pgmq releases before 1.13.1, with either trigger, because the listener derives the queue name from the partition's table name. pgflow creates queues with pgmq.create only.

Feature Comparison

Shared SQL Core

Both call the same PostgreSQL functions:

Core tables include flows, steps, deps (definition) and runs, step_states, step_tasks, workers (runtime).

Schema Divergences

The Elixir implementation adds or omits the following relative to the pinned upstream commit (PgFlow.upstream_sha/0). The compatibility harness manifest records each one.

Existing flow records default to flow_type = 'flow'. Helpers are installed with mix pgflow.gen.helpers_migration.

Compilation

TypeScript: CLI sends HTTP request to ControlPlane edge function, which extracts the flow shape and generates SQL. Written to supabase/migrations/.

Elixir: mix pgflow.gen.flow_migration MyFlow reads __pgflow_definition__/0 and generates an Ecto migration. Worker startup uses PgFlow.Worker.Bootstrap and pgflow.ensure_flow_compiled/2 to compile or verify definitions. FlowCompiler remains available for existing migrations and explicit upserts.