Federation async (#848)

* Asyncify more

* I guess these changed

* Clean PR a bit

* Convert more away from failure error

* config changes for testing federation

* It was DNS

So actix-web's client relies on TRust DNS Resolver to figure out
where to send data, but TRust DNS Resolver seems to not play nice
with docker, which expressed itself as not resolving the name to
an IP address _the first time_ when making a request. The fix was
literally to make the request again (which I limited to 3 times
total, and not exceeding the request timeout in total)

* Only retry for connecterror

Since TRust DNS Resolver was causing ConnectError::Timeout,
this change limits the retry to only this error, returning
immediately for any other error

* Use http sig norm 0.4.0-alpha for actix-web 3.0 support

* Blocking function, retry http requests

* cargo +nightly fmt

* Only create one pictrs dir

* Don't yarn build

* cargo +nightly fmt

1 files changed, 229 insertions, 158 deletions

diff --git a/server/src/api/community.rs b/server/src/api/community.rs
index 3fc67eb3..02071c57 100644
--- a/server/src/api/community.rs
+++ b/server/src/api/community.rs
@@ -7,6 +7,7 @@ use crate::{
     ActorType,
     EndpointType,
   },
+  blocking,
   db::{Bannable, Crud, Followable, Joinable, SortType},
   is_valid_community_name,
   naive_from_unix,
@@ -18,12 +19,9 @@ use crate::{
     UserOperation,
     WebsocketInfo,
   },
+  DbPool,
+  LemmyError,
 };
-use diesel::{
-  r2d2::{ConnectionManager, Pool},
-  PgConnection,
-};
-use failure::Error;
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
 
@@ -138,14 +136,15 @@ pub struct TransferCommunity {
   auth: String,
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<GetCommunity> {
   type Response = GetCommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     websocket_info: Option<WebsocketInfo>,
-  ) -> Result<GetCommunityResponse, Error> {
+  ) -> Result<GetCommunityResponse, LemmyError> {
     let data: &GetCommunity = &self.data;
 
     let user_id: Option<i32> = match &data.auth {
@@ -159,33 +158,38 @@ impl Perform for Oper<GetCommunity> {
       None => None,
     };
 
-    let conn = pool.get()?;
-
+    let name = data.name.to_owned().unwrap_or_else(|| "main".to_string());
     let community = match data.id {
-      Some(id) => Community::read(&conn, id)?,
-      None => {
-        match Community::read_from_name(
-          &conn,
-          &data.name.to_owned().unwrap_or_else(|| "main".to_string()),
-        ) {
-          Ok(community) => community,
-          Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
-        }
-      }
+      Some(id) => blocking(pool, move |conn| Community::read(conn, id)).await??,
+      None => match blocking(pool, move |conn| Community::read_from_name(conn, &name)).await? {
+        Ok(community) => community,
+        Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
+      },
     };
 
-    let community_view = match CommunityView::read(&conn, community.id, user_id) {
+    let community_id = community.id;
+    let community_view = match blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, user_id)
+    })
+    .await?
+    {
       Ok(community) => community,
       Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
     };
 
-    let moderators = match CommunityModeratorView::for_community(&conn, community.id) {
+    let community_id = community.id;
+    let moderators: Vec<CommunityModeratorView> = match blocking(pool, move |conn| {
+      CommunityModeratorView::for_community(conn, community_id)
+    })
+    .await?
+    {
       Ok(moderators) => moderators,
       Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
     };
 
-    let site_creator_id = Site::read(&conn, 1)?.creator_id;
-    let mut admins = UserView::admins(&conn)?;
+    let site = blocking(pool, move |conn| Site::read(conn, 1)).await??;
+    let site_creator_id = site.creator_id;
+    let mut admins = blocking(pool, move |conn| UserView::admins(conn)).await??;
     let creator_index = admins.iter().position(|r| r.id == site_creator_id).unwrap();
     let creator_user = admins.remove(creator_index);
     admins.insert(0, creator_user);
@@ -220,14 +224,15 @@ impl Perform for Oper<GetCommunity> {
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<CreateCommunity> {
   type Response = CommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     _websocket_info: Option<WebsocketInfo>,
-  ) -> Result<CommunityResponse, Error> {
+  ) -> Result<CommunityResponse, LemmyError> {
     let data: &CreateCommunity = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -255,10 +260,9 @@ impl Perform for Oper<CreateCommunity> {
 
     let user_id = claims.id;
 
-    let conn = pool.get()?;
-
     // Check for a site ban
-    if UserView::read(&conn, user_id)?.banned {
+    let user_view = blocking(pool, move |conn| UserView::read(conn, user_id)).await??;
+    if user_view.banned {
       return Err(APIError::err("site_ban").into());
     }
 
@@ -283,34 +287,36 @@ impl Perform for Oper<CreateCommunity> {
       published: None,
     };
 
-    let inserted_community = match Community::create(&conn, &community_form) {
-      Ok(community) => community,
-      Err(_e) => return Err(APIError::err("community_already_exists").into()),
-    };
+    let inserted_community =
+      match blocking(pool, move |conn| Community::create(conn, &community_form)).await? {
+        Ok(community) => community,
+        Err(_e) => return Err(APIError::err("community_already_exists").into()),
+      };
 
     let community_moderator_form = CommunityModeratorForm {
       community_id: inserted_community.id,
       user_id,
     };
 
-    let _inserted_community_moderator =
-      match CommunityModerator::join(&conn, &community_moderator_form) {
-        Ok(user) => user,
-        Err(_e) => return Err(APIError::err("community_moderator_already_exists").into()),
-      };
+    let join = move |conn: &'_ _| CommunityModerator::join(conn, &community_moderator_form);
+    if blocking(pool, join).await?.is_err() {
+      return Err(APIError::err("community_moderator_already_exists").into());
+    }
 
     let community_follower_form = CommunityFollowerForm {
       community_id: inserted_community.id,
       user_id,
     };
 
-    let _inserted_community_follower =
-      match CommunityFollower::follow(&conn, &community_follower_form) {
-        Ok(user) => user,
-        Err(_e) => return Err(APIError::err("community_follower_already_exists").into()),
-      };
+    let follow = move |conn: &'_ _| CommunityFollower::follow(conn, &community_follower_form);
+    if blocking(pool, follow).await?.is_err() {
+      return Err(APIError::err("community_follower_already_exists").into());
+    }
 
-    let community_view = CommunityView::read(&conn, inserted_community.id, Some(user_id))?;
+    let community_view = blocking(pool, move |conn| {
+      CommunityView::read(conn, inserted_community.id, Some(user_id))
+    })
+    .await??;
 
     Ok(CommunityResponse {
       community: community_view,
@@ -318,14 +324,15 @@ impl Perform for Oper<CreateCommunity> {
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<EditCommunity> {
   type Response = CommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     websocket_info: Option<WebsocketInfo>,
-  ) -> Result<CommunityResponse, Error> {
+  ) -> Result<CommunityResponse, LemmyError> {
     let data: &EditCommunity = &self.data;
 
     if let Err(slurs) = slur_check(&data.name) {
@@ -353,28 +360,34 @@ impl Perform for Oper<EditCommunity> {
 
     let user_id = claims.id;
 
-    let conn = pool.get()?;
-
     // Check for a site ban
-    let user = User_::read(&conn, user_id)?;
+    let user = blocking(pool, move |conn| User_::read(conn, user_id)).await??;
     if user.banned {
       return Err(APIError::err("site_ban").into());
     }
 
     // Verify its a mod
+    let edit_id = data.edit_id;
     let mut editors: Vec<i32> = Vec::new();
     editors.append(
-      &mut CommunityModeratorView::for_community(&conn, data.edit_id)?
-        .into_iter()
-        .map(|m| m.user_id)
-        .collect(),
+      &mut blocking(pool, move |conn| {
+        CommunityModeratorView::for_community(conn, edit_id)
+          .map(|v| v.into_iter().map(|m| m.user_id).collect())
+      })
+      .await??,
+    );
+    editors.append(
+      &mut blocking(pool, move |conn| {
+        UserView::admins(conn).map(|v| v.into_iter().map(|a| a.id).collect())
+      })
+      .await??,
     );
-    editors.append(&mut UserView::admins(&conn)?.into_iter().map(|a| a.id).collect());
     if !editors.contains(&user_id) {
       return Err(APIError::err("no_community_edit_allowed").into());
     }
 
-    let read_community = Community::read(&conn, data.edit_id)?;
+    let edit_id = data.edit_id;
+    let read_community = blocking(pool, move |conn| Community::read(conn, edit_id)).await??;
 
     let community_form = CommunityForm {
       name: data.name.to_owned(),
@@ -394,7 +407,12 @@ impl Perform for Oper<EditCommunity> {
       published: None,
     };
 
-    let updated_community = match Community::update(&conn, data.edit_id, &community_form) {
+    let edit_id = data.edit_id;
+    let updated_community = match blocking(pool, move |conn| {
+      Community::update(conn, edit_id, &community_form)
+    })
+    .await?
+    {
       Ok(community) => community,
       Err(_e) => return Err(APIError::err("couldnt_update_community").into()),
     };
@@ -412,24 +430,36 @@ impl Perform for Oper<EditCommunity> {
         reason: data.reason.to_owned(),
         expires,
       };
-      ModRemoveCommunity::create(&conn, &form)?;
+      blocking(pool, move |conn| ModRemoveCommunity::create(conn, &form)).await??;
     }
 
     if let Some(deleted) = data.deleted.to_owned() {
       if deleted {
-        updated_community.send_delete(&user, &conn)?;
+        updated_community
+          .send_delete(&user, &self.client, pool)
+          .await?;
       } else {
-        updated_community.send_undo_delete(&user, &conn)?;
+        updated_community
+          .send_undo_delete(&user, &self.client, pool)
+          .await?;
       }
     } else if let Some(removed) = data.removed.to_owned() {
       if removed {
-        updated_community.send_remove(&user, &conn)?;
+        updated_community
+          .send_remove(&user, &self.client, pool)
+          .await?;
       } else {
-        updated_community.send_undo_remove(&user, &conn)?;
+        updated_community
+          .send_undo_remove(&user, &self.client, pool)
+          .await?;
       }
     }
 
-    let community_view = CommunityView::read(&conn, data.edit_id, Some(user_id))?;
+    let edit_id = data.edit_id;
+    let community_view = blocking(pool, move |conn| {
+      CommunityView::read(conn, edit_id, Some(user_id))
+    })
+    .await??;
 
     let res = CommunityResponse {
       community: community_view,
@@ -453,14 +483,15 @@ impl Perform for Oper<EditCommunity> {
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<ListCommunities> {
   type Response = ListCommunitiesResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     _websocket_info: Option<WebsocketInfo>,
-  ) -> Result<ListCommunitiesResponse, Error> {
+  ) -> Result<ListCommunitiesResponse, LemmyError> {
     let data: &ListCommunities = &self.data;
 
     let user_claims: Option<Claims> = match &data.auth {
@@ -483,29 +514,33 @@ impl Perform for Oper<ListCommunities> {
 
     let sort = SortType::from_str(&data.sort)?;
 
-    let conn = pool.get()?;
-
-    let communities = CommunityQueryBuilder::create(&conn)
-      .sort(&sort)
-      .for_user(user_id)
-      .show_nsfw(show_nsfw)
-      .page(data.page)
-      .limit(data.limit)
-      .list()?;
+    let page = data.page;
+    let limit = data.limit;
+    let communities = blocking(pool, move |conn| {
+      CommunityQueryBuilder::create(conn)
+        .sort(&sort)
+        .for_user(user_id)
+        .show_nsfw(show_nsfw)
+        .page(page)
+        .limit(limit)
+        .list()
+    })
+    .await??;
 
     // Return the jwt
     Ok(ListCommunitiesResponse { communities })
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<FollowCommunity> {
   type Response = CommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     _websocket_info: Option<WebsocketInfo>,
-  ) -> Result<CommunityResponse, Error> {
+  ) -> Result<CommunityResponse, LemmyError> {
     let data: &FollowCommunity = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -515,9 +550,8 @@ impl Perform for Oper<FollowCommunity> {
 
     let user_id = claims.id;
 
-    let conn = pool.get()?;
-
-    let community = Community::read(&conn, data.community_id)?;
+    let community_id = data.community_id;
+    let community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
     let community_follower_form = CommunityFollowerForm {
       community_id: data.community_id,
       user_id,
@@ -525,34 +559,44 @@ impl Perform for Oper<FollowCommunity> {
 
     if community.local {
       if data.follow {
-        match CommunityFollower::follow(&conn, &community_follower_form) {
-          Ok(user) => user,
-          Err(_e) => return Err(APIError::err("community_follower_already_exists").into()),
-        };
+        let follow = move |conn: &'_ _| CommunityFollower::follow(conn, &community_follower_form);
+        if blocking(pool, follow).await?.is_err() {
+          return Err(APIError::err("community_follower_already_exists").into());
+        }
       } else {
-        match CommunityFollower::unfollow(&conn, &community_follower_form) {
-          Ok(user) => user,
-          Err(_e) => return Err(APIError::err("community_follower_already_exists").into()),
-        };
+        let unfollow =
+          move |conn: &'_ _| CommunityFollower::unfollow(conn, &community_follower_form);
+        if blocking(pool, unfollow).await?.is_err() {
+          return Err(APIError::err("community_follower_already_exists").into());
+        }
       }
     } else {
-      let user = User_::read(&conn, user_id)?;
+      let user = blocking(pool, move |conn| User_::read(conn, user_id)).await??;
 
       if data.follow {
         // Dont actually add to the community followers here, because you need
         // to wait for the accept
-        user.send_follow(&community.actor_id, &conn)?;
+        user
+          .send_follow(&community.actor_id, &self.client, pool)
+          .await?;
       } else {
-        user.send_unfollow(&community.actor_id, &conn)?;
-        match CommunityFollower::unfollow(&conn, &community_follower_form) {
-          Ok(user) => user,
-          Err(_e) => return Err(APIError::err("community_follower_already_exists").into()),
-        };
+        user
+          .send_unfollow(&community.actor_id, &self.client, pool)
+          .await?;
+        let unfollow =
+          move |conn: &'_ _| CommunityFollower::unfollow(conn, &community_follower_form);
+        if blocking(pool, unfollow).await?.is_err() {
+          return Err(APIError::err("community_follower_already_exists").into());
+        }
       }
       // TODO: this needs to return a "pending" state, until Accept is received from the remote server
     }
 
-    let community_view = CommunityView::read(&conn, data.community_id, Some(user_id))?;
+    let community_id = data.community_id;
+    let community_view = blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, Some(user_id))
+    })
+    .await??;
 
     Ok(CommunityResponse {
       community: community_view,
@@ -560,14 +604,15 @@ impl Perform for Oper<FollowCommunity> {
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<GetFollowedCommunities> {
   type Response = GetFollowedCommunitiesResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     _websocket_info: Option<WebsocketInfo>,
-  ) -> Result<GetFollowedCommunitiesResponse, Error> {
+  ) -> Result<GetFollowedCommunitiesResponse, LemmyError> {
     let data: &GetFollowedCommunities = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -577,27 +622,29 @@ impl Perform for Oper<GetFollowedCommunities> {
 
     let user_id = claims.id;
 
-    let conn = pool.get()?;
-
-    let communities: Vec<CommunityFollowerView> =
-      match CommunityFollowerView::for_user(&conn, user_id) {
-        Ok(communities) => communities,
-        Err(_e) => return Err(APIError::err("system_err_login").into()),
-      };
+    let communities = match blocking(pool, move |conn| {
+      CommunityFollowerView::for_user(conn, user_id)
+    })
+    .await?
+    {
+      Ok(communities) => communities,
+      _ => return Err(APIError::err("system_err_login").into()),
+    };
 
     // Return the jwt
     Ok(GetFollowedCommunitiesResponse { communities })
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<BanFromCommunity> {
   type Response = BanFromCommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     websocket_info: Option<WebsocketInfo>,
-  ) -> Result<BanFromCommunityResponse, Error> {
+  ) -> Result<BanFromCommunityResponse, LemmyError> {
     let data: &BanFromCommunity = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -612,18 +659,16 @@ impl Perform for Oper<BanFromCommunity> {
       user_id: data.user_id,
     };
 
-    let conn = pool.get()?;
-
     if data.ban {
-      match CommunityUserBan::ban(&conn, &community_user_ban_form) {
-        Ok(user) => user,
-        Err(_e) => return Err(APIError::err("community_user_already_banned").into()),
-      };
+      let ban = move |conn: &'_ _| CommunityUserBan::ban(conn, &community_user_ban_form);
+      if blocking(pool, ban).await?.is_err() {
+        return Err(APIError::err("community_user_already_banned").into());
+      }
     } else {
-      match CommunityUserBan::unban(&conn, &community_user_ban_form) {
-        Ok(user) => user,
-        Err(_e) => return Err(APIError::err("community_user_already_banned").into()),
-      };
+      let unban = move |conn: &'_ _| CommunityUserBan::unban(conn, &community_user_ban_form);
+      if blocking(pool, unban).await?.is_err() {
+        return Err(APIError::err("community_user_already_banned").into());
+      }
     }
 
     // Mod tables
@@ -640,9 +685,10 @@ impl Perform for Oper<BanFromCommunity> {
       banned: Some(data.ban),
       expires,
     };
-    ModBanFromCommunity::create(&conn, &form)?;
+    blocking(pool, move |conn| ModBanFromCommunity::create(conn, &form)).await??;
 
-    let user_view = UserView::read(&conn, data.user_id)?;
+    let user_id = data.user_id;
+    let user_view = blocking(pool, move |conn| UserView::read(conn, user_id)).await??;
 
     let res = BanFromCommunityResponse {
       user: user_view,
@@ -662,14 +708,15 @@ impl Perform for Oper<BanFromCommunity> {
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<AddModToCommunity> {
   type Response = AddModToCommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     websocket_info: Option<WebsocketInfo>,
-  ) -> Result<AddModToCommunityResponse, Error> {
+  ) -> Result<AddModToCommunityResponse, LemmyError> {
     let data: &AddModToCommunity = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -684,18 +731,16 @@ impl Perform for Oper<AddModToCommunity> {
       user_id: data.user_id,
     };
 
-    let conn = pool.get()?;
-
     if data.added {
-      match CommunityModerator::join(&conn, &community_moderator_form) {
-        Ok(user) => user,
-        Err(_e) => return Err(APIError::err("community_moderator_already_exists").into()),
-      };
+      let join = move |conn: &'_ _| CommunityModerator::join(conn, &community_moderator_form);
+      if blocking(pool, join).await?.is_err() {
+        return Err(APIError::err("community_moderator_already_exists").into());
+      }
     } else {
-      match CommunityModerator::leave(&conn, &community_moderator_form) {
-        Ok(user) => user,
-        Err(_e) => return Err(APIError::err("community_moderator_already_exists").into()),
-      };
+      let leave = move |conn: &'_ _| CommunityModerator::leave(conn, &community_moderator_form);
+      if blocking(pool, leave).await?.is_err() {
+        return Err(APIError::err("community_moderator_already_exists").into());
+      }
     }
 
     // Mod tables
@@ -705,9 +750,13 @@ impl Perform for Oper<AddModToCommunity> {
       community_id: data.community_id,
       removed: Some(!data.added),
     };
-    ModAddCommunity::create(&conn, &form)?;
+    blocking(pool, move |conn| ModAddCommunity::create(conn, &form)).await??;
 
-    let moderators = CommunityModeratorView::for_community(&conn, data.community_id)?;
+    let community_id = data.community_id;
+    let moderators = blocking(pool, move |conn| {
+      CommunityModeratorView::for_community(conn, community_id)
+    })
+    .await??;
 
     let res = AddModToCommunityResponse { moderators };
 
@@ -724,14 +773,15 @@ impl Perform for Oper<AddModToCommunity> {
   }
 }
 
+#[async_trait::async_trait(?Send)]
 impl Perform for Oper<TransferCommunity> {
   type Response = GetCommunityResponse;
 
-  fn perform(
+  async fn perform(
     &self,
-    pool: Pool<ConnectionManager<PgConnection>>,
+    pool: &DbPool,
     _websocket_info: Option<WebsocketInfo>,
-  ) -> Result<GetCommunityResponse, Error> {
+  ) -> Result<GetCommunityResponse, LemmyError> {
     let data: &TransferCommunity = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -741,12 +791,14 @@ impl Perform for Oper<TransferCommunity> {
 
     let user_id = claims.id;
 
-    let conn = pool.get()?;
+    let community_id = data.community_id;
+    let read_community = blocking(pool, move |conn| Community::read(conn, community_id)).await??;
+
+    let site_creator_id =
+      blocking(pool, move |conn| Site::read(conn, 1).map(|s| s.creator_id)).await??;
 
-    let read_community = Community::read(&conn, data.community_id)?;
+    let mut admins = blocking(pool, move |conn| UserView::admins(conn)).await??;
 
-    let site_creator_id = Site::read(&conn, 1)?.creator_id;
-    let mut admins = UserView::admins(&conn)?;
     let creator_index = admins.iter().position(|r| r.id == site_creator_id).unwrap();
     let creator_user = admins.remove(creator_index);
     admins.insert(0, creator_user);
@@ -774,13 +826,18 @@ impl Perform for Oper<TransferCommunity> {
       published: None,
     };
 
-    let _updated_community = match Community::update(&conn, data.community_id, &community_form) {
-      Ok(community) => community,
-      Err(_e) => return Err(APIError::err("couldnt_update_community").into()),
+    let community_id = data.community_id;
+    let update = move |conn: &'_ _| Community::update(conn, community_id, &community_form);
+    if blocking(pool, update).await?.is_err() {
+      return Err(APIError::err("couldnt_update_community").into());
     };
 
     // You also have to re-do the community_moderator table, reordering it.
-    let mut community_mods = CommunityModeratorView::for_community(&conn, data.community_id)?;
+    let community_id = data.community_id;
+    let mut community_mods = blocking(pool, move |conn| {
+      CommunityModeratorView::for_community(conn, community_id)
+    })
+    .await??;
     let creator_index = community_mods
       .iter()
       .position(|r| r.user_id == data.user_id)
@@ -788,19 +845,23 @@ impl Perform for Oper<TransferCommunity> {
     let creator_user = community_mods.remove(creator_index);
     community_mods.insert(0, creator_user);
 
-    CommunityModerator::delete_for_community(&conn, data.community_id)?;
+    let community_id = data.community_id;
+    blocking(pool, move |conn| {
+      CommunityModerator::delete_for_community(conn, community_id)
+    })
+    .await??;
 
+    // TODO: this should probably be a bulk operation
     for cmod in &community_mods {
       let community_moderator_form = CommunityModeratorForm {
         community_id: cmod.community_id,
         user_id: cmod.user_id,
       };
 
-      let _inserted_community_moderator =
-        match CommunityModerator::join(&conn, &community_moderator_form) {
-          Ok(user) => user,
-          Err(_e) => return Err(APIError::err("community_moderator_already_exists").into()),
-        };
+      let join = move |conn: &'_ _| CommunityModerator::join(conn, &community_moderator_form);
+      if blocking(pool, join).await?.is_err() {
+        return Err(APIError::err("community_moderator_already_exists").into());
+      }
     }
 
     // Mod tables
@@ -810,14 +871,24 @@ impl Perform for Oper<TransferCommunity> {
       community_id: data.community_id,
       removed: Some(false),
     };
-    ModAddCommunity::create(&conn, &form)?;
+    blocking(pool, move |conn| ModAddCommunity::create(conn, &form)).await??;
 
-    let community_view = match CommunityView::read(&conn, data.community_id, Some(user_id)) {
+    let community_id = data.community_id;
+    let community_view = match blocking(pool, move |conn| {
+      CommunityView::read(conn, community_id, Some(user_id))
+    })
+    .await?
+    {
       Ok(community) => community,
       Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
     };
 
-    let moderators = match CommunityModeratorView::for_community(&conn, data.community_id) {
+    let community_id = data.community_id;
+    let moderators = match blocking(pool, move |conn| {
+      CommunityModeratorView::for_community(conn, community_id)
+    })
+    .await?
+    {
       Ok(moderators) => moderators,
       Err(_e) => return Err(APIError::err("couldnt_find_community").into()),
     };