Merge pull request #2608 from get10101/feat/notify-traders-on-new-fun…

…ding-rate

Notify traders when new funding rates are added

coordinator/src/db/mod.rs

@@ -6,15 +6,12 @@ pub mod custom_types;
 pub mod dlc_channels;
 pub mod dlc_messages;
 pub mod dlc_protocols;
-pub mod funding_fee_events;
-pub mod funding_rates;
 pub mod hodl_invoice;
 pub mod last_outbound_dlc_message;
 pub mod liquidity_options;
 pub mod metrics;
 pub mod polls;
 pub mod positions;
-pub mod protocol_funding_fee_events;
 pub mod reported_errors;
 pub mod rollover_params;
 pub mod spendable_outputs;

coordinator/src/dlc_protocol.rs

@@ -1,4 +1,6 @@
 use crate::db;
+use crate::funding_fee::insert_protocol_funding_fee_event;
+use crate::funding_fee::mark_funding_fee_event_as_paid;
 use crate::position::models::PositionState;
 use crate::trade::models::NewTrade;
 use crate::trade::websocket::InternalPositionUpdateMessage;
@@ -210,7 +212,7 @@ impl DlcProtocolExecutor {
                 &trader_pubkey,
             )?;
 
-            db::protocol_funding_fee_events::insert(conn, protocol_id, &funding_fee_event_ids)?;
+            insert_protocol_funding_fee_event(conn, protocol_id, &funding_fee_event_ids)?;
 
             db::trade_params::insert(
                 conn,
@@ -246,7 +248,7 @@ impl DlcProtocolExecutor {
                 &trader_pubkey,
             )?;
 
-            db::protocol_funding_fee_events::insert(conn, protocol_id, &funding_fee_event_ids)?;
+            insert_protocol_funding_fee_event(conn, protocol_id, &funding_fee_event_ids)?;
 
             db::trade_params::insert(conn, &TradeParams::new(trade_params, protocol_id, None))?;
 
@@ -280,7 +282,7 @@ impl DlcProtocolExecutor {
                 &trader_pubkey,
             )?;
 
-            db::protocol_funding_fee_events::insert(conn, protocol_id, &funding_fee_event_ids)?;
+            insert_protocol_funding_fee_event(conn, protocol_id, &funding_fee_event_ids)?;
 
             db::rollover_params::insert(conn, &rollover_params)?;
 
@@ -521,7 +523,7 @@ impl DlcProtocolExecutor {
 
         db::trades::insert(conn, new_trade)?;
 
-        db::funding_fee_events::mark_as_paid(conn, protocol_id)?;
+        mark_funding_fee_event_as_paid(conn, protocol_id)?;
 
         Ok(())
     }
@@ -621,7 +623,7 @@ impl DlcProtocolExecutor {
 
         db::trades::insert(conn, new_trade)?;
 
-        db::funding_fee_events::mark_as_paid(conn, protocol_id)?;
+        mark_funding_fee_event_as_paid(conn, protocol_id)?;
 
         Ok(())
     }
@@ -657,7 +659,7 @@ impl DlcProtocolExecutor {
             rollover_params.liquidation_price_trader,
         )?;
 
-        db::funding_fee_events::mark_as_paid(conn, protocol_id)?;
+        mark_funding_fee_event_as_paid(conn, protocol_id)?;
 
         Ok(())
     }

coordinator/src/funding_fee.rs

@@ -1,4 +1,3 @@
-use crate::db;
 use crate::decimal_from_f32;
 use crate::message::OrderbookMessage;
 use crate::FundingFee;
@@ -19,12 +18,22 @@ use std::time::Duration;
 use time::ext::NumericalDuration;
 use time::format_description;
 use time::OffsetDateTime;
+use tokio::sync::broadcast;
 use tokio::task::block_in_place;
 use tokio_cron_scheduler::JobScheduler;
 use xxi_node::commons::ContractSymbol;
 use xxi_node::commons::Direction;
+use xxi_node::commons::FundingRate;
 use xxi_node::commons::Message;
 
+mod db;
+
+pub use db::get_funding_fee_events_for_active_trader_positions;
+pub use db::get_next_funding_rate;
+pub use db::get_outstanding_funding_fee_events;
+pub use db::insert_protocol_funding_fee_event;
+pub use db::mark_funding_fee_event_as_paid;
+
 const RETRY_INTERVAL: Duration = Duration::from_secs(5);
 
 /// A record that a funding fee is owed between the coordinator and a trader.
@@ -155,7 +164,7 @@ fn generate_funding_fee_events(
     }
 
     // We exclude active positions which were open after this funding period ended.
-    let positions = db::positions::Position::get_all_active_positions_open_before(
+    let positions = crate::db::positions::Position::get_all_active_positions_open_before(
         &mut conn,
         funding_rate.end_date(),
     )?;
@@ -304,6 +313,26 @@ pub fn funding_fee_from_funding_fee_events(events: &[FundingFeeEvent]) -> Fundin
     }
 }
 
+pub fn insert_funding_rates(
+    conn: &mut PgConnection,
+    tx_orderbook_feed: broadcast::Sender<Message>,
+    funding_rates: &[FundingRate],
+) -> Result<()> {
+    db::insert_funding_rates(conn, funding_rates)?;
+
+    // There is no guarantee that the next funding rate has changed, but sending the message
+    // unconditionally is simpler and should cause no problems.
+    let next_funding_rate = get_next_funding_rate(conn)?;
+
+    if let Some(next_funding_rate) = next_funding_rate {
+        if let Err(e) = tx_orderbook_feed.send(Message::NextFundingRate(next_funding_rate)) {
+            tracing::error!("Failed to notify traders about next funding rate: {e}");
+        }
+    }
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

coordinator/src/funding_fee/db.rs

@@ -0,0 +1,7 @@
+pub mod funding_fee_events;
+pub mod funding_rates;
+pub mod protocol_funding_fee_events;
+
+pub use funding_fee_events::*;
+pub use funding_rates::*;
+pub use protocol_funding_fee_events::*;

coordinator/src/db/funding_fee_events.rs → coordinator/src/funding_fee/db/funding_fee_events.rs

@@ -30,7 +30,7 @@ struct FundingFeeEvent {
     _timestamp: OffsetDateTime,
 }
 
-pub(crate) fn insert(
+pub fn insert(
     conn: &mut PgConnection,
     amount: SignedAmount,
     trader_pubkey: PublicKey,
@@ -79,7 +79,7 @@ pub(crate) fn insert(
 ///
 /// A list of [`xxi_node::FundingFeeEvent`]s, since these are to be sent to the trader via the
 /// `xxi_node::Message::AllFundingFeeEvents` message.
-pub(crate) fn get_for_active_trader_positions(
+pub fn get_funding_fee_events_for_active_trader_positions(
     conn: &mut PgConnection,
     trader_pubkey: PublicKey,
 ) -> QueryResult<Vec<xxi_node::FundingFeeEvent>> {
@@ -110,7 +110,7 @@ pub(crate) fn get_for_active_trader_positions(
 }
 
 /// Get the unpaid [`funding_fee::FundingFeeEvent`]s for a trader position.
-pub(crate) fn get_outstanding_fees(
+pub fn get_outstanding_funding_fee_events(
     conn: &mut PgConnection,
     trader_pubkey: PublicKey,
     position_id: i32,
@@ -131,7 +131,10 @@ pub(crate) fn get_outstanding_fees(
         .collect())
 }
 
-pub(crate) fn mark_as_paid(conn: &mut PgConnection, protocol_id: ProtocolId) -> QueryResult<()> {
+pub fn mark_funding_fee_event_as_paid(
+    conn: &mut PgConnection,
+    protocol_id: ProtocolId,
+) -> QueryResult<()> {
     conn.transaction(|conn| {
         // Find all funding fee event IDs that were just paid.
         let funding_fee_event_ids: Vec<i32> = protocol_funding_fee_events::table

coordinator/src/db/funding_rates.rs → coordinator/src/funding_fee/db/funding_rates.rs

@@ -27,7 +27,7 @@ struct FundingRate {
     _timestamp: OffsetDateTime,
 }
 
-pub(crate) fn insert(
+pub fn insert_funding_rates(
     conn: &mut PgConnection,
     funding_rates: &[xxi_node::commons::FundingRate],
 ) -> Result<()> {
@@ -48,7 +48,7 @@ pub(crate) fn insert(
     Ok(())
 }
 
-pub(crate) fn get_next_funding_rate(
+pub fn get_next_funding_rate(
     conn: &mut PgConnection,
 ) -> QueryResult<Option<xxi_node::commons::FundingRate>> {
     let funding_rate: Option<FundingRate> = funding_rates::table
@@ -62,7 +62,7 @@ pub(crate) fn get_next_funding_rate(
 }
 
 /// Get the funding rate with an end date that is equal to the current date to the nearest hour.
-pub(crate) fn get_funding_rate_charged_in_the_last_hour(
+pub fn get_funding_rate_charged_in_the_last_hour(
     conn: &mut PgConnection,
 ) -> QueryResult<Option<xxi_node::commons::FundingRate>> {
     let now = OffsetDateTime::now_utc();

coordinator/src/db/protocol_funding_fee_events.rs → coordinator/src/funding_fee/db/protocol_funding_fee_events.rs

@@ -5,7 +5,7 @@ use crate::schema::protocol_funding_fee_events;
 use diesel::prelude::*;
 use xxi_node::node::ProtocolId;
 
-pub(crate) fn insert(
+pub fn insert_protocol_funding_fee_event(
     conn: &mut PgConnection,
     protocol_id: ProtocolId,
     funding_fee_event_ids: &[i32],

coordinator/src/node/liquidated_positions.rs

@@ -1,5 +1,6 @@
 use crate::db;
 use crate::funding_fee::funding_fee_from_funding_fee_events;
+use crate::funding_fee::get_outstanding_funding_fee_events;
 use crate::node::Node;
 use crate::orderbook;
 use crate::orderbook::db::orders;
@@ -51,7 +52,7 @@ async fn check_if_positions_need_to_get_liquidated(
         // Update position based on the outstanding funding fee events _before_ considering
         // liquidation.
         let funding_fee_events =
-            db::funding_fee_events::get_outstanding_fees(&mut conn, position.trader, position.id)?;
+            get_outstanding_funding_fee_events(&mut conn, position.trader, position.id)?;
 
         let funding_fee = funding_fee_from_funding_fee_events(&funding_fee_events);
 

coordinator/src/node/rollover.rs

@@ -5,6 +5,7 @@ use crate::decimal_from_f32;
 use crate::dlc_protocol;
 use crate::dlc_protocol::RolloverParams;
 use crate::funding_fee::funding_fee_from_funding_fee_events;
+use crate::funding_fee::get_outstanding_funding_fee_events;
 use crate::node::Node;
 use crate::notifications::Notification;
 use crate::notifications::NotificationKind;
@@ -217,7 +218,7 @@ impl Node {
             Decimal::try_from(maintenance_margin_rate).expect("to fit into decimal");
 
         let funding_fee_events =
-            db::funding_fee_events::get_outstanding_fees(conn, trader_pubkey, position.id)?;
+            get_outstanding_funding_fee_events(conn, trader_pubkey, position.id)?;
 
         let funding_fee = funding_fee_from_funding_fee_events(&funding_fee_events);
 

coordinator/src/orderbook/websocket.rs

@@ -1,7 +1,7 @@
 use crate::db;
-use crate::db::funding_fee_events;
-use crate::db::funding_rates;
 use crate::db::user;
+use crate::funding_fee::get_funding_fee_events_for_active_trader_positions;
+use crate::funding_fee::get_next_funding_rate;
 use crate::message::NewUserMessage;
 use crate::orderbook::db::orders;
 use crate::orderbook::trading::NewOrderMessage;
@@ -253,7 +253,7 @@ pub async fn websocket_connection(stream: WebSocket, state: Arc<AppState>) {
 
                             // Send over all the funding fee events that the trader may have missed
                             // whilst they were offline.
-                            match funding_fee_events::get_for_active_trader_positions(
+                            match get_funding_fee_events_for_active_trader_positions(
                                 &mut conn, trader_id,
                             ) {
                                 Ok(funding_fee_events) => {
@@ -277,7 +277,7 @@ pub async fn websocket_connection(stream: WebSocket, state: Arc<AppState>) {
                                 }
                             }
 
-                            match funding_rates::get_next_funding_rate(&mut conn) {
+                            match get_next_funding_rate(&mut conn) {
                                 Ok(Some(funding_rate)) => {
                                     if let Err(e) = local_sender
                                         .send(Message::NextFundingRate(funding_rate))

coordinator/src/routes/admin.rs

@@ -1,5 +1,6 @@
 use crate::collaborative_revert;
 use crate::db;
+use crate::funding_fee::insert_funding_rates;
 use crate::parse_dlc_channel_id;
 use crate::position::models::Position;
 use crate::referrals;
@@ -684,7 +685,7 @@ pub async fn post_funding_rates(
             .map(xxi_node::commons::FundingRate::from)
             .collect::<Vec<_>>();
 
-        db::funding_rates::insert(&mut conn, &funding_rates)
+        insert_funding_rates(&mut conn, state.tx_orderbook_feed.clone(), &funding_rates)
             .map_err(|e| AppError::BadRequest(format!("{e:#}")))?;
 
         Ok(())

coordinator/src/trade/mod.rs

@@ -3,6 +3,7 @@ use crate::db;
 use crate::decimal_from_f32;
 use crate::dlc_protocol;
 use crate::funding_fee::funding_fee_from_funding_fee_events;
+use crate::funding_fee::get_outstanding_funding_fee_events;
 use crate::message::OrderbookMessage;
 use crate::node::Node;
 use crate::orderbook::db::matches;
@@ -798,7 +799,7 @@ impl TradeExecutor {
 
         // Update position based on the outstanding funding fee events _before_ applying resize.
         let funding_fee_events =
-            db::funding_fee_events::get_outstanding_fees(conn, position.trader, position.id)?;
+            get_outstanding_funding_fee_events(conn, position.trader, position.id)?;
 
         let funding_fee = funding_fee_from_funding_fee_events(&funding_fee_events);
 
@@ -1063,7 +1064,7 @@ impl TradeExecutor {
         // Update position based on the outstanding funding fee events _before_ calculating
         // `position_settlement_amount_coordinator`.
         let funding_fee_events =
-            db::funding_fee_events::get_outstanding_fees(conn, position.trader, position.id)?;
+            get_outstanding_funding_fee_events(conn, position.trader, position.id)?;
 
         let funding_fee = funding_fee_from_funding_fee_events(&funding_fee_events);