systemprompt_database/repository/
cleanup.rs

1//! Periodic cleanup queries for orphaned rows and expired tokens.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use sqlx::PgPool;
7
8use crate::error::DatabaseResult;
9
10#[derive(Debug)]
11pub struct CleanupRepository {
12    write_pool: PgPool,
13}
14
15impl CleanupRepository {
16    pub const fn new(pool: PgPool) -> Self {
17        Self { write_pool: pool }
18    }
19
20    pub const fn new_with_write_pool(write_pool: PgPool) -> Self {
21        Self { write_pool }
22    }
23
24    pub async fn delete_orphaned_logs(&self) -> DatabaseResult<u64> {
25        let result = sqlx::query!(
26            r#"
27            DELETE FROM logs
28            WHERE user_id IS NOT NULL
29              AND user_id NOT IN (SELECT id FROM users)
30            "#
31        )
32        .execute(&self.write_pool)
33        .await?;
34        Ok(result.rows_affected())
35    }
36
37    pub async fn delete_orphaned_mcp_executions(&self) -> DatabaseResult<u64> {
38        let result = sqlx::query!(
39            r#"
40            DELETE FROM mcp_tool_executions
41            WHERE context_id IS NOT NULL
42              AND context_id NOT IN (SELECT context_id FROM user_contexts)
43            "#
44        )
45        .execute(&self.write_pool)
46        .await?;
47        Ok(result.rows_affected())
48    }
49
50    pub async fn delete_old_logs(&self, days: i32) -> DatabaseResult<u64> {
51        let cutoff = chrono::Utc::now() - chrono::Duration::days(i64::from(days));
52        let result = sqlx::query!("DELETE FROM logs WHERE timestamp < $1", cutoff)
53            .execute(&self.write_pool)
54            .await?;
55        Ok(result.rows_affected())
56    }
57
58    pub async fn count_old_logs(&self, days: i32) -> DatabaseResult<i64> {
59        let cutoff = chrono::Utc::now() - chrono::Duration::days(i64::from(days));
60        let count = sqlx::query_scalar!(
61            r#"SELECT COUNT(*) as "count!" FROM logs WHERE timestamp < $1"#,
62            cutoff
63        )
64        .fetch_one(&self.write_pool)
65        .await?;
66        Ok(count)
67    }
68
69    pub async fn delete_expired_oauth_tokens(&self) -> DatabaseResult<u64> {
70        let result = sqlx::query!("DELETE FROM oauth_refresh_tokens WHERE expires_at < NOW()")
71            .execute(&self.write_pool)
72            .await?;
73        Ok(result.rows_affected())
74    }
75
76    pub async fn delete_expired_oauth_codes(&self) -> DatabaseResult<u64> {
77        let result = sqlx::query!(
78            "DELETE FROM oauth_auth_codes WHERE expires_at < NOW() OR used_at IS NOT NULL"
79        )
80        .execute(&self.write_pool)
81        .await?;
82        Ok(result.rows_affected())
83    }
84
85    pub async fn delete_expired_oauth_state_bindings(&self) -> DatabaseResult<u64> {
86        let result = sqlx::query!("DELETE FROM oauth_state_bindings WHERE expires_at < NOW()")
87            .execute(&self.write_pool)
88            .await?;
89        Ok(result.rows_affected())
90    }
91
92    pub async fn delete_expired_oauth_jti_revocations(&self) -> DatabaseResult<u64> {
93        let result = sqlx::query!("DELETE FROM oauth_jti_revocations WHERE exp < NOW()")
94            .execute(&self.write_pool)
95            .await?;
96        Ok(result.rows_affected())
97    }
98
99    pub async fn delete_expired_id_jag_replays(&self) -> DatabaseResult<u64> {
100        let result = sqlx::query!("DELETE FROM id_jag_replay WHERE expires_at < NOW()")
101            .execute(&self.write_pool)
102            .await?;
103        Ok(result.rows_affected())
104    }
105}
systemprompt_database/repository/cleanup.rs

systemprompt_database/repository/
cleanup.rs
