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};
use std::io;
use byteorder::{ReadBytesExt, WriteBytesExt, BigEndian};

use allocator::Allocator;
use address::Address;
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;
use std::fmt;
use record_scanner::{RecordScanner, RecordScannerTx};

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 {
    config: Arc<Config>,
    journal: Journal,
    address: Address,
    allocator: Arc<Allocator>,
}

/// prepared transaction state 
pub struct TransactionFinalize {
    transaction: Transaction,
}

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

#[derive(Debug)]
pub enum PersyError {
    IO(io::Error),
    Err(String),
    Encoding(str::Utf8Error),
    VersionNotLastest,
    RecordNotFound,
    SegmentNotFound,
    SegmentAlreadyExists,
    CannotDropSegmentCreatedInTx,
    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 {
            config: config.clone(),
            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, self.config.tx_strategy())?)
    }

    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);
                }
            }
        }
        let segment_id = self.address.create_temp_segment(&segment.into())?;
        tx.add_create_segment(&self.journal, segment, segment_id)?;
        Ok(())

    }

    pub fn drop_segment(&self, tx: &mut Transaction, segment: &str) -> PRes<()> {
        let (_, segment_id) = self.check_segment_tx(tx, &segment.into())?;
        tx.add_drop_segment(&self.journal, segment, segment_id)?;
        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 => Ok(false),
            CREATED(_) => Ok(true),
            NONE => self.address.exists_segment(&segment.into()),
        }
    }

    /// 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, u32)> {
        match tx.exists_segment(segment) {
            DROPPED => Err(PersyError::SegmentNotFound),
            CREATED(segment_id) => Ok((true, segment_id)),
            NONE => {
                if let Some(id) = self.address.segment_id(segment)? {
                    Ok((false, id))
                } else {
                    Err(PersyError::SegmentNotFound)
                }
            }
        }
    }

    pub fn insert_record(&self, tx: &mut Transaction, segment: &str, rec: &Vec<u8>) -> PRes<RecRef> {
        let (in_tx, segment_id) = 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 in_tx {
            address.allocate_temp(segment_id)?
        } else {
            address.allocate(segment_id)?
        };
        tx.add_insert(journal, segment_id, &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_segment(&self, tx: &Transaction, segment_id: u32, rec_ref: &RecRef) -> PRes<Option<(u64, u16, u32)>> {
        match tx.read(rec_ref) {
            TxRead::RECORD(rec) => return Ok(Some((rec.0, rec.1, segment_id))),
            TxRead::DELETED => return Ok(None),
            TxRead::NONE => Ok(self.address.read(rec_ref, segment_id)?.map(|(pos, version)| (pos, version, segment_id))), 
        }
    }

    fn read_ref(&self, tx: &Transaction, segment: &String, rec_ref: &RecRef) -> PRes<Option<(u64, u16, u32)>> {
        let (_, segment_id) = self.check_segment_tx(tx, segment)?;
        self.read_ref_segment(tx, segment_id, rec_ref)
    }

    fn read_page(&self, page: u64) -> PRes<Vec<u8>> {
        let mut pg = self.allocator.load_page(page)?;
        let len = pg.read_u64::<BigEndian>()?;
        let mut buffer = Vec::<u8>::with_capacity(len as usize);
        pg.take(len).read_to_end(&mut buffer)?;
        return Ok(buffer);
    }

    pub fn read_record_scan_tx(&self, tx: &Transaction, segment_id: u32, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
        if let Some(page) = self.read_ref_segment(tx, segment_id, rec_ref)? {
            return Ok(Some(self.read_page(page.0)?));
        }
        Ok(None)
    }

    pub fn read_record_tx(&self, tx: &mut Transaction, segment: &str, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
        let seg = segment.into();
        if let Some(page) = self.read_ref(tx, &seg, rec_ref)? {
            tx.add_read(&self.journal, page.2, rec_ref, page.1)?;
            return Ok(Some(self.read_page(page.0)?));
        }
        Ok(None)
    }

    pub fn read_record(&self, segment: &str, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
        if let Some(segment_id) = self.address.segment_id(&segment.into())? {
            self.read_record_scan(segment_id, rec_ref)
        } else {
            Err(PersyError::SegmentNotFound)
        }
    }

    pub fn read_record_scan(&self, segment_id: u32, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
        if let Some((page, _)) = self.address.read(rec_ref, segment_id)? {
            Ok(Some(self.read_page(page)?))
        } else {
            Ok(None)
        }
    }

    pub fn scan_records(&self, segment: &str) -> PRes<RecordScanner> {
        let segment_id;
        if let Some(id) = self.address.segment_id(&segment.into())? {
            segment_id = id;
        } else {
            return Err(PersyError::SegmentNotFound);
        }
        Ok(RecordScanner::new(&self, segment_id, self.address.scan(segment_id)?))
    }


    pub fn scan_records_tx<'a>(&'a self, tx: &'a Transaction, segment: &str) -> PRes<RecordScannerTx<'a>> {
        let seg = segment.into();
        let (_, segment_id) = self.check_segment_tx(tx, &seg)?;
        Ok(RecordScannerTx::<'a>::new(&self, &tx, segment_id, self.address.scan(segment_id)?))
    }

    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) = 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, old.2, &rec_ref, page, old.0, old.1)?;
            {
                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) = self.read_ref(tx, &segment.into(), rec_ref)? {
            tx.add_delete(journal, old.2, &rec_ref, old.0, old.1)?;
            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<io::Error> for PersyError {
    fn from(erro: io::Error) -> PersyError {
        // TODO: this seems to miss some detail, find another way to extract IO error message
        PersyError::IO(erro)
    }
}

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,
        }
    }
}

impl fmt::Display for PersyError {
    fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
        write!(f, "{}", (self as &error::Error).description())
    }
}

impl PartialEq for PersyError {
    fn eq(&self, other: &PersyError) -> bool {

        match *self {

            PersyError::IO(ref pio) => {
                match *other {
                    PersyError::IO(ref io) => (pio as &error::Error).description().eq((io as &error::Error).description()),
                    _ => false,
                }
            }
            PersyError::Err(ref perr) => {
                match *other {
                    PersyError::Err(ref err) => perr.eq(err),
                    _ => false,
                }
            }
            PersyError::Encoding(ref penc) => {
                match *other {
                    PersyError::Encoding(ref enc) => penc.eq(enc),
                    _ => false,
                }
            }
            PersyError::RecordNotFound => {
                match *other {
                    PersyError::RecordNotFound => true,
                    _ => false,
                }
            }
            PersyError::SegmentNotFound => {
                match *other {
                    PersyError::SegmentNotFound => true,
                    _ => false,
                }
            }
            PersyError::SegmentAlreadyExists => {
                match *other {
                    PersyError::SegmentAlreadyExists => true,
                    _ => false,
                }
            }

            PersyError::VersionNotLastest => {
                match *other {
                    PersyError::VersionNotLastest => true,
                    _ => false,
                }
            }

            PersyError::Lock => {
                match *other {
                    PersyError::Lock => true,
                    _ => false,
                }
            }
            PersyError::CannotDropSegmentCreatedInTx => {
                match *other {
                    PersyError::CannotDropSegmentCreatedInTx => true,
                    _ => false,
                }
            }
        }
    }
}
impl error::Error for PersyError {
    fn description(&self) -> &str {
        match *self {
            PersyError::IO(ref io) => &io.description(),
            PersyError::Err(ref err) => &err,
            PersyError::Encoding(ref enc) => &enc.description(),
            PersyError::RecordNotFound => "Record Not Found",
            PersyError::SegmentNotFound => "Segment Not Found",
            PersyError::SegmentAlreadyExists => "Segment Already Exist",
            PersyError::Lock => "Error Locking resource",
            PersyError::VersionNotLastest => "Persistent version is more recent that transaction version",
            PersyError::CannotDropSegmentCreatedInTx => "Impossible to drop a segment in the same transaction where is created",
        }
    }
}