diff --git a/src/monitoring/universal_block_listener.py b/src/monitoring/universal_block_listener.py index be90396..0403c9d 100644 --- a/src/monitoring/universal_block_listener.py +++ b/src/monitoring/universal_block_listener.py @@ -3,14 +3,17 @@ Universal block listener that works with any platform through the interface syst """ import asyncio +import base64 import json from collections.abc import Awaitable, Callable import websockets +from solders.transaction import VersionedTransaction +from core.client import SolanaClient from interfaces.core import Platform, TokenInfo from monitoring.base_listener import BaseTokenListener -from platforms import get_platform_implementations +from platforms import get_platform_implementations, platform_factory from utils.logger import get_logger logger = get_logger(__name__) @@ -23,7 +26,7 @@ class UniversalBlockListener(BaseTokenListener): self, wss_endpoint: str, platforms: list[Platform] | None = None, - ): + ) -> None: """Initialize universal block listener. Args: @@ -34,9 +37,7 @@ class UniversalBlockListener(BaseTokenListener): self.wss_endpoint = wss_endpoint self.ping_interval = 20 # seconds - # Import platform factory and get supported platforms - from platforms import platform_factory - + # Get supported platforms if platforms is None: # Monitor all supported platforms self.platforms = platform_factory.get_supported_platforms() @@ -46,15 +47,14 @@ class UniversalBlockListener(BaseTokenListener): # Get event parsers for all platforms self.platform_parsers = {} self.platform_program_ids = [] + # Map program IDs to their parsers for faster lookup + self.program_id_to_parser = {} for platform in self.platforms: try: - # Create a simple dummy client that doesn't start blockhash updater - from core.client import SolanaClient - # Create a mock client class to avoid network operations class DummyClient(SolanaClient): - def __init__(self): + def __init__(self) -> None: # Skip SolanaClient.__init__ to avoid starting blockhash updater self.rpc_endpoint = "http://dummy" self._client = None @@ -67,7 +67,9 @@ class UniversalBlockListener(BaseTokenListener): implementations = get_platform_implementations(platform, dummy_client) parser = implementations.event_parser self.platform_parsers[platform] = parser - self.platform_program_ids.append(str(parser.get_program_id())) + program_id_str = str(parser.get_program_id()) + self.platform_program_ids.append(program_id_str) + self.program_id_to_parser[program_id_str] = (platform, parser) logger.info( f"Registered platform {platform.value} with program ID {parser.get_program_id()}" @@ -139,7 +141,9 @@ class UniversalBlockListener(BaseTokenListener): logger.info("Reconnecting in 5 seconds...") await asyncio.sleep(5) - async def _subscribe_to_programs(self, websocket) -> None: + async def _subscribe_to_programs( + self, websocket: websockets.WebSocketServerProtocol + ) -> None: """Subscribe to blocks mentioning any of the monitored program IDs. Args: @@ -169,7 +173,7 @@ class UniversalBlockListener(BaseTokenListener): await websocket.send(subscription_message) logger.info(f"Subscribed to blocks mentioning program: {program_id}") - async def _ping_loop(self, websocket) -> None: + async def _ping_loop(self, websocket: websockets.WebSocketServerProtocol) -> None: """Keep connection alive with pings. Args: @@ -191,7 +195,9 @@ class UniversalBlockListener(BaseTokenListener): except Exception: logger.exception("Ping error") - async def _wait_for_token_creation(self, websocket) -> TokenInfo | None: + async def _wait_for_token_creation( + self, websocket: websockets.WebSocketServerProtocol + ) -> TokenInfo | None: """Wait for token creation event from any platform. Args: @@ -204,6 +210,14 @@ class UniversalBlockListener(BaseTokenListener): response = await asyncio.wait_for(websocket.recv(), timeout=30) data = json.loads(response) + # Handle subscription errors + if "error" in data: + logger.error(f"Block subscription error: {data['error']}") + return None + elif "result" in data: + # Subscription confirmation - continue waiting for notifications + return None + if "method" not in data or data["method"] != "blockNotification": return None @@ -218,21 +232,8 @@ class UniversalBlockListener(BaseTokenListener): if "transactions" not in block: return None - # Try each platform's event parser on each transaction - for tx in block["transactions"]: - if not isinstance(tx, dict) or "transaction" not in tx: - continue - - for platform, parser in self.platform_parsers.items(): - # Check if the parser has a block parsing method - if hasattr(parser, "parse_token_creation_from_block"): - token_info = parser.parse_token_creation_from_block( - {"transactions": [tx]} - ) - if token_info: - return token_info - - return None + # Process all transactions in the block for token creations + return self._process_block_transactions(block["transactions"]) except TimeoutError: logger.debug("No data received for 30 seconds") @@ -243,3 +244,117 @@ class UniversalBlockListener(BaseTokenListener): logger.exception("Error processing WebSocket message") return None + + def _process_block_transactions(self, transactions: list) -> TokenInfo | None: + """Process all transactions in a block looking for token creations. + + Args: + transactions: List of transaction data from block + + Returns: + TokenInfo if a token creation is found, None otherwise + """ + for tx in transactions: + if not isinstance(tx, dict) or "transaction" not in tx: + continue + + tx_data = tx["transaction"] + + # Handle base64 encoded transaction data + if isinstance(tx_data, list) and len(tx_data) > 0: + token_info = self._parse_encoded_transaction(tx, tx_data[0]) + if token_info: + return token_info + + # Handle already decoded transaction data (shouldn't happen in blockSubscribe) + elif isinstance(tx_data, dict) and "message" in tx_data: + token_info = self._parse_decoded_transaction(tx, tx_data) + if token_info: + return token_info + + return None + + def _parse_encoded_transaction( + self, tx: dict, encoded_data: str + ) -> TokenInfo | None: + """Parse base64 encoded transaction data. + + Args: + tx: Transaction wrapper from block + encoded_data: Base64 encoded transaction data + + Returns: + TokenInfo if token creation found, None otherwise + """ + try: + tx_bytes = base64.b64decode(encoded_data) + transaction = VersionedTransaction.from_bytes(tx_bytes) + + # Check if any of the instructions use our monitored programs + for instruction in transaction.message.instructions: + program_id = str( + transaction.message.account_keys[instruction.program_id_index] + ) + + # Check if this program ID is one we're monitoring + if program_id in self.program_id_to_parser: + platform, parser = self.program_id_to_parser[program_id] + + # Try to parse with the appropriate parser + try: + if hasattr(parser, "parse_token_creation_from_block"): + token_info = parser.parse_token_creation_from_block( + {"transactions": [tx]} + ) + if token_info: + return token_info + except Exception: + # Expected for non-creation transactions + continue + + except Exception: + # Failed to decode transaction - skip it + pass + + return None + + def _parse_decoded_transaction(self, tx: dict, tx_data: dict) -> TokenInfo | None: + """Parse already decoded transaction data. + + Args: + tx: Transaction wrapper from block + tx_data: Decoded transaction data + + Returns: + TokenInfo if token creation found, None otherwise + """ + message = tx_data["message"] + if "instructions" not in message or "accountKeys" not in message: + return None + + for ix in message["instructions"]: + if "programIdIndex" not in ix: + continue + + program_idx = ix["programIdIndex"] + if program_idx >= len(message["accountKeys"]): + continue + + program_id = message["accountKeys"][program_idx] + + # Check if this program ID is one we're monitoring + if program_id in self.program_id_to_parser: + platform, parser = self.program_id_to_parser[program_id] + + try: + if hasattr(parser, "parse_token_creation_from_block"): + token_info = parser.parse_token_creation_from_block( + {"transactions": [tx]} + ) + if token_info: + return token_info + except Exception: + # Expected for non-creation transactions + continue + + return None