bright_ln_client/

websocket.rs

use std::sync::Arc;

use futures_util::{
    SinkExt, StreamExt,
    stream::{SplitSink, SplitStream},
};
use httparse::Header;
use serde::{Serialize, de::DeserializeOwned};
use tokio::{net::TcpStream, sync::RwLock};
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, tungstenite::Message};

use bright_ln_models::{LndError, LndResponse};

type LndWebsocketWriterHalf =
    Option<SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>;
type LndWebsocketReaderHalf = Option<SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>>;

#[derive(Debug, thiserror::Error)]
pub enum LndWebsocketError {
    #[error("NoWriter")]
    NoWriter,
    #[error("UnparseableMessage")]
    UnparseableMessage,
    #[error("WebSocketError: {0}")]
    Tungstenite(Box<tokio_tungstenite::tungstenite::Error>),
    #[error("TlsError: {0}")]
    NativeTls(#[from] native_tls::Error),
}

type Result<T> = std::result::Result<T, LndWebsocketError>;

#[derive(Clone, Default, Debug)]
pub struct LndWebsocketWriter(Arc<RwLock<LndWebsocketWriterHalf>>);
impl LndWebsocketWriter {
    pub fn new(writer: LndWebsocketWriterHalf) -> Self {
        Self(Arc::new(RwLock::new(writer)))
    }
    pub async fn send<S>(&self, message: S) -> Result<()>
    where
        S: Send + Sync + 'static,
        String: TryFrom<S, Error = serde_json::Error>,
    {
        let message_string =
            String::try_from(message).map_err(|_e| LndWebsocketError::UnparseableMessage)?;
        let message = Message::Text(message_string.into());
        let mut writer = self.0.write().await;
        let writer = writer.as_mut().ok_or(LndWebsocketError::NoWriter)?;
        writer
            .send(message)
            .await
            .map_err(|e| LndWebsocketError::Tungstenite(Box::new(e)))
    }
}
#[derive(Clone, Default, Debug)]
pub struct LndWebsocketReader(Arc<RwLock<LndWebsocketReaderHalf>>);
impl LndWebsocketReader {
    #[must_use]
    pub fn new(reader: LndWebsocketReaderHalf) -> Self {
        Self(Arc::new(RwLock::new(reader)))
    }
    pub async fn read<R>(&self) -> Option<LndWebsocketMessage<R>>
    where
        R: std::str::FromStr + Send + Sync + 'static + Serialize + DeserializeOwned + Clone,
    {
        let value = self.0.write().await.as_mut()?.next().await?;
        match value {
            Ok(message) => match message {
                Message::Text(text) => match text.to_string().parse::<LndResponse<R>>() {
                    Ok(response) => Some(LndWebsocketMessage::Response(response.inner())),
                    Err(_e) => {
                        let lnd_error = text.to_string().parse::<LndError>().ok()?;
                        Some(LndWebsocketMessage::Error(lnd_error))
                    }
                },
                Message::Ping(_) => Some(LndWebsocketMessage::Ping),
                _ => None,
            },
            Err(e) => Some(LndWebsocketMessage::Error(
                e.to_string().parse::<LndError>().unwrap_or_default(),
            )),
        }
    }
}

#[derive(Debug)]
pub enum LndWebsocketMessage<R> {
    Response(R),
    Error(LndError),
    Ping,
}

#[derive(Debug, Default)]
pub struct LndWebsocket {
    pub receiver: LndWebsocketReader,
    pub sender: LndWebsocketWriter,
}

impl LndWebsocket {
    pub async fn connect(&self, url: String, macaroon: String, request: String) -> Result<Self> {
        let random_key = b"dGhlIHNhbXBsZSBub25jZQ2342qdfsdgfsdfg";
        let mut headers = [
            Header {
                name: "Grpc-Metadata-macaroon",
                value: macaroon.as_bytes(),
            },
            Header {
                name: "Sec-WebSocket-Key",
                value: random_key,
            },
            Header {
                name: "Host",
                value: url.as_bytes(),
            },
            Header {
                name: "Connection",
                value: b"Upgrade",
            },
            Header {
                name: "Upgrade",
                value: b"websocket",
            },
            httparse::Header {
                name: "Sec-WebSocket-Version",
                value: b"13",
            },
        ];
        let mut req = httparse::Request::new(&mut headers);
        req.method = Some("GET");
        req.path = Some(&request);
        req.version = Some(1);

        // Prepare the websocket connection with SSL
        let danger_conf = Some(
            tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
                .accept_unmasked_frames(true),
        );

        let tls = native_tls::TlsConnector::builder()
            .danger_accept_invalid_certs(true)
            .build()?;
        let (ws, _response) = tokio_tungstenite::connect_async_tls_with_config(
            req,
            danger_conf,
            false,
            Some(tokio_tungstenite::Connector::NativeTls(tls)),
        )
        .await
        .map_err(|e| LndWebsocketError::Tungstenite(Box::new(e)))?;
        let (websocket_sender, websocket_reader) = ws.split();
        let sender = LndWebsocketWriter::new(Some(websocket_sender));
        let receiver = LndWebsocketReader::new(Some(websocket_reader));
        Ok(Self { receiver, sender })
    }
}
#[cfg(test)]
mod test {

    use super::{LndWebsocketMessage, Result};
    use bright_ln_models::{LndHodlInvoiceState, LndInvoiceRequestBody};
    use std::io::Read;
    use tracing_test::traced_test;

    #[tokio::test]
    #[traced_test]
    async fn check_invoice_paid() -> Result<()> {
        let url = "lnd.illuminodes.com";
        let client = crate::LndRestClient::new(url, "./admin.macaroon").expect("No client");
        let invoice = client
            .get_invoice(LndInvoiceRequestBody {
                value: 1000.to_string(),
                memo: Some("Hello".to_string()),
            })
            .await
            .expect("No invoice");
        tracing::info!("Invoice: {:#?}", invoice);
        let query = format!(
            "wss://{}/v2/invoices/subscribe/{}",
            url,
            invoice.r_hash_url_safe().expect("No hash")
        );
        let mut macaroon = vec![];
        let mut file = std::fs::File::open("../admin.macaroon").expect("No file");
        file.read_to_end(&mut macaroon).expect("No macaroon");
        let lnd_ws = super::LndWebsocket::default()
            .connect(
                url.to_string(),
                macaroon.iter().fold(String::new(), |mut acc, x| {
                    acc.push_str(&format!("{x:02x}"));
                    acc
                }),
                query,
            )
            .await?;
        loop {
            match lnd_ws.receiver.read::<LndHodlInvoiceState>().await {
                Some(LndWebsocketMessage::Response(state)) => {
                    tracing::info!("State: {:#?}", state);
                    break;
                }
                Some(LndWebsocketMessage::Error(e)) => {
                    tracing::error!("Error: {:#?}", e);
                    panic!("Error: {:#?}", e);
                }
                Some(LndWebsocketMessage::Ping) => {
                    tracing::info!("Ping");
                }
                None => {
                    tracing::info!("None");
                    panic!("None");
                }
            }
        }
        Ok(())
    }
}