scryer-engine 0.1.0

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};

use scryer_db::{CodeGraphEdge, Scope, ScryerDb, SourceFile, Symbol, SymbolReference};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::payload::{ParsedFilePayload, RawCodeGraphEdge, RawSymbolReference};

/// Known persistence status of a source file before ingestion.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ExistingFileStatus {
    #[default]
    Unknown,
    New,
    Existing(u64),
}

/// Ingestion actions processed by the transactional writer task.
#[derive(Debug)]
pub enum IngestionMessage {
    UpsertFile {
        payload: ParsedFilePayload,
        status: ExistingFileStatus,
        dependency_package_id: Option<u64>,
    },
    DeleteFile(PathBuf),
}

/// Handle to a running ingestion background actor with a bounded channel.
pub struct IngestionHandle {
    sender: mpsc::Sender<IngestionMessage>,
    join_handle: JoinHandle<anyhow::Result<usize>>,
}

impl IngestionHandle {
    pub fn sender(&self) -> mpsc::Sender<IngestionMessage> {
        self.sender.clone()
    }

    /// Await the completion of the background ingestion writer task.
    pub async fn finish(self) -> anyhow::Result<usize> {
        drop(self.sender);
        self.join_handle.await?
    }
}

/// Transactional batch ingestion actor persisting parsed entities into Turso.
pub struct BatchIngestionActor {
    db: ScryerDb,
    project_id: u64,
}

impl BatchIngestionActor {
    pub fn new(db: ScryerDb, project_id: u64) -> Self {
        Self { db, project_id }
    }

    /// Spawn the writer actor as a Tokio background task with a bounded channel.
    pub fn spawn(db: ScryerDb, project_id: u64, buffer_capacity: usize) -> IngestionHandle {
        let (sender, mut receiver) = mpsc::channel::<IngestionMessage>(buffer_capacity);
        let actor = Self::new(db, project_id);

        let join_handle = tokio::spawn(async move {
            let mut processed = 0;
            let mut batch = Vec::new();
            while let Some(msg) = receiver.recv().await {
                batch.push(msg);
                while batch.len() < 500 {
                    match receiver.try_recv() {
                        Ok(m) => batch.push(m),
                        Err(_) => break,
                    }
                }
                let count = batch.len();
                actor.ingest_batch(std::mem::take(&mut batch)).await?;
                processed += count;
            }
            Ok(processed)
        });

        IngestionHandle {
            sender,
            join_handle,
        }
    }

    /// Transactionally process a batch of upsert and delete operations under a single commit.
    pub async fn ingest_batch(&self, messages: Vec<IngestionMessage>) -> anyhow::Result<()> {
        if messages.is_empty() {
            return Ok(());
        }
        let mut db_guard = self.db.lock().await;
        let mut tx = db_guard.transaction().await?;

        // Insert every file's symbols before linking any references, so references into
        // files later in the batch still resolve.
        let mut pending = Vec::new();
        for msg in messages {
            match msg {
                IngestionMessage::UpsertFile {
                    payload,
                    status,
                    dependency_package_id,
                } => {
                    pending.push(
                        self.insert_file_entities(&mut tx, payload, status, dependency_package_id)
                            .await?,
                    );
                }
                IngestionMessage::DeleteFile(rel_path) => {
                    self.delete_file_in_tx(&mut tx, &rel_path).await?;
                }
            }
        }

        let mut cache = HashMap::new();
        for links in pending {
            self.link_in_tx(&mut tx, links, &mut cache).await?;
        }

        tx.commit().await?;
        Ok(())
    }

    /// Ingest a single parsed file transactionally.
    pub async fn ingest_payload(&self, payload: ParsedFilePayload) -> anyhow::Result<()> {
        self.ingest_batch(vec![IngestionMessage::UpsertFile {
            payload,
            status: ExistingFileStatus::Unknown,
            dependency_package_id: None,
        }])
        .await
    }

    /// Ingest a single parsed file inside an active transaction.
    pub async fn ingest_payload_in_tx(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        payload: ParsedFilePayload,
        status: ExistingFileStatus,
    ) -> anyhow::Result<()> {
        let links = self.insert_file_entities(tx, payload, status, None).await?;
        self.link_in_tx(tx, links, &mut HashMap::new()).await
    }

    /// Upsert the file row and insert its scopes and symbols, re-pointing references and
    /// edges from other files at the new symbol IDs. Returns the file's references and
    /// edges for [`Self::link_in_tx`].
    async fn insert_file_entities(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        payload: ParsedFilePayload,
        status: ExistingFileStatus,
        dependency_package_id: Option<u64>,
    ) -> anyhow::Result<PendingLinks> {
        let rel_path_str = payload.relative_path.to_string_lossy().to_string();
        let effective_project_id = if dependency_package_id.is_some() {
            0
        } else {
            self.project_id
        };

        let existing = match status {
            ExistingFileStatus::New => None,
            ExistingFileStatus::Existing(existing_id) => SourceFile::filter(
                SourceFile::fields()
                    .project_id()
                    .eq(effective_project_id)
                    .and(SourceFile::fields().id().eq(existing_id)),
            )
            .first()
            .exec(&mut *tx)
            .await?
            .map(|f| (existing_id, Some(f)))
            // The row vanished but its children may not have: purge them anyway.
            .or(Some((existing_id, None))),
            ExistingFileStatus::Unknown => {
                if let Some(dep_pkg_id) = dependency_package_id {
                    SourceFile::filter(
                        SourceFile::fields()
                            .project_id()
                            .eq(0)
                            .and(
                                SourceFile::fields()
                                    .dependency_package_id()
                                    .eq(Some(dep_pkg_id)),
                            )
                            .and(SourceFile::fields().path().eq(&rel_path_str)),
                    )
                    .first()
                    .exec(&mut *tx)
                    .await?
                    .map(|f| (f.id, Some(f)))
                } else {
                    SourceFile::filter(
                        SourceFile::fields()
                            .project_id()
                            .eq(effective_project_id)
                            .and(SourceFile::fields().path().eq(&rel_path_str)),
                    )
                    .first()
                    .exec(&mut *tx)
                    .await?
                    .map(|f| (f.id, Some(f)))
                }
            }
        };

        let mut old_symbols = Vec::new();
        let file_id = match existing {
            Some((existing_id, file)) => {
                old_symbols = Symbol::filter(
                    Symbol::fields()
                        .project_id()
                        .eq(effective_project_id)
                        .and(Symbol::fields().file_id().eq(existing_id)),
                )
                .exec(&mut *tx)
                .await?;
                purge_file_records(
                    tx,
                    effective_project_id,
                    existing_id,
                    PurgeMode::KeepIncoming,
                )
                .await?;
                match file {
                    Some(mut file) => {
                        file.update()
                            .content_hash(payload.content_hash.clone())
                            .line_count(payload.line_count)
                            .byte_size(payload.byte_size)
                            .exec(&mut *tx)
                            .await?;
                        Some(file.id)
                    }
                    None => None,
                }
            }
            None => None,
        };
        let file_id = match file_id {
            Some(id) => id,
            None => {
                let mut builder = SourceFile::create();
                builder = builder
                    .project_id(effective_project_id)
                    .path(rel_path_str)
                    .content_hash(payload.content_hash.clone())
                    .language(payload.language.clone())
                    .line_count(payload.line_count)
                    .byte_size(payload.byte_size);

                if let Some(dep_pkg_id) = dependency_package_id {
                    builder = builder.dependency_package_id(dep_pkg_id);
                }

                builder.exec(&mut *tx).await?.id
            }
        };

        // 2. Insert Scope records and map local_id -> db_id
        let mut local_to_db_id = HashMap::new();
        for raw_scope in payload.scopes {
            let parent_db_id = raw_scope
                .parent_local_id
                .and_then(|p| local_to_db_id.get(&p).copied());

            let mut builder = Scope::create();
            builder = builder
                .project_id(effective_project_id)
                .file_id(file_id)
                .scope_kind(raw_scope.scope_kind)
                .start_byte(raw_scope.start_byte)
                .end_byte(raw_scope.end_byte)
                .start_line(raw_scope.start_line)
                .end_line(raw_scope.end_line);

            if let Some(p_id) = parent_db_id {
                builder = builder.parent_scope_id(p_id);
            }

            let inserted_scope = builder.exec(&mut *tx).await?;
            local_to_db_id.insert(raw_scope.local_id, inserted_scope.id);
        }

        // 3. Insert Symbol records referencing mapped scope_id
        let mut symbol_name_to_id = HashMap::new();
        for raw_symbol in payload.symbols {
            let scope_db_id = raw_symbol
                .scope_local_id
                .and_then(|s| local_to_db_id.get(&s).copied());

            let mut builder = Symbol::create();
            builder = builder
                .project_id(effective_project_id)
                .file_id(file_id)
                .name(raw_symbol.name.clone())
                .qualified_name(raw_symbol.qualified_name.clone())
                .kind(raw_symbol.kind)
                .visibility(raw_symbol.visibility)
                .signature(raw_symbol.signature)
                .start_byte(raw_symbol.start_byte)
                .end_byte(raw_symbol.end_byte)
                .start_line(raw_symbol.start_line)
                .end_line(raw_symbol.end_line);

            if let Some(dep_pkg_id) = dependency_package_id {
                builder = builder.dependency_package_id(dep_pkg_id);
            }

            if let Some(s_id) = scope_db_id {
                builder = builder.scope_id(s_id);
            }

            if let Some(doc) = raw_symbol.docstring {
                builder = builder.docstring(doc);
            }

            let inserted_sym = builder.exec(&mut *tx).await?;
            // Keep the first symbol for duplicate names, matching the database lookup order.
            symbol_name_to_id
                .entry(raw_symbol.qualified_name)
                .or_insert(inserted_sym.id);
            symbol_name_to_id
                .entry(raw_symbol.name)
                .or_insert(inserted_sym.id);
        }

        // 4. Re-point references and edges from other files at the replacement symbols,
        // dropping those whose target no longer exists.
        let p = effective_project_id;
        for old in old_symbols {
            let stmts = match symbol_name_to_id.get(&old.qualified_name) {
                Some(&new_id) => [
                    format!(
                        "UPDATE symbol_reference SET symbol_id = {new_id} WHERE project_id = {p} AND symbol_id = {};",
                        old.id
                    ),
                    format!(
                        "UPDATE code_graph_edge SET target_symbol_id = {new_id} WHERE project_id = {p} AND target_symbol_id = {};",
                        old.id
                    ),
                ],
                None => [
                    format!(
                        "DELETE FROM symbol_reference WHERE project_id = {p} AND symbol_id = {};",
                        old.id
                    ),
                    format!(
                        "DELETE FROM code_graph_edge WHERE project_id = {p} AND target_symbol_id = {};",
                        old.id
                    ),
                ],
            };
            for stmt in &stmts {
                toasty::sql::statement(stmt).exec(&mut *tx).await?;
            }
        }

        Ok(PendingLinks {
            file_id,
            local_symbols: symbol_name_to_id,
            references: payload.references,
            edges: payload.edges,
        })
    }

    /// Insert a file's references and call edges, resolving target names against the
    /// file's own symbols first and then the whole project.
    async fn link_in_tx(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        links: PendingLinks,
        cache: &mut HashMap<String, Option<u64>>,
    ) -> anyhow::Result<()> {
        let PendingLinks {
            file_id,
            local_symbols,
            references,
            edges,
        } = links;

        for raw_ref in references {
            let target = self
                .resolve_name(tx, &local_symbols, cache, &raw_ref.target_symbol_name)
                .await?;
            if let Some(sym_id) = target {
                SymbolReference::create()
                    .project_id(self.project_id)
                    .symbol_id(sym_id)
                    .file_id(file_id)
                    .role(raw_ref.role)
                    .start_byte(raw_ref.start_byte)
                    .end_byte(raw_ref.end_byte)
                    .line_number(raw_ref.line_number)
                    .exec(&mut *tx)
                    .await?;
            }
        }

        let mut seen = HashSet::new();
        for raw_edge in edges {
            let source = local_symbols.get(&raw_edge.source_symbol_name).copied();
            let target = self
                .resolve_name(tx, &local_symbols, cache, &raw_edge.target_symbol_name)
                .await?;
            if let (Some(src), Some(tgt)) = (source, target) {
                if !seen.insert((src, tgt, raw_edge.edge_type.clone())) {
                    continue;
                }
                CodeGraphEdge::create()
                    .project_id(self.project_id)
                    .source_symbol_id(src)
                    .target_symbol_id(tgt)
                    .edge_type(raw_edge.edge_type)
                    .exec(&mut *tx)
                    .await?;
            }
        }

        Ok(())
    }

    async fn resolve_name(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        local_symbols: &HashMap<String, u64>,
        cache: &mut HashMap<String, Option<u64>>,
        target: &str,
    ) -> anyhow::Result<Option<u64>> {
        if let Some(&id) = local_symbols.get(target) {
            return Ok(Some(id));
        }
        if let Some(&cached) = cache.get(target) {
            return Ok(cached);
        }
        let found = find_symbol_for_target(&mut *tx, self.project_id, target, None)
            .await?
            .map(|s| s.id);
        cache.insert(target.to_string(), found);
        Ok(found)
    }

    /// Transactionally delete a file and all its cascaded entities inside an active transaction.
    pub async fn delete_file_in_tx(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        rel_path: &Path,
    ) -> anyhow::Result<()> {
        let rel_path_str = rel_path.to_string_lossy().to_string();

        let existing_file = SourceFile::filter(
            SourceFile::fields()
                .project_id()
                .eq(self.project_id)
                .and(SourceFile::fields().path().eq(&rel_path_str)),
        )
        .first()
        .exec(&mut *tx)
        .await?;

        if let Some(file) = existing_file {
            purge_file_records(tx, self.project_id, file.id, PurgeMode::All).await?;

            let del_source_file = format!(
                "DELETE FROM source_file WHERE project_id = {} AND id = {};",
                self.project_id, file.id
            );
            toasty::sql::statement(&del_source_file)
                .exec(&mut *tx)
                .await?;
        }

        Ok(())
    }

    /// Transactionally delete a file and all its cascaded entities.
    pub async fn delete_file(&self, rel_path: &Path) -> anyhow::Result<()> {
        self.ingest_batch(vec![IngestionMessage::DeleteFile(rel_path.to_path_buf())])
            .await
    }
}

/// References and edges produced for a file, awaiting target resolution.
struct PendingLinks {
    file_id: u64,
    /// The file's own symbols by qualified and bare name.
    local_symbols: HashMap<String, u64>,
    references: Vec<RawSymbolReference>,
    edges: Vec<RawCodeGraphEdge>,
}

/// Whether purging a file also removes references and edges from other files into it.
#[derive(Clone, Copy, PartialEq, Eq)]
enum PurgeMode {
    /// The file is being deleted: drop everything pointing into it.
    All,
    /// The file is being re-indexed: keep incoming references and edges so they can be
    /// re-pointed at the replacement symbols.
    KeepIncoming,
}

/// Executes cascading SQL deletes to remove any existing child records for a file.
async fn purge_file_records(
    tx: &mut toasty::db::Transaction<'_>,
    project_id: u64,
    file_id: u64,
    mode: PurgeMode,
) -> anyhow::Result<()> {
    let file_symbols =
        format!("SELECT id FROM symbol WHERE project_id = {project_id} AND file_id = {file_id}");
    let mut statements = vec![
        format!(
            "DELETE FROM symbol_reference WHERE project_id = {project_id} AND file_id = {file_id};"
        ),
        format!(
            "DELETE FROM code_graph_edge WHERE project_id = {project_id} AND source_symbol_id IN ({file_symbols});"
        ),
    ];
    if mode == PurgeMode::All {
        statements.push(format!(
            "DELETE FROM symbol_reference WHERE project_id = {project_id} AND symbol_id IN ({file_symbols});"
        ));
        statements.push(format!(
            "DELETE FROM code_graph_edge WHERE project_id = {project_id} AND target_symbol_id IN ({file_symbols});"
        ));
    }
    statements.push(format!(
        "DELETE FROM symbol WHERE project_id = {project_id} AND file_id = {file_id};"
    ));
    statements.push(format!(
        "DELETE FROM scope WHERE project_id = {project_id} AND file_id = {file_id};"
    ));

    for stmt in &statements {
        toasty::sql::statement(stmt).exec(&mut *tx).await?;
    }

    Ok(())
}

/// Find the project symbol a reference target names.
///
/// Path targets (`a::b::f`) match a symbol whose qualified name equals the path or ends
/// with `::a::b::f`; bare names match by symbol name. Symbols in `prefer_file` win ties.
pub async fn find_symbol_for_target(
    executor: &mut dyn toasty::Executor,
    project_id: u64,
    target: &str,
    prefer_file: Option<u64>,
) -> anyhow::Result<Option<Symbol>> {
    let name = target.rsplit("::").next().unwrap_or(target);
    if name.is_empty() {
        return Ok(None);
    }
    let mut candidates = Symbol::filter(
        Symbol::fields()
            .project_id()
            .eq(project_id)
            .and(Symbol::fields().name().eq(name)),
    )
    .exec(executor)
    .await?;
    candidates.sort_by_key(|s| s.id);

    // If not found in current project, search external dependencies (project_id = 0)
    if candidates.is_empty() && project_id != 0 {
        candidates = Symbol::filter(
            Symbol::fields()
                .project_id()
                .eq(0)
                .and(Symbol::fields().name().eq(name)),
        )
        .exec(executor)
        .await?;
        candidates.sort_by_key(|s| s.id);
    }

    if target.contains("::") {
        let suffix = format!("::{target}");
        candidates.retain(|s| s.qualified_name == target || s.qualified_name.ends_with(&suffix));
        if let Some(pos) = candidates.iter().position(|s| s.qualified_name == target) {
            return Ok(Some(candidates.swap_remove(pos)));
        }
    }

    let preferred = prefer_file.and_then(|f| candidates.iter().position(|s| s.file_id == f));
    Ok(match preferred {
        Some(pos) => Some(candidates.swap_remove(pos)),
        None => candidates.into_iter().next(),
    })
}