Mariusz Klochowicz
3 years ago
5 changed files with 192 additions and 282 deletions
@ -0,0 +1,149 @@ |
|||
use crate::model::cfd::{Attestation, Cfd, CfdState, CfdStateChangeEvent, OrderId}; |
|||
use crate::wallet::Wallet; |
|||
use crate::{db, monitor, oracle}; |
|||
use anyhow::{bail, Result}; |
|||
use sqlx::pool::PoolConnection; |
|||
use sqlx::Sqlite; |
|||
use tokio::sync::watch; |
|||
|
|||
pub async fn insert_cfd( |
|||
cfd: Cfd, |
|||
conn: &mut PoolConnection<Sqlite>, |
|||
update_sender: &watch::Sender<Vec<Cfd>>, |
|||
) -> Result<()> { |
|||
db::insert_cfd(cfd, conn).await?; |
|||
update_sender.send(db::load_all_cfds(conn).await?)?; |
|||
Ok(()) |
|||
} |
|||
|
|||
pub async fn insert_new_cfd_state_by_order_id( |
|||
order_id: OrderId, |
|||
new_state: &CfdState, |
|||
conn: &mut PoolConnection<Sqlite>, |
|||
update_sender: &watch::Sender<Vec<Cfd>>, |
|||
) -> Result<()> { |
|||
db::insert_new_cfd_state_by_order_id(order_id, new_state, conn).await?; |
|||
update_sender.send(db::load_all_cfds(conn).await?)?; |
|||
Ok(()) |
|||
} |
|||
|
|||
pub async fn try_cet_publication( |
|||
cfd: &mut Cfd, |
|||
conn: &mut PoolConnection<Sqlite>, |
|||
wallet: &Wallet, |
|||
update_sender: &watch::Sender<Vec<Cfd>>, |
|||
) -> Result<()> { |
|||
match cfd.cet()? { |
|||
Ok(cet) => { |
|||
let txid = wallet.try_broadcast_transaction(cet).await?; |
|||
tracing::info!("CET published with txid {}", txid); |
|||
|
|||
if cfd.handle(CfdStateChangeEvent::CetSent)?.is_none() { |
|||
bail!("If we can get the CET we should be able to transition") |
|||
} |
|||
|
|||
insert_new_cfd_state_by_order_id(cfd.order.id, &cfd.state, conn, update_sender).await?; |
|||
} |
|||
Err(not_ready_yet) => { |
|||
tracing::debug!("{:#}", not_ready_yet); |
|||
return Ok(()); |
|||
} |
|||
}; |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
pub async fn handle_monitoring_event( |
|||
event: monitor::Event, |
|||
conn: &mut PoolConnection<Sqlite>, |
|||
wallet: &Wallet, |
|||
update_sender: &watch::Sender<Vec<Cfd>>, |
|||
) -> Result<()> { |
|||
let order_id = event.order_id(); |
|||
|
|||
let mut cfd = db::load_cfd_by_order_id(order_id, conn).await?; |
|||
|
|||
if cfd.handle(CfdStateChangeEvent::Monitor(event))?.is_none() { |
|||
// early exit if there was not state change
|
|||
// this is for cases where we are already in a final state
|
|||
return Ok(()); |
|||
} |
|||
|
|||
insert_new_cfd_state_by_order_id(order_id, &cfd.state, conn, update_sender).await?; |
|||
|
|||
if let CfdState::OpenCommitted { .. } = cfd.state { |
|||
try_cet_publication(&mut cfd, conn, wallet, update_sender).await?; |
|||
} else if let CfdState::MustRefund { .. } = cfd.state { |
|||
let signed_refund_tx = cfd.refund_tx()?; |
|||
let txid = wallet.try_broadcast_transaction(signed_refund_tx).await?; |
|||
|
|||
tracing::info!("Refund transaction published on chain: {}", txid); |
|||
} |
|||
Ok(()) |
|||
} |
|||
|
|||
pub async fn handle_commit( |
|||
order_id: OrderId, |
|||
conn: &mut PoolConnection<Sqlite>, |
|||
wallet: &Wallet, |
|||
update_sender: &watch::Sender<Vec<Cfd>>, |
|||
) -> Result<()> { |
|||
let mut cfd = db::load_cfd_by_order_id(order_id, conn).await?; |
|||
|
|||
let signed_commit_tx = cfd.commit_tx()?; |
|||
|
|||
let txid = wallet.try_broadcast_transaction(signed_commit_tx).await?; |
|||
|
|||
if cfd.handle(CfdStateChangeEvent::CommitTxSent)?.is_none() { |
|||
bail!("If we can get the commit tx we should be able to transition") |
|||
} |
|||
|
|||
insert_new_cfd_state_by_order_id(cfd.order.id, &cfd.state, conn, update_sender).await?; |
|||
tracing::info!("Commit transaction published on chain: {}", txid); |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
pub async fn handle_oracle_attestation( |
|||
attestation: oracle::Attestation, |
|||
conn: &mut PoolConnection<Sqlite>, |
|||
wallet: &Wallet, |
|||
update_sender: &watch::Sender<Vec<Cfd>>, |
|||
) -> Result<()> { |
|||
tracing::debug!( |
|||
"Learnt latest oracle attestation for event: {}", |
|||
attestation.id |
|||
); |
|||
|
|||
let mut cfds = db::load_cfds_by_oracle_event_id(attestation.id, conn).await?; |
|||
|
|||
for (cfd, dlc) in cfds |
|||
.iter_mut() |
|||
.filter_map(|cfd| cfd.dlc().map(|dlc| (cfd, dlc))) |
|||
{ |
|||
if cfd |
|||
.handle(CfdStateChangeEvent::OracleAttestation(Attestation::new( |
|||
attestation.id, |
|||
attestation.price, |
|||
attestation.scalars.clone(), |
|||
dlc, |
|||
cfd.role(), |
|||
)?))? |
|||
.is_none() |
|||
{ |
|||
// if we don't transition to a new state after oracle attestation we ignore the cfd
|
|||
// this is for cases where we cannot handle the attestation which should be in a
|
|||
// final state
|
|||
continue; |
|||
} |
|||
|
|||
insert_new_cfd_state_by_order_id(cfd.order.id, &cfd.state, conn, update_sender).await?; |
|||
|
|||
if let Err(e) = try_cet_publication(cfd, conn, wallet, update_sender).await { |
|||
tracing::error!("Error when trying to publish CET: {:#}", e); |
|||
continue; |
|||
} |
|||
} |
|||
|
|||
Ok(()) |
|||
} |
|||
Loading…
Reference in new issue