1import * as Effect from "effect/Effect";
2import * as Stream from "effect/Stream";
3import {
4 consumeTopicNotifications,
5 type TopicNotification,
6} from "../SNS/TopicEventSource.ts";
7import type { Topic } from "../SNS/Topic.ts";
8import type { Directory } from "./Directory.ts";
9import { EventTopic } from "./EventTopic.ts";
10
11/**
12 * A directory status notification delivered to the handler. Directory
13 * Service publishes a message whenever the directory changes stage — e.g.
14 * from `Active` to `Impaired` or `Inoperable`, and back to `Active`.
15 */
16export interface DirectoryStatusEvent {
17 /** The raw SNS notification as delivered. */
18 notification: TopicNotification;
19 /**
20 * The parsed JSON body of the notification, when the message is JSON
21 * (AWS does not document a strict schema for directory status messages).
22 */
23 detail: Record<string, unknown> | undefined;
24}
25
26const parseDirectoryStatus = (
27 notification: TopicNotification,
28): DirectoryStatusEvent => {
29 let detail: Record<string, unknown> | undefined;
30 try {
31 const parsed: unknown = JSON.parse(notification.Message);
32 detail =
33 typeof parsed === "object" && parsed !== null
34 ? (parsed as Record<string, unknown>)
35 : undefined;
36 } catch {
37 detail = undefined;
38 }
39 return { notification, detail };
40};
41
42/**
43 * Event source connecting an AWS Directory Service {@link Directory}'s
44 * status notifications to the hosting Lambda function. Directory Service's
45 * native event mechanism is Amazon SNS: at deploy time this registers the
46 * directory as a publisher to the given {@link Topic} (an
47 * {@link EventTopic} association) and subscribes the host function to the
48 * topic; at runtime delivered status messages are dispatched to the
49 * handler.
50 *
51 * Provide `AWS.Lambda.TopicEventSource` on the Function effect (the SNS
52 * subscription machinery this event source delegates to).
53 *
54 * @example Alert When the Directory Becomes Impaired
55 * ```typescript
56 * export default AlertFunction.make(
57 * { main: import.meta.url },
58 * Effect.gen(function* () {
59 * const topic = yield* SNS.Topic("DirectoryStatus", {});
60 * const directory = yield* DirectoryService.Directory("Corp", { ... });
61 *
62 * yield* DirectoryService.consumeDirectoryStatus(directory, topic, (events) =>
63 * Stream.runForEach(events, (event) =>
64 * Effect.logError(`directory status: ${event.notification.Message}`),
65 * ),
66 * );
67 * return {};
68 * }).pipe(Effect.provide(Lambda.TopicEventSource)),
69 * );
70 * ```
71 */
72export const consumeDirectoryStatus = <StreamReq = never, Req = never>(
73 directory: Directory,
74 topic: Topic,
75 process: (
76 events: Stream.Stream<DirectoryStatusEvent, never, StreamReq>,
77 ) => Effect.Effect<void, never, Req>,
78) =>
79 Effect.gen(function* () {
80 // Deploy-time: associate the directory with the SNS topic so status
81 // changes are published to it.
82 yield* EventTopic(`${directory.LogicalId}${topic.LogicalId}Status`, {
83 directoryId: directory.directoryId,
84 topicName: topic.topicName,
85 });
86 // Subscribe the host function to the topic and dispatch parsed status
87 // messages to the handler.
88 yield* consumeTopicNotifications(topic, (stream) =>
89 process(stream.pipe(Stream.map(parseDirectoryStatus))),
90 );
91 });