From e65439b723c50f41e3cd8f34cf784890724bb407 Mon Sep 17 00:00:00 2001 From: c_w_xiaohei <1641233466@qq.com> Date: Thu, 9 Jul 2026 01:48:11 +0800 Subject: [PATCH] Support existing Cloudflare workers for queue and cron --- platform/src/components/cloudflare/cron.ts | 68 +++++++----- .../cloudflare/queue-worker-subscriber.ts | 22 +++- platform/src/components/cloudflare/queue.ts | 30 +++++- .../cloudflare-existing-worker.test.ts | 102 ++++++++++++++++++ .../components/cloudflare-worker-api-types.ts | 20 ++++ 5 files changed, 207 insertions(+), 35 deletions(-) create mode 100644 platform/test/components/cloudflare-existing-worker.test.ts create mode 100644 platform/test/components/cloudflare-worker-api-types.ts diff --git a/platform/src/components/cloudflare/cron.ts b/platform/src/components/cloudflare/cron.ts index 6b6598b282..a45b107bc4 100644 --- a/platform/src/components/cloudflare/cron.ts +++ b/platform/src/components/cloudflare/cron.ts @@ -2,7 +2,7 @@ import { all, ComponentResourceOptions, Output } from "@pulumi/pulumi"; import * as cf from "@pulumi/cloudflare"; import * as cloudflare from "@pulumi/cloudflare"; import { Component, Transform, transform } from "../component"; -import { WorkerArgs } from "./worker"; +import { Worker, WorkerArgs } from "./worker"; import { DEFAULT_ACCOUNT_ID } from "./account-id.js"; import { Input } from "../input.js"; import { WorkerBuilder, workerBuilder } from "./helpers/worker-builder"; @@ -10,52 +10,41 @@ import { VisibleError } from "../error"; export interface CronArgs { /** - * The worker that'll be executed when the cron job runs. + * The worker that's executed when the cron job runs. * @deprecated Use `worker` instead. + */ + job?: Input | Worker; + /** + * The worker that's executed when the cron job runs. * * @example * * ```ts * { - * job: "src/cron.ts" + * worker: "src/cron.ts" * } * ``` * - * You can pass in the full worker props. + * Pass full worker props. * * ```ts * { - * job: { + * worker: { * handler: "src/cron.ts", * link: [bucket] * } * } * ``` - */ - job?: Input; - /** - * The worker that'll be executed when the cron job runs. * - * @example + * Or pass an existing worker. * * ```ts * { - * worker: "src/cron.ts" - * } - * ``` - * - * You can pass in the full worker props. - * - * ```ts - * { - * worker: { - * handler: "src/cron.ts", - * link: [bucket] - * } + * worker * } * ``` */ - worker?: Input; + worker?: Input | Worker; /** * The schedule for the cron job. * @@ -115,7 +104,7 @@ export interface CronArgs { * }; * ``` * - * Pass in a `schedules` and a `worker` that'll be executed. + * Pass `schedules` and a `worker`. * * ```ts title="sst.config.ts" * new sst.cloudflare.Cron("MyCronJob", { @@ -124,9 +113,9 @@ export interface CronArgs { * }); * ``` * - * #### Customize the worker + * #### Pass full worker props * - * ```js title="sst.config.ts" + * ```ts title="sst.config.ts" * new sst.cloudflare.Cron("MyCronJob", { * schedules: ["* * * * *"], * worker: { @@ -135,6 +124,19 @@ export interface CronArgs { * } * }); * ``` + * + * #### Use an existing worker + * + * ```ts title="sst.config.ts" + * const worker = new sst.cloudflare.Worker("MyWorker", { + * handler: "worker.ts" + * }); + * + * new sst.cloudflare.Cron("MyCronJob", { + * schedules: ["* * * * *"], + * worker + * }); + * ``` */ export class Cron extends Component { private worker: WorkerBuilder; @@ -164,7 +166,19 @@ export class Cron extends Component { throw new VisibleError( `You must provide a "worker" for the "${name}" Cron component.`, ); - return workerBuilder(`${name}Handler`, workerArgs, undefined, undefined, args.accountId); + if (workerArgs instanceof Worker) + return all([workerArgs.nodes.worker]).apply(([script]) => ({ + getWorker: () => workerArgs, + script, + })); + + return workerBuilder( + `${name}Handler`, + workerArgs as Input, + undefined, + undefined, + args.accountId, + ); } function createTrigger() { diff --git a/platform/src/components/cloudflare/queue-worker-subscriber.ts b/platform/src/components/cloudflare/queue-worker-subscriber.ts index 7cc991e1b1..ca5bd58219 100644 --- a/platform/src/components/cloudflare/queue-worker-subscriber.ts +++ b/platform/src/components/cloudflare/queue-worker-subscriber.ts @@ -7,8 +7,9 @@ import { toMilliseconds, } from "../duration"; import { WorkerBuilder, workerBuilder } from "./helpers/worker-builder"; -import { WorkerArgs } from "./worker"; +import { Worker, WorkerArgs } from "./worker"; import { DEFAULT_ACCOUNT_ID } from "./account-id"; +import { VisibleError } from "../error"; export interface QueueWorkerSubscriberArgs { /** @@ -21,9 +22,10 @@ export interface QueueWorkerSubscriberArgs { id: Input; }>; /** - * The subscriber worker. + * The subscriber worker. Accepts a handler path, full worker props, or an + * existing Cloudflare Worker. */ - subscriber: Input; + subscriber: Input | Worker; /** * The Cloudflare account ID to use for this subscriber and its consumer. * Overrides the default account ID set via `CLOUDFLARE_DEFAULT_ACCOUNT_ID`. @@ -133,9 +135,21 @@ export class QueueWorkerSubscriber extends Component { this.consumer = consumer; function createWorker() { + if (args.subscriber instanceof Worker) { + if (args.transform?.worker) + throw new VisibleError( + `Cannot transform the "${name}" Worker because it is already created.`, + ); + + return output({ + getWorker: () => args.subscriber as Worker, + script: args.subscriber.nodes.worker, + }); + } + return workerBuilder( `${name}Function`, - args.subscriber, + args.subscriber as Input, args.transform?.worker, { parent: self }, accountId, diff --git a/platform/src/components/cloudflare/queue.ts b/platform/src/components/cloudflare/queue.ts index 255c223659..3495b5dd36 100644 --- a/platform/src/components/cloudflare/queue.ts +++ b/platform/src/components/cloudflare/queue.ts @@ -4,7 +4,7 @@ import { Component, Transform, transform } from "../component"; import { Link } from "../link"; import { binding } from "./binding"; import { DEFAULT_ACCOUNT_ID } from "./account-id"; -import { WorkerArgs } from "./worker"; +import { Worker, WorkerArgs } from "./worker"; import { VisibleError } from "../error"; import { QueueWorkerSubscriber } from "./queue-worker-subscriber"; import { DurationMinutes, DurationSeconds } from "../duration"; @@ -168,6 +168,17 @@ export interface QueueSubscribeArgs { * link: [bucket], * }); * ``` + * + * #### Subscribe with an existing worker + * + * ```ts title="sst.config.ts" + * const worker = new sst.cloudflare.Worker("MyWorker", { + * handler: "worker.ts", + * link: [queue], + * }); + * + * queue.subscribe(worker); + * ``` */ export class Queue extends Component implements Link.Linkable { private queue: cloudflare.Queue; @@ -204,7 +215,7 @@ export class Queue extends Component implements Link.Linkable { } /** - * Subscribe to the queue with a worker. + * Subscribe with a worker. * * @param subscriber The worker that'll process messages from the queue. * @param args Configure the subscription. @@ -218,7 +229,7 @@ export class Queue extends Component implements Link.Linkable { * queue.subscribe("consumer.ts"); * ``` * - * Pass in full worker props. + * Pass full worker props. * * ```ts title="sst.config.ts" * const bucket = new sst.cloudflare.Bucket("MyBucket"); @@ -229,6 +240,17 @@ export class Queue extends Component implements Link.Linkable { * }); * ``` * + * Pass an existing worker. + * + * ```ts title="sst.config.ts" + * const worker = new sst.cloudflare.Worker("MyWorker", { + * handler: "worker.ts", + * link: [queue], + * }); + * + * queue.subscribe(worker); + * ``` + * * Configure batch settings. * * ```ts title="sst.config.ts" @@ -241,7 +263,7 @@ export class Queue extends Component implements Link.Linkable { * ``` */ public subscribe( - subscriber: Input, + subscriber: Input | Worker, args?: QueueSubscribeArgs, opts?: ComponentResourceOptions, ) { diff --git a/platform/test/components/cloudflare-existing-worker.test.ts b/platform/test/components/cloudflare-existing-worker.test.ts new file mode 100644 index 0000000000..bdb3f47246 --- /dev/null +++ b/platform/test/components/cloudflare-existing-worker.test.ts @@ -0,0 +1,102 @@ +import * as pulumi from "@pulumi/pulumi"; +import type * as cloudflare from "@pulumi/cloudflare"; +import { describe, expect, it } from "vitest"; +import type { Worker } from "../../src/components/cloudflare/worker"; + +// @ts-ignore +global.$app = { + name: "app", + stage: "test", + providers: {}, +}; + +pulumi.runtime.setMocks( + { + newResource: function (args: pulumi.runtime.MockResourceArgs): { + id: string; + state: any; + } { + return { + id: args.name + "_id", + state: args.inputs, + }; + }, + call: function (args: pulumi.runtime.MockCallArgs) { + return args.inputs; + }, + }, + "project", + "stack", + false, +); + +async function resolveOutput(value: pulumi.Output) { + return await new Promise((resolve) => { + value.apply((resolved) => { + resolve(resolved); + return resolved; + }); + }); +} + +function createMockWorker() { + const script = { + scriptName: pulumi.output("existing-worker"), + } as cloudflare.WorkerScript; + return import("../../src/components/cloudflare/worker").then(({ Worker }) => { + const worker = Object.create(Worker.prototype) as Worker; + Object.defineProperty(worker, "nodes", { value: { worker: script } }); + return worker; + }); +} + +describe("Cloudflare existing Worker targets", () => { + it("attaches a queue consumer to an existing Worker", async () => { + const { QueueWorkerSubscriber } = await import( + "../../src/components/cloudflare/queue-worker-subscriber" + ); + const worker = await createMockWorker(); + + const subscriber = new QueueWorkerSubscriber("Subscriber", { + queue: { id: "queue-id" }, + subscriber: worker, + }); + + const scriptName = await resolveOutput(subscriber.nodes.consumer.scriptName); + + expect(scriptName).toBe("existing-worker"); + }); + + it("rejects worker transforms when subscribing an existing Worker", async () => { + const { QueueWorkerSubscriber } = await import( + "../../src/components/cloudflare/queue-worker-subscriber" + ); + const worker = await createMockWorker(); + + expect( + () => + new QueueWorkerSubscriber("SubscriberWithTransform", { + queue: { id: "queue-id" }, + subscriber: worker, + transform: { + worker: () => undefined, + }, + }), + ).toThrow(/already created/); + }); + + it("attaches a cron trigger to an existing Worker", async () => { + const { Cron } = await import("../../src/components/cloudflare/cron"); + const worker = await createMockWorker(); + + const cron = new Cron("Cron", { + worker, + schedules: ["*/5 * * * *"], + }); + + const trigger = await resolveOutput(cron.nodes.trigger); + const scriptName = await resolveOutput(trigger.scriptName); + + expect(scriptName).toBe("existing-worker"); + }); +}); diff --git a/platform/test/components/cloudflare-worker-api-types.ts b/platform/test/components/cloudflare-worker-api-types.ts new file mode 100644 index 0000000000..d79055d422 --- /dev/null +++ b/platform/test/components/cloudflare-worker-api-types.ts @@ -0,0 +1,20 @@ +import { Cron } from "../../src/components/cloudflare/cron"; +import { Queue } from "../../src/components/cloudflare/queue"; +import { Worker } from "../../src/components/cloudflare/worker"; + +// This file is a compile-time fixture covered by `bun run typecheck:platform`. +declare const queue: Queue; +declare const worker: Worker; +declare const bucket: unknown; + +queue.subscribe(worker); +queue.subscribe("consumer.ts"); +queue.subscribe({ handler: "consumer.ts", link: [bucket] }); + +new Cron("CronWorker", { worker, schedules: ["*/5 * * * *"] }); +new Cron("CronJobWorker", { job: worker, schedules: ["*/5 * * * *"] }); +new Cron("CronString", { worker: "cron.ts", schedules: ["* * * * *"] }); +new Cron("CronArgs", { + worker: { handler: "cron.ts", link: [bucket] }, + schedules: ["* * * * *"], +});