1import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { GcpEnvironment } from "../Environment.ts";
9import type { Providers } from "../Providers.ts";
10import {
11 listChildResources,
12 listEntities,
13 listLakes,
14 listPartitions,
15 listZones,
16 parseResourceName,
17 replaceIfChanged,
18 hasAlchemyLabelMap,
19} from "./shared.ts";
20
21export type LakesEntitiesPartitionProps = {
22 /**
23 * Parent entity. Full name
24 * `projects/{project}/locations/{location}/lakes/{lake}/zones/{zone}/entities/{entity}`.
25 * Immutable — changing it replaces the partition.
26 */
27 entity: string;
28 /**
29 * Ordered partition values matching the parent entity's partition
30 * schema. Immutable — changing them replaces the partition.
31 */
32 values: string[];
33 /**
34 * Location of the partition data (`gs://bucket/path/key=value` or a
35 * BigQuery table). Immutable — changing it replaces the partition.
36 */
37 location: string;
38};
39
40export type LakesEntitiesPartition = Resource<
41 "GCP.Dataplex.LakesEntitiesPartition",
42 LakesEntitiesPartitionProps,
43 {
44 /** Full resource name `.../entities/{entity}/partitions/{values}`. */
45 name: string;
46 /** Parent entity resource name. */
47 entity: string;
48 /** Project id. */
49 project: string;
50 /** Region id. */
51 region: string;
52 /** Ordered partition values. */
53 values: string[];
54 /** Data location. */
55 location: string;
56 /** Server etag. */
57 etag: string | undefined;
58 },
59 never,
60 Providers
61>;
62
63/**
64 * A Dataplex metadata partition under a zone entity.
65 *
66 * Partitions have no labels or description field: `read` reports a
67 * partition it finds without prior state as unowned, and `list` returns
68 * the partitions of entities in Alchemy-labeled lakes. Identity is the parent entity plus `values`; changing
69 * `entity`, `values`, or `location` replaces the partition.
70 *
71 * ### Creating a Partition
72 * **Example:** Hive-style date partition
73 * ```typescript
74 * const partition = yield* GCP.Dataplex.LakesEntitiesPartition("Day", {
75 * entity: entity.name,
76 * values: ["2024-01-01"],
77 * location: `gs://${bucket.bucketName}/events/dt=2024-01-01`,
78 * });
79 * ```
80 *
81 * @resource
82 * @category Dataplex
83 */
84export const LakesEntitiesPartition = Resource<LakesEntitiesPartition>(
85 "GCP.Dataplex.LakesEntitiesPartition",
86);
87
88export class LakesEntitiesPartitionNotResolved extends Data.TaggedError(
89 "GCP.Dataplex.LakesEntitiesPartitionNotResolved",
90)<{
91 name: string;
92}> {}
93
94const encodeValue = (value: string) =>
95 encodeURIComponent(encodeURIComponent(value));
96
97const resourceNameOf = (entity: string, values: readonly string[]) =>
98 `${entity}/partitions/${values.map(encodeValue).join("/")}`;
99
100const toAttrs = (
101 partition: dataplex.GoogleCloudDataplexV1Partition,
102 project: string,
103 entity: string,
104) => {
105 const name = partition.name ?? "";
106 const parsed = parseResourceName(name, "partitions");
107 return {
108 name,
109 entity: parsed.parent || entity,
110 project: parsed.project || project,
111 region: parsed.location,
112 values: partition.values ?? [],
113 location: partition.location ?? "",
114 etag: partition.etag,
115 };
116};
117
118const getByName = (name: string) =>
119 dataplex
120 .getProjectsLocationsLakesZonesEntitiesPartitions({ name })
121 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
122
123const listOwnedEntityNames = (project: string, region: string) =>
124 Effect.gen(function* () {
125 const lakes = (yield* listLakes(project, region)).filter((lake) =>
126 hasAlchemyLabelMap(lake.labels),
127 );
128 const zones = yield* listChildResources(lakes, listZones);
129 const named = zones.filter((zone) => (zone.name ?? "").length > 0);
130 const tables = yield* Effect.forEach(
131 named,
132 (zone) => listEntities(zone.name!, "TABLES"),
133 { concurrency: 4 },
134 );
135 const filesets = yield* Effect.forEach(
136 named,
137 (zone) => listEntities(zone.name!, "FILESETS"),
138 { concurrency: 4 },
139 );
140 return [...tables.flat(), ...filesets.flat()]
141 .map((entity) => entity.name)
142 .filter((name): name is string => (name ?? "").length > 0);
143 });
144
145export const LakesEntitiesPartitionProvider = () =>
146 Provider.succeed(LakesEntitiesPartition, {
147 stables: ["name", "entity", "project", "region", "values"],
148
149 diff: Effect.fn(function* ({ news, olds, output }) {
150 if (!isResolved(news)) return undefined;
151 const previousEntity = olds?.entity ?? output?.entity;
152 const nextEntity = news.entity ?? previousEntity;
153 const previousValues = olds?.values ?? output?.values;
154 const nextValues = news.values ?? previousValues;
155 const previousLocation = olds?.location ?? output?.location;
156 const nextLocation = news.location ?? previousLocation;
157 if (
158 replaceIfChanged(previousEntity, nextEntity) ||
159 (previousValues !== undefined &&
160 nextValues !== undefined &&
161 JSON.stringify(previousValues) !== JSON.stringify(nextValues)) ||
162 replaceIfChanged(previousLocation, nextLocation)
163 ) {
164 return {
165 action: "replace" as const,
166 deleteFirst:
167 previousEntity === nextEntity &&
168 previousLocation === nextLocation &&
169 previousValues !== undefined &&
170 nextValues !== undefined &&
171 JSON.stringify(previousValues) === JSON.stringify(nextValues),
172 };
173 }
174 return undefined;
175 }),
176
177 read: Effect.fn(function* ({ olds, output }) {
178 const env = yield* GcpEnvironment.current;
179 const entity = olds?.entity ?? output?.entity ?? "";
180 const values = olds?.values ?? output?.values ?? [];
181 const name =
182 output?.name ?? (entity ? resourceNameOf(entity, values) : "");
183 if (name.length === 0) return undefined;
184 const existing = yield* getByName(name);
185 if (existing === undefined) return undefined;
186 const attrs = toAttrs(existing, env.project, entity);
187 // No labels field: without prior state the partition may not be ours.
188 return output !== undefined ? attrs : Unowned(attrs);
189 }),
190
191 list: () =>
192 Effect.gen(function* () {
193 const env = yield* GcpEnvironment.current;
194 const entities = yield* listOwnedEntityNames(env.project, env.region);
195 const groups = yield* Effect.forEach(
196 entities,
197 (entity) =>
198 listPartitions(entity).pipe(
199 Effect.map((partitions) =>
200 partitions.map((partition) =>
201 toAttrs(partition, env.project, entity),
202 ),
203 ),
204 ),
205 { concurrency: 4 },
206 );
207 return groups.flat();
208 }),
209
210 reconcile: Effect.fn(function* ({ news, output }) {
211 const env = yield* GcpEnvironment.current;
212 const entity = news.entity;
213 const name = output?.name ?? resourceNameOf(entity, news.values);
214 let current = yield* getByName(name);
215
216 if (current === undefined) {
217 const created = yield* dataplex
218 .createProjectsLocationsLakesZonesEntitiesPartitions({
219 parent: entity,
220 body: {
221 values: news.values,
222 location: news.location,
223 },
224 })
225 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
226 current = created ?? undefined;
227 if (current === undefined) {
228 current = yield* getByName(name);
229 }
230 }
231
232 if (current === undefined) {
233 return yield* new LakesEntitiesPartitionNotResolved({ name });
234 }
235
236 return toAttrs(current, env.project, entity);
237 }),
238
239 delete: Effect.fn(function* ({ output }) {
240 yield* dataplex
241 .deleteProjectsLocationsLakesZonesEntitiesPartitions({
242 name: output.name,
243 etag: output.etag,
244 })
245 .pipe(Effect.catchTag("NotFound", () => Effect.void));
246 }),
247 });