| 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 | */ |
| 24 | import * as AWS from "@/AWS"; |
| 25 | import * as Endpoint from "@/AWS/Endpoint.ts"; |
| 26 | import * as Region from "@/AWS/Region.ts"; |
| 27 | import * as Test from "@/Test/Alchemy"; |
| 28 | import { zipCode } from "@/Util/zip.ts"; |
| 29 | import { Credentials } from "@distilled.cloud/aws/Credentials"; |
| 30 | import type { RegionName } from "@distilled.cloud/aws/Region"; |
| 31 | import * as Lambda from "@distilled.cloud/aws/lambda"; |
| 32 | import * as SQS from "@distilled.cloud/aws/sqs"; |
| 33 | import { expect } from "alchemy-test"; |
| 34 | import * as Effect from "effect/Effect"; |
| 35 | import * as Layer from "effect/Layer"; |
| 36 | import * as Redacted from "effect/Redacted"; |
| 37 | import * as Schedule from "effect/Schedule"; |
| 38 | import { |
| 39 | dockerAvailable, |
| 40 | FLOCI_ENDPOINT, |
| 41 | rawS3GetObject, |
| 42 | } from "./fixtures/raw.ts"; |
| 43 | |
| 44 | const { 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 | */ |
| 51 | const 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 | |
| 65 | const FUNCTION_NAME = "alchemy-local-esm-consumer"; |
| 66 | const 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 | */ |
| 75 | const handlerSource = /* js */ ` |
| 76 | import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3"; |
| 77 | const s3 = new S3Client({ |
| 78 | endpoint: process.env.AWS_ENDPOINT_URL, |
| 79 | forcePathStyle: true, |
| 80 | }); |
| 81 | export 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 | |
| 95 | const 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 | |
| 106 | test.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 | ); |