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