client

@pgflow/client

TypeScript client library for interacting with pgflow workflows and observing real-time progress updates.

Caution

Security Notice: pgflow does not handle security yet. YOU are responsible for securing access to schemas, tables, and functions.

See SECURITY.md for required permission grants and their implications.

Note

This project and all its components are licensed under Apache 2.0 license.

Overview

@pgflow/client provides a type-safe client for starting pgflow workflows, monitoring their execution, and handling real-time updates. It integrates seamlessly with Supabase for database operations and real-time subscriptions.

Key features:

• Type-Safe Workflow Management - Full TypeScript support with automatic type inference
• Real-Time Updates - Live progress monitoring via Supabase broadcasts
• Event-Driven Architecture - Subscribe to workflow and step events
• Resource Management - Automatic cleanup and disposal of resources
• Error Handling - Comprehensive error recovery and retry mechanisms

Installation

npm install @pgflow/client

Note

Real-time updates use Supabase broadcast channels. Ensure real-time is enabled in your Supabase project.

Quick Start

import { createClient } from '@supabase/supabase-js';
import { PgflowClient, FlowRunStatus } from '@pgflow/client';

// Initialize clients
const supabase = createClient('https://your-project.supabase.co', 'your-anon-key');
const pgflow = new PgflowClient(supabase);

// Start a workflow
const run = await pgflow.startFlow('analyze_website', {
  url: 'https://example.com',
});

console.log(`Started workflow: ${run.run_id}`);

// Wait for completion
const completed = await run.waitForStatus(FlowRunStatus.Completed, {
  timeoutMs: 30000,
});

console.log('Workflow completed:', completed.output);

Core Concepts

FlowRun

A FlowRun represents a single execution instance of a workflow. It provides methods to monitor progress and interact with individual steps.

// Get run status and metadata
console.log(run.status); // 'started' | 'completed' | 'failed'
console.log(run.flow_slug); // 'analyze_website'
console.log(run.input); // Original input data
console.log(run.output); // Final output (when completed)

// Access individual steps
const step = run.step('website_scraper');
console.log(step.status); // Step-specific status
console.log(step.output); // Step output data

FlowStep

Individual steps within a workflow run. Each step has its own status, input, output, and timing information.

const step = run.step('sentiment_analysis');

// Step metadata
console.log(step.step_slug); // 'sentiment_analysis'
console.log(step.status); // 'created' | 'started' | 'completed' | 'failed'
console.log(step.started_at); // Timestamp when step started
console.log(step.completed_at); // Timestamp when step completed
console.log(step.error_message); // Error message if failed

API Reference

PgflowClient

startFlow<TFlow>(flow_slug, input, run_id?)

Starts a new workflow execution.

const run = await pgflow.startFlow('my_flow', { data: 'input' });

// With custom run ID
const run = await pgflow.startFlow('my_flow', { data: 'input' }, 'custom-id');

// With type safety (requires flow definition)
import { MyFlow } from './flows/my-flow';
const run = await pgflow.startFlow<typeof MyFlow>(MyFlow.slug, {
  data: 'input',
});

getRun(run_id)

Retrieves an existing workflow run.

const run = await pgflow.getRun('run-uuid');
if (run) {
  console.log('Found run:', run.status);
} else {
  console.log('Run not found');
}

dispose(run_id) / disposeAll()

Clean up resources for specific runs or all runs.

// Dispose specific run
pgflow.dispose('run-uuid');

// Dispose all runs
pgflow.disposeAll();

FlowRun Methods

waitForStatus(status, options?)

Wait for the run to reach a specific status.

// Wait for completion
const completed = await run.waitForStatus(FlowRunStatus.Completed);

// With timeout
const completed = await run.waitForStatus(FlowRunStatus.Completed, {
  timeoutMs: 30000,
});

// Wait for any terminal status
const terminal = await run.waitForStatus([
  FlowRunStatus.Completed,
  FlowRunStatus.Failed,
]);

Event Subscription

Subscribe to run-level events.

// Listen to all events
run.on('*', (event) => {
  console.log('Event:', event.status);
});

// Listen to specific events
run.on('completed', (event) => {
  console.log('Run completed:', event.output);
});

run.on('failed', (event) => {
  console.log('Run failed:', event.error_message);
});

// Unsubscribe
const unsubscribe = run.on('completed', handler);
unsubscribe();

FlowStep Methods

waitForStatus(status, options?)

Wait for a step to reach a specific status.

import { FlowStepStatus } from '@pgflow/client';

const step = run.step('data_processing');

// Wait for step completion
await step.waitForStatus(FlowStepStatus.Completed);

// With timeout
await step.waitForStatus(FlowStepStatus.Completed, {
  timeoutMs: 15000,
});

Event Subscription

Subscribe to step-level events.

const step = run.step('analysis');

step.on('started', (event) => {
  console.log(`Step started at: ${event.started_at}`);
});

step.on('completed', (event) => {
  console.log(`Step completed:`, event.output);
});

step.on('failed', (event) => {
  console.log(`Step failed: ${event.error_message}`);
});

Type Safety with Flow Definitions

When using with @pgflow/dsl, you get full type safety:

import { Flow } from '@pgflow/dsl';

// Define your flow
const AnalyzeWebsite = new Flow<{ url: string }>({ slug: 'analyzeWebsite' })
  .step({ slug: 'scrape' }, async (input) => ({ content: 'html...' }))
  .step({ slug: 'analyze' }, async (input) => ({ sentiment: 0.8 }));

// Type-safe client usage
const run = await pgflow.startFlow<typeof AnalyzeWebsite>(
  AnalyzeWebsite.slug,
  { url: 'https://example.com' } // TypeScript validates this matches Flow input
);

// Typed step access
const scrapeStep = run.step('scrape'); // TypeScript knows this step exists
const analyzeStep = run.step('analyze');

// Typed output access
await run.waitForStatus(FlowRunStatus.Completed);
console.log(run.output); // TypeScript knows the output structure

Advanced Usage

Custom Error Handling

try {
  const run = await pgflow.startFlow('risky_workflow', { data: 'test' });

  // Monitor for failures
  run.on('failed', (event) => {
    console.error('Workflow failed:', event.error_message);
    // Handle failure (retry, alert, etc.)
  });

  const result = await run.waitForStatus([
    FlowRunStatus.Completed,
    FlowRunStatus.Failed,
  ]);

  if (result.status === FlowRunStatus.Failed) {
    throw new Error(`Workflow failed: ${result.error_message}`);
  }
} catch (error) {
  console.error('Error managing workflow:', error);
}

Monitoring Multiple Steps

const run = await pgflow.startFlow('complex_workflow', { data: 'input' });

// Monitor specific steps in parallel
const steps = ['step1', 'step2', 'step3'];
const stepPromises = steps.map((stepSlug) =>
  run.step(stepSlug).waitForStatus(FlowStepStatus.Completed)
);

// Wait for all steps to complete
await Promise.all(stepPromises);
console.log('All monitored steps completed');

Resource Cleanup

const run = await pgflow.startFlow('my_workflow', { data: 'input' });

// Auto-cleanup when workflow finishes
run.on('*', (event) => {
  if (['completed', 'failed'].includes(event.status)) {
    pgflow.dispose(run.run_id);
  }
});

// Or cleanup manually after waiting
await run.waitForStatus(FlowRunStatus.Completed);
pgflow.dispose(run.run_id);

Requirements

• Supabase Project with pgflow schema installed
• Real-time enabled for live updates
• Proper permissions for the pgflow schema

Schema Setup

To use the client, your Supabase project needs:

1. pgflow schema installed (via @pgflow/core migrations)

2. Proper permissions - See SECURITY.md for required grants

3. Exposed schema - Add pgflow to your exposed schemas:

   • Dashboard: Settings → API → Data API Settings → Exposed schemas
   • Config: Add to api.schemas in supabase/config.toml:

     [api]
     schemas = ["public", "graphql_public", "pgflow"]
     extra_search_path = ["public", "extensions"]
     max_rows = 1000

Development

Building

nx build client

For detailed information about the build process and output formats, see BUILD_AND_RELEASE.md.

Testing

# Unit tests
nx test client

# Integration tests (requires local Supabase)
nx test:integration client

Example Usage

See examples/basic.ts for comprehensive usage examples.

Browser Usage

<script src="https://unpkg.com/@pgflow/client"></script>
<script>
  const pgflow = window.pgflow.createClient(supabase);
</script>

See BUILD_AND_RELEASE.md for full CDN usage details.

Related Packages

• @pgflow/core - PostgreSQL workflow engine
• @pgflow/dsl - TypeScript workflow definitions
• @pgflow/edge-worker - Task execution runtime

Documentation

For more detailed documentation, visit:

• pgflow Documentation
• Start Flows from TypeScript Client
• Client API Reference