aitzol/Plume
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add async/.await until all our errors are the same: that our Connection is not Send-safe.

Browse Source
This commit is contained in:
Mina Galić
2020-05-23 23:32:10 +02:00
parent 850b3c1337
commit d2881ee3f7
12 changed files with 264 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/mod.rs
+1
View File
@@ -66,6 +66,7 @@ pub struct OAuthRequest {
}
#[get("/oauth2?<query..>")]
#[rocket::async_trait(?Send)]
pub async fn oauth(
query: Form<OAuthRequest>,
rockets: PlumeRocket,
src/inbox.rs
+2 -1
View File
@@ -11,7 +11,7 @@ use rocket_contrib::json::*;
use serde::Deserialize;
use tokio::io::AsyncReadExt;
pub fn handle_incoming(
pub async fn handle_incoming(
rockets: PlumeRocket,
data: SignedJson<serde_json::Value>,
headers: Headers<'_>,
@@ -32,6 +32,7 @@ pub fn handle_incoming(
// maybe we just know an old key?
actor
.refetch(conn)
.await
.and_then(|_| User::get(conn, actor.id))
.and_then(|u| {
if verify_http_headers(&u, &headers.0, &sig).is_secure() || act.clone().verify(&u) {
src/routes/blogs.rs
+17 -9
View File
@@ -19,10 +19,14 @@ use plume_models::{
};
#[get("/~/<name>?<page>", rank = 2)]
pub fn details(name: String, page: Option<Page>, rockets: PlumeRocket) -> Result<Ructe, ErrorPage> {
pub async fn details(
name: String,
page: Option<Page>,
rockets: PlumeRocket,
) -> Result<Ructe, ErrorPage> {
let page = page.unwrap_or_default();
let conn = &*rockets.conn;
let blog = Blog::find_by_fqn(&rockets, &name)?;
let blog = Blog::find_by_fqn(&rockets, &name).await?;
let posts = Post::blog_page(conn, &blog, page.limits())?;
let articles_count = Post::count_for_blog(conn, &blog)?;
let authors = &blog.list_authors(conn)?;
@@ -43,7 +47,7 @@ pub async fn activity_details(
rockets: PlumeRocket,
_ap: ApRequest,
) -> Option<ActivityStream<CustomGroup>> {
let blog = Blog::find_by_fqn(&rockets, &name).await?.ok()?;
let blog = Blog::find_by_fqn(&rockets, &name).await?;
Some(ActivityStream::new(blog.to_activity(&*rockets.conn).ok()?))
}
@@ -83,7 +87,7 @@ fn valid_slug(title: &str) -> Result<(), ValidationError> {
}
#[post("/blogs/new", data = "<form>")]
pub fn create(form: LenientForm<NewBlogForm>, rockets: PlumeRocket) -> RespondOrRedirect {
pub async fn create(form: LenientForm<NewBlogForm>, rockets: PlumeRocket) -> RespondOrRedirect {
let slug = utils::make_actor_id(&form.title);
let conn = &*rockets.conn;
let intl = &rockets.intl.catalog;
@@ -143,9 +147,11 @@ pub fn create(form: LenientForm<NewBlogForm>, rockets: PlumeRocket) -> RespondOr
}
#[post("/~/<name>/delete")]
pub fn delete(name: String, rockets: PlumeRocket) -> RespondOrRedirect {
pub async fn delete(name: String, rockets: PlumeRocket) -> RespondOrRedirect {
let conn = &*rockets.conn;
let blog = Blog::find_by_fqn(&rockets, &name).expect("blog::delete: blog not found");
let blog = Blog::find_by_fqn(&rockets, &name)
.await
.expect("blog::delete: blog not found");
if rockets
.user
@@ -348,9 +354,10 @@ pub async fn outbox(
name: String,
rockets: PlumeRocket,
) -> Option<ActivityStream<OrderedCollection>> {
let blog = Blog::find_by_fqn(&rockets, &name).await?.ok()?;
let blog = Blog::find_by_fqn(&rockets, &name).await?;
Some(blog.outbox(&*rockets.conn).ok()?)
}
#[allow(unused_variables)]
#[get("/~/<name>/outbox?<page>")]
pub async fn outbox_page(
@@ -358,12 +365,13 @@ pub async fn outbox_page(
page: Page,
rockets: PlumeRocket,
) -> Option<ActivityStream<OrderedCollectionPage>> {
let blog = Blog::find_by_fqn(&rockets, &name).await?.ok()?;
let blog = Blog::find_by_fqn(&rockets, &name).await?;
Some(blog.outbox_page(&*rockets.conn, page.limits()).ok()?)
}
#[get("/~/<name>/atom.xml")]
pub async fn atom_feed(name: String, rockets: PlumeRocket) -> Option<Content<String>> {
let blog = Blog::find_by_fqn(&rockets, &name).await?.ok()?;
let blog = Blog::find_by_fqn(&rockets, &name).await?;
let conn = &*rockets.conn;
let entries = Post::get_recents_for_blog(&*conn, &blog, 15).ok()?;
let uri = Instance::get_local()
src/routes/comments.rs
+9 -8
View File
@@ -28,7 +28,7 @@ pub struct NewCommentForm {
}
#[post("/~/<blog_name>/<slug>/comment", data = "<form>")]
pub fn create(
pub async fn create(
blog_name: String,
slug: String,
form: LenientForm<NewCommentForm>,
@@ -39,9 +39,7 @@ pub fn create(
let blog = Blog::find_by_fqn(&rockets, &blog_name)
.await
.expect("comments::create: blog error");
let post = Post::find_by_slug(&*conn, &slug, blog.id)
.await
.expect("comments::create: post error");
let post = Post::find_by_slug(&*conn, &slug, blog.id).expect("comments::create: post error");
form.validate()
.map(|_| {
let (html, mentions, _hashtags) = utils::md_to_html(
@@ -70,6 +68,7 @@ pub fn create(
.expect("comments::create: insert error");
let new_comment = comm
.create_activity(&rockets)
.await
.expect("comments::create: activity error");
// save mentions
@@ -77,6 +76,7 @@ pub fn create(
Mention::from_activity(
&*conn,
&Mention::build_activity(&rockets, &ment)
.await
.expect("comments::create: build mention error"),
comm.id,
false,
@@ -185,8 +185,9 @@ pub fn activity_pub(
_ap: ApRequest,
rockets: PlumeRocket,
) -> Option<ActivityStream<Note>> {
Comment::get(&*rockets.conn, id)
.and_then(|c| c.to_activity(&rockets))
.ok()
.map(ActivityStream::new)
let c = match Comment::get(&*rockets.conn, id) {
Ok(c) => c.to_activity(&rockets).await.ok(),
Err(_) => None,
};
c.map(ActivityStream::new)
}
src/routes/errors.rs
+6 -6
View File
@@ -41,20 +41,20 @@ impl<'r> Responder<'r> for ErrorPage {
}
#[catch(404)]
pub fn not_found(req: &Request<'_>) -> Ructe {
let rockets = req.guard::<PlumeRocket>().unwrap();
pub async fn not_found(req: &Request<'_>) -> Ructe {
let rockets = req.guard::<PlumeRocket>().await.unwrap();
render!(errors::not_found(&rockets.to_context()))
}
#[catch(422)]
pub fn unprocessable_entity(req: &Request<'_>) -> Ructe {
let rockets = req.guard::<PlumeRocket>().unwrap();
pub async fn unprocessable_entity(req: &Request<'_>) -> Ructe {
let rockets = req.guard::<PlumeRocket>().await.unwrap();
render!(errors::unprocessable_entity(&rockets.to_context()))
}
#[catch(500)]
pub fn server_error(req: &Request<'_>) -> Ructe {
let rockets = req.guard::<PlumeRocket>().unwrap();
pub async fn server_error(req: &Request<'_>) -> Ructe {
let rockets = req.guard::<PlumeRocket>().await.unwrap();
render!(errors::server_error(&rockets.to_context()))
}
src/routes/posts.rs
+4 -3
View File
@@ -304,11 +304,11 @@ pub async fn update(
if post.published {
post.update_mentions(
&conn,
mentions
.into_iter()
.filter_map(|m| Mention::build_activity(&rockets, &m).await.ok())
.collect(),
&conn,
)
.expect("post::update: mentions error");
}
@@ -645,9 +645,10 @@ pub async fn remote_interact_post(
let target = Blog::find_by_fqn(&rockets, &blog_name)
.await
.and_then(|blog| Post::find_by_slug(&rockets.conn, &slug, blog.id))?;
if let Some(uri) = User::fetch_remote_interact_uri(&remote.remote)
if let uri = User::fetch_remote_interact_uri(&remote.remote)
.await
.map(|uri| uri.replace("{uri}", format!("{}", target.ap_url)).ok())
.map(|uri| uri.replace("{uri}", &format!("{}", target.ap_url)))
.unwrap()
{
Ok(Redirect::to(uri).into())
} else {
src/routes/reshares.rs
+2 -2
View File
@@ -10,14 +10,14 @@ use plume_models::{
};
#[post("/~/<blog>/<slug>/reshare")]
pub fn create(
pub async fn create(
blog: String,
slug: String,
user: User,
rockets: PlumeRocket,
) -> Result<Redirect, ErrorPage> {
let conn = &*rockets.conn;
let b = Blog::find_by_fqn(&rockets, &blog)?;
let b = Blog::find_by_fqn(&rockets, &blog).await?;
let post = Post::find_by_slug(&*conn, &slug, b.id)?;
if !user.has_reshared(&*conn, &post)? {
src/routes/session.rs
+5 -3
View File
@@ -42,14 +42,16 @@ pub struct LoginForm {
}
#[post("/login", data = "<form>")]
pub fn create(
pub async fn create(
form: LenientForm<LoginForm>,
mut cookies: Cookies<'_>,
rockets: PlumeRocket,
) -> RespondOrRedirect {
let conn = &*rockets.conn;
let user = User::find_by_email(&*conn, &form.email_or_name)
.or_else(|_| User::find_by_fqn(&rockets, &form.email_or_name));
let user = match User::find_by_email(&*conn, &form.email_or_name) {
Ok(user) => Ok(user),
Err(_) => User::find_by_fqn(&rockets, &form.email_or_name).await,
};
let mut errors = match form.validate() {
Ok(_) => ValidationErrors::new(),
Err(e) => e,
src/routes/user.rs
+15 -15
View File
@@ -50,7 +50,7 @@ pub async fn details(
update_conn: DbConn,
) -> Result<Ructe, ErrorPage> {
let conn = &*rockets.conn;
let user = User::find_by_fqn(&rockets, &name)?;
let user = User::find_by_fqn(&rockets, &name).await?;
let recents = Post::get_recents_for_author(&*conn, &user, 6)?;
let reshares = Reshare::get_recents_for_author(&*conn, &user, 6)?;
let worker = &rockets.worker;
@@ -149,7 +149,7 @@ pub fn dashboard_auth(i18n: I18n) -> Flash<Redirect> {
}
#[post("/@/<name>/follow")]
pub fn follow(
pub async fn follow(
name: String,
user: User,
rockets: PlumeRocket,
@@ -193,7 +193,7 @@ pub fn follow(
}
#[post("/@/<name>/follow", data = "<remote_form>", rank = 2)]
pub fn follow_not_connected(
pub async fn follow_not_connected(
rockets: PlumeRocket,
name: String,
remote_form: Option<LenientForm<RemoteForm>>,
@@ -205,14 +205,14 @@ pub fn follow_not_connected(
.await
.ok()
.and_then(|uri| {
uri.replace(
Some(uri.replace(
"{uri}",
format!(
&format!(
"{}@{}",
target.fqn,
target.get_instance(&rockets.conn).ok()?.public_domain
),
)
))
})
{
Ok(Redirect::to(uri).into())
@@ -270,14 +270,14 @@ pub fn follow_auth(name: String, i18n: I18n) -> Flash<Redirect> {
#[get("/@/<name>/followers?<page>", rank = 2)]
pub async fn followers(
String,
name: String,
page: Option<Page>,
rockets: PlumeRocket,
) -> Result<Ructe, ErrorPage> {
let conn = &*rockets.conn;
let page = page.unwrap_or_default();
let user = User::find_by_fqn(&rockets, &name).await?;
let followers_count = user.count_followers(&*;
let user: User = User::find_by_fqn(&rockets, &name).await?;
let followers_count = user.count_followers(&conn);
Ok(render!(users::followers(
&rockets.to_context(),
@@ -296,7 +296,7 @@ pub async fn followers(
}
#[get("/@/<name>/followed?<page>", rank = 2)]
pub fn followed(
pub async fn followed(
name: String,
page: Option<Page>,
rockets: PlumeRocket,
@@ -328,8 +328,8 @@ pub async fn activity_details(
rockets: PlumeRocket,
_ap: ApRequest,
) -> Option<ActivityStream<CustomPerson>> {
let user = User::find_by_fqn(&rockets, &name).await?.ok()?;
Some(ActivityStream::new(user.to_activity(&*roonn).ok()?))
let user: User = User::find_by_fqn(&rockets, &name).await?;
Some(ActivityStream::new(user.to_activity(&*rockets.conn).ok()?))
}
#[get("/users/new")]
@@ -595,7 +595,7 @@ pub async fn inbox(
User::find_by_fqn(&rockets, &name)
.await
.map_err(|_| status::BadRequest(Some("User not found")))?;
inbox::handle_incoming(rockets, data, headers)
inbox::handle_incoming(rockets, data, headers).await
}
#[get("/@/<name>/followers", rank = 1)]
@@ -604,7 +604,7 @@ pub async fn ap_followers(
rockets: PlumeRocket,
_ap: ApRequest,
) -> Option<ActivityStream<OrderedCollection>> {
let user = User::find_by_fqn(&rockets, &name).await?.ok()?;
let user = User::find_by_fqn(&rockets, &name).await?;
let followers = user
.get_followers(&*rockets.conn)
.ok()?
@@ -626,7 +626,7 @@ pub async fn ap_followers(
#[get("/@/<name>/atom.xml")]
pub async fn atom_feed(name: String, rockets: PlumeRocket) -> Option<Content<String>> {
let conn = &*rockets.conn;
let author = User::find_by_fqn(&rockets, &name).await?.ok()?;
let author = User::find_by_fqn(&rockets, &name).await?;
let entries = Post::get_recents_for_author(conn, &author, 15).ok()?;
let uri = Instance::get_local()
.ok()?
src/routes/well_known.rs
+13 -7
View File
@@ -1,3 +1,4 @@
use async_trait;
use rocket::http::ContentType;
use rocket::response::Content;
use serde_json;
@@ -43,12 +44,14 @@ pub fn host_meta() -> String {
struct WebfingerResolver;
impl Resolver<PlumeRocket> for WebfingerResolver {
fn instance_domain<'a>(&self) -> &'a str {
#[async_trait::async_trait]
impl AsyncResolver for WebfingerResolver {
type Repo = PlumeRocket;
async fn instance_domain<'a>(&self) -> &'a str {
CONFIG.base_url.as_str()
}
fn find(
async fn find(
&self,
prefix: Prefix,
acct: String,
@@ -66,9 +69,9 @@ impl Resolver<PlumeRocket> for WebfingerResolver {
Prefix::Custom(_) => Err(ResolverError::NotFound),
}
}
fn endpoint(
async fn endpoint(
&self,
resource: impl Into<String>,
resource: impl Into<String> + 'async_trait,
resource_repo: PlumeRocket,
) -> Result<Webfinger, ResolverError> {
let resource = resource.into();
@@ -80,7 +83,9 @@ impl Resolver<PlumeRocket> for WebfingerResolver {
let user = parsed_res.next().ok_or(ResolverError::InvalidResource)?;
let domain = parsed_res.next().ok_or(ResolverError::InvalidResource)?;
if domain == webfinger.instance_domain() {
webfinger.find(res_prefix, user.to_string(), resource_repo)
webfinger
.find(res_prefix, user.to_string(), resource_repo)
.await
} else {
Err(ResolverError::WrongDomain)
}
@@ -88,8 +93,9 @@ impl Resolver<PlumeRocket> for WebfingerResolver {
}
#[get("/.well-known/webfinger?<resource>")]
pub fn webfinger(resource: String, rockets: PlumeRocket) -> Content<String> {
pub async fn webfinger(resource: String, rockets: PlumeRocket) -> Content<String> {
match WebfingerResolver::endpoint(resource, rockets)
.await
.and_then(|wf| serde_json::to_string(&wf).map_err(|_| ResolverError::NotFound))
{
Ok(wf) => Content(ContentType::new("application", "jrd+json"), wf),