1import * as GCP from "alchemy/GCP";
2import * as Effect from "effect/Effect";
3import { HttpServerRequest } from "effect/http/HttpServerRequest";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import { Jobs, Results, resultPath, type JobMessage } from "./resources.ts";
6
7/**
8 * The public front of the job queue.
9 *
10 * - `POST /jobs` — enqueue `{ payload }` and return `202 { id }`.
11 * - `GET /jobs/:id` — `200` with the result once the worker is done,
12 * `202 { status: "pending" }` until then.
13 *
14 * The API never processes a job itself, so a slow or crashed worker only
15 * delays results; submissions keep succeeding.
16 */
17export default class Api extends GCP.Function<Api>()(
18 "Api",
19 {
20 main: import.meta.url,
21 invokerIamDisabled: true,
22 },
23 Effect.gen(function* () {
24 const jobs = yield* GCP.PubSub.WriteTopic(Jobs);
25 const results = yield* GCP.Firestore.ReadDatabase(Results);
26
27 return {
28 fetch: Effect.gen(function* () {
29 const request = yield* HttpServerRequest;
30 const url = new URL(request.originalUrl);
31 const segments = url.pathname.split("/").filter(Boolean);
32
33 if (request.method === "GET" && segments.length === 0) {
34 return HttpServerResponse.text("ok");
35 }
36
37 if (
38 request.method === "POST" &&
39 segments.length === 1 &&
40 segments[0] === "jobs"
41 ) {
42 const body = (yield* request.json.pipe(
43 Effect.orElseSucceed(() => ({})),
44 )) as { payload?: unknown };
45 if (typeof body.payload !== "string") {
46 return yield* HttpServerResponse.json(
47 { error: "payload must be a string" },
48 { status: 400 },
49 );
50 }
51
52 const job: JobMessage = {
53 id: crypto.randomUUID(),
54 payload: body.payload,
55 submittedAt: new Date().toISOString(),
56 };
57 yield* jobs
58 .publish({
59 data: JSON.stringify(job),
60 // Attributes are readable without decoding the body, e.g. by
61 // a subscription filter.
62 attributes: { jobId: job.id },
63 })
64 .pipe(Effect.orDie);
65
66 return yield* HttpServerResponse.json(
67 { id: job.id, status: "pending" },
68 { status: 202 },
69 );
70 }
71
72 if (
73 request.method === "GET" &&
74 segments.length === 2 &&
75 segments[0] === "jobs"
76 ) {
77 const id = segments[1]!;
78 const result = yield* results.get(resultPath(id)).pipe(Effect.orDie);
79 // No document yet means the job is queued or in flight. The API
80 // keeps no record of submissions, so an unknown id also reads
81 // as pending.
82 if (result === undefined) {
83 return yield* HttpServerResponse.json(
84 { id, status: "pending" },
85 { status: 202 },
86 );
87 }
88 return yield* HttpServerResponse.json({ id, ...result.fields });
89 }
90
91 return yield* HttpServerResponse.json(
92 { error: "not found" },
93 { status: 404 },
94 );
95 }),
96 };
97 }).pipe(
98 Effect.provide([GCP.PubSub.WriteTopicHttp, GCP.Firestore.ReadDatabaseHttp]),
99 ),
100) {}