1import * as storagetransfer from "@distilled.cloud/gcp/storagetransfer_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import {
8 alchemyLabelKeys,
9 createInternalLabels,
10 hasAlchemyLabels,
11} from "../Labels.ts";
12
13export const MAX_AGENT_POOL_ID = 128;
14export const MAX_AGENT_POOL_DISPLAY_NAME = 127;
15export const MAX_JOB_ID = 115;
16export const MAX_JOB_DESCRIPTION = 1024;
17export const JOB_NAME_PREFIX = "transferJobs/";
18export const DEFAULT_JOB_STATUS = "ENABLED";
19export const DEFAULT_AGENT_POOL = "transfer_service_default";
20
21export class ResourceNotResolved extends Data.TaggedError(
22 "GCP.StorageTransfer.ResourceNotResolved",
23)<{
24 name: string;
25}> {}
26
27export class ResourceNotReady extends Data.TaggedError(
28 "GCP.StorageTransfer.ResourceNotReady",
29)<{
30 name: string;
31 state: string;
32}> {}
33
34export class ResourceStillExists extends Data.TaggedError(
35 "GCP.StorageTransfer.ResourceStillExists",
36)<{
37 name: string;
38}> {}
39
40export const lastSegment = (value: string) => {
41 const trimmed = value.replace(/\/+$/, "");
42 const parts = trimmed.split("/");
43 return parts[parts.length - 1] || trimmed;
44};
45
46export const agentPoolName = (project: string, agentPoolId: string) =>
47 `projects/${project}/agentPools/${agentPoolId}`;
48
49export const transferJobName = (jobId: string) =>
50 jobId.startsWith(JOB_NAME_PREFIX) ? jobId : `${JOB_NAME_PREFIX}${jobId}`;
51
52export const jobIdOf = (name: string) =>
53 name.startsWith(JOB_NAME_PREFIX)
54 ? name.slice(JOB_NAME_PREFIX.length)
55 : lastSegment(name);
56
57export const agentPoolIdOf = (name: string) => lastSegment(name);
58
59const startsWithLetter = (value: string) => /^[a-z]/.test(value);
60
61const endsAlnum = (value: string) => /[a-z0-9]$/.test(value);
62
63export const sanitizeAgentPoolId = (value: string) => {
64 let next = lastSegment(value)
65 .toLowerCase()
66 .replace(/[^a-z0-9-._~]/g, "-")
67 .replace(/^-+|-+$/g, "");
68 if (next.startsWith("goog")) next = `x${next}`;
69 if (!startsWithLetter(next)) next = `p${next}`;
70 next = next.slice(0, MAX_AGENT_POOL_ID);
71 if (next.length > 1 && !endsAlnum(next)) {
72 next = `${next.slice(0, MAX_AGENT_POOL_ID - 1)}0`;
73 }
74 return next.length > 0 ? next : "pool";
75};
76
77export const sanitizeJobId = (value: string) => {
78 let next = jobIdOf(value)
79 .replace(/[^A-Za-z0-9-._~]/g, "-")
80 .replace(/^-+|-+$/g, "");
81 if (next.toUpperCase().startsWith("OPI")) next = `x${next}`;
82 next = next.slice(0, MAX_JOB_ID);
83 if (next.length === 0) next = "job";
84 if (!endsAlnum(next.toLowerCase())) {
85 next = `${next.slice(0, MAX_JOB_ID - 1)}0`;
86 }
87 return next;
88};
89
90export const toAgentPoolId = (
91 id: string,
92 explicit: string | undefined,
93 existing?: string,
94) =>
95 Effect.gen(function* () {
96 if (explicit !== undefined) return sanitizeAgentPoolId(explicit);
97 if (existing !== undefined) return existing;
98 return sanitizeAgentPoolId(
99 yield* createPhysicalName({
100 id,
101 maxLength: 63,
102 lowercase: true,
103 forbiddenPrefixes: ["goog"],
104 }),
105 );
106 });
107
108export const toJobId = (
109 id: string,
110 explicit: string | undefined,
111 existing?: string,
112) =>
113 Effect.gen(function* () {
114 if (explicit !== undefined) return sanitizeJobId(explicit);
115 if (existing !== undefined) return existing;
116 return sanitizeJobId(
117 yield* createPhysicalName({
118 id,
119 maxLength: 63,
120 lowercase: true,
121 forbiddenPrefixes: ["opi"],
122 }),
123 );
124 });
125
126const markerOf = (stack: string, stage: string, id: string) =>
127 `[alchemy ${alchemyLabelKeys.stack}=${stack} ${alchemyLabelKeys.stage}=${stage} ${alchemyLabelKeys.id}=${id}]`;
128
129const fitMarker = (labels: Record<string, string>, maxLength: number) => {
130 let stack = labels[alchemyLabelKeys.stack] ?? "x";
131 let stage = labels[alchemyLabelKeys.stage] ?? "x";
132 let id = labels[alchemyLabelKeys.id] ?? "x";
133 let marker = markerOf(stack, stage, id);
134 while (
135 marker.length > maxLength &&
136 (stack.length > 1 || stage.length > 1 || id.length > 1)
137 ) {
138 if (stack.length >= stage.length && stack.length >= id.length) {
139 stack = stack.slice(0, -1);
140 } else if (stage.length >= id.length) {
141 stage = stage.slice(0, -1);
142 } else {
143 id = id.slice(0, -1);
144 }
145 marker = markerOf(stack, stage, id);
146 }
147 return marker.slice(0, Math.max(0, maxLength));
148};
149
150export const encodeOwnership = (
151 labels: Record<string, string>,
152 text: string | undefined,
153 maxLength = MAX_JOB_DESCRIPTION,
154): string => {
155 const trimmed = text?.replace(/[\r\n]+/g, " ").trim();
156 if (!trimmed) return fitMarker(labels, maxLength);
157 const minMarker = 24;
158 const reserved = Math.min(
159 trimmed.length + 1,
160 Math.max(0, maxLength - minMarker),
161 );
162 const marker = fitMarker(labels, maxLength - reserved);
163 return `${marker} ${trimmed}`.slice(0, maxLength);
164};
165
166export const parseOwnership = (
167 text: string | undefined,
168): {
169 labels: Record<string, string>;
170 text: string | undefined;
171} => {
172 if (!text?.startsWith("[alchemy ")) {
173 return { labels: {}, text };
174 }
175 const end = text.indexOf("]");
176 if (end < 0) return { labels: {}, text };
177 const labels: Record<string, string> = {};
178 for (const part of text.slice("[alchemy ".length, end).split(/\s+/)) {
179 const eq = part.indexOf("=");
180 if (eq > 0) {
181 labels[part.slice(0, eq)] = part.slice(eq + 1);
182 }
183 }
184 const rest = text.slice(end + 1).replace(/^[\s\n]+/, "");
185 return { labels, text: rest.length > 0 ? rest : undefined };
186};
187
188export const hasOwnershipMarker = (text: string | undefined) =>
189 Object.keys(parseOwnership(text).labels).some((key) =>
190 key.startsWith("alchemy-"),
191 );
192
193const prefixMatch = (expected: string, observed: string) =>
194 expected === observed ||
195 expected.startsWith(observed) ||
196 observed.startsWith(expected);
197
198export const ownedByAlchemy = (id: string, text: string | undefined) =>
199 Effect.gen(function* () {
200 const expected = yield* createInternalLabels(id);
201 const { labels } = parseOwnership(text);
202 if (!hasOwnershipMarker(text)) return false;
203 const exact = yield* hasAlchemyLabels(id, labels);
204 if (exact) return true;
205 return (
206 prefixMatch(
207 expected[alchemyLabelKeys.stack] ?? "",
208 labels[alchemyLabelKeys.stack] ?? "",
209 ) &&
210 prefixMatch(
211 expected[alchemyLabelKeys.stage] ?? "",
212 labels[alchemyLabelKeys.stage] ?? "",
213 ) &&
214 prefixMatch(
215 expected[alchemyLabelKeys.id] ?? "",
216 labels[alchemyLabelKeys.id] ?? "",
217 )
218 );
219 });
220
221export const createOwnership = (id: string) => createInternalLabels(id);
222
223export const canonical = (value: unknown): unknown => {
224 if (value === undefined || value === null) return undefined;
225 if (typeof value === "boolean" || typeof value === "number") return value;
226 if (typeof value === "string") {
227 return value.length === 0 ? undefined : value;
228 }
229 if (Array.isArray(value)) {
230 const items = value.map(canonical).filter((item) => item !== undefined);
231 return items.length === 0 ? undefined : items;
232 }
233 if (typeof value === "object") {
234 const entries = Object.entries(value as Record<string, unknown>)
235 .map(([key, item]) => [key, canonical(item)] as const)
236 .filter(([, item]) => item !== undefined)
237 .sort(([left], [right]) => left.localeCompare(right));
238 if (entries.length === 0) return undefined;
239 return Object.fromEntries(entries);
240 }
241 return undefined;
242};
243
244export const fingerprint = (value: unknown): string =>
245 JSON.stringify(canonical(value) ?? null);
246
247export const sameValue = (left: unknown, right: unknown) =>
248 fingerprint(left) === fingerprint(right);
249
250export const fieldMask = (fields: Array<string | false | undefined>) =>
251 fields
252 .filter((field): field is string => typeof field === "string")
253 .join(",");
254
255export const retryApiDisabled = <A, E extends { readonly _tag: string }, R>(
256 effect: Effect.Effect<A, E, R>,
257) =>
258 effect.pipe(
259 Effect.retry({
260 // The API may still be propagating right after it is enabled.
261 while: (error) => error._tag === "ServiceDisabled",
262 times: 8,
263 schedule: Schedule.spaced("2 seconds"),
264 }),
265 );
266
267export const collectPages = <Page, Item, E, R>(
268 pages: Stream.Stream<Page, E, R>,
269 items: (page: Page) => readonly Item[] | null | undefined,
270) =>
271 pages.pipe(
272 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
273 Stream.runCollect,
274 Effect.map((chunk) => Array.from(chunk)),
275 );
276
277export const getAgentPool = (name: string) =>
278 name.length === 0
279 ? Effect.succeed(undefined)
280 : storagetransfer.getProjectsAgentPools({ name }).pipe(
281 retryApiDisabled,
282 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
283 );
284
285export const getTransferJob = (jobName: string, projectId: string) =>
286 jobName.length === 0
287 ? Effect.succeed(undefined)
288 : storagetransfer.getTransferJobs({ jobName, projectId }).pipe(
289 retryApiDisabled,
290 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
291 );
292
293export const isDeletedJob = (job: storagetransfer.TransferJob | undefined) =>
294 job?.status === "DELETED";
295
296export const waitAgentPoolCreated = (name: string) =>
297 getAgentPool(name).pipe(
298 Effect.filterOrFail(
299 (pool): pool is storagetransfer.AgentPool => pool !== undefined,
300 () => new ResourceNotResolved({ name }),
301 ),
302 Effect.filterOrFail(
303 (pool) => (pool.state ?? "") !== "DELETING",
304 () => new ResourceNotResolved({ name }),
305 ),
306 Effect.filterOrFail(
307 (pool) => {
308 const state = pool.state ?? "CREATED";
309 return state === "CREATED" || state === "STATE_UNSPECIFIED";
310 },
311 (pool) =>
312 new ResourceNotReady({
313 name,
314 state: pool.state ?? "UNKNOWN",
315 }),
316 ),
317 Effect.retry({
318 while: (error) => error._tag === "GCP.StorageTransfer.ResourceNotReady",
319 times: 10,
320 schedule: Schedule.spaced("1 second"),
321 }),
322 Effect.catchTag("GCP.StorageTransfer.ResourceNotResolved", () =>
323 Effect.succeed(undefined),
324 ),
325 );
326
327export const waitAgentPoolGone = (name: string) =>
328 getAgentPool(name).pipe(
329 Effect.flatMap((pool) =>
330 pool === undefined
331 ? Effect.void
332 : Effect.fail(new ResourceStillExists({ name })),
333 ),
334 Effect.retry({
335 while: (error) =>
336 error._tag === "GCP.StorageTransfer.ResourceStillExists",
337 times: 10,
338 schedule: Schedule.spaced("1 second"),
339 }),
340 Effect.catchTag(
341 "GCP.StorageTransfer.ResourceStillExists",
342 () => Effect.void,
343 ),
344 );
345
346export const listAgentPools = (projectId: string) =>
347 retryApiDisabled(
348 collectPages(
349 storagetransfer.listProjectsAgentPools.pages({
350 projectId,
351 pageSize: 256,
352 }),
353 (page) => page.agentPools,
354 ),
355 ).pipe(
356 Effect.catchTag("NotFound", () =>
357 Effect.succeed([] as storagetransfer.AgentPool[]),
358 ),
359 );
360
361export const listTransferJobs = (projectId: string) =>
362 retryApiDisabled(
363 collectPages(
364 storagetransfer.listTransferJobs.pages({
365 filter: JSON.stringify({ projectId }),
366 pageSize: 256,
367 }),
368 (page) => page.transferJobs,
369 ),
370 ).pipe(
371 Effect.catchTag("NotFound", () =>
372 Effect.succeed([] as storagetransfer.TransferJob[]),
373 ),
374 );