Fixed content_id persistence

Cargo.lock

@@ -1844,7 +1844,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
 
 [[package]]
 name = "furumusic"
-version = "0.6.3-fd"
+version = "0.6.5-fd"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumusic"
-version = "0.6.4-fd"
+version = "0.6.6-fd"
 edition = "2024"
 description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
 

src/federation/mod.rs

@@ -15,6 +15,7 @@
 mod serve;
 mod storage;
 
+use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
@@ -42,12 +43,19 @@ struct Running {
     tasks: Vec<tokio::task::JoinHandle<()>>,
 }
 
+struct ContentHashJob {
+    media_file_id: i64,
+    sha256_hash: String,
+    file_path: String,
+}
+
 pub struct Federation {
     /// Transport data directory; server-side DHT state and identity live in PostgreSQL.
     data_dir: PathBuf,
     database_url: std::sync::Mutex<String>,
     storage_dir: std::sync::Mutex<String>,
-    content_cache: std::sync::Mutex<std::collections::HashMap<i64, (String, String)>>,
+    content_cache: std::sync::Mutex<HashMap<i64, (String, String)>>,
+    content_pending: std::sync::Mutex<HashSet<i64>>,
     pool: tokio::sync::OnceCell<PgPool>,
     running: tokio::sync::Mutex<Option<Running>>,
     last_sync: std::sync::Mutex<Option<String>>,
@@ -73,6 +81,7 @@ pub fn handle() -> Arc<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()),
+            content_pending: std::sync::Mutex::new(Default::default()),
             pool: tokio::sync::OnceCell::new(),
             running: tokio::sync::Mutex::new(None),
             last_sync: std::sync::Mutex::new(None),
@@ -286,7 +295,7 @@ impl Federation {
         Ok(())
     }
 
-    async fn sync_once(&self, service: &MusicDhtService) -> Result<SyncStats> {
+    async fn sync_once(self: &Arc<Self>, service: &MusicDhtService) -> Result<SyncStats> {
         let specs = match self.collect_specs().await {
             Ok(specs) => specs,
             Err(err) => {
@@ -346,7 +355,7 @@ impl Federation {
 
     /// Everything the regular player shows, as DHT item specs: non-hidden
     /// artists, releases and tracks (a track also hides with its release).
-    async fn collect_specs(&self) -> Result<Vec<ItemSpec>> {
+    async fn collect_specs(self: &Arc<Self>) -> Result<Vec<ItemSpec>> {
         let pool = self.pool().await?;
         let mut specs = Vec::new();
 
@@ -434,24 +443,37 @@ 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,
-                    t.audio_file_id, m.file_path, m.sha256_hash
+                    t.audio_file_id, m.file_path, m.sha256_hash, c.content_id
              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
+             LEFT JOIN furumusic__federation_content_id_cache c
+                    ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash
              WHERE t.is_hidden = false AND r.is_hidden = false",
         )
         .fetch_all(&pool)
         .await?;
         let storage_dir = lock(&self.storage_dir).clone();
+        let mut content_hash_jobs = Vec::new();
         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
+            let cached_content_id: Option<String> = row.get(11);
-                .content_id_for_media(media_file_id, sha256_hash, file_path, &storage_dir)
+            let content_id = cached_content_id
-                .await;
+                .or_else(|| self.cached_content_id_for_media(media_file_id, &sha256_hash));
+            if content_id.is_none()
+                && !storage_dir.trim().is_empty()
+                && self.mark_content_hash_pending(media_file_id)
+            {
+                content_hash_jobs.push(ContentHashJob {
+                    media_file_id,
+                    sha256_hash,
+                    file_path,
+                });
+            }
             specs.push(ItemSpec {
                 local_key: format!("track:{id}"),
                 kind: ItemKind::Track,
@@ -467,30 +489,67 @@ impl Federation {
                 content_id,
             });
         }
+        self.spawn_content_warmer(pool.clone(), storage_dir, content_hash_jobs);
 
         Ok(specs)
     }
 
-    async fn content_id_for_media(
+    fn cached_content_id_for_media(&self, media_file_id: i64, sha256_hash: &str) -> Option<String> {
-        &self,
-        media_file_id: i64,
-        sha256_hash: String,
-        file_path: String,
-        storage_dir: &str,
-    ) -> Option<String> {
         if let Some((cached_hash, content_id)) = lock(&self.content_cache).get(&media_file_id)
-            && cached_hash == &sha256_hash
+            && cached_hash == sha256_hash
         {
             return Some(content_id.clone());
         }
-        let storage_dir = storage_dir.to_string();
+        None
-        let content_id =
+    }
-            tokio::task::spawn_blocking(move || audio_content_id(&storage_dir, &file_path))
+
+    fn mark_content_hash_pending(&self, media_file_id: i64) -> bool {
+        lock(&self.content_pending).insert(media_file_id)
+    }
+
+    fn spawn_content_warmer(
+        self: &Arc<Self>,
+        pool: PgPool,
+        storage_dir: String,
+        jobs: Vec<ContentHashJob>,
+    ) {
+        if jobs.is_empty() {
+            return;
+        }
+        let fed = Arc::clone(self);
+        tokio::spawn(async move {
+            let total = jobs.len();
+            let mut stored = 0usize;
+            for job in jobs {
+                let job_storage_dir = storage_dir.clone();
+                let job_file_path = job.file_path.clone();
+                let content_id = tokio::task::spawn_blocking(move || {
+                    audio_content_id(&job_storage_dir, &job_file_path)
+                })
                 .await
                 .ok()
-                .flatten()?;
+                .flatten();
-        lock(&self.content_cache).insert(media_file_id, (sha256_hash, content_id.clone()));
+                lock(&fed.content_pending).remove(&job.media_file_id);
-        Some(content_id)
+                if let Some(content_id) = content_id {
+                    lock(&fed.content_cache).insert(
+                        job.media_file_id,
+                        (job.sha256_hash.clone(), content_id.clone()),
+                    );
+                    if let Err(err) =
+                        persist_content_id(&pool, job.media_file_id, &job.sha256_hash, &content_id)
+                            .await
+                    {
+                        tracing::warn!(
+                            media_file_id = job.media_file_id,
+                            "federation content-id cache write failed: {err:#}"
+                        );
+                    } else {
+                        stored += 1;
+                    }
+                }
+            }
+            tracing::info!(total, stored, "federation content-id cache warm finished");
+        });
     }
 
     /// Live status for the admin page.
@@ -550,6 +609,30 @@ impl Federation {
     }
 }
 
+async fn persist_content_id(
+    pool: &PgPool,
+    media_file_id: i64,
+    sha256_hash: &str,
+    content_id: &str,
+) -> Result<()> {
+    sqlx::query(
+        "INSERT INTO furumusic__federation_content_id_cache
+         (media_file_id, sha256_hash, content_id, updated_at)
+         VALUES ($1, $2, $3, $4)
+         ON CONFLICT (media_file_id) DO UPDATE SET
+            sha256_hash = EXCLUDED.sha256_hash,
+            content_id = EXCLUDED.content_id,
+            updated_at = EXCLUDED.updated_at",
+    )
+    .bind(media_file_id)
+    .bind(sha256_hash)
+    .bind(content_id)
+    .bind(now_iso())
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
 fn audio_content_id(storage_dir: &str, file_path: &str) -> Option<String> {
     if storage_dir.trim().is_empty() {
         return None;

src/federation/serve.rs

@@ -522,7 +522,7 @@ async fn serve_catalog_one(
 
     match request.want.as_deref() {
         None | Some("catalog") => {
-            let response = match build_catalog(&pool, &own, &storage_dir, &request.artist).await {
+            let response = match build_catalog(&pool, &own, &request.artist).await {
                 Ok(response) => response,
                 Err(err) => CatalogResponse {
                     ok: false,
@@ -582,12 +582,7 @@ async fn serve_catalog_one(
     Ok(())
 }
 
-async fn build_catalog(
+async fn build_catalog(pool: &PgPool, own: &EndpointId, artist: &str) -> Result<CatalogResponse> {
-    pool: &PgPool,
-    own: &EndpointId,
-    storage_dir: &str,
-    artist: &str,
-) -> Result<CatalogResponse> {
     let Some(artist_row) = sqlx::query(
         "SELECT id, name FROM furumusic__artist
          WHERE LOWER(name) = LOWER($1) AND is_hidden = false
@@ -620,9 +615,11 @@ async fn build_catalog(
         let release_id: i64 = release_row.get(0);
         let track_rows = sqlx::query(
             "SELECT t.id, t.title, t.track_number, t.disc_number, t.duration_seconds,
-                    m.file_path
+                    c.content_id
              FROM furumusic__track t
              JOIN furumusic__media_file m ON m.id = t.audio_file_id
+             LEFT JOIN furumusic__federation_content_id_cache c
+                    ON c.media_file_id = m.id AND c.sha256_hash = m.sha256_hash
              WHERE t.release_id = $1 AND t.is_hidden = false
              ORDER BY t.disc_number NULLS FIRST, t.track_number NULLS LAST, t.title",
         )
@@ -633,14 +630,12 @@ async fn build_catalog(
         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,
+                content_id: row.get(5),
                 item_id: item_id_of(own, track_id),
             });
         }
@@ -662,14 +657,6 @@ async fn build_catalog(
     })
 }
 
-async fn catalog_content_id(storage_dir: &str, file_path: String) -> Option<String> {
-    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<Option<(String, String)>> {
     let row = sqlx::query(
         "SELECT m.file_path, m.mime_type FROM furumusic__artist a

src/federation/storage.rs

@@ -44,6 +44,14 @@ const SCHEMA: &[&str] = &[
         ticket        TEXT NOT NULL,
         last_seen_ms  BIGINT NOT NULL
     )",
+    "CREATE TABLE IF NOT EXISTS furumusic__federation_content_id_cache (
+        media_file_id  BIGINT PRIMARY KEY,
+        sha256_hash    TEXT NOT NULL,
+        content_id     TEXT NOT NULL,
+        updated_at     TEXT NOT NULL
+    )",
+    "CREATE INDEX IF NOT EXISTS idx_furumusic_federation_content_id_cache_content_id
+        ON furumusic__federation_content_id_cache(content_id)",
 ];
 
 #[derive(Debug, Clone)]