add graphile queue driver

alex-holovach · alex-holovach · commit d2709e6646e4 · 2025-11-30T12:25:50.000-06:00
diff --git a/packages/utils/package.json b/packages/utils/package.json
@@ -37,6 +37,7 @@
     "@types/ms": "2.1.0",
     "@types/node": "catalog:",
     "@workflow/tsconfig": "workspace:*",
+    "execa": "9.6.1",
     "vitest": "catalog:"
   },
   "dependencies": {
diff --git a/packages/world-postgres/package.json b/packages/world-postgres/package.json
@@ -52,6 +52,7 @@
     "cbor-x": "1.6.0",
     "dotenv": "16.4.5",
     "drizzle-orm": "0.44.7",
+    "graphile-worker": "0.16.6",
     "pg-boss": "11.0.7",
     "postgres": "3.4.7",
     "ulid": "3.0.1",
diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts
@@ -6,10 +6,14 @@ import { createFunctionProxy } from './proxies/function-proxy.js';
 import { createHttpProxy } from './proxies/http-proxy.js';
 import { createQueue } from './queue.js';
 import {
+  createGraphileWorkerFunctionProxyQueue,
+  createGraphileWorkerHttpProxyQueue,
   createPgBossFunctionProxyQueue,
   createPgBossHttpProxyQueue,
 } from './queue-drivers/factories.js';
+import { createGraphileWorkerQueue } from './queue-drivers/graphile.js';
 import { createPgBossQueue } from './queue-drivers/pgboss.js';
+import type { QueueDriver } from './queue-drivers/types.js';
 import {
   createEventsStorage,
   createHooksStorage,
@@ -18,14 +22,30 @@ import {
 } from './storage.js';
 import { createStreamer } from './streamer.js';
 
+/**
+ * Get the default queue factory based on WORKFLOW_QUEUE_DRIVER env var.
+ * Defaults to pg-boss for backwards compatibility.
+ *
+ * Set WORKFLOW_QUEUE_DRIVER=graphile to use Graphile Worker.
+ */
+function getDefaultQueueFactory(): () => QueueDriver {
+  const driver = process.env.WORKFLOW_QUEUE_DRIVER || 'pgboss';
+
+  if (driver === 'graphile') {
+    return createGraphileWorkerHttpProxyQueue;
+  }
+
+  return createPgBossHttpProxyQueue;
+}
+
 export function createWorld(
   opts: PostgresWorldConfig = {}
 ): World & { start(): Promise<void> } {
   const config = loadWorldConfig(opts);
 
   const queueDriver = opts.queueFactory
     ? opts.queueFactory()
-    : createPgBossHttpProxyQueue();
+    : getDefaultQueueFactory()();
 
   const postgres = createPostgres(config.connectionString);
   const drizzle = createClient(postgres);
@@ -63,3 +83,8 @@ export {
   createPgBossFunctionProxyQueue,
   createPgBossHttpProxyQueue,
 };
+export {
+  createGraphileWorkerQueue,
+  createGraphileWorkerFunctionProxyQueue,
+  createGraphileWorkerHttpProxyQueue,
+};
diff --git a/packages/world-postgres/src/queue-drivers/factories.ts b/packages/world-postgres/src/queue-drivers/factories.ts
@@ -1,6 +1,7 @@
 import { getQueueConfig, getWorldConfig } from '../config.js';
 import { createFunctionProxy } from '../proxies/function-proxy.js';
 import { createHttpProxy } from '../proxies/http-proxy.js';
+import { createGraphileWorkerQueue } from './graphile.js';
 import { createPgBossQueue } from './pgboss.js';
 import type { QueueDriver } from './types.js';
 
@@ -85,3 +86,91 @@ export function createPgBossHttpProxyQueue(
     })
   );
 }
+
+/**
+ * QueueDriver implementation using Graphile Worker for job management
+ * and direct function calls for execution.
+ *
+ * Graphile Worker uses PostgreSQL LISTEN/NOTIFY for near-instant job pickup
+ * (~3ms latency vs 500ms polling with pg-boss).
+ */
+export function createGraphileWorkerFunctionProxyQueue(opts: {
+  jobPrefix?: 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 createGraphileWorkerQueue(
+    {
+      jobPrefix: config.jobPrefix,
+      connectionString: config.connectionString,
+      queueConcurrency: config.queueConcurrency,
+    },
+    createFunctionProxy({
+      securityToken: config.securityToken,
+      stepEntrypoint: opts.stepEntrypoint,
+      workflowEntrypoint: opts.workflowEntrypoint,
+    })
+  );
+}
+
+/**
+ * QueueDriver implementation using Graphile Worker for job management
+ * and HTTP for execution.
+ *
+ * Graphile Worker uses PostgreSQL LISTEN/NOTIFY for near-instant job pickup
+ * (~3ms latency vs 500ms polling with pg-boss).
+ */
+export function createGraphileWorkerHttpProxyQueue(
+  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 createGraphileWorkerQueue(
+    {
+      jobPrefix: config.jobPrefix,
+      connectionString: config.connectionString,
+      queueConcurrency: config.queueConcurrency,
+    },
+    createHttpProxy({
+      port: config.port,
+      baseUrl: config.baseUrl,
+      securityToken: config.securityToken,
+    })
+  );
+}
diff --git a/packages/world-postgres/src/queue-drivers/graphile.ts b/packages/world-postgres/src/queue-drivers/graphile.ts
@@ -0,0 +1,111 @@
+import {
+  makeWorkerUtils,
+  run,
+  type Task,
+  type WorkerUtils,
+} from 'graphile-worker';
+import type { WkfProxy } from '../proxies/types.js';
+import { MessageData, type QueueDriver } from './types.js';
+
+/**
+ * QueueDriver implementation using Graphile Worker for job management.
+ * Uses PostgreSQL LISTEN/NOTIFY for near-instant job pickup (~3ms latency).
+ * Takes in a proxy that will handle the actual step/flow execution.
+ */
+export function createGraphileWorkerQueue(
+  opts: {
+    jobPrefix: string;
+    connectionString: string;
+    queueConcurrency: number;
+  },
+  proxy: WkfProxy
+): QueueDriver {
+  let workerUtils: WorkerUtils | null = null;
+
+  const stepTaskName = `${opts.jobPrefix}step`;
+  const flowTaskName = `${opts.jobPrefix}flow`;
+
+  const ensureUtils = async (): Promise<WorkerUtils> => {
+    if (!workerUtils) {
+      workerUtils = await makeWorkerUtils({
+        connectionString: opts.connectionString,
+      });
+      await workerUtils.migrate();
+    }
+    return workerUtils;
+  };
+
+  const createTaskHandler = (
+    proxyFn: WkfProxy[keyof WkfProxy],
+    taskName: string
+  ): Task => {
+    return async (payload, helpers) => {
+      const message = MessageData.parse(payload);
+
+      helpers.logger.info(`Running: ${message.queueName}`);
+
+      try {
+        const response = await proxyFn(message);
+
+        if (response.status === 503) {
+          const body = (await response.json()) as { timeoutSeconds?: number };
+
+          if (body.timeoutSeconds) {
+            // Requeue the job with a delay
+            const utils = await ensureUtils();
+            await utils.addJob(taskName, MessageData.encode(message), {
+              jobKey: message.idempotencyKey ?? message.messageId,
+              maxAttempts: 3,
+              runAt: new Date(Date.now() + body.timeoutSeconds * 1000),
+            });
+
+            helpers.logger.info(
+              `Requeued: ${message.queueName} for ${body.timeoutSeconds}s`
+            );
+
+            return;
+          }
+        }
+
+        if (!response.ok) {
+          const text = await response.text();
+          throw new Error(`Step failed: ${text}`);
+        }
+      } catch (error) {
+        helpers.logger.error(`Error handling: ${message.queueName}`, { error });
+        throw error;
+      }
+    };
+  };
+
+  return {
+    pushStep: async (message: MessageData) => {
+      const utils = await ensureUtils();
+      await utils.addJob(stepTaskName, MessageData.encode(message), {
+        jobKey: message.idempotencyKey ?? message.messageId,
+        maxAttempts: 3,
+      });
+    },
+
+    pushFlow: async (message: MessageData) => {
+      const utils = await ensureUtils();
+      await utils.addJob(flowTaskName, MessageData.encode(message), {
+        jobKey: message.idempotencyKey ?? message.messageId,
+        maxAttempts: 3,
+      });
+    },
+
+    start: async () => {
+      await ensureUtils();
+
+      await run({
+        connectionString: opts.connectionString,
+        concurrency: opts.queueConcurrency,
+        taskList: {
+          [stepTaskName]: createTaskHandler(proxy.proxyStep, stepTaskName),
+          [flowTaskName]: createTaskHandler(proxy.proxyWorkflow, flowTaskName),
+        },
+      });
+    },
+  };
+}
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
