Add eth_listener.py

eth_listener.py

@@ -0,0 +1,197 @@
+#!/usr/bin/env python3
+"""
+Vector Zulu — Ethereum Event Listener
+Watches SV token contracts for Mint, Burn, Transfer events.
+Converts them to KoreID payloads and queues for posting to KoreChain.
+"""
+
+import os, json, time, logging
+from datetime import datetime, timezone
+from web3 import Web3
+import redis
+from koreid import KoreIDGenerator, wrap_ethereum_event
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s [ETH-LISTENER] %(levelname)s %(message)s'
+)
+log = logging.getLogger(__name__)
+
+# ── Config ────────────────────────────────────────────────────────────────────
+
+ETH_RPC   = os.environ.get('ETH_RPC', 'https://eth-mainnet.g.alchemy.com/v2/nRVT_IneMUDF3Pokubawx')
+POLL_SECS = int(os.environ.get('POLL_INTERVAL', '60'))
+
+# SV token contracts to monitor
+SV_TOKENS = {
+    '0x0A925D0F18090d2953CaC1F90b71c79fEd0efdB2': {'symbol': 'SVUSD', 'currency': 'USD'},
+    '0xec2Cb021F1d768bC987480381d79640AfD52Af54': {'symbol': 'SVGBP', 'currency': 'GBP'},
+    '0x37A09BF6288Fbe451588992BEb3760Afdb330c1c': {'symbol': 'SVEUR', 'currency': 'EUR'},
+    '0xE2fE8f436a154c9E9C81Ef1C4049070cE3374581': {'symbol': 'SVCHF', 'currency': 'CHF'},
+    '0xACa671CfEc4956F43A1770BccC02D663BA5BB80e': {'symbol': 'SVJPY', 'currency': 'JPY'},
+    '0x79b12192451B8cE7c70488E4C179183E93266f93': {'symbol': 'SVSGD', 'currency': 'SGD'},
+}
+
+# ERC-20 events we care about
+MINT_TOPIC     = Web3.keccak(text='Mint(address,uint256,uint256)').hex()
+BURN_TOPIC     = Web3.keccak(text='Burn(address,uint256,uint256)').hex()
+TRANSFER_TOPIC = Web3.keccak(text='Transfer(address,address,uint256)').hex()
+
+ZERO_ADDRESS = '0x0000000000000000000000000000000000000000'
+
+ERC20_ABI = [
+    {"type":"event","name":"Transfer","inputs":[
+        {"name":"from","type":"address","indexed":True},
+        {"name":"to","type":"address","indexed":True},
+        {"name":"value","type":"uint256","indexed":False}
+    ]},
+    {"type":"event","name":"Mint","inputs":[
+        {"name":"to","type":"address","indexed":True},
+        {"name":"amount","type":"uint256","indexed":False},
+        {"name":"newTotalSupply","type":"uint256","indexed":False}
+    ]},
+    {"type":"event","name":"Burn","inputs":[
+        {"name":"from","type":"address","indexed":True},
+        {"name":"amount","type":"uint256","indexed":False},
+        {"name":"newTotalSupply","type":"uint256","indexed":False}
+    ]},
+]
+
+
+class EthereumListener:
+
+    def __init__(self, rpc_url: str, redis_client, pg_conn=None):
+        self.w3       = Web3(Web3.HTTPProvider(rpc_url))
+        self.r        = redis_client
+        self.gen      = KoreIDGenerator(pg_conn)
+        self.last_block = self._get_start_block()
+        log.info(f"Connected to Ethereum: block {self.w3.eth.block_number}")
+        log.info(f"Starting from block: {self.last_block}")
+
+    def _get_start_block(self) -> int:
+        """Resume from last processed block or start from recent."""
+        saved = None
+        try:
+            saved = self.r.get('eth_listener:last_block')
+        except Exception:
+            pass
+        if saved:
+            return int(saved)
+        # Start from 100 blocks ago on first run
+        return self.w3.eth.block_number - 100
+
+    def _save_block(self, block_num: int):
+        try:
+            self.r.set('eth_listener:last_block', block_num)
+        except Exception:
+            pass
+
+    def _queue_payload(self, payload: dict):
+        """Queue KoreID payload for posting to KoreChain."""
+        try:
+            self.r.lpush('korechain:outbound_queue', json.dumps(payload))
+            log.info(f"Queued KoreID: {payload['koreid']} for {payload['source']['event_type']} {payload['source']['token']}")
+        except Exception as e:
+            log.error(f"Queue error: {e}")
+
+    def process_transfer_log(self, log_entry, token_addr: str):
+        """Process an ERC-20 Transfer event."""
+        token_info = SV_TOKENS.get(token_addr.lower(), SV_TOKENS.get(token_addr))
+        if not token_info:
+            return
+
+        try:
+            contract = self.w3.eth.contract(
+                address=Web3.to_checksum_address(token_addr),
+                abi=ERC20_ABI
+            )
+
+            # Decode Transfer event
+            decoded = contract.events.Transfer().process_log(log_entry)
+            from_addr = decoded['args']['from']
+            to_addr   = decoded['args']['to']
+            value     = decoded['args']['value']
+            amount    = value / 1e6  # 6 decimals
+
+            # Determine event type
+            if from_addr == ZERO_ADDRESS:
+                event_type = 'MINT'
+            elif to_addr == ZERO_ADDRESS:
+                event_type = 'BURN'
+            else:
+                event_type = 'TRANSFER'
+
+            # Skip small transfers (dust)
+            if amount < 1.0:
+                return
+
+            payload = wrap_ethereum_event(
+                event_type=event_type,
+                amount=amount,
+                currency=token_info['currency'],
+                tx_hash=log_entry['transactionHash'].hex(),
+                block_number=log_entry['blockNumber'],
+                from_address=from_addr,
+                to_address=to_addr,
+                token_symbol=token_info['symbol'],
+                koreid_generator=self.gen,
+            )
+
+            self._queue_payload(payload)
+
+        except Exception as e:
+            log.error(f"Log processing error: {e}")
+
+    def poll(self):
+        """Poll for new events since last block."""
+        try:
+            current_block = self.w3.eth.block_number
+            if current_block <= self.last_block:
+                return
+
+            from_block = self.last_block + 1
+            to_block   = min(current_block, from_block + 1000)  # max 1000 blocks per poll
+
+            log.info(f"Scanning blocks {from_block} → {to_block}")
+
+            addresses = list(SV_TOKENS.keys())
+
+            # Get Transfer events (covers Mint/Burn via from/to zero address)
+            logs = self.w3.eth.get_logs({
+                'fromBlock': from_block,
+                'toBlock':   to_block,
+                'address':   [Web3.to_checksum_address(a) for a in addresses],
+                'topics':    [TRANSFER_TOPIC],
+            })
+
+            for log_entry in logs:
+                token_addr = log_entry['address'].lower()
+                self.process_transfer_log(log_entry, token_addr)
+
+            self.last_block = to_block
+            self._save_block(to_block)
+
+            if logs:
+                log.info(f"Processed {len(logs)} events up to block {to_block}")
+
+        except Exception as e:
+            log.error(f"Poll error: {e}")
+
+    def run(self):
+        log.info("Ethereum event listener starting...")
+        while True:
+            self.poll()
+            time.sleep(POLL_SECS)
+
+
+def get_redis():
+    return redis.Redis(
+        host=os.environ.get('REDIS_HOST', 'localhost'),
+        port=6379, decode_responses=True
+    )
+
+
+if __name__ == '__main__':
+    r = get_redis()
+    listener = EthereumListener(ETH_RPC, r)
+    listener.run()