use super::{
    DEFAULT_EYE_SOURCE_HEIGHT, DEFAULT_EYE_SOURCE_WIDTH, INLINE_PREVIEW_MAX_KBIT,
    INLINE_PREVIEW_REQUEST_FPS, INLINE_PREVIEW_REQUEST_HEIGHT, INLINE_PREVIEW_REQUEST_WIDTH,
    LauncherPreview, PREVIEW_HEIGHT, PREVIEW_WIDTH, PreviewSurface, PreviewTelemetry,
    preview_render_size, sanitize_preview_request,
};
use crate::launcher::state::{CaptureSizePreset, LauncherState};
use futures::stream;
use lesavka_common::lesavka::relay_server::{Relay, RelayServer};
use lesavka_common::lesavka::{MonitorRequest, VideoPacket};
use serial_test::serial;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};

#[derive(Clone, Default)]
struct ProbeRelay {
    requests: Arc<Mutex<Vec<MonitorRequest>>>,
}

#[tonic::async_trait]
impl Relay for ProbeRelay {
    type StreamKeyboardStream = Pin<
        Box<
            dyn futures::Stream<Item = Result<lesavka_common::lesavka::KeyboardReport, Status>>
                + Send,
        >,
    >;
    type StreamMouseStream = Pin<
        Box<
            dyn futures::Stream<Item = Result<lesavka_common::lesavka::MouseReport, Status>> + Send,
        >,
    >;
    type CaptureVideoStream =
        Pin<Box<dyn futures::Stream<Item = Result<VideoPacket, Status>> + Send>>;
    type CaptureAudioStream = Pin<
        Box<
            dyn futures::Stream<Item = Result<lesavka_common::lesavka::AudioPacket, Status>> + Send,
        >,
    >;
    type StreamMicrophoneStream =
        Pin<Box<dyn futures::Stream<Item = Result<lesavka_common::lesavka::Empty, Status>> + Send>>;
    type StreamCameraStream =
        Pin<Box<dyn futures::Stream<Item = Result<lesavka_common::lesavka::Empty, Status>> + Send>>;

    async fn stream_keyboard(
        &self,
        _request: Request<tonic::Streaming<lesavka_common::lesavka::KeyboardReport>>,
    ) -> Result<Response<Self::StreamKeyboardStream>, Status> {
        Ok(Response::new(Box::pin(stream::empty())))
    }

    async fn stream_mouse(
        &self,
        _request: Request<tonic::Streaming<lesavka_common::lesavka::MouseReport>>,
    ) -> Result<Response<Self::StreamMouseStream>, Status> {
        Ok(Response::new(Box::pin(stream::empty())))
    }

    async fn capture_video(
        &self,
        request: Request<MonitorRequest>,
    ) -> Result<Response<Self::CaptureVideoStream>, Status> {
        self.requests.lock().unwrap().push(request.into_inner());
        let (_tx, rx) = mpsc::channel(1);
        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
    }

    async fn capture_audio(
        &self,
        _request: Request<MonitorRequest>,
    ) -> Result<Response<Self::CaptureAudioStream>, Status> {
        Ok(Response::new(Box::pin(stream::empty())))
    }

    async fn stream_microphone(
        &self,
        _request: Request<tonic::Streaming<lesavka_common::lesavka::AudioPacket>>,
    ) -> Result<Response<Self::StreamMicrophoneStream>, Status> {
        Ok(Response::new(Box::pin(stream::empty())))
    }

    async fn stream_camera(
        &self,
        _request: Request<tonic::Streaming<VideoPacket>>,
    ) -> Result<Response<Self::StreamCameraStream>, Status> {
        Ok(Response::new(Box::pin(stream::empty())))
    }

    async fn paste_text(
        &self,
        _request: Request<lesavka_common::lesavka::PasteRequest>,
    ) -> Result<Response<lesavka_common::lesavka::PasteReply>, Status> {
        Ok(Response::new(lesavka_common::lesavka::PasteReply {
            ok: true,
            error: String::new(),
        }))
    }

    async fn reset_usb(
        &self,
        _request: Request<lesavka_common::lesavka::Empty>,
    ) -> Result<Response<lesavka_common::lesavka::ResetUsbReply>, Status> {
        Ok(Response::new(lesavka_common::lesavka::ResetUsbReply {
            ok: true,
        }))
    }

    async fn get_capture_power(
        &self,
        _request: Request<lesavka_common::lesavka::Empty>,
    ) -> Result<Response<lesavka_common::lesavka::CapturePowerState>, Status> {
        Ok(Response::new(lesavka_common::lesavka::CapturePowerState {
            available: true,
            enabled: true,
            unit: "relay.service".to_string(),
            detail: "active/running".to_string(),
            active_leases: 1,
            mode: "auto".to_string(),
            detected_devices: 2,
        }))
    }

    async fn set_capture_power(
        &self,
        _request: Request<lesavka_common::lesavka::SetCapturePowerRequest>,
    ) -> Result<Response<lesavka_common::lesavka::CapturePowerState>, Status> {
        self.get_capture_power(Request::new(lesavka_common::lesavka::Empty {}))
            .await
    }
}

#[test]
fn inline_preview_profile_uses_lightweight_defaults() {
    let profile = PreviewSurface::Inline.profile();
    assert_eq!(profile.display_width, PREVIEW_WIDTH);
    assert_eq!(profile.display_height, PREVIEW_HEIGHT);
    assert_eq!(profile.requested_width, INLINE_PREVIEW_REQUEST_WIDTH);
    assert_eq!(profile.requested_height, INLINE_PREVIEW_REQUEST_HEIGHT);
    assert_eq!(profile.requested_fps, INLINE_PREVIEW_REQUEST_FPS);
    assert_eq!(profile.max_bitrate_kbit, INLINE_PREVIEW_MAX_KBIT);
}

#[test]
fn breakout_preview_profile_defaults_to_higher_quality() {
    let profile = PreviewSurface::Window.profile();
    assert_eq!(profile.display_width, 1280);
    assert_eq!(profile.display_height, 720);
    assert_eq!(profile.requested_width, DEFAULT_EYE_SOURCE_WIDTH);
    assert_eq!(profile.requested_height, DEFAULT_EYE_SOURCE_HEIGHT);
    assert_eq!(profile.requested_fps, 60);
    assert_eq!(profile.max_bitrate_kbit, 18_000);
}

#[test]
fn preview_render_size_fits_source_into_display_budget() {
    let profile = PreviewSurface::Inline.profile();
    assert_eq!(preview_render_size(profile, 1920, 1080), (960, 540));
}

#[test]
fn preview_render_size_never_upscales_beyond_source_geometry() {
    let profile = PreviewSurface::Window.profile();
    assert_eq!(preview_render_size(profile, 1280, 720), (1280, 720));
}

#[test]
fn preview_request_sanitizer_keeps_requested_source_geometry() {
    let adapted = sanitize_preview_request(1920, 1080, 60, 18_000);
    assert_eq!(adapted, (1920, 1080, 60, 18_000));
}

#[test]
fn preview_request_sanitizer_clamps_invalid_values() {
    let adapted = sanitize_preview_request(0, 0, 0, 0);
    assert_eq!(adapted, (2, 2, 1, 800));
}

#[test]
fn preview_telemetry_reports_fps_jitter_loss_and_drop_metrics() {
    let mut telemetry = PreviewTelemetry::default();
    let start = Instant::now();
    telemetry.note_decoder("nvh264dec");
    telemetry.record_packet_at(start, 1, 30, 0, 1, 41, 38, 2, "x264enc", 215);
    telemetry.record_presented_frame_at(start + Duration::from_millis(5));
    telemetry.record_packet_at(
        start + Duration::from_millis(33),
        2,
        30,
        0,
        1,
        41,
        38,
        2,
        "x264enc",
        215,
    );
    telemetry.record_presented_frame_at(start + Duration::from_millis(37));
    telemetry.record_packet_at(
        start + Duration::from_millis(80),
        4,
        27,
        2,
        3,
        77,
        88,
        4,
        "x264enc",
        382,
    );
    telemetry.record_presented_frame_at(start + Duration::from_millis(90));

    let snapshot = telemetry.snapshot_at(start + Duration::from_millis(120));
    assert!(snapshot.receive_fps >= 12.0);
    assert!(snapshot.present_fps >= 12.0);
    assert_eq!(snapshot.server_fps, 27.0);
    assert!(snapshot.stream_spread_ms > 0.0);
    assert!(snapshot.packet_loss_pct > 0.0);
    assert_eq!(snapshot.dropped_frames, 2);
    assert_eq!(snapshot.queue_depth, 3);
    assert_eq!(snapshot.queue_depth_peak, 3);
    assert!(snapshot.packet_gap_peak_ms >= 47.0);
    assert!(snapshot.present_gap_peak_ms >= 53.0);
    assert_eq!(snapshot.server_source_gap_peak_ms, 77.0);
    assert_eq!(snapshot.server_send_gap_peak_ms, 88.0);
    assert_eq!(snapshot.server_queue_peak, 4);
    assert_eq!(snapshot.server_process_cpu_pct, 38.2);
    assert_eq!(snapshot.server_encoder_label, "x264enc");
    assert_eq!(snapshot.decoder_label, "nvh264dec");
}

#[test]
#[serial]
fn inline_preview_requests_selected_source_profile_on_wire() {
    let relay = ProbeRelay::default();
    let requests = relay.requests.clone();
    let rt = tokio::runtime::Runtime::new().expect("runtime");
    let addr = rt.block_on(async move {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        tokio::spawn(async move {
            let _ = tonic::transport::Server::builder()
                .add_service(RelayServer::new(relay))
                .serve(addr)
                .await;
        });
        addr
    });

    let preview = LauncherPreview::new(format!("http://{addr}")).expect("preview");
    let state = LauncherState::default();
    let capture = state.capture_size_choice(1);
    preview.set_capture_profile(
        1,
        1,
        capture.width,
        capture.height,
        capture.fps,
        capture.max_bitrate_kbit,
    );
    preview.activate_surface_for_test(1, PreviewSurface::Inline);

    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if let Some(request) = requests.lock().unwrap().last().cloned() {
            assert_eq!(request.id, 1);
            assert_eq!(request.source_id, Some(1));
            assert_eq!(request.requested_width, 1920);
            assert_eq!(request.requested_height, 1080);
            assert_eq!(request.requested_fps, 60);
            assert_eq!(request.max_bitrate, 18_000);
            preview.shutdown_all();
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    preview.shutdown_all();
    panic!("preview did not issue a capture request within timeout");
}

#[test]
#[serial]
fn inline_preview_requests_honest_source_profile_on_wire() {
    let relay = ProbeRelay::default();
    let requests = relay.requests.clone();
    let rt = tokio::runtime::Runtime::new().expect("runtime");
    let addr = rt.block_on(async move {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        tokio::spawn(async move {
            let _ = tonic::transport::Server::builder()
                .add_service(RelayServer::new(relay))
                .serve(addr)
                .await;
        });
        addr
    });

    let preview = LauncherPreview::new(format!("http://{addr}")).expect("preview");
    let mut state = LauncherState::default();
    state.set_capture_size_preset(1, CaptureSizePreset::P1080);
    let capture = state.capture_size_choice(1);
    preview.set_capture_profile(
        1,
        1,
        capture.width,
        capture.height,
        capture.fps,
        capture.max_bitrate_kbit,
    );
    preview.activate_surface_for_test(1, PreviewSurface::Inline);

    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if let Some(request) = requests.lock().unwrap().last().cloned() {
            assert_eq!(request.id, 1);
            assert_eq!(request.source_id, Some(1));
            assert_eq!(request.requested_width, 1920);
            assert_eq!(request.requested_height, 1080);
            assert_eq!(request.requested_fps, 60);
            assert_eq!(request.max_bitrate, 18_000);
            preview.shutdown_all();
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    preview.shutdown_all();
    panic!("preview did not issue a source capture request within timeout");
}

#[test]
#[serial]
fn inline_preview_requests_native_720p_source_mode_on_wire() {
    let relay = ProbeRelay::default();
    let requests = relay.requests.clone();
    let rt = tokio::runtime::Runtime::new().expect("runtime");
    let addr = rt.block_on(async move {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        tokio::spawn(async move {
            let _ = tonic::transport::Server::builder()
                .add_service(RelayServer::new(relay))
                .serve(addr)
                .await;
        });
        addr
    });

    let preview = LauncherPreview::new(format!("http://{addr}")).expect("preview");
    let mut state = LauncherState::default();
    state.set_capture_size_preset(1, CaptureSizePreset::P720);
    let capture = state.capture_size_choice(1);
    preview.set_capture_profile(
        1,
        1,
        capture.width,
        capture.height,
        capture.fps,
        capture.max_bitrate_kbit,
    );
    preview.activate_surface_for_test(1, PreviewSurface::Inline);

    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if let Some(request) = requests.lock().unwrap().last().cloned() {
            assert_eq!(request.id, 1);
            assert_eq!(request.source_id, Some(1));
            assert_eq!(request.requested_width, 1280);
            assert_eq!(request.requested_height, 720);
            assert_eq!(request.requested_fps, 60);
            assert_eq!(request.max_bitrate, 12_000);
            preview.shutdown_all();
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    preview.shutdown_all();
    panic!("preview did not issue a 720p source capture request within timeout");
}

#[test]
#[serial]
fn inline_preview_legacy_low_modes_fall_forward_to_720p_on_wire() {
    let relay = ProbeRelay::default();
    let requests = relay.requests.clone();
    let rt = tokio::runtime::Runtime::new().expect("runtime");
    let addr = rt.block_on(async move {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        tokio::spawn(async move {
            let _ = tonic::transport::Server::builder()
                .add_service(RelayServer::new(relay))
                .serve(addr)
                .await;
        });
        addr
    });

    let preview = LauncherPreview::new(format!("http://{addr}")).expect("preview");
    let mut state = LauncherState::default();
    state.set_capture_size_preset(1, CaptureSizePreset::P480);
    let capture = state.capture_size_choice(1);
    preview.set_capture_profile(
        1,
        1,
        capture.width,
        capture.height,
        capture.fps,
        capture.max_bitrate_kbit,
    );
    preview.activate_surface_for_test(1, PreviewSurface::Inline);

    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if let Some(request) = requests.lock().unwrap().last().cloned() {
            assert_eq!(request.id, 1);
            assert_eq!(request.source_id, Some(1));
            assert_eq!(request.requested_width, 1280);
            assert_eq!(request.requested_height, 720);
            assert_eq!(request.requested_fps, 60);
            assert_eq!(request.max_bitrate, 12_000);
            preview.shutdown_all();
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    preview.shutdown_all();
    panic!("preview did not issue a 720p fallback source capture request within timeout");
}

#[test]
#[serial]
fn preview_can_request_other_eye_as_a_distinct_stream() {
    let relay = ProbeRelay::default();
    let requests = relay.requests.clone();
    let rt = tokio::runtime::Runtime::new().expect("runtime");
    let addr = rt.block_on(async move {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        tokio::spawn(async move {
            let _ = tonic::transport::Server::builder()
                .add_service(RelayServer::new(relay))
                .serve(addr)
                .await;
        });
        addr
    });

    let preview = LauncherPreview::new(format!("http://{addr}")).expect("preview");
    preview.set_capture_profile(0, 1, 1920, 1080, 30, 12_000);
    preview.activate_surface_for_test(0, PreviewSurface::Inline);

    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if let Some(request) = requests.lock().unwrap().last().cloned() {
            assert_eq!(request.id, 0);
            assert_eq!(request.source_id, Some(1));
            assert_eq!(request.requested_width, 1920);
            assert_eq!(request.requested_height, 1080);
            preview.shutdown_all();
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    preview.shutdown_all();
    panic!("preview did not issue a mirrored capture request within timeout");
}