map_iced 0.1.0

//! This is the real core of the tile cache.
//! It provides asynchronous access to the the caching system.
//!
//! It is the responsibility of the user to make sure that no two requests for the same
//! tile type are in flight at the same time.

use crate::tile_cache::file_util::{FileUtil, LRU_TABLE_FILE, round_to_final_consumption};
use crate::tile_cache::lru_list::LastRecentlyUsedList;
use crate::tile_cache::tile_name_conversion::TileSpecification;
use crate::tile_cache::web_requester::{Requester, RequesterError};
use bytes::Bytes;
use std::fmt::Debug;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::{Arc, Mutex};
use tokio::sync::RwLock;
use tokio::sync::mpsc;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::task::JoinHandle;

///  The maximum channel size we currently allow for.
const MAXIMUM_MESSAGE_CHANNEL: usize = 100;

/// The amount of idle seconds we use till saving.
const AMOUNT_OF_SECONDS_TILL_SAVE: u8 = 5;

/// This struct contains the elements that must be shared across different tasks.
struct ShareableEntries {
    /// The file utility for file access.
    file_util: FileUtil,
    /// The lru list behind a mutex.
    lru_list: Mutex<LastRecentlyUsedList>,
    /// The requester we use for making web requests.
    requester: Requester,
    /// Flags that the initialization is completed.
    initialization_completed: AtomicBool,
    /// Whether downloaded tiles and the LRU table go to disc, switched off for good by
    /// a clear without persisting afterwards. Writers hold it shared, the clear holds
    /// it exclusively, so no file is written into a directory that is being deleted.
    disc_writing: RwLock<bool>,
}

/// The different messages that come from the caching system,
#[derive(Clone, PartialEq, Eq)]
pub enum CachingResultMessage {
    /// We have encountered an error, message included.
    Error { message: String },
    /// The initialization request has been competed.
    InitializationCompleted,
    /// The files of the cache are deleted, see [`CachingSystem::clear_cache`].
    CacheCleared,
    /// The tile data, that has been retrieved.
    TileData {
        /// Zoom level.
        level: u8,
        /// x coordinate.
        x: u32,
        /// y coordinate.
        y: u32,
        /// The data of the tile, put behind an arc to prevent expensive cloning.
        data: Bytes,
        /// The same tile decoded to RGBA, `None` if `data` is no decodable image.
        decoded: Option<DecodedTile>,
    },
    /// Contains the information that a tile id has failed, needed  for book keeping.
    TileFailed {
        /// Zoom level.
        level: u8,
        /// x coordinate.
        x: u32,
        /// y coordinate.
        y: u32,
        /// Why did the tile not arrive..
        message: String,
    },
}

/// A tile decoded to 8 bit RGBA. iced uploads such an image synchronously, while
/// an encoded one goes through a single background worker and is not drawn until
/// that worker gets round to it.
#[derive(Clone, PartialEq, Eq)]
pub struct DecodedTile {
    pub width: u32,
    pub height: u32,
    pub pixels: Bytes,
}

impl DecodedTile {
    /// Decodes `data`, run in the tokio task so the UI thread does not pay for it.
    fn decode(data: &[u8]) -> Option<Self> {
        let image = image::load_from_memory(data).ok()?.into_rgba8();
        Some(Self {
            width: image.width(),
            height: image.height(),
            pixels: Bytes::from(image.into_raw()),
        })
    }
}

impl Debug for CachingResultMessage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CachingResultMessage::Error { message } => write!(f, "Caching error: {}", message),
            CachingResultMessage::InitializationCompleted => write!(f, "Initialization completed"),
            CachingResultMessage::CacheCleared => write!(f, "Cache cleared"),
            CachingResultMessage::TileData {
                level, x, y, data, ..
            } => {
                write!(
                    f,
                    "Tile data level {}, x {}, y {}, size {}",
                    level,
                    x,
                    y,
                    data.len()
                )
            }
            CachingResultMessage::TileFailed {
                level,
                x,
                y,
                message,
            } => {
                write!(
                    f,
                    "Tile failed level {}, x {}, y {}, message {}",
                    level, x, y, message
                )
            }
        }
    }
}

/// The caching system we also represent to the outside.
pub struct CachingSystem {
    /// The entry that gets cloned for every working thread.
    cloneable_entry: Arc<ShareableEntries>,
    /// The stream reader for the tokio stream.
    stream_reader: Option<Receiver<CachingResultMessage>>,
    /// The sender used for data,
    stream_sender: Sender<CachingResultMessage>,
    /// The atomic counter we use for saving the lru cache.
    savings_counter: Arc<AtomicU8>,
    /// The join handle for the timer to kill in drop trait.
    timer_handle: JoinHandle<()>,
    /// The copyright text that belongs to the tile system.
    copyright_text: String,
}

impl CachingSystem {
    pub fn new(
        requester: Requester,
        cache_base_dir: impl AsRef<Path>,
        maximum_amount_of_data: u64,
        copyright_text: String,
    ) -> CachingSystem {
        let (tx, rx) = mpsc::channel::<CachingResultMessage>(MAXIMUM_MESSAGE_CHANNEL);
        let sharable_entry = ShareableEntries {
            file_util: FileUtil::new(cache_base_dir),
            lru_list: Mutex::new(LastRecentlyUsedList::new(maximum_amount_of_data)),
            requester,
            initialization_completed: AtomicBool::new(false),
            disc_writing: RwLock::new(true),
        };

        let arc_sharable_entry = Arc::new(sharable_entry);
        let savings_counter = Arc::new(AtomicU8::new(0));
        let timer_handle = tokio::spawn(Self::savings_timer(
            savings_counter.clone(),
            arc_sharable_entry.clone(),
            tx.clone(),
        ));

        Self {
            cloneable_entry: arc_sharable_entry,
            stream_reader: Some(rx),
            stream_sender: tx,
            savings_counter,
            timer_handle,
            copyright_text,
        }
    }

    /// Gets a copy of the copyright text.
    pub fn get_copyright_text(&self) -> String {
        self.copyright_text.clone()
    }

    /// Timer function that does an autosave after a couple of idle seconds.
    async fn savings_timer(
        counter: Arc<AtomicU8>,
        shareable_entries: Arc<ShareableEntries>,
        sender: Sender<CachingResultMessage>,
    ) {
        loop {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            let previous = counter.try_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
                Some(current.saturating_sub(1))
            });
            if previous == Ok(1) {
                Self::save_lru_table(&shareable_entries, &sender).await;
            }
        }
    }

    /// Extracts the receiver out of the class. Can only be done one time.
    pub fn get_receiver(&mut self) -> Option<Receiver<CachingResultMessage>> {
        self.stream_reader.take()
    }

    /// Asynchronous initialization function to set up the system.
    async fn process_initialize(
        sharable_entry: Arc<ShareableEntries>,
        sender: Sender<CachingResultMessage>,
        savings_counter: Arc<AtomicU8>,
    ) {
        // Remove all orphan files.
        sharable_entry.file_util.remove_temps_from_base().await;
        // Now we load all the data from the images on the cache.
        let data = sharable_entry
            .file_util
            .get_all_pngs_interpreted_as_u64()
            .await;
        // Let us check if we get an lru table.
        let lru_data = FileUtil::convert_to_u64(
            &sharable_entry
                .file_util
                .try_load_plain(LRU_TABLE_FILE)
                .await
                .unwrap_or_default(),
        );
        let deletion_list = sharable_entry
            .lru_list
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .reconstruct_from(&lru_data, &data);

        // Remove any orphan files, we do not cache anymore.
        sharable_entry.file_util.remove_files(&deletion_list).await;

        sharable_entry
            .initialization_completed
            .store(true, Ordering::Relaxed);

        // We also register a save because the contents of the LRU may have changed.
        savings_counter.store(AMOUNT_OF_SECONDS_TILL_SAVE, Ordering::SeqCst);

        // If an error occurred the receiver has been dropped in the meantime.
        let _ = sender
            .send(CachingResultMessage::InitializationCompleted)
            .await;
    }

    /// Initializes the system. Starts an internal tokio task for the work. Called
    /// once, right after construction.
    pub fn initialize(&self) {
        debug_assert!(
            !self
                .cloneable_entry
                .initialization_completed
                .load(Ordering::Relaxed),
            "CachingSystem::initialize called twice"
        );
        tokio::spawn(Self::process_initialize(
            self.cloneable_entry.clone(),
            self.stream_sender.clone(),
            self.savings_counter.clone(),
        ));
    }

    /// Deletes all files of the cache and empties its bookkeeping. Starts an internal
    /// tokio task for the work, which answers with [`CachingResultMessage::CacheCleared`].
    /// With `persist_afterwards` false nothing gets written to disc any more for the
    /// lifetime of this system, tiles are only delivered. Must not be called before
    /// the initialization is completed, the startup scan would race the deletion.
    pub fn clear_cache(&self, persist_afterwards: bool) {
        debug_assert!(
            self.cloneable_entry
                .initialization_completed
                .load(Ordering::Relaxed),
            "CachingSystem::clear_cache before initialization"
        );
        tokio::spawn(Self::process_clear_cache(
            self.cloneable_entry.clone(),
            self.stream_sender.clone(),
            persist_afterwards,
        ));
    }

    /// Asynchronous part of [`Self::clear_cache`].
    async fn process_clear_cache(
        sharable_entry: Arc<ShareableEntries>,
        sender: Sender<CachingResultMessage>,
        persist_afterwards: bool,
    ) {
        let result = {
            // Waits for the saves running, later ones wait for the deletion.
            let mut disc_writing = sharable_entry.disc_writing.write().await;
            *disc_writing &= persist_afterwards;
            sharable_entry
                .lru_list
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .reset();
            sharable_entry.file_util.remove_cache_files().await
        };
        if let Err(text) = result {
            let _ = sender
                .send(CachingResultMessage::Error {
                    message: "Error on clearing the cache: ".to_string() + text.as_str(),
                })
                .await;
        }
        let _ = sender.send(CachingResultMessage::CacheCleared).await;
    }

    /// Poses a request for a tile. Starts an internal tokio task for the work.
    pub fn request_tile(&self, level: u8, x: u32, y: u32) -> Result<(), String> {
        if !self
            .cloneable_entry
            .initialization_completed
            .load(Ordering::Relaxed)
        {
            return Err(String::from(
                "CachingSystem::request_tile not initialized yet",
            ));
        }
        let shareable_entry = self.cloneable_entry.clone();
        let sender = self.stream_sender.clone();
        tokio::spawn(Self::process_request_tile(
            level,
            x,
            y,
            shareable_entry,
            sender,
            self.savings_counter.clone(),
        ));
        Ok(())
    }

    /// Internal work for requesting a tile basically distinguishes between cache hit and miss.
    async fn process_request_tile(
        level: u8,
        x: u32,
        y: u32,
        sharable_entry: Arc<ShareableEntries>,
        sender: Sender<CachingResultMessage>,
        savings_counter: Arc<AtomicU8>,
    ) {
        let destination = TileSpecification::new(level, x, y);
        // Here we have to distinguish if we have it on cache or not.
        if let Some(image_data) = sharable_entry
            .file_util
            .try_load_plain(destination.filename())
            .await
        {
            // First send off the result.
            let _ = sender
                .send(CachingResultMessage::TileData {
                    level,
                    x,
                    y,
                    decoded: DecodedTile::decode(&image_data),
                    data: Bytes::from(image_data),
                })
                .await;
            // Now we have to check with the cache.
            sharable_entry
                .lru_list
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .touch(destination.into());
        } else {
            Self::deal_with_cache_miss(&sharable_entry, sender, destination).await;
        }
        // Flag for cache flush.
        savings_counter.store(AMOUNT_OF_SECONDS_TILL_SAVE, Ordering::SeqCst);
    }

    /// The cache miss part is more complicated but tries to serve the data as fast as possible.
    async fn deal_with_cache_miss(
        sharable_entry: &Arc<ShareableEntries>,
        sender: Sender<CachingResultMessage>,
        destination: TileSpecification,
    ) {
        // In this case the file is not on the cache so we have to get it.
        let web_access = sharable_entry.requester.get_image_data(destination).await;
        let raw_data: Bytes = match web_access {
            Ok(data) if data.is_empty() => {
                let _ = sender
                    .send(CachingResultMessage::TileFailed {
                        x: destination.x(),
                        y: destination.y(),
                        level: destination.level(),
                        message: "Empty data received from server".to_string(),
                    })
                    .await;
                return;
            }
            Ok(data) => data,
            Err(text) => {
                let _ = sender
                    .send(CachingResultMessage::TileFailed {
                        x: destination.x(),
                        y: destination.y(),
                        level: destination.level(),
                        message: text,
                    })
                    .await;
                return;
            }
        };

        // First send the result.
        let _ = sender
            .send(CachingResultMessage::TileData {
                level: destination.level(),
                x: destination.x(),
                y: destination.y(),
                decoded: DecodedTile::decode(&raw_data),
                data: raw_data.clone(),
            })
            .await;

        let disc_writing = sharable_entry.disc_writing.read().await;
        if !*disc_writing {
            return;
        }
        let new_memory = round_to_final_consumption(raw_data.len() as u64);
        // Now we have to save the data on the disc.
        if let Err(text) = sharable_entry
            .file_util
            .safe_save(destination.filename(), &raw_data)
            .await
        {
            let _ = sender
                .send(CachingResultMessage::Error { message: text })
                .await;

            return;
        }

        let to_remove = sharable_entry
            .lru_list
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert_and_clear(destination.into(), new_memory);
        sharable_entry.file_util.remove_files(&to_remove).await;
    }

    /// Helper routine to write out the cache file.
    async fn save_lru_table(
        sharable_entry: &Arc<ShareableEntries>,
        sender: &Sender<CachingResultMessage>,
    ) {
        let disc_writing = sharable_entry.disc_writing.read().await;
        if !*disc_writing {
            return;
        }
        let raw_data = sharable_entry
            .lru_list
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .generate_usage_list();
        let res = sharable_entry
            .file_util
            .safe_save(LRU_TABLE_FILE, &FileUtil::convert_to_u8(&raw_data))
            .await;
        match res {
            Ok(_) => {} // Nothing to do here.
            Err(x) => {
                let _ = sender
                    .send(CachingResultMessage::Error {
                        message: "Error on writing cache file: ".to_string() + x.as_str(),
                    })
                    .await;
            }
        }
    }
}

impl Drop for CachingSystem {
    fn drop(&mut self) {
        self.timer_handle.abort();
    }
}

/// The dummy cache we use for testing purposes.
pub fn generate_dummy_cache(
    cache_base_dir: impl AsRef<Path>,
    maximum_amount_of_data: u64,
) -> CachingSystem {
    CachingSystem::new(
        Requester::dummy(),
        cache_base_dir,
        maximum_amount_of_data,
        "".to_string(),
    )
}

/// Generates the real cache. The url parts and the user agent go to the [`Requester`], see
/// [`Requester::new`]. The copyright text gets displayed with the tiles. Cache base directory is
/// the place where we store data and the maximum amount is the maximum amount of data we want to take.
pub fn generate_cache(
    intro_url: &str,
    post_url: &str,
    user_agent: &str,
    copyright_text: String,
    cache_base_dir: impl AsRef<Path>,
    maximum_amount_of_data: u64,
) -> Result<CachingSystem, RequesterError> {
    Ok(CachingSystem::new(
        Requester::new(intro_url, post_url, user_agent)?,
        cache_base_dir,
        maximum_amount_of_data,
        copyright_text,
    ))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::assert_matches;
    use std::time::Duration;

    #[tokio::test]
    async fn first_setup() {
        let mut cache = generate_dummy_cache(tempfile::tempdir().unwrap(), 10_000);
        cache.initialize();
        let mut receiver = cache.get_receiver().unwrap();
        let message = receiver.recv().await.unwrap();
        assert_eq!(message, CachingResultMessage::InitializationCompleted);
    }

    #[tokio::test]
    async fn first_fill() {
        let mut cache = generate_dummy_cache(tempfile::tempdir().unwrap(), 20_000);
        cache.initialize();
        let mut receiver = cache.get_receiver().unwrap();
        let message = receiver.recv().await.unwrap();
        assert_eq!(message, CachingResultMessage::InitializationCompleted);
        for y in 0..100 {
            cache
                .request_tile(0, 1, y)
                .expect("Initialization uncompleted.");
        }

        for _ in 0..100 {
            let message = receiver.recv().await.unwrap();
            assert_matches!(
                message,
                CachingResultMessage::TileData { level: 0, x: 1, .. }
            );
        }
        // Wait for the data written to disc.
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    /// Requests the tiles and waits for all of them to arrive.
    async fn fetch_tiles(
        cache: &CachingSystem,
        receiver: &mut Receiver<CachingResultMessage>,
        count: u32,
    ) {
        for y in 0..count {
            cache.request_tile(2, 1, y).expect("Initialized");
        }
        for _ in 0..count {
            let message = receiver.recv().await.unwrap();
            assert_matches!(message, CachingResultMessage::TileData { level: 2, .. });
        }
        // Wait for the data written to disc.
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    #[tokio::test]
    async fn clear_cache_test() {
        for persist_afterwards in [true, false] {
            let dir = tempfile::tempdir().unwrap();
            let util = FileUtil::new(&dir);
            let mut cache = generate_dummy_cache(&dir, 100_000);
            cache.initialize();
            let mut receiver = cache.get_receiver().unwrap();
            let message = receiver.recv().await.unwrap();
            assert_eq!(message, CachingResultMessage::InitializationCompleted);
            fetch_tiles(&cache, &mut receiver, 4).await;
            assert_eq!(util.get_all_pngs_interpreted_as_u64().await.len(), 4);

            cache.clear_cache(persist_afterwards);
            let message = receiver.recv().await.unwrap();
            assert_eq!(message, CachingResultMessage::CacheCleared);
            assert!(util.get_all_pngs_interpreted_as_u64().await.is_empty());

            // The tiles still arrive, but only go to disc if persisting.
            fetch_tiles(&cache, &mut receiver, 3).await;
            let expected = if persist_afterwards { 3 } else { 0 };
            assert_eq!(util.get_all_pngs_interpreted_as_u64().await.len(), expected);
        }
    }
}