fix(jobs): add a job to clear expired sessions

Author: ImShyMike

rustytime/Cargo.lock

@@ -1,7 +1,7 @@
 [package]
 name = "rustytime-server"
 description = "🕒 blazingly fast time tracking for developers"
-version = "0.17.0"
+version = "0.17.1"
 edition = "2024"
 authors = ["ImShyMike"]
 readme = "../README.md"

rustytime/Cargo.toml

@@ -521,8 +521,8 @@ pub async fn setup(
 
     WorkerBuilder::new("import-worker")
         .backend(import_store)
-        .layer(PrometheusLayer::default())
         .enable_tracing()
+        .layer(PrometheusLayer::default())
         .catch_panic()
         .concurrency(2)
         .data(diesel_pool)

rustytime/src/jobs/import.rs

@@ -155,7 +155,7 @@ fn regenerate_leaderboard_period(
 
 fn cleanup_old_entries(pool: &DbPool) -> Result<(), diesel::result::Error> {
     let mut conn = pool.get().map_err(|e| {
-        tracing::error!(error = ?e, "Failed to get connection for cleanup");
+        tracing::error!(error = ?e, "Failed to get connection for leaderboard cleanup");
         diesel::result::Error::DatabaseError(
             diesel::result::DatabaseErrorKind::Unknown,
             Box::new(e.to_string()),

rustytime/src/jobs/leaderboard.rs

@@ -1,9 +1,11 @@
 pub mod import;
 mod leaderboard;
+mod sessions;
 
 use apalis_postgres::PostgresStorage;
 use axum_prometheus::metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
 use sqlx::PgPool;
+use std::future::Future;
 
 use crate::db::connection::DbPool;
 
@@ -17,15 +19,22 @@ pub async fn setup_jobs(
     sqlx_pool: PgPool,
     diesel_pool: DbPool,
 ) -> (
-    impl std::future::Future<Output = ()>,
-    impl std::future::Future<Output = ()>,
+    impl Future<Output = ()>,
+    impl Future<Output = ()>,
+    impl Future<Output = ()>,
     import::ImportStore,
 ) {
     PostgresStorage::setup(&sqlx_pool).await.unwrap();
 
     let import_store = import::create_storage(&sqlx_pool).await;
     let leaderboard_worker = leaderboard::setup(diesel_pool.clone()).await;
-    let import_worker = import::setup(sqlx_pool, diesel_pool).await;
+    let import_worker = import::setup(sqlx_pool, diesel_pool.clone()).await;
+    let sessions_worker = sessions::setup(diesel_pool).await;
 
-    (leaderboard_worker, import_worker, import_store)
+    (
+        leaderboard_worker,
+        import_worker,
+        sessions_worker,
+        import_store,
+    )
 }

rustytime/src/jobs/mod.rs

@@ -0,0 +1,62 @@
+use std::str::FromStr;
+
+use apalis::{
+    layers::{WorkerBuilderExt, prometheus::PrometheusLayer},
+    prelude::{Data, WorkerBuilder},
+};
+use apalis_cron::{CronStream, Tick};
+use cron::Schedule;
+use tokio::signal::ctrl_c;
+
+use diesel::Connection;
+
+use crate::db::connection::DbPool;
+use crate::models::session::Session;
+
+fn cleanup_expired_sessions(pool: &DbPool) -> Result<(), diesel::result::Error> {
+    let mut conn = pool.get().map_err(|e| {
+        tracing::error!(error = ?e, "Failed to get connection for session cleanup");
+        diesel::result::Error::DatabaseError(
+            diesel::result::DatabaseErrorKind::Unknown,
+            Box::new(e.to_string()),
+        )
+    })?;
+
+    conn.transaction(|conn| {
+        let deleted = Session::delete_expired(conn)?;
+
+        tracing::debug!(deleted, "Cleaned up old leaderboard entries");
+
+        Ok(())
+    })
+}
+
+async fn run_cleanup(_tick: Tick, pool: Data<DbPool>) {
+    if let Err(e) = cleanup_expired_sessions(&pool) {
+        tracing::error!(error = ?e, "Failed to cleanup old leaderboard entries");
+    }
+}
+
+pub async fn setup(diesel_pool: DbPool) -> impl std::future::Future<Output = ()> {
+    let cleanup_schedule =
+        Schedule::from_str("0 0 0 * * *").expect("valid cron: daily at midnight");
+
+    let cleanup_pool = diesel_pool;
+
+    let cleanup_worker = WorkerBuilder::new("sessions-cleanup")
+        .backend(CronStream::new(cleanup_schedule))
+        .enable_tracing()
+        .layer(PrometheusLayer::default())
+        .catch_panic()
+        .data(cleanup_pool)
+        .build(run_cleanup);
+
+    async move {
+        tokio::select! {
+            _ = cleanup_worker.run() => {}
+            _ = ctrl_c() => {
+                tracing::info!("Shutting down session workers");
+            }
+        }
+    }
+}

rustytime/src/jobs/sessions.rs

@@ -145,11 +145,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let sqlx_pool = sqlx::PgPool::connect(&database_url)
         .await
         .expect("Failed to connect to database for jobs");
-    let (leaderboard_worker, import_worker, import_store) =
+    let (leaderboard_worker, import_worker, sessions_worker, import_store) =
         jobs::setup_jobs(sqlx_pool, pool.clone()).await;
     app_state.set_import_store(import_store);
     tokio::spawn(leaderboard_worker);
     tokio::spawn(import_worker);
+    tokio::spawn(sessions_worker);
     info!("✅ Jobs system started");
 
     let rate_period = if is_production {

rustytime/src/main.rs

@@ -8,7 +8,7 @@ use crate::models::user::User;
 use crate::schema::sessions;
 use crate::schema::sessions::dsl;
 
-const SESSION_EXPIRY_DAYS: i64 = 7;
+pub const SESSION_EXPIRY_DAYS: i64 = 7;
 
 #[derive(Queryable, Selectable, Serialize, Deserialize, Debug, Clone)]
 #[diesel(table_name = sessions)]
@@ -92,10 +92,16 @@ impl Session {
     }
 
     #[allow(dead_code)]
+    #[inline(always)]
     pub fn delete_by_user_id(conn: &mut PgConnection, user_id: i32) -> QueryResult<usize> {
         diesel::delete(sessions::table.filter(dsl::user_id.eq(user_id))).execute(conn)
     }
 
+    #[inline(always)]
+    pub fn delete_expired(conn: &mut PgConnection) -> QueryResult<usize> {
+        diesel::delete(sessions::table.filter(dsl::expires_at.le(now))).execute(conn)
+    }
+
     pub fn impersonate(
         conn: &mut PgConnection,
         session_id: Uuid,