use std::{
fs,
path::{Path, PathBuf},
process::Command,
sync::Arc,
time::{SystemTime, UNIX_EPOCH},
};
use relay_knowledge::{
api::{CodeRepositoryRegisterRequest, InterfaceKind, RequestContext},
application::{RelayKnowledgeService, RuntimeConfiguration},
domain::{
CodeIndexMode, CodeIndexRequest, CodeQueryKind, CodeRepositorySelector,
CodeRetrievalRequest, FreshnessPolicy, SoftwareGlobalKind, SoftwareGlobalRequest,
},
env::{EnvironmentConfig, PlatformKind},
paths::RuntimePaths,
storage::{KnowledgeStore, PartitionedSqliteKnowledgeStore, SqliteGraphStore},
};
use rusqlite::{Connection, params};
#[tokio::test]
async fn software_projection_resolves_symbolic_refs_to_indexed_commit_scope() {
let repo = FixtureRepo::create("code-software-ref");
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.1.0"
edition = "2021"
[dependencies]
serde = "1"
"#,
);
repo.write("src/lib.rs", "pub fn uses_dependency_manifest() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "software manifest"]);
repo.git(["branch", "software-main"]);
let service = service_with_memory_store().await;
register_fixture_repo(&service, &repo, Vec::new(), "register-software-ref").await;
let indexed = service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "software-main"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-ref"),
)
.await
.expect("symbolic ref should index");
let projection =
software_projection(&service, "software-main", FreshnessPolicy::WaitUntilFresh)
.await
.expect("symbolic software ref should resolve to indexed commit");
assert_eq!(projection.scope.requested_ref, "software-main");
assert_eq!(projection.scope.scope_id, indexed.scope.scope_id);
assert!(
projection
.components
.iter()
.any(|component| component.name == "serde")
);
}
#[tokio::test]
async fn stale_software_projection_uses_a_durable_repair_task() {
let repo = FixtureRepo::create("code-software-fast-refresh");
let db_path = repo.path.join("relay-knowledge.sqlite");
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.1.0"
edition = "2021"
[dependencies]
serde = "1"
"#,
);
repo.write("src/lib.rs", "pub fn uses_fast_projection_refresh() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "software manifest"]);
let service = service_with_store_path(&db_path).await;
register_fixture_repo(
&service,
&repo,
Vec::new(),
"register-software-fast-refresh",
)
.await;
let first = service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-fast-first"),
)
.await
.expect("initial index should refresh software projection");
mark_software_projection_stale(&db_path, &first.summary.source_scope);
let started = service
.start_code_repository_index(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-fast-second"),
)
.await
.expect("stale projection should queue a durable repair");
assert!(started.summary.is_none());
let task = started
.task
.expect("projection repair should return a task");
let completed = service
.run_code_index_task_once(
Some(task.task_id),
context("run-software-projection-repair"),
)
.await
.expect("projection repair worker should run")
.expect("projection repair worker should claim the task");
let projection = software_projection(&service, "HEAD", FreshnessPolicy::WaitUntilFresh)
.await
.expect("durable repair should clear stale software status");
assert_eq!(completed.source_scope, first.summary.source_scope);
assert!(!projection.status.stale);
assert_eq!(projection.status.last_error, None);
assert!(
projection
.components
.iter()
.any(|component| component.name == "serde")
);
}
#[tokio::test]
async fn fresh_full_index_start_does_not_rewrite_software_projection() {
let repo = FixtureRepo::create("code-software-fast-read-only");
let db_path = repo.path.join("relay-knowledge.sqlite");
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.1.0"
edition = "2021"
[dependencies]
serde = "1"
"#,
);
repo.write("src/lib.rs", "pub fn stable_projection() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "stable projection"]);
let service = service_with_store_path(&db_path).await;
register_fixture_repo(&service, &repo, Vec::new(), "register-fast-read-only").await;
let first = service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-fast-read-only-first"),
)
.await
.expect("initial index should succeed");
let observer = Connection::open(&db_path).expect("observer connection should open");
let before = sqlite_data_version(&observer);
let started = service
.start_code_repository_index(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("start-fast-read-only-second"),
)
.await
.expect("fresh index start should reuse the completed scope");
assert!(started.task.is_none());
assert_eq!(
started
.summary
.as_ref()
.map(|summary| summary.source_scope.as_str()),
Some(first.summary.source_scope.as_str())
);
assert_eq!(sqlite_data_version(&observer), before);
}
#[tokio::test]
async fn partitioned_stale_software_projection_uses_a_fenced_repair_task() {
let repo = FixtureRepo::create("partitioned-software-repair");
let runtime_root = repo.path.join("runtime");
repo.write("src/lib.rs", "pub fn partitioned_projection() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "partitioned projection"]);
let (service, paths) = service_with_partitioned_store(&runtime_root).await;
register_fixture_repo(&service, &repo, Vec::new(), "register-partitioned-repair").await;
let first = service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-partitioned-repair-first"),
)
.await
.expect("initial partitioned index should succeed");
mark_software_projection_stale(
&paths.repository_shard_database_file(&first.summary.repository_id),
&first.summary.source_scope,
);
let started = service
.start_code_repository_index(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("start-partitioned-repair"),
)
.await
.expect("stale partitioned projection should queue a repair");
assert!(started.summary.is_none());
let task = started
.task
.expect("partitioned repair should return a task");
let completed = service
.run_code_index_task_once(Some(task.task_id), context("run-partitioned-repair"))
.await
.expect("partitioned repair worker should run")
.expect("partitioned repair worker should claim the task");
let projection = software_projection(&service, "HEAD", FreshnessPolicy::WaitUntilFresh)
.await
.expect("partitioned projection should be fresh after fenced repair");
assert_eq!(completed.source_scope, first.summary.source_scope);
assert!(!projection.status.stale);
assert_eq!(projection.status.source_scope, first.summary.source_scope);
}
#[tokio::test]
async fn software_projection_metadata_reports_selected_old_scope() {
let repo = FixtureRepo::create("code-software-old-scope");
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.1.0"
edition = "2021"
[dependencies]
serde = "1"
"#,
);
repo.write("src/lib.rs", "pub fn old_dependency_manifest() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "old software manifest"]);
let old_commit = repo.git_text(["rev-parse", "HEAD"]);
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.2.0"
edition = "2021"
[dependencies]
tokio = "1"
"#,
);
repo.write("src/lib.rs", "pub fn new_dependency_manifest() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "new software manifest"]);
let service = service_with_memory_store().await;
register_fixture_repo(&service, &repo, Vec::new(), "register-software-old").await;
let old_index = service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", &old_commit),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-old"),
)
.await
.expect("old commit should index");
service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-head"),
)
.await
.expect("head should index");
let projection = software_projection(&service, &old_commit, FreshnessPolicy::WaitUntilFresh)
.await
.expect("old software scope should load");
assert_eq!(projection.scope.scope_id, old_index.scope.scope_id);
assert_eq!(projection.scope.resolved_commit_sha, old_commit);
assert!(
projection
.components
.iter()
.any(|component| component.name == "serde")
);
assert!(
!projection
.components
.iter()
.any(|component| component.name == "tokio")
);
}
#[tokio::test]
async fn software_projection_accepts_canonical_path_filter_spellings() {
let repo = FixtureRepo::create("code-software-filter");
repo.write(
"src/Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.1.0"
edition = "2021"
[dependencies]
serde = "1"
"#,
);
repo.write("src/lib.rs", "pub fn filtered_dependency_manifest() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "filtered software manifest"]);
let service = service_with_memory_store().await;
register_fixture_repo(
&service,
&repo,
vec!["src".to_owned()],
"register-software-filter",
)
.await;
service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-filter"),
)
.await
.expect("filtered scope should index");
let projection = service
.software_global_projection(
SoftwareGlobalRequest::new(
filtered_selector("fixture", "HEAD", vec!["./src/".to_owned()]),
SoftwareGlobalKind::Dependencies,
FreshnessPolicy::WaitUntilFresh,
10,
)
.expect("software request should validate"),
context("software-filter"),
)
.await
.expect("canonical filter spelling should match indexed software scope");
assert_eq!(projection.scope.path_filters, ["src"]);
assert!(
projection
.components
.iter()
.any(|component| component.name == "serde")
);
}
#[tokio::test]
async fn software_projection_allow_stale_serves_completed_scope_during_active_index() {
let repo = FixtureRepo::create("code-software-active");
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.1.0"
edition = "2021"
[dependencies]
serde = "1"
"#,
);
repo.write("src/lib.rs", "pub fn old_dependency_manifest() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "old software manifest"]);
let service = service_with_memory_store().await;
register_fixture_repo(&service, &repo, Vec::new(), "register-software-active").await;
service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-active-old"),
)
.await
.expect("old head should index");
repo.write(
"Cargo.toml",
r#"
[package]
name = "fixture"
version = "0.2.0"
edition = "2021"
[dependencies]
tokio = "1"
"#,
);
repo.write("src/lib.rs", "pub fn new_dependency_manifest() {}\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "new software manifest"]);
service
.start_code_repository_index(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::AllowStale,
reuse_historical: false,
},
context("queue-software-active-new"),
)
.await
.expect("new head should queue");
let fresh_error = software_projection(&service, "HEAD", FreshnessPolicy::WaitUntilFresh)
.await
.expect_err("wait-until-fresh should reject stale code scope");
assert!(
fresh_error.message.contains("is stale") || fresh_error.message.contains("has no index")
);
let projection = software_projection(&service, "HEAD", FreshnessPolicy::AllowStale)
.await
.expect("allow-stale should serve completed projection");
assert!(projection.metadata.stale);
assert!(projection.scope.stale);
assert!(projection.status.stale);
assert!(
projection
.components
.iter()
.any(|component| component.name == "serde")
);
assert!(
!projection
.components
.iter()
.any(|component| component.name == "tokio")
);
}
#[tokio::test]
async fn software_projection_links_document_topics_config_and_code_files() {
let repo = FixtureRepo::create("code-software-doc-config");
repo.write(
"docs/runtime.md",
"# Runtime Configuration\n\n`payments.enabled` controls checkout rollout.\n",
);
repo.write("config/flags.yaml", "payments:\n enabled: true\n");
repo.write(
"src/lib.rs",
"pub fn checkout_enabled(config: &Config) -> bool {\n config.get_bool(\"payments.enabled\")\n}\n",
);
repo.git(["add", "."]);
repo.git(["commit", "-m", "document config software graph"]);
let service = service_with_memory_store().await;
register_fixture_repo(&service, &repo, Vec::new(), "register-software-doc-config").await;
service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "HEAD"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-software-doc-config"),
)
.await
.expect("doc/config scope should index");
let projection = software_projection(&service, "HEAD", FreshnessPolicy::WaitUntilFresh)
.await
.expect("software projection should load");
assert!(
projection
.files
.iter()
.any(|file| { file.path == "docs/runtime.md" && file.file_role == "documentation" })
);
assert!(
projection
.topics
.iter()
.any(|topic| topic.name == "Runtime Configuration")
);
assert!(projection.relationships.iter().any(|relationship| {
relationship.relationship_kind == "documents"
&& relationship.evidence_path == "docs/runtime.md"
}));
assert!(projection.relationships.iter().any(|relationship| {
relationship.relationship_kind == "configures"
&& relationship.target_hint.as_deref() == Some("payments.enabled")
}));
}
#[tokio::test]
async fn moved_branch_requires_new_scope_and_queries_rebased_head() {
let repo = FixtureRepo::create("code-rebase-scope");
repo.write("src/lib.rs", "pub fn old_topic_policy() -> u32 { 1 }\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "old topic"]);
repo.git(["branch", "topic"]);
let old_commit = repo.git_text(["rev-parse", "topic"]);
repo.write("src/lib.rs", "pub fn new_topic_policy() -> u32 { 2 }\n");
repo.git(["add", "."]);
repo.git(["commit", "-m", "new topic"]);
repo.git(["branch", "-f", "topic", "HEAD"]);
let service = service_with_memory_store().await;
register_fixture_repo(&service, &repo, vec!["src".to_owned()], "register-rebase").await;
service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", &old_commit),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-old-topic"),
)
.await
.expect("old topic commit should index");
let stale_topic = service
.query_code_repository(
CodeRetrievalRequest::new(
"old_topic_policy",
selector("fixture", "topic"),
CodeQueryKind::Definition,
10,
FreshnessPolicy::AllowStale,
)
.expect("query request should validate"),
context("query-topic-before-new-index"),
)
.await
.expect_err("moved branch should require indexing the new snapshot");
assert!(stale_topic.message.contains("no index for ref"));
service
.index_code_repository(
CodeIndexRequest {
repository: selector("fixture", "topic"),
mode: CodeIndexMode::Full,
workspace_detection: Default::default(),
freshness_policy: FreshnessPolicy::WaitUntilFresh,
reuse_historical: false,
},
context("index-new-topic"),
)
.await
.expect("new topic should index");
let topic = service
.query_code_repository(
CodeRetrievalRequest::new(
"new_topic_policy",
selector("fixture", "topic"),
CodeQueryKind::Definition,
10,
FreshnessPolicy::AllowStale,
)
.expect("query request should validate"),
context("query-new-topic"),
)
.await
.expect("topic should read new scope");
assert!(
topic
.results
.iter()
.any(|hit| hit.excerpt.contains("new_topic_policy"))
);
assert!(
!topic
.results
.iter()
.any(|hit| hit.excerpt.contains("old_topic_policy"))
);
}
#[allow(clippy::result_large_err)]
async fn software_projection(
service: &RelayKnowledgeService,
ref_selector: &str,
freshness_policy: FreshnessPolicy,
) -> Result<relay_knowledge::api::SoftwareGlobalResponse, relay_knowledge::api::ApiError> {
service
.software_global_projection(
SoftwareGlobalRequest::new(
selector("fixture", ref_selector),
SoftwareGlobalKind::All,
freshness_policy,
10,
)
.expect("software request should validate"),
context("software"),
)
.await
}
async fn register_fixture_repo(
service: &RelayKnowledgeService,
repo: &FixtureRepo,
path_filters: Vec<String>,
name: &str,
) {
service
.register_code_repository(
CodeRepositoryRegisterRequest {
root_path: repo.path.display().to_string(),
alias: "fixture".to_owned(),
path_filters,
language_filters: Vec::new(),
},
context(name),
)
.await
.expect("repository should register");
}
fn selector(alias: &str, ref_selector: &str) -> CodeRepositorySelector {
CodeRepositorySelector::new(alias, ref_selector, Vec::new(), Vec::new())
.expect("selector should validate")
}
fn filtered_selector(
alias: &str,
ref_selector: &str,
path_filters: Vec<String>,
) -> CodeRepositorySelector {
CodeRepositorySelector::new(alias, ref_selector, path_filters, Vec::new())
.expect("selector should validate")
}
fn context(name: &str) -> RequestContext {
RequestContext::with_ids(
InterfaceKind::Cli,
format!("req-{name}"),
format!("trace-{name}"),
)
}
async fn service_with_memory_store() -> RelayKnowledgeService {
service_with_store(Arc::new(
SqliteGraphStore::open_in_memory().expect("store should open"),
))
.await
}
async fn service_with_store_path(path: &Path) -> RelayKnowledgeService {
service_with_store(Arc::new(
SqliteGraphStore::open(path).expect("persistent store should open"),
))
.await
}
async fn service_with_partitioned_store(root: &Path) -> (RelayKnowledgeService, RuntimePaths) {
fs::create_dir_all(root).expect("partitioned runtime root should exist");
let environment = EnvironmentConfig::from_pairs(
PlatformKind::Unix,
[
("HOME", "/home/alice"),
("TMPDIR", "/tmp"),
(
"RELAY_KNOWLEDGE_HOME",
root.to_str().expect("runtime root should be UTF-8"),
),
],
)
.expect("partitioned environment should parse");
let paths = RuntimePaths::resolve(&environment.platform, &environment.paths)
.expect("partitioned runtime paths should resolve");
let store = PartitionedSqliteKnowledgeStore::open(paths.database_file(), paths.clone())
.expect("partitioned store should open");
(service_with_store(Arc::new(store)).await, paths)
}
async fn service_with_store(store: Arc<dyn KnowledgeStore>) -> RelayKnowledgeService {
let environment = EnvironmentConfig::from_pairs(
PlatformKind::Unix,
[
("HOME", "/home/alice"),
("TMPDIR", "/tmp"),
("RELAY_KNOWLEDGE_HOME", "/srv/relay"),
],
)
.expect("environment should parse");
let runtime = RuntimeConfiguration::from_environment(&environment)
.await
.expect("runtime should compose");
RelayKnowledgeService::with_store(runtime, store)
}
fn sqlite_data_version(connection: &Connection) -> i64 {
connection
.query_row("PRAGMA data_version", [], |row| row.get(0))
.expect("SQLite data version should load")
}
fn mark_software_projection_stale(path: &Path, source_scope: &str) {
let connection = Connection::open(path).expect("store should open for test mutation");
let changed = connection
.execute(
"
UPDATE software_global_status
SET stale = 1,
last_error = 'previous projection failure'
WHERE source_scope = ?1
",
params![source_scope],
)
.expect("software status should be marked stale");
assert_eq!(changed, 1);
}
struct FixtureRepo {
path: PathBuf,
}
impl FixtureRepo {
fn create(name: &str) -> Self {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock should be after epoch")
.as_nanos();
let path = std::env::temp_dir().join(format!("relay-knowledge-{name}-{nanos}"));
fs::create_dir_all(path.join("src")).expect("repo directory should be created");
let repo = Self { path };
repo.git(["init"]);
repo.git(["config", "user.email", "relay@example.invalid"]);
repo.git(["config", "user.name", "Relay Test"]);
repo
}
fn write(&self, relative: &str, content: &str) {
let path = self.path.join(relative);
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).expect("parent directory should exist");
}
fs::write(path, content).expect("fixture file should be written");
}
fn git<const N: usize>(&self, args: [&str; N]) {
let output = git_command(&self.path, args)
.output()
.expect("git should run");
assert!(
output.status.success(),
"git failed: {}",
String::from_utf8_lossy(&output.stderr)
);
}
fn git_text<const N: usize>(&self, args: [&str; N]) -> String {
let output = git_command(&self.path, args)
.output()
.expect("git should run");
assert!(output.status.success());
String::from_utf8_lossy(&output.stdout).trim().to_owned()
}
}
fn git_command<const N: usize>(path: &Path, args: [&str; N]) -> Command {
let mut command = Command::new("git");
command.current_dir(path).args(args);
command
}