1/**
2 * Dev-mode remote-producer shim for `Alchemy.remote()` queues.
3 *
4 * Cloudflare's preview/remote-binding sessions reject queue bindings
5 * outright (a preview worker carrying one serves 503 for every request;
6 * wrangler has the same gap — cloudflare/workers-sdk#9929), so a locally
7 * running worker cannot produce to a live queue through the usual
8 * remote-binding proxy. Instead, alchemy deploys a REAL shim worker that
9 * holds the actual queue binding:
10 *
11 * local worker ── env.QUEUE.send() ──▶ forwarder service (local workerd)
12 * POST /message | /batch + Authorization: Bearer <token>
13 * │
14 * ▼
15 * deployed shim worker (workers.dev)
16 * │ env.QUEUE.send(body, { contentType })
17 * ▼
18 * real Cloudflare Queue
19 *
20 * The shim and its bearer token are ordinary engine-managed resources
21 * ({@link Worker} + {@link Random}), conditionally instantiated at eval
22 * time when a LOCAL worker binds a LIVE queue — the same idiom as
23 * `AccountApiToken` (capability layers minting token resources) and
24 * event-source mappings. The engine gives the full lifecycle for free:
25 * Output-based ordering (shim deploys before the local worker serves),
26 * stable-name updates, deletion on destroy, and orphan GC on promotion
27 * (`alchemy deploy` no longer registers the shim, so it is cleaned up).
28 *
29 * NOT exported from `index.ts` — binding-internal scaffolding.
30 */
31import * as Effect from "effect/Effect";
32import type * as Redacted from "effect/Redacted";
33import type * as Output from "../../Output.ts";
34import {
35 defaultProviderMode,
36 remote,
37 type ProviderMode,
38} from "../../ProviderMode.ts";
39import { Random } from "../../Random.ts";
40import { Worker } from "../Workers/Worker.ts";
41import type { Queue } from "./Queue.ts";
42
43/**
44 * The deployed shim's script: decodes workerd's queue wire protocol
45 * (`POST /message` with raw bytes + `X-Msg-Fmt`, `POST /batch` with base64
46 * payloads) back into `send()`/`sendBatch()` calls on its real queue
47 * binding, gated by a bearer token.
48 *
49 * `v8`-serialized messages cannot cross the boundary — deserializing V8
50 * payloads in userland is not possible — so they are rejected with a clear
51 * error telling the caller to pass an explicit contentType.
52 */
53const QUEUE_SHIM_SCRIPT = `class HttpError extends Error {
54 constructor(status, message) {
55 super(message);
56 this.status = status;
57 }
58}
59
60function decodeBody(contentType, bytes) {
61 switch (contentType) {
62 case "text":
63 return { contentType, body: new TextDecoder().decode(bytes) };
64 case "json":
65 return { contentType, body: JSON.parse(new TextDecoder().decode(bytes)) };
66 case "bytes":
67 return { contentType, body: bytes.slice().buffer };
68 case "v8":
69 throw new HttpError(
70 422,
71 'v8-serialized queue messages cannot be produced to a live queue from local dev; pass an explicit contentType of "json", "text", or "bytes" to send()/sendBatch()',
72 );
73 default:
74 throw new HttpError(400, "message content type " + contentType + " is invalid");
75 }
76}
77
78function base64ToBytes(b64) {
79 const binary = atob(b64);
80 const bytes = new Uint8Array(binary.length);
81 for (let i = 0; i < binary.length; i++) {
82 bytes[i] = binary.charCodeAt(i);
83 }
84 return bytes;
85}
86
87function getDelay(value) {
88 if (!value) return undefined;
89 const delay = Number(value);
90 if (!Number.isInteger(delay) || delay < 0) {
91 throw new HttpError(400, "message delay " + value + " is invalid");
92 }
93 return delay;
94}
95
96// workerd's queue client requires this response shape (SendMetadata);
97// the metrics are informational and synthesized as zeros here.
98const sendResponse = () =>
99 Response.json({
100 metadata: {
101 metrics: { backlogCount: 0, backlogBytes: 0, oldestMessageTimestamp: 0 },
102 },
103 });
104
105export default {
106 async fetch(request, env) {
107 try {
108 const auth = request.headers.get("authorization");
109 if (auth !== "Bearer " + env.SHIM_TOKEN) {
110 throw new HttpError(401, "unauthorized");
111 }
112 if (request.method !== "POST") {
113 throw new HttpError(405, "method " + request.method + " not allowed");
114 }
115 const url = new URL(request.url);
116 switch (url.pathname) {
117 case "/message": {
118 // workerd's absent-header default is "v8" — mirrored here so an
119 // unmarked payload is rejected rather than misparsed.
120 const contentType = request.headers.get("X-Msg-Fmt") ?? "v8";
121 const delaySeconds = getDelay(request.headers.get("X-Msg-Delay-Secs"));
122 const bytes = new Uint8Array(await request.arrayBuffer());
123 const decoded = decodeBody(contentType, bytes);
124 await env.QUEUE.send(decoded.body, {
125 contentType: decoded.contentType,
126 ...(delaySeconds !== undefined && { delaySeconds }),
127 });
128 return sendResponse();
129 }
130 case "/batch": {
131 const delaySeconds = getDelay(request.headers.get("X-Msg-Delay-Secs"));
132 const body = await request.json();
133 const messages = body.messages.map((message) => {
134 const decoded = decodeBody(message.contentType, base64ToBytes(message.body));
135 const delay = message.delaySecs ?? delaySeconds;
136 return {
137 body: decoded.body,
138 contentType: decoded.contentType,
139 ...(delay !== undefined && { delaySeconds: delay }),
140 };
141 });
142 await env.QUEUE.sendBatch(messages);
143 return sendResponse();
144 }
145 default:
146 throw new HttpError(404, "not found");
147 }
148 } catch (e) {
149 if (e instanceof HttpError) {
150 return new Response(e.message, { status: e.status });
151 }
152 const message = e instanceof Error ? e.message : String(e);
153 return new Response("queue shim failed: " + message, { status: 500 });
154 }
155 },
156};`;
157
158export interface QueueShimBinding {
159 url: Output.Output<string | undefined>;
160 token: Output.Output<Redacted.Redacted<string>>;
161}
162
163/**
164 * Conditionally instantiate the queue shim for a producer binding: when a
165 * LOCAL host worker binds a LIVE queue (dev run + `Alchemy.remote()` on the
166 * queue), register the shim worker + its token and return their outputs
167 * for the binding data. Any other mode combination returns `undefined` —
168 * including the shim worker's own eval of its queue binding (the shim is
169 * pinned live, so no recursion).
170 */
171export const maybeQueueShim = Effect.fn(function* (
172 queue: Queue,
173 host: { Mode?: ProviderMode | undefined },
174) {
175 const runDefault = yield* defaultProviderMode;
176 const hostMode = host.Mode ?? runDefault;
177 const queueMode = queue.Mode ?? runDefault;
178 if (hostMode !== "local" || queueMode !== "live") {
179 return undefined;
180 }
181 const token = yield* Random(`${queue.LogicalId}ShimToken`);
182 const shim = yield* Worker(`${queue.LogicalId}Shim`, {
183 script: QUEUE_SHIM_SCRIPT,
184 env: {
185 QUEUE: queue,
186 SHIM_TOKEN: token.text,
187 },
188 }).pipe(remote());
189 return {
190 url: shim.url,
191 token: token.text,
192 } satisfies QueueShimBinding;
193});