1import * as Lambda from "@/AWS/Lambda";
2import * as SQS from "@/AWS/SQS";
3import * as Cause from "effect/Cause";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "handler.ts");
12
13// Shared Lambda fixture hosting every SQS message-plane binding, one HTTP
14// route per operation (see test/AWS/DynamoDB/handler.ts for the pattern).
15export class SQSTestFunction extends Lambda.Function<Lambda.Function>()(
16 "SQSTestFunction",
17) {}
18
19export default SQSTestFunction.make(
20 {
21 main,
22 functionUrl: true,
23 // The /receive route long-polls SQS (WaitTimeSeconds up to 2s); AWS's 3s
24 // default function timeout intermittently kills the invocation AFTER SQS
25 // has consumed the message server-side, trapping it invisible for the
26 // queue's 30s visibility window while the caller only sees a 500.
27 timeout: Duration.seconds(30),
28 },
29 Effect.gen(function* () {
30 // BindingsDLQ doubles as the redrive source for the message-move-task
31 // bindings; BindingsQueue's redrivePolicy is what makes it a valid DLQ.
32 const dlq = yield* SQS.Queue("BindingsDLQ");
33 const queue = yield* SQS.Queue("BindingsQueue", {
34 redrivePolicy: {
35 deadLetterTargetArn: dlq.queueArn,
36 // High enough that test polling (bounded re-receives) never trips a
37 // real redrive; the DLQ only receives messages sent to it directly.
38 maxReceiveCount: 1000,
39 },
40 });
41 // Dedicated source for MessageMoveTasks. Messages must have actually
42 // redriven from a source queue to be eligible for DLQ redrive; directly
43 // sending to the DLQ is not deterministic. Keep this separate from the
44 // shared binding queue because maxReceiveCount=1 would interfere with the
45 // receive/visibility tests below.
46 const moveSource = yield* SQS.Queue("BindingsMoveSource", {
47 redrivePolicy: {
48 deadLetterTargetArn: dlq.queueArn,
49 maxReceiveCount: 1,
50 },
51 });
52
53 const sendMessage = yield* SQS.SendMessage(queue);
54 const sendMessageBatch = yield* SQS.SendMessageBatch(queue);
55 const receiveMessage = yield* SQS.ReceiveMessage(queue);
56 const deleteMessage = yield* SQS.DeleteMessage(queue);
57 const deleteMessageBatch = yield* SQS.DeleteMessageBatch(queue);
58 const changeMessageVisibility = yield* SQS.ChangeMessageVisibility(queue);
59 const changeMessageVisibilityBatch =
60 yield* SQS.ChangeMessageVisibilityBatch(queue);
61 const getQueueAttributes = yield* SQS.GetQueueAttributes(queue);
62 const purgeQueue = yield* SQS.PurgeQueue(queue);
63 const listDeadLetterSourceQueues =
64 yield* SQS.ListDeadLetterSourceQueues(dlq);
65 const startMessageMoveTask = yield* SQS.StartMessageMoveTask(dlq, {
66 destination: moveSource,
67 });
68 const cancelMessageMoveTask = yield* SQS.CancelMessageMoveTask(dlq);
69 const listMessageMoveTasks = yield* SQS.ListMessageMoveTasks(dlq);
70 const queueUrl = yield* queue.queueUrl;
71 const dlqUrl = yield* dlq.queueUrl;
72 const moveSourceUrl = yield* moveSource.queueUrl;
73
74 return {
75 fetch: Effect.gen(function* () {
76 const request = yield* HttpServerRequest;
77 const url = new URL(request.originalUrl);
78 const pathname = url.pathname;
79
80 yield* Effect.logInfo(`Request: ${request.method} ${pathname}`);
81
82 if (request.method === "GET" && pathname === "/ready") {
83 return yield* HttpServerResponse.json({
84 ok: true,
85 queueUrl: yield* queueUrl,
86 dlqUrl: yield* dlqUrl,
87 moveSourceUrl: yield* moveSourceUrl,
88 });
89 }
90
91 if (request.method === "POST" && pathname === "/send") {
92 const body = (yield* request.json) as unknown as {
93 messageBody: string;
94 };
95 const result = yield* sendMessage({
96 MessageBody: body.messageBody,
97 });
98 return yield* HttpServerResponse.json({
99 messageId: result.MessageId,
100 });
101 }
102
103 if (request.method === "POST" && pathname === "/send-batch") {
104 const body = (yield* request.json) as unknown as {
105 entries: { id: string; messageBody: string }[];
106 };
107 const result = yield* sendMessageBatch({
108 Entries: body.entries.map((entry) => ({
109 Id: entry.id,
110 MessageBody: entry.messageBody,
111 })),
112 });
113 return yield* HttpServerResponse.json({
114 successful: (result.Successful ?? []).map((s) => ({
115 id: s.Id,
116 messageId: s.MessageId,
117 })),
118 failed: (result.Failed ?? []).map((f) => ({
119 id: f.Id,
120 code: f.Code,
121 })),
122 });
123 }
124
125 if (request.method === "POST" && pathname === "/receive") {
126 const body = (yield* request.json) as unknown as {
127 maxNumberOfMessages?: number;
128 waitTimeSeconds?: number;
129 visibilityTimeout?: number;
130 };
131 const result = yield* receiveMessage({
132 MaxNumberOfMessages: body.maxNumberOfMessages ?? 10,
133 WaitTimeSeconds: body.waitTimeSeconds ?? 2,
134 ...(body.visibilityTimeout !== undefined
135 ? { VisibilityTimeout: body.visibilityTimeout }
136 : {}),
137 });
138 return yield* HttpServerResponse.json({
139 messages: (result.Messages ?? []).map((message) => ({
140 messageId: message.MessageId,
141 body: message.Body,
142 receiptHandle: message.ReceiptHandle,
143 })),
144 });
145 }
146
147 if (request.method === "POST" && pathname === "/delete") {
148 const body = (yield* request.json) as unknown as {
149 receiptHandle: string;
150 };
151 yield* deleteMessage({ ReceiptHandle: body.receiptHandle });
152 return yield* HttpServerResponse.json({ success: true });
153 }
154
155 if (request.method === "POST" && pathname === "/delete-batch") {
156 const body = (yield* request.json) as unknown as {
157 entries: { id: string; receiptHandle: string }[];
158 };
159 const result = yield* deleteMessageBatch({
160 Entries: body.entries.map((entry) => ({
161 Id: entry.id,
162 ReceiptHandle: entry.receiptHandle,
163 })),
164 });
165 return yield* HttpServerResponse.json({
166 successful: (result.Successful ?? []).map((s) => s.Id),
167 failed: (result.Failed ?? []).map((f) => ({
168 id: f.Id,
169 code: f.Code,
170 })),
171 });
172 }
173
174 if (request.method === "POST" && pathname === "/change-visibility") {
175 const body = (yield* request.json) as unknown as {
176 receiptHandle: string;
177 visibilityTimeout: number;
178 };
179 yield* changeMessageVisibility({
180 ReceiptHandle: body.receiptHandle,
181 VisibilityTimeout: body.visibilityTimeout,
182 });
183 return yield* HttpServerResponse.json({ success: true });
184 }
185
186 if (
187 request.method === "POST" &&
188 pathname === "/change-visibility-batch"
189 ) {
190 const body = (yield* request.json) as unknown as {
191 entries: {
192 id: string;
193 receiptHandle: string;
194 visibilityTimeout: number;
195 }[];
196 };
197 const result = yield* changeMessageVisibilityBatch({
198 Entries: body.entries.map((entry) => ({
199 Id: entry.id,
200 ReceiptHandle: entry.receiptHandle,
201 VisibilityTimeout: entry.visibilityTimeout,
202 })),
203 });
204 return yield* HttpServerResponse.json({
205 successful: (result.Successful ?? []).map((s) => s.Id),
206 failed: (result.Failed ?? []).map((f) => ({
207 id: f.Id,
208 code: f.Code,
209 })),
210 });
211 }
212
213 if (request.method === "POST" && pathname === "/attributes") {
214 const body = (yield* request.json) as unknown as {
215 attributeNames?: string[];
216 };
217 const result = yield* getQueueAttributes({
218 AttributeNames: body.attributeNames ?? ["All"],
219 });
220 return yield* HttpServerResponse.json({
221 attributes: result.Attributes ?? {},
222 });
223 }
224
225 if (request.method === "POST" && pathname === "/purge") {
226 yield* purgeQueue();
227 return yield* HttpServerResponse.json({ success: true });
228 }
229
230 if (request.method === "POST" && pathname === "/dlq-sources") {
231 const result = yield* listDeadLetterSourceQueues();
232 return yield* HttpServerResponse.json({
233 queueUrls: result.queueUrls,
234 });
235 }
236
237 if (request.method === "POST" && pathname === "/move/start") {
238 const body = (yield* request.json) as unknown as {
239 maxNumberOfMessagesPerSecond?: number;
240 };
241 const result = yield* startMessageMoveTask({
242 ...(body.maxNumberOfMessagesPerSecond !== undefined
243 ? {
244 MaxNumberOfMessagesPerSecond:
245 body.maxNumberOfMessagesPerSecond,
246 }
247 : {}),
248 });
249 return yield* HttpServerResponse.json({
250 taskHandle: result.TaskHandle,
251 });
252 }
253
254 if (request.method === "POST" && pathname === "/move/list") {
255 const result = yield* listMessageMoveTasks({ MaxResults: 10 });
256 return yield* HttpServerResponse.json({
257 tasks: (result.Results ?? []).map((task) => ({
258 taskHandle: task.TaskHandle,
259 status: task.Status,
260 moved: task.ApproximateNumberOfMessagesMoved,
261 })),
262 });
263 }
264
265 if (request.method === "POST" && pathname === "/move/cancel") {
266 const body = (yield* request.json) as unknown as {
267 taskHandle: string;
268 };
269 // A task that already finished is a benign race: SQS returns either
270 // ResourceNotFoundException or this exact typed validation message.
271 // Preserve every other InvalidParameterValueException as a failure.
272 const canceled = yield* cancelMessageMoveTask({
273 TaskHandle: body.taskHandle,
274 }).pipe(
275 Effect.map(() => true),
276 Effect.catchTag("ResourceNotFoundException", () =>
277 Effect.succeed(false),
278 ),
279 Effect.catchTag("InvalidParameterValueException", (error) =>
280 error.message === "Only active tasks can be cancelled."
281 ? Effect.succeed(false)
282 : Effect.fail(error),
283 ),
284 );
285 return yield* HttpServerResponse.json({ canceled });
286 }
287
288 return yield* HttpServerResponse.json(
289 { error: "Not found", method: request.method, pathname },
290 { status: 404 },
291 );
292 }).pipe(
293 // Lambda's default failure response is only "Internal Server Error",
294 // which hides the typed SQS rejection that made the route fail. This
295 // is a test-only handler: return the pretty Effect cause so the live
296 // test log identifies the exact capability/IAM/consistency failure.
297 Effect.catchCause((cause) =>
298 HttpServerResponse.json(
299 { error: Cause.pretty(cause) },
300 { status: 500 },
301 ),
302 ),
303 ),
304 };
305 }).pipe(
306 Effect.provide(
307 Layer.mergeAll(
308 SQS.SendMessageHttp,
309 SQS.SendMessageBatchHttp,
310 SQS.ReceiveMessageHttp,
311 SQS.DeleteMessageHttp,
312 SQS.DeleteMessageBatchHttp,
313 SQS.ChangeMessageVisibilityHttp,
314 SQS.ChangeMessageVisibilityBatchHttp,
315 SQS.GetQueueAttributesHttp,
316 SQS.PurgeQueueHttp,
317 SQS.ListDeadLetterSourceQueuesHttp,
318 SQS.StartMessageMoveTaskHttp,
319 SQS.CancelMessageMoveTaskHttp,
320 SQS.ListMessageMoveTasksHttp,
321 ),
322 ),
323 ),
324);