fix(cloudflare/workers): simulate inbound email in alchemy dev

packages/alchemy/src/Cloudflare/Workers/EmailEventSource.ts

@@ -4,6 +4,7 @@ import * as Context from "effect/Context";
 import * as Data from "effect/Data";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import { AlchemyContext } from "../../AlchemyContext.ts";
 import type { Input } from "../../Input.ts";
 import * as Namespace from "../../Namespace.ts";
 import * as RemovalPolicy from "../../RemovalPolicy.ts";
@@ -263,56 +264,75 @@ export const EmailEventSourceLive = Layer.effect(
         message: ForwardableEmailMessage,
       ) => Effect.Effect<void, E, Req>,
     ) {
-      // Deploy-time: provision the Email.Routing toggle plus the routing
-      // resource that hands matched mail to this Worker. Skipped once
-      // running inside the deployed Worker (the global guard) and when
-      // `zone` is omitted (bring-your-own routing). Namespaced under the
-      // host so logical identity is stable per Worker.
-      if (!globalThis.__ALCHEMY_RUNTIME__ && props.zone !== undefined) {
-        const zone = props.zone;
-        const matchers = props.matchers;
-        yield* Namespace.push(
-          host.LogicalId,
-          Effect.gen(function* () {
-            // Routing is a per-zone singleton shared with other rules on
-            // the zone, so destroying this Worker must not disable it.
-            yield* Routing("EmailRouting", {
-              zone,
-              enabled: true,
-            }).pipe(RemovalPolicy.retain());
+      // Deploy-time only: provision the Email.Routing toggle plus the
+      // routing resource that hands matched mail to this Worker.
+      //
+      // The guard is a bare `__ALCHEMY_RUNTIME__` check with every other
+      // condition nested inside, so the bundler folds it away and drops the
+      // whole block — and with it `AlchemyContext`, `Namespace`, `Routing`,
+      // `CatchAll` and `Rule` — from the deployed Worker. Hoisting the `dev`
+      // lookup out into a compound condition would keep it (and the context
+      // it reaches for) live in the runtime bundle.
+      if (!globalThis.__ALCHEMY_RUNTIME__) {
+        // Under `alchemy dev` the Worker only exists locally, so Cloudflare's
+        // mail pipeline has nothing to deliver to. Pointing a real zone's
+        // catch-all at a script that was never uploaded would fail — and if it
+        // did land, it would silently take over inbound mail for the whole zone
+        // and drop it. Local inbound is driven by the runtime's
+        // `POST /cdn-cgi/handler/email?from=&to=` trigger route instead, which
+        // dispatches to the same listener registered below.
+        const dev = yield* Effect.serviceOption(AlchemyContext).pipe(
+          Effect.map((ctx) => (ctx._tag === "Some" ? ctx.value.dev : false)),
+        );
+
+        // Also skipped when `zone` is omitted (bring-your-own routing).
+        // Namespaced under the host so logical identity is stable per Worker.
+        if (props.zone !== undefined && !dev) {
+          const zone = props.zone;
+          const matchers = props.matchers;
+          yield* Namespace.push(
+            host.LogicalId,
+            Effect.gen(function* () {
+              // Routing is a per-zone singleton shared with other rules on
+              // the zone, so destroying this Worker must not disable it.
+              yield* Routing("EmailRouting", {
+                zone,
+                enabled: true,
+              }).pipe(RemovalPolicy.retain());
+
+              const action = {
+                type: "worker" as const,
+                value: [host.workerName],
+              };
 
-            const action = {
-              type: "worker" as const,
-              value: [host.workerName],
-            };
+              // Catch-all is a per-zone SINGLETON living behind
+              // `/rules/catch_all`, not an ordinary rule. Cloudflare surfaces
+              // it in `listRules` but rejects mutating it through the rule
+              // endpoint ("Invalid rule operation"), so creating it as an
+              // `Email.Rule` would produce a row the engine cannot delete.
+              // Route an all-matcher subscription to `Email.CatchAll`
+              // instead — the resource that owns that endpoint.
+              if (isCatchAll(matchers)) {
+                yield* CatchAll("EmailCatchAll", {
+                  zone,
+                  name: props.ruleName ?? host.LogicalId,
+                  enabled: props.enabled ?? true,
+                  actions: [action],
+                });
+                return;
+              }
 
-            // Catch-all is a per-zone SINGLETON living behind
-            // `/rules/catch_all`, not an ordinary rule. Cloudflare surfaces
-            // it in `listRules` but rejects mutating it through the rule
-            // endpoint ("Invalid rule operation"), so creating it as an
-            // `Email.Rule` would produce a row the engine cannot delete.
-            // Route an all-matcher subscription to `Email.CatchAll`
-            // instead — the resource that owns that endpoint.
-            if (isCatchAll(matchers)) {
-              yield* CatchAll("EmailCatchAll", {
+              yield* Rule("EmailRule", {
                 zone,
                 name: props.ruleName ?? host.LogicalId,
                 enabled: props.enabled ?? true,
+                priority: props.priority ?? 0,
+                matchers: matchers!,
                 actions: [action],
               });
-              return;
-            }
-
-            yield* Rule("EmailRule", {
-              zone,
-              name: props.ruleName ?? host.LogicalId,
-              enabled: props.enabled ?? true,
-              priority: props.priority ?? 0,
-              matchers: matchers!,
-              actions: [action],
-            });
-          }),
-        );
+            }),
+          );
+        }
       }
 
       // Resolve the runtime context per-call rather than at layer

packages/alchemy/src/Cloudflare/Workers/WorkerBridge.ts

@@ -152,23 +152,6 @@ export const makeWorkerBridge = (
     ) {
       super(ctx, env);
 
-      for (const methodName of ExportedHandlerMethods) {
-        (this as any)[methodName] = async (input: any) =>
-          processEvent(
-            (built) =>
-              built.export[methodName](input, this.env, this.ctx) as [
-                Effect.Effect<any>,
-                Context.Context<never>,
-              ],
-            this.ctx,
-            this.env,
-            (exit) =>
-              exit._tag === "Success"
-                ? Promise.resolve(exit.value)
-                : Promise.reject(Cause.squash(exit.cause)),
-          );
-      }
-
       return new Proxy(this, {
         get: (target, prop) => {
           if (typeof prop !== "string") return (target as any)[prop];
@@ -213,14 +196,40 @@ export const makeWorkerBridge = (
     }
   }
 
-  // Stub prototype methods so Cloudflare's script-validate detects the
-  // standard handler set; per-instance overrides above are what actually
-  // run.
+  // The standard handler set lives on the *prototype*, never as own
+  // properties of the instance. Two things depend on that:
+  //
+  //  - Cloudflare's script-validate reads the prototype to detect which
+  //    handlers a class entrypoint implements.
+  //  - workerd's JSRPC method lookup only resolves methods found on the
+  //    prototype chain. An own instance property of the same name *shadows*
+  //    the prototype entry and makes the lookup fail outright with
+  //    `The RPC receiver does not implement the method "..."` — it does not
+  //    fall back to the shadowed prototype method.
+  //
+  // Most handlers never exercise the JSRPC path: workerd dispatches
+  // `fetch`/`scheduled`/`queue` as built-in events, and Cloudflare's mail
+  // pipeline delivers `email` as an event too. The local runtime is the
+  // exception — its entry worker forwards the `/cdn-cgi/handler/email`
+  // trigger route to the user worker as `env[USER_WORKER].email(message)`,
+  // a plain JSRPC call — so assigning the handlers in the constructor made
+  // inbound email the one handler that worked deployed but not in
+  // `alchemy dev`. See `EmailEventSource.local.test.ts`.
   for (const method of ExportedHandlerMethods) {
     Object.defineProperty(WorkerBridge.prototype, method, {
-      value: function () {
-        throw new Error(
-          `Bridge method '${method}' was called before instance setup`,
+      value: function (this: any, input: any) {
+        return processEvent(
+          (built) =>
+            built.export[method](input, this.env, this.ctx) as [
+              Effect.Effect<any>,
+              Context.Context<never>,
+            ],
+          this.ctx,
+          this.env,
+          (exit) =>
+            exit._tag === "Success"
+              ? Promise.resolve(exit.value)
+              : Promise.reject(Cause.squash(exit.cause)),
         );
       },
       writable: true,

packages/alchemy/test/Cloudflare/Workers/EmailEventSource.local.test.ts

@@ -0,0 +1,140 @@
+import * as Cloudflare from "@/Cloudflare/index.ts";
+import * as Test from "@/Test/Alchemy";
+import { expect } from "alchemy-test";
+import * as Data from "effect/Data";
+import * as Effect from "effect/Effect";
+import { MinimumLogLevel } from "effect/References";
+import * as Schedule from "effect/Schedule";
+import * as HttpClient from "effect/unstable/http/HttpClient";
+import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
+import EmailSubscribeLocalWorker from "./fixtures/email-subscribe-local-worker.ts";
+
+// `dev: true` runs local providers behind the RPC sidecar proxy by default,
+// matching the process topology of the real `alchemy dev` command.
+const { test } = Test.make({
+  providers: Cloudflare.providers(),
+  dev: true,
+});
+
+const logLevel = Effect.provideService(
+  MinimumLogLevel,
+  process.env.DEBUG ? "Debug" : "Info",
+);
+
+class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
+  status: number;
+}> {}
+
+const getReady = (url: string) =>
+  Effect.gen(function* () {
+    const client = yield* HttpClient.HttpClient;
+    return yield* client.get(url).pipe(
+      Effect.flatMap((res) =>
+        res.status === 200
+          ? Effect.succeed(res)
+          : Effect.fail(new WorkerNotReady({ status: res.status })),
+      ),
+      Effect.retry({
+        while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
+        schedule: Schedule.max([
+          Schedule.min([
+            Schedule.exponential("500 millis"),
+            Schedule.spaced("2 seconds"),
+          ]),
+          Schedule.recurs(10),
+        ]),
+      }),
+    );
+  }).pipe(Effect.orDie);
+
+const FROM = "someone@example.com";
+const TO = "inbox@example.com";
+
+const incomingEmail = (subject: string) =>
+  [
+    `From: someone <${FROM}>`,
+    `To: inbox <${TO}>`,
+    `Subject: ${subject}`,
+    "Message-ID: <local-subscribe@example.com>",
+    "MIME-Version: 1.0",
+    "Content-Type: text/plain",
+    "",
+    "hello from the local trigger route",
+  ].join("\n");
+
+const postEmail = (workerUrl: string, raw: string) =>
+  Effect.gen(function* () {
+    const client = yield* HttpClient.HttpClient;
+    return yield* client.execute(
+      HttpClientRequest.post(
+        `${workerUrl}/cdn-cgi/handler/email?from=${encodeURIComponent(FROM)}&to=${encodeURIComponent(TO)}`,
+      ).pipe(HttpClientRequest.bodyText(raw)),
+    );
+  });
+
+/**
+ * `Cloudflare.email().subscribe(...)` against the local simulator.
+ *
+ * The local entry worker delivers inbound mail as a JSRPC call,
+ * `env[USER_WORKER].email(message)` — the same shape Cloudflare's mail
+ * pipeline uses against a deployed Worker. workerd resolves JSRPC methods
+ * only on the target entrypoint's *prototype chain*, so this is a
+ * regression guard for `WorkerBridge` keeping the handler set there: an own
+ * instance property of the same name shadows the prototype entry and the
+ * call fails with `The RPC receiver does not implement the method "email"`.
+ *
+ * `fetch`/`scheduled`/`queue` cannot catch that — workerd dispatches those
+ * through its built-in event path rather than over RPC — which is why this
+ * test earns its keep alongside `CronEventSource.local.test.ts`.
+ */
+test.provider(
+  "the local email trigger dispatches to a subscribe() handler",
+  (stack) =>
+    Effect.gen(function* () {
+      yield* stack.destroy();
+
+      // Yield the fixture class itself: it is an Effect-native Worker, so
+      // its `main: import.meta.filename` and init effect have to come from
+      // the class rather than a generic Worker pointed at the same file.
+      const deployed = yield* stack.deploy(
+        Effect.gen(function* () {
+          const worker = yield* EmailSubscribeLocalWorker;
+          return { worker };
+        }),
+      );
+
+      // Serving from the local dev proxy — proof nothing was deployed.
+      expect(deployed.worker.url).toMatch(/^http:\/\/localhost:\d+$/);
+      yield* getReady(deployed.worker.url!);
+
+      // 1. Accepted message reaches the subscribe handler.
+      const raw = incomingEmail("accept-me");
+      const res = yield* postEmail(deployed.worker.url!, raw);
+      expect(res.status).toBe(200);
+
+      const snapshot = (yield* (yield* getReady(
+        `${deployed.worker.url}/received`,
+      )).json) as { received: Array<Record<string, unknown>> };
+      expect(snapshot.received).toHaveLength(1);
+      const message = snapshot.received[0]!;
+      // Envelope addresses come from the trigger route's query parameters.
+      expect(message.from).toBe(FROM);
+      expect(message.to).toBe(TO);
+      expect(message.subject).toBe("accept-me");
+      // `bodySize` is the wrapper's name for cf's `rawSize`.
+      expect(message.bodySize).toBe(new TextEncoder().encode(raw).byteLength);
+      expect(message.body).toBe(raw);
+
+      // 2. `setReject` — an Effect on the wrapper — surfaces as 400 with the
+      //    reason, same as the raw-handler path.
+      const rejected = yield* postEmail(
+        deployed.worker.url!,
+        incomingEmail("reject-me"),
+      );
+      expect(rejected.status).toBe(400);
+      expect(yield* rejected.text).toContain("I don't like this email");
+
+      yield* stack.destroy();
+    }).pipe(logLevel),
+  { timeout: 180_000 },
+);