Merge branch 'cli-print-timeline'

5 files changed, 137 insertions, 2 deletions

diff --git a/cli/src/cli.rs b/cli/src/cli.rs
index 058b649..ee8d608 100644
--- a/cli/src/cli.rs
+++ b/cli/src/cli.rs
@@ -51,6 +51,20 @@ pub fn app<'a>() -> App<'a> {
                 )
             )
 
+            .subcommand(App::new("cat")
+                .author(crate_authors!())
+                .version(crate_version!())
+                .about("Read complete timeline of profile")
+
+                .arg(Arg::new("name")
+                    .long("name")
+                    .required(true)
+                    .takes_value(true)
+                    .value_name("NAME")
+                    .about("Name of the profile")
+                )
+            )
+
             .subcommand(App::new("post")
                 .author(crate_authors!())
                 .version(crate_version!())
@@ -84,8 +98,6 @@ pub fn app<'a>() -> App<'a> {
                     .args(&["text", "editor"])
                     .required(true) // one must be present
                 )
-
-
             )
         )
 
diff --git a/cli/src/profile.rs b/cli/src/profile.rs
index ca59c2b..a0025b6 100644
--- a/cli/src/profile.rs
+++ b/cli/src/profile.rs
@@ -8,12 +8,14 @@ use clap::ArgMatches;
 
 use distrox_lib::config::Config;
 use distrox_lib::profile::Profile;
+use distrox_lib::types::Payload;
 
 pub async fn profile(matches: &ArgMatches) -> Result<()> {
     match matches.subcommand() {
         Some(("create", m)) => profile_create(m).await,
         Some(("serve", m)) => profile_serve(m).await,
         Some(("post", m)) => profile_post(m).await,
+        Some(("cat", m)) => profile_cat(m).await,
         _ => unimplemented!(),
     }
 }
@@ -78,6 +80,7 @@ async fn profile_post(matches: &ArgMatches) -> Result<()> {
             unreachable!()
         }
     };
+
     let name = matches.value_of("name").map(String::from).unwrap(); // required
     let state_dir = Profile::state_dir_path(&name)?;
     log::info!("Creating '{}' in {}", name, state_dir.display());
@@ -95,3 +98,70 @@ async fn profile_post(matches: &ArgMatches) -> Result<()> {
     profile.exit().await
 }
 
+async fn profile_cat(matches: &ArgMatches) -> Result<()> {
+    use distrox_lib::stream::NodeStreamBuilder;
+    use futures::stream::StreamExt;
+
+    let name = matches.value_of("name").map(String::from).unwrap(); // required
+    let state_dir = Profile::state_dir_path(&name)?;
+    log::info!("Creating '{}' in {}", name, state_dir.display());
+
+    log::info!("Loading '{}' from {}", name, state_dir.display());
+    let profile = Profile::load(Config::default(), &name).await?;
+    log::info!("Profile loaded");
+    if let Some(head) = profile.head() {
+        log::info!("Profile HEAD = {:?}", head);
+        NodeStreamBuilder::starting_from(head.clone())
+            .into_stream(profile.client())
+            .then(|node| async {
+                match node {
+                    Err(e) => Err(e),
+                    Ok(node) => {
+                        profile.client()
+                            .get_payload(node.payload())
+                            .await
+                    }
+                }
+            })
+            .then(|payload| async {
+                match payload {
+                    Err(e) => Err(e),
+                    Ok(payload) => {
+                        profile.client()
+                            .get_content_text(payload.content())
+                            .await
+                            .map(|text| (payload, text))
+                    }
+                }
+            })
+            .then(|res| async {
+                use std::io::Write;
+                match res {
+                    Err(e) => {
+                        let out = std::io::stderr();
+                        let mut lock = out.lock();
+                        writeln!(lock, "Error: {:?}", e)?;
+                    }
+                    Ok((payload, text)) => {
+                        let out = std::io::stdout();
+                        let mut lock = out.lock();
+                        writeln!(lock, "{time} - {cid}",
+                            time = payload.timestamp().inner(),
+                            cid = payload.content())?;
+
+                        writeln!(lock, "{text}", text = text)?;
+                        writeln!(lock, "")?;
+                    },
+                }
+                Ok(())
+            })
+            .collect::<Vec<Result<()>>>()
+            .await
+            .into_iter()
+            .collect::<Result<()>>()?;
+    } else {
+        eprintln!("Profile has no posts");
+    }
+
+    Ok(())
+}
diff --git a/lib/src/lib.rs b/lib/src/lib.rs
index 759ea14..0cccef9 100644
--- a/lib/src/lib.rs
+++ b/lib/src/lib.rs
@@ -3,4 +3,5 @@ pub mod config;
 pub mod consts;
 pub mod ipfs_client;
 pub mod profile;
+pub mod stream;
 pub mod types;
diff --git a/lib/src/stream.rs b/lib/src/stream.rs
new file mode 100644
index 0000000..01548e6
--- /dev/null
+++ b/lib/src/stream.rs
@@ -0,0 +1,46 @@
+use anyhow::Result;
+
+use crate::client::Client;
+use crate::types::Node;
+
+#[derive(Debug)]
+pub struct NodeStreamBuilder {
+    state: Vec<cid::Cid>
+}
+
+impl NodeStreamBuilder {
+    pub fn starting_from(node_cid: cid::Cid) -> Self {
+        Self {
+            state: vec![node_cid]
+        }
+    }
+
+    pub fn into_stream<'a>(self, client: &'a Client) -> impl futures::stream::Stream<Item = Result<Node>> + 'a {
+        futures::stream::unfold((client, self.state), move |(client, mut state)| {
+                async move {
+                    if let Some(node_cid) = state.pop() {
+                        match client
+                            .get_node(node_cid)
+                            .await
+                            .map(move |node| {
+                                node.parents().iter().for_each(|parent| {
+                                    state.push(parent.clone())
+                                });
+
+                                (node, state)
+                            })
+                            .map(Some)
+                            .transpose()
+                        {
+                            Some(Ok((item, state))) => Some((Ok(item), (client, state))),
+                            Some(Err(e)) => Some((Err(e), (client, vec![]))),
+                            None => None,
+                        }
+                    } else {
+                        None
+                    }
+                }
+            })
+    }
+
+}
diff --git a/lib/src/types/datetime.rs b/lib/src/types/datetime.rs
index 00d739a..70098d7 100644
--- a/lib/src/types/datetime.rs
+++ b/lib/src/types/datetime.rs
@@ -33,3 +33,9 @@ impl From<chrono::DateTime<chrono::Utc>> for DateTime {
     }
 }
 
+impl DateTime {
+    pub fn inner(&self) -> &chrono::DateTime<chrono::Utc> {
+        &self.0
+    }
+}
+