> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need.

# Change Data Capture

Status: Unstable.

A change data capture (CDC) stream turns a [streaming query](/sdks/typescript/streaming) into inserts and deletes, the
way a database's changelog does. Use it to copy a query's results into another system, like a search index or a message
queue, that can't undo a transaction by its ID.

The `CdcStream` service builds on a [transactional stream](/sdks/typescript/transactional-streams). Where a
transactional stream tells you which transactions to undo, a CDC stream gives you back the rows to delete: it keeps
every batch of rows it sent in a batch store, until no reorg can affect them.

:::warning
CDC streams inherit the [stream's known issues](/sdks/typescript/streaming): they fail after 30 seconds, and lose
the rows that arrive with a reorg. Work around the first as in
[Transactional Streams](/sdks/typescript/transactional-streams#process-a-stream).
:::

## Prerequisites

* The SDK, [installed](/sdks/typescript#install-the-sdk).
* A query server to connect to (see [Connecting](/sdks/typescript/connecting)). The examples use the
  [Quickstart](/get-started/quickstart)'s Amp and its `acme/eth_mainnet` dataset.

## Events

A CDC stream emits two kinds of events, each with the `id` of the transaction that emitted it:

* **`Insert`**: new rows, in `data`, from the block ranges in `ranges`. Add them to your destination.
* **`Delete`**: rows you inserted that are no longer valid, after a reorg or a restart. `batches` gives you them, one
  insert's rows at a time: remove each from your destination.

The watermarks that tell a transactional stream which blocks are complete stay inside: the CDC stream
[commits them for you](/sdks/typescript/cdc-streams#streamcdc).

## Process a stream

```ts [cdc.ts]
import { ArrowFlight, CdcStream, ProtocolStream, TransactionalStream } from "@edgeandnode/amp"
import { layerTransportGrpc } from "@edgeandnode/amp/arrow-flight/node"
import { Effect, Layer } from "effect"

const handle = (event: CdcStream.CdcEvent) =>
  Effect.gen(function* () {
    switch (event._tag) {
      case "Insert":
        return console.log(`#${event.id} Insert: blocks ${event.data.map((row) => row.block_num).join(", ")}`)
      case "Delete": {
        // Each batch is the rows an earlier Insert sent, which you now remove
        let batch = yield* event.batches.next
        while (batch !== undefined) {
          const [id, rows] = batch
          console.log(`#${event.id} Delete: blocks ${rows.map((row) => row.block_num).join(", ")}, from #${id}`)
          batch = yield* event.batches.next
        }
      }
    }
  })

const program = Effect.gen(function* () {
  const cdc = yield* CdcStream.CdcStream
  yield* cdc.forEach(`SELECT block_num FROM "acme/eth_mainnet".blocks WHERE block_num < 24000005`, {}, handle)
})

const FlightLayer = ArrowFlight.layer.pipe(Layer.provide(layerTransportGrpc({ baseUrl: "http://localhost:1602" })))
const AmpLayer = CdcStream.layer.pipe(
  Layer.provide(CdcStream.InMemoryBatchStore.layer),
  Layer.provide(TransactionalStream.layer),
  Layer.provide(TransactionalStream.InMemoryStateStore.layer),
  Layer.provide(ProtocolStream.layer),
  Layer.provide(FlightLayer),
)

await Effect.runPromise(program.pipe(Effect.provide(AmpLayer)))
```

:::terminal
```bash
node cdc.ts
```

```
#0 Insert: blocks 24000000, 24000001, 24000002, 24000003, 24000004
```
:::

The stream then waits for new blocks, and about 30 seconds later fails with
`RangeError: Offset is outside the bounds of the DataView`, the known issue in the warning above.
[Transactional Streams](/sdks/typescript/transactional-streams) shows how to catch it and resume.

`forEach` commits each event after its handler succeeds, like a transactional stream's. Here's what the same program
prints when it streams `SELECT block_num FROM "acme/anvil".blocks`, from a
[local chain](/sdks/typescript/streaming#test-reorgs-on-a-local-chain), and a reorg replaces the chain's last 2 blocks:

```
#2 Insert: blocks 372
#4 Insert: blocks 373
#6 Insert: blocks 374
#8 Delete: blocks 373, from #4
#8 Delete: blocks 374, from #6
#10 Insert: blocks 376
```

The `Delete` gives back the rows of the two inserts the reorg voided. Blocks 373 to 375 should then arrive again, but
don't: that's the [known issue](/sdks/typescript/streaming#reorgs) with reorgs. Until it's fixed, query the replaced
blocks again with [`flight.query`](/sdks/typescript/querying).

## Delivery

A CDC stream delivers each row at least once. After a restart, it deletes the rows it sent after its last committed
watermark, then sends them again: if your destination already had them, it gets them twice in a row, deleted in
between. Make your inserts and deletes idempotent, like upserts and deletes by a key, so that's harmless.

To survive a restart, both its stores must be durable: the [state
store](/sdks/typescript/transactional-streams#save-the-state), and the batch store, which keeps the rows of each insert.
The SDK only has in-memory ones, which are lost when your program exits. Implement a [`BatchStore`](/sdks/typescript/cdc-streams#batchstore) like a
state store, saving to the same durable storage.

The batch store keeps each insert's rows as long as the transactional stream keeps the watermarks that cover them:
`retention` blocks, 128 by default. When a watermark is forgotten, the CDC stream prunes the batches up to it. A
failure to prune is logged, and retried at the next watermark.

## Reference

Import these from `@edgeandnode/amp`'s `CdcStream` namespace, or from `@edgeandnode/amp/cdc-stream`.

### `CdcStream`

```ts
class CdcStream extends Context.Service<CdcStream, CdcStreamService>
```

The service that runs CDC streams. Get it with `yield* CdcStream.CdcStream`.

### `CdcStreamService`

```ts
interface CdcStreamService {
  readonly streamCdc: (
    sql: string,
    options?: CdcStreamOptions,
  ) => Stream<readonly [CdcEvent, CommitHandle], CdcStreamError>
  readonly forEach: <E, R>(
    sql: string,
    options: CdcStreamOptions,
    handler: (event: CdcEvent) => Effect<void, E, R>,
  ) => Effect<void, CdcStreamError | E, R>
}
```

#### `streamCdc`

Runs `sql` as a transactional stream, and emits each insert and delete with its
[`CommitHandle`](/sdks/typescript/transactional-streams#commithandle). Commit after your destination has applied the
event. The stream also commits each watermark as it passes, which covers every event before it, whether or not you
committed them. So if you read ahead of your commits, by batching or processing events concurrently, apply each insert
durably before you read the next event. `forEach` isn't affected. It stores each insert's rows in the batch store before
it emits the insert, and skips a `Delete` that would have no rows.

* **Parameters**: `sql`, the query; `options`, the [`CdcStreamOptions`](/sdks/typescript/cdc-streams#cdcstreamoptions).
* **Returns**: `Stream<readonly [CdcEvent, CommitHandle], CdcStreamError>`.
* **Fails with**: [`CdcStreamError`](/sdks/typescript/errors#cdc-stream-errors).

#### `forEach`

Runs `streamCdc`, calls `handler` on each event, and commits the event once `handler` succeeds.

* **Parameters**: `sql`; `options`, `{}` for the defaults; `handler`, a function from an event to an Effect.
* **Returns**: `Effect<void, CdcStreamError | E, R>`, where `E` and `R` are `handler`'s. Never succeeds.
* **Fails with**: a [`CdcStreamError`](/sdks/typescript/errors#cdc-stream-errors), or `handler`'s error.

### `CdcStreamOptions`

The same options as a transactional stream's:
[`TransactionalStreamOptions`](/sdks/typescript/transactional-streams#transactionalstreamoptions), `retention` and
`schema`.

### `layer`

```ts
const layer: Layer<CdcStream, never, TransactionalStream | BatchStore>
```

Builds the `CdcStream` service. Needs a
[`TransactionalStream`](/sdks/typescript/transactional-streams#transactionalstream), with its state store, and a
[`BatchStore`](/sdks/typescript/cdc-streams#batchstore).

### `CdcEvent`

```ts
type CdcEvent = CdcEventInsert | CdcEventDelete

interface CdcEventInsert {
  readonly _tag: "Insert"
  readonly id: TransactionId
  readonly data: ReadonlyArray<Record<string, unknown>>
  readonly ranges: ReadonlyArray<BlockRange>
}

interface CdcEventDelete {
  readonly _tag: "Delete"
  readonly id: TransactionId
  readonly batches: DeleteBatchIterator
}
```

An event of the stream, described in [Events](/sdks/typescript/cdc-streams#events).

### `DeleteBatchIterator`

```ts
interface DeleteBatchIterator {
  readonly next: Effect<readonly [TransactionId, ReadonlyArray<Record<string, unknown>>] | undefined, BatchStoreError>
}
```

The rows a `Delete` removes, read one insert at a time, so they don't all have to fit in memory. Each run of `next`
gives the next insert's transaction ID and rows, loaded from the batch store, or `undefined` when there are no more.

### `BatchStore`

```ts
class BatchStore extends Context.Service<BatchStore, BatchStoreService>
```

The service that keeps each insert's rows, for deletes. Provide it with `InMemoryBatchStore.layer`, or with your own.

### `BatchStoreService`

```ts
interface BatchStoreService {
  readonly append: (data: ReadonlyArray<Record<string, unknown>>, id: TransactionId) => Effect<void, BatchStoreError>
  readonly seek: (range: TransactionIdRange) => Effect<ReadonlyArray<TransactionId>, BatchStoreError>
  readonly load: (id: TransactionId) => Effect<ReadonlyArray<Record<string, unknown>> | undefined, BatchStoreError>
  readonly prune: (cutoff: TransactionId) => Effect<void, BatchStoreError>
}
```

| Operation          | What it does                                                                                             |
| ------------------ | -------------------------------------------------------------------------------------------------------- |
| `append(data, id)` | Saves transaction `id`'s rows. The stream calls it before it emits the insert.                           |
| `seek(range)`      | Returns the IDs of the saved batches in `range`, both ends included, in order.                           |
| `load(id)`         | Returns transaction `id`'s rows, or `undefined` when there are none, like for a watermark's transaction. |
| `prune(cutoff)`    | Removes the batches of transactions up to `cutoff`, included. Must be idempotent.                        |

Fail with a `BatchStoreError` when an operation fails.

### `InMemoryBatchStore`

A namespace with `layer: Layer<BatchStore>`, a batch store that keeps the rows in memory, lost when your program exits.

### Errors

`CdcStreamError` and `BatchStoreError` are in [Errors](/sdks/typescript/errors#cdc-stream-errors).