random hash username and clear users when restart

Cargo.toml

@@ -15,7 +15,7 @@ rocket = { version = "=0.5.0-rc.2", features = ["json"] }
 rocket_sync_db_pools = { version = "=0.1.0-rc.2", features = ["diesel_postgres_pool"] }
 diesel = { version = "1.4.8", features = ["postgres", "chrono"] }
 diesel_migrations = "1.4.0"
-redis = { version="0.21.5", features = ["aio", "tokio-comp"] }
+redis = { version="0.23.0", features = ["aio", "tokio-comp"] }
 chrono = { version="0.4.19", features = ["serde"] }
 rand = "0.8.5"
 dotenv = "0.15.0"
@@ -25,5 +25,6 @@ env_logger = "0.9.0"
 web-push = "0.9.2"
 url = "2.2.2"
 futures = "0.3.24"
+futures-util = "0.3.24"
 
 reqwest = { version = "0.11.10", features = ["json"], optional = true }

src/cache.rs

@@ -1,11 +1,12 @@
 use crate::api::CurrentUser;
 use crate::models::{Comment, Post, User};
 use crate::rds_conn::RdsConn;
-use crate::rds_models::{init, BlockedUsers};
+use crate::rds_models::{clear_all, init, BlockedUsers};
 use rand::Rng;
 use redis::{AsyncCommands, RedisError, RedisResult};
 use rocket::serde::json::serde_json;
 // can use rocket::serde::json::to_string in master version
+use futures_util::stream::StreamExt;
 use rocket::futures::future;
 use std::collections::HashMap;
 
@@ -28,6 +29,8 @@ pub struct PostCache {
 impl PostCache {
     init!();
 
+    clear_all!("hole_v2:cache::post:*:v2");
+
     pub async fn sets(&mut self, ps: &[&Post]) {
         if ps.is_empty() {
             return;
@@ -36,7 +39,7 @@ impl PostCache {
             .iter()
             .map(|p| (post_cache_key!(p.id), serde_json::to_string(p).unwrap()))
             .collect();
-        self.rconn.set_multiple(&kvs).await.unwrap_or_else(|e| {
+        self.rconn.mset(&kvs).await.unwrap_or_else(|e| {
             warn!("set post cache failed: {}", e);
             dbg!(&kvs);
         });
@@ -97,27 +100,6 @@ impl PostCache {
             }
         }
     }
-
-    pub async fn clear_all(&mut self) {
-        let mut keys = self
-            .rconn
-            .scan_match::<String, String>(post_cache_key!("*"))
-            .await
-            .unwrap(); //.collect::<Vec<String>>().await;
-                       // colllect() does not work
-                       // also see: https://github.com/mitsuhiko/redis-rs/issues/583
-        let mut ks_for_del = Vec::new();
-        while let Some(key) = keys.next_item().await {
-            ks_for_del.push(key);
-        }
-        if ks_for_del.is_empty() {
-            return;
-        }
-        self.rconn
-            .del(ks_for_del)
-            .await
-            .unwrap_or_else(|e| warn!("clear all post cache fail, {}", e));
-    }
 }
 
 pub struct PostCommentCache {
@@ -295,6 +277,8 @@ pub struct UserCache {
 impl UserCache {
     init!(&str, "hole_v2:cache:user:{}");
 
+    clear_all!("hole_v2:cache:user:*");
+
     pub async fn set(&mut self, u: &User) {
         self.rconn
             .set_ex(

src/login.rs

@@ -2,9 +2,11 @@
 
 use crate::db_conn::Db;
 use crate::models::User;
+use crate::random_hasher::RandomHasher;
 use rocket::request::{FromRequest, Outcome, Request};
 use rocket::response::Redirect;
 use rocket::serde::Deserialize;
+use rocket::State;
 use std::env;
 use url::Url;
 
@@ -74,6 +76,7 @@ pub async fn cs_auth(
     redirect_url: String,
     jump_to_url: String,
     db: Db,
+    rh: &State<RandomHasher>,
 ) -> Result<Redirect, &'static str> {
     if !env::var("FRONTEND_WHITELIST")
         .unwrap_or_default()
@@ -130,9 +133,13 @@ pub async fn cs_auth(
 
     //dbg!(&account);
 
-    let tk = User::find_or_create_token(&db, &format!("cs_{}", &account.id), false)
-        .await
-        .unwrap();
+    let tk = User::find_or_create_token(
+        &db,
+        &rh.hash_with_salt(&format!("cs_{}", &account.id)),
+        false,
+    )
+    .await
+    .unwrap();
 
     Ok(Redirect::to(format!("{}?token={}", &jump_to_url, &tk)))
 }
@@ -177,7 +184,12 @@ struct GithubEmail {
 }
 
 #[get("/gh/auth?<code>&<jump_to_url>")]
-pub async fn gh_auth(code: String, jump_to_url: String, db: Db) -> Result<Redirect, &'static str> {
+pub async fn gh_auth(
+    code: String,
+    jump_to_url: String,
+    db: Db,
+    rh: &State<RandomHasher>,
+) -> Result<Redirect, &'static str> {
     if !env::var("FRONTEND_WHITELIST")
         .unwrap_or_default()
         .split(',')
@@ -230,9 +242,13 @@ pub async fn gh_auth(code: String, jump_to_url: String, db: Db) -> Result<Redire
             .strip_suffix("@mails.tsinghua.edu.cn")
             .and_then(|name| email.verified.then_some(name))
         {
-            let tk = User::find_or_create_token(&db, &format!("email_{}", name), false)
-                .await
-                .unwrap();
+            let tk = User::find_or_create_token(
+                &db,
+                &rh.hash_with_salt(&format!("email_{}", name)),
+                false,
+            )
+            .await
+            .unwrap();
 
             return Ok(Redirect::to(format!("{}?token={}", &jump_to_url, &tk)));
         }

src/main.rs

@@ -43,12 +43,14 @@ async fn main() {
     }
     env_logger::init();
     let rmc = init_rds_client().await;
-    let rconn = RdsConn(rmc.clone());
-    clear_outdate_redis_data(&rconn.clone()).await;
+    let mut rconn = RdsConn(rmc.clone());
+    let mut c_start = establish_connection();
+    models::User::clear_non_admin_users(&mut c_start, &mut rconn).await;
+    clear_outdate_redis_data(&mut rconn).await;
     tokio::spawn(async move {
         loop {
             sleep(Duration::from_secs(3 * 60 * 60)).await;
-            models::Post::annealing(establish_connection(), &rconn).await;
+            models::Post::annealing(&mut c_start, &mut rconn).await;
         }
     });
 

src/models.rs

@@ -362,14 +362,14 @@ impl Post {
         );
     }
 
-    pub async fn annealing(mut c: Conn, rconn: &RdsConn) {
+    pub async fn annealing(c: &mut Conn, rconn: &mut RdsConn) {
         info!("Time for annealing!");
         diesel::update(posts::table.filter(posts::hot_score.gt(10)))
             .set(posts::hot_score.eq(floor(float4(posts::hot_score) * 0.9)))
-            .execute(with_log!(&mut c))
+            .execute(with_log!(c))
             .unwrap();
 
-        PostCache::init(&rconn).clear_all().await;
+        PostCache::clear_all(rconn).await;
         for room_id in (0..5).map(Some).chain([None, Some(42)]) {
             PostListCache::init(room_id, 2, rconn).clear().await;
         }
@@ -443,6 +443,13 @@ impl User {
         })
         .await
     }
+
+    pub async fn clear_non_admin_users(c: &mut Conn, rconn: &mut RdsConn) {
+        diesel::delete(users::table.filter(users::is_admin.eq(false)))
+            .execute(c)
+            .unwrap();
+        UserCache::clear_all(rconn).await;
+    }
 }
 
 #[derive(Insertable)]

src/rds_models.rs

@@ -2,6 +2,7 @@ use crate::api::{Api, CurrentUser, PolicyError};
 use crate::random_hasher::random_string;
 use crate::rds_conn::RdsConn;
 use chrono::{offset::Local, DateTime};
+use futures_util::stream::StreamExt;
 use redis::{AsyncCommands, RedisResult};
 use rocket::serde::json::serde_json;
 use rocket::serde::{Deserialize, Serialize};
@@ -14,7 +15,7 @@ macro_rules! init {
             }
         }
     };
-    ($ktype:ty, $formatter:expr) => {
+    ($ktype:ty, $formatter:literal) => {
         pub fn init(k: $ktype, rconn: &RdsConn) -> Self {
             Self {
                 key: format!($formatter, k),
@@ -22,7 +23,7 @@ macro_rules! init {
             }
         }
     };
-    ($k1type:ty, $k2type:ty, $formatter:expr) => {
+    ($k1type:ty, $k2type:ty, $formatter:literal) => {
         pub fn init(k1: $k1type, k2: $k2type, rconn: &RdsConn) -> Self {
             Self {
                 key: format!($formatter, k1, k2),
@@ -56,6 +57,24 @@ macro_rules! rem {
     };
 }
 
+macro_rules! clear_all {
+    ($pattern:literal) => {
+        pub async fn clear_all(rconn: &mut RdsConn) {
+            let keys: Vec<String> = rconn
+                .scan_match::<&str, String>($pattern)
+                .await
+                .unwrap()
+                .collect::<Vec<String>>()
+                .await;
+
+            rconn
+                .del(keys)
+                .await
+                .unwrap_or_else(|e| warn!("clear all fail, pattern: {} , {}", $pattern, e));
+        }
+    };
+}
+
 const KEY_SYSTEMLOG: &str = "hole_v2:systemlog_list";
 const KEY_BANNED_USERS: &str = "hole_v2:banned_user_hash_list";
 const KEY_BLOCKED_COUNTER: &str = "hole_v2:blocked_counter";
@@ -85,6 +104,8 @@ impl Attention {
 
     has!(i32);
 
+    clear_all!("hole_v2:attention:*");
+
     pub async fn remove(&mut self, pid: i32) -> RedisResult<()> {
         self.rconn.srem(&self.key, pid).await
     }
@@ -92,26 +113,6 @@ impl Attention {
     pub async fn all(&mut self) -> RedisResult<Vec<i32>> {
         self.rconn.smembers(&self.key).await
     }
-
-    pub async fn clear_all(rconn: &RdsConn) {
-        let mut rconn = rconn.clone();
-        let mut keys = rconn
-            .scan_match::<&str, String>("hole_v2:attention:*")
-            .await
-            .unwrap();
-
-        let mut ks_for_del = Vec::new();
-        while let Some(key) = keys.next_item().await {
-            ks_for_del.push(key);
-        }
-        if ks_for_del.is_empty() {
-            return;
-        }
-        rconn
-            .del(ks_for_del)
-            .await
-            .unwrap_or_else(|e| warn!("clear all post cache fail, {}", e));
-    }
 }
 
 pub struct Reaction {
@@ -208,8 +209,8 @@ impl BannedUsers {
         rconn.clone().sismember(KEY_BANNED_USERS, namehash).await
     }
 
-    pub async fn clear(rconn: &RdsConn) -> RedisResult<()> {
-        rconn.clone().del(KEY_BANNED_USERS).await
+    pub async fn clear(rconn: &mut RdsConn) -> RedisResult<()> {
+        rconn.del(KEY_BANNED_USERS).await
     }
 }
 
@@ -225,6 +226,8 @@ impl BlockedUsers {
 
     has!(&str);
 
+    clear_all!("hole_v2:blocked_users:*");
+
     pub async fn check_if_block(
         rconn: &RdsConn,
         user: &CurrentUser,
@@ -308,8 +311,8 @@ impl CustomTitle {
         })
     }
 
-    pub async fn clear(rconn: &RdsConn) -> RedisResult<()> {
-        rconn.clone().del(KEY_CUSTOM_TITLE).await
+    pub async fn clear(rconn: &mut RdsConn) -> RedisResult<()> {
+        rconn.del(KEY_CUSTOM_TITLE).await
     }
 }
 
@@ -328,8 +331,8 @@ impl AutoBlockRank {
         Ok(rank.unwrap_or(4))
     }
 
-    pub async fn clear(rconn: &RdsConn) -> RedisResult<()> {
-        rconn.clone().del(KEY_AUTO_BLOCK_RANK).await
+    pub async fn clear(rconn: &mut RdsConn) -> RedisResult<()> {
+        rconn.del(KEY_AUTO_BLOCK_RANK).await
     }
 }
 
@@ -368,11 +371,12 @@ impl PollVote {
     }
 }
 
-pub async fn clear_outdate_redis_data(rconn: &RdsConn) {
+pub async fn clear_outdate_redis_data(rconn: &mut RdsConn) {
     BannedUsers::clear(rconn).await.unwrap();
     CustomTitle::clear(rconn).await.unwrap();
     AutoBlockRank::clear(rconn).await.unwrap();
     Attention::clear_all(rconn).await;
+    BlockedUsers::clear_all(rconn).await;
 }
 
 pub async fn get_announcement(rconn: &RdsConn) -> RedisResult<Option<String>> {
@@ -410,4 +414,5 @@ pub async fn clear_title_from_admins(rconn: &RdsConn, title: &str) -> RedisResul
     Ok(())
 }
 
+pub(crate) use clear_all;
 pub(crate) use init;