src/scheduled_tasks.rs

   1 // Scheduler, and trait for .seconds(), .minutes(), etc.
   2 use clokwerk::{Scheduler, TimeUnits};
   3 // Import week days and WeekDay
   4 use diesel::{sql_query, PgConnection, RunQueryDsl};
   5 use lemmy_db_queries::{source::activity::Activity_, DbPool};
   6 use lemmy_db_schema::source::activity::Activity;
   7 use log::info;
   8 use std::{thread, time::Duration};
   9
  10 /// Schedules various cleanup tasks for lemmy in a background thread
  11 pub fn setup(pool: DbPool) {
  12   let mut scheduler = Scheduler::new();
  13
  14   let conn = pool.get().unwrap();
  15   active_counts(&conn);
  16   reindex_aggregates_tables(&conn);
  17   scheduler.every(1.hour()).run(move || {
  18     active_counts(&conn);
  19     reindex_aggregates_tables(&conn);
  20   });
  21
  22   let conn = pool.get().unwrap();
  23   clear_old_activities(&conn);
  24   scheduler.every(1.weeks()).run(move || {
  25     clear_old_activities(&conn);
  26   });
  27
  28   // Manually run the scheduler in an event loop
  29   loop {
  30     scheduler.run_pending();
  31     thread::sleep(Duration::from_millis(1000));
  32   }
  33 }
  34
  35 /// Reindex the aggregates tables every one hour
  36 /// This is necessary because hot_rank is actually a mutable function:
  37 /// https://dba.stackexchange.com/questions/284052/how-to-create-an-index-based-on-a-time-based-function-in-postgres?noredirect=1#comment555727_284052
  38 fn reindex_aggregates_tables(conn: &PgConnection) {
  39   for table_name in &[
  40     "post_aggregates",
  41     "comment_aggregates",
  42     "community_aggregates",
  43   ] {
  44     reindex_table(&conn, &table_name);
  45   }
  46 }
  47
  48 fn reindex_table(conn: &PgConnection, table_name: &str) {
  49   info!("Reindexing table {} ...", table_name);
  50   let query = format!("reindex table concurrently {}", table_name);
  51   sql_query(query).execute(conn).expect("reindex table");
  52   info!("Done.");
  53 }
  54
  55 /// Clear old activities (this table gets very large)
  56 fn clear_old_activities(conn: &PgConnection) {
  57   info!("Clearing old activities...");
  58   Activity::delete_olds(&conn).expect("clear old activities");
  59   info!("Done.");
  60 }
  61
  62 /// Re-calculate the site and community active counts every 12 hours
  63 fn active_counts(conn: &PgConnection) {
  64   info!("Updating active site and community aggregates ...");
  65
  66   let intervals = vec![
  67     ("1 day", "day"),
  68     ("1 week", "week"),
  69     ("1 month", "month"),
  70     ("6 months", "half_year"),
  71   ];
  72
  73   for i in &intervals {
  74     let update_site_stmt = format!(
  75       "update site_aggregates set users_active_{} = (select * from site_aggregates_activity('{}'))",
  76       i.1, i.0
  77     );
  78     sql_query(update_site_stmt)
  79       .execute(conn)
  80       .expect("update site stats");
  81
  82     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);
  83     sql_query(update_community_stmt)
  84       .execute(conn)
  85       .expect("update community stats");
  86   }
  87
  88   info!("Done.");
  89 }