use std::{
    io::{Read, Write},
    sync::{Arc, Mutex},
    thread,
};

use anyhow::{bail, Context};
use dashmap::DashMap;
use portable_pty::{native_pty_system, Child, CommandBuilder, PtySize};
use serde::{Deserialize, Serialize};
use tokio::sync::{broadcast, mpsc};
use uuid::Uuid;

use crate::{
    pty_agent::{default_args, default_command, AgentKind, SessionConfig, SessionInput},
    signals::{detect_approval_request, detect_signal, AgentSignal},
};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionRecord {
    pub id: Uuid,
    pub agent: AgentKind,
    pub status: SessionStatus,
    pub signal: Option<AgentSignal>,
    pub approval_pending: bool,
    pub exit_code: Option<i32>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum SessionStatus {
    Running,
    Exited,
    Cancelled,
    Failed,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum SessionEvent {
    Output { text: String },
    Signal { signal: AgentSignal },
    ApprovalRequested { prompt: String },
    ApprovalDecision { approved: bool },
    Exit { exit_code: Option<i32> },
    Error { message: String },
}

#[derive(Debug)]
pub enum ApprovalDecisionError {
    NotFound,
    NoPending,
    Send(anyhow::Error),
}

struct SessionHandle {
    record: Mutex<SessionRecord>,
    transcript: Mutex<String>,
    input_tx: mpsc::UnboundedSender<SessionInput>,
    resize_tx: mpsc::UnboundedSender<(u16, u16)>,
    event_tx: broadcast::Sender<SessionEvent>,
    approval_requested: Mutex<bool>,
    approval_scan_offset: Mutex<usize>,
    child: Mutex<Option<Box<dyn Child + Send + Sync>>>,
}

#[derive(Default)]
pub struct SessionStore {
    sessions: DashMap<Uuid, Arc<SessionHandle>>,
}

impl SessionStore {
    pub async fn spawn(&self, cfg: SessionConfig) -> anyhow::Result<Uuid> {
        let id = Uuid::new_v4();
        let (event_tx, _) = broadcast::channel(1024);
        let (input_tx, input_rx) = mpsc::unbounded_channel::<SessionInput>();
        let (resize_tx, resize_rx) = mpsc::unbounded_channel::<(u16, u16)>();

        let handle = Arc::new(SessionHandle {
            record: Mutex::new(SessionRecord {
                id,
                agent: cfg.agent,
                status: SessionStatus::Running,
                signal: None,
                approval_pending: false,
                exit_code: None,
            }),
            transcript: Mutex::new(String::new()),
            input_tx,
            resize_tx,
            event_tx,
            approval_requested: Mutex::new(false),
            approval_scan_offset: Mutex::new(0),
            child: Mutex::new(None),
        });
        self.sessions.insert(id, handle.clone());

        thread::Builder::new()
            .name(format!("pty-session-{id}"))
            .spawn(move || {
                if let Err(err) = run_session_thread(cfg, handle.clone(), input_rx, resize_rx) {
                    let _ = handle.event_tx.send(SessionEvent::Error {
                        message: err.to_string(),
                    });
                    if let Ok(mut rec) = handle.record.lock() {
                        rec.status = SessionStatus::Failed;
                    }
                    clear_pending_approval(&handle);
                }
            })
            .context("spawn session thread")?;

        Ok(id)
    }

    pub fn get(&self, id: Uuid) -> Option<SessionRecord> {
        self.sessions
            .get(&id)
            .and_then(|entry| session_record(&entry))
    }

    pub fn list(&self) -> Vec<SessionRecord> {
        let mut records: Vec<_> = self
            .sessions
            .iter()
            .filter_map(|entry| session_record(&entry))
            .collect();
        records.sort_by_key(|record| record.id);
        records
    }

    pub async fn send(&self, id: Uuid, input: SessionInput) -> anyhow::Result<()> {
        self.sessions
            .get(&id)
            .ok_or_else(|| anyhow::anyhow!("session not found"))?
            .input_tx
            .send(input)
            .context("send input")
    }

    pub async fn approval_decision(
        &self,
        id: Uuid,
        approved: bool,
    ) -> Result<(), ApprovalDecisionError> {
        let session = self
            .sessions
            .get(&id)
            .ok_or(ApprovalDecisionError::NotFound)?;
        {
            let record = session.record.lock().expect("session record lock");
            if !matches!(record.status, SessionStatus::Running) {
                drop(record);
                clear_pending_approval(&session);
                return Err(ApprovalDecisionError::NoPending);
            }
        }
        {
            let transcript = session.transcript.lock().expect("transcript lock");
            let mut scan_offset = session
                .approval_scan_offset
                .lock()
                .expect("approval scan offset lock");
            let mut approval_requested = session
                .approval_requested
                .lock()
                .expect("approval requested lock");
            if !*approval_requested {
                return Err(ApprovalDecisionError::NoPending);
            }
            *approval_requested = false;
            *scan_offset = transcript.len();
        }
        session
            .input_tx
            .send(SessionInput {
                text: if approved { "y" } else { "n" }.to_string(),
                enter: true,
            })
            .context("send approval decision")
            .map_err(ApprovalDecisionError::Send)?;
        let _ = session
            .event_tx
            .send(SessionEvent::ApprovalDecision { approved });
        Ok(())
    }

    pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> anyhow::Result<()> {
        self.sessions
            .get(&id)
            .ok_or_else(|| anyhow::anyhow!("session not found"))?
            .resize_tx
            .send((cols, rows))
            .context("send resize")
    }

    pub async fn cancel(&self, id: Uuid) -> anyhow::Result<()> {
        let session = self
            .sessions
            .get(&id)
            .ok_or_else(|| anyhow::anyhow!("session not found"))?;
        if let Some(child) = session.child.lock().expect("child lock").as_mut() {
            child.kill().context("kill child")?;
        }
        if let Ok(mut rec) = session.record.lock() {
            rec.status = SessionStatus::Cancelled;
        }
        clear_pending_approval(&session);
        Ok(())
    }

    pub fn transcript(&self, id: Uuid) -> Option<String> {
        self.sessions
            .get(&id)?
            .transcript
            .lock()
            .ok()
            .map(|t| t.clone())
    }

    pub fn subscribe(&self, id: Uuid) -> Option<broadcast::Receiver<SessionEvent>> {
        Some(self.sessions.get(&id)?.event_tx.subscribe())
    }
}

fn session_record(handle: &SessionHandle) -> Option<SessionRecord> {
    let mut record = handle.record.lock().ok()?.clone();
    record.approval_pending = *handle.approval_requested.lock().ok()?;
    Some(record)
}

fn run_session_thread(
    cfg: SessionConfig,
    handle: Arc<SessionHandle>,
    mut input_rx: mpsc::UnboundedReceiver<SessionInput>,
    mut resize_rx: mpsc::UnboundedReceiver<(u16, u16)>,
) -> anyhow::Result<()> {
    let pty_system = native_pty_system();
    let pair = pty_system
        .openpty(PtySize {
            rows: cfg.rows,
            cols: cfg.cols,
            pixel_width: 0,
            pixel_height: 0,
        })
        .context("open pty")?;

    let command = cfg
        .command
        .clone()
        .unwrap_or_else(|| default_command(cfg.agent).to_string());
    let mut args = if cfg.args.is_empty() {
        default_args(cfg.agent)
    } else {
        cfg.args.clone()
    };
    let mut cmd = CommandBuilder::new(command);
    for arg in args.drain(..) {
        cmd.arg(arg);
    }
    if let Some(cwd) = &cfg.cwd {
        cmd.cwd(cwd);
    }

    let child = pair
        .slave
        .spawn_command(cmd)
        .context("spawn agent command")?;
    drop(pair.slave);
    *handle.child.lock().expect("child lock") = Some(child);

    let master = Arc::new(Mutex::new(pair.master));
    let mut reader = master
        .lock()
        .expect("pty master lock")
        .try_clone_reader()
        .context("clone pty reader")?;
    let writer = Arc::new(Mutex::new(
        master
            .lock()
            .expect("pty master lock")
            .take_writer()
            .context("take pty writer")?,
    ));

    {
        let writer = writer.clone();
        let prompt = cfg.prompt.clone();
        thread::spawn(move || {
            // Paste after the terminal app has had a moment to draw its prompt.
            thread::sleep(std::time::Duration::from_millis(800));
            if let Ok(mut w) = writer.lock() {
                let _ = w.write_all(prompt.as_bytes());
                let _ = w.write_all(b"\n");
                let _ = w.flush();
            }
        });
    }

    {
        let writer = writer.clone();
        thread::spawn(move || {
            while let Some(input) = input_rx.blocking_recv() {
                if let Ok(mut w) = writer.lock() {
                    let _ = w.write_all(input.text.as_bytes());
                    if input.enter {
                        let _ = w.write_all(b"\n");
                    }
                    let _ = w.flush();
                }
            }
        });
    }

    {
        let master = master.clone();
        let event_tx = handle.event_tx.clone();
        thread::spawn(move || {
            while let Some((cols, rows)) = resize_rx.blocking_recv() {
                if let Err(err) = master.lock().expect("pty master lock").resize(PtySize {
                    rows,
                    cols,
                    pixel_width: 0,
                    pixel_height: 0,
                }) {
                    let _ = event_tx.send(SessionEvent::Error {
                        message: format!("resize pty: {err}"),
                    });
                }
            }
        });
    }

    let mut buf = [0u8; 8192];
    loop {
        match reader.read(&mut buf) {
            Ok(0) => break,
            Ok(n) => {
                let text = String::from_utf8_lossy(&buf[..n]).to_string();
                append_output(&handle, &text);
            }
            Err(err) => bail!("read pty: {err}"),
        }
    }

    let exit_code = handle
        .child
        .lock()
        .expect("child lock")
        .as_mut()
        .and_then(|child| child.wait().ok())
        .map(|status| status.exit_code() as i32);

    if let Ok(mut rec) = handle.record.lock() {
        if !matches!(rec.status, SessionStatus::Cancelled) {
            rec.status = SessionStatus::Exited;
        }
        rec.exit_code = exit_code;
    }
    clear_pending_approval(&handle);
    let _ = handle.event_tx.send(SessionEvent::Exit { exit_code });
    Ok(())
}

fn clear_pending_approval(handle: &SessionHandle) {
    if let Ok(transcript) = handle.transcript.lock() {
        if let Ok(mut scan_offset) = handle.approval_scan_offset.lock() {
            *scan_offset = transcript.len();
        }
    }
    if let Ok(mut approval_requested) = handle.approval_requested.lock() {
        *approval_requested = false;
    }
}

fn append_output(handle: &Arc<SessionHandle>, text: &str) {
    if let Ok(mut transcript) = handle.transcript.lock() {
        transcript.push_str(text);
        let approval_scan = if let Ok(mut scan_offset) = handle.approval_scan_offset.lock() {
            let scan_from = floor_char_boundary(&transcript, (*scan_offset).min(transcript.len()));
            let scan = transcript[scan_from..].to_string();
            *scan_offset = if detect_approval_request(&scan) {
                transcript.len()
            } else {
                scan_from.max(floor_char_boundary(
                    &transcript,
                    transcript.len().saturating_sub(256),
                ))
            };
            scan
        } else {
            text.to_string()
        };
        if detect_approval_request(&approval_scan) {
            if let Ok(mut approval_requested) = handle.approval_requested.lock() {
                if !*approval_requested {
                    *approval_requested = true;
                    let _ = handle.event_tx.send(SessionEvent::ApprovalRequested {
                        prompt: transcript_tail(&transcript, 500),
                    });
                }
            }
        }
        if let Some(signal) = detect_signal(&transcript) {
            if let Ok(mut rec) = handle.record.lock() {
                if rec.signal.as_ref() != Some(&signal) {
                    rec.signal = Some(signal.clone());
                    let _ = handle.event_tx.send(SessionEvent::Signal { signal });
                }
            }
        }
    }
    let _ = handle.event_tx.send(SessionEvent::Output {
        text: text.to_string(),
    });
}

fn floor_char_boundary(text: &str, mut index: usize) -> usize {
    while index > 0 && !text.is_char_boundary(index) {
        index -= 1;
    }
    index
}

fn transcript_tail(transcript: &str, max_chars: usize) -> String {
    transcript
        .chars()
        .rev()
        .take(max_chars)
        .collect::<String>()
        .chars()
        .rev()
        .collect()
}

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

    #[test]
    fn append_output_emits_approval_requested_event_once() {
        let (event_tx, mut event_rx) = broadcast::channel(16);
        let (input_tx, _) = mpsc::unbounded_channel::<SessionInput>();
        let (resize_tx, _) = mpsc::unbounded_channel::<(u16, u16)>();
        let handle = Arc::new(SessionHandle {
            record: Mutex::new(SessionRecord {
                id: Uuid::new_v4(),
                agent: AgentKind::Codex,
                status: SessionStatus::Running,
                signal: None,
                approval_pending: false,
                exit_code: None,
            }),
            transcript: Mutex::new(String::new()),
            input_tx,
            resize_tx,
            event_tx,
            approval_requested: Mutex::new(false),
            approval_scan_offset: Mutex::new(0),
            child: Mutex::new(None),
        });

        append_output(&handle, "PLAN_READY");
        append_output(
            &handle,
            "Need approval before running this command. Approve?",
        );
        append_output(&handle, "Approve?");
        append_output(&handle, "COMPLETED");

        let mut events = Vec::new();
        while let Ok(event) = event_rx.try_recv() {
            events.push(event);
        }
        let approval_events = events
            .iter()
            .filter(|event| matches!(event, SessionEvent::ApprovalRequested { .. }))
            .count();
        assert_eq!(approval_events, 1, "{events:?}");
        assert!(
            events.iter().any(|event| matches!(
                event,
                SessionEvent::Signal {
                    signal: AgentSignal::PlanReady
                }
            )),
            "{events:?}"
        );
        assert!(
            events.iter().any(|event| matches!(
                event,
                SessionEvent::Signal {
                    signal: AgentSignal::Completed
                }
            )),
            "{events:?}"
        );
    }
}
