1import * as file from "@distilled.cloud/gcp/file_v1";
2import * as Data from "effect/Data";
3import type { GcpOpContext } from "@distilled.cloud/gcp/Protocol";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import { tagRecord } from "../../Tags.ts";
9import { stripInternalLabels } from "../Labels.ts";
10import { waitForOperation as waitForLongRunningOperation } from "../Operation.ts";
11
12export const DEFAULT_ZONE = "us-central1-a";
13export const DEFAULT_SHARE_NAME = "share1";
14export const MAX_NAME_LENGTH = 63;
15
16export class ResourceNotResolved extends Data.TaggedError(
17 "GCP.Filestore.ResourceNotResolved",
18)<{
19 name: string;
20}> {}
21
22export class ResourceStillExists extends Data.TaggedError(
23 "GCP.Filestore.ResourceStillExists",
24)<{
25 name: string;
26}> {}
27
28export class ResourceNotReady extends Data.TaggedError(
29 "GCP.Filestore.ResourceNotReady",
30)<{
31 name: string;
32 state: string;
33}> {}
34
35export class ResourceFailed extends Data.TaggedError(
36 "GCP.Filestore.ResourceFailed",
37)<{
38 name: string;
39 state: string;
40}> {}
41
42export const lastSegment = (value: string | undefined) => {
43 if (value === undefined || value.length === 0) return "";
44 const trimmed = value.replace(/\/+$/, "");
45 const parts = trimmed.split("/");
46 return parts[parts.length - 1] || trimmed;
47};
48
49export const rfc1035 = (name: string, fallback = "filestore"): string => {
50 let next = name
51 .toLowerCase()
52 .replace(/[^a-z0-9-]/g, "-")
53 .replace(/-+/g, "-")
54 .replace(/^-+|-+$/g, "");
55 if (!/^[a-z]/.test(next)) next = `f${next}`;
56 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
57 if (next.length === 0) return fallback;
58 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, MAX_NAME_LENGTH - 1)}0`;
59 return next.slice(0, MAX_NAME_LENGTH);
60};
61
62export const normalizeLocation = (
63 location: string | undefined,
64 fallback: string,
65) => lastSegment(location ?? fallback).toLowerCase();
66
67export const regionOf = (location: string) => {
68 const parts = location.split("-");
69 const last = parts[parts.length - 1];
70 if (last !== undefined && last.length === 1) {
71 return parts.slice(0, -1).join("-");
72 }
73 return location;
74};
75
76export const toPhysicalId = (
77 id: string,
78 explicit: string | undefined,
79 existing: string | undefined,
80 fallback = "filestore",
81) =>
82 Effect.gen(function* () {
83 if (explicit !== undefined) return rfc1035(explicit, fallback);
84 if (existing !== undefined) return existing;
85 return rfc1035(
86 yield* createPhysicalName({
87 id,
88 maxLength: MAX_NAME_LENGTH,
89 lowercase: true,
90 }),
91 fallback,
92 );
93 });
94
95export const parseName = (
96 name: string,
97 collection: string,
98 fallbackLocation: string,
99) => {
100 const parts = name.split("/").filter((part) => part.length > 0);
101 const collectionAt = parts.lastIndexOf(collection);
102 const locationsAt = parts.lastIndexOf("locations");
103 const projectsAt = parts.lastIndexOf("projects");
104 return {
105 project:
106 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
107 location:
108 locationsAt >= 0 && parts[locationsAt + 1]
109 ? parts[locationsAt + 1]!
110 : fallbackLocation,
111 id:
112 collectionAt >= 0 && parts[collectionAt + 1]
113 ? parts[collectionAt + 1]!
114 : lastSegment(name),
115 parent:
116 collectionAt > 0
117 ? parts.slice(0, collectionAt).join("/")
118 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
119 };
120};
121
122export const expandParent = (
123 value: string,
124 project: string,
125 location: string,
126 collection: string,
127) => {
128 if (value.includes("/")) return value.replace(/\/+$/, "");
129 return `projects/${project}/locations/${location}/${collection}/${value}`;
130};
131
132export const userLabels = (
133 labels: Record<string, string | undefined> | null | undefined,
134): Record<string, string> => stripInternalLabels(tagRecord(labels));
135
136export const hasAlchemyLabelMap = (
137 labels: Record<string, string | undefined> | null | undefined,
138) => Object.keys(labels ?? {}).some((key) => key.startsWith("alchemy-"));
139
140export const fieldMask = (fields: Array<string | false | undefined>) =>
141 fields
142 .filter((field): field is string => typeof field === "string")
143 .join(",");
144
145export const replaceOnIdentity = (input: {
146 previousId: string | undefined;
147 nextId: string | undefined;
148 previousLocation: string;
149 nextLocation: string;
150 extra?: boolean;
151 previousParent?: string;
152 nextParent?: string;
153}) => {
154 const parentChanged =
155 (input.previousParent ?? "") !== "" &&
156 (input.nextParent ?? "") !== "" &&
157 (input.previousParent ?? "") !== (input.nextParent ?? "");
158 const replace =
159 (input.extra ?? false) ||
160 parentChanged ||
161 (input.previousId !== undefined &&
162 input.nextId !== undefined &&
163 input.nextId !== input.previousId) ||
164 input.previousLocation !== input.nextLocation;
165 if (!replace) return undefined;
166 const samePhysical =
167 input.previousLocation === input.nextLocation &&
168 !parentChanged &&
169 input.previousId !== undefined &&
170 input.nextId === input.previousId;
171 return {
172 action: "replace" as const,
173 deleteFirst: samePhysical,
174 };
175};
176
177const READY_STATES = new Set(["READY"]);
178const FAILED_STATES = new Set(["INVALID", "ERROR"]);
179
180export const isReadyState = (state: string | undefined) =>
181 READY_STATES.has((state ?? "").toUpperCase());
182
183export const isFailedState = (state: string | undefined) =>
184 FAILED_STATES.has((state ?? "").toUpperCase());
185
186export const isDeletingState = (state: string | undefined) =>
187 (state ?? "").toUpperCase() === "DELETING";
188
189/**
190 * Wait on a Filestore long-running operation. Instances, backups, and snapshots take 5–20 minutes.
191 * `ALREADY_EXISTS` (a create race) counts as success; so does `NOT_FOUND`
192 * when `notFoundOk` (deletes). Returns the final operation.
193 */
194export const waitForOperation = (
195 operation: file.Operation,
196 options?: { notFoundOk?: boolean },
197) =>
198 Effect.suspend(() => {
199 let latest = operation;
200 return waitForLongRunningOperation(
201 operation,
202 (name) => {
203 const get = file.getProjectsLocationsOperations({ name }).pipe(
204 Effect.tap((current) =>
205 Effect.sync(() => {
206 latest = current;
207 }),
208 ),
209 );
210 const observe: Effect.Effect<
211 file.Operation,
212 file.GetProjectsLocationsOperationsError,
213 GcpOpContext
214 > =
215 options?.notFoundOk === true
216 ? get.pipe(
217 Effect.catchTag("NotFound", () =>
218 Effect.succeed<file.Operation>({ name, done: true }),
219 ),
220 )
221 : get.pipe(
222 // A just-returned operation can briefly 404 on read.
223 Effect.retry({
224 while: (error) => error._tag === "NotFound",
225 times: 5,
226 schedule: Schedule.exponential("250 millis"),
227 }),
228 );
229 return observe;
230 },
231 { budget: "30 minutes", interval: "10 seconds" },
232 ).pipe(
233 Effect.map(() => latest),
234 // google.rpc.Code ALREADY_EXISTS = 6, NOT_FOUND = 5.
235 Effect.catchTag("GCP.OperationFailed", (error) =>
236 error.code === 6 || (options?.notFoundOk === true && error.code === 5)
237 ? Effect.succeed(latest)
238 : Effect.fail(error),
239 ),
240 );
241 });
242
243export const waitUntilExists = <A, E extends { readonly _tag: string }, R>(
244 get: Effect.Effect<A | undefined, E, R>,
245 name: string,
246): Effect.Effect<NonNullable<A>, E | ResourceNotResolved, R> =>
247 get.pipe(
248 Effect.filterOrFail(
249 (value): value is NonNullable<A> => value != null,
250 () => new ResourceNotResolved({ name }),
251 ),
252 Effect.retry({
253 while: (error) => error._tag === "GCP.Filestore.ResourceNotResolved",
254 times: 8,
255 schedule: Schedule.spaced("1 second"),
256 }),
257 );
258
259export const waitUntilGone = <A, E extends { readonly _tag: string }, R>(
260 get: Effect.Effect<A | undefined, E, R>,
261 name: string,
262): Effect.Effect<void, E | ResourceStillExists, R> =>
263 get.pipe(
264 Effect.filterOrFail(
265 (value) => value === undefined,
266 () => new ResourceStillExists({ name }),
267 ),
268 Effect.retry({
269 while: (error) => error._tag === "GCP.Filestore.ResourceStillExists",
270 // Backup and snapshot deletion takes several minutes.
271 times: 120,
272 schedule: Schedule.spaced("10 seconds"),
273 }),
274 Effect.asVoid,
275 );
276
277export const waitUntilReady = <A, E extends { readonly _tag: string }, R>(
278 get: Effect.Effect<A | undefined, E, R>,
279 name: string,
280 stateOf: (value: NonNullable<A>) => string | undefined,
281): Effect.Effect<
282 NonNullable<A>,
283 E | ResourceNotResolved | ResourceFailed | ResourceNotReady,
284 R
285> =>
286 get.pipe(
287 Effect.filterOrFail(
288 (value): value is NonNullable<A> => value != null,
289 () => new ResourceNotResolved({ name }),
290 ),
291 Effect.filterOrFail(
292 (value) => !isFailedState(stateOf(value)),
293 (value) =>
294 new ResourceFailed({
295 name,
296 state: stateOf(value) ?? "",
297 }),
298 ),
299 Effect.filterOrFail(
300 (value) => {
301 const state = stateOf(value) ?? "";
302 return isReadyState(state) || state.length === 0;
303 },
304 (value) => new ResourceNotReady({ name, state: stateOf(value) ?? "" }),
305 ),
306 Effect.retry({
307 while: (error) =>
308 error._tag === "GCP.Filestore.ResourceNotReady" ||
309 error._tag === "GCP.Filestore.ResourceNotResolved",
310 // Backups and snapshots take several minutes to become READY.
311 times: 120,
312 schedule: Schedule.spaced("10 seconds"),
313 }),
314 );
315
316export const listLabeledPages = <Page, A, E, R>(
317 pages: Stream.Stream<Page, E, R>,
318 items: (page: Page) => readonly A[] | undefined,
319 labelsOf: (item: A) => Record<string, string | undefined> | null | undefined,
320) =>
321 pages.pipe(
322 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
323 Stream.filter((item) => hasAlchemyLabelMap(labelsOf(item))),
324 Stream.runCollect,
325 Effect.map((chunk) => Array.from(chunk)),
326 );