use rusqlite::Connection;
pub fn run_migrations(conn: &Connection) -> Result<(), rusqlite::Error> {
conn.execute_batch(
"
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
title TEXT,
created_at TEXT,
updated_at TEXT,
project_path TEXT
);
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY,
session_id TEXT,
role TEXT,
content TEXT,
tool_call_id TEXT,
name TEXT,
created_at TEXT,
FOREIGN KEY (session_id) REFERENCES sessions(id)
);
-- Speeds delete/list/count by session (full replace on every persist).
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id);
CREATE TABLE IF NOT EXISTS state (
key TEXT PRIMARY KEY,
value TEXT
);
",
)?;
ensure_column(
conn,
"sessions",
"input_tokens",
"INTEGER NOT NULL DEFAULT 0",
)?;
ensure_column(
conn,
"sessions",
"output_tokens",
"INTEGER NOT NULL DEFAULT 0",
)?;
ensure_column(conn, "sessions", "cache_creation_input_tokens", "INTEGER")?;
ensure_column(conn, "sessions", "cache_read_input_tokens", "INTEGER")?;
conn.execute_batch(
"
CREATE TABLE IF NOT EXISTS memories (
id TEXT PRIMARY KEY,
project_key TEXT NOT NULL,
text TEXT NOT NULL,
embedding BLOB NOT NULL,
source_session TEXT,
created_at TEXT NOT NULL,
last_recalled_at TEXT,
recall_count INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_memories_project ON memories(project_key);
CREATE TABLE IF NOT EXISTS code_chunks (
id TEXT PRIMARY KEY,
project_key TEXT NOT NULL,
path TEXT NOT NULL,
start_line INTEGER NOT NULL,
end_line INTEGER NOT NULL,
text TEXT NOT NULL,
embedding BLOB NOT NULL,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_code_chunks_project ON code_chunks(project_key);
CREATE TABLE IF NOT EXISTS session_chunks (
id TEXT PRIMARY KEY,
project_key TEXT NOT NULL,
session_id TEXT NOT NULL,
turn_index INTEGER NOT NULL,
text TEXT NOT NULL,
embedding BLOB NOT NULL,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_session_chunks_project ON session_chunks(project_key);
",
)?;
conn.execute_batch(BRANCH_TABLES)?;
for table in FTS_TABLES {
create_fts(conn, table, "fts5")?;
}
Ok(())
}
const BRANCH_TABLES: &str = "
CREATE TABLE IF NOT EXISTS session_branches (
session_id TEXT NOT NULL,
ord INTEGER NOT NULL,
name TEXT,
PRIMARY KEY (session_id, ord)
);
CREATE TABLE IF NOT EXISTS branch_messages (
session_id TEXT NOT NULL,
ord INTEGER NOT NULL,
seq INTEGER NOT NULL,
content TEXT NOT NULL,
PRIMARY KEY (session_id, ord, seq)
);
";
pub(crate) const FTS_TABLES: [&str; 3] = ["memories", "code_chunks", "session_chunks"];
pub(crate) fn create_fts(
conn: &Connection,
table: &str,
module: &str,
) -> Result<bool, rusqlite::Error> {
let fts = format!("{table}_fts");
let existed = table_exists(conn, &fts)?;
let created = conn.execute_batch(&format!(
"
CREATE VIRTUAL TABLE IF NOT EXISTS {fts}
USING {module}(text, content='{table}', content_rowid='rowid');
CREATE TRIGGER IF NOT EXISTS {fts}_ai AFTER INSERT ON {table} BEGIN
INSERT INTO {fts}(rowid, text) VALUES (new.rowid, new.text);
END;
CREATE TRIGGER IF NOT EXISTS {fts}_ad AFTER DELETE ON {table} BEGIN
INSERT INTO {fts}({fts}, rowid, text) VALUES ('delete', old.rowid, old.text);
END;
CREATE TRIGGER IF NOT EXISTS {fts}_au AFTER UPDATE OF text ON {table} BEGIN
INSERT INTO {fts}({fts}, rowid, text) VALUES ('delete', old.rowid, old.text);
INSERT INTO {fts}(rowid, text) VALUES (new.rowid, new.text);
END;
"
));
match created {
Ok(()) => {}
Err(e) if e.to_string().contains("no such module") => return Ok(false),
Err(e) => return Err(e),
}
if !existed {
conn.execute_batch(&format!("INSERT INTO {fts}({fts}) VALUES ('rebuild');"))?;
}
Ok(true)
}
fn table_exists(conn: &Connection, table: &str) -> Result<bool, rusqlite::Error> {
let mut stmt =
conn.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 LIMIT 1")?;
let mut rows = stmt.query(rusqlite::params![table])?;
Ok(rows.next()?.is_some())
}
fn ensure_column(
conn: &Connection,
table: &str,
column: &str,
decl: &str,
) -> Result<(), rusqlite::Error> {
if column_exists(conn, table, column)? {
return Ok(());
}
match conn.execute(
&format!("ALTER TABLE {table} ADD COLUMN {column} {decl}"),
[],
) {
Ok(_) => Ok(()),
Err(e) if e.to_string().contains("duplicate column") => Ok(()),
Err(e) => Err(e),
}
}
fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool, rusqlite::Error> {
let mut stmt = conn.prepare("SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2 LIMIT 1")?;
let mut rows = stmt.query(rusqlite::params![table, column])?;
Ok(rows.next()?.is_some())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn migrations_add_usage_columns_on_legacy_schema() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
title TEXT,
created_at TEXT,
updated_at TEXT,
project_path TEXT
);
CREATE TABLE messages (
id TEXT PRIMARY KEY,
session_id TEXT,
role TEXT,
content TEXT,
tool_call_id TEXT,
name TEXT,
created_at TEXT
);
CREATE TABLE state (key TEXT PRIMARY KEY, value TEXT);
",
)
.unwrap();
run_migrations(&conn).unwrap();
assert!(column_exists(&conn, "sessions", "input_tokens").unwrap());
assert!(column_exists(&conn, "sessions", "output_tokens").unwrap());
assert!(column_exists(&conn, "sessions", "cache_creation_input_tokens").unwrap());
assert!(column_exists(&conn, "sessions", "cache_read_input_tokens").unwrap());
run_migrations(&conn).unwrap();
assert!(column_exists(&conn, "memories", "embedding").unwrap());
assert!(table_exists(&conn, "memories").unwrap());
assert!(table_exists(&conn, "session_branches").unwrap());
assert!(table_exists(&conn, "branch_messages").unwrap());
}
fn fts_hits(conn: &Connection, table: &str, word: &str) -> i64 {
conn.query_row(
&format!("SELECT COUNT(*) FROM {table}_fts WHERE {table}_fts MATCH ?1"),
[word],
|r| r.get(0),
)
.unwrap()
}
#[test]
fn fts_backfills_existing_rows_once_and_tracks_writes() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"
CREATE TABLE memories (
id TEXT PRIMARY KEY, project_key TEXT NOT NULL, text TEXT NOT NULL,
embedding BLOB NOT NULL, source_session TEXT, created_at TEXT NOT NULL,
last_recalled_at TEXT, recall_count INTEGER NOT NULL DEFAULT 0
);
INSERT INTO memories (id, project_key, text, embedding, created_at)
VALUES ('m1', 'p', 'bundled sqlite on windows', x'', 'now');
",
)
.unwrap();
run_migrations(&conn).unwrap();
assert_eq!(fts_hits(&conn, "memories", "bundled"), 1);
run_migrations(&conn).unwrap();
assert_eq!(fts_hits(&conn, "memories", "bundled"), 1);
conn.execute(
"UPDATE memories SET text = 'msvc linker' WHERE id = 'm1'",
[],
)
.unwrap();
assert_eq!(fts_hits(&conn, "memories", "bundled"), 0);
assert_eq!(fts_hits(&conn, "memories", "linker"), 1);
conn.execute("DELETE FROM memories WHERE id = 'm1'", [])
.unwrap();
assert_eq!(fts_hits(&conn, "memories", "linker"), 0);
}
#[test]
fn create_fts_reports_a_missing_module_without_failing() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch("CREATE TABLE notes (text TEXT NOT NULL);")
.unwrap();
assert!(!create_fts(&conn, "notes", "no_such_fts").unwrap());
assert!(!table_exists(&conn, "notes_fts").unwrap());
assert!(create_fts(&conn, "notes", "fts5").unwrap());
}
#[test]
fn create_fts_propagates_other_errors() {
let conn = Connection::open_in_memory().unwrap();
assert!(create_fts(&conn, "missing", "fts5").is_err());
}
#[test]
fn memories_table_created() {
let conn = Connection::open_in_memory().unwrap();
run_migrations(&conn).unwrap();
assert!(table_exists(&conn, "memories").unwrap());
run_migrations(&conn).unwrap();
}
#[test]
fn messages_session_index_created() {
let conn = Connection::open_in_memory().unwrap();
run_migrations(&conn).unwrap();
let mut stmt = conn
.prepare(
"SELECT 1 FROM sqlite_master WHERE type='index' AND name='idx_messages_session'",
)
.unwrap();
let mut rows = stmt.query([]).unwrap();
assert!(rows.next().unwrap().is_some());
run_migrations(&conn).unwrap();
}
#[test]
fn run_migrations_fails_when_first_batch_cannot_write() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch("PRAGMA query_only=ON;").unwrap();
assert!(run_migrations(&conn).is_err());
}
#[test]
fn run_migrations_fails_adding_input_tokens_on_a_view() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"
CREATE VIEW sessions AS
SELECT 'x' AS id, 't' AS title, 'c' AS created_at,
'u' AS updated_at, 'p' AS project_path;
CREATE TABLE messages (
id TEXT PRIMARY KEY, session_id TEXT, role TEXT, content TEXT,
tool_call_id TEXT, name TEXT, created_at TEXT
);
CREATE TABLE state (key TEXT PRIMARY KEY, value TEXT);
",
)
.unwrap();
assert!(run_migrations(&conn).is_err());
}
#[test]
fn run_migrations_fails_adding_output_tokens_at_column_limit() {
let conn = Connection::open_in_memory().unwrap();
let mut cols = vec![
"id TEXT PRIMARY KEY".to_string(),
"title TEXT".to_string(),
"created_at TEXT".to_string(),
"updated_at TEXT".to_string(),
"project_path TEXT".to_string(),
"input_tokens INTEGER NOT NULL DEFAULT 0".to_string(),
];
for i in cols.len()..2000 {
cols.push(format!("pad_{i} INTEGER"));
}
conn.execute(&format!("CREATE TABLE sessions ({})", cols.join(",")), [])
.unwrap();
conn.execute_batch(
"
CREATE TABLE messages (
id TEXT PRIMARY KEY, session_id TEXT, role TEXT, content TEXT,
tool_call_id TEXT, name TEXT, created_at TEXT
);
CREATE TABLE state (key TEXT PRIMARY KEY, value TEXT);
",
)
.unwrap();
assert!(run_migrations(&conn).is_err());
}
#[test]
fn run_migrations_fails_when_memory_index_cannot_be_created() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"
CREATE TABLE sessions (
id TEXT PRIMARY KEY, title TEXT, created_at TEXT,
updated_at TEXT, project_path TEXT
);
CREATE TABLE messages (
id TEXT PRIMARY KEY, session_id TEXT, role TEXT, content TEXT,
tool_call_id TEXT, name TEXT, created_at TEXT
);
CREATE TABLE state (key TEXT PRIMARY KEY, value TEXT);
CREATE TABLE memories (id TEXT PRIMARY KEY);
",
)
.unwrap();
assert!(run_migrations(&conn).is_err());
}
#[test]
fn ensure_column_treats_duplicate_name_as_already_migrated() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch("CREATE TABLE t (col INTEGER)").unwrap();
ensure_column(&conn, "t", "COL", "INTEGER").unwrap();
}
#[test]
fn ensure_column_propagates_alter_errors() {
let conn = Connection::open_in_memory().unwrap();
assert!(ensure_column(&conn, "missing", "col", "INTEGER").is_err());
}
}