1import * as sns from "@distilled.cloud/aws/sns";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Stream from "effect/Stream";
5import { isResolved } from "../../Diff.ts";
6import type { Input } from "../../Input.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10
11export type SubscriptionArn = string;
12
13export interface SubscriptionProps {
14 /**
15 * ARN of the topic to subscribe to.
16 */
17 topicArn: Input<string>;
18 /**
19 * SNS subscription protocol, for example `lambda`, `sqs`, `https`, or `email`.
20 */
21 protocol: string;
22 /**
23 * Endpoint for the selected protocol, such as a Lambda function ARN or queue ARN.
24 */
25 endpoint?: Input<string>;
26 /**
27 * Raw SNS subscription attributes keyed by AWS attribute name.
28 */
29 attributes?: Record<string, string>;
30 /**
31 * Whether SNS should return the subscription ARN immediately, even while pending confirmation.
32 * @default true
33 */
34 returnSubscriptionArn?: boolean;
35}
36
37export interface Subscription extends Resource<
38 "AWS.SNS.Subscription",
39 SubscriptionProps,
40 {
41 /** ARN of the subscription. */
42 subscriptionArn: SubscriptionArn;
43 /** ARN of the topic the subscription is attached to. */
44 topicArn: string;
45 /** Delivery protocol of the subscription (e.g. `lambda`, `sqs`, `https`). */
46 protocol: string;
47 /** Endpoint receiving deliveries, such as a Lambda function or queue ARN. */
48 endpoint: string | undefined;
49 /** AWS account ID that owns the subscription. */
50 owner: string | undefined;
51 /** Whether the subscription is still awaiting endpoint confirmation. */
52 pendingConfirmation: boolean;
53 /** Raw SNS subscription attributes keyed by AWS attribute name. */
54 attributes: Record<string, string>;
55 },
56 never,
57 Providers
58> {}
59
60/**
61 * An Amazon SNS subscription that attaches an endpoint to a topic.
62 *
63 * `Subscription` keeps the lifecycle of the subscription itself separate from the
64 * topic, which lets Lambda event sources and manually managed subscriptions share
65 * the same canonical resource model.
66 * ### Creating Subscriptions
67 * **Example:** Lambda Subscription
68 * ```typescript
69 * const subscription = yield* Subscription("TopicSubscription", {
70 * topicArn: topic.topicArn,
71 * protocol: "lambda",
72 * endpoint: fn.functionArn,
73 * });
74 * ```
75 *
76 * **Example:** Fan Out a Topic to an SQS Queue
77 * ```typescript
78 * const topic = yield* SNS.Topic("Events");
79 * const queue = yield* SQS.Queue("Notifications");
80 *
81 * const subscription = yield* Subscription("QueueSubscription", {
82 * topicArn: topic.topicArn,
83 * protocol: "sqs",
84 * endpoint: queue.queueArn,
85 * returnSubscriptionArn: true,
86 * });
87 * ```
88 *
89 * **Example:** Filtered Subscription
90 * ```typescript
91 * const subscription = yield* Subscription("OrderSubscription", {
92 * topicArn: topic.topicArn,
93 * protocol: "sqs",
94 * endpoint: queue.queueArn,
95 * attributes: {
96 * FilterPolicy: JSON.stringify({ type: ["order"] }),
97 * },
98 * });
99 * ```
100 *
101 * @resource
102 */
103export const Subscription = Resource<Subscription>("AWS.SNS.Subscription");
104
105export const SubscriptionProvider = () =>
106 Provider.succeed(Subscription, {
107 read: Effect.fn(function* ({ olds, output }) {
108 return yield* readSubscription({
109 subscriptionArn: output?.subscriptionArn,
110 topicArn: (output?.topicArn ?? olds.topicArn) as string | undefined,
111 protocol: output?.protocol ?? olds.protocol,
112 endpoint: (output?.endpoint ?? olds.endpoint) as string | undefined,
113 });
114 }),
115 stables: ["subscriptionArn"],
116 // Account/region-scoped collection: paginate `listSubscriptions`
117 // exhaustively, then hydrate each concrete-ARN subscription via
118 // `getSubscriptionAttributes` (inside `readSubscription`) so each element
119 // matches the exact `read` Attributes shape. Pending-confirmation entries
120 // have no real ARN to hydrate or delete against, so they are skipped.
121 list: Effect.fn(function* () {
122 const subscriptions = yield* sns.listSubscriptions.pages({}).pipe(
123 Stream.runCollect,
124 Effect.map((chunk) =>
125 Array.from(chunk).flatMap((page) => page.Subscriptions ?? []),
126 ),
127 );
128
129 const concrete = subscriptions.filter(
130 (
131 subscription,
132 ): subscription is sns.Subscription & {
133 SubscriptionArn: string;
134 } =>
135 typeof subscription.SubscriptionArn === "string" &&
136 !isPendingConfirmation(subscription.SubscriptionArn),
137 );
138
139 const rows = yield* Effect.forEach(
140 concrete,
141 (subscription) =>
142 readSubscription({
143 subscriptionArn: subscription.SubscriptionArn,
144 topicArn: subscription.TopicArn,
145 protocol: subscription.Protocol,
146 endpoint: subscription.Endpoint,
147 }),
148 { concurrency: 10 },
149 );
150
151 return rows.filter(
152 (row): row is NonNullable<typeof row> => row !== undefined,
153 );
154 }),
155 diff: Effect.fn(function* ({ news, olds }) {
156 if (!isResolved(news)) return undefined;
157 if (news.protocol !== olds.protocol) {
158 return { action: "replace" } as const;
159 }
160
161 if (
162 typeof news.topicArn === "string" &&
163 typeof olds.topicArn === "string" &&
164 news.topicArn !== olds.topicArn
165 ) {
166 return { action: "replace" } as const;
167 }
168
169 if (
170 typeof news.endpoint === "string" &&
171 typeof olds.endpoint === "string" &&
172 news.endpoint !== olds.endpoint
173 ) {
174 return { action: "replace" } as const;
175 }
176 }),
177 reconcile: Effect.fn(function* ({ news, output, session }) {
178 const topicArn = news.topicArn as string;
179 const protocol = news.protocol;
180 const endpoint = news.endpoint as string | undefined;
181 const desiredAttributes = toAttributeMap(news.attributes);
182
183 // Observe — derive the live SubscriptionArn. We can't trust a
184 // pending-confirmation ARN from `output`, since the user may have
185 // confirmed it out of band. Cached ARN is preferred when concrete;
186 // otherwise we list-and-match by (topicArn, protocol, endpoint).
187 let subscriptionArn: string | undefined;
188 if (
189 output?.subscriptionArn &&
190 !isPendingConfirmation(output.subscriptionArn)
191 ) {
192 const observed = yield* sns
193 .getSubscriptionAttributes({
194 SubscriptionArn: output.subscriptionArn,
195 })
196 .pipe(
197 Effect.catchTag("NotFoundException", () =>
198 Effect.succeed(undefined),
199 ),
200 Effect.catchTag("InvalidParameterException", () =>
201 Effect.succeed(undefined),
202 ),
203 );
204 if (observed) {
205 subscriptionArn = output.subscriptionArn;
206 }
207 } else {
208 subscriptionArn = yield* findSubscription({
209 topicArn,
210 protocol,
211 endpoint,
212 });
213 }
214
215 // Ensure — subscribe if no live subscription matches.
216 if (!subscriptionArn) {
217 const response = yield* sns.subscribe({
218 TopicArn: topicArn,
219 Protocol: protocol,
220 Endpoint: endpoint,
221 Attributes: news.attributes,
222 ReturnSubscriptionArn: news.returnSubscriptionArn ?? true,
223 });
224 if (!response.SubscriptionArn) {
225 return yield* Effect.die(new Error(`subscribe returned no ARN`));
226 }
227 subscriptionArn = response.SubscriptionArn;
228 yield* session.note(subscriptionArn);
229 return {
230 subscriptionArn,
231 topicArn,
232 protocol,
233 endpoint,
234 owner: undefined,
235 pendingConfirmation: isPendingConfirmation(subscriptionArn),
236 attributes: desiredAttributes,
237 };
238 }
239
240 // Sync attributes — fetch observed cloud attributes (FilterPolicy,
241 // RawMessageDelivery, etc.) and apply only the delta. We can only
242 // do this once the subscription is confirmed (pending confirmation
243 // returns no real ARN to mutate against).
244 const attrsResponse = yield* sns
245 .getSubscriptionAttributes({ SubscriptionArn: subscriptionArn })
246 .pipe(
247 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
248 Effect.catchTag("InvalidParameterException", () =>
249 Effect.succeed(undefined),
250 ),
251 );
252 const observedAttributes = toAttributeMap(attrsResponse?.Attributes);
253
254 for (const [name, value] of Object.entries(desiredAttributes)) {
255 if (observedAttributes[name] !== value) {
256 yield* sns.setSubscriptionAttributes({
257 SubscriptionArn: subscriptionArn,
258 AttributeName: name,
259 AttributeValue: value,
260 });
261 }
262 }
263
264 // Reset attributes the user no longer specifies. We only clear
265 // user-specified keys (those present in `news.attributes`/`olds`),
266 // not all observed keys, because SNS exposes many read-only system
267 // attributes we should not touch.
268 const previousKeys = new Set(
269 Object.keys(toAttributeMap(output?.attributes)),
270 );
271 for (const name of previousKeys) {
272 if (!(name in desiredAttributes)) {
273 yield* sns.setSubscriptionAttributes({
274 SubscriptionArn: subscriptionArn,
275 AttributeName: name,
276 });
277 }
278 }
279
280 yield* session.note(subscriptionArn);
281
282 return {
283 subscriptionArn,
284 topicArn,
285 protocol,
286 endpoint,
287 owner: output?.owner,
288 pendingConfirmation: isPendingConfirmation(subscriptionArn),
289 attributes: desiredAttributes,
290 };
291 }),
292 delete: Effect.fn(function* ({ olds, output }) {
293 const subscriptionArn = isPendingConfirmation(output.subscriptionArn)
294 ? yield* findSubscription({
295 topicArn: (output.topicArn ?? olds.topicArn) as string | undefined,
296 protocol: output.protocol ?? olds.protocol,
297 endpoint: (output.endpoint ?? olds.endpoint) as string | undefined,
298 }).pipe(
299 // The topic is already gone — SNS drops its subscriptions with
300 // it, so there is nothing left to unsubscribe.
301 Effect.catchTag("NotFoundException", () =>
302 Effect.succeed(undefined),
303 ),
304 )
305 : output.subscriptionArn;
306
307 if (!subscriptionArn) {
308 return;
309 }
310
311 yield* sns
312 .unsubscribe({
313 SubscriptionArn: subscriptionArn,
314 })
315 .pipe(
316 Effect.catchTag("NotFoundException", () => Effect.void),
317 Effect.catchTag("InvalidParameterException", () => Effect.void),
318 );
319 }),
320 });
321
322const isPendingConfirmation = (subscriptionArn: string | undefined) =>
323 subscriptionArn === undefined ||
324 subscriptionArn.toLowerCase() === "pending confirmation";
325
326const toAttributeMap = (
327 attributes: Record<string, string | undefined> | undefined,
328): Record<string, string> =>
329 Object.fromEntries(
330 Object.entries(attributes ?? {}).filter(
331 (entry): entry is [string, string] => typeof entry[1] === "string",
332 ),
333 );
334
335const findSubscription = Effect.fn(function* ({
336 topicArn,
337 protocol,
338 endpoint,
339}: {
340 topicArn: string | undefined;
341 protocol: string | undefined;
342 endpoint: string | undefined;
343}) {
344 if (!topicArn || !protocol) {
345 return undefined;
346 }
347
348 const match = yield* sns.listSubscriptionsByTopic
349 .items({ TopicArn: topicArn })
350 .pipe(
351 Stream.filter(
352 (subscription) =>
353 subscription.Protocol === protocol &&
354 subscription.Endpoint === endpoint &&
355 !!subscription.SubscriptionArn,
356 ),
357 Stream.runHead,
358 Effect.map(Option.getOrUndefined),
359 );
360
361 return match?.SubscriptionArn;
362});
363
364const readSubscription = Effect.fn(function* ({
365 subscriptionArn,
366 topicArn,
367 protocol,
368 endpoint,
369}: {
370 subscriptionArn?: string;
371 topicArn?: string;
372 protocol?: string;
373 endpoint: string | undefined;
374}) {
375 let resolvedSubscriptionArn: string | undefined;
376 if (subscriptionArn && !isPendingConfirmation(subscriptionArn)) {
377 resolvedSubscriptionArn = subscriptionArn;
378 } else {
379 // No concrete ARN to hydrate — list the topic's subscriptions and match
380 // by (protocol, endpoint). If the topic itself no longer exists (deleted
381 // out of band, or by an interrupted destroy that committed the topic's
382 // state row late) then the subscription is gone too: report "missing"
383 // rather than failing the read, so a subsequent destroy can proceed.
384 const found = yield* findSubscription({
385 topicArn,
386 protocol,
387 endpoint,
388 }).pipe(
389 Effect.map((arn) => ({ arn })),
390 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
391 );
392 if (found === undefined) {
393 return undefined;
394 }
395 resolvedSubscriptionArn = found.arn;
396 }
397
398 if (!resolvedSubscriptionArn) {
399 if (!topicArn || !protocol) {
400 return undefined;
401 }
402
403 return {
404 subscriptionArn: subscriptionArn ?? "pending confirmation",
405 topicArn,
406 protocol,
407 endpoint,
408 owner: undefined,
409 pendingConfirmation: true,
410 attributes: {},
411 };
412 }
413
414 const response = yield* sns
415 .getSubscriptionAttributes({
416 SubscriptionArn: resolvedSubscriptionArn,
417 })
418 .pipe(
419 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
420 Effect.catchTag("InvalidParameterException", () =>
421 Effect.succeed(undefined),
422 ),
423 );
424
425 if (!response) {
426 return undefined;
427 }
428
429 const attributes = toAttributeMap(response.Attributes);
430 const resolvedTopicArn = attributes.TopicArn ?? topicArn;
431 const resolvedProtocol = attributes.Protocol ?? protocol;
432
433 if (!resolvedTopicArn || !resolvedProtocol) {
434 return undefined;
435 }
436
437 return {
438 subscriptionArn: resolvedSubscriptionArn,
439 topicArn: resolvedTopicArn,
440 protocol: resolvedProtocol,
441 endpoint: attributes.Endpoint ?? endpoint,
442 owner: attributes.Owner,
443 pendingConfirmation:
444 attributes.PendingConfirmation === "true" ||
445 isPendingConfirmation(resolvedSubscriptionArn),
446 attributes,
447 };
448});