src/scheduled_tasks.rs

   1 use chrono::NaiveDateTime;
   2 use clokwerk::{Scheduler, TimeUnits as CTimeUnits};
   3 use diesel::{
   4   dsl::{now, IntervalDsl},
   5   sql_types::{Integer, Timestamp},
   6   Connection,
   7   ExpressionMethods,
   8   NullableExpressionMethods,
   9   QueryDsl,
  10   QueryableByName,
  11 };
  12 // Import week days and WeekDay
  13 use diesel::{sql_query, PgConnection, RunQueryDsl};
  14 use lemmy_api_common::context::LemmyContext;
  15 use lemmy_db_schema::{
  16   schema::{activity, captcha_answer, comment, community_person_ban, instance, person, post},
  17   source::instance::{Instance, InstanceForm},
  18   utils::{naive_now, DELETED_REPLACEMENT_TEXT},
  19 };
  20 use lemmy_routes::nodeinfo::NodeInfo;
  21 use lemmy_utils::{error::LemmyError, REQWEST_TIMEOUT};
  22 use reqwest::blocking::Client;
  23 use std::{thread, time::Duration};
  24 use tracing::{error, info};
  25
  26 /// Schedules various cleanup tasks for lemmy in a background thread
  27 pub fn setup(
  28   db_url: String,
  29   user_agent: String,
  30   context_1: LemmyContext,
  31 ) -> Result<(), LemmyError> {
  32   // Setup the connections
  33   let mut scheduler = Scheduler::new();
  34
  35   startup_jobs(&db_url);
  36
  37   // Update active counts every hour
  38   let url = db_url.clone();
  39   scheduler.every(CTimeUnits::hour(1)).run(move || {
  40     let mut conn = PgConnection::establish(&url).expect("could not establish connection");
  41     active_counts(&mut conn);
  42     update_banned_when_expired(&mut conn);
  43   });
  44
  45   // Update hot ranks every 15 minutes
  46   let url = db_url.clone();
  47   scheduler.every(CTimeUnits::minutes(15)).run(move || {
  48     let mut conn = PgConnection::establish(&url).expect("could not establish connection");
  49     update_hot_ranks(&mut conn, true);
  50   });
  51
  52   // Delete any captcha answers older than ten minutes, every ten minutes
  53   let url = db_url.clone();
  54   scheduler.every(CTimeUnits::minutes(10)).run(move || {
  55     let mut conn = PgConnection::establish(&url).expect("could not establish connection");
  56     delete_expired_captcha_answers(&mut conn);
  57   });
  58
  59   // Clear old activities every week
  60   let url = db_url.clone();
  61   scheduler.every(CTimeUnits::weeks(1)).run(move || {
  62     let mut conn = PgConnection::establish(&url).expect("could not establish connection");
  63     clear_old_activities(&mut conn);
  64   });
  65
  66   // Remove old rate limit buckets after 1 to 2 hours of inactivity
  67   scheduler.every(CTimeUnits::hour(1)).run(move || {
  68     let hour = Duration::from_secs(3600);
  69     context_1.settings_updated_channel().remove_older_than(hour);
  70   });
  71
  72   // Overwrite deleted & removed posts and comments every day
  73   let url = db_url.clone();
  74   scheduler.every(CTimeUnits::days(1)).run(move || {
  75     let mut conn = PgConnection::establish(&url).expect("could not establish connection");
  76     overwrite_deleted_posts_and_comments(&mut conn);
  77   });
  78
  79   // Update the Instance Software
  80   scheduler.every(CTimeUnits::days(1)).run(move || {
  81     let mut conn = PgConnection::establish(&db_url).expect("could not establish connection");
  82     update_instance_software(&mut conn, &user_agent);
  83   });
  84
  85   // Manually run the scheduler in an event loop
  86   loop {
  87     scheduler.run_pending();
  88     thread::sleep(Duration::from_millis(1000));
  89   }
  90 }
  91
  92 /// Run these on server startup
  93 fn startup_jobs(db_url: &str) {
  94   let mut conn = PgConnection::establish(db_url).expect("could not establish connection");
  95   active_counts(&mut conn);
  96   update_hot_ranks(&mut conn, false);
  97   update_banned_when_expired(&mut conn);
  98   clear_old_activities(&mut conn);
  99   overwrite_deleted_posts_and_comments(&mut conn);
 100 }
 101
 102 /// Update the hot_rank columns for the aggregates tables
 103 /// Runs in batches until all necessary rows are updated once
 104 fn update_hot_ranks(conn: &mut PgConnection, last_week_only: bool) {
 105   let process_start_time = if last_week_only {
 106     info!("Updating hot ranks for last week...");
 107     naive_now() - chrono::Duration::days(7)
 108   } else {
 109     info!("Updating hot ranks for all history...");
 110     NaiveDateTime::from_timestamp_opt(0, 0).expect("0 timestamp creation")
 111   };
 112
 113   process_hot_ranks_in_batches(
 114     conn,
 115     "post_aggregates",
 116     "SET hot_rank = hot_rank(a.score, a.published),
 117          hot_rank_active = hot_rank(a.score, a.newest_comment_time_necro)",
 118     process_start_time,
 119   );
 120
 121   process_hot_ranks_in_batches(
 122     conn,
 123     "comment_aggregates",
 124     "SET hot_rank = hot_rank(a.score, a.published)",
 125     process_start_time,
 126   );
 127
 128   process_hot_ranks_in_batches(
 129     conn,
 130     "community_aggregates",
 131     "SET hot_rank = hot_rank(a.subscribers, a.published)",
 132     process_start_time,
 133   );
 134
 135   info!("Finished hot ranks update!");
 136 }
 137
 138 #[derive(QueryableByName)]
 139 struct HotRanksUpdateResult {
 140   #[diesel(sql_type = Timestamp)]
 141   published: NaiveDateTime,
 142 }
 143
 144 /// Runs the hot rank update query in batches until all rows after `process_start_time` have been
 145 /// processed.
 146 /// In `set_clause`, "a" will refer to the current aggregates table.
 147 /// Locked rows are skipped in order to prevent deadlocks (they will likely get updated on the next
 148 /// run)
 149 fn process_hot_ranks_in_batches(
 150   conn: &mut PgConnection,
 151   table_name: &str,
 152   set_clause: &str,
 153   process_start_time: NaiveDateTime,
 154 ) {
 155   let update_batch_size = 1000; // Bigger batches than this tend to cause seq scans
 156   let mut previous_batch_result = Some(process_start_time);
 157   while let Some(previous_batch_last_published) = previous_batch_result {
 158     // Raw `sql_query` is used as a performance optimization - Diesel does not support doing this
 159     // in a single query (neither as a CTE, nor using a subquery)
 160     let result = sql_query(format!(
 161       r#"WITH batch AS (SELECT a.id
 162                FROM {aggregates_table} a
 163                WHERE a.published > $1
 164                ORDER BY a.published
 165                LIMIT $2
 166                FOR UPDATE SKIP LOCKED)
 167          UPDATE {aggregates_table} a {set_clause}
 168              FROM batch WHERE a.id = batch.id RETURNING a.published;
 169     "#,
 170       aggregates_table = table_name,
 171       set_clause = set_clause
 172     ))
 173     .bind::<Timestamp, _>(previous_batch_last_published)
 174     .bind::<Integer, _>(update_batch_size)
 175     .get_results::<HotRanksUpdateResult>(conn);
 176
 177     match result {
 178       Ok(updated_rows) => previous_batch_result = updated_rows.last().map(|row| row.published),
 179       Err(e) => {
 180         error!("Failed to update {} hot_ranks: {}", table_name, e);
 181         break;
 182       }
 183     }
 184   }
 185   info!(
 186     "Finished process_hot_ranks_in_batches execution for {}",
 187     table_name
 188   );
 189 }
 190
 191 fn delete_expired_captcha_answers(conn: &mut PgConnection) {
 192   match diesel::delete(
 193     captcha_answer::table.filter(captcha_answer::published.lt(now - IntervalDsl::minutes(10))),
 194   )
 195   .execute(conn)
 196   {
 197     Ok(_) => {
 198       info!("Done.");
 199     }
 200     Err(e) => {
 201       error!("Failed to clear old captcha answers: {}", e)
 202     }
 203   }
 204 }
 205
 206 /// Clear old activities (this table gets very large)
 207 fn clear_old_activities(conn: &mut PgConnection) {
 208   info!("Clearing old activities...");
 209   match diesel::delete(activity::table.filter(activity::published.lt(now - 6.months())))
 210     .execute(conn)
 211   {
 212     Ok(_) => {
 213       info!("Done.");
 214     }
 215     Err(e) => {
 216       error!("Failed to clear old activities: {}", e)
 217     }
 218   }
 219 }
 220
 221 /// overwrite posts and comments 30d after deletion
 222 fn overwrite_deleted_posts_and_comments(conn: &mut PgConnection) {
 223   info!("Overwriting deleted posts...");
 224   match diesel::update(
 225     post::table
 226       .filter(post::deleted.eq(true))
 227       .filter(post::updated.lt(now.nullable() - 1.months()))
 228       .filter(post::body.ne(DELETED_REPLACEMENT_TEXT)),
 229   )
 230   .set((
 231     post::body.eq(DELETED_REPLACEMENT_TEXT),
 232     post::name.eq(DELETED_REPLACEMENT_TEXT),
 233   ))
 234   .execute(conn)
 235   {
 236     Ok(_) => {
 237       info!("Done.");
 238     }
 239     Err(e) => {
 240       error!("Failed to overwrite deleted posts: {}", e)
 241     }
 242   }
 243
 244   info!("Overwriting deleted comments...");
 245   match diesel::update(
 246     comment::table
 247       .filter(comment::deleted.eq(true))
 248       .filter(comment::updated.lt(now.nullable() - 1.months()))
 249       .filter(comment::content.ne(DELETED_REPLACEMENT_TEXT)),
 250   )
 251   .set(comment::content.eq(DELETED_REPLACEMENT_TEXT))
 252   .execute(conn)
 253   {
 254     Ok(_) => {
 255       info!("Done.");
 256     }
 257     Err(e) => {
 258       error!("Failed to overwrite deleted comments: {}", e)
 259     }
 260   }
 261 }
 262
 263 /// Re-calculate the site and community active counts every 12 hours
 264 fn active_counts(conn: &mut PgConnection) {
 265   info!("Updating active site and community aggregates ...");
 266
 267   let intervals = vec![
 268     ("1 day", "day"),
 269     ("1 week", "week"),
 270     ("1 month", "month"),
 271     ("6 months", "half_year"),
 272   ];
 273
 274   for i in &intervals {
 275     let update_site_stmt = format!(
 276       "update site_aggregates set users_active_{} = (select * from site_aggregates_activity('{}')) where site_id = 1",
 277       i.1, i.0
 278     );
 279     match sql_query(update_site_stmt).execute(conn) {
 280       Ok(_) => {}
 281       Err(e) => {
 282         error!("Failed to update site stats: {}", e)
 283       }
 284     }
 285
 286     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);
 287     match sql_query(update_community_stmt).execute(conn) {
 288       Ok(_) => {}
 289       Err(e) => {
 290         error!("Failed to update community stats: {}", e)
 291       }
 292     }
 293   }
 294
 295   info!("Done.");
 296 }
 297
 298 /// Set banned to false after ban expires
 299 fn update_banned_when_expired(conn: &mut PgConnection) {
 300   info!("Updating banned column if it expires ...");
 301
 302   match diesel::update(
 303     person::table
 304       .filter(person::banned.eq(true))
 305       .filter(person::ban_expires.lt(now)),
 306   )
 307   .set(person::banned.eq(false))
 308   .execute(conn)
 309   {
 310     Ok(_) => {}
 311     Err(e) => {
 312       error!("Failed to update person.banned when expires: {}", e)
 313     }
 314   }
 315   match diesel::delete(community_person_ban::table.filter(community_person_ban::expires.lt(now)))
 316     .execute(conn)
 317   {
 318     Ok(_) => {}
 319     Err(e) => {
 320       error!("Failed to remove community_ban expired rows: {}", e)
 321     }
 322   }
 323 }
 324
 325 /// Updates the instance software and version
 326 fn update_instance_software(conn: &mut PgConnection, user_agent: &str) {
 327   info!("Updating instances software and versions...");
 328
 329   let client = match Client::builder()
 330     .user_agent(user_agent)
 331     .timeout(REQWEST_TIMEOUT)
 332     .build()
 333   {
 334     Ok(client) => client,
 335     Err(e) => {
 336       error!("Failed to build reqwest client: {}", e);
 337       return;
 338     }
 339   };
 340
 341   let instances = match instance::table.get_results::<Instance>(conn) {
 342     Ok(instances) => instances,
 343     Err(e) => {
 344       error!("Failed to get instances: {}", e);
 345       return;
 346     }
 347   };
 348
 349   for instance in instances {
 350     let node_info_url = format!("https://{}/nodeinfo/2.0.json", instance.domain);
 351
 352     // Skip it if it can't connect
 353     let res = client
 354       .get(&node_info_url)
 355       .send()
 356       .ok()
 357       .and_then(|t| t.json::<NodeInfo>().ok());
 358
 359     if let Some(node_info) = res {
 360       let software = node_info.software.as_ref();
 361       let form = InstanceForm::builder()
 362         .domain(instance.domain)
 363         .software(software.and_then(|s| s.name.clone()))
 364         .version(software.and_then(|s| s.version.clone()))
 365         .updated(Some(naive_now()))
 366         .build();
 367
 368       match diesel::update(instance::table.find(instance.id))
 369         .set(form)
 370         .execute(conn)
 371       {
 372         Ok(_) => {
 373           info!("Done.");
 374         }
 375         Err(e) => {
 376           error!("Failed to update site instance software: {}", e);
 377           return;
 378         }
 379       }
 380     }
 381   }
 382 }
 383
 384 #[cfg(test)]
 385 mod tests {
 386   use lemmy_routes::nodeinfo::NodeInfo;
 387   use reqwest::Client;
 388
 389   #[tokio::test]
 390   #[ignore]
 391   async fn test_nodeinfo() {
 392     let client = Client::builder().build().unwrap();
 393     let lemmy_ml_nodeinfo = client
 394       .get("https://lemmy.ml/nodeinfo/2.0.json")
 395       .send()
 396       .await
 397       .unwrap()
 398       .json::<NodeInfo>()
 399       .await
 400       .unwrap();
 401
 402     assert_eq!(lemmy_ml_nodeinfo.software.unwrap().name.unwrap(), "lemmy");
 403   }
 404 }