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: StartCtx) => 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: StopCtx) => void | Promise<void>. Called once when the pipe stops.
fork: (optional) (cursor: BlockCursor, ctx: Ctx) => void | Promise<void>. Called before the next batch whenever the source detects a chain reorg. cursor identifies the last safe block. 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`	`BatchStreamContext`	Per-stream state (see below).
`batch`	`BatchMetadata`	Per-batch volume info (see below).

`ctx.stream: BatchStreamContext`

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: StartCtx)`

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.

`fork(cursor, ctx: Ctx)`

Fired before the next batch whenever a reorg is detected. 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: StopCtx)`

Fired once when the pipe stops.

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

Overview

Pipes SDK

Squid SDK

Alternative clients

createTransformer

Context variables

`transform(data, ctx: BatchContext)`

`ctx.stream: BatchStreamContext`

`ctx.batch: BatchMetadata`

`start(ctx: StartCtx)`

`fork(cursor, ctx: Ctx)`

`stop(ctx: StopCtx)`

​createTransformer

​Context variables

​transform(data, ctx: BatchContext)

​ctx.stream: BatchStreamContext

​ctx.batch: BatchMetadata

​start(ctx: StartCtx)

​fork(cursor, ctx: Ctx)

​stop(ctx: StopCtx)

createTransformer

Context variables

`transform(data, ctx: BatchContext)`

`ctx.stream: BatchStreamContext`

`ctx.batch: BatchMetadata`

`start(ctx: StartCtx)`

`fork(cursor, ctx: Ctx)`

`stop(ctx: StopCtx)`