1import * as Alchemy from "alchemy";
2import * as GCP from "alchemy/GCP";
3import * as Test from "alchemy/Test/Bun";
4import * as bigquery from "@distilled.cloud/gcp/bigquery_v2";
5import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
6import * as run from "@distilled.cloud/gcp/run_v2";
7import * as storage from "@distilled.cloud/gcp/storage_v1";
8import { describe, expect } from "bun:test";
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 Stack from "../alchemy.run.ts";
17import { MAX_DELIVERY_ATTEMPTS } from "../src/Email.ts";
18import { emailObjectFor, type OrderEvent } from "../src/resources.ts";
19
20const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
21 providers: GCP.providers(),
22 state: Alchemy.localState(),
23});
24
25const { getWhenReady } = Test;
26
27// Out-of-band calls to the Google APIs resolve the same stored credentials
28// the deploy uses, so the test runs against the configured profile.
29const GcpHttp = Layer.mergeAll(
30 GCP.GcpAuth,
31 GCP.fromAuthProvider(),
32 FetchHttpClient.layer,
33);
34
35// The services are built from `main`, which needs a local image build.
36const dockerAvailable =
37 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 }).status ===
38 0;
39
40// Deploy, tests and destroy all sit behind the same Docker guard.
41describe.skipIf(!dockerAvailable)("gcp-pubsub-fanout", () => {
42 const stack = beforeAll(deploy(Stack), { timeout: 900_000 });
43
44 const baseUrlOf = (url: string | undefined) => {
45 if (url === undefined) throw new Error("the service has no URL");
46 return url.replace(/\/+$/, "");
47 };
48
49 /** `projects/{p}/datasets/{d}/tables/{t}` → its parts. */
50 const tableRefOf = (tableName: string) => {
51 const [, projectId, , datasetId, , tableId] = tableName.split("/");
52 return { projectId: projectId!, datasetId: datasetId!, tableId: tableId! };
53 };
54
55 /** The push subscriptions the two consumers' event sources created. */
56 // Pub/Sub reads are eventually consistent: a just-created subscription can
57 // briefly read as missing, so re-list until every listed one resolves.
58 const subscriptionsOf = (topic: string) =>
59 pubsub.listProjectsTopicsSubscriptions({ topic }).pipe(
60 Effect.map((page) => page.subscriptions ?? []),
61 Effect.flatMap((names) =>
62 Effect.forEach(names, (subscription) =>
63 pubsub.getProjectsSubscriptions({ subscription }),
64 ),
65 ),
66 Effect.retry({
67 while: (error) => error._tag === "NotFound",
68 schedule: Schedule.spaced("2 seconds"),
69 times: 15,
70 }),
71 Effect.provide(GcpHttp),
72 );
73
74 const placeOrder = (baseUrl: string, email: string, total: number) =>
75 Effect.gen(function* () {
76 const res = yield* HttpClient.execute(
77 HttpClientRequest.post(`${baseUrl}/orders`).pipe(
78 HttpClientRequest.bodyJsonUnsafe({ email, total }),
79 ),
80 );
81 expect(res.status).toBe(202);
82 return (yield* res.json) as { orderId: string; eventId: string };
83 });
84
85 const cancelOrder = (baseUrl: string, orderId: string, email: string) =>
86 Effect.gen(function* () {
87 const res = yield* HttpClient.execute(
88 HttpClientRequest.post(`${baseUrl}/orders/${orderId}/cancel`).pipe(
89 HttpClientRequest.bodyJsonUnsafe({ email }),
90 ),
91 );
92 expect(res.status).toBe(202);
93 return (yield* res.json) as { orderId: string; eventId: string };
94 });
95
96 /** The outbox object for an event, or `undefined` while it is missing. */
97 const emailFor = (bucket: string, eventId: string) =>
98 storage.getObjects({ bucket, object: emailObjectFor(eventId) }).pipe(
99 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
100 Effect.provide(GcpHttp),
101 );
102
103 /** Poll until the email consumer has written the event's confirmation. */
104 const awaitEmail = (bucket: string, eventId: string) =>
105 emailFor(bucket, eventId).pipe(
106 Effect.repeat({
107 schedule: Schedule.spaced("5 seconds"),
108 until: (object) => object !== undefined,
109 times: 60,
110 }),
111 );
112
113 interface AnalyticsRow {
114 eventId: string;
115 type: string;
116 orderId: string;
117 }
118
119 /** Rows the analytics consumer recorded for these events. */
120 const analyticsRowsFor = (tableName: string, eventIds: string[]) =>
121 Effect.gen(function* () {
122 const { projectId, datasetId, tableId } = tableRefOf(tableName);
123 const result = yield* bigquery.queryJobs({
124 projectId,
125 body: {
126 query: `SELECT eventId, type, orderId FROM \`${datasetId}.${tableId}\` WHERE eventId IN UNNEST(@ids)`,
127 useLegacySql: false,
128 parameterMode: "NAMED",
129 queryParameters: [
130 {
131 name: "ids",
132 parameterType: { type: "ARRAY", arrayType: { type: "STRING" } },
133 parameterValue: {
134 arrayValues: eventIds.map((value) => ({ value })),
135 },
136 },
137 ],
138 timeoutMs: 20_000,
139 },
140 });
141 return (result.rows ?? []).map((row): AnalyticsRow => ({
142 eventId: String(row.f?.[0]?.v),
143 type: String(row.f?.[1]?.v),
144 orderId: String(row.f?.[2]?.v),
145 }));
146 }).pipe(Effect.provide(GcpHttp));
147
148 /** Poll until every event has a row (streaming inserts are queryable at once). */
149 const awaitAnalytics = (tableName: string, eventIds: string[]) =>
150 analyticsRowsFor(tableName, eventIds).pipe(
151 Effect.repeat({
152 schedule: Schedule.spaced("5 seconds"),
153 until: (rows) =>
154 eventIds.every((id) => rows.some((row) => row.eventId === id)),
155 times: 60,
156 }),
157 );
158
159 test(
160 "each consumer gets its own push subscription on the one topic",
161 Effect.gen(function* () {
162 const {
163 topicName,
164 deadLetterTopicName,
165 deadLetterSubscription,
166 project,
167 } = yield* stack;
168
169 const subscriptions = yield* subscriptionsOf(topicName);
170 expect(subscriptions.length).toBe(2);
171 for (const subscription of subscriptions) {
172 expect(subscription.pushConfig?.pushEndpoint).toMatch(
173 /^https:\/\/.+\/__alchemy\/pubsub\/orderevents$/,
174 );
175 expect(subscription.pushConfig?.oidcToken?.serviceAccountEmail).toMatch(
176 /@.+\.iam\.gserviceaccount\.com$/,
177 );
178 }
179
180 const filtered = subscriptions.filter((s) => s.filter);
181 expect(filtered.length).toBe(1);
182 const email = filtered[0]!;
183 expect(email.filter).toBe('attributes.type = "order.created"');
184 expect(email.deadLetterPolicy).toEqual({
185 deadLetterTopic: deadLetterTopicName,
186 maxDeliveryAttempts: MAX_DELIVERY_ATTEMPTS,
187 });
188 // Backoff keeps the attempts from being spent on a transient failure.
189 expect(email.retryPolicy).toEqual({
190 minimumBackoff: "10s",
191 maximumBackoff: "600s",
192 });
193
194 const analytics = subscriptions.find((s) => !s.filter)!;
195 expect(analytics.deadLetterPolicy).toBeUndefined();
196
197 // Dead letters are forwarded by the project's Pub/Sub service agent,
198 // which needs to publish to the dead-letter topic and ack the source.
199 const topicPolicy = yield* pubsub
200 .getIamPolicyProjectsTopics({ resource: deadLetterTopicName })
201 .pipe(Effect.provide(GcpHttp));
202 const agent = (topicPolicy.bindings ?? [])
203 .find((binding) => binding.role === "roles/pubsub.publisher")
204 ?.members?.find((member) => member.includes("@gcp-sa-pubsub."));
205 expect(agent).toBeDefined();
206 const subscriptionPolicy = yield* pubsub
207 .getIamPolicyProjectsSubscriptions({ resource: email.name! })
208 .pipe(Effect.provide(GcpHttp));
209 expect(
210 (subscriptionPolicy.bindings ?? []).some(
211 (binding) =>
212 binding.role === "roles/pubsub.subscriber" &&
213 binding.members?.includes(agent!),
214 ),
215 ).toBe(true);
216
217 expect(deadLetterSubscription).toStartWith(`projects/${project}/`);
218 }),
219 { timeout: 120_000 },
220 );
221
222 test(
223 "rejects an order without an email or total",
224 Effect.gen(function* () {
225 const { url } = yield* stack;
226 const baseUrl = baseUrlOf(url);
227 yield* getWhenReady(`${baseUrl}/`);
228
229 for (const body of [{ total: 10 }, { email: "a@example.com" }]) {
230 const res = yield* HttpClient.execute(
231 HttpClientRequest.post(`${baseUrl}/orders`).pipe(
232 HttpClientRequest.bodyJsonUnsafe(body),
233 ),
234 );
235 expect(res.status).toBe(400);
236 }
237 }),
238 { timeout: 120_000 },
239 );
240
241 test(
242 "every order.created reaches both the email and the analytics consumer",
243 Effect.gen(function* () {
244 const { url, bucketName, tableName } = yield* stack;
245 const baseUrl = baseUrlOf(url);
246 yield* getWhenReady(`${baseUrl}/`);
247
248 const orders = [];
249 for (const [i, total] of [12.5, 99, 7.25].entries()) {
250 orders.push(yield* placeOrder(baseUrl, `buyer${i}@example.com`, total));
251 }
252
253 // A fresh deploy's first pushes can wait on IAM propagation for a few
254 // minutes; Pub/Sub retries them until they land.
255 for (const [i, order] of orders.entries()) {
256 const email = yield* awaitEmail(bucketName, order.eventId);
257 expect(email).toBeDefined();
258 expect(email!.metadata).toEqual({
259 orderId: order.orderId,
260 to: `buyer${i}@example.com`,
261 });
262 }
263
264 const rows = yield* awaitAnalytics(
265 tableName,
266 orders.map((order) => order.eventId),
267 );
268 for (const order of orders) {
269 expect(rows).toContainEqual({
270 eventId: order.eventId,
271 type: "order.created" satisfies OrderEvent["type"],
272 orderId: order.orderId,
273 });
274 }
275 }),
276 { timeout: 900_000 },
277 );
278
279 test(
280 "order.cancelled only reaches analytics",
281 Effect.gen(function* () {
282 const { url, bucketName, tableName } = yield* stack;
283 const baseUrl = baseUrlOf(url);
284 yield* getWhenReady(`${baseUrl}/`);
285
286 const created = yield* placeOrder(baseUrl, "changed@example.com", 40);
287 const cancelled = yield* cancelOrder(
288 baseUrl,
289 created.orderId,
290 "changed@example.com",
291 );
292
293 const rows = yield* awaitAnalytics(tableName, [
294 created.eventId,
295 cancelled.eventId,
296 ]);
297 expect(rows).toContainEqual({
298 eventId: cancelled.eventId,
299 type: "order.cancelled",
300 orderId: created.orderId,
301 });
302
303 // The confirmation for the created event lands; the cancellation is
304 // filtered out by Pub/Sub before it is ever pushed to the email service.
305 expect(yield* awaitEmail(bucketName, created.eventId)).toBeDefined();
306 expect(yield* emailFor(bucketName, cancelled.eventId)).toBeUndefined();
307 }),
308 { timeout: 900_000 },
309 );
310
311 test(
312 "an email that can never be sent is dead-lettered after max attempts",
313 Effect.gen(function* () {
314 const { url, bucketName, tableName, deadLetterSubscription } =
315 yield* stack;
316 const baseUrl = baseUrlOf(url);
317 yield* getWhenReady(`${baseUrl}/`);
318
319 const order = yield* placeOrder(baseUrl, "nobody@example.invalid", 5);
320
321 // Analytics is a separate subscription, so the email failures don't
322 // hold it back.
323 yield* awaitAnalytics(tableName, [order.eventId]);
324
325 const pullDeadLetters = pubsub
326 .pullProjectsSubscriptions({
327 subscription: deadLetterSubscription,
328 body: { maxMessages: 10, returnImmediately: true },
329 })
330 .pipe(Effect.provide(GcpHttp));
331
332 const dead: pubsub.ReceivedMessage[] = [];
333 yield* pullDeadLetters.pipe(
334 Effect.tap(({ receivedMessages = [] }) =>
335 Effect.gen(function* () {
336 dead.push(...receivedMessages);
337 const ackIds = receivedMessages.flatMap((m) =>
338 m.ackId ? [m.ackId] : [],
339 );
340 if (ackIds.length > 0) {
341 yield* pubsub
342 .acknowledgeProjectsSubscriptions({
343 subscription: deadLetterSubscription,
344 body: { ackIds },
345 })
346 .pipe(Effect.provide(GcpHttp));
347 }
348 }),
349 ),
350 Effect.repeat({
351 schedule: Schedule.spaced("10 seconds"),
352 until: () => dead.some((m) => decode(m).eventId === order.eventId),
353 times: 48,
354 }),
355 );
356
357 const letter = dead.find((m) => decode(m).eventId === order.eventId)!;
358 expect(letter.message?.attributes?.type).toBe("order.created");
359 // Pub/Sub stamps where the dead letter came from.
360 expect(
361 letter.message?.attributes?.CloudPubSubDeadLetterSourceSubscription,
362 ).toBeDefined();
363
364 expect(yield* emailFor(bucketName, order.eventId)).toBeUndefined();
365 }),
366 { timeout: 900_000 },
367 );
368
369 const decode = (received: pubsub.ReceivedMessage) =>
370 JSON.parse(
371 Buffer.from(received.message?.data ?? "e30=", "base64").toString("utf8"),
372 ) as OrderEvent;
373
374 /** Poll until a deleted resource reads as `NotFound`. */
375 const expectGone = <E, R>(
376 what: string,
377 status: Effect.Effect<"found" | "gone", E, R>,
378 ) =>
379 status.pipe(
380 Effect.repeat({
381 schedule: Schedule.spaced("5 seconds"),
382 until: (status) => status === "gone",
383 times: 12,
384 }),
385 Effect.tap((status) =>
386 Effect.sync(() => expect(`${what}: ${status}`).toBe(`${what}: gone`)),
387 ),
388 );
389
390 const statusOf = <A, E extends { _tag: string }, R>(
391 effect: Effect.Effect<A, E, R>,
392 ) =>
393 effect.pipe(
394 Effect.as("found" as const),
395 Effect.catchIf(
396 (error) => error._tag === "NotFound",
397 () => Effect.succeed("gone" as const),
398 ),
399 Effect.provide(GcpHttp),
400 );
401
402 afterAll.skipIf(!!process.env.NO_DESTROY)(
403 Effect.gen(function* () {
404 const outputs = yield* stack;
405 const subscriptions =
406 outputs === undefined
407 ? []
408 : (yield* subscriptionsOf(outputs.topicName)).map((s) => s.name!);
409
410 yield* destroy(Stack);
411 if (outputs === undefined) return;
412
413 for (const [what, name] of [
414 ["email service", outputs.emailService],
415 ["analytics service", outputs.analyticsService],
416 ] as const) {
417 yield* expectGone(
418 what,
419 statusOf(run.getProjectsLocationsServices({ name })),
420 );
421 }
422 for (const subscription of [
423 ...subscriptions,
424 outputs.deadLetterSubscription,
425 ]) {
426 yield* expectGone(
427 subscription,
428 statusOf(pubsub.getProjectsSubscriptions({ subscription })),
429 );
430 }
431 for (const topic of [outputs.topicName, outputs.deadLetterTopicName]) {
432 yield* expectGone(topic, statusOf(pubsub.getProjectsTopics({ topic })));
433 }
434 yield* expectGone(
435 "bucket",
436 statusOf(storage.getBuckets({ bucket: outputs.bucketName })),
437 );
438 const { projectId, datasetId } = tableRefOf(outputs.tableName);
439 yield* expectGone(
440 "dataset",
441 statusOf(bigquery.getDatasets({ projectId, datasetId })),
442 );
443 }),
444 { timeout: 600_000 },
445 );
446});