bctx-cloud-core 0.1.39

use anyhow::Result;
use rusqlite::{params, Connection};
use std::path::Path;
use std::sync::{Arc, Mutex};

pub struct CloudDb {
    conn: Arc<Mutex<Connection>>,
}

impl CloudDb {
    pub fn open(path: &Path) -> Result<Self> {
        let conn = Connection::open(path)?;
        // WAL mode: readers don't block writers, writers don't block readers.
        conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;")?;
        let db = Self {
            conn: Arc::new(Mutex::new(conn)),
        };
        db.migrate()?;
        Ok(db)
    }

    pub fn open_in_memory() -> Result<Self> {
        let conn = Connection::open_in_memory()?;
        let db = Self {
            conn: Arc::new(Mutex::new(conn)),
        };
        db.migrate()?;
        Ok(db)
    }

    fn migrate(&self) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute_batch(SCHEMA)?;
        // Idempotent: fails silently if column already exists.
        let _ = conn.execute(
            "ALTER TABLE users ADD COLUMN is_admin INTEGER NOT NULL DEFAULT 0",
            [],
        );
        // Seed a baseline 'signup' event for any user that predates tier tracking
        // so conversion history is complete back to their join date. Idempotent.
        conn.execute(
            "INSERT INTO tier_events(user_id, old_tier, new_tier, reason, occurred_at)
             SELECT id, NULL, tier, 'backfill', created_at FROM users u
             WHERE NOT EXISTS (SELECT 1 FROM tier_events te WHERE te.user_id = u.id)",
            [],
        )?;
        Ok(())
    }

    pub fn conn(&self) -> std::sync::MutexGuard<'_, Connection> {
        self.conn.lock().unwrap()
    }
}

impl Clone for CloudDb {
    fn clone(&self) -> Self {
        Self {
            conn: self.conn.clone(),
        }
    }
}

pub struct User {
    pub id: String,
    pub email: String,
    pub tier: String,
    pub created_at: String,
    pub is_admin: bool,
}

pub struct AdminStats {
    pub total_users: i64,
    pub users_by_tier: Vec<(String, i64)>,
    pub mrr_usd: f64,
    pub total_tokens_saved_all_time: i64,
    pub tokens_saved_last_30d: i64,
    pub top_commands: Vec<(String, i64)>,
    pub new_users_last_7d: i64,
    pub new_users_last_30d: i64,
}

pub struct VaultRecord {
    pub project_hash: String,
    pub user_id: String,
    pub facts_json: String,
    pub server_version: i64,
    pub updated_at: String,
}

pub struct UsageRecord {
    pub user_id: String,
    pub tokens_sent: i64,
    pub tokens_saved: i64,
    pub skill: String,
    pub recorded_at: String,
}

pub struct TierEvent {
    pub old_tier: Option<String>,
    pub new_tier: String,
    pub reason: String,
    pub occurred_at: String,
}

pub struct UserRow {
    pub id: String,
    pub tier: String,
    pub created_at: String,
}

impl CloudDb {
    pub fn upsert_user(&self, id: &str, email: &str, tier: &str) -> Result<()> {
        let conn = self.conn();
        conn.execute(
            "INSERT INTO users(id, email, tier, created_at) VALUES(?1,?2,?3,datetime('now'))
             ON CONFLICT(id) DO UPDATE SET tier=excluded.tier",
            params![id, email, tier],
        )?;
        Ok(())
    }

    /// Record an explicit tier transition. Used for signup (old_tier = None)
    /// where the user row is created separately by `upsert_user`.
    pub fn record_tier_event(
        &self,
        user_id: &str,
        old_tier: Option<&str>,
        new_tier: &str,
        reason: &str,
    ) -> Result<()> {
        let conn = self.conn();
        conn.execute(
            "INSERT INTO tier_events(user_id, old_tier, new_tier, reason, occurred_at)
             VALUES(?1,?2,?3,?4,datetime('now'))",
            params![user_id, old_tier, new_tier, reason],
        )?;
        Ok(())
    }

    /// Update a user's tier and append a tier event in one transaction.
    /// No-op (returns false) when the user is unknown or already on `new_tier`,
    /// so retried Stripe webhooks don't produce duplicate events.
    pub fn change_user_tier(&self, id: &str, new_tier: &str, reason: &str) -> Result<bool> {
        let conn = self.conn();
        let old: Option<String> = conn
            .query_row("SELECT tier FROM users WHERE id=?1", params![id], |r| {
                r.get(0)
            })
            .ok();
        match old.as_deref() {
            None => Ok(false),
            Some(t) if t == new_tier => Ok(false),
            Some(_) => {
                conn.execute(
                    "UPDATE users SET tier=?1 WHERE id=?2",
                    params![new_tier, id],
                )?;
                conn.execute(
                    "INSERT INTO tier_events(user_id, old_tier, new_tier, reason, occurred_at)
                     VALUES(?1,?2,?3,?4,datetime('now'))",
                    params![id, old, new_tier, reason],
                )?;
                Ok(true)
            }
        }
    }

    pub fn count_users(&self) -> i64 {
        let conn = self.conn();
        conn.query_row("SELECT COUNT(*) FROM users", [], |r| r.get(0))
            .unwrap_or(0)
    }

    /// One page of users ordered by join date — for the paginated admin export.
    pub fn users_page(&self, limit: i64, offset: i64) -> Vec<UserRow> {
        let conn = self.conn();
        let mut stmt = match conn.prepare(
            "SELECT id, tier, created_at FROM users
             ORDER BY created_at ASC, id ASC LIMIT ?1 OFFSET ?2",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map(params![limit, offset], |row| {
            Ok(UserRow {
                id: row.get(0)?,
                tier: row.get(1)?,
                created_at: row.get(2)?,
            })
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }

    /// Conversion funnel counts computed in SQL over all users: (signups,
    /// ever_paid, paid_now). `ever_paid` is any user who ever entered a paid tier.
    pub fn conversion_counts(&self) -> (i64, i64, i64) {
        let conn = self.conn();
        let signups = conn
            .query_row("SELECT COUNT(*) FROM users", [], |r| r.get(0))
            .unwrap_or(0);
        let ever_paid = conn
            .query_row(
                "SELECT COUNT(DISTINCT user_id) FROM tier_events
                 WHERE new_tier IN ('beacon','studio','enterprise')",
                [],
                |r| r.get(0),
            )
            .unwrap_or(0);
        let paid_now = conn
            .query_row(
                "SELECT COUNT(*) FROM users WHERE tier IN ('beacon','studio','enterprise')",
                [],
                |r| r.get(0),
            )
            .unwrap_or(0);
        (signups, ever_paid, paid_now)
    }

    /// Per-signup-month (cohort, signups, converted), oldest first. `converted`
    /// counts cohort members who ever entered a paid tier.
    pub fn cohorts(&self) -> Vec<(String, i64, i64)> {
        let conn = self.conn();
        let mut stmt = match conn.prepare(
            "SELECT strftime('%Y-%m', u.created_at) AS m,
                    COUNT(*),
                    COUNT(DISTINCT p.user_id)
             FROM users u
             LEFT JOIN (
                 SELECT DISTINCT user_id FROM tier_events
                 WHERE new_tier IN ('beacon','studio','enterprise')
             ) p ON p.user_id = u.id
             GROUP BY m ORDER BY m ASC",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, i64>(2)?,
            ))
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }

    /// Days from signup to first paid event, one entry per converted user.
    /// The caller derives the median.
    pub fn days_to_convert(&self) -> Vec<i64> {
        let conn = self.conn();
        let mut stmt = match conn.prepare(
            "SELECT CAST(julianday(MIN(te.occurred_at)) - julianday(u.created_at) AS INTEGER)
             FROM users u
             JOIN tier_events te ON te.user_id = u.id
                AND te.new_tier IN ('beacon','studio','enterprise')
             GROUP BY u.id",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map([], |row| row.get::<_, i64>(0))
            .ok()
            .map(|rows| rows.filter_map(|r| r.ok()).map(|d| d.max(0)).collect())
            .unwrap_or_default()
    }

    /// Paid users ranked by token usage in `month` (YYYY-MM): (user_id, tier,
    /// tokens_used). Uses a LEFT JOIN so paid subscribers with no usage this
    /// month still appear (at 0) — an idle paying user is a churn signal.
    pub fn quota_usage_top(&self, month: &str, limit: i64) -> Vec<(String, String, i64)> {
        let conn = self.conn();
        let mut stmt = match conn.prepare(
            "SELECT u.id, u.tier, COALESCE(SUM(g.tokens_sent),0) AS used
             FROM users u
             LEFT JOIN usage_all g ON g.user_id = u.id AND strftime('%Y-%m', g.recorded_at) = ?1
             WHERE u.tier IN ('beacon','studio')
             GROUP BY u.id ORDER BY used DESC, u.id ASC LIMIT ?2",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map(params![month, limit], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, i64>(2)?,
            ))
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }

    /// Full tier-change timeline for a user, oldest first.
    pub fn tier_events_for(&self, user_id: &str) -> Vec<TierEvent> {
        let conn = self.conn();
        let mut stmt = match conn.prepare(
            "SELECT old_tier, new_tier, reason, occurred_at
             FROM tier_events WHERE user_id=?1 ORDER BY id ASC",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map(params![user_id], |row| {
            Ok(TierEvent {
                old_tier: row.get(0)?,
                new_tier: row.get(1)?,
                reason: row.get(2)?,
                occurred_at: row.get(3)?,
            })
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }

    /// Raw per-event usage rows for a user, oldest first. The export layer
    /// aggregates these into calendar months and walks them to find the date a
    /// monthly token quota was first crossed.
    /// Returns vec of (recorded_at, tokens_sent, tokens_saved).
    pub fn usage_events(&self, user_id: &str) -> Vec<(String, i64, i64)> {
        let conn = self.conn();
        let mut stmt = match conn.prepare(
            "SELECT recorded_at, tokens_sent, tokens_saved
             FROM usage_all WHERE user_id=?1 ORDER BY recorded_at ASC",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map(params![user_id], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, i64>(2)?,
            ))
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }

    pub fn set_admin_by_email(&self, email: &str, is_admin: bool) -> Result<bool> {
        let conn = self.conn();
        let n = conn.execute(
            "UPDATE users SET is_admin=?1 WHERE email=?2",
            params![is_admin as i64, email],
        )?;
        Ok(n > 0)
    }

    pub fn get_user(&self, id: &str) -> Option<User> {
        let conn = self.conn();
        conn.query_row(
            "SELECT id,email,tier,created_at,COALESCE(is_admin,0) FROM users WHERE id=?1",
            params![id],
            |row| {
                Ok(User {
                    id: row.get(0)?,
                    email: row.get(1)?,
                    tier: row.get(2)?,
                    created_at: row.get(3)?,
                    is_admin: row.get::<_, i64>(4)? != 0,
                })
            },
        )
        .ok()
    }

    pub fn admin_stats(&self) -> AdminStats {
        let conn = self.conn();

        let total_users: i64 = conn
            .query_row("SELECT COUNT(*) FROM users", [], |r| r.get(0))
            .unwrap_or(0);

        let users_by_tier: Vec<(String, i64)> = {
            let mut stmt = conn
                .prepare("SELECT tier, COUNT(*) FROM users GROUP BY tier ORDER BY COUNT(*) DESC")
                .unwrap();
            stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)))
                .ok()
                .map(|rows| rows.filter_map(|r| r.ok()).collect())
                .unwrap_or_default()
        };

        // Approximate MRR: count paid users × tier price
        let mrr_usd: f64 = users_by_tier.iter().fold(0.0, |acc, (tier, count)| {
            acc + match tier.as_str() {
                "beacon" => 9.0 * (*count as f64),
                "studio" => 29.0 * (*count as f64),
                "enterprise" => 499.0 * (*count as f64),
                _ => 0.0,
            }
        });

        let total_tokens_saved_all_time: i64 = conn
            .query_row(
                "SELECT COALESCE(SUM(tokens_saved),0) FROM usage_all",
                [],
                |r| r.get(0),
            )
            .unwrap_or(0);

        let tokens_saved_last_30d: i64 = conn
            .query_row(
                "SELECT COALESCE(SUM(tokens_saved),0) FROM usage_all WHERE recorded_at >= date('now','-29 days')",
                [],
                |r| r.get(0),
            )
            .unwrap_or(0);

        let top_commands: Vec<(String, i64)> = {
            let mut stmt = conn
                .prepare(
                    "SELECT skill, COALESCE(SUM(tokens_saved),0) as saved
                     FROM usage_all WHERE skill!=''
                     GROUP BY skill ORDER BY saved DESC LIMIT 10",
                )
                .unwrap();
            stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)))
                .ok()
                .map(|rows| rows.filter_map(|r| r.ok()).collect())
                .unwrap_or_default()
        };

        let new_users_last_7d: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM users WHERE created_at >= datetime('now','-7 days')",
                [],
                |r| r.get(0),
            )
            .unwrap_or(0);

        let new_users_last_30d: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM users WHERE created_at >= datetime('now','-30 days')",
                [],
                |r| r.get(0),
            )
            .unwrap_or(0);

        AdminStats {
            total_users,
            users_by_tier,
            mrr_usd,
            total_tokens_saved_all_time,
            tokens_saved_last_30d,
            top_commands,
            new_users_last_7d,
            new_users_last_30d,
        }
    }

    pub fn upsert_vault(&self, user_id: &str, project_hash: &str, facts_json: &str) -> Result<i64> {
        let conn = self.conn();
        conn.execute(
            "INSERT INTO vaults(project_hash,user_id,facts_json,server_version,updated_at)
             VALUES(?1,?2,?3,1,datetime('now'))
             ON CONFLICT(project_hash,user_id) DO UPDATE SET
               facts_json=excluded.facts_json,
               server_version=server_version+1,
               updated_at=excluded.updated_at",
            params![project_hash, user_id, facts_json],
        )?;
        let version: i64 = conn.query_row(
            "SELECT server_version FROM vaults WHERE project_hash=?1 AND user_id=?2",
            params![project_hash, user_id],
            |row| row.get(0),
        )?;
        Ok(version)
    }

    pub fn get_vault(&self, user_id: &str, project_hash: &str) -> Option<VaultRecord> {
        let conn = self.conn();
        conn.query_row(
            "SELECT project_hash,user_id,facts_json,server_version,updated_at
             FROM vaults WHERE project_hash=?1 AND user_id=?2",
            params![project_hash, user_id],
            |row| {
                Ok(VaultRecord {
                    project_hash: row.get(0)?,
                    user_id: row.get(1)?,
                    facts_json: row.get(2)?,
                    server_version: row.get(3)?,
                    updated_at: row.get(4)?,
                })
            },
        )
        .ok()
    }

    /// Returns true if `key` (user_id or "ip:x.x.x.x") is within the daily compress limit.
    pub fn check_compress_rate_limit(&self, key: &str, daily_limit: i64) -> bool {
        if daily_limit == i64::MAX {
            return true;
        }
        let conn = self.conn();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM usage WHERE user_id=?1 AND skill='web_compress' AND recorded_at >= date('now')",
                rusqlite::params![key],
                |r| r.get(0),
            )
            .unwrap_or(0);
        count < daily_limit
    }

    /// Returns true if the user is within rate limits (120/min, 10_000/day).
    pub fn check_signal_rate_limit(&self, user_id: &str) -> bool {
        let conn = self.conn();
        let per_min: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM usage WHERE user_id=?1 AND recorded_at >= datetime('now','-1 minute')",
                params![user_id],
                |r| r.get(0),
            )
            .unwrap_or(0);
        if per_min >= 120 {
            return false;
        }
        let per_day: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM usage WHERE user_id=?1 AND recorded_at >= datetime('now','-1 day')",
                params![user_id],
                |r| r.get(0),
            )
            .unwrap_or(0);
        per_day < 10_000
    }

    pub fn record_usage(
        &self,
        user_id: &str,
        tokens_sent: i64,
        tokens_saved: i64,
        skill: &str,
    ) -> Result<()> {
        // Clients up to 0.1.37 send what they recorded; keep only the program's name.
        let skill = if skill.is_empty() {
            String::new()
        } else {
            crate::usage::skill_name(skill)
        };
        let conn = self.conn();
        conn.execute(
            "INSERT INTO usage(user_id,tokens_sent,tokens_saved,skill,recorded_at)
             VALUES(?1,?2,?3,?4,datetime('now'))",
            params![user_id, tokens_sent, tokens_saved, skill],
        )?;
        Ok(())
    }

    /// Distinct program names one user may store per day; a real day uses a few dozen.
    pub const SKILLS_PER_DAY: i64 = 200;

    /// Live one-time web codes one user may hold; each lives 60 seconds.
    pub const WEB_CODES_PER_MINUTE: i64 = 10;

    /// The `date('now', …)` modifier that starts a window of `days` whole UTC days ending today.
    pub fn day_window(days: i64) -> String {
        format!("-{} days", (days - 1).max(0))
    }

    /// Stores a device's per-day totals with replace semantics: re-sending a day
    /// overwrites it, so retries and repeated snapshots never double-count. A program new
    /// to a day is skipped once the user has [`Self::SKILLS_PER_DAY`] that day; returns
    /// how many rows were skipped.
    pub fn upsert_usage_daily(
        &self,
        user_id: &str,
        device_id: &str,
        rows: &[crate::usage::UsageRow],
    ) -> Result<usize> {
        let mut skipped = 0;
        let mut conn = self.conn();
        let tx = conn.transaction()?;
        {
            let mut stmt = tx.prepare(
                    "INSERT INTO usage_daily(user_id, device_id, day, skill, tokens_sent, tokens_saved, runs, updated_at)
                     SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, datetime('now')
                     WHERE EXISTS(SELECT 1 FROM usage_daily WHERE user_id = ?1 AND day = ?3 AND skill = ?4)
                        OR (SELECT COUNT(DISTINCT skill) FROM usage_daily WHERE user_id = ?1 AND day = ?3) < ?8
                     ON CONFLICT(user_id, device_id, day, skill) DO UPDATE SET
                         tokens_sent = excluded.tokens_sent,
                         tokens_saved = excluded.tokens_saved,
                         runs = excluded.runs,
                         updated_at = excluded.updated_at",
                )?;
            for r in rows {
                let stored = stmt.execute(params![
                    user_id,
                    device_id,
                    r.day,
                    r.skill,
                    r.tokens_sent,
                    r.tokens_saved,
                    r.runs,
                    Self::SKILLS_PER_DAY
                ])?;
                if stored == 0 {
                    skipped += 1;
                }
            }
        }
        tx.commit()?;
        Ok(skipped)
    }

    /// A 60-second, single-use code that a browser exchanges for a session token,
    /// so long-lived tokens never appear in URLs, history or access logs.
    /// Fails with [`TooManyWebCodes`] once the user holds [`Self::WEB_CODES_PER_MINUTE`] live codes.
    pub fn create_web_code(&self, user_id: &str) -> Result<String> {
        let conn = self.conn();
        conn.execute(
            "DELETE FROM web_codes WHERE expires_at < datetime('now')",
            [],
        )?;
        let live: i64 = conn.query_row(
            "SELECT COUNT(*) FROM web_codes WHERE user_id = ?1",
            params![user_id],
            |row| row.get(0),
        )?;
        if live >= Self::WEB_CODES_PER_MINUTE {
            return Err(TooManyWebCodes.into());
        }
        let code = format!(
            "{}{}",
            uuid::Uuid::new_v4().simple(),
            uuid::Uuid::new_v4().simple()
        );
        conn.execute(
            "INSERT INTO web_codes(code, user_id, expires_at, used)
                 VALUES (?1, ?2, datetime('now','+60 seconds'), 0)",
            params![code, user_id],
        )?;
        Ok(code)
    }

    pub fn redeem_web_code(&self, code: &str) -> Option<String> {
        self.conn()
            .query_row(
                "UPDATE web_codes SET used = 1
                     WHERE code = ?1 AND used = 0 AND expires_at >= datetime('now')
                     RETURNING user_id",
                params![code],
                |row| row.get(0),
            )
            .ok()
    }

    /// One snapshot per user per minute, whatever device it claims to come from:
    /// clients send every 6 hours, and the device id is the client's to choose.
    pub fn snapshot_too_soon(&self, user_id: &str) -> bool {
        let conn = self.conn();
        conn.query_row(
            "SELECT EXISTS(SELECT 1 FROM usage_daily
                               WHERE user_id=?1
                                 AND updated_at >= datetime('now','-60 seconds'))",
            params![user_id],
            |row| row.get::<_, bool>(0),
        )
        .unwrap_or(false)
    }

    /// Devices one user may sync from within 24 hours; every login creates one.
    pub const DEVICES_PER_DAY: i64 = 10;

    /// True when `device_id` is new to this user and [`Self::DEVICES_PER_DAY`]
    /// other devices already synced in the last 24 hours.
    pub fn too_many_devices(&self, user_id: &str, device_id: &str) -> bool {
        let conn = self.conn();
        conn.query_row(
            "SELECT NOT EXISTS(SELECT 1 FROM usage_daily WHERE user_id=?1 AND device_id=?2)
                    AND (SELECT COUNT(DISTINCT device_id) FROM usage_daily
                          WHERE user_id=?1 AND updated_at >= datetime('now','-1 day')) >= ?3",
            params![user_id, device_id, Self::DEVICES_PER_DAY],
            |row| row.get::<_, bool>(0),
        )
        .unwrap_or(true)
    }

    pub fn usage_summary(&self, user_id: &str, days: i64) -> (i64, i64) {
        let conn = self.conn();
        conn.query_row(
            "SELECT COALESCE(SUM(tokens_sent),0), COALESCE(SUM(tokens_saved),0)
             FROM usage_all WHERE user_id=?1 AND recorded_at >= date('now', ?2)",
            params![user_id, Self::day_window(days)],
            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
        )
        .unwrap_or((0, 0))
    }

    pub fn store_device_code(
        &self,
        device_code: &str,
        user_code: &str,
        email: &str,
        expires_at: &str,
    ) -> Result<()> {
        let conn = self.conn();
        conn.execute(
            "INSERT OR REPLACE INTO device_codes(device_code,user_code,email,authorized,expires_at)
             VALUES(?1,?2,?3,0,?4)",
            params![device_code, user_code, email, expires_at],
        )?;
        Ok(())
    }

    pub fn authorize_device_code(&self, user_code: &str, user_id: &str) -> Result<bool> {
        let conn = self.conn();
        let n = conn.execute(
            "UPDATE device_codes SET authorized=1,user_id=?1 WHERE user_code=?2 AND datetime('now')<expires_at",
            params![user_id, user_code],
        )?;
        Ok(n > 0)
    }

    pub fn poll_device_code(&self, device_code: &str) -> Option<String> {
        let conn = self.conn();
        conn.query_row(
            "SELECT user_id FROM device_codes WHERE device_code=?1 AND authorized=1 AND datetime('now')<expires_at",
            params![device_code],
            |row| row.get(0),
        ).ok()
    }

    /// Per-day usage over the last `days` days.
    /// Returns vec of (date_str, tokens_sent, tokens_saved).
    pub fn usage_by_day(&self, user_id: &str, days: i64) -> Vec<(String, i64, i64)> {
        let conn = self.conn();
        let window = Self::day_window(days);
        let mut stmt = match conn.prepare(
            "SELECT date(recorded_at) as day,
                    COALESCE(SUM(tokens_sent),0),
                    COALESCE(SUM(tokens_saved),0)
             FROM usage_all
             WHERE user_id=?1 AND recorded_at >= date('now', ?2)
             GROUP BY date(recorded_at)
             ORDER BY day ASC",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map(params![user_id, window], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, i64>(2)?,
            ))
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }

    /// Per-skill token totals over the last `days` days.
    /// Returns vec of (skill_name, tokens_sent) sorted descending.
    pub fn skill_breakdown(&self, user_id: &str, days: i64) -> Vec<(String, i64)> {
        let conn = self.conn();
        let window = Self::day_window(days);
        let mut stmt = match conn.prepare(
            "SELECT skill, COALESCE(SUM(tokens_sent),0) as total
             FROM usage_all
             WHERE user_id=?1 AND skill!='' AND recorded_at >= date('now', ?2)
             GROUP BY skill
             ORDER BY total DESC
             LIMIT 10",
        ) {
            Ok(s) => s,
            Err(_) => return vec![],
        };
        stmt.query_map(params![user_id, window], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
        })
        .ok()
        .map(|rows| rows.filter_map(|r| r.ok()).collect())
        .unwrap_or_default()
    }
}

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS users (
    id TEXT PRIMARY KEY,
    email TEXT NOT NULL UNIQUE,
    tier TEXT NOT NULL DEFAULT 'free',
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS vaults (
    project_hash TEXT NOT NULL,
    user_id TEXT NOT NULL,
    facts_json TEXT NOT NULL,
    server_version INTEGER NOT NULL DEFAULT 1,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (project_hash, user_id)
);

CREATE TABLE IF NOT EXISTS usage (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id TEXT NOT NULL,
    tokens_sent INTEGER NOT NULL DEFAULT 0,
    tokens_saved INTEGER NOT NULL DEFAULT 0,
    skill TEXT NOT NULL DEFAULT '',
    recorded_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS device_codes (
    device_code TEXT PRIMARY KEY,
    user_code TEXT NOT NULL UNIQUE,
    email TEXT NOT NULL,
    authorized INTEGER NOT NULL DEFAULT 0,
    user_id TEXT,
    expires_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS tier_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id TEXT NOT NULL,
    old_tier TEXT,
    new_tier TEXT NOT NULL,
    reason TEXT NOT NULL,
    occurred_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_tier_events_user ON tier_events(user_id, id);

CREATE INDEX IF NOT EXISTS idx_usage_user ON usage(user_id, recorded_at);

CREATE TABLE IF NOT EXISTS usage_daily (
    user_id TEXT NOT NULL,
    device_id TEXT NOT NULL,
    day TEXT NOT NULL,
    skill TEXT NOT NULL DEFAULT '',
    tokens_sent INTEGER NOT NULL DEFAULT 0,
    tokens_saved INTEGER NOT NULL DEFAULT 0,
    runs INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (user_id, device_id, day, skill)
);

DROP VIEW IF EXISTS usage_all;
CREATE VIEW usage_all AS
    SELECT user_id, tokens_sent, tokens_saved, skill, 1 AS runs, recorded_at FROM usage
    UNION ALL
    SELECT user_id, tokens_sent, tokens_saved, skill, runs, day || ' 12:00:00' AS recorded_at FROM usage_daily;

CREATE TABLE IF NOT EXISTS web_codes (
    code TEXT PRIMARY KEY,
    user_id TEXT NOT NULL,
    expires_at TEXT NOT NULL,
    used INTEGER NOT NULL DEFAULT 0
);
";

/// The user already holds [`CloudDb::WEB_CODES_PER_MINUTE`] live web codes.
#[derive(Debug)]
pub struct TooManyWebCodes;

impl std::fmt::Display for TooManyWebCodes {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("too many web codes; wait a minute")
    }
}

impl std::error::Error for TooManyWebCodes {}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn signup_then_upgrade_builds_timeline() {
        let db = CloudDb::open_in_memory().unwrap();
        db.upsert_user("u1", "a@b.com", "free").unwrap();
        db.record_tier_event("u1", None, "free", "signup").unwrap();

        assert!(db
            .change_user_tier("u1", "beacon", "stripe_checkout")
            .unwrap());
        // Idempotent: same tier again produces no second event.
        assert!(!db
            .change_user_tier("u1", "beacon", "stripe_checkout")
            .unwrap());
        assert!(db
            .change_user_tier("u1", "free", "downgrade_to_free")
            .unwrap());

        let events = db.tier_events_for("u1");
        assert_eq!(events.len(), 3);
        assert_eq!(events[0].new_tier, "free");
        assert_eq!(events[0].old_tier, None);
        assert_eq!(events[0].reason, "signup");
        assert_eq!(events[1].new_tier, "beacon");
        assert_eq!(events[1].old_tier.as_deref(), Some("free"));
        assert_eq!(events[2].new_tier, "free");
        assert_eq!(events[2].old_tier.as_deref(), Some("beacon"));
        assert_eq!(db.get_user("u1").unwrap().tier, "free");
    }

    #[test]
    fn change_tier_on_unknown_user_is_noop() {
        let db = CloudDb::open_in_memory().unwrap();
        assert!(!db.change_user_tier("ghost", "beacon", "x").unwrap());
        assert!(db.tier_events_for("ghost").is_empty());
    }

    #[test]
    fn migrate_backfills_existing_users() {
        // A user inserted directly (no tier event) gets a backfill event on
        // the next open() against the same connection-less path simulation.
        let db = CloudDb::open_in_memory().unwrap();
        {
            let conn = db.conn();
            conn.execute(
                "INSERT INTO users(id,email,tier,created_at) VALUES('old','o@b.com','beacon','2026-01-01 00:00:00')",
                [],
            )
            .unwrap();
        }
        db.migrate().unwrap();
        let events = db.tier_events_for("old");
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].reason, "backfill");
        assert_eq!(events[0].new_tier, "beacon");
        assert_eq!(events[0].occurred_at, "2026-01-01 00:00:00");
    }

    #[test]
    fn usage_events_returns_ordered_rows() {
        let db = CloudDb::open_in_memory().unwrap();
        db.upsert_user("u1", "a@b.com", "free").unwrap();
        {
            let conn = db.conn();
            conn.execute(
                "INSERT INTO usage(user_id,tokens_sent,tokens_saved,skill,recorded_at)
                 VALUES('u1',200,80,'scout','2026-05-20 10:00:00'),
                       ('u1',100,40,'scout','2026-05-10 10:00:00'),
                       ('u1',300,90,'scout','2026-06-01 10:00:00')",
                [],
            )
            .unwrap();
        }
        let events = db.usage_events("u1");
        assert_eq!(events.len(), 3);
        assert_eq!(events[0], ("2026-05-10 10:00:00".to_string(), 100, 40));
        assert_eq!(events[1], ("2026-05-20 10:00:00".to_string(), 200, 80));
        assert_eq!(events[2], ("2026-06-01 10:00:00".to_string(), 300, 90));
    }

    /// Seeds three users (free / beacon / studio) with tier history and usage,
    /// then exercises every aggregate + pagination method off one fixture.
    fn seed_aggregate_fixture() -> CloudDb {
        let db = CloudDb::open_in_memory().unwrap();
        let conn = db.conn();
        conn.execute_batch(
            "INSERT INTO users(id,email,tier,created_at) VALUES
               ('a','a@x.io','free','2026-05-01 09:00:00'),
               ('b','b@x.io','beacon','2026-05-10 09:00:00'),
               ('c','c@x.io','studio','2026-06-02 09:00:00');
             INSERT INTO tier_events(user_id,old_tier,new_tier,reason,occurred_at) VALUES
               ('a',NULL,'free','signup','2026-05-01 09:00:00'),
               ('b',NULL,'free','signup','2026-05-10 09:00:00'),
               ('b','free','beacon','stripe_checkout','2026-05-12 09:00:00'),
               ('c',NULL,'free','signup','2026-06-02 09:00:00'),
               ('c','free','studio','stripe_checkout','2026-06-05 09:00:00');
             INSERT INTO usage(user_id,tokens_sent,tokens_saved,skill,recorded_at) VALUES
               ('a',9999,10,'scout','2026-06-03 09:00:00'),
               ('b',30000,5000,'scout','2026-06-03 09:00:00'),
               ('c',500000,90000,'scout','2026-06-04 09:00:00');",
        )
        .unwrap();
        drop(conn);
        db
    }

    #[test]
    fn count_and_paginate_users() {
        let db = seed_aggregate_fixture();
        assert_eq!(db.count_users(), 3);

        let first = db.users_page(2, 0);
        assert_eq!(
            first.iter().map(|u| u.id.as_str()).collect::<Vec<_>>(),
            ["a", "b"]
        );
        let second = db.users_page(2, 2);
        assert_eq!(
            second.iter().map(|u| u.id.as_str()).collect::<Vec<_>>(),
            ["c"]
        );
        assert!(db.users_page(2, 10).is_empty());
    }

    #[test]
    fn conversion_counts_and_cohorts() {
        let db = seed_aggregate_fixture();

        let (signups, ever_paid, paid_now) = db.conversion_counts();
        assert_eq!((signups, ever_paid, paid_now), (3, 2, 2));

        let cohorts = db.cohorts();
        assert_eq!(
            cohorts,
            vec![
                ("2026-05".to_string(), 2, 1), // a (free) + b (converted)
                ("2026-06".to_string(), 1, 1), // c (converted)
            ]
        );

        let mut days = db.days_to_convert();
        days.sort_unstable();
        assert_eq!(days, vec![2, 3]); // b: 10th→12th, c: 2nd→5th
    }

    #[test]
    fn quota_usage_top_ranks_paid_users_for_month() {
        let db = seed_aggregate_fixture();
        let top = db.quota_usage_top("2026-06", 50);
        // Free user 'a' is excluded; studio 'c' outranks beacon 'b'.
        assert_eq!(
            top,
            vec![
                ("c".to_string(), "studio".to_string(), 500000),
                ("b".to_string(), "beacon".to_string(), 30000),
            ]
        );
        // A month with no usage still lists every paid subscriber, at 0.
        let idle = db.quota_usage_top("2026-01", 50);
        assert_eq!(
            idle,
            vec![
                ("b".to_string(), "beacon".to_string(), 0),
                ("c".to_string(), "studio".to_string(), 0),
            ]
        );
    }

    const DEV_A: &str = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa";
    const DEV_B: &str = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";

    fn today_row(sent: i64, saved: i64) -> crate::usage::UsageRow {
        crate::usage::UsageRow {
            day: chrono::Utc::now().format("%Y-%m-%d").to_string(),
            skill: "git".into(),
            tokens_sent: sent,
            tokens_saved: saved,
            runs: 5,
        }
    }

    fn total_runs(db: &CloudDb, user: &str) -> i64 {
        db.conn()
            .query_row(
                "SELECT COALESCE(SUM(runs), 0) FROM usage_all WHERE user_id = ?1",
                params![user],
                |r| r.get(0),
            )
            .unwrap()
    }

    #[test]
    fn snapshot_replaces_instead_of_adding() {
        let db = CloudDb::open_in_memory().unwrap();
        db.upsert_usage_daily("u1", DEV_A, &[today_row(100, 400)])
            .unwrap();
        db.upsert_usage_daily("u1", DEV_A, &[today_row(150, 600)])
            .unwrap();
        assert_eq!(db.usage_summary("u1", 30), (150, 600));
    }

    #[test]
    fn totals_sum_devices_and_legacy_rows() {
        let db = CloudDb::open_in_memory().unwrap();
        db.record_usage("u1", 10, 20, "git").unwrap();
        db.upsert_usage_daily("u1", DEV_A, &[today_row(100, 400)])
            .unwrap();
        db.upsert_usage_daily("u1", DEV_B, &[today_row(50, 50)])
            .unwrap();
        assert_eq!(db.usage_summary("u1", 30), (160, 470));
        let by_day = db.usage_by_day("u1", 30);
        assert_eq!(by_day.len(), 1);
        assert_eq!((by_day[0].1, by_day[0].2), (160, 470));
        assert_eq!(db.skill_breakdown("u1", 30), vec![("git".to_string(), 160)]);
    }

    #[test]
    fn a_users_second_snapshot_within_a_minute_is_too_soon_from_any_device() {
        // device_id is chosen by the client, so a per-device limit is no limit at all.
        let db = CloudDb::open_in_memory().unwrap();
        assert!(!db.snapshot_too_soon("u1"));
        db.upsert_usage_daily("u1", DEV_A, &[today_row(1, 1)])
            .unwrap();
        assert!(db.snapshot_too_soon("u1"));
        assert!(!db.snapshot_too_soon("u2"));
    }

    #[test]
    fn a_user_cannot_add_endless_new_devices() {
        let db = CloudDb::open_in_memory().unwrap();
        for i in 0..10 {
            let dev = format!("{i:08}-aaaa-4aaa-8aaa-aaaaaaaaaaaa");
            assert!(!db.too_many_devices("u1", &dev), "device {i}");
            db.upsert_usage_daily("u1", &dev, &[today_row(1, 1)])
                .unwrap();
        }
        assert!(db.too_many_devices("u1", "99999999-aaaa-4aaa-8aaa-aaaaaaaaaaaa"));
        assert!(
            !db.too_many_devices("u1", "00000003-aaaa-4aaa-8aaa-aaaaaaaaaaaa"),
            "a known device may sync"
        );
    }

    #[test]
    fn runs_count_commands_not_rows() {
        let db = CloudDb::open_in_memory().unwrap();
        db.record_usage("u1", 10, 20, "git").unwrap();
        db.upsert_usage_daily("u1", DEV_A, &[today_row(100, 400)])
            .unwrap();
        assert_eq!(
            total_runs(&db, "u1"),
            6,
            "one legacy command plus five from the snapshot"
        );
    }

    #[test]
    fn web_code_is_single_use() {
        let db = CloudDb::open_in_memory().unwrap();
        let code = db.create_web_code("u1").unwrap();
        assert_eq!(db.redeem_web_code(&code), Some("u1".to_string()));
        assert_eq!(db.redeem_web_code(&code), None);
    }

    #[test]
    fn expired_web_code_is_rejected() {
        let db = CloudDb::open_in_memory().unwrap();
        db.conn()
            .execute(
                "INSERT INTO web_codes(code,user_id,expires_at,used) VALUES('old','u1',datetime('now','-1 second'),0)",
                [],
            )
            .unwrap();
        assert_eq!(db.redeem_web_code("old"), None);
    }
}