1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as Lambda from "@distilled.cloud/aws/lambda";
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 IoTEventSourceFunctionLive, {
12 IoTEventSourceFunction,
13} from "./iot-event-source-handler.ts";
14
15const { test } = Test.make({ providers: AWS.providers() });
16
17describe.sequential(
18 "AWS.IoT.TopicRuleEventSource",
19 {
20 tags: [
21 "provider:aws",
22 "provider:aws:iot",
23 "provider:aws:lambda",
24 "provider:aws:sqs",
25 "live",
26 ],
27 },
28 () => {
29 test.provider(
30 "publishes via the Publish binding, routes through the topic rule, and observes delivery",
31 (stack) =>
32 Effect.gen(function* () {
33 yield* stack.destroy();
34
35 const fn = yield* stack.deploy(
36 IoTEventSourceFunction.pipe(
37 Effect.provide(IoTEventSourceFunctionLive),
38 ),
39 );
40 const functionUrl = fn.functionUrl!.replace(/\/+$/, "");
41
42 // Ride out cold-start / URL propagation until /ready reports the queue.
43 const { resultQueueUrl } = yield* HttpClient.get(
44 `${functionUrl}/ready`,
45 ).pipe(
46 // Bound each fetch attempt so a transient Function URL DNS/socket
47 // stall reaches the retry schedule instead of consuming the whole
48 // test timeout.
49 Effect.timeout("4 seconds"),
50 Effect.mapError(() => new FunctionNotReady("ready request")),
51 Effect.flatMap((response) =>
52 response.status === 200
53 ? (response.json as Effect.Effect<{ resultQueueUrl?: string }>)
54 : Effect.fail(
55 new FunctionNotReady(`status ${response.status}`),
56 ),
57 ),
58 Effect.flatMap((body) =>
59 body.resultQueueUrl
60 ? Effect.succeed(body as { resultQueueUrl: string })
61 : Effect.fail(new FunctionNotReady("no result queue")),
62 ),
63 Effect.retry({
64 schedule: Schedule.max([
65 Schedule.fixed("5 seconds"),
66 Schedule.recurs(10),
67 ]),
68 }),
69 );
70
71 // Publish a uniquely-marked message from inside the Lambda via the IoT
72 // Publish binding. The topic rule matches it and re-invokes the Lambda,
73 // whose event-source handler forwards it into the result queue.
74 const marker = `iot-${crypto.randomUUID()}`;
75 yield* HttpClient.execute(
76 HttpClientRequest.post(`${functionUrl}/publish`).pipe(
77 HttpClientRequest.bodyJsonUnsafe({ marker }),
78 ),
79 ).pipe(
80 Effect.timeout("4 seconds"),
81 Effect.mapError(() => new FunctionNotReady("publish request")),
82 Effect.filterOrFail(
83 (response) => response.status === 200,
84 (response) => new FunctionNotReady(`publish ${response.status}`),
85 ),
86 Effect.retry({
87 while: (e) => e._tag === "FunctionNotReady",
88 schedule: Schedule.max([
89 Schedule.fixed("5 seconds"),
90 Schedule.recurs(10),
91 ]),
92 }),
93 );
94
95 // Poll the result queue until the forwarded message (carrying the
96 // marker) shows up. IoT rule and permission propagation are eventually
97 // consistent, so republish while waiting, bounded to about 50 seconds.
98 const received = yield* Effect.gen(function* () {
99 const result = yield* SQS.receiveMessage({
100 QueueUrl: resultQueueUrl,
101 MaxNumberOfMessages: 10,
102 WaitTimeSeconds: 2,
103 });
104 const match = (result.Messages ?? []).find((message) =>
105 message.Body?.includes(marker),
106 );
107 if (!match?.ReceiptHandle) {
108 // Republish in case the earlier publish predated rule/permission
109 // readiness.
110 yield* HttpClient.execute(
111 HttpClientRequest.post(`${functionUrl}/publish`).pipe(
112 HttpClientRequest.bodyJsonUnsafe({ marker }),
113 ),
114 ).pipe(Effect.timeout("4 seconds"), Effect.ignore);
115 return yield* Effect.fail(new MessageNotDelivered());
116 }
117 yield* SQS.deleteMessage({
118 QueueUrl: resultQueueUrl,
119 ReceiptHandle: match.ReceiptHandle,
120 });
121 return match.Body!;
122 }).pipe(
123 Effect.retry({
124 while: (error) => error._tag === "MessageNotDelivered",
125 schedule: Schedule.max([
126 Schedule.fixed("3 seconds"),
127 Schedule.recurs(10),
128 ]),
129 }),
130 );
131
132 expect(received).toContain(marker);
133
134 yield* stack.destroy();
135
136 // Assert the stack's observable resources are gone: the Lambda
137 // function (which hosted the event source) and the result queue.
138 yield* Lambda.getFunction({ FunctionName: fn.functionName }).pipe(
139 Effect.flatMap(() =>
140 Effect.fail(new ResourceStillExists("lambda")),
141 ),
142 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
143 Effect.retry({
144 while: (e) => e._tag === "ResourceStillExists",
145 schedule: Schedule.max([
146 Schedule.fixed("2 seconds"),
147 Schedule.recurs(10),
148 ]),
149 }),
150 );
151 // SQS DeleteQueue propagation is documented at up to ~60s.
152 yield* SQS.getQueueAttributes({
153 QueueUrl: resultQueueUrl,
154 AttributeNames: ["All"],
155 }).pipe(
156 Effect.flatMap(() => Effect.fail(new ResourceStillExists("queue"))),
157 Effect.catchTag("QueueDoesNotExist", () => Effect.void),
158 Effect.retry({
159 while: (e) => e._tag === "ResourceStillExists",
160 schedule: Schedule.max([
161 Schedule.spaced("5 seconds"),
162 Schedule.recurs(10),
163 ]),
164 }),
165 );
166 }),
167 { timeout: 180_000 },
168 );
169 },
170);
171
172class MessageNotDelivered extends Data.TaggedError("MessageNotDelivered") {}
173
174class ResourceStillExists extends Data.TaggedError("ResourceStillExists")<{
175 what: string;
176}> {
177 constructor(what: string) {
178 super({ what });
179 }
180}
181
182class FunctionNotReady extends Data.TaggedError("FunctionNotReady")<{
183 message: string;
184}> {
185 constructor(message: string) {
186 super({ message });
187 }
188}