pinakes/packages/pinakes-server/src/routes/streaming.rs

use axum::{
  extract::{Path, State},
  http::StatusCode,
};
use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
use pinakes_core::{
  model::MediaId,
  transcode::{estimate_bandwidth, parse_resolution},
};
use uuid::Uuid;

use crate::{error::ApiError, state::AppState};

fn build_response(
  content_type: &str,
  body: impl Into<axum::body::Body>,
) -> Result<axum::response::Response, ApiError> {
  axum::response::Response::builder()
    .header("Content-Type", content_type)
    .body(body.into())
    .map_err(|e| {
      ApiError(pinakes_core::error::PinakesError::InvalidOperation(
        format!("failed to build response: {e}"),
      ))
    })
}

fn build_response_with_status(
  status: StatusCode,
  headers: &[(&str, &str)],
  body: impl Into<axum::body::Body>,
) -> Result<axum::response::Response, ApiError> {
  let mut builder = axum::response::Response::builder().status(status);
  for (name, value) in headers {
    builder = builder.header(*name, *value);
  }
  builder.body(body.into()).map_err(|e| {
    ApiError(pinakes_core::error::PinakesError::InvalidOperation(
      format!("failed to build response: {e}"),
    ))
  })
}

fn escape_xml(s: &str) -> String {
  s.replace('&', "&amp;")
    .replace('<', "&lt;")
    .replace('>', "&gt;")
    .replace('"', "&quot;")
    .replace('\'', "&apos;")
}

#[utoipa::path(
    get,
    path = "/api/v1/media/{id}/stream/hls/master.m3u8",
    tag = "streaming",
    params(("id" = Uuid, Path, description = "Media item ID")),
    responses(
        (status = 200, description = "HLS master playlist"),
        (status = 401, description = "Unauthorized"),
        (status = 404, description = "Not found"),
    ),
    security(("bearer_auth" = []))
)]
pub async fn hls_master_playlist(
  State(state): State<AppState>,
  Path(id): Path<Uuid>,
) -> Result<axum::response::Response, ApiError> {
  // Verify media exists
  let _item = state.storage.get_media(MediaId(id)).await?;

  let config = state.config.read().await;
  let profiles = &config.transcoding.profiles;

  let mut playlist = String::from("#EXTM3U\n#EXT-X-VERSION:3\n\n");

  for profile in profiles {
    let (w, h) = parse_resolution(&profile.max_resolution);
    let bandwidth = estimate_bandwidth(profile);
    let encoded_name =
      utf8_percent_encode(&profile.name, NON_ALPHANUMERIC).to_string();
    playlist.push_str(&format!(
      "#EXT-X-STREAM-INF:BANDWIDTH={bandwidth},RESOLUTION={w}x{h}\n/api/v1/\
       media/{id}/stream/hls/{encoded_name}/playlist.m3u8\n\n",
    ));
  }

  build_response("application/vnd.apple.mpegurl", playlist)
}

#[utoipa::path(
    get,
    path = "/api/v1/media/{id}/stream/hls/{profile}/playlist.m3u8",
    tag = "streaming",
    params(
        ("id" = Uuid, Path, description = "Media item ID"),
        ("profile" = String, Path, description = "Transcode profile name"),
    ),
    responses(
        (status = 200, description = "HLS variant playlist"),
        (status = 400, description = "Bad request"),
        (status = 401, description = "Unauthorized"),
        (status = 404, description = "Not found"),
    ),
    security(("bearer_auth" = []))
)]
pub async fn hls_variant_playlist(
  State(state): State<AppState>,
  Path((id, profile)): Path<(Uuid, String)>,
) -> Result<axum::response::Response, ApiError> {
  let item = state.storage.get_media(MediaId(id)).await?;
  let duration = item.duration_secs.unwrap_or(0.0);
  if duration <= 0.0 {
    return Err(ApiError(
      pinakes_core::error::PinakesError::InvalidOperation(
        "cannot generate HLS playlist for media with unknown or zero duration"
          .into(),
      ),
    ));
  }
  let segment_duration = 10.0;
  let num_segments = (duration / segment_duration).ceil() as usize;

  let mut playlist = String::from(
    "#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-TARGETDURATION:10\n#\
     EXT-X-MEDIA-SEQUENCE:0\n",
  );
  for i in 0..num_segments.max(1) {
    let seg_dur = if i == num_segments.saturating_sub(1) && duration > 0.0 {
      (i as f64).mul_add(-segment_duration, duration)
    } else {
      segment_duration
    };
    playlist.push_str(&format!("#EXTINF:{seg_dur:.3},\n"));
    playlist.push_str(&format!(
      "/api/v1/media/{id}/stream/hls/{profile}/segment{i}.ts\n"
    ));
  }
  playlist.push_str("#EXT-X-ENDLIST\n");

  build_response("application/vnd.apple.mpegurl", playlist)
}

#[utoipa::path(
    get,
    path = "/api/v1/media/{id}/stream/hls/{profile}/{segment}",
    tag = "streaming",
    params(
        ("id" = Uuid, Path, description = "Media item ID"),
        ("profile" = String, Path, description = "Transcode profile name"),
        ("segment" = String, Path, description = "Segment filename"),
    ),
    responses(
        (status = 200, description = "HLS segment data"),
        (status = 202, description = "Segment not yet available"),
        (status = 400, description = "Bad request"),
        (status = 401, description = "Unauthorized"),
    ),
    security(("bearer_auth" = []))
)]
pub async fn hls_segment(
  State(state): State<AppState>,
  Path((id, profile, segment)): Path<(Uuid, String, String)>,
) -> Result<axum::response::Response, ApiError> {
  // Strict validation: reject path traversal, null bytes, leading dots
  if segment.is_empty()
    || segment.starts_with('.')
    || segment.contains('\0')
    || segment.contains("..")
    || segment.contains('/')
    || segment.contains('\\')
  {
    return Err(ApiError(
      pinakes_core::error::PinakesError::InvalidOperation(
        "invalid segment name".into(),
      ),
    ));
  }

  let media_id = MediaId(id);

  // Look for an active/completed transcode session
  if let Some(transcode_service) = &state.transcode_service
    && let Some(session) =
      transcode_service.find_session(media_id, &profile).await
  {
    let segment_path = session.cache_path.join(&segment);

    if segment_path.exists() {
      let data = tokio::fs::read(&segment_path).await.map_err(|e| {
        ApiError(pinakes_core::error::PinakesError::InvalidOperation(
          format!("failed to read segment: {e}"),
        ))
      })?;

      return build_response("video/MP2T", data);
    }

    // Session exists but segment not ready yet
    return build_response_with_status(
      StatusCode::ACCEPTED,
      &[("Retry-After", "2")],
      "segment not yet available",
    );
  }

  Err(ApiError(
    pinakes_core::error::PinakesError::InvalidOperation(
      "no transcode session found; start a transcode first via POST \
       /media/{id}/transcode"
        .into(),
    ),
  ))
}

#[utoipa::path(
    get,
    path = "/api/v1/media/{id}/stream/dash/manifest.mpd",
    tag = "streaming",
    params(("id" = Uuid, Path, description = "Media item ID")),
    responses(
        (status = 200, description = "DASH manifest"),
        (status = 400, description = "Bad request"),
        (status = 401, description = "Unauthorized"),
        (status = 404, description = "Not found"),
    ),
    security(("bearer_auth" = []))
)]
pub async fn dash_manifest(
  State(state): State<AppState>,
  Path(id): Path<Uuid>,
) -> Result<axum::response::Response, ApiError> {
  let item = state.storage.get_media(MediaId(id)).await?;
  let duration = item.duration_secs.unwrap_or(0.0);
  if duration <= 0.0 {
    return Err(ApiError(
      pinakes_core::error::PinakesError::InvalidOperation(
        "cannot generate DASH manifest for media with unknown or zero duration"
          .into(),
      ),
    ));
  }
  let hours = (duration / 3600.0) as u32;
  let minutes = ((duration % 3600.0) / 60.0) as u32;
  let seconds = duration % 60.0;

  let config = state.config.read().await;
  let profiles = &config.transcoding.profiles;

  let mut representations = String::new();
  for profile in profiles {
    let (w, h) = parse_resolution(&profile.max_resolution);
    let bandwidth = estimate_bandwidth(profile);
    let xml_name = escape_xml(&profile.name);
    let url_name =
      utf8_percent_encode(&profile.name, NON_ALPHANUMERIC).to_string();
    representations.push_str(&format!(
            r#"      <Representation id="{xml_name}" bandwidth="{bandwidth}" width="{w}" height="{h}">
        <SegmentTemplate media="/api/v1/media/{id}/stream/dash/{url_name}/segment$Number$.m4s" initialization="/api/v1/media/{id}/stream/dash/{url_name}/init.mp4" duration="10000" timescale="1000" startNumber="0"/>
      </Representation>
"#,
        ));
  }

  let mpd = format!(
    r#"<?xml version="1.0" encoding="UTF-8"?>
<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" mediaPresentationDuration="PT{hours}H{minutes}M{seconds:.1}S" minBufferTime="PT1.5S">
  <Period>
    <AdaptationSet mimeType="video/mp4" segmentAlignment="true">
{representations}    </AdaptationSet>
  </Period>
</MPD>"#
  );

  build_response("application/dash+xml", mpd)
}

#[utoipa::path(
    get,
    path = "/api/v1/media/{id}/stream/dash/{profile}/{segment}",
    tag = "streaming",
    params(
        ("id" = Uuid, Path, description = "Media item ID"),
        ("profile" = String, Path, description = "Transcode profile name"),
        ("segment" = String, Path, description = "Segment filename"),
    ),
    responses(
        (status = 200, description = "DASH segment data"),
        (status = 202, description = "Segment not yet available"),
        (status = 400, description = "Bad request"),
        (status = 401, description = "Unauthorized"),
    ),
    security(("bearer_auth" = []))
)]
pub async fn dash_segment(
  State(state): State<AppState>,
  Path((id, profile, segment)): Path<(Uuid, String, String)>,
) -> Result<axum::response::Response, ApiError> {
  // Strict validation: reject path traversal, null bytes, leading dots
  if segment.is_empty()
    || segment.starts_with('.')
    || segment.contains('\0')
    || segment.contains("..")
    || segment.contains('/')
    || segment.contains('\\')
  {
    return Err(ApiError(
      pinakes_core::error::PinakesError::InvalidOperation(
        "invalid segment name".into(),
      ),
    ));
  }

  let media_id = MediaId(id);

  if let Some(transcode_service) = &state.transcode_service
    && let Some(session) =
      transcode_service.find_session(media_id, &profile).await
  {
    let segment_path = session.cache_path.join(&segment);

    if segment_path.exists() {
      let data = tokio::fs::read(&segment_path).await.map_err(|e| {
        ApiError(pinakes_core::error::PinakesError::InvalidOperation(
          format!("failed to read segment: {e}"),
        ))
      })?;

      return build_response("video/mp4", data);
    }

    return build_response_with_status(
      StatusCode::ACCEPTED,
      &[("Retry-After", "2")],
      "segment not yet available",
    );
  }

  Err(ApiError(
    pinakes_core::error::PinakesError::InvalidOperation(
      "no transcode session found; start a transcode first via POST \
       /media/{id}/transcode"
        .into(),
    ),
  ))
}