use crate::embedding::{EMBEDDING_DIMS, EmbeddingEngine};
use crate::embedding_profiles::{EmbeddingRole, profile_for};
use crate::errors::Error;
use crate::sqlite::embedding::{EmbeddingClass, classify_embedding};
use crate::sqlite::identity::ModelIdentity;
use crate::sqlite::migration_types::{MigrationReport, MigrationRowFailure};
use crate::sqlite::{self, Database, Error as SqliteError};
use std::cell::RefCell;
use std::path::Path;
pub fn migrate_model(
db_path: &Path,
config: &crate::config::Config,
) -> Result<MigrationReport, Error> {
let _profile = profile_for(&config.embedding_model)?;
let mut db = Database::open(db_path)?;
db.set_busy_timeout(std::time::Duration::ZERO)?;
let engine = RefCell::new(EmbeddingEngine::new(&config.embedding_model)?);
let report = migrate_model_with_embedder(
&mut db,
config,
&mut |content| {
engine
.borrow_mut()
.embed_passage(content)
.map_err(|e| SqliteError::Sqlite(e.to_string()))
},
&mut |content| {
engine
.borrow()
.token_count(EmbeddingRole::Passage, content)
.map_err(|e| SqliteError::Sqlite(e.to_string()))
},
)?;
Ok(report)
}
pub(crate) fn migrate_model_with_embedder<E, C>(
db: &mut Database,
config: &crate::config::Config,
embed: &mut E,
token_count: &mut C,
) -> Result<MigrationReport, SqliteError>
where
E: FnMut(&str) -> Result<Vec<f32>, SqliteError>,
C: FnMut(&str) -> Result<usize, SqliteError>,
{
let profile = profile_for(&config.embedding_model)
.map_err(|e| SqliteError::InvalidInput(e.to_string()))?;
let target = ModelIdentity {
model_id: profile.model_id.to_string(),
revision: profile.revision.to_string(),
};
let projects = db.list_all_project_ids()?;
let offending = preflight_over_limit(db, &projects, &mut *token_count)?;
if !offending.is_empty() {
return Err(SqliteError::MigrationRefused { offending });
}
write_migration_marker(db, &target)?;
let mut reindexed: usize = 0;
let mut skipped: usize = 0;
let mut failures: Vec<MigrationRowFailure> = vec![];
for project_id in &projects {
let (r, s, f) = reembed_project(db, project_id, &mut *embed)?;
reindexed += r;
skipped += s;
failures.extend(f);
}
if !failures.is_empty() {
return Err(SqliteError::MigrationIncomplete {
report: MigrationReport {
reindexed,
skipped,
failures,
},
});
}
record_identity_and_clear_marker(db, &target)?;
Ok(MigrationReport {
reindexed,
skipped,
failures: Vec::new(),
})
}
pub(crate) fn preflight_over_limit<C>(
db: &Database,
projects: &[String],
token_count: &mut C,
) -> Result<Vec<String>, SqliteError>
where
C: FnMut(&str) -> Result<usize, SqliteError>,
{
let mut offending: Vec<String> = Vec::new();
for project_id in projects {
let rows = db.list_all_rows_for_project(project_id)?;
for (id, content, _embedding) in rows {
let count = token_count(&content)?;
if count > crate::embedding::MAX_EMBEDDING_TOKENS {
offending.push(id);
}
}
}
Ok(offending)
}
pub(crate) fn reembed_project<F>(
db: &mut Database,
project_id: &str,
mut embed: F,
) -> Result<(usize, usize, Vec<MigrationRowFailure>), SqliteError>
where
F: FnMut(&str) -> Result<Vec<f32>, SqliteError>,
{
let rows = db.list_all_rows_for_project(project_id)?;
let mut reindexed: usize = 0;
let mut skipped: usize = 0;
let mut failed: Vec<MigrationRowFailure> = Vec::new();
let tx = db.begin_transaction()?;
for (id, content, embedding) in rows {
if classify_embedding(&embedding) == EmbeddingClass::Unknown {
skipped += 1;
continue;
}
match embed(&content) {
Ok(new_vec) if new_vec.len() == EMBEDDING_DIMS => {
update_embedding_in(&tx, &id, &new_vec)?;
reindexed += 1;
}
Ok(_) => failed.push(MigrationRowFailure {
id,
error: "embedder returned wrong vector length".to_string(),
}),
Err(e) => failed.push(MigrationRowFailure {
id,
error: e.to_string(),
}),
}
}
commit_tx(tx)?;
Ok((reindexed, skipped, failed))
}
fn update_embedding_in(
tx: &rusqlite::Transaction<'_>,
id: &str,
embedding: &[f32],
) -> Result<(), SqliteError> {
let blob = sqlite::vec_to_blob(embedding)?;
tx.execute(
"UPDATE memories SET embedding = ?1 WHERE id = ?2",
rusqlite::params![&blob, id],
)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
Ok(())
}
pub(crate) fn write_migration_marker(
db: &Database,
target: &ModelIdentity,
) -> Result<(), SqliteError> {
let marker = migration_marker_for(target);
db.conn()
.execute(
"INSERT INTO model_identity (id, model_id, model_revision, migration_marker)
VALUES (1, NULL, NULL, ?1)
ON CONFLICT(id) DO UPDATE SET migration_marker = excluded.migration_marker",
[marker],
)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
Ok(())
}
pub(crate) fn record_identity_and_clear_marker(
db: &mut Database,
identity: &ModelIdentity,
) -> Result<(), SqliteError> {
let tx = db.begin_transaction()?;
tx.execute(
"INSERT INTO model_identity (id, model_id, model_revision, migration_marker)
VALUES (1, ?1, ?2, NULL)
ON CONFLICT(id) DO UPDATE SET model_id = excluded.model_id,
model_revision = excluded.model_revision,
migration_marker = NULL",
(&identity.model_id, &identity.revision),
)
.map_err(|e| SqliteError::Sqlite(e.to_string()))?;
commit_tx(tx).map_err(|e| SqliteError::Sqlite(e.to_string()))?;
Ok(())
}
pub(crate) fn migration_marker_for(identity: &ModelIdentity) -> String {
format!("migrating to {}", identity.display())
}
fn commit_tx(tx: rusqlite::Transaction<'_>) -> rusqlite::Result<()> {
tx.commit()
}
#[cfg(test)]
#[path = "migration_tests.rs"]
mod migration_tests;