1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import SFNTestFunctionLive, { SFNTestFunction } from "./handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "SFNBindings");
16
17// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
18// over 60s on a fresh deploy under parallel-suite load.
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(75),
22]);
23
24let baseUrl: string;
25let functionArn: string;
26
27class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
28 readonly status: number;
29 readonly body: string;
30}> {}
31
32// Retry transient 5xx only; genuine 4xx/assertion failures surface
33// immediately.
34const send = (request: HttpClientRequest.HttpClientRequest) =>
35 HttpClient.execute(request).pipe(
36 Effect.flatMap((response) =>
37 response.status >= 500
38 ? response.text.pipe(
39 Effect.flatMap((body) =>
40 Effect.fail(
41 new TransientUpstream({ status: response.status, body }),
42 ),
43 ),
44 )
45 : Effect.succeed(response),
46 ),
47 Effect.retry({
48 while: (e) => e._tag === "TransientUpstream",
49 schedule: Schedule.max([
50 Schedule.exponential("500 millis"),
51 Schedule.recurs(6),
52 ]),
53 }),
54 );
55
56const postJson = (url: string, body: unknown) =>
57 send(
58 HttpClientRequest.post(url).pipe(HttpClientRequest.bodyJsonUnsafe(body)),
59 ).pipe(Effect.flatMap((r) => r.json));
60
61const getJson = (url: string) =>
62 send(HttpClientRequest.get(url)).pipe(Effect.flatMap((r) => r.json));
63
64/** Poll an execution's status via the fixture until it leaves RUNNING. */
65const pollExecution = (route: string, executionArn: string) =>
66 getJson(
67 `${baseUrl}/${route}?executionArn=${encodeURIComponent(executionArn)}`,
68 ).pipe(
69 Effect.repeat({
70 schedule: Schedule.spaced("2 seconds"),
71 until: (r) => (r as { status: string }).status !== "RUNNING",
72 times: 20,
73 }),
74 ) as Effect.Effect<
75 { status: string; output?: string; error?: string },
76 unknown
77 >;
78
79/**
80 * Drain the callback queue until we hold the task token for the given
81 * execution. Tokens for other executions (stale from earlier tests in the
82 * sequential block) are consumed and discarded.
83 */
84const receiveTokenFor = (executionArn: string) =>
85 getJson(`${baseUrl}/receive-token`).pipe(
86 Effect.repeat({
87 schedule: Schedule.spaced("1 second"),
88 until: (r) =>
89 (r as { executionArn?: string }).executionArn === executionArn,
90 times: 15,
91 }),
92 Effect.map((r) => (r as { token: string }).token),
93 );
94
95describe(
96 "StepFunctions Bindings",
97 {
98 tags: [
99 "provider:aws",
100 "provider:aws:lambda",
101 "provider:aws:sqs",
102 "provider:aws:stepfunctions",
103 "live",
104 ],
105 },
106 () => {
107 beforeAll(
108 Effect.gen(function* () {
109 yield* Effect.logInfo("SFN test setup: destroying previous resources");
110 yield* sharedStack.destroy();
111
112 yield* Effect.logInfo("SFN test setup: deploying fixture");
113 const deployed = yield* sharedStack.deploy(
114 Effect.gen(function* () {
115 return yield* SFNTestFunction;
116 }).pipe(Effect.provide(SFNTestFunctionLive)),
117 );
118
119 expect(deployed.functionUrl).toBeTruthy();
120 baseUrl = deployed.functionUrl!.replace(/\/+$/, "");
121 functionArn = deployed.functionArn;
122 const readinessUrl = `${baseUrl}/health`;
123
124 yield* Effect.logInfo(
125 `SFN test setup: probing readiness at ${readinessUrl}`,
126 );
127 yield* HttpClient.get(readinessUrl).pipe(
128 Effect.flatMap((response) =>
129 response.status === 200
130 ? Effect.succeed(response)
131 : Effect.fail(
132 new Error(`Function not ready: ${response.status}`),
133 ),
134 ),
135 Effect.tapError((error) =>
136 Effect.logWarning(
137 `SFN test setup: fixture not ready yet (${String(error)})`,
138 ),
139 ),
140 Effect.retry({ schedule: readinessPolicy }),
141 );
142 }),
143 { timeout: 240_000 },
144 );
145
146 afterAll(sharedStack.destroy(), { timeout: 120_000 });
147
148 describe("StartSyncExecution", () => {
149 test.provider(
150 "runs an EXPRESS workflow synchronously and returns its output",
151 (_stack) =>
152 Effect.gen(function* () {
153 const response = (yield* postJson(`${baseUrl}/start-sync`, {
154 input: JSON.stringify({ value: 42 }),
155 })) as {
156 executionArn: string;
157 status: string;
158 output?: string;
159 error?: string;
160 cause?: string;
161 };
162
163 expect(response.status).toBe("SUCCEEDED");
164 expect(response.executionArn).toContain(":express:");
165 const output = JSON.parse(response.output!);
166 expect(output.greeting).toBe("hello");
167 expect(output.echo.value).toBe(42);
168 }),
169 { timeout: 60_000 },
170 );
171 });
172
173 describe("StartExecution", () => {
174 test.provider(
175 "starts a STANDARD execution asynchronously",
176 (_stack) =>
177 Effect.gen(function* () {
178 const response = (yield* postJson(`${baseUrl}/start`, {
179 input: JSON.stringify({ order: 1 }),
180 })) as { executionArn: string };
181
182 expect(response.executionArn).toContain(":execution:");
183 }),
184 { timeout: 60_000 },
185 );
186 });
187
188 describe("DescribeExecution", () => {
189 test.provider(
190 "observes a STANDARD execution completing",
191 (_stack) =>
192 Effect.gen(function* () {
193 const started = (yield* postJson(`${baseUrl}/start`, {
194 input: JSON.stringify({ order: 2 }),
195 })) as { executionArn: string };
196
197 const final = yield* pollExecution(
198 "describe-standard",
199 started.executionArn,
200 );
201 expect(final.status).toBe("SUCCEEDED");
202 expect(JSON.parse(final.output!)).toEqual({ done: true });
203 }),
204 { timeout: 90_000 },
205 );
206 });
207
208 describe("ValidateStateMachineDefinition", () => {
209 test.provider(
210 "returns OK for a valid definition and FAIL with diagnostics for an invalid one",
211 (_stack) =>
212 Effect.gen(function* () {
213 const ok = (yield* postJson(`${baseUrl}/validate-definition`, {
214 definition: JSON.stringify({
215 StartAt: "Done",
216 States: { Done: { Type: "Pass", End: true } },
217 }),
218 })) as { result: string; diagnostics: unknown[] };
219 expect(ok.result).toBe("OK");
220 expect(ok.diagnostics).toEqual([]);
221
222 const fail = (yield* postJson(`${baseUrl}/validate-definition`, {
223 definition: JSON.stringify({
224 StartAt: "Missing",
225 States: { Done: { Type: "Pass", End: true } },
226 }),
227 })) as { result: string; diagnostics: { code: string }[] };
228 expect(fail.result).toBe("FAIL");
229 expect(fail.diagnostics.length).toBeGreaterThan(0);
230 }),
231 { timeout: 60_000 },
232 );
233 });
234
235 describe("TestState", () => {
236 test.provider(
237 "executes a single JSONata Pass state against an input",
238 (_stack) =>
239 Effect.gen(function* () {
240 const response = (yield* postJson(`${baseUrl}/test-state`, {
241 definition: JSON.stringify({
242 Type: "Pass",
243 QueryLanguage: "JSONata",
244 Output: "{% $states.input.value * 2 %}",
245 End: true,
246 }),
247 input: JSON.stringify({ value: 21 }),
248 })) as { status: string; output?: string };
249 expect(response.status).toBe("SUCCEEDED");
250 expect(response.output).toBe("42");
251 }),
252 { timeout: 60_000 },
253 );
254 });
255
256 // The callback flows share ONE SQS queue carrying task tokens — receives
257 // steal each other's messages under concurrency, so run them sequentially
258 // and drain only the matching token per test.
259 describe.sequential("callback pattern", () => {
260 describe("SendTaskSuccess", () => {
261 test.provider(
262 "heartbeats then completes a waiting execution",
263 (_stack) =>
264 Effect.gen(function* () {
265 const started = (yield* postJson(
266 `${baseUrl}/start-callback`,
267 {},
268 )) as { executionArn: string };
269
270 const token = yield* receiveTokenFor(started.executionArn);
271 expect(token).toBeTruthy();
272
273 // SendTaskHeartbeat keeps the task alive without completing it
274 const heartbeat = (yield* postJson(`${baseUrl}/task-heartbeat`, {
275 token,
276 })) as { sent: boolean };
277 expect(heartbeat.sent).toBe(true);
278
279 const success = (yield* postJson(`${baseUrl}/task-success`, {
280 token,
281 output: JSON.stringify({ approved: true }),
282 })) as { sent: boolean };
283 expect(success.sent).toBe(true);
284
285 const final = yield* pollExecution(
286 "describe-callback",
287 started.executionArn,
288 );
289 expect(final.status).toBe("SUCCEEDED");
290 expect(JSON.parse(final.output!)).toEqual({ approved: true });
291 }),
292 { timeout: 120_000 },
293 );
294 });
295
296 describe("SendTaskFailure", () => {
297 test.provider(
298 "fails a waiting execution with a typed error",
299 (_stack) =>
300 Effect.gen(function* () {
301 const started = (yield* postJson(
302 `${baseUrl}/start-callback`,
303 {},
304 )) as { executionArn: string };
305
306 const token = yield* receiveTokenFor(started.executionArn);
307
308 const failed = (yield* postJson(`${baseUrl}/task-failure`, {
309 token,
310 error: "ApprovalRejected",
311 cause: "rejected by integration test",
312 })) as { sent: boolean };
313 expect(failed.sent).toBe(true);
314
315 const final = yield* pollExecution(
316 "describe-callback",
317 started.executionArn,
318 );
319 expect(final.status).toBe("FAILED");
320 expect(final.error).toBe("ApprovalRejected");
321 }),
322 { timeout: 120_000 },
323 );
324 });
325
326 describe("StopExecution", () => {
327 test.provider(
328 "aborts a running execution",
329 (_stack) =>
330 Effect.gen(function* () {
331 const started = (yield* postJson(
332 `${baseUrl}/start-callback`,
333 {},
334 )) as { executionArn: string };
335
336 // consume this execution's token first so it never pollutes the
337 // queue for other flows
338 yield* receiveTokenFor(started.executionArn);
339
340 const stopped = (yield* postJson(`${baseUrl}/stop`, {
341 executionArn: started.executionArn,
342 })) as { stopDate: string };
343 expect(stopped.stopDate).toBeTruthy();
344
345 const final = yield* pollExecution(
346 "describe-callback",
347 started.executionArn,
348 );
349 expect(final.status).toBe("ABORTED");
350 }),
351 { timeout: 120_000 },
352 );
353 });
354
355 describe("SendTaskHeartbeat", () => {
356 test.provider(
357 "rejects an invalid task token with a typed error surfaced as 500",
358 (_stack) =>
359 Effect.gen(function* () {
360 // an obviously-invalid token dies in the fixture (orDie) — the
361 // route answers 500; we only assert the request round-trips and
362 // does not hang (IAM wiring for states:SendTaskHeartbeat works,
363 // otherwise this would be AccessDenied at a different layer).
364 const response = yield* HttpClient.execute(
365 HttpClientRequest.post(`${baseUrl}/task-heartbeat`).pipe(
366 HttpClientRequest.bodyJsonUnsafe({ token: "not-a-token" }),
367 ),
368 );
369 expect(response.status).toBeGreaterThanOrEqual(400);
370 }),
371 { timeout: 60_000 },
372 );
373 });
374
375 describe("GetActivityTask", () => {
376 test.provider(
377 "polls a scheduled activity task and completes it",
378 (_stack) =>
379 Effect.gen(function* () {
380 const started = (yield* postJson(
381 `${baseUrl}/start-activity`,
382 {},
383 )) as { executionArn: string };
384
385 // GetActivityTask long-polls; repeat until the scheduled task's
386 // token arrives.
387 const task = yield* getJson(`${baseUrl}/activity-task`).pipe(
388 Effect.repeat({
389 schedule: Schedule.spaced("2 seconds"),
390 until: (r): boolean =>
391 (r as { taskToken: string | null }).taskToken !== null,
392 times: 8,
393 }),
394 );
395 const token = (task as { taskToken: string }).taskToken;
396 expect(token).toBeTruthy();
397
398 const success = (yield* postJson(`${baseUrl}/task-success`, {
399 token,
400 output: JSON.stringify({ workedBy: "activity-worker" }),
401 })) as { sent: boolean };
402 expect(success.sent).toBe(true);
403
404 const final = yield* pollExecution(
405 "describe-activity",
406 started.executionArn,
407 );
408 expect(final.status).toBe("SUCCEEDED");
409 expect(JSON.parse(final.output!)).toEqual({
410 workedBy: "activity-worker",
411 });
412 }),
413 { timeout: 120_000 },
414 );
415 });
416 });
417
418 describe("ListExecutions + GetExecutionHistory + RedriveExecution", () => {
419 test.provider(
420 "lists executions, reads history, and rejects redrive of a succeeded execution",
421 (_stack) =>
422 Effect.gen(function* () {
423 const started = (yield* postJson(`${baseUrl}/start`, {
424 input: JSON.stringify({ order: 4 }),
425 })) as { executionArn: string };
426
427 const final = yield* pollExecution(
428 "describe-standard",
429 started.executionArn,
430 );
431 expect(final.status).toBe("SUCCEEDED");
432
433 const list = (yield* getJson(`${baseUrl}/list-executions`)) as {
434 count: number;
435 executionArns: string[];
436 };
437 expect(list.count).toBeGreaterThanOrEqual(1);
438 expect(list.executionArns).toContain(started.executionArn);
439
440 const history = (yield* getJson(
441 `${baseUrl}/history?executionArn=${encodeURIComponent(started.executionArn)}`,
442 )) as { count: number; lastType: string | null };
443 expect(history.count).toBeGreaterThan(0);
444 expect(history.lastType).toBe("ExecutionSucceeded");
445
446 // a SUCCEEDED execution is not redrivable — the typed
447 // ExecutionNotRedrivable tag proves the IAM grant and error union
448 const redrive = (yield* postJson(`${baseUrl}/redrive`, {
449 executionArn: started.executionArn,
450 })) as { redriven: boolean; error?: string };
451 expect(redrive.redriven).toBe(false);
452 expect(redrive.error).toBe("ExecutionNotRedrivable");
453 }),
454 { timeout: 90_000 },
455 );
456 });
457
458 describe("Distributed Map Runs", () => {
459 test.provider(
460 "ListMapRuns finds the run; DescribeMapRun and UpdateMapRun round-trip",
461 (_stack) =>
462 Effect.gen(function* () {
463 const started = (yield* postJson(`${baseUrl}/start-map`, {})) as {
464 executionArn: string;
465 };
466
467 const final = yield* pollExecution(
468 "describe-map",
469 started.executionArn,
470 );
471 expect(final.status).toBe("SUCCEEDED");
472
473 const runs = (yield* getJson(
474 `${baseUrl}/map-runs?executionArn=${encodeURIComponent(started.executionArn)}`,
475 )) as { mapRunArns: string[] };
476 expect(runs.mapRunArns.length).toBe(1);
477 const mapRunArn = runs.mapRunArns[0]!;
478
479 const mapRun = (yield* getJson(
480 `${baseUrl}/describe-map-run?mapRunArn=${encodeURIComponent(mapRunArn)}`,
481 )) as { status: string; succeeded?: number };
482 expect(mapRun.status).toBe("SUCCEEDED");
483 expect(mapRun.succeeded).toBe(2);
484
485 // UpdateMapRun targets in-progress runs; on a completed run the
486 // API answers with the typed ValidationException — either result
487 // proves the grant + typed union.
488 const updated = (yield* postJson(`${baseUrl}/update-map-run`, {
489 mapRunArn,
490 maxConcurrency: 2,
491 })) as { updated: boolean; error?: string };
492 expect(
493 updated.updated || updated.error === "ValidationException",
494 ).toBe(true);
495 }),
496 { timeout: 120_000 },
497 );
498 });
499
500 describe(
501 "consumeExecutionEvents",
502 { tags: ["provider:aws:eventbridge"] },
503 () => {
504 test.provider(
505 "the deploy created an EventBridge rule targeting the function",
506 (_stack) =>
507 Effect.gen(function* () {
508 // Out-of-band via distilled: the fixture's consumeExecutionEvents
509 // must have materialized as a rule on the default bus with the
510 // Lambda as target.
511 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
512 TargetArn: functionArn,
513 });
514 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
515 }),
516 );
517 },
518 );
519 },
520);