//! A journal is an ordered append-only list of possibly duplicate entries. An
//! entry is simply a bytestring of arbitrary length (currently up to 4 GiB
//! each).
//!
//! A journal is stored using two files: the checkpoints file and the entries
//! file. The checkpoints file stores the offset and length of each entry in the
//! entries file. The entries file simply stores the raw entry data, and nothing
//! else.

pub mod checkpoints;
pub mod entries;

use database::journal::checkpoints::{Checkpoint, Checkpoints, CheckpointsIter};
use database::journal::entries::Entries;

use r2d2::{Pool, PooledConnection, ManageConnection};
use r2d2;

use smallvec::SmallVec;

use std::io;
use std::iter;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

/// A sequence of entries on disk.
#[derive(Debug)]
pub struct Journal {
    checkpoints: Checkpoints,
    entries: Entries,
}

impl Journal {
    /// Open a journal or create it if it does not yet exist.
    pub fn open<P, Q>(checkpoints_path: P, entries_path: Q) -> io::Result<Self>
        where P: AsRef<Path>, Q: AsRef<Path> {
        let checkpoints = Checkpoints::open(checkpoints_path)?;
        let entries = Entries::open(entries_path)?;
        Ok(Journal{checkpoints, entries})
    }

    /// Append an entry onto the journal and fsync it. This method can be
    /// invoked concurrently with readers, but not concurrently with other
    /// writers for the same journal. See [`ConcurrentJournal`], which contains
    /// a mutex to prevent concurrent writes.
    pub fn append(&mut self, entry: &[u8]) -> io::Result<()> {
        let offset = self.entries.append(entry)?;
        let length = entry.len() as u32;
        let checkpoint = Checkpoint{offset, length};
        self.checkpoints.append(checkpoint)?;
        Ok(())
    }

    /// Iterator over all the entries in a journal. See [`JournalIter`] for more
    /// information.
    pub fn iter(&mut self) -> io::Result<JournalIter> {
        JournalIter::new(self)
    }
}

/// Iterator over all the entries in a journal.
///
/// This iterator can be used concurrently with other reads from and writes to
/// the same journal. It will ignore the last entry if it is still being
/// written; no incomplete or corrupt entries will be read.
#[derive(Debug)]
pub struct JournalIter<'a> {
    checkpoints: iter::Fuse<CheckpointsIter<'a>>,
    entries: &'a mut Entries,
}

impl<'a> JournalIter<'a> {
    pub fn new(journal: &'a mut Journal) -> io::Result<Self> {
        let checkpoints = journal.checkpoints.iter()?;
        let entries = &mut journal.entries;
        Ok(JournalIter{checkpoints, entries})
    }
}

impl<'a> Iterator for JournalIter<'a> {
    type Item = io::Result<SmallVec<[u8; 256]>>;

    fn next(&mut self) -> Option<Self::Item> {
        let checkpoint = match self.checkpoints.next() {
            None => return None,
            Some(Err(error)) => return Some(Err(error)),
            Some(Ok(checkpoint)) => checkpoint,
        };

        let entry = self.entries.read(checkpoint.offset,
                                      checkpoint.length);
        Some(entry)
    }
}

aggregate_errors!(
    ConcurrentJournalOpenError {
        R2d2Error(r2d2::Error),
    }

    ConcurrentJournalAppendError {
        IoError(io::Error),
        R2d2Error(r2d2::Error),
    }

    ConcurrentJournalReadError {
        R2d2Error(r2d2::Error),
    }
);

/// Like [`Journal`], but [`Sync`].
///
/// Uses a pool of file handles, and a lock for preventing concurrent writes. A
/// write and a read can concur, as can multiple reads. This is because journals
/// are append-only, and readers stop when they find an incomplete checkpoint.
#[derive(Debug)]
pub struct ConcurrentJournal {
    pool: Pool<JournalManager>,
    write: Mutex<()>,
}

impl ConcurrentJournal {
    /// Open a concurrent journal.
    pub fn open(checkpoints_path: PathBuf, entries_path: PathBuf) ->
        Result<Self, ConcurrentJournalOpenError> {
        let journal_manager = JournalManager{checkpoints_path, entries_path};
        let pool = Pool::new(journal_manager)?;
        let write = Mutex::new(());
        Ok(ConcurrentJournal{pool, write})
    }

    /// Append an entry onto the journal.
    pub fn append(&self, entry: &[u8]) -> Result<(), ConcurrentJournalAppendError> {
        let mut journal = self.pool.get()?;
        let _ = self.write.lock().unwrap();
        journal.append(entry)?;
        Ok(())
    }

    /// Get a non-concurrent journal for reading.
    pub fn read(&self) -> Result<PooledConnection<JournalManager>,
                                 ConcurrentJournalReadError> {
        let journal = self.pool.get()?;
        Ok(journal)
    }
}

/// Connection manager for journals. This can be used with the r2d2 pooling
/// library to keep a pool of open journals.
#[derive(Debug)]
pub struct JournalManager {
    pub checkpoints_path: PathBuf,
    pub entries_path: PathBuf,
}

impl ManageConnection for JournalManager {
    type Connection = Journal;
    type Error = io::Error;

    fn connect(&self) -> Result<Self::Connection, Self::Error> {
        Journal::open(&self.checkpoints_path, &self.entries_path)
    }

    fn is_valid(&self, _conn: &mut Self::Connection) -> Result<(), Self::Error> {
        Ok(())
    }

    fn has_broken(&self, _conn: &mut Self::Connection) -> bool {
        false
    }
}
