1import * as resourcemanager from "@distilled.cloud/gcp/cloudresourcemanager_v3";
2import * as eventarc from "@distilled.cloud/gcp/eventarc_v1";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { tagRecord } from "../../Tags.ts";
13import { GcpEnvironment } from "../Environment.ts";
14import {
15 createInternalLabels,
16 diffLabels,
17 hasAlchemyLabels,
18 stripInternalLabels,
19 toLabels,
20} from "../Labels.ts";
21import type { Providers } from "../Providers.ts";
22import { waitForOperation } from "./internal.ts";
23
24const DEFAULT_CONTENT_TYPE = "application/json";
25const MAX_NAME_LENGTH = 63;
26
27export type EventFilter = {
28 /**
29 * CloudEvents attribute to match. Every trigger must include a filter
30 * on `type`.
31 */
32 attribute: string;
33 /**
34 * Value the attribute must equal (or match via `operator`).
35 */
36 value: string;
37 /**
38 * Optional match operator. Allowed values: `path_pattern`,
39 * `match-path-pattern`. Omitted means exact match.
40 */
41 operator?: string;
42};
43
44export type CloudRunDestination = {
45 /**
46 * Cloud Run service name (the `{service}` segment, not a full resource
47 * name). Must live in the same project as the trigger.
48 */
49 service: string;
50 /**
51 * Region the Cloud Run service is deployed in.
52 */
53 region: string;
54 /**
55 * Relative path on the service events are sent to, e.g. `/events`.
56 */
57 path?: string;
58};
59
60export type GkeDestination = {
61 /**
62 * GKE cluster name. Must be in the same project as the trigger.
63 */
64 cluster: string;
65 /**
66 * Zone (zonal clusters) or region (regional clusters) of the cluster.
67 */
68 location: string;
69 /**
70 * GKE service name.
71 */
72 service: string;
73 /**
74 * Kubernetes namespace the service runs in.
75 */
76 namespace: string;
77 /**
78 * Relative path on the GKE service events are sent to.
79 */
80 path?: string;
81};
82
83export type HttpEndpointDestination = {
84 /**
85 * RFC2396 URI of the HTTP endpoint (`http://` or `https://`).
86 */
87 uri: string;
88};
89
90export type NetworkConfig = {
91 /**
92 * Network attachment granting Eventarc access to the VPC, as
93 * `projects/{project}/regions/{region}/networkAttachments/{name}`.
94 * Required with `httpEndpoint`.
95 */
96 networkAttachment: string;
97};
98
99export type Destination = {
100 /**
101 * Cloud Run service that receives events.
102 */
103 cloudRun?: CloudRunDestination;
104 /**
105 * GKE service that receives events.
106 */
107 gke?: GkeDestination;
108 /**
109 * Workflow resource name
110 * (`projects/{project}/locations/{location}/workflows/{workflow}`).
111 */
112 workflow?: string;
113 /**
114 * Internal HTTP endpoint that receives events.
115 */
116 httpEndpoint?: HttpEndpointDestination;
117 /**
118 * VPC network config. Only valid with `httpEndpoint`.
119 */
120 networkConfig?: NetworkConfig;
121};
122
123export type PubsubTransport = {
124 /**
125 * Existing Pub/Sub topic Eventarc uses as the event source, as
126 * `projects/{project}/topics/{topic}`. Required for
127 * `google.cloud.pubsub.topic.v1.messagePublished`. Immutable —
128 * changing it replaces the trigger. Eventarc-managed topics are not
129 * deleted with the trigger.
130 */
131 topic?: string;
132 /**
133 * Output only. Pub/Sub subscription Eventarc manages for delivery.
134 */
135 subscription?: string;
136};
137
138export type Transport = {
139 /**
140 * Pub/Sub transport intermediary.
141 */
142 pubsub?: PubsubTransport;
143};
144
145export type RetryPolicy = {
146 /**
147 * Maximum delivery attempts. The only valid value is `1`. Cloud Run
148 * destinations only.
149 */
150 maxAttempts?: number;
151};
152
153export type TriggerProps = {
154 /**
155 * Trigger id (the `{trigger}` segment of
156 * `projects/{project}/locations/{location}/triggers/{trigger}`). If
157 * omitted, a unique name is generated from the stack, stage, and
158 * logical id. Must match `[a-z]([a-z0-9-]*[a-z0-9])?` and be 1-63
159 * characters. Immutable — changing it replaces the trigger.
160 */
161 triggerId?: string;
162 /**
163 * Eventarc location (`us-central1`, `us-east1`, …). Immutable —
164 * changing it replaces the trigger. `US-CENTRAL1` is accepted and
165 * normalized to `us-central1`.
166 * @default the stack's GCP region (`GCP.Region`, else the profile
167 * region, else `us-central1`)
168 */
169 location?: string;
170 /**
171 * CloudEvents attribute filters. Every event must match all filters.
172 * A `type` filter is required. Immutable — changing filters replaces
173 * the trigger.
174 */
175 eventFilters: EventFilter[];
176 /**
177 * Where matching events are delivered (Cloud Run, GKE, Workflows, or
178 * an HTTP endpoint).
179 */
180 destination: Destination;
181 /**
182 * Transport intermediary. For Pub/Sub events, set `pubsub.topic` to an
183 * existing topic. Immutable — changing the topic replaces the trigger.
184 */
185 transport?: Transport;
186 /**
187 * Partner channel,
188 * `projects/{project}/locations/{location}/channels/{channel}`.
189 * Required for Eventarc SaaS partner events. Immutable — changing it
190 * replaces the trigger.
191 */
192 channel?: string;
193 /**
194 * IAM service account email Eventarc uses as the trigger identity.
195 * The caller needs `iam.serviceAccounts.actAs` on this account. If
196 * omitted, Alchemy uses the project's Compute Engine default service
197 * account (`{PROJECT_NUMBER}-compute@developer.gserviceaccount.com`).
198 * Required by the Eventarc API for Workflows (and most other)
199 * destinations.
200 */
201 serviceAccount?: string;
202 /**
203 * MIME type expected in the CloudEvent `data` payload.
204 * @default "application/json"
205 */
206 eventDataContentType?: string;
207 /**
208 * Retry policy. Cloud Run destinations only; omit for the 24-hour
209 * default.
210 */
211 retryPolicy?: RetryPolicy;
212 /**
213 * User labels. Alchemy ownership labels are merged in automatically.
214 */
215 labels?: Record<string, string>;
216};
217
218export type Trigger = Resource<
219 "GCP.Eventarc.Trigger",
220 TriggerProps,
221 {
222 /** Full resource name `projects/{project}/locations/{location}/triggers/{trigger}`. */
223 name: string;
224 /** Trigger id (last path segment). */
225 triggerId: string;
226 /** Project id. */
227 project: string;
228 /** Location id (`us-central1`, …). */
229 location: string;
230 /** CloudEvents attribute filters. */
231 eventFilters: EventFilter[];
232 /** Destination currently configured on the trigger. */
233 destination: Destination | undefined;
234 /** Transport intermediary, including Eventarc-managed Pub/Sub resources. */
235 transport: Transport | undefined;
236 /** Partner channel, if any. */
237 channel: string | undefined;
238 /** Trigger identity service account. */
239 serviceAccount: string | undefined;
240 /** CloudEvent data MIME type. */
241 eventDataContentType: string | undefined;
242 /** Retry policy, if set. */
243 retryPolicy: RetryPolicy | undefined;
244 /** User labels (Alchemy ownership labels stripped). */
245 labels: Record<string, string>;
246 /** Server-assigned UUID4, stable until delete. */
247 uid: string | undefined;
248 /** Server checksum of the resource. */
249 etag: string | undefined;
250 /** RFC3339 creation timestamp. */
251 createTime: string | undefined;
252 /** RFC3339 last-update timestamp. */
253 updateTime: string | undefined;
254 /** Whether the trigger satisfies physical zone separation. */
255 satisfiesPzs: boolean | undefined;
256 /** Reasons the trigger is in FAILED state, if any. */
257 conditions: eventarc.StateConditionMap | undefined;
258 },
259 never,
260 Providers
261>;
262
263/**
264 * An Eventarc trigger that routes events matching a set of CloudEvents
265 * filters to a Cloud Run service, GKE service, Workflow, or HTTP
266 * endpoint.
267 *
268 * `triggerId`, `location`, `eventFilters`, `channel`, and
269 * `transport.pubsub.topic` are identity — changing any of them replaces
270 * the trigger. Destination, service account, content type, retry policy,
271 * and labels update in place.
272 *
273 * ### Creating a Trigger
274 * **Example:** Pub/Sub messages to a Workflow
275 * ```typescript
276 * const topic = yield* GCP.PubSub.Topic("events", {});
277 * const trigger = yield* GCP.Eventarc.Trigger("orders", {
278 * eventFilters: [
279 * {
280 * attribute: "type",
281 * value: "google.cloud.pubsub.topic.v1.messagePublished",
282 * },
283 * ],
284 * destination: {
285 * workflow:
286 * "projects/my-project/locations/us-central1/workflows/orders",
287 * },
288 * transport: { pubsub: { topic: topic.name } },
289 * });
290 * ```
291 *
292 * **Example:** Pub/Sub messages to Cloud Run
293 * ```typescript
294 * const trigger = yield* GCP.Eventarc.Trigger("orders", {
295 * triggerId: "order-events",
296 * location: "us-central1",
297 * eventFilters: [
298 * {
299 * attribute: "type",
300 * value: "google.cloud.pubsub.topic.v1.messagePublished",
301 * },
302 * ],
303 * destination: {
304 * cloudRun: { service: "order-handler", region: "us-central1" },
305 * },
306 * transport: { pubsub: { topic: topic.name } },
307 * labels: { env: "prod" },
308 * });
309 * ```
310 *
311 * ### Updating a Trigger
312 * Re-declare the same logical id with changed props; the engine keeps the
313 * physical trigger and patches it in place.
314 *
315 * **Example:** Change labels and the CloudEvent content type
316 * ```typescript
317 * const trigger = yield* GCP.Eventarc.Trigger("orders", {
318 * eventFilters: [
319 * {
320 * attribute: "type",
321 * value: "google.cloud.pubsub.topic.v1.messagePublished",
322 * },
323 * ],
324 * destination: {
325 * workflow:
326 * "projects/my-project/locations/us-central1/workflows/orders",
327 * },
328 * transport: { pubsub: { topic: topic.name } },
329 * eventDataContentType: "application/json",
330 * labels: { env: "prod", role: "events" },
331 * });
332 * ```
333 *
334 * @resource
335 * @category Eventarc
336 */
337export const Trigger = Resource<Trigger>("GCP.Eventarc.Trigger");
338
339export class TriggerNotHealthy extends Data.TaggedError(
340 "GCP.Eventarc.TriggerNotHealthy",
341)<{ name: string; conditions: string }> {}
342
343export class TriggerNotResolved extends Data.TaggedError(
344 "GCP.Eventarc.TriggerNotResolved",
345)<{
346 name: string;
347}> {}
348
349export class TriggerStillExists extends Data.TaggedError(
350 "GCP.Eventarc.TriggerStillExists",
351)<{
352 name: string;
353}> {}
354
355export class TriggerServiceAccountMissing extends Data.TaggedError(
356 "GCP.Eventarc.TriggerServiceAccountMissing",
357)<{
358 project: string;
359}> {}
360
361const lastSegment = (value: string) => {
362 const trimmed = value.replace(/\/+$/, "");
363 const parts = trimmed.split("/");
364 return parts[parts.length - 1] || trimmed;
365};
366
367const normalizeLocation = (
368 location: string | undefined,
369 defaultLocation: string,
370) => lastSegment(location ?? defaultLocation).toLowerCase();
371
372const rfc1035 = (name: string): string => {
373 let next = name
374 .toLowerCase()
375 .replace(/[^a-z0-9-]/g, "-")
376 .replace(/-+/g, "-")
377 .replace(/^-+|-+$/g, "");
378 if (!/^[a-z]/.test(next)) next = `t${next}`;
379 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
380 return next.length > 0 ? next : "trigger";
381};
382
383const resourceName = (project: string, location: string, triggerId: string) =>
384 `projects/${project}/locations/${location}/triggers/${triggerId}`;
385
386const parseName = (name: string, defaultLocation: string) => {
387 const parts = name.split("/").filter((part) => part.length > 0);
388 const triggersAt = parts.lastIndexOf("triggers");
389 const locationsAt = parts.lastIndexOf("locations");
390 const projectsAt = parts.lastIndexOf("projects");
391 return {
392 project:
393 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
394 location:
395 locationsAt >= 0 && parts[locationsAt + 1]
396 ? parts[locationsAt + 1]!
397 : defaultLocation,
398 triggerId:
399 triggersAt >= 0 && parts[triggersAt + 1]
400 ? parts[triggersAt + 1]!
401 : lastSegment(name),
402 };
403};
404
405const userLabels = (
406 labels: Record<string, string | undefined> | null | undefined,
407): Record<string, string> => stripInternalLabels(tagRecord(labels));
408
409const toId = (id: string, triggerId: string | undefined, existing?: string) =>
410 Effect.gen(function* () {
411 if (triggerId !== undefined) return rfc1035(triggerId);
412 if (existing !== undefined) return existing;
413 return rfc1035(
414 yield* createPhysicalName({
415 id,
416 maxLength: MAX_NAME_LENGTH,
417 lowercase: true,
418 }),
419 );
420 });
421
422const compact = <T extends Record<string, unknown>>(value: T): T => {
423 const next: Record<string, unknown> = {};
424 for (const [key, entry] of Object.entries(value)) {
425 if (entry !== undefined && entry !== "") next[key] = entry;
426 }
427 return next as T;
428};
429
430const toEventFilters = (
431 filters: eventarc.EventFilterList | undefined,
432): EventFilter[] =>
433 (filters ?? [])
434 .filter(
435 (
436 filter,
437 ): filter is eventarc.EventFilter & {
438 attribute: string;
439 value: string;
440 } =>
441 typeof filter.attribute === "string" &&
442 typeof filter.value === "string",
443 )
444 .map((filter) =>
445 compact({
446 attribute: filter.attribute,
447 value: filter.value,
448 operator: filter.operator,
449 }),
450 );
451
452const toDestination = (
453 destination: eventarc.Destination | undefined,
454): Destination | undefined => {
455 if (destination === undefined) return undefined;
456 const next = compact({
457 cloudRun: destination.cloudRun
458 ? compact({
459 service: destination.cloudRun.service ?? "",
460 region: destination.cloudRun.region ?? "",
461 path: destination.cloudRun.path,
462 })
463 : undefined,
464 gke: destination.gke
465 ? compact({
466 cluster: destination.gke.cluster ?? "",
467 location: destination.gke.location ?? "",
468 service: destination.gke.service ?? "",
469 namespace: destination.gke.namespace ?? "",
470 path: destination.gke.path,
471 })
472 : undefined,
473 workflow: destination.workflow,
474 httpEndpoint: destination.httpEndpoint
475 ? compact({ uri: destination.httpEndpoint.uri ?? "" })
476 : undefined,
477 networkConfig: destination.networkConfig
478 ? compact({
479 networkAttachment: destination.networkConfig.networkAttachment ?? "",
480 })
481 : undefined,
482 });
483 return Object.keys(next).length > 0 ? next : undefined;
484};
485
486const fromDestination = (destination: Destination): eventarc.Destination =>
487 compact({
488 cloudRun: destination.cloudRun
489 ? compact({
490 service: destination.cloudRun.service,
491 region: destination.cloudRun.region,
492 path: destination.cloudRun.path,
493 })
494 : undefined,
495 gke: destination.gke
496 ? compact({
497 cluster: destination.gke.cluster,
498 location: destination.gke.location,
499 service: destination.gke.service,
500 namespace: destination.gke.namespace,
501 path: destination.gke.path,
502 })
503 : undefined,
504 workflow: destination.workflow,
505 httpEndpoint: destination.httpEndpoint
506 ? compact({ uri: destination.httpEndpoint.uri })
507 : undefined,
508 networkConfig: destination.networkConfig
509 ? compact({
510 networkAttachment: destination.networkConfig.networkAttachment,
511 })
512 : undefined,
513 });
514
515const toTransport = (
516 transport: eventarc.Transport | undefined,
517): Transport | undefined => {
518 if (transport?.pubsub === undefined) return undefined;
519 return {
520 pubsub: compact({
521 topic: transport.pubsub.topic,
522 subscription: transport.pubsub.subscription,
523 }),
524 };
525};
526
527const topicKey = (topic: string | undefined, project: string) => {
528 if (topic === undefined || topic.length === 0) return "";
529 if (topic.includes("/")) return topic;
530 return `projects/${project}/topics/${topic}`;
531};
532
533const saKey = (serviceAccount: string | undefined) => {
534 if (serviceAccount === undefined || serviceAccount.length === 0) return "";
535 return serviceAccount.includes("/")
536 ? lastSegment(serviceAccount)
537 : serviceAccount;
538};
539
540const filtersKey = (filters: EventFilter[] | undefined) =>
541 JSON.stringify(
542 [...(filters ?? [])]
543 .map((filter) => ({
544 attribute: filter.attribute,
545 value: filter.value,
546 operator: filter.operator ?? "",
547 }))
548 .sort((left, right) =>
549 left.attribute === right.attribute
550 ? left.value.localeCompare(right.value)
551 : left.attribute.localeCompare(right.attribute),
552 ),
553 );
554
555const destinationKey = (destination: Destination | undefined) =>
556 JSON.stringify(toDestination(fromDestination(destination ?? {})) ?? {});
557
558const contentTypeKey = (value: string | undefined) =>
559 (value === undefined || value.length === 0
560 ? DEFAULT_CONTENT_TYPE
561 : value
562 ).toLowerCase();
563
564const retryKey = (policy: RetryPolicy | undefined) =>
565 JSON.stringify({ maxAttempts: policy?.maxAttempts ?? null });
566
567const toAttrs = (
568 trigger: eventarc.Trigger,
569 project: string,
570 region: string,
571) => {
572 const name = trigger.name ?? "";
573 const parsed = parseName(name, region);
574 return {
575 name,
576 triggerId: parsed.triggerId,
577 project: parsed.project || project,
578 location: parsed.location,
579 eventFilters: toEventFilters(trigger.eventFilters),
580 destination: toDestination(trigger.destination),
581 transport: toTransport(trigger.transport),
582 channel: trigger.channel,
583 serviceAccount: trigger.serviceAccount,
584 eventDataContentType: trigger.eventDataContentType ?? DEFAULT_CONTENT_TYPE,
585 retryPolicy: trigger.retryPolicy
586 ? compact({ maxAttempts: trigger.retryPolicy.maxAttempts })
587 : undefined,
588 labels: userLabels(trigger.labels),
589 uid: trigger.uid,
590 etag: trigger.etag,
591 createTime: trigger.createTime,
592 updateTime: trigger.updateTime,
593 satisfiesPzs: trigger.satisfiesPzs,
594 conditions: trigger.conditions,
595 };
596};
597
598const getByName = (name: string) =>
599 eventarc
600 .getProjectsLocationsTriggers({ name })
601 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
602
603/**
604 * Block until Eventarc reports every trigger condition healthy (`OK`), so
605 * nothing downstream assumes a trigger that is still provisioning its
606 * transport. A trigger whose conditions settle on an error fails.
607 */
608const waitUntilHealthy = (name: string) =>
609 waitUntilExists(name).pipe(
610 Effect.flatMap((trigger) => {
611 const failing = Object.entries(trigger.conditions ?? {}).filter(
612 ([, condition]) =>
613 condition?.code !== undefined && condition.code !== "OK",
614 );
615 return failing.length === 0
616 ? Effect.succeed(trigger)
617 : Effect.fail(
618 new TriggerNotHealthy({
619 name,
620 conditions: failing
621 .map(
622 ([key, condition]) =>
623 `${key}: ${condition?.code} ${condition?.message ?? ""}`,
624 )
625 .join("; "),
626 }),
627 );
628 }),
629 Effect.retry({
630 while: (error) => error._tag === "GCP.Eventarc.TriggerNotHealthy",
631 times: 30,
632 schedule: Schedule.spaced("5 seconds"),
633 }),
634 );
635
636const waitUntilExists = (name: string) =>
637 getByName(name).pipe(
638 Effect.flatMap((trigger) =>
639 trigger
640 ? Effect.succeed(trigger)
641 : Effect.fail(new TriggerNotResolved({ name })),
642 ),
643 Effect.retry({
644 while: (error) => error._tag === "GCP.Eventarc.TriggerNotResolved",
645 times: 8,
646 schedule: Schedule.spaced("1 second"),
647 }),
648 );
649
650const waitUntilGone = (name: string) =>
651 getByName(name).pipe(
652 Effect.flatMap((trigger) =>
653 trigger === undefined
654 ? Effect.void
655 : Effect.fail(new TriggerStillExists({ name })),
656 ),
657 Effect.retry({
658 while: (error) => error._tag === "GCP.Eventarc.TriggerStillExists",
659 times: 10,
660 schedule: Schedule.spaced("2 seconds"),
661 }),
662 );
663
664const desiredTransport = (
665 news: TriggerProps,
666): eventarc.Transport | undefined => {
667 const topic = news.transport?.pubsub?.topic;
668 if (topic === undefined || topic.length === 0) return undefined;
669 return { pubsub: { topic } };
670};
671
672const defaultServiceAccount = (project: string) =>
673 resourcemanager.getProjects({ name: `projects/${project}` }).pipe(
674 Effect.flatMap((resource) => {
675 const projectNumber = lastSegment(resource.name ?? "");
676 if (!/^\d+$/.test(projectNumber)) {
677 return Effect.fail(new TriggerServiceAccountMissing({ project }));
678 }
679 return Effect.succeed(
680 `${projectNumber}-compute@developer.gserviceaccount.com`,
681 );
682 }),
683 );
684
685const resolveServiceAccount = (
686 project: string,
687 news: TriggerProps,
688 current: eventarc.Trigger | undefined,
689) =>
690 Effect.gen(function* () {
691 if (news.serviceAccount !== undefined && news.serviceAccount.length > 0) {
692 return news.serviceAccount;
693 }
694 if (
695 current?.serviceAccount !== undefined &&
696 current.serviceAccount.length > 0
697 ) {
698 return current.serviceAccount;
699 }
700 return yield* defaultServiceAccount(project);
701 });
702
703export const TriggerProvider = () =>
704 Provider.succeed(Trigger, {
705 stables: [
706 "name",
707 "triggerId",
708 "project",
709 "location",
710 "uid",
711 "createTime",
712 "channel",
713 ],
714
715 diff: Effect.fn(function* ({ news, olds, output }) {
716 if (!isResolved(news)) return undefined;
717 const env = yield* GcpEnvironment.current;
718
719 const previousId = olds?.triggerId ?? output?.triggerId;
720 const nextId = news.triggerId ? rfc1035(news.triggerId) : previousId;
721 const previousLocation = normalizeLocation(
722 olds?.location ?? output?.location,
723 env.region,
724 );
725 const nextLocation = normalizeLocation(
726 news.location ?? output?.location,
727 env.region,
728 );
729 const project = output?.project ?? "";
730 const previousChannel = olds?.channel ?? output?.channel ?? "";
731 const nextChannel = news.channel ?? previousChannel;
732 const previousTopic = topicKey(
733 olds?.transport?.pubsub?.topic ?? output?.transport?.pubsub?.topic,
734 project,
735 );
736 const nextTopic =
737 news.transport?.pubsub?.topic !== undefined
738 ? topicKey(news.transport.pubsub.topic, project)
739 : previousTopic;
740 const previousFilters = filtersKey(
741 olds?.eventFilters ?? output?.eventFilters,
742 );
743 const nextFilters = filtersKey(news.eventFilters);
744
745 const replace =
746 (previousId !== undefined &&
747 nextId !== undefined &&
748 nextId !== previousId) ||
749 previousLocation !== nextLocation ||
750 previousChannel !== nextChannel ||
751 (nextTopic.length > 0 &&
752 previousTopic.length > 0 &&
753 nextTopic !== previousTopic) ||
754 previousFilters !== nextFilters;
755
756 if (!replace) return undefined;
757 return {
758 action: "replace" as const,
759 deleteFirst:
760 previousLocation === nextLocation &&
761 previousId !== undefined &&
762 nextId === previousId,
763 };
764 }),
765
766 read: Effect.fn(function* ({ id, olds, output }) {
767 const env = yield* GcpEnvironment.current;
768 const triggerId = yield* toId(id, olds?.triggerId, output?.triggerId);
769 const location = normalizeLocation(
770 olds?.location ?? output?.location,
771 env.region,
772 );
773 const name =
774 output?.name ?? resourceName(env.project, location, triggerId);
775 const existing = yield* getByName(name);
776 if (existing === undefined) return undefined;
777 const attrs = toAttrs(existing, env.project, env.region);
778 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
779 ? attrs
780 : Unowned(attrs);
781 }),
782
783 list: () =>
784 Effect.gen(function* () {
785 const env = yield* GcpEnvironment.current;
786 return yield* eventarc.listProjectsLocationsTriggers
787 .pages({
788 parent: `projects/${env.project}/locations/-`,
789 pageSize: 1000,
790 })
791 .pipe(
792 Stream.flatMap((page) => Stream.fromIterable(page.triggers ?? [])),
793 Stream.filter((trigger) =>
794 Object.keys(trigger.labels ?? {}).some((key) =>
795 key.startsWith("alchemy-"),
796 ),
797 ),
798 Stream.map((trigger) => toAttrs(trigger, env.project, env.region)),
799 Stream.runCollect,
800 Effect.map((chunk) => Array.from(chunk)),
801 );
802 }),
803
804 reconcile: Effect.fn(function* ({ id, news, output }) {
805 const env = yield* GcpEnvironment.current;
806 const triggerId = yield* toId(id, news.triggerId, output?.triggerId);
807 const location = normalizeLocation(
808 news.location ?? output?.location,
809 env.region,
810 );
811 const name = resourceName(env.project, location, triggerId);
812 const desiredLabels = {
813 ...toLabels(news.labels),
814 ...(yield* createInternalLabels(id)),
815 };
816 const eventFilters = news.eventFilters.map((filter) =>
817 compact({
818 attribute: filter.attribute,
819 value: filter.value,
820 operator: filter.operator,
821 }),
822 );
823 const destination = fromDestination(news.destination);
824 const transport = desiredTransport(news);
825
826 let current = yield* getByName(name);
827 const serviceAccount = yield* resolveServiceAccount(
828 env.project,
829 news,
830 current,
831 );
832
833 if (current === undefined) {
834 const created = yield* eventarc
835 .createProjectsLocationsTriggers({
836 parent: `projects/${env.project}/locations/${location}`,
837 triggerId,
838 body: {
839 name,
840 labels: desiredLabels,
841 eventFilters,
842 destination,
843 transport,
844 channel: news.channel,
845 serviceAccount,
846 eventDataContentType: news.eventDataContentType,
847 retryPolicy: news.retryPolicy,
848 },
849 })
850 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
851 if (created !== undefined) {
852 yield* waitForOperation(created);
853 }
854 current = yield* waitUntilExists(name);
855 }
856
857 if (current === undefined) {
858 return yield* new TriggerNotResolved({ name });
859 }
860
861 const observedLabels = tagRecord(current.labels);
862 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
863 const labelsChanged = upsert.length > 0 || removed.length > 0;
864 const destinationChanged =
865 destinationKey(toDestination(current.destination)) !==
866 destinationKey(news.destination);
867 const serviceAccountChanged =
868 saKey(serviceAccount) !== saKey(current.serviceAccount);
869 const contentTypeChanged =
870 news.eventDataContentType !== undefined &&
871 contentTypeKey(current.eventDataContentType) !==
872 contentTypeKey(news.eventDataContentType);
873 const retryChanged =
874 news.retryPolicy !== undefined &&
875 retryKey(current.retryPolicy) !== retryKey(news.retryPolicy);
876
877 if (
878 labelsChanged ||
879 destinationChanged ||
880 serviceAccountChanged ||
881 contentTypeChanged ||
882 retryChanged
883 ) {
884 const updateMask = [
885 labelsChanged ? "labels" : undefined,
886 destinationChanged ? "destination" : undefined,
887 serviceAccountChanged ? "serviceAccount" : undefined,
888 contentTypeChanged ? "eventDataContentType" : undefined,
889 retryChanged ? "retryPolicy" : undefined,
890 ].filter((field): field is string => field !== undefined);
891
892 const patched = yield* eventarc.patchProjectsLocationsTriggers({
893 name,
894 updateMask: updateMask.join(","),
895 body: {
896 name,
897 labels: desiredLabels,
898 destination,
899 serviceAccount,
900 eventDataContentType: news.eventDataContentType,
901 retryPolicy: news.retryPolicy,
902 },
903 });
904 yield* waitForOperation(patched);
905 current = yield* waitUntilExists(name);
906 }
907
908 current = yield* waitUntilHealthy(name);
909 return toAttrs(current, env.project, env.region);
910 }),
911
912 delete: Effect.fn(function* ({ output }) {
913 const existing = yield* getByName(output.name);
914 if (existing === undefined) return;
915 const operation = yield* eventarc
916 .deleteProjectsLocationsTriggers({
917 name: output.name,
918 allowMissing: true,
919 })
920 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
921 if (operation !== undefined) {
922 yield* waitForOperation(operation, { notFoundOk: true });
923 }
924 yield* waitUntilGone(output.name);
925 }),
926 });