1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as Lambda from "@distilled.cloud/aws/lambda";
4import * as SQS from "@distilled.cloud/aws/sqs";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import QueueEventSourceFunctionLive, {
11 QueueEventSourceFunction,
12} from "./event-source-handler.ts";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16describe.sequential(
17 "AWS.SQS.QueueEventSource",
18 { tags: ["provider:aws", "provider:aws:lambda", "provider:aws:sqs", "live"] },
19 () => {
20 test.provider(
21 "delivers real SQS messages through the Lambda event source",
22 (stack) =>
23 Effect.gen(function* () {
24 yield* stack.destroy();
25
26 const fn = yield* stack.deploy(
27 QueueEventSourceFunction.pipe(
28 Effect.provide(QueueEventSourceFunctionLive),
29 ),
30 );
31
32 const functionUrl = fn.functionUrl!;
33
34 // First request rides out cold-start / URL propagation; keep polling
35 // until the fixture reports both queue identifiers.
36 const { sourceQueueUrl, sourceQueueArn, resultQueueUrl } =
37 yield* HttpClient.get(functionUrl).pipe(
38 Effect.timeout("4 seconds"),
39 Effect.mapError(
40 () => new FunctionNotReady("Function URL request timed out"),
41 ),
42 Effect.flatMap((response) =>
43 response.status === 200
44 ? (response.json as Effect.Effect<{
45 sourceQueueUrl?: string;
46 sourceQueueArn?: string;
47 resultQueueUrl?: string;
48 }>)
49 : Effect.fail(
50 new FunctionNotReady(
51 `Function not ready: ${response.status}`,
52 ),
53 ),
54 ),
55 Effect.flatMap((body) =>
56 body.sourceQueueUrl &&
57 body.sourceQueueArn &&
58 body.resultQueueUrl
59 ? Effect.succeed(
60 body as {
61 sourceQueueUrl: string;
62 sourceQueueArn: string;
63 resultQueueUrl: string;
64 },
65 )
66 : Effect.fail(
67 new FunctionNotReady(
68 "Function returned empty queue identifiers",
69 ),
70 ),
71 ),
72 Effect.retry({
73 schedule: Schedule.max([
74 Schedule.fixed("4 seconds"),
75 Schedule.recurs(10),
76 ]),
77 }),
78 );
79
80 // The event-source mapping activates asynchronously after deploy.
81 const mapping = yield* waitForEventSourceMappingEnabled(
82 fn.functionName,
83 sourceQueueArn,
84 );
85 expect(mapping.State).toEqual("Enabled");
86
87 // Send a message to the source queue out-of-band; the Lambda handler
88 // forwards its body to the result queue via QueueSink.
89 const messageBody = `event-source-${crypto.randomUUID()}`;
90 yield* SQS.sendMessage({
91 QueueUrl: sourceQueueUrl,
92 MessageBody: messageBody,
93 });
94
95 // Poll the result queue until the forwarded body shows up. Bounded:
96 // ~30 polls, each an SQS long-poll of 2s.
97 const received = yield* Effect.gen(function* () {
98 const result = yield* SQS.receiveMessage({
99 QueueUrl: resultQueueUrl,
100 MaxNumberOfMessages: 10,
101 WaitTimeSeconds: 2,
102 });
103 const match = (result.Messages ?? []).find(
104 (message) => message.Body === messageBody,
105 );
106 if (!match?.ReceiptHandle) {
107 return yield* Effect.fail(new MessageNotDelivered());
108 }
109 yield* SQS.deleteMessage({
110 QueueUrl: resultQueueUrl,
111 ReceiptHandle: match.ReceiptHandle,
112 });
113 return match.Body!;
114 }).pipe(
115 Effect.retry({
116 while: (error) => error._tag === "MessageNotDelivered",
117 schedule: Schedule.max([
118 Schedule.fixed("3 seconds"),
119 Schedule.recurs(10),
120 ]),
121 }),
122 );
123
124 expect(received).toEqual(messageBody);
125
126 yield* stack.destroy();
127
128 // Out-of-band: both queues are actually gone after the destroy.
129 yield* assertQueueDeleted(sourceQueueUrl);
130 yield* assertQueueDeleted(resultQueueUrl);
131 }),
132 { timeout: 240_000 },
133 );
134 },
135);
136
137const waitForEventSourceMappingEnabled = Effect.fn(function* (
138 functionName: string,
139 eventSourceArn: string,
140) {
141 return yield* Lambda.listEventSourceMappings({
142 FunctionName: functionName,
143 EventSourceArn: eventSourceArn,
144 }).pipe(
145 Effect.flatMap((result) => {
146 const mapping = result.EventSourceMappings?.[0];
147 if (!mapping || mapping.State !== "Enabled") {
148 return Effect.fail(new EventSourceMappingNotReady());
149 }
150 return Effect.succeed(mapping);
151 }),
152 Effect.retry({
153 while: (error) => error._tag === "EventSourceMappingNotReady",
154 schedule: Schedule.max([
155 Schedule.fixed("5 seconds"),
156 Schedule.recurs(10),
157 ]),
158 }),
159 );
160});
161
162class QueueStillExists extends Data.TaggedError("QueueStillExists") {}
163
164/** Poll (bounded) until GetQueueAttributes reports the queue gone. */
165const assertQueueDeleted = Effect.fn(function* (queueUrl: string) {
166 yield* SQS.getQueueAttributes({
167 QueueUrl: queueUrl,
168 AttributeNames: ["All"],
169 }).pipe(
170 Effect.flatMap(() => Effect.fail(new QueueStillExists())),
171 Effect.retry({
172 // SQS DeleteQueue propagation is documented at ~60s; poll on a fixed
173 // cadence with a bounded budget.
174 while: (e) => e._tag === "QueueStillExists",
175 schedule: Schedule.max([
176 Schedule.spaced("5 seconds"),
177 Schedule.recurs(10),
178 ]),
179 }),
180 Effect.catchTag("QueueDoesNotExist", () => Effect.void),
181 );
182});
183
184class EventSourceMappingNotReady extends Data.TaggedError(
185 "EventSourceMappingNotReady",
186) {}
187
188class MessageNotDelivered extends Data.TaggedError("MessageNotDelivered") {}
189
190class FunctionNotReady extends Data.TaggedError("FunctionNotReady")<{
191 message: string;
192}> {
193 constructor(message: string) {
194 super({ message });
195 }
196}