//! Functions for working with checkpoint files.
//!
//! Checkpoints in the checkpoints file are all of the same size and packed
//! together. This allows for fast discovery of the number of checkpoints, and
//! whether there is any additional garbage data at the end of the file. This
//! allows for reliable recovery after an unclean shutdown.

use utility::binary::{decode_little_endian_u32, decode_little_endian_u64,
                      encode_little_endian_u32, encode_little_endian_u64};

use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, Write};
use std::io;
use std::iter;
use std::mem;
use std::path::Path;

/// The size in bytes of a checkpoint in the checkpoints file.
pub const CHECKPOINT_SIZE: usize = OFFSET_SIZE + LENGTH_SIZE;

/// The size in bytes of the offset of an entry referenced by a checkpoint.
pub const OFFSET_SIZE: usize = 8;

/// The size in bytes of the length of an entry referenced by a checkpoint.
pub const LENGTH_SIZE: usize = 4;

/// A checkpoint.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct Checkpoint {
    pub offset: u64,
    pub length: u32,
}

impl Checkpoint {
    /// Encode the checkpoint for storage.
    pub fn encode(&self) -> [u8; CHECKPOINT_SIZE] {
        let offset = encode_little_endian_u64(self.offset);
        let length = encode_little_endian_u32(self.length);
        unsafe {
            let mut buffer = mem::uninitialized::<[u8; CHECKPOINT_SIZE]>();
            buffer[          0 ..           0 + OFFSET_SIZE].copy_from_slice(&offset);
            buffer[OFFSET_SIZE .. OFFSET_SIZE + LENGTH_SIZE].copy_from_slice(&length);
            buffer
        }
    }

    /// Decode a checkpoint from storage.
    pub fn decode(buffer: [u8; CHECKPOINT_SIZE]) -> Self {
        let offset_buffer = &buffer[          0 ..           0 + OFFSET_SIZE];
        let length_buffer = &buffer[OFFSET_SIZE .. OFFSET_SIZE + LENGTH_SIZE];
        let offset = decode_little_endian_u64(offset_buffer);
        let length = decode_little_endian_u32(length_buffer);
        Checkpoint{offset, length}
    }
}

/// A checkpoints file contains timestamps and offsets of committed entries.
///
/// Storing this information separately in fixed-sized chunks allows to ensure
/// atomicity of appends, even in the case of crash or power loss.
#[derive(Debug)]
pub struct Checkpoints {
    file: File,
}

impl Checkpoints {
    /// Open a checkpoints file or create it if it does not yet exist.
    pub fn open<P>(path: P) -> io::Result<Self>
        where P: AsRef<Path> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .open(path)?;
        Ok(Checkpoints{file})
    }

    /// Seek a checkpoints file to its logical end.
    ///
    /// Any corrupt checkpoint at the physical end will be overwritten by a
    /// consecutive write.
    fn seek_to_logical_end(&mut self) -> io::Result<()> {
        let physical_end = self.file.seek(io::SeekFrom::End(0))?;
        let logical_end = physical_end / CHECKPOINT_SIZE as u64 * CHECKPOINT_SIZE as u64;
        self.file.seek(io::SeekFrom::Start(logical_end))?;
        Ok(())
    }

    /// Append a checkpoint and fsync the checkpoints file.
    ///
    /// This should not not be called until after the entries file has been
    /// fsynced.
    pub fn append(&mut self, checkpoint: Checkpoint) -> io::Result<()> {
        let buffer = checkpoint.encode();
        self.seek_to_logical_end()?;
        self.file.write_all(&buffer)?;
        self.file.sync_data()?;
        Ok(())
    }

    pub fn iter(&mut self) -> io::Result<iter::Fuse<CheckpointsIter>> {
        Ok(CheckpointsIter::new(&mut self.file)?.fuse())
    }
}

#[derive(Debug)]
pub struct CheckpointsIter<'a> {
    file: &'a mut File,
}

impl<'a> CheckpointsIter<'a> {
    fn new(file: &'a mut File) -> io::Result<Self> {
        file.seek(io::SeekFrom::Start(0))?;
        Ok(CheckpointsIter{file})
    }
}

impl<'a> Iterator for CheckpointsIter<'a> {
    type Item = io::Result<Checkpoint>;

    fn next(&mut self) -> Option<Self::Item> {
        let mut buffer = [0; CHECKPOINT_SIZE];
        match self.file.read(&mut buffer) {
            Err(error) => Some(Err(error)),
            Ok(n) if n != CHECKPOINT_SIZE => None,
            Ok(_) => Some(Ok(Checkpoint::decode(buffer))),
        }
    }
}

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

    use utility::path::push_path;

    use tempdir::TempDir;

    #[test]
    fn test_checkpoint_encode() {
        let examples = &[
            /* ( offset, length, expected buffer                      ) */
               (      0,      0, [0; 12]                              ),
               (      0,      1, [0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0] ),
               (      1,      0, [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] ),
               (     !0,     !0, [255; 12]                            ),
        ];

        for &(offset, length, expected_buffer) in examples {
            let checkpoint = Checkpoint{offset, length};
            let actual_buffer = checkpoint.encode();
            assert!(actual_buffer == expected_buffer,
                    "{:?} != {:?}", actual_buffer, expected_buffer);
        }
    }

    #[test]
    fn test_checkpoints() {
        let directory = TempDir::new("").unwrap();
        let file_path = push_path(directory.path().to_path_buf(), "CHECKPOINTS");
        let mut checkpoints = Checkpoints::open(&file_path).unwrap();
        checkpoints.append(Checkpoint{offset: 0, length: 8}).unwrap();
        checkpoints.append(Checkpoint{offset: 8, length: 14}).unwrap();
        checkpoints.append(Checkpoint{offset: 100, length: 20}).unwrap();
        let mut iter = checkpoints.iter().unwrap();
        assert!(iter.next().unwrap().unwrap() == Checkpoint{offset: 0, length: 8});
        assert!(iter.next().unwrap().unwrap() == Checkpoint{offset: 8, length: 14});
        assert!(iter.next().unwrap().unwrap() == Checkpoint{offset: 100, length: 20});
        assert!(iter.next().is_none());
    }
}
