use crate::api::ClientVersion;
use crate::providers::Provider;
use crate::status::*;
use crate::util::cache::VideoCache;
use crate::util::requester::Requester;
use crate::videos::{ServerOptions, VideoFormat, VideoItem};
use crate::{DbPool, db};

use async_trait::async_trait;
use chrono::NaiveDateTime;
use error_chain::error_chain;
use futures::stream::{self, StreamExt};
use htmlentity::entity::{ICodedDataTrait, decode};
use regex::Regex;
use serde::Deserialize;
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex, OnceLock, RwLock};
use std::vec;
use wreq::Version;

// How long a cached listing/search entry is considered usable at all.
const HARD_TTL_SECS: u64 = 60 * 60 * 24;
// Past this age we still answer instantly from cache/DB but trigger a
// background refresh so the next request gets fresh data.
const SOFT_TTL_SECS: u64 = 60 * 60;

const SITE: &str = "https://hentaihaven.xxx";
// Poster art lives on a separate image host; the catalogue only stores the path.
const IMG_BASE: &str = "https://img.hentaihaven.xxx/";
// The catalogue API rejects the request outright above this page size.
const MAX_PER_PAGE: usize = 40;

pub const CHANNEL_METADATA: crate::providers::ProviderChannelMetadata =
    crate::providers::ProviderChannelMetadata {
        group_id: "hentai-animation",
        tags: &["hentai", "anime", "curated"],
    };

error_chain! {
    foreign_links {
        Io(std::io::Error);
        HttpRequest(wreq::Error);
        Json(serde_json::Error);
    }
    errors {
        Parse(msg: String) {
            description("parse error")
            display("parse error: {}", msg)
        }
    }
}

#[derive(Debug, Deserialize, Default)]
struct ApiTitle {
    #[serde(default)]
    rendered: String,
}

#[derive(Debug, Deserialize, Default)]
struct ApiMeta {
    #[serde(default)]
    vraven_remote_thumbnail: String,
}

#[derive(Debug, Deserialize)]
struct ApiEntry {
    id: u64,
    #[serde(default)]
    date: String,
    slug: String,
    #[serde(default)]
    title: ApiTitle,
    #[serde(default)]
    meta: ApiMeta,
}

#[derive(Debug, Deserialize, Default)]
struct ApiEngagement {
    #[serde(default)]
    rating: Option<f32>,
    #[serde(default)]
    views: Option<u32>,
}

#[derive(Debug, Deserialize)]
struct ApiResponse {
    #[serde(default)]
    data: Vec<ApiEntry>,
    #[serde(default)]
    engagement: HashMap<String, ApiEngagement>,
}

/// Everything the catalogue API already tells us about a series, so resolving
/// an item only has to go looking for its episodes.
#[derive(Debug, Clone)]
struct ListingEntry {
    id: u64,
    slug: String,
    url: String,
    title: String,
    thumb: String,
    views: u32,
    rating: f32,
    uploaded_at: u64,
}

#[derive(Debug, Clone)]
pub struct HentaihavenProvider {
    categories: Arc<RwLock<Vec<FilterOption>>>,
}

impl HentaihavenProvider {
    pub fn new() -> Self {
        Self {
            categories: Arc::new(RwLock::new(vec![])),
        }
    }

    fn build_channel(&self, clientversion: ClientVersion) -> Channel {
        let _ = clientversion;
        Channel {
            id: "hentaihaven".to_string(),
            name: "Hentai Haven".to_string(),
            description: "Watch Free Hentai Videos HD!".to_string(),
            premium: false,
            favicon: "https://www.google.com/s2/favicons?sz=64&domain=hentaihaven.xxx".to_string(),
            status: "active".to_string(),
            categories: self
                .categories
                .read()
                .map(|categories| categories.iter().map(|c| c.title.clone()).collect())
                .unwrap_or_else(|e| {
                    crate::providers::report_provider_error_background(
                        "hentaihaven",
                        "build_channel.categories_read",
                        &e.to_string(),
                    );
                    vec![]
                }),
            options: vec![],
            nsfw: true,
            cacheDuration: None,
            ytdlpCommand: Some("yt-dlp --impersonate chrome-120".to_string()),
        }
    }

    fn push_unique(target: &Arc<RwLock<Vec<FilterOption>>>, item: FilterOption) {
        if let Ok(mut vec) = target.write() {
            if !vec.iter().any(|x| x.id == item.id) {
                vec.push(item);
            }
        }
    }

    /// Build the catalogue API URL backing a listing or search request. The URL
    /// doubles as the cache key, so every parameter that changes the result set
    /// has to be part of it.
    fn api_url(page: u8, per_page: usize, sort: &str, query: Option<&str>) -> String {
        let per_page = per_page.clamp(1, MAX_PER_PAGE);
        let mut serializer = url::form_urlencoded::Serializer::new(String::new());
        serializer
            .append_pair("per_page", &per_page.to_string())
            .append_pair("page", &page.max(1).to_string());
        if let Some(query) = query {
            serializer.append_pair("search", query);
        } else if matches!(
            sort,
            "views" | "popular" | "most-viewed" | "trending" | "hot" | "rating"
        ) {
            // The catalogue's ranked chart, which is what the site itself serves
            // on /browse/trending/.
            serializer
                .append_pair("trending_period", "monthly")
                .append_pair("live", "1");
        }
        format!("{SITE}/api/manga/?{}", serializer.finish())
    }

    async fn list(
        &self,
        cache: VideoCache,
        page: u8,
        per_page: usize,
        sort: &str,
        query: Option<&str>,
        options: ServerOptions,
        pool: DbPool,
    ) -> Result<Vec<VideoItem>> {
        let api_url = Self::api_url(page, per_page, sort, query);

        // Fast path: a usable in-memory entry exists. Answer immediately; once it
        // is older than the soft TTL, kick a background refresh so the next caller
        // sees fresher data without anyone waiting on it now.
        if let Some((time, items)) = cache.get(&api_url) {
            let age = time.elapsed().unwrap_or_default().as_secs();
            if age < HARD_TTL_SECS && !items.is_empty() {
                if age >= SOFT_TTL_SECS {
                    let requester = crate::providers::requester_or_default(
                        &options,
                        module_path!(),
                        "missing_requester",
                    );
                    self.spawn_refresh(requester, pool, cache, api_url, None);
                }
                return Ok(items);
            }
        }

        // One cheap JSON request gives us the whole page of series with titles,
        // posters and engagement counts already attached.
        let mut requester =
            crate::providers::requester_or_default(&options, module_path!(), "missing_requester");
        let entries = match Self::fetch_listing(&mut requester, &api_url).await {
            Ok(entries) => entries,
            Err(e) => {
                crate::providers::report_provider_error(
                    "hentaihaven",
                    "list.request",
                    &format!("url={api_url}; error={e}"),
                )
                .await;
                return Ok(Self::cached(&cache, &api_url));
            }
        };
        if entries.is_empty() {
            return Ok(Self::cached(&cache, &api_url));
        }

        // Serve the page straight from the DB when it covers *every* series on it,
        // then refresh the whole listing in the background. Partial coverage is
        // not good enough: returning only the handful of already-resolved series
        // would silently hand the client a short page.
        let db_items = Self::items_from_db(&entries, &pool);
        if db_items.len() == entries.len() {
            cache.insert(api_url.clone(), db_items.clone());
            self.spawn_refresh(requester, pool, cache, api_url, Some(entries));
            return Ok(db_items);
        }

        // Cold start (or a page with newly published series): resolve now so the
        // caller gets the full page.
        let items = self.resolve_entries(entries, &requester, pool).await;
        if !items.is_empty() {
            cache.insert(api_url.clone(), items.clone());
            return Ok(items);
        }
        Ok(Self::cached(&cache, &api_url))
    }

    fn cached(cache: &VideoCache, key: &str) -> Vec<VideoItem> {
        cache.get(key).map(|(_, items)| items).unwrap_or_default()
    }

    /// Fetch and flatten one page of the catalogue API.
    async fn fetch_listing(
        requester: &mut Requester,
        api_url: &str,
    ) -> std::result::Result<Vec<ListingEntry>, String> {
        let body = Self::get_with_retry(requester, api_url, 3).await?;
        let response: ApiResponse = serde_json::from_str(&body).map_err(|e| {
            format!(
                "failed to parse catalogue json: {e}; body={}",
                body.chars().take(200).collect::<String>()
            )
        })?;

        Ok(response
            .data
            .into_iter()
            .map(|entry| {
                let engagement = response.engagement.get(&entry.id.to_string());
                let title = decode(entry.title.rendered.as_bytes())
                    .to_string()
                    .unwrap_or_else(|_| entry.title.rendered.clone());
                ListingEntry {
                    id: entry.id,
                    url: format!("{SITE}/watch/{}/", entry.slug),
                    slug: entry.slug,
                    title,
                    thumb: Self::thumb_url(&entry.meta.vraven_remote_thumbnail),
                    views: engagement.and_then(|e| e.views).unwrap_or(0),
                    rating: engagement.and_then(|e| e.rating).unwrap_or(0.0),
                    uploaded_at: Self::parse_timestamp(&entry.date),
                }
            })
            .collect())
    }

    /// Poster paths are stored raw and routinely contain spaces, so they have to
    /// go through the URL parser rather than plain concatenation.
    fn thumb_url(path: &str) -> String {
        let path = path.trim();
        if path.is_empty() {
            return String::new();
        }
        if path.starts_with("http") {
            return path.to_string();
        }
        match url::Url::parse(&format!("{IMG_BASE}{}", path.trim_start_matches('/'))) {
            Ok(url) => url.to_string(),
            Err(_) => String::new(),
        }
    }

    /// Catalogue dates are naive local timestamps like `2026-09-01T01:27:10`.
    fn parse_timestamp(date: &str) -> u64 {
        NaiveDateTime::parse_from_str(date.trim(), "%Y-%m-%dT%H:%M:%S")
            .map(|dt| dt.and_utc().timestamp().max(0) as u64)
            .unwrap_or(0)
    }

    /// Build a response from already-resolved items stored in the DB, preserving
    /// the order of `entries`. Items not yet in the DB are simply skipped.
    fn items_from_db(entries: &[ListingEntry], pool: &DbPool) -> Vec<VideoItem> {
        let mut conn = match pool.get() {
            Ok(conn) => conn,
            Err(_) => return vec![],
        };
        entries
            .iter()
            .filter_map(|entry| match db::get_video(&mut conn, entry.url.clone()) {
                Ok(Some(json)) => VideoItem::from(json).ok(),
                _ => None,
            })
            .collect()
    }

    /// Resolve each series into a full `VideoItem`, persisting every success to
    /// the DB. On failure we fall back to any stored copy so a transient error
    /// does not drop the item from the page.
    async fn resolve_entries(
        &self,
        entries: Vec<ListingEntry>,
        requester: &Requester,
        pool: DbPool,
    ) -> Vec<VideoItem> {
        stream::iter(entries.into_iter().map(|entry| {
            let provider = self.clone();
            let mut req = requester.clone();
            let pool = pool.clone();
            async move {
                let url = entry.url.clone();
                match provider.fetch_video_item(&entry, &mut req).await {
                    Ok((item, complete)) => {
                        if let Ok(mut conn) = pool.get() {
                            let old_item = db::get_video(&mut conn, url.clone())
                                .ok()
                                .flatten()
                                .and_then(|json| VideoItem::from(json).ok());
                            let old_len = old_item
                                .as_ref()
                                .and_then(|o| o.formats.as_ref())
                                .map_or(0, |f| f.len());
                            let new_len = item.formats.as_ref().map_or(0, |f| f.len());
                            if complete || new_len >= old_len {
                                // Every episode the series page advertises resolved,
                                // so this is authoritative even when it is shorter
                                // than what we stored earlier.
                                let _ = db::upsert_video(
                                    &mut conn,
                                    &url,
                                    &serde_json::to_string(&item).unwrap_or_default(),
                                );
                                Some(item)
                            } else {
                                // Some episode pages failed this round — keep the
                                // richer stored copy rather than degrading it.
                                old_item.or(Some(item))
                            }
                        } else {
                            Some(item)
                        }
                    }
                    Err(e) => {
                        eprintln!("Hentai Haven Provider: Failed to resolve {url}: {e}");
                        if let Ok(mut conn) = pool.get() {
                            if let Ok(Some(cached)) = db::get_video(&mut conn, url.clone()) {
                                if let Ok(item) = VideoItem::from(cached) {
                                    return Some(item);
                                }
                            }
                        }
                        None
                    }
                }
            }
        }))
        .buffered(2)
        .filter_map(|item| async move { item })
        .collect::<Vec<_>>()
        .await
    }

    /// Per-listing in-flight guard so we never run two background refreshes for
    /// the same page concurrently.
    fn refresh_in_flight() -> &'static Mutex<HashSet<String>> {
        static SET: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();
        SET.get_or_init(|| Mutex::new(HashSet::new()))
    }

    fn try_begin_refresh(key: &str) -> bool {
        match Self::refresh_in_flight().lock() {
            Ok(mut set) => set.insert(key.to_string()),
            Err(_) => false,
        }
    }

    fn end_refresh(key: &str) {
        if let Ok(mut set) = Self::refresh_in_flight().lock() {
            set.remove(key);
        }
    }

    /// Spawn a non-blocking refresh of a listing/search page. `entries` may be
    /// supplied when the caller already fetched the catalogue page; otherwise the
    /// refresh re-fetches it itself.
    fn spawn_refresh(
        &self,
        requester: Requester,
        pool: DbPool,
        cache: VideoCache,
        key: String,
        entries: Option<Vec<ListingEntry>>,
    ) {
        if !Self::try_begin_refresh(&key) {
            return;
        }
        let provider = self.clone();
        tokio::spawn(async move {
            provider
                .refresh(requester, pool, cache, key.clone(), entries)
                .await;
            Self::end_refresh(&key);
        });
    }

    async fn refresh(
        &self,
        mut requester: Requester,
        pool: DbPool,
        cache: VideoCache,
        key: String,
        entries: Option<Vec<ListingEntry>>,
    ) {
        let entries = match entries {
            Some(entries) => entries,
            None => match Self::fetch_listing(&mut requester, &key).await {
                Ok(entries) => entries,
                Err(e) => {
                    crate::providers::report_provider_error_background(
                        "hentaihaven",
                        "refresh.request",
                        &format!("url={key}; error={e}"),
                    );
                    return;
                }
            },
        };
        if entries.is_empty() {
            return;
        }
        let items = self.resolve_entries(entries, &requester, pool).await;
        if !items.is_empty() {
            cache.insert(key, items);
        }
    }

    fn episode_regex(slug: &str) -> Result<Regex> {
        Regex::new(&format!(
            r#"href="(/watch/{}/([^"/]+)/)""#,
            regex::escape(slug)
        ))
        .map_err(|e| Error::from(format!("failed to build episode regex: {e}")))
    }

    fn genre_regex() -> &'static Regex {
        static RE: OnceLock<Regex> = OnceLock::new();
        RE.get_or_init(|| {
            Regex::new(r#"href="/series/([a-z0-9-]+)/"[^>]*>([^<]+)<"#)
                .expect("valid hentaihaven genre regex")
        })
    }

    fn content_url_regex() -> &'static Regex {
        static RE: OnceLock<Regex> = OnceLock::new();
        RE.get_or_init(|| {
            Regex::new(r#""contentUrl"\s*:\s*"([^"]+)""#)
                .expect("valid hentaihaven contentUrl regex")
        })
    }

    fn duration_regex() -> &'static Regex {
        static RE: OnceLock<Regex> = OnceLock::new();
        RE.get_or_init(|| {
            Regex::new(r#""duration"\s*:\s*"(PT[^"]+)""#)
                .expect("valid hentaihaven duration regex")
        })
    }

    /// The series page also renders sidebars full of *other* series' episodes, so
    /// every candidate link is scoped to this series' own slug.
    fn parse_episode_paths(html: &str, slug: &str) -> Result<Vec<(u32, String)>> {
        let regex = Self::episode_regex(slug)?;
        let mut seen: HashSet<String> = HashSet::new();
        let mut episodes: Vec<(u32, String)> = vec![];
        for capture in regex.captures_iter(html) {
            let path = capture[1].to_string();
            let segment = capture[2].to_string();
            if !segment.starts_with("episode") {
                continue;
            }
            if !seen.insert(path.clone()) {
                continue;
            }
            let number = segment
                .rsplit('-')
                .next()
                .and_then(|s| s.parse::<u32>().ok())
                .unwrap_or(u32::MAX);
            episodes.push((number, format!("{SITE}{path}")));
        }
        episodes.sort_by_key(|(number, _)| *number);
        Ok(episodes)
    }

    fn parse_genres(html: &str) -> Vec<FilterOption> {
        let mut seen: HashSet<String> = HashSet::new();
        Self::genre_regex()
            .captures_iter(html)
            .filter_map(|capture| {
                let id = capture[1].to_string();
                let title = decode(capture[2].as_bytes())
                    .to_string()
                    .unwrap_or_else(|_| capture[2].to_string())
                    .trim()
                    .to_string();
                if title.is_empty() || !seen.insert(id.clone()) {
                    return None;
                }
                Some(FilterOption { id, title })
            })
            .collect()
    }

    /// Parse an ISO-8601 duration such as `PT6M32S` into seconds.
    fn parse_iso8601_duration(value: &str) -> u32 {
        let mut seconds = 0u32;
        let mut number = 0u32;
        for c in value.trim_start_matches("PT").chars() {
            match c {
                '0'..='9' => number = number.saturating_mul(10).saturating_add(c as u32 - '0' as u32),
                'H' => {
                    seconds = seconds.saturating_add(number.saturating_mul(3600));
                    number = 0;
                }
                'M' => {
                    seconds = seconds.saturating_add(number.saturating_mul(60));
                    number = 0;
                }
                'S' => {
                    seconds = seconds.saturating_add(number);
                    number = 0;
                }
                _ => number = 0,
            }
        }
        seconds
    }

    /// Resolve one series into a `VideoItem`. The boolean is `true` when every
    /// episode the series page advertises produced a playable format, which lets
    /// the caller tell a genuinely short series from a partly failed fetch.
    async fn fetch_video_item(
        &self,
        entry: &ListingEntry,
        requester: &mut Requester,
    ) -> Result<(VideoItem, bool)> {
        let html = Self::get_with_retry(requester, &entry.url, 3)
            .await
            .map_err(|e| Error::from(format!("Failed to fetch series page: {e}")))?;

        let genres = Self::parse_genres(&html);
        for genre in &genres {
            Self::push_unique(&self.categories, genre.clone());
        }
        let tags: Vec<String> = genres.into_iter().map(|g| g.title).collect();

        let episodes = Self::parse_episode_paths(&html, &entry.slug)?;
        let expected = episodes.len();
        let formats: Vec<(VideoFormat, u32)> = if episodes.is_empty() {
            // A handful of one-shot entries render the player straight onto the
            // series page instead of linking out to an episode.
            vec![Self::format_from_player_html(&html, "Episode 1").ok_or_else(|| {
                ErrorKind::Parse(format!("no episodes and no player on {}", entry.url))
            })?]
        } else {
            stream::iter(episodes.into_iter().map(|(number, url)| {
                let mut req = requester.clone();
                async move {
                    let label = if number == u32::MAX {
                        "Episode".to_string()
                    } else {
                        format!("Episode {number}")
                    };
                    match Self::get_with_retry(&mut req, &url, 3).await {
                        Ok(html) => Self::format_from_player_html(&html, &label),
                        Err(e) => {
                            eprintln!("Hentai Haven Provider: Failed to fetch episode {url}: {e}");
                            None
                        }
                    }
                }
            }))
            .buffered(2)
            .filter_map(|format| async move { format })
            .collect::<Vec<_>>()
            .await
        };

        if formats.is_empty() {
            return Err(Error::from(format!(
                "No formats found for series: {}",
                entry.url
            )));
        }

        // A multi-episode series plays its first episode by default, so that is
        // the duration the client should show.
        let format_count = formats.len();
        let duration = formats.first().map(|(_, secs)| *secs).unwrap_or(0);
        let mut title = entry.title.clone();
        if format_count > 1 {
            title = format!("{} ({} Episodes)", title, format_count);
        }

        let mut item = VideoItem::new(
            entry.id.to_string(),
            title,
            entry.url.clone(),
            "hentaihaven".into(),
            entry.thumb.clone(),
            duration,
        )
        .formats(formats.into_iter().map(|(format, _)| format).collect())
        .tags(tags)
        .views(entry.views)
        .aspect_ratio(0.667);

        if entry.rating > 0.0 {
            item = item.rating(entry.rating);
        }
        if entry.uploaded_at > 0 {
            item = item.uploaded_at(entry.uploaded_at);
        }

        // The single-player fallback has nothing to compare against, so it only
        // gets here when it did produce a format — treat that as complete too.
        let complete = expected == 0 || format_count == expected;
        Ok((item, complete))
    }

    /// Episode pages publish the stream in their `VideoObject` JSON-LD block, so
    /// no player handshake is needed any more. The page also carries an
    /// `ImageObject` whose `contentUrl` is the poster and which is emitted
    /// *before* the video, so the playlist has to be picked out by extension
    /// rather than by taking the first match.
    fn format_from_player_html(html: &str, label: &str) -> Option<(VideoFormat, u32)> {
        let src = Self::content_url_regex()
            .captures_iter(html)
            .map(|c| c[1].to_string())
            .find(|url| Self::is_stream_url(url))?;
        let duration = Self::duration_regex()
            .captures(html)
            .map(|c| Self::parse_iso8601_duration(&c[1]))
            .unwrap_or(0);
        let container = if src.to_lowercase().contains(".mp4") {
            "mp4"
        } else {
            "m3u8"
        };
        let format = VideoFormat::new(src, "auto".to_string(), container.to_string())
            .format_id(label.to_string())
            .format_note(label.to_string());
        Some((format, duration))
    }

    /// Distinguishes the `VideoObject` stream from the poster art that shares
    /// the `contentUrl` key.
    fn is_stream_url(url: &str) -> bool {
        let path = url.split(['?', '#']).next().unwrap_or(url).to_lowercase();
        path.ends_with(".m3u8") || path.ends_with(".mp4")
    }

    async fn get_with_retry(
        requester: &mut Requester,
        url: &str,
        attempts: u32,
    ) -> std::result::Result<String, String> {
        let mut last_err = String::new();
        for attempt in 0..attempts {
            if attempt > 0 {
                let backoff_ms = 500u64 * (1u64 << (attempt - 1).min(3));
                tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await;
            }
            match requester.get(url, Some(Version::HTTP_2)).await {
                Ok(text) => return Ok(text),
                Err(e) => last_err = e.to_string(),
            }
        }
        Err(last_err)
    }
}

#[async_trait]
impl Provider for HentaihavenProvider {
    async fn get_videos(
        &self,
        cache: VideoCache,
        pool: DbPool,
        sort: String,
        query: Option<String>,
        page: String,
        per_page: String,
        options: ServerOptions,
    ) -> Vec<VideoItem> {
        let page = page.parse::<u8>().unwrap_or(1);
        let per_page = per_page.parse::<usize>().unwrap_or(20);

        let res = self
            .list(
                cache,
                page,
                per_page,
                &sort,
                query.as_deref(),
                options,
                pool,
            )
            .await;

        res.unwrap_or_else(|e| {
            eprintln!("hentai haven error: {e}");
            vec![]
        })
    }

    fn get_channel(&self, v: ClientVersion) -> Option<Channel> {
        Some(self.build_channel(v))
    }
}