use std::fs::OpenOptions;
use std::fs::File;
use std::sync;
use std::rc::Rc;
use std::sync::Arc;
use std::error;
use std::io::{Read, Write, Error};
use byteorder::{ReadBytesExt, WriteBytesExt, BigEndian};

use allocator::Allocator;
use address::Address;
use segment::{SegmentScanner, SegmentIterator};
use transaction::{Transaction, TxRead};
use discref::DiscRef;
use journal::Journal;
use config::Config;
use std::str;
use transaction::TxSegCheck::{CREATED, DROPPED, NONE};
use std::collections::HashMap;

const DEFAULT_PAGE_EXP: u8 = 10; // 2^10

#[derive(PartialEq,Eq,PartialOrd,Ord,Hash,Clone,Debug)]
pub struct RecRef {
    pub page: u64,
    pub pos: u32,
}

pub struct PersyImpl {
    journal: Journal,
    address: Address,
    allocator: Arc<Allocator>,
}

pub struct RecordScanner<'a> {
    persy: &'a PersyImpl,
    segment: String,
    scanner: SegmentScanner<'a>,
}

pub struct RecordIterator<'a> {
    persy: &'a PersyImpl,
    segment: String,
    iterator: SegmentIterator<'a>,
}

impl<'a> RecordScanner<'a> {
    fn new<'b>(persy: &'b PersyImpl, segment: &String, scanner: SegmentScanner<'b>) -> RecordScanner<'b> {
        RecordScanner::<'b> {
            persy: persy,
            segment: segment.clone(),
            scanner: scanner,
        }
    }
}

impl<'a> IntoIterator for RecordScanner<'a> {
    type Item = Vec<u8>;
    type IntoIter = RecordIterator<'a>;

    fn into_iter(self) -> Self::IntoIter {
        RecordIterator::<'a> {
            persy: self.persy,
            segment: self.segment,
            iterator: self.scanner.into_iter(),
        }
    }
}

impl<'a> Iterator for RecordIterator<'a> {
    type Item = Vec<u8>;
    fn next(&mut self) -> Option<Vec<u8>> {
        let rec;
        loop {
            let iter = self.iterator.next();
            if let Some(id) = iter {
                let res = self.persy.read_record(&self.segment, &id);
                if !res.is_err() {
                    let tp = res.unwrap();
                    if tp.is_some() {
                        rec = tp;
                        break;
                    }
                }
            } else {
                rec = None;
                break;
            }
        }

        rec
    }
}


pub struct TransactionFinalize {
    transaction: Transaction,
}

#[derive(PartialEq,Debug)]
pub enum RecoverStatus {
    Started,
    PrepareCommit,
    Rollback,
    Commit,
}

#[derive(Debug,PartialEq,Eq)]
pub enum PersyError {
    IO(String),
    Err(String),
    Encoding(str::Utf8Error),
    RecordNotFound,
    SegmentNotFound,
    SegmentAlreadyExists,
    Lock,
}

pub type PRes<T> = Result<T, PersyError>;

impl PersyImpl {
    pub fn create<P: Into<String>>(path: P) -> PRes<()> {
        let f = OpenOptions::new().write(true)
            .read(true)
            .create_new(true)
            .open(path.into())?;
        PersyImpl::init_file(f)?;
        Ok(())
    }

    fn init_file(fl: File) -> PRes<()> {
        let mut disc = DiscRef::new(fl);
        // root_page is everytime 0
        let root_page = disc.create_page_raw(DEFAULT_PAGE_EXP)?;
        let allocator_page = Allocator::init(&mut disc)?;
        let ref allocator = Allocator::new(disc, &Rc::new(Config::new()), allocator_page)?;
        let address_page = Address::init(allocator)?;
        let journal_page = Journal::init(allocator)?;
        {
            let mut root = allocator.disc().load_page_raw(root_page, DEFAULT_PAGE_EXP)?;
            // Version of the disc format
            root.write_u16::<BigEndian>(0)?;
            // Position of the start of address structure
            root.write_u64::<BigEndian>(address_page)?;
            // Start of the Log data, if shutdown wel this will be everytime 0
            root.write_u64::<BigEndian>(journal_page)?;
            root.write_u64::<BigEndian>(allocator_page)?;
            allocator.flush_page(&mut root)?;
            // TODO: check this never go over the first page
        }
        allocator.disc().flush()?;
        Ok(())
    }

    fn new(file: File, config: Config) -> PRes<PersyImpl> {
        let disc = DiscRef::new(file);
        let address_page;
        let journal_page;
        let allocator_page;
        {
            let mut pg = disc.load_page_raw(0, DEFAULT_PAGE_EXP)?;
            pg.read_u16::<BigEndian>()?;//THIS NOW is 0 all the time
            address_page = pg.read_u64::<BigEndian>()?;
            journal_page = pg.read_u64::<BigEndian>()?;
            allocator_page = pg.read_u64::<BigEndian>()?;
        }
        let config = Arc::new(config);
        let allocator = Arc::new(Allocator::new(disc, &config, allocator_page)?);
        let address = Address::new(&allocator, &config, address_page)?;
        let journal = Journal::new(&allocator, journal_page)?;
        Ok(PersyImpl {
            journal: journal,
            address: address,
            allocator: allocator,
        })

    }

    fn recover(&self) -> PRes<()> {
        let mut last_id = None;
        let mut commit_order = Vec::new();
        let mut transactions = HashMap::new();
        self.journal
            .recover(|record, id| {
                {
                    let tx = transactions.entry(id.clone()).or_insert_with(|| (RecoverStatus::Started, Transaction::recover(id.clone())));
                    let res = record.recover(&mut tx.1);
                    if res.is_err() {
                        tx.0 = RecoverStatus::Rollback;
                    } else {
                        // TODO: check if is in rollback do something else.
                        match res.unwrap() {
                            RecoverStatus::Started => {
                                if tx.0 != RecoverStatus::Rollback {
                                    tx.0 = RecoverStatus::Started;
                                }
                            }
                            RecoverStatus::PrepareCommit => {
                                if tx.0 != RecoverStatus::Rollback {
                                    tx.0 = RecoverStatus::PrepareCommit;
                                    commit_order.push(id.clone());
                                }
                            }
                            RecoverStatus::Rollback => {
                                tx.0 = RecoverStatus::Rollback;
                            }
                            RecoverStatus::Commit => {
                                if tx.0 != RecoverStatus::Rollback {
                                    tx.0 = RecoverStatus::Commit;
                                }
                            }
                        }
                    }
                }
                last_id = Some(id.clone());
            })?;

        let ref allocator = self.allocator;
        let ref address = self.address;
        for id in commit_order {
            if let Some(mut rec) = transactions.remove(&id) {
                if rec.0 == RecoverStatus::PrepareCommit {
                    rec.1.recover_prepare_commit(address, allocator)?;
                    rec.1.recover_commit(address, allocator)?;
                }
            }
        }

        for (_, rec) in transactions.iter_mut() {
            rec.1.recover_rollback(address, allocator)?;
        }
        if let Some(id) = last_id {
            self.journal.clear(&id)?;
        }
        allocator.flush_free_list()?;
        Ok(())
    }


    pub fn open<P: Into<String>>(path: P, config: Config) -> PRes<PersyImpl> {
        let f = OpenOptions::new().write(true)
            .read(true)
            .create(false)
            .truncate(false)
            .open(path.into())?;
        let persy = PersyImpl::new(f, config)?;
        persy.recover()?;
        Ok(persy)
    }

    pub fn begin(&self) -> PRes<Transaction> {
        let ref journal = self.journal;
        Ok(Transaction::new(journal)?)
    }

    pub fn create_segment(&self, tx: &mut Transaction, segment: &str) -> PRes<()> {
        match tx.exists_segment(&segment.into()) {
            DROPPED => {}
            CREATED => {
                return Err(PersyError::SegmentAlreadyExists);
            }
            NONE => {
                if self.address.exists_segment(&segment.into())? {
                    return Err(PersyError::SegmentAlreadyExists);
                }
            }
        }
        self.address.create_temp_segment(&segment.into())?;
        tx.add_create_segment(&self.journal, segment)?;
        Ok(())

    }

    pub fn drop_segment(&self, tx: &mut Transaction, segment: &str) -> PRes<()> {
        self.check_segment_tx(tx, &segment.into())?;
        tx.add_drop_segment(&self.journal, segment)?;
        Ok(())
    }

    pub fn exists_segment(&self, segment: &str) -> PRes<bool> {
        self.address.exists_segment(&segment.into())
    }

    pub fn exists_segment_tx(&self, tx: &Transaction, segment: &str) -> PRes<bool> {
        match tx.exists_segment(&segment.into()) {
            DROPPED => {
                return Ok(false);
            }
            CREATED => return Ok(true),
            NONE => {
                if !self.address.exists_segment(&segment.into())? {
                    return Ok(false);
                } else {

                    return Ok(true);
                }
            }
        }
    }

    /// check if a segment exist persistent or in tx.
    ///
    /// @return true if the segment was created in tx.
    fn check_segment_tx(&self, tx: &Transaction, segment: &String) -> PRes<bool> {
        match tx.exists_segment(segment) {
            DROPPED => {
                return Err(PersyError::SegmentNotFound);
            }
            CREATED => return Ok(true),
            NONE => {
                if !self.address.exists_segment(segment)? {
                    return Err(PersyError::SegmentNotFound);
                }
            }
        }
        return Ok(false);
    }

    pub fn insert_record(&self, tx: &mut Transaction, segment: &str, rec: &Vec<u8>) -> PRes<RecRef> {
        let temp = self.check_segment_tx(tx, &segment.into())?;
        let len = rec.len();
        let allocation_exp = exp_from_size((len + 8) as u64);
        let ref allocator = self.allocator;
        let ref address = self.address;
        let ref journal = self.journal;
        let page = allocator.allocate(allocation_exp)?;
        let rec_ref;
        if temp {
            rec_ref = address.allocate_temp(&segment.into())?
        } else {
            rec_ref = address.allocate(&segment.into())?;
        }
        tx.add_insert(journal, &segment.into(), &rec_ref, page)?;
        {
            let mut pg = allocator.write_page(page)?;
            pg.write_u64::<BigEndian>(len as u64)?;
            pg.write_all(rec)?;
            allocator.flush_page(&mut pg)?;
        }
        Ok(rec_ref)
    }

    fn read_ref(&self, tx: &Transaction, segment: &String, rec_ref: &RecRef) -> PRes<Option<u64>> {
        self.check_segment_tx(tx, segment)?;
        match tx.read(rec_ref) {
            TxRead::RECORD(rec) => return Ok(Some(rec)),
            TxRead::DELETED => return Ok(None),
            TxRead::NONE => {
                return self.address.read(rec_ref, segment);
            }
        }
    }

    pub fn read_record_tx(&self, tx: &Transaction, segment: &str, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
        if let Some(page) = self.read_ref(tx, &segment.into(), rec_ref)? {
            let mut buffer;
            {
                let mut pg = self.allocator.load_page(page)?;
                let len = pg.read_u64::<BigEndian>()?;
                buffer = Vec::<u8>::with_capacity(len as usize);
                pg.take(len).read_to_end(&mut buffer)?;
            }
            return Ok(Some(buffer));
        }
        Ok(None)
    }

    pub fn read_record(&self, segment: &str, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
        if !self.exists_segment(segment)? {
            return Err(PersyError::SegmentNotFound);
        }
        let page;
        if let Some(pag) = self.address.read(rec_ref, &segment.into())? {
            page = pag;
        } else {
            return Ok(None);
        }
        let mut buffer;
        {
            let mut pg = self.allocator.load_page(page)?;
            let len = pg.read_u64::<BigEndian>()?;
            buffer = Vec::<u8>::with_capacity(len as usize);
            pg.take(len).read_to_end(&mut buffer)?;
        }
        return Ok(Some(buffer));
    }

    pub fn scan_records(&self, segment: &str) -> PRes<RecordScanner> {
        let seg = segment.into();
        Ok(RecordScanner::new(&self, &seg, self.address.scan(&seg)?))
    }

    pub fn update_record(&self, tx: &mut Transaction, segment: &str, rec_ref: &RecRef, rec: &Vec<u8>) -> PRes<()> {
        let ref allocator = self.allocator;
        let ref journal = self.journal;
        if let Some(old_record) = self.read_ref(tx, &segment.into(), rec_ref)? {
            let len = rec.len();
            let allocation_exp = exp_from_size((len + 8) as u64);
            let page = allocator.allocate(allocation_exp)?;
            tx.add_update(journal, &segment.into(), &rec_ref, page, old_record)?;
            {
                let mut pg = allocator.write_page(page)?;
                pg.write_u64::<BigEndian>(len as u64)?;
                pg.write_all(rec)?;
                allocator.flush_page(&mut pg)?;
            }
            return Ok(());
        }
        Err(PersyError::RecordNotFound)
    }

    pub fn delete_record(&self, tx: &mut Transaction, segment: &str, rec_ref: &RecRef) -> PRes<()> {
        let ref journal = self.journal;
        if let Some(old_record) = self.read_ref(tx, &segment.into(), rec_ref)? {
            tx.add_delete(journal, &segment.into(), &rec_ref, old_record)?;
            return Ok(());
        }
        Err(PersyError::RecordNotFound)
    }

    pub fn rollback(&self, mut tx: Transaction) -> PRes<()> {
        let ref allocator = self.allocator;
        let ref journal = self.journal;
        let ref address = self.address;
        tx.rollback(journal, address, allocator)
    }

    pub fn prepare_commit(&self, mut tx: Transaction) -> PRes<TransactionFinalize> {
        let ref allocator = self.allocator;
        let ref journal = self.journal;
        let ref address = self.address;
        tx.prepare_commit(journal, address, allocator)?;

        Ok(TransactionFinalize { transaction: tx })
    }

    pub fn rollback_prepared(&self, mut finalizer: TransactionFinalize) -> PRes<()> {
        let ref allocator = self.allocator;
        let ref journal = self.journal;
        let ref address = self.address;
        finalizer.transaction.rollback_prepared(journal, address, allocator)
    }

    pub fn commit(&self, mut finalizer: TransactionFinalize) -> PRes<()> {
        let ref allocator = self.allocator;
        let ref journal = self.journal;
        let ref address = self.address;
        finalizer.transaction.commit(address, journal, allocator)
    }
}


fn exp_from_size(size: u64) -> u8 {
    // Should be there a better way, so far is ok.
    let mut res: u8 = 1;
    loop {
        if size < (1 << res) {
            return res;
        }
        res += 1;
    }
}


impl From<Error> for PersyError {
    fn from(erro: Error) -> PersyError {
        let mess: &error::Error = &erro;
        // TODO: this seems to miss some detail, find another way to extract IO error message
        PersyError::IO(mess.description().into())
    }
}

impl<T> From<sync::PoisonError<T>> for PersyError {
    fn from(_: sync::PoisonError<T>) -> PersyError {
        PersyError::Lock
    }
}

impl From<str::Utf8Error> for PersyError {
    fn from(err: str::Utf8Error) -> PersyError {
        PersyError::Encoding(err)
    }
}

impl RecRef {
    pub fn new(page: u64, pos: u32) -> RecRef {
        RecRef {
            page: page,
            pos: pos,
        }
    }
}