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 expandTopic,
25 hasAlchemyLabelKeys,
26 loggingKey,
27 normalizeLocation,
28 parentOf,
29 parseName,
30 resourceName,
31 rfc1035,
32 sameJson,
33 textKey,
34 toLoggingConfig,
35 toPhysicalId,
36 userAnnotations,
37 userLabels,
38 retryOnTransient,
39 waitForOperation,
40 waitUntilGone,
41 waitUntilPresent,
42} from "./internal.ts";
43
44const COLLECTION = "pipelines";
45
46export type PipelinePayloadSchema = {
47 /** Entire schema definition stored inline. */
48 schemaDefinition?: string;
49};
50
51export type PipelinePayloadFormat = {
52 /** Protobuf payload format. */
53 protobuf?: PipelinePayloadSchema;
54 /** Avro payload format. */
55 avro?: PipelinePayloadSchema;
56 /** JSON payload format. An empty object selects JSON. */
57 json?: Record<string, never>;
58};
59
60export type PipelineTransformation = {
61 /**
62 * CEL expression template applied to transform messages.
63 */
64 transformationTemplate?: string;
65};
66
67export type PipelineMediation = {
68 /**
69 * How the pipeline transforms messages. Currently the only mediation.
70 */
71 transformation?: PipelineTransformation;
72};
73
74export type PipelineRetryPolicy = {
75 /**
76 * Maximum delivery attempts (1-100). Default 5.
77 */
78 maxAttempts?: number;
79 /**
80 * Minimum wait between retries, as a duration string (e.g. `"5s"`).
81 * 1-600 seconds. Default `"5s"`.
82 */
83 minRetryDelay?: string;
84 /**
85 * Maximum wait between retries, as a duration string (e.g. `"60s"`).
86 * 1-600 seconds. Default `"60s"`.
87 */
88 maxRetryDelay?: string;
89};
90
91export type PipelineDestinationNetworkConfig = {
92 /**
93 * Network attachment granting the pipeline access to the consumer
94 * VPC, as
95 * `projects/{project}/regions/{region}/networkAttachments/{name}`.
96 */
97 networkAttachment?: string;
98};
99
100export type PipelineDestinationHttpEndpoint = {
101 /**
102 * HTTPS URI of the HTTP endpoint (`https://svc.example.com/route`).
103 */
104 uri: string;
105 /**
106 * CEL expression that rewrites the destination-bound HTTP request.
107 */
108 messageBindingTemplate?: string;
109};
110
111export type PipelineOauthToken = {
112 /**
113 * Service account email used to mint the OAuth token. The caller
114 * needs `iam.serviceAccounts.actAs` on this account.
115 */
116 serviceAccount?: string;
117 /**
118 * OAuth scope. Defaults to `https://www.googleapis.com/auth/cloud-platform`.
119 */
120 scope?: string;
121};
122
123export type PipelineOidcToken = {
124 /**
125 * Service account email used to mint the OIDC token.
126 */
127 serviceAccount?: string;
128 /**
129 * Audience claim. Defaults to the destination URI.
130 */
131 audience?: string;
132};
133
134export type PipelineAuthenticationConfig = {
135 /** Attach an OAuth2 access token as an `Authorization` header. */
136 oauthToken?: PipelineOauthToken;
137 /** Attach a Google OIDC token as an `Authorization` header. */
138 googleOidc?: PipelineOidcToken;
139};
140
141export type PipelineDestination = {
142 /**
143 * HTTP endpoint destination. HTTPS only.
144 */
145 httpEndpoint?: PipelineDestinationHttpEndpoint;
146 /**
147 * Workflow whose executions are triggered by events, as
148 * `projects/{project}/locations/{location}/workflows/{workflow}`.
149 */
150 workflow?: string;
151 /**
152 * Pub/Sub topic to publish to, as `projects/{project}/topics/{topic}`.
153 */
154 topic?: string;
155 /**
156 * MessageBus to publish to, as
157 * `projects/{project}/locations/{location}/messageBuses/{messageBus}`.
158 */
159 messageBus?: string;
160 /**
161 * VPC network config used to resolve and reach the destination.
162 */
163 networkConfig?: PipelineDestinationNetworkConfig;
164 /**
165 * Authentication attached to destination requests.
166 */
167 authenticationConfig?: PipelineAuthenticationConfig;
168 /**
169 * Message format delivered to the destination. Requires
170 * `inputPayloadFormat` on the pipeline.
171 */
172 outputPayloadFormat?: PipelinePayloadFormat;
173};
174
175export type PipelineProps = {
176 /**
177 * Pipeline id (the `{pipeline}` segment of
178 * `projects/{project}/locations/{location}/pipelines/{pipeline}`). If
179 * omitted, a unique name is generated. Must match
180 * `[a-z]([a-z0-9-]*[a-z0-9])?` and be 1-63 characters. Immutable —
181 * changing it replaces the pipeline.
182 */
183 pipelineId?: string;
184 /**
185 * Eventarc Advanced location (`us-central1`, `us-east4`, …). Immutable
186 * — changing it replaces the pipeline. `US-CENTRAL1` is accepted and
187 * normalized to `us-central1`.
188 * @default the stack's GCP region (`GCP.Region`, else the profile
189 * region, else `us-central1`)
190 */
191 location?: string;
192 /**
193 * Destinations messages are forwarded to. The API currently accepts
194 * exactly one destination.
195 */
196 destinations: PipelineDestination[];
197 /**
198 * Payload format expected of incoming messages. Unset treats the
199 * payload as opaque bytes.
200 */
201 inputPayloadFormat?: PipelinePayloadFormat;
202 /**
203 * Mediation operations. Currently at most one transformation.
204 */
205 mediations?: PipelineMediation[];
206 /**
207 * Retry policy for non-responsive or retryable destinations.
208 */
209 retryPolicy?: PipelineRetryPolicy;
210 /**
211 * Human-readable display name.
212 */
213 displayName?: string;
214 /**
215 * Customer-managed Cloud KMS key used to encrypt/decrypt event data,
216 * as
217 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
218 */
219 cryptoKeyName?: string;
220 /**
221 * Platform logging configuration.
222 */
223 loggingConfig?: LoggingConfig;
224 /**
225 * User labels. Alchemy ownership labels are merged in automatically.
226 */
227 labels?: Record<string, string>;
228 /**
229 * Free-form annotations.
230 */
231 annotations?: Record<string, string>;
232};
233
234export type Pipeline = Resource<
235 "GCP.Eventarc.Pipeline",
236 PipelineProps,
237 {
238 /** Full resource name `projects/{project}/locations/{location}/pipelines/{pipeline}`. */
239 name: string;
240 /** Pipeline id (last path segment). */
241 pipelineId: string;
242 /** Project id. */
243 project: string;
244 /** Location id (`us-central1`, …). */
245 location: string;
246 /** Configured destinations. */
247 destinations: PipelineDestination[];
248 /** Incoming payload format. */
249 inputPayloadFormat: PipelinePayloadFormat | undefined;
250 /** Mediation operations. */
251 mediations: PipelineMediation[] | undefined;
252 /** Retry policy. */
253 retryPolicy: PipelineRetryPolicy | undefined;
254 /** Display name. */
255 displayName: string | undefined;
256 /** Customer-managed KMS key, if any. */
257 cryptoKeyName: string | undefined;
258 /** Platform logging configuration. */
259 loggingConfig: LoggingConfig | undefined;
260 /** User labels (Alchemy ownership labels stripped). */
261 labels: Record<string, string>;
262 /** User annotations. */
263 annotations: Record<string, string>;
264 /** Whether the pipeline satisfies physical zone separation. */
265 satisfiesPzs: boolean | undefined;
266 /** Server-assigned UUID4, stable until delete. */
267 uid: string | undefined;
268 /** Server checksum of the resource. */
269 etag: string | undefined;
270 /** RFC3339 creation timestamp. */
271 createTime: string | undefined;
272 /** RFC3339 last-update timestamp. */
273 updateTime: string | undefined;
274 },
275 never,
276 Providers
277>;
278
279/**
280 * An Eventarc Advanced pipeline that delivers matched enrollment
281 * messages to a destination (HTTP, Workflows, Pub/Sub, or another
282 * MessageBus), optionally transforming and retrying along the way.
283 *
284 * `pipelineId` and `location` are identity — changing either replaces
285 * the pipeline. Destinations, mediations, retry policy, logging, CMEK,
286 * display name, labels, and annotations update in place.
287 *
288 * ### Creating a Pipeline
289 * **Example:** Publish to a Pub/Sub topic
290 * ```typescript
291 * const topic = yield* GCP.PubSub.Topic("Sink", {});
292 * const pipeline = yield* GCP.Eventarc.Pipeline("Forward", {
293 * location: "us-central1",
294 * destinations: [{ topic: topic.name }],
295 * labels: { env: "prod" },
296 * });
297 * ```
298 *
299 * **Example:** HTTPS destination with OIDC
300 * ```typescript
301 * const pipeline = yield* GCP.Eventarc.Pipeline("HttpSink", {
302 * location: "us-central1",
303 * destinations: [{
304 * httpEndpoint: { uri: "https://service-xyz-uc.a.run.app" },
305 * authenticationConfig: {
306 * googleOidc: { serviceAccount: sa.email },
307 * },
308 * }],
309 * });
310 * ```
311 *
312 * ### Updating a Pipeline
313 * Re-declare the same logical id with changed props; the engine keeps the
314 * physical pipeline and patches it in place.
315 *
316 * **Example:** Change labels and retry policy
317 * ```typescript
318 * const pipeline = yield* GCP.Eventarc.Pipeline("Forward", {
319 * location: "us-central1",
320 * destinations: [{ topic: topic.name }],
321 * retryPolicy: { maxAttempts: 8 },
322 * labels: { env: "prod", role: "pipeline" },
323 * });
324 * ```
325 *
326 * @resource
327 * @category Eventarc
328 */
329export const Pipeline = Resource<Pipeline>("GCP.Eventarc.Pipeline");
330
331const toPayloadFormat = (
332 format:
333 | PipelinePayloadFormat
334 | eventarc.GoogleCloudEventarcV1PipelineMessagePayloadFormat
335 | undefined,
336): PipelinePayloadFormat | undefined => {
337 if (format === undefined) return undefined;
338 const next: PipelinePayloadFormat = {};
339 if (format.protobuf) {
340 next.protobuf = {
341 schemaDefinition: format.protobuf.schemaDefinition,
342 };
343 }
344 if (format.avro) {
345 next.avro = { schemaDefinition: format.avro.schemaDefinition };
346 }
347 if (format.json !== undefined) {
348 next.json = {};
349 }
350 return next.protobuf || next.avro || next.json ? next : undefined;
351};
352
353const toRetryPolicy = (
354 policy:
355 | PipelineRetryPolicy
356 | eventarc.GoogleCloudEventarcV1PipelineRetryPolicy
357 | undefined,
358): PipelineRetryPolicy | undefined => {
359 if (policy === undefined) return undefined;
360 if (
361 policy.maxAttempts === undefined &&
362 policy.minRetryDelay === undefined &&
363 policy.maxRetryDelay === undefined
364 ) {
365 return undefined;
366 }
367 return {
368 maxAttempts: policy.maxAttempts,
369 minRetryDelay: policy.minRetryDelay,
370 maxRetryDelay: policy.maxRetryDelay,
371 };
372};
373
374const toMediations = (
375 mediations:
376 | readonly PipelineMediation[]
377 | eventarc.GoogleCloudEventarcV1PipelineMediationList
378 | undefined,
379): PipelineMediation[] | undefined => {
380 if (mediations === undefined || mediations.length === 0) return undefined;
381 return mediations.map((mediation) => ({
382 transformation: mediation.transformation
383 ? {
384 transformationTemplate:
385 mediation.transformation.transformationTemplate,
386 }
387 : undefined,
388 }));
389};
390
391const toDestination = (
392 destination:
393 | PipelineDestination
394 | eventarc.GoogleCloudEventarcV1PipelineDestination,
395 project: string,
396 location: string,
397): PipelineDestination => {
398 const auth = destination.authenticationConfig;
399 return {
400 httpEndpoint: destination.httpEndpoint
401 ? {
402 uri: destination.httpEndpoint.uri ?? "",
403 messageBindingTemplate:
404 destination.httpEndpoint.messageBindingTemplate,
405 }
406 : undefined,
407 workflow: destination.workflow
408 ? expandResource(destination.workflow, project, location, "workflows")
409 : undefined,
410 topic: destination.topic
411 ? expandTopic(destination.topic, project)
412 : undefined,
413 messageBus: destination.messageBus
414 ? expandResource(
415 destination.messageBus,
416 project,
417 location,
418 "messageBuses",
419 )
420 : undefined,
421 networkConfig: destination.networkConfig?.networkAttachment
422 ? { networkAttachment: destination.networkConfig.networkAttachment }
423 : undefined,
424 authenticationConfig: auth
425 ? {
426 oauthToken: auth.oauthToken
427 ? {
428 serviceAccount: auth.oauthToken.serviceAccount,
429 scope: auth.oauthToken.scope,
430 }
431 : undefined,
432 googleOidc: auth.googleOidc
433 ? {
434 serviceAccount: auth.googleOidc.serviceAccount,
435 audience: auth.googleOidc.audience,
436 }
437 : undefined,
438 }
439 : undefined,
440 outputPayloadFormat: toPayloadFormat(destination.outputPayloadFormat),
441 };
442};
443
444const toDestinations = (
445 destinations:
446 | readonly PipelineDestination[]
447 | eventarc.GoogleCloudEventarcV1PipelineDestinationList
448 | undefined,
449 project: string,
450 location: string,
451): PipelineDestination[] =>
452 (destinations ?? []).map((destination) =>
453 toDestination(destination, project, location),
454 );
455
456const toAttrs = (
457 pipeline: eventarc.Pipeline,
458 project: string,
459 region: string,
460) => {
461 const name = pipeline.name ?? "";
462 const parsed = parseName(name, COLLECTION, region);
463 const location = parsed.location;
464 const resolvedProject = parsed.project || project;
465 return {
466 name,
467 pipelineId: parsed.id,
468 project: resolvedProject,
469 location,
470 destinations: toDestinations(
471 pipeline.destinations,
472 resolvedProject,
473 location,
474 ),
475 inputPayloadFormat: toPayloadFormat(pipeline.inputPayloadFormat),
476 mediations: toMediations(pipeline.mediations),
477 retryPolicy: toRetryPolicy(pipeline.retryPolicy),
478 displayName: pipeline.displayName,
479 cryptoKeyName: pipeline.cryptoKeyName,
480 loggingConfig: toLoggingConfig(pipeline.loggingConfig),
481 labels: userLabels(pipeline.labels),
482 annotations: userAnnotations(pipeline.annotations),
483 satisfiesPzs: pipeline.satisfiesPzs,
484 uid: pipeline.uid,
485 etag: pipeline.etag,
486 createTime: pipeline.createTime,
487 updateTime: pipeline.updateTime,
488 };
489};
490
491const getByName = (name: string) =>
492 eventarc
493 .getProjectsLocationsPipelines({ name })
494 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
495
496export const PipelineProvider = () =>
497 Provider.succeed(Pipeline, {
498 stables: ["name", "pipelineId", "project", "location", "uid", "createTime"],
499
500 diff: Effect.fn(function* ({ news, olds, output }) {
501 if (!isResolved(news)) return undefined;
502 const env = yield* GcpEnvironment.current;
503 const previousId = olds?.pipelineId ?? output?.pipelineId;
504 const nextId = news.pipelineId
505 ? rfc1035(news.pipelineId, "pipeline")
506 : previousId;
507 const previousLocation = normalizeLocation(
508 olds?.location ?? output?.location,
509 env.region,
510 );
511 const nextLocation = normalizeLocation(
512 news.location ?? olds?.location ?? output?.location,
513 env.region,
514 );
515 if (
516 (previousId !== undefined &&
517 nextId !== undefined &&
518 nextId !== previousId) ||
519 previousLocation !== nextLocation
520 ) {
521 return { action: "replace" as const };
522 }
523 return undefined;
524 }),
525
526 read: Effect.fn(function* ({ id, olds, output }) {
527 const env = yield* GcpEnvironment.current;
528 const pipelineId = yield* toPhysicalId(
529 id,
530 olds?.pipelineId,
531 output?.pipelineId,
532 "pipeline",
533 );
534 const location = normalizeLocation(
535 olds?.location ?? output?.location,
536 env.region,
537 );
538 const name =
539 output?.name ??
540 resourceName(env.project, location, COLLECTION, pipelineId);
541 const existing = yield* getByName(name);
542 if (existing === undefined) return undefined;
543 const attrs = toAttrs(existing, env.project, env.region);
544 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
545 ? attrs
546 : Unowned(attrs);
547 }),
548
549 list: () =>
550 Effect.gen(function* () {
551 const env = yield* GcpEnvironment.current;
552 const items = yield* collectPages(
553 eventarc.listProjectsLocationsPipelines.pages({
554 parent: parentOf(env.project, "-"),
555 pageSize: 1000,
556 }),
557 (page) => page.pipelines,
558 );
559 return items
560 .filter((item) => hasAlchemyLabelKeys(item.labels))
561 .map((item) => toAttrs(item, env.project, env.region));
562 }),
563
564 reconcile: Effect.fn(function* ({ id, news, output }) {
565 const env = yield* GcpEnvironment.current;
566 const pipelineId = yield* toPhysicalId(
567 id,
568 news.pipelineId,
569 output?.pipelineId,
570 "pipeline",
571 );
572 const location = normalizeLocation(
573 news.location ?? output?.location,
574 env.region,
575 );
576 const name = resourceName(env.project, location, COLLECTION, pipelineId);
577 const desiredLabels = {
578 ...toLabels(news.labels),
579 ...(yield* createInternalLabels(id)),
580 };
581 const destinations = toDestinations(
582 news.destinations,
583 env.project,
584 location,
585 );
586 const desiredLogging = toLoggingConfig(news.loggingConfig);
587 const desiredCrypto =
588 news.cryptoKeyName && news.cryptoKeyName.length > 0
589 ? news.cryptoKeyName
590 : undefined;
591 const desiredAnnotations = news.annotations
592 ? tagRecord(news.annotations)
593 : undefined;
594 const desiredFormat = toPayloadFormat(news.inputPayloadFormat);
595 const desiredMediations = toMediations(news.mediations);
596 const desiredRetry = toRetryPolicy(news.retryPolicy);
597
598 let current = yield* getByName(name);
599
600 if (current === undefined) {
601 const created = yield* eventarc
602 .createProjectsLocationsPipelines({
603 parent: parentOf(env.project, location),
604 pipelineId,
605 body: compact({
606 name,
607 destinations,
608 inputPayloadFormat: desiredFormat,
609 mediations: desiredMediations,
610 retryPolicy: desiredRetry,
611 displayName: news.displayName,
612 cryptoKeyName: desiredCrypto,
613 loggingConfig: desiredLogging,
614 labels: desiredLabels,
615 annotations: desiredAnnotations,
616 }),
617 })
618 .pipe(
619 Effect.retry({
620 while: (error) => error._tag === "Conflict",
621 times: 5,
622 schedule: Schedule.spaced("2 seconds"),
623 }),
624 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
625 );
626 if (created !== undefined) {
627 yield* waitForOperation(created);
628 }
629 current = yield* waitUntilPresent(getByName(name), name);
630 }
631
632 const observed = toAttrs(current, env.project, env.region);
633 const observedLabels = tagRecord(current.labels);
634 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
635 const updateMask = changedFields([
636 ["labels", upsert.length > 0 || removed.length > 0],
637 ["destinations", !sameJson(observed.destinations, destinations)],
638 [
639 "inputPayloadFormat",
640 !sameJson(observed.inputPayloadFormat, desiredFormat),
641 ],
642 ["mediations", !sameJson(observed.mediations, desiredMediations)],
643 ["retryPolicy", !sameJson(observed.retryPolicy, desiredRetry)],
644 [
645 "displayName",
646 textKey(current.displayName) !== textKey(news.displayName),
647 ],
648 [
649 "cryptoKeyName",
650 cryptoKeyKey(current.cryptoKeyName) !== cryptoKeyKey(desiredCrypto),
651 ],
652 [
653 "loggingConfig",
654 loggingKey(current.loggingConfig) !== loggingKey(desiredLogging),
655 ],
656 [
657 "annotations",
658 !sameJson(tagRecord(current.annotations), desiredAnnotations ?? {}),
659 ],
660 ]);
661
662 if (updateMask.length > 0) {
663 const patched = yield* eventarc.patchProjectsLocationsPipelines({
664 name: current.name ?? name,
665 updateMask: updateMask.join(","),
666 body: {
667 name: current.name ?? name,
668 destinations,
669 inputPayloadFormat: desiredFormat,
670 mediations: desiredMediations,
671 retryPolicy: desiredRetry,
672 displayName: news.displayName,
673 cryptoKeyName: desiredCrypto,
674 loggingConfig: desiredLogging,
675 labels: desiredLabels,
676 annotations: desiredAnnotations ?? {},
677 },
678 });
679 yield* waitForOperation(patched);
680 current = yield* waitUntilPresent(
681 getByName(current.name ?? name),
682 current.name ?? name,
683 );
684 }
685
686 return toAttrs(current, env.project, env.region);
687 }),
688
689 delete: Effect.fn(function* ({ output }) {
690 yield* retryOnTransient(
691 Effect.gen(function* () {
692 const existing = yield* getByName(output.name);
693 if (existing === undefined) return;
694 const operation = yield* eventarc
695 .deleteProjectsLocationsPipelines({
696 name: output.name,
697 allowMissing: true,
698 })
699 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
700 if (operation !== undefined) {
701 yield* waitForOperation(operation, { notFoundOk: true });
702 }
703 }),
704 ).pipe(
705 Effect.catchTag("GCP.OperationFailed", (error) =>
706 getByName(output.name).pipe(
707 Effect.flatMap((current) =>
708 current === undefined ? Effect.void : Effect.fail(error),
709 ),
710 ),
711 ),
712 );
713 yield* waitUntilGone(getByName(output.name), output.name);
714 }),
715 });