scryer-engine 0.1.0

use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use rayon::iter::{IntoParallelIterator, IntoParallelRefIterator, ParallelIterator};
use scryer_db::{DependencyPackage, Project, ProjectDependency, ScryerDb};

use crate::dependency::{CargoDependencyProvider, DependencyProvider, DependencySourceType};
use crate::hasher::{ChangeDetector, FileChange};
use crate::ingest::{BatchIngestionActor, ExistingFileStatus, IngestionMessage};
use crate::parsers::{ParseMode, parse_file};
use crate::payload::{ParsedFilePayload, RawTypeContract};
use crate::scanner::WorkspaceScanner;
use crate::stack_graph::engine::StepBoundedCancellationFlag;
use crate::stack_graph::pool::CachedDependencyPackage;
use crate::stack_graph::{CrossFileResolver, ProjectEnginePool, ResolvedDefinition};
use crate::watcher::WorkspaceWatcher;

/// Outcome report summarizing an indexing pass.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct IndexReport {
    pub scanned_files: usize,
    pub added_files: usize,
    pub modified_files: usize,
    pub unchanged_files: usize,
    pub deleted_files: usize,
    pub total_scopes: usize,
    pub total_symbols: usize,
    pub total_references: usize,
    pub total_edges: usize,
    pub dependencies_indexed: usize,
}

/// Central engine service coordinating scanning, incremental hashing, parallel AST parsing, Stack Graph resolution, and Turso persistence.
#[derive(Clone)]
pub struct EngineService {
    db: ScryerDb,
    rayon_pool: Arc<rayon::ThreadPool>,
    engine_pool: Arc<ProjectEnginePool>,
    /// Serializes index writes per project so watchers, tools and CLI runs never overlap.
    index_locks: Arc<Mutex<HashMap<u64, Arc<tokio::sync::Mutex<()>>>>>,
}

impl EngineService {
    /// Create a new `EngineService` with a default Rayon thread pool matching CPU core count and 16-project engine pool.
    pub fn new(db: ScryerDb) -> Self {
        let pool = rayon::ThreadPoolBuilder::new()
            .build()
            .expect("Failed to initialize Rayon thread pool");
        let engine_pool =
            Arc::new(ProjectEnginePool::new(16).expect("Failed to initialize ProjectEnginePool"));

        Self {
            db,
            rayon_pool: Arc::new(pool),
            engine_pool,
            index_locks: Arc::default(),
        }
    }

    /// Create an `EngineService` with a custom number of Rayon worker threads.
    pub fn with_thread_count(db: ScryerDb, thread_count: usize) -> Self {
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(thread_count)
            .build()
            .expect("Failed to initialize custom Rayon thread pool");
        let engine_pool =
            Arc::new(ProjectEnginePool::new(16).expect("Failed to initialize ProjectEnginePool"));

        Self {
            db,
            rayon_pool: Arc::new(pool),
            engine_pool,
            index_locks: Arc::default(),
        }
    }

    /// Access the underlying database handle.
    pub fn db(&self) -> &ScryerDb {
        &self.db
    }

    /// Access the underlying `ProjectEnginePool`.
    pub fn pool(&self) -> &Arc<ProjectEnginePool> {
        &self.engine_pool
    }

    /// Acquire the per-project index write guard.
    async fn index_guard(&self, project_id: u64) -> tokio::sync::OwnedMutexGuard<()> {
        let lock = {
            let mut locks = self.index_locks.lock().unwrap_or_else(|e| e.into_inner());
            Arc::clone(locks.entry(project_id).or_default())
        };
        lock.lock_owned().await
    }

    /// Fail if `project_id` is no longer registered, so a watcher or tool racing a project
    /// removal can't re-insert rows for it. Call while holding the index guard.
    async fn ensure_project_exists(&self, project_id: u64) -> anyhow::Result<()> {
        let mut guard = self.db.lock().await;
        let project = Project::filter(Project::fields().id().eq(project_id))
            .first()
            .exec(&mut *guard)
            .await?;
        anyhow::ensure!(project.is_some(), "Project {project_id} is not registered");
        Ok(())
    }

    /// Record that the project's index changed by bumping `project.updated_at`.
    async fn touch_project(&self, project_id: u64) -> anyhow::Result<()> {
        let stmt = format!(
            "UPDATE project SET updated_at = '{}' WHERE id = {project_id};",
            scryer_db::time::now_rfc3339()
        );
        let mut guard = self.db.lock().await;
        toasty::sql::statement(&stmt).exec(&mut *guard).await?;
        Ok(())
    }

    /// Delete every index row (files, scopes, symbols, references, edges) for a project and
    /// drop its in-memory stack graph.
    ///
    /// Waits for any in-flight indexing of the project to finish first.
    pub async fn purge_project(&self, project_id: u64) -> anyhow::Result<()> {
        let _guard = self.index_guard(project_id).await;
        {
            let mut db_guard = self.db.lock().await;
            let mut tx = db_guard.transaction().await?;
            for table in [
                "symbol_reference",
                "code_graph_edge",
                "symbol",
                "scope",
                "source_file",
            ] {
                let stmt = format!("DELETE FROM {table} WHERE project_id = {project_id};");
                toasty::sql::statement(&stmt).exec(&mut tx).await?;
            }
            tx.commit().await?;
        }
        self.engine_pool.evict(project_id).await;
        Ok(())
    }

    /// Index an entire workspace project incrementally:
    /// 1. Walks files honoring `.gitignore`.
    /// 2. Hashes contents via BLAKE3 and compares to Turso cache.
    /// 3. Parses modified/added files in parallel across Rayon workers.
    /// 4. Executes Stack Graph scope resolution and extracts references and call graph edges.
    /// 5. Streams parsed payloads through bounded channel into Tokio writer actor.
    /// 6. Cleans up deleted files with cascading deletes.
    pub async fn index_project(&self, project_id: u64, root: &Path) -> anyhow::Result<IndexReport> {
        let _guard = self.index_guard(project_id).await;
        self.ensure_project_exists(project_id).await?;
        let canonical_root = dunce::canonicalize(root).unwrap_or_else(|_| root.to_path_buf());
        let scanner = WorkspaceScanner::new(&canonical_root);
        let scanned = scanner.scan();

        let change_report =
            ChangeDetector::detect_changes(&self.db, project_id, &canonical_root, &scanned).await?;

        let mut report = IndexReport {
            scanned_files: scanned.len(),
            added_files: change_report.added_count,
            modified_files: change_report.modified_count,
            unchanged_files: change_report.unchanged_count,
            deleted_files: change_report.deleted_count,
            total_scopes: 0,
            total_symbols: 0,
            total_references: 0,
            total_edges: 0,
            dependencies_indexed: 0,
        };

        if change_report.is_empty() {
            if canonical_root.join("Cargo.toml").exists() {
                match self.index_dependencies(project_id, &canonical_root).await {
                    Ok(deps_indexed) => {
                        report.dependencies_indexed = deps_indexed;
                    }
                    Err(err) => {
                        tracing::warn!(
                            "Failed to index dependencies for project {project_id}: {err}"
                        );
                    }
                }
            }
            return Ok(report);
        }

        // Spawn Tokio batch ingestion actor
        let ingestion_handle = BatchIngestionActor::spawn(self.db.clone(), project_id, 4096);
        let sender = ingestion_handle.sender();

        // Collect files to parse and files to delete
        let mut files_to_parse = Vec::new();
        let mut file_statuses = HashMap::new();
        let mut deleted_paths = Vec::new();

        for change in change_report.changes {
            match change {
                FileChange::Added {
                    rel_path,
                    content_hash,
                    content,
                } => {
                    file_statuses.insert(rel_path.clone(), ExistingFileStatus::New);
                    files_to_parse.push((rel_path, content_hash, content));
                }
                FileChange::Modified {
                    file_id,
                    rel_path,
                    content_hash,
                    content,
                } => {
                    file_statuses.insert(rel_path.clone(), ExistingFileStatus::Existing(file_id));
                    files_to_parse.push((rel_path, content_hash, content));
                }
                FileChange::Deleted { rel_path, .. } => {
                    deleted_paths.push(rel_path.clone());
                    let _ = sender.send(IngestionMessage::DeleteFile(rel_path)).await;
                }
                FileChange::Unchanged { .. } => {}
            }
        }

        // 1. Parallel AST parsing across Rayon workers
        let pool = Arc::clone(&self.rayon_pool);
        let files_to_parse = Arc::new(files_to_parse);
        let files_for_parse = Arc::clone(&files_to_parse);
        let parsed_files: Vec<anyhow::Result<ParsedFilePayload>> =
            tokio::task::spawn_blocking(move || {
                pool.install(|| {
                    files_for_parse
                        .par_iter()
                        .map(|(rel_path, hash, content)| {
                            parse_file(rel_path, content, hash, ParseMode::Workspace)
                        })
                        .collect()
                })
            })
            .await?;

        let mut payloads = Vec::new();
        for res in parsed_files {
            payloads.push(res?);
        }

        // 2. Prepare structures for cross-file reference resolution
        let engine_lock = self.engine_pool.get_or_create(project_id).await;
        if !deleted_paths.is_empty() {
            let mut engine = engine_lock.lock().await;
            for path in &deleted_paths {
                engine.remove_file(path);
            }
        }
        let mut file_sources = Vec::new();
        let mut all_symbols = HashMap::new();
        let mut all_scopes = HashMap::new();

        let content_map: HashMap<&Path, &[u8]> = files_to_parse
            .iter()
            .map(|(p, _, c)| (p.as_path(), c.as_slice()))
            .collect();

        for p in &payloads {
            if let Some(bytes) = content_map.get(p.relative_path.as_path()) {
                if let Ok(src) = std::str::from_utf8(bytes) {
                    file_sources.push((p.relative_path.clone(), src.to_string()));
                }
            } else {
                let full_path = canonical_root.join(&p.relative_path);
                if let Ok(src) = fs::read_to_string(&full_path) {
                    file_sources.push((p.relative_path.clone(), src));
                }
            }
            all_symbols.insert(p.relative_path.clone(), p.symbols.clone());
            all_scopes.insert(p.relative_path.clone(), p.scopes.clone());
        }

        // 3. Resolve cross-file references and call graph edges
        let extracted_map = CrossFileResolver::resolve_project(
            Arc::clone(&engine_lock),
            &file_sources,
            &all_symbols,
            &all_scopes,
        )
        .await?;

        // 4. Attach references and edges to payloads and send to ingestion writer actor
        let mut total_scopes = 0;
        let mut total_symbols = 0;
        let mut total_references = 0;
        let mut total_edges = 0;

        for mut payload in payloads {
            total_scopes += payload.scopes.len();
            total_symbols += payload.symbols.len();

            if let Some(extracted) = extracted_map.get(&payload.relative_path) {
                total_references += extracted.references.len();
                total_edges += extracted.edges.len();
                payload.references = extracted.references.clone();
                payload.edges = extracted.edges.clone();
            }

            let status = file_statuses
                .remove(&payload.relative_path)
                .unwrap_or(ExistingFileStatus::Unknown);

            sender
                .send(IngestionMessage::UpsertFile {
                    payload,
                    status,
                    dependency_package_id: None,
                })
                .await
                .map_err(|e| anyhow::anyhow!("Writer channel closed: {e}"))?;
        }

        report.total_scopes = total_scopes;
        report.total_symbols = total_symbols;
        report.total_references = total_references;
        report.total_edges = total_edges;

        drop(sender);

        // Await completion of writer actor
        ingestion_handle.finish().await?;

        // 5. Index dependencies if Cargo manifest is present
        if canonical_root.join("Cargo.toml").exists() {
            match self.index_dependencies(project_id, &canonical_root).await {
                Ok(deps_indexed) => {
                    report.dependencies_indexed = deps_indexed;
                }
                Err(err) => {
                    tracing::warn!("Failed to index dependencies for project {project_id}: {err}");
                }
            }
        }

        self.touch_project(project_id).await?;

        Ok(report)
    }

    /// Incrementally index or re-index a single file.
    pub async fn index_file(
        &self,
        project_id: u64,
        root: &Path,
        rel_path: &Path,
    ) -> anyhow::Result<()> {
        let _guard = self.index_guard(project_id).await;
        self.ensure_project_exists(project_id).await?;
        let full_path = root.join(rel_path);
        let content = fs::read(&full_path)?;
        let content_hash = crate::hasher::hash_bytes(&content);

        let mut parsed = parse_file(rel_path, &content, &content_hash, ParseMode::Workspace)?;

        if let Ok(src) = std::str::from_utf8(&content) {
            let engine_lock = self.engine_pool.get_or_create(project_id).await;
            let mut all_symbols = HashMap::new();
            let mut all_scopes = HashMap::new();
            all_symbols.insert(rel_path.to_path_buf(), parsed.symbols.clone());
            all_scopes.insert(rel_path.to_path_buf(), parsed.scopes.clone());

            if let Ok(extracted) = CrossFileResolver::resolve_file_references(
                Arc::clone(&engine_lock),
                rel_path,
                src,
                &all_symbols,
                &all_scopes,
            )
            .await
            {
                parsed.references = extracted.references;
                parsed.edges = extracted.edges;
            }
        }

        let actor = BatchIngestionActor::new(self.db.clone(), project_id);
        actor.ingest_payload(parsed).await?;
        self.touch_project(project_id).await?;

        Ok(())
    }

    /// Transactionally purge records for a deleted file.
    pub async fn remove_file(&self, project_id: u64, rel_path: &Path) -> anyhow::Result<()> {
        let _guard = self.index_guard(project_id).await;
        let actor = BatchIngestionActor::new(self.db.clone(), project_id);
        actor.delete_file(rel_path).await?;
        self.touch_project(project_id).await?;
        let engine_lock = self.engine_pool.get_or_create(project_id).await;
        engine_lock.lock().await.remove_file(rel_path);
        Ok(())
    }

    /// Resolve a definition at a specific `(line, col)` location in `rel_path`.
    pub async fn resolve_definition(
        &self,
        project_id: u64,
        root: &Path,
        rel_path: &Path,
        line: u32,
        col: u32,
    ) -> anyhow::Result<Option<ResolvedDefinition>> {
        let engine_lock = self.engine_pool.get_or_create(project_id).await;
        let mut engine = engine_lock.lock().await;

        if !engine.files.contains_key(rel_path) {
            let full_path = root.join(rel_path);
            if let Ok(content) = fs::read_to_string(&full_path)
                && let Ok(file) = engine.add_file(rel_path, &content)
            {
                let cancellation = StepBoundedCancellationFlag::new(engine.max_steps);
                let _ = engine.precompute_file_paths(file, &cancellation);
            }
        }

        let resolved = engine.resolve_at_location(rel_path, line, col)?;
        if let Some(mut def) = resolved {
            if def.is_external || def.file_path.is_absolute() {
                def.is_external = true;
                if def.crate_name.is_none()
                    && let Some(parent) = def.file_path.parent()
                {
                    let name_candidate = parent
                        .file_name()
                        .and_then(|s| s.to_str())
                        .unwrap_or_default();
                    let clean = name_candidate.split('-').next().unwrap_or(name_candidate);
                    if !clean.is_empty() {
                        def.crate_name = Some(clean.to_string());
                    }
                }
                let mut guard = self.db.lock().await;
                let syms = scryer_db::Symbol::filter(
                    scryer_db::Symbol::fields()
                        .project_id()
                        .eq(0)
                        .and(scryer_db::Symbol::fields().name().eq(&def.symbol_name)),
                )
                .exec(&mut *guard)
                .await?;
                if let Some(s) = syms
                    .into_iter()
                    .find(|s| s.start_line <= def.start_line && def.start_line <= s.end_line)
                    && def.syntax_type.is_none()
                {
                    def.syntax_type = Some(s.signature);
                }
            }
            return Ok(Some(def));
        }

        Ok(None)
    }

    /// Index external dependencies for a workspace project using Cargo metadata.
    ///
    /// 1. Discovers workspace dependencies via [`CargoDependencyProvider`].
    /// 2. For each dependency:
    ///    - Checks if `(name, version, package_hash)` already exists in `DependencyPackage`.
    ///    - If found (cache hit): links `ProjectDependency` (O(1)) and skips parsing.
    ///    - If missing (cache miss): creates `DependencyPackage`, scans `.rs` files under crate root/src,
    ///      parses public surface with [`ParseMode::DependencyPublicSurface`], streams records via
    ///      [`BatchIngestionActor`] with `project_id = 0` and `dependency_package_id = Some(id)`,
    ///      and inserts `ProjectDependency` linking record.
    pub async fn index_dependencies(&self, project_id: u64, root: &Path) -> anyhow::Result<usize> {
        let manifest_path = if root.is_file() {
            root.to_path_buf()
        } else {
            root.join("Cargo.toml")
        };

        if !manifest_path.exists() {
            return Ok(0);
        }

        let provider = CargoDependencyProvider::new();
        let discovered = match provider.discover_dependencies(root).await {
            Ok(deps) => deps,
            Err(err) => {
                tracing::warn!(
                    "Failed to discover cargo dependencies for {}: {err}",
                    root.display()
                );
                return Ok(0);
            }
        };

        if discovered.is_empty() {
            return Ok(0);
        }

        let direct_deps: Vec<_> = discovered.into_iter().filter(|d| d.is_direct).collect();
        if direct_deps.is_empty() {
            return Ok(0);
        }

        // Phase 1: Check existing packages and register new DependencyPackages & ProjectDependency links
        let mut to_index = Vec::new();
        let mut existing = Vec::new();

        {
            let mut db_guard = self.db.lock().await;
            for dep in direct_deps {
                let pkg_hash = dep
                    .package_hash
                    .clone()
                    .unwrap_or_else(|| format!("{}-{}", dep.name, dep.version));

                let existing_pkg = DependencyPackage::filter(
                    DependencyPackage::fields()
                        .name()
                        .eq(&dep.name)
                        .and(DependencyPackage::fields().version().eq(&dep.version))
                        .and(DependencyPackage::fields().package_hash().eq(&pkg_hash)),
                )
                .first()
                .exec(&mut *db_guard)
                .await?;

                let (package_id, is_new) = match existing_pkg {
                    Some(pkg) => (pkg.id, false),
                    None => {
                        let source_type_str = match &dep.source_type {
                            DependencySourceType::CratesIo => "cratesio",
                            DependencySourceType::Git => "git",
                            DependencySourceType::Path => "path",
                            DependencySourceType::Unknown(s) => s.as_str(),
                        };

                        let new_pkg = DependencyPackage::create()
                            .name(dep.name.clone())
                            .version(dep.version.clone())
                            .source_type(source_type_str.to_string())
                            .package_hash(pkg_hash)
                            .root_path(dep.root_path.to_string_lossy().to_string())
                            .manifest_path(dep.manifest_path.to_string_lossy().to_string())
                            .indexed_at(scryer_db::time::now_rfc3339())
                            .exec(&mut *db_guard)
                            .await?;

                        (new_pkg.id, true)
                    }
                };

                let existing_link = ProjectDependency::filter(
                    ProjectDependency::fields().project_id().eq(project_id).and(
                        ProjectDependency::fields()
                            .dependency_package_id()
                            .eq(package_id),
                    ),
                )
                .first()
                .exec(&mut *db_guard)
                .await?;

                if existing_link.is_none() {
                    ProjectDependency::create()
                        .project_id(project_id)
                        .dependency_package_id(package_id)
                        .is_direct(dep.is_direct)
                        .features(dep.features.join(","))
                        .exec(&mut *db_guard)
                        .await?;
                }

                let source_dir = provider
                    .resolve_source_directory(&dep)
                    .unwrap_or_else(|_| dep.root_path.clone());

                let entry_file = if source_dir.join("src/lib.rs").is_file() {
                    source_dir.join("src/lib.rs")
                } else if source_dir.join("lib.rs").is_file() {
                    source_dir.join("lib.rs")
                } else {
                    source_dir.clone()
                };

                if is_new {
                    to_index.push((dep, package_id, source_dir, entry_file));
                } else {
                    existing.push((dep, package_id, source_dir, entry_file));
                }
            }
        }

        let engine_lock = self.engine_pool.get_or_create(project_id).await;

        // Apply existing dependencies to engine and cache
        for (dep, package_id, _source_dir, entry_file) in existing {
            let mut cached_files = Vec::new();
            {
                let mut engine = engine_lock.lock().await;
                let _ = engine.register_crate_root(&dep.name, &entry_file);
                if entry_file.is_file()
                    && let Ok(src) = fs::read_to_string(&entry_file)
                {
                    if let Ok(h) = engine.add_file(&entry_file, &src) {
                        engine.dependency_files.insert(h);
                        engine.file_to_crate.insert(h, dep.name.clone());
                    }
                    cached_files.push((entry_file.clone(), src));
                }
            }

            self.engine_pool
                .cache_dependency(CachedDependencyPackage {
                    package_id,
                    crate_name: dep.name.clone(),
                    entry_file: entry_file.clone(),
                    files: cached_files,
                    paths_count: 0,
                })
                .await;
        }

        if to_index.is_empty() {
            return Ok(0);
        }

        // Phase 2: Parallel walk, parse, and batched streaming ingestion for new dependency crates
        let ingestion_handle = BatchIngestionActor::spawn(self.db.clone(), 0, 2000);
        let sender = ingestion_handle.sender();

        let concurrency = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4)
            .clamp(2, 8);
        let semaphore = Arc::new(tokio::sync::Semaphore::new(concurrency));
        let mut join_set = tokio::task::JoinSet::new();

        for (dep, package_id, source_dir, entry_file) in to_index {
            let permit = Arc::clone(&semaphore)
                .acquire_owned()
                .await
                .map_err(|e| anyhow::anyhow!("Semaphore closed: {e}"))?;
            let sender = sender.clone();
            let rayon_pool = Arc::clone(&self.rayon_pool);
            let engine_lock = Arc::clone(&engine_lock);
            let engine_pool = Arc::clone(&self.engine_pool);

            join_set.spawn(async move {
                let _permit = permit;
                let target_dir = if source_dir.join("src").is_dir() {
                    source_dir.join("src")
                } else {
                    source_dir.clone()
                };

                let mut files_to_parse = Vec::new();
                for entry in ignore::WalkBuilder::new(&target_dir)
                    .hidden(false)
                    .parents(false)
                    .build()
                    .flatten()
                {
                    let path = entry.path();
                    if path.is_file()
                        && path.extension().and_then(|s| s.to_str()) == Some("rs")
                        && let Ok(rel_path) = path.strip_prefix(&source_dir)
                        && let Ok(content) = fs::read(path)
                    {
                        let hash = crate::hasher::hash_bytes(&content);
                        files_to_parse.push((rel_path.to_path_buf(), hash, content));
                    }
                }

                if !files_to_parse.is_empty() {
                    let crate_name = dep.name.clone();
                    let parsed_files: Vec<anyhow::Result<ParsedFilePayload>> =
                        tokio::task::spawn_blocking(move || {
                            rayon_pool.install(|| {
                                files_to_parse
                                    .into_par_iter()
                                    .map(|(rel_path, hash, content)| {
                                        crate::parsers::rust::RustAstParser::parse(
                                            &rel_path,
                                            &content,
                                            &hash,
                                            ParseMode::DependencyPublicSurface {
                                                crate_name: crate_name.clone(),
                                            },
                                        )
                                    })
                                    .collect()
                            })
                        })
                        .await
                        .map_err(|e| anyhow::anyhow!("Join error in parse task: {e}"))?;

                    for payload in parsed_files.into_iter().flatten() {
                        sender
                            .send(IngestionMessage::UpsertFile {
                                payload,
                                status: ExistingFileStatus::New,
                                dependency_package_id: Some(package_id),
                            })
                            .await
                            .map_err(|e| anyhow::anyhow!("Writer channel closed: {e}"))?;
                    }
                }

                // Add entry file to engine and cache
                let mut cached_files = Vec::new();
                {
                    let mut engine = engine_lock.lock().await;
                    let _ = engine.register_crate_root(&dep.name, &entry_file);
                    if entry_file.is_file()
                        && let Ok(src) = fs::read_to_string(&entry_file)
                    {
                        if let Ok(h) = engine.add_file(&entry_file, &src) {
                            engine.dependency_files.insert(h);
                            engine.file_to_crate.insert(h, dep.name.clone());
                        }
                        cached_files.push((entry_file.clone(), src));
                    }
                }

                engine_pool
                    .cache_dependency(CachedDependencyPackage {
                        package_id,
                        crate_name: dep.name.clone(),
                        entry_file: entry_file.clone(),
                        files: cached_files,
                        paths_count: 0,
                    })
                    .await;

                Ok::<(), anyhow::Error>(())
            });
        }

        let mut indexed_count = 0;
        while let Some(res) = join_set.join_next().await {
            res.map_err(|e| anyhow::anyhow!("Join set error: {e}"))??;
            indexed_count += 1;
        }

        drop(sender);
        ingestion_handle.finish().await?;

        Ok(indexed_count)
    }

    /// Retrieve type contract for a struct, enum, trait, or type alias across workspace and dependencies.
    pub async fn get_type_contract(
        &self,
        project_id: u64,
        root: &Path,
        type_name: &str,
        file_path_hint: Option<&str>,
        member_query: Option<&str>,
    ) -> anyhow::Result<RawTypeContract> {
        let mut guard = self.db.lock().await;

        // 1. Try finding matching type symbol in workspace project
        let symbols =
            scryer_db::Symbol::filter(scryer_db::Symbol::fields().project_id().eq(project_id))
                .exec(&mut *guard)
                .await?;

        let matching_type = symbols.iter().find(|s| {
            if let Some(hint) = file_path_hint {
                // If hint provided, can narrow file if needed
                let _ = hint;
            }
            (s.name == type_name || s.qualified_name == type_name)
                && matches!(
                    s.kind.as_str(),
                    "struct" | "enum" | "trait" | "type" | "class" | "interface"
                )
        });

        // 2. If not found in workspace, search external dependencies linked to project
        let (type_sym, is_external, dep_root) = if let Some(sym) = matching_type {
            (sym.clone(), false, None)
        } else {
            let dep_links = scryer_db::ProjectDependency::filter(
                scryer_db::ProjectDependency::fields()
                    .project_id()
                    .eq(project_id),
            )
            .exec(&mut *guard)
            .await?;
            let dep_pkg_ids: Vec<u64> = dep_links
                .into_iter()
                .map(|l| l.dependency_package_id)
                .collect();

            let dep_symbols =
                scryer_db::Symbol::filter(scryer_db::Symbol::fields().project_id().eq(0))
                    .exec(&mut *guard)
                    .await?;

            let target_name = type_name.rsplit("::").next().unwrap_or(type_name);
            let found_dep_sym = dep_symbols.into_iter().find(|s| {
                let matches_name = if type_name.contains("::") {
                    s.qualified_name == type_name
                        || s.qualified_name.ends_with(&format!("::{type_name}"))
                } else {
                    s.name == target_name
                };
                matches_name
                    && matches!(
                        s.kind.as_str(),
                        "struct" | "enum" | "trait" | "type" | "class" | "interface"
                    )
                    && s.dependency_package_id
                        .is_some_and(|id| dep_pkg_ids.contains(&id))
            });

            if let Some(sym) = found_dep_sym {
                let pkg_id = sym.dependency_package_id.unwrap();
                let pkg = scryer_db::DependencyPackage::filter(
                    scryer_db::DependencyPackage::fields().id().eq(pkg_id),
                )
                .first()
                .exec(&mut *guard)
                .await?;
                (sym, true, pkg.map(|p| PathBuf::from(p.root_path)))
            } else {
                return Ok(RawTypeContract {
                    found: false,
                    type_name: type_name.to_string(),
                    kind: None,
                    signature: None,
                    docstring: None,
                    file_path: None,
                    members: Vec::new(),
                });
            }
        };

        // Resolve declaring source file
        let file_proj_id = if is_external { 0 } else { project_id };
        let file = scryer_db::SourceFile::filter(
            scryer_db::SourceFile::fields()
                .project_id()
                .eq(file_proj_id)
                .and(scryer_db::SourceFile::fields().id().eq(type_sym.file_id)),
        )
        .first()
        .exec(&mut *guard)
        .await?;
        let file_rel_path = file.map(|f| f.path);

        let mut members = Vec::new();
        let mut final_file_path = None;

        if let Some(rel) = &file_rel_path {
            let full_disk_path = if is_external && let Some(dep_dir) = &dep_root {
                dep_dir.join(rel)
            } else {
                root.join(rel)
            };
            final_file_path = Some(full_disk_path.to_string_lossy().to_string());

            if let Ok(source) = fs::read_to_string(&full_disk_path) {
                members = crate::parsers::rust::extract_type_members(&source, &type_sym.name);
            }

            // Extract associated methods from `impl` blocks in the same crate/project
            let pool_symbols = if is_external {
                scryer_db::Symbol::filter(
                    scryer_db::Symbol::fields().project_id().eq(0).and(
                        scryer_db::Symbol::fields()
                            .dependency_package_id()
                            .eq(type_sym.dependency_package_id),
                    ),
                )
                .exec(&mut *guard)
                .await?
            } else {
                symbols
            };

            let crate_root = type_sym.qualified_name.split("::").next();
            for s in &pool_symbols {
                let Some((parent, _)) = s.qualified_name.rsplit_once("::") else {
                    continue;
                };
                let owned_by_type = parent == type_sym.qualified_name
                    || (parent.rsplit("::").next() == Some(type_sym.name.as_str())
                        && s.qualified_name.split("::").next() == crate_root);
                if s.kind == "fn" && owned_by_type {
                    let entry = format!("method: {}", s.signature);
                    if !members.contains(&entry) {
                        members.push(entry);
                    }
                }
            }
        }

        if let Some(mq) = member_query {
            let q_lower = mq.to_lowercase();
            members.retain(|m| m.to_lowercase().contains(&q_lower));
        }

        Ok(RawTypeContract {
            found: true,
            type_name: type_sym.name,
            kind: Some(type_sym.kind),
            signature: Some(type_sym.signature),
            docstring: type_sym.docstring,
            file_path: final_file_path,
            members,
        })
    }

    /// Start a watcher (200ms debounce) with no projects; add them with
    /// [`WorkspaceWatcher::add_project`].
    pub fn watcher(&self) -> anyhow::Result<WorkspaceWatcher> {
        WorkspaceWatcher::start(self.clone(), Duration::from_millis(200))
    }

    /// Start a watcher (200ms debounce) for a single project.
    pub fn watch_project(&self, project_id: u64, root: &Path) -> anyhow::Result<WorkspaceWatcher> {
        let watcher = self.watcher()?;
        watcher.add_project(project_id, root)?;
        Ok(watcher)
    }
}