feat: Add collab close dlc protocol event

coordinator/migrations/2024-05-08-104311_nullable_contract_id_on_dlc_protocols/down.sql

@@ -0,0 +1 @@
+ALTER TABLE dlc_protocols ALTER COLUMN contract_id SET NOT NULL;

coordinator/migrations/2024-05-08-104311_nullable_contract_id_on_dlc_protocols/up.sql

@@ -0,0 +1 @@
+ALTER TABLE dlc_protocols ALTER COLUMN contract_id DROP NOT NULL;

coordinator/src/bin/coordinator.rs

@@ -285,8 +285,7 @@ async fn main() -> Result<()> {
         network,
     );
 
-    node.spawn_shadow_dlc_channels_task();
-    node.spawn_watch_closing_channels();
+    node.spawn_watch_dlc_channel_events_task();
 
     tokio::spawn({
         let node = node.clone();

coordinator/src/db/dlc_channels.rs

@@ -1,4 +1,3 @@
-use crate::dlc_protocol::ProtocolId;
 use crate::node::channel;
 use crate::schema::dlc_channels;
 use crate::schema::sql_types::DlcChannelStateType;
@@ -24,6 +23,7 @@ use std::any::TypeId;
 use std::str::FromStr;
 use time::OffsetDateTime;
 use uuid::Uuid;
+use xxi_node::node::ProtocolId;
 
 #[derive(Debug, Clone, Copy, PartialEq, FromSqlRow, AsExpression)]
 #[diesel(sql_type = DlcChannelStateType)]
@@ -191,7 +191,7 @@ pub(crate) fn set_channel_collab_closing(
         .execute(conn)
 }
 
-pub(crate) fn set_channel_collab_closed(
+pub(crate) fn set_channel_closed(
     conn: &mut PgConnection,
     channel_id: &DlcChannelId,
     close_txid: Txid,

coordinator/src/db/dlc_protocols.rs

@@ -1,6 +1,5 @@
 use crate::db;
 use crate::dlc_protocol;
-use crate::dlc_protocol::ProtocolId;
 use crate::schema::dlc_protocols;
 use crate::schema::sql_types::ProtocolStateType;
 use crate::schema::sql_types::ProtocolTypeType;
@@ -21,6 +20,7 @@ use std::any::TypeId;
 use std::str::FromStr;
 use time::OffsetDateTime;
 use uuid::Uuid;
+use xxi_node::node::ProtocolId;
 
 #[derive(Debug, Clone, Copy, PartialEq, FromSqlRow, AsExpression, Eq, Hash)]
 #[diesel(sql_type = ProtocolStateType)]
@@ -68,7 +68,7 @@ pub(crate) struct DlcProtocol {
     pub protocol_id: Uuid,
     pub previous_protocol_id: Option<Uuid>,
     pub channel_id: String,
-    pub contract_id: String,
+    pub contract_id: Option<String>,
     pub protocol_state: DlcProtocolState,
     pub trader_pubkey: String,
     pub timestamp: OffsetDateTime,
@@ -113,9 +113,14 @@ pub(crate) fn get_dlc_protocol(
 
     let protocol = dlc_protocol::DlcProtocol {
         id: dlc_protocol.protocol_id.into(),
+        previous_id: dlc_protocol
+            .previous_protocol_id
+            .map(|previous_id| previous_id.into()),
         timestamp: dlc_protocol.timestamp,
         channel_id: DlcChannelId::from_hex(&dlc_protocol.channel_id).expect("valid dlc channel id"),
-        contract_id: ContractId::from_hex(&dlc_protocol.contract_id).expect("valid contract id"),
+        contract_id: dlc_protocol
+            .contract_id
+            .map(|cid| ContractId::from_hex(cid).expect("valid contract id")),
         trader: PublicKey::from_str(&dlc_protocol.trader_pubkey).expect("valid public key"),
         protocol_state: dlc_protocol.protocol_state.into(),
         protocol_type,
@@ -130,7 +135,7 @@ pub(crate) fn set_dlc_protocol_state_to_failed(
 ) -> QueryResult<()> {
     let affected_rows = diesel::update(dlc_protocols::table)
         .filter(dlc_protocols::protocol_id.eq(protocol_id.to_uuid()))
-        .set((dlc_protocols::protocol_state.eq(DlcProtocolState::Failed),))
+        .set(dlc_protocols::protocol_state.eq(DlcProtocolState::Failed))
         .execute(conn)?;
 
     if affected_rows == 0 {
@@ -143,14 +148,14 @@ pub(crate) fn set_dlc_protocol_state_to_failed(
 pub(crate) fn set_dlc_protocol_state_to_success(
     conn: &mut PgConnection,
     protocol_id: ProtocolId,
-    contract_id: &ContractId,
+    contract_id: Option<&ContractId>,
     channel_id: &DlcChannelId,
 ) -> QueryResult<()> {
     let affected_rows = diesel::update(dlc_protocols::table)
         .filter(dlc_protocols::protocol_id.eq(protocol_id.to_uuid()))
         .set((
             dlc_protocols::protocol_state.eq(DlcProtocolState::Success),
-            dlc_protocols::contract_id.eq(hex::encode(contract_id)),
+            dlc_protocols::contract_id.eq(contract_id.map(hex::encode)),
             dlc_protocols::channel_id.eq(hex::encode(channel_id)),
         ))
         .execute(conn)?;
@@ -166,7 +171,7 @@ pub(crate) fn create(
     conn: &mut PgConnection,
     protocol_id: ProtocolId,
     previous_protocol_id: Option<ProtocolId>,
-    contract_id: &ContractId,
+    contract_id: Option<&ContractId>,
     channel_id: &DlcChannelId,
     protocol_type: dlc_protocol::DlcProtocolType,
     trader: &PublicKey,
@@ -175,7 +180,7 @@ pub(crate) fn create(
         .values(&(
             dlc_protocols::protocol_id.eq(protocol_id.to_uuid()),
             dlc_protocols::previous_protocol_id.eq(previous_protocol_id.map(|ppid| ppid.to_uuid())),
-            dlc_protocols::contract_id.eq(hex::encode(contract_id)),
+            dlc_protocols::contract_id.eq(contract_id.map(hex::encode)),
             dlc_protocols::channel_id.eq(hex::encode(channel_id)),
             dlc_protocols::protocol_state.eq(DlcProtocolState::Pending),
             dlc_protocols::trader_pubkey.eq(trader.to_string()),

coordinator/src/db/trade_params.rs

@@ -1,5 +1,4 @@
 use crate::dlc_protocol;
-use crate::dlc_protocol::ProtocolId;
 use crate::orderbook::db::custom_types::Direction;
 use crate::schema::trade_params;
 use bitcoin::secp256k1::PublicKey;
@@ -14,6 +13,7 @@ use diesel::RunQueryDsl;
 use std::str::FromStr;
 use uuid::Uuid;
 use xxi_node::commons;
+use xxi_node::node::ProtocolId;
 
 #[derive(Queryable, Debug)]
 #[diesel(table_name = trade_params)]

coordinator/src/dlc_protocol.rs

@@ -14,99 +14,23 @@ use diesel::Connection;
 use diesel::PgConnection;
 use diesel::QueryResult;
 use dlc_manager::ContractId;
-use dlc_manager::ReferenceId;
 use rust_decimal::prelude::FromPrimitive;
 use rust_decimal::prelude::ToPrimitive;
 use rust_decimal::Decimal;
-use std::fmt::Display;
-use std::fmt::Formatter;
-use std::str::from_utf8;
 use time::OffsetDateTime;
 use tokio::sync::broadcast::Sender;
-use uuid::Uuid;
 use xxi_node::cfd::calculate_pnl;
 use xxi_node::commons;
 use xxi_node::commons::Direction;
 use xxi_node::node::rust_dlc_manager::DlcChannelId;
-
-#[derive(Debug, Copy, Clone, PartialEq)]
-pub struct ProtocolId(Uuid);
-
-impl ProtocolId {
-    pub fn new() -> Self {
-        ProtocolId(Uuid::new_v4())
-    }
-
-    pub fn to_uuid(&self) -> Uuid {
-        self.0
-    }
-}
-
-impl Default for ProtocolId {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl Display for ProtocolId {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        self.0.to_string().fmt(f)
-    }
-}
-
-impl From<ProtocolId> for ReferenceId {
-    fn from(value: ProtocolId) -> Self {
-        let uuid = value.to_uuid();
-
-        // 16 bytes.
-        let uuid_bytes = uuid.as_bytes();
-
-        // 32-digit hex string.
-        let hex = hex::encode(uuid_bytes);
-
-        // Derived `ReferenceId`: 32-bytes.
-        let hex_bytes = hex.as_bytes();
-
-        let mut array = [0u8; 32];
-        array.copy_from_slice(hex_bytes);
-
-        array
-    }
-}
-
-impl TryFrom<ReferenceId> for ProtocolId {
-    type Error = anyhow::Error;
-
-    fn try_from(value: ReferenceId) -> Result<Self> {
-        // 32-digit hex string.
-        let hex = from_utf8(&value)?;
-
-        // 16 bytes.
-        let uuid_bytes = hex::decode(hex)?;
-
-        let uuid = Uuid::from_slice(&uuid_bytes)?;
-
-        Ok(ProtocolId(uuid))
-    }
-}
-
-impl From<Uuid> for ProtocolId {
-    fn from(value: Uuid) -> Self {
-        ProtocolId(value)
-    }
-}
-
-impl From<ProtocolId> for Uuid {
-    fn from(value: ProtocolId) -> Self {
-        value.0
-    }
-}
+use xxi_node::node::ProtocolId;
 
 pub struct DlcProtocol {
     pub id: ProtocolId,
+    pub previous_id: Option<ProtocolId>,
     pub timestamp: OffsetDateTime,
     pub channel_id: DlcChannelId,
-    pub contract_id: ContractId,
+    pub contract_id: Option<ContractId>,
     pub trader: PublicKey,
     pub protocol_state: DlcProtocolState,
     pub protocol_type: DlcProtocolType,
@@ -247,7 +171,7 @@ impl DlcProtocolExecutor {
         &self,
         protocol_id: ProtocolId,
         previous_protocol_id: Option<ProtocolId>,
-        contract_id: &ContractId,
+        contract_id: Option<&ContractId>,
         channel_id: &DlcChannelId,
         protocol_type: DlcProtocolType,
     ) -> Result<()> {
@@ -325,15 +249,14 @@ impl DlcProtocolExecutor {
                     )
                 }
                 DlcProtocolType::Settle { trade_params } => {
-                    let settled_contract = &dlc_protocol.contract_id;
-
+                    let settled_contract = dlc_protocol.contract_id;
                     self.finish_close_trade_dlc_protocol(
                         conn,
                         trade_params,
                         protocol_id,
                         // If the contract got settled, we do not get a new contract id, hence we
                         // copy the contract id of the settled contract.
-                        settled_contract,
+                        settled_contract.as_ref(),
                         channel_id,
                     )
                 }
@@ -349,7 +272,10 @@ impl DlcProtocolExecutor {
                         channel_id,
                     )
                 }
-                DlcProtocolType::Close { .. } | DlcProtocolType::ForceClose { .. } => {
+                DlcProtocolType::Close { .. } => {
+                    self.finish_close_channel_dlc_protocol(conn, trader_id, protocol_id, channel_id)
+                }
+                DlcProtocolType::ForceClose { .. } => {
                     debug_assert!(false, "Finishing unexpected dlc protocol types");
                     Ok(())
                 }
@@ -395,7 +321,7 @@ impl DlcProtocolExecutor {
         conn: &mut PgConnection,
         trade_params: &TradeParams,
         protocol_id: ProtocolId,
-        settled_contract: &ContractId,
+        settled_contract: Option<&ContractId>,
         channel_id: &DlcChannelId,
     ) -> QueryResult<()> {
         db::dlc_protocols::set_dlc_protocol_state_to_success(
@@ -501,7 +427,7 @@ impl DlcProtocolExecutor {
         db::dlc_protocols::set_dlc_protocol_state_to_success(
             conn,
             protocol_id,
-            contract_id,
+            Some(contract_id),
             channel_id,
         )?;
 
@@ -547,7 +473,7 @@ impl DlcProtocolExecutor {
         db::dlc_protocols::set_dlc_protocol_state_to_success(
             conn,
             protocol_id,
-            contract_id,
+            Some(contract_id),
             channel_id,
         )?;
 
@@ -594,13 +520,25 @@ impl DlcProtocolExecutor {
         db::dlc_protocols::set_dlc_protocol_state_to_success(
             conn,
             protocol_id,
-            contract_id,
+            Some(contract_id),
             channel_id,
         )?;
 
         db::positions::Position::set_position_to_open(conn, trader.to_string(), *contract_id)?;
         Ok(())
     }
+
+    /// Completes the collab close dlc protocol as successful
+    fn finish_close_channel_dlc_protocol(
+        &self,
+        conn: &mut PgConnection,
+        trader: &PublicKey,
+        protocol_id: ProtocolId,
+        channel_id: &DlcChannelId,
+    ) -> QueryResult<()> {
+        tracing::debug!(%trader, %protocol_id, "Finalizing channel close");
+        db::dlc_protocols::set_dlc_protocol_state_to_success(conn, protocol_id, None, channel_id)
+    }
 }
 
 #[cfg(test)]