diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 08bea5d..80d76cd 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -1069,19 +1069,34 @@ impl App for AdminApp { ), "admin_releases_edit", ), - Route::with_handler_and_name( - "/releases/{id}/delete", - cot::router::method::post( - |session: Session, db: Database, path: Path| async move { - let admin = match auth::require_admin_or_redirect(&session, &db).await { - Ok(u) => u, - Err(resp) => return Ok(resp), - }; - views::releases_delete(admin, &db, path.0.id).await - }, - ), - "admin_releases_delete", - ), + { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + Route::with_handler_and_name( + "/releases/{id}/delete", + cot::router::method::post(move |session: Session, db: Database, path: Path| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + async move { + let admin = match auth::require_admin_or_redirect(&session, &db).await { + Ok(u) => u, + Err(resp) => return Ok(resp), + }; + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("admin pool") + }) + .await; + views::releases_delete(admin, &db, pg_pool, path.0.id).await + } + }), + "admin_releases_delete", + ) + }, // -- Media Files -------------------------------------------------- Route::with_handler_and_name( "/media-files", diff --git a/src/admin/v2.rs b/src/admin/v2.rs index 34345a8..eef8d2a 100644 --- a/src/admin/v2.rs +++ b/src/admin/v2.rs @@ -2066,7 +2066,12 @@ pub async fn bulk_library( .into_response(); } - let affected = apply_library_action(pool, &kind, action, &ids).await?; + let storage_dir = if action == "delete" && matches!(kind.as_str(), "releases" | "tracks") { + AppConfig::load_with_db(&db).await.0.agent_storage_dir + } else { + String::new() + }; + let affected = apply_library_action(pool, &kind, action, &ids, &storage_dir).await?; Json(MutationResponse { ok: true, affected }).into_response() } @@ -3620,10 +3625,11 @@ async fn apply_library_action( kind: &str, action: &str, ids: &[i64], + storage_dir: &str, ) -> cot::Result { match action { "hide" | "show" => set_library_visibility(pool, kind, ids, action == "hide").await, - "delete" => delete_library_items(pool, kind, ids).await, + "delete" => delete_library_items(pool, kind, ids, storage_dir).await, _ => Ok(0), } } @@ -3674,10 +3680,15 @@ async fn set_library_visibility( Ok(result.rows_affected()) } -async fn delete_library_items(pool: &PgPool, kind: &str, ids: &[i64]) -> cot::Result { +async fn delete_library_items( + pool: &PgPool, + kind: &str, + ids: &[i64], + storage_dir: &str, +) -> cot::Result { match kind { - "releases" => delete_releases(pool, ids).await, - "tracks" => delete_tracks(pool, ids).await, + "releases" => delete_releases(pool, ids, storage_dir).await, + "tracks" => delete_tracks(pool, ids, storage_dir).await, "playlists" => delete_playlists(pool, ids).await, _ => delete_artists(pool, ids).await, } @@ -3707,95 +3718,16 @@ async fn delete_artists(pool: &PgPool, ids: &[i64]) -> cot::Result { Ok(result.rows_affected()) } -async fn delete_releases(pool: &PgPool, ids: &[i64]) -> cot::Result { - let track_ids = - sqlx::query_as::<_, IdRow>("SELECT id FROM furumusic__track WHERE release_id = ANY($1)") - .bind(ids) - .fetch_all(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))? - .into_iter() - .map(|row| row.id) - .collect::>(); - - if !track_ids.is_empty() { - sqlx::query("DELETE FROM furumusic__playlist_track WHERE track_id = ANY($1)") - .bind(&track_ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__user_liked_track WHERE track_id = ANY($1)") - .bind(&track_ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__play_history WHERE track_id = ANY($1)") - .bind(&track_ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__track_genre WHERE track_id = ANY($1)") - .bind(&track_ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__track_artist WHERE track_id = ANY($1)") - .bind(&track_ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - } - - sqlx::query("DELETE FROM furumusic__track WHERE release_id = ANY($1)") - .bind(ids) - .execute(pool) +async fn delete_releases(pool: &PgPool, ids: &[i64], storage_dir: &str) -> cot::Result { + crate::library_cleanup::delete_releases(pool, ids, storage_dir) .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__release_artist WHERE release_id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - let result = sqlx::query("DELETE FROM furumusic__release WHERE id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - Ok(result.rows_affected()) + .map_err(|error| cot::Error::internal(error.to_string())) } -async fn delete_tracks(pool: &PgPool, ids: &[i64]) -> cot::Result { - sqlx::query("DELETE FROM furumusic__playlist_track WHERE track_id = ANY($1)") - .bind(ids) - .execute(pool) +async fn delete_tracks(pool: &PgPool, ids: &[i64], storage_dir: &str) -> cot::Result { + crate::library_cleanup::delete_tracks(pool, ids, storage_dir) .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__user_liked_track WHERE track_id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__play_history WHERE track_id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__track_genre WHERE track_id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - sqlx::query("DELETE FROM furumusic__track_artist WHERE track_id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - let result = sqlx::query("DELETE FROM furumusic__track WHERE id = ANY($1)") - .bind(ids) - .execute(pool) - .await - .map_err(|e| cot::Error::internal(e.to_string()))?; - Ok(result.rows_affected()) + .map_err(|error| cot::Error::internal(error.to_string())) } async fn delete_playlists(pool: &PgPool, ids: &[i64]) -> cot::Result { diff --git a/src/admin/views.rs b/src/admin/views.rs index d03f548..19fdf11 100644 --- a/src/admin/views.rs +++ b/src/admin/views.rs @@ -1262,9 +1262,11 @@ pub async fn releases_update( pub async fn releases_delete( _admin: AuthenticatedUser, db: &Database, + pool: &sqlx::PgPool, release_id: i64, ) -> cot::Result> { - Release::delete_by_id(db, release_id) + let (config, _) = AppConfig::load_with_db(db).await; + crate::library_cleanup::delete_releases(pool, &[release_id], &config.agent_storage_dir) .await .map_err(|e| cot::Error::internal(format!("failed to delete release: {e}")))?; Ok(auth::redirect("/admin/releases")) diff --git a/src/jobs/inbox_process.rs b/src/jobs/inbox_process.rs index 709b121..b6aa6ad 100644 --- a/src/jobs/inbox_process.rs +++ b/src/jobs/inbox_process.rs @@ -988,23 +988,59 @@ pub async fn finalize_approved( })? }; - let media_file = MediaFile::create( - db, - "audio", - &storage_path, - original_filename, - mime_type, - file_size, - sha256, - Some(ext), - audio_bitrate, - audio_sample_rate, - audio_bit_depth, - uploaded_by_user_id, - Some(uploader_name), + let reusable_media_id: Option = sqlx::query_scalar( + r#"SELECT media.id + FROM furumusic__media_file media + WHERE media.file_type = 'audio' + AND media.file_path = $1 + AND media.sha256_hash = $2 + AND NOT EXISTS ( + SELECT 1 FROM furumusic__track track + WHERE track.audio_file_id = media.id OR track.cover_file_id = media.id + ) + AND NOT EXISTS ( + SELECT 1 FROM furumusic__release release + WHERE release.cover_file_id = media.id + ) + AND NOT EXISTS ( + SELECT 1 FROM furumusic__artist artist + WHERE artist.image_file_id = media.id + ) + AND NOT EXISTS ( + SELECT 1 FROM furumusic__playlist playlist + WHERE playlist.cover_file_id = media.id + ) + ORDER BY media.id + LIMIT 1"#, ) - .await - .map_err(|e| anyhow::anyhow!("failed to create media file: {e}"))?; + .bind(&storage_path) + .bind(sha256) + .fetch_optional(pool) + .await?; + let media_file = if let Some(media_file_id) = reusable_media_id { + MediaFile::get_by_id(db, media_file_id) + .await + .map_err(|error| anyhow::anyhow!("failed to load reusable media file: {error}"))? + .ok_or_else(|| anyhow::anyhow!("reusable media file disappeared"))? + } else { + MediaFile::create( + db, + "audio", + &storage_path, + original_filename, + mime_type, + file_size, + sha256, + Some(ext), + audio_bitrate, + audio_sample_rate, + audio_bit_depth, + uploaded_by_user_id, + Some(uploader_name), + ) + .await + .map_err(|e| anyhow::anyhow!("failed to create media file: {e}"))? + }; let track = Track::create( db, @@ -1020,6 +1056,35 @@ pub async fn finalize_approved( .await .map_err(|e| anyhow::anyhow!("failed to create track: {e}"))?; + if let Err(error) = sqlx::query( + r#"INSERT INTO furumusic__youtube_import_media (item_id, media_file_id) + SELECT DISTINCT item.id, $1 + FROM furumusic__youtube_download_item item + JOIN furumusic__pending_review review + ON item.inbox_path IS NOT NULL + AND (review.input_path = item.inbox_path + OR left(review.input_path, length(item.inbox_path) + 1) + = item.inbox_path || '/') + WHERE review.context_json IS NOT NULL + AND substring( + review.context_json + from '"sha256"[[:space:]]*:[[:space:]]*"([0-9a-fA-F]{64})"' + ) = $2 + ON CONFLICT (item_id, media_file_id) DO NOTHING"#, + ) + .bind(media_file.id_val()) + .bind(sha256) + .execute(pool) + .await + { + tracing::warn!( + track_id = track.id_val(), + media_file_id = media_file.id_val(), + error = %error, + "failed to link imported media to its YouTube download item" + ); + } + TrackArtist::create(db, track.id_val(), artist.id_val(), "main", 0) .await .map_err(|e| anyhow::anyhow!("failed to link track-artist: {e}"))?; diff --git a/src/library_cleanup.rs b/src/library_cleanup.rs new file mode 100644 index 0000000..9ef7c27 --- /dev/null +++ b/src/library_cleanup.rs @@ -0,0 +1,568 @@ +use std::collections::HashSet; +use std::io::ErrorKind; +use std::path::PathBuf; + +use anyhow::{Context, bail}; +use sqlx::{FromRow, PgPool, Postgres, Transaction}; +use uuid::Uuid; + +#[derive(Debug, FromRow)] +struct MediaFileRow { + id: i64, + file_type: String, + file_path: String, + sha256_hash: String, +} + +#[derive(Debug, FromRow)] +struct PlaybackStateRow { + id: i64, + current_track_id: Option, + position_ms: i32, + queue_json: String, + queue_position: i32, +} + +#[derive(Debug)] +struct QuarantinedFile { + original: PathBuf, + quarantined: PathBuf, +} + +#[derive(Debug, Default)] +struct Quarantine { + root: Option, + files: Vec, +} + +pub async fn delete_tracks( + pool: &PgPool, + requested_track_ids: &[i64], + storage_dir: &str, +) -> anyhow::Result { + delete_scope(pool, requested_track_ids, &[], storage_dir).await +} + +pub async fn delete_releases( + pool: &PgPool, + requested_release_ids: &[i64], + storage_dir: &str, +) -> anyhow::Result { + let mut transaction = pool.begin().await?; + let release_ids: Vec = sqlx::query_scalar( + "SELECT id FROM furumusic__release WHERE id = ANY($1) ORDER BY id FOR UPDATE", + ) + .bind(requested_release_ids) + .fetch_all(&mut *transaction) + .await?; + if release_ids.is_empty() { + transaction.rollback().await?; + return Ok(0); + } + let track_ids: Vec = sqlx::query_scalar( + "SELECT id FROM furumusic__track WHERE release_id = ANY($1) ORDER BY id FOR UPDATE", + ) + .bind(&release_ids) + .fetch_all(&mut *transaction) + .await?; + delete_locked_scope(transaction, track_ids, release_ids, storage_dir, true).await +} + +async fn delete_scope( + pool: &PgPool, + requested_track_ids: &[i64], + release_ids: &[i64], + storage_dir: &str, +) -> anyhow::Result { + let mut transaction = pool.begin().await?; + let track_ids: Vec = sqlx::query_scalar( + "SELECT id FROM furumusic__track WHERE id = ANY($1) ORDER BY id FOR UPDATE", + ) + .bind(requested_track_ids) + .fetch_all(&mut *transaction) + .await?; + if track_ids.is_empty() { + transaction.rollback().await?; + return Ok(0); + } + delete_locked_scope( + transaction, + track_ids, + release_ids.to_vec(), + storage_dir, + false, + ) + .await +} + +async fn delete_locked_scope( + mut transaction: Transaction<'_, Postgres>, + track_ids: Vec, + release_ids: Vec, + storage_dir: &str, + delete_release_rows: bool, +) -> anyhow::Result { + let media_files = deletable_media_files(&mut transaction, &track_ids, &release_ids).await?; + let quarantine = match quarantine_media_files(storage_dir, &media_files).await { + Ok(quarantine) => quarantine, + Err(error) => { + transaction.rollback().await?; + return Err(error); + } + }; + + let deletion = delete_database_rows( + &mut transaction, + &track_ids, + &release_ids, + &media_files, + delete_release_rows, + ) + .await; + let affected = match deletion { + Ok(affected) => affected, + Err(error) => { + transaction.rollback().await?; + restore_quarantine(&quarantine).await; + return Err(error); + } + }; + + if let Err(error) = transaction.commit().await { + restore_quarantine(&quarantine).await; + return Err(error.into()); + } + purge_quarantine(&quarantine).await; + remove_empty_storage_parents(storage_dir, &quarantine.files).await; + Ok(affected) +} + +async fn deletable_media_files( + transaction: &mut Transaction<'_, Postgres>, + track_ids: &[i64], + release_ids: &[i64], +) -> anyhow::Result> { + Ok(sqlx::query_as( + r#"WITH seed_media(id) AS ( + SELECT audio_file_id FROM furumusic__track WHERE id = ANY($1) + UNION + SELECT cover_file_id FROM furumusic__track + WHERE id = ANY($1) AND cover_file_id IS NOT NULL + UNION + SELECT cover_file_id FROM furumusic__release + WHERE id = ANY($2) AND cover_file_id IS NOT NULL + ), candidate_media(id) AS ( + SELECT id FROM seed_media + UNION + SELECT duplicate.id + FROM furumusic__media_file duplicate + JOIN furumusic__media_file seed + ON duplicate.file_path = seed.file_path + AND duplicate.sha256_hash = seed.sha256_hash + JOIN seed_media ON seed_media.id = seed.id + ) + SELECT mf.id, mf.file_type::text AS file_type, mf.file_path, + mf.sha256_hash::text AS sha256_hash + FROM furumusic__media_file mf + JOIN candidate_media candidate ON candidate.id = mf.id + WHERE NOT EXISTS ( + SELECT 1 + FROM furumusic__track track + JOIN furumusic__media_file linked + ON linked.id = track.audio_file_id + OR linked.id = track.cover_file_id + WHERE linked.file_path = mf.file_path + AND linked.sha256_hash = mf.sha256_hash + AND NOT (track.id = ANY($1)) + ) + AND NOT EXISTS ( + SELECT 1 + FROM furumusic__release release + JOIN furumusic__media_file linked + ON linked.id = release.cover_file_id + WHERE linked.file_path = mf.file_path + AND linked.sha256_hash = mf.sha256_hash + AND NOT (release.id = ANY($2)) + ) + AND NOT EXISTS ( + SELECT 1 + FROM furumusic__artist artist + JOIN furumusic__media_file linked + ON linked.id = artist.image_file_id + WHERE linked.file_path = mf.file_path + AND linked.sha256_hash = mf.sha256_hash + ) + AND NOT EXISTS ( + SELECT 1 + FROM furumusic__playlist playlist + JOIN furumusic__media_file linked + ON linked.id = playlist.cover_file_id + WHERE linked.file_path = mf.file_path + AND linked.sha256_hash = mf.sha256_hash + ) + ORDER BY mf.id + FOR UPDATE OF mf"#, + ) + .bind(track_ids) + .bind(release_ids) + .fetch_all(&mut **transaction) + .await?) +} + +async fn quarantine_media_files( + storage_dir: &str, + media_files: &[MediaFileRow], +) -> anyhow::Result { + if media_files.is_empty() { + return Ok(Quarantine::default()); + } + if storage_dir.trim().is_empty() { + bail!("agent_storage_dir is not configured; refusing to leave deleted tracks on disk"); + } + + let storage_root = crate::media_paths::resolve_config_path_buf(storage_dir); + if storage_root.parent().is_none() { + bail!("agent_storage_dir must not be a filesystem root"); + } + let quarantine_root = storage_root + .join(".furumusic-trash") + .join(Uuid::new_v4().to_string()); + let mut quarantine = Quarantine { + root: Some(quarantine_root.clone()), + files: Vec::new(), + }; + let mut seen = HashSet::new(); + let result: anyhow::Result<()> = + async { + for media in media_files { + let original = checked_storage_path(storage_dir, &media.file_path)?; + let mut paths = vec![original.clone()]; + if media.file_type == "cover_art" { + paths.extend(crate::agent::cover_variants::COVER_VARIANTS.iter().map( + |variant| crate::agent::cover_variants::variant_path(&original, *variant), + )); + } + + for (index, path) in paths.into_iter().enumerate() { + if !seen.insert(path.clone()) { + continue; + } + match tokio::fs::symlink_metadata(&path).await { + Ok(metadata) if metadata.is_file() || metadata.file_type().is_symlink() => { + } + Ok(_) => bail!("media path is not a regular file: {}", path.display()), + Err(error) if error.kind() == ErrorKind::NotFound => continue, + Err(error) => return Err(error.into()), + } + tokio::fs::create_dir_all(&quarantine_root).await?; + let extension = path + .extension() + .and_then(|value| value.to_str()) + .unwrap_or("bin"); + let quarantined = quarantine_root.join(format!( + "{}-{index}-{}.{}", + media.id, + Uuid::new_v4(), + extension + )); + tokio::fs::rename(&path, &quarantined) + .await + .with_context(|| format!("failed to remove {}", path.display()))?; + quarantine.files.push(QuarantinedFile { + original: path, + quarantined, + }); + } + } + Ok(()) + } + .await; + match result { + Ok(()) => Ok(quarantine), + Err(error) => { + restore_quarantine(&quarantine).await; + Err(error) + } + } +} + +fn checked_storage_path(storage_dir: &str, stored_path: &str) -> anyhow::Result { + let resolved = crate::media_paths::resolve_media_file_path(storage_dir, stored_path); + crate::media_paths::path_for_root(storage_dir, &resolved).with_context(|| { + format!( + "refusing to delete media outside agent_storage_dir: {}", + resolved.display() + ) + })?; + Ok(resolved) +} + +async fn restore_quarantine(quarantine: &Quarantine) { + for file in quarantine.files.iter().rev() { + if let Some(parent) = file.original.parent() { + let _ = tokio::fs::create_dir_all(parent).await; + } + let _ = tokio::fs::rename(&file.quarantined, &file.original).await; + } + if let Some(root) = &quarantine.root { + let _ = tokio::fs::remove_dir_all(root).await; + } +} + +async fn purge_quarantine(quarantine: &Quarantine) { + if let Some(root) = &quarantine.root + && let Err(error) = tokio::fs::remove_dir_all(root).await + && error.kind() != ErrorKind::NotFound + { + tracing::warn!(path = %root.display(), error = %error, "failed to purge deleted media quarantine"); + } + if let Some(parent) = quarantine.root.as_deref().and_then(|root| root.parent()) { + let _ = tokio::fs::remove_dir(parent).await; + } +} + +async fn remove_empty_storage_parents(storage_dir: &str, files: &[QuarantinedFile]) { + let storage_root = crate::media_paths::resolve_config_path_buf(storage_dir); + let mut seen = HashSet::new(); + for file in files { + let mut current = file.original.parent(); + while let Some(directory) = current { + if directory == storage_root || !directory.starts_with(&storage_root) { + break; + } + if !seen.insert(directory.to_path_buf()) { + break; + } + match tokio::fs::remove_dir(directory).await { + Ok(()) => current = directory.parent(), + Err(_) => break, + } + } + } +} + +async fn delete_database_rows( + transaction: &mut Transaction<'_, Postgres>, + track_ids: &[i64], + release_ids: &[i64], + media_files: &[MediaFileRow], + delete_release_rows: bool, +) -> anyhow::Result { + cleanup_playback_states(transaction, track_ids).await?; + + for table in [ + "furumusic__playlist_track", + "furumusic__user_liked_track", + "furumusic__play_history", + "furumusic__track_popularity_history", + "furumusic__lastfm_scrobble_outbox", + "furumusic__track_genre", + "furumusic__track_artist", + "furumusic__track_embedding", + ] { + let query = format!("DELETE FROM {table} WHERE track_id = ANY($1)"); + sqlx::query(&query) + .bind(track_ids) + .execute(&mut **transaction) + .await?; + } + for table in [ + "furumusic__entity_genre_tag", + "furumusic__external_metadata_id", + "furumusic__artwork_lookup_state", + ] { + let query = + format!("DELETE FROM {table} WHERE entity_kind = 'track' AND entity_id = ANY($1)"); + sqlx::query(&query) + .bind(track_ids) + .execute(&mut **transaction) + .await?; + } + for table in [ + "furumusic__fed_state_like", + "furumusic__fed_state_playlist_item", + "furumusic__track_ref", + "furumusic__listen_event", + ] { + let query = + format!("UPDATE {table} SET local_track_id = NULL WHERE local_track_id = ANY($1)"); + sqlx::query(&query) + .bind(track_ids) + .execute(&mut **transaction) + .await?; + } + + let tracks_deleted = sqlx::query("DELETE FROM furumusic__track WHERE id = ANY($1)") + .bind(track_ids) + .execute(&mut **transaction) + .await? + .rows_affected(); + + if delete_release_rows { + for table in [ + "furumusic__entity_genre_tag", + "furumusic__external_metadata_id", + "furumusic__artwork_lookup_state", + ] { + let query = format!( + "DELETE FROM {table} WHERE entity_kind = 'release' AND entity_id = ANY($1)" + ); + sqlx::query(&query) + .bind(release_ids) + .execute(&mut **transaction) + .await?; + } + sqlx::query("DELETE FROM furumusic__release_artist WHERE release_id = ANY($1)") + .bind(release_ids) + .execute(&mut **transaction) + .await?; + sqlx::query("DELETE FROM furumusic__release WHERE id = ANY($1)") + .bind(release_ids) + .execute(&mut **transaction) + .await?; + } + + let media_ids: Vec = media_files.iter().map(|media| media.id).collect(); + if !media_ids.is_empty() { + let media_hashes: Vec = media_files + .iter() + .map(|media| media.sha256_hash.clone()) + .collect(); + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = 'failed', progress_percent = 0, + downloaded_bytes = 0, total_bytes = NULL, + speed_bytes_per_sec = NULL, eta_seconds = NULL, + error = 'Imported library files were deleted; this source can be imported again', + completed_at = NULL, updated_at = $2 + WHERE id IN ( + SELECT item_id + FROM furumusic__youtube_import_media + WHERE media_file_id = ANY($1) + )"#, + ) + .bind(&media_ids) + .bind(chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()) + .execute(&mut **transaction) + .await?; + + let review_ids: Vec = sqlx::query_scalar( + r#"SELECT id FROM furumusic__pending_review + WHERE context_json IS NOT NULL + AND substring( + context_json + from '"sha256"[[:space:]]*:[[:space:]]*"([0-9a-fA-F]{64})"' + ) = ANY($1)"#, + ) + .bind(&media_hashes) + .fetch_all(&mut **transaction) + .await?; + if !review_ids.is_empty() { + sqlx::query( + "DELETE FROM furumusic__processing_stats WHERE pending_review_id = ANY($1)", + ) + .bind(&review_ids) + .execute(&mut **transaction) + .await?; + sqlx::query("DELETE FROM furumusic__pending_review WHERE id = ANY($1)") + .bind(&review_ids) + .execute(&mut **transaction) + .await?; + } + sqlx::query( + "DELETE FROM furumusic__federation_content_id_cache WHERE media_file_id = ANY($1)", + ) + .bind(&media_ids) + .execute(&mut **transaction) + .await?; + sqlx::query("DELETE FROM furumusic__media_file WHERE id = ANY($1)") + .bind(&media_ids) + .execute(&mut **transaction) + .await?; + } + Ok(if delete_release_rows { + u64::try_from(release_ids.len()).unwrap_or(u64::MAX) + } else { + tracks_deleted + }) +} + +async fn cleanup_playback_states( + transaction: &mut Transaction<'_, Postgres>, + track_ids: &[i64], +) -> anyhow::Result<()> { + let deleted: HashSet = track_ids.iter().copied().collect(); + let states: Vec = sqlx::query_as( + "SELECT id, current_track_id, position_ms, queue_json, queue_position FROM furumusic__playback_state FOR UPDATE", + ) + .fetch_all(&mut **transaction) + .await?; + + for state in states { + let mut queue: Vec = serde_json::from_str(&state.queue_json).unwrap_or_default(); + let original_queue = queue.clone(); + queue.retain(|track_id| !deleted.contains(track_id)); + let current_track_id = state.current_track_id.filter(|id| !deleted.contains(id)); + if queue == original_queue && current_track_id == state.current_track_id { + continue; + } + let queue_position = current_track_id + .and_then(|current| queue.iter().position(|id| *id == current)) + .map(|position| i32::try_from(position).unwrap_or(i32::MAX)) + .unwrap_or_else(|| { + if queue.is_empty() { + 0 + } else { + state + .queue_position + .clamp(0, i32::try_from(queue.len() - 1).unwrap_or(i32::MAX)) + } + }); + let position_ms = if current_track_id.is_some() { + state.position_ms + } else { + 0 + }; + let queue_json = serde_json::to_string(&queue)?; + sqlx::query( + r#"UPDATE furumusic__playback_state + SET current_track_id = $2, position_ms = $3, + queue_json = $4, queue_position = $5 + WHERE id = $1"#, + ) + .bind(state.id) + .bind(current_track_id) + .bind(position_ms) + .bind(queue_json) + .bind(queue_position) + .execute(&mut **transaction) + .await?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_media_paths_outside_storage() { + assert!(checked_storage_path("/srv/music", "/etc/passwd").is_err()); + assert!(checked_storage_path("/srv/music", "../outside.flac").is_err()); + assert_eq!( + checked_storage_path("/srv/music", "Artist/Album/01.flac").unwrap(), + PathBuf::from("/srv/music/Artist/Album/01.flac") + ); + } + + #[test] + fn deleted_track_ids_are_removed_from_saved_queue() { + let deleted = HashSet::from([2_i64, 4]); + let mut queue = vec![1_i64, 2, 3, 4, 5]; + queue.retain(|track_id| !deleted.contains(track_id)); + assert_eq!(queue, vec![1, 3, 5]); + let serialized: serde_json::Value = + serde_json::from_str(&serde_json::to_string(&queue).unwrap()).unwrap(); + assert_eq!(serialized, serde_json::json!([1, 3, 5])); + } +} diff --git a/src/main.rs b/src/main.rs index 7dd6d80..b485a42 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,6 +7,7 @@ mod federation; mod i18n; mod jobs; mod lastfm; +mod library_cleanup; mod local_uploads; mod media_paths; mod metrics; diff --git a/src/music/mod.rs b/src/music/mod.rs index 5821dcb..e6d1130 100644 --- a/src/music/mod.rs +++ b/src/music/mod.rs @@ -2715,6 +2715,73 @@ pub mod db_migrations { &[Operation::custom(create_local_upload_history).build()]; } + // -- M0047: durable YouTube item -> imported media links --------------- + + #[cot::db::migrations::migration_op] + async fn create_youtube_import_media_links( + ctx: migrations::MigrationContext<'_>, + ) -> cot::db::Result<()> { + ctx.db + .raw( + "CREATE TABLE IF NOT EXISTS furumusic__youtube_import_media ( + item_id VARCHAR(36) NOT NULL + REFERENCES furumusic__youtube_download_item(id) ON DELETE CASCADE, + media_file_id BIGINT NOT NULL + REFERENCES furumusic__media_file(id) ON DELETE CASCADE, + PRIMARY KEY (item_id, media_file_id) + )", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_youtube_import_media_file + ON furumusic__youtube_import_media (media_file_id)", + ) + .await?; + // Backfill links for existing imports through the inbox review hash. + ctx.db + .raw( + "INSERT INTO furumusic__youtube_import_media (item_id, media_file_id) + SELECT DISTINCT item.id, media.id + FROM furumusic__youtube_download_item item + JOIN furumusic__pending_review review + ON item.inbox_path IS NOT NULL + AND (review.input_path = item.inbox_path + OR left(review.input_path, length(item.inbox_path) + 1) + = item.inbox_path || '/') + JOIN furumusic__media_file media + ON media.sha256_hash::text = substring( + review.context_json + from '\"sha256\"[[:space:]]*:[[:space:]]*\"([0-9a-fA-F]{64})\"' + ) + JOIN furumusic__track track ON track.audio_file_id = media.id + WHERE review.context_json IS NOT NULL + ON CONFLICT (item_id, media_file_id) DO NOTHING", + ) + .await?; + Ok(()) + } + + #[derive(Debug, Copy, Clone)] + pub struct M0047CreateYouTubeImportMediaLinks; + + impl migrations::Migration for M0047CreateYouTubeImportMediaLinks { + const APP_NAME: &'static str = "furumusic"; + const MIGRATION_NAME: &'static str = "m_0047_create_youtube_import_media_links"; + const DEPENDENCIES: &'static [migrations::MigrationDependency] = &[ + migrations::MigrationDependency::migration( + "furumusic", + "m_0046_create_local_upload_history", + ), + migrations::MigrationDependency::migration( + "furumusic", + "m_0027_create_processing_stats", + ), + ]; + const OPERATIONS: &'static [Operation] = + &[Operation::custom(create_youtube_import_media_links).build()]; + } + pub const MIGRATIONS: &[&SyncDynMigration] = &[ &M0006CreateMediaFile, &M0007CreateArtist, @@ -2752,5 +2819,6 @@ pub mod db_migrations { &M0044AddSimilarityRoutingSignature, &M0045CreateYouTubeDownloads, &M0046CreateLocalUploadHistory, + &M0047CreateYouTubeImportMediaLinks, ]; } diff --git a/src/youtube.rs b/src/youtube.rs index 659d519..40b5dec 100644 --- a/src/youtube.rs +++ b/src/youtube.rs @@ -1246,10 +1246,17 @@ async fn prepare_downloaded_folder( if tokio::fs::try_exists(&destination).await? { let existing = find_audio_files(&destination).await?; if existing.is_empty() { - bail!("YouTube inbox destination already exists without audio"); + // A completed inbox import moves audio into the media library but + // may leave cover.jpg behind. Treat that directory as an orphan so + // deleting a release does not make the same source impossible to + // import again. + tokio::fs::remove_dir_all(&destination).await?; + tokio::fs::rename(stage, &destination).await?; + audio_files = find_audio_files(&destination).await?; + } else { + tokio::fs::remove_dir_all(stage).await?; + audio_files = existing; } - tokio::fs::remove_dir_all(stage).await?; - audio_files = existing; } else { tokio::fs::rename(stage, &destination).await?; } @@ -1645,9 +1652,18 @@ async fn source_already_imported( FROM furumusic__youtube_download_item i JOIN furumusic__youtube_download j ON j.id = i.job_id WHERE j.user_id = $1 AND i.job_id <> $2 AND i.source_id = $3 - AND i.status IN ( - 'queued', 'downloading', 'postprocessing', 'awaiting_ai', - 'ai_processing', 'complete', 'needs_review', 'skipped' + AND ( + i.status IN ( + 'queued', 'downloading', 'postprocessing', 'awaiting_ai', + 'ai_processing', 'needs_review' + ) + OR ( + i.status IN ('complete', 'skipped') + AND (SELECT COUNT(*) + FROM furumusic__youtube_import_media imported + WHERE imported.item_id = i.id) >= i.audio_file_count + AND i.audio_file_count > 0 + ) ) )"#, ) @@ -1669,9 +1685,18 @@ async fn already_imported_source_ids( FROM furumusic__youtube_download_item i JOIN furumusic__youtube_download j ON j.id = i.job_id WHERE j.user_id = $1 AND i.source_id = ANY($2) - AND i.status IN ( - 'queued', 'downloading', 'postprocessing', 'awaiting_ai', - 'ai_processing', 'complete', 'needs_review', 'skipped' + AND ( + i.status IN ( + 'queued', 'downloading', 'postprocessing', 'awaiting_ai', + 'ai_processing', 'needs_review' + ) + OR ( + i.status IN ('complete', 'skipped') + AND (SELECT COUNT(*) + FROM furumusic__youtube_import_media imported + WHERE imported.item_id = i.id) >= i.audio_file_count + AND i.audio_file_count > 0 + ) )"#, ) .bind(user_id)