use crate::metastore::{
    BaseRocksTable, IdRow, IndexId, RocksEntity, RocksSecondaryIndex, RocksTable, TableId,
    TableInfo,
};
use crate::table::{Row, TableValue};
use crate::{base_rocks_secondary_index, rocks_table_new, CubeError};
use chrono::serde::{ts_seconds, ts_seconds_option};
use chrono::{DateTime, Duration, Utc};
use cuberockstore::rocksdb::WriteBatch;
use std::cmp::Ordering;
use std::sync::Arc;

use crate::cachestore::QueueKey;
use serde::{Deserialize, Deserializer, Serialize};

// It's the x-process-id header, a uuid (36 chars), bounded on ingress in src/http
pub const QUEUE_ITEM_PROCESS_ID_MAX_LEN: usize = 64;
// We use ${uuidv4()}, it's 36, let's limit to 48
pub const QUEUE_ITEM_EXTERNAL_ID_MAX_LEN: usize = 48;

fn merge(a: serde_json::Value, b: serde_json::Value) -> Option<serde_json::Value> {
    match (a, b) {
        (mut root @ serde_json::Value::Object(_), serde_json::Value::Object(b)) => {
            let r = root.as_object_mut().unwrap();
            for (k, v) in b {
                if r.contains_key(&k) {
                    r.remove(&k);
                }

                r.insert(k, v);
            }

            Some(root)
        }
        // Special case to truncate extra
        (_a, serde_json::Value::Null) => None,
        (_a, b) => Some(b),
    }
}

#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq, Hash)]
pub enum QueueResultAckEventResult {
    Empty,
    WithResult {
        // It can be a large string, 20 mb
        // Arc is used as temporarily solution to protect cloning on receiving from broadcast::channel
        // TODO(ovr): Rewrite Queue without holding queue result in channel
        result: Arc<String>,
    },
}

#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq, Hash)]
pub struct QueueResultAckEvent {
    pub id: u64,
    pub path: String,
    pub result: QueueResultAckEventResult,
}

#[repr(u8)]
#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq, Hash)]
pub enum QueueItemStatus {
    Pending = 0,
    Active = 1,
    Finished = 2,
}

impl ToString for QueueItemStatus {
    fn to_string(&self) -> String {
        match self {
            QueueItemStatus::Pending => "pending".to_string(),
            QueueItemStatus::Active => "active".to_string(),
            QueueItemStatus::Finished => "finished".to_string(),
        }
    }
}

#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq)]
pub struct QueueItem {
    prefix: Option<String>,
    pub(crate) key: String,
    pub(crate) extra: Option<String>,
    #[serde(default = "QueueItem::status_default")]
    pub(crate) status: QueueItemStatus,
    #[serde(default)]
    priority: i64,
    #[serde(with = "ts_seconds")]
    created: DateTime<Utc>,
    #[serde(with = "ts_seconds_option")]
    pub(crate) heartbeat: Option<DateTime<Utc>>,
    #[serde(with = "ts_seconds_option")]
    orphaned: Option<DateTime<Utc>>,
    #[serde(with = "ts_seconds")]
    expire: DateTime<Utc>,
    #[serde(default)]
    pub(crate) process_id: Option<String>,
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub(crate) exclusive: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) external_id: Option<String>,
}

impl RocksEntity for QueueItem {
    fn version() -> u32 {
        5
    }
}

impl Ord for QueueItem {
    fn cmp(&self, other: &Self) -> Ordering {
        if self.priority == other.priority {
            other.created.cmp(&self.created)
        } else {
            self.priority.cmp(&other.priority)
        }
    }
}

impl PartialOrd for QueueItem {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl QueueItem {
    pub fn extract_prefix(path: String) -> Option<String> {
        let parts: Vec<&str> = path.rsplitn(2, ":").collect();

        match parts.len() {
            2 => Some(parts[1].to_string()),
            _ => None,
        }
    }

    pub fn parse_path(path: String) -> (Option<String>, String) {
        let parts: Vec<&str> = path.rsplitn(2, ":").collect();

        match parts.len() {
            2 => (Some(parts[1].to_string()), parts[0].to_string()),
            _ => (None, path),
        }
    }

    pub fn new(
        path: String,
        status: QueueItemStatus,
        priority: i64,
        orphaned: Option<u32>,
        process_id: Option<String>,
        exclusive: bool,
        external_id: Option<String>,
    ) -> Self {
        let (prefix, key) = QueueItem::parse_path(path);
        let created = Utc::now();

        QueueItem {
            prefix,
            key,
            status,
            priority,
            extra: None,
            heartbeat: None,
            orphaned: if let Some(orphaned) = orphaned {
                Some(created + Duration::seconds(orphaned as i64))
            } else {
                None
            },
            expire: if let Some(orphaned) = orphaned {
                created + Duration::seconds(orphaned as i64) + Duration::hours(2)
            } else {
                created.clone() + Duration::hours(4)
            },
            created,
            process_id,
            exclusive,
            external_id,
        }
    }

    pub fn queue_to_cancel_row(item: IdRow<QueueItem>) -> Row {
        let row_id = item.get_id();
        let row = item.into_row();

        Row::new(vec![
            TableValue::String(row.key),
            TableValue::String(row_id.to_string()),
        ])
    }

    pub fn get_key(&self) -> &String {
        &self.key
    }

    pub fn get_prefix(&self) -> &Option<String> {
        &self.prefix
    }

    pub fn get_path(&self) -> String {
        if let Some(prefix) = &self.prefix {
            format!("{}:{}", prefix, self.key)
        } else {
            self.key.clone()
        }
    }

    pub fn get_priority(&self) -> &i64 {
        &self.priority
    }

    pub fn get_extra(&self) -> &Option<String> {
        &self.extra
    }

    pub fn get_status(&self) -> &QueueItemStatus {
        &self.status
    }

    pub fn get_heartbeat(&self) -> &Option<DateTime<Utc>> {
        &self.heartbeat
    }

    pub fn get_created(&self) -> &DateTime<Utc> {
        &self.created
    }

    pub fn get_orphaned(&self) -> &Option<DateTime<Utc>> {
        &self.orphaned
    }

    pub fn get_expire(&self) -> &DateTime<Utc> {
        &self.expire
    }

    pub fn get_process_id(&self) -> &Option<String> {
        &self.process_id
    }

    pub fn get_exclusive(&self) -> bool {
        self.exclusive
    }

    pub fn get_external_id(&self) -> &Option<String> {
        &self.external_id
    }

    /// Returns whether this item should be visible to the given caller process.
    /// Exclusive items with a process_id are only visible to the owning process.
    pub fn is_visible_for(&self, caller_process_id: &Option<String>) -> bool {
        if self.exclusive {
            match (&self.process_id, caller_process_id) {
                (Some(item_pid), Some(caller_pid)) => item_pid == caller_pid,
                (Some(_), None) => false,
                _ => true,
            }
        } else {
            true
        }
    }

    pub fn status_default() -> QueueItemStatus {
        QueueItemStatus::Pending
    }

    pub fn update_heartbeat(&mut self) {
        self.heartbeat = Some(Utc::now());
    }

    pub fn merge_extra(&self, payload: String) -> Result<Self, CubeError> {
        let mut new = self.clone();

        if let Some(extra) = &self.extra {
            let prev = serde_json::from_str(&extra)?;
            let next = serde_json::from_str(&payload)?;

            let extra = merge(prev, next);

            new.extra = extra.map(|v| v.to_string())
        } else {
            new.extra = Some(payload);
        }

        Ok(new)
    }
}

#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq)]
pub enum QueueRetrieveResponse {
    Success {
        id: u64,
        item: QueueItem,
        payload: String,
        pending: u64,
        active: Vec<String>,
    },
    LockFailed {
        pending: u64,
        active: Vec<String>,
    },
    NotEnoughConcurrency {
        pending: u64,
        active: Vec<String>,
    },
    NotFound {
        pending: u64,
        active: Vec<String>,
    },
    ExclusiveAccessFailed {
        pending: u64,
        active: Vec<String>,
    },
}

/// Wire format of the `active` column, shared by all commands which report it
pub fn active_keys_to_value(active: Vec<String>) -> TableValue {
    if active.is_empty() {
        TableValue::Null
    } else {
        TableValue::String(active.join(","))
    }
}

impl QueueRetrieveResponse {
    pub fn into_queue_retrieve_rows(self, extended: bool) -> Vec<Row> {
        match self {
            QueueRetrieveResponse::Success {
                id,
                item,
                payload,
                pending,
                active,
            } => vec![Row::new(vec![
                TableValue::String(payload),
                if let Some(extra) = item.extra {
                    TableValue::String(extra)
                } else {
                    TableValue::Null
                },
                TableValue::Int(pending as i64),
                active_keys_to_value(active),
                TableValue::String(id.to_string()),
            ])],
            QueueRetrieveResponse::LockFailed { pending, active }
            | QueueRetrieveResponse::NotEnoughConcurrency { pending, active }
            | QueueRetrieveResponse::NotFound { pending, active }
            | QueueRetrieveResponse::ExclusiveAccessFailed { pending, active } => {
                if extended {
                    vec![Row::new(vec![
                        TableValue::Null,
                        TableValue::Null,
                        TableValue::Int(pending as i64),
                        active_keys_to_value(active),
                        TableValue::Null,
                    ])]
                } else {
                    vec![]
                }
            }
        }
    }
}

#[derive(Clone, Copy, Debug)]
pub(crate) enum QueueItemRocksIndex {
    ByPath = 1,
    ByPrefixAndStatus = 2,
    ByPrefix = 3,
    ByPathAndExternalId = 4,
}

pub struct QueueItemRocksTable<'a> {
    db: crate::metastore::DbTableRef<'a>,
}

impl<'a> QueueItemRocksTable<'a> {
    pub fn new(db: crate::metastore::DbTableRef<'a>) -> Self {
        Self { db }
    }

    pub fn get_row_by_key(&self, key: QueueKey) -> Result<Option<IdRow<QueueItem>>, CubeError> {
        match key {
            QueueKey::ByPath(path) => {
                let index_key = QueueItemIndexKey::ByPath(path.clone());
                self.get_single_opt_row_by_index(&index_key, &QueueItemRocksIndex::ByPath)
            }
            QueueKey::ById(id) => self.get_row(id.clone()),
        }
    }
}

impl<'a> BaseRocksTable for QueueItemRocksTable<'a> {
    fn enable_delete_event(&self) -> bool {
        false
    }

    fn enable_update_event(&self) -> bool {
        false
    }

    fn migrate_table(
        &self,
        batch: &mut WriteBatch,
        _table_info: TableInfo,
    ) -> Result<(), CubeError> {
        self.migrate_table_by_truncate(batch)
    }
}

rocks_table_new!(QueueItem, QueueItemRocksTable, TableId::QueueItems, {
    vec![
        Box::new(QueueItemRocksIndex::ByPath),
        Box::new(QueueItemRocksIndex::ByPrefixAndStatus),
        Box::new(QueueItemRocksIndex::ByPrefix),
        Box::new(QueueItemRocksIndex::ByPathAndExternalId),
    ]
});

#[derive(Hash, Clone, Debug)]
pub enum QueueItemIndexKey {
    ByPath(String),
    ByPrefixAndStatus(String, QueueItemStatus),
    ByPrefix(String),
    ByPathAndExternalId(String, Option<String>),
}

base_rocks_secondary_index!(QueueItem, QueueItemRocksIndex);

impl RocksSecondaryIndex<QueueItem, QueueItemIndexKey> for QueueItemRocksIndex {
    fn typed_key_by(&self, row: &QueueItem) -> QueueItemIndexKey {
        match self {
            QueueItemRocksIndex::ByPath => QueueItemIndexKey::ByPath(row.get_path()),
            QueueItemRocksIndex::ByPrefixAndStatus => QueueItemIndexKey::ByPrefixAndStatus(
                row.get_prefix().clone().unwrap_or("".to_string()),
                row.get_status().clone(),
            ),
            QueueItemRocksIndex::ByPrefix => {
                QueueItemIndexKey::ByPrefix(row.get_prefix().clone().unwrap_or("".to_string()))
            }
            QueueItemRocksIndex::ByPathAndExternalId => QueueItemIndexKey::ByPathAndExternalId(
                row.get_path(),
                row.get_external_id().clone(),
            ),
        }
    }

    fn key_to_bytes(&self, key: &QueueItemIndexKey) -> Vec<u8> {
        match key {
            QueueItemIndexKey::ByPath(s) => s.as_bytes().to_vec(),
            QueueItemIndexKey::ByPrefix(s) => s.as_bytes().to_vec(),
            QueueItemIndexKey::ByPathAndExternalId(path, s) => {
                let mut r = Vec::new();
                r.extend_from_slice(path.as_bytes());
                r.push(0u8);
                r.extend_from_slice(s.as_deref().unwrap_or("__null__").as_bytes());
                r
            }
            QueueItemIndexKey::ByPrefixAndStatus(prefix, s) => {
                let mut r = Vec::with_capacity(prefix.len() + 1);
                r.extend_from_slice(&prefix.as_bytes());

                match s {
                    QueueItemStatus::Pending => r.push(0_u8),
                    QueueItemStatus::Active => r.push(1_u8),
                    QueueItemStatus::Finished => r.push(2_u8),
                }

                r
            }
        }
    }

    fn is_unique(&self) -> bool {
        match self {
            QueueItemRocksIndex::ByPath => true,
            QueueItemRocksIndex::ByPrefixAndStatus => false,
            QueueItemRocksIndex::ByPrefix => false,
            QueueItemRocksIndex::ByPathAndExternalId => true,
        }
    }

    fn version(&self) -> u32 {
        match self {
            QueueItemRocksIndex::ByPath => 1,
            QueueItemRocksIndex::ByPrefixAndStatus => 2,
            QueueItemRocksIndex::ByPrefix => 1,
            QueueItemRocksIndex::ByPathAndExternalId => 2,
        }
    }

    fn is_ttl(&self) -> bool {
        true
    }

    fn get_expire(&self, row: &QueueItem) -> Option<DateTime<Utc>> {
        Some(row.expire.clone())
    }

    fn get_id(&self) -> IndexId {
        *self as IndexId
    }

    fn should_index_row(&self, row: &QueueItem) -> bool {
        match self {
            Self::ByPathAndExternalId => row.external_id.is_some(),
            _ => true,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::CubeError;
    use itertools::Itertools;

    #[test]
    fn test_queue_item_sort() -> Result<(), CubeError> {
        let priority0_1 = QueueItem::new(
            "1".to_string(),
            QueueItemStatus::Active,
            0,
            None,
            None,
            false,
            None,
        );
        let mut priority0_2 = QueueItem::new(
            "2".to_string(),
            QueueItemStatus::Active,
            0,
            None,
            None,
            false,
            None,
        );
        let mut priority0_3 = QueueItem::new(
            "3".to_string(),
            QueueItemStatus::Active,
            0,
            None,
            None,
            false,
            None,
        );
        let mut priority10_4 = QueueItem::new(
            "4".to_string(),
            QueueItemStatus::Active,
            10,
            None,
            None,
            false,
            None,
        );
        let mut priority0_5 = QueueItem::new(
            "5".to_string(),
            QueueItemStatus::Active,
            0,
            None,
            None,
            false,
            None,
        );
        let mut priority_n5_6 = QueueItem::new(
            "6".to_string(),
            QueueItemStatus::Active,
            -5,
            None,
            None,
            false,
            None,
        );

        // Force timestamps to be distinct (on systems that are too fast or have low clock resolution)
        for (i, item) in (1..).zip([
            &mut priority0_2,
            &mut priority0_3,
            &mut priority10_4,
            &mut priority0_5,
            &mut priority_n5_6,
        ]) {
            item.created = priority0_1.created + Duration::milliseconds(i);
        }

        assert_eq!(
            vec![
                priority0_1.clone(),
                priority0_2.clone(),
                priority0_3.clone(),
                priority10_4.clone(),
                priority_n5_6.clone(),
                priority0_5.clone()
            ]
            .into_iter()
            .sorted_by(|a, b| b.cmp(&a))
            .map(|item| item.get_key().clone())
            .collect::<Vec<String>>(),
            vec![
                "4".to_string(),
                "1".to_string(),
                "2".to_string(),
                "3".to_string(),
                "5".to_string(),
                "6".to_string()
            ]
        );

        assert_eq!(
            vec![
                priority10_4,
                priority0_1,
                priority0_5,
                priority0_2,
                priority0_3,
                priority_n5_6
            ]
            .into_iter()
            .sorted_by(|a, b| b.cmp(&a))
            .map(|item| item.get_key().clone())
            .collect::<Vec<String>>(),
            vec![
                "4".to_string(),
                "1".to_string(),
                "2".to_string(),
                "3".to_string(),
                "5".to_string(),
                "6".to_string()
            ],
        );

        Ok(())
    }

    #[test]
    fn test_is_visible_for() {
        // Non-exclusive item
        let non_exclusive = QueueItem::new(
            "prefix:key".to_string(),
            QueueItemStatus::Pending,
            0,
            None,
            Some("pid-1".to_string()),
            false,
            None,
        );
        assert!(non_exclusive.is_visible_for(&None));
        assert!(non_exclusive.is_visible_for(&Some("pid-1".to_string())));
        assert!(non_exclusive.is_visible_for(&Some("pid-other".to_string())));

        // Exclusive item with process_id
        let exclusive = QueueItem::new(
            "prefix:key".to_string(),
            QueueItemStatus::Pending,
            0,
            None,
            Some("pid-1".to_string()),
            true,
            None,
        );
        assert!(exclusive.is_visible_for(&Some("pid-1".to_string())));
        assert!(!exclusive.is_visible_for(&Some("pid-other".to_string())));
        assert!(!exclusive.is_visible_for(&None));
    }
}
