1import * as Alchemy from "alchemy";
2import * as GCP from "alchemy/GCP";
3import * as Test from "alchemy/Test/Bun";
4import { describe, expect } from "bun:test";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import * as HttpClient from "effect/http/HttpClient";
8import * as HttpClientRequest from "effect/http/HttpClientRequest";
9import { spawnSync } from "node:child_process";
10import Stack from "../alchemy.run.ts";
11
12const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
13 providers: GCP.providers(),
14 state: Alchemy.localState(),
15});
16
17const { getWhenReady } = Test;
18
19// Both hosts are built from `main`, which needs a local image build.
20const dockerAvailable =
21 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 }).status ===
22 0;
23
24// Deploy, tests and destroy all sit behind the same Docker guard.
25describe.skipIf(!dockerAvailable)("gcp-event-pipeline", () => {
26 const stack = beforeAll(deploy(Stack), { timeout: 900_000 });
27
28 afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), {
29 timeout: 600_000,
30 });
31
32 const baseUrlOf = (url: string | undefined) => {
33 if (url === undefined) throw new Error("the service has no URL");
34 return url.replace(/\/+$/, "");
35 };
36
37 const publish = (baseUrl: string, type: string, payload: unknown) =>
38 HttpClient.execute(
39 HttpClientRequest.post(`${baseUrl}/events`).pipe(
40 HttpClientRequest.bodyJsonUnsafe({ type, payload }),
41 ),
42 );
43
44 const countOf = (baseUrl: string, type: string) =>
45 Effect.gen(function* () {
46 const res = yield* HttpClient.execute(
47 HttpClientRequest.get(
48 `${baseUrl}/events/count?type=${encodeURIComponent(type)}`,
49 ),
50 );
51 return ((yield* res.json) as { count: number }).count;
52 });
53
54 test(
55 "accepts events without touching BigQuery on the request path",
56 Effect.gen(function* () {
57 const { url } = yield* stack;
58 expect(url).toMatch(/^https:\/\//);
59 const baseUrl = baseUrlOf(url);
60 yield* getWhenReady(`${baseUrl}/`);
61
62 const accepted = yield* publish(baseUrl, "integ.smoke", { n: 1 });
63 expect(accepted.status).toBe(202);
64 const { id } = (yield* accepted.json) as { id: string };
65 expect(id).toMatch(
66 /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/,
67 );
68
69 const missingType = yield* HttpClient.execute(
70 HttpClientRequest.post(`${baseUrl}/events`).pipe(
71 HttpClientRequest.bodyJsonUnsafe({ payload: {} }),
72 ),
73 );
74 expect(missingType.status).toBe(400);
75 }),
76 { timeout: 180_000 },
77 );
78
79 test(
80 "the drain job moves published events into BigQuery",
81 Effect.gen(function* () {
82 const { url } = yield* stack;
83 const baseUrl = baseUrlOf(url);
84 yield* getWhenReady(`${baseUrl}/`);
85
86 // A unique type keeps the count independent of other runs against the
87 // same project.
88 const type = `integ.drain.${crypto.randomUUID().slice(0, 8)}`;
89 for (let i = 0; i < 3; i++) {
90 const res = yield* publish(baseUrl, type, { i });
91 expect(res.status).toBe(202);
92 }
93
94 expect(yield* countOf(baseUrl, type)).toBe(0);
95
96 const drain = yield* HttpClient.execute(
97 HttpClientRequest.post(`${baseUrl}/drain`),
98 );
99 expect(drain.status).toBe(202);
100
101 // Starting a Cloud Run Job returns immediately; the rows show up once
102 // the execution finishes and BigQuery's streaming buffer is queryable.
103 const landed = yield* countOf(baseUrl, type).pipe(
104 Effect.repeat({
105 schedule: Schedule.spaced("10 seconds"),
106 until: (count) => count >= 3,
107 times: 24,
108 }),
109 );
110 expect(landed).toBeGreaterThanOrEqual(3);
111 }),
112 { timeout: 600_000 },
113 );
114});