From a8262ac8f8d86bf48f399dba3b607445e90e561c Mon Sep 17 00:00:00 2001 From: Alexander Date: Sat, 27 Jun 2026 15:17:13 +0200 Subject: [PATCH] Split and refactor the classes --- src/db/entities.rs | 62 + src/db/mod.rs | 2 + src/db/sync.rs | 74 ++ src/file_watcher.rs | 3 - src/item.rs | 84 ++ src/local.rs | 1196 ------------------ src/main.rs | 15 +- src/music/db.rs | 191 +++ src/{music_metadata.rs => music/metadata.rs} | 0 src/music/mod.rs | 2 + src/origins/local/file_io.rs | 77 ++ src/origins/local/mod.rs | 468 +++++++ src/origins/local/snapshot.rs | 107 ++ src/origins/local/watcher.rs | 53 + src/origins/mod.rs | 1 + src/virtual_dirs.rs | 121 ++ 16 files changed, 1249 insertions(+), 1207 deletions(-) create mode 100644 src/db/entities.rs create mode 100644 src/db/mod.rs create mode 100644 src/db/sync.rs delete mode 100644 src/file_watcher.rs create mode 100644 src/item.rs delete mode 100644 src/local.rs create mode 100644 src/music/db.rs rename src/{music_metadata.rs => music/metadata.rs} (100%) create mode 100644 src/music/mod.rs create mode 100644 src/origins/local/file_io.rs create mode 100644 src/origins/local/mod.rs create mode 100644 src/origins/local/snapshot.rs create mode 100644 src/origins/local/watcher.rs create mode 100644 src/origins/mod.rs create mode 100644 src/virtual_dirs.rs diff --git a/src/db/entities.rs b/src/db/entities.rs new file mode 100644 index 0000000..502062e --- /dev/null +++ b/src/db/entities.rs @@ -0,0 +1,62 @@ +use std::{fs, path::PathBuf}; + +use fuser::INodeNo; +use sea_orm::entity::prelude::*; + +use crate::item::{FileType, Item}; +use crate::virtual_dirs::parent_inode_from_path; + +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "items")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub inode: i64, + pub name: String, + pub original_path: String, + pub local_path: String, + pub file_type: String, + pub hash: i64, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} + +impl From<&Item> for ActiveModel { + fn from(item: &Item) -> Self { + use sea_orm::ActiveValue::Set; + ActiveModel { + inode: Set(item.inode.0 as i64), + name: Set(item.name.clone()), + original_path: Set(item.original_path.to_string_lossy().into_owned()), + local_path: Set(item.local_path.to_string_lossy().into_owned()), + file_type: Set(match item.file_type { + FileType::Directory => "directory".to_string(), + FileType::File => "file".to_string(), + }), + hash: Set(item.hash as i64), + } + } +} + +impl From<(Model, fs::Metadata)> for Item { + fn from((entity, metadata): (Model, fs::Metadata)) -> Self { + let local_path = PathBuf::from(&entity.local_path); + let parent_inode = parent_inode_from_path(&local_path); + let file_type = match entity.file_type.as_str() { + "directory" => FileType::Directory, + _ => FileType::File, + }; + Item::new( + INodeNo(entity.inode as u64), + parent_inode, + entity.name, + PathBuf::from(entity.original_path), + local_path, + file_type, + metadata, + None, + ) + } +} diff --git a/src/db/mod.rs b/src/db/mod.rs new file mode 100644 index 0000000..bce2a87 --- /dev/null +++ b/src/db/mod.rs @@ -0,0 +1,2 @@ +pub mod entities; +pub mod sync; diff --git a/src/db/sync.rs b/src/db/sync.rs new file mode 100644 index 0000000..eaf35a1 --- /dev/null +++ b/src/db/sync.rs @@ -0,0 +1,74 @@ +use std::collections::{BTreeMap, HashMap, HashSet}; + +use fuser::INodeNo; +use sea_orm::entity::prelude::*; + +use crate::db::entities::{ActiveModel, Entity, Model}; +use crate::item::Item; +use crate::music::db::save_music_metadata; + +pub async fn sync_items_to_db( + snapshot: &BTreeMap, + db_items: &HashMap, + client: &sea_orm::DatabaseConnection, +) { + let mut to_insert: Vec = vec![]; + let mut to_update: Vec = vec![]; + let mut to_delete: Vec = vec![]; + let mut to_save_music: Vec = vec![]; + + for (ino, item) in snapshot { + let ino_i64 = ino.0 as i64; + match db_items.get(&ino_i64) { + None => { + to_insert.push(ActiveModel::from(item)); + if item.music_metadata.is_some() { + to_save_music.push(ino_i64); + } + } + Some(db_item) if db_item.hash != item.hash as i64 => { + to_update.push(ActiveModel::from(item)); + if item.music_metadata.is_some() { + to_save_music.push(ino_i64); + } + } + _ => {} + } + } + + let fresh_inodes: HashSet = snapshot.keys().map(|i| i.0 as i64).collect(); + for ino in db_items.keys().filter(|i| !fresh_inodes.contains(i)) { + to_delete.push(*ino); + } + + if !to_insert.is_empty() { + Entity::insert_many(to_insert).exec(client).await.unwrap(); + } + for model in to_update { + model.update(client).await.unwrap(); + } + for ino in to_delete { + Entity::delete_by_id(ino).exec(client).await.unwrap(); + } + + for ino_i64 in &to_save_music { + let ino = INodeNo(*ino_i64 as u64); + if let Some(music_metadata) = snapshot + .get(&ino) + .and_then(|item| item.music_metadata.as_ref()) + { + save_music_metadata(*ino_i64, music_metadata, client) + .await + .unwrap(); + } + } +} + +pub fn run_db_blocking(future: F) -> F::Output { + return tokio::runtime::Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .unwrap() + .block_on(future); +} diff --git a/src/file_watcher.rs b/src/file_watcher.rs deleted file mode 100644 index 2c1ddff..0000000 --- a/src/file_watcher.rs +++ /dev/null @@ -1,3 +0,0 @@ -pub trait FileWatcher { - fn watch(&self); -} diff --git a/src/item.rs b/src/item.rs new file mode 100644 index 0000000..4a447b3 --- /dev/null +++ b/src/item.rs @@ -0,0 +1,84 @@ +use std::{fs, hash::Hasher, path::PathBuf, time::SystemTime}; + +use std::os::unix::ffi::OsStrExt; +use std::os::unix::fs::MetadataExt; + +use fuser::INodeNo; +use twox_hash::XxHash64; + +use crate::music::metadata::MusicMetadata; + +#[derive(PartialEq, Eq, Copy, Clone, Debug)] +pub enum FileType { + Directory, + File, +} + +#[derive(Debug)] +pub struct Item { + pub inode: INodeNo, + pub parent_inode: INodeNo, + pub name: String, + pub original_path: PathBuf, + pub local_path: PathBuf, + pub file_type: FileType, + pub metadata: fs::Metadata, + pub music_metadata: Option, + pub hash: u64, +} + +impl Item { + #[allow(clippy::too_many_arguments)] + pub fn new( + inode: INodeNo, + parent_inode: INodeNo, + name: String, + original_path: PathBuf, + local_path: PathBuf, + file_type: FileType, + metadata: fs::Metadata, + music_metadata: Option, + ) -> Item { + let mut item = Item { + inode, + parent_inode, + name, + original_path, + local_path, + file_type, + metadata, + music_metadata, + hash: 0, + }; + item.hash = item.compute_hash(); + + return item; + } + + pub fn compute_hash(&self) -> u64 { + let seed = 1234; + let mut hasher = XxHash64::with_seed(seed); + + let metadata = &self.metadata; + let mtime: u64 = metadata + .modified() + .unwrap_or(SystemTime::UNIX_EPOCH) + .duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let crtime: u64 = metadata + .created() + .unwrap_or(SystemTime::UNIX_EPOCH) + .duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + + hasher.write_u64(self.inode.0); + hasher.write(self.original_path.as_os_str().as_bytes()); + hasher.write_u64(metadata.ctime() as u64); + hasher.write_u64(mtime); + hasher.write_u64(crtime); + + return hasher.finish(); + } +} diff --git a/src/local.rs b/src/local.rs deleted file mode 100644 index 8440f49..0000000 --- a/src/local.rs +++ /dev/null @@ -1,1196 +0,0 @@ -use std::{ - collections::BTreeMap, - fs, - hash::Hasher, - io::{self, Read, Seek, SeekFrom}, - os::unix::{ - ffi::OsStrExt, - fs::{MetadataExt, PermissionsExt}, - }, - path::{Path, PathBuf}, - sync::{Arc, Mutex, mpsc}, - thread, - time::{Duration, SystemTime}, -}; - -use fuser::{Errno, FileAttr, Filesystem, Generation, INodeNo, Request}; -use notify::{Event, EventKind, RecursiveMode, Watcher}; -use sea_orm::entity::prelude::*; -use twox_hash::XxHash64; - -use crate::file_watcher::FileWatcher; -use crate::music_metadata::MusicMetadata; -use crate::music_metadata::db::{ - artists, music_metadata as music_metadata_entity, other_tags, pictures, -}; - -#[derive(PartialEq, Eq, Copy, Clone, Debug)] -enum FileType { - Directory, - File, -} - -#[derive(Debug)] -struct LocalItem { - inode: INodeNo, - parent_inode: INodeNo, - name: String, - original_path: PathBuf, - local_path: PathBuf, - file_type: FileType, - metadata: fs::Metadata, - music_metadata: Option, - hash: u64, -} - -impl LocalItem { - #[allow(clippy::too_many_arguments)] - fn new( - inode: INodeNo, - parent_inode: INodeNo, - name: String, - original_path: PathBuf, - local_path: PathBuf, - file_type: FileType, - metadata: fs::Metadata, - music_metadata: Option, - ) -> LocalItem { - let mut item = LocalItem { - inode, - parent_inode, - name, - original_path, - local_path, - file_type, - metadata, - music_metadata, - hash: 0, - }; - item.hash = item.compute_hash(); - - return item; - } - - fn compute_hash(&self) -> u64 { - let seed = 1234; - let mut hasher = XxHash64::with_seed(seed); - - let metadata = &self.metadata; - let mtime: u64 = metadata - .modified() - .unwrap_or(SystemTime::UNIX_EPOCH) - .duration_since(SystemTime::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0); - let crtime: u64 = metadata - .created() - .unwrap_or(SystemTime::UNIX_EPOCH) - .duration_since(SystemTime::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0); - - hasher.write_u64(self.inode.0); - hasher.write(self.original_path.as_os_str().as_bytes()); - hasher.write_u64(metadata.ctime() as u64); - hasher.write_u64(mtime); - hasher.write_u64(crtime); - - return hasher.finish(); - } -} - -pub struct LocalOrigin { - source: PathBuf, - destination: PathBuf, - files: Arc>>, - - client: sea_orm::DatabaseConnection, -} - -impl std::fmt::Debug for LocalOrigin { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - return f - .debug_struct("LocalOrigin") - .field("source", &self.source) - .field("destination", &self.destination) - .field("files", &self.files) - .finish(); - } -} - -impl LocalOrigin { - pub async fn new( - source: String, - destination: String, - client: sea_orm::DatabaseConnection, - ) -> Result { - println!("Initializing the LocalOrigin"); - - let mut snapshot = - LocalOrigin::build_snapshot(Path::new(&source), Path::new(&destination))?; - - let db_items: std::collections::HashMap = Entity::find() - .all(&client) - .await - .unwrap() - .into_iter() - .map(|e| (e.inode, e)) - .collect(); - - LocalOrigin::sync_items_to_db(&snapshot, &db_items, &client).await; - LocalOrigin::restore_music_metadata_from_db(&mut snapshot, &db_items, &client).await; - LocalOrigin::restore_virtual_paths(&mut snapshot, &db_items, Path::new(&source)); - - let local_origin = LocalOrigin { - source: source.into(), - destination: destination.into(), - files: Arc::new(Mutex::new(snapshot)), - client, - }; - //dbg!(&local_origin); - return Ok(local_origin); - } - - async fn sync_items_to_db( - snapshot: &BTreeMap, - db_items: &std::collections::HashMap, - client: &sea_orm::DatabaseConnection, - ) { - let mut to_insert: Vec = vec![]; - let mut to_update: Vec = vec![]; - let mut to_delete: Vec = vec![]; - let mut to_save_music: Vec = vec![]; - - for (ino, item) in snapshot { - let ino_i64 = ino.0 as i64; - match db_items.get(&ino_i64) { - None => { - to_insert.push(ActiveModel::from(item)); - if item.music_metadata.is_some() { - to_save_music.push(ino_i64); - } - } - Some(db_item) if db_item.hash != item.hash as i64 => { - to_update.push(ActiveModel::from(item)); - if item.music_metadata.is_some() { - to_save_music.push(ino_i64); - } - } - _ => {} - } - } - - let fresh_inodes: std::collections::HashSet = - snapshot.keys().map(|i| i.0 as i64).collect(); - for ino in db_items.keys().filter(|i| !fresh_inodes.contains(i)) { - to_delete.push(*ino); - } - - if !to_insert.is_empty() { - Entity::insert_many(to_insert).exec(client).await.unwrap(); - } - for model in to_update { - model.update(client).await.unwrap(); - } - for ino in to_delete { - Entity::delete_by_id(ino).exec(client).await.unwrap(); - } - - for ino_i64 in &to_save_music { - let ino = INodeNo(*ino_i64 as u64); - if let Some(music_metadata) = snapshot - .get(&ino) - .and_then(|item| item.music_metadata.as_ref()) - { - LocalOrigin::save_music_metadata(*ino_i64, music_metadata, client) - .await - .unwrap(); - } - } - } - - async fn restore_music_metadata_from_db( - snapshot: &mut BTreeMap, - db_items: &std::collections::HashMap, - client: &sea_orm::DatabaseConnection, - ) { - use artists::Column as ArtCol; - use music_metadata_entity::Column as MmCol; - use other_tags::Column as OtCol; - use pictures::Column as PicCol; - - let unchanged_music_inodes: Vec = db_items - .values() - .filter_map(|db_item| { - let ino = INodeNo(db_item.inode as u64); - snapshot - .get(&ino) - .filter(|item| { - item.hash as i64 == db_item.hash && item.music_metadata.is_some() - }) - .map(|_| db_item.inode) - }) - .collect(); - - if unchanged_music_inodes.is_empty() { - return; - } - - let mm_rows: std::collections::HashMap = - music_metadata_entity::Entity::find() - .filter(MmCol::Inode.is_in(unchanged_music_inodes.clone())) - .all(client) - .await - .unwrap() - .into_iter() - .map(|m| (m.inode, m)) - .collect(); - - let mut artists_by_inode: std::collections::HashMap> = - std::collections::HashMap::new(); - for row in artists::Entity::find() - .filter(ArtCol::Inode.is_in(unchanged_music_inodes.clone())) - .all(client) - .await - .unwrap() - { - artists_by_inode - .entry(row.inode) - .or_default() - .push(row.artist); - } - - let mut other_tags_by_inode: std::collections::HashMap> = - std::collections::HashMap::new(); - for row in other_tags::Entity::find() - .filter(OtCol::Inode.is_in(unchanged_music_inodes.clone())) - .all(client) - .await - .unwrap() - { - other_tags_by_inode - .entry(row.inode) - .or_default() - .push((row.position, row.tag)); - } - - let mut pictures_by_inode: std::collections::HashMap> = - std::collections::HashMap::new(); - for row in pictures::Entity::find() - .filter(PicCol::Inode.is_in(unchanged_music_inodes)) - .all(client) - .await - .unwrap() - { - pictures_by_inode.entry(row.inode).or_default().push(row); - } - - for (inode, mm_row) in mm_rows { - let ino = INodeNo(inode as u64); - let Some(item) = snapshot.get_mut(&ino) else { - continue; - }; - - let mut sorted_tags = other_tags_by_inode.remove(&inode).unwrap_or_default(); - sorted_tags.sort_by_key(|(pos, _)| *pos); - - let mut sorted_pics = pictures_by_inode.remove(&inode).unwrap_or_default(); - sorted_pics.sort_by_key(|p| p.position); - - let picture_block_headers: Vec<[u8; 4]> = sorted_pics - .iter() - .map(|p| { - let mut hdr = [0u8; 4]; - let len = p.block_header.len().min(4); - hdr[..len].copy_from_slice(&p.block_header[..len]); - hdr - }) - .collect(); - - let picture_data_ranges: Vec<(u64, u64)> = sorted_pics - .iter() - .map(|p| (p.data_offset as u64, p.data_length as u64)) - .collect(); - - let mut music_metadata = MusicMetadata { - artist: artists_by_inode.remove(&inode).unwrap_or_default(), - album: mm_row.album, - track_number: mm_row.track_number, - track_title: mm_row.track_title, - other_tags: sorted_tags.into_iter().map(|(_, tag)| tag).collect(), - header: mm_row.header, - picture_block_headers, - picture_data_ranges, - real_audio_start: mm_row.real_audio_start as u64, - vorbis_comment_offset: 0, - vorbis_comment_length: 0, - }; - music_metadata.find_vorbis_offsets(); - item.music_metadata = Some(music_metadata); - } - } - - fn restore_virtual_paths( - snapshot: &mut BTreeMap, - db_items: &std::collections::HashMap, - source: &Path, - ) { - let restorations: Vec<(INodeNo, String, PathBuf)> = db_items - .values() - .filter_map(|db_item| { - let ino = INodeNo(db_item.inode as u64); - if ino == INodeNo::ROOT { - return None; - } - snapshot - .get(&ino) - .filter(|item| item.hash as i64 == db_item.hash) - .map(|_| { - ( - ino, - db_item.name.clone(), - PathBuf::from(&db_item.local_path), - ) - }) - }) - .collect(); - - for (_, _, local_path) in &restorations { - LocalOrigin::ensure_virtual_dirs(local_path, source, snapshot); - } - for (ino, name, local_path) in restorations { - if let Some(item) = snapshot.get_mut(&ino) { - item.name = name; - item.parent_inode = LocalOrigin::parent_inode_from_path(&local_path); - item.local_path = local_path; - } - } - } - - async fn save_music_metadata( - inode: i64, - music_metadata: &MusicMetadata, - client: &sea_orm::DatabaseConnection, - ) -> Result<(), sea_orm::DbErr> { - use sea_orm::ActiveValue::Set; - - music_metadata_entity::Entity::delete_by_id(inode) - .exec(client) - .await?; - - music_metadata_entity::Entity::insert(music_metadata_entity::ActiveModel { - inode: Set(inode), - track_title: Set(music_metadata.track_title.clone()), - album: Set(music_metadata.album.clone()), - track_number: Set(music_metadata.track_number), - header: Set(music_metadata.header.clone()), - real_audio_start: Set(music_metadata.real_audio_start as i64), - }) - .exec(client) - .await?; - - if !music_metadata.artist.is_empty() { - artists::Entity::insert_many(music_metadata.artist.iter().map(|a| { - artists::ActiveModel { - inode: Set(inode), - artist: Set(a.clone()), - } - })) - .exec(client) - .await?; - } - - if !music_metadata.other_tags.is_empty() { - other_tags::Entity::insert_many(music_metadata.other_tags.iter().enumerate().map( - |(pos, tag)| other_tags::ActiveModel { - inode: Set(inode), - position: Set(pos as i32), - tag: Set(tag.clone()), - }, - )) - .exec(client) - .await?; - } - - if !music_metadata.picture_block_headers.is_empty() { - pictures::Entity::insert_many( - music_metadata - .picture_block_headers - .iter() - .zip(music_metadata.picture_data_ranges.iter()) - .enumerate() - .map(|(pos, (hdr, (offset, len)))| pictures::ActiveModel { - inode: Set(inode), - position: Set(pos as i32), - block_header: Set(hdr.to_vec()), - data_offset: Set(*offset as i64), - data_length: Set(*len as i64), - }), - ) - .exec(client) - .await?; - } - - return Ok(()); - } - - fn fill_fileset( - map: &Arc>>, - source: &Path, - destination: &Path, - ) { - match LocalOrigin::build_snapshot(source, destination) { - Ok(new_snapshot) => { - let mut files = map.lock().unwrap(); - - files.retain(|ino, _| new_snapshot.contains_key(ino)); - - for (ino, new_item) in new_snapshot { - match files.get(&ino) { - Some(existing) if existing.hash == new_item.hash => {} - _ => { - files.insert(ino, new_item); - } - } - } - } - Err(e) => eprintln!("Error while reading source: {e}"), - } - } - - fn build_snapshot( - source: &Path, - destination: &Path, - ) -> Result, io::Error> { - let mut map = BTreeMap::new(); - - let local_root = LocalItem::new( - INodeNo::ROOT, - INodeNo::ROOT, - "/".to_string(), - source.to_path_buf(), - destination.to_path_buf(), - FileType::Directory, - fs::metadata(source)?, - None, - ); - map.insert(INodeNo::ROOT, local_root); - - LocalOrigin::read_into_map(source, destination, &mut map)?; - return Ok(map); - } - - fn read_into_map( - source: &Path, - destination: &Path, - map: &mut BTreeMap, - ) -> Result<(), io::Error> { - for item in fs::read_dir(source)? { - let entry = item?; - let name = entry.file_name().to_string_lossy().into_owned(); - let item_path = entry.path(); - let metadata = entry.metadata()?; - let file_type = if entry.file_type()?.is_dir() { - FileType::Directory - } else { - FileType::File - }; - let music_metadata = - if item_path.extension().and_then(std::ffi::OsStr::to_str) == Some("flac") { - MusicMetadata::parse_music_metadata(&item_path) - } else { - None - }; - - let mut local_path = PathBuf::new(); - if music_metadata.is_some() { - let music_metadata = music_metadata.clone().unwrap(); - local_path.push(music_metadata.artist.join("-")); - local_path.push(music_metadata.album); - } - local_path.push(name.clone()); - - let inode = INodeNo(metadata.ino()); - let parent_inode = LocalOrigin::ensure_virtual_dirs(&local_path, source, map); - let local_item = LocalItem::new( - inode, - parent_inode, - name, - item_path.clone(), - local_path, - file_type, - metadata, - music_metadata, - ); - - map.insert(inode, local_item); - - if file_type == FileType::Directory { - LocalOrigin::read_into_map(&item_path, destination, map)?; - } - } - return Ok(()); - } - - fn local_file_to_file_attr(local_file: &LocalItem) -> FileAttr { - let metadata = &local_file.metadata; - let ctime = std::time::UNIX_EPOCH + Duration::from_secs(metadata.ctime() as u64); - let atime = metadata.accessed().unwrap_or(SystemTime::UNIX_EPOCH); - let mtime = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH); - let crtime = metadata.created().unwrap_or(SystemTime::UNIX_EPOCH); - let kind = LocalOrigin::file_type(local_file.file_type); - - let size = match &local_file.music_metadata { - Some(mm) if mm.real_audio_start > 0 => mm.virtual_size(metadata.size()), - _ => metadata.size(), - }; - - return FileAttr { - ino: local_file.inode, - size, - blocks: metadata.blocks(), - atime, - mtime, - ctime, - crtime, - kind, - perm: metadata.permissions().mode() as u16, - nlink: metadata.nlink() as u32, - uid: metadata.uid(), - gid: metadata.gid(), - rdev: metadata.rdev() as u32, - blksize: metadata.blksize() as u32, - flags: 0, // TODO this is for macos only but perhaps we could get it from metadata still? - }; - } - - fn file_type(file_type: FileType) -> fuser::FileType { - match file_type { - FileType::Directory => fuser::FileType::Directory, - FileType::File => fuser::FileType::RegularFile, - } - } - - fn run_db_blocking(future: F) -> F::Output { - return tokio::runtime::Builder::new_current_thread() - .enable_io() - .enable_time() - .build() - .unwrap() - .block_on(future); - } - - fn read_bytes_at(path: &Path, offset: u64, len: usize) -> io::Result> { - let mut f = fs::File::open(path)?; - f.seek(SeekFrom::Start(offset))?; - let mut buf = vec![0u8; len]; - let n = f.read(&mut buf)?; - buf.truncate(n); - - return Ok(buf); - } - - fn assemble_flac_read( - original_path: &Path, - header: &[u8], - pic_hdrs: &[[u8; 4]], - pic_ranges: &[(u64, u64)], - real_audio_start: u64, - offset: u64, - size: u32, - ) -> io::Result> { - let end = offset + size as u64; - let header_end = header.len() as u64; - - // Pure header read — most common for tag readers. - if end <= header_end { - return Ok(header[offset as usize..end as usize].to_vec()); - } - - let mut buf = Vec::with_capacity(size as usize); - if offset < header_end { - buf.extend_from_slice(&header[offset as usize..]); - } - - let mut virt_pos = header_end; - for (pic_hdr, (data_real_offset, data_len)) in pic_hdrs.iter().zip(pic_ranges.iter()) { - let pic_hdr_end = virt_pos + 4; - let pic_end = pic_hdr_end + data_len; - - if end <= virt_pos { - break; - } - if offset >= pic_end { - virt_pos = pic_end; - continue; - } - - let hdr_from = (offset.max(virt_pos) - virt_pos) as usize; - let hdr_to = ((end.min(pic_hdr_end)) - virt_pos) as usize; - if hdr_from < hdr_to { - buf.extend_from_slice(&pic_hdr[hdr_from..hdr_to.min(4)]); - } - - let data_start = offset.max(pic_hdr_end); - let data_end = end.min(pic_end); - if data_start < data_end { - let real_off = data_real_offset + (data_start - pic_hdr_end); - let bytes = LocalOrigin::read_bytes_at( - original_path, - real_off, - (data_end - data_start) as usize, - )?; - buf.extend_from_slice(&bytes); - } - - virt_pos = pic_end; - } - - let audio_start = offset.max(virt_pos); - if audio_start < end { - let real_off = real_audio_start + (audio_start - virt_pos); - let bytes = - LocalOrigin::read_bytes_at(original_path, real_off, (end - audio_start) as usize)?; - buf.extend_from_slice(&bytes); - } - - return Ok(buf); - } - - fn parent_inode_from_path(local_path: &Path) -> INodeNo { - let components: Vec<_> = local_path - .components() - .filter(|c| matches!(c, std::path::Component::Normal(_))) - .collect(); - if components.len() <= 1 { - return INodeNo::ROOT; - } - let mut parent_path = String::new(); - for (i, comp) in components[..components.len() - 1].iter().enumerate() { - if i > 0 { - parent_path.push('/'); - } - parent_path.push_str(comp.as_os_str().to_str().unwrap_or_default()); - } - - return LocalOrigin::virtual_inode(&parent_path); - } - - fn virtual_inode(path: &str) -> INodeNo { - let mut hasher = XxHash64::with_seed(5678); - hasher.write(path.as_bytes()); - - return INodeNo(hasher.finish() | (1u64 << 63)); - } - - fn ensure_virtual_dirs( - local_path: &Path, - source: &Path, - map: &mut BTreeMap, - ) -> INodeNo { - let components: Vec<_> = local_path - .components() - .filter(|c| matches!(c, std::path::Component::Normal(_))) - .collect(); - - if components.len() <= 1 { - return INodeNo::ROOT; - } - - let mut current_parent = INodeNo::ROOT; - let mut current_path = String::new(); - - for component in &components[..components.len() - 1] { - let comp_str = component.as_os_str().to_str().unwrap_or_default(); - if !current_path.is_empty() { - current_path.push('/'); - } - current_path.push_str(comp_str); - - let virt_ino = LocalOrigin::virtual_inode(¤t_path); - - if !map.contains_key(&virt_ino) { - let virt_item = LocalItem::new( - virt_ino, - current_parent, - comp_str.to_string(), - source.to_path_buf(), - PathBuf::from(¤t_path), - FileType::Directory, - fs::metadata(source).unwrap(), - None, - ); - map.insert(virt_ino, virt_item); - } - - current_parent = virt_ino; - } - - return current_parent; - } -} - -impl Filesystem for LocalOrigin { - fn open( - &self, - _req: &Request, - ino: INodeNo, - _flags: fuser::OpenFlags, - reply: fuser::ReplyOpen, - ) { - if self.files.lock().unwrap().contains_key(&ino) { - reply.opened(fuser::FileHandle(ino.0), fuser::FopenFlags::empty()); - } else { - reply.error(Errno::ENOENT); - } - } - - fn setattr( - &self, - _req: &Request, - ino: INodeNo, - _mode: Option, - _uid: Option, - _gid: Option, - _size: Option, - _atime: Option, - _mtime: Option, - _ctime: Option, - _fh: Option, - _crtime: Option, - _chgtime: Option, - _bkuptime: Option, - _flags: Option, - reply: fuser::ReplyAttr, - ) { - match self.files.lock().unwrap().get(&ino) { - Some(file) => reply.attr( - &Duration::new(1, 0), - &LocalOrigin::local_file_to_file_attr(file), - ), - None => reply.error(Errno::ENOENT), - } - } - - fn write( - &self, - _req: &Request, - ino: INodeNo, - _fh: fuser::FileHandle, - offset: u64, - data: &[u8], - _write_flags: fuser::WriteFlags, - _flags: fuser::OpenFlags, - _lock_owner: Option, - reply: fuser::ReplyWrite, - ) { - let written = data.len() as u32; - let write_start = offset; - let write_end = write_start + data.len() as u64; - - let updated_music_metadata = { - let mut files = self.files.lock().unwrap(); - let item = match files.get_mut(&ino) { - Some(item) => item, - None => { - reply.written(written); - return; - } - }; - let music_metadata = match &mut item.music_metadata { - Some(mm) if mm.vorbis_comment_length > 0 => mm, - _ => { - reply.written(written); - return; - } - }; - - let vc_data_offset = music_metadata.vorbis_comment_offset; - let vc_hdr_offset = vc_data_offset - 4; - - // Must cover the 4-byte block header to read the (possibly new) data length - if write_start <= vc_hdr_offset && write_end >= vc_data_offset { - let hdr_from = (vc_hdr_offset - write_start) as usize; - let new_length = u32::from_be_bytes([ - 0, - data[hdr_from + 1], - data[hdr_from + 2], - data[hdr_from + 3], - ]) as u64; - let vc_data_end = vc_data_offset + new_length; - - if write_end >= vc_data_end { - let from = (vc_data_offset - write_start) as usize; - let to = (vc_data_end - write_start) as usize; - music_metadata.update_from_vorbis_comment_data(&data[from..to]); - Some(music_metadata.clone()) - } else { - None - } - } else { - None - } - }; - - if let Some(music_metadata) = updated_music_metadata { - let client = self.client.clone(); - let ino_i64 = ino.0 as i64; - LocalOrigin::run_db_blocking(async move { - LocalOrigin::save_music_metadata(ino_i64, &music_metadata, &client) - .await - .unwrap(); - }); - } - - reply.written(written); - } - - fn getattr( - &self, - _req: &fuser::Request, - ino: INodeNo, - _fh: Option, - reply: fuser::ReplyAttr, - ) { - println!("getattr(ino={})", ino); - // TODO fix unwrap - match self.files.lock().unwrap().get(&ino) { - Some(file) => { - let ttl = Duration::new(1, 0); - let attr = LocalOrigin::local_file_to_file_attr(file); - reply.attr(&ttl, &attr); - } - None => reply.error(Errno::ENOENT), - } - } - - fn readdir( - &self, - _req: &fuser::Request, - ino: INodeNo, - fh: fuser::FileHandle, - offset: u64, - mut reply: fuser::ReplyDirectory, - ) { - println!("readdir(ino={}, fh={}, offset={})", ino, fh, offset); - - let files = self.files.lock().unwrap(); - - let parent_inode = match files.get(&ino) { - Some(dir) => dir.parent_inode, - None => { - reply.error(Errno::ENOENT); - return; - } - }; - - // Offsets are 1-based cursors: "." = 1, ".." = 2, real entries = 3+. - // The kernel passes back the offset of the last entry it received; - // we return entries with offset > that value. - if offset < 1 { - if reply.add(ino, 1, fuser::FileType::Directory, ".") { - reply.ok(); - return; - } - } - - if offset < 2 { - if reply.add(parent_inode, 2, fuser::FileType::Directory, "..") { - reply.ok(); - return; - } - } - - let skip_count = if offset <= 2 { - 0 - } else { - (offset - 2) as usize - }; - - for (i, (key, value)) in files - .iter() - .filter(|(_, v)| v.parent_inode == ino && v.inode != ino) - .skip(skip_count) - .enumerate() - { - let entry_offset = (skip_count + i + 3) as u64; - let file_type: fuser::FileType = LocalOrigin::file_type(value.file_type); - if reply.add(*key, entry_offset, file_type, &value.name) { - break; - } - } - reply.ok(); - } - - fn rename( - &self, - _req: &Request, - parent: INodeNo, - name: &std::ffi::OsStr, - newparent: INodeNo, - newname: &std::ffi::OsStr, - _flags: fuser::RenameFlags, - reply: fuser::ReplyEmpty, - ) { - let name_str = match name.to_str() { - Some(s) => s, - None => { - reply.error(Errno::EINVAL); - return; - } - }; - let newname_str = match newname.to_str() { - Some(s) => s, - None => { - reply.error(Errno::EINVAL); - return; - } - }; - - let mut files = self.files.lock().unwrap(); - - let item_ino = match files - .iter() - .find(|(_, v)| v.parent_inode == parent && v.name == name_str) - { - Some((ino, _)) => *ino, - None => { - reply.error(Errno::ENOENT); - return; - } - }; - - let new_local_path = if newparent == INodeNo::ROOT { - PathBuf::from(newname_str) - } else { - match files.get(&newparent) { - Some(dir) => dir.local_path.join(newname_str), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - - let new_parent_inode = LocalOrigin::parent_inode_from_path(&new_local_path); - - let item = files.get_mut(&item_ino).unwrap(); - item.name = newname_str.to_string(); - item.local_path = new_local_path.clone(); - item.parent_inode = new_parent_inode; - - let inode_i64 = item_ino.0 as i64; - let new_name_owned = newname_str.to_string(); - let new_local_path_str = new_local_path.to_string_lossy().into_owned(); - let client = self.client.clone(); - - drop(files); - - LocalOrigin::run_db_blocking(async move { - use sea_orm::ActiveValue::Set; - ActiveModel { - inode: Set(inode_i64), - name: Set(new_name_owned), - local_path: Set(new_local_path_str), - ..Default::default() - } - .update(&client) - .await - .unwrap(); - }); - - reply.ok(); - } - - fn read( - &self, - _req: &Request, - ino: INodeNo, - _fh: fuser::FileHandle, - offset: u64, - size: u32, - _flags: fuser::OpenFlags, - _lock_owner: Option, - reply: fuser::ReplyData, - ) { - let files = self.files.lock().unwrap(); - let item = match files.get(&ino) { - Some(f) => f, - None => { - reply.error(Errno::ENOENT); - return; - } - }; - - let original_path = item.original_path.clone(); - let flac = item - .music_metadata - .as_ref() - .filter(|mm| mm.real_audio_start > 0) - .map(|mm| { - ( - mm.header.clone(), - mm.picture_block_headers.clone(), - mm.picture_data_ranges.clone(), - mm.real_audio_start, - ) - }); - drop(files); - - // Non-FLAC files are streamed straight from the real file. - let Some((header, pic_hdrs, pic_ranges, real_audio_start)) = flac else { - match LocalOrigin::read_bytes_at(&original_path, offset, size as usize) { - Ok(bytes) => reply.data(&bytes), - Err(_) => reply.error(Errno::EIO), - } - return; - }; - - match LocalOrigin::assemble_flac_read( - &original_path, - &header, - &pic_hdrs, - &pic_ranges, - real_audio_start, - offset, - size, - ) { - Ok(bytes) => reply.data(&bytes), - Err(_) => reply.error(Errno::EIO), - } - } - - fn lookup( - &self, - _req: &Request, - parent: INodeNo, - name: &std::ffi::OsStr, - reply: fuser::ReplyEntry, - ) { - println!("lookup(parent={}, name={})", parent, name.display()); - - match self - .files - .lock() - .unwrap() - .iter() - .find(|item| item.1.parent_inode == parent && item.1.name == name.to_str().unwrap()) - { - Some(item) => { - let ttl = Duration::new(1, 0); - let attr = LocalOrigin::local_file_to_file_attr(item.1); - - reply.entry(&ttl, &attr, Generation(0)); - } - None => reply.error(Errno::ENOENT), - } - } -} - -impl FileWatcher for LocalOrigin { - fn watch(&self) { - let files = self.files.clone(); - let source = self.source.clone(); - let destination = self.destination.clone(); - - println!("Starting to watch the source files in another thread"); - thread::spawn(move || { - let (tx, rx): ( - mpsc::Sender>, - mpsc::Receiver>, - ) = mpsc::channel(); - - // Use recommended_watcher() to automatically select the best implementation - // for your platform. The `EventHandler` passed to this constructor can be a - // closure, a `std::sync::mpsc::Sender`, a `crossbeam_channel::Sender`, or - // another type the trait is implemented for. - let mut watcher: notify::INotifyWatcher = notify::recommended_watcher(tx).unwrap(); - - // Add a path to be watched. All files and directories at that path and - // below will be monitored for changes. - watcher.watch(&source, RecursiveMode::Recursive).unwrap(); - // Block forever, printing out events as they come in - for res in rx { - match res { - Ok(event) => { - println!("event: {:?}", event); - - match event.kind { - EventKind::Any => { - println!("Something happened to item, ignoring"); - } - EventKind::Access(_access_kind) => { - println!("Item was read"); - } - EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => { - println!("Item was removed"); - LocalOrigin::fill_fileset(&files, &source, &destination); - } - EventKind::Other => { - println!("Some other action happened to item, ignoring"); - } - } - } - Err(e) => println!("watch error: {:?}", e), - } - } - }); - } -} - -#[derive(Clone, Debug, PartialEq, Eq, sea_orm::DeriveEntityModel)] -#[sea_orm(table_name = "items")] -pub struct Model { - #[sea_orm(primary_key, auto_increment = false)] - pub inode: i64, - pub name: String, - pub original_path: String, - pub local_path: String, - pub file_type: String, - pub hash: i64, -} - -pub type LocalItemEntity = Model; - -#[derive(Copy, Clone, Debug, sea_orm::EnumIter, sea_orm::DeriveRelation)] -pub enum Relation {} - -impl sea_orm::ActiveModelBehavior for ActiveModel {} - -impl From<&LocalItem> for ActiveModel { - fn from(item: &LocalItem) -> Self { - use sea_orm::ActiveValue::Set; - ActiveModel { - inode: Set(item.inode.0 as i64), - name: Set(item.name.clone()), - original_path: Set(item.original_path.to_string_lossy().into_owned()), - local_path: Set(item.local_path.to_string_lossy().into_owned()), - file_type: Set(match item.file_type { - FileType::Directory => "directory".to_string(), - FileType::File => "file".to_string(), - }), - hash: Set(item.hash as i64), - } - } -} - -impl From<(LocalItemEntity, fs::Metadata)> for LocalItem { - fn from((entity, metadata): (LocalItemEntity, fs::Metadata)) -> Self { - let local_path = PathBuf::from(&entity.local_path); - let parent_inode = LocalOrigin::parent_inode_from_path(&local_path); - let file_type = match entity.file_type.as_str() { - "directory" => FileType::Directory, - _ => FileType::File, - }; - LocalItem::new( - INodeNo(entity.inode as u64), - parent_inode, - entity.name, - PathBuf::from(entity.original_path), - local_path, - file_type, - metadata, - None, - ) - } -} diff --git a/src/main.rs b/src/main.rs index b801712..aed6239 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,12 +1,14 @@ use clap::Parser; -use file_watcher::FileWatcher; +use origins::local::watcher::FileWatcher; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -mod file_watcher; -mod local; -mod music_metadata; +mod db; +mod item; +mod music; +mod origins; +mod virtual_dirs; #[derive(Parser, Debug)] #[command(version, about, long_about = None)] @@ -37,13 +39,10 @@ async fn main() { let db = sea_orm::Database::connect(&args.database).await.unwrap(); - // TODO don't think clone is necessary - // TODO unwrap might not be safe and better would be to handle it properly - let fs = local::LocalOrigin::new(args.source.clone(), mountpoint.clone(), db) + let fs = origins::local::LocalOrigin::new(args.source.clone(), mountpoint.clone(), db) .await .unwrap(); - // start watching for changes in source files fs.watch(); let cfg = fuser::Config::default(); diff --git a/src/music/db.rs b/src/music/db.rs new file mode 100644 index 0000000..6c1e108 --- /dev/null +++ b/src/music/db.rs @@ -0,0 +1,191 @@ +use std::collections::{BTreeMap, HashMap}; + +use fuser::INodeNo; +use sea_orm::entity::prelude::*; + +use crate::db::entities::Model; +use crate::item::Item; +use crate::music::metadata::MusicMetadata; +use crate::music::metadata::db::{ + artists, music_metadata as music_metadata_entity, other_tags, pictures, +}; + +pub async fn save_music_metadata( + inode: i64, + music_metadata: &MusicMetadata, + client: &sea_orm::DatabaseConnection, +) -> Result<(), sea_orm::DbErr> { + use sea_orm::ActiveValue::Set; + + music_metadata_entity::Entity::delete_by_id(inode) + .exec(client) + .await?; + + music_metadata_entity::Entity::insert(music_metadata_entity::ActiveModel { + inode: Set(inode), + track_title: Set(music_metadata.track_title.clone()), + album: Set(music_metadata.album.clone()), + track_number: Set(music_metadata.track_number), + header: Set(music_metadata.header.clone()), + real_audio_start: Set(music_metadata.real_audio_start as i64), + }) + .exec(client) + .await?; + + if !music_metadata.artist.is_empty() { + artists::Entity::insert_many(music_metadata.artist.iter().map(|a| artists::ActiveModel { + inode: Set(inode), + artist: Set(a.clone()), + })) + .exec(client) + .await?; + } + + if !music_metadata.other_tags.is_empty() { + other_tags::Entity::insert_many(music_metadata.other_tags.iter().enumerate().map( + |(pos, tag)| other_tags::ActiveModel { + inode: Set(inode), + position: Set(pos as i32), + tag: Set(tag.clone()), + }, + )) + .exec(client) + .await?; + } + + if !music_metadata.picture_block_headers.is_empty() { + pictures::Entity::insert_many( + music_metadata + .picture_block_headers + .iter() + .zip(music_metadata.picture_data_ranges.iter()) + .enumerate() + .map(|(pos, (hdr, (offset, len)))| pictures::ActiveModel { + inode: Set(inode), + position: Set(pos as i32), + block_header: Set(hdr.to_vec()), + data_offset: Set(*offset as i64), + data_length: Set(*len as i64), + }), + ) + .exec(client) + .await?; + } + + return Ok(()); +} + +pub async fn restore_music_metadata_from_db( + snapshot: &mut BTreeMap, + db_items: &HashMap, + client: &sea_orm::DatabaseConnection, +) { + use artists::Column as ArtCol; + use music_metadata_entity::Column as MmCol; + use other_tags::Column as OtCol; + use pictures::Column as PicCol; + + let unchanged_music_inodes: Vec = db_items + .values() + .filter_map(|db_item| { + let ino = INodeNo(db_item.inode as u64); + snapshot + .get(&ino) + .filter(|item| item.hash as i64 == db_item.hash && item.music_metadata.is_some()) + .map(|_| db_item.inode) + }) + .collect(); + + if unchanged_music_inodes.is_empty() { + return; + } + + let mm_rows: HashMap = music_metadata_entity::Entity::find() + .filter(MmCol::Inode.is_in(unchanged_music_inodes.clone())) + .all(client) + .await + .unwrap() + .into_iter() + .map(|m| (m.inode, m)) + .collect(); + + let mut artists_by_inode: HashMap> = HashMap::new(); + for row in artists::Entity::find() + .filter(ArtCol::Inode.is_in(unchanged_music_inodes.clone())) + .all(client) + .await + .unwrap() + { + artists_by_inode + .entry(row.inode) + .or_default() + .push(row.artist); + } + + let mut other_tags_by_inode: HashMap> = HashMap::new(); + for row in other_tags::Entity::find() + .filter(OtCol::Inode.is_in(unchanged_music_inodes.clone())) + .all(client) + .await + .unwrap() + { + other_tags_by_inode + .entry(row.inode) + .or_default() + .push((row.position, row.tag)); + } + + let mut pictures_by_inode: HashMap> = HashMap::new(); + for row in pictures::Entity::find() + .filter(PicCol::Inode.is_in(unchanged_music_inodes)) + .all(client) + .await + .unwrap() + { + pictures_by_inode.entry(row.inode).or_default().push(row); + } + + for (inode, mm_row) in mm_rows { + let ino = INodeNo(inode as u64); + let Some(item) = snapshot.get_mut(&ino) else { + continue; + }; + + let mut sorted_tags = other_tags_by_inode.remove(&inode).unwrap_or_default(); + sorted_tags.sort_by_key(|(pos, _)| *pos); + + let mut sorted_pics = pictures_by_inode.remove(&inode).unwrap_or_default(); + sorted_pics.sort_by_key(|p| p.position); + + let picture_block_headers: Vec<[u8; 4]> = sorted_pics + .iter() + .map(|p| { + let mut hdr = [0u8; 4]; + let len = p.block_header.len().min(4); + hdr[..len].copy_from_slice(&p.block_header[..len]); + hdr + }) + .collect(); + + let picture_data_ranges: Vec<(u64, u64)> = sorted_pics + .iter() + .map(|p| (p.data_offset as u64, p.data_length as u64)) + .collect(); + + let mut music_metadata = MusicMetadata { + artist: artists_by_inode.remove(&inode).unwrap_or_default(), + album: mm_row.album, + track_number: mm_row.track_number, + track_title: mm_row.track_title, + other_tags: sorted_tags.into_iter().map(|(_, tag)| tag).collect(), + header: mm_row.header, + picture_block_headers, + picture_data_ranges, + real_audio_start: mm_row.real_audio_start as u64, + vorbis_comment_offset: 0, + vorbis_comment_length: 0, + }; + music_metadata.find_vorbis_offsets(); + item.music_metadata = Some(music_metadata); + } +} diff --git a/src/music_metadata.rs b/src/music/metadata.rs similarity index 100% rename from src/music_metadata.rs rename to src/music/metadata.rs diff --git a/src/music/mod.rs b/src/music/mod.rs new file mode 100644 index 0000000..ab71815 --- /dev/null +++ b/src/music/mod.rs @@ -0,0 +1,2 @@ +pub mod db; +pub mod metadata; diff --git a/src/origins/local/file_io.rs b/src/origins/local/file_io.rs new file mode 100644 index 0000000..e4792f2 --- /dev/null +++ b/src/origins/local/file_io.rs @@ -0,0 +1,77 @@ +use std::{ + fs, + io::{self, Read, Seek, SeekFrom}, + path::Path, +}; + +pub fn read_bytes_at(path: &Path, offset: u64, len: usize) -> io::Result> { + let mut f = fs::File::open(path)?; + f.seek(SeekFrom::Start(offset))?; + let mut buf = vec![0u8; len]; + let n = f.read(&mut buf)?; + buf.truncate(n); + + return Ok(buf); +} + +pub fn assemble_flac_read( + original_path: &Path, + header: &[u8], + pic_hdrs: &[[u8; 4]], + pic_ranges: &[(u64, u64)], + real_audio_start: u64, + offset: u64, + size: u32, +) -> io::Result> { + let end = offset + size as u64; + let header_end = header.len() as u64; + + // Pure header read — most common for tag readers. + if end <= header_end { + return Ok(header[offset as usize..end as usize].to_vec()); + } + + let mut buf = Vec::with_capacity(size as usize); + if offset < header_end { + buf.extend_from_slice(&header[offset as usize..]); + } + + let mut virt_pos = header_end; + for (pic_hdr, (data_real_offset, data_len)) in pic_hdrs.iter().zip(pic_ranges.iter()) { + let pic_hdr_end = virt_pos + 4; + let pic_end = pic_hdr_end + data_len; + + if end <= virt_pos { + break; + } + if offset >= pic_end { + virt_pos = pic_end; + continue; + } + + let hdr_from = (offset.max(virt_pos) - virt_pos) as usize; + let hdr_to = ((end.min(pic_hdr_end)) - virt_pos) as usize; + if hdr_from < hdr_to { + buf.extend_from_slice(&pic_hdr[hdr_from..hdr_to.min(4)]); + } + + let data_start = offset.max(pic_hdr_end); + let data_end = end.min(pic_end); + if data_start < data_end { + let real_off = data_real_offset + (data_start - pic_hdr_end); + let bytes = read_bytes_at(original_path, real_off, (data_end - data_start) as usize)?; + buf.extend_from_slice(&bytes); + } + + virt_pos = pic_end; + } + + let audio_start = offset.max(virt_pos); + if audio_start < end { + let real_off = real_audio_start + (audio_start - virt_pos); + let bytes = read_bytes_at(original_path, real_off, (end - audio_start) as usize)?; + buf.extend_from_slice(&bytes); + } + + return Ok(buf); +} diff --git a/src/origins/local/mod.rs b/src/origins/local/mod.rs new file mode 100644 index 0000000..6615fbe --- /dev/null +++ b/src/origins/local/mod.rs @@ -0,0 +1,468 @@ +pub mod file_io; +pub mod snapshot; +pub mod watcher; + +use std::{ + collections::BTreeMap, + io, + path::{Path, PathBuf}, + sync::{Arc, Mutex}, + time::Duration, +}; + +use fuser::{Errno, FileAttr, Filesystem, Generation, INodeNo, Request}; +use sea_orm::entity::prelude::*; + +use crate::db::entities::{ActiveModel, Entity, Model}; +use crate::db::sync::{run_db_blocking, sync_items_to_db}; +use crate::item::{FileType, Item}; +use crate::music::db::{restore_music_metadata_from_db, save_music_metadata}; +use crate::virtual_dirs::{parent_inode_from_path, restore_virtual_paths}; + +use std::os::unix::fs::{MetadataExt, PermissionsExt}; + +pub struct LocalOrigin { + pub(crate) source: PathBuf, + pub(crate) destination: PathBuf, + pub(crate) files: Arc>>, + client: sea_orm::DatabaseConnection, +} + +impl std::fmt::Debug for LocalOrigin { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return f + .debug_struct("LocalOrigin") + .field("source", &self.source) + .field("destination", &self.destination) + .field("files", &self.files) + .finish(); + } +} + +impl LocalOrigin { + pub async fn new( + source: String, + destination: String, + client: sea_orm::DatabaseConnection, + ) -> Result { + println!("Initializing the LocalOrigin"); + + let mut snapshot = snapshot::build_snapshot(Path::new(&source), Path::new(&destination))?; + + let db_items: std::collections::HashMap = Entity::find() + .all(&client) + .await + .unwrap() + .into_iter() + .map(|e| (e.inode, e)) + .collect(); + + sync_items_to_db(&snapshot, &db_items, &client).await; + restore_music_metadata_from_db(&mut snapshot, &db_items, &client).await; + restore_virtual_paths(&mut snapshot, &db_items, Path::new(&source)); + + let local_origin = LocalOrigin { + source: source.into(), + destination: destination.into(), + files: Arc::new(Mutex::new(snapshot)), + client, + }; + return Ok(local_origin); + } + + fn file_to_attr(item: &Item) -> FileAttr { + let metadata = &item.metadata; + let ctime = std::time::UNIX_EPOCH + Duration::from_secs(metadata.ctime() as u64); + let atime = metadata + .accessed() + .unwrap_or(std::time::SystemTime::UNIX_EPOCH); + let mtime = metadata + .modified() + .unwrap_or(std::time::SystemTime::UNIX_EPOCH); + let crtime = metadata + .created() + .unwrap_or(std::time::SystemTime::UNIX_EPOCH); + let kind = match item.file_type { + FileType::Directory => fuser::FileType::Directory, + FileType::File => fuser::FileType::RegularFile, + }; + + let size = match &item.music_metadata { + Some(mm) if mm.real_audio_start > 0 => mm.virtual_size(metadata.size()), + _ => metadata.size(), + }; + + return FileAttr { + ino: item.inode, + size, + blocks: metadata.blocks(), + atime, + mtime, + ctime, + crtime, + kind, + perm: metadata.permissions().mode() as u16, + nlink: metadata.nlink() as u32, + uid: metadata.uid(), + gid: metadata.gid(), + rdev: metadata.rdev() as u32, + blksize: metadata.blksize() as u32, + flags: 0, + }; + } +} + +impl Filesystem for LocalOrigin { + fn open( + &self, + _req: &Request, + ino: INodeNo, + _flags: fuser::OpenFlags, + reply: fuser::ReplyOpen, + ) { + if self.files.lock().unwrap().contains_key(&ino) { + reply.opened(fuser::FileHandle(ino.0), fuser::FopenFlags::empty()); + } else { + reply.error(Errno::ENOENT); + } + } + + fn setattr( + &self, + _req: &Request, + ino: INodeNo, + _mode: Option, + _uid: Option, + _gid: Option, + _size: Option, + _atime: Option, + _mtime: Option, + _ctime: Option, + _fh: Option, + _crtime: Option, + _chgtime: Option, + _bkuptime: Option, + _flags: Option, + reply: fuser::ReplyAttr, + ) { + match self.files.lock().unwrap().get(&ino) { + Some(file) => reply.attr(&Duration::new(1, 0), &LocalOrigin::file_to_attr(file)), + None => reply.error(Errno::ENOENT), + } + } + + fn write( + &self, + _req: &Request, + ino: INodeNo, + _fh: fuser::FileHandle, + offset: u64, + data: &[u8], + _write_flags: fuser::WriteFlags, + _flags: fuser::OpenFlags, + _lock_owner: Option, + reply: fuser::ReplyWrite, + ) { + let written = data.len() as u32; + let write_start = offset; + let write_end = write_start + data.len() as u64; + + let updated_music_metadata = { + let mut files = self.files.lock().unwrap(); + let item = match files.get_mut(&ino) { + Some(item) => item, + None => { + reply.written(written); + return; + } + }; + let music_metadata = match &mut item.music_metadata { + Some(mm) if mm.vorbis_comment_length > 0 => mm, + _ => { + reply.written(written); + return; + } + }; + + let vc_data_offset = music_metadata.vorbis_comment_offset; + let vc_hdr_offset = vc_data_offset - 4; + + // Must cover the 4-byte block header to read the (possibly new) data length + if write_start <= vc_hdr_offset && write_end >= vc_data_offset { + let hdr_from = (vc_hdr_offset - write_start) as usize; + let new_length = u32::from_be_bytes([ + 0, + data[hdr_from + 1], + data[hdr_from + 2], + data[hdr_from + 3], + ]) as u64; + let vc_data_end = vc_data_offset + new_length; + + if write_end >= vc_data_end { + let from = (vc_data_offset - write_start) as usize; + let to = (vc_data_end - write_start) as usize; + music_metadata.update_from_vorbis_comment_data(&data[from..to]); + Some(music_metadata.clone()) + } else { + None + } + } else { + None + } + }; + + if let Some(music_metadata) = updated_music_metadata { + let client = self.client.clone(); + let ino_i64 = ino.0 as i64; + run_db_blocking(async move { + save_music_metadata(ino_i64, &music_metadata, &client) + .await + .unwrap(); + }); + } + + reply.written(written); + } + + fn getattr( + &self, + _req: &fuser::Request, + ino: INodeNo, + _fh: Option, + reply: fuser::ReplyAttr, + ) { + println!("getattr(ino={})", ino); + match self.files.lock().unwrap().get(&ino) { + Some(file) => { + let ttl = Duration::new(1, 0); + let attr = LocalOrigin::file_to_attr(file); + reply.attr(&ttl, &attr); + } + None => reply.error(Errno::ENOENT), + } + } + + fn readdir( + &self, + _req: &fuser::Request, + ino: INodeNo, + fh: fuser::FileHandle, + offset: u64, + mut reply: fuser::ReplyDirectory, + ) { + println!("readdir(ino={}, fh={}, offset={})", ino, fh, offset); + + let files = self.files.lock().unwrap(); + + let parent_inode = match files.get(&ino) { + Some(dir) => dir.parent_inode, + None => { + reply.error(Errno::ENOENT); + return; + } + }; + + // Offsets are 1-based cursors: "." = 1, ".." = 2, real entries = 3+. + // The kernel passes back the offset of the last entry it received; + // we return entries with offset > that value. + if offset < 1 { + if reply.add(ino, 1, fuser::FileType::Directory, ".") { + reply.ok(); + return; + } + } + + if offset < 2 { + if reply.add(parent_inode, 2, fuser::FileType::Directory, "..") { + reply.ok(); + return; + } + } + + let skip_count = if offset <= 2 { + 0 + } else { + (offset - 2) as usize + }; + + for (i, (key, value)) in files + .iter() + .filter(|(_, v)| v.parent_inode == ino && v.inode != ino) + .skip(skip_count) + .enumerate() + { + let entry_offset = (skip_count + i + 3) as u64; + let file_type = match value.file_type { + FileType::Directory => fuser::FileType::Directory, + FileType::File => fuser::FileType::RegularFile, + }; + if reply.add(*key, entry_offset, file_type, &value.name) { + break; + } + } + reply.ok(); + } + + fn rename( + &self, + _req: &Request, + parent: INodeNo, + name: &std::ffi::OsStr, + newparent: INodeNo, + newname: &std::ffi::OsStr, + _flags: fuser::RenameFlags, + reply: fuser::ReplyEmpty, + ) { + let name_str = match name.to_str() { + Some(s) => s, + None => { + reply.error(Errno::EINVAL); + return; + } + }; + let newname_str = match newname.to_str() { + Some(s) => s, + None => { + reply.error(Errno::EINVAL); + return; + } + }; + + let mut files = self.files.lock().unwrap(); + + let item_ino = match files + .iter() + .find(|(_, v)| v.parent_inode == parent && v.name == name_str) + { + Some((ino, _)) => *ino, + None => { + reply.error(Errno::ENOENT); + return; + } + }; + + let new_local_path = if newparent == INodeNo::ROOT { + PathBuf::from(newname_str) + } else { + match files.get(&newparent) { + Some(dir) => dir.local_path.join(newname_str), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + + let new_parent_inode = parent_inode_from_path(&new_local_path); + + let item = files.get_mut(&item_ino).unwrap(); + item.name = newname_str.to_string(); + item.local_path = new_local_path.clone(); + item.parent_inode = new_parent_inode; + + let inode_i64 = item_ino.0 as i64; + let new_name_owned = newname_str.to_string(); + let new_local_path_str = new_local_path.to_string_lossy().into_owned(); + let client = self.client.clone(); + + drop(files); + + run_db_blocking(async move { + use sea_orm::ActiveValue::Set; + ActiveModel { + inode: Set(inode_i64), + name: Set(new_name_owned), + local_path: Set(new_local_path_str), + ..Default::default() + } + .update(&client) + .await + .unwrap(); + }); + + reply.ok(); + } + + fn read( + &self, + _req: &Request, + ino: INodeNo, + _fh: fuser::FileHandle, + offset: u64, + size: u32, + _flags: fuser::OpenFlags, + _lock_owner: Option, + reply: fuser::ReplyData, + ) { + let files = self.files.lock().unwrap(); + let item = match files.get(&ino) { + Some(f) => f, + None => { + reply.error(Errno::ENOENT); + return; + } + }; + + let original_path = item.original_path.clone(); + let flac = item + .music_metadata + .as_ref() + .filter(|mm| mm.real_audio_start > 0) + .map(|mm| { + ( + mm.header.clone(), + mm.picture_block_headers.clone(), + mm.picture_data_ranges.clone(), + mm.real_audio_start, + ) + }); + drop(files); + + let Some((header, pic_hdrs, pic_ranges, real_audio_start)) = flac else { + match file_io::read_bytes_at(&original_path, offset, size as usize) { + Ok(bytes) => reply.data(&bytes), + Err(_) => reply.error(Errno::EIO), + } + return; + }; + + match file_io::assemble_flac_read( + &original_path, + &header, + &pic_hdrs, + &pic_ranges, + real_audio_start, + offset, + size, + ) { + Ok(bytes) => reply.data(&bytes), + Err(_) => reply.error(Errno::EIO), + } + } + + fn lookup( + &self, + _req: &Request, + parent: INodeNo, + name: &std::ffi::OsStr, + reply: fuser::ReplyEntry, + ) { + println!("lookup(parent={}, name={})", parent, name.display()); + + match self + .files + .lock() + .unwrap() + .iter() + .find(|item| item.1.parent_inode == parent && item.1.name == name.to_str().unwrap()) + { + Some(item) => { + let ttl = Duration::new(1, 0); + let attr = LocalOrigin::file_to_attr(item.1); + + reply.entry(&ttl, &attr, Generation(0)); + } + None => reply.error(Errno::ENOENT), + } + } +} diff --git a/src/origins/local/snapshot.rs b/src/origins/local/snapshot.rs new file mode 100644 index 0000000..c88ebd6 --- /dev/null +++ b/src/origins/local/snapshot.rs @@ -0,0 +1,107 @@ +use std::{ + collections::BTreeMap, + fs, io, + path::{Path, PathBuf}, + sync::{Arc, Mutex}, +}; + +use fuser::INodeNo; +use std::os::unix::fs::MetadataExt; + +use crate::item::{FileType, Item}; +use crate::music::metadata::MusicMetadata; +use crate::virtual_dirs::ensure_virtual_dirs; + +pub fn fill_fileset(map: &Arc>>, source: &Path, destination: &Path) { + match build_snapshot(source, destination) { + Ok(new_snapshot) => { + let mut files = map.lock().unwrap(); + + files.retain(|ino, _| new_snapshot.contains_key(ino)); + + for (ino, new_item) in new_snapshot { + match files.get(&ino) { + Some(existing) if existing.hash == new_item.hash => {} + _ => { + files.insert(ino, new_item); + } + } + } + } + Err(e) => eprintln!("Error while reading source: {e}"), + } +} + +pub fn build_snapshot( + source: &Path, + destination: &Path, +) -> Result, io::Error> { + let mut map = BTreeMap::new(); + + let local_root = Item::new( + INodeNo::ROOT, + INodeNo::ROOT, + "/".to_string(), + source.to_path_buf(), + destination.to_path_buf(), + FileType::Directory, + fs::metadata(source)?, + None, + ); + map.insert(INodeNo::ROOT, local_root); + + read_into_map(source, destination, &mut map)?; + return Ok(map); +} + +pub fn read_into_map( + source: &Path, + destination: &Path, + map: &mut BTreeMap, +) -> Result<(), io::Error> { + for item in fs::read_dir(source)? { + let entry = item?; + let name = entry.file_name().to_string_lossy().into_owned(); + let item_path = entry.path(); + let metadata = entry.metadata()?; + let file_type = if entry.file_type()?.is_dir() { + FileType::Directory + } else { + FileType::File + }; + let music_metadata = + if item_path.extension().and_then(std::ffi::OsStr::to_str) == Some("flac") { + MusicMetadata::parse_music_metadata(&item_path) + } else { + None + }; + + let mut local_path = PathBuf::new(); + if music_metadata.is_some() { + let music_metadata = music_metadata.clone().unwrap(); + local_path.push(music_metadata.artist.join("-")); + local_path.push(music_metadata.album); + } + local_path.push(name.clone()); + + let inode = INodeNo(metadata.ino()); + let parent_inode = ensure_virtual_dirs(&local_path, source, map); + let local_item = Item::new( + inode, + parent_inode, + name, + item_path.clone(), + local_path, + file_type, + metadata, + music_metadata, + ); + + map.insert(inode, local_item); + + if file_type == FileType::Directory { + read_into_map(&item_path, destination, map)?; + } + } + return Ok(()); +} diff --git a/src/origins/local/watcher.rs b/src/origins/local/watcher.rs new file mode 100644 index 0000000..09d171c --- /dev/null +++ b/src/origins/local/watcher.rs @@ -0,0 +1,53 @@ +use std::{sync::mpsc, thread}; + +use notify::{Event, EventKind, RecursiveMode, Watcher}; + +pub trait FileWatcher { + fn watch(&self); +} + +use super::LocalOrigin; + +impl FileWatcher for LocalOrigin { + fn watch(&self) { + let files = self.files.clone(); + let source = self.source.clone(); + let destination = self.destination.clone(); + + println!("Starting to watch the source files in another thread"); + thread::spawn(move || { + let (tx, rx): ( + mpsc::Sender>, + mpsc::Receiver>, + ) = mpsc::channel(); + + let mut watcher: notify::INotifyWatcher = notify::recommended_watcher(tx).unwrap(); + + watcher.watch(&source, RecursiveMode::Recursive).unwrap(); + for res in rx { + match res { + Ok(event) => { + println!("event: {:?}", event); + + match event.kind { + EventKind::Any => { + println!("Something happened to item, ignoring"); + } + EventKind::Access(_access_kind) => { + println!("Item was read"); + } + EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => { + println!("Item was removed"); + super::snapshot::fill_fileset(&files, &source, &destination); + } + EventKind::Other => { + println!("Some other action happened to item, ignoring"); + } + } + } + Err(e) => println!("watch error: {:?}", e), + } + } + }); + } +} diff --git a/src/origins/mod.rs b/src/origins/mod.rs new file mode 100644 index 0000000..2709962 --- /dev/null +++ b/src/origins/mod.rs @@ -0,0 +1 @@ +pub mod local; diff --git a/src/virtual_dirs.rs b/src/virtual_dirs.rs new file mode 100644 index 0000000..a3a38ee --- /dev/null +++ b/src/virtual_dirs.rs @@ -0,0 +1,121 @@ +use std::{ + collections::BTreeMap, + fs, + hash::Hasher, + path::{Path, PathBuf}, +}; + +use fuser::INodeNo; +use twox_hash::XxHash64; + +use crate::db::entities::Model; +use crate::item::{FileType, Item}; + +pub fn virtual_inode(path: &str) -> INodeNo { + let mut hasher = XxHash64::with_seed(5678); + hasher.write(path.as_bytes()); + + return INodeNo(hasher.finish() | (1u64 << 63)); +} + +pub fn parent_inode_from_path(local_path: &Path) -> INodeNo { + let components: Vec<_> = local_path + .components() + .filter(|c| matches!(c, std::path::Component::Normal(_))) + .collect(); + if components.len() <= 1 { + return INodeNo::ROOT; + } + let mut parent_path = String::new(); + for (i, comp) in components[..components.len() - 1].iter().enumerate() { + if i > 0 { + parent_path.push('/'); + } + parent_path.push_str(comp.as_os_str().to_str().unwrap_or_default()); + } + + return virtual_inode(&parent_path); +} + +pub fn ensure_virtual_dirs( + local_path: &Path, + source: &Path, + map: &mut BTreeMap, +) -> INodeNo { + let components: Vec<_> = local_path + .components() + .filter(|c| matches!(c, std::path::Component::Normal(_))) + .collect(); + + if components.len() <= 1 { + return INodeNo::ROOT; + } + + let mut current_parent = INodeNo::ROOT; + let mut current_path = String::new(); + + for component in &components[..components.len() - 1] { + let comp_str = component.as_os_str().to_str().unwrap_or_default(); + if !current_path.is_empty() { + current_path.push('/'); + } + current_path.push_str(comp_str); + + let virt_ino = virtual_inode(¤t_path); + + if !map.contains_key(&virt_ino) { + let virt_item = Item::new( + virt_ino, + current_parent, + comp_str.to_string(), + source.to_path_buf(), + PathBuf::from(¤t_path), + FileType::Directory, + fs::metadata(source).unwrap(), + None, + ); + map.insert(virt_ino, virt_item); + } + + current_parent = virt_ino; + } + + return current_parent; +} + +pub fn restore_virtual_paths( + snapshot: &mut BTreeMap, + db_items: &std::collections::HashMap, + source: &Path, +) { + let restorations: Vec<(INodeNo, String, PathBuf)> = db_items + .values() + .filter_map(|db_item| { + let ino = INodeNo(db_item.inode as u64); + if ino == INodeNo::ROOT { + return None; + } + snapshot + .get(&ino) + .filter(|item| item.hash as i64 == db_item.hash) + .map(|_| { + ( + ino, + db_item.name.clone(), + PathBuf::from(&db_item.local_path), + ) + }) + }) + .collect(); + + for (_, _, local_path) in &restorations { + ensure_virtual_dirs(local_path, source, snapshot); + } + for (ino, name, local_path) in restorations { + if let Some(item) = snapshot.get_mut(&ino) { + item.name = name; + item.parent_inode = parent_inode_from_path(&local_path); + item.local_path = local_path; + } + } +}