//! An HLS [`SourceStream`]: fetches an `.m3u8` media playlist and streams its
//! mp3 segments in order as one continuous byte stream.
//!
//! SoundCloud (and other HLS sources) serve audio as a playlist of short mp3
//! segments rather than one file. mp3 frames byte-concatenate into a valid
//! stream (the fact `ffmpeg -c copy` relies on), so streaming the segments in
//! order yields bytes rodio's symphonia mp3 decoder handles unchanged
//! (architecture/soundcloud-provider.md D4).
//!
//! Sibling to [`crate::windowed_http::WindowedHttpStream`]. This is **forward
//! only**: it reports the source as non-seekable and length-less, so symphonia
//! does not attempt an end-seek that would need a known total length (the
//! rodio-0.22 panic the opus work documented).
//!
//! Never logs the playlist or segment URLs — they are signed and ephemeral.

use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};

use bytes::Bytes;
use futures::{Future, Stream};
use stream_download::source::{DecodeError, SourceStream};
use tracing::{debug, trace, warn};
use url::Url;

/// Parameters for [`HlsStream::create`].
#[derive(Clone, Debug)]
pub struct HlsParams {
    /// The `.m3u8` media-playlist URL.
    pub url: Url,
    pub client: reqwest::Client,
}

impl HlsParams {
    pub fn new(url: Url, client: reqwest::Client) -> Self {
        Self { url, client }
    }
}

/// Error creating the stream (playlist fetch/parse failed).
#[derive(Debug, thiserror::Error)]
#[error("{0}")]
pub struct HlsError(String);

impl DecodeError for HlsError {}

type BytesStream = Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send + Sync>>;
type SegmentFuture = Pin<Box<dyn Future<Output = io::Result<BytesStream>> + Send + Sync>>;

/// A parsed playlist: either the media (segment) playlist we want, or a master
/// playlist pointing at variant playlists (we follow the first).
enum Playlist {
    Media(Vec<Url>),
    Master(Url),
}

enum State {
    /// Ready to fetch `segments[cursor]`.
    Idle,
    /// Waiting for a segment response.
    Requesting(SegmentFuture),
    /// Draining a segment body.
    Streaming(BytesStream),
    Finished,
}

/// See the module docs. Streams `segments[cursor]` bytes, advancing at segment
/// boundaries; finishes after the last.
pub struct HlsStream {
    client: reqwest::Client,
    /// Ordered mp3 segment URLs parsed from the media playlist.
    segments: Vec<Url>,
    /// Index of the next segment to fetch.
    cursor: usize,
    state: State,
}

impl std::fmt::Debug for HlsStream {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HlsStream")
            .field("segments", &self.segments.len())
            .field("cursor", &self.cursor)
            .finish_non_exhaustive()
    }
}

impl HlsStream {
    /// Schedules the fetch of the segment at `cursor`, or finishes when none
    /// remain.
    fn schedule_next(&mut self) {
        let Some(url) = self.segments.get(self.cursor).cloned() else {
            self.state = State::Finished;
            return;
        };
        self.cursor += 1;
        self.state = State::Requesting(Box::pin(fetch_segment(self.client.clone(), url)));
    }
}

/// GETs one segment, returning its body byte-stream. A non-success status is an
/// error carrying the status only (never the signed URL).
async fn fetch_segment(client: reqwest::Client, url: Url) -> io::Result<BytesStream> {
    trace!("fetching hls segment");
    let resp = client.get(url).send().await.map_err(|err| {
        io::Error::other(format!("segment request failed: {}", err.without_url()))
    })?;
    if !resp.status().is_success() {
        return Err(io::Error::other(format!(
            "segment request rejected: {}",
            resp.status()
        )));
    }
    Ok(Box::pin(resp.bytes_stream()))
}

impl Stream for HlsStream {
    type Item = io::Result<Bytes>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = &mut *self;
        loop {
            match &mut this.state {
                State::Finished => return Poll::Ready(None),
                State::Idle => this.schedule_next(),
                State::Requesting(future) => match future.as_mut().poll(cx) {
                    Poll::Pending => return Poll::Pending,
                    Poll::Ready(Ok(stream)) => this.state = State::Streaming(stream),
                    Poll::Ready(Err(err)) => {
                        this.state = State::Finished;
                        return Poll::Ready(Some(Err(err)));
                    }
                },
                State::Streaming(stream) => match stream.as_mut().poll_next(cx) {
                    Poll::Pending => return Poll::Pending,
                    Poll::Ready(Some(Ok(bytes))) => return Poll::Ready(Some(Ok(bytes))),
                    Poll::Ready(Some(Err(err))) => {
                        return Poll::Ready(Some(Err(io::Error::other(format!(
                            "segment body failed: {}",
                            err.without_url()
                        )))));
                    }
                    // Segment drained: move to the next one.
                    Poll::Ready(None) => this.schedule_next(),
                },
            }
        }
    }
}

impl SourceStream for HlsStream {
    type Params = HlsParams;
    type StreamCreationError = HlsError;

    async fn create(params: Self::Params) -> Result<Self, Self::StreamCreationError> {
        let segments = load_segments(&params.client, params.url).await?;
        debug!(segments = segments.len(), "hls stream open");
        if segments.is_empty() {
            return Err(HlsError("hls playlist has no segments".to_string()));
        }
        Ok(Self {
            client: params.client,
            segments,
            cursor: 0,
            state: State::Idle,
        })
    }

    /// Unknown up front (segment sizes are not in the playlist), so `None` —
    /// which, with `supports_seek() == false`, keeps symphonia off the
    /// end-seek path.
    fn content_length(&self) -> Option<u64> {
        None
    }

    /// Forward-only. In-track seeking (open at a segment offset) is a later,
    /// additive change (architecture/soundcloud-provider.md D7).
    async fn seek_range(&mut self, _start: u64, _end: Option<u64>) -> io::Result<()> {
        Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "HLS stream is forward-only",
        ))
    }

    /// Best-effort resume after a dropped connection: re-open the current
    /// segment. `stream-download`'s temp storage already holds everything up to
    /// `current_position`, and its `Read` side serves that; this only needs to
    /// resume producing fresh bytes, so re-fetching the in-flight segment is
    /// acceptable (a short overlap at worst).
    async fn reconnect(&mut self, _current_position: u64) -> io::Result<()> {
        let restart = self.cursor.saturating_sub(1);
        warn!(
            segment = restart,
            "reconnecting hls stream at current segment"
        );
        self.cursor = restart;
        self.state = State::Idle;
        Ok(())
    }

    fn supports_seek(&self) -> bool {
        false
    }
}

/// Fetches the playlist at `url`, following one level of master playlist, and
/// returns the ordered media-segment URLs.
async fn load_segments(client: &reqwest::Client, url: Url) -> Result<Vec<Url>, HlsError> {
    let body = fetch_text(client, url.clone()).await?;
    match parse_playlist(&body, &url)? {
        Playlist::Media(segments) => Ok(segments),
        Playlist::Master(variant) => {
            let body = fetch_text(client, variant.clone()).await?;
            match parse_playlist(&body, &variant)? {
                Playlist::Media(segments) => Ok(segments),
                // A master pointing at another master is not something
                // SoundCloud produces; refuse rather than recurse.
                Playlist::Master(_) => Err(HlsError("nested master playlist".to_string())),
            }
        }
    }
}

async fn fetch_text(client: &reqwest::Client, url: Url) -> Result<String, HlsError> {
    let resp = client
        .get(url)
        .send()
        .await
        .map_err(|e| HlsError(format!("playlist request failed: {}", e.without_url())))?;
    if !resp.status().is_success() {
        return Err(HlsError(format!("playlist rejected: {}", resp.status())));
    }
    resp.text()
        .await
        .map_err(|e| HlsError(format!("playlist read failed: {e}")))
}

/// Parses an m3u8 body. A master playlist (`#EXT-X-STREAM-INF`) yields the
/// first variant URI; otherwise every non-comment line is a media segment.
/// Relative URIs resolve against `base`.
fn parse_playlist(body: &str, base: &Url) -> Result<Playlist, HlsError> {
    let is_master = body.contains("#EXT-X-STREAM-INF");
    let mut uris = Vec::new();
    for line in body.lines() {
        let line = line.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        let resolved = base
            .join(line)
            .map_err(|e| HlsError(format!("bad segment URI: {e}")))?;
        uris.push(resolved);
        if is_master {
            // Only the first variant is needed.
            break;
        }
    }
    if is_master {
        uris.into_iter()
            .next()
            .map(Playlist::Master)
            .ok_or_else(|| HlsError("master playlist has no variant".to_string()))
    } else {
        Ok(Playlist::Media(uris))
    }
}

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

    fn base() -> Url {
        Url::parse("https://cf-hls.sndcdn.com/media/0/playlist.m3u8?token=x").unwrap()
    }

    #[test]
    fn parses_media_playlist_absolute_and_relative() {
        let body = "#EXTM3U\n\
                    #EXT-X-VERSION:6\n\
                    #EXTINF:10.0,\n\
                    https://cdn.sndcdn.com/media/0/1.ts\n\
                    #EXTINF:9.9,\n\
                    2.ts\n\
                    #EXT-X-ENDLIST\n";
        let Playlist::Media(segs) = parse_playlist(body, &base()).unwrap() else {
            panic!("expected media playlist");
        };
        assert_eq!(segs.len(), 2);
        assert_eq!(segs[0].as_str(), "https://cdn.sndcdn.com/media/0/1.ts");
        // Relative URI resolved against the playlist URL.
        assert_eq!(segs[1].as_str(), "https://cf-hls.sndcdn.com/media/0/2.ts");
    }

    #[test]
    fn follows_first_master_variant() {
        let body = "#EXTM3U\n\
                    #EXT-X-STREAM-INF:BANDWIDTH=128000\n\
                    variant-128.m3u8\n\
                    #EXT-X-STREAM-INF:BANDWIDTH=64000\n\
                    variant-64.m3u8\n";
        let Playlist::Master(url) = parse_playlist(body, &base()).unwrap() else {
            panic!("expected master playlist");
        };
        assert_eq!(
            url.as_str(),
            "https://cf-hls.sndcdn.com/media/0/variant-128.m3u8"
        );
    }

    #[test]
    fn ignores_comments_and_blank_lines() {
        let body = "#EXTM3U\n\n#EXTINF:1,\nonly.ts\n\n";
        let Playlist::Media(segs) = parse_playlist(body, &base()).unwrap() else {
            panic!("expected media playlist");
        };
        assert_eq!(segs.len(), 1);
    }
}