use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;

use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine;
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
use tokio::process::{Child, ChildStdin, Command};
use tokio::sync::{broadcast, Mutex, RwLock};

use sandbox_agent_error::SandboxError;

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum ProcessStatus {
    Running,
    Exited,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum ProcessStream {
    Stdout,
    Stderr,
    Pty,
}

#[derive(Debug, Clone)]
pub struct ProcessStartSpec {
    pub command: String,
    pub args: Vec<String>,
    pub cwd: Option<String>,
    pub env: HashMap<String, String>,
    pub tty: bool,
    pub interactive: bool,
}

#[derive(Debug, Clone)]
pub struct RunSpec {
    pub command: String,
    pub args: Vec<String>,
    pub cwd: Option<String>,
    pub env: HashMap<String, String>,
    pub timeout_ms: Option<u64>,
    pub max_output_bytes: Option<usize>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RunOutput {
    pub exit_code: Option<i32>,
    pub timed_out: bool,
    pub stdout: String,
    pub stderr: String,
    pub stdout_truncated: bool,
    pub stderr_truncated: bool,
    pub duration_ms: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProcessLogLine {
    pub sequence: u64,
    pub stream: ProcessStream,
    pub timestamp_ms: i64,
    pub data: String,
    pub encoding: &'static str,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProcessSnapshot {
    pub id: String,
    pub command: String,
    pub args: Vec<String>,
    pub cwd: Option<String>,
    pub tty: bool,
    pub interactive: bool,
    pub status: ProcessStatus,
    pub pid: Option<u32>,
    pub exit_code: Option<i32>,
    pub created_at_ms: i64,
    pub exited_at_ms: Option<i64>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProcessRuntimeConfig {
    pub max_concurrent_processes: usize,
    pub default_run_timeout_ms: u64,
    pub max_run_timeout_ms: u64,
    pub max_output_bytes: usize,
    pub max_log_bytes_per_process: usize,
    pub max_input_bytes_per_request: usize,
}

impl Default for ProcessRuntimeConfig {
    fn default() -> Self {
        Self {
            max_concurrent_processes: 64,
            default_run_timeout_ms: 30_000,
            max_run_timeout_ms: 300_000,
            max_output_bytes: 1_048_576,
            max_log_bytes_per_process: 10_485_760,
            max_input_bytes_per_request: 65_536,
        }
    }
}

#[derive(Debug, Clone)]
pub struct ProcessRuntime {
    config: Arc<RwLock<ProcessRuntimeConfig>>,
    inner: Arc<ProcessRuntimeInner>,
}

#[derive(Debug)]
struct ProcessRuntimeInner {
    next_id: AtomicU64,
    processes: RwLock<HashMap<String, Arc<ManagedProcess>>>,
}

#[derive(Debug)]
struct ManagedProcess {
    id: String,
    command: String,
    args: Vec<String>,
    cwd: Option<String>,
    tty: bool,
    interactive: bool,
    created_at_ms: i64,
    pid: Option<u32>,
    max_log_bytes: usize,
    stdin: Mutex<Option<ProcessStdin>>,
    #[cfg(unix)]
    pty_resize_fd: Mutex<Option<std::fs::File>>,
    status: RwLock<ManagedStatus>,
    sequence: AtomicU64,
    logs: Mutex<VecDeque<StoredLog>>,
    total_log_bytes: Mutex<usize>,
    log_tx: broadcast::Sender<ProcessLogLine>,
}

#[derive(Debug)]
enum ProcessStdin {
    Pipe(ChildStdin),
    Pty(tokio::fs::File),
}

#[derive(Debug, Clone)]
struct StoredLog {
    line: ProcessLogLine,
    byte_len: usize,
}

#[derive(Debug, Clone)]
struct ManagedStatus {
    status: ProcessStatus,
    exit_code: Option<i32>,
    exited_at_ms: Option<i64>,
}

struct SpawnedPipeProcess {
    process: Arc<ManagedProcess>,
    child: Child,
    stdout: tokio::process::ChildStdout,
    stderr: tokio::process::ChildStderr,
}

#[cfg(unix)]
struct SpawnedTtyProcess {
    process: Arc<ManagedProcess>,
    child: Child,
    reader: tokio::fs::File,
}

impl ProcessRuntime {
    pub fn new() -> Self {
        Self {
            config: Arc::new(RwLock::new(ProcessRuntimeConfig::default())),
            inner: Arc::new(ProcessRuntimeInner {
                next_id: AtomicU64::new(1),
                processes: RwLock::new(HashMap::new()),
            }),
        }
    }

    pub async fn get_config(&self) -> ProcessRuntimeConfig {
        self.config.read().await.clone()
    }

    pub async fn set_config(
        &self,
        mut value: ProcessRuntimeConfig,
    ) -> Result<ProcessRuntimeConfig, SandboxError> {
        if value.max_concurrent_processes == 0 {
            return Err(SandboxError::InvalidRequest {
                message: "maxConcurrentProcesses must be greater than 0".to_string(),
            });
        }
        if value.default_run_timeout_ms == 0 || value.max_run_timeout_ms == 0 {
            return Err(SandboxError::InvalidRequest {
                message: "timeouts must be greater than 0".to_string(),
            });
        }
        if value.default_run_timeout_ms > value.max_run_timeout_ms {
            value.default_run_timeout_ms = value.max_run_timeout_ms;
        }
        if value.max_output_bytes == 0
            || value.max_log_bytes_per_process == 0
            || value.max_input_bytes_per_request == 0
        {
            return Err(SandboxError::InvalidRequest {
                message: "byte limits must be greater than 0".to_string(),
            });
        }

        *self.config.write().await = value.clone();
        Ok(value)
    }

    pub async fn start_process(
        &self,
        spec: ProcessStartSpec,
    ) -> Result<ProcessSnapshot, SandboxError> {
        let config = self.get_config().await;

        let process_refs = {
            let processes = self.inner.processes.read().await;
            processes.values().cloned().collect::<Vec<_>>()
        };

        let mut running_count = 0usize;
        for process in process_refs {
            if process.status.read().await.status == ProcessStatus::Running {
                running_count += 1;
            }
        }

        if running_count >= config.max_concurrent_processes {
            return Err(SandboxError::Conflict {
                message: format!(
                    "max concurrent process limit reached ({})",
                    config.max_concurrent_processes
                ),
            });
        }

        if spec.command.trim().is_empty() {
            return Err(SandboxError::InvalidRequest {
                message: "command must not be empty".to_string(),
            });
        }

        let id_num = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
        let id = format!("proc_{id_num}");

        if spec.tty {
            #[cfg(unix)]
            {
                let spawned = self
                    .spawn_tty_process(id.clone(), spec, config.max_log_bytes_per_process)
                    .await?;
                let process = spawned.process.clone();
                self.inner
                    .processes
                    .write()
                    .await
                    .insert(id, process.clone());

                let p = process.clone();
                tokio::spawn(async move {
                    pump_output(p, spawned.reader, ProcessStream::Pty).await;
                });

                let p = process.clone();
                tokio::spawn(async move {
                    watch_exit(p, spawned.child).await;
                });

                return Ok(process.snapshot().await);
            }
            #[cfg(not(unix))]
            {
                return Err(SandboxError::StreamError {
                    message: "tty process mode is not supported on this platform".to_string(),
                });
            }
        }

        let spawned = self
            .spawn_pipe_process(id.clone(), spec, config.max_log_bytes_per_process)
            .await?;
        let process = spawned.process.clone();
        self.inner
            .processes
            .write()
            .await
            .insert(id, process.clone());

        let p = process.clone();
        tokio::spawn(async move {
            pump_output(p, spawned.stdout, ProcessStream::Stdout).await;
        });

        let p = process.clone();
        tokio::spawn(async move {
            pump_output(p, spawned.stderr, ProcessStream::Stderr).await;
        });

        let p = process.clone();
        tokio::spawn(async move {
            watch_exit(p, spawned.child).await;
        });

        Ok(process.snapshot().await)
    }

    pub async fn run_once(&self, spec: RunSpec) -> Result<RunOutput, SandboxError> {
        if spec.command.trim().is_empty() {
            return Err(SandboxError::InvalidRequest {
                message: "command must not be empty".to_string(),
            });
        }

        let config = self.get_config().await;
        let mut timeout_ms = spec.timeout_ms.unwrap_or(config.default_run_timeout_ms);
        if timeout_ms == 0 {
            timeout_ms = config.default_run_timeout_ms;
        }
        timeout_ms = timeout_ms.min(config.max_run_timeout_ms);

        let max_output_bytes = spec.max_output_bytes.unwrap_or(config.max_output_bytes);

        let mut cmd = Command::new(&spec.command);
        cmd.args(&spec.args)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        if let Some(cwd) = &spec.cwd {
            cmd.current_dir(cwd);
        }

        for (key, value) in &spec.env {
            cmd.env(key, value);
        }

        let mut child = cmd.spawn().map_err(|err| SandboxError::StreamError {
            message: format!("failed to spawn process: {err}"),
        })?;

        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| SandboxError::StreamError {
                message: "failed to capture stdout".to_string(),
            })?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| SandboxError::StreamError {
                message: "failed to capture stderr".to_string(),
            })?;

        let started = Instant::now();
        let stdout_task = tokio::spawn(capture_output(stdout, max_output_bytes));
        let stderr_task = tokio::spawn(capture_output(stderr, max_output_bytes));

        let wait_result =
            tokio::time::timeout(std::time::Duration::from_millis(timeout_ms), child.wait()).await;

        let (exit_code, timed_out) = match wait_result {
            Ok(Ok(status)) => (status.code(), false),
            Ok(Err(err)) => {
                let _ = child.kill().await;
                return Err(SandboxError::StreamError {
                    message: format!("failed to wait on process: {err}"),
                });
            }
            Err(_) => {
                let _ = child.kill().await;
                let _ = child.wait().await;
                (None, true)
            }
        };

        let (stdout, stdout_truncated) = match stdout_task.await {
            Ok(Ok(captured)) => captured,
            _ => (Vec::new(), false),
        };
        let (stderr, stderr_truncated) = match stderr_task.await {
            Ok(Ok(captured)) => captured,
            _ => (Vec::new(), false),
        };

        Ok(RunOutput {
            exit_code,
            timed_out,
            stdout: String::from_utf8_lossy(&stdout).to_string(),
            stderr: String::from_utf8_lossy(&stderr).to_string(),
            stdout_truncated,
            stderr_truncated,
            duration_ms: started.elapsed().as_millis() as u64,
        })
    }

    pub async fn list_processes(&self) -> Vec<ProcessSnapshot> {
        let processes = self.inner.processes.read().await;
        let mut items = Vec::with_capacity(processes.len());
        for process in processes.values() {
            items.push(process.snapshot().await);
        }
        items.sort_by(|a, b| a.id.cmp(&b.id));
        items
    }

    pub async fn snapshot(&self, id: &str) -> Result<ProcessSnapshot, SandboxError> {
        Ok(self.lookup_process(id).await?.snapshot().await)
    }

    pub async fn is_tty(&self, id: &str) -> Result<bool, SandboxError> {
        Ok(self.lookup_process(id).await?.tty)
    }

    pub async fn max_input_bytes(&self) -> usize {
        self.get_config().await.max_input_bytes_per_request
    }

    pub async fn delete_process(&self, id: &str) -> Result<(), SandboxError> {
        let process = self.lookup_process(id).await?;
        let status = process.status.read().await.clone();
        if status.status == ProcessStatus::Running {
            return Err(SandboxError::Conflict {
                message: "process is still running; stop or kill it before delete".to_string(),
            });
        }

        self.inner.processes.write().await.remove(id);
        Ok(())
    }

    pub async fn stop_process(
        &self,
        id: &str,
        wait_ms: Option<u64>,
    ) -> Result<ProcessSnapshot, SandboxError> {
        let process = self.lookup_process(id).await?;
        process.send_signal(SIGTERM).await?;
        maybe_wait_for_exit(process.clone(), wait_ms.unwrap_or(2_000)).await;
        Ok(process.snapshot().await)
    }

    pub async fn kill_process(
        &self,
        id: &str,
        wait_ms: Option<u64>,
    ) -> Result<ProcessSnapshot, SandboxError> {
        let process = self.lookup_process(id).await?;
        process.send_signal(SIGKILL).await?;
        maybe_wait_for_exit(process.clone(), wait_ms.unwrap_or(1_000)).await;
        Ok(process.snapshot().await)
    }

    pub async fn write_input(&self, id: &str, data: &[u8]) -> Result<usize, SandboxError> {
        self.lookup_process(id).await?.write_input(data).await
    }

    pub async fn resize_terminal(
        &self,
        id: &str,
        cols: u16,
        rows: u16,
    ) -> Result<(), SandboxError> {
        let process = self.lookup_process(id).await?;
        if !process.tty {
            return Err(SandboxError::Conflict {
                message: "process is not running in tty mode".to_string(),
            });
        }

        process.resize_pty(cols, rows).await?;
        process.send_signal(SIGWINCH).await
    }

    pub async fn logs(
        &self,
        id: &str,
        filter: ProcessLogFilter,
    ) -> Result<Vec<ProcessLogLine>, SandboxError> {
        self.lookup_process(id).await?.read_logs(filter).await
    }

    pub async fn subscribe_logs(
        &self,
        id: &str,
    ) -> Result<broadcast::Receiver<ProcessLogLine>, SandboxError> {
        let process = self.lookup_process(id).await?;
        Ok(process.log_tx.subscribe())
    }

    async fn lookup_process(&self, id: &str) -> Result<Arc<ManagedProcess>, SandboxError> {
        let process = self.inner.processes.read().await.get(id).cloned();
        process.ok_or_else(|| SandboxError::NotFound {
            resource: "process".to_string(),
            id: id.to_string(),
        })
    }

    async fn spawn_pipe_process(
        &self,
        id: String,
        spec: ProcessStartSpec,
        max_log_bytes: usize,
    ) -> Result<SpawnedPipeProcess, SandboxError> {
        let mut cmd = Command::new(&spec.command);
        cmd.args(&spec.args)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        if let Some(cwd) = &spec.cwd {
            cmd.current_dir(cwd);
        }

        for (key, value) in &spec.env {
            cmd.env(key, value);
        }

        let mut child = cmd.spawn().map_err(|err| SandboxError::StreamError {
            message: format!("failed to spawn process: {err}"),
        })?;

        let stdin = child.stdin.take();
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| SandboxError::StreamError {
                message: "failed to capture stdout".to_string(),
            })?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| SandboxError::StreamError {
                message: "failed to capture stderr".to_string(),
            })?;
        let pid = child.id();

        let (tx, _rx) = broadcast::channel(512);
        let process = Arc::new(ManagedProcess {
            id,
            command: spec.command,
            args: spec.args,
            cwd: spec.cwd,
            tty: false,
            interactive: spec.interactive,
            created_at_ms: now_ms(),
            pid,
            max_log_bytes,
            stdin: Mutex::new(stdin.map(ProcessStdin::Pipe)),
            #[cfg(unix)]
            pty_resize_fd: Mutex::new(None),
            status: RwLock::new(ManagedStatus {
                status: ProcessStatus::Running,
                exit_code: None,
                exited_at_ms: None,
            }),
            sequence: AtomicU64::new(1),
            logs: Mutex::new(VecDeque::new()),
            total_log_bytes: Mutex::new(0),
            log_tx: tx,
        });

        Ok(SpawnedPipeProcess {
            process,
            child,
            stdout,
            stderr,
        })
    }

    #[cfg(unix)]
    async fn spawn_tty_process(
        &self,
        id: String,
        spec: ProcessStartSpec,
        max_log_bytes: usize,
    ) -> Result<SpawnedTtyProcess, SandboxError> {
        use std::os::fd::AsRawFd;
        use std::process::Stdio;

        let (master_fd, slave_fd) = open_pty(80, 24)?;
        let slave_raw = slave_fd.as_raw_fd();

        let stdin_fd = dup_fd(slave_raw)?;
        let stdout_fd = dup_fd(slave_raw)?;
        let stderr_fd = dup_fd(slave_raw)?;

        let mut cmd = Command::new(&spec.command);
        cmd.args(&spec.args)
            .stdin(Stdio::from(std::fs::File::from(stdin_fd)))
            .stdout(Stdio::from(std::fs::File::from(stdout_fd)))
            .stderr(Stdio::from(std::fs::File::from(stderr_fd)));

        if let Some(cwd) = &spec.cwd {
            cmd.current_dir(cwd);
        }

        for (key, value) in &spec.env {
            cmd.env(key, value);
        }

        unsafe {
            cmd.pre_exec(move || {
                if libc::setsid() == -1 {
                    return Err(std::io::Error::last_os_error());
                }
                if libc::ioctl(slave_raw, libc::TIOCSCTTY as _, 0) == -1 {
                    return Err(std::io::Error::last_os_error());
                }
                Ok(())
            });
        }

        let child = cmd.spawn().map_err(|err| SandboxError::StreamError {
            message: format!("failed to spawn tty process: {err}"),
        })?;

        let pid = child.id();
        drop(slave_fd);

        let master_raw = master_fd.as_raw_fd();
        let writer_fd = dup_fd(master_raw)?;
        let resize_fd = dup_fd(master_raw)?;

        let reader_file = tokio::fs::File::from_std(std::fs::File::from(master_fd));
        let writer_file = tokio::fs::File::from_std(std::fs::File::from(writer_fd));
        let resize_file = std::fs::File::from(resize_fd);

        let (tx, _rx) = broadcast::channel(512);
        let process = Arc::new(ManagedProcess {
            id,
            command: spec.command,
            args: spec.args,
            cwd: spec.cwd,
            tty: true,
            interactive: spec.interactive,
            created_at_ms: now_ms(),
            pid,
            max_log_bytes,
            stdin: Mutex::new(Some(ProcessStdin::Pty(writer_file))),
            pty_resize_fd: Mutex::new(Some(resize_file)),
            status: RwLock::new(ManagedStatus {
                status: ProcessStatus::Running,
                exit_code: None,
                exited_at_ms: None,
            }),
            sequence: AtomicU64::new(1),
            logs: Mutex::new(VecDeque::new()),
            total_log_bytes: Mutex::new(0),
            log_tx: tx,
        });

        Ok(SpawnedTtyProcess {
            process,
            child,
            reader: reader_file,
        })
    }
}

#[derive(Debug, Clone, Copy)]
pub enum ProcessLogFilterStream {
    Stdout,
    Stderr,
    Combined,
    Pty,
}

#[derive(Debug, Clone, Copy)]
pub struct ProcessLogFilter {
    pub stream: ProcessLogFilterStream,
    pub tail: Option<usize>,
    pub since: Option<u64>,
}

impl ManagedProcess {
    async fn snapshot(&self) -> ProcessSnapshot {
        let status = self.status.read().await.clone();
        ProcessSnapshot {
            id: self.id.clone(),
            command: self.command.clone(),
            args: self.args.clone(),
            cwd: self.cwd.clone(),
            tty: self.tty,
            interactive: self.interactive,
            status: status.status,
            pid: self.pid,
            exit_code: status.exit_code,
            created_at_ms: self.created_at_ms,
            exited_at_ms: status.exited_at_ms,
        }
    }

    async fn append_log(&self, stream: ProcessStream, data: &[u8]) {
        if data.is_empty() {
            return;
        }

        let stream = if self.tty { ProcessStream::Pty } else { stream };
        let line = ProcessLogLine {
            sequence: self.sequence.fetch_add(1, Ordering::Relaxed),
            stream,
            timestamp_ms: now_ms(),
            data: BASE64.encode(data),
            encoding: "base64",
        };
        let stored = StoredLog {
            line: line.clone(),
            byte_len: data.len(),
        };

        {
            let mut logs = self.logs.lock().await;
            let mut total = self.total_log_bytes.lock().await;
            logs.push_back(stored);
            *total += data.len();

            while *total > self.max_log_bytes {
                if let Some(front) = logs.pop_front() {
                    *total = total.saturating_sub(front.byte_len);
                } else {
                    break;
                }
            }
        }

        let _ = self.log_tx.send(line);
    }

    async fn write_input(&self, data: &[u8]) -> Result<usize, SandboxError> {
        if self.status.read().await.status != ProcessStatus::Running {
            return Err(SandboxError::Conflict {
                message: "process is not running".to_string(),
            });
        }

        let mut guard = self.stdin.lock().await;
        let stdin = guard.as_mut().ok_or_else(|| SandboxError::Conflict {
            message: "process does not accept stdin".to_string(),
        })?;

        match stdin {
            ProcessStdin::Pipe(pipe) => {
                pipe.write_all(data)
                    .await
                    .map_err(|err| SandboxError::StreamError {
                        message: format!("failed to write stdin: {err}"),
                    })?;
                pipe.flush()
                    .await
                    .map_err(|err| SandboxError::StreamError {
                        message: format!("failed to flush stdin: {err}"),
                    })?;
            }
            ProcessStdin::Pty(pty_writer) => {
                pty_writer
                    .write_all(data)
                    .await
                    .map_err(|err| SandboxError::StreamError {
                        message: format!("failed to write PTY input: {err}"),
                    })?;
                pty_writer
                    .flush()
                    .await
                    .map_err(|err| SandboxError::StreamError {
                        message: format!("failed to flush PTY input: {err}"),
                    })?;
            }
        }

        Ok(data.len())
    }

    async fn read_logs(
        &self,
        filter: ProcessLogFilter,
    ) -> Result<Vec<ProcessLogLine>, SandboxError> {
        let logs = self.logs.lock().await;

        let mut entries: Vec<ProcessLogLine> = logs
            .iter()
            .filter_map(|entry| {
                if let Some(since) = filter.since {
                    if entry.line.sequence <= since {
                        return None;
                    }
                }
                if stream_matches(entry.line.stream, filter.stream) {
                    Some(entry.line.clone())
                } else {
                    None
                }
            })
            .collect();

        if let Some(tail) = filter.tail {
            if entries.len() > tail {
                let start = entries.len() - tail;
                entries = entries.split_off(start);
            }
        }

        Ok(entries)
    }

    async fn send_signal(&self, signal: i32) -> Result<(), SandboxError> {
        if self.status.read().await.status != ProcessStatus::Running {
            return Ok(());
        }
        let Some(pid) = self.pid else {
            return Ok(());
        };

        send_signal(pid, signal)
    }

    async fn resize_pty(&self, cols: u16, rows: u16) -> Result<(), SandboxError> {
        if !self.tty {
            return Ok(());
        }

        #[cfg(unix)]
        {
            use std::os::fd::AsRawFd;
            let guard = self.pty_resize_fd.lock().await;
            let Some(fd) = guard.as_ref() else {
                return Err(SandboxError::Conflict {
                    message: "PTY resize handle unavailable".to_string(),
                });
            };
            resize_pty(fd.as_raw_fd(), cols, rows)?;
        }

        #[cfg(not(unix))]
        {
            let _ = cols;
            let _ = rows;
        }

        Ok(())
    }
}

fn stream_matches(stream: ProcessStream, filter: ProcessLogFilterStream) -> bool {
    match filter {
        ProcessLogFilterStream::Stdout => stream == ProcessStream::Stdout,
        ProcessLogFilterStream::Stderr => stream == ProcessStream::Stderr,
        ProcessLogFilterStream::Combined => {
            stream == ProcessStream::Stdout || stream == ProcessStream::Stderr
        }
        ProcessLogFilterStream::Pty => stream == ProcessStream::Pty,
    }
}

async fn maybe_wait_for_exit(process: Arc<ManagedProcess>, wait_ms: u64) {
    let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_millis(wait_ms);
    while tokio::time::Instant::now() < deadline {
        if process.status.read().await.status == ProcessStatus::Exited {
            break;
        }
        tokio::time::sleep(tokio::time::Duration::from_millis(25)).await;
    }
}

async fn pump_output<R>(process: Arc<ManagedProcess>, mut reader: R, stream: ProcessStream)
where
    R: AsyncRead + Unpin,
{
    let mut buffer = [0_u8; 8192];
    loop {
        match reader.read(&mut buffer).await {
            Ok(0) => break,
            Ok(n) => {
                process.append_log(stream, &buffer[..n]).await;
            }
            Err(err) => {
                let msg = format!("\n[process stream error: {err}]\n");
                process
                    .append_log(
                        if process.tty {
                            ProcessStream::Pty
                        } else {
                            ProcessStream::Stderr
                        },
                        msg.as_bytes(),
                    )
                    .await;
                break;
            }
        }
    }
}

async fn watch_exit(process: Arc<ManagedProcess>, mut child: Child) {
    let wait = child.wait().await;
    let (exit_code, exited_at_ms) = match wait {
        Ok(status) => (status.code(), Some(now_ms())),
        Err(_) => (None, Some(now_ms())),
    };

    {
        let mut state = process.status.write().await;
        state.status = ProcessStatus::Exited;
        state.exit_code = exit_code;
        state.exited_at_ms = exited_at_ms;
    }

    let _ = process.stdin.lock().await.take();
}

async fn capture_output<R>(mut reader: R, max_bytes: usize) -> std::io::Result<(Vec<u8>, bool)>
where
    R: AsyncRead + Unpin,
{
    let mut output = Vec::new();
    let mut buffer = [0_u8; 8192];
    let mut truncated = false;

    loop {
        let n = reader.read(&mut buffer).await?;
        if n == 0 {
            break;
        }

        if output.len() < max_bytes {
            let remaining = max_bytes - output.len();
            let to_copy = remaining.min(n);
            output.extend_from_slice(&buffer[..to_copy]);
            if to_copy < n {
                truncated = true;
            }
        } else {
            truncated = true;
        }
    }

    Ok((output, truncated))
}

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| duration.as_millis() as i64)
        .unwrap_or(0)
}

#[cfg(unix)]
const SIGTERM: i32 = libc::SIGTERM;
#[cfg(unix)]
const SIGKILL: i32 = libc::SIGKILL;
#[cfg(unix)]
const SIGWINCH: i32 = libc::SIGWINCH;

#[cfg(unix)]
fn send_signal(pid: u32, signal: i32) -> Result<(), SandboxError> {
    let result = unsafe { libc::kill(pid as libc::pid_t, signal) };
    if result == 0 {
        return Ok(());
    }

    let err = std::io::Error::last_os_error();
    if err.raw_os_error() == Some(libc::ESRCH) {
        return Ok(());
    }

    Err(SandboxError::StreamError {
        message: format!("failed to signal process {pid}: {err}"),
    })
}

#[cfg(not(unix))]
const SIGTERM: i32 = 15;
#[cfg(not(unix))]
const SIGKILL: i32 = 9;
#[cfg(not(unix))]
const SIGWINCH: i32 = 28;

#[cfg(not(unix))]
fn send_signal(_pid: u32, _signal: i32) -> Result<(), SandboxError> {
    Err(SandboxError::StreamError {
        message: "process signaling not supported on this platform".to_string(),
    })
}

#[cfg(unix)]
fn open_pty(
    cols: u16,
    rows: u16,
) -> Result<(std::os::fd::OwnedFd, std::os::fd::OwnedFd), SandboxError> {
    use std::os::fd::FromRawFd;

    let mut master: libc::c_int = -1;
    let mut slave: libc::c_int = -1;
    let mut winsize = libc::winsize {
        ws_row: rows,
        ws_col: cols,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };

    let rc = unsafe {
        libc::openpty(
            &mut master,
            &mut slave,
            std::ptr::null_mut(),
            std::ptr::null_mut(),
            &mut winsize,
        )
    };

    if rc != 0 {
        return Err(SandboxError::StreamError {
            message: format!(
                "failed to allocate PTY: {}",
                std::io::Error::last_os_error()
            ),
        });
    }

    let master_fd = unsafe { std::os::fd::OwnedFd::from_raw_fd(master) };
    let slave_fd = unsafe { std::os::fd::OwnedFd::from_raw_fd(slave) };
    Ok((master_fd, slave_fd))
}

#[cfg(unix)]
fn dup_fd(fd: std::os::fd::RawFd) -> Result<std::os::fd::OwnedFd, SandboxError> {
    use std::os::fd::FromRawFd;

    let duplicated = unsafe { libc::dup(fd) };
    if duplicated == -1 {
        return Err(SandboxError::StreamError {
            message: format!("failed to dup fd: {}", std::io::Error::last_os_error()),
        });
    }

    Ok(unsafe { std::os::fd::OwnedFd::from_raw_fd(duplicated) })
}

#[cfg(unix)]
fn resize_pty(fd: std::os::fd::RawFd, cols: u16, rows: u16) -> Result<(), SandboxError> {
    let winsize = libc::winsize {
        ws_row: rows,
        ws_col: cols,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };

    let rc = unsafe { libc::ioctl(fd, libc::TIOCSWINSZ as _, &winsize) };
    if rc == -1 {
        return Err(SandboxError::StreamError {
            message: format!("failed to resize PTY: {}", std::io::Error::last_os_error()),
        });
    }

    Ok(())
}

pub fn decode_input_bytes(data: &str, encoding: &str) -> Result<Vec<u8>, SandboxError> {
    match encoding {
        "base64" => BASE64
            .decode(data)
            .map_err(|err| SandboxError::InvalidRequest {
                message: format!("invalid base64 input: {err}"),
            }),
        "utf8" | "text" => Ok(data.as_bytes().to_vec()),
        _ => Err(SandboxError::InvalidRequest {
            message: "encoding must be one of: base64, utf8, text".to_string(),
        }),
    }
}
