use std::collections::{BTreeSet, HashMap};
use std::fs;
use std::path::{Path, PathBuf};
use chrono::{DateTime, Utc};
use khive_runtime::{KhiveRuntime, NamespaceToken, RuntimeError};
use khive_storage::{Edge, Entity, LinkId};
use khive_types::EdgeRelation;
use serde_json::{json, Value};
use uuid::Uuid;
use crate::imports::{self, Resolved};
use crate::ingest::CODE_INGEST_NAMESPACE;
use crate::manifest;
#[derive(Debug, Default, serde::Serialize)]
pub struct CodeSourceIngestReport {
pub projects_created: u64,
pub projects_updated: u64,
pub modules_created: u64,
pub modules_updated: u64,
pub edges_created: u64,
pub edges_updated: u64,
pub unresolved_recorded: u64,
pub unresolved_resolved: u64,
pub languages: Vec<String>,
pub warnings: Vec<String>,
pub db_path: String,
}
#[derive(Debug, thiserror::Error)]
pub enum CodeSourceIngestError {
#[error("path {0:?} does not exist or is not a directory")]
InvalidPath(PathBuf),
#[error("runtime error: {0}")]
Runtime(#[from] RuntimeError),
#[error("storage error: {0}")]
Storage(String),
}
pub struct CodeSourceIngestOptions<'a> {
pub path: &'a Path,
pub languages: BTreeSet<&'static str>,
pub sweep_time: DateTime<Utc>,
}
fn uuid5_json(value: &Value) -> Uuid {
let bytes = serde_json::to_vec(value).expect("Value always serializes");
Uuid::new_v5(&CODE_INGEST_NAMESPACE, &bytes)
}
fn project_uuid(source_project: &str) -> Uuid {
uuid5_json(&json!({
"kind": "code-source-project",
"source_project": source_project,
}))
}
fn module_uuid(source_project: &str, language: &str, module_path: &str) -> Uuid {
uuid5_json(&json!({
"kind": "code-source-symbol",
"source_project": source_project,
"language": language,
"module_path": module_path,
"name": module_path,
"symbol_kind": "module",
}))
}
fn edge_uuid(relation: EdgeRelation, source_id: Uuid, target_id: Uuid) -> Uuid {
uuid5_json(&json!({
"kind": "code-source-edge",
"relation": relation.as_str(),
"source_id": source_id.to_string(),
"target_id": target_id.to_string(),
}))
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
struct UnresolvedSpec {
specifier: String,
target_kind: String,
dependency_kind: String,
language: String,
}
fn read_unresolved(properties: &Value) -> Vec<UnresolvedSpec> {
properties
.get("unresolved_specifiers")
.and_then(|v| serde_json::from_value(v.clone()).ok())
.unwrap_or_default()
}
async fn get_entity_opt(
rt: &KhiveRuntime,
token: &NamespaceToken,
id: Uuid,
) -> Result<Option<Entity>, CodeSourceIngestError> {
rt.entities(token)?
.get_entity(id)
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))
}
async fn upsert_entity(
rt: &KhiveRuntime,
token: &NamespaceToken,
entity: Entity,
) -> Result<(), CodeSourceIngestError> {
rt.entities(token)?
.upsert_entity(entity)
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))
}
#[allow(clippy::too_many_arguments)]
async fn upsert_edge(
rt: &KhiveRuntime,
token: &NamespaceToken,
id: Uuid,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
metadata: Value,
now: DateTime<Utc>,
) -> Result<bool, CodeSourceIngestError> {
let link_id = LinkId::from(id);
let graph = rt.graph(token)?;
let existed = graph
.get_edge(link_id)
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))?
.is_some();
let edge = Edge {
id: link_id,
namespace: token.namespace().as_str().to_string(),
source_id,
target_id,
relation,
weight: 1.0,
created_at: now,
updated_at: now,
deleted_at: None,
metadata: Some(metadata),
target_backend: None,
};
graph
.upsert_edge(edge)
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))?;
Ok(!existed)
}
fn ts(dt: DateTime<Utc>) -> i64 {
dt.timestamp_micros()
}
async fn upsert_project(
rt: &KhiveRuntime,
token: &NamespaceToken,
name: &str,
language: &str,
sweep_time: DateTime<Utc>,
report: &mut CodeSourceIngestReport,
) -> Result<Uuid, CodeSourceIngestError> {
let id = project_uuid(name);
let existing = get_entity_opt(rt, token, id).await?;
let is_new = existing.is_none();
let mut sweep_clock = existing
.as_ref()
.and_then(|e| e.properties.as_ref())
.and_then(|p| p.get("sweep_clock"))
.and_then(|v| v.as_object().cloned())
.unwrap_or_default();
sweep_clock.insert(language.to_string(), json!(sweep_time.to_rfc3339()));
let unresolved = existing
.as_ref()
.and_then(|e| e.properties.as_ref())
.map(read_unresolved)
.unwrap_or_default();
let mut props = serde_json::Map::new();
props.insert("source_project".into(), json!(name));
props.insert("last_seen_at".into(), json!(sweep_time.to_rfc3339()));
props.insert("sweep_clock".into(), Value::Object(sweep_clock));
if !unresolved.is_empty() {
props.insert(
"unresolved_specifiers".into(),
serde_json::to_value(&unresolved).expect("serializes"),
);
}
let mut entity = Entity::new(token.namespace().as_str(), "project", name);
entity.id = id;
entity.properties = Some(Value::Object(props));
let now = ts(sweep_time);
entity.created_at = existing.as_ref().map(|e| e.created_at).unwrap_or(now);
entity.updated_at = now;
upsert_entity(rt, token, entity).await?;
if is_new {
report.projects_created += 1;
} else {
report.projects_updated += 1;
}
Ok(id)
}
#[allow(clippy::too_many_arguments)]
async fn upsert_module(
rt: &KhiveRuntime,
token: &NamespaceToken,
source_project: &str,
language: &str,
module_path: &str,
content_hash: &str,
sweep_time: DateTime<Utc>,
report: &mut CodeSourceIngestReport,
) -> Result<Uuid, CodeSourceIngestError> {
let id = module_uuid(source_project, language, module_path);
let existing = get_entity_opt(rt, token, id).await?;
let is_new = existing.is_none();
let unresolved = existing
.as_ref()
.and_then(|e| e.properties.as_ref())
.map(read_unresolved)
.unwrap_or_default();
let mut props = serde_json::Map::new();
props.insert("source_project".into(), json!(source_project));
props.insert("language".into(), json!(language));
props.insert("module_path".into(), json!(module_path));
props.insert("content_hash".into(), json!(content_hash));
props.insert("last_seen_at".into(), json!(sweep_time.to_rfc3339()));
if !unresolved.is_empty() {
props.insert(
"unresolved_specifiers".into(),
serde_json::to_value(&unresolved).expect("serializes"),
);
}
let mut entity = Entity::new(token.namespace().as_str(), "concept", module_path)
.with_entity_type(Some("module"));
entity.id = id;
entity.properties = Some(Value::Object(props));
let now = ts(sweep_time);
entity.created_at = existing.as_ref().map(|e| e.created_at).unwrap_or(now);
entity.updated_at = now;
upsert_entity(rt, token, entity).await?;
if is_new {
report.modules_created += 1;
} else {
report.modules_updated += 1;
}
Ok(id)
}
async fn record_unresolved(
rt: &KhiveRuntime,
token: &NamespaceToken,
entity_id: Uuid,
spec: UnresolvedSpec,
report: &mut CodeSourceIngestReport,
) -> Result<(), CodeSourceIngestError> {
let Some(mut entity) = get_entity_opt(rt, token, entity_id).await? else {
return Ok(());
};
let mut list = entity
.properties
.as_ref()
.map(read_unresolved)
.unwrap_or_default();
if list.contains(&spec) {
return Ok(());
}
list.push(spec);
report.unresolved_recorded += 1;
let mut props = entity
.properties
.clone()
.and_then(|v| v.as_object().cloned())
.unwrap_or_default();
props.insert(
"unresolved_specifiers".into(),
serde_json::to_value(&list).expect("serializes"),
);
entity.properties = Some(Value::Object(props));
upsert_entity(rt, token, entity).await
}
fn module_path_separator(language: &str) -> &'static str {
match language {
"python" => ".",
"typescript" => "/",
_ => "::",
}
}
fn module_candidate_specifiers(language: &str, specifier: &str) -> Vec<String> {
let sep = module_path_separator(language);
let segments: Vec<&str> = specifier.split(sep).filter(|s| !s.is_empty()).collect();
if segments.is_empty() {
return vec![specifier.to_string()];
}
(1..=segments.len())
.rev()
.map(|n| segments[..n].join(sep))
.collect()
}
fn target_ids_for(source_project: &str, spec: &UnresolvedSpec) -> Vec<Uuid> {
match spec.target_kind.as_str() {
"module" => module_candidate_specifiers(&spec.language, &spec.specifier)
.into_iter()
.map(|path| module_uuid(source_project, &spec.language, &path))
.collect(),
_ => vec![project_uuid(&spec.specifier)],
}
}
fn merge_dependency_kinds(
existing_metadata: Option<&Value>,
new_kind: &str,
language: &str,
) -> Value {
let mut kinds: BTreeSet<String> = existing_metadata
.and_then(|m| m.get("dependency_kinds"))
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
kinds.insert(new_kind.to_string());
json!({
"dependency_kinds": kinds.into_iter().collect::<Vec<_>>(),
"language": language,
})
}
#[allow(clippy::too_many_arguments)]
async fn upsert_dependency_edge(
rt: &KhiveRuntime,
token: &NamespaceToken,
source_id: Uuid,
target_id: Uuid,
dependency_kind: &str,
language: &str,
now: DateTime<Utc>,
report: &mut CodeSourceIngestReport,
) -> Result<(), CodeSourceIngestError> {
let edge_id = edge_uuid(EdgeRelation::DependsOn, source_id, target_id);
let link_id = LinkId::from(edge_id);
let graph = rt.graph(token)?;
let existing = graph
.get_edge(link_id)
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))?;
let existed = existing.is_some();
let metadata = merge_dependency_kinds(
existing.as_ref().and_then(|e| e.metadata.as_ref()),
dependency_kind,
language,
);
let edge = Edge {
id: link_id,
namespace: token.namespace().as_str().to_string(),
source_id,
target_id,
relation: EdgeRelation::DependsOn,
weight: 1.0,
created_at: existing.as_ref().map(|e| e.created_at).unwrap_or(now),
updated_at: now,
deleted_at: None,
metadata: Some(metadata),
target_backend: None,
};
graph
.upsert_edge(edge)
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))?;
if existed {
report.edges_updated += 1;
} else {
report.edges_created += 1;
}
Ok(())
}
async fn reresolve_pass(
rt: &KhiveRuntime,
token: &NamespaceToken,
now: DateTime<Utc>,
report: &mut CodeSourceIngestReport,
) -> Result<(), CodeSourceIngestError> {
use khive_storage::types::{SqlStatement, SqlValue};
let sql = rt.sql();
let mut reader = sql
.reader()
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))?;
let rows = reader
.query_all(SqlStatement {
sql: "SELECT id, kind, properties FROM entities WHERE namespace=?1 \
AND deleted_at IS NULL \
AND json_extract(properties,'$.unresolved_specifiers') IS NOT NULL"
.into(),
params: vec![SqlValue::Text(token.namespace().as_str().to_string())],
label: Some("code_ingest_reresolve_scan".into()),
})
.await
.map_err(|e| CodeSourceIngestError::Storage(e.to_string()))?;
for row in rows {
let id = match row.get("id") {
Some(SqlValue::Uuid(u)) => *u,
Some(SqlValue::Text(s)) => match Uuid::parse_str(s) {
Ok(u) => u,
Err(_) => continue,
},
_ => continue,
};
let Some(mut entity) = get_entity_opt(rt, token, id).await? else {
continue;
};
let source_project = entity
.properties
.as_ref()
.and_then(|p| p.get("source_project"))
.and_then(Value::as_str)
.unwrap_or(entity.name.as_str())
.to_string();
let mut list = entity
.properties
.as_ref()
.map(read_unresolved)
.unwrap_or_default();
if list.is_empty() {
continue;
}
let mut still_unresolved = Vec::new();
let mut changed = false;
for spec in list.drain(..) {
let mut resolved_target = None;
for target_id in target_ids_for(&source_project, &spec) {
if get_entity_opt(rt, token, target_id).await?.is_some() {
resolved_target = Some(target_id);
break;
}
}
match resolved_target {
Some(target_id) => {
upsert_dependency_edge(
rt,
token,
entity.id,
target_id,
&spec.dependency_kind,
&spec.language,
now,
report,
)
.await?;
report.unresolved_resolved += 1;
changed = true;
}
None => still_unresolved.push(spec),
}
}
if changed {
let mut props = entity
.properties
.clone()
.and_then(|v| v.as_object().cloned())
.unwrap_or_default();
if still_unresolved.is_empty() {
props.remove("unresolved_specifiers");
} else {
props.insert(
"unresolved_specifiers".into(),
serde_json::to_value(&still_unresolved).expect("serializes"),
);
}
entity.properties = Some(Value::Object(props));
upsert_entity(rt, token, entity).await?;
}
}
Ok(())
}
fn collect_source_files(root: &Path, ext: &str, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
const SKIP_DIRS: &[&str] = &[
".git",
"target",
"node_modules",
"__pycache__",
".venv",
"venv",
"dist",
"build",
];
for entry in fs::read_dir(root)? {
let entry = entry?;
let path = entry.path();
let file_type = entry.file_type()?;
if file_type.is_dir() {
let name = entry.file_name();
let name = name.to_string_lossy();
if SKIP_DIRS.contains(&name.as_ref()) || name.starts_with('.') {
continue;
}
collect_source_files(&path, ext, out)?;
} else if path.extension().and_then(|e| e.to_str()) == Some(ext) {
out.push(path);
}
}
Ok(())
}
fn content_hash(content: &str) -> String {
let mut hash: u64 = 0xcbf29ce484222325;
for b in content.as_bytes() {
hash ^= *b as u64;
hash = hash.wrapping_mul(0x100000001b3);
}
format!("{hash:016x}")
}
pub async fn run_code_ingest(
rt: &KhiveRuntime,
token: &NamespaceToken,
opts: CodeSourceIngestOptions<'_>,
) -> Result<CodeSourceIngestReport, CodeSourceIngestError> {
if !opts.path.is_dir() {
return Err(CodeSourceIngestError::InvalidPath(opts.path.to_path_buf()));
}
let mut report = CodeSourceIngestReport {
languages: opts.languages.iter().map(|s| s.to_string()).collect(),
..Default::default()
};
let manifests = manifest::discover_manifests(opts.path, &opts.languages)
.map_err(|e| CodeSourceIngestError::InvalidPath(opts.path.join(e.to_string())))?;
let mut project_ids: HashMap<String, Uuid> = HashMap::new();
for m in &manifests {
let id =
upsert_project(rt, token, &m.name, m.language, opts.sweep_time, &mut report).await?;
project_ids.insert(m.name.clone(), id);
}
for m in &manifests {
let source_id = project_ids[&m.name];
for (dep_name, dep_kind) in &m.dependencies {
let spec = UnresolvedSpec {
specifier: dep_name.clone(),
target_kind: "project".to_string(),
dependency_kind: dep_kind.clone(),
language: m.language.to_string(),
};
record_unresolved(rt, token, source_id, spec, &mut report).await?;
}
}
for language in opts.languages.iter().copied() {
run_import_scan(
rt,
token,
language,
opts.path,
opts.sweep_time,
&mut project_ids,
&mut report,
)
.await?;
}
reresolve_pass(rt, token, opts.sweep_time, &mut report).await?;
Ok(report)
}
fn basename_project_name(ingest_root: &Path) -> String {
ingest_root
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| ingest_root.display().to_string())
}
#[allow(clippy::too_many_arguments)]
async fn run_import_scan(
rt: &KhiveRuntime,
token: &NamespaceToken,
language: &'static str,
ingest_root: &Path,
sweep_time: DateTime<Utc>,
project_ids: &mut HashMap<String, Uuid>,
report: &mut CodeSourceIngestReport,
) -> Result<(), CodeSourceIngestError> {
let Some(ext) = imports::extension_for_language(language) else {
return Ok(());
};
let mut files = Vec::new();
if let Err(e) = collect_source_files(ingest_root, ext, &mut files) {
report
.warnings
.push(format!("walking {}: {e}", ingest_root.display()));
return Ok(());
}
for file in files {
let Some(file_dir) = file.parent() else {
continue;
};
let (proj_root, proj_name) =
manifest::find_governing_manifest(file_dir, ingest_root, language).unwrap_or_else(
|| {
(
ingest_root.to_path_buf(),
basename_project_name(ingest_root),
)
},
);
let Some(module_path) = imports::module_path_for_file(&file, &proj_root, language) else {
continue;
};
let proj_id = match project_ids.get(&proj_name) {
Some(id) => *id,
None => {
let id =
upsert_project(rt, token, &proj_name, language, sweep_time, report).await?;
project_ids.insert(proj_name.clone(), id);
id
}
};
let content = match fs::read_to_string(&file) {
Ok(c) => c,
Err(e) => {
report
.warnings
.push(format!("reading {}: {e}", file.display()));
continue;
}
};
let hash = content_hash(&content);
let module_id = upsert_module(
rt,
token,
&proj_name,
language,
&module_path,
&hash,
sweep_time,
report,
)
.await?;
let contains_edge_id = edge_uuid(EdgeRelation::Contains, proj_id, module_id);
let contains_created = upsert_edge(
rt,
token,
contains_edge_id,
proj_id,
module_id,
EdgeRelation::Contains,
json!({}),
sweep_time,
)
.await?;
if contains_created {
report.edges_created += 1;
} else {
report.edges_updated += 1;
}
for raw in imports::extract_raw_imports(language, &content) {
let resolved = if language == "typescript" && raw.starts_with('.') {
let rel_dir = file_dir.strip_prefix(&proj_root).unwrap_or(Path::new(""));
Resolved::IntraModule(imports::resolve_relative_ts_module(rel_dir, &raw))
} else {
imports::classify_import(language, &raw, &module_path, &proj_name)
};
match resolved {
Resolved::Skip => {}
Resolved::IntraModule(target_module_path) => {
let spec = UnresolvedSpec {
specifier: target_module_path,
target_kind: "module".to_string(),
dependency_kind: "import".to_string(),
language: language.to_string(),
};
record_unresolved(rt, token, module_id, spec, report).await?;
}
Resolved::ExternalProject(target_name) => {
let spec = UnresolvedSpec {
specifier: target_name,
target_kind: "project".to_string(),
dependency_kind: "import".to_string(),
language: language.to_string(),
};
record_unresolved(rt, token, proj_id, spec, report).await?;
}
}
}
}
Ok(())
}