1import * as Lambda from "@/AWS/Lambda";
2import * as SQS from "@/AWS/SQS";
3import * as StepFunctions from "@/AWS/StepFunctions";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Redacted from "effect/Redacted";
8import * as Stream from "effect/Stream";
9import { HttpServerRequest } from "effect/http/HttpServerRequest";
10import * as HttpServerResponse from "effect/http/HttpServerResponse";
11import path from "pathe";
12
13const main = path.resolve(import.meta.dirname, "handler.ts");
14
15const plain = (
16 value: string | Redacted.Redacted<string> | undefined,
17): string | undefined =>
18 value === undefined
19 ? undefined
20 : typeof value === "string"
21 ? value
22 : Redacted.value(value);
23
24export class SFNTestFunction extends Lambda.Function<Lambda.Function>()(
25 "SFNTestFunction",
26) {}
27
28export default SFNTestFunction.make(
29 {
30 main,
31 functionUrl: true,
32 // routes long-poll SQS/GetActivityTask (up to 60s when no task is
33 // scheduled) and round-trip whole workflows — AWS's 3s default
34 // intermittently times out under cold starts
35 timeout: Duration.seconds(75),
36 },
37 Effect.gen(function* () {
38 // EXPRESS workflow for synchronous invocation
39 const express = yield* StepFunctions.StateMachine("TestExpressMachine", {
40 type: "EXPRESS",
41 definition: {
42 StartAt: "Echo",
43 States: {
44 Echo: {
45 Type: "Pass",
46 Parameters: { "echo.$": "$", greeting: "hello" },
47 End: true,
48 },
49 },
50 },
51 });
52
53 // STANDARD workflow for async start + describe polling
54 const standard = yield* StepFunctions.StateMachine("TestStandardMachine", {
55 definition: {
56 StartAt: "Done",
57 States: {
58 Done: { Type: "Pass", Result: { done: true }, End: true },
59 },
60 },
61 });
62
63 // Callback-pattern workflow: parks on .waitForTaskToken after sending
64 // the task token to an SQS queue.
65 const callbackQueue = yield* SQS.Queue("CallbackQueue");
66 const callback = yield* StepFunctions.StateMachine("TestCallbackMachine", {
67 definition: {
68 StartAt: "WaitForCallback",
69 States: {
70 WaitForCallback: {
71 Type: "Task",
72 Resource: "arn:aws:states:::sqs:sendMessage.waitForTaskToken",
73 Parameters: {
74 QueueUrl: callbackQueue.queueUrl,
75 MessageBody: {
76 "token.$": "$$.Task.Token",
77 "executionArn.$": "$$.Execution.Id",
78 },
79 },
80 TimeoutSeconds: 300,
81 End: true,
82 },
83 },
84 },
85 policyStatements: [
86 {
87 Effect: "Allow",
88 Action: ["sqs:SendMessage"],
89 Resource: [callbackQueue.queueArn],
90 },
91 ],
92 });
93
94 // Activity-worker workflow: a Task state that parks on an Activity ARN
95 // until a worker polls it with GetActivityTask and completes it.
96 const activity = yield* StepFunctions.Activity("PollActivity", {});
97 const activityMachine = yield* StepFunctions.StateMachine(
98 "TestActivityMachine",
99 {
100 definition: {
101 StartAt: "WaitForWorker",
102 States: {
103 WaitForWorker: {
104 Type: "Task",
105 Resource: activity.activityArn,
106 TimeoutSeconds: 300,
107 End: true,
108 },
109 },
110 },
111 },
112 );
113
114 // Distributed Map workflow: produces a Map Run per execution for the
115 // ListMapRuns/DescribeMapRun/UpdateMapRun bindings.
116 const mapMachine = yield* StepFunctions.StateMachine("TestMapMachine", {
117 definition: {
118 StartAt: "MapItems",
119 States: {
120 MapItems: {
121 Type: "Map",
122 ItemsPath: "$.items",
123 MaxConcurrency: 1,
124 ItemProcessor: {
125 ProcessorConfig: {
126 Mode: "DISTRIBUTED",
127 ExecutionType: "EXPRESS",
128 },
129 StartAt: "PassItem",
130 States: { PassItem: { Type: "Pass", End: true } },
131 },
132 End: true,
133 },
134 },
135 },
136 // A Distributed Map's execution role starts (and observes) child
137 // EXPRESS executions of the machine itself — self-referential, so
138 // grant service-wide.
139 policyStatements: [
140 {
141 Effect: "Allow",
142 Action: [
143 "states:StartExecution",
144 "states:DescribeExecution",
145 "states:StopExecution",
146 ],
147 Resource: ["*"],
148 },
149 ],
150 });
151
152 // Event source: subscribe the host to this machine's execution status
153 // changes. The deploy proves the EventBridge rule + invoke permission
154 // wiring (and that Output-valued ARNs resolve inside the pattern).
155 yield* StepFunctions.consumeExecutionEvents(
156 { stateMachines: [standard], statuses: ["SUCCEEDED", "FAILED"] },
157 (events) =>
158 Stream.runForEach(events, (event) =>
159 Effect.log(
160 `execution ${event.detail.executionArn}: ${event.detail.status}`,
161 ),
162 ),
163 );
164
165 const startSyncExecution = yield* StepFunctions.StartSyncExecution(express);
166 const startExecution = yield* StepFunctions.StartExecution(standard);
167 const startCallbackExecution =
168 yield* StepFunctions.StartExecution(callback);
169 const describeStandardExecution =
170 yield* StepFunctions.DescribeExecution(standard);
171 const describeCallbackExecution =
172 yield* StepFunctions.DescribeExecution(callback);
173 const stopExecution = yield* StepFunctions.StopExecution(callback);
174 const sendTaskSuccess = yield* StepFunctions.SendTaskSuccess();
175 const sendTaskFailure = yield* StepFunctions.SendTaskFailure();
176 const sendTaskHeartbeat = yield* StepFunctions.SendTaskHeartbeat();
177 const validateDefinition =
178 yield* StepFunctions.ValidateStateMachineDefinition();
179 const testState = yield* StepFunctions.TestState();
180 const receiveMessage = yield* SQS.ReceiveMessage(callbackQueue);
181 const deleteMessage = yield* SQS.DeleteMessage(callbackQueue);
182
183 const startActivityExecution =
184 yield* StepFunctions.StartExecution(activityMachine);
185 const describeActivityExecution =
186 yield* StepFunctions.DescribeExecution(activityMachine);
187 const getActivityTask = yield* StepFunctions.GetActivityTask(activity);
188 const listExecutions = yield* StepFunctions.ListExecutions(standard);
189 const getExecutionHistory =
190 yield* StepFunctions.GetExecutionHistory(standard);
191 const redriveExecution = yield* StepFunctions.RedriveExecution(standard);
192 const startMapExecution = yield* StepFunctions.StartExecution(mapMachine);
193 const describeMapExecution =
194 yield* StepFunctions.DescribeExecution(mapMachine);
195 const listMapRuns = yield* StepFunctions.ListMapRuns(mapMachine);
196 const describeMapRun = yield* StepFunctions.DescribeMapRun(mapMachine);
197 const updateMapRun = yield* StepFunctions.UpdateMapRun(mapMachine);
198
199 return {
200 fetch: Effect.gen(function* () {
201 const request = yield* HttpServerRequest;
202 const url = new URL(request.originalUrl);
203 const pathname = url.pathname;
204
205 if (request.method === "GET" && pathname === "/health") {
206 return HttpServerResponse.text("ok");
207 }
208
209 if (request.method === "POST" && pathname === "/start-sync") {
210 const body = (yield* request.json) as unknown as { input?: string };
211 const result = yield* startSyncExecution({ input: body.input });
212 return yield* HttpServerResponse.json({
213 executionArn: result.executionArn,
214 status: result.status,
215 output: plain(result.output),
216 error: plain(result.error),
217 cause: plain(result.cause),
218 });
219 }
220
221 if (request.method === "POST" && pathname === "/start") {
222 const body = (yield* request.json) as unknown as { input?: string };
223 const result = yield* startExecution({ input: body.input });
224 return yield* HttpServerResponse.json({
225 executionArn: result.executionArn,
226 });
227 }
228
229 if (request.method === "POST" && pathname === "/start-callback") {
230 const result = yield* startCallbackExecution();
231 return yield* HttpServerResponse.json({
232 executionArn: result.executionArn,
233 });
234 }
235
236 if (request.method === "GET" && pathname === "/describe-standard") {
237 const executionArn = url.searchParams.get("executionArn")!;
238 const result = yield* describeStandardExecution({ executionArn });
239 return yield* HttpServerResponse.json({
240 status: result.status,
241 output: plain(result.output),
242 });
243 }
244
245 if (request.method === "GET" && pathname === "/describe-callback") {
246 const executionArn = url.searchParams.get("executionArn")!;
247 const result = yield* describeCallbackExecution({ executionArn });
248 return yield* HttpServerResponse.json({
249 status: result.status,
250 output: plain(result.output),
251 error: plain(result.error),
252 });
253 }
254
255 if (request.method === "GET" && pathname === "/receive-token") {
256 const result = yield* receiveMessage({
257 MaxNumberOfMessages: 1,
258 WaitTimeSeconds: 5,
259 // resurface quickly if a failed invocation consumed the message
260 VisibilityTimeout: 15,
261 });
262 const message = result.Messages?.[0];
263 if (!message?.Body) {
264 return yield* HttpServerResponse.json({ token: null });
265 }
266 yield* deleteMessage({ ReceiptHandle: message.ReceiptHandle! });
267 const parsed = JSON.parse(message.Body) as {
268 token: string;
269 executionArn: string;
270 };
271 return yield* HttpServerResponse.json({
272 token: parsed.token,
273 executionArn: parsed.executionArn,
274 });
275 }
276
277 if (request.method === "POST" && pathname === "/task-success") {
278 const body = (yield* request.json) as unknown as {
279 token: string;
280 output: string;
281 };
282 yield* sendTaskSuccess({
283 taskToken: body.token,
284 output: body.output,
285 });
286 return yield* HttpServerResponse.json({ sent: true });
287 }
288
289 if (request.method === "POST" && pathname === "/task-failure") {
290 const body = (yield* request.json) as unknown as {
291 token: string;
292 error?: string;
293 cause?: string;
294 };
295 yield* sendTaskFailure({
296 taskToken: body.token,
297 error: body.error,
298 cause: body.cause,
299 });
300 return yield* HttpServerResponse.json({ sent: true });
301 }
302
303 if (request.method === "POST" && pathname === "/task-heartbeat") {
304 const body = (yield* request.json) as unknown as { token: string };
305 yield* sendTaskHeartbeat({ taskToken: body.token });
306 return yield* HttpServerResponse.json({ sent: true });
307 }
308
309 if (request.method === "POST" && pathname === "/validate-definition") {
310 const body = (yield* request.json) as unknown as {
311 definition: string;
312 type?: "STANDARD" | "EXPRESS";
313 };
314 const report = yield* validateDefinition({
315 definition: body.definition,
316 type: body.type,
317 severity: "ERROR",
318 });
319 return yield* HttpServerResponse.json({
320 result: report.result,
321 diagnostics: report.diagnostics.map((diagnostic) => ({
322 severity: diagnostic.severity,
323 code: plain(diagnostic.code),
324 message: plain(diagnostic.message),
325 })),
326 });
327 }
328
329 if (request.method === "POST" && pathname === "/test-state") {
330 const body = (yield* request.json) as unknown as {
331 definition: string;
332 input?: string;
333 };
334 const result = yield* testState({
335 definition: body.definition,
336 input: body.input,
337 });
338 return yield* HttpServerResponse.json({
339 status: result.status,
340 output: plain(result.output),
341 error: plain(result.error),
342 });
343 }
344
345 if (request.method === "POST" && pathname === "/stop") {
346 const body = (yield* request.json) as unknown as {
347 executionArn: string;
348 };
349 const result = yield* stopExecution({
350 executionArn: body.executionArn,
351 error: "TestStop",
352 cause: "stopped by integration test",
353 });
354 return yield* HttpServerResponse.json({
355 stopDate: result.stopDate.toISOString(),
356 });
357 }
358
359 if (request.method === "POST" && pathname === "/start-activity") {
360 const result = yield* startActivityExecution();
361 return yield* HttpServerResponse.json({
362 executionArn: result.executionArn,
363 });
364 }
365
366 if (request.method === "GET" && pathname === "/activity-task") {
367 const task = yield* getActivityTask({
368 workerName: "sfn-bindings-test",
369 });
370 return yield* HttpServerResponse.json({
371 taskToken: task.taskToken ?? null,
372 input: plain(task.input) ?? null,
373 });
374 }
375
376 if (request.method === "GET" && pathname === "/describe-activity") {
377 const executionArn = url.searchParams.get("executionArn")!;
378 const result = yield* describeActivityExecution({ executionArn });
379 return yield* HttpServerResponse.json({
380 status: result.status,
381 output: plain(result.output),
382 });
383 }
384
385 if (request.method === "GET" && pathname === "/list-executions") {
386 const statusFilter = url.searchParams.get("status") ?? undefined;
387 const result = yield* listExecutions({
388 statusFilter: statusFilter as "SUCCEEDED" | undefined,
389 });
390 return yield* HttpServerResponse.json({
391 count: result.executions.length,
392 executionArns: result.executions.map((e) => e.executionArn),
393 });
394 }
395
396 if (request.method === "GET" && pathname === "/history") {
397 const executionArn = url.searchParams.get("executionArn")!;
398 const result = yield* getExecutionHistory({
399 executionArn,
400 reverseOrder: true,
401 maxResults: 20,
402 });
403 return yield* HttpServerResponse.json({
404 count: result.events.length,
405 lastType: result.events[0]?.type ?? null,
406 });
407 }
408
409 if (request.method === "POST" && pathname === "/redrive") {
410 const body = (yield* request.json) as unknown as {
411 executionArn: string;
412 };
413 const result = yield* redriveExecution({
414 executionArn: body.executionArn,
415 }).pipe(
416 Effect.map(() => ({ redriven: true as const })),
417 Effect.catchTag("ExecutionNotRedrivable", () =>
418 Effect.succeed({
419 redriven: false as const,
420 error: "ExecutionNotRedrivable",
421 }),
422 ),
423 );
424 return yield* HttpServerResponse.json(result);
425 }
426
427 if (request.method === "POST" && pathname === "/start-map") {
428 const result = yield* startMapExecution({
429 input: JSON.stringify({ items: [1, 2] }),
430 });
431 return yield* HttpServerResponse.json({
432 executionArn: result.executionArn,
433 });
434 }
435
436 if (request.method === "GET" && pathname === "/describe-map") {
437 const executionArn = url.searchParams.get("executionArn")!;
438 const result = yield* describeMapExecution({ executionArn });
439 return yield* HttpServerResponse.json({ status: result.status });
440 }
441
442 if (request.method === "GET" && pathname === "/map-runs") {
443 const executionArn = url.searchParams.get("executionArn")!;
444 const result = yield* listMapRuns({ executionArn });
445 return yield* HttpServerResponse.json({
446 mapRunArns: result.mapRuns.map((run) => run.mapRunArn),
447 });
448 }
449
450 if (request.method === "GET" && pathname === "/describe-map-run") {
451 const mapRunArn = url.searchParams.get("mapRunArn")!;
452 const result = yield* describeMapRun({ mapRunArn });
453 return yield* HttpServerResponse.json({
454 status: result.status,
455 maxConcurrency: result.maxConcurrency,
456 succeeded: result.itemCounts?.succeeded,
457 });
458 }
459
460 if (request.method === "POST" && pathname === "/update-map-run") {
461 const body = (yield* request.json) as unknown as {
462 mapRunArn: string;
463 maxConcurrency?: number;
464 };
465 // UpdateMapRun targets in-progress Map Runs; a completed one
466 // answers with the typed ValidationException — either way the
467 // IAM grant and typed union are exercised.
468 const result = yield* updateMapRun({
469 mapRunArn: body.mapRunArn,
470 maxConcurrency: body.maxConcurrency ?? 2,
471 }).pipe(
472 Effect.map(() => ({ updated: true as const })),
473 Effect.catchTag("ValidationException", () =>
474 Effect.succeed({
475 updated: false as const,
476 error: "ValidationException",
477 }),
478 ),
479 );
480 return yield* HttpServerResponse.json(result);
481 }
482
483 return yield* HttpServerResponse.json(
484 { error: "Not found", method: request.method, pathname },
485 { status: 404 },
486 );
487 }).pipe(Effect.orDie),
488 };
489 }).pipe(
490 Effect.provide(
491 Layer.mergeAll(
492 StepFunctions.StartExecutionHttp,
493 StepFunctions.StartSyncExecutionHttp,
494 StepFunctions.DescribeExecutionHttp,
495 StepFunctions.StopExecutionHttp,
496 StepFunctions.SendTaskSuccessHttp,
497 StepFunctions.SendTaskFailureHttp,
498 StepFunctions.SendTaskHeartbeatHttp,
499 StepFunctions.ValidateStateMachineDefinitionHttp,
500 StepFunctions.TestStateHttp,
501 StepFunctions.GetActivityTaskHttp,
502 StepFunctions.ListExecutionsHttp,
503 StepFunctions.GetExecutionHistoryHttp,
504 StepFunctions.RedriveExecutionHttp,
505 StepFunctions.ListMapRunsHttp,
506 StepFunctions.DescribeMapRunHttp,
507 StepFunctions.UpdateMapRunHttp,
508 Lambda.EventSource,
509 SQS.ReceiveMessageHttp,
510 SQS.DeleteMessageHttp,
511 ),
512 ),
513 ),
514);