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