use super::{Database, EMBEDDING_COLUMN, Error, Memory};
pub type Result<T> = std::result::Result<T, Error>;
fn classify_db_failure(err: rusqlite::Error) -> Error {
Error::from(err)
}
impl Error {
pub(crate) fn with_context(self, context: &str) -> Self {
match self {
Error::Sqlite(message) => Error::Sqlite(format!("{context}: {message}")),
other => other,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct FtsDesyncReport {
pub underpopulated_by_project: std::collections::HashMap<String, usize>,
pub underpopulated_global: usize,
pub orphans: usize,
}
impl FtsDesyncReport {
pub fn is_desynced(&self) -> bool {
self.underpopulated_global > 0 || self.orphans > 0
}
pub fn total_desynced(&self) -> usize {
self.underpopulated_global + self.orphans
}
}
impl Database {
pub fn initialize_fts(&self) -> Result<()> {
let fts_exists: bool = self
.conn
.query_row(
"SELECT name FROM sqlite_master WHERE type='table' AND name='memories_fts'",
[],
|_row| Ok(true),
)
.unwrap_or(false);
if fts_exists {
let has_project_id: bool = self.conn.query_row(
"SELECT COUNT(*) FROM pragma_table_info('memories_fts') WHERE name = 'project_id'",
[],
|row| row.get::<_, i64>(0).map(|count| count > 0),
)?;
if !has_project_id {
let tx = self.conn.unchecked_transaction()?;
let memories_exists: bool = tx.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='memories'",
[],
|row| row.get::<_, i64>(0).map(|count| count > 0),
)?;
if !memories_exists {
return Err(Error::Sqlite(
"External content table 'memories' does not exist".to_string(),
));
}
let memory_count: i64 =
tx.query_row("SELECT COUNT(*) FROM memories", [], |row| row.get(0))?;
tx.execute_batch(
"DROP TABLE IF EXISTS memories_fts;
DROP TRIGGER IF EXISTS memories_fts_insert;
DROP TRIGGER IF EXISTS memories_fts_delete;
DROP TRIGGER IF EXISTS memories_fts_update;
CREATE VIRTUAL TABLE memories_fts USING fts5(
content,
project_id UNINDEXED,
tokenize='porter unicode61',
content_rowid='rowid',
content='memories'
);
CREATE TRIGGER memories_fts_insert AFTER INSERT ON memories BEGIN
INSERT INTO memories_fts(rowid, content, project_id)
VALUES (new.rowid, new.content, new.project_id);
END;
CREATE TRIGGER memories_fts_delete AFTER DELETE ON memories BEGIN
INSERT INTO memories_fts(memories_fts, rowid, content, project_id)
VALUES('delete', old.rowid, old.content, old.project_id);
END;
CREATE TRIGGER memories_fts_update AFTER UPDATE ON memories BEGIN
INSERT INTO memories_fts(memories_fts, rowid, content, project_id)
VALUES('delete', old.rowid, old.content, old.project_id);
INSERT INTO memories_fts(rowid, content, project_id)
VALUES (new.rowid, new.content, new.project_id);
END;
INSERT INTO memories_fts(rowid, content, project_id)
SELECT rowid, content, project_id FROM memories;",
)
.map_err(|e| classify_db_failure(e).with_context("FTS5 schema migration failed"))?;
let fts_count: i64 =
tx.query_row("SELECT COUNT(*) FROM memories_fts", [], |row| row.get(0))?;
if fts_count != memory_count {
tx.rollback()?;
return Err(Error::Sqlite(format!(
"FTS5 migration incomplete: expected {} rows, got {} rows",
memory_count, fts_count
)));
}
tx.commit()?;
}
}
Ok(())
}
pub fn search_bm25(
&self,
query: &str,
project_id: &str,
limit: usize,
memory_types: Option<&[&str]>,
statuses: Option<&[&str]>,
) -> Result<Vec<Memory>> {
super::search::validate_limit(limit)?;
if !self.is_fts_initialized()? {
self.initialize_fts()?;
}
let escaped_query = Self::escape_fts_query(query);
if escaped_query.is_empty() {
return Ok(Vec::new());
}
let mut where_clauses = vec![
"memories_fts MATCH ?1".to_string(),
"m.project_id = ?2".to_string(),
];
let mut params: Vec<&dyn rusqlite::ToSql> = vec![&escaped_query, &project_id];
let param_index = super::build_filters(
&mut where_clauses,
&mut params,
3,
statuses,
memory_types,
"m.",
);
let where_clause = where_clauses.join(" AND ");
let sql = format!(
r#"
SELECT m.id, m.project_id, m.content, m.metadata, m.embedding, m.created_at, m.updated_at, m.type, m.status, m.superseded_by, m.retrieval_count, m.last_retrieved_at, m.importance,
bm25(memories_fts) as bm25_score
FROM memories_fts
JOIN memories m ON m.rowid = memories_fts.rowid
WHERE {}
ORDER BY bm25(memories_fts)
LIMIT ?{}
"#,
where_clause, param_index
);
let limit_i64 = limit as i64;
params.push(&limit_i64);
let mut stmt = self.conn.prepare(&sql)?;
let memories: rusqlite::Result<Vec<Memory>> = stmt
.query_map(params.as_slice(), |row| {
let id: String = row.get(0)?;
let blob: Vec<u8> = row.get(EMBEDDING_COLUMN)?;
let embedding = super::embedding::blob_to_vec(&blob)
.map_err(|e| super::query_mod::corrupt_embedding_error(id.clone(), e))?;
Ok(Memory {
id: row.get(0)?,
project_id: row.get(1)?,
content: row.get(2)?,
metadata: row.get(3)?,
embedding,
created_at: row.get(5)?,
updated_at: row.get(6)?,
memory_type: row.get(7)?,
status: row.get(8)?,
superseded_by: row.get(9)?,
retrieval_count: row.get(10)?,
last_retrieved_at: row.get(11)?,
importance: row.get(12)?,
similarity: Some(row.get::<_, f64>(13)?),
})
})?
.collect();
Ok(memories?)
}
pub fn detect_fts_desync(&self) -> Result<FtsDesyncReport> {
let fts_exists: i64 = self.conn.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='memories_fts'",
[],
|row| row.get(0),
)?;
if fts_exists == 0 {
return Ok(FtsDesyncReport::default());
}
let mut by_project: std::collections::HashMap<String, usize> =
std::collections::HashMap::new();
let mut underpopulated_global: usize = 0;
let mut stmt = self.conn.prepare(
"SELECT m.project_id, COUNT(*) AS c FROM memories m WHERE m.rowid IS NOT (SELECT fts.rowid FROM memories_fts fts WHERE fts.rowid = m.rowid) GROUP BY m.project_id",
)?;
let rows = stmt.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
})?;
for row_res in rows {
let (pid, c) = row_res?;
let count = c as usize;
underpopulated_global += count;
*by_project.entry(pid).or_insert(0) += count;
}
let orphan_rowids: Vec<i64> = {
let mut ostmt = self.conn.prepare(
"SELECT fts.rowid FROM memories_fts fts WHERE fts.rowid IS NOT (SELECT m.rowid FROM memories m WHERE m.rowid = fts.rowid)",
)?;
let orows = ostmt.query_map([], |row| row.get::<_, i64>(0))?;
let mut out = Vec::new();
for row_res in orows {
out.push(row_res?);
}
out
};
let orphans = orphan_rowids.len();
Ok(FtsDesyncReport {
underpopulated_by_project: by_project,
underpopulated_global,
orphans,
})
}
pub fn rebuild_fts(&self) -> Result<()> {
self.conn.execute(
"INSERT INTO memories_fts(memories_fts) VALUES('rebuild')",
[],
)?;
Ok(())
}
fn is_fts_initialized(&self) -> Result<bool> {
let count: i64 = self.conn.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='memories_fts'",
[],
|row| row.get(0),
)?;
if count == 0 {
return Ok(false);
}
let fts_count: i64 =
self.conn
.query_row("SELECT COUNT(*) FROM memories_fts", [], |row| row.get(0))?;
Ok(fts_count > 0)
}
fn escape_fts_query(query: &str) -> String {
query
.split_whitespace()
.filter(|word| !word.is_empty())
.map(|word| {
let escaped = word.replace('\\', "\\\\").replace('"', "\"\"");
format!("\"{}\"", escaped)
})
.collect::<Vec<_>>()
.join(" ")
}
}