Something went wrong. Try again.
experiments of a tiny cytube-like player with yt-dlp
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430//! HTTP server: Axum router, handlers, and API routes.//!//! Provides the HTTP API for creating rooms, ingesting URLs, serving media//! files, and WebSocket connections for room state + chat.
use std::collections::HashMap;use std::net::SocketAddr;use std::path::PathBuf;use std::sync::Arc;
use axum::{ extract::{ConnectInfo, Path, Query, State, WebSocketUpgrade}, http::{header, StatusCode}, response::{IntoResponse, Json, Response}, routing::{get, post}, Router,};use tokio::sync::Mutex;
use crate::media;use crate::room;use crate::types::PlaylistEntryId;
/// Per-IP rate limiter for POST /api/ingest (max 1 req/5s per IP).#[derive(Clone)]pub(crate) struct RateLimiter { inner: Arc<Mutex<HashMap<SocketAddr, tokio::time::Instant>>>,}
impl RateLimiter { pub(crate) fn new() -> Self { Self { inner: Arc::new(Mutex::new(HashMap::new())), } }
/// Returns `Ok(())` if allowed, or `Err(duration_to_wait)` if rate limited. pub(crate) async fn check(&self, addr: SocketAddr) -> Result<(), std::time::Duration> { let mut map = self.inner.lock().await; if let Some(last) = map.get(&addr) { let elapsed = last.elapsed(); if elapsed < std::time::Duration::from_secs(5) { return Err(std::time::Duration::from_secs(5) - elapsed); } } map.insert(addr, tokio::time::Instant::now()); Ok(()) }
/// Remove entries older than 30 seconds to prevent memory leak. pub(crate) async fn cleanup(&self) { let mut map = self.inner.lock().await; map.retain(|_, last| last.elapsed() < std::time::Duration::from_secs(30)); }}
#[derive(Clone)]pub(crate) struct AppState { pub rooms: room::Registry, pub cache_dir: PathBuf, pub rate_limiter: Option<RateLimiter>,}
pub(crate) fn router(rooms: room::Registry, cache_dir: PathBuf) -> Router { let rate_limiter = RateLimiter::new();
let cleanup_limiter = rate_limiter.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(30)); loop { interval.tick().await; cleanup_limiter.cleanup().await; } });
let state = AppState { rooms, cache_dir, rate_limiter: Some(rate_limiter), };
Router::new() .route("/api/v1/rooms", post(create_room)) .route("/api/v1/ingest", post(ingest_url)) .route("/api/v1/skip", post(skip_handler)) .route("/api/v1/playlist/add", post(add_playlist_entry)) .route("/api/v1/playlist/remove", post(remove_playlist_entry)) .route("/api/v1/rooms/{room_id}/media/{hash}", get(media_handler)) .route("/api/v1/ws/{room_id}", get(ws_handler)) // SPA: serve built frontend from static/dist/ .nest_service( "/assets", tower_http::services::ServeDir::new("static/dist/assets"), ) .route("/", get(spa_index)) .route("/r/{name}", get(spa_room)) .route("/{*path}", get(spa_handler)) .with_state(state)}
async fn spa_index() -> Response { serve_dist_html().await}
async fn spa_room() -> Response { serve_dist_html().await}
/// Read the built index.html from disk.async fn serve_dist_html() -> Response { let path = std::path::Path::new("static/dist/index.html"); match tokio::fs::read_to_string(path).await { Ok(html) => ([(header::CONTENT_TYPE, "text/html; charset=utf-8")], html).into_response(), Err(_) => ( StatusCode::OK, [(header::CONTENT_TYPE, "text/html; charset=utf-8")], include_str!("../static/index.html"), ) .into_response(), }}
async fn create_room(State(state): State<AppState>) -> Result<impl IntoResponse, StatusCode> { let (room_id, room_name) = state.rooms.create().await; Ok(Json( serde_json::json!({ "room_id": room_id.to_string(), "room_name": room_name }), ))}
/// Find a media file in the cache directory by hash (derived from URL).async fn find_media_file(dir: &std::path::Path, hash: &str) -> Option<std::path::PathBuf> { // Try exact hash path, with mp4 extension. let path = dir.join(hash).with_extension("mp4"); if path.exists() { return Some(path); } // Fall back to bare hash (no extension), for any legacy files. let bare = dir.join(hash); if bare.exists() { return Some(bare); } None}
#[derive(serde::Deserialize)]struct IngestRequest { url: String, /// UUID or room name. The API accepts either; the backend /// resolves to the internal UUID. room_id: String,}
fn err_response(status: StatusCode, msg: &str) -> axum::response::Response { (status, Json(serde_json::json!({ "error": msg }))).into_response()}
async fn ingest_url( State(state): State<AppState>, ConnectInfo(addr): ConnectInfo<SocketAddr>, axum::Json(req): axum::Json<IngestRequest>,) -> Result<Json<serde_json::Value>, axum::response::Response> { if req.url.len() > 2048 { // 2048 is the practical URL max across browsers return Err(err_response(StatusCode::BAD_REQUEST, "URL too long")); }
if let Some(limiter) = &state.rate_limiter { if let Err(retry_after) = limiter.check(addr).await { let mut res = err_response(StatusCode::TOO_MANY_REQUESTS, "rate limited"); res.headers_mut().insert( axum::http::header::RETRY_AFTER, retry_after .as_secs() .to_string() .parse() .expect("valid retry-after seconds"), ); return Err(res); } }
let room_id = state .rooms .resolve_ref(&req.room_id) .await .ok_or_else(|| err_response(StatusCode::NOT_FOUND, "room not found"))?;
state.rooms.push_url(&room_id, req.url).await;
Ok(Json(serde_json::json!({ "ok": true })))}
#[derive(serde::Deserialize)]struct SkipRequest { /// UUID or room name. room_id: String,}
async fn skip_handler( State(state): State<AppState>, axum::Json(req): axum::Json<SkipRequest>,) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { let room_id = state.rooms.resolve_ref(&req.room_id).await.ok_or_else(|| { ( StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "room not found" })), ) })?;
state.rooms.skip(&room_id).await;
Ok(Json(serde_json::json!({ "ok": true })))}
#[derive(serde::Deserialize)]struct AddPlaylistReq { /// UUID or room name. room_id: String, url: String,}
async fn add_playlist_entry( State(state): State<AppState>, axum::Json(req): axum::Json<AddPlaylistReq>,) -> Result<Json<serde_json::Value>, axum::response::Response> { if req.url.len() > 2048 { return Err(err_response(StatusCode::BAD_REQUEST, "URL too long")); } let room_id = state .rooms .resolve_ref(&req.room_id) .await .ok_or_else(|| err_response(StatusCode::NOT_FOUND, "room not found"))?; state.rooms.add_playlist_entry(&room_id, req.url).await; Ok(Json(serde_json::json!({ "ok": true })))}
#[derive(serde::Deserialize)]struct RemovePlaylistReq { /// UUID or room name. room_id: String, id: String,}
async fn remove_playlist_entry( State(state): State<AppState>, axum::Json(req): axum::Json<RemovePlaylistReq>,) -> Result<Json<serde_json::Value>, axum::response::Response> { let room_id = state .rooms .resolve_ref(&req.room_id) .await .ok_or_else(|| err_response(StatusCode::NOT_FOUND, "room not found"))?; let entry_id = PlaylistEntryId::parse(&req.id) .ok_or_else(|| err_response(StatusCode::BAD_REQUEST, "invalid entry ID"))?; state.rooms.remove_playlist_entry(&room_id, entry_id).await; Ok(Json(serde_json::json!({ "ok": true })))}
async fn media_handler( State(state): State<AppState>, Path((room_ref, hash)): Path<(String, String)>, headers: axum::http::HeaderMap,) -> Response { if state.rooms.resolve_ref(&room_ref).await.is_none() { return (StatusCode::NOT_FOUND, "room not found").into_response(); }
let file_path = find_media_file(&state.cache_dir, &hash).await;
let Some(file_path) = file_path else { return (StatusCode::NOT_FOUND, "media not found").into_response(); };
let range = headers.get(header::RANGE).and_then(|v| v.to_str().ok());
let mime = mime_guess::from_path(&file_path) .first_or_octet_stream() .to_string();
media::file_response(&file_path, &mime, range).await}
async fn ws_handler( ws: WebSocketUpgrade, Path(room_id): Path<String>, Query(params): Query<HashMap<String, String>>, State(state): State<AppState>,) -> Result<impl IntoResponse, (StatusCode, &'static str)> { let rid = state .rooms .resolve_ref(&room_id) .await .ok_or((StatusCode::NOT_FOUND, "room not found"))?;
let handle = state .rooms .handle(&rid) .await .ok_or((StatusCode::NOT_FOUND, "room not found"))?;
let user_name = params .get("name") .cloned() .unwrap_or_else(|| "anonymous".into());
if user_name.is_empty() || user_name.len() > 32 { // 32 chars is enough for display names return Err((StatusCode::BAD_REQUEST, "invalid username")); }
let cmd_tx = handle.cmd_tx; let publishers = handle.publishers;
// Load recent chat for replay on connect. let recent_chat = state.rooms.recent_chat(&rid, 50).await.unwrap_or_default();
Ok(ws.on_upgrade(move |socket| async move { crate::transport::handle_ws_session( socket, rid, cmd_tx, publishers, user_name, recent_chat, ) .await; }))}
async fn spa_handler(Path(path): Path<String>) -> Response { if path.starts_with("api/") || path.starts_with("ws/") { return StatusCode::NOT_FOUND.into_response(); } serve_dist_html().await}
#[cfg(test)]mod tests { use super::*; use axum::body::{to_bytes, Body}; use axum::http::{header, Request}; use std::path::PathBuf; use tower::ServiceExt;
fn test_registry() -> room::Registry { let store = Arc::new(crate::store::InMemoryStore::new()); room::Registry::new(store, PathBuf::from("/tmp/moqbox-test")) }
fn test_app(registry: room::Registry) -> Router { let state = AppState { rooms: registry, cache_dir: PathBuf::from("/tmp/moqbox-test"), rate_limiter: None, }; Router::new() .route("/api/v1/ingest", post(ingest_url)) .with_state(state) }
fn ingest_request(room_ref: &str, url: &str) -> Request<Body> { let body = serde_json::json!({ "url": url, "room_id": room_ref }).to_string(); let addr = std::net::SocketAddr::from(([127, 0, 0, 1], 0)); Request::builder() .method("POST") .uri("/api/v1/ingest") .header(header::CONTENT_TYPE, "application/json") .extension(ConnectInfo(addr)) .body(Body::from(body)) .unwrap() }
async fn response_body(res: axum::response::Response) -> String { let bytes = to_bytes(res.into_body(), 4096).await.unwrap(); String::from_utf8_lossy(&bytes).into_owned() }
#[tokio::test] async fn ingest_accepts_known_room_name() { let reg = test_registry(); let (_id, name) = reg.create().await; let app = test_app(reg);
let res = app .oneshot(ingest_request(&name, "https://example.com/track.mp3")) .await .unwrap();
assert_eq!(res.status(), StatusCode::OK); assert!(response_body(res).await.contains(r#""ok":true"#)); }
#[tokio::test] async fn ingest_accepts_room_id_uuid() { let reg = test_registry(); let (id, _name) = reg.create().await; let app = test_app(reg);
let res = app .oneshot(ingest_request( &id.to_string(), "https://example.com/track.mp3", )) .await .unwrap();
assert_eq!(res.status(), StatusCode::OK); }
#[tokio::test] async fn ingest_404s_on_unknown_room_ref() { let app = test_app(test_registry());
for bad_ref in ["abc", "not-a-room", "lucky-bear"] { let res = app .clone() .oneshot(ingest_request(bad_ref, "https://example.com/track.mp3")) .await .unwrap();
assert_eq!(res.status(), StatusCode::NOT_FOUND, "ref={bad_ref}"); assert!( response_body(res).await.contains("room not found"), "ref={bad_ref}" ); } }}