1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as SQS from "@distilled.cloud/aws/sqs";
4import { describe, expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import PipeTargetFunctionLive, { PipeTargetFunction } from "./pipe-handler.ts";
10
11const { test } = Test.make({ providers: AWS.providers() });
12
13describe.sequential(
14 "AWS.Pipes delivery",
15 {
16 tags: [
17 "provider:aws",
18 "provider:aws:lambda",
19 "provider:aws:pipes",
20 "provider:aws:sqs",
21 "live",
22 ],
23 },
24 () => {
25 test.provider(
26 "delivers SQS messages through a pipe to the Lambda target",
27 (stack) =>
28 Effect.gen(function* () {
29 yield* stack.destroy();
30
31 const fn = yield* stack.deploy(
32 PipeTargetFunction.pipe(Effect.provide(PipeTargetFunctionLive)),
33 );
34
35 const functionUrl = fn.functionUrl!;
36
37 // First request rides out cold-start / URL propagation; keep polling
38 // until the fixture reports both queue identifiers.
39 const { sourceQueueUrl, sinkQueueUrl } = yield* HttpClient.get(
40 functionUrl,
41 ).pipe(
42 Effect.flatMap((response) =>
43 response.status === 200
44 ? (response.json as Effect.Effect<{
45 sourceQueueUrl?: string;
46 sinkQueueUrl?: string;
47 }>)
48 : Effect.fail(
49 new FunctionNotReady(
50 `Function not ready: ${response.status}`,
51 ),
52 ),
53 ),
54 Effect.flatMap((body) =>
55 body.sourceQueueUrl && body.sinkQueueUrl
56 ? Effect.succeed(
57 body as { sourceQueueUrl: string; sinkQueueUrl: string },
58 )
59 : Effect.fail(
60 new FunctionNotReady(
61 "Function returned empty queue identifiers",
62 ),
63 ),
64 ),
65 Effect.retry({
66 schedule: Schedule.max([
67 Schedule.fixed("1 seconds"),
68 Schedule.recurs(45),
69 ]),
70 }),
71 );
72
73 // The Pipe provider waits for RUNNING inside reconcile, so the pipe
74 // is already active. Send a message to the source queue out-of-band;
75 // the pipe delivers it to the Lambda target, whose listener forwards
76 // the body into the sink queue.
77 const messageBody = `pipe-delivery-${crypto.randomUUID()}`;
78 yield* SQS.sendMessage({
79 QueueUrl: sourceQueueUrl,
80 MessageBody: messageBody,
81 });
82
83 // Poll the sink queue until the forwarded body shows up. Nine 8s
84 // long polls with 500ms spacing give delivery a bounded ~76s budget
85 // without spending half the time asleep between polls.
86 const received = yield* Effect.gen(function* () {
87 const result = yield* SQS.receiveMessage({
88 QueueUrl: sinkQueueUrl,
89 MaxNumberOfMessages: 10,
90 WaitTimeSeconds: 8,
91 });
92 const match = (result.Messages ?? []).find(
93 (message) => message.Body === messageBody,
94 );
95 if (!match?.ReceiptHandle) {
96 return yield* Effect.fail(new MessageNotDelivered());
97 }
98 yield* SQS.deleteMessage({
99 QueueUrl: sinkQueueUrl,
100 ReceiptHandle: match.ReceiptHandle,
101 });
102 return match.Body!;
103 }).pipe(
104 Effect.retry({
105 while: (error) => error._tag === "MessageNotDelivered",
106 schedule: Schedule.max([
107 Schedule.spaced("500 millis"),
108 Schedule.recurs(8),
109 ]),
110 }),
111 );
112
113 expect(received).toEqual(messageBody);
114
115 yield* stack.destroy();
116 }),
117 { timeout: 240_000 },
118 );
119 },
120);
121
122class MessageNotDelivered extends Data.TaggedError("MessageNotDelivered") {}
123
124class FunctionNotReady extends Data.TaggedError("FunctionNotReady")<{
125 message: string;
126}> {
127 constructor(message: string) {
128 super({ message });
129 }
130}