mempal_runtime/ingest/
reindex.rs

1use std::path::PathBuf;
2
3use thiserror::Error;
4
5use crate::core::{db::Database, types::ReindexSource};
6use crate::embed::Embedder;
7
8use super::{
9    IngestError, IngestOptions, ingest_file_with_options, normalize::CURRENT_NORMALIZE_VERSION,
10};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum ReindexMode {
14    Stale,
15    Force,
16}
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub struct ReindexOptions {
20    pub mode: ReindexMode,
21    pub dry_run: bool,
22}
23
24#[derive(Debug, Clone, PartialEq, Eq, Default)]
25pub struct ReindexReport {
26    pub candidate_drawers: u64,
27    pub candidate_sources: u64,
28    pub processed_sources: u64,
29    pub reingested_files: usize,
30    pub reingested_chunks: usize,
31    pub skipped_existing_chunks: usize,
32    pub skipped_missing_sources: u64,
33    pub skipped_missing_drawers: u64,
34}
35
36#[derive(Debug, Error)]
37pub enum ReindexError {
38    #[error(transparent)]
39    Db(#[from] crate::core::db::DbError),
40    #[error("failed to reindex source {source_file}")]
41    Ingest {
42        source_file: String,
43        #[source]
44        source: IngestError,
45    },
46}
47
48pub async fn reindex_sources<E: Embedder + ?Sized>(
49    db: &Database,
50    embedder: &E,
51    options: ReindexOptions,
52) -> Result<ReindexReport, ReindexError> {
53    let sources = match options.mode {
54        ReindexMode::Stale => db.reindex_sources_stale(CURRENT_NORMALIZE_VERSION)?,
55        ReindexMode::Force => db.reindex_sources_force()?,
56    };
57
58    let mut report = ReindexReport {
59        candidate_drawers: sources.iter().map(|source| source.drawer_count).sum(),
60        candidate_sources: sources.len() as u64,
61        ..ReindexReport::default()
62    };
63
64    if options.dry_run {
65        return Ok(report);
66    }
67
68    for source in sources {
69        let Some(source_file) = source.source_file.as_deref() else {
70            report.skipped_missing_sources += 1;
71            report.skipped_missing_drawers += source.drawer_count;
72            continue;
73        };
74        let source_path = PathBuf::from(source_file);
75        if !source_path.is_file() {
76            report.skipped_missing_sources += 1;
77            report.skipped_missing_drawers += source.drawer_count;
78            continue;
79        }
80
81        let stats = reindex_one_source(db, embedder, &source, source_file, source_path).await?;
82        report.processed_sources += 1;
83        report.reingested_files += stats.files;
84        report.reingested_chunks += stats.chunks;
85        report.skipped_existing_chunks += stats.skipped;
86    }
87
88    Ok(report)
89}
90
91async fn reindex_one_source<E: Embedder + ?Sized>(
92    db: &Database,
93    embedder: &E,
94    source: &ReindexSource,
95    source_file: &str,
96    source_path: PathBuf,
97) -> Result<super::IngestStats, ReindexError> {
98    ingest_file_with_options(
99        db,
100        embedder,
101        &source_path,
102        &source.wing,
103        IngestOptions {
104            room: source.room.as_deref(),
105            source_root: source_path.parent(),
106            dry_run: false,
107            source_file_override: Some(source_file),
108            replace_existing_source: true,
109            replace_across_rooms: true,
110            no_strip_noise: false,
111            ..IngestOptions::default()
112        },
113    )
114    .await
115    .map_err(|source| ReindexError::Ingest {
116        source_file: source_file.to_string(),
117        source,
118    })
119}
mempal_runtime/ingest/reindex.rs

mempal_runtime/ingest/
reindex.rs
