use std::{ collections::{VecDeque, HashSet}, fs, path::{Path, PathBuf}, process::{Child, Command, Stdio}, sync::{mpsc, Arc, Mutex}, thread, time::{Duration, Instant}, net::TcpStream, io::{Read, Write}, }; use serde_json::{json, Value}; use crate::core::{config::Config, jellyfin::{JellyfinClient, JfItem, LibraryQuery}, paths}; #[derive(Clone, Debug)] pub enum DownloadKind { Movie, Episode { series: String, season: Option, episode: Option }, Track { artist: Option, album: Option, track_no: Option }, Other, } #[derive(Clone, Debug)] pub enum DownloadStatus { Queued, Downloading, Finalizing, Completed, Failed(String), Cancelled, } #[derive(Clone, Debug)] pub struct DownloadItem { pub item_id: String, pub title: String, pub kind: DownloadKind, pub final_rel_path: PathBuf, // relative to downloads root pub status: DownloadStatus, pub added_at: Instant, pub bytes_downloaded: u64, pub bytes_total: Option, pub download_speed: u64, // bytes/sec pub eta_secs: Option, // seconds remaining // aria2 pub gid: Option, pub playlist_links: Vec, } #[derive(Debug)] pub enum DownloadCommand { Enqueue(Vec), EnqueuePlaylistItem { item: JfItem, playlist_name: String, playlist_index: usize, }, Cancel { index: usize }, RemoveQueued { index: usize }, Shutdown, } #[derive(Default)] pub struct DownloadState { pub queue: VecDeque, pub active_gid: Option, pub last_message: Option, } pub struct DownloadManager { tx: mpsc::Sender, pub state: Arc>, } #[derive(Clone, Debug)] pub struct LocalEntry { pub item_id: String, pub title: String, pub path: PathBuf, } #[derive(Clone, Debug)] pub struct PlaylistLink { pub playlist_name: String, pub playlist_index: usize, // 0-based pub title: String, } /// Scan downloads root for marker files and return local playable entries. /// Title is taken from marker JSON if present, else from the media filename stem. pub fn scan_local_entries() -> Vec { fn walk(dir: &Path, out: &mut Vec) { let rd = match std::fs::read_dir(dir) { Ok(r) => r, Err(_) => return, }; for ent in rd.flatten() { let p = ent.path(); if p.is_dir() { walk(&p, out); continue; } let name = match p.file_name().and_then(|s| s.to_str()) { Some(s) => s, None => continue, }; if !name.ends_with(DOWNLOAD_MARKER_EXT) { continue; } let raw = match std::fs::read_to_string(&p) { Ok(s) => s, Err(_) => continue, }; let v: serde_json::Value = match serde_json::from_str(raw.trim()) { Ok(v) => v, Err(_) => continue, }; let item_id = match v.get("item_id").and_then(|x| x.as_str()) { Some(s) if !s.is_empty() => s.to_string(), _ => continue, }; // derive media path: "." -> "" let suffix = format!(".{DOWNLOAD_MARKER_EXT}"); if !name.ends_with(&suffix) { continue; } let media_name = &name[..name.len() - suffix.len()]; let media_path = p.with_file_name(media_name); if !media_path.exists() { continue; } // title: prefer json field if you ever add it, else from filename let title = v .get("title") .and_then(|x| x.as_str()) .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) .unwrap_or_else(|| { media_path .file_stem() .and_then(|s| s.to_str()) .unwrap_or(media_name) .to_string() }); out.push(LocalEntry { item_id, title, path: media_path, }); } } let mut out = Vec::new(); if let Ok(root) = downloads_root() { walk(&root, &mut out); } // Stable sort: title out.sort_by(|a, b| a.title.to_lowercase().cmp(&b.title.to_lowercase())); out } impl DownloadManager { pub fn start(cfg: Config) -> Self { let state = Arc::new(Mutex::new(DownloadState::default())); let (tx, rx) = mpsc::channel::(); // cleanup temp on start let _ = cleanup_temp_dir(); let state_bg = Arc::clone(&state); thread::spawn(move || worker_loop(cfg, state_bg, rx)); Self { tx, state } } pub fn send(&self, cmd: DownloadCommand) { let _ = self.tx.send(cmd); } } // ---------- directory helpers ---------- const DOWNLOAD_MARKER_EXT: &str = "cjtui.json"; fn marker_path_for_media(media_path: &Path) -> PathBuf { // Example: // Movie.mkv -> Movie.mkv.cjtui.json let file_name = media_path .file_name() .and_then(|s| s.to_str()) .unwrap_or("download"); media_path .parent() .unwrap_or_else(|| Path::new(".")) .join(format!("{file_name}.{DOWNLOAD_MARKER_EXT}")) } fn write_download_marker(media_path: &Path, item_id: &str) -> Result<(), String> { let marker = marker_path_for_media(media_path); let payload = json!({ "item_id": item_id, "saved_at_unix": std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .ok() .map(|d| d.as_secs()) .unwrap_or(0), "media_file": media_path.file_name().and_then(|s| s.to_str()).unwrap_or(""), }); fs::write(&marker, payload.to_string()) .map_err(|e| format!("write marker failed: {e}"))?; Ok(()) } fn scan_dir_for_markers(dir: &Path, out: &mut HashSet) -> Result<(), String> { if !dir.exists() { return Ok(()); } for ent in fs::read_dir(dir).map_err(|e| format!("read_dir failed: {e}"))? { let p = ent.map_err(|e| format!("dir entry failed: {e}"))?.path(); if p.is_dir() { scan_dir_for_markers(&p, out)?; continue; } // Look for *.cjtui.json let name = p.file_name().and_then(|s| s.to_str()).unwrap_or(""); if !name.ends_with(DOWNLOAD_MARKER_EXT) { continue; } let raw = match fs::read_to_string(&p) { Ok(s) => s, Err(_) => continue, }; let v: Value = match serde_json::from_str(raw.trim()) { Ok(v) => v, Err(_) => continue, }; if let Some(id) = v.get("item_id").and_then(|x| x.as_str()) { if !id.trim().is_empty() { out.insert(id.to_string()); } } } Ok(()) } /// Public: scan the downloads root and return set of downloaded Jellyfin item ids. pub fn scan_downloaded_ids() -> Result, String> { let root = downloads_root()?; let mut out = HashSet::new(); scan_dir_for_markers(&root, &mut out)?; Ok(out) } pub fn downloads_root() -> Result { let dir = paths::app_config_dir().ok_or_else(|| "no config dir".to_string())?; Ok(dir.join("downloads")) } pub fn temp_root() -> Result { let dir = paths::app_config_dir().ok_or_else(|| "no config dir".to_string())?; Ok(dir.join("temp")) } pub fn ensure_dirs() -> Result<(), String> { fs::create_dir_all(downloads_root().map_err(|e| e.to_string())?).map_err(|e| e.to_string())?; fs::create_dir_all(temp_root().map_err(|e| e.to_string())?).map_err(|e| e.to_string())?; Ok(()) } pub fn cleanup_temp_dir() -> Result<(), String> { let tmp = temp_root()?; if tmp.exists() { fs::remove_dir_all(&tmp).map_err(|e| format!("remove temp: {e}"))?; } fs::create_dir_all(&tmp).map_err(|e| format!("create temp: {e}"))?; fs::create_dir_all(tmp.join("jobs")).map_err(|e| format!("create temp/jobs: {e}"))?; Ok(()) } // ---------- queue building (expansion) ---------- pub fn expand_for_enqueue( jf: &JellyfinClient, user_id: &str, item: &JfItem, ) -> Result, String> { let t = item.item_type.as_deref().unwrap_or(""); match t { "Movie" => Ok(vec![make_download_item_movie(item)?]), "Episode" => Ok(vec![make_download_item_episode(item)?]), "Audio" => Ok(vec![make_download_item_track(item)?]), // Containers: "Season" => { // episodes under Season (non-recursive) list_all_items(jf, user_id, LibraryQuery::all().with_parent(item.id.clone()).with_item_types("Episode"), false)? .into_iter() .map(|ep| make_download_item_episode(&ep)) .collect() } "Series" => { // episodes under Series (recursive) let mut q = LibraryQuery::all() .with_parent(item.id.clone()) .with_item_types("Episode"); q.recursive = true; list_all_items(jf, user_id, q, true)? .into_iter() .map(|ep| make_download_item_episode(&ep)) .collect() } "MusicAlbum" => { // tracks under album list_all_items(jf, user_id, LibraryQuery::all().with_parent(item.id.clone()).with_item_types("Audio"), false)? .into_iter() .map(|tr| make_download_item_track(&tr)) .collect() } "Playlist" => { // playlist items via special endpoint list_all_playlist_items(jf, user_id, &item.id)? .into_iter() .map(|it| { let tt = it.item_type.as_deref().unwrap_or(""); match tt { "Audio" => make_download_item_track(&it), "Episode" => make_download_item_episode(&it), "Movie" => make_download_item_movie(&it), _ => make_download_item_other(&it), } }) .collect() } _ => Ok(vec![make_download_item_other(item)?]), } } fn list_all_items( jf: &JellyfinClient, user_id: &str, query: LibraryQuery, _maybe_large: bool, ) -> Result, String> { // user-driven action, so paging here is fine const LIMIT: usize = 200; let mut start = 0usize; let mut out = Vec::new(); loop { let page = jf.list_items(user_id, query.clone(), start, LIMIT)?; out.extend(page.items.into_iter()); if out.len() >= page.total || page.total == 0 { break; } start += LIMIT; } Ok(out) } fn list_all_playlist_items(jf: &JellyfinClient, user_id: &str, playlist_id: &str) -> Result, String> { const LIMIT: usize = 200; let mut start = 0usize; let mut out = Vec::new(); loop { let page = jf.playlist_items(user_id, playlist_id, start, LIMIT)?; out.extend(page.items.into_iter()); if out.len() >= page.total || page.total == 0 { break; } start += LIMIT; } Ok(out) } // ---------- naming ---------- fn playlist_link_abs_path( downloads_root: &Path, playlist_name: &str, playlist_index: usize, title: &str, ext: &str, ) -> PathBuf { let plist = sanitize_component(playlist_name); let t = sanitize_component(title); let idx = format!("{:03}", playlist_index + 1); let file = if ext.is_empty() { format!("{idx} {t}") } else { format!("{idx} {t}.{ext}") }; downloads_root .join("playlists") .join(plist) .join(file) } fn ensure_playlist_link( final_media_path: &Path, playlist_name: &str, playlist_index: usize, title: &str, ) -> Result<(), String> { let dl_root = downloads_root()?; let ext = final_media_path .extension() .and_then(|s| s.to_str()) .unwrap_or(""); let link_abs = playlist_link_abs_path(&dl_root, playlist_name, playlist_index, title, ext); if let Some(parent) = link_abs.parent() { fs::create_dir_all(parent).map_err(|e| format!("mk playlist dir: {e}"))?; } // If exists, do nothing (idempotent) if link_abs.exists() { return Ok(()); } // Prefer hardlink (fast, no duplication). If cross-device, fall back to copy. match fs::hard_link(final_media_path, &link_abs) { Ok(_) => Ok(()), Err(e) => { // EXDEV / permission / etc -> fallback copy fs::copy(final_media_path, &link_abs) .map(|_| ()) .map_err(|e2| format!("playlist link failed: {e} / copy failed: {e2}")) } } } fn sanitize_component(s: &str) -> String { let mut out = String::with_capacity(s.len()); for ch in s.chars() { // conservative sanitization for Linux/Windows compatibility if matches!(ch, '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*' ) { out.push('_'); } else { out.push(ch); } } let out = out.trim().to_string(); if out.is_empty() { "_".into() } else { out.chars().take(100).collect() } } fn make_download_item_movie(item: &JfItem) -> Result { let name = sanitize_component(&item.name); let rel = PathBuf::from("movies").join(format!("{name}.watch")); Ok(DownloadItem { item_id: item.id.clone(), title: item.name.clone(), kind: DownloadKind::Movie, final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), bytes_downloaded: 0, bytes_total: None, download_speed: 0, eta_secs: None, gid: None, playlist_links: Vec::new(), }) } fn make_download_item_episode(item: &JfItem) -> Result { let series = item.series_name.clone().unwrap_or_else(|| "Unknown Show".into()); let series_s = sanitize_component(&series); let season_no = item.parent_index_number; let ep_no = item.index_number; let season_folder = season_no .map(|n| format!("Season-{}", n)) .unwrap_or_else(|| "Season-0".into()); let title_s = sanitize_component(&item.name); let filename = match (season_no, ep_no) { (Some(s), Some(e)) => format!("S{:02}E{:02} - {title_s}.watch", s, e), _ => format!("{title_s}.watch"), }; let rel = PathBuf::from("shows") .join(series_s) .join(season_folder) .join(filename); Ok(DownloadItem { item_id: item.id.clone(), title: format!("{series} - {}", item.name), kind: DownloadKind::Episode { series, season: season_no, episode: ep_no, }, final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), bytes_downloaded: 0, bytes_total: None, download_speed: 0, eta_secs: None, gid: None, playlist_links: Vec::new(), }) } fn make_download_item_track(item: &JfItem) -> Result { let artist = item .album_artist .clone() .or_else(|| item.artists.get(0).cloned()); let album = item.album.clone(); let track_no = item.index_number; let artist_s = sanitize_component(artist.as_deref().unwrap_or("Unknown Artist")); let album_s = sanitize_component(album.as_deref().unwrap_or("Unknown Album")); let title_s = sanitize_component(&item.name); let filename = match track_no { Some(n) => format!("{:02} - {title_s}.listen", n), None => format!("{title_s}.listen"), }; let rel = PathBuf::from("music") .join(artist_s) .join(album_s) .join(filename); Ok(DownloadItem { item_id: item.id.clone(), title: item.name.clone(), kind: DownloadKind::Track { artist, album, track_no, }, final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), bytes_downloaded: 0, bytes_total: None, download_speed: 0, eta_secs: None, gid: None, playlist_links: Vec::new(), }) } fn make_download_item_other(item: &JfItem) -> Result { let name = sanitize_component(&item.name); let rel = PathBuf::from("other").join(format!("{name}.watch")); Ok(DownloadItem { item_id: item.id.clone(), title: item.name.clone(), kind: DownloadKind::Other, final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), bytes_downloaded: 0, bytes_total: None, download_speed: 0, eta_secs: None, gid: None, playlist_links: Vec::new(), }) } fn tmp_dir_of_item(it: &DownloadItem) -> Result { temp_dir_for(it) } // Find the actual downloaded media file in tmp dir. // We ignore aria2 metadata (*.aria2) and partials (*.part). fn find_real_downloaded_file(tmp_dir: &Path) -> Result { let mut best: Option<(u64, PathBuf)> = None; let rd = fs::read_dir(tmp_dir).map_err(|e| format!("read temp dir: {e}"))?; for ent in rd { let p = ent.map_err(|e| format!("read dir entry: {e}"))?.path(); if !p.is_file() { continue; } let name = p.file_name().and_then(|s| s.to_str()).unwrap_or(""); if name.ends_with(".aria2") || name.ends_with(".part") { continue; } let sz = fs::metadata(&p).map_err(|e| format!("stat temp file: {e}"))?.len(); match &best { None => best = Some((sz, p)), Some((best_sz, _)) => { if sz > *best_sz { best = Some((sz, p)); } } } } best.map(|(_, p)| p) .ok_or_else(|| "no downloaded media file found in temp dir".to_string()) } // ---------- worker loop ---------- fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver) { let _ = ensure_dirs(); let token = cfg.access_token.clone().unwrap_or_default(); let mut shutdown = false; let mut aria2_daemon = match start_aria2_daemon() { Ok(c) => c, Err(e) => { if let Ok(mut st) = state.lock() { st.last_message = Some(format!("downloads disabled: {e}")); } return; } }; while !shutdown { // ------------------------- // 1) Handle incoming commands // ------------------------- while let Ok(cmd) = rx.try_recv() { match cmd { DownloadCommand::Enqueue(mut items) => { if let Ok(mut st) = state.lock() { // dedupe by item_id + final_rel_path for it in items.drain(..) { let already = st.queue.iter().any(|q| { q.item_id == it.item_id && q.final_rel_path == it.final_rel_path }); if !already { st.queue.push_back(it); } } st.last_message = Some("queued".into()); } } DownloadCommand::Cancel { index } => { if index == 0 { // cancel the active download (front) let gid = state.lock().ok().and_then(|s| s.active_gid.clone()); if let Some(g) = gid { let _ = aria2_rpc_call("aria2.forceRemove", json!([g.clone()])); let _ = aria2_rpc_call("aria2.removeDownloadResult", json!([g])); } // mark cancelled + delete temp let active_it = state.lock().ok().and_then(|s| s.queue.front().cloned()); if let Some(it) = active_it { if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Cancelled; } st.last_message = Some("cancelled active download".into()); st.active_gid = None; } let _ = delete_temp_for(&it); } // remove from queue if let Ok(mut st) = state.lock() { let _ = st.queue.pop_front(); } } else { // cancel queued item (not active) if let Ok(mut st) = state.lock() { if index < st.queue.len() { let mut v: Vec<_> = st.queue.drain(..).collect(); if index < v.len() { let mut it = v.remove(index); it.status = DownloadStatus::Cancelled; let _ = delete_temp_for(&it); } st.queue = v.into(); st.last_message = Some("removed from queue".into()); } } } } DownloadCommand::RemoveQueued { index } => { if let Ok(mut st) = state.lock() { if index > 0 && index < st.queue.len() { let mut v: Vec<_> = st.queue.drain(..).collect(); let mut it = v.remove(index); it.status = DownloadStatus::Cancelled; let _ = delete_temp_for(&it); st.queue = v.into(); st.last_message = Some("removed from queue".into()); } } } DownloadCommand::EnqueuePlaylistItem { item, playlist_name, playlist_index } => { // If already downloaded, just create the playlist link immediately. if let Some(local) = local_media_path_for_item_id(&item.id) { let _ = ensure_playlist_link(&local, &playlist_name, playlist_index, &item.name); if let Ok(mut st) = state.lock() { st.last_message = Some("playlist link created".into()); } continue; } // Build a DownloadItem like normal, but attach a playlist link intent. let tt = item.item_type.as_deref().unwrap_or(""); let mut di = match tt { "Audio" => make_download_item_track(&item), "Episode" => make_download_item_episode(&item), "Movie" => make_download_item_movie(&item), _ => make_download_item_other(&item), }; let mut di = match di { Ok(v) => v, Err(e) => { if let Ok(mut st) = state.lock() { st.last_message = Some(format!("enqueue playlist item failed: {e}")); } continue; } }; di.playlist_links.push(PlaylistLink { playlist_name, playlist_index, title: item.name.clone(), }); if let Ok(mut st) = state.lock() { // If same media already queued, merge playlist link intents. if let Some(existing) = st.queue.iter_mut().find(|q| q.item_id == di.item_id && q.final_rel_path == di.final_rel_path) { for pl in di.playlist_links.drain(..) { let already = existing.playlist_links.iter().any(|x| x.playlist_name == pl.playlist_name && x.playlist_index == pl.playlist_index ); if !already { existing.playlist_links.push(pl); } } st.last_message = Some("queued (merged playlist link)".into()); } else { st.queue.push_back(di); st.last_message = Some("queued (playlist)".into()); } } } DownloadCommand::Shutdown => { shutdown = true; } } } if shutdown { break; } // ------------------------- // 2) If nothing active, start next // ------------------------- let (has_queue, has_active_gid, next_item) = { let st = match state.lock() { Ok(s) => s, Err(_) => break, }; let has_queue = !st.queue.is_empty(); let has_active_gid = st.active_gid.is_some(); let next_item = st.queue.front().cloned(); (has_queue, has_active_gid, next_item) }; if !has_queue { thread::sleep(Duration::from_millis(120)); continue; } if !has_active_gid { if let Some(it) = next_item { // mark downloading if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Downloading; } } match aria2_add_download(&cfg.base_url, &token, &it) { Ok(gid) => { if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.gid = Some(gid.clone()); front.status = DownloadStatus::Downloading; } st.active_gid = Some(gid); st.last_message = Some(format!("downloading {}", it.title)); } } Err(e) => { if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Failed(e.clone()); } st.last_message = Some(format!("download failed: {e}")); let _ = st.queue.pop_front(); st.active_gid = None; } } } } thread::sleep(Duration::from_millis(80)); continue; } // ------------------------- // 3) Poll progress + handle completion // ------------------------- poll_progress_for_front(&state); match aria2_front_status(&state).as_deref() { Some("complete") => { // mark finalizing if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Finalizing; } } // finalize if let Err(e) = finalize_front(&state) { if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Failed(e.clone()); } st.last_message = Some(e); let _ = st.queue.pop_front(); st.active_gid = None; } } else { // remove aria2 result let gid = state.lock().ok().and_then(|s| s.active_gid.clone()); if let Some(g) = gid { let _ = aria2_rpc_call("aria2.removeDownloadResult", json!([g])); } if let Ok(mut st) = state.lock() { let _ = st.queue.pop_front(); st.active_gid = None; st.last_message = Some("download complete".into()); } } } Some("error") => { if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Failed("aria2 status=error".into()); } st.last_message = Some("aria2 failed".into()); let _ = st.queue.pop_front(); st.active_gid = None; } } Some("active") | Some("waiting") => { thread::sleep(Duration::from_millis(200)); } _ => { thread::sleep(Duration::from_millis(200)); } } } // ------------------------- // shutdown cleanup // ------------------------- let _ = aria2_rpc_call("aria2.shutdown", json!([])); let _ = aria2_daemon.kill(); let _ = aria2_daemon.wait(); let _ = cleanup_temp_dir(); } // Each download gets its own isolated temp folder so deletes can't affect others. // temp/jobs// fn temp_dir_for(it: &DownloadItem) -> Result { let tmp = temp_root()?; Ok(tmp.join("jobs").join(&it.item_id)) } fn delete_temp_for(it: &DownloadItem) -> Result<(), String> { let dir = temp_dir_for(it)?; if dir.exists() { fs::remove_dir_all(&dir).map_err(|e| format!("rm temp: {e}"))?; } Ok(()) } fn aria2_add_download(base_url: &str, token: &str, it: &DownloadItem) -> Result { let base = base_url.trim_end_matches('/'); let url = format!("{}/Items/{}/Download", base, it.item_id); let tmp_dir = temp_dir_for(it)?; if tmp_dir.exists() { let _ = fs::remove_dir_all(&tmp_dir); } fs::create_dir_all(&tmp_dir).map_err(|e| format!("mk temp dir: {e}"))?; // aria2 options let opts = json!({ "dir": tmp_dir.to_string_lossy(), "allow-overwrite": "true", "auto-file-renaming": "false", "continue": "true", "check-certificate": "true", "content-disposition": "true", "header": format!("X-Emby-Token: {token}"), }); // params: [ [uri], options ] let result = aria2_rpc_call("aria2.addUri", json!([[url], opts]))?; result.as_str() .map(|s| s.to_string()) .ok_or_else(|| format!("aria2.addUri returned unexpected result: {result}")) } fn poll_progress_for_front(state: &Arc>) { let gid = { let st = match state.lock() { Ok(s) => s, Err(_) => return, }; st.queue.front().and_then(|it| it.gid.clone()) }; let gid = match gid { Some(g) => g, None => return, }; let res = match aria2_rpc_call("aria2.tellStatus", json!([gid, ["status","completedLength","totalLength","downloadSpeed"]])) { Ok(v) => v, Err(_) => return, }; let status = res.get("status").and_then(|v| v.as_str()).unwrap_or(""); let completed = res.get("completedLength").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()).unwrap_or(0); let total = res.get("totalLength").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()); let speed = res.get("downloadSpeed").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()).unwrap_or(0); let eta = match (total, speed) { (Some(t), s) if t > completed && s > 0 => Some((t - completed) / s), _ => None, }; if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.bytes_downloaded = completed; front.bytes_total = total; front.download_speed = speed; front.eta_secs = eta; // optional: if aria2 says error, reflect it if status == "error" { front.status = DownloadStatus::Failed("aria2 error".into()); } } } } fn finalize_front(state: &Arc>) -> Result<(), String> { let (it, tmp_dir, final_abs_watch) = { let st = state.lock().map_err(|_| "lock poisoned".to_string())?; let it = st.queue.front().ok_or_else(|| "no front".to_string())?.clone(); let tmp_dir = tmp_dir_of_item(&it)?; let dl_root = downloads_root()?; let final_abs_watch = dl_root.join(&it.final_rel_path); (it, tmp_dir, final_abs_watch) }; // Ensure final dir exists let final_parent = final_abs_watch .parent() .ok_or_else(|| "no final parent".to_string())?; fs::create_dir_all(final_parent).map_err(|e| format!("mk final dir: {e}"))?; // Find real downloaded file (with real extension) inside tmp_dir let real_file = find_real_downloaded_file(&tmp_dir)?; // Build final filename: // final_rel_path ends in "...something.watch/.listen" // Keep stem, replace extension with actual media extension let real_ext = real_file .extension() .and_then(|s| s.to_str()) .unwrap_or("") .to_string(); let stem = final_abs_watch .file_stem() .and_then(|s| s.to_str()) .ok_or_else(|| "bad final filename stem".to_string())? .to_string(); let final_abs = if real_ext.is_empty() { final_abs_watch.clone() } else { final_parent.join(format!("{stem}.{real_ext}")) }; // Move file into place fs::rename(&real_file, &final_abs).map_err(|e| format!("move to final failed: {e}"))?; // ✅ NEW: write marker so we can detect "downloaded" later // (If marker fails, we still consider download successful.) let _ = write_download_marker(&final_abs, &it.item_id); // NEW: create playlist links (best-effort) for pl in &it.playlist_links { let _ = ensure_playlist_link(&final_abs, &pl.playlist_name, pl.playlist_index, &pl.title); } // Cleanup temp folder fully let _ = fs::remove_dir_all(&tmp_dir); Ok(()) } const ARIA2_RPC_HOST: &str = "127.0.0.1"; const ARIA2_RPC_PORT: u16 = 6800; fn aria2_http_post(body: &str) -> Result { let mut stream = TcpStream::connect((ARIA2_RPC_HOST, ARIA2_RPC_PORT)) .map_err(|e| format!("aria2 rpc connect failed: {e}"))?; let req = format!( "POST /jsonrpc HTTP/1.1\r\nHost: {}:{}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", ARIA2_RPC_HOST, ARIA2_RPC_PORT, body.len(), body ); stream.write_all(req.as_bytes()).map_err(|e| format!("aria2 rpc write failed: {e}"))?; let mut resp = String::new(); stream.read_to_string(&mut resp).map_err(|e| format!("aria2 rpc read failed: {e}"))?; // split headers/body let (_hdrs, body) = resp.split_once("\r\n\r\n").ok_or("aria2 rpc bad http response")?; Ok(body.to_string()) } fn aria2_rpc_call(method: &str, params: Value) -> Result { let payload = json!({ "jsonrpc": "2.0", "id": "cjtui", "method": method, "params": params, }); let body = payload.to_string(); let raw = aria2_http_post(&body)?; let v: Value = serde_json::from_str(raw.trim()).map_err(|e| format!("aria2 rpc json parse failed: {e} ({raw})"))?; if let Some(err) = v.get("error") { return Err(format!("aria2 rpc error: {err}")); } Ok(v.get("result").cloned().unwrap_or(Value::Null)) } fn start_aria2_daemon() -> Result { let mut cmd = Command::new("aria2c"); cmd.arg("--enable-rpc=true") .arg("--rpc-listen-all=false") .arg(format!("--rpc-listen-port={}", ARIA2_RPC_PORT)) .arg("--rpc-allow-origin-all=true") .arg("--rpc-max-request-size=2M") .arg("--check-certificate=true") .arg("--continue=true") .arg("--max-tries=0") .arg("--retry-wait=5") .arg("--timeout=60") .arg("--connect-timeout=10") .arg("--summary-interval=1") .arg("--console-log-level=warn") .arg("--file-allocation=none") .arg("--content-disposition=true") .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()); let child = cmd.spawn().map_err(|e| format!("spawn aria2c rpc daemon failed: {e}"))?; // quick readiness check (try rpc a few times) let start = Instant::now(); while start.elapsed() < Duration::from_millis(800) { if aria2_rpc_call("aria2.getVersion", json!([])).is_ok() { return Ok(child); } thread::sleep(Duration::from_millis(40)); } Err("aria2 rpc daemon did not become ready".into()) } fn aria2_front_status(state: &Arc>) -> Option { let gid = { let st = state.lock().ok()?; st.queue.front().and_then(|it| it.gid.clone()) }?; let res = aria2_rpc_call("aria2.tellStatus", json!([gid, ["status"]])).ok()?; res.get("status").and_then(|v| v.as_str()).map(|s| s.to_string()) } /// If this item_id has a local download marker, return the local media file path. /// Marker files are written alongside the media file as: /// . /// where DOWNLOAD_MARKER_EXT == "cjtui.json" pub fn local_media_path_for_item_id(item_id: &str) -> Option { fn walk(dir: &Path, item_id: &str) -> Option { if !dir.exists() { return None; } let rd = fs::read_dir(dir).ok()?; for ent in rd { let p = ent.ok()?.path(); if p.is_dir() { if let Some(found) = walk(&p, item_id) { return Some(found); } continue; } let name = p.file_name().and_then(|s| s.to_str()).unwrap_or(""); if !name.ends_with(DOWNLOAD_MARKER_EXT) { continue; } let raw = fs::read_to_string(&p).ok()?; let v: Value = serde_json::from_str(raw.trim()).ok()?; let id = v.get("item_id").and_then(|x| x.as_str())?; if id == item_id { // Marker file is "." // So the media file path is the same path with that suffix stripped. let suffix = format!(".{DOWNLOAD_MARKER_EXT}"); if !name.ends_with(&suffix) { continue; } let media_name = &name[..name.len() - suffix.len()]; let media_path = p.with_file_name(media_name); if media_path.exists() { return Some(media_path); } } } None } let root = downloads_root().ok()?; walk(&root, item_id) } // Delete ALL local copies of an item by scanning *.cjtui.json markers recursively. // Returns how many media files we attempted to remove (count of matched markers). pub fn delete_local_by_item_id(item_id: &str) -> Result { let root = downloads_root()?; let mut deleted = 0usize; delete_in_dir_by_item_id(&root, item_id, &mut deleted)?; Ok(deleted) } // Delete local content based on a concrete local file path. // If there is a sibling marker, we read item_id and delete ALL copies (same as online behavior). // If there is no marker, we delete just that file + sidecars. pub fn delete_local_by_path(path: &Path) -> Result { // If there is a sibling marker, prefer item_id-based deletion (removes hardlinks elsewhere too). if let Some(item_id) = item_id_for_media_path(path)? { return delete_local_by_item_id(&item_id); } // No marker => delete just this file + any obvious sidecars. delete_file_and_sidecars(path)?; Ok(1) } // -------------------- // Internal helpers // -------------------- fn delete_in_dir_by_item_id(dir: &Path, item_id: &str, deleted: &mut usize) -> Result<(), String> { if !dir.exists() { return Ok(()); } for ent in fs::read_dir(dir).map_err(|e| format!("read_dir failed: {e}"))? { let p = ent.map_err(|e| format!("dir entry failed: {e}"))?.path(); if p.is_dir() { delete_in_dir_by_item_id(&p, item_id, deleted)?; continue; } // marker files only let name = p.file_name().and_then(|s| s.to_str()).unwrap_or(""); if !name.ends_with(DOWNLOAD_MARKER_EXT) { continue; } let raw = match fs::read_to_string(&p) { Ok(s) => s, Err(_) => continue, }; let v: Value = match serde_json::from_str(raw.trim()) { Ok(v) => v, Err(_) => continue, }; let mid = match v.get("item_id").and_then(|x| x.as_str()) { Some(s) => s.trim(), None => continue, }; if mid != item_id { continue; } // If we matched the item_id, delete: // - media file referenced by marker (if present) // - marker itself // - media sidecars (.aria2/.part) if let Some(media_name) = v.get("media_file").and_then(|x| x.as_str()) { let media_path = p .parent() .unwrap_or(dir) .join(media_name); let _ = delete_file_and_sidecars(&media_path); } let _ = fs::remove_file(&p); // remove marker *deleted += 1; } Ok(()) } // Reads sibling marker ".cjtui.json" and returns item_id if present. fn item_id_for_media_path(media_path: &Path) -> Result, String> { let fname = match media_path.file_name().and_then(|s| s.to_str()) { Some(s) => s, None => return Ok(None), }; let marker = media_path .parent() .unwrap_or_else(|| Path::new(".")) .join(format!("{fname}.{DOWNLOAD_MARKER_EXT}")); if !marker.exists() { return Ok(None); } let raw = fs::read_to_string(&marker).map_err(|e| format!("read marker failed: {e}"))?; let v: Value = serde_json::from_str(raw.trim()).map_err(|e| format!("marker json parse failed: {e}"))?; Ok(v.get("item_id").and_then(|x| x.as_str()).map(|s| s.to_string())) } fn delete_file_and_sidecars(path: &Path) -> Result<(), String> { // delete main file let _ = fs::remove_file(path); // delete sibling marker (if any) if let Some(fname) = path.file_name().and_then(|s| s.to_str()) { let marker = path .parent() .unwrap_or_else(|| Path::new(".")) .join(format!("{fname}.{DOWNLOAD_MARKER_EXT}")); let _ = fs::remove_file(marker); } // delete obvious aria2 artifacts if let Some(fname) = path.file_name().and_then(|s| s.to_str()) { let parent = path.parent().unwrap_or_else(|| Path::new(".")); let aria2 = parent.join(format!("{fname}.aria2")); let part = parent.join(format!("{fname}.part")); let _ = fs::remove_file(aria2); let _ = fs::remove_file(part); } Ok(()) }