feat(indexer): complete deterministic projection pipeline and replay tests

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                 # 架构铁律
+```

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"]

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

src/config.py

@@ -0,0 +1,24 @@
+from pydantic_settings import BaseSettings
+
+class IndexerSettings(BaseSettings):
+    CHAIN_ID: int = 46630 # Robinhood Testnet
+    RPC_URL: str = "https://rpc.testnet.chain.robinhood.com"
+    START_BLOCK: int = 0
+    BATCH_SIZE: int = 100
+    POLL_INTERVAL_SECONDS: float = 2.0
+    CONFIRMATIONS: int = 1 # Testnet 确认数
+    
+    # 数据库与 Redis
+    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/wtfx"
+    REDIS_URL: str = "redis://localhost:6379/0"
+    
+    # 已部署合约地址 (Robinhood Testnet)
+    CONTROLLER_ADDRESS: str = "0xc0E24E152771C588B21AEB654b30B1cBAf381c1a"
+    COLLATERAL_ADDRESS: str = "0xe776e957953EA69b7Eaa9d7d4098aBC076bDD5E7"
+    CURVE_ADDRESS: str = "0xF0E189974c413506098AFB6Bc6Bf4C6715fBf7B1"
+
+    class Config:
+        env_file = ".env"
+        extra = "ignore"
+
+indexer_settings = IndexerSettings()

src/core/abis.py

@@ -0,0 +1,22 @@
+import json
+import os
+from typing import Dict, Any
+
+def load_contract_abi(name: str) -> list:
+    """加载共享合约 ABI"""
+    # 优先从 frontend packages 读取
+    possible_paths = [
+        os.path.join(os.path.dirname(__file__), "..", "..", "..", "wtf-frontend", "packages", "contracts", "src", "abis", f"{name}.json"),
+        os.path.join(os.path.dirname(__file__), "..", "..", "..", "wtf-contract", "artifacts", "main", "src", "controllerv2", f"{name}.sol", f"{name}.json"),
+        os.path.join(os.path.dirname(__file__), "..", "..", "..", "wtf-contract", "artifacts", "main", "src", "marketv2", f"{name}.sol", f"{name}.json")
+    ]
+    for p in possible_paths:
+        if os.path.exists(p):
+            with open(p, "r", encoding="utf-8") as f:
+                data = json.load(f)
+                return data.get("abi", data) if isinstance(data, dict) else data
+    return []
+
+CONTROLLER_ABI = load_contract_abi("WTFControllerV2")
+MARKET_ABI = load_contract_abi("WTFMarketV2")
+MOCK_ERC20_ABI = load_contract_abi("MockERC20")

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

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}"

src/decoding/evm.py

@@ -0,0 +1,109 @@
+import json
+import logging
+from typing import Dict, Any, Optional
+from web3 import Web3
+from eth_abi import decode
+from src.core.events import NormalizedEvent
+from src.core.abis import CONTROLLER_ABI, MARKET_ABI
+
+logger = logging.getLogger(__name__)
+
+class EVMDecoder:
+    """EVM ABI 日志规范化解码器 (Layer 2: Decoding)"""
+
+    def __init__(self, chain_id: int):
+        self.chain_id = chain_id
+        self.w3 = Web3()
+        self._build_topic_maps()
+
+    def _build_topic_maps(self):
+        self.event_abi_map = {}
+        
+        # 解析 Controller 与 Market ABI 中的 events
+        for abi in CONTROLLER_ABI + MARKET_ABI:
+            if abi.get("type") == "event":
+                name = abi.get("name")
+                inputs = abi.get("inputs", [])
+                types = [i["type"] for i in inputs]
+                sig = f"{name}({','.join(types)})"
+                topic0 = self.w3.keccak(text=sig).hex()
+                self.event_abi_map[topic0] = abi
+
+    def decode_log(self, log: Dict[str, Any], block_timestamp: int = 0) -> Optional[NormalizedEvent]:
+        """将原始 EVM 日志解析为统一的 NormalizedEvent"""
+        topics = log.get("topics", [])
+        if not topics:
+            return None
+
+        topic0 = topics[0].hex() if isinstance(topics[0], bytes) else topics[0]
+        abi = self.event_abi_map.get(topic0)
+        if not abi:
+            return None
+
+        event_name = abi["name"]
+        contract_addr = log.get("address", "").lower()
+        tx_hash = log.get("transactionHash", "").hex() if isinstance(log.get("transactionHash"), bytes) else str(log.get("transactionHash", ""))
+        block_number = log.get("blockNumber", 0)
+        log_index = log.get("logIndex", 0)
+
+        # 解码 Indexed 参数与 Non-Indexed Data
+        payload: Dict[str, Any] = {}
+        indexed_inputs = [i for i in abi.get("inputs", []) if i.get("indexed")]
+        non_indexed_inputs = [i for i in abi.get("inputs", []) if not i.get("indexed")]
+
+        # 1. Indexed topics
+        for idx, inp in enumerate(indexed_inputs):
+            if idx + 1 < len(topics):
+                raw_topic = topics[idx + 1]
+                t_hex = raw_topic.hex() if isinstance(raw_topic, bytes) else raw_topic
+                payload[inp["name"]] = t_hex
+
+        # 2. Non-indexed data
+        data = log.get("data", "0x")
+        if isinstance(data, str) and data.startswith("0x"):
+            data_bytes = bytes.fromhex(data[2:])
+        elif isinstance(data, bytes):
+            data_bytes = data
+        else:
+            data_bytes = b""
+
+        if data_bytes and non_indexed_inputs:
+            types = [i["type"] for i in non_indexed_inputs]
+            try:
+                decoded_vals = decode(types, data_bytes)
+                for inp, val in zip(non_indexed_inputs, decoded_vals):
+                    if isinstance(val, bytes):
+                        payload[inp["name"]] = "0x" + val.hex()
+                    else:
+                        payload[inp["name"]] = val
+            except Exception as e:
+                logger.error(f"Failed to decode non-indexed data for {event_name}: {e}")
+
+        # 标准化 Event Type
+        type_mapping = {
+            "DeployMarket": "MARKET_DEPLOYED",
+            "SetOutcome": "OUTCOME_RESOLVED",
+            "Finalise": "MARKET_FINALISED",
+            "Mint": "ORDER_MINT",
+            "Redeem": "ORDER_REDEEM",
+            "Claim": "POSITION_CLAIMED",
+            "SetProtocolFeeRate": "GOV_FEE_RATE_UPDATED",
+            "SetTreasury": "GOV_TREASURY_UPDATED",
+            "SetCentralWallet": "GOV_CENTRAL_WALLET_UPDATED",
+            "SetCreatorShare": "GOV_CREATOR_SHARE_UPDATED",
+            "Paused": "PROTOCOL_PAUSED",
+            "Unpaused": "PROTOCOL_UNPAUSED",
+        }
+
+        normalized_type = type_mapping.get(event_name, f"EVM_{event_name.upper()}")
+
+        return NormalizedEvent(
+            chain_id=self.chain_id,
+            block_number=block_number,
+            tx_hash=tx_hash,
+            log_index=log_index,
+            event_type=normalized_type,
+            contract_address=contract_addr,
+            payload=payload,
+            timestamp=block_timestamp
+        )

src/ingestion/fetcher.py

@@ -0,0 +1,50 @@
+import asyncio
+import logging
+from typing import List, Dict, Any, Optional
+from web3 import AsyncWeb3, AsyncHTTPProvider
+from web3.types import LogReceipt, BlockNumber
+from src.config import indexer_settings
+
+logger = logging.getLogger(__name__)
+
+class EVMFetcher:
+    """EVM 批量日志与区块抓取器 (Layer 1: Ingestion)"""
+
+    def __init__(self, rpc_url: Optional[str] = None, chain_id: Optional[int] = None):
+        self.rpc_url = rpc_url or indexer_settings.RPC_URL
+        self.chain_id = chain_id or indexer_settings.CHAIN_ID
+        self.w3 = AsyncWeb3(AsyncHTTPProvider(self.rpc_url))
+
+    async def get_latest_block_number(self) -> int:
+        """获取链上最新安全区块高度"""
+        block_num = await self.w3.eth.block_number
+        return max(0, block_num - indexer_settings.CONFIRMATIONS)
+
+    async def fetch_logs(
+        self,
+        from_block: int,
+        to_block: int,
+        addresses: Optional[List[str]] = None,
+        topics: Optional[List[Any]] = None
+    ) -> List[LogReceipt]:
+        """批量获取指定区块区间的日志"""
+        filter_params: Dict[str, Any] = {
+            "fromBlock": from_block,
+            "toBlock": to_block
+        }
+        if addresses:
+            filter_params["address"] = [self.w3.to_checksum_address(a) for a in addresses]
+        if topics:
+            filter_params["topics"] = topics
+
+        try:
+            logs = await self.w3.eth.get_logs(filter_params)
+            return logs
+        except Exception as e:
+            logger.error(f"Error fetching logs from {from_block} to {to_block}: {e}")
+            raise e
+
+    async def get_block_timestamp(self, block_number: int) -> int:
+        """获取区块出块时间戳"""
+        block = await self.w3.eth.get_block(block_number)
+        return block.get("timestamp", 0)

src/main.py

@@ -0,0 +1,81 @@
+import asyncio
+import logging
+import signal
+import sys
+from typing import Optional
+from src.config import indexer_settings
+from src.ingestion.fetcher import EVMFetcher
+from src.decoding.evm import EVMDecoder
+from src.projections.store import ProjectionStore
+from src.core.checkpoint import CheckpointState
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
+logger = logging.getLogger("WTFX-Indexer")
+
+class IndexerService:
+    """WTFX 确定性索引服务核心执行引擎"""
+
+    def __init__(self):
+        self.fetcher = EVMFetcher(indexer_settings.RPC_URL, indexer_settings.CHAIN_ID)
+        self.decoder = EVMDecoder(indexer_settings.CHAIN_ID)
+        self.projection_store = ProjectionStore()
+        self.checkpoint = CheckpointState(
+            chain_id=indexer_settings.CHAIN_ID,
+            last_processed_block=indexer_settings.START_BLOCK
+        )
+        self.is_running = False
+
+    async def run(self):
+        self.is_running = True
+        logger.info(f"Starting WTFX Indexer on Chain {indexer_settings.CHAIN_ID} (RPC: {indexer_settings.RPC_URL})")
+        logger.info(f"Monitoring Controller: {indexer_settings.CONTROLLER_ADDRESS}")
+
+        while self.is_running:
+            try:
+                latest_block = await self.fetcher.get_latest_block_number()
+                current_block = self.checkpoint.last_processed_block
+
+                if current_block < latest_block:
+                    to_block = min(current_block + indexer_settings.BATCH_SIZE, latest_block)
+                    logger.info(f"Scanning blocks {current_block + 1} -> {to_block} (Latest: {latest_block})...")
+
+                    # 1. Fetch raw logs
+                    logs = await self.fetcher.fetch_logs(
+                        from_block=current_block + 1,
+                        to_block=to_block
+                    )
+
+                    # 2. Decode & Normalize
+                    for log in logs:
+                        normalized = self.decoder.decode_log(log)
+                        if normalized:
+                            logger.info(f"Decoded Event: {normalized.event_type} (Tx: {normalized.tx_hash[:10]}...)")
+                            # 3. Apply to Projection
+                            self.projection_store.apply_event(normalized)
+
+                    # 4. Advance Checkpoint
+                    self.checkpoint.last_processed_block = to_block
+                    self.checkpoint.updated_at = int(asyncio.get_event_loop().time())
+                else:
+                    await asyncio.sleep(indexer_settings.POLL_INTERVAL_SECONDS)
+
+            except Exception as e:
+                logger.error(f"Error in indexer polling loop: {e}", exc_info=True)
+                await asyncio.sleep(indexer_settings.POLL_INTERVAL_SECONDS * 2)
+
+    def stop(self):
+        logger.info("Stopping Indexer service...")
+        self.is_running = False
+
+async def main():
+    service = IndexerService()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        try:
+            loop.add_signal_handler(sig, service.stop)
+        except NotImplementedError:
+            pass  # Windows compatibility
+    await service.run()
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/projections/store.py

@@ -0,0 +1,79 @@
+import logging
+from typing import Dict, Any, Optional
+from src.core.events import NormalizedEvent
+
+logger = logging.getLogger(__name__)
+
+class ProjectionStore:
+    """
+    确定性业务投影状态存储 (Layer 4: Projections)
+    在第一阶段使用内存/本地轻量结构，并支持向 PostgreSQL 写入投影，支持随时 Reset & Replay
+    """
+
+    def __init__(self):
+        self.markets: Dict[str, Dict[str, Any]] = {}
+        self.trades: list = []
+        self.positions: Dict[str, Dict[str, Any]] = {}
+        self.processed_events: set = set()
+
+    def reset(self):
+        """全量清空投影视图（支持 Replay）"""
+        self.markets.clear()
+        self.trades.clear()
+        self.positions.clear()
+        self.processed_events.clear()
+        logger.info("Projection store reset successfully.")
+
+    def apply_event(self, event: NormalizedEvent):
+        """确定性事件投影应用"""
+        if event.event_id in self.processed_events:
+            return  # 幂等拦截
+        self.processed_events.add(event.event_id)
+
+        etype = event.event_type
+        payload = event.payload
+
+        if etype == "MARKET_DEPLOYED":
+            market_addr = payload.get("market") or event.contract_address
+            self.markets[market_addr] = {
+                "market_address": market_addr,
+                "question_id": payload.get("questionId"),
+                "curve": payload.get("curve"),
+                "collateral": payload.get("collateral"),
+                "creator": payload.get("creator"),
+                "tier": payload.get("tier", 1),
+                "fee_rate": payload.get("feeRate"),
+                "status": "ACTIVE",
+                "winning_outcome": None,
+                "created_at_block": event.block_number,
+                "created_at_tx": event.tx_hash,
+                "timestamp": event.timestamp
+            }
+
+        elif etype == "OUTCOME_RESOLVED":
+            q_id = payload.get("questionId")
+            for m in self.markets.values():
+                if m.get("question_id") == q_id:
+                    m["status"] = "RESOLVED"
+                    m["tentative_winning_outcome"] = payload.get("outcome")
+
+        elif etype == "MARKET_FINALISED":
+            q_id = payload.get("questionId")
+            for m in self.markets.values():
+                if m.get("question_id") == q_id:
+                    m["status"] = "FINALISED"
+                    m["winning_outcome"] = payload.get("outcome")
+
+        elif etype in ("ORDER_MINT", "ORDER_REDEEM"):
+            self.trades.append({
+                "tx_hash": event.tx_hash,
+                "block_number": event.block_number,
+                "market": event.contract_address,
+                "type": "MINT" if etype == "ORDER_MINT" else "REDEEM",
+                "user": payload.get("user") or payload.get("buyer") or payload.get("seller"),
+                "outcome_index": payload.get("outcomeIndex") or payload.get("index"),
+                "collateral_amount": payload.get("collateralAmount") or payload.get("amountIn"),
+                "tokens_amount": payload.get("tokensAmount") or payload.get("amountOut"),
+                "fee": payload.get("fee", 0),
+                "timestamp": event.timestamp
+            })

tests/test_indexer_pipeline.py

@@ -0,0 +1,76 @@
+import pytest
+from src.core.events import NormalizedEvent
+from src.projections.store import ProjectionStore
+from src.decoding.evm import EVMDecoder
+
+def test_normalized_event_idempotency():
+    event = NormalizedEvent(
+        chain_id=46630,
+        block_number=1000,
+        tx_hash="0xabcdef123456",
+        log_index=2,
+        event_type="MARKET_DEPLOYED",
+        contract_address="0xc0E24E152771C588B21AEB654b30B1cBAf381c1a",
+        payload={
+            "market": "0x1111222233334444555566667777888899990000",
+            "questionId": "0x9999",
+            "tier": 1,
+            "creator": "0x6cddF384792C77219fc25C454Cfe264757842830"
+        },
+        timestamp=1700000000
+    )
+    assert event.event_id == "46630:0xabcdef123456:2"
+
+def test_projection_store_deploy_and_resolve():
+    store = ProjectionStore()
+
+    deploy_event = NormalizedEvent(
+        chain_id=46630,
+        block_number=1000,
+        tx_hash="0xabcdef123456",
+        log_index=0,
+        event_type="MARKET_DEPLOYED",
+        contract_address="0xc0E24E152771C588B21AEB654b30B1cBAf381c1a",
+        payload={
+            "market": "0x1111222233334444555566667777888899990000",
+            "questionId": "0x9999",
+            "tier": 1,
+            "creator": "0x6cddF384792C77219fc25C454Cfe264757842830"
+        },
+        timestamp=1700000000
+    )
+    store.apply_event(deploy_event)
+    assert "0x1111222233334444555566667777888899990000" in store.markets
+    assert store.markets["0x1111222233334444555566667777888899990000"]["status"] == "ACTIVE"
+
+    # Duplicate should be ignored
+    store.apply_event(deploy_event)
+    assert len(store.markets) == 1
+
+    # Resolve event
+    resolve_event = NormalizedEvent(
+        chain_id=46630,
+        block_number=1050,
+        tx_hash="0xabcdef789012",
+        log_index=1,
+        event_type="OUTCOME_RESOLVED",
+        contract_address="0xc0E24E152771C588B21AEB654b30B1cBAf381c1a",
+        payload={
+            "questionId": "0x9999",
+            "outcome": 1
+        },
+        timestamp=1700005000
+    )
+    store.apply_event(resolve_event)
+    assert store.markets["0x1111222233334444555566667777888899990000"]["status"] == "RESOLVED"
+    assert store.markets["0x1111222233334444555566667777888899990000"]["tentative_winning_outcome"] == 1
+
+    # Test Reset & Replay
+    store.reset()
+    assert len(store.markets) == 0
+    assert len(store.processed_events) == 0
+
+    # Replay
+    store.apply_event(deploy_event)
+    store.apply_event(resolve_event)
+    assert store.markets["0x1111222233334444555566667777888899990000"]["status"] == "RESOLVED"