use std::{
    net::SocketAddr,
    sync::Arc,
    time::{Duration, Instant},
};

use async_broadcast::broadcast;
use axum::{
    extract::{ConnectInfo, Path, State},
    http::{HeaderMap, StatusCode, header},
    response::IntoResponse,
};
use bytes::Bytes;
use chrono::{Local, Utc};
use dashmap::DashMap;
use entity::{stream_key, stream_session, users};
use sea_orm::{DatabaseConnection, EntityTrait, IntoActiveModel};
use str0m::{
    Candidate, Event, Input, Output, Rtc,
    change::SdpOffer,
    media::{MediaKind, Mid},
    net::{Protocol, Receive},
};
use tokio::{net::UdpSocket, sync::mpsc::Receiver};
use tracing::{debug, error, info, trace, warn};

use crate::{
    StreamSession, audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer,
    http_error::HttpError,
};

/// Kill a WHIP publish if it delivers no media (video or audio) this long.
const NO_MEDIA_TIMEOUT: Duration = Duration::from_secs(30);

pub async fn handle_whip_injest_delete(
    State(state): State<Arc<HttpServer>>,
    ConnectInfo(remote): ConnectInfo<SocketAddr>,
    Path(_slug): Path<String>,
    headers: HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
    let token = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(|s| s.trim().to_string());
    let Some(token) = token else {
        warn!("Whip: missing bearer token from {}", remote);
        return Err(HttpError::Unauthorized);
    };

    let key = stream_key::Entity::find_by_key(&state.db, &token)
        .await?
        .ok_or(HttpError::Unauthorized)?;

    // The detach task may have already cleaned up the session when ICE
    // disconnected (OBS closes the PeerConnection before sending DELETE).
    // Deleting an already-gone session is still a successful delete.
    if let Some(active_session) =
        stream_session::Model::get_active_by_stream_key_id(&state.db, key.id).await?
    {
        active_session
            .into_active_model()
            .finish_stream_session(&state.db, Utc::now())
            .await?;
    }

    state.appstate.lock().await.stream_sessions.remove(&key.id);

    Ok(StatusCode::OK)
}

/// PATCH /api/whip/{id} — Trickle ICE candidate delivery.
pub async fn handle_whip_injest_patch(
    State(state): State<Arc<HttpServer>>,
    Path(slug): Path<String>,
    headers: HeaderMap,
    body: String,
) -> Result<impl IntoResponse, HttpError> {
    // The slug is the stream_key_id (set by the POST handler's Location header).
    let stream_key_id: i32 = slug.parse().map_err(|e| {
        warn!(%slug, "Whip PATCH: bad slug: {:?}", e);
        HttpError::NotFound
    })?;

    info!(stream_key_id, ct = ?headers.get(header::CONTENT_TYPE), "Whip PATCH: trickle candidate");

    // Look up the trickle sender for this session.
    let trickle_map = state.appstate.lock().await.webrtc_proxy.trickle_tx.clone();
    let tx = trickle_map.get(&stream_key_id).ok_or_else(|| {
        warn!(stream_key_id, "Whip PATCH: no trickle channel for session");
        HttpError::NotFound
    })?;

    debug!(stream_key_id, %body, "Whip PATCH: forwarding trickle candidate");
    tx.send(body).ok();

    Ok(StatusCode::NO_CONTENT)
}

pub async fn handle_whip_injest(
    State(state): State<Arc<HttpServer>>,
    ConnectInfo(remote): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    offer: String,
) -> axum::response::Response {
    let err = |status, msg: &str| -> axum::response::Response {
        (
            status,
            [(header::CONTENT_TYPE, "text/plain")],
            msg.to_string(),
        )
            .into_response()
    };
    let public_addr = state.appstate.lock().await.webrtc_proxy.public_addr();

    let token = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(|s| s.trim().to_string());
    let Some(token) = token else {
        warn!("Whip: missing bearer token from {}", remote);
        return err(StatusCode::UNAUTHORIZED, "missing bearer token");
    };

    let key = match stream_key::Entity::find_by_key(&state.db, &token).await {
        Ok(Some(key)) => key,
        Ok(None) => {
            warn!("Whip: stream key not found, rejecting {}", remote);
            return err(StatusCode::UNAUTHORIZED, "invalid stream key");
        }
        Err(e) => {
            error!("Whip: DB error looking up stream key: {:?}", e);
            return err(StatusCode::INTERNAL_SERVER_ERROR, "db error");
        }
    };

    match stream_session::Model::get_active_by_stream_key_id(&state.db, key.id).await {
        Ok(Some(_)) => {
            warn!(stream_key_id = key.id, label = %key.label, "Whip: stream key already live, rejecting duplicate publish from {}", remote);
            return err(StatusCode::CONFLICT, "stream already live");
        }
        Ok(None) => {}
        Err(e) => {
            error!("Whip: DB error checking active stream: {:?}", e);
            return err(StatusCode::INTERNAL_SERVER_ERROR, "db error");
        }
    };
    info!(stream_key_id = key.id, label = %key.label, "Whip authenticated stream key from {}", remote);

    // Parse the SDP offer first so we can detect the codec and
    // configure the Rtc before accepting.
    let sdp_offer = match SdpOffer::from_sdp_string(&offer) {
        Ok(o) => o,
        Err(e) => {
            error!("Whip: cant parse offer from {}: {:?}", remote, e);
            return err(StatusCode::BAD_REQUEST, "invalid SDP offer");
        }
    };
    info!(
        "Whip offer from {} — {} media line(s):",
        remote,
        sdp_offer.media_lines.len()
    );
    for x in &sdp_offer.media_lines {
        info!("  {}", x);
    }

    // Detect codec from the offer so we can enable matching codecs.
    let stream_codec = video_codec_from_sdp_offer(&sdp_offer);
    info!(?stream_codec, "detected codec from WHIP offer");

    // Build Rtc with ICE-Lite (required by WHIP RFC 9728 §4.1) and
    // matching codecs enabled.
    // Not using ICE-Lite: full ICE lets the server initiate checks
    // when OBS hasn't sent its candidates yet (Trickle ICE without PATCH).
    let mut builder = Rtc::builder();
    {
        let cc = builder.codec_config();
        cc.clear();
        cc.enable_opus(true);
        match &stream_codec {
            Some(crate::StreamCodec::H264) => {
                info!("Whip: enabling H.264 codec");
                cc.enable_h264(true);
            }
            Some(crate::StreamCodec::H265) => {
                info!("Whip: enabling H.265 codec");
                cc.enable_h265(true);
            }
            Some(crate::StreamCodec::AV1) => {
                info!("Whip: enabling AV1 codec");
                cc.enable_av1(true);
            }
            None => {
                warn!("Whip: no video codec detected in offer, enabling H.264 as fallback");
                cc.enable_h264(true);
            }
        }
    }
    let mut rtc = builder.build(Instant::now());
    let candidate = Candidate::host(public_addr, Protocol::Udp).unwrap();
    rtc.add_local_candidate(candidate);
    info!(%public_addr, "Whip: added local ICE candidate, accepting offer…");

    let offer_answe = match rtc.sdp_api().accept_offer(sdp_offer) {
        Ok(a) => a,
        Err(e) => {
            error!("cant accept inject offer: {:?}", e);
            return err(StatusCode::BAD_REQUEST, "could not accept offer");
        }
    };

    // OBS sends no a=candidate: lines (disableAutoGathering). Derive a
    // remote host candidate from the HTTP source address so the server
    // has somewhere to send STUN checks.
    if let Ok(c) = Candidate::host(remote, Protocol::Udp) {
        info!(%remote, "Whip: no candidates in offer, adding HTTP-derived remote host candidate");
        rtc.add_remote_candidate(c);
    }

    // Extract negotiated codec, PT, and profile from the answer
    // so WHEP viewers can use the exact same codec config.
    let (negotiated_codec, video_pt, video_profile) = extract_negotiated_codec_info(&offer_answe);
    info!(
        ?negotiated_codec,
        video_pt,
        ?video_profile,
        "negotiated codec from WHIP answer"
    );
    if negotiated_codec.is_none() || video_pt.is_none() {
        warn!("Whip: no common video codec negotiated, rejecting");
        return err(StatusCode::NOT_ACCEPTABLE, "no common video codec");
    }

    let answer_sdp = offer_answe
        .to_sdp_string()
        // Strip a=ice-options:trickle so OBS starts ICE immediately.
        .replace("a=ice-options:trickle\r\n", "")
        .replace("a=ice-options:trickle\n", "");
    // Fix up the answer for libdatachannel (OBS WHIP):
    // 1. Strip a=group:BUNDLE — OBS doesn't negotiate it.
    // 2. Add the host candidate to the video m= line.
    let answer_sdp = answer_sdp
        .replace("a=group:BUNDLE 0 1\r\n", "")
        .replace("a=group:BUNDLE 0 1\n", "");
    let answer_sdp =
        if let Some(cand_line) = answer_sdp.lines().find(|l| l.starts_with("a=candidate:")) {
            // Insert the candidate line after the video m= line.
            let cand_replacement = format!(
                "m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 {}\r\n{}\r\n",
                public_addr.ip(),
                cand_line
            );
            answer_sdp.replace(
                "m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 0.0.0.0\r\n",
                &cand_replacement,
            )
        } else {
            answer_sdp
        };
    info!("Serving Whip SDP answer to {}:\n{}", remote, answer_sdp);

    let ufrag = answer_sdp
        .lines()
        .find(|l| l.starts_with("a=ice-ufrag:"))
        .and_then(|l| l.strip_prefix("a=ice-ufrag:"))
        .map(|s| s.trim().to_string());
    let Some(ufrag) = ufrag else {
        error!("Whip: answer has no a=ice-ufrag");
        return err(StatusCode::INTERNAL_SERVER_ERROR, "no ice-ufrag in answer");
    };
    let ice_pwd = answer_sdp
        .lines()
        .find(|l| l.starts_with("a=ice-pwd:"))
        .and_then(|l| l.strip_prefix("a=ice-pwd:"))
        .map(|s| s.trim().to_string());
    info!(
        ?ufrag,
        ?ice_pwd,
        "Whip: registering ICE credentials with proxy"
    );
    let (socket, rx) = state.appstate.lock().await.webrtc_proxy.add_client(ufrag);

    let stream_sessions = state.appstate.lock().await.stream_sessions.clone();
    let (mut video_tx, video_rx) = broadcast::<Arc<VideoFrame>>(32);
    let (mut audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(32);
    // Never block the ingest loop on slow/missing viewers: overwrite the
    // oldest frame instead (same as the RTMP path; the viewer re-waits for a
    // keyframe on Overflowed). The undrained rx below must not stall sends.
    video_tx.set_overflow(true);
    audio_tx.set_overflow(true);

    let user = users::Entity::find_by_id(key.user_id)
        .one(&state.db)
        .await
        .unwrap()
        .unwrap();
    stream_sessions.insert(
        key.id,
        StreamSession {
            stream_key_id: key.id,
            stream_key_label: key.label,
            stream_key_user: user.username,
            custom_id: key.custom_id,
            is_unlisted: key.is_unlisted,
            password: key.password,
            frame_channel: video_tx,
            audio_channel: audio_tx,
            codec: negotiated_codec,
            started_at: Utc::now(),
            active_clients: 0.into(),
            video_pt,
            video_profile_level_id: video_profile,
        },
    );
    info!(
        stream_key_id = key.id,
        "Whip: StreamSession inserted, spawning detach task"
    );

    // Trickle-ICE channel: OBS can send candidates via PATCH after the
    // initial offer. We forward them to the Rtc task for add_remote_candidate.
    let (trickle_tx, trickle_rx) = tokio::sync::mpsc::unbounded_channel();
    let trickle_map = state.appstate.lock().await.webrtc_proxy.trickle_tx.clone();
    trickle_map.insert(key.id, trickle_tx);

    let db = state.db.clone();
    tokio::spawn(async move {
        detach_inject_rtc(
            stream_sessions,
            key.id,
            db,
            socket,
            rx,
            trickle_rx,
            rtc,
            public_addr,
            video_rx,
            audio_rx,
        )
        .await;
        // Clean up trickle channel when done.
        trickle_map.remove(&key.id);
    });

    stream_session::Model::create_stream_session(&state.db, key.id, Local::now().into())
        .await
        .ok();

    let location = format!("/api/whip/{}", key.id);
    axum::response::Response::builder()
        .status(StatusCode::CREATED)
        .header(header::CONTENT_TYPE, "application/sdp")
        .header(header::LOCATION, &location)
        // // Provide a STUN server via Link header so OBS can gather
        // // ICE candidates even without explicit STUN configuration.
        // .header(
        //     header::LINK,
        //     "<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
        // )
        .body(axum::body::Body::from(answer_sdp))
        .unwrap()
}

async fn detach_inject_rtc(
    sessions_ref: Arc<DashMap<i32, StreamSession>>,
    stream_key_id: i32,
    db: DatabaseConnection,
    socket: Arc<UdpSocket>,
    mut rx: Receiver<(Bytes, SocketAddr)>,
    mut trickle_rx: tokio::sync::mpsc::UnboundedReceiver<String>,
    mut rtc: Rtc,
    local_addr: SocketAddr,
    video_rx: async_broadcast::Receiver<Arc<VideoFrame>>,
    audio_rx: async_broadcast::Receiver<Arc<OpusAudioFrame>>,
) {
    let cleanup = || {
        let sessions_ref = sessions_ref.clone();
        let db = db.clone();
        async move {
            debug!("Cleaning up {:?}", &stream_key_id);
            sessions_ref.remove(&stream_key_id);
            if let Ok(Some(s)) =
                stream_session::Model::get_active_by_stream_key_id(&db, stream_key_id).await
            {
                s.into_active_model()
                    .finish_stream_session(&db, Local::now().into())
                    .await
                    .ok();
            }
        }
    };
    let mut video_mid: Option<Mid> = None;
    let mut audio_mid: Option<Mid> = None;
    let mut video_tx: Option<async_broadcast::Sender<Arc<VideoFrame>>> = None;
    let mut audio_tx: Option<async_broadcast::Sender<Arc<OpusAudioFrame>>> = None;
    let mut _connected = false;
    let mut disconnect_timer: Option<Instant> = None;

    loop {
        // Blankly using them so they don't drop (like RTMP).
        let _ = video_rx.is_closed();
        let _ = audio_rx.is_closed();

        let deadline = loop {
            match rtc.poll_output() {
                Ok(Output::Timeout(t)) => break t,
                Ok(Output::Transmit(t)) => {
                    // The keep alive loop, by default is every 1 second.
                    trace!(
                        "Whip TX: {} bytes → {}:{}",
                        t.contents.len(),
                        t.destination.ip(),
                        t.destination.port()
                    );
                    if let Err(e) = socket.send_to(&t.contents, t.destination).await {
                        warn!("Whip UDP send error: {:?}", e);
                        cleanup().await;
                        return;
                    }
                    // 30 Sec time out if disconnected, then we clean up and disconnect.
                    if let Some(instant_since_disconnet) = disconnect_timer
                        && Instant::now()
                            .duration_since(instant_since_disconnet)
                            .as_secs()
                            > NO_MEDIA_TIMEOUT.as_secs()
                    {
                        info!(
                            "WHIP connection on stream_key_id: {:?}, has been disconnected for {:?} seconds. Cleaning up and destroying the connection,",
                            stream_key_id,
                            NO_MEDIA_TIMEOUT.as_secs()
                        );
                        cleanup().await;
                        rtc.disconnect();
                        return;
                    }
                }
                Ok(Output::Event(e)) => match e {
                    Event::MediaAdded(ma) => {
                        info!(stream_key_id, kind = ?ma.kind, mid = ?ma.mid, "Whip MediaAdded");
                        if ma.kind == MediaKind::Video {
                            video_mid = Some(ma.mid);
                        }
                        if ma.kind == MediaKind::Audio {
                            audio_mid = Some(ma.mid);
                        }
                    }
                    Event::MediaData(md) => {
                        // str0m depacketizes RTP into full codec frames (no decode).
                        // H.264/H.265 → Annex-B, AV1 → OBU, Opus → raw Opus packets.
                        // Same format as our RTMP codec parsers produce — push directly.
                        if Some(md.mid) == video_mid {
                            let ts_ms = (md.time.as_seconds() * 1000.0) as u32;
                            let frame = VideoFrame {
                                data: Bytes::copy_from_slice(&md.data),
                                is_keyframe: md.is_keyframe(),
                                timestamp_ms: ts_ms,
                            };
                            if let Some(tx) = &video_tx {
                                tx.broadcast(Arc::new(frame)).await.ok();
                            }
                        }
                        if Some(md.mid) == audio_mid {
                            let ts_ms = (md.time.as_seconds() * 1000.0) as u32;
                            let frame = OpusAudioFrame {
                                data: Bytes::copy_from_slice(&md.data),
                                timestamp_ms: ts_ms,
                            };
                            if let Some(tx) = &audio_tx {
                                tx.broadcast(Arc::new(frame)).await.ok();
                            }
                        }
                    }
                    Event::IceConnectionStateChange(state) => {
                        info!(stream_key_id, ?state, "Whip ICE state change");
                        match state {
                            str0m::IceConnectionState::Disconnected => {
                                info!(
                                    "Whip ICE disconnected... (State changed to Disconnected for {:?}) ((This is usually due to network jitter))",
                                    &stream_key_id
                                );
                                disconnect_timer = Some(Instant::now());
                            }
                            str0m::IceConnectionState::Connected
                            | str0m::IceConnectionState::Completed => {
                                disconnect_timer = None;
                            }
                            _ => {}
                        }
                    }
                    Event::Connected => {
                        info!(
                            stream_key_id,
                            "Whip DTLS+ICE connected, wiring broadcast channels"
                        );
                        if let Some(session) = sessions_ref.get(&stream_key_id) {
                            video_tx = Some(session.frame_channel.clone());
                            audio_tx = Some(session.audio_channel.clone());
                        }
                        _connected = true;
                    }
                    _ => {}
                },
                Err(e) => {
                    error!("Whip poll_output error (closing): {:?}", e);
                    cleanup().await;
                    return;
                }
            }
        };

        let sleep = tokio::time::sleep_until(deadline.max(Instant::now()).into());
        tokio::select! {
                _ = sleep => {
                    if let Err(e) = rtc.handle_input(Input::Timeout(Instant::now())) {
                        error!(stream_key_id, "Whip handle_input(Timeout) error: {:?}", e);
                        cleanup().await;
                        return;
                    }
                }
                // Trickle-ICE candidates from PATCH /api/whip/{id}
                Some(candidate_line) = trickle_rx.recv() => {
                    info!(stream_key_id, %candidate_line, "Whip: received trickle candidate");
                    // Parse the candidate string (without "a=candidate:" prefix if present).
                    let cand_str = candidate_line
                        .strip_prefix("a=candidate:")
                        .unwrap_or(&candidate_line);
                    match str0m::Candidate::from_sdp_string(cand_str) {
                        Ok(c) => {
                            info!(stream_key_id, "Whip: adding remote candidate");
                            rtc.add_remote_candidate(c);
                        }
                        Err(e) => {
                            warn!(stream_key_id, %cand_str, "Whip: bad trickle candidate: {:?}", e);
                        }
                    }
                }
                    // No UDP input at all for 30s: closing connection.
                    // (ICE keepalives still arrive when only the encoder is stalled,
                    // so this fires on a dead peer, not a paused one.)
                    result = tokio::time::timeout(NO_MEDIA_TIMEOUT, rx.recv()) => {
                    let Ok(Some((data, from))) = result else {
                        if result.is_err() {
                            warn!(stream_key_id, "Whip: no input for {NO_MEDIA_TIMEOUT:?}, closing connection");
                            rtc.disconnect();
                        } else {
                            info!(stream_key_id, "Whip proxy channel closed, cleaning up");
                        }
                        cleanup().await;
                        return;
                    };
                trace!(
                    stream_key_id,
                    len = data.len(),
                    %from,
                    "Whip RX: {} bytes",
                    data.len()
                );
                if let Ok(contents) = (&data[..]).try_into()
                    && let Err(e) = rtc.handle_input(Input::Receive(
                        Instant::now(),
                        Receive {
                            proto: Protocol::Udp,
                            source: from,
                            destination: local_addr,
                            contents,
                        },
                    )) {
                        error!("Whip handle_input(Receive) error: {:?}", e);
                        cleanup().await;
                        return;
                    }
            }
        }
    }
}

/// Extract the negotiated video codec, PT, and profile_level_id from an SDP answer.
///
/// Walks the answer's media lines, finds the first video m-line with
/// negotiated rtp_params, and returns the codec + PT + H.264 profile.
pub fn extract_negotiated_codec_info(
    answer: &str0m::change::SdpAnswer,
) -> (Option<crate::StreamCodec>, Option<u8>, Option<u32>) {
    use str0m::format::Codec;

    for line in answer.media_lines.iter() {
        // Iterate rtp_params on every m-line; video check via
        // codec.is_video() avoids needing str0m's private MediaType.
        for p in line.rtp_params() {
            if p.spec().codec.is_video() {
                let pt = Some(*p.pt());
                let profile = p.spec().format.profile_level_id;
                let codec = match p.spec().codec {
                    Codec::H264 => Some(crate::StreamCodec::H264),
                    Codec::H265 => Some(crate::StreamCodec::H265),
                    Codec::Av1 => Some(crate::StreamCodec::AV1),
                    _ => None,
                };
                return (codec, pt, profile);
            }
        }
    }
    (None, None, None)
}

/// Extract the video codec from an SDP offer's media lines.
///
/// Walks every m= line's `rtp_params()`, maps str0m's `Codec` enum
/// to our `StreamCodec`. Returns the first video codec found.
pub fn video_codec_from_sdp_offer(
    sdp_offer: &str0m::change::SdpOffer,
) -> Option<crate::StreamCodec> {
    use str0m::format::Codec;

    // SdpOffer derefs to Sdp, which has pub media_lines.
    // MediaLine is pub(crate) in str0m, but we can call its pub methods.
    for line in sdp_offer.media_lines.iter() {
        for p in line.rtp_params() {
            if p.spec().codec.is_video() {
                return match p.spec().codec {
                    Codec::H264 => Some(crate::StreamCodec::H264),
                    Codec::H265 => Some(crate::StreamCodec::H265),
                    Codec::Av1 => Some(crate::StreamCodec::AV1),
                    _ => None,
                };
            }
        }
    }
    None
}