mod file_ext;
mod page;
mod page_no;
mod size_info;

pub use page_no::*;

use file_ext::FileExt;
use page::Page;
use size_info::SizeInfo;
use std::fs::File;
use std::io::{Error, ErrorKind, Result};
use std::path::Path;
use std::sync::atomic::{AtomicU32, Ordering};

use page_lock::PageLocker;
use tokio::task::spawn_blocking;

/// **NOTE**: Caller must ensure that closure must not panic.
/// If it does, it will be undefined behavior.
macro_rules! into_async { [$body:expr] => { unsafe { spawn_blocking(move || $body).await.unwrap_unchecked() } }; }
macro_rules! err { [$cond: expr, $msg: expr] => { if $cond { return Err(Error::new(ErrorKind::InvalidInput, $msg)); } }; }

pub struct Pages<N, const SIZE: usize> {
    /// Total Page number
    len: AtomicU32,
    file: &'static File,
    locker: PageLocker<N>,
    size_info: SizeInfo,
}

impl<N: PageNo, const SIZE: usize> Pages<N, SIZE> {
    /// Create a new `Pages` instance.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        err!(SIZE < 64, "Page size should >= 64 bytes");
        err!(SIZE > 1024 * 256, "Page size should > 256 kilobytes");

        let size_info = SizeInfo {
            block_size: SIZE as u32,
            pages_len_nbytes: N::SIZE as u8,
        };
        let file = File::options()
            .read(true)
            .write(true)
            .create(true)
            .open(path)?;

        let file_len = file.metadata()?.len();
        // So that, There is no residue bytes.
        if file_len % SIZE as u64 != 0 {
            return Err(ErrorKind::InvalidData.into());
        }
        let mut len = file_len as u32 / SIZE as u32;
        // Is new file? set the length to 1.
        if file_len == 0 {
            len = 1;
            file.set_len(SIZE as u64)?;
        }
        // Exist File?  Check size info.
        else {
            let mut buf = [0; 4];
            file.read_exact_at(&mut buf, 0)?;
            let info = SizeInfo::from(buf);
            err!(
                info != size_info,
                format!("Expected {:?}, but got: {:?}", info, size_info)
            );
        }
        Ok(Self {
            len: AtomicU32::new(len),
            file: Box::leak(Box::new(file)),
            size_info,
            locker: PageLocker::new(),
        })
    }

    pub async fn sync_data(&self) -> Result<()> {
        let file = self.file;
        into_async!(file.sync_data())
    }

    pub async fn sync_all(&self) -> Result<()> {
        let file = self.file;
        into_async!(file.sync_all())
    }

    pub async fn read(&self, num: N) -> Result<[u8; SIZE]> {
        debug_assert!(num.as_u32() < self.len());
        self.locker.unlock(num).await;
        let num = num.as_u32() as u64;
        let file = self.file;
        into_async!({
            let mut buf = [0; SIZE];
            file.read_exact_at(&mut buf, SIZE as u64 * num)?;
            Ok(buf)
        })
    }

    pub async fn goto(&self, num: N) -> Result<Page<'_, N, SIZE>> {
        let buf = self.read(num).await?;
        let _lock = self.locker.lock(num);
        Ok(Page {
            _lock,
            num,
            buf,
            pages: self,
        })
    }

    pub async fn write(&self, num: N, buf: [u8; SIZE]) -> Result<()> {
        debug_assert!(num.as_u32() < self.len());
        let num = num.as_u32() as u64;
        let file = self.file;
        into_async!(file.write_all_at(&buf, SIZE as u64 * num))
    }

    /// Increase page number by `count`
    pub async fn alloc(&self, count: u32) -> Result<u32> {
        let old_len = self.len.fetch_add(count, Ordering::SeqCst);
        let file = self.file;
        into_async!(file.set_len(SIZE as u64 * (old_len + count) as u64))?;
        Ok(old_len)
    }

    pub async fn create(&self, buf: [u8; SIZE]) -> Result<()> {
        let num = self.len.fetch_add(1, Ordering::SeqCst);
        self.write(PageNo::new(num), buf).await
    }

    pub fn alloc_sync(&self, count: u32) -> Result<u32> {
        let old_len = self.len.fetch_add(count, Ordering::SeqCst);
        self.file
            .set_len(SIZE as u64 * (old_len + count) as u64)?;
        Ok(old_len)
    }

    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> u32 {
        self.len.load(Ordering::SeqCst)
    }

    pub fn inner(&self) -> &File {
        self.file
    }
}

impl<N, const SIZE: usize> Drop for Pages<N, SIZE> {
    fn drop(&mut self) {
        self.file
            .write_all_at(&self.size_info.to_bytes(), 0)
            .unwrap();
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    async fn read_write() -> Result<()> {
        let pages = Pages::<u16, 64>::open("test1.db")?;

        assert_eq!(pages.len(), 1);
        pages.create([0; 64]).await?;
        assert_eq!(pages.len(), 2);

        let mut page = pages.goto(1).await?;
        page.buf = [1; 64];
        page.write().await?;

        assert_eq!(pages.read(1).await?, [1; 64]);
        pages.alloc(1).await?;

        assert_eq!(pages.len(), 3);

        pages.write(2, [2; 64]).await?;
        assert_eq!(pages.read(2).await?, [2; 64]);

        std::fs::remove_file("test1.db")
    }

    async fn alloc_page() -> Result<()> {
        let pages = Pages::<u16, 64>::open("test2.db")?;

        pages.create([1; 64]).await?;
        assert_eq!(pages.len(), 2);
        pages.alloc(1).await?;
        assert_eq!(pages.len(), 3);

        std::fs::remove_file("test2.db")
    }

    #[test]
    fn all() {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("Failed building the Runtime")
            .block_on(async {
                read_write().await.unwrap();
                alloc_page().await.unwrap();
            });
    }
}