use chrono::{DateTime, Utc}; use sqlx::FromRow; use sqlx::SqlitePool; use crate::error::{Error, Result}; pub type DbPool = SqlitePool; pub async fn init_pool(path: &str) -> Result { if let Some(parent) = std::path::Path::new(path).parent() { tokio::fs::create_dir_all(parent).await.map_err(Error::Io)?; } let pool = SqlitePool::connect(&format!("sqlite:{path}?mode=rwc")).await?; run_migrations(&pool).await?; // Clear stale voice activity records from previous run. sqlx::query("DELETE FROM voice_activity") .execute(&pool) .await?; Ok(pool) } async fn run_migrations(pool: &DbPool) -> Result<()> { sqlx::query( "CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL )", ) .execute(pool) .await?; sqlx::query( "CREATE TABLE IF NOT EXISTS voice_activity ( user_id TEXT PRIMARY KEY, join_time REAL NOT NULL DEFAULT 0, total_active REAL NOT NULL DEFAULT 0, total_afk REAL NOT NULL DEFAULT 0, is_afk INTEGER NOT NULL DEFAULT 0, afk_start REAL NOT NULL DEFAULT 0, active_start REAL NOT NULL DEFAULT 0, channel_name TEXT NOT NULL DEFAULT '', last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP )", ) .execute(pool) .await?; sqlx::query( "CREATE TABLE IF NOT EXISTS cron_jobs ( id INTEGER PRIMARY KEY, cron_expression TEXT NOT NULL, type TEXT NOT NULL CHECK(type IN ('audio', 'wiadomość')), channel_id INTEGER NOT NULL, active INTEGER NOT NULL DEFAULT 1, audio_filename TEXT, message TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_run TIMESTAMP, guild_id INTEGER NOT NULL DEFAULT 0 )", ) .execute(pool) .await?; // Add columns for pre-existing databases that lack them. sqlx::query("ALTER TABLE cron_jobs ADD COLUMN last_run TIMESTAMP") .execute(pool) .await .ok(); sqlx::query("ALTER TABLE cron_jobs ADD COLUMN guild_id INTEGER NOT NULL DEFAULT 0") .execute(pool) .await .ok(); sqlx::query( "CREATE TABLE IF NOT EXISTS triggers ( id INTEGER PRIMARY KEY AUTOINCREMENT, guild_id INTEGER NOT NULL DEFAULT 0, name TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(guild_id, name) )", ) .execute(pool) .await?; // Add guild_id for pre-existing trigger tables. (The old table also kept a // global UNIQUE(name), so existing databases still require unique trigger // names across guilds; new databases use UNIQUE(guild_id, name).) sqlx::query("ALTER TABLE triggers ADD COLUMN guild_id INTEGER NOT NULL DEFAULT 0") .execute(pool) .await .ok(); sqlx::query( "CREATE TABLE IF NOT EXISTS trigger_keywords ( id INTEGER PRIMARY KEY AUTOINCREMENT, trigger_id INTEGER NOT NULL REFERENCES triggers(id) ON DELETE CASCADE, keyword TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(trigger_id, keyword) )", ) .execute(pool) .await?; sqlx::query( "CREATE TABLE IF NOT EXISTS trigger_responses ( id INTEGER PRIMARY KEY AUTOINCREMENT, trigger_id INTEGER NOT NULL REFERENCES triggers(id) ON DELETE CASCADE, response_text TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )", ) .execute(pool) .await?; sqlx::query( "CREATE TABLE IF NOT EXISTS user_stats ( user_id TEXT PRIMARY KEY, message_count INTEGER NOT NULL DEFAULT 0, total_voice_active REAL NOT NULL DEFAULT 0, total_voice_afk REAL NOT NULL DEFAULT 0, last_message_at TIMESTAMP )", ) .execute(pool) .await?; Ok(()) } // -- Voice Activity -- #[derive(Debug, Clone, FromRow)] pub struct VoiceActivity { pub user_id: String, pub join_time: f64, pub total_active: f64, pub total_afk: f64, pub is_afk: i32, pub afk_start: f64, pub active_start: f64, pub channel_name: String, #[allow(dead_code)] pub last_updated: Option>, } #[allow(dead_code)] pub async fn get_voice_activity(pool: &DbPool, user_id: &str) -> Result> { Ok( sqlx::query_as("SELECT * FROM voice_activity WHERE user_id = ?") .bind(user_id) .fetch_optional(pool) .await?, ) } pub async fn upsert_voice_activity(pool: &DbPool, va: &VoiceActivity) -> Result<()> { sqlx::query( "INSERT INTO voice_activity (user_id, join_time, total_active, total_afk, is_afk, afk_start, active_start, channel_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET join_time = excluded.join_time, total_active = excluded.total_active, total_afk = excluded.total_afk, is_afk = excluded.is_afk, afk_start = excluded.afk_start, active_start = excluded.active_start, channel_name = excluded.channel_name, last_updated = CURRENT_TIMESTAMP", ) .bind(&va.user_id) .bind(va.join_time) .bind(va.total_active) .bind(va.total_afk) .bind(va.is_afk) .bind(va.afk_start) .bind(va.active_start) .bind(&va.channel_name) .execute(pool) .await?; Ok(()) } // -- User Stats (leaderboard) -- #[derive(Debug, Clone, FromRow)] pub struct UserStats { pub user_id: String, pub message_count: i64, pub total_voice_active: f64, pub total_voice_afk: f64, } /// Increment message count for a user. pub async fn increment_message_count(pool: &DbPool, user_id: &str) -> Result<()> { sqlx::query( "INSERT INTO user_stats (user_id, message_count, last_message_at) VALUES (?, 1, CURRENT_TIMESTAMP) ON CONFLICT(user_id) DO UPDATE SET message_count = message_count + 1, last_message_at = CURRENT_TIMESTAMP", ) .bind(user_id) .execute(pool) .await?; Ok(()) } /// Add voice time to a user's cumulative stats. pub async fn add_voice_time( pool: &DbPool, user_id: &str, active_seconds: f64, afk_seconds: f64, ) -> Result<()> { sqlx::query( "INSERT INTO user_stats (user_id, total_voice_active, total_voice_afk) VALUES (?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET total_voice_active = total_voice_active + ?, total_voice_afk = total_voice_afk + ?", ) .bind(user_id) .bind(active_seconds) .bind(afk_seconds) .bind(active_seconds) .bind(afk_seconds) .execute(pool) .await?; Ok(()) } /// Get stats for a single user. pub async fn get_user_stats(pool: &DbPool, user_id: &str) -> Result> { Ok( sqlx::query_as("SELECT * FROM user_stats WHERE user_id = ?") .bind(user_id) .fetch_optional(pool) .await?, ) } /// Get leaderboard entries sorted by a metric. /// Returns (entries, total_count) where total_count is used for pagination. /// SAFETY: `order_clause` and `order` are derived from fixed match arms, not user input. pub async fn get_leaderboard( pool: &DbPool, metric: &str, ascending: bool, limit: i64, offset: i64, ) -> Result<(Vec, i64)> { let order = if ascending { "ASC" } else { "DESC" }; // SAFETY: all arms are compile-time constant strings #[allow(clippy::disallowed_methods)] let order_clause = match metric { "messages" => "message_count", "active" => "total_voice_active", "afk" => "total_voice_afk", "total" => "total_voice_active + total_voice_afk", _ => "message_count", }; let total: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM user_stats") .fetch_one(pool) .await?; // SAFETY: order_clause and order are not user-controlled let query = format!( "SELECT * FROM user_stats ORDER BY {} {} LIMIT ? OFFSET ?", order_clause, order ); let entries: Vec = sqlx::query_as(sqlx::AssertSqlSafe(query)) .bind(limit) .bind(offset) .fetch_all(pool) .await?; Ok((entries, total.0)) } /// Get a user's rank for a metric (1-indexed, 0 if not found). /// SAFETY: `order_clause` is derived from fixed match arms, not user input. pub async fn get_user_rank(pool: &DbPool, user_id: &str, metric: &str) -> Result { // SAFETY: all arms are compile-time constant strings #[allow(clippy::disallowed_methods)] let order_clause = match metric { "messages" => "message_count", "active" => "total_voice_active", "afk" => "total_voice_afk", "total" => "total_voice_active + total_voice_afk", _ => "message_count", }; // SAFETY: order_clause is not user-controlled let query = format!( "SELECT COUNT(*) + 1 FROM user_stats WHERE {} > ( SELECT {} FROM user_stats WHERE user_id = ? )", order_clause, order_clause ); let result: (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(query)) .bind(user_id) .fetch_one(pool) .await?; // If user doesn't exist, return 0 let user_exists: Option<(i64,)> = sqlx::query_as( "SELECT 1 FROM user_stats WHERE user_id = ?" ) .bind(user_id) .fetch_optional(pool) .await?; Ok(if user_exists.is_some() { result.0 } else { 0 }) } // -- Cron Jobs -- #[derive(Debug, Clone, FromRow)] pub struct CronJob { pub id: i64, #[allow(dead_code)] pub guild_id: i64, pub cron_expression: String, pub r#type: String, pub channel_id: i64, pub active: i32, pub audio_filename: Option, pub message: Option, #[allow(dead_code)] pub created_at: Option>, pub last_run: Option>, } #[allow(dead_code)] pub async fn list_cron_jobs(pool: &DbPool, guild_id: i64) -> Result> { Ok( // `guild_id = 0` covers rows created before guild scoping existed; // they stay manageable until recreated in a guild. sqlx::query_as("SELECT * FROM cron_jobs WHERE guild_id = ? OR guild_id = 0 ORDER BY id") .bind(guild_id) .fetch_all(pool) .await?, ) } #[allow(dead_code)] pub async fn list_active_cron_jobs(pool: &DbPool) -> Result> { Ok(sqlx::query_as("SELECT * FROM cron_jobs WHERE active = 1") .fetch_all(pool) .await?) } pub async fn add_cron_job( pool: &DbPool, guild_id: i64, cron_expression: &str, r#type: &str, channel_id: i64, audio_filename: Option<&str>, message: Option<&str>, ) -> Result { // Let SQLite allocate the id atomically; a SELECT-then-INSERT scheme // races when two users create jobs at the same time. let result = sqlx::query( "INSERT INTO cron_jobs (guild_id, cron_expression, type, channel_id, audio_filename, message) VALUES (?, ?, ?, ?, ?, ?)", ) .bind(guild_id) .bind(cron_expression) .bind(r#type) .bind(channel_id) .bind(audio_filename) .bind(message) .execute(pool) .await?; Ok(result.last_insert_rowid()) } pub async fn delete_cron_job(pool: &DbPool, id: i64, guild_id: i64) -> Result { let result = sqlx::query("DELETE FROM cron_jobs WHERE id = ? AND (guild_id = ? OR guild_id = 0)") .bind(id) .bind(guild_id) .execute(pool) .await?; Ok(result.rows_affected() > 0) } pub async fn update_cron_last_run(pool: &DbPool, id: i64, last_run: DateTime) -> Result<()> { sqlx::query("UPDATE cron_jobs SET last_run = ? WHERE id = ?") .bind(last_run) .bind(id) .execute(pool) .await?; Ok(()) } pub async fn set_cron_job_active( pool: &DbPool, id: i64, guild_id: i64, active: bool, ) -> Result { let result = sqlx::query("UPDATE cron_jobs SET active = ? WHERE id = ? AND (guild_id = ? OR guild_id = 0)") .bind(if active { 1 } else { 0 }) .bind(id) .bind(guild_id) .execute(pool) .await?; Ok(result.rows_affected() > 0) } // -- Triggers -- #[derive(Debug, Clone, FromRow)] pub struct Trigger { pub id: i64, pub guild_id: i64, pub name: String, pub created_at: Option>, pub last_updated: Option>, } #[allow(dead_code)] #[derive(Debug, Clone, FromRow)] pub struct TriggerKeyword { pub id: i64, pub trigger_id: i64, pub keyword: String, } #[allow(dead_code)] #[derive(Debug, Clone, FromRow)] pub struct TriggerResponse { pub id: i64, pub trigger_id: i64, pub response_text: String, } pub async fn create_trigger(pool: &DbPool, guild_id: i64, name: &str) -> Result { let result = sqlx::query("INSERT INTO triggers (guild_id, name) VALUES (?, ?)") .bind(guild_id) .bind(name) .execute(pool) .await?; Ok(result.last_insert_rowid()) } pub async fn delete_trigger(pool: &DbPool, id: i64, guild_id: i64) -> Result { let result = sqlx::query("DELETE FROM triggers WHERE id = ? AND (guild_id = ? OR guild_id = 0)") .bind(id) .bind(guild_id) .execute(pool) .await?; Ok(result.rows_affected() > 0) } pub async fn get_trigger_by_name( pool: &DbPool, guild_id: i64, name: &str, ) -> Result> { Ok( sqlx::query_as( "SELECT * FROM triggers WHERE (guild_id = ? OR guild_id = 0) AND name = ?", ) .bind(guild_id) .bind(name) .fetch_optional(pool) .await?, ) } pub async fn list_triggers(pool: &DbPool, guild_id: i64) -> Result> { Ok( sqlx::query_as( "SELECT * FROM triggers WHERE guild_id = ? OR guild_id = 0 ORDER BY name", ) .bind(guild_id) .fetch_all(pool) .await?, ) } pub async fn add_keyword(pool: &DbPool, trigger_id: i64, keyword: &str) -> Result<()> { sqlx::query("INSERT OR IGNORE INTO trigger_keywords (trigger_id, keyword) VALUES (?, ?)") .bind(trigger_id) .bind(keyword) .execute(pool) .await?; update_trigger_timestamp(pool, trigger_id).await } pub async fn remove_keyword(pool: &DbPool, trigger_id: i64, keyword: &str) -> Result { let result = sqlx::query("DELETE FROM trigger_keywords WHERE trigger_id = ? AND keyword = ?") .bind(trigger_id) .bind(keyword) .execute(pool) .await?; update_trigger_timestamp(pool, trigger_id).await?; Ok(result.rows_affected() > 0) } pub async fn add_response(pool: &DbPool, trigger_id: i64, response: &str) -> Result<()> { sqlx::query("INSERT INTO trigger_responses (trigger_id, response_text) VALUES (?, ?)") .bind(trigger_id) .bind(response) .execute(pool) .await?; update_trigger_timestamp(pool, trigger_id).await } pub async fn remove_response(pool: &DbPool, trigger_id: i64, response: &str) -> Result { let result = sqlx::query("DELETE FROM trigger_responses WHERE trigger_id = ? AND response_text = ?") .bind(trigger_id) .bind(response) .execute(pool) .await?; update_trigger_timestamp(pool, trigger_id).await?; Ok(result.rows_affected() > 0) } pub async fn get_keywords(pool: &DbPool, trigger_id: i64) -> Result> { let rows: Vec<(String,)> = sqlx::query_as("SELECT keyword FROM trigger_keywords WHERE trigger_id = ?") .bind(trigger_id) .fetch_all(pool) .await?; Ok(rows.into_iter().map(|r| r.0).collect()) } pub async fn get_responses(pool: &DbPool, trigger_id: i64) -> Result> { let rows: Vec<(String,)> = sqlx::query_as("SELECT response_text FROM trigger_responses WHERE trigger_id = ?") .bind(trigger_id) .fetch_all(pool) .await?; Ok(rows.into_iter().map(|r| r.0).collect()) } pub async fn load_all_triggers_with_keywords_and_responses( pool: &DbPool, ) -> Result, Vec)>> { // The event cache needs every guild's triggers; `/trigger list` and // mutations use the guild-scoped helpers above. let triggers: Vec = sqlx::query_as("SELECT * FROM triggers ORDER BY name") .fetch_all(pool) .await?; let mut result = Vec::new(); for t in triggers { let keywords = get_keywords(pool, t.id).await?; let responses = get_responses(pool, t.id).await?; result.push((t, keywords, responses)); } Ok(result) } async fn update_trigger_timestamp(pool: &DbPool, id: i64) -> Result<()> { sqlx::query("UPDATE triggers SET last_updated = CURRENT_TIMESTAMP WHERE id = ?") .bind(id) .execute(pool) .await?; Ok(()) }