make plume-models async (again)

plume-models/src/comments.rs

@@ -17,6 +17,7 @@ use activitypub::{
 };
 use chrono::{self, NaiveDateTime};
 use diesel::{self, ExpressionMethods, QueryDsl, RunQueryDsl, SaveChangesDsl};
+use futures::stream::{self, StreamExt};
 use plume_common::{
     activity_pub::{
         inbox::{AsActor, AsObject, FromId},
@@ -105,7 +106,7 @@ impl Comment {
                 .unwrap_or(false)
     }
 
-    pub fn to_activity(&self, c: &PlumeRocket) -> Result<Note> {
+    pub async fn to_activity(&self, c: &PlumeRocket) -> Result<Note> {
         let author = User::get(&c.conn, self.author_id)?;
         let (html, mentions, _hashtags) = utils::md_to_html(
             self.content.get().as_ref(),
@@ -132,18 +133,18 @@ impl Comment {
         note.object_props.set_attributed_to_link(author.into_id())?;
         note.object_props.set_to_link_vec(to)?;
         note.object_props.set_tag_link_vec(
-            mentions
-                .into_iter()
-                .filter_map(|m| Mention::build_activity(c, &m).ok())
-                .collect::<Vec<link::Mention>>(),
+            stream::iter(mentions)
+                .filter_map(|m| async move { Mention::build_activity(c, &m).await.ok() })
+                .collect::<Vec<link::Mention>>()
+                .await,
         )?;
         Ok(note)
     }
 
-    pub fn create_activity(&self, c: &PlumeRocket) -> Result<Create> {
+    pub async fn create_activity(&self, c: &PlumeRocket) -> Result<Create> {
         let author = User::get(&c.conn, self.author_id)?;
 
-        let note = self.to_activity(c)?;
+        let note = self.to_activity(c).await?;
         let mut act = Create::default();
         act.create_props.set_actor_link(author.into_id())?;
         act.create_props.set_object_object(note.clone())?;

plume-models/src/mentions.rs

@@ -52,8 +52,8 @@ impl Mention {
         }
     }
 
-    pub fn build_activity(c: &PlumeRocket, ment: &str) -> Result<link::Mention> {
-        let user = User::find_by_fqn(c, ment)?;
+    pub async fn build_activity(c: &PlumeRocket, ment: &str) -> Result<link::Mention> {
+        let user = User::find_by_fqn(c, ment).await?;
         let mut mention = link::Mention::default();
         mention.link_props.set_href_string(user.ap_url)?;
         mention.link_props.set_name_string(format!("@{}", ment))?;

plume-models/src/timeline/query.rs

@@ -311,10 +311,19 @@ impl WithList {
                     })
                 }
                 WithList::Author { boosts, likes } => match kind {
-                    Kind::Original => Ok(list
-                        .iter()
-                        .filter_map(|a| User::find_by_fqn(rocket, a).ok())
-                        .any(|a| post.is_author(&rocket.conn, a.id).unwrap_or(false))),
+                    Kind::Original => {
+                        let mut rt = Runtime::new().unwrap();
+                        rt.block_on(async move {
+                            Ok(stream::iter(list)
+                                .filter_map(|a| async move {
+                                    Some(User::find_by_fqn(rocket, a).await.ok().unwrap())
+                                })
+                                .collect::<Vec<_>>()
+                                .await
+                                .into_iter()
+                                .any(|a| post.is_author(&rocket.conn, a.id).unwrap_or(false)))
+                        })
+                    }
                     Kind::Reshare(u) => {
                         if *boosts {
                             Ok(list.iter().any(|user| &u.fqn == user))

plume-models/src/users.rs

@@ -190,29 +190,31 @@ impl User {
             .map_err(Error::from)
     }
 
-    pub fn find_by_fqn(c: &PlumeRocket, fqn: &str) -> Result<User> {
+    pub async fn find_by_fqn(c: &PlumeRocket, fqn: &str) -> Result<User> {
         let from_db = users::table
             .filter(users::fqn.eq(fqn))
             .first(&*c.conn)
             .optional()?;
         if let Some(from_db) = from_db {
             Ok(from_db)
         } else {
-            User::fetch_from_webfinger(c, fqn)
+            User::fetch_from_webfinger(c, fqn).await
         }
     }
 
-    fn fetch_from_webfinger(c: &PlumeRocket, acct: &str) -> Result<User> {
-        let link = resolve(acct.to_owned(), true)?
+    async fn fetch_from_webfinger(c: &PlumeRocket, acct: &str) -> Result<User> {
+        let link = resolve(acct.to_owned(), true)
+            .await?
             .links
             .into_iter()
             .find(|l| l.mime_type == Some(String::from("application/activity+json")))
             .ok_or(Error::Webfinger)?;
         User::from_id(c, link.href.as_ref()?, None).map_err(|(_, e)| e)
     }
 
-    pub fn fetch_remote_interact_uri(acct: &str) -> Result<String> {
-        resolve(acct.to_owned(), true)?
+    pub async fn fetch_remote_interact_uri(acct: &str) -> Result<String> {
+        resolve(acct.to_owned(), true)
+            .await?
             .links
             .into_iter()
             .find(|l| l.rel == "http://ostatus.org/schema/1.0/subscribe")