Code organisation

Author: paulhenri-l

packages/world-postgres/src/index.ts

@@ -2,8 +2,14 @@ import type { Storage, World } from '@workflow/world';
 import createPostgres from 'postgres';
 import 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';
 import { createQueue } from './queue.js';
-import { createPgBossHttpProxy } from './queue-drivers/pgboss-http-proxy.js';
+import {
+  createPgBossFunctionProxy,
+  createPgBossHttpProxy,
+} from './queue-drivers/factories.js';
+import { createPgBossQueue } from './queue-drivers/pgboss.js';
 import {
   createEventsStorage,
   createHooksStorage,
@@ -50,10 +56,10 @@ function createStorage(drizzle: Drizzle): Storage {
 
 function defaultQueueFactory() {
   return createPgBossHttpProxy({
-    securityToken:
-      process.env.WORKFLOW_POSTGRES_SECURITY_TOKEN || 'this-is-not-safe',
     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',
     queueConcurrency:
       parseInt(process.env.WORKFLOW_POSTGRES_WORKER_CONCURRENCY || '10', 10) ||
       10,
@@ -63,8 +69,6 @@ function defaultQueueFactory() {
 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 type { PgBossFunctionProxyConfig } from './queue-drivers/pgboss-function-proxy.js';
-export { createPgBossFunctionProxy } from './queue-drivers/pgboss-function-proxy.js';
-export type { PgBossHttpProxyConfig } from './queue-drivers/pgboss-http-proxy.js';
-// Re-export queue drivers for custom configurations
-export { createPgBossHttpProxy } from './queue-drivers/pgboss-http-proxy.js';
+
+export { createFunctionProxy, createHttpProxy };
+export { createPgBossQueue, createPgBossFunctionProxy, createPgBossHttpProxy };

packages/world-postgres/src/proxies/function-proxy.ts

@@ -0,0 +1,29 @@
+import type { MessageData } from '../queue-drivers/types.js';
+import type { WkfProxy } from './types.js';
+import { prepareRequestParams } from './utils.js';
+
+export function createFunctionProxy(opts: {
+  securityToken: string;
+  stepEntrypoint: (request: Request) => Promise<Response>;
+  workflowEntrypoint: (request: Request) => Promise<Response>;
+}): WkfProxy {
+  return {
+    proxyWorkflow: async (message: MessageData): Promise<Response> => {
+      const request = new Request(
+        'https://world-postgres.local/wkf-direct-call',
+        prepareRequestParams(message, opts.securityToken)
+      );
+
+      return opts.workflowEntrypoint(request);
+    },
+
+    proxyStep: async (message: MessageData): Promise<Response> => {
+      const request = new Request(
+        'https://world-postgres.local/wkf-direct-call',
+        prepareRequestParams(message, opts.securityToken)
+      );
+
+      return opts.stepEntrypoint(request);
+    },
+  };
+}

packages/world-postgres/src/proxies/http-proxy.ts

@@ -0,0 +1,33 @@
+import type { MessageData } from '../queue-drivers/types.js';
+import type { WkfProxy } from './types.js';
+import { prepareRequestParams } from './utils.js';
+
+export function createHttpProxy(opts: {
+  port?: number;
+  baseUrl?: string;
+  securityToken: string;
+}): WkfProxy {
+  const resolveBaseUrl = (): string => {
+    if (opts.baseUrl) return opts.baseUrl;
+    if (opts.port) return `http://localhost:${opts.port}`;
+    return 'http://localhost:3000';
+  };
+
+  const baseUrl = resolveBaseUrl();
+
+  return {
+    proxyWorkflow: async (message: MessageData): Promise<Response> => {
+      return fetch(
+        `${baseUrl}/.well-known/workflow/v1/flow`,
+        prepareRequestParams(message, opts.securityToken)
+      );
+    },
+
+    proxyStep: async (message: MessageData): Promise<Response> => {
+      return fetch(
+        `${baseUrl}/.well-known/workflow/v1/step`,
+        prepareRequestParams(message, opts.securityToken)
+      );
+    },
+  };
+}

packages/world-postgres/src/proxies/types.ts

@@ -0,0 +1,6 @@
+import type { MessageData } from '../queue-drivers/types.js';
+
+export interface WkfProxy {
+  proxyWorkflow: (message: MessageData) => Promise<Response>;
+  proxyStep: (message: MessageData) => Promise<Response>;
+}

packages/world-postgres/src/proxies/utils.ts

@@ -0,0 +1,15 @@
+import { MessageData } from '../queue-drivers/types.js';
+
+export const prepareRequestParams = (
+  message: MessageData,
+  securityToken: string
+) => {
+  return {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      'X-Workflow-Secret': securityToken,
+    },
+    body: JSON.stringify(MessageData.encode(message)),
+  };
+};

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

@@ -0,0 +1,56 @@
+import { createFunctionProxy } from '../proxies/function-proxy.js';
+import { createHttpProxy } from '../proxies/http-proxy.js';
+import { createPgBossQueue } from './pgboss.js';
+import type { QueueDriver } from './types.js';
+
+/**
+ * QueueDriver implementation using pg-boss for job management
+ * and direct function calls for execution.
+ */
+export function createPgBossFunctionProxy(opts: {
+  jobPrefix?: string;
+  securityToken: string;
+  connectionString: string;
+  queueConcurrency?: number;
+  stepEntrypoint: (request: Request) => Promise<Response>;
+  workflowEntrypoint: (request: Request) => Promise<Response>;
+}): QueueDriver {
+  return createPgBossQueue(
+    {
+      jobPrefix: opts.jobPrefix,
+      connectionString: opts.connectionString,
+      queueConcurrency: opts.queueConcurrency,
+    },
+    createFunctionProxy({
+      securityToken: opts.securityToken,
+      stepEntrypoint: opts.stepEntrypoint,
+      workflowEntrypoint: opts.workflowEntrypoint,
+    })
+  );
+}
+
+/**
+ * QueueDriver implementation using pg-boss for job management
+ * and HTTP for execution.
+ */
+export function createPgBossHttpProxy(config: {
+  port?: number;
+  baseUrl?: string;
+  jobPrefix?: string;
+  securityToken: string;
+  connectionString: string;
+  queueConcurrency?: number;
+}): QueueDriver {
+  return createPgBossQueue(
+    {
+      jobPrefix: config.jobPrefix,
+      connectionString: config.connectionString,
+      queueConcurrency: config.queueConcurrency,
+    },
+    createHttpProxy({
+      port: config.port,
+      baseUrl: config.baseUrl,
+      securityToken: config.securityToken,
+    })
+  );
+}