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