caesura 0.29.0

use crate::prelude::*;
use flat_db::{Hash, Table};
use futures::stream::{StreamExt, iter};
use std::fs::create_dir;

/// Queue of FLAC sources and their statuses.
///
/// Each source is represented by a [`QueueItem`] stored by 20 byte SHA-1 hash.
///
/// Items are stored and retrieved as chunks by [`Table<20, 1, QueueItem>`].
///
/// Chunks are determined by taking the first byte of the hash.
///
/// As a byte represents 256 (2^8) values you can determine the approximate
/// number of sources per chunk with `total / 256` therefore:
///   `1,000` total ≈   `4` per chunk
///   `5,000` total ≈  `20` per chunk
///  `10,000` total ≈  `39` per chunk
///  `50,000` total ≈ `195` per chunk
/// `100,000` total ≈ `390` per chunk
pub(crate) struct Queue {
    /// Path to the queue file
    table: Table<20, 1, QueueItem>,
}

#[injectable]
impl Queue {
    /// Create a new [`Queue`]
    pub(crate) fn from_path(path: PathBuf) -> Self {
        Self {
            table: Table::new(path),
        }
    }

    /// DI constructor for [`Queue`]
    #[inject]
    pub(crate) fn from_options(options: Ref<CacheOptions>) -> Self {
        let path = options.cache.join("queue");
        if !path.exists() {
            create_dir(&path)
                .expect("should be able to create queue directory if it does not exist");
        }
        Self::from_path(path)
    }

    /// Get an item from the queue
    pub(crate) async fn get(
        &self,
        hash: Hash<20>,
    ) -> Result<Option<QueueItem>, Failure<QueueAction>> {
        self.table
            .get(hash)
            .await
            .map_err(Failure::wrap(QueueAction::Get))
    }

    /// Get the keys of the items that have not been processed.
    ///
    /// Items are filtered to ensure they have:
    /// - the correct indexer
    /// - not been verified, unless `transcode_enabled` is true
    /// - not been transcoded, unless `upload_enabled` is true
    /// - not been verified OR have been and `verified` is true
    /// - not been transcoded OR have been and `success` is true
    /// - not been uploaded
    ///
    /// Items are sorted by name
    pub(crate) async fn get_unprocessed(
        &self,
        indexer: Indexer,
        transcode_enabled: bool,
        upload_enabled: bool,
        retry_failed_transcodes: bool,
    ) -> Result<Vec<Hash<20>>, Failure<QueueAction>> {
        let is_red = indexer == Indexer::Red;
        let mut items = self
            .get_unprocessed_internal(
                indexer,
                transcode_enabled,
                upload_enabled,
                retry_failed_transcodes,
            )
            .await?;
        if is_red {
            let mut pth_items = self
                .get_unprocessed_internal(
                    Indexer::Pth,
                    transcode_enabled,
                    upload_enabled,
                    retry_failed_transcodes,
                )
                .await?;
            items.append(&mut pth_items);
        }
        Ok(items)
    }

    /// Get the keys of the items that have not been processed.
    ///
    /// Items are filtered to ensure they have:
    /// - the correct indexer
    /// - not been verified, unless `transcode_enabled` is true
    /// - not been transcoded, unless `upload_enabled` is true
    /// - not been verified OR have been and `verified` is true
    /// - not been transcoded OR have been and `success` is true
    /// - not been uploaded
    ///
    /// Items are sorted by name
    async fn get_unprocessed_internal(
        &self,
        indexer: Indexer,
        transcode_enabled: bool,
        upload_enabled: bool,
        retry_failed_transcodes: bool,
    ) -> Result<Vec<Hash<20>>, Failure<QueueAction>> {
        let items = self
            .table
            .get_all()
            .await
            .map_err(Failure::wrap(QueueAction::GetAll))?;
        let mut items: Vec<&QueueItem> = items
            .values()
            .filter(|item| {
                item.indexer.as_ref() == Some(&indexer)
                    && exclude_verified_if_transcode_disabled(item, transcode_enabled)
                    && exclude_transcoded_if_upload_disabled(item, upload_enabled)
                    && exclude_verify_failures(item)
                    && exclude_transcode_failures(item, retry_failed_transcodes)
                    && item.upload.is_none()
            })
            .collect();
        items.sort_by_key(|x| &x.name);
        let hashes = items.iter().map(|x| x.hash).collect();
        Ok(hashes)
    }

    /// Get all items.
    ///
    /// Items are unsorted.
    pub(crate) async fn get_all(
        &self,
    ) -> Result<BTreeMap<Hash<20>, QueueItem>, Failure<QueueAction>> {
        self.table
            .get_all()
            .await
            .map_err(Failure::wrap(QueueAction::GetAll))
    }

    /// Update an item into the queue
    pub(crate) async fn set(&self, item: QueueItem) -> Result<(), Failure<QueueAction>> {
        self.table
            .set(item.hash, item)
            .await
            .map_err(Failure::wrap(QueueAction::Set))
    }

    /// Add many items.
    ///
    /// If `replace` is true then existing items are replaced
    ///
    /// Items are chunked together to minimize IO operations.
    ///
    /// Returns the number of items added
    pub(crate) async fn set_many(
        &self,
        items: BTreeMap<Hash<20>, QueueItem>,
        replace: bool,
    ) -> Result<usize, Failure<QueueAction>> {
        self.table
            .set_many(items, replace)
            .await
            .map_err(Failure::wrap(QueueAction::SetMany))
    }

    /// Remove an item from the queue
    pub(crate) async fn remove(
        &self,
        hash: Hash<20>,
    ) -> Result<Option<QueueItem>, Failure<QueueAction>> {
        self.table
            .remove(hash)
            .await
            .map_err(Failure::wrap(QueueAction::Remove))
    }

    /// Insert torrent files into the queue if they are not already present
    /// Returns the number of items added
    pub(crate) async fn insert_new_torrent_files(
        &self,
        paths: Vec<PathBuf>,
    ) -> Result<usize, Failure<QueueAction>> {
        let stream = iter(paths.into_iter());
        let items: BTreeMap<_, _> = stream
            .filter_map(|path| async {
                let torrent = match TorrentReader::execute(&path).await {
                    Ok(torrent) => Some(torrent),
                    Err(error) => {
                        warn!("{}", error.render());
                        None
                    }
                };
                let item = QueueItem::from_torrent(path, &torrent?);
                Some((item.hash, item))
            })
            .collect()
            .await;
        self.table
            .set_many(items, false)
            .await
            .map_err(Failure::wrap(QueueAction::SetMany))
    }
}

fn exclude_verify_failures(item: &QueueItem) -> bool {
    !matches!(
        item.verify,
        Some(VerifyStatus {
            verified: false,
            ..
        })
    )
}

fn exclude_transcode_failures(item: &QueueItem, retry_failed_transcodes: bool) -> bool {
    retry_failed_transcodes
        || !matches!(item.transcode, Some(TranscodeStatus { success: false, .. }))
}

fn exclude_verified_if_transcode_disabled(item: &QueueItem, transcode_enabled: bool) -> bool {
    transcode_enabled || item.verify.is_none()
}

fn exclude_transcoded_if_upload_disabled(item: &QueueItem, upload_enabled: bool) -> bool {
    upload_enabled || !matches!(item.transcode, Some(TranscodeStatus { success: true, .. }))
}