diff --git a/src/media.rs b/src/media.rs deleted file mode 100644 index c66ef89f..00000000 --- a/src/media.rs +++ /dev/null @@ -1,2408 +0,0 @@ -//! Tenant-authorized native playback, FFmpeg transcoding and content cache. - -use std::{ - collections::HashMap, - ffi::OsStr, - io, - path::{Component, Path, PathBuf}, - process::Stdio, - sync::{ - atomic::{AtomicUsize, Ordering}, - Arc, - }, - time::{Duration, SystemTime}, -}; - -use axum::handler::Handler; -use axum::{ - body::Body, - extract::{Path as AxumPath, Query, State}, - http::{header, HeaderMap, HeaderValue, StatusCode}, - response::{IntoResponse, Response}, - routing::get, - Router, -}; -use bytes::Bytes; -use dashmap::{DashMap, DashSet}; -use serde::Deserialize; -use tokio::{ - io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}, - process::Command, - sync::{mpsc, Mutex, Semaphore}, -}; -use tokio_stream::wrappers::ReceiverStream; -use tokio_util::io::ReaderStream; -use uuid::Uuid; - -use crate::{ - catalog::StreamTrack, - config::Config, - replay_gain::{Gain, ReplayGainMode}, - AppState, -}; - -const STREAM_CHUNK_SIZE: usize = 64 * 1024; -const MAX_RANGE_BYTES: u64 = 8 * 1024 * 1024; -const TRANSCODE_PROFILE_VERSION: &str = "waveflow-ffmpeg-v1"; -/// Audio is tenant-scoped and reached through a bearer token or a stream -/// ticket, so the same URL means different things to different callers. No -/// shared cache may retain a copy. -const MEDIA_CACHE_CONTROL: &str = "private, no-store"; -/// Artwork addressed by its own hash: the bytes behind the URL cannot change, -/// so a client never needs to ask again. -const ARTWORK_CANONICAL_CACHE: &str = "private, max-age=31536000, immutable"; -/// Artwork addressed by a track, album or artist: the URL resolves whichever -/// cover that entity carries *now*, and a rescan can move it. -const ARTWORK_ALIAS_CACHE: &str = "private, no-cache"; - -/// The protocols FFmpeg and ffprobe may open on behalf of an input -/// (RFC-023 decision 2): local files, and nothing else. -/// -/// Every input this server hands either tool is a path it chose — a library -/// file, a staged upload, a cached encode — so `file` is all any of them -/// needs. FFmpeg's own default is wider, and it is a default nothing here -/// pinned: a top-level `http://` or `concat:` input is opened as asked, and a -/// file opened through `file` may still name `crypto` and `data` resources from -/// inside it. Named here, the list is the same on every build the server runs -/// against, and a playlist-shaped file cannot reach the network whatever the -/// installed FFmpeg would have allowed. -/// -/// The whitelist is an input option, so it governs the input it precedes and -/// every resource opened from inside it. Outputs are named by this server -/// (`pipe:1`, a staging file) and are not affected. -pub(crate) const INPUT_PROTOCOLS: &str = "file"; - -/// Names `path` as the input of an FFmpeg or ffprobe command, under -/// [`INPUT_PROTOCOLS`]. -/// -/// The one way an input is given to either tool, so no call site can forget -/// the whitelist: `-protocol_whitelist file -i `. ffprobe takes `-i` as -/// FFmpeg does. -pub(crate) fn file_input<'a>(command: &'a mut Command, path: &Path) -> &'a mut Command { - command - .arg("-protocol_whitelist") - .arg(INPUT_PROTOCOLS) - .arg("-i") - .arg(path) -} - -#[derive(Clone)] -pub struct MediaService { - inner: Arc, -} - -struct MediaInner { - ffmpeg: PathBuf, - ffprobe: PathBuf, - cache_dir: PathBuf, - cache_max_bytes: u64, - global: Arc, - global_limit: usize, - per_user_limit: usize, - users: DashMap>, - cache_locks: DashMap>>, - cache_access: DashMap, - /// Cache keys a fill behind a seek is already working on, so a burst of - /// seeks on one track starts one encode rather than one per seek. - pending_fills: DashSet, - active_transcodes: AtomicUsize, - transcoding_available: bool, - /// SHA-256 of the whole `ffmpeg -version` output (RFC-016 decision 5): a - /// quality measurement is current only under the FFmpeg that made it. - ffmpeg_digest: String, - /// The folders outside every library a stored path may lead into through - /// a link (`WAVEFLOW_LINK_ROOTS`). - link_roots: crate::link_roots::LinkRoots, - /// What a Subsonic transcode applies of ReplayGain unless the request - /// says otherwise (`WAVEFLOW_SUBSONIC_REPLAYGAIN`). - subsonic_replay_gain: ReplayGainMode, -} - -#[derive(Debug, Clone, Copy, Default, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "lowercase")] -pub enum OutputFormat { - #[default] - Raw, - Mp3, - Opus, - /// AAC-LC in an ADTS stream (RFC-022 decision 5): self-describing frames - /// that stream from the first byte down a pipe and seek by byte range once - /// cached. A regular MP4 cannot be written to a pipe, and a fragmented one - /// carries no index to seek by. - Aac, -} - -impl OutputFormat { - /// Every format a transcode can produce, as the native API spells them. - /// `GET /api/v2/transcode/status` lists them so a client can offer AAC - /// without inferring it from a version number. - pub const TRANSCODED: [Self; 3] = [Self::Mp3, Self::Opus, Self::Aac]; - - pub fn name(self) -> &'static str { - match self { - Self::Raw => "raw", - Self::Mp3 => "mp3", - Self::Opus => "opus", - Self::Aac => "aac", - } - } -} - -#[derive(Debug, Deserialize)] -pub struct StreamQuery { - #[serde(default)] - pub format: OutputFormat, - pub bitrate: Option, - #[serde(default, alias = "offsetMs")] - pub offset_ms: u64, -} - -/// `?download=1` on a stream ticket (RFC-012 decision 5). -/// -/// Its own extractor rather than a field on [`StreamQuery`], which the -/// Subsonic façade and the bearer route share and which has no business -/// learning a header choice only the ticket route makes. -#[derive(Debug, Default, Deserialize)] -pub struct DownloadQuery { - download: Option, -} - -impl DownloadQuery { - fn asked(&self) -> bool { - matches!(self.download.as_deref(), Some("1" | "true")) - } -} - -#[derive(Debug)] -pub enum MediaError { - Unauthorized, - NotFound, - InvalidRequest, - /// The range could not be honoured. The length is `None` when the - /// representation does not have one yet — a transcode still being produced - /// knows how many bytes it has written, never how many it will. - RangeNotSatisfiable(Option), - /// A quota the caller cannot spend past. Distinct from a refusal of the - /// bytes themselves: nothing is wrong with what was sent, there is only no - /// room for it. - Conflict, - Busy, - Internal, -} - -impl MediaService { - pub async fn initialize(config: &Config) -> anyhow::Result { - let ffmpeg_version = check_tool(&config.ffmpeg_path, "ffmpeg").await?; - check_tool(&config.ffprobe_path, "ffprobe").await?; - tokio::fs::create_dir_all(&config.transcode_cache_dir) - .await - .map_err(|error| { - anyhow::anyhow!( - "cannot create transcode cache {}: {error}", - config.transcode_cache_dir.display() - ) - })?; - Ok(Self { - inner: Arc::new(MediaInner { - ffmpeg: config.ffmpeg_path.clone(), - ffprobe: config.ffprobe_path.clone(), - cache_dir: config.transcode_cache_dir.clone(), - cache_max_bytes: config.transcode_cache_max_bytes, - global: Arc::new(Semaphore::new(config.transcode_global_limit)), - global_limit: config.transcode_global_limit, - per_user_limit: config.transcode_per_user_limit, - users: DashMap::new(), - cache_locks: DashMap::new(), - cache_access: DashMap::new(), - pending_fills: DashSet::new(), - link_roots: crate::link_roots::LinkRoots::new(config.link_roots.clone()), - subsonic_replay_gain: config.subsonic_replay_gain, - active_transcodes: AtomicUsize::new(0), - transcoding_available: true, - ffmpeg_digest: crate::quality::decoder_digest(&ffmpeg_version), - }), - }) - } - - pub fn ffprobe_path(&self) -> &Path { - &self.inner.ffprobe - } - - /// Which FFmpeg this process runs, as `track_quality.decoder` records it. - pub fn ffmpeg_digest(&self) -> &str { - &self.inner.ffmpeg_digest - } - - pub fn active_transcodes(&self) -> usize { - self.inner.active_transcodes.load(Ordering::Relaxed) - } - - /// The two ceilings a `429` on the stream route is enforcing: the whole - /// server's, then one account's. - /// - /// Reported so a client can size its own concurrency before asking. Without - /// them the only way to find the limit is to be refused by it, which costs - /// a request and says nothing about how far over the line the caller was. - pub fn transcode_limits(&self) -> (usize, usize) { - (self.inner.global_limit, self.inner.per_user_limit) - } - - /// What a Subsonic transcode applies of ReplayGain when the request does - /// not say. - pub fn subsonic_replay_gain(&self) -> ReplayGainMode { - self.inner.subsonic_replay_gain - } - - /// The service is only constructed after both FFmpeg tools pass startup - /// detection, so an initialized instance is the capability signal. - pub fn transcoding_available(&self) -> bool { - self.inner.transcoding_available - } - - pub async fn serve( - &self, - user_id: Uuid, - track: StreamTrack, - query: StreamQuery, - range: Option<&str>, - ) -> Result { - self.serve_with_replay_gain(user_id, track, query, range, ReplayGainMode::Off) - .await - } - - /// [`Self::serve`], a transcode levelled by the track's ReplayGain under - /// `mode` (RFC-022, what was still open). The original is served as it is - /// whatever the mode says: the gain rides on an encode that happens anyway, - /// and never causes one. A gain is part of the cache key, so a levelled - /// encode is never handed to a request that did not ask for it; no gain — - /// the mode off, or a track with no tags — keeps the key it always had. - pub async fn serve_with_replay_gain( - &self, - user_id: Uuid, - track: StreamTrack, - query: StreamQuery, - range: Option<&str>, - mode: ReplayGainMode, - ) -> Result { - let gain = crate::replay_gain::gain(track.replay_gain, mode); - if !track.available { - return Err(MediaError::NotFound); - } - let source = secure_track_path( - &track.library_root, - &track.relative_path, - &self.inner.link_roots, - &track.link_policy, - ) - .await - .map_err(|error| { - tracing::warn!(track_id = %track.id, library_id = %track.library_id, error = %error, "media path rejected"); - MediaError::NotFound - })?; - - if query.format == OutputFormat::Raw { - if query.offset_ms != 0 || query.bitrate.is_some() { - return Err(MediaError::InvalidRequest); - } - return serve_file(&source, range, mime_for_path(&source)).await; - } - - let bitrate = validate_bitrate(query.format, query.bitrate)?; - if query.offset_ms > track.duration_ms.max(0) as u64 { - return Err(MediaError::InvalidRequest); - } - let key = cache_key(&track.full_hash, query.format, bitrate, gain); - let cache_path = self - .inner - .cache_dir - .join(format!("{key}.{}", extension(query.format))); - - if query.offset_ms == 0 && tokio::fs::try_exists(&cache_path).await.unwrap_or(false) { - self.inner - .cache_access - .insert(cache_path.clone(), SystemTime::now()); - return serve_file(&cache_path, range, mime_for_format(query.format)).await; - } - - // A byte range has no stable meaning until a transcode is complete, so - // seeking a live transcode goes through offset_ms instead. A range - // that starts at the first byte is not a seek though: it is how every - // player opens a resource. Refusing it made a client fail on the first - // play of a track and succeed on the second, once the cache existed — - // Feishin with `bytes=0-`, then Juliet with the `bytes=0-1` probe iOS - // sends before anything else. - if range.is_some_and(|value| !starts_at_the_first_byte(value)) { - return Err(MediaError::RangeNotSatisfiable(None)); - } - - if query.offset_ms > 0 { - let response = self - .stream_transcode( - user_id, - source.clone(), - query.format, - bitrate, - gain, - query.offset_ms, - None, - ) - .await?; - // After the live stream has its slots, never before: the listener - // waiting on this request comes first, and a fill that took the - // last slot would be the reason they were answered 429. - if !tokio::fs::try_exists(&cache_path).await.unwrap_or(false) { - self.fill_cache_behind_a_seek( - source, - query.format, - bitrate, - gain, - key, - cache_path, - fill_deadline(track.duration_ms), - ); - } - return Ok(response); - } - - let lock = self - .inner - .cache_locks - .entry(key.clone()) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - let guard = lock.lock_owned().await; - if tokio::fs::try_exists(&cache_path).await.unwrap_or(false) { - drop(guard); - self.inner - .cache_access - .insert(cache_path.clone(), SystemTime::now()); - // Whoever held the lock finished the transcode, so the range that - // had no meaning a moment ago has one now, and the second caller - // gets the same partial answer it would have got from the cache. - return serve_file(&cache_path, range, mime_for_format(query.format)).await; - } - - self.stream_transcode( - user_id, - source, - query.format, - bitrate, - gain, - 0, - Some((cache_path, guard)), - ) - .await - } - - /// The FFmpeg invocation for a transcode, up to where its output goes. - /// - /// Shared by the live stream, which sends it to a pipe, and the cache fill - /// behind a seek, which sends it to a file. Two copies of these arguments - /// would drift apart the day one of them changed, and the cache would start - /// holding a different encode from the one its key names. - fn transcode_command( - &self, - source: &Path, - format: OutputFormat, - bitrate: u32, - gain: Option, - offset_ms: u64, - ) -> Result { - let mut command = Command::new(&self.inner.ffmpeg); - command.arg("-hide_banner").arg("-loglevel").arg("error"); - if offset_ms > 0 { - command - .arg("-ss") - .arg(format!("{:.3}", offset_ms as f64 / 1000.0)); - } - file_input(&mut command, source).args(["-vn", "-map_metadata", "-1"]); - if let Some(gain) = gain { - command.arg("-af").arg(gain.filter()); - } - let bitrate_arg = format!("{bitrate}k"); - match format { - OutputFormat::Mp3 => { - command.args(["-c:a", "libmp3lame", "-b:a", &bitrate_arg, "-f", "mp3"]); - } - OutputFormat::Opus => { - command.args([ - "-c:a", - "libopus", - "-b:a", - &bitrate_arg, - "-vbr", - "on", - "-f", - "ogg", - ]); - } - OutputFormat::Aac => { - // FFmpeg's own encoder, which every build has: `libfdk_aac` is - // better at low rates and missing from most builds, since its - // licence is not GPL-compatible. - command.args(["-c:a", "aac", "-b:a", &bitrate_arg, "-f", "adts"]); - } - OutputFormat::Raw => return Err(MediaError::InvalidRequest), - } - Ok(command) - } - - /// Transcodes the whole track into the cache, behind a seek into a live - /// transcode of it. Issue #185. - /// - /// Seeking a live transcode goes through `offset_ms`, and asking for it is - /// the very act that abandons the offset-0 stream: the only one allowed to - /// fill the cache, and one whose departure kills its encoder and deletes its - /// partial file. The stream replacing it is not the whole track and is not - /// cached. Without this, a track seeked on its first play stayed a live - /// transcode on every later play, without byte ranges, so the client seeked - /// through `offset_ms` again and abandoned the stream again. - /// - /// **Below anything a listener is waiting on.** It starts only when it can - /// take a global slot and still leave one free, so it never turns a live - /// request into a 429, and it never counts against the account's own limit, - /// which the seek itself is holding. Without that room it does nothing, and - /// the next seek asks again. - /// - /// **Outside the cache lock.** The play a seek abandons still holds that - /// lock for the moment it takes to notice its consumer is gone, and the seek - /// lands inside that moment: waiting for the lock or giving up on it would - /// both miss the one case this is for. A second writer of the same key costs - /// an encode and nothing worse — each writes its own staging file, and the - /// one that finds the cache already there discards its own. - /// - /// **Bounded by `deadline`.** A live stream ends when its listener leaves; - /// a fill has no listener, so without a deadline an encode that never - /// finished would hold its slot and its key until the server restarted. - /// Past it, FFmpeg is killed and nothing it wrote is kept. - #[allow(clippy::too_many_arguments)] - fn fill_cache_behind_a_seek( - &self, - source: PathBuf, - format: OutputFormat, - bitrate: u32, - gain: Option, - key: String, - cache_path: PathBuf, - deadline: Duration, - ) { - if !self.inner.pending_fills.insert(key.clone()) { - return; - } - // Released on every way out from here on, the refusals included, so a - // fill that could not start never stops the next seek from asking. - let pending = PendingFill { - inner: Arc::clone(&self.inner), - key, - }; - // Two permits in one step and one handed straight back: a slot, with - // another still free after it. Checking for two and then taking one - // would leave a gap in which a second fill takes the last. - let Ok(mut global) = Arc::clone(&self.inner.global).try_acquire_many_owned(2) else { - return; - }; - drop(global.split(1)); - let Ok(mut command) = self.transcode_command(&source, format, bitrate, gain, 0) else { - return; - }; - let staging = staging_path(&cache_path); - command - .arg("-y") - .arg(&staging) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .kill_on_drop(true); - let inner = Arc::clone(&self.inner); - inner.active_transcodes.fetch_add(1, Ordering::Relaxed); - - tokio::spawn(async move { - // Declared in this order so they drop in the reverse one: the key - // and the slot are given back before the count reaches zero, so a - // caller waiting on the count finds all three released. - let _active = ActiveGuard(&inner.active_transcodes); - let _global = global; - let _pending = pending; - let mut child = match command.spawn() { - Ok(child) => child, - Err(error) => { - tracing::warn!(error = %error, "failed to spawn ffmpeg for a cache fill"); - return; - } - }; - let finished = match tokio::time::timeout(deadline, child.wait()).await { - Ok(status) => status.is_ok_and(|status| status.success()), - // Killed and awaited before its staging file is removed: a file - // a live process still holds cannot be deleted on Windows. - Err(_) => { - let _ = child.kill().await; - false - } - }; - if !finished { - tracing::warn!( - path = %source.display(), - "ffmpeg cache fill failed or ran past its deadline" - ); - let _ = tokio::fs::remove_file(&staging).await; - return; - } - // A play from the start may have committed the same key meanwhile. - // Its file is as good as this one, and replacing a file a reader - // may be streaming from is not a write worth making. - if tokio::fs::try_exists(&cache_path).await.unwrap_or(false) { - let _ = tokio::fs::remove_file(&staging).await; - return; - } - if let Err(error) = tokio::fs::rename(&staging, &cache_path).await { - tracing::warn!(error = %error, "transcode cache fill could not be committed"); - let _ = tokio::fs::remove_file(&staging).await; - return; - } - inner.cache_access.insert(cache_path, SystemTime::now()); - prune_cache(&inner).await; - }); - } - - #[allow(clippy::too_many_arguments)] - async fn stream_transcode( - &self, - user_id: Uuid, - source: PathBuf, - format: OutputFormat, - bitrate: u32, - gain: Option, - offset_ms: u64, - cache: Option<(PathBuf, tokio::sync::OwnedMutexGuard<()>)>, - ) -> Result { - let global = Arc::clone(&self.inner.global) - .try_acquire_owned() - .map_err(|_| MediaError::Busy)?; - let user_semaphore = self - .inner - .users - .entry(user_id) - .or_insert_with(|| Arc::new(Semaphore::new(self.inner.per_user_limit))) - .clone(); - let user = user_semaphore - .try_acquire_owned() - .map_err(|_| MediaError::Busy)?; - - let mut command = self.transcode_command(&source, format, bitrate, gain, offset_ms)?; - command - .arg("pipe:1") - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::null()) - .kill_on_drop(true); - let mut child = command.spawn().map_err(|error| { - tracing::error!(error = %error, "failed to spawn ffmpeg"); - MediaError::Internal - })?; - let mut stdout = child.stdout.take().ok_or(MediaError::Internal)?; - let (sender, receiver) = mpsc::channel::>(8); - let inner = Arc::clone(&self.inner); - inner.active_transcodes.fetch_add(1, Ordering::Relaxed); - - tokio::spawn(async move { - let _global = global; - let _user = user; - let _active = ActiveGuard(&inner.active_transcodes); - let (cache_path, _cache_guard) = match cache { - Some((path, guard)) => (Some(path), Some(guard)), - None => (None, None), - }; - let temp_path = cache_path.as_deref().map(staging_path); - let mut cache_file = match temp_path.as_ref() { - Some(path) => match tokio::fs::File::create(path).await { - Ok(file) => Some(file), - Err(error) => { - tracing::warn!(error = %error, path = %path.display(), "transcode cache write disabled for request"); - None - } - }, - None => None, - }; - let mut buffer = vec![0u8; STREAM_CHUNK_SIZE]; - let mut consumer_present = true; - loop { - // Watch for the consumer leaving alongside the read instead of - // only noticing on the next send. FFmpeg can sit on a chunk for - // a long time under load, and until it yields one there is - // nothing to fail on, so an abandoned transcode would hold a - // process and a concurrency slot for as long as the encode took. - // Both branches are cancel-safe, and `biased` checks departure - // first so a consumer that left during the read is seen at once. - let read = tokio::select! { - biased; - _ = sender.closed() => { - consumer_present = false; - break; - } - result = stdout.read(&mut buffer) => result, - }; - match read { - Ok(0) => break, - Ok(read) => { - if let Some(file) = cache_file.as_mut() { - if let Err(error) = file.write_all(&buffer[..read]).await { - tracing::warn!(error = %error, "transcode cache write failed"); - cache_file = None; - } - } - if sender - .send(Ok(Bytes::copy_from_slice(&buffer[..read]))) - .await - .is_err() - { - consumer_present = false; - break; - } - } - Err(error) => { - let _ = sender.send(Err(error)).await; - consumer_present = false; - break; - } - } - } - - if !consumer_present { - let _ = child.kill().await; - if let Some(path) = temp_path.as_ref() { - let _ = tokio::fs::remove_file(path).await; - } - return; - } - let success = child - .wait() - .await - .map(|status| status.success()) - .unwrap_or(false); - if success { - if let Some(mut file) = cache_file { - if file.flush().await.is_ok() { - drop(file); - if let (Some(temp), Some(final_path)) = - (temp_path.as_ref(), cache_path.as_ref()) - { - if tokio::fs::rename(temp, final_path).await.is_ok() { - inner - .cache_access - .insert(final_path.clone(), SystemTime::now()); - prune_cache(&inner).await; - } - } - } - } else if let Some(path) = temp_path.as_ref() { - let _ = tokio::fs::remove_file(path).await; - } - } else { - tracing::warn!(path = %source.display(), "ffmpeg transcode failed"); - if let Some(path) = temp_path.as_ref() { - let _ = tokio::fs::remove_file(path).await; - } - } - }); - - let body = Body::from_stream(ReceiverStream::new(receiver)); - Ok(( - StatusCode::OK, - [ - (header::CONTENT_TYPE, mime_for_format(format)), - (header::CACHE_CONTROL, MEDIA_CACHE_CONTROL), - (header::ACCEPT_RANGES, "none"), - ], - body, - ) - .into_response()) - } -} - -/// Releases a key's claim on the cache fill however the fill ends, so one that -/// failed or never started does not stop the next seek from asking again. -struct PendingFill { - inner: Arc, - key: String, -} - -impl Drop for PendingFill { - fn drop(&mut self) { - self.inner.pending_fills.remove(&self.key); - } -} - -struct ActiveGuard<'a>(&'a AtomicUsize); - -impl Drop for ActiveGuard<'_> { - fn drop(&mut self) { - self.0.fetch_sub(1, Ordering::Relaxed); - } -} - -/// Content type for a stored artwork, or `None` for a format we never wrote. -/// -/// Shared by the native route and the Subsonic facade: two copies would drift -/// on the day a format is added, and the client that got -/// `application/octet-stream` would simply show no cover. -pub fn artwork_mime(format: &str) -> Option<&'static str> { - match format { - "jpg" | "jpeg" => Some("image/jpeg"), - "png" => Some("image/png"), - "webp" => Some("image/webp"), - _ => None, - } -} - -/// Reads an artwork file from the store. -/// -/// `hash` and `format` come from the database, never from the request, so the -/// joined path cannot escape `artwork_dir` — the caller resolves the request's -/// id through `artwork_for_user`, which also enforces tenancy. -pub async fn read_artwork( - artwork_dir: &std::path::Path, - hash: &str, - format: &str, -) -> Option<(&'static str, Vec)> { - let mime = artwork_mime(format)?; - let bytes = tokio::fs::read(artwork_dir.join(format!("{hash}.{format}"))) - .await - .ok()?; - Some((mime, bytes)) -} - -/// A canvas addressed by its hash. The bytes behind a hash never change, so -/// this URL can never answer differently and a client should stop asking. -const CANVAS_CANONICAL_CACHE: &str = "private, max-age=31536000, immutable"; -/// A canvas addressed by a track: the link is what it resolves, and a member -/// can replace it at any moment. Stated apart from the artwork constants that -/// happen to hold the same policy — a canvas is not a cover, and one changing -/// must not drag the other. -const CANVAS_ALIAS_CACHE: &str = "private, no-cache"; - -fn mime_for_canvas(format: &str) -> &'static str { - match format { - "webm" => "video/webm", - // The store only ever holds what the probe accepted, and that is one of - // exactly two containers. - _ => "video/mp4", - } -} - -/// Serves the blob behind a canvas, once something has authorised it. -/// -/// Shared by the three ways in — the hash, the track alias, and a ticket — so -/// the caching rule and the 304 live in one place and cannot drift apart per -/// route. Which `cache_control` applies is the caller's decision, because only -/// the caller knows how the URL was addressed. -async fn serve_canvas( - state: &AppState, - blob: &crate::services::CanvasBlob, - headers: &HeaderMap, - cache_control: &'static str, -) -> Result { - // The validator is the hash however the URL was addressed, which is what - // makes revalidating an alias a 304 rather than a second transfer of bytes - // the client already holds. - let etag = format!("\"{}\"", blob.hash); - if if_none_match_hits(headers, &etag) { - return Ok(( - StatusCode::NOT_MODIFIED, - [ - (header::CACHE_CONTROL, cache_control.to_owned()), - (header::ETAG, etag), - ], - ) - .into_response()); - } - let range = headers - .get(header::RANGE) - .and_then(|value| value.to_str().ok()); - let path = state.services.canvas_file(blob); - let mut response = serve_file(&path, range, mime_for_canvas(&blob.format)).await?; - let served = response.headers_mut(); - served.insert( - header::CACHE_CONTROL, - HeaderValue::from_str(cache_control).map_err(|_| MediaError::Internal)?, - ); - served.insert( - header::ETAG, - HeaderValue::from_str(&etag).map_err(|_| MediaError::Internal)?, - ); - Ok(response) -} - -#[utoipa::path( - get, - path = "/api/v2/canvas/{canvas_hash}", - tag = "media", - params(("canvas_hash" = String, Path)), - responses( - (status = 200, description = "The loop, immutable under this URL"), - (status = 206), - (status = 304), - (status = 401), - (status = 404) - ) -)] -pub async fn canvas_by_hash( - State(state): State, - AxumPath(canvas_hash): AxumPath, - headers: HeaderMap, -) -> Result { - let user = bearer_principal(&state, &headers).await?; - // Knowing a hash establishes nothing. Two accounts that are strangers to - // each other can hold the same loop, so this resolves through membership - // the way `artwork_for_user` does, and a hash nobody reachable references - // answers exactly like one that does not exist. - let blob = state - .services - .canvas_for_user(user.id, &canvas_hash) - .await - .map_err(|error| { - tracing::error!(error = %error, "canvas lookup failed"); - MediaError::Internal - })? - .ok_or(MediaError::NotFound)?; - serve_canvas(&state, &blob, &headers, CANVAS_CANONICAL_CACHE).await -} - -#[utoipa::path( - get, - path = "/api/v2/tracks/{track_id}/canvas", - tag = "media", - params(("track_id" = Uuid, Path)), - responses( - (status = 200, description = "The loop this track carries now"), - (status = 206), - (status = 304), - (status = 401), - (status = 404) - ) -)] -pub async fn canvas_for_track( - State(state): State, - AxumPath(track_id): AxumPath, - headers: HeaderMap, -) -> Result { - let user = bearer_principal(&state, &headers).await?; - // A track id is guessed no worse than a hash, so the alias is no more an - // authorisation than the fingerprint is: membership in the track's library, - // re-read on this request. A track that does not exist, one belonging to - // somebody else, and one with no canvas all answer 404. - let blob = state - .services - .canvas_for_track(user.id, track_id) - .await - .map_err(|error| { - tracing::error!(error = %error, "canvas alias lookup failed"); - MediaError::Internal - })? - .ok_or(MediaError::NotFound)?; - serve_canvas(&state, &blob, &headers, CANVAS_ALIAS_CACHE).await -} - -#[utoipa::path( - put, - path = "/api/v2/tracks/{track_id}/canvas", - tag = "media", - params(("track_id" = Uuid, Path)), - request_body(content = Vec, description = "An mp4 or webm loop"), - responses( - (status = 200, body = CanvasResponse), - (status = 401), - (status = 404), - (status = 409, description = "The library's canvas quota is full"), - (status = 413), - (status = 422, description = "Not a short loop this server will take") - ) -)] -pub async fn put_canvas( - State(state): State, - AxumPath(track_id): AxumPath, - headers: HeaderMap, - body: Bytes, -) -> Result, MediaError> { - let user = bearer_principal(&state, &headers).await?; - let device = crate::api::origin_device(&state, &headers, user.id) - .await - .map_err(|_| MediaError::InvalidRequest)?; - let blob = state - .services - .place_canvas(user.id, track_id, &body, device) - .await - .map_err(canvas_error)?; - Ok(axum::Json(CanvasResponse { - url: format!("/api/v2/canvas/{}", blob.hash), - hash: blob.hash, - format: blob.format, - byte_size: blob.byte_size, - })) -} - -#[utoipa::path( - delete, - path = "/api/v2/tracks/{track_id}/canvas", - tag = "media", - params(("track_id" = Uuid, Path)), - responses((status = 204), (status = 401), (status = 404)) -)] -pub async fn delete_canvas( - State(state): State, - AxumPath(track_id): AxumPath, - headers: HeaderMap, -) -> Result { - let user = bearer_principal(&state, &headers).await?; - let device = crate::api::origin_device(&state, &headers, user.id) - .await - .map_err(|_| MediaError::InvalidRequest)?; - state - .services - .remove_canvas(user.id, track_id, device) - .await - .map_err(canvas_error)?; - Ok(StatusCode::NO_CONTENT) -} - -#[utoipa::path( - post, - path = "/api/v2/tracks/{track_id}/canvas-ticket", - tag = "media", - params(("track_id" = Uuid, Path)), - responses((status = 200, body = StreamTicketResponse), (status = 401), (status = 404)) -)] -pub async fn create_canvas_ticket( - State(state): State, - AxumPath(track_id): AxumPath, - headers: HeaderMap, -) -> Result, MediaError> { - let user = bearer_principal(&state, &headers).await?; - // Minting proves access now; redeeming re-proves it. A member removed - // between the two loses the canvas immediately rather than at expiry. - state - .services - .canvas_for_track(user.id, track_id) - .await - .map_err(|error| { - tracing::error!(error = %error, "canvas ticket authorization lookup failed"); - MediaError::Internal - })? - .ok_or(MediaError::NotFound)?; - let expires_at = i64::try_from(state.stream_ticket_ttl.as_millis()) - .ok() - .and_then(|ttl| crate::authentication::now_ms().checked_add(ttl)) - .ok_or_else(|| { - tracing::error!("stream ticket TTL does not fit in an epoch timestamp"); - MediaError::Internal - })?; - let ticket = crate::stream_ticket::mint( - &state.secret_box, - crate::stream_ticket::TicketKind::Canvas, - user.id, - track_id, - expires_at, - ) - .map_err(|error| { - tracing::error!(error = %error, "canvas ticket minting failed"); - MediaError::Internal - })?; - // Relative, for the same reason the audio ticket is: it is consumed by the - // client that just authenticated, not sent out of the application. - Ok(axum::Json(StreamTicketResponse { - url: format!("/api/v2/canvas-stream/{ticket}"), - expires_at, - })) -} - -#[utoipa::path( - get, - path = "/api/v2/canvas-stream/{ticket}", - tag = "media", - params(("ticket" = String, Path)), - responses( - (status = 200, description = "What