refactor(iroh): extract docs RPC into iroh-docs (#2868)

Depends on n0-computer/iroh-docs#5

---------

Co-authored-by: Diva M <divma@protonmail.com>
Co-authored-by: Ruediger Klaehn <rklaehn@protonmail.com>

iroh-cli/Cargo.toml

@@ -42,6 +42,7 @@ human-time = "0.1.6"
 indicatif = { version = "0.17", features = ["tokio"] }
 iroh = { version = "0.28.1", path = "../iroh", features = ["metrics"] }
 iroh-gossip = "0.28.1"
+iroh-docs = { version = "0.28.0", features = ["rpc"]}
 iroh-metrics = { version = "0.28.0" }
 parking_lot = "0.12.1"
 pkarr = { version = "2.2.0", default-features = false }

iroh-cli/src/commands/docs.rs

@@ -18,17 +18,17 @@ use indicatif::{HumanBytes, HumanDuration, MultiProgress, ProgressBar, ProgressS
 use iroh::{
     base::{base32::fmt_short, node_addr::AddrInfoOptions},
     blobs::{provider::AddProgress, util::SetTagOption, Hash, Tag},
-    client::{
-        blobs::WrapOption,
-        docs::{Doc, Entry, LiveEvent, Origin, ShareMode},
-        Iroh,
-    },
+    client::{blobs::WrapOption, Doc, Iroh},
     docs::{
         store::{DownloadPolicy, FilterKind, Query, SortDirection},
         AuthorId, DocTicket, NamespaceId,
     },
     util::fs::{path_content_info, path_to_key, PathContent},
 };
+use iroh_docs::{
+    engine::Origin,
+    rpc::client::docs::{Entry, LiveEvent, ShareMode},
+};
 use tokio::io::AsyncReadExt;
 
 use crate::config::ConsoleEnv;
@@ -414,7 +414,7 @@ impl DocCommands {
 
                 let mut stream = doc.get_many(query).await?;
                 while let Some(entry) = stream.try_next().await? {
-                    println!("{}", fmt_entry(&doc, &entry, mode).await);
+                    println!("{}", fmt_entry(&iroh.blobs(), &entry, mode).await);
                 }
             }
             Self::Keys {
@@ -440,7 +440,7 @@ impl DocCommands {
                 query = query.sort_by(sort.into(), direction);
                 let mut stream = doc.get_many(query).await?;
                 while let Some(entry) = stream.try_next().await? {
-                    println!("{}", fmt_entry(&doc, &entry, mode).await);
+                    println!("{}", fmt_entry(&iroh.blobs(), &entry, mode).await);
                 }
             }
             Self::Leave { doc } => {
@@ -516,7 +516,7 @@ impl DocCommands {
                     }
                     Some(e) => e,
                 };
-                match entry.content_reader(&doc).await {
+                match iroh.blobs().read(entry.content_hash()).await {
                     Ok(mut content) => {
                         if let Some(dir) = path.parent() {
                             if let Err(err) = std::fs::create_dir_all(dir) {
@@ -547,13 +547,14 @@ impl DocCommands {
             Self::Watch { doc } => {
                 let doc = get_doc(iroh, env, doc).await?;
                 let mut stream = doc.subscribe().await?;
+                let blobs = iroh.blobs();
                 while let Some(event) = stream.next().await {
                     let event = event?;
                     match event {
                         LiveEvent::InsertLocal { entry } => {
                             println!(
                                 "local change:  {}",
-                                fmt_entry(&doc, &entry, DisplayContentMode::Auto).await
+                                fmt_entry(&blobs, &entry, DisplayContentMode::Auto).await
                             )
                         }
                         LiveEvent::InsertRemote {
@@ -563,17 +564,17 @@ impl DocCommands {
                         } => {
                             let content = match content_status {
                                 iroh::docs::ContentStatus::Complete => {
-                                    fmt_entry(&doc, &entry, DisplayContentMode::Auto).await
+                                    fmt_entry(&blobs, &entry, DisplayContentMode::Auto).await
                                 }
                                 iroh::docs::ContentStatus::Incomplete => {
                                     let (Ok(content) | Err(content)) =
-                                        fmt_content(&doc, &entry, DisplayContentMode::ShortHash)
+                                        fmt_content(&blobs, &entry, DisplayContentMode::ShortHash)
                                             .await;
                                     format!("<incomplete: {} ({})>", content, human_len(&entry))
                                 }
                                 iroh::docs::ContentStatus::Missing => {
                                     let (Ok(content) | Err(content)) =
-                                        fmt_content(&doc, &entry, DisplayContentMode::ShortHash)
+                                        fmt_content(&blobs, &entry, DisplayContentMode::ShortHash)
                                             .await;
                                     format!("<missing: {} ({})>", content, human_len(&entry))
                                 }
@@ -679,14 +680,19 @@ impl DocCommands {
 
 /// Gets the document given the client, the environment (and maybe the [`NamespaceID`]).
 async fn get_doc(iroh: &Iroh, env: &ConsoleEnv, id: Option<NamespaceId>) -> anyhow::Result<Doc> {
+    let doc_id = env.doc(id)?;
     iroh.docs()
-        .open(env.doc(id)?)
+        .open(doc_id)
         .await?
         .context("Document not found")
 }
 
 /// Formats the content. If an error occurs it's returned in a formatted, friendly way.
-async fn fmt_content(doc: &Doc, entry: &Entry, mode: DisplayContentMode) -> Result<String, String> {
+async fn fmt_content(
+    blobs: &iroh::client::blobs::Client,
+    entry: &Entry,
+    mode: DisplayContentMode,
+) -> Result<String, String> {
     let read_failed = |err: anyhow::Error| format!("<failed to get content: {err}>");
     let encode_hex = |err: std::string::FromUtf8Error| format!("0x{}", hex::encode(err.as_bytes()));
     let as_utf8 = |buf: Vec<u8>| String::from_utf8(buf).map(|repr| format!("\"{repr}\""));
@@ -695,11 +701,17 @@ async fn fmt_content(doc: &Doc, entry: &Entry, mode: DisplayContentMode) -> Resu
         DisplayContentMode::Auto => {
             if entry.content_len() < MAX_DISPLAY_CONTENT_LEN {
                 // small content: read fully as UTF-8
-                let bytes = entry.content_bytes(doc).await.map_err(read_failed)?;
+                let bytes = blobs
+                    .read_to_bytes(entry.content_hash())
+                    .await
+                    .map_err(read_failed)?;
                 Ok(as_utf8(bytes.into()).unwrap_or_else(encode_hex))
             } else {
                 // large content: read just the first part as UTF-8
-                let mut blob_reader = entry.content_reader(doc).await.map_err(read_failed)?;
+                let mut blob_reader = blobs
+                    .read(entry.content_hash())
+                    .await
+                    .map_err(read_failed)?;
                 let mut buf = Vec::with_capacity(MAX_DISPLAY_CONTENT_LEN as usize + 5);
 
                 blob_reader
@@ -714,7 +726,10 @@ async fn fmt_content(doc: &Doc, entry: &Entry, mode: DisplayContentMode) -> Resu
         }
         DisplayContentMode::Content => {
             // read fully as UTF-8
-            let bytes = entry.content_bytes(doc).await.map_err(read_failed)?;
+            let bytes = blobs
+                .read_to_bytes(entry.content_hash())
+                .await
+                .map_err(read_failed)?;
             Ok(as_utf8(bytes.into()).unwrap_or_else(encode_hex))
         }
         DisplayContentMode::ShortHash => {
@@ -735,12 +750,16 @@ fn human_len(entry: &Entry) -> HumanBytes {
 
 /// Formats an entry for display as a `String`.
 #[must_use = "this won't be printed, you need to print it yourself"]
-async fn fmt_entry(doc: &Doc, entry: &Entry, mode: DisplayContentMode) -> String {
+async fn fmt_entry(
+    blobs: &iroh::client::blobs::Client,
+    entry: &Entry,
+    mode: DisplayContentMode,
+) -> String {
     let key = std::str::from_utf8(entry.key())
         .unwrap_or("<bad key>")
         .bold();
     let author = fmt_short(entry.author());
-    let (Ok(content) | Err(content)) = fmt_content(doc, entry, mode).await;
+    let (Ok(content) | Err(content)) = fmt_content(blobs, entry, mode).await;
     let len = human_len(entry);
     format!("@{author}: {key} = {content} ({len})")
 }

iroh/Cargo.toml

@@ -36,7 +36,7 @@ iroh-router = { version = "0.28.0" }
 nested_enum_utils = "0.1.0"
 num_cpus = { version = "1.15.0" }
 portable-atomic = "1"
-iroh-docs = { version = "0.28.0" }
+iroh-docs = { version = "0.28.0", features = ["rpc"] }
 iroh-gossip = "0.28.1"
 parking_lot = "0.12.1"
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }

iroh/examples/client.rs

@@ -16,6 +16,7 @@ async fn main() -> anyhow::Result<()> {
     // Could also use `node` directly, as it derefs to the client.
     let client = node.client();
 
+    let blobs = client.blobs();
     let doc = client.docs().create().await?;
     let author = client.authors().default().await?;
 
@@ -24,8 +25,7 @@ async fn main() -> anyhow::Result<()> {
     let mut stream = doc.get_many(Query::all()).await?;
     while let Some(entry) = stream.try_next().await? {
         println!("entry {}", fmt_entry(&entry));
-        // You can pass either `&doc` or the `client`.
-        let content = entry.content_bytes(&doc).await?;
+        let content = blobs.read_to_bytes(entry.content_hash()).await?;
         println!("  content {}", std::str::from_utf8(&content)?)
     }
 

iroh/src/client.rs

@@ -14,13 +14,12 @@ pub use crate::rpc_protocol::RpcService;
 
 mod quic;
 
-pub(crate) use self::quic::{connect_raw as quic_connect_raw, RPC_ALPN};
-pub use self::{docs::Doc, net::NodeStatus};
-
-pub mod authors;
 pub use iroh_blobs::rpc::client::{blobs, tags};
+pub use iroh_docs::rpc::client::{authors, docs, docs::Doc};
 pub use iroh_gossip::rpc::client as gossip;
-pub mod docs;
+
+pub use self::net::NodeStatus;
+pub(crate) use self::quic::{connect_raw as quic_connect_raw, RPC_ALPN};
 pub mod net;
 
 // Keep this type exposed, otherwise every occurrence of `RpcClient` in the API
@@ -61,13 +60,13 @@ impl Iroh {
     }
 
     /// Returns the docs client.
-    pub fn docs(&self) -> &docs::Client {
-        docs::Client::ref_cast(&self.rpc)
+    pub fn docs(&self) -> iroh_docs::rpc::client::docs::Client {
+        iroh_docs::rpc::client::docs::Client::new(self.rpc.clone().map().boxed())
     }
 
-    /// Returns the authors client.
-    pub fn authors(&self) -> &authors::Client {
-        authors::Client::ref_cast(&self.rpc)
+    /// Returns the docs client.
+    pub fn authors(&self) -> iroh_docs::rpc::client::authors::Client {
+        iroh_docs::rpc::client::authors::Client::new(self.rpc.clone().map().boxed())
     }
 
     /// Returns the tags client.