diff --git a/eth_listener.py b/eth_listener.py new file mode 100644 index 0000000..817d05b --- /dev/null +++ b/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()