1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as cloudrun from "@distilled.cloud/gcp/run_v2";
4import * as storage from "@distilled.cloud/gcp/storage_v1";
5import { expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10import { spawnSync } from "node:child_process";
11import MarkerJob, { MARKER_OBJECT } from "./fixtures/job.ts";
12
13const { test } = Test.make({ providers: GCP.providers() });
14
15const logLevel = Effect.provideService(
16 MinimumLogLevel,
17 process.env.DEBUG ? "Debug" : "Info",
18);
19
20const dockerAvailable = (() => {
21 try {
22 return (
23 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 })
24 .status === 0
25 );
26 } catch {
27 return false;
28 }
29})();
30
31const IMAGE = "us-docker.pkg.dev/cloudrun/container/job:latest";
32
33class JobRunNotReady extends Data.TaggedError("JobRunNotReady")<{
34 reason: string;
35}> {}
36
37const waitUntilGone = (name: string) =>
38 cloudrun.getProjectsLocationsJobs({ name }).pipe(
39 Effect.as("found" as const),
40 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
41 Effect.repeat({
42 schedule: Schedule.spaced("1 second"),
43 until: (status) => status === "gone",
44 times: 10,
45 }),
46 );
47
48test.provider(
49 "create, update, and delete a job",
50 (stack) =>
51 Effect.gen(function* () {
52 yield* stack.destroy();
53
54 const created = yield* stack.deploy(
55 Effect.gen(function* () {
56 return yield* GCP.Run.Job("Migrate", {
57 location: "us-central1",
58 labels: { env: "test" },
59 containers: [{ image: IMAGE }],
60 });
61 }),
62 );
63
64 expect(created.name).toContain("/jobs/");
65 expect(created.jobId).toEqual(expect.any(String));
66 expect(created.location).toEqual("us-central1");
67 expect(created.labels).toMatchObject({ env: "test" });
68 expect(created.image).toEqual(IMAGE);
69
70 const fetched = yield* cloudrun.getProjectsLocationsJobs({
71 name: created.name,
72 });
73 expect(fetched.name).toEqual(created.name);
74 expect(fetched.labels?.env).toEqual("test");
75 expect(fetched.template?.template?.containers?.[0]?.image).toEqual(IMAGE);
76
77 const updated = yield* stack.deploy(
78 Effect.gen(function* () {
79 return yield* GCP.Run.Job("Migrate", {
80 jobId: created.jobId,
81 location: "us-central1",
82 labels: { env: "prod", role: "migrate" },
83 maxRetries: 1,
84 timeout: "120s",
85 containers: [
86 {
87 image: IMAGE,
88 env: [{ name: "STAGE", value: "prod" }],
89 },
90 ],
91 });
92 }),
93 );
94
95 expect(updated.name).toEqual(created.name);
96 expect(updated.labels).toMatchObject({ env: "prod", role: "migrate" });
97 expect(updated.maxRetries).toEqual(1);
98 expect(updated.timeout).toEqual("120s");
99
100 const refetched = yield* cloudrun.getProjectsLocationsJobs({
101 name: created.name,
102 });
103 expect(refetched.labels?.env).toEqual("prod");
104 expect(refetched.labels?.role).toEqual("migrate");
105 expect(refetched.template?.template?.maxRetries).toEqual(1);
106 expect(refetched.template?.template?.timeout).toEqual("120s");
107 expect(refetched.template?.template?.containers?.[0]?.env).toEqual(
108 expect.arrayContaining([
109 expect.objectContaining({ name: "STAGE", value: "prod" }),
110 ]),
111 );
112
113 yield* stack.destroy();
114
115 const gone = yield* waitUntilGone(created.name);
116 expect(gone).toEqual("gone");
117 }).pipe(logLevel),
118 { tags: ["provider:gcp", "provider:gcp:run", "live"], timeout: 120_000 },
119);
120
121test.provider.skipIf(!dockerAvailable)(
122 "effect-native Job run writes a Storage object",
123 (stack) =>
124 Effect.gen(function* () {
125 yield* stack.destroy();
126
127 const out = yield* stack.deploy(
128 Effect.gen(function* () {
129 const job = yield* MarkerJob;
130 return { name: job.name, project: job.project };
131 }),
132 );
133
134 yield* cloudrun.runProjectsLocationsJobs({ name: out.name });
135
136 const execution = yield* cloudrun
137 .listProjectsLocationsJobsExecutions({
138 parent: out.name,
139 pageSize: 10,
140 })
141 .pipe(
142 Effect.map((page) => page.executions?.[0]),
143 Effect.filterOrFail(
144 (item): item is cloudrun.GoogleCloudRunV2Execution =>
145 item !== undefined &&
146 (item.completionTime ?? "").length > 0 &&
147 (item.succeededCount ?? 0) >= 1,
148 (item) =>
149 new JobRunNotReady({
150 reason: item?.completionTime
151 ? `failed:${item.failedCount ?? 0}`
152 : "pending",
153 }),
154 ),
155 Effect.retry({
156 while: (error): error is JobRunNotReady =>
157 error._tag === "JobRunNotReady" && error.reason === "pending",
158 // Cloud Run schedules a fresh execution slowly: even Google's
159 // sample job takes ~2 minutes to start in the test project.
160 schedule: Schedule.spaced("10 seconds"),
161 times: 36,
162 }),
163 );
164 expect((execution.succeededCount ?? 0) >= 1).toEqual(true);
165
166 const listed = yield* storage.listBuckets({
167 project: out.project,
168 maxResults: 1000,
169 });
170 const markerBucket = (listed.items ?? []).find((bucket) =>
171 (bucket.labels?.["alchemy-id"] ?? "").includes("jobrunmarker"),
172 );
173 expect(markerBucket?.name).toEqual(expect.any(String));
174 const object = yield* storage.getObjects({
175 bucket: markerBucket!.name!,
176 object: MARKER_OBJECT,
177 });
178 expect(object.name).toEqual(MARKER_OBJECT);
179
180 yield* stack.destroy();
181 const gone = yield* waitUntilGone(out.name);
182 expect(gone).toEqual("gone");
183 }).pipe(logLevel),
184 { tags: ["provider:gcp", "provider:gcp:run", "live"], timeout: 540_000 },
185);