fix(core): handle block listener errors

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