1import * as dsql from "@distilled.cloud/aws/dsql";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as EStream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11
12export interface StreamProps {
13 /**
14 * Identifier of the DSQL cluster whose committed row-level changes the
15 * stream captures. The cluster must be in `ACTIVE` status when the stream
16 * is created (an `IDLE` cluster rejects `CreateStream` with a validation
17 * error). Changing the cluster replaces the stream.
18 */
19 clusterId: string;
20 /**
21 * ARN of the Amazon Kinesis data stream that receives the change records.
22 * Must be in the same account and region as the cluster. Configure the
23 * Kinesis stream with `maxRecordSizeInKiB: 10240` — CDC records can
24 * approach 10 MiB and an undersized stream becomes `IMPAIRED` with
25 * `KINESIS_OVERSIZE_RECORD`. Changing the target replaces the stream.
26 */
27 kinesisStreamArn: string;
28 /**
29 * ARN of the IAM role Aurora DSQL assumes to write records into the
30 * Kinesis stream. The role's trust policy must allow the
31 * `dsql.amazonaws.com` service principal and its permissions policy must
32 * grant `kinesis:PutRecord`, `kinesis:PutRecords`,
33 * `kinesis:DescribeStreamSummary` and `kinesis:ListShards` on the target
34 * stream. Changing the role replaces the stream.
35 */
36 roleArn: string;
37 /**
38 * Record ordering guarantee. Only `UNORDERED` is currently supported —
39 * consumers deduplicate and order by `source.ts_ns`.
40 * @default "UNORDERED"
41 */
42 ordering?: dsql.StreamOrdering;
43 /**
44 * Record serialization format.
45 * @default "JSON"
46 */
47 format?: dsql.StreamFormat;
48 /**
49 * User-defined tags for the stream.
50 */
51 tags?: Record<string, string>;
52}
53
54export interface Stream extends Resource<
55 "AWS.DSQL.Stream",
56 StreamProps,
57 {
58 /** Identifier of the source cluster. */
59 clusterId: string;
60 /** The unique stream identifier assigned by DSQL. */
61 streamId: string;
62 /** The ARN of the stream. */
63 streamArn: string;
64 /** The current status of the stream, e.g. `ACTIVE`. */
65 status: string;
66 /** Record ordering guarantee. */
67 ordering: string;
68 /** Record serialization format. */
69 format: string;
70 /** ARN of the target Kinesis data stream. */
71 kinesisStreamArn: string;
72 /** ARN of the IAM role DSQL assumes to write to Kinesis. */
73 roleArn: string;
74 },
75 never,
76 Providers
77> {}
78
79/**
80 * A change data capture (CDC) stream on an Aurora DSQL cluster — delivers
81 * committed row-level changes (Debezium-shaped JSON envelopes) to an Amazon
82 * Kinesis data stream.
83 *
84 * Creation is asynchronous (`CREATING` -> `ACTIVE`, typically one to three
85 * minutes); the provider waits for `ACTIVE` (bounded) before returning. A
86 * stream has no update operation — every property except tags replaces it.
87 *
88 * Functions consume the change records through the existing Kinesis event
89 * source on the **target** stream; DSQL itself never invokes compute
90 * directly.
91 *
92 * ### Creating a CDC Stream
93 * **Example:** Stream Cluster Changes into Kinesis
94 * ```typescript
95 * const cluster = yield* DSQL.Cluster("AppDb", {});
96 * const target = yield* Kinesis.Stream("Changes", {
97 * streamMode: "ON_DEMAND",
98 * maxRecordSizeInKiB: 10240,
99 * });
100 * const role = yield* IAM.Role("CdcRole", {
101 * assumeRolePolicyDocument: JSON.stringify({
102 * Version: "2012-10-17",
103 * Statement: [
104 * {
105 * Effect: "Allow",
106 * Principal: { Service: "dsql.amazonaws.com" },
107 * Action: "sts:AssumeRole",
108 * },
109 * ],
110 * }),
111 * inlinePolicies: {
112 * kinesis: JSON.stringify({
113 * Version: "2012-10-17",
114 * Statement: [
115 * {
116 * Effect: "Allow",
117 * Action: [
118 * "kinesis:PutRecord",
119 * "kinesis:PutRecords",
120 * "kinesis:DescribeStreamSummary",
121 * "kinesis:ListShards",
122 * ],
123 * Resource: target.streamArn,
124 * },
125 * ],
126 * }),
127 * },
128 * });
129 * const cdc = yield* DSQL.Stream("Cdc", {
130 * clusterId: cluster.clusterId,
131 * kinesisStreamArn: target.streamArn,
132 * roleArn: role.roleArn,
133 * });
134 * ```
135 *
136 * **Example:** Consume Change Records with a Function
137 * ```typescript
138 * // DSQL delivers into the Kinesis stream; consume it with the
139 * // Kinesis event source on the target stream.
140 * yield* Kinesis.consume(target, (records) =>
141 * Effect.forEach(records, (record) => handleChange(record)),
142 * );
143 * ```
144 *
145 * @resource
146 */
147export const Stream = Resource<Stream>("AWS.DSQL.Stream");
148
149const defaultOrdering = "UNORDERED";
150const defaultFormat = "JSON";
151
152export const StreamProvider = () =>
153 Provider.effect(
154 Stream,
155 Effect.gen(function* () {
156 const readStream = Effect.fn(function* (
157 clusterIdentifier: string,
158 streamIdentifier: string,
159 ) {
160 return yield* dsql
161 .getStream({ clusterIdentifier, streamIdentifier })
162 .pipe(
163 Effect.catchTag("ResourceNotFoundException", () =>
164 Effect.succeed(undefined),
165 ),
166 );
167 });
168
169 // Bounded readiness wait — stream creation typically completes in one
170 // to three minutes; budget 5 min (60 x 5s). FAILED is terminal.
171 const waitForActive = Effect.fn(function* (
172 clusterIdentifier: string,
173 streamIdentifier: string,
174 ) {
175 const policy = Schedule.max([
176 Schedule.fixed("5 seconds"),
177 Schedule.recurs(60),
178 ]);
179 return yield* readStream(clusterIdentifier, streamIdentifier).pipe(
180 Effect.flatMap((stream) => {
181 if (stream === undefined) {
182 return Effect.fail(
183 new Error(`DSQL stream '${streamIdentifier}' not found`),
184 );
185 }
186 if (stream.status === "FAILED") {
187 return Effect.die(
188 `DSQL stream '${streamIdentifier}' failed to provision` +
189 (stream.statusReason
190 ? ` (${stream.statusReason.error})`
191 : ""),
192 );
193 }
194 if (stream.status !== "ACTIVE") {
195 return Effect.fail(
196 new Error(
197 `DSQL stream '${streamIdentifier}' not active (status: ${stream.status})`,
198 ),
199 );
200 }
201 return Effect.succeed(stream);
202 }),
203 Effect.retry({ schedule: policy }),
204 );
205 });
206
207 const toAttrs = (stream: dsql.GetStreamOutput) => ({
208 clusterId: stream.clusterIdentifier,
209 streamId: stream.streamIdentifier,
210 streamArn: stream.arn,
211 status: stream.status,
212 ordering: stream.ordering,
213 format: stream.format,
214 kinesisStreamArn: stream.targetDefinition?.kinesis.streamArn ?? "",
215 roleArn: stream.targetDefinition?.kinesis.roleArn ?? "",
216 });
217
218 return {
219 stables: ["clusterId", "streamId", "streamArn"],
220
221 diff: Effect.fn(function* ({ olds, news }) {
222 if (!isResolved(news)) return undefined;
223 // There is no UpdateStream — everything except tags replaces.
224 if (
225 news.clusterId !== olds?.clusterId ||
226 news.kinesisStreamArn !== olds?.kinesisStreamArn ||
227 news.roleArn !== olds?.roleArn ||
228 (news.ordering ?? defaultOrdering) !==
229 (olds?.ordering ?? defaultOrdering) ||
230 (news.format ?? defaultFormat) !== (olds?.format ?? defaultFormat)
231 ) {
232 return { action: "replace" } as const;
233 }
234 }),
235
236 read: Effect.fn(function* ({ id, output }) {
237 if (output?.streamId === undefined) return undefined;
238 const stream = yield* readStream(output.clusterId, output.streamId);
239 if (stream === undefined || stream.status === "DELETED") {
240 return undefined;
241 }
242 const tags = Object.fromEntries(
243 Object.entries(stream.tags ?? {}).filter(
244 (entry): entry is [string, string] =>
245 typeof entry[1] === "string",
246 ),
247 );
248 const attrs = toAttrs(stream);
249 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
250 }),
251
252 reconcile: Effect.fn(function* ({ id, news, output, session }) {
253 const props = news!;
254 const internalTags = yield* createInternalTags(id);
255 const desiredTags = { ...internalTags, ...props.tags };
256
257 // 1. Observe — cloud state is authoritative; output caches the id.
258 let observed =
259 output?.streamId === undefined
260 ? undefined
261 : yield* readStream(output.clusterId, output.streamId);
262
263 // 2. Ensure — create if missing. DSQL assigns the identifier.
264 let streamId = observed?.streamIdentifier ?? output?.streamId;
265 if (observed === undefined) {
266 const created = yield* dsql
267 .createStream({
268 clusterIdentifier: props.clusterId,
269 targetDefinition: {
270 kinesis: {
271 streamArn: props.kinesisStreamArn,
272 roleArn: props.roleArn,
273 },
274 },
275 ordering: props.ordering ?? defaultOrdering,
276 format: props.format ?? defaultFormat,
277 tags: desiredTags,
278 })
279 .pipe(
280 // Two transient races surface as typed errors here:
281 // - ConflictException while another cluster operation is in
282 // flight;
283 // - ValidationException "Aurora DSQL can't access the
284 // specified IAM role ..." while a freshly created service
285 // role propagates through IAM.
286 Effect.retry({
287 while: (e): boolean =>
288 e._tag === "ConflictException" ||
289 (e._tag === "ValidationException" &&
290 e.message.includes("IAM role")),
291 schedule: Schedule.max([
292 Schedule.fixed("5 seconds"),
293 Schedule.recurs(24),
294 ]),
295 }),
296 );
297 streamId = created.streamIdentifier;
298 }
299
300 // Wait for ACTIVE so consumers observe a delivering stream.
301 observed = yield* waitForActive(props.clusterId, streamId!);
302
303 // 3. Sync tags — diff against OBSERVED cloud tags.
304 const observedTags = Object.fromEntries(
305 Object.entries(observed.tags ?? {}).filter(
306 (entry): entry is [string, string] =>
307 typeof entry[1] === "string",
308 ),
309 );
310 const { upsert, removed } = diffTags(observedTags, desiredTags);
311 if (upsert.length > 0) {
312 yield* dsql.tagResource({
313 resourceArn: observed.arn,
314 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
315 });
316 }
317 if (removed.length > 0) {
318 yield* dsql.untagResource({
319 resourceArn: observed.arn,
320 tagKeys: removed,
321 });
322 }
323
324 yield* session.note(streamId!);
325 return toAttrs(observed);
326 }),
327
328 delete: Effect.fn(function* ({ output }) {
329 yield* dsql
330 .deleteStream({
331 clusterIdentifier: output.clusterId,
332 streamIdentifier: output.streamId,
333 })
334 .pipe(
335 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
336 // A stream still CREATING rejects delete with a typed
337 // conflict; retry briefly until it settles.
338 Effect.retry({
339 while: (e): boolean => e._tag === "ConflictException",
340 schedule: Schedule.max([
341 Schedule.fixed("5 seconds"),
342 Schedule.recurs(24),
343 ]),
344 }),
345 );
346 // Wait until fully gone (bounded) so the parent cluster's delete
347 // does not trip over a stream still DELETING.
348 yield* readStream(output.clusterId, output.streamId).pipe(
349 Effect.flatMap((stream) =>
350 stream === undefined || stream.status === "DELETED"
351 ? Effect.void
352 : Effect.fail(
353 new Error(
354 `DSQL stream '${output.streamId}' still ${stream.status}`,
355 ),
356 ),
357 ),
358 Effect.retry({
359 while: (e): boolean => e instanceof Error,
360 schedule: Schedule.max([
361 Schedule.fixed("5 seconds"),
362 Schedule.recurs(36),
363 ]),
364 }),
365 // Best-effort — deletion continues server-side either way.
366 Effect.catch(() => Effect.void),
367 );
368 }),
369
370 list: () =>
371 Effect.gen(function* () {
372 const clusters = yield* dsql.listClusters.items({}).pipe(
373 EStream.runCollect,
374 Effect.map((c) => Array.from(c)),
375 );
376 const nested = yield* Effect.forEach(
377 clusters,
378 (cluster) =>
379 dsql.listStreams
380 .items({ clusterIdentifier: cluster.identifier })
381 .pipe(
382 EStream.runCollect,
383 Effect.map((c) => Array.from(c)),
384 Effect.catchTag("ResourceNotFoundException", () =>
385 Effect.succeed([]),
386 ),
387 ),
388 { concurrency: 4 },
389 );
390 return yield* Effect.forEach(
391 nested.flat(),
392 (summary) =>
393 readStream(
394 summary.clusterIdentifier,
395 summary.streamIdentifier,
396 ).pipe(
397 Effect.map((stream) =>
398 stream === undefined || stream.status === "DELETED"
399 ? undefined
400 : toAttrs(stream),
401 ),
402 ),
403 { concurrency: 4 },
404 ).pipe(
405 Effect.map((attrs) =>
406 attrs.filter(
407 (a): a is NonNullable<typeof a> => a !== undefined,
408 ),
409 ),
410 );
411 }),
412 };
413 }),
414 );