1import * as eventbridge from "@distilled.cloud/aws/eventbridge";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import {
11 createInternalTags,
12 createTagsList,
13 diffTags,
14 hasAlchemyTags,
15} from "../../Tags.ts";
16import { AWSEnvironment, type AccountID } from "../Environment.ts";
17import type { Providers } from "../Providers.ts";
18import type { RegionID } from "../Region.ts";
19import type { QueueArn } from "../SQS/Queue.ts";
20
21export type {
22 IncludeDetail,
23 Level,
24 LogConfig,
25} from "@distilled.cloud/aws/eventbridge";
26
27export type EventBusName = string;
28export type EventBusArn =
29 `arn:aws:events:${RegionID}:${AccountID}:event-bus/${EventBusName}`;
30
31export interface EventBusDeadLetterConfig {
32 /** ARN of the SQS queue used as the dead-letter queue. */
33 Arn?: QueueArn;
34}
35
36export interface EventBusProps {
37 /**
38 * Name of the event bus. Must match [/\.\-_A-Za-z0-9]+, 1-256 characters.
39 * If omitted, a unique name will be generated.
40 * Cannot be "default" — use the default event bus by omitting eventBusName on rules.
41 */
42 name?: EventBusName;
43
44 /**
45 * The partner event source to associate with this event bus.
46 * Only used when creating a partner event bus.
47 */
48 eventSourceName?: string;
49
50 /**
51 * Description of the event bus.
52 */
53 description?: string;
54
55 /**
56 * The identifier of the KMS customer managed key for EventBridge to use
57 * to encrypt events on this event bus.
58 */
59 kmsKeyIdentifier?: string;
60
61 /**
62 * Dead-letter queue configuration for undeliverable events.
63 */
64 deadLetterConfig?: EventBusDeadLetterConfig;
65
66 /**
67 * Logging configuration for the event bus.
68 */
69 logConfig?: eventbridge.LogConfig;
70
71 /**
72 * Whether to delete any rules remaining on the bus (removing their
73 * targets first) when the bus is destroyed. Rules managed by the same
74 * stack are always deleted before the bus by the engine, and AWS-managed
75 * rules (e.g. the hidden archival rule an Archive leaves behind while
76 * AWS's async cleanup lags) are always force-swept; `forceDestroy`
77 * additionally sweeps rules created out-of-band (or leaked by an
78 * interrupted deploy), which would otherwise block bus deletion with
79 * `EventBusHasRules` forever.
80 * @default false
81 */
82 forceDestroy?: boolean;
83
84 /**
85 * Tags to assign to the event bus.
86 */
87 tags?: Record<string, string>;
88}
89
90/**
91 * An Amazon EventBridge event bus for receiving and routing events.
92 * ### Creating Event Buses
93 * **Example:** Custom Event Bus
94 * ```typescript
95 * const bus = yield* EventBus("MyAppEvents", {
96 * description: "Custom event bus for my application",
97 * });
98 * ```
99 *
100 * **Example:** Event Bus with Dead Letter Queue
101 * ```typescript
102 * const bus = yield* EventBus("ReliableBus", {
103 * deadLetterConfig: {
104 * Arn: yield* dlq.queueArn,
105 * },
106 * });
107 * ```
108 *
109 * **Example:** Event Bus with KMS Encryption
110 * ```typescript
111 * const bus = yield* EventBus("EncryptedBus", {
112 * kmsKeyIdentifier: yield* key.keyArn(),
113 * });
114 * ```
115 *
116 * @resource
117 */
118export interface EventBus extends Resource<
119 "AWS.EventBridge.EventBus",
120 EventBusProps,
121 {
122 /** The name of the event bus. */
123 eventBusName: EventBusName;
124 /** The ARN of the event bus. */
125 eventBusArn: EventBusArn;
126 /** Description of the event bus, if set. */
127 description?: string;
128 },
129 never,
130 Providers
131> {}
132export const EventBus = Resource<EventBus>("AWS.EventBridge.EventBus");
133
134export const EventBusProvider = () =>
135 Provider.effect(
136 EventBus,
137 Effect.gen(function* () {
138 const createEventBusName = (id: string, props: { name?: string } = {}) =>
139 Effect.gen(function* () {
140 if (props.name) {
141 return props.name;
142 }
143 return yield* createPhysicalName({
144 id,
145 maxLength: 256,
146 });
147 });
148
149 return {
150 stables: ["eventBusName", "eventBusArn"],
151 diff: Effect.fn(function* ({ id, news, olds }) {
152 if (!isResolved(news)) return;
153 const oldName = yield* createEventBusName(id, olds);
154 const newName = yield* createEventBusName(id, news);
155 if (oldName !== newName) {
156 return { action: "replace" } as const;
157 }
158 if ((olds.eventSourceName ?? "") !== (news.eventSourceName ?? "")) {
159 return { action: "replace" } as const;
160 }
161 }),
162 read: Effect.fn(function* ({ id, olds, output }) {
163 const eventBusName =
164 output?.eventBusName ?? (yield* createEventBusName(id, olds ?? {}));
165 const described = yield* eventbridge
166 .describeEventBus({
167 Name: eventBusName,
168 })
169 .pipe(
170 Effect.catchTag("ResourceNotFoundException", () =>
171 Effect.succeed(undefined),
172 ),
173 );
174
175 if (!described?.Arn || !described.Name) {
176 return undefined;
177 }
178
179 const { Tags } = yield* eventbridge.listTagsForResource({
180 ResourceARN: described.Arn,
181 });
182 const attrs = {
183 eventBusName: described.Name,
184 eventBusArn: described.Arn as EventBusArn,
185 description: described.Description,
186 };
187 return (yield* hasAlchemyTags(id, Tags ?? []))
188 ? attrs
189 : Unowned(attrs);
190 }),
191 list: () =>
192 Effect.gen(function* () {
193 // Enumerate every event bus in the ambient account/region via
194 // manual NextToken pagination (listEventBuses is not a paginated
195 // distilled op). The AWS-managed `default` bus is excluded — the
196 // EventBus resource cannot manage it (name "default" is reserved).
197 const attrs: {
198 eventBusName: EventBusName;
199 eventBusArn: EventBusArn;
200 description?: string;
201 }[] = [];
202 let nextToken: string | undefined;
203 do {
204 const page = yield* eventbridge.listEventBuses({
205 NextToken: nextToken,
206 });
207 for (const bus of page.EventBuses ?? []) {
208 if (!bus.Name || !bus.Arn || bus.Name === "default") {
209 continue;
210 }
211 attrs.push({
212 eventBusName: bus.Name,
213 eventBusArn: bus.Arn as EventBusArn,
214 description: bus.Description,
215 });
216 }
217 nextToken = page.NextToken;
218 } while (nextToken);
219 return attrs;
220 }),
221 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
222 const { accountId, region } = yield* AWSEnvironment.current;
223 const eventBusName =
224 output?.eventBusName ?? (yield* createEventBusName(id, news));
225 const eventBusArn = (output?.eventBusArn ??
226 `arn:aws:events:${region}:${accountId}:event-bus/${eventBusName}`) as EventBusArn;
227 const internalTags = yield* createInternalTags(id);
228 const desiredTags = {
229 ...internalTags,
230 ...(news.tags as Record<string, string> | undefined),
231 };
232
233 // Observe — fetch live cloud state. We don't trust `output`
234 // blindly: a bus deleted out of band shows up as missing and we
235 // recreate. Foreign-tagged buses have already been screened by
236 // `read` upstream.
237 let described = yield* eventbridge
238 .describeEventBus({
239 Name: eventBusName,
240 })
241 .pipe(
242 Effect.catchTag("ResourceNotFoundException", () =>
243 Effect.succeed(undefined),
244 ),
245 );
246
247 // Ensure — create the bus if missing. Tolerate
248 // `ResourceAlreadyExistsException` as a race with a peer
249 // reconciler: re-read and continue with the sync path.
250 if (!described?.Arn) {
251 yield* eventbridge
252 .createEventBus({
253 Name: eventBusName,
254 EventSourceName: news.eventSourceName,
255 Description: news.description,
256 KmsKeyIdentifier: news.kmsKeyIdentifier as string | undefined,
257 DeadLetterConfig: news.deadLetterConfig
258 ? { Arn: news.deadLetterConfig.Arn as string | undefined }
259 : undefined,
260 LogConfig: news.logConfig,
261 Tags: createTagsList(desiredTags),
262 })
263 .pipe(
264 Effect.catchTag(
265 "ResourceAlreadyExistsException",
266 () => Effect.void,
267 ),
268 );
269
270 described = yield* eventbridge
271 .describeEventBus({
272 Name: eventBusName,
273 })
274 .pipe(
275 Effect.catchTag("ResourceNotFoundException", () =>
276 Effect.succeed(undefined),
277 ),
278 );
279 }
280
281 // Sync mutable bus configuration — `updateEventBus` overwrites
282 // `description`, KMS key, DLQ, and log config in one shot, so we
283 // call it unconditionally (idempotent for matching values).
284 yield* eventbridge.updateEventBus({
285 Name: eventBusName,
286 Description: news.description,
287 KmsKeyIdentifier: news.kmsKeyIdentifier as string | undefined,
288 DeadLetterConfig: news.deadLetterConfig
289 ? { Arn: news.deadLetterConfig.Arn as string | undefined }
290 : undefined,
291 LogConfig: news.logConfig,
292 });
293
294 // Sync tags — diff observed cloud tags against desired. Adoption
295 // may bring us a bus with its own tag set; diffing against the
296 // freshly-fetched tags lets the reconciler converge regardless.
297 const observedTagsList = yield* eventbridge
298 .listTagsForResource({
299 ResourceARN: eventBusArn,
300 })
301 .pipe(Effect.map((r) => r.Tags ?? []));
302 const observedTags: Record<string, string> = {};
303 for (const tag of observedTagsList) {
304 if (tag.Key && tag.Value !== undefined) {
305 observedTags[tag.Key] = tag.Value;
306 }
307 }
308 const { removed, upsert } = diffTags(observedTags, desiredTags);
309
310 if (removed.length > 0) {
311 yield* eventbridge.untagResource({
312 ResourceARN: eventBusArn,
313 TagKeys: removed,
314 });
315 }
316
317 if (upsert.length > 0) {
318 yield* eventbridge.tagResource({
319 ResourceARN: eventBusArn,
320 Tags: upsert,
321 });
322 }
323
324 yield* session.note(eventBusArn);
325 return {
326 eventBusName,
327 eventBusArn,
328 description: news.description,
329 };
330 }),
331 delete: Effect.fn(function* ({ olds = {}, output }) {
332 const eventBusName = output.eventBusName;
333
334 // Rules the engine knows about are deleted first (they depend on
335 // the bus), but AWS-managed rules can linger: deleting an archive
336 // removes its hidden archival rule *asynchronously*, sometimes
337 // minutes later, during which deleteEventBus keeps failing with
338 // EventBusHasRules. Sweep remaining rules with Force (required for
339 // managed rules) instead of waiting out AWS's async cleanup:
340 // AWS-managed rules (`ManagedBy` set) are always swept; rules
341 // created out-of-band by the user are only swept when
342 // `forceDestroy` is enabled.
343 yield* Effect.gen(function* () {
344 let nextToken: string | undefined;
345 do {
346 const page = yield* eventbridge.listRules({
347 EventBusName: eventBusName,
348 NextToken: nextToken,
349 });
350 for (const rule of page.Rules ?? []) {
351 if (!rule.Name) continue;
352 if (!rule.ManagedBy && !olds.forceDestroy) continue;
353 const targets = yield* eventbridge
354 .listTargetsByRule({
355 Rule: rule.Name,
356 EventBusName: eventBusName,
357 })
358 .pipe(
359 Effect.catchTag("ResourceNotFoundException", () =>
360 Effect.succeed({ Targets: [] }),
361 ),
362 );
363 const targetIds = (targets.Targets ?? []).map((t) => t.Id);
364 if (targetIds.length > 0) {
365 yield* eventbridge
366 .removeTargets({
367 Rule: rule.Name,
368 EventBusName: eventBusName,
369 Ids: targetIds,
370 Force: true,
371 })
372 .pipe(
373 Effect.catchTag(
374 ["ResourceNotFoundException", "ManagedRuleException"],
375 () => Effect.void,
376 ),
377 );
378 }
379 yield* eventbridge
380 .deleteRule({
381 Name: rule.Name,
382 EventBusName: eventBusName,
383 Force: true,
384 })
385 .pipe(
386 Effect.catchTag(
387 ["ResourceNotFoundException", "ManagedRuleException"],
388 () => Effect.void,
389 ),
390 );
391 }
392 nextToken = page.NextToken;
393 } while (nextToken);
394 }).pipe(
395 // Bus already gone (or vanishes mid-sweep) — nothing to sweep.
396 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
397 );
398
399 // The sweep and EventBridge's own bookkeeping are eventually
400 // consistent — a just-deleted rule can still count against the bus
401 // for a few seconds. Retry the typed dependency violation on a
402 // bounded schedule.
403 yield* eventbridge
404 .deleteEventBus({
405 Name: eventBusName,
406 })
407 .pipe(
408 Effect.retry({
409 while: (e): boolean => e._tag === "EventBusHasRules",
410 schedule: Schedule.spaced("3 seconds"),
411 times: 10,
412 }),
413 // Bus already gone (deleted out-of-band, or a previous
414 // destroy partially succeeded) — delete is idempotent.
415 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
416 );
417 }),
418 };
419 }),
420 );