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 { 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 type { AccountID } from "../Environment.ts";
12import type { Providers } from "../Providers.ts";
13import type { RegionID } from "../Region.ts";
14
15export type TopicName = string;
16export type TopicArn = `arn:aws:sns:${RegionID}:${AccountID}:${TopicName}`;
17
18export interface TopicProps {
19 /**
20 * Name of the topic.
21 * @default ${app}-${stage}-${id}?.fifo
22 */
23 topicName?: string;
24 /**
25 * Whether to create a FIFO topic.
26 * @default false
27 */
28 fifo?: boolean;
29 /**
30 * Raw SNS topic attributes keyed by AWS attribute name.
31 * Use this for delivery policies, tracing, KMS, signatures, archive policy, and
32 * other SNS topic attributes not modeled as first-class props.
33 */
34 attributes?: Record<string, string>;
35 /**
36 * SNS data protection policy JSON for the topic.
37 *
38 * TODO(sam): should this be a typed object that we serialize/deserialize?
39 */
40 dataProtectionPolicy?: string;
41 /**
42 * User-defined tags to apply to the topic.
43 */
44 tags?: Record<string, string>;
45}
46
47export interface Topic extends Resource<
48 "AWS.SNS.Topic",
49 TopicProps,
50 {
51 topicArn: TopicArn;
52 topicName: TopicName;
53 fifo: boolean;
54 attributes: Record<string, string>;
55 dataProtectionPolicy: string | undefined;
56 tags: Record<string, string>;
57 },
58 never,
59 Providers
60> {}
61
62/**
63 * An Amazon SNS topic for fan-out messaging and notifications.
64 *
65 * `Topic` owns the SNS topic lifecycle while raw AWS topic attributes remain
66 * available through the `attributes` prop so the full core pub/sub surface can
67 * be configured without waiting on additional typed wrappers. A topic name is
68 * auto-generated unless you provide one explicitly.
69 * ### Creating Topics
70 * **Example:** Standard Topic
71 * ```typescript
72 * import * as SNS from "alchemy/AWS/SNS";
73 *
74 * const topic = yield* SNS.Topic("OrdersTopic");
75 * ```
76 *
77 * **Example:** Topic with Display Name
78 * ```typescript
79 * const topic = yield* SNS.Topic("NotificationsTopic", {
80 * attributes: {
81 * DisplayName: "App Notifications",
82 * },
83 * });
84 * ```
85 *
86 * **Example:** FIFO Topic
87 * ```typescript
88 * const topic = yield* SNS.Topic("OrdersFifoTopic", {
89 * fifo: true,
90 * attributes: {
91 * ContentBasedDeduplication: "true",
92 * },
93 * });
94 * ```
95 *
96 * ### Runtime Publishing
97 * Bind publish operations in the init phase and use them in runtime
98 * handlers.
99 *
100 * **Example:** Publish from a handler
101 * ```typescript
102 * // init
103 * const publish = yield* SNS.Publish(topic);
104 *
105 * return {
106 * fetch: Effect.gen(function* () {
107 * // runtime
108 * yield* publish({
109 * Message: JSON.stringify({ orderId: "123" }),
110 * Subject: "OrderCreated",
111 * });
112 * return HttpServerResponse.text("Published");
113 * }),
114 * };
115 * ```
116 *
117 * ### Subscriptions
118 * Subscribe a Lambda function to process messages published to the
119 * topic. The subscription and invoke permissions are created
120 * automatically.
121 *
122 * **Example:** Process topic notifications
123 * ```typescript
124 * // init
125 * yield* SNS.consumeTopicNotifications(topic, (stream) =>
126 * stream.pipe(
127 * Stream.runForEach((message) =>
128 * Effect.log(`Received: ${message.Message}`),
129 * ),
130 * ),
131 * );
132 * ```
133 *
134 * @resource
135 */
136export const Topic = Resource<Topic>("AWS.SNS.Topic");
137
138export const TopicProvider = () =>
139 Provider.succeed(Topic, {
140 // AWS account/region collection: `listTopics` enumerates every topic ARN
141 // in the ambient account+region (paginated, ARN-only), then each ARN is
142 // hydrated via `readTopic` into the exact `read` Attributes shape
143 // (attributes + tags + data protection policy). Per-item not-found is
144 // handled inside `readTopic` (returns undefined) for topics deleted
145 // between enumeration and hydration.
146 list: Effect.fn(function* () {
147 const topicArns = yield* sns.listTopics.pages({}).pipe(
148 Stream.runCollect,
149 Effect.map((chunk) =>
150 Array.from(chunk).flatMap((page) =>
151 (page.Topics ?? [])
152 .map((topic) => topic.TopicArn)
153 .filter((arn): arn is string => typeof arn === "string"),
154 ),
155 ),
156 );
157
158 const rows = yield* Effect.forEach(
159 topicArns,
160 (topicArn) =>
161 readTopic({
162 id: "",
163 topicArn,
164 topicName: topicArn.split(":").at(-1) ?? topicArn,
165 }),
166 { concurrency: 10 },
167 );
168
169 return rows.filter(
170 (row): row is NonNullable<typeof row> => row !== undefined,
171 );
172 }),
173 read: Effect.fn(function* ({ id, olds, output }) {
174 const topicName =
175 output?.topicName ?? (yield* toTopicName(id, olds ?? {}));
176
177 const state = yield* readTopic({
178 id,
179 topicArn: output?.topicArn,
180 topicName,
181 });
182 if (!state) return undefined;
183 return (yield* hasAlchemyTags(id, state.tags)) ? state : Unowned(state);
184 }),
185 stables: ["topicArn", "topicName", "fifo"],
186 diff: Effect.fn(function* ({ id, news = {}, olds = {} }) {
187 if (!isResolved(news)) return undefined;
188 if ((news.fifo ?? false) !== (olds.fifo ?? false)) {
189 return { action: "replace" } as const;
190 }
191
192 const oldTopicName = yield* toTopicName(id, olds);
193 const newTopicName = yield* toTopicName(id, news);
194
195 if (oldTopicName !== newTopicName) {
196 return { action: "replace" } as const;
197 }
198
199 if (
200 olds.dataProtectionPolicy !== undefined &&
201 news.dataProtectionPolicy === undefined
202 ) {
203 return { action: "replace" } as const;
204 }
205 }),
206 reconcile: Effect.fn(function* ({ id, news = {}, olds, output, session }) {
207 const topicName = yield* toTopicName(id, news);
208 const internalTags = yield* createInternalTags(id);
209 const desiredTags = { ...internalTags, ...news.tags };
210 const desiredAttributes = toAttributes(news);
211 const isFifo = news.fifo ?? false;
212
213 // Observe + ensure — `createTopic` is idempotent for an identical
214 // (name, FifoTopic) pair, so we use it both to ensure-on-greenfield
215 // and to fetch the ARN for adoption. We deliberately pass ONLY the
216 // FifoTopic attribute (which is stable at create time): AWS rejects
217 // createTopic for an existing topic if any other attribute, tag,
218 // or DataProtectionPolicy differs from what's already attached.
219 // Mutable attributes are reconciled below against observed state.
220 const createResponse = yield* sns.createTopic({
221 Name: topicName,
222 Attributes: isFifo ? { FifoTopic: "true" } : undefined,
223 });
224 const topicArn = createResponse.TopicArn;
225 if (!topicArn) {
226 return yield* Effect.die(new Error(`createTopic returned no ARN`));
227 }
228
229 // Sync attributes — fetch observed cloud attributes and apply only
230 // the delta.
231 const observedState = yield* sns
232 .getTopicAttributes({ TopicArn: topicArn })
233 .pipe(
234 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
235 );
236 const observedAttributes = toAttributeMap(observedState?.Attributes);
237
238 for (const [name, value] of Object.entries(desiredAttributes)) {
239 if (observedAttributes[name] !== value) {
240 yield* sns.setTopicAttributes({
241 TopicArn: topicArn,
242 AttributeName: name,
243 AttributeValue: value,
244 });
245 }
246 }
247
248 // Reset previously-user-specified attributes that the user no
249 // longer specifies. Use `olds.attributes` (the prior props) — NOT
250 // `output.attributes`, which on adoption holds every attribute SNS
251 // returns and would erroneously try to reset SNS-managed values
252 // like the auto-generated Policy.
253 const previouslyManaged = olds?.attributes ?? {};
254 for (const name of Object.keys(previouslyManaged)) {
255 if (!(name in desiredAttributes) && name !== "FifoTopic") {
256 yield* sns.setTopicAttributes({
257 TopicArn: topicArn,
258 AttributeName: name,
259 });
260 }
261 }
262
263 // Sync tags — diff observed cloud tags against desired so that an
264 // adoption-takeover (cloud tags identify a different logical id)
265 // correctly rewrites ownership tags.
266 const observedTagsResp = yield* sns
267 .listTagsForResource({ ResourceArn: topicArn })
268 .pipe(
269 Effect.catchTag("ResourceNotFoundException", () =>
270 Effect.succeed({ Tags: [] as sns.Tag[] }),
271 ),
272 );
273 const observedTags = toTagMap(observedTagsResp.Tags);
274 const { removed, upsert } = diffTags(observedTags, desiredTags);
275
276 if (upsert.length > 0) {
277 yield* sns.tagResource({
278 ResourceArn: topicArn,
279 Tags: upsert,
280 });
281 }
282
283 if (removed.length > 0) {
284 yield* sns.untagResource({
285 ResourceArn: topicArn,
286 TagKeys: removed,
287 });
288 }
289
290 // Sync data protection policy — observed ↔ desired. SNS rejects
291 // getDataProtectionPolicy on FIFO topics, so we skip it there.
292 let observedPolicy: string | undefined;
293 if (!isFifo) {
294 observedPolicy = yield* sns
295 .getDataProtectionPolicy({ ResourceArn: topicArn })
296 .pipe(
297 Effect.map((r) => r.DataProtectionPolicy),
298 Effect.catchTag("NotFoundException", () =>
299 Effect.succeed(undefined),
300 ),
301 Effect.catchTag("InvalidParameterException", () =>
302 Effect.succeed(undefined),
303 ),
304 );
305 }
306 if (
307 !isFifo &&
308 news.dataProtectionPolicy !== undefined &&
309 news.dataProtectionPolicy !== observedPolicy
310 ) {
311 yield* sns.putDataProtectionPolicy({
312 ResourceArn: topicArn,
313 DataProtectionPolicy: news.dataProtectionPolicy,
314 });
315 }
316
317 yield* session.note(topicArn);
318
319 return {
320 topicArn: topicArn as TopicArn,
321 topicName,
322 fifo: isFifo,
323 attributes: desiredAttributes,
324 dataProtectionPolicy: news.dataProtectionPolicy ?? observedPolicy,
325 tags: desiredTags,
326 };
327 }),
328 delete: Effect.fn(function* ({ output }) {
329 yield* sns
330 .deleteTopic({
331 TopicArn: output.topicArn,
332 })
333 .pipe(
334 Effect.catchTag("NotFoundException", () => Effect.void),
335 Effect.catchTag("InvalidParameterException", () => Effect.void),
336 );
337 }),
338 });
339
340const toTopicName = Effect.fn(function* (id: string, props: TopicProps) {
341 if (props.topicName) {
342 return props.topicName;
343 }
344
345 const baseName = yield* createPhysicalName({
346 id,
347 maxLength: props.fifo ? 256 - ".fifo".length : 256,
348 });
349
350 return props.fifo ? `${baseName}.fifo` : baseName;
351});
352
353const toAttributes = (props: TopicProps): Record<string, string> => ({
354 ...props.attributes,
355 ...(props.fifo ? { FifoTopic: "true" } : undefined),
356});
357
358const toTagMap = (tags: sns.Tag[] | undefined): Record<string, string> =>
359 Object.fromEntries(
360 (tags ?? [])
361 .filter(
362 (tag): tag is Required<Pick<sns.Tag, "Key" | "Value">> =>
363 typeof tag.Key === "string" && typeof tag.Value === "string",
364 )
365 .map((tag) => [tag.Key, tag.Value]),
366 );
367
368const toAttributeMap = (
369 attributes: Record<string, string | undefined> | undefined,
370): Record<string, string> =>
371 Object.fromEntries(
372 Object.entries(attributes ?? {}).filter(
373 (entry): entry is [string, string] => typeof entry[1] === "string",
374 ),
375 );
376
377const findTopicArnByName = Effect.fn(function* (topicName: string) {
378 const match = yield* sns.listTopics.items({}).pipe(
379 Stream.filter((topic) => topic.TopicArn?.split(":").at(-1) === topicName),
380 Stream.runHead,
381 Effect.map(Option.getOrUndefined),
382 );
383 return match?.TopicArn;
384});
385
386const readTopic = Effect.fn(function* ({
387 id,
388 topicArn,
389 topicName,
390}: {
391 id: string;
392 topicArn?: string;
393 topicName: string;
394}) {
395 const resolvedTopicArn = topicArn ?? (yield* findTopicArnByName(topicName));
396
397 if (!resolvedTopicArn) {
398 return undefined;
399 }
400
401 const topicState = yield* Effect.all(
402 [
403 sns.getTopicAttributes({
404 TopicArn: resolvedTopicArn,
405 }),
406 sns.listTagsForResource({
407 ResourceArn: resolvedTopicArn,
408 }),
409 sns
410 .getDataProtectionPolicy({
411 ResourceArn: resolvedTopicArn,
412 })
413 .pipe(
414 Effect.map((response) => response.DataProtectionPolicy),
415 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
416 ),
417 ],
418 { concurrency: "unbounded" },
419 ).pipe(
420 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
421 Effect.catchTag("InvalidParameterException", () =>
422 Effect.succeed(undefined),
423 ),
424 // `list()` hydrates every topic in the account, so a topic deleted by a
425 // parallel test between enumeration and hydration surfaces here —
426 // `listTagsForResource` reports it as `ResourceNotFoundException`. Treat a
427 // vanished topic as "not present" rather than failing the whole listing.
428 Effect.catchTag("ResourceNotFoundException", () =>
429 Effect.succeed(undefined),
430 ),
431 );
432
433 if (!topicState) {
434 return undefined;
435 }
436
437 const [attributes, tags, dataProtectionPolicy] = topicState;
438
439 const topicAttributes = toAttributeMap(attributes.Attributes);
440
441 const apiTags = toTagMap(tags.Tags);
442 return {
443 topicArn: resolvedTopicArn as TopicArn,
444 topicName: topicAttributes.TopicArn?.split(":").at(-1) ?? topicName,
445 fifo: topicAttributes.FifoTopic === "true",
446 attributes: topicAttributes,
447 dataProtectionPolicy,
448 tags: apiTags,
449 };
450});