Factory transformers

Use the factory pattern when you need to index events from contracts that are deployed dynamically by a known factory contract — for example, Uniswap V3 pools created by the UniswapV3Factory.

The examples below use typegen-generated ABI modules. See Specifying events for how to generate them from a JSON ABI.

Basic factory

Track events from contracts created by a factory. The factory() helper discovers child contracts from the factory’s creation events and maintains the address list in a local SQLite database.

import { evmPortalStream, evmDecoder, factory, factorySqliteDatabase } from "@subsquid/pipes/evm";
import { createTarget } from "@subsquid/pipes";
import * as factoryAbi from "./abi/uniswap-v3-factory";
import * as poolAbi from "./abi/uniswap-v3-pool";

await evmPortalStream({
  portal: "https://portal.sqd.dev/datasets/ethereum-mainnet",
  outputs: evmDecoder({
    range: { from: 12369621 },
    contracts: factory({
      address: "0x1f98431c8ad98523631ae4a59f267346ea31f984",
      event: factoryAbi.events.PoolCreated,
      parameter: "pool",
      database: factorySqliteDatabase({ path: "./uniswap-v3-pools.sqlite" }),
    }),
    events: { swap: poolAbi.events.Swap },
  }),
}).pipeTo(createTarget({
  write: async ({ logger, read }) => {
    for await (const { data } of read()) {
      logger.info(`Parsed ${data.swap.length} swaps`);
    }
  },
}));

Filtering factory events

To narrow which child contracts are tracked, pass an event object with a params field. Only creation events matching the specified parameter values are stored — unmatched contracts are ignored at both the portal and the local database level.

contracts: factory({
  address: "0x1f98431c8ad98523631ae4a59f267346ea31f984",
  event: {
    event: factoryAbi.events.PoolCreated,
    params: {
      token0: "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", // WETH
    },
  },
  parameter: "pool",
  database: factorySqliteDatabase({ path: "./uniswap-v3-weth-pools.sqlite" }),
})

Filter rules:

Only indexed parameters can be used for filtering.
Multiple parameters are combined with AND logic.
Passing an array of values for a parameter matches any of them (OR logic).
Address matching is case-insensitive.

Show Full example: filter by token0 with WETH pools

import { evmPortalStream, evmDecoder, factory, factorySqliteDatabase } from "@subsquid/pipes/evm";
import { createTarget } from "@subsquid/pipes";
import * as factoryAbi from "./abi/uniswap-v3-factory";
import * as poolAbi from "./abi/uniswap-v3-pool";

const WETH = "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2";

await evmPortalStream({
  portal: "https://portal.sqd.dev/datasets/ethereum-mainnet",
  outputs: evmDecoder({
    range: { from: 12369621 },
    contracts: factory({
      address: "0x1f98431c8ad98523631ae4a59f267346ea31f984",
      event: {
        event: factoryAbi.events.PoolCreated,
        params: { token0: WETH },
      },
      parameter: "pool",
      database: factorySqliteDatabase({ path: "./uniswap-v3-weth-pools.sqlite" }),
    }),
    events: { swap: poolAbi.events.Swap },
  }),
}).pipeTo(createTarget({
  write: async ({ logger, read }) => {
    for await (const { data } of read()) {
      logger.info(`Parsed ${data.swap.length} swaps from WETH pools`);
    }
  },
}));

Including factory event data

DecodedEvent<T, F> carries a .factory field with the creation event. Use it when you need to include factory context (e.g. pool token addresses) alongside each decoded event.

Show Full example: access factory event metadata

import {
  evmPortalStream, evmDecoder, factory, DecodedEvent, factorySqliteDatabase,
} from "@subsquid/pipes/evm";
import { createTarget } from "@subsquid/pipes";
import * as factoryAbi from "./abi/uniswap-v3-factory";
import * as poolAbi from "./abi/uniswap-v3-pool";

function addFactoryMetadata<T, F>(event: DecodedEvent<T, F>) {
  return {
    ...event.event,
    blockNumber: event.block.number,
    factoryEvent: event.factory?.event,
  };
}

const decoder = evmDecoder({
  range: { from: 12369621 },
  contracts: factory({
    address: "0x1f98431c8ad98523631ae4a59f267346ea31f984",
    event: factoryAbi.events.PoolCreated,
    parameter: "pool",
    database: factorySqliteDatabase({ path: "./uniswap-v3-pools.sqlite" }),
  }),
  events: { swap: poolAbi.events.Swap, mint: poolAbi.events.Mint },
}).pipe(({ swap, mint }) => ({
  swap: swap.map(addFactoryMetadata),
  mint: mint.map(addFactoryMetadata),
}));

await evmPortalStream({
  portal: "https://portal.sqd.dev/datasets/ethereum-mainnet",
  outputs: decoder,
}).pipeTo(createTarget({
  write: async ({ logger, read }) => {
    for await (const { data } of read()) {
      for (const s of data.swap) {
        logger.info({
          pool: s.factoryEvent?.pool,
          token0: s.factoryEvent?.token0,
          token1: s.factoryEvent?.token1,
          amount0: s.amount0.toString(),
          amount1: s.amount1.toString(),
        });
      }
    }
  },
}));

Multiple factories

Pass separate evmDecoder outputs to track contracts from different factory addresses in a single pipeline.

Show Full example: Uniswap V2 and V3 in one pipeline

import {
  evmPortalStream, evmDecoder, factory, factorySqliteDatabase,
} from "@subsquid/pipes/evm";
import { createTarget } from "@subsquid/pipes";
import * as uniswapV3FactoryAbi from "./abi/uniswap-v3-factory";
import * as uniswapV2FactoryAbi from "./abi/uniswap-v2-factory";
import * as poolAbi from "./abi/pool";

await evmPortalStream({
  portal: "https://portal.sqd.dev/datasets/ethereum-mainnet",
  outputs: {
    v3: evmDecoder({
      range: { from: 12369621 },
      contracts: factory({
        address: "0x1f98431c8ad98523631ae4a59f267346ea31f984",
        event: uniswapV3FactoryAbi.events.PoolCreated,
        parameter: "pool",
        database: factorySqliteDatabase({ path: "./v3-pools.sqlite" }),
      }),
      events: { swap: poolAbi.events.Swap },
    }),
    v2: evmDecoder({
      range: { from: 10000835 },
      contracts: factory({
        address: "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f",
        event: uniswapV2FactoryAbi.events.PairCreated,
        parameter: "pair",
        database: factorySqliteDatabase({ path: "./v2-pairs.sqlite" }),
      }),
      events: { swap: poolAbi.events.Swap },
    }),
  },
}).pipeTo(createTarget({
  write: async ({ logger, read }) => {
    for await (const { data } of read()) {
      logger.info({ v3Swaps: data.v3.swap.length, v2Swaps: data.v2.swap.length });
    }
  },
}));

Pre-indexing factory Experimental

Pre-populate the factory database before the main pipeline to ensure all historical child contracts are known before live indexing begins.

This is an experimental feature. The pre-indexing request is limited and this approach won’t work for thousands of addresses.

Show Full example: pre-index all pools then run main pipeline

import {
  evmPortalStream, evmDecoder, factory, factorySqliteDatabase,
} from "@subsquid/pipes/evm";
import { createTarget } from "@subsquid/pipes";
import * as factoryAbi from "./abi/uniswap-v3-factory";
import * as poolAbi from "./abi/uniswap-v3-pool";

const factoryDb = factorySqliteDatabase({ path: "./uniswap-v3-pools.sqlite" });

// Step 1: pre-index historical pool creations
await evmPortalStream({
  portal: "https://portal.sqd.dev/datasets/ethereum-mainnet",
  outputs: evmDecoder({
    range: { from: 12369621, to: 20000000 },
    contracts: ["0x1f98431c8ad98523631ae4a59f267346ea31f984"],
    events: { poolCreated: factoryAbi.events.PoolCreated },
  }),
}).pipeTo(createTarget({
  write: async ({ logger, read }) => {
    for await (const { data } of read()) {
      for (const event of data.poolCreated) {
        await factoryDb.add(event.event.pool);
        logger.info(`Added pool: ${event.event.pool}`);
      }
    }
  },
}));

// Step 2: run main pipeline with populated factory database
await evmPortalStream({
  portal: "https://portal.sqd.dev/datasets/ethereum-mainnet",
  outputs: evmDecoder({
    range: { from: 20000000 },
    contracts: factory({
      address: "0x1f98431c8ad98523631ae4a59f267346ea31f984",
      event: factoryAbi.events.PoolCreated,
      parameter: "pool",
      database: factoryDb,
    }),
    events: { swap: poolAbi.events.Swap },
  }),
}).pipeTo(createTarget({
  write: async ({ logger, read }) => {
    for await (const { data } of read()) {
      logger.info(`Processed ${data.swap.length} swaps`);
    }
  },
}));

Overview

Pipes SDK

Squid SDK

Basic factory

Filtering factory events

Including factory event data

Multiple factories

Pre-indexing factory Experimental

Overview

Pipes SDK

Squid SDK

Documentation Index

​Basic factory

​Filtering factory events

​Including factory event data

​Multiple factories

​Pre-indexing factory Experimental

Basic factory

Filtering factory events

Including factory event data

Multiple factories

Pre-indexing factory Experimental