Розробка системи індексації блокчейн-подій

Розробка систем індексації блокчейн-подій

Проблема виникає коли вам потрібно відповісти на запитання "покажи все транзакції цього користувача за останні 30 днів" або "який загальний обсяг торгів пройшов через наш DEX". Прямі виклики до вузла через eth_getLogs з діапазонами блоків — це не архітектура, це костиль. На архівній ноді з блоку 0 такий запрос займе хвилини, на публічному RPC — швидше за все впаде з timeout або block range too large.

Правильна система індексації — це pipeline: нода → event listener → parser → база даних → API. Кожен компонент з незалежним масштабуванням та гарантією exactly-once або at-least-once семантики.

Архітектура event pipeline

Шар отримання подій

Три підходи до отримання подій з вузла, кожен зі своєю ціною надійності:

Polling через eth_getLogs — найпростіший та найнадійніший при правильній реалізації. Воркер періодично запитує події за діапазон блоків, зберігає lastIndexedBlock, при перезапуску продовжує з останньої оброблюваної блоку.

interface IndexerState {
    lastIndexedBlock: number
    lastIndexedBlockHash: string // для детекції реорга
}

async function pollEvents(
    provider: ethers.JsonRpcProvider,
    contracts: ContractConfig[],
    state: IndexerState
): Promise<ProcessedEvent[]> {
    const currentBlock = await provider.getBlockNumber()
    const toBlock = Math.min(state.lastIndexedBlock + BATCH_SIZE, currentBlock - CONFIRMATIONS)

    if (toBlock <= state.lastIndexedBlock) return []

    // Верифікація непрервиності chain
    const lastBlock = await provider.getBlock(state.lastIndexedBlock)
    if (lastBlock?.hash !== state.lastIndexedBlockHash) {
        throw new ReorgDetectedError(state.lastIndexedBlock)
    }

    const logs = await provider.getLogs({
        address: contracts.map(c => c.address),
        topics: [contracts.flatMap(c => c.topics)],
        fromBlock: state.lastIndexedBlock + 1,
        toBlock,
    })

    return logs.map(parseLog)
}

WebSocket subscriptions — низька latency (нові события приходять одразу), але WebSocket з'єднання нестабільні. Потрібен automatic reconnect з backoff та синхронізація пропущених блоків при переподключенні:

async function subscribeWithFallback(provider: ethers.WebSocketProvider) {
    const filter = { address: CONTRACT, topics: [EVENT_TOPIC] }

    provider.on(filter, async (log) => {
        await processLog(log)
    })

    provider.websocket.on('close', async () => {
        console.log('WebSocket closed, syncing missed blocks...')
        await syncMissedBlocks(lastProcessedBlock)
        reconnect() // exponential backoff
    })
}

Firehose / StreamingFast — enterprise-рівень. Бінарний стриміння блоків з усім контентом, мінімальна latency, вбудована обробка реоргів. Використовується як джерело даних для The Graph protocol nodes. Істотно складніше в налаштуванні.

Обробка реорганізацій

Реорги — найчастіший джерело багів у індексерах. На Ethereum з PoS финальність приходить після двох епох (~12-13 хвилин). На PoW мережах та молодих EVM ланцюгах (BSC, Polygon) реорги глибиною 3-5 блоків — норма.

Стратегія: індексувати з затримкою N блоків (safe confirmations), зберігати хеш кожної індексованої блоку, при невідповідності — откат до останнього консистентного стану.

-- Таблиця стану індексера
CREATE TABLE indexer_blocks (
    block_number    BIGINT PRIMARY KEY,
    block_hash      VARCHAR(66) NOT NULL,
    indexed_at      TIMESTAMPTZ DEFAULT NOW()
);

-- Подія з прив'язкою до блоку для можливості rollback
CREATE TABLE indexed_events (
    id              BIGSERIAL PRIMARY KEY,
    block_number    BIGINT NOT NULL REFERENCES indexer_blocks(block_number),
    log_index       INT NOT NULL,
    tx_hash         VARCHAR(66) NOT NULL,
    contract_addr   VARCHAR(42) NOT NULL,
    event_name      VARCHAR(100) NOT NULL,
    decoded_data    JSONB NOT NULL,
    UNIQUE(tx_hash, log_index)
);

При детекції реорга — DELETE FROM indexed_events WHERE block_number >= reorg_depth, потім переіндексація з точки реорга.

Декодування подій

ABI-декодування подій — тривіальна задача для ethers.js або viem, але є нюанси.

Indexed vs non-indexed параметри: indexed параметри потрапляють у topics, non-indexed — у data. Подія з 3 indexed параметрів + event signature займає 4 topics. Декодування topics для non-primitive типів (structs, dynamic arrays) неможливе — вони хешуються через keccak256 і втрачають вихідні дані.

import { decodeEventLog } from 'viem'

function parseSwapEvent(log: Log, abi: Abi): SwapEvent {
    const decoded = decodeEventLog({
        abi,
        eventName: 'Swap',
        data: log.data,
        topics: log.topics,
    })

    return {
        blockNumber: log.blockNumber,
        txHash: log.transactionHash,
        sender: decoded.args.sender,
        recipient: decoded.args.recipient,
        amount0: decoded.args.amount0,
        amount1: decoded.args.amount1,
        sqrtPriceX96: decoded.args.sqrtPriceX96,
        liquidity: decoded.args.liquidity,
        tick: decoded.args.tick,
    }
}

Анонімні события (без event selector у topic0) — рідкість, але зустрічається в старих контрактах. Декодування вимагає знання структури даних без ABI.

Proxy контракти: события емітяться з адреси proxy, але ABI від реалізації. Потрібно резолвити implementation address через EIP-1967 storage slot: 0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc.

База даних та продуктивність

PostgreSQL — стандарт де-факто для зберігання подій. Ключові рішення по схемі:

Партиціонування за block_number: для контрактів з високим обсягом подій (Uniswap V3, великі ERC-20) таблиця швидко вирастає до сотень мільйонів рядків. Партиціонування за діапазоном дає планові сканування замість seq scan:

CREATE TABLE swap_events (
    block_number BIGINT NOT NULL,
    -- ...решта полів
) PARTITION BY RANGE (block_number);

CREATE TABLE swap_events_0_5m    PARTITION OF swap_events FOR VALUES FROM (0) TO (5000000);
CREATE TABLE swap_events_5m_10m  PARTITION OF swap_events FOR VALUES FROM (5000000) TO (10000000);
-- тощо

JSONB для decoded_data: зберігання декодованих даних у JSONB дозволяє додавати нові типи подій без міграцій схеми. GIN індекси на часто запитувані поля всередину JSONB. Для критичних полів — вилучення в окремі типізовані колонки.

TimescaleDB — розширення для PostgreSQL, яке дає автоматичне time-based партиціонування (hypertables), стиснення старих даних та continuous aggregates для OHLCV/метрик без фонових джобів:

-- Continuous aggregate: обсяг за годину за пул
CREATE MATERIALIZED VIEW pool_hourly_volume
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', event_timestamp) AS hour,
    pool_address,
    SUM(amount_usd) AS volume_usd,
    COUNT(*) AS tx_count
FROM swap_events
GROUP BY 1, 2;

SELECT add_continuous_aggregate_policy('pool_hourly_volume',
    start_offset => INTERVAL '3 hours',
    end_offset   => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour'
);

Мониторинг та алерти

Індексер повинен мати метрики: indexer_lag_blocks (розрив між head та останньої індексованою блоком), events_per_second, reorg_count. Алерт якщо indexer_lag_blocks > 50 — індексер відстає або завис.

Деплой через Docker Compose / Kubernetes з health check endpoint, який повертає 503 якщо lag перевищує поріг. Це дозволяє load balancer-у виключити нездоровий інстанс.

Типовий стек: Go або Rust для воркера (продуктивність), PostgreSQL / TimescaleDB для зберігання, Redis для стану та черг, Grafana + Prometheus для мониторинга.