# Author: Simon-Pierre Boucher # Mail: contact@spboucher.ai """Tron adapter — TRC-20 stablecoin transfers via the HTTP wallet API. How a TRC-20 transfer appears on Tron: the tx is a `TriggerSmartContract` call, and the token contract emits the SAME Transfer event as ERC-20 (Tron's TVM is EVM-derived). We read one block's worth of transaction infos with /wallet/gettransactioninfobyblocknum and filter the `log` entries by contract address + Transfer topic. Addresses: Tron shows Base58Check "T..." addresses; on-chain they are 21 bytes (0x41 prefix + 20-byte EVM-style address). Event topics carry the bare 20 bytes left-padded to 32 — we convert back to T-addresses so stored rows use the chain-native format. Base58Check is implemented by hand below (double-SHA256 checksum) — no dependency. Finality: DPoS, ~3s blocks; a block is effectively final once 2/3 of the 27 SRs confirm (≈ 19 blocks) — we trail head by `confirmations` (20). """ import hashlib import logging import time from .. import db from ..rpc import AllEndpointsDown, RestError, RestPool log = logging.getLogger("tron") # keccak256("Transfer(address,address,uint256)") — Tron logs omit the 0x TRANSFER_TOPIC = "ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" _B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" _B58_INDEX = {c: i for i, c in enumerate(_B58)} def _sha256d(b): return hashlib.sha256(hashlib.sha256(b).digest()).digest() def b58check_decode(s): num = 0 for c in s: num = num * 58 + _B58_INDEX[c] raw = num.to_bytes((num.bit_length() + 7) // 8, "big") raw = b"\x00" * (len(s) - len(s.lstrip("1"))) + raw # leading '1's = zero bytes payload, checksum = raw[:-4], raw[-4:] if _sha256d(payload)[:4] != checksum: raise ValueError(f"bad Base58Check checksum for {s}") return payload def b58check_encode(payload): raw = payload + _sha256d(payload)[:4] num = int.from_bytes(raw, "big") out = "" while num: num, rem = divmod(num, 58) out = _B58[rem] + out return "1" * (len(raw) - len(raw.lstrip(b"\x00"))) + out def taddr_to_hex20(t_addr): """T-address → bare 20-byte hex (what appears in event logs/topics).""" payload = b58check_decode(t_addr) if len(payload) != 21 or payload[0] != 0x41: raise ValueError(f"not a Tron address: {t_addr}") return payload[1:].hex() def hex20_to_taddr(h): """Bare 20-byte hex → Base58Check T-address (0x41 mainnet prefix).""" return b58check_encode(b"\x41" + bytes.fromhex(h)) class TronIndexer: def __init__(self, chain, cfg, tokens, db_path): self.chain = chain self.cfg = cfg self.db_path = db_path self.conn = None self.pool = RestPool(cfg["rpcs"]) # hex20 (as seen in logs) → token meta self.tokens = {taddr_to_hex20(t["id"]): t for t in tokens} self.native = cfg.get("native") # {symbol: TRX, min, decimals: 6} # -- API helpers ----------------------------------------------------- def head(self): blk = self.pool.post("/wallet/getnowblock") return blk["block_header"]["raw_data"]["number"] def block_id(self, n): blk = self.pool.post("/wallet/getblockbynum", {"num": n}) return (blk or {}).get("blockID") def block_infos(self, n): return self.pool.post("/wallet/gettransactioninfobyblocknum", {"num": n}) or [] # -- pipeline ---------------------------------------------------------- def process_block(self, n): rows = [] for info in self.block_infos(n): receipt = info.get("receipt") or {} if receipt.get("result") not in (None, "SUCCESS"): continue # reverted contract call — logs are not effective for i, lg in enumerate(info.get("log") or []): addr = (lg.get("address") or "").lower()[-40:] # strip 41 prefix if present meta = self.tokens.get(addr) topics = lg.get("topics") or [] if meta is None or len(topics) != 3 or topics[0].lower() != TRANSFER_TOPIC: continue data = (lg.get("data") or "").strip() or "0" rows.append({ "chain": self.chain, "block": n, "block_hash": None, "tx_hash": info["id"], "log_index": i, "timestamp": info.get("blockTimeStamp", 0) // 1000 or None, "token": meta["id"], # T-address, chain-native format "symbol": meta["symbol"], "from": hex20_to_taddr(topics[1][-40:]), "to": hex20_to_taddr(topics[2][-40:]), "amount": str(int(data, 16)), "decimals": meta["decimals"], }) if self.native: rows.extend(self.native_rows(n)) db.insert_transfers(self.conn, rows) return len(rows) def native_rows(self, n): """Whale-only native TRX: TransferContract txs above the config min. Addresses come back as 41-prefixed hex — convert to T-addresses.""" min_sun = int(self.native["min"] * 10 ** self.native["decimals"]) blk = self.pool.post("/wallet/getblockbynum", {"num": n}) or {} ts = (blk.get("block_header", {}).get("raw_data", {}).get("timestamp", 0)) // 1000 or None rows = [] for tx in blk.get("transactions") or []: for ci, c in enumerate(tx.get("raw_data", {}).get("contract") or []): if c.get("type") != "TransferContract": continue v = c.get("parameter", {}).get("value", {}) amount = v.get("amount", 0) if amount < min_sun: continue def cvt(h): h = (h or "").lower() return hex20_to_taddr(h[-40:]) if len(h) >= 40 else (h or None) rows.append({ "chain": self.chain, "block": n, "block_hash": None, "tx_hash": tx["txID"], "log_index": 100000 + ci, "timestamp": ts, "token": "native", "symbol": self.native["symbol"], "from": cvt(v.get("owner_address")), "to": cvt(v.get("to_address")), "amount": str(amount), "decimals": self.native["decimals"], }) return rows def reorged(self, cursor_block, cursor_hash): if not cursor_hash: return False bid = self.block_id(cursor_block) return bid is not None and bid.lower() != cursor_hash.lower() # -- main loop ----------------------------------------------------- def run(self, stop): self.conn = db.connect(self.db_path) cur = db.get_cursor(self.conn, self.chain) confirmations = self.cfg.get("confirmations", 20) while not stop.is_set(): try: head = self.head() safe = head - confirmations if cur is None: start = max(0, safe - self.cfg.get("start_offset", 100)) cur = (start, None, start) db.set_cursor(self.conn, self.chain, start, None) db.set_backfill(self.conn, self.chain, start) log.info("%s: fresh start at block %d (head %d)", self.chain, start, head) cursor_block, cursor_hash, _ = cur if self.reorged(cursor_block, cursor_hash): to_block = cursor_block - 2 * confirmations log.warning("%s: reorg at %d — rolling back to %d", self.chain, cursor_block, to_block) db.rollback(self.conn, self.chain, to_block) cursor_block, cursor_hash = to_block, None moved = 0 while cursor_block < safe and not stop.is_set(): n = self.process_block(cursor_block + 1) cursor_block += 1 moved += 1 if n: log.info("%s: %d transfers in block %d (lag %d)", self.chain, n, cursor_block, head - cursor_block) if moved % 20 == 0: # persist progress during catch-up db.set_cursor(self.conn, self.chain, cursor_block, None) if moved: cursor_hash = self.block_id(cursor_block) db.set_cursor(self.conn, self.chain, cursor_block, cursor_hash, head=head) cur = (cursor_block, cursor_hash, None) db.save_rpc_health(self.conn, self.chain, self.pool.stats(), int(time.time())) stop.wait(max(float(self.cfg.get("block_time", 3)), 2.0)) except (RestError, AllEndpointsDown, Exception) as e: log.error("%s: %s — retrying in 10s", self.chain, e) stop.wait(10)