use std::io;
use std::process::Stdio;

use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::mpsc::UnboundedSender;
use tokio::task::JoinHandle;
use uuid::Uuid;

use crate::session::{MonitorEvent, TranscriptEvent};

pub struct MonitorRunner {
    pub task_id: String,
    pub description: String,
    handle: JoinHandle<()>,
}

impl MonitorRunner {
    pub fn spawn(
        command: impl Into<String>,
        description: impl Into<String>,
        tx: UnboundedSender<TranscriptEvent>,
    ) -> io::Result<Self> {
        let command = command.into();
        let description = description.into();
        let task_id = format!("mon-{}", &Uuid::new_v4().simple().to_string()[..8]);

        let mut child = Command::new("sh")
            .arg("-c")
            .arg(&command)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true)
            .spawn()?;

        let stdout = child.stdout.take().expect("monitor stdout piped");
        let stderr = child.stderr.take().expect("monitor stderr piped");

        tokio::spawn(async move {
            let mut lines = BufReader::new(stderr).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                eprintln!(
                    "{}",
                    serde_json::json!({
                        "level": "info",
                        "event": "monitor.stderr",
                        "line": line,
                    })
                );
            }
        });

        let task_id_for_task = task_id.clone();
        let description_for_task = description.clone();
        let handle = tokio::spawn(async move {
            let mut lines = BufReader::new(stdout).lines();
            loop {
                match lines.next_line().await {
                    Ok(Some(line)) => {
                        let event = MonitorEvent {
                            task_id: task_id_for_task.clone(),
                            description: description_for_task.clone(),
                            event: line,
                            source_uuid: Uuid::new_v4().to_string(),
                            delivery_id: Uuid::new_v4().to_string(),
                            human_turn: false,
                            note: None,
                        };
                        let _ = tx.send(TranscriptEvent::MonitorEvent(event.clone()));
                        let _ = tx.send(TranscriptEvent::MonitorConsumed {
                            event,
                            reason: "absorbed_mid_turn".to_string(),
                        });
                    }
                    Ok(None) => break,
                    Err(error) => {
                        eprintln!(
                            "{}",
                            serde_json::json!({
                                "level": "error",
                                "event": "monitor.read_failed",
                                "error": error.to_string(),
                            })
                        );
                        break;
                    }
                }
            }
            let _ = wait_child(&mut child).await;
        });

        Ok(Self {
            task_id,
            description,
            handle,
        })
    }

    pub fn abort(&self) {
        self.handle.abort();
    }

    pub async fn wait(self) {
        let _ = self.handle.await;
    }

    pub async fn finish(mut self, grace: std::time::Duration) {
        if tokio::time::timeout(grace, &mut self.handle).await.is_err() {
            self.handle.abort();
        }
    }
}

async fn wait_child(child: &mut Child) -> io::Result<std::process::ExitStatus> {
    child.wait().await
}

#[cfg(test)]
mod tests {
    use std::fs;

    use super::*;
    use crate::session::harnesses::claude_code_cli::ClaudeCodeCliHarness;
    use crate::session::SessionLogger;

    #[tokio::test]
    async fn monitor_stdout_becomes_task_notification_records() {
        let temp = tempfile::tempdir().unwrap();
        let cwd = temp.path().join("work");
        fs::create_dir_all(&cwd).unwrap();
        let root = temp.path().join("claude");
        let harness = ClaudeCodeCliHarness::create(Some(&root), &cwd, "test-model").unwrap();
        let session = SessionLogger::start(vec![Box::new(harness)], "alice").unwrap();
        let path = session.session_file().to_path_buf();
        let tx = session.transcript_sender();

        let runner = MonitorRunner::spawn("printf '%s\\n' one two", "monitor & test", tx).unwrap();
        runner.wait().await;
        session.close().await;

        let records: Vec<serde_json::Value> = fs::read_to_string(&path)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        assert_eq!(records.len(), 6);
        for chunk in records.chunks(3) {
            assert_eq!(chunk[0]["type"], "queue-operation");
            assert_eq!(chunk[0]["operation"], "enqueue");
            assert_eq!(chunk[1]["type"], "attachment");
            assert_eq!(chunk[1]["attachment"]["type"], "queued_command");
            assert_eq!(chunk[2]["type"], "queue-operation");
            assert_eq!(chunk[2]["operation"], "remove");
            assert_eq!(
                chunk[2]["commandUuid"],
                chunk[1]["attachment"]["source_uuid"]
            );
            assert_eq!(
                chunk[2]["deliveryId"],
                chunk[1]["attachment"]["delivery_id"]
            );
        }
        let content = records[0]["content"].as_str().unwrap();
        assert!(content.contains("monitor &amp; test"));
        assert!(content.contains("one"));
    }
}