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

src/config.py

@@ -1,20 +1,24 @@
 from pydantic_settings import BaseSettings
 
 class IndexerSettings(BaseSettings):
-    CHAIN_ID: int = 46630 # Robinhood Testnet default
+    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 = "0x0000000000000000000000000000000000000000"
+    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/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

@@ -1,22 +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
 
-async def run_indexer_pipeline():
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
-    """
+logger = logging.getLogger("WTFX-Indexer")
-    可重放的确定性流水线:
-    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
+class IndexerService:
-    while True:
+    """WTFX 确定性索引服务核心执行引擎"""
-        # 1. 抓取安全区间块日志 (Fetcher)
+
-        # 2. 解码并生成 NormalizedEvent (Decoder)
+    def __init__(self):
-        # 3. 幂等去重检查 (Idempotency Check)
+        self.fetcher = EVMFetcher(indexer_settings.RPC_URL, indexer_settings.CHAIN_ID)
-        # 4. 业务处理与 Projection 视图更新 (Processor & Projection)
+        self.decoder = EVMDecoder(indexer_settings.CHAIN_ID)
-        # 5. 更新 Checkpoint
+        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(run_indexer_pipeline())
+    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"