diff --git a/.changeset/next-base-path.md b/.changeset/next-base-path.md new file mode 100644 index 0000000000..1234dcb3a9 --- /dev/null +++ b/.changeset/next-base-path.md @@ -0,0 +1,12 @@ +--- +"@workflow/builders": patch +"@workflow/astro": patch +"@workflow/core": patch +"@workflow/next": patch +"@workflow/sveltekit": 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/astro/src/builder.ts b/packages/astro/src/builder.ts index 8702500f8f..8bb32ca19c 100644 --- a/packages/astro/src/builder.ts +++ b/packages/astro/src/builder.ts @@ -176,16 +176,13 @@ export const prerender = false;`, // Normalize request, needed for preserving request through astro workflowsRouteContent = replaceGeneratedRouteExport( workflowsRouteContent, - /const handler = workflowEntrypoint\(workflowCode(?[^)]*)\);\s*export const HEAD = handler;\s*export const POST = handler;?\s*$/m, + /export const POST = workflowEntrypoint\(workflowCode(?[^)]*)\);?$/m, (_match, options = '') => `${NORMALIZE_REQUEST_CODE} -const handleWorkflowRequest = async ({request}) => { +export const POST = async ({request}) => { const normalRequest = await normalizeRequest(request); return workflowEntrypoint(workflowCode${options})(normalRequest); }; -export const HEAD = handleWorkflowRequest; -export const POST = handleWorkflowRequest; - export const prerender = false;`, 'Failed to wrap generated Astro workflow route' ); diff --git a/packages/builders/src/base-builder.ts b/packages/builders/src/base-builder.ts index 5b416563d0..31f50ce81a 100644 --- a/packages/builders/src/base-builder.ts +++ b/packages/builders/src/base-builder.ts @@ -21,7 +21,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 { fastDiscoverEntries, @@ -1342,8 +1345,11 @@ export const __steps_registered = true; } } - const workflowEntrypointOptionsCode = - createWorkflowEntrypointOptionsCode(); + const workflowEntrypointOptionsCode = createWorkflowEntrypointOptionsCode( + { + basePath: this.config.basePath, + } + ); const bundleFinal = async (interimBundle: string) => { const workflowBundleCode = interimBundle; @@ -1354,10 +1360,7 @@ import { workflowEntrypoint } from 'workflow/runtime'; const workflowCode = \`${workflowBundleCode.replace(/[\\`$]/g, '\\$&')}\`; -const handler = workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode}); - -export const HEAD = handler; -export const POST = handler;`; +${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntrypointOptionsCode})`)}`; // we skip the final bundling step for Next.js so it can bundle itself if (!bundleFinalOutput) { diff --git a/packages/builders/src/constants.test.ts b/packages/builders/src/constants.test.ts index 2474a7e1a5..3363ec9b14 100644 --- a/packages/builders/src/constants.test.ts +++ b/packages/builders/src/constants.test.ts @@ -48,4 +48,13 @@ describe('createWorkflowEntrypointOptionsCode', () => { ', { namespace: "custom" }' ); }); + + it('inlines basePath alongside namespace options', () => { + expect( + createWorkflowEntrypointOptionsCode({ + namespace: 'custom', + basePath: '/v2', + }) + ).toBe(', { namespace: "custom", basePath: "/v2" }'); + }); }); diff --git a/packages/builders/src/constants.ts b/packages/builders/src/constants.ts index 3a00a368eb..6ca43d847d 100644 --- a/packages/builders/src/constants.ts +++ b/packages/builders/src/constants.ts @@ -86,17 +86,35 @@ export function createWorkflowQueueTrigger(options?: { namespace?: string }) { */ export function createWorkflowEntrypointOptionsCode(options?: { namespace?: string; + basePath?: string; }) { const namespace = resolveQueueNamespace(options?.namespace); + const fields: string[] = []; - if (!namespace) { + if (namespace) { + // Reuse prefix construction for namespace validation. + getQueueTopicPrefix('workflow', namespace); + fields.push(`namespace: ${JSON.stringify(namespace)}`); + } + + if (options?.basePath !== undefined) { + fields.push(`basePath: ${JSON.stringify(options.basePath)}`); + } + + if (fields.length === 0) { return ''; } - // Reuse prefix construction for namespace validation. - getQueueTopicPrefix('workflow', namespace); + return `, { ${fields.join(', ')} }`; +} - return `, { namespace: ${JSON.stringify(namespace)} }`; +export function createWorkflowRouteHandlersCode( + workflowEntrypointCall: string +) { + return `export const POST = ${workflowEntrypointCall}; +export const GET = POST; +export const HEAD = POST; +export const OPTIONS = POST;`; } /** diff --git a/packages/builders/src/types.ts b/packages/builders/src/types.ts index a17dc69ea2..c84513c636 100644 --- a/packages/builders/src/types.ts +++ b/packages/builders/src/types.ts @@ -40,6 +40,9 @@ interface BaseWorkflowConfig { // Optional prefix for debug files (e.g., "_" for Astro to ignore them) debugFilePrefix?: 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 8585437378..d8c897ffd7 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -7,6 +7,7 @@ import { WorkflowRunFailedError, WorkflowWorldError, } from '@workflow/errors'; +import { createWorkflowUrl } from '@workflow/utils'; import { SPEC_VERSION_CURRENT, type World } from '@workflow/world'; import { afterAll, @@ -147,6 +148,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]; @@ -584,17 +588,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 @@ -639,86 +637,55 @@ 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"}'); }); test( 'parallelStepsThenWebhookWorkflow - no hook_conflict from same-tick replay race', - { timeout: 180_000 }, + { + timeout: 180_000, + }, async () => { // Regression test for https://github.com/vercel/workflow/issues/1665 // and https://github.com/vercel/workflow/issues/2283. @@ -782,17 +749,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); } @@ -838,11 +799,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({}), @@ -1806,7 +1763,9 @@ describe('e2e', () => { }, ])( '$workflow - hook.getConflict() does not block step execution', - { timeout: 60_000 }, + { + timeout: 60_000, + }, async ({ workflow, hookGetConflictTestData }) => { const token = Math.random().toString(36).slice(2); const customData = Math.random().toString(36).slice(2); @@ -1837,7 +1796,9 @@ describe('e2e', () => { test( 'hookGetConflictThenStepParallelWorkflow - hook.getConflict() continuation step runs alongside other steps', - { timeout: 90_000 }, + { + timeout: 90_000, + }, async () => { const token = Math.random().toString(36).slice(2); const customData = Math.random().toString(36).slice(2); @@ -1869,7 +1830,9 @@ describe('e2e', () => { test( 'hookGetConflictWorkflow - hook.getConflict() resolves with the conflicting run when token is already registered', - { timeout: 60_000 }, + { + timeout: 60_000, + }, async () => { const token = Math.random().toString(36).slice(2); const customData = Math.random().toString(36).slice(2); @@ -1922,7 +1885,9 @@ describe('e2e', () => { test( 'hookClaimOnlyMutexWorkflow - hook works as a pure run mutex without payload data', - { timeout: 90_000 }, + { + timeout: 90_000, + }, async () => { const token = Math.random().toString(36).slice(2); @@ -1989,7 +1954,9 @@ describe('e2e', () => { test( 'hookAdoptOwnerResultWorkflow - duplicate adopts the owner result via conflict.returnValue', - { timeout: 120_000 }, + { + timeout: 120_000, + }, async () => { const token = Math.random().toString(36).slice(2); @@ -2053,7 +2020,9 @@ describe('e2e', () => { test( 'hookSignalOwnerWorkflow - duplicate forwards its payload to the owner via resumeHook', - { timeout: 90_000 }, + { + timeout: 90_000, + }, async () => { const token = Math.random().toString(36).slice(2); @@ -2090,7 +2059,9 @@ describe('e2e', () => { test( 'hookSupersedeOwnerWorkflow - duplicate cancels the owner and claims the released token', - { timeout: 90_000 }, + { + timeout: 90_000, + }, async () => { const token = Math.random().toString(36).slice(2); @@ -2132,7 +2103,9 @@ describe('e2e', () => { test( 'resume-or-start route pattern - resumeHook retried after start() reaches the new run', - { timeout: 90_000 }, + { + timeout: 90_000, + }, async () => { const token = Math.random().toString(36).slice(2); @@ -2395,10 +2368,9 @@ describe('e2e', () => { // bypasses protection by sending messages through the Queue infrastructure. // Test the flow endpoint health check - const flowHealthUrl = new URL( - '/.well-known/workflow/v1/flow?__health', - deploymentUrl - ); + const flowHealthUrl = createWorkflowUrl(deploymentUrl, { + type: 'health', + }); const flowHeadRes = await fetch(flowHealthUrl, { method: 'HEAD', headers: await getTrustedSourcesHeaders(), @@ -3101,7 +3073,9 @@ describe('e2e', () => { test( 'hookWithSleepFinalStepWorkflow - step only on final payload', - { timeout: 120_000 }, + { + timeout: 120_000, + }, async () => { // Regression test for the v0chat incident. Mirrors the production // shape: a hook + fire-and-forget sleep, where the step runs only diff --git a/packages/core/e2e/utils.ts b/packages/core/e2e/utils.ts index 6515e81b69..ac982842fb 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'; @@ -352,7 +353,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 1a26839c0e..78962ab1b7 100644 --- a/packages/core/src/create-hook.ts +++ b/packages/core/src/create-hook.ts @@ -164,8 +164,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 9452fc9594..9fd68c3f56 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -6,6 +6,7 @@ import { RunExpiredError, WorkflowRuntimeError, } from '@workflow/errors'; +import { setWorkflowBasePath } from '@workflow/utils'; import { parseWorkflowName } from '@workflow/utils/parse-name'; import { type Event, @@ -133,8 +134,10 @@ function hasRecordedTerminalRunEvent(events: Event[], runId: string): boolean { */ export function workflowEntrypoint( workflowCode: string, - options?: { namespace?: string } + options?: { namespace?: string; basePath?: string } ): (req: Request) => Promise { + setWorkflowBasePath(options?.basePath); + const namespace = resolveQueueNamespace(options?.namespace); const workflowPrefix = getQueueTopicPrefix('workflow', namespace); diff --git a/packages/core/src/runtime/step-handler.ts b/packages/core/src/runtime/step-handler.ts index e532f90841..440612826d 100644 --- a/packages/core/src/runtime/step-handler.ts +++ b/packages/core/src/runtime/step-handler.ts @@ -9,7 +9,7 @@ import { WorkflowRuntimeError, WorkflowWorldError, } from '@workflow/errors'; -import { pluralize } from '@workflow/utils'; +import { createWorkflowBaseUrl, pluralize } from '@workflow/utils'; import { getPort } from '@workflow/utils/get-port'; import { getQueueTopicPrefix, @@ -548,9 +548,11 @@ const stepHandler = createQueueHandler( 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: createWorkflowBaseUrl( + isVercel + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:${port ?? 3000}` + ), }, workflowDeploymentId: process.env.VERCEL_DEPLOYMENT_ID, ops, diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index db59e6d3fe..2d79cda6e3 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 * as nanoid from 'nanoid'; @@ -110,10 +110,16 @@ export async function runWorkflow( // Get the port before creating VM context to avoid async operations // affecting the deterministic timestamp const isVercel = process.env.VERCEL_URL !== undefined; - // The resolved port is cached per process (see get-port-lazy.ts), so this - // is cheap on replays after the first — `getPort()` otherwise re-runs OS - // port discovery (spawning `lsof` on macOS, ~60ms) on every replay. - const port = isVercel ? undefined : await getPortLazy(); + // Load getPort lazily to prevent Turbopack from tracing get-port's + // 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 — `getPort()` otherwise re-runs OS port + // discovery (spawning `lsof` on macOS, ~60ms) on every replay. + const workflowBaseUrl = createWorkflowBaseUrl( + isVercel + ? `https://${process.env.VERCEL_URL}` + : `http://localhost:${(await getPortLazy()) ?? 3000}` + ); const { context, @@ -208,18 +214,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, }; // @ts-expect-error - `@types/node` says symbol is not valid, but it does work diff --git a/packages/core/src/workflow/create-hook.ts b/packages/core/src/workflow/create-hook.ts index 4e604628f3..a52c043183 100644 --- a/packages/core/src/workflow/create-hook.ts +++ b/packages/core/src/workflow/create-hook.ts @@ -1,3 +1,4 @@ +import { createWorkflowUrl } from '@workflow/utils'; import type { Hook, HookOptions, @@ -48,7 +49,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 5e255d5808..3fef19c1da 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'); } @@ -145,6 +144,21 @@ describe('withWorkflow builder config', () => { expect(prewarmWorkflowSwcPluginCacheMock).not.toHaveBeenCalled(); }); + 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('removes workflow packages from serverExternalPackages for this build', async () => { const projectDir = mkdtempSync( join(realTmpDir, 'workflow-next-server-external-') diff --git a/packages/next/src/index.ts b/packages/next/src/index.ts index c4ba7cb31a..0d7edb6f2c 100644 --- a/packages/next/src/index.ts +++ b/packages/next/src/index.ts @@ -18,6 +18,14 @@ const turbopackWorkflowContentPattern = 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; @@ -328,6 +336,8 @@ export function withWorkflow( } // shallow clone to avoid read-only on top-level nextConfig = Object.assign({}, nextConfig); + const workflowBasePath = nextConfig.basePath; + setWorkflowBasePath(workflowBasePath); const configuredServerExternalPackages = Array.isArray( nextConfig.serverExternalPackages @@ -403,6 +413,7 @@ export function withWorkflow( moduleSpecifierRoot: workingDir, workingDir, distDir: nextConfig.distDir || '.next', + basePath: workflowBasePath, buildTarget: 'next', workflowsBundlePath: '', // not used in base stepsBundlePath: '', // not used in base diff --git a/packages/sveltekit/src/builder.ts b/packages/sveltekit/src/builder.ts index 2680445653..232eb4c4f3 100644 --- a/packages/sveltekit/src/builder.ts +++ b/packages/sveltekit/src/builder.ts @@ -192,15 +192,12 @@ export const POST = handleStepRequest;`, // Replace the default export with SvelteKit-compatible handler workflowsRouteContent = replaceGeneratedRouteExport( workflowsRouteContent, - /const handler = workflowEntrypoint\(workflowCode(?[^)]*)\);\s*export const HEAD = handler;\s*export const POST = handler;?\s*$/m, + /export const POST = workflowEntrypoint\(workflowCode(?[^)]*)\);?$/m, (_match, options = '') => `${NORMALIZE_REQUEST_CODE} -const handleWorkflowRequest = async ({request}) => { +export const POST = async ({request}) => { const normalRequest = await normalizeRequest(request); return workflowEntrypoint(workflowCode${options})(normalRequest); -}; - -export const HEAD = handleWorkflowRequest; -export const POST = handleWorkflowRequest;`, +};`, 'Failed to wrap generated SvelteKit workflow route' ); await writeFile(workflowsRouteFile, workflowsRouteContent); diff --git a/packages/utils/src/get-port.ts b/packages/utils/src/get-port.ts index 02e136c244..10ed823d67 100644 --- a/packages/utils/src/get-port.ts +++ b/packages/utils/src/get-port.ts @@ -1,6 +1,7 @@ import { execFile } from 'node:child_process'; import { readdir, readFile, readlink } from 'node:fs/promises'; import { promisify } from 'node:util'; +import { WORKFLOW_ROUTE_BASE } from './workflow-routes.js'; const execFileAsync = promisify(execFile); @@ -280,7 +281,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 5aff0af4e5..0126f1b104 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -6,6 +6,14 @@ export { } from './parse-name.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 { isVercelWorldTarget, resolveWorkflowTargetWorld, 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 2cc9259dac..480c90b056 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(); @@ -191,7 +211,9 @@ describe('queue timeout re-enqueue', () => { }); it('logs actionable guidance for detached ArrayBuffer proxy failures', async () => { - const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}); + const consoleError = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); const fetchError = new TypeError('fetch failed'); (fetchError as TypeError & { cause?: unknown }).cause = new TypeError( 'Cannot perform ArrayBuffer.prototype.slice on a detached ArrayBuffer' diff --git a/packages/world-local/src/queue.ts b/packages/world-local/src/queue.ts index 6a7984b964..c7d32af272 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'; @@ -184,7 +185,9 @@ export function createQueue(config: Partial): LocalQueue { if (directHandler) { const req = new Request( - `http://localhost/.well-known/workflow/v1/${pathname}`, + createWorkflowUrl(resolveDirectBaseUrl(config), { + type: pathname, + }), { method: 'POST', headers, @@ -196,7 +199,7 @@ export function createQueue(config: Partial): LocalQueue { 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 35c051e5ef..2734f16fd6 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',