Adapt endpoints

electrumx/server/session/session_manager.py

@@ -837,7 +837,11 @@ def validate_raw_tx_blueprint(self, raw_tx, raise_if_burned=True) -> AtomicalsVa
         )
 
     # Helper method to decode the transaction and returns formatted structure.
-    async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
+    async def transaction_decode_raw_tx_blueprint(
+        self,
+        raw_tx: bytes,
+        tap_leafs: Optional[List[bytes]],
+    ) -> dict:
         # Deserialize the transaction
         tx, tx_hash = self.env.coin.DESERIALIZER(raw_tx, 0).read_tx_and_hash()
         cache_res = self._tx_decode_cache.get(tx_hash)
@@ -848,7 +852,10 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
             return cache_res
 
         # Determine if there are any other operations at the transfer
-        operations_found_at_inputs = parse_protocols_operations_from_witness_array(tx, tx_hash, True)
+        if tap_leafs:
+            found_operations = parse_atomicals_operations_from_tap_leafs(tap_leafs, True)
+        else:
+            found_operations = parse_protocols_operations_from_witness_array(tx, tx_hash, True)
         # Build the map of the atomicals potential spent at the tx
         atomicals_spent_at_inputs: Dict[int:List] = self.bp.build_atomicals_spent_at_inputs_for_validation_only(tx)
         # Build a structure of organizing into NFT and FTs
@@ -857,7 +864,7 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
         blueprint_builder = AtomicalsTransferBlueprintBuilder(
             self.logger,
             atomicals_spent_at_inputs,
-            operations_found_at_inputs,
+            found_operations,
             tx_hash,
             tx,
             self.bp.get_atomicals_id_mint_info,
@@ -870,8 +877,8 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
         encoded_spent_at_inputs = encode_atomical_ids_hex(atomicals_spent_at_inputs)
         encoded_ft_output_blueprint: Dict[str, Dict] = dict(encode_atomical_ids_hex(ft_output_blueprint))
         encoded_nft_output_blueprint: Dict[str, Dict] = dict(encode_atomical_ids_hex(nft_output_blueprint))
-        op = operations_found_at_inputs.get("op") or "transfer"
-        payload = operations_found_at_inputs.get("payload")
+        op = found_operations.get("op") or "transfer"
+        payload = found_operations.get("payload")
         ret = {
             "op": [op],
             "burned": {
@@ -905,7 +912,7 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
                 if not outputs.get(k3):
                     outputs[k3] = {}
                 outputs[k3][atomical_id] = item3.atomical_value
-        mint_info = {}
+        mint_info: Dict | None = None
         if blueprint_builder.is_mint:
             if op in ["dmt", "ft"]:
                 tx_out = tx.outputs[0]
@@ -915,7 +922,6 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
                 mint_info = {
                     "atomical_id": atomical_id,
                     "outputs": {
-                        "atomical_id": atomical_id,
                         "index": 0,
                         "value": tx_out.value,
                     },
@@ -927,7 +933,6 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
                 mint_info = {
                     "atomical_id": atomical_id,
                     "outputs": {
-                        "atomical_id": atomical_id,
                         "index": 0,
                         "value": tx_out.value,
                     },
@@ -939,7 +944,7 @@ async def transaction_decode_raw_tx_blueprint(self, raw_tx: bytes) -> dict:
             if not outputs.get(index):
                 outputs[index] = {}
             outputs[index][atomical_id] = value
-        payment_info = {}
+        payment_info: Dict | None = None
         (
             payment_id,
             payment_idx,

electrumx/server/session/shared_session.py

@@ -1,14 +1,14 @@
 import asyncio
 import datetime
 from logging import LoggerAdapter
-from typing import TYPE_CHECKING, Callable, Optional, Union
+from typing import TYPE_CHECKING, Callable, Union, Optional
 
 from aiorpcx import RPCError
 
 from electrumx.lib import util
 from electrumx.lib.atomicals_blueprint_builder import AtomicalsValidationError
+from electrumx.lib.psbt import parse_psbt_hex_and_operations
 from electrumx.lib.script2addr import get_address_from_output_script
-from electrumx.lib.tx import psbt_hex_to_tx_hex
 from electrumx.lib.util_atomicals import *
 from electrumx.server.daemon import DaemonError
 from electrumx.server.session import ATOMICALS_INVALID_TX, BAD_REQUEST
@@ -964,7 +964,7 @@ async def transaction_broadcast_force(self, raw_tx: str):
             return hex_hash
 
     def transaction_validate_psbt_blueprint(self, psbt_hex: str):
-        raw_tx = psbt_hex_to_tx_hex(psbt_hex)
+        raw_tx, _ = parse_psbt_hex_and_operations(psbt_hex)
         return self.transaction_validate_tx_blueprint(raw_tx)
 
     def transaction_validate_tx_blueprint(self, raw_tx: str):
@@ -973,13 +973,16 @@ def transaction_validate_tx_blueprint(self, raw_tx: str):
         return {"result": dict(result)}
 
     async def transaction_decode_psbt(self, psbt_hex: str):
-        tx = psbt_hex_to_tx_hex(psbt_hex)
-        return await self.transaction_decode_tx(tx)
+        tx, tap_leafs = parse_psbt_hex_and_operations(psbt_hex)
+        return await self._transaction_decode(tx, tap_leafs)
 
     async def transaction_decode_tx(self, tx: str):
+        return await self._transaction_decode(tx)
+
+    async def _transaction_decode(self, tx: str, tap_leafs=None):
         raw_tx = bytes.fromhex(tx)
         self.bump_cost(0.25 + len(raw_tx) / 5000)
-        result = await self.session_mgr.transaction_decode_raw_tx_blueprint(raw_tx)
+        result = await self.session_mgr.transaction_decode_raw_tx_blueprint(raw_tx, tap_leafs)
         self.logger.debug(f"transaction_decode: {result}")
         return {"result": result}