1import * as eventarc from "@distilled.cloud/gcp/eventarc_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 changedFields,
19 compact,
20 collectPages,
21 expandResource,
22 hasAlchemyLabelKeys,
23 normalizeLocation,
24 parentOf,
25 parseName,
26 resourceName,
27 rfc1035,
28 sameJson,
29 textKey,
30 toPhysicalId,
31 userAnnotations,
32 userLabels,
33 retryOnTransient,
34 waitForOperation,
35 waitUntilGone,
36 waitUntilPresent,
37} from "./internal.ts";
38
39const COLLECTION = "enrollments";
40
41export type EnrollmentProps = {
42 /**
43 * Enrollment id (the `{enrollment}` segment of
44 * `projects/{project}/locations/{location}/enrollments/{enrollment}`).
45 * If omitted, a unique name is generated. Must match
46 * `[a-z]([a-z0-9-]*[a-z0-9])?` and be 1-63 characters. Immutable —
47 * changing it replaces the enrollment.
48 */
49 enrollmentId?: string;
50 /**
51 * Eventarc Advanced location (`us-central1`, `us-east4`, …). Immutable
52 * — changing it replaces the enrollment. `US-CENTRAL1` is accepted
53 * and normalized to `us-central1`.
54 * @default the stack's GCP region (`GCP.Region`, else the profile
55 * region, else `us-central1`)
56 */
57 location?: string;
58 /**
59 * MessageBus this enrollment reads from, as
60 * `projects/{project}/locations/{location}/messageBuses/{messageBus}`
61 * or the `{messageBus}` segment. Immutable — changing it replaces the
62 * enrollment.
63 */
64 messageBus: string;
65 /**
66 * Pipeline that receives matched messages, as
67 * `projects/{project}/locations/{location}/pipelines/{pipeline}` or
68 * the `{pipeline}` segment.
69 */
70 destination: string;
71 /**
72 * CEL expression identifying which messages this enrollment applies
73 * to, e.g. `message.type == 'google.cloud.pubsub.topic.v1.messagePublished'`
74 * or `true` to match every message.
75 */
76 celMatch: string;
77 /**
78 * Human-readable display name.
79 */
80 displayName?: string;
81 /**
82 * User labels. Alchemy ownership labels are merged in automatically.
83 */
84 labels?: Record<string, string>;
85 /**
86 * Free-form annotations.
87 */
88 annotations?: Record<string, string>;
89};
90
91export type Enrollment = Resource<
92 "GCP.Eventarc.Enrollment",
93 EnrollmentProps,
94 {
95 /** Full resource name `projects/{project}/locations/{location}/enrollments/{enrollment}`. */
96 name: string;
97 /** Enrollment id (last path segment). */
98 enrollmentId: string;
99 /** Project id. */
100 project: string;
101 /** Location id (`us-central1`, …). */
102 location: string;
103 /** Source MessageBus resource name. */
104 messageBus: string | undefined;
105 /** Destination Pipeline resource name. */
106 destination: string | undefined;
107 /** CEL match expression. */
108 celMatch: string | undefined;
109 /** Display name. */
110 displayName: string | undefined;
111 /** User labels (Alchemy ownership labels stripped). */
112 labels: Record<string, string>;
113 /** User annotations. */
114 annotations: Record<string, string>;
115 /** Server-assigned UUID4, stable until delete. */
116 uid: string | undefined;
117 /** Server checksum of the resource. */
118 etag: string | undefined;
119 /** RFC3339 creation timestamp. */
120 createTime: string | undefined;
121 /** RFC3339 last-update timestamp. */
122 updateTime: string | undefined;
123 },
124 never,
125 Providers
126>;
127
128/**
129 * An Eventarc Advanced enrollment that subscribes to messages on a
130 * MessageBus and delivers matches to a Pipeline.
131 *
132 * `enrollmentId`, `location`, and `messageBus` are identity — changing
133 * any of them replaces the enrollment. Destination, CEL match, display
134 * name, labels, and annotations update in place.
135 *
136 * ### Creating an Enrollment
137 * **Example:** Match every message
138 * ```typescript
139 * const bus = yield* GCP.Eventarc.MessageBus("Events", {
140 * location: "us-central1",
141 * });
142 * const pipeline = yield* GCP.Eventarc.Pipeline("Sink", {
143 * location: "us-central1",
144 * destinations: [{ topic: topic.name }],
145 * });
146 * const enrollment = yield* GCP.Eventarc.Enrollment("All", {
147 * location: "us-central1",
148 * messageBus: bus.name,
149 * destination: pipeline.name,
150 * celMatch: "true",
151 * labels: { env: "prod" },
152 * });
153 * ```
154 *
155 * ### Updating an Enrollment
156 * Re-declare the same logical id with changed props; the engine keeps the
157 * physical enrollment and patches it in place.
158 *
159 * **Example:** Tighten the CEL match
160 * ```typescript
161 * const enrollment = yield* GCP.Eventarc.Enrollment("All", {
162 * location: "us-central1",
163 * messageBus: bus.name,
164 * destination: pipeline.name,
165 * celMatch: "message.type == 'google.cloud.pubsub.topic.v1.messagePublished'",
166 * labels: { env: "prod", role: "enrollment" },
167 * });
168 * ```
169 *
170 * @resource
171 * @category Eventarc
172 */
173export const Enrollment = Resource<Enrollment>("GCP.Eventarc.Enrollment");
174
175const toAttrs = (
176 enrollment: eventarc.Enrollment,
177 project: string,
178 region: string,
179) => {
180 const name = enrollment.name ?? "";
181 const parsed = parseName(name, COLLECTION, region);
182 return {
183 name,
184 enrollmentId: parsed.id,
185 project: parsed.project || project,
186 location: parsed.location,
187 messageBus: enrollment.messageBus,
188 destination: enrollment.destination,
189 celMatch: enrollment.celMatch,
190 displayName: enrollment.displayName,
191 labels: userLabels(enrollment.labels),
192 annotations: userAnnotations(enrollment.annotations),
193 uid: enrollment.uid,
194 etag: enrollment.etag,
195 createTime: enrollment.createTime,
196 updateTime: enrollment.updateTime,
197 };
198};
199
200const getByName = (name: string) =>
201 eventarc
202 .getProjectsLocationsEnrollments({ name })
203 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
204
205export const EnrollmentProvider = () =>
206 Provider.succeed(Enrollment, {
207 stables: [
208 "name",
209 "enrollmentId",
210 "project",
211 "location",
212 "messageBus",
213 "uid",
214 "createTime",
215 ],
216
217 diff: Effect.fn(function* ({ news, olds, output }) {
218 if (!isResolved(news)) return undefined;
219 const env = yield* GcpEnvironment.current;
220 const previousId = olds?.enrollmentId ?? output?.enrollmentId;
221 const nextId = news.enrollmentId
222 ? rfc1035(news.enrollmentId, "enrollment")
223 : previousId;
224 const previousLocation = normalizeLocation(
225 olds?.location ?? output?.location,
226 env.region,
227 );
228 const nextLocation = normalizeLocation(
229 news.location ?? olds?.location ?? output?.location,
230 env.region,
231 );
232 const previousBus = olds?.messageBus ?? output?.messageBus ?? "";
233 const nextBus = news.messageBus;
234 if (
235 (previousId !== undefined &&
236 nextId !== undefined &&
237 nextId !== previousId) ||
238 previousLocation !== nextLocation ||
239 (previousBus.length > 0 && previousBus !== nextBus)
240 ) {
241 return {
242 action: "replace" as const,
243 deleteFirst:
244 previousId !== undefined &&
245 nextId !== undefined &&
246 previousId === nextId &&
247 previousLocation === nextLocation,
248 };
249 }
250 return undefined;
251 }),
252
253 read: Effect.fn(function* ({ id, olds, output }) {
254 const env = yield* GcpEnvironment.current;
255 const enrollmentId = yield* toPhysicalId(
256 id,
257 olds?.enrollmentId,
258 output?.enrollmentId,
259 "enrollment",
260 );
261 const location = normalizeLocation(
262 olds?.location ?? output?.location,
263 env.region,
264 );
265 const name =
266 output?.name ??
267 resourceName(env.project, location, COLLECTION, enrollmentId);
268 const existing = yield* getByName(name);
269 if (existing === undefined) return undefined;
270 const attrs = toAttrs(existing, env.project, env.region);
271 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
272 ? attrs
273 : Unowned(attrs);
274 }),
275
276 list: () =>
277 Effect.gen(function* () {
278 const env = yield* GcpEnvironment.current;
279 const items = yield* collectPages(
280 eventarc.listProjectsLocationsEnrollments.pages({
281 parent: parentOf(env.project, "-"),
282 pageSize: 1000,
283 }),
284 (page) => page.enrollments,
285 );
286 return items
287 .filter((item) => hasAlchemyLabelKeys(item.labels))
288 .map((item) => toAttrs(item, env.project, env.region));
289 }),
290
291 reconcile: Effect.fn(function* ({ id, news, output }) {
292 const env = yield* GcpEnvironment.current;
293 const enrollmentId = yield* toPhysicalId(
294 id,
295 news.enrollmentId,
296 output?.enrollmentId,
297 "enrollment",
298 );
299 const location = normalizeLocation(
300 news.location ?? output?.location,
301 env.region,
302 );
303 const name = resourceName(
304 env.project,
305 location,
306 COLLECTION,
307 enrollmentId,
308 );
309 const desiredLabels = {
310 ...toLabels(news.labels),
311 ...(yield* createInternalLabels(id)),
312 };
313 const messageBus = expandResource(
314 news.messageBus,
315 env.project,
316 location,
317 "messageBuses",
318 );
319 const destination = expandResource(
320 news.destination,
321 env.project,
322 location,
323 "pipelines",
324 );
325 const desiredAnnotations = news.annotations
326 ? tagRecord(news.annotations)
327 : undefined;
328
329 let current = yield* getByName(name);
330
331 if (current === undefined) {
332 const created = yield* eventarc
333 .createProjectsLocationsEnrollments({
334 parent: parentOf(env.project, location),
335 enrollmentId,
336 body: compact({
337 name,
338 messageBus,
339 destination,
340 celMatch: news.celMatch,
341 displayName: news.displayName,
342 labels: desiredLabels,
343 annotations: desiredAnnotations,
344 }),
345 })
346 .pipe(
347 Effect.retry({
348 while: (error) => error._tag === "Conflict",
349 times: 5,
350 schedule: Schedule.spaced("2 seconds"),
351 }),
352 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
353 );
354 if (created !== undefined) {
355 yield* waitForOperation(created);
356 }
357 current = yield* waitUntilPresent(getByName(name), name);
358 }
359
360 const observedLabels = tagRecord(current.labels);
361 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
362 const updateMask = changedFields([
363 ["labels", upsert.length > 0 || removed.length > 0],
364 ["destination", textKey(current.destination) !== destination],
365 ["celMatch", textKey(current.celMatch) !== news.celMatch],
366 [
367 "displayName",
368 textKey(current.displayName) !== textKey(news.displayName),
369 ],
370 [
371 "annotations",
372 !sameJson(tagRecord(current.annotations), desiredAnnotations ?? {}),
373 ],
374 ]);
375
376 if (updateMask.length > 0) {
377 const patched = yield* eventarc.patchProjectsLocationsEnrollments({
378 name: current.name ?? name,
379 updateMask: updateMask.join(","),
380 body: {
381 name: current.name ?? name,
382 destination,
383 celMatch: news.celMatch,
384 displayName: news.displayName,
385 labels: desiredLabels,
386 annotations: desiredAnnotations ?? {},
387 },
388 });
389 yield* waitForOperation(patched);
390 current = yield* waitUntilPresent(
391 getByName(current.name ?? name),
392 current.name ?? name,
393 );
394 }
395
396 return toAttrs(current, env.project, env.region);
397 }),
398
399 delete: Effect.fn(function* ({ output }) {
400 yield* retryOnTransient(
401 Effect.gen(function* () {
402 const existing = yield* getByName(output.name);
403 if (existing === undefined) return;
404 const operation = yield* eventarc
405 .deleteProjectsLocationsEnrollments({
406 name: output.name,
407 allowMissing: true,
408 })
409 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
410 if (operation !== undefined) {
411 yield* waitForOperation(operation, { notFoundOk: true });
412 }
413 }),
414 ).pipe(
415 Effect.catchTag("GCP.OperationFailed", (error) =>
416 getByName(output.name).pipe(
417 Effect.flatMap((current) =>
418 current === undefined ? Effect.void : Effect.fail(error),
419 ),
420 ),
421 ),
422 );
423 yield* waitUntilGone(getByName(output.name), output.name);
424 }),
425 });