1/**
2 * Event-source-mapping conformance against the floci emulator, driven
3 * entirely through OUR client stack (distilled SDK + alchemy providers).
4 *
5 * Topology (all local, dev mode, no profile stub):
6 * - `AWS.SQS.Queue` + `AWS.S3.Bucket` + `AWS.IAM.Role` deploy through the
7 * dualized providers into floci;
8 * - a RAW floci Lambda is created via distilled `lambda.createFunction`
9 * against the emulator endpoint (the Lambda.Function RESOURCE is out of
10 * scope for F1) — its handler writes each consumed SQS message into the
11 * bucket;
12 * - the event source mapping is created via distilled
13 * `lambda.createEventSourceMapping` (the alchemy EventSourceMapping
14 * resource is NOT dualized: its ownership scan needs lambda ListTags on
15 * `event-source-mapping:` ARNs, which floci rejects — see the F1
16 * fork-patch candidates);
17 * - `sqs.sendMessage` through distilled, then a bounded poll for the
18 * object the function writes — pinning floci's local event pumping
19 * (ESM poller -> containerized Lambda -> S3 write-back) end to end.
20 *
21 * Requires Docker (floci + the Lambda runtime container); skipped when the
22 * daemon is unavailable.
23 */
24import * as AWS from "@/AWS";
25import * as Endpoint from "@/AWS/Endpoint.ts";
26import * as Region from "@/AWS/Region.ts";
27import * as Test from "@/Test/Alchemy";
28import { zipCode } from "@/Util/zip.ts";
29import { Credentials } from "@distilled.cloud/aws/Credentials";
30import type { RegionName } from "@distilled.cloud/aws/Region";
31import * as Lambda from "@distilled.cloud/aws/lambda";
32import * as SQS from "@distilled.cloud/aws/sqs";
33import { expect } from "alchemy-test";
34import * as Effect from "effect/Effect";
35import * as Layer from "effect/Layer";
36import * as Redacted from "effect/Redacted";
37import * as Schedule from "effect/Schedule";
38import {
39 dockerAvailable,
40 FLOCI_ENDPOINT,
41 rawS3GetObject,
42} from "./fixtures/raw.ts";
43
44const { test } = Test.make({ providers: AWS.providers(), dev: true });
45
46/**
47 * Floci-scoped context for the raw distilled calls this test makes itself
48 * (create/delete the Lambda function, send the SQS message) — the same
49 * client stack the wrapped local providers use.
50 */
51const flociContext = Layer.mergeAll(
52 Endpoint.of(FLOCI_ENDPOINT),
53 Region.of("us-east-1"),
54 Layer.succeed(
55 Credentials,
56 Effect.succeed({
57 accessKeyId: Redacted.make("test"),
58 secretAccessKey: Redacted.make("test"),
59 sessionToken: undefined,
60 region: "us-east-1" as RegionName,
61 }),
62 ),
63);
64
65const FUNCTION_NAME = "alchemy-local-esm-consumer";
66const MARKER_KEY = "consumed-marker.txt";
67
68/**
69 * nodejs20.x handler: writes each consumed SQS record into the target
70 * bucket under the key carried in the message body. The Lambda runtime
71 * container gets `AWS_ENDPOINT_URL` + placeholder credentials from floci,
72 * so the bundled AWS SDK v3 targets the emulator; path-style is forced
73 * because virtual-host bucket DNS does not resolve locally.
74 */
75const handlerSource = /* js */ `
76import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3";
77const s3 = new S3Client({
78 endpoint: process.env.AWS_ENDPOINT_URL,
79 forcePathStyle: true,
80});
81export const handler = async (event) => {
82 for (const record of event.Records ?? []) {
83 await s3.send(
84 new PutObjectCommand({
85 Bucket: process.env.TARGET_BUCKET,
86 Key: record.body,
87 Body: record.messageId,
88 }),
89 );
90 }
91 return { batchItemFailures: [] };
92};
93`;
94
95const assumeRolePolicy = {
96 Version: "2012-10-17" as const,
97 Statement: [
98 {
99 Effect: "Allow" as const,
100 Principal: { Service: "lambda.amazonaws.com" },
101 Action: ["sts:AssumeRole"],
102 },
103 ],
104};
105
106test.provider.skipIf(!dockerAvailable)(
107 "SQS -> raw floci Lambda -> S3 event pumping through our client stack",
108 (stack) =>
109 Effect.gen(function* () {
110 yield* stack.destroy();
111
112 // Phase 1: the data plane the function needs (all floci-local).
113 const base = yield* stack.deploy(
114 Effect.gen(function* () {
115 const queue = yield* AWS.SQS.Queue("EsmQueue");
116 const bucket = yield* AWS.S3.Bucket("EsmTarget", {
117 forceDestroy: true,
118 });
119 const role = yield* AWS.IAM.Role("EsmRole", {
120 assumeRolePolicyDocument: assumeRolePolicy,
121 });
122 return { queue, bucket, role };
123 }),
124 );
125 expect(base.queue.queueArn).toContain(":000000000000:");
126 expect(base.bucket.bucketName).toBeDefined();
127
128 // Pre-clean: the emulator container is long-lived, so a previous
129 // interrupted run may have left the function (with a stale
130 // TARGET_BUCKET env) and event source mappings whose pollers spin on
131 // deleted queues. Delete mappings first, then the function.
132 const leftover = yield* Lambda.listEventSourceMappings({
133 FunctionName: FUNCTION_NAME,
134 }).pipe(
135 Effect.catchTag("ResourceNotFoundException", () =>
136 Effect.succeed({ EventSourceMappings: [] }),
137 ),
138 Effect.provide(flociContext),
139 );
140 yield* Effect.forEach(leftover.EventSourceMappings ?? [], (mapping) =>
141 mapping.UUID
142 ? Lambda.deleteEventSourceMapping({ UUID: mapping.UUID }).pipe(
143 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
144 Effect.provide(flociContext),
145 )
146 : Effect.void,
147 );
148 yield* Lambda.deleteFunction({ FunctionName: FUNCTION_NAME }).pipe(
149 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
150 Effect.provide(flociContext),
151 );
152
153 // Raw floci Lambda via distilled against the emulator endpoint.
154 const archive = yield* zipCode(handlerSource);
155 yield* Lambda.createFunction({
156 FunctionName: FUNCTION_NAME,
157 Runtime: "nodejs20.x",
158 Handler: "index.handler",
159 Role: base.role.roleArn,
160 Code: { ZipFile: archive },
161 Timeout: 30,
162 Environment: {
163 Variables: { TARGET_BUCKET: base.bucket.bucketName },
164 },
165 }).pipe(Effect.provide(flociContext));
166
167 // Wait (bounded) for the function to be Active before wiring the ESM.
168 const fn = yield* Lambda.getFunction({
169 FunctionName: FUNCTION_NAME,
170 }).pipe(
171 Effect.provide(flociContext),
172 Effect.repeat({
173 schedule: Schedule.spaced("2 seconds"),
174 until: (res): boolean => res.Configuration?.State === "Active",
175 times: 30,
176 }),
177 );
178 expect(fn.Configuration?.State).toBe("Active");
179
180 // Wire queue -> fn via distilled against the emulator (our client
181 // stack end to end).
182 const esm = yield* Lambda.createEventSourceMapping({
183 FunctionName: FUNCTION_NAME,
184 EventSourceArn: base.queue.queueArn,
185 BatchSize: 1,
186 Enabled: true,
187 }).pipe(Effect.provide(flociContext));
188 expect(esm.UUID).toBeDefined();
189
190 // Send through OUR client stack (distilled sqs against the emulator).
191 yield* SQS.sendMessage({
192 QueueUrl: base.queue.queueUrl,
193 MessageBody: MARKER_KEY,
194 }).pipe(Effect.provide(flociContext));
195
196 // Bounded poll: floci's ESM poller must deliver the message to the
197 // containerized function, whose S3 write-back proves consumption.
198 const consumed = yield* rawS3GetObject(
199 base.bucket.bucketName,
200 MARKER_KEY,
201 ).pipe(
202 Effect.repeat({
203 schedule: Schedule.spaced("3 seconds"),
204 until: (res): boolean => res.status === 200,
205 times: 40,
206 }),
207 );
208 expect(consumed.status).toBe(200);
209
210 // Cleanup: remove the raw mapping + function from the emulator, then
211 // destroy the stack (queue, bucket incl. the object, role).
212 if (esm.UUID) {
213 yield* Lambda.deleteEventSourceMapping({ UUID: esm.UUID }).pipe(
214 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
215 Effect.provide(flociContext),
216 );
217 }
218 yield* Lambda.deleteFunction({ FunctionName: FUNCTION_NAME }).pipe(
219 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
220 Effect.provide(flociContext),
221 );
222 yield* stack.destroy();
223 }),
224 {
225 tags: [
226 "provider:aws",
227 "provider:aws:iam",
228 "provider:aws:lambda",
229 "provider:aws:s3",
230 "provider:aws:sqs",
231 "local",
232 ],
233 timeout: 540_000,
234 },
235);