1import type * as runtime from "@cloudflare/workers-types";
2import * as Effect from "effect/Effect";
3import { Worker, WorkerEnvironment } from "../Workers/Worker.ts";
4import type { Queue } from "./Queue.ts";
5import { maybeQueueShim } from "./QueueShim.ts";
6import { SendError } from "./QueueTypes.ts";
7
8/**
9 * Shared scaffolding for the Worker-binding implementations of the Queue
10 * services.
11 *
12 * Resolves the {@link WorkerEnvironment} and host {@link Worker}, registers
13 * the `queue` binding at deploy time, then delegates to `makeClient` with the
14 * shared {@link makeQueueHelpers} to build the producer client.
15 */
16export const makeQueueBinding = <Client>(options: {
17 makeClient: (helpers: ReturnType<typeof makeQueueHelpers>) => Client;
18}) =>
19 Effect.gen(function* () {
20 const env = yield* WorkerEnvironment;
21 const host = yield* Worker;
22 // Deploy-time ambient context (providers, stack). Captured at layer
23 // init so the shim registration below can run inside the callable —
24 // whose contract requires `R = never` — without leaking engine
25 // requirements into the binding's public type. Only dereferenced under
26 // the `__ALCHEMY_RUNTIME__` guard, where the context is the stack
27 // eval's (at runtime the captured context lacks providers, but the
28 // guarded branch never runs there).
29 const context = yield* Effect.context<never>();
30
31 return Effect.fn(function* (queue: Queue) {
32 if (!globalThis.__ALCHEMY_RUNTIME__) {
33 // A LOCAL host binding a LIVE queue needs the dev-mode
34 // remote-producer shim (see `QueueShim.ts`) — registered here as
35 // ordinary engine-managed resources, like the `AccountApiToken`
36 // the HTTP capability layers mint.
37 const shim = yield* maybeQueueShim(queue, host).pipe(
38 Effect.provideContext(context),
39 ) as Effect.Effect<Effect.Success<ReturnType<typeof maybeQueueShim>>>;
40 yield* host.bind`${queue}`({
41 bindings: [
42 {
43 type: "queue",
44 name: queue.LogicalId,
45 queueName: queue.queueName,
46 // Alchemy-only mode discriminator for dev (stripped before
47 // upload): a `dev:` id keeps the local broker, a real id
48 // (Alchemy.remote()) routes through the deployed shim.
49 queueId: queue.queueId,
50 ...(shim ? { shim } : {}),
51 },
52 ],
53 });
54 }
55
56 return options.makeClient(makeQueueHelpers(env, queue));
57 });
58 });
59
60/** Primitives shared by the Worker-binding producer clients. */
61export const makeQueueHelpers = (env: Record<string, any>, queue: Queue) => {
62 const raw = Effect.sync(
63 () => (env as Record<string, runtime.Queue<unknown>>)[queue.LogicalId]!,
64 );
65
66 const tryPromise = <T>(fn: () => Promise<T>): Effect.Effect<T, SendError> =>
67 Effect.tryPromise({
68 try: fn,
69 catch: (error: any) =>
70 new SendError({
71 message: error?.message ?? "Unknown queue error",
72 cause: error,
73 }),
74 });
75
76 const use = <T>(
77 fn: (raw: runtime.Queue<unknown>) => Promise<T>,
78 ): Effect.Effect<T, SendError> =>
79 raw.pipe(Effect.flatMap((raw) => tryPromise(() => fn(raw))));
80
81 return { raw, use, tryPromise };
82};