code_moniker_workspace/code/

index.rs

// code-moniker: ignore-file[smell-clone-reflex]
// Code index refresh and graph diffing clone stable IDs into owned snapshots/diffs.
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use code_moniker_core::core::moniker::Moniker;
use rayon::prelude::*;
use rustc_hash::FxHashMap;

use crate::code::{def_kind, is_navigable_def, last_name, ref_kind};
use crate::environment::SourceFileSet;
use crate::lines::LineIndex;
use crate::snapshot::{
	CodeIndex, CodeIndexTimings, ExtractionMeasurement, RecordTable, ReferenceId, ReferenceRecord,
	SourceCatalog, SourceFileRecord, SourceId, SourceUnit, SymbolId, SymbolInventoryIndex,
	SymbolRecord, WorkspaceCancellation, WorkspaceFailure, WorkspaceResource, WorkspaceResult,
};
use crate::source::{
	CodeIndexMaterial, IndexedSourceFile, LocalResourceCache, SourceCatalogMaterial,
};

use crate::source::LocalIdentityResolver;

pub trait CodeIndexPort {
	fn build_index(&mut self, catalog: &SourceCatalog) -> WorkspaceResult<CodeIndex>;
	fn build_index_cancellable(
		&mut self,
		catalog: &SourceCatalog,
		cancellation: &WorkspaceCancellation,
	) -> WorkspaceResult<CodeIndex> {
		cancellation.check(WorkspaceResource::CodeIndex)?;
		let index = self.build_index(catalog)?;
		cancellation.check(WorkspaceResource::CodeIndex)?;
		Ok(index)
	}
	fn refresh_paths(
		&mut self,
		current: &CodeIndex,
		paths: &[PathBuf],
	) -> WorkspaceResult<CodeIndexRefresh>;
	fn refresh_paths_cancellable(
		&mut self,
		current: &CodeIndex,
		paths: &[PathBuf],
		cancellation: &WorkspaceCancellation,
	) -> WorkspaceResult<CodeIndexRefresh> {
		cancellation.check(WorkspaceResource::CodeIndex)?;
		let refresh = self.refresh_paths(current, paths)?;
		cancellation.check(WorkspaceResource::CodeIndex)?;
		Ok(refresh)
	}
	fn refresh_catalog_paths(
		&mut self,
		current: &CodeIndex,
		catalog: &SourceCatalog,
		paths: &[PathBuf],
	) -> WorkspaceResult<CodeIndexRefresh>;
	fn refresh_catalog_paths_cancellable(
		&mut self,
		current: &CodeIndex,
		catalog: &SourceCatalog,
		paths: &[PathBuf],
		cancellation: &WorkspaceCancellation,
	) -> WorkspaceResult<CodeIndexRefresh> {
		cancellation.check(WorkspaceResource::CodeIndex)?;
		let refresh = self.refresh_catalog_paths(current, catalog, paths)?;
		cancellation.check(WorkspaceResource::CodeIndex)?;
		Ok(refresh)
	}
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CodeIndexRefresh {
	pub index: CodeIndex,
	pub changed_sources: Vec<SourceId>,
	pub graph_diff: CodeIndexGraphDiff,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct CodeIndexGraphDiff {
	/// Symbols added by the refresh.
	pub added_symbols: Vec<SymbolId>,
	/// Existing symbols whose linkage-relevant fields changed.
	pub modified_symbols: Vec<SymbolId>,
	/// Added or linkage-modified symbols used to refresh graph consumers.
	pub changed_symbols: Vec<SymbolId>,
	/// Symbols removed by the refresh.
	pub removed_symbols: Vec<SymbolId>,
	pub modified_symbol_identities: Vec<String>,
	/// Existing symbols whose inventory-only fields changed.
	///
	/// These changes invalidate inventory consumers such as workspace statistics,
	/// but do not require linkage reconstruction.
	pub modified_inventory_symbols: Vec<SymbolId>,
	pub modified_inventory_symbol_identities: Vec<String>,
	pub removed_symbol_identities: Vec<String>,
	pub changed_references: Vec<ReferenceId>,
	pub removed_references: Vec<ReferenceId>,
	pub removed_reference_kinds: Vec<String>,
	pub symbol_id_remaps: Vec<(SymbolId, SymbolId)>,
	pub reference_id_remaps: Vec<(ReferenceId, ReferenceId)>,
	pub unchanged_symbols: usize,
	pub unchanged_references: usize,
}

impl CodeIndexGraphDiff {
	pub fn changed_symbol_count(&self) -> usize {
		self.changed_linkage_symbol_count() + self.changed_inventory_symbol_count()
	}

	pub fn changed_linkage_symbol_count(&self) -> usize {
		self.changed_symbols.len() + self.removed_symbols.len()
	}

	pub fn changed_inventory_symbol_count(&self) -> usize {
		self.modified_inventory_symbols.len()
	}

	pub fn changed_reference_count(&self) -> usize {
		self.changed_references.len() + self.removed_references.len()
	}
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct LocalCodeIndexOptions {
	pub cache_dir: Option<PathBuf>,
	pub detailed_telemetry: bool,
}

impl LocalCodeIndexOptions {
	pub fn new(cache_dir: Option<PathBuf>) -> Self {
		Self {
			cache_dir,
			detailed_telemetry: false,
		}
	}

	pub fn with_detailed_telemetry(mut self, enabled: bool) -> Self {
		self.detailed_telemetry = enabled;
		self
	}
}

pub struct LocalCodeIndex {
	options: LocalCodeIndexOptions,
	cache: LocalResourceCache,
}

impl LocalCodeIndex {
	pub fn new(options: LocalCodeIndexOptions, cache: LocalResourceCache) -> Self {
		Self { options, cache }
	}

	pub fn build_index_from_extracted(
		&mut self,
		sources: SourceFileSet,
		identity: LocalIdentityResolver,
		files: Vec<IndexedSourceFile>,
	) -> WorkspaceResult<(SourceCatalog, CodeIndex)> {
		build_local_code_index_from_extracted(&self.cache, sources, identity, files)
	}
}

impl CodeIndexPort for LocalCodeIndex {
	fn build_index(&mut self, catalog: &SourceCatalog) -> WorkspaceResult<CodeIndex> {
		build_local_code_index(
			&self.cache,
			&self.options,
			catalog,
			&WorkspaceCancellation::default(),
		)
	}

	fn build_index_cancellable(
		&mut self,
		catalog: &SourceCatalog,
		cancellation: &WorkspaceCancellation,
	) -> WorkspaceResult<CodeIndex> {
		build_local_code_index(&self.cache, &self.options, catalog, cancellation)
	}

	fn refresh_paths(
		&mut self,
		current: &CodeIndex,
		paths: &[PathBuf],
	) -> WorkspaceResult<CodeIndexRefresh> {
		refresh_local_code_index(
			&self.cache,
			&self.options,
			current,
			None,
			paths,
			&WorkspaceCancellation::default(),
		)
	}

	fn refresh_paths_cancellable(
		&mut self,
		current: &CodeIndex,
		paths: &[PathBuf],
		cancellation: &WorkspaceCancellation,
	) -> WorkspaceResult<CodeIndexRefresh> {
		refresh_local_code_index(
			&self.cache,
			&self.options,
			current,
			None,
			paths,
			cancellation,
		)
	}

	fn refresh_catalog_paths(
		&mut self,
		current: &CodeIndex,
		catalog: &SourceCatalog,
		paths: &[PathBuf],
	) -> WorkspaceResult<CodeIndexRefresh> {
		refresh_local_code_index(
			&self.cache,
			&self.options,
			current,
			Some(catalog),
			paths,
			&WorkspaceCancellation::default(),
		)
	}

	fn refresh_catalog_paths_cancellable(
		&mut self,
		current: &CodeIndex,
		catalog: &SourceCatalog,
		paths: &[PathBuf],
		cancellation: &WorkspaceCancellation,
	) -> WorkspaceResult<CodeIndexRefresh> {
		refresh_local_code_index(
			&self.cache,
			&self.options,
			current,
			Some(catalog),
			paths,
			cancellation,
		)
	}
}

fn build_local_code_index(
	cache: &LocalResourceCache,
	options: &LocalCodeIndexOptions,
	catalog: &SourceCatalog,
	cancellation: &WorkspaceCancellation,
) -> WorkspaceResult<CodeIndex> {
	cancellation.check(WorkspaceResource::CodeIndex)?;
	let total_timer = Instant::now();
	let source_material = source_material(cache, catalog)?;
	let generation = cache.next_generation();
	let extract_timer = Instant::now();
	let (files, extraction_workers) = extract_source_files(
		&source_material,
		options.cache_dir.as_deref(),
		cancellation,
		options.detailed_telemetry,
	)?;
	let extraction_jobs = files.len();
	let extract_sources = extract_timer.elapsed();
	let extraction = if options.detailed_telemetry {
		extraction_measurements(&files, None)
	} else {
		Default::default()
	};
	cancellation.check(WorkspaceResource::CodeIndex)?;
	let semantic_timer = Instant::now();
	let (symbols, references, material) =
		build_semantic_index(source_material, files, cancellation)?;
	let semantic_index = semantic_timer.elapsed();
	let mut sources = source_records(&material);
	let identity_scheme = material.identity.scheme().to_string();
	cache.insert_index(generation, material);
	sources.shrink_to_fit();
	let inventory = Arc::new(SymbolInventoryIndex::build(generation, &sources, &symbols));
	Ok(CodeIndex {
		generation,
		catalog_generation: catalog.generation,
		identity_scheme,
		sources,
		symbols,
		references,
		inventory,
		timings: CodeIndexTimings {
			extract_sources,
			semantic_index,
			total: total_timer.elapsed(),
			extraction,
			extraction_jobs,
			extraction_workers,
		},
	})
}

fn build_code_index_from_extracted(
	cache: &LocalResourceCache,
	catalog: &SourceCatalog,
	source_material: SourceCatalogMaterial,
	files: Vec<Arc<IndexedSourceFile>>,
) -> WorkspaceResult<CodeIndex> {
	let generation = cache.next_generation();
	let cancellation = WorkspaceCancellation::default();
	let (symbols, references, material) =
		build_semantic_index(source_material, files, &cancellation)?;
	let mut sources = source_records(&material);
	let identity_scheme = material.identity.scheme().to_string();
	cache.insert_index(generation, material);
	sources.shrink_to_fit();
	let inventory = Arc::new(SymbolInventoryIndex::build(generation, &sources, &symbols));
	Ok(CodeIndex {
		generation,
		catalog_generation: catalog.generation,
		identity_scheme,
		sources,
		symbols,
		references,
		inventory,
		timings: CodeIndexTimings {
			extract_sources: Duration::ZERO,
			semantic_index: Duration::ZERO,
			total: Duration::ZERO,
			extraction: Vec::new(),
			extraction_jobs: 0,
			extraction_workers: 0,
		},
	})
}

fn build_local_code_index_from_extracted(
	cache: &LocalResourceCache,
	sources: SourceFileSet,
	identity: LocalIdentityResolver,
	files: Vec<IndexedSourceFile>,
) -> WorkspaceResult<(SourceCatalog, CodeIndex)> {
	validate_extracted_files(&sources, &identity, &files)?;
	let catalog_generation = cache.next_generation();
	let units = sources
		.files
		.iter()
		.enumerate()
		.map(|(file_idx, file)| {
			SourceUnit::with_language(
				identity.source_id(file_idx, &file.rel_path),
				crate::path_util::portable_path(&file.rel_path),
				file.lang.tag(),
			)
		})
		.collect();
	let catalog = SourceCatalog::new(catalog_generation, units);
	let source_material = SourceCatalogMaterial {
		sources,
		identity,
		memory_sources: BTreeMap::new(),
		memory_slots: BTreeSet::new(),
		memory_revisions: BTreeMap::new(),
	};
	cache.insert_sources(catalog_generation, source_material.clone());
	let index = build_code_index_from_extracted(
		cache,
		&catalog,
		source_material,
		files.into_iter().map(Arc::new).collect(),
	)?;
	Ok((catalog, index))
}

fn validate_extracted_files(
	sources: &SourceFileSet,
	identity: &LocalIdentityResolver,
	files: &[IndexedSourceFile],
) -> WorkspaceResult<()> {
	if sources.files.len() != files.len() {
		return Err(WorkspaceFailure::new(
			WorkspaceResource::CodeIndex,
			format!(
				"extracted file count {} does not match source catalog count {}",
				files.len(),
				sources.files.len()
			),
		));
	}
	for (file_idx, (source, extracted)) in sources.files.iter().zip(files).enumerate() {
		let expected_id = identity.source_id(file_idx, &source.rel_path);
		let matches = source.source == extracted.source_root
			&& expected_id == extracted.source_id
			&& source.path == extracted.path
			&& source.rel_path == extracted.rel_path
			&& source.anchor == extracted.anchor
			&& source.lang == extracted.lang
			&& extracted.identity == *identity;
		if !matches {
			return Err(WorkspaceFailure::new(
				WorkspaceResource::CodeIndex,
				format!(
					"extracted file {} does not match source catalog slot {file_idx}",
					extracted.rel_path.display()
				),
			));
		}
	}
	Ok(())
}

fn refresh_local_code_index(
	cache: &LocalResourceCache,
	options: &LocalCodeIndexOptions,
	current: &CodeIndex,
	extended_catalog: Option<&SourceCatalog>,
	paths: &[PathBuf],
	cancellation: &WorkspaceCancellation,
) -> WorkspaceResult<CodeIndexRefresh> {
	cancellation.check(WorkspaceResource::CodeIndex)?;
	let total_timer = Instant::now();
	let current_material = cache.index_material(current.generation).ok_or_else(|| {
		WorkspaceFailure::new(
			WorkspaceResource::CodeIndex,
			"code index material is unavailable",
		)
	})?;
	let source_catalog = match extended_catalog {
		Some(catalog) => cache.source_material(catalog.generation).ok_or_else(|| {
			WorkspaceFailure::new(
				WorkspaceResource::CodeIndex,
				"extended source catalog material is unavailable",
			)
		})?,
		None => current_material.source_catalog.clone(),
	};
	let mut files = current_material.files.clone();
	let mut changed_sources = Vec::new();
	let mut changed_file_indexes = BTreeSet::new();
	let mut extraction_jobs = BTreeSet::new();
	let extract_timer = Instant::now();
	let extraction_parent = options
		.detailed_telemetry
		.then(tracing::Span::current)
		.unwrap_or_else(tracing::Span::none);
	refresh_retired_slots(RetiredSlotRefresh {
		previous_catalog: &current_material.source_catalog,
		source_catalog: &source_catalog,
		files: &mut files,
		changed_sources: &mut changed_sources,
		changed_file_indexes: &mut changed_file_indexes,
		extraction_jobs: &mut extraction_jobs,
	})?;
	for file_idx in files.len()..source_catalog.sources.files.len() {
		extraction_jobs.insert(file_idx);
	}
	for path in paths {
		let Some(source) = source_catalog.resolve_source(path) else {
			continue;
		};
		let Some(file_idx) = source.eager_index else {
			continue;
		};
		if changed_file_indexes.contains(&file_idx)
			|| source_catalog.sources.files[file_idx].retired
		{
			continue;
		}
		extraction_jobs.insert(file_idx);
	}
	let extraction_job_count = extraction_jobs.len();
	let extracted = extract_source_file_jobs(
		&source_catalog,
		extraction_jobs,
		options.cache_dir.as_deref(),
		cancellation,
		&extraction_parent,
		options.detailed_telemetry,
	)?
	.ready_to_merge(cancellation)?;
	let extraction_workers = extracted.workers;
	for (file_idx, indexed) in extracted.files {
		push_unique_source(&mut changed_sources, indexed.source_id);
		changed_file_indexes.insert(file_idx);
		if file_idx == files.len() {
			files.push(indexed);
		} else if let Some(slot) = files.get_mut(file_idx) {
			*slot = indexed;
		}
	}
	let extract_sources = extract_timer.elapsed();
	if changed_sources.is_empty() {
		let mut index = current.clone();
		index.catalog_generation = extended_catalog
			.map(|catalog| catalog.generation)
			.unwrap_or(current.catalog_generation);
		index.timings = CodeIndexTimings {
			extract_sources,
			semantic_index: Duration::ZERO,
			total: total_timer.elapsed(),
			extraction: Vec::new(),
			extraction_jobs: 0,
			extraction_workers: 0,
		};
		if extended_catalog.is_some() {
			let mut material = current_material.as_ref().clone();
			material.source_catalog = source_catalog;
			cache.insert_index(current.generation, material);
		}
		return Ok(CodeIndexRefresh {
			index,
			changed_sources,
			graph_diff: CodeIndexGraphDiff::default(),
		});
	}
	let semantic_timer = Instant::now();
	let extraction = if options.detailed_telemetry {
		extraction_measurements(&files, Some(&changed_file_indexes))
	} else {
		Default::default()
	};
	let material = material_from_files(source_catalog, files, cancellation)?;
	let sources = source_records(&material);
	let graph_diff = graph_diff(current_material.as_ref(), &material, &changed_file_indexes);
	let mut symbols = current.symbols.clone();
	let mut references = current.references.clone();
	for file_idx in &changed_file_indexes {
		let (file_symbols, file_references) =
			records_for_file(*file_idx, &material.files[*file_idx]);
		symbols.replace(*file_idx, Arc::from(file_symbols));
		references.replace(*file_idx, Arc::from(file_references));
	}
	let semantic_index = semantic_timer.elapsed();
	let generation = cache.next_generation();
	let identity_scheme = material.identity.scheme().to_string();
	cache_refreshed_index(cache, current, generation, material, &graph_diff);
	let inventory = Arc::new(current.inventory.refresh(
		generation,
		&sources,
		&symbols,
		&changed_file_indexes,
	));
	Ok(CodeIndexRefresh {
		index: CodeIndex {
			generation,
			catalog_generation: extended_catalog
				.map(|catalog| catalog.generation)
				.unwrap_or(current.catalog_generation),
			identity_scheme,
			sources,
			symbols,
			references,
			inventory,
			timings: CodeIndexTimings {
				extract_sources,
				semantic_index,
				total: total_timer.elapsed(),
				extraction,
				extraction_jobs: extraction_job_count,
				extraction_workers,
			},
		},
		changed_sources,
		graph_diff,
	})
}

fn cache_refreshed_index(
	cache: &LocalResourceCache,
	current: &CodeIndex,
	generation: crate::snapshot::ResourceGeneration,
	material: CodeIndexMaterial,
	graph_diff: &CodeIndexGraphDiff,
) {
	cache.insert_index(generation, material);
	cache.insert_index_diff(generation, current.generation, graph_diff.clone());
}

struct RetiredSlotRefresh<'a> {
	previous_catalog: &'a SourceCatalogMaterial,
	source_catalog: &'a SourceCatalogMaterial,
	files: &'a mut Vec<Arc<IndexedSourceFile>>,
	changed_sources: &'a mut Vec<SourceId>,
	changed_file_indexes: &'a mut BTreeSet<usize>,
	extraction_jobs: &'a mut BTreeSet<usize>,
}

fn refresh_retired_slots(refresh: RetiredSlotRefresh<'_>) -> WorkspaceResult<()> {
	let slots = refresh
		.files
		.len()
		.min(refresh.source_catalog.sources.files.len());
	for file_idx in 0..slots {
		let was_retired = refresh.previous_catalog.sources.files[file_idx].retired;
		let is_retired = refresh.source_catalog.sources.files[file_idx].retired;
		if was_retired == is_retired {
			continue;
		}
		if !is_retired {
			refresh.extraction_jobs.insert(file_idx);
			continue;
		}
		let indexed = tombstone_file(&refresh.files[file_idx]);
		push_unique_source(refresh.changed_sources, indexed.source_id);
		refresh.changed_file_indexes.insert(file_idx);
		refresh.files[file_idx] = Arc::new(indexed);
	}
	Ok(())
}

fn tombstone_file(previous: &IndexedSourceFile) -> IndexedSourceFile {
	IndexedSourceFile {
		source_root: previous.source_root,
		source_id: previous.source_id,
		source_uri: previous.source_uri.clone(),
		identity: previous.identity.clone(),
		path: previous.path.clone(),
		rel_path: previous.rel_path.clone(),
		anchor: previous.anchor.clone(),
		lang: previous.lang,
		graph: code_moniker_core::core::code_graph::CodeGraph::from_records(Vec::new(), Vec::new()),
		source: String::new(),
		extraction_cache: "retired",
		extraction_duration: Duration::ZERO,
	}
}

fn source_material(
	cache: &LocalResourceCache,
	catalog: &SourceCatalog,
) -> WorkspaceResult<SourceCatalogMaterial> {
	cache.source_material(catalog.generation).ok_or_else(|| {
		WorkspaceFailure::new(
			WorkspaceResource::CodeIndex,
			"source catalog material is unavailable",
		)
	})
}

fn extract_source_files(
	source_material: &SourceCatalogMaterial,
	cache_dir: Option<&std::path::Path>,
	cancellation: &WorkspaceCancellation,
	detailed_telemetry: bool,
) -> WorkspaceResult<(Vec<Arc<IndexedSourceFile>>, usize)> {
	let parent = detailed_telemetry
		.then(tracing::Span::current)
		.unwrap_or_else(tracing::Span::none);
	let extracted = extract_source_file_jobs(
		source_material,
		0..source_material.sources.files.len(),
		cache_dir,
		cancellation,
		&parent,
		detailed_telemetry,
	)?
	.ready_to_merge(cancellation)?;
	Ok((
		extracted.files.into_iter().map(|(_, file)| file).collect(),
		extracted.workers,
	))
}

struct ExtractedFileBatch {
	files: Vec<(usize, Arc<IndexedSourceFile>)>,
	workers: usize,
}

impl ExtractedFileBatch {
	fn ready_to_merge(self, cancellation: &WorkspaceCancellation) -> WorkspaceResult<Self> {
		cancellation.check(WorkspaceResource::CodeIndex)?;
		Ok(self)
	}
}

fn extract_source_file_jobs(
	source_material: &SourceCatalogMaterial,
	file_indexes: impl IntoParallelIterator<Item = usize>,
	cache_dir: Option<&Path>,
	cancellation: &WorkspaceCancellation,
	parent: &tracing::Span,
	detailed_telemetry: bool,
) -> WorkspaceResult<ExtractedFileBatch> {
	let worker_usage = (0..rayon::current_num_threads())
		.map(|_| AtomicBool::new(false))
		.collect::<Vec<_>>();
	let external_worker_used = AtomicBool::new(false);
	let mut files = file_indexes
		.into_par_iter()
		.map(|file_idx| {
			if let Some(worker_idx) = rayon::current_thread_index() {
				if let Some(used) = worker_usage.get(worker_idx) {
					used.store(true, Ordering::Relaxed);
				}
			} else {
				external_worker_used.store(true, Ordering::Relaxed);
			}
			cancellation.check(WorkspaceResource::CodeIndex)?;
			let file = source_material.sources.files.get(file_idx).ok_or_else(|| {
				WorkspaceFailure::new(
					WorkspaceResource::CodeIndex,
					format!("source file index {file_idx} is unavailable"),
				)
			})?;
			extract_source_file(
				source_material,
				file_idx,
				&file.path,
				cache_dir,
				parent,
				detailed_telemetry,
			)
			.map(|file| (file_idx, Arc::new(file)))
		})
		.collect::<WorkspaceResult<Vec<_>>>()?;
	files.sort_by_key(|(file_idx, _)| *file_idx);
	let workers = worker_usage
		.iter()
		.filter(|used| used.load(Ordering::Relaxed))
		.count()
		+ usize::from(external_worker_used.load(Ordering::Relaxed));
	Ok(ExtractedFileBatch { files, workers })
}

fn extract_source_file(
	source_material: &SourceCatalogMaterial,
	file_idx: usize,
	path: &Path,
	cache_dir: Option<&Path>,
	parent: &tracing::Span,
	detailed_telemetry: bool,
) -> WorkspaceResult<IndexedSourceFile> {
	let file = source_material.sources.files.get(file_idx).ok_or_else(|| {
		WorkspaceFailure::new(
			WorkspaceResource::CodeIndex,
			format!("source file index {file_idx} is unavailable"),
		)
	})?;
	let root = source_material
		.sources
		.roots
		.get(file.source)
		.ok_or_else(|| {
			WorkspaceFailure::new(
				WorkspaceResource::CodeIndex,
				format!("source root {} is unavailable", file.source),
			)
		})?;
	let ctx = file.extraction_context(root);
	let started = detailed_telemetry.then(Instant::now);
	let span = if detailed_telemetry {
		tracing::info_span!(
			parent: parent,
			"workspace.extract_file",
			file.path = %file.rel_path.display(),
			file.language = file.lang.tag(),
			file.source_bytes = tracing::field::Empty,
			cache.result = tracing::field::Empty,
			graph.definitions = tracing::field::Empty,
			graph.references = tracing::field::Empty,
		)
	} else {
		tracing::Span::none()
	};
	let _entered = span.enter();
	let (graph, source, cache_status) = match source_material.memory_source(path) {
		Some(source) => (
			crate::environment::extract_source_with(file.lang, source, &file.anchor, &ctx),
			source.to_owned(),
			"memory",
		),
		None => {
			let (graph, extracted_source, cache_outcome) =
				crate::cache::load_or_extract_workspace_result(
					path,
					&file.anchor,
					file.lang,
					cache_dir,
					&ctx,
				)
				.map_err(|err| {
					WorkspaceFailure::new(
						WorkspaceResource::CodeIndex,
						format!("cannot extract {}: {err}", path.display()),
					)
				})?;
			let source = match extracted_source {
				Some(source) => source,
				None => crate::cache::read_source_lossy(path).map_err(|err| {
					WorkspaceFailure::new(
						WorkspaceResource::CodeIndex,
						format!("cannot read {}: {err}", path.display()),
					)
				})?,
			};
			(graph, source, cache_outcome.as_str())
		}
	};
	let elapsed = started.map_or(Duration::ZERO, |started| started.elapsed());
	if detailed_telemetry {
		span.record("file.source_bytes", source.len());
		span.record("cache.result", cache_status);
		span.record("graph.definitions", graph.def_count());
		span.record("graph.references", graph.ref_count());
	}
	Ok(IndexedSourceFile {
		source_root: file.source,
		source_id: source_material
			.source_id_for_file(file_idx)
			.ok_or_else(|| {
				WorkspaceFailure::new(
					WorkspaceResource::CodeIndex,
					format!("source id is unavailable for {}", file.rel_path.display()),
				)
			})?,
		source_uri: source_material
			.source_uri_for_path(&file.path)
			.ok_or_else(|| {
				WorkspaceFailure::new(
					WorkspaceResource::CodeIndex,
					format!("source uri is unavailable for {}", file.path.display()),
				)
			})?,
		identity: source_material.identity.clone(),
		path: file.path.clone(),
		rel_path: file.rel_path.clone(),
		anchor: file.anchor.clone(),
		lang: file.lang,
		graph,
		source,
		extraction_cache: cache_status,
		extraction_duration: elapsed,
	})
}

fn extraction_measurements(
	files: &[Arc<IndexedSourceFile>],
	selected: Option<&BTreeSet<usize>>,
) -> Vec<ExtractionMeasurement> {
	let mut groups = BTreeMap::<(&'static str, &'static str), ExtractionMeasurement>::new();
	for (file_idx, file) in files.iter().enumerate() {
		if selected.is_some_and(|selected| !selected.contains(&file_idx)) {
			continue;
		}
		let language = file.lang.tag();
		let cache = file.extraction_cache;
		let entry = groups
			.entry((language, cache))
			.or_insert_with(|| ExtractionMeasurement {
				language,
				cache,
				..ExtractionMeasurement::default()
			});
		entry.files += 1;
		entry.source_bytes += file.source.len();
		entry.duration += file.extraction_duration;
	}
	groups.into_values().collect()
}

fn build_semantic_index(
	source_material: SourceCatalogMaterial,
	files: Vec<Arc<IndexedSourceFile>>,
	cancellation: &WorkspaceCancellation,
) -> WorkspaceResult<(
	RecordTable<SymbolRecord>,
	RecordTable<ReferenceRecord>,
	CodeIndexMaterial,
)> {
	let mut symbol_shards = Vec::with_capacity(files.len());
	let mut reference_shards = Vec::with_capacity(files.len());
	for (file_idx, file) in files.iter().enumerate() {
		cancellation.check(WorkspaceResource::CodeIndex)?;
		let (symbols, references) = records_for_file(file_idx, file);
		symbol_shards.push(Arc::from(symbols));
		reference_shards.push(Arc::from(references));
	}
	let material = material_from_files(source_material, files, cancellation)?;
	Ok((
		RecordTable::from_shards(symbol_shards),
		RecordTable::from_shards(reference_shards),
		material,
	))
}

fn material_from_files(
	source_material: SourceCatalogMaterial,
	mut files: Vec<Arc<IndexedSourceFile>>,
	cancellation: &WorkspaceCancellation,
) -> WorkspaceResult<CodeIndexMaterial> {
	let symbol_count = files.iter().map(|file| file.graph.def_count()).sum();
	let mut symbols_by_moniker = rustc_hash::FxHashMap::default();
	symbols_by_moniker.reserve(symbol_count);
	for (file_idx, file) in files.iter().enumerate() {
		cancellation.check(WorkspaceResource::CodeIndex)?;
		for (def_idx, def) in file.graph.defs().enumerate() {
			symbols_by_moniker.insert(
				def.moniker.clone(),
				file.identity.symbol_id(file_idx, def_idx),
			);
		}
	}
	symbols_by_moniker.shrink_to_fit();
	files.shrink_to_fit();
	let identity = source_material.identity.clone();
	Ok(CodeIndexMaterial {
		source_catalog: source_material,
		files,
		identity,
		symbols_by_moniker,
	})
}

fn graph_diff(
	previous: &CodeIndexMaterial,
	next: &CodeIndexMaterial,
	changed_files: &BTreeSet<usize>,
) -> CodeIndexGraphDiff {
	let mut diff = CodeIndexGraphDiff::default();
	for file_idx in changed_files {
		let Some(next_file) = next.files.get(*file_idx) else {
			continue;
		};
		let (previous_symbols, previous_references) = match previous.files.get(*file_idx) {
			Some(previous_file) => records_for_file(*file_idx, previous_file),
			None => (Vec::new(), Vec::new()),
		};
		let (next_symbols, next_references) = records_for_file(*file_idx, next_file);
		diff_symbols(&previous_symbols, &next_symbols, &mut diff);
		diff_references(
			&previous_references,
			previous,
			&next_references,
			next,
			&mut diff,
		);
	}
	diff
}

fn records_for_file(
	file_idx: usize,
	file: &IndexedSourceFile,
) -> (Vec<SymbolRecord>, Vec<ReferenceRecord>) {
	let line_index = LineIndex::new(&file.source);
	let mut symbols = Vec::with_capacity(file.graph.def_count());
	collect_symbols(file_idx, file, &line_index, &mut symbols);
	let mut reference_identity_pool = TargetIdentityPool::default();
	let mut references = Vec::with_capacity(file.graph.ref_count());
	collect_references(
		file_idx,
		file,
		&line_index,
		&mut references,
		&mut reference_identity_pool,
	);
	(symbols, references)
}

fn diff_symbols(previous: &[SymbolRecord], next: &[SymbolRecord], diff: &mut CodeIndexGraphDiff) {
	let mut next_by_key = symbol_record_indexes(next);
	for previous_symbol in previous {
		let key = symbol_key(previous_symbol);
		let Some(next_idx) = pop_index(&mut next_by_key, &key) else {
			diff.removed_symbols.push(previous_symbol.id);
			diff.removed_symbol_identities
				.push(previous_symbol.identity.to_string());
			continue;
		};
		let next_symbol = &next[next_idx];
		if symbol_linkage_fields_changed(previous_symbol, next_symbol) {
			diff.modified_symbols.push(next_symbol.id);
			diff.modified_symbol_identities
				.push(next_symbol.identity.to_string());
			diff.changed_symbols.push(next_symbol.id);
			continue;
		}
		let inventory_changed = symbol_inventory_fields_changed(previous_symbol, next_symbol);
		if inventory_changed {
			diff.modified_inventory_symbols.push(next_symbol.id);
			diff.modified_inventory_symbol_identities
				.push(next_symbol.identity.to_string());
		}
		if previous_symbol.id != next_symbol.id {
			diff.symbol_id_remaps
				.push((previous_symbol.id, next_symbol.id));
		}
		if inventory_changed {
			continue;
		}
		diff.unchanged_symbols += 1;
	}
	for indexes in next_by_key.into_values() {
		for idx in indexes {
			diff.added_symbols.push(next[idx].id);
			diff.changed_symbols.push(next[idx].id);
		}
	}
}

fn diff_references(
	previous: &[ReferenceRecord],
	previous_material: &CodeIndexMaterial,
	next: &[ReferenceRecord],
	next_material: &CodeIndexMaterial,
	diff: &mut CodeIndexGraphDiff,
) {
	let mut next_by_key = reference_record_indexes(next, next_material);
	for previous_reference in previous {
		let Some(key) = reference_key(previous_reference, previous_material) else {
			diff.removed_references.push(previous_reference.id);
			diff.removed_reference_kinds
				.push(previous_reference.kind.clone());
			continue;
		};
		let Some(next_idx) = pop_index(&mut next_by_key, &key) else {
			diff.removed_references.push(previous_reference.id);
			diff.removed_reference_kinds
				.push(previous_reference.kind.clone());
			continue;
		};
		let next_reference = &next[next_idx];
		if previous_reference.id != next_reference.id {
			diff.reference_id_remaps
				.push((previous_reference.id, next_reference.id));
		}
		diff.unchanged_references += 1;
	}
	for indexes in next_by_key.into_values() {
		for idx in indexes {
			diff.changed_references.push(next[idx].id);
		}
	}
}

fn symbol_record_indexes(records: &[SymbolRecord]) -> FxHashMap<Arc<str>, Vec<usize>> {
	let mut by_key = FxHashMap::<Arc<str>, Vec<usize>>::default();
	for (idx, record) in records.iter().enumerate() {
		by_key.entry(symbol_key(record)).or_default().push(idx);
	}
	by_key
}

fn reference_record_indexes(
	records: &[ReferenceRecord],
	material: &CodeIndexMaterial,
) -> FxHashMap<ReferenceKey, Vec<usize>> {
	let mut by_key = FxHashMap::<ReferenceKey, Vec<usize>>::default();
	for (idx, record) in records.iter().enumerate() {
		if let Some(key) = reference_key(record, material) {
			by_key.entry(key).or_default().push(idx);
		}
	}
	by_key
}

fn pop_index<K: Eq + std::hash::Hash>(
	by_key: &mut FxHashMap<K, Vec<usize>>,
	key: &K,
) -> Option<usize> {
	let indexes = by_key.get_mut(key)?;
	let idx = indexes.remove(0);
	if indexes.is_empty() {
		by_key.remove(key);
	}
	Some(idx)
}

fn symbol_key(symbol: &SymbolRecord) -> Arc<str> {
	Arc::clone(&symbol.identity)
}

fn symbol_linkage_fields_changed(previous: &SymbolRecord, next: &SymbolRecord) -> bool {
	previous.identity != next.identity
		|| previous.name != next.name
		|| previous.kind != next.kind
		|| previous.visibility != next.visibility
		|| previous.signature != next.signature
		|| previous.call_name != next.call_name
		|| previous.call_arity != next.call_arity
		|| previous.navigable != next.navigable
}

fn symbol_inventory_fields_changed(previous: &SymbolRecord, next: &SymbolRecord) -> bool {
	previous.line_range != next.line_range || previous.parent != next.parent
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct ReferenceKey {
	source_symbol_identity: String,
	target_identity: String,
	kind: String,
	call_name: Option<String>,
	call_arity: Option<usize>,
	confidence: Option<String>,
	receiver: Option<String>,
	alias: Option<String>,
}

fn reference_key(
	reference: &ReferenceRecord,
	material: &CodeIndexMaterial,
) -> Option<ReferenceKey> {
	let source_symbol_identity = material
		.symbol_moniker(&reference.source_symbol)
		.map(|moniker| material.identity.moniker_uri(moniker))?;
	Some(ReferenceKey {
		source_symbol_identity,
		target_identity: reference.target_identity.to_string(),
		kind: reference.kind.clone(),
		call_name: reference.call_name.clone(),
		call_arity: reference.call_arity,
		confidence: reference.confidence.clone(),
		receiver: reference.receiver.clone(),
		alias: reference.alias.clone(),
	})
}

fn push_unique_source(sources: &mut Vec<SourceId>, source: SourceId) {
	if !sources.iter().any(|existing| existing == &source) {
		sources.push(source);
	}
}

fn collect_symbols(
	file_idx: usize,
	file: &IndexedSourceFile,
	line_index: &LineIndex,
	symbols: &mut Vec<SymbolRecord>,
) {
	for (def_idx, def) in file.graph.defs().enumerate() {
		let id = file.identity.symbol_id(file_idx, def_idx);
		let parent = def
			.parent
			.map(|parent_idx| file.identity.symbol_id(file_idx, parent_idx));
		symbols.push(SymbolRecord {
			id,
			source: file.source_id,
			identity: Arc::from(file.identity.moniker_uri(&def.moniker)),
			name: last_name(&def.moniker),
			kind: def_kind(def),
			visibility: def_visibility(def),
			signature: String::from_utf8_lossy(&def.signature).to_string(),
			call_name: (!def.call_name.is_empty())
				.then(|| String::from_utf8_lossy(&def.call_name).to_string()),
			call_arity: def.call_arity,
			navigable: is_navigable_def(file.lang, def),
			line_range: def
				.position
				.map(|(start, end)| line_index.line_range(start, end)),
			parent,
		});
	}
}

fn def_visibility(def: &code_moniker_core::core::code_graph::DefRecord) -> String {
	std::str::from_utf8(&def.visibility)
		.unwrap_or("")
		.to_string()
}

fn collect_references(
	file_idx: usize,
	file: &IndexedSourceFile,
	line_index: &LineIndex,
	references: &mut Vec<ReferenceRecord>,
	reference_identity_pool: &mut TargetIdentityPool,
) {
	for (ref_idx, reference) in file.graph.refs().enumerate() {
		let id = file.identity.reference_id(file_idx, ref_idx);
		let source_symbol = file.identity.symbol_id(file_idx, reference.source);
		let target_identity = reference_identity_pool.intern(&file.identity, &reference.target);
		references.push(
			ReferenceRecord::new(
				id,
				file.source_id,
				source_symbol,
				target_identity,
				ref_kind(reference),
				reference
					.position
					.map(|(start, end)| line_index.line_range(start, end)),
			)
			.with_call_metadata(ref_attr(&reference.call_name), reference.call_arity)
			.with_metadata(
				ref_attr(&reference.confidence),
				ref_attr(&reference.receiver_hint),
				ref_attr(&reference.alias),
			),
		);
	}
}

#[derive(Default)]
struct TargetIdentityPool {
	values: rustc_hash::FxHashMap<Moniker, Arc<str>>,
}

impl TargetIdentityPool {
	fn intern(&mut self, identity: &LocalIdentityResolver, target: &Moniker) -> Arc<str> {
		if let Some(existing) = self.values.get(target) {
			return Arc::clone(existing);
		}
		let shared = Arc::<str>::from(identity.moniker_uri(target));
		self.values.insert(target.clone(), Arc::clone(&shared));
		shared
	}
}

fn source_records(material: &CodeIndexMaterial) -> Vec<SourceFileRecord> {
	material
		.files
		.iter()
		.map(|file| SourceFileRecord {
			id: file.source_id,
			uri: file.source_uri.clone(),
			source_root: file.source_root,
			path: file.path.display().to_string(),
			rel_path: crate::path_util::portable_path(&file.rel_path),
			anchor: crate::path_util::portable_path(&file.anchor),
			language: file.lang.tag().to_string(),
			text: if material.source_catalog.is_memory_slot(&file.path) {
				file.source.to_owned()
			} else {
				String::new()
			},
		})
		.collect()
}

fn ref_attr(bytes: &[u8]) -> Option<String> {
	if bytes.is_empty() {
		return None;
	}
	std::str::from_utf8(bytes)
		.ok()
		.filter(|value| !value.is_empty())
		.map(ToOwned::to_owned)
}

#[cfg(test)]
mod tests {
	use super::*;
	use crate::snapshot::WorkspaceRequest;
	use crate::source::{
		LocalSourceCatalog, LocalSourceCatalogOptions, MemorySourceDocument, MemorySourceSet,
		SourceCatalogPort,
	};

	#[test]
	fn extraction_collection_uses_multiple_workers_and_cancels_before_merge() {
		let temp = tempfile::tempdir().expect("tempdir");
		let cache = LocalResourceCache::default();
		cache.replace_memory_source_set(MemorySourceSet {
			srcset: "parallel".to_string(),
			revision: Some("1".to_string()),
			documents: (0..128)
				.map(|index| MemorySourceDocument {
					uri: format!("schema/table_{index:03}.sql"),
					lang: code_moniker_core::lang::Lang::Sql,
					content: Arc::from(format!(
						"CREATE TABLE app.table_{index:03} (id bigint, parent_id bigint, label text, metadata jsonb);"
					)),
				})
				.collect(),
		});
		let mut catalog_port = LocalSourceCatalog::new(
			LocalSourceCatalogOptions::new(vec![temp.path().to_path_buf()], None),
			cache.clone(),
		);
		let catalog = catalog_port
			.load_catalog(&WorkspaceRequest::new("parallel-extraction-test"))
			.expect("memory source catalog");
		let material = source_material(&cache, &catalog).expect("source material");
		let cancellation = WorkspaceCancellation::default();
		let pool = rayon::ThreadPoolBuilder::new()
			.num_threads(2)
			.build()
			.expect("two-worker pool");

		let extracted = pool
			.install(|| {
				extract_source_file_jobs(
					&material,
					0..material.sources.files.len(),
					None,
					&cancellation,
					&tracing::Span::none(),
					false,
				)
			})
			.expect("parallel extraction collection");

		assert_eq!(extracted.files.len(), 128);
		assert_eq!(extracted.workers, 2);
		assert_eq!(
			extracted
				.files
				.iter()
				.map(|(file_idx, _)| *file_idx)
				.collect::<Vec<_>>(),
			(0..128).collect::<Vec<_>>()
		);
		cancellation.cancel();
		let error = match extracted.ready_to_merge(&cancellation) {
			Ok(_) => panic!("cancelled extraction batch must not become mergeable"),
			Err(error) => error,
		};
		assert_eq!(error.message, "workspace build cancelled");
	}

	#[test]
	fn line_range_changes_are_inventory_deltas_not_linkage_deltas() {
		let mut previous =
			SymbolRecord::new(SymbolId::at(0, 0), SourceId::at(0), "Invoice", "class");
		previous.identity = Arc::from("code+moniker://./lang:java/class:Invoice");
		previous.line_range = Some((4, 4));
		let mut next = previous.clone();
		next.line_range = Some((4, 13));
		let mut diff = CodeIndexGraphDiff::default();

		diff_symbols(&[previous], &[next.clone()], &mut diff);

		assert_eq!(diff.modified_inventory_symbols, vec![next.id]);
		assert_eq!(
			diff.modified_inventory_symbol_identities,
			vec![next.identity.to_string()]
		);
		assert!(diff.modified_symbols.is_empty());
		assert!(diff.modified_symbol_identities.is_empty());
		assert_eq!(diff.unchanged_symbols, 0);
		assert_eq!(diff.changed_symbol_count(), 1);
		assert_eq!(diff.changed_linkage_symbol_count(), 0);
	}
}