use crate::core::codec::Codec;
use crate::core::context::decoder_stream::DecoderStream;
use crate::core::context::pre_mux_queue::PreMuxQueueSender;
use crate::core::context::PacketBox;
use crate::core::hwaccel::HWAccelID;
use crate::core::scheduler::input_controller::SchNode;
use crate::error::OpenInputError;
use crate::filter::frame_pipeline::FramePipeline;
use crate::raw::FormatContext;
use crossbeam_channel::Sender;
use ffmpeg_sys_next::AVHWDeviceType::AV_HWDEVICE_TYPE_NONE;
use ffmpeg_sys_next::AVMediaType::{AVMEDIA_TYPE_AUDIO, AVMEDIA_TYPE_SUBTITLE, AVMEDIA_TYPE_VIDEO};
use ffmpeg_sys_next::AVPixelFormat::{
    AV_PIX_FMT_CUDA, AV_PIX_FMT_MEDIACODEC, AV_PIX_FMT_NONE, AV_PIX_FMT_QSV,
};
use ffmpeg_sys_next::{
    av_channel_layout_default, av_codec_is_decoder, av_codec_iterate, av_get_pix_fmt,
    av_hwdevice_find_type_by_name, av_hwdevice_get_type_name, avcodec_descriptor_get,
    avcodec_descriptor_get_by_name, avcodec_find_decoder, avcodec_find_decoder_by_name,
    avcodec_get_hw_config, AVChannelOrder, AVCodecID, AVCodecParameters, AVFormatContext,
    AVHWDeviceType, AVMediaType, AVPixelFormat, AVRational, AVERROR, EINVAL,
};
use log::{debug, error, warn};
use std::collections::HashMap;
use std::ffi::{CStr, CString};
use std::ptr::{null, null_mut};
use std::sync::atomic::AtomicBool;
use std::sync::Arc;

/// Muxer-side wiring a streamcopy destination needs to route its packets
/// through the encoders' deferred-start gate + per-stream pre-mux queue, rather
/// than blocking the shared demuxer on the live bounded queue before the muxer
/// starts. Built by `Muxer::new_copy_stream` and carried lockstep with `dsts`.
/// fftools routes demux->mux copy packets through the same `sch_send` path as
/// encoders (ffmpeg_sched.c:2038-2077).
pub(crate) struct CopyMuxHandle {
    pub(crate) pre_sender: PreMuxQueueSender,
    pub(crate) gate: Arc<crate::core::context::MuxStartGate>,
}

pub(crate) struct Demuxer {
    pub(crate) url: String,
    /// Owns the input `AVFormatContext`. `Some` from construction until the demux
    /// worker `take()`s it (see `demux_task`); `None` afterward. Dropping the
    /// `Demuxer` before that hand-off (build succeeded but the job never started)
    /// frees the context via `FormatContext`'s Drop — no manual `Drop` needed.
    pub(crate) in_fmt_ctx: Option<FormatContext>,
    pub(crate) ts_offset: i64,
    pub(crate) frame_pipelines: Option<Vec<FramePipeline>>,

    pub(crate) readrate: Option<f32>,
    pub(crate) start_time_us: Option<i64>,
    pub(crate) recording_time_us: Option<i64>,
    pub(crate) exit_on_error: Option<bool>,
    pub(crate) stream_loop: Option<i32>,
    pub(crate) copy_ts: bool,

    /// Automatically rotate video based on display matrix metadata.
    /// Default is true (enabled).
    ///
    /// FFmpeg CLI: `-autorotate 0/1`
    /// FFmpeg source: `ffmpeg_demux.c:1270`, `ffmpeg_filter.c:1744-1778` (FFmpeg 7.x)
    pub(crate) autorotate: bool,

    /// Timestamp scale factor for pts/dts values.
    /// Default is 1.0 (no scaling).
    ///
    /// FFmpeg CLI: `-itsscale <scale>`
    /// FFmpeg source: `ffmpeg_demux.c:404-406` (FFmpeg 7.x)
    pub(crate) ts_scale: f64,

    /// Forced framerate for the input video stream.
    /// When set (num != 0), overrides the DTS estimation logic to use framerate-based
    /// grid calculation. When `{0, 0}` (default), packet duration is used for DTS
    /// estimation, matching FFmpeg CLI behavior when `-r` is not specified.
    ///
    /// FFmpeg CLI: `-r <rate>`
    /// FFmpeg source: `ffmpeg.h:452`, `ffmpeg_demux.c:329-333` (FFmpeg 7.x)
    pub(crate) framerate: AVRational,

    #[cfg(windows)]
    pub(crate) hwaccel: Option<String>,

    pub(crate) start_time_effective: i64,

    pub(crate) node: Arc<SchNode>,
    streams: Vec<DecoderStream>,
    dsts: Vec<(Sender<PacketBox>, usize, Option<usize>)>,
    /// Parallel to `dsts`: for a `-shortest` streamcopy destination, the mux
    /// stream's `source_finished` flag (`SchNode::MuxStream`), so the demux can
    /// stop producing a follower the muxer's `sq_mux` has cascade-finished
    /// (Architecture Y'). `None` for every non-shortest / non-copy destination.
    dst_source_finished: Vec<Option<Arc<AtomicBool>>>,
    /// Parallel to `dsts`: a streamcopy destination's muxer gate + pre-mux
    /// queue, so the demux routes copy packets through the same deferred-start
    /// path encoders use. `None` for every decoder destination (which keeps the
    /// direct live send). Kept strictly in lockstep with `dsts`.
    dst_mux_gate: Vec<Option<CopyMuxHandle>>,
}

// SAFETY: Demuxer can be sent to another thread. The raw FFmpeg pointers are only
// accessed from the owning thread. Note: Demuxer is NOT Sync because it contains
// frame_pipelines with Box<dyn FrameFilter> which only implements Send.
unsafe impl Send for Demuxer {}

// No manual `Drop`: the `Option<FormatContext>` field owns the context and frees it
// on drop (input path, custom-IO-aware) whether it was taken by the worker (`None`)
// or the job never started (`Some`). This replaces the old hand-written Drop that
// called `in_fmt_ctx_free` + nulled the raw pointer.

impl Demuxer {
    /// Raw input `AVFormatContext` pointer for read-only FFI (field reads,
    /// `av_read_frame`, seek, `av_guess_frame_rate`). Returns null once the demux
    /// worker has taken ownership. Callers must not free through it.
    pub(crate) fn in_fmt_ctx_ptr(&self) -> *mut AVFormatContext {
        // SAFETY: as_ptr just returns the stored pointer.
        self.in_fmt_ctx
            .as_ref()
            .map_or(std::ptr::null_mut(), |fc| unsafe { fc.as_ptr() })
    }

    pub(crate) fn new(
        url: String,
        in_fmt_ctx: FormatContext,
        ts_offset: i64,
        frame_pipelines: Option<Vec<FramePipeline>>,
        video_codec: Option<String>,
        audio_codec: Option<String>,
        subtitle_codec: Option<String>,
        video_codec_opts: Option<HashMap<CString, CString>>,
        audio_codec_opts: Option<HashMap<CString, CString>>,
        subtitle_codec_opts: Option<HashMap<CString, CString>>,
        readrate: Option<f32>,
        start_time_us: Option<i64>,
        recording_time_us: Option<i64>,
        exit_on_error: Option<bool>,
        stream_loop: Option<i32>,
        hwaccel: Option<String>,
        hwaccel_device: Option<String>,
        hwaccel_output_format: Option<String>,
        copy_ts: bool,
        autorotate: bool,
        ts_scale: f64,
        framerate: AVRational,
        log_level_offset: i32,
    ) -> crate::error::Result<Self> {
        // `in_fmt_ctx` is owned here: if `init_streams` fails, it drops exactly
        // once (input-side free), which is why no external guard is needed across
        // this fallible constructor. `init_streams` only reads the context.
        // SAFETY: as_ptr returns the live pointer for read-only stream probing.
        let streams = Self::init_streams(
            unsafe { in_fmt_ctx.as_ptr() },
            video_codec,
            audio_codec,
            subtitle_codec,
            video_codec_opts,
            audio_codec_opts,
            subtitle_codec_opts,
            hwaccel.clone(),
            hwaccel_device,
            hwaccel_output_format,
            framerate,
            log_level_offset,
        )?;

        Ok(Self {
            url,
            in_fmt_ctx: Some(in_fmt_ctx),
            ts_offset,
            frame_pipelines,
            readrate,
            start_time_us,
            recording_time_us,
            exit_on_error,
            stream_loop,
            copy_ts,
            autorotate,
            ts_scale,
            framerate,
            #[cfg(windows)]
            hwaccel,
            node: Arc::new(SchNode::Demux {
                waiter: Arc::new(Default::default()),
                task_exited: Arc::new(Default::default()),
            }),
            streams,
            dsts: vec![],
            dst_source_finished: vec![],
            dst_mux_gate: vec![],
            start_time_effective: 0,
        })
    }

    fn init_streams(
        fmt_ctx: *mut AVFormatContext,
        video_codec: Option<String>,
        audio_codec: Option<String>,
        subtitle_codec: Option<String>,
        video_codec_opts: Option<HashMap<CString, CString>>,
        audio_codec_opts: Option<HashMap<CString, CString>>,
        subtitle_codec_opts: Option<HashMap<CString, CString>>,
        hwaccel: Option<String>,
        hwaccel_device: Option<String>,
        hwaccel_output_format: Option<String>,
        forced_framerate: AVRational,
        log_level_offset: i32,
    ) -> crate::error::Result<Vec<DecoderStream>> {
        unsafe {
            let stream_count = (*fmt_ctx).nb_streams;
            let mut streams = Vec::with_capacity(stream_count as usize);

            for i in 0..stream_count {
                let st = *(*fmt_ctx).streams.add(i as usize);

                let duration = (*st).duration;
                let time_base = (*st).time_base;
                // A user-forced framerate replaces the container's average
                // for video (ffmpeg_demux.c:1002-1006: forced -> flags |=
                // FRAMERATE_FORCED, else framerate = avg_frame_rate).
                let codec_type_early = (*(*st).codecpar).codec_type;
                let framerate_forced =
                    forced_framerate.num != 0 && codec_type_early == AVMEDIA_TYPE_VIDEO;
                let avg_framerate = if framerate_forced {
                    forced_framerate
                } else {
                    (*st).avg_frame_rate
                };
                let codec_parameters = (*st).codecpar;
                let codec_type = (*codec_parameters).codec_type;

                let (hwaccel_id, hwaccel_device_type, hwaccel_device, hwaccel_output_format) =
                    find_hwaccel(
                        codec_type,
                        hwaccel.clone(),
                        hwaccel_device.clone(),
                        hwaccel_output_format.clone(),
                    )?;

                let codec_id = (*codec_parameters).codec_id;

                if codec_type == AVMEDIA_TYPE_AUDIO
                    && (*codec_parameters).ch_layout.order
                        == AVChannelOrder::AV_CHANNEL_ORDER_UNSPEC
                    && (*codec_parameters).ch_layout.nb_channels > 0
                {
                    av_channel_layout_default(
                        &mut (*codec_parameters).ch_layout,
                        (*codec_parameters).ch_layout.nb_channels,
                    );
                }

                let codec_name =
                    get_codec_name(codec_type, &video_codec, &audio_codec, &subtitle_codec);
                let decoder = choose_decoder(
                    codec_name,
                    codec_type,
                    codec_parameters,
                    codec_id,
                    hwaccel_id,
                    hwaccel_device_type,
                )?;
                let codec_desc = avcodec_descriptor_get(codec_id);
                let codec_opts = get_codec_opts(
                    codec_type,
                    &video_codec_opts,
                    &audio_codec_opts,
                    &subtitle_codec_opts,
                );

                let stream = DecoderStream::new(
                    i as usize,
                    st,
                    codec_parameters,
                    codec_type,
                    match decoder {
                        Some(decoder) => decoder.as_ptr(),
                        None => null(),
                    },
                    codec_desc,
                    codec_opts,
                    duration,
                    time_base,
                    avg_framerate,
                    framerate_forced,
                    hwaccel_id,
                    hwaccel_device_type,
                    hwaccel_device,
                    hwaccel_output_format,
                    log_level_offset,
                );
                streams.push(stream);
            }

            Ok(streams)
        }
    }

    pub(crate) fn add_packet_dst(
        &mut self,
        packet_dst: Sender<PacketBox>,
        input_stream_index: usize,
        output_stream_index: usize,
        source_finished: Option<Arc<AtomicBool>>,
        mux_gate: CopyMuxHandle,
    ) {
        self.dsts
            .push((packet_dst, input_stream_index, Some(output_stream_index)));
        // Kept strictly in lockstep with `dsts` so `dst_source_finished[i]` and
        // `dst_mux_gate[i]` always describe `dsts[i]`.
        self.dst_source_finished.push(source_finished);
        // Every streamcopy destination carries a gate handle so it routes
        // through the encoders' pre-mux queue instead of blocking the demuxer.
        self.dst_mux_gate.push(Some(mux_gate));
    }

    pub(crate) fn get_streams(&self) -> &Vec<DecoderStream> {
        &self.streams
    }

    pub(crate) fn get_streams_mut(&mut self) -> &mut Vec<DecoderStream> {
        &mut self.streams
    }

    pub(crate) fn get_stream_mut(&mut self, index: usize) -> &mut DecoderStream {
        &mut self.streams[index]
    }

    pub(crate) fn get_stream(&self, index: usize) -> &DecoderStream {
        &self.streams[index]
    }

    pub(crate) fn connect_stream(&mut self, index: usize) {
        if self.streams[index].is_used() {
            return;
        }
        let (sender, receiver) = crossbeam_channel::bounded(8);
        self.dsts.push((sender, index, None));
        // A decoded (non-copy) destination is never an `sq_mux` follower and
        // never routes through the mux gate (it feeds a decoder, not the mux).
        self.dst_source_finished.push(None);
        self.dst_mux_gate.push(None);
        self.streams[index].set_src(receiver);
    }

    pub(crate) fn take_dsts(&mut self) -> Vec<(Sender<PacketBox>, usize, Option<usize>)> {
        std::mem::take(&mut self.dsts)
    }

    pub(crate) fn take_dst_source_finished(&mut self) -> Vec<Option<Arc<AtomicBool>>> {
        std::mem::take(&mut self.dst_source_finished)
    }

    pub(crate) fn take_dst_mux_gate(&mut self) -> Vec<Option<CopyMuxHandle>> {
        std::mem::take(&mut self.dst_mux_gate)
    }

    pub(crate) fn destination_is_empty(&mut self) -> bool {
        self.dsts.is_empty()
    }
}

fn get_codec_name(
    codec_type: AVMediaType,
    video_codec: &Option<String>,
    audio_codec: &Option<String>,
    subtitle_codec: &Option<String>,
) -> Option<String> {
    if codec_type == AVMEDIA_TYPE_VIDEO {
        video_codec.clone()
    } else if codec_type == AVMEDIA_TYPE_AUDIO {
        audio_codec.clone()
    } else if codec_type == AVMEDIA_TYPE_SUBTITLE {
        subtitle_codec.clone()
    } else {
        None
    }
}

fn get_codec_opts(
    codec_type: AVMediaType,
    video_codec_opts: &Option<HashMap<CString, CString>>,
    audio_codec_opts: &Option<HashMap<CString, CString>>,
    subtitle_codec_opts: &Option<HashMap<CString, CString>>,
) -> Option<HashMap<CString, CString>> {
    if codec_type == AVMEDIA_TYPE_VIDEO {
        video_codec_opts.clone()
    } else if codec_type == AVMEDIA_TYPE_AUDIO {
        audio_codec_opts.clone()
    } else if codec_type == AVMEDIA_TYPE_SUBTITLE {
        subtitle_codec_opts.clone()
    } else {
        None
    }
}

fn choose_decoder(
    codec_name: Option<String>,
    codec_type: AVMediaType,
    codec_parameters: *mut AVCodecParameters,
    codec_id: AVCodecID,
    hwaccel_id: HWAccelID,
    hwaccel_device_type: AVHWDeviceType,
) -> crate::error::Result<Option<Codec>> {
    match codec_name {
        Some(codec_name) => unsafe {
            let codec_cstr = CString::new(codec_name.clone())?;

            let mut codec = avcodec_find_decoder_by_name(codec_cstr.as_ptr());
            let desc = avcodec_descriptor_get_by_name(codec_cstr.as_ptr());

            if codec.is_null() && !desc.is_null() {
                codec = avcodec_find_decoder((*desc).id);
                if !codec.is_null() {
                    let codec_name = (*codec).name;
                    let codec_name = CStr::from_ptr(codec_name).to_str();
                    let desc_name = (*desc).name;
                    let desc_name = CStr::from_ptr(desc_name).to_str();
                    if let (Ok(codec_name), Ok(desc_name)) = (codec_name, desc_name) {
                        debug!("Matched decoder '{codec_name}' for codec '{desc_name}'.");
                    }
                }
            }

            if codec.is_null() {
                error!("Unknown decoder '{codec_name}'");
                return Err(crate::error::DecoderError::NotFound(codec_name).into());
            }

            if (*codec).type_ != codec_type {
                error!("Invalid decoder type '{codec_name}'");
                return Err(OpenInputError::InvalidArgument.into());
            }
            let codec_id = (*codec).id;

            (*codec_parameters).codec_id = codec_id;
            if (*codec_parameters).codec_type != codec_type {
                (*codec_parameters).codec_type = codec_type;
            }

            Ok(Some(Codec::new(codec)))
        },
        None => {
            if codec_type == AVMEDIA_TYPE_VIDEO
                && hwaccel_id == HWAccelID::HwaccelGeneric
                && hwaccel_device_type != AV_HWDEVICE_TYPE_NONE
            {
                let mut i = null_mut();
                loop {
                    let c = unsafe { av_codec_iterate(&mut i) };
                    if c.is_null() {
                        break;
                    }
                    unsafe {
                        if (*c).id != codec_id || av_codec_is_decoder(c) == 0 {
                            continue;
                        }
                    }

                    let mut j = 0;
                    loop {
                        unsafe {
                            let config = avcodec_get_hw_config(c, j);
                            if config.is_null() {
                                break;
                            }
                            if (*config).device_type == hwaccel_device_type {
                                let name = (*c).name;
                                let name = CStr::from_ptr(name).to_str();
                                let type_name = av_hwdevice_get_type_name(hwaccel_device_type);
                                let type_name = CStr::from_ptr(type_name).to_str();
                                if let (Ok(name), Ok(type_name)) = (name, type_name) {
                                    debug!("Selecting decoder '{name}' because of requested hwaccel method {type_name}");
                                }

                                return Ok(Some(Codec::new(c)));
                            }
                        }
                        j += 1;
                    }
                }
            }

            let c = unsafe { avcodec_find_decoder(codec_id) };
            if c.is_null() {
                Ok(None)
            } else {
                Ok(Some(Codec::new(c)))
            }
        }
    }
}

fn find_hwaccel(
    codec_type: AVMediaType,
    hwaccel: Option<String>,
    hwaccel_device: Option<String>,
    hwaccel_output_format: Option<String>,
) -> crate::error::Result<(HWAccelID, AVHWDeviceType, Option<String>, AVPixelFormat)> {
    if codec_type != AVMediaType::AVMEDIA_TYPE_VIDEO {
        return Ok((
            HWAccelID::HwaccelNone,
            AVHWDeviceType::AV_HWDEVICE_TYPE_NONE,
            None,
            AV_PIX_FMT_NONE,
        ));
    }
    let mut out_hwaccel_output_format = AV_PIX_FMT_NONE;

    match (&hwaccel, hwaccel_output_format) {
        (Some(hwaccel), None) if hwaccel == "cuvid" => {
            warn!("WARNING: Defaulting hwaccel_output_format to cuda for compatibility with older.This behavior is DEPRECATED and will be removed in the future.Please explicitly set \"hwaccel_output_format\" to \"cuda\" using the appropriate API method.");
            out_hwaccel_output_format = AV_PIX_FMT_CUDA;
        }
        (Some(hwaccel), None) if hwaccel == "qsv" => {
            warn!("WARNING: Defaulting hwaccel_output_format to qsv for compatibility with older.This behavior is DEPRECATED and will be removed in the future.Please explicitly set \"hwaccel_output_format\" to \"qsv\" using the appropriate API method.");
            out_hwaccel_output_format = AV_PIX_FMT_QSV;
        }
        (Some(hwaccel), None) if hwaccel == "mediacodec" => {
            // There is no real AVHWFrameContext implementation. Set
            // hwaccel_output_format to avoid av_hwframe_transfer_data error.
            out_hwaccel_output_format = AV_PIX_FMT_MEDIACODEC;
        }
        (_, Some(hwaccel_output_format)) => {
            let hwaccel_output_format_cstr = CString::new(hwaccel_output_format)?;

            let hwaccel_format = unsafe { av_get_pix_fmt(hwaccel_output_format_cstr.as_ptr()) };
            if hwaccel_format == AV_PIX_FMT_NONE {
                error!("Unrecognised hwaccel output format: {:?}", hwaccel_format);
            } else {
                out_hwaccel_output_format = hwaccel_format;
            }
        }
        _ => {}
    }

    let mut out_hwaccel_id = HWAccelID::HwaccelNone;
    let mut out_hwaccel_device_type = AV_HWDEVICE_TYPE_NONE;
    if let Some(mut hwaccel) = hwaccel {
        // The NVDEC hwaccels use a CUDA device, so remap the name here.
        if hwaccel == "nvdec" || hwaccel == "cuvid" {
            hwaccel = "cuda".to_string();
        }
        if hwaccel == "none" {
        } else if hwaccel == "auto" {
            out_hwaccel_id = HWAccelID::HwaccelAuto;
        } else {
            let hwaccel_cstr = CString::new(hwaccel.clone())?;
            let device_type = unsafe { av_hwdevice_find_type_by_name(hwaccel_cstr.as_ptr()) };
            if device_type != AV_HWDEVICE_TYPE_NONE {
                out_hwaccel_id = HWAccelID::HwaccelGeneric;
                out_hwaccel_device_type = device_type;
            }

            if out_hwaccel_id == HWAccelID::HwaccelNone {
                error!("Unrecognized hwaccel: {hwaccel}.");

                let mut hwaccels = Vec::new();
                loop {
                    let device_type =
                        unsafe { av_hwdevice_find_type_by_name(hwaccel_cstr.as_ptr()) };
                    if device_type == AV_HWDEVICE_TYPE_NONE {
                        break;
                    }
                    hwaccels.push(device_type);
                }
                if !hwaccels.is_empty() {
                    error!("Supported hwaccels: {:?}.", hwaccels);
                } else {
                    error!("No hardware acceleration.");
                }
                return Err(OpenInputError::from(AVERROR(EINVAL)).into());
            }
        }
    }

    Ok((
        out_hwaccel_id,
        out_hwaccel_device_type,
        hwaccel_device,
        out_hwaccel_output_format,
    ))
}
