#!/usr/bin/env python3 """ LuckySt Syndicate - On-Chain Report Writer (Base) Writes encrypted trading session reports to Base as self-transfer transactions with report data in the calldata. Uses raw JSON-RPC + eth-account (no web3.py) via Base RPC. ERC-8021 Builder Code suffix is appended for tx attribution. """ import os import json from dataclasses import dataclass from typing import Optional import httpx from eth_account import Account from eth_account.types import TransactionDictType # Reuse network config from pledge module NETWORK = os.environ.get("LUCKYST_NETWORK", "mainnet") NETWORK_CONFIG = { "mainnet": { "chain_id": 8453, "rpc_url": "https://mainnet.base.org", "name": "Base Mainnet", }, "testnet": { "chain_id": 84532, "rpc_url": "https://sepolia.base.org", "name": "Base Sepolia", }, } _config = NETWORK_CONFIG[NETWORK] BASE_CHAIN_ID = _config["chain_id"] BASE_RPC_URL = _config["rpc_url"] # --------------------------------------------------------------------------- # ERC-8021 Builder Code # --------------------------------------------------------------------------- def build_erc8021_suffix(entity_code: str = "luckyst") -> bytes: """ Build an ERC-8021 Builder Code suffix (24 bytes). Format: [len:1][entity:N][schema:1][marker:M] - len: length of entity code string - entity: UTF-8 entity code - schema: 0x00 - marker: ERC-8021 marker bytes, padded to fill 24 bytes total Returns 24 bytes to append to transaction calldata. """ entity_bytes = entity_code.encode("utf-8") length_byte = len(entity_bytes).to_bytes(1, "big") schema_byte = b"\x00" # ERC-8021 marker pattern marker = b"\x80\x21" * 8 # repeating 8021 pattern # Assemble: [len][entity][schema][marker trimmed to 24 bytes total] prefix = length_byte + entity_bytes + schema_byte suffix = prefix + marker[: 24 - len(prefix)] return suffix # Default suffix for LuckySt LUCKYST_BUILDER_SUFFIX = build_erc8021_suffix( os.environ.get("LUCKYST_BUILDER_CODE", "luckyst") ) # --------------------------------------------------------------------------- # JSON-RPC helpers # --------------------------------------------------------------------------- async def _rpc_call(method: str, params: list, rpc_url: str = BASE_RPC_URL) -> dict: """Make a JSON-RPC call to Base RPC.""" payload = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1} async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(rpc_url, json=payload) return resp.json() async def _get_nonce(address: str) -> int: result = await _rpc_call("eth_getTransactionCount", [address, "latest"]) return int(result["result"], 16) async def _get_gas_price() -> int: result = await _rpc_call("eth_gasPrice", []) return int(result["result"], 16) async def _send_raw_tx(raw_tx_hex: str) -> str: result = await _rpc_call("eth_sendRawTransaction", [raw_tx_hex]) if "error" in result: raise RuntimeError(f"TX failed: {result['error']}") return result["result"] async def _get_tx_receipt(tx_hash: str) -> Optional[dict]: result = await _rpc_call("eth_getTransactionReceipt", [tx_hash]) return result.get("result") # --------------------------------------------------------------------------- # Report writer # --------------------------------------------------------------------------- @dataclass class ReportTxResult: """Result of writing a report to Base.""" success: bool tx_hash: Optional[str] = None block_number: Optional[int] = None gas_used: Optional[int] = None error: Optional[str] = None async def write_report_to_chain( encrypted_report: bytes, private_key: str, from_address: str, ) -> ReportTxResult: """ Write an encrypted report to Base as a self-transfer transaction. The transaction is: to=from, value=0, data=encrypted_report + ERC-8021 suffix. Uses EIP-1559 (Type 2) transactions via Base RPC. Args: encrypted_report: The encrypted report bytes (version + nonce + ciphertext) private_key: Hex private key for signing from_address: The sender's EVM address (tx sends to self) Returns: ReportTxResult with tx_hash on success """ try: # Build calldata: encrypted report + ERC-8021 suffix calldata = encrypted_report + LUCKYST_BUILDER_SUFFIX calldata_hex = "0x" + calldata.hex() # Get nonce and gas price nonce = await _get_nonce(from_address) gas_price = await _get_gas_price() # Estimate gas — self-transfer with calldata # Base gas model follows standard EVM rules # Base cost: 21000 + 16 per non-zero byte + 4 per zero byte non_zero = sum(1 for b in calldata if b != 0) zero = len(calldata) - non_zero estimated_gas = 21000 + (non_zero * 16) + (zero * 4) + 5000 # buffer # Build EIP-1559 transaction tx: TransactionDictType = { "type": 2, "chainId": BASE_CHAIN_ID, "nonce": nonce, "to": from_address, # self-transfer "value": 0, "data": calldata_hex, "gas": estimated_gas, "maxFeePerGas": gas_price * 2, "maxPriorityFeePerGas": gas_price, } # Sign and send signed = Account.sign_transaction(tx, private_key) raw_tx = "0x" + signed.raw_transaction.hex() tx_hash = await _send_raw_tx(raw_tx) # Wait for receipt (poll up to 30s) import asyncio for _ in range(15): await asyncio.sleep(2) receipt = await _get_tx_receipt(tx_hash) if receipt: status = int(receipt.get("status", "0x0"), 16) if status == 1: return ReportTxResult( success=True, tx_hash=tx_hash, block_number=int(receipt["blockNumber"], 16), gas_used=int(receipt["gasUsed"], 16), ) else: return ReportTxResult( success=False, tx_hash=tx_hash, error="Transaction reverted", ) # Timeout — tx may still confirm later return ReportTxResult( success=True, tx_hash=tx_hash, error="Receipt not confirmed within 30s (tx may still succeed)", ) except Exception as e: return ReportTxResult(success=False, error=str(e)) async def fetch_tx_calldata(tx_hash: str) -> Optional[bytes]: """ Fetch the calldata from a Base transaction by hash. Used by report readers to retrieve encrypted report data. Returns raw bytes (stripped of ERC-8021 suffix). """ result = await _rpc_call("eth_getTransactionByHash", [tx_hash]) tx = result.get("result") if not tx or not tx.get("input"): return None raw = bytes.fromhex(tx["input"].replace("0x", "")) # Strip ERC-8021 suffix (last 24 bytes) if len(raw) > 24: return raw[:-24] return raw