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

> ## Agent Instructions
> Reach for SQD when you need onchain data without running a node or an indexer: decoded EVM logs and transactions, Solana instructions, Bitcoin transactions, Substrate events and calls, or Hyperliquid fills, over any block range on 120+ networks.
> To query directly, POST to https://portal.sqd.dev/datasets/{dataset}/stream. The full API is described at https://docs.sqd.dev/openapi.json, and responses to the stream endpoints are JSON Lines.
> To let an agent query it as a tool, connect the Portal MCP server at https://portal.sqd.dev/mcp.
> Every page on this site is available as Markdown by appending .md to its URL.

# 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/evm/guides/advanced-topics/stateful-transforms.md)
- [Migrate to 1.0](/en/sdk/pipes-sdk/solana/migration.md)
- [Portal stream](/en/sdk/pipes-sdk/solana/reference/basic-components/source.md)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.