//! Display-server-agnostic capture pipeline. The video *source* element is the
//! only part that differs between Wayland (`pipewiresrc`, after a portal
//! handshake) and X11 (`ximagesrc`); everything downstream — the videorate cap,
//! the encoder, `h264parse`, `mpegtsmux`, the whole audio branch, the gst spawn,
//! the [`Serve`] fanout binding, and the [`CaptureHandle`] lifecycle — is shared
//! and lives here. Backends call [`spawn`] with just their source-element args.

use anyhow::{Context, Result};
use nix::sys::signal::Signal;
use std::process::Stdio;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::process::{Child, Command};
use tokio::time::timeout;

use super::audio_plan::CapturePlan;
use super::health;
use super::quality::EffectiveQuality;
use super::serve::Serve;
use crate::cli::HostOpts;
use crate::common::contained;

const GST_TERM_BUDGET: Duration = Duration::from_secs(1);
const GST_KILL_BUDGET: Duration = Duration::from_secs(1);

/// Owns the contained GStreamer process from spawn through confirmed reap.
///
/// Keeping this guard alive during `Serve::bind` closes the old constructor
/// leak: any error after spawn kills the whole process group, not merely the
/// direct child. An unconfirmed Drop poisons the host so the supervisor cannot
/// start another capture on top of a possibly-live portal/PipeWire owner.
struct CaptureProcess {
    child: Child,
    leader_pid: u32,
    reaped: bool,
    health: health::Reporter,
}

impl CaptureProcess {
    fn new(child: Child, health: health::Reporter) -> Result<Self> {
        let leader_pid = child
            .id()
            .context("gst-launch-1.0 exited before its process id was recorded")?;
        Ok(Self {
            child,
            leader_pid,
            reaped: false,
            health,
        })
    }

    fn take_stdout(&mut self) -> Option<tokio::process::ChildStdout> {
        self.child.stdout.take()
    }

    async fn shutdown(&mut self) {
        // Reap an already-dead child before addressing its process group. A
        // zombie still reserves its pid, but after `try_wait` succeeds that pid
        // may be reused; returning here avoids ever signalling a new group that
        // inherited the old numeric id.
        match self.child.try_wait() {
            Ok(Some(_)) => {
                self.reaped = true;
                self.health
                    .poison("gst-launch-1.0 exited before PixelPass began capture shutdown");
                return;
            }
            Ok(None) => {}
            Err(e) => tracing::warn!(
                "capture: could not inspect gst before SIGTERM ({e}); continuing teardown"
            ),
        }

        let _ = contained::signal_group(self.leader_pid, Signal::SIGTERM);
        match timeout(GST_TERM_BUDGET, self.child.wait()).await {
            Ok(Ok(_)) => {
                self.reaped = true;
                return;
            }
            Ok(Err(e)) => tracing::warn!(
                "capture: gst wait after SIGTERM failed ({e}); escalating to SIGKILL"
            ),
            Err(_) => tracing::warn!(
                "capture: gst did not exit within {GST_TERM_BUDGET:?}; escalating to SIGKILL"
            ),
        }

        let _ = contained::signal_group(self.leader_pid, Signal::SIGKILL);
        let _ = self.child.start_kill();
        match timeout(GST_KILL_BUDGET, self.child.wait()).await {
            Ok(Ok(_)) => self.reaped = true,
            Ok(Err(e)) => {
                self.health.poison(format!(
                    "gst-launch-1.0 could not be reaped after SIGKILL: {e}"
                ));
            }
            Err(_) => {
                self.health.poison(format!(
                    "gst-launch-1.0 did not exit within {GST_KILL_BUDGET:?} after SIGKILL"
                ));
            }
        }
    }
}

impl Drop for CaptureProcess {
    fn drop(&mut self) {
        if self.reaped {
            return;
        }
        let _ = contained::signal_group(self.leader_pid, Signal::SIGKILL);
        let _ = self.child.start_kill();
        self.health.poison(
            "gst-launch-1.0 was dropped before a confirmed reap; its process group was killed",
        );
    }
}

pub(super) struct CaptureHandle {
    gst: Option<CaptureProcess>,
    audio: Option<CapturePlan>,
    serve: Option<Serve>,
    stopping: Arc<AtomicBool>,
}

impl CaptureHandle {
    pub fn local_port(&self) -> u16 {
        self.serve
            .as_ref()
            .expect("serve is always Some until shutdown")
            .local_port()
    }

    /// Graceful teardown: SIGTERM the gst process group, give it ~1s to exit,
    /// then SIGKILL and a second bounded reap,
    /// unload audio routing (if any), then tear down the serve layer.
    /// The serve reader will see EOF on gst stdout and exit on its own;
    /// serve.shutdown() is the backstop.
    pub async fn shutdown(mut self) {
        self.stopping.store(true, Ordering::Release);
        if let Some(mut gst) = self.gst.take() {
            gst.shutdown().await;
        }
        if let Some(audio_plan) = self.audio.take() {
            audio_plan.shutdown().await;
        }
        if let Some(serve) = self.serve.take() {
            serve.shutdown().await;
        }
    }
}

impl Drop for CaptureHandle {
    fn drop(&mut self) {
        self.stopping.store(true, Ordering::Release);
        // CaptureProcess kills the whole process group and poisons the host;
        // the typed plan's inner owner and Serve handle their own Drop layers.
    }
}

/// Spawn the shared gst pipeline for a backend that supplies `source_args`
/// (the video-source element + its properties, e.g. `["pipewiresrc", "fd=7",
/// …]` or `["ximagesrc", "use-damage=false", …]`). `source_dims` is the source
/// pixel size when the backend knows it (Wayland from the portal, X11 from
/// root/window geometry); it lets a downscale preset compute an exact even
/// target resolution and skip scaling when the source is already small enough.
/// `after_spawn` runs once, immediately after the gst child is launched —
/// Wayland uses it to `close` the pipewire fd it leaked into the child; X11
/// passes a no-op.
pub(super) async fn spawn(
    opts: &HostOpts,
    quality: &EffectiveQuality,
    source_dims: Option<(u32, u32)>,
    source_args: Vec<String>,
    health: health::Reporter,
    after_spawn: impl FnOnce(),
) -> Result<CaptureHandle> {
    let audio_plan = CapturePlan::start(opts, health.clone()).await?;
    let args = build_args(&source_args, &audio_plan, opts, quality, source_dims);

    let mut gst_cmd = Command::new("gst-launch-1.0");
    gst_cmd
        .args(&args)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::inherit())
        .kill_on_drop(true);
    if std::env::var_os("PIXELPASS_GST_DEBUG").is_some() {
        gst_cmd.env("GST_DEBUG", "3");
    }
    let gst = contained::spawn_tokio(&mut gst_cmd).context("failed to spawn gst-launch-1.0")?;
    let mut gst = CaptureProcess::new(gst, health.clone())?;

    // Backend-specific post-spawn cleanup (Wayland closes its leaked pw fd here,
    // once gst has inherited its own copy).
    after_spawn();

    let gst_stdout = gst
        .take_stdout()
        .context("gst-launch-1.0 stdout pipe unavailable")?;

    // Hand stdout to the serve layer, which binds the localhost HTTP listener
    // and runs the broadcast fanout. No demux/remux, no codec assumptions.
    let stopping = Arc::new(AtomicBool::new(false));
    let serve = Serve::bind(gst_stdout, health.clone(), Arc::clone(&stopping)).await?;

    Ok(CaptureHandle {
        gst: Some(gst),
        audio: Some(audio_plan),
        serve: Some(serve),
        stopping,
    })
}

/// Build the full gst-launch argument vector: MPEG-TS mux + fdsink, then the
/// video branch (caller's `source` → videorate cap → optional downscale →
/// encoder → h264parse → mux.), then the audio branch (pulsesrc → AAC → mux.).
/// Bitrate, framerate, and the downscale height come from the resolved
/// [`EffectiveQuality`]; the encoder and the `videoconvert` target format are
/// selected by `opts.no_hwencode` (hardware VAAPI wants NV12, software x264
/// wants I420).
fn build_args(
    source: &[String],
    audio_plan: &CapturePlan,
    opts: &HostOpts,
    quality: &EffectiveQuality,
    source_dims: Option<(u32, u32)>,
) -> Vec<String> {
    let key_interval = (quality.framerate * 2).to_string();
    let bitrate = quality.bitrate.to_string();
    let framerate_caps = format!("video/x-raw,framerate={}/1", quality.framerate);

    let (raw_format, encoder_args): (&str, Vec<String>) = if opts.no_hwencode {
        (
            "video/x-raw,format=I420",
            vec![
                "x264enc".into(),
                "tune=zerolatency".into(),
                "speed-preset=ultrafast".into(),
                format!("bitrate={bitrate}"),
                format!("key-int-max={key_interval}"),
            ],
        )
    } else {
        (
            "video/x-raw,format=NV12",
            vec![
                "vah264enc".into(),
                "rate-control=cbr".into(),
                format!("bitrate={bitrate}"),
                format!("key-int-max={key_interval}"),
            ],
        )
    };

    // muxer + sink
    let mut args: Vec<String> = vec![
        "mpegtsmux".into(),
        "name=mux".into(),
        "!".into(),
        "queue".into(),
        "!".into(),
    ];
    // Debug A/V-drift tap: when PIXELPASS_TS_DUMP=<path> is set, tee the exact
    // muxed TS both to fd=1 (normal serve path, unchanged) and to a file, so the
    // host-side stream can be ffprobe'd for capture-side audio/video PTS drift.
    // Each tee branch has its own queue so the disk sink can't backpressure the
    // live serve branch. No effect when unset. (Mirrors PIXELPASS_GST_DEBUG.)
    if let Some(dump) = std::env::var_os("PIXELPASS_TS_DUMP") {
        let path = dump.to_string_lossy().into_owned();
        args.extend([
            "tee".into(),
            "name=dbgtee".into(),
            "!".into(),
            "queue".into(),
            "!".into(),
            "fdsink".into(),
            "fd=1".into(),
            "dbgtee.".into(),
            "!".into(),
            "queue".into(),
            "!".into(),
            "filesink".into(),
            format!("location={path}"),
        ]);
    } else {
        args.extend(["fdsink".into(), "fd=1".into()]);
    }

    // Downscale step for the quality presets. `None` = encode at native size
    // (the "Source" preset, or a source already at/below the target height — we
    // never upscale). When the source dimensions are known we pin an exact even
    // WxH preserving the source aspect; H.264 4:2:0 needs even dims, so width is
    // rounded to even and height is forced even (preset heights already are; a
    // raw --max-height override is rounded down). When dims are unknown (a rare
    // X11 geometry-read failure) we fall back to height-only + square pixels +
    // an even-stepped width range and let videoscale negotiate.
    let scale_caps: Option<String> = match quality.max_height {
        None => {
            tracing::info!(preset = %quality.label, "encoding at native resolution (no downscale)");
            None
        }
        Some(max_h) => {
            let h = (max_h & !1).max(2);
            match source_dims {
                Some((sw, sh)) if sh > h => {
                    let w = ((sw as u64 * h as u64 + sh as u64 / 2) / sh as u64) as u32;
                    let w = (w & !1).max(2);
                    tracing::info!(
                        preset = %quality.label,
                        from = %format!("{sw}x{sh}"),
                        to = %format!("{w}x{h}"),
                        "downscaling video"
                    );
                    Some(format!("{raw_format},width={w},height={h}"))
                }
                Some((sw, sh)) => {
                    tracing::info!(
                        preset = %quality.label,
                        source = %format!("{sw}x{sh}"),
                        max_height = h,
                        "source already at/below preset height — encoding native (no upscale)"
                    );
                    None
                }
                None => {
                    tracing::info!(
                        preset = %quality.label,
                        max_height = h,
                        "downscaling to max height (source size unknown — width follows negotiation)"
                    );
                    Some(format!(
                        "{raw_format},height={h},pixel-aspect-ratio=1/1,width=[2,8192,2]"
                    ))
                }
            }
        }
    };

    // video branch — videorate caps to the target fps so we don't ship at the
    // monitor's refresh rate (e.g. 180Hz) and pile up frames in the demuxer
    // queue faster than realtime. videoscale (when scaling) runs *after*
    // videoconvert so it operates on system-memory NV12/I420: scaling
    // pipewiresrc's raw output directly can hit a format/memory (e.g. DMABuf)
    // that software videoscale won't negotiate.
    args.extend(source.iter().cloned());
    args.extend([
        "!".into(),
        "videorate".into(),
        "!".into(),
        framerate_caps,
        "!".into(),
        "queue".into(),
        "!".into(),
        "videoconvert".into(),
        "!".into(),
        raw_format.into(),
        "!".into(),
    ]);
    if let Some(caps) = scale_caps {
        args.extend(["videoscale".into(), "!".into(), caps, "!".into()]);
    }
    args.extend(encoder_args);
    args.extend([
        "!".into(),
        "h264parse".into(),
        "config-interval=-1".into(),
        "!".into(),
        "video/x-h264,stream-format=byte-stream,alignment=au".into(),
        "!".into(),
        "mux.".into(),
    ]);

    // audio branch — capture the default sink's MONITOR (system audio out),
    // not the default source (which is the mic).
    args.extend([
        "pulsesrc".into(),
        audio_plan.gst_device_arg(),
        "do-timestamp=true".into(),
        "!".into(),
        "queue".into(),
        "!".into(),
        "audioconvert".into(),
        "!".into(),
        "audioresample".into(),
        "!".into(),
        "audio/x-raw,rate=48000,channels=2".into(),
        "!".into(),
        "avenc_aac".into(),
        "bitrate=128000".into(),
        "!".into(),
        "aacparse".into(),
        "!".into(),
        "mux.".into(),
    ]);

    args
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cli::{CaptureMode, Quality};
    use std::time::{Duration, Instant};

    fn legacy_opts() -> HostOpts {
        HostOpts {
            window: false,
            app: None,
            strict_audio: false,
            display_server: None,
            quality: Quality::Source,
            bitrate: None,
            framerate: None,
            max_height: None,
            no_hwencode: false,
            max_viewers: None,
            interactive: false,
            capture_mode: CaptureMode::Legacy,
            legacy_null_sink: false,
            relay: None,
        }
    }

    #[test]
    fn legacy_desktop_audio_tail_is_byte_identical() {
        let opts = legacy_opts();
        let quality = super::super::quality::resolve(&opts, 1);
        let plan = CapturePlan::legacy_fixture("alsa_output.fixture.monitor");
        let args = build_args(&["ximagesrc".to_string()], &plan, &opts, &quality, None);
        let audio_start = args
            .iter()
            .position(|arg| arg == "pulsesrc")
            .expect("pipeline has an audio branch");
        assert_eq!(
            &args[audio_start..],
            [
                "pulsesrc",
                "device=alsa_output.fixture.monitor",
                "do-timestamp=true",
                "!",
                "queue",
                "!",
                "audioconvert",
                "!",
                "audioresample",
                "!",
                "audio/x-raw,rate=48000,channels=2",
                "!",
                "avenc_aac",
                "bitrate=128000",
                "!",
                "aacparse",
                "!",
                "mux.",
            ]
        );
    }

    fn process_is_running(pid: u32) -> bool {
        let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else {
            return false;
        };
        stat.rsplit_once(") ")
            .and_then(|(_, rest)| rest.as_bytes().first().copied())
            .is_some_and(|state| state != b'Z' && state != b'X')
    }

    fn sleeping_capture(health: health::Reporter) -> CaptureProcess {
        let mut command = Command::new("sleep");
        command.arg("30").kill_on_drop(true);
        let child = contained::spawn_tokio(&mut command).expect("spawn contained fixture");
        CaptureProcess::new(child, health).expect("capture process guard")
    }

    #[tokio::test]
    async fn graceful_capture_shutdown_confirms_reap_without_poison() {
        let (health, _) = health::channel();
        let mut capture = sleeping_capture(health.clone());
        capture.shutdown().await;
        assert!(health.fault().is_none());
    }

    #[tokio::test]
    async fn an_already_exited_capture_is_not_misreported_as_a_clean_shutdown() {
        let (health, _) = health::channel();
        let mut command = Command::new("true");
        command.kill_on_drop(true);
        let child = contained::spawn_tokio(&mut command).expect("spawn short contained fixture");
        let mut capture = CaptureProcess::new(child, health.clone()).expect("capture guard");

        let deadline = Instant::now() + Duration::from_secs(1);
        while process_is_running(capture.leader_pid) && Instant::now() < deadline {
            tokio::task::yield_now().await;
        }
        assert!(
            !process_is_running(capture.leader_pid),
            "short fixture must exit before shutdown begins"
        );

        capture.shutdown().await;
        assert_eq!(
            health.fault().as_deref(),
            Some("gst-launch-1.0 exited before PixelPass began capture shutdown")
        );
    }

    #[tokio::test]
    async fn dropping_live_capture_kills_its_group_and_poisons() {
        let (health, _) = health::channel();
        let capture = sleeping_capture(health.clone());
        let pid = capture.leader_pid;
        drop(capture);
        assert!(health.fault().is_some());

        let deadline = Instant::now() + Duration::from_secs(2);
        while process_is_running(pid) && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert!(!process_is_running(pid));
    }
}