> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sqd.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# createTarget

> Use createTarget in an EVM Pipes SDK pipeline.

Build a custom data target. A target drains batches from the pipe and is responsible for persisting them.

```ts theme={"system"}
createTarget<In>(config: Target<In>): Target<In>
```

**Config fields:**

* `write`: (required) Async function `({ read, logger }) => Promise<void>`. Iterate the stream by calling `read()` and consuming `{ data, ctx }` batches. The function returns when the stream ends.
* `resolveFork`: (optional) `(canonicalBlocks: BlockCursor[]) => Promise<BlockCursor | null>`. Called when the source detects a chain reorg. `canonicalBlocks` is the portal's view of the canonical chain (a.k.a. `previousBlocks` in Portal API `/stream` 409 responses). Find the common ancestor, roll back everything persisted above it, and return the cursor to resume from — or `null` if no common ancestor can be determined (the stream will throw). See [Fork handling](../../../guides/architecture-deep-dives/fork-handling). You don't need this callback when the source is configured to [read only finalized blocks](../source#finalized-blocks).

## The `write` context

```ts theme={"system"}
type WriteCtx<In> = {
  read: (cursor?: BlockCursor) => AsyncIterableIterator<PortalBatch<In>>
  logger: Logger
}
```

| Field    | Description                                                                                                            |
| -------- | ---------------------------------------------------------------------------------------------------------------------- |
| `read`   | Opens an async iterator over pipeline batches. Pass `cursor` to resume from a specific block the target has persisted. |
| `logger` | Pino-compatible logger scoped to this target.                                                                          |

## Per-batch context (`ctx`)

Each `{ data, ctx }` yielded by `read()` carries the same `BatchContext` that transformers receive. Fields:

| Field                        | Type                        | Description                                                                                                 |
| ---------------------------- | --------------------------- | ----------------------------------------------------------------------------------------------------------- |
| `id`                         | `string`                    | Pipeline ID — the `id` passed to `evmPortalStream()`.                                                       |
| `logger`                     | `Logger`                    | Batch-scoped logger.                                                                                        |
| `metrics`                    | `Metrics`                   | Prometheus metrics registry. See [Metrics](../../../guides/advanced-topics/metrics).                        |
| `profiler`                   | `Profiler`                  | Open a span with `ctx.profiler.start('label')`. See [Profiling](../../../guides/advanced-topics/profiling). |
| `stream.dataset`             | `ApiDataset`                | Dataset metadata.                                                                                           |
| `stream.head.finalized`      | `BlockCursor \| undefined`  | Current finalized head.                                                                                     |
| `stream.head.latest`         | `BlockCursor \| undefined`  | Current unfinalized head.                                                                                   |
| `stream.state.initial`       | `number`                    | First block number the stream was configured to read.                                                       |
| `stream.state.last`          | `number`                    | Last block number the stream intends to read.                                                               |
| `stream.state.current`       | `BlockCursor`               | Latest block in this batch.                                                                                 |
| `stream.state.rollbackChain` | `BlockCursor[]`             | Tail of unfinalized cursors subject to rollback.                                                            |
| `stream.progress`            | `ProgressEvent['progress']` | Progress metrics when `progress` is enabled.                                                                |
| `stream.query`               | `{ url, hash, raw }`        | Portal query details for the batch.                                                                         |
| `batch.blocksCount`          | `number`                    | Number of blocks in this batch.                                                                             |
| `batch.bytesSize`            | `number`                    | Compressed payload size received from the portal.                                                           |
| `batch.requests`             | `Record<number, number>`    | Map of HTTP status code → count of responses that produced this batch.                                      |
| `batch.lastBlockReceivedAt`  | `Date`                      | Wall-clock time the last block was received.                                                                |

## Example

```ts theme={"system"}
const target = createTarget({
  write: async ({ read, logger }) => {
    for await (const { data, ctx } of read()) {
      const span = ctx.profiler.start('save')
      await database.save(data)
      span.end()
      logger.info(
        { block: ctx.stream.state.current.number, rows: ctx.batch.blocksCount },
        'saved batch',
      )
    }
  },
  resolveFork: async (canonicalBlocks) => {
    // Return a cursor from your persisted state; null to fail hard.
    return canonicalBlocks[canonicalBlocks.length - 1] ?? null
  },
})
```

## Resuming from a persisted cursor

Stateful targets typically persist a cursor and resume from it on restart:

```ts theme={"system"}
createTarget({
  write: async ({ read, logger }) => {
    const lastSaved = await database.getCursor() // BlockCursor | undefined
    for await (const { data, ctx } of read(lastSaved)) {
      await database.save(data)
      await database.saveCursor(ctx.stream.state.current)
    }
  },
})
```


## Related topics

- [createTarget](/en/sdk/pipes-sdk/solana/reference/basic-components/target/create-target.md)
- [Cursor management](/en/sdk/pipes-sdk/solana/guides/architecture-deep-dives/cursor-management.md)
- [Developing pipes](/en/sdk/pipes-sdk/solana/guides/basic-development/flow.md)
- [Migrate to 1.0](/en/sdk/pipes-sdk/evm/migration.md)
- [Choosing your tool](/en/sdk/options-comparison.md)