//! Bounded, deduplicating HTTP download dispatcher.

#![allow(clippy::missing_errors_doc)] // Public result shapes are fixed by the compatibility map.
#![allow(clippy::needless_pass_by_value)] // Mapped APIs preserve owned CLR argument shapes.

use crate::{Error, GridClient, HttpCapsClientProgressReport};
use libremetaverse_types::compat::{
    CancellationToken, CancellationTokenSource, HttpResponse, IProgress, Subscription,
    TaskCompletionSource, Uri,
};
use std::collections::HashMap;
use std::fmt;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use tokio::sync::{Notify, mpsc, oneshot};
use tokio::task::JoinSet;

const DEFAULT_PARALLEL_DOWNLOADS: usize = 8;
const MAX_PARALLEL_DOWNLOADS: usize = 32;
const DOWNLOAD_QUEUE_CAPACITY: usize = 256;

type DownloadValue = (HttpResponse, Vec<u8>);
type DownloadCompletion = TaskCompletionSource<DownloadValue>;
type DownloadResult = Result<DownloadValue, Error>;

fn mutex<T>(value: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    value
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Mapped request state with the same defaults as the C# constructor.
pub struct DownloadRequest {
    pub address: Uri,
    pub attempt: i32,
    pub cancellation_token: CancellationToken,
    pub completion_tcs: Option<DownloadCompletion>,
    pub content_type: Option<String>,
    pub download_progress_callback: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>,
    pub retries: i32,
}

impl DownloadRequest {
    pub fn new(
        address: Uri,
        content_type: Option<String>,
        download_progress_callback: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>,
    ) -> Result<Self, Error> {
        validate_http_uri(&address)?;
        Ok(Self {
            address,
            attempt: 0,
            cancellation_token: CancellationToken::default(),
            completion_tcs: None,
            content_type,
            download_progress_callback,
            retries: 5,
        })
    }
}

impl fmt::Debug for DownloadRequest {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DownloadRequest")
            .field("address", &"<redacted capability URI>")
            .field("attempt", &self.attempt)
            .field(
                "cancellation_requested",
                &self.cancellation_token.is_cancellation_requested(),
            )
            .field("has_completion", &self.completion_tcs.is_some())
            .field("has_content_type", &self.content_type.is_some())
            .field("has_progress", &self.download_progress_callback.is_some())
            .field("retries", &self.retries)
            .finish()
    }
}

struct ActiveDownload {
    cancellation: CancellationTokenSource,
    cancellation_guards: Mutex<Vec<Subscription>>,
    progress: Mutex<Vec<Arc<dyn IProgress<HttpCapsClientProgressReport>>>>,
    completion_sources: Mutex<Vec<DownloadCompletion>>,
    waiters: Mutex<Vec<oneshot::Sender<DownloadResult>>>,
    completed: AtomicBool,
}

impl ActiveDownload {
    fn new() -> Self {
        Self {
            cancellation: CancellationTokenSource::new(),
            cancellation_guards: Mutex::new(Vec::new()),
            progress: Mutex::new(Vec::new()),
            completion_sources: Mutex::new(Vec::new()),
            waiters: Mutex::new(Vec::new()),
            completed: AtomicBool::new(false),
        }
    }

    fn attach_cancellation(&self, token: &CancellationToken) {
        let cancellation = self.cancellation.clone();
        let guard = token.register_callback(Arc::new(move || cancellation.cancel()));
        mutex(&self.cancellation_guards).push(guard);
    }

    fn attach_progress(&self, progress: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>) {
        if let Some(progress) = progress {
            mutex(&self.progress).push(Arc::from(progress));
        }
    }

    fn add_waiter(&self) -> oneshot::Receiver<DownloadResult> {
        let (sender, receiver) = oneshot::channel();
        mutex(&self.waiters).push(sender);
        receiver
    }

    fn attach_completion_source(&self, completion_source: Option<DownloadCompletion>) {
        if let Some(completion_source) = completion_source {
            mutex(&self.completion_sources).push(completion_source);
        }
    }

    fn complete(&self, result: DownloadResult) {
        if self.completed.swap(true, Ordering::AcqRel) {
            return;
        }
        for waiter in std::mem::take(&mut *mutex(&self.waiters)) {
            let _ = waiter.send(result.clone());
        }
        for completion_source in std::mem::take(&mut *mutex(&self.completion_sources)) {
            match result.clone() {
                Ok(value) => {
                    let _ = completion_source.try_set_result(value);
                }
                Err(Error::Cancelled) => {
                    let _ = completion_source.try_set_cancelled();
                }
                Err(error) => {
                    let _ = completion_source.try_set_error(error);
                }
            }
        }
        mutex(&self.progress).clear();
        mutex(&self.cancellation_guards).clear();
    }

    fn report(&self, report: HttpCapsClientProgressReport) {
        let handlers = mutex(&self.progress).clone();
        for handler in handlers {
            let _ = catch_unwind(AssertUnwindSafe(|| handler.report(report)));
        }
    }
}

struct ProgressFanout(Weak<ActiveDownload>);

impl IProgress<HttpCapsClientProgressReport> for ProgressFanout {
    fn report(&self, report: HttpCapsClientProgressReport) {
        if let Some(active) = self.0.upgrade() {
            active.report(report);
        }
    }
}

struct DownloadJob {
    key: String,
    address: Uri,
    attempt: i32,
    retries: i32,
    active: Arc<ActiveDownload>,
    client: GridClient,
}

struct GateState {
    active: usize,
}

struct DynamicGate {
    limit: AtomicUsize,
    state: Mutex<GateState>,
    notify: Notify,
}

impl DynamicGate {
    fn new(limit: usize) -> Self {
        Self {
            limit: AtomicUsize::new(limit),
            state: Mutex::new(GateState { active: 0 }),
            notify: Notify::new(),
        }
    }

    fn set_limit(&self, value: usize) {
        self.limit.store(value, Ordering::Release);
        self.notify.notify_waiters();
    }

    async fn acquire(
        self: &Arc<Self>,
        cancellation: CancellationToken,
    ) -> Result<GatePermit, Error> {
        loop {
            cancellation.throw_if_cancellation_requested()?;
            let notified = self.notify.notified();
            {
                let mut state = mutex(&self.state);
                if state.active < self.limit.load(Ordering::Acquire) {
                    state.active += 1;
                    return Ok(GatePermit(Arc::clone(self)));
                }
            }
            tokio::select! {
                () = notified => {}
                () = cancellation.cancelled() => return Err(Error::Cancelled),
            }
        }
    }
}

struct GatePermit(Arc<DynamicGate>);

impl Drop for GatePermit {
    fn drop(&mut self) {
        let mut state = mutex(&self.0.state);
        state.active = state.active.saturating_sub(1);
        drop(state);
        self.0.notify.notify_one();
    }
}

struct DownloadManagerInner {
    client: GridClient,
    active: Mutex<HashMap<String, Arc<ActiveDownload>>>,
    sender: Mutex<Option<mpsc::Sender<DownloadJob>>>,
    dispatcher: Mutex<Option<JoinHandle<()>>>,
    shutdown: CancellationTokenSource,
    gate: Arc<DynamicGate>,
    parallel_downloads: AtomicUsize,
    disposed: AtomicBool,
}

impl DownloadManagerInner {
    fn remove_active(&self, key: &str, active: &Arc<ActiveDownload>) {
        let mut downloads = mutex(&self.active);
        if downloads
            .get(key)
            .is_some_and(|candidate| Arc::ptr_eq(candidate, active))
        {
            downloads.remove(key);
        }
    }

    fn shutdown(&self) {
        if self.disposed.swap(true, Ordering::AcqRel) {
            return;
        }
        self.shutdown.cancel();
        for active in mutex(&self.active).values() {
            active.cancellation.cancel();
        }
        mutex(&self.sender).take();
        if let Some(dispatcher) = mutex(&self.dispatcher).take()
            && dispatcher.thread().id() != thread::current().id()
        {
            let _ = dispatcher.join();
        }
        let remaining = std::mem::take(&mut *mutex(&self.active));
        for active in remaining.into_values() {
            active.complete(Err(Error::Cancelled));
        }
    }
}

impl Drop for DownloadManagerInner {
    fn drop(&mut self) {
        self.shutdown();
    }
}

/// Native, bounded download manager with URI-level request deduplication.
#[derive(Clone)]
pub struct DownloadManager(Arc<DownloadManagerInner>);

impl DownloadManager {
    pub fn new(client: GridClient) -> Result<Self, Error> {
        let (sender, receiver) = mpsc::channel(DOWNLOAD_QUEUE_CAPACITY);
        let gate = Arc::new(DynamicGate::new(DEFAULT_PARALLEL_DOWNLOADS));
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|_| Error::InvalidOperation)?;
        let inner = Arc::new(DownloadManagerInner {
            client,
            active: Mutex::new(HashMap::new()),
            sender: Mutex::new(Some(sender)),
            dispatcher: Mutex::new(None),
            shutdown: CancellationTokenSource::new(),
            gate: Arc::clone(&gate),
            parallel_downloads: AtomicUsize::new(DEFAULT_PARALLEL_DOWNLOADS),
            disposed: AtomicBool::new(false),
        });
        let weak = Arc::downgrade(&inner);
        let shutdown = inner.shutdown.token();
        let dispatcher = thread::Builder::new()
            .name("libremetaverse-download-dispatcher".to_owned())
            .spawn(move || download_dispatcher(runtime, receiver, weak, gate, shutdown))
            .map_err(|_| Error::InvalidOperation)?;
        *mutex(&inner.dispatcher) = Some(dispatcher);
        Ok(Self(inner))
    }

    pub fn dispose(&self) -> Result<(), Error> {
        self.0.shutdown();
        Ok(())
    }

    /// Number of URI-deduplicated downloads currently queued or executing.
    #[must_use]
    pub fn active_download_count(&self) -> usize {
        mutex(&self.0.active).len()
    }

    /// Whether the dispatcher has completed its idempotent teardown.
    #[must_use]
    pub fn is_disposed(&self) -> bool {
        self.0.disposed.load(Ordering::Acquire)
    }

    /// Whether a dispatcher thread is still owned by this manager.
    #[must_use]
    pub fn dispatcher_running(&self) -> bool {
        mutex(&self.0.dispatcher).is_some()
    }

    #[must_use]
    pub fn parallel_downloads(&self) -> i32 {
        i32::try_from(self.0.parallel_downloads.load(Ordering::Acquire)).unwrap_or(i32::MAX)
    }

    pub fn set_parallel_downloads(&mut self, value: i32) {
        let value = usize::try_from(value)
            .unwrap_or(1)
            .clamp(1, MAX_PARALLEL_DOWNLOADS);
        self.0.parallel_downloads.store(value, Ordering::Release);
        self.0.gate.set_limit(value);
    }

    pub fn queue_download_with_download_request(
        &self,
        request: DownloadRequest,
    ) -> Result<(), Error> {
        self.enqueue(request, false).map(|_| ())
    }

    pub fn queue_download_with_download_request_cancellation_token(
        &self,
        mut request: DownloadRequest,
        cancellation_token: CancellationToken,
    ) -> Result<(), Error> {
        request.cancellation_token = cancellation_token;
        self.queue_download_with_download_request(request)
    }

    pub async fn queue_download_with_download_request_ba727387(
        &self,
        request: DownloadRequest,
    ) -> DownloadResult {
        let cancellation = request.cancellation_token.clone();
        let (active, receiver) = self.enqueue(request, true)?;
        await_download(
            active,
            receiver.ok_or(Error::InvalidOperation)?,
            cancellation,
        )
        .await
    }

    pub async fn queue_download_with_uri_string_i_progress_cancellation_token_int32(
        &self,
        address: Uri,
        content_type: Option<String>,
        progress_callback: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>,
        cancellation_token: Option<CancellationToken>,
        retries: Option<i32>,
    ) -> DownloadResult {
        let cancellation = cancellation_token.unwrap_or_default();
        let mut request = DownloadRequest::new(address, content_type, progress_callback)?;
        request.cancellation_token = cancellation.clone();
        request.retries = retries.unwrap_or(5).max(0);
        let (active, receiver) = self.enqueue(request, true)?;
        await_download(
            active,
            receiver.ok_or(Error::InvalidOperation)?,
            cancellation,
        )
        .await
    }

    pub async fn download_with_uri_string_i_progress_cancellation_token(
        &self,
        address: Uri,
        content_type: String,
        progress: Box<dyn IProgress<HttpCapsClientProgressReport>>,
        cancellation_token: CancellationToken,
    ) -> DownloadResult {
        self.queue_download_with_uri_string_i_progress_cancellation_token_int32(
            address,
            Some(content_type),
            Some(progress),
            Some(cancellation_token),
            Some(5),
        )
        .await
    }

    pub async fn download_with_uri_i_progress_cancellation_token(
        &self,
        address: Uri,
        progress: Box<dyn IProgress<HttpCapsClientProgressReport>>,
        cancellation_token: CancellationToken,
    ) -> DownloadResult {
        self.download_with_uri_string_i_progress_cancellation_token(
            address,
            String::new(),
            progress,
            cancellation_token,
        )
        .await
    }

    fn enqueue(
        &self,
        mut request: DownloadRequest,
        with_waiter: bool,
    ) -> Result<
        (
            Arc<ActiveDownload>,
            Option<oneshot::Receiver<DownloadResult>>,
        ),
        Error,
    > {
        if self.0.disposed.load(Ordering::Acquire) {
            return Err(Error::InvalidOperation);
        }
        let key = validate_http_uri(&request.address)?;
        let selected = {
            let mut active = mutex(&self.0.active);
            if let Some(existing) = active.get(&key) {
                Some((Arc::clone(existing), false))
            } else if request.cancellation_token.is_cancellation_requested() {
                None
            } else {
                let download = Arc::new(ActiveDownload::new());
                active.insert(key.clone(), Arc::clone(&download));
                Some((download, true))
            }
        };
        let Some((active, is_new)) = selected else {
            if let Some(completion_source) = request.completion_tcs.take() {
                let _ = completion_source.try_set_cancelled();
            }
            return Err(Error::Cancelled);
        };
        active.attach_cancellation(&request.cancellation_token);
        active.attach_progress(request.download_progress_callback.take());
        active.attach_completion_source(request.completion_tcs.take());
        let receiver = with_waiter.then(|| active.add_waiter());
        if !is_new {
            return Ok((active, receiver));
        }

        let job = DownloadJob {
            key: key.clone(),
            address: request.address,
            attempt: request.attempt.max(0),
            retries: request.retries.max(0),
            active: Arc::clone(&active),
            client: self.0.client.clone(),
        };
        let sender = mutex(&self.0.sender)
            .clone()
            .ok_or(Error::InvalidOperation)?;
        if sender.try_send(job).is_err() {
            self.0.remove_active(&key, &active);
            active.complete(Err(Error::InvalidOperation));
            return Err(Error::InvalidOperation);
        }
        Ok((active, receiver))
    }
}

impl fmt::Debug for DownloadManager {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DownloadManager")
            .field("active_downloads", &mutex(&self.0.active).len())
            .field("parallel_downloads", &self.parallel_downloads())
            .field("queue_capacity", &DOWNLOAD_QUEUE_CAPACITY)
            .field("disposed", &self.0.disposed.load(Ordering::Acquire))
            .finish()
    }
}

async fn await_download(
    active: Arc<ActiveDownload>,
    receiver: oneshot::Receiver<DownloadResult>,
    cancellation: CancellationToken,
) -> DownloadResult {
    tokio::select! {
        result = receiver => result.map_err(|_| Error::Cancelled)?,
        () = cancellation.cancelled() => {
            active.cancellation.cancel();
            Err(Error::Cancelled)
        }
    }
}

fn download_dispatcher(
    runtime: tokio::runtime::Runtime,
    mut receiver: mpsc::Receiver<DownloadJob>,
    manager: Weak<DownloadManagerInner>,
    gate: Arc<DynamicGate>,
    shutdown: CancellationToken,
) {
    runtime.block_on(async move {
        let mut jobs = JoinSet::new();
        loop {
            tokio::select! {
                biased;
                () = shutdown.cancelled() => break,
                Some(job) = receiver.recv() => {
                    let manager = manager.clone();
                    let gate = Arc::clone(&gate);
                    jobs.spawn(async move {
                        run_download_job(job, manager, gate).await;
                    });
                }
                Some(_) = jobs.join_next(), if !jobs.is_empty() => {}
                else => break,
            }
        }
        while let Ok(job) = receiver.try_recv() {
            job.active.complete(Err(Error::Cancelled));
            if let Some(manager) = manager.upgrade() {
                manager.remove_active(&job.key, &job.active);
            }
        }
        while jobs.join_next().await.is_some() {}
    });
}

async fn run_download_job(
    job: DownloadJob,
    manager: Weak<DownloadManagerInner>,
    gate: Arc<DynamicGate>,
) {
    let cancellation = job.active.cancellation.token();
    let permit = gate.acquire(cancellation.clone()).await;
    let result = match permit {
        Ok(_permit) => perform_download(&job, cancellation).await,
        Err(error) => Err(error),
    };
    if let Some(manager) = manager.upgrade() {
        manager.remove_active(&job.key, &job.active);
    }
    job.active.complete(result);
}

async fn perform_download(job: &DownloadJob, cancellation: CancellationToken) -> DownloadResult {
    let mut attempt = job.attempt;
    loop {
        cancellation.throw_if_cancellation_requested()?;
        let progress: Box<dyn IProgress<HttpCapsClientProgressReport>> =
            Box::new(ProgressFanout(Arc::downgrade(&job.active)));
        let result = job
            .client
            .native_http_caps_client()
            .get(job.address.clone(), cancellation.clone(), Some(progress))
            .await;
        match result {
            Ok((response, data)) if response.is_success_status_code() => {
                return Ok((response, data));
            }
            Ok((response, _)) => {
                if is_permanent_status(response.status_code) || attempt >= job.retries {
                    return Err(Error::HttpRequest);
                }
            }
            Err(Error::Cancelled) => return Err(Error::Cancelled),
            Err(error) if attempt >= job.retries => return Err(error),
            Err(_) => {}
        }
        attempt = attempt.saturating_add(1);
        let base_delay = u64::try_from(200_i32.saturating_mul(attempt).min(2_000)).unwrap_or(2_000);
        let delay =
            Duration::from_millis(base_delay.saturating_add(retry_jitter(&job.key, attempt)));
        tokio::select! {
            () = tokio::time::sleep(delay) => {}
            () = cancellation.cancelled() => return Err(Error::Cancelled),
        }
    }
}

fn is_permanent_status(status: u16) -> bool {
    matches!(status, 401 | 403 | 404 | 410)
}

fn retry_jitter(key: &str, attempt: i32) -> u64 {
    let hash = key.bytes().fold(
        0xcbf2_9ce4_8422_2325_u64 ^ u64::try_from(attempt).unwrap_or_default(),
        |hash, byte| hash.wrapping_mul(0x0000_0100_0000_01b3) ^ u64::from(byte),
    );
    hash % 200
}

fn validate_http_uri(uri: &Uri) -> Result<String, Error> {
    let parsed = reqwest::Url::parse(&uri.0).map_err(|_| Error::Argument)?;
    if !matches!(parsed.scheme(), "http" | "https") || parsed.host().is_none() {
        return Err(Error::Argument);
    }
    Ok(parsed.to_string())
}