1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as SNS from "@distilled.cloud/aws/sns";
4import * as SQS from "@distilled.cloud/aws/sqs";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import {
12 SNSApiFunction,
13 SNSApiFunctionLive,
14 TopicAndQueue,
15} from "./handler.ts";
16
17const { test } = Test.make({ providers: AWS.providers() });
18
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(14),
22]);
23
24describe.sequential(
25 "SNS Bindings",
26 {
27 tags: [
28 "provider:aws",
29 "provider:aws:lambda",
30 "provider:aws:sns",
31 "provider:aws:sqs",
32 "live",
33 ],
34 },
35 () => {
36 test.provider(
37 "exercises the SNS bindings surface against a live fixture",
38 (stack) =>
39 Effect.gen(function* () {
40 yield* Effect.logInfo(
41 "SNS test setup: destroying previous resources",
42 );
43 if (!process.env.NO_DESTROY) {
44 yield* stack.destroy();
45 }
46
47 yield* Effect.logInfo("SNS test setup: deploying fixture");
48 const deployed = yield* stack.deploy(
49 Effect.gen(function* () {
50 const { topic, queue, subscription } = yield* TopicAndQueue;
51
52 const apiFunction = yield* SNSApiFunction;
53
54 return {
55 apiFunction,
56 topic,
57 queue,
58 subscription,
59 };
60 }).pipe(Effect.provide(SNSApiFunctionLive)),
61 );
62
63 const baseUrl = deployed.apiFunction.functionUrl!.replace(/\/+$/, "");
64 const queueUrl = deployed.queue.queueUrl;
65 const topicArn = deployed.topic.topicArn;
66 const subscriptionArn = deployed.subscription.subscriptionArn;
67
68 // Gate on an IAM-backed binding call (read-only GetTopicAttributes),
69 // not the no-op `/ready` route. A fresh role's inline policy is
70 // eventually consistent: the function code can be live (200 on
71 // `/ready`) seconds before STS has propagated the `sns:*` grants, so a
72 // `/ready` gate lets the first `/publish` race the propagation window
73 // and 500 with AuthorizationErrorException. Hitting `/topic-attributes`
74 // here waits until the policy is actually live before any assertions.
75 const readinessUrl = `${baseUrl}/topic-attributes`;
76 yield* Effect.logInfo(
77 `SNS test setup: probing IAM readiness at ${readinessUrl} (150s bounded budget)`,
78 );
79
80 yield* HttpClient.get(readinessUrl).pipe(
81 Effect.timeout("8 seconds"),
82 Effect.flatMap((response) =>
83 response.status === 200
84 ? Effect.succeed(response)
85 : Effect.fail(
86 new Error(`Function not ready: ${response.status}`),
87 ),
88 ),
89 Effect.tap(() =>
90 Effect.logInfo("SNS test setup: fixture responded successfully"),
91 ),
92 Effect.tapError((error) =>
93 Effect.logWarning(
94 `SNS test setup: fixture not ready yet (${String(error)})`,
95 ),
96 ),
97 Effect.retry({ schedule: readinessPolicy }),
98 );
99
100 const getJson = (path: string) =>
101 HttpClient.get(`${baseUrl}${path}`).pipe(
102 Effect.tap((response) =>
103 Effect.flatMap(response.text, (text) =>
104 Effect.logInfo(`GET ${path} -> ${text.slice(0, 400)}`),
105 ),
106 ),
107 Effect.flatMap((response) => response.json),
108 Effect.timeout("20 seconds"),
109 );
110
111 const postJson = (path: string, body: unknown) =>
112 HttpClient.execute(
113 HttpClientRequest.bodyJsonUnsafe(
114 HttpClientRequest.post(`${baseUrl}${path}`),
115 body,
116 ),
117 ).pipe(
118 Effect.tap((response) =>
119 Effect.flatMap(response.text, Effect.logInfo),
120 ),
121 Effect.flatMap((response) => response.json),
122 Effect.timeout("20 seconds"),
123 );
124
125 const deleteJson = (path: string, body: unknown) =>
126 HttpClient.execute(
127 HttpClientRequest.bodyJsonUnsafe(
128 HttpClientRequest.delete(`${baseUrl}${path}`),
129 body,
130 ),
131 ).pipe(
132 Effect.flatMap((response) => response.json),
133 Effect.timeout("20 seconds"),
134 );
135
136 const bufferedQueueMessages: Array<{
137 message: string;
138 topicArn: string;
139 subject?: string;
140 }> = [];
141
142 const waitForQueueMessage = Effect.fn(function* (
143 predicate: (body: {
144 message: string;
145 topicArn: string;
146 subject?: string;
147 }) => boolean,
148 ) {
149 const takeBuffered = () => {
150 const index = bufferedQueueMessages.findIndex(predicate);
151 return index < 0
152 ? undefined
153 : bufferedQueueMessages.splice(index, 1)[0];
154 };
155
156 return yield* Effect.gen(function* () {
157 const buffered = takeBuffered();
158 if (buffered) return buffered;
159
160 const result = yield* SQS.receiveMessage({
161 QueueUrl: queueUrl,
162 MaxNumberOfMessages: 10,
163 WaitTimeSeconds: 5,
164 VisibilityTimeout: 30,
165 });
166 const messages = (result.Messages ?? []).filter(
167 (message) => message.Body && message.ReceiptHandle,
168 );
169 if (messages.length === 0) {
170 return yield* Effect.fail(new QueueMessageNotReady());
171 }
172
173 yield* Effect.forEach(
174 messages,
175 (message) =>
176 SQS.deleteMessage({
177 QueueUrl: queueUrl,
178 ReceiptHandle: message.ReceiptHandle!,
179 }),
180 { concurrency: "unbounded" },
181 );
182 bufferedQueueMessages.push(
183 ...messages.map(
184 (message) =>
185 JSON.parse(message.Body!) as {
186 message: string;
187 topicArn: string;
188 subject?: string;
189 },
190 ),
191 );
192
193 const received = takeBuffered();
194 return yield* received
195 ? Effect.succeed(received)
196 : Effect.fail(new QueueMessageNotReady());
197 }).pipe(
198 // Five-second long polls keep the complete retry window below 45s.
199 Effect.retry({
200 while: (error) => error._tag === "QueueMessageNotReady",
201 schedule: Schedule.recurs(8),
202 }),
203 );
204 });
205
206 const waitForQueueMessages = Effect.fn(function* (count: number) {
207 const bodies: Array<{
208 message: string;
209 topicArn: string;
210 subject?: string;
211 }> = [];
212
213 while (bodies.length < count) {
214 bodies.push(yield* waitForQueueMessage(() => true));
215 }
216
217 return bodies;
218 });
219
220 // Publish
221 yield* Effect.gen(function* () {
222 const marker = `publish-${crypto.randomUUID()}`;
223 const response = yield* postJson("/publish", {
224 message: marker,
225 subject: "PublishTest",
226 });
227
228 expect((response as any).MessageId).toBeTruthy();
229
230 const queued = yield* waitForQueueMessage(
231 (body) => body.message === marker,
232 );
233 expect((queued as any).topicArn).toBe(topicArn);
234 expect((queued as any).subject).toBe("PublishTest");
235 });
236
237 // PublishBatch
238 yield* Effect.gen(function* () {
239 const first = `batch-1-${crypto.randomUUID()}`;
240 const second = `batch-2-${crypto.randomUUID()}`;
241
242 const response = yield* postJson("/publish-batch", {
243 messages: [first, second],
244 });
245
246 expect(((response as any).Successful ?? []).length).toBe(2);
247
248 const bodies = yield* waitForQueueMessages(2);
249 const messages = bodies.map((body) => body.message);
250 expect(messages).toContain(first);
251 expect(messages).toContain(second);
252 });
253
254 // GetTopicAttributes
255 yield* Effect.gen(function* () {
256 const response = yield* getJson("/topic-attributes");
257 expect((response as any).Attributes.TopicArn).toBe(topicArn);
258 });
259
260 // SetTopicAttributes — eventual consistency. SNS can take a few
261 // seconds to propagate a SetTopicAttributes change, and during
262 // that window GetTopicAttributes may return a payload without
263 // the updated key. Poll briefly.
264 yield* Effect.gen(function* () {
265 yield* postJson("/topic-attributes", {
266 name: "DisplayName",
267 value: "updated-display-name",
268 });
269
270 yield* Effect.gen(function* () {
271 const response = yield* getJson("/topic-attributes");
272 const displayName = (response as any).Attributes?.DisplayName;
273 if (displayName !== "updated-display-name") {
274 return yield* Effect.fail(new TopicAttributeNotPropagated());
275 }
276 }).pipe(
277 Effect.retry({
278 while: (e) => e._tag === "TopicAttributeNotPropagated",
279 schedule: Schedule.max([
280 Schedule.fixed("1 second"),
281 Schedule.recurs(15),
282 ]),
283 }),
284 );
285 });
286
287 // AddPermission
288 yield* Effect.gen(function* () {
289 yield* postJson("/add-permission", {});
290 const response = yield* getJson("/topic-attributes");
291 expect((response as any).Attributes.Policy).toContain(
292 "FixturePublishPermission",
293 );
294 });
295
296 // RemovePermission
297 yield* Effect.gen(function* () {
298 yield* postJson("/add-permission", {});
299 yield* postJson("/remove-permission", {});
300 const response = yield* getJson("/topic-attributes");
301 expect((response as any).Attributes.Policy ?? "").not.toContain(
302 "FixturePublishPermission",
303 );
304 });
305
306 // GetDataProtectionPolicy
307 yield* Effect.gen(function* () {
308 const response = yield* getJson("/data-protection-policy");
309 if ((response as any).ok === false) {
310 expect((response as any).error).toBeTruthy();
311 } else {
312 expect(response).toBeDefined();
313 }
314 });
315
316 // PutDataProtectionPolicy
317 yield* Effect.gen(function* () {
318 const response = yield* postJson("/data-protection-policy", {
319 policy: "{}",
320 });
321 expect(response).toBeDefined();
322 });
323
324 // ListTopics (account-wide) — the alchemy binding wraps SNS's
325 // single-page `ListTopics` operation, which returns up to 100 topics.
326 // On a busy account our brand-new topic may simply be on a later page,
327 // so just assert the binding works (returns an Array of topics). Our
328 // specific ARN is verified via the topic-scoped `topic-attributes`
329 // call above.
330 yield* Effect.gen(function* () {
331 const response = yield* getJson("/topics");
332 const topics = (response as any).Topics ?? [];
333 expect(Array.isArray(topics)).toBe(true);
334 });
335
336 // ListSubscriptions (account-wide) — the alchemy binding wraps
337 // SNS's single-page `ListSubscriptions` operation, which returns
338 // up to 100 subscriptions. On a busy account our brand-new
339 // subscription may simply be on a later page. Just assert the
340 // binding works (returns an Array of subscriptions). Our specific
341 // ARN is verified via the topic-scoped call below.
342 yield* Effect.gen(function* () {
343 const response = yield* getJson("/subscriptions");
344 const subscriptions = (response as any).Subscriptions ?? [];
345 expect(Array.isArray(subscriptions)).toBe(true);
346 });
347
348 // ListSubscriptionsByTopic — scoped to the topic we just created,
349 // so propagation is tight. SNS still has eventual consistency on
350 // brand-new subscriptions; poll for ~30s.
351 yield* Effect.gen(function* () {
352 const arns = yield* Effect.gen(function* () {
353 const response = yield* getJson("/subscriptions-by-topic");
354 const arns = ((response as any).Subscriptions ?? []).map(
355 (subscription: any) => subscription.SubscriptionArn,
356 );
357 if (!arns.includes(subscriptionArn)) {
358 return yield* Effect.fail(new SubscriptionNotListed());
359 }
360 return arns;
361 }).pipe(
362 Effect.retry({
363 while: (e) => e._tag === "SubscriptionNotListed",
364 schedule: Schedule.max([
365 Schedule.fixed("2 seconds"),
366 Schedule.recurs(15),
367 ]),
368 }),
369 );
370 expect(arns).toContain(subscriptionArn);
371 });
372
373 // ListTagsForResource
374 yield* Effect.gen(function* () {
375 const response = yield* getJson("/tags");
376 const keys = ((response as any).Tags ?? []).map(
377 (tag: any) => tag.Key,
378 );
379 expect(keys).toContain("alchemy::stack");
380 expect(keys).toContain("alchemy::stage");
381 expect(keys).toContain("alchemy::id");
382 });
383
384 // TagResource
385 yield* Effect.gen(function* () {
386 yield* postJson("/tags", {
387 key: "sns-binding-test",
388 value: "true",
389 });
390 const response = yield* getJson("/tags");
391 const tags = Object.fromEntries(
392 ((response as any).Tags ?? []).map((tag: any) => [
393 tag.Key,
394 tag.Value,
395 ]),
396 );
397 expect(tags["sns-binding-test"]).toBe("true");
398 });
399
400 // UntagResource
401 yield* Effect.gen(function* () {
402 yield* postJson("/tags", {
403 key: "sns-remove-test",
404 value: "true",
405 });
406 yield* deleteJson("/tags", {
407 keys: ["sns-remove-test"],
408 });
409 const response = yield* getJson("/tags");
410 const keys = ((response as any).Tags ?? []).map(
411 (tag: any) => tag.Key,
412 );
413 expect(keys).not.toContain("sns-remove-test");
414 });
415
416 // GetSubscriptionAttributes
417 yield* Effect.gen(function* () {
418 const response = yield* getJson("/subscription-attributes");
419 expect((response as any).Attributes.Protocol).toBe("sqs");
420 });
421
422 // SetSubscriptionAttributes
423 yield* Effect.gen(function* () {
424 const response = yield* postJson("/subscription-attributes", {
425 name: "RawMessageDelivery",
426 value: "true",
427 }).pipe(
428 Effect.flatMap(() => getJson("/subscription-attributes")),
429 Effect.ensuring(
430 postJson("/subscription-attributes", {
431 name: "RawMessageDelivery",
432 value: "false",
433 }).pipe(Effect.ignore),
434 ),
435 );
436 expect((response as any).Attributes.RawMessageDelivery).toBe(
437 "true",
438 );
439 });
440
441 // ConfirmSubscription
442 yield* Effect.gen(function* () {
443 const response = yield* postJson("/confirm-subscription", {
444 token: "invalid-token",
445 });
446 expect((response as any).ok).toBe(false);
447 expect((response as any).error).toBeTruthy();
448 });
449
450 // Subscribe + Unsubscribe — subscribe the notifications queue at
451 // runtime, assert a concrete ARN came back, then unsubscribe it in
452 // the same request so no stray subscription survives.
453 yield* Effect.gen(function* () {
454 const response = yield* postJson("/subscribe-cycle", {});
455 expect((response as any).ok).toBe(true);
456 expect((response as any).subscriptionArn).toContain("arn:aws:sns:");
457 });
458
459 // GetSMSAttributes
460 yield* Effect.gen(function* () {
461 const response = yield* getJson("/sms/attributes");
462 expect((response as any).attributes).toBeDefined();
463 });
464
465 // SetSMSAttributes — set the default type to Transactional (a safe
466 // idempotent default for the testing account) and read it back.
467 yield* Effect.gen(function* () {
468 yield* postJson("/sms/attributes", { type: "Transactional" });
469 const response = yield* getJson("/sms/attributes");
470 expect((response as any).attributes.DefaultSMSType).toBe(
471 "Transactional",
472 );
473 });
474
475 // ListPhoneNumbersOptedOut
476 yield* Effect.gen(function* () {
477 const response = yield* getJson("/sms/opted-out");
478 expect(Array.isArray((response as any).phoneNumbers)).toBe(true);
479 });
480
481 // CheckIfPhoneNumberIsOptedOut — reserved fictional number; never
482 // opted out.
483 yield* Effect.gen(function* () {
484 const response = yield* postJson("/sms/check-opt-out", {
485 phoneNumber: "+15555550100",
486 });
487 expect((response as any).isOptedOut).toBe(false);
488 });
489
490 // OptInPhoneNumber — the fictional number was never opted out, so
491 // SNS either succeeds (no-op) or rejects with a typed error; both
492 // prove the binding + IAM wiring.
493 yield* Effect.gen(function* () {
494 const response = yield* postJson("/sms/opt-in", {
495 phoneNumber: "+15555550100",
496 });
497 expect(response).toBeDefined();
498 });
499
500 // ListOriginationNumbers
501 yield* Effect.gen(function* () {
502 const response = yield* getJson("/sms/origination-numbers");
503 expect(Array.isArray((response as any).PhoneNumbers)).toBe(true);
504 });
505
506 // GetSMSSandboxAccountStatus
507 yield* Effect.gen(function* () {
508 const response = yield* getJson("/sms/sandbox-status");
509 expect(typeof (response as any).IsInSandbox).toBe("boolean");
510 });
511
512 // ListSMSSandboxPhoneNumbers
513 yield* Effect.gen(function* () {
514 const response = yield* getJson("/sms/sandbox-numbers");
515 expect(Array.isArray((response as any).PhoneNumbers)).toBe(true);
516 });
517
518 // CreateSMSSandboxPhoneNumber — malformed number surfaces the typed
519 // error (never sends a real OTP text).
520 yield* Effect.gen(function* () {
521 const response = yield* postJson("/sms/sandbox-create", {
522 phoneNumber: "not-a-phone-number",
523 });
524 expect((response as any).ok).toBe(false);
525 expect((response as any).error).toBeTruthy();
526 });
527
528 // VerifySMSSandboxPhoneNumber — unregistered number surfaces the
529 // typed error.
530 yield* Effect.gen(function* () {
531 const response = yield* postJson("/sms/sandbox-verify", {
532 phoneNumber: "+15555550100",
533 otp: "000000",
534 });
535 expect((response as any).ok).toBe(false);
536 expect((response as any).error).toBeTruthy();
537 });
538
539 // DeleteSMSSandboxPhoneNumber — unregistered number surfaces the
540 // typed error (idempotent wiring proof).
541 yield* Effect.gen(function* () {
542 const response = yield* postJson("/sms/sandbox-delete", {
543 phoneNumber: "+15555550100",
544 });
545 expect((response as any).ok).toBe(false);
546 expect((response as any).error).toBeTruthy();
547 });
548
549 // PublishSms — in the SMS sandbox the fictional number is not
550 // verified so SNS rejects with a typed error; out of the sandbox it
551 // accepts and returns a MessageId. Either proves the binding.
552 yield* Effect.gen(function* () {
553 const response = yield* postJson("/sms/publish", {
554 phoneNumber: "+15555550100",
555 message: "alchemy sns binding test",
556 });
557 const ok =
558 typeof (response as any).MessageId === "string" ||
559 (response as any).ok === false;
560 expect(ok).toBe(true);
561 });
562
563 // ListPlatformApplications
564 yield* Effect.gen(function* () {
565 const response = yield* getJson("/platform/applications");
566 expect(Array.isArray((response as any).PlatformApplications)).toBe(
567 true,
568 );
569 });
570
571 // TopicSink — 12 messages > the PublishBatch limit of 10, so the
572 // batched sink must split the chunk into 2 sequential API calls
573 // (10 + 2) and every message must still arrive.
574 yield* Effect.gen(function* () {
575 const prefix = `sink-${crypto.randomUUID()}`;
576 const markers = Array.from(
577 { length: 12 },
578 (_, i) => `${prefix}-${i}`,
579 );
580
581 const response = yield* postJson("/sink", { messages: markers });
582 expect((response as any).ok).toBe(true);
583
584 const received: string[] = [];
585 while (received.length < markers.length) {
586 const body = yield* waitForQueueMessage((candidate) =>
587 candidate.message.startsWith(prefix),
588 );
589 received.push(body.message);
590 }
591 expect(received.sort()).toEqual([...markers].sort());
592 });
593
594 if (!process.env.NO_DESTROY) {
595 yield* stack.destroy();
596 yield* assertTopicGone(topicArn);
597 }
598 }),
599 { timeout: 240_000 },
600 );
601 },
602);
603
604class TopicStillExists extends Data.TaggedError("TopicStillExists") {}
605
606// Out-of-band proof the trailing destroy left nothing behind: the fixture
607// topic must be observably gone (typed NotFoundException) after destroy.
608const assertTopicGone = Effect.fn(function* (topicArn: string) {
609 yield* SNS.getTopicAttributes({ TopicArn: topicArn }).pipe(
610 Effect.flatMap(() => Effect.fail(new TopicStillExists())),
611 Effect.retry({
612 while: (error) => error._tag === "TopicStillExists",
613 schedule: Schedule.exponential(100),
614 times: 8,
615 }),
616 Effect.catchTag("NotFoundException", () => Effect.void),
617 Effect.catchTag("InvalidParameterException", () => Effect.void),
618 );
619});
620
621class QueueMessageNotReady extends Data.TaggedError("QueueMessageNotReady") {}
622class SubscriptionNotListed extends Data.TaggedError("SubscriptionNotListed") {}
623class TopicAttributeNotPropagated extends Data.TaggedError(
624 "TopicAttributeNotPropagated",
625) {}