1import * as AWS from "@/AWS";
2import { AWSEnvironment } from "@/AWS/Environment.ts";
3import * as Test from "@/Test/Alchemy";
4import * as keyspacesstreams from "@distilled.cloud/aws/keyspacesstreams";
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";
9
10import KeyspacesStreamsTestFunctionLive, {
11 KeyspacesStreamsTestFunction,
12} from "./streams-handler.ts";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16// keyspacesstreams is a data-plane read API over CDC streams of Keyspaces
17// tables. The ungated probes assert the distilled wiring surfaces typed
18// errors; the live traverse (which needs a CDC-enabled table + a deployed
19// Lambda exercising the TableStreams binding) is gated behind
20// AWS_TEST_SLOW=1.
21describe(
22 "AWS.Keyspaces.TableStreams",
23 { tags: ["provider:aws", "provider:aws:keyspaces", "live"] },
24 () => {
25 test.provider(
26 "getStream on a nonexistent stream yields a typed error",
27 (_stack) =>
28 Effect.gen(function* () {
29 const { accountId, region } = yield* AWSEnvironment.current;
30 const error = yield* keyspacesstreams
31 .getStream({
32 streamArn: `arn:aws:cassandra:${region}:${accountId}:/keyspace/alchemy_nonexistent_ks/table/nonexistent_tbl/stream/2024-01-01T00:00:00.000`,
33 })
34 .pipe(Effect.flip);
35 expect([
36 "ResourceNotFoundException",
37 "ValidationException",
38 ]).toContain(error._tag);
39 }),
40 { timeout: 60_000 },
41 );
42
43 test.provider(
44 "listStreams on a nonexistent table returns an empty stream list",
45 (_stack) =>
46 Effect.gen(function* () {
47 // The API treats an unknown keyspace/table as "no streams" rather
48 // than ResourceNotFound — the round-trip proves the distilled
49 // wiring decodes ListStreamsOutput.
50 const listed = yield* keyspacesstreams.listStreams({
51 keyspaceName: "alchemy_nonexistent_ks",
52 tableName: "nonexistent_tbl",
53 });
54 expect(listed.streams ?? []).toHaveLength(0);
55 }),
56 { timeout: 60_000 },
57 );
58
59 test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
60 "CDC table exposes latestStreamArn and Lambda traverses the stream via the binding",
61 (stack) =>
62 Effect.gen(function* () {
63 yield* stack.destroy();
64
65 const { functionUrl } = yield* stack.deploy(
66 Effect.gen(function* () {
67 return yield* KeyspacesStreamsTestFunction;
68 }).pipe(Effect.provide(KeyspacesStreamsTestFunctionLive)),
69 );
70 expect(functionUrl).toBeTruthy();
71 const baseUrl = functionUrl!.replace(/\/+$/, "");
72
73 // Out-of-band: the CDC stream is enumerable via distilled directly.
74 const listed = yield* keyspacesstreams.listStreams({
75 keyspaceName: "alchemy_streams_test_ks",
76 tableName: "orders",
77 });
78 expect((listed.streams ?? []).length).toBeGreaterThanOrEqual(1);
79
80 // Drive the deployed Lambda through the TableStreams binding —
81 // proves the cassandra:* stream IAM actions and the client wiring.
82 // Retry through function-URL cold start / IAM propagation.
83 const body = yield* HttpClient.get(`${baseUrl}/traverse`).pipe(
84 Effect.flatMap((response) =>
85 response.status === 200
86 ? response.json
87 : Effect.fail(
88 new Error(`traverse not ready: ${response.status}`),
89 ),
90 ),
91 Effect.retry({
92 schedule: Schedule.max([
93 Schedule.exponential("1 second"),
94 Schedule.recurs(8),
95 ]),
96 }),
97 );
98 const traverse = body as {
99 streamArn: string;
100 streamStatus: string;
101 viewType: string;
102 shardCount: number;
103 recordCount: number;
104 };
105 expect(traverse.streamArn).toContain("/stream/");
106 expect(["ENABLED", "ENABLING"]).toContain(traverse.streamStatus);
107 expect(traverse.viewType).toBe("NEW_AND_OLD_IMAGES");
108 expect(traverse.shardCount).toBeGreaterThanOrEqual(1);
109 // Idle table — no writes were made, so no change records.
110 expect(traverse.recordCount).toBeGreaterThanOrEqual(0);
111
112 yield* stack.destroy();
113 }),
114 { tags: ["provider:aws:lambda"], timeout: 900_000 },
115 );
116 },
117);