//! Tauri command surface. Delegates to the modules; holds no parsing logic. use crate::db::Db; use crate::downloader::{self, Progress}; use crate::feed; use crate::models::{ Channel, ChannelWithCount, DownloadState, FeedFilter, FeedItem, ImportPreview, Prereqs, }; use crate::net; use crate::playlist_server::PlaylistServer; use crate::takeout; use crate::thumbs; use futures::stream::StreamExt; use serde::Serialize; use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; use tauri::{AppHandle, Emitter, Manager, State}; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::sync::{Mutex, Semaphore}; /// How many channel feeds we fetch at once. YouTube serves these fine in /// parallel; 8 keeps a 300-channel refresh brisk without hammering them. const FEED_CONCURRENCY: usize = 8; /// Video downloads are bandwidth-bound, so more than a couple at once just /// makes every one of them slower. const DOWNLOAD_CONCURRENCY: usize = 2; /// Cap thumbnail work per refresh so a first-run import doesn't stall for ages. const THUMB_BATCH: i64 = 600; pub struct AppState { pub db: Arc>, pub http: reqwest::Client, pub library: Arc>, pub app_data: PathBuf, pub children: Arc>>, pub download_slots: Arc, pub playlists: PlaylistServer, /// yt-dlp's PyInstaller bundle costs several seconds to start, so its /// version is read once and kept. pub prereqs: Arc>>, /// Resolved stream URLs, keyed by video and quality cap. YouTube's signed /// URLs last hours, so replaying a video should not pay for yt-dlp again. pub streams: Arc), (String, std::time::Instant)>>>, } /// Comfortably inside the ~6h lifetime of YouTube's signed URLs. const STREAM_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(3 * 3600); #[derive(Clone, Serialize)] struct RefreshProgress { done: usize, total: usize, channel: String, } #[derive(Clone, Serialize)] struct DownloadProgressEvent { video_id: String, pct: Option, bytes_done: u64, bytes_total: Option, speed: Option, eta: Option, } #[derive(Clone, Serialize)] struct DownloadStateEvent { video_id: String, state: DownloadState, error: Option, path: Option, } #[derive(Serialize)] pub struct RefreshSummary { pub channels: usize, pub new_videos: usize, pub failures: Vec, } /// Resolves a helper binary. /// /// The app ships `yt-dlp`, `ffmpeg` and `ffprobe` as sidecars, so a fresh Mac /// needs nothing installed. Tauri places them beside the executable inside /// `Contents/MacOS/`, which is checked first. A copy on PATH still wins nothing /// — but the Homebrew fallbacks remain for `cargo run` during development, /// where there is no bundle. fn bin(name: &str) -> String { if let Ok(exe) = std::env::current_exe() { if let Some(dir) = exe.parent() { let bundled = dir.join(name); if bundled.exists() { return bundled.to_string_lossy().to_string(); } } } // GUI apps launched from Finder don't inherit a login shell PATH, so // Homebrew's bin dir is invisible to them; name the paths explicitly. for prefix in ["/opt/homebrew/bin", "/usr/local/bin", "/usr/bin"] { let candidate = format!("{prefix}/{name}"); if std::path::Path::new(&candidate).exists() { return candidate; } } name.to_string() } /// True when the binary we resolved is the one inside the app bundle. fn is_bundled(name: &str) -> bool { std::env::current_exe() .ok() .and_then(|e| e.parent().map(|d| d.join(name).exists())) .unwrap_or(false) } /// `flag` differs per tool: yt-dlp takes `--version`, ffmpeg only accepts /// `-version` (it exits non-zero on `--version` and writes to stderr), so the /// flag is passed in and both streams are consulted. async fn version_of(name: &str, flag: &str) -> Option { // Must be the async Command: yt-dlp is a PyInstaller bundle that unpacks // ~37MB on its first run, so a blocking call here would stall a runtime // worker for the better part of a minute. let out = tokio::process::Command::new(bin(name)) .arg(flag) .output() .await .ok()?; if !out.status.success() { return None; } let text = if out.stdout.is_empty() { String::from_utf8_lossy(&out.stderr).to_string() } else { String::from_utf8_lossy(&out.stdout).to_string() }; let first = text.lines().next().unwrap_or("").trim().to_string(); (!first.is_empty()).then_some(first) } fn label(version: &str, name: &str) -> String { if is_bundled(name) { format!("{version} (bundled)") } else { format!("{version} (system)") } } #[tauri::command] pub async fn check_prereqs(state: State<'_, AppState>) -> Result { // The library path can change, so only the tool versions are cached. let library = state.library.lock().await.clone(); if let Some(cached) = state.prereqs.lock().await.clone() { return Ok(Prereqs { library_path: library.to_string_lossy().to_string(), ..cached }); } let fresh = Prereqs { yt_dlp: version_of("yt-dlp", "--version").await.map(|v| label(&v, "yt-dlp")), ffmpeg: version_of("ffmpeg", "-version").await.map(|v| { // ffmpeg's first line is long; keep the useful head of it. let head = v.split_whitespace().take(3).collect::>().join(" "); label(&head, "ffmpeg") }), library_path: library.to_string_lossy().to_string(), }; *state.prereqs.lock().await = Some(fresh.clone()); Ok(fresh) } async fn read_channels(path: &str) -> Result, String> { let raw = tokio::fs::read(path) .await .map_err(|e| format!("Cannot read {path}: {e}"))?; // Takeout exports are UTF-8, sometimes with a BOM. let text = String::from_utf8_lossy(&raw); let text = text.strip_prefix('\u{feff}').unwrap_or(&text); let channels = takeout::parse_csv(text)?; if channels.is_empty() { return Err("No channels found in that file.".into()); } Ok(channels) } /// Reports what a replacing import would add and destroy, so the UI can name /// the consequences before the user commits to them. #[tauri::command] pub async fn preview_takeout_import( path: String, state: State<'_, AppState>, ) -> Result { let channels = read_channels(&path).await?; state.db.lock().await.preview_replace(&channels) } /// The imported CSV becomes the entire subscription list. Channels that are no /// longer in it are removed along with their videos, download records, and the /// downloaded files themselves — leaving those on disk would orphan gigabytes /// the app can no longer show or delete. #[tauri::command] pub async fn import_takeout_csv( path: String, state: State<'_, AppState>, ) -> Result { let channels = read_channels(&path).await?; let doomed = state .db .lock() .await .paths_dropped_by_replace(&channels)?; for p in doomed { let _ = tokio::fs::remove_file(&p).await; } let mut db = state.db.lock().await; db.replace_channels(&channels) } /// Resolves a directly playable URL for a video we have NOT downloaded, so it /// can stream inside the app's own player. /// /// YouTube's iframe embed is not an option here: it rejects a Tauri window with /// "Error 153" because the page origin is `tauri://localhost` rather than an /// http(s) origin it will accept. Instead we ask yt-dlp for YouTube's HLS master /// playlist, which lists H.264 + AAC variants up to 1080p with separate audio /// tracks — exactly the shape AVFoundation plays natively in WKWebView, with /// adaptive bitrate for free. #[derive(Serialize)] pub struct Stream { /// Direct URL to hand the player, when no filtering was needed. pub url: Option, /// A rewritten HLS master playlist, when a quality cap was applied. The /// frontend turns this into a Blob URL — every URL inside is absolute, so /// the playlist works from anywhere. pub playlist: Option, } #[tauri::command] pub async fn resolve_stream( video_id: String, max_height: Option, state: State<'_, AppState>, ) -> Result { let key = (video_id.clone(), max_height); { let mut cache = state.streams.lock().await; cache.retain(|_, (_, at)| at.elapsed() < STREAM_CACHE_TTL); if let Some((cached, _)) = cache.get(&key) { return Ok(Stream { url: Some(cached.clone()), playlist: None }); } } let url = format!("https://www.youtube.com/watch?v={video_id}"); // The HLS master playlist. Every m3u8 format shares the same manifest_url, // so any one of them yields the master. if let Some(master) = yt_dlp_print( &["-f", "bv*[protocol^=m3u8]", "--print", "%(manifest_url)s", &url], ) .await { let Some(cap) = max_height else { remember(&state, key, &master).await; return Ok(Stream { url: Some(master), playlist: None }); }; // Fetch and cut it down. If anything about that fails, the adaptive // master still plays — a quality preference is not worth an error. return match state.http.get(&master).send().await { Ok(resp) => match resp.text().await { Ok(body) => Ok(match filter_master_playlist(&body, cap) { Some(filtered) => { let served = state.playlists.publish(filtered).await; remember(&state, key, &served).await; Stream { url: Some(served), playlist: None } } None => Stream { url: Some(master), playlist: None }, }), Err(_) => Ok(Stream { url: Some(master), playlist: None }), }, Err(_) => Ok(Stream { url: Some(master), playlist: None }), }; } // Rare fallback: an old-style progressive muxed MP4. if let Some(u) = yt_dlp_print(&[ "-f", "b[ext=mp4][acodec!=none][vcodec!=none]", "--print", "%(url)s", &url, ]) .await { return Ok(Stream { url: Some(u), playlist: None }); } Err("Could not find a playable stream for this video.".into()) } async fn remember( state: &State<'_, AppState>, key: (String, Option), url: &str, ) { state .streams .lock() .await .insert(key, (url.to_string(), std::time::Instant::now())); } /// Keeps only the highest video variant at or below `max_height`, along with /// every `EXT-X-MEDIA` line (the audio and subtitle groups it references). /// /// Returns `None` if nothing matched, so the caller can fall back to adaptive /// rather than hand the player an empty playlist. pub fn filter_master_playlist(body: &str, max_height: u32) -> Option { let lines: Vec<&str> = body.lines().collect(); let mut media = Vec::new(); // (height, stream-inf line, url line) let mut variants: Vec<(u32, &str, &str)> = Vec::new(); for (i, line) in lines.iter().enumerate() { if line.starts_with("#EXT-X-MEDIA:") { media.push(*line); } else if line.starts_with("#EXT-X-STREAM-INF:") { let Some(url) = lines.get(i + 1) else { continue }; if url.starts_with('#') || url.trim().is_empty() { continue; } if let Some(h) = resolution_height(line) { if h <= max_height { variants.push((h, *line, *url)); } } } } let best = variants.iter().max_by_key(|(h, _, _)| *h)?; let mut out = String::from("#EXTM3U #EXT-X-INDEPENDENT-SEGMENTS "); for m in media { out.push_str(m); out.push('\n'); } out.push_str(best.1); out.push('\n'); out.push_str(best.2); out.push('\n'); Some(out) } /// Pulls the vertical size out of a `RESOLUTION=1920x1080` attribute. fn resolution_height(stream_inf: &str) -> Option { let at = stream_inf.find("RESOLUTION=")? + "RESOLUTION=".len(); let rest = &stream_inf[at..]; let value = rest.split(&[',', ' '][..]).next()?; value.split(&['x', 'X'][..]).nth(1)?.parse().ok() } /// Runs yt-dlp and returns its first non-empty stdout line, or None. async fn yt_dlp_print(args: &[&str]) -> Option { let mut cmd = tokio::process::Command::new(bin("yt-dlp")); cmd.args(["--no-warnings", "--no-playlist", "--simulate"]); cmd.args(args); let out = cmd.output().await.ok()?; if !out.status.success() { return None; } String::from_utf8_lossy(&out.stdout) .lines() .map(str::trim) .find(|l| l.starts_with("http")) .map(str::to_string) } /// Called periodically while a video plays, and once when the player closes. #[tauri::command] pub async fn save_playback( video_id: String, position: f64, duration: f64, state: State<'_, AppState>, ) -> Result<(), String> { state.db.lock().await.save_playback(&video_id, position, duration) } #[tauri::command] pub async fn list_channels(state: State<'_, AppState>) -> Result, String> { state.db.lock().await.list_channels() } #[tauri::command] pub async fn list_feed( filter: FeedFilter, state: State<'_, AppState>, ) -> Result, String> { state.db.lock().await.list_feed(&filter) } #[tauri::command] pub async fn get_connectivity(state: State<'_, AppState>) -> Result { Ok(net::is_online(&state.http).await) } #[tauri::command] pub async fn refresh_feeds( app: AppHandle, state: State<'_, AppState>, ) -> Result { let channel_ids = state.db.lock().await.channel_ids()?; if channel_ids.is_empty() { return Err("No subscriptions imported yet. Import your Takeout CSV first.".into()); } let total = channel_ids.len(); let http = state.http.clone(); // Fetch with bounded concurrency. One channel failing must not abort the run. let results = futures::stream::iter(channel_ids.into_iter().map(|cid| { let http = http.clone(); async move { let res = feed::fetch_channel(&http, &cid).await; (cid, res) } })) .buffer_unordered(FEED_CONCURRENCY) .collect::>() .await; let mut new_videos = 0usize; let mut failures = Vec::new(); let mut done = 0usize; for (cid, res) in results { done += 1; match res { Ok(videos) => { if !videos.is_empty() { let mut db = state.db.lock().await; new_videos += db.upsert_videos(&videos)?; } } Err(e) => failures.push(format!("{cid}: {e}")), } let _ = app.emit( "refresh:progress", RefreshProgress { done, total, channel: cid, }, ); } cache_thumbnails(&state).await; Ok(RefreshSummary { channels: total, new_videos, failures, }) } /// Mirrors thumbnails to disk so the feed still renders with no network. async fn cache_thumbnails(state: &State<'_, AppState>) { let pending = match state.db.lock().await.videos_missing_thumbs(THUMB_BATCH) { Ok(p) => p, Err(_) => return, }; if pending.is_empty() { return; } let dir = thumbs::cache_dir(&state.app_data); let http = state.http.clone(); let cached = futures::stream::iter(pending.into_iter().map(|(id, url)| { let http = http.clone(); let dir = dir.clone(); async move { let res = thumbs::cache_one(&http, &id, &url, &dir).await; (id, res) } })) .buffer_unordered(FEED_CONCURRENCY) .collect::>() .await; let db = state.db.lock().await; for (id, res) in cached { if let Ok(path) = res { let _ = db.set_thumb_path(&id, &path.to_string_lossy()); } } } #[tauri::command] pub async fn download_video( video_id: String, quality: String, app: AppHandle, state: State<'_, AppState>, ) -> Result<(), String> { let library = state.library.lock().await.clone(); tokio::fs::create_dir_all(&library) .await .map_err(|e| format!("Cannot create library dir: {e}"))?; { let db = state.db.lock().await; db.set_download_state(&video_id, DownloadState::Queued, None)?; } let _ = app.emit( "download:state", DownloadStateEvent { video_id: video_id.clone(), state: DownloadState::Queued, error: None, path: None, }, ); let permit = state .download_slots .clone() .acquire_owned() .await .map_err(|e| format!("Download queue closed: {e}"))?; let out_template = library .join(downloader::OUTPUT_TEMPLATE) .to_string_lossy() .to_string(); let mut args = downloader::build_args(&video_id, &out_template, &quality); // Without this yt-dlp looks for ffmpeg on PATH, which a bundled app has no // reason to have. Merging video and audio would fail on a clean machine. args.push("--ffmpeg-location".into()); args.push(bin("ffmpeg")); let mut child = tokio::process::Command::new(bin("yt-dlp")) .args(&args) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) .spawn() .map_err(|e| { format!("Could not start yt-dlp: {e}. Install it with: brew install yt-dlp ffmpeg") })?; let stdout = child.stdout.take().ok_or("yt-dlp produced no stdout")?; let stderr = child.stderr.take().ok_or("yt-dlp produced no stderr")?; state .children .lock() .await .insert(video_id.clone(), child); { let db = state.db.lock().await; db.set_download_state(&video_id, DownloadState::Running, None)?; } let _ = app.emit( "download:state", DownloadStateEvent { video_id: video_id.clone(), state: DownloadState::Running, error: None, path: None, }, ); // Collect stderr concurrently so a failure has a real message, and so a // full stderr pipe can't deadlock the child. let stderr_task = tokio::spawn(async move { let mut lines = BufReader::new(stderr).lines(); let mut buf = Vec::new(); while let Ok(Some(l)) = lines.next_line().await { buf.push(l); } buf }); let mut final_path: Option = None; let mut reader = BufReader::new(stdout).lines(); while let Ok(Some(line)) = reader.next_line().await { if let Some(p) = downloader::parse_final_path(&line) { final_path = Some(p); continue; } if let Some(Progress { downloaded, total, speed, eta, }) = downloader::parse_progress_line(&line) { let pct = Progress { downloaded, total, speed, eta, } .pct(); { let db = state.db.lock().await; let _ = db.set_download_progress(&video_id, downloaded, total, pct); } let _ = app.emit( "download:progress", DownloadProgressEvent { video_id: video_id.clone(), pct, bytes_done: downloaded, bytes_total: total, speed, eta, }, ); } } let mut child = state .children .lock() .await .remove(&video_id) .ok_or_else(|| "Download was cancelled.".to_string())?; let status = child .wait() .await .map_err(|e| format!("yt-dlp did not exit cleanly: {e}"))?; let stderr_lines = stderr_task.await.unwrap_or_default(); drop(permit); if status.success() { // yt-dlp normally reports the path via `--print after_move:`; if that // line went missing, find the file it wrote by its embedded video id. let path = match final_path { Some(p) => p, None => find_by_video_id(&library, &video_id) .await .ok_or("Download finished but the file could not be located.")?, }; let db = state.db.lock().await; db.set_download_state(&video_id, DownloadState::Done, None)?; db.set_download_path(&video_id, &path)?; drop(db); let _ = app.emit( "download:state", DownloadStateEvent { video_id, state: DownloadState::Done, error: None, path: Some(path), }, ); Ok(()) } else { let msg = stderr_lines .iter() .rev() .find(|l| l.contains("ERROR")) .cloned() .unwrap_or_else(|| format!("yt-dlp exited with {status}")); let db = state.db.lock().await; db.set_download_state(&video_id, DownloadState::Failed, Some(&msg))?; drop(db); let _ = app.emit( "download:state", DownloadStateEvent { video_id, state: DownloadState::Failed, error: Some(msg.clone()), path: None, }, ); Err(msg) } } #[tauri::command] pub async fn cancel_download( video_id: String, app: AppHandle, state: State<'_, AppState>, ) -> Result<(), String> { if let Some(mut child) = state.children.lock().await.remove(&video_id) { let _ = child.kill().await; } { let db = state.db.lock().await; db.set_download_state(&video_id, DownloadState::Cancelled, None)?; } cleanup_partials(state.library.lock().await.clone(), &video_id).await; let _ = app.emit( "download:state", DownloadStateEvent { video_id, state: DownloadState::Cancelled, error: None, path: None, }, ); Ok(()) } /// Locates a finished download by the `[]` tag in its filename. async fn find_by_video_id(library: &std::path::Path, video_id: &str) -> Option { let mut entries = tokio::fs::read_dir(library).await.ok()?; while let Ok(Some(entry)) = entries.next_entry().await { let name = entry.file_name().to_string_lossy().to_string(); if name.contains(video_id) && !name.contains(".part") && !name.ends_with(".ytdl") { return Some(entry.path().to_string_lossy().to_string()); } } None } /// yt-dlp leaves `.part`, `.ytdl` and format-specific fragments behind when /// killed; without this the library slowly fills with dead bytes. async fn cleanup_partials(library: PathBuf, video_id: &str) { let Ok(mut entries) = tokio::fs::read_dir(&library).await else { return; }; while let Ok(Some(entry)) = entries.next_entry().await { let name = entry.file_name().to_string_lossy().to_string(); if name.contains(video_id) && (name.contains(".part") || name.ends_with(".ytdl")) { let _ = tokio::fs::remove_file(entry.path()).await; } } } #[tauri::command] pub async fn delete_download( video_id: String, state: State<'_, AppState>, ) -> Result<(), String> { let path = state.db.lock().await.get_download_path(&video_id)?; if let Some(p) = path { let _ = tokio::fs::remove_file(&p).await; } cleanup_partials(state.library.lock().await.clone(), &video_id).await; state.db.lock().await.clear_download(&video_id) } #[tauri::command] pub async fn set_library_path( path: String, state: State<'_, AppState>, ) -> Result { let p = PathBuf::from(&path); tokio::fs::create_dir_all(&p) .await .map_err(|e| format!("Cannot use that folder: {e}"))?; *state.library.lock().await = p.clone(); Ok(p.to_string_lossy().to_string()) } #[tauri::command] pub async fn open_external(app: AppHandle, url: String) -> Result<(), String> { use tauri_plugin_opener::OpenerExt; app.opener() .open_url(url, None::<&str>) .map_err(|e| format!("Could not open link: {e}")) } pub fn default_library() -> PathBuf { dirs_home() .map(|h| h.join("Movies").join("FlightTube")) .unwrap_or_else(|| PathBuf::from("FlightTube")) } fn dirs_home() -> Option { std::env::var_os("HOME").map(PathBuf::from) } pub fn build_state(app: &AppHandle) -> Result { let app_data = app .path() .app_data_dir() .map_err(|e| format!("No app data dir: {e}"))?; std::fs::create_dir_all(&app_data).map_err(|e| format!("Cannot create app data dir: {e}"))?; let db = Db::open(&app_data.join("flighttube.db"))?; let library = default_library(); let _ = std::fs::create_dir_all(&library); let http = reqwest::Client::builder() .user_agent("FlightTube/0.1 (+desktop)") .timeout(std::time::Duration::from_secs(20)) .build() .map_err(|e| format!("Cannot build HTTP client: {e}"))?; Ok(AppState { db: Arc::new(Mutex::new(db)), http, library: Arc::new(Mutex::new(library)), app_data, children: Arc::new(Mutex::new(HashMap::new())), download_slots: Arc::new(Semaphore::new(DOWNLOAD_CONCURRENCY)), playlists: PlaylistServer::start()?, prereqs: Arc::new(Mutex::new(None)), streams: Arc::new(Mutex::new(HashMap::new())), }) } #[cfg(test)] mod tests { use super::filter_master_playlist; const MASTER: &str = concat!( "#EXTM3U\n", "#EXT-X-INDEPENDENT-SEGMENTS\n", "#EXT-X-MEDIA:URI=\"https://a/audio.m3u8\",TYPE=AUDIO,GROUP-ID=\"234\",DEFAULT=YES\n", "#EXT-X-STREAM-INF:BANDWIDTH=756324,CODECS=\"avc1,mp4a\",RESOLUTION=640x360,AUDIO=\"234\"\n", "https://a/360.m3u8\n", "#EXT-X-STREAM-INF:BANDWIDTH=3878958,CODECS=\"avc1,mp4a\",RESOLUTION=1280x720,AUDIO=\"234\"\n", "https://a/720.m3u8\n", "#EXT-X-STREAM-INF:BANDWIDTH=6039686,CODECS=\"avc1,mp4a\",RESOLUTION=1920x1080,AUDIO=\"234\"\n", "https://a/1080.m3u8\n", ); #[test] fn keeps_the_best_variant_at_or_below_the_cap() { let out = filter_master_playlist(MASTER, 720).unwrap(); assert!(out.contains("https://a/720.m3u8")); assert!(!out.contains("https://a/1080.m3u8")); assert!(!out.contains("https://a/360.m3u8")); } #[test] fn always_carries_the_audio_group_across() { // A video-only variant would play silently, so EXT-X-MEDIA must survive. let out = filter_master_playlist(MASTER, 360).unwrap(); assert!(out.contains("TYPE=AUDIO")); assert!(out.contains("https://a/audio.m3u8")); assert!(out.starts_with("#EXTM3U")); } #[test] fn an_exact_match_is_included_not_excluded() { let out = filter_master_playlist(MASTER, 1080).unwrap(); assert!(out.contains("https://a/1080.m3u8")); } #[test] fn nothing_below_the_cap_yields_none_so_the_caller_can_fall_back() { assert!(filter_master_playlist(MASTER, 144).is_none()); assert!(filter_master_playlist("#EXTM3U\n", 1080).is_none()); assert!(filter_master_playlist("", 1080).is_none()); } #[test] fn a_stream_inf_with_no_following_url_is_skipped() { let truncated = "#EXTM3U\n#EXT-X-STREAM-INF:RESOLUTION=1280x720\n"; assert!(filter_master_playlist(truncated, 1080).is_none()); } }