1import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { tagRecord } from "../../Tags.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import {
11 createInternalLabels,
12 diffLabels,
13 hasAlchemyLabels,
14 toLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17import {
18 DataplexNotResolved,
19 collectPages,
20 fingerprint,
21 hasAlchemyLabelMap,
22 listAtLocation,
23 normalizeLocation,
24 parentOf,
25 parseName,
26 replaceOnIdentity,
27 retryQuota,
28 toPhysicalId,
29 userLabels,
30 waitForOperation,
31 waitUntilExists,
32 waitUntilGone,
33} from "./internal.ts";
34
35export type DataScanDataSource = dataplex.GoogleCloudDataplexV1DataSource;
36export type DataScanExecutionSpec =
37 dataplex.GoogleCloudDataplexV1DataScanExecutionSpec;
38export type DataScanExecutionIdentity =
39 dataplex.GoogleCloudDataplexV1ExecutionIdentity;
40export type DataQualitySpec = dataplex.GoogleCloudDataplexV1DataQualitySpec;
41export type DataProfileSpec = dataplex.GoogleCloudDataplexV1DataProfileSpec;
42export type DataDiscoverySpec = dataplex.GoogleCloudDataplexV1DataDiscoverySpec;
43export type DataDocumentationSpec =
44 dataplex.GoogleCloudDataplexV1DataDocumentationSpec;
45export type UnstructuredDataProfileSpec =
46 dataplex.GoogleCloudDataplexV1UnstructuredDataProfileSpec;
47
48export type DataScanProps = {
49 /**
50 * DataScan id. If omitted, a unique name is generated. Must contain
51 * only lowercase letters, numbers and hyphens, start with a letter,
52 * end with a letter or number, and be 1-63 characters. Immutable —
53 * changing it replaces the scan.
54 */
55 dataScanId?: string;
56 /**
57 * Region (`us-central1`, …). Immutable — changing it replaces the
58 * scan.
59 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
60 */
61 location?: string;
62 /**
63 * Description (1-1024 characters).
64 */
65 description?: string;
66 /**
67 * User-friendly display name (1-256 characters).
68 */
69 displayName?: string;
70 /**
71 * User labels. Alchemy ownership labels are merged in automatically.
72 */
73 labels?: Record<string, string>;
74 /**
75 * Immutable data source (Dataplex entity or BigQuery / GCS resource).
76 * Changing it replaces the scan.
77 */
78 data: DataScanDataSource;
79 /**
80 * Execution settings. Defaults to on-demand.
81 */
82 executionSpec?: DataScanExecutionSpec;
83 /**
84 * Immutable identity used to run the scan. Changing it replaces the
85 * scan.
86 */
87 executionIdentity?: DataScanExecutionIdentity;
88 /**
89 * Data quality scan settings. Mutually exclusive with the other spec
90 * fields.
91 */
92 dataQualitySpec?: DataQualitySpec;
93 /**
94 * Data profile scan settings.
95 */
96 dataProfileSpec?: DataProfileSpec;
97 /**
98 * Data discovery scan settings.
99 */
100 dataDiscoverySpec?: DataDiscoverySpec;
101 /**
102 * Data documentation scan settings.
103 */
104 dataDocumentationSpec?: DataDocumentationSpec;
105 /**
106 * Unstructured data profile scan settings.
107 */
108 unstructuredDataProfileSpec?: UnstructuredDataProfileSpec;
109};
110
111export type DataScan = Resource<
112 "GCP.Dataplex.DataScan",
113 DataScanProps,
114 {
115 /** Full resource name. */
116 name: string;
117 /** DataScan id (last path segment). */
118 dataScanId: string;
119 /** Project id. */
120 project: string;
121 /** Location id. */
122 location: string;
123 /** Description. */
124 description: string | undefined;
125 /** Display name. */
126 displayName: string | undefined;
127 /** User labels (Alchemy ownership labels stripped). */
128 labels: Record<string, string>;
129 /** Current state. */
130 state: string | undefined;
131 /** Data source. */
132 data: DataScanDataSource | undefined;
133 /** Execution spec. */
134 executionSpec: DataScanExecutionSpec | undefined;
135 /** Scan type. */
136 type: string | undefined;
137 /** Data quality spec, if any. */
138 dataQualitySpec: DataQualitySpec | undefined;
139 /** Data profile spec, if any. */
140 dataProfileSpec: DataProfileSpec | undefined;
141 /** Data discovery spec, if any. */
142 dataDiscoverySpec: DataDiscoverySpec | undefined;
143 /** Data documentation spec, if any. */
144 dataDocumentationSpec: DataDocumentationSpec | undefined;
145 /** Unstructured data profile spec, if any. */
146 unstructuredDataProfileSpec: UnstructuredDataProfileSpec | undefined;
147 /** System uid. */
148 uid: string | undefined;
149 /** RFC3339 creation timestamp. */
150 createTime: string | undefined;
151 /** RFC3339 last-update timestamp. */
152 updateTime: string | undefined;
153 },
154 never,
155 Providers
156>;
157
158/**
159 * A Dataplex DataScan that profiles, quality-checks, discovers, or
160 * documents a BigQuery table, dataset, or Cloud Storage bucket.
161 *
162 * Location, scan id, data source, and execution identity are immutable.
163 * Description, display name, labels, execution spec, and scan specs
164 * update in place.
165 *
166 * ### Creating a DataScan
167 * **Example:** On-demand profile scan of a BigQuery table
168 * ```typescript
169 * const scan = yield* GCP.Dataplex.DataScan("OrdersProfile", {
170 * displayName: "orders profile",
171 * labels: { env: "test" },
172 * data: {
173 * resource:
174 * "//bigquery.googleapis.com/projects/my-project/datasets/sales/tables/orders",
175 * },
176 * dataProfileSpec: {},
177 * });
178 * ```
179 *
180 * @resource
181 * @category Dataplex
182 */
183export const DataScan = Resource<DataScan>("GCP.Dataplex.DataScan");
184
185const resourceName = (project: string, location: string, dataScanId: string) =>
186 `projects/${project}/locations/${location}/dataScans/${dataScanId}`;
187
188const toAttrs = (
189 scan: dataplex.GoogleCloudDataplexV1DataScan,
190 project: string,
191) => {
192 const name = scan.name ?? "";
193 const parsed = parseName(name, "dataScans");
194 return {
195 name,
196 dataScanId: parsed.id,
197 project: parsed.project || project,
198 location: parsed.location,
199 description: scan.description,
200 displayName: scan.displayName,
201 labels: userLabels(scan.labels),
202 state: scan.state,
203 data: scan.data,
204 executionSpec: scan.executionSpec,
205 type: scan.type,
206 dataQualitySpec: scan.dataQualitySpec,
207 dataProfileSpec: scan.dataProfileSpec,
208 dataDiscoverySpec: scan.dataDiscoverySpec,
209 dataDocumentationSpec: scan.dataDocumentationSpec,
210 unstructuredDataProfileSpec: scan.unstructuredDataProfileSpec,
211 uid: scan.uid,
212 createTime: scan.createTime,
213 updateTime: scan.updateTime,
214 };
215};
216
217const getByName = (name: string) =>
218 retryQuota(
219 dataplex.getProjectsLocationsDataScans({ name, view: "FULL" }),
220 ).pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
221
222const listScans = (project: string, region: string) => {
223 const collect = (parent: string) =>
224 collectPages(
225 dataplex.listProjectsLocationsDataScans.pages({
226 parent,
227 pageSize: 1000,
228 }),
229 (page) => page.dataScans,
230 ).pipe(
231 Effect.map((items) =>
232 items.filter((item) => hasAlchemyLabelMap(item.labels)),
233 ),
234 Effect.catchTag("NotFound", () => Effect.succeed([])),
235 );
236 return listAtLocation(project, region, collect);
237};
238
239const dataSourceOf = (data: DataScanDataSource | undefined) => ({
240 entity: data?.entity ?? "",
241 resource: data?.resource ?? "",
242});
243
244export const DataScanProvider = () =>
245 Provider.succeed(DataScan, {
246 stables: ["name", "dataScanId", "project", "location", "uid", "createTime"],
247
248 diff: Effect.fn(function* ({ news, olds, output }) {
249 if (!isResolved(news)) return undefined;
250 const env = yield* GcpEnvironment.current;
251 return replaceOnIdentity({
252 previousId: olds?.dataScanId ?? output?.dataScanId,
253 nextId: news.dataScanId ?? olds?.dataScanId ?? output?.dataScanId,
254 previousLocation: normalizeLocation(
255 olds?.location ?? output?.location,
256 env.region,
257 ),
258 nextLocation: normalizeLocation(
259 news.location ?? olds?.location ?? output?.location,
260 env.region,
261 ),
262 extra:
263 fingerprint(dataSourceOf(news.data)) !==
264 fingerprint(dataSourceOf(olds?.data ?? output?.data)) ||
265 fingerprint(news.executionIdentity ?? olds?.executionIdentity) !==
266 fingerprint(olds?.executionIdentity),
267 });
268 }),
269
270 read: Effect.fn(function* ({ id, olds, output }) {
271 const env = yield* GcpEnvironment.current;
272 const dataScanId = yield* toPhysicalId(
273 id,
274 olds?.dataScanId,
275 output?.dataScanId,
276 "datascan",
277 );
278 const location = normalizeLocation(
279 olds?.location ?? output?.location,
280 env.region,
281 );
282 const name =
283 output?.name ?? resourceName(env.project, location, dataScanId);
284 const existing = yield* getByName(name);
285 if (existing === undefined) return undefined;
286 const attrs = toAttrs(existing, env.project);
287 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
288 ? attrs
289 : Unowned(attrs);
290 }),
291
292 list: () =>
293 Effect.gen(function* () {
294 const env = yield* GcpEnvironment.current;
295 const items = yield* listScans(env.project, env.region);
296 return items.map((item) => toAttrs(item, env.project));
297 }),
298
299 reconcile: Effect.fn(function* ({ id, news, output }) {
300 const env = yield* GcpEnvironment.current;
301 const dataScanId = yield* toPhysicalId(
302 id,
303 news.dataScanId,
304 output?.dataScanId,
305 "datascan",
306 );
307 const location = normalizeLocation(
308 news.location ?? output?.location,
309 env.region,
310 );
311 const name = resourceName(env.project, location, dataScanId);
312 const desiredLabels = {
313 ...toLabels(news.labels),
314 ...(yield* createInternalLabels(id)),
315 };
316
317 let current = yield* getByName(output?.name ?? name);
318
319 if (current === undefined) {
320 const created = yield* retryQuota(
321 dataplex.createProjectsLocationsDataScans({
322 parent: parentOf(env.project, location),
323 dataScanId,
324 body: {
325 description: news.description,
326 displayName: news.displayName,
327 labels: desiredLabels,
328 data: news.data,
329 executionSpec: news.executionSpec,
330 executionIdentity: news.executionIdentity,
331 dataQualitySpec: news.dataQualitySpec,
332 dataProfileSpec: news.dataProfileSpec,
333 dataDiscoverySpec: news.dataDiscoverySpec,
334 dataDocumentationSpec: news.dataDocumentationSpec,
335 unstructuredDataProfileSpec: news.unstructuredDataProfileSpec,
336 },
337 }),
338 ).pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
339 if (created !== undefined) {
340 yield* waitForOperation(created);
341 }
342 current = yield* waitUntilExists(getByName(name), name);
343 }
344
345 if (current === undefined) {
346 return yield* new DataplexNotResolved({ name });
347 }
348
349 const observedLabels = tagRecord(current.labels);
350 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
351 const labelsChanged = upsert.length > 0 || removed.length > 0;
352 const descriptionChanged =
353 (current.description ?? "") !== (news.description ?? "");
354 const displayNameChanged =
355 (current.displayName ?? "") !== (news.displayName ?? "");
356 const executionChanged =
357 fingerprint(current.executionSpec) !== fingerprint(news.executionSpec);
358 const qualityChanged =
359 fingerprint(current.dataQualitySpec) !==
360 fingerprint(news.dataQualitySpec);
361 const profileChanged =
362 fingerprint(current.dataProfileSpec) !==
363 fingerprint(news.dataProfileSpec);
364 const discoveryChanged =
365 fingerprint(current.dataDiscoverySpec) !==
366 fingerprint(news.dataDiscoverySpec);
367 const documentationChanged =
368 fingerprint(current.dataDocumentationSpec) !==
369 fingerprint(news.dataDocumentationSpec);
370 const unstructuredChanged =
371 fingerprint(current.unstructuredDataProfileSpec) !==
372 fingerprint(news.unstructuredDataProfileSpec);
373
374 if (
375 labelsChanged ||
376 descriptionChanged ||
377 displayNameChanged ||
378 executionChanged ||
379 qualityChanged ||
380 profileChanged ||
381 discoveryChanged ||
382 documentationChanged ||
383 unstructuredChanged
384 ) {
385 const operation = yield* retryQuota(
386 dataplex.patchProjectsLocationsDataScans({
387 name: current.name ?? name,
388 updateMask: [
389 labelsChanged ? "labels" : undefined,
390 descriptionChanged ? "description" : undefined,
391 displayNameChanged ? "displayName" : undefined,
392 executionChanged ? "executionSpec" : undefined,
393 qualityChanged ? "dataQualitySpec" : undefined,
394 profileChanged ? "dataProfileSpec" : undefined,
395 discoveryChanged ? "dataDiscoverySpec" : undefined,
396 documentationChanged ? "dataDocumentationSpec" : undefined,
397 unstructuredChanged ? "unstructuredDataProfileSpec" : undefined,
398 ]
399 .filter((field): field is string => field !== undefined)
400 .join(","),
401 body: {
402 name: current.name ?? name,
403 description: news.description,
404 displayName: news.displayName,
405 labels: desiredLabels,
406 executionSpec: news.executionSpec,
407 dataQualitySpec: news.dataQualitySpec,
408 dataProfileSpec: news.dataProfileSpec,
409 dataDiscoverySpec: news.dataDiscoverySpec,
410 dataDocumentationSpec: news.dataDocumentationSpec,
411 unstructuredDataProfileSpec: news.unstructuredDataProfileSpec,
412 },
413 }),
414 );
415 yield* waitForOperation(operation);
416 current = yield* waitUntilExists(
417 getByName(current.name ?? name),
418 current.name ?? name,
419 );
420 }
421
422 return toAttrs(current, env.project);
423 }),
424
425 delete: Effect.fn(function* ({ output }) {
426 const operation = yield* retryQuota(
427 dataplex.deleteProjectsLocationsDataScans({
428 name: output.name,
429 force: true,
430 }),
431 ).pipe(
432 Effect.retry({
433 while: (error) => error._tag === "Conflict",
434 times: 8,
435 schedule: Schedule.spaced("2 seconds"),
436 }),
437 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
438 );
439 if (operation !== undefined) {
440 yield* waitForOperation(operation, { notFoundOk: true });
441 }
442 yield* waitUntilGone(getByName(output.name), output.name);
443 }),
444 });