Port crates#index to use Diesel

I decided to tackle this one next since we have an N+1 queries bug on
the max versions, it's a relatively high traffic endpoint, and the code
benefitted the most from Diesel.

There was probably more code added to support migrating the tests over
than to support the actual `index` function, but all those pieces will
start to get used as more endpoints are ported over. There were a few
tests that were hitting the followings and owners endpoints which I have
temporarily changed to hit the database directly instead.

Porting those endpoints would mean porting over more tests, which might
mean porting over more endpoints, etc. We didn't lose any coverage
overall, but we should still go back to hitting the endpoints directly
in tests when we can.

Author: sgrif

Cargo.lock

@@ -35,10 +35,11 @@ rustc-serialize = "0.3"
 license-exprs = "^1.3"
 dotenv = "0.8.0"
 toml = "0.2"
-diesel = { version = "0.11.0", features = ["postgres", "serde_json"] }
+diesel = { version = "0.11.0", features = ["postgres", "serde_json", "deprecated-time"] }
 diesel_codegen = { version = "0.11.0", features = ["postgres"] }
 r2d2-diesel = "0.11.0"
 diesel_full_text_search = "0.11.0"
+serde_json = "0.9.0"
 
 conduit = "0.8"
 conduit-conditional-get = "0.8"

Cargo.toml

@@ -0,0 +1 @@
+ALTER TABLE versions ALTER COLUMN yanked DROP NOT NULL;

migrations/20170307211844_versions_yanked_is_not_nullalbe/down.sql

@@ -0,0 +1 @@
+ALTER TABLE versions ALTER COLUMN yanked SET NOT NULL;

migrations/20170307211844_versions_yanked_is_not_nullalbe/up.sql

@@ -54,7 +54,8 @@ impl App {
             .helper_threads(if config.env == ::Env::Production {3} else {1})
             .build();
         let diesel_db_config = r2d2::Config::builder()
-            .pool_size(if config.env == ::Env::Production {1} else {1})
+            .pool_size(if config.env == ::Env::Production {50} else {1})
+            .min_idle(if config.env == ::Env::Production {Some(5)} else {None})
             .helper_threads(if config.env == ::Env::Production {3} else {1})
             .build();
 

src/app.rs

@@ -1,12 +1,16 @@
-use util::CargoResult;
-use krate::Crate;
 use Model;
+use krate::Crate;
+use schema::badges;
+use util::CargoResult;
 
-use std::collections::HashMap;
+use diesel::pg::Pg;
+use diesel::prelude::*;
 use pg::GenericConnection;
 use pg::rows::Row;
 use rustc_serialize::Decodable;
 use rustc_serialize::json::{Json, Decoder};
+use serde_json;
+use std::collections::HashMap;
 
 #[derive(Debug, PartialEq, Clone)]
 pub enum Badge {
@@ -27,6 +31,17 @@ pub struct EncodableBadge {
     pub attributes: HashMap<String, String>,
 }
 
+impl Queryable<badges::SqlType, Pg> for Badge {
+    type Row = (i32, String, serde_json::Value);
+
+    fn build((_, badge_type, attributes): Self::Row) -> Self {
+        let attributes = serde_json::from_value::<HashMap<String, String>>(attributes)
+            .expect("attributes was not a map in the database");
+        Self::from_attributes(&badge_type, &attributes)
+            .expect("invalid badge in the database")
+    }
+}
+
 impl Model for Badge {
     fn from_row(row: &Row) -> Badge {
         let attributes: Json = row.get("attributes");

src/badge.rs

@@ -3,22 +3,37 @@ use time::Timespec;
 
 use conduit::{Request, Response};
 use conduit_router::RequestParams;
+use diesel::*;
+use diesel::pg::PgConnection;
 use pg::GenericConnection;
 use pg::rows::Row;
 
 use {Model, Crate};
 use db::RequestTransaction;
+use schema::*;
 use util::{RequestUtils, CargoResult, ChainError};
 use util::errors::NotFound;
 
-#[derive(Clone)]
+#[derive(Clone, Identifiable, Associations, Queryable)]
+#[has_many(crates_categories)]
+#[table_name="categories"]
 pub struct Category {
     pub id: i32,
     pub category: String,
     pub slug: String,
     pub description: String,
-    pub created_at: Timespec,
     pub crates_cnt: i32,
+    pub created_at: Timespec,
+}
+
+#[derive(Associations, Insertable, Identifiable)]
+#[belongs_to(Category)]
+#[belongs_to(Crate)]
+#[table_name="crates_categories"]
+#[primary_key(crate_id, category_id)]
+struct CrateCategory {
+    crate_id: i32,
+    category_id: i32,
 }
 
 #[derive(RustcEncodable, RustcDecodable)]
@@ -77,7 +92,27 @@ impl Category {
         }
     }
 
-    pub fn update_crate(conn: &GenericConnection,
+    pub fn update_crate<'a>(conn: &PgConnection,
+                            krate: &Crate,
+                            slugs: &[&'a str]) -> QueryResult<Vec<&'a str>> {
+        use diesel::expression::dsl::any;
+
+        let categories = categories::table
+            .filter(categories::slug.eq(any(slugs)))
+            .load::<Category>(conn)?;
+        let invalid_categories = slugs.iter().cloned()
+            .filter(|s| !categories.iter().any(|c| c.slug == *s))
+            .collect();
+        let crate_categories = categories.iter()
+            .map(|c| CrateCategory { category_id: c.id, crate_id: krate.id })
+            .collect::<Vec<_>>();
+
+        delete(CrateCategory::belonging_to(krate)).execute(conn)?;
+        insert(&crate_categories).into(crates_categories::table).execute(conn)?;
+        Ok(invalid_categories)
+    }
+
+    pub fn update_crate_old(conn: &GenericConnection,
                         krate: &Crate,
                         categories: &[String]) -> CargoResult<Vec<String>> {
         let old_categories = krate.categories(conn)?;
@@ -194,6 +229,32 @@ impl Category {
     }
 }
 
+#[derive(Insertable, Default)]
+#[table_name="categories"]
+pub struct NewCategory<'a> {
+    pub category: &'a str,
+    pub slug: &'a str,
+}
+
+impl<'a> NewCategory<'a> {
+    pub fn create_or_update(&self, conn: &PgConnection) -> QueryResult<Category> {
+        use schema::categories::dsl::*;
+        use diesel::pg::upsert::*;
+
+        let maybe_inserted = insert(&self.on_conflict_do_nothing())
+            .into(categories)
+            .get_result(conn)
+            .optional()?;
+
+        if let Some(c) = maybe_inserted {
+            return Ok(c);
+        }
+
+        categories.filter(slug.eq(self.slug))
+            .first(conn)
+    }
+}
+
 impl Model for Category {
     fn from_row(row: &Row) -> Category {
         Category {

src/category.rs

@@ -4,20 +4,35 @@ use time::Timespec;
 
 use conduit::{Request, Response};
 use conduit_router::RequestParams;
+use diesel::pg::PgConnection;
+use diesel::prelude::*;
+use diesel;
 use pg::GenericConnection;
 use pg::rows::Row;
 
 use {Model, Crate};
 use db::RequestTransaction;
+use schema::*;
 use util::{RequestUtils, CargoResult, ChainError, internal};
 use util::errors::NotFound;
 
-#[derive(Clone)]
+#[derive(Clone, Identifiable, Associations, Queryable)]
+#[has_many(crates_keywords)]
 pub struct Keyword {
     pub id: i32,
     pub keyword: String,
-    pub created_at: Timespec,
     pub crates_cnt: i32,
+    pub created_at: Timespec,
+}
+
+#[derive(Associations, Insertable, Identifiable)]
+#[belongs_to(Keyword)]
+#[belongs_to(Crate)]
+#[table_name="crates_keywords"]
+#[primary_key(crate_id, keyword_id)]
+struct CrateKeyword {
+    crate_id: i32,
+    keyword_id: i32,
 }
 
 #[derive(RustcEncodable, RustcDecodable)]
@@ -37,6 +52,27 @@ impl Keyword {
         Ok(rows.iter().next().map(|r| Model::from_row(&r)))
     }
 
+    pub fn find_all(conn: &PgConnection, names: &[&str]) -> QueryResult<Vec<Keyword>> {
+        use diesel::pg::upsert::*;
+        use diesel::expression::dsl::any;
+
+        #[derive(Insertable)]
+        #[table_name="keywords"]
+        struct NewKeyword<'a> {
+            keyword: &'a str,
+        }
+        sql_function!(lower, lower_t, (x: ::diesel::types::Text) -> ::diesel::types::Text);
+
+        let (lowercase_names, new_keywords): (Vec<_>, Vec<_>) = names.iter()
+            .map(|s| (s.to_lowercase(), NewKeyword { keyword: *s }))
+            .unzip();
+
+        diesel::insert(&new_keywords.on_conflict_do_nothing()).into(keywords::table)
+            .execute(conn)?;
+        keywords::table.filter(lower(keywords::keyword).eq(any(lowercase_names)))
+            .load(conn)
+    }
+
     pub fn find_or_insert(conn: &GenericConnection, name: &str)
                           -> CargoResult<Keyword> {
         // TODO: racy (the select then insert is not atomic)
@@ -91,7 +127,21 @@ impl Keyword {
         }
     }
 
-    pub fn update_crate(conn: &GenericConnection,
+    pub fn update_crate(conn: &PgConnection,
+                        krate: &Crate,
+                        keywords: &[&str]) -> QueryResult<()> {
+        let keywords = Keyword::find_all(conn, keywords)?;
+        diesel::delete(CrateKeyword::belonging_to(krate))
+            .execute(conn)?;
+        let crate_keywords = keywords.into_iter().map(|kw| {
+            CrateKeyword { crate_id: krate.id, keyword_id: kw.id }
+        }).collect::<Vec<_>>();
+        diesel::insert(&crate_keywords).into(crates_keywords::table)
+            .execute(conn)?;
+        Ok(())
+    }
+
+    pub fn update_crate_old(conn: &GenericConnection,
                         krate: &Crate,
                         keywords: &[String]) -> CargoResult<()> {
         let old_kws = krate.keywords(conn)?;