fix identifier!

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<PrismaValue> = 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<FileCreated> = block_on(db._query_raw(raw!(
+				println!("Creating {} unique file records. {:?}", values.len(), values);
+				
+				let files: Vec<FileCreated> = 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<usize>,
 }
 
-pub async fn count_orphan_file_paths(ctx: &CoreContext) -> Result<usize, FileError> {
+pub async fn count_orphan_file_paths(
+	ctx: &CoreContext,
+	location_id: i64,
+) -> Result<usize, FileError> {
 	let db = &ctx.database;
 	let files_count = db
 		._query_raw::<CountRes>(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<Vec<PrismaValue>> {
-	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)])
 }

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