1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
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 * as HttpClientRequest from "effect/http/HttpClientRequest";
10import PipesBindingsFunctionLive, {
11 PipesBindingsFunction,
12} from "./bindings-handler.ts";
13
14const testOptions = { providers: AWS.providers() };
15const { test, beforeAll, afterAll } = Test.make(testOptions);
16const sharedStack = Core.scratchStack(testOptions, "PipesBindings");
17
18// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
19// over 60s on a fresh deploy.
20const readinessPolicy = Schedule.max([
21 Schedule.fixed("2 seconds"),
22 Schedule.recurs(75),
23]);
24
25let baseUrl: string;
26
27class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
28 readonly status: number;
29 readonly body: string;
30}> {}
31
32class PipeNotSettled extends Data.TaggedError("PipeNotSettled")<{
33 readonly currentState: string | undefined;
34}> {}
35
36// The shared Lambda fixture occasionally answers a transient 5xx under load
37// (cold re-init, IAM propagation on the freshly attached policy). Retry only
38// 5xx; a genuine 4xx/assertion failure surfaces immediately.
39const send = (request: HttpClientRequest.HttpClientRequest) =>
40 HttpClient.execute(request).pipe(
41 Effect.flatMap((response) =>
42 response.status >= 500
43 ? response.text.pipe(
44 Effect.flatMap((body) =>
45 Effect.fail(
46 new TransientUpstream({ status: response.status, body }),
47 ),
48 ),
49 )
50 : Effect.succeed(response),
51 ),
52 Effect.retry({
53 while: (e) => e._tag === "TransientUpstream",
54 schedule: Schedule.max([
55 Schedule.exponential("500 millis"),
56 Schedule.recurs(6),
57 ]),
58 }),
59 );
60
61const getJson = (path: string) =>
62 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
63 Effect.flatMap((r) => r.json),
64 // The response body has no compile-time shape; widen the distilled Json
65 // union to `unknown` so call sites can assert their expected shape.
66 Effect.map((json): unknown => json),
67 );
68
69interface DescribeResponse {
70 name: string | undefined;
71 currentState: string | undefined;
72 desiredState: string | undefined;
73}
74
75// Poll the fixture's /describe route (bounded) until the pipe settles into
76// the expected state after a stop/start.
77const awaitState = (expected: string) =>
78 getJson("/describe").pipe(
79 Effect.flatMap((raw) => {
80 const described = raw as DescribeResponse;
81 return described.currentState === expected
82 ? Effect.succeed(described)
83 : Effect.fail(
84 new PipeNotSettled({ currentState: described.currentState }),
85 );
86 }),
87 Effect.retry({
88 while: (e): boolean => e._tag === "PipeNotSettled",
89 schedule: Schedule.max([
90 Schedule.fixed("3 seconds"),
91 Schedule.recurs(30),
92 ]),
93 }),
94 );
95
96describe.sequential(
97 "AWS.Pipes bindings",
98 {
99 tags: [
100 "provider:aws",
101 "provider:aws:iam",
102 "provider:aws:lambda",
103 "provider:aws:pipes",
104 "provider:aws:sqs",
105 "live",
106 ],
107 },
108 () => {
109 beforeAll(
110 Effect.gen(function* () {
111 yield* Effect.logInfo(
112 "Pipes test setup: destroying previous resources",
113 );
114 yield* sharedStack.destroy();
115
116 yield* Effect.logInfo("Pipes test setup: deploying fixture");
117 const attrs = yield* sharedStack.deploy(
118 Effect.gen(function* () {
119 return yield* PipesBindingsFunction;
120 }).pipe(Effect.provide(PipesBindingsFunctionLive)),
121 );
122
123 expect(attrs.functionUrl).toBeTruthy();
124 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
125
126 const readinessUrl = `${baseUrl}/bindings`;
127 yield* Effect.logInfo(
128 `Pipes test setup: probing readiness at ${readinessUrl}`,
129 );
130 yield* HttpClient.get(readinessUrl).pipe(
131 Effect.flatMap((response) =>
132 response.status === 200
133 ? Effect.succeed(response)
134 : Effect.fail(
135 new Error(`Function not ready: ${response.status}`),
136 ),
137 ),
138 Effect.tapError((error) =>
139 Effect.logWarning(
140 `Pipes test setup: fixture not ready yet (${String(error)})`,
141 ),
142 ),
143 Effect.retry({ schedule: readinessPolicy }),
144 );
145 }),
146 { timeout: 240_000 },
147 );
148
149 afterAll(sharedStack.destroy(), { timeout: 180_000 });
150
151 describe("binding registration", () => {
152 test.provider(
153 "all four capabilities initialize in the runtime",
154 (_stack) =>
155 Effect.gen(function* () {
156 const response = (yield* getJson("/bindings")) as {
157 bound: string[];
158 };
159 expect(response.bound).toHaveLength(4);
160 }),
161 );
162 });
163
164 describe("DescribePipe", () => {
165 test.provider("reads the pipe's live state", (_stack) =>
166 Effect.gen(function* () {
167 const described = (yield* getJson("/describe")) as DescribeResponse;
168 expect(described.name).toBeTruthy();
169 // The provider waits for the pipe to settle out of CREATING before
170 // returning, so the fixture pipe is already RUNNING.
171 expect(described.currentState).toBe("RUNNING");
172 expect(described.desiredState).toBe("RUNNING");
173 }),
174 );
175 });
176
177 describe("ListPipes", () => {
178 test.provider("finds the pipe by name prefix", (_stack) =>
179 Effect.gen(function* () {
180 const described = (yield* getJson("/describe")) as DescribeResponse;
181 const { names } = (yield* getJson("/list")) as { names: string[] };
182 expect(names).toContain(described.name);
183 }),
184 );
185 });
186
187 describe("StopPipe + StartPipe", () => {
188 test.provider(
189 "stops the running pipe, then starts it again",
190 (_stack) =>
191 Effect.gen(function* () {
192 const stopped = (yield* getJson("/stop")) as {
193 desiredState: string;
194 currentState: string;
195 };
196 expect(stopped.desiredState).toBe("STOPPED");
197
198 // Wait (bounded) for the pipe to settle — StartPipe during
199 // STOPPING would be a ConflictException.
200 yield* awaitState("STOPPED");
201
202 const started = (yield* getJson("/start")) as {
203 desiredState: string;
204 currentState: string;
205 };
206 expect(started.desiredState).toBe("RUNNING");
207
208 yield* awaitState("RUNNING");
209 }),
210 { timeout: 240_000 },
211 );
212 });
213 },
214);