use std::collections::HashMap;
use std::io::{BufRead, BufReader, BufWriter, Read, Write};
use std::path::PathBuf;
use std::time::Duration;

use orfail::OrFail;

use crate::json::JsonObject;
use crate::media::{MediaSample, MediaStreamId, MediaStreamName, MediaStreamNameRegistry};
use crate::metadata::SourceId;
use crate::processor::{
    MediaProcessor, MediaProcessorInput, MediaProcessorOutput, MediaProcessorSpec,
    MediaProcessorWorkloadHint,
};
use crate::stats::ProcessorStats;
use crate::types::EvenUsize;

#[derive(Debug, Clone)]
pub struct PluginCommand {
    pub command: PathBuf,
    pub args: Vec<String>,
    pub input_stream_names: Vec<MediaStreamName>,
    pub output_stream_names: Vec<MediaStreamName>,
}

impl PluginCommand {
    pub fn start(
        &self,
        registry: &mut MediaStreamNameRegistry,
    ) -> orfail::Result<PluginCommandProcessor> {
        let mut input_stream_ids = Vec::new();
        for name in &self.input_stream_names {
            input_stream_ids.push(registry.get_id(name).or_fail()?);
        }

        let mut output_stream_ids = HashMap::new();
        for name in &self.output_stream_names {
            output_stream_ids.insert(
                name.clone(),
                registry.register_name(name.clone()).or_fail()?,
            );
        }

        let mut process = std::process::Command::new(&self.command)
            .args(&self.args)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::inherit())
            .spawn()
            .or_fail_with(|e| format!("failed to start plugin command: {e}"))?;

        let stdin = process
            .stdin
            .take()
            .or_fail_with(|()| "failed to get stdin handle".to_owned())?;

        let stdout = process
            .stdout
            .take()
            .or_fail_with(|()| "failed to get stdout handle".to_owned())?;

        Ok(PluginCommandProcessor {
            process,
            stdin: BufWriter::new(stdin),
            stdout: BufReader::new(stdout),
            input_stream_ids,
            next_request_id: 0,
            output_stream_ids,
        })
    }
}

impl<'text, 'raw> TryFrom<nojson::RawJsonValue<'text, 'raw>> for PluginCommand {
    type Error = nojson::JsonParseError;

    fn try_from(value: nojson::RawJsonValue<'text, 'raw>) -> Result<Self, Self::Error> {
        let obj = JsonObject::new(value)?;

        Ok(Self {
            command: obj.get_required("command")?,
            args: obj.get("args")?.unwrap_or_default(),
            input_stream_names: obj.get("input_stream")?.unwrap_or_default(),
            output_stream_names: obj.get("output_stream")?.unwrap_or_default(),
        })
    }
}

#[derive(Debug)]
pub struct PluginCommandProcessor {
    process: std::process::Child,
    stdin: BufWriter<std::process::ChildStdin>,
    stdout: BufReader<std::process::ChildStdout>,
    input_stream_ids: Vec<MediaStreamId>,
    output_stream_ids: HashMap<MediaStreamName, MediaStreamId>,
    next_request_id: u64,
}

impl PluginCommandProcessor {
    fn cast<T>(
        &mut self,
        notification: &JsonRpcRequest<T>,
        payload: Option<&[u8]>,
    ) -> orfail::Result<()>
    where
        T: nojson::DisplayJson,
    {
        let notification = nojson::Json(notification).to_string();
        writeln!(self.stdin, "Content-Length: {}", notification.len()).or_fail()?;
        writeln!(self.stdin, "Content-Type: application/json").or_fail()?;
        writeln!(self.stdin).or_fail()?;
        write!(self.stdin, "{notification}").or_fail()?;

        if let Some(payload) = payload {
            writeln!(self.stdin, "Content-Length: {}", payload.len()).or_fail()?;
            writeln!(self.stdin, "Content-Type: application/octet-stream").or_fail()?;
            writeln!(self.stdin).or_fail()?;
            self.stdin.write_all(payload).or_fail()?;
        }

        self.stdin.flush().or_fail()?;
        Ok(())
    }

    fn call<T, U>(&mut self, request: &JsonRpcRequest<T>) -> orfail::Result<U>
    where
        T: nojson::DisplayJson,
        U: for<'text, 'raw> TryFrom<nojson::RawJsonValue<'text, 'raw>>,
    {
        let request = nojson::Json(request).to_string();
        writeln!(self.stdin, "Content-Length: {}", request.len()).or_fail()?;
        writeln!(self.stdin, "Content-Type: application/json").or_fail()?;
        writeln!(self.stdin).or_fail()?;
        write!(self.stdin, "{request}").or_fail()?;
        self.stdin.flush().or_fail()?;

        // ヘッダーを読み取ってコンテンツ長を取得
        let mut content_length = None;
        let mut line = String::new();

        loop {
            line.clear();
            self.stdout.read_line(&mut line).or_fail()?;

            if line.trim().is_empty() {
                // 空行はヘッダー終了を示す
                break;
            }

            if let Some(header_value) = line.strip_prefix("Content-Length: ") {
                content_length = Some(
                    header_value
                        .trim()
                        .parse::<usize>()
                        .or_fail_with(|e| format!("invalid content length: {e}"))?,
                );
            }
        }

        let content_length = content_length
            .or_fail_with(|()| "missing Content-Length header in response".to_owned())?;

        // JSON レスポンス本体を読み取り
        let mut response_buffer = vec![0u8; content_length];
        self.stdout.read_exact(&mut response_buffer).or_fail()?;

        let response_text = std::str::from_utf8(&response_buffer)
            .or_fail_with(|e| format!("invalid UTF-8 in response: {e}"))?;

        // JSON-RPC レスポンスをパース
        let json = nojson::RawJson::parse(response_text)
            .or_fail_with(|e| format!("failed to parse JSON response: {e}"))?;
        if let Some(error) = json.value().to_member("error").or_fail()?.get() {
            return Err(orfail::Failure::new(format!("JSON-RPC error: {error}",)));
        }

        let result = json
            .value()
            .to_member("result")
            .or_fail()?
            .required()
            .or_fail()?;
        U::try_from(result).map_err(|_| {
            orfail::Failure::new("failed to convert response to expected type".to_owned())
        })
    }

    fn read_payload(&mut self) -> orfail::Result<Vec<u8>> {
        let mut content_length = None;
        let mut line = String::new();

        loop {
            line.clear();
            self.stdout.read_line(&mut line).or_fail()?;

            if line.trim().is_empty() {
                break;
            }

            if let Some(header_value) = line.strip_prefix("Content-Length: ") {
                content_length = Some(
                    header_value
                        .trim()
                        .parse::<usize>()
                        .or_fail_with(|e| format!("invalid content length: {e}"))?,
                );
            }
        }

        let content_length = content_length
            .or_fail_with(|()| "missing Content-Length header for payload".to_owned())?;

        let mut payload_data = vec![0u8; content_length];
        self.stdout.read_exact(&mut payload_data).or_fail()?;

        Ok(payload_data)
    }
}

impl MediaProcessor for PluginCommandProcessor {
    fn spec(&self) -> MediaProcessorSpec {
        MediaProcessorSpec {
            input_stream_ids: self.input_stream_ids.clone(),
            output_stream_ids: self.output_stream_ids.values().copied().collect(),
            stats: ProcessorStats::other("plugin_command"),
            workload_hint: MediaProcessorWorkloadHint::PLUGIN,
        }
    }

    fn process_input(&mut self, input: MediaProcessorInput) -> orfail::Result<()> {
        match input.sample {
            None => {
                self.input_stream_ids.retain(|id| *id != input.stream_id);

                let req = JsonRpcRequest::notification(
                    "notify_eos",
                    nojson::object(|f| f.member("stream_id", input.stream_id)),
                );
                self.cast(&req, None).or_fail()?;
            }
            Some(MediaSample::Audio(data)) => {
                (data.format == crate::audio::AudioFormat::I16Be).or_fail()?;
                let req = JsonRpcRequest::notification(
                    "notify_audio",
                    nojson::object(|f| {
                        f.member("stream_id", input.stream_id)?;
                        f.member("stereo", data.stereo)?;
                        f.member("sample_rate", data.sample_rate)?;
                        f.member("timestamp_us", data.timestamp.as_micros())?;
                        f.member("duration_us", data.duration.as_micros())?;
                        Ok(())
                    }),
                );
                self.cast(&req, Some(&data.data)).or_fail()?;
            }
            Some(MediaSample::Video(frame)) => {
                let req = JsonRpcRequest::notification(
                    "notify_video",
                    nojson::object(|f| {
                        f.member("stream_id", input.stream_id)?;
                        f.member("width", frame.width)?;
                        f.member("height", frame.height)?;
                        f.member("timestamp_us", frame.timestamp.as_micros())?;
                        f.member("duration_us", frame.duration.as_micros())?;
                        Ok(())
                    }),
                );
                let bgr_data = frame.to_bgr_data().or_fail()?;
                self.cast(&req, Some(&bgr_data)).or_fail()?;
            }
        }
        Ok(())
    }

    fn process_output(&mut self) -> orfail::Result<MediaProcessorOutput> {
        let id = self.next_request_id;
        self.next_request_id += 1;

        let req = JsonRpcRequest::request("poll_output", id, ());
        let res: PollOutputResponse = self.call(&req).or_fail()?;

        let output = match res {
            PollOutputResponse::WaitingInputAny => MediaProcessorOutput::awaiting_any(),
            PollOutputResponse::WaitingInput { stream_id } => {
                MediaProcessorOutput::pending(stream_id)
            }
            PollOutputResponse::AudioData {
                stream_name,
                stereo,
                sample_rate,
                timestamp,
                duration,
            } => {
                let stream_id = self
                    .output_stream_ids
                    .get(&stream_name)
                    .copied()
                    .or_fail()?;
                let audio_data = self.read_payload().or_fail()?;

                let mut audio_sample = crate::audio::AudioData {
                    source_id: None,
                    data: audio_data,
                    format: crate::audio::AudioFormat::I16Be,
                    stereo,
                    sample_rate: sample_rate as u16,
                    timestamp,
                    duration,
                    sample_entry: None,
                };
                audio_sample.source_id = Some(SourceId::new(stream_name.get()));

                MediaProcessorOutput::audio_data(stream_id, audio_sample)
            }
            PollOutputResponse::VideoFrame {
                stream_name,
                width,
                height,
                timestamp,
                duration,
            } => {
                let stream_id = self
                    .output_stream_ids
                    .get(&stream_name)
                    .copied()
                    .or_fail()?;
                let frame_data = self.read_payload().or_fail()?;

                let mut video_frame = crate::video::VideoFrame::from_bgr_data(
                    &frame_data,
                    width,
                    height,
                    timestamp,
                    duration,
                )
                .or_fail()?;
                video_frame.source_id = Some(SourceId::new(stream_name.get()));
                MediaProcessorOutput::video_frame(stream_id, video_frame)
            }
            PollOutputResponse::Finished => MediaProcessorOutput::Finished,
        };
        Ok(output)
    }
}

impl Drop for PluginCommandProcessor {
    fn drop(&mut self) {
        let _ = self.process.kill();
        let _ = self.process.wait();
    }
}

#[derive(Debug)]
pub struct JsonRpcRequest<'a, T> {
    method: &'a str,
    id: Option<u64>,
    params: T,
}

impl<'a, T> JsonRpcRequest<'a, T> {
    pub fn notification(method: &'a str, params: T) -> Self {
        Self {
            method,
            id: None,
            params,
        }
    }

    pub fn request(method: &'a str, id: u64, params: T) -> Self {
        Self {
            method,
            id: Some(id),
            params,
        }
    }
}

impl<'a, T> nojson::DisplayJson for JsonRpcRequest<'a, T>
where
    T: nojson::DisplayJson,
{
    fn fmt(&self, f: &mut nojson::JsonFormatter<'_, '_>) -> std::fmt::Result {
        f.object(|f| {
            f.member("jsonrpc", "2.0")?;
            f.member("method", self.method)?;
            if let Some(id) = self.id {
                f.member("id", id)?;
            }
            f.member("params", &self.params)?;
            Ok(())
        })
    }
}

#[derive(Debug)]
pub enum PollOutputResponse {
    WaitingInputAny,
    WaitingInput {
        stream_id: MediaStreamId,
    },
    AudioData {
        stream_name: MediaStreamName,
        stereo: bool,
        sample_rate: u32,
        timestamp: Duration,
        duration: Duration,
    },
    VideoFrame {
        stream_name: MediaStreamName,
        width: EvenUsize,
        height: EvenUsize,
        timestamp: Duration,
        duration: Duration,
    },
    Finished,
}

impl<'text, 'raw> TryFrom<nojson::RawJsonValue<'text, 'raw>> for PollOutputResponse {
    type Error = nojson::JsonParseError;

    fn try_from(value: nojson::RawJsonValue<'text, 'raw>) -> Result<Self, Self::Error> {
        let obj = JsonObject::new(value)?;
        let response_type: String = obj.get_required("type")?;

        match response_type.as_str() {
            "waiting_input_any" => Ok(Self::WaitingInputAny),
            "waiting_input" => {
                let stream_id = obj.get_required("stream_id")?;
                Ok(Self::WaitingInput { stream_id })
            }
            "audio_data" => {
                let stream_name = obj.get_required("stream_name")?;
                let stereo = obj.get_required("stereo")?;
                let sample_rate = obj.get_required("sample_rate")?;
                let timestamp_us: u64 = obj.get_required("timestamp_us")?;
                let duration_us: u64 = obj.get_required("duration_us")?;
                Ok(Self::AudioData {
                    stream_name,
                    stereo,
                    sample_rate,
                    timestamp: Duration::from_micros(timestamp_us),
                    duration: Duration::from_micros(duration_us),
                })
            }
            "video_frame" => {
                let stream_name = obj.get_required("stream_name")?;
                let width_raw: u32 = obj.get_required("width")?;
                let height_raw: u32 = obj.get_required("height")?;
                let timestamp_us: u64 = obj.get_required("timestamp_us")?;
                let duration_us: u64 = obj.get_required("duration_us")?;

                let width = EvenUsize::new(width_raw as usize)
                    .ok_or_else(|| value.invalid("width must be even"))?;
                let height = EvenUsize::new(height_raw as usize)
                    .ok_or_else(|| value.invalid("height must be even"))?;

                Ok(Self::VideoFrame {
                    stream_name,
                    width,
                    height,
                    timestamp: Duration::from_micros(timestamp_us),
                    duration: Duration::from_micros(duration_us),
                })
            }
            "finished" => Ok(Self::Finished),
            unknown => {
                Err(value.invalid(format!("unknown poll output response type: {unknown:?}")))
            }
        }
    }
}
