use std::collections::BTreeMap;
use std::path::Path;
use std::process::Stdio;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result, anyhow};
use tokio::fs::OpenOptions;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStdin, Command};
use tokio::sync::Mutex;

use crate::protocol::{EventStream, HarnessConfig, HarnessEvent, HarnessKind};

#[derive(Debug)]
pub struct RunningHarness {
    name: String,
    child: Child,
    stdin: Option<ChildStdin>,
    log_path: String,
}

impl RunningHarness {
    pub fn pid(&self) -> Option<u32> {
        self.child.id()
    }

    pub async fn is_alive(&mut self) -> bool {
        match self.child.try_wait() {
            Ok(None) => true,
            Ok(Some(_)) | Err(_) => false,
        }
    }

    pub async fn write(&mut self, line: &str) -> Result<()> {
        let stdin = self
            .stdin
            .as_mut()
            .ok_or_else(|| anyhow!("harness stdin is closed"))?;
        stdin.write_all(line.as_bytes()).await?;
        stdin.write_all(b"\n").await?;
        stdin.flush().await?;
        append_event(
            Path::new(&self.log_path),
            &HarnessEvent {
                harness: self.name.clone(),
                stream: EventStream::Stdin,
                line: line.to_string(),
                at_unix_ms: now_ms(),
            },
        )
        .await?;
        Ok(())
    }

    pub async fn stop(&mut self) -> Result<()> {
        let _ = self.child.kill().await;
        append_event(
            Path::new(&self.log_path),
            &HarnessEvent {
                harness: self.name.clone(),
                stream: EventStream::Status,
                line: "stopped".to_string(),
                at_unix_ms: now_ms(),
            },
        )
        .await?;
        Ok(())
    }

    pub fn log_path(&self) -> &str {
        &self.log_path
    }
}

pub type ProcessTable = Arc<Mutex<BTreeMap<String, RunningHarness>>>;

pub fn new_process_table() -> ProcessTable {
    Arc::new(Mutex::new(BTreeMap::new()))
}

pub fn command_for(config: &HarnessConfig) -> (String, Vec<String>) {
    let mut args = Vec::new();
    match config.kind {
        HarnessKind::Claude => {
            args.extend([
                "--continue".to_string(),
                "--output-format".to_string(),
                "stream-json".to_string(),
                "--input-format".to_string(),
                "stream-json".to_string(),
                "--print".to_string(),
                "--verbose".to_string(),
                "--dangerously-skip-permissions".to_string(),
            ]);
            if !config.personality.trim().is_empty() {
                args.push("--system-prompt".to_string());
                args.push(config.personality.clone());
            }
            args.extend(config.args.clone());
        }
        HarnessKind::Codex => {
            args.push("exec".to_string());
            args.push(config.personality.clone());
            args.push("--full-auto".to_string());
            args.extend(config.args.clone());
        }
        HarnessKind::OpenClaw | HarnessKind::Generic => {
            args.extend(config.args.clone());
        }
    }
    (config.command.clone(), args)
}

pub async fn spawn_harness(config: &HarnessConfig, log_dir: &Path) -> Result<RunningHarness> {
    let (bin, args) = command_for(config);
    tokio::fs::create_dir_all(log_dir).await?;
    let log_path = log_dir.join(format!("{}.jsonl", config.name));

    let mut cmd = Command::new(bin);
    cmd.args(args)
        .current_dir(&config.cwd)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    for (k, v) in &config.env {
        cmd.env(k, v);
    }

    let mut child = cmd.spawn().context("spawn harness process")?;
    let stdin = child.stdin.take();
    if let Some(stdout) = child.stdout.take() {
        emit_output_task(
            config.name.clone(),
            log_path.clone(),
            stdout,
            EventStream::Stdout,
        );
    }
    if let Some(stderr) = child.stderr.take() {
        emit_output_task(
            config.name.clone(),
            log_path.clone(),
            stderr,
            EventStream::Stderr,
        );
    }
    append_event(
        &log_path,
        &HarnessEvent {
            harness: config.name.clone(),
            stream: EventStream::Status,
            line: "started".to_string(),
            at_unix_ms: now_ms(),
        },
    )
    .await?;
    Ok(RunningHarness {
        name: config.name.clone(),
        child,
        stdin,
        log_path: log_path.display().to_string(),
    })
}

fn emit_output_task<R>(
    harness: String,
    log_path: std::path::PathBuf,
    reader: R,
    stream: EventStream,
) where
    R: tokio::io::AsyncRead + Send + Unpin + 'static,
{
    tokio::spawn(async move {
        let mut lines = BufReader::new(reader).lines();
        loop {
            match lines.next_line().await {
                Ok(Some(line)) => {
                    let event = HarnessEvent {
                        harness: harness.clone(),
                        stream,
                        line,
                        at_unix_ms: now_ms(),
                    };
                    if let Err(err) = append_event(&log_path, &event).await {
                        tracing::warn!(harness = %harness, error = %err, "failed to append harness event");
                    }
                }
                Ok(None) => break,
                Err(err) => {
                    let event = HarnessEvent {
                        harness: harness.clone(),
                        stream: EventStream::Status,
                        line: format!("read_error:{err}"),
                        at_unix_ms: now_ms(),
                    };
                    let _ = append_event(&log_path, &event).await;
                    break;
                }
            }
        }
    });
}

pub async fn append_event(path: &Path, event: &HarnessEvent) -> Result<()> {
    if let Some(parent) = path.parent() {
        tokio::fs::create_dir_all(parent).await?;
    }
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .await?;
    let mut line = serde_json::to_vec(event)?;
    line.push(b'\n');
    file.write_all(&line).await?;
    file.flush().await?;
    Ok(())
}

pub async fn tail_events(path: &Path, max_lines: usize) -> Result<Vec<HarnessEvent>> {
    let file = match tokio::fs::File::open(path).await {
        Ok(file) => file,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(err) => return Err(err.into()),
    };
    let mut lines = Vec::new();
    let mut reader = BufReader::new(file).lines();
    while let Some(line) = reader.next_line().await? {
        if line.trim().is_empty() {
            continue;
        }
        lines.push(serde_json::from_str(&line)?);
    }
    if lines.len() > max_lines {
        Ok(lines.split_off(lines.len() - max_lines))
    } else {
        Ok(lines)
    }
}

pub fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .try_into()
        .unwrap_or(u64::MAX)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::os::unix::fs::PermissionsExt;

    fn cfg(kind: HarnessKind) -> HarnessConfig {
        HarnessConfig {
            name: "brain".into(),
            kind,
            command: "bin".into(),
            args: vec!["--x".into()],
            cwd: "/tmp".into(),
            personality: "reviewer".into(),
            env: vec![],
        }
    }

    #[test]
    fn claude_command_matches_stream_json_reference_pattern() {
        let (_bin, args) = command_for(&cfg(HarnessKind::Claude));
        assert!(args.contains(&"--input-format".to_string()));
        assert!(args.contains(&"stream-json".to_string()));
        assert!(args.contains(&"--system-prompt".to_string()));
        assert!(args.contains(&"reviewer".to_string()));
    }

    #[test]
    fn codex_command_uses_exec_full_auto() {
        let (_bin, args) = command_for(&cfg(HarnessKind::Codex));
        assert_eq!(args[0], "exec");
        assert!(args.contains(&"--full-auto".to_string()));
    }

    #[tokio::test]
    async fn event_log_round_trips_jsonl() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("brain.jsonl");
        append_event(
            &path,
            &HarnessEvent {
                harness: "brain".into(),
                stream: EventStream::Status,
                line: "started".into(),
                at_unix_ms: 1,
            },
        )
        .await
        .unwrap();
        let events = tail_events(&path, 10).await.unwrap();
        assert_eq!(events[0].line, "started");
    }

    #[tokio::test]
    async fn spawn_harness_records_status_stdout_stdin_and_stop() {
        let dir = tempfile::tempdir().unwrap();
        let bin = dir.path().join("mock-agent");
        std::fs::write(
            &bin,
            r#"#!/usr/bin/env bash
echo ready
while IFS= read -r line; do
  if [ -n "$MOCK_STDERR" ]; then echo "err:$line" >&2; fi
  echo "out:$line"
done
"#,
        )
        .unwrap();
        let mut perms = std::fs::metadata(&bin).unwrap().permissions();
        perms.set_mode(0o755);
        std::fs::set_permissions(&bin, perms).unwrap();
        let config = HarnessConfig {
            name: "mock".into(),
            kind: HarnessKind::Generic,
            command: bin.display().to_string(),
            args: vec![],
            cwd: dir.path().display().to_string(),
            personality: "p".into(),
            env: vec![("MOCK_STDERR".into(), "1".into())],
        };

        let mut proc = spawn_harness(&config, &dir.path().join("logs"))
            .await
            .unwrap();
        assert!(proc.pid().is_some());
        assert!(proc.is_alive().await);
        proc.write("hello").await.unwrap();
        let events = wait_for_events(Path::new(proc.log_path()), |events| {
            events.iter().any(|e| e.stream == EventStream::Stdout)
                && events.iter().any(|e| e.stream == EventStream::Stderr)
        })
        .await;
        proc.stop().await.unwrap();
        assert!(events.iter().any(|e| e.stream == EventStream::Status));
        assert!(events.iter().any(|e| e.stream == EventStream::Stdout));
        assert!(events.iter().any(|e| e.stream == EventStream::Stderr));
        assert!(
            events
                .iter()
                .any(|e| e.stream == EventStream::Stdin && e.line == "hello")
        );
    }

    #[tokio::test]
    async fn tail_events_missing_file_is_empty_and_limits_results() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("events.jsonl");
        assert!(tail_events(&path, 2).await.unwrap().is_empty());
        for i in 0..3 {
            append_event(
                &path,
                &HarnessEvent {
                    harness: "h".into(),
                    stream: EventStream::Status,
                    line: format!("{i}"),
                    at_unix_ms: i,
                },
            )
            .await
            .unwrap();
        }
        let events = tail_events(&path, 2).await.unwrap();
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].line, "1");
    }

    async fn wait_for_events<F>(path: &Path, predicate: F) -> Vec<HarnessEvent>
    where
        F: Fn(&[HarnessEvent]) -> bool,
    {
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                let events = tail_events(path, 50).await.unwrap();
                if predicate(&events) {
                    return events;
                }
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap()
    }
}
