Integrate with db
This commit is contained in:
Generated
+2424
-13
File diff suppressed because it is too large
Load Diff
+3
-1
@@ -10,8 +10,10 @@ fuser = "0.17.0"
|
|||||||
libc = "0.2.186"
|
libc = "0.2.186"
|
||||||
notify = "8.2.0"
|
notify = "8.2.0"
|
||||||
time = "0.3.49"
|
time = "0.3.49"
|
||||||
|
|
||||||
symphonia = { version = "0.5", default-features = false, features = [
|
symphonia = { version = "0.5", default-features = false, features = [
|
||||||
"aac", "alac", "flac", "mp3", "ogg", "vorbis", "wav"
|
"aac", "alac", "flac", "mp3", "ogg", "vorbis", "wav"
|
||||||
] }
|
] }
|
||||||
twox-hash = "2.1.2"
|
twox-hash = "2.1.2"
|
||||||
|
|
||||||
|
sea-orm = { version = "2.0.0-rc", features = [ "sqlx-postgres", "runtime-tokio", "macros" ] }
|
||||||
|
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
|
||||||
|
|||||||
@@ -0,0 +1,20 @@
|
|||||||
|
CREATE TABLE items (
|
||||||
|
inode BIGINT PRIMARY KEY,
|
||||||
|
name TEXT NOT NULL,
|
||||||
|
original_path TEXT NOT NULL,
|
||||||
|
local_path TEXT NOT NULL,
|
||||||
|
file_type TEXT NOT NULL CHECK (file_type IN ('directory', 'file')),
|
||||||
|
hash BIGINT NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE music_metadata (
|
||||||
|
inode BIGINT PRIMARY KEY REFERENCES items(inode) ON DELETE CASCADE,
|
||||||
|
track_title TEXT NOT NULL,
|
||||||
|
album TEXT NOT NULL,
|
||||||
|
track_number INTEGER NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE music_metadata_artists (
|
||||||
|
inode BIGINT NOT NULL REFERENCES music_metadata(inode) ON DELETE CASCADE,
|
||||||
|
artist TEXT NOT NULL
|
||||||
|
);
|
||||||
+10
@@ -31,6 +31,16 @@
|
|||||||
opencode
|
opencode
|
||||||
];
|
];
|
||||||
|
|
||||||
|
services.postgres = {
|
||||||
|
enable = true;
|
||||||
|
initialDatabases = [
|
||||||
|
{
|
||||||
|
name = "musicfs";
|
||||||
|
schema = ./db/schema.sql;
|
||||||
|
}
|
||||||
|
];
|
||||||
|
};
|
||||||
|
|
||||||
outputs = {
|
outputs = {
|
||||||
rust-app = config.languages.rust.import ./. { };
|
rust-app = config.languages.rust.import ./. { };
|
||||||
};
|
};
|
||||||
|
|||||||
+130
-6
@@ -13,9 +13,9 @@ use std::{
|
|||||||
time::{Duration, SystemTime},
|
time::{Duration, SystemTime},
|
||||||
};
|
};
|
||||||
|
|
||||||
use clap::builder::OsStr;
|
|
||||||
use fuser::{Errno, FileAttr, Filesystem, Generation, INodeNo, Request};
|
use fuser::{Errno, FileAttr, Filesystem, Generation, INodeNo, Request};
|
||||||
use notify::{Event, EventKind, RecursiveMode, Watcher};
|
use notify::{Event, EventKind, RecursiveMode, Watcher};
|
||||||
|
use sea_orm::entity::prelude::*;
|
||||||
use twox_hash::XxHash64;
|
use twox_hash::XxHash64;
|
||||||
|
|
||||||
use crate::file_watcher::FileWatcher;
|
use crate::file_watcher::FileWatcher;
|
||||||
@@ -36,10 +36,11 @@ struct LocalItem {
|
|||||||
file_type: FileType,
|
file_type: FileType,
|
||||||
metadata: fs::Metadata,
|
metadata: fs::Metadata,
|
||||||
music_metadata: Option<MusicMetadata>,
|
music_metadata: Option<MusicMetadata>,
|
||||||
|
hash: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl LocalItem {
|
impl LocalItem {
|
||||||
fn hash(&self) -> u64 {
|
fn compute_hash(&self) -> u64 {
|
||||||
let seed = 1234;
|
let seed = 1234;
|
||||||
let mut hasher = XxHash64::with_seed(seed);
|
let mut hasher = XxHash64::with_seed(seed);
|
||||||
|
|
||||||
@@ -69,25 +70,80 @@ impl LocalItem {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
|
||||||
pub struct LocalOrigin {
|
pub struct LocalOrigin {
|
||||||
source: PathBuf,
|
source: PathBuf,
|
||||||
destination: PathBuf,
|
destination: PathBuf,
|
||||||
files: Arc<Mutex<BTreeMap<INodeNo, LocalItem>>>,
|
files: Arc<Mutex<BTreeMap<INodeNo, LocalItem>>>,
|
||||||
|
|
||||||
|
client: sea_orm::DatabaseConnection,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Debug for LocalOrigin {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
f.debug_struct("LocalOrigin")
|
||||||
|
.field("source", &self.source)
|
||||||
|
.field("destination", &self.destination)
|
||||||
|
.field("files", &self.files)
|
||||||
|
.finish()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl LocalOrigin {
|
impl LocalOrigin {
|
||||||
pub fn new(source: String, destination: String) -> Result<LocalOrigin, io::Error> {
|
pub async fn new(
|
||||||
|
source: String,
|
||||||
|
destination: String,
|
||||||
|
client: sea_orm::DatabaseConnection,
|
||||||
|
) -> Result<LocalOrigin, io::Error> {
|
||||||
println!("Initializing the LocalOrigin");
|
println!("Initializing the LocalOrigin");
|
||||||
|
|
||||||
let snapshot = LocalOrigin::build_snapshot(Path::new(&source), Path::new(&destination))?;
|
let snapshot = LocalOrigin::build_snapshot(Path::new(&source), Path::new(&destination))?;
|
||||||
|
|
||||||
|
let db_map: std::collections::HashMap<i64, i64> = Entity::find()
|
||||||
|
.all(&client)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.into_iter()
|
||||||
|
.map(|e| (e.inode, e.hash))
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
let mut to_insert: Vec<ActiveModel> = vec![];
|
||||||
|
let mut to_update: Vec<ActiveModel> = vec![];
|
||||||
|
let mut to_delete: Vec<i64> = vec![];
|
||||||
|
|
||||||
|
for (ino, item) in &snapshot {
|
||||||
|
let ino_i64 = ino.0 as i64;
|
||||||
|
match db_map.get(&ino_i64) {
|
||||||
|
None => to_insert.push(ActiveModel::from(item)),
|
||||||
|
Some(&db_hash) if db_hash != item.hash as i64 => {
|
||||||
|
to_update.push(ActiveModel::from(item))
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let fresh_inodes: std::collections::HashSet<i64> =
|
||||||
|
snapshot.keys().map(|i| i.0 as i64).collect();
|
||||||
|
for ino in db_map.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();
|
||||||
|
}
|
||||||
|
|
||||||
let local_origin = LocalOrigin {
|
let local_origin = LocalOrigin {
|
||||||
source: source.into(),
|
source: source.into(),
|
||||||
destination: destination.into(),
|
destination: destination.into(),
|
||||||
files: Arc::new(Mutex::new(snapshot)),
|
files: Arc::new(Mutex::new(snapshot)),
|
||||||
|
client,
|
||||||
};
|
};
|
||||||
dbg!(&local_origin);
|
//dbg!(&local_origin);
|
||||||
Ok(local_origin)
|
Ok(local_origin)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -104,7 +160,7 @@ impl LocalOrigin {
|
|||||||
|
|
||||||
for (ino, new_item) in new_snapshot {
|
for (ino, new_item) in new_snapshot {
|
||||||
match files.get(&ino) {
|
match files.get(&ino) {
|
||||||
Some(existing) if existing.hash() == new_item.hash() => {}
|
Some(existing) if existing.hash == new_item.hash => {}
|
||||||
_ => {
|
_ => {
|
||||||
files.insert(ino, new_item);
|
files.insert(ino, new_item);
|
||||||
}
|
}
|
||||||
@@ -129,6 +185,11 @@ impl LocalOrigin {
|
|||||||
file_type: FileType::Directory,
|
file_type: FileType::Directory,
|
||||||
metadata: fs::metadata(source)?,
|
metadata: fs::metadata(source)?,
|
||||||
music_metadata: None,
|
music_metadata: None,
|
||||||
|
hash: 0,
|
||||||
|
};
|
||||||
|
let local_root = LocalItem {
|
||||||
|
hash: local_root.compute_hash(),
|
||||||
|
..local_root
|
||||||
};
|
};
|
||||||
map.insert(INodeNo::ROOT, local_root);
|
map.insert(INodeNo::ROOT, local_root);
|
||||||
|
|
||||||
@@ -171,6 +232,11 @@ impl LocalOrigin {
|
|||||||
file_type,
|
file_type,
|
||||||
metadata,
|
metadata,
|
||||||
music_metadata,
|
music_metadata,
|
||||||
|
hash: 0,
|
||||||
|
};
|
||||||
|
let local_item = LocalItem {
|
||||||
|
hash: local_item.compute_hash(),
|
||||||
|
..local_item
|
||||||
};
|
};
|
||||||
|
|
||||||
map.insert(inode, local_item);
|
map.insert(inode, local_item);
|
||||||
@@ -366,3 +432,61 @@ impl FileWatcher for LocalOrigin {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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_item = LocalItem {
|
||||||
|
inode: INodeNo(entity.inode as u64),
|
||||||
|
name: entity.name,
|
||||||
|
original_path: PathBuf::from(entity.original_path),
|
||||||
|
local_path: PathBuf::from(entity.local_path),
|
||||||
|
file_type: match entity.file_type.as_str() {
|
||||||
|
"directory" => FileType::Directory,
|
||||||
|
_ => FileType::File,
|
||||||
|
},
|
||||||
|
metadata,
|
||||||
|
music_metadata: None,
|
||||||
|
hash: 0,
|
||||||
|
};
|
||||||
|
LocalItem {
|
||||||
|
hash: local_item.compute_hash(),
|
||||||
|
..local_item
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+10
-2
@@ -17,9 +17,13 @@ struct Args {
|
|||||||
|
|
||||||
#[arg(short, long, required = true)]
|
#[arg(short, long, required = true)]
|
||||||
source: String,
|
source: String,
|
||||||
|
|
||||||
|
#[arg(short, long, required = true)]
|
||||||
|
database: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn main() {
|
#[tokio::main]
|
||||||
|
async fn main() {
|
||||||
let args = Args::parse();
|
let args = Args::parse();
|
||||||
let mountpoint = args.mountpoint;
|
let mountpoint = args.mountpoint;
|
||||||
|
|
||||||
@@ -32,9 +36,13 @@ fn main() {
|
|||||||
})
|
})
|
||||||
.expect("Error setting Ctrl+C handler");
|
.expect("Error setting Ctrl+C handler");
|
||||||
|
|
||||||
|
let db = sea_orm::Database::connect(&args.database).await.unwrap();
|
||||||
|
|
||||||
// TODO don't think clone is necessary
|
// TODO don't think clone is necessary
|
||||||
// TODO unwrap might not be safe and better would be to handle it properly
|
// TODO unwrap might not be safe and better would be to handle it properly
|
||||||
let fs = local::LocalOrigin::new(args.source.clone(), mountpoint.clone()).unwrap();
|
let fs = local::LocalOrigin::new(args.source.clone(), mountpoint.clone(), db)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
// start watching for changes in source files
|
// start watching for changes in source files
|
||||||
fs.watch();
|
fs.watch();
|
||||||
|
|||||||
Reference in New Issue
Block a user