1import * as resourcemanager from "@distilled.cloud/gcp/cloudresourcemanager_v3";
2import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
3import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { tagRecord } from "../../Tags.ts";
12import { GcpEnvironment } from "../Environment.ts";
13import {
14 createInternalLabels,
15 diffLabels,
16 hasAlchemyLabels,
17 toLabels,
18} from "../Labels.ts";
19import type { Providers } from "../Providers.ts";
20import { waitForOperation } from "./operations.ts";
21import {
22 fingerprint,
23 hasAlchemyLabelMap,
24 lastSegment,
25 listMetadataFeeds,
26 locationParent,
27 normalizeLocation,
28 parseResourceName,
29 replaceIfChanged,
30 toPhysicalRfc1035,
31 userLabels,
32} from "./shared.ts";
33
34export type MetadataFeedScope = {
35 /** Listen to every entry in the organization. */
36 organizationLevel?: boolean;
37 /** Projects whose entries are published (`projects/{project}`). */
38 projects?: string[];
39 /** Entry groups whose entries are published. */
40 entryGroups?: string[];
41};
42
43export type MetadataFeedFilters = {
44 /** Entry types to listen to. */
45 entryTypes?: string[];
46 /** Aspect types to listen to. */
47 aspectTypes?: string[];
48 /** Change types (`CREATE`, `UPDATE`, `DELETE`). */
49 changeTypes?: Array<
50 | dataplex.GoogleCloudDataplexV1MetadataFeedFiltersChangeTypesItemEnum
51 | (string & {})
52 >;
53};
54
55export type MetadataFeedProps = {
56 /**
57 * Feed id (the `{metadataFeed}` segment of
58 * `projects/{project}/locations/{location}/metadataFeeds/{metadataFeed}`).
59 * If omitted, a unique name is generated. Immutable — changing it
60 * replaces the feed.
61 */
62 metadataFeedId?: string;
63 /**
64 * Region. Immutable — changing it replaces the feed.
65 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
66 */
67 location?: string;
68 /**
69 * Scope of catalog changes to publish. Defaults to the current project.
70 */
71 scope?: MetadataFeedScope;
72 /**
73 * Filters on entry types, aspect types, and change types.
74 */
75 filters?: MetadataFeedFilters;
76 /**
77 * Pub/Sub topic that receives feed messages
78 * (`projects/{project}/topics/{topic}`). Grant the Dataplex service
79 * account `roles/pubsub.publisher` on the topic.
80 */
81 pubsubTopic?: string;
82 /**
83 * User labels. Alchemy ownership labels are merged in automatically.
84 */
85 labels?: Record<string, string>;
86};
87
88export type MetadataFeed = Resource<
89 "GCP.Dataplex.MetadataFeed",
90 MetadataFeedProps,
91 {
92 /** Full resource name `.../metadataFeeds/{metadataFeed}`. */
93 name: string;
94 /** Feed id. */
95 metadataFeedId: string;
96 /** Project id. */
97 project: string;
98 /** Location id. */
99 location: string;
100 /** Organization-level scope. */
101 organizationLevel: boolean;
102 /** Scoped projects. */
103 projects: string[];
104 /** Scoped entry groups. */
105 entryGroups: string[];
106 /** Destination Pub/Sub topic. */
107 pubsubTopic: string | undefined;
108 /** User labels (Alchemy ownership labels stripped). */
109 labels: Record<string, string>;
110 /** Server-assigned uid. */
111 uid: string | undefined;
112 /** RFC3339 creation timestamp. */
113 createTime: string | undefined;
114 /** RFC3339 last-update timestamp. */
115 updateTime: string | undefined;
116 },
117 never,
118 Providers
119>;
120
121/**
122 * A Dataplex metadata feed that publishes catalog changes to Pub/Sub.
123 *
124 * Changing `metadataFeedId` or `location` replaces the feed. Scope,
125 * filters, destination topic, and labels update in place.
126 *
127 * ### Creating a Feed
128 * **Example:** Project-scoped feed
129 * ```typescript
130 * const feed = yield* GCP.Dataplex.MetadataFeed("Catalog", {
131 * pubsubTopic: topic.name,
132 * scope: { projects: [`projects/${project}`] },
133 * labels: { env: "dev" },
134 * });
135 * ```
136 *
137 * **Example:** Filtered create events
138 * ```typescript
139 * const feed = yield* GCP.Dataplex.MetadataFeed("Catalog", {
140 * pubsubTopic: topic.name,
141 * filters: { changeTypes: ["CREATE"] },
142 * });
143 * ```
144 *
145 * @resource
146 * @category Dataplex
147 */
148export const MetadataFeed = Resource<MetadataFeed>("GCP.Dataplex.MetadataFeed");
149
150export class MetadataFeedNotResolved extends Data.TaggedError(
151 "GCP.Dataplex.MetadataFeedNotResolved",
152)<{
153 name: string;
154}> {}
155
156export class MetadataFeedStillExists extends Data.TaggedError(
157 "GCP.Dataplex.MetadataFeedStillExists",
158)<{
159 name: string;
160}> {}
161
162const resourceName = (project: string, location: string, feedId: string) =>
163 `${locationParent(project, location)}/metadataFeeds/${feedId}`;
164
165const scopeBody = (
166 project: string,
167 scope: MetadataFeedScope | undefined,
168): dataplex.GoogleCloudDataplexV1MetadataFeedScope => ({
169 organizationLevel: scope?.organizationLevel === true ? true : undefined,
170 projects:
171 scope?.projects ??
172 (scope?.organizationLevel === true || (scope?.entryGroups?.length ?? 0) > 0
173 ? undefined
174 : [`projects/${project}`]),
175 entryGroups: scope?.entryGroups,
176});
177
178const toAttrs = (
179 feed: dataplex.GoogleCloudDataplexV1MetadataFeed,
180 project: string,
181) => {
182 const name = feed.name ?? "";
183 const parsed = parseResourceName(name, "metadataFeeds");
184 return {
185 name,
186 metadataFeedId: parsed.id,
187 project: parsed.project || project,
188 location: parsed.location,
189 organizationLevel: feed.scope?.organizationLevel === true,
190 projects: feed.scope?.projects ?? [],
191 entryGroups: feed.scope?.entryGroups ?? [],
192 pubsubTopic: feed.pubsubTopic,
193 labels: userLabels(feed.labels),
194 uid: feed.uid,
195 createTime: feed.createTime,
196 updateTime: feed.updateTime,
197 };
198};
199
200const getByName = (name: string) =>
201 dataplex
202 .getProjectsLocationsMetadataFeeds({ name })
203 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
204
205const topicResource = (topic: string, project: string) =>
206 topic.includes("/") ? topic : `projects/${project}/topics/${topic}`;
207
208const grantDataplexPubsub = (project: string, topic: string | undefined) =>
209 Effect.gen(function* () {
210 if (topic === undefined || topic.length === 0) return;
211 const resource = topicResource(topic, project);
212 const projectResource = yield* resourcemanager
213 .getProjects({ name: `projects/${project}` })
214 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
215 const projectNumber = lastSegment(projectResource?.name ?? "");
216 if (projectNumber.length === 0) return;
217 const member = `serviceAccount:service-${projectNumber}@gcp-sa-dataplex.iam.gserviceaccount.com`;
218 const policy = yield* pubsub.getIamPolicyProjectsTopics({ resource });
219 const bindings = (policy.bindings ?? []).map((binding) => ({
220 ...binding,
221 members: [...(binding.members ?? [])],
222 }));
223 for (const role of ["roles/pubsub.publisher", "roles/pubsub.viewer"]) {
224 const binding = bindings.find((item) => item.role === role);
225 if (binding?.members?.includes(member)) continue;
226 if (binding) {
227 binding.members = [...(binding.members ?? []), member];
228 } else {
229 bindings.push({ role, members: [member] });
230 }
231 }
232 yield* pubsub.setIamPolicyProjectsTopics({
233 resource,
234 body: { policy: { ...policy, bindings } },
235 });
236 }).pipe(
237 Effect.retry({
238 while: (error) =>
239 error._tag === "Conflict" || error._tag === "TooManyRequests",
240 times: 5,
241 schedule: Schedule.spaced("2 seconds"),
242 }),
243 );
244
245const waitUntilExists = (name: string) =>
246 getByName(name).pipe(
247 Effect.filterOrFail(
248 (feed): feed is dataplex.GoogleCloudDataplexV1MetadataFeed =>
249 feed !== undefined,
250 () => new MetadataFeedNotResolved({ name }),
251 ),
252 Effect.retry({
253 while: (error) => error._tag === "GCP.Dataplex.MetadataFeedNotResolved",
254 times: 8,
255 schedule: Schedule.spaced("2 seconds"),
256 }),
257 );
258
259const waitUntilGone = (name: string) =>
260 getByName(name).pipe(
261 Effect.flatMap((feed) =>
262 feed === undefined
263 ? Effect.void
264 : Effect.fail(new MetadataFeedStillExists({ name })),
265 ),
266 Effect.retry({
267 while: (error) => error._tag === "GCP.Dataplex.MetadataFeedStillExists",
268 times: 10,
269 schedule: Schedule.spaced("3 seconds"),
270 }),
271 );
272
273export const MetadataFeedProvider = () =>
274 Provider.succeed(MetadataFeed, {
275 stables: [
276 "name",
277 "metadataFeedId",
278 "project",
279 "location",
280 "uid",
281 "createTime",
282 ],
283
284 diff: Effect.fn(function* ({ news, olds, output }) {
285 if (!isResolved(news)) return undefined;
286 const env = yield* GcpEnvironment.current;
287 const previousId = olds?.metadataFeedId ?? output?.metadataFeedId;
288 const nextId = news.metadataFeedId ?? previousId;
289 const previousLocation = normalizeLocation(
290 olds?.location ?? output?.location,
291 env.region,
292 );
293 const nextLocation = normalizeLocation(
294 news.location ?? olds?.location ?? output?.location,
295 env.region,
296 );
297 if (
298 replaceIfChanged(previousId, nextId) ||
299 (output !== undefined && previousLocation !== nextLocation)
300 ) {
301 return {
302 action: "replace" as const,
303 deleteFirst:
304 previousLocation === nextLocation &&
305 previousId !== undefined &&
306 nextId === previousId,
307 };
308 }
309 return undefined;
310 }),
311
312 read: Effect.fn(function* ({ id, olds, output }) {
313 const env = yield* GcpEnvironment.current;
314 const location = normalizeLocation(
315 olds?.location ?? output?.location,
316 env.region,
317 );
318 const feedId = yield* toPhysicalRfc1035(
319 id,
320 olds?.metadataFeedId,
321 output?.metadataFeedId,
322 );
323 const name = output?.name ?? resourceName(env.project, location, feedId);
324 const existing = yield* getByName(name);
325 if (existing === undefined) return undefined;
326 const attrs = toAttrs(existing, env.project);
327 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
328 ? attrs
329 : Unowned(attrs);
330 }),
331
332 list: () =>
333 Effect.gen(function* () {
334 const env = yield* GcpEnvironment.current;
335 const feeds = yield* listMetadataFeeds(env.project, env.region);
336 return feeds
337 .filter((feed) => hasAlchemyLabelMap(feed.labels))
338 .map((feed) => toAttrs(feed, env.project));
339 }),
340
341 reconcile: Effect.fn(function* ({ id, news, output }) {
342 const env = yield* GcpEnvironment.current;
343 const location = normalizeLocation(
344 news.location ?? output?.location,
345 env.region,
346 );
347 const feedId = yield* toPhysicalRfc1035(
348 id,
349 news.metadataFeedId,
350 output?.metadataFeedId,
351 );
352 const name = output?.name ?? resourceName(env.project, location, feedId);
353 const desiredLabels = {
354 ...toLabels(news.labels),
355 ...(yield* createInternalLabels(id)),
356 };
357 const scope = scopeBody(env.project, news.scope);
358 const filters = news.filters;
359 yield* grantDataplexPubsub(env.project, news.pubsubTopic);
360
361 let current = yield* getByName(name);
362
363 if (current === undefined) {
364 const created = yield* dataplex
365 .createProjectsLocationsMetadataFeeds({
366 parent: locationParent(env.project, location),
367 metadataFeedId: feedId,
368 body: {
369 scope,
370 filters,
371 labels: desiredLabels,
372 pubsubTopic: news.pubsubTopic,
373 },
374 })
375 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
376 if (created !== undefined) {
377 yield* waitForOperation(created);
378 }
379 current = yield* waitUntilExists(name);
380 }
381
382 if (current === undefined) {
383 return yield* new MetadataFeedNotResolved({ name });
384 }
385
386 const observedLabels = tagRecord(current.labels);
387 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
388 const labelsChanged = upsert.length > 0 || removed.length > 0;
389 const scopeChanged = fingerprint(current.scope) !== fingerprint(scope);
390 const filtersChanged =
391 fingerprint(current.filters) !== fingerprint(filters);
392 const topicChanged =
393 (current.pubsubTopic ?? "") !== (news.pubsubTopic ?? "");
394
395 if (labelsChanged || scopeChanged || filtersChanged || topicChanged) {
396 const updateMask = [
397 labelsChanged ? "labels" : undefined,
398 scopeChanged ? "scope" : undefined,
399 filtersChanged ? "filters" : undefined,
400 topicChanged ? "pubsub_topic" : undefined,
401 ].filter((field): field is string => field !== undefined);
402 const operation = yield* dataplex.patchProjectsLocationsMetadataFeeds({
403 name: current.name ?? name,
404 updateMask: updateMask.join(","),
405 body: {
406 scope,
407 filters,
408 labels: desiredLabels,
409 pubsubTopic: news.pubsubTopic,
410 },
411 });
412 yield* waitForOperation(operation);
413 current = yield* waitUntilExists(current.name ?? name);
414 }
415
416 return toAttrs(current, env.project);
417 }),
418
419 delete: Effect.fn(function* ({ output }) {
420 const operation = yield* dataplex
421 .deleteProjectsLocationsMetadataFeeds({ name: output.name })
422 .pipe(
423 Effect.retry({
424 while: (error) => error._tag === "Conflict",
425 times: 8,
426 schedule: Schedule.spaced("2 seconds"),
427 }),
428 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
429 );
430 if (operation !== undefined) {
431 yield* waitForOperation(operation, { notFoundOk: true });
432 }
433 yield* waitUntilGone(output.name);
434 }),
435 });