Merge pull request 'Add TanglesThread struct import for args' (#61) from tangle_docs_fix into main

Reviewed-on: #61

.gitignore

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

CHANGELOG.md

@@ -0,0 +1,5 @@
+# 0.1.1
+
+_26 February 2022_
+
+ - [PR #32](https://git.coopcloud.tech/golgi-ssb/golgi/pulls/32): Fix two filter-related bugs which resulted in `get_about_message_stream(ssb_id)` returning **all** messages authored by `ssb_id`.

Cargo.toml

@@ -1,18 +1,26 @@
 [package]
 name = "golgi"
-version = "0.1.0"
-authors = ["glyph <glyph@mycelial.technology>"]
+version = "0.2.4"
 edition = "2021"
+authors = ["Max Fowler <max@mfowler.info>", "Andrew Reid <glyph@mycelial.technology>"]
+readme = "README.md"
+description = "An asynchronous, experimental Scuttlebutt client library"
+repository = "https://git.coopcloud.tech/golgi-ssb/golgi"
+homepage = "http://golgi.mycelial.technology"
+license = "LGPL-3.0"
+keywords = ["scuttlebutt", "ssb", "decentralized", "peer-for-peer", "p4p"]
+exclude = ["git_hooks/", "examples/"]
 
 [dependencies]
 async-std = "1.10.0"
+async-stream = "0.3.2"
 base64 = "0.13.0"
-futures = "0.3.18"
+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"
-# waiting for a pr merge upstream
-kuska-ssb = { path = "../ssb" }
-# try to replace with miniserde
-serde = "1"
+kuska-ssb = { git = "https://github.com/Kuska-ssb/ssb" }
+serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+sha2 = "0.10.2"

LICENSE.txt

@@ -0,0 +1,165 @@
+                   GNU LESSER GENERAL PUBLIC LICENSE
+                       Version 3, 29 June 2007
+
+ Copyright (C) 2007 Free Software Foundation, Inc. <https://fsf.org/>
+ Everyone is permitted to copy and distribute verbatim copies
+ of this license document, but changing it is not allowed.
+
+
+  This version of the GNU Lesser General Public License incorporates
+the terms and conditions of version 3 of the GNU General Public
+License, supplemented by the additional permissions listed below.
+
+  0. Additional Definitions.
+
+  As used herein, "this License" refers to version 3 of the GNU Lesser
+General Public License, and the "GNU GPL" refers to version 3 of the GNU
+General Public License.
+
+  "The Library" refers to a covered work governed by this License,
+other than an Application or a Combined Work as defined below.
+
+  An "Application" is any work that makes use of an interface provided
+by the Library, but which is not otherwise based on the Library.
+Defining a subclass of a class defined by the Library is deemed a mode
+of using an interface provided by the Library.
+
+  A "Combined Work" is a work produced by combining or linking an
+Application with the Library.  The particular version of the Library
+with which the Combined Work was made is also called the "Linked
+Version".
+
+  The "Minimal Corresponding Source" for a Combined Work means the
+Corresponding Source for the Combined Work, excluding any source code
+for portions of the Combined Work that, considered in isolation, are
+based on the Application, and not on the Linked Version.
+
+  The "Corresponding Application Code" for a Combined Work means the
+object code and/or source code for the Application, including any data
+and utility programs needed for reproducing the Combined Work from the
+Application, but excluding the System Libraries of the Combined Work.
+
+  1. Exception to Section 3 of the GNU GPL.
+
+  You may convey a covered work under sections 3 and 4 of this License
+without being bound by section 3 of the GNU GPL.
+
+  2. Conveying Modified Versions.
+
+  If you modify a copy of the Library, and, in your modifications, a
+facility refers to a function or data to be supplied by an Application
+that uses the facility (other than as an argument passed when the
+facility is invoked), then you may convey a copy of the modified
+version:
+
+   a) under this License, provided that you make a good faith effort to
+   ensure that, in the event an Application does not supply the
+   function or data, the facility still operates, and performs
+   whatever part of its purpose remains meaningful, or
+
+   b) under the GNU GPL, with none of the additional permissions of
+   this License applicable to that copy.
+
+  3. Object Code Incorporating Material from Library Header Files.
+
+  The object code form of an Application may incorporate material from
+a header file that is part of the Library.  You may convey such object
+code under terms of your choice, provided that, if the incorporated
+material is not limited to numerical parameters, data structure
+layouts and accessors, or small macros, inline functions and templates
+(ten or fewer lines in length), you do both of the following:
+
+   a) Give prominent notice with each copy of the object code that the
+   Library is used in it and that the Library and its use are
+   covered by this License.
+
+   b) Accompany the object code with a copy of the GNU GPL and this license
+   document.
+
+  4. Combined Works.
+
+  You may convey a Combined Work under terms of your choice that,
+taken together, effectively do not restrict modification of the
+portions of the Library contained in the Combined Work and reverse
+engineering for debugging such modifications, if you also do each of
+the following:
+
+   a) Give prominent notice with each copy of the Combined Work that
+   the Library is used in it and that the Library and its use are
+   covered by this License.
+
+   b) Accompany the Combined Work with a copy of the GNU GPL and this license
+   document.
+
+   c) For a Combined Work that displays copyright notices during
+   execution, include the copyright notice for the Library among
+   these notices, as well as a reference directing the user to the
+   copies of the GNU GPL and this license document.
+
+   d) Do one of the following:
+
+       0) Convey the Minimal Corresponding Source under the terms of this
+       License, and the Corresponding Application Code in a form
+       suitable for, and under terms that permit, the user to
+       recombine or relink the Application with a modified version of
+       the Linked Version to produce a modified Combined Work, in the
+       manner specified by section 6 of the GNU GPL for conveying
+       Corresponding Source.
+
+       1) Use a suitable shared library mechanism for linking with the
+       Library.  A suitable mechanism is one that (a) uses at run time
+       a copy of the Library already present on the user's computer
+       system, and (b) will operate properly with a modified version
+       of the Library that is interface-compatible with the Linked
+       Version.
+
+   e) Provide Installation Information, but only if you would otherwise
+   be required to provide such information under section 6 of the
+   GNU GPL, and only to the extent that such information is
+   necessary to install and execute a modified version of the
+   Combined Work produced by recombining or relinking the
+   Application with a modified version of the Linked Version. (If
+   you use option 4d0, the Installation Information must accompany
+   the Minimal Corresponding Source and Corresponding Application
+   Code. If you use option 4d1, you must provide the Installation
+   Information in the manner specified by section 6 of the GNU GPL
+   for conveying Corresponding Source.)
+
+  5. Combined Libraries.
+
+  You may place library facilities that are a work based on the
+Library side by side in a single library together with other library
+facilities that are not Applications and are not covered by this
+License, and convey such a combined library under terms of your
+choice, if you do both of the following:
+
+   a) Accompany the combined library with a copy of the same work based
+   on the Library, uncombined with any other library facilities,
+   conveyed under the terms of this License.
+
+   b) Give prominent notice with the combined library that part of it
+   is a work based on the Library, and explaining where to find the
+   accompanying uncombined form of the same work.
+
+  6. Revised Versions of the GNU Lesser General Public License.
+
+  The Free Software Foundation may publish revised and/or new versions
+of the GNU Lesser General Public License from time to time. Such new
+versions will be similar in spirit to the present version, but may
+differ in detail to address new problems or concerns.
+
+  Each version is given a distinguishing version number. If the
+Library as you received it specifies that a certain numbered version
+of the GNU Lesser General Public License "or any later version"
+applies to it, you have the option of following the terms and
+conditions either of that published version or of any later version
+published by the Free Software Foundation. If the Library as you
+received it does not specify a version number of the GNU Lesser
+General Public License, you may choose any version of the GNU Lesser
+General Public License ever published by the Free Software Foundation.
+
+  If the Library as you received it specifies that a proxy can decide
+whether future versions of the GNU Lesser General Public License shall
+apply, that proxy's public statement of acceptance of any version is
+permanent authorization for you to choose that version for the
+Library.

README.md

@@ -1,18 +1,72 @@
 # golgi
 
-_The Golgi complex (aka. Golgi apparatus or Golgi body) packages proteins into membrane-bound vesicles inside the cell before the vesicles are sent to their destination._
+_The Golgi complex (aka. Golgi apparatus or Golgi body) packages proteins
+into membrane-bound vesicles inside the cell before the vesicles are sent
+to their destination._
 
 -----
 
-Golgi is an experimental Scuttlebutt client which uses the [kuska-ssb](https://github.com/Kuska-ssb) libraries and aims to provide a high-level API for interacting with an sbot instance. Development efforts are currently oriented towards [go-sbot](https://github.com/cryptoscope/ssb) interoperability.
+## Introduction
+
+Golgi is an asynchronous, experimental Scuttlebutt client that aims to
+facilitate Scuttlebutt application development. It provides a high-level
+API for interacting with an sbot instance and uses the
+[kuska-ssb](https://github.com/Kuska-ssb) libraries to make RPC calls.
+
+## Features
+
+Golgi offers the ability to invoke individual RPC methods while also
+providing a number of convenience methods which may involve multiple RPC
+calls and / or the processing of data received from those calls. The
+`Sbot` `struct` is the primary means of interacting with the library.
+
+Features include the ability to publish messages of various kinds; to
+retrieve messages (e.g. `about` and `description` messages) and formulate
+queries; to follow, unfollow, block and unblock a peer; to query the social
+graph; and to generate pub invite codes.
 
 ## Example Usage
 
-```rust
-pub async fn run() -> Result<(), GolgiError> {
-    let mut sbot_client = Sbot::init(None, None).await?;
+Basic usage is demonstrated below. Visit the [examples directory](https://git.coopcloud.tech/golgi-ssb/golgi/src/branch/main/examples) in the `golgi` repository for
+more comprehensive examples.
 
+```rust
+use golgi::{GolgiError, Sbot, sbot::Keystore};
+
+pub async fn run() -> Result<(), GolgiError> {
+    // Attempt to initialise a connection to an sbot instance using the
+    // secret file at the Patchwork path and the default IP address, port
+    // and network key (aka. capabilities key).
+    let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+
+    // Call the `whoami` RPC method to retrieve the public key for the sbot
+    // identity.
     let id = sbot_client.whoami().await?;
+    
+    // Print the public key (identity) to `stdout`.
     println!("{}", id);
+
+    // Compose an SSB post message type.
+    let post = SsbMessageContent::Post {
+        text: "Biology, eh?!".to_string(),
+        mentions: None,
+    };
+
+    // Publish the post.
+    let post_msg_reference = sbot_client.publish(post).await?;
+
+    // Print the reference (sigil-link) for the published post.
+    println!("{}", post_msg_reference);
+
+    Ok(())
 }
 ```
+
+## Authors
+
+ - [notplants](https://mfowler.info/)
+ - [glyph](https://mycelial.technology/)
+
+## License
+
+LGPL-3.0.

examples/basic.rs

@@ -0,0 +1,99 @@
+use std::process;
+
+use golgi::{messages::SsbMessageContent, sbot::Keystore, GolgiError, Sbot};
+
+// Golgi is an asynchronous library so we must call it from within an
+// async function. The `GolgiError` type encapsulates all possible
+// error variants for the library.
+async fn run() -> Result<(), GolgiError> {
+    // Attempt to initialise a connection to an sbot instance using the
+    // secret file at the Patchwork path and the default IP address, port
+    // and network key (aka. capabilities key).
+    let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+
+    // Alternatively, we could specify a non-standard IP and port.
+    // let ip_port = "127.0.0.1:8021".to_string();
+    // let mut sbot_client = Sbot::init(Some(ip_port), None).await?;
+
+    // Call the `whoami` RPC method to retrieve the public key for the sbot
+    // identity. This is our 'local' public key.
+    let id = sbot_client.whoami().await?;
+
+    // Print the public key (identity) to `stdout`.
+    println!("whoami: {}", id);
+
+    // Compose an SSB `about` message type.
+    // The `SsbMessageContent` type has many variants and allows for a high
+    // degree of control when creating messages.
+    let name = SsbMessageContent::About {
+        about: id.clone(),
+        name: Some("golgi".to_string()),
+        title: None,
+        branch: None,
+        image: None,
+        description: None,
+        location: None,
+        start_datetime: None,
+    };
+
+    // Publish the name message. The `publish` method returns a reference to
+    // the published message.
+    let name_msg_ref = sbot_client.publish(name).await?;
+
+    // Print the message reference to `stdout`.
+    println!("name_msg_ref: {}", name_msg_ref);
+
+    // Compose an SSB `post` message type.
+    let post = SsbMessageContent::Post {
+        text: "golgi go womp womp".to_string(),
+        mentions: None,
+    };
+
+    // Publish the post.
+    let post_msg_ref = sbot_client.publish(post).await?;
+
+    // Print the post reference to `stdout`.
+    println!("post_msg_ref: {}", post_msg_ref);
+
+    // Golgi also exposes convenience methods for some of the most common
+    // message types. Here we see an example of a convenience method for
+    // posting a description message. The description is for the local
+    // identity, ie. we are publishing this about "ourself".
+    let post_msg_ref = sbot_client
+        .publish_description("this is a description")
+        .await?;
+
+    // Print the description message reference to `stdout`.
+    println!("description: {}", post_msg_ref);
+
+    let author: String = id.clone();
+    println!("author: {:?}", author);
+
+    // Retrieve the description for the given public key (identity).
+    let description = sbot_client.get_description(&author).await?;
+
+    // Print the description to `stdout`.
+    println!("found description: {:?}", description);
+
+    // Compose and publish another `post` message type.
+    let post = SsbMessageContent::Post {
+        text: "golgi go womp womp2".to_string(),
+        mentions: None,
+    };
+
+    let post_msg_ref = sbot_client.publish(post).await?;
+    println!("post_msg_ref2: {}", post_msg_ref);
+
+    Ok(())
+}
+
+// Enable an async main function and execute the `run()` function,
+// catching any errors and printing them to `stderr` before exiting the
+// process.
+#[async_std::main]
+async fn main() {
+    if let Err(e) = run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

examples/friends.rs

@@ -0,0 +1,129 @@
+use std::process;
+
+use golgi::{
+    api::friends::{FriendsHops, RelationshipQuery},
+    sbot::Keystore,
+    GolgiError, Sbot,
+};
+
+// Golgi is an asynchronous library so we must call it from within an
+// async function. The `GolgiError` type encapsulates all possible
+// error variants for the library.
+async fn run() -> Result<(), GolgiError> {
+    // Attempt to initialise a connection to an sbot instance using the
+    // secret file at the Patchwork path and the default IP address, port
+    // and network key (aka. capabilities key).
+    let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+
+    // Call the `whoami` RPC method to retrieve the public key for the sbot
+    // identity. This is our 'local' public key.
+    let id = sbot_client.whoami().await?;
+
+    // Print the public key (identity) to `stdout`.
+    println!("whoami: {}", id);
+
+    // Define IDs (public keys) to follow and block.
+    let to_follow = String::from("@5Pt3dKy2HTJ0mWuS78oIiklIX0gBz6BTfEnXsbvke9c=.ed25519");
+    let to_block = String::from("@7Y4nwfQmVtAilEzi5knXdS2gilW7cGKSHXdXoT086LM=.ed25519");
+
+    // Set the relationship of the local identity to the `to_follow` identity.
+    // In this case, the `set_relationship` method publishes a `contact`
+    // message which defines following as `true`.
+    // A message reference is returned for the published `contact` message.
+    let response = sbot_client
+        .set_relationship(&to_follow, Some(true), None)
+        .await?;
+
+    // Print the message reference to `stdout`.
+    println!("follow_response: {:?}", response);
+
+    // Set the relationship of the local identity to the `to_block` identity.
+    // In this case, the `set_relationship` method publishes a `contact`
+    // message which defines blocking as `true`.
+    // A message reference is returned for the published `contact` message.
+    let response = sbot_client
+        .set_relationship(&to_block, None, Some(true))
+        .await?;
+
+    // Print the message reference to `stdout`.
+    println!("follow_response: {:?}", response);
+
+    // Golgi also exposes convenience methods for following and blocking.
+    // Here is an example of a simpler way to follow an identity.
+    let _follow_response = sbot_client.follow(&to_follow).await?;
+
+    // Blocking can be achieved in a similar fashion.
+    let _block_response = sbot_client.block(&to_block).await?;
+
+    // Get a list of peers within 0 hops of the local identity.
+    // This returns a list of peers whom we follow.
+    // If `max` is set to 1, the list will include the peers we follow plus
+    // the peers that they follow.
+    let follows = sbot_client
+        .friends_hops(FriendsHops {
+            max: 0,
+            start: None,
+            // The `reverse` parameter is not currently implemented in `go-sbot`.
+            reverse: Some(false),
+        })
+        .await?;
+
+    // Print the list of peers to `stdout`.
+    println!("follows: {:?}", follows);
+
+    // Determine if an identity (`source`) is following a second identity (`dest`).
+    // This method will return `true` or `false`.
+    let mref = sbot_client
+        .friends_is_following(RelationshipQuery {
+            source: id.clone(),
+            dest: to_follow.clone(),
+        })
+        .await?;
+
+    // Print the follow status to `stdout`.
+    println!("isfollowingmref: {}", mref);
+
+    // Determine if an identity (`source`) is blocking a second identity (`dest`).
+    let mref = sbot_client
+        .friends_is_blocking(RelationshipQuery {
+            source: id.clone(),
+            dest: to_block.clone(),
+        })
+        .await?;
+
+    // Print the block status to `stdout`.
+    println!("isblockingmref: {}", mref);
+
+    let mref = sbot_client
+        .friends_is_blocking(RelationshipQuery {
+            source: id.clone(),
+            dest: to_follow,
+        })
+        .await?;
+
+    // Output should be `false`.
+    println!("isblockingmref(should be false): {}", mref);
+
+    let mref = sbot_client
+        .friends_is_following(RelationshipQuery {
+            source: id,
+            dest: to_block.clone(),
+        })
+        .await?;
+
+    // Output should be `false`.
+    println!("isfollowingmref(should be false): {}", mref);
+
+    Ok(())
+}
+
+// Enable an async main function and execute the `run()` function,
+// catching any errors and printing them to `stderr` before exiting the
+// process.
+#[async_std::main]
+async fn main() {
+    if let Err(e) = run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

examples/invite.rs

@@ -0,0 +1,77 @@
+use std::process;
+
+use kuska_ssb::api::dto::content::PubAddress;
+
+use golgi::{messages::SsbMessageContent, sbot::Keystore, GolgiError, Sbot};
+
+// Golgi is an asynchronous library so we must call it from within an
+// async function. The `GolgiError` type encapsulates all possible
+// error variants for the library.
+async fn run() -> Result<(), GolgiError> {
+    // Attempt to initialise a connection to an sbot instance using the
+    // secret file at the Patchwork path and the default IP address, port
+    // and network key (aka. capabilities key).
+    let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+
+    // Call the `whoami` RPC method to retrieve the public key for the sbot
+    // identity. This is our 'local' public key.
+    let id = sbot_client.whoami().await?;
+
+    // Print the public key (identity) to `stdout`.
+    println!("whoami: {}", id);
+
+    // Compose a `pub` address type message.
+    let pub_address_msg = SsbMessageContent::Pub {
+        address: Some(PubAddress {
+            // IP address.
+            host: Some("127.0.0.1".to_string()),
+            // Port.
+            port: 8009,
+            // Public key.
+            key: id,
+        }),
+    };
+
+    // Publish the `pub` address message.
+    // This step is required for successful invite code creation.
+    let pub_msg_ref = sbot_client.publish(pub_address_msg).await?;
+
+    // Print the message reference to `stdout`.
+    println!("pub_msg_ref: {}", pub_msg_ref);
+
+    // Generate an invite code that can be used 1 time.
+    let invite_code = sbot_client.invite_create(1).await?;
+
+    // Print the invite code to `stdout`.
+    println!("invite (1 use): {:?}", invite_code);
+
+    // Generate an invite code that can be used 7 times.
+    let invite_code_2 = sbot_client.invite_create(7).await?;
+
+    // Print the invite code to `stdout`.
+    println!("invite (7 uses): {:?}", invite_code_2);
+
+    // Define an invite code.
+    let test_invite =
+        "net:ssbroom2.commoninternet.net:8009~shs:wm8a1zHWjtESv4XSKMWU/rPRhnAoAiSAe4hQSY0UF5A=";
+
+    // Redeem an invite code (initiating a mutual follow between the local
+    // identity and the identity which generated the code (`wm8a1z...`).
+    let mref = sbot_client.invite_use(test_invite).await?;
+
+    // Print the message reference to `stdout`.
+    println!("mref: {:?}", mref);
+
+    Ok(())
+}
+
+// Enable an async main function and execute the `run()` function,
+// catching any errors and printing them to `stderr` before exiting the
+// process.
+#[async_std::main]
+async fn main() {
+    if let Err(e) = run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

examples/ssb-client.rs

@@ -1,59 +0,0 @@
-use std::process;
-
-use kuska_ssb::api::dto::content::{SubsetQuery, TypedMessage};
-
-use golgi::error::GolgiError;
-use golgi::sbot::Sbot;
-
-async fn run() -> Result<(), GolgiError> {
-    let mut sbot_client = Sbot::init(None, None).await?;
-
-    let id = sbot_client.whoami().await?;
-    println!("{}", id);
-
-    /*
-    let name = TypedMessage::About {
-        about: id,
-        name: Some("golgi".to_string()),
-        title: None,
-        branch: None,
-        image: None,
-        description: None,
-        location: None,
-        start_datetime: None,
-    };
-
-    let name_msg_ref = sbot_client.publish(name).await?;
-    println!("{}", name_msg_ref);
-
-    let post = TypedMessage::Post {
-        text: "golgi go womp womp".to_string(),
-        mentions: None,
-    };
-
-    let post_msg_ref = sbot_client.publish(post).await?;
-    println!("{}", post_msg_ref);
-
-    let post_msg_ref = sbot_client
-        .publish_description("this is a description")
-        .await?;
-    println!("description: {}", post_msg_ref);
-    */
-
-    let query = SubsetQuery::Type {
-        op: "type".to_string(),
-        string: "vote".to_string(),
-    };
-    let query_response = sbot_client.getsubset(query).await?;
-    println!("{}", query_response);
-
-    Ok(())
-}
-
-#[async_std::main]
-async fn main() {
-    if let Err(e) = run().await {
-        eprintln!("Application error: {}", e);
-        process::exit(1);
-    }
-}

examples/streams.rs

@@ -0,0 +1,166 @@
+use std::process;
+
+use async_std::stream::StreamExt;
+use futures::TryStreamExt;
+
+use golgi::{
+    api::{
+        get_subset::{SubsetQuery, SubsetQueryOptions},
+        history_stream::CreateHistoryStream,
+    },
+    messages::{SsbMessageContentType, SsbMessageKVT},
+    sbot::Keystore,
+    GolgiError, Sbot,
+};
+
+// Golgi is an asynchronous library so we must call it from within an
+// async function. The `GolgiError` type encapsulates all possible
+// error variants for the library.
+async fn run() -> Result<(), GolgiError> {
+    // Attempt to initialise a connection to an sbot instance using the
+    // secret file at the Patchwork path and the default IP address, port
+    // and network key (aka. capabilities key).
+    let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+
+    // Call the `whoami` RPC method to retrieve the public key for the sbot
+    // identity. This is our 'local' public key.
+    let id = sbot_client.whoami().await?;
+
+    // Print the public key (identity) to `stdout`.
+    println!("whoami: {}", id);
+
+    let author = id.clone();
+
+    /* HISTORY STREAM EXAMPLE */
+
+    let history_stream_args = CreateHistoryStream::new(author.to_string());
+    // Create an ordered stream of all messages authored by the `author`
+    // identity. Messages are returned as KVTs (Key Value Timestamp).
+    let history_stream = sbot_client
+        .create_history_stream(history_stream_args)
+        .await?;
+
+    // Pin the stream to the stack to allow polling of the `future`.
+    futures::pin_mut!(history_stream);
+
+    println!("looping through stream");
+
+    // Iterate through each element in the stream and match on the `Result`.
+    // In this case, each element has type `Result<SsbMessageKVT, GolgiError>`.
+    while let Some(res) = history_stream.next().await {
+        match res {
+            Ok(kvt) => {
+                // Print the `SsbMessageKVT` of this element to `stdout`.
+                println!("kvt: {:?}", kvt);
+            }
+            Err(err) => {
+                // Print the `GolgiError` of this element to `stderr`.
+                eprintln!("err: {:?}", err);
+            }
+        }
+    }
+
+    println!("reached end of stream");
+
+    // Create an ordered stream of all messages authored by the `author`
+    // identity.
+    let history_stream = sbot_client
+        .create_history_stream(CreateHistoryStream::new(author.to_string()))
+        .await?;
+
+    // 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<SsbMessageKVT> = history_stream.try_collect().await?;
+
+    // Loop through the `SsbMessageKVT` elements, printing each one
+    // to `stdout`.
+    for x in results {
+        println!("x: {:?}", x);
+    }
+
+    // Create an ordered stream of all messages authored by the `author`
+    // identity.
+    let history_stream = sbot_client
+        .create_history_stream(CreateHistoryStream::new(author.to_string()))
+        .await?;
+
+    // Iterate through the elements in the stream and use `map` to convert
+    // 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(kvt) => {
+            let message_type = kvt.value.get_message_type()?;
+            // Return the message key and type.
+            let tuple: (String, SsbMessageContentType) = (kvt.key, message_type);
+            Ok(tuple)
+        }
+        Err(err) => Err(err),
+    });
+
+    // Pin the stream to the stack to allow polling of the `future`.
+    futures::pin_mut!(type_stream);
+
+    println!("looping through type stream");
+
+    // Iterate through each element in the stream and match on the `Result`.
+    // In this case, each element has type
+    // `Result<(String, SsbMessageContentType), GolgiError>`.
+    while let Some(res) = type_stream.next().await {
+        match res {
+            Ok(value) => {
+                println!("value: {:?}", value);
+            }
+            Err(err) => {
+                println!("err: {:?}", err);
+            }
+        }
+    }
+
+    println!("reached end of type stream");
+
+    /* SUBSET STREAM EXAMPLE */
+
+    // Compose a subset query for `post` message types.
+    let post_query = SubsetQuery::Type {
+        op: "type".to_string(),
+        string: "post".to_string(),
+    };
+
+    // Define the options for the query.
+    let post_query_opts = SubsetQueryOptions {
+        descending: Some(true),
+        keys: None,
+        page_limit: Some(5),
+    };
+
+    // Return 5 `post` type messages from any author in descending order.
+    let query_stream = sbot_client
+        .get_subset_stream(post_query, Some(post_query_opts))
+        .await?;
+
+    println!("looping through post query stream");
+
+    // Iterate over the stream and pretty-print each returned message
+    // value while ignoring any errors.
+    query_stream.for_each(|msg| match msg {
+        Ok(val) => println!("{:#?}", val),
+        Err(_) => (),
+    });
+
+    println!("reached end of post query stream");
+
+    Ok(())
+}
+
+// Enable an async main function and execute the `run()` function,
+// catching any errors and printing them to `stderr` before exiting the
+// process.
+#[async_std::main]
+async fn main() {
+    if let Err(e) = run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

examples/tangles.rs

@@ -0,0 +1,84 @@
+use std::process;
+
+use async_std::stream::StreamExt;
+
+use golgi::{api::tangles::TanglesThread, sbot::Keystore, GolgiError, Sbot};
+
+// Golgi is an asynchronous library so we must call it from within an
+// async function. The `GolgiError` type encapsulates all possible
+// error variants for the library.
+async fn run() -> Result<(), GolgiError> {
+    // Attempt to initialise a connection to an sbot instance using the
+    // secret file at the Patchwork path and the default IP address, port
+    // and network key (aka. capabilities key).
+    //let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    let mut sbot_client =
+        Sbot::init(Keystore::GoSbot, Some("127.0.0.1:8021".to_string()), None).await?;
+
+    // Call the `whoami` RPC method to retrieve the public key for the sbot
+    // identity. This is our 'local' public key.
+    let id = sbot_client.whoami().await?;
+
+    // Print the public key (identity) to `stdout`.
+    println!("whoami: {}", id);
+
+    /* TANGLES THREAD EXAMPLE */
+
+    // Define the message key that will be used to generate the tangle stream.
+    // This must be the key of the root message in the thread (ie. the first
+    // message of the thread).
+    let msg_key = "%vY53ethgEG1tNsKqmLm6isNSbsu+jwMf/UNGMfUiLm0=.sha256".to_string();
+
+    // Instantiate the arguments for the `tangles.thread` RPC by instantiating
+    // a new instant of the `TanglesThread` struct.
+    let args = TanglesThread::new(msg_key).keys_values(true, true);
+
+    // It's also possible to define the maximum number of messages
+    // returned from the stream by setting the `limit` value:
+    // Note: a limit of 3 will return a maximum of 4 messages! The root
+    // message + 3 other messages in the thread.
+    // let args = TanglesThread::new(msg_key).keys_values(true, true).limit(3);
+
+    // Create an ordered stream of all messages comprising the thread
+    // in which the given message resides. Messages are returned as KVTs
+    // (Key Value Timestamp).
+    let thread_stream = sbot_client.tangles_thread(args).await?;
+
+    // Pin the stream to the stack to allow polling of the `future`.
+    futures::pin_mut!(thread_stream);
+
+    println!("looping through stream");
+
+    // Iterate through each element in the stream and match on the `Result`.
+    // In this case, each element has type `Result<SsbMessageKVT, GolgiError>`.
+    while let Some(res) = thread_stream.next().await {
+        match res {
+            Ok(kvt) => {
+                // Print the `SsbMessageKVT` of this element to `stdout`.
+                println!("kvt: {:?}", kvt);
+            }
+            Err(err) => {
+                // Print the `GolgiError` of this element to `stderr`.
+                eprintln!("err: {:?}", err);
+            }
+        }
+    }
+
+    println!("reached end of stream");
+
+    // Take a look at the `examples/streams.rs` file in this repo for more
+    // ways of handling streams.
+
+    Ok(())
+}
+
+// Enable an async main function and execute the `run()` function,
+// catching any errors and printing them to `stderr` before exiting the
+// process.
+#[async_std::main]
+async fn main() {
+    if let Err(e) = run().await {
+        eprintln!("Application error: {}", e);
+        process::exit(1);
+    }
+}

git_hooks/pre-commit

@@ -0,0 +1 @@
+cargo fmt

src/api/about.rs

@@ -0,0 +1,473 @@
+//! Retrieve data about a peer.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::get_about_info`]
+//! - [`Sbot::get_about_message_stream`]
+//! - [`Sbot::get_description`]
+//! - [`Sbot::get_image`]
+//! - [`Sbot::get_latest_about_message`]
+//! - [`Sbot::get_name`]
+//! - [`Sbot::get_name_and_image`]
+//! - [`Sbot::get_profile_info`]
+//! - [`Sbot::get_signifier`]
+
+use std::collections::HashMap;
+
+use async_std::stream::{Stream, StreamExt};
+
+use crate::{
+    api::get_subset::{SubsetQuery, SubsetQueryOptions},
+    error::GolgiError,
+    messages::{SsbMessageContentType, SsbMessageValue},
+    sbot::Sbot,
+    utils,
+};
+
+impl Sbot {
+    /// Get all the `about` type messages for a peer in order of recency
+    /// (ie. most recent messages first).
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use async_std::stream::{Stream, StreamExt};
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn about_message_stream() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     let about_message_stream = sbot_client.get_about_message_stream(ssb_id).await?;
+    ///
+    ///     // Make the stream into an iterator.
+    ///     futures::pin_mut!(about_message_stream);
+    ///
+    ///     about_message_stream.for_each(|msg| {
+    ///         match msg {
+    ///             Ok(val) => println!("msg value: {:?}", val),
+    ///             Err(e) => eprintln!("error: {}", e),
+    ///         }
+    ///     }).await;
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_about_message_stream(
+        &mut self,
+        ssb_id: &str,
+    ) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>>, 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 the latest `image` value for a peer. The value is a blob reference.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn image_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     let image = sbot_client.get_image(ssb_id).await?;
+    ///
+    ///     println!("peer {} is identified by {}", ssb_id, image);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_image(&mut self, ssb_id: &str) -> Result<String, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .names_get_image_for_req_send(ssb_id)
+            .await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+
+    /// Get the value of the latest `about` type message, containing the given
+    /// `key`, for a peer.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn name_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///     let key = "name";
+    ///
+    ///     let name_info = sbot_client.get_latest_about_message(ssb_id, key).await?;
+    ///
+    ///     match name_info {
+    ///         Some(name) => println!("peer {} is named {}", ssb_id, name),
+    ///         None => println!("no name found for peer {}", ssb_id)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_latest_about_message(
+        &mut self,
+        ssb_id: &str,
+        key: &str,
+    ) -> Result<Option<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
+        futures::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<Result<SsbMessageValue, GolgiError>> =
+            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<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())
+        });
+
+        // return value is either `Ok(Some(String))` or `Ok(None)`
+        Ok(latest_about_value)
+    }
+
+    /// Get the latest `name`, `description` and `image` values for a peer,
+    /// as defined in their `about` type messages.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn profile_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     let profile_info = sbot_client.get_profile_info(ssb_id).await?;
+    ///
+    ///     let name = profile_info.get("name");
+    ///     let description = profile_info.get("description");
+    ///     let image = profile_info.get("image");
+    ///
+    ///     match (name, description, image) {
+    ///         (Some(name), Some(desc), Some(image)) => {
+    ///             println!(
+    ///                 "peer {} is named {}. their profile image blob reference is {} and they describe themself as follows: {}",
+    ///                 ssb_id, name, image, desc,
+    ///             )
+    ///         },
+    ///         (_, _, _) => {
+    ///             eprintln!("failed to retrieve all profile info values")
+    ///         }
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_profile_info(
+        &mut self,
+        ssb_id: &str,
+    ) -> Result<HashMap<String, String>, GolgiError> {
+        let key_to_search_for = vec!["description"];
+        let description = self.get_about_info(ssb_id, key_to_search_for).await?;
+
+        let mut profile_info = self.get_name_and_image(ssb_id).await?;
+        // extend the `profile_info` HashMap by adding the key-value from the
+        // `description` HashMap; `profile_info` now contains all three
+        // key-value pairs
+        profile_info.extend(description.into_iter());
+
+        Ok(profile_info)
+    }
+
+    /// Get the latest `name` and `image` values for a peer. This method can
+    /// be used to display profile images of a list of users.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn name_and_image_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     let profile_info = sbot_client.get_name_and_image(ssb_id).await?;
+    ///
+    ///     let name = profile_info.get("name");
+    ///     let image = profile_info.get("image");
+    ///
+    ///     match (name, image) {
+    ///         (Some(name), Some(image)) => {
+    ///             println!(
+    ///                 "peer {} is named {}. their profile image blob reference is {}.",
+    ///                 ssb_id, name, image,
+    ///             )
+    ///         },
+    ///         (Some(name), None) => {
+    ///             println!(
+    ///                 "peer {} is named {}. no image blob reference was found for them.",
+    ///                 ssb_id, name,
+    ///             )
+    ///         },
+    ///         (_, _) => {
+    ///             eprintln!("failed to retrieve all profile info values")
+    ///         }
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    pub async fn get_name_and_image(
+        &mut self,
+        ssb_id: &str,
+    ) -> Result<HashMap<String, String>, GolgiError> {
+        let mut name_and_image: HashMap<String, String> = HashMap::new();
+
+        let name = self.get_name(ssb_id).await?;
+        let image = self.get_image(ssb_id).await?;
+
+        name_and_image.insert("name".to_string(), name);
+        name_and_image.insert("image".to_string(), image);
+
+        Ok(name_and_image)
+    }
+
+    /// Get the latest values for the provided keys from the `about` type
+    /// messages of a peer. The method will return once a value has been
+    /// found for each key, or once all messages have been checked.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn about_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///     let keys_to_search_for = vec!["name", "description"];
+    ///
+    ///     let about_info = sbot_client.get_about_info(ssb_id, keys_to_search_for).await?;
+    ///
+    ///     let name = about_info.get("name");
+    ///     let description = about_info.get("description");
+    ///
+    ///     match (name, description) {
+    ///         (Some(name), Some(desc)) => {
+    ///             println!(
+    ///                 "peer {} is named {}. they describe themself as: {}",
+    ///                 ssb_id, name, desc,
+    ///             )
+    ///         },
+    ///         (Some(name), None) => {
+    ///             println!(
+    ///                 "peer {} is named {}. no description was found for them.",
+    ///                 ssb_id, name,
+    ///             )
+    ///         },
+    ///         (_, _) => {
+    ///             eprintln!("failed to retrieve all profile info values")
+    ///         }
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_about_info(
+        &mut self,
+        ssb_id: &str,
+        mut keys_to_search_for: Vec<&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!` is needed for iteration
+        futures::pin_mut!(about_message_stream);
+
+        let mut profile_info: HashMap<String, String> = 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.is_empty() {
+                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 about_val = msg.content.get("about").and_then(|val| val.as_str());
+                        let option_val = msg
+                            .content
+                            .get(key)
+                            .and_then(|val| val.as_str())
+                            .map(|val| val.to_string());
+                        match option_val {
+                            // only return val if this msg is about the given ssb_id
+                            Some(val) if about_val == Some(ssb_id) => {
+                                // if a value is found, then insert it
+                                profile_info.insert(key.to_string(), val);
+                                // remove this key from keys_to_search_for,
+                                // since we are no longer searching for it
+                                keys_to_search_for.retain(|val| val != key)
+                            }
+                            _ => continue,
+                        }
+                    }
+                }
+                Err(_err) => {
+                    // skip errors
+                    continue;
+                }
+            }
+        }
+
+        Ok(profile_info)
+    }
+
+    /// Get the latest `name` value for a peer. The public key of the peer
+    /// will be returned if no `name` value is found.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn name_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     let name = sbot_client.get_name(ssb_id).await?;
+    ///
+    ///     println!("peer {} is named {}", ssb_id, name);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_name(&mut self, ssb_id: &str) -> Result<String, GolgiError> {
+        self.get_signifier(ssb_id).await
+    }
+
+    /// Get the latest `description` value for a peer.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn description_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     if let Some(desc) = sbot_client.get_description(ssb_id).await? {
+    ///         println!("peer {} describes themself as follows: {}", ssb_id, desc)
+    ///     } else {
+    ///         eprintln!("no description found for peer {}", ssb_id)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_description(&mut self, ssb_id: &str) -> Result<Option<String>, GolgiError> {
+        self.get_latest_about_message(ssb_id, "description").await
+    }
+
+    /// Get the latest `name` value (signifier) for a peer. The public key of
+    /// the peer will be returned if no `name` value is found.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn name_info() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     let name = sbot_client.get_signifier(ssb_id).await?;
+    ///
+    ///     println!("peer {} is named {}", ssb_id, name);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_signifier(&mut self, ssb_id: &str) -> Result<String, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .names_get_signifier_req_send(ssb_id)
+            .await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+}

src/api/friends.rs

@@ -0,0 +1,432 @@
+//! Define peer relationships and query the social graph.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::block`]
+//! - [`Sbot::unblock`]
+//! - [`Sbot::follow`]
+//! - [`Sbot::unfollow`]
+//! - [`Sbot::friends_hops`]
+//! - [`Sbot::friends_is_blocking`]
+//! - [`Sbot::friends_is_following`]
+//! - [`Sbot::get_blocks`]
+//! - [`Sbot::get_follows`]
+//! - [`Sbot::set_relationship`]
+
+use crate::{error::GolgiError, messages::SsbMessageContent, sbot::Sbot, utils};
+
+// re-export friends-related kuska types
+pub use kuska_ssb::api::dto::content::{FriendsHops, RelationshipQuery};
+
+impl Sbot {
+    /// Follow a peer.
+    ///
+    /// This is a convenience method to publish a contact message with
+    /// following: `true`.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn follow_peer() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     match sbot_client.follow(ssb_id).await {
+    ///         Ok(msg_ref) => {
+    ///             println!("follow msg reference is: {}", msg_ref)
+    ///         },
+    ///         Err(e) => eprintln!("failed to follow {}: {}", ssb_id, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn follow(&mut self, contact: &str) -> Result<String, GolgiError> {
+        self.set_relationship(contact, Some(true), None).await
+    }
+
+    /// Unfollow a peer.
+    ///
+    /// This is a convenience method to publish a contact message with
+    /// following: `false`.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn unfollow_peer() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     match sbot_client.unfollow(ssb_id).await {
+    ///         Ok(msg_ref) => {
+    ///             println!("unfollow msg reference is: {}", msg_ref)
+    ///         },
+    ///         Err(e) => eprintln!("failed to unfollow {}: {}", ssb_id, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn unfollow(&mut self, contact: &str) -> Result<String, GolgiError> {
+        self.set_relationship(contact, Some(false), None).await
+    }
+
+    /// Block a peer.
+    ///
+    /// This is a convenience method to publish a contact message with
+    /// following: `false` and blocking: `true`.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn block_peer() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     match sbot_client.block(ssb_id).await {
+    ///         Ok(msg_ref) => {
+    ///             println!("block msg reference is: {}", msg_ref)
+    ///         },
+    ///         Err(e) => eprintln!("failed to block {}: {}", ssb_id, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn block(&mut self, contact: &str) -> Result<String, GolgiError> {
+        // we want to unfollow and block
+        self.set_relationship(contact, Some(false), Some(true))
+            .await
+    }
+
+    /// Unblock a peer.
+    ///
+    /// This is a convenience method to publish a contact message with
+    /// blocking: `false`.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn unblock_peer() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///
+    ///     match sbot_client.unblock(ssb_id).await {
+    ///         Ok(msg_ref) => {
+    ///             println!("unblock msg reference is: {}", msg_ref)
+    ///         },
+    ///         Err(e) => eprintln!("failed to unblock {}: {}", ssb_id, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn unblock(&mut self, contact: &str) -> Result<String, GolgiError> {
+        self.set_relationship(contact, None, Some(false)).await
+    }
+
+    /// Publish a contact message defining the relationship for a peer.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn relationship() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///     let following = Some(true);
+    ///     // Could also be `None` to only publish the following relationship.
+    ///     let blocking = Some(false);
+    ///
+    ///     match sbot_client.set_relationship(ssb_id, following, blocking).await {
+    ///         Ok(msg_ref) => {
+    ///             println!("contact msg reference is: {}", msg_ref)
+    ///         },
+    ///         Err(e) => eprintln!("failed to set relationship for {}: {}", ssb_id, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn set_relationship(
+        &mut self,
+        contact: &str,
+        following: Option<bool>,
+        blocking: Option<bool>,
+    ) -> Result<String, GolgiError> {
+        let msg = SsbMessageContent::Contact {
+            contact: Some(contact.to_string()),
+            following,
+            blocking,
+            autofollow: None,
+        };
+        self.publish(msg).await
+    }
+
+    /// Get the follow status of two peers (ie. does one peer follow the other?).
+    ///
+    /// A `RelationshipQuery` `struct` must be defined and passed into this method.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, api::friends::RelationshipQuery, sbot::Keystore};
+    ///
+    /// async fn relationship() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let peer_a = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///     let peer_b = "@3QoWCcy46X9a4jTnOl8m3+n1gKfbsukWuODDxNGN0W8=.ed25519";
+    ///
+    ///     let query = RelationshipQuery {
+    ///         source: peer_a.to_string(),
+    ///         dest: peer_b.to_string(),
+    ///     };
+    ///
+    ///     match sbot_client.friends_is_following(query).await {
+    ///         Ok(following) if following == "true" => {
+    ///             println!("{} is following {}", peer_a, peer_b)
+    ///         },
+    ///         Ok(_) => println!("{} is not following {}", peer_a, peer_b),
+    ///         Err(e) => eprintln!("failed to query relationship status for {} and {}: {}", peer_a,
+    ///         peer_b, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn friends_is_following(
+        &mut self,
+        args: RelationshipQuery,
+    ) -> Result<String, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .friends_is_following_req_send(args)
+            .await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+
+    /// Get the block status of two peers (ie. does one peer block the other?).
+    ///
+    /// A `RelationshipQuery` `struct` must be defined and passed into this method.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, api::friends::RelationshipQuery, sbot::Keystore};
+    ///
+    /// async fn relationship() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let peer_a = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519";
+    ///     let peer_b = "@3QoWCcy46X9a4jTnOl8m3+n1gKfbsukWuODDxNGN0W8=.ed25519";
+    ///
+    ///     let query = RelationshipQuery {
+    ///         source: peer_a.to_string(),
+    ///         dest: peer_b.to_string(),
+    ///     };
+    ///
+    ///     match sbot_client.friends_is_blocking(query).await {
+    ///         Ok(blocking) if blocking == "true" => {
+    ///             println!("{} is blocking {}", peer_a, peer_b)
+    ///         },
+    ///         Ok(_) => println!("{} is not blocking {}", peer_a, peer_b),
+    ///         Err(e) => eprintln!("failed to query relationship status for {} and {}: {}", peer_a,
+    ///         peer_b, e)
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn friends_is_blocking(
+        &mut self,
+        args: RelationshipQuery,
+    ) -> Result<String, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .friends_is_blocking_req_send(args)
+            .await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+
+    /// Get a list of peers blocked by the local peer.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn follows() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let follows = sbot_client.get_blocks().await?;
+    ///
+    ///     if follows.is_empty() {
+    ///         println!("we do not block any peers")
+    ///     } else {
+    ///         follows.iter().for_each(|peer| println!("we block {}", peer))
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_blocks(&mut self) -> Result<Vec<String>, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection.client.friends_blocks_req_send().await?;
+        utils::get_source_until_eof(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+
+    /// Get a list of peers followed by the local peer.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn follows() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let follows = sbot_client.get_follows().await?;
+    ///
+    ///     if follows.is_empty() {
+    ///         println!("we do not follow any peers")
+    ///     } else {
+    ///         follows.iter().for_each(|peer| println!("we follow {}", peer))
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_follows(&mut self) -> Result<Vec<String>, GolgiError> {
+        self.friends_hops(FriendsHops {
+            max: 0,
+            start: None,
+            reverse: Some(false),
+        })
+        .await
+    }
+
+    /// Get a list of peers following the local peer.
+    ///
+    /// NOTE: this method is not currently working as expected.
+    ///
+    /// go-sbot does not currently implement the `reverse=True` parameter.
+    /// As a result, the parameter is ignored and this method returns follows
+    /// instead of followers.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn followers() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     // let followers = sbot_client.get_followers().await?;
+    ///
+    ///     // if followers.is_empty() {
+    ///     //     println!("no peers follow us")
+    ///     // } else {
+    ///     //     followers.iter().for_each(|peer| println!("{} is following us", peer))
+    ///     // }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    async fn _get_followers(&mut self) -> Result<Vec<String>, GolgiError> {
+        self.friends_hops(FriendsHops {
+            max: 0,
+            start: None,
+            reverse: Some(true),
+        })
+        .await
+    }
+
+    /// Get a list of peers within the specified hops range.
+    ///
+    /// A `RelationshipQuery` `struct` must be defined and passed into this method.
+    ///
+    /// Hops = 0 returns a list of peers followed by the local identity.
+    /// Those peers may or may not be mutual follows (ie. friends).
+    ///
+    /// Hops = 1 includes the peers followed by the peers we follow.
+    /// For example, if the local identity follows Aiko and Aiko follows
+    /// Bridgette and Chris, hops = 1 will return a list including the public
+    /// keys for Aiko, Bridgette and Chris (even though Bridgette and Chris
+    /// are not followed by the local identity).
+    ///
+    /// When reverse = True, hops = 0 should return a list of peers who
+    /// follow the local identity, ie. followers (but this is not currently
+    /// implemented in go-sbot).
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, api::friends::FriendsHops, sbot::Keystore};
+    ///
+    /// async fn peers_within_range() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let hops = 2;
+    ///
+    ///     let query = FriendsHops {
+    ///         max: hops,
+    ///         reverse: Some(false),
+    ///         start: None,
+    ///     };
+    ///
+    ///     let peers = sbot_client.friends_hops(query).await?;
+    ///
+    ///     if peers.is_empty() {
+    ///         println!("no peers found within {} hops", hops)
+    ///     } else {
+    ///         peers.iter().for_each(|peer| println!("{} is within {} hops", peer, hops))
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn friends_hops(&mut self, args: FriendsHops) -> Result<Vec<String>, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection.client.friends_hops_req_send(args).await?;
+        utils::get_source_until_eof(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+}

src/api/get_subset.rs

@@ -0,0 +1,87 @@
+//! Perform subset queries.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::get_subset_stream`]
+
+use async_std::stream::Stream;
+
+use crate::{error::GolgiError, messages::SsbMessageValue, sbot::Sbot, utils};
+
+// re-export subset-related kuska types
+pub use kuska_ssb::api::dto::content::{SubsetQuery, SubsetQueryOptions};
+
+impl Sbot {
+    /// Make a subset query, as defined by the [Subset replication for SSB specification](https://github.com/ssbc/ssb-subset-replication-spec).
+    ///
+    /// Calls the `partialReplication. getSubset` RPC method.
+    ///
+    /// # Arguments
+    ///
+    /// * `query` - A `SubsetQuery` which specifies the desired query filters.
+    /// * `options` - An Option<`SubsetQueryOptions`> which, if provided, adds
+    /// additional specifications to the query, such as page limit and/or
+    /// descending results.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use async_std::stream::StreamExt;
+    /// use golgi::{
+    ///     Sbot,
+    ///     GolgiError,
+    ///     api::get_subset::{
+    ///         SubsetQuery,
+    ///         SubsetQueryOptions
+    ///     },
+    ///     sbot::Keystore
+    /// };
+    ///
+    /// async fn query() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let post_query = SubsetQuery::Type {
+    ///         op: "type".to_string(),
+    ///         string: "post".to_string()
+    ///     };
+    ///
+    ///     let post_query_opts = SubsetQueryOptions {
+    ///         descending: Some(true),
+    ///         keys: None,
+    ///         page_limit: Some(5),
+    ///     };
+    ///
+    ///     // Return 5 `post` type messages from any author in descending order.
+    ///     let query_stream = sbot_client
+    ///         .get_subset_stream(post_query, Some(post_query_opts))
+    ///         .await?;
+    ///
+    ///     // Iterate over the stream and pretty-print each returned message
+    ///     // value while ignoring any errors.
+    ///     query_stream.for_each(|msg| match msg {
+    ///         Ok(val) => println!("{:#?}", val),
+    ///         Err(_) => (),
+    ///     });
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn get_subset_stream(
+        &mut self,
+        query: SubsetQuery,
+        options: Option<SubsetQueryOptions>,
+    ) -> Result<impl Stream<Item = Result<SsbMessageValue, GolgiError>>, 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 = utils::get_source_stream(
+            sbot_connection.rpc_reader,
+            req_id,
+            utils::ssb_message_res_parse,
+        )
+        .await;
+        Ok(get_subset_stream)
+    }
+}

src/api/history_stream.rs

@@ -0,0 +1,59 @@
+//! Return a history stream.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::create_history_stream`]
+
+use async_std::stream::Stream;
+pub use kuska_ssb::api::dto::CreateHistoryStreamIn as CreateHistoryStream;
+
+use crate::{error::GolgiError, messages::SsbMessageKVT, sbot::Sbot, utils};
+
+impl Sbot {
+    /// Call the `createHistoryStream` RPC method. Returns messages in the form
+    /// of KVTs (Key Value Timestamp).
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use async_std::stream::StreamExt;
+    /// use golgi::{
+    ///     Sbot,
+    ///     GolgiError,
+    ///     sbot::Keystore,
+    ///     api::history_stream::CreateHistoryStream
+    /// };
+    ///
+    /// async fn history() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let ssb_id = "@zqshk7o2Rpd/OaZ/MxH6xXONgonP1jH+edK9+GZb/NY=.ed25519".to_string();
+    ///
+    ///     let args = CreateHistoryStream::new(ssb_id);
+    ///     let history_stream = sbot_client.create_history_stream(args).await?;
+    ///
+    ///     history_stream.for_each(|msg| {
+    ///         match msg {
+    ///             Ok(val) => println!("msg value: {:?}", val),
+    ///             Err(e) => eprintln!("error: {}", e),
+    ///         }
+    ///     }).await;
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn create_history_stream(
+        &mut self,
+        args: CreateHistoryStream,
+    ) -> Result<impl Stream<Item = Result<SsbMessageKVT, GolgiError>>, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .create_history_stream_req_send(&args)
+            .await?;
+        let history_stream =
+            utils::get_source_stream(sbot_connection.rpc_reader, req_id, utils::kvt_res_parse)
+                .await;
+        Ok(history_stream)
+    }
+}

src/api/invite.rs

@@ -0,0 +1,79 @@
+//! Create and use invite codes.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::invite_create`]
+//! - [`Sbot::invite_use`]
+
+use crate::{error::GolgiError, sbot::Sbot, utils};
+
+impl Sbot {
+    /// Generate an invite code.
+    ///
+    /// Calls the `invite.create` RPC method and returns the code.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn invite_code_generator() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let invite_code = sbot_client.invite_create(5).await?;
+    ///
+    ///     println!("this invite code can be used 5 times: {}", invite_code);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn invite_create(&mut self, uses: u16) -> Result<String, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection.client.invite_create_req_send(uses).await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+
+    /// Use an invite code.
+    ///
+    /// Calls the `invite.use` RPC method and returns a reference to the follow
+    /// message.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn invite_code_consumer() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let invite_code = "127.0.0.1:8008:@0iMa+vP7B2aMrV3dzRxlch/iqZn/UM3S3Oo2oVeILY8=.ed25519~ZHNjeajPB/84NjjsrglZInlh46W55RcNDPcffTPgX/Q=";
+    ///     
+    ///     match sbot_client.invite_use(invite_code).await {
+    ///         Ok(msg_ref) => println!("consumed invite code. msg reference: {}", msg_ref),
+    ///         Err(e) => eprintln!("failed to consume the invite code: {}", e),
+    ///     }
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn invite_use(&mut self, invite_code: &str) -> Result<String, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .invite_use_req_send(invite_code)
+            .await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+}

src/api/mod.rs

@@ -0,0 +1,12 @@
+//! API for interacting with a running go-sbot instance.
+pub mod about;
+pub mod friends;
+pub mod get_subset;
+pub mod history_stream;
+pub mod invite;
+pub mod private;
+pub mod publish;
+pub mod tangles;
+pub mod whoami;
+
+pub use crate::sbot::*;

src/api/private.rs

@@ -0,0 +1,66 @@
+//! Publish Scuttlebutt private messages.
+//!
+//! Implements the following method:
+//!
+//! - [`Sbot::publish_private`]
+
+use crate::{error::GolgiError, messages::SsbMessageContent, sbot::Sbot, utils};
+
+impl Sbot {
+    /// Publish a private message.
+    ///
+    /// # Arguments
+    ///
+    /// * `msg` - A `PrivateMessage` `struct` whose fields include `text` and
+    /// `recipients`.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, messages::SsbMessageContent, sbot::Keystore};
+    ///
+    /// async fn publish_a_private_msg() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let text = String::from("Hi friends, I have a super secrect message to share with you about the wonders of intra-cellular transport mechanics.");
+    ///
+    ///     // We must also include the local identity (public key) here if we wish
+    ///     // to be able to read the message. Ie. the sender must be included in
+    ///     // the list of recipients.
+    ///     let recipients = vec![
+    ///         String::from("@OKRij/n7Uu42A0Z75ty0JI0cZxcieD2NyjXrRdYKNOQ=.ed25519"),
+    ///         String::from("@Sih4JGgs5oQPXehRyHS5qrYbx/0hQVUqChojX0LNtcQ=.ed25519"),
+    ///         String::from("@BVA85B7a/a17v2ZVcLkMgPE+v7X5rQVAHEgQBbCaKMs=.ed25519"),
+    ///     ];
+    ///         
+    ///     let msg_ref = sbot_client.publish_private(text, recipients).await?;
+    ///
+    ///     println!("msg reference for the private msg: {}", msg_ref);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn publish_private(
+        &mut self,
+        text: String,
+        recipients: Vec<String>,
+    ) -> Result<String, GolgiError> {
+        let msg = SsbMessageContent::Post {
+            text: text.to_string(),
+            mentions: None,
+        };
+
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .private_publish_req_send(msg, recipients)
+            .await?;
+
+        utils::get_async(
+            &mut sbot_connection.rpc_reader,
+            req_id,
+            utils::string_res_parse,
+        )
+        .await
+    }
+}

src/api/publish.rs

@@ -0,0 +1,196 @@
+//! Publish Scuttlebutt messages.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::publish`]
+//! - [`Sbot::publish_description`]
+//! - [`Sbot::publish_image`]
+//! - [`Sbot::publish_name`]
+//! - [`Sbot::publish_post`]
+
+use kuska_ssb::api::dto::content::Image;
+
+use crate::{error::GolgiError, messages::SsbMessageContent, sbot::Sbot, utils};
+
+impl Sbot {
+    /// Publish a message.
+    ///
+    /// # Arguments
+    ///
+    /// * `msg` - A `SsbMessageContent` `enum` whose variants include `Pub`,
+    /// `Post`, `Contact`, `About`, `Channel` and `Vote`.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, messages::SsbMessageContent, sbot::Keystore};
+    ///
+    /// async fn publish_a_msg() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     // Construct an SSB message of type `post`.
+    ///     let post = SsbMessageContent::Post {
+    ///         text: "And then those vesicles, filled with the Golgi products, move to the rest of the cell".to_string(),
+    ///         mentions: None,
+    ///     };
+    ///
+    ///     let msg_ref = sbot_client.publish(post).await?;
+    ///
+    ///     println!("msg reference for the golgi post: {}", msg_ref);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn publish(&mut self, msg: SsbMessageContent) -> Result<String, GolgiError> {
+        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
+    }
+
+    /// Publish a post.
+    ///
+    /// Convenient wrapper around the `publish` method which constructs and
+    /// publishes a `post` type message appropriately from a string.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn publish_a_post() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let text = "The Golgi is located right near the nucleus.";
+    ///
+    ///     let msg_ref = sbot_client.publish_post(text).await?;
+    ///
+    ///     println!("msg reference for the golgi post: {}", msg_ref);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn publish_post(&mut self, text: &str) -> Result<String, GolgiError> {
+        let msg = SsbMessageContent::Post {
+            text: text.to_string(),
+            mentions: None,
+        };
+        self.publish(msg).await
+    }
+
+    /// Publish a description for the local identity.
+    ///
+    /// Convenient wrapper around the `publish` method which constructs and
+    /// publishes an `about` type description message appropriately from a string.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn publish_a_description() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let description = "The Golgi apparatus was identified by the Italian scientist Camillo Golgi in 1897.";
+    ///
+    ///     let msg_ref = sbot_client.publish_description(description).await?;
+    ///
+    ///     println!("msg reference for the golgi description: {}", msg_ref);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn publish_description(&mut self, description: &str) -> Result<String, GolgiError> {
+        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
+    }
+
+    /// Publish an image for the local identity.
+    ///
+    /// Requires the image to have been added to the local blobstore. The
+    /// ID of the blob (sigil-link in the form `&...=.sha256`) must be provided.
+    ///
+    /// Convenient wrapper around the `publish` method which constructs and
+    /// publishes an `about` type name message appropriately from a string.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn publish_an_image() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let blob_id = "&JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8=.sha256";
+    ///
+    ///     let msg_ref = sbot_client.publish_image(blob_id).await?;
+    ///
+    ///     println!("msg reference: {}", msg_ref);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn publish_image(&mut self, blob_id: &str) -> Result<String, GolgiError> {
+        let msg = SsbMessageContent::About {
+            about: self.id.to_string(),
+            name: None,
+            title: None,
+            branch: None,
+            image: Some(Image::OnlyLink(blob_id.to_string())),
+            description: None,
+            location: None,
+            start_datetime: None,
+        };
+        self.publish(msg).await
+    }
+
+    /// Publish a name for the local identity.
+    ///
+    /// Convenient wrapper around the `publish` method which constructs and
+    /// publishes an `about` type name message appropriately from a string.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn publish_a_name() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let name = "glyphski_golgionikus";
+    ///
+    ///     let msg_ref = sbot_client.publish_name(name).await?;
+    ///
+    ///     println!("msg reference: {}", msg_ref);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    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
+    }
+}

src/api/tangles.rs

@@ -0,0 +1,61 @@
+//! Take a reference to the root message of a thread and return a stream of all
+//! messages in the thread. This includes the root message and all replies.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::tangles_thread`]
+
+use async_std::stream::Stream;
+pub use kuska_ssb::api::dto::TanglesThread;
+
+use crate::{error::GolgiError, messages::SsbMessageKVT, sbot::Sbot, utils};
+
+impl Sbot {
+    /// Call the `tanglesThread` RPC method. Returns messages in the form
+    /// of KVTs (Key Value Timestamp).
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use async_std::stream::StreamExt;
+    /// use golgi::{
+    ///     Sbot,
+    ///     GolgiError,
+    ///     api::tangles::TanglesThread,
+    ///     sbot::Keystore
+    /// };
+    ///
+    /// async fn tangle() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let msg_key = "%kmXb3MXtBJaNugcEL/Q7G40DgcAkMNTj3yhmxKHjfCM=.sha256";
+    ///
+    ///     let args = TanglesThread::new(msg_key).keys_values(true, true);
+    ///     let tangles_thread = sbot_client.tangles_thread(msg_id).await?;
+    ///
+    ///     tangles_thread.for_each(|msg| {
+    ///         match msg {
+    ///             Ok(kvt) => println!("msg kvt: {:?}", kvt),
+    ///             Err(e) => eprintln!("error: {}", e),
+    ///         }
+    ///     }).await;
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn tangles_thread(
+        &mut self,
+        args: TanglesThread,
+    ) -> Result<impl Stream<Item = Result<SsbMessageKVT, GolgiError>>, GolgiError> {
+        let mut sbot_connection = self.get_sbot_connection().await?;
+        let req_id = sbot_connection
+            .client
+            .tangles_thread_req_send(&args)
+            .await?;
+
+        let thread_stream =
+            utils::get_source_stream(sbot_connection.rpc_reader, req_id, utils::kvt_res_parse)
+                .await;
+        Ok(thread_stream)
+    }
+}

src/api/whoami.rs

@@ -0,0 +1,45 @@
+//! Return the SSB ID of the local sbot instance.
+//!
+//! Implements the following methods:
+//!
+//! - [`Sbot::whoami`]
+
+use crate::{error::GolgiError, sbot::Sbot, utils};
+
+impl Sbot {
+    /// Get the public key of the local identity.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use golgi::{Sbot, GolgiError, sbot::Keystore};
+    ///
+    /// async fn fetch_id() -> Result<(), GolgiError> {
+    ///     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
+    ///
+    ///     let pub_key = sbot_client.whoami().await?;
+    ///
+    ///     println!("local ssb id: {}", pub_key);
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn whoami(&mut self) -> Result<String, GolgiError> {
+        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())
+    }
+}

src/blobs.rs

@@ -0,0 +1,145 @@
+//! Blob utilities which do not require RPC calls.
+//!
+//! Offers the following functions:
+//!
+//! - [`get_blob_path()`]
+//! - [`hash_blob()`]
+
+use base64;
+use sha2::{Digest, Sha256};
+
+use crate::error::GolgiError;
+
+/// Lookup the filepath for a blob.
+///
+/// # Example
+///
+/// ```rust
+/// use golgi::{blobs, GolgiError};
+///
+/// fn lookup_blob_path() -> Result<(), GolgiError> {
+///     let blob_ref = "&JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8=.sha256";
+///
+///     let blob_path = blobs::get_blob_path(blob_ref)?;
+///
+///     println!("{}", blob_path);
+///
+///     // 26/524773dc9e1b5129640f5f20f18bcd42831f415e477f408b9edeba1293d46f
+///
+///     Ok(())
+/// }
+/// ```
+pub fn get_blob_path(blob_id: &str) -> Result<String, GolgiError> {
+    // ensure the id starts with the correct sigil link
+    if !blob_id.starts_with('&') {
+        return Err(GolgiError::SigilLink(format!(
+            "incorrect first character, expected '&' sigil: {}",
+            blob_id
+        )));
+    }
+
+    // find the dot index denoting the start of the algorithm definition tag
+    let dot_index = blob_id
+        .rfind('.')
+        .ok_or_else(|| GolgiError::SigilLink(format!("no dot index was found: {}", blob_id)))?;
+
+    // obtain the base64 portion (substring) of the blob id
+    let base64_str = &blob_id[1..dot_index];
+
+    // decode blob substring from base64 (to bytes)
+    let blob_bytes = base64::decode_config(base64_str, base64::STANDARD)?;
+
+    // represent the blob bytes as hex, removing all unnecessary characters
+    let blob_hex = format!("{:02x?}", blob_bytes)
+        .replace('[', "")
+        .replace(']', "")
+        .replace(',', "")
+        .replace(' ', "");
+
+    // split the hex representation of the decoded base64
+    // this is how paths are formatted for the blobstore
+    // e.g. 26/524773dc9e1b5129640f5f20f18bcd42831f415e477f408b9edeba1293d46f
+    // full path would be: `/home/user/.ssb-go/blobs/sha256/26/524773dc...`
+    let blob_path = format!("{}/{}", &blob_hex[..2], &blob_hex[2..]);
+
+    Ok(blob_path)
+}
+
+/// Hash the given blob byte slice and return the hex representation and
+/// blob ID (sigil-link).
+///
+/// This function can be used when adding a blob to the local blobstore.
+pub fn hash_blob(buffer: &[u8]) -> Result<(String, String), GolgiError> {
+    // hash the bytes
+    let hash = Sha256::digest(buffer);
+
+    // generate a hex representation of the hash
+    let hex_hash = format!("{:02x?}", hash)
+        .replace('[', "")
+        .replace(']', "")
+        .replace(',', "")
+        .replace(' ', "");
+
+    // encode the hash
+    let b64_hash = base64::encode(&hash[..]);
+
+    // format the base64 encoding as a blob sigil-link (blob id)
+    let blob_id = format!("&{}.sha256", b64_hash);
+
+    Ok((hex_hash, blob_id))
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::blobs;
+
+    /* HAPPY PATH TESTS */
+
+    #[test]
+    fn blob_path() {
+        let blob_ref = "&JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8=.sha256";
+        let blob_path = blobs::get_blob_path(blob_ref);
+        assert!(blob_path.is_ok());
+        assert_eq!(
+            blob_path.unwrap(),
+            "26/524773dc9e1b5129640f5f20f18bcd42831f415e477f408b9edeba1293d46f"
+        );
+    }
+
+    #[test]
+    fn hashed_blob() {
+        // pretend this represents a file which has been written to a buffer
+        let buffer = vec![7; 128];
+        let hash_result = blobs::hash_blob(&buffer);
+        assert!(hash_result.is_ok());
+        let (hex_hash, blob_id) = hash_result.unwrap();
+        assert_eq!(
+            hex_hash,
+            "4c1398e54d53e925cff04da532f4bbaf15f75b5981fc76c2072dfdc6491a9fb1"
+        );
+        assert_eq!(
+            blob_id,
+            "&TBOY5U1T6SXP8E2lMvS7rxX3W1mB/HbCBy39xkkan7E=.sha256"
+        );
+    }
+
+    /* SAD PATH TESTS */
+
+    #[test]
+    fn blob_id_without_sigil() {
+        let blob_ref = "JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8=.sha256";
+        match blobs::get_blob_path(blob_ref) {
+            Err(e) => assert_eq!(e.to_string(), "SSB blob ID error: incorrect first character, expected '&' sigil: JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8=.sha256"),
+            _ => panic!(),
+        }
+    }
+
+    #[test]
+    fn blob_id_without_algo() {
+        let blob_ref = "&JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8=";
+        match blobs::get_blob_path(blob_ref) {
+            Err(e) => assert_eq!(e.to_string(), "SSB blob ID error: no dot index was found: &JlJHc9yeG1EpZA9fIPGLzUKDH0FeR39Ai57euhKT1G8="),
+            _ => panic!(),
+        }
+    }
+}

src/error.rs

@@ -1,6 +1,7 @@
-//! Custom error type for `golgi`.
+//! Custom error type.
 
 use std::io::Error as IoError;
+use std::str::Utf8Error;
 
 use base64::DecodeError;
 use kuska_handshake::async_std::Error as HandshakeError;
@@ -33,8 +34,17 @@ pub enum GolgiError {
     Rpc(RpcError),
     /// Go-sbot error.
     Sbot(String),
+    /// SSB sigil-link error.
+    SigilLink(String),
     /// JSON serialization or deserialization error.
     SerdeJson(JsonError),
+    /// Error decoding typed SSB message from content.
+    ContentType(String),
+    /// Error decoding UTF8 string from bytes
+    Utf8Parse {
+        /// The underlying parse error.
+        source: Utf8Error,
+    },
 }
 
 impl std::error::Error for GolgiError {
@@ -47,7 +57,10 @@ impl std::error::Error for GolgiError {
             GolgiError::Feed(ref err) => Some(err),
             GolgiError::Rpc(ref err) => Some(err),
             GolgiError::Sbot(_) => None,
+            GolgiError::SigilLink(_) => None,
             GolgiError::SerdeJson(ref err) => Some(err),
+            GolgiError::ContentType(_) => None,
+            GolgiError::Utf8Parse { ref source } => Some(source),
         }
     }
 }
@@ -58,15 +71,24 @@ impl std::fmt::Display for GolgiError {
             // TODO: add context (what were we trying to decode?)
             GolgiError::DecodeBase64(_) => write!(f, "Failed to decode base64"),
             GolgiError::Io { ref context, .. } => write!(f, "IO error: {}", context),
-            GolgiError::Handshake(ref err) => write!(f, "{}", err),
+            GolgiError::Handshake(ref err) => write!(f, "Handshake failure: {}", err),
             GolgiError::Api(ref err) => write!(f, "SSB API failure: {}", err),
             GolgiError::Feed(ref err) => write!(f, "SSB feed error: {}", err),
             // TODO: improve this variant with a context message
             // then have the core display msg be: "SSB RPC error: {}", context
             GolgiError::Rpc(ref err) => write!(f, "SSB RPC failure: {}", err),
-            GolgiError::Sbot(ref err) => write!(f, "Sbot returned an error response: {}", err),
+            GolgiError::Sbot(ref err) => write!(f, "Sbot encountered an error: {}", err),
+            GolgiError::SigilLink(ref context) => write!(f, "SSB blob ID error: {}", context),
             GolgiError::SerdeJson(_) => write!(f, "Failed to serialize JSON slice"),
             //GolgiError::WhoAmI(ref err) => write!(f, "{}", err),
+            GolgiError::ContentType(ref err) => write!(
+                f,
+                "Failed to decode typed message from SSB message content: {}",
+                err
+            ),
+            GolgiError::Utf8Parse { source } => {
+                write!(f, "Failed to deserialize UTF8 from bytes: {}", source)
+            }
         }
     }
 }
@@ -106,3 +128,9 @@ impl From<JsonError> for GolgiError {
         GolgiError::SerdeJson(err)
     }
 }
+
+impl From<Utf8Error> for GolgiError {
+    fn from(err: Utf8Error) -> Self {
+        GolgiError::Utf8Parse { source: err }
+    }
+}

src/lib.rs

@@ -2,30 +2,78 @@
 
 //! # golgi
 //!
-//! _The Golgi complex (aka. Golgi apparatus or Golgi body) packages proteins into membrane-bound vesicles inside the cell before the vesicles are sent to their destination._
+//! _The Golgi complex (aka. Golgi apparatus or Golgi body) packages proteins
+//! into membrane-bound vesicles inside the cell before the vesicles are sent
+//! to their destination._
 //!
 //! -----
 //!
-//! Golgi is an experimental Scuttlebutt client which uses the [kuska-ssb](https://github.com/Kuska-ssb) libraries and aims to provide a high-level API for interacting with an sbot instance. Development efforts are currently oriented towards [go-sbot](https://github.com/cryptoscope/ssb) interoperability.
+//! ## Introduction
+//!
+//! Golgi is an asynchronous, experimental Scuttlebutt client that aims to
+//! facilitate Scuttlebutt application development. It provides a high-level
+//! API for interacting with an sbot instance and uses the
+//! [kuska-ssb](https://github.com/Kuska-ssb) libraries to make RPC calls.
+//! Development efforts are currently oriented towards
+//! [go-sbot](https://github.com/ssbc/go-ssb) interoperability.
+//!
+//! ## Features
+//!
+//! Golgi offers the ability to invoke individual RPC methods while also
+//! providing a number of convenience methods which may involve multiple RPC
+//! calls and / or the processing of data received from those calls. The
+//! [`Sbot`](crate::sbot::Sbot) `struct` is the primary means of interacting
+//! with the library.
+//!
+//! Features include the ability to publish messages of various kinds; to
+//! retrieve messages (e.g. `about` and `description` messages) and formulate
+//! queries; to follow, unfollow, block and unblock a peer; to query the social
+//! graph; and to generate pub invite codes.
+//!
+//! Visit the [API modules](crate::api) to view the available methods.
 //!
 //! ## Example Usage
 //!
+//! Basic usage is demonstrated below. Visit the [examples directory](https://git.coopcloud.tech/golgi-ssb/golgi/src/branch/main/examples) in the `golgi` repository for
+//! more comprehensive examples.
+//!
 //! ```rust
-//! use golgi::GolgiError;
-//! use golgi::sbot::Sbot;
+//! use golgi::{messages::SsbMessageContent, GolgiError, Sbot, sbot::Keystore};
 //!
 //! pub async fn run() -> Result<(), GolgiError> {
-//!     let mut sbot_client = Sbot::init(None, None).await?;
+//!     // Attempt to connect to an sbot instance using the default IP address,
+//!     // port and network key (aka. capabilities key).
+//!     let mut sbot_client = Sbot::init(Keystore::Patchwork, None, None).await?;
 //!
+//!     // Call the `whoami` RPC method to retrieve the public key for the sbot
+//!     // identity.
 //!     let id = sbot_client.whoami().await?;
+//!
+//!     // Print the public key (identity) to `stdout`.
 //!     println!("{}", id);
 //!
+//!     // Compose an SSB post message type.
+//!     let post = SsbMessageContent::Post {
+//!         text: "Biology, eh?!".to_string(),
+//!         mentions: None,
+//!     };
+//!
+//!     // Publish the post.
+//!     let post_msg_reference = sbot_client.publish(post).await?;
+//!
+//!     // Print the reference (sigil-link) for the published post.
+//!     println!("{}", post_msg_reference);
+//!
 //!     Ok(())
 //! }
 //! ```
 
+pub mod api;
+pub mod blobs;
 pub mod error;
+pub mod messages;
 pub mod sbot;
-mod utils;
+pub mod utils;
 
-pub use crate::error::GolgiError;
+pub use crate::{error::GolgiError, sbot::Sbot};
+pub use kuska_ssb;

src/messages.rs

@@ -0,0 +1,104 @@
+//! Message types and conversion methods.
+
+use kuska_ssb::api::dto::content::TypedMessage;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::fmt::Debug;
+
+use crate::error::GolgiError;
+
+/// `SsbMessageContent` is a type alias for `TypedMessage` from the `kuska_ssb` library.
+/// It is aliased in golgi to fit the naming convention of the other message
+/// types: `SsbMessageKVT` and `SsbMessageValue`.
+///
+/// See the [kuska source code](https://github.com/Kuska-ssb/ssb/blob/master/src/api/dto/content.rs#L103) for the type definition of `TypedMessage`.
+pub type SsbMessageContent = TypedMessage;
+
+/// The `value` of an SSB message (the `V` in `KVT`).
+///
+/// More information concerning the data model can be found in the
+/// [`Metadata` documentation](https://spec.scuttlebutt.nz/feed/messages.html#metadata).
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
+#[allow(missing_docs)]
+pub struct SsbMessageValue {
+    pub previous: Option<String>,
+    pub author: String,
+    pub sequence: u64,
+    pub timestamp: f64,
+    pub hash: String,
+    pub content: Value,
+    pub signature: String,
+}
+
+/// Message content types.
+#[derive(Debug, Eq, PartialEq)]
+#[allow(missing_docs)]
+pub enum SsbMessageContentType {
+    About,
+    Vote,
+    Post,
+    Contact,
+    Unrecognized,
+}
+
+impl SsbMessageValue {
+    /// Get the type field of the message content as an enum, if found.
+    ///
+    /// If no `type` field is found or the `type` field is not a string,
+    /// it returns an `Err(GolgiError::ContentType)`.
+    ///
+    /// If a `type` field is found but with an unknown string,
+    /// it returns an `Ok(SsbMessageContentType::Unrecognized)`.
+    pub fn get_message_type(&self) -> Result<SsbMessageContentType, GolgiError> {
+        let msg_type = self
+            .content
+            .get("type")
+            .ok_or_else(|| GolgiError::ContentType("type field not found".to_string()))?;
+        let mtype_str: &str = msg_type.as_str().ok_or_else(|| {
+            GolgiError::ContentType("type field value is not a string as expected".to_string())
+        })?;
+        let enum_type = match mtype_str {
+            "about" => SsbMessageContentType::About,
+            "post" => SsbMessageContentType::Post,
+            "vote" => SsbMessageContentType::Vote,
+            "contact" => SsbMessageContentType::Contact,
+            _ => SsbMessageContentType::Unrecognized,
+        };
+        Ok(enum_type)
+    }
+
+    /// Helper function which returns `true` if this message is of the given type,
+    /// and `false` if the type does not match or is not found.
+    pub fn is_message_type(&self, message_type: SsbMessageContentType) -> bool {
+        let self_message_type = self.get_message_type();
+        match self_message_type {
+            Ok(mtype) => mtype == message_type,
+            Err(_err) => false,
+        }
+    }
+
+    /// Convert the content JSON value into an `SsbMessageContent` `enum`,
+    /// using the `type` field as a tag to select which variant of the `enum`
+    /// to deserialize into.
+    ///
+    /// See the [Serde docs on internally-tagged enum representations](https://serde.rs/enum-representations.html#internally-tagged) for further details.
+    pub fn into_ssb_message_content(self) -> Result<SsbMessageContent, GolgiError> {
+        let m: SsbMessageContent = serde_json::from_value(self.content)?;
+        Ok(m)
+    }
+}
+
+/// An SSB message represented as a key-value-timestamp (`KVT`).
+///
+/// More information concerning the data model can be found in the
+/// [`Metadata` documentation](https://spec.scuttlebutt.nz/feed/messages.html#metadata).
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(deny_unknown_fields)]
+#[allow(missing_docs)]
+pub struct SsbMessageKVT {
+    pub key: String,
+    pub value: SsbMessageValue,
+    pub timestamp: Option<f64>,
+    pub rts: Option<f64>,
+}

src/sbot.rs

@@ -1,73 +1,158 @@
-//! Sbot type and associated methods.
-
+//! Sbot type and connection-related methods.
 use async_std::net::TcpStream;
 
 use kuska_handshake::async_std::BoxStream;
 use kuska_sodiumoxide::crypto::{auth, sign::ed25519};
 use kuska_ssb::{
-    api::{
-        dto::{
-            //content::{About, Post},
-            content::{SubsetQuery, TypedMessage},
-            CreateHistoryStreamIn,
-        },
-        ApiCaller,
-    },
+    api::ApiCaller,
     discovery, keystore,
     keystore::OwnedIdentity,
     rpc::{RpcReader, RpcWriter},
 };
 
 use crate::error::GolgiError;
-use crate::utils;
 
-/// 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.
+/// Keystore selector to specify the location of the secret file.
+///
+/// This enum is used when initiating a connection with an sbot instance.
+pub enum Keystore {
+    /// Patchwork default keystore path: `.ssb/secret` in the user's home directory.
+    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.
+/// 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 for writing requests to go-bot
+    pub client: ApiCaller<TcpStream>,
+    /// RpcReader object for reading responses from go-sbot
+    pub rpc_reader: RpcReader<TcpStream>,
+}
+
+/// Holds the Scuttlebutt identity, keys and configuration parameters for
+/// connecting to a local sbot and implements all Golgi API methods.
 pub struct Sbot {
-    id: String,
+    /// The ID (public key value) of the account associated with the local sbot instance.
+    pub id: String,
     public_key: ed25519::PublicKey,
     private_key: ed25519::SecretKey,
     address: String,
     // aka caps key (scuttleverse identifier)
     network_id: auth::Key,
-    client: ApiCaller<TcpStream>,
-    rpc_reader: RpcReader<TcpStream>,
 }
 
 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<String>, net_id: Option<String>) -> Result<Sbot, GolgiError> {
-        let address = if ip_port.is_none() {
+    /// 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.
+    pub async fn init(
+        keystore: Keystore,
+        ip_port: Option<String>,
+        net_id: Option<String>,
+    ) -> Result<Sbot, GolgiError> {
+        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 {
             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");
+        let OwnedIdentity { pk, sk, id } = match keystore {
+            Keystore::Patchwork => keystore::from_patchwork_local().await.map_err(|_err| {
+                GolgiError::Sbot(
+                    "sbot initialization error: couldn't read local patchwork secret from default location".to_string(),
+                )
+            })?,
+            Keystore::GoSbot => keystore::from_gosbot_local().await.map_err(|_err| {
+                GolgiError::Sbot(
+                    "sbot initialization error: couldn't read local go-sbot secret from default location".to_string(),
+                )
+            })?,
+            Keystore::CustomGoSbot(key_path) => {
+                keystore::from_custom_gosbot_keypath(key_path.to_string())
+                    .await
+                    .map_err(|_err| {
+                        GolgiError::Sbot(format!(
+                            "sbot initialization error: couldn't read local go-sbot secret from: {}",
+                            key_path
+                        ))
+                    })?
+            }
+            Keystore::CustomPatchwork(key_path) => {
+                keystore::from_custom_patchwork_keypath(key_path.to_string())
+                    .await
+                    .map_err(|_err| {
+                        GolgiError::Sbot(format!(
+                            "sbot initialization error: couldn't read local patchwork secret from: {}",
+                            key_path
+                        ))
+                    })?
+            }
+        };
 
+        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<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
+    }
+
+    /// Private helper function which creates a new connection with sbot,
+    /// but with all variables passed as arguments.
+    ///
+    /// 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.
+    async fn _get_sbot_connection_helper(
+        address: String,
+        network_id: auth::Key,
+        public_key: ed25519::PublicKey,
+        private_key: ed25519::SecretKey,
+    ) -> Result<SbotConnection, GolgiError> {
         let socket = TcpStream::connect(&address)
             .await
             .map_err(|source| GolgiError::Io {
                 source,
-                context: "socket error; failed to initiate tcp stream connection".to_string(),
+                context: "failed to initiate tcp stream connection".to_string(),
             })?;
 
         let handshake = kuska_handshake::async_std::handshake_client(
             &mut &socket,
             network_id.clone(),
-            pk,
-            sk.clone(),
-            pk,
+            public_key,
+            private_key.clone(),
+            public_key,
         )
         .await
         .map_err(GolgiError::Handshake)?;
@@ -77,94 +162,7 @@ impl Sbot {
 
         let rpc_reader = RpcReader::new(box_stream_read);
         let client = ApiCaller::new(RpcWriter::new(box_stream_write));
-
-        Ok(Self {
-            id,
-            public_key: pk,
-            private_key: sk,
-            address,
-            network_id,
-            client,
-            rpc_reader,
-        })
-    }
-
-    /// Call the `partialReplication getSubset` RPC method and return a vector
-    /// of messages as KVTs (key, value, timestamp).
-    // TODO: add args for `descending` and `page` (max number of msgs in response)
-    pub async fn getsubset(&mut self, query: SubsetQuery) -> Result<String, GolgiError> {
-        let req_id = self.client.getsubset_req_send(query).await?;
-
-        utils::get_async(&mut self.rpc_reader, req_id, utils::getsubset_res_parse).await
-    }
-
-    /// Call the `whoami` RPC method and return an `id`.
-    pub async fn whoami(&mut self) -> Result<String, GolgiError> {
-        let req_id = self.client.whoami_req_send().await?;
-
-        utils::get_async(&mut self.rpc_reader, req_id, utils::whoami_res_parse)
-            .await
-            .map(|whoami| whoami.id)
-    }
-
-    /// Call the `publish` RPC method and return a message reference.
-    ///
-    /// # Arguments
-    ///
-    /// * `msg` - A `TypedMessage` `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: TypedMessage) -> Result<String, GolgiError> {
-        let req_id = self.client.publish_req_send(msg).await?;
-
-        utils::get_async(&mut self.rpc_reader, req_id, utils::publish_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<String, GolgiError> {
-        let msg = TypedMessage::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<String, GolgiError> {
-        let msg = TypedMessage::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
-    }
-
-    /*
-    pub async fn publish_post(&mut self, post: Post) -> Result<String, GolgiError> {
-        let req_id = self.client.publish_req_send(post).await?;
-
-        utils::get_async(&mut self.rpc_reader, req_id, utils::publish_res_parse).await
-    }
-    */
-
-    /// Call the `createHistoryStream` RPC method and print the output.
-    async fn create_history_stream(&mut self, id: String) -> Result<(), GolgiError> {
-        let args = CreateHistoryStreamIn::new(id);
-        let req_id = self.client.create_history_stream_req_send(&args).await?;
-        // TODO: we should return a vector of messages instead of printing them
-        utils::print_source_until_eof(&mut self.rpc_reader, req_id, utils::feed_res_parse).await
+        let sbot_connection = SbotConnection { rpc_reader, client };
+        Ok(sbot_connection)
     }
 }

src/utils.rs

@@ -1,41 +1,66 @@
-/*
-
-use kuska_handshake::async_std::BoxStream;
-use kuska_sodiumoxide::crypto::sign::ed25519;
-use kuska_ssb::discovery;
-use kuska_ssb::keystore;
-use kuska_ssb::keystore::OwnedIdentity;
-*/
+//! Utility methods.
 use std::fmt::Debug;
 
-use async_std::io::Read;
-
-use kuska_ssb::api::dto::WhoAmIOut;
-use kuska_ssb::feed::Feed;
+use async_std::{io::Read, net::TcpStream, stream::Stream};
+use async_stream::stream;
 use kuska_ssb::rpc::{RecvMsg, RequestNo, RpcReader};
+use serde_json::Value;
 
 use crate::error::GolgiError;
+use crate::messages::{SsbMessageKVT, SsbMessageValue};
 
-pub fn getsubset_res_parse(body: &[u8]) -> Result<String, GolgiError> {
-    // TODO: cleanup with proper error handling etc.
-    Ok(std::str::from_utf8(body).unwrap().to_string())
+/// Parse an array of bytes (returned by an rpc call) into a `KVT`.
+///
+/// # Arguments
+///
+/// * `body` - An array of u8 to be parsed.
+pub fn kvt_res_parse(body: &[u8]) -> Result<SsbMessageKVT, GolgiError> {
+    let value: Value = serde_json::from_slice(body)?;
+    let kvt: SsbMessageKVT = serde_json::from_value(value)?;
+    Ok(kvt)
 }
 
-pub fn feed_res_parse(body: &[u8]) -> Result<Feed, GolgiError> {
-    Ok(Feed::from_slice(body)?)
+/// Parse an array of bytes (returned by an rpc call) into a `String`.
+///
+/// # Arguments
+///
+/// * `body` - An array of u8 to be parsed.
+pub fn string_res_parse(body: &[u8]) -> Result<String, GolgiError> {
+    Ok(std::str::from_utf8(body)?.to_string())
 }
 
-//pub fn publish_res_parse(body: &[u8]) -> Result<PublishOut, GolgiError> {
-pub fn publish_res_parse(body: &[u8]) -> Result<String, GolgiError> {
-    //Ok(serde_json::from_slice(body)?)
-    // TODO: cleanup with proper error handling etc.
-    Ok(std::str::from_utf8(body).unwrap().to_string())
+/// Parse an array of bytes (returned by an rpc call) into a `serde_json::Value`.
+///
+/// # Arguments
+///
+/// * `body` - An array of u8 to be parsed.
+pub fn json_res_parse(body: &[u8]) -> Result<Value, GolgiError> {
+    let message: Value = serde_json::from_slice(body)?;
+    Ok(message)
 }
 
-pub fn whoami_res_parse(body: &[u8]) -> Result<WhoAmIOut, GolgiError> {
-    Ok(serde_json::from_slice(body)?)
+/// Parse an array of bytes (returned by an rpc call) into an `SsbMessageValue`.
+///
+/// # Arguments
+///
+/// * `body` - An array of u8 to be parsed.
+pub fn ssb_message_res_parse(body: &[u8]) -> Result<SsbMessageValue, GolgiError> {
+    let message: SsbMessageValue = serde_json::from_slice(body)?;
+    Ok(message)
 }
 
+/// Take in an RPC request number along with a handling function and wait for
+/// an RPC response which matches the request number. Then, call the handling
+/// function on the response.
+///
+/// # Arguments
+///
+/// * `rpc_reader` - A `RpcReader` which can return Messages in a loop
+/// * `req_no` - A `RequestNo` of the response to listen for
+/// * `f` - A function which takes in an array of `u8` and returns a
+/// `Result<T, GolgiError>`. This is a function which parses the response from
+/// the `RpcReader`. `T` is a generic type, so this parse function can return
+/// multiple possible types (`String`, JSON, custom struct etc.)
 pub async fn get_async<'a, R, T, F>(
     rpc_reader: &mut RpcReader<R>,
     req_no: RequestNo,
@@ -51,17 +76,86 @@ 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));
                 }
+                RecvMsg::CancelStreamRespose() => {
+                    return Err(GolgiError::Sbot(
+                        "sbot returned CancelStreamResponse before any content".to_string(),
+                    ));
+                }
                 _ => {}
             }
         }
     }
 }
 
+/// Take in an RPC request number along with a handling function and call
+/// the handling function on all RPC responses which match the request number,
+/// appending the result of each parsed message to a vector until a
+/// `CancelStreamResponse` is found (marking the end of the stream).
+///
+/// # Arguments
+///
+/// * `rpc_reader` - A `RpcReader` which can return Messages in a loop
+/// * `req_no` - A `RequestNo` of the response to listen for
+/// * `f` - A function which takes in an array of `u8` and returns a
+/// `Result<T, GolgiError>`. This is a function which parses the response from
+/// the `RpcReader`. `T` is a generic type, so this parse function can return
+/// multiple possible types (`String`, JSON, custom struct etc.)
+pub async fn get_source_until_eof<'a, R, T, F>(
+    rpc_reader: &mut RpcReader<R>,
+    req_no: RequestNo,
+    f: F,
+) -> Result<Vec<T>, GolgiError>
+where
+    R: Read + Unpin,
+    F: Fn(&[u8]) -> Result<T, GolgiError>,
+    T: Debug,
+{
+    let mut messages: Vec<T> = Vec::new();
+    loop {
+        let (id, msg) = rpc_reader.recv().await?;
+        if id == req_no {
+            match msg {
+                RecvMsg::RpcResponse(_type, body) => {
+                    let parsed_response: Result<T, GolgiError> = f(&body);
+                    match parsed_response {
+                        Ok(parsed_message) => {
+                            messages.push(parsed_message);
+                        }
+                        Err(err) => {
+                            return Err(err);
+                        }
+                    }
+                }
+                RecvMsg::ErrorResponse(message) => {
+                    return Err(GolgiError::Sbot(message));
+                }
+                RecvMsg::CancelStreamRespose() => break,
+                _ => {}
+            }
+        }
+    }
+    Ok(messages)
+}
+
+/// Take in an RPC request number along with a handling function and call the
+/// handling function on all responses which match the request number. Then,
+/// prints out the result of the handling function.
+///
+/// This function useful for debugging and only prints the output.
+///
+/// # Arguments
+///
+/// * `rpc_reader` - A `RpcReader` which can return Messages in a loop
+/// * `req_no` - A `RequestNo` of the response to listen for
+/// * `f` - A function which takes in an array of `u8` and returns a
+/// `Result<T, GolgiError>`. This is a function which parses the response from
+/// the `RpcReader`. `T` is a generic type, so this parse function can return
+/// multiple possible types (`String`, JSON, custom struct etc.)
 pub async fn print_source_until_eof<'a, R, T, F>(
     rpc_reader: &mut RpcReader<R>,
     req_no: RequestNo,
@@ -90,3 +184,61 @@ where
     }
     Ok(())
 }
+
+/// Take in an RPC request number along with a handling function (parsing
+/// results of type `T`) and produce an `async_std::stream::Stream` of results
+/// of type `T`, where the handling function is called on all `RpcReader`
+/// responses which match the request number.
+///
+/// # Arguments
+///
+/// * `req_no` - A `RequestNo` of the response to listen for
+/// * `f` - A function which takes in an array of `u8` and returns a
+/// `Result<T, GolgiError>`. This is a function which parses the response from
+/// the `RpcReader`. `T` is a generic type, so this parse function can return
+/// multiple possible types (`String`, JSON, custom struct etc.)
+pub async fn get_source_stream<'a, F, T>(
+    mut rpc_reader: RpcReader<TcpStream>,
+    req_no: RequestNo,
+    f: F,
+) -> impl Stream<Item = Result<T, GolgiError>>
+where
+    F: Fn(&[u8]) -> Result<T, GolgiError>,
+    T: Debug + serde::Deserialize<'a>,
+{
+    // we use the async_stream::stream macro to allow for creating a stream which calls async functions
+    // see https://users.rust-lang.org/t/how-to-create-async-std-stream-which-calls-async-function-in-poll-next/69760
+    let source_stream = stream! {
+        loop {
+            // get the next message from the rpc_reader
+            let (id, msg) = rpc_reader.recv().await?;
+            let x : i32 = id.clone();
+            // check if the next message from rpc_reader matches the req_no we are looking for
+            // if it matches, then this rpc response is for the given request
+            // and if it doesn't match, then we ignore it
+            if x == req_no {
+                match msg {
+                    RecvMsg::RpcResponse(_type, body) => {
+                        // parse an item of type T from the message body using the provided
+                        // function for parsing
+                        let item = f(&body)?;
+                        // return Ok(item) as the next value in the stream
+                        yield Ok(item)
+                    }
+                    RecvMsg::ErrorResponse(message) => {
+                        // if an error is received
+                        // return an Err(err) as the next value in the stream
+                        yield Err(GolgiError::Sbot(message.to_string()));
+                    }
+                    // if we find a CancelStreamResponse
+                    // this is the end of the stream
+                    RecvMsg::CancelStreamRespose() => break,
+                    // if we find an unknown response, we just continue the loop
+                    _ => {}
+                }
+            }
+        }
+    };
+    // finally return the stream object
+    source_stream
+}