Substreams Broker

高性能的 Broker 数据索引系统，使用 Substreams 替代传统的 Graph Node。基于 newBroker 的逻辑实现，提供更高效的数据处理和实时同步能力。

🎯 核心功能

• 🚀 高性能事件处理: 基于 Substreams 的原生并行处理能力
• 📊 实时关注关系跟踪: 精确追踪用户与 Broker 的关注状态
• 💰 准确的统计数据: 实时计算粉丝数量和托管金额
• 🔄 断点续传: 支持中断后从上次位置继续同步
• 🌐 多客户端支持: JavaScript Stream Client + PostgreSQL 数据库
• 📈 可扩展架构: 易于添加新的事件类型和数据处理逻辑

📋 监听的事件

基于 BSC 网络上的 Broker 合约 (0xC74848F603f723A4878131952c9238908f096266):

• BrokerRegistered(indexed address) - Broker注册事件
• Follow(indexed address,indexed address,uint256) - 关注事件
• Unfollow(indexed address,indexed address,uint256) - 取消关注事件

🗄️ 数据结构

BrokerItem (broker_items 表)

• account: Broker 地址
• time: 注册时间戳
• follower_count: 当前粉丝数量
• total_amount: 总托管金额
• 区块链相关字段 (block_number, block_hash, transaction_hash)

FollowRelation (follow_relations 表)

• user_address: 用户地址
• broker: Broker 地址
• amount: 托管金额
• is_active: 关注状态
• follow_time: 关注时间
• unfollow_time: 取消关注时间
• 区块链相关字段

1. 环境准备

# 安装 Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
source ~/.cargo/env
rustup target add wasm32-unknown-unknown

# 安装 Substreams CLI
curl -sSL https://github.com/streamingfast/substreams/releases/download/v1.1.20/substreams_linux_x86_64.tar.gz | tar -xz -C /tmp
sudo mv /tmp/substreams /usr/local/bin/

# 安装 Node.js (用于 JavaScript 客户端)
nvm install 18
nvm use 18

2. 构建 Substreams

# 构建 WASM 模块和打包
./build.sh

3. 设置数据库

# 使用 Docker 启动 PostgreSQL
docker-compose up -d postgres

# 或手动部署到现有数据库
./deploy-db.sh

4. 配置认证

# 设置 Substreams API Token
export SUBSTREAMS_API_TOKEN="your_token_here"

# 或创建 .substreams.env 文件
echo "SUBSTREAMS_API_TOKEN=your_token_here" > .substreams.env

5. 启动数据同步

# 启动 JavaScript 流客户端
./start-stream.sh

# 或直接使用 substreams CLI
substreams run substreams.yaml map_events -e bsc.substreams.pinax.network:443 -s 64218055

📁 项目结构

substreams-broker/
├── README.md                 # 项目文档
├── Cargo.toml               # Rust 依赖配置
├── build.rs                 # 构建脚本
├── substreams.yaml          # Substreams 配置
├── schema.sql               # PostgreSQL 数据库 schema
├── docker-compose.yml       # Docker 服务配置
├── build.sh                 # 构建脚本
├── deploy-db.sh            # 数据库部署脚本
├── start-stream.sh         # 启动流客户端
├── proto/                   # Protocol Buffers 定义
│   ├── events.proto         # 事件消息定义
│   └── database.proto       # 数据库更改定义
├── src/                     # Rust 源代码
│   ├── lib.rs              # 主要逻辑
│   ├── events.rs           # 事件签名计算
│   └── constants.rs        # 常量定义
└── abi/                     # 合约 ABI
    └── Broker.json         # Broker 合约 ABI

🔧 配置说明

Substreams 配置

• 起始区块: 64218055 (与 newBroker 相同)
• 网络: BSC (Binance Smart Chain)
• 合约地址: 0xC74848F603f723A4878131952c9238908f096266

数据库配置

环境变量:

• DB_HOST: 数据库主机 (默认: localhost)
• DB_PORT: 数据库端口 (默认: 5432)
• DB_NAME: 数据库名称 (默认: broker_substreams)
• DB_USER: 数据库用户 (默认: postgres)
• DB_PASSWORD: 数据库密码 (默认: postgres)

📊 性能优势

相比传统的 Graph Node 实现:

• 处理速度: 10-100x 更快的事件处理
• 资源占用: 更低的内存和 CPU 使用
• 实时性: 接近实时的数据同步 (<1s 延迟)
• 可靠性: 内置的断点续传和错误重试
• 可扩展性: 易于水平扩展和分布式部署

🔍 监控和调试

查看事件日志

# 使用 substreams CLI 工具实时查看事件
substreams run substreams.yaml map_events -e bsc.substreams.pinax.network:443 -s 64218055 -t +100

数据库查询

-- 查看 broker 统计
SELECT * FROM broker_summary;

-- 查看活跃关注关系
SELECT * FROM active_follow_relations;

-- 查看处理进度
SELECT * FROM substreams_cursors;

Web UI 访问

启动 Docker 服务后，可以通过 Adminer 访问数据库:

• URL: http://localhost:8080
• 系统: PostgreSQL
• 服务器: postgres
• 用户名: postgres
• 密码: postgres
• 数据库: broker_substreams

🤝 与原有系统的对比

特性	newBroker (Graph Node)	substreams-broker
处理速度	较慢	10-100x 更快
资源占用	高	低
实时性	5-30s 延迟	<1s 延迟
断点续传	有限支持	完整支持
横向扩展	困难	简单
开发复杂度	中等	较高
生态支持	丰富	发展中

📝 开发和贡献

添加新事件类型

1. 在 proto/events.proto 中添加新的消息定义
2. 在 src/events.rs 中添加事件签名
3. 在 src/lib.rs 中添加解析和处理逻辑
4. 更新数据库 schema
5. 重新构建和部署

测试

# 运行 Rust 测试
cargo test

# 测试特定区块范围
substreams run substreams.yaml map_events -s 64218055 -t +100

📞 支持和问题

如果遇到问题，请检查:

1. 认证: 确保 SUBSTREAMS_API_TOKEN 正确设置
2. 网络: 确保可以访问 bsc.substreams.pinax.network:443
3. 依赖: 确保安装了所有必要的工具和依赖
4. 日志: 查看详细的错误日志进行诊断

更多技术支持，请参考:

• Substreams 官方文档
• Protocol Buffers 文档
• BSC 网络信息

📄 许可证

本项目遵循与 newBroker 相同的许可证。