1import * as bigquery from "@distilled.cloud/gcp/bigquery_v2";
2import * as scheduler from "@distilled.cloud/gcp/cloudscheduler_v1";
3import { Credentials } from "@distilled.cloud/gcp/Credentials";
4import * as run from "@distilled.cloud/gcp/run_v2";
5import * as storage from "@distilled.cloud/gcp/storage_v1";
6import * as Alchemy from "alchemy";
7import * as GCP from "alchemy/GCP";
8import * as Test from "alchemy/Test/Bun";
9import { describe, expect } from "bun:test";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Schedule from "effect/Schedule";
13import * as FetchHttpClient from "effect/http/FetchHttpClient";
14import * as HttpClient from "effect/http/HttpClient";
15import * as HttpClientRequest from "effect/http/HttpClientRequest";
16import { spawnSync } from "node:child_process";
17import Stack from "../alchemy.run.ts";
18import { SUMMARY_PREFIX, type DailySummary } from "../src/resources.ts";
19
20const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
21 providers: GCP.providers(),
22 state: Alchemy.localState(),
23});
24
25const { getWhenReady } = Test;
26
27// Out-of-band calls to the Google APIs resolve the same stored credentials
28// the deploy uses, so the test runs against the configured profile.
29const GcpHttp = Layer.mergeAll(
30 GCP.GcpAuth,
31 GCP.fromAuthProvider(),
32 FetchHttpClient.layer,
33);
34
35// The job and the service are built from `main`, which needs a local
36// image build.
37const dockerAvailable =
38 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 }).status ===
39 0;
40
41// Deploy, tests and destroy all sit behind the same Docker guard.
42describe.skipIf(!dockerAvailable)("gcp-scheduled-job", () => {
43 const stack = beforeAll(deploy(Stack), { timeout: 900_000 });
44
45 afterAll.skipIf(!!process.env.NO_DESTROY)(
46 Effect.gen(function* () {
47 const { jobName, schedulerJobName, bucketName } = yield* stack;
48 yield* destroy(Stack);
49
50 const found = Effect.as("found" as const);
51 const gone = <A, E extends { readonly _tag: string }, R>(
52 self: Effect.Effect<A, E, R>,
53 ) =>
54 self.pipe(
55 Effect.catchIf(
56 (error) => error._tag === "NotFound",
57 () => Effect.succeed("gone" as const),
58 ),
59 );
60 const [job, schedulerJob, bucket] = yield* Effect.all([
61 run.getProjectsLocationsJobs({ name: jobName }).pipe(found, gone),
62 scheduler
63 .getProjectsLocationsJobs({ name: schedulerJobName })
64 .pipe(found, gone),
65 storage.getBuckets({ bucket: bucketName }).pipe(found, gone),
66 ]).pipe(Effect.orDie, Effect.provide(GcpHttp));
67 expect({ job, schedulerJob, bucket }).toEqual({
68 job: "gone",
69 schedulerJob: "gone",
70 bucket: "gone",
71 });
72 }),
73 { timeout: 600_000 },
74 );
75
76 const baseUrlOf = (url: string | undefined) => {
77 if (url === undefined) throw new Error("the service has no URL");
78 return url.replace(/\/+$/, "");
79 };
80
81 /** Every execution the Cloud Run job has, newest first. */
82 const executionsOf = (jobName: string) =>
83 run
84 .listProjectsLocationsJobsExecutions({ parent: jobName, pageSize: 100 })
85 .pipe(
86 Effect.map((page) => page.executions ?? []),
87 Effect.orDie,
88 Effect.provide(GcpHttp),
89 );
90
91 /** Download an object's content with the test's own credentials. */
92 const download = (bucket: string, object: string) =>
93 Effect.gen(function* () {
94 const { accessToken } = yield* yield* Credentials;
95 const res = yield* HttpClient.execute(
96 HttpClientRequest.get(
97 `https://storage.googleapis.com/storage/v1/b/${bucket}/o/${encodeURIComponent(object)}`,
98 ).pipe(
99 HttpClientRequest.setUrlParams({ alt: "media" }),
100 HttpClientRequest.bearerToken(accessToken),
101 ),
102 );
103 expect(res.status).toBe(200);
104 return yield* res.text;
105 }).pipe(Effect.orDie, Effect.provide(GcpHttp));
106
107 /** Every summary currently in the bucket. */
108 const summariesIn = (bucket: string) =>
109 Effect.gen(function* () {
110 const { items = [] } = yield* storage
111 .listObjects({ bucket, prefix: SUMMARY_PREFIX })
112 .pipe(Effect.orDie, Effect.provide(GcpHttp));
113 const summaries: DailySummary[] = [];
114 for (const item of items) {
115 if (item.name === undefined) continue;
116 summaries.push(
117 JSON.parse(yield* download(bucket, item.name)) as DailySummary,
118 );
119 }
120 return summaries;
121 });
122
123 test(
124 "POST /run summarizes seeded orders into the bucket",
125 Effect.gen(function* () {
126 const { url, project, datasetId, tableId, bucketName } = yield* stack;
127 const baseUrl = baseUrlOf(url);
128
129 // Unique region names keep the check independent of other rows.
130 const tag = crypto.randomUUID().slice(0, 8);
131 const east = `east-${tag}`;
132 const west = `west-${tag}`;
133 const occurredAt = new Date(Date.now() - 5 * 60_000).toISOString();
134 const seed = [
135 { region: east, amountCents: 1000 },
136 { region: east, amountCents: 2500 },
137 { region: east, amountCents: 499 },
138 { region: west, amountCents: 1200 },
139 { region: west, amountCents: 800 },
140 ].map((row, i) => ({ id: `${tag}-${i}`, occurredAt, ...row }));
141
142 // A just-created table can briefly 404 on the streaming path.
143 const inserted = yield* bigquery
144 .insertAllTabledata({
145 projectId: project,
146 datasetId,
147 tableId,
148 body: { rows: seed.map((json) => ({ insertId: json.id, json })) },
149 })
150 .pipe(
151 Effect.retry({
152 while: (error) => error._tag === "NotFound",
153 schedule: Schedule.spaced("5 seconds"),
154 times: 12,
155 }),
156 Effect.orDie,
157 Effect.provide(GcpHttp),
158 );
159 expect(inserted.insertErrors ?? []).toEqual([]);
160
161 yield* getWhenReady(`${baseUrl}/`);
162 const started = yield* HttpClient.execute(
163 HttpClientRequest.post(`${baseUrl}/run`),
164 );
165 expect(started.status).toBe(202);
166 const { operation } = (yield* started.json) as {
167 operation: string | null;
168 };
169 expect(operation).toContain("/operations/");
170
171 // An execution can take ~2 minutes to start, then runs for seconds.
172 const hasOurs = (summary: DailySummary) =>
173 summary.regions.some((r) => r.region === east) &&
174 summary.regions.some((r) => r.region === west);
175 const summary = yield* summariesIn(bucketName).pipe(
176 Effect.map((summaries) => summaries.find(hasOurs)),
177 Effect.repeat({
178 schedule: Schedule.spaced("10 seconds"),
179 until: (found) => found !== undefined,
180 times: 48,
181 }),
182 );
183 expect(summary).toBeDefined();
184
185 const regionOf = (name: string) =>
186 summary!.regions.find((r) => r.region === name);
187 expect(regionOf(east)).toEqual({
188 region: east,
189 orders: 3,
190 revenueCents: 3999,
191 });
192 expect(regionOf(west)).toEqual({
193 region: west,
194 orders: 2,
195 revenueCents: 2000,
196 });
197 expect(summary!.orders).toBeGreaterThanOrEqual(5);
198 expect(summary!.revenueCents).toBeGreaterThanOrEqual(5999);
199 expect(summary!.execution).toBeTruthy();
200 expect(
201 new Date(summary!.to).getTime() - new Date(summary!.from).getTime(),
202 ).toBe(24 * 60 * 60 * 1000);
203 }),
204 { timeout: 720_000 },
205 );
206
207 test(
208 "the Cloud Scheduler job starts an execution through the Admin API",
209 Effect.gen(function* () {
210 const { jobName, schedulerJobName } = yield* stack;
211
212 const schedulerJob = yield* scheduler
213 .getProjectsLocationsJobs({ name: schedulerJobName })
214 .pipe(Effect.orDie, Effect.provide(GcpHttp));
215 expect(schedulerJob.schedule).toEqual("0 2 * * *");
216 expect(schedulerJob.httpTarget?.uri).toEqual(
217 `https://run.googleapis.com/v2/${jobName}:run`,
218 );
219 const job = yield* run
220 .getProjectsLocationsJobs({ name: jobName })
221 .pipe(Effect.orDie, Effect.provide(GcpHttp));
222 expect(schedulerJob.httpTarget?.oauthToken?.serviceAccountEmail).toEqual(
223 job.template?.template?.serviceAccount,
224 );
225
226 const before = (yield* executionsOf(jobName)).map((e) => e.name);
227
228 // Don't wait for 02:00: force a run now.
229 yield* scheduler
230 .runProjectsLocationsJobs({ name: schedulerJobName, body: {} })
231 .pipe(Effect.orDie, Effect.provide(GcpHttp));
232
233 const fresh = yield* executionsOf(jobName).pipe(
234 Effect.map((executions) =>
235 executions.filter((e) => !before.includes(e.name)),
236 ),
237 Effect.repeat({
238 schedule: Schedule.spaced("10 seconds"),
239 until: (executions) => executions.length > 0,
240 times: 30,
241 }),
242 );
243 expect(fresh.length).toBeGreaterThanOrEqual(1);
244
245 // Cloud Scheduler records the Admin API's answer; code 0 (or unset)
246 // means `jobs:run` accepted the OAuth token.
247 const attempted = yield* scheduler
248 .getProjectsLocationsJobs({ name: schedulerJobName })
249 .pipe(
250 Effect.orDie,
251 Effect.provide(GcpHttp),
252 Effect.repeat({
253 schedule: Schedule.spaced("5 seconds"),
254 until: (j) => j.lastAttemptTime !== undefined,
255 times: 24,
256 }),
257 );
258 expect(attempted.status?.code ?? 0).toBe(0);
259 }),
260 { timeout: 600_000 },
261 );
262});