1import * as AWS from "@/AWS";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Stream from "effect/Stream";
6import { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import path from "pathe";
9
10const main = path.resolve(import.meta.dirname, "iot-event-source-handler.ts");
11
12// The MQTT topic the rule subscribes to and the fixture publishes to.
13export const TOPIC = "alchemy/iot/eventsource";
14
15// End-to-end IoT TopicRuleEventSource fixture. The Lambda:
16// 1. Publishes MQTT messages to `TOPIC` via the IoT Publish binding
17// (POST /publish).
18// 2. Consumes messages routed back to it by the IoT topic rule
19// (`consumeTopicMessages`) and forwards each into a result SQS queue via
20// QueueSink, so the test can observe delivery out-of-band.
21export class IoTEventSourceFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
22 "IoTEventSourceFunction",
23) {}
24
25export class ResultQueue extends Context.Service<
26 ResultQueue,
27 { result: AWS.SQS.Queue }
28>()("IoTResultQueue") {}
29
30export const ResultQueueLive = Layer.effect(
31 ResultQueue,
32 Effect.gen(function* () {
33 const result = yield* AWS.SQS.Queue("IoTEventSourceResultQueue");
34 return { result };
35 }),
36);
37
38export default IoTEventSourceFunction.make(
39 {
40 main,
41 functionUrl: true,
42 },
43 Effect.gen(function* () {
44 const { result } = yield* ResultQueue;
45 const sink = yield* AWS.SQS.QueueSink(result);
46 const publish = yield* AWS.IoT.Publish(TOPIC);
47
48 yield* AWS.IoT.consumeTopicMessages(TOPIC, (messages) =>
49 messages.pipe(
50 Stream.map((message) => ({ MessageBody: JSON.stringify(message) })),
51 Stream.run(sink),
52 Effect.orDie,
53 ),
54 );
55
56 const resultQueueUrl = yield* result.queueUrl;
57
58 return {
59 fetch: Effect.gen(function* () {
60 const request = yield* HttpServerRequest;
61 const pathname = new URL(request.originalUrl).pathname;
62
63 if (request.method === "GET" && pathname === "/ready") {
64 return yield* HttpServerResponse.json({
65 ok: true,
66 resultQueueUrl: yield* resultQueueUrl,
67 });
68 }
69
70 if (request.method === "POST" && pathname === "/publish") {
71 const body = (yield* request.json) as { marker: string };
72 yield* publish({
73 topic: TOPIC,
74 payload: JSON.stringify({ marker: body.marker }),
75 });
76 return yield* HttpServerResponse.json({ ok: true });
77 }
78
79 return yield* HttpServerResponse.json(
80 { error: "Not found", method: request.method, pathname },
81 { status: 404 },
82 );
83 }).pipe(Effect.orDie),
84 };
85 }).pipe(
86 Effect.provide(
87 Layer.provideMerge(
88 Layer.mergeAll(
89 AWS.Lambda.TopicRuleEventSource,
90 AWS.SQS.QueueSinkHttp,
91 AWS.IoT.PublishHttp,
92 ),
93 Layer.mergeAll(AWS.SQS.SendMessageBatchHttp, ResultQueueLive),
94 ),
95 ),
96 ),
97);