111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
"""
|
|
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 |