1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import { describe, expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import IoTBindingsFunctionLive, {
11 IoTBindingsFunction,
12 RETAINED_TOPIC,
13} from "./iot-bindings-handler.ts";
14
15const testOptions = { providers: AWS.providers() };
16const { test, beforeAll, afterAll } = Test.make(testOptions);
17const sharedStack = Core.scratchStack(testOptions, "IoTBindings");
18
19// Bound environment propagation can lag the code update; keep the readiness
20// wait bounded to about 50 seconds.
21const readinessPolicy = Schedule.max([
22 Schedule.fixed("5 seconds"),
23 Schedule.recurs(10),
24]);
25
26let baseUrl: string;
27let thingName: string;
28
29class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
30 readonly status: number;
31 readonly body: string;
32}> {}
33
34class NotYetConsistent extends Data.TaggedError("NotYetConsistent")<{
35 readonly what: string;
36}> {}
37
38// Retry transient 5xx (cold re-init, IAM propagation surfaced by the
39// handler's Effect.orDie as a 500). Genuine 4xx returns immediately.
40const send = (request: HttpClientRequest.HttpClientRequest) =>
41 HttpClient.execute(request).pipe(
42 Effect.flatMap((response) =>
43 response.status >= 500
44 ? response.text.pipe(
45 Effect.flatMap((body) =>
46 Effect.fail(
47 new TransientUpstream({ status: response.status, body }),
48 ),
49 ),
50 )
51 : Effect.succeed(response),
52 ),
53 Effect.retry({
54 while: (e) => e._tag === "TransientUpstream",
55 schedule: Schedule.max([
56 Schedule.exponential("500 millis"),
57 Schedule.recurs(6),
58 ]),
59 }),
60 );
61
62const getJson = (path: string) =>
63 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
64 Effect.flatMap((r) => r.json),
65 );
66
67const postJson = (path: string, body: unknown) =>
68 send(
69 HttpClientRequest.bodyJsonUnsafe(
70 HttpClientRequest.post(`${baseUrl}${path}`),
71 body,
72 ),
73 ).pipe(Effect.flatMap((r) => r.json));
74
75const del = (path: string) =>
76 send(HttpClientRequest.delete(`${baseUrl}${path}`)).pipe(
77 Effect.flatMap((r) => r.json),
78 );
79
80describe(
81 "IoT Bindings",
82 { tags: ["provider:aws", "provider:aws:iot", "provider:aws:lambda", "live"] },
83 () => {
84 beforeAll(
85 Effect.gen(function* () {
86 yield* sharedStack.destroy();
87
88 const { functionUrl } = yield* sharedStack.deploy(
89 Effect.gen(function* () {
90 return yield* IoTBindingsFunction;
91 }).pipe(Effect.provide(IoTBindingsFunctionLive)),
92 );
93
94 expect(functionUrl).toBeTruthy();
95 baseUrl = functionUrl!.replace(/\/+$/, "");
96
97 // Readiness requires `thingName` to be present, not just a 200: the
98 // function is pre-created as a stub (no env) and the env-carrying
99 // UpdateFunctionConfiguration is async, so the very first execution
100 // environment can serve with stale env. Retry until a fresh
101 // environment picks up the bound env.
102 const ready = yield* HttpClient.get(`${baseUrl}/ready`).pipe(
103 Effect.flatMap((response) =>
104 response.status === 200
105 ? (response.json as Effect.Effect<{ thingName?: string }>)
106 : Effect.fail(
107 new Error(`Function not ready: ${response.status}`),
108 ),
109 ),
110 Effect.flatMap((body) =>
111 body.thingName
112 ? Effect.succeed(body as { thingName: string })
113 : Effect.fail(new Error("Function env not yet propagated")),
114 ),
115 Effect.retry({ schedule: readinessPolicy }),
116 );
117 thingName = ready.thingName;
118 expect(thingName).toBeTruthy();
119 }),
120 { timeout: 240_000 },
121 );
122
123 afterAll.skipIf(!!process.env.NO_DESTROY)(sharedStack.destroy(), {
124 timeout: 120_000,
125 });
126
127 describe("UpdateThingShadow / GetThingShadow", () => {
128 test.provider(
129 "writes and reads the classic shadow",
130 (_stack) =>
131 Effect.gen(function* () {
132 const update = (yield* postJson("/shadow", {
133 state: { desired: { led: "on" } },
134 })) as { ok: boolean };
135 expect(update.ok).toBe(true);
136
137 const shadow = (yield* getJson("/shadow")) as {
138 found: boolean;
139 payload?: string;
140 };
141 expect(shadow.found).toBe(true);
142 expect(JSON.parse(shadow.payload!).state.desired.led).toBe("on");
143 }),
144 { timeout: 120_000 },
145 );
146
147 test.provider(
148 "writes and reads a named shadow",
149 (_stack) =>
150 Effect.gen(function* () {
151 yield* postJson("/shadow?shadowName=telemetry", {
152 shadowName: "telemetry",
153 state: { reported: { t: 22.5 } },
154 });
155
156 const shadow = (yield* getJson("/shadow?shadowName=telemetry")) as {
157 found: boolean;
158 payload?: string;
159 };
160 expect(shadow.found).toBe(true);
161 expect(JSON.parse(shadow.payload!).state.reported.t).toBe(22.5);
162 }),
163 { timeout: 120_000 },
164 );
165 });
166
167 describe("ListNamedShadowsForThing", () => {
168 test.provider(
169 "lists the named shadow",
170 (_stack) =>
171 Effect.gen(function* () {
172 yield* postJson("/shadow", {
173 shadowName: "inventory",
174 state: { reported: { count: 1 } },
175 });
176
177 // Named-shadow listing is eventually consistent.
178 const result = yield* getJson("/shadow/list").pipe(
179 Effect.flatMap((r) => {
180 const { results } = r as { results: string[] };
181 return results.includes("inventory")
182 ? Effect.succeed(results)
183 : Effect.fail(new NotYetConsistent({ what: "shadow list" }));
184 }),
185 Effect.retry({
186 while: (e): boolean => e._tag === "NotYetConsistent",
187 schedule: Schedule.max([
188 Schedule.fixed("2 seconds"),
189 Schedule.recurs(10),
190 ]),
191 }),
192 );
193 expect(result).toContain("inventory");
194 }),
195 { timeout: 120_000 },
196 );
197 });
198
199 describe("DeleteThingShadow", () => {
200 test.provider(
201 "deletes a named shadow",
202 (_stack) =>
203 Effect.gen(function* () {
204 yield* postJson("/shadow", {
205 shadowName: "doomed",
206 state: { reported: { alive: true } },
207 });
208
209 const deleted = (yield* del("/shadow?shadowName=doomed")) as {
210 ok: boolean;
211 };
212 expect(deleted.ok).toBe(true);
213
214 const shadow = (yield* getJson("/shadow?shadowName=doomed")) as {
215 found: boolean;
216 };
217 expect(shadow.found).toBe(false);
218 }),
219 { timeout: 120_000 },
220 );
221 });
222
223 describe("DescribeThing", () => {
224 test.provider(
225 "reads the thing's registry entry",
226 (_stack) =>
227 Effect.gen(function* () {
228 const result = (yield* getJson("/thing")) as {
229 thingName: string;
230 attributes: Record<string, string>;
231 };
232 expect(result.thingName).toBe(thingName);
233 expect(result.attributes.purpose).toBe("bindings-test");
234 }),
235 { timeout: 120_000 },
236 );
237 });
238
239 describe("ListThings", () => {
240 test.provider(
241 "lists the deployed thing",
242 (_stack) =>
243 Effect.gen(function* () {
244 // Registry listing is eventually consistent.
245 const names = yield* getJson("/things").pipe(
246 Effect.flatMap((r) => {
247 const { thingNames } = r as { thingNames: string[] };
248 return thingNames.includes(thingName)
249 ? Effect.succeed(thingNames)
250 : Effect.fail(new NotYetConsistent({ what: "thing list" }));
251 }),
252 Effect.retry({
253 while: (e): boolean => e._tag === "NotYetConsistent",
254 schedule: Schedule.max([
255 Schedule.fixed("2 seconds"),
256 Schedule.recurs(10),
257 ]),
258 }),
259 );
260 expect(names).toContain(thingName);
261 }),
262 { timeout: 120_000 },
263 );
264 });
265
266 describe("DescribeEndpoint", () => {
267 test.provider(
268 "returns the account's ATS data endpoint",
269 (_stack) =>
270 Effect.gen(function* () {
271 const result = (yield* getJson("/endpoint")) as {
272 endpointAddress?: string;
273 };
274 expect(result.endpointAddress).toBeTruthy();
275 expect(result.endpointAddress).toContain("iot");
276 expect(result.endpointAddress).toContain("amazonaws.com");
277 }),
278 { timeout: 120_000 },
279 );
280 });
281
282 describe("Publish / GetRetainedMessage / ListRetainedMessages", () => {
283 test.provider(
284 "publishes a retained message, reads it back, lists it, and clears it",
285 (_stack) =>
286 Effect.gen(function* () {
287 const marker = "retained-marker-1";
288 yield* postJson("/retained", {
289 topic: RETAINED_TOPIC,
290 payload: JSON.stringify({ marker }),
291 });
292
293 // Retained-message read is eventually consistent after publish.
294 const read = yield* getJson(
295 `/retained?topic=${encodeURIComponent(RETAINED_TOPIC)}`,
296 ).pipe(
297 Effect.flatMap((r) => {
298 const result = r as { found: boolean; payload?: string };
299 return result.found
300 ? Effect.succeed(result)
301 : Effect.fail(
302 new NotYetConsistent({ what: "retained message" }),
303 );
304 }),
305 Effect.retry({
306 while: (e): boolean => e._tag === "NotYetConsistent",
307 schedule: Schedule.max([
308 Schedule.fixed("2 seconds"),
309 Schedule.recurs(10),
310 ]),
311 }),
312 );
313 expect(JSON.parse(read.payload!).marker).toBe(marker);
314
315 const listed = (yield* getJson("/retained/list")) as {
316 topics: string[];
317 };
318 expect(listed.topics).toContain(RETAINED_TOPIC);
319
320 // Clear the retained message (empty retained payload) so the
321 // account carries no leftovers after the test.
322 yield* postJson("/retained", { topic: RETAINED_TOPIC });
323 const cleared = yield* getJson(
324 `/retained?topic=${encodeURIComponent(RETAINED_TOPIC)}`,
325 ).pipe(
326 Effect.flatMap((r) => {
327 const result = r as { found: boolean };
328 return result.found
329 ? Effect.fail(
330 new NotYetConsistent({ what: "retained clear" }),
331 )
332 : Effect.succeed(result);
333 }),
334 Effect.retry({
335 while: (e): boolean => e._tag === "NotYetConsistent",
336 schedule: Schedule.max([
337 Schedule.fixed("2 seconds"),
338 Schedule.recurs(10),
339 ]),
340 }),
341 );
342 expect(cleared.found).toBe(false);
343 }),
344 { timeout: 120_000 },
345 );
346 });
347
348 describe("GetConnection", () => {
349 test.provider(
350 "returns a typed ResourceNotFoundException for a never-connected client",
351 (_stack) =>
352 Effect.gen(function* () {
353 const result = (yield* getJson(
354 "/connection?clientId=alchemy-bindings-nonexistent",
355 )) as { ok: boolean; tag?: string; connected?: boolean };
356 if (result.ok) {
357 // Some accounts report never-connected clients as disconnected
358 // instead of missing.
359 expect(result.connected).toBe(false);
360 } else {
361 expect(result.tag).toBe("ResourceNotFoundException");
362 }
363 }),
364 { timeout: 120_000 },
365 );
366 });
367
368 describe("ListSubscriptions", () => {
369 test.provider(
370 "returns a typed ResourceNotFoundException for a never-connected client",
371 (_stack) =>
372 Effect.gen(function* () {
373 const result = (yield* getJson(
374 "/subscriptions?clientId=alchemy-bindings-nonexistent",
375 )) as { ok: boolean; tag?: string; subscriptions?: unknown[] };
376 if (result.ok) {
377 expect(result.subscriptions).toEqual([]);
378 } else {
379 expect(result.tag).toBe("ResourceNotFoundException");
380 }
381 }),
382 { timeout: 120_000 },
383 );
384 });
385
386 describe("DeleteConnection", () => {
387 test.provider(
388 "returns a typed ResourceNotFoundException for a never-connected client",
389 (_stack) =>
390 Effect.gen(function* () {
391 const result = (yield* del(
392 "/connection?clientId=alchemy-bindings-nonexistent",
393 )) as { ok: boolean; tag?: string };
394 expect(result.ok).toBe(false);
395 expect(result.tag).toBe("ResourceNotFoundException");
396 }),
397 { timeout: 120_000 },
398 );
399 });
400
401 describe("SendDirectMessage", () => {
402 test.provider(
403 "returns a typed ResourceNotFoundException for a never-connected client",
404 (_stack) =>
405 Effect.gen(function* () {
406 const result = (yield* postJson("/direct", {
407 clientId: "alchemy-bindings-nonexistent",
408 topic: "commands/noop",
409 payload: JSON.stringify({ noop: true }),
410 })) as { ok: boolean; tag?: string };
411 expect(result.ok).toBe(false);
412 expect(result.tag).toBe("ResourceNotFoundException");
413 }),
414 { timeout: 120_000 },
415 );
416 });
417 },
418);