1import * as securitylake from "@distilled.cloud/aws/securitylake";
2import * as Effect from "effect/Effect";
3import * as Redacted from "effect/Redacted";
4import * as Stream from "effect/Stream";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import type { Providers } from "../Providers.ts";
9import { retryWhileConflict } from "./internal.ts";
10
11/**
12 * HTTPS delivery settings for a subscriber notification.
13 */
14export interface SubscriberHttpsNotificationConfiguration {
15 /** The HTTPS endpoint Security Lake POSTs/PUTs object notifications to. */
16 endpoint: string;
17 /**
18 * ARN of the EventBridge API-destination role Security Lake assumes to
19 * invoke the endpoint.
20 */
21 targetRoleArn: string;
22 /**
23 * Name of the API-key header sent with each notification.
24 */
25 authorizationApiKeyName?: string;
26 /**
27 * Value of the API-key header sent with each notification. Held as a
28 * `Redacted` secret; it is never persisted or logged in plaintext.
29 */
30 authorizationApiKeyValue?: Redacted.Redacted<string>;
31 /**
32 * The HTTP method used to deliver notifications.
33 * @default "POST"
34 */
35 httpMethod?: "POST" | "PUT";
36}
37
38export interface SubscriberNotificationProps {
39 /**
40 * ID of the `SecurityLake.Subscriber` the notification is configured for.
41 * Changing this replaces the notification.
42 */
43 subscriberId: string;
44
45 /**
46 * Deliver new-object notifications to an AWS-managed SQS queue (Security
47 * Lake creates the queue). Exactly one of `sqs` or
48 * `httpsNotificationConfiguration` must be set.
49 * @default false
50 */
51 sqs?: boolean;
52
53 /**
54 * Deliver new-object notifications to a custom HTTPS endpoint. Exactly one
55 * of `sqs` or `httpsNotificationConfiguration` must be set.
56 */
57 httpsNotificationConfiguration?: SubscriberHttpsNotificationConfiguration;
58}
59
60/** @resource */
61export interface SubscriberNotification extends Resource<
62 "AWS.SecurityLake.SubscriberNotification",
63 SubscriberNotificationProps,
64 {
65 /** ID of the subscriber the notification belongs to. */
66 subscriberId: string;
67 /**
68 * The notification endpoint — the ARN of the AWS-managed SQS queue, or
69 * the configured HTTPS endpoint.
70 */
71 subscriberEndpoint: string | undefined;
72 },
73 never,
74 Providers
75> {}
76
77/**
78 * A Security Lake subscriber notification — notifies a data-access
79 * subscriber whenever new objects land in its Security Lake bucket, either
80 * via an AWS-managed SQS queue or a custom HTTPS endpoint.
81 *
82 * ### Notifying subscribers
83 * **Example:** SQS notifications
84 * ```typescript
85 * const notification = yield* SecurityLake.SubscriberNotification("Notify", {
86 * subscriberId: subscriber.subscriberId,
87 * sqs: true,
88 * });
89 * ```
90 *
91 * **Example:** HTTPS notifications with an API key
92 * ```typescript
93 * const notification = yield* SecurityLake.SubscriberNotification("Notify", {
94 * subscriberId: subscriber.subscriberId,
95 * httpsNotificationConfiguration: {
96 * endpoint: "https://ingest.example.com/securitylake",
97 * targetRoleArn: eventsRole.roleArn,
98 * authorizationApiKeyName: "x-api-key",
99 * authorizationApiKeyValue: Redacted.make("super-secret"),
100 * },
101 * });
102 * ```
103 */
104const SubscriberNotificationResource = Resource<SubscriberNotification>(
105 "AWS.SecurityLake.SubscriberNotification",
106);
107
108export { SubscriberNotificationResource as SubscriberNotification };
109
110// Build the wire NotificationConfiguration from props. Redacted secrets are
111// passed through intact — distilled's SensitiveString accepts Redacted.
112const toWireConfiguration = (
113 props: SubscriberNotificationProps,
114): securitylake.NotificationConfiguration =>
115 props.httpsNotificationConfiguration !== undefined
116 ? { httpsNotificationConfiguration: props.httpsNotificationConfiguration }
117 : { sqsNotificationConfiguration: {} };
118
119// A comparable, secret-free serialization of the desired configuration used
120// to skip no-op updates (the API does not expose the current configuration,
121// only the resulting endpoint — `olds` is the hint here).
122const configFingerprint = (props: {
123 sqs?: boolean;
124 httpsNotificationConfiguration?: SubscriberHttpsNotificationConfiguration;
125}): string => {
126 const https = props.httpsNotificationConfiguration;
127 return https !== undefined
128 ? JSON.stringify({
129 endpoint: https.endpoint,
130 targetRoleArn: https.targetRoleArn,
131 authorizationApiKeyName: https.authorizationApiKeyName,
132 // `olds` may have round-tripped through persisted state, so the
133 // secret may no longer be a live Redacted — guard before unwrapping.
134 authorizationApiKeyValue:
135 https.authorizationApiKeyValue !== undefined
136 ? Redacted.isRedacted(https.authorizationApiKeyValue)
137 ? Redacted.value(https.authorizationApiKeyValue)
138 : String(https.authorizationApiKeyValue)
139 : undefined,
140 httpMethod: https.httpMethod ?? "POST",
141 })
142 : "sqs";
143};
144
145const getSubscriberEndpoint = (subscriberId: string) =>
146 securitylake
147 .getSubscriber({ subscriberId })
148 .pipe(Effect.map((response) => response.subscriber?.subscriberEndpoint));
149
150export const SubscriberNotificationProvider = () =>
151 Provider.effect(
152 SubscriberNotificationResource,
153 Effect.gen(function* () {
154 return {
155 read: Effect.fn(function* ({ olds, output }) {
156 const subscriberId = output?.subscriberId ?? olds?.subscriberId;
157 if (subscriberId === undefined) return undefined;
158 // A missing subscriber (or an account that never onboarded
159 // Security Lake) means "no notification", not a failure.
160 const endpoint = yield* getSubscriberEndpoint(subscriberId).pipe(
161 Effect.catchTag(
162 ["ResourceNotFoundException", "UnauthorizedException"],
163 () => Effect.succeed(undefined),
164 ),
165 );
166 // Notifications carry no tags, so ownership can't be
167 // distinguished — read reports observed state directly.
168 return endpoint === undefined
169 ? undefined
170 : { subscriberId, subscriberEndpoint: endpoint };
171 }),
172
173 // A notification exists only for subscribers that report a
174 // notification endpoint.
175 list: () =>
176 securitylake.listSubscribers.items({}).pipe(
177 Stream.filter(
178 (subscriber) => subscriber.subscriberEndpoint !== undefined,
179 ),
180 Stream.runCollect,
181 Effect.map((subscribers) =>
182 [...subscribers].map((subscriber) => ({
183 subscriberId: subscriber.subscriberId,
184 subscriberEndpoint: subscriber.subscriberEndpoint,
185 })),
186 ),
187 // An account that never onboarded Security Lake has no
188 // subscribers to enumerate.
189 Effect.catchTag(
190 [
191 "AccessDeniedException",
192 "ResourceNotFoundException",
193 "UnauthorizedException",
194 ],
195 () => Effect.succeed([]),
196 ),
197 ),
198
199 // The notification is per-subscriber; changing the subscriber
200 // replaces it.
201 diff: Effect.fn(function* ({ news, olds }) {
202 if (!isResolved(news)) return undefined;
203 if (news.subscriberId !== olds.subscriberId) {
204 return { action: "replace" } as const;
205 }
206 }),
207
208 reconcile: Effect.fn(function* ({ news, olds, output, session }) {
209 const configuration = toWireConfiguration(news);
210
211 // 1. OBSERVE — the subscriber's current notification endpoint.
212 const observedEndpoint = yield* getSubscriberEndpoint(
213 news.subscriberId,
214 );
215
216 let endpoint = observedEndpoint;
217 if (observedEndpoint === undefined) {
218 // 2. ENSURE — create when missing. A ConflictException means a
219 // concurrent create won the race; converge via update.
220 endpoint = yield* securitylake
221 .createSubscriberNotification({
222 subscriberId: news.subscriberId,
223 configuration,
224 })
225 .pipe(
226 Effect.map((response) => response.subscriberEndpoint),
227 Effect.catchTag("ConflictException", () =>
228 securitylake
229 .updateSubscriberNotification({
230 subscriberId: news.subscriberId,
231 configuration,
232 })
233 .pipe(
234 retryWhileConflict,
235 Effect.map((response) => response.subscriberEndpoint),
236 ),
237 ),
238 );
239 } else if (
240 output === undefined ||
241 olds === undefined ||
242 configFingerprint(olds) !== configFingerprint(news)
243 ) {
244 // 3. SYNC — the API reports only the endpoint, not the
245 // configuration, so `olds` is the no-op hint; adoption always
246 // converges via an idempotent update.
247 endpoint = yield* securitylake
248 .updateSubscriberNotification({
249 subscriberId: news.subscriberId,
250 configuration,
251 })
252 .pipe(
253 retryWhileConflict,
254 Effect.map(
255 (response) => response.subscriberEndpoint ?? endpoint,
256 ),
257 );
258 }
259
260 // 4. RETURN fresh attributes.
261 const final =
262 endpoint ?? (yield* getSubscriberEndpoint(news.subscriberId));
263 yield* session.note(final ?? news.subscriberId);
264 return {
265 subscriberId: news.subscriberId,
266 subscriberEndpoint: final,
267 };
268 }),
269
270 delete: Effect.fn(function* ({ output }) {
271 yield* securitylake
272 .deleteSubscriberNotification({
273 subscriberId: output.subscriberId,
274 })
275 .pipe(
276 retryWhileConflict,
277 // Gone already, the subscriber was deleted first, or the data
278 // lake itself was offboarded.
279 Effect.catchTag(
280 [
281 "AccessDeniedException",
282 "ResourceNotFoundException",
283 "UnauthorizedException",
284 ],
285 () => Effect.void,
286 ),
287 );
288 }),
289 };
290 }),
291 );