use std::fs;
use std::io::{Read, Seek, SeekFrom, Write};
use std::thread;
use std::time::{Duration, Instant};

use crate::base64_encode;
use crate::process::{ProcessDir, ProcessState, is_valid_process_id, remote_base, unix_timestamp};
use crate::protocol::{ProcessSummary, Response};

const WRITE_STDIN_TIMEOUT: Duration = Duration::from_secs(30);

fn invalid_process_id(id: &str) -> Response {
    Response::error(format!("invalid process ID: {id}"))
}

/// Get process status with self-healing: if status says "running" but process
/// is dead, update to "exited(unknown)".
pub fn status(id: &str) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);
    let state = match pdir.read_status() {
        Ok(s) => s,
        Err(_) => return Response::error(format!("process not found: {id}")),
    };

    // Self-healing: check if process is actually alive
    let state = if matches!(state, ProcessState::Running) {
        if let Ok(Some(pid)) = pdir.read_pid() {
            let alive = unsafe { libc::kill(pid as i32, 0) } == 0;
            if !alive {
                let runner_alive = pdir
                    .read_runner_pid()
                    .ok()
                    .flatten()
                    .is_some_and(|rp| unsafe { libc::kill(rp as i32, 0) } == 0);
                if !runner_alive {
                    let _ = fs::write(pdir.status_path(), "exited(unknown)");
                    let _ = fs::write(pdir.ended_path(), unix_timestamp().to_string());
                    ProcessState::ExitedUnknown
                } else {
                    state
                }
            } else {
                state
            }
        } else {
            state
        }
    } else {
        state
    };

    Response::Status {
        id: id.to_string(),
        state: state.to_string(),
        cmd: pdir.read_cmd().unwrap_or_default(),
        started: pdir.read_started().unwrap_or(0),
        ended: pdir.read_ended().unwrap_or(None),
        stdout_size: pdir.stream_size("stdout"),
        stderr_size: pdir.stream_size("stderr"),
    }
}

/// Default read limit: 1 MiB.
const DEFAULT_READ_LIMIT: u64 = 1024 * 1024;

/// Read process output (stdout or stderr) with optional byte offset and limit.
pub fn read_output(id: &str, stream: &str, offset: Option<u64>, limit: Option<u64>) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);
    let path = match stream {
        "stdout" => pdir.stdout_path(),
        "stderr" => pdir.stderr_path(),
        _ => return Response::error(format!("invalid stream: {stream}")),
    };

    let file_size = fs::metadata(&path).map(|m| m.len()).unwrap_or(0);

    let mut file = match fs::File::open(&path) {
        Ok(f) => f,
        Err(_) => return Response::error(format!("process not found: {id}")),
    };

    let offset = offset.unwrap_or(0);
    if offset > 0
        && let Err(e) = file.seek(SeekFrom::Start(offset))
    {
        return Response::error(format!("seek failed: {e}"));
    }

    let limit = limit.unwrap_or(DEFAULT_READ_LIMIT);
    let mut data = vec![0u8; limit.min(file_size.saturating_sub(offset)) as usize];
    let bytes_read = match file.read(&mut data) {
        Ok(n) => n,
        Err(e) => return Response::error(format!("read failed: {e}")),
    };
    data.truncate(bytes_read);

    Response::Output {
        data: base64_encode(&data),
        offset,
        size: file_size,
    }
}

/// Get the byte size of a stream file.
pub fn size(id: &str, stream: &str) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);
    let sz = pdir.stream_size(stream);
    Response::Output {
        data: String::new(),
        offset: 0,
        size: sz,
    }
}

/// Write input to the process's stdin FIFO.
///
/// If `raw` is false (default), a newline is appended. If `raw` is true, the
/// input is sent as-is. Uses O_NONBLOCK to avoid hanging if the process has
/// already exited.
pub fn write_stdin(id: &str, input: &str, raw: bool) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);
    let fifo = pdir.stdin_pipe_path();

    if !pdir.dir.exists() {
        return Response::error(format!("process not found: {id}"));
    }

    if let Ok(state) = pdir.read_status()
        && !matches!(state, ProcessState::Running)
    {
        return Response::error(format!("process already exited: {id}"));
    }

    let fifo_cstr = match std::ffi::CString::new(fifo.to_str().unwrap_or("")) {
        Ok(c) => c,
        Err(e) => return Response::error(format!("invalid path: {e}")),
    };
    let fd = unsafe { libc::open(fifo_cstr.as_ptr(), libc::O_WRONLY | libc::O_NONBLOCK) };
    if fd < 0 {
        let err = std::io::Error::last_os_error();
        return Response::error(format!("failed to open stdin pipe: {err}"));
    }

    let data = if raw {
        input.as_bytes().to_vec()
    } else {
        format!("{input}\n").into_bytes()
    };

    let write_result = write_all_nonblocking(fd, &data);
    unsafe { libc::close(fd) };

    match write_result {
        Ok(bytes) => Response::Written { bytes },
        Err(err) => Response::error(format!("write failed: {err}")),
    }
}

fn write_all_nonblocking(fd: libc::c_int, data: &[u8]) -> std::io::Result<usize> {
    let started = Instant::now();
    let mut offset = 0;

    while offset < data.len() {
        let written = unsafe {
            libc::write(
                fd,
                data[offset..].as_ptr() as *const libc::c_void,
                data.len() - offset,
            )
        };

        if written > 0 {
            offset += written as usize;
            continue;
        }

        if written == 0 {
            return Err(std::io::Error::new(
                std::io::ErrorKind::WriteZero,
                "stdin pipe write returned zero bytes",
            ));
        }

        let err = std::io::Error::last_os_error();
        match err.kind() {
            std::io::ErrorKind::Interrupted => {}
            std::io::ErrorKind::WouldBlock => {
                if started.elapsed() >= WRITE_STDIN_TIMEOUT {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "timed out writing to stdin pipe",
                    ));
                }
                thread::sleep(Duration::from_millis(10));
            }
            _ => return Err(err),
        }
    }

    Ok(offset)
}

/// Kill the stdin holder process to signal EOF to the command.
/// Returns true if the holder was alive and killed, false otherwise.
fn kill_stdin_holder(pdir: &ProcessDir) -> bool {
    if let Ok(Some(holder_pid)) = pdir.read_stdin_holder_pid() {
        let alive = unsafe { libc::kill(holder_pid as i32, 0) } == 0;
        if alive {
            unsafe { libc::kill(holder_pid as i32, libc::SIGTERM) };
            return true;
        }
    }
    false
}

/// Close the stdin FIFO (send EOF to the process).
///
/// Kills the holder process, which is the last writer on the FIFO. Once the
/// holder dies, the command sees EOF on stdin.
pub fn close_stdin(id: &str) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);

    if !pdir.dir.exists() {
        return Response::error(format!("process not found: {id}"));
    }

    // Don't touch exited processes
    if let Ok(state) = pdir.read_status()
        && !matches!(state, ProcessState::Running)
    {
        return Response::error(format!("process already exited: {id}"));
    }

    if kill_stdin_holder(&pdir) {
        Response::Written { bytes: 0 } // 0 bytes = EOF marker
    } else {
        Response::error("stdin already closed or no holder found")
    }
}

/// Pipe data from own stdin (SSH channel) to a process's stdin FIFO.
/// The reverse of `follow`: reads stdin in chunks, writes to FIFO.
/// No JSON output — this is a raw data channel.
pub fn pipe_stdin(id: &str, no_close: bool) {
    if !is_valid_process_id(id) {
        std::process::exit(1);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);

    if !pdir.dir.exists() {
        std::process::exit(1);
    }

    if let Ok(state) = pdir.read_status()
        && !matches!(state, ProcessState::Running)
    {
        std::process::exit(1);
    }

    let fifo = pdir.stdin_pipe_path();
    let fifo_cstr = match std::ffi::CString::new(fifo.to_str().unwrap_or("")) {
        Ok(c) => c,
        Err(_) => std::process::exit(1),
    };

    // Blocking O_WRONLY — provides backpressure when command's buffer is full
    let fd = unsafe { libc::open(fifo_cstr.as_ptr(), libc::O_WRONLY) };
    if fd < 0 {
        std::process::exit(1);
    }

    // Ignore SIGPIPE — we handle EPIPE from write() return value
    unsafe { libc::signal(libc::SIGPIPE, libc::SIG_IGN) };

    let stdin = std::io::stdin();
    let mut stdin = stdin.lock();
    let mut buf = [0u8; 8192];

    loop {
        let n = match stdin.read(&mut buf) {
            Ok(0) => break,
            Ok(n) => n,
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
            Err(_) => break,
        };

        let mut offset = 0;
        while offset < n {
            let written = unsafe {
                libc::write(
                    fd,
                    buf[offset..n].as_ptr() as *const libc::c_void,
                    n - offset,
                )
            };
            if written <= 0 {
                unsafe { libc::close(fd) };
                if !no_close {
                    kill_stdin_holder(&pdir);
                }
                std::process::exit(1);
            }
            offset += written as usize;
        }
    }

    unsafe { libc::close(fd) };

    if !no_close {
        kill_stdin_holder(&pdir);
    }
}

/// Kill a process with SIGTERM→wait→SIGKILL escalation.
///
/// After setsid(), the runner is the process group leader (PGID == runner_pid).
/// The command and holder inherit that PGID. We kill the entire group.
pub fn kill(id: &str) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);

    if !pdir.dir.exists() {
        return Response::error(format!("process not found: {id}"));
    }

    // Don't overwrite a real exit code with "exited(killed)"
    if let Ok(state) = pdir.read_status()
        && !matches!(state, ProcessState::Running)
    {
        return Response::error(format!("process already exited: {id}"));
    }

    let rpid = pdir.read_runner_pid().ok().flatten();
    let cpid = pdir.read_pid().ok().flatten();

    // Phase 1: SIGTERM to process group
    let sent = if let Some(rpid) = rpid {
        (unsafe { libc::kill(-(rpid as i32), libc::SIGTERM) }) == 0
    } else if let Some(cpid) = cpid {
        (unsafe { libc::kill(cpid as i32, libc::SIGTERM) }) == 0
    } else {
        false
    };

    if !sent {
        // Process already dead — just update state
        let _ = fs::write(pdir.status_path(), "exited(killed)");
        let _ = fs::write(pdir.ended_path(), unix_timestamp().to_string());
        return Response::Killed { id: id.to_string() };
    }

    // Phase 2: wait briefly for graceful exit
    thread::sleep(Duration::from_millis(200));

    // Phase 3: check if still alive, escalate to SIGKILL
    let still_alive = cpid
        .map(|p| unsafe { libc::kill(p as i32, 0) } == 0)
        .unwrap_or(false);

    if still_alive {
        if let Some(rpid) = rpid {
            unsafe { libc::kill(-(rpid as i32), libc::SIGKILL) };
        } else if let Some(cpid) = cpid {
            unsafe { libc::kill(cpid as i32, libc::SIGKILL) };
        }
        thread::sleep(Duration::from_millis(50));
    }

    let _ = fs::write(pdir.status_path(), "exited(killed)");
    let _ = fs::write(pdir.ended_path(), unix_timestamp().to_string());

    Response::Killed { id: id.to_string() }
}

/// List all managed processes.
pub fn list() -> Response {
    let base = remote_base();
    let mut processes = Vec::new();

    if let Ok(entries) = fs::read_dir(&base) {
        for entry in entries.flatten() {
            if !entry.path().is_dir() {
                continue;
            }
            let id = entry.file_name().to_string_lossy().to_string();
            let pdir = ProcessDir::new(&base, &id);
            if !pdir.status_path().exists() {
                continue;
            }
            let state = pdir
                .read_status()
                .map(|s| s.to_string())
                .unwrap_or_else(|_| "unknown".to_string());
            let cmd = pdir.read_cmd().unwrap_or_default();
            processes.push(ProcessSummary { id, state, cmd });
        }
    }

    Response::List { processes }
}

/// Clean up exited processes.
pub fn clean() -> Response {
    let base = remote_base();
    let mut removed = Vec::new();

    if let Ok(entries) = fs::read_dir(&base) {
        for entry in entries.flatten() {
            if !entry.path().is_dir() {
                continue;
            }
            let id = entry.file_name().to_string_lossy().to_string();
            let pdir = ProcessDir::new(&base, &id);
            if !pdir.status_path().exists() {
                continue;
            }

            if let Ok(state) = pdir.read_status()
                && !matches!(state, ProcessState::Running)
            {
                if let Ok(Some(rpid)) = pdir.read_runner_pid() {
                    unsafe { libc::kill(rpid as i32, libc::SIGTERM) };
                }
                if let Ok(Some(hpid)) = pdir.read_stdin_holder_pid() {
                    unsafe { libc::kill(hpid as i32, libc::SIGTERM) };
                }
                let _ = fs::remove_dir_all(&pdir.dir);
                removed.push(id);
            }
        }
    }

    Response::Cleaned { removed }
}

/// Follow (tail) a stream file, writing raw bytes to stdout.
/// Streams until the process exits and all output is flushed.
/// If `offset` is provided, seeks to that position first (for resume).
pub fn follow(id: &str, stream: &str, offset: Option<u64>) -> Response {
    if !is_valid_process_id(id) {
        return invalid_process_id(id);
    }

    let base = remote_base();
    let pdir = ProcessDir::new(&base, id);
    let path = match stream {
        "stdout" => pdir.stdout_path(),
        "stderr" => pdir.stderr_path(),
        _ => return Response::error(format!("invalid stream: {stream}")),
    };

    let mut file = match fs::File::open(&path) {
        Ok(f) => f,
        Err(_) => return Response::error(format!("process not found: {id}")),
    };

    if let Some(off) = offset
        && off > 0
    {
        let _ = file.seek(SeekFrom::Start(off));
    }

    let stdout = std::io::stdout();
    let mut stdout = stdout.lock();
    let mut buf = [0u8; 8192];

    loop {
        match file.read(&mut buf) {
            Ok(0) => {
                if let Ok(state) = pdir.read_status()
                    && !matches!(state, ProcessState::Running)
                {
                    // Final drain
                    while let Ok(n) = file.read(&mut buf) {
                        if n == 0 {
                            break;
                        }
                        let _ = stdout.write_all(&buf[..n]);
                    }
                    let _ = stdout.flush();
                    break;
                }
                thread::sleep(Duration::from_millis(100));
            }
            Ok(n) => {
                let _ = stdout.write_all(&buf[..n]);
                let _ = stdout.flush();
            }
            Err(_) => break,
        }
    }

    Response::error("follow completed")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn write_all_nonblocking_writes_payload_larger_than_pipe_capacity() {
        let mut fds = [0; 2];
        assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
        let read_fd = fds[0];
        let write_fd = fds[1];

        let flags = unsafe { libc::fcntl(write_fd, libc::F_GETFL) };
        assert!(flags >= 0);
        assert_eq!(
            unsafe { libc::fcntl(write_fd, libc::F_SETFL, flags | libc::O_NONBLOCK) },
            0
        );

        let data = vec![b'x'; 128 * 1024];
        let expected_len = data.len();
        let reader = thread::spawn(move || {
            let mut out = Vec::with_capacity(expected_len);
            let mut buf = [0u8; 4096];

            while out.len() < expected_len {
                let n = unsafe {
                    libc::read(read_fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len())
                };
                if n > 0 {
                    out.extend_from_slice(&buf[..n as usize]);
                } else if n == 0 {
                    break;
                } else {
                    let err = std::io::Error::last_os_error();
                    if err.kind() != std::io::ErrorKind::Interrupted {
                        panic!("pipe read failed: {err}");
                    }
                }
            }

            unsafe { libc::close(read_fd) };
            out
        });

        let written = write_all_nonblocking(write_fd, &data).unwrap();
        unsafe { libc::close(write_fd) };

        let out = reader.join().unwrap();
        assert_eq!(written, data.len());
        assert_eq!(out, data);
    }
}
