1import * as DynamoDB from "alchemy/AWS/DynamoDB";
2import * as Lambda from "alchemy/AWS/Lambda";
3import * as S3 from "alchemy/AWS/S3";
4import * as SNS from "alchemy/AWS/SNS";
5import * as SQS from "alchemy/AWS/SQS";
6import * as Config from "effect/Config";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Stream from "effect/Stream";
10import { HttpServerRequest } from "effect/http/HttpServerRequest";
11import * as HttpServerResponse from "effect/http/HttpServerResponse";
12import { MARKER } from "./marker.ts";
13
14/**
15 * Kitchen-sink dev-mode Lambda: one Function owning an S3 bucket, a
16 * DynamoDB table, an SQS queue, and an SNS topic, exercising each runtime
17 * binding AND each event-source glue resource (Subscription, Permission,
18 * EventSourceMapping) over HTTP routes so `alchemy dev` — and a live
19 * `alchemy deploy` — can be driven end-to-end from a test.
20 *
21 * Routes:
22 * - GET / → marker + env (Config-provided MY_VARIABLE)
23 * - GET /s3 → PutObject/GetObject roundtrip
24 * - GET /dynamo → PutItem/GetItem roundtrip
25 * - POST /queue/send → SendMessage over the binding
26 * - GET /queue/messages → reads what the queue consumer recorded
27 * - POST /topic/send → SNS Publish over the binding
28 * - GET /topic/messages → reads what the topic consumer recorded
29 * - POST /items → PutItem with a caller-chosen id (stream feed)
30 * - GET /changes → reads what the table-changes consumer recorded
31 *
32 * The consumers all run on this same Function:
33 * - `consumeQueueMessages` — the event-source-mapping poller invokes it
34 * with SQS batches
35 * - `consumeTopicNotifications` — creates the SNS.Subscription and the
36 * Lambda invoke Permission (the glue that must be local in dev mode)
37 * - `consumeTableChanges` — enables the table stream and creates the
38 * stream EventSourceMapping
39 * Each consumer records into the table under a distinct key prefix so every
40 * produce → deliver → consume path is observable over HTTP.
41 */
42export default class ApiFunction extends Lambda.Function<ApiFunction>()(
43 "ApiFunction",
44 {
45 main: import.meta.url,
46 functionUrl: true,
47 // The default 128 MB runs this fixture at its ceiling (observed
48 // Max Memory Used: 127 MB live) — leave headroom for event batches.
49 memorySize: 512,
50 env: { MY_VARIABLE: "my-variable-abc123" },
51 },
52 Effect.gen(function* () {
53 // `/s3` writes into the bucket, so destroy must empty it first.
54 const bucket = yield* S3.Bucket("DevBucket", { forceDestroy: true });
55 const table = yield* DynamoDB.Table("MessagesTable", {
56 partitionKey: "id",
57 attributes: { id: "S" },
58 });
59 const queue = yield* SQS.Queue("DevQueue");
60 const topic = yield* SNS.Topic("DevTopic");
61
62 const getObject = yield* S3.GetObject(bucket);
63 const putObject = yield* S3.PutObject(bucket);
64 const getItem = yield* DynamoDB.GetItem(table);
65 const putItem = yield* DynamoDB.PutItem(table);
66 const sendMessage = yield* SQS.SendMessage(queue);
67 const publish = yield* SNS.Publish(topic);
68
69 // Consume produced messages and record them into the table so the
70 // test can observe the produce → deliver → consume roundtrip.
71 yield* SQS.consumeQueueMessages(queue, (records) =>
72 records.pipe(
73 Stream.mapEffect((record) => {
74 const parsed = JSON.parse(record.body) as { id: string };
75 return putItem({
76 Item: {
77 id: { S: `msg:${parsed.id}` },
78 body: { S: record.body },
79 },
80 });
81 }),
82 Stream.runDrain,
83 Effect.orDie,
84 ),
85 );
86
87 // Consume topic notifications. This is the SNS → Lambda glue: it
88 // creates the AWS.SNS.Subscription and the AWS.Lambda.Permission
89 // (lambda:InvokeFunction for sns.amazonaws.com) — the two resources
90 // that MUST resolve against the emulator in dev mode, since their
91 // props embed the local topic/function ARNs.
92 yield* SNS.consumeTopicNotifications(topic, (notifications) =>
93 notifications.pipe(
94 Stream.mapEffect((notification) => {
95 const parsed = JSON.parse(notification.Message) as { id: string };
96 return putItem({
97 Item: {
98 id: { S: `topic:${parsed.id}` },
99 body: { S: notification.Message },
100 },
101 });
102 }),
103 Stream.runDrain,
104 Effect.orDie,
105 ),
106 );
107
108 // Consume the table's change stream. This enables the DynamoDB stream
109 // on the table and creates the stream AWS.Lambda.EventSourceMapping.
110 // Only plain (un-prefixed) item ids are recorded, so the consumers'
111 // own `msg:`/`topic:`/`change:` writes never feed back into the stream.
112 // TRIM_HORIZON, not LATEST: on real AWS the poller starts reading from
113 // the stream TIP whenever it actually begins polling (minutes after
114 // mapping creation), so LATEST permanently skips records written in
115 // that window — a write shortly after deploy would never arrive.
116 yield* DynamoDB.consumeTableChanges(
117 table,
118 {
119 streamViewType: "NEW_AND_OLD_IMAGES",
120 startingPosition: "TRIM_HORIZON",
121 },
122 (changes) =>
123 changes.pipe(
124 Stream.mapEffect((record) => {
125 const id = record.dynamodb.Keys?.id?.S;
126 if (typeof id !== "string" || id.includes(":")) {
127 return Effect.void;
128 }
129 return putItem({
130 Item: {
131 id: { S: `change:${id}` },
132 body: { S: record.eventName ?? "UNKNOWN" },
133 },
134 });
135 }),
136 Stream.runDrain,
137 Effect.orDie,
138 ),
139 );
140
141 return {
142 fetch: Effect.gen(function* () {
143 const request = yield* HttpServerRequest;
144 const url = new URL(request.originalUrl);
145
146 if (url.pathname === "/") {
147 const variable = yield* Config.String("MY_VARIABLE");
148 return yield* HttpServerResponse.json({ marker: MARKER, variable });
149 }
150
151 if (url.pathname === "/s3") {
152 yield* putObject({ Key: "hello.txt", Body: "hello from s3" });
153 const object = yield* getObject({ Key: "hello.txt" });
154 const text = yield* (
155 object.Body?.pipe(Stream.decodeText, Stream.mkString) ??
156 Effect.succeed("")
157 );
158 return yield* HttpServerResponse.json({ text });
159 }
160
161 if (url.pathname === "/dynamo") {
162 yield* putItem({
163 Item: {
164 id: { S: "hello" },
165 content: { S: "hello from dynamo" },
166 },
167 });
168 const item = yield* getItem({ Key: { id: { S: "hello" } } });
169 return yield* HttpServerResponse.json({
170 text: item.Item?.content?.S ?? null,
171 });
172 }
173
174 if (url.pathname === "/queue/send" && request.method === "POST") {
175 const body = yield* request.text;
176 yield* sendMessage({ MessageBody: body });
177 return yield* HttpServerResponse.json({ sent: true });
178 }
179
180 if (url.pathname === "/queue/messages") {
181 const id = url.searchParams.get("id") ?? "";
182 const item = yield* getItem({ Key: { id: { S: `msg:${id}` } } });
183 return yield* HttpServerResponse.json({
184 body: item.Item?.body?.S ?? null,
185 });
186 }
187
188 if (url.pathname === "/topic/send" && request.method === "POST") {
189 const body = yield* request.text;
190 yield* publish({ Message: body });
191 return yield* HttpServerResponse.json({ published: true });
192 }
193
194 if (url.pathname === "/topic/messages") {
195 const id = url.searchParams.get("id") ?? "";
196 const item = yield* getItem({ Key: { id: { S: `topic:${id}` } } });
197 return yield* HttpServerResponse.json({
198 body: item.Item?.body?.S ?? null,
199 });
200 }
201
202 if (url.pathname === "/items" && request.method === "POST") {
203 const body = yield* request.text;
204 const parsed = JSON.parse(body) as { id: string };
205 yield* putItem({
206 Item: {
207 id: { S: parsed.id },
208 body: { S: body },
209 },
210 });
211 return yield* HttpServerResponse.json({ put: true });
212 }
213
214 if (url.pathname === "/changes") {
215 const id = url.searchParams.get("id") ?? "";
216 const item = yield* getItem({ Key: { id: { S: `change:${id}` } } });
217 return yield* HttpServerResponse.json({
218 body: item.Item?.body?.S ?? null,
219 });
220 }
221
222 return HttpServerResponse.text("Not found", { status: 404 });
223 }).pipe(Effect.orDie),
224 };
225 }).pipe(
226 Effect.provide(
227 Layer.mergeAll(
228 Lambda.QueueEventSource,
229 Lambda.TopicEventSource,
230 Lambda.TableEventSource,
231 S3.GetObjectHttp,
232 S3.PutObjectHttp,
233 DynamoDB.GetItemHttp,
234 DynamoDB.PutItemHttp,
235 SQS.SendMessageHttp,
236 SNS.PublishHttp,
237 ),
238 ),
239 ),
240) {}