From ec9ca6d99d406bd00693b66335351af677e1c809 Mon Sep 17 00:00:00 2001 From: Collin Pendleton Date: Fri, 23 Oct 2026 09:06:52 -0500 Subject: [PATCH 1/2] Cheaper YouTube channel refresh --- rust-api/src/config.rs | 5 + rust-api/src/database.rs | 261 +++---- rust-api/src/handlers/refresh.rs | 251 +++--- rust-api/src/handlers/youtube.rs | 731 +++++++++--------- rust-api/src/services/ytdlp.rs | 12 + rust-api/tests/integration/harness/app.rs | 1 + rust-api/tests/integration/harness/env.rs | 5 + rust-api/tests/integration/harness/mod.rs | 1 + rust-api/tests/integration/harness/youtube.rs | 215 ++++++ rust-api/tests/integration/main.rs | 1 + rust-api/tests/integration/youtube.rs | 216 ++++++ 11 files changed, 1107 insertions(+), 592 deletions(-) create mode 100644 rust-api/tests/integration/harness/youtube.rs create mode 100644 rust-api/tests/integration/youtube.rs diff --git a/rust-api/src/config.rs b/rust-api/src/config.rs index 480a06d26..19238f74a 100644 --- a/rust-api/src/config.rs +++ b/rust-api/src/config.rs @@ -13,10 +13,14 @@ pub struct Config { pub api: ApiConfig, } +pub const YOUTUBE_FEED_URL: &str = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"; + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ApiConfig { pub search_api_url: String, pub people_api_url: String, + /// A YouTube channel's RSS feed, with `{channel_id}` in place of the channel. + pub youtube_feed_url: String, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -335,6 +339,7 @@ impl Config { let api = ApiConfig { search_api_url: env::var("SEARCH_API_URL").unwrap(), people_api_url: env::var("PEOPLE_API_URL").unwrap(), + youtube_feed_url: YOUTUBE_FEED_URL.to_string(), }; // Validate OIDC configuration diff --git a/rust-api/src/database.rs b/rust-api/src/database.rs index 9a6c5b859..212257b1e 100644 --- a/rust-api/src/database.rs +++ b/rust-api/src/database.rs @@ -16807,190 +16807,145 @@ impl DatabasePool { } } - // Remove old YouTube videos - deletes videos and all their references from dependent tables + // Remove a channel's videos published before the cutoff, except those a user still has in + // their history, queue, saved list, downloads, playlists or collections. pub async fn remove_old_youtube_videos(&self, podcast_id: i32, cutoff_date: chrono::DateTime) -> AppResult<()> { - debug!("Removing old YouTube videos for podcast {} before {}", podcast_id, cutoff_date); - let cutoff_naive = cutoff_date.naive_utc(); - let rows_affected = match self { DatabasePool::Postgres(pool) => { - // First, delete all references from dependent tables - let cleanup_queries = vec![ - r#"DELETE FROM "UserVideoHistory" WHERE videoid IN (SELECT videoid FROM "YouTubeVideos" WHERE podcastid = $1 AND publishedat < $2)"#, - r#"DELETE FROM "SavedVideos" WHERE videoid IN (SELECT videoid FROM "YouTubeVideos" WHERE podcastid = $1 AND publishedat < $2)"#, - r#"DELETE FROM "DownloadedVideos" WHERE videoid IN (SELECT videoid FROM "YouTubeVideos" WHERE podcastid = $1 AND publishedat < $2)"#, - r#"DELETE FROM "PlaylistContents" WHERE videoid IN (SELECT videoid FROM "YouTubeVideos" WHERE podcastid = $1 AND publishedat < $2)"#, - r#"DELETE FROM "EpisodeQueue" WHERE episodeid IN (SELECT videoid FROM "YouTubeVideos" WHERE podcastid = $1 AND publishedat < $2)"#, - ]; - - for query in cleanup_queries { - sqlx::query(query) - .bind(podcast_id) - .bind(cutoff_naive) - .execute(pool) - .await?; - } - - // Now delete the videos themselves - sqlx::query(r#"DELETE FROM "YouTubeVideos" WHERE podcastid = $1 AND publishedat < $2"#) - .bind(podcast_id) - .bind(cutoff_naive) - .execute(pool) - .await? - .rows_affected() + sqlx::query(r#" + DELETE FROM "YouTubeVideos" AS v + WHERE v.podcastid = $1 AND v.publishedat < $2 + AND NOT EXISTS (SELECT 1 FROM "UserVideoHistory" h WHERE h.videoid = v.videoid) + AND NOT EXISTS (SELECT 1 FROM "SavedVideos" s WHERE s.videoid = v.videoid) + AND NOT EXISTS (SELECT 1 FROM "DownloadedVideos" d WHERE d.videoid = v.videoid) + AND NOT EXISTS (SELECT 1 FROM "PlaylistContents" p WHERE p.videoid = v.videoid) + AND NOT EXISTS (SELECT 1 FROM "CollectionEpisodes" c WHERE c.videoid = v.videoid) + AND NOT EXISTS (SELECT 1 FROM "EpisodeQueue" q WHERE q.episodeid = v.videoid AND q.is_youtube = TRUE) + "#) + .bind(podcast_id) + .bind(cutoff_naive) + .execute(pool) + .await? + .rows_affected() } DatabasePool::MySQL(pool) => { - // First, delete all references from dependent tables - let cleanup_queries = vec![ - "DELETE FROM UserVideoHistory WHERE VideoID IN (SELECT VideoID FROM YouTubeVideos WHERE PodcastID = ? AND PublishedAt < ?)", - "DELETE FROM SavedVideos WHERE VideoID IN (SELECT VideoID FROM YouTubeVideos WHERE PodcastID = ? AND PublishedAt < ?)", - "DELETE FROM DownloadedVideos WHERE VideoID IN (SELECT VideoID FROM YouTubeVideos WHERE PodcastID = ? AND PublishedAt < ?)", - "DELETE FROM PlaylistContents WHERE VideoID IN (SELECT VideoID FROM YouTubeVideos WHERE PodcastID = ? AND PublishedAt < ?)", - "DELETE FROM EpisodeQueue WHERE EpisodeID IN (SELECT VideoID FROM YouTubeVideos WHERE PodcastID = ? AND PublishedAt < ?)", - ]; - - for query in cleanup_queries { - sqlx::query(query) - .bind(podcast_id) - .bind(cutoff_naive) - .execute(pool) - .await?; - } - - // Now delete the videos themselves - sqlx::query("DELETE FROM YouTubeVideos WHERE PodcastID = ? AND PublishedAt < ?") - .bind(podcast_id) - .bind(cutoff_naive) - .execute(pool) - .await? - .rows_affected() + sqlx::query(" + DELETE v FROM YouTubeVideos v + WHERE v.PodcastID = ? AND v.PublishedAt < ? + AND NOT EXISTS (SELECT 1 FROM UserVideoHistory h WHERE h.VideoID = v.VideoID) + AND NOT EXISTS (SELECT 1 FROM SavedVideos s WHERE s.VideoID = v.VideoID) + AND NOT EXISTS (SELECT 1 FROM DownloadedVideos d WHERE d.VideoID = v.VideoID) + AND NOT EXISTS (SELECT 1 FROM PlaylistContents p WHERE p.VideoID = v.VideoID) + AND NOT EXISTS (SELECT 1 FROM CollectionEpisodes c WHERE c.VideoID = v.VideoID) + AND NOT EXISTS (SELECT 1 FROM EpisodeQueue q WHERE q.EpisodeID = v.VideoID AND q.is_youtube = 1) + ") + .bind(podcast_id) + .bind(cutoff_naive) + .execute(pool) + .await? + .rows_affected() } }; - - debug!("Removed {} old YouTube videos", rows_affected); + if rows_affected > 0 { + debug!("Removed {} old YouTube videos of podcast {}", rows_affected, podcast_id); + } Ok(()) } - - // Get existing YouTube videos - matches Python get_existing_youtube_videos function exactly - pub async fn get_existing_youtube_videos(&self, podcast_id: i32) -> AppResult> { - info!("Getting existing YouTube videos for podcast {}", podcast_id); - - let mut video_urls = Vec::new(); - - match self { + + // YouTube IDs of a channel podcast's stored videos. + pub async fn youtube_video_ids(&self, podcast_id: i32) -> AppResult> { + let ids: Vec> = match self { DatabasePool::Postgres(pool) => { - let rows = sqlx::query(r#"SELECT videourl FROM "YouTubeVideos" WHERE podcastid = $1"#) + sqlx::query_scalar(r#"SELECT youtubevideoid FROM "YouTubeVideos" WHERE podcastid = $1"#) .bind(podcast_id) .fetch_all(pool) - .await?; - - for row in rows { - let url: String = row.try_get("videourl")?; - video_urls.push(url); - } + .await? } DatabasePool::MySQL(pool) => { - let rows = sqlx::query("SELECT VideoURL FROM YouTubeVideos WHERE PodcastID = ?") + sqlx::query_scalar("SELECT YouTubeVideoID FROM YouTubeVideos WHERE PodcastID = ?") .bind(podcast_id) .fetch_all(pool) - .await?; - - for row in rows { - let url: String = row.try_get("VideoURL")?; - video_urls.push(url); - } + .await? } - } - - debug!("Found {} existing videos", video_urls.len()); - Ok(video_urls) + }; + Ok(ids.into_iter().flatten().collect()) } - - // Add YouTube videos - matches Python add_youtube_videos function exactly - pub async fn add_youtube_videos(&self, podcast_id: i32, videos: &[serde_json::Value]) -> AppResult<()> { - debug!("Adding {} YouTube videos for podcast {}", videos.len(), podcast_id); - + + // Insert videos into a channel podcast; returns the inserted rows. + pub async fn add_youtube_videos( + &self, + podcast_id: i32, + videos: &[&crate::handlers::youtube::VideoDetails], + ) -> AppResult> { + let mut inserted = Vec::with_capacity(videos.len()); for video in videos { - let video_id = video.get("id").and_then(|v| v.as_str()).unwrap_or(""); - let raw_title = video.get("title").and_then(|v| v.as_str()).unwrap_or(""); - let raw_description = video.get("description").and_then(|v| v.as_str()).unwrap_or(""); - let title = self.clean_and_normalize_title(raw_title); - let description = self.decode_html_entities(raw_description); - let url = video.get("url").and_then(|v| v.as_str()).unwrap_or(""); - let thumbnail = video.get("thumbnail").and_then(|v| v.as_str()).unwrap_or(""); - - debug!("Processing video {} for database insertion", video_id); - info!("Video data: {:?}", video); - - let duration = if let Some(duration_str) = video.get("duration").and_then(|v| v.as_str()) { - info!("Duration as string: '{}'", duration_str); - let parsed = crate::handlers::youtube::parse_youtube_duration(duration_str).unwrap_or(0) as i32; - debug!("Parsed duration: {}", parsed); - parsed - } else { - let int_duration = video.get("duration").and_then(|v| v.as_i64()).unwrap_or(0) as i32; - info!("Duration as integer: {}", int_duration); - int_duration - }; - - // Parse publish date - let publish_date = if let Some(date_str) = video.get("publish_date").and_then(|v| v.as_str()) { - chrono::DateTime::parse_from_rfc3339(date_str) - .map(|dt| dt.naive_utc()) - .unwrap_or_else(|_| chrono::Utc::now().naive_utc()) - } else { - chrono::Utc::now().naive_utc() - }; - - match self { + let title = self.clean_and_normalize_title(&video.title); + let description = self.decode_html_entities(&video.description); + let url = format!("https://www.youtube.com/watch?v={}", video.id); + let published = video.published.naive_utc(); + let video_id: i32 = match self { DatabasePool::Postgres(pool) => { - let _ = sqlx::query(r#" + sqlx::query_scalar(r#" INSERT INTO "YouTubeVideos" ( podcastid, youtubevideoid, videotitle, videodescription, videourl, thumbnailurl, publishedat, duration, completed, listenposition - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, FALSE, 0) + RETURNING videoid "#) - .bind(podcast_id) - .bind(video_id) - .bind(title) - .bind(description) - .bind(url) - .bind(thumbnail) - .bind(publish_date) - .bind(duration) - .bind(false) // Not completed - .bind(0) // Listen position 0 - .execute(pool) - .await; + .bind(podcast_id) + .bind(&video.id) + .bind(&title) + .bind(&description) + .bind(&url) + .bind(&video.thumbnail) + .bind(published) + .bind(video.duration) + .fetch_one(pool) + .await? } DatabasePool::MySQL(pool) => { - let _ = sqlx::query(r#" - INSERT IGNORE INTO YouTubeVideos ( + sqlx::query(r#" + INSERT INTO YouTubeVideos ( PodcastID, YouTubeVideoID, VideoTitle, VideoDescription, VideoURL, ThumbnailURL, PublishedAt, Duration, Completed, ListenPosition - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, 0) "#) - .bind(podcast_id) - .bind(video_id) - .bind(title) - .bind(description) - .bind(url) - .bind(thumbnail) - .bind(publish_date) - .bind(duration) - .bind(false) // Not completed - .bind(0) // Listen position 0 - .execute(pool) - .await; + .bind(podcast_id) + .bind(&video.id) + .bind(&title) + .bind(&description) + .bind(&url) + .bind(&video.thumbnail) + .bind(published) + .bind(video.duration) + .execute(pool) + .await? + .last_insert_id() as i32 } - } + }; + inserted.push(crate::handlers::podcasts::Episode { + podcastid: podcast_id, + podcastname: String::new(), + episodetitle: title, + episodepubdate: video.published.format("%Y-%m-%dT%H:%M:%S").to_string(), + episodedescription: description, + episodeartwork: video.thumbnail.clone(), + episodeurl: url, + episodeduration: video.duration, + listenduration: None, + episodeid: video_id, + completed: false, + saved: false, + queued: false, + downloaded: false, + is_youtube: true, + is_video: true, + }); } - - debug!("Successfully added {} YouTube videos", videos.len()); - Ok(()) + Ok(inserted) } - + // Get video date using web scraping - matches Python get_video_date function exactly pub async fn get_video_date(&self, video_id: &str) -> AppResult> { let client = reqwest::Client::new(); @@ -17022,7 +16977,7 @@ impl DatabasePool { // Update episode count for podcast - matches Python update_episode_count function exactly pub async fn update_episode_count(&self, podcast_id: i32) -> AppResult<()> { - info!("Updating episode count for podcast {}", podcast_id); + debug!("Updating episode count for podcast {}", podcast_id); // Count episodes and YouTube videos let (episode_count, youtube_count) = match self { @@ -20310,7 +20265,7 @@ impl DatabasePool { // Update YouTube video duration after download - updates duration from MP3 file pub async fn update_youtube_video_duration(&self, video_id: &str, duration_seconds: i32) -> AppResult<()> { - info!("Updating duration for YouTube video {} to {} seconds", video_id, duration_seconds); + debug!("Updating duration for YouTube video {} to {} seconds", video_id, duration_seconds); match self { DatabasePool::Postgres(pool) => { diff --git a/rust-api/src/handlers/refresh.rs b/rust-api/src/handlers/refresh.rs index ded2295ac..f763a977b 100644 --- a/rust-api/src/handlers/refresh.rs +++ b/rust-api/src/handlers/refresh.rs @@ -455,13 +455,32 @@ async fn mark_refreshed(state: &AppState, ids: &[i32], next: Schedule) { feed_health::after_success(&state.task_manager, recovered).await; } -/// Import a newly added YouTube channel's videos and schedule its refreshes. Holds the channel's -/// refresh lock so a scheduled refresh can't run alongside. +/// Import a newly added YouTube channel's videos and schedule its refreshes. The channel is +/// refreshed for all its subscribers, holding its refresh lock so a scheduled refresh can't run +/// alongside. pub async fn import_added_channel(state: &AppState, podcast_id: i32, channel_id: &str, feed_cutoff: i32) -> AppResult<()> { let url = format!("https://www.youtube.com/channel/{}", channel_id); let _lock = FeedLock::acquire(FeedLock::key(&url, None, None)).await; - crate::handlers::youtube::process_youtube_channel(podcast_id, channel_id, feed_cutoff, state).await?; - mark_refreshed(state, &[podcast_id], youtube_schedule(state, podcast_id).await).await; + let feed_id = Some(state.db_pool.feed_of_podcast(podcast_id).await?); + let item = PodcastRefreshItem { + podcast_id, + feed_id, + feed_url: url, + artwork_url: None, + auto_download: false, + auto_queue: false, + username: None, + password: None, + is_youtube: true, + user_id: 0, + feed_cutoff: Some(feed_cutoff), + etag: None, + last_modified: None, + content_hash: None, + refresh_interval: None, + }; + let group = feed_group(state, feed_id, item).await?; + refresh_youtube_group(state, &group).await?; Ok(()) } @@ -474,43 +493,94 @@ async fn mark_failed(state: &AppState, ids: &[i32], kind: FeedErrorKind, error: feed_health::after_failure(&state.db_pool, &state.task_manager, ids).await; } -/// Refresh a single YouTube channel podcast. -async fn refresh_youtube_item(state: &AppState, item: &PodcastRefreshItem) { - let channel_id = if item.feed_url.contains("channel/") { - item.feed_url - .split("channel/") - .nth(1) - .unwrap_or(&item.feed_url) - .split('/') - .next() - .unwrap_or(&item.feed_url) - .split('?') - .next() - .unwrap_or(&item.feed_url) - .to_string() - } else { - item.feed_url.clone() +/// Refresh one YouTube channel for every subscriber podcast in the group: its feed is fetched once +/// (conditional GET + body hash, yt-dlp's listing when the feed fails), yt-dlp fetches only the +/// videos a subscriber is missing, and each subscriber gets its own new rows. The caller holds the +/// channel's lock. Returns every inserted video across the group; Err only when the channel failed. +async fn refresh_youtube_group( + state: &AppState, + group: &[PodcastRefreshItem], +) -> AppResult> { + use crate::handlers::youtube::{self, ChannelListing, ChannelSubscriber}; + let rep = &group[0]; + let ids: Vec = group.iter().map(|i| i.podcast_id).collect(); + let channel_id = youtube::channel_id_from_url(&rep.feed_url); + + // Validators are only trusted when every subscriber applied the stored feed; a new + // subscription gets a full fetch. + let content_hash = rep + .content_hash + .clone() + .filter(|h| group.iter().all(|i| i.content_hash.as_deref() == Some(h.as_str()))); + let validators = match &content_hash { + Some(hash) => crate::database::FeedValidators { + etag: rep.etag.as_deref(), + last_modified: rep.last_modified.as_deref(), + content_hash: Some(hash), + }, + None => crate::database::FeedValidators::default(), }; - let ids = [item.podcast_id]; - match crate::handlers::youtube::process_youtube_channel( - item.podcast_id, - &channel_id, - item.feed_cutoff.unwrap_or(30), - state, - ) - .await - { - Ok(_) => { - info!("Successfully refreshed YouTube channel {}", item.podcast_id); - mark_refreshed(state, &ids, youtube_schedule(state, item.podcast_id).await).await; + + match youtube::list_channel_videos(state, channel_id, validators).await { + Ok(ChannelListing::Unchanged { etag, last_modified }) => { + if etag.is_some() || last_modified.is_some() { + let _ = state + .db_pool + .store_feed_cache_validators(&ids, etag.as_deref(), last_modified.as_deref(), None) + .await; + } + mark_refreshed(state, &ids, youtube_schedule(state, rep.podcast_id).await).await; + Ok(Vec::new()) + } + Ok(ChannelListing::Listed { videos, etag, last_modified, content_hash }) => { + let subscribers: Vec = group + .iter() + .map(|i| ChannelSubscriber { podcast_id: i.podcast_id, feed_cutoff: i.feed_cutoff.unwrap_or(30) }) + .collect(); + match youtube::process_youtube_channel(state, &subscribers, &videos).await { + Ok(new_videos) => { + if !new_videos.is_empty() { + info!("YouTube channel {}: {} new videos", channel_id, new_videos.len()); + } + let _ = state + .db_pool + .store_feed_cache_validators(&ids, etag.as_deref(), last_modified.as_deref(), content_hash.as_deref()) + .await; + mark_refreshed(state, &ids, youtube_schedule(state, rep.podcast_id).await).await; + Ok(new_videos) + } + Err(e) => { + warn!("Error refreshing YouTube channel {}: {}", channel_id, e); + mark_failed(state, &ids, FeedErrorKind::of(&e), &e.to_string(), None).await; + Err(e) + } + } + } + Ok(ChannelListing::Throttled { status, retry_after }) => { + let msg = format!("YouTube throttled refresh (HTTP {})", status); + mark_failed(state, &ids, FeedErrorKind::Throttled, &msg, retry_after).await; + Err(AppError::external_error(&msg)) } Err(e) => { - warn!("Error refreshing YouTube channel {}: {}", item.podcast_id, e); + warn!("Error refreshing YouTube channel {}: {}", channel_id, e); mark_failed(state, &ids, FeedErrorKind::of(&e), &e.to_string(), None).await; + Err(e) } } } +/// The podcasts subscribed to this feed, with `item` among them. +async fn feed_group(state: &AppState, feed_id: Option, item: PodcastRefreshItem) -> AppResult> { + let mut group = match feed_id { + Some(feed_id) => load_refresh_items(state, None, Some(&[feed_id]), true).await?, + None => Vec::new(), + }; + if !group.iter().any(|i| i.podcast_id == item.podcast_id) { + group.insert(0, item); + } + Ok(group) +} + /// The podcast rows to refresh (all, or those of the `only` feeds), skipping feeds a host asked us /// to leave alone (Retry-After) and, unless `ignore_backoff`, feeds in failure backoff. async fn load_refresh_items( @@ -655,20 +725,19 @@ pub async fn refresh_feed_urls(state: &AppState, urls: &[String]) -> AppResult) -> usize { - // Partition into YouTube items (refreshed per-podcast) and RSS feeds grouped by feed, so a - // feed subscribed by many users is fetched + parsed ONCE. A row not linked to its feed yet - // groups by address and credentials, which is what a feed is. + // Group by feed, so a feed or channel subscribed by many users is fetched + parsed ONCE. A row + // not linked to its feed yet groups by address and credentials, which is what a feed is. use std::collections::HashMap; #[derive(PartialEq, Eq, Hash)] enum Group { Feed(i32), Unlinked(FeedKey), } - let mut youtube_items: Vec = Vec::new(); + let mut youtube_groups: HashMap> = HashMap::new(); let mut feed_groups: HashMap> = HashMap::new(); let owners: HashMap = refresh_items.iter().map(|i| (i.podcast_id, i.user_id)).collect(); @@ -677,28 +746,26 @@ async fn refresh_items(state: &AppState, refresh_items: Vec) debug!("Skipping local podcast {} - not an RSS feed", item.podcast_id); continue; } - if item.is_youtube { - youtube_items.push(item); - } else { - let key = match item.feed_id { - Some(id) => Group::Feed(id), - None => Group::Unlinked(FeedLock::key(&item.feed_url, item.username.as_deref(), item.password.as_deref())), - }; - feed_groups.entry(key).or_default().push(item); - } + let key = match item.feed_id { + Some(id) => Group::Feed(id), + None => Group::Unlinked(FeedLock::key(&item.feed_url, item.username.as_deref(), item.password.as_deref())), + }; + let groups = if item.is_youtube { &mut youtube_groups } else { &mut feed_groups }; + groups.entry(key).or_default().push(item); } use futures::stream::{self, StreamExt}; // Refresh unique RSS feeds with bounded concurrency. let groups: Vec> = feed_groups.into_values().collect(); - let feeds = groups.len() + youtube_items.len(); + let channels: Vec> = youtube_groups.into_values().collect(); + let feeds = groups.len() + channels.len(); let subscription_count: usize = groups.iter().map(|g| g.len()).sum(); debug!( "Running refresh: {} unique RSS feeds ({} subscriptions), {} YouTube channels (concurrency={})", groups.len(), subscription_count, - youtube_items.len(), + channels.len(), REFRESH_CONCURRENCY ); // Users whose podcasts got new episodes, for their playlist counts. @@ -717,11 +784,16 @@ async fn refresh_items(state: &AppState, refresh_items: Vec) }) .await; - // Refresh YouTube channels with bounded concurrency. - stream::iter(youtube_items) - .for_each_concurrent(Some(REFRESH_CONCURRENCY), |item| async move { - let _lock = FeedLock::acquire(FeedLock::key(&item.feed_url, None, None)).await; - refresh_youtube_item(state, &item).await; + // Refresh YouTube channels with bounded concurrency (yt-dlp has its own, lower limit). + stream::iter(channels) + .for_each_concurrent(Some(REFRESH_CONCURRENCY), |group| { + let (owners, touched) = (&owners, &touched); + async move { + let _lock = FeedLock::acquire(FeedLock::key(&group[0].feed_url, None, None)).await; + if let Ok(new_videos) = refresh_youtube_group(state, &group).await { + touched.lock().unwrap().extend(new_videos.iter().filter_map(|e| owners.get(&e.podcastid))); + } + } }) .await; @@ -1050,7 +1122,7 @@ async fn refresh_single_podcast( info!("Refreshing podcast: {} (ID: {})", podcast.name, podcast.id); if podcast.is_youtube { - refresh_youtube_channel(state, podcast).await + refresh_youtube_channel(state, podcast, user_id).await } else { refresh_rss_feed(state, podcast, user_id).await } @@ -1101,13 +1173,7 @@ async fn refresh_rss_feed( }; // The feed is refreshed for everyone subscribed to it: its schedule and validators are // shared, so refreshing one subscription alone would leave the others behind. - let mut group = match podcast.feed_id { - Some(feed_id) => load_refresh_items(state, None, Some(&[feed_id]), true).await?, - None => Vec::new(), - }; - if !group.iter().any(|i| i.podcast_id == item.podcast_id) { - group.insert(0, item); - } + let group = feed_group(state, podcast.feed_id, item).await?; let mut new_episodes = refresh_feed_group(state, &group).await?; new_episodes.retain(|e| e.podcastid == podcast.id); @@ -1121,40 +1187,43 @@ async fn refresh_rss_feed( async fn refresh_youtube_channel( state: &AppState, podcast: &PodcastForRefresh, + user_id: i32, ) -> AppResult> { debug!("Refreshing YouTube channel: {}", podcast.name); - let channel_id = if podcast.feed_url.contains("channel/") { - podcast.feed_url.split("channel/").nth(1).unwrap_or(&podcast.feed_url).split('/').next().unwrap_or(&podcast.feed_url).split('?').next().unwrap_or(&podcast.feed_url) - } else { - &podcast.feed_url - }; - + if podcast.recently_refreshed || podcast.throttled { + debug!( + "Skipping manual refresh of channel {} (recently refreshed: {}, throttled: {})", + podcast.id, podcast.recently_refreshed, podcast.throttled + ); + return Ok(Vec::new()); + } let Some(_lock) = FeedLock::try_acquire(FeedLock::key(&podcast.feed_url, None, None)) else { debug!("Skipping manual refresh of channel {}: it's being refreshed right now", podcast.id); return Ok(Vec::new()); }; - let ids = [podcast.id]; - match crate::handlers::youtube::process_youtube_channel( - podcast.id, - channel_id, - podcast.feed_cutoff_days.unwrap_or(30), - state, - ).await { - Ok(_) => { - info!("Successfully refreshed YouTube channel: {}", podcast.name); - mark_refreshed(state, &ids, youtube_schedule(state, podcast.id).await).await; - // YouTube new-episode surfacing over the websocket is not yet wired up - // (process_youtube_channel does not return the inserted rows); the channel is still - // refreshed and new videos appear on the next list load. - Ok(Vec::new()) - } - Err(e) => { - warn!("Error refreshing YouTube channel {}: {}", podcast.name, e); - mark_failed(state, &ids, FeedErrorKind::of(&e), &e.to_string(), None).await; - Err(e) - } - } + // Like a feed, the channel is refreshed for everyone subscribed to it. + let item = PodcastRefreshItem { + podcast_id: podcast.id, + feed_id: podcast.feed_id, + feed_url: podcast.feed_url.clone(), + artwork_url: podcast.artwork_url.clone(), + auto_download: podcast.auto_download, + auto_queue: podcast.auto_queue, + username: None, + password: None, + is_youtube: true, + user_id, + feed_cutoff: podcast.feed_cutoff_days, + etag: podcast.etag.clone(), + last_modified: podcast.last_modified.clone(), + content_hash: podcast.content_hash.clone(), + refresh_interval: podcast.refresh_interval, + }; + let group = feed_group(state, podcast.feed_id, item).await?; + let mut new_videos = refresh_youtube_group(state, &group).await?; + new_videos.retain(|e| e.podcastid == podcast.id); + Ok(new_videos) } // Define sync result structure to match our database return type diff --git a/rust-api/src/handlers/youtube.rs b/rust-api/src/handlers/youtube.rs index 2ba28e9c9..2a2b575fe 100644 --- a/rust-api/src/handlers/youtube.rs +++ b/rust-api/src/handlers/youtube.rs @@ -248,7 +248,7 @@ pub async fn subscribe_to_youtube_channel( let feed_cutoff = query.feed_cutoff.unwrap_or(30); - info!("Starting subscription for channel {}", query.channel_id); + debug!("Starting subscription for channel {}", query.channel_id); // Check if channel already exists let existing_id = state.db_pool.check_existing_channel_subscription( @@ -257,7 +257,7 @@ pub async fn subscribe_to_youtube_channel( ).await?; if let Some(podcast_id) = existing_id { - info!("Channel {} already subscribed", query.channel_id); + debug!("Channel {} already subscribed", query.channel_id); return Ok(Json(serde_json::json!({ "success": true, "podcast_id": podcast_id, @@ -265,7 +265,7 @@ pub async fn subscribe_to_youtube_channel( }))); } - info!("Getting channel info"); + debug!("Getting channel info"); let channel_info = get_youtube_channel_info(&query.channel_id).await?; debug!("Adding channel to database"); @@ -291,156 +291,65 @@ pub async fn subscribe_to_youtube_channel( }))) } -// Converts yt-dlp float seconds to ISO 8601 duration (e.g. 253.0 → "PT4M13S"). -// parse_youtube_duration() below expects this format. -fn seconds_to_pt_duration(seconds: f64) -> String { - let total = seconds as u64; - let h = total / 3600; - let m = (total % 3600) / 60; - let s = total % 60; - if h > 0 { - format!("PT{}H{}M{}S", h, m, s) - } else if m > 0 { - format!("PT{}M{}S", m, s) - } else { - format!("PT{}S", s) - } +// Run the local yt-dlp binary (#793). +async fn run_ytdlp(args: &[&str]) -> Result { + let bin = ytdlp_binary(); + Command::new(&bin).args(args).output().await.map_err(|e| { + error!("Failed to execute yt-dlp {}: {}", bin, e); + AppError::external_error(&format!("Failed to execute yt-dlp: {}", e)) + }) } -// Converts yt-dlp upload_date "YYYYMMDD" to RFC3339 "YYYY-MM-DDT00:00:00Z". -// process_youtube_channel parses publishedAt with DateTime::parse_from_rfc3339. -fn upload_date_to_rfc3339(upload_date: &str) -> String { - if upload_date.len() == 8 { - format!("{}-{}-{}T00:00:00Z", &upload_date[..4], &upload_date[4..6], &upload_date[6..8]) - } else { - chrono::Utc::now().to_rfc3339() - } +// yt-dlp prints one JSON object per line for --dump-json. +fn json_lines(stdout: &[u8]) -> Vec { + String::from_utf8_lossy(stdout) + .lines() + .filter_map(|line| serde_json::from_str(line).ok()) + .collect() } -// Fetch YouTube channel metadata + recent videos by running yt-dlp locally (#793). -// Returns the same camelCase JSON shape the hosted Backend used to return -// ({ name, description, thumbnailUrl, recentVideos: [{ id, title, description, url, -// thumbnail, publishedAt, duration }] }) so callers below are unchanged. This replaces the -// old HTTP hop to SEARCH_API_URL — YouTube now resolves entirely on the local, self-updating -// yt-dlp binary. -async fn fetch_youtube_channel_data(channel_id: &str) -> Result { - let channel_url = format!("https://www.youtube.com/channel/{}/videos", channel_id); - let bin = ytdlp_binary(); - let output = Command::new(&bin) - .args(&[ - "--quiet", - "--no-warnings", - "--skip-download", - "--dump-json", - "--playlist-end", "15", - &channel_url, - ]) - .output() - .await - .map_err(|e| { - error!("yt-dlp channel fetch: failed to execute {} for {}: {}", bin, channel_id, e); - AppError::external_error(&format!("Failed to execute yt-dlp: {}", e)) - })?; +fn video_thumbnail(entry: &serde_json::Value, video_id: &str) -> String { + entry.get("thumbnail").and_then(|v| v.as_str()).map(str::to_string) + .unwrap_or_else(|| format!("https://i.ytimg.com/vi/{}/hqdefault.jpg", video_id)) +} +// Channel name and artwork for a new subscription, from yt-dlp's data on its newest video. +pub async fn get_youtube_channel_info(channel_id: &str) -> Result, AppError> { + debug!("Getting channel info for {} via local yt-dlp", channel_id); + let channel_url = format!("https://www.youtube.com/channel/{}/videos", channel_id); + let output = run_ytdlp(&[ + "--quiet", + "--no-warnings", + "--skip-download", + "--dump-json", + "--playlist-end", "1", + &channel_url, + ]).await?; if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); error!("yt-dlp channel fetch failed for {}: {}", channel_id, stderr.trim()); return Err(AppError::external_error(&format!("yt-dlp channel fetch failed: {}", stderr.trim()))); } + let entries = json_lines(&output.stdout); + let first = entries.first().ok_or_else(|| AppError::not_found("Channel not found or has no videos"))?; - let stdout = String::from_utf8_lossy(&output.stdout); - let mut entries: Vec = Vec::new(); - for line in stdout.lines() { - if let Ok(entry) = serde_json::from_str::(line) { - entries.push(entry); - } - } - - if entries.is_empty() { - return Err(AppError::not_found("Channel not found or has no videos")); - } - - let first = &entries[0]; - let channel_name = first.get("channel").and_then(|v| v.as_str()) + let name = first.get("channel").and_then(|v| v.as_str()) .or_else(|| first.get("uploader").and_then(|v| v.as_str())) - .unwrap_or("").to_string(); - // channel_thumbnail is a URL string in full-metadata mode; fall back to first video thumbnail + .unwrap_or(""); let first_video_id = first.get("id").and_then(|v| v.as_str()).unwrap_or(""); - let thumbnail_url = first.get("channel_thumbnail").and_then(|v| v.as_str()) - .or_else(|| first.get("thumbnail").and_then(|v| v.as_str())) - .map(|s| s.to_string()) - .or_else(|| { - if !first_video_id.is_empty() { - Some(format!("https://i.ytimg.com/vi/{}/hqdefault.jpg", first_video_id)) - } else { - None - } - }) - .unwrap_or_default(); - - let recent_videos: Vec = entries.iter().map(|entry| { - let video_id = entry.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(); - let upload_date = entry.get("upload_date").and_then(|v| v.as_str()).unwrap_or(""); - let duration_secs = entry.get("duration").and_then(|v| v.as_f64()).unwrap_or(0.0); - let thumb = entry.get("thumbnail").and_then(|v| v.as_str()) - .map(|s| s.to_string()) - .or_else(|| { - if !video_id.is_empty() { - Some(format!("https://i.ytimg.com/vi/{}/hqdefault.jpg", video_id)) - } else { - None - } - }) - .unwrap_or_default(); - serde_json::json!({ - "id": video_id, - "title": entry.get("title").and_then(|v| v.as_str()).unwrap_or(""), - "description": entry.get("description").and_then(|v| v.as_str()).unwrap_or(""), - "url": format!("https://www.youtube.com/watch?v={}", video_id), - "thumbnail": thumb, - "publishedAt": upload_date_to_rfc3339(upload_date), - "duration": seconds_to_pt_duration(duration_secs), - }) - }).collect(); - - Ok(serde_json::json!({ - "channelId": channel_id, - "name": channel_name, - "description": "", - "thumbnailUrl": thumbnail_url, - "url": format!("https://www.youtube.com/channel/{}", channel_id), - "recentVideos": recent_videos, - })) -} - -// Helper function to get YouTube channel info using the local yt-dlp binary -pub async fn get_youtube_channel_info(channel_id: &str) -> Result, AppError> { - info!("Getting channel info for {} via local yt-dlp", channel_id); - - let channel_data = fetch_youtube_channel_data(channel_id).await?; + let thumbnail = first.get("channel_thumbnail").and_then(|v| v.as_str()).map(str::to_string) + .unwrap_or_else(|| video_thumbnail(first, first_video_id)); - // Extract channel info (same shape the Backend previously returned) let mut channel_info = HashMap::new(); - channel_info.insert("channel_id".to_string(), channel_id.to_string()); - channel_info.insert("name".to_string(), - channel_data.get("name").and_then(|v| v.as_str()).unwrap_or("").to_string()); - - let description = channel_data.get("description") - .and_then(|v| v.as_str()) - .unwrap_or("") - .chars() - .take(500) - .collect::(); - channel_info.insert("description".to_string(), description); - - channel_info.insert("thumbnail_url".to_string(), - channel_data.get("thumbnailUrl").and_then(|v| v.as_str()).unwrap_or("").to_string()); - - info!("Successfully extracted channel info for: {}", channel_info.get("name").cloned().unwrap_or_default()); + channel_info.insert("name".to_string(), name.to_string()); + channel_info.insert("description".to_string(), String::new()); + channel_info.insert("thumbnail_url".to_string(), thumbnail); + debug!("Extracted channel info for: {}", name); Ok(channel_info) } + // Helper function to get MP3 duration from file pub fn get_mp3_duration(file_path: &str) -> Option { match mp3_metadata::read_from_file(file_path) { @@ -452,251 +361,343 @@ pub fn get_mp3_duration(file_path: &str) -> Option { } } -// Helper function to parse YouTube duration format (PT4M13S) to seconds -pub fn parse_youtube_duration(duration_str: &str) -> Option { - if !duration_str.starts_with("PT") { - return None; +/// The channel ID in a YouTube channel podcast's feed URL (`https://www.youtube.com/channel/`). +pub fn channel_id_from_url(feed_url: &str) -> &str { + match feed_url.split_once("channel/") { + Some((_, rest)) => rest.split(['/', '?']).next().unwrap_or(rest), + None => feed_url, } - - let duration_part = &duration_str[2..]; // Remove "PT" - let mut total_seconds = 0i64; - let mut current_number = String::new(); - - for ch in duration_part.chars() { - if ch.is_ascii_digit() { - current_number.push(ch); - } else { - if let Ok(num) = current_number.parse::() { - match ch { - 'H' => total_seconds += num * 3600, - 'M' => total_seconds += num * 60, - 'S' => total_seconds += num, - _ => {} - } - } - current_number.clear(); - } - } - - Some(total_seconds) } -// Process YouTube channel videos using the local yt-dlp binary -pub async fn process_youtube_channel( - podcast_id: i32, - channel_id: &str, - feed_cutoff: i32, +/// One of a channel's newest videos. +#[derive(Debug, Clone)] +pub struct ListedVideo { + pub id: String, + /// Unknown when the list came from yt-dlp's flat listing. + pub published: Option>, +} + +/// What a look at a channel found (see `list_channel_videos`). +pub enum ChannelListing { + /// The feed hasn't changed since its subscribers last applied it. + Unchanged { etag: Option, last_modified: Option }, + /// The newest videos, with the validators and body hash to store once they're applied (no + /// hash when the list came from yt-dlp). + Listed { + videos: Vec, + etag: Option, + last_modified: Option, + content_hash: Option, + }, + /// 429/503 from YouTube. + Throttled { status: u16, retry_after: Option }, +} + +/// A channel's newest videos from its RSS feed (conditional GET, body hash), or from yt-dlp's flat +/// listing when the feed can't be read. Err when neither works, with the feed's error. +pub async fn list_channel_videos( state: &AppState, -) -> Result<(), AppError> { - debug!("Processing YouTube channel: podcast_id={} channel_id={}", podcast_id, channel_id); - - let cutoff_date = chrono::Utc::now() - chrono::Duration::days(feed_cutoff as i64); - debug!("Cutoff date set to: {}", cutoff_date); - - // Clean up old videos - debug!("Cleaning up videos older than cutoff date..."); - state.db_pool.remove_old_youtube_videos(podcast_id, cutoff_date).await?; - - // Fetch channel data via the local yt-dlp binary (#793) - debug!("Fetching channel data via local yt-dlp for {}", channel_id); - let channel_data = fetch_youtube_channel_data(channel_id).await?; - - // Self-heal: update podcast name/artwork if they were empty from a broken initial subscription - let channel_name_update = channel_data.get("name").and_then(|v| v.as_str()).unwrap_or(""); - let channel_thumb_update = channel_data.get("thumbnailUrl").and_then(|v| v.as_str()).unwrap_or(""); - if !channel_name_update.is_empty() || !channel_thumb_update.is_empty() { - match &state.db_pool { - crate::database::DatabasePool::Postgres(pool) => { - let _ = sqlx::query(r#" - UPDATE "Podcasts" - SET podcastname = CASE WHEN podcastname = '' OR podcastname IS NULL THEN $2 ELSE podcastname END, - artworkurl = CASE WHEN artworkurl = '' OR artworkurl IS NULL THEN $3 ELSE artworkurl END, - author = CASE WHEN author = '' OR author IS NULL THEN $2 ELSE author END - WHERE podcastid = $1 - "#) - .bind(podcast_id) - .bind(channel_name_update) - .bind(channel_thumb_update) - .execute(pool) - .await; + channel_id: &str, + prev: crate::database::FeedValidators<'_>, +) -> Result { + use crate::database::{FeedFetch, FeedValidators}; + let url = state.config.api.youtube_feed_url.replace("{channel_id}", channel_id); + // The body hash is our own (see channel_feed_hash), so it's compared here. + let validators = FeedValidators { content_hash: None, ..prev }; + let error = match state.db_pool.fetch_feed_conditional(&url, None, None, validators).await { + Ok(FeedFetch::NotModified { etag, last_modified, .. }) => { + return Ok(ChannelListing::Unchanged { etag, last_modified }); + } + Ok(FeedFetch::Fetched { body, etag, last_modified, .. }) => { + let hash = channel_feed_hash(&body); + if prev.content_hash == Some(hash.as_str()) { + debug!("Channel feed unchanged (hash match): {}", channel_id); + return Ok(ChannelListing::Unchanged { etag, last_modified }); } - crate::database::DatabasePool::MySQL(pool) => { - let _ = sqlx::query(r#" - UPDATE Podcasts - SET PodcastName = CASE WHEN PodcastName = '' OR PodcastName IS NULL THEN ? ELSE PodcastName END, - ArtworkURL = CASE WHEN ArtworkURL = '' OR ArtworkURL IS NULL THEN ? ELSE ArtworkURL END, - Author = CASE WHEN Author = '' OR Author IS NULL THEN ? ELSE Author END - WHERE PodcastID = ? - "#) - .bind(channel_name_update) - .bind(channel_thumb_update) - .bind(channel_name_update) - .bind(podcast_id) - .execute(pool) - .await; + match parse_channel_feed(&body) { + Ok(videos) => { + return Ok(ChannelListing::Listed { videos, etag, last_modified, content_hash: Some(hash) }); + } + Err(e) => e, } } - info!("Healed podcast {} metadata: name='{}', thumbnail='{}'", podcast_id, channel_name_update, channel_thumb_update); + Ok(FeedFetch::Throttled { status, retry_after }) => { + return Ok(ChannelListing::Throttled { status, retry_after }); + } + Err(e) => e, + }; + warn!("Feed of YouTube channel {} failed ({}), listing it with yt-dlp", channel_id, error); + match list_channel_videos_flat(channel_id).await { + Ok(videos) => Ok(ChannelListing::Listed { videos, etag: None, last_modified: None, content_hash: None }), + Err(e) => { + warn!("yt-dlp listing of YouTube channel {} failed too: {}", channel_id, e); + Err(error) + } } +} - let empty_vec = vec![]; - let recent_videos_data = channel_data.get("recentVideos") - .and_then(|v| v.as_array()) - .unwrap_or(&empty_vec); - - debug!("Found {} total videos from Backend service", recent_videos_data.len()); - - let mut recent_videos = Vec::new(); - - // Process each video from Backend service response - for video_entry in recent_videos_data { - let video_id = video_entry.get("id").and_then(|v| v.as_str()).unwrap_or(""); - if video_id.is_empty() { - debug!("Skipping video with missing ID"); - continue; +/// Hash of a channel feed without its view counts and ratings, which change on every fetch. +fn channel_feed_hash(body: &str) -> String { + const OPEN: &str = ""; + const CLOSE: &str = ""; + let mut stable = String::with_capacity(body.len()); + let mut rest = body; + while let Some(start) = rest.find(OPEN) { + stable.push_str(&rest[..start]); + match rest[start..].find(CLOSE) { + Some(end) => rest = &rest[start + end + CLOSE.len()..], + None => { + rest = ""; + break; + } } + } + stable.push_str(rest); + crate::services::http::content_hash(&stable) +} - debug!("Processing video ID: {}", video_id); +/// The videos in a channel's Atom feed, newest first. +fn parse_channel_feed(body: &str) -> Result, AppError> { + let feed = feed_rs::parser::parse(body.as_bytes()) + .map_err(|e| AppError::external_error(&format!("Not a valid feed: {}", e)))?; + Ok(feed + .entries + .iter() + .filter_map(|entry| { + let id = entry.id.strip_prefix("yt:video:")?; + Some(ListedVideo { id: id.to_string(), published: entry.published.or(entry.updated) }) + }) + .collect()) +} - // Parse the publishedAt date from Backend service - let published_str = video_entry.get("publishedAt").and_then(|v| v.as_str()).unwrap_or(""); - let published = chrono::DateTime::parse_from_rfc3339(published_str) - .map(|dt| dt.with_timezone(&chrono::Utc)) - .unwrap_or_else(|_| { - warn!("Failed to parse date {}, using current time", published_str); - chrono::Utc::now() - }); +/// A channel's 15 newest videos from yt-dlp, without fetching each one. +async fn list_channel_videos_flat(channel_id: &str) -> Result, AppError> { + let channel_url = format!("https://www.youtube.com/channel/{}/videos", channel_id); + let _permit = crate::services::ytdlp::refresh_permit().await; + let output = run_ytdlp(&[ + "--quiet", + "--no-warnings", + "--flat-playlist", + "--dump-json", + "--playlist-end", "15", + &channel_url, + ]).await?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(AppError::external_error(&format!("yt-dlp channel listing failed: {}", stderr.trim()))); + } + Ok(json_lines(&output.stdout) + .iter() + .filter_map(|entry| entry.get("id").and_then(|v| v.as_str())) + .map(|id| ListedVideo { id: id.to_string(), published: None }) + .collect()) +} - info!("Video publish date: {}", published); +/// A video's details from `yt-dlp --dump-json`. +#[derive(Debug, Clone)] +pub struct VideoDetails { + pub id: String, + pub title: String, + pub description: String, + pub thumbnail: String, + pub published: chrono::DateTime, + /// Seconds. + pub duration: i32, + pub channel: String, +} - if published <= cutoff_date { - info!("Video {} from {} is too old, stopping processing", video_id, published); - break; - } +impl VideoDetails { + fn from_json(entry: &serde_json::Value) -> Option { + let id = entry.get("id").and_then(|v| v.as_str())?.to_string(); + let text = |key: &str| entry.get(key).and_then(|v| v.as_str()).unwrap_or("").to_string(); + let published = entry + .get("timestamp") + .and_then(|v| v.as_i64()) + .and_then(|ts| chrono::DateTime::from_timestamp(ts, 0)) + .or_else(|| { + let date = chrono::NaiveDate::parse_from_str(entry.get("upload_date")?.as_str()?, "%Y%m%d").ok()?; + Some(date.and_hms_opt(0, 0, 0)?.and_utc()) + }) + .unwrap_or_else(chrono::Utc::now); + Some(VideoDetails { + thumbnail: video_thumbnail(entry, &id), + title: text("title"), + description: text("description"), + published, + duration: entry.get("duration").and_then(|v| v.as_f64()).unwrap_or(0.0) as i32, + channel: entry.get("channel").or_else(|| entry.get("uploader")).and_then(|v| v.as_str()).unwrap_or("").to_string(), + id, + }) + } +} - // Debug: print what we got from Backend for this video - info!("Backend video data for {}: {:?}", video_id, video_entry); - let duration_str = video_entry.get("duration").and_then(|v| v.as_str()).unwrap_or(""); - info!("Duration string from Backend: '{}'", duration_str); - let parsed_duration = if !duration_str.is_empty() { - parse_youtube_duration(duration_str).unwrap_or(0) - } else { - 0 - }; - debug!("Parsed duration: {}", parsed_duration); - - let video_data = serde_json::json!({ - "id": video_id, - "title": video_entry.get("title").and_then(|v| v.as_str()).unwrap_or(""), - "description": video_entry.get("description").and_then(|v| v.as_str()).unwrap_or(""), - "url": format!("https://www.youtube.com/watch?v={}", video_id), - "thumbnail": video_entry.get("thumbnail").and_then(|v| v.as_str()).unwrap_or(""), - "publish_date": published.to_rfc3339(), - "duration": duration_str // Store as string for proper parsing in database - }); - - debug!("Successfully added video {} to processing queue", video_id); - recent_videos.push(video_data); +/// Details of these videos from a single yt-dlp run. Videos yt-dlp can't read (members-only, +/// private, upcoming premieres) are left out; Err only when yt-dlp itself failed. +async fn fetch_video_details(ids: &[String]) -> Result, AppError> { + let urls: Vec = ids.iter().map(|id| format!("https://www.youtube.com/watch?v={}", id)).collect(); + let mut args = vec!["--quiet", "--no-warnings", "--skip-download", "--ignore-errors", "--socket-timeout", "30", "--dump-json"]; + args.extend(urls.iter().map(String::as_str)); + let _permit = crate::services::ytdlp::refresh_permit().await; + let output = run_ytdlp(&args).await?; + let details: HashMap = json_lines(&output.stdout) + .iter() + .filter_map(VideoDetails::from_json) + .map(|d| (d.id.clone(), d)) + .collect(); + let stderr = String::from_utf8_lossy(&output.stderr); + // A failing video makes yt-dlp exit non-zero; only a run with no per-video errors to show for + // it is a broken yt-dlp. + if !output.status.success() && details.is_empty() && !stderr.contains("ERROR: [youtube]") { + return Err(AppError::external_error(&format!("yt-dlp video fetch failed: {}", stderr.trim()))); } + if details.len() < ids.len() { + debug!("yt-dlp skipped {} of {} videos: {}", ids.len() - details.len(), ids.len(), stderr.trim()); + } + Ok(details) +} - debug!("Processing complete - Found {} recent videos", recent_videos.len()); - - if !recent_videos.is_empty() { - info!("Starting database updates"); - - // Get existing videos - let existing_videos = state.db_pool.get_existing_youtube_videos(podcast_id).await?; - - // Filter out videos that already exist - let mut new_videos = Vec::new(); - for video in &recent_videos { - let video_url = format!("https://www.youtube.com/watch?v={}", - video.get("id").and_then(|v| v.as_str()).unwrap_or("")); - if !existing_videos.contains(&video_url) { - new_videos.push(video.clone()); - } else { - debug!("Video already exists, skipping: {}", - video.get("title").and_then(|v| v.as_str()).unwrap_or("")); - } - } +/// A channel subscription brought up to date by `process_youtube_channel`. +pub struct ChannelSubscriber { + pub podcast_id: i32, + pub feed_cutoff: i32, +} - if !new_videos.is_empty() { - state.db_pool.add_youtube_videos(podcast_id, &new_videos).await?; - debug!("Successfully added {} new videos", new_videos.len()); - } else { - info!("No new videos to add"); +/// Bring a channel's subscriptions up to date with its newest videos: drop videos past each +/// subscriber's cutoff, fetch the details of the listed videos a subscriber is missing (one +/// yt-dlp run, never past the longest cutoff), insert them, and download their audio. Returns +/// the inserted rows. +pub async fn process_youtube_channel( + state: &AppState, + subscribers: &[ChannelSubscriber], + listed: &[ListedVideo], +) -> Result, AppError> { + let now = chrono::Utc::now(); + let cutoff = |sub: &ChannelSubscriber| now - chrono::Duration::days(sub.feed_cutoff as i64); + + let mut have: Vec> = Vec::with_capacity(subscribers.len()); + for sub in subscribers { + state.db_pool.remove_old_youtube_videos(sub.podcast_id, cutoff(sub)).await?; + have.push(state.db_pool.youtube_video_ids(sub.podcast_id).await?); + } + let missing = |video: &ListedVideo, sub: &ChannelSubscriber, have: &HashSet| { + !have.contains(&video.id) && video.published.map_or(true, |p| p > cutoff(sub)) + }; + let wanted: Vec = listed + .iter() + .filter(|v| subscribers.iter().zip(&have).any(|(sub, have)| missing(v, sub, have))) + .map(|v| v.id.clone()) + .collect(); + let details = if wanted.is_empty() { HashMap::new() } else { fetch_video_details(&wanted).await? }; + + let mut new_episodes = Vec::new(); + let mut stored: HashSet = have.iter().flatten().cloned().collect(); + for (sub, have) in subscribers.iter().zip(&have) { + let videos: Vec<&VideoDetails> = wanted + .iter() + .filter(|id| !have.contains(*id)) + .filter_map(|id| details.get(id)) + .filter(|d| d.published > cutoff(sub)) + .collect(); + if let Some(first) = videos.first() { + heal_channel_metadata(state, sub.podcast_id, first).await; + let inserted = state.db_pool.add_youtube_videos(sub.podcast_id, &videos).await?; + debug!("Podcast {}: {} new videos", sub.podcast_id, inserted.len()); + stored.extend(videos.iter().map(|d| d.id.clone())); + new_episodes.extend(inserted); } + state.db_pool.update_episode_count(sub.podcast_id).await?; + } - // Download audio for recent videos - info!("Starting audio downloads"); - let mut successful_downloads = 0; - let mut failed_downloads = 0; - - for video in &recent_videos { - let video_id = video.get("id").and_then(|v| v.as_str()).unwrap_or(""); - let title = video.get("title").and_then(|v| v.as_str()).unwrap_or(""); - - let output_path = format!("/opt/pinepods/downloads/youtube/{}.mp3", video_id); - let output_path_double = format!("{}.mp3", output_path); + // Audio of the listed videos a subscriber keeps (so never past the longest cutoff). + let keep: Vec<&str> = listed.iter().map(|v| v.id.as_str()).filter(|id| stored.contains(*id)).collect(); + download_missing_audio(state, &keep).await; - debug!("Processing download for video: {}", video_id); - info!("Title: {}", title); - info!("Target path: {}", output_path); + Ok(new_episodes) +} - // Check if file already exists - if tokio::fs::metadata(&output_path).await.is_ok() || - tokio::fs::metadata(&output_path_double).await.is_ok() { - debug!("Audio file already exists, skipping download"); - continue; - } +// Fill in a podcast's name/artwork if they were empty from a broken initial subscription. +async fn heal_channel_metadata(state: &AppState, podcast_id: i32, video: &VideoDetails) { + if video.channel.is_empty() && video.thumbnail.is_empty() { + return; + } + let result = match &state.db_pool { + crate::database::DatabasePool::Postgres(pool) => { + sqlx::query(r#" + UPDATE "Podcasts" + SET podcastname = CASE WHEN podcastname = '' OR podcastname IS NULL THEN $2 ELSE podcastname END, + artworkurl = CASE WHEN artworkurl = '' OR artworkurl IS NULL THEN $3 ELSE artworkurl END, + author = CASE WHEN author = '' OR author IS NULL THEN $2 ELSE author END + WHERE podcastid = $1 + "#) + .bind(podcast_id) + .bind(&video.channel) + .bind(&video.thumbnail) + .execute(pool) + .await + .map(|r| r.rows_affected()) + } + crate::database::DatabasePool::MySQL(pool) => { + sqlx::query(r#" + UPDATE Podcasts + SET PodcastName = CASE WHEN PodcastName = '' OR PodcastName IS NULL THEN ? ELSE PodcastName END, + ArtworkURL = CASE WHEN ArtworkURL = '' OR ArtworkURL IS NULL THEN ? ELSE ArtworkURL END, + Author = CASE WHEN Author = '' OR Author IS NULL THEN ? ELSE Author END + WHERE PodcastID = ? + "#) + .bind(&video.channel) + .bind(&video.thumbnail) + .bind(&video.channel) + .bind(podcast_id) + .execute(pool) + .await + .map(|r| r.rows_affected()) + } + }; + if let Err(e) = result { + warn!("Updating metadata of YouTube podcast {} failed: {}", podcast_id, e); + } +} - info!("Starting download..."); - match download_youtube_audio(video_id, &output_path).await { - Ok(_) => { - info!("Download completed successfully"); - successful_downloads += 1; - - // Get duration from the downloaded MP3 file and update database - if let Some(duration) = get_mp3_duration(&output_path) { +// Download the audio of these videos where it isn't on disk yet. +async fn download_missing_audio(state: &AppState, video_ids: &[&str]) { + let (mut downloaded, mut failed) = (0, 0); + for video_id in video_ids { + let output_path = format!("/opt/pinepods/downloads/youtube/{}.mp3", video_id); + let output_path_double = format!("{}.mp3", output_path); + if tokio::fs::metadata(&output_path).await.is_ok() || tokio::fs::metadata(&output_path_double).await.is_ok() { + continue; + } + debug!("Downloading audio of video {} to {}", video_id, output_path); + let result = { + let _permit = crate::services::ytdlp::refresh_permit().await; + download_youtube_audio(video_id, &output_path).await + }; + match result { + Ok(_) => { + downloaded += 1; + match get_mp3_duration(&output_path) { + Some(duration) => { if let Err(e) = state.db_pool.update_youtube_video_duration(video_id, duration).await { warn!("Failed to update duration for video {}: {}", video_id, e); - } else { - debug!("Updated duration for video {} to {} seconds", video_id, duration); } - } else { - warn!("Could not read duration from MP3 file: {}", output_path); } + None => debug!("Could not read duration from MP3 file: {}", output_path), } - Err(e) => { - failed_downloads += 1; - let error_msg = e.to_string(); - if error_msg.to_lowercase().contains("members-only") { - debug!("Skipping video {} - Members-only content: {}", video_id, title); - } else if error_msg.to_lowercase().contains("private") { - debug!("Skipping video {} - Private video: {}", video_id, title); - } else if error_msg.to_lowercase().contains("unavailable") { - debug!("Skipping video {} - Unavailable video: {}", video_id, title); - } else { - warn!("Failed to download video {}: {}", video_id, title); - warn!("Error: {}", error_msg); - } + } + Err(e) => { + failed += 1; + let error_msg = e.to_string().to_lowercase(); + if ["members-only", "private", "unavailable"].iter().any(|w| error_msg.contains(w)) { + debug!("Skipping video {}: {}", video_id, e); + } else { + warn!("Failed to download video {}: {}", video_id, e); } } } - - warn!("Download summary: {} successful, {} failed", successful_downloads, failed_downloads); - } else { - info!("No new videos to process"); } - - // Update episode count - state.db_pool.update_episode_count(podcast_id).await?; - - info!("{}", "=".repeat(50)); - debug!("Channel processing complete"); - info!("{}", "=".repeat(50)); - - Ok(()) + if downloaded + failed > 0 { + debug!("YouTube audio downloads: {} successful, {} failed", downloaded, failed); + } } @@ -774,4 +775,38 @@ pub async fn check_youtube_channel( ).await?; Ok(Json(serde_json::json!({ "exists": exists }))) -} \ No newline at end of file +} +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn channel_ids_from_feed_urls() { + assert_eq!(channel_id_from_url("https://www.youtube.com/channel/UCabc"), "UCabc"); + assert_eq!(channel_id_from_url("https://www.youtube.com/channel/UCabc/videos?x=1"), "UCabc"); + assert_eq!(channel_id_from_url("UCabc"), "UCabc"); + } + + #[test] + fn view_counts_dont_change_the_feed_hash() { + let feed = |views: u32, title: &str| { + format!("{title}") + }; + assert_eq!(channel_feed_hash(&feed(1, "A")), channel_feed_hash(&feed(999, "A"))); + assert_ne!(channel_feed_hash(&feed(1, "A")), channel_feed_hash(&feed(1, "B"))); + } + + #[test] + fn video_ids_from_a_channel_feed() { + let body = r#" + +yt:channel:UCabcChan +yt:video:vid1vid1One2026-10-20T10:00:00+00:00 +"#; + let videos = parse_channel_feed(body).unwrap(); + assert_eq!(videos.len(), 1); + assert_eq!(videos[0].id, "vid1"); + assert_eq!(videos[0].published.unwrap().to_rfc3339(), "2026-10-20T10:00:00+00:00"); + assert!(parse_channel_feed("nope").is_err()); + } +} diff --git a/rust-api/src/services/ytdlp.rs b/rust-api/src/services/ytdlp.rs index 7bab494a2..8e1858cbd 100644 --- a/rust-api/src/services/ytdlp.rs +++ b/rust-api/src/services/ytdlp.rs @@ -23,6 +23,18 @@ pub fn ytdlp_binary() -> String { std::env::var("YTDLP_PATH").unwrap_or_else(|_| "yt-dlp".to_string()) } +/// yt-dlp processes the feed refresh may run at once. Each one is heavy and talks to YouTube, so +/// this is far below the refresh's feed concurrency. +const REFRESH_PROCESSES: usize = 3; + +static REFRESH_PERMITS: std::sync::LazyLock = + std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(REFRESH_PROCESSES)); + +/// Wait for a turn to run yt-dlp from the refresh. +pub async fn refresh_permit() -> tokio::sync::SemaphorePermit<'static> { + REFRESH_PERMITS.acquire().await.expect("yt-dlp permits are never closed") +} + /// Normalize an arbitrary channel string to one yt-dlp accepts. Anything other than `nightly` /// falls back to `stable` so the UI can't persist a value that bricks updates. pub fn normalize_channel(channel: &str) -> &'static str { diff --git a/rust-api/tests/integration/harness/app.rs b/rust-api/tests/integration/harness/app.rs index 6cdc237fd..73138aefd 100644 --- a/rust-api/tests/integration/harness/app.rs +++ b/rust-api/tests/integration/harness/app.rs @@ -128,6 +128,7 @@ impl TestApp { api: ApiConfig { search_api_url: fixtures.url("search-api/api/search"), people_api_url: fixtures.url("people-api"), + youtube_feed_url: fixtures.url("youtube/{channel_id}.xml"), }, }; tweak(&mut config); diff --git a/rust-api/tests/integration/harness/env.rs b/rust-api/tests/integration/harness/env.rs index df62d3d99..cdeb7b5a1 100644 --- a/rust-api/tests/integration/harness/env.rs +++ b/rust-api/tests/integration/harness/env.rs @@ -42,6 +42,8 @@ pub struct TestEnv { /// Name of the fully-migrated database every test clones. pub template: String, pub fixtures: FixtureServer, + /// Where the yt-dlp stub and its answers live (see `youtube.rs`). + pub ytdlp_dir: PathBuf, } static ENV: OnceLock = OnceLock::new(); @@ -64,6 +66,8 @@ pub fn env() -> &'static TestEnv { fn init() -> TestEnv { sandbox_network(); pinepods_api::services::url_guard::allow_loopback_for_tests(); + let ytdlp_dir = std::env::temp_dir().join(format!("pinepods-ytdlp-{}", std::process::id())); + super::youtube::install_stub(&ytdlp_dir); let backend = match var("PINEPODS_TEST_DB_TYPE", "postgresql").to_lowercase().as_str() { "postgresql" | "postgres" => Backend::Postgres, @@ -85,6 +89,7 @@ fn init() -> TestEnv { valkey_port: var("PINEPODS_TEST_VALKEY_PORT", "56380").parse().expect("PINEPODS_TEST_VALKEY_PORT"), template: String::new(), fixtures: FixtureServer::start(), + ytdlp_dir, }; let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); diff --git a/rust-api/tests/integration/harness/mod.rs b/rust-api/tests/integration/harness/mod.rs index 504714a8f..618e95bc9 100644 --- a/rust-api/tests/integration/harness/mod.rs +++ b/rust-api/tests/integration/harness/mod.rs @@ -11,6 +11,7 @@ pub mod http; pub mod mock_gpodder; pub mod rss; pub mod sql; +pub mod youtube; #[allow(unused_imports)] pub use app::{hash_password, TestApp, TestUser}; diff --git a/rust-api/tests/integration/harness/youtube.rs b/rust-api/tests/integration/harness/youtube.rs new file mode 100644 index 000000000..3e77565bb --- /dev/null +++ b/rust-api/tests/integration/harness/youtube.rs @@ -0,0 +1,215 @@ +//! YouTube without YouTube: channel feeds (videos.xml) come from the fixture server and yt-dlp is +//! a stub script that answers from files and logs every call. +//! +//! The stub is installed once per test process (`YTDLP_PATH`). For each `watch?v=` argument +//! it prints `videos/.json`, for a `/channel/` argument `channels/.jsonl`, and fails +//! like yt-dlp for anything it has no file for. Tests isolate themselves by id (channel and video +//! ids embed the test's tag). + +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Duration, Utc}; +use serde_json::json; + +use super::{eventually, int, TestApp, TestUser}; +use super::fixtures::Fixture; + +const STUB: &str = r#"#!/bin/sh +dir="$(dirname "$0")" +echo "$*" >> "$dir/calls.log" +status=0 +for a in "$@"; do + case "$a" in + *watch\?v=*) + id="${a##*v=}" + if [ -f "$dir/videos/$id.json" ]; then cat "$dir/videos/$id.json"; echo; else echo "ERROR: [youtube] $id: Video unavailable" >&2; status=1; fi ;; + */channel/*) + ch="${a#*/channel/}"; ch="${ch%%/*}" + if [ -f "$dir/channels/$ch.jsonl" ]; then cat "$dir/channels/$ch.jsonl"; else echo "ERROR: [youtube:tab] $ch: channel unavailable" >&2; status=1; fi ;; + esac +done +exit $status +"#; + +/// Write the stub into `dir` and point the server at it. Call before any server code runs. +pub fn install_stub(dir: &Path) { + use std::os::unix::fs::PermissionsExt; + std::fs::create_dir_all(dir.join("videos")).unwrap(); + std::fs::create_dir_all(dir.join("channels")).unwrap(); + let bin = dir.join("yt-dlp"); + std::fs::write(&bin, STUB).unwrap(); + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); + // SAFETY (edition 2021 API): runs once, before the server code under test starts threads. + std::env::set_var("YTDLP_PATH", &bin); +} + +#[derive(Clone)] +pub struct Video { + pub id: String, + pub title: String, + pub published: DateTime, + pub duration: u32, +} + +#[derive(Clone)] +pub struct Channel { + pub id: String, + pub name: String, + /// Newest first, like the real feed. + pub videos: Vec