From 239031ad9e1d27942f8e30a59fd6fef254544fff Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Mon, 6 Jul 2026 16:43:35 -0700 Subject: [PATCH] fix(next): respect basePath for workflow routes (#2732) * fix(next): respect basePath for workflow routes * docs(core): note workflow URL resolution gap * fix(next): expose workflow health route methods * test(utils): remove workflow route helper tests * test(builders): remove route handler string test * fix(next): defer basePath validation to Next.js * refactor(utils): remove workflow url helper wrappers * Test Next basePath builder wiring --- .changeset/next-base-path.md | 10 ++ packages/builders/src/base-builder.ts | 14 ++- packages/builders/src/constants.test.ts | 3 +- packages/builders/src/constants.ts | 14 +++ packages/builders/src/types.ts | 3 + packages/core/e2e/e2e.test.ts | 124 +++++++-------------- packages/core/e2e/utils.ts | 3 +- packages/core/src/create-hook.ts | 4 +- packages/core/src/runtime.ts | 9 +- packages/core/src/runtime/step-executor.ts | 16 ++- packages/core/src/runtime/step-handler.ts | 21 +++- packages/core/src/workflow.ts | 16 ++- packages/core/src/workflow/create-hook.ts | 11 +- packages/next/src/index.test.ts | 16 ++- packages/next/src/index.ts | 11 ++ packages/utils/src/get-port.ts | 3 +- packages/utils/src/index.ts | 8 ++ packages/utils/src/workflow-routes.ts | 54 +++++++++ packages/world-local/src/config.test.ts | 16 +++ packages/world-local/src/config.ts | 22 +++- packages/world-local/src/queue.test.ts | 22 +++- packages/world-local/src/queue.ts | 15 ++- packages/world-postgres/src/queue.test.ts | 34 ++++++ packages/world-postgres/src/queue.ts | 17 ++- 24 files changed, 327 insertions(+), 139 deletions(-) create mode 100644 .changeset/next-base-path.md create mode 100644 packages/utils/src/workflow-routes.ts diff --git a/.changeset/next-base-path.md b/.changeset/next-base-path.md new file mode 100644 index 0000000000..1aa840d488 --- /dev/null +++ b/.changeset/next-base-path.md @@ -0,0 +1,10 @@ +--- +"@workflow/builders": patch +"@workflow/core": patch +"@workflow/next": patch +"@workflow/utils": patch +"@workflow/world-local": patch +"@workflow/world-postgres": patch +--- + +Respect framework base paths when routing workflow traffic and expose health checks on generated Next.js workflow routes. diff --git a/packages/builders/src/base-builder.ts b/packages/builders/src/base-builder.ts index 61921f4564..ef070b0f26 100644 --- a/packages/builders/src/base-builder.ts +++ b/packages/builders/src/base-builder.ts @@ -23,7 +23,10 @@ import { applySwcTransform, type WorkflowManifest, } from './apply-swc-transform.js'; -import { createWorkflowEntrypointOptionsCode } from './constants.js'; +import { + createWorkflowEntrypointOptionsCode, + createWorkflowRouteHandlersCode, +} from './constants.js'; import { getEsbuildTsconfigOptions } from './esbuild-tsconfig.js'; import { type DiscoveredEntries, @@ -1564,6 +1567,7 @@ export const __steps_registered = true; const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode( { + basePath: this.config.basePath, routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', } ); @@ -1578,7 +1582,7 @@ import { workflowEntrypoint } from 'workflow/runtime'; const workflowRouteModuleBodyStartedAt = Date.now(); const workflowCode = \`${workflowBundleCode.replace(/[\\`$]/g, '\\$&')}\`; -export const POST = workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode});`; +${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; // we skip the final bundling step for Next.js so it can bundle itself if (!bundleFinalOutput) { @@ -1771,6 +1775,7 @@ export const POST = workflowEntrypoint(workflowCode${workflowEntrypointOptionsCo const stepsRelativePath = `./${basename(stepsOutfile).replace(/\\/g, '/')}`; const escapedVMCode = workflowVMCode.replace(/[\\`$]/g, '\\$&'); const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode({ + basePath: this.config.basePath, routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', }); @@ -1786,7 +1791,7 @@ void __steps_registered; const workflowCode = \`${escapedVMCode}\`; -export const POST = workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode});`; +${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; if (!bundleFinalOutput) { await this.writeGeneratedFile(flowOutfile, combinedFunctionCode); @@ -1849,6 +1854,7 @@ export const POST = workflowEntrypoint(workflowCode${workflowEntrypointOptionsCo const escaped = interimBundleText.replace(/[\\`$]/g, '\\$&'); const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode( { + basePath: this.config.basePath, routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', } ); @@ -1864,7 +1870,7 @@ void __steps_registered; const workflowCode = \`${escaped}\`; -export const POST = workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode});`; +${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; const outputDir = dirname(flowOutfile); await mkdir(outputDir, { recursive: true }); diff --git a/packages/builders/src/constants.test.ts b/packages/builders/src/constants.test.ts index 3e086100f9..d257434428 100644 --- a/packages/builders/src/constants.test.ts +++ b/packages/builders/src/constants.test.ts @@ -53,10 +53,11 @@ describe('createWorkflowEntrypointOptionsCode', () => { expect( createWorkflowEntrypointOptionsCode({ namespace: 'custom', + basePath: '/v2', routeModuleBodyStartedAt: 'workflowRouteModuleBodyStartedAt', }) ).toBe( - ', { namespace: "custom", routeModuleBodyStartedAt: workflowRouteModuleBodyStartedAt }' + ', { namespace: "custom", basePath: "/v2", routeModuleBodyStartedAt: workflowRouteModuleBodyStartedAt }' ); }); }); diff --git a/packages/builders/src/constants.ts b/packages/builders/src/constants.ts index 7d063890e8..ac087ba726 100644 --- a/packages/builders/src/constants.ts +++ b/packages/builders/src/constants.ts @@ -54,6 +54,7 @@ export function createWorkflowQueueTrigger(options?: { namespace?: string }) { */ export function createWorkflowEntrypointOptionsCode(options?: { namespace?: string; + basePath?: string; /** Raw code identifier/expression emitted into generated route files, not data. */ routeModuleBodyStartedAt?: string; }) { @@ -66,6 +67,10 @@ export function createWorkflowEntrypointOptionsCode(options?: { fields.push(`namespace: ${JSON.stringify(namespace)}`); } + if (options?.basePath !== undefined) { + fields.push(`basePath: ${JSON.stringify(options.basePath)}`); + } + if (options?.routeModuleBodyStartedAt) { fields.push( `routeModuleBodyStartedAt: ${options.routeModuleBodyStartedAt}` @@ -79,6 +84,15 @@ export function createWorkflowEntrypointOptionsCode(options?: { return `, { ${fields.join(', ')} }`; } +export function createWorkflowRouteHandlersCode( + workflowEntrypointCall: string +) { + return `export const POST = ${workflowEntrypointCall}; +export const GET = POST; +export const HEAD = POST; +export const OPTIONS = POST;`; +} + /** * Default queue trigger (no namespace). Backward compatible. */ diff --git a/packages/builders/src/types.ts b/packages/builders/src/types.ts index 74fcd3b885..d0d65998df 100644 --- a/packages/builders/src/types.ts +++ b/packages/builders/src/types.ts @@ -60,6 +60,9 @@ interface BaseWorkflowConfig { // artifact locations. distDir?: string; + // Optional route prefix for apps deployed below the origin root. + basePath?: string; + // Suppress informational logs emitted by createWorkflowsBundle() // (e.g. intermediate/final workflow bundle timing logs). suppressCreateWorkflowsBundleLogs?: boolean; diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 7c1e909879..6db2fca8c7 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -10,6 +10,7 @@ import { WorkflowRunFailedError, WorkflowWorldError, } from '@workflow/errors'; +import { createWorkflowUrl } from '@workflow/utils'; import { SPEC_VERSION_CURRENT, type World } from '@workflow/world'; import { afterAll, @@ -111,6 +112,9 @@ function writeE2EMetadata() { const e2e = (fn: string) => getWorkflowMetadata(deploymentUrl, 'workflows/99_e2e.ts', fn); +const workflowWebhookUrl = (token: string) => + createWorkflowUrl(deploymentUrl, { type: 'webhook', token }); + type WorkflowEvent = Awaited< ReturnType >['data'][number]; @@ -532,17 +536,11 @@ describe('e2e', () => { expect(hook.runId).toBe(run.runId); // Attempt to resume via the public webhook endpoint — should get 404 - const res = await fetch( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token)}`, - deploymentUrl - ), - { - method: 'POST', - headers: await getTrustedSourcesHeaders(), - body: JSON.stringify({ message: 'should-be-rejected' }), - } - ); + const res = await fetch(workflowWebhookUrl(token), { + method: 'POST', + headers: await getTrustedSourcesHeaders(), + body: JSON.stringify({ message: 'should-be-rejected' }), + }); expect(res.status).toBe(404); // Now resume via server-side resumeHook() — should work @@ -587,79 +585,46 @@ describe('e2e', () => { const [token, token2, token3] = hooks.map((h) => h.token); // Webhook with default response - const res = await fetch( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token)}`, - deploymentUrl - ), - { - method: 'POST', - headers: await getTrustedSourcesHeaders(), - body: JSON.stringify({ message: 'one' }), - } - ); + const res = await fetch(workflowWebhookUrl(token), { + method: 'POST', + headers: await getTrustedSourcesHeaders(), + body: JSON.stringify({ message: 'one' }), + }); expect(res.status).toBe(202); const body = await res.text(); expect(body).toBe(''); // Webhook with static response - const res2 = await fetch( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token2)}`, - deploymentUrl - ), - { - method: 'POST', - headers: await getTrustedSourcesHeaders(), - body: JSON.stringify({ message: 'two' }), - } - ); + const res2 = await fetch(workflowWebhookUrl(token2), { + method: 'POST', + headers: await getTrustedSourcesHeaders(), + body: JSON.stringify({ message: 'two' }), + }); expect(res2.status).toBe(402); const body2 = await res2.text(); expect(body2).toBe('Hello from static response!'); // Webhook with manual response - const res3 = await fetch( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token3)}`, - deploymentUrl - ), - { - method: 'POST', - headers: await getTrustedSourcesHeaders(), - body: JSON.stringify({ message: 'three' }), - } - ); + const res3 = await fetch(workflowWebhookUrl(token3), { + method: 'POST', + headers: await getTrustedSourcesHeaders(), + body: JSON.stringify({ message: 'three' }), + }); expect(res3.status).toBe(200); const body3 = await res3.text(); expect(body3).toBe('Hello from webhook!'); const returnValue = await run.returnValue; expect(returnValue).toHaveLength(3); - expect(returnValue[0].url).toBe( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token)}`, - deploymentUrl - ).href - ); + expect(returnValue[0].url).toBe(workflowWebhookUrl(token)); expect(returnValue[0].method).toBe('POST'); expect(returnValue[0].body).toBe('{"message":"one"}'); - expect(returnValue[1].url).toBe( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token2)}`, - deploymentUrl - ).href - ); + expect(returnValue[1].url).toBe(workflowWebhookUrl(token2)); expect(returnValue[1].method).toBe('POST'); expect(returnValue[1].body).toBe('{"message":"two"}'); - expect(returnValue[2].url).toBe( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(token3)}`, - deploymentUrl - ).href - ); + expect(returnValue[2].url).toBe(workflowWebhookUrl(token3)); expect(returnValue[2].method).toBe('POST'); expect(returnValue[2].body).toBe('{"message":"three"}'); }); @@ -730,17 +695,11 @@ describe('e2e', () => { continue; } - const res = await fetch( - new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent(unservedHook.token)}`, - deploymentUrl - ), - { - method: 'POST', - headers: await getTrustedSourcesHeaders(), - body: `body-${unservedHook.token}`, - } - ); + const res = await fetch(workflowWebhookUrl(unservedHook.token), { + method: 'POST', + headers: await getTrustedSourcesHeaders(), + body: `body-${unservedHook.token}`, + }); expect(res.status).toBe(202); servedTokens.add(unservedHook.token); } @@ -786,11 +745,7 @@ describe('e2e', () => { ); test('webhook route with invalid token', { timeout: 60_000 }, async () => { - const invalidWebhookUrl = new URL( - `/.well-known/workflow/v1/webhook/${encodeURIComponent('invalid')}`, - deploymentUrl - ); - const res = await fetch(invalidWebhookUrl, { + const res = await fetch(workflowWebhookUrl('invalid'), { method: 'POST', headers: await getTrustedSourcesHeaders(), body: JSON.stringify({}), @@ -2555,14 +2510,13 @@ describe('e2e', () => { // bypasses protection by sending messages through the Queue infrastructure. // Test the flow endpoint health check (V2: combined handler for both workflow + step) - const flowHealthUrl = new URL( - '/.well-known/workflow/v1/flow?__health', - deploymentUrl + const flowRes = await fetch( + createWorkflowUrl(deploymentUrl, { type: 'health' }), + { + method: 'POST', + headers: await getTrustedSourcesHeaders(), + } ); - const flowRes = await fetch(flowHealthUrl, { - method: 'POST', - headers: await getTrustedSourcesHeaders(), - }); expect(flowRes.status).toBe(200); expect(flowRes.headers.get('Content-Type')).toBe('application/json'); const flowBody = await flowRes.json(); diff --git a/packages/core/e2e/utils.ts b/packages/core/e2e/utils.ts index a2fe1b207c..064e169021 100644 --- a/packages/core/e2e/utils.ts +++ b/packages/core/e2e/utils.ts @@ -3,6 +3,7 @@ import fs from 'node:fs'; import path, { dirname } from 'node:path'; import { setTimeout as sleep } from 'node:timers/promises'; import { fileURLToPath } from 'node:url'; +import { createWorkflowUrl } from '@workflow/utils'; import { createVercelWorld } from '@workflow/world-vercel'; import { onTestFailed } from 'vitest'; import { getTrustedSourcesHeaders } from '../../../scripts/trusted-sources-headers.mjs'; @@ -373,7 +374,7 @@ export async function fetchManifest( const forceRefresh = options?.forceRefresh ?? false; if (cachedManifest && !forceRefresh) return cachedManifest; - const url = new URL('/.well-known/workflow/v1/manifest.json', deploymentUrl); + const url = createWorkflowUrl(deploymentUrl, { type: 'manifest' }); const res = await fetch(url, { headers: await getTrustedSourcesHeaders(), }); diff --git a/packages/core/src/create-hook.ts b/packages/core/src/create-hook.ts index 108fc205ea..c44daafbf2 100644 --- a/packages/core/src/create-hook.ts +++ b/packages/core/src/create-hook.ts @@ -167,8 +167,8 @@ export interface HookOptions { * Whether this hook can be resumed via the public webhook endpoint. * * When `true`, the hook can be triggered by sending an HTTP request to the - * public `/.well-known/workflow/v1/webhook/{token}` URL. This is automatically - * set when using `createWebhook()`. + * public workflow webhook URL. This is automatically set when using + * `createWebhook()`. * * When `false` (the default), the hook can only be resumed server-side * via `resumeHook()`. diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index b57468862a..0ae0283271 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -9,6 +9,7 @@ import { RunExpiredError, WorkflowRuntimeError, } from '@workflow/errors'; +import { setWorkflowBasePath } from '@workflow/utils'; import { parseWorkflowName, workflowDisplayName, @@ -293,8 +294,14 @@ function hasOpenHookOrWait(events: Event[]): boolean { */ export function workflowEntrypoint( workflowCode: string, - options?: { namespace?: string; routeModuleBodyStartedAt?: number } + options?: { + namespace?: string; + routeModuleBodyStartedAt?: number; + basePath?: string; + } ): (req: Request) => Promise { + setWorkflowBasePath(options?.basePath); + const NO_INLINE_REPLAY_AFTER_MS = Number(process.env.WORKFLOW_V2_TIMEOUT_MS) || 120_000; diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index 7f21bba0fc..130a73fce9 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -8,7 +8,11 @@ import { TooEarlyError, WorkflowRuntimeError, } from '@workflow/errors'; -import { pluralize, stepDisplayName } from '@workflow/utils'; +import { + createWorkflowBaseUrl, + pluralize, + stepDisplayName, +} from '@workflow/utils'; import type { Event, SerializedData, Step, World } from '@workflow/world'; import { SPEC_VERSION_CURRENT, @@ -563,7 +567,11 @@ export async function executeStep( const args = hydratedInput.args; const thisVal = hydratedInput.thisVal ?? null; - const port = isVercel ? undefined : await getPortLazy(); + const workflowBaseUrl = createWorkflowBaseUrl( + isVercel + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:${(await getPortLazy()) ?? 3000}` + ); const executionStartTime = Date.now(); result = await trace('step.execute', {}, async () => { @@ -579,9 +587,7 @@ export async function executeStep( workflowName, workflowRunId, workflowStartedAt: new Date(+workflowStartedAt), - url: isVercel - ? `https://${process.env.VERCEL_URL}` - : `http://localhost:${port ?? 3000}`, + url: workflowBaseUrl, features: { encryption: !!encryptionKey }, }, workflowDeploymentId: params.workflowDeploymentId, diff --git a/packages/core/src/runtime/step-handler.ts b/packages/core/src/runtime/step-handler.ts index 228a2740ca..adbcd6e495 100644 --- a/packages/core/src/runtime/step-handler.ts +++ b/packages/core/src/runtime/step-handler.ts @@ -9,7 +9,12 @@ import { WorkflowRuntimeError, WorkflowWorldError, } from '@workflow/errors'; -import { formatStepName, pluralize, stepDisplayName } from '@workflow/utils'; +import { + createWorkflowBaseUrl, + formatStepName, + pluralize, + stepDisplayName, +} from '@workflow/utils'; import { getPort } from '@workflow/utils/get-port'; import { getQueueTopicPrefix, @@ -223,6 +228,14 @@ function createStepHandler(namespace?: string) { getSpanKind('CONSUMER'), ]); + // TODO: resolve the workflow base URL through the World interface. + // This fallback cannot see local/Postgres baseUrl overrides or custom-world routing. + const workflowBaseUrl = createWorkflowBaseUrl( + isVercel + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:${port ?? 3000}` + ); + return trace( `step.execute ${stepDisplayName(stepName)}`, { kind: spanKind, links: spanLinks }, @@ -667,11 +680,7 @@ function createStepHandler(namespace?: string) { workflowName, workflowRunId, workflowStartedAt: new Date(+workflowStartedAt), - // TODO: there should be a getUrl method on the world interface itself. This - // solution only works for vercel + local worlds. - url: isVercel - ? `https://${process.env.VERCEL_URL}` - : `http://localhost:${port ?? 3000}`, + url: workflowBaseUrl, features: { encryption: !!encryptionKey }, }, workflowDeploymentId: process.env.VERCEL_DEPLOYMENT_ID, diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index af78708ceb..4d8a138dcc 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -4,7 +4,7 @@ import { WorkflowNotRegisteredError, WorkflowRuntimeError, } from '@workflow/errors'; -import { withResolvers } from '@workflow/utils'; +import { createWorkflowBaseUrl, withResolvers } from '@workflow/utils'; import { parseWorkflowName } from '@workflow/utils/parse-name'; import type { Event, WorkflowRun } from '@workflow/world'; import { SPEC_VERSION_SUPPORTS_COMPRESSION } from '@workflow/world'; @@ -194,7 +194,11 @@ export async function runWorkflow( // fs ops (readdir, readFile) into the flow route bundle. The resolved // port is cached per process (see get-port-lazy.ts), so this is cheap // on replays after the first. - const port = isVercel ? undefined : await getPortLazy(); + const workflowBaseUrl = createWorkflowBaseUrl( + isVercel + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:${(await getPortLazy()) ?? 3000}` + ); const { context, @@ -311,18 +315,12 @@ export async function runWorkflow( vmGlobalThis[WORKFLOW_GET_STREAM_ID] = (namespace?: string) => getWorkflowRunStreamId(workflowRun.runId, namespace); - // TODO: there should be a getUrl method on the world interface itself. This - // solution only works for vercel + local worlds. - const url = isVercel - ? `https://${process.env.VERCEL_URL}` - : `http://localhost:${port ?? 3000}`; - // For the workflow VM, we store the context in a symbol on the `globalThis` object const ctx: WorkflowMetadata = { workflowName: workflowRun.workflowName, workflowRunId: workflowRun.runId, workflowStartedAt: new vmGlobalThis.Date(+startedAt), - url, + url: workflowBaseUrl, features: { encryption: !!encryptionKey }, }; diff --git a/packages/core/src/workflow/create-hook.ts b/packages/core/src/workflow/create-hook.ts index c56bcd4834..96e31d1b9f 100644 --- a/packages/core/src/workflow/create-hook.ts +++ b/packages/core/src/workflow/create-hook.ts @@ -1,3 +1,8 @@ +import { createWorkflowUrl } from '@workflow/utils'; +import { + aliasSerializationClass, + RUN_CLASS_ID, +} from '../class-serialization.js'; import { throwNotInWorkflowContext } from '../context-errors.js'; import type { Hook, @@ -6,10 +11,6 @@ import type { Webhook, WebhookOptions, } from '../create-hook.js'; -import { - aliasSerializationClass, - RUN_CLASS_ID, -} from '../class-serialization.js'; import { Run } from '../runtime/run.js'; import { WORKFLOW_CREATE_HOOK } from '../symbols.js'; import { getWorkflowMetadata } from './get-workflow-metadata.js'; @@ -75,7 +76,7 @@ export function createWebhook( | Webhook; const { url } = getWorkflowMetadata(); - hook.url = `${url}/.well-known/workflow/v1/webhook/${encodeURIComponent(hook.token)}`; + hook.url = createWorkflowUrl(url, { type: 'webhook', token: hook.token }); return hook; } diff --git a/packages/next/src/index.test.ts b/packages/next/src/index.test.ts index 229965962b..15a95eaeb6 100644 --- a/packages/next/src/index.test.ts +++ b/packages/next/src/index.test.ts @@ -71,7 +71,6 @@ describe('withWorkflow builder config', () => { builderConfigs.length = 0; getNextBuilderMock.mockClear(); prewarmWorkflowSwcPluginCacheMock.mockClear(); - if (!hadLoaderStub) { writeFileSync(loaderStubPath, 'module.exports = {};\n', 'utf-8'); } @@ -173,6 +172,21 @@ describe('withWorkflow builder config', () => { }); }); + it('passes Next basePath to the workflow builder', async () => { + const config = withWorkflow({ + basePath: '/v2', + }); + + await config('phase-production-build', { + defaultConfig: {}, + }); + + expect(builderConfigs).toHaveLength(1); + expect(builderConfigs[0]).toMatchObject({ + basePath: '/v2', + }); + }); + it('externalizes the built-in Vercel world while preserving user externals', async () => { const config = withWorkflow({ serverExternalPackages: ['@node-rs/xxhash'], diff --git a/packages/next/src/index.ts b/packages/next/src/index.ts index 8d7e44202b..73f85b7cf8 100644 --- a/packages/next/src/index.ts +++ b/packages/next/src/index.ts @@ -41,6 +41,14 @@ const workflowSerdeComputedPropertyPattern = const PSEUDO_EXTERNAL_PACKAGES = new Set(['server-only', 'client-only']); const warnedAutoRemovedServerExternalPackages = new Set(); +const BASE_PATH_SYMBOL = Symbol.for('@workflow/core/basePath'); +const globalConfig = globalThis as typeof globalThis & + Record; + +// Keep this local: @workflow/next is CommonJS, while @workflow/utils is ESM-only. +function setWorkflowBasePath(basePath: string | undefined): void { + globalConfig[BASE_PATH_SYMBOL] = basePath ?? ''; +} interface WorkflowPatternMatch { hasUseWorkflow: boolean; @@ -386,6 +394,8 @@ export function withWorkflow( } // shallow clone to avoid read-only on top-level nextConfig = Object.assign({}, nextConfig); + const workflowBasePath = nextConfig.basePath; + setWorkflowBasePath(workflowBasePath); nextConfig.env = { ...nextConfig.env, WORKFLOW_TARGET_WORLD: workflowTargetWorld, @@ -516,6 +526,7 @@ export function withWorkflow( moduleSpecifierRoot: workingDir, workingDir, distDir, + basePath: workflowBasePath, diagnosticsDir: `${distDir}/diagnostics`, buildTarget: 'next', workflowsBundlePath: '', // not used in base diff --git a/packages/utils/src/get-port.ts b/packages/utils/src/get-port.ts index 0903ab9ef3..d7c44b1005 100644 --- a/packages/utils/src/get-port.ts +++ b/packages/utils/src/get-port.ts @@ -2,6 +2,7 @@ import { execFile } from 'node:child_process'; import { readdir, readFile, readlink } from 'node:fs/promises'; import { promisify } from 'node:util'; import { parseWindowsNetstatPortsForPid } from './get-port-internals.js'; +import { WORKFLOW_ROUTE_BASE } from './workflow-routes.js'; const execFileAsync = promisify(execFile); @@ -215,7 +216,7 @@ export async function getPort(): Promise { // Configuration for HTTP probing const PROBE_TIMEOUT_MS = 500; -const PROBE_ENDPOINT = '/.well-known/workflow/v1/flow?__health'; +const PROBE_ENDPOINT = `${WORKFLOW_ROUTE_BASE}/flow?__health`; export interface ProbeOptions { endpoint?: string; diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts index ab5beecb1a..55b883452b 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -10,6 +10,14 @@ export { export { pluralize } from './pluralize.js'; export { once, type PromiseWithResolvers, withResolvers } from './promise.js'; export { parseDurationToDate } from './time.js'; +export { + createWorkflowBaseUrl, + createWorkflowHealthEndpoint, + createWorkflowUrl, + setWorkflowBasePath, + type WorkflowUrlRoute, + WORKFLOW_ROUTE_BASE, +} from './workflow-routes.js'; export { getWorldImport, isVercelWorldTarget, diff --git a/packages/utils/src/workflow-routes.ts b/packages/utils/src/workflow-routes.ts new file mode 100644 index 0000000000..2ecd214553 --- /dev/null +++ b/packages/utils/src/workflow-routes.ts @@ -0,0 +1,54 @@ +export const WORKFLOW_ROUTE_BASE = '/.well-known/workflow/v1'; +const BASE_PATH_SYMBOL = Symbol.for('@workflow/core/basePath'); +const globalConfig = globalThis as typeof globalThis & + Record; + +export function setWorkflowBasePath(basePath: string | undefined): void { + globalConfig[BASE_PATH_SYMBOL] = basePath ?? ''; +} + +function getWorkflowBasePath(): string { + return globalConfig[BASE_PATH_SYMBOL] ?? ''; +} + +export function createWorkflowBaseUrl(origin: string): string { + new URL(origin); + return `${origin.replace(/[?#].*$/, '').replace(/\/+$/, '')}${getWorkflowBasePath()}`; +} + +export type WorkflowUrlRoute = + | { type: 'flow' | 'step' } + | { type: 'manifest' } + | { type: 'webhook'; token: string } + | { type: 'health' }; + +export function createWorkflowUrl( + baseUrl: string, + route: WorkflowUrlRoute +): string { + const url = new URL(baseUrl); + url.pathname = `${url.pathname.replace(/\/+$/, '')}${WORKFLOW_ROUTE_BASE}/${getWorkflowRouteEndpoint(route)}`; + url.search = route.type === 'health' ? '__health' : ''; + url.hash = ''; + return url.toString(); +} + +function getWorkflowRouteEndpoint(route: WorkflowUrlRoute): string { + switch (route.type) { + case 'flow': + case 'health': + return 'flow'; + case 'step': + return 'step'; + case 'manifest': + return 'manifest.json'; + case 'webhook': + return `webhook/${encodeURIComponent(route.token)}`; + } + const exhaustive: never = route; + return exhaustive; +} + +export function createWorkflowHealthEndpoint(): string { + return `${getWorkflowBasePath()}${WORKFLOW_ROUTE_BASE}/flow?__health`; +} diff --git a/packages/world-local/src/config.test.ts b/packages/world-local/src/config.test.ts index 3bc9a5b779..8f5ec7bb5d 100644 --- a/packages/world-local/src/config.test.ts +++ b/packages/world-local/src/config.test.ts @@ -1,3 +1,4 @@ +import { setWorkflowBasePath } from '@workflow/utils'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { resolveBaseUrl } from './config'; @@ -15,6 +16,7 @@ describe('resolveBaseUrl', () => { afterEach(() => { process.env = originalEnv; + setWorkflowBasePath(undefined); vi.clearAllMocks(); }); @@ -182,6 +184,20 @@ describe('resolveBaseUrl', () => { expect(result).toBe('http://localhost:3000'); }); + it('should probe and return local URLs under the workflow base path', async () => { + const { getWorkflowPort } = await import('@workflow/utils/get-port'); + vi.mocked(getWorkflowPort).mockResolvedValue(3000); + delete process.env.PORT; + setWorkflowBasePath('/v2'); + + const result = await resolveBaseUrl({}); + + expect(result).toBe('http://localhost:3000/v2'); + expect(getWorkflowPort).toHaveBeenCalledWith({ + endpoint: '/v2/.well-known/workflow/v1/flow?__health', + }); + }); + it('should throw error when auto-detection fails', async () => { const { getWorkflowPort } = await import('@workflow/utils/get-port'); vi.mocked(getWorkflowPort).mockResolvedValue(undefined); diff --git a/packages/world-local/src/config.ts b/packages/world-local/src/config.ts index 923bda8e01..0915b48435 100644 --- a/packages/world-local/src/config.ts +++ b/packages/world-local/src/config.ts @@ -1,3 +1,7 @@ +import { + createWorkflowBaseUrl, + createWorkflowHealthEndpoint, +} from '@workflow/utils'; import { getWorkflowPort } from '@workflow/utils/get-port'; import { once } from './util.js'; @@ -42,6 +46,14 @@ export const config = once(() => { return { dataDir, baseUrl }; }); +export function resolveDirectBaseUrl(config: Partial): string { + return ( + config.baseUrl ?? + process.env.WORKFLOW_LOCAL_BASE_URL ?? + createWorkflowBaseUrl('http://localhost') + ); +} + /** * Resolves the base URL for queue requests following the priority order: * 1. config.baseUrl (highest priority - full override from args) @@ -62,16 +74,18 @@ export async function resolveBaseUrl(config: Partial): Promise { } if (typeof config.port === 'number') { - return `http://localhost:${config.port}`; + return createWorkflowBaseUrl(`http://localhost:${config.port}`); } if (process.env.PORT) { - return `http://localhost:${process.env.PORT}`; + return createWorkflowBaseUrl(`http://localhost:${process.env.PORT}`); } - const detectedPort = await getWorkflowPort(); + const detectedPort = await getWorkflowPort({ + endpoint: createWorkflowHealthEndpoint(), + }); if (detectedPort) { - return `http://localhost:${detectedPort}`; + return createWorkflowBaseUrl(`http://localhost:${detectedPort}`); } throw new Error('Unable to resolve base URL for workflow queue.'); diff --git a/packages/world-local/src/queue.test.ts b/packages/world-local/src/queue.test.ts index 9e190e1ac7..0973413ebe 100644 --- a/packages/world-local/src/queue.test.ts +++ b/packages/world-local/src/queue.test.ts @@ -1,3 +1,4 @@ +import { setWorkflowBasePath } from '@workflow/utils'; import type { StepInvokePayload } from '@workflow/world'; import { MessageId, ValidQueueName } from '@workflow/world'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; @@ -48,6 +49,7 @@ describe('queue timeout re-enqueue', () => { afterEach(async () => { await localQueue.close(); + setWorkflowBasePath(undefined); vi.restoreAllMocks(); vi.unstubAllGlobals(); }); @@ -165,6 +167,24 @@ describe('queue timeout re-enqueue', () => { }); }); + it('uses basePath when delivering to direct in-process handlers', async () => { + await localQueue.close(); + localQueue = createQueue({}); + setWorkflowBasePath('/v2'); + const handler = vi.fn(async () => Response.json({ ok: true })); + + localQueue.registerHandler('__wkf_step_', handler); + await localQueue.queue('__wkf_step_test' as any, stepPayload); + + await vi.waitFor(() => { + expect(handler).toHaveBeenCalledTimes(1); + }); + + expect(handler.mock.calls[0]?.[0].url).toBe( + 'http://localhost/v2/.well-known/workflow/v1/step' + ); + }); + it('queue retries immediately when handler returns timeoutSeconds: 0', async () => { const { setTimeout: mockSetTimeout } = await import('node:timers/promises'); vi.mocked(mockSetTimeout).mockClear(); @@ -262,7 +282,7 @@ describe('queue delaySeconds', () => { // Real-ish sleep: never resolves, rejects with AbortError on signal // abort — mirrors node:timers/promises semantics for long delays. vi.mocked(mockSetTimeout).mockImplementationOnce( - (_delay?: number, value?: unknown, opts?: { signal?: AbortSignal }) => + (_delay?: number, _value?: unknown, opts?: { signal?: AbortSignal }) => new Promise((_resolve, reject) => { opts?.signal?.addEventListener('abort', () => { const err = new Error('The operation was aborted'); diff --git a/packages/world-local/src/queue.ts b/packages/world-local/src/queue.ts index b803fd43a3..07b6f4ce5b 100644 --- a/packages/world-local/src/queue.ts +++ b/packages/world-local/src/queue.ts @@ -1,5 +1,6 @@ import { setTimeout } from 'node:timers/promises'; import type { Transport } from '@vercel/queue'; +import { createWorkflowUrl } from '@workflow/utils'; import { MessageId, parseQueueName, @@ -12,7 +13,7 @@ import { monotonicFactory } from 'ulid'; import { Agent } from 'undici'; import { z } from 'zod/v4'; import type { Config } from './config.js'; -import { resolveBaseUrl } from './config.js'; +import { resolveBaseUrl, resolveDirectBaseUrl } from './config.js'; import { jsonReplacer, jsonReviver } from './fs.js'; import { getPackageInfo } from './init.js'; @@ -213,19 +214,17 @@ export function createQueue(config: Partial): LocalQueue { try { if (directHandler) { const req = new Request( - `http://localhost/.well-known/workflow/v1/${pathname}`, - { - method: 'POST', - headers, - body, - } + createWorkflowUrl(resolveDirectBaseUrl(config), { + type: pathname, + }), + { method: 'POST', headers, body } ); response = await directHandler(req); } else { const baseUrl = await resolveBaseUrl(config); // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici v7 dispatcher types don't match @types/node's RequestInit response = await fetch( - `${baseUrl}/.well-known/workflow/v1/${pathname}`, + createWorkflowUrl(baseUrl, { type: pathname }), { method: 'POST', duplex: 'half', diff --git a/packages/world-postgres/src/queue.test.ts b/packages/world-postgres/src/queue.test.ts index f7d975d799..9a48b0d37b 100644 --- a/packages/world-postgres/src/queue.test.ts +++ b/packages/world-postgres/src/queue.test.ts @@ -1,5 +1,6 @@ import { createServer, type Server } from 'node:http'; import { JsonTransport } from '@vercel/queue'; +import { setWorkflowBasePath } from '@workflow/utils'; import { getWorkflowPort } from '@workflow/utils/get-port'; import { MessageId, parseQueueName, type QueuePayload } from '@workflow/world'; import { createLocalWorld } from '@workflow/world-local'; @@ -80,6 +81,7 @@ describe('postgres queue http execution', () => { vi.useRealTimers(); delete process.env.WORKFLOW_LOCAL_BASE_URL; delete process.env.PORT; + setWorkflowBasePath(undefined); }); it('uses the workflow http step route when the real runtime step handler would fail in-process with Step not found', async () => { @@ -365,6 +367,38 @@ describe('postgres queue http execution', () => { } }); + it('uses basePath for local postgres queue HTTP delivery', async () => { + const fetchMock = vi.fn(async () => Response.json({ ok: true })); + vi.stubGlobal('fetch', fetchMock); + const port = await getUnusedLoopbackPort(); + await startWorkflowHttpServer([], port); + process.env.PORT = String(port); + setWorkflowBasePath('/v2'); + + const queue = buildQueue({ connectionString: 'postgres://test' }, pool); + try { + await queue.start(); + + const task = getTaskHandler('workflow_steps'); + const payload = buildMessageData('__wkf_step_test-step', { + workflowName: 'test-workflow', + workflowRunId: 'run_01ABC', + workflowStartedAt: Date.now(), + stepId: 'step_01ABC', + }); + + await expect(task(payload, {} as any)).resolves.toBeUndefined(); + + expect(fetchMock).toHaveBeenCalledWith( + `http://localhost:${port}/v2/.well-known/workflow/v1/step`, + expect.objectContaining({ method: 'POST' }) + ); + expect(getWorkflowPort).not.toHaveBeenCalled(); + } finally { + vi.unstubAllGlobals(); + } + }); + it('queues producer delays and headers in graphile job metadata', async () => { vi.useFakeTimers(); vi.setSystemTime(new Date('2024-01-01T00:00:00.000Z')); diff --git a/packages/world-postgres/src/queue.ts b/packages/world-postgres/src/queue.ts index 962f74f6d4..abad28f4a2 100644 --- a/packages/world-postgres/src/queue.ts +++ b/packages/world-postgres/src/queue.ts @@ -2,6 +2,11 @@ import { connect } from 'node:net'; import * as Stream from 'node:stream'; import { setTimeout as sleep } from 'node:timers/promises'; import type { Transport } from '@vercel/queue'; +import { + createWorkflowBaseUrl, + createWorkflowHealthEndpoint, + createWorkflowUrl, +} from '@workflow/utils'; import { getWorkflowPort } from '@workflow/utils/get-port'; import { getQueuePrefixKind, @@ -217,16 +222,18 @@ export function createQueue( } if (typeof port === 'number') { - return `http://localhost:${port}`; + return createWorkflowBaseUrl(`http://localhost:${port}`); } if (process.env.PORT) { - return `http://localhost:${process.env.PORT}`; + return createWorkflowBaseUrl(`http://localhost:${process.env.PORT}`); } - const detectedPort = await getWorkflowPort(); + const detectedPort = await getWorkflowPort({ + endpoint: createWorkflowHealthEndpoint(), + }); if (typeof detectedPort === 'number') { - return `http://localhost:${detectedPort}`; + return createWorkflowBaseUrl(`http://localhost:${detectedPort}`); } return undefined; @@ -360,7 +367,7 @@ export function createQueue( const pathname = getQueueRoute(queueName); const response = await fetch( - `${baseUrl}/.well-known/workflow/v1/${pathname}`, + createWorkflowUrl(baseUrl, { type: pathname }), { method: 'POST', duplex: 'half',