periodic VPN session & stats purge

Cargo.toml

@@ -21,6 +21,7 @@ defguard_proto = { path = "./crates/defguard_proto", version = "0.0.0" }
 defguard_proxy_manager = { path = "./crates/defguard_proxy_manager", version = "0.0.0" }
 defguard_session_manager = { path = "./crates/defguard_session_manager", version = "0.0.0" }
 defguard_version = { path = "./crates/defguard_version", version = "0.0.0" }
+defguard_vpn_stats_purge = { path = "./crates/defguard_vpn_stats_purge", version = "0.0.0" }
 defguard_web_ui = { path = "./crates/defguard_web_ui", version = "0.0.0" }
 defguard_certs = { path = "./crates/defguard_certs", version = "0.0.0" }
 model_derive = { path = "./crates/model_derive", version = "0.0.0" }

crates/defguard/Cargo.toml

@@ -17,6 +17,7 @@ defguard_mail = { workspace = true }
 defguard_proxy_manager = { workspace = true }
 defguard_session_manager = { workspace = true }
 defguard_version = { workspace = true }
+defguard_vpn_stats_purge = { workspace = true }
 defguard_certs = { workspace = true }
 
 # external dependencies

crates/defguard/src/main.rs

@@ -34,13 +34,13 @@ use defguard_core::{
     init_dev_env, init_vpn_location, run_web_server,
     utility_thread::run_utility_thread,
     version::IncompatibleComponents,
-    wireguard_stats_purge::run_periodic_stats_purge,
 };
 use defguard_event_logger::{message::EventLoggerMessage, run_event_logger};
 use defguard_event_router::{RouterReceiverSet, run_event_router};
 use defguard_mail::{Mail, run_mail_handler};
 use defguard_proxy_manager::{ProxyManager, ProxyTxSet};
 use defguard_session_manager::{events::SessionManagerEvent, run_session_manager};
+use defguard_vpn_stats_purge::run_periodic_stats_purge;
 use secrecy::ExposeSecret;
 use tokio::sync::{
     broadcast,

crates/defguard_common/src/db/models/wireguard_peer_stats.rs

@@ -1,12 +1,8 @@
-use std::time::Duration;
-
-use chrono::{DateTime, NaiveDateTime, TimeDelta, Utc};
-use humantime::format_duration;
+use chrono::NaiveDateTime;
 use ipnetwork::IpNetwork;
 use model_derive::Model;
 use serde::{Deserialize, Serialize};
-use sqlx::{PgExecutor, PgPool, query, query_as, query_scalar};
-use tracing::{debug, info};
+use sqlx::{PgPool, query_as};
 
 use crate::db::{Id, NoId};
 
@@ -28,90 +24,6 @@ pub struct WireguardPeerStats<I = NoId> {
     pub allowed_ips: Option<String>,
 }
 
-impl WireguardPeerStats {
-    /// Delete stats older than a configured threshold.
-    /// This is done to prevent unnecessary table growth.
-    /// At least one record is retained for each device and network combination,
-    /// even when older than set threshold.
-    pub async fn purge_old_stats(
-        pool: &PgPool,
-        stats_purge_threshold: Duration,
-    ) -> Result<(), sqlx::Error> {
-        let start = Utc::now();
-        info!(
-            "Purging stats older than {}",
-            format_duration(stats_purge_threshold)
-        );
-
-        let threshold = (Utc::now()
-            - TimeDelta::from_std(stats_purge_threshold).expect("Failed to parse duration"))
-        .naive_utc();
-        let result = query!(
-            "DELETE FROM wireguard_peer_stats \
-            WHERE collected_at < $1 \
-            AND (device_id, network, collected_at) NOT IN ( \
-                SELECT device_id, network, MAX(collected_at) \
-                FROM wireguard_peer_stats \
-                GROUP BY device_id, network)",
-            threshold
-        )
-        .execute(pool)
-        .await?;
-
-        let end = Utc::now();
-        let rows_count = result.rows_affected();
-
-        info!("Removed {rows_count} old records from wireguard_peer_stats",);
-
-        // Store successful stats purge in database.
-        Self::record_stats_purge(pool, start, end, threshold, rows_count as i64).await?;
-
-        Ok(())
-    }
-
-    // Check how much time has elapsed since last recorded stats purge
-    pub async fn time_since_last_purge<'e, E>(executor: E) -> Result<Option<Duration>, sqlx::Error>
-    where
-        E: PgExecutor<'e>,
-    {
-        debug!("Checking time since last stats purge");
-
-        let timestamp = query_scalar!("SELECT MAX(started_at) FROM wireguard_stats_purge")
-            .fetch_one(executor)
-            .await?;
-
-        match timestamp {
-            Some(timestamp) => {
-                let time_since = Utc::now().signed_duration_since(timestamp.and_utc());
-                let time_since = time_since.to_std().expect("Failed to parse duration");
-                debug!(
-                    "Time since last stats purge: {}",
-                    format_duration(time_since)
-                );
-                Ok(Some(time_since))
-            }
-            None => Ok(None),
-        }
-    }
-
-    async fn record_stats_purge<'e, E>(
-        executor: E,
-        start: DateTime<Utc>,
-        end: DateTime<Utc>,
-        removal_threshold: NaiveDateTime,
-        records_removed: i64,
-    ) -> Result<(), sqlx::Error>
-    where
-        E: PgExecutor<'e>,
-    {
-        debug!("Recording successful stats purge in database");
-        query!("INSERT INTO wireguard_stats_purge (started_at, finished_at, removal_threshold, records_removed) VALUES ($1, $2, $3, $4)",
-        start.naive_utc(), end.naive_utc(), removal_threshold, records_removed).execute(executor).await?;
-
-        Ok(())
-    }
-}
-
 impl WireguardPeerStats<Id> {
     pub async fn fetch_latest(
         conn: &PgPool,
@@ -167,6 +79,8 @@ impl WireguardPeerStats<Id> {
 
 #[cfg(test)]
 mod test {
+    use chrono::Utc;
+
     use super::*;
 
     #[test]

crates/defguard_core/src/lib.rs

@@ -178,7 +178,6 @@ pub mod user_management;
 pub mod utility_thread;
 pub mod version;
 pub mod wg_config;
-pub mod wireguard_stats_purge;
 
 #[macro_use]
 extern crate tracing;

crates/defguard_vpn_stats_purge/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "defguard_vpn_stats_purge"
+version = "0.0.0"
+edition.workspace = true
+license-file.workspace = true
+homepage.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+defguard_common.workspace = true
+
+chrono.workspace = true
+humantime.workspace = true
+sqlx.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+