//! SQLite implementation of `KnowledgeStore`.
//!
//! Follows the same pattern as `repo_schedule.rs`: async trait with a
//! `run_blocking` helper that acquires the shared `Arc<Mutex<Connection>>`.

use crate::knowledge::{
    ConsolidateRequest, Consolidation, IngestRequest, KnowledgeEntry, KnowledgeError,
    KnowledgeFilter, KnowledgeQueryResult, KnowledgeStats, KnowledgeStore, QueryOpts,
    RetentionPolicy, RetentionResult, RetrievalMode,
};
use async_trait::async_trait;
use rusqlite::{Connection, params, types::Value};
use std::sync::{Arc, Mutex};
use time::OffsetDateTime;

/// SQLite-backed `KnowledgeStore`.
#[derive(Clone)]
pub struct SqliteKnowledgeStore {
    conn: Arc<Mutex<Connection>>,
}

impl SqliteKnowledgeStore {
    pub fn new(conn: Arc<Mutex<Connection>>) -> Self {
        Self { conn }
    }

    async fn run_blocking<F, R>(&self, f: F) -> Result<R, KnowledgeError>
    where
        F: FnOnce(&mut Connection) -> Result<R, KnowledgeError> + Send + 'static,
        R: Send + 'static,
    {
        let conn_arc = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let mut conn = conn_arc.lock().unwrap();
            f(&mut conn)
        })
        .await
        .map_err(|e| KnowledgeError::Other(format!("Knowledge task execution failed: {}", e)))?
    }
}

// ─── Helpers ─────────────────────────────────────────────────────────────────

/// Format an `OffsetDateTime` as RFC 3339 for SQLite TEXT storage.
fn format_dt(dt: &OffsetDateTime) -> String {
    dt.format(&time::format_description::well_known::Rfc3339)
        .unwrap_or_default()
}

/// Parse an RFC 3339 string back into `OffsetDateTime`.
fn parse_dt(s: &str) -> Result<OffsetDateTime, KnowledgeError> {
    OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339).map_err(|e| {
        KnowledgeError::DatabaseError(format!("Failed to parse datetime '{}': {}", s, e))
    })
}

/// Parse an optional RFC 3339 string.
fn parse_dt_opt(s: &Option<String>) -> Result<Option<OffsetDateTime>, KnowledgeError> {
    match s {
        Some(v) => Ok(Some(parse_dt(v)?)),
        None => Ok(None),
    }
}

/// Read a `KnowledgeEntry` from a rusqlite `Row`.
fn row_to_entry(row: &rusqlite::Row<'_>) -> Result<KnowledgeEntry, rusqlite::Error> {
    let entities_json: String = row.get("entities_json")?;
    let topics_json: String = row.get("topics_json")?;
    let connections_json: String = row.get("connections_json")?;
    let created_at_str: String = row.get("created_at")?;
    let consolidated_at_str: Option<String> = row.get("consolidated_at")?;

    let entities: Vec<String> = serde_json::from_str(&entities_json).unwrap_or_default();
    let topics: Vec<String> = serde_json::from_str(&topics_json).unwrap_or_default();
    let connections: Vec<String> = serde_json::from_str(&connections_json).unwrap_or_default();

    let created_at = OffsetDateTime::parse(
        &created_at_str,
        &time::format_description::well_known::Rfc3339,
    )
    .map_err(|_| rusqlite::Error::InvalidQuery)?;
    let consolidated_at = consolidated_at_str
        .as_deref()
        .map(|s| {
            OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339)
                .map_err(|_| rusqlite::Error::InvalidQuery)
        })
        .transpose()?;

    Ok(KnowledgeEntry {
        id: row.get("id")?,
        public_id: row.get("public_id")?,
        scope: row.get("scope")?,
        source: row.get("source")?,
        raw_text: row.get("raw_text")?,
        summary: row.get("summary")?,
        entities,
        topics,
        connections,
        importance: row.get("importance")?,
        consolidated_at,
        created_at,
    })
}

/// Read a `Consolidation` from a rusqlite `Row`.
fn row_to_consolidation(row: &rusqlite::Row<'_>) -> Result<Consolidation, rusqlite::Error> {
    let source_ids_json: String = row.get("source_entry_public_ids_json")?;
    let connections_json: String = row.get("connections_json")?;
    let created_at_str: String = row.get("created_at")?;

    let source_entry_public_ids: Vec<String> =
        serde_json::from_str(&source_ids_json).unwrap_or_default();
    let connections: Vec<String> = serde_json::from_str(&connections_json).unwrap_or_default();

    let created_at = OffsetDateTime::parse(
        &created_at_str,
        &time::format_description::well_known::Rfc3339,
    )
    .map_err(|_| rusqlite::Error::InvalidQuery)?;

    Ok(Consolidation {
        id: row.get("id")?,
        public_id: row.get("public_id")?,
        scope: row.get("scope")?,
        source_entry_public_ids,
        summary: row.get("summary")?,
        insight: row.get("insight")?,
        connections,
        created_at,
    })
}

const ENTRY_COLS: &str = "id, public_id, scope, source, raw_text, summary, \
    entities_json, topics_json, connections_json, importance, \
    consolidated_at, created_at";

/// Table-qualified column list for use in JOIN queries (FTS5).
const ENTRY_COLS_QUALIFIED: &str = "e.id, e.public_id, e.scope, e.source, e.raw_text, e.summary, \
    e.entities_json, e.topics_json, e.connections_json, e.importance, \
    e.consolidated_at, e.created_at";

const CONSOLIDATION_COLS: &str = "id, public_id, scope, source_entry_public_ids_json, \
    summary, insight, connections_json, created_at";

/// Table-qualified column list for use in JOIN queries (FTS5).
const CONSOLIDATION_COLS_QUALIFIED: &str = "e.id, e.public_id, e.scope, e.source_entry_public_ids_json, \
    e.summary, e.insight, e.connections_json, e.created_at";

// ─── KnowledgeStore implementation ───────────────────────────────────────────

#[async_trait]
impl KnowledgeStore for SqliteKnowledgeStore {
    async fn ingest(
        &self,
        scope: &str,
        entry: IngestRequest,
    ) -> Result<KnowledgeEntry, KnowledgeError> {
        let scope = scope.to_string();
        let public_id = uuid::Uuid::now_v7().to_string();
        let now = OffsetDateTime::now_utc();
        let now_str = format_dt(&now);

        let entities_json = serde_json::to_string(&entry.entities)?;
        let topics_json = serde_json::to_string(&entry.topics)?;
        let connections_json = serde_json::to_string(&entry.connections)?;

        let pid = public_id.clone();
        let scope_clone = scope.clone();
        let source = entry.source.clone();
        let raw_text = entry.raw_text.clone();
        let summary = entry.summary.clone();
        let importance = entry.importance;
        let entities = entry.entities.clone();
        let topics = entry.topics.clone();
        let connections = entry.connections.clone();

        let id = self
            .run_blocking(move |conn| {
                conn.execute(
                    "INSERT INTO knowledge_entries (
                        public_id, scope, source, raw_text, summary,
                        entities_json, topics_json, connections_json,
                        importance, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                    params![
                        pid,
                        scope_clone,
                        entry.source,
                        entry.raw_text,
                        entry.summary,
                        entities_json,
                        topics_json,
                        connections_json,
                        entry.importance,
                        now_str,
                    ],
                )
                .map_err(KnowledgeError::from)?;
                let row_id = conn.last_insert_rowid();

                // Populate junction tables with normalized forms for indexed matching
                for entity in &entry.entities {
                    let normalized = crate::knowledge::text_processing::normalize_term(entity);
                    conn.execute(
                        "INSERT INTO knowledge_entry_entities (entry_id, entity, entity_normalized) \
                         VALUES (?, ?, ?)",
                        params![row_id, entity, normalized],
                    )
                    .map_err(KnowledgeError::from)?;
                }
                for topic in &entry.topics {
                    let normalized = crate::knowledge::text_processing::normalize_term(topic);
                    conn.execute(
                        "INSERT INTO knowledge_entry_topics (entry_id, topic, topic_normalized) \
                         VALUES (?, ?, ?)",
                        params![row_id, topic, normalized],
                    )
                    .map_err(KnowledgeError::from)?;
                }

                Ok(row_id)
            })
            .await?;

        Ok(KnowledgeEntry {
            id,
            public_id,
            scope,
            source,
            raw_text: Some(raw_text),
            summary,
            entities,
            topics,
            connections,
            importance,
            consolidated_at: None,
            created_at: now,
        })
    }

    async fn list_unconsolidated(
        &self,
        scope: &str,
        limit: usize,
    ) -> Result<Vec<KnowledgeEntry>, KnowledgeError> {
        let scope = scope.to_string();
        self.run_blocking(move |conn| {
            let mut stmt = conn
                .prepare(&format!(
                    "SELECT {ENTRY_COLS} FROM knowledge_entries \
                     WHERE scope = ? AND consolidated_at IS NULL \
                     ORDER BY created_at ASC LIMIT ?"
                ))
                .map_err(KnowledgeError::from)?;
            let rows = stmt
                .query_map(params![scope, limit as i64], row_to_entry)
                .map_err(KnowledgeError::from)?;
            rows.collect::<Result<Vec<_>, _>>()
                .map_err(KnowledgeError::from)
        })
        .await
    }

    async fn list(
        &self,
        scope: &str,
        filter: KnowledgeFilter,
    ) -> Result<Vec<KnowledgeEntry>, KnowledgeError> {
        let scope = scope.to_string();
        self.run_blocking(move |conn| {
            let mut conditions = vec!["scope = ?1".to_string()];
            let mut param_idx = 2usize;

            // We use explicit named index binding via stmt.raw_bind_parameter
            // to avoid issues with dynamic param slices.
            let mut bindings: Vec<(usize, Value)> = vec![(1, Value::Text(scope))];

            if let Some(ref since) = filter.since {
                conditions.push(format!("created_at >= ?{}", param_idx));
                bindings.push((param_idx, Value::Text(format_dt(since))));
                param_idx += 1;
            }

            match filter.consolidated {
                Some(true) => conditions.push("consolidated_at IS NOT NULL".to_string()),
                Some(false) => conditions.push("consolidated_at IS NULL".to_string()),
                None => {}
            }

            // Topic/entity filtering uses normalized junction tables.
            // Filter values are normalized (lowercased + stemmed) to match
            // the stored normalized forms, enabling index-backed exact matching.
            if let Some(ref topics) = filter.topics {
                let topic_conditions: Vec<String> = topics
                    .iter()
                    .map(|t| {
                        let normalized = crate::knowledge::text_processing::normalize_term(t);
                        let cond = format!(
                            "EXISTS (SELECT 1 FROM knowledge_entry_topics \
                             WHERE entry_id = knowledge_entries.id AND topic_normalized = ?{})",
                            param_idx
                        );
                        bindings.push((param_idx, Value::Text(normalized)));
                        param_idx += 1;
                        cond
                    })
                    .collect();
                if !topic_conditions.is_empty() {
                    conditions.push(format!("({})", topic_conditions.join(" OR ")));
                }
            }

            if let Some(ref entities) = filter.entities {
                let entity_conditions: Vec<String> = entities
                    .iter()
                    .map(|e| {
                        let normalized = crate::knowledge::text_processing::normalize_term(e);
                        let cond = format!(
                            "EXISTS (SELECT 1 FROM knowledge_entry_entities \
                             WHERE entry_id = knowledge_entries.id AND entity_normalized = ?{})",
                            param_idx
                        );
                        bindings.push((param_idx, Value::Text(normalized)));
                        param_idx += 1;
                        cond
                    })
                    .collect();
                if !entity_conditions.is_empty() {
                    conditions.push(format!("({})", entity_conditions.join(" OR ")));
                }
            }

            let where_clause = conditions.join(" AND ");
            let limit_idx = param_idx;
            let sql = format!(
                "SELECT {ENTRY_COLS} FROM knowledge_entries \
                 WHERE {where_clause} \
                 ORDER BY created_at DESC LIMIT ?{limit_idx}"
            );
            bindings.push((limit_idx, Value::Integer(filter.limit as i64)));

            let mut stmt = conn.prepare(&sql).map_err(KnowledgeError::from)?;
            for (idx, val) in &bindings {
                stmt.raw_bind_parameter(*idx, val)
                    .map_err(KnowledgeError::from)?;
            }
            let mut rows = stmt.raw_query();
            let mut results = Vec::new();
            while let Some(row) = rows.next().map_err(KnowledgeError::from)? {
                results.push(row_to_entry(row).map_err(KnowledgeError::from)?);
            }
            Ok(results)
        })
        .await
    }

    async fn consolidate(
        &self,
        scope: &str,
        request: ConsolidateRequest,
    ) -> Result<Consolidation, KnowledgeError> {
        let scope = scope.to_string();
        let public_id = uuid::Uuid::now_v7().to_string();
        let now = OffsetDateTime::now_utc();
        let now_str = format_dt(&now);

        let source_ids_json = serde_json::to_string(&request.source_entry_public_ids)?;
        let connections_json = serde_json::to_string(&request.connections)?;

        let pid = public_id.clone();
        let scope_clone = scope.clone();
        let summary = request.summary.clone();
        let insight = request.insight.clone();
        let connections = request.connections.clone();
        let source_entry_public_ids = request.source_entry_public_ids.clone();

        let id = self
            .run_blocking(move |conn| {
                let tx = conn.transaction().map_err(KnowledgeError::from)?;

                // Insert the consolidation
                tx.execute(
                    "INSERT INTO knowledge_consolidations (
                        public_id, scope, source_entry_public_ids_json,
                        summary, insight, connections_json, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    params![
                        pid,
                        scope_clone,
                        source_ids_json,
                        request.summary,
                        request.insight,
                        connections_json,
                        now_str,
                    ],
                )
                .map_err(KnowledgeError::from)?;
                let consolidation_id = tx.last_insert_rowid();

                // Mark source entries as consolidated
                for entry_public_id in &request.source_entry_public_ids {
                    tx.execute(
                        "UPDATE knowledge_entries SET consolidated_at = ? \
                         WHERE public_id = ? AND scope = ? AND consolidated_at IS NULL",
                        params![now_str, entry_public_id, scope_clone],
                    )
                    .map_err(KnowledgeError::from)?;
                }

                tx.commit().map_err(KnowledgeError::from)?;
                Ok(consolidation_id)
            })
            .await?;

        Ok(Consolidation {
            id,
            public_id,
            scope,
            source_entry_public_ids,
            summary,
            insight,
            connections,
            created_at: now,
        })
    }

    async fn query(
        &self,
        scope: &str,
        question: &str,
        opts: QueryOpts,
    ) -> Result<KnowledgeQueryResult, KnowledgeError> {
        let scope = scope.to_string();
        let question = question.to_string();

        self.run_blocking(move |conn| {
            // Extract keywords with multilingual stopword filtering and stemming.
            // Raw keywords go to FTS5 (which applies its own Porter stemming).
            // Stemmed keywords go to LIKE-based entity/topic matching for consistency.
            let (keywords, stemmed_keywords) = extract_keywords(&question);
            let fts_query = build_fts5_query(&keywords);

            let entries =
                query_entries(conn, &scope, fts_query.as_deref(), &stemmed_keywords, &opts)?;
            let consolidations = if opts.include_consolidations {
                query_consolidations(conn, &scope, fts_query.as_deref(), opts.limit)?
            } else {
                vec![]
            };

            Ok(KnowledgeQueryResult {
                entries,
                consolidations,
            })
        })
        .await
    }

    async fn stats(&self, scope: &str) -> Result<KnowledgeStats, KnowledgeError> {
        let scope = scope.to_string();
        self.run_blocking(move |conn| {
            let total_entries: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM knowledge_entries WHERE scope = ?",
                    params![scope],
                    |row| row.get(0),
                )
                .map_err(KnowledgeError::from)?;

            let unconsolidated_entries: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM knowledge_entries \
                     WHERE scope = ? AND consolidated_at IS NULL",
                    params![scope],
                    |row| row.get(0),
                )
                .map_err(KnowledgeError::from)?;

            let total_consolidations: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM knowledge_consolidations WHERE scope = ?",
                    params![scope],
                    |row| row.get(0),
                )
                .map_err(KnowledgeError::from)?;

            let latest_entry_at: Option<String> = conn
                .query_row(
                    "SELECT MAX(created_at) FROM knowledge_entries WHERE scope = ?",
                    params![scope],
                    |row| row.get(0),
                )
                .map_err(KnowledgeError::from)?;

            let latest_consolidation_at: Option<String> = conn
                .query_row(
                    "SELECT MAX(created_at) FROM knowledge_consolidations WHERE scope = ?",
                    params![scope],
                    |row| row.get(0),
                )
                .map_err(KnowledgeError::from)?;

            Ok(KnowledgeStats {
                total_entries,
                unconsolidated_entries,
                total_consolidations,
                latest_entry_at: parse_dt_opt(&latest_entry_at)?,
                latest_consolidation_at: parse_dt_opt(&latest_consolidation_at)?,
            })
        })
        .await
    }

    async fn apply_retention(
        &self,
        scope: &str,
        policy: &RetentionPolicy,
    ) -> Result<RetentionResult, KnowledgeError> {
        let scope = scope.to_string();
        let policy = policy.clone();

        self.run_blocking(move |conn| {
            let tx = conn.transaction().map_err(KnowledgeError::from)?;
            let mut archived: i64 = 0;
            let mut deleted: i64 = 0;

            // Age-based retention
            if let Some(max_age_days) = policy.max_age_days {
                let cutoff = OffsetDateTime::now_utc() - time::Duration::days(max_age_days as i64);
                let cutoff_str = format_dt(&cutoff);

                if policy.archive_raw_text {
                    // Archive: set raw_text to NULL but keep summaries
                    let count = tx
                        .execute(
                            "UPDATE knowledge_entries SET raw_text = NULL \
                             WHERE scope = ? AND created_at < ? AND raw_text IS NOT NULL",
                            params![scope, cutoff_str],
                        )
                        .map_err(KnowledgeError::from)?;
                    archived += count as i64;
                } else {
                    // Hard delete old entries
                    let count = tx
                        .execute(
                            "DELETE FROM knowledge_entries \
                             WHERE scope = ? AND created_at < ?",
                            params![scope, cutoff_str],
                        )
                        .map_err(KnowledgeError::from)?;
                    deleted += count as i64;
                }
            }

            // Count-based retention
            if let Some(max_entries) = policy.max_entries {
                let current_count: i64 = tx
                    .query_row(
                        "SELECT COUNT(*) FROM knowledge_entries WHERE scope = ?",
                        params![scope],
                        |row| row.get(0),
                    )
                    .map_err(KnowledgeError::from)?;

                if current_count > max_entries {
                    let excess = current_count - max_entries;

                    if policy.archive_raw_text {
                        // Archive the oldest entries by setting raw_text to NULL
                        let count = tx
                            .execute(
                                "UPDATE knowledge_entries SET raw_text = NULL \
                                 WHERE scope = ? AND raw_text IS NOT NULL \
                                 AND id IN (
                                     SELECT id FROM knowledge_entries \
                                     WHERE scope = ? \
                                     ORDER BY created_at ASC LIMIT ?
                                 )",
                                params![scope, scope, excess],
                            )
                            .map_err(KnowledgeError::from)?;
                        archived += count as i64;
                    } else {
                        // Hard delete the oldest entries
                        let count = tx
                            .execute(
                                "DELETE FROM knowledge_entries \
                                 WHERE scope = ? AND id IN (
                                     SELECT id FROM knowledge_entries \
                                     WHERE scope = ? \
                                     ORDER BY created_at ASC LIMIT ?
                                 )",
                                params![scope, scope, excess],
                            )
                            .map_err(KnowledgeError::from)?;
                        deleted += count as i64;
                    }
                }
            }

            tx.commit().map_err(KnowledgeError::from)?;
            Ok(RetentionResult { archived, deleted })
        })
        .await
    }

    async fn is_source_ingested(
        &self,
        scope: &str,
        source_key: &str,
    ) -> Result<bool, KnowledgeError> {
        let scope = scope.to_string();
        let source_key = source_key.to_string();
        self.run_blocking(move |conn| {
            let exists: bool = conn
                .query_row(
                    "SELECT EXISTS(SELECT 1 FROM knowledge_ingestion_log \
                     WHERE scope = ? AND source_key = ?)",
                    params![scope, source_key],
                    |row| row.get(0),
                )
                .map_err(KnowledgeError::from)?;
            Ok(exists)
        })
        .await
    }

    async fn mark_source_ingested(
        &self,
        scope: &str,
        source_key: &str,
    ) -> Result<(), KnowledgeError> {
        let scope = scope.to_string();
        let source_key = source_key.to_string();
        let now_str = format_dt(&OffsetDateTime::now_utc());

        self.run_blocking(move |conn| {
            conn.execute(
                "INSERT OR IGNORE INTO knowledge_ingestion_log \
                 (scope, source_key, processed_at) VALUES (?, ?, ?)",
                params![scope, source_key, now_str],
            )
            .map_err(KnowledgeError::from)?;
            Ok(())
        })
        .await
    }
}

// ─── Query Helpers ───────────────────────────────────────────────────────────

/// Extract keywords from a question string for keyword-based search.
///
/// Uses the `text_processing` module for multilingual stopword filtering.
/// Returns `(keywords, stemmed_keywords)` — raw keywords for FTS5 (which
/// does its own Porter stemming) and stemmed keywords for LIKE-based
/// entity/topic matching where we need consistent normalization.
fn extract_keywords(question: &str) -> (Vec<String>, Vec<String>) {
    let (keywords, lang) = crate::knowledge::text_processing::extract_keywords(question);
    let stemmed = crate::knowledge::text_processing::stem_keywords(&keywords, &lang);
    (keywords, stemmed)
}

/// Build a safe FTS5 query string from extracted keywords.
///
/// Produces an OR-joined query of quoted tokens, e.g. `"rust" OR "memory"`.
/// Each token is double-quote-escaped to prevent FTS5 syntax injection
/// (e.g. a user typing `NOT` or `NEAR` in a question). Tokens with the
/// porter stemmer applied at index time will be matched via stemmed forms
/// automatically by FTS5.
///
/// Returns `None` if no valid keywords remain after stop-word filtering.
/// Build a safe FTS5 query string from pre-extracted keywords.
///
/// Produces an OR-joined query of quoted tokens, e.g. `"rust" OR "memory"`.
/// Each token is double-quote-escaped to prevent FTS5 syntax injection
/// (e.g. a user typing `NOT` or `NEAR` in a question). Tokens with the
/// porter stemmer applied at index time will be matched via stemmed forms
/// automatically by FTS5.
///
/// Takes raw (unstemmed) keywords since FTS5 applies Porter stemming
/// internally at both index and query time.
///
/// Returns `None` if no keywords are provided.
fn build_fts5_query(keywords: &[String]) -> Option<String> {
    if keywords.is_empty() {
        return None;
    }

    // Quote each token to prevent FTS5 operator injection.
    // Escape any internal double-quotes by doubling them.
    let terms: Vec<String> = keywords
        .iter()
        .map(|kw| format!("\"{}\"", kw.replace('"', "\"\"")))
        .collect();

    Some(terms.join(" OR "))
}

/// Query entries using FTS5 BM25 ranking with optional hybrid scoring.
///
/// Uses FTS5 `MATCH` for candidate selection and `bm25()` for relevance
/// ranking. The bm25 weights are: summary=10.0, raw_text=1.0, source=0.5.
///
/// In `Hybrid` mode, the BM25 score is combined with structured boosts
/// from entity/topic LIKE matches and the importance field. The `keywords`
/// parameter should contain **stemmed** keywords for consistent matching
/// against stored entity/topic values.
fn query_entries(
    conn: &Connection,
    scope: &str,
    fts_query: Option<&str>,
    keywords: &[String],
    opts: &QueryOpts,
) -> Result<Vec<KnowledgeEntry>, KnowledgeError> {
    let fts_query = match fts_query {
        Some(q) => q,
        None => {
            // No keywords: return most recent entries
            let mut stmt = conn
                .prepare(&format!(
                    "SELECT {ENTRY_COLS} FROM knowledge_entries \
                     WHERE scope = ?1 ORDER BY created_at DESC LIMIT ?2"
                ))
                .map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(1, scope)
                .map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(2, opts.limit as i64)
                .map_err(KnowledgeError::from)?;
            let mut rows = stmt.raw_query();
            let mut results = Vec::new();
            while let Some(row) = rows.next().map_err(KnowledgeError::from)? {
                results.push(row_to_entry(row).map_err(KnowledgeError::from)?);
            }
            return Ok(results);
        }
    };

    // BM25 weights: summary=10.0, raw_text=1.0, source=0.5
    // bm25() returns negative values (lower = better), so we negate for
    // combination with positive boost terms in hybrid mode.
    //
    // Hybrid mode uses a UNION to pull candidates from both FTS text matches
    // AND entity/topic LIKE matches, then ranks everything together with
    // BM25 + structured boosts + importance. This ensures entries that match
    // only via entities/topics (not in summary/raw_text/source) are still
    // discovered and ranked.
    if opts.retrieval_mode == RetrievalMode::Hybrid {
        // Build junction-table conditions for entity/topic candidate selection + boosting.
        // Uses normalized (stemmed) keywords matched against the indexed
        // `entity_normalized` / `topic_normalized` columns for exact matching.
        //
        // The FTS query param is ?1, scope is ?2.
        let mut junction_conditions = Vec::new();
        let mut junction_boosts = Vec::new();
        let mut bindings: Vec<(usize, Value)> = vec![
            (1, Value::Text(fts_query.to_string())),
            (2, Value::Text(scope.to_string())),
        ];
        let mut param_idx = 3usize;

        for keyword in keywords {
            let kw_idx = param_idx;
            bindings.push((kw_idx, Value::Text(keyword.clone())));
            param_idx += 1;

            junction_conditions.push(format!(
                "EXISTS (SELECT 1 FROM knowledge_entry_entities WHERE entry_id = knowledge_entries.id AND entity_normalized = ?{kw_idx}) OR \
                 EXISTS (SELECT 1 FROM knowledge_entry_topics WHERE entry_id = knowledge_entries.id AND topic_normalized = ?{kw_idx})"
            ));
            junction_boosts.push(format!(
                "CASE WHEN EXISTS (SELECT 1 FROM knowledge_entry_entities kee WHERE kee.entry_id = e.id AND kee.entity_normalized = ?{kw_idx}) THEN 2.0 ELSE 0.0 END + \
                 CASE WHEN EXISTS (SELECT 1 FROM knowledge_entry_topics ket WHERE ket.entry_id = e.id AND ket.topic_normalized = ?{kw_idx}) THEN 2.0 ELSE 0.0 END"
            ));
        }

        let junction_where = if junction_conditions.is_empty() {
            "0".to_string() // never true — no entity/topic candidates
        } else {
            format!("({})", junction_conditions.join(" OR "))
        };

        let boost_expr = if junction_boosts.is_empty() {
            "0.0".to_string()
        } else {
            junction_boosts.join(" + ")
        };

        let limit_idx = param_idx;
        bindings.push((limit_idx, Value::Integer(opts.limit as i64)));

        // UNION strategy:
        // 1. FTS candidates: entries matching the text query (have BM25 score)
        // 2. Entity/topic candidates: entries matching via junction tables
        //    (assigned BM25 score of 0.0 since they didn't match text)
        //
        // Both sets are scored with: bm25_score + entity/topic boosts + importance
        // Note: FTS5 requires the real table name, not an alias.
        let sql = format!(
            "SELECT {ENTRY_COLS}, _rank FROM ( \
                SELECT {ENTRY_COLS_QUALIFIED}, \
                    (-bm25(knowledge_entries_fts, 10.0, 1.0, 0.5) + ({boost_expr}) + (e.importance * 5.0)) AS _rank \
                FROM knowledge_entries_fts \
                JOIN knowledge_entries e ON e.id = knowledge_entries_fts.rowid \
                WHERE knowledge_entries_fts MATCH ?1 AND e.scope = ?2 \
              UNION \
                SELECT {ENTRY_COLS}, \
                    (0.0 + ({boost_expr_unaliased}) + (importance * 5.0)) AS _rank \
                FROM knowledge_entries \
                WHERE scope = ?2 AND {junction_where} \
             ) \
             ORDER BY _rank DESC, created_at DESC, id DESC \
             LIMIT ?{limit_idx}",
            boost_expr = boost_expr,
            boost_expr_unaliased = boost_expr.replace("e.id", "knowledge_entries.id"),
        );

        let mut stmt = conn.prepare(&sql).map_err(KnowledgeError::from)?;
        for (idx, val) in &bindings {
            stmt.raw_bind_parameter(*idx, val)
                .map_err(KnowledgeError::from)?;
        }
        let mut rows = stmt.raw_query();
        let mut results = Vec::new();
        while let Some(row) = rows.next().map_err(KnowledgeError::from)? {
            results.push(row_to_entry(row).map_err(KnowledgeError::from)?);
        }
        Ok(results)
    } else {
        // Keyword mode: pure BM25 ranking, no structured boosts.
        // Note: FTS5 requires the real table name, not an alias.
        let sql = format!(
            "SELECT {ENTRY_COLS_QUALIFIED} FROM knowledge_entries_fts \
             JOIN knowledge_entries e ON e.id = knowledge_entries_fts.rowid \
             WHERE knowledge_entries_fts MATCH ?1 AND e.scope = ?2 \
             ORDER BY bm25(knowledge_entries_fts, 10.0, 1.0, 0.5), e.created_at DESC, e.id DESC \
             LIMIT ?3"
        );

        let mut stmt = conn.prepare(&sql).map_err(KnowledgeError::from)?;
        stmt.raw_bind_parameter(1, fts_query)
            .map_err(KnowledgeError::from)?;
        stmt.raw_bind_parameter(2, scope)
            .map_err(KnowledgeError::from)?;
        stmt.raw_bind_parameter(3, opts.limit as i64)
            .map_err(KnowledgeError::from)?;
        let mut rows = stmt.raw_query();
        let mut results = Vec::new();
        while let Some(row) = rows.next().map_err(KnowledgeError::from)? {
            results.push(row_to_entry(row).map_err(KnowledgeError::from)?);
        }
        Ok(results)
    }
}

/// Query consolidations using FTS5 BM25 ranking.
///
/// Uses FTS5 `MATCH` for candidate selection and `bm25()` for relevance
/// ranking. The bm25 weights are: summary=10.0, insight=5.0.
fn query_consolidations(
    conn: &Connection,
    scope: &str,
    fts_query: Option<&str>,
    limit: usize,
) -> Result<Vec<Consolidation>, KnowledgeError> {
    match fts_query {
        None => {
            // No keywords: return most recent consolidations
            let mut stmt = conn
                .prepare(&format!(
                    "SELECT {CONSOLIDATION_COLS} FROM knowledge_consolidations \
                     WHERE scope = ?1 ORDER BY created_at DESC LIMIT ?2"
                ))
                .map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(1, scope)
                .map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(2, limit as i64)
                .map_err(KnowledgeError::from)?;
            let mut rows = stmt.raw_query();
            let mut results = Vec::new();
            while let Some(row) = rows.next().map_err(KnowledgeError::from)? {
                results.push(row_to_consolidation(row).map_err(KnowledgeError::from)?);
            }
            Ok(results)
        }
        Some(fts_q) => {
            // BM25 weights: summary=10.0, insight=5.0
            let sql = format!(
                "SELECT {CONSOLIDATION_COLS_QUALIFIED} FROM knowledge_consolidations_fts \
                 JOIN knowledge_consolidations e ON e.id = knowledge_consolidations_fts.rowid \
                 WHERE knowledge_consolidations_fts MATCH ?1 AND e.scope = ?2 \
                 ORDER BY bm25(knowledge_consolidations_fts, 10.0, 5.0), e.created_at DESC, e.id DESC \
                 LIMIT ?3"
            );

            let mut stmt = conn.prepare(&sql).map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(1, fts_q)
                .map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(2, scope)
                .map_err(KnowledgeError::from)?;
            stmt.raw_bind_parameter(3, limit as i64)
                .map_err(KnowledgeError::from)?;
            let mut rows = stmt.raw_query();
            let mut results = Vec::new();
            while let Some(row) = rows.next().map_err(KnowledgeError::from)? {
                results.push(row_to_consolidation(row).map_err(KnowledgeError::from)?);
            }
            Ok(results)
        }
    }
}

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

    /// Create an in-memory SQLite DB with schema initialized.
    fn setup_db() -> Arc<Mutex<Connection>> {
        let mut conn = Connection::open_in_memory().unwrap();
        schema::init_schema(&mut conn).unwrap();
        Arc::new(Mutex::new(conn))
    }

    fn make_ingest(source: &str, summary: &str) -> IngestRequest {
        IngestRequest {
            source: source.to_string(),
            raw_text: format!("Raw text for: {}", summary),
            summary: summary.to_string(),
            entities: vec!["entity1".to_string()],
            topics: vec!["topic1".to_string()],
            connections: vec![],
            importance: 0.5,
        }
    }

    // ── Ingest + basic retrieval ─────────────────────────────────────────

    #[tokio::test]
    async fn ingest_creates_entry_with_public_id() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        let entry = store
            .ingest("test-scope", make_ingest("user", "Test summary"))
            .await
            .unwrap();

        assert!(!entry.public_id.is_empty());
        assert_eq!(entry.scope, "test-scope");
        assert_eq!(entry.source, "user");
        assert_eq!(entry.summary, "Test summary");
        assert_eq!(
            entry.raw_text.as_deref(),
            Some("Raw text for: Test summary")
        );
        assert!(entry.consolidated_at.is_none());
    }

    #[tokio::test]
    async fn list_unconsolidated_returns_only_unconsolidated() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest("scope1", make_ingest("src", "Entry 1"))
            .await
            .unwrap();
        store
            .ingest("scope1", make_ingest("src", "Entry 2"))
            .await
            .unwrap();
        let entry3 = store
            .ingest("scope1", make_ingest("src", "Entry 3"))
            .await
            .unwrap();

        // Consolidate entry3
        store
            .consolidate(
                "scope1",
                ConsolidateRequest {
                    source_entry_public_ids: vec![entry3.public_id.clone()],
                    summary: "Consolidated".to_string(),
                    insight: "Insight".to_string(),
                    connections: vec![],
                },
            )
            .await
            .unwrap();

        let unconsolidated = store.list_unconsolidated("scope1", 10).await.unwrap();
        assert_eq!(unconsolidated.len(), 2);
        // Should not contain entry3
        assert!(
            unconsolidated
                .iter()
                .all(|e| e.public_id != entry3.public_id)
        );
    }

    #[tokio::test]
    async fn list_unconsolidated_respects_scope() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest("scope-a", make_ingest("src", "A"))
            .await
            .unwrap();
        store
            .ingest("scope-b", make_ingest("src", "B"))
            .await
            .unwrap();

        let list_a = store.list_unconsolidated("scope-a", 10).await.unwrap();
        assert_eq!(list_a.len(), 1);
        assert_eq!(list_a[0].scope, "scope-a");

        let list_b = store.list_unconsolidated("scope-b", 10).await.unwrap();
        assert_eq!(list_b.len(), 1);
    }

    // ── List with filters ───────────────────────────────────────────────

    #[tokio::test]
    async fn list_with_topic_filter() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest(
                "scope",
                IngestRequest {
                    source: "src".to_string(),
                    raw_text: "text".to_string(),
                    summary: "Has rust topic".to_string(),
                    entities: vec![],
                    topics: vec!["rust".to_string(), "programming".to_string()],
                    connections: vec![],
                    importance: 0.5,
                },
            )
            .await
            .unwrap();
        store
            .ingest(
                "scope",
                IngestRequest {
                    source: "src".to_string(),
                    raw_text: "text".to_string(),
                    summary: "Has python topic".to_string(),
                    entities: vec![],
                    topics: vec!["python".to_string()],
                    connections: vec![],
                    importance: 0.5,
                },
            )
            .await
            .unwrap();

        let results = store
            .list(
                "scope",
                KnowledgeFilter {
                    topics: Some(vec!["rust".to_string()]),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].summary, "Has rust topic");
    }

    // ── Consolidation ───────────────────────────────────────────────────

    #[tokio::test]
    async fn consolidate_marks_entries_and_creates_consolidation() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        let e1 = store
            .ingest("scope", make_ingest("src", "Entry 1"))
            .await
            .unwrap();
        let e2 = store
            .ingest("scope", make_ingest("src", "Entry 2"))
            .await
            .unwrap();

        let consolidation = store
            .consolidate(
                "scope",
                ConsolidateRequest {
                    source_entry_public_ids: vec![e1.public_id.clone(), e2.public_id.clone()],
                    summary: "Combined summary".to_string(),
                    insight: "Key insight from both entries".to_string(),
                    connections: vec!["connection1".to_string()],
                },
            )
            .await
            .unwrap();

        assert!(!consolidation.public_id.is_empty());
        assert_eq!(consolidation.scope, "scope");
        assert_eq!(consolidation.source_entry_public_ids.len(), 2);
        assert_eq!(consolidation.summary, "Combined summary");
        assert_eq!(consolidation.insight, "Key insight from both entries");

        // Entries should now be marked as consolidated
        let unconsolidated = store.list_unconsolidated("scope", 10).await.unwrap();
        assert_eq!(unconsolidated.len(), 0);
    }

    // ── Query semantics ─────────────────────────────────────────────────

    #[tokio::test]
    async fn query_keyword_matches_summary() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest(
                "scope",
                make_ingest("src", "Rust memory management patterns"),
            )
            .await
            .unwrap();
        store
            .ingest("scope", make_ingest("src", "Python web frameworks"))
            .await
            .unwrap();

        let result = store
            .query("scope", "rust memory", QueryOpts::default())
            .await
            .unwrap();
        assert!(!result.entries.is_empty());
        assert_eq!(result.entries[0].summary, "Rust memory management patterns");
    }

    #[tokio::test]
    async fn query_returns_consolidations_when_enabled() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        let e1 = store
            .ingest("scope", make_ingest("src", "Rust ownership model"))
            .await
            .unwrap();

        store
            .consolidate(
                "scope",
                ConsolidateRequest {
                    source_entry_public_ids: vec![e1.public_id.clone()],
                    summary: "Ownership patterns in Rust".to_string(),
                    insight: "Rust uses ownership for memory safety".to_string(),
                    connections: vec![],
                },
            )
            .await
            .unwrap();

        let result = store
            .query(
                "scope",
                "rust ownership",
                QueryOpts {
                    include_consolidations: true,
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        assert!(!result.consolidations.is_empty());
    }

    #[tokio::test]
    async fn query_hybrid_mode_boosts_entities() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        // Entry with matching entity
        store
            .ingest(
                "scope",
                IngestRequest {
                    source: "src".to_string(),
                    raw_text: "Some text about coding".to_string(),
                    summary: "General coding notes".to_string(),
                    entities: vec!["tokio".to_string()],
                    topics: vec!["async".to_string()],
                    connections: vec![],
                    importance: 0.9,
                },
            )
            .await
            .unwrap();

        // Entry without matching entity
        store
            .ingest(
                "scope",
                IngestRequest {
                    source: "src".to_string(),
                    raw_text: "tokio runtime details".to_string(),
                    summary: "Other notes".to_string(),
                    entities: vec![],
                    topics: vec![],
                    connections: vec![],
                    importance: 0.1,
                },
            )
            .await
            .unwrap();

        let result = store
            .query(
                "scope",
                "tokio async",
                QueryOpts {
                    retrieval_mode: RetrievalMode::Hybrid,
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        assert!(!result.entries.is_empty());
        // The entry with matching entity+topic and high importance should rank first
        assert_eq!(result.entries[0].summary, "General coding notes");
    }

    #[tokio::test]
    async fn query_empty_question_returns_recent() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest("scope", make_ingest("src", "Old entry"))
            .await
            .unwrap();
        store
            .ingest("scope", make_ingest("src", "New entry"))
            .await
            .unwrap();

        // Empty question with no keywords
        let result = store
            .query("scope", "", QueryOpts::default())
            .await
            .unwrap();
        assert_eq!(result.entries.len(), 2);
        // Most recent first
        assert_eq!(result.entries[0].summary, "New entry");
    }

    // ── Stats ───────────────────────────────────────────────────────────

    #[tokio::test]
    async fn stats_returns_correct_counts() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        let e1 = store
            .ingest("scope", make_ingest("src", "Entry 1"))
            .await
            .unwrap();
        store
            .ingest("scope", make_ingest("src", "Entry 2"))
            .await
            .unwrap();
        store
            .ingest("scope", make_ingest("src", "Entry 3"))
            .await
            .unwrap();

        store
            .consolidate(
                "scope",
                ConsolidateRequest {
                    source_entry_public_ids: vec![e1.public_id.clone()],
                    summary: "Consolidated".to_string(),
                    insight: "Insight".to_string(),
                    connections: vec![],
                },
            )
            .await
            .unwrap();

        let stats = store.stats("scope").await.unwrap();
        assert_eq!(stats.total_entries, 3);
        assert_eq!(stats.unconsolidated_entries, 2);
        assert_eq!(stats.total_consolidations, 1);
        assert!(stats.latest_entry_at.is_some());
        assert!(stats.latest_consolidation_at.is_some());
    }

    #[tokio::test]
    async fn stats_empty_scope() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        let stats = store.stats("empty-scope").await.unwrap();
        assert_eq!(stats.total_entries, 0);
        assert_eq!(stats.unconsolidated_entries, 0);
        assert_eq!(stats.total_consolidations, 0);
        assert!(stats.latest_entry_at.is_none());
    }

    // ── Retention ───────────────────────────────────────────────────────

    #[tokio::test]
    async fn retention_archive_sets_raw_text_null() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest("scope", make_ingest("src", "Old entry"))
            .await
            .unwrap();

        let result = store
            .apply_retention(
                "scope",
                &RetentionPolicy {
                    max_age_days: Some(0), // Everything is "old"
                    max_entries: None,
                    archive_raw_text: true,
                },
            )
            .await
            .unwrap();

        assert_eq!(result.archived, 1);
        assert_eq!(result.deleted, 0);

        // Verify raw_text is NULL but summary survives
        let entries = store
            .list(
                "scope",
                KnowledgeFilter {
                    limit: 10,
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert!(entries[0].raw_text.is_none());
        assert_eq!(entries[0].summary, "Old entry");
    }

    #[tokio::test]
    async fn retention_delete_removes_entries() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store
            .ingest("scope", make_ingest("src", "Entry to delete"))
            .await
            .unwrap();

        let result = store
            .apply_retention(
                "scope",
                &RetentionPolicy {
                    max_age_days: Some(0),
                    max_entries: None,
                    archive_raw_text: false,
                },
            )
            .await
            .unwrap();

        assert_eq!(result.deleted, 1);

        let stats = store.stats("scope").await.unwrap();
        assert_eq!(stats.total_entries, 0);
    }

    #[tokio::test]
    async fn retention_max_entries_caps_count() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        for i in 0..5 {
            store
                .ingest("scope", make_ingest("src", &format!("Entry {}", i)))
                .await
                .unwrap();
        }

        let result = store
            .apply_retention(
                "scope",
                &RetentionPolicy {
                    max_age_days: None,
                    max_entries: Some(3),
                    archive_raw_text: false,
                },
            )
            .await
            .unwrap();

        assert_eq!(result.deleted, 2);

        let stats = store.stats("scope").await.unwrap();
        assert_eq!(stats.total_entries, 3);
    }

    // ── Ingestion dedup ─────────────────────────────────────────────────

    #[tokio::test]
    async fn ingestion_dedup_prevents_reprocessing() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        assert!(!store.is_source_ingested("scope", "doc:123").await.unwrap());

        store
            .mark_source_ingested("scope", "doc:123")
            .await
            .unwrap();

        assert!(store.is_source_ingested("scope", "doc:123").await.unwrap());

        // Different scope should not collide
        assert!(
            !store
                .is_source_ingested("other-scope", "doc:123")
                .await
                .unwrap()
        );
    }

    #[tokio::test]
    async fn mark_source_ingested_is_idempotent() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        store.mark_source_ingested("scope", "key1").await.unwrap();
        // Second call should not error (INSERT OR IGNORE)
        store.mark_source_ingested("scope", "key1").await.unwrap();

        assert!(store.is_source_ingested("scope", "key1").await.unwrap());
    }

    // ── Keyword extraction ──────────────────────────────────────────────

    #[test]
    fn extract_keywords_filters_stop_words() {
        let (kw, _stemmed) = extract_keywords("what are the patterns and refactoring tasks?");
        assert!(kw.contains(&"patterns".to_string()));
        assert!(kw.contains(&"refactoring".to_string()));
        assert!(kw.contains(&"tasks".to_string()));
        assert!(!kw.contains(&"the".to_string()));
        assert!(!kw.contains(&"are".to_string()));
        assert!(!kw.contains(&"what".to_string()));
    }

    #[test]
    fn extract_keywords_returns_empty_for_only_stop_words() {
        let (kw, _stemmed) = extract_keywords("is it a the");
        assert!(kw.is_empty());
    }

    #[test]
    fn extract_keywords_handles_punctuation() {
        let (kw, _stemmed) = extract_keywords("rust's memory-safe, concurrent!");
        assert!(kw.contains(&"rust".to_string()));
        assert!(kw.contains(&"memory-safe".to_string()));
        assert!(kw.contains(&"concurrent".to_string()));
    }

    #[test]
    fn extract_keywords_returns_stemmed_forms() {
        let (kw, stemmed) = extract_keywords("the configurations are running smoothly today");
        assert!(kw.contains(&"configurations".to_string()));
        assert!(kw.contains(&"running".to_string()));
        // Stemmed forms should reduce to root
        assert!(stemmed.contains(&"configur".to_string()));
        assert!(stemmed.contains(&"run".to_string()));
    }

    // ── Citation contract ───────────────────────────────────────────────

    #[tokio::test]
    async fn query_results_use_public_ids_not_row_ids() {
        let db = setup_db();
        let store = SqliteKnowledgeStore::new(db);

        let entry = store
            .ingest("scope", make_ingest("src", "Test citation"))
            .await
            .unwrap();

        let result = store
            .query("scope", "test citation", QueryOpts::default())
            .await
            .unwrap();

        assert!(!result.entries.is_empty());
        // public_id should be set and match the ingested entry
        assert_eq!(result.entries[0].public_id, entry.public_id);
        // internal id should be 0 (serde skip)
        let json = serde_json::to_string(&result.entries[0]).unwrap();
        assert!(!json.contains("\"id\""));
    }
}
