// Copyright (C) 2024 Huawei Device Co., Ltd.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Callback handling for download operations.
//!
//! This module provides a prime callback implementation that handles download
//! events and communicates with cache storage, manages download state, and
//! notifies registered callbacks about download progress, success, failure, and
//! cancellation.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use cache_core::{CacheManager, Updater};
use netstack_rs::info::DownloadInfo;
use request_utils::task_id::TaskId;

use super::common::{CommonError, CommonResponse};
use super::{CacheDownloadError, RUNNING};
use crate::download::{CANCEL, FAIL, SUCCESS};
use crate::info::RustDownloadInfo;
use crate::services::{CacheDownloadService, PreloadCallback};

/// Interval for reporting progress updates.
const PROGRESS_INTERVAL: usize = 8;

/// Configuration options for download task retry and timeout behavior.
///
/// Groups related timeout and retry settings into a single structure
/// for cleaner function signatures.
pub(crate) struct TaskConfig {
    /// Maximum retry count (task override or global setting)
    pub(crate) max_retry: Option<usize>,
    /// Network check timeout in seconds (task override or global setting)
    pub(crate) network_check_timeout: Option<u32>,
    /// HTTP total timeout in seconds (task override or global setting)
    pub(crate) http_total_timeout: Option<u32>,
}

impl TaskConfig {
    /// Creates a new TaskConfig with the specified values.
    pub(crate) fn new(
        max_retry: Option<usize>,
        network_check_timeout: Option<u32>,
        http_total_timeout: Option<u32>,
    ) -> Self {
        Self {
            max_retry,
            network_check_timeout,
            http_total_timeout,
        }
    }
}

/// Primary callback handler for managing download operations and notifications.
pub(crate) struct PrimeCallback {
    /// Unique identifier for the download task
    task_id: TaskId,
    /// Flag indicating whether the download has finished
    finish: Arc<AtomicBool>,
    /// Current state of the download (running, success, fail, cancel)
    state: Arc<AtomicUsize>,
    /// Handle for updating cache storage with downloaded data
    cache_handle: Updater,
    /// Queue of user-registered callbacks to notify about download events
    callbacks: Arc<Mutex<VecDeque<Box<dyn PreloadCallback>>>>,
    /// Restricts how frequently progress updates are reported
    progress_restriction: ProgressRestriction,
    /// Sequence number for task ordering
    seq: usize,
    /// Task configuration (retry and timeout settings)
    config: TaskConfig,
}

/// Restricts the frequency of progress updates.
struct ProgressRestriction {
    /// Last reported downloaded byte position.
    processed: u64,
    /// Counter used to throttle progress notifications.
    count: usize,
    /// Whether any data has been received yet for this task.
    data_receive: bool,
}

impl ProgressRestriction {
    /// Creates a new progress restriction with zeroed counters.
    fn new() -> Self {
        Self {
            processed: 0,
            count: 0,
            data_receive: false,
        }
    }
}

impl PrimeCallback {
    /// Creates a new prime callback for a download task.
    pub(crate) fn new(
        task_id: TaskId,
        cache_manager: &'static CacheManager,
        finish: Arc<AtomicBool>,
        state: Arc<AtomicUsize>,
        callbacks: Arc<Mutex<VecDeque<Box<dyn PreloadCallback>>>>,
        seq: usize,
        config: TaskConfig,
    ) -> Self {
        Self {
            task_id: task_id.clone(),
            finish,
            state,
            cache_handle: Updater::new(task_id, cache_manager),
            callbacks,
            progress_restriction: ProgressRestriction::new(),
            seq,
            config,
        }
    }

    /// Marks the task state as running.
    pub(crate) fn set_running(&self) {
        self.state.store(RUNNING, Ordering::Release);
    }

    /// Returns a clone of the task identifier.
    pub(crate) fn task_id(&self) -> TaskId {
        self.task_id.clone()
    }

    /// Returns the configured maximum retry count, if any.
    pub(crate) fn max_retry(&self) -> Option<usize> {
        self.config.max_retry
    }

    /// Returns the configured network check timeout in seconds, if any.
    pub(crate) fn network_check_timeout(&self) -> Option<u32> {
        self.config.network_check_timeout
    }

    /// Returns the configured HTTP total timeout in seconds, if any.
    pub(crate) fn http_total_timeout(&self) -> Option<u32> {
        self.config.http_total_timeout
    }
}

impl PrimeCallback {
    /// Handles successful download completion.
    ///
    /// Updates the cache, changes the download state to success, and notifies
    /// all registered callbacks of the successful completion. Reports 100%
    /// progress before calling each callback's success method.
    ///
    /// # Type Parameters
    /// * `R` - Type implementing `CommonResponse` containing the HTTP status
    ///   code
    ///
    /// # Arguments
    /// * `response` - Response object containing the status code
    pub(crate) fn common_success<R>(&mut self, response: R)
    where
        R: CommonResponse,
    {
        let code = response.code();
        info!("{} status {}", self.task_id.brief(), code);

        // Finalize cache storage
        let cache = self.cache_handle.cache_finish();
        // Update task state to success
        self.state.store(SUCCESS, Ordering::Release);
        self.finish.store(true, Ordering::Release);

        // Notify all registered callbacks
        let mut callbacks = self.callbacks.lock().unwrap();

        while let Some(mut callback) = callbacks.pop_front() {
            let clone_cache = cache.clone();
            let task_id = self.task_id.brief().to_string();
            // Spawn in separate tasks to avoid blocking
            crate::spawn(move || {
                // Report 100% progress before success
                callback.on_progress(clone_cache.size() as u64, clone_cache.size() as u64);
                callback.on_success(clone_cache, &task_id)
            });
        }

        // Explicit drop to release the mutex
        drop(callbacks);
        // Notify the service that the task has finished
        self.notify_agent_finish();
    }

    /// Handles download failure.
    ///
    /// Updates the download state to failed, and notifies all registered
    /// callbacks of the failure with the appropriate error information.
    ///
    /// # Type Parameters
    /// * `E` - Type implementing `CommonError` containing error information
    ///
    /// # Arguments
    /// * `error` - Error object containing the failure details
    pub(crate) fn common_fail<E>(&mut self, error: E, info: DownloadInfo)
    where
        E: CommonError,
    {
        info!("{} download failed {}", self.task_id.brief(), error.code());
        // Update task state to failed
        self.state.store(FAIL, Ordering::Release);
        self.finish.store(true, Ordering::Release);

        // Notify all registered callbacks
        let mut callbacks = self.callbacks.lock().unwrap();

        while let Some(mut callback) = callbacks.pop_front() {
            let task_id = self.task_id.brief().to_string();
            // Convert to the standard cache download error type
            let error = CacheDownloadError::from(&error);
            let info = RustDownloadInfo::from_download_info(info.clone());
            // Spawn in separate tasks to avoid blocking
            crate::spawn(move || callback.on_fail(error, info, &task_id));
        }

        // Explicit drop to release the mutex
        drop(callbacks);
        // Notify the service that the task has finished
        self.notify_agent_finish();
    }

    /// Handles download cancellation.
    ///
    /// Updates the download state to canceled, and notifies all registered
    /// callbacks of the cancellation.
    pub(crate) fn common_cancel(&mut self) {
        info!("{} is cancel", self.task_id.brief());
        // Update task state to canceled
        self.state.store(CANCEL, Ordering::Release);
        self.finish.store(true, Ordering::Release);

        // Notify all registered callbacks
        let mut callbacks = self.callbacks.lock().unwrap();

        while let Some(mut callback) = callbacks.pop_front() {
            // Spawn in separate tasks to avoid blocking
            crate::spawn(move || callback.on_cancel());
        }

        // Explicit drop to release the mutex
        drop(callbacks);
        // Notify the service that the task has finished
        self.notify_agent_finish();
    }

    /// Reports download progress to registered callbacks.
    ///
    /// Implements throttling to prevent excessive progress notifications, only
    /// reporting progress when the download has advanced and at a limited
    /// frequency.
    ///
    /// # Arguments
    /// * `dl_total` - Total number of bytes to download
    /// * `dl_now` - Current number of bytes downloaded
    /// * `_ul_total` - Total number of bytes to upload (unused in downloads)
    /// * `_ul_now` - Current number of bytes uploaded (unused in downloads)
    pub(crate) fn common_progress(
        &mut self,
        dl_total: u64,
        dl_now: u64,
        _ul_total: u64,
        _ul_now: u64,
    ) {
        // Skip if no data has been received yet, or if no progress has been made,
        // or if the download is complete (which will be handled by common_success)
        if !self.progress_restriction.data_receive
            || dl_now == self.progress_restriction.processed
            || dl_now == dl_total
        {
            return;
        }

        // Update the last processed position
        self.progress_restriction.processed = dl_now;

        // Implement throttling using a counter
        let count = self.progress_restriction.count;
        self.progress_restriction.count += 1;
        if count % PROGRESS_INTERVAL != 0 {
            return;
        }

        // Reset counter for next interval
        self.progress_restriction.count = 1;

        // Notify all registered callbacks of progress
        let mut callbacks = self.callbacks.lock().unwrap();
        for callback in callbacks.iter_mut() {
            callback.on_progress(dl_now, dl_total);
        }
    }

    /// Processes received data and updates the cache.
    ///
    /// Marks that data reception has started and forwards the data to the cache
    /// handler.
    ///
    /// # Type Parameters
    /// * `F` - Function type that returns the content length when called
    ///
    /// # Arguments
    /// * `data` - Buffer containing the received data
    /// * `content_length` - Function that returns the total content length if
    ///   available
    pub(crate) fn common_data_receive<F>(&mut self, data: &[u8], content_length: F)
    where
        F: FnOnce() -> Option<usize>,
    {
        // Mark that data reception has started
        self.progress_restriction.data_receive = true;
        // Forward data to cache storage
        self.cache_handle.cache_receive(data, content_length);
    }

    /// Restarts the download by resetting the cache.
    ///
    /// # Notes
    /// Only available when the `netstack` feature is enabled.
    #[cfg(feature = "netstack")]
    pub(crate) fn common_restart(&mut self) {
        self.cache_handle.reset_cache();
    }

    /// Notifies the cache download service that the task has finished.
    ///
    /// Used to trigger any necessary cleanup or notification operations in the
    /// service.
    fn notify_agent_finish(&self) {
        CacheDownloadService::get_instance().task_finish(&self.task_id, self.seq);
    }
}