1import * as AWS from "@/AWS";
2import { Stream } from "@/AWS/DSQL";
3import * as Test from "@/Test/Alchemy";
4import * as dsql from "@distilled.cloud/aws/dsql";
5import { expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9
10const { test } = Test.make({ providers: AWS.providers() });
11
12const getStream = (clusterIdentifier: string, streamIdentifier: string) =>
13 dsql
14 .getStream({ clusterIdentifier, streamIdentifier })
15 .pipe(
16 Effect.catchTag("ResourceNotFoundException", () =>
17 Effect.succeed(undefined),
18 ),
19 );
20
21class StreamStillPresent extends Data.TaggedError("StreamStillPresent")<{
22 readonly streamId: string;
23 readonly status: string;
24}> {}
25
26// Gated behind AWS_TEST_SLOW: a DSQL CDC stream transitions
27// CREATING -> ACTIVE in one to three minutes (AWS docs), which is beyond the
28// ~90s async-provisioning budget for the always-on suite. The full lifecycle
29// (cluster + Kinesis target + IAM service role + stream + destroy) measures
30// ~5-7 minutes wall.
31test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
32 "create and delete a DSQL CDC stream into Kinesis",
33 (stack) =>
34 Effect.gen(function* () {
35 yield* stack.destroy();
36
37 const created = yield* stack.deploy(
38 Effect.gen(function* () {
39 const cluster = yield* AWS.DSQL.Cluster("CdcDb", {});
40 // CDC records can approach 10 MiB — the docs require
41 // maxRecordSizeInKiB 10240 and ON_DEMAND mode on the target.
42 const target = yield* AWS.Kinesis.Stream("CdcTarget", {
43 streamMode: "ON_DEMAND",
44 maxRecordSizeInKiB: 10240,
45 });
46 const role = yield* AWS.IAM.Role("CdcRole", {
47 assumeRolePolicyDocument: {
48 Version: "2012-10-17",
49 Statement: [
50 {
51 Sid: "DSQLAccess",
52 Effect: "Allow",
53 Principal: { Service: "dsql.amazonaws.com" },
54 Action: ["sts:AssumeRole"],
55 },
56 ],
57 },
58 inlinePolicies: {
59 kinesis: {
60 Version: "2012-10-17",
61 Statement: [
62 {
63 Sid: "KinesisAccess",
64 Effect: "Allow",
65 Action: [
66 "kinesis:PutRecord",
67 "kinesis:PutRecords",
68 "kinesis:DescribeStreamSummary",
69 "kinesis:ListShards",
70 ],
71 Resource: target.streamArn,
72 },
73 ],
74 },
75 },
76 });
77 const cdc = yield* Stream("Cdc", {
78 clusterId: cluster.clusterId,
79 kinesisStreamArn: target.streamArn,
80 roleArn: role.roleArn,
81 tags: { app: "alchemy-test" },
82 });
83 return {
84 clusterId: cdc.clusterId,
85 streamId: cdc.streamId,
86 streamArn: cdc.streamArn,
87 status: cdc.status,
88 kinesisStreamArn: cdc.kinesisStreamArn,
89 roleArn: cdc.roleArn,
90 };
91 }),
92 );
93
94 expect(created.streamId).toBeDefined();
95 expect(created.status).toBe("ACTIVE");
96 expect(created.streamArn).toContain(
97 `:cluster/${created.clusterId}/stream/${created.streamId}`,
98 );
99 expect(created.kinesisStreamArn).toContain(":stream/");
100 expect(created.roleArn).toContain(":role/");
101
102 // out-of-band verification via distilled
103 const observed = yield* getStream(created.clusterId, created.streamId);
104 expect(observed?.status).toBe("ACTIVE");
105 expect(observed?.targetDefinition?.kinesis.streamArn).toEqual(
106 created.kinesisStreamArn,
107 );
108 expect(observed?.tags?.app).toEqual("alchemy-test");
109
110 // destroy everything; typed wait-until-gone on the CDC stream
111 yield* stack.destroy();
112 yield* getStream(created.clusterId, created.streamId).pipe(
113 Effect.flatMap((stream) =>
114 stream === undefined || stream.status === "DELETED"
115 ? Effect.void
116 : Effect.fail(
117 new StreamStillPresent({
118 streamId: created.streamId,
119 status: stream.status,
120 }),
121 ),
122 ),
123 Effect.retry({
124 while: (e): boolean => e._tag === "StreamStillPresent",
125 schedule: Schedule.max([
126 Schedule.spaced("5 seconds"),
127 Schedule.recurs(10),
128 ]),
129 }),
130 );
131 }),
132 {
133 tags: [
134 "provider:aws",
135 "provider:aws:dsql",
136 "provider:aws:iam",
137 "provider:aws:kinesis",
138 "live",
139 ],
140 timeout: 600_000,
141 },
142);