//! Sbot type and associated methods. use async_std::{ net::TcpStream, stream::{Stream, StreamExt}, }; use futures::pin_mut; use std::collections::HashMap; use kuska_handshake::async_std::BoxStream; use kuska_sodiumoxide::crypto::{auth, sign::ed25519}; use kuska_ssb::{ api::{dto::CreateHistoryStreamIn, ApiCaller}, discovery, keystore, keystore::OwnedIdentity, rpc::{RpcReader, RpcWriter}, }; use crate::error::GolgiError; use crate::messages::{SsbMessageContent, SsbMessageContentType, SsbMessageKVT, SsbMessageValue}; use crate::utils; use crate::utils::get_source_stream; // re-export types from kuska pub use kuska_ssb::api::dto::content::{SubsetQuery, SubsetQueryOptions}; /// A struct representing a connection with a running sbot. /// A client and an rpc_reader can together be used to make requests to the sbot /// and read the responses. /// Note there can be multiple SbotConnection at the same time. pub struct SbotConnection { client: ApiCaller, rpc_reader: RpcReader, } /// The Scuttlebutt identity, keys and configuration parameters for connecting to a local sbot pub struct Sbot { pub id: String, public_key: ed25519::PublicKey, private_key: ed25519::SecretKey, address: String, // aka caps key (scuttleverse identifier) network_id: auth::Key, } impl Sbot { /// Initiate a connection with an sbot instance. Define the IP address, port and network key /// for the sbot, then retrieve the public key, private key (secret) and identity from the /// `.ssb-go/secret` file. Open a TCP stream to the sbot and perform the secret handshake. If successful, create a box stream and split it into a writer and reader. Return RPC handles to the sbot as part of the `struct` output. pub async fn init(ip_port: Option, net_id: Option) -> Result { let address = if ip_port.is_none() { "127.0.0.1:8008".to_string() } else { ip_port.unwrap() }; let network_id = if net_id.is_none() { discovery::ssb_net_id() } else { auth::Key::from_slice(&hex::decode(net_id.unwrap()).unwrap()).unwrap() }; let OwnedIdentity { pk, sk, id } = keystore::from_gosbot_local() .await .expect("couldn't read local secret"); Ok(Self { id, public_key: pk, private_key: sk, address, network_id, }) } /// Creates a new connection with the sbot, /// using the address, network_id, public_key and private_key supplied when Sbot was initialized. /// /// Note that a single Sbot can have multiple SbotConnection at the same time. pub async fn get_sbot_connection(&self) -> Result { let address = self.address.clone(); let network_id = self.network_id.clone(); let public_key = self.public_key; let private_key = self.private_key.clone(); Sbot::_get_sbot_connection_helper(address, network_id, public_key, private_key).await } /// Private helper function which creates a new connection with sbot, /// but with all variables passed as arguments. async fn _get_sbot_connection_helper( address: String, network_id: auth::Key, public_key: ed25519::PublicKey, private_key: ed25519::SecretKey, ) -> Result { let socket = TcpStream::connect(&address) .await .map_err(|source| GolgiError::Io { source, context: "socket error; failed to initiate tcp stream connection".to_string(), })?; let handshake = kuska_handshake::async_std::handshake_client( &mut &socket, network_id.clone(), public_key, private_key.clone(), public_key, ) .await .map_err(GolgiError::Handshake)?; let (box_stream_read, box_stream_write) = BoxStream::from_handshake(socket.clone(), socket, handshake, 0x8000).split_read_write(); let rpc_reader = RpcReader::new(box_stream_read); let client = ApiCaller::new(RpcWriter::new(box_stream_write)); let sbot_connection = SbotConnection { rpc_reader, client }; Ok(sbot_connection) } /// Call the `partialReplication getSubset` RPC method /// and return a Stream of Result /// /// # Arguments /// /// * `query` - A `SubsetQuery` which specifies what filters to use. /// * `option` - An Option<`SubsetQueryOptions`> which, if provided, adds additional /// specifications to the query, such as specifying page limit and/or descending. pub async fn get_subset_stream( &mut self, query: SubsetQuery, options: Option, ) -> Result>, GolgiError> { let mut sbot_connection = self.get_sbot_connection().await?; let req_id = sbot_connection .client .getsubset_req_send(query, options) .await?; let get_subset_stream = get_source_stream( sbot_connection.rpc_reader, req_id, utils::ssb_message_res_parse, ) .await; Ok(get_subset_stream) } /// Call the `whoami` RPC method and return an `id`. pub async fn whoami(&mut self) -> Result { let mut sbot_connection = self.get_sbot_connection().await?; let req_id = sbot_connection.client.whoami_req_send().await?; let result = utils::get_async( &mut sbot_connection.rpc_reader, req_id, utils::json_res_parse, ) .await?; let id = result .get("id") .ok_or_else(|| GolgiError::Sbot("id key not found on whoami call".to_string()))? .as_str() .ok_or_else(|| GolgiError::Sbot("whoami returned non-string value".to_string()))?; Ok(id.to_string()) } /// Call the `publish` RPC method and return a message reference. /// /// # Arguments /// /// * `msg` - A `SsbMessageContent` `enum` whose variants include `Pub`, `Post`, `Contact`, `About`, /// `Channel` and `Vote`. See the `kuska_ssb` documentation for further details such as field /// names and accepted values for each variant. pub async fn publish(&mut self, msg: SsbMessageContent) -> Result { let mut sbot_connection = self.get_sbot_connection().await?; let req_id = sbot_connection.client.publish_req_send(msg).await?; utils::get_async( &mut sbot_connection.rpc_reader, req_id, utils::string_res_parse, ) .await } /// Call the `friends follow` RPC method and return a message reference. /// if state is true, then sets the following status to true /// if state is false, then sets the following status to false (unfollow) pub async fn friends_follow( &mut self, ssb_id: &str, state: bool, ) -> Result { let mut sbot_connection = self.get_sbot_connection().await?; let req_id = sbot_connection .client .friends_follow_req_send(ssb_id, state) .await?; utils::get_async( &mut sbot_connection.rpc_reader, req_id, utils::string_res_parse, ) .await } /// Call the `friends isfollowing` RPC method and return a message reference. /// returns true if src_id is following dest_id and false otherwise pub async fn friends_is_following( &mut self, src_id: &str, dest_id: &str, ) -> Result { let mut sbot_connection = self.get_sbot_connection().await?; let req_id = sbot_connection .client .friends_isfollowing_req_send(src_id, dest_id) .await?; utils::get_async( &mut sbot_connection.rpc_reader, req_id, utils::string_res_parse, ) .await } /// Wrapper for publish which constructs and publishes a post message appropriately from a string. /// /// # Arguments /// /// * `text` - A reference to a string slice which represents the text to be published in the post pub async fn publish_post(&mut self, text: &str) -> Result { let msg = SsbMessageContent::Post { text: text.to_string(), mentions: None, }; self.publish(msg).await } /// Wrapper for publish which constructs and publishes an about description message appropriately from a string. /// /// # Arguments /// /// * `description` - A reference to a string slice which represents the text to be published as an about description. pub async fn publish_description(&mut self, description: &str) -> Result { let msg = SsbMessageContent::About { about: self.id.to_string(), name: None, title: None, branch: None, image: None, description: Some(description.to_string()), location: None, start_datetime: None, }; self.publish(msg).await } /// Wrapper for publish which constructs and publishes an about name message appropriately from a string. /// /// # Arguments /// /// * `name` - A reference to a string slice which represents the text to be published as an about name. pub async fn publish_name(&mut self, name: &str) -> Result { let msg = SsbMessageContent::About { about: self.id.to_string(), name: Some(name.to_string()), title: None, branch: None, image: None, description: None, location: None, start_datetime: None, }; self.publish(msg).await } /// Get the about messages for a particular user in order of recency. pub async fn get_about_message_stream( &mut self, ssb_id: &str, ) -> Result>, GolgiError> { let query = SubsetQuery::Author { op: "author".to_string(), feed: ssb_id.to_string(), }; // specify that most recent messages should be returned first let query_options = SubsetQueryOptions { descending: Some(true), keys: None, page_limit: None, }; let get_subset_stream = self.get_subset_stream(query, Some(query_options)).await?; // TODO: after fixing sbot regression, // change this subset query to filter by type about in addition to author // and remove this filter section // filter down to about messages let about_message_stream = get_subset_stream.filter(|msg| match msg { Ok(val) => val.is_message_type(SsbMessageContentType::About), Err(_err) => false, }); // return about message stream Ok(about_message_stream) } /// Get value of latest about message with given key from given user pub async fn get_latest_about_message( &mut self, ssb_id: &str, key: &str, ) -> Result, GolgiError> { // get about_message_stream let about_message_stream = self.get_about_message_stream(ssb_id).await?; // now we have a stream of about messages with most recent at the front of the vector pin_mut!(about_message_stream); // iterate through the vector looking for most recent about message with the given key let latest_about_message_res: Option> = about_message_stream // find the first msg that contains the field `key` .find(|res| match res { Ok(msg) => msg.content.get(key).is_some(), Err(_) => false, }) .await; // Option> -> Option let latest_about_message = latest_about_message_res.and_then(|msg| msg.ok()); // Option -> Option let latest_about_value = latest_about_message.and_then(|msg| { msg // SsbMessageValue -> Option<&Value> .content .get(key) // Option<&Value> -> .and_then(|value| value.as_str()) // Option<&str> -> Option .map(|value| value.to_string()) }); // return value is either `Ok(Some(String))` or `Ok(None)` Ok(latest_about_value) } /// Get HashMap of profile info for given user pub async fn get_profile_info( &mut self, ssb_id: &str, ) -> Result, GolgiError> { let mut keys_to_search_for = vec!["name", "description", "image"]; self.get_about_info(ssb_id, keys_to_search_for).await } /// Get HashMap of name and image for given user /// (this is can be used to display profile images of a list of users) pub async fn get_name_and_image( &mut self, ssb_id: &str, ) -> Result, GolgiError> { let mut keys_to_search_for = vec!["name", "image"]; self.get_about_info(ssb_id, keys_to_search_for).await } /// Get HashMap of about keys to values for given user /// by iteratively searching through a stream of about messages, /// in order of recency, /// until we find all about messages for all needed info /// or reach the end of the stream. /// /// # Arguments /// /// * `ssb_id` - A reference to a string slice which represents the id of the user to get info about. /// * `keys_to_search_for` - A mutable vector of string slice, which represent the about keys /// that will be searched for. As they are found, keys are removed from the vector. pub async fn get_about_info( &mut self, ssb_id: &str, mut keys_to_search_for: Vec<&str>, ) -> Result, GolgiError> { // get about_message_stream let about_message_stream = self.get_about_message_stream(ssb_id).await?; // now we have a stream of about messages with most recent at the front of the vector pin_mut!(about_message_stream); // needed for iteration let mut profile_info: HashMap = HashMap::new(); // iterate through the stream while it still has more values and // we still have keys we are looking for while let Some(res) = about_message_stream.next().await { // if there are no more keys we are looking for, then we are done if keys_to_search_for.len() == 0 { break; } // if there are still keys we are looking for, then continue searching match res { Ok(msg) => { // for each key we are searching for, check if this about // message contains a value for that key for key in &keys_to_search_for.clone() { let option_val = msg .content .get(key) .and_then(|val| val.as_str()) .map(|val| val.to_string()); match option_val { Some(val) => { // if a value is found, then insert it profile_info.insert(key.to_string(), val); // remove this key fom keys_to_search_for, since we are no longer searching for it keys_to_search_for.retain(|val| val != key) } None => continue, } } } Err(err) => { // skip errors continue; } } } Ok(profile_info) } /// Get latest about name from given user /// /// # Arguments /// /// * `ssb_id` - A reference to a string slice which represents the ssb user /// to lookup the about name for. pub async fn get_name(&mut self, ssb_id: &str) -> Result, GolgiError> { self.get_latest_about_message(ssb_id, "name").await } /// Get latest about description from given user /// /// # Arguments /// /// * `ssb_id` - A reference to a string slice which represents the ssb user /// to lookup the about description for. pub async fn get_description(&mut self, ssb_id: &str) -> Result, GolgiError> { self.get_latest_about_message(ssb_id, "description").await } /// Call the `createHistoryStream` RPC method /// and return a Stream of Result pub async fn create_history_stream( &mut self, id: String, ) -> Result>, GolgiError> { let mut sbot_connection = self.get_sbot_connection().await?; let args = CreateHistoryStreamIn::new(id); let req_id = sbot_connection .client .create_history_stream_req_send(&args) .await?; let history_stream = get_source_stream( sbot_connection.rpc_reader, req_id, utils::ssb_message_res_parse, ) .await; Ok(history_stream) } }