Merge pull request #2585 from get10101/chore/only-settle-invoice-when…

…-proposal-succeeded

feat: Only settle invoice if proposal was successful

coordinator/migrations/2024-05-30-112618_add-invoice-state-to-hodl-invoice/down.sql

@@ -0,0 +1,5 @@
+ALTER TABLE "hodl_invoices"
+    DROP COLUMN "invoice_state",
+    DROP COLUMN "order_id";
+
+DROP TYPE "InvoiceState_Type";

coordinator/migrations/2024-05-30-112618_add-invoice-state-to-hodl-invoice/up.sql

@@ -0,0 +1,5 @@
+CREATE TYPE "InvoiceState_Type" AS ENUM ('Open', 'Accepted', 'Settled', 'Failed');
+
+ALTER TABLE "hodl_invoices"
+    ADD COLUMN "invoice_state" "InvoiceState_Type" NOT NULL DEFAULT 'Open',
+    ADD COLUMN "order_id" UUID;

coordinator/src/bin/coordinator.rs

@@ -177,6 +177,7 @@ async fn main() -> Result<()> {
         settings.to_node_settings(),
         tx_position_feed.clone(),
         auth_users_notifier.clone(),
+        lnd_bridge.clone(),
     );
 
     // TODO: Pass the tokio metrics into Prometheus

coordinator/src/db/custom_types.rs

@@ -3,13 +3,15 @@ use crate::db::dlc_channels::DlcChannelState;
 use crate::db::dlc_messages::MessageType;
 use crate::db::dlc_protocols::DlcProtocolState;
 use crate::db::dlc_protocols::DlcProtocolType;
+use crate::db::hodl_invoice::InvoiceState;
 use crate::db::polls::PollType;
 use crate::db::positions::ContractSymbol;
 use crate::db::positions::PositionState;
 use crate::schema::sql_types::BonusStatusType;
 use crate::schema::sql_types::ContractSymbolType;
 use crate::schema::sql_types::DirectionType;
 use crate::schema::sql_types::DlcChannelStateType;
+use crate::schema::sql_types::InvoiceStateType;
 use crate::schema::sql_types::MessageTypeType;
 use crate::schema::sql_types::PollTypeType;
 use crate::schema::sql_types::PositionStateType;
@@ -265,3 +267,27 @@ impl FromSql<BonusStatusType, Pg> for BonusType {
         }
     }
 }
+
+impl ToSql<InvoiceStateType, Pg> for InvoiceState {
+    fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
+        match *self {
+            InvoiceState::Open => out.write_all(b"Open")?,
+            InvoiceState::Accepted => out.write_all(b"Accepted")?,
+            InvoiceState::Settled => out.write_all(b"Settled")?,
+            InvoiceState::Failed => out.write_all(b"Failed")?,
+        }
+        Ok(IsNull::No)
+    }
+}
+
+impl FromSql<InvoiceStateType, Pg> for InvoiceState {
+    fn from_sql(bytes: PgValue<'_>) -> deserialize::Result<Self> {
+        match bytes.as_bytes() {
+            b"Open" => Ok(InvoiceState::Open),
+            b"Accepted" => Ok(InvoiceState::Accepted),
+            b"Settled" => Ok(InvoiceState::Settled),
+            b"Failed" => Ok(InvoiceState::Failed),
+            _ => Err("Unrecognized enum variant".into()),
+        }
+    }
+}

coordinator/src/db/hodl_invoice.rs

@@ -1,12 +1,37 @@
 use crate::schema::hodl_invoices;
+use crate::schema::sql_types::InvoiceStateType;
 use anyhow::ensure;
 use anyhow::Result;
 use bitcoin::secp256k1::PublicKey;
 use bitcoin::Amount;
+use diesel::query_builder::QueryId;
+use diesel::AsExpression;
 use diesel::ExpressionMethods;
+use diesel::FromSqlRow;
 use diesel::PgConnection;
+use diesel::QueryResult;
 use diesel::RunQueryDsl;
+use std::any::TypeId;
 use time::OffsetDateTime;
+use uuid::Uuid;
+
+#[derive(Debug, Clone, Copy, PartialEq, FromSqlRow, AsExpression)]
+#[diesel(sql_type = InvoiceStateType)]
+pub enum InvoiceState {
+    Open,
+    Accepted,
+    Settled,
+    Failed,
+}
+
+impl QueryId for InvoiceStateType {
+    type QueryId = InvoiceStateType;
+    const HAS_STATIC_QUERY_ID: bool = false;
+
+    fn query_id() -> Option<TypeId> {
+        None
+    }
+}
 
 pub fn create_hodl_invoice(
     conn: &mut PgConnection,
@@ -18,6 +43,7 @@ pub fn create_hodl_invoice(
         .values((
             hodl_invoices::r_hash.eq(r_hash),
             hodl_invoices::trader_pubkey.eq(trader_pubkey.to_string()),
+            hodl_invoices::invoice_state.eq(InvoiceState::Open),
             hodl_invoices::amount_sats.eq(amount_sats as i64),
         ))
         .execute(conn)?;
@@ -27,19 +53,62 @@ pub fn create_hodl_invoice(
     Ok(())
 }
 
-pub fn update_hodl_invoice_pre_image(
+pub fn update_hodl_invoice_to_accepted(
     conn: &mut PgConnection,
     hash: &str,
     pre_image: &str,
+    order_id: Uuid,
 ) -> Result<Amount> {
     let amount: i64 = diesel::update(hodl_invoices::table)
         .filter(hodl_invoices::r_hash.eq(hash))
         .set((
             hodl_invoices::pre_image.eq(pre_image),
             hodl_invoices::updated_at.eq(OffsetDateTime::now_utc()),
+            hodl_invoices::invoice_state.eq(InvoiceState::Accepted),
+            hodl_invoices::order_id.eq(order_id),
         ))
         .returning(hodl_invoices::amount_sats)
         .get_result(conn)?;
 
     Ok(Amount::from_sat(amount as u64))
 }
+
+pub fn update_hodl_invoice_to_settled(
+    conn: &mut PgConnection,
+    order_id: Uuid,
+) -> QueryResult<Option<String>> {
+    diesel::update(hodl_invoices::table)
+        .filter(hodl_invoices::order_id.eq(order_id))
+        .set((
+            hodl_invoices::updated_at.eq(OffsetDateTime::now_utc()),
+            hodl_invoices::invoice_state.eq(InvoiceState::Settled),
+        ))
+        .returning(hodl_invoices::pre_image)
+        .get_result(conn)
+}
+
+pub fn update_hodl_invoice_to_failed(
+    conn: &mut PgConnection,
+    order_id: Uuid,
+) -> QueryResult<usize> {
+    diesel::update(hodl_invoices::table)
+        .filter(hodl_invoices::order_id.eq(order_id))
+        .set((
+            hodl_invoices::updated_at.eq(OffsetDateTime::now_utc()),
+            hodl_invoices::invoice_state.eq(InvoiceState::Failed),
+        ))
+        .execute(conn)
+}
+
+pub fn update_hodl_invoice_to_failed_by_r_hash(
+    conn: &mut PgConnection,
+    r_hash: String,
+) -> QueryResult<usize> {
+    diesel::update(hodl_invoices::table)
+        .filter(hodl_invoices::r_hash.eq(r_hash))
+        .set((
+            hodl_invoices::updated_at.eq(OffsetDateTime::now_utc()),
+            hodl_invoices::invoice_state.eq(InvoiceState::Failed),
+        ))
+        .execute(conn)
+}

coordinator/src/db/last_outbound_dlc_message.rs

@@ -28,6 +28,12 @@ pub(crate) struct LastOutboundDlcMessage {
     pub timestamp: OffsetDateTime,
 }
 
+pub(crate) fn delete(conn: &mut PgConnection, peer_id: &PublicKey) -> QueryResult<usize> {
+    diesel::delete(last_outbound_dlc_messages::table)
+        .filter(last_outbound_dlc_messages::peer_id.eq(peer_id.to_string()))
+        .execute(conn)
+}
+
 pub(crate) fn get(
     conn: &mut PgConnection,
     peer_id: &PublicKey,

coordinator/src/node.rs

@@ -21,6 +21,7 @@ use dlc_messages::channel::Reject;
 use dlc_messages::channel::RenewFinalize;
 use dlc_messages::channel::SettleFinalize;
 use dlc_messages::channel::SignChannel;
+use lnd_bridge::LndBridge;
 use std::sync::Arc;
 use tokio::sync::broadcast::Sender;
 use tokio::sync::mpsc;
@@ -78,6 +79,7 @@ pub struct Node {
     pub settings: Arc<RwLock<NodeSettings>>,
     pub tx_position_feed: Sender<InternalPositionUpdateMessage>,
     trade_notifier: mpsc::Sender<OrderbookMessage>,
+    pub lnd_bridge: LndBridge,
 }
 
 impl Node {
@@ -94,6 +96,7 @@ impl Node {
         settings: NodeSettings,
         tx_position_feed: Sender<InternalPositionUpdateMessage>,
         trade_notifier: mpsc::Sender<OrderbookMessage>,
+        lnd_bridge: LndBridge,
     ) -> Self {
         Self {
             inner,
@@ -102,6 +105,7 @@ impl Node {
             _running: Arc::new(running),
             tx_position_feed,
             trade_notifier,
+            lnd_bridge,
         }
     }
 
@@ -208,7 +212,7 @@ impl Node {
     /// inconsistent state. One way of fixing that could be to: (1) use a single data source for the
     /// 10101 data and the `rust-dlc` data; (2) wrap the function into a DB transaction which can be
     /// atomically rolled back on error or committed on success.
-    fn process_dlc_message(&self, node_id: PublicKey, msg: &TenTenOneMessage) -> Result<()> {
+    pub fn process_dlc_message(&self, node_id: PublicKey, msg: &TenTenOneMessage) -> Result<()> {
         tracing::info!(
             from = %node_id,
             kind = %tentenone_message_name(msg),

coordinator/src/node/invoice.rs

@@ -1,13 +1,19 @@
+use crate::db;
 use bitcoin::Amount;
+use diesel::r2d2::ConnectionManager;
+use diesel::r2d2::Pool;
+use diesel::PgConnection;
 use futures_util::TryStreamExt;
 use lnd_bridge::InvoiceState;
 use lnd_bridge::LndBridge;
 use tokio::sync::broadcast;
+use tokio::task::spawn_blocking;
 use xxi_node::commons;
 use xxi_node::commons::Message;
 
 /// Watches a hodl invoice with the given r_hash
 pub fn spawn_invoice_watch(
+    pool: Pool<ConnectionManager<PgConnection>>,
     trader_sender: broadcast::Sender<Message>,
     lnd_bridge: LndBridge,
     invoice_params: commons::HodlInvoiceParams,
@@ -31,6 +37,22 @@ pub fn spawn_invoice_watch(
                     }
                     InvoiceState::Canceled => {
                         tracing::warn!(%trader_pubkey, invoice.r_hash, "Pending hodl invoice has been canceled.");
+                        if let Err(e) = spawn_blocking(move || {
+                            let mut conn = pool.get()?;
+                            db::hodl_invoice::update_hodl_invoice_to_failed_by_r_hash(
+                                &mut conn,
+                                invoice.r_hash,
+                            )?;
+                            anyhow::Ok(())
+                        })
+                        .await
+                        .expect("task to finish")
+                        {
+                            tracing::error!(
+                                r_hash,
+                                "Failed to set hodl invoice to failed. Error: {e:#}"
+                            );
+                        }
                         break;
                     }
                     InvoiceState::Accepted => {

coordinator/src/routes.rs

@@ -676,6 +676,7 @@ async fn create_invoice(
 
     // watch for the created hodl invoice
     invoice::spawn_invoice_watch(
+        state.pool.clone(),
         state.tx_orderbook_feed.clone(),
         state.lnd_bridge.clone(),
         invoice_params,

coordinator/src/routes/orderbook.rs

@@ -77,7 +77,7 @@ pub async fn post_order(
         .map_err(|_| AppError::Unauthorized)?;
 
     let new_order = new_order_request.value;
-    let trader_pubkey_string = new_order.trader_id().to_string();
+    let order_id = new_order.id();
 
     // TODO(holzeis): We should add a similar check eventually for limit orders (makers).
     if let NewOrder::Market(new_order) = &new_order {
@@ -114,13 +114,13 @@ pub async fn post_order(
         .clone()
         .and_then(|c| c.pre_image)
     {
-        Some(pre_image) => {
-            let pre_image = commons::PreImage::from_url_safe_encoded_pre_image(pre_image.as_str())
-                .map_err(|_| AppError::BadRequest("Invalid pre_image provided".to_string()))?;
-            let inner_pre_image = pre_image.get_pre_image_as_string();
+        Some(pre_image_str) => {
+            let pre_image =
+                commons::PreImage::from_url_safe_encoded_pre_image(pre_image_str.as_str())
+                    .map_err(|_| AppError::BadRequest("Invalid pre_image provided".to_string()))?;
 
             tracing::debug!(
-                pre_image = inner_pre_image,
+                pre_image_str,
                 hash = pre_image.hash,
                 "Received pre-image, updating records"
             );
@@ -129,29 +129,19 @@ pub async fn post_order(
             let funding_amount = spawn_blocking(move || {
                 let mut conn = pool.get()?;
 
-                let amount = db::hodl_invoice::update_hodl_invoice_pre_image(
+                let amount = db::hodl_invoice::update_hodl_invoice_to_accepted(
                     &mut conn,
                     inner_hash.as_str(),
-                    inner_pre_image.as_str(),
+                    pre_image_str.as_str(),
+                    order_id,
                 )?;
 
                 anyhow::Ok(amount)
             })
             .await
             .expect("task to complete")
-            .map_err(|e| AppError::BadRequest(format!("Invalid preimage provided: {e:#}")))?;
+            .map_err(|e| AppError::BadRequest(format!("Invalid pre_image provided: {e:#}")))?;
 
-            state
-                .lnd_bridge
-                .settle_invoice(pre_image.get_base64_encoded_pre_image())
-                .await
-                .map_err(|err| AppError::BadRequest(format!("Could not settle invoice {err:#}")))?;
-
-            tracing::info!(
-                hash = pre_image.hash,
-                trader_pubkey = trader_pubkey_string,
-                "Settled invoice"
-            );
             // we have received funding via lightning and can now open the channel with funding
             // only from the coordinator
             Some(funding_amount)

coordinator/src/schema.rs

@@ -25,6 +25,10 @@ pub mod sql_types {
     #[diesel(postgres_type(name = "Htlc_Status_Type"))]
     pub struct HtlcStatusType;
 
+    #[derive(diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "InvoiceState_Type"))]
+    pub struct InvoiceStateType;
+
     #[derive(diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "MatchState_Type"))]
     pub struct MatchStateType;
@@ -213,6 +217,9 @@ diesel::table! {
 }
 
 diesel::table! {
+    use diesel::sql_types::*;
+    use super::sql_types::InvoiceStateType;
+
     hodl_invoices (id) {
         id -> Int4,
         trader_pubkey -> Text,
@@ -221,6 +228,8 @@ diesel::table! {
         pre_image -> Nullable<Text>,
         created_at -> Timestamptz,
         updated_at -> Nullable<Timestamptz>,
+        invoice_state -> InvoiceStateType,
+        order_id -> Nullable<Uuid>,
     }
 }