1import * as Alchemy from "alchemy";
2import * as GCP from "alchemy/GCP";
3import * as Test from "alchemy/Test/Bun";
4import * as eventarc from "@distilled.cloud/gcp/eventarc_v1";
5import * as firestore from "@distilled.cloud/gcp/firestore_v1";
6import { describe, expect } from "bun:test";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Schedule from "effect/Schedule";
10import * as FetchHttpClient from "effect/http/FetchHttpClient";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import { spawnSync } from "node:child_process";
14import Stack from "../alchemy.run.ts";
15import { LOCATION } from "../src/resources.ts";
16
17const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
18 providers: GCP.providers(),
19 state: Alchemy.localState(),
20});
21
22const { getWhenReady } = Test;
23
24// Out-of-band calls to the Google APIs resolve the same stored credentials
25// the deploy uses, so the test runs against the configured profile.
26const GcpHttp = Layer.mergeAll(
27 GCP.GcpAuth,
28 GCP.fromAuthProvider(),
29 FetchHttpClient.layer,
30);
31
32// Both services are built from `main`, which needs a local image build.
33const dockerAvailable =
34 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 }).status ===
35 0;
36
37const TRIGGER_PATH = "/__alchemy/eventarc/ordercreated";
38
39// Kept outside the handle so teardown can verify even after a failed test.
40let deployed:
41 | { project: string; auditorService: string; databaseName: string }
42 | undefined;
43
44// Deploy, tests and destroy all sit behind the same Docker guard.
45describe.skipIf(!dockerAvailable)("gcp-eventarc-firestore", () => {
46 const stack = beforeAll(
47 deploy(Stack).pipe(
48 Effect.tap((outputs) =>
49 Effect.sync(() => {
50 deployed = outputs;
51 }),
52 ),
53 ),
54 { timeout: 900_000 },
55 );
56
57 /** The example's trigger, found by its delivery path and target service. */
58 const findTrigger = (project: string, auditor: string) =>
59 eventarc
60 .listProjectsLocationsTriggers({
61 parent: `projects/${project}/locations/${LOCATION}`,
62 })
63 .pipe(
64 Effect.map((page) =>
65 (page.triggers ?? []).find(
66 (trigger) =>
67 trigger.destination?.cloudRun?.path === TRIGGER_PATH &&
68 trigger.destination.cloudRun.service === auditor,
69 ),
70 ),
71 Effect.orDie,
72 Effect.provide(GcpHttp),
73 );
74
75 // Destroy, then prove the trigger and the database are actually gone.
76 afterAll.skipIf(!!process.env.NO_DESTROY)(
77 Effect.gen(function* () {
78 yield* destroy(Stack);
79 if (deployed === undefined) return;
80 const { project, auditorService, databaseName } = deployed;
81
82 expect(yield* findTrigger(project, auditorService)).toBeUndefined();
83
84 const database = yield* firestore
85 .getProjectsDatabases({ name: databaseName })
86 .pipe(
87 Effect.map((db) => db.name),
88 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
89 Effect.orDie,
90 Effect.provide(GcpHttp),
91 );
92 expect(database).toBeUndefined();
93 }),
94 { timeout: 600_000 },
95 );
96
97 const baseUrlOf = (url: string | undefined) => {
98 if (url === undefined) throw new Error("the service has no URL");
99 return url.replace(/\/+$/, "");
100 };
101
102 const createOrder = (baseUrl: string, item: string, quantity: number) =>
103 HttpClient.execute(
104 HttpClientRequest.post(`${baseUrl}/orders`).pipe(
105 HttpClientRequest.bodyJsonUnsafe({ item, quantity }),
106 ),
107 );
108
109 /**
110 * A fresh deploy's project-level Firestore grant can take several minutes
111 * to propagate, and not to every IAM backend at once; until it has, the
112 * API answers 500 (Forbidden from Firestore).
113 */
114 const createOrderWhenGranted = (
115 baseUrl: string,
116 item: string,
117 quantity: number,
118 ) =>
119 createOrder(baseUrl, item, quantity).pipe(
120 Effect.repeat({
121 schedule: Schedule.spaced("10 seconds"),
122 until: (response) => response.status !== 500,
123 times: 42,
124 }),
125 );
126
127 /** The auditor's `audit/{id}` entry, polled every 10s up to `times`. */
128 const auditOf = (databaseName: string, id: string, times: number) =>
129 firestore
130 .getProjectsDatabasesDocuments({
131 name: `${databaseName}/documents/audit/${id}`,
132 })
133 .pipe(
134 Effect.retry({
135 while: (error) => error._tag === "NotFound",
136 schedule: Schedule.spaced("10 seconds"),
137 times,
138 }),
139 Effect.provide(GcpHttp),
140 );
141
142 test(
143 "the trigger watches created orders in the named database",
144 Effect.gen(function* () {
145 const { project, auditorService, databaseId } = yield* stack;
146 const trigger = yield* findTrigger(project, auditorService);
147 expect(trigger?.name).toEqual(expect.any(String));
148
149 const filters = Object.fromEntries(
150 (trigger?.eventFilters ?? []).map((filter) => [
151 filter.attribute,
152 { value: filter.value, operator: filter.operator },
153 ]),
154 );
155 expect(filters.type?.value).toEqual(
156 "google.cloud.firestore.document.v1.created",
157 );
158 expect(filters.database?.value).toEqual(databaseId);
159 expect(filters.document).toEqual({
160 value: "orders/{id}",
161 operator: "match-path-pattern",
162 });
163 }),
164 { timeout: 120_000 },
165 );
166
167 test(
168 "rejects an order without an item",
169 Effect.gen(function* () {
170 const { url } = yield* stack;
171 const baseUrl = baseUrlOf(url);
172 yield* getWhenReady(`${baseUrl}/`);
173
174 const res = yield* HttpClient.execute(
175 HttpClientRequest.post(`${baseUrl}/orders`).pipe(
176 HttpClientRequest.bodyJsonUnsafe({ quantity: 1 }),
177 ),
178 );
179 expect(res.status).toBe(400);
180 }),
181 { timeout: 120_000 },
182 );
183
184 test(
185 "every created order gets an audit entry from the change event",
186 Effect.gen(function* () {
187 const { url, databaseName } = yield* stack;
188 const baseUrl = baseUrlOf(url);
189 yield* getWhenReady(`${baseUrl}/`);
190
191 // Eventarc can take a couple of minutes after the trigger reports
192 // healthy before it starts routing events, and a Firestore event
193 // emitted before then is never delivered. Probe with throwaway orders
194 // until one is audited; after that, every order must be.
195 yield* Effect.gen(function* () {
196 const probe = yield* createOrderWhenGranted(baseUrl, "probe", 1);
197 const { id } = (yield* probe.json) as { id: string };
198 return yield* auditOf(databaseName, id, 6);
199 }).pipe(Effect.retry({ times: 6 }), Effect.orDie);
200
201 const first = yield* createOrderWhenGranted(baseUrl, "widget", 2);
202 expect(first.status).toBe(201);
203 const second = yield* createOrderWhenGranted(baseUrl, "gadget", 1);
204 expect(second.status).toBe(201);
205
206 const ids = [
207 ((yield* first.json) as { id: string }).id,
208 ((yield* second.json) as { id: string }).id,
209 ];
210
211 for (const id of ids) {
212 // The order is a real Firestore document.
213 const order = yield* firestore
214 .getProjectsDatabasesDocuments({
215 name: `${databaseName}/documents/orders/${id}`,
216 })
217 .pipe(Effect.orDie, Effect.provide(GcpHttp));
218 expect(order.fields?.item?.stringValue).toBeDefined();
219
220 // A failed delivery (e.g. the auditor's grant still propagating) is
221 // redelivered with backoff, so give each audit a few minutes.
222 const audit = yield* auditOf(databaseName, id, 30).pipe(Effect.orDie);
223 expect(audit.fields?.order?.stringValue).toEqual(`orders/${id}`);
224 expect(audit.fields?.eventType?.stringValue).toEqual(
225 "google.cloud.firestore.document.v1.created",
226 );
227 // The protobuf payload reached the handler as bytes.
228 expect(
229 Number(audit.fields?.payloadBytes?.integerValue ?? 0),
230 ).toBeGreaterThan(0);
231 }
232 }),
233 { timeout: 900_000 },
234 );
235});