diff --git a/chia/rpc/full_node_rpc_api.py b/chia/rpc/full_node_rpc_api.py index c51426e015c6..521534cce6ca 100644 --- a/chia/rpc/full_node_rpc_api.py +++ b/chia/rpc/full_node_rpc_api.py @@ -15,7 +15,7 @@ from chia.types.blockchain_format.coin import Coin from chia.types.blockchain_format.sized_bytes import bytes32 from chia.types.coin_record import CoinRecord -from chia.types.coin_spend import CoinSpend +from chia.types.coin_spend import CoinSpend, compute_additions from chia.types.full_block import FullBlock from chia.types.generator_types import BlockGenerator from chia.types.mempool_inclusion_status import MempoolInclusionStatus @@ -72,6 +72,7 @@ def get_routes(self) -> Dict[str, Endpoint]: "/push_tx": self.push_tx, "/get_puzzle_and_solution": self.get_puzzle_and_solution, "/get_puzzles_and_solutions_by_names":self.get_puzzles_and_solutions_by_names, + "/get_singleton_by_launcher_id": self.get_singleton_by_launcher_id, # Mempool "/get_all_mempool_tx_ids": self.get_all_mempool_tx_ids, "/get_all_mempool_items": self.get_all_mempool_items, @@ -944,6 +945,62 @@ async def get_puzzles_and_solutions_by_names(self, request: Dict[str, Any]) -> E return {'coin_solutions': coin_spends} + async def get_coin_spend_for_coin_record(self, coin_record: CoinRecord) -> Optional[CoinSpend]: + if not coin_record.spent: + return + + height= coin_record.spent_block_index + + header_hash = self.service.blockchain.height_to_hash(height) + assert header_hash is not None + block: Optional[FullBlock] = await self.service.block_store.get_full_block(header_hash) + + if block is None or block.transactions_generator is None: + return + + block_generator: Optional[BlockGenerator] = await self.service.blockchain.get_block_generator(block) + if block_generator is None: + return + + spend_info = get_puzzle_and_solution_for_coin(block_generator, coin_record.coin) + + return CoinSpend(coin_record.coin, spend_info.puzzle, spend_info.solution) + + async def get_singleton_addition(self, parent_spend: CoinSpend) -> Optional[CoinRecord]: + additions: List[Coin] = compute_additions(parent_spend) + + filtered_additions: List[Coin] = list(filter(lambda coin: coin.amount % 2 == 1,additions)) + + if len(filtered_additions) != 1: + raise ValueError(f"Invalid singleton no single odd child coin.") + + coin_record: Optional[CoinRecord] = await self.service.blockchain.coin_store.get_coin_record(filtered_additions[0].name()) + + return coin_record + + async def get_singleton_by_launcher_id(self, request: Dict[str, Any]) -> EndpointResult: + if "launcher_id" not in request: + raise ValueError("Launcher ID not in request") + launcher_id = bytes32.from_hexstr(request["launcher_id"]) + + singleton_coin_record: Optional[CoinRecord] = await self.service.blockchain.coin_store.get_coin_record(launcher_id) + + if singleton_coin_record is None: + raise ValueError(f"Launcher coin not found for ID {launcher_id.hex()}") + + while singleton_coin_record.spent_block_index > 0: + singleton_parent_spend = await self.get_coin_spend_for_coin_record(singleton_coin_record) + + singleton_coin_record = await self.get_singleton_addition(singleton_parent_spend) + + if singleton_coin_record is None: + raise ValueError("Singleton coin record not found") + + return { + "coin_record": coin_record_dict_backwards_compat(singleton_coin_record.to_json_dict()), + "parent_spend": singleton_parent_spend.to_json_dict() + } + async def get_additions_and_removals(self, request: Dict[str, Any]) -> EndpointResult: if "header_hash" not in request: raise ValueError("No header_hash in request")