lykin/src/sbot.rs

144 lines
4.7 KiB
Rust

//! Scuttlebutt functionality.
use std::env;
use async_std::stream::StreamExt;
use chrono::NaiveDateTime;
use golgi::{
api::{friends::RelationshipQuery, history_stream::CreateHistoryStream},
messages::{SsbMessageContentType, SsbMessageKVT},
sbot::Keystore,
GolgiError, Sbot,
};
use log::{debug, info, warn};
use serde_json::value::Value;
use crate::db::Post;
/// Initialise a connection to a Scuttlebutt server.
async fn init_sbot() -> Result<Sbot, String> {
// Define the port on which the go-sbot is running.
// Defaults to 8021 if the environment variable is not set.
let go_sbot_port = env::var("GO_SBOT_PORT").unwrap_or("8021".to_string());
let keystore = Keystore::GoSbot;
let ip_port = Some(format!("127.0.0.1:{}", go_sbot_port));
let net_id = None;
debug!("Initialising the sbot connection");
Sbot::init(keystore, ip_port, net_id)
.await
.map_err(|e| e.to_string())
}
/// Return the public key of the local sbot instance.
pub async fn whoami() -> Result<String, String> {
let mut sbot = init_sbot().await?;
info!("Executing `whoami` call");
sbot.whoami().await.map_err(|e| e.to_string())
}
/// Follow a peer.
pub async fn follow_peer(public_key: &str) -> Result<String, String> {
let mut sbot = init_sbot().await?;
sbot.follow(public_key).await.map_err(|e| e.to_string())
}
/// Check follow status.
///
/// Is peer A (`public_key_a`) following peer B (`public_key_b`)?
pub async fn is_following(public_key_a: &str, public_key_b: &str) -> Result<String, String> {
let mut sbot = init_sbot().await?;
let query = RelationshipQuery {
source: public_key_a.to_string(),
dest: public_key_b.to_string(),
};
sbot.friends_is_following(query)
.await
.map_err(|e| e.to_string())
}
/// Return a stream of messages authored by the given public key.
///
/// This returns all messages regardless of type.
pub async fn get_message_stream(
public_key: &str,
sequence_number: u64,
) -> impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>> {
let mut sbot = init_sbot().await.unwrap();
let history_stream_args = CreateHistoryStream::new(public_key.to_string())
.keys_values(true, true)
.after_seq(sequence_number);
sbot.create_history_stream(history_stream_args)
.await
.unwrap()
}
/// Return the name (self-identifier) for the peer associated with the given
/// public key.
///
/// The public key of the peer will be returned if a name is not found.
pub async fn get_name(public_key: &str) -> Result<String, String> {
let mut sbot = init_sbot().await?;
sbot.get_name(public_key).await.map_err(|e| e.to_string())
}
/// Filter a stream of messages and return a vector of root posts.
///
/// Each returned vector element includes the key of the post, the content
/// text, the date the post was published, the sequence number of the post
/// and whether it is read or unread.
pub async fn get_root_posts(
history_stream: impl futures::Stream<Item = Result<SsbMessageKVT, GolgiError>>,
) -> (u64, Vec<Post>) {
let mut latest_sequence = 0;
let mut posts = Vec::new();
futures::pin_mut!(history_stream);
while let Some(res) = history_stream.next().await {
match res {
Ok(msg) => {
if msg.value.is_message_type(SsbMessageContentType::Post) {
let content = msg.value.content.to_owned();
if let Value::Object(content_map) = content {
if !content_map.contains_key("root") {
let timestamp = msg.value.timestamp.round() as i64 / 1000;
let datetime = NaiveDateTime::from_timestamp(timestamp, 0);
let date = datetime.format("%d %b %Y").to_string();
let text = content_map.get_key_value("text").unwrap().1.to_string();
let subject = text.get(0..52).map(|s| s.to_string());
latest_sequence = msg.value.sequence;
posts.push(Post {
key: msg.key.to_owned(),
text,
timestamp,
date,
sequence: msg.value.sequence,
read: false,
subject,
})
}
}
}
}
Err(err) => {
// Print the `GolgiError` of this element to `stderr`.
warn!("err: {:?}", err);
}
}
}
(latest_sequence, posts)
}