Support existing Cloudflare workers for queue and cron

platform/src/components/cloudflare/cron.ts

@@ -2,60 +2,49 @@ 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";
 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<string | WorkerArgs> | 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<string | WorkerArgs>;
-  /**
-   * 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<string | WorkerArgs>;
+  worker?: Input<string | WorkerArgs> | 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<string | WorkerArgs>,
+        undefined,
+        undefined,
+        args.accountId,
+      );
     }
 
     function createTrigger() {

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<string>;
   }>;
   /**
-   * The subscriber worker.
+   * The subscriber worker. Accepts a handler path, full worker props, or an
+   * existing Cloudflare Worker.
    */
-  subscriber: Input<string | WorkerArgs>;
+  subscriber: Input<string | WorkerArgs> | 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<string | WorkerArgs>,
         args.transform?.worker,
         { parent: self },
         accountId,

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<string | WorkerArgs>,
+    subscriber: Input<string | WorkerArgs> | Worker,
     args?: QueueSubscribeArgs,
     opts?: ComponentResourceOptions,
   ) {

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<T>(value: pulumi.Output<T>) {
+  return await new Promise<T>((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");
+  });
+});

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: ["* * * * *"],
+});