diff --git a/core/src/file/cas/identifier.rs b/core/src/file/cas/identifier.rs index bf49e768c..dcc48dbda 100644 --- a/core/src/file/cas/identifier.rs +++ b/core/src/file/cas/identifier.rs @@ -12,6 +12,7 @@ use crate::{ use anyhow::Result; use futures::executor::block_on; use prisma_client_rust::prisma_models::PrismaValue; +use prisma_client_rust::raw::Raw; use prisma_client_rust::{raw, Direction}; use serde::{Deserialize, Serialize}; @@ -35,7 +36,11 @@ impl Job for FileIdentifierJob { } async fn run(&self, ctx: WorkerContext) -> Result<()> { println!("Identifying files"); - let total_count = count_orphan_file_paths(&ctx.core_ctx).await?; + let location = get_location(&ctx.core_ctx, self.location_id).await?; + let location_path = location.path.unwrap_or("".to_string()); + + let total_count = count_orphan_file_paths(&ctx.core_ctx, location.id.into()).await?; + println!("Found {} orphan file paths", total_count); let task_count = (total_count as f64 / 100f64).ceil() as usize; @@ -47,10 +52,6 @@ impl Job for FileIdentifierJob { let db = ctx.core_ctx.database.clone(); - let location = get_location(&ctx.core_ctx, self.location_id).await?; - - let location_path = location.path.unwrap_or("".to_string()); - let ctx = tokio::task::spawn_blocking(move || { let mut completed: usize = 0; let mut cursor: i32 = 1; @@ -64,12 +65,12 @@ impl Job for FileIdentifierJob { task_count ); - let mut rows = Vec::new(); + let mut values: Vec = Vec::new(); // only rows that have a valid cas_id to be inserted for file_path in file_paths.iter() { match prepare_file_values(&location_path, file_path) { Ok(data) => { - rows.extend(data); + values.extend(data); } Err(e) => { println!("Error processing file: {}", e); @@ -77,20 +78,26 @@ impl Job for FileIdentifierJob { } }; } - if rows.len() == 0 { + if values.len() == 0 { println!("No orphan files to process, finishing..."); break; } - let files: Vec = block_on(db._query_raw(raw!( + println!("Creating {} unique file records. {:?}", values.len(), values); + + let files: Vec = block_on(db._query_raw(Raw::new( &format!( "INSERT INTO files (cas_id, size_in_bytes) VALUES {} ON CONFLICT (cas_id) DO NOTHING RETURNING id, cas_id", - vec!["({}, {})"; rows.len()].join(",") + vec!["({}, {})"; file_paths.len()].join(",") ), - PrismaValue::List(rows) - ))).unwrap(); + values + ))).unwrap_or_else(|e| { + println!("Error inserting files: {}", e); + Vec::new() + }); // assign unique file to file path + println!("Assigning {} unique file ids to origin file_paths", files.len()); for (index, file) in files.iter().enumerate() { let file_path_id = file_paths[index].id; block_on( @@ -121,7 +128,7 @@ impl Job for FileIdentifierJob { }) .await?; - let remaining = count_orphan_file_paths(&ctx.core_ctx).await?; + let remaining = count_orphan_file_paths(&ctx.core_ctx, location.id.into()).await?; println!( "Finished with {} files remaining because your code is bad.", @@ -141,11 +148,15 @@ struct CountRes { count: Option, } -pub async fn count_orphan_file_paths(ctx: &CoreContext) -> Result { +pub async fn count_orphan_file_paths( + ctx: &CoreContext, + location_id: i64, +) -> Result { let db = &ctx.database; let files_count = db ._query_raw::(raw!( - "SELECT COUNT(*) AS count FROM file_paths WHERE file_id IS NULL AND is_dir IS FALSE" + "SELECT COUNT(*) AS count FROM file_paths WHERE file_id IS NULL AND is_dir IS FALSE AND location_id = {}", + PrismaValue::Int(location_id) )) .await?; Ok(files_count[0].count.unwrap_or(0)) @@ -174,16 +185,9 @@ pub async fn get_orphan_file_paths( pub fn prepare_file_values( location_path: &str, file_path: &file_path::Data, -) -> Result> { - println!( - "Location: {:?}, Path: {:?}", - location_path, - file_path.materialized_path.as_str() - ); - - let path = Path::new(&location_path).join(file_path.materialized_path.as_str()); - - println!("Processing file: {:?}", path); +) -> Result<[PrismaValue; 2]> { + let path = Path::new(&location_path).join(Path::new(file_path.materialized_path.as_str())); + // println!("Processing file: {:?}", path); let metadata = fs::metadata(&path)?; let cas_id = { if !file_path.is_dir { @@ -195,5 +199,5 @@ pub fn prepare_file_values( } }; - Ok(vec![PrismaValue::String(cas_id), PrismaValue::Int(0)]) + Ok([PrismaValue::String(cas_id), PrismaValue::Int(0)]) } diff --git a/core/src/file/indexer/scan.rs b/core/src/file/indexer/scan.rs index 97a51e28b..0c4980d2d 100644 --- a/core/src/file/indexer/scan.rs +++ b/core/src/file/indexer/scan.rs @@ -150,6 +150,8 @@ pub async fn scan_path( ); } + println!("Creating {} file paths. {:?}", files.len(), files); + let raw = Raw::new( &format!(" INSERT INTO file_paths (id, is_dir, location_id, materialized_path, name, extension, parent_id) @@ -183,7 +185,7 @@ fn prepare_values( is_dir: bool, ) -> Result<[PrismaValue; 7]> { let metadata = fs::metadata(&file_path)?; - let location_path = location.path.as_ref().unwrap().as_str(); + let location_path = Path::new(location.path.as_ref().unwrap().as_str()); // let size = metadata.len(); let name; let extension; @@ -200,20 +202,14 @@ fn prepare_values( name = extract_name(file_path.file_stem()); } - let materialized_path = match file_path.to_str() { - Some(p) => p - .clone() - .strip_prefix(&location_path) - // .and_then(|p| p.strip_suffix(format!("{}{}", name, extension).as_str())) - .unwrap_or_default(), - None => return Err(anyhow!("{}", file_path.to_str().unwrap_or_default())), - }; + let materialized_path = file_path.strip_prefix(location_path)?; + let materialized_path_as_string = materialized_path.to_str().unwrap_or("").to_owned(); let values = [ PrismaValue::Int(id as i64), PrismaValue::Boolean(metadata.is_dir()), PrismaValue::Int(location.id as i64), - PrismaValue::String(materialized_path.to_string()), + PrismaValue::String(materialized_path_as_string), PrismaValue::String(name), PrismaValue::String(extension.to_lowercase()), parent_id