Add a search engine into Plume (#324)

* Add search engine to the model

Add a Tantivy based search engine to the model
Implement most required functions for it

* Implement indexing and plm subcommands

Implement indexation on insert, update and delete
Modify func args to get the indexer where required
Add subcommand to initialize, refill and unlock search db

* Move to a new threadpool engine allowing scheduling

* Autocommit search index every half an hour

* Implement front part of search

Add default fields for search
Add new routes and templates for search and result
Implement FromFormValue for Page to reuse it on search result pagination
Add optional query parameters to paginate template's macro
Update to newer rocket_csrf, don't get csrf token on GET forms

* Handle process termination to release lock

Handle process termination
Add tests to search

* Add proper support for advanced search

Add an advanced search form to /search, in template and route
Modify Tantivy schema, add new tokenizer for some properties
Create new String query parser
Create Tantivy query AST from our own

* Split search.rs, add comment and tests

Split search.rs into multiple submodules
Add comments and tests for Query
Make user@domain be treated as one could assume

src/api/posts.rs

@@ -9,18 +9,20 @@ use plume_models::{
     Connection,
     db_conn::DbConn,
     posts::Post,
+    search::Searcher as UnmanagedSearcher,
 };
 use api::authorization::*;
+use Searcher;
 
 #[get("/posts/<id>")]
-fn get(id: i32, conn: DbConn, auth: Option<Authorization<Read, Post>>) -> Json<serde_json::Value> {
-    let post = <Post as Provider<(&Connection, Option<i32>)>>::get(&(&*conn, auth.map(|a| a.0.user_id)), id).ok();
+fn get(id: i32, conn: DbConn, auth: Option<Authorization<Read, Post>>, search: Searcher) -> Json<serde_json::Value> {
+    let post = <Post as Provider<(&Connection, &UnmanagedSearcher, Option<i32>)>>::get(&(&*conn, &search, auth.map(|a| a.0.user_id)), id).ok();
     Json(json!(post))
 }
 
 #[get("/posts")]
-fn list(conn: DbConn, uri: &Origin, auth: Option<Authorization<Read, Post>>) -> Json<serde_json::Value> {
+fn list(conn: DbConn, uri: &Origin, auth: Option<Authorization<Read, Post>>, search: Searcher) -> Json<serde_json::Value> {
     let query: PostEndpoint = serde_qs::from_str(uri.query().unwrap_or("")).expect("api::list: invalid query error");
-    let post = <Post as Provider<(&Connection, Option<i32>)>>::list(&(&*conn, auth.map(|a| a.0.user_id)), query);
+    let post = <Post as Provider<(&Connection, &UnmanagedSearcher, Option<i32>)>>::list(&(&*conn, &search, auth.map(|a| a.0.user_id)), query);
     Json(json!(post))
 }

src/inbox.rs

@@ -19,11 +19,11 @@ use plume_common::activity_pub::{
 };
 use plume_models::{
     comments::Comment, follows::Follow, instance::Instance, likes, posts::Post, reshares::Reshare,
-    users::User, Connection,
+    users::User, search::Searcher, Connection,
 };
 
 pub trait Inbox {
-    fn received(&self, conn: &Connection, act: serde_json::Value) -> Result<(), Error> {
+    fn received(&self, conn: &Connection, searcher: &Searcher, act: serde_json::Value) -> Result<(), Error> {
         let actor_id = Id::new(act["actor"].as_str().unwrap_or_else(|| {
             act["actor"]["id"]
                 .as_str()
@@ -37,7 +37,7 @@ pub trait Inbox {
                 }
                 "Create" => {
                     let act: Create = serde_json::from_value(act.clone())?;
-                    if Post::try_from_activity(conn, act.clone())
+                    if Post::try_from_activity(&(conn, searcher), act.clone())
                         || Comment::try_from_activity(conn, act)
                     {
                         Ok(())
@@ -53,7 +53,7 @@ pub trait Inbox {
                             .object_props
                             .id_string()?,
                         actor_id.as_ref(),
-                        conn,
+                        &(conn, searcher),
                     );
                     Ok(())
                 }
@@ -113,7 +113,7 @@ pub trait Inbox {
                 }
                 "Update" => {
                     let act: Update = serde_json::from_value(act.clone())?;
-                    Post::handle_update(conn, &act.update_props.object_object()?);
+                    Post::handle_update(conn, &act.update_props.object_object()?, searcher);
                     Ok(())
                 }
                 _ => Err(InboxError::InvalidType)?,

src/main.rs

@@ -6,6 +6,7 @@ extern crate atom_syndication;
 extern crate canapi;
 extern crate chrono;
 extern crate colored;
+extern crate ctrlc;
 extern crate diesel;
 extern crate dotenv;
 extern crate failure;
@@ -13,6 +14,7 @@ extern crate gettextrs;
 extern crate guid_create;
 extern crate heck;
 extern crate multipart;
+extern crate num_cpus;
 extern crate plume_api;
 extern crate plume_common;
 extern crate plume_models;
@@ -22,6 +24,7 @@ extern crate rocket_contrib;
 extern crate rocket_csrf;
 extern crate rocket_i18n;
 extern crate rpassword;
+extern crate scheduled_thread_pool;
 extern crate serde;
 #[macro_use]
 extern crate serde_derive;
@@ -32,20 +35,24 @@ extern crate validator;
 #[macro_use]
 extern crate validator_derive;
 extern crate webfinger;
-extern crate workerpool;
 
 use diesel::r2d2::ConnectionManager;
 use rocket::State;
 use rocket_contrib::Template;
 use rocket_csrf::CsrfFairingBuilder;
-use plume_models::{DATABASE_URL, Connection, db_conn::DbPool};
-use workerpool::{Pool, thunk::ThunkWorker};
+use plume_models::{DATABASE_URL, Connection,
+    db_conn::DbPool, search::Searcher as UnmanagedSearcher};
+use scheduled_thread_pool::ScheduledThreadPool;
+use std::process::exit;
+use std::sync::Arc;
+use std::time::Duration;
 
 mod api;
 mod inbox;
 mod routes;
 
-type Worker<'a> = State<'a, Pool<ThunkWorker<()>>>;
+type Worker<'a> = State<'a, ScheduledThreadPool>;
+type Searcher<'a> = State<'a, Arc<UnmanagedSearcher>>;
 
 /// Initializes a database pool.
 fn init_pool() -> Option<DbPool> {
@@ -56,7 +63,21 @@ fn init_pool() -> Option<DbPool> {
 }
 
 fn main() {
-    let pool = init_pool().expect("main: database pool initialization error");
+
+    let dbpool = init_pool().expect("main: database pool initialization error");
+    let workpool = ScheduledThreadPool::with_name("worker {}", num_cpus::get());
+    let searcher = Arc::new(UnmanagedSearcher::open(&"search_index").unwrap());
+
+    let commiter = searcher.clone();
+    workpool.execute_with_fixed_delay(Duration::from_secs(5), Duration::from_secs(60*30), move || commiter.commit());
+
+    let search_unlocker = searcher.clone();
+    ctrlc::set_handler(move || {
+        search_unlocker.drop_writer();
+        exit(0);
+    }).expect("Error setting Ctrl-c handler");
+
+
     rocket::ignite()
         .mount("/", routes![
             routes::blogs::paginated_details,
@@ -119,6 +140,9 @@ fn main() {
             routes::reshares::create,
             routes::reshares::create_auth,
 
+            routes::search::index,
+            routes::search::query,
+
             routes::session::new,
             routes::session::new_message,
             routes::session::create,
@@ -167,8 +191,9 @@ fn main() {
             routes::errors::not_found,
             routes::errors::server_error
         ])
-        .manage(pool)
-        .manage(Pool::<ThunkWorker<()>>::new(4))
+        .manage(dbpool)
+        .manage(workpool)
+        .manage(searcher)
         .attach(Template::custom(|engines| {
             rocket_i18n::tera(&mut engines.tera);
         }))

src/routes/blogs.rs

@@ -21,6 +21,7 @@ use plume_models::{
     users::User
 };
 use routes::Page;
+use Searcher;
 
 #[get("/~/<name>?<page>", rank = 2)]
 fn paginated_details(name: String, conn: DbConn, user: Option<User>, page: Page) -> Template {
@@ -130,10 +131,10 @@ fn create(conn: DbConn, data: LenientForm<NewBlogForm>, user: User) -> Result<Re
 }
 
 #[post("/~/<name>/delete")]
-fn delete(conn: DbConn, name: String, user: Option<User>) -> Result<Redirect, Option<Template>>{
+fn delete(conn: DbConn, name: String, user: Option<User>, searcher: Searcher) -> Result<Redirect, Option<Template>>{
     let blog = Blog::find_local(&*conn, &name).ok_or(None)?;
     if user.map(|u| u.is_author_in(&*conn, &blog)).unwrap_or(false) {
-        blog.delete(&conn);
+        blog.delete(&conn, &searcher);
         Ok(Redirect::to(uri!(super::instance::index)))
     } else {
         Err(Some(Template::render("errors/403", json!({// TODO actually return 403 error code

src/routes/comments.rs

@@ -1,13 +1,11 @@
 use activitypub::object::Note;
 use rocket::{
-    State,
     request::LenientForm,
     response::Redirect
 };
 use rocket_contrib::Template;
 use serde_json;
 use validator::Validate;
-use workerpool::{Pool, thunk::*};
 
 use plume_common::{utils, activity_pub::{broadcast, ApRequest, ActivityStream}};
 use plume_models::{
@@ -19,6 +17,7 @@ use plume_models::{
     safe_string::SafeString,
     users::User
 };
+use Worker;
 
 #[derive(FromForm, Debug, Validate, Serialize)]
 struct NewCommentForm {
@@ -29,7 +28,7 @@ struct NewCommentForm {
 }
 
 #[post("/~/<blog_name>/<slug>/comment", data = "<data>")]
-fn create(blog_name: String, slug: String, data: LenientForm<NewCommentForm>, user: User, conn: DbConn, worker: State<Pool<ThunkWorker<()>>>)
+fn create(blog_name: String, slug: String, data: LenientForm<NewCommentForm>, user: User, conn: DbConn, worker: Worker)
     -> Result<Redirect, Option<Template>> {
     let blog = Blog::find_by_fqn(&*conn, &blog_name).ok_or(None)?;
     let post = Post::find_by_slug(&*conn, &slug, blog.id).ok_or(None)?;
@@ -56,7 +55,7 @@ fn create(blog_name: String, slug: String, data: LenientForm<NewCommentForm>, us
             // federate
             let dest = User::one_by_instance(&*conn);
             let user_clone = user.clone();
-            worker.execute(Thunk::of(move || broadcast(&user_clone, new_comment, dest)));
+            worker.execute(move || broadcast(&user_clone, new_comment, dest));
 
             Redirect::to(uri!(super::posts::details: blog = blog_name, slug = slug))
         })

src/routes/instance.rs

@@ -18,6 +18,7 @@ use plume_models::{
 };
 use inbox::Inbox;
 use routes::Page;
+use Searcher;
 
 #[get("/")]
 fn index(conn: DbConn, user: Option<User>) -> Template {
@@ -190,15 +191,15 @@ fn admin_users_paginated(admin: Admin, conn: DbConn, page: Page) -> Template {
 }
 
 #[post("/admin/users/<id>/ban")]
-fn ban(_admin: Admin, conn: DbConn, id: i32) -> Redirect {
+fn ban(_admin: Admin, conn: DbConn, id: i32, searcher: Searcher) -> Redirect {
     if let Some(u) = User::get(&*conn, id) {
-        u.delete(&*conn);
+        u.delete(&*conn, &searcher);
     }
     Redirect::to(uri!(admin_users))
 }
 
 #[post("/inbox", data = "<data>")]
-fn shared_inbox(conn: DbConn, data: String, headers: Headers) -> Result<String, status::BadRequest<&'static str>> {
+fn shared_inbox(conn: DbConn, data: String, headers: Headers, searcher: Searcher) -> Result<String, status::BadRequest<&'static str>> {
     let act: serde_json::Value = serde_json::from_str(&data[..]).expect("instance::shared_inbox: deserialization error");
 
     let activity = act.clone();
@@ -216,7 +217,7 @@ fn shared_inbox(conn: DbConn, data: String, headers: Headers) -> Result<String,
         return Ok(String::new());
     }
     let instance = Instance::get_local(&*conn).expect("instance::shared_inbox: local instance not found error");
-    Ok(match instance.received(&*conn, act) {
+    Ok(match instance.received(&*conn, &searcher, act) {
         Ok(_) => String::new(),
         Err(e) => {
             println!("Shared inbox error: {}\n{}", e.as_fail(), e.backtrace());

src/routes/likes.rs

@@ -1,5 +1,4 @@
-use rocket::{State, response::{Redirect, Flash}};
-use workerpool::{Pool, thunk::*};
+use rocket::{response::{Redirect, Flash}};
 
 use plume_common::activity_pub::{broadcast, inbox::{Notify, Deletable}};
 use plume_common::utils;
@@ -10,9 +9,10 @@ use plume_models::{
     posts::Post,
     users::User
 };
+use Worker;
 
 #[post("/~/<blog>/<slug>/like")]
-fn create(blog: String, slug: String, user: User, conn: DbConn, worker: State<Pool<ThunkWorker<()>>>) -> Option<Redirect> {
+fn create(blog: String, slug: String, user: User, conn: DbConn, worker: Worker) -> Option<Redirect> {
     let b = Blog::find_by_fqn(&*conn, &blog)?;
     let post = Post::find_by_slug(&*conn, &slug, b.id)?;
 
@@ -27,12 +27,12 @@ fn create(blog: String, slug: String, user: User, conn: DbConn, worker: State<Po
 
         let dest = User::one_by_instance(&*conn);
         let act = like.to_activity(&*conn);
-        worker.execute(Thunk::of(move || broadcast(&user, act, dest)));
+        worker.execute(move || broadcast(&user, act, dest));
     } else {
         let like = likes::Like::find_by_user_on_post(&*conn, user.id, post.id).expect("likes::create: like exist but not found error");
         let delete_act = like.delete(&*conn);
         let dest = User::one_by_instance(&*conn);
-        worker.execute(Thunk::of(move || broadcast(&user, delete_act, dest)));
+        worker.execute(move || broadcast(&user, delete_act, dest));
     }
 
     Some(Redirect::to(uri!(super::posts::details: blog = blog, slug = slug)))

src/routes/mod.rs

@@ -1,6 +1,8 @@
 use atom_syndication::{ContentBuilder, Entry, EntryBuilder, LinkBuilder, Person, PersonBuilder};
 use rocket::{
-    http::uri::{FromUriParam, UriDisplay},
+    http::{RawStr,
+        uri::{FromUriParam, UriDisplay}},
+    request::FromFormValue,
     response::NamedFile
 };
 use std::{
@@ -57,6 +59,17 @@ impl FromUriParam<i32> for Page {
     }
 }
 
+impl<'v> FromFormValue<'v> for Page {
+    type Error = &'v RawStr;
+
+    fn from_form_value(form_value: &'v RawStr) -> Result<Page, &'v RawStr> {
+        match form_value.parse::<i32>() {
+            Ok(page) => Ok(Page{page}),
+            _ => Err(form_value),
+        }
+    }
+}
+
 impl Page {
     pub fn first() -> Page {
         Page {
@@ -109,6 +122,7 @@ pub mod reshares;
 pub mod session;
 pub mod tags;
 pub mod user;
+pub mod search;
 pub mod well_known;
 
 #[get("/static/<file..>", rank = 2)]

src/routes/posts.rs

@@ -1,13 +1,12 @@
 use activitypub::object::Article;
 use chrono::Utc;
 use heck::{CamelCase, KebabCase};
-use rocket::{State, request::LenientForm};
+use rocket::{request::LenientForm};
 use rocket::response::{Redirect, Flash};
 use rocket_contrib::Template;
 use serde_json;
 use std::{collections::{HashMap, HashSet}, borrow::Cow};
 use validator::{Validate, ValidationError, ValidationErrors};
-use workerpool::{Pool, thunk::*};
 
 use plume_common::activity_pub::{broadcast, ActivityStream, ApRequest, inbox::Deletable};
 use plume_common::utils;
@@ -24,6 +23,8 @@ use plume_models::{
     tags::*,
     users::User
 };
+use Worker;
+use Searcher;
 
 #[derive(FromForm)]
 struct CommentQuery {
@@ -163,7 +164,7 @@ fn edit(blog: String, slug: String, user: User, conn: DbConn) -> Option<Template
 }
 
 #[post("/~/<blog>/<slug>/edit", data = "<data>")]
-fn update(blog: String, slug: String, user: User, conn: DbConn, data: LenientForm<NewPostForm>, worker: State<Pool<ThunkWorker<()>>>)
+fn update(blog: String, slug: String, user: User, conn: DbConn, data: LenientForm<NewPostForm>, worker: Worker, searcher: Searcher)
     -> Result<Redirect, Option<Template>> {
     let b = Blog::find_by_fqn(&*conn, &blog).ok_or(None)?;
     let mut post = Post::find_by_slug(&*conn, &slug, b.id).ok_or(None)?;
@@ -217,7 +218,7 @@ fn update(blog: String, slug: String, user: User, conn: DbConn, data: LenientFor
             post.source = form.content.clone();
             post.license = license;
             post.cover_id = form.cover;
-            post.update(&*conn);
+            post.update(&*conn, &searcher);
             let post = post.update_ap_url(&*conn);
 
             if post.published {
@@ -236,11 +237,11 @@ fn update(blog: String, slug: String, user: User, conn: DbConn, data: LenientFor
                 if newly_published {
                     let act = post.create_activity(&conn);
                     let dest = User::one_by_instance(&*conn);
-                    worker.execute(Thunk::of(move || broadcast(&user, act, dest)));
+                    worker.execute(move || broadcast(&user, act, dest));
                 } else {
                     let act = post.update_activity(&*conn);
                     let dest = User::one_by_instance(&*conn);
-                    worker.execute(Thunk::of(move || broadcast(&user, act, dest)));
+                    worker.execute(move || broadcast(&user, act, dest));
                 }
             }
 
@@ -284,7 +285,7 @@ fn valid_slug(title: &str) -> Result<(), ValidationError> {
 }
 
 #[post("/~/<blog_name>/new", data = "<data>")]
-fn create(blog_name: String, data: LenientForm<NewPostForm>, user: User, conn: DbConn, worker: State<Pool<ThunkWorker<()>>>) -> Result<Redirect, Option<Template>> {
+fn create(blog_name: String, data: LenientForm<NewPostForm>, user: User, conn: DbConn, worker: Worker, searcher: Searcher) -> Result<Redirect, Option<Template>> {
     let blog = Blog::find_by_fqn(&*conn, &blog_name).ok_or(None)?;
     let form = data.get();
     let slug = form.title.to_string().to_kebab_case();
@@ -324,7 +325,9 @@ fn create(blog_name: String, data: LenientForm<NewPostForm>, user: User, conn: D
                 subtitle: form.subtitle.clone(),
                 source: form.content.clone(),
                 cover_id: form.cover,
-            });
+                },
+                &searcher,
+            );
             let post = post.update_ap_url(&*conn);
             PostAuthor::insert(&*conn, NewPostAuthor {
                 post_id: post.id,
@@ -357,7 +360,7 @@ fn create(blog_name: String, data: LenientForm<NewPostForm>, user: User, conn: D
 
                 let act = post.create_activity(&*conn);
                 let dest = User::one_by_instance(&*conn);
-                worker.execute(Thunk::of(move || broadcast(&user, act, dest)));
+                worker.execute(move || broadcast(&user, act, dest));
             }
 
             Ok(Redirect::to(uri!(details: blog = blog_name, slug = slug)))
@@ -377,7 +380,7 @@ fn create(blog_name: String, data: LenientForm<NewPostForm>, user: User, conn: D
 }
 
 #[post("/~/<blog_name>/<slug>/delete")]
-fn delete(blog_name: String, slug: String, conn: DbConn, user: User, worker: State<Pool<ThunkWorker<()>>>) -> Redirect {
+fn delete(blog_name: String, slug: String, conn: DbConn, user: User, worker: Worker, searcher: Searcher) -> Redirect {
     let post = Blog::find_by_fqn(&*conn, &blog_name)
         .and_then(|blog| Post::find_by_slug(&*conn, &slug, blog.id));
 
@@ -386,8 +389,8 @@ fn delete(blog_name: String, slug: String, conn: DbConn, user: User, worker: Sta
             Redirect::to(uri!(details: blog = blog_name.clone(), slug = slug.clone()))
         } else {
             let dest = User::one_by_instance(&*conn);
-            let delete_activity = post.delete(&*conn);
-            worker.execute(Thunk::of(move || broadcast(&user, delete_activity, dest)));
+            let delete_activity = post.delete(&(&conn, &searcher));
+            worker.execute(move || broadcast(&user, delete_activity, dest));
 
             Redirect::to(uri!(super::blogs::details: name = blog_name))
         }

src/routes/reshares.rs

@@ -1,5 +1,4 @@
-use rocket::{State, response::{Redirect, Flash}};
-use workerpool::{Pool, thunk::*};
+use rocket::{response::{Redirect, Flash}};
 
 use plume_common::activity_pub::{broadcast, inbox::{Deletable, Notify}};
 use plume_common::utils;
@@ -10,9 +9,10 @@ use plume_models::{
     reshares::*,
     users::User
 };
+use Worker;
 
 #[post("/~/<blog>/<slug>/reshare")]
-fn create(blog: String, slug: String, user: User, conn: DbConn, worker: State<Pool<ThunkWorker<()>>>) -> Option<Redirect> {
+fn create(blog: String, slug: String, user: User, conn: DbConn, worker: Worker) -> Option<Redirect> {
     let b = Blog::find_by_fqn(&*conn, &blog)?;
     let post = Post::find_by_slug(&*conn, &slug, b.id)?;
 
@@ -27,13 +27,13 @@ fn create(blog: String, slug: String, user: User, conn: DbConn, worker: State<Po
 
         let dest = User::one_by_instance(&*conn);
         let act = reshare.to_activity(&*conn);
-        worker.execute(Thunk::of(move || broadcast(&user, act, dest)));
+        worker.execute(move || broadcast(&user, act, dest));
     } else {
         let reshare = Reshare::find_by_user_on_post(&*conn, user.id, post.id)
             .expect("reshares::create: reshare exist but not found error");
         let delete_act = reshare.delete(&*conn);
         let dest = User::one_by_instance(&*conn);
-        worker.execute(Thunk::of(move || broadcast(&user, delete_act, dest)));
+        worker.execute(move || broadcast(&user, delete_act, dest));
     }
 
     Some(Redirect::to(uri!(super::posts::details: blog = blog, slug = slug)))

src/routes/search.rs

@@ -0,0 +1,84 @@
+use chrono::offset::Utc;
+use rocket_contrib::Template;
+use serde_json;
+
+use plume_models::{
+    db_conn::DbConn, users::User,
+    search::Query};
+use routes::Page;
+use Searcher;
+
+#[get("/search")]
+fn index(conn: DbConn, user: Option<User>) -> Template {
+    Template::render("search/index", json!({
+        "account": user.map(|u| u.to_json(&*conn)),
+        "now": format!("{}", Utc::today().format("%Y-%m-d")),
+    }))
+}
+
+#[derive(FromForm)]
+struct SearchQuery {
+    q: Option<String>,
+    title: Option<String>,
+    subtitle: Option<String>,
+    content: Option<String>,
+    instance: Option<String>,
+    author: Option<String>,
+    tag: Option<String>,
+    blog: Option<String>,
+    lang: Option<String>,
+    license: Option<String>,
+    after: Option<String>,
+    before: Option<String>,
+    page: Option<Page>,
+}
+
+macro_rules! param_to_query {
+    ( $query:ident, $parsed_query:ident; normal: $($field:ident),*; date: $($date:ident),*) => {
+        $(
+            if let Some(field) = $query.$field {
+                let mut rest = field.as_str();
+                while !rest.is_empty() {
+                    let (token, r) = Query::get_first_token(rest);
+                    rest = r;
+                    $parsed_query.$field(token, None);
+                }
+            }
+        )*
+        $(
+            if let Some(field) = $query.$date {
+                let mut rest = field.as_str();
+                while !rest.is_empty() {
+                    use chrono::naive::NaiveDate;
+                    let (token, r) = Query::get_first_token(rest);
+                    rest = r;
+                    if let Ok(token) = NaiveDate::parse_from_str(token, "%Y-%m-%d") {
+                        $parsed_query.$date(&token);
+                    }
+                }
+            }
+        )*
+    }
+}
+
+
+#[get("/search?<query>")]
+fn query(query: SearchQuery, conn: DbConn, searcher: Searcher, user: Option<User>) -> Template {
+    let page = query.page.unwrap_or(Page::first());
+    let mut parsed_query = Query::from_str(&query.q.unwrap_or_default());
+
+    param_to_query!(query, parsed_query; normal: title, subtitle, content, tag,
+              instance, author, blog, lang, license;
+              date: before, after);
+
+    let str_q = parsed_query.to_string();
+    let res = searcher.search_document(&conn, parsed_query, page.limits());
+
+    Template::render("search/result", json!({
+        "query":str_q,
+        "account": user.map(|u| u.to_json(&*conn)),
+        "next_page": if res.is_empty() { 0 } else { page.page+1 },
+        "posts": res.into_iter().map(|p| p.to_json(&*conn)).collect::<Vec<serde_json::Value>>(),
+        "page": page.page,
+    }))
+}

src/routes/user.rs

@@ -8,7 +8,6 @@ use rocket::{
 use rocket_contrib::Template;
 use serde_json;
 use validator::{Validate, ValidationError};
-use workerpool::thunk::*;
 
 use inbox::Inbox;
 use plume_common::activity_pub::{
@@ -24,6 +23,7 @@ use plume_models::{
 };
 use routes::Page;
 use Worker;
+use Searcher;
 
 #[get("/me")]
 fn me(user: Option<User>) -> Result<Redirect, Flash<Redirect>> {
@@ -39,9 +39,10 @@ fn details(
     conn: DbConn,
     account: Option<User>,
     worker: Worker,
-    fecth_articles_conn: DbConn,
-    fecth_followers_conn: DbConn,
+    fetch_articles_conn: DbConn,
+    fetch_followers_conn: DbConn,
     update_conn: DbConn,
+    searcher: Searcher,
 ) -> Template {
     may_fail!(
         account.map(|a| a.to_json(&*conn)),
@@ -56,12 +57,13 @@ fn details(
             if !user.get_instance(&*conn).local {
                 // Fetch new articles
                 let user_clone = user.clone();
-                worker.execute(Thunk::of(move || {
+                let searcher = searcher.clone();
+                worker.execute(move || {
                     for create_act in user_clone.fetch_outbox::<Create>() {
                         match create_act.create_props.object_object::<Article>() {
                             Ok(article) => {
                                 Post::from_activity(
-                                    &*fecth_articles_conn,
+                                    &(&fetch_articles_conn, &searcher),
                                     article,
                                     user_clone.clone().into_id(),
                                 );
@@ -72,20 +74,20 @@ fn details(
                             }
                         }
                     }
-                }));
+                });
 
                 // Fetch followers
                 let user_clone = user.clone();
-                worker.execute(Thunk::of(move || {
+                worker.execute(move || {
                     for user_id in user_clone.fetch_followers_ids() {
                         let follower =
-                            User::find_by_ap_url(&*fecth_followers_conn, &user_id)
+                            User::find_by_ap_url(&*fetch_followers_conn, &user_id)
                                 .unwrap_or_else(|| {
-                                    User::fetch_from_url(&*fecth_followers_conn, &user_id)
+                                    User::fetch_from_url(&*fetch_followers_conn, &user_id)
                                         .expect("user::details: Couldn't fetch follower")
                                 });
                         follows::Follow::insert(
-                            &*fecth_followers_conn,
+                            &*fetch_followers_conn,
                             follows::NewFollow {
                                 follower_id: follower.id,
                                 following_id: user_clone.id,
@@ -93,14 +95,14 @@ fn details(
                             },
                         );
                     }
-                }));
+                });
 
                 // Update profile information if needed
                 let user_clone = user.clone();
                 if user.needs_update() {
-                    worker.execute(Thunk::of(move || {
+                    worker.execute(move || {
                         user_clone.refetch(&*update_conn);
-                    }))
+                    });
                 }
             }
 
@@ -148,9 +150,9 @@ fn follow(name: String, conn: DbConn, user: User, worker: Worker) -> Option<Redi
     let target = User::find_by_fqn(&*conn, &name)?;
     if let Some(follow) = follows::Follow::find(&*conn, user.id, target.id) {
         let delete_act = follow.delete(&*conn);
-        worker.execute(Thunk::of(move || {
+        worker.execute(move || {
             broadcast(&user, delete_act, vec![target])
-        }));
+        });
     } else {
         let f = follows::Follow::insert(
             &*conn,
@@ -163,7 +165,7 @@ fn follow(name: String, conn: DbConn, user: User, worker: Worker) -> Option<Redi
         f.notify(&*conn);
 
         let act = f.to_activity(&*conn);
-        worker.execute(Thunk::of(move || broadcast(&user, act, vec![target])));
+        worker.execute(move || broadcast(&user, act, vec![target]));
     }
     Some(Redirect::to(uri!(details: name = name)))
 }
@@ -285,10 +287,10 @@ fn update(_name: String, conn: DbConn, user: User, data: LenientForm<UpdateUserF
 }
 
 #[post("/@/<name>/delete")]
-fn delete(name: String, conn: DbConn, user: User, mut cookies: Cookies) -> Option<Redirect> {
+fn delete(name: String, conn: DbConn, user: User, mut cookies: Cookies, searcher: Searcher) -> Option<Redirect> {
     let account = User::find_by_fqn(&*conn, &name)?;
     if user.id == account.id {
-        account.delete(&*conn);
+        account.delete(&*conn, &searcher);
 
     if let Some(cookie) = cookies.get_private(AUTH_COOKIE) {
         cookies.remove_private(cookie);
@@ -393,6 +395,7 @@ fn inbox(
     conn: DbConn,
     data: String,
     headers: Headers,
+    searcher: Searcher,
 ) -> Result<String, Option<status::BadRequest<&'static str>>> {
     let user = User::find_local(&*conn, &name).ok_or(None)?;
     let act: serde_json::Value =
@@ -420,7 +423,7 @@ fn inbox(
     if Instance::is_blocked(&*conn, actor_id) {
         return Ok(String::new());
     }
-    Ok(match user.received(&*conn, act) {
+    Ok(match user.received(&*conn, &searcher, act) {
         Ok(_) => String::new(),
         Err(e) => {
             println!("User inbox error: {}\n{}", e.as_fail(), e.backtrace());