src/scheduled_tasks.rs

   1 use clokwerk::{Scheduler, TimeUnits as CTimeUnits};
   2 use diesel::{
   3   dsl::{now, IntervalDsl},
   4   Connection,
   5   ExpressionMethods,
   6   QueryDsl,
   7 };
   8 // Import week days and WeekDay
   9 use diesel::{sql_query, PgConnection, RunQueryDsl};
  10 use lemmy_db_schema::{
  11   schema::{
  12     activity,
  13     comment_aggregates,
  14     community_aggregates,
  15     community_person_ban,
  16     instance,
  17     person,
  18     post_aggregates,
  19   },
  20   source::instance::{Instance, InstanceForm},
  21   utils::{functions::hot_rank, naive_now},
  22 };
  23 use lemmy_routes::nodeinfo::NodeInfo;
  24 use lemmy_utils::{error::LemmyError, REQWEST_TIMEOUT};
  25 use reqwest::blocking::Client;
  26 use std::{thread, time::Duration};
  27 use tracing::info;
  28
  29 /// Schedules various cleanup tasks for lemmy in a background thread
  30 pub fn setup(db_url: String, user_agent: String) -> Result<(), LemmyError> {
  31   // Setup the connections
  32   let mut scheduler = Scheduler::new();
  33
  34   let mut conn_1 = PgConnection::establish(&db_url).expect("could not establish connection");
  35   let mut conn_2 = PgConnection::establish(&db_url).expect("could not establish connection");
  36   let mut conn_3 = PgConnection::establish(&db_url).expect("could not establish connection");
  37   let mut conn_4 = PgConnection::establish(&db_url).expect("could not establish connection");
  38
  39   // Run on startup
  40   active_counts(&mut conn_1);
  41   update_hot_ranks(&mut conn_1, false);
  42   update_banned_when_expired(&mut conn_1);
  43   clear_old_activities(&mut conn_1);
  44
  45   // Update active counts every hour
  46   scheduler.every(CTimeUnits::hour(1)).run(move || {
  47     active_counts(&mut conn_1);
  48     update_banned_when_expired(&mut conn_1);
  49   });
  50
  51   // Update hot ranks every 5 minutes
  52   scheduler.every(CTimeUnits::minutes(5)).run(move || {
  53     update_hot_ranks(&mut conn_2, true);
  54   });
  55
  56   // Clear old activities every week
  57   scheduler.every(CTimeUnits::weeks(1)).run(move || {
  58     clear_old_activities(&mut conn_3);
  59   });
  60
  61   scheduler.every(CTimeUnits::days(1)).run(move || {
  62     update_instance_software(&mut conn_4, &user_agent);
  63   });
  64
  65   // Manually run the scheduler in an event loop
  66   loop {
  67     scheduler.run_pending();
  68     thread::sleep(Duration::from_millis(1000));
  69   }
  70 }
  71
  72 /// Update the hot_rank columns for the aggregates tables
  73 fn update_hot_ranks(conn: &mut PgConnection, last_week_only: bool) {
  74   let mut post_update = diesel::update(post_aggregates::table).into_boxed();
  75   let mut comment_update = diesel::update(comment_aggregates::table).into_boxed();
  76   let mut community_update = diesel::update(community_aggregates::table).into_boxed();
  77
  78   // Only update for the last week of content
  79   if last_week_only {
  80     info!("Updating hot ranks for last week...");
  81     let last_week = now - diesel::dsl::IntervalDsl::weeks(1);
  82
  83     post_update = post_update.filter(post_aggregates::published.gt(last_week));
  84     comment_update = comment_update.filter(comment_aggregates::published.gt(last_week));
  85     community_update = community_update.filter(community_aggregates::published.gt(last_week));
  86   } else {
  87     info!("Updating hot ranks for all history...");
  88   }
  89
  90   post_update
  91     .set((
  92       post_aggregates::hot_rank.eq(hot_rank(post_aggregates::score, post_aggregates::published)),
  93       post_aggregates::hot_rank_active.eq(hot_rank(
  94         post_aggregates::score,
  95         post_aggregates::newest_comment_time_necro,
  96       )),
  97     ))
  98     .execute(conn)
  99     .expect("update post_aggregate hot_ranks");
 100
 101   comment_update
 102     .set(comment_aggregates::hot_rank.eq(hot_rank(
 103       comment_aggregates::score,
 104       comment_aggregates::published,
 105     )))
 106     .execute(conn)
 107     .expect("update comment_aggregate hot_ranks");
 108
 109   community_update
 110     .set(community_aggregates::hot_rank.eq(hot_rank(
 111       community_aggregates::subscribers,
 112       community_aggregates::published,
 113     )))
 114     .execute(conn)
 115     .expect("update community_aggregate hot_ranks");
 116   info!("Done.");
 117 }
 118
 119 /// Clear old activities (this table gets very large)
 120 fn clear_old_activities(conn: &mut PgConnection) {
 121   info!("Clearing old activities...");
 122   diesel::delete(activity::table.filter(activity::published.lt(now - 6.months())))
 123     .execute(conn)
 124     .expect("clear old activities");
 125   info!("Done.");
 126 }
 127
 128 /// Re-calculate the site and community active counts every 12 hours
 129 fn active_counts(conn: &mut PgConnection) {
 130   info!("Updating active site and community aggregates ...");
 131
 132   let intervals = vec![
 133     ("1 day", "day"),
 134     ("1 week", "week"),
 135     ("1 month", "month"),
 136     ("6 months", "half_year"),
 137   ];
 138
 139   for i in &intervals {
 140     let update_site_stmt = format!(
 141       "update site_aggregates set users_active_{} = (select * from site_aggregates_activity('{}'))",
 142       i.1, i.0
 143     );
 144     sql_query(update_site_stmt)
 145       .execute(conn)
 146       .expect("update site stats");
 147
 148     let update_community_stmt = format!("update community_aggregates ca set users_active_{} = mv.count_ from community_aggregates_activity('{}') mv where ca.community_id = mv.community_id_", i.1, i.0);
 149     sql_query(update_community_stmt)
 150       .execute(conn)
 151       .expect("update community stats");
 152   }
 153
 154   info!("Done.");
 155 }
 156
 157 /// Set banned to false after ban expires
 158 fn update_banned_when_expired(conn: &mut PgConnection) {
 159   info!("Updating banned column if it expires ...");
 160
 161   diesel::update(
 162     person::table
 163       .filter(person::banned.eq(true))
 164       .filter(person::ban_expires.lt(now)),
 165   )
 166   .set(person::banned.eq(false))
 167   .execute(conn)
 168   .expect("update person.banned when expires");
 169
 170   diesel::delete(community_person_ban::table.filter(community_person_ban::expires.lt(now)))
 171     .execute(conn)
 172     .expect("remove community_ban expired rows");
 173 }
 174
 175 /// Updates the instance software and version
 176 fn update_instance_software(conn: &mut PgConnection, user_agent: &str) {
 177   info!("Updating instances software and versions...");
 178
 179   let client = Client::builder()
 180     .user_agent(user_agent)
 181     .timeout(REQWEST_TIMEOUT)
 182     .build()
 183     .expect("couldnt build reqwest client");
 184
 185   let instances = instance::table
 186     .get_results::<Instance>(conn)
 187     .expect("no instances found");
 188
 189   for instance in instances {
 190     let node_info_url = format!("https://{}/nodeinfo/2.0.json", instance.domain);
 191
 192     // Skip it if it can't connect
 193     let res = client
 194       .get(&node_info_url)
 195       .send()
 196       .ok()
 197       .and_then(|t| t.json::<NodeInfo>().ok());
 198
 199     if let Some(node_info) = res {
 200       let software = node_info.software.as_ref();
 201       let form = InstanceForm::builder()
 202         .domain(instance.domain)
 203         .software(software.and_then(|s| s.name.clone()))
 204         .version(software.and_then(|s| s.version.clone()))
 205         .updated(Some(naive_now()))
 206         .build();
 207
 208       diesel::update(instance::table.find(instance.id))
 209         .set(form)
 210         .execute(conn)
 211         .expect("update site instance software");
 212     }
 213   }
 214   info!("Done.");
 215 }
 216
 217 #[cfg(test)]
 218 mod tests {
 219   use lemmy_routes::nodeinfo::NodeInfo;
 220   use reqwest::Client;
 221
 222   #[tokio::test]
 223   #[ignore]
 224   async fn test_nodeinfo() {
 225     let client = Client::builder().build().unwrap();
 226     let lemmy_ml_nodeinfo = client
 227       .get("https://lemmy.ml/nodeinfo/2.0.json")
 228       .send()
 229       .await
 230       .unwrap()
 231       .json::<NodeInfo>()
 232       .await
 233       .unwrap();
 234
 235     assert_eq!(lemmy_ml_nodeinfo.software.unwrap().name.unwrap(), "lemmy");
 236   }
 237 }