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 type LoggingConfig,
19 changedFields,
20 compact,
21 cryptoKeyKey,
22 collectPages,
23 expandResource,
24 hasAlchemyLabelKeys,
25 loggingKey,
26 normalizeLocation,
27 parentOf,
28 parseName,
29 resourceName,
30 rfc1035,
31 sameJson,
32 textKey,
33 toLoggingConfig,
34 toPhysicalId,
35 userAnnotations,
36 userLabels,
37 retryOnTransient,
38 waitForOperation,
39 waitUntilGone,
40 waitUntilPresent,
41} from "./internal.ts";
42
43const COLLECTION = "googleApiSources";
44
45export type ProjectSubscriptions = {
46 /**
47 * Projects to receive events from, as `projects/{id}` or
48 * `projects/{number}`. All projects must share an organization. At
49 * most 100.
50 */
51 list?: string[];
52};
53
54export type OrganizationSubscription = {
55 /**
56 * When true, subscribe to events from every project in the source's
57 * organization.
58 */
59 enabled?: boolean;
60};
61
62export type GoogleApiSourceProps = {
63 /**
64 * Google API source id (the `{googleApiSource}` segment of
65 * `projects/{project}/locations/{location}/googleApiSources/{googleApiSource}`).
66 * If omitted, a unique name is generated. Must match
67 * `[a-z]([a-z0-9-]*[a-z0-9])?` and be 1-63 characters. Immutable —
68 * changing it replaces the source.
69 */
70 googleApiSourceId?: string;
71 /**
72 * Eventarc Advanced location (`us-central1`, `us-east4`, …). Immutable
73 * — changing it replaces the source. `US-CENTRAL1` is accepted and
74 * normalized to `us-central1`. One source is allowed per project per
75 * region.
76 * @default the stack's GCP region (`GCP.Region`, else the profile
77 * region, else `us-central1`)
78 */
79 location?: string;
80 /**
81 * MessageBus that receives Google API events, as
82 * `projects/{project}/locations/{location}/messageBuses/{messageBus}`
83 * or the `{messageBus}` segment.
84 */
85 destination: string;
86 /**
87 * Human-readable display name.
88 */
89 displayName?: string;
90 /**
91 * Customer-managed Cloud KMS key used to encrypt/decrypt event data,
92 * as
93 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
94 */
95 cryptoKeyName?: string;
96 /**
97 * Platform logging configuration.
98 */
99 loggingConfig?: LoggingConfig;
100 /**
101 * Subscribe to events from every project in the organization.
102 */
103 organizationSubscription?: OrganizationSubscription;
104 /**
105 * Subscribe to events from an explicit list of projects in the same
106 * organization.
107 */
108 projectSubscriptions?: ProjectSubscriptions;
109 /**
110 * User labels. Alchemy ownership labels are merged in automatically.
111 */
112 labels?: Record<string, string>;
113 /**
114 * Free-form annotations.
115 */
116 annotations?: Record<string, string>;
117};
118
119export type GoogleApiSource = Resource<
120 "GCP.Eventarc.GoogleApiSource",
121 GoogleApiSourceProps,
122 {
123 /** Full resource name `projects/{project}/locations/{location}/googleApiSources/{googleApiSource}`. */
124 name: string;
125 /** Google API source id (last path segment). */
126 googleApiSourceId: string;
127 /** Project id. */
128 project: string;
129 /** Location id (`us-central1`, …). */
130 location: string;
131 /** Destination MessageBus resource name. */
132 destination: string | undefined;
133 /** Display name. */
134 displayName: string | undefined;
135 /** Customer-managed KMS key, if any. */
136 cryptoKeyName: string | undefined;
137 /** Platform logging configuration. */
138 loggingConfig: LoggingConfig | undefined;
139 /** Organization-wide subscription config. */
140 organizationSubscription: OrganizationSubscription | undefined;
141 /** Explicit project subscription list. */
142 projectSubscriptions: ProjectSubscriptions | undefined;
143 /** User labels (Alchemy ownership labels stripped). */
144 labels: Record<string, string>;
145 /** User annotations. */
146 annotations: Record<string, string>;
147 /** Server-assigned UUID4, stable until delete. */
148 uid: string | undefined;
149 /** Server checksum of the resource. */
150 etag: string | undefined;
151 /** RFC3339 creation timestamp. */
152 createTime: string | undefined;
153 /** RFC3339 last-update timestamp. */
154 updateTime: string | undefined;
155 },
156 never,
157 Providers
158>;
159
160/**
161 * An Eventarc Advanced GoogleApiSource that delivers first-party Google
162 * API events onto a MessageBus. One source is allowed per project per
163 * region.
164 *
165 * `googleApiSourceId` and `location` are identity — changing either
166 * replaces the source. Destination, display name, labels, annotations,
167 * logging, CMEK, and subscription config update in place.
168 *
169 * ### Creating a GoogleApiSource
170 * **Example:** Collect Google events onto a bus
171 * ```typescript
172 * const bus = yield* GCP.Eventarc.MessageBus("Events", {
173 * location: "us-central1",
174 * });
175 * const source = yield* GCP.Eventarc.GoogleApiSource("GoogleEvents", {
176 * location: "us-central1",
177 * destination: bus.name,
178 * labels: { env: "prod" },
179 * });
180 * ```
181 *
182 * ### Updating a GoogleApiSource
183 * Re-declare the same logical id with changed props; the engine keeps the
184 * physical source and patches it in place.
185 *
186 * **Example:** Change labels and display name
187 * ```typescript
188 * const source = yield* GCP.Eventarc.GoogleApiSource("GoogleEvents", {
189 * location: "us-central1",
190 * destination: bus.name,
191 * displayName: "google events v2",
192 * labels: { env: "prod", role: "source" },
193 * });
194 * ```
195 *
196 * @resource
197 * @category Eventarc
198 */
199export const GoogleApiSource = Resource<GoogleApiSource>(
200 "GCP.Eventarc.GoogleApiSource",
201);
202
203const toProjectSubscriptions = (
204 value: ProjectSubscriptions | eventarc.ProjectSubscriptions | undefined,
205): ProjectSubscriptions | undefined => {
206 const list = [...(value?.list ?? [])].filter((item) => item.length > 0);
207 if (list.length === 0) return undefined;
208 return { list };
209};
210
211const toOrganizationSubscription = (
212 value:
213 | OrganizationSubscription
214 | eventarc.OrganizationSubscription
215 | undefined,
216): OrganizationSubscription | undefined => {
217 if (value === undefined || value.enabled === undefined) return undefined;
218 return { enabled: value.enabled };
219};
220
221const toAttrs = (
222 source: eventarc.GoogleApiSource,
223 project: string,
224 region: string,
225) => {
226 const name = source.name ?? "";
227 const parsed = parseName(name, COLLECTION, region);
228 return {
229 name,
230 googleApiSourceId: parsed.id,
231 project: parsed.project || project,
232 location: parsed.location,
233 destination: source.destination,
234 displayName: source.displayName,
235 cryptoKeyName: source.cryptoKeyName,
236 loggingConfig: toLoggingConfig(source.loggingConfig),
237 organizationSubscription: toOrganizationSubscription(
238 source.organizationSubscription,
239 ),
240 projectSubscriptions: toProjectSubscriptions(source.projectSubscriptions),
241 labels: userLabels(source.labels),
242 annotations: userAnnotations(source.annotations),
243 uid: source.uid,
244 etag: source.etag,
245 createTime: source.createTime,
246 updateTime: source.updateTime,
247 };
248};
249
250const getByName = (name: string) =>
251 eventarc
252 .getProjectsLocationsGoogleApiSources({ name })
253 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
254
255export const GoogleApiSourceProvider = () =>
256 Provider.succeed(GoogleApiSource, {
257 stables: [
258 "name",
259 "googleApiSourceId",
260 "project",
261 "location",
262 "uid",
263 "createTime",
264 ],
265
266 diff: Effect.fn(function* ({ news, olds, output }) {
267 if (!isResolved(news)) return undefined;
268 const env = yield* GcpEnvironment.current;
269 const previousId = olds?.googleApiSourceId ?? output?.googleApiSourceId;
270 const nextId = news.googleApiSourceId
271 ? rfc1035(news.googleApiSourceId, "google-api-source")
272 : previousId;
273 const previousLocation = normalizeLocation(
274 olds?.location ?? output?.location,
275 env.region,
276 );
277 const nextLocation = normalizeLocation(
278 news.location ?? olds?.location ?? output?.location,
279 env.region,
280 );
281 if (
282 (previousId !== undefined &&
283 nextId !== undefined &&
284 nextId !== previousId) ||
285 previousLocation !== nextLocation
286 ) {
287 return { action: "replace" as const };
288 }
289 return undefined;
290 }),
291
292 read: Effect.fn(function* ({ id, olds, output }) {
293 const env = yield* GcpEnvironment.current;
294 const googleApiSourceId = yield* toPhysicalId(
295 id,
296 olds?.googleApiSourceId,
297 output?.googleApiSourceId,
298 "google-api-source",
299 );
300 const location = normalizeLocation(
301 olds?.location ?? output?.location,
302 env.region,
303 );
304 const name =
305 output?.name ??
306 resourceName(env.project, location, COLLECTION, googleApiSourceId);
307 const existing = yield* getByName(name);
308 if (existing === undefined) return undefined;
309 const attrs = toAttrs(existing, env.project, env.region);
310 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
311 ? attrs
312 : Unowned(attrs);
313 }),
314
315 list: () =>
316 Effect.gen(function* () {
317 const env = yield* GcpEnvironment.current;
318 const items = yield* collectPages(
319 eventarc.listProjectsLocationsGoogleApiSources.pages({
320 parent: parentOf(env.project, "-"),
321 pageSize: 1000,
322 }),
323 (page) => page.googleApiSources,
324 );
325 return items
326 .filter((item) => hasAlchemyLabelKeys(item.labels))
327 .map((item) => toAttrs(item, env.project, env.region));
328 }),
329
330 reconcile: Effect.fn(function* ({ id, news, output }) {
331 const env = yield* GcpEnvironment.current;
332 const googleApiSourceId = yield* toPhysicalId(
333 id,
334 news.googleApiSourceId,
335 output?.googleApiSourceId,
336 "google-api-source",
337 );
338 const location = normalizeLocation(
339 news.location ?? output?.location,
340 env.region,
341 );
342 const name = resourceName(
343 env.project,
344 location,
345 COLLECTION,
346 googleApiSourceId,
347 );
348 const desiredLabels = {
349 ...toLabels(news.labels),
350 ...(yield* createInternalLabels(id)),
351 };
352 const destination = expandResource(
353 news.destination,
354 env.project,
355 location,
356 "messageBuses",
357 );
358 const desiredLogging = toLoggingConfig(news.loggingConfig);
359 const desiredCrypto =
360 news.cryptoKeyName && news.cryptoKeyName.length > 0
361 ? news.cryptoKeyName
362 : undefined;
363 const desiredAnnotations = news.annotations
364 ? tagRecord(news.annotations)
365 : undefined;
366 const desiredOrg = toOrganizationSubscription(
367 news.organizationSubscription,
368 );
369 const desiredProjects = toProjectSubscriptions(news.projectSubscriptions);
370
371 let current = yield* getByName(name);
372
373 if (current === undefined) {
374 const created = yield* eventarc
375 .createProjectsLocationsGoogleApiSources({
376 parent: parentOf(env.project, location),
377 googleApiSourceId,
378 body: compact({
379 name,
380 destination,
381 displayName: news.displayName,
382 cryptoKeyName: desiredCrypto,
383 loggingConfig: desiredLogging,
384 organizationSubscription: desiredOrg,
385 projectSubscriptions: desiredProjects,
386 labels: desiredLabels,
387 annotations: desiredAnnotations,
388 }),
389 })
390 .pipe(
391 Effect.retry({
392 while: (error) => error._tag === "Conflict",
393 times: 5,
394 schedule: Schedule.spaced("2 seconds"),
395 }),
396 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
397 );
398 if (created !== undefined) {
399 yield* waitForOperation(created);
400 }
401 current = yield* waitUntilPresent(getByName(name), name);
402 }
403
404 const observedLabels = tagRecord(current.labels);
405 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
406 const updateMask = changedFields([
407 ["labels", upsert.length > 0 || removed.length > 0],
408 ["destination", textKey(current.destination) !== destination],
409 [
410 "displayName",
411 textKey(current.displayName) !== textKey(news.displayName),
412 ],
413 [
414 "cryptoKeyName",
415 cryptoKeyKey(current.cryptoKeyName) !== cryptoKeyKey(desiredCrypto),
416 ],
417 [
418 "loggingConfig",
419 loggingKey(current.loggingConfig) !== loggingKey(desiredLogging),
420 ],
421 [
422 "organizationSubscription",
423 !sameJson(
424 toOrganizationSubscription(current.organizationSubscription),
425 desiredOrg,
426 ),
427 ],
428 [
429 "projectSubscriptions",
430 !sameJson(
431 toProjectSubscriptions(current.projectSubscriptions),
432 desiredProjects,
433 ),
434 ],
435 [
436 "annotations",
437 !sameJson(tagRecord(current.annotations), desiredAnnotations ?? {}),
438 ],
439 ]);
440
441 if (updateMask.length > 0) {
442 const patched = yield* eventarc.patchProjectsLocationsGoogleApiSources({
443 name: current.name ?? name,
444 updateMask: updateMask.join(","),
445 body: {
446 name: current.name ?? name,
447 destination,
448 displayName: news.displayName,
449 cryptoKeyName: desiredCrypto,
450 loggingConfig: desiredLogging,
451 organizationSubscription: desiredOrg,
452 projectSubscriptions: desiredProjects,
453 labels: desiredLabels,
454 annotations: desiredAnnotations ?? {},
455 },
456 });
457 yield* waitForOperation(patched);
458 current = yield* waitUntilPresent(
459 getByName(current.name ?? name),
460 current.name ?? name,
461 );
462 }
463
464 return toAttrs(current, env.project, env.region);
465 }),
466
467 delete: Effect.fn(function* ({ output }) {
468 yield* retryOnTransient(
469 Effect.gen(function* () {
470 const existing = yield* getByName(output.name);
471 if (existing === undefined) return;
472 const operation = yield* eventarc
473 .deleteProjectsLocationsGoogleApiSources({
474 name: output.name,
475 allowMissing: true,
476 })
477 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
478 if (operation !== undefined) {
479 yield* waitForOperation(operation, { notFoundOk: true });
480 }
481 }),
482 ).pipe(
483 Effect.catchTag("GCP.OperationFailed", (error) =>
484 getByName(output.name).pipe(
485 Effect.flatMap((current) =>
486 current === undefined ? Effect.void : Effect.fail(error),
487 ),
488 ),
489 ),
490 );
491 yield* waitUntilGone(getByName(output.name), output.name);
492 }),
493 });