1import * as redshift from "@distilled.cloud/aws/redshift";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
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 { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import { AWSEnvironment } from "../Environment.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 applyRedshiftTagDelta,
15 redshiftArn,
16 sameStringSet,
17 toTagRecord,
18} from "./internal.ts";
19
20/**
21 * The kind of Redshift resource an {@link EventSubscription} filters on.
22 */
23export type EventSubscriptionSourceType =
24 | "cluster"
25 | "cluster-parameter-group"
26 | "cluster-security-group"
27 | "cluster-snapshot"
28 | "scheduled-action";
29
30/**
31 * Redshift event categories an {@link EventSubscription} can filter on.
32 */
33export type EventSubscriptionCategory =
34 | "configuration"
35 | "management"
36 | "monitoring"
37 | "security"
38 | "pending";
39
40export interface EventSubscriptionProps {
41 /**
42 * Name of the event subscription. Must be 1-255 ASCII letters, digits or
43 * hyphens, starting with a letter, and must not end with a hyphen or
44 * contain two consecutive hyphens. If omitted, a deterministic physical
45 * name is generated. Changing the name replaces the subscription.
46 */
47 subscriptionName?: string;
48 /**
49 * ARN of the SNS topic Redshift publishes matching events to. The topic
50 * must exist and allow Redshift to publish.
51 */
52 snsTopicArn: string;
53 /**
54 * The kind of source to filter on (e.g. `"cluster"`). Required when
55 * `sourceIds` is set.
56 * @default all source types
57 */
58 sourceType?: EventSubscriptionSourceType;
59 /**
60 * Identifiers of the specific sources to filter on (e.g. cluster
61 * identifiers when `sourceType` is `"cluster"`).
62 * @default all sources of the subscribed type
63 */
64 sourceIds?: string[];
65 /**
66 * Event categories to subscribe to (e.g. `["monitoring", "management"]`).
67 * @default all categories
68 */
69 eventCategories?: EventSubscriptionCategory[];
70 /**
71 * Only deliver events of this severity.
72 * @default both `ERROR` and `INFO`
73 */
74 severity?: "ERROR" | "INFO";
75 /**
76 * Whether event delivery is active.
77 * @default true
78 */
79 enabled?: boolean;
80 /**
81 * User-defined tags for the event subscription.
82 */
83 tags?: Record<string, string>;
84}
85
86export interface EventSubscription extends Resource<
87 "AWS.Redshift.EventSubscription",
88 EventSubscriptionProps,
89 {
90 /**
91 * Name of the event subscription.
92 */
93 subscriptionName: string;
94 /**
95 * ARN of the event subscription.
96 */
97 eventSubscriptionArn: string;
98 /**
99 * ARN of the SNS topic events are delivered to.
100 */
101 snsTopicArn: string | undefined;
102 /**
103 * Status of the subscription (`"active"`, or `"no-permission"` /
104 * `"topic-not-exist"` when the SNS topic became unreachable after
105 * creation).
106 */
107 status: string | undefined;
108 /**
109 * The source type filter, if any.
110 */
111 sourceType: string | undefined;
112 /**
113 * The source identifier filters, if any.
114 */
115 sourceIds: string[];
116 /**
117 * The event category filters, if any.
118 */
119 eventCategories: string[];
120 /**
121 * The severity filter, if any.
122 */
123 severity: string | undefined;
124 /**
125 * Whether event delivery is active.
126 */
127 enabled: boolean | undefined;
128 /**
129 * Tags on the event subscription (including internal Alchemy tags).
130 */
131 tags: Record<string, string>;
132 },
133 never,
134 Providers
135> {}
136
137/**
138 * An Amazon Redshift event notification subscription — routes provisioned
139 * cluster lifecycle events (maintenance, resizes, snapshots, failures,
140 * security changes) to an SNS topic.
141 *
142 * SNS is Redshift's native event channel for provisioned clusters: the only
143 * events Redshift publishes *directly* to EventBridge are the zero-ETL
144 * integration detail-types, so cluster events reach compute through an
145 * `EventSubscription` → `SNS.Topic` → `SNS.consumeTopicNotifications` chain.
146 * Subscriptions are free and provision instantly.
147 * ### Subscribing to Cluster Events
148 * **Example:** Route Cluster Events to an SNS Topic
149 * ```typescript
150 * const alerts = yield* SNS.Topic("WarehouseAlerts", {});
151 * const subscription = yield* Redshift.EventSubscription("WarehouseEvents", {
152 * snsTopicArn: alerts.topicArn,
153 * sourceType: "cluster",
154 * sourceIds: [cluster.clusterIdentifier],
155 * });
156 * ```
157 * **Example:** Only Error-Severity Monitoring Events
158 * ```typescript
159 * const subscription = yield* Redshift.EventSubscription("WarehouseErrors", {
160 * snsTopicArn: alerts.topicArn,
161 * eventCategories: ["monitoring"],
162 * severity: "ERROR",
163 * });
164 * ```
165 * **Example:** Consume the Events in a Function
166 * ```typescript
167 * // inside a Lambda Function definition:
168 * yield* SNS.consumeTopicNotifications(alerts, (messages) =>
169 * Stream.runForEach(messages, (message) =>
170 * Effect.logInfo(`redshift event: ${message.Message}`),
171 * ),
172 * );
173 * ```
174 *
175 * @resource
176 */
177export const EventSubscription = Resource<EventSubscription>(
178 "AWS.Redshift.EventSubscription",
179);
180
181/**
182 * Retry an effect while the subscription is mid-transition
183 * (`InvalidSubscriptionStateFault`), bounded to ~2 minutes.
184 *
185 * Expressed as an explicitly-typed module-scope helper: inlining
186 * `Effect.retry` in a lifecycle operation leaves `Retry.Return`'s
187 * conditional type unresolved in the provider's inferred layer type, which
188 * declaration emit widens to an `unknown` R — poisoning `AWS.providers()`
189 * for every downstream consumer.
190 */
191const retryWhileSubscriptionTransitioning = <A, E extends { _tag: string }, R>(
192 self: Effect.Effect<A, E, R>,
193): Effect.Effect<A, E, R> =>
194 Effect.retry(self, {
195 while: (e) => e._tag === "InvalidSubscriptionStateFault",
196 schedule: Schedule.max([Schedule.fixed("5 seconds"), Schedule.recurs(24)]),
197 });
198
199export const EventSubscriptionProvider = () =>
200 Provider.effect(
201 EventSubscription,
202 Effect.gen(function* () {
203 const toName = (id: string, props: Partial<EventSubscriptionProps>) =>
204 props.subscriptionName
205 ? Effect.succeed(props.subscriptionName)
206 : createPhysicalName({ id, maxLength: 255, lowercase: true });
207
208 const readSubscription = Effect.fn(function* (name: string) {
209 const response = yield* redshift
210 .describeEventSubscriptions({ SubscriptionName: name })
211 .pipe(
212 Effect.catchTag("SubscriptionNotFoundFault", () =>
213 Effect.succeed(undefined),
214 ),
215 );
216 return response?.EventSubscriptionsList?.[0];
217 });
218
219 const toAttrs = Effect.fn(function* (
220 subscription: redshift.EventSubscription,
221 ) {
222 const { accountId, region } = yield* AWSEnvironment.current;
223 if (!subscription.CustSubscriptionId) {
224 return yield* Effect.fail(
225 new Error("Event subscription is missing its name"),
226 );
227 }
228 return {
229 subscriptionName: subscription.CustSubscriptionId,
230 eventSubscriptionArn: redshiftArn(
231 region,
232 accountId,
233 "eventsubscription",
234 subscription.CustSubscriptionId,
235 ),
236 snsTopicArn: subscription.SnsTopicArn,
237 status: subscription.Status,
238 sourceType: subscription.SourceType,
239 sourceIds: subscription.SourceIdsList ?? [],
240 eventCategories: subscription.EventCategoriesList ?? [],
241 severity: subscription.Severity,
242 enabled: subscription.Enabled,
243 tags: toTagRecord(subscription.Tags),
244 };
245 });
246
247 return {
248 stables: ["subscriptionName", "eventSubscriptionArn"],
249
250 diff: Effect.fn(function* ({ id, olds, news }) {
251 if (!isResolved(news)) return undefined;
252 if (
253 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
254 ) {
255 return { action: "replace" } as const;
256 }
257 }),
258
259 read: Effect.fn(function* ({ id, olds, output }) {
260 const name =
261 output?.subscriptionName ?? (yield* toName(id, olds ?? {}));
262 const subscription = yield* readSubscription(name);
263 if (!subscription?.CustSubscriptionId) return undefined;
264 const attrs = yield* toAttrs(subscription);
265 return (yield* hasAlchemyTags(id, attrs.tags))
266 ? attrs
267 : Unowned(attrs);
268 }),
269
270 reconcile: Effect.fn(function* ({ id, news, output, session }) {
271 const name = output?.subscriptionName ?? (yield* toName(id, news));
272 const internalTags = yield* createInternalTags(id);
273 const desiredTags = { ...internalTags, ...news.tags };
274
275 // 1. Observe — cloud state is authoritative.
276 let observed = yield* readSubscription(name);
277
278 // 2. Ensure — create if missing; tolerate AlreadyExists as a race
279 // with a peer reconciler by re-reading.
280 if (!observed?.CustSubscriptionId) {
281 yield* redshift
282 .createEventSubscription({
283 SubscriptionName: name,
284 SnsTopicArn: news.snsTopicArn,
285 SourceType: news.sourceType,
286 SourceIds: news.sourceIds,
287 EventCategories: news.eventCategories,
288 Severity: news.severity,
289 Enabled: news.enabled ?? true,
290 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
291 Key,
292 Value,
293 })),
294 })
295 .pipe(
296 Effect.catchTag(
297 "SubscriptionAlreadyExistFault",
298 () => Effect.void,
299 ),
300 );
301 observed = yield* readSubscription(name);
302 if (!observed?.CustSubscriptionId) {
303 return yield* Effect.fail(
304 new Error(`Failed to create event subscription '${name}'`),
305 );
306 }
307 }
308
309 // 3. Sync — diff OBSERVED filters against desired; modify replaces
310 // the whole filter configuration, so send the full desired
311 // shape when anything drifted.
312 const desiredEnabled = news.enabled ?? true;
313 if (
314 observed.SnsTopicArn !== news.snsTopicArn ||
315 observed.SourceType !== news.sourceType ||
316 !sameStringSet(observed.SourceIdsList, news.sourceIds) ||
317 !sameStringSet(
318 observed.EventCategoriesList,
319 news.eventCategories,
320 ) ||
321 observed.Severity !== news.severity ||
322 observed.Enabled !== desiredEnabled
323 ) {
324 yield* retryWhileSubscriptionTransitioning(
325 redshift.modifyEventSubscription({
326 SubscriptionName: name,
327 SnsTopicArn: news.snsTopicArn,
328 SourceType: news.sourceType,
329 SourceIds: news.sourceIds,
330 EventCategories: news.eventCategories,
331 Severity: news.severity,
332 Enabled: desiredEnabled,
333 }),
334 );
335 observed = yield* readSubscription(name);
336 if (!observed?.CustSubscriptionId) {
337 return yield* Effect.fail(
338 new Error(
339 `Event subscription '${name}' not found after update`,
340 ),
341 );
342 }
343 }
344
345 // 3b. Sync tags — diff against OBSERVED cloud tags (describe
346 // surfaces them inline).
347 const { accountId, region } = yield* AWSEnvironment.current;
348 const arn = redshiftArn(region, accountId, "eventsubscription", name);
349 const { removed, upsert } = diffTags(
350 toTagRecord(observed.Tags),
351 desiredTags,
352 );
353 yield* applyRedshiftTagDelta({ arn, upsert, removed });
354
355 yield* session.note(arn);
356 const attrs = yield* toAttrs(observed);
357 return { ...attrs, tags: desiredTags };
358 }),
359
360 delete: Effect.fn(function* ({ output }) {
361 yield* retryWhileSubscriptionTransitioning(
362 redshift
363 .deleteEventSubscription({
364 SubscriptionName: output.subscriptionName,
365 })
366 .pipe(
367 Effect.catchTag("SubscriptionNotFoundFault", () =>
368 Effect.succeed(undefined),
369 ),
370 ),
371 );
372 }),
373
374 list: () =>
375 // Top-level account/region collection: exhaustively paginate
376 // describeEventSubscriptions; tags come inline on each entry.
377 redshift.describeEventSubscriptions.pages({}).pipe(
378 Stream.runCollect,
379 Effect.map((chunk) =>
380 Array.from(chunk).flatMap((page) =>
381 (page.EventSubscriptionsList ?? []).filter(
382 (subscription) =>
383 subscription.CustSubscriptionId !== undefined,
384 ),
385 ),
386 ),
387 Effect.flatMap(
388 Effect.forEach((subscription) => toAttrs(subscription), {
389 concurrency: 4,
390 }),
391 ),
392 ),
393 };
394 }),
395 );