区块链数据索引实战：从智能合约事件到高性能查询接口的设计与实现

很多团队第一次做链上应用时，都会有一个误区：区块链已经存了数据，那我直接查链不就行了？

理论上没错，实际上很快就会撞墙。

因为链上节点擅长的是状态验证和共识，不是给业务系统做高频、复杂、低延迟查询。你要是让前端页面每次都去扫区块、读日志、拼历史数据，页面会慢得让人怀疑人生；如果后端服务每来一个请求都实时回放合约事件，节点和接口基本都扛不住。

所以，**“链上数据索引层”**几乎是稍微认真一点的区块链应用都绕不过去的一层基础设施。

这篇文章我会从架构角度，带你把这件事讲透，并给出一套可运行的示例：
从智能合约事件出发，落库到 PostgreSQL，再通过 API 提供高性能查询接口。

背景与问题

先明确问题：为什么不能直接查链？

常见场景包括：

• 查询某个地址的全部转账历史
• 按时间范围筛选 NFT Mint 记录
• 按代币、用户、区块高度做聚合统计
• 返回分页列表，并支持排序、过滤
• 快速展示“最近 24 小时交易量”“排行榜”“持仓分布”

这些需求放到以太坊 JSON-RPC 上，通常会遇到几个硬伤：

1. 查询粒度粗

   • 节点原生接口更偏底层，比如 eth_getLogs、eth_call
   • 想做复杂过滤、分页、关联查询，非常不顺手

2. 性能不可控

   • 大范围扫日志非常慢
   • 公共节点常有 rate limit
   • 自建节点也不是为复杂 OLAP/检索优化的

3. 重组（Reorg）处理麻烦

   • 你以为数据已经确认，结果链重组了
   • 如果没有索引层，业务端很难正确回滚状态

4. 业务语义缺失

   • 链上只有原始事件和状态
   • 产品需要的是“订单”“成交”“用户资产快照”“活跃地址统计”

所以索引系统的目标，不是“把链上数据复制一份”，而是：

把区块链上的底层事件，转换成可查询、可回放、可纠错、可扩展的业务数据模型。

先看整体方案

这类系统通常可以拆成 4 层：

1. 链上数据源

   • 区块、交易、收据、事件日志

2. 索引器（Indexer）

   • 按区块拉取日志
   • 解码合约事件
   • 写入数据库
   • 处理重试与重组

3. 存储层

   • PostgreSQL/MySQL：结构化查询
   • Redis：热点缓存
   • 对于大分析量，可再接 ClickHouse/Elasticsearch

4. 查询接口层

   • REST/GraphQL
   • 分页、过滤、排序、聚合
   • 对外隐藏链底层细节

下面这张图是一个比较典型的生产形态。

flowchart LR
    A[Blockchain Node / RPC] --> B[Indexer Worker]
    B --> C[(PostgreSQL)]
    B --> D[(Redis Cache)]
    C --> E[REST API]
    D --> E
    E --> F[Web / BI / Internal Service]

如果再细一点，从“区块推进”到“接口返回”通常是这样的：

sequenceDiagram
    participant N as RPC Node
    participant I as Indexer
    participant DB as PostgreSQL
    participant R as Redis
    participant API as Query API
    participant U as User

    I->>N: 获取 latest block
    I->>N: 按区块范围拉取 logs
    N-->>I: 事件日志
    I->>I: 解码事件 + 幂等处理
    I->>DB: UPSERT 事件/聚合数据
    I->>R: 刷新热点缓存
    U->>API: 查询地址历史
    API->>R: 先查缓存
    alt 命中缓存
        R-->>API: 返回结果
    else 未命中
        API->>DB: 查询分页结果
        DB-->>API: 数据集
        API->>R: 回填缓存
    end
    API-->>U: JSON 响应

方案对比：为什么“事件驱动索引”是主流

区块链数据索引一般有三种思路。

方案一：接口请求时实时查链

优点：

• 开发快
• 不需要维护数据库

缺点：

• 延迟高
• 无法支持复杂查询
• 很难稳定分页
• 节点压力大

适合：

• PoC、Demo、低频内部工具

方案二：定时扫链并存库

优点：

• 结构简单
• 好理解，好上手

缺点：

• 实时性较差
• 区块回滚处理容易遗漏
• 扫描窗口设计不好会重复或漏数

适合：

• 数据量不大、时效性要求不高的项目

方案三：基于智能合约事件构建增量索引

优点：

• 实时性好
• 增量处理成本低
• 数据建模清晰
• 容易支持业务查询接口

缺点：

• 强依赖合约事件设计质量
• 需要认真处理幂等、重组、补数

适合：

• 大多数正式业务系统

这篇文章选择的就是第三种。

核心原理

核心原理其实只有一句话：

把“链上的 append-only 事件流”，转换成“数据库中的可检索状态和历史记录”。

这个过程包含几个关键点。

1. 事件是事实源，不一定是最终查询模型

假设合约里有一个标准 ERC-20 Transfer 事件：

event Transfer(address indexed from, address indexed to, uint256 value);

原始事件很适合作为事实记录，但前端常常想查的是：

• 某地址收到过哪些 token
• 某 token 最近 100 笔转账
• 某地址净流入是多少
• 某时间段交易量

所以我们通常会维护两类表：

• 事实表：保存原始事件
• 派生表/聚合表：保存适合查询的业务数据

2. 幂等性比“跑通一次”重要得多

在索引器里，重复消费几乎不可避免：

• 服务重启
• 区块重复扫描
• 补历史数据
• 链重组回滚后重放

因此数据库设计时必须有天然唯一键，比如：

• chain_id + tx_hash + log_index

只要这组唯一键在，重复插入就不会造成脏数据。

3. 重组处理是区块链索引系统的分水岭

这是很多初版系统最容易忽略的问题。

在某些链上，区块不是 100% 立即最终确定的。你已经处理过区块 #1000，过一会儿它可能因为重组变成另一条链上的新区块。
如果你把旧事件直接当成真相写死了，业务就会出现：

• 明明显示到账，后来又消失
• 排行榜突然跳变
• API 数据前后不一致

常见做法：

• 只处理“确认数达到 N”的区块
• 或者维护最近 K 个区块的可回滚窗口
• 存储 block_hash
• 检测父块不连续时，回滚本地数据再重放

4. 查询接口不能暴露链底层复杂性

对业务方来说，最好不要让他们关心：

• log topic 是什么
• ABI 如何解码
• 某条链确认数设多少
• 一次 getLogs 最多扫多少块

API 应该提供的是：

• /transfers?address=0x...&page=1&pageSize=20
• /tokens/:address/stats
• /holders/top?n=100

这才是索引层的价值。

数据模型设计

下面用 ERC-20 Transfer 为例做一个最小可用设计。

1. 事实表：transfers

CREATE TABLE IF NOT EXISTS transfers (
  id BIGSERIAL PRIMARY KEY,
  chain_id INTEGER NOT NULL,
  block_number BIGINT NOT NULL,
  block_hash VARCHAR(66) NOT NULL,
  tx_hash VARCHAR(66) NOT NULL,
  log_index INTEGER NOT NULL,
  contract_address VARCHAR(42) NOT NULL,
  from_address VARCHAR(42) NOT NULL,
  to_address VARCHAR(42) NOT NULL,
  amount NUMERIC(78, 0) NOT NULL,
  removed BOOLEAN NOT NULL DEFAULT FALSE,
  block_timestamp TIMESTAMP NOT NULL,
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  UNIQUE (chain_id, tx_hash, log_index)
);

CREATE INDEX idx_transfers_contract_block
  ON transfers (contract_address, block_number DESC);

CREATE INDEX idx_transfers_from_block
  ON transfers (from_address, block_number DESC);

CREATE INDEX idx_transfers_to_block
  ON transfers (to_address, block_number DESC);

几点说明：

• NUMERIC(78, 0) 用来兼容大整数，不要用普通 BIGINT
• removed 用于标记重组移除的日志
• block_hash 很关键，后面做重组校验会用到

2. 游标表：indexer_state

CREATE TABLE IF NOT EXISTS indexer_state (
  chain_id INTEGER PRIMARY KEY,
  last_processed_block BIGINT NOT NULL DEFAULT 0,
  last_safe_block BIGINT NOT NULL DEFAULT 0,
  updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);

这张表的作用是：

• 记录当前同步到哪里了
• 服务重启后可恢复
• 支持多链场景

3. 如果业务需要高频余额查询

可以额外维护一张派生表：

CREATE TABLE IF NOT EXISTS token_balances (
  chain_id INTEGER NOT NULL,
  contract_address VARCHAR(42) NOT NULL,
  wallet_address VARCHAR(42) NOT NULL,
  balance NUMERIC(78, 0) NOT NULL DEFAULT 0,
  updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
  PRIMARY KEY (chain_id, contract_address, wallet_address)
);

这张表不是事实源，而是派生状态。
它适合做余额页、排行榜，但必须能通过事件重新构建。

实战代码（可运行）

下面我给一个最小可运行版本，用 Node.js + Express + PostgreSQL + Ethers 来实现：

• 从 RPC 拉取 ERC-20 Transfer 事件
• 写入 PostgreSQL
• 提供查询接口

为了控制篇幅，示例先聚焦主流程，不把 Redis 和完整重组回滚都塞进同一个 demo。

项目结构

blockchain-indexer-demo/
├─ package.json
├─ .env
├─ schema.sql
├─ indexer.js
└─ api.js

安装依赖

npm init -y
npm install ethers express pg dotenv

环境变量

.env

RPC_URL=https://mainnet.infura.io/v3/YOUR_KEY
CHAIN_ID=1
PG_URL=postgres://postgres:postgres@localhost:5432/indexer_demo
TOKEN_ADDRESS=0xA0b86991c6218b36c1d19d4a2e9eb0ce3606eb48
START_BLOCK=20000000
CONFIRMATIONS=12
PORT=3000

这里我用 USDC 合约地址举例。你也可以换成测试网或自己的合约。

建表 SQL

schema.sql

CREATE TABLE IF NOT EXISTS transfers (
  id BIGSERIAL PRIMARY KEY,
  chain_id INTEGER NOT NULL,
  block_number BIGINT NOT NULL,
  block_hash VARCHAR(66) NOT NULL,
  tx_hash VARCHAR(66) NOT NULL,
  log_index INTEGER NOT NULL,
  contract_address VARCHAR(42) NOT NULL,
  from_address VARCHAR(42) NOT NULL,
  to_address VARCHAR(42) NOT NULL,
  amount NUMERIC(78, 0) NOT NULL,
  removed BOOLEAN NOT NULL DEFAULT FALSE,
  block_timestamp TIMESTAMP NOT NULL,
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  UNIQUE (chain_id, tx_hash, log_index)
);

CREATE INDEX IF NOT EXISTS idx_transfers_contract_block
  ON transfers (contract_address, block_number DESC);

CREATE INDEX IF NOT EXISTS idx_transfers_from_block
  ON transfers (from_address, block_number DESC);

CREATE INDEX IF NOT EXISTS idx_transfers_to_block
  ON transfers (to_address, block_number DESC);

CREATE TABLE IF NOT EXISTS indexer_state (
  chain_id INTEGER PRIMARY KEY,
  last_processed_block BIGINT NOT NULL DEFAULT 0,
  last_safe_block BIGINT NOT NULL DEFAULT 0,
  updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);

索引器实现

indexer.js

require('dotenv').config();
const { ethers } = require('ethers');
const { Pool } = require('pg');

const RPC_URL = process.env.RPC_URL;
const CHAIN_ID = Number(process.env.CHAIN_ID || 1);
const PG_URL = process.env.PG_URL;
const TOKEN_ADDRESS = process.env.TOKEN_ADDRESS.toLowerCase();
const START_BLOCK = Number(process.env.START_BLOCK || 0);
const CONFIRMATIONS = Number(process.env.CONFIRMATIONS || 12);

const provider = new ethers.JsonRpcProvider(RPC_URL);
const pool = new Pool({ connectionString: PG_URL });

const abi = [
  'event Transfer(address indexed from, address indexed to, uint256 value)'
];
const iface = new ethers.Interface(abi);

async function getState(client) {
  const res = await client.query(
    'SELECT * FROM indexer_state WHERE chain_id = $1',
    [CHAIN_ID]
  );
  if (res.rowCount > 0) return res.rows[0];

  await client.query(
    `INSERT INTO indexer_state (chain_id, last_processed_block, last_safe_block)
     VALUES ($1, $2, $2)`,
    [CHAIN_ID, START_BLOCK - 1]
  );

  return {
    chain_id: CHAIN_ID,
    last_processed_block: START_BLOCK - 1,
    last_safe_block: START_BLOCK - 1
  };
}

async function saveState(client, lastProcessed, lastSafe) {
  await client.query(
    `UPDATE indexer_state
     SET last_processed_block = $2,
         last_safe_block = $3,
         updated_at = NOW()
     WHERE chain_id = $1`,
    [CHAIN_ID, lastProcessed, lastSafe]
  );
}

async function fetchBlockTimestamp(blockNumber) {
  const block = await provider.getBlock(blockNumber);
  return new Date(Number(block.timestamp) * 1000);
}

async function upsertTransfer(client, row) {
  const sql = `
    INSERT INTO transfers (
      chain_id, block_number, block_hash, tx_hash, log_index,
      contract_address, from_address, to_address, amount,
      removed, block_timestamp
    ) VALUES (
      $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11
    )
    ON CONFLICT (chain_id, tx_hash, log_index)
    DO UPDATE SET
      block_number = EXCLUDED.block_number,
      block_hash = EXCLUDED.block_hash,
      contract_address = EXCLUDED.contract_address,
      from_address = EXCLUDED.from_address,
      to_address = EXCLUDED.to_address,
      amount = EXCLUDED.amount,
      removed = EXCLUDED.removed,
      block_timestamp = EXCLUDED.block_timestamp
  `;
  await client.query(sql, row);
}

async function processRange(fromBlock, toBlock) {
  const topic = ethers.id('Transfer(address,address,uint256)');
  const logs = await provider.getLogs({
    address: TOKEN_ADDRESS,
    fromBlock,
    toBlock,
    topics: [topic]
  });

  const timestamps = new Map();
  const client = await pool.connect();

  try {
    await client.query('BEGIN');

    for (const log of logs) {
      const parsed = iface.parseLog(log);
      const from = parsed.args.from.toLowerCase();
      const to = parsed.args.to.toLowerCase();
      const value = parsed.args.value.toString();

      if (!timestamps.has(log.blockNumber)) {
        timestamps.set(log.blockNumber, await fetchBlockTimestamp(log.blockNumber));
      }

      await upsertTransfer(client, [
        CHAIN_ID,
        log.blockNumber,
        log.blockHash,
        log.transactionHash,
        log.index,
        log.address.toLowerCase(),
        from,
        to,
        value,
        false,
        timestamps.get(log.blockNumber)
      ]);
    }

    const latest = await provider.getBlockNumber();
    const safeBlock = Math.max(0, latest - CONFIRMATIONS);

    await saveState(client, toBlock, Math.min(toBlock, safeBlock));
    await client.query('COMMIT');

    console.log(`Processed blocks ${fromBlock} -> ${toBlock}, logs=${logs.length}`);
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

async function main() {
  const client = await pool.connect();
  let state;
  try {
    state = await getState(client);
  } finally {
    client.release();
  }

  while (true) {
    try {
      const latest = await provider.getBlockNumber();
      const target = latest - CONFIRMATIONS;

      if (target <= state.last_processed_block) {
        console.log('No new safe blocks, sleep 5s...');
        await new Promise(r => setTimeout(r, 5000));
        continue;
      }

      const batchSize = 500;
      const fromBlock = state.last_processed_block + 1;
      const toBlock = Math.min(fromBlock + batchSize - 1, target);

      await processRange(fromBlock, toBlock);
      state.last_processed_block = toBlock;
      state.last_safe_block = toBlock;
    } catch (err) {
      console.error('Indexer error:', err.message);
      await new Promise(r => setTimeout(r, 5000));
    }
  }
}

main().catch(err => {
  console.error(err);
  process.exit(1);
});

这个版本做了什么

• 从 START_BLOCK 开始同步
• 只处理已确认区块，降低重组影响
• 用 (chain_id, tx_hash, log_index) 做幂等
• 按批次拉日志，避免一次请求过大
• 状态落库，可断点续跑

它还缺什么

一个正式生产版通常还要加：

• 动态批次控制
• 多合约/多事件支持
• 重组检测与回滚
• 失败任务队列
• Prometheus 指标
• Redis 缓存
• API 限流和鉴权

查询接口实现

api.js

require('dotenv').config();
const express = require('express');
const { Pool } = require('pg');

const PORT = Number(process.env.PORT || 3000);
const PG_URL = process.env.PG_URL;
const pool = new Pool({ connectionString: PG_URL });

const app = express();
app.use(express.json());

app.get('/health', async (req, res) => {
  try {
    await pool.query('SELECT 1');
    res.json({ ok: true });
  } catch (err) {
    res.status(500).json({ ok: false, error: err.message });
  }
});

app.get('/transfers', async (req, res) => {
  try {
    const {
      address,
      contract,
      page = '1',
      pageSize = '20'
    } = req.query;

    const pageNum = Math.max(1, Number(page));
    const sizeNum = Math.min(100, Math.max(1, Number(pageSize)));
    const offset = (pageNum - 1) * sizeNum;

    const conditions = ['removed = FALSE'];
    const params = [];
    let idx = 1;

    if (address) {
      conditions.push(`(from_address = $${idx} OR to_address = $${idx})`);
      params.push(String(address).toLowerCase());
      idx++;
    }

    if (contract) {
      conditions.push(`contract_address = $${idx}`);
      params.push(String(contract).toLowerCase());
      idx++;
    }

    const whereClause = conditions.length ? `WHERE ${conditions.join(' AND ')}` : '';

    const countSql = `
      SELECT COUNT(*)::INT AS total
      FROM transfers
      ${whereClause}
    `;

    const listSql = `
      SELECT
        block_number,
        tx_hash,
        log_index,
        contract_address,
        from_address,
        to_address,
        amount,
        block_timestamp
      FROM transfers
      ${whereClause}
      ORDER BY block_number DESC, log_index DESC
      LIMIT $${idx} OFFSET $${idx + 1}
    `;

    const countResult = await pool.query(countSql, params);
    const listResult = await pool.query(listSql, [...params, sizeNum, offset]);

    res.json({
      page: pageNum,
      pageSize: sizeNum,
      total: countResult.rows[0].total,
      items: listResult.rows
    });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

app.get('/addresses/:address/stats', async (req, res) => {
  try {
    const address = req.params.address.toLowerCase();

    const sql = `
      SELECT
        COALESCE(SUM(CASE WHEN to_address = $1 THEN amount ELSE 0 END), 0) AS total_in,
        COALESCE(SUM(CASE WHEN from_address = $1 THEN amount ELSE 0 END), 0) AS total_out,
        COUNT(*)::INT AS transfer_count
      FROM transfers
      WHERE removed = FALSE
        AND (from_address = $1 OR to_address = $1)
    `;

    const result = await pool.query(sql, [address]);
    res.json(result.rows[0]);
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

app.listen(PORT, () => {
  console.log(`API listening on :${PORT}`);
});

启动方式

先建库并执行建表：

psql postgres://postgres:postgres@localhost:5432/indexer_demo -f schema.sql

启动索引器：

node indexer.js

启动 API：

node api.js

测试接口：

curl "http://localhost:3000/transfers?contract=0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&page=1&pageSize=5"

curl "http://localhost:3000/addresses/0x55fe002aeff02f77364de339a1292923a15844b8/stats"

架构演进：从 Demo 到生产

如果系统开始承载真实流量，建议往下面这个方向演进。

flowchart TD
    A[RPC / Archive Node] --> B[Block Scanner]
    B --> C[Decode Worker]
    C --> D[(Raw Events Table)]
    C --> E[(Derived Tables)]
    E --> F[Query API]
    E --> G[Materialized Views]
    F --> H[(Redis)]
    F --> I[Clients]
    D --> J[Reorg Handler]
    J --> E

这个阶段通常会加入：

• Raw Events Table
  • 先把原始日志完整存下来
  • 派生逻辑可以离线重算
• Derived Tables
  • 面向业务查询建模
• Materialized Views
  • 做榜单、汇总、时间窗口统计
• Reorg Handler
  • 专门处理回滚与重放
• Redis
  • 缓解热点查询压力

容量估算与取舍分析

这部分经常被忽略，但架构设计离不开它。

1. 吞吐估算

假设你的目标合约平均每天产生 300 万条事件：

• 每条事件落库后约 200~500 字节（粗略）
• 每天数据量约 600MB ~ 1.5GB
• 一年就是 200GB 级别起步，还没算索引和备份

这时就要考虑：

• PostgreSQL 单表是否需要分区
• 历史冷数据是否归档
• API 查询是否只覆盖最近 N 天热数据

2. 查询模式决定索引方式

如果请求主要是：

• 按地址查历史
• 按合约查最新记录

那 B-Tree 足够。

如果你要：

• 全文检索
• 复杂聚合分析
• 多维明细钻取

那 PostgreSQL 可能不够，至少要考虑：

• ClickHouse：分析型聚合
• Elasticsearch：复杂检索
• TimescaleDB：时序增强

不要一上来就“全家桶”，先看访问模式再决定。

3. 一致性 vs 实时性

这是典型取舍：

• 如果你追求秒级实时，可能要接受少量短暂回滚风险
• 如果你追求强一致，就要等更多确认数

我的建议是：

• 资金类、资产类接口：偏保守，增加确认数
• 资讯类、动态流接口：可接受更低确认数，但前端标注“待确认”

常见坑与排查

这部分我自己踩过不少，尤其是第一版索引器总是“看起来在跑，实际数据有洞”。

1. getLogs 范围过大，被 RPC 拒绝

现象：

• 返回超时
• 报错 response size exceeded
• 公共节点直接限流

排查思路：

• 缩小区块批次，比如从 5000 改成 200
• 分链看限制，有些节点对日志查询非常严格
• 对热点合约，按块分段并发，但要控制并发数

建议：

• 做动态批次控制
• 失败后指数退避重试

2. 事件没漏，但查询结果还是不对

常见原因：

• 地址大小写不统一
• amount 被当成 JS number 丢精度
• 分页排序不稳定
• 统计接口没过滤 removed = false

建议：

• 地址全部转小写存储
• 金额一律字符串/大整数处理
• 排序使用 (block_number DESC, log_index DESC)
• 所有业务查询都明确过滤重组移除记录

3. 断点续跑后重复数据暴增

原因通常是：

• 没有唯一约束
• 状态表更新时机不对
• 事务没包住“写数据 + 更新游标”

正确做法：

• 事实表必须有唯一键
• 数据写入与游标更新放同一事务
• 成功提交后再推进内存状态

这个点非常关键。我见过不少系统，明明加了断点恢复，却因为游标提前更新导致丢数。

4. 合约升级后 ABI 变了

如果你的合约可升级，或者不同版本事件格式有变化，就会出现：

• 老事件解码正常，新事件报错
• 同一个业务实体跨版本字段不一致

建议：

• 以合约地址 + 生效区块 + ABI 版本做映射
• 不要把 ABI 写死成单版本常量
• 派生层尽量输出统一业务模型

5. 重组导致余额表错误

这是派生表最容易出错的地方。
如果你只在插入时更新余额，不处理回滚，那余额一定会漂。

排查方式：

• 找某个异常地址，回放其相关事件
• 对比事实表与派生表计算结果
• 检查是否有 recent blocks 的回滚机制

建议：

• 余额这种状态表一定要可重建
• 最近 N 个块的派生结果尽量支持重算

安全/性能最佳实践

这一节给一些更偏工程化的建议。

安全最佳实践

1. 不要盲信单一 RPC

单一 RPC 可能：

• 数据延迟
• 偶发返回不一致
• 对归档数据支持不完整

建议：

• 关键业务至少配主备 RPC
• 对最新区块高度、区块 hash 做抽样校验

2. API 层要做输入校验

比如：

• 地址格式校验
• pageSize 上限
• 时间范围上限
• 排序字段白名单

否则别人一个超大分页就能把数据库拖慢。

3. 把“链上数据正确”与“业务可见”分层

不要让未确认数据直接进入最终资产视图。
比较稳妥的方式是：

• 原始事件可以先入库
• 对外接口只展示已确认或标记状态的数据

性能最佳实践

1. 索引器与查询接口分离

不要把拉链、解码、写库、对外查询混成一个服务。
最少也应拆成：

• indexer
• api

这样故障域更清晰。

2. 热点数据做缓存

适合缓存的内容：

• 首页统计
• Top N 排行
• 最近区块动态
• 高频地址画像

不适合盲目缓存的内容：

• 强一致资产余额
• 高度个性化且更新频繁的数据

3. 优先做“覆盖查询路径”的索引

不是索引越多越好。
多一个索引，就多一份写放大。

例如你的主要 SQL 是：

SELECT *
FROM transfers
WHERE contract_address = $1
ORDER BY block_number DESC
LIMIT 20;

那就优先建：

CREATE INDEX idx_transfers_contract_block
ON transfers (contract_address, block_number DESC);

而不是先堆一堆无关索引。

4. 大表要考虑分区

当 transfers 进入亿级别时，建议考虑：

• 按 block_number 范围分区
• 或按时间月分区
• 定期归档冷数据

5. 派生聚合尽量异步

例如：

• 日交易量
• 用户活跃统计
• 排行榜

这些不一定要在写事件时同步更新，可以：

• 先落事实表
• 再由异步任务批量聚合

这样写入链路更稳。

一个更稳的重组处理思路

如果你的系统已经进入生产，建议用下面这个状态模型来处理区块。

stateDiagram-v2
    [*] --> Pending
    Pending --> Confirmed: 达到确认数
    Pending --> Removed: 检测到重组
    Confirmed --> Finalized: 超过回滚窗口
    Removed --> Replayed: 新链重放完成

具体做法可以是：

1. 先把最近 K 个块的数据标记为 pending
2. 达到确认数后转为 confirmed
3. 如果检测到区块 hash 不一致：
   • 找到分叉点
   • 删除或标记该分叉点之后的派生数据
   • 重放新区块
4. 旧数据超过回滚窗口后视作 finalized

这样做虽然复杂一点，但对资产类业务非常值得。

边界条件：什么时候不需要自建索引器

并不是所有项目都要自己维护这一套。

如果你的需求只是：

• 查基础交易历史
• 展示常见代币数据
• 内部低频使用

可以优先考虑：

• 区块链浏览器 API
• 第三方数据服务
• 托管索引平台

但如果你满足下面任意一条，我会建议尽早自建：

• 有明确的业务语义建模需求
• 查询量大
• 对延迟和稳定性有要求
• 需要可控的数据质量和回溯能力
• 需要跨合约、跨事件、跨链做统一查询

总结

把区块链数据索引这件事做好，关键不在于“能不能把事件抓下来”，而在于你是否把下面几件事真正设计清楚了：

• 事件模型：哪些是事实源，哪些是派生状态
• 幂等机制：重复消费怎么不出脏数据
• 重组策略：回滚和重放怎么做
• 查询模型：面向业务而不是面向 RPC
• 容量规划：数据量和热点查询如何演进

如果你现在准备落地一版，我建议按这个顺序推进：

1. 先确定业务查询接口长什么样
2. 再反推需要哪些事实表和派生表
3. 用事件做增量索引，保证唯一键和事务一致性
4. 先支持确认数窗口，后续再补完整重组回滚
5. 热点查询接缓存，大表提前考虑分区

一句话收尾：

链是事实来源，索引层才是业务数据入口。

只要你接受这个分层思路，后面的架构设计就会清晰很多。