Index Crust storage activity

This tutorial builds a Substrate squid that indexes three kinds of Crust activity:

files added to or deleted from storage worker reports;
storage orders placed by accounts;
storage groups joined by accounts.

The example uses a bounded historical range so you can reproduce the result and let the processor exit cleanly.

Prerequisites

Node.js 20 or newer
Docker with Docker Compose
An SQD Network API key

Scaffold the project with the latest CLI

Install the latest Squid CLI, then create a project from the Substrate template:

npm install --global @subsquid/cli@latest
sqd init substrate-crust-tutorial --template substrate --no-interactive
cd substrate-crust-tutorial

Install the template dependencies and update the packages used in this tutorial to their latest releases:

npm install
npm install @subsquid/substrate-processor@latest @subsquid/typeorm-store@latest @subsquid/typeorm-migration@latest
npm install --save-dev @subsquid/substrate-typegen@latest @subsquid/typeorm-codegen@latest

Configure gateway and RPC access

Add your SQD Network API key and the Crust RPC URL to .env:

SQD_API_KEY=<YOUR_SQD_NETWORK_API_KEY>
RPC_CRUST_HTTP=https://crust.api.onfinality.io/public

The API key authenticates requests to the v2 SQD gateway. The processor uses the RPC endpoint to obtain chain metadata.

Define the entities

Replace schema.graphql with the following schema:

schema.graphql

type Account @entity {
  id: ID!
  workReports: [WorkReport!]! @derivedFrom(field: "account")
  joinGroups: [JoinGroup!]! @derivedFrom(field: "member")
  storageOrders: [StorageOrder!]! @derivedFrom(field: "account")
}

type WorkReport @entity {
  id: ID!
  account: Account!
  addedFiles: [[String!]]
  deletedFiles: [[String!]]
  extrinsicHash: String
  createdAt: DateTime
  blockNum: Int!
}

type JoinGroup @entity {
  id: ID!
  member: Account!
  owner: String!
  extrinsicHash: String
  createdAt: DateTime
  blockNum: Int!
}

type StorageOrder @entity {
  id: ID!
  account: Account!
  fileCid: String!
  extrinsicHash: String
  createdAt: DateTime
  blockNum: Int!
}

Generate the TypeORM entity classes:

npx squid-typeorm-codegen

Generate current Crust event types

Replace typegen.json with this configuration:

typegen.json

{
  "outDir": "src/types",
  "specVersions": "https://v2.archive.subsquid.io/metadata/crust",
  "pallets": {
    "Swork": {
      "events": [
        "WorksReportSuccess",
        "JoinGroupSuccess"
      ]
    },
    "Market": {
      "events": [
        "FileSuccess"
      ]
    }
  }
}

Run Substrate typegen:

npx squid-substrate-typegen ./typegen.json

The generated wrappers reflect the actual Crust runtime types for each event.

Configure the processor

Replace src/processor.ts with the following code:

src/processor.ts

import {
  DataHandlerContext,
  SubstrateBatchProcessor,
  SubstrateBatchProcessorFields,
} from '@subsquid/substrate-processor'
import {events} from './types'

export const processor = new SubstrateBatchProcessor()
  .setGateway({
    url: 'https://v2.archive.subsquid.io/network/crust',
    apiKey: process.env.SQD_API_KEY,
  })
  .setRpcEndpoint({
    url: process.env.RPC_CRUST_HTTP ?? 'https://crust.api.onfinality.io/public',
    rateLimit: 10,
  })
  .setRpcDataIngestionSettings({disabled: true})
  .setBlockRange({from: 584_300, to: 589_350})
  .addEvent({
    name: [
      events.market.fileSuccess.name,
      events.swork.joinGroupSuccess.name,
      events.swork.worksReportSuccess.name,
    ],
    call: true,
    extrinsic: true,
  })
  .setFields({
    block: {timestamp: true},
    event: {args: true},
    call: {args: true},
    extrinsic: {hash: true},
  })

type Fields = SubstrateBatchProcessorFields<typeof processor>
export type ProcessorContext<Store> = DataHandlerContext<Store, Fields>

Selecting call.args is required because the added and deleted file lists belong to the Swork.report_works call. The bounded range contains all three requested event types.

To follow the chain continuously, remove the upper to bound and .setRpcDataIngestionSettings({disabled: true}). Keep a reliable RPC endpoint configured for real-time ingestion.

Decode and persist each event

Replace src/main.ts with this batch handler:

src/main.ts

import {Store, TypeormDatabase} from '@subsquid/typeorm-store'
import * as ss58 from '@subsquid/ss58'
import {In} from 'typeorm'
import {Account, JoinGroup, StorageOrder, WorkReport} from './model'
import {processor, ProcessorContext} from './processor'
import {events} from './types'

type EntityWithAccount<T> = [entity: T, accountId: string]

interface EventsInfo {
  joinGroups: EntityWithAccount<JoinGroup>[]
  storageOrders: EntityWithAccount<StorageOrder>[]
  workReports: EntityWithAccount<WorkReport>[]
  accountIds: Set<string>
}

processor.run(new TypeormDatabase(), async (ctx) => {
  const eventsInfo = getEventsInfo(ctx)
  const accounts = await ctx.store
    .findBy(Account, {id: In([...eventsInfo.accountIds])})
    .then((rows) => new Map(rows.map((account) => [account.id, account])))

  for (const accountId of eventsInfo.accountIds) {
    if (!accounts.has(accountId)) {
      accounts.set(accountId, new Account({id: accountId}))
    }
  }

  for (const [joinGroup, accountId] of eventsInfo.joinGroups) {
    joinGroup.member = getAccount(accounts, accountId)
  }
  for (const [storageOrder, accountId] of eventsInfo.storageOrders) {
    storageOrder.account = getAccount(accounts, accountId)
  }
  for (const [workReport, accountId] of eventsInfo.workReports) {
    workReport.account = getAccount(accounts, accountId)
  }

  await ctx.store.upsert([...accounts.values()])
  await ctx.store.insert(eventsInfo.joinGroups.map(([entity]) => entity))
  await ctx.store.insert(eventsInfo.storageOrders.map(([entity]) => entity))
  await ctx.store.insert(eventsInfo.workReports.map(([entity]) => entity))
})

function getEventsInfo(ctx: ProcessorContext<Store>): EventsInfo {
  const codec = ss58.codec('crust')
  const result: EventsInfo = {
    joinGroups: [],
    storageOrders: [],
    workReports: [],
    accountIds: new Set<string>(),
  }

  for (const block of ctx.blocks) {
    const createdAt = block.header.timestamp == null
      ? undefined
      : new Date(block.header.timestamp)

    for (const event of block.events) {
      if (event.name === events.swork.joinGroupSuccess.name) {
        const [member, owner] = events.swork.joinGroupSuccess.v1.decode(event)
        const memberId = codec.encode(member)
        result.joinGroups.push([
          new JoinGroup({
            id: event.id,
            owner: codec.encode(owner),
            blockNum: block.header.height,
            createdAt,
            extrinsicHash: event.extrinsic?.hash,
          }),
          memberId,
        ])
        result.accountIds.add(memberId)
      } else if (event.name === events.market.fileSuccess.name) {
        const [account, fileCid] = events.market.fileSuccess.v1.decode(event)
        const accountId = codec.encode(account)
        result.storageOrders.push([
          new StorageOrder({
            id: event.id,
            fileCid,
            blockNum: block.header.height,
            createdAt,
            extrinsicHash: event.extrinsic?.hash,
          }),
          accountId,
        ])
        result.accountIds.add(accountId)
      } else if (event.name === events.swork.worksReportSuccess.name) {
        const [account] = events.swork.worksReportSuccess.v1.decode(event)
        const accountId = codec.encode(account)
        const args = event.call?.args as {
          addedFiles?: unknown
          deletedFiles?: unknown
        } | undefined
        const addedFiles = toStringMatrix(args?.addedFiles)
        const deletedFiles = toStringMatrix(args?.deletedFiles)

        if (addedFiles.length > 0 || deletedFiles.length > 0) {
          result.workReports.push([
            new WorkReport({
              id: event.id,
              addedFiles,
              deletedFiles,
              blockNum: block.header.height,
              createdAt,
              extrinsicHash: event.extrinsic?.hash,
            }),
            accountId,
          ])
          result.accountIds.add(accountId)
        }
      }
    }
  }

  return result
}

function toStringMatrix(value: unknown): string[][] {
  if (!Array.isArray(value)) return []
  return value.map((row) => Array.isArray(row) ? row.map(String) : [String(row)])
}

function getAccount(accounts: Map<string, Account>, id: string): Account {
  const account = accounts.get(id)
  if (account == null) throw new Error(`Missing account ${id}`)
  return account
}

The generated byte values, including fileCid, are already hex strings. They do not need another toHex() conversion.

Build and prepare PostgreSQL

Build the project, start PostgreSQL, replace the template migration, and apply the new migration:

npm run build
docker compose up -d
rm -rf db/migrations
npx squid-typeorm-migration generate
npx squid-typeorm-migration apply

Run and verify the processor

Start the processor:

node -r dotenv/config lib/main.js

The processor exits after block 589350. Query the stored row counts:

docker compose exec db psql -U postgres -d squid -c '
SELECT (SELECT COUNT(*) FROM account) AS accounts,
       (SELECT COUNT(*) FROM join_group) AS join_groups,
       (SELECT COUNT(*) FROM storage_order) AS storage_orders,
       (SELECT COUNT(*) FROM work_report) AS work_reports;
'

This range produces 181 accounts, 178 group joins, 6 storage orders, and 1 work report containing a file change.

Query the GraphQL API

Start the GraphQL server:

npx squid-graphql-server

Open localhost:4350/graphql and run:

query CrustActivity {
  workReports(limit: 10, orderBy: blockNum_DESC) {
    blockNum
    addedFiles
    deletedFiles
    account {
      id
    }
  }
  storageOrders(limit: 10, orderBy: blockNum_DESC) {
    blockNum
    fileCid
  }
  joinGroups(limit: 10, orderBy: blockNum_DESC) {
    blockNum
    owner
  }
}

Overview

Pipes SDK

Squid SDK

Alternative clients

Prerequisites

​Prerequisites

Prerequisites