// 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.

//! Notification publishing and dispatching module for download/upload tasks.
//!
//! This module provides functionality for dispatching various types of
//! notifications related to download and upload tasks, including progress
//! updates, success/failure notifications, and group management. It uses a
//! singleton pattern to maintain a central notification dispatcher.

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, LazyLock, Mutex};

use ylong_runtime::fastrand::fast_random;
use ylong_runtime::sync::mpsc::{self, unbounded_channel};

use super::database::NotificationDb;
use super::notify_flow::{EventualNotify, NotifyFlow, NotifyInfo, ProgressNotify};
use super::task_handle::{cancel_notification, NotificationCheck};
use crate::info::TaskInfo;
use crate::service::notification_bar::NotificationConfig;
use crate::task::request_task::RequestTask;
use crate::utils::get_current_duration;

/// Interval in milliseconds between progress notifications (500 ms).
pub(crate) const NOTIFY_PROGRESS_INTERVAL: u64 = 500;

/// Central dispatcher for managing and publishing task notifications.
///
/// This struct serves as a singleton and provides functionality to register
/// tasks, dispatch notifications, and manage notification groups for
/// download/upload tasks.
pub struct NotificationDispatcher {
    /// Database for storing notification configuration and task information.
    database: Arc<NotificationDb>,
    /// Map of task IDs to notification visibility flags (gauge).
    task_gauge: Mutex<HashMap<u32, Arc<AtomicBool>>>,
    /// Channel for sending notification information to the notification flow.
    flow: mpsc::UnboundedSender<NotifyInfo>,
}

impl NotificationDispatcher {
    /// Creates a new NotificationDispatcher instance.
    ///
    /// Initializes the database, creates a notification channel, and starts the
    /// notification flow for processing notifications asynchronously.
    fn new() -> Self {
        // Create notification database
        let database = Arc::new(NotificationDb::new());
        // Set up channel for notification messages
        let (tx, rx) = unbounded_channel();
        // Start notification flow processor
        NotifyFlow::new(rx, database.clone()).run();

        Self {
            database: database.clone(),
            task_gauge: Mutex::new(HashMap::new()),
            flow: tx,
        }
    }

    /// Returns the singleton instance of NotificationDispatcher.
    ///
    /// Uses LazyLock for thread-safe initialization of the singleton instance.
    ///
    /// # Returns
    ///
    /// Static reference to the singleton NotificationDispatcher instance
    pub(crate) fn get_instance() -> &'static Self {
        static INSTANCE: LazyLock<NotificationDispatcher> =
            LazyLock::new(NotificationDispatcher::new);
        &INSTANCE
    }

    /// Clears notification information for a specific task.
    ///
    /// # Arguments
    ///
    /// * `task_id` - ID of the task whose notification information should be
    ///   cleared
    pub(crate) fn clear_task_info(&self, task_id: u32) {
        self.database.clear_task_info(task_id);
    }

    /// Clears group notification information that is older than a week.
    ///
    /// This method helps maintain database size by removing outdated group
    /// information automatically.
    pub(crate) fn clear_group_info(&self) {
        self.database.clear_group_info_a_week_ago();
    }

    /// Disables notifications for a specific task.
    ///
    /// # Arguments
    ///
    /// * `uid` - User ID associated with the task
    /// * `task_id` - ID of the task to disable notifications for
    pub(crate) fn disable_task_notification(&self, uid: u64, task_id: u32) {
        self.database.disable_task_notification(task_id);
        self.unregister_task(uid, task_id, true);
    }

    /// Enables progress notifications for a specific task.
    ///
    /// Sets the notification gauge flag to true if the task is registered.
    ///
    /// # Arguments
    ///
    /// * `task_id` - ID of the task to enable progress notifications for
    pub(crate) fn enable_task_progress_notification(&self, task_id: u32) {
        if let Some(gauge) = self.task_gauge.lock().unwrap().get(&task_id) {
            gauge.store(true, Ordering::Release);
        }
    }

    /// Updates customized notification configuration for a task.
    ///
    /// # Arguments
    ///
    /// * `config` - New notification configuration to apply
    pub(crate) fn update_task_customized_notification(&self, config: &NotificationConfig) {
        self.database.update_task_customized_notification(config);
    }

    /// Checks if notifications are available for a specific task.
    ///
    /// # Arguments
    ///
    /// * `task_id` - ID of the task to check
    ///
    /// # Returns
    ///
    /// Boolean indicating whether notifications are available for the task
    pub(crate) fn check_task_notification_available(&self, task_id: u32) -> bool {
        self.database.check_task_notification_available(&task_id)
    }

    /// Gets the notification visibility flag (gauge) for a specific task.
    ///
    /// # Arguments
    ///
    /// * `task_id` - ID of the task to get the gauge for
    ///
    /// # Returns
    ///
    /// `Some(bool)` with the gauge value if the task exists, `None` otherwise
    pub(crate) fn get_task_gauge(&self, task_id: u32) -> Option<bool> {
        self.task_gauge
            .lock()
            .ok()?
            .get(&task_id)
            .map(|gauge| gauge.load(Ordering::Acquire))
    }

    /// Registers a task for notification tracking.
    ///
    /// Creates a notification visibility gauge for the task based on its group
    /// configuration or individual settings.
    ///
    /// # Arguments
    ///
    /// * `task` - Reference to the task to register for notifications
    ///
    /// # Returns
    ///
    /// Arc<AtomicBool> representing the notification visibility gauge for the
    /// task
    pub(crate) fn register_task(&self, task: &RequestTask) -> Arc<AtomicBool> {
        // Determine gauge value based on group membership or individual task settings
        let gauge = if let Some(gid) = self.database.query_task_gid(task.task_id()) {
            if self.database.check_group_notification_available(&gid) {
                Arc::new(AtomicBool::new(true))
            } else {
                Arc::new(AtomicBool::new(false))
            }
        } else {
            let gauge = task.notification_check(&self.database);
            Arc::new(AtomicBool::new(gauge))
        };

        // Store gauge in the task map
        self.task_gauge
            .lock()
            .unwrap()
            .insert(task.task_id(), gauge.clone());
        gauge
    }

    /// Unregisters a task from notification tracking.
    ///
    /// Stops notifications for the specified task and optionally affects group
    /// notifications.
    ///
    /// # Arguments
    ///
    /// * `uid` - User ID associated with the task
    /// * `task_id` - ID of the task to unregister
    /// * `affect_group` - Whether to update the task's group notification state
    pub(crate) fn unregister_task(&self, uid: u64, task_id: u32, affect_group: bool) {
        match (
            self.task_gauge.lock().unwrap().get(&task_id).cloned(),
            self.database.query_task_gid(task_id),
        ) {
            (Some(gauge), Some(gid)) => {
                if affect_group {
                    gauge.store(false, Ordering::Release);
                    let _ = self.flow.send(NotifyInfo::Unregister(uid, task_id, gid));
                }
            }
            (None, Some(gid)) => {
                if affect_group {
                    let _ = self.flow.send(NotifyInfo::Unregister(uid, task_id, gid));
                }
            }
            (Some(gauge), None) => {
                gauge.store(false, Ordering::Release);
                cancel_notification(task_id);
            }
            (None, None) => {}
        }
    }

    /// Publishes a progress notification for a task.
    ///
    /// Creates and sends a progress notification with current task status
    /// information. Handles multi-file uploads and calculates total size
    /// when available.
    ///
    /// # Arguments
    ///
    /// * `task` - Reference to the task whose progress should be published
    pub(crate) fn publish_progress_notification(&self, task: &RequestTask) {
        let progress = task.progress.lock().unwrap();
        let mut total = Some(0);

        // Calculate total size if all sizes are non-negative
        for size in progress.sizes.iter() {
            if *size < 0 {
                total = None; // Unknown total size
                break;
            }
            *total.as_mut().unwrap() += *size as u64;
        }

        // Determine if this is a multi-file upload
        let multi_upload = match progress.sizes.len() {
            0 | 1 => None,
            len => Some((progress.common_data.index, len)),
        };

        // Create progress notification
        let notify = ProgressNotify {
            action: task.action(),
            task_id: task.task_id(),
            uid: task.uid(),
            file_name: match task.conf.file_specs.first() {
                Some(spec) => spec.file_name.clone(),
                None => {
                    error!("Failed to get the first file_spec from an empty vector in TaskConfig");
                    String::new()
                }
            },
            processed: progress.common_data.total_processed as u64,
            total,
            multi_upload,
            version: task.conf.version,
        };

        // Send notification through the channel
        let _ = self.flow.send(NotifyInfo::Progress(notify));
    }

    /// Publishes a success notification when a task completes successfully.
    ///
    /// Removes the task from notification tracking and sends a success
    /// notification if notifications are available for this task.
    ///
    /// # Arguments
    ///
    /// * `info` - Reference to the completed task information
    pub(crate) fn publish_success_notification(&self, info: &TaskInfo) {
        // Remove task from gauge map as it's completed
        self.task_gauge
            .lock()
            .unwrap()
            .remove(&info.common_data.task_id);

        // Only send notification if task notifications are available
        if !info.notification_check(&self.database) {
            return;
        }

        // Create success notification
        let notify = EventualNotify {
            action: info.action(),
            task_id: info.common_data.task_id,
            processed: info.progress.common_data.total_processed as u64,
            uid: info.uid(),
            file_name: match info.file_specs.first() {
                Some(spec) => spec.file_name.clone(),
                None => {
                    error!("Failed to get the first file_spec from an empty vector in TaskInfo");
                    String::new()
                }
            },
            is_successful: true,
        };

        // Send notification through the channel
        let _ = self.flow.send(NotifyInfo::Eventual(notify));
    }

    /// Publishes a failure notification when a task fails to complete.
    ///
    /// Removes the task from notification tracking and sends a failure
    /// notification if notifications are available for this task.
    ///
    /// # Arguments
    ///
    /// * `info` - Reference to the failed task information
    pub(crate) fn publish_failed_notification(&self, info: &TaskInfo) {
        // Remove task from gauge map as it's completed (failed)
        self.task_gauge
            .lock()
            .unwrap()
            .remove(&info.common_data.task_id);

        // Only send notification if task notifications are available
        if !info.notification_check(&self.database) {
            return;
        }

        // Create failure notification
        let notify = EventualNotify {
            action: info.action(),
            task_id: info.common_data.task_id,
            processed: info.progress.common_data.total_processed as u64,
            uid: info.uid(),
            file_name: match info.file_specs.first() {
                Some(spec) => spec.file_name.clone(),
                None => {
                    error!("Failed to get the first file_spec from an empty vector in TaskInfo");
                    String::new()
                }
            },
            is_successful: false,
        };

        // Send notification through the channel
        let _ = self.flow.send(NotifyInfo::Eventual(notify));
    }

    /// Attaches multiple tasks to a notification group.
    ///
    /// Updates the notification gauge for all specified tasks to match the
    /// group's visibility setting and sends a notification to update the
    /// group.
    ///
    /// # Arguments
    ///
    /// * `task_ids` - List of task IDs to attach to the group
    /// * `group_id` - ID of the target group
    /// * `uid` - User ID associated with the tasks
    ///
    /// # Returns
    ///
    /// Boolean indicating whether the operation was successful
    pub(crate) fn attach_group(&self, task_ids: Vec<u32>, group_id: u32, uid: u64) -> bool {
        // Verify the caller owns this group; prevents cross-UID binding of one's
        // own task to another group to reuse its customized notification content
        // and visibility configuration.
        if !self.check_group_owner(group_id, uid) {
            return false;
        }
        if !self.database.attach_able(group_id) {
            return false;
        }
        info!("Attach task {:?} to group {}", task_ids, group_id);
        let is_gauge = self.database.is_gauge(group_id);

        // Update each task's group and gauge setting
        for task_id in task_ids.iter().copied() {
            self.database.update_task_group(task_id, group_id);
            if let Some(gauge) = self.task_gauge.lock().unwrap().get(&task_id) {
                gauge.store(is_gauge, std::sync::atomic::Ordering::Release);
            }
        }

        // Only send notification if group notifications are available
        if !self.database.check_group_notification_available(&group_id) {
            return true;
        }

        let _ = self
            .flow
            .send(NotifyInfo::AttachGroup(group_id, uid, task_ids));
        true
    }

    /// Deletes a notification group and disables further attachments.
    ///
    /// Marks the group as disabled and sends a notification to update the UI.
    ///
    /// # Arguments
    ///
    /// * `group_id` - ID of the group to delete
    /// * `uid` - User ID associated with the group
    ///
    /// # Returns
    ///
    /// Boolean indicating whether the operation was successful
    pub(crate) fn delete_group(&self, group_id: u32, uid: u64) -> bool {
        info!("Delete group {}", group_id);
        // Only the group's creator may delete it; reject cross-UID deletion.
        if !self.check_group_owner(group_id, uid) {
            return false;
        }
        // Check if group exists and can be deleted
        if !self.database.attach_able(group_id) {
            return false;
        }

        // Disable the group to prevent further attachments
        self.database.disable_attach_group(group_id);

        // Only send notification if group notifications are available
        if !self.database.check_group_notification_available(&group_id) {
            return true;
        }

        // Send group deletion notification
        let notify = NotifyInfo::GroupEventual(group_id, uid);
        let _ = self.flow.send(notify);
        true
    }

    /// Checks whether the given UID owns the specified notification group.
    ///
    /// Ownership is established at group creation time and stored in the
    /// `group_notification_config.uid` column. Three cases are handled:
    /// - Group exists with a recorded owner: the owner must match `uid`.
    /// - Group exists but `uid` is NULL (legacy rows created before owner
    ///   tracking was introduced): the check is relaxed to `true` to keep
    ///   upgrade data usable. Only legacy rows are relaxed; newly created
    ///   groups are always owner-protected.
    /// - Group does not exist, or the database query fails: `false`.
    ///
    /// # Arguments
    ///
    /// * `group_id` - The ID of the group to check
    /// * `uid` - The UID to verify against the group's recorded owner
    ///
    /// # Returns
    ///
    /// `true` if the group exists and is either owned by `uid` or is a legacy
    /// owner-less row; `false` otherwise.
    pub(crate) fn check_group_owner(&self, group_id: u32, uid: u64) -> bool {
        match self.database.query_group_owner(group_id) {
            // New data: strict owner match.
            Some(owner) => owner == uid,
            // Legacy owner-less row (uid NULL) is relaxed to keep upgrade data
            // usable; a genuinely absent group fails the contains_group check.
            None => self.database.contains_group(group_id),
        }
    }

    /// Creates a new notification group with the specified configuration.
    ///
    /// Generates a unique group ID, stores the group configuration in the
    /// database, and optionally sets up customized notification text.
    ///
    /// # Arguments
    ///
    /// * `gauge` - Whether to show progress gauge in notifications
    /// * `title` - Optional custom title for the group notification
    /// * `text` - Optional custom text for the group notification
    /// * `want_agent` - Optional agent identifier
    /// * `disable` - Whether to disable notifications for this group
    /// * `visibility` - Visibility level of the notifications
    /// * `uid` - The UID of the calling process, recorded as the group owner
    ///
    /// # Returns
    ///
    /// The newly created group ID
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn create_group(
        &self,
        gauge: bool,
        title: Option<String>,
        text: Option<String>,
        want_agent: Option<String>,
        disable: bool,
        visibility: u32,
        uid: u64,
    ) -> u32 {
        // Generate a unique group ID using random number generation
        let new_group_id = loop {
            let candidate = fast_random() as u32;
            if !self.database.contains_group(candidate) {
                break candidate;
            }
        };

        info!(
            "Create group {} gauge {} customized_title {:?} customized_text {:?} want_agent {:?} disable {} visibility {} uid {}",
            new_group_id, gauge, title, text, want_agent, disable, visibility, uid
        );

        // Get current time for group creation timestamp
        let current_time = get_current_duration().as_millis() as u64;

        // Store group configuration
        self.database.update_group_config(
            new_group_id,
            gauge,
            current_time,
            !disable,
            visibility,
            uid,
        );

        // Set up customized notification if provided
        if title.is_some() || text.is_some() || want_agent.is_some() {
            self.database.update_group_customized_notification(
                new_group_id,
                title,
                text,
                want_agent,
            );
        }

        new_group_id
    }
}