Backend Postgres

PostgreSQL-based implementation of the Backend interface for the workflow orchestrator.

Features

• Event Sourcing: Immutable append-only event storage in PostgreSQL
• Automatic Schema Initialization: Idempotent migrations run automatically on first connection
• Race-Safe Step Claiming: Atomic step claiming using SELECT FOR UPDATE SKIP LOCKED
• Events-as-Queue Pattern: Queue state derived from events (no separate queue table)
• JSONB Storage: Flexible schema evolution with JSONB columns
• Full TypeScript Support: Complete type safety across all operations
• Easy Database Migration: All pg usage isolated in db.ts for easy library swapping

Installation

bun add backend-postgres pg
bun add -d @types/pg

Quick Start

Basic Usage

import { PostgresBackend } from "backend-postgres";

// Create backend with connection string
const backend = new PostgresBackend("postgres://user:password@localhost:5432/workflow_db");

// The database schema is automatically initialized on first use
// No manual migration commands needed!

// Use with the runner
import { runAll } from "runner";
const result = await runAll(workflow, input, { backend });

// Don't forget to close the connection pool when done
await backend.close();

With CLI

You can use the Postgres backend with the CLI by setting the POSTGRES_URL environment variable:

# Set the Postgres connection string
export POSTGRES_URL="postgres://user:password@localhost:5432/workflow_db"

# If POSTGRES_URL uses transaction-pooled PgBouncer, use a direct connection
# for schema migrations and concurrent index creation.
export POSTGRES_MIGRATIONS_URL="postgres://user:password@direct-host:5432/workflow_db"

# Run workflows
wfo run my-workflow --input '{"key": "value"}'

# Queue-based execution
wfo worker start  # Uses Postgres backend
wfo submit my-workflow --input '{...}'

Programmatic Usage

import { PostgresBackend } from "backend-postgres";
import { Client } from "client";

const backend = new PostgresBackend(process.env.POSTGRES_URL!);
const client = new Client({ backend });

// Submit a workflow run
const runId = await client.submit({
  workflowSlug: "my-workflow",
  input: { message: "Hello" },
  tags: ["production"],
  idempotencyKey: "unique-key-123", // Optional deduplication
});

// Wait for completion
const result = await client.waitForCompletion(runId);
console.log("Output:", result.output);

// Cleanup
await backend.close();

When constructing the backend directly with a transaction-pooled runtime URL, pass the direct migration URL as the fourth constructor argument:

const backend = new PostgresBackend(
  process.env.POSTGRES_URL!,
  "cascadeflow",
  undefined,
  process.env.POSTGRES_MIGRATIONS_URL
);

Connection String Format

The connection string follows the standard PostgreSQL format:

postgres://[user[:password]@][host][:port][/database][?option=value]

Examples:

# Local development
postgres://postgres:postgres@localhost:5432/workflow_dev

# With SSL
postgres://user:[email protected]:5432/workflows?sslmode=require

# Unix socket
postgres://user@/dbname?host=/var/run/postgresql

# Cloud providers (example: Neon, Supabase, etc.)
postgres://user:[email protected]/dbname?sslmode=require

Database Schema

The backend creates and manages the following tables:

Event Tables

workflow_events - Workflow-level events

• id (serial) - Primary key
• event_id (text) - Microsecond timestamp string
• workflow_slug (text) - Workflow identifier
• run_id (text) - Run identifier
• timestamp_us (bigint) - Event timestamp in microseconds
• category (text) - Always 'workflow'
• type (text) - Event type (WorkflowStarted, WorkflowCompleted, etc.)
• data (jsonb) - Full event payload

step_events - Step-level events

• id (serial) - Primary key
• event_id (text) - Microsecond timestamp string
• workflow_slug (text) - Workflow identifier
• run_id (text) - Run identifier
• step_id (text) - Step identifier
• timestamp_us (bigint) - Event timestamp in microseconds
• category (text) - Always 'step'
• type (text) - Event type (StepScheduled, StepStarted, etc.)
• data (jsonb) - Full event payload

Registry Tables

workflow_metadata - Workflow registry

• slug (text) - Primary key
• name (text) - Display name
• description (text) - Optional description
• input_schema_json (jsonb) - JSON Schema for input validation
• tags (text[]) - Tags array

step_definitions - Step definitions

• workflow_slug (text) - References workflow_metadata(slug)
• id (text) - Step identifier
• dependencies (jsonb) - Step dependencies
• export_output (boolean) - Whether output is exported
• input_schema_json (jsonb) - JSON Schema for step input
• timeout_ms (integer) - Step timeout
• max_retries (integer) - Maximum retry attempts
• retry_delay_ms (integer) - Delay between retries

Supporting Tables

step_outputs - Serialized step outputs

• workflow_slug, run_id, step_id, attempt_number - Composite primary key
• output (jsonb) - Step output data

idempotency_keys - Deduplication

• hash (text) - Primary key (SHA256 of idempotency key)
• run_id (text) - Associated run ID

Migration Strategy

The backend uses idempotent migrations that are safe to rerun. Migrations are automatically executed on the first database connection.

Current Migrations

1. Migration 001: Create tables - Uses CREATE TABLE IF NOT EXISTS
2. Migration 002: Create indexes - Uses CREATE INDEX IF NOT EXISTS

Adding New Migrations

To add a new migration:

1. Add a new migration function to src/migrations.ts:

async function migration003_addNewFeature(pool: Pool): Promise<void> {
  const client = await pool.connect();
  try {
    // Your migration code here (must be idempotent!)
    await client.query(`
      ALTER TABLE workflow_metadata
      ADD COLUMN IF NOT EXISTS new_field TEXT
    `);

    console.log("[Migration 003] Added new feature");
  } catch (error) {
    console.error("[Migration 003] Error:", error);
    throw error;
  } finally {
    client.release();
  }
}

2. Add it to the runMigrations() function:

export async function runMigrations(pool: Pool): Promise<void> {
  await migration001_createTables(pool);
  await migration002_createIndexes(pool);
  await migration003_addNewFeature(pool);  // Add here
}

Idempotency Best Practices

All migrations must be idempotent (safe to run multiple times):

✅ Good (idempotent):

CREATE TABLE IF NOT EXISTS my_table (...);
CREATE INDEX IF NOT EXISTS idx_name ON table (column);
ALTER TABLE table ADD COLUMN IF NOT EXISTS new_col TEXT;

❌ Bad (not idempotent):

CREATE TABLE my_table (...);  -- Fails if table exists
ALTER TABLE table ADD COLUMN new_col TEXT;  -- Fails if column exists

Architecture Details

Event Sourcing

All state changes are stored as immutable events:

// Events are never updated or deleted
await backend.appendEvent(workflowSlug, runId, {
  category: "step",
  type: "StepStarted",
  eventId: "1234567890123456",
  timestampUs: 1234567890123456,
  stepId: "my-step",
  workerId: "worker-1",
  // ... other event-specific fields
});

// Current state is computed by projecting events
const events = await backend.loadEvents(workflowSlug, runId);
const currentState = projectRunStateFromEvents(events, workflowSlug);

Events-as-Queue Pattern

There is no separate "queue" table. Queue state is derived by projecting the RunSubmitted event along with workflow events:

// Submitting creates a RunSubmitted event
const runId = await backend.submitRun({
  workflowSlug: "my-workflow",
  input: { data: "..." },
});

// Listing runs projects events to compute status
const runs = await backend.listRuns({
  statuses: ["pending", "running"],
});

Atomic Step Claiming

Step claiming uses PostgreSQL's SELECT FOR UPDATE SKIP LOCKED for race-safe concurrency:

// Multiple workers can safely claim different steps
const claimed = await backend.claimScheduledStep(
  workflowSlug,
  runId,
  stepId,
  "worker-1"
);

// Only one worker succeeds per step
console.log(claimed); // true or false

Implementation (simplified):

BEGIN;

-- Acquire lock (SKIP LOCKED prevents blocking)
SELECT * FROM step_events
WHERE workflow_slug = $1 AND run_id = $2 AND step_id = $3
ORDER BY timestamp_us DESC
LIMIT 1
FOR UPDATE SKIP LOCKED;

-- Verify step is in scheduled state
-- Write StepStarted event

COMMIT;

Time Precision

All timestamps use microsecond precision (timestampUs fields) for correct event ordering in concurrent execution:

// Example: 1762916097401523 (microseconds since epoch)
const timestamp = getMicrosecondTimestamp();

Testing

Integration Tests

Integration tests require a running PostgreSQL database:

# Set test database URL
export POSTGRES_TEST_URL="postgres://postgres:postgres@localhost:5432/workflow_test"

# Run tests
bun test

# Run specific test file
bun test tests/integration/postgres-backend.integration.test.ts

Docker PostgreSQL for Testing

# Start PostgreSQL in Docker
docker run -d \
  --name postgres-test \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=workflow_test \
  -p 5432:5432 \
  postgres:16

# Run tests
export POSTGRES_TEST_URL="postgres://postgres:postgres@localhost:5432/workflow_test"
bun test

# Cleanup
docker stop postgres-test
docker rm postgres-test

Performance Considerations

Indexes

The backend creates indexes optimized for common query patterns:

• Event lookups: (workflow_slug, run_id, timestamp_us)
• Step claiming: (workflow_slug, run_id, step_id, type, timestamp_us)
• Registry queries: (workflow_slug)

Connection Pooling

The backend uses pg.Pool for efficient connection management. Configure pool size via connection string:

postgres://user:pass@host/db?max=20&min=5&idle_timeout=10000

JSONB Storage

JSONB columns allow flexible schema evolution but have storage/query trade-offs:

• ✅ Flexible schema changes without migrations
• ✅ Can index specific JSONB fields if needed
• ⚠️ Slightly larger storage than normalized tables
• ⚠️ Complex queries on nested JSONB can be slower

For high-volume production use, consider adding indexes on frequently-queried JSONB fields:

CREATE INDEX idx_event_type ON workflow_events ((data->>'type'));

Switching Database Libraries

All pg usage is isolated in src/db.ts, making it easy to swap to another library:

1. Install new library (e.g., postgres, kysely, drizzle)
2. Update src/db.ts to use new library's API
3. Keep function signatures the same
4. Update src/migrations.ts if needed

The PostgresBackend class in src/index.ts doesn't need changes.

Comparison to FileSystemBackend

Troubleshooting

Connection Issues

// Error: connection refused
// Fix: Check PostgreSQL is running and connection string is correct

// Error: authentication failed
// Fix: Verify username/password in connection string

// Error: database does not exist
// Fix: Create database first: createdb workflow_db

Migration Failures

// Error: relation already exists
// Fix: Ensure migrations use IF NOT EXISTS

// Error: permission denied
// Fix: Grant required permissions to database user

Performance Issues

# Check for missing indexes
EXPLAIN ANALYZE SELECT ...;

# Monitor connection pool
SELECT * FROM pg_stat_activity WHERE datname = 'workflow_db';

# Check table sizes
SELECT pg_size_pretty(pg_total_relation_size('step_events'));

See Also

• Main CLAUDE.md - Backend architecture overview
• Backend Interface - Abstract backend contract
• Backend Filesystem - Filesystem implementation
• Worker README - Distributed worker architecture
• Client README - Programmatic API usage