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 expandParent,
22 fingerprint,
23 hasAlchemyLabelMap,
24 isPendingState,
25 listChildResources,
26 listLakes,
27 listZones,
28 normalizeLocation,
29 parseResourceName,
30 replaceIfChanged,
31 toPhysicalRfc1035,
32 userLabels,
33} from "./shared.ts";
34
35const DEFAULT_TYPE = "RAW";
36const DEFAULT_LOCATION_TYPE = "SINGLE_REGION";
37
38export type ZoneDiscoveryCsvOptions = {
39 /** Header rows to skip. */
40 headerRows?: number;
41 /** Field delimiter. @default "," */
42 delimiter?: string;
43 /** Character encoding. @default "UTF-8" */
44 encoding?: string;
45 /** Register all columns as strings. */
46 disableTypeInference?: boolean;
47};
48
49export type ZoneDiscoveryJsonOptions = {
50 /** Character encoding. @default "UTF-8" */
51 encoding?: string;
52 /** Disable type inference. */
53 disableTypeInference?: boolean;
54};
55
56export type ZoneDiscoverySpec = {
57 /**
58 * Whether discovery is enabled.
59 * @default false
60 */
61 enabled?: boolean;
62 /** Include glob / table-name patterns. */
63 includePatterns?: string[];
64 /** Exclude glob / table-name patterns. */
65 excludePatterns?: string[];
66 /** CSV discovery options. */
67 csvOptions?: ZoneDiscoveryCsvOptions;
68 /** JSON discovery options. */
69 jsonOptions?: ZoneDiscoveryJsonOptions;
70 /** Cron schedule (at least 60 minutes apart). */
71 schedule?: string;
72};
73
74export type LakesZoneProps = {
75 /**
76 * Parent lake. Full name
77 * `projects/{project}/locations/{location}/lakes/{lake}` or the lake id
78 * (combined with `location`). Immutable — changing it replaces the zone.
79 */
80 lake: string;
81 /**
82 * Region used when `lake` is a bare id.
83 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
84 */
85 location?: string;
86 /**
87 * Zone id. Must contain only lowercase letters, numbers, and hyphens;
88 * start with a letter; end with a letter or number; and be 1-63
89 * characters. Unique across lakes in the project. Immutable — changing
90 * it replaces the zone.
91 */
92 zoneId?: string;
93 /**
94 * Zone type. Immutable — changing it replaces the zone.
95 * @default "RAW"
96 */
97 type?: dataplex.GoogleCloudDataplexV1ZoneTypeEnum | (string & {});
98 /**
99 * Location type of attached assets. Immutable — changing it replaces
100 * the zone.
101 * @default "SINGLE_REGION"
102 */
103 locationType?:
104 | dataplex.GoogleCloudDataplexV1ZoneResourceSpecLocationTypeEnum
105 | (string & {});
106 /**
107 * User-friendly display name.
108 */
109 displayName?: string;
110 /**
111 * Human-readable description.
112 */
113 description?: string;
114 /**
115 * User labels. Alchemy ownership labels are merged in automatically.
116 */
117 labels?: Record<string, string>;
118 /**
119 * Discovery settings applied to data in this zone.
120 */
121 discoverySpec?: ZoneDiscoverySpec;
122};
123
124export type LakesZone = Resource<
125 "GCP.Dataplex.LakesZone",
126 LakesZoneProps,
127 {
128 /** Full resource name `.../lakes/{lake}/zones/{zone}`. */
129 name: string;
130 /** Zone id. */
131 zoneId: string;
132 /** Parent lake resource name. */
133 lake: string;
134 /** Project id. */
135 project: string;
136 /** Location id. */
137 location: string;
138 /** Zone type (`RAW` or `CURATED`). */
139 type: string;
140 /** Attached-asset location type. */
141 locationType: string;
142 /** User-friendly display name. */
143 displayName: string | undefined;
144 /** Description. */
145 description: string | undefined;
146 /** User labels (Alchemy ownership labels stripped). */
147 labels: Record<string, string>;
148 /** Whether discovery is enabled. */
149 discoveryEnabled: boolean;
150 /** Lifecycle state. */
151 state: string | undefined;
152 /** Server-assigned uid. */
153 uid: string | undefined;
154 /** RFC3339 creation timestamp. */
155 createTime: string | undefined;
156 /** RFC3339 last-update timestamp. */
157 updateTime: string | undefined;
158 },
159 never,
160 Providers
161>;
162
163/**
164 * A Dataplex zone — a logical grouping of assets inside a lake (`RAW` or
165 * `CURATED`).
166 *
167 * Changing `lake`, `zoneId`, `location`, `type`, or `locationType`
168 * replaces the zone. Display name, description, labels, and discovery
169 * spec update in place.
170 *
171 * ### Creating a Zone
172 * **Example:** RAW zone in a lake
173 * ```typescript
174 * const zone = yield* GCP.Dataplex.LakesZone("Landing", {
175 * lake: lake.name,
176 * type: "RAW",
177 * labels: { env: "dev" },
178 * });
179 * ```
180 *
181 * **Example:** Named zone with discovery disabled
182 * ```typescript
183 * const zone = yield* GCP.Dataplex.LakesZone("Landing", {
184 * lake: lake.name,
185 * zoneId: "landing-raw",
186 * discoverySpec: { enabled: false },
187 * });
188 * ```
189 *
190 * @resource
191 * @category Dataplex
192 */
193export const LakesZone = Resource<LakesZone>("GCP.Dataplex.LakesZone");
194
195export class LakesZoneNotResolved extends Data.TaggedError(
196 "GCP.Dataplex.LakesZoneNotResolved",
197)<{
198 name: string;
199}> {}
200
201export class LakesZoneStillExists extends Data.TaggedError(
202 "GCP.Dataplex.LakesZoneStillExists",
203)<{
204 name: string;
205}> {}
206
207const lakeOf = (lake: string, project: string, location: string) =>
208 expandParent(lake, project, location, "lakes");
209
210const resourceName = (lake: string, zoneId: string) =>
211 `${lake}/zones/${zoneId}`;
212
213const desiredType = (type: string | undefined) =>
214 (type ?? DEFAULT_TYPE).toUpperCase();
215
216const desiredLocationType = (value: string | undefined) =>
217 (value ?? DEFAULT_LOCATION_TYPE).toUpperCase();
218
219const discoveryBody = (
220 spec: ZoneDiscoverySpec | undefined,
221): dataplex.GoogleCloudDataplexV1ZoneDiscoverySpec => ({
222 enabled: spec?.enabled === true,
223 includePatterns: spec?.includePatterns,
224 excludePatterns: spec?.excludePatterns,
225 csvOptions: spec?.csvOptions,
226 jsonOptions: spec?.jsonOptions,
227 schedule: spec?.schedule,
228});
229
230const toAttrs = (zone: dataplex.GoogleCloudDataplexV1Zone, project: string) => {
231 const name = zone.name ?? "";
232 const parsed = parseResourceName(name, "zones");
233 return {
234 name,
235 zoneId: parsed.id,
236 lake: parsed.parent,
237 project: parsed.project || project,
238 location: parsed.location,
239 type: zone.type ?? DEFAULT_TYPE,
240 locationType: zone.resourceSpec?.locationType ?? DEFAULT_LOCATION_TYPE,
241 displayName: zone.displayName,
242 description: zone.description,
243 labels: userLabels(zone.labels),
244 discoveryEnabled: zone.discoverySpec?.enabled === true,
245 state: zone.state,
246 uid: zone.uid,
247 createTime: zone.createTime,
248 updateTime: zone.updateTime,
249 };
250};
251
252const getByName = (name: string) =>
253 name.length === 0 || name.includes("//")
254 ? Effect.succeed(undefined)
255 : dataplex
256 .getProjectsLocationsLakesZones({ name })
257 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
258
259const waitUntilReady = (name: string) =>
260 getByName(name).pipe(
261 Effect.filterOrFail(
262 (zone): zone is dataplex.GoogleCloudDataplexV1Zone => zone !== undefined,
263 () => new LakesZoneNotResolved({ name }),
264 ),
265 Effect.filterOrFail(
266 (zone) => !isPendingState(zone.state),
267 () => new DataplexNotResolved({ name }),
268 ),
269 Effect.retry({
270 while: (error) =>
271 error._tag === "GCP.Dataplex.LakesZoneNotResolved" ||
272 error._tag === "GCP.Dataplex.NotResolved" ||
273 error._tag === "TooManyRequests",
274 // Lakes provision a metastore-backed control plane for 2-5 minutes.
275 times: 36,
276 schedule: Schedule.spaced("10 seconds"),
277 }),
278 );
279
280const waitUntilGone = (name: string) =>
281 getByName(name).pipe(
282 Effect.flatMap((zone) =>
283 zone === undefined
284 ? Effect.void
285 : Effect.fail(new LakesZoneStillExists({ name })),
286 ),
287 Effect.retry({
288 while: (error) => error._tag === "GCP.Dataplex.LakesZoneStillExists",
289 times: 10,
290 schedule: Schedule.spaced("3 seconds"),
291 }),
292 );
293
294export const LakesZoneProvider = () =>
295 Provider.succeed(LakesZone, {
296 stables: [
297 "name",
298 "zoneId",
299 "lake",
300 "project",
301 "location",
302 "type",
303 "locationType",
304 "uid",
305 "createTime",
306 ],
307
308 diff: Effect.fn(function* ({ news, olds, output }) {
309 if (!isResolved(news)) return undefined;
310 const env = yield* GcpEnvironment.current;
311 const previousId = olds?.zoneId ?? output?.zoneId;
312 const nextId = news.zoneId ?? previousId;
313 const previousLake = olds?.lake ?? output?.lake;
314 const nextLake = news.lake ?? previousLake;
315 const previousLocation = normalizeLocation(
316 olds?.location ?? output?.location,
317 env.region,
318 );
319 const nextLocation = normalizeLocation(
320 news.location ?? olds?.location ?? output?.location,
321 env.region,
322 );
323 const previousType = desiredType(olds?.type ?? output?.type);
324 const nextType = desiredType(news.type ?? previousType);
325 const previousLocationType = desiredLocationType(
326 olds?.locationType ?? output?.locationType,
327 );
328 const nextLocationType = desiredLocationType(
329 news.locationType ?? previousLocationType,
330 );
331 if (
332 replaceIfChanged(previousId, nextId) ||
333 replaceIfChanged(previousLake, nextLake) ||
334 (output !== undefined && previousLocation !== nextLocation) ||
335 previousType !== nextType ||
336 previousLocationType !== nextLocationType
337 ) {
338 return {
339 action: "replace" as const,
340 deleteFirst:
341 previousLake === nextLake &&
342 previousLocation === nextLocation &&
343 previousId !== undefined &&
344 nextId === previousId,
345 };
346 }
347 return undefined;
348 }),
349
350 read: Effect.fn(function* ({ id, olds, output }) {
351 const env = yield* GcpEnvironment.current;
352 const location = normalizeLocation(
353 olds?.location ?? output?.location,
354 env.region,
355 );
356 const lake = lakeOf(
357 olds?.lake ?? output?.lake ?? "",
358 env.project,
359 location,
360 );
361 const zoneId = yield* toPhysicalRfc1035(id, olds?.zoneId, output?.zoneId);
362 const name = output?.name ?? resourceName(lake, zoneId);
363 const existing = yield* getByName(name);
364 if (existing === undefined) return undefined;
365 const attrs = toAttrs(existing, env.project);
366 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
367 ? attrs
368 : Unowned(attrs);
369 }),
370
371 list: () =>
372 Effect.gen(function* () {
373 const env = yield* GcpEnvironment.current;
374 const lakes = yield* listLakes(env.project, env.region);
375 const zones = yield* listChildResources(lakes, listZones);
376 return zones
377 .filter((zone) => hasAlchemyLabelMap(zone.labels))
378 .map((zone) => toAttrs(zone, env.project));
379 }),
380
381 reconcile: Effect.fn(function* ({ id, news, output }) {
382 const env = yield* GcpEnvironment.current;
383 const location = normalizeLocation(
384 news.location ?? output?.location,
385 env.region,
386 );
387 const lake = lakeOf(news.lake, env.project, location);
388 const zoneId = yield* toPhysicalRfc1035(id, news.zoneId, output?.zoneId);
389 const name = output?.name ?? resourceName(lake, zoneId);
390 const type = desiredType(news.type);
391 const locationType = desiredLocationType(news.locationType);
392 const desiredLabels = {
393 ...toLabels(news.labels),
394 ...(yield* createInternalLabels(id)),
395 };
396 const desiredDiscovery = discoveryBody(news.discoverySpec);
397
398 let current = yield* getByName(name);
399
400 if (current === undefined) {
401 const created = yield* dataplex
402 .createProjectsLocationsLakesZones({
403 parent: lake,
404 zoneId,
405 body: {
406 type,
407 resourceSpec: { locationType },
408 displayName: news.displayName,
409 description: news.description,
410 labels: desiredLabels,
411 discoverySpec: desiredDiscovery,
412 },
413 })
414 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
415 if (created !== undefined) {
416 yield* waitForOperation(created);
417 }
418 current = yield* waitUntilReady(name);
419 }
420
421 if (current === undefined) {
422 return yield* new LakesZoneNotResolved({ name });
423 }
424
425 const observedLabels = tagRecord(current.labels);
426 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
427 const labelsChanged = upsert.length > 0 || removed.length > 0;
428 const displayNameChanged =
429 (current.displayName ?? "") !== (news.displayName ?? "");
430 const descriptionChanged =
431 (current.description ?? "") !== (news.description ?? "");
432 const discoveryChanged =
433 fingerprint(current.discoverySpec) !== fingerprint(desiredDiscovery);
434
435 if (
436 labelsChanged ||
437 displayNameChanged ||
438 descriptionChanged ||
439 discoveryChanged
440 ) {
441 const updateMask = [
442 labelsChanged ? "labels" : undefined,
443 displayNameChanged ? "display_name" : undefined,
444 descriptionChanged ? "description" : undefined,
445 discoveryChanged ? "discovery_spec" : undefined,
446 ].filter((field): field is string => field !== undefined);
447 const operation = yield* dataplex.patchProjectsLocationsLakesZones({
448 name: current.name ?? name,
449 updateMask: updateMask.join(","),
450 body: {
451 displayName: news.displayName,
452 description: news.description,
453 labels: desiredLabels,
454 discoverySpec: desiredDiscovery,
455 },
456 });
457 yield* waitForOperation(operation);
458 current = yield* waitUntilReady(current.name ?? name);
459 }
460
461 return toAttrs(current, env.project);
462 }),
463
464 delete: Effect.fn(function* ({ output }) {
465 if (!output.name || output.name.includes("//")) return;
466 const operation = yield* dataplex
467 .deleteProjectsLocationsLakesZones({ name: output.name })
468 .pipe(
469 Effect.retry({
470 while: (error) => error._tag === "Conflict",
471 times: 8,
472 schedule: Schedule.spaced("2 seconds"),
473 }),
474 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
475 Effect.catchTag("BadRequest", () => Effect.succeed(undefined)),
476 );
477 if (operation !== undefined) {
478 yield* waitForOperation(operation, { notFoundOk: true });
479 }
480 yield* waitUntilGone(output.name);
481 }),
482 });