1import * as queues from "@distilled.cloud/cloudflare/queues";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import { authorizeWith } from "../HttpClientUtils.ts";
5import {
6 makeHttpQueueBinding,
7 makeQueueHttpScope,
8 toQueueSendError,
9 type QueueAuth,
10} from "./QueueHttp.ts";
11import { SendError, type SendMessage } from "./QueueTypes.ts";
12import { WriteQueue, type WriteQueueClient } from "./WriteQueue.ts";
13
14/**
15 * HTTP-backed implementation of the {@link WriteQueue} service.
16 *
17 * It creates a scoped {@link AccountApiToken} with the `Queues Write`
18 * permission and pushes messages via the Cloudflare Queues bulk-push
19 * HTTP API (`POST /messages/batch`). The bulk endpoint takes the raw
20 * JSON value as the message `body` (the single-message endpoint
21 * expects a pre-encoded string), so it round-trips arbitrary
22 * JSON-serializable payloads the same way the native producer binding
23 * does — `send` is just a batch of one.
24 */
25export const WriteQueueHttp = Layer.effect(
26 WriteQueue,
27 Effect.suspend(() =>
28 makeHttpQueueBinding({
29 permissionGroups: ["Queues Write"],
30 makeClient: (token, queueId) =>
31 makeWriteQueueHttpClient(
32 { authorize: authorizeWith(token), accountId: token.accountId },
33 queueId,
34 ),
35 }),
36 ),
37);
38
39/** Convert a message into the Cloudflare bulk-push `messages[]` shape. */
40const toMessage = (message: SendMessage) =>
41 message.contentType === "text"
42 ? {
43 body:
44 typeof message.body === "string"
45 ? message.body
46 : String(message.body),
47 contentType: "text" as const,
48 }
49 : { body: message.body, contentType: "json" as const };
50
51/**
52 * Build the producer client over the Queues bulk-push HTTP API.
53 *
54 * Shared by {@link WriteQueueHttp} (scoped token auth) and
55 * `WriteQueueLocal` (current-credentials auth) — they differ only in the
56 * {@link QueueAuth} they inject.
57 */
58export const makeWriteQueueHttpClient = (
59 auth: QueueAuth,
60 queueId: Effect.Effect<string>,
61): WriteQueueClient => {
62 const scope = makeQueueHttpScope(auth, queueId);
63
64 const push = (messages: ReadonlyArray<SendMessage>) =>
65 scope.pipe(
66 Effect.flatMap(({ accountId, queueId }) =>
67 auth.authorize(
68 queues.bulkPushMessages({
69 accountId,
70 queueId,
71 messages: messages.map(toMessage),
72 }),
73 ),
74 ),
75 Effect.mapError(toQueueSendError),
76 Effect.asVoid,
77 );
78
79 return {
80 raw: Effect.die(
81 new SendError({
82 message:
83 "Queue HTTP client does not expose a native Queue binding; use send/sendBatch.",
84 cause: new Error("unsupported"),
85 }),
86 ),
87 send: (body: unknown, options?: { contentType?: "json" | "text" }) =>
88 push([{ body, contentType: options?.contentType }]),
89 sendBatch: (messages: ReadonlyArray<SendMessage>) => push(messages),
90 };
91};