extern crate tokio_core;
extern crate futures;
extern crate rand;
extern crate serde_json;
extern crate base64;
extern crate atomicwrites;

use std::time::Duration;
use std::io;
use std::str;
use std::rc::Rc;
use std::sync::{Arc,Mutex};
use std::error::Error;
use std::cell::{Cell,RefCell};
use std::fs::File;
use std::io::Read;

use tokio_core::reactor::Interval;
use tokio_core::reactor::Handle;
use futures::{Stream,Future,future};
use self::rand::Rng;
use std::str::FromStr;
use hyper;
use hyper::header::{Headers, ETag, EntityTag};
use hyper::client::Client;

use types::*;
use ::Config;

#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct Params {
    pub fridge_setpoint: f32,
    pub fridge_difference: f32,
    pub overshoot_delay: u64,
    pub overshoot_factor: f32,
    pub disabled: bool,
    pub nowort: bool,
    pub fridge_range_lower: f32,
    pub fridge_range_upper: f32,
}

impl Params {
    pub fn defaults() -> Params {
        Params {
            fridge_setpoint: 16.0,
            fridge_difference: 0.2,
            overshoot_delay: 720, // 12 minutes
            overshoot_factor: 1.0,
            disabled: false,
            nowort: false,
            fridge_range_lower: 3.0,
            fridge_range_upper: 3.0,
            }
    }

    fn try_load(filename: &str) -> Result<Params, TemplogError> {
        let mut s = String::new();
        File::open(filename)?.read_to_string(&mut s)?;
        Ok(serde_json::from_str(&s)?)
    }

    pub fn load(config: &Config) -> Params {
        Self::try_load(&config.PARAMS_FILE)
            .unwrap_or_else(|_| Params::defaults())
    }
}

#[derive(Deserialize, Debug)]
struct Response {
    // sent as an opaque etag: header. Has format "epoch-nonce",
    // responses where the epoch do not match ParamWaiter::epoch are dropped
    etag: String,
    params: Params,
}

#[derive(Clone)]
pub struct ParamWaiter {
    limitlog: NotTooOften,
    // last_etag is used for long-polling.
    last_etag: RefCell<String>,
    epoch: String,

    config: Config,
    handle: Handle,
}

const LOG_MINUTES: u64 = 15;
const MAX_RESPONSE_SIZE: usize = 10000;
const TIMEOUT_MINUTES: u64 = 5;

impl ParamWaiter {
    fn make_request(&self) -> hyper::client::FutureResponse {
        let uri = self.config.SETTINGS_URL.parse().expect("Bad SETTINGS_URL in config");
        let mut req = hyper::client::Request::new(hyper::Method::Get, uri);
        {
            let mut headers = req.headers_mut();
            headers.set(hyper::header::ETag(EntityTag::new(false, self.last_etag.borrow().clone())));
        }
        hyper::client::Client::new(&self.handle).request(req)
        // XXX how do we do this?
        // req.timeout(Duration::new(TIMEOUT_MINUTES * 60, 0)).unwrap();
    }


    fn handle_response(&self, buf : hyper::Chunk, status: hyper::StatusCode) -> Result<Params, TemplogError> {
        let text = String::from_utf8_lossy(buf.as_ref());
        match status {
            hyper::StatusCode::Ok => {
                // new params
                let r: Response = serde_json::from_str(&text)?;
                let mut le = self.last_etag.borrow_mut();
                *le = r.etag;

                // update params if the epoch is correct
                if let Some(e) = le.split('-').next() {
                    if e == &self.epoch {
                        self.write_params(&r.params);
                        return Ok(r.params);
                    }
                }
                Err(TemplogError::new(&format!("Bad epoch from server '{}' expected '{}'", *le, self.epoch)))
            }
            hyper::StatusCode::NotModified => {
                // XXX this isn't really an error
                Err(TemplogError::new("304 unmodified (long polling timeout at the server)"))
            },
            _ => {
                Err(TemplogError::new(&format!("Wrong server response code {}: {}", status.as_u16(), text)))
            },
        }
    }

    fn write_params(&self, params: &Params) {
        let p = atomicwrites::AtomicFile::new(&self.config.PARAMS_FILE, atomicwrites::AllowOverwrite);
        p.write(|f| {
            serde_json::to_writer(f, params)
        });
    }

    fn step(rself: Rc<Self>) -> Box<Future<Item=Params, Error=TemplogError>> {
        let resp = rself.make_request();

        let s = resp.map_err(|e| TemplogError::new_hyper("response", e))
                    .and_then(move |r| {
                        let status = r.status();
                        r.body().concat2()
                            .map_err(|e| TemplogError::new_hyper("body", e))
                            .and_then(move |b| {
                                rself.handle_response(b, status)
                            })
                    });
        Box::new(s)
    }

    fn new(config: &Config, handle: &Handle) -> Self {
        let mut b = [0u8; 15]; // 15 bytes -> 20 characters base64
        rand::OsRng::new().expect("Opening system RNG failed").fill_bytes(&mut b);
        let epoch = base64::encode(&b);
        ParamWaiter {
            limitlog: NotTooOften::new(LOG_MINUTES*60),
            last_etag: RefCell::new(String::new()),
            epoch: epoch,
            config: config.clone(),
            handle: handle.clone(),
        }
    }

    pub fn stream(config: &Config, handle: &Handle) -> Box<Stream<Item=Params, Error=TemplogError>> {
        let rcself = Rc::new(ParamWaiter::new(config, handle));

        let dur = Duration::from_millis(4000);
        let i = Interval::new(dur, &rcself.handle).unwrap()
            .map_err(|e| TemplogError::new_io("interval failed", e))
            .and_then(move |()| {
                Self::step(rcself.clone())
            });

        // TODO use consume_errors() instead once "impl trait" is stable
        //Box::new(consume_errors(i))
        let f = i.then(|r| {
            match r {
                Ok(v) => Ok(Some(v)),
                Err(e) => {
                    debug!("Params stream error: {}", e);
                    Ok(None)
                }
            }
        })
        .filter_map(|p| p);
        Box::new(f)
    }
}