Load swaps from the database

Load each accepted swap from the database and:
- Populate state store
- Spawn Alice/Bob

Author: Tobin C. Harding

cnd/src/db/mod.rs

@@ -14,6 +14,7 @@ pub mod with_swap_types;
 embed_migrations!("./migrations");
 
 pub use self::{
+    load_swaps::*,
     save_message::{SaveMessage, SaveRfc003Messages},
     swap::*,
     swap_types::*,

cnd/src/db/swap_types.rs

@@ -110,7 +110,7 @@ struct QueryableSwap {
     swap_id: Text<SwapId>,
 }
 
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, Copy, Debug, PartialEq)]
 pub struct SwapTypes {
     pub alpha_ledger: LedgerKind,
     pub beta_ledger: LedgerKind,
@@ -119,7 +119,7 @@ pub struct SwapTypes {
     pub role: Role,
 }
 
-#[derive(Debug, Clone, Display, EnumString, PartialEq)]
+#[derive(Debug, Clone, Copy, Display, EnumString, PartialEq)]
 pub enum LedgerKind {
     Bitcoin,
     Ethereum,
@@ -137,7 +137,7 @@ impl From<ledger::LedgerKind> for LedgerKind {
     }
 }
 
-#[derive(Clone, Debug, Display, EnumString, PartialEq)]
+#[derive(Clone, Copy, Debug, Display, EnumString, PartialEq)]
 pub enum AssetKind {
     Bitcoin,
     Ether,

cnd/src/lib.rs

@@ -22,6 +22,7 @@ pub mod bitcoin;
 pub mod comit_api;
 pub mod config;
 pub mod http_api;
+pub mod load_swaps;
 pub mod logging;
 pub mod network;
 #[cfg(test)]

cnd/src/load_swaps.rs

@@ -0,0 +1,75 @@
+#![allow(clippy::type_repetition_in_bounds)]
+use crate::{
+    db::{AcceptedSwap, DetermineTypes, LoadAcceptedSwap, Retrieve, Sqlite},
+    seed::Seed,
+    swap_protocols::{
+        ledger::LedgerConnectors,
+        rfc003::{
+            state_store::{InMemoryStateStore, StateStore},
+            Spawn,
+        },
+    },
+};
+use futures::Stream;
+use std::sync::Arc;
+
+#[allow(clippy::cognitive_complexity)]
+pub async fn load_swaps_from_database(
+    ledger_events: LedgerConnectors,
+    state_store: Arc<InMemoryStateStore>,
+    seed: Seed,
+    db: Sqlite,
+) -> anyhow::Result<()> {
+    log::debug!("loading swaps from database ...");
+
+    for swap in db.all().await?.iter() {
+        let swap_id = swap.swap_id;
+        log::debug!("got swap from database: {}", swap_id);
+
+        let types = db.determine_types(&swap_id).await?;
+
+        with_swap_types!(types, {
+            let accepted: Result<AcceptedSwap<AL, BL, AA, BA>, anyhow::Error> =
+                db.load_accepted_swap(swap_id.clone()).await;
+
+            match accepted {
+                Ok((request, accept)) => {
+                    match types.role {
+                        Role::Alice => {
+                            let state = alice::State::accepted(request.clone(), accept, seed);
+                            state_store.insert(swap_id, state);
+
+                            let receiver = ledger_events.spawn(request, accept);
+
+                            tokio::spawn(receiver.for_each({
+                                let state_store = state_store.clone();
+                                move |update| {
+                                    state_store
+                                        .update::<alice::State<AL, BL, AA, BA>>(&swap_id, update);
+                                    Ok(())
+                                }
+                            }));
+                        }
+                        Role::Bob => {
+                            let state = bob::State::accepted(request.clone(), accept, seed);
+                            state_store.insert(swap_id, state);
+
+                            let receiver = ledger_events.spawn(request, accept);
+
+                            tokio::spawn(receiver.for_each({
+                                let state_store = state_store.clone();
+                                move |update| {
+                                    state_store
+                                        .update::<bob::State<AL, BL, AA, BA>>(&swap_id, update);
+                                    Ok(())
+                                }
+                            }));
+                        }
+                    };
+                }
+                Err(e) => log::error!("failed to load swap: {}, continuing ...", e),
+            };
+        });
+    }
+    Ok(())
+}

cnd/src/main.rs

@@ -7,6 +7,7 @@ use cnd::{
     config::{self, Settings},
     db::{DetermineTypes, Retrieve, Save, SaveRfc003Messages, Sqlite},
     http_api::{self, route_factory},
+    load_swaps,
     network::{self, Network, SendRequest},
     seed::{Seed, SwapSeed},
     swap_protocols::{
@@ -18,6 +19,7 @@ use cnd::{
     },
 };
 use futures::{stream, Future, Stream};
+use futures_core::{FutureExt, TryFutureExt};
 use libp2p::{
     identity::{self, ed25519},
     PeerId, Swarm,
@@ -53,8 +55,6 @@ fn main() -> anyhow::Result<()> {
 
     let mut runtime = tokio::runtime::Runtime::new()?;
 
-    let state_store = Arc::new(InMemoryStateStore::default());
-
     let bitcoin_connector = {
         let config::Bitcoin { node_url, network } = settings.clone().bitcoin;
         BitcoindConnector::new(node_url, network)?
@@ -68,11 +68,24 @@ fn main() -> anyhow::Result<()> {
         ethereum_connector,
     };
 
+    let state_store = Arc::new(InMemoryStateStore::default());
+
+    let database = Sqlite::new(&settings.database.sqlite)?;
+
     let local_key_pair = derive_key_pair(&seed);
     let local_peer_id = PeerId::from(local_key_pair.clone().public());
     log::info!("Starting with peer_id: {}", local_peer_id);
 
-    let database = Sqlite::new(&settings.database.sqlite)?;
+    runtime.block_on(
+        load_swaps::load_swaps_from_database(
+            ledger_events.clone(),
+            state_store.clone(),
+            seed,
+            database.clone(),
+        )
+        .boxed()
+        .compat(),
+    )?;
 
     let transport = libp2p::build_development_transport(local_key_pair);
     let behaviour = network::ComitNode::new(