Reactor & update dependencies

Author: buffrr

Cargo.lock

@@ -18,21 +18,20 @@ path = "src/lib.rs"
 [dependencies]
 spaces_wallet = { path = "../wallet" }
 spaces_protocol = { path = "../protocol", version = "*", features = ["std"]}
-spacedb = { git = "https://github.com/spacesprotocol/spacedb", version = "0.0.5" }
+spacedb = { git = "https://github.com/spacesprotocol/spacedb", version = "0.0.6" }
 
 tokio = { version = "1.37.0", features = ["signal"] }
 ctrlc = "3.4.4"
 anyhow = "1.0.86"
 clap = { version = "4.5.6", features = ["derive", "env"] }
 log = "0.4.21"
 serde = { version = "1.0.200", features = ["derive"] }
-toml = "0.8.14"
 hex = "0.4.3"
 jsonrpsee = { version = "0.22.5", features = ["server", "http-client", "macros"] }
 directories = "5.0.1"
 env_logger = "0.11.3"
 serde_json = "1.0.116"
-bincode = {version = "2.0.0-rc.3", features = ["serde", "derive"]}
+bincode = {version = "2.0.1", features = ["serde", "derive"]}
 base64 = "0.22.1"
 futures = "0.3.30"
 reqwest = { version = "0.12.5", default-features = false, features = ["json", "blocking", "rustls-tls"] }

client/Cargo.toml

@@ -0,0 +1,139 @@
+use std::path::PathBuf;
+use std::sync::Arc;
+use anyhow::anyhow;
+use tokio::sync::{broadcast, mpsc};
+use tokio::task::{JoinHandle, JoinSet};
+use crate::config::Args;
+use crate::rpc::{AsyncChainState, RpcServerImpl, WalletLoadRequest, WalletManager};
+use crate::source::{BitcoinBlockSource, BitcoinRpc};
+use crate::spaces::Spaced;
+use crate::store::LiveSnapshot;
+use crate::wallets::RpcWallet;
+
+pub struct App {
+    shutdown: broadcast::Sender<()>,
+    services: JoinSet<anyhow::Result<()>>,
+}
+
+impl App {
+    pub fn new(shutdown: broadcast::Sender<()>) -> Self {
+        Self {
+            shutdown,
+            services: JoinSet::new(),
+        }
+    }
+
+    async fn setup_rpc_wallet(&mut self, spaced: &Spaced, rx: mpsc::Receiver<WalletLoadRequest>, cbf: bool) {
+        let wallet_service = RpcWallet::service(
+            spaced.network,
+            spaced.rpc.clone(),
+            spaced.chain.state.clone(),
+            rx,
+            self.shutdown.clone(),
+            spaced.num_workers,
+            cbf
+        );
+
+        self.services.spawn(async move {
+            wallet_service
+                .await
+                .map_err(|e| anyhow!("Wallet service error: {}", e))
+        });
+    }
+
+    async fn setup_rpc_services(&mut self, spaced: &Spaced) {
+        let (wallet_loader_tx, wallet_loader_rx) = mpsc::channel(1);
+
+        let wallet_manager = WalletManager {
+            data_dir: spaced.data_dir.join("wallets"),
+            network: spaced.network,
+            rpc: spaced.rpc.clone(),
+            wallet_loader: wallet_loader_tx,
+            wallets: Arc::new(Default::default()),
+        };
+
+        let (async_chain_state, async_chain_state_handle) = create_async_store(
+            spaced.rpc.clone(),
+            spaced.anchors_path.clone(),
+            spaced.chain.state.clone(),
+            spaced.block_index.as_ref().map(|index| index.state.clone()),
+            self.shutdown.subscribe(),
+        )
+            .await;
+
+        self.services.spawn(async {
+            async_chain_state_handle
+                .await
+                .map_err(|e| anyhow!("Chain state error: {}", e))
+        });
+        let rpc_server = RpcServerImpl::new(async_chain_state.clone(), wallet_manager);
+
+        let bind = spaced.bind.clone();
+        let shutdown = self.shutdown.clone();
+
+        self.services.spawn(async move {
+            rpc_server
+                .listen(bind, shutdown)
+                .await
+                .map_err(|e| anyhow!("RPC Server error: {}", e))
+        });
+
+        self.setup_rpc_wallet(spaced, wallet_loader_rx, spaced.cbf).await;
+    }
+
+    async fn setup_sync_service(&mut self, mut spaced: Spaced) {
+        let (spaced_sender, spaced_receiver) = tokio::sync::oneshot::channel();
+
+        let shutdown = self.shutdown.clone();
+        let rpc = spaced.rpc.clone();
+
+        std::thread::spawn(move || {
+            let source = BitcoinBlockSource::new(rpc);
+            _ = spaced_sender.send(spaced.protocol_sync(source, shutdown));
+        });
+
+        self.services.spawn(async move {
+            spaced_receiver
+                .await?
+                .map_err(|e| anyhow!("Protocol sync error: {}", e))
+        });
+    }
+
+    pub async fn run(&mut self, args: Vec<String>) -> anyhow::Result<()> {
+        let shutdown_receiver = self.shutdown.subscribe();
+        let spaced = Args::configure(args, shutdown_receiver).await?;
+        self.setup_rpc_services(&spaced).await;
+        self.setup_sync_service(spaced).await;
+
+        while let Some(res) = self.services.join_next().await {
+            res??
+        }
+
+        Ok(())
+    }
+}
+
+async fn create_async_store(
+    rpc: BitcoinRpc,
+    anchors: Option<PathBuf>,
+    chain_state: LiveSnapshot,
+    block_index: Option<LiveSnapshot>,
+    shutdown: broadcast::Receiver<()>,
+) -> (AsyncChainState, JoinHandle<()>) {
+    let (tx, rx) = mpsc::channel(32);
+    let async_store = AsyncChainState::new(tx);
+    let client = reqwest::Client::new();
+    let handle = tokio::spawn(async move {
+        AsyncChainState::handler(
+            &client,
+            rpc,
+            anchors,
+            chain_state,
+            block_index,
+            rx,
+            shutdown,
+        )
+            .await
+    });
+    (async_store, handle)
+}