1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as appflow from "@distilled.cloud/aws/appflow";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
5import * as s3 from "@distilled.cloud/aws/s3";
6import * as sqs from "@distilled.cloud/aws/sqs";
7import { expect } from "alchemy-test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import {
14 AppFlowApiFunction,
15 AppFlowApiFunctionLive,
16 FLOW_BUCKET,
17 FLOW_NAME,
18 FlowBucket,
19 FlowFixture,
20} from "./handler.ts";
21
22const { test } = Test.make({ providers: AWS.providers() });
23
24// The full fixture: bucket + flow + events queue + the API function with all
25// four Flow bindings and the flowEvents consume loop.
26const program = Effect.gen(function* () {
27 const { flow, eventsQueue } = yield* FlowFixture;
28 const fn = yield* AppFlowApiFunction;
29 return { fn, flow, eventsQueue };
30}).pipe(Effect.provide(AppFlowApiFunctionLive));
31
32// Lambda Function URLs cold-start and a fresh role's IAM grants are
33// eventually consistent; retrying on any non-200 lets the FIRST request wait
34// through that window.
35const readinessSchedule = Schedule.max([
36 Schedule.fixed("2 seconds"),
37 Schedule.recurs(10),
38]);
39
40// Function URLs come back with a trailing slash; strip it before joining so
41// pathnames match the fixture's routes.
42const urlOf = (baseUrl: string, path: string) =>
43 `${baseUrl.replace(/\/+$/, "")}${path}`;
44
45const getJson = (baseUrl: string, path: string) =>
46 HttpClient.get(urlOf(baseUrl, path)).pipe(
47 Effect.flatMap((response) =>
48 response.status === 200
49 ? response.json
50 : Effect.fail(new Error(`Request failed: ${response.status}`)),
51 ),
52 Effect.retry({ schedule: readinessSchedule }),
53 );
54
55const postJson = (baseUrl: string, path: string, body: unknown) =>
56 HttpClient.execute(
57 HttpClientRequest.bodyJsonUnsafe(
58 HttpClientRequest.post(urlOf(baseUrl, path)),
59 body,
60 ),
61 ).pipe(
62 Effect.flatMap((response) =>
63 response.status === 200
64 ? response.json
65 : Effect.fail(new Error(`Request failed: ${response.status}`)),
66 ),
67 Effect.retry({ schedule: readinessSchedule }),
68 );
69
70class ExecutionNotFinished extends Data.TaggedError("ExecutionNotFinished")<{
71 lastStatus: string | undefined;
72}> {
73 override get message() {
74 return this.lastStatus === undefined
75 ? "execution record never appeared"
76 : `execution still '${this.lastStatus}' when the poll budget exhausted`;
77 }
78}
79
80/**
81 * Poll the fixture's /executions route until the given run finishes.
82 *
83 * Budget: every observed pass completes the execution well inside 50s of
84 * polling (whole test 104–117s); ~90s is the speed-doctrine ceiling for
85 * async platform waits. If this exhausts again, the error's `lastStatus`
86 * says whether AppFlow was still queueing (`InProgress`) — the signal to
87 * skipIf-gate the execution wait rather than widen the poll further.
88 */
89const waitForExecution = (baseUrl: string, executionId: string) =>
90 getJson(baseUrl, "/executions").pipe(
91 Effect.flatMap((response) => {
92 const record = (
93 response as {
94 flowExecutions: Array<{
95 executionId?: string;
96 executionStatus?: string;
97 }>;
98 }
99 ).flowExecutions.find((r) => r.executionId === executionId);
100 return record?.executionStatus && record.executionStatus !== "InProgress"
101 ? Effect.succeed(record)
102 : Effect.fail(
103 new ExecutionNotFinished({ lastStatus: record?.executionStatus }),
104 );
105 }),
106 Effect.retry({
107 while: (e): boolean => e instanceof ExecutionNotFinished,
108 schedule: Schedule.max([
109 Schedule.fixed("5 seconds"),
110 Schedule.recurs(18),
111 ]),
112 }),
113 );
114
115class EventNotDelivered extends Data.TaggedError("EventNotDelivered") {}
116
117interface DeliveredFlowEvent {
118 detailType: string;
119 detail: { "flow-name"?: string; status?: string; "execution-id"?: string };
120}
121
122/**
123 * Re-publish a run report until the fresh EventBridge rule routes it.
124 *
125 * AppFlow's native run-report delivery is explicitly best-effort, so a live
126 * flow can finish without emitting an event. Publishing the documented event
127 * envelope tests `flowEvents` deterministically while also riding out the
128 * new rule's propagation window (EventBridge never retro-delivers events).
129 */
130const publishUntilFlowEvent = (queueUrl: string, executionId: string) =>
131 Effect.gen(function* () {
132 const published = yield* eventbridge.putEvents({
133 Entries: [
134 {
135 Source: "alchemy.test.appflow",
136 DetailType: "AppFlow End Flow Run Report",
137 Detail: JSON.stringify({
138 "flow-name": FLOW_NAME,
139 "execution-id": executionId,
140 status: "Execution Successful",
141 }),
142 },
143 ],
144 });
145 if ((published.FailedEntryCount ?? 0) !== 0) {
146 return yield* Effect.fail(new EventNotDelivered());
147 }
148
149 const result = yield* sqs.receiveMessage({
150 QueueUrl: queueUrl,
151 MaxNumberOfMessages: 10,
152 WaitTimeSeconds: 2,
153 });
154 const match = (result.Messages ?? [])
155 .flatMap((message) =>
156 message.Body ? [JSON.parse(message.Body) as DeliveredFlowEvent] : [],
157 )
158 .find(
159 (body) =>
160 body.detail?.["flow-name"] === FLOW_NAME &&
161 body.detail?.["execution-id"] === executionId,
162 );
163 if (!match) {
164 return yield* Effect.fail(new EventNotDelivered());
165 }
166 return match;
167 }).pipe(
168 Effect.retry({
169 while: (e): boolean => e._tag === "EventNotDelivered",
170 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(8)]),
171 }),
172 );
173
174const assertFlowGone = (flowName: string) =>
175 Effect.gen(function* () {
176 const result = yield* appflow.describeFlow({ flowName }).pipe(
177 Effect.map(() => "present" as const),
178 Effect.catchTag("ResourceNotFoundException", () =>
179 Effect.succeed("gone" as const),
180 ),
181 );
182 if (result === "present") {
183 return yield* Effect.fail(new Error(`Flow '${flowName}' still exists`));
184 }
185 }).pipe(
186 Effect.retry({
187 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(8)]),
188 }),
189 );
190
191// One sequential lifecycle: the S3-to-S3 path is credential-free, but the
192// source prefix must be seeded BEFORE the flow exists, so the bucket deploys
193// first, the object lands out-of-band, then the full fixture deploys.
194test.provider(
195 "StartFlow / DescribeFlowExecutionRecords / StopFlow / CancelFlowExecutions / flowEvents",
196 (stack) =>
197 Effect.gen(function* () {
198 yield* stack.destroy();
199
200 // Phase 1: the bucket alone; seed the source prefix.
201 yield* stack.deploy(FlowBucket);
202 yield* s3.putObject({
203 Bucket: FLOW_BUCKET,
204 Key: "input/data.csv",
205 Body: new TextEncoder().encode("id,name\n1,alpha\n2,beta\n"),
206 ContentType: "text/csv",
207 });
208
209 // Phase 2: flow + queue + function with all bindings.
210 const outputs = yield* stack.deploy(program);
211 const url = outputs.fn.functionUrl!;
212 const queueUrl = outputs.eventsQueue.queueUrl;
213 expect(outputs.flow.flowName).toBe(FLOW_NAME);
214 expect(url).toBeDefined();
215
216 yield* getJson(url, "/ready");
217
218 // StartFlow — trigger an on-demand run.
219 const started = (yield* postJson(url, "/start", {})) as {
220 executionId?: string;
221 flowArn?: string;
222 };
223 expect(started.executionId).toBeTruthy();
224 expect(started.flowArn).toContain(":appflow:");
225
226 // StopFlow — on-demand flows cannot be stopped; the binding surfaces
227 // the typed tag.
228 const stopped = (yield* postJson(url, "/stop", {})) as {
229 ok: boolean;
230 error?: string;
231 };
232 expect(stopped.ok).toBe(false);
233 expect(stopped.error).toBe("UnsupportedOperationException");
234
235 // DescribeFlowExecutionRecords — poll until the run leaves InProgress.
236 const record = yield* waitForExecution(url, started.executionId!);
237 expect(record.executionId).toBe(started.executionId);
238 expect(record.executionStatus).toBeTruthy();
239
240 // flowEvents — AppFlow's native delivery is best-effort. Publish the
241 // documented run-report envelope until the fresh rule is active, then
242 // verify the consume loop forwards it into the SQS sink.
243 const event = yield* publishUntilFlowEvent(
244 queueUrl,
245 started.executionId!,
246 );
247 expect(event.detail["flow-name"]).toBe(FLOW_NAME);
248 expect(event.detailType).toContain("Flow Run Report");
249
250 // CancelFlowExecutions — start a second run and cancel it right away.
251 // The run may already be finishing, so only the response shape is
252 // asserted (finished runs land in invalidExecutions).
253 const second = (yield* postJson(url, "/start", {})) as {
254 executionId?: string;
255 };
256 const canceled = (yield* postJson(url, "/cancel", {
257 executionIds: [second.executionId],
258 })) as { ok: boolean; invalidExecutions?: string[] };
259 expect(canceled.ok).toBe(true);
260 expect(Array.isArray(canceled.invalidExecutions)).toBe(true);
261
262 // Destroy and verify the flow is gone.
263 yield* stack.destroy();
264 yield* assertFlowGone(FLOW_NAME);
265 }),
266 // Observed pass baseline is 104–117s under full-suite load with the old
267 // ~50s execution poll; the poll is now ~90s, so 210s + the 40s poll delta.
268 // Every constituent readiness poll remains independently bounded.
269 {
270 tags: [
271 "provider:aws",
272 "provider:aws:appflow",
273 "provider:aws:eventbridge",
274 "provider:aws:iam",
275 "provider:aws:lambda",
276 "provider:aws:s3",
277 "provider:aws:sqs",
278 "live",
279 ],
280 timeout: 250_000,
281 retry: 0,
282 },
283);