1import * as Alchemy from "alchemy";
2import * as GCP from "alchemy/GCP";
3import * as Test from "alchemy/Test/Bun";
4import * as firestore from "@distilled.cloud/gcp/firestore_v1";
5import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
6import * as run from "@distilled.cloud/gcp/run_v2";
7import { describe, expect } from "bun:test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Schedule from "effect/Schedule";
12import * as FetchHttpClient from "effect/http/FetchHttpClient";
13import * as HttpClient from "effect/http/HttpClient";
14import * as HttpClientRequest from "effect/http/HttpClientRequest";
15import { spawnSync } from "node:child_process";
16import { createHash } from "node:crypto";
17import Stack from "../alchemy.run.ts";
18
19const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
20 providers: GCP.providers(),
21 state: Alchemy.localState(),
22});
23
24const { getWhenReady } = Test;
25
26// Out-of-band calls to the Google APIs resolve the same stored credentials
27// the deploy uses, so the test runs against the configured profile.
28const GcpHttp = Layer.mergeAll(
29 GCP.GcpAuth,
30 GCP.fromAuthProvider(),
31 FetchHttpClient.layer,
32);
33
34// Both hosts are built from `main`, which needs a local image build.
35const dockerAvailable =
36 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 }).status ===
37 0;
38
39// Deploy, tests and destroy all sit behind the same Docker guard.
40describe.skipIf(!dockerAvailable)("gcp-pubsub-worker", () => {
41 const stack = beforeAll(deploy(Stack), { timeout: 900_000 });
42
43 /** Pull subscriptions the worker's event source created on the topic. */
44 const subscriptionsOf = (topic: string) =>
45 pubsub.listProjectsTopicsSubscriptions({ topic }).pipe(
46 Effect.map((page) => page.subscriptions ?? []),
47 Effect.provide(GcpHttp),
48 );
49
50 let subscriptions: string[] = [];
51
52 /** Poll until a deleted resource reads as `NotFound`. */
53 const expectGone = <E, R>(
54 what: string,
55 status: Effect.Effect<"found" | "gone", E, R>,
56 ) =>
57 status.pipe(
58 // Deletes can take a few seconds to become visible.
59 Effect.repeat({
60 schedule: Schedule.spaced("5 seconds"),
61 until: (status) => status === "gone",
62 times: 12,
63 }),
64 Effect.tap((status) =>
65 Effect.sync(() => expect(`${what}: ${status}`).toBe(`${what}: gone`)),
66 ),
67 );
68
69 const found = Effect.as("found" as const);
70 const gone = () => Effect.succeed("gone" as const);
71
72 afterAll.skipIf(!!process.env.NO_DESTROY)(
73 Effect.gen(function* () {
74 const outputs = yield* stack;
75 yield* destroy(Stack);
76 if (outputs === undefined) return;
77
78 yield* expectGone(
79 "worker pool",
80 run
81 .getProjectsLocationsWorkerPools({ name: outputs.workerPoolName })
82 .pipe(
83 found,
84 Effect.catchTag("NotFound", gone),
85 Effect.provide(GcpHttp),
86 ),
87 );
88 yield* expectGone(
89 "topic",
90 pubsub
91 .getProjectsTopics({ topic: outputs.topicName })
92 .pipe(
93 found,
94 Effect.catchTag("NotFound", gone),
95 Effect.provide(GcpHttp),
96 ),
97 );
98 for (const subscription of subscriptions) {
99 yield* expectGone(
100 subscription,
101 pubsub
102 .getProjectsSubscriptions({ subscription })
103 .pipe(
104 found,
105 Effect.catchTag("NotFound", gone),
106 Effect.provide(GcpHttp),
107 ),
108 );
109 }
110 yield* expectGone(
111 "database",
112 firestore
113 .getProjectsDatabases({ name: outputs.databaseName })
114 .pipe(
115 found,
116 Effect.catchTag("NotFound", gone),
117 Effect.provide(GcpHttp),
118 ),
119 );
120 }),
121 { timeout: 600_000 },
122 );
123
124 const baseUrlOf = (url: string | undefined) => {
125 if (url === undefined) throw new Error("the service has no URL");
126 return url.replace(/\/+$/, "");
127 };
128
129 interface JobStatus {
130 id: string;
131 status: "pending" | "done";
132 sha256?: string;
133 words?: number;
134 chars?: number;
135 }
136
137 class UnexpectedStatus extends Data.TaggedError("UnexpectedStatus")<{
138 status: number;
139 body: string;
140 }> {}
141
142 const statusOf = (baseUrl: string, id: string) =>
143 Effect.gen(function* () {
144 const res = yield* HttpClient.execute(
145 HttpClientRequest.get(`${baseUrl}/jobs/${id}`),
146 );
147 if (res.status !== 200 && res.status !== 202) {
148 return yield* new UnexpectedStatus({
149 status: res.status,
150 body: yield* res.text,
151 });
152 }
153 return (yield* res.json) as unknown as JobStatus;
154 });
155
156 const PAYLOADS = [
157 "hello world",
158 "the quick brown fox jumps over the lazy dog",
159 " alchemy on google cloud ",
160 "one",
161 "",
162 ];
163
164 test(
165 "rejects a job without a string payload",
166 Effect.gen(function* () {
167 const { url } = yield* stack;
168 const baseUrl = baseUrlOf(url);
169 yield* getWhenReady(`${baseUrl}/`);
170
171 const res = yield* HttpClient.execute(
172 HttpClientRequest.post(`${baseUrl}/jobs`).pipe(
173 HttpClientRequest.bodyJsonUnsafe({ payload: 42 }),
174 ),
175 );
176 expect(res.status).toBe(400);
177 }),
178 { timeout: 120_000 },
179 );
180
181 test(
182 "processes submitted jobs through Pub/Sub into Firestore",
183 Effect.gen(function* () {
184 const { url, topicName, databaseName } = yield* stack;
185 const baseUrl = baseUrlOf(url);
186 yield* getWhenReady(`${baseUrl}/`);
187
188 subscriptions = yield* subscriptionsOf(topicName);
189 expect(subscriptions.length).toBe(1);
190
191 const ids: string[] = [];
192 for (const payload of PAYLOADS) {
193 const res = yield* HttpClient.execute(
194 HttpClientRequest.post(`${baseUrl}/jobs`).pipe(
195 HttpClientRequest.bodyJsonUnsafe({ payload }),
196 ),
197 );
198 expect(res.status).toBe(202);
199 const body = (yield* res.json) as unknown as JobStatus;
200 expect(body.status).toBe("pending");
201 ids.push(body.id);
202 }
203
204 // The worker pool's first pull, and project-level Firestore grants on
205 // a fresh deploy, can each take a few minutes. Until then the API
206 // answers 202 pending — or 500 while its own read grant propagates.
207 const results = yield* Effect.forEach(ids, (id) =>
208 statusOf(baseUrl, id).pipe(
209 Effect.tapError((error) => Effect.logWarning(error)),
210 Effect.retry({ schedule: Schedule.spaced("5 seconds"), times: 24 }),
211 Effect.repeat({
212 schedule: Schedule.spaced("10 seconds"),
213 until: (status) => status.status === "done",
214 times: 48,
215 }),
216 ),
217 );
218
219 for (const [i, payload] of PAYLOADS.entries()) {
220 const expected = {
221 id: ids[i],
222 status: "done",
223 sha256: createHash("sha256").update(payload, "utf8").digest("hex"),
224 words: payload.split(/\s+/).filter(Boolean).length,
225 chars: payload.length,
226 };
227 expect(results[i]).toMatchObject(expected);
228
229 // The result the API served is a real Firestore document.
230 const document = yield* firestore
231 .getProjectsDatabasesDocuments({
232 name: `${databaseName}/documents/jobs/${ids[i]}`,
233 })
234 .pipe(Effect.provide(GcpHttp));
235 expect(document.fields?.status?.stringValue).toBe("done");
236 expect(document.fields?.sha256?.stringValue).toBe(expected.sha256);
237 expect(Number(document.fields?.words?.integerValue)).toBe(
238 expected.words,
239 );
240 }
241 }),
242 { timeout: 900_000 },
243 );
244});