//! Exec channels over an existing SSH connection (port of `_execFixed`/`_execStream`).
//! Output is decoded incrementally, bounded by an output limit and a deadline, and the
//! channel is cancelled when the session is lost. No API accepts commands from the UI:
//! callers are Core services with fixed scripts.

use crate::core::error::{CoreError, CoreResult};
use crate::core::ssh::utf8::Utf8Decoder;
use russh::client::Msg;
use russh::{ChannelMsg, ChannelReadHalf, ChannelWriteHalf};
use std::time::Duration;
use tokio::sync::watch;
use tokio::time::Instant;

pub const MAX_COMMAND_OUTPUT: usize = 512 * 1024;

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ExecEvent {
    Stdout,
    Stderr,
    Done,
}

#[derive(Debug, Clone)]
pub struct ExecResult {
    pub code: Option<i64>,
    pub signal: Option<String>,
    pub stdout: String,
    pub stderr: String,
}

pub struct ExecChannel {
    read: ChannelReadHalf,
    write: ChannelWriteHalf<Msg>,
    pub stdout: String,
    pub stderr: String,
    decode_out: Utf8Decoder,
    decode_err: Utf8Decoder,
    received: usize,
    limit: usize,
    deadline: Instant,
    lost: watch::Receiver<bool>,
    code: Option<i64>,
    signal: Option<String>,
    done: bool,
    eof_sent: bool,
}

impl ExecChannel {
    pub fn new(read: ChannelReadHalf, write: ChannelWriteHalf<Msg>, timeout: Duration, limit: usize, lost: watch::Receiver<bool>) -> Self {
        Self {
            read,
            write,
            stdout: String::new(),
            stderr: String::new(),
            decode_out: Utf8Decoder::default(),
            decode_err: Utf8Decoder::default(),
            received: 0,
            limit,
            deadline: Instant::now() + timeout,
            lost,
            code: None,
            signal: None,
            done: false,
            eof_sent: false,
        }
    }

    pub async fn write(&self, data: &[u8]) {
        if self.done || self.eof_sent {
            return;
        }
        let _ = self.write.data_bytes(data.to_vec()).await;
    }

    pub async fn eof(&mut self) {
        if !self.eof_sent && !self.done {
            self.eof_sent = true;
            let _ = self.write.eof().await;
        }
    }

    pub async fn close(&self) {
        let _ = self.write.close().await;
    }

    pub fn is_done(&self) -> bool {
        self.done
    }

    /// Waits for the next output chunk (appended to `stdout`/`stderr`) or the end of the channel.
    pub async fn next(&mut self) -> CoreResult<ExecEvent> {
        if self.done {
            return Ok(ExecEvent::Done);
        }
        if *self.lost.borrow() {
            return Err(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost."));
        }
        loop {
            let message = tokio::select! {
                _ = tokio::time::sleep_until(self.deadline) => {
                    self.close().await;
                    return Err(CoreError::new("REMOTE_COMMAND_TIMEOUT", "The remote command timed out."));
                }
                changed = self.lost.changed() => {
                    if changed.is_err() || *self.lost.borrow() {
                        return Err(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost."));
                    }
                    continue;
                }
                message = self.read.wait() => message,
            };
            match message {
                Some(ChannelMsg::Data { data }) => {
                    self.account(data.len()).await?;
                    let text = self.decode_out.write(&data);
                    self.stdout.push_str(&text);
                    return Ok(ExecEvent::Stdout);
                }
                Some(ChannelMsg::ExtendedData { data, ext }) => {
                    self.account(data.len()).await?;
                    if ext == 1 {
                        let text = self.decode_err.write(&data);
                        self.stderr.push_str(&text);
                        return Ok(ExecEvent::Stderr);
                    }
                }
                Some(ChannelMsg::ExitStatus { exit_status }) => self.code = Some(exit_status as i64),
                Some(ChannelMsg::ExitSignal { signal_name, .. }) => self.signal = Some(format!("{signal_name:?}")),
                Some(ChannelMsg::Close) | None => {
                    self.done = true;
                    let rest = self.decode_out.end();
                    self.stdout.push_str(&rest);
                    let rest = self.decode_err.end();
                    self.stderr.push_str(&rest);
                    return Ok(ExecEvent::Done);
                }
                Some(ChannelMsg::OpenFailure(reason)) => {
                    return Err(CoreError::new("REMOTE_COMMAND_FAILED", format!("The remote channel could not be opened: {reason:?}")));
                }
                Some(_) => {}
            }
        }
    }

    async fn account(&mut self, length: usize) -> CoreResult<()> {
        self.received += length;
        if self.received > self.limit {
            self.close().await;
            return Err(CoreError::new("REMOTE_OUTPUT_TOO_LARGE", "The remote response is too large."));
        }
        Ok(())
    }

    /// Reads until the channel closes; returns the exit code instead of failing on non-zero.
    pub async fn finish(mut self) -> CoreResult<ExecResult> {
        while self.next().await? != ExecEvent::Done {}
        Ok(ExecResult { code: self.code, signal: self.signal.clone(), stdout: std::mem::take(&mut self.stdout), stderr: std::mem::take(&mut self.stderr) })
    }
}