diff --git a/bun.lock b/bun.lock index ebbae4ca..8ccc6d9b 100644 --- a/bun.lock +++ b/bun.lock @@ -4,7 +4,7 @@ "": { "name": "@upstash/workflow", "dependencies": { - "@upstash/qstash": "^2.8.4", + "@upstash/qstash": "2.9.0-rc", }, "devDependencies": { "@commitlint/cli": "^19.5.0", @@ -536,7 +536,7 @@ "@ungap/structured-clone": ["@ungap/structured-clone@1.3.0", "", {}, "sha512-WmoN8qaIAo7WTYWbAZuG8PYEhn5fkz7dZrqTBZ7dtt//lL2Gwms1IcnQ5yHqjDfX8Ft5j4YzDM23f87zBfDe9g=="], - "@upstash/qstash": ["@upstash/qstash@2.8.4", "", { "dependencies": { "crypto-js": ">=4.2.0", "jose": "^5.2.3", "neverthrow": "^7.0.1" } }, "sha512-iojHWUlRoC3M2e4XQ1NFEgC+7Orurrm5uIrPn5WilU7LvWQoocyjYBXR0VUalXkeMAmFyk4blF0EOYZY4igdIQ=="], + "@upstash/qstash": ["@upstash/qstash@2.9.0-rc", "", { "dependencies": { "crypto-js": ">=4.2.0", "jose": "^5.2.3", "neverthrow": "^7.0.1" } }, "sha512-T2VGOuK/XOFe+kppsANdiaXxpqPvnXl8XWmZzAtqZKsHWkBXPYqAvr/Pek8TZEKeOLAgc7wrl4Rm0KNqec9Fww=="], "@vercel/nft": ["@vercel/nft@0.30.3", "", { "dependencies": { "@mapbox/node-pre-gyp": "^2.0.0", "@rollup/pluginutils": "^5.1.3", "acorn": "^8.6.0", "acorn-import-attributes": "^1.9.5", "async-sema": "^3.1.1", "bindings": "^1.4.0", "estree-walker": "2.0.2", "glob": "^10.4.5", "graceful-fs": "^4.2.9", "node-gyp-build": "^4.2.2", "picomatch": "^4.0.2", "resolve-from": "^5.0.0" }, "bin": { "nft": "out/cli.js" } }, "sha512-UEq+eF0ocEf9WQCV1gktxKhha36KDs7jln5qii6UpPf5clMqDc0p3E7d9l2Smx0i9Pm1qpq4S4lLfNl97bbv6w=="], diff --git a/examples/ci/app/ci/utils.ts b/examples/ci/app/ci/utils.ts index d82bd822..c94f9a73 100644 --- a/examples/ci/app/ci/utils.ts +++ b/examples/ci/app/ci/utils.ts @@ -75,12 +75,7 @@ export const initiateTest = async (params: Pick) => { try { await redis.checkRedisForResults(route, randomTestId, expectedCallCount, expectedResult) } catch (error) { - try { - const logs = await qstash.getWorkflowLogs(workflowRunId) - console.error("Test Failed. Logs of the started workflow:", JSON.stringify(logs, null, 2)) - } catch (error) { - console.error("Failed to get workflow logs:", error) - } + console.error("Test Failed. no results found.") throw error } } diff --git a/package.json b/package.json index f0a876fc..fe3bca61 100644 --- a/package.json +++ b/package.json @@ -105,7 +105,7 @@ "typescript-eslint": "^8.18.0" }, "dependencies": { - "@upstash/qstash": "^2.8.4" + "@upstash/qstash": "2.9.0-rc" }, "directories": { "example": "examples" diff --git a/src/client/index.ts b/src/client/index.ts index 6b91f93b..55e4c438 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -23,17 +23,7 @@ export class Client { private client: QStashClient; constructor(clientConfig: ClientConfig) { - if (!clientConfig?.token) { - console.error( - "QStash token is required for Upstash Workflow!\n\n" + - "To fix this:\n" + - "1. Get your token from the Upstash Console (https://console.upstash.com/qstash)\n" + - "2. Initialize the workflow client with:\n\n" + - " const client = new Client({\n" + - " token: ''\n" + - " });" - ); - } + // TODO: add warning back this.client = new QStashClient(clientConfig); } diff --git a/src/serve/index.ts b/src/serve/index.ts index c4d010d8..c8f818dd 100644 --- a/src/serve/index.ts +++ b/src/serve/index.ts @@ -25,6 +25,7 @@ import { verifyRequest, } from "../workflow-requests"; import { DisabledWorkflowContext } from "./authorization"; +import { getHandlersForRequest } from "./multi-region/handlers"; import { AUTH_FAIL_MESSAGE, createResponseData, @@ -61,10 +62,8 @@ export const serveBase = < // Prepares options with defaults if they are not provided. const { - qstashClient, initialPayloadParser, url, - receiver, failureFunction, baseUrl, env, @@ -100,12 +99,20 @@ export const serveBase = < middlewareManager.dispatchDebug.bind(middlewareManager) ); + // validation & parsing to get isFirstInvocation early + const { isFirstInvocation, workflowRunId, unknownSdk } = validateRequest(request); + + // Get the appropriate handlers based on region + const regionHeader = request.headers.get("upstash-region"); + const { client: regionalClient, receiver: regionalReceiver } = getHandlersForRequest( + internal.qstashHandlers, + regionHeader, + isFirstInvocation + ); + // get payload as raw string const requestPayload = (await getPayload(request)) ?? ""; - await verifyRequest(requestPayload, request.headers.get("upstash-signature"), receiver); - - // validation & parsing - const { isFirstInvocation, workflowRunId, unknownSdk } = validateRequest(request); + await verifyRequest(requestPayload, request.headers.get("upstash-signature"), regionalReceiver); middlewareManager.assignWorkflowRunId(workflowRunId); await middlewareManager.dispatchDebug("onInfo", { @@ -118,7 +125,7 @@ export const serveBase = < isFirstInvocation, unknownSdk, workflowRunId, - requester: qstashClient.http, + requester: regionalClient.http, messageId: request.headers.get("upstash-message-id")!, dispatchDebug: middlewareManager.dispatchDebug.bind(middlewareManager), }); @@ -144,7 +151,7 @@ export const serveBase = < const failureCheck = await handleFailure({ request, requestPayload, - qstashClient, + qstashClient: regionalClient, initialPayloadParser, routeFunction, failureFunction, @@ -181,7 +188,7 @@ export const serveBase = < // create context const workflowContext = new WorkflowContext({ - qstashClient, + qstashClient: regionalClient, workflowRunId, initialPayload: initialPayloadParser(rawInitialPayload), headers: recreateUserHeaders(request.headers as Headers), @@ -219,7 +226,7 @@ export const serveBase = < const callReturnCheck = await handleThirdPartyCallResult({ request, requestPayload: rawInitialPayload, - client: qstashClient, + client: regionalClient, workflowUrl, telemetry, middlewareManager, diff --git a/src/serve/multi-region-integration.test.ts b/src/serve/multi-region-integration.test.ts new file mode 100644 index 00000000..3b550b57 --- /dev/null +++ b/src/serve/multi-region-integration.test.ts @@ -0,0 +1,790 @@ +/** + * Integration tests for multi-region support in workflow SDK. + * These tests modify process.env and test the full flow from environment variables + * through to actual workflow execution with multi-region support. + */ + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { Client } from "@upstash/qstash"; +import { serve } from "./index"; +import { + MOCK_QSTASH_SERVER_URL, + mockQStashServer, + WORKFLOW_ENDPOINT, + getRequest, +} from "../test-utils"; +import { WorkflowContext } from "../context"; +import { WORKFLOW_INIT_HEADER } from "../constants"; + +// Store original environment to restore after each test +let originalEnvironment: Record = {}; + +/** + * Helper to set up environment variables for a test + */ +function setupEnvironment(environmentVariables: Record) { + // Clear relevant env vars + const keysToManage = [ + "QSTASH_TOKEN", + "QSTASH_URL", + "QSTASH_CURRENT_SIGNING_KEY", + "QSTASH_NEXT_SIGNING_KEY", + "QSTASH_REGION", + "US_EAST_1_QSTASH_TOKEN", + "US_EAST_1_QSTASH_URL", + "US_EAST_1_QSTASH_CURRENT_SIGNING_KEY", + "US_EAST_1_QSTASH_NEXT_SIGNING_KEY", + "EU_CENTRAL_1_QSTASH_TOKEN", + "EU_CENTRAL_1_QSTASH_URL", + "EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY", + "EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY", + ]; + + for (const key of keysToManage) { + if (key in environmentVariables) { + process.env[key] = environmentVariables[key]; + } else { + delete process.env[key]; + } + } +} + +/** + * Helper to restore environment variables after a test + */ +function restoreEnvironment() { + for (const [key, value] of Object.entries(originalEnvironment)) { + if (value === undefined) { + delete process.env[key]; + } else { + process.env[key] = value; + } + } +} + +describe("Multi-Region Integration Tests", () => { + beforeEach(() => { + // Save current environment + originalEnvironment = { + QSTASH_TOKEN: process.env.QSTASH_TOKEN, + QSTASH_URL: process.env.QSTASH_URL, + QSTASH_CURRENT_SIGNING_KEY: process.env.QSTASH_CURRENT_SIGNING_KEY, + QSTASH_NEXT_SIGNING_KEY: process.env.QSTASH_NEXT_SIGNING_KEY, + QSTASH_REGION: process.env.QSTASH_REGION, + US_EAST_1_QSTASH_TOKEN: process.env.US_EAST_1_QSTASH_TOKEN, + US_EAST_1_QSTASH_URL: process.env.US_EAST_1_QSTASH_URL, + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: process.env.US_EAST_1_QSTASH_CURRENT_SIGNING_KEY, + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: process.env.US_EAST_1_QSTASH_NEXT_SIGNING_KEY, + EU_CENTRAL_1_QSTASH_TOKEN: process.env.EU_CENTRAL_1_QSTASH_TOKEN, + EU_CENTRAL_1_QSTASH_URL: process.env.EU_CENTRAL_1_QSTASH_URL, + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: process.env.EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY, + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: process.env.EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY, + }; + }); + + afterEach(() => { + restoreEnvironment(); + }); + + describe("Single-Region Mode (Default)", () => { + test("should handle workflow request in single-region mode", async () => { + setupEnvironment({ + QSTASH_URL: MOCK_QSTASH_SERVER_URL, + QSTASH_TOKEN: "default-token", + QSTASH_CURRENT_SIGNING_KEY: "default-current-key", + QSTASH_NEXT_SIGNING_KEY: "default-next-key", + }); + + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + const result = await context.run("step1", () => { + return "workflow result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, // Disable signature verification for this test + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest(WORKFLOW_ENDPOINT, "wfr_123", { message: "test" }, [], { + [WORKFLOW_INIT_HEADER]: "true", + }); + + const response = await handler(request); + expect(response.status).toBe(200); + + const body = (await response.json()) as { workflowRunId?: string }; + expect(body.workflowRunId).toBeDefined(); + }, + responseFields: { + body: { messageId: "msg_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "default-token", + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":1,"stepName":"step1","stepType":"Run","out":"\\"workflow result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + }); + + describe("Multi-Region Mode - US East 1", () => { + test("should handle first invocation with US region as default", async () => { + setupEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + const result = await context.run("step1", () => { + return "us workflow result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, // Disable signature verification + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + "wfr_us_123", + { message: "test from US" }, + [], + { + [WORKFLOW_INIT_HEADER]: "true", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + + const body = (await response.json()) as { workflowRunId?: string }; + expect(body.workflowRunId).toBeDefined(); + }, + responseFields: { + body: { messageId: "msg_us_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "us-token", // Should use US token + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_us_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":1,"stepName":"step1","stepType":"Run","out":"\\"us workflow result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + + test("should handle subsequent invocation with US region header", async () => { + setupEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const workflowRunId = "wfr_us_123"; + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + await context.run("step1", () => { + return "step 1 result"; + }); + const result = await context.run("step2", () => { + return "step 2 result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, + }); + + // Mock the second invocation (after step1) + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + workflowRunId, + { message: "test from US" }, + [ + { + stepId: 1, + stepName: "step1", + stepType: "Run", + out: "step 1 result", + concurrent: 1, + }, + ], + { + "upstash-region": "US-EAST-1", + "upstash-message-id": "msg_123", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: { messageId: "msg_us_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "us-token", // Should use US token based on region header + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-message-id": "msg_123", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_us_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":2,"stepName":"step2","stepType":"Run","out":"\\"step 2 result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + }); + + describe("Multi-Region Mode - EU Central 1", () => { + test("should handle first invocation with EU region as default", async () => { + setupEnvironment({ + QSTASH_REGION: "EU_CENTRAL_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + const result = await context.run("step1", () => { + return "eu workflow result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + "wfr_eu_123", + { message: "test from EU" }, + [], + { + [WORKFLOW_INIT_HEADER]: "true", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + + const body = (await response.json()) as { workflowRunId?: string }; + expect(body.workflowRunId).toBeDefined(); + }, + responseFields: { + body: { messageId: "msg_eu_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "eu-token", // Should use EU token + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_eu_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":1,"stepName":"step1","stepType":"Run","out":"\\"eu workflow result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + + test("should handle subsequent invocation with EU region header", async () => { + setupEnvironment({ + QSTASH_REGION: "EU_CENTRAL_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const workflowRunId = "wfr_eu_123"; + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + await context.run("step1", () => { + return "step 1 result"; + }); + const result = await context.run("step2", () => { + return "step 2 result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + workflowRunId, + { message: "test from EU" }, + [ + { + stepId: 1, + stepName: "step1", + stepType: "Run", + out: "step 1 result", + concurrent: 1, + }, + ], + { + "upstash-region": "EU-CENTRAL-1", + "upstash-message-id": "msg_123", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: { messageId: "msg_eu_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "eu-token", // Should use EU token based on region header + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-message-id": "msg_123", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_eu_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":2,"stepName":"step2","stepType":"Run","out":"\\"step 2 result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + }); + + describe("Region Switching", () => { + test("should use correct region based on header in subsequent invocations", async () => { + setupEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const workflowRunId = "wfr_multi_123"; + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + await context.run("step1", () => { + return "step 1 result"; + }); + return "done"; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, + }); + + // Simulate request coming from EU region (even though default is US) + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + workflowRunId, + { message: "test" }, + [ + { + stepId: 1, + stepName: "step1", + stepType: "Run", + out: "step 1 result", + concurrent: 1, + }, + ], + { + "upstash-region": "EU-CENTRAL-1", + "upstash-message-id": "msg_123", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: {}, + status: 200, + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs/${workflowRunId}?cancel=false`, + token: "eu-token", // Should use EU token because of region header + body: "done", + }, + }); + }); + }); + + describe("Client Configuration Options", () => { + test("should pass client config options to all regions in multi-region mode", async () => { + setupEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + US_EAST_1_QSTASH_TOKEN: "us-token", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + const result = await context.run("step1", () => { + return "result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + qstashClient: { + retry: { + retries: 3, + }, + }, + receiver: undefined, + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest(WORKFLOW_ENDPOINT, "wfr_config_123", { message: "test" }, [], { + [WORKFLOW_INIT_HEADER]: "true", + }); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: { messageId: "msg_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "us-token", + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_config_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":1,"stepName":"step1","stepType":"Run","out":"\\"result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + }); + + describe("Fallback Behavior", () => { + test("should fallback to default region when region header is invalid", async () => { + setupEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const workflowRunId = "wfr_fallback_123"; + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + await context.run("step1", () => { + return "step 1 result"; + }); + return "done"; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + workflowRunId, + { message: "test" }, + [ + { + stepId: 1, + stepName: "step1", + stepType: "Run", + out: "step 1 result", + concurrent: 1, + }, + ], + { + "upstash-region": "INVALID-REGION", + "upstash-message-id": "msg_123", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: {}, + status: 200, + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs/${workflowRunId}?cancel=false`, + token: "us-token", // Should fallback to default region (US) + body: "done", + }, + }); + }); + + test("should use default region when region header is missing", async () => { + setupEnvironment({ + QSTASH_REGION: "EU_CENTRAL_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + EU_CENTRAL_1_QSTASH_URL: MOCK_QSTASH_SERVER_URL, + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const workflowRunId = "wfr_missing_header_123"; + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + await context.run("step1", () => { + return "step 1 result"; + }); + return "done"; + }; + + const { handler } = serve(routeFunction, { + receiver: undefined, + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest( + WORKFLOW_ENDPOINT, + workflowRunId, + { message: "test" }, + [ + { + stepId: 1, + stepName: "step1", + stepType: "Run", + out: "step 1 result", + concurrent: 1, + }, + ], + { + "upstash-message-id": "msg_123", + } + ); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: {}, + status: 200, + }, + receivesRequest: { + method: "DELETE", + url: `${MOCK_QSTASH_SERVER_URL}/v2/workflows/runs/${workflowRunId}?cancel=false`, + token: "eu-token", // Should use default region (EU) + body: "done", + }, + }); + }); + }); + + describe("Force Single-Region Mode", () => { + test("should use single-region mode when client instance is provided", async () => { + setupEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + }); + + const customClient = new Client({ + baseUrl: MOCK_QSTASH_SERVER_URL, + token: "custom-token", + }); + + const routeFunction = async (context: WorkflowContext<{ message: string }>) => { + const result = await context.run("step1", () => { + return "result"; + }); + return result; + }; + + const { handler } = serve(routeFunction, { + qstashClient: customClient, // Forces single-region mode + receiver: undefined, + }); + + await mockQStashServer({ + execute: async () => { + const request = getRequest(WORKFLOW_ENDPOINT, "wfr_custom_123", { message: "test" }, [], { + [WORKFLOW_INIT_HEADER]: "true", + }); + + const response = await handler(request); + expect(response.status).toBe(200); + }, + responseFields: { + body: { messageId: "msg_123" }, + status: 200, + }, + receivesRequest: { + method: "POST", + url: `${MOCK_QSTASH_SERVER_URL}/v2/batch`, + token: "custom-token", // Should use custom client token, not region tokens + body: [ + { + destination: "https://requestcatcher.com/api", + headers: { + "content-type": "application/json", + "upstash-feature-set": "LazyFetch,InitialBody,WF_DetectTrigger,WF_TriggerOnConfig", + "upstash-forward-upstash-workflow-sdk-version": "1", + "upstash-method": "POST", + "upstash-telemetry-framework": "unknown", + "upstash-telemetry-runtime": expect.any(String), + "upstash-telemetry-sdk": expect.stringMatching(/^@upstash\/workflow@v1\.0\./), + "upstash-workflow-init": "false", + "upstash-workflow-runid": "wfr_custom_123", + "upstash-workflow-sdk-version": "1", + "upstash-workflow-url": "https://requestcatcher.com/api", + }, + body: `{"stepId":1,"stepName":"step1","stepType":"Run","out":"\\"result\\"","concurrent":1}`, + }, + ], + }, + }); + }); + }); +}); diff --git a/src/serve/multi-region/handlers.ts b/src/serve/multi-region/handlers.ts new file mode 100644 index 00000000..6410f0a6 --- /dev/null +++ b/src/serve/multi-region/handlers.ts @@ -0,0 +1,203 @@ +import { Client, Receiver } from "@upstash/qstash"; +import { + getRegionFromEnvironment, + normalizeRegionHeader, + QStashHandlers, + QStashRegion, + readClientEnvironmentVariables, + readReceiverEnvironmentVariables, + RegionalHandler, +} from "./utils"; +import { QStashClientExtraConfig, WorkflowClient, WorkflowReceiver } from "../../types"; + +/** + * Get the appropriate QStash client and receiver based on the request region header + * + * @param qstashHandlers - The QStash handlers configuration + * @param regionHeader - The UPSTASH-REGION header from the request + * @param isFirstInvocation - Whether this is the first invocation + * @returns Regional handler with client and receiver + */ +export const getHandlersForRequest = ( + qstashHandlers: QStashHandlers, + regionHeader: string | null, + isFirstInvocation: boolean +): RegionalHandler => { + if (qstashHandlers.mode === "single-region") { + return qstashHandlers.handlers; + } + + // Multi-region mode + let targetRegion: QStashRegion; + + if (isFirstInvocation && !regionHeader) { + // Use the default region for first non-qstash invocation + targetRegion = qstashHandlers.defaultRegion; + } else { + // Use the region from the header for subsequent invocations + const normalizedRegion = regionHeader ? normalizeRegionHeader(regionHeader) : undefined; + targetRegion = normalizedRegion ?? qstashHandlers.defaultRegion; + } + + const handler = qstashHandlers.handlers[targetRegion]; + + if (!handler) { + console.warn( + `[Upstash Workflow] No handler found for region "${targetRegion}". Falling back to default region.` + ); + return qstashHandlers.handlers[qstashHandlers.defaultRegion]; + } + + return handler; +}; + +/** + * Creates a regional handler with client and receiver + */ +const createRegionalHandler = ( + environment: Record, + receiverConfig: WorkflowReceiver | "set-to-undefined" | "not-set", + region?: QStashRegion, + clientOptions?: Omit[0], "baseUrl" | "token"> +): RegionalHandler => { + const clientEnv = readClientEnvironmentVariables(environment, region); + + const client = new Client({ + ...clientOptions, + baseUrl: clientEnv.QSTASH_URL!, + token: clientEnv.QSTASH_TOKEN!, + }); + const receiver = getReceiver(environment, receiverConfig, region); + + return { client, receiver }; +}; + +/** + * Determines if multi-region mode should be enabled + */ +const shouldUseMultiRegionMode = ( + environment: Record, + qstashClientOption?: WorkflowClient | QStashClientExtraConfig +): + | { isMultiRegion: true; defaultRegion: QStashRegion; clientOptions?: QStashClientExtraConfig } + | { isMultiRegion: false } => { + // Multi-region mode is enabled when: + // 1. QSTASH_REGION env variable is set + // 2. qstashClient option is not a WorkflowClient instance (either undefined or config object) + const hasRegionEnv = Boolean(getRegionFromEnvironment(environment)); + if (hasRegionEnv && (!qstashClientOption || !("http" in qstashClientOption))) { + return { + isMultiRegion: true, + defaultRegion: getRegionFromEnvironment(environment)!, + clientOptions: qstashClientOption, + }; + } else { + return { isMultiRegion: false }; + } +}; + +const getQStashHandlers = ({ + environment, + qstashClientOption, + receiverConfig, +}: { + environment: Record; + qstashClientOption?: WorkflowClient | QStashClientExtraConfig; + /** + * - "set-to-undefined" if user explicitly set receiver to undefined in options + * - "not-set" if user did not pass receiver in options + * - WorkflowReceiver if user passed a receiver instance in options + */ + receiverConfig: WorkflowReceiver | "set-to-undefined" | "not-set"; +}): QStashHandlers => { + const multiRegion = shouldUseMultiRegionMode(environment, qstashClientOption); + + if (multiRegion.isMultiRegion) { + // Multi-region mode + + const regions: QStashRegion[] = ["US_EAST_1", "EU_CENTRAL_1"]; + const handlers: Record = {} as Record< + QStashRegion, + RegionalHandler + >; + + for (const region of regions) { + try { + handlers[region] = createRegionalHandler( + environment, + receiverConfig, + region, + multiRegion.clientOptions + ); + } catch (error) { + console.warn(`[Upstash Workflow] Failed to create handler for region ${region}:`, error); + } + } + + return { + mode: "multi-region", + handlers, + defaultRegion: multiRegion.defaultRegion, + }; + } else { + // Single-region mode + return { + mode: "single-region", + handlers: { + client: + qstashClientOption && "http" in qstashClientOption + ? qstashClientOption + : new Client({ + ...qstashClientOption, + baseUrl: environment.QSTASH_URL!, + token: environment.QSTASH_TOKEN!, + }), + receiver: getReceiver(environment, receiverConfig), + }, + }; + } +}; + +const getReceiver = ( + environment: Record, + receiverConfig: WorkflowReceiver | "set-to-undefined" | "not-set", + region?: QStashRegion +) => { + if (typeof receiverConfig === "string") { + if (receiverConfig === "set-to-undefined") { + return undefined; + } + + const receiverEnv = readReceiverEnvironmentVariables(environment, region); + return receiverEnv.QSTASH_CURRENT_SIGNING_KEY && receiverEnv.QSTASH_NEXT_SIGNING_KEY + ? new Receiver({ + currentSigningKey: receiverEnv.QSTASH_CURRENT_SIGNING_KEY, + nextSigningKey: receiverEnv.QSTASH_NEXT_SIGNING_KEY, + }) + : undefined; + } else { + return receiverConfig; + } +}; + +export const getQStashHandlerOptions = ( + ...params: Parameters +): { + qstashHandlers: ReturnType; + defaultReceiver: WorkflowReceiver | undefined; + defaultClient: WorkflowClient; +} => { + const handlers = getQStashHandlers(...params); + + return { + qstashHandlers: handlers, + defaultReceiver: + handlers.mode === "single-region" + ? handlers.handlers.receiver + : handlers.handlers[handlers.defaultRegion].receiver, + defaultClient: + handlers.mode === "single-region" + ? handlers.handlers.client + : handlers.handlers[handlers.defaultRegion].client, + }; +}; diff --git a/src/serve/multi-region/multi-region.test.ts b/src/serve/multi-region/multi-region.test.ts new file mode 100644 index 00000000..3a106ec0 --- /dev/null +++ b/src/serve/multi-region/multi-region.test.ts @@ -0,0 +1,525 @@ +/** + * Tests for multi-region support utilities. + * Tests credential resolution and region handling. + */ + +import { describe, expect, test } from "bun:test"; +import { Client, Receiver } from "@upstash/qstash"; +import { + getRegionFromEnvironment, + normalizeRegionHeader, + readClientEnvironmentVariables, + readReceiverEnvironmentVariables, +} from "./utils"; +import { getQStashHandlerOptions } from "./handlers"; + +// Helper to create a clean environment for each test +const createEnvironment = ( + environment: Record +): Record => { + return { ...environment }; +}; + +describe("Multi-Region Utilities", () => { + describe("normalizeRegionHeader", () => { + test("should normalize hyphenated region to underscores", () => { + expect(normalizeRegionHeader("us-east-1")).toBe("US_EAST_1"); + expect(normalizeRegionHeader("eu-central-1")).toBe("EU_CENTRAL_1"); + }); + + test("should normalize lowercase region to uppercase", () => { + expect(normalizeRegionHeader("us_east_1")).toBe("US_EAST_1"); + expect(normalizeRegionHeader("eu_central_1")).toBe("EU_CENTRAL_1"); + }); + + test("should accept correctly formatted regions", () => { + expect(normalizeRegionHeader("US_EAST_1")).toBe("US_EAST_1"); + expect(normalizeRegionHeader("EU_CENTRAL_1")).toBe("EU_CENTRAL_1"); + }); + + test("should return undefined for invalid regions", () => { + expect(normalizeRegionHeader("INVALID_REGION")).toBeUndefined(); + expect(normalizeRegionHeader("ap-south-1")).toBeUndefined(); + }); + + test("should return undefined for undefined input", () => { + expect(normalizeRegionHeader(undefined)).toBeUndefined(); + }); + + test("should return undefined for empty string", () => { + expect(normalizeRegionHeader("")).toBeUndefined(); + }); + }); + + describe("getRegionFromEnvironment", () => { + test("should read QSTASH_REGION from environment", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + }); + + expect(getRegionFromEnvironment(environment)).toBe("US_EAST_1"); + }); + + test("should normalize region from environment", () => { + const environment = createEnvironment({ + QSTASH_REGION: "us-east-1", + }); + + expect(getRegionFromEnvironment(environment)).toBe("US_EAST_1"); + }); + + test("should return undefined when QSTASH_REGION is not set", () => { + const environment = createEnvironment({}); + + expect(getRegionFromEnvironment(environment)).toBeUndefined(); + }); + + test("should return undefined for invalid region in environment", () => { + const environment = createEnvironment({ + QSTASH_REGION: "INVALID", + }); + + expect(getRegionFromEnvironment(environment)).toBeUndefined(); + }); + }); + + describe("readClientEnvironmentVariables", () => { + test("should read default client credentials", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + }); + + const result = readClientEnvironmentVariables(environment); + + expect(result.QSTASH_URL).toBe("https://qstash.upstash.io"); + expect(result.QSTASH_TOKEN).toBe("test-token"); + }); + + test("should read region-specific client credentials", () => { + const environment = createEnvironment({ + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + }); + + const result = readClientEnvironmentVariables(environment, "US_EAST_1"); + + expect(result.QSTASH_URL).toBe("https://us-qstash.upstash.io"); + expect(result.QSTASH_TOKEN).toBe("us-token"); + }); + + test("should return undefined for missing credentials", () => { + const environment = createEnvironment({}); + + const result = readClientEnvironmentVariables(environment); + + expect(result.QSTASH_URL).toBeUndefined(); + expect(result.QSTASH_TOKEN).toBeUndefined(); + }); + + test("should read EU region credentials", () => { + const environment = createEnvironment({ + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const result = readClientEnvironmentVariables(environment, "EU_CENTRAL_1"); + + expect(result.QSTASH_URL).toBe("https://eu-qstash.upstash.io"); + expect(result.QSTASH_TOKEN).toBe("eu-token"); + }); + }); + + describe("readReceiverEnvironmentVariables", () => { + test("should read default receiver credentials", () => { + const environment = createEnvironment({ + QSTASH_CURRENT_SIGNING_KEY: "current-key", + QSTASH_NEXT_SIGNING_KEY: "next-key", + }); + + const result = readReceiverEnvironmentVariables(environment); + + expect(result.QSTASH_CURRENT_SIGNING_KEY).toBe("current-key"); + expect(result.QSTASH_NEXT_SIGNING_KEY).toBe("next-key"); + }); + + test("should read region-specific receiver credentials", () => { + const environment = createEnvironment({ + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + }); + + const result = readReceiverEnvironmentVariables(environment, "US_EAST_1"); + + expect(result.QSTASH_CURRENT_SIGNING_KEY).toBe("us-current-key"); + expect(result.QSTASH_NEXT_SIGNING_KEY).toBe("us-next-key"); + }); + + test("should return undefined for missing credentials", () => { + const environment = createEnvironment({}); + + const result = readReceiverEnvironmentVariables(environment); + + expect(result.QSTASH_CURRENT_SIGNING_KEY).toBeUndefined(); + expect(result.QSTASH_NEXT_SIGNING_KEY).toBeUndefined(); + }); + + test("should read EU region credentials", () => { + const environment = createEnvironment({ + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const result = readReceiverEnvironmentVariables(environment, "EU_CENTRAL_1"); + + expect(result.QSTASH_CURRENT_SIGNING_KEY).toBe("eu-current-key"); + expect(result.QSTASH_NEXT_SIGNING_KEY).toBe("eu-next-key"); + }); + }); +}); + +describe("QStash Handler Options - Multi-Region Mode Detection", () => { + describe("Single-Region Mode (Default)", () => { + test("should use single-region mode with default credentials", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + QSTASH_CURRENT_SIGNING_KEY: "current-key", + QSTASH_NEXT_SIGNING_KEY: "next-key", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("single-region"); + expect(result.defaultClient).toBeDefined(); + expect(result.defaultReceiver).toBeDefined(); + }); + + test("should use single-region mode when QSTASH_REGION is not set", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("single-region"); + }); + + test("should use single-region mode when client instance is provided", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + }); + + const client = new Client({ + baseUrl: "https://custom.upstash.io", + token: "custom-token", + }); + + const result = getQStashHandlerOptions({ + environment, + qstashClientOption: client, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("single-region"); + if (result.qstashHandlers.mode === "single-region") { + expect(result.qstashHandlers.handlers.client).toBe(client); + } + }); + }); + + describe("Multi-Region Mode", () => { + test("should enable multi-region mode when QSTASH_REGION is set", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("multi-region"); + if (result.qstashHandlers.mode === "multi-region") { + expect(result.qstashHandlers.defaultRegion).toBe("US_EAST_1"); + expect(result.qstashHandlers.handlers["US_EAST_1"]).toBeDefined(); + expect(result.qstashHandlers.handlers["EU_CENTRAL_1"]).toBeDefined(); + } + }); + + test("should use US_EAST_1 as default region when specified", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("multi-region"); + if (result.qstashHandlers.mode === "multi-region") { + expect(result.qstashHandlers.defaultRegion).toBe("US_EAST_1"); + } + }); + + test("should use EU_CENTRAL_1 as default region when specified", () => { + const environment = createEnvironment({ + QSTASH_REGION: "EU_CENTRAL_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("multi-region"); + if (result.qstashHandlers.mode === "multi-region") { + expect(result.qstashHandlers.defaultRegion).toBe("EU_CENTRAL_1"); + } + }); + + test("should enable multi-region mode with client config options", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const result = getQStashHandlerOptions({ + environment, + qstashClientOption: { + retry: { + retries: 3, + }, + }, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("multi-region"); + }); + + test("should create handlers for both regions in multi-region mode", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("multi-region"); + if (result.qstashHandlers.mode === "multi-region") { + const usHandler = result.qstashHandlers.handlers["US_EAST_1"]; + const euHandler = result.qstashHandlers.handlers["EU_CENTRAL_1"]; + + expect(usHandler.client).toBeDefined(); + expect(usHandler.receiver).toBeDefined(); + expect(euHandler.client).toBeDefined(); + expect(euHandler.receiver).toBeDefined(); + } + }); + }); + + describe("Receiver Configuration", () => { + test("should use receiver from config when provided", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + }); + + const customReceiver = new Receiver({ + currentSigningKey: "config-current", + nextSigningKey: "config-next", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: customReceiver, + }); + + expect(result.defaultReceiver).toBe(customReceiver); + }); + + test("should not create receiver when explicitly set to undefined", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + QSTASH_CURRENT_SIGNING_KEY: "current-key", + QSTASH_NEXT_SIGNING_KEY: "next-key", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "set-to-undefined", + }); + + expect(result.defaultReceiver).toBeUndefined(); + }); + + test("should create receiver from env when not set", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + QSTASH_CURRENT_SIGNING_KEY: "current-key", + QSTASH_NEXT_SIGNING_KEY: "next-key", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.defaultReceiver).toBeDefined(); + }); + + test("should not create receiver when env vars are missing", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.defaultReceiver).toBeUndefined(); + }); + + test("should create region-specific receivers in multi-region mode", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + US_EAST_1_QSTASH_CURRENT_SIGNING_KEY: "us-current-key", + US_EAST_1_QSTASH_NEXT_SIGNING_KEY: "us-next-key", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + EU_CENTRAL_1_QSTASH_CURRENT_SIGNING_KEY: "eu-current-key", + EU_CENTRAL_1_QSTASH_NEXT_SIGNING_KEY: "eu-next-key", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + if (result.qstashHandlers.mode === "multi-region") { + expect(result.qstashHandlers.handlers["US_EAST_1"].receiver).toBeDefined(); + expect(result.qstashHandlers.handlers["EU_CENTRAL_1"].receiver).toBeDefined(); + } + }); + }); + + describe("Fallback Behavior", () => { + test("should handle missing region credentials gracefully", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + // EU_CENTRAL_1 credentials missing + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.qstashHandlers.mode).toBe("multi-region"); + if (result.qstashHandlers.mode === "multi-region") { + // US should be created + expect(result.qstashHandlers.handlers["US_EAST_1"]).toBeDefined(); + // EU might not be created or might be undefined, depending on implementation + } + }); + + test("should normalize invalid QSTASH_REGION and fallback to single-region", () => { + const environment = createEnvironment({ + QSTASH_REGION: "INVALID_REGION", + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + // Should fallback to single-region mode + expect(result.qstashHandlers.mode).toBe("single-region"); + }); + }); + + describe("Default Client and Receiver", () => { + test("should return default client in single-region mode", () => { + const environment = createEnvironment({ + QSTASH_URL: "https://qstash.upstash.io", + QSTASH_TOKEN: "test-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.defaultClient).toBeDefined(); + expect(result.defaultClient.http).toBeDefined(); + }); + + test("should return default region client in multi-region mode", () => { + const environment = createEnvironment({ + QSTASH_REGION: "US_EAST_1", + US_EAST_1_QSTASH_URL: "https://us-qstash.upstash.io", + US_EAST_1_QSTASH_TOKEN: "us-token", + EU_CENTRAL_1_QSTASH_URL: "https://eu-qstash.upstash.io", + EU_CENTRAL_1_QSTASH_TOKEN: "eu-token", + }); + + const result = getQStashHandlerOptions({ + environment, + receiverConfig: "not-set", + }); + + expect(result.defaultClient).toBeDefined(); + // Default client should be from US_EAST_1 since that's the default region + expect(result.defaultClient.http).toBeDefined(); + }); + }); +}); diff --git a/src/serve/multi-region/utils.ts b/src/serve/multi-region/utils.ts new file mode 100644 index 00000000..0e1a9798 --- /dev/null +++ b/src/serve/multi-region/utils.ts @@ -0,0 +1,88 @@ +import { WorkflowClient, WorkflowReceiver } from "../../types"; + +const VALID_REGIONS = ["EU_CENTRAL_1", "US_EAST_1"] as const; + +export type QStashRegion = (typeof VALID_REGIONS)[number]; + +/** + * Regional handler containing client and optional receiver + */ +export type RegionalHandler = { + client: WorkflowClient; + receiver?: WorkflowReceiver; +}; + +/** + * QStash handlers for single or multi-region mode + */ +export type QStashHandlers = + | { + mode: "single-region"; + handlers: RegionalHandler; + } + | { + mode: "multi-region"; + handlers: Record; + defaultRegion: QStashRegion; + }; + +export const DEFAULT_QSTASH_URL = "https://qstash.upstash.io"; + +export const getRegionFromEnvironment = ( + environment: Record +): QStashRegion | undefined => { + const region = environment.QSTASH_REGION as QStashRegion | undefined; + return normalizeRegionHeader(region); +}; + +function readEnvironmentVariables( + environmentVariables: T, + environment: Record, + region?: QStashRegion +): Record { + const result: Record = {}; + + for (const variable of environmentVariables) { + const key = region ? `${region}_${variable}` : variable; + result[variable] = environment[key]; + } + + return result as Record; +} + +export function readClientEnvironmentVariables( + environment: Record, + region?: QStashRegion +) { + return readEnvironmentVariables(["QSTASH_URL", "QSTASH_TOKEN"] as const, environment, region); +} + +export function readReceiverEnvironmentVariables( + environment: Record, + region?: QStashRegion +) { + return readEnvironmentVariables( + ["QSTASH_CURRENT_SIGNING_KEY", "QSTASH_NEXT_SIGNING_KEY"] as const, + environment, + region + ); +} + +export function normalizeRegionHeader(region: string | undefined): QStashRegion | undefined { + if (!region) { + return undefined; + } + + region = region.replaceAll("-", "_").toUpperCase(); + if (VALID_REGIONS.includes(region as QStashRegion)) { + return region as QStashRegion; + } + + console.warn( + `[Upstash Workflow] Invalid UPSTASH-REGION header value: "${region}". Expected one of: ${VALID_REGIONS.join( + ", " + )}.` + ); + + return undefined; +} diff --git a/src/serve/options.ts b/src/serve/options.ts index ebf62ebb..ca6f682c 100644 --- a/src/serve/options.ts +++ b/src/serve/options.ts @@ -1,10 +1,10 @@ -import { Receiver } from "@upstash/qstash"; -import { Client } from "@upstash/qstash"; import { VERSION, WORKFLOW_PROTOCOL_VERSION, WORKFLOW_PROTOCOL_VERSION_HEADER } from "../constants"; import type { DetailedFinishCondition, RequiredExceptFields, WorkflowServeOptions } from "../types"; import { formatWorkflowError, WorkflowError } from "../error"; import { loggingMiddleware } from "../middleware"; import { DispatchDebug } from "../middleware/types"; +import { getQStashHandlerOptions } from "./multi-region/handlers"; +import { QStashHandlers } from "./multi-region/utils"; export type ResponseData = { text: string; @@ -25,6 +25,10 @@ export type InternalServeOptions = { * in `triggerFirstInvocation`. */ useJSONContent: boolean; + /** + * QStash handlers for single or multi-region mode + */ + qstashHandlers: QStashHandlers; }; /** @@ -127,17 +131,23 @@ export const processOptions = < const environment = options?.env ?? (typeof process === "undefined" ? ({} as Record) : process.env); - const receiverEnvironmentVariablesSet = Boolean( - environment.QSTASH_CURRENT_SIGNING_KEY && environment.QSTASH_NEXT_SIGNING_KEY - ); + const { + qstashHandlers, + defaultClient: qstashClient, + defaultReceiver: receiver, + } = getQStashHandlerOptions({ + environment, + qstashClientOption: options?.qstashClient, + receiverConfig: + options && "receiver" in options + ? options.receiver + ? options.receiver + : "set-to-undefined" + : "not-set", + }); return { - qstashClient: - options?.qstashClient ?? - new Client({ - baseUrl: environment.QSTASH_URL!, - token: environment.QSTASH_TOKEN!, - }), + qstashClient, initialPayloadParser: (initialRequest: string) => { // if there is no payload, simply return undefined if (!initialRequest) { @@ -158,12 +168,7 @@ export const processOptions = < throw error; } }, - receiver: receiverEnvironmentVariablesSet - ? new Receiver({ - currentSigningKey: environment.QSTASH_CURRENT_SIGNING_KEY!, - nextSigningKey: environment.QSTASH_NEXT_SIGNING_KEY!, - }) - : undefined, + receiver, baseUrl: environment.UPSTASH_WORKFLOW_URL, env: environment, disableTelemetry: false, @@ -180,6 +185,7 @@ export const processOptions = < }) as TResponse; }), useJSONContent: internalOptions?.useJSONContent ?? false, + qstashHandlers, }, }; }; diff --git a/src/types.ts b/src/types.ts index a6f15972..6dd5ad27 100644 --- a/src/types.ts +++ b/src/types.ts @@ -172,11 +172,20 @@ type WorkflowContextWithoutMethods = Omit< | "waitForWebhook" >; +export type QStashClientExtraConfig = Omit< + NonNullable[0]>, + "baseUrl" | "token" +>; + export type WorkflowServeOptions = { /** - * QStash client + * QStash client or client configuration + * + * Can be either: + * - A WorkflowClient instance + * - Client configuration options (omitting baseUrl and token, which will be read from env vars) */ - qstashClient?: WorkflowClient; + qstashClient?: WorkflowClient | QStashClientExtraConfig; /** * Url of the endpoint where the workflow is set up. * diff --git a/src/workflow-requests.ts b/src/workflow-requests.ts index 7b4558b0..aa50cadd 100644 --- a/src/workflow-requests.ts +++ b/src/workflow-requests.ts @@ -271,7 +271,8 @@ export const recreateUserHeaders = (headers: Headers): Headers => { const headerLowerCase = header.toLowerCase(); const isUserHeader = - (!headerLowerCase.startsWith("upstash-workflow-") && + (headerLowerCase !== "upstash-region" && + !headerLowerCase.startsWith("upstash-workflow-") && // https://vercel.com/docs/edge-network/headers/request-headers#x-vercel-id !headerLowerCase.startsWith("x-vercel-") && !headerLowerCase.startsWith("x-forwarded-") &&