vercel__workflow/packages/world-postgres/HOW_IT_WORKS.md

How PostgreSQL World Works

This document explains the architecture and components of the PostgreSQL world implementation for workflow management.

This implementation is using Drizzle Schema that can be pushed or migrated into your PostgreSQL schema and backed by node-postgres (pg). createWorld uses a single pg.Pool for Drizzle and graphile-worker (via pgPool), and a dedicated pg.Client for LISTEN/NOTIFY derived from the same connection options. You may pass your own pool to share query connections with application code.

If you want to use any other ORM, query builder or underlying database client, you should be able to fork this implementation and replace the Drizzle parts with your own.

Job Queue System

graph LR
    Client --> PG[graphile-worker queue]
    PG --> Worker[Embedded Worker]
    Worker --> HTTP[Workflow HTTP routes]
    HTTP --> Handler[Workflow or Step Handler]

    PG -.-> F["${prefix}flows<br/>(workflows)"]
    PG -.-> S["${prefix}steps<br/>(steps)"]

Jobs include retry logic (3 attempts), idempotency keys, durable delayed rescheduling, and configurable worker concurrency (default: 10).

Streaming

Real-time data streaming via PostgreSQL LISTEN/NOTIFY:

• Stream chunks stored in workflow_stream_chunks table
• pg_notify triggers sent on writes to workflow_event_chunk topic
• Subscribers receive notifications and fetch chunk data
• ULID-based ordering ensures correct sequence
• One long-lived dedicated LISTEN client, with an in-process EventEmitter for distributing events to multiple subscribers

Setup

Call world.start() to initialize graphile-worker workers. When .start() is called, workers begin listening to graphile-worker queues. When a job arrives, the worker executes the queue message over the workflow HTTP routes and awaits completion before acknowledging the Graphile job.

When the runtime returns { timeoutSeconds }, the worker schedules a new Graphile job with a future runAt time before finishing the current task.

The worker targets the HTTP-compatible workflow endpoints directly: .well-known/workflow/v1/flow for workflows and .well-known/workflow/v1/step for steps.

In Next.js, the world.start() call needs to be added to instrumentation.ts|js to ensure workers start before request handling. Use workflow/runtime for getWorld (same as the testing server and other framework plugins):

// instrumentation.ts

if (process.env.NEXT_RUNTIME !== "edge") {
  import("workflow/runtime").then(async ({ getWorld }) => {
    // start listening to the jobs.
    const world = await getWorld();
    await world.start?.();
  });
}