//! # TwilioMediaStreamsServer -- the inbound WSS pump task.

use std::time::Duration;

use axum::{
    Router,
    extract::{
        State,
        ws::{Message, WebSocket, WebSocketUpgrade},
    },
    response::IntoResponse,
    routing::get,
};
use base64::Engine as _;
use rutster_call_model::ChannelId;
use rutster_media::PcmFrame;
use serde::{Deserialize, Serialize};
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, info, warn};

use crate::g711::G711Codec;

const MEDIA_CHANNEL_CAPACITY: usize = 16;

pub struct RegisterTrunkInboundChannel {
    pub call_sid: String,
    pub inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
    pub outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
    pub tap_url: url::Url,
    pub reply: oneshot::Sender<ChannelId>,
}

/// Router state: the media-thread register channel + the app-level
/// keepalive interval (`RUTSTER_WS_PING_SECS`, deploy slice A §5.2).
#[derive(Clone)]
struct MediaStreamState {
    register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
    ping_interval: Duration,
}

pub struct TwilioMediaStreamsServer;

impl TwilioMediaStreamsServer {
    pub fn router(
        register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
        ping_interval: Duration,
    ) -> Router {
        Router::new()
            .route("/twilio/media-stream", get(handle_media_stream))
            .with_state(MediaStreamState {
                register_tx,
                ping_interval,
            })
    }
}

#[derive(Debug, Deserialize)]
#[serde(tag = "event", rename_all = "lowercase")]
pub enum TwilioMediaEvent {
    Connected,
    Start { start: StartPayload },
    Media { media: MediaPayload },
    Stop,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StartPayload {
    pub stream_sid: String,
    pub call_sid: String,
}

#[derive(Debug, Deserialize)]
pub struct MediaPayload {
    pub payload: String,
}

#[derive(Debug, Serialize)]
struct OutboundMediaFrame<'a> {
    event: &'a str,
    #[serde(rename = "streamSid")]
    stream_sid: &'a str,
    media: OutboundMediaPayload,
}

#[derive(Debug, Serialize)]
struct OutboundMediaPayload {
    payload: String,
}

async fn handle_media_stream(
    ws: WebSocketUpgrade,
    State(state): State<MediaStreamState>,
) -> impl IntoResponse {
    ws.on_upgrade(move |socket| run_media_stream(socket, state.register_tx, state.ping_interval))
}

async fn run_media_stream(
    mut socket: WebSocket,
    register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
    ping_interval: Duration,
) {
    match socket.recv().await {
        Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
            Ok(TwilioMediaEvent::Connected) => {
                debug!("twilio media streams: connected handshake received")
            }
            Ok(other) => warn!(?other, "expected 'connected' envelope as first frame"),
            Err(e) => {
                warn!(error = ?e, "first WS frame not a Twilio Media Streams envelope");
                let _ = socket.close().await;
                return;
            }
        },
        _ => {
            warn!("twilio media streams: socket sent no connected handshake; closing");
            let _ = socket.close().await;
            return;
        }
    }

    let (call_sid, stream_sid): (String, String) = match socket.recv().await {
        Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
            Ok(TwilioMediaEvent::Start { start }) => (start.call_sid, start.stream_sid),
            Ok(other) => {
                warn!(?other, "expected 'start' envelope as second frame");
                let _ = socket.close().await;
                return;
            }
            Err(e) => {
                warn!(error = ?e, "second WS frame not a Twilio 'start' envelope");
                let _ = socket.close().await;
                return;
            }
        },
        _ => {
            warn!("twilio media streams: socket sent no start frame; closing");
            let _ = socket.close().await;
            return;
        }
    };

    let (inbound_tx, inbound_rx) = mpsc::channel::<PcmFrame>(MEDIA_CHANNEL_CAPACITY);
    let (outbound_tx, mut outbound_rx) = mpsc::channel::<PcmFrame>(MEDIA_CHANNEL_CAPACITY);

    let (reply_tx, reply_rx) = oneshot::channel::<ChannelId>();
    let register = RegisterTrunkInboundChannel {
        call_sid: call_sid.clone(),
        inbound_from_twilio_rx: inbound_rx,
        outbound_to_twilio_tx: outbound_tx,
        tap_url: url::Url::parse("ws://127.0.0.1:8082/echo")
            .expect("hardcoded loopback URL is always valid"),
        reply: reply_tx,
    };
    if register_tx.send(register).await.is_err() {
        warn!(%call_sid, "media-thread register channel closed; closing WSS");
        let _ = socket.close().await;
        return;
    }
    let channel_id = match reply_rx.await {
        Ok(id) => id,
        Err(_) => {
            warn!(%call_sid, "media-thread dropped RegisterTrunk reply; closing WSS");
            let _ = socket.close().await;
            return;
        }
    };
    info!(%channel_id, %call_sid, %stream_sid, "twilio media streams: registered, audio loop active");

    // App-level keepalive (deploy slice A §5.2): engine-originated pings
    // at RUTSTER_WS_PING_SECS cadence. `interval_at` skips the immediate
    // first tick (a fresh call doesn't need a ping at t=0);
    // MissedTickBehavior::Delay because a late ping must not be followed
    // by a compensating burst. The pre-registration handshake recvs above
    // are un-pinged — Twilio sends connected+start immediately on
    // connect, so that window is sub-second by protocol.
    let mut ping =
        tokio::time::interval_at(tokio::time::Instant::now() + ping_interval, ping_interval);
    ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        tokio::select! {
            biased;

            msg = socket.recv() => match msg {
                Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
                    Ok(TwilioMediaEvent::Media { media }) => {
                        if let Some(frame) = decode_media_payload(&media.payload) {
                            if inbound_tx.try_send(frame).is_err() {
                                warn!(%call_sid, "inbound mpsc full; dropping caller PCM frame");
                            }
                        }
                    }
                    Ok(TwilioMediaEvent::Stop) => {
                        info!(%call_sid, "twilio: stop frame; ending pump");
                        break;
                    }
                    Ok(TwilioMediaEvent::Connected) => {
                        debug!(%call_sid, "twilio sent a duplicate 'connected' frame; ignoring");
                    }
                    Ok(TwilioMediaEvent::Start { .. }) => {
                        debug!(%call_sid, "twilio sent a duplicate 'start' frame; ignoring");
                    }
                    Err(e) => {
                        warn!(error = ?e, %call_sid, "twilio media streams JSON parse failed; skipping frame");
                    }
                },
                Some(Ok(Message::Close(_))) => {
                    info!(%call_sid, "twilio: WS closed by peer");
                    break;
                }
                Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}
                Some(Ok(Message::Binary(_))) => {
                    warn!(%call_sid, "twilio: unexpected binary frame on a text-only protocol; dropping");
                }
                Some(Err(e)) => {
                    warn!(error = ?e, %call_sid, "twilio WS recv error; ending pump");
                    break;
                }
                None => {
                    info!(%call_sid, "twilio: WS stream ended by peer");
                    break;
                }
            },

            frame = outbound_rx.recv() => match frame {
                Some(reply_frame) => {
                    let payload = encode_media_payload(&reply_frame);
                    let envelope = OutboundMediaFrame {
                        event: "media",
                        stream_sid: &stream_sid,
                        media: OutboundMediaPayload { payload },
                    };
                    match serde_json::to_string(&envelope) {
                        Ok(json) => {
                            if socket.send(Message::Text(json)).await.is_err() {
                                warn!(%call_sid, "twilio WS send (outbound) failed; ending pump");
                                break;
                            }
                        }
                        Err(e) => {
                            warn!(error = ?e, %call_sid, "outbound media frame JSON serialization failed; dropping frame");
                        }
                    }
                }
                None => {
                    info!(%call_sid, "outbound_to_twilio_rx closed; ending pump");
                    break;
                }
            },

            _ = ping.tick() => {
                // Vec::new(): an empty application payload — the frame
                // itself is the keepalive; peers must Pong per RFC 6455.
                if socket.send(Message::Ping(Vec::new())).await.is_err() {
                    warn!(%call_sid, "twilio WS ping send failed; ending pump");
                    break;
                }
            },
        }
    }

    let _ = socket.close().await;
    debug!(%call_sid, "twilio media streams: pump task exited");
}

fn decode_media_payload(base64_payload: &str) -> Option<PcmFrame> {
    if base64_payload.is_empty() {
        debug!("twilio media streams: empty media payload; dropping frame");
        return None;
    }
    let mulaw_bytes = base64::engine::general_purpose::STANDARD
        .decode(base64_payload)
        .map_err(|e| warn!(error = ?e, "twilio media streams: base64 decode failed"))
        .ok()?;
    Some(G711Codec::decode_mulaw_to_pcm(&mulaw_bytes))
}

fn encode_media_payload(frame: &PcmFrame) -> String {
    let mulaw_bytes = G711Codec::encode_pcm_to_mulaw(frame);
    base64::engine::general_purpose::STANDARD.encode(&mulaw_bytes)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn media_event_parses_connected_envelope() {
        let s = r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#;
        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
        assert!(matches!(e, TwilioMediaEvent::Connected));
    }

    #[test]
    fn media_event_parses_start_envelope() {
        let s = r#"{
            "event":"start","sequenceNumber":1,
            "start":{"streamSid":"MZdeadbeef","callSid":"CAdeadbeef","accountSid":"ACdeadbeef","tracks":["inbound","outbound"]}
        }"#;
        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
        match e {
            TwilioMediaEvent::Start { start } => {
                assert_eq!(start.stream_sid, "MZdeadbeef");
                assert_eq!(start.call_sid, "CAdeadbeef");
            }
            other => panic!("expected Start, got {other:?}"),
        }
    }

    #[test]
    fn media_event_parses_media_envelope() {
        let s = r#"{"event":"media","sequenceNumber":2,"media":{"track":"outbound","chunk":"1","timestamp":"2023","payload":"//MA"}}"#;
        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
        match e {
            TwilioMediaEvent::Media { media } => assert_eq!(media.payload, "//MA"),
            other => panic!("expected Media, got {other:?}"),
        }
    }

    #[test]
    fn media_event_parses_stop_envelope() {
        let s = r#"{"event":"stop","sequenceNumber":3,"stop":{}}"#;
        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
        assert!(matches!(e, TwilioMediaEvent::Stop));
    }

    #[test]
    fn media_event_forwards_unknown_protocol_fields() {
        let s = r#"{"event":"stop","stop":{},"futureField":"whatever"}"#;
        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
        assert!(matches!(e, TwilioMediaEvent::Stop));
    }

    #[test]
    fn decode_media_payload_round_trips_a_loud_signal() {
        let mut input_frame = PcmFrame::zeroed();
        for s in &mut input_frame.samples {
            *s = 10_000;
        }
        let encoded = encode_media_payload(&input_frame);
        let decoded = decode_media_payload(&encoded).expect("decode OK");
        assert_eq!(decoded.samples.len(), 480);

        let orig_rms = rms(&input_frame.samples);
        let dec_rms = rms(&decoded.samples);
        let drift = (dec_rms - orig_rms).abs() / orig_rms.max(1.0);
        assert!(
            drift <= 0.12,
            "WS-pump decode energy drift = {:.2}% (budget 12%, spec §6.5)",
            drift * 100.0
        );
    }

    #[test]
    fn decode_media_payload_returns_none_on_invalid_base64() {
        assert!(decode_media_payload("!!!not-valid-base64!!!").is_none());
    }

    #[test]
    fn decode_media_payload_returns_none_on_empty_payload() {
        assert!(decode_media_payload("").is_none());
    }

    #[test]
    fn encode_media_payload_round_trips_negative_amplitude_without_polarity_flip() {
        let mut frame = PcmFrame::zeroed();
        for s in &mut frame.samples {
            *s = -5_000;
        }
        let encoded = encode_media_payload(&frame);
        let decoded = decode_media_payload(&encoded).expect("decode OK");
        assert!(
            decoded.samples.iter().all(|&s| s <= 0),
            "negative PCM input must round-trip without polarity flip"
        );
    }

    #[tokio::test]
    async fn router_constructs_with_register_channel() {
        let (register_tx, _register_rx) = mpsc::channel::<RegisterTrunkInboundChannel>(4);
        let _app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20));
    }

    fn rms(samples: &[i16]) -> f64 {
        let sum_sq: u64 = samples.iter().map(|&s| (s as i64 * s as i64) as u64).sum();
        (sum_sq as f64 / samples.len() as f64).sqrt()
    }
}
