Config

Author: paulhenri-l

packages/world-postgres/src/cli.ts

@@ -4,6 +4,7 @@ import { config } from 'dotenv';
 import { drizzle } from 'drizzle-orm/postgres-js';
 import { migrate } from 'drizzle-orm/postgres-js/migrator';
 import postgres from 'postgres';
+import { DEFAULT_PG_URL } from './config.js';
 
 const __dirname = dirname(fileURLToPath(import.meta.url));
 
@@ -14,7 +15,7 @@ async function setupDatabase() {
   const connectionString =
     process.env.WORKFLOW_POSTGRES_URL ||
     process.env.DATABASE_URL ||
-    'postgres://world:world@localhost:5432/world';
+    DEFAULT_PG_URL;
 
   console.log('🔧 Setting up database schema...');
   console.log(

packages/world-postgres/src/config.ts

@@ -1,7 +1,58 @@
 import type { QueueDriver } from './queue-drivers/types.js';
 
-export type PostgresWorldConfig = {
-  securityToken: string;
-  connectionString: string;
-  queueFactory: () => QueueDriver;
+export type BaseWorldConfig = {
+  connectionString?: string;
+  securityToken?: string;
 };
+
+export type PostgresWorldConfig = BaseWorldConfig & {
+  queueFactory?: () => QueueDriver;
+};
+
+export type ResolvedBaseWorldConfig = Required<BaseWorldConfig>;
+
+export const DEFAULT_PG_URL = 'postgres://world:world@localhost:5432/world';
+export const DEFAULT_SECURITY_TOKEN = 'secret';
+export const DEFAULT_JOB_PREFIX = 'workflow_';
+export const DEFAULT_QUEUE_CONCURRENCY = 10;
+
+let worldConfig: ResolvedBaseWorldConfig | null = null;
+
+export function loadWorldConfig(
+  config: BaseWorldConfig = {}
+): ResolvedBaseWorldConfig {
+  worldConfig = {
+    connectionString:
+      config.connectionString ??
+      process.env.WORKFLOW_POSTGRES_URL ??
+      process.env.DATABASE_URL ??
+      DEFAULT_PG_URL,
+
+    securityToken:
+      config.securityToken ??
+      process.env.WORKFLOW_POSTGRES_SECURITY_TOKEN ??
+      DEFAULT_SECURITY_TOKEN,
+  };
+
+  return worldConfig;
+}
+
+export function getWorldConfig(): ResolvedBaseWorldConfig {
+  if (!worldConfig) {
+    throw new Error(
+      'World config not loaded. Call createWorld() or loadWorldConfig().'
+    );
+  }
+
+  return worldConfig;
+}
+
+export function getQueueConfig() {
+  return {
+    jobPrefix: process.env.WORKFLOW_POSTGRES_JOB_PREFIX ?? DEFAULT_JOB_PREFIX,
+    queueConcurrency:
+      (process.env.WORKFLOW_POSTGRES_WORKER_CONCURRENCY
+        ? parseInt(process.env.WORKFLOW_POSTGRES_WORKER_CONCURRENCY, 10)
+        : undefined) ?? DEFAULT_QUEUE_CONCURRENCY,
+  };
+}

packages/world-postgres/src/index.ts

@@ -1,6 +1,6 @@
 import type { Storage, World } from '@workflow/world';
 import createPostgres from 'postgres';
-import type { PostgresWorldConfig } from './config.js';
+import { loadWorldConfig, type PostgresWorldConfig } from './config.js';
 import { createClient, type Drizzle } from './drizzle/index.js';
 import { createFunctionProxy } from './proxies/function-proxy.js';
 import { createHttpProxy } from './proxies/http-proxy.js';
@@ -18,16 +18,15 @@ import {
 } from './storage.js';
 import { createStreamer } from './streamer.js';
 
-export const DEFAULT_PG_URL = 'postgres://world:world@localhost:5432/world';
-
 export function createWorld(
-  config: PostgresWorldConfig = {
-    connectionString: process.env.WORKFLOW_POSTGRES_URL || DEFAULT_PG_URL,
-    securityToken: process.env.WORKFLOW_POSTGRES_SECURITY_TOKEN || 'secret',
-    queueFactory: defaultQueueFactory,
-  }
+  opts: PostgresWorldConfig = {}
 ): World & { start(): Promise<void> } {
-  const queueDriver = config.queueFactory();
+  const config = loadWorldConfig(opts);
+
+  const queueDriver = opts.queueFactory
+    ? opts.queueFactory()
+    : createPgBossHttpProxyQueue();
+
   const postgres = createPostgres(config.connectionString);
   const drizzle = createClient(postgres);
 
@@ -54,26 +53,13 @@ function createStorage(drizzle: Drizzle): Storage {
   };
 }
 
-function defaultQueueFactory() {
-  return createPgBossHttpProxyQueue({
-    baseUrl: process.env.WORKFLOW_POSTGRES_BASE_URL,
-    connectionString: process.env.WORKFLOW_POSTGRES_URL || DEFAULT_PG_URL,
-    securityToken:
-      process.env.WORKFLOW_POSTGRES_SECURITY_TOKEN || 'this-is-not-safe',
-    jobPrefix: process.env.WORKFLOW_POSTGRES_JOB_PREFIX,
-    queueConcurrency:
-      parseInt(process.env.WORKFLOW_POSTGRES_WORKER_CONCURRENCY || '10', 10) ||
-      10,
-  });
-}
-
 export type { PostgresWorldConfig } from './config.js';
 // Re-export schema for users who want to extend or inspect the database schema
 export * from './drizzle/schema.js';
 
 export { createFunctionProxy, createHttpProxy };
 export {
   createPgBossQueue,
-  createPgBossFunctionProxyQueue as createPgBossFunctionProxy,
-  createPgBossHttpProxyQueue as createPgBossHttpProxy,
+  createPgBossFunctionProxyQueue,
+  createPgBossHttpProxyQueue,
 };

packages/world-postgres/src/queue-drivers/factories.ts

@@ -1,3 +1,4 @@
+import { getQueueConfig, getWorldConfig } from '../config.js';
 import { createFunctionProxy } from '../proxies/function-proxy.js';
 import { createHttpProxy } from '../proxies/http-proxy.js';
 import { createPgBossQueue } from './pgboss.js';
@@ -9,20 +10,30 @@ import type { QueueDriver } from './types.js';
  */
 export function createPgBossFunctionProxyQueue(opts: {
   jobPrefix?: string;
-  securityToken: string;
-  connectionString: string;
+  securityToken?: string;
+  connectionString?: string;
   queueConcurrency?: number;
   stepEntrypoint: (request: Request) => Promise<Response>;
   workflowEntrypoint: (request: Request) => Promise<Response>;
 }): QueueDriver {
+  const worldDefaults = getWorldConfig();
+  const queueDefaults = getQueueConfig();
+
+  const config = {
+    connectionString: opts.connectionString ?? worldDefaults.connectionString,
+    securityToken: opts.securityToken ?? worldDefaults.securityToken,
+    jobPrefix: opts.jobPrefix ?? queueDefaults.jobPrefix,
+    queueConcurrency: opts.queueConcurrency ?? queueDefaults.queueConcurrency,
+  };
+
   return createPgBossQueue(
     {
-      jobPrefix: opts.jobPrefix,
-      connectionString: opts.connectionString,
-      queueConcurrency: opts.queueConcurrency,
+      jobPrefix: config.jobPrefix,
+      connectionString: config.connectionString,
+      queueConcurrency: config.queueConcurrency,
     },
     createFunctionProxy({
-      securityToken: opts.securityToken,
+      securityToken: config.securityToken,
       stepEntrypoint: opts.stepEntrypoint,
       workflowEntrypoint: opts.workflowEntrypoint,
     })
@@ -33,14 +44,34 @@ export function createPgBossFunctionProxyQueue(opts: {
  * QueueDriver implementation using pg-boss for job management
  * and HTTP for execution.
  */
-export function createPgBossHttpProxyQueue(config: {
-  port?: number;
-  baseUrl?: string;
-  jobPrefix?: string;
-  securityToken: string;
-  connectionString: string;
-  queueConcurrency?: number;
-}): QueueDriver {
+export function createPgBossHttpProxyQueue(
+  opts: {
+    port?: number;
+    baseUrl?: string;
+    jobPrefix?: string;
+    securityToken?: string;
+    connectionString?: string;
+    queueConcurrency?: number;
+  } = {}
+): QueueDriver {
+  const worldDefaults = getWorldConfig();
+  const queueDefaults = getQueueConfig();
+
+  const config = {
+    connectionString: opts.connectionString ?? worldDefaults.connectionString,
+    securityToken: opts.securityToken ?? worldDefaults.securityToken,
+    jobPrefix: opts.jobPrefix ?? queueDefaults.jobPrefix,
+    queueConcurrency: opts.queueConcurrency ?? queueDefaults.queueConcurrency,
+
+    port:
+      opts.port ??
+      (process.env.WORKFLOW_POSTGRES_APP_PORT
+        ? parseInt(process.env.WORKFLOW_POSTGRES_APP_PORT, 10)
+        : undefined),
+
+    baseUrl: opts.baseUrl ?? process.env.WORKFLOW_POSTGRES_APP_URL,
+  };
+
   return createPgBossQueue(
     {
       jobPrefix: config.jobPrefix,

packages/world-postgres/src/queue-drivers/pgboss.ts

@@ -7,19 +7,18 @@ import { MessageData, type QueueDriver } from './types.js';
  * Takes in a proxy that will handle the actual step/flow execution.
  */
 export function createPgBossQueue(
-  config: {
-    jobPrefix?: string;
+  opts: {
+    jobPrefix: string;
     connectionString: string;
-    queueConcurrency?: number;
+    queueConcurrency: number;
   },
   proxy: WkfProxy
 ): QueueDriver {
   let startPromise: Promise<unknown> | null = null;
-  const boss = new PgBoss(config.connectionString);
+  const boss = new PgBoss(opts.connectionString);
 
-  const prefix = config.jobPrefix || 'workflow_';
-  const stepQueueName = `${prefix}steps`;
-  const workflowQueueName = `${prefix}flows`;
+  const stepQueueName = `${opts.jobPrefix}steps`;
+  const workflowQueueName = `${opts.jobPrefix}flows`;
 
   const ensureStarted = async () => {
     if (!startPromise) {
@@ -59,7 +58,7 @@ export function createPgBossQueue(
       const stepWorker = createWorker(proxy.proxyStep);
       const workflowWorker = createWorker(proxy.proxyWorkflow);
 
-      for (let i = 0; i < (config.queueConcurrency || 10); i++) {
+      for (let i = 0; i < opts.queueConcurrency; i++) {
         await boss.work(workflowQueueName, workflowWorker);
         await boss.work(stepQueueName, stepWorker);
       }