1import { BetterAuth } from "@alchemy.run/better-auth";
2import { CloudflareD1 } from "@alchemy.run/better-auth/CloudflareD1";
3import * as Cloudflare from "alchemy/Cloudflare";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7import * as HttpBody from "effect/http/HttpBody";
8import * as HttpClientRequest from "effect/http/HttpClientRequest";
9import { HttpServerRequest } from "effect/http/HttpServerRequest";
10import * as HttpServerResponse from "effect/http/HttpServerResponse";
11import Agent from "./Agent.ts";
12import { Gateway } from "./AiGateway.ts";
13import { Bucket } from "./Bucket.ts";
14import { KV } from "./KV.ts";
15import NotifyWorkflow from "./NotifyWorkflow.ts";
16import { Queue } from "./Queue.ts";
17import { Repos } from "./Repos.ts";
18import Room from "./Room.ts";
19
20interface QueueMessageBody {
21 id: string;
22 text: string;
23 sentAt: number;
24}
25
26/** D1 database backing Better Auth (auto-migrated at deploy). */
27export const AuthDb = Cloudflare.D1.Database("AuthDb");
28
29export default class Api extends Cloudflare.Worker<Api>()(
30 "Api",
31 {
32 main: import.meta.url,
33 observability: {
34 enabled: true,
35 },
36 assets: "./assets",
37 build: {
38 bundleAnalyzer: true,
39 },
40 compatibility: {
41 date: "2026-08-31",
42 },
43 },
44 Effect.gen(function* () {
45 const auth = yield* BetterAuth({
46 basePath: "/auth",
47 emailAndPassword: { enabled: true },
48 });
49 const agents = yield* Agent;
50 const rooms = yield* Room;
51 const notifier = yield* NotifyWorkflow;
52 const loader = yield* Cloudflare.WorkerLoader("Loader");
53 const bucket = yield* Cloudflare.R2.ReadWriteBucket(Bucket);
54 const kv = yield* Cloudflare.KV.ReadWriteNamespace(KV);
55 const queueResource = yield* Queue;
56 const queue = yield* Cloudflare.Queues.WriteQueue(queueResource);
57 const repos = yield* Cloudflare.Artifacts.ReadWriteNamespace(Repos);
58 const aiGateway = yield* Cloudflare.AI.QueryGateway(Gateway);
59
60 // Effect-style queue consumer. Each batch is piped through the
61 // handler; success ack()s every message in the batch, failure
62 // retry()s. The persisted JSON at /queue/<id> on R2 lets the
63 // integ test verify the producer→consumer round-trip.
64 yield* Cloudflare.Queues.consumeQueueMessages<QueueMessageBody>(
65 queueResource,
66 (stream) =>
67 Stream.runForEach(stream, (msg) =>
68 bucket
69 .put(`/queue/${msg.body.id}`, JSON.stringify(msg.body), {
70 httpMetadata: { contentType: "application/json" },
71 })
72 .pipe(Effect.asVoid),
73 ),
74 );
75
76 return {
77 fetch: Effect.gen(function* () {
78 const request = yield* HttpServerRequest;
79
80 if (request.url.startsWith("/auth/")) {
81 return yield* auth.fetch;
82 } else if (request.url.startsWith("/kv/")) {
83 if (request.method === "GET") {
84 const key = request.url.split("/").pop()!;
85 return yield* kv.get(key).pipe(
86 Effect.map((value) =>
87 value
88 ? HttpServerResponse.text(value)
89 : HttpServerResponse.empty({ status: 404 }),
90 ),
91 Effect.catch(() =>
92 Effect.succeed(HttpServerResponse.empty({ status: 404 })),
93 ),
94 );
95 } else if (request.method === "POST") {
96 const key = request.url.split("/").pop()!;
97 const value = yield* request.text;
98 return yield* kv.put(key, value).pipe(
99 Effect.map(() => HttpServerResponse.empty({ status: 200 })),
100 Effect.catch(() =>
101 Effect.succeed(HttpServerResponse.empty({ status: 500 })),
102 ),
103 );
104 }
105 } else if (request.url.startsWith("/object/")) {
106 if (request.method === "GET") {
107 return yield* bucket.get(request.url.split("/").pop()!).pipe(
108 Effect.flatMap((object) =>
109 object === null
110 ? Effect.succeed(
111 HttpServerResponse.text("Object not found", {
112 status: 404,
113 }),
114 )
115 : object.text().pipe(
116 Effect.map((text) =>
117 HttpServerResponse.text(text, {
118 headers: { "content-type": "application/json" },
119 }),
120 ),
121 ),
122 ),
123 Effect.catchTag("R2Error", (error) =>
124 Effect.succeed(
125 HttpServerResponse.text(error.message, {
126 status: 500,
127 statusText: error.message,
128 }),
129 ),
130 ),
131 );
132 } else if (request.method === "POST" || request.method === "PUT") {
133 // const request = yield* Cloudflare.Workers.Request
134 const key = request.url.split("/").pop()!;
135 return yield* bucket
136 .put(key, request.stream, {
137 contentLength: Number(request.headers["content-length"] ?? 0),
138 })
139 .pipe(
140 Effect.map(() => HttpServerResponse.empty({ status: 201 })),
141 Effect.catch((err) =>
142 HttpServerResponse.json(
143 {
144 error: err.message,
145 headers: request.headers,
146 },
147 { status: 500 },
148 ),
149 ),
150 );
151 } else {
152 return HttpServerResponse.text("Method not allowed", {
153 status: 405,
154 });
155 }
156 } else if (request.url === "/sandbox/increment") {
157 const agent = agents.getByName("sandbox-test");
158 const body = yield* agent.increment().pipe(Effect.orDie);
159 const room = rooms.getByName("default");
160 yield* room.broadcast(`[container] ${body}`);
161 return HttpServerResponse.text(body, {
162 headers: { "content-type": "application/json" },
163 });
164 } else if (request.url.startsWith("/sandbox")) {
165 const agent = agents.getByName("sandbox-test");
166 const body = yield* agent.hello().pipe(Effect.orDie);
167 return HttpServerResponse.text(body);
168 } else if (request.url.startsWith("/workflow/start/")) {
169 const roomId = request.url.split("/workflow/start/")[1];
170 if (!roomId) {
171 return yield* HttpServerResponse.json(
172 { error: "roomId is required" },
173 { status: 400 },
174 );
175 }
176 const instance = yield* notifier.create({
177 params: {
178 roomId,
179 message: "hello from workflow",
180 },
181 });
182 return yield* HttpServerResponse.json({ instanceId: instance.id });
183 } else if (request.url.startsWith("/workflow/status/")) {
184 const instanceId = request.url.split("/workflow/status/")[1];
185 if (!instanceId) {
186 return yield* HttpServerResponse.json(
187 { error: "instanceId is required" },
188 { status: 400 },
189 );
190 }
191 const instance = yield* notifier.get(instanceId);
192 const status = yield* instance.status();
193 return yield* HttpServerResponse.json(status);
194 } else if (request.url.startsWith("/eval")) {
195 if (request.method === "POST") {
196 const code = yield* request.text;
197 const worker = yield* loader.load({
198 compatibilityDate: "2026-08-31",
199 mainModule: "worker.js",
200 modules: {
201 "worker.js": `
202 export default {
203 async fetch(request) {
204 try {
205 const result = (0, eval)(${"`${await request.text()}`"});
206 return new Response(String(result), { status: 200 });
207 } catch (e) {
208 return new Response(e.message, { status: 500 });
209 }
210 }
211 }
212 `,
213 },
214 globalOutbound: null,
215 });
216 return yield* worker
217 .fetch(
218 HttpClientRequest.post("https://worker/").pipe(
219 HttpClientRequest.setBody(HttpBody.text(code)),
220 ),
221 )
222 .pipe(
223 Effect.map(HttpServerResponse.fromClientResponse),
224 Effect.orDie,
225 );
226 }
227 } else if (request.url.startsWith("/connect/")) {
228 const agentId = request.url.split("/").pop()!;
229 const agent = agents.getByName(agentId);
230 const response = yield* agent.fetch(request);
231 return response;
232 } else if (request.url.startsWith("/room/")) {
233 const upgradeHeader = request.headers.upgrade;
234 const roomId = request.url.split("/").pop()!;
235 if (!upgradeHeader || upgradeHeader !== "websocket") {
236 return HttpServerResponse.text(
237 "Worker expected Upgrade: websocket",
238 { status: 426 },
239 );
240 } else if (request.method !== "GET") {
241 return HttpServerResponse.text("Method not allowed", {
242 status: 405,
243 });
244 }
245 const room = rooms.getByName(roomId);
246 const response = yield* room.fetch(request);
247 return response;
248 }
249 // Cloudflare Artifacts — Git-compatible versioned repos.
250 // Exercises Cloudflare.ArtifactsConnection by creating a repo,
251 // looking it up, and minting short-lived clone tokens.
252 if (
253 request.url.startsWith("/repos/create") &&
254 request.method === "POST"
255 ) {
256 const text = yield* request.text;
257 const body = JSON.parse(text || "{}") as {
258 name?: string;
259 description?: string;
260 };
261 const name = body.name?.trim();
262 if (!name) {
263 return yield* HttpServerResponse.json(
264 { error: "name is required" },
265 { status: 400 },
266 );
267 }
268 return yield* repos
269 .create(name, {
270 description: body.description,
271 setDefaultBranch: "main",
272 })
273 .pipe(
274 Effect.flatMap((created) =>
275 HttpServerResponse.json({
276 id: created.id,
277 name: created.name,
278 remote: created.remote,
279 token: created.token,
280 tokenExpiresAt: created.tokenExpiresAt,
281 defaultBranch: created.defaultBranch,
282 }),
283 ),
284 Effect.catchTag("ArtifactsError", (err) =>
285 HttpServerResponse.json(
286 { error: err.message },
287 { status: 409 },
288 ),
289 ),
290 );
291 }
292 if (request.url.startsWith("/repos/list") && request.method === "GET") {
293 return yield* repos.list({ limit: 50 }).pipe(
294 Effect.flatMap((res) => HttpServerResponse.json(res)),
295 Effect.catchTag("ArtifactsError", (err) =>
296 HttpServerResponse.json({ error: err.message }, { status: 500 }),
297 ),
298 );
299 }
300 if (request.url.startsWith("/repos/info") && request.method === "GET") {
301 const name = new URL(request.url, "http://x").searchParams.get(
302 "name",
303 );
304 if (!name) {
305 return yield* HttpServerResponse.json(
306 { error: "name is required" },
307 { status: 400 },
308 );
309 }
310 return yield* repos.get(name).pipe(
311 Effect.flatMap((repo) =>
312 HttpServerResponse.json({
313 id: repo.raw.id,
314 name: repo.raw.name,
315 description: repo.raw.description,
316 defaultBranch: repo.raw.defaultBranch,
317 remote: repo.raw.remote,
318 createdAt: repo.raw.createdAt,
319 updatedAt: repo.raw.updatedAt,
320 lastPushAt: repo.raw.lastPushAt,
321 readOnly: repo.raw.readOnly,
322 }),
323 ),
324 Effect.catchTag("ArtifactsError", (err) =>
325 HttpServerResponse.json(
326 { name, error: err.message },
327 { status: 404 },
328 ),
329 ),
330 );
331 }
332 if (
333 request.url.startsWith("/repos/token") &&
334 request.method === "POST"
335 ) {
336 const text = yield* request.text;
337 const body = JSON.parse(text || "{}") as {
338 name?: string;
339 scope?: "read" | "write";
340 ttl?: number;
341 };
342 const name = body.name?.trim();
343 if (!name) {
344 return yield* HttpServerResponse.json(
345 { error: "name is required" },
346 { status: 400 },
347 );
348 }
349 return yield* repos.get(name).pipe(
350 Effect.flatMap((repo) =>
351 repo.createToken(body.scope ?? "read", body.ttl ?? 3600),
352 ),
353 Effect.flatMap((token) =>
354 HttpServerResponse.json({ name, ...token }),
355 ),
356 Effect.catchTag("ArtifactsError", (err) =>
357 HttpServerResponse.json(
358 { name, error: err.message },
359 { status: 404 },
360 ),
361 ),
362 );
363 }
364 // Queue producer + consumer smoke test.
365 //
366 // POST /queue/send sends a message with a generated id.
367 // GET /queue/result/:id reads the bucket entry the consumer
368 // wrote when it processed that message.
369 //
370 // Producer side: `Cloudflare.Queues.WriteQueue`. Consumer side:
371 // `Cloudflare.Queues.consumeQueueMessages(Queue, handler)` registered in
372 // the init phase (above), with `EventSourceLive` on the
373 // worker layer.
374 // AI Gateway smoke test — POST /ai with { prompt }.
375 //
376 // Routes a Workers AI inference call through the gateway resource so
377 // every request is observable in the Cloudflare.AI. Gateway UI and
378 // benefits from caching/rate limiting configured on the resource.
379 if (request.url.startsWith("/ai") && request.method === "POST") {
380 const text = yield* request.text;
381 const body = (() => {
382 try {
383 return JSON.parse(text || "{}") as { prompt?: string };
384 } catch {
385 return {} as { prompt?: string };
386 }
387 })();
388 const prompt =
389 body.prompt?.trim() || "Say hello in one short sentence.";
390 const response = yield* aiGateway.run({
391 provider: "workers-ai",
392 endpoint: "@cf/meta/llama-3.1-8b-instruct",
393 headers: { "content-type": "application/json" },
394 query: { prompt },
395 });
396 return HttpServerResponse.fromWeb(response);
397 }
398 if (request.url === "/queue/send" && request.method === "POST") {
399 const text = yield* request.text;
400 const msg: QueueMessageBody = {
401 id: crypto.randomUUID(),
402 text,
403 sentAt: Date.now(),
404 };
405 yield* queue.send(msg).pipe(Effect.orDie);
406 return yield* HttpServerResponse.json({ sent: msg }, { status: 202 });
407 }
408 if (request.url.startsWith("/queue/result/")) {
409 const id = request.url.split("/queue/result/")[1];
410 if (!id) {
411 return HttpServerResponse.text("missing id", { status: 400 });
412 }
413 if (request.method === "GET") {
414 return yield* bucket.get(`/queue/${id}`).pipe(
415 Effect.flatMap((object) =>
416 object === null
417 ? Effect.succeed(
418 HttpServerResponse.text("not yet", { status: 404 }),
419 )
420 : object.text().pipe(
421 Effect.map((body) =>
422 HttpServerResponse.text(body, {
423 headers: { "content-type": "application/json" },
424 }),
425 ),
426 ),
427 ),
428 Effect.catchTag("R2Error", (error) =>
429 Effect.succeed(
430 HttpServerResponse.text(error.message, { status: 500 }),
431 ),
432 ),
433 );
434 }
435 // DELETE — used by the integ test to clear consumed
436 // entries before stack.destroy(), so Bucket delete
437 // doesn't fail with "bucket not empty".
438 if (request.method === "DELETE") {
439 return yield* bucket.delete(`/queue/${id}`).pipe(
440 Effect.map(() => HttpServerResponse.empty({ status: 204 })),
441 Effect.catchTag("R2Error", (error) =>
442 Effect.succeed(
443 HttpServerResponse.text(error.message, { status: 500 }),
444 ),
445 ),
446 );
447 }
448 }
449 return HttpServerResponse.text("Not Found", { status: 404 });
450 }).pipe(
451 Effect.catch(() =>
452 Effect.succeed(
453 HttpServerResponse.text("Internal Server Error", {
454 status: 500,
455 }),
456 ),
457 ),
458 ),
459 };
460 }).pipe(
461 Effect.provide(
462 Layer.mergeAll(
463 CloudflareD1(AuthDb),
464 Cloudflare.R2.ReadWriteBucketBinding,
465 Cloudflare.KV.ReadWriteNamespaceBinding,
466 Cloudflare.Queues.WriteQueueBinding,
467 Cloudflare.Queues.EventSourceLive,
468 Cloudflare.Artifacts.ReadWriteNamespaceBinding,
469 Cloudflare.AI.QueryGatewayBinding,
470 ),
471 ),
472 ),
473) {}