leankg 0.19.1

use cozo::{DataValue, NamedRows, ScriptMutability};
use sha2::{Digest, Sha256};
use std::path::Path;

pub type CozoDb = cozo::DbInstance;

/// Thin wrapper around `DbInstance::run_script` that absorbs the CozoDB 0.7.x
/// API changes (third `ScriptMutability` argument, `BTreeMap<String, DataValue>`
/// params type) so the rest of the codebase keeps the old 2-arg
/// `BTreeMap<String, serde_json::Value>` calling convention.
///
/// Mutability is auto-detected from the leading operator in the query string.
/// The full operator surface in this codebase is `:put :rm :create :replace`
/// and `PRAGMA` (writes) plus bare `?`/`::relations` reads — verified by grep.
/// New write operators (e.g. `::hnsw create`) must be added to `WRITE_PREFIXES`.
pub fn run_script(
    db: &CozoDb,
    query: &str,
    params: std::collections::BTreeMap<String, serde_json::Value>,
) -> Result<NamedRows, cozo::Error> {
    let cozo_params: std::collections::BTreeMap<String, DataValue> = params
        .into_iter()
        .map(|(k, v)| (k, json_to_datavalue(v)))
        .collect();
    db.run_script(query, cozo_params, mutability_for(query))
}

/// Convert serde_json::Value to cozo::DataValue using CozoDB's own conversion
/// semantics (matches `impl From<JsonValue> for DataValue` in cozo 0.7.6).
/// Importantly, `JsonValue::Null` becomes `DataValue::Null` (not `Bot` — `Bot`
/// is the bottom type and is rejected by nullable column coercion).
fn json_to_datavalue(v: serde_json::Value) -> DataValue {
    DataValue::from(v)
}

fn mutability_for(query: &str) -> ScriptMutability {
    // CozoDB 0.7.x enforces that Immutable scripts cannot acquire write locks.
    // A Datalog statement can combine a read head (`?[...] := ...`) with an
    // action operator (`:put`, `:rm`, etc.) in one script — the leading char
    // is `?` but the action still needs a write lock. So we scan the whole
    // query for any write operator, not just the prefix.
    const WRITE_TOKENS: &[&str] = &[
        ":put",
        ":rm",
        ":create",
        ":replace",
        ":delete",
        ":update",
        ":insert",
        "PRAGMA",
        "::set_triggers",
        "::hnsw",
        "::lsh",
        "::fts",
        "::index",
    ];
    if WRITE_TOKENS.iter().any(|t| query.contains(t)) {
        ScriptMutability::Mutable
    } else {
        ScriptMutability::Immutable
    }
}

const DEFAULT_ROCKSDB_ROOT: &str = ".leankg-rocksdb";

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StorageEngine {
    Sqlite,
    RocksDb,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageConfig {
    pub engine: StorageEngine,
    pub path: std::path::PathBuf,
}

fn get_env_mmap_size() -> u64 {
    std::env::var("LEANKG_MMAP_SIZE")
        .ok()
        .and_then(|v| v.parse().ok())
        // Default 64 MiB. SQLite's mmap'd region is resident in the process RSS,
        // so a smaller mmap dramatically reduces the container's memory footprint
        // for large codebases. 256 MiB was far too aggressive for the common case
        // (it pushed containers past their memory limits and triggered OOM kills).
        .unwrap_or(64 * 1024 * 1024)
}

pub fn init_db(db_path: &Path) -> Result<CozoDb, Box<dyn std::error::Error>> {
    let storage = resolve_storage_config(db_path);
    if let Some(parent) = storage.path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    let path_str = storage.path.to_string_lossy().to_string();
    let db = match storage.engine {
        StorageEngine::Sqlite => cozo::DbInstance::new("sqlite", &path_str, "")?,
        StorageEngine::RocksDb => {
            std::fs::create_dir_all(&storage.path)?;
            cozo::DbInstance::new("rocksdb", &path_str, "")?
        }
    };

    let mmap_size = get_env_mmap_size();
    tracing::info!(
        "Cozo storage = {:?} at {} (LEANKG_MMAP_SIZE={})",
        storage.engine,
        storage.path.display(),
        mmap_size
    );

    // CozoDB 0.7.x no longer accepts raw PRAGMA statements — the parser rejects
    // them silently. SQLite tuning is now done by CozoDB internally based on
    // the storage engine's own defaults. LEANKG_MMAP_SIZE is captured for
    // diagnostic logging only.
    let _ = mmap_size;

    init_schema(&db)?;

    Ok(db)
}

pub fn resolve_storage_config(db_path: &Path) -> StorageConfig {
    match std::env::var("LEANKG_DB_ENGINE")
        .unwrap_or_else(|_| "sqlite".to_string())
        .to_ascii_lowercase()
        .as_str()
    {
        "rocksdb" | "rocks" | "rockdb" => StorageConfig {
            engine: StorageEngine::RocksDb,
            path: central_project_storage_path(db_path),
        },
        _ => StorageConfig {
            engine: StorageEngine::Sqlite,
            path: if db_path.is_dir() {
                db_path.join("leankg.db")
            } else {
                db_path.to_path_buf()
            },
        },
    }
}

pub(crate) fn central_project_storage_path(db_path: &Path) -> std::path::PathBuf {
    let root = std::env::var_os("LEANKG_ROCKSDB_ROOT")
        .map(std::path::PathBuf::from)
        .or_else(|| dirs::home_dir().map(|home| home.join(DEFAULT_ROCKSDB_ROOT)))
        .unwrap_or_else(|| std::path::PathBuf::from(DEFAULT_ROCKSDB_ROOT));

    // Walk up to the project root regardless of which form the caller passed.
    //
    // Callers historically varied on what they passed to `init_db`:
    //   * `Index` / `McpHttp`     -> `<project>/.leankg`        (the directory)
    //   * `Embed` / `SemanticContext` / `SmokeTest` -> `<project>/.leankg/leankg.db` (the file)
    //
    // If we used the file path directly, the SHA-256 hash of the project key
    // would differ from the indexer, so `leankg embed` would open a brand
    // new (empty) RocksDB at a different path, find zero `code_elements`,
    // and silently report "Embedded: 0" — leaving the agent with a working
    // ontology-first `semantic_search` and a non-functional HNSW path.
    //
    // FR-HNSW-C fix: accept both forms and always hash the project root,
    // not the .leankg directory or the database file inside it.
    let project_root = project_root_from_db_path(db_path);
    let project_key = project_root
        .canonicalize()
        .unwrap_or_else(|_| project_root.to_path_buf());
    let project_key = project_key.to_string_lossy();
    let mut hasher = Sha256::new();
    hasher.update(project_key.as_bytes());
    let hash = format!("{:x}", hasher.finalize());
    let name = project_root
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("project")
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
                ch
            } else {
                '-'
            }
        })
        .collect::<String>();

    root.join("projects")
        .join(format!("{}-{}", name, &hash[..12]))
}

/// Reduce a `db_path` to the project root, regardless of which form the
/// caller passed (the `.leankg` directory, the `leankg.db` file inside it,
/// or an unrelated path).
///
/// * `<project>/.leankg/leankg.db` -> `<project>`
/// * `<project>/.leankg`           -> `<project>`
/// * `<project>`                   -> `<project>`
/// * `leankg.db`                   -> `.` (no parent anchor — caller error)
fn project_root_from_db_path(db_path: &Path) -> std::path::PathBuf {
    let file_name = db_path.file_name().and_then(|n| n.to_str());
    if file_name == Some("leankg.db") {
        // The file lives inside `.leankg/`. Walk up two levels to the
        // project root, tolerating a missing `.leankg` parent (the file
        // may have been moved out by a manual export).
        if let Some(leankg_dir) = db_path.parent() {
            if leankg_dir.file_name().and_then(|n| n.to_str()) == Some(".leankg") {
                if let Some(project) = leankg_dir.parent() {
                    return project.to_path_buf();
                }
            }
            return leankg_dir.to_path_buf();
        }
        return std::path::PathBuf::from(".");
    }
    if file_name == Some(".leankg") {
        return db_path.parent().unwrap_or(db_path).to_path_buf();
    }
    db_path.to_path_buf()
}

fn init_schema(db: &CozoDb) -> Result<(), Box<dyn std::error::Error>> {
    let check_relations = r#"::relations"#;
    let relations_result = run_script(db, check_relations, Default::default())?;
    let existing_relations: std::collections::HashSet<String> = relations_result
        .rows
        .iter()
        .filter_map(|row| row.first().and_then(|v| v.get_str().map(String::from)))
        .collect();

    if !existing_relations.contains("code_elements") {
        let create_code_elements = r#":create code_elements {qualified_name: String, element_type: String, name: String, file_path: String, line_start: Int, line_end: Int, language: String, parent_qualified: String?, cluster_id: String?, cluster_label: String?, metadata: String, env: String default 'local', ontology_layer: String default 'procedural'}"#;
        if let Err(e) = run_script(db, create_code_elements, Default::default()) {
            eprintln!("Failed to create code_elements: {:?}", e);
        }
    } else {
        let create_file_path_index =
            r#"::index create code_elements:file_path_index { file_path }"#;
        if let Err(e) = run_script(db, create_file_path_index, Default::default()) {
            tracing::debug!("file_path index may already exist: {:?}", e);
        }

        let create_qualified_name_index =
            r#"::index create code_elements:qualified_name_index { qualified_name }"#;
        if let Err(e) = run_script(db, create_qualified_name_index, Default::default()) {
            tracing::debug!("qualified_name index may already exist: {:?}", e);
        }

        let create_element_type_index =
            r#"::index create code_elements:element_type_index { element_type }"#;
        if let Err(e) = run_script(db, create_element_type_index, Default::default()) {
            tracing::debug!("element_type index may already exist: {:?}", e);
        }

        let create_parent_qualified_index =
            r#"::index create code_elements:parent_qualified_index { parent_qualified }"#;
        if let Err(e) = run_script(db, create_parent_qualified_index, Default::default()) {
            tracing::debug!("parent_qualified index may already exist: {:?}", e);
        }

        validate_code_elements_schema(db)?;
    }

    if !existing_relations.contains("relationships") {
        let create_relationships = r#":create relationships {source_qualified: String, target_qualified: String, rel_type: String, confidence: Float, metadata: String, env: String default 'local'}"#;
        if let Err(e) = run_script(db, create_relationships, Default::default()) {
            eprintln!("Failed to create relationships: {:?}", e);
        }
    } else {
        let create_rel_type_index = r#"::index create relationships:rel_type_index { rel_type }"#;
        if let Err(e) = run_script(db, create_rel_type_index, Default::default()) {
            tracing::debug!("rel_type index may already exist: {:?}", e);
        }

        let create_target_index =
            r#"::index create relationships:target_qualified_index { target_qualified }"#;
        if let Err(e) = run_script(db, create_target_index, Default::default()) {
            tracing::debug!("target_qualified index may already exist: {:?}", e);
        }

        // NOTE: source_qualified_index is created for each DB to avoid migration issues
        // See get_relationships_for_elements_optimized in query.rs

        validate_relationships_schema(db)?;
    }

    if !existing_relations.contains("business_logic") {
        let create_business_logic = r#":create business_logic {element_qualified: String, description: String, user_story_id: String?, feature_id: String?}"#;
        if let Err(e) = run_script(db, create_business_logic, Default::default()) {
            eprintln!("Failed to create business_logic: {:?}", e);
        }
    }

    if !existing_relations.contains("context_metrics") {
        let create_context_metrics = r#":create context_metrics {tool_name: String, timestamp: Int, project_path: String, input_tokens: Int, output_tokens: Int, output_elements: Int, execution_time_ms: Int, baseline_tokens: Int, baseline_lines_scanned: Int, tokens_saved: Int, savings_percent: Float, correct_elements: Int?, total_expected: Int?, f1_score: Float?, query_pattern: String?, query_file: String?, query_depth: Int?, success: Bool, is_deleted: Bool}"#;
        if let Err(e) = run_script(db, create_context_metrics, Default::default()) {
            eprintln!("Failed to create context_metrics: {:?}", e);
        }

        let create_tool_index = r#"::index create context_metrics:tool_name_index { tool_name }"#;
        if let Err(e) = run_script(db, create_tool_index, Default::default()) {
            tracing::debug!("tool_name index may already exist: {:?}", e);
        }

        let create_timestamp_index =
            r#"::index create context_metrics:timestamp_index { timestamp }"#;
        if let Err(e) = run_script(db, create_timestamp_index, Default::default()) {
            tracing::debug!("timestamp index may already exist: {:?}", e);
        }

        let create_project_index =
            r#"::index create context_metrics:project_path_index { project_path }"#;
        if let Err(e) = run_script(db, create_project_index, Default::default()) {
            tracing::debug!("project_path index may already exist: {:?}", e);
        }
    }

    if !existing_relations.contains("query_cache") {
        let create_query_cache = r#":create query_cache {cache_key: String, value_json: String, created_at: Int, ttl_seconds: Int, tool_name: String, project_path: String, metadata: String}"#;
        if let Err(e) = run_script(db, create_query_cache, Default::default()) {
            eprintln!("Failed to create query_cache: {:?}", e);
        }

        let create_key_index = r#"::index create query_cache:cache_key_index { cache_key }"#;
        if let Err(e) = run_script(db, create_key_index, Default::default()) {
            tracing::debug!("cache_key index may already exist: {:?}", e);
        }

        let create_tool_index = r#"::index create query_cache:tool_name_index { tool_name }"#;
        if let Err(e) = run_script(db, create_tool_index, Default::default()) {
            tracing::debug!("tool_name index may already exist: {:?}", e);
        }
    }

    // Run migrations for schema evolution
    run_migrations(db, &existing_relations)?;

    // Do not rely solely on the migration ledger for canonical graph arity.
    // Some older databases recorded migration 006 while retaining the
    // pre-env 11-column code_elements relation, which breaks all current
    // graph queries at runtime.
    repair_canonical_schema(db, &existing_relations)?;

    // Create service_metadata table if not exists (idempotent via migration)
    if !existing_relations.contains("service_metadata") {
        let create_svc = r#":create service_metadata {service_name: String, env: String default 'local', team: String?, on_call: String?, repo_url: String?, language: String?, health_endpoint: String?, slo_p99_ms: Int?, incident_count: Int, last_incident: Int?, tags: String, version: String?, deploy_envs: String, created_at: Int, updated_at: Int}"#;
        if let Err(e) = run_script(db, create_svc, Default::default()) {
            tracing::warn!("Failed to create service_metadata: {:?}", e);
        }
        let svc_indexes = [
            r#"::index create service_metadata:svc_name_index { service_name }"#,
            r#"::index create service_metadata:svc_env_index { env }"#,
        ];
        for idx in &svc_indexes {
            if let Err(e) = run_script(db, idx, Default::default()) {
                tracing::debug!("service_metadata index note: {:?}", e);
            }
        }
    }

    // Create teams table for shared graph management
    if !existing_relations.contains("teams") {
        let create_teams = r#":create teams {id: String, name: String, description: String, owner_id: String, created_at: Int, updated_at: Int, graph_read_users: String, graph_write_users: String, members: String}"#;
        if let Err(e) = run_script(db, create_teams, Default::default()) {
            tracing::warn!("Failed to create teams: {:?}", e);
        }
        let team_indexes = [r#"::index create teams:owner_index { owner_id }"#];
        for idx in &team_indexes {
            if let Err(e) = run_script(db, idx, Default::default()) {
                tracing::debug!("teams index note: {:?}", e);
            }
        }
    }

    // Create team_invites table for onboarding workflow
    if !existing_relations.contains("team_invites") {
        let create_invites = r#":create team_invites {token: String, team_id: String, email: String?, role: String, created_by: String, created_at: Int, expires_at: Int, accepted: Bool, accepted_by: String?}"#;
        if let Err(e) = run_script(db, create_invites, Default::default()) {
            tracing::warn!("Failed to create team_invites: {:?}", e);
        }
        let invite_indexes = [
            r#"::index create team_invites:team_index { team_id }"#,
            r#"::index create team_invites:token_index { token }"#,
        ];
        for idx in &invite_indexes {
            if let Err(e) = run_script(db, idx, Default::default()) {
                tracing::debug!("team_invites index note: {:?}", e);
            }
        }
    }

    // Embedding-state table (only when the `embeddings` feature is compiled in).
    // Without the feature, the table is never created and `embeddings::*`
    // calls are absent from the binary — keeps default builds lean.
    #[cfg(feature = "embeddings")]
    {
        crate::embeddings::state::ensure_embedding_state_table(db)?;
    }

    Ok(())
}

fn run_migrations(
    db: &CozoDb,
    existing_relations: &std::collections::HashSet<String>,
) -> Result<(), Box<dyn std::error::Error>> {
    // Create migrations tracking table if not exists
    if !existing_relations.contains("migrations") {
        let create_migrations = r#":create migrations {id: String, applied_at: Int}"#;
        if let Err(e) = run_script(db, create_migrations, Default::default()) {
            tracing::warn!("Failed to create migrations table: {:?}", e);
        }
    }

    // Get already-applied migrations
    let applied: std::collections::HashSet<String> =
        run_script(db, "?[id] := *migrations[id, _]", Default::default())
            .map(|r| {
                r.rows
                    .iter()
                    .filter_map(|row| row.first().and_then(|v| v.get_str().map(String::from)))
                    .collect()
            })
            .unwrap_or_default();

    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64;

    // Migration 001: Create knowledge_entries table
    if !applied.contains("001_knowledge_entries") {
        tracing::info!("Running migration 001_knowledge_entries...");
        let create_knowledge = r#":create knowledge_entries {id: String, knowledge_type: String, title: String, content: String, element_qualified: String?, user_story_id: String?, feature_id: String?, tags: String, environment: String, branch: String?, author: String, created_at: Int, updated_at: Int}"#;
        if let Err(e) = run_script(db, create_knowledge, Default::default()) {
            tracing::warn!("Migration 001 failed (may already exist): {:?}", e);
        }

        // Create indexes
        let indexes = [
            r#"::index create knowledge_entries:type_index { knowledge_type }"#,
            r#"::index create knowledge_entries:element_index { element_qualified }"#,
            r#"::index create knowledge_entries:env_index { environment }"#,
            r#"::index create knowledge_entries:author_index { author }"#,
        ];
        for idx in &indexes {
            if let Err(e) = run_script(db, idx, Default::default()) {
                tracing::debug!("Index creation note: {:?}", e);
            }
        }

        record_migration(db, "001_knowledge_entries", now)?;
    }

    // Migration 002-005 have been consolidated into migration 006.
    // The old migration IDs are recorded as applied to skip the stacked
    // :replace chain that caused schema drift (environment vs env columns).
    mark_legacy_migrations_as_applied(db, &applied, now)?;

    // Migration 006: Safe canonical schema repair for code_elements and
    // relationships, plus incident table creation. Replaces the old 002-005
    // stacked :replace chain. This migration is idempotent: it inspects the
    // current column count and only performs a :replace when the schema
    // does not match the canonical 13-column (code_elements) or 6-column
    // (relationships) layout. Non-matching schemas (e.g. with extra
    // environment/branch/version_tag columns from old partial migrations)
    // are repaired to the canonical form.
    if !applied.contains("006_safe_canonical_schema_repair") {
        tracing::info!("Running migration 006_safe_canonical_schema_repair...");
        repair_canonical_schema(db, existing_relations)?;
        record_migration(db, "006_safe_canonical_schema_repair", now)?;
    }

    Ok(())
}

fn mark_legacy_migrations_as_applied(
    db: &CozoDb,
    applied: &std::collections::HashSet<String>,
    now: i64,
) -> Result<(), Box<dyn std::error::Error>> {
    let legacy_ids = [
        "002_code_elements_versioning",
        "003_business_logic_versioning",
        "004_env_and_incidents",
        "005_canonical_env_graph_schema",
    ];
    for id in &legacy_ids {
        if !applied.contains(*id) {
            record_migration(db, id, now)?;
        }
    }
    Ok(())
}

fn repair_canonical_schema(
    db: &CozoDb,
    existing_relations: &std::collections::HashSet<String>,
) -> Result<(), Box<dyn std::error::Error>> {
    ensure_canonical_code_elements(db, existing_relations)?;
    ensure_canonical_relationships(db, existing_relations)?;
    if let Err(e) = ensure_incidents_table(db) {
        tracing::warn!("incidents table creation failed: {:?}", e);
    }
    Ok(())
}

const REPAIR_LEGACY_CODE_ELEMENTS_11_TO_13: &str = r#"
?[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata, env, ontology_layer] :=
    *code_elements[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata],
    env = "local",
    ontology_layer = "procedural"
:replace code_elements {qualified_name: String, element_type: String, name: String, file_path: String, line_start: Int, line_end: Int, language: String, parent_qualified: String?, cluster_id: String?, cluster_label: String?, metadata: String, env: String default 'local', ontology_layer: String default 'procedural'}
"#;
const REPAIR_LEGACY_CODE_ELEMENTS_12_TO_13: &str = r#"
?[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata, env, ontology_layer] :=
    *code_elements[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata, env],
    ontology_layer = "procedural"
:replace code_elements {qualified_name: String, element_type: String, name: String, file_path: String, line_start: Int, line_end: Int, language: String, parent_qualified: String?, cluster_id: String?, cluster_label: String?, metadata: String, env: String default 'local', ontology_layer: String default 'procedural'}
"#;
const REPAIR_LEGACY_RELATIONSHIPS_5_TO_6: &str = r#"
?[source_qualified, target_qualified, rel_type, confidence, metadata, env] :=
    *relationships[source_qualified, target_qualified, rel_type, confidence, metadata],
    env = "local"
:replace relationships {source_qualified: String, target_qualified: String, rel_type: String, confidence: Float, metadata: String, env: String default 'local'}
"#;

fn get_column_count(db: &CozoDb, relation: &str) -> usize {
    let arity_probe = match relation {
        "code_elements" => Some(vec![
            (
                13,
                "?[qualified_name] := *code_elements[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata, env, ontology_layer] :limit 0",
            ),
            (
                12,
                "?[qualified_name] := *code_elements[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata, env] :limit 0",
            ),
            (
                11,
                "?[qualified_name] := *code_elements[qualified_name, element_type, name, file_path, line_start, line_end, language, parent_qualified, cluster_id, cluster_label, metadata] :limit 0",
            ),
        ]),
        "relationships" => Some(vec![
            (
                6,
                "?[source_qualified] := *relationships[source_qualified, target_qualified, rel_type, confidence, metadata, env] :limit 0",
            ),
            (
                5,
                "?[source_qualified] := *relationships[source_qualified, target_qualified, rel_type, confidence, metadata] :limit 0",
            ),
        ]),
        _ => None,
    };

    if let Some(probes) = arity_probe {
        for (arity, query) in probes {
            if run_script(db, query, Default::default()).is_ok() {
                return arity;
            }
        }
    }

    let query = format!(":schema {}", relation);
    run_script(db, &query, Default::default())
        .map(|r| r.rows.len())
        .unwrap_or(0)
}

/// Canonical column lists per arity, keyed by relation name. Used by
/// `get_relation_schema` to translate an arity probe into a concrete list
/// of column names. Keep in sync with `:create` statements above and with
/// the repair scripts (`REPAIR_LEGACY_CODE_ELEMENTS_*`, `REPAIR_LEGACY_RELATIONSHIPS_5_TO_6`).
const CODE_ELEMENTS_13_COLUMNS: &[&str] = &[
    "qualified_name",
    "element_type",
    "name",
    "file_path",
    "line_start",
    "line_end",
    "language",
    "parent_qualified",
    "cluster_id",
    "cluster_label",
    "metadata",
    "env",
    "ontology_layer",
];

const CODE_ELEMENTS_12_COLUMNS: &[&str] = &[
    "qualified_name",
    "element_type",
    "name",
    "file_path",
    "line_start",
    "line_end",
    "language",
    "parent_qualified",
    "cluster_id",
    "cluster_label",
    "metadata",
    "env",
];

const CODE_ELEMENTS_11_COLUMNS: &[&str] = &[
    "qualified_name",
    "element_type",
    "name",
    "file_path",
    "line_start",
    "line_end",
    "language",
    "parent_qualified",
    "cluster_id",
    "cluster_label",
    "metadata",
];

const RELATIONSHIPS_6_COLUMNS: &[&str] = &[
    "source_qualified",
    "target_qualified",
    "rel_type",
    "confidence",
    "metadata",
    "env",
];

const RELATIONSHIPS_5_COLUMNS: &[&str] = &[
    "source_qualified",
    "target_qualified",
    "rel_type",
    "confidence",
    "metadata",
];

/// Schema snapshot for one CozoDB relation. Returned by `get_relation_schema`
/// and consumed by callers that need to write arity-correct Datalog rules
/// (e.g. the ontology self-test in `mcp/kg_self_test.rs`).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RelationSchema {
    pub name: String,
    pub arity: usize,
    pub columns: Vec<String>,
    pub canonical: bool,
}

/// Returns the live schema for the named relation. `columns` is the ordered
/// list of column names as the relation is currently defined; `canonical`
/// is true when the live arity matches the current canonical schema
/// (13 for code_elements, 6 for relationships).
pub fn get_relation_schema(db: &CozoDb, relation: &str) -> RelationSchema {
    let arity = get_column_count(db, relation);
    let columns: Vec<String> = match relation {
        "code_elements" => match arity {
            13 => CODE_ELEMENTS_13_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect(),
            12 => CODE_ELEMENTS_12_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect(),
            11 => CODE_ELEMENTS_11_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect(),
            _ => Vec::new(),
        },
        "relationships" => match arity {
            6 => RELATIONSHIPS_6_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect(),
            5 => RELATIONSHIPS_5_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect(),
            _ => Vec::new(),
        },
        _ => Vec::new(),
    };
    let canonical = match relation {
        "code_elements" => arity == 13,
        "relationships" => arity == 6,
        _ => arity > 0,
    };
    RelationSchema {
        name: relation.to_string(),
        arity,
        columns,
        canonical,
    }
}

/// Convenience accessor for the code_elements relation.
pub fn code_elements_schema(db: &CozoDb) -> RelationSchema {
    get_relation_schema(db, "code_elements")
}

/// Convenience accessor for the relationships relation.
pub fn relationships_schema(db: &CozoDb) -> RelationSchema {
    get_relation_schema(db, "relationships")
}

fn ensure_canonical_code_elements(
    db: &CozoDb,
    existing_relations: &std::collections::HashSet<String>,
) -> Result<(), Box<dyn std::error::Error>> {
    if !existing_relations.contains("code_elements") {
        return Ok(());
    }
    const EXPECTED: usize = 13;
    let current = get_column_count(db, "code_elements");
    if current == EXPECTED {
        tracing::info!(
            "code_elements schema already canonical ({} columns), skipping replace",
            current
        );
        return Ok(());
    }
    tracing::info!(
        "code_elements schema has {} columns (expected {}), applying canonical :replace",
        current,
        EXPECTED
    );
    // CozoDB 0.7.6 refuses to :replace a relation that has indices — drop them
    // first, then recreate after the replace. Best-effort: legacy DBs may have
    // a different index set, so we tolerate "index not found" errors.
    for idx in &[
        "file_path_index",
        "qualified_name_index",
        "element_type_index",
        "parent_qualified_index",
    ] {
        let _ = run_script(
            db,
            &format!("::index drop code_elements:{}", idx),
            Default::default(),
        );
    }
    match current {
        11 => {
            run_script(db, REPAIR_LEGACY_CODE_ELEMENTS_11_TO_13, Default::default())?;
        }
        12 => {
            run_script(db, REPAIR_LEGACY_CODE_ELEMENTS_12_TO_13, Default::default())?;
        }
        _ => {
            tracing::warn!(
                "code_elements schema has unsupported arity {}; canonical repair only supports legacy 11- or 12-column schema",
                current
            );
            return Ok(());
        }
    }
    tracing::info!("code_elements :replace successful, recreating indices");
    // Recreate indices dropped before the :replace (CozoDB 0.7.6 requirement).
    for idx_query in &[
        r#"::index create code_elements:file_path_index { file_path }"#,
        r#"::index create code_elements:qualified_name_index { qualified_name }"#,
        r#"::index create code_elements:element_type_index { element_type }"#,
        r#"::index create code_elements:parent_qualified_index { parent_qualified }"#,
    ] {
        let _ = run_script(db, idx_query, Default::default());
    }
    Ok(())
}

fn ensure_canonical_relationships(
    db: &CozoDb,
    existing_relations: &std::collections::HashSet<String>,
) -> Result<(), Box<dyn std::error::Error>> {
    if !existing_relations.contains("relationships") {
        return Ok(());
    }
    const EXPECTED: usize = 6;
    let current = get_column_count(db, "relationships");
    if current == EXPECTED {
        tracing::info!(
            "relationships schema already canonical ({} columns), skipping replace",
            current
        );
        return Ok(());
    }
    tracing::info!(
        "relationships schema has {} columns (expected {}), applying canonical :replace",
        current,
        EXPECTED
    );
    if current != 5 {
        tracing::warn!(
            "relationships schema has unsupported arity {}; canonical repair only supports legacy 5-column schema",
            current
        );
        return Ok(());
    }
    // Drop indices before :replace (CozoDB 0.7.6 requirement).
    for idx in &["rel_type_index", "target_qualified_index"] {
        let _ = run_script(
            db,
            &format!("::index drop relationships:{}", idx),
            Default::default(),
        );
    }
    run_script(db, REPAIR_LEGACY_RELATIONSHIPS_5_TO_6, Default::default())?;
    // Recreate indices after :replace.
    for idx_query in &[
        r#"::index create relationships:rel_type_index { rel_type }"#,
        r#"::index create relationships:target_qualified_index { target_qualified }"#,
    ] {
        let _ = run_script(db, idx_query, Default::default());
    }
    tracing::info!("relationships :replace successful");
    Ok(())
}

fn ensure_incidents_table(db: &CozoDb) -> Result<(), Box<dyn std::error::Error>> {
    let existing = run_script(db, "::relations", Default::default())
        .map(|r| {
            r.rows
                .iter()
                .filter_map(|row| row.first().and_then(|v| v.get_str().map(String::from)))
                .collect::<std::collections::HashSet<_>>()
        })
        .unwrap_or_default();
    if existing.contains("incidents") {
        return Ok(());
    }
    let create_incidents = r#":create incidents {id: String, env: String, title: String, severity: String, occurred_at: Int, resolved_at: Int?, root_cause: String, resolution: String, affected_services: String, trigger_pattern: String?, prevention: String?, tags: String, author: String, linked_ticket: String?}"#;
    run_script(db, create_incidents, Default::default())?;
    for idx in &[
        r#"::index create incidents:env_index { env }"#,
        r#"::index create incidents:severity_index { severity }"#,
        r#"::index create incidents:author_index { author }"#,
    ] {
        if let Err(e) = run_script(db, idx, Default::default()) {
            tracing::debug!("Incident index note: {:?}", e);
        }
    }
    Ok(())
}

fn record_migration(
    db: &CozoDb,
    id: &str,
    applied_at: i64,
) -> Result<(), Box<dyn std::error::Error>> {
    let query = r#"?[id, applied_at] <- [[$mid, $ts]] :put migrations {id, applied_at}"#;
    let mut params = std::collections::BTreeMap::new();
    params.insert("mid".to_string(), serde_json::Value::String(id.to_string()));
    params.insert(
        "ts".to_string(),
        serde_json::Value::Number(applied_at.into()),
    );
    run_script(db, query, params)?;
    Ok(())
}

fn validate_code_elements_schema(db: &CozoDb) -> Result<(), Box<dyn std::error::Error>> {
    let schema_query = r#":schema code_elements"#;
    match run_script(db, schema_query, Default::default()) {
        Ok(result) => {
            let column_count = result.rows.len();
            const EXPECTED_COLUMNS: usize = 13;
            if column_count != EXPECTED_COLUMNS {
                eprintln!(
                    "WARNING: code_elements schema has {} columns, expected {}. \
                     Schema may be from an older version. Consider re-indexing.",
                    column_count, EXPECTED_COLUMNS
                );
            }
        }
        Err(e) => {
            tracing::debug!("Could not validate code_elements schema: {:?}", e);
        }
    }
    Ok(())
}

fn validate_relationships_schema(db: &CozoDb) -> Result<(), Box<dyn std::error::Error>> {
    let schema_query = r#":schema relationships"#;
    match run_script(db, schema_query, Default::default()) {
        Ok(result) => {
            let column_count = result.rows.len();
            const EXPECTED_COLUMNS: usize = 6;
            if column_count != EXPECTED_COLUMNS {
                eprintln!(
                    "WARNING: relationships schema has {} columns, expected {}. \
                     Schema may be from an older version. Consider re-indexing.",
                    column_count, EXPECTED_COLUMNS
                );
            }
        }
        Err(e) => {
            tracing::debug!("Could not validate relationships schema: {:?}", e);
        }
    }
    Ok(())
}

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

    fn make_db(path: &std::path::Path) -> CozoDb {
        let s = path.to_string_lossy().to_string();
        cozo::DbInstance::new("sqlite", &s, "").unwrap()
    }

    #[test]
    fn code_elements_schema_on_canonical_db() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("ce.db");
        let db = make_db(&db_path);
        run_script(&db,
            r#":create code_elements {qualified_name: String, element_type: String, name: String, file_path: String, line_start: Int, line_end: Int, language: String, parent_qualified: String?, cluster_id: String?, cluster_label: String?, metadata: String, env: String default 'local', ontology_layer: String default 'procedural'}"#,
            Default::default(),
        )
        .unwrap();

        let schema = code_elements_schema(&db);
        assert_eq!(schema.name, "code_elements");
        assert_eq!(schema.arity, 13);
        assert!(schema.canonical, "fresh 13-col DB must report canonical");
        assert_eq!(
            schema.columns,
            CODE_ELEMENTS_13_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect::<Vec<_>>()
        );
        assert_eq!(
            schema.columns.last().map(String::as_str),
            Some("ontology_layer")
        );
    }

    #[test]
    fn relationships_schema_on_canonical_db() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("rel.db");
        let db = make_db(&db_path);
        run_script(&db,
            r#":create relationships {source_qualified: String, target_qualified: String, rel_type: String, confidence: Float, metadata: String, env: String default 'local'}"#,
            Default::default(),
        )
        .unwrap();

        let schema = relationships_schema(&db);
        assert_eq!(schema.name, "relationships");
        assert_eq!(schema.arity, 6);
        assert!(schema.canonical);
        assert_eq!(
            schema.columns,
            RELATIONSHIPS_6_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect::<Vec<_>>()
        );
    }

    #[test]
    fn code_elements_schema_reports_legacy_11_columns_as_non_canonical() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("legacy.db");
        let db = make_db(&db_path);
        run_script(&db,
            r#":create code_elements {qualified_name: String, element_type: String, name: String, file_path: String, line_start: Int, line_end: Int, language: String, parent_qualified: String?, cluster_id: String?, cluster_label: String?, metadata: String}"#,
            Default::default(),
        )
        .unwrap();

        let schema = code_elements_schema(&db);
        assert_eq!(schema.arity, 11);
        assert!(!schema.canonical, "11-col schema must not be canonical");
        assert_eq!(
            schema.columns,
            CODE_ELEMENTS_11_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect::<Vec<_>>()
        );
        assert!(!schema.columns.contains(&"env".to_string()));
        assert!(!schema.columns.contains(&"ontology_layer".to_string()));
    }

    #[test]
    fn relationships_schema_reports_legacy_5_columns_as_non_canonical() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("legacy-rel.db");
        let db = make_db(&db_path);
        run_script(&db,
            r#":create relationships {source_qualified: String, target_qualified: String, rel_type: String, confidence: Float, metadata: String}"#,
            Default::default(),
        )
        .unwrap();

        let schema = relationships_schema(&db);
        assert_eq!(schema.arity, 5);
        assert!(!schema.canonical);
        assert_eq!(
            schema.columns,
            RELATIONSHIPS_5_COLUMNS
                .iter()
                .map(|s| s.to_string())
                .collect::<Vec<_>>()
        );
        assert!(!schema.columns.contains(&"env".to_string()));
    }

    #[test]
    fn get_relation_schema_unknown_relation_returns_zero_columns() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("unknown.db");
        let db = make_db(&db_path);
        let schema = get_relation_schema(&db, "no_such_relation");
        assert_eq!(schema.name, "no_such_relation");
        assert_eq!(schema.arity, 0);
        assert!(schema.columns.is_empty());
        assert!(!schema.canonical);
    }

    // CozoDB 0.7.x migration wrapper tests — verify the `run_script` adapter,
    // `mutability_for` auto-detection, `json_to_datavalue` conversion, and
    // storage config resolution introduced when upgrading from vendored
    // cozo-0.2.2 to cozo-0.7.6.

    static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[test]
    fn mutability_for_read_query_is_immutable() {
        assert_eq!(
            mutability_for("?[a, b] := *code_elements[a, b]"),
            ScriptMutability::Immutable
        );
        assert_eq!(mutability_for("::relations"), ScriptMutability::Immutable);
    }

    #[test]
    fn mutability_for_put_query_is_mutable() {
        assert_eq!(
            mutability_for("?[a, b] <- [[$a, $b]] :put code_elements {a, b}"),
            ScriptMutability::Mutable
        );
    }

    #[test]
    fn mutability_for_rm_query_is_mutable() {
        assert_eq!(
            mutability_for("?[a] <- [[$a]] :rm code_elements {a}"),
            ScriptMutability::Mutable
        );
    }

    #[test]
    fn mutability_for_create_query_is_mutable() {
        assert_eq!(
            mutability_for(":create code_elements {a: String, b: String}"),
            ScriptMutability::Mutable
        );
    }

    #[test]
    fn mutability_for_hnsw_query_is_mutable() {
        assert_eq!(
            mutability_for("::hnsw create embedding_vectors:vec_idx { dim: 384 }"),
            ScriptMutability::Mutable
        );
    }

    #[test]
    fn mutability_for_index_create_is_mutable() {
        assert_eq!(
            mutability_for("::index create code_elements:name_idx { name }"),
            ScriptMutability::Mutable
        );
    }

    #[test]
    fn mutability_for_combined_read_head_with_put_is_mutable() {
        // Read head (?) + write action (:put) → Mutable (scans whole query).
        assert_eq!(
            mutability_for("?[a, b] := *rel[a, b], b = $val :put rel2 {a, b}"),
            ScriptMutability::Mutable
        );
    }

    #[test]
    fn json_to_datavalue_null_becomes_datavalue_null() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("null.db");
        let db = make_db(&db_path);
        run_script(&db, ":create t {k: String, v: String?}", Default::default()).unwrap();
        let mut params = std::collections::BTreeMap::new();
        params.insert("k".to_string(), serde_json::json!("key1"));
        params.insert("v".to_string(), serde_json::Value::Null);
        run_script(&db, "?[k, v] <- [[$k, $v]] :put t {k, v}", params).unwrap();
        let result = run_script(&db, "?[k, v] := *t[k, v]", Default::default()).unwrap();
        assert_eq!(result.rows.len(), 1);
    }

    #[test]
    fn json_to_datavalue_string_preserves_value() {
        let dv = json_to_datavalue(serde_json::json!("hello"));
        assert_eq!(dv.get_str(), Some("hello"));
    }

    #[test]
    fn json_to_datavalue_int_preserves_value() {
        let dv = json_to_datavalue(serde_json::json!(42));
        assert_eq!(dv.get_int(), Some(42));
    }

    #[test]
    fn run_script_put_then_get_roundtrip() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("rt.db");
        let db = make_db(&db_path);
        run_script(
            &db,
            ":create kv {k: String => v: String}",
            Default::default(),
        )
        .unwrap();
        let mut params = std::collections::BTreeMap::new();
        params.insert("k".to_string(), serde_json::json!("alpha"));
        params.insert("v".to_string(), serde_json::json!("beta"));
        run_script(&db, "?[k, v] <- [[$k, $v]] :put kv {k => v}", params).unwrap();
        let mut qparams = std::collections::BTreeMap::new();
        qparams.insert("k".to_string(), serde_json::json!("alpha"));
        let result = run_script(&db, "?[v] := *kv[k, v], k = $k", qparams).unwrap();
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.rows[0][0].get_str(), Some("beta"));
    }

    #[test]
    fn run_script_immutable_read_after_write() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("imm.db");
        let db = make_db(&db_path);
        run_script(
            &db,
            ":create simple {a: String, b: Int}",
            Default::default(),
        )
        .unwrap();
        let mut params = std::collections::BTreeMap::new();
        params.insert("a".to_string(), serde_json::json!("x"));
        params.insert("b".to_string(), serde_json::json!(10));
        run_script(&db, "?[a, b] <- [[$a, $b]] :put simple {a, b}", params).unwrap();
        let result = run_script(&db, "?[a, b] := *simple[a, b]", Default::default()).unwrap();
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.rows[0][0].get_str(), Some("x"));
        assert_eq!(result.rows[0][1].get_int(), Some(10));
    }

    #[test]
    fn resolve_storage_config_defaults_to_sqlite() {
        let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var("LEANKG_DB_ENGINE").ok();
        std::env::remove_var("LEANKG_DB_ENGINE");
        let cfg = resolve_storage_config(std::path::Path::new("/tmp/test.db"));
        assert_eq!(cfg.engine, StorageEngine::Sqlite);
        if let Some(v) = prev {
            std::env::set_var("LEANKG_DB_ENGINE", v);
        }
    }

    #[test]
    fn resolve_storage_config_rocksdb_when_env_set() {
        let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var("LEANKG_DB_ENGINE").ok();
        std::env::set_var("LEANKG_DB_ENGINE", "rocksdb");
        let cfg = resolve_storage_config(std::path::Path::new("/tmp/test.db"));
        assert_eq!(cfg.engine, StorageEngine::RocksDb);
        match prev {
            Some(v) => std::env::set_var("LEANKG_DB_ENGINE", v),
            None => std::env::remove_var("LEANKG_DB_ENGINE"),
        }
    }

    #[test]
    fn resolve_storage_config_sqlite_dir_appends_leankg_db() {
        let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var("LEANKG_DB_ENGINE").ok();
        std::env::remove_var("LEANKG_DB_ENGINE");
        // resolve_storage_config checks is_dir() — create a real temp dir
        // so the SQLite path appends "leankg.db".
        let tmp = TempDir::new().unwrap();
        let cfg = resolve_storage_config(tmp.path());
        assert_eq!(cfg.engine, StorageEngine::Sqlite);
        assert!(cfg.path.ends_with("leankg.db"));
        if let Some(v) = prev {
            std::env::set_var("LEANKG_DB_ENGINE", v);
        }
    }

    #[test]
    fn get_env_mmap_size_default_is_64_mib() {
        let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var("LEANKG_MMAP_SIZE").ok();
        std::env::remove_var("LEANKG_MMAP_SIZE");
        assert_eq!(get_env_mmap_size(), 64 * 1024 * 1024);
        if let Some(v) = prev {
            std::env::set_var("LEANKG_MMAP_SIZE", v);
        }
    }

    #[test]
    fn get_env_mmap_size_env_override() {
        let _guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var("LEANKG_MMAP_SIZE").ok();
        std::env::set_var("LEANKG_MMAP_SIZE", "134217728");
        assert_eq!(get_env_mmap_size(), 134217728);
        match prev {
            Some(v) => std::env::set_var("LEANKG_MMAP_SIZE", v),
            None => std::env::remove_var("LEANKG_MMAP_SIZE"),
        }
    }

    #[test]
    fn central_project_storage_path_includes_project_name() {
        let path =
            central_project_storage_path(std::path::Path::new("/home/user/myproject/.leankg"));
        let path_str = path.to_string_lossy();
        assert!(path_str.contains("projects"), "path: {path_str}");
        assert!(path_str.contains("myproject"), "path: {path_str}");
    }

    /// FR-HNSW-C: the embed / semantic-context / smoke-test CLI commands pass
    /// `<project>/.leankg/leankg.db` to `init_db` (the file path), while
    /// `leankg index` and `leankg mcp-http` pass the `.leankg` directory.
    /// Both must resolve to the SAME RocksDB project root, otherwise
    /// `leankg embed` opens a fresh empty database and silently embeds
    /// nothing — leaving `semantic_search` stuck on the ontology-only path.
    #[test]
    fn central_project_storage_path_resolves_leankg_db_to_same_root_as_dot_leankg() {
        let dir_path =
            central_project_storage_path(std::path::Path::new("/home/user/myproject/.leankg"));
        let file_path = central_project_storage_path(std::path::Path::new(
            "/home/user/myproject/.leankg/leankg.db",
        ));
        assert_eq!(
            dir_path, file_path,
            "leankg.db file path must resolve to the same project root as .leankg directory"
        );
        let path_str = dir_path.to_string_lossy();
        assert!(path_str.contains("myproject"), "path: {path_str}");
    }

    /// Edge case: the database file is detached from `.leankg/` (e.g., a
    /// manual copy). Falls back to the parent directory as the project root
    /// rather than producing a hash of the file's actual location.
    #[test]
    fn central_project_storage_path_handles_detached_db_file() {
        let file_path = central_project_storage_path(std::path::Path::new("/tmp/loose-leankg.db"));
        let path_str = file_path.to_string_lossy();
        // Falls back to "loose-leankg" as the project name; this is the
        // graceful degradation path — caller will get a fresh DB at a
        // distinct path, and the embed step will report 0 vectors.
        assert!(
            path_str.contains("loose-leankg"),
            "expected detached db to use parent dir as project: {path_str}"
        );
    }

    #[test]
    fn init_db_creates_canonical_schema_on_sqlite() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("init.db");
        let db = init_db(&db_path).expect("init_db");
        let schema = code_elements_schema(&db);
        assert_eq!(schema.name, "code_elements");
        assert_eq!(schema.arity, 13);
        assert!(schema.canonical);
        assert!(schema.columns.contains(&"env".to_string()));
        assert!(schema.columns.contains(&"ontology_layer".to_string()));
    }

    #[test]
    fn init_db_relationships_has_six_columns() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("rel_init.db");
        let db = init_db(&db_path).expect("init_db");
        let schema = relationships_schema(&db);
        assert_eq!(schema.name, "relationships");
        assert_eq!(schema.arity, 6);
        assert!(schema.canonical);
        assert!(schema.columns.contains(&"env".to_string()));
    }
}