Postgres Conductor

Durable execution using only Postgres

Documentation

Why Postgres Conductor?

Durable Execution - Tasks survive crashes and restarts through automatic step memoization

Multiple Triggers - Invocable APIs, cron schedules, custom events, and database triggers

Workflows - Invoke child tasks and wait for results with full type safety

Performance - Process up to 20,000 jobs per second with just Postgres

Type-Safety - Share common event and task schemas across your services

Simple Infrastructure - No Redis, no message queues - just Postgres

Installation

Postgres Conductor is available on npm:

pnpm install pgconductor-js

Quick Start

Here's a minimal example to get you started:

Define a Task

Tasks are defined with a name and optional payload schema using Standard Schema:

import { z } from "zod";
import { defineTask } from "pgconductor-js";

const greetTask = defineTask({
  name: "greet",
  payload: z.object({ name: z.string() }),
});

Create the Conductor

The Conductor manages task definitions and the database connection:

import { Conductor } from "pgconductor-js";
import { TaskSchemas } from "pgconductor-js";

const sql = postgres();

const conductor = Conductor.create({
  connectionString: "postgres://localhost:5432/mydb",
  tasks: TaskSchemas.fromSchema([greetTask])
});

Implement the Task Handler

Create a task handler that defines what happens when the task runs:

const greet = conductor.createTask(
  { name: "greet" },
  { invocable: true }, // Allow invocation
  async (event, ctx) => {
    const { name } = event.payload;
    ctx.logger.info(`Hello, ${name}!`);
  }
);

Set Up the Orchestrator

The Orchestrator runs workers that execute tasks:

import { Orchestrator } from "pgconductor-js";

const orchestrator = Orchestrator.create({
  conductor,
  tasks: [greet],
});

await orchestrator.start();
console.log("Worker running. Press Ctrl+C to stop.");
await orchestrator.stopped;

Invoke the Task

In a separate process, invoke the task via an instance of the Conductor:

await conductor.invoke(
  { name: "greet" },
  { name: "World" }
);