Stream

Source: src/AWS/DSQL/Stream.ts

A change data capture (CDC) stream on an Aurora DSQL cluster — delivers committed row-level changes (Debezium-shaped JSON envelopes) to an Amazon Kinesis data stream.

Creation is asynchronous (CREATING -> ACTIVE, typically one to three minutes); the provider waits for ACTIVE (bounded) before returning. A stream has no update operation — every property except tags replaces it.

Functions consume the change records through the existing Kinesis event source on the target stream; DSQL itself never invokes compute directly.

Creating a CDC Stream

Stream Cluster Changes into Kinesis

const cluster = yield* DSQL.Cluster("AppDb", {});
const target = yield* Kinesis.Stream("Changes", {
  streamMode: "ON_DEMAND",
  maxRecordSizeInKiB: 10240,
});
const role = yield* IAM.Role("CdcRole", {
  assumeRolePolicyDocument: JSON.stringify({
    Version: "2012-10-17",
    Statement: [
      {
        Effect: "Allow",
        Principal: { Service: "dsql.amazonaws.com" },
        Action: "sts:AssumeRole",
      },
    ],
  }),
  inlinePolicies: {
    kinesis: JSON.stringify({
      Version: "2012-10-17",
      Statement: [
        {
          Effect: "Allow",
          Action: [
            "kinesis:PutRecord",
            "kinesis:PutRecords",
            "kinesis:DescribeStreamSummary",
            "kinesis:ListShards",
          ],
          Resource: target.streamArn,
        },
      ],
    }),
  },
});
const cdc = yield* DSQL.Stream("Cdc", {
  clusterId: cluster.clusterId,
  kinesisStreamArn: target.streamArn,
  roleArn: role.roleArn,
});

Consume Change Records with a Function

// DSQL delivers into the Kinesis stream; consume it with the
// Kinesis event source on the target stream.
yield* Kinesis.consume(target, (records) =>
  Effect.forEach(records, (record) => handleChange(record)),
);