Use queue prefix

Author: paulhenri-l

packages/world-postgres/src/index.ts

@@ -6,8 +6,8 @@ import { createFunctionProxy } from './proxies/function-proxy.js';
 import { createHttpProxy } from './proxies/http-proxy.js';
 import { createQueue } from './queue.js';
 import {
-  createPgBossFunctionProxy,
-  createPgBossHttpProxy,
+  createPgBossFunctionProxyQueue,
+  createPgBossHttpProxyQueue,
 } from './queue-drivers/factories.js';
 import { createPgBossQueue } from './queue-drivers/pgboss.js';
 import {
@@ -55,11 +55,12 @@ function createStorage(drizzle: Drizzle): Storage {
 }
 
 function defaultQueueFactory() {
-  return createPgBossHttpProxy({
+  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,
@@ -71,4 +72,8 @@ export type { PostgresWorldConfig } from './config.js';
 export * from './drizzle/schema.js';
 
 export { createFunctionProxy, createHttpProxy };
-export { createPgBossQueue, createPgBossFunctionProxy, createPgBossHttpProxy };
+export {
+  createPgBossQueue,
+  createPgBossFunctionProxyQueue as createPgBossFunctionProxy,
+  createPgBossHttpProxyQueue as createPgBossHttpProxy,
+};

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

@@ -7,7 +7,7 @@ import type { QueueDriver } from './types.js';
  * QueueDriver implementation using pg-boss for job management
  * and direct function calls for execution.
  */
-export function createPgBossFunctionProxy(opts: {
+export function createPgBossFunctionProxyQueue(opts: {
   jobPrefix?: string;
   securityToken: string;
   connectionString: string;
@@ -33,7 +33,7 @@ export function createPgBossFunctionProxy(opts: {
  * QueueDriver implementation using pg-boss for job management
  * and HTTP for execution.
  */
-export function createPgBossHttpProxy(config: {
+export function createPgBossHttpProxyQueue(config: {
   port?: number;
   baseUrl?: string;
   jobPrefix?: string;

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

@@ -17,8 +17,9 @@ export function createPgBossQueue(
   let startPromise: Promise<unknown> | null = null;
   const boss = new PgBoss(config.connectionString);
 
-  const stepQueueName = 'workflow_steps';
-  const workflowQueueName = 'workflow_flows';
+  const prefix = config.jobPrefix || 'workflow_';
+  const stepQueueName = `${prefix}steps`;
+  const workflowQueueName = `${prefix}flows`;
 
   const ensureStarted = async () => {
     if (!startPromise) {