1import * as eventarc from "@distilled.cloud/gcp/eventarc_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.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 stripInternalLabels,
18 toLabels,
19} from "../Labels.ts";
20import type { Providers } from "../Providers.ts";
21import { waitForOperation } from "./internal.ts";
22
23const MAX_NAME_LENGTH = 63;
24
25export type ChannelState = eventarc.ChannelStateEnum | (string & {});
26
27export type ChannelProps = {
28 /**
29 * Channel id (the `{channel}` segment of
30 * `projects/{project}/locations/{location}/channels/{channel}`). If
31 * omitted, a unique name is generated from the stack, stage, and
32 * logical id. Must match `[a-z]([a-z0-9-]*[a-z0-9])?` and be 1-63
33 * characters. Immutable — changing it replaces the channel.
34 */
35 channelId?: string;
36 /**
37 * Eventarc location (`us-central1`, `us-east1`, …). Immutable —
38 * changing it replaces the channel. `US-CENTRAL1` is accepted and
39 * normalized to `us-central1`.
40 * @default the stack's GCP region (`GCP.Region`, else the profile
41 * region, else `us-central1`)
42 */
43 location?: string;
44 /**
45 * Event provider associated with the channel (an Eventarc SaaS
46 * partner). Format
47 * `projects/{project}/locations/{location}/providers/{providerId}` or
48 * the `{providerId}` segment. The provider is granted permission to
49 * publish events on the channel. Omit for a Google channel with no
50 * third-party partner. Immutable — changing it replaces the channel.
51 */
52 provider?: string;
53 /**
54 * Customer-managed Cloud KMS key used to encrypt/decrypt event data,
55 * as
56 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
57 */
58 cryptoKeyName?: string;
59 /**
60 * User labels. Alchemy ownership labels are merged in automatically.
61 */
62 labels?: Record<string, string>;
63};
64
65export type Channel = Resource<
66 "GCP.Eventarc.Channel",
67 ChannelProps,
68 {
69 /** Full resource name `projects/{project}/locations/{location}/channels/{channel}`. */
70 name: string;
71 /** Channel id (last path segment). */
72 channelId: string;
73 /** Project id. */
74 project: string;
75 /** Location id (`us-central1`, …). */
76 location: string;
77 /** Event provider resource name, if any. */
78 provider: string | undefined;
79 /** Customer-managed KMS key, if any. */
80 cryptoKeyName: string | undefined;
81 /** User labels (Alchemy ownership labels stripped). */
82 labels: Record<string, string>;
83 /**
84 * Pub/Sub topic Eventarc manages as the channel transport
85 * (`projects/{project}/topics/{topic}`).
86 */
87 pubsubTopic: string | undefined;
88 /** Channel lifecycle state (`PENDING`, `ACTIVE`, `INACTIVE`, …). */
89 state: ChannelState | undefined;
90 /**
91 * Activation token the provider uses to register the channel. Present
92 * while the channel is `PENDING`.
93 */
94 activationToken: string | undefined;
95 /** Server-assigned UUID4, stable until delete. */
96 uid: string | undefined;
97 /** RFC3339 creation timestamp. */
98 createTime: string | undefined;
99 /** RFC3339 last-update timestamp. */
100 updateTime: string | undefined;
101 /** Whether the channel satisfies physical zone separation. */
102 satisfiesPzs: boolean | undefined;
103 },
104 never,
105 Providers
106>;
107
108/**
109 * An Eventarc channel representing a subscriber's intent to receive
110 * events from an event provider. Published events are delivered over
111 * the transport Eventarc associates with the channel.
112 *
113 * `channelId`, `location`, and `provider` are identity — changing any
114 * of them replaces the channel. Labels and `cryptoKeyName` update in
115 * place. A Google channel (no third-party `provider`) is valid; SaaS
116 * partner channels stay `PENDING` until the partner activates them.
117 *
118 * ### Creating a Channel
119 * **Example:** Google channel
120 * ```typescript
121 * const channel = yield* GCP.Eventarc.Channel("events", {
122 * location: "us-central1",
123 * labels: { env: "prod" },
124 * });
125 * ```
126 *
127 * **Example:** Third-party provider channel
128 * ```typescript
129 * const channel = yield* GCP.Eventarc.Channel("datadog", {
130 * channelId: "datadog-events",
131 * location: "us-central1",
132 * provider:
133 * "projects/my-project/locations/us-central1/providers/datadog",
134 * });
135 * ```
136 *
137 * ### Updating a Channel
138 * Re-declare the same logical id with changed props; the engine keeps the
139 * physical channel and patches it in place.
140 *
141 * **Example:** Change labels and CMEK
142 * ```typescript
143 * const channel = yield* GCP.Eventarc.Channel("events", {
144 * location: "us-central1",
145 * cryptoKeyName:
146 * "projects/my-project/locations/us-central1/keyRings/keys/cryptoKeys/events",
147 * labels: { env: "prod", role: "events" },
148 * });
149 * ```
150 *
151 * @resource
152 * @category Eventarc
153 */
154export const Channel = Resource<Channel>("GCP.Eventarc.Channel");
155
156export class ChannelNotResolved extends Data.TaggedError(
157 "GCP.Eventarc.ChannelNotResolved",
158)<{
159 name: string;
160}> {}
161
162export class ChannelStillExists extends Data.TaggedError(
163 "GCP.Eventarc.ChannelStillExists",
164)<{
165 name: string;
166}> {}
167
168const lastSegment = (value: string) => {
169 const trimmed = value.replace(/\/+$/, "");
170 const parts = trimmed.split("/");
171 return parts[parts.length - 1] || trimmed;
172};
173
174const normalizeLocation = (
175 location: string | undefined,
176 defaultLocation: string,
177) => lastSegment(location ?? defaultLocation).toLowerCase();
178
179const rfc1035 = (name: string): string => {
180 let next = name
181 .toLowerCase()
182 .replace(/[^a-z0-9-]/g, "-")
183 .replace(/-+/g, "-")
184 .replace(/^-+|-+$/g, "");
185 if (!/^[a-z]/.test(next)) next = `c${next}`;
186 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
187 return next.length > 0 ? next : "channel";
188};
189
190const resourceName = (project: string, location: string, channelId: string) =>
191 `projects/${project}/locations/${location}/channels/${channelId}`;
192
193const parseName = (name: string, defaultLocation: string) => {
194 const parts = name.split("/").filter((part) => part.length > 0);
195 const channelsAt = parts.lastIndexOf("channels");
196 const locationsAt = parts.lastIndexOf("locations");
197 const projectsAt = parts.lastIndexOf("projects");
198 return {
199 project:
200 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
201 location:
202 locationsAt >= 0 && parts[locationsAt + 1]
203 ? parts[locationsAt + 1]!
204 : defaultLocation,
205 channelId:
206 channelsAt >= 0 && parts[channelsAt + 1]
207 ? parts[channelsAt + 1]!
208 : lastSegment(name),
209 };
210};
211
212const userLabels = (
213 labels: Record<string, string | undefined> | null | undefined,
214): Record<string, string> => stripInternalLabels(tagRecord(labels));
215
216const toId = (id: string, channelId: string | undefined, existing?: string) =>
217 Effect.gen(function* () {
218 if (channelId !== undefined) return rfc1035(channelId);
219 if (existing !== undefined) return existing;
220 return rfc1035(
221 yield* createPhysicalName({
222 id,
223 maxLength: MAX_NAME_LENGTH,
224 lowercase: true,
225 }),
226 );
227 });
228
229const compact = <T extends Record<string, unknown>>(value: T): T => {
230 const next: Record<string, unknown> = {};
231 for (const [key, entry] of Object.entries(value)) {
232 if (entry !== undefined && entry !== "") next[key] = entry;
233 }
234 return next as T;
235};
236
237const providerKey = (
238 provider: string | undefined,
239 project: string,
240 location: string,
241) => {
242 if (provider === undefined || provider.length === 0) return "";
243 if (provider.includes("/")) return provider;
244 return `projects/${project}/locations/${location}/providers/${provider}`;
245};
246
247const cryptoKeyKey = (name: string | undefined) =>
248 name === undefined || name.length === 0 ? "" : name;
249
250const toAttrs = (
251 channel: eventarc.Channel,
252 project: string,
253 region: string,
254) => {
255 const name = channel.name ?? "";
256 const parsed = parseName(name, region);
257 return {
258 name,
259 channelId: parsed.channelId,
260 project: parsed.project || project,
261 location: parsed.location,
262 provider: channel.provider,
263 cryptoKeyName: channel.cryptoKeyName,
264 labels: userLabels(channel.labels),
265 pubsubTopic: channel.pubsubTopic,
266 state: channel.state,
267 activationToken: channel.activationToken,
268 uid: channel.uid,
269 createTime: channel.createTime,
270 updateTime: channel.updateTime,
271 satisfiesPzs: channel.satisfiesPzs,
272 };
273};
274
275const getByName = (name: string) =>
276 eventarc
277 .getProjectsLocationsChannels({ name })
278 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
279
280const waitUntilExists = (name: string) =>
281 getByName(name).pipe(
282 Effect.flatMap((channel) =>
283 channel
284 ? Effect.succeed(channel)
285 : Effect.fail(new ChannelNotResolved({ name })),
286 ),
287 Effect.retry({
288 while: (error) => error._tag === "GCP.Eventarc.ChannelNotResolved",
289 times: 8,
290 schedule: Schedule.spaced("1 second"),
291 }),
292 );
293
294const waitUntilGone = (name: string) =>
295 getByName(name).pipe(
296 Effect.flatMap((channel) =>
297 channel === undefined
298 ? Effect.void
299 : Effect.fail(new ChannelStillExists({ name })),
300 ),
301 Effect.retry({
302 while: (error) => error._tag === "GCP.Eventarc.ChannelStillExists",
303 times: 10,
304 schedule: Schedule.spaced("2 seconds"),
305 }),
306 );
307
308export const ChannelProvider = () =>
309 Provider.succeed(Channel, {
310 stables: [
311 "name",
312 "channelId",
313 "project",
314 "location",
315 "uid",
316 "createTime",
317 "provider",
318 ],
319
320 diff: Effect.fn(function* ({ news, olds, output }) {
321 if (!isResolved(news)) return undefined;
322 const env = yield* GcpEnvironment.current;
323
324 const previousId = olds?.channelId ?? output?.channelId;
325 const nextId = news.channelId ? rfc1035(news.channelId) : previousId;
326 const previousLocation = normalizeLocation(
327 olds?.location ?? output?.location,
328 env.region,
329 );
330 const nextLocation = normalizeLocation(
331 news.location ?? output?.location,
332 env.region,
333 );
334 const project = output?.project ?? "";
335 const previousProvider = providerKey(
336 olds?.provider ?? output?.provider,
337 project,
338 previousLocation,
339 );
340 const nextProvider =
341 news.provider !== undefined
342 ? providerKey(news.provider, project, nextLocation)
343 : previousProvider;
344
345 const replace =
346 (previousId !== undefined &&
347 nextId !== undefined &&
348 nextId !== previousId) ||
349 previousLocation !== nextLocation ||
350 previousProvider !== nextProvider;
351
352 if (!replace) return undefined;
353 return {
354 action: "replace" as const,
355 deleteFirst:
356 previousLocation === nextLocation &&
357 previousId !== undefined &&
358 nextId === previousId,
359 };
360 }),
361
362 read: Effect.fn(function* ({ id, olds, output }) {
363 const env = yield* GcpEnvironment.current;
364 const channelId = yield* toId(id, olds?.channelId, output?.channelId);
365 const location = normalizeLocation(
366 olds?.location ?? output?.location,
367 env.region,
368 );
369 const name =
370 output?.name ?? resourceName(env.project, location, channelId);
371 const existing = yield* getByName(name);
372 if (existing === undefined) return undefined;
373 const attrs = toAttrs(existing, env.project, env.region);
374 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
375 ? attrs
376 : Unowned(attrs);
377 }),
378
379 list: () =>
380 Effect.gen(function* () {
381 const env = yield* GcpEnvironment.current;
382 return yield* eventarc.listProjectsLocationsChannels
383 .pages({
384 parent: `projects/${env.project}/locations/-`,
385 pageSize: 1000,
386 })
387 .pipe(
388 Stream.flatMap((page) => Stream.fromIterable(page.channels ?? [])),
389 Stream.filter((channel) =>
390 Object.keys(channel.labels ?? {}).some((key) =>
391 key.startsWith("alchemy-"),
392 ),
393 ),
394 Stream.map((channel) => toAttrs(channel, env.project, env.region)),
395 Stream.runCollect,
396 Effect.map((chunk) => Array.from(chunk)),
397 Effect.catchTag("NotFound", () => Effect.succeed([])),
398 );
399 }),
400
401 reconcile: Effect.fn(function* ({ id, news, output }) {
402 const env = yield* GcpEnvironment.current;
403 const channelId = yield* toId(id, news.channelId, output?.channelId);
404 const location = normalizeLocation(
405 news.location ?? output?.location,
406 env.region,
407 );
408 const name = resourceName(env.project, location, channelId);
409 const desiredLabels = {
410 ...toLabels(news.labels),
411 ...(yield* createInternalLabels(id)),
412 };
413 const desiredProvider = news.provider
414 ? providerKey(news.provider, env.project, location)
415 : undefined;
416 const desiredCryptoKey =
417 news.cryptoKeyName && news.cryptoKeyName.length > 0
418 ? news.cryptoKeyName
419 : undefined;
420
421 let current = yield* getByName(name);
422
423 if (current === undefined) {
424 const created = yield* eventarc
425 .createProjectsLocationsChannels({
426 parent: `projects/${env.project}/locations/${location}`,
427 channelId,
428 body: compact({
429 name,
430 labels: desiredLabels,
431 provider: desiredProvider,
432 cryptoKeyName: desiredCryptoKey,
433 }),
434 })
435 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
436 if (created !== undefined) {
437 yield* waitForOperation(created);
438 }
439 current = yield* waitUntilExists(name);
440 }
441
442 if (current === undefined) {
443 return yield* new ChannelNotResolved({ name });
444 }
445
446 const observedLabels = tagRecord(current.labels);
447 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
448 const labelsChanged = upsert.length > 0 || removed.length > 0;
449 const cryptoKeyChanged =
450 cryptoKeyKey(current.cryptoKeyName) !== cryptoKeyKey(desiredCryptoKey);
451
452 if (labelsChanged || cryptoKeyChanged) {
453 const updateMask = [
454 labelsChanged ? "labels" : undefined,
455 cryptoKeyChanged ? "cryptoKeyName" : undefined,
456 ].filter((field): field is string => field !== undefined);
457
458 const patched = yield* eventarc.patchProjectsLocationsChannels({
459 name,
460 updateMask: updateMask.join(","),
461 body: {
462 name,
463 labels: desiredLabels,
464 cryptoKeyName: desiredCryptoKey,
465 },
466 });
467 yield* waitForOperation(patched);
468 current = yield* waitUntilExists(name);
469 }
470
471 return toAttrs(current, env.project, env.region);
472 }),
473
474 delete: Effect.fn(function* ({ output }) {
475 const existing = yield* getByName(output.name);
476 if (existing === undefined) return;
477 const operation = yield* eventarc
478 .deleteProjectsLocationsChannels({
479 name: output.name,
480 })
481 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
482 if (operation !== undefined) {
483 yield* waitForOperation(operation, { notFoundOk: true });
484 }
485 yield* waitUntilGone(output.name);
486 }),
487 });