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 scheduler from "@distilled.cloud/gcp/cloudscheduler_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";
15
16const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
17 providers: GCP.providers(),
18 state: Alchemy.localState(),
19});
20
21const { getWhenReady } = Test;
22
23// Out-of-band calls to the Google APIs resolve the same stored credentials
24// the deploy uses, so the test runs against the configured profile.
25const GcpHttp = Layer.mergeAll(
26 GCP.GcpAuth,
27 GCP.fromAuthProvider(),
28 FetchHttpClient.layer,
29);
30
31// The service is built from `main`, which needs a local image build.
32const dockerAvailable =
33 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 }).status ===
34 0;
35
36interface Heartbeat {
37 kind: "heartbeat" | "daily";
38 jobName: string;
39 scheduleTime: string;
40 receivedAt: string;
41 value: number;
42}
43
44const baseUrlOf = (url: string | undefined) => {
45 if (url === undefined) throw new Error("the service has no URL");
46 return url.replace(/\/+$/, "");
47};
48
49/** The Cloud Scheduler jobs that deliver to this service. */
50const jobsFor = (project: string, location: string, baseUrl: string) =>
51 scheduler
52 .listProjectsLocationsJobs({
53 parent: `projects/${project}/locations/${location}`,
54 pageSize: 500,
55 })
56 .pipe(
57 Effect.map(({ jobs = [] }) =>
58 jobs.filter((job) =>
59 job.httpTarget?.uri?.startsWith(`${baseUrl}/__alchemy/scheduler/`),
60 ),
61 ),
62 Effect.orDie,
63 Effect.provide(GcpHttp),
64 );
65
66// Deploy, tests and destroy all sit behind the same Docker guard.
67describe.skipIf(!dockerAvailable)("gcp-cron", () => {
68 const stack = beforeAll(deploy(Stack), { timeout: 900_000 });
69
70 afterAll.skipIf(!!process.env.NO_DESTROY)(
71 Effect.gen(function* () {
72 const { url, project, location, tableName } = yield* stack;
73 yield* destroy(Stack);
74
75 expect(yield* jobsFor(project, location, baseUrlOf(url))).toEqual([]);
76
77 // `projects/{project}/datasets/{dataset}/tables/{table}`
78 const [, projectId, , datasetId, , tableId] = tableName.split("/");
79 const table = yield* bigquery
80 .getTables({
81 projectId: projectId!,
82 datasetId: datasetId!,
83 tableId: tableId!,
84 })
85 .pipe(
86 Effect.as("found" as const),
87 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
88 Effect.orDie,
89 Effect.provide(GcpHttp),
90 );
91 expect(table).toEqual("gone");
92 }),
93 { timeout: 600_000 },
94 );
95
96 const heartbeatsOf = (baseUrl: string, kind: Heartbeat["kind"]) =>
97 Effect.gen(function* () {
98 const res = yield* HttpClient.execute(
99 HttpClientRequest.get(`${baseUrl}/heartbeats?kind=${kind}&limit=50`),
100 );
101 expect(res.status).toBe(200);
102 return ((yield* res.json) as unknown as { heartbeats: Heartbeat[] })
103 .heartbeats;
104 });
105
106 test(
107 "creates one Cloud Scheduler job per schedule, both OIDC-authenticated",
108 Effect.gen(function* () {
109 const { url, project, location } = yield* stack;
110 const baseUrl = baseUrlOf(url);
111
112 const jobs = yield* jobsFor(project, location, baseUrl);
113 const byPath = Object.fromEntries(
114 jobs.map((job) => [new URL(job.httpTarget!.uri!).pathname, job]),
115 );
116 expect(Object.keys(byPath).sort()).toEqual([
117 "/__alchemy/scheduler/dailyrollup",
118 "/__alchemy/scheduler/heartbeat",
119 ]);
120 expect(byPath["/__alchemy/scheduler/heartbeat"]!.schedule).toEqual(
121 "* * * * *",
122 );
123 expect(byPath["/__alchemy/scheduler/dailyrollup"]!.schedule).toEqual(
124 "0 0 * * *",
125 );
126 for (const job of jobs) {
127 expect(job.httpTarget?.httpMethod).toEqual("POST");
128 expect(job.httpTarget?.oidcToken?.audience).toEqual(
129 job.httpTarget?.uri,
130 );
131 }
132
133 // The service is public, but the schedule routes are not: a request
134 // without Cloud Scheduler's token is refused.
135 yield* getWhenReady(`${baseUrl}/`);
136 const forged = yield* HttpClient.execute(
137 HttpClientRequest.post(`${baseUrl}/__alchemy/scheduler/heartbeat`),
138 );
139 expect(forged.status).toBe(401);
140 }),
141 { timeout: 180_000 },
142 );
143
144 test(
145 "a forced run of each schedule records a row in BigQuery",
146 Effect.gen(function* () {
147 const { url, project, location, datasetId, tableId } = yield* stack;
148 const baseUrl = baseUrlOf(url);
149 yield* getWhenReady(`${baseUrl}/`);
150
151 const jobs = yield* jobsFor(project, location, baseUrl);
152 const jobNamed = (path: string) => {
153 const job = jobs.find((candidate) =>
154 candidate.httpTarget?.uri?.endsWith(path),
155 );
156 if (job?.name === undefined) throw new Error(`no job for ${path}`);
157 return job.name;
158 };
159 const heartbeatJob = jobNamed("/heartbeat");
160 const dailyJob = jobNamed("/dailyrollup");
161
162 // Don't wait for the cron: run both jobs now. Each job's retry policy
163 // redelivers until the fresh service and its grants are ready.
164 for (const name of [heartbeatJob, dailyJob]) {
165 yield* scheduler
166 .runProjectsLocationsJobs({ name, body: {} })
167 .pipe(Effect.orDie, Effect.provide(GcpHttp));
168 }
169
170 const shortName = (name: string) => name.split("/").pop()!;
171
172 const heartbeat = yield* heartbeatsOf(baseUrl, "heartbeat").pipe(
173 Effect.map((rows) =>
174 rows.find((row) => row.jobName === shortName(heartbeatJob)),
175 ),
176 Effect.repeat({
177 schedule: Schedule.spaced("10 seconds"),
178 until: (row) => row !== undefined,
179 times: 30,
180 }),
181 );
182 expect(heartbeat).toBeDefined();
183 expect(heartbeat!.scheduleTime).toMatch(/^\d{4}-\d{2}-\d{2}T/);
184 expect(typeof heartbeat!.value).toBe("number");
185
186 // The daily cron cannot fire during the test, so its row is the forced run.
187 const daily = yield* heartbeatsOf(baseUrl, "daily").pipe(
188 Effect.map((rows) =>
189 rows.find((row) => row.jobName === shortName(dailyJob)),
190 ),
191 Effect.repeat({
192 schedule: Schedule.spaced("10 seconds"),
193 until: (row) => row !== undefined,
194 times: 30,
195 }),
196 );
197 expect(daily).toBeDefined();
198 expect(daily!.value).toBeGreaterThanOrEqual(0);
199
200 // The rows are real BigQuery rows, not just what the service answers.
201 const result = yield* bigquery
202 .queryJobs({
203 projectId: project,
204 body: {
205 query: `SELECT kind, COUNT(*) AS n FROM \`${project}.${datasetId}.${tableId}\` GROUP BY kind`,
206 useLegacySql: false,
207 location,
208 timeoutMs: 30_000,
209 },
210 })
211 .pipe(Effect.orDie, Effect.provide(GcpHttp));
212 expect(result.jobComplete).toBe(true);
213 const counts = Object.fromEntries(
214 (result.rows ?? []).map((row) => [
215 row.f?.[0]?.v as string,
216 Number(row.f?.[1]?.v),
217 ]),
218 );
219 expect(counts.heartbeat).toBeGreaterThanOrEqual(1);
220 expect(counts.daily).toBeGreaterThanOrEqual(1);
221 }),
222 { timeout: 600_000 },
223 );
224});