From 4b7756c36e83943e8da42b423e3521666c73448c Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Mon, 20 Jul 2026 18:05:16 +0300 Subject: [PATCH] Extend DHT scheme with content_id --- Cargo.lock | 19 ++++++------- Cargo.toml | 1 + src/federation/mod.rs | 52 +++++++++++++++++++++++++++++++++++- src/federation/serve.rs | 59 +++++++++++++++++++++++++++-------------- 4 files changed, 101 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4989eef..6387b51 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -690,9 +690,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.6.2" +version = "4.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd059f9da4f5c36b3787f65d38ccaab1cc315f07b01f89abc8359ee6a8205011" +checksum = "0fb99565819980999fb7b4a1796046a5c949e6d4ff132cf5fadf5a641e20d776" dependencies = [ "clap_builder", "clap_derive", @@ -712,9 +712,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.6.1" +version = "4.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +checksum = "32f2392eae7f16557a3d727ef3a12e57b2b2ca6f98566a5f4fb41ffe305df077" dependencies = [ "heck", "proc-macro2", @@ -1717,7 +1717,7 @@ dependencies = [ [[package]] name = "federation-net" version = "0.1.0" -source = "git+https://gt.hexor.cy/ab/frid.git#a897737978d476c223b66e65b960b70376084f2a" +source = "git+https://gt.hexor.cy/ab/frid.git#8ee1db9cf89ea604c6b8a8c3fe089a714c1e321f" dependencies = [ "blake3", "data-encoding", @@ -1844,11 +1844,12 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "furumusic" -version = "0.6.2-fd" +version = "0.6.3-fd" dependencies = [ "anyhow", "async-trait", "base64 0.22.1", + "blake3", "chrono", "cot", "croner", @@ -2470,9 +2471,9 @@ dependencies = [ [[package]] name = "hyper" -version = "1.10.1" +version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" +checksum = "d22053281f852e11534f5198498373cbb59295120a20771d90f7ed1897490a72" dependencies = [ "atomic-waker", "bytes", @@ -3621,7 +3622,7 @@ dependencies = [ [[package]] name = "music-dht" version = "0.1.0" -source = "git+https://gt.hexor.cy/ab/frid.git#a897737978d476c223b66e65b960b70376084f2a" +source = "git+https://gt.hexor.cy/ab/frid.git#8ee1db9cf89ea604c6b8a8c3fe089a714c1e321f" dependencies = [ "async-trait", "blake3", diff --git a/Cargo.toml b/Cargo.toml index 033f73c..5b032a3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls" tokio = { version = "1", features = ["sync", "fs", "io-util"] } tower = "0.5" base64 = "0.22" +blake3 = "1" serde_json = "1" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/src/federation/mod.rs b/src/federation/mod.rs index 79e5041..abe0076 100644 --- a/src/federation/mod.rs +++ b/src/federation/mod.rs @@ -46,6 +46,8 @@ pub struct Federation { /// Transport data directory; server-side DHT state and identity live in PostgreSQL. data_dir: PathBuf, database_url: std::sync::Mutex, + storage_dir: std::sync::Mutex, + content_cache: std::sync::Mutex>, pool: tokio::sync::OnceCell, running: tokio::sync::Mutex>, last_sync: std::sync::Mutex>, @@ -69,6 +71,8 @@ pub fn handle() -> Arc { Arc::new(Federation { data_dir: PathBuf::from(crate::media_paths::resolve_config_path("federation")), database_url: std::sync::Mutex::new(String::new()), + storage_dir: std::sync::Mutex::new(String::new()), + content_cache: std::sync::Mutex::new(Default::default()), pool: tokio::sync::OnceCell::new(), running: tokio::sync::Mutex::new(None), last_sync: std::sync::Mutex::new(None), @@ -149,6 +153,7 @@ impl Federation { /// node. Called at boot and every time the admin settings are saved. pub async fn apply(self: &Arc, config: &AppConfig) { *lock(&self.database_url) = config.database_url.clone(); + *lock(&self.storage_dir) = config.agent_storage_dir.clone(); let network = config.federation_network_id.trim().to_string(); if config.federation_enabled && !network.is_empty() { if let Err(err) = self.start(network, config.agent_storage_dir.clone()).await { @@ -362,6 +367,7 @@ impl Federation { track_number: None, disc_number: None, duration_seconds: None, + content_id: None, }); } @@ -400,6 +406,7 @@ impl Federation { track_number: None, disc_number: None, duration_seconds: None, + content_id: None, }); } @@ -426,16 +433,25 @@ impl Federation { } let tracks = sqlx::query( "SELECT t.id, t.title, COALESCE(t.year, r.year), t.duration_seconds, - r.title, r.release_type, t.track_number, t.disc_number + r.title, r.release_type, t.track_number, t.disc_number, + t.audio_file_id, m.file_path, m.sha256_hash FROM furumusic__track t JOIN furumusic__release r ON r.id = t.release_id + JOIN furumusic__media_file m ON m.id = t.audio_file_id WHERE t.is_hidden = false AND r.is_hidden = false", ) .fetch_all(&pool) .await?; + let storage_dir = lock(&self.storage_dir).clone(); for row in &tracks { let id: i64 = row.get(0); let duration: f64 = row.get(3); + let media_file_id: i64 = row.get(8); + let file_path: String = row.get(9); + let sha256_hash: String = row.get(10); + let content_id = self + .content_id_for_media(media_file_id, sha256_hash, file_path, &storage_dir) + .await; specs.push(ItemSpec { local_key: format!("track:{id}"), kind: ItemKind::Track, @@ -448,12 +464,35 @@ impl Federation { track_number: row.get(6), disc_number: row.get(7), duration_seconds: (duration > 0.0).then_some(duration), + content_id, }); } Ok(specs) } + async fn content_id_for_media( + &self, + media_file_id: i64, + sha256_hash: String, + file_path: String, + storage_dir: &str, + ) -> Option { + if let Some((cached_hash, content_id)) = lock(&self.content_cache).get(&media_file_id) + && cached_hash == &sha256_hash + { + return Some(content_id.clone()); + } + let storage_dir = storage_dir.to_string(); + let content_id = + tokio::task::spawn_blocking(move || audio_content_id(&storage_dir, &file_path)) + .await + .ok() + .flatten()?; + lock(&self.content_cache).insert(media_file_id, (sha256_hash, content_id.clone())); + Some(content_id) + } + /// Live status for the admin page. pub async fn status(&self) -> Value { let guard = self.running.lock().await; @@ -511,6 +550,17 @@ impl Federation { } } +fn audio_content_id(storage_dir: &str, file_path: &str) -> Option { + if storage_dir.trim().is_empty() { + return None; + } + let path = crate::media_paths::resolve_media_file_path(storage_dir, file_path); + let mut file = std::fs::File::open(path).ok()?; + let mut hasher = blake3::Hasher::new(); + std::io::copy(&mut file, &mut hasher).ok()?; + Some(format!("b3:{}", hasher.finalize().to_hex())) +} + async fn stop_running(running: Option) { let Some(running) = running else { return }; for task in &running.tasks { diff --git a/src/federation/serve.rs b/src/federation/serve.rs index 27ad077..4bb6f0e 100644 --- a/src/federation/serve.rs +++ b/src/federation/serve.rs @@ -101,6 +101,8 @@ struct CatalogTrack { track_number: Option, disc_number: Option, duration_seconds: Option, + #[serde(skip_serializing_if = "Option::is_none")] + content_id: Option, item_id: String, } @@ -520,7 +522,7 @@ async fn serve_catalog_one( match request.want.as_deref() { None | Some("catalog") => { - let response = match build_catalog(&pool, &own, &request.artist).await { + let response = match build_catalog(&pool, &own, &storage_dir, &request.artist).await { Ok(response) => response, Err(err) => CatalogResponse { ok: false, @@ -580,7 +582,12 @@ async fn serve_catalog_one( Ok(()) } -async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result { +async fn build_catalog( + pool: &PgPool, + own: &EndpointId, + storage_dir: &str, + artist: &str, +) -> Result { let Some(artist_row) = sqlx::query( "SELECT id, name FROM furumusic__artist WHERE LOWER(name) = LOWER($1) AND is_hidden = false @@ -612,32 +619,36 @@ async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result< for release_row in release_rows { let release_id: i64 = release_row.get(0); let track_rows = sqlx::query( - "SELECT id, title, track_number, disc_number, duration_seconds - FROM furumusic__track - WHERE release_id = $1 AND is_hidden = false - ORDER BY disc_number NULLS FIRST, track_number NULLS LAST, title", + "SELECT t.id, t.title, t.track_number, t.disc_number, t.duration_seconds, + m.file_path + FROM furumusic__track t + JOIN furumusic__media_file m ON m.id = t.audio_file_id + WHERE t.release_id = $1 AND t.is_hidden = false + ORDER BY t.disc_number NULLS FIRST, t.track_number NULLS LAST, t.title", ) .bind(release_id) .fetch_all(pool) .await?; + let mut tracks = Vec::with_capacity(track_rows.len()); + for row in track_rows { + let track_id: i64 = row.get(0); + let duration: f64 = row.get(4); + let file_path: String = row.get(5); + let content_id = catalog_content_id(storage_dir, file_path).await; + tracks.push(CatalogTrack { + title: row.get(1), + track_number: row.get(2), + disc_number: row.get(3), + duration_seconds: (duration > 0.0).then_some(duration), + content_id, + item_id: item_id_of(own, track_id), + }); + } releases.push(CatalogRelease { title: release_row.get(1), release_type: release_row.get(2), year: release_row.get(3), - tracks: track_rows - .into_iter() - .map(|row| { - let track_id: i64 = row.get(0); - let duration: f64 = row.get(4); - CatalogTrack { - title: row.get(1), - track_number: row.get(2), - disc_number: row.get(3), - duration_seconds: (duration > 0.0).then_some(duration), - item_id: item_id_of(own, track_id), - } - }) - .collect(), + tracks, }); } @@ -651,6 +662,14 @@ async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result< }) } +async fn catalog_content_id(storage_dir: &str, file_path: String) -> Option { + let storage_dir = storage_dir.to_string(); + tokio::task::spawn_blocking(move || super::audio_content_id(&storage_dir, &file_path)) + .await + .ok() + .flatten() +} + async fn artist_image_by_name(pool: &PgPool, artist: &str) -> Result> { let row = sqlx::query( "SELECT m.file_path, m.mime_type FROM furumusic__artist a