use crate::buffer::Buffer;
use crate::buffer::BufferError;
use crate::logsink;
use crate::logsink::Logsink;
use crate::logsink::SinkError;
use crate::parse::parse_lines;
use crate::signals;
use crate::signals::SignalError;
use io::Read;
use std::io;
use std::path;
use std::path::PathBuf;
use std::process;
use std::process::ExitStatus;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use std::time::Instant;
use thiserror::Error;

const LONGLINE_CUT: usize = 512;

#[allow(unused)]
pub const DO_TRACE: bool = false;

#[allow(unused)]
pub const DO_DEBUG: bool = false || DO_TRACE;

#[allow(unused)]
#[macro_export]
macro_rules! trace {
    ($($arg:tt)*) => { if $crate::logappend::DO_TRACE { eprintln!($($arg)*) } };
}

#[allow(unused)]
#[macro_export]
macro_rules! debug {
    ($($arg:tt)*) => { if $crate::logappend::DO_DEBUG { eprintln!($($arg)*) } };
}

#[allow(unused)]
#[macro_export]
macro_rules! error {
    ($($arg:tt)*) => { if true { eprintln!($($arg)*) } };
}

#[derive(Debug, Error)]
pub enum LogAppendError {
    #[error("IOError({0})")]
    IO(#[from] io::Error),
    #[error("SinkError({0})")]
    Sink(#[from] SinkError),
    #[error("BufferError({0})")]
    Buffer(#[from] BufferError),
    #[error("LogicError({0})")]
    Logic(String),
    #[error("SystemSignalError({0})")]
    SystemSignal(#[from] SignalError),
}

#[derive(Debug, Error)]
pub enum BufDropError {
    #[error("AdvanceError({0})")]
    Advance(#[from] BufferError),
}

#[test]
fn buf_drop_error() {
    let e: Result<(), _> = Err(BufferError::RpAdv(6, 4, 3)).map_err(BufDropError::from);
    let s = e.unwrap_err().to_string();
    assert_eq!("AdvanceError(ReadPointerAdvanceError(6, 4, 3))", s);
}

fn buf_drop_until_newline(buf: &mut Buffer) -> Result<usize, LogAppendError> {
    let bb = buf.readable();
    let mut n = bb.len();
    for (i, b) in bb.iter().enumerate() {
        if *b == 0xa {
            n = 1 + i;
            break;
        }
    }
    buf.advance(n)?;
    Ok(n)
}

fn push_lines_to_sinks(lines: &[&[u8]], sinks: &mut [&mut Logsink]) -> Result<(), LogAppendError> {
    for sink in sinks.iter_mut() {
        sink.push_lines(&lines)?;
    }
    Ok(())
}

fn output_append_running<INP: Read>(
    dir: PathBuf,
    total_max: usize,
    file_max: usize,
    mut inp: INP,
    prefix: &str,
) -> Result<(), LogAppendError> {
    let mut sink_all = Logsink::new(
        dir.clone(),
        logsink::Filter::All,
        prefix,
        total_max,
        file_max,
    )?;
    let mut buf = Buffer::new(1024 * 16);
    loop {
        let bmut = buf.writable();
        let nread = inp.read(bmut)?;
        if nread > bmut.len() {
            error!("ERROR read returned more than buffer space");
            return Err(LogAppendError::Logic(
                "read returned more than buffer space".into(),
            ));
        }
        buf.adv_wp(nread)?;
        sink_all.push_data(buf.readable())?;
        buf.reset();
        sink_all.flush()?;
        if nread == 0 {
            break;
        }
    }
    Ok(())
}

fn log_append_running<INP: Read>(
    dir: PathBuf,
    total_max: usize,
    file_max: usize,
    mut inp: INP,
) -> Result<(), LogAppendError> {
    let mut sink_all = Logsink::new(
        dir.clone(),
        logsink::Filter::All,
        "all",
        total_max,
        file_max,
    )?;
    let mut sink_info = Logsink::new(dir, logsink::Filter::Info, "info", total_max, file_max)?;
    let mut sinks = [&mut sink_all, &mut sink_info];
    let mut buf = Buffer::new(1024 * 16);
    loop {
        if buf.free_len() == 0 {
            let x = buf_drop_until_newline(&mut buf)?;
            debug!("[BUF-FULL-DROP {}]", x);
        }
        let bmut = buf.writable();
        let nread = inp.read(bmut)?;
        if nread > bmut.len() {
            error!("ERROR read returned more than buffer space");
            return Err(LogAppendError::Logic(
                "ERROR read returned more than buffer space".into(),
            ));
        }
        buf.adv_wp(nread)?;
        trace!("[READ {:5}  HAVE {:5}]", nread, buf.len());
        let (lines, n2) = parse_lines(buf.readable());
        trace!("[PARSED-LINES {}]", lines.len());
        push_lines_to_sinks(&lines, &mut sinks)?;
        buf.advance(n2)?;
        if buf.len() > LONGLINE_CUT {
            debug!("[TRUNCATED]");
            let lines = [buf.readable()[..LONGLINE_CUT].as_ref()];
            push_lines_to_sinks(&lines, &mut sinks)?;
            let x = buf_drop_until_newline(&mut buf)?;
            debug!("[TRUNC-DROP {}]", x);
        }
        if nread == 0 {
            if buf.len() > 0 {
                let lines = [buf.readable()];
                push_lines_to_sinks(&lines, &mut sinks)?;
            }
            break;
        }
        for sink in sinks.iter_mut() {
            sink.flush()?;
        }
    }
    Ok(())
}

pub enum InputStreamType {
    FeedThrough,
    LogSplit,
}

pub fn logappend_wrap(
    dirname: &str,
    total_max: usize,
    file_max: usize,
    exe: String,
    args: Vec<String>,
    stdout_stream_type: InputStreamType,
    stderr_stream_type: InputStreamType,
) -> Result<ExitStatus, LogAppendError> {
    debug!("exe {exe:?}");
    debug!("args {args:?}");
    let dir = path::PathBuf::from(dirname);

    signals::init();
    signals::ignore_signal(libc::SIGINT)?;
    signals::ignore_signal(libc::SIGTERM).unwrap();
    signals::ignore_signal(libc::SIGHUP).unwrap();

    // TODO make sure that this chikd process receives our std input.

    let mut proc = process::Command::new(exe)
        .args(args)
        .stdout(process::Stdio::piped())
        .stderr(process::Stdio::piped())
        .spawn()
        .unwrap();
    let chout = proc.stdout.take().unwrap();
    let cherr = proc.stderr.take().unwrap();

    let count_running = Arc::new(AtomicUsize::new(2));

    // TODO handle possible error from thread
    let jh1 = thread::Builder::new()
        .spawn({
            let count_running = count_running.clone();
            let dir = dir.to_owned();
            move || {
                match stdout_stream_type {
                    InputStreamType::FeedThrough => {
                        output_append_running(dir, total_max, file_max, chout, "stdout").unwrap();
                    }
                    InputStreamType::LogSplit => {
                        log_append_running(dir, total_max, file_max, chout).unwrap();
                    }
                }
                count_running.fetch_sub(1, Ordering::SeqCst);
            }
        })
        .unwrap();
    let jh2 = thread::Builder::new()
        .spawn({
            let count_running = count_running.clone();
            let dir = dir.to_owned();
            move || {
                match stderr_stream_type {
                    InputStreamType::FeedThrough => {
                        output_append_running(dir, total_max, file_max, cherr, "stderr").unwrap();
                    }
                    InputStreamType::LogSplit => {
                        log_append_running(dir, total_max, file_max, cherr).unwrap();
                    }
                }
                count_running.fetch_sub(1, Ordering::SeqCst);
            }
        })
        .unwrap();

    let recv = signals::receiver();
    let deadline = Instant::now() + Duration::from_millis(30000);
    loop {
        let h = count_running.load(Ordering::SeqCst);
        trace!("msg loop {h}");
        let tsnow = Instant::now();
        if false && tsnow >= deadline {
            debug!("msg timeout break");
            break;
        }
        let _timeout = deadline - tsnow;
        let timeout = Duration::from_millis(100);
        match recv.recv_timeout(timeout) {
            Ok(e) => {
                debug!("msg: {e}");
            }
            Err(_e) => {
                if count_running.load(Ordering::SeqCst) == 0 {
                    break;
                }
            }
        }
    }
    let ec = proc.wait().unwrap();
    jh1.join().unwrap();
    jh2.join().unwrap();
    Ok(ec)
}

// Only as playground.
pub fn play_signals() -> Result<(), LogAppendError> {
    let count_running = Arc::new(AtomicUsize::new(0));
    signals::init();
    if false {
        signals::set_signal_handler(libc::SIGUSR1).unwrap();
    }
    signals::ignore_signal(libc::SIGINT).unwrap();
    signals::ignore_signal(libc::SIGTERM).unwrap();
    signals::ignore_signal(libc::SIGHUP).unwrap();
    let proc = process::Command::new("/bin/bash")
        .args(&[
            "-c",
            "while true; do date; >&2 echo stderr; sleep 0.3; done",
        ])
        .stdout(process::Stdio::piped())
        .stderr(process::Stdio::piped())
        .spawn()
        .unwrap();
    count_running.fetch_add(2, Ordering::SeqCst);
    let mut chout = proc.stdout.unwrap();
    let mut cherr = proc.stderr.unwrap();
    let jh1 = thread::Builder::new()
        .spawn({
            let count_running = count_running.clone();
            move || {
                let mut buf = vec![0; 256];
                loop {
                    debug!("thread-1 read");
                    let n = chout.read(&mut buf).unwrap();
                    debug!("thread-1 read done");
                    if n == 0 {
                        debug!("thread-1 break");
                        break;
                    } else {
                        signals::sender()
                            .send(format!(
                                "thread-1 {}",
                                std::str::from_utf8(&buf[..n]).unwrap()
                            ))
                            .unwrap();
                    }
                }
                count_running.fetch_sub(1, Ordering::SeqCst);
            }
        })
        .unwrap();
    let jh2 = thread::Builder::new()
        .spawn({
            let count_running = count_running.clone();
            move || {
                let mut buf = vec![0; 256];
                loop {
                    debug!("thread-2 read");
                    let n = cherr.read(&mut buf).unwrap();
                    debug!("thread-2 read done");
                    if n == 0 {
                        debug!("thread-2 break");
                        break;
                    } else {
                        signals::sender()
                            .send(format!(
                                "thread-2 {}",
                                std::str::from_utf8(&buf[..n]).unwrap()
                            ))
                            .unwrap();
                    }
                }
                count_running.fetch_sub(1, Ordering::SeqCst);
            }
        })
        .unwrap();
    if false {
        thread::sleep(Duration::from_millis(1000));
        debug!("sending USR1");
        unsafe { libc::kill(0, libc::SIGUSR1) };
        debug!("sending USR1 done");
    }
    let recv = signals::receiver();
    let deadline = Instant::now() + Duration::from_millis(30000);
    loop {
        let tsnow = Instant::now();
        if tsnow >= deadline {
            debug!("msg timeout break");
            break;
        }
        let _timeout = deadline - tsnow;
        let timeout = Duration::from_millis(100);
        match recv.recv_timeout(timeout) {
            Ok(e) => {
                debug!("msg: {e}");
            }
            Err(_e) => {
                if count_running.load(Ordering::SeqCst) == 0 {
                    break;
                }
            }
        }
    }
    debug!("await join handles");
    jh1.join().unwrap();
    jh2.join().unwrap();
    debug!("DONE");
    Ok(())
}
