merge main

.gitignore

@@ -1 +1,2 @@
 /target
+Cargo.lock

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "golgi"
-version = "0.1.1"
+version = "0.2.2"
 edition = "2021"
 authors = ["Max Fowler <max@mfowler.info>", "Andrew Reid <glyph@mycelial.technology>"]
 readme = "README.md"
@@ -16,6 +16,7 @@ async-std = "1.10.0"
 async-stream = "0.3.2"
 base64 = "0.13.0"
 futures = "0.3.21"
+log = "0.4"
 hex = "0.4.3"
 kuska-handshake = { version = "0.2.0", features = ["async_std"] }
 kuska-sodiumoxide = "0.2.5-0"

examples/streams.rs

@@ -5,7 +5,7 @@ use futures::TryStreamExt;
 
 use golgi::{
     api::get_subset::{SubsetQuery, SubsetQueryOptions},
-    messages::{SsbMessageContentType, SsbMessageValue},
+    messages::{SsbMessageContentType, SsbMessageKVT},
     sbot::Keystore,
     GolgiError, Sbot,
 };
@@ -31,7 +31,7 @@ async fn run() -> Result<(), GolgiError> {
     /* HISTORY STREAM EXAMPLE */
 
     // Create an ordered stream of all messages authored by the `author`
-    // identity.
+    // identity. Messages are returned as KVTs (Key Value Timestamp).
     let history_stream = sbot_client
         .create_history_stream(author.to_string())
         .await?;
@@ -42,12 +42,12 @@ async fn run() -> Result<(), GolgiError> {
     println!("looping through stream");
 
     // Iterate through each element in the stream and match on the `Result`.
-    // In this case, each element has type `Result<SsbMessageValue, GolgiError>`.
+    // In this case, each element has type `Result<SsbMessageKVT, GolgiError>`.
     while let Some(res) = history_stream.next().await {
         match res {
-            Ok(value) => {
+            Ok(kvt) => {
-                // Print the `SsbMessageValue` of this element to `stdout`.
+                // Print the `SsbMessageKVT` of this element to `stdout`.
-                println!("value: {:?}", value);
+                println!("kvt: {:?}", kvt);
             }
             Err(err) => {
                 // Print the `GolgiError` of this element to `stderr`.
@@ -64,12 +64,12 @@ async fn run() -> Result<(), GolgiError> {
         .create_history_stream(author.to_string())
         .await?;
 
-    // Collect the stream elements into a `Vec<SsbMessageValue>` using
+    // Collect the stream elements into a `Vec<SsbMessageKVT>` using
     // `try_collect`. A `GolgiError` will be returned from the `run`
     // function if any element contains an error.
-    let results: Vec<SsbMessageValue> = history_stream.try_collect().await?;
+    let results: Vec<SsbMessageKVT> = history_stream.try_collect().await?;
 
-    // Loop through the `SsbMessageValue` elements, printing each one
+    // Loop through the `SsbMessageKVT` elements, printing each one
     // to `stdout`.
     for x in results {
         println!("x: {:?}", x);
@@ -82,13 +82,14 @@ async fn run() -> Result<(), GolgiError> {
         .await?;
 
     // Iterate through the elements in the stream and use `map` to convert
-    // each `SsbMessageValue` element into a tuple of
+    // each `SsbMessageKVT` element into a tuple of
     // `(String, SsbMessageContentType)`. This is an example of stream
     // conversion.
     let type_stream = history_stream.map(|msg| match msg {
-        Ok(val) => {
+        Ok(kvt) => {
-            let message_type = val.get_message_type()?;
+            let message_type = kvt.value.get_message_type()?;
-            let tuple: (String, SsbMessageContentType) = (val.signature, message_type);
+            // Return the message key and type.
+            let tuple: (String, SsbMessageContentType) = (kvt.key, message_type);
             Ok(tuple)
         }
         Err(err) => Err(err),

src/api/history_stream.rs

@@ -7,10 +7,11 @@
 use async_std::stream::Stream;
 use kuska_ssb::api::dto::CreateHistoryStreamIn;
 
-use crate::{error::GolgiError, messages::SsbMessageValue, sbot::Sbot, utils};
+use crate::{error::GolgiError, messages::SsbMessageKVT, sbot::Sbot, utils};
 
 impl Sbot {
-    /// Call the `createHistoryStream` RPC method.
+    /// Call the `createHistoryStream` RPC method. Returns messages in the form
+    /// of KVTs (Key Value Timestamp).
     ///
     /// # Example
     ///
@@ -38,18 +39,15 @@ impl Sbot {
     pub async fn create_history_stream(
         &mut self,
         id: String,
-    ) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>>, GolgiError> {
+    ) -> Result<impl Stream<Item = Result<SsbMessageKVT, GolgiError>>, GolgiError> {
         let mut sbot_connection = self.get_sbot_connection().await?;
-        let args = CreateHistoryStreamIn::new(id);
+        let args = CreateHistoryStreamIn::new(id).keys_values(true, true);
         let req_id = sbot_connection
             .client
             .create_history_stream_req_send(&args)
             .await?;
-        let history_stream = utils::get_source_stream(
+        let history_stream =
-            sbot_connection.rpc_reader,
+            utils::get_source_stream(sbot_connection.rpc_reader, req_id, utils::kvt_res_parse)
-            req_id,
-            utils::ssb_message_res_parse,
-        )
                 .await;
         Ok(history_stream)
     }

src/lib.rs

@@ -76,3 +76,4 @@ pub mod sbot;
 pub mod utils;
 
 pub use crate::{error::GolgiError, sbot::Sbot};
+pub use kuska_ssb;

src/sbot.rs

@@ -20,6 +20,10 @@ pub enum Keystore {
     Patchwork,
     /// GoSbot default keystore path: `.ssb-go/secret` in the user's home directory.
     GoSbot,
+    /// GoSbot keystore in a custom location
+    CustomGoSbot(String),
+    /// Patchwork keystore in a custom location
+    CustomPatchwork(String),
 }
 
 /// A struct representing a connection with a running sbot.
@@ -54,12 +58,16 @@ impl Sbot {
         ip_port: Option<String>,
         net_id: Option<String>,
     ) -> Result<Sbot, GolgiError> {
-        let address = if ip_port.is_none() {
+        let mut address = if ip_port.is_none() {
             "127.0.0.1:8008".to_string()
         } else {
             ip_port.unwrap()
         };
 
+        if address.starts_with(':') {
+            address = format!("127.0.0.1{}", address);
+        }
+
         let network_id = if net_id.is_none() {
             discovery::ssb_net_id()
         } else {
@@ -69,10 +77,24 @@ impl Sbot {
         let OwnedIdentity { pk, sk, id } = match keystore {
             Keystore::Patchwork => keystore::from_patchwork_local()
                 .await
-                .expect("couldn't read local secret"),
+                .expect("couldn't read local patchwork secret from default location"),
             Keystore::GoSbot => keystore::from_gosbot_local()
                 .await
-                .expect("couldn't read local secret"),
+                .expect("couldn't read local go-sbot secret from default location"),
+            Keystore::CustomGoSbot(key_path) => {
+                keystore::from_custom_gosbot_keypath(key_path.to_string())
+                    .await
+                    .unwrap_or_else(|_| {
+                        panic!("couldn't read local go-sbot secret from: {}", key_path)
+                    })
+            }
+            Keystore::CustomPatchwork(key_path) => {
+                keystore::from_custom_patchwork_keypath(key_path.to_string())
+                    .await
+                    .unwrap_or_else(|_| {
+                        panic!("couldn't read local patchwork secret from: {}", key_path)
+                    })
+            }
         };
 
         Ok(Self {

src/utils.rs

@@ -76,7 +76,7 @@ where
         if id == req_no {
             match msg {
                 RecvMsg::RpcResponse(_type, body) => {
-                    return f(&body).map_err(|err| err);
+                    return f(&body);
                 }
                 RecvMsg::ErrorResponse(message) => {
                     return Err(GolgiError::Sbot(message));