Initial public release

src/analytics/aggregate.rs

@@ -0,0 +1,140 @@
+use rusqlite::{Connection, params};
+use std::collections::HashMap;
+use crate::db::analytics::{parse_ua, clean_referrer};
+
+// Run aggregation for a specific day
+pub fn aggregate_day(conn: &mut Connection, date: &str) -> rusqlite::Result<()> {
+    let mut visits = Vec::new();
+    {
+        // 1. Fetch all visits on that day
+        let mut stmt = conn.prepare(
+            "SELECT target_type, target_id, user_agent, referer, country, status_code FROM visits WHERE date(timestamp) = ?1;"
+        )?;
+        
+        struct RawVisit {
+            target_type: String,
+            target_id: String,
+            user_agent: String,
+            referer: String,
+            country: String,
+        }
+        
+        let rows = stmt.query_map(params![date], |row| {
+            Ok(RawVisit {
+                target_type: row.get(0)?,
+                target_id: row.get(1)?,
+                user_agent: row.get(2)?,
+                referer: row.get(3)?,
+                country: row.get(4)?,
+            })
+        })?;
+        
+        for r in rows {
+            visits.push(r?);
+        }
+    }
+
+    if visits.is_empty() {
+        return Ok(());
+    }
+
+    // 2. Compute metrics in-memory
+    // Key structure: (target_type, target_id, metric_type, metric_key) -> count
+    let mut aggregates: HashMap<(String, String, String, String), i64> = HashMap::new();
+
+    // Also track total per day (all targets combined) using target_id = "all"
+    for v in visits {
+        let (browser, os, device) = parse_ua(&v.user_agent);
+        let referrer = clean_referrer(&v.referer);
+        let country = if v.country.is_empty() { "Unknown".to_string() } else { v.country.clone() };
+
+        let targets = vec![
+            (v.target_type.clone(), v.target_id.clone()),
+            (v.target_type.clone(), "all".to_string()),
+        ];
+
+        for (t_type, t_id) in targets {
+            // Clicks
+            *aggregates.entry((t_type.clone(), t_id.clone(), "clicks".to_string(), "".to_string())).or_insert(0) += 1;
+            
+            // Country
+            *aggregates.entry((t_type.clone(), t_id.clone(), "country".to_string(), country.clone())).or_insert(0) += 1;
+
+            // Browser
+            *aggregates.entry((t_type.clone(), t_id.clone(), "browser".to_string(), browser.clone())).or_insert(0) += 1;
+
+            // OS
+            *aggregates.entry((t_type.clone(), t_id.clone(), "os".to_string(), os.clone())).or_insert(0) += 1;
+
+            // Device
+            *aggregates.entry((t_type.clone(), t_id.clone(), "device".to_string(), device.clone())).or_insert(0) += 1;
+
+            // Referrer
+            *aggregates.entry((t_type.clone(), t_id.clone(), "referrer".to_string(), referrer.clone())).or_insert(0) += 1;
+        }
+    }
+
+    // 3. Save to database in a transaction
+    let tx = conn.transaction()?;
+    {
+        // Delete old aggregates for this day
+        tx.execute("DELETE FROM daily_summaries WHERE date = ?1;", params![date])?;
+
+        let mut insert_stmt = tx.prepare(
+            "INSERT INTO daily_summaries (date, target_type, target_id, metric_type, metric_key, metric_value)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6);"
+        )?;
+
+        for ((t_type, t_id, m_type, m_key), value) in aggregates {
+            insert_stmt.execute(params![
+                date,
+                t_type,
+                t_id,
+                m_type,
+                m_key,
+                value
+            ])?;
+        }
+    }
+    tx.commit()?;
+
+    // Update monthly and yearly summaries using the daily summaries
+    aggregate_month_from_daily(conn, &date[0..7])?;
+    aggregate_year_from_daily(conn, &date[0..4])?;
+
+    Ok(())
+}
+
+pub fn aggregate_month_from_daily(conn: &mut Connection, year_month: &str) -> rusqlite::Result<()> {
+    let tx = conn.transaction()?;
+    {
+        tx.execute("DELETE FROM monthly_summaries WHERE year_month = ?1;", params![year_month])?;
+        tx.execute(
+            "INSERT INTO monthly_summaries (year_month, target_type, target_id, metric_type, metric_key, metric_value)
+             SELECT ?1, target_type, target_id, metric_type, metric_key, SUM(metric_value)
+             FROM daily_summaries
+             WHERE date LIKE ?2
+             GROUP BY target_type, target_id, metric_type, metric_key;",
+            params![year_month, format!("{}-%", year_month)],
+        )?;
+    }
+    tx.commit()?;
+    Ok(())
+}
+
+pub fn aggregate_year_from_daily(conn: &mut Connection, year: &str) -> rusqlite::Result<()> {
+    let tx = conn.transaction()?;
+    {
+        tx.execute("DELETE FROM yearly_summaries WHERE year = ?1;", params![year])?;
+        tx.execute(
+            "INSERT INTO yearly_summaries (year, target_type, target_id, metric_type, metric_key, metric_value)
+             SELECT ?1, target_type, target_id, metric_type, metric_key, SUM(metric_value)
+             FROM daily_summaries
+             WHERE date LIKE ?2
+             GROUP BY target_type, target_id, metric_type, metric_key;",
+            params![year, format!("{}-%", year)],
+        )?;
+    }
+    tx.commit()?;
+    Ok(())
+}

src/analytics/events.rs

@@ -0,0 +1,48 @@
+use serde::{Serialize, Deserialize};
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub enum AnalyticsEvent {
+    RedirectVisit {
+        url_id: String,
+        code: String,
+        timestamp: String,
+        ip_address: String,
+        user_agent: String,
+        referer: String,
+        accept_language: String,
+        country: String,
+        status_code: u16,
+    },
+    LandingPageVisit {
+        page_id: String,
+        code: String,
+        slug: String,
+        timestamp: String,
+        ip_address: String,
+        user_agent: String,
+        referer: String,
+        accept_language: String,
+        country: String,
+        status_code: u16,
+    },
+    AdminLogin {
+        username: String,
+        timestamp: String,
+        ip_address: Option<String>,
+        user_agent: Option<String>,
+        success: bool,
+    },
+    ApiRequest {
+        username: String,
+        endpoint: String,
+        method: String,
+        timestamp: String,
+        ip_address: Option<String>,
+        status_code: u16,
+    },
+    SystemEvent {
+        event_type: String, // 'startup', 'shutdown', 'backup', etc.
+        details: String,
+        timestamp: String,
+    },
+}

src/analytics/location.rs

@@ -0,0 +1,19 @@
+use axum::http::HeaderMap;
+
+// Guess or extract client country from headers
+pub fn get_client_country(headers: &HeaderMap) -> String {
+    if let Some(country) = headers.get("cf-ipcountry").and_then(|h| h.to_str().ok()) {
+        return country.to_uppercase();
+    }
+    if let Some(lang) = headers.get("accept-language").and_then(|h| h.to_str().ok()) {
+        if let Some(dash_idx) = lang.find('-') {
+            if lang.len() > dash_idx + 2 {
+                let code = &lang[dash_idx + 1..dash_idx + 3];
+                if code.chars().all(|c| c.is_ascii_alphabetic()) {
+                    return code.to_uppercase();
+                }
+            }
+        }
+    }
+    "Unknown".to_string()
+}

src/analytics/mod.rs

@@ -0,0 +1,10 @@
+pub mod queue;
+pub mod worker;
+pub mod location;
+pub mod events;
+pub mod aggregate;
+
+pub use queue::AnalyticsQueue;
+pub use location::get_client_country;
+pub use events::AnalyticsEvent;
+pub use aggregate::{aggregate_day, aggregate_month_from_daily, aggregate_year_from_daily};

src/analytics/queue.rs

@@ -0,0 +1,29 @@
+use tokio::sync::mpsc;
+use crate::models::VisitRecord;
+use crate::db::Db;
+
+#[derive(Clone)]
+pub struct AnalyticsQueue {
+    sender: mpsc::Sender<VisitRecord>,
+}
+
+impl AnalyticsQueue {
+    pub fn new(db: Db, capacity: usize) -> Self {
+        let (sender, receiver) = mpsc::channel(capacity);
+
+        // Spawn background worker to batch-write records
+        tokio::spawn(async move {
+            super::worker::run_worker(db, receiver).await;
+        });
+
+        Self { sender }
+    }
+
+    // Attempt to queue a visit. Non-blocking.
+    pub fn push(&self, record: VisitRecord) {
+        use tracing::error;
+        if let Err(e) = self.sender.try_send(record) {
+            error!("Failed to queue analytics record: {:?}", e);
+        }
+    }
+}

src/analytics/worker.rs

@@ -0,0 +1,56 @@
+use tokio::sync::mpsc;
+use tokio::time::{interval, MissedTickBehavior};
+use std::time::Duration;
+use tracing::{info, error};
+
+use crate::db::Db;
+use crate::models::VisitRecord;
+use crate::db::analytics::insert_visits_batch;
+
+pub async fn run_worker(db: Db, mut receiver: mpsc::Receiver<VisitRecord>) {
+    let mut batch = Vec::new();
+    let batch_size = 50;
+    let flush_interval = Duration::from_secs(2);
+
+    let mut timer = interval(flush_interval);
+    timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
+
+    loop {
+        tokio::select! {
+            record_opt = receiver.recv() => {
+                match record_opt {
+                    Some(record) => {
+                        batch.push(record);
+                        if batch.len() >= batch_size {
+                            flush_batch(&db, &mut batch);
+                        }
+                    }
+                    None => {
+                        info!("Analytics channel closed. Flushing remaining records.");
+                        flush_batch(&db, &mut batch);
+                        break;
+                    }
+                }
+            }
+            _ = timer.tick() => {
+                if !batch.is_empty() {
+                    flush_batch(&db, &mut batch);
+                }
+            }
+        }
+    }
+}
+
+fn flush_batch(db: &Db, batch: &mut Vec<VisitRecord>) {
+    if batch.is_empty() {
+        return;
+    }
+
+    info!("Flushing {} visits to analytics database", batch.len());
+    let mut conn_lock = db.analytics.lock().unwrap();
+    if let Err(e) = insert_visits_batch(&mut conn_lock, batch) {
+        error!("Failed to write analytics batch to database: {:?}", e);
+    } else {
+        batch.clear();
+    }
+}