hunk.code_window: 							.join(THUMBNAIL_CACHE_DIR_NAME)
   							.join(&object.cas_id)
   							.with_extension("webp");
   						object.has_thumbnail = thumb_path.exists();
   						ExplorerItem::Object(Box::new(object))
   					})
  
hunk.labels: keep keep keep keep replace keep keep keep
hunk.after_edit: 						object.has_thumbnail = thumb_path.try_exists().unwrap();
  
hunk.file_path: core/src/api/tags.rs
hunk.type: replace
file: use crate::{
  	job::{JobError, JobReportUpdate, JobResult, JobState, StatefulJob, WorkerContext},
  	library::LibraryContext,
  	prisma::{file_path, location, object},
  };
  
  use chrono::{DateTime, FixedOffset};
  use prisma_client_rust::{prisma_models::PrismaValue, raw::Raw, Direction};
  use serde::{Deserialize, Serialize};
  use std::{
  	collections::{HashMap, HashSet},
  	path::{Path, PathBuf},
  };
  use tokio::{fs, io};
  use tracing::{error, info};
  
  use super::cas::generate_cas_id;
  
  // we break this job into chunks of 100 to improve performance
  static CHUNK_SIZE: usize = 100;
  pub const IDENTIFIER_JOB_NAME: &str = "file_identifier";
  
  pub struct FileIdentifierJob {}
  
  // FileIdentifierJobInit takes file_paths without a file_id and uniquely identifies them
  // first: generating the cas_id and extracting metadata
  // finally: creating unique file records, and linking them to their file_paths
  #[derive(Serialize, Deserialize, Clone)]
  pub struct FileIdentifierJobInit {
  	pub location_id: i32,
  	pub sub_path: Option<PathBuf>, // subpath to start from
  }
  
  #[derive(Serialize, Deserialize, Debug)]
  pub struct FilePathIdAndLocationIdCursor {
  	file_path_id: i32,
  	location_id: i32,
  }
  
  impl From<&FilePathIdAndLocationIdCursor> for file_path::UniqueWhereParam {
  	fn from(cursor: &FilePathIdAndLocationIdCursor) -> Self {
  		file_path::location_id_id(cursor.location_id, cursor.file_path_id)
  	}
  }
  
  #[derive(Serialize, Deserialize)]
  pub struct FileIdentifierJobState {
  	total_count: usize,
  	task_count: usize,
  	location: location::Data,
  	location_path: PathBuf,
  	cursor: FilePathIdAndLocationIdCursor,
  }
  
  #[async_trait::async_trait]
  impl StatefulJob for FileIdentifierJob {
  	type Init = FileIdentifierJobInit;
  	type Data = FileIdentifierJobState;
  	type Step = ();
  
  	fn name(&self) -> &'static str {
  		IDENTIFIER_JOB_NAME
  	}
  
  	async fn init(
  		&self,
  		ctx: WorkerContext,
  		state: &mut JobState<Self::Init, Self::Data, Self::Step>,
  	) -> Result<(), JobError> {
  		info!("Identifying orphan Paths...");
  
  		let library = ctx.library_ctx();
  
  		let location_id = state.init.location_id;
  
  		let location = library
  			.db
  			.location()
  			.find_unique(location::id::equals(location_id))
  			.exec()
  			.await?
  			.unwrap();
  
  		let location_path = location
  			.local_path
  			.as_ref()
  			.map(PathBuf::from)
  			.unwrap_or_default();
  
  		let total_count = count_orphan_file_paths(&library, state.init.location_id).await?;
  		info!("Found {} orphan file paths", total_count);
  
  		let task_count = (total_count as f64 / CHUNK_SIZE as f64).ceil() as usize;
  		info!(
  			"Found {} orphan Paths. Will execute {} tasks...",
  			total_count, task_count
  		);
  
  		// update job with total task count based on orphan file_paths count
  		ctx.progress(vec![JobReportUpdate::TaskCount(task_count)]);
  
  		let first_path_id = library
  			.db
  			.file_path()
  			.find_first(orphan_path_filters(location_id, None))
  			.exec()
  			.await?
  			.map(|d| d.id)
  			.unwrap_or(1);
  
  		state.data = Some(FileIdentifierJobState {
  			total_count,
  			task_count,
  			location,
  			location_path,
  			cursor: FilePathIdAndLocationIdCursor {
  				file_path_id: first_path_id,
  				location_id: state.init.location_id,
  			},
  		});
  
  		state.steps = (0..task_count).map(|_| ()).collect();
  
  		Ok(())
  	}
  
  	async fn execute_step(
  		&self,
  		ctx: WorkerContext,
  		state: &mut JobState<Self::Init, Self::Data, Self::Step>,
  	) -> Result<(), JobError> {
  		let db = ctx.library_ctx().db;
  
  		// link file_path ids to a CreateFile struct containing unique file data
  		let mut chunk: HashMap<i32, CreateFile> = HashMap::new();
  		let mut cas_lookup: HashMap<String, i32> = HashMap::new();
  
  		let data = state
  			.data
  			.as_mut()
  			.expect("Critical error: missing data on job state");
  
  		// get chunk of orphans to process
  		let file_paths =
  			get_orphan_file_paths(&ctx.library_ctx(), &data.cursor, data.location.id).await?;
  
  		// if no file paths found, abort entire job early
  		if file_paths.is_empty() {
  			return Err(JobError::JobDataNotFound(
  				"Expected orphan Paths not returned from database query for this chunk".to_string(),
  			));
  		}
  
  		info!(
  			"Processing {:?} orphan Paths. ({} completed of {})",
  			file_paths.len(),
  			state.step_number,
  			data.task_count
  		);
  
  		// analyze each file_path
  		for file_path in &file_paths {
  			// get the cas_id and extract metadata
  			match assemble_object_metadata(&data.location_path, file_path).await {
  				Ok(object) => {
  					let cas_id = object.cas_id.clone();
  					// create entry into chunks for created file data
  					chunk.insert(file_path.id, object);
  					cas_lookup.insert(cas_id, file_path.id);
  				}
  				Err(e) => {
  					error!("Error assembling Object metadata: {:#?}", e);
  					continue;
  				}
  			};
  		}
  
  		// find all existing files by cas id
  		let generated_cas_ids = chunk.values().map(|c| c.cas_id.clone()).collect();
  		let existing_objects = db
  			.object()
  			.find_many(vec![object::cas_id::in_vec(generated_cas_ids)])
  			.exec()
  			.await?;
  
  		info!("Found {} existing files", existing_objects.len());
  
  		for existing_object in &existing_objects {
  			if let Err(e) = db
  				.file_path()
  				.update(
  					file_path::location_id_id(
  						state.init.location_id,
  						*cas_lookup.get(&existing_object.cas_id).unwrap(),
  					),
  					vec![file_path::object_id::set(Some(existing_object.id))],
  				)
  				.exec()
  				.await
  			{
  				error!("Error updating file_id: {:#?}", e);
  			}
  		}
  
  		let existing_object_cas_ids = existing_objects
  			.iter()
  			.map(|object| object.cas_id.clone())
  			.collect::<HashSet<_>>();
  
  		// extract objects that don't already exist in the database
  		let new_objects = chunk
  			.iter()
  			.map(|(_id, create_file)| create_file)
  			.filter(|create_file| !existing_object_cas_ids.contains(&create_file.cas_id))
  			.collect::<Vec<_>>();
  
  		if !new_objects.is_empty() {
  			// assemble prisma values for new unique files
  			let mut values = Vec::with_capacity(new_objects.len() * 3);
  			for object in &new_objects {
  				values.extend([
  					PrismaValue::String(object.cas_id.clone()),
  					PrismaValue::Int(object.size_in_bytes),
  					PrismaValue::DateTime(object.date_created),
  				]);
  			}
  
  			// create new file records with assembled values
  			// TODO: Use create_many with skip_duplicates. Waiting on https://github.com/Brendonovich/prisma-client-rust/issues/143
  			let created_files: Vec<FileCreated> = db
  				._query_raw(Raw::new(
  					&format!(
  						"INSERT INTO object (cas_id, size_in_bytes, date_created) VALUES {}
  						ON CONFLICT (cas_id) DO NOTHING RETURNING id, cas_id",
  						vec!["({}, {}, {})"; new_objects.len()].join(",")
  					),
  					values,
  				))
  				.exec()
  				.await
  				.unwrap_or_else(|e| {
  					error!("Error inserting files: {:#?}", e);
  					Vec::new()
  				});
  
  			for created_file in created_files {
  				// associate newly created files with their respective file_paths
  				// TODO: this is potentially bottle necking the chunk system, individually linking file_path to file, 100 queries per chunk
  				// - insert many could work, but I couldn't find a good way to do this in a single SQL query
  				if let Err(e) = ctx
  					.library_ctx()
  					.db
  					.file_path()
  					.update(
  						file_path::location_id_id(
  							state.init.location_id,
  							*cas_lookup.get(&created_file.cas_id).unwrap(),
  						),
  						vec![file_path::object_id::set(Some(created_file.id))],
  					)
  					.exec()
  					.await
  				{
  					info!("Error updating file_id: {:#?}", e);
  				}
  			}
  		}
  
  		// set the step data cursor to the last row of this chunk
  		if let Some(last_row) = file_paths.last() {
  			data.cursor.file_path_id = last_row.id;
  		}
  
  		ctx.progress(vec![
  			JobReportUpdate::CompletedTaskCount(state.step_number),
  			JobReportUpdate::Message(format!(
  				"Processed {} of {} orphan Paths",
  				state.step_number * CHUNK_SIZE,
  				data.total_count
  			)),
  		]);
  
  		// let _remaining = count_orphan_file_paths(&ctx.core_ctx, location_id.into()).await?;
  		Ok(())
  	}
  
  	async fn finalize(
  		&self,
  		_ctx: WorkerContext,
  		state: &mut JobState<Self::Init, Self::Data, Self::Step>,
  	) -> JobResult {
  		let data = state
  			.data
  			.as_ref()
  			.expect("critical error: missing data on job state");
  		info!(
  			"Finalizing identifier job at {}, total of {} tasks",
  			data.location_path.display(),
  			data.task_count
  		);
  
  		Ok(Some(serde_json::to_value(&state.init)?))
  	}
  }
  
  fn orphan_path_filters(location_id: i32, file_path_id: Option<i32>) -> Vec<file_path::WhereParam> {
  	let mut params = vec![
  		file_path::object_id::equals(None),
  		file_path::is_dir::equals(false),
  		file_path::location_id::equals(location_id),
  	];
  	// this is a workaround for the cursor not working properly
  	if let Some(file_path_id) = file_path_id {
  		params.push(file_path::id::gte(file_path_id))
  	}
  	params
  }
  
  #[derive(Deserialize, Serialize, Debug)]
  struct CountRes {
  	count: Option<usize>,
  }
  
  async fn count_orphan_file_paths(
  	ctx: &LibraryContext,
  	location_id: i32,
  ) -> Result<usize, prisma_client_rust::QueryError> {
  	let files_count = ctx
  		.db
  		.file_path()
  		.count(vec![
  			file_path::object_id::equals(None),
  			file_path::is_dir::equals(false),
  			file_path::location_id::equals(location_id),
  		])
  		.exec()
  		.await?;
  	// Is this
  	Ok(files_count as usize)
  }
  
  async fn get_orphan_file_paths(
  	ctx: &LibraryContext,
  	cursor: &FilePathIdAndLocationIdCursor,
  	location_id: i32,
  ) -> Result<Vec<file_path::Data>, prisma_client_rust::QueryError> {
  	info!(
  		"Querying {} orphan Paths at cursor: {:?}",
  		CHUNK_SIZE, cursor
  	);
  	ctx.db
  		.file_path()
  		.find_many(orphan_path_filters(location_id, Some(cursor.file_path_id)))
  		.order_by(file_path::id::order(Direction::Asc))
  		// .cursor(cursor.into())
  		.take(CHUNK_SIZE as i64)
  		.skip(1)
  		.exec()
  		.await
  }
  
  #[derive(Deserialize, Serialize, Debug)]
  struct CreateFile {
  	pub cas_id: String,
  	pub size_in_bytes: i64,
  	pub date_created: DateTime<FixedOffset>,
  }
  
  #[derive(Deserialize, Serialize, Debug)]
  struct FileCreated {
  	pub id: i32,
  	pub cas_id: String,
  }
  
  async fn assemble_object_metadata(
  	location_path: impl AsRef<Path>,
  	file_path: &file_path::Data,
  ) -> Result<CreateFile, io::Error> {
  	let path = location_path
  		.as_ref()
  		.join(file_path.materialized_path.as_str());
  
  	info!("Reading path: {:?}", path);
  
  	let metadata = fs::metadata(&path).await?;
  
  	// let date_created: DateTime<Utc> = metadata.created().unwrap().into();
  
  	let size = metadata.len();
  
  	let cas_id = {
  		if !file_path.is_dir {
  			let mut ret = generate_cas_id(path, size).await?;
  			ret.truncate(16);
  			ret
  		} else {
  			"".to_string()
  		}
  	};
  
  	Ok(CreateFile {
  		cas_id,
  		size_in_bytes: size as i64,
  		date_created: file_path.date_created,
  	})
  }
  