""" Parser for the Solana Token Program (SPL) """ import base64 from typing import Dict, List, Optional, Tuple from app.parsers.base import TransferParser # Token Program Instructions TOKEN_TRANSFER_INSTRUCTION = 3 TOKEN_PROGRAM_ID = "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" class TokenProgramParser(TransferParser): """ Parser for the Solana Token Program instructions """ @property def program_id(self) -> str: return TOKEN_PROGRAM_ID def parse_instruction(self, instruction: Dict, accounts: List[str], instruction_data: bytes) -> Dict: """ Parse a token program instruction """ if not instruction_data: return {"type": "unknown", "error": "No instruction data"} # First byte is the instruction type instruction_type = instruction_data[0] if instruction_type == TOKEN_TRANSFER_INSTRUCTION: # Parse transfer instruction return self._parse_transfer(instruction, accounts, instruction_data) # For other instruction types return { "type": f"token_instruction_{instruction_type}", "program": "token_program", "program_id": TOKEN_PROGRAM_ID, "accounts": [accounts[idx] for idx in instruction.get("accounts", [])], "data": base64.b64encode(instruction_data).decode("utf-8"), } def _parse_transfer(self, instruction: Dict, accounts: List[str], instruction_data: bytes) -> Dict: """ Parse a token transfer instruction """ # For token transfers, the accounts are: # 0: Source account # 1: Destination account # 2: Owner of source account # (3: Signers if multisig, optional) instruction_accounts = instruction.get("accounts", []) if len(instruction_accounts) < 3: return { "type": "token_transfer", "error": "Not enough accounts for transfer", "program": "token_program", "program_id": TOKEN_PROGRAM_ID, } # Parse amount from instruction data # The format is: [instruction_type(1 byte), amount(8 bytes)] try: amount_data = instruction_data[1:9] amount = int.from_bytes(amount_data, byteorder="little") except Exception as e: return { "type": "token_transfer", "error": f"Failed to parse amount: {str(e)}", "program": "token_program", "program_id": TOKEN_PROGRAM_ID, } source_idx = instruction_accounts[0] dest_idx = instruction_accounts[1] owner_idx = instruction_accounts[2] return { "type": "token_transfer", "program": "token_program", "program_id": TOKEN_PROGRAM_ID, "source": accounts[source_idx] if source_idx < len(accounts) else None, "destination": accounts[dest_idx] if dest_idx < len(accounts) else None, "owner": accounts[owner_idx] if owner_idx < len(accounts) else None, "amount": amount, "decimals": None, # We don't know the token decimals from instruction alone } def extract_transfer_info(self, parsed_instruction: Dict) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[float]]: """ Extract transfer information from a parsed instruction Returns: Tuple of (token_address, from_address, to_address, amount) """ if parsed_instruction.get("type") != "token_transfer": return None, None, None, None source = parsed_instruction.get("source") destination = parsed_instruction.get("destination") amount = parsed_instruction.get("amount") # Note: For SPL tokens, the actual token mint (token_address) is not directly # included in the instruction, so we would need additional context to determine it return None, source, destination, amount