1import * as GCP from "@/GCP";
2import { GcpEnvironment } from "@/GCP/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as Core from "@/Test/Core";
5import * as resourcemanager from "@distilled.cloud/gcp/cloudresourcemanager_v3";
6import * as storage from "@distilled.cloud/gcp/storage_v1";
7import * as storagetransfer from "@distilled.cloud/gcp/storagetransfer_v1";
8import { describe, expect } from "alchemy-test";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import { dockerAvailable, expectProbe } from "../bindingHost.ts";
12import StorageTransferBindingsHost, {
13 Copy,
14 SEED_KEY,
15 SEED_TEXT,
16 Sink,
17 Source,
18} from "./fixtures/bindings-host.ts";
19
20const testOptions = { providers: GCP.providers() };
21const { test, beforeAll, afterAll } = Test.make(testOptions);
22const sharedStack = Core.scratchStack(testOptions, "StorageTransferBindings");
23
24let baseUrl: string;
25let hostAccount: string;
26let project: string;
27let jobName: string;
28let sinkBucket: string;
29let transferAccount: string;
30
31/** Project-level roles held by the host's service account. */
32const projectRoles = () =>
33 resourcemanager
34 .getIamPolicyProjects({
35 resource: `projects/${project}`,
36 body: { options: { requestedPolicyVersion: 3 } },
37 })
38 .pipe(
39 Effect.map((policy) =>
40 (policy.bindings ?? [])
41 .filter((binding) =>
42 (binding.members ?? []).includes(`serviceAccount:${hostAccount}`),
43 )
44 .map((binding) => ({
45 role: binding.role,
46 condition: binding.condition?.expression,
47 }))
48 .sort((a, b) => (a.role ?? "").localeCompare(b.role ?? "")),
49 ),
50 );
51
52/**
53 * Let the Google-managed transfer agent read the source and write the
54 * sink (the transfer itself runs as that agent, not as the host).
55 */
56const grantBucketRole = (bucket: string, member: string, role: string) =>
57 Effect.gen(function* () {
58 const policy = yield* storage.getIamPolicyBuckets({
59 bucket,
60 optionsRequestedPolicyVersion: 3,
61 });
62 const bindings = [...(policy.bindings ?? [])];
63 const existing = bindings.find((binding) => binding.role === role);
64 if (existing?.members?.includes(member)) return;
65 if (existing) {
66 existing.members = [...(existing.members ?? []), member];
67 } else {
68 bindings.push({ role, members: [member] });
69 }
70 yield* storage.setIamPolicyBuckets({
71 bucket,
72 body: { ...policy, bindings },
73 });
74 }).pipe(
75 Effect.retry({
76 while: (error) => error._tag === "Conflict",
77 times: 4,
78 schedule: Schedule.spaced("500 millis"),
79 }),
80 );
81
82describe.skipIf(!dockerAvailable)(
83 "StorageTransfer Bindings",
84 {
85 tags: [
86 "provider:gcp",
87 "provider:gcp:storagetransfer",
88 "provider:gcp:run",
89 "live",
90 ],
91 },
92 () => {
93 beforeAll(
94 Effect.gen(function* () {
95 yield* sharedStack.destroy();
96 // The job's buckets must admit the transfer agent before the job
97 // exists, so deploy them first.
98 const buckets = yield* sharedStack.deploy(
99 Effect.gen(function* () {
100 const source = yield* Source;
101 const sink = yield* Sink;
102 const { project } = yield* GcpEnvironment.current;
103 return {
104 project,
105 source: source.bucketName,
106 sink: sink.bucketName,
107 };
108 }),
109 );
110 // Out-of-band calls in a hook need the provider services.
111 yield* Core.withProviders(
112 Effect.gen(function* () {
113 const account = yield* storagetransfer.getGoogleServiceAccounts({
114 projectId: buckets.project,
115 });
116 const member = `serviceAccount:${account.accountEmail}`;
117 yield* grantBucketRole(
118 buckets.source,
119 member,
120 "roles/storage.objectViewer",
121 );
122 yield* grantBucketRole(
123 buckets.source,
124 member,
125 "roles/storage.legacyBucketReader",
126 );
127 yield* grantBucketRole(
128 buckets.sink,
129 member,
130 "roles/storage.legacyBucketWriter",
131 );
132 yield* grantBucketRole(
133 buckets.sink,
134 member,
135 "roles/storage.objectAdmin",
136 );
137 }),
138 testOptions,
139 "StorageTransferBindings",
140 );
141 const out = yield* sharedStack.deploy(
142 Effect.gen(function* () {
143 const host = yield* StorageTransferBindingsHost;
144 const job = yield* Copy;
145 return {
146 uri: host.uri,
147 serviceAccount: host.serviceAccount,
148 project: host.project,
149 job: job.name,
150 };
151 }),
152 );
153 baseUrl = out.uri!;
154 hostAccount = out.serviceAccount!;
155 project = out.project;
156 jobName = out.job;
157 sinkBucket = buckets.sink;
158 }),
159 { timeout: 900_000 },
160 );
161
162 afterAll(sharedStack.destroy(), { timeout: 600_000 });
163
164 describe("GetGoogleServiceAccount", () => {
165 test.provider(
166 "returns the project's transfer agent as the host",
167 (_stack) =>
168 Effect.gen(function* () {
169 const out = yield* expectProbe<{ accountEmail: string }>(
170 baseUrl,
171 "getGoogleServiceAccount",
172 );
173 const account = yield* storagetransfer.getGoogleServiceAccounts({
174 projectId: project,
175 });
176 expect(out.accountEmail).toEqual(account.accountEmail);
177 expect(out.accountEmail).toMatch(
178 /^project-\d+@storage-transfer-service\.iam\.gserviceaccount\.com$/,
179 );
180 expect(yield* projectRoles()).toContainEqual({
181 role: "roles/storagetransfer.viewer",
182 condition: undefined,
183 });
184 }),
185 {
186 tags: ["provider:gcp", "provider:gcp:storagetransfer", "live"],
187 timeout: 600_000,
188 },
189 );
190 });
191
192 describe("RunTransferJob", () => {
193 test.provider(
194 "runs the bound job as the host and the object lands in the sink",
195 (_stack) =>
196 Effect.gen(function* () {
197 const started = yield* expectProbe<{ name: string }>(
198 baseUrl,
199 "runTransferJob",
200 );
201 expect(started.name).toMatch(/^transferOperations\//);
202
203 const operation = yield* storagetransfer
204 .getTransferOperations({ name: started.name })
205 .pipe(
206 Effect.repeat({
207 schedule: Schedule.spaced("5 seconds"),
208 until: (current) => current.done === true,
209 times: 30,
210 }),
211 );
212 expect(operation.done).toEqual(true);
213 expect(operation.error).toEqual(undefined);
214 expect(operation.metadata?.transferJobName).toEqual(jobName);
215
216 const copied = yield* storage.getObjects({
217 bucket: sinkBucket,
218 object: SEED_KEY,
219 });
220 expect(copied.size).toEqual(String(SEED_TEXT.length));
221
222 expect(yield* projectRoles()).toEqual([
223 { role: "roles/storagetransfer.user", condition: undefined },
224 { role: "roles/storagetransfer.viewer", condition: undefined },
225 ]);
226 }),
227 {
228 tags: ["provider:gcp", "provider:gcp:storagetransfer", "live"],
229 timeout: 600_000,
230 },
231 );
232 });
233 },
234);