1import * as AWS from "@/AWS";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import { HttpServerRequest } from "effect/http/HttpServerRequest";
6import * as HttpServerResponse from "effect/http/HttpServerResponse";
7import path from "pathe";
8
9const main = path.resolve(import.meta.dirname, "iot-bindings-handler.ts");
10
11// The MQTT topic namespace this fixture publishes retained messages into.
12export const TOPIC_FILTER = "alchemy/iot/bindings/*";
13export const RETAINED_TOPIC = "alchemy/iot/bindings/retained";
14
15// Exercises every IoT runtime binding from inside a deployed Lambda:
16// device shadows + DescribeThing on a Thing resource, retained messages via
17// Publish/GetRetainedMessage/ListRetainedMessages, DescribeEndpoint and
18// ListThings at account level, and the MQTT connection-management data
19// plane (GetConnection/ListSubscriptions/DeleteConnection/SendDirectMessage)
20// against a client id filter.
21export class IoTBindingsFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
22 "IoTBindingsFunction",
23) {}
24
25export default IoTBindingsFunction.make(
26 {
27 main,
28 functionUrl: true,
29 },
30 Effect.gen(function* () {
31 const thing = yield* AWS.IoT.Thing("BindingsThing", {
32 attributes: { purpose: "bindings-test" },
33 });
34
35 // thing-scoped
36 const getShadow = yield* AWS.IoT.GetThingShadow(thing);
37 const updateShadow = yield* AWS.IoT.UpdateThingShadow(thing);
38 const deleteShadow = yield* AWS.IoT.DeleteThingShadow(thing);
39 const listShadows = yield* AWS.IoT.ListNamedShadowsForThing(thing);
40 const describeThing = yield* AWS.IoT.DescribeThing(thing);
41 // topic-scoped
42 const publish = yield* AWS.IoT.Publish(TOPIC_FILTER);
43 const getRetained = yield* AWS.IoT.GetRetainedMessage(TOPIC_FILTER);
44 // account-level
45 const listRetained = yield* AWS.IoT.ListRetainedMessages();
46 const describeEndpoint = yield* AWS.IoT.DescribeEndpoint();
47 const listThings = yield* AWS.IoT.ListThings();
48 // client-scoped
49 const getConnection = yield* AWS.IoT.GetConnection("alchemy-bindings-*");
50 const listSubscriptions =
51 yield* AWS.IoT.ListSubscriptions("alchemy-bindings-*");
52 const deleteConnection =
53 yield* AWS.IoT.DeleteConnection("alchemy-bindings-*");
54 const sendDirectMessage =
55 yield* AWS.IoT.SendDirectMessage("alchemy-bindings-*");
56
57 const thingName = yield* thing.thingName;
58
59 const decodeShadowPayload = (
60 payload: Stream.Stream<Uint8Array, Error> | undefined,
61 ) =>
62 payload === undefined
63 ? Effect.succeed(undefined)
64 : Stream.mkString(Stream.decodeText(payload));
65
66 return {
67 fetch: Effect.gen(function* () {
68 const request = yield* HttpServerRequest;
69 const url = new URL(request.originalUrl);
70 const pathname = url.pathname;
71 const shadowName = url.searchParams.get("shadowName") ?? undefined;
72
73 if (request.method === "GET" && pathname === "/ready") {
74 // `updateFunctionConfiguration` (which delivers the accessor env
75 // vars) is eventually consistent: a sandbox that boots between the
76 // code update and the config update runs the REAL code with the
77 // STALE (pre-binding) env, and stays warm serving it. Report 503
78 // until the env is visible so the test's readiness retry keeps
79 // polling until Lambda retires the stale environment.
80 const resolvedThingName = yield* thingName;
81 if (!resolvedThingName) {
82 return yield* HttpServerResponse.json(
83 { ok: false, error: "binding env not yet propagated" },
84 { status: 503 },
85 );
86 }
87 return yield* HttpServerResponse.json({
88 ok: true,
89 thingName: resolvedThingName,
90 });
91 }
92
93 if (request.method === "POST" && pathname === "/shadow") {
94 const body = (yield* request.json) as {
95 shadowName?: string;
96 state: unknown;
97 };
98 const result = yield* updateShadow({
99 shadowName: body.shadowName,
100 payload: JSON.stringify({ state: body.state }),
101 });
102 const payload = yield* decodeShadowPayload(result.payload);
103 return yield* HttpServerResponse.json({ ok: true, payload });
104 }
105
106 if (request.method === "GET" && pathname === "/shadow") {
107 const result = yield* getShadow({ shadowName }).pipe(
108 Effect.map((r) => ({ found: true as const, payload: r.payload })),
109 Effect.catchTag("ResourceNotFoundException", () =>
110 Effect.succeed({ found: false as const, payload: undefined }),
111 ),
112 );
113 const payload = yield* decodeShadowPayload(result.payload);
114 return yield* HttpServerResponse.json({
115 found: result.found,
116 payload,
117 });
118 }
119
120 if (request.method === "GET" && pathname === "/shadow/list") {
121 const result = yield* listShadows();
122 return yield* HttpServerResponse.json({
123 results: result.results ?? [],
124 });
125 }
126
127 if (request.method === "DELETE" && pathname === "/shadow") {
128 yield* deleteShadow({ shadowName }).pipe(
129 Effect.catchTag("ResourceNotFoundException", () =>
130 Effect.succeed(undefined),
131 ),
132 );
133 return yield* HttpServerResponse.json({ ok: true });
134 }
135
136 if (request.method === "GET" && pathname === "/thing") {
137 const result = yield* describeThing();
138 return yield* HttpServerResponse.json({
139 thingName: result.thingName,
140 attributes: result.attributes ?? {},
141 });
142 }
143
144 if (request.method === "GET" && pathname === "/things") {
145 const result = yield* listThings();
146 return yield* HttpServerResponse.json({
147 thingNames: (result.things ?? []).map((t) => t.thingName),
148 });
149 }
150
151 if (request.method === "GET" && pathname === "/endpoint") {
152 const result = yield* describeEndpoint({
153 endpointType: "iot:Data-ATS",
154 });
155 return yield* HttpServerResponse.json({
156 endpointAddress: result.endpointAddress,
157 });
158 }
159
160 if (request.method === "POST" && pathname === "/retained") {
161 const body = (yield* request.json) as {
162 topic: string;
163 payload?: string;
164 };
165 // An empty retained payload clears the retained message.
166 yield* publish({
167 topic: body.topic,
168 retain: true,
169 qos: 1,
170 payload: body.payload,
171 });
172 return yield* HttpServerResponse.json({ ok: true });
173 }
174
175 if (request.method === "GET" && pathname === "/retained") {
176 const topic = url.searchParams.get("topic")!;
177 const result = yield* getRetained({ topic }).pipe(
178 Effect.map((r) => ({ found: true as const, payload: r.payload })),
179 Effect.catchTag("ResourceNotFoundException", () =>
180 Effect.succeed({ found: false as const, payload: undefined }),
181 ),
182 );
183 const payload = result.payload
184 ? yield* Effect.sync(() => new TextDecoder().decode(result.payload))
185 : undefined;
186 return yield* HttpServerResponse.json({
187 found: result.found,
188 payload,
189 });
190 }
191
192 if (request.method === "GET" && pathname === "/retained/list") {
193 const result = yield* listRetained();
194 return yield* HttpServerResponse.json({
195 topics: (result.retainedTopics ?? []).map((t) => t.topic),
196 });
197 }
198
199 if (request.method === "GET" && pathname === "/connection") {
200 const clientId = url.searchParams.get("clientId")!;
201 const result = yield* getConnection({ clientId }).pipe(
202 Effect.map((r) => ({
203 ok: true as const,
204 connected: r.connected ?? false,
205 })),
206 Effect.catchTag("ResourceNotFoundException", (e) =>
207 Effect.succeed({
208 ok: false as const,
209 tag: "ResourceNotFoundException",
210 message: e.message,
211 }),
212 ),
213 );
214 return yield* HttpServerResponse.json(result);
215 }
216
217 if (request.method === "GET" && pathname === "/subscriptions") {
218 const clientId = url.searchParams.get("clientId")!;
219 const result = yield* listSubscriptions({ clientId }).pipe(
220 Effect.map((r) => ({
221 ok: true as const,
222 subscriptions: r.subscriptions ?? [],
223 })),
224 Effect.catchTag("ResourceNotFoundException", (e) =>
225 Effect.succeed({
226 ok: false as const,
227 tag: "ResourceNotFoundException",
228 message: e.message,
229 }),
230 ),
231 );
232 return yield* HttpServerResponse.json(result);
233 }
234
235 if (request.method === "DELETE" && pathname === "/connection") {
236 const clientId = url.searchParams.get("clientId")!;
237 const result = yield* deleteConnection({ clientId }).pipe(
238 Effect.map(() => ({ ok: true as const })),
239 Effect.catchTag("ResourceNotFoundException", (e) =>
240 Effect.succeed({
241 ok: false as const,
242 tag: "ResourceNotFoundException",
243 message: e.message,
244 }),
245 ),
246 );
247 return yield* HttpServerResponse.json(result);
248 }
249
250 if (request.method === "POST" && pathname === "/direct") {
251 const body = (yield* request.json) as {
252 clientId: string;
253 topic: string;
254 payload?: string;
255 };
256 const result = yield* sendDirectMessage({
257 clientId: body.clientId,
258 topic: body.topic,
259 payload: body.payload,
260 }).pipe(
261 Effect.map(() => ({ ok: true as const })),
262 Effect.catchTag("ResourceNotFoundException", (e) =>
263 Effect.succeed({
264 ok: false as const,
265 tag: "ResourceNotFoundException",
266 message: e.message,
267 }),
268 ),
269 );
270 return yield* HttpServerResponse.json(result);
271 }
272
273 return yield* HttpServerResponse.json(
274 { error: "Not found", method: request.method, pathname },
275 { status: 404 },
276 );
277 }).pipe(Effect.orDie),
278 };
279 }).pipe(
280 Effect.provide(
281 Layer.mergeAll(
282 AWS.IoT.GetThingShadowHttp,
283 AWS.IoT.UpdateThingShadowHttp,
284 AWS.IoT.DeleteThingShadowHttp,
285 AWS.IoT.ListNamedShadowsForThingHttp,
286 AWS.IoT.DescribeThingHttp,
287 AWS.IoT.PublishHttp,
288 AWS.IoT.GetRetainedMessageHttp,
289 AWS.IoT.ListRetainedMessagesHttp,
290 AWS.IoT.DescribeEndpointHttp,
291 AWS.IoT.ListThingsHttp,
292 AWS.IoT.GetConnectionHttp,
293 AWS.IoT.ListSubscriptionsHttp,
294 AWS.IoT.DeleteConnectionHttp,
295 AWS.IoT.SendDirectMessageHttp,
296 ),
297 ),
298 ),
299);