1import * as AWS from "@/AWS";
2import * as Alchemy from "@/index.ts";
3import * as State from "@/State";
4import * as Test from "@/Test/Alchemy";
5import { describe, expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import KinesisApiFunctionLive, {
11 KinesisApiFunction,
12 StreamAndConsumer,
13} from "./handler.ts";
14
15const providers = AWS.providers();
16const state = State.localState();
17const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
18 providers,
19 state,
20});
21
22const Stack = Alchemy.Stack(
23 "kinesis-bindings",
24 { providers, state },
25 Effect.gen(function* () {
26 // Share one stream/consumer between the deployed function and the stack
27 // outputs so the test can assert the live API responses against the known
28 // names without round-tripping them through the fixture's runtime.
29 const { stream, consumer } = yield* StreamAndConsumer;
30 const fn = yield* KinesisApiFunction;
31 return {
32 url: fn.functionUrl.as<string>(),
33 streamName: stream.streamName.as<string>(),
34 consumerName: consumer.consumerName.as<string>(),
35 };
36 }).pipe(Effect.provide(KinesisApiFunctionLive)),
37);
38
39const stack = beforeAll(
40 Effect.gen(function* () {
41 yield* Effect.logInfo("Kinesis test setup: destroying previous resources");
42 yield* destroy(Stack);
43
44 yield* Effect.logInfo("Kinesis test setup: deploying fixture");
45 return yield* deploy(Stack);
46 }),
47 { timeout: 240_000 },
48);
49afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), {
50 timeout: 300_000,
51});
52
53// Lambda Function URLs cold-start (DNS, init) and a fresh role's IAM grants
54// (eventual consistency) can both take a while on the first hit. Retrying on
55// any non-200 lets the first request wait through that window; warm calls
56// return on the first try and never retry.
57const readinessSchedule = Schedule.max([
58 Schedule.fixed("4 seconds"),
59 Schedule.recurs(10),
60]);
61
62// Lambda Function URLs come back with a trailing slash (`https://…on.aws/`).
63// Naively concatenating `${baseUrl}${path}` would yield a double slash
64// (`…on.aws//stream`), whose pathname (`//stream`) never matches the fixture's
65// `/stream` route, so every request 404s and the readiness retry spins until
66// the test times out. Strip the trailing slash before joining.
67const urlOf = (baseUrl: string, path: string) =>
68 `${baseUrl.replace(/\/+$/, "")}${path}`;
69
70const getJson = (baseUrl: string, path: string) =>
71 HttpClient.get(urlOf(baseUrl, path)).pipe(
72 Effect.timeout("5 seconds"),
73 Effect.flatMap((response) =>
74 response.status === 200
75 ? response.json
76 : Effect.fail(new Error(`Request failed: ${response.status}`)),
77 ),
78 Effect.retry({ schedule: readinessSchedule }),
79 );
80
81const postJson = (baseUrl: string, path: string, body: unknown) =>
82 HttpClient.execute(
83 HttpClientRequest.bodyJsonUnsafe(
84 HttpClientRequest.post(urlOf(baseUrl, path)),
85 body,
86 ),
87 ).pipe(
88 // Runtime routes are normally sub-second. The record-polling and sink
89 // routes can legitimately spend several seconds on bounded retries.
90 Effect.timeout("15 seconds"),
91 Effect.flatMap((response) =>
92 response.status === 200
93 ? response.json
94 : Effect.fail(new Error(`Request failed: ${response.status}`)),
95 ),
96 Effect.retry({ schedule: readinessSchedule }),
97 );
98
99const getFirstShardId = (baseUrl: string) =>
100 getJson(baseUrl, "/shards").pipe(
101 Effect.map((response) => (response as any).Shards?.[0]?.ShardId as string),
102 );
103
104describe.sequential(
105 "Kinesis Bindings",
106 {
107 tags: [
108 "provider:aws",
109 "provider:aws:kinesis",
110 "provider:aws:lambda",
111 "live",
112 ],
113 },
114 () => {
115 describe("DescribeAccountSettings", () => {
116 test(
117 "returns the account settings payload",
118 Effect.gen(function* () {
119 const { url } = yield* stack;
120 const response = yield* getJson(url, "/account-settings");
121 if ((response as any).ok === false) {
122 expect((response as any).error).toBeTruthy();
123 } else {
124 expect((response as any).value).toBeDefined();
125 }
126 }),
127 );
128 });
129
130 describe("DescribeLimits", () => {
131 test(
132 "returns shard and stream limits",
133 Effect.gen(function* () {
134 const { url } = yield* stack;
135 const response = yield* getJson(url, "/limits");
136 if ((response as any).ok === false) {
137 expect((response as any).error).toBeTruthy();
138 } else {
139 expect((response as any).value.ShardLimit).toBeGreaterThan(0);
140 }
141 }),
142 );
143 });
144
145 describe("ListStreams", () => {
146 test(
147 "lists the deployed stream",
148 Effect.gen(function* () {
149 // Kinesis ListStreams is paginated and the alchemy binding wraps
150 // the single-page operation. On an account with > 100 streams our
151 // brand-new stream may simply not be on page 1. Just verify the
152 // binding returns an Array; the specific stream is verified via
153 // DescribeStream below.
154 const { url } = yield* stack;
155 const response = yield* getJson(url, "/streams");
156 const names = (response as any).StreamNames ?? [];
157 expect(Array.isArray(names)).toBe(true);
158 }),
159 );
160 });
161
162 describe("DescribeStream", () => {
163 test(
164 "describes the bound stream",
165 Effect.gen(function* () {
166 const { url, streamName } = yield* stack;
167 const response = yield* getJson(url, "/stream");
168 expect((response as any).StreamDescription.StreamName).toBe(
169 streamName,
170 );
171 }),
172 );
173 });
174
175 describe("DescribeStreamSummary", () => {
176 test(
177 "describes the bound stream summary",
178 Effect.gen(function* () {
179 const { url, streamName } = yield* stack;
180 const response = yield* getJson(url, "/stream-summary");
181 expect((response as any).StreamDescriptionSummary.StreamName).toBe(
182 streamName,
183 );
184 }),
185 );
186 });
187
188 describe("GetResourcePolicy", () => {
189 test(
190 "returns the stream policy or a structured error",
191 Effect.gen(function* () {
192 const { url } = yield* stack;
193 const response = yield* getJson(url, "/resource-policy");
194 if ((response as any).ok === false) {
195 expect((response as any).error).toBeTruthy();
196 } else {
197 expect((response as any).value).toBeDefined();
198 }
199 }),
200 );
201 });
202
203 describe("ListShards", () => {
204 test(
205 "lists shards for the stream",
206 Effect.gen(function* () {
207 const { url } = yield* stack;
208 const response = yield* getJson(url, "/shards");
209 expect(((response as any).Shards ?? []).length).toBeGreaterThan(0);
210 }),
211 );
212 });
213
214 describe("GetShardIterator", () => {
215 test(
216 "returns a shard iterator for the first shard",
217 Effect.gen(function* () {
218 const { url } = yield* stack;
219 const shardId = yield* getFirstShardId(url);
220 const response = yield* postJson(url, "/iterator", { shardId });
221 expect((response as any).ShardIterator).toBeTruthy();
222 }),
223 );
224 });
225
226 describe("GetRecords", () => {
227 test(
228 "reads a just-written record through the shard iterator",
229 Effect.gen(function* () {
230 const { url } = yield* stack;
231 const shardId = yield* getFirstShardId(url);
232 const marker = `records-${crypto.randomUUID()}`;
233 const response = yield* postJson(url, "/records", {
234 shardId,
235 partitionKey: "records-test",
236 data: marker,
237 });
238 const records = (response as any).records ?? [];
239 expect(records.some((record: any) => record.data === marker)).toBe(
240 true,
241 );
242 }),
243 );
244 });
245
246 describe("ListStreamConsumers", () => {
247 test(
248 "lists the registered consumer",
249 Effect.gen(function* () {
250 const { url, consumerName } = yield* stack;
251 const response = yield* getJson(url, "/stream-consumers");
252 const consumers = (response as any).Consumers ?? [];
253 expect(
254 consumers.some(
255 (consumer: any) => consumer.ConsumerName === consumerName,
256 ),
257 ).toBe(true);
258 }),
259 );
260 });
261
262 describe("DescribeStreamConsumer", () => {
263 test(
264 "describes the registered consumer",
265 Effect.gen(function* () {
266 const { url, consumerName } = yield* stack;
267 const response = yield* getJson(url, "/consumer");
268 expect((response as any).ConsumerDescription.ConsumerName).toBe(
269 consumerName,
270 );
271 }),
272 );
273 });
274
275 describe("SubscribeToShard", () => {
276 test(
277 "opens a subscribe-to-shard stream",
278 Effect.gen(function* () {
279 const { url } = yield* stack;
280 const shardId = yield* getFirstShardId(url);
281 const response = yield* postJson(url, "/subscribe", { shardId });
282 expect((response as any).ok).toBe(true);
283 }),
284 );
285 });
286
287 describe("ListTagsForResource", () => {
288 test(
289 "lists the stream ownership tags",
290 Effect.gen(function* () {
291 const { url } = yield* stack;
292 const response = yield* getJson(url, "/tags");
293 const keys = ((response as any).Tags ?? []).map(
294 (tag: any) => tag.Key,
295 );
296 expect(keys).toContain("alchemy::stack");
297 expect(keys).toContain("alchemy::stage");
298 expect(keys).toContain("alchemy::id");
299 expect(keys).toContain("fixture");
300 }),
301 );
302 });
303
304 describe("PutRecord", () => {
305 test(
306 "writes a single record",
307 Effect.gen(function* () {
308 const { url } = yield* stack;
309 const response = yield* postJson(url, "/put-record", {
310 partitionKey: "put-record",
311 data: `put-record-${crypto.randomUUID()}`,
312 });
313 expect((response as any).ShardId).toBeTruthy();
314 expect((response as any).SequenceNumber).toBeTruthy();
315 }),
316 );
317 });
318
319 describe("PutRecords", () => {
320 test(
321 "writes a batch of records",
322 Effect.gen(function* () {
323 const { url } = yield* stack;
324 const response = yield* postJson(url, "/put-records", {
325 records: [
326 {
327 partitionKey: "put-records",
328 data: `batch-1-${crypto.randomUUID()}`,
329 },
330 {
331 partitionKey: "put-records",
332 data: `batch-2-${crypto.randomUUID()}`,
333 },
334 ],
335 });
336 expect((response as any).FailedRecordCount ?? 0).toBe(0);
337 expect(((response as any).Records ?? []).length).toBe(2);
338 }),
339 );
340 });
341
342 describe("StreamSink", () => {
343 test(
344 "writes records through the sink helper",
345 Effect.gen(function* () {
346 const { url } = yield* stack;
347 const response = yield* postJson(url, "/sink", {
348 records: [
349 { partitionKey: "sink", data: `sink-1-${crypto.randomUUID()}` },
350 { partitionKey: "sink", data: `sink-2-${crypto.randomUUID()}` },
351 ],
352 });
353 expect((response as any).ok).toBe(true);
354 }),
355 );
356
357 test(
358 "splits more than 500 records into multiple PutRecords calls",
359 Effect.gen(function* () {
360 const { url } = yield* stack;
361 // 501 records > the PutRecords limit of 500, so the batched sink
362 // must split the chunk into 2 sequential API calls (500 + 1). Any
363 // per-record throttling failures are retried by the sink engine.
364 const marker = crypto.randomUUID();
365 const response = yield* postJson(url, "/sink", {
366 records: Array.from({ length: 501 }, (_, i) => ({
367 partitionKey: `sink-${i % 7}`,
368 data: `sink-${marker}-${i}`,
369 })),
370 });
371 expect((response as any).ok).toBe(true);
372 }),
373 { timeout: 120_000 },
374 );
375 });
376
377 // Shard-topology tests run LAST (describe.sequential) because they change
378 // the stream's shard layout: earlier tests assume `Shards[0]` is the
379 // single open shard the fixture was created with.
380 describe("SplitShard", () => {
381 test(
382 "splits the open shard into two children",
383 Effect.gen(function* () {
384 const { url } = yield* stack;
385
386 const openShards = yield* getOpenShards(url);
387 expect(openShards.length).toBe(1);
388 const shard = openShards[0]!;
389
390 // Split at the midpoint of the shard's hash-key range.
391 const start = BigInt(shard.HashKeyRange.StartingHashKey);
392 const end = BigInt(shard.HashKeyRange.EndingHashKey);
393 const midpoint = (start + end) / 2n + 1n;
394
395 const response = yield* postJson(url, "/split-shard", {
396 shardToSplit: shard.ShardId,
397 newStartingHashKey: midpoint.toString(),
398 });
399 expect((response as any).ok).toBe(true);
400
401 // The split completes asynchronously: the stream transitions
402 // UPDATING -> ACTIVE and the parent shard closes, leaving two
403 // open children.
404 const children = yield* waitForOpenShardCount(url, 2);
405 for (const child of children) {
406 expect(child.ParentShardId).toBe(shard.ShardId);
407 }
408 }),
409 { timeout: 120_000 },
410 );
411 });
412
413 describe("MergeShards", () => {
414 test(
415 "merges the two adjacent children back into one shard",
416 Effect.gen(function* () {
417 const { url } = yield* stack;
418
419 const openShards = yield* waitForOpenShardCount(url, 2);
420 // Order the two children by hash-key range so they are passed as
421 // (shard, adjacent-shard) the way MergeShards expects.
422 const [low, high] = [...openShards].sort((a, b) =>
423 BigInt(a.HashKeyRange.StartingHashKey) <
424 BigInt(b.HashKeyRange.StartingHashKey)
425 ? -1
426 : 1,
427 );
428
429 const response = yield* postJson(url, "/merge-shards", {
430 shardToMerge: low!.ShardId,
431 adjacentShardToMerge: high!.ShardId,
432 });
433 expect((response as any).ok).toBe(true);
434
435 const merged = yield* waitForOpenShardCount(url, 1);
436 expect(merged[0]!.ParentShardId).toBe(low!.ShardId);
437 }),
438 { timeout: 120_000 },
439 );
440 });
441 },
442);
443
444interface ShardInfo {
445 ShardId: string;
446 ParentShardId?: string;
447 HashKeyRange: { StartingHashKey: string; EndingHashKey: string };
448 SequenceNumberRange: { EndingSequenceNumber?: string };
449}
450
451// A shard is open while its sequence-number range has no upper bound.
452const getOpenShards = (baseUrl: string) =>
453 getJson(baseUrl, "/shards").pipe(
454 Effect.map((response) =>
455 (((response as any).Shards ?? []) as ShardInfo[]).filter(
456 (shard) =>
457 shard.SequenceNumberRange?.EndingSequenceNumber === undefined,
458 ),
459 ),
460 );
461
462const waitForOpenShardCount = (baseUrl: string, count: number) =>
463 getOpenShards(baseUrl).pipe(
464 Effect.repeat({
465 schedule: Schedule.spaced("5 seconds"),
466 until: (shards): boolean => shards.length === count,
467 times: 20,
468 }),
469 Effect.flatMap((shards) =>
470 shards.length === count
471 ? Effect.succeed(shards)
472 : Effect.fail(
473 new Error(
474 `expected ${count} open shards, saw ${shards.length} after polling`,
475 ),
476 ),
477 ),
478 );