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