path: root/src/database/mod.rs

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<DbPool> {
    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
        )",
    )
    .execute(pool)
    .await?;

    // Add last_run column for pre-existing databases that lack it.
    sqlx::query("ALTER TABLE cron_jobs ADD COLUMN last_run TIMESTAMP")
        .execute(pool)
        .await
        .ok();

    sqlx::query(
        "CREATE TABLE IF NOT EXISTS triggers (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )",
    )
    .execute(pool)
    .await?;

    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<DateTime<Utc>>,
}

#[allow(dead_code)]
pub async fn get_voice_activity(pool: &DbPool, user_id: &str) -> Result<Option<VoiceActivity>> {
    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<Option<UserStats>> {
    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<UserStats>, 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<UserStats> = 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<i64> {
    // 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,
    pub cron_expression: String,
    pub r#type: String,
    pub channel_id: i64,
    pub active: i32,
    pub audio_filename: Option<String>,
    pub message: Option<String>,
    #[allow(dead_code)]
    pub created_at: Option<DateTime<Utc>>,
    pub last_run: Option<DateTime<Utc>>,
}

#[allow(dead_code)]
pub async fn list_cron_jobs(pool: &DbPool) -> Result<Vec<CronJob>> {
    Ok(sqlx::query_as("SELECT * FROM cron_jobs ORDER BY id")
        .fetch_all(pool)
        .await?)
}

#[allow(dead_code)]
pub async fn list_active_cron_jobs(pool: &DbPool) -> Result<Vec<CronJob>> {
    Ok(sqlx::query_as("SELECT * FROM cron_jobs WHERE active = 1")
        .fetch_all(pool)
        .await?)
}

pub async fn add_cron_job(
    pool: &DbPool,
    cron_expression: &str,
    r#type: &str,
    channel_id: i64,
    audio_filename: Option<&str>,
    message: Option<&str>,
) -> Result<i64> {
    // Find the lowest unused ID to reuse freed slots
    let existing: Vec<i64> = sqlx::query_scalar("SELECT id FROM cron_jobs ORDER BY id")
        .fetch_all(pool)
        .await?;

    let next_id = existing.iter().enumerate().find_map(|(i, &id)| {
        let expected = (i as i64) + 1;
        if id != expected {
            Some(expected)
        } else {
            None
        }
    }).unwrap_or(existing.len() as i64 + 1);

    let result = sqlx::query(
        "INSERT INTO cron_jobs (id, cron_expression, type, channel_id, audio_filename, message) VALUES (?, ?, ?, ?, ?, ?)",
    )
    .bind(next_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) -> Result<bool> {
    let result = sqlx::query("DELETE FROM cron_jobs WHERE id = ?")
        .bind(id)
        .execute(pool)
        .await?;
    Ok(result.rows_affected() > 0)
}

pub async fn update_cron_last_run(pool: &DbPool, id: i64, last_run: DateTime<Utc>) -> 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, active: bool) -> Result<bool> {
    let result = sqlx::query("UPDATE cron_jobs SET active = ? WHERE id = ?")
        .bind(if active { 1 } else { 0 })
        .bind(id)
        .execute(pool)
        .await?;
    Ok(result.rows_affected() > 0)
}

// -- Triggers --
#[derive(Debug, Clone, FromRow)]
pub struct Trigger {
    pub id: i64,
    pub name: String,
    pub created_at: Option<DateTime<Utc>>,
    pub last_updated: Option<DateTime<Utc>>,
}

#[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, name: &str) -> Result<i64> {
    let result = sqlx::query("INSERT INTO triggers (name) VALUES (?)")
        .bind(name)
        .execute(pool)
        .await?;
    Ok(result.last_insert_rowid())
}

pub async fn delete_trigger(pool: &DbPool, id: i64) -> Result<bool> {
    let result = sqlx::query("DELETE FROM triggers WHERE id = ?")
        .bind(id)
        .execute(pool)
        .await?;
    Ok(result.rows_affected() > 0)
}

pub async fn get_trigger_by_name(pool: &DbPool, name: &str) -> Result<Option<Trigger>> {
    Ok(sqlx::query_as("SELECT * FROM triggers WHERE name = ?")
        .bind(name)
        .fetch_optional(pool)
        .await?)
}

pub async fn list_triggers(pool: &DbPool) -> Result<Vec<Trigger>> {
    Ok(sqlx::query_as("SELECT * FROM triggers ORDER BY name")
        .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<bool> {
    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<bool> {
    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<Vec<String>> {
    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<Vec<String>> {
    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<(Trigger, Vec<String>, Vec<String>)>> {
    let triggers = list_triggers(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(())
}