bcode_usage 0.0.1-alpha.0

//! Transactional, disposable usage snapshots stored through Switchy/Turso.

use bcode_session_models::{SessionId, SessionUsageGeneration, SessionUsagePage};
use bcode_usage_models::{UsageQuery, UsageReport, UsageRequestRow};
use std::path::{Path, PathBuf};
use switchy::database::{
    Database, DatabaseValue, Row,
    query::{FilterableQuery, SortDirection},
    schema::{Column, DataType, create_index, create_table},
};

mod upgrade;

const SCHEMA: i64 = 2;

fn column(name: &str, data_type: DataType, nullable: bool) -> Column {
    Column {
        name: name.into(),
        nullable,
        auto_increment: false,
        data_type,
        default: None,
    }
}

async fn create_row_schema(db: &dyn Database) -> Result<(), String> {
    create_table("usage_rows")
        .columns(vec![
            column("ordinal", DataType::BigInt, true),
            column("session_id", DataType::Text, false),
            column("request_key", DataType::Text, false),
            column("entry_json", DataType::Text, false),
            column("staged", DataType::BigInt, false),
        ])
        .primary_key("ordinal")
        .unique_columns(["session_id", "request_key", "staged"])
        .execute(db)
        .await
        .map_err(error)?;
    for (name, columns) in [
        (
            "usage_session_rows",
            vec!["session_id", "staged", "ordinal"],
        ),
        ("usage_published_rows", vec!["staged", "ordinal"]),
    ] {
        create_index(name)
            .table("usage_rows")
            .columns(columns)
            .execute(db)
            .await
            .map_err(error)?;
    }
    Ok(())
}

async fn create_time_schema(db: &dyn Database) -> Result<(), String> {
    create_table("usage_time_nodes")
        .columns(vec![
            // SQLite TEXT PRIMARY KEY permits NULL unless NOT NULL is explicit.
            column("node_key", DataType::Text, true),
            column("tree", DataType::Text, false),
            column("totals_json", DataType::Text, false),
        ])
        .primary_key("node_key")
        .execute(db)
        .await
        .map_err(error)?;
    create_index("usage_time_tree")
        .table("usage_time_nodes")
        .column("tree")
        .execute(db)
        .await
        .map_err(error)
}

/// State-location-local reporting storage. Opening a dashboard never initializes or repairs it.
#[derive(Debug)]
pub struct UsageIndex {
    path: PathBuf,
}

pub(super) fn error(_: impl std::fmt::Display) -> String {
    "usage index unavailable or incompatible; explicit collection/maintenance required".into()
}
pub(super) fn text(row: &Row, name: &str) -> Result<String, String> {
    row.get(name)
        .and_then(|value| value.as_str().map(str::to_owned))
        .ok_or_else(|| error(name))
}
fn integer(row: &Row, name: &str) -> Result<i64, String> {
    row.get(name)
        .and_then(|value| value.as_i64())
        .ok_or_else(|| error(name))
}

/// Bounded decision for a source whose current accounting generation was verified by its owner.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CollectionProgress {
    /// No published or staged collection exists.
    Missing,
    /// Published contributions already represent the verified generation.
    Current,
    /// Continue durable staging at this exclusive source key.
    Continue(String),
    /// Recorded source generation differs; explicit restart is required.
    Changed,
}

impl UsageIndex {
    /// Prepare a known reporting schema upgrade during explicit collection only.
    /// # Errors
    /// Rejects unknown versions or unverifiable exclusive access; preserves existing data.
    pub async fn prepare_collection(&self) -> Result<(), String> {
        let (_lock, _db) = self.open(true).await?;
        Ok(())
    }

    /// Inspect one source checkpoint without creating storage or reading request contributions.
    /// # Errors
    /// Rejects corrupt checkpoints or incompatible storage; a missing index is reported separately.
    pub async fn collection_progress(
        &self,
        session_id: SessionId,
        generation: SessionUsageGeneration,
    ) -> Result<CollectionProgress, String> {
        match std::fs::symlink_metadata(&self.path) {
            Err(cause) if cause.kind() == std::io::ErrorKind::NotFound => {
                return Ok(CollectionProgress::Missing);
            }
            Err(cause) => return Err(error(cause)),
            Ok(_) => {}
        }
        let (_lock, db) = self.open(false).await?;
        let Some(row) = db
            .select("usage_sessions")
            .where_eq("session_id", session_id.to_string())
            .execute_first(&*db)
            .await
            .map_err(error)?
        else {
            return Ok(CollectionProgress::Missing);
        };
        let recorded: SessionUsageGeneration =
            serde_json::from_str(&text(&row, "generation")?).map_err(error)?;
        if recorded != generation {
            return Ok(CollectionProgress::Changed);
        }
        match integer(&row, "collecting")? {
            0 => Ok(CollectionProgress::Current),
            1 => {
                let next = text(&row, "next_key")?;
                if next.is_empty() || next.len() > 4096 {
                    return Err(error("invalid collection cursor"));
                }
                Ok(CollectionProgress::Continue(next))
            }
            _ => Err(error("invalid collection status")),
        }
    }
    /// Resolve reporting storage beneath an already resolved state location.
    #[must_use]
    pub fn in_state_root(root: &Path) -> Self {
        Self {
            path: root.join("usage.db"),
        }
    }

    async fn open(&self, initialize: bool) -> Result<(std::fs::File, Box<dyn Database>), String> {
        // Reject symlinks before access, including the supplied state directory.
        let parent = self.path.parent().ok_or_else(|| error("parent"))?;
        let canonical_parent = parent.canonicalize().map_err(error)?;
        if canonical_parent != parent {
            return Err(error("noncanonical state root"));
        }
        match std::fs::symlink_metadata(&self.path) {
            Ok(metadata) if !metadata.is_file() || metadata.file_type().is_symlink() => {
                return Err(error("unsafe index path"));
            }
            Ok(_) => {}
            Err(cause) if cause.kind() == std::io::ErrorKind::NotFound && initialize => {}
            Err(cause) => return Err(error(cause)),
        }
        // Fence every access across daemon artifact versions before opening Turso.
        let lock_path = self.path.with_extension("lock");
        if let Ok(metadata) = std::fs::symlink_metadata(&lock_path)
            && (!metadata.is_file() || metadata.file_type().is_symlink())
        {
            return Err(error("unsafe lock"));
        }
        let lock = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(initialize)
            .truncate(false)
            .open(lock_path)
            .map_err(error)?;
        lock.try_lock().map_err(error)?;
        let db = switchy::database_connection::builder()
            .turso()
            .with_path(&self.path)
            .with_busy_timeout(std::time::Duration::from_secs(5))
            .with_multiprocess_wal(false)
            .build()
            .await
            .map_err(error)?;
        let tx = db.begin_transaction().await.map_err(error)?;
        let tables = tx
            .select("sqlite_master")
            .columns(&["name"])
            .where_eq("type", "table")
            .where_not_like("name", "sqlite_%")
            .limit(8)
            .execute(&*tx)
            .await
            .map_err(error)?;
        if tables.is_empty() && initialize {
            create_table("usage_meta")
                .columns(vec![
                    column("id", DataType::BigInt, true),
                    column("version", DataType::BigInt, false),
                    column("revision", DataType::BigInt, false),
                    column("ordinal", DataType::BigInt, false),
                ])
                .primary_key("id")
                .check_integer_equals("id", 1)
                .execute(&*tx)
                .await
                .map_err(error)?;
            tx.insert("usage_meta")
                .value("id", 1)
                .value("version", SCHEMA)
                .value("revision", 0)
                .value("ordinal", 0)
                .execute(&*tx)
                .await
                .map_err(error)?;
            create_time_schema(&*tx).await?;
            create_table("usage_sessions")
                .columns(vec![
                    column("session_id", DataType::Text, true),
                    column("generation", DataType::Text, false),
                    column("next_key", DataType::Text, true),
                    column("collecting", DataType::Int, false),
                ])
                .primary_key("session_id")
                .execute(&*tx)
                .await
                .map_err(error)?;
            create_row_schema(&*tx).await?;
        }
        let meta = tx
            .select("usage_meta")
            .where_eq("id", 1)
            .execute_first(&*tx)
            .await
            .map_err(error)?
            .ok_or_else(|| error("missing schema"))?;
        match integer(&meta, "version")? {
            1 if initialize => upgrade::upgrade_v1(&*tx).await?,
            SCHEMA => {}
            _ => return Err(error("unsupported schema")),
        }
        tx.commit().await.map_err(error)?;
        Ok((lock, db))
    }

    /// Stage one bounded complete-range source page and atomically publish at end-of-source.
    /// Incomplete staging survives restart. Concurrent conflicting collectors fail closed.
    /// # Errors
    /// Rejects invalid pages, changed generations, duplicate contributions, or incompatible storage.
    pub async fn collect(
        &self,
        session_id: SessionId,
        after: Option<&str>,
        page: SessionUsagePage,
    ) -> Result<bool, String> {
        validate_page(after, &page)?;
        let (_lock, db) = self.open(true).await?;
        let tx = db.begin_transaction().await.map_err(error)?;
        let id = session_id.to_string();
        let generation = serde_json::to_string(&page.generation).map_err(error)?;
        let current = tx
            .select("usage_sessions")
            .where_eq("session_id", id.clone())
            .execute_first(&*tx)
            .await
            .map_err(error)?;
        if let Some(after) = after {
            let current = current.ok_or_else(|| error("collection missing"))?;
            if integer(&current, "collecting")? != 1
                || text(&current, "generation")? != generation
                || text(&current, "next_key")? != after
            {
                return Err(error("collection changed"));
            }
        } else {
            validate_restart(current.as_ref(), &generation)?;
            tx.delete("usage_time_nodes")
                .where_eq("tree", format!("{id}/{generation}"))
                .execute(&*tx)
                .await
                .map_err(error)?;
            tx.delete("usage_rows")
                .where_eq("session_id", id.clone())
                .where_eq("staged", 1)
                .execute(&*tx)
                .await
                .map_err(error)?;
        }
        let meta = tx
            .select("usage_meta")
            .where_eq("id", 1)
            .execute_first(&*tx)
            .await
            .map_err(error)?
            .ok_or_else(|| error("meta"))?;
        let mut ordinal = integer(&meta, "ordinal")?;
        for entry in &page.entries {
            index_contribution(&*tx, &id, &generation, entry).await?;
            ordinal = ordinal
                .checked_add(1)
                .ok_or_else(|| error("ordinal overflow"))?;
            tx.insert("usage_rows")
                .value("ordinal", ordinal)
                .value("session_id", id.clone())
                .value("request_key", entry.key.clone())
                .value("entry_json", serde_json::to_string(entry).map_err(error)?)
                .value("staged", 1)
                .execute(&*tx)
                .await
                .map_err(error)?;
        }
        let complete = page.next_after.is_none();
        tx.upsert("usage_sessions")
            .unique(&["session_id"])
            .value("session_id", id.clone())
            .value("generation", generation)
            .value(
                "next_key",
                page.next_after
                    .map_or(DatabaseValue::Null, DatabaseValue::String),
            )
            .value("collecting", i32::from(!complete))
            .execute(&*tx)
            .await
            .map_err(error)?;
        let mut revision = integer(&meta, "revision")?;
        if complete {
            tx.delete("usage_rows")
                .where_eq("session_id", id.clone())
                .where_eq("staged", 0)
                .execute(&*tx)
                .await
                .map_err(error)?;
            tx.update("usage_rows")
                .value("staged", 0)
                .where_eq("session_id", id)
                .where_eq("staged", 1)
                .execute(&*tx)
                .await
                .map_err(error)?;
            revision = revision
                .checked_add(1)
                .ok_or_else(|| error("revision overflow"))?;
        }
        tx.update("usage_meta")
            .value("ordinal", ordinal)
            .value("revision", revision)
            .where_eq("id", 1)
            .execute(&*tx)
            .await
            .map_err(error)?;
        tx.commit().await.map_err(error)?;
        Ok(complete)
    }

    /// Read exact whole-range totals for one completely published session snapshot.
    /// # Errors
    /// Rejects missing, collecting, incompatible, or corrupt state. Source freshness is not implied.
    pub async fn session_range_totals(
        &self,
        session_id: SessionId,
        range: bcode_session_models::SessionCostRange,
    ) -> Result<bcode_usage_models::UsageTotals, String> {
        range.validate()?;
        let (_lock, db) = self.open(false).await?;
        let row = db
            .select("usage_sessions")
            .where_eq("session_id", session_id.to_string())
            .execute_first(&*db)
            .await
            .map_err(error)?
            .ok_or_else(|| error("missing source"))?;
        if integer(&row, "collecting")? != 0 {
            return Err(error("source collecting"));
        }
        crate::time_index::range(
            &*db,
            &format!("{session_id}/{}", text(&row, "generation")?),
            range.from_timestamp_ms,
            range.to_timestamp_ms,
        )
        .await
    }

    /// Read a bounded generation-fenced report page without initializing missing storage.
    /// # Errors
    /// Rejects invalid queries, incompatible storage, corrupt contributions, and changed revisions.
    pub async fn query(&self, query: &UsageQuery) -> Result<UsageReport, String> {
        query.validate()?;
        let (_lock, db) = self.open(false).await?;
        let tx = db.begin_transaction().await.map_err(error)?;
        let meta = tx
            .select("usage_meta")
            .where_eq("id", 1)
            .execute_first(&*tx)
            .await
            .map_err(error)?
            .ok_or_else(|| error("meta"))?;
        let revision = u64::try_from(integer(&meta, "revision")?).map_err(error)?;
        if query.revision.is_some_and(|expected| expected != revision) {
            return Err(error("revision changed"));
        }
        let after = i64::try_from(query.after.unwrap_or_default()).map_err(error)?;
        let rows = tx
            .select("usage_rows")
            .where_eq("staged", 0)
            .where_gt("ordinal", after)
            .sort("ordinal", SortDirection::Asc)
            .limit(query.limit as usize)
            .execute(&*tx)
            .await
            .map_err(error)?;
        let mut requests = Vec::with_capacity(rows.len());
        for row in rows {
            requests.push(UsageRequestRow {
                ordinal: u64::try_from(integer(&row, "ordinal")?).map_err(error)?,
                session_id: text(&row, "session_id")?.parse().map_err(error)?,
                entry: serde_json::from_str(&text(&row, "entry_json")?).map_err(error)?,
            });
        }
        let next = if requests.len() == query.limit as usize {
            requests.last().map(|row| row.ordinal)
        } else {
            None
        };
        let mut query = query.clone();
        query.revision = Some(revision);
        let report = crate::report_page(&query, requests, next,
            "Page subtotal of durable explicitly collected snapshots. Source freshness unverified; recollect before relying on totals.".into())?;
        tx.commit().await.map_err(error)?;
        Ok(report)
    }

    /// Remove unverifiable source contributions and invalidate report cursors atomically.
    /// # Errors
    /// Returns an error for unavailable storage or revision overflow.
    pub async fn invalidate(&self, session_id: SessionId) -> Result<(), String> {
        let (_lock, db) = self.open(false).await?;
        let tx = db.begin_transaction().await.map_err(error)?;
        tx.delete("usage_rows")
            .where_eq("session_id", session_id.to_string())
            .execute(&*tx)
            .await
            .map_err(error)?;
        tx.delete("usage_sessions")
            .where_eq("session_id", session_id.to_string())
            .execute(&*tx)
            .await
            .map_err(error)?;
        let current = tx
            .select("usage_meta")
            .where_eq("id", 1)
            .execute_first(&*tx)
            .await
            .map_err(error)?
            .ok_or_else(|| error("meta"))?;
        let revision = integer(&current, "revision")?
            .checked_add(1)
            .ok_or_else(|| error("revision overflow"))?;
        tx.update("usage_meta")
            .value("revision", revision)
            .where_eq("id", 1)
            .execute(&*tx)
            .await
            .map_err(error)?;
        tx.commit().await.map_err(error)?;
        Ok(())
    }
}

async fn index_contribution(
    db: &dyn Database,
    id: &str,
    generation: &str,
    entry: &bcode_session_models::SessionUsageEntry,
) -> Result<(), String> {
    let mut contribution = bcode_usage_models::UsageTotals::default();
    crate::contribute(&mut contribution, &entry.usage)?;
    crate::time_index::add(
        db,
        &format!("{id}/{generation}"),
        entry.first_observed_at_ms,
        &contribution,
    )
    .await
}

fn validate_restart(current: Option<&Row>, generation: &str) -> Result<(), String> {
    if let Some(current) = current {
        match integer(current, "collecting")? {
            1 if text(current, "generation")? == generation => {
                return Err(error("collection already active; continue its cursor"));
            }
            0 | 1 => {}
            _ => return Err(error("invalid collection status")),
        }
    }
    Ok(())
}

fn validate_page(after: Option<&str>, page: &SessionUsagePage) -> Result<(), String> {
    if page.scanned > 256 || page.entries.len() != page.scanned as usize {
        return Err(error("incomplete range"));
    }
    let mut previous = after.unwrap_or_default();
    for entry in &page.entries {
        if entry.key.as_str() <= previous {
            return Err(error("unordered page"));
        }
        entry.usage.validate()?;
        previous = &entry.key;
    }
    if page
        .next_after
        .as_deref()
        .is_some_and(|next| next <= after.unwrap_or_default() || next != previous)
    {
        return Err(error("continuation"));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use bcode_session_models::{
        SessionCostRange, SessionTokenUsage, SessionUsageEntry, SessionUsageGeneration,
    };
    use bcode_usage_models::USAGE_VERSION;
    fn query() -> UsageQuery {
        UsageQuery {
            version: USAGE_VERSION,
            range: SessionCostRange {
                from_timestamp_ms: 0,
                to_timestamp_ms: 100,
            },
            models: std::collections::BTreeSet::new(),
            providers: std::collections::BTreeSet::new(),
            session_id: None,
            bucket_ms: 100,
            after: None,
            revision: None,
            limit: 256,
        }
    }
    fn page(key: &str, next: Option<&str>) -> SessionUsagePage {
        SessionUsagePage {
            generation: SessionUsageGeneration {
                through_sequence: Some(2),
                cost_revision: 0,
            },
            scanned: 1,
            entries: vec![SessionUsageEntry {
                key: key.into(),
                first_observed_at_ms: 1,
                usage: SessionTokenUsage::default(),
                evidence: None,
            }],
            next_after: next.map(str::to_owned),
        }
    }
    #[tokio::test]
    async fn restart_preserves_staging_and_atomic_publication() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        assert!(index.query(&query()).await.is_err());
        assert!(!root.join("usage.db").exists());
        let session = SessionId::new();
        assert!(
            !index
                .collect(session, None, page("a", Some("a")))
                .await
                .unwrap()
        );
        assert_eq!(index.query(&query()).await.unwrap().totals.requests, 0);
        drop(index);
        let index = UsageIndex::in_state_root(&root);
        assert!(
            index
                .collect(session, Some("a"), page("b", None))
                .await
                .unwrap()
        );
        let report = index.query(&query()).await.unwrap();
        assert_eq!(report.totals.requests, 2);
        assert!(index.collect(session, None, page("a", None)).await.unwrap());
        assert_eq!(index.query(&query()).await.unwrap().totals.requests, 1);
        let mut stale = query();
        stale.revision = Some(report.revision);
        assert!(index.query(&stale).await.is_err());
        index.invalidate(session).await.unwrap();
        assert_eq!(index.query(&query()).await.unwrap().totals.requests, 0);
    }
    #[tokio::test]
    async fn progress_distinguishes_missing_staged_current_and_repriced() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        let session = SessionId::new();
        let generation = page("a", None).generation;
        assert_eq!(
            index
                .collection_progress(session, generation)
                .await
                .unwrap(),
            CollectionProgress::Missing
        );
        assert!(!root.join("usage.db").exists());
        index
            .collect(session, None, page("a", Some("a")))
            .await
            .unwrap();
        assert_eq!(
            index
                .collection_progress(session, generation)
                .await
                .unwrap(),
            CollectionProgress::Continue("a".into())
        );
        index
            .collect(session, Some("a"), page("b", None))
            .await
            .unwrap();
        assert_eq!(
            index
                .collection_progress(session, generation)
                .await
                .unwrap(),
            CollectionProgress::Current
        );
        let mut repriced = generation;
        repriced.cost_revision += 1;
        assert_eq!(
            index.collection_progress(session, repriced).await.unwrap(),
            CollectionProgress::Changed
        );
    }

    #[tokio::test]
    async fn changed_source_restarts_staging_and_rejects_old_continuation() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        let session = SessionId::new();
        index
            .collect(session, None, page("a", Some("a")))
            .await
            .unwrap();
        let mut changed = page("a", Some("a"));
        changed.generation.cost_revision = 1;
        index.collect(session, None, changed).await.unwrap();
        assert!(
            index
                .collect(session, Some("a"), page("b", None))
                .await
                .is_err()
        );
        let mut final_page = page("b", None);
        final_page.generation.cost_revision = 1;
        assert!(index.collect(session, Some("a"), final_page).await.unwrap());
        assert_eq!(index.query(&query()).await.unwrap().totals.requests, 2);
    }

    #[tokio::test]
    async fn time_range_totals_are_exact_across_restart() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        let id = SessionId::new();
        let mut first = page("a", Some("a"));
        first.entries[0].first_observed_at_ms = 10;
        index.collect(id, None, first).await.unwrap();
        let mut second = page("b", None);
        second.entries[0].first_observed_at_ms = 20;
        index.collect(id, Some("a"), second).await.unwrap();
        let index = UsageIndex::in_state_root(&root);
        for (from, to, count) in [(0, 10, 0), (10, 20, 1), (10, 21, 2), (21, 100, 0)] {
            let totals = index
                .session_range_totals(
                    id,
                    SessionCostRange {
                        from_timestamp_ms: from,
                        to_timestamp_ms: to,
                    },
                )
                .await
                .unwrap();
            assert_eq!(totals.requests, count);
        }
    }

    #[tokio::test]
    async fn structured_schema_preserves_constraints_and_indexes() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        let (_lock, db) = index.open(true).await.unwrap();
        for name in [
            "usage_time_tree",
            "usage_session_rows",
            "usage_published_rows",
        ] {
            assert!(
                db.select("sqlite_master")
                    .where_eq("type", "index")
                    .where_eq("name", name)
                    .limit(1)
                    .execute_first(&*db)
                    .await
                    .unwrap()
                    .is_some()
            );
        }
        // Preserve SQLite's nullable TEXT primary keys, not an implicit NOT NULL.
        db.insert("usage_time_nodes")
            .value("node_key", DatabaseValue::Null)
            .value("tree", "tree")
            .value("totals_json", "{}")
            .execute(&*db)
            .await
            .unwrap();
        assert!(
            db.insert("usage_time_nodes")
                .value("node_key", "key")
                .value("tree", DatabaseValue::Null)
                .value("totals_json", "{}")
                .execute(&*db)
                .await
                .is_err()
        );
        db.insert("usage_sessions")
            .value("session_id", DatabaseValue::Null)
            .value("generation", "generation")
            .value("next_key", DatabaseValue::Null)
            .value("collecting", 0)
            .execute(&*db)
            .await
            .unwrap();
        assert!(
            db.update("usage_meta")
                .value("id", 2)
                .where_eq("id", 1)
                .execute(&*db)
                .await
                .is_err()
        );
        for ordinal in [1, 2] {
            let result = db
                .insert("usage_rows")
                .value("ordinal", ordinal)
                .value("session_id", "session")
                .value("request_key", "request")
                .value("entry_json", "{}")
                .value("staged", 0)
                .execute(&*db)
                .await;
            assert_eq!(result.is_ok(), ordinal == 1);
        }
        // The staged copy is distinct from the published copy of the same request.
        db.insert("usage_rows")
            .value("ordinal", DatabaseValue::Null)
            .value("session_id", "session")
            .value("request_key", "request")
            .value("entry_json", "{}")
            .value("staged", 1)
            .execute(&*db)
            .await
            .unwrap();
        let rows = db.select("usage_rows").execute(&*db).await.unwrap();
        assert_eq!(rows.len(), 2);
        assert!(rows.iter().all(|row| integer(row, "ordinal").is_ok()));
    }

    #[tokio::test]
    async fn unrelated_schema_is_not_initialized() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        std::fs::create_dir_all(index.path.parent().unwrap()).unwrap();
        let db = switchy::database_connection::builder()
            .turso()
            .with_path(&index.path)
            .with_multiprocess_wal(false)
            .build()
            .await
            .unwrap();
        for n in 0..10 {
            create_table(&format!("unrelated_{n}"))
                .column(column("id", DataType::BigInt, true))
                .execute(&*db)
                .await
                .unwrap();
        }
        drop(db);
        assert!(index.open(true).await.is_err());
        let db = switchy::database_connection::builder()
            .turso()
            .with_path(&index.path)
            .with_multiprocess_wal(false)
            .build()
            .await
            .unwrap();
        assert!(
            db.select("sqlite_master")
                .where_eq("name", "usage_meta")
                .limit(1)
                .execute(&*db)
                .await
                .unwrap()
                .is_empty()
        );
        assert_eq!(
            db.select("sqlite_master")
                .where_eq("type", "table")
                .where_not_like("name", "sqlite_%")
                .limit(8)
                .execute(&*db)
                .await
                .unwrap()
                .len(),
            8
        );
    }

    #[tokio::test]
    async fn v1_upgrade_is_explicit_and_preserves_contributions() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        let id = SessionId::new();
        index.collect(id, None, page("a", None)).await.unwrap();
        let (lock, db) = index.open(false).await.unwrap();
        switchy::database::schema::drop_table("usage_time_nodes")
            .execute(&*db)
            .await
            .unwrap();
        db.update("usage_meta")
            .value("version", 1)
            .where_eq("id", 1)
            .execute(&*db)
            .await
            .unwrap();
        drop(db);
        drop(lock);
        assert!(index.query(&query()).await.is_err());
        index.prepare_collection().await.unwrap();
        assert_eq!(index.query(&query()).await.unwrap().totals.requests, 1);
        assert_eq!(
            index
                .session_range_totals(id, query().range)
                .await
                .unwrap()
                .requests,
            1
        );
    }

    #[tokio::test]
    async fn future_schema_is_preserved_and_rejected() {
        let root = tempfile::tempdir().unwrap();
        let root = root.path().canonicalize().unwrap();
        let index = UsageIndex::in_state_root(&root);
        index
            .collect(SessionId::new(), None, page("a", None))
            .await
            .unwrap();
        let (lock, db) = index.open(false).await.unwrap();
        db.update("usage_meta")
            .value("version", 999)
            .where_eq("id", 1)
            .execute(&*db)
            .await
            .unwrap();
        drop(db);
        drop(lock);
        assert!(index.query(&query()).await.is_err());
        assert!(
            index
                .collect(SessionId::new(), None, page("b", None))
                .await
                .is_err()
        );
    }
}