$ show pgflow-dev/pgflow
pgflow is a PostgreSQL/Supabase-native workflow orchestration engine for defining and executing typed DAGs as background jobs. The system is split into six coordinated packages: a PL/pgSQL core (pkgs/core) that implements the complete workflow state machine in the database — flows, steps, runs, step states, tasks, retries with exponential backoff, map/reduce fan-out, conditional step execution, skip-cascade propagation, and realtime event broadcasting — plus a type-safe TypeScript DSL (pkgs/dsl) for declaring flows with compile-time-enforced inputs/outputs, a Deno/edge worker runtime (pkgs/edge-worker) that consumes PGMQ messages and executes step handlers via platform adapters (Supabase vs. plain process), a realtime-subscription client library (pkgs/client), and a CLI (pkgs/cli) for scaffolding new Supabase projects. Execution relies on pgmq for task queues, pg_cron for stalled-task recovery and worker health, pg_net for HTTP worker invocation, supabase_vault for credentials, and Supabase Realtime for atomic state-transition broadcasts.
The LLM identified 5 files worth highlighting for reuse:
Efficiently updates message visibility timeouts for a batch of PGMQ tasks using a single unnest-join UPDATE instead of per-row set_vt calls. This is a directly reusable SQL optimization pattern for any PGMQ consumer that needs to claim multiple messages in a transaction, and is cited throughout the codebase as the preferred path over the older per-row approach.
Lazy flow-input loader that fetches the flow input exactly once per run, caching the result and de-duplicating concurrent pending promises so that parallel steps waiting on the same input share a single in-flight query. A clean, self-contained caching pattern (fetch → memoize → dedup) that transfers directly to any serverless or edge workload where the same upstream resource is needed by multiple concurrent handlers.
Event-driven state machine for a workflow run, built on nanoevents: it ingests typed run:* and step:* realtime events, applies a deterministic status-precedence map, protects terminal states from regression, exposes waitForStatus with timeout/abort, and auto-disposes listeners. The pattern — realtime event ingestion + idempotent state transition + guarded wait — is a reusable template for building resilient realtime UIs against out-of-order or duplicate broadcasts.
Type-safe Flow builder implementing a fully immutable fluent DSL: each .step()/.map() call returns a new frozen Flow instance. Contains the recursive ContainmentPattern<T> type (mirroring Postgres @> containment semantics for step conditions) and a conditional-options discriminated union that forces whenUnmet to be present exactly when if/ifNot is specified. Demonstrates advanced TypeScript generic design — compile-time enforcement of dependency shape and JSON-serializable outputs — that is broadly applicable to any typed workflow/state-machine library.
Low-level typed client for the pgflow SQL functions (readMessages, startTasks, completeTask, failTask, startFlow) that encodes the two-phase polling contract — read from PGMQ in one statement, then start tasks with a fresh snapshot in a second — that avoids claiming races. A clean, testable database-access seam that decouples the worker runtime from SQL details, and a good reference for any queue consumer needing to avoid snapshot-staleness bugs.