1import * as notifications from "@distilled.cloud/aws/notifications";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10import { pinNotificationsRegion } from "./internal.ts";
11
12export interface EventRuleProps {
13 /**
14 * The ARN of the {@link NotificationConfiguration} this rule feeds.
15 * Changing the configuration replaces the rule.
16 */
17 notificationConfigurationArn: string;
18
19 /**
20 * The EventBridge event source, e.g. `aws.s3`, `aws.cloudwatch`,
21 * `aws.ec2`. Changing the source replaces the rule.
22 */
23 source: string;
24
25 /**
26 * The event type emitted by the source, e.g. `Object Created`,
27 * `CloudWatch Alarm State Change`, `EC2 Instance State-change
28 * Notification`. Changing the event type replaces the rule.
29 */
30 eventType: string;
31
32 /**
33 * Optional EventBridge event pattern that further restricts which events
34 * match. Provided as a JSON string or a plain pattern object. When
35 * omitted, all events of `source`/`eventType` match.
36 */
37 eventPattern?: string | Record<string, any>;
38
39 /**
40 * The AWS regions whose events this rule captures. User Notifications
41 * creates a managed EventBridge rule in each listed region.
42 */
43 regions: string[];
44}
45
46export interface EventRule extends Resource<
47 "AWS.Notifications.EventRule",
48 EventRuleProps,
49 {
50 /** The ARN of the event rule. */
51 eventRuleArn: string;
52 /** The ARN of the parent notification configuration. */
53 notificationConfigurationArn: string;
54 /** The EventBridge event source. */
55 source: string;
56 /** The event type. */
57 eventType: string;
58 /** The regions whose events the rule captures. */
59 regions: string[];
60 },
61 never,
62 Providers
63> {}
64
65/**
66 * An AWS User Notifications **event rule** — attaches an EventBridge
67 * source/event-type (optionally narrowed by an event pattern) to a
68 * {@link NotificationConfiguration}, across one or more regions.
69 *
70 * `source`, `eventType` and the parent configuration are immutable
71 * (changing them replaces the rule); `eventPattern` and `regions` update
72 * in place. User Notifications materializes a managed EventBridge rule in
73 * every listed region.
74 *
75 * ### Creating an Event Rule
76 * **Example:** Notify on S3 object creation
77 * ```typescript
78 * import * as Notifications from "alchemy/AWS/Notifications";
79 *
80 * const config = yield* Notifications.NotificationConfiguration("Alerts");
81 * const rule = yield* Notifications.EventRule("S3Created", {
82 * notificationConfigurationArn: config.notificationConfigurationArn,
83 * source: "aws.s3",
84 * eventType: "Object Created",
85 * regions: ["us-west-2"],
86 * });
87 * ```
88 *
89 * **Example:** Restrict matches with an event pattern
90 * ```typescript
91 * const rule = yield* Notifications.EventRule("BucketRule", {
92 * notificationConfigurationArn: config.notificationConfigurationArn,
93 * source: "aws.s3",
94 * eventType: "Object Created",
95 * eventPattern: { detail: { bucket: { name: ["my-bucket"] } } },
96 * regions: ["us-west-2", "us-east-2"],
97 * });
98 * ```
99 *
100 * @resource
101 */
102export const EventRule = Resource<EventRule>("AWS.Notifications.EventRule");
103
104/** Normalize an `eventPattern` prop to the wire JSON string ("" = none). */
105const toPatternString = (
106 pattern: string | Record<string, any> | undefined,
107): string =>
108 pattern === undefined
109 ? ""
110 : typeof pattern === "string"
111 ? pattern
112 : JSON.stringify(pattern);
113
114/** Order-insensitive region list equality. */
115const sameRegions = (a: readonly string[], b: readonly string[]): boolean =>
116 a.length === b.length && [...a].sort().join(",") === [...b].sort().join(",");
117
118/**
119 * Poll a rule until no region is in a transitional state (`CREATING` /
120 * `UPDATING` / `DELETING`). Bounded and fail-open — regional propagation
121 * completes in seconds and a still-pending region is not an error.
122 *
123 * Expressed as an explicitly-typed helper: inlining `Effect.repeat` in the
124 * provider leaves its conditional return type unresolved in declaration
125 * emit, widening the provider layer to an `unknown` R.
126 */
127const untilSettled = <A, E, R>(
128 self: Effect.Effect<A, E, R>,
129 pending: (a: A) => boolean,
130): Effect.Effect<A, E, R> =>
131 Effect.repeat(self, {
132 schedule: Schedule.fixed("2 seconds"),
133 until: (a) => !pending(a),
134 times: 20,
135 });
136
137/**
138 * Retry an operation while the service reports a transitional-lock
139 * `ConflictException` (e.g. "event rule ... has not reached terminal status
140 * in region(s): us-west-2 - DELETING"). User Notifications serializes rule
141 * mutations per configuration, so a create that races a sibling delete is
142 * rejected until the delete settles. Bounded (~40s).
143 */
144const retryWhileConflict = <A, E extends { _tag: string }, R>(
145 self: Effect.Effect<A, E, R>,
146): Effect.Effect<A, E, R> =>
147 Effect.retry(self, {
148 while: (e) => e._tag === "ConflictException",
149 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(20)]),
150 });
151
152/**
153 * Poll an observation until the resource is gone (bounded, fail-open).
154 * Rule deletion is asynchronous (regions transition through `DELETING`);
155 * returning before it settles trips ConflictExceptions in follow-up
156 * operations on the same configuration.
157 */
158const pollUntilGone = <A, E, R>(
159 self: Effect.Effect<A | undefined, E, R>,
160): Effect.Effect<A | undefined, E, R> =>
161 Effect.repeat(self, {
162 schedule: Schedule.fixed("2 seconds"),
163 until: (a) => a === undefined,
164 times: 20,
165 });
166
167const hasPendingRegion = (rule: {
168 statusSummaryByRegion: {
169 [key: string]: { status: string } | undefined;
170 };
171}): boolean =>
172 Object.values(rule.statusSummaryByRegion).some(
173 (s) =>
174 s !== undefined &&
175 (s.status === "CREATING" ||
176 s.status === "UPDATING" ||
177 s.status === "DELETING"),
178 );
179
180export const EventRuleProvider = () =>
181 Provider.effect(
182 EventRule,
183 Effect.gen(function* () {
184 const getByArn = Effect.fn(function* (arn: string) {
185 return yield* pinNotificationsRegion(
186 notifications
187 .getEventRule({ arn })
188 .pipe(
189 Effect.catchTag("ResourceNotFoundException", () =>
190 Effect.succeed(undefined),
191 ),
192 ),
193 );
194 });
195
196 // Identity fallback when the ARN cache is missing: scan the parent
197 // configuration's rules for the (source, eventType) pair.
198 const findBySignature = Effect.fn(function* (
199 notificationConfigurationArn: string,
200 source: string,
201 eventType: string,
202 ) {
203 return yield* pinNotificationsRegion(
204 notifications.listEventRules
205 .items({ notificationConfigurationArn })
206 .pipe(
207 Stream.filter(
208 (rule) =>
209 rule.source === source && rule.eventType === eventType,
210 ),
211 Stream.runHead,
212 Effect.map(Option.getOrUndefined),
213 Effect.catchTag("ResourceNotFoundException", () =>
214 Effect.succeed(undefined),
215 ),
216 ),
217 );
218 });
219
220 const toAttrs = (live: {
221 arn: string;
222 notificationConfigurationArn: string;
223 source: string;
224 eventType: string;
225 regions: readonly string[];
226 }) => ({
227 eventRuleArn: live.arn,
228 notificationConfigurationArn: live.notificationConfigurationArn,
229 source: live.source,
230 eventType: live.eventType,
231 regions: [...live.regions],
232 });
233
234 return EventRule.Provider.of({
235 stables: [
236 "eventRuleArn",
237 "notificationConfigurationArn",
238 "source",
239 "eventType",
240 ],
241
242 // Sub-resource keyed by its parent notification configuration —
243 // there is no account-wide enumeration without a parent ARN.
244 list: () => Effect.succeed([]),
245
246 read: Effect.fn(function* ({ olds, output }) {
247 const found = output?.eventRuleArn
248 ? yield* getByArn(output.eventRuleArn)
249 : olds
250 ? yield* findBySignature(
251 olds.notificationConfigurationArn,
252 olds.source,
253 olds.eventType,
254 )
255 : undefined;
256 // Event rules are not taggable, so ownership can't be verified —
257 // identity via the parent-scoped (source, eventType) signature.
258 return found ? toAttrs(found) : undefined;
259 }),
260
261 diff: Effect.fn(function* ({ news, olds }) {
262 if (!isResolved(news)) return undefined;
263 if (
264 news.notificationConfigurationArn !==
265 olds.notificationConfigurationArn ||
266 news.source !== olds.source ||
267 news.eventType !== olds.eventType
268 ) {
269 return { action: "replace" } as const;
270 }
271 return undefined;
272 }),
273
274 reconcile: Effect.fn(function* ({ news, output, session }) {
275 const desiredPattern = toPatternString(news.eventPattern);
276
277 // OBSERVE — cloud state is authoritative; the cached ARN falls
278 // through to a parent-scoped signature scan.
279 let live = output?.eventRuleArn
280 ? yield* getByArn(output.eventRuleArn)
281 : yield* findBySignature(
282 news.notificationConfigurationArn,
283 news.source,
284 news.eventType,
285 );
286
287 // ENSURE — create when missing; ConflictException means a peer
288 // created the same rule concurrently → re-observe.
289 if (live === undefined) {
290 // Retry the transitional lock (a sibling rule still DELETING);
291 // a conflict that persists past the budget means the same rule
292 // already exists → re-observe by signature.
293 const created = yield* retryWhileConflict(
294 pinNotificationsRegion(
295 notifications.createEventRule({
296 notificationConfigurationArn:
297 news.notificationConfigurationArn,
298 source: news.source,
299 eventType: news.eventType,
300 ...(desiredPattern !== ""
301 ? { eventPattern: desiredPattern }
302 : {}),
303 regions: news.regions,
304 }),
305 ),
306 ).pipe(
307 Effect.catchTag("ConflictException", () =>
308 Effect.succeed(undefined),
309 ),
310 );
311 live = created
312 ? yield* getByArn(created.arn)
313 : yield* findBySignature(
314 news.notificationConfigurationArn,
315 news.source,
316 news.eventType,
317 );
318 }
319 const arn = live!.arn;
320
321 // SYNC — diff observed eventPattern/regions against desired and
322 // apply only the delta.
323 if (
324 live!.eventPattern !== desiredPattern ||
325 !sameRegions(live!.regions, news.regions)
326 ) {
327 yield* retryWhileConflict(
328 pinNotificationsRegion(
329 notifications.updateEventRule({
330 arn,
331 eventPattern: desiredPattern,
332 regions: news.regions,
333 }),
334 ),
335 );
336 }
337
338 // Wait for regional managed rules to settle (bounded, fail-open).
339 const settled = yield* untilSettled(
340 pinNotificationsRegion(notifications.getEventRule({ arn })),
341 hasPendingRegion,
342 );
343
344 yield* session.note(arn);
345 return toAttrs(settled);
346 }),
347
348 delete: Effect.fn(function* ({ output }) {
349 // Idempotent — the parent configuration's deletion cascades rules,
350 // so the rule may already be gone. A ConflictException means a
351 // sibling mutation is still settling — retry through it.
352 yield* retryWhileConflict(
353 pinNotificationsRegion(
354 notifications.deleteEventRule({ arn: output.eventRuleArn }),
355 ),
356 ).pipe(
357 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
358 );
359 // Deletion is asynchronous — wait until the rule is actually gone
360 // so follow-up mutations on the configuration don't hit the
361 // transitional lock (bounded, fail-open).
362 yield* pollUntilGone(getByArn(output.eventRuleArn));
363 }),
364 });
365 }),
366 );