1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as Kinesis from "@distilled.cloud/aws/kinesis";
4import * as Lambda from "@distilled.cloud/aws/lambda";
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 KinesisStreamFunctionLive, {
11 KinesisStreamFunction,
12} from "./stream-handler.ts";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16describe.sequential(
17 "AWS.Kinesis.StreamEventSource",
18 {
19 tags: [
20 "provider:aws",
21 "provider:aws:kinesis",
22 "provider:aws:lambda",
23 "provider:aws:sqs",
24 "live",
25 ],
26 },
27 () => {
28 test.provider(
29 "processes real Kinesis records through Lambda",
30 (stack) =>
31 Effect.gen(function* () {
32 yield* stack.destroy();
33
34 const streamFunction = yield* stack.deploy(
35 KinesisStreamFunction.pipe(
36 Effect.provide(KinesisStreamFunctionLive),
37 ),
38 );
39
40 const functionUrl = streamFunction.functionUrl!;
41
42 const { streamName, streamArn } = yield* HttpClient.get(
43 functionUrl,
44 ).pipe(
45 Effect.flatMap((response) =>
46 response.status === 200
47 ? (response.json as Effect.Effect<{
48 streamName: string;
49 streamArn: string;
50 }>)
51 : Effect.fail(
52 new FunctionNotReady(
53 `Function not ready: ${response.status}`,
54 ),
55 ),
56 ),
57 Effect.flatMap((body) =>
58 body.streamName && body.streamArn
59 ? Effect.succeed(body)
60 : Effect.fail(
61 new FunctionNotReady(
62 "Function returned empty stream identifiers",
63 ),
64 ),
65 ),
66 Effect.retry({
67 schedule: Schedule.max([
68 Schedule.fixed("1 seconds"),
69 Schedule.recurs(60),
70 ]),
71 }),
72 );
73
74 yield* waitForEventSourceMappingEnabled(
75 streamFunction.functionName,
76 streamArn,
77 );
78 yield* Effect.sleep("10 seconds");
79
80 yield* Kinesis.putRecord({
81 StreamName: streamName,
82 PartitionKey: "stream-event-source",
83 Data: new TextEncoder().encode("payload"),
84 });
85 yield* Effect.sleep("5 seconds");
86
87 const mapping = yield* waitForEventSourceMappingEnabled(
88 streamFunction.functionName,
89 streamArn,
90 );
91 expect(mapping.State).toEqual("Enabled");
92
93 yield* stack.destroy();
94 }),
95 { timeout: 240_000 },
96 );
97 },
98);
99
100const waitForEventSourceMappingEnabled = Effect.fn(function* (
101 functionName: string,
102 eventSourceArn: string,
103) {
104 return yield* Lambda.listEventSourceMappings({
105 FunctionName: functionName,
106 EventSourceArn: eventSourceArn,
107 }).pipe(
108 Effect.flatMap((result) => {
109 const mapping = result.EventSourceMappings?.[0];
110 if (!mapping || mapping.State !== "Enabled") {
111 return Effect.fail(new EventSourceMappingNotReady());
112 }
113 return Effect.succeed(mapping);
114 }),
115 Effect.retry({
116 while: (error) => error._tag === "EventSourceMappingNotReady",
117 schedule: Schedule.max([
118 Schedule.fixed("2 seconds"),
119 Schedule.recurs(20),
120 ]),
121 }),
122 );
123});
124
125class EventSourceMappingNotReady extends Data.TaggedError(
126 "EventSourceMappingNotReady",
127) {}
128
129class FunctionNotReady extends Data.TaggedError("FunctionNotReady")<{
130 message: string;
131}> {
132 constructor(message: string) {
133 super({ message });
134 }
135}