bctx-cloud-core 0.1.30

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 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 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", [], |r| {
                r.get(0)
            })
            .unwrap_or(0);

        let tokens_saved_last_30d: i64 = conn
            .query_row(
                "SELECT COALESCE(SUM(tokens_saved),0) FROM usage WHERE recorded_at >= datetime('now','-30 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 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<()> {
        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(())
    }

    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 WHERE user_id=?1 AND recorded_at >= datetime('now', ?2)",
            params![user_id, format!("-{days} 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 = format!("-{days} 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
             WHERE user_id=?1 AND recorded_at >= datetime('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 = format!("-{days} days");
        let mut stmt = match conn.prepare(
            "SELECT skill, COALESCE(SUM(tokens_sent),0) as total
             FROM usage
             WHERE user_id=?1 AND skill!='' AND recorded_at >= datetime('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);
";

#[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),
            ]
        );
    }
}