pgflow Workflow Engine

pgflow allows you to move Retrieval-Augmented Generation (RAG) orchestration directly into Postgres. This approach addresses the complexities of vector embedding pipelines, such as detecting content changes, chunking content, calling embedding APIs, storing results, and handling retries without blocking database writes.

Key capabilities covered in the RAG series include:

• Automatic Embeddings: Ingesting, chunking, and embedding data automatically upon INSERT.
• Keeping Data Fresh: Handling updates and deletes to maintain synchronization.
• Advanced Retrieval: Implementing techniques like Hypothetical Document Embeddings (HyDE), reranking, and LLM-augmented retrieval.
• Conversational RAG: Implementing chat functionality with conversation history.

The Edge Worker wraps Supabase Background Tasks to provide enterprise-grade queue management features:

• Retries with delays: Automatically retries failed tasks after a specified delay.
• Concurrency control: Manages how many tasks are processed simultaneously.
• Observability: Provides visibility through heartbeats and logging.
• Horizontal scaling: You can scale throughput by deploying multiple edge functions to listen to the same queue.

pgflow is currently in public beta. While core functionality is considered stable and reliable, users should be aware of the following limitations:

• Performance: While core operations use batching and indexing, extremely large-scale workloads may require additional optimization.
• Fair Processing: There is currently no built-in fair processing across flows. A single flow with many tasks (particularly those using map steps to process large arrays) can saturate the queue and delay other flows. For multi-tenant scenarios requiring fairness, consult the community for workarounds.
• Permissions: pgflow does not include built-in access controls. If using the TypeScript Client or Supabase RPC from client applications, you must manually configure Database permissions.

The context object is pgflow's mechanism for providing handlers with access to platform resources like database connections, environment variables, and service clients. Instead of handlers creating their own connections (which leads to resource leaks and boilerplate), pgflow manages these resources centrally and injects them through a standardized second parameter in handler functions.

When pgflow executes a handler, it passes two parameters:

1. The input object.
2. The context object.

The context parameter is optional; handlers that do not require platform resources can omit it.

When defining a flow, you use different methods to specify how steps behave and how they process data:

• .step(): Defines a Regular Step. This is the standard building block for sequential or dependent tasks.
• .array(): Defines an Array-Returning Step. Use this when a step is designed to return a collection of items rather than a single value.
• .map(): Defines Parallel Array Processing. This method is used to iterate over an array and execute a step for each element in parallel.

By using these methods, pgflow automatically manages dependencies and parallel execution of independent steps.

To prevent a race condition where a worker attempts to process a message before the corresponding step_tasks record exists in the database, pgflow uses a Two-Phase Polling mechanism involving the Worker and SQL Core.

The Process:

1. Phase 1: The Worker calls read_with_poll(). This reserves messages and returns a set of msg_ids.
2. Phase 2: The Worker calls start_tasks(flow_slug, msg_ids, worker_id, queue_name). This creates the necessary step_tasks records and returns the task details.

This ensures that by the time a worker is actively executing a task, the orchestration state is fully persisted in the SQL Core.

The client uses two primary abstractions to manage workflow execution:

• FlowRun: Represents a single execution instance of a workflow. It tracks the overall status (started, completed, failed), input, and final output. It also acts as a container for individual steps.
• FlowStep: Represents an individual step within a FlowRun. Each step has its own lifecycle (created, started, completed, failed), timing metadata (started_at, completed_at), and specific output or error messages.

The TypeScript DSL allows you to define complex workflows using type-safe method chaining. It translates high-level concepts (like mapping over an array) into Directed Acyclic Graph (DAG) primitives that the SQL Core can understand.

Key Features:

• Type-safe method chaining using .step(), .array(), and .map().
• Full type inference across step dependencies.
• Pattern recognition for complex workflows.

Note: The DSL layer does not handle task execution, database state, or queue mechanics; it only defines the structure and intent of the flow.

new Flow<Input>({ slug: "process_users" })
  .step({ slug: "fetch_users" }, fetchUsers)
  .array({ slug: "users_array" }, (input) => input.fetch_users)
  .map({ slug: "send_email" }, sendEmail)

The layers interact through strict, one-way contracts to prevent abstraction leakage:

1. DSL $\rightarrow$ SQL Core: The DSL translates TypeScript code into step definitions. The contract consists of step definitions containing step_type, dependencies, and options.
2. SQL Core $\rightarrow$ Worker: The SQL Core spawns tasks when dependencies are met. The contract is a task object containing a handler function, input data, and retry configuration.
3. Worker $\rightarrow$ SQL Core: The Worker executes the handler and reports the outcome. The contract is a task completion message containing either the output or an error.