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<i32>, episode: Option<i32> },
    Track { artist: Option<String>, album: Option<String>, track_no: Option<i32> },
    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<u64>,
    pub download_speed: u64,      // bytes/sec
    pub eta_secs: Option<u64>,    // seconds remaining

    // aria2
    pub gid: Option<String>,
}

#[derive(Debug)]
pub enum DownloadCommand {
    Enqueue(Vec<DownloadItem>),
    Cancel { index: usize }, // cancel by queue index (including current 0)
    RemoveQueued { index: usize }, // remove if queued (not current)
    Shutdown,
}

#[derive(Default)]
pub struct DownloadState {
    pub queue: VecDeque<DownloadItem>,
    pub active_gid: Option<String>,
    pub last_message: Option<String>,
}

pub struct DownloadManager {
    tx: mpsc::Sender<DownloadCommand>,
    pub state: Arc<Mutex<DownloadState>>,
}

impl DownloadManager {
    pub fn start(cfg: Config) -> Self {
        let state = Arc::new(Mutex::new(DownloadState::default()));
        let (tx, rx) = mpsc::channel::<DownloadCommand>();

        // 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 = "cj-jftui.json";

fn marker_path_for_media(media_path: &Path) -> PathBuf {
    // Example:
    // Movie.mkv  -> Movie.mkv.cj-jftui.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<String>) -> 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 *.cj-jftui.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<HashSet<String>, String> {
    let root = downloads_root()?;
    let mut out = HashSet::new();
    scan_dir_for_markers(&root, &mut out)?;
    Ok(out)
}


pub fn downloads_root() -> Result<PathBuf, String> {
    let dir = paths::app_config_dir().ok_or_else(|| "no config dir".to_string())?;
    Ok(dir.join("downloads"))
}

pub fn temp_root() -> Result<PathBuf, String> {
    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<Vec<DownloadItem>, 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<Vec<JfItem>, 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<Vec<JfItem>, 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 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<DownloadItem, String> {
    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,
    })
}

fn make_download_item_episode(item: &JfItem) -> Result<DownloadItem, String> {
    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,
    })
}

fn make_download_item_track(item: &JfItem) -> Result<DownloadItem, String> {
    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,
    })
}

fn make_download_item_other(item: &JfItem) -> Result<DownloadItem, String> {
    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,
    })
}


fn tmp_dir_of_item(it: &DownloadItem) -> Result<PathBuf, String> {
    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<PathBuf, String> {
    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<Mutex<DownloadState>>, rx: mpsc::Receiver<DownloadCommand>) {
    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::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/<item_id>/
fn temp_dir_for(it: &DownloadItem) -> Result<PathBuf, String> {
    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<String, String> {
    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<Mutex<DownloadState>>) {
    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::<u64>().ok()).unwrap_or(0);
    let total = res.get("totalLength").and_then(|v| v.as_str()).and_then(|s| s.parse::<u64>().ok());
    let speed = res.get("downloadSpeed").and_then(|v| v.as_str()).and_then(|s| s.parse::<u64>().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<Mutex<DownloadState>>) -> 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);

    // 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<String, String> {
    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<Value, String> {
    let payload = json!({
        "jsonrpc": "2.0",
        "id": "cj-jftui",
        "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<Child, String> {
    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<Mutex<DownloadState>>) -> Option<String> {
    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:
///   <media_filename>.<DOWNLOAD_MARKER_EXT>
/// where DOWNLOAD_MARKER_EXT == "cj-jftui.json"
pub fn local_media_path_for_item_id(item_id: &str) -> Option<PathBuf> {
    fn walk(dir: &Path, item_id: &str) -> Option<PathBuf> {
        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 "<media_filename>.<DOWNLOAD_MARKER_EXT>"
                // 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)
}