1import * as GCP from "@/GCP";
2import { makeObjectMedia } from "@/GCP/Storage/ObjectMedia.ts";
3import * as Test from "@/Test/Alchemy";
4import * as scheduler from "@distilled.cloud/gcp/cloudscheduler_v1";
5import { expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import { MinimumLogLevel } from "effect/References";
8import * as Schedule from "effect/Schedule";
9import { spawnSync } from "node:child_process";
10import ScheduleService, {
11 Markers,
12 markerFor,
13 SCHEDULE_BODY,
14} from "./fixtures/schedule-service.ts";
15
16const { test } = Test.make({ providers: GCP.providers() });
17
18const logLevel = Effect.provideService(
19 MinimumLogLevel,
20 process.env.DEBUG ? "Debug" : "Info",
21);
22
23const dockerAvailable = (() => {
24 try {
25 return (
26 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 })
27 .status === 0
28 );
29 } catch {
30 return false;
31 }
32})();
33
34test.provider.skipIf(!dockerAvailable)(
35 "a Cloud Scheduler job invokes an effect-native Function",
36 (stack) =>
37 Effect.gen(function* () {
38 yield* stack.destroy();
39
40 const out = yield* stack.deploy(
41 Effect.gen(function* () {
42 const service = yield* ScheduleService;
43 const bucket = yield* Markers;
44 return {
45 uri: service.uri,
46 project: service.project,
47 bucket: bucket.bucketName,
48 };
49 }),
50 );
51
52 // The job POSTs to the service with an OIDC token.
53 const { jobs = [] } = yield* scheduler.listProjectsLocationsJobs({
54 parent: `projects/${out.project}/locations/us-central1`,
55 pageSize: 500,
56 });
57 const job = jobs.find((candidate) =>
58 candidate.httpTarget?.uri?.startsWith(
59 `${out.uri}/__alchemy/scheduler/`,
60 ),
61 );
62 expect(job).toBeDefined();
63 expect(job!.httpTarget?.httpMethod).toEqual("POST");
64 expect(job!.httpTarget?.oidcToken?.serviceAccountEmail).toBeDefined();
65 expect(job!.httpTarget?.oidcToken?.audience).toEqual(
66 `${out.uri}/__alchemy/scheduler/heartbeat`,
67 );
68
69 // Force one run instead of waiting for the (yearly) cron. The job's
70 // default retry policy redelivers until the host accepts it.
71 const jobName = job!.name!.split("/").pop()!;
72 const media = yield* makeObjectMedia;
73 yield* scheduler.runProjectsLocationsJobs({ name: job!.name!, body: {} });
74 const marker = yield* media
75 .download({ bucket: out.bucket, object: markerFor(jobName) })
76 .pipe(
77 Effect.retry({
78 while: (error) => error._tag === "GCP.Storage.ObjectNotFound",
79 schedule: Schedule.spaced("5 seconds"),
80 times: 48,
81 }),
82 );
83 const event = JSON.parse(new TextDecoder().decode(marker.body));
84 expect(event.jobName).toEqual(jobName);
85 expect(event.body).toEqual(SCHEDULE_BODY);
86 expect(event.scheduleTime).toMatch(/^\d{4}-\d{2}-\d{2}T/);
87
88 yield* stack.destroy();
89
90 const gone = yield* scheduler
91 .getProjectsLocationsJobs({ name: job!.name! })
92 .pipe(
93 Effect.as("found" as const),
94 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
95 );
96 expect(gone).toEqual("gone");
97 }).pipe(logLevel),
98 {
99 tags: ["provider:gcp", "provider:gcp:cloudscheduler", "live"],
100 timeout: 600_000,
101 },
102);