diff --git a/packages/alchemy/src/Cloudflare/Workers/EmailEventSource.ts b/packages/alchemy/src/Cloudflare/Workers/EmailEventSource.ts index 13a4d5508a..88e7315f00 100644 --- a/packages/alchemy/src/Cloudflare/Workers/EmailEventSource.ts +++ b/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, ) { - // 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 diff --git a/packages/alchemy/src/Cloudflare/Workers/WorkerBridge.ts b/packages/alchemy/src/Cloudflare/Workers/WorkerBridge.ts index 21ee2e9b15..8592fd9afd 100644 --- a/packages/alchemy/src/Cloudflare/Workers/WorkerBridge.ts +++ b/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, - Context.Context, - ], - 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, + Context.Context, + ], + this.ctx, + this.env, + (exit) => + exit._tag === "Success" + ? Promise.resolve(exit.value) + : Promise.reject(Cause.squash(exit.cause)), ); }, writable: true, diff --git a/packages/alchemy/test/Cloudflare/Workers/EmailEventSource.local.test.ts b/packages/alchemy/test/Cloudflare/Workers/EmailEventSource.local.test.ts new file mode 100644 index 0000000000..317d0aee59 --- /dev/null +++ b/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: ", + "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> }; + 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 }, +); diff --git a/packages/alchemy/test/Cloudflare/Workers/fixtures/email-subscribe-local-worker.ts b/packages/alchemy/test/Cloudflare/Workers/fixtures/email-subscribe-local-worker.ts new file mode 100644 index 0000000000..a3ca93ec63 --- /dev/null +++ b/packages/alchemy/test/Cloudflare/Workers/fixtures/email-subscribe-local-worker.ts @@ -0,0 +1,62 @@ +import * as Cloudflare from "@/Cloudflare/index.ts"; +import * as Effect from "effect/Effect"; +import { HttpServerRequest } from "effect/unstable/http/HttpServerRequest"; +import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse"; + +interface ReceivedMessage { + from: string; + to: string; + subject: string | null; + bodySize: number; + body: string; +} + +/** + * Effect-native Worker driven through `Cloudflare.email().subscribe(...)`, + * exercised against the local simulator rather than a deployed Worker. + * + * `zone` is deliberately omitted: under `alchemy dev` the event source skips + * the deploy-time half anyway, and leaving it out keeps this fixture from + * naming a real zone. The runtime listener is what's under test — the local + * runtime's `POST /cdn-cgi/handler/email` trigger route dispatches to the + * same `email` export the deployed Worker uses. + */ +export default class EmailSubscribeLocalWorker extends Cloudflare.Worker()( + "EmailSubscribeLocalWorker", + { main: import.meta.filename }, + Effect.gen(function* () { + const received: ReceivedMessage[] = []; + + yield* Cloudflare.email().subscribe((message) => + Effect.gen(function* () { + // Reject on demand so the trigger route's 400-with-reason path is + // covered alongside the accept path. + const subject = message.headers.get("subject"); + if (subject === "reject-me") { + return yield* message.setReject("I don't like this email"); + } + const body = yield* Effect.promise(() => + new Response(message.body as any).text(), + ); + received.push({ + from: message.from, + to: message.to, + subject, + bodySize: message.bodySize, + body, + }); + }), + ); + + return { + fetch: Effect.gen(function* () { + const request = yield* HttpServerRequest; + const url = new URL(request.url, "http://x"); + if (url.pathname === "/received") { + return yield* HttpServerResponse.json({ received }); + } + return HttpServerResponse.text("ok"); + }), + }; + }).pipe(Effect.provide(Cloudflare.EmailEventSourceLive)), +) {}