Transformer

createTransformer

Construct a whole-pipe transformer.

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 block; undo any internal state above it. See Fork handling.
profiler: (optional) { name: string; hidden?: boolean }. Overrides the transformer’s node name in the profiler tree.

Example:

const transformer = createTransformer({
  transform: async (data, ctx) => {
    ctx.logger.info({ block: ctx.stream.state.current.number }, 'batch')
    return data.map((b) => b.logs)
  },
})

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 `evmPortalStream()`.
`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.
`profiler`	`Profiler`	Open a span with `ctx.profiler.start('label')`. See 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 block number the stream was configured to read.
`state.last`	`number`	Last block number the stream intends to read (often `Infinity`).
`state.current`	`BlockCursor`	Latest block 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 blocks 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 block 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 block 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:

import { OutputOf } from '@subsquid/pipes'
import { evmQuery } from '@subsquid/pipes/evm'

function myDecoder() {
  return evmQuery()
    .addFields({
      block: { timestamp: true },
      log: { address: true, transactionHash: true },
    })
    .build()
    .pipe((blocks) => blocks.map((b) => ({ timestamp: b.header.timestamp })))
}

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

function myTransformation(): (data: MyDecoderOut) => { ts: number }[] {
  return (data) => data.map((i) => ({ ts: i.timestamp }))
}

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

Overview

Pipes SDK

Squid SDK

Alternative clients

createTransformer

Context variables

`transform(data, ctx: BatchContext)`

`ctx.stream: StreamInfo`

`ctx.batch: BatchMetadata`

`start(ctx: StartContext)`

`rollback(cursor, ctx: HookContext)`

`stop(ctx: StopContext)`

The `OutputOf` type helper

​createTransformer

​Context variables

​transform(data, ctx: BatchContext)

​ctx.stream: StreamInfo

​ctx.batch: BatchMetadata

​start(ctx: StartContext)

​rollback(cursor, ctx: HookContext)

​stop(ctx: StopContext)

​The OutputOf type helper

createTransformer

Context variables

`transform(data, ctx: BatchContext)`

`ctx.stream: StreamInfo`

`ctx.batch: BatchMetadata`

`start(ctx: StartContext)`

`rollback(cursor, ctx: HookContext)`

`stop(ctx: StopContext)`

The `OutputOf` type helper