use std::time::Duration;

use std::str;



use std::cell::{RefCell};
use std::fs::File;
use std::io::Read;

use serde::{Serialize,Deserialize};

use rand::rngs::{OsRng};
use rand::{RngCore};


use hyper;

// for try_concat()
use futures::stream::TryStreamExt;
use futures::executor::block_on;

use riker::actors::*;

use super::types::*;
use super::config::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: true,
            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())
    }

}

pub struct ParamWaiter {
    limitlog: NotTooOften,
    // last_etag is used for long-polling.
    last_etag: RefCell<String>,
    epoch: String,
    notify: ChannelRef<Params>,

    config: Config,
}

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

impl ParamWaiter {
    pub fn new_actor((config, notify): (Config, ChannelRef<Params>)) -> Self {
        Self::new(config, notify)
    }

    pub fn new(config: Config, notify: ChannelRef<Params>) -> Self {
        let mut b = [0u8; 15]; // 15 bytes -> 20 characters base64
        OsRng.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,
            notify: notify,
        }
    }

    async fn wait_updates(uri: &str, etag: &str) -> Result<(hyper::Chunk, hyper::StatusCode), TemplogError> {
        debug!("wait_updates {}", uri);
        let req = hyper::Request::get(uri)
            .header(hyper::header::ETAG, etag)//*self.last_etag.borrow())
            .body(hyper::Body::from("")).unwrap();

        // TODO timeout?
        let resp = hyper::Client::new().request(req).await?;
        let status = resp.status();
        let chunk = resp.into_body().try_concat().await?;

        Ok((chunk, status))
    }

    fn handle_response(&self, buf : hyper::Chunk, status: hyper::StatusCode) -> Result<Params, TemplogError> {

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

        match status {
            hyper::StatusCode::OK => {
                // new params
                let r: Response = serde_json::from_slice(&buf)?;
                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::NOT_MODIFIED => {
                // XXX this isn't really an error. Should handle_response() return 
                // Result<Option<Params>, TemplogError> instead?
                Err(TemplogError::new("304 unmodified (long polling timeout at the server)"))
            },
            _ => {
                let text = String::from_utf8_lossy(buf.as_ref());
                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);
        let r = p.write(|f| {
            serde_json::to_writer(f, params)
        });
        if let Err(e) = r {
            // XXX notify?
            error!("Couldn't write to {}: {}", self.config.params_file, e)
        };
    }

    fn do_poll(&mut self, ctx: &Context<<Self as Actor>::Msg>) -> Result<(), TemplogError> {
        let url = self.config.settings_url.clone();
        let etag = self.last_etag.borrow().clone();
        let h = ctx.run(async move { 
            Self::wait_updates(&url, &etag).await 
        }).expect("spawn failed"); // XXX error handling
        let (chunk, stat) = block_on(h)?;
        let new_params = self.handle_response(chunk, stat)?;
        self.notify.tell(Publish{msg: new_params, topic: "params".into()}, None);
        Ok(())
    }
}

#[derive(Clone,Debug)]
pub struct PollForParams;

impl Actor for ParamWaiter {
    type Msg = PollForParams;

    fn recv(&mut self,
            ctx: &Context<Self::Msg>,
            _msg: Self::Msg,
            _sender: Sender) {
        // schedule a retry once this iteration finishes
        ctx.schedule_once(Duration::from_secs(1), ctx.myself(), None, PollForParams);

        if let Err(e) = self.do_poll(ctx) {
            self.limitlog.and_then(|| {
                warn!("Problem fetching params: {}", e)
            });
        }
    }

    fn pre_start(&mut self, ctx: &Context<Self::Msg>) {
        ctx.schedule_once(Duration::from_secs(1), ctx.myself(), None, PollForParams);
    }
}

    // pub fn stream(config: Config, handle: Handle) -> Result<(), TemplogError> {
    //     let rcself = Rc::new(ParamWaiter::new(config, handle));

    //     let dur = Duration::from_millis(4000);
    //     for _ in Interval::new(dur, &rcself.handle).unwrap() {
    //         // fetch params
    //         // TODO - skip if inflight.
    //         let r = await!(rcself.make_request()).map_err(|e| TemplogError::new_hyper("response", e))?;
    //         let status = r.status();
    //         let b = await!(r.body().concat2()).map_err(|e| TemplogError::new_hyper("body", e))?;
    //         if let Ok(params) = rcself.handle_response(b, status) {
    //             stream_yield!(params);
    //         }
    //     }
    //     Ok(())
    // }