1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as storage from "@distilled.cloud/gcp/storage_v1";
4import * as storagetransfer from "@distilled.cloud/gcp/storagetransfer_v1";
5import { expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import { MinimumLogLevel } from "effect/References";
8import * as Schedule from "effect/Schedule";
9import { GcpEnvironment } from "@/GCP/Environment";
10
11const { test } = Test.make({ providers: GCP.providers() });
12
13const logLevel = Effect.provideService(
14 MinimumLogLevel,
15 process.env.DEBUG ? "Debug" : "Info",
16);
17
18const waitUntilGone = (jobName: string) =>
19 Effect.gen(function* () {
20 const { project } = yield* GcpEnvironment.current;
21
22 return yield* storagetransfer
23 .getTransferJobs({ jobName, projectId: project })
24 .pipe(
25 Effect.map((job) =>
26 job.status === "DELETED" ? ("gone" as const) : ("found" as const),
27 ),
28 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
29 Effect.repeat({
30 schedule: Schedule.spaced("1 second"),
31 until: (status) => status === "gone",
32 times: 10,
33 }),
34 );
35 });
36
37const grantBucketRole = (bucket: string, member: string, role: string) =>
38 Effect.gen(function* () {
39 const policy = yield* storage.getIamPolicyBuckets({
40 bucket,
41 optionsRequestedPolicyVersion: 3,
42 });
43 const bindings = [...(policy.bindings ?? [])];
44 const existing = bindings.find((binding) => binding.role === role);
45 if (existing?.members?.includes(member)) return;
46 if (existing) {
47 existing.members = [...(existing.members ?? []), member];
48 } else {
49 bindings.push({ role, members: [member] });
50 }
51 yield* storage.setIamPolicyBuckets({
52 bucket,
53 body: {
54 ...policy,
55 bindings,
56 },
57 });
58 }).pipe(
59 Effect.retry({
60 while: (error) => error._tag === "Conflict",
61 times: 4,
62 schedule: Schedule.spaced("500 millis"),
63 }),
64 );
65
66const grantTransferAccess = (
67 sourceBucket: string,
68 sinkBucket: string,
69 email: string,
70) => {
71 const member = `serviceAccount:${email}`;
72 return Effect.all(
73 [
74 grantBucketRole(sourceBucket, member, "roles/storage.legacyBucketReader"),
75 grantBucketRole(sourceBucket, member, "roles/storage.objectViewer"),
76 grantBucketRole(sinkBucket, member, "roles/storage.legacyBucketReader"),
77 grantBucketRole(sinkBucket, member, "roles/storage.objectAdmin"),
78 ],
79 { concurrency: 1 },
80 );
81};
82
83test.provider(
84 "create, update, and delete a transfer job",
85 (stack) =>
86 Effect.gen(function* () {
87 const { project } = yield* GcpEnvironment.current;
88
89 yield* stack.destroy();
90
91 const buckets = yield* stack.deploy(
92 Effect.gen(function* () {
93 const source = yield* GCP.Storage.Bucket("Src", {
94 location: "US-CENTRAL1",
95 forceDestroy: true,
96 });
97 const sink = yield* GCP.Storage.Bucket("Dst", {
98 location: "US-CENTRAL1",
99 forceDestroy: true,
100 });
101 return {
102 sourceBucket: source.bucketName.as<string>(),
103 sinkBucket: sink.bucketName.as<string>(),
104 };
105 }),
106 );
107
108 const account = yield* storagetransfer.getGoogleServiceAccounts({
109 projectId: project,
110 });
111 if (account.accountEmail) {
112 yield* grantTransferAccess(
113 buckets.sourceBucket,
114 buckets.sinkBucket,
115 account.accountEmail,
116 );
117 }
118
119 const created = yield* stack.deploy(
120 Effect.gen(function* () {
121 yield* GCP.Storage.Bucket("Src", {
122 bucketName: buckets.sourceBucket,
123 location: "US-CENTRAL1",
124 forceDestroy: true,
125 });
126 yield* GCP.Storage.Bucket("Dst", {
127 bucketName: buckets.sinkBucket,
128 location: "US-CENTRAL1",
129 forceDestroy: true,
130 });
131 return yield* GCP.StorageTransfer.TransferJob("Copy", {
132 description: "copy uploads",
133 status: "DISABLED",
134 transferSpec: {
135 gcsDataSource: { bucketName: buckets.sourceBucket },
136 gcsDataSink: { bucketName: buckets.sinkBucket },
137 },
138 });
139 }),
140 );
141
142 expect(created.jobId).toEqual(expect.any(String));
143 expect(created.name).toEqual(`transferJobs/${created.jobId}`);
144 expect(created.project).toEqual(project);
145 expect(created.description).toEqual("copy uploads");
146 expect(created.status).toEqual("DISABLED");
147 expect(created.transferSpec?.gcsDataSource?.bucketName).toEqual(
148 buckets.sourceBucket,
149 );
150 expect(created.transferSpec?.gcsDataSink?.bucketName).toEqual(
151 buckets.sinkBucket,
152 );
153
154 const fetched = yield* storagetransfer.getTransferJobs({
155 jobName: created.name,
156 projectId: project,
157 });
158 expect(fetched.name).toEqual(created.name);
159 expect(fetched.status).toEqual("DISABLED");
160 expect(fetched.description).toContain("alchemy-id=");
161 expect(fetched.description).toContain("copy uploads");
162 expect(fetched.transferSpec?.gcsDataSource?.bucketName).toEqual(
163 buckets.sourceBucket,
164 );
165
166 const updated = yield* stack.deploy(
167 Effect.gen(function* () {
168 yield* GCP.Storage.Bucket("Src", {
169 bucketName: buckets.sourceBucket,
170 location: "US-CENTRAL1",
171 forceDestroy: true,
172 });
173 yield* GCP.Storage.Bucket("Dst", {
174 bucketName: buckets.sinkBucket,
175 location: "US-CENTRAL1",
176 forceDestroy: true,
177 });
178 return yield* GCP.StorageTransfer.TransferJob("Copy", {
179 jobId: created.jobId,
180 description: "copy inbox",
181 status: "DISABLED",
182 transferSpec: {
183 gcsDataSource: {
184 bucketName: buckets.sourceBucket,
185 path: "inbox/",
186 },
187 gcsDataSink: { bucketName: buckets.sinkBucket },
188 },
189 });
190 }),
191 );
192
193 expect(updated.name).toEqual(created.name);
194 expect(updated.jobId).toEqual(created.jobId);
195 expect(updated.description).toEqual("copy inbox");
196 expect(updated.transferSpec?.gcsDataSource?.path).toEqual("inbox/");
197
198 const fetchedUpdate = yield* storagetransfer.getTransferJobs({
199 jobName: created.name,
200 projectId: project,
201 });
202 expect(fetchedUpdate.description).toContain("copy inbox");
203 expect(fetchedUpdate.transferSpec?.gcsDataSource?.path).toEqual("inbox/");
204 expect(fetchedUpdate.status).toEqual("DISABLED");
205
206 yield* stack.destroy();
207
208 const gone = yield* waitUntilGone(created.name);
209 expect(gone).toEqual("gone");
210 }).pipe(logLevel),
211 {
212 tags: ["provider:gcp", "provider:gcp:storagetransfer", "live"],
213 timeout: 120_000,
214 },
215);