Deliver Postgres queue messages in process

.changeset/framework-direct-queue-bootstrap.md

@@ -0,0 +1,10 @@
+---
+'@workflow/astro': patch
+'@workflow/nest': patch
+'@workflow/nitro': patch
+'@workflow/next': patch
+'@workflow/sveltekit': patch
+'@workflow/world-testing': patch
+---
+
+Initialize generated flow handlers before starting direct-delivery Worlds.

.changeset/postgres-direct-delivery.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Deliver Postgres workflow queue messages directly to the in-process handler and create runs atomically with their first event.

packages/astro/src/builder.ts

@@ -84,10 +84,11 @@ export class LocalBuilder extends BaseBuilder {
     workflowsRouteContent = workflowsRouteContent.replace(
       /export const POST = workflowEntrypoint\(workflowCode(?<options>[^)]*)\);?$/m,
       (_match, options = '') => `${NORMALIZE_REQUEST_CODE}
-export const POST = async ({request}) => {
+const workflowHandler = workflowEntrypoint(workflowCode${options});
+export const POST = Object.assign(async ({request}) => {
   const normalRequest = await normalizeRequest(request);
-  return workflowEntrypoint(workflowCode${options})(normalRequest);
-}
+  return workflowHandler(normalRequest);
+}, { initialize: workflowHandler.initialize });
 
 export const prerender = false;`
     );

packages/nest/src/workflow.controller.ts

@@ -91,6 +91,12 @@ function getOutDir(): string {
   return controllerConfig.outDir;
 }
 
+export async function loadWorkflowHandler() {
+  const outDir = getOutDir();
+  await import(pathToFileURL(join(outDir, 'steps.mjs')).href);
+  return import(pathToFileURL(join(outDir, 'workflows.mjs')).href);
+}
+
 /**
  * Controller that handles the well-known workflow endpoints.
  * Dynamically imports the generated bundles and handles request/response conversion.
@@ -99,12 +105,7 @@ function getOutDir(): string {
 export class WorkflowController {
   @Post('flow')
   async handleFlow(@Req() req: any, @Res() res: any) {
-    const outDir = getOutDir();
-    // Import step registrations (side effects) before the combined handler
-    await import(pathToFileURL(join(outDir, 'steps.mjs')).href);
-    const { POST } = await import(
-      pathToFileURL(join(outDir, 'workflows.mjs')).href
-    );
+    const { POST } = await loadWorkflowHandler();
     const webRequest = toWebRequest(req);
     const webResponse = await POST(webRequest);
     await sendWebResponse(res, webResponse);

packages/nest/src/workflow.module.ts

@@ -9,6 +9,7 @@ import { join } from 'pathe';
 import type { NestBuilderOptions } from './builder.js';
 import {
   configureWorkflowController,
+  loadWorkflowHandler,
   WorkflowController,
 } from './workflow.controller.js';
 
@@ -86,19 +87,23 @@ export class WorkflowModule implements OnModuleInit, OnModuleDestroy {
 
   async onModuleInit() {
     const options = WorkflowModule.state.options;
-    if (!options || options.skipBuild) {
-      return;
+    const outDir = WorkflowModule.state.outDir;
+    if (!options || !outDir) return;
+
+    if (!options.skipBuild) {
+      // Lazy-load the toolchain so it never enters the runtime bundle.
+      const [{ NestLocalBuilder }, { createBuildQueue }] = await Promise.all([
+        import('./builder.js'),
+        import('@workflow/builders'),
+      ]);
+      const builder = new NestLocalBuilder({ ...options, outDir });
+      await createBuildQueue()(() => builder.build());
+    }
+
+    if (process.env.WORKFLOW_TARGET_WORLD === '@workflow/world-postgres') {
+      const { POST } = await loadWorkflowHandler();
+      await POST.initialize();
     }
-    // Lazy-load the toolchain so it never enters the runtime bundle.
-    const [{ NestLocalBuilder }, { createBuildQueue }] = await Promise.all([
-      import('./builder.js'),
-      import('@workflow/builders'),
-    ]);
-    const builder = new NestLocalBuilder({
-      ...options,
-      outDir: WorkflowModule.state.outDir ?? undefined,
-    });
-    await createBuildQueue()(() => builder.build());
   }
 
   async onModuleDestroy() {

packages/next/src/index.test.ts

@@ -144,6 +144,30 @@ describe('withWorkflow builder config', () => {
     expect(prewarmWorkflowSwcPluginCacheMock).not.toHaveBeenCalled();
   });
 
+  it('inlines the selected World during development', async () => {
+    process.env.WORKFLOW_TARGET_WORLD = '@workflow/world-postgres';
+    const config = withWorkflow({ env: { EXISTING: 'value' } });
+
+    const nextConfig = await config('phase-development-server', {
+      defaultConfig: {},
+    });
+
+    expect(nextConfig.env).toEqual({
+      EXISTING: 'value',
+      WORKFLOW_TARGET_WORLD: '@workflow/world-postgres',
+    });
+  });
+
+  it('keeps World selection dynamic in production', async () => {
+    const config = withWorkflow({ env: { EXISTING: 'value' } });
+
+    const nextConfig = await config('phase-production-build', {
+      defaultConfig: {},
+    });
+
+    expect(nextConfig.env).toEqual({ EXISTING: 'value' });
+  });
+
   it('configures diagnostics inside the default Next.js dist dir', async () => {
     const config = withWorkflow({});
 

packages/next/src/index.ts

@@ -488,6 +488,12 @@ export function withWorkflow(
     }
     // shallow clone to avoid read-only on top-level
     nextConfig = Object.assign({}, nextConfig);
+    if (phase === 'phase-development-server') {
+      nextConfig.env = {
+        ...nextConfig.env,
+        WORKFLOW_TARGET_WORLD: process.env.WORKFLOW_TARGET_WORLD,
+      };
+    }
     const workflowBasePath = nextConfig.basePath;
     setWorkflowBasePath(workflowBasePath);
     nextConfig.serverExternalPackages = [

packages/nitro/src/index.ts

@@ -476,6 +476,18 @@ function addVirtualHandler(
     'workflow/webhook.mjs': '',
     'workflow/workflows.mjs': `await import(/* @vite-ignore */ pathToFileURL(${stepsImportPath}).href + "?t=" + version);`,
   };
+  const initializeWorkflow: Record<VirtualHandlerPath, string> = {
+    'workflow/webhook.mjs': '',
+    'workflow/workflows.mjs': `
+      export async function initializeWorkflow() {
+        await (await loadHandler()).POST.initialize();
+      }`,
+  };
+  const handlerImports: Record<VirtualHandlerPath, string> = {
+    'workflow/webhook.mjs': `import { POST } from ${handlerImportPath};`,
+    'workflow/workflows.mjs': `import { POST } from ${handlerImportPath};
+    export const initializeWorkflow = POST.initialize;`,
+  };
 
   if (nitro.options.dev) {
     // Dev mode: load generated workflow bundles from disk at request time.
@@ -492,19 +504,21 @@ function addVirtualHandler(
       let currentVersion = "";
       let currentImportPath = "";
 
-      async function loadPOST() {
+      async function loadHandler() {
         const version = String(statSync(handlerPath).mtimeMs);
         if (version !== currentVersion) {
           currentVersion = version;
           currentImportPath = pathToFileURL(handlerPath).href + "?t=" + version;
           ${preloadSteps[buildPath]}
         }
-        return (await import(currentImportPath)).POST;
+        return import(currentImportPath);
       }
 
+      ${initializeWorkflow[buildPath]}
+
       export default fromWebHandler(async (request, context) => {
-        const POST = await loadPOST();
-        return POST(request, context);
+        const handler = await loadHandler();
+        return handler.POST(request, context);
       });
     `;
     } else {
@@ -516,20 +530,22 @@ function addVirtualHandler(
       let currentVersion = "";
       let currentImportPath = "";
 
-      async function loadPOST() {
+      async function loadHandler() {
         const version = String(statSync(handlerPath).mtimeMs);
         if (version !== currentVersion) {
           currentVersion = version;
           currentImportPath = pathToFileURL(handlerPath).href + "?t=" + version;
           ${preloadSteps[buildPath]}
         }
-        return (await import(currentImportPath)).POST;
+        return import(currentImportPath);
       }
 
+      ${initializeWorkflow[buildPath]}
+
       export default async ({ req }) => {
         try {
-          const POST = await loadPOST();
-          return await POST(req);
+          const handler = await loadHandler();
+          return await handler.POST(req);
         } catch (error) {
           console.error('Handler error:', error);
           return new Response('Internal Server Error', { status: 500 });
@@ -550,14 +566,14 @@ function addVirtualHandler(
     nitro.options.virtual[`#${buildPath}`] = /* js */ `
     import ${handlerImportPath};
     import { fromWebHandler } from "h3";
-    import { POST } from ${handlerImportPath};
+    ${handlerImports[buildPath]}
     export default fromWebHandler(POST);
   `;
   } else {
     // Nitro v3+ (native web handlers)
     nitro.options.virtual[`#${buildPath}`] = /* js */ `
     import ${handlerImportPath};
-    import { POST } from ${handlerImportPath};
+    ${handlerImports[buildPath]}
     export default async ({ req }) => {
       try {
         return await POST(req);

packages/sveltekit/src/builder.ts

@@ -94,10 +94,11 @@ export class SvelteKitBuilder extends BaseBuilder {
     workflowsRouteContent = workflowsRouteContent.replace(
       /export const POST = workflowEntrypoint\(workflowCode(?<options>[^)]*)\);?$/m,
       (_match, options = '') => `${NORMALIZE_REQUEST_CODE}
-export const POST = async ({request}) => {
+const workflowHandler = workflowEntrypoint(workflowCode${options});
+export const POST = Object.assign(async ({request}) => {
   const normalRequest = await normalizeRequest(request);
-  return workflowEntrypoint(workflowCode${options})(normalRequest);
-}`
+  return workflowHandler(normalRequest);
+}, { initialize: workflowHandler.initialize });`
     );
     // These files are generated into the SvelteKit routes directory, which
     // Vite watches in dev, so an identical rewrite would still invalidate the

packages/world-postgres/package.json

@@ -46,11 +46,9 @@
     "db:push": "node dist/cli.js"
   },
   "dependencies": {
-    "@vercel/queue": "catalog:",
     "@workflow/errors": "workspace:*",
     "@workflow/utils": "workspace:*",
     "@workflow/world": "workspace:*",
-    "@workflow/world-local": "workspace:*",
     "cbor-x": "1.6.0",
     "dotenv": "17.3.1",
     "drizzle-orm": "0.45.2",

packages/world-postgres/src/index.ts

@@ -75,6 +75,7 @@ export function createWorld(
   return {
     specVersion: mintedSpecVersion(),
     capabilities: {
+      directQueueDelivery: true,
       hookRetention: { active: true },
     },
     ...storage,