1import * as AWS from "@/AWS";
2import * as Context from "effect/Context";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "handler.ts");
12
13const formatError = (error: unknown) =>
14 typeof error === "object" && error !== null && "_tag" in error
15 ? { ok: false as const, error: (error as { _tag: string })._tag }
16 : { ok: false as const, error: `${error}` };
17
18export class TopicAndQueue extends Context.Service<
19 TopicAndQueue,
20 {
21 topic: AWS.SNS.Topic;
22 queue: AWS.SQS.Queue;
23 subscription: AWS.SNS.Subscription;
24 subscriptionAttrsQueue: AWS.SQS.Queue;
25 queueSubscription: AWS.SNS.Subscription;
26 }
27>()("TopicAndQueue") {}
28
29export const TopicAndQueueLive = Layer.effect(
30 TopicAndQueue,
31 Effect.gen(function* () {
32 const topic = yield* AWS.SNS.Topic("TestTopic", {
33 attributes: {
34 DisplayName: "sns-test-topic",
35 },
36 });
37 const queue = yield* AWS.SQS.Queue("NotificationsQueue");
38 const subscriptionAttrsQueue = yield* AWS.SQS.Queue(
39 "SubscriptionAttrsQueue",
40 );
41 const queueSubscription = yield* AWS.SNS.Subscription(
42 "QueueFixtureSubscription",
43 {
44 topicArn: topic.topicArn,
45 protocol: "sqs",
46 endpoint: subscriptionAttrsQueue.queueArn,
47 returnSubscriptionArn: true,
48 },
49 );
50 return {
51 topic,
52 queue,
53 // ConfirmSubscription only injects the parent TopicArn, so the existing
54 // queue subscription exercises that binding without a second Lambda.
55 subscription: queueSubscription,
56 subscriptionAttrsQueue,
57 queueSubscription,
58 };
59 }),
60);
61
62export class SNSApiFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
63 "SNSApiFunction",
64) {}
65
66export const SNSApiFunctionLive = SNSApiFunction.make(
67 {
68 main,
69 functionUrl: true,
70 // The sink's bounded partial-failure retry can sleep up to ~6s, which
71 // exceeds Lambda's 3s default timeout (see PATTERNS §7).
72 timeout: Duration.seconds(30),
73 env: {
74 DEBUG: "true",
75 },
76 },
77 Effect.gen(function* () {
78 const { topic, queue, subscription, queueSubscription } =
79 yield* TopicAndQueue;
80
81 const publish = yield* AWS.SNS.Publish(topic);
82 const publishBatch = yield* AWS.SNS.PublishBatch(topic);
83 const getTopicAttributes = yield* AWS.SNS.GetTopicAttributes(topic);
84 const setTopicAttributes = yield* AWS.SNS.SetTopicAttributes(topic);
85 const addPermission = yield* AWS.SNS.AddPermission(topic);
86 const removePermission = yield* AWS.SNS.RemovePermission(topic);
87 const getDataProtectionPolicy =
88 yield* AWS.SNS.GetDataProtectionPolicy(topic);
89 const putDataProtectionPolicy =
90 yield* AWS.SNS.PutDataProtectionPolicy(topic);
91 const listTopics = yield* AWS.SNS.ListTopics();
92 const listSubscriptions = yield* AWS.SNS.ListSubscriptions();
93 const listSubscriptionsByTopic =
94 yield* AWS.SNS.ListSubscriptionsByTopic(topic);
95 const listTagsForResource = yield* AWS.SNS.ListTagsForResource(topic);
96 const tagResource = yield* AWS.SNS.TagResource(topic);
97 const untagResource = yield* AWS.SNS.UntagResource(topic);
98 const getSubscriptionAttributes =
99 yield* AWS.SNS.GetSubscriptionAttributes(queueSubscription);
100 const setSubscriptionAttributes =
101 yield* AWS.SNS.SetSubscriptionAttributes(queueSubscription);
102 const confirmSubscription =
103 yield* AWS.SNS.ConfirmSubscription(subscription);
104 const subscribeToTopic = yield* AWS.SNS.Subscribe(topic);
105 const unsubscribe = yield* AWS.SNS.Unsubscribe();
106 const publishSms = yield* AWS.SNS.PublishSms();
107 const checkIfPhoneNumberIsOptedOut =
108 yield* AWS.SNS.CheckIfPhoneNumberIsOptedOut();
109 const optInPhoneNumber = yield* AWS.SNS.OptInPhoneNumber();
110 const listPhoneNumbersOptedOut = yield* AWS.SNS.ListPhoneNumbersOptedOut();
111 const getSmsAttributes = yield* AWS.SNS.GetSMSAttributes();
112 const setSmsAttributes = yield* AWS.SNS.SetSMSAttributes();
113 const listOriginationNumbers = yield* AWS.SNS.ListOriginationNumbers();
114 const getSandboxStatus = yield* AWS.SNS.GetSMSSandboxAccountStatus();
115 const listSandboxNumbers = yield* AWS.SNS.ListSMSSandboxPhoneNumbers();
116 const createSandboxNumber = yield* AWS.SNS.CreateSMSSandboxPhoneNumber();
117 const verifySandboxNumber = yield* AWS.SNS.VerifySMSSandboxPhoneNumber();
118 const deleteSandboxNumber = yield* AWS.SNS.DeleteSMSSandboxPhoneNumber();
119 const listPlatformApplications = yield* AWS.SNS.ListPlatformApplications();
120 const sink = yield* AWS.SNS.TopicSink(topic);
121 const QueueArn = yield* queue.queueArn;
122 const TopicArn = yield* topic.topicArn;
123 const accountId = TopicArn.pipe(
124 Effect.map((topicArn) => topicArn.split(":")[4] ?? ""),
125 );
126
127 const queueSink = yield* AWS.SQS.QueueSink(queue);
128
129 yield* AWS.SNS.consumeTopicNotifications(topic, (stream) =>
130 stream.pipe(
131 Stream.map((notification) => ({
132 MessageBody: JSON.stringify({
133 topicArn: notification.TopicArn,
134 message: notification.Message,
135 subject: notification.Subject,
136 messageId: notification.MessageId,
137 }),
138 })),
139 Stream.run(queueSink),
140 Effect.orDie,
141 ),
142 );
143
144 return {
145 fetch: Effect.gen(function* () {
146 const request = yield* HttpServerRequest;
147 const url = new URL(request.originalUrl);
148 const pathname = url.pathname;
149
150 if (request.method === "GET" && pathname === "/ready") {
151 return yield* HttpServerResponse.json({ ok: true });
152 }
153
154 if (request.method === "POST" && pathname === "/publish") {
155 const body = (yield* request.json) as {
156 message: string;
157 subject?: string;
158 };
159 const response = yield* publish({
160 Message: body.message,
161 Subject: body.subject,
162 });
163 return yield* HttpServerResponse.json(response);
164 }
165
166 if (request.method === "POST" && pathname === "/publish-batch") {
167 const body = (yield* request.json) as { messages: string[] };
168 const response = yield* publishBatch({
169 PublishBatchRequestEntries: body.messages.map((message, index) => ({
170 Id: `${index}`,
171 Message: message,
172 })),
173 });
174 return yield* HttpServerResponse.json(response);
175 }
176
177 if (request.method === "POST" && pathname === "/sink") {
178 const body = (yield* request.json) as { messages: string[] };
179 yield* Stream.fromIterable(body.messages).pipe(
180 Stream.map((message) => ({ Message: message })),
181 Stream.run(sink),
182 );
183 return yield* HttpServerResponse.json({ ok: true });
184 }
185
186 if (request.method === "GET" && pathname === "/topic-attributes") {
187 return yield* HttpServerResponse.json(yield* getTopicAttributes());
188 }
189
190 if (request.method === "POST" && pathname === "/topic-attributes") {
191 const body = (yield* request.json) as {
192 name: string;
193 value?: string;
194 };
195 return yield* HttpServerResponse.json(
196 yield* setTopicAttributes({
197 AttributeName: body.name,
198 AttributeValue: body.value,
199 }),
200 );
201 }
202
203 if (request.method === "POST" && pathname === "/add-permission") {
204 const label = "FixturePublishPermission";
205 const response = yield* addPermission({
206 Label: label,
207 AWSAccountId: [yield* accountId],
208 ActionName: ["Publish"],
209 });
210 return yield* HttpServerResponse.json({ label, response });
211 }
212
213 if (request.method === "POST" && pathname === "/remove-permission") {
214 const response = yield* removePermission({
215 Label: "FixturePublishPermission",
216 });
217 return yield* HttpServerResponse.json(response);
218 }
219
220 if (
221 request.method === "GET" &&
222 pathname === "/data-protection-policy"
223 ) {
224 return yield* HttpServerResponse.json(
225 yield* getDataProtectionPolicy().pipe(
226 Effect.catch((error) => Effect.succeed(formatError(error))),
227 ),
228 );
229 }
230
231 if (
232 request.method === "POST" &&
233 pathname === "/data-protection-policy"
234 ) {
235 const body = (yield* request.json) as { policy: string };
236 const response = yield* putDataProtectionPolicy({
237 DataProtectionPolicy: body.policy,
238 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
239 return yield* HttpServerResponse.json(response);
240 }
241
242 if (request.method === "GET" && pathname === "/topics") {
243 return yield* HttpServerResponse.json(yield* listTopics());
244 }
245
246 if (request.method === "GET" && pathname === "/subscriptions") {
247 return yield* HttpServerResponse.json(yield* listSubscriptions());
248 }
249
250 if (
251 request.method === "GET" &&
252 pathname === "/subscriptions-by-topic"
253 ) {
254 return yield* HttpServerResponse.json(
255 yield* listSubscriptionsByTopic(),
256 );
257 }
258
259 if (request.method === "GET" && pathname === "/tags") {
260 return yield* HttpServerResponse.json(yield* listTagsForResource());
261 }
262
263 if (request.method === "POST" && pathname === "/tags") {
264 const body = (yield* request.json) as {
265 key: string;
266 value: string;
267 };
268 return yield* HttpServerResponse.json(
269 yield* tagResource({
270 Tags: [{ Key: body.key, Value: body.value }],
271 }),
272 );
273 }
274
275 if (request.method === "DELETE" && pathname === "/tags") {
276 const body = (yield* request.json) as { keys: string[] };
277 return yield* HttpServerResponse.json(
278 yield* untagResource({
279 TagKeys: body.keys,
280 }),
281 );
282 }
283
284 if (
285 request.method === "GET" &&
286 pathname === "/subscription-attributes"
287 ) {
288 return yield* HttpServerResponse.json(
289 yield* getSubscriptionAttributes(),
290 );
291 }
292
293 if (
294 request.method === "POST" &&
295 pathname === "/subscription-attributes"
296 ) {
297 const body = (yield* request.json) as {
298 name: string;
299 value?: string;
300 };
301 return yield* HttpServerResponse.json(
302 yield* setSubscriptionAttributes({
303 AttributeName: body.name,
304 AttributeValue: body.value,
305 }),
306 );
307 }
308
309 if (request.method === "POST" && pathname === "/subscribe-cycle") {
310 // Subscribe the (otherwise unsubscribed) notifications queue to the
311 // topic at runtime and immediately unsubscribe it — proves both the
312 // `Subscribe` and `Unsubscribe` bindings against the live API
313 // without leaving a subscription behind that would double-deliver
314 // later publishes.
315 const subscribed = yield* subscribeToTopic({
316 Protocol: "sqs",
317 Endpoint: yield* QueueArn,
318 ReturnSubscriptionArn: true,
319 });
320 if (!subscribed.SubscriptionArn) {
321 return yield* HttpServerResponse.json(
322 { ok: false, error: "no SubscriptionArn" },
323 { status: 500 },
324 );
325 }
326 yield* unsubscribe({ SubscriptionArn: subscribed.SubscriptionArn });
327 return yield* HttpServerResponse.json({
328 ok: true,
329 subscriptionArn: subscribed.SubscriptionArn,
330 });
331 }
332
333 if (request.method === "GET" && pathname === "/sms/attributes") {
334 return yield* HttpServerResponse.json(
335 yield* getSmsAttributes().pipe(
336 Effect.catch((error) => Effect.succeed(formatError(error))),
337 ),
338 );
339 }
340
341 if (request.method === "POST" && pathname === "/sms/attributes") {
342 const body = (yield* request.json) as { type: string };
343 return yield* HttpServerResponse.json(
344 yield* setSmsAttributes({
345 attributes: { DefaultSMSType: body.type },
346 }).pipe(
347 Effect.catch((error) => Effect.succeed(formatError(error))),
348 ),
349 );
350 }
351
352 if (request.method === "GET" && pathname === "/sms/opted-out") {
353 return yield* HttpServerResponse.json(
354 yield* listPhoneNumbersOptedOut().pipe(
355 Effect.catch((error) => Effect.succeed(formatError(error))),
356 ),
357 );
358 }
359
360 if (request.method === "POST" && pathname === "/sms/check-opt-out") {
361 const body = (yield* request.json) as { phoneNumber: string };
362 return yield* HttpServerResponse.json(
363 yield* checkIfPhoneNumberIsOptedOut({
364 phoneNumber: body.phoneNumber,
365 }).pipe(
366 Effect.catch((error) => Effect.succeed(formatError(error))),
367 ),
368 );
369 }
370
371 if (request.method === "POST" && pathname === "/sms/opt-in") {
372 const body = (yield* request.json) as { phoneNumber: string };
373 const response = yield* optInPhoneNumber({
374 phoneNumber: body.phoneNumber,
375 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
376 return yield* HttpServerResponse.json(response);
377 }
378
379 if (
380 request.method === "GET" &&
381 pathname === "/sms/origination-numbers"
382 ) {
383 return yield* HttpServerResponse.json(
384 yield* listOriginationNumbers().pipe(
385 Effect.catch((error) => Effect.succeed(formatError(error))),
386 ),
387 );
388 }
389
390 if (request.method === "GET" && pathname === "/sms/sandbox-status") {
391 return yield* HttpServerResponse.json(
392 yield* getSandboxStatus().pipe(
393 Effect.catch((error) => Effect.succeed(formatError(error))),
394 ),
395 );
396 }
397
398 if (request.method === "GET" && pathname === "/sms/sandbox-numbers") {
399 return yield* HttpServerResponse.json(
400 yield* listSandboxNumbers().pipe(
401 Effect.catch((error) => Effect.succeed(formatError(error))),
402 ),
403 );
404 }
405
406 if (request.method === "POST" && pathname === "/sms/sandbox-create") {
407 const body = (yield* request.json) as { phoneNumber: string };
408 const response = yield* createSandboxNumber({
409 PhoneNumber: body.phoneNumber,
410 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
411 return yield* HttpServerResponse.json(response);
412 }
413
414 if (request.method === "POST" && pathname === "/sms/sandbox-verify") {
415 const body = (yield* request.json) as {
416 phoneNumber: string;
417 otp: string;
418 };
419 const response = yield* verifySandboxNumber({
420 PhoneNumber: body.phoneNumber,
421 OneTimePassword: body.otp,
422 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
423 return yield* HttpServerResponse.json(response);
424 }
425
426 if (request.method === "POST" && pathname === "/sms/sandbox-delete") {
427 const body = (yield* request.json) as { phoneNumber: string };
428 const response = yield* deleteSandboxNumber({
429 PhoneNumber: body.phoneNumber,
430 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
431 return yield* HttpServerResponse.json(response);
432 }
433
434 if (request.method === "POST" && pathname === "/sms/publish") {
435 const body = (yield* request.json) as {
436 phoneNumber: string;
437 message: string;
438 };
439 const response = yield* publishSms({
440 PhoneNumber: body.phoneNumber,
441 Message: body.message,
442 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
443 return yield* HttpServerResponse.json(response);
444 }
445
446 if (request.method === "GET" && pathname === "/platform/applications") {
447 return yield* HttpServerResponse.json(
448 yield* listPlatformApplications().pipe(
449 Effect.catch((error) => Effect.succeed(formatError(error))),
450 ),
451 );
452 }
453
454 if (request.method === "POST" && pathname === "/confirm-subscription") {
455 const body = (yield* request.json) as { token: string };
456 const response = yield* confirmSubscription({
457 Token: body.token,
458 }).pipe(Effect.catch((error) => Effect.succeed(formatError(error))));
459 return yield* HttpServerResponse.json(response);
460 }
461
462 return yield* HttpServerResponse.json(
463 { error: "Not found", method: request.method, pathname },
464 { status: 404 },
465 );
466 }).pipe(Effect.orDie),
467 };
468 }).pipe(
469 Effect.provide(
470 Layer.provideMerge(
471 Layer.mergeAll(
472 TopicAndQueueLive,
473 AWS.Lambda.TopicEventSource,
474 AWS.SNS.TopicSinkHttp,
475 AWS.SQS.QueueSinkHttp,
476 ),
477 Layer.mergeAll(
478 AWS.SNS.AddPermissionHttp,
479 AWS.SNS.CheckIfPhoneNumberIsOptedOutHttp,
480 AWS.SNS.ConfirmSubscriptionHttp,
481 AWS.SNS.CreateSMSSandboxPhoneNumberHttp,
482 AWS.SNS.DeleteSMSSandboxPhoneNumberHttp,
483 AWS.SNS.GetSMSAttributesHttp,
484 AWS.SNS.GetSMSSandboxAccountStatusHttp,
485 AWS.SNS.ListOriginationNumbersHttp,
486 AWS.SNS.ListPhoneNumbersOptedOutHttp,
487 AWS.SNS.ListPlatformApplicationsHttp,
488 AWS.SNS.ListSMSSandboxPhoneNumbersHttp,
489 AWS.SNS.OptInPhoneNumberHttp,
490 AWS.SNS.PublishSmsHttp,
491 AWS.SNS.SetSMSAttributesHttp,
492 AWS.SNS.SubscribeHttp,
493 AWS.SNS.UnsubscribeHttp,
494 AWS.SNS.VerifySMSSandboxPhoneNumberHttp,
495 AWS.SNS.GetDataProtectionPolicyHttp,
496 AWS.SNS.GetSubscriptionAttributesHttp,
497 AWS.SNS.GetTopicAttributesHttp,
498 AWS.SNS.ListSubscriptionsByTopicHttp,
499 AWS.SNS.ListSubscriptionsHttp,
500 AWS.SNS.ListTagsForResourceHttp,
501 AWS.SNS.ListTopicsHttp,
502 AWS.SNS.PublishBatchHttp,
503 AWS.SNS.PublishHttp,
504 AWS.SNS.PutDataProtectionPolicyHttp,
505 AWS.SNS.RemovePermissionHttp,
506 AWS.SNS.SetSubscriptionAttributesHttp,
507 AWS.SNS.SetTopicAttributesHttp,
508 AWS.SNS.TagResourceHttp,
509 AWS.SNS.UntagResourceHttp,
510 AWS.SQS.SendMessageBatchHttp,
511 ),
512 ),
513 ),
514 ),
515).pipe(Layer.provideMerge(TopicAndQueueLive));
516
517export default SNSApiFunctionLive;