@sqd-pipes/delta-db

@subsquid/delta-db

Embedded rollback-aware computation engine for blockchain data. Routes raw events through reducers and materialized views, producing delta records (insert/update/delete) for downstream targets.

Build

Requires Rust (stable) and Node.js >= 18.

cd packages/delta-db-node
npm install

# Debug build
npx napi build --cargo-cwd ../.. --features napi

# Release build
npx napi build --cargo-cwd ../.. --features napi --release

This produces delta-db.node (native binary) and regenerates index.d.ts.

If index.js is missing after build, create it:

const { DeltaDb } = require('./delta-db.node')
module.exports.DeltaDb = DeltaDb

Test

npm test

Usage

Direct API

import { DeltaDb } from '@subsquid/delta-db'

const db = DeltaDb.open({
  schema: `
    CREATE TABLE swaps (
      block_number UInt64,
      pool         String,
      amount       Float64
    );
    CREATE MATERIALIZED VIEW volume AS
      SELECT pool, sum(amount) AS total, count() AS cnt
      FROM swaps GROUP BY pool;
  `,
  dataDir: './data', // optional, enables persistence
})

const batch = db.ingest({
  data: {
    swaps: [
      { block_number: 1000, pool: 'ETH/USDC', amount: 100 },
    ],
  },
  rollbackChain: [{ number: 1000, hash: '0x...' }],
  finalizedHead: { number: 999, hash: '0x...' },
})
// batch.records → [{ table: 'swaps', op: 'insert', ... }, { table: 'volume', op: 'insert', ... }]

Pipes SDK

import { deltaDbTarget } from '@subsquid/delta-db/pipes'

await source
  .pipe(decoder)
  .pipeTo(deltaDbTarget({
    schema: '...',
    dataDir: './data',
    transform: (data) => ({ transfers: data.transfers.map(formatRow) }),
    onDelta: async ({ batch }) => {
      await clickhouse.insert(batch.records)
    },
  }))