scryer-engine 0.3.0

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

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

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

/// 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),
}

/// Messages one writer transaction takes by default.
pub const DEFAULT_WRITER_BATCH_CAP: usize = 500;

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

/// What a writer run stored, and the links it could not make yet.
#[derive(Debug, Default)]
pub struct IngestOutcome {
    /// Messages processed.
    pub processed: usize,
    pub inserted_references: usize,
    pub inserted_edges: usize,
    /// References and edges whose target no batch so far defined. Pass them to
    /// [`BatchIngestionActor::link_deferred`] once every file (and dependency) is stored.
    pub deferred: Vec<Deferred>,
    /// Edges the first pass stored, so a late link cannot duplicate one.
    pub stored_edges: HashSet<(u64, u64, String)>,
    /// Bare names of the symbols this run stored. A persisted unresolved reference to one of
    /// them may link now.
    pub new_names: HashSet<String>,
    /// Files this run stored or deleted.
    pub touched_files: HashSet<u64>,
}

/// A reference or edge awaiting its target.
#[derive(Debug, Clone)]
pub struct Deferred {
    /// The project the link belongs to: the referencing file's, which is not the writer's own
    /// project for a reference into a dependency that was just re-indexed.
    project_id: u64,
    file_id: u64,
    file_path: String,
    item: DeferredItem,
    origin: Origin,
}

/// Where a deferred link came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Origin {
    /// Extracted from a file stored in this run.
    Fresh,
    /// Read back from the `unresolved_reference` table: an earlier run could not link it.
    Revived,
    /// Pointed at a symbol that a re-indexed file no longer defines. It is dropped when its own
    /// file was stored in the same run, because that file's links were made again.
    Orphan,
}

#[derive(Debug, Clone)]
enum DeferredItem {
    Reference(RawSymbolReference),
    Edge {
        source_id: u64,
        edge: RawCodeGraphEdge,
    },
}

impl Deferred {
    fn target(&self) -> &str {
        match &self.item {
            DeferredItem::Reference(r) => &r.target_symbol_name,
            DeferredItem::Edge { edge, .. } => &edge.target_symbol_name,
        }
    }

    fn sort_key(&self) -> (&str, usize, &str) {
        let at = match &self.item {
            DeferredItem::Reference(r) => r.start_byte,
            DeferredItem::Edge { .. } => usize::MAX,
        };
        (&self.file_path, at, self.target())
    }
}

/// What [`BatchIngestionActor::link_deferred`] did.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct LinkResult {
    pub linked_references: usize,
    pub linked_edges: usize,
    /// Still no symbol of that name anywhere: external or unknown.
    pub unresolved_references: usize,
    pub unresolved_edges: usize,
}

/// Counts from linking one batch.
#[derive(Debug, Default, Clone, Copy)]
struct LinkCounts {
    references: usize,
    edges: usize,
    /// Edges whose calling symbol is not a stored symbol; they can never link.
    orphan_edges: usize,
}

impl IngestOutcome {
    /// Add one writer transaction's result.
    fn absorb(&mut self, result: BatchResult) {
        self.inserted_references += result.counts.references;
        self.inserted_edges += result.counts.edges;
        self.deferred.extend(result.deferred);
        self.stored_edges.extend(result.stored_edges);
        self.new_names.extend(result.new_names);
        self.touched_files.extend(result.touched_files);
    }
}

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<IngestOutcome> {
        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.
    ///
    /// `batch_cap` bounds how many queued messages one transaction takes (at least 1).
    pub fn spawn(
        db: ScryerDb,
        project_id: u64,
        buffer_capacity: usize,
        batch_cap: usize,
    ) -> IngestionHandle {
        let batch_cap = batch_cap.max(1);
        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 outcome = IngestOutcome::default();
            let mut batch = Vec::new();
            while let Some(msg) = receiver.recv().await {
                batch.push(msg);
                while batch.len() < batch_cap {
                    match receiver.try_recv() {
                        Ok(m) => batch.push(m),
                        Err(_) => break,
                    }
                }
                let count = batch.len();
                let result = actor.ingest_batch(std::mem::take(&mut batch)).await?;
                outcome.processed += count;
                outcome.absorb(result);
            }
            Ok(outcome)
        });

        IngestionHandle {
            sender,
            join_handle,
        }
    }

    /// Transactionally process a batch of upsert and delete operations under a single commit.
    async fn ingest_batch(&self, messages: Vec<IngestionMessage>) -> anyhow::Result<BatchResult> {
        if messages.is_empty() {
            return Ok(BatchResult::default());
        }
        let batch_timer = std::time::Instant::now();
        let n_messages = messages.len();
        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 result = BatchResult::default();
        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) => {
                    if let Some((file_id, orphans)) =
                        self.delete_file_in_tx(&mut tx, &rel_path).await?
                    {
                        result.touched_files.insert(file_id);
                        result.deferred.extend(orphans);
                    }
                }
            }
        }

        let insert_ms = batch_timer.elapsed().as_millis();
        for mut links in pending {
            result.new_names.extend(links.new_names.iter().cloned());
            result.touched_files.insert(links.file_id);
            result.deferred.append(&mut links.orphans);
            let counts = self
                .link_in_tx(
                    &mut tx,
                    links,
                    &mut result.deferred,
                    &mut result.stored_edges,
                )
                .await?;
            result.counts.references += counts.references;
            result.counts.edges += counts.edges;
            result.counts.orphan_edges += counts.orphan_edges;
        }
        let link_ms = batch_timer.elapsed().as_millis() - insert_ms;

        tx.commit().await?;
        tracing::info!(
            "ingest_batch project={} messages={n_messages}: entities={insert_ms}ms link={link_ms}ms commit={}ms",
            self.project_id,
            batch_timer.elapsed().as_millis() - insert_ms - link_ms
        );
        Ok(result)
    }

    /// Link the references and edges a writer run could not resolve, now that every file
    /// (and dependency) is stored, and link the ones an earlier run could not resolve to the
    /// symbols this run added. What stays unresolved names a symbol that does not exist in the
    /// index; for a project it is kept in `unresolved_reference` so a later run or edit that
    /// adds the symbol can link it, and counted.
    pub async fn link_deferred(&self, outcome: IngestOutcome) -> anyhow::Result<LinkResult> {
        let IngestOutcome {
            mut deferred,
            stored_edges,
            new_names,
            touched_files,
            ..
        } = outcome;
        let project_run = self.project_id != 0;
        let mut result = LinkResult::default();
        // A link to a symbol a re-indexed file dropped is stale when the file it comes from
        // was stored in the same run: that file's links were made again.
        deferred.retain(|d| d.origin != Origin::Orphan || !touched_files.contains(&d.file_id));
        if deferred.is_empty() && (!project_run || new_names.is_empty()) {
            if project_run {
                let mut db_guard = self.db.lock().await;
                result.unresolved_references =
                    count_unresolved(&mut *db_guard, self.project_id).await?;
            }
            return Ok(result);
        }

        let mut db_guard = self.db.lock().await;
        let mut tx = db_guard.transaction().await?;

        if project_run {
            deferred.extend(self.revive_unresolved(&mut tx, &new_names).await?);
        }
        deferred.sort_by(|a, b| a.sort_key().cmp(&b.sort_key()));

        let names: Vec<String> = deferred
            .iter()
            .map(|d| bare_name(d.target()).to_string())
            .filter(|n| !n.is_empty())
            .collect::<HashSet<_>>()
            .into_iter()
            .collect();
        let mut cache = CandidateCache::new();
        self.load_candidates(&mut tx, &names, &mut cache).await?;

        let mut seen = stored_edges;

        let mut ref_rows = Vec::new();
        let mut edge_rows = Vec::new();
        let mut unresolved_rows = Vec::new();
        for d in deferred {
            let picked = cache.get(bare_name(d.target())).and_then(|c| {
                pick_candidate(
                    c,
                    d.target(),
                    Some(d.file_id),
                    Some(Path::new(&d.file_path)),
                )
            });
            let target = picked.map(|(c, _)| c.id);
            match (d.item, target) {
                (DeferredItem::Reference(raw), Some(sym_id)) => {
                    // `exact` only when nothing else matched; otherwise it is a name match.
                    let unique = picked.is_some_and(|(_, unique)| unique);
                    let via = match raw.via {
                        Via::Exact if !unique => Via::Name,
                        via => via,
                    };
                    ref_rows.push(reference_row(d.project_id, sym_id, d.file_id, raw, via));
                    result.linked_references += 1;
                }
                (DeferredItem::Reference(raw), None) => {
                    result.unresolved_references += 1;
                    if d.project_id != 0 {
                        unresolved_rows.push(unresolved_row(d.project_id, d.file_id, raw.into()));
                    }
                }
                (DeferredItem::Edge { source_id, edge }, Some(tgt)) => {
                    // An edge from an earlier run may duplicate one that a different spelling
                    // of the same target already stored.
                    let stored = d.origin != Origin::Fresh
                        && edge_exists(&mut tx, source_id, tgt, &edge.edge_type).await?;
                    if !stored && seen.insert((source_id, tgt, edge.edge_type.clone())) {
                        edge_rows.push(
                            CodeGraphEdge::create()
                                .project_id(d.project_id)
                                .source_symbol_id(source_id)
                                .target_symbol_id(tgt)
                                .file_id(d.file_id)
                                .edge_type(edge.edge_type),
                        );
                        result.linked_edges += 1;
                    }
                }
                (DeferredItem::Edge { source_id, edge }, None) => {
                    result.unresolved_edges += 1;
                    if d.project_id != 0 {
                        unresolved_rows.push(unresolved_row(
                            d.project_id,
                            d.file_id,
                            Unresolved::edge(source_id, edge),
                        ));
                    }
                }
            }
        }
        insert_all(&mut tx, ref_rows, SymbolReference::create_many).await?;
        insert_all(&mut tx, edge_rows, CodeGraphEdge::create_many).await?;
        insert_all(&mut tx, unresolved_rows, UnresolvedReference::create_many).await?;
        if project_run {
            // The table is the whole project's, not only this run's.
            result.unresolved_references = count_unresolved(&mut tx, self.project_id).await?;
        }
        tx.commit().await?;
        Ok(result)
    }

    /// Take the persisted unresolved references whose target is named like one of `new_names`
    /// out of the table, as deferred links to try again.
    async fn revive_unresolved(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        new_names: &HashSet<String>,
    ) -> anyhow::Result<Vec<Deferred>> {
        if new_names.is_empty() {
            return Ok(Vec::new());
        }
        // One scan for the names that have rows, instead of a probe per new symbol.
        let rows = toasty::sql::query(
            "SELECT DISTINCT bare_name FROM unresolved_reference WHERE project_id = ?1",
        )
        .bind(self.project_id as i64)
        .exec(&mut *tx)
        .await?;
        let persisted: HashSet<String> = rows
            .iter()
            .filter_map(|row| match row {
                toasty::stmt::Value::Record(record) => record.fields.first()?.as_str(),
                other => other.as_str(),
            })
            .map(str::to_string)
            .collect();
        let mut wanted: Vec<&str> = new_names
            .iter()
            .map(String::as_str)
            .filter(|n| persisted.contains(*n))
            .collect();
        wanted.sort_unstable();

        let mut revived = Vec::new();
        for name in wanted {
            let rows = toasty::sql::query(
                "SELECT u.file_id, f.path, u.target_name, u.kind, u.role, u.start_byte, \
                 u.end_byte, u.line_number, u.via, u.source_symbol_id, u.edge_type \
                 FROM unresolved_reference u JOIN source_file f ON f.id = u.file_id \
                 WHERE u.bare_name = ?1 AND u.project_id = ?2",
            )
            .bind(name)
            .bind(self.project_id as i64)
            .exec(&mut *tx)
            .await?;
            for row in rows {
                let toasty::stmt::Value::Record(record) = row else {
                    anyhow::bail!("expected a record row, got {row:?}");
                };
                let [
                    file_id,
                    path,
                    target,
                    kind,
                    role,
                    start_byte,
                    end_byte,
                    line_number,
                    via,
                    source_id,
                    edge_type,
                ] = record.fields.as_slice()
                else {
                    anyhow::bail!("unexpected unresolved row shape: {record:?}");
                };
                let target = target.as_str().unwrap_or_default().to_string();
                let item = if kind.as_str() == Some("edge") {
                    DeferredItem::Edge {
                        source_id: value_u64(source_id)?,
                        edge: RawCodeGraphEdge {
                            source_symbol_name: String::new(),
                            target_symbol_name: target,
                            edge_type: edge_type.as_str().unwrap_or_default().to_string(),
                        },
                    }
                } else {
                    DeferredItem::Reference(RawSymbolReference {
                        target_symbol_name: target,
                        role: role.as_str().unwrap_or_default().to_string(),
                        start_byte: value_u64(start_byte)? as usize,
                        end_byte: value_u64(end_byte)? as usize,
                        line_number: u32::try_from(value_u64(line_number)?)?,
                        via: Via::parse(via.as_str().unwrap_or_default()),
                    })
                };
                revived.push(Deferred {
                    project_id: self.project_id,
                    file_id: value_u64(file_id)?,
                    file_path: path.as_str().unwrap_or_default().to_string(),
                    item,
                    origin: Origin::Revived,
                });
            }
            toasty::sql::statement(
                "DELETE FROM unresolved_reference WHERE bare_name = ?1 AND project_id = ?2",
            )
            .bind(name)
            .bind(self.project_id as i64)
            .exec(&mut *tx)
            .await?;
        }
        Ok(revived)
    }

    /// Ingest a single parsed file transactionally. The links it could not make yet come back
    /// in the outcome; pass it to [`Self::link_deferred`] to finish them.
    pub async fn ingest_payload(
        &self,
        payload: ParsedFilePayload,
    ) -> anyhow::Result<IngestOutcome> {
        self.ingest_one(IngestionMessage::UpsertFile {
            payload,
            status: ExistingFileStatus::Unknown,
            dependency_package_id: None,
        })
        .await
    }

    async fn ingest_one(&self, message: IngestionMessage) -> anyhow::Result<IngestOutcome> {
        let result = self.ingest_batch(vec![message]).await?;
        let mut outcome = IngestOutcome {
            processed: 1,
            ..IngestOutcome::default()
        };
        outcome.absorb(result);
        Ok(outcome)
    }

    /// 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 Vec::new(), &mut HashSet::new())
            .await
            .map(drop)
    }

    /// 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,
            // By id alone: this engine's planner scans for `project_id = ? AND id = ?`. Ids are
            // unique, so the project is checked on the row instead.
            ExistingFileStatus::Existing(existing_id) => {
                SourceFile::filter(SourceFile::fields().id().eq(existing_id))
                    .first()
                    .exec(&mut *tx)
                    .await?
                    .filter(|f| f.project_id == effective_project_id)
                    .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)) => {
                // By file id alone, as in `purge_file_records`.
                old_symbols = Symbol::filter_by_file_id(existing_id)
                    .exec(&mut *tx)
                    .await?;
                purge_file_records(tx, existing_id).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.clone())
                    .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 rows a depth level at a time, so each level can name the database
        // ids of its parents (parents always precede their children in the payload).
        let scopes = payload.scopes;
        let mut depth = Vec::with_capacity(scopes.len());
        let mut local_depth = HashMap::new();
        for raw_scope in &scopes {
            let d = raw_scope
                .parent_local_id
                .and_then(|p| local_depth.get(&p).copied())
                .map_or(0, |parent_depth| parent_depth + 1);
            local_depth.insert(raw_scope.local_id, d);
            depth.push(d);
        }
        let max_depth = depth.iter().copied().max().unwrap_or(0);
        let mut local_to_db_id: HashMap<_, u64> = HashMap::with_capacity(scopes.len());
        for level in 0..=max_depth {
            let level_scopes: Vec<_> = scopes
                .iter()
                .zip(&depth)
                .filter(|&(_, &d)| d == level)
                .map(|(scope, _)| scope)
                .collect();
            for chunk in level_scopes.chunks(INSERT_CHUNK) {
                let mut create = Scope::create_many();
                for raw_scope in chunk {
                    let mut builder = Scope::create()
                        .project_id(effective_project_id)
                        .file_id(file_id)
                        .scope_kind(raw_scope.scope_kind.clone())
                        .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(parent_id) = raw_scope
                        .parent_local_id
                        .and_then(|p| local_to_db_id.get(&p).copied())
                    {
                        builder = builder.parent_scope_id(parent_id);
                    }
                    create = create.item(builder);
                }
                let inserted = create.exec(&mut *tx).await?;
                anyhow::ensure!(
                    inserted.len() == chunk.len(),
                    "scope batch insert returned {} rows for {}",
                    inserted.len(),
                    chunk.len()
                );
                for (raw_scope, row) in chunk.iter().zip(inserted) {
                    anyhow::ensure!(
                        row.start_byte == raw_scope.start_byte
                            && row.end_byte == raw_scope.end_byte,
                        "scope batch insert returned rows out of order"
                    );
                    local_to_db_id.insert(raw_scope.local_id, row.id);
                }
            }
        }

        // 3. Symbols. One that continues an old symbol of the file keeps its id, so the
        // references and edges from other files into it stay valid and need no re-pointing;
        // the rest are inserted.
        let old_view: Vec<OldSymbol> = old_symbols
            .iter()
            .map(|s| OldSymbol {
                id: s.id,
                qualified_name: &s.qualified_name,
                kind: &s.kind,
                start_byte: s.start_byte,
            })
            .collect();
        let reuse = match_symbols(&old_view, &payload.symbols);
        let mut ids = vec![0u64; payload.symbols.len()];
        let fresh: Vec<usize> = (0..payload.symbols.len())
            .filter(|&i| reuse[i].is_none())
            .collect();
        for chunk in fresh.chunks(INSERT_CHUNK) {
            let mut create = Symbol::create_many();
            for &i in chunk {
                let raw_symbol = &payload.symbols[i];
                let mut builder = Symbol::create()
                    .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.clone())
                    .visibility(raw_symbol.visibility.clone())
                    .signature(raw_symbol.signature.clone())
                    .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(scope_id) = raw_symbol
                    .scope_local_id
                    .and_then(|s| local_to_db_id.get(&s).copied())
                {
                    builder = builder.scope_id(scope_id);
                }
                if let Some(doc) = &raw_symbol.docstring {
                    builder = builder.docstring(doc.clone());
                }
                create = create.item(builder);
            }
            let inserted = create.exec(&mut *tx).await?;
            anyhow::ensure!(
                inserted.len() == chunk.len(),
                "symbol batch insert returned {} rows for {}",
                inserted.len(),
                chunk.len()
            );
            for (&i, row) in chunk.iter().zip(inserted) {
                let raw_symbol = &payload.symbols[i];
                anyhow::ensure!(
                    row.qualified_name == raw_symbol.qualified_name
                        && row.start_byte == raw_symbol.start_byte,
                    "symbol batch insert returned rows out of order"
                );
                ids[i] = row.id;
            }
        }
        let kept: Vec<(u64, &RawSymbol)> = reuse
            .iter()
            .zip(&payload.symbols)
            .filter_map(|(old_id, raw_symbol)| Some((((*old_id)?), raw_symbol)))
            .collect();
        insert_kept_symbols(
            tx,
            SymbolOwner {
                project_id: effective_project_id,
                file_id,
                dependency_package_id,
            },
            &local_to_db_id,
            &kept,
        )
        .await?;
        for (id, old_id) in ids.iter_mut().zip(&reuse) {
            if let Some(old_id) = old_id {
                *id = *old_id;
            }
        }
        // Keep the first symbol for duplicate names, matching the database lookup order.
        let mut symbol_name_to_id = HashMap::new();
        for (raw_symbol, &id) in payload.symbols.iter().zip(&ids) {
            symbol_name_to_id
                .entry(raw_symbol.qualified_name.clone())
                .or_insert(id);
            symbol_name_to_id
                .entry(raw_symbol.name.clone())
                .or_insert(id);
        }

        // 4. What other files pointed at a symbol this file no longer defines goes back to the
        // deferred links: it may name a symbol elsewhere, or stay unresolved until one exists.
        let kept_ids: HashSet<u64> = reuse.iter().flatten().copied().collect();
        let removed: Vec<&Symbol> = old_symbols
            .iter()
            .filter(|s| !kept_ids.contains(&s.id))
            .collect();
        let orphans = take_incoming_links(tx, &removed).await?;

        Ok(PendingLinks {
            file_id,
            file_path: rel_path_str,
            new_names: fresh
                .iter()
                .map(|&i| payload.symbols[i].name.clone())
                .collect(),
            orphans,
            local_symbols: symbol_name_to_id,
            references: payload.references,
            edges: payload.edges,
        })
    }

    /// Insert the references and call edges of a file that target the file's own symbols.
    ///
    /// Anything else is pushed to `deferred`: which of several same-named symbols a target
    /// means depends on every file being stored, and a batch only sees the files stored so
    /// far. Resolving it here would make the result depend on batch boundaries. A same-file
    /// target is the one case that does not (it always wins, see [`pick_candidate`]).
    async fn link_in_tx(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        links: PendingLinks,
        deferred: &mut Vec<Deferred>,
        stored_edges: &mut HashSet<(u64, u64, String)>,
    ) -> anyhow::Result<LinkCounts> {
        let PendingLinks {
            file_id,
            file_path,
            new_names: _,
            orphans: _,
            local_symbols,
            references,
            edges,
        } = links;
        let mut counts = LinkCounts::default();

        let mut ref_rows = Vec::with_capacity(references.len());
        for raw_ref in references {
            match local_symbols.get(&raw_ref.target_symbol_name) {
                Some(&sym_id) => {
                    let via = raw_ref.via;
                    ref_rows.push(reference_row(
                        self.project_id,
                        sym_id,
                        file_id,
                        raw_ref,
                        via,
                    ))
                }
                None => deferred.push(Deferred {
                    project_id: self.project_id,
                    file_id,
                    file_path: file_path.clone(),
                    item: DeferredItem::Reference(raw_ref),
                    origin: Origin::Fresh,
                }),
            }
        }
        counts.references = ref_rows.len();
        insert_all(tx, ref_rows, SymbolReference::create_many).await?;

        let mut seen = HashSet::new();
        let mut edge_rows = Vec::with_capacity(edges.len());
        for raw_edge in edges {
            let Some(src) = local_symbols.get(&raw_edge.source_symbol_name).copied() else {
                counts.orphan_edges += 1;
                continue;
            };
            match local_symbols.get(&raw_edge.target_symbol_name) {
                Some(&tgt) => {
                    if !seen.insert((src, tgt, raw_edge.edge_type.clone())) {
                        continue;
                    }
                    edge_rows.push(
                        CodeGraphEdge::create()
                            .project_id(self.project_id)
                            .source_symbol_id(src)
                            .target_symbol_id(tgt)
                            .file_id(file_id)
                            .edge_type(raw_edge.edge_type),
                    );
                }
                None => deferred.push(Deferred {
                    project_id: self.project_id,
                    file_id,
                    file_path: file_path.clone(),
                    item: DeferredItem::Edge {
                        source_id: src,
                        edge: raw_edge,
                    },
                    origin: Origin::Fresh,
                }),
            }
        }
        counts.edges = edge_rows.len();
        insert_all(tx, edge_rows, CodeGraphEdge::create_many).await?;
        stored_edges.extend(seen);

        Ok(counts)
    }

    /// Fill `cache` with the symbols named `names`. Dependency symbols are only consulted for
    /// names the project does not define.
    async fn load_candidates(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        names: &[String],
        cache: &mut CandidateCache,
    ) -> anyhow::Result<()> {
        let mut by_name = fetch_candidates(&mut *tx, self.project_id, names).await?;
        if self.project_id != 0 {
            let missing: Vec<String> = names
                .iter()
                .filter(|n| !by_name.contains_key(*n))
                .cloned()
                .collect();
            by_name.extend(fetch_candidates(&mut *tx, 0, &missing).await?);
        }
        for name in names {
            cache
                .entry(name.clone())
                .or_insert_with(|| by_name.remove(name).unwrap_or_default());
        }
        Ok(())
    }

    /// Transactionally delete a file and all its cascaded entities inside an active
    /// transaction. Returns the deleted file's id and the links from other files into its
    /// symbols, which may name a symbol elsewhere (see [`Self::link_deferred`]).
    pub async fn delete_file_in_tx(
        &self,
        tx: &mut toasty::db::Transaction<'_>,
        rel_path: &Path,
    ) -> anyhow::Result<Option<(u64, Vec<Deferred>)>> {
        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?;

        let Some(file) = existing_file else {
            return Ok(None);
        };
        let symbols = Symbol::filter_by_file_id(file.id).exec(&mut *tx).await?;
        let orphans = take_incoming_links(tx, &symbols.iter().collect::<Vec<_>>()).await?;
        purge_file_records(tx, file.id).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(Some((file.id, orphans)))
    }

    /// Transactionally delete a file and all its cascaded entities. The outcome carries the
    /// links into the file's symbols for [`Self::link_deferred`].
    pub async fn delete_file(&self, rel_path: &Path) -> anyhow::Result<IngestOutcome> {
        self.ingest_one(IngestionMessage::DeleteFile(rel_path.to_path_buf()))
            .await
    }
}

/// Rows per batched `INSERT`. Bounds the statement's bind-parameter count.
const INSERT_CHUNK: usize = 500;

/// Insert records through the ORM with one multi-row `INSERT` per chunk, instead of one
/// statement per row.
async fn insert_all<M, C>(
    tx: &mut toasty::db::Transaction<'_>,
    items: Vec<C>,
    create_many: fn() -> toasty::stmt::CreateMany<M>,
) -> anyhow::Result<()>
where
    M: toasty::schema::Model,
    C: toasty::stmt::IntoInsert<Model = M>,
{
    let mut items = items.into_iter().peekable();
    while items.peek().is_some() {
        let mut create = create_many();
        for item in items.by_ref().take(INSERT_CHUNK) {
            create = create.item(item);
        }
        create.exec(&mut *tx).await?;
    }
    Ok(())
}

/// References and edges produced for a file, awaiting target resolution.
struct PendingLinks {
    file_id: u64,
    /// Project-relative path of the file, for the same-directory tie-break.
    file_path: String,
    /// Bare names of the file's symbols that were stored (not kept from an earlier version).
    new_names: Vec<String>,
    /// Links from other files to symbols the file no longer defines.
    orphans: Vec<Deferred>,
    /// The file's own symbols by qualified and bare name.
    local_symbols: HashMap<String, u64>,
    references: Vec<RawSymbolReference>,
    edges: Vec<RawCodeGraphEdge>,
}

/// Delete the rows a file owns: its symbols and scopes, the references and edges that start in
/// it and its unresolved references. Links from other files into its symbols stay; the caller
/// decides what becomes of them (see [`take_incoming_links`]).
async fn purge_file_records(
    tx: &mut toasty::db::Transaction<'_>,
    file_id: u64,
) -> anyhow::Result<()> {
    // Everything here goes by one column that has its own index (`file_id`): this engine's
    // planner scans the table for `project_id = ? AND ...` (it cost ~200 ms per replaced file
    // on a 1,400-file project) and for `IN` lists or subqueries (~70 ms). File ids are unique,
    // so the project is implied. A fixed number of statements, however many symbols the file
    // has.
    for table in [
        "symbol_reference",
        "unresolved_reference",
        "code_graph_edge",
        "symbol",
        "scope",
    ] {
        let stmt = format!("DELETE FROM {table} WHERE file_id = {file_id};");
        toasty::sql::statement(&stmt).exec(&mut *tx).await?;
    }
    Ok(())
}

/// Take the references and call edges that point at `removed` out of the tables and return them
/// as deferred links whose target is the removed symbol's qualified name: they re-link only to a
/// symbol that name still describes (the same path defined elsewhere, or defined again later),
/// never to an unrelated symbol that merely shares the bare name.
async fn take_incoming_links(
    tx: &mut toasty::db::Transaction<'_>,
    removed: &[&Symbol],
) -> anyhow::Result<Vec<Deferred>> {
    let mut orphans = Vec::new();
    for symbol in removed {
        // By symbol id alone, as above.
        let refs = toasty::sql::query(
            "SELECT r.project_id, r.file_id, f.path, r.role, r.start_byte, r.end_byte, \
             r.line_number, r.via \
             FROM symbol_reference r JOIN source_file f ON f.id = r.file_id \
             WHERE r.symbol_id = ?1",
        )
        .bind(symbol.id as i64)
        .exec(&mut *tx)
        .await?;
        for row in &refs {
            let toasty::stmt::Value::Record(record) = row else {
                anyhow::bail!("expected a record row, got {row:?}");
            };
            let [
                project_id,
                file_id,
                path,
                role,
                start_byte,
                end_byte,
                line_number,
                via,
            ] = record.fields.as_slice()
            else {
                anyhow::bail!("unexpected reference row shape: {record:?}");
            };
            orphans.push(Deferred {
                project_id: value_u64(project_id)?,
                file_id: value_u64(file_id)?,
                file_path: path.as_str().unwrap_or_default().to_string(),
                item: DeferredItem::Reference(RawSymbolReference {
                    target_symbol_name: symbol.qualified_name.clone(),
                    role: role.as_str().unwrap_or_default().to_string(),
                    start_byte: value_u64(start_byte)? as usize,
                    end_byte: value_u64(end_byte)? as usize,
                    line_number: u32::try_from(value_u64(line_number)?)?,
                    via: Via::parse(via.as_str().unwrap_or_default()),
                }),
                origin: Origin::Orphan,
            });
        }
        if !refs.is_empty() {
            let stmt = format!(
                "DELETE FROM symbol_reference WHERE symbol_id = {};",
                symbol.id
            );
            toasty::sql::statement(&stmt).exec(&mut *tx).await?;
        }

        let edges = toasty::sql::query(
            "SELECT e.project_id, e.source_symbol_id, e.file_id, f.path, e.edge_type \
             FROM code_graph_edge e JOIN source_file f ON f.id = e.file_id \
             WHERE e.target_symbol_id = ?1",
        )
        .bind(symbol.id as i64)
        .exec(&mut *tx)
        .await?;
        for row in &edges {
            let toasty::stmt::Value::Record(record) = row else {
                anyhow::bail!("expected a record row, got {row:?}");
            };
            let [project_id, source_id, file_id, path, edge_type] = record.fields.as_slice() else {
                anyhow::bail!("unexpected edge row shape: {record:?}");
            };
            orphans.push(Deferred {
                project_id: value_u64(project_id)?,
                file_id: value_u64(file_id)?,
                file_path: path.as_str().unwrap_or_default().to_string(),
                item: DeferredItem::Edge {
                    source_id: value_u64(source_id)?,
                    edge: RawCodeGraphEdge {
                        source_symbol_name: String::new(),
                        target_symbol_name: symbol.qualified_name.clone(),
                        edge_type: edge_type.as_str().unwrap_or_default().to_string(),
                    },
                },
                origin: Origin::Orphan,
            });
        }
        if !edges.is_empty() {
            let stmt = format!(
                "DELETE FROM code_graph_edge WHERE target_symbol_id = {};",
                symbol.id
            );
            toasty::sql::statement(&stmt).exec(&mut *tx).await?;
        }
    }
    Ok(orphans)
}

/// The part of a stored symbol that decides whether a re-indexed symbol continues it.
struct OldSymbol<'a> {
    id: u64,
    qualified_name: &'a str,
    kind: &'a str,
    start_byte: usize,
}

/// For each new symbol of a file, the id of the old symbol it continues, if any.
///
/// Symbols match by `(qualified_name, kind)`; of several with the same key (overloads in other
/// languages, `cfg` variants) the first in the new file continues the first in the old one, and
/// so on, by position. A symbol that moves within the file, or whose body changes, keeps its id.
fn match_symbols(old: &[OldSymbol], new: &[RawSymbol]) -> Vec<Option<u64>> {
    let mut old_order: Vec<&OldSymbol> = old.iter().collect();
    old_order.sort_by_key(|s| (s.start_byte, s.id));
    let mut old_by_key: HashMap<(&str, &str), Vec<u64>> = HashMap::new();
    for symbol in old_order {
        old_by_key
            .entry((symbol.qualified_name, symbol.kind))
            .or_default()
            .push(symbol.id);
    }

    let mut new_order: Vec<usize> = (0..new.len()).collect();
    new_order.sort_by_key(|&i| (new[i].start_byte, i));
    let mut taken: HashMap<(&str, &str), usize> = HashMap::new();
    let mut matched = vec![None; new.len()];
    for i in new_order {
        let key = (new[i].qualified_name.as_str(), new[i].kind.as_str());
        let n = taken.entry(key).or_default();
        matched[i] = old_by_key.get(&key).and_then(|ids| ids.get(*n)).copied();
        *n += 1;
    }
    matched
}

/// Queue `value` as the next bound parameter and return its placeholder.
fn placeholder<'a>(binds: &mut Vec<&'a str>, value: &'a str) -> String {
    binds.push(value);
    format!("?{}", binds.len())
}

/// Whose symbols are being stored.
struct SymbolOwner {
    project_id: u64,
    file_id: u64,
    dependency_package_id: Option<u64>,
}

/// Insert symbol rows under the ids they had before the file was re-indexed. The ORM assigns
/// ids itself, so this is a plain multi-row `INSERT`; only the text columns are bound (the
/// numbers come from the database or the parser, and a NULL cannot be bound untyped).
async fn insert_kept_symbols(
    tx: &mut toasty::db::Transaction<'_>,
    owner: SymbolOwner,
    scope_ids: &HashMap<usize, u64>,
    kept: &[(u64, &RawSymbol)],
) -> anyhow::Result<()> {
    let SymbolOwner {
        project_id,
        file_id,
        dependency_package_id,
    } = owner;
    let package = dependency_package_id.map_or("NULL".to_string(), |id| id.to_string());
    for chunk in kept.chunks(INSERT_CHUNK) {
        let mut binds: Vec<&str> = Vec::with_capacity(chunk.len() * 6);
        let mut rows = Vec::with_capacity(chunk.len());
        for (id, symbol) in chunk {
            // Evaluated in column order: the placeholders must be numbered in bind order.
            let name = placeholder(&mut binds, &symbol.name);
            let qualified = placeholder(&mut binds, &symbol.qualified_name);
            let kind = placeholder(&mut binds, &symbol.kind);
            let visibility = placeholder(&mut binds, &symbol.visibility);
            let signature = placeholder(&mut binds, &symbol.signature);
            let docstring = match &symbol.docstring {
                Some(doc) => placeholder(&mut binds, doc),
                None => "NULL".to_string(),
            };
            let scope = symbol
                .scope_local_id
                .and_then(|local| scope_ids.get(&local))
                .map_or("NULL".to_string(), |id| id.to_string());
            rows.push(format!(
                "({id},{project_id},{file_id},{package},{scope},{name},{qualified},{kind},\
                 {visibility},{signature},{docstring},{},{},{},{})",
                symbol.start_byte, symbol.end_byte, symbol.start_line, symbol.end_line
            ));
        }
        let mut statement = toasty::sql::statement(format!(
            "INSERT INTO symbol (id, project_id, file_id, dependency_package_id, scope_id, name, \
             qualified_name, kind, visibility, signature, docstring, start_byte, end_byte, \
             start_line, end_line) VALUES {}",
            rows.join(",")
        ));
        for value in binds {
            statement = statement.bind(value);
        }
        statement.exec(&mut *tx).await?;
    }
    Ok(())
}

/// What the tie-break needs to know about a stored symbol, and the path of its file. A slim
/// row: loading whole `Symbol` models for thousands of candidates dominated the final pass.
struct Candidate {
    id: u64,
    file_id: u64,
    qualified_name: String,
    start_byte: usize,
    path: String,
}

/// Candidates by bare name, in [`candidate_order`].
type CandidateCache = HashMap<String, Vec<Candidate>>;

/// Candidates are ordered by where they are defined, not by database id, so which of several
/// same-named symbols a reference picks does not depend on the order files were stored in.
fn candidate_order(a: &Candidate, b: &Candidate) -> std::cmp::Ordering {
    (&a.path, a.start_byte, &a.qualified_name, a.file_id).cmp(&(
        &b.path,
        b.start_byte,
        &b.qualified_name,
        b.file_id,
    ))
}

fn value_u64(value: &toasty::stmt::Value) -> anyhow::Result<u64> {
    use toasty::stmt::Value;
    match value {
        Value::I64(v) => Ok(u64::try_from(*v)?),
        Value::U64(v) => Ok(*v),
        Value::I32(v) => Ok(u64::try_from(*v)?),
        other => anyhow::bail!("expected an integer column, got {other:?}"),
    }
}

/// Load candidate symbols for the given bare names, grouped by name.
async fn fetch_candidates(
    executor: &mut dyn toasty::Executor,
    project_id: u64,
    names: &[String],
) -> anyhow::Result<CandidateCache> {
    let mut by_name = CandidateCache::new();
    for name in names {
        let rows = toasty::sql::query(
            "SELECT s.id, s.name, s.qualified_name, s.file_id, s.start_byte, f.path \
             FROM symbol s JOIN source_file f ON f.id = s.file_id \
             WHERE s.project_id = ?1 AND s.name = ?2",
        )
        .bind(project_id as i64)
        .bind(name.as_str())
        .exec(&mut *executor)
        .await?;
        for row in rows {
            let toasty::stmt::Value::Record(record) = row else {
                anyhow::bail!("expected a record row, got {row:?}");
            };
            let [id, name, qualified, file_id, start_byte, path] = record.fields.as_slice() else {
                anyhow::bail!("unexpected candidate row shape: {record:?}");
            };
            by_name
                .entry(name.as_str().unwrap_or_default().to_string())
                .or_default()
                .push(Candidate {
                    id: value_u64(id)?,
                    file_id: value_u64(file_id)?,
                    qualified_name: qualified.as_str().unwrap_or_default().to_string(),
                    start_byte: value_u64(start_byte)? as usize,
                    path: path.as_str().unwrap_or_default().to_string(),
                });
        }
    }
    for list in by_name.values_mut() {
        list.sort_by(candidate_order);
    }
    Ok(by_name)
}

/// The last segment of a qualified name: `crate::a::f` (Rust) and `pkg.mod.f` (Python,
/// TypeScript) both give `f`.
fn bare_name(target: &str) -> &str {
    let last = target.rsplit("::").next().unwrap_or(target);
    last.rsplit('.').next().unwrap_or(last)
}

/// Choose the symbol a reference target names from same-name candidates.
///
/// Path targets (`a::b::f`) match a symbol whose qualified name equals the path, else one
/// that ends with `::a::b::f`. Candidates in another language than `from` never match. Among
/// the matches, and for bare names: a symbol in `prefer_file`, then one in `from`'s directory, then the first in [`candidate_order`] (path, then
/// position). See ADR 0013.
fn pick_candidate<'a>(
    candidates: &'a [Candidate],
    target: &str,
    prefer_file: Option<u64>,
    from: Option<&Path>,
) -> Option<(&'a Candidate, bool)> {
    // `matching` is in candidate order, so its first item is the fallback. The flag says the
    // match was the only one.
    fn prefer<'a>(
        matching: impl Iterator<Item = &'a Candidate> + Clone,
        prefer_file: Option<u64>,
        from: Option<&Path>,
    ) -> Option<(&'a Candidate, bool)> {
        // Only symbols in the referencing file's language can be what it means.
        let matching = matching.filter(move |c| {
            from.is_none_or(|from| crate::parsers::same_language(Path::new(&c.path), from))
        });
        let from_dir = from.and_then(Path::parent);
        let unique = matching.clone().take(2).count() == 1;
        prefer_file
            .and_then(|f| matching.clone().find(|c| c.file_id == f))
            .or_else(|| {
                from_dir.and_then(|dir| {
                    matching
                        .clone()
                        .find(|c| Path::new(&c.path).parent() == Some(dir))
                })
            })
            .or_else(|| matching.clone().next())
            .map(|c| (c, unique))
    }

    if target.contains("::") || target.contains('.') {
        let exact = candidates.iter().filter(|c| c.qualified_name == target);
        if exact.clone().next().is_some() {
            return prefer(exact, prefer_file, from);
        }
        let rust_suffix = format!("::{target}");
        let dotted_suffix = format!(".{target}");
        let suffixed = candidates.iter().filter(|c| {
            c.qualified_name.ends_with(&rust_suffix) || c.qualified_name.ends_with(&dotted_suffix)
        });
        if suffixed.clone().next().is_some() {
            return prefer(suffixed, prefer_file, from);
        }
        // A path through a re-export (`ops::Filter::new` for `ops::cargo_compile::Filter::new`):
        // its segments appear, in order, in the definition's qualified name. Never `exact`.
        let reexported = candidates
            .iter()
            .filter(|c| is_reexported_path(&c.qualified_name, target));
        return prefer(reexported, prefer_file, from).map(|(c, _)| (c, false));
    }
    prefer(candidates.iter(), prefer_file, from)
}

/// Whether `target` (`a::b::f`, `pkg.mod.f`) names `qualified` through a re-export: all of its
/// segments appear in `qualified` in the same order and the last is the same. A path of three
/// or more segments qualifies on that alone; one of two (`crate_name::Item`, a glob re-export
/// at a crate root) only when its first segment is the definition's own crate root, because
/// `io::Error` could as well be somebody else's.
fn is_reexported_path(qualified: &str, target: &str) -> bool {
    let sep = if target.contains("::") { "::" } else { "." };
    let wanted: Vec<&str> = target.split(sep).collect();
    let mut have = qualified.split(sep);
    let last_matches = qualified.rsplit(sep).next() == wanted.last().copied();
    let own_crate = qualified.split(sep).next() == wanted.first().copied();
    if !last_matches || wanted.len() < 2 || (wanted.len() == 2 && !own_crate) {
        return false;
    }
    wanted
        .iter()
        .all(|segment| have.any(|part| part == *segment))
}

/// A stored reference row for `raw`.
fn reference_row(
    project_id: u64,
    symbol_id: u64,
    file_id: u64,
    raw: RawSymbolReference,
    via: Via,
) -> impl toasty::stmt::IntoInsert<Model = SymbolReference> {
    SymbolReference::create()
        .project_id(project_id)
        .symbol_id(symbol_id)
        .file_id(file_id)
        .role(raw.role)
        .start_byte(raw.start_byte)
        .end_byte(raw.end_byte)
        .line_number(raw.line_number)
        .via(via.as_str().to_string())
}

/// What differs between a persisted unresolved reference and a persisted unresolved edge.
struct Unresolved {
    target: String,
    kind: &'static str,
    role: String,
    start_byte: usize,
    end_byte: usize,
    line_number: u32,
    via: Via,
    source_id: u64,
    edge_type: String,
}

impl From<RawSymbolReference> for Unresolved {
    fn from(raw: RawSymbolReference) -> Self {
        Self {
            target: raw.target_symbol_name,
            kind: "reference",
            role: raw.role,
            start_byte: raw.start_byte,
            end_byte: raw.end_byte,
            line_number: raw.line_number,
            via: raw.via,
            source_id: 0,
            edge_type: String::new(),
        }
    }
}

impl Unresolved {
    fn edge(source_id: u64, edge: RawCodeGraphEdge) -> Self {
        Self {
            target: edge.target_symbol_name,
            kind: "edge",
            role: String::new(),
            start_byte: 0,
            end_byte: 0,
            line_number: 0,
            via: Via::Name,
            source_id,
            edge_type: edge.edge_type,
        }
    }
}

/// A persisted unresolved reference or edge.
fn unresolved_row(
    project_id: u64,
    file_id: u64,
    item: Unresolved,
) -> impl toasty::stmt::IntoInsert<Model = UnresolvedReference> {
    UnresolvedReference::create()
        .project_id(project_id)
        .file_id(file_id)
        .bare_name(bare_name(&item.target).to_string())
        .target_name(item.target)
        .kind(item.kind.to_string())
        .role(item.role)
        .start_byte(item.start_byte)
        .end_byte(item.end_byte)
        .line_number(item.line_number)
        .via(item.via.as_str().to_string())
        .source_symbol_id(item.source_id)
        .edge_type(item.edge_type)
}

/// How many unresolved references (not edges) the project has stored.
async fn count_unresolved(
    executor: &mut dyn toasty::Executor,
    project_id: u64,
) -> anyhow::Result<usize> {
    let rows = toasty::sql::query(
        "SELECT COUNT(*) FROM unresolved_reference WHERE project_id = ?1 AND kind = 'reference'",
    )
    .bind(project_id as i64)
    .exec(&mut *executor)
    .await?;
    let count = match rows.first() {
        Some(toasty::stmt::Value::Record(record)) => record.fields.first().map(value_u64),
        Some(other) => Some(value_u64(other)),
        None => None,
    };
    Ok(count.transpose()?.unwrap_or(0) as usize)
}

/// Whether an edge of this type already links the two symbols.
async fn edge_exists(
    executor: &mut dyn toasty::Executor,
    source_id: u64,
    target_id: u64,
    edge_type: &str,
) -> anyhow::Result<bool> {
    let rows = toasty::sql::query(
        "SELECT 1 FROM code_graph_edge WHERE source_symbol_id = ?1 AND target_symbol_id = ?2 \
         AND edge_type = ?3",
    )
    .bind(source_id as i64)
    .bind(target_id as i64)
    .bind(edge_type)
    .exec(&mut *executor)
    .await?;
    Ok(!rows.is_empty())
}

/// Find the project symbol a reference target names.
///
/// Falls back to external dependencies (project 0) when the project defines no symbol
/// with that name. See [`pick_candidate`] for the matching rules.
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 = bare_name(target);
    if name.is_empty() {
        return Ok(None);
    }
    let names = [name.to_string()];
    let mut by_name = fetch_candidates(&mut *executor, project_id, &names).await?;
    if by_name.is_empty() && project_id != 0 {
        by_name = fetch_candidates(&mut *executor, 0, &names).await?;
    }
    let candidates = by_name.remove(name).unwrap_or_default();
    let Some((picked, _)) = pick_candidate(&candidates, target, prefer_file, None) else {
        return Ok(None);
    };
    Ok(Symbol::filter(Symbol::fields().id().eq(picked.id))
        .first()
        .exec(&mut *executor)
        .await?)
}

/// Result of one writer transaction.
#[derive(Default)]
struct BatchResult {
    counts: LinkCounts,
    deferred: Vec<Deferred>,
    stored_edges: HashSet<(u64, u64, String)>,
    new_names: HashSet<String>,
    touched_files: HashSet<u64>,
}

#[cfg(test)]
mod tests {
    use super::*;

    fn raw(qualified: &str, kind: &str, start_byte: usize) -> RawSymbol {
        RawSymbol {
            scope_local_id: None,
            name: qualified.rsplit("::").next().unwrap().to_string(),
            qualified_name: qualified.to_string(),
            kind: kind.to_string(),
            visibility: "public".to_string(),
            signature: String::new(),
            docstring: None,
            start_byte,
            end_byte: start_byte + 1,
            start_line: 1,
            end_line: 1,
        }
    }

    fn old(
        id: u64,
        qualified_name: &'static str,
        kind: &'static str,
        start_byte: usize,
    ) -> OldSymbol<'static> {
        OldSymbol {
            id,
            qualified_name,
            kind,
            start_byte,
        }
    }

    #[test]
    fn symbols_match_by_qualified_name_and_kind_wherever_they_moved() {
        let before = [old(1, "crate::a", "fn", 0), old(2, "crate::b", "fn", 10)];
        // `b` now comes first, `a` moved down, and `c` is new.
        let after = [
            raw("crate::b", "fn", 0),
            raw("crate::a", "fn", 40),
            raw("crate::c", "fn", 80),
        ];
        assert_eq!(match_symbols(&before, &after), [Some(2), Some(1), None]);
    }

    #[test]
    fn a_different_kind_is_a_different_symbol() {
        let before = [old(1, "crate::a", "fn", 0)];
        let after = [raw("crate::a", "struct", 0)];
        assert_eq!(match_symbols(&before, &after), [None]);
    }

    #[test]
    fn duplicates_pair_up_by_position_not_by_payload_or_row_order() {
        // Two `cfg` variants of one function; the stored rows come back in any order.
        let before = [old(7, "crate::f", "fn", 50), old(3, "crate::f", "fn", 10)];
        // The new payload lists them out of order too, and one line was added above both.
        let after = [raw("crate::f", "fn", 70), raw("crate::f", "fn", 30)];
        assert_eq!(match_symbols(&before, &after), [Some(7), Some(3)]);
    }

    #[test]
    fn a_removed_duplicate_leaves_the_last_old_one_unmatched() {
        let before = [old(1, "crate::f", "fn", 0), old(2, "crate::f", "fn", 20)];
        let after = [raw("crate::f", "fn", 0)];
        assert_eq!(match_symbols(&before, &after), [Some(1)]);
        let grown = [
            raw("crate::f", "fn", 0),
            raw("crate::f", "fn", 20),
            raw("crate::f", "fn", 40),
        ];
        assert_eq!(match_symbols(&before, &grown), [Some(1), Some(2), None]);
    }
}