use cozo::{DataValue, NamedRows, ScriptMutability};
use sha2::{Digest, Sha256};
use std::path::Path;
pub type CozoDb = cozo::DbInstance;
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))
}
fn json_to_datavalue(v: serde_json::Value) -> DataValue {
DataValue::from(v)
}
fn mutability_for(query: &str) -> ScriptMutability {
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())
.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
);
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));
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]))
}
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") {
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);
}
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(db, &existing_relations)?;
repair_canonical_schema(db, &existing_relations)?;
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);
}
}
}
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);
}
}
}
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);
}
}
}
#[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>> {
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);
}
}
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;
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);
}
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)?;
}
if !applied.contains("002_feature_workflow_links") {
tracing::info!("Running migration 002_feature_workflow_links...");
let create_fw_links =
r#":create feature_workflow_links {feature_id: String, workflow_id: String}"#;
if let Err(e) = run_script(db, create_fw_links, Default::default()) {
tracing::warn!("Migration 002 failed (may already exist): {:?}", e);
}
let fw_index = r#"::index create feature_workflow_links:feature_id_index { feature_id }"#;
if let Err(e) = run_script(db, fw_index, Default::default()) {
tracing::debug!("feature_workflow_links index creation note: {:?}", e);
}
record_migration(db, "002_feature_workflow_links", now)?;
}
mark_legacy_migrations_as_applied(db, &applied, now)?;
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)
}
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",
];
#[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,
}
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,
}
}
pub fn code_elements_schema(db: &CozoDb) -> RelationSchema {
get_relation_schema(db, "code_elements")
}
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
);
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");
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(());
}
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())?;
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);
}
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() {
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");
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}");
}
#[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}");
}
#[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();
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()));
}
}