Processor in action

V2 gateway requests require an API key. Set SQD_API_KEY in .env before running gateway-based examples, or pass apiKey: process.env.SQD_API_KEY in GatewaySettings.

EVM
Substrate

An end-to-end idiomatic usage of EvmBatchProcessor can be inspected in the gravatar template repository and also learned from more elaborate examples.In order to illustrate the concepts covered in the development guide, here we highlight the key steps, put together a processor configuration and a data handling definition.

Pre-requisites: NodeJS, Git, Docker, Squid CLI, any of the EVM templates.

1. Model the target schema and generate entity classes

Create or edit schema.graphql to define the target entities and relations. Consult the schema reference.Update the entity classes, start a fresh database and regenerate migrations:

npx squid-typeorm-codegen

docker compose down

docker compose up -d

rm -r db/migrations

npx squid-typeorm-migration generate

Apply the migrations with

npx squid-typeorm-migration apply

2. Generate Typescript ABI modules

Use evm-typegen to generate the facade classes, for example like this:

npx squid-evm-typegen src/abi 0x2E645469f354BB4F5c8a05B3b30A929361cf77eC#Gravity --clean

3. Configuration

See Configuration section for more details.

const processor = new EvmBatchProcessor()
  .setGateway('https://v2.archive.subsquid.io/network/ethereum-mainnet')
  .setRpcEndpoint('<my_eth_rpc_url>')
  .setFinalityConfirmation(75)
  .setBlockRange({ from: 6175243 })
  // fetch logs emitted by '0x2E645469f354BB4F5c8a05B3b30A929361cf77eC'
  // matching either `NewGravatar` or `UpdatedGravatar`
  .addLog({
    address: ['0x2E645469f354BB4F5c8a05B3b30A929361cf77eC'],
    topic0: [
      events.NewGravatar.topic,
      events.UpdatedGravatar.topic,
    ]
  })

4. Iterate over the batch items and group events

The following code snippet illustrates a typical data transformation in a batch. The strategy is to

Iterate over ctx.blocks and block.logs
Decode each log using a suitable facade class
Enrich and transform the data
Upsert arrays of entities in batches using ctx.save()

The processor.run() method the looks as follows:

processor.run(new TypeormDatabase(), async (ctx) => {
  // storing the new/updated entities in
  // an in-memory identity map
  const gravatars: Map<string, Gravatar> = new Map()
  // iterate over the data batch stored in ctx.blocks
  for (const c of ctx.blocks) {
    for (const log of c.logs) {
      // decode the log data
      const { id, owner, displayName, imageUrl } = extractData(log)
      // transform and normalize to match the target entity (Gravatar)
      gravatars.set(id.toHexString(), new Gravatar({
        id: id.toHexString(),
        owner: decodeHex(owner),
        displayName,
        imageUrl
      }))
    }
  }
  // Upsert the entities that were updated.
  // Note that store.save() automatically updates
  // the existing entities and creates new ones.
  // It splits the data into suitable chunks to
  // guarantee an adequate performance.
  await ctx.store.save([...gravatars.values()])
});

In the snippet above, we decode both NewGravatar and UpdatedGravatar with a single helper function that uses the generated events facade class:

function extractData(evmLog: any): { id: ethers.BigNumber, owner: string, displayName: string, imageUrl: string} {
  if (evmLog.topics[0] === events.NewGravatar.topic) {
    return events.NewGravatar.decode(evmLog)
  }
  if (evmLog.topics[0] === events.UpdatedGravatar.topic) {
    return events.UpdatedGravatar.decode(evmLog)
  }
  throw new Error('Unsupported topic')
}

5. Run the processor and store the transformed data into the target database

Run the processor with

node -r dotenv/config lib/main.js

The script will build the code automatically before running.In a separate terminal window, run

npx squid-graphql-server

Inspect the GraphQL API at http://localhost:4350/graphql.

An end-to-end idiomatic usage of SubstrateBatchProcessor can be inspected in the squid-substrate-template and also learned from more elaborate examples.Here we highlight the key steps and put together a configuration and a data handling definition to illustrate the concepts covered in the squid development guide.

Pre-requisites: NodeJS, Git, Docker, Squid CLI, any of the Substrate templates.

1. Set up the processor

See Setup section for more details.

src/processor.ts

import {
  SubstrateBatchProcessor
} from '@subsquid/substrate-processor'

export const processor = new SubstrateBatchProcessor()
  // set the data source
  .setGateway('https://v2.archive.subsquid.io/network/kusama')
  .setRpcEndpoint('https://kusama-rpc.dwellir.com')
  // add data requests
  .addEvent({
    name: ['Balances.Transfer'],
    extrinsic: true
  })
  .addCall({
    name: ['Balances.transfer_keep_alive'],
    extrinsic: true
  })
  // define the data
  // to be fetched for each data item
  .setFields({
    extrinsic: {
      hash: true,
      fee: true
    },
    call: {
      success: true
    },
    block: {
      timestamp: true
    }
  })

2. Define a custom data facade and extract normalized data from `BatchContext`

The following code snippet illustrates how the data is extracted and normalized into some user-specific facade interface TransferData. Note how the data items in each of the block data iterables (events, calls, extrinsics) are filtered in the batch handler to make sure they match the data requests.Type-safe access and decoding of the call and event data is done with the help of the classes generated by a suitable typegen tool, in this case squid-substrate-typegen.

src/main.ts

import {Store} from '@subsquid/typeorm-store'
import * as ss58 from '@subsquid/ss58'

import {events} from './types'
import {ProcessorContext} from './processor'

// some normalized user-define data
interface TransferData {
  id: string
  timestamp: Date | undefined
  extrinsicHash: string | undefined
  from: string
  to: string
  amount: bigint
  fee?: bigint
}

// extract and normalize
function getTransfers(ctx: ProcessorContext<Store>): TransferData[] {
  let transfers: TransferData[] = []
  for (let block of ctx.blocks) {
    for (let event of block.events) {
      if (event.name==='Balances.Transfer') {
        // decode the event data with runtime-versioned
        // classes generated by typegen
        let rec: {from: string, to: string, amount: bigint}
        if (events.balances.transfer.v1020.is(event)) {
          let [from, to, amount,] =
            events.balances.transfer.v1020.decode(event)
          rec = {from, to, amount}
        } else if (events.balances.transfer.v1050.is(event)) {
          let [from, to, amount] =
            events.balances.transfer.v1050.decode(event)
          rec = {from, to, amount}
        } else {
          rec = events.balances.transfer.v9130.decode(event)
        }
        // extract and normalize the `item.event` data
        let timestamp =
          block.header.timestamp == null ?
          undefined :
          new Date(block.header.timestamp)
        transfers.push({
          id: event.id,
          timestamp,
          extrinsicHash: event.extrinsic?.hash,
          from: ss58.codec('kusama').encode(rec.from),
          to: ss58.codec('kusama').encode(rec.to),
          amount: rec.amount,
          fee: event.extrinsic?.fee || 0n
        })
      }
    }
    for (let call of block.calls) {
      if (call.name==='Balances.transfer_keep_alive') {
        // extranct and normalize the call data
        transfer.push({
          /* some data manipulation on the call data */
        })
      }
    }
  }
  return transfers
}

3. Run the processor and store the transformed data into the target database

The snippet below assumes that we are using TypeormDatabase and that the database schema together with the model entity types has already been prepared. Consult the Schema file section for more details.

src/main.ts

import {TypeormDatabase} from '@subsquid/typeorm-store'
// an entity type generated from the schema file
import {Transfer} from './model'
import {processor} from './processor'

// ... `getTransfers(Ctx)` definition

processor.run(new TypeormDatabase(), async ctx => {
  // get the normalized data from the context
  let transfersData = getTransfers(ctx)

  // an array of entity class instances
  let transfers: Transfer[] = []

  for (let t of transfersData) {
    let {id, blockNumber, timestamp, extrinsicHash, amount, fee} = t

    // join some extra data
    let from = getAccount(accounts, t.from)
    let to = getAccount(accounts, t.to)

    // populate entity class instances
    // with the extracted data
    transfers.push(new Transfer({
      id,
      blockNumber,
      timestamp,
      extrinsicHash,
      from,
      to,
      amount,
      fee
    }))
  }

  // persist an array of entities
  // in a single statement
  await ctx.store.insert(transfers)
})

Overview

Pipes SDK

Squid SDK

Alternative clients

1. Model the target schema and generate entity classes

2. Generate Typescript ABI modules

3. Configuration

4. Iterate over the batch items and group events

5. Run the processor and store the transformed data into the target database

1. Set up the processor

2. Define a custom data facade and extract normalized data from `BatchContext`

3. Run the processor and store the transformed data into the target database

​1. Model the target schema and generate entity classes

​2. Generate Typescript ABI modules

​3. Configuration

​4. Iterate over the batch items and group events

​5. Run the processor and store the transformed data into the target database

​1. Set up the processor

​2. Define a custom data facade and extract normalized data from BatchContext

​3. Run the processor and store the transformed data into the target database

1. Model the target schema and generate entity classes

2. Generate Typescript ABI modules

3. Configuration

4. Iterate over the batch items and group events

5. Run the processor and store the transformed data into the target database

1. Set up the processor

2. Define a custom data facade and extract normalized data from `BatchContext`

3. Run the processor and store the transformed data into the target database