2021-12-03 09:37:22 +00:00
//! Sbot type and associated methods.
2022-01-12 19:58:07 +00:00
use std ::collections ::HashMap ;
2022-01-05 20:08:51 +00:00
use async_std ::{
net ::TcpStream ,
stream ::{ Stream , StreamExt } ,
} ;
2022-01-12 17:42:50 +00:00
use futures ::pin_mut ;
2021-12-02 17:55:31 +00:00
2021-12-02 17:48:16 +00:00
use kuska_handshake ::async_std ::BoxStream ;
use kuska_sodiumoxide ::crypto ::{ auth , sign ::ed25519 } ;
2021-12-02 17:55:31 +00:00
use kuska_ssb ::{
2022-01-05 18:58:48 +00:00
api ::{ dto ::CreateHistoryStreamIn , ApiCaller } ,
2021-12-29 16:38:03 +00:00
discovery , keystore ,
2021-12-02 17:55:31 +00:00
keystore ::OwnedIdentity ,
2022-01-05 18:58:48 +00:00
rpc ::{ RpcReader , RpcWriter } ,
2021-12-02 17:48:16 +00:00
} ;
use crate ::error ::GolgiError ;
2022-01-05 18:58:48 +00:00
use crate ::messages ::{ SsbMessageContent , SsbMessageContentType , SsbMessageKVT , SsbMessageValue } ;
2021-12-02 17:48:16 +00:00
use crate ::utils ;
2022-01-05 18:58:48 +00:00
use crate ::utils ::get_source_stream ;
2021-12-02 17:48:16 +00:00
2021-12-29 20:12:20 +00:00
// re-export types from kuska
2022-01-12 17:32:09 +00:00
pub use kuska_ssb ::api ::dto ::content ::{ SubsetQuery , SubsetQueryOptions } ;
2021-12-29 20:12:20 +00:00
2022-01-05 18:58:48 +00:00
/// 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.
2022-01-05 15:58:07 +00:00
pub struct SbotConnection {
client : ApiCaller < TcpStream > ,
rpc_reader : RpcReader < TcpStream > ,
}
2021-12-03 09:37:22 +00:00
/// The Scuttlebutt identity, keys and configuration parameters for connecting to a local sbot
/// instance, as well as handles for calling RPC methods and receiving responses.
2021-12-02 17:48:16 +00:00
pub struct Sbot {
2022-01-05 20:08:51 +00:00
pub id : String ,
2021-12-02 17:48:16 +00:00
public_key : ed25519 ::PublicKey ,
private_key : ed25519 ::SecretKey ,
address : String ,
// aka caps key (scuttleverse identifier)
network_id : auth ::Key ,
2022-01-05 18:58:48 +00:00
// the primary connection with sbot which can be re-used for non-stream calls
// note that stream calls will each need their own SbotConnection
sbot_connection : SbotConnection ,
2021-12-02 17:48:16 +00:00
}
impl Sbot {
2021-12-03 09:37:22 +00:00
/// 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.
2021-12-02 17:48:16 +00:00
pub async fn init ( ip_port : Option < String > , net_id : Option < String > ) -> Result < Sbot , GolgiError > {
2021-12-22 19:47:41 +00:00
let address = if ip_port . is_none ( ) {
" 127.0.0.1:8008 " . to_string ( )
2021-12-02 17:48:16 +00:00
} else {
2021-12-22 19:47:41 +00:00
ip_port . unwrap ( )
} ;
2021-12-02 17:48:16 +00:00
2021-12-22 19:47:41 +00:00
let network_id = if net_id . is_none ( ) {
discovery ::ssb_net_id ( )
2021-12-02 17:48:16 +00:00
} else {
2021-12-22 19:47:41 +00:00
auth ::Key ::from_slice ( & hex ::decode ( net_id . unwrap ( ) ) . unwrap ( ) ) . unwrap ( )
} ;
2021-12-02 17:48:16 +00:00
let OwnedIdentity { pk , sk , id } = keystore ::from_gosbot_local ( )
. await
. expect ( " couldn't read local secret " ) ;
2022-01-05 18:58:48 +00:00
let sbot_connection =
Sbot ::_get_sbot_connection_helper ( address . clone ( ) , network_id . clone ( ) , pk , sk . clone ( ) )
. await ? ;
2021-12-02 17:48:16 +00:00
Ok ( Self {
id ,
public_key : pk ,
private_key : sk ,
address ,
network_id ,
2022-01-05 18:58:48 +00:00
sbot_connection ,
2021-12-02 17:48:16 +00:00
} )
}
2022-01-05 18:58:48 +00:00
/// 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 < SbotConnection , GolgiError > {
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
}
2022-01-05 13:50:57 +00:00
2022-01-05 18:58:48 +00:00
/// 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 < SbotConnection , GolgiError > {
2022-01-05 13:50:57 +00:00
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 ( ) ,
2022-01-05 18:58:48 +00:00
public_key ,
private_key . clone ( ) ,
public_key ,
2022-01-05 13:50:57 +00:00
)
2022-01-05 18:58:48 +00:00
. await
. map_err ( GolgiError ::Handshake ) ? ;
2022-01-05 13:50:57 +00:00
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 ) ;
2022-01-05 15:58:07 +00:00
let client = ApiCaller ::new ( RpcWriter ::new ( box_stream_write ) ) ;
2022-01-05 18:58:48 +00:00
let sbot_connection = SbotConnection { rpc_reader , client } ;
2022-01-05 15:58:07 +00:00
Ok ( sbot_connection )
2022-01-05 13:50:57 +00:00
}
2022-01-05 20:08:51 +00:00
/// Call the `partialReplication getSubset` RPC method
2022-01-12 20:22:26 +00:00
/// and return a Stream of Result<SsbMessageValue, GolgiError>
2022-01-12 17:32:09 +00:00
///
/// # 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.
2022-01-05 20:08:51 +00:00
pub async fn get_subset_stream (
2022-01-05 18:58:48 +00:00
& mut self ,
query : SubsetQuery ,
2022-01-12 17:37:15 +00:00
options : Option < SubsetQueryOptions > ,
2022-01-12 20:22:26 +00:00
) -> Result < impl Stream < Item = Result < SsbMessageValue , GolgiError > > , GolgiError > {
2022-01-05 20:08:51 +00:00
let mut sbot_connection = self . get_sbot_connection ( ) . await ? ;
2022-01-12 17:37:15 +00:00
let req_id = sbot_connection
. client
. getsubset_req_send ( query , options )
. await ? ;
2022-01-05 20:08:51 +00:00
let get_subset_stream =
2022-01-12 20:22:26 +00:00
get_source_stream ( sbot_connection . rpc_reader , req_id , utils ::ssb_message_res_parse ) . await ;
2022-01-05 20:08:51 +00:00
Ok ( get_subset_stream )
2021-12-27 17:56:43 +00:00
}
2021-12-03 09:37:22 +00:00
/// Call the `whoami` RPC method and return an `id`.
2021-12-02 17:48:16 +00:00
pub async fn whoami ( & mut self ) -> Result < String , GolgiError > {
2022-01-05 18:58:48 +00:00
let req_id = self . sbot_connection . client . whoami_req_send ( ) . await ? ;
2021-12-02 17:48:16 +00:00
2022-01-05 20:08:51 +00:00
let result = utils ::get_async (
2022-01-05 18:58:48 +00:00
& mut self . sbot_connection . rpc_reader ,
req_id ,
2022-01-05 20:08:51 +00:00
utils ::json_res_parse ,
2022-01-05 18:58:48 +00:00
)
2022-01-05 20:08:51 +00:00
. await ? ;
let id = result
. get ( " id " )
2022-01-12 17:42:50 +00:00
. ok_or_else ( | | GolgiError ::Sbot ( " id key not found on whoami call " . to_string ( ) ) ) ?
2022-01-05 20:08:51 +00:00
. as_str ( )
2022-01-12 17:42:50 +00:00
. ok_or_else ( | | GolgiError ::Sbot ( " whoami returned non-string value " . to_string ( ) ) ) ? ;
2022-01-05 20:08:51 +00:00
Ok ( id . to_string ( ) )
2021-12-02 17:48:16 +00:00
}
2021-12-03 09:37:22 +00:00
/// Call the `publish` RPC method and return a message reference.
///
/// # Arguments
///
2021-12-29 15:59:05 +00:00
/// * `msg` - A `SsbMessageContent` `enum` whose variants include `Pub`, `Post`, `Contact`, `About`,
2021-12-03 09:37:22 +00:00
/// `Channel` and `Vote`. See the `kuska_ssb` documentation for further details such as field
/// names and accepted values for each variant.
2021-12-29 15:59:05 +00:00
pub async fn publish ( & mut self , msg : SsbMessageContent ) -> Result < String , GolgiError > {
2022-01-05 18:58:48 +00:00
let req_id = self . sbot_connection . client . publish_req_send ( msg ) . await ? ;
2021-12-03 09:37:22 +00:00
2022-01-05 18:58:48 +00:00
utils ::get_async (
& mut self . sbot_connection . rpc_reader ,
req_id ,
utils ::string_res_parse ,
)
. await
2021-12-03 09:37:22 +00:00
}
2021-12-22 19:47:41 +00:00
/// Wrapper for publish which constructs and publishes a post message appropriately from a string.
2021-12-22 19:42:01 +00:00
///
/// # Arguments
///
2021-12-24 15:34:07 +00:00
/// * `text` - A reference to a string slice which represents the text to be published in the post
2021-12-22 19:42:01 +00:00
pub async fn publish_post ( & mut self , text : & str ) -> Result < String , GolgiError > {
2021-12-29 15:59:05 +00:00
let msg = SsbMessageContent ::Post {
2021-12-27 17:56:43 +00:00
text : text . to_string ( ) ,
mentions : None ,
} ;
2021-12-22 19:42:01 +00:00
self . publish ( msg ) . await
}
2021-12-22 21:19:52 +00:00
/// Wrapper for publish which constructs and publishes an about description message appropriately from a string.
///
/// # Arguments
///
2021-12-24 15:34:07 +00:00
/// * `description` - A reference to a string slice which represents the text to be published as an about description.
2021-12-22 21:19:52 +00:00
pub async fn publish_description ( & mut self , description : & str ) -> Result < String , GolgiError > {
2021-12-29 15:59:05 +00:00
let msg = SsbMessageContent ::About {
2021-12-22 21:19:52 +00:00
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
}
2021-12-30 01:51:17 +00:00
/// 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 < String , GolgiError > {
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
}
2021-12-29 16:48:53 +00:00
2021-12-29 20:12:20 +00:00
/// Get the about messages for a particular user in order of recency.
2022-01-05 20:08:51 +00:00
pub async fn get_about_message_stream (
2022-01-05 18:58:48 +00:00
& mut self ,
ssb_id : & str ,
2022-01-05 20:08:51 +00:00
) -> Result < impl Stream < Item = Result < SsbMessageValue , GolgiError > > , GolgiError > {
2022-01-05 18:58:48 +00:00
let query = SubsetQuery ::Author {
2021-12-29 20:12:20 +00:00
op : " author " . to_string ( ) ,
2021-12-30 01:51:17 +00:00
feed : ssb_id . to_string ( ) ,
2021-12-29 20:12:20 +00:00
} ;
2022-01-12 17:32:09 +00:00
// specify that most recent messages should be returned first
let query_options = SubsetQueryOptions {
descending : Some ( true ) ,
keys : None ,
2022-01-12 17:37:15 +00:00
page_limit : None ,
2022-01-12 17:32:09 +00:00
} ;
2022-01-12 20:22:26 +00:00
let get_subset_stream = self . get_subset_stream ( query , Some ( query_options ) ) . await ? ;
2021-12-29 20:12:20 +00:00
// 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
2022-01-12 20:22:26 +00:00
let about_message_stream = get_subset_stream . filter ( | msg | match msg {
Ok ( val ) = > {
val . is_message_type ( SsbMessageContentType ::About )
}
2022-01-05 20:08:51 +00:00
Err ( _err ) = > false ,
} ) ;
// return about message stream
Ok ( about_message_stream )
2021-12-29 16:48:53 +00:00
}
2021-12-30 01:51:17 +00:00
/// Get value of latest about message with given key from given user
2022-01-05 18:58:48 +00:00
pub async fn get_latest_about_message (
& mut self ,
ssb_id : & str ,
key : & str ,
) -> Result < Option < String > , GolgiError > {
2022-01-05 20:08:51 +00:00
// get about_message_stream
let about_message_stream = self . get_about_message_stream ( ssb_id ) . await ? ;
2022-01-12 17:32:09 +00:00
// now we have a stream of about messages with most recent at the front of the vector
pin_mut! ( about_message_stream ) ;
2021-12-30 18:35:39 +00:00
// iterate through the vector looking for most recent about message with the given key
2022-01-12 19:58:07 +00:00
let latest_about_message_res : Option < Result < SsbMessageValue , GolgiError > > = about_message_stream
2021-12-30 18:35:39 +00:00
// find the first msg that contains the field `key`
2022-01-12 17:37:15 +00:00
. find ( | res | match res {
Ok ( msg ) = > msg . content . get ( key ) . is_some ( ) ,
Err ( _ ) = > false ,
} )
2022-01-12 19:58:07 +00:00
. await ;
// Option<Result<SsbMessageValue, GolgiError>> -> Option<SsbMessageValue>
let latest_about_message = latest_about_message_res . and_then ( | msg | msg . ok ( ) ) ;
// Option<SsbMessageValue> -> Option<String>
let latest_about_value = latest_about_message . and_then ( | msg | {
msg
// SsbMessageValue -> Option<&Value>
. content
. get ( key )
// Option<&Value> -> <Option<&str>
. and_then ( | value | value . as_str ( ) )
// Option<&str> -> Option<String>
. map ( | value | value . to_string ( ) )
} ) ;
2021-12-30 18:35:39 +00:00
// return value is either `Ok(Some(String))` or `Ok(None)`
2022-01-12 17:32:09 +00:00
Ok ( latest_about_value )
2021-12-30 01:51:17 +00:00
}
2022-01-12 19:58:07 +00:00
/// Get HashMap of profile info 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.
pub async fn get_profile_info (
& mut self ,
ssb_id : & str
) -> Result < HashMap < String , String > , 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 < String , String > = HashMap ::new ( ) ;
// these are the about message keys we are looking for
// as we find values for each of these keys,
// we will remove them from this vector (as we are no longer searching for them)
let mut keys_to_search_for = vec! [ " name " , " description " , " profile " ] ;
// 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 {
println! ( " res: {:?} " , res ) ;
// 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 )
}
2022-01-05 18:58:48 +00:00
/// 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.
2021-12-30 01:51:17 +00:00
pub async fn get_name ( & mut self , ssb_id : & str ) -> Result < Option < String > , GolgiError > {
self . get_latest_about_message ( ssb_id , " name " ) . await
2021-12-29 16:48:53 +00:00
}
2022-01-05 20:08:51 +00:00
/// Get latest about description from given user
2022-01-05 18:58:48 +00:00
///
/// # Arguments
///
/// * `ssb_id` - A reference to a string slice which represents the ssb user
/// to lookup the about description for.
2021-12-30 02:03:06 +00:00
pub async fn get_description ( & mut self , ssb_id : & str ) -> Result < Option < String > , GolgiError > {
self . get_latest_about_message ( ssb_id , " description " ) . await
}
2022-01-04 19:09:49 +00:00
/// Call the `createHistoryStream` RPC method
/// and return a Stream of Result<SsbMessageValue, GolgiError>
2022-01-05 13:50:57 +00:00
pub async fn create_history_stream (
& mut self ,
2021-12-29 16:31:53 +00:00
id : String ,
2022-01-05 13:50:57 +00:00
) -> Result < impl Stream < Item = Result < SsbMessageValue , GolgiError > > , GolgiError > {
2022-01-05 20:08:51 +00:00
let mut sbot_connection = self . get_sbot_connection ( ) . await ? ;
2021-12-02 17:48:16 +00:00
let args = CreateHistoryStreamIn ::new ( id ) ;
2022-01-05 18:58:48 +00:00
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 ;
2022-01-04 19:09:49 +00:00
Ok ( history_stream )
2021-12-02 17:48:16 +00:00
}
}