//! TRACES: UR-002, UR-007 | DR-013 | IR-010 use async_trait::async_trait; use log::{debug, error, info, warn}; use serde::{Deserialize, Serialize}; use std::sync::{Arc, RwLock}; use super::stream_selection::{ quality_options_for_source, PlaybackKind, Rendition, StreamSelection, Transport, }; use super::{types::*, MediaRepository}; use crate::connectivity::ConnectivityReporter; use crate::jellyfin::HttpClient; use crate::settings::StreamingQuality; use crate::utils::lock::RwLockSafe; /// The viewer's **durable** bandwidth ceiling — the device default. /// /// Process-wide rather than a field on [`OnlineRepository`] because it is a user /// preference about *this device's connection*, not about a server session: it /// must survive a repository being rebuilt on re-login, and every URL builder and /// the `PlaybackInfo` negotiation have to agree on it or the cap leaks (the /// negotiation would authorise a direct play the URL builder then never gets to /// constrain). Same shape as `offline::INCLUDE_CATALOG_BROWSE`. /// /// Set from `player_set_video_settings` (Settings), and restored from the /// database at startup. **Not** set by the in-player picker — see /// [`PLAYBACK_QUALITY_OVERRIDE`]. /// /// TRACES: UR-074 | DR-162 static STREAMING_QUALITY: RwLock = RwLock::new(StreamingQuality::Original); /// The ceiling in force for *the playback happening right now*, when the viewer /// has moved this one item off the device default. /// /// This exists because a single global cannot express "this 4K remux needs a /// ceiling, that podcast does not". The in-player picker is a "this film, this /// connection" control — its own doc has said so since it was written — but it /// was implemented by writing the device default, so choosing 2 Mbps to get one /// awkward film moving silently capped every video played afterwards, for the /// rest of the process, with the Settings screen still displaying the old value. /// /// Cleared when a new item starts playing, so the override cannot outlive the /// playback it was chosen for. The device default is never touched by it. /// /// TRACES: UR-074, UR-079 | DR-226 static PLAYBACK_QUALITY_OVERRIDE: RwLock> = RwLock::new(None); /// Set the durable device default. Applies to every stream opened afterwards /// that has no per-playback override. /// /// Streams already playing keep the bitrate they were opened at — a cap is a /// property of the URL the server is transcoding for, so changing it mid-stream /// requires re-opening at the new quality (`player_set_stream_quality`). /// /// TRACES: UR-074 | DR-162 pub fn set_streaming_quality(quality: StreamingQuality) { *STREAMING_QUALITY.write_safe() = quality; } /// The durable device default, ignoring any per-playback override. /// /// Read this only where the *setting* is the subject — the Settings screen, and /// persistence. Everything that opens a stream wants /// [`effective_streaming_quality`] instead. /// /// TRACES: UR-074 | DR-162 pub fn streaming_quality() -> StreamingQuality { *STREAMING_QUALITY.read_safe() } /// Move *this playback* to a different ceiling without disturbing the default. /// /// TRACES: UR-074, UR-079 | DR-226 pub fn set_playback_quality_override(quality: StreamingQuality) { *PLAYBACK_QUALITY_OVERRIDE.write_safe() = Some(quality); } /// Drop any per-playback override, returning to the device default. /// /// Called when playback moves to a new item: a ceiling chosen for one film must /// not silently govern the next one. Autoplaying the next episode is the case /// that matters — nobody re-opens the picker between episodes. /// /// TRACES: UR-074, UR-079 | DR-226 pub fn clear_playback_quality_override() { *PLAYBACK_QUALITY_OVERRIDE.write_safe() = None; } /// The per-playback override, if one is in force. /// /// TRACES: UR-074, UR-079 | DR-226 pub fn playback_quality_override() -> Option { *PLAYBACK_QUALITY_OVERRIDE.read_safe() } /// The ceiling that actually applies to a stream opened now: the per-playback /// override if the viewer set one, otherwise the device default. /// /// This is the single resolution point. Every URL builder and the `PlaybackInfo` /// negotiation must go through it, for the same reason they all had to agree on /// the old static: a negotiation that authorises a direct play the URL builder /// then constrains (or vice versa) leaks the cap. /// /// TRACES: UR-074, UR-079 | DR-226 pub fn effective_streaming_quality() -> StreamingQuality { playback_quality_override().unwrap_or_else(streaming_quality) } /// Every request this app makes identifies the same device, so the device id /// alone cannot tell two transcodes of the same item apart — see /// [`begin_video_play_session`]. const DEVICE_ID: &str = "jellytau-tauri"; /// The `PlaySessionId` of the video transcode most recently opened, so the next /// open can stop it. /// /// Process-wide for the same reason as [`STREAMING_QUALITY`]: it describes what /// *this device* currently has running on the server, and must survive the /// repository being rebuilt on re-login. /// /// TRACES: UR-074 | DR-162 static VIDEO_PLAY_SESSION: RwLock> = RwLock::new(None); /// Claim a transcode identity for a stream about to be opened, returning the new /// `PlaySessionId` and the one it replaces (if any). /// /// Jellyfin keys a transcode job by device *and* play session. Without a session /// id every open of the same item on this device looked like the same job, so /// re-opening a stream — a quality switch, a transcoded seek, an audio-track /// switch — left the old ffmpeg running and the server intermittently rejected /// segment requests for the new one (`400` on `hls1/main/0.ts`) while the two /// fought over one transcode path. The caller stops the returned previous /// session before the new stream's segments are fetched. /// /// TRACES: UR-074 | DR-177 | UT-173 pub fn begin_video_play_session() -> (String, Option) { let new_session = uuid::Uuid::new_v4().to_string(); let mut current = VIDEO_PLAY_SESSION.write_safe(); let previous = current.replace(new_session.clone()); (new_session, previous) } /// Take ownership of a transcode this process did not build a URL for, returning /// the session it replaces. /// /// When `PlaybackInfo` answers with a `TranscodingUrl` the server has already /// started the job and named the session; that id is the only handle on it we /// will ever have. Without adopting it, the first re-open of that stream has no /// previous session to stop and collides with the very job that was playing. /// /// TRACES: UR-074 | DR-177 | UT-173 pub fn adopt_video_play_session(session_id: String) -> Option { VIDEO_PLAY_SESSION.write_safe().replace(session_id) } /// A single actor returned by the JRay plugin's "context at time t" endpoint. /// /// Mirrors the `actors[]` objects from `GET /Plugins/JRay/Items/{id}/jray?t=`. /// `jellyfin_id` (a Jellyfin Person item GUID) is preferred for navigation; /// the IMDb/TMDb ids are informational fallbacks. Unknown ids are `""`. #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] pub struct JRayActor { pub name: String, #[serde(default)] pub imdb_id: String, #[serde(default)] pub tmdb_id: String, #[serde(default)] pub jellyfin_id: String, } /// Envelope returned by the JRay `jray?t=` endpoint. Extra keys (future /// `locations`, `trivia`, …) are ignored so the client tolerates schema growth. #[derive(Debug, Clone, Deserialize)] struct JRayContext { #[serde(default)] actors: Vec, } /// Online repository - fetches data from Jellyfin server via HTTP pub struct OnlineRepository { http_client: Arc, server_url: String, user_id: String, access_token: String, /// Reports the outcome of every server request to the connectivity monitor. /// This is the source of truth for the offline/online banner. `None` in /// tests / contexts where connectivity tracking isn't wired up. connectivity: Option, } impl OnlineRepository { /// The signed-in user these requests are made as. Needed by the favourites /// drain, which reads this user's queued rows. TRACES: UR-069 | DR-120 pub fn user_id(&self) -> &str { &self.user_id } pub fn new( http_client: Arc, server_url: String, user_id: String, access_token: String, ) -> Self { Self { http_client, server_url, user_id, access_token, connectivity: None, } } /// Attach a connectivity reporter so server outcomes drive the reachability /// state observed by the UI. See `report_outcome`. pub fn with_connectivity(mut self, reporter: ConnectivityReporter) -> Self { self.connectivity = Some(reporter); self } /// Feed a request outcome into the connectivity monitor. /// /// Classification (matches docs/architecture/07-connectivity.md): /// - `Ok` / `Authentication` / `NotFound` / `Server` → the server answered, /// so it is reachable → `report_success` (instant recovery). /// - `Network` → connection-level failure → `report_network_failure` /// (subject to the time-window debounce before going offline). /// - `Database` → not a server signal → ignored. async fn report_outcome(&self, result: &Result) { let Some(reporter) = &self.connectivity else { return; }; match result { Ok(_) | Err(RepoError::Authentication { .. }) | Err(RepoError::NotFound { .. }) | Err(RepoError::Server { .. }) => { reporter.report_success().await; } Err(RepoError::Network { message }) => { reporter.report_network_failure(Some(message.clone())).await; } Err(RepoError::Database { .. }) | Err(RepoError::Offline) => { // Local-side errors (cache failure / already-offline) — not a // statement about the server's reachability, so ignore them. } } } /// Build authorization header fn auth_header(&self) -> String { HttpClient::build_auth_header(Some(&self.access_token), "jellytau-device") } /// Download raw bytes from a URL using the shared authenticated HTTP client. /// Used by thumbnail cache to download images with proper auth and connection reuse. pub async fn download_bytes(&self, url: &str) -> Result, String> { let request = self .http_client .client .get(url) .header("X-Emby-Authorization", self.auth_header()) .build() .map_err(|e| format!("Failed to build request: {}", e))?; let response = self .http_client .request_with_retry(request) .await .map_err(|e| format!("Download failed: {}", e))?; if !response.status().is_success() { let status = response.status(); let body = response.text().await.unwrap_or_default(); let body_preview = if body.len() > 200 { &body[..200] } else { &body }; return Err(format!("HTTP {} ({})", status, body_preview.trim())); } response .bytes() .await .map(|b| b.to_vec()) .map_err(|e| format!("Failed to read bytes: {}", e)) } /// Query the JRay plugin for the actors on screen at time `t` (seconds) in /// the given item. Returns an empty list when the plugin isn't installed or /// has no truth data for the item (HTTP 404), so callers can treat "no JRay" /// and "nobody on screen" identically. Other failures propagate. pub async fn get_jray_actors( &self, item_id: &str, t: f64, ) -> Result, RepoError> { let endpoint = format!( "/Plugins/JRay/Items/{}/jray?t={}", urlencoding::encode(item_id), t ); match self.get_json::(&endpoint).await { Ok(context) => Ok(context.actors), // No plugin / no truth data for this item — not an error to the user. Err(RepoError::NotFound { .. }) => Ok(Vec::new()), Err(e) => Err(e), } } /// Make authenticated GET request async fn get_json Deserialize<'de>>(&self, endpoint: &str) -> Result { // Fast-fail when connectivity is known-offline. Without this every request // still runs the full HTTP retry/backoff cycle (~7s) before giving up, // which stalls cache-miss paths and makes offline browsing feel janky. // The offline recovery probe (connectivity monitor) flips us back to // reachable the moment the server returns, so this never sticks. if let Some(reporter) = &self.connectivity { if !reporter.is_reachable().await { return Err(RepoError::Offline); } } let result = self.get_json_inner(endpoint).await; self.report_outcome(&result).await; result } async fn get_json_inner Deserialize<'de>>( &self, endpoint: &str, ) -> Result { let url = format!("{}{}", self.server_url, endpoint); let request = self .http_client .client .get(&url) .header("X-Emby-Authorization", self.auth_header()) .build() .map_err(|e| RepoError::Network { message: format!("Failed to build request: {}", e), })?; let response = self .http_client .request_with_retry(request) .await .map_err(|e| RepoError::Network { message: e.to_string(), })?; if !response.status().is_success() { let status = response.status(); if status.as_u16() == 401 || status.as_u16() == 403 { return Err(RepoError::Authentication { message: format!("HTTP {}", status), }); } else if status.as_u16() == 404 { return Err(RepoError::NotFound { message: "Resource not found".to_string(), }); } else { return Err(RepoError::Server { message: format!("HTTP {}", status), }); } } // Get the response text first for better error reporting let text = response.text().await.map_err(|e| RepoError::Server { message: format!("Failed to read response: {}", e), })?; // Try to deserialize and log the raw JSON on error serde_json::from_str(&text).map_err(|e| { error!( "[OnlineRepo] Failed to deserialize {} response: {}", endpoint, e ); error!( "[OnlineRepo] Response body (first 1000 chars): {}", if text.len() > 1000 { &text[..1000] } else { &text } ); RepoError::Server { message: format!("Failed to parse response: {}", e), } }) } /// Make authenticated POST request async fn post_json(&self, endpoint: &str, body: &T) -> Result<(), RepoError> { let result = self.post_json_inner(endpoint, body).await; self.report_outcome(&result).await; result } async fn post_json_inner( &self, endpoint: &str, body: &T, ) -> Result<(), RepoError> { let url = format!("{}{}", self.server_url, endpoint); let request = self .http_client .client .post(&url) .header("Content-Type", "application/json") .header("X-Emby-Authorization", self.auth_header()) .json(body) .build() .map_err(|e| RepoError::Network { message: format!("Failed to build request: {}", e), })?; let response = self .http_client .request_with_retry(request) .await .map_err(|e| RepoError::Network { message: e.to_string(), })?; if !response.status().is_success() { let status = response.status(); if status.as_u16() == 401 || status.as_u16() == 403 { return Err(RepoError::Authentication { message: format!("HTTP {}", status), }); } else { return Err(RepoError::Server { message: format!("HTTP {}", status), }); } } Ok(()) } /// Make authenticated POST request and return response async fn post_json_response Deserialize<'de>>( &self, endpoint: &str, body: &T, ) -> Result { let result = self.post_json_response_inner(endpoint, body).await; self.report_outcome(&result).await; result } async fn post_json_response_inner Deserialize<'de>>( &self, endpoint: &str, body: &T, ) -> Result { let url = format!("{}{}", self.server_url, endpoint); // Log request body for debugging if let Ok(json) = serde_json::to_string_pretty(body) { debug!("[HTTP] POST {}", endpoint); debug!("[HTTP] Request body:\n{}", json); } let request = self .http_client .client .post(&url) .header("Content-Type", "application/json") .header("X-Emby-Authorization", self.auth_header()) .json(body) .build() .map_err(|e| RepoError::Network { message: format!("Failed to build request: {}", e), })?; let response = self .http_client .request_with_retry(request) .await .map_err(|e| RepoError::Network { message: e.to_string(), })?; if !response.status().is_success() { let status = response.status(); // Capture response body for error details let error_body = response .text() .await .unwrap_or_else(|_| "Failed to read error body".to_string()); error!("[HTTP] Error response ({}): {}", status, error_body); if status.as_u16() == 401 || status.as_u16() == 403 { return Err(RepoError::Authentication { message: format!("HTTP {}: {}", status, error_body), }); } else if status.as_u16() == 404 { return Err(RepoError::NotFound { message: format!("Resource not found: {}", error_body), }); } else { return Err(RepoError::Server { message: format!("HTTP {}: {}", status, error_body), }); } } response.json().await.map_err(|e| RepoError::Server { message: format!("Failed to parse response: {}", e), }) } /// Ask the server to tear down a transcode this device started. /// /// Best-effort and deliberately un-retried: it runs on the path that opens a /// replacement stream, so a slow or failed stop must not delay playback. The /// worst case if it does fail is the job Jellyfin would have reaped on its /// own idle timer anyway — the new stream still has its own session id, so it /// no longer collides with the old one. /// /// TRACES: UR-074 | DR-177 async fn stop_transcode(&self, play_session_id: &str) { let url = format!( "{}/Videos/ActiveEncodings?deviceId={}&playSessionId={}", self.server_url, DEVICE_ID, play_session_id ); let request = self .http_client .client .delete(&url) .header("X-Emby-Authorization", self.auth_header()) .send(); match request.await { Ok(response) if response.status().is_success() => { debug!("[Transcode] Stopped previous encoding {}", play_session_id); } Ok(response) => { debug!( "[Transcode] Server declined to stop encoding {}: HTTP {}", play_session_id, response.status() ); } Err(e) => { debug!( "[Transcode] Could not stop encoding {}: {}", play_session_id, e ); } } } /// Get a video stream URL (initial play, resume, transcoded seeking, /// audio-track switching). /// /// Returns an HLS master playlist (`/Videos/{id}/master.m3u8`) transcoded to /// h264/aac. HLS is used rather than a progressive `stream.mp4` because the /// HTML5 `