> ## 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.

# Transformer

> Use createTransformer and transforms in a Solana Pipes SDK pipeline.

## createTransformer

Construct a whole-pipe transformer.

```ts theme={"system"}
createTransformer<I, O>(config: TransformerOptions<I, O>): Transformer<I, O>
```

**Config fields:**

* `transform`: (required) `(data: I, ctx: BatchContext) => O | Promise<O>`. Called once per batch.
* `start`: (optional) `(ctx: StartContext) => void | Promise<void>`. Called once when the pipe starts. Use this to load state, warm up caches, or query the portal for historical data before the main stream begins.
* `stop`: (optional) `(ctx: StopContext) => void | Promise<void>`. Called once when the pipe stops.
* `rollback`: (optional) `(cursor: BlockCursor, ctx: HookContext) => void | Promise<void>`. Called after the source detects a chain reorg and resolves the safe cursor. `cursor` identifies the last safe slot; undo any internal state above it. See [Fork handling](../../guides/architecture-deep-dives/fork-handling).
* `profiler`: (optional) `{ name: string; hidden?: boolean }`. Overrides the transformer's node name in the [profiler](../../guides/advanced-topics/profiling) tree.

**Example:**

```ts theme={"system"}
const transformer = createTransformer({
  transform: async (data, ctx) => {
    ctx.logger.info({ slot: ctx.stream.state.current.number }, 'batch')
    return data.map((b) => b.instructions)
  },
})
```

## Context variables

Each callback receives a context object. The fields differ by callback.

### `transform(data, ctx: BatchContext)`

`ctx` is the full per-batch context. Fields:

| Field      | Type            | Description                                                                                                                                                                                    |
| ---------- | --------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `id`       | `string`        | Pipeline ID — the `id` passed to `solanaPortalStream()`.                                                                                                                                       |
| `logger`   | `Logger`        | Pino-compatible logger scoped to this batch. Defaults to the source-level logger.                                                                                                              |
| `metrics`  | `Metrics`       | Prometheus metrics registry. Use `ctx.metrics.counter()`, `.gauge()`, `.histogram()`, `.summary()` to register and update custom metrics. See [Metrics](../../guides/advanced-topics/metrics). |
| `profiler` | `Profiler`      | Open a span with `ctx.profiler.start('label')`. See [Profiling](../../guides/advanced-topics/profiling).                                                                                       |
| `stream`   | `StreamInfo`    | Per-stream state (see below).                                                                                                                                                                  |
| `batch`    | `BatchMetadata` | Per-batch volume info (see below).                                                                                                                                                             |

#### `ctx.stream: StreamInfo`

| Field                 | Type                        | Description                                                                               |
| --------------------- | --------------------------- | ----------------------------------------------------------------------------------------- |
| `dataset`             | `ApiDataset`                | Dataset metadata returned by the portal (chain name, genesis, tier).                      |
| `head.finalized`      | `BlockCursor \| undefined`  | Current finalized head known to the portal, if advertised.                                |
| `head.latest`         | `BlockCursor \| undefined`  | Current unfinalized head.                                                                 |
| `state.initial`       | `number`                    | First slot the stream was configured to read.                                             |
| `state.last`          | `number`                    | Last slot the stream intends to read (often `Infinity`).                                  |
| `state.current`       | `BlockCursor`               | Latest slot in this batch. Cursor has `{ number, hash?, timestamp? }`.                    |
| `state.rollbackChain` | `BlockCursor[]`             | Unfinalized-chain tail — cursors the stream will need to roll back if a fork is detected. |
| `progress`            | `ProgressEvent['progress']` | Progress metrics when `progress` is configured on the source; otherwise undefined.        |
| `query`               | `{ url, hash, raw }`        | Debug info for the portal query feeding this batch.                                       |

#### `ctx.batch: BatchMetadata`

| Field                 | Type                     | Description                                                             |
| --------------------- | ------------------------ | ----------------------------------------------------------------------- |
| `blocksCount`         | `number`                 | Number of slots in this batch.                                          |
| `bytesSize`           | `number`                 | Compressed payload size received from the portal.                       |
| `requests`            | `Record<number, number>` | Map of HTTP status code → number of responses that produced this batch. |
| `lastBlockReceivedAt` | `Date`                   | Wall-clock time the last block was received.                            |

### `start(ctx: StartContext)`

Fired once, before any batch. Use to warm up caches or run one-off queries.

| Field           | Type                       | Description                                                                 |
| --------------- | -------------------------- | --------------------------------------------------------------------------- |
| `id`            | `string`                   | Pipeline ID.                                                                |
| `logger`        | `Logger`                   | Same as in `BatchContext`.                                                  |
| `metrics`       | `Metrics`                  | Same as in `BatchContext`.                                                  |
| `portal`        | `PortalClient`             | Live portal client. Use `portal.getStream(query)` for warm-up reads.        |
| `state.initial` | `number`                   | First slot the stream was configured to read.                               |
| `state.current` | `BlockCursor \| undefined` | Cursor persisted by the previous run, if any. `undefined` on a fresh start. |

### `rollback(cursor, ctx: HookContext)`

Fired before the next batch whenever a reorg is detected, after the safe cursor has been resolved. `cursor` is the last slot to keep; drop state produced for anything after it.

| Field      | Type       | Description                |
| ---------- | ---------- | -------------------------- |
| `logger`   | `Logger`   | Same as in `BatchContext`. |
| `profiler` | `Profiler` | Same as in `BatchContext`. |

### `stop(ctx: StopContext)`

Fired once when the pipe stops.

| Field    | Type     | Description                |
| -------- | -------- | -------------------------- |
| `logger` | `Logger` | Same as in `BatchContext`. |

## The `OutputOf` type helper

`OutputOf<T>` infers the output type of a query-transform combo, a transformer, a transform function, or a record of named outputs. Use it to type downstream transform functions without spelling out intermediate types:

```ts theme={"system"}
import { OutputOf } from '@subsquid/pipes'
import { solanaQuery } from '@subsquid/pipes/solana'

function myDecoder() {
  return solanaQuery()
    .addFields({ block: { number: true, timestamp: true } })
    .includeAllBlocks()
    .build()
    .pipe((blocks) => blocks.map((b) => ({ slot: b.header.number })))
}

type MyDecoderOut = OutputOf<typeof myDecoder> // { slot: number }[]

function myTransformation(): (data: MyDecoderOut) => { s: number }[] {
  return (data) => data.map((i) => ({ s: i.slot }))
}
```

It also works on a whole `outputs` record: `OutputOf<typeof outputs>` yields `{ name1: ..., name2: ... }`.


## Related topics

- [Transformer](/en/sdk/pipes-sdk/evm/reference/basic-components/transformer.md)
- [Factory transformers](/en/sdk/pipes-sdk/evm/guides/advanced-topics/factory-transformers.md)
- [Stateful transforms](/en/sdk/pipes-sdk/solana/guides/advanced-topics/stateful-transforms.md)
- [Migrate to 1.0](/en/sdk/pipes-sdk/solana/migration.md)
- [Portal stream](/en/sdk/pipes-sdk/evm/reference/basic-components/source.md)