use std::process::Stdio;
use std::str::from_utf8;

use tokio::select;
use tokio::sync::{mpsc,watch};
use tokio::time::{delay_for,Duration};
use tokio::process::{Child,ChildStdout};
use tokio::io::AsyncReadExt;
use std::default::Default;

use futures_util::future;
use futures_util::future::FutureExt;

use std::iter;

use anyhow::anyhow;

use apigpio::Connection;
use apigpio::Level::*;

use picollar::ksmode::*;
use picollar::ksmode::Mode::*;
use picollar::ledpins::*;
use picollar::local::*;
use picollar::local::DemandDelta::Delta;
use picollar::local::DemandDeltaSign::*;
use picollar::config::{Config,NO_MORE};
use picollar::RecvError;

type E = anyhow::Error;

pub struct VariableDemandDeltaSender {
  sent : OptionSeverity,
  output : mpsc::Sender<DemandDelta>,
}

impl VariableDemandDeltaSender {
  pub fn new(output : mpsc::Sender<DemandDelta>) -> Self {
    VariableDemandDeltaSender { sent : None.into(), output }
  }
  pub async fn send(&mut self, wanted : OptionSeverity) -> Result<(),E> {
    if self.sent != wanted {
      if let Some(decrement) = *self.sent {
        self.output.send(Delta(decrement,Down)).await?;
        self.sent = None.into()
      }
      if let Some(increment) = *wanted {
        self.output.send(Delta(increment,Up)).await?;
        self.sent = wanted;
      }
    }
    Ok(())
  }
}

impl Drop for VariableDemandDeltaSender {
  fn drop(&mut self) { assert!(self.sent.is_none()); }
}

type Seq = u16;

type Timeout = u16;
const NO_TIMEOUT : u16 = 0xffff;

type Stdout = ChildStdout;

struct Running {
  child : Child,
  stdout : Stdout,
  next_seq : Seq,
}

const PACKET_LEN : usize = 15;
type RawPacket = [u8;PACKET_LEN];
// webserver's packets, fixed length (15 bytes) but also ascii
//   ssss llll tttt\n
//   ^seq ^sev ^timeout
// llll is SeverityKeyword or "none" (see local.rs FromStr impl)
// others u16 encoded as hex.
// severity ffff means None
// timeout  ffff means None, used iff severity is ffff
#[derive(Debug)]
struct CheckedPacket {
  severity : OptionSeverity,
  timeout : Timeout,
}

async fn read_packet(stdout : &mut Stdout) -> Result<RawPacket,E> {
  let mut buf = [0;PACKET_LEN];
  stdout.read_exact(&mut buf).await?;
  Ok(buf)
}

fn check_packet(buf : &[u8], next_seq : &mut Seq) -> Result<CheckedPacket,E> {
  let mut strs = from_utf8(buf)?.split_ascii_whitespace()
    .map(|s| Ok(s))
    .chain(iter::once(Err(anyhow!("too few words in packet"))));

  let num = |s : Option<Result<&str,E>>| {
    <Result<_,E>>::Ok( u16::from_str_radix(s.unwrap()?,16)? )
  };

  let seq = num(strs.next())?;
  if seq != *next_seq { Err(anyhow!("out of sequence"))? }
  *next_seq += 1;

  let severity = strs.next().unwrap()?.parse()?;

  let timeout = num(strs.next())?;
  if severity == None.into() && timeout != NO_TIMEOUT {
    Err(anyhow!("timeout set on no severity"))?
  }
  Ok(CheckedPacket { severity, timeout })
}

async fn startup(cfg : &Config, running : &mut Option<Running>)
                 -> Result<(),E> {
  let mut cmd = cfg.command_simple(&["web_ui","command"],"web-ui",&NO_MORE)?;
//  let mut cmd = cfg.command_simple("web_ui.command","web-ui",&NO_MORE)?;
  cmd.stdout(Stdio::piped());
  let mut child = cmd.spawn()?;
  let stdout = child.stdout.take().unwrap();
  *running = Some(Running{ child, stdout, next_seq : 0 });
  Ok(())
}

async fn processing(running : &mut Option<Running>,
                    vdds : &mut VariableDemandDeltaSender)
                    -> Result<(),E> {
  let mut timeout = NO_TIMEOUT;

  loop {
    let Running { ref mut child, ref mut stdout, ref mut next_seq } =
      running.as_mut().unwrap();

    select!{
      got = read_packet(stdout) => {
        let ok = check_packet(&got?, next_seq)?;
        println!("application: webtalk: received {:?}", &ok);
        vdds.send(ok.severity).await?;
        timeout = ok.timeout;
      },
      
      () = async {
        if timeout == NO_TIMEOUT {
          future::pending().await
        } else {
          delay_for(Duration::from_millis(timeout.into())).await
        }
      } => {
        println!("application: webtalk: timeout tripped");
        vdds.send(None.into()).await?;
        timeout = NO_TIMEOUT;
      },

      status = child => {
        let status = status?;
        *running = None;
        Err(anyhow!("application terminated! {:?}", status))?;
      }
    }
  }
}

async fn shutdown(vdds : &mut VariableDemandDeltaSender,
                  running : &mut Option<Running>) -> Result<(),E> {
  vdds.send(None.into()).await?;
  if let Some(Running { ref mut child, .. }) = running {
    println!("application: killing webserver");
    child.kill()?;
    child.await?;
  }
  *running = None;
  Ok(())
}

pub async fn mainloop(pi : Connection, cfg : Config,
             mut keyswitch : watch::Receiver<Mode>,
             output : mpsc::Sender<DemandDelta>) -> Result<(),E> {
  let mut vdds = VariableDemandDeltaSender::new(output);
  let mut mode = keyswitch.recv().await.ok_or(RecvError{})?;
  loop {
    println!("application: webtalk: implementing mode {:?}", mode);

    let mut running = None;
    select!{
      r = async {
        if let Err(e) = match mode {
          Local | Dev => future::pending().await,
          Net => async {
            startup(&cfg, &mut running).await?;
            processing(&mut running, &mut vdds).await?;
            <Result<(),E>>::Ok(())
          }.await,
        } {
          let info = if let Some(Running{ ref mut child, ..}) = running {
            match child.now_or_never() {
              Some(Ok(status)) => format!("status {:?}",status),
              Some(Err(we))    => format!("wait error {:?}",we),
              None             => "still running".to_owned(),
            }
          } else {
            "ended".to_owned()
          };
          println!("application: webtalk: failed: {}, {:?}",info,&e);
          flashled(&pi, [H;3], FLASH_SPECIAL).await?;
        }
        <Result<(),E>>::Ok(())
      } => r?,

      r = keyswitch.recv() => {
        mode = r.ok_or(RecvError{})?;
      }
    }

    shutdown(&mut vdds, &mut running).await?;
  }
}