1import * as AWS from "@/AWS";
2import * as Output from "@/Output";
3import * as Core from "@/Test/Core";
4import * as Test from "@/Test/Alchemy";
5import * as agw2 from "@distilled.cloud/aws/apigatewayv2";
6import * as cip from "@distilled.cloud/aws/cognito-identity-provider";
7import * as ddb from "@distilled.cloud/aws/dynamodb";
8import * as lambda from "@distilled.cloud/aws/lambda";
9import * as S3 from "@distilled.cloud/aws/s3";
10import * as sfn from "@distilled.cloud/aws/sfn";
11import * as sqs from "@distilled.cloud/aws/sqs";
12import { describe, expect } from "alchemy-test";
13import * as Data from "effect/Data";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Schedule from "effect/Schedule";
17import * as Stream from "effect/Stream";
18import * as HttpClient from "effect/http/HttpClient";
19import * as HttpClientRequest from "effect/http/HttpClientRequest";
20import SmokeApiFunctionLive, {
21 SmokeApiFunction,
22} from "./fixtures/api-handler.ts";
23import { ServerlessResources } from "./fixtures/serverless-resources.ts";
24import SmokeWorkerFunctionLive, {
25 SmokeWorkerFunction,
26} from "./fixtures/worker-handler.ts";
27
28/**
29 * Phase-1 exit flagship: the full-stack serverless story in ONE stack.
30 *
31 * Cognito UserPool + Client → API Gateway v2 HTTP API (JWT authorizer)
32 * → API Lambda binding DynamoDB, S3 presign, SQS, and an EXPRESS Step
33 * Function → worker Lambda consuming the jobs queue via the SQS event
34 * source. The test drives the story over HTTP and verifies every side
35 * effect out-of-band via distilled.
36 */
37
38const testOptions = { providers: AWS.providers() };
39const { test, beforeAll, afterAll } = Test.make(testOptions);
40const sharedStack = Core.scratchStack(testOptions, "ServerlessSmoke");
41
42// Lambda cold start + API Gateway route/permission propagation can take
43// well over 60s on a fresh deploy under parallel-suite load.
44const readinessPolicy = Schedule.max([
45 Schedule.fixed("2 seconds"),
46 Schedule.recurs(90),
47]);
48
49interface StackOutputs {
50 url: string;
51 apiId: string;
52 userPoolId: string;
53 clientId: string;
54 tableName: string;
55 bucketName: string;
56 jobsQueueUrl: string;
57 jobsQueueArn: string;
58 resultsQueueUrl: string;
59 machineArn: string;
60 workerFunctionName: string;
61 apiFunctionName: string;
62}
63
64let outputs: StackOutputs;
65let baseUrl: string;
66
67class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
68 readonly status: number;
69 readonly body: string;
70}> {}
71
72class UnexpectedStatus extends Data.TaggedError("UnexpectedStatus")<{
73 readonly status: number;
74 readonly expected: number;
75}> {}
76
77class StillExists extends Data.TaggedError("StillExists")<{
78 readonly what: string;
79}> {}
80
81class MessageNotDelivered extends Data.TaggedError("MessageNotDelivered") {}
82
83class EventSourceMappingNotReady extends Data.TaggedError(
84 "EventSourceMappingNotReady",
85) {}
86
87// Retry transient 5xx only (cold re-init, IAM propagation); 4xx and
88// assertion failures surface immediately.
89const send = (request: HttpClientRequest.HttpClientRequest) =>
90 HttpClient.execute(request).pipe(
91 Effect.flatMap((response) =>
92 response.status >= 500
93 ? response.text.pipe(
94 Effect.flatMap((body) =>
95 Effect.fail(
96 new TransientUpstream({ status: response.status, body }),
97 ),
98 ),
99 )
100 : Effect.succeed(response),
101 ),
102 Effect.retry({
103 while: (e) => e._tag === "TransientUpstream",
104 schedule: Schedule.max([
105 Schedule.exponential("500 millis"),
106 Schedule.recurs(6),
107 ]),
108 }),
109 );
110
111// Poll until the API answers a request with the expected status — rides out
112// the short window while a freshly-created route/authorizer propagates.
113const awaitStatus = (
114 request: HttpClientRequest.HttpClientRequest,
115 expected: number,
116 times = 20,
117) =>
118 HttpClient.execute(request).pipe(
119 Effect.flatMap((response) =>
120 response.status === expected
121 ? Effect.succeed(response)
122 : Effect.fail(
123 new UnexpectedStatus({ status: response.status, expected }),
124 ),
125 ),
126 Effect.retry({
127 while: (e) => e._tag === "UnexpectedStatus",
128 schedule: Schedule.max([
129 Schedule.fixed("2 seconds"),
130 Schedule.recurs(times),
131 ]),
132 }),
133 );
134
135// Bounded wait until an out-of-band probe reports the resource gone.
136const waitUntilGone = <E, R>(
137 what: string,
138 probe: Effect.Effect<boolean, E, R>,
139) =>
140 probe.pipe(
141 Effect.flatMap((gone) =>
142 gone ? Effect.void : Effect.fail(new StillExists({ what })),
143 ),
144 Effect.retry({
145 // `instanceof`, not `_tag`: the probe's error type `E` is generic and
146 // carries no tag constraint.
147 while: (e) => e instanceof StillExists,
148 schedule: Schedule.max([
149 Schedule.fixed("3 seconds"),
150 Schedule.recurs(30),
151 ]),
152 }),
153 );
154
155const deployProgram = Effect.gen(function* () {
156 const { pool, client, table, bucket, jobsQueue, resultsQueue, machine } =
157 yield* ServerlessResources;
158 const apiFn = yield* SmokeApiFunction;
159 const worker = yield* SmokeWorkerFunction;
160
161 const { api, integration, url } = yield* AWS.ApiGatewayV2.HttpApi(
162 "SmokeHttpApi",
163 { handler: apiFn },
164 );
165
166 // JWT authorizer wired to the Cognito user pool. The pool id embeds its
167 // region (`us-west-2_Abc123`), so the issuer URL is derived from it.
168 const authorizer = yield* AWS.ApiGatewayV2.Authorizer("SmokeJwtAuthorizer", {
169 api,
170 authorizerType: "JWT",
171 identitySource: ["$request.header.Authorization"],
172 jwtConfiguration: {
173 Issuer: Output.map(
174 pool.userPoolId,
175 (id) => `https://cognito-idp.${id.split("_")[0]}.amazonaws.com/${id}`,
176 ),
177 Audience: [client.clientId],
178 },
179 });
180
181 yield* AWS.ApiGatewayV2.Route("TodoPostRoute", {
182 api,
183 routeKey: "POST /todo",
184 integration,
185 authorizationType: "JWT",
186 authorizerId: authorizer.authorizerId,
187 });
188 yield* AWS.ApiGatewayV2.Route("TodoGetRoute", {
189 api,
190 routeKey: "GET /todo",
191 integration,
192 authorizationType: "JWT",
193 authorizerId: authorizer.authorizerId,
194 });
195
196 return {
197 url,
198 apiId: api.apiId,
199 userPoolId: pool.userPoolId,
200 clientId: client.clientId,
201 tableName: table.tableName,
202 bucketName: bucket.bucketName,
203 jobsQueueUrl: jobsQueue.queueUrl,
204 jobsQueueArn: jobsQueue.queueArn,
205 resultsQueueUrl: resultsQueue.queueUrl,
206 machineArn: machine.stateMachineArn,
207 workerFunctionName: worker.functionName,
208 apiFunctionName: apiFn.functionName,
209 };
210}).pipe(
211 Effect.provide(Layer.mergeAll(SmokeApiFunctionLive, SmokeWorkerFunctionLive)),
212);
213
214describe.sequential(
215 "Serverless smoke",
216 {
217 tags: [
218 "provider:aws",
219 "provider:aws:apigatewayv2",
220 "provider:aws:cognito",
221 "provider:aws:dynamodb",
222 "provider:aws:lambda",
223 "provider:aws:s3",
224 "provider:aws:sqs",
225 "provider:aws:stepfunctions",
226 "live",
227 ],
228 },
229 () => {
230 beforeAll(
231 Effect.gen(function* () {
232 yield* Effect.logInfo("Serverless smoke: destroying previous stack");
233 yield* sharedStack.destroy();
234
235 yield* Effect.logInfo("Serverless smoke: deploying stack");
236 outputs = (yield* sharedStack.deploy(deployProgram)) as StackOutputs;
237 baseUrl = outputs.url.replace(/\/+$/, "");
238
239 yield* Effect.logInfo(`Serverless smoke: probing ${baseUrl}/config`);
240 const config = (yield* HttpClient.get(`${baseUrl}/config`).pipe(
241 Effect.flatMap((response) =>
242 response.status === 200
243 ? response.json
244 : Effect.fail(new Error(`API not ready: ${response.status}`)),
245 ),
246 Effect.tapError((error) =>
247 Effect.logWarning(
248 `Serverless smoke: API not ready yet (${String(error)})`,
249 ),
250 ),
251 Effect.retry({ schedule: readinessPolicy }),
252 )) as {
253 userPoolId: string;
254 clientId: string;
255 tableName: string;
256 bucketName: string;
257 jobsQueueUrl: string;
258 resultsQueueUrl: string;
259 machineArn: string;
260 };
261
262 // The fixture Lambdas and the stack program declare the shared
263 // resources through ONE layer — the identifiers the deployed Lambda
264 // observes must be the same physical resources the stack returned.
265 expect(config.userPoolId).toBe(outputs.userPoolId);
266 expect(config.clientId).toBe(outputs.clientId);
267 expect(config.tableName).toBe(outputs.tableName);
268 expect(config.bucketName).toBe(outputs.bucketName);
269 expect(config.jobsQueueUrl).toBe(outputs.jobsQueueUrl);
270 expect(config.resultsQueueUrl).toBe(outputs.resultsQueueUrl);
271 expect(config.machineArn).toBe(outputs.machineArn);
272 }),
273 { timeout: 300_000 },
274 );
275
276 afterAll(sharedStack.destroy(), { timeout: 120_000 });
277
278 test.provider(
279 "unauthenticated $default route serves; JWT routes are gated",
280 (_stack) =>
281 Effect.gen(function* () {
282 const free = yield* send(
283 HttpClientRequest.get(`${baseUrl}/signup-free`),
284 );
285 expect(free.status).toBe(200);
286 expect((yield* free.json) as object).toEqual({ ok: true });
287
288 // No token → the JWT authorizer must answer 401 before the Lambda
289 // is ever invoked. Polls through route/authorizer propagation.
290 const unauthorized = yield* awaitStatus(
291 HttpClientRequest.post(`${baseUrl}/todo`).pipe(
292 HttpClientRequest.bodyJsonUnsafe({ id: "nope", text: "nope" }),
293 ),
294 401,
295 );
296 expect(unauthorized.status).toBe(401);
297
298 // Garbage token → still 401.
299 const garbage = yield* HttpClient.execute(
300 HttpClientRequest.post(`${baseUrl}/todo`).pipe(
301 HttpClientRequest.setHeader("Authorization", "Bearer not-a-jwt"),
302 HttpClientRequest.bodyJsonUnsafe({ id: "nope", text: "nope" }),
303 ),
304 );
305 expect(garbage.status).toBe(401);
306
307 // The gated write never reached DynamoDB.
308 const item = yield* ddb.getItem({
309 TableName: outputs.tableName,
310 Key: { pk: { S: "todo" }, sk: { S: "nope" } },
311 ConsistentRead: true,
312 });
313 expect(item.Item).toBeUndefined();
314 }),
315 { timeout: 120_000 },
316 );
317
318 test.provider(
319 "Cognito JWT (minted via UserPoolAdmin/UserPoolAuth bindings) unlocks the todo API",
320 (_stack) =>
321 Effect.gen(function* () {
322 const tokens = (yield* send(
323 HttpClientRequest.post(
324 `${baseUrl}/auth?username=serverless-smoke-user`,
325 ),
326 ).pipe(Effect.flatMap((r) => r.json))) as {
327 idToken: string | undefined;
328 accessToken: string | undefined;
329 tokenType: string | undefined;
330 };
331 expect(tokens.tokenType).toBe("Bearer");
332 expect(tokens.idToken).toBeTruthy();
333
334 const authorize = HttpClientRequest.setHeader(
335 "Authorization",
336 `Bearer ${tokens.idToken}`,
337 );
338
339 const text = "ship the serverless smoke";
340 const wrote = yield* awaitStatus(
341 HttpClientRequest.post(`${baseUrl}/todo`).pipe(
342 authorize,
343 HttpClientRequest.bodyJsonUnsafe({ id: "todo-1", text }),
344 ),
345 200,
346 10,
347 );
348 expect((yield* wrote.json) as object).toEqual({
349 ok: true,
350 id: "todo-1",
351 });
352
353 const read = yield* send(
354 HttpClientRequest.get(`${baseUrl}/todo?id=todo-1`).pipe(authorize),
355 );
356 expect(read.status).toBe(200);
357 expect((yield* read.json) as object).toEqual({
358 item: { id: "todo-1", text },
359 });
360
361 // Out-of-band: the item really landed in the table.
362 const raw = yield* ddb.getItem({
363 TableName: outputs.tableName,
364 Key: { pk: { S: "todo" }, sk: { S: "todo-1" } },
365 ConsistentRead: true,
366 });
367 expect(raw.Item?.text?.S).toBe(text);
368 }),
369 { timeout: 120_000 },
370 );
371
372 test.provider(
373 "presigned PUT URL uploads into the bucket",
374 (_stack) =>
375 Effect.gen(function* () {
376 const key = "smoke/upload.txt";
377 const body = "uploaded through the serverless smoke story";
378
379 const presigned = (yield* send(
380 HttpClientRequest.get(
381 `${baseUrl}/upload-url?key=${encodeURIComponent(key)}&contentType=text/plain`,
382 ),
383 ).pipe(Effect.flatMap((r) => r.json))) as { url: string };
384 expect(presigned.url).toContain(outputs.bucketName);
385
386 const put = yield* send(
387 HttpClientRequest.put(presigned.url).pipe(
388 HttpClientRequest.bodyText(body, "text/plain"),
389 ),
390 );
391 expect(put.status).toBe(200);
392
393 // Out-of-band: fetch the object back via distilled.
394 const got = yield* S3.getObject({
395 Bucket: outputs.bucketName,
396 Key: key,
397 });
398 expect(got.ContentType).toBe("text/plain");
399 const text = yield* Stream.mkString(Stream.decodeText(got.Body!));
400 expect(text).toBe(body);
401 }),
402 { timeout: 120_000 },
403 );
404
405 test.provider(
406 "enqueued message is observed by the worker Lambda via the SQS event source",
407 (_stack) =>
408 Effect.gen(function* () {
409 // The event-source mapping activates asynchronously after deploy.
410 yield* lambda
411 .listEventSourceMappings({
412 FunctionName: outputs.workerFunctionName,
413 EventSourceArn: outputs.jobsQueueArn,
414 })
415 .pipe(
416 Effect.flatMap((result) => {
417 const mapping = result.EventSourceMappings?.[0];
418 return mapping?.State === "Enabled"
419 ? Effect.succeed(mapping)
420 : Effect.fail(new EventSourceMappingNotReady());
421 }),
422 Effect.retry({
423 while: (e) => e._tag === "EventSourceMappingNotReady",
424 schedule: Schedule.max([
425 Schedule.fixed("2 seconds"),
426 Schedule.recurs(30),
427 ]),
428 }),
429 );
430
431 const message = yield* Effect.sync(
432 () => `smoke-job-${crypto.randomUUID()}`,
433 );
434 const enqueued = (yield* send(
435 HttpClientRequest.post(`${baseUrl}/enqueue`).pipe(
436 HttpClientRequest.bodyJsonUnsafe({ message }),
437 ),
438 ).pipe(Effect.flatMap((r) => r.json))) as { messageId: string };
439 expect(enqueued.messageId).toBeTruthy();
440
441 // The worker forwards `processed:{body}` into the results queue —
442 // poll it out-of-band (bounded: ~30 polls × 2s long-poll).
443 const received = yield* Effect.gen(function* () {
444 const result = yield* sqs.receiveMessage({
445 QueueUrl: outputs.resultsQueueUrl,
446 MaxNumberOfMessages: 10,
447 WaitTimeSeconds: 2,
448 });
449 const match = (result.Messages ?? []).find(
450 (m) => m.Body === `processed:${message}`,
451 );
452 if (!match?.ReceiptHandle) {
453 return yield* Effect.fail(new MessageNotDelivered());
454 }
455 yield* sqs.deleteMessage({
456 QueueUrl: outputs.resultsQueueUrl,
457 ReceiptHandle: match.ReceiptHandle,
458 });
459 return match.Body!;
460 }).pipe(
461 Effect.retry({
462 while: (e) => e._tag === "MessageNotDelivered",
463 schedule: Schedule.max([
464 Schedule.fixed("2 seconds"),
465 Schedule.recurs(30),
466 ]),
467 }),
468 );
469 expect(received).toBe(`processed:${message}`);
470 }),
471 { timeout: 120_000 },
472 );
473
474 test.provider(
475 "EXPRESS Step Function round-trips synchronously from the API Lambda",
476 (_stack) =>
477 Effect.gen(function* () {
478 const result = (yield* send(
479 HttpClientRequest.post(`${baseUrl}/compute`).pipe(
480 HttpClientRequest.bodyJsonUnsafe({ value: 21 }),
481 ),
482 ).pipe(Effect.flatMap((r) => r.json))) as {
483 status: string;
484 output: string | undefined;
485 error: string | undefined;
486 };
487
488 expect(result.error).toBeUndefined();
489 expect(result.status).toBe("SUCCEEDED");
490 const output = JSON.parse(result.output!) as {
491 computed: boolean;
492 echo: { value: number };
493 };
494 expect(output.computed).toBe(true);
495 expect(output.echo).toEqual({ value: 21 });
496 }),
497 { timeout: 120_000 },
498 );
499
500 test.provider(
501 "destroy removes the pool, api, table, bucket, queues, and state machine",
502 (_stack) =>
503 Effect.gen(function* () {
504 yield* sharedStack.destroy();
505
506 yield* waitUntilGone(
507 "user pool",
508 cip.describeUserPool({ UserPoolId: outputs.userPoolId }).pipe(
509 Effect.map(() => false),
510 Effect.catchTag("ResourceNotFoundException", () =>
511 Effect.succeed(true),
512 ),
513 ),
514 );
515
516 yield* waitUntilGone(
517 "http api",
518 agw2.getApi({ ApiId: outputs.apiId }).pipe(
519 Effect.map(() => false),
520 Effect.catchTag("NotFoundException", () => Effect.succeed(true)),
521 ),
522 );
523
524 yield* waitUntilGone(
525 "table",
526 ddb.describeTable({ TableName: outputs.tableName }).pipe(
527 Effect.map(() => false),
528 Effect.catchTag("ResourceNotFoundException", () =>
529 Effect.succeed(true),
530 ),
531 ),
532 );
533
534 yield* waitUntilGone(
535 "bucket",
536 S3.headBucket({ Bucket: outputs.bucketName }).pipe(
537 Effect.map(() => false),
538 Effect.catchTag("NotFound", () => Effect.succeed(true)),
539 ),
540 );
541
542 yield* waitUntilGone(
543 "jobs queue",
544 sqs
545 .getQueueUrl({
546 QueueName: queueNameFromUrl(outputs.jobsQueueUrl),
547 })
548 .pipe(
549 Effect.map(() => false),
550 Effect.catchTag("QueueDoesNotExist", () =>
551 Effect.succeed(true),
552 ),
553 ),
554 );
555
556 yield* waitUntilGone(
557 "results queue",
558 sqs
559 .getQueueUrl({
560 QueueName: queueNameFromUrl(outputs.resultsQueueUrl),
561 })
562 .pipe(
563 Effect.map(() => false),
564 Effect.catchTag("QueueDoesNotExist", () =>
565 Effect.succeed(true),
566 ),
567 ),
568 );
569
570 yield* waitUntilGone(
571 "api lambda",
572 lambda.getFunction({ FunctionName: outputs.apiFunctionName }).pipe(
573 Effect.map(() => false),
574 Effect.catchTag("ResourceNotFoundException", () =>
575 Effect.succeed(true),
576 ),
577 ),
578 );
579
580 yield* waitUntilGone(
581 "worker lambda",
582 lambda
583 .getFunction({ FunctionName: outputs.workerFunctionName })
584 .pipe(
585 Effect.map(() => false),
586 Effect.catchTag("ResourceNotFoundException", () =>
587 Effect.succeed(true),
588 ),
589 ),
590 );
591
592 yield* waitUntilGone(
593 "state machine",
594 sfn
595 .describeStateMachine({ stateMachineArn: outputs.machineArn })
596 .pipe(
597 Effect.map((machine) => machine.status === "DELETING"),
598 Effect.catchTag("StateMachineDoesNotExist", () =>
599 Effect.succeed(true),
600 ),
601 ),
602 );
603 }),
604 { timeout: 120_000 },
605 );
606 },
607);
608
609const queueNameFromUrl = (queueUrl: string) => queueUrl.split("/").at(-1)!;