1import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { tagRecord } from "../../Tags.ts";
10import { GcpEnvironment } from "../Environment.ts";
11import {
12 createInternalLabels,
13 diffLabels,
14 hasAlchemyLabels,
15 toLabels,
16} from "../Labels.ts";
17import type { Providers } from "../Providers.ts";
18import { waitForOperation } from "./operations.ts";
19import {
20 DataplexNotResolved,
21 hasAlchemyLabelMap,
22 isPendingState,
23 listLakes,
24 locationParent,
25 normalizeLocation,
26 parseResourceName,
27 replaceIfChanged,
28 toPhysicalRfc1035,
29 userLabels,
30} from "./shared.ts";
31
32export type LakeMetastore = {
33 /** Dataproc Metastore service `projects/{project}/locations/{location}/services/{service}`. */
34 service?: string;
35};
36
37export type LakeProps = {
38 /**
39 * Lake id (the `{lake}` segment of
40 * `projects/{project}/locations/{location}/lakes/{lake}`). If omitted, a
41 * unique name is generated from the stack, stage, and logical id. Must
42 * contain only lowercase letters, numbers, and hyphens; start with a
43 * letter; end with a letter or number; and be 1-63 characters. Immutable
44 * — changing it replaces the lake.
45 */
46 lakeId?: string;
47 /**
48 * Region (`us-central1`, …). Immutable — changing it replaces the lake.
49 * `US-CENTRAL1` is accepted and normalized to `us-central1`.
50 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
51 */
52 location?: string;
53 /**
54 * User-friendly display name.
55 */
56 displayName?: string;
57 /**
58 * Human-readable description.
59 */
60 description?: string;
61 /**
62 * User labels. Alchemy ownership labels are merged in automatically.
63 */
64 labels?: Record<string, string>;
65 /**
66 * Optional Dataproc Metastore association.
67 */
68 metastore?: LakeMetastore;
69};
70
71export type Lake = Resource<
72 "GCP.Dataplex.Lake",
73 LakeProps,
74 {
75 /** Full resource name `projects/{project}/locations/{location}/lakes/{lake}`. */
76 name: string;
77 /** Lake id (last path segment). */
78 lakeId: string;
79 /** Project id. */
80 project: string;
81 /** Location id. */
82 location: string;
83 /** User-friendly display name. */
84 displayName: string | undefined;
85 /** Description. */
86 description: string | undefined;
87 /** User labels (Alchemy ownership labels stripped). */
88 labels: Record<string, string>;
89 /** Lifecycle state (`ACTIVE`, `CREATING`, …). */
90 state: string | undefined;
91 /** Lake service account. */
92 serviceAccount: string | undefined;
93 /** Associated Dataproc Metastore service, if any. */
94 metastoreService: string | undefined;
95 /** Server-assigned uid. */
96 uid: string | undefined;
97 /** RFC3339 creation timestamp. */
98 createTime: string | undefined;
99 /** RFC3339 last-update timestamp. */
100 updateTime: string | undefined;
101 },
102 never,
103 Providers
104>;
105
106/**
107 * A Dataplex lake — a regional container for zones, assets, and tasks.
108 *
109 * Changing `lakeId` or `location` replaces the lake. Display name,
110 * description, labels, and metastore association update in place.
111 *
112 * ### Creating a Lake
113 * **Example:** Generated name
114 * ```typescript
115 * const lake = yield* GCP.Dataplex.Lake("Warehouse", {
116 * labels: { env: "dev" },
117 * });
118 * ```
119 *
120 * **Example:** Named lake with a description
121 * ```typescript
122 * const lake = yield* GCP.Dataplex.Lake("Warehouse", {
123 * lakeId: "analytics-lake",
124 * location: "us-central1",
125 * displayName: "Analytics",
126 * description: "curated warehouse",
127 * labels: { env: "prod" },
128 * });
129 * ```
130 *
131 * @resource
132 * @category Dataplex
133 */
134export const Lake = Resource<Lake>("GCP.Dataplex.Lake");
135
136export class LakeNotResolved extends Data.TaggedError(
137 "GCP.Dataplex.LakeNotResolved",
138)<{
139 name: string;
140}> {}
141
142export class LakeStillExists extends Data.TaggedError(
143 "GCP.Dataplex.LakeStillExists",
144)<{
145 name: string;
146}> {}
147
148const resourceName = (project: string, location: string, lakeId: string) =>
149 `${locationParent(project, location)}/lakes/${lakeId}`;
150
151const toAttrs = (lake: dataplex.GoogleCloudDataplexV1Lake, project: string) => {
152 const name = lake.name ?? "";
153 const parsed = parseResourceName(name, "lakes");
154 return {
155 name,
156 lakeId: parsed.id,
157 project: parsed.project || project,
158 location: parsed.location,
159 displayName: lake.displayName,
160 description: lake.description,
161 labels: userLabels(lake.labels),
162 state: lake.state,
163 serviceAccount: lake.serviceAccount,
164 metastoreService: lake.metastore?.service,
165 uid: lake.uid,
166 createTime: lake.createTime,
167 updateTime: lake.updateTime,
168 };
169};
170
171const getByName = (name: string) =>
172 dataplex
173 .getProjectsLocationsLakes({ name })
174 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
175
176// Lakes provision a metastore-backed control plane for 2-5 minutes.
177const waitUntilReady = (name: string) =>
178 Effect.gen(function* () {
179 const lake = yield* getByName(name);
180 if (lake === undefined) {
181 return yield* new LakeNotResolved({ name });
182 }
183 if (isPendingState(lake.state)) {
184 return yield* new DataplexNotResolved({ name });
185 }
186 return lake;
187 }).pipe(
188 Effect.retry({
189 while: (error) =>
190 error._tag === "GCP.Dataplex.LakeNotResolved" ||
191 error._tag === "GCP.Dataplex.NotResolved" ||
192 error._tag === "TooManyRequests",
193 times: 36,
194 schedule: Schedule.spaced("10 seconds"),
195 }),
196 );
197
198const waitUntilGone = (name: string) =>
199 getByName(name).pipe(
200 Effect.flatMap((lake) =>
201 lake === undefined
202 ? Effect.void
203 : Effect.fail(new LakeStillExists({ name })),
204 ),
205 Effect.retry({
206 while: (error) => error._tag === "GCP.Dataplex.LakeStillExists",
207 times: 10,
208 schedule: Schedule.spaced("3 seconds"),
209 }),
210 );
211
212export const LakeProvider = () =>
213 Provider.succeed(Lake, {
214 stables: ["name", "lakeId", "project", "location", "uid", "createTime"],
215
216 diff: Effect.fn(function* ({ news, olds, output }) {
217 if (!isResolved(news)) return undefined;
218 const env = yield* GcpEnvironment.current;
219 const previousId = olds?.lakeId ?? output?.lakeId;
220 const nextId = news.lakeId ?? previousId;
221 const previousLocation = normalizeLocation(
222 olds?.location ?? output?.location,
223 env.region,
224 );
225 const nextLocation = normalizeLocation(
226 news.location ?? olds?.location ?? output?.location,
227 env.region,
228 );
229 if (
230 replaceIfChanged(previousId, nextId) ||
231 (output !== undefined && previousLocation !== nextLocation)
232 ) {
233 return {
234 action: "replace" as const,
235 deleteFirst:
236 previousLocation === nextLocation &&
237 previousId !== undefined &&
238 nextId === previousId,
239 };
240 }
241 return undefined;
242 }),
243
244 read: Effect.fn(function* ({ id, olds, output }) {
245 const env = yield* GcpEnvironment.current;
246 const location = normalizeLocation(
247 olds?.location ?? output?.location,
248 env.region,
249 );
250 const lakeId = yield* toPhysicalRfc1035(id, olds?.lakeId, output?.lakeId);
251 const name = output?.name ?? resourceName(env.project, location, lakeId);
252 const existing = yield* getByName(name);
253 if (existing === undefined) return undefined;
254 const attrs = toAttrs(existing, env.project);
255 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
256 ? attrs
257 : Unowned(attrs);
258 }),
259
260 list: () =>
261 Effect.gen(function* () {
262 const env = yield* GcpEnvironment.current;
263 const lakes = yield* listLakes(env.project, env.region);
264 return lakes
265 .filter((lake) => hasAlchemyLabelMap(lake.labels))
266 .map((lake) => toAttrs(lake, env.project));
267 }),
268
269 reconcile: Effect.fn(function* ({ id, news, output }) {
270 const env = yield* GcpEnvironment.current;
271 const location = normalizeLocation(
272 news.location ?? output?.location,
273 env.region,
274 );
275 const lakeId = yield* toPhysicalRfc1035(id, news.lakeId, output?.lakeId);
276 const name = output?.name ?? resourceName(env.project, location, lakeId);
277 const desiredLabels = {
278 ...toLabels(news.labels),
279 ...(yield* createInternalLabels(id)),
280 };
281 const desiredMetastore = news.metastore?.service;
282
283 let current = yield* getByName(name);
284
285 if (current === undefined) {
286 const created = yield* dataplex
287 .createProjectsLocationsLakes({
288 parent: locationParent(env.project, location),
289 lakeId,
290 body: {
291 displayName: news.displayName,
292 description: news.description,
293 labels: desiredLabels,
294 metastore: desiredMetastore
295 ? { service: desiredMetastore }
296 : undefined,
297 },
298 })
299 .pipe(
300 Effect.retry({
301 while: (error) => error._tag === "TooManyRequests",
302 times: 8,
303 schedule: Schedule.spaced("2 seconds"),
304 }),
305 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
306 );
307 if (created !== undefined) {
308 yield* waitForOperation(created);
309 }
310 current = yield* waitUntilReady(name);
311 }
312
313 if (current === undefined) {
314 return yield* new LakeNotResolved({ name });
315 }
316
317 const observedLabels = tagRecord(current.labels);
318 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
319 const labelsChanged = upsert.length > 0 || removed.length > 0;
320 const displayNameChanged =
321 (current.displayName ?? "") !== (news.displayName ?? "");
322 const descriptionChanged =
323 (current.description ?? "") !== (news.description ?? "");
324 const metastoreChanged =
325 (current.metastore?.service ?? "") !== (desiredMetastore ?? "");
326
327 if (
328 labelsChanged ||
329 displayNameChanged ||
330 descriptionChanged ||
331 metastoreChanged
332 ) {
333 const updateMask = [
334 labelsChanged ? "labels" : undefined,
335 displayNameChanged ? "display_name" : undefined,
336 descriptionChanged ? "description" : undefined,
337 metastoreChanged ? "metastore.service" : undefined,
338 ].filter((field): field is string => field !== undefined);
339 const operation = yield* dataplex.patchProjectsLocationsLakes({
340 name: current.name ?? name,
341 updateMask: updateMask.join(","),
342 body: {
343 displayName: news.displayName,
344 description: news.description,
345 labels: desiredLabels,
346 metastore: desiredMetastore
347 ? { service: desiredMetastore }
348 : undefined,
349 },
350 });
351 yield* waitForOperation(operation);
352 current = yield* waitUntilReady(current.name ?? name);
353 }
354
355 return toAttrs(current, env.project);
356 }),
357
358 delete: Effect.fn(function* ({ output }) {
359 if (!output.name) return;
360 const operation = yield* dataplex
361 .deleteProjectsLocationsLakes({ name: output.name })
362 .pipe(
363 Effect.retry({
364 while: (error) => error._tag === "Conflict",
365 times: 8,
366 schedule: Schedule.spaced("2 seconds"),
367 }),
368 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
369 );
370 if (operation !== undefined) {
371 yield* waitForOperation(operation, { notFoundOk: true });
372 }
373 yield* waitUntilGone(output.name);
374 }),
375 });