Merge pull request #41 from BitcoinDevShop/connection

Connection to another node

Author: AnthonyRonning

.github/workflows/preview.yml

@@ -35,7 +35,7 @@ jobs:
 
       - name: Build wasm
         working-directory: ./node-manager
-        run: wasm-pack build --target web
+        run: wasm-pack build --release --target web
 
       - name: Use Node.js 18.x
         uses: actions/setup-node@v3

.github/workflows/production.yml

@@ -35,7 +35,7 @@ jobs:
 
       - name: Build wasm
         working-directory: ./node-manager
-        run: wasm-pack build --target web
+        run: wasm-pack build --release --target web
 
       - name: Use Node.js 18.x
         uses: actions/setup-node@v3

frontend/src/routes/App.tsx

@@ -44,6 +44,7 @@ function App() {
       <header className='p-8'>
         <img src={logo} className="App-logo" alt="logo" />
         <h2>You're probably looking for <a href="/tests">the tests</a></h2>
+        <p>View the <a href="https://github.com/BitcoinDevShop/mutiny-web-poc">source</a></p>
         {wasmSupported ? <p>WASM works!</p> :
           <p>
             WASM does not seem supported in your browser, this might not work for you!

frontend/src/routes/KitchenSink.tsx

@@ -196,6 +196,7 @@ function App() {
                         {newPubkey &&
                             <form onSubmit={connect_peer} className="flex flex-col items-start gap-4 my-4">
                                 <h2>Connect Peer:</h2>
+                                <p>You may want to use "wss://websocket-tcp-proxy-fywbx.ondigitalocean.app" as the example websocket proxy</p>
                                 <input type="text" placeholder='Websocket Proxy Address' onChange={handleProxyAddressChange}></input>
                                 <input type="text" placeholder='Peer Connection String' onChange={handleConnectPeerChange}></input>
                                 <input type="submit" value="Connect" />

justfile

@@ -1,10 +1,10 @@
 dev:
     cd ./frontend && npm run start-ssl
 pack:
-    wasm-pack build ./node-manager --target web
+    wasm-pack build ./node-manager --dev --target web
 
 pack-mac:
-    AR=/opt/homebrew/opt/llvm/bin/llvm-ar CC=/opt/homebrew/opt/llvm/bin/clang wasm-pack build ./node-manager --target web
+    AR=/opt/homebrew/opt/llvm/bin/llvm-ar CC=/opt/homebrew/opt/llvm/bin/clang wasm-pack build ./node-manager --dev --target web
 
 test:
     cargo test --package websocket-tcp-proxy

node-manager/Cargo.toml

@@ -57,8 +57,7 @@ wasm-bindgen-test = "0.3.13"
 web-sys = { version = "0.3.60", features = ["console"] }
 
 [features]
-# default = ["console_error_panic_hook" ,"wee_alloc"]
-default = ["console_error_panic_hook"]
+default = ["console_error_panic_hook" ,"wee_alloc"]
 
 [package.metadata.wasm-pack.profile.release]
-wasm-opt = false
+wasm-opt = true

node-manager/src/error.rs

@@ -1,3 +1,4 @@
+use lightning::ln::peer_handler::PeerHandleError;
 use thiserror::Error;
 use wasm_bindgen::JsValue;
 
@@ -91,6 +92,13 @@ impl From<bdk::Error> for MutinyError {
     }
 }
 
+impl From<PeerHandleError> for MutinyError {
+    fn from(_e: PeerHandleError) -> Self {
+        // TODO handle the case where `no_connection_possible`
+        Self::ConnectionFailed
+    }
+}
+
 impl From<MutinyStorageError> for bdk::Error {
     fn from(e: MutinyStorageError) -> Self {
         match e {

node-manager/src/node.rs

@@ -3,10 +3,14 @@ use crate::localstorage::MutinyBrowserStorage;
 use crate::wallet::{esplora_from_network, MutinyWallet};
 use bdk::blockchain::EsploraBlockchain;
 use bitcoin::Network;
+use futures::StreamExt;
+use gloo_net::websocket::Message;
 use lightning::chain::{chainmonitor, Filter};
+use lightning::ln::msgs::NetAddress;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::str::FromStr;
 use std::sync::Arc;
+use wasm_bindgen_futures::spawn_local;
 
 use crate::tcpproxy::{SocketDescriptor, TcpProxy};
 use crate::{
@@ -21,10 +25,10 @@ use bitcoin::secp256k1::PublicKey;
 use lightning::chain::keysinterface::{InMemorySigner, KeysInterface, KeysManager, Recipient};
 use lightning::ln::peer_handler::{
     ErroringMessageHandler, IgnoringMessageHandler, MessageHandler as LdkMessageHandler,
-    PeerManager as LdkPeerManager,
+    PeerManager as LdkPeerManager, SocketDescriptor as LdkSocketDescriptor,
 };
 use lightning::routing::gossip;
-use log::info;
+use log::{debug, error, info, warn};
 
 pub(crate) type NetworkGraph = gossip::NetworkGraph<Arc<MutinyLogger>>;
 
@@ -138,9 +142,14 @@ impl Node {
             }
         };
 
-        if connect_peer_if_necessary(websocket_proxy_addr, pubkey, peer_addr)
-            .await
-            .is_err()
+        if connect_peer_if_necessary(
+            websocket_proxy_addr,
+            pubkey,
+            peer_addr,
+            self.peer_manager.clone(),
+        )
+        .await
+        .is_err()
         {
             Err(MutinyError::PeerInfoParseFailed)
                 .with_context(|| format!("could not connect to peer: {pubkey}"))?
@@ -152,10 +161,10 @@ impl Node {
 
 pub(crate) async fn connect_peer_if_necessary(
     websocket_proxy_addr: String,
-    _pubkey: PublicKey,
+    pubkey: PublicKey,
     peer_addr: SocketAddr,
-    // peer_manager: Arc<PeerManager>,
-) -> Result<(), ()> {
+    peer_manager: Arc<PeerManager>,
+) -> Result<(), MutinyError> {
     // TODO add this when the peer manager is ready
     /*
     for node_pubkey in peer_manager.get_peer_node_ids() {
@@ -166,18 +175,68 @@ pub(crate) async fn connect_peer_if_necessary(
     */
 
     // first make a connection to the node
-    let tcp_proxy = TcpProxy::new(websocket_proxy_addr, peer_addr).await;
-
-    // TODO remove the test send
-    tcp_proxy.send(String::from("test\n").into_bytes().to_vec());
-
-    // TODO then give that connection to the peer manager
+    let tcp_proxy = Arc::new(TcpProxy::new(websocket_proxy_addr, peer_addr).await);
+    let mut descriptor = SocketDescriptor::new(tcp_proxy);
+
+    // then give that connection to the peer manager
+    let initial_bytes = peer_manager.new_outbound_connection(
+        pubkey,
+        descriptor.clone(),
+        Some(get_net_addr_from_socket(peer_addr)),
+    )?;
+
+    let sent_bytes = descriptor.send_data(&initial_bytes, true);
+    debug!("sent {sent_bytes} to node: {pubkey}");
+
+    // schedule a reader on the connection
+    let mut new_descriptor = descriptor.clone();
+    spawn_local(async move {
+        while let Some(msg) = descriptor.conn.read.lock().await.next().await {
+            if let Ok(msg_contents) = msg {
+                match msg_contents {
+                    Message::Text(t) => {
+                        warn!(
+                            "received text from websocket when we should only receive binary: {}",
+                            t
+                        )
+                    }
+                    Message::Bytes(b) => {
+                        debug!("received binary data from websocket");
+
+                        let read_res = peer_manager.read_event(&mut new_descriptor, &b);
+                        match read_res {
+                            // TODO handle read boolean event
+                            Ok(_read_bool) => {
+                                debug!("read event from the node");
+                                peer_manager.process_events();
+                            }
+                            Err(e) => error!("got an error reading event: {}", e),
+                        }
+                    }
+                };
+            }
+        }
 
-    // TODO then schedule a reader on the connection
+        // TODO when we detect an error, lock the writes and close connection.
+        debug!("WebSocket Closed")
+    });
 
     Ok(())
 }
 
+fn get_net_addr_from_socket(socket_addr: SocketAddr) -> NetAddress {
+    match socket_addr {
+        SocketAddr::V4(sockaddr) => NetAddress::IPv4 {
+            addr: sockaddr.ip().octets(),
+            port: sockaddr.port(),
+        },
+        SocketAddr::V6(sockaddr) => NetAddress::IPv6 {
+            addr: sockaddr.ip().octets(),
+            port: sockaddr.port(),
+        },
+    }
+}
+
 pub(crate) fn create_peer_manager(
     km: Arc<KeysManager>,
     lightning_msg_handler: MessageHandler,

node-manager/src/tcpproxy.rs

@@ -2,48 +2,28 @@ use std::hash::Hash;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::{net::SocketAddr, sync::Arc};
 
+use futures::stream::SplitStream;
 use futures::{lock::Mutex, stream::SplitSink, SinkExt, StreamExt};
 use gloo_net::websocket::{futures::WebSocket, Message};
 use lightning::ln::peer_handler;
 use log::{debug, info};
 use wasm_bindgen_futures::spawn_local;
 
 pub struct TcpProxy {
-    connection: WsSplit,
+    pub write: WsSplit,
+    pub read: ReadSplit,
 }
 
 type WsSplit = Arc<Mutex<SplitSink<WebSocket, Message>>>;
+type ReadSplit = Arc<Mutex<SplitStream<WebSocket>>>;
 
 impl TcpProxy {
     pub async fn new(proxy_url: String, peer_addr: SocketAddr) -> Self {
         let ws = WebSocket::open(String::as_str(&proxy_to_url(proxy_url, peer_addr))).unwrap();
-        let (write, mut read) = ws.split();
-
-        spawn_local(async move {
-            // TODO callback or pass bytes over to some stream reader.
-            // need to figure out how LDK wants this incoming data.
-            while let Some(msg) = read.next().await {
-                if let Ok(msg_contents) = msg {
-                    match msg_contents {
-                        Message::Text(t) => {
-                            info!("receive text from websocket {}", t)
-                        }
-                        Message::Bytes(b) => {
-                            info!(
-                                "receive binary from websocket {}",
-                                String::from_utf8_lossy(&b)
-                            )
-                        }
-                    };
-                }
-            }
-
-            // TODO when we detect an error, lock the writes and close connection.
-            debug!("WebSocket Closed")
-        });
-
+        let (write, read) = ws.split();
         TcpProxy {
-            connection: Arc::new(Mutex::new(write)),
+            write: Arc::new(Mutex::new(write)),
+            read: Arc::new(Mutex::new(read)),
         }
     }
 
@@ -53,7 +33,7 @@ impl TcpProxy {
         // There can only be one sender at a time
         // Cannot send and write at the same time either
         // TODO check if the connection is closed before trying to send.
-        let cloned_conn = self.connection.clone();
+        let cloned_conn = self.write.clone();
         spawn_local(async move {
             let mut write = cloned_conn.lock().await;
             write.send(Message::Bytes(data)).await.unwrap();
@@ -73,11 +53,11 @@ fn proxy_to_url(proxy_url: String, peer_addr: SocketAddr) -> String {
 static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
 
 pub struct SocketDescriptor {
-    conn: Arc<TcpProxy>,
+    pub conn: Arc<TcpProxy>,
     id: u64,
 }
 impl SocketDescriptor {
-    fn new(conn: Arc<TcpProxy>) -> Self {
+    pub fn new(conn: Arc<TcpProxy>) -> Self {
         let id = ID_COUNTER.fetch_add(1, Ordering::AcqRel);
         Self { conn, id }
     }
@@ -90,7 +70,7 @@ impl peer_handler::SocketDescriptor for SocketDescriptor {
     }
 
     fn disconnect_socket(&mut self) {
-        let cloned = self.conn.connection.clone();
+        let cloned = self.conn.write.clone();
         spawn_local(async move {
             let mut conn = cloned.lock().await;
             let _ = conn.close();