diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 0000000..8fd7e85 --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,73 @@ +# WTFX Indexer Architecture & Principles (索引器架构与核心铁律) + +> **定位**:Indexer 不是数据库搬运工,而是 **“区块链事实 → WTFX 可查询状态” 的确定性投影系统**。 +> **运行形态**:第一阶段单进程运行,但严格按照 **Fetch -> Decode -> Process -> Project** 四层边界设计,支持**零停机热追高、全量可重放、幂等去重、防链重组 (Reorg)**。 + +--- + +## 1. 五大核心架构铁律 (Non-Negotiable Principles) + +1. **【三层数据流转:Raw -> Normalized -> Projection】**: + - **Raw Layer**:原样保存链上原始日志 (`raw_logs`, `raw_blocks`)。 + - **Normalized Layer**:解析为跨链统一的标准事件字典 (`chain_events`: `ORDER_FILLED`, `MARKET_CREATED`, `SETTLEMENT`)。 + - **Projection Layer**:生成业务最终可查询的只读视图 (`markets`, `trades`, `positions`, `token_balances`)。 +2. **【绝对可重放性 (Replayability)】**: + - 当业务投影算法调整或修复时,**严禁手工修数据库**; + - 必须支持:清空 Projection 视图 $ ightarrow$ 从指定 Checkpoint 或 Block 0 重新重放 Normalized Events $ ightarrow$ 确定性重新生成 Projection。 +3. **【幂等去重保证 (Idempotency)】**: + - 每个事件唯一凭证为 `(chain_id, tx_hash, log_index)`; + - 数据库唯一约束拦截重复写入,重复消费或追高扫描时无脑跳过已存在事件。 +4. **【安全检查点与防重组 (Checkpoint & Reorg Protection)】**: + - 维护持久化 `indexer_state` 记录安全落盘水位; + - 区分 `CONFIRMED`(快速响应)与 `FINALIZED`(安全落盘),为最新区块预留回退重测窗口。 +5. **【不直接越权控制业务资金】**: + - Indexer 只负责**“发现链上发生了什么”**,严禁直接在 Indexer 内部修改用户余额; + - 资金结算与账本生成交由 Backend 的 Settlement Service 处理。 + +--- + +## 2. 目录结构全景 + +``` +wtf-indexer/ +├── src/ +│ ├── main.py # 索引器服务入口与管道调度器 (Pipeline Runner) +│ ├── config.py # RPC、链配置、数据库与扫描区间配置 +│ │ +│ ├── core/ # 核心抽象与不可变数据模型 +│ │ ├── events.py # 规范化事件定义 (Normalized Event Schema) +│ │ ├── checkpoint.py # Checkpoint 水位管理器与 Reorg 检测 +│ │ └── models.py # 基础类型与接口抽象 +│ │ +│ ├── ingestion/ # 链上数据拉取层 (Fetcher) +│ │ ├── fetcher.py # RPC 批量抓块与重试调度 (Batch eth_getLogs) +│ │ └── queue.py # 内存/Redis 缓冲队列 +│ │ +│ ├── decoding/ # 事件解析层 (Decoder) +│ │ ├── evm.py # EVM ABI 日志解码器 (WTFMarketV2, Controller) +│ │ └── base.py # 解码器抽象基类 +│ │ +│ ├── processors/ # 业务领域处理器 (Processor) +│ │ ├── market.py # 市场创建/状态流转处理 +│ │ ├── trading.py # 订单成交与代币铸造/赎回处理 +│ │ └── settlement.py # 到期裁决与兑付结算处理 +│ │ +│ ├── projections/ # 确定性投影层 (Projection Views) +│ │ ├── markets.py # 市场聚合状态表投影 +│ │ ├── trades.py # 逐笔交易记录投影 +│ │ └── positions.py # 用户持仓与成本投影 +│ │ +│ ├── chains/ # 链适配器 (Chain Adapters) +│ │ ├── robinhood.py # Robinhood Chain (Testnet/Mainnet) 适配器 +│ │ └── base.py # 通用 EVM 适配器 +│ │ +│ └── infrastructure/ # 底层设施 +│ ├── postgres.py # 异步 PostgreSQL 连接与投影持久化 +│ ├── redis.py # Redis 锁与事件总线广播 +│ └── rpc.py # 健壮的多节点 RPC 客户端 +│ +├── tests/ # 可重放测试与解码单元测试 +├── Dockerfile # 容器构建镜像 +├── requirements.txt # 依赖包配置 +└── ARCHITECTURE.md # 架构铁律 +``` diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..862f08c --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential \ + libpq-dev \ + curl \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY . . + +CMD ["python", "-m", "src.main"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..83bd15b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +web3>=6.19.0 +eth-abi>=5.1.0 +pydantic>=2.7.0 +pydantic-settings>=2.3.0 +asyncpg>=0.29.0 +sqlalchemy[asyncio]>=2.0.30 +redis>=5.0.4 +pytest>=8.2.0 +pytest-asyncio>=0.23.7 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/chains/__init__.py b/src/chains/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..f6f4439 --- /dev/null +++ b/src/config.py @@ -0,0 +1,20 @@ +from pydantic_settings import BaseSettings + +class IndexerSettings(BaseSettings): + CHAIN_ID: int = 46630 # Robinhood Testnet default + RPC_URL: str = "https://rpc.testnet.chain.robinhood.com" + START_BLOCK: int = 0 + BATCH_SIZE: int = 100 + POLL_INTERVAL_SECONDS: float = 2.0 + + # 数据库与 Redis + DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/wtfx" + REDIS_URL: str = "redis://localhost:6379/0" + + # 控制器合约地址 + CONTROLLER_ADDRESS: str = "0x0000000000000000000000000000000000000000" + + class Config: + env_file = ".env" + +indexer_settings = IndexerSettings() diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/checkpoint.py b/src/core/checkpoint.py new file mode 100644 index 0000000..07b9d92 --- /dev/null +++ b/src/core/checkpoint.py @@ -0,0 +1,7 @@ +from pydantic import BaseModel + +class CheckpointState(BaseModel): + chain_id: int + last_processed_block: int + last_processed_tx: str = "" + updated_at: int = 0 diff --git a/src/core/events.py b/src/core/events.py new file mode 100644 index 0000000..5496f18 --- /dev/null +++ b/src/core/events.py @@ -0,0 +1,18 @@ +from pydantic import BaseModel, Field +from typing import Dict, Any, Optional + +class NormalizedEvent(BaseModel): + """跨链与跨合约的标准归一化事件 (Layer 2)""" + chain_id: int + block_number: int + tx_hash: str + log_index: int + event_type: str = Field(description="e.g. MARKET_CREATED, MINT_COLLATERAL, REDEEM_OT, RESOLVE_OUTCOME, CLAIM") + contract_address: str + payload: Dict[str, Any] + timestamp: int + + @property + def event_id(self) -> str: + """全局唯一幂等标识 (Idempotency Key)""" + return f"{self.chain_id}:{self.tx_hash}:{self.log_index}" diff --git a/src/decoding/__init__.py b/src/decoding/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/infrastructure/__init__.py b/src/infrastructure/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/ingestion/__init__.py b/src/ingestion/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..fdd70cc --- /dev/null +++ b/src/main.py @@ -0,0 +1,22 @@ +import asyncio +from src.config import indexer_settings + +async def run_indexer_pipeline(): + """ + 可重放的确定性流水线: + Fetcher (拉块) -> Decoder (规范化) -> Processor (业务处理) -> Projection (确定性视图落盘) + """ + print(f"Starting WTFX Indexer for Chain [{indexer_settings.CHAIN_ID}]...") + print(f"Target RPC: {indexer_settings.RPC_URL}") + + current_block = indexer_settings.START_BLOCK + while True: + # 1. 抓取安全区间块日志 (Fetcher) + # 2. 解码并生成 NormalizedEvent (Decoder) + # 3. 幂等去重检查 (Idempotency Check) + # 4. 业务处理与 Projection 视图更新 (Processor & Projection) + # 5. 更新 Checkpoint + await asyncio.sleep(indexer_settings.POLL_INTERVAL_SECONDS) + +if __name__ == "__main__": + asyncio.run(run_indexer_pipeline()) diff --git a/src/processors/__init__.py b/src/processors/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/projections/__init__.py b/src/projections/__init__.py new file mode 100644 index 0000000..e69de29