1import * as GCP from "@/GCP";
2import { decodeFields } from "@/GCP/Firestore/Values.ts";
3import * as Test from "@/Test/Alchemy";
4import * as Core from "@/Test/Core";
5import * as scheduler from "@distilled.cloud/gcp/cloudscheduler_v1";
6import * as eventarc from "@distilled.cloud/gcp/eventarc_v1";
7import * as firestore from "@distilled.cloud/gcp/firestore_v1";
8import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
9import * as cloudrun from "@distilled.cloud/gcp/run_v2";
10import * as storage from "@distilled.cloud/gcp/storage_v1";
11import * as iam from "@distilled.cloud/gcp/unstable/iam_v1";
12import { describe, expect } from "alchemy-test";
13import * as Data from "effect/Data";
14import * as Effect from "effect/Effect";
15import * as Schedule from "effect/Schedule";
16import * as HttpClient from "effect/http/HttpClient";
17import * as HttpClientRequest from "effect/http/HttpClientRequest";
18import { spawnSync } from "node:child_process";
19import SmokeApi from "./fixtures/api.ts";
20import SmokeJob from "./fixtures/job.ts";
21import {
22 docId,
23 Jobs,
24 SCHEDULE_BODY,
25 Store,
26 Uploads,
27} from "./fixtures/serverless-resources.ts";
28import SmokeWorker from "./fixtures/worker.ts";
29
30/**
31 * The GCP counterpart of the AWS serverless smoke, in ONE stack.
32 *
33 * Public API Function (Firestore ReadWriteDatabase on a named
34 * FIRESTORE_NATIVE database, Storage ReadWriteBucket + signed URLs,
35 * Pub/Sub WriteTopic, Run.RunJob) → private worker Function consuming the
36 * topic over push, a Cloud Scheduler job, and Eventarc Storage finalize
37 * events → an Effect-native Cloud Run Job. The test drives the story over
38 * HTTP and verifies every side effect out-of-band via distilled.
39 */
40
41const testOptions = { providers: GCP.providers() };
42const { test, beforeAll, afterAll } = Test.make(testOptions);
43const sharedStack = Core.scratchStack(
44 testOptions,
45 "ServerlessSmoke",
46 "test/GCP/Smoke/Serverless.smoke.test.ts",
47);
48
49const dockerAvailable = (() => {
50 try {
51 return (
52 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 })
53 .status === 0
54 );
55 } catch {
56 return false;
57 }
58})();
59
60const skip = !!process.env.FAST || !dockerAvailable;
61
62interface StackOutputs {
63 url: string;
64 project: string;
65 apiName: string;
66 apiServiceAccount: string;
67 workerName: string;
68 workerServiceId: string;
69 workerUrl: string;
70 workerServiceAccount: string;
71 jobName: string;
72 jobServiceAccount: string;
73 databaseName: string;
74 bucketName: string;
75 topicName: string;
76}
77
78let outputs: StackOutputs;
79let baseUrl: string;
80
81class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
82 readonly status: number;
83 readonly body: string;
84}> {}
85
86class NotReady extends Data.TaggedError("NotReady")<{
87 readonly status: number;
88 readonly body: string;
89}> {}
90
91class StillExists extends Data.TaggedError("StillExists")<{
92 readonly what: string;
93}> {}
94
95class DocumentMissing extends Data.TaggedError("DocumentMissing")<{
96 readonly path: string;
97}> {}
98
99class ExecutionPending extends Data.TaggedError("ExecutionPending") {}
100
101// Retry transient 5xx only (cold start, IAM propagation); 4xx and
102// assertion failures surface immediately.
103const send = (request: HttpClientRequest.HttpClientRequest) =>
104 HttpClient.execute(request).pipe(
105 Effect.flatMap((response) =>
106 response.status >= 500
107 ? response.text.pipe(
108 Effect.flatMap((body) =>
109 Effect.fail(
110 new TransientUpstream({ status: response.status, body }),
111 ),
112 ),
113 )
114 : Effect.succeed(response),
115 ),
116 Effect.retry({
117 while: (e) => e._tag === "TransientUpstream",
118 schedule: Schedule.max([
119 Schedule.exponential("1 second"),
120 Schedule.recurs(8),
121 ]),
122 }),
123 );
124
125// Poll a URL until it answers 200 — rides out IAM grant propagation to
126// the API's runtime service account (up to a few minutes on GCP).
127const awaitOk = (url: string) =>
128 HttpClient.get(url).pipe(
129 Effect.flatMap((response) =>
130 response.status === 200
131 ? Effect.succeed(response)
132 : response.text.pipe(
133 Effect.tap((body) =>
134 Effect.logInfo(
135 `Serverless smoke: ${url} not ready (${response.status}: ${body.slice(0, 200)})`,
136 ),
137 ),
138 Effect.flatMap((body) =>
139 Effect.fail(new NotReady({ status: response.status, body })),
140 ),
141 ),
142 ),
143 Effect.retry({
144 while: (e) => e._tag === "NotReady",
145 schedule: Schedule.max([
146 Schedule.fixed("5 seconds"),
147 Schedule.recurs(72),
148 ]),
149 }),
150 );
151
152const documentName = (path: string) =>
153 `${outputs.databaseName}/documents/${path}`;
154
155// Out-of-band Firestore read: `undefined` when the document is missing.
156const readDocument = (path: string) =>
157 firestore.getProjectsDatabasesDocuments({ name: documentName(path) }).pipe(
158 Effect.map((document) => decodeFields(document.fields ?? {})),
159 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
160 );
161
162// Bounded wait until a consumer writes `path`.
163const awaitDocument = (path: string, times = 48) =>
164 readDocument(path).pipe(
165 Effect.flatMap((fields) =>
166 fields === undefined
167 ? Effect.fail(new DocumentMissing({ path }))
168 : Effect.succeed(fields),
169 ),
170 Effect.retry({
171 while: (e) => e._tag === "DocumentMissing",
172 schedule: Schedule.max([
173 Schedule.fixed("5 seconds"),
174 Schedule.recurs(times),
175 ]),
176 }),
177 );
178
179// Bounded wait until an out-of-band probe reports the resource gone.
180const waitUntilGone = <E, R>(
181 what: string,
182 probe: Effect.Effect<boolean, E, R>,
183) =>
184 probe.pipe(
185 Effect.flatMap((gone) =>
186 gone ? Effect.void : Effect.fail(new StillExists({ what })),
187 ),
188 Effect.retry({
189 // `instanceof`, not `_tag`: the probe's error type `E` is generic.
190 while: (e) => e instanceof StillExists,
191 schedule: Schedule.max([
192 Schedule.fixed("3 seconds"),
193 Schedule.recurs(30),
194 ]),
195 }),
196 );
197
198const schedulerJobFor = (workerUrl: string) =>
199 scheduler
200 .listProjectsLocationsJobs({
201 parent: `projects/${outputs.project}/locations/us-central1`,
202 pageSize: 500,
203 })
204 .pipe(
205 Effect.map(({ jobs = [] }) =>
206 jobs.find((job) =>
207 job.httpTarget?.uri?.startsWith(`${workerUrl}/__alchemy/scheduler/`),
208 ),
209 ),
210 );
211
212const eventarcTriggerFor = (workerServiceId: string) =>
213 eventarc
214 .listProjectsLocationsTriggers({
215 parent: `projects/${outputs.project}/locations/us-central1`,
216 })
217 .pipe(
218 Effect.map(({ triggers = [] }) =>
219 triggers.find(
220 (trigger) =>
221 trigger.destination?.cloudRun?.service === workerServiceId,
222 ),
223 ),
224 );
225
226const deployProgram = Effect.gen(function* () {
227 const store = yield* Store;
228 const uploads = yield* Uploads;
229 const jobs = yield* Jobs;
230 const api = yield* SmokeApi;
231 const worker = yield* SmokeWorker;
232 const job = yield* SmokeJob;
233 return {
234 url: api.uri,
235 project: api.project,
236 apiName: api.name,
237 apiServiceAccount: api.serviceAccount,
238 workerName: worker.name,
239 workerServiceId: worker.serviceId,
240 workerUrl: worker.uri,
241 workerServiceAccount: worker.serviceAccount,
242 jobName: job.name,
243 jobServiceAccount: job.serviceAccount,
244 databaseName: store.name,
245 bucketName: uploads.bucketName,
246 topicName: jobs.name,
247 };
248});
249
250describe.sequential(
251 "Serverless smoke",
252 {
253 tags: [
254 "provider:gcp",
255 "provider:gcp:run",
256 "provider:gcp:firestore",
257 "provider:gcp:storage",
258 "provider:gcp:pubsub",
259 "provider:gcp:eventarc",
260 "provider:gcp:cloudscheduler",
261 "live",
262 ],
263 },
264 () => {
265 // No `beforeAll.skipIf`: a skipped run deploys nothing.
266 beforeAll(
267 Effect.gen(function* () {
268 if (skip) return;
269 yield* Effect.logInfo("Serverless smoke: destroying previous stack");
270 yield* sharedStack.destroy();
271
272 yield* Effect.logInfo("Serverless smoke: deploying stack");
273 outputs = (yield* sharedStack.deploy(deployProgram)) as StackOutputs;
274 baseUrl = outputs.url.replace(/\/+$/, "");
275
276 yield* Effect.logInfo(`Serverless smoke: probing ${baseUrl}/config`);
277 const config = (yield* (yield* awaitOk(`${baseUrl}/config`)).json) as {
278 databaseName: string;
279 bucketName: string;
280 topicName: string;
281 jobName: string;
282 };
283
284 // The API observes the same physical resources the stack returned.
285 expect(config).toEqual({
286 databaseName: outputs.databaseName,
287 bucketName: outputs.bucketName,
288 topicName: outputs.topicName,
289 jobName: outputs.jobName,
290 });
291
292 // Firestore/Storage grants are live once /ready answers 200.
293 yield* awaitOk(`${baseUrl}/ready`);
294 }),
295 { timeout: 1_200_000 },
296 );
297
298 afterAll.skipIf(skip)(sharedStack.destroy(), { timeout: 600_000 });
299
300 test.provider.skipIf(skip)(
301 "the API and worker are deployed with their own runtime identities",
302 (_stack) =>
303 Effect.gen(function* () {
304 const database = yield* firestore.getProjectsDatabases({
305 name: outputs.databaseName,
306 });
307 expect(database.type).toEqual("FIRESTORE_NATIVE");
308 expect(database.name?.endsWith("/databases/(default)")).toBe(false);
309
310 const worker = yield* cloudrun.getProjectsLocationsServices({
311 name: outputs.workerName,
312 });
313 expect(worker.template?.serviceAccount).toEqual(
314 outputs.workerServiceAccount,
315 );
316 expect(outputs.workerServiceAccount).not.toEqual(
317 outputs.apiServiceAccount,
318 );
319
320 // The worker is private: an anonymous request is rejected.
321 const anonymous = yield* HttpClient.get(outputs.workerUrl);
322 expect([401, 403]).toContain(anonymous.status);
323 }),
324 {
325 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
326 timeout: 120_000,
327 },
328 );
329
330 test.provider.skipIf(skip)(
331 "todo CRUD through the API lands in Firestore",
332 (_stack) =>
333 Effect.gen(function* () {
334 const text = "ship the gcp serverless smoke";
335 const wrote = yield* send(
336 HttpClientRequest.put(`${baseUrl}/todos/todo-1`).pipe(
337 HttpClientRequest.bodyJsonUnsafe({ text }),
338 ),
339 );
340 expect(wrote.status).toBe(200);
341 expect((yield* wrote.json) as object).toEqual({
342 ok: true,
343 id: "todo-1",
344 });
345
346 // Out-of-band: the document really landed in the named database.
347 expect(yield* readDocument("todos/todo-1")).toEqual({
348 id: "todo-1",
349 text,
350 });
351
352 const read = yield* send(
353 HttpClientRequest.get(`${baseUrl}/todos/todo-1`),
354 );
355 expect(read.status).toBe(200);
356 expect((yield* read.json) as object).toEqual({
357 item: { id: "todo-1", text },
358 });
359
360 const listed = yield* send(HttpClientRequest.get(`${baseUrl}/todos`));
361 expect((yield* listed.json) as object).toEqual({
362 todos: [{ id: "todo-1", text }],
363 });
364
365 const updated = yield* send(
366 HttpClientRequest.put(`${baseUrl}/todos/todo-1`).pipe(
367 HttpClientRequest.bodyJsonUnsafe({ text: "updated" }),
368 ),
369 );
370 expect(updated.status).toBe(200);
371 expect(yield* readDocument("todos/todo-1")).toEqual({
372 id: "todo-1",
373 text: "updated",
374 });
375
376 const deleted = yield* send(
377 HttpClientRequest.delete(`${baseUrl}/todos/todo-1`),
378 );
379 expect(deleted.status).toBe(204);
380 expect(yield* readDocument("todos/todo-1")).toBeUndefined();
381
382 const missing = yield* send(
383 HttpClientRequest.get(`${baseUrl}/todos/todo-1`),
384 );
385 expect(missing.status).toBe(404);
386 }),
387 {
388 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
389 timeout: 180_000,
390 },
391 );
392
393 test.provider.skipIf(skip)(
394 "uploaded files read back directly and through a signed URL",
395 (_stack) =>
396 Effect.gen(function* () {
397 const name = "hello.txt";
398 const body = "uploaded through the gcp serverless smoke";
399
400 const put = yield* send(
401 HttpClientRequest.put(`${baseUrl}/files/${name}`).pipe(
402 HttpClientRequest.bodyText(body, "text/plain"),
403 ),
404 );
405 expect(put.status).toBe(200);
406
407 // Out-of-band: the object is in the bucket.
408 const object = yield* storage.getObjects({
409 bucket: outputs.bucketName,
410 object: name,
411 });
412 expect(object.contentType).toBe("text/plain");
413 expect(object.size).toBe(String(body.length));
414
415 const direct = yield* send(
416 HttpClientRequest.get(`${baseUrl}/files/${name}`),
417 );
418 expect(direct.status).toBe(200);
419 expect(yield* direct.text).toBe(body);
420
421 // signBlob permission on the API's own account may still be
422 // propagating; `send` retries the 5xx.
423 const { url } = (yield* send(
424 HttpClientRequest.get(`${baseUrl}/files/${name}/signed`),
425 ).pipe(Effect.flatMap((r) => r.json))) as { url: string };
426 expect(url).toContain(
427 `https://storage.googleapis.com/${outputs.bucketName}/${name}?`,
428 );
429 expect(url).toContain("X-Goog-Signature=");
430
431 // The signed URL downloads without any Google credentials.
432 const signed = yield* awaitOk(url);
433 expect(yield* signed.text).toBe(body);
434
435 // Tampering with the object path invalidates the signature.
436 const tampered = yield* HttpClient.get(
437 url.replace(`/${name}?`, "/other.txt?"),
438 );
439 expect(tampered.status).toBe(403);
440 }),
441 {
442 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
443 timeout: 300_000,
444 },
445 );
446
447 test.provider.skipIf(skip)(
448 "an upload is delivered to the worker through Eventarc",
449 (_stack) =>
450 Effect.gen(function* () {
451 const trigger = yield* eventarcTriggerFor(outputs.workerServiceId);
452 expect(trigger?.name).toEqual(expect.any(String));
453
454 const name = "eventarc/drop.txt";
455 const put = yield* send(
456 HttpClientRequest.put(
457 `${baseUrl}/files/${encodeURIComponent(name)}`,
458 ).pipe(HttpClientRequest.bodyText("eventarc!", "text/plain")),
459 );
460 expect(put.status).toBe(200);
461
462 const marker = yield* awaitDocument(`uploads/${docId(name)}`);
463 expect(marker).toMatchObject({
464 type: "google.cloud.storage.object.v1.finalized",
465 bucket: outputs.bucketName,
466 name,
467 });
468 }),
469 {
470 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
471 timeout: 360_000,
472 },
473 );
474
475 test.provider.skipIf(skip)(
476 "a published job is processed by the worker over Pub/Sub push",
477 (_stack) =>
478 Effect.gen(function* () {
479 const payload = "process me please";
480 const published = (yield* send(
481 HttpClientRequest.post(`${baseUrl}/jobs`).pipe(
482 HttpClientRequest.bodyJsonUnsafe({ payload }),
483 ),
484 ).pipe(Effect.flatMap((r) => r.json))) as {
485 id: string;
486 messageId: string;
487 };
488 expect(published.messageId).toEqual(expect.any(String));
489
490 const result = yield* awaitDocument(`results/${published.id}`);
491 expect(result).toEqual({
492 id: published.id,
493 payload,
494 processed: payload.toUpperCase(),
495 messageId: published.messageId,
496 });
497
498 // Out-of-band: the push subscription targets the worker.
499 const { subscriptions = [] } =
500 yield* pubsub.listProjectsTopicsSubscriptions({
501 topic: outputs.topicName,
502 });
503 expect(subscriptions.length).toBeGreaterThan(0);
504 }),
505 {
506 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
507 timeout: 360_000,
508 },
509 );
510
511 test.provider.skipIf(skip)(
512 "a forced Cloud Scheduler run reaches the worker",
513 (_stack) =>
514 Effect.gen(function* () {
515 const job = yield* schedulerJobFor(outputs.workerUrl);
516 expect(job?.name).toEqual(expect.any(String));
517 expect(job!.httpTarget?.oidcToken?.serviceAccountEmail).toEqual(
518 outputs.workerServiceAccount,
519 );
520
521 yield* scheduler.runProjectsLocationsJobs({
522 name: job!.name!,
523 body: {},
524 });
525 const jobName = job!.name!.split("/").pop()!;
526 const marker = yield* awaitDocument(`schedules/${docId(jobName)}`);
527 expect(marker.jobName).toEqual(jobName);
528 expect(marker.body).toEqual(SCHEDULE_BODY);
529 expect(String(marker.scheduleTime)).toMatch(/^\d{4}-\d{2}-\d{2}T/);
530 }),
531 {
532 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
533 timeout: 360_000,
534 },
535 );
536
537 test.provider.skipIf(skip)(
538 "the API triggers a Cloud Run Job execution that records itself",
539 (_stack) =>
540 Effect.gen(function* () {
541 const started = (yield* send(
542 HttpClientRequest.post(`${baseUrl}/run-job`),
543 ).pipe(Effect.flatMap((r) => r.json))) as {
544 operation: string;
545 execution: string | undefined;
546 };
547 expect(started.execution).toContain(`${outputs.jobName}/executions/`);
548 const executionId = started.execution!.split("/").pop()!;
549
550 // Cloud Run schedules a fresh execution slowly (~2 minutes).
551 const execution = yield* cloudrun
552 .getProjectsLocationsJobsExecutions({ name: started.execution! })
553 .pipe(
554 Effect.filterOrFail(
555 (item) => (item.completionTime ?? "").length > 0,
556 () => new ExecutionPending(),
557 ),
558 Effect.retry({
559 while: (e) => e._tag === "ExecutionPending",
560 schedule: Schedule.max([
561 Schedule.fixed("10 seconds"),
562 Schedule.recurs(36),
563 ]),
564 }),
565 );
566 expect(execution.succeededCount).toBe(1);
567
568 const marker = yield* awaitDocument(`executions/${executionId}`, 6);
569 expect(marker.execution).toEqual(executionId);
570 expect(marker.ranAt).toBeInstanceOf(Date);
571 }),
572 {
573 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
574 timeout: 480_000,
575 },
576 );
577
578 test.provider.skipIf(skip)(
579 "destroy removes the functions, job, triggers, topic, bucket, and database",
580 (_stack) =>
581 Effect.gen(function* () {
582 const schedulerJob = yield* schedulerJobFor(outputs.workerUrl);
583 const trigger = yield* eventarcTriggerFor(outputs.workerServiceId);
584 expect(schedulerJob?.name).toEqual(expect.any(String));
585 expect(trigger?.name).toEqual(expect.any(String));
586
587 yield* sharedStack.destroy();
588
589 yield* waitUntilGone(
590 "api service",
591 cloudrun
592 .getProjectsLocationsServices({ name: outputs.apiName })
593 .pipe(
594 Effect.as(false),
595 Effect.catchTag("NotFound", () => Effect.succeed(true)),
596 ),
597 );
598 yield* waitUntilGone(
599 "worker service",
600 cloudrun
601 .getProjectsLocationsServices({
602 name: outputs.workerName,
603 })
604 .pipe(
605 Effect.as(false),
606 Effect.catchTag("NotFound", () => Effect.succeed(true)),
607 ),
608 );
609 yield* waitUntilGone(
610 "run job",
611 cloudrun.getProjectsLocationsJobs({ name: outputs.jobName }).pipe(
612 Effect.as(false),
613 Effect.catchTag("NotFound", () => Effect.succeed(true)),
614 ),
615 );
616 yield* waitUntilGone(
617 "scheduler job",
618 scheduler
619 .getProjectsLocationsJobs({ name: schedulerJob!.name! })
620 .pipe(
621 Effect.as(false),
622 Effect.catchTag("NotFound", () => Effect.succeed(true)),
623 ),
624 );
625 yield* waitUntilGone(
626 "eventarc trigger",
627 eventarc
628 .getProjectsLocationsTriggers({ name: trigger!.name! })
629 .pipe(
630 Effect.as(false),
631 Effect.catchTag("NotFound", () => Effect.succeed(true)),
632 ),
633 );
634 yield* waitUntilGone(
635 "topic",
636 pubsub.getProjectsTopics({ topic: outputs.topicName }).pipe(
637 Effect.as(false),
638 Effect.catchTag("NotFound", () => Effect.succeed(true)),
639 ),
640 );
641 yield* waitUntilGone(
642 "bucket",
643 storage.getBuckets({ bucket: outputs.bucketName }).pipe(
644 Effect.as(false),
645 Effect.catchTag("NotFound", () => Effect.succeed(true)),
646 ),
647 );
648 yield* waitUntilGone(
649 "database",
650 firestore.getProjectsDatabases({ name: outputs.databaseName }).pipe(
651 Effect.map((database) => database.deleteTime !== undefined),
652 Effect.catchTag("NotFound", () => Effect.succeed(true)),
653 ),
654 );
655 for (const email of [
656 outputs.apiServiceAccount,
657 outputs.workerServiceAccount,
658 outputs.jobServiceAccount,
659 ]) {
660 yield* waitUntilGone(
661 `service account ${email}`,
662 iam
663 .getProjectsServiceAccounts({
664 name: `projects/${outputs.project}/serviceAccounts/${email}`,
665 })
666 .pipe(
667 Effect.as(false),
668 Effect.catchTag("NotFound", () => Effect.succeed(true)),
669 ),
670 );
671 }
672 }),
673 {
674 tags: ["provider:gcp", "provider:gcp:smoke", "live"],
675 timeout: 600_000,
676 },
677 );
678 },
679);