Move shared libraries into workspace; Spin rsky-repo out into its own crate

Author: rudyfraser

.idea/rsky.iml

@@ -1,14 +1,28 @@
 [workspace]
-members = [ "rsky-common", "rsky-crypto","rsky-feedgen", "rsky-firehose", "rsky-identity", "rsky-labeler", "rsky-lexicon", "rsky-pds", "rsky-syntax", "rsky-jetstream-subscriber"]
+members = [ "rsky-common", "rsky-crypto","rsky-feedgen", "rsky-firehose", "rsky-identity", "rsky-labeler", "rsky-lexicon", "rsky-pds", "rsky-syntax", "rsky-jetstream-subscriber", "rsky-repo"]
 resolver = "2"
 
 [workspace.dependencies]
 cargo = { version = "0.84.0",features = ["vendored-openssl"] }
+serde = { version = "1.0.160", features = ["derive"] }
+serde_derive = "^1.0"
+serde_ipld_dagcbor = {  version = "0.6.1" ,features = ["codec"]}
+lexicon_cid = { package = "cid", version = "0.10.1", features = ["serde-codec"] }
+libipld = "0.16.0"
+serde_cbor = "0.11.2"
+serde_bytes = "0.11.15"
+tokio = { version = "1.28.2",features = ["full"] }
+sha2 = "0.11.0-pre.3"
+rand = "0.8.5"
+rand_core = "0.6.4"
+secp256k1 = { version = "0.28.2", features = ["global-context", "serde", "rand", "hashes","rand-std"] }
+serde_json = { version = "1.0.96",features = ["preserve_order"] }
 rsky-lexicon = {path = "rsky-lexicon", version = "0.2.3"}
 rsky-identity = {path = "rsky-identity", version = "0.1.0"}
 rsky-crypto = {path = "rsky-crypto", version = "0.1.1"}
 rsky-syntax = {path = "rsky-syntax", version = "0.1.0"}
-rsky-common = {path = "rsky-common", version = "0.1.0"}
+rsky-common = {path = "rsky-common", version = "0.1.1"}
+rsky-repo = {path = "rsky-repo", version = "0.0.1"}
 
 [profile.release]
 debug = 2  # Or any level from 0 to 2

Cargo.toml

@@ -37,12 +37,14 @@ rsky (/ˈrɪski/) is intended to be a full implementation of [AT Protocol](https
 | `rsky-lexicon`: schema definition language                 | [README](./rsky-lexicon/README.md)  | [![Crate](https://img.shields.io/crates/v/rsky-lexicon?logo=rust&style=flat-square&logoColor=E05D44&color=E05D44)](https://crates.io/crates/rsky-lexicon)   |
 | `rsky-syntax`: string parsers for identifiers              | [README](./rsky-syntax/README.md)   | [![Crate](https://img.shields.io/crates/v/rsky-syntax?logo=rust&style=flat-square&logoColor=E05D44&color=E05D44)](https://crates.io/crates/rsky-syntax)     |
 | `rsky-common`: shared code                                 | [README](./rsky-common/README.md)   | [![Crate](https://img.shields.io/crates/v/rsky-common?logo=rust&style=flat-square&logoColor=E05D44&color=E05D44)](https://crates.io/crates/rsky-common)     |
+| `rsky-repo`: data storage structure, including MST         | [README](./rsky-repo/README.md)     | [![Crate](https://img.shields.io/crates/v/rsky-repo?logo=rust&style=flat-square&logoColor=E05D44&color=E05D44)](https://crates.io/crates/rsky-repo)         |
 
 **Rust Services:**
 
 - `rsky-pds`: "Personal Data Server", hosting repo content for atproto accounts. It differs from the canonical Typescript implementation by using Postgres instead of SQLite, s3 compatible blob storage instead of on-disk, and mailgun for emailing. All to make the PDS easier to migrate between cloud hosting providers and more maintainable.
 - `rsky-feedgen`: Bluesky feed generator that closely follows the use cases of the Blacksky community.
 - `rsky-firehose`: Firehose consumer.
+- `rsky-jetstream-subscriber`: Firehose consumer for Jetstream.
 - `rsky-labeler`: Firehose consumer that labels content.
 
 ## About AT Protocol

README.md

@@ -1,6 +1,6 @@
 [package]
 name = "rsky-common"
-version = "0.1.0"
+version = "0.1.1"
 authors = ["Rudy Fraser <him@rudyfraser.com>"]
 description = "Shared code for rsky"
 license = "Apache-2.0"
@@ -13,6 +13,25 @@ documentation = "https://docs.rs/rsky-common"
 
 [dependencies]
 regex = "1.8.4"
+serde = { version = "1.0.217", features = ["derive"] }
+thiserror = "2.0.11"
+serde_ipld_dagcbor = { workspace = true }
+anyhow = "1.0.79"
+chrono = "0.4.39"
+rand = {workspace = true}
+rand_core = { workspace = true }
+url = "2.5.4"
+serde_json = "1.0.138"
+tracing = "0.1.41" # @TODO: Remove anyhow in lib
+rsky-identity = {workspace = true}
+base64ct = "1.6.0"
+urlencoding = "2.1.3"
+futures = "0.3.28"
+libipld = {workspace = true}
+indexmap = { version = "1.9.3",features = ["serde-1"] }
+secp256k1 = {workspace = true}
+sha2 = {workspace = true}
+lexicon_cid = {workspace = true}
 
 [dev-dependencies]
 temp-env = { version = "0.3.6"}

rsky-common/Cargo.toml

@@ -1,5 +1,5 @@
-use crate::common::time::SECOND;
-use crate::common::wait;
+use crate::time::SECOND;
+use crate::wait;
 use futures::stream::Stream;
 use futures::task::{Context, Poll};
 use std::cmp;

rsky-pds/src/common/async.rs rsky-common/src/async.rsrsky-pds/src/common/async.rs renamed to rsky-common/src/async.rs

@@ -1,4 +1,3 @@
-use crate::common;
 use anyhow::Result;
 use lexicon_cid::Cid;
 use libipld::cbor::DagCborCodec;
@@ -8,7 +7,7 @@ use libipld::raw::RawCodec;
 use serde::Serialize;
 
 pub fn cid_for_cbor<T: Serialize>(data: &T) -> Result<Cid> {
-    let bytes = common::struct_to_cbor(data)?;
+    let bytes = crate::struct_to_cbor(data)?;
     let cid = Cid::new_v1(
         u64::from(DagCborCodec),
         Code::Sha2_256.digest(bytes.as_slice()),

rsky-pds/src/common/ipld.rs rsky-common/src/ipld.rsrsky-pds/src/common/ipld.rs renamed to rsky-common/src/ipld.rs

@@ -1,2 +1,177 @@
+use anyhow::Result;
+use base64ct::{Base64, Encoding};
+use chrono::offset::Utc as UtcOffset;
+use chrono::DateTime;
+use rand::{distributions::Alphanumeric, Rng};
+use rsky_identity::did::atproto_data::VerificationMaterial;
+use rsky_identity::types::DidDocument;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::thread;
+use std::time::{Duration, SystemTime};
+use thiserror::Error;
+use url::Url;
+use urlencoding::encode;
+
+pub const RFC3339_VARIANT: &str = "%Y-%m-%dT%H:%M:%S%.3fZ";
+
+#[derive(Error, Debug)]
+pub enum BadContentTypeError {
+    #[error("BadType: `{0}`")]
+    BadType(String),
+    #[error("Content-Type header is missing")]
+    MissingType,
+}
+
+#[derive(Debug)]
+pub struct GetServiceEndpointOpts {
+    pub id: String,
+    pub r#type: Option<String>,
+}
+
+pub fn now() -> String {
+    let system_time = SystemTime::now();
+    let dt: DateTime<UtcOffset> = system_time.into();
+    format!("{}", dt.format(RFC3339_VARIANT))
+}
+
+pub fn wait(ms: u64) {
+    thread::sleep(Duration::from_millis(ms))
+}
+
+pub fn beginning_of_time() -> String {
+    let beginning_of_time = SystemTime::UNIX_EPOCH;
+    let dt: DateTime<UtcOffset> = beginning_of_time.into();
+    format!("{}", dt.format(RFC3339_VARIANT))
+}
+
+pub fn get_random_str() -> String {
+    let token: String = rand::thread_rng()
+        .sample_iter(&Alphanumeric)
+        .take(32)
+        .map(char::from)
+        .collect();
+    token
+}
+
+pub fn struct_to_cbor<T: Serialize>(obj: &T) -> Result<Vec<u8>> {
+    Ok(serde_ipld_dagcbor::to_vec(obj)?)
+}
+
+pub fn cbor_to_struct<T: DeserializeOwned>(bytes: Vec<u8>) -> Result<T> {
+    Ok(serde_ipld_dagcbor::from_slice::<T>(bytes.as_slice())?)
+}
+
+pub fn json_to_b64url<T: Serialize>(obj: &T) -> Result<String> {
+    Ok(Base64::encode_string((&serde_json::to_string(obj)?).as_ref()).replace("=", ""))
+}
+
+pub fn encode_uri_component(input: &String) -> String {
+    encode(input).to_string()
+}
+
+// based on did-doc.ts
+pub fn get_did(doc: &DidDocument) -> String {
+    doc.id.clone()
+}
+
+pub fn get_handle(doc: &DidDocument) -> Option<String> {
+    match &doc.also_known_as {
+        None => None,
+        Some(aka) => {
+            let found = aka.into_iter().find(|name| name.starts_with("at://"));
+            match found {
+                None => None,
+                // strip off at:// prefix
+                Some(found) => Some(found[5..].to_string()),
+            }
+        }
+    }
+}
+
+pub fn get_verification_material(
+    doc: &DidDocument,
+    key_id: &String,
+) -> Option<VerificationMaterial> {
+    let did = get_did(doc);
+    let keys = &doc.verification_method;
+    if let Some(keys) = keys {
+        let found = keys
+            .into_iter()
+            .find(|key| key.id == format!("#{key_id}") || key.id == format!("{did}#{key_id}"));
+        match found {
+            Some(found) if found.public_key_multibase.is_some() => {
+                let found = found.clone();
+                Some(VerificationMaterial {
+                    r#type: found.r#type,
+                    public_key_multibase: found.public_key_multibase.unwrap(),
+                })
+            }
+            _ => None,
+        }
+    } else {
+        None
+    }
+}
+
+pub fn get_notif_endpoint(doc: DidDocument) -> Option<String> {
+    get_service_endpoint(
+        doc,
+        GetServiceEndpointOpts {
+            id: "#bsky_notif".to_string(),
+            r#type: Some("BskyNotificationService".to_string()),
+        },
+    )
+}
+
+#[tracing::instrument(skip_all)]
+pub fn get_service_endpoint(doc: DidDocument, opts: GetServiceEndpointOpts) -> Option<String> {
+    tracing::info!(
+        "@LOG: common::get_service_endpoint() doc: {:?}; opts: {:?}",
+        doc,
+        opts
+    );
+    let did = get_did(&doc);
+    match doc.service {
+        None => None,
+        Some(services) => {
+            let found = services.iter().find(|service| {
+                service.id == opts.id || service.id == format!("{}{}", did, opts.id)
+            });
+            match found {
+                None => None,
+                Some(found) => match opts.r#type {
+                    None => None,
+                    Some(opts_type) if found.r#type == opts_type => {
+                        validate_url(&found.service_endpoint)
+                    }
+                    _ => None,
+                },
+            }
+        }
+    }
+}
+
+// Check protocol and hostname to prevent potential SSRF
+pub fn validate_url(url_str: &String) -> Option<String> {
+    match Url::parse(url_str) {
+        Err(_) => None,
+        Ok(url) => {
+            return if !vec!["http", "https"].contains(&url.scheme()) {
+                None
+            } else if url.host().is_none() {
+                None
+            } else {
+                Some(url_str.clone())
+            }
+        }
+    }
+}
+
+pub mod r#async;
 pub mod env;
 pub mod explicit_slurs;
+pub mod ipld;
+pub mod sign;
+pub mod tid;
+pub mod time;

rsky-common/src/lib.rs

@@ -28,7 +28,7 @@ pub fn dedash(str: String) -> String {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
-pub struct TID(pub(crate) String);
+pub struct TID(pub String);
 
 impl TID {
     pub fn new(str: String) -> Result<Self> {

rsky-pds/src/common/sign.rs rsky-common/src/sign.rsrsky-pds/src/common/sign.rs renamed to rsky-common/src/sign.rs

@@ -1,4 +1,4 @@
-use crate::common::RFC3339_VARIANT;
+use crate::RFC3339_VARIANT;
 use anyhow::Result;
 use chrono::offset::Utc as UtcOffset;
 use chrono::{DateTime, NaiveDateTime, Utc};