1import * as AWS from "@/AWS";
2import * as Kinesis from "@distilled.cloud/aws/kinesis";
3import * as Context from "effect/Context";
4import * as Data from "effect/Data";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import * as Schedule from "effect/Schedule";
9import * as Stream from "effect/Stream";
10import { HttpServerRequest } from "effect/http/HttpServerRequest";
11import * as HttpServerResponse from "effect/http/HttpServerResponse";
12
13export class KinesisApiFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
14 "KinesisApiFunction",
15) {}
16
17export class StreamAndConsumer extends Context.Service<
18 StreamAndConsumer,
19 {
20 stream: AWS.Kinesis.Stream;
21 consumer: AWS.Kinesis.StreamConsumer;
22 }
23>()("StreamAndConsumer") {}
24
25export const StreamAndConsumerLive = Layer.effect(
26 StreamAndConsumer,
27 Effect.gen(function* () {
28 const stream = yield* AWS.Kinesis.Stream("FixtureStream", {
29 streamMode: "PROVISIONED",
30 shardCount: 1,
31 tags: {
32 fixture: "kinesis-bindings",
33 },
34 });
35
36 const consumer = yield* AWS.Kinesis.StreamConsumer("FixtureConsumer", {
37 streamArn: stream.streamArn,
38 tags: {
39 fixture: "kinesis-bindings",
40 },
41 });
42
43 return {
44 stream,
45 consumer,
46 };
47 }),
48);
49
50export const KinesisApiFunctionLive = KinesisApiFunction.make(
51 {
52 main: import.meta.url,
53 functionUrl: true,
54 // The sink's bounded partial-failure retry can sleep up to ~6s, which
55 // exceeds Lambda's 3s default timeout (see PATTERNS §7).
56 timeout: Duration.seconds(30),
57 },
58 Effect.gen(function* () {
59 const { stream, consumer } = yield* StreamAndConsumer;
60
61 const describeAccountSettings =
62 yield* AWS.Kinesis.DescribeAccountSettings();
63 const describeLimits = yield* AWS.Kinesis.DescribeLimits();
64 const listStreams = yield* AWS.Kinesis.ListStreams();
65 const describeStream = yield* AWS.Kinesis.DescribeStream(stream);
66 const describeStreamSummary =
67 yield* AWS.Kinesis.DescribeStreamSummary(stream);
68 const listShards = yield* AWS.Kinesis.ListShards(stream);
69 const getShardIterator = yield* AWS.Kinesis.GetShardIterator(stream);
70 const getRecords = yield* AWS.Kinesis.GetRecords(stream);
71 const getResourcePolicy = yield* AWS.Kinesis.GetResourcePolicy(stream);
72 const listStreamConsumers = yield* AWS.Kinesis.ListStreamConsumers(stream);
73 const describeStreamConsumer =
74 yield* AWS.Kinesis.DescribeStreamConsumer(consumer);
75 const subscribeToShard = yield* AWS.Kinesis.SubscribeToShard(consumer);
76 const listTagsForResource = yield* AWS.Kinesis.ListTagsForResource(stream);
77 const putRecord = yield* AWS.Kinesis.PutRecord(stream);
78 const putRecords = yield* AWS.Kinesis.PutRecords(stream);
79 const sink = yield* AWS.Kinesis.StreamSink(stream);
80 const splitShard = yield* AWS.Kinesis.SplitShard(stream);
81 const mergeShards = yield* AWS.Kinesis.MergeShards(stream);
82
83 return {
84 fetch: Effect.gen(function* () {
85 const request = yield* HttpServerRequest;
86 const url = new URL(request.originalUrl);
87 const pathname = url.pathname;
88
89 if (request.method === "GET" && pathname === "/ready") {
90 return yield* HttpServerResponse.json({ ok: true });
91 }
92
93 if (request.method === "GET" && pathname === "/account-settings") {
94 const response = yield* describeAccountSettings().pipe(
95 Effect.match({
96 onFailure: (error) => ({
97 ok: false as const,
98 error:
99 typeof error === "object" && error !== null && "_tag" in error
100 ? (error as { _tag: string })._tag
101 : `${error}`,
102 }),
103 onSuccess: (value) => ({
104 ok: true as const,
105 value,
106 }),
107 }),
108 );
109 return yield* HttpServerResponse.json(response);
110 }
111
112 if (request.method === "GET" && pathname === "/limits") {
113 const response = yield* describeLimits().pipe(
114 Effect.match({
115 onFailure: (error) => ({
116 ok: false as const,
117 error:
118 typeof error === "object" && error !== null && "_tag" in error
119 ? (error as { _tag: string })._tag
120 : `${error}`,
121 }),
122 onSuccess: (value) => ({
123 ok: true as const,
124 value,
125 }),
126 }),
127 );
128 return yield* HttpServerResponse.json(response);
129 }
130
131 if (request.method === "GET" && pathname === "/streams") {
132 return yield* HttpServerResponse.json(yield* listStreams());
133 }
134
135 if (request.method === "GET" && pathname === "/stream") {
136 return yield* HttpServerResponse.json(yield* describeStream());
137 }
138
139 if (request.method === "GET" && pathname === "/stream-summary") {
140 return yield* HttpServerResponse.json(yield* describeStreamSummary());
141 }
142
143 if (request.method === "GET" && pathname === "/resource-policy") {
144 const response = yield* getResourcePolicy().pipe(
145 Effect.match({
146 onFailure: (error) => ({
147 ok: false as const,
148 error:
149 typeof error === "object" && error !== null && "_tag" in error
150 ? (error as { _tag: string })._tag
151 : `${error}`,
152 }),
153 onSuccess: (value) => ({
154 ok: true as const,
155 value,
156 }),
157 }),
158 );
159 return yield* HttpServerResponse.json(response);
160 }
161
162 if (request.method === "GET" && pathname === "/shards") {
163 return yield* HttpServerResponse.json(yield* listShards());
164 }
165
166 if (request.method === "GET" && pathname === "/stream-consumers") {
167 return yield* HttpServerResponse.json(yield* listStreamConsumers());
168 }
169
170 if (request.method === "GET" && pathname === "/consumer") {
171 return yield* HttpServerResponse.json(
172 yield* describeStreamConsumer(),
173 );
174 }
175
176 if (request.method === "GET" && pathname === "/tags") {
177 return yield* HttpServerResponse.json(yield* listTagsForResource());
178 }
179
180 if (request.method === "POST" && pathname === "/put-record") {
181 const body = (yield* request.json) as {
182 partitionKey: string;
183 data: string;
184 };
185 return yield* HttpServerResponse.json(
186 yield* putRecord({
187 PartitionKey: body.partitionKey,
188 Data: new TextEncoder().encode(body.data),
189 }),
190 );
191 }
192
193 if (request.method === "POST" && pathname === "/put-records") {
194 const body = (yield* request.json) as {
195 records: Array<{ partitionKey: string; data: string }>;
196 };
197 return yield* HttpServerResponse.json(
198 yield* putRecords({
199 Records: body.records.map((record) => ({
200 PartitionKey: record.partitionKey,
201 Data: new TextEncoder().encode(record.data),
202 })),
203 }),
204 );
205 }
206
207 if (request.method === "POST" && pathname === "/sink") {
208 const body = (yield* request.json) as {
209 records: Array<{ partitionKey: string; data: string }>;
210 };
211 yield* Stream.fromIterable(
212 body.records.map((record) => ({
213 PartitionKey: record.partitionKey,
214 Data: new TextEncoder().encode(record.data),
215 })),
216 ).pipe(Stream.run(sink));
217 return yield* HttpServerResponse.json({ ok: true });
218 }
219
220 if (request.method === "POST" && pathname === "/iterator") {
221 const body = (yield* request.json) as { shardId: string };
222 return yield* HttpServerResponse.json(
223 yield* getShardIterator({
224 ShardId: body.shardId,
225 ShardIteratorType: "LATEST",
226 }),
227 );
228 }
229
230 if (request.method === "POST" && pathname === "/records") {
231 const body = (yield* request.json) as {
232 shardId: string;
233 partitionKey: string;
234 data: string;
235 };
236
237 const iterator = yield* getShardIterator({
238 ShardId: body.shardId,
239 ShardIteratorType: "LATEST",
240 });
241 const shardIterator = iterator.ShardIterator;
242
243 if (!shardIterator) {
244 return yield* HttpServerResponse.json(
245 { error: "No shard iterator returned" },
246 { status: 500 },
247 );
248 }
249
250 yield* putRecord({
251 PartitionKey: body.partitionKey,
252 Data: new TextEncoder().encode(body.data),
253 });
254
255 const result = yield* waitForRecords(getRecords, shardIterator);
256 return yield* HttpServerResponse.json({
257 records: (result.Records ?? []).map((record) => ({
258 partitionKey: record.PartitionKey,
259 data: decodeText(record.Data),
260 })),
261 millisBehindLatest: result.MillisBehindLatest,
262 });
263 }
264
265 if (request.method === "POST" && pathname === "/split-shard") {
266 const body = (yield* request.json) as {
267 shardToSplit: string;
268 newStartingHashKey: string;
269 };
270 yield* splitShard({
271 ShardToSplit: body.shardToSplit,
272 NewStartingHashKey: body.newStartingHashKey,
273 });
274 return yield* HttpServerResponse.json({ ok: true });
275 }
276
277 if (request.method === "POST" && pathname === "/merge-shards") {
278 const body = (yield* request.json) as {
279 shardToMerge: string;
280 adjacentShardToMerge: string;
281 };
282 yield* mergeShards({
283 ShardToMerge: body.shardToMerge,
284 AdjacentShardToMerge: body.adjacentShardToMerge,
285 });
286 return yield* HttpServerResponse.json({ ok: true });
287 }
288
289 if (request.method === "POST" && pathname === "/subscribe") {
290 const body = (yield* request.json) as { shardId: string };
291 const result = yield* subscribeToShard({
292 ShardId: body.shardId,
293 StartingPosition: {
294 Type: "LATEST",
295 },
296 });
297 return yield* HttpServerResponse.json({
298 ok: result.EventStream !== undefined,
299 });
300 }
301
302 return yield* HttpServerResponse.json(
303 { error: "Not found", method: request.method, pathname },
304 { status: 404 },
305 );
306 }).pipe(Effect.orDie),
307 };
308 }).pipe(
309 Effect.provide(
310 Layer.provideMerge(
311 Layer.mergeAll(AWS.Kinesis.StreamSinkHttp, AWS.Kinesis.PutRecordsHttp),
312 Layer.mergeAll(
313 AWS.Kinesis.DescribeAccountSettingsHttp,
314 AWS.Kinesis.DescribeLimitsHttp,
315 AWS.Kinesis.DescribeStreamHttp,
316 AWS.Kinesis.DescribeStreamConsumerHttp,
317 AWS.Kinesis.DescribeStreamSummaryHttp,
318 AWS.Kinesis.GetRecordsHttp,
319 AWS.Kinesis.GetResourcePolicyHttp,
320 AWS.Kinesis.GetShardIteratorHttp,
321 AWS.Kinesis.ListShardsHttp,
322 AWS.Kinesis.ListStreamConsumersHttp,
323 AWS.Kinesis.ListStreamsHttp,
324 AWS.Kinesis.ListTagsForResourceHttp,
325 AWS.Kinesis.MergeShardsHttp,
326 AWS.Kinesis.PutRecordHttp,
327 AWS.Kinesis.PutRecordsHttp,
328 AWS.Kinesis.SplitShardHttp,
329 AWS.Kinesis.SubscribeToShardHttp,
330 StreamAndConsumerLive,
331 ),
332 ),
333 ),
334 ),
335 // Re-merge so the deploying Stack can `yield* StreamAndConsumer` and expose
336 // the stream/consumer names as deploy-time outputs. Reusing the same
337 // `StreamAndConsumerLive` reference keeps it a single shared stream/consumer.
338).pipe(Layer.provideMerge(StreamAndConsumerLive));
339
340export default KinesisApiFunctionLive;
341
342const decodeText = (value: Uint8Array<ArrayBufferLike>) =>
343 new TextDecoder().decode(value);
344
345class RecordsNotReady extends Data.TaggedError("RecordsNotReady") {}
346
347const waitForRecords = (
348 getRecords: (
349 request: Kinesis.GetRecordsInput,
350 ) => Effect.Effect<Kinesis.GetRecordsOutput, Kinesis.GetRecordsError>,
351 shardIterator: string,
352) =>
353 getRecords({
354 ShardIterator: shardIterator,
355 }).pipe(
356 Effect.flatMap((result) =>
357 (result.Records?.length ?? 0) > 0
358 ? Effect.succeed(result)
359 : Effect.fail(new RecordsNotReady()),
360 ),
361 Effect.retry({
362 while: (error) => error._tag === "RecordsNotReady",
363 schedule: Schedule.max([Schedule.fixed("1 second"), Schedule.recurs(10)]),
364 }),
365 );