Add finalize_transfers API

rpc/src/client.rs

@@ -22,7 +22,7 @@ use psbt::Psbt;
 use rgb::schema::TransitionType;
 use rgb::{Contract, ContractId, ContractState, ContractStateMap, SealEndpoint, StateTransfer};
 
-use crate::messages::{HelloReq, TransferFinalize};
+use crate::messages::{FinalizeTransfersRes, HelloReq, TransferFinalize, TransfersReq};
 use crate::{
     AcceptReq, BusMsg, ComposeReq, ContractValidity, Error, FailureCode, OutpointFilter, Reveal,
     RpcMsg, ServiceId, TransferReq,
@@ -241,6 +241,22 @@ impl Client {
         }
     }
 
+    pub fn finalize_transfers(
+        &mut self,
+        transfers: Vec<(StateTransfer, Vec<SealEndpoint>)>,
+        psbt: Psbt,
+        progress: impl Fn(String),
+    ) -> Result<FinalizeTransfersRes, Error> {
+        self.request(RpcMsg::FinalizeTransfers(TransfersReq { transfers, psbt }))?;
+        loop {
+            match self.response()?.failure_to_error()? {
+                RpcMsg::FinalizedTransfers(transfers) => return Ok(transfers),
+                RpcMsg::Progress(info) => progress(info),
+                _ => return Err(Error::UnexpectedServerResponse),
+            }
+        }
+    }
+
     pub fn consume_transfer(
         &mut self,
         transfer: StateTransfer,

rpc/src/lib.rs

@@ -32,8 +32,8 @@ pub use client::Client;
 pub use error::{Error, FailureCode};
 pub(crate) use messages::BusMsg;
 pub use messages::{
-    AcceptReq, ComposeReq, ContractValidity, HelloReq, OutpointFilter, RpcMsg, TransferFinalize,
-    TransferReq,
+    AcceptReq, ComposeReq, ContractValidity, FinalizeTransfersRes, HelloReq, OutpointFilter,
+    RpcMsg, TransferFinalize, TransferReq, TransfersReq,
 };
 pub use reveal::Reveal;
 pub use service_id::ServiceId;

rpc/src/messages.rs

@@ -74,6 +74,9 @@ pub enum RpcMsg {
     #[display(inner)]
     Transfer(TransferReq),
 
+    #[display(inner)]
+    FinalizeTransfers(TransfersReq),
+
     #[display("memorize_seal({0})")]
     MemorizeSeal(seal::Revealed),
 
@@ -97,6 +100,9 @@ pub enum RpcMsg {
     #[display("state_transfer_finalize(...)")]
     StateTransferFinalize(TransferFinalize),
 
+    #[display("state_transfer_finalize(...)")]
+    FinalizedTransfers(FinalizeTransfersRes),
+
     #[display("progress(\"{0}\")")]
     #[from]
     Progress(String),
@@ -194,14 +200,30 @@ pub struct TransferReq {
     pub beneficiary: Option<NodeAddr>,
 }
 
-impl From<&str> for RpcMsg {
-    fn from(s: &str) -> Self { RpcMsg::Progress(s.to_owned()) }
-}
-
 #[derive(Clone, PartialEq, Eq, Debug, Display)]
 #[derive(NetworkEncode, NetworkDecode)]
 #[display("transfer_complete(...)")]
 pub struct TransferFinalize {
     pub consignment: StateTransfer,
     pub psbt: Psbt,
 }
+
+#[derive(Clone, PartialEq, Eq, Debug, Display)]
+#[derive(NetworkEncode, NetworkDecode)]
+#[display("transfers_req(...)")]
+pub struct TransfersReq {
+    pub transfers: Vec<(StateTransfer, Vec<SealEndpoint>)>,
+    pub psbt: Psbt,
+}
+
+#[derive(Clone, PartialEq, Eq, Debug, Display)]
+#[derive(NetworkEncode, NetworkDecode)]
+#[display("finalize_transfers_res(...)")]
+pub struct FinalizeTransfersRes {
+    pub consignments: Vec<StateTransfer>,
+    pub psbt: Psbt,
+}
+
+impl From<&str> for RpcMsg {
+    fn from(s: &str) -> Self { RpcMsg::Progress(s.to_owned()) }
+}

src/bucketd/processor.rs

@@ -24,7 +24,7 @@ use rgb::{
     OwnedRights, PedersenStrategy, Schema, SchemaId, SealEndpoint, StateTransfer, Transition,
     TransitionBundle, TypedAssignments, Validator, Validity,
 };
-use rgb_rpc::{OutpointFilter, Reveal, TransferFinalize};
+use rgb_rpc::{FinalizeTransfersRes, OutpointFilter, Reveal, TransferFinalize};
 use storm::chunk::ChunkIdExt;
 use storm::{ChunkId, Container, ContainerId};
 use strict_encoding::StrictDecode;
@@ -574,6 +574,52 @@ impl Runtime {
 
         Ok(TransferFinalize { consignment, psbt })
     }
+
+    pub(super) fn finalize_transfers(
+        &mut self,
+        transfers: Vec<(StateTransfer, Vec<SealEndpoint>)>,
+        mut psbt: Psbt,
+    ) -> Result<FinalizeTransfersRes, DaemonError> {
+        // 1. Pack LNPBP-4 and anchor information.
+        let mut bundles = psbt.rgb_bundles()?;
+        debug!("Found {} bundles", bundles.len());
+        trace!("Bundles: {:?}", bundles);
+
+        let anchor = Anchor::commit(&mut psbt)?;
+        trace!("Anchor: {:?}", anchor);
+
+        let mut consignments = vec![];
+        for (state_transfer, seal_endpoints) in &transfers {
+            let mut consignment = state_transfer.clone();
+            let contract_id = consignment.contract_id();
+            info!("Finalizing transfer for {}", contract_id);
+
+            // 2. Extract contract-related state transition from PSBT and put it
+            //    into consignment.
+            let bundle = bundles.remove(&contract_id).ok_or(FinalizeError::ContractBundleMissed)?;
+            let bundle_id = bundle.bundle_id();
+            consignment.push_anchored_bundle(anchor.to_merkle_proof(contract_id)?, bundle)?;
+
+            // 3. Add seal endpoints.
+            let endseals = seal_endpoints.clone();
+            for endseal in endseals {
+                consignment.push_seal_endpoint(bundle_id, endseal);
+            }
+
+            consignments.push(consignment);
+        }
+
+        // 4. Conceal all the state not related to the transfer.
+        // TODO: Conceal all the amounts except the last transition
+        // TODO: Conceal all seals outside of the paths from the endpoint to genesis
+
+        // 5. Construct and store disclosure for the blank transfers.
+        let txid = anchor.txid;
+        let disclosure = Disclosure::with(anchor, bundles, None);
+        self.store.store_sten(db::DISCLOSURES, txid, &disclosure)?;
+
+        Ok(FinalizeTransfersRes { consignments, psbt })
+    }
 }
 
 struct Collector {

src/bucketd/service.rs

@@ -40,8 +40,9 @@ use storm_rpc::AddressedMsg;
 use strict_encoding::{MediumVec, StrictEncode};
 
 use crate::bus::{
-    BusMsg, ConsignReq, CtlMsg, DaemonId, Endpoints, FinalizeTransferReq, OutpointStateReq,
-    ProcessDisclosureReq, ProcessReq, Responder, ServiceBus, ServiceId, ValidityResp,
+    BusMsg, ConsignReq, CtlMsg, DaemonId, Endpoints, FinalizeTransferReq, FinalizeTransfersReq,
+    OutpointStateReq, ProcessDisclosureReq, ProcessReq, Responder, ServiceBus, ServiceId,
+    ValidityResp,
 };
 use crate::{Config, DaemonError, LaunchError};
 
@@ -250,6 +251,14 @@ impl Runtime {
                 )?;
             }
 
+            CtlMsg::FinalizeTransfers(FinalizeTransfersReq {
+                client_id,
+                transfers,
+                psbt,
+            }) => {
+                self.handle_finalize_transfers(endpoints, client_id, transfers, psbt)?;
+            }
+
             wrong_msg => {
                 error!("Request is not supported by the CTL interface");
                 return Err(DaemonError::wrong_esb_msg(ServiceBus::Ctl, &wrong_msg));
@@ -485,4 +494,24 @@ impl Runtime {
         }
         Ok(())
     }
+
+    fn handle_finalize_transfers(
+        &mut self,
+        endpoints: &mut Endpoints,
+        client_id: ClientId,
+        transfers: Vec<(StateTransfer, Vec<SealEndpoint>)>,
+        psbt: Psbt,
+    ) -> Result<(), DaemonError> {
+        match self.finalize_transfers(transfers, psbt) {
+            Err(err) => {
+                let _ = self.send_rpc(endpoints, client_id, err);
+                self.send_ctl(endpoints, ServiceId::rgbd(), CtlMsg::ProcessingFailed)?
+            }
+            Ok(transfers) => {
+                let _ = self.send_rpc(endpoints, client_id, RpcMsg::FinalizedTransfers(transfers));
+                self.send_ctl(endpoints, ServiceId::rgbd(), CtlMsg::ProcessingComplete)?;
+            }
+        }
+        Ok(())
+    }
 }

src/bus/ctl.rs

@@ -54,6 +54,9 @@ pub enum CtlMsg {
     #[display(inner)]
     FinalizeTransfer(FinalizeTransferReq),
 
+    #[display(inner)]
+    FinalizeTransfers(FinalizeTransfersReq),
+
     #[display(inner)]
     #[from]
     Validity(ValidityResp),
@@ -117,3 +120,12 @@ pub struct FinalizeTransferReq {
     pub psbt: Psbt,
     pub beneficiary: Option<NodeAddr>,
 }
+
+#[derive(Clone, PartialEq, Eq, Debug, Display)]
+#[derive(NetworkEncode, NetworkDecode)]
+#[display("finalize_transfers({client_id}, ...)")]
+pub struct FinalizeTransfersReq {
+    pub client_id: ClientId,
+    pub transfers: Vec<(StateTransfer, Vec<SealEndpoint>)>,
+    pub psbt: Psbt,
+}

src/bus/mod.rs

@@ -16,8 +16,8 @@ use rgb_rpc::RpcMsg;
 use storm_ext::ExtMsg as StormMsg;
 
 pub use self::ctl::{
-    ConsignReq, CtlMsg, FinalizeTransferReq, OutpointStateReq, ProcessDisclosureReq, ProcessReq,
-    ValidityResp,
+    ConsignReq, CtlMsg, FinalizeTransferReq, FinalizeTransfersReq, OutpointStateReq,
+    ProcessDisclosureReq, ProcessReq, ValidityResp,
 };
 pub use self::services::{DaemonId, ServiceId};
 pub(crate) use self::services::{Endpoints, Responder, ServiceBus};

src/rgbd/service.rs

@@ -28,15 +28,16 @@ use rgb::{
 };
 use rgb_rpc::{
     AcceptReq, ComposeReq, FailureCode, HelloReq, OutpointFilter, Reveal, RpcMsg, TransferReq,
+    TransfersReq,
 };
 use storm::ContainerId;
 use storm_ext::ExtMsg as StormMsg;
 use storm_rpc::AddressedMsg;
 
 use crate::bucketd::StashError;
 use crate::bus::{
-    BusMsg, ConsignReq, CtlMsg, DaemonId, Endpoints, FinalizeTransferReq, OutpointStateReq,
-    ProcessDisclosureReq, ProcessReq, Responder, ServiceBus, ServiceId,
+    BusMsg, ConsignReq, CtlMsg, DaemonId, Endpoints, FinalizeTransferReq, FinalizeTransfersReq,
+    OutpointStateReq, ProcessDisclosureReq, ProcessReq, Responder, ServiceBus, ServiceId,
 };
 use crate::db::ChunkHolder;
 use crate::rgbd::daemons::Daemon;
@@ -277,6 +278,11 @@ impl Runtime {
                     beneficiary,
                 )?;
             }
+
+            RpcMsg::FinalizeTransfers(TransfersReq { transfers, psbt }) => {
+                self.complete_transfers(endpoints, client_id, transfers, psbt)?;
+            }
+
             wrong_msg => {
                 error!("Request is not supported by the RPC interface");
                 return Err(DaemonError::wrong_esb_msg(ServiceBus::Rpc, &wrong_msg));
@@ -558,4 +564,19 @@ impl Runtime {
         }));
         self.pick_or_start(endpoints, client_id)
     }
+
+    fn complete_transfers(
+        &mut self,
+        endpoints: &mut Endpoints,
+        client_id: ClientId,
+        transfers: Vec<(StateTransfer, Vec<SealEndpoint>)>,
+        psbt: Psbt,
+    ) -> Result<(), DaemonError> {
+        self.ctl_queue.push_back(CtlMsg::FinalizeTransfers(FinalizeTransfersReq {
+            client_id,
+            transfers,
+            psbt,
+        }));
+        self.pick_or_start(endpoints, client_id)
+    }
 }