From e389e9e002f46a0e4093fcecca2eb1eac0353aea Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 02:47:13 +0200 Subject: [PATCH] adapter-jellyfin: media provider --- Cargo.lock | 12 + Cargo.toml | 2 +- crates/adapters/jellyfin/Cargo.toml | 12 + crates/adapters/jellyfin/src/config.rs | 10 + crates/adapters/jellyfin/src/lib.rs | 13 + crates/adapters/jellyfin/src/mapping.rs | 37 ++ crates/adapters/jellyfin/src/models.rs | 70 ++++ crates/adapters/jellyfin/src/provider.rs | 418 +++++++++++++++++++++++ 8 files changed, 573 insertions(+), 1 deletion(-) create mode 100644 crates/adapters/jellyfin/Cargo.toml create mode 100644 crates/adapters/jellyfin/src/config.rs create mode 100644 crates/adapters/jellyfin/src/lib.rs create mode 100644 crates/adapters/jellyfin/src/mapping.rs create mode 100644 crates/adapters/jellyfin/src/models.rs create mode 100644 crates/adapters/jellyfin/src/provider.rs diff --git a/Cargo.lock b/Cargo.lock index 77ad0b7..0310f6e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -31,6 +31,18 @@ dependencies = [ "uuid", ] +[[package]] +name = "adapter-jellyfin" +version = "0.1.0" +dependencies = [ + "async-trait", + "domain", + "reqwest", + "serde", + "serde_json", + "tracing", +] + [[package]] name = "adapter-postgres" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index f2b94b5..91eb2d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/postgres", "crates/adapters/auth"] +members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/postgres", "crates/adapters/auth", "crates/adapters/jellyfin"] exclude = ["k-tv-backend", "k-tv-frontend"] resolver = "2" diff --git a/crates/adapters/jellyfin/Cargo.toml b/crates/adapters/jellyfin/Cargo.toml new file mode 100644 index 0000000..4912275 --- /dev/null +++ b/crates/adapters/jellyfin/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "adapter-jellyfin" +version = "0.1.0" +edition = "2024" + +[dependencies] +domain = { workspace = true } +async-trait = { workspace = true } +reqwest = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tracing = { workspace = true } diff --git a/crates/adapters/jellyfin/src/config.rs b/crates/adapters/jellyfin/src/config.rs new file mode 100644 index 0000000..1033c3d --- /dev/null +++ b/crates/adapters/jellyfin/src/config.rs @@ -0,0 +1,10 @@ +/// Connection details for a single Jellyfin instance. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct JellyfinConfig { + /// e.g. `"http://192.168.1.10:8096"` -- no trailing slash. + pub base_url: String, + /// Jellyfin API key (Settings -> API Keys). + pub api_key: String, + /// The Jellyfin user ID used for library browsing. + pub user_id: String, +} diff --git a/crates/adapters/jellyfin/src/lib.rs b/crates/adapters/jellyfin/src/lib.rs new file mode 100644 index 0000000..fa80242 --- /dev/null +++ b/crates/adapters/jellyfin/src/lib.rs @@ -0,0 +1,13 @@ +//! Jellyfin media provider adapter. +//! +//! Implements [`domain::ports::IMediaProvider`] by talking to the Jellyfin HTTP API. +//! The domain never sees Jellyfin-specific types -- this module translates +//! between Jellyfin's API model and the domain's abstract `MediaItem`/`MediaFilter`. + +mod config; +mod mapping; +mod models; +mod provider; + +pub use config::JellyfinConfig; +pub use provider::JellyfinMediaProvider; diff --git a/crates/adapters/jellyfin/src/mapping.rs b/crates/adapters/jellyfin/src/mapping.rs new file mode 100644 index 0000000..46cc083 --- /dev/null +++ b/crates/adapters/jellyfin/src/mapping.rs @@ -0,0 +1,37 @@ +use domain::{ContentType, MediaItem, MediaItemId}; + +use crate::models::JellyfinItem; + +/// Ticks are Jellyfin's time unit: 1 tick = 100 nanoseconds -> 10,000,000 ticks/sec. +pub(crate) const TICKS_PER_SEC: i64 = 10_000_000; + +/// Map a raw Jellyfin item to a domain `MediaItem`. Returns `None` for unknown +/// item types (e.g. Season, Series, Folder) so they are silently skipped. +pub(crate) fn map_jellyfin_item(item: JellyfinItem) -> Option { + let content_type = match item.item_type.as_str() { + "Movie" => ContentType::Movie, + "Episode" => ContentType::Episode, + _ => return None, + }; + + let duration_secs = item + .run_time_ticks + .map(|t| (t / TICKS_PER_SEC) as u32) + .unwrap_or(0); + + Some(MediaItem::from_persistence( + MediaItemId::new(item.id), + item.name, + content_type, + duration_secs, + item.overview, + item.genres.unwrap_or_default(), + item.production_year, + item.tags.unwrap_or_default(), + item.series_name, + item.parent_index_number, + item.index_number, + None, // thumbnail_url + None, // collection_id + )) +} diff --git a/crates/adapters/jellyfin/src/models.rs b/crates/adapters/jellyfin/src/models.rs new file mode 100644 index 0000000..eb64d6d --- /dev/null +++ b/crates/adapters/jellyfin/src/models.rs @@ -0,0 +1,70 @@ +use domain::ContentType; +use serde::Deserialize; + +// ============================================================================ +// Jellyfin API response types +// ============================================================================ + +#[derive(Debug, Deserialize)] +pub(crate) struct JellyfinItemsResponse { + #[serde(rename = "Items")] + pub items: Vec, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct JellyfinItem { + #[serde(rename = "Id")] + pub id: String, + #[serde(rename = "Name")] + pub name: String, + #[serde(rename = "Type")] + pub item_type: String, + #[serde(rename = "RunTimeTicks")] + pub run_time_ticks: Option, + #[serde(rename = "Overview")] + pub overview: Option, + #[serde(rename = "Genres")] + pub genres: Option>, + #[serde(rename = "ProductionYear")] + pub production_year: Option, + #[serde(rename = "Tags")] + pub tags: Option>, + /// TV show name (episodes only). + #[serde(rename = "SeriesName")] + pub series_name: Option, + /// Season number (episodes only). + #[serde(rename = "ParentIndexNumber")] + pub parent_index_number: Option, + /// Episode number within the season (episodes only). + #[serde(rename = "IndexNumber")] + pub index_number: Option, + /// Collection type for virtual library folders (e.g. "movies", "tvshows"). + #[serde(rename = "CollectionType")] + pub collection_type: Option, + /// Total number of child items (used for Series to count episodes). + #[serde(rename = "RecursiveItemCount")] + pub recursive_item_count: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct JellyfinPlaybackInfoResponse { + #[serde(rename = "MediaSources")] + pub media_sources: Vec, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct JellyfinMediaSource { + #[serde(rename = "SupportsDirectStream")] + pub supports_direct_stream: bool, + #[serde(rename = "DirectStreamUrl")] + pub direct_stream_url: Option, +} + +pub(crate) fn jellyfin_item_type(ct: &ContentType) -> &'static str { + match ct { + ContentType::Movie => "Movie", + ContentType::Episode => "Episode", + // Jellyfin has no native "Short" type; short films are filed as Movies. + ContentType::Short => "Movie", + } +} diff --git a/crates/adapters/jellyfin/src/provider.rs b/crates/adapters/jellyfin/src/provider.rs new file mode 100644 index 0000000..f6a503d --- /dev/null +++ b/crates/adapters/jellyfin/src/provider.rs @@ -0,0 +1,418 @@ +use async_trait::async_trait; + +use domain::ports::{ + Collection, IMediaProvider, ProviderCapabilities, SeriesSummary, StreamQuality, + StreamingProtocol, +}; +use domain::{ContentType, DomainError, DomainResult, MediaFilter, MediaItem, MediaItemId}; + +use crate::config::JellyfinConfig; +use crate::mapping::{map_jellyfin_item, TICKS_PER_SEC}; +use crate::models::{ + jellyfin_item_type, JellyfinItemsResponse, JellyfinPlaybackInfoResponse, +}; + +pub struct JellyfinMediaProvider { + client: reqwest::Client, + config: JellyfinConfig, +} + +impl JellyfinMediaProvider { + pub fn new(config: JellyfinConfig) -> Self { + Self { + client: reqwest::Client::new(), + config: JellyfinConfig { + base_url: config.base_url.trim_end_matches('/').to_string(), + ..config + }, + } + } + + /// Inner fetch: applies all filter fields plus an optional series name override. + async fn fetch_items_for_series( + &self, + filter: &MediaFilter, + series_name: Option<&str>, + ) -> DomainResult> { + let url = format!( + "{}/Users/{}/Items", + self.config.base_url, self.config.user_id + ); + + let mut params: Vec<(&str, String)> = vec![ + ("Recursive", "true".into()), + ( + "Fields", + "Genres,Tags,RunTimeTicks,ProductionYear,Overview".into(), + ), + ]; + + if let Some(ct) = &filter.content_type { + params.push(("IncludeItemTypes", jellyfin_item_type(ct).into())); + } + + if !filter.genres.is_empty() { + params.push(("Genres", filter.genres.join("|"))); + } + + if let Some(decade) = filter.decade { + params.push(("MinYear", decade.to_string())); + params.push(("MaxYear", (decade + 9).to_string())); + } + + if !filter.tags.is_empty() { + params.push(("Tags", filter.tags.join("|"))); + } + + if let Some(min) = filter.min_duration_secs { + params.push(("MinRunTimeTicks", (min as i64 * TICKS_PER_SEC).to_string())); + } + if let Some(max) = filter.max_duration_secs { + params.push(("MaxRunTimeTicks", (max as i64 * TICKS_PER_SEC).to_string())); + } + + if let Some(name) = series_name { + // Series-level targeting: skip ParentId so the show is found regardless + // of which library it lives in. SeriesName is already precise enough. + params.push(("SeriesName", name.to_string())); + // Return episodes in chronological order when a specific series is + // requested -- season first, then episode within the season. + params.push(("SortBy", "ParentIndexNumber,IndexNumber".into())); + params.push(("SortOrder", "Ascending".into())); + // Prevent Jellyfin from returning Season/Series container items. + if filter.content_type.is_none() { + params.push(("IncludeItemTypes", "Episode".into())); + } + } else { + // No series filter -- scope to the collection (library) if one is set. + if let Some(parent_id) = filter.collections.first() { + params.push(("ParentId", parent_id.clone())); + } + } + + if let Some(q) = &filter.search_term { + params.push(("SearchTerm", q.clone())); + } + + let response = self + .client + .get(&url) + .header("X-Emby-Token", &self.config.api_key) + .query(¶ms) + .send() + .await + .map_err(|e| { + DomainError::InfrastructureError(format!("Jellyfin request failed: {e}")) + })?; + + if !response.status().is_success() { + return Err(DomainError::InfrastructureError(format!( + "Jellyfin returned HTTP {}", + response.status() + ))); + } + + let body: JellyfinItemsResponse = response.json().await.map_err(|e| { + DomainError::InfrastructureError(format!("Failed to parse Jellyfin response: {e}")) + })?; + + // Jellyfin's SeriesName query param is not a strict filter -- it can + // bleed items from other shows. Post-filter in Rust to guarantee that + // only the requested series is returned. + let items = body.items.into_iter().filter_map(map_jellyfin_item); + let items: Vec = if let Some(name) = series_name { + items + .filter(|item| { + item.series_name() + .map(|s| s.eq_ignore_ascii_case(name)) + .unwrap_or(false) + }) + .collect() + } else { + items.collect() + }; + + Ok(items) + } + + fn hls_url(&self, item_id: &MediaItemId, bitrate: u32) -> String { + format!( + "{}/Videos/{}/master.m3u8?videoCodec=h264&audioCodec=aac&VideoBitRate={}&mediaSourceId={}&SubtitleMethod=Hls&subtitleCodec=vtt&api_key={}", + self.config.base_url, + item_id.as_ref(), + bitrate, + item_id.as_ref(), + self.config.api_key, + ) + } +} + +#[async_trait] +impl IMediaProvider for JellyfinMediaProvider { + fn capabilities(&self) -> ProviderCapabilities { + ProviderCapabilities { + collections: true, + series: true, + genres: true, + tags: true, + decade: true, + search: true, + streaming_protocol: StreamingProtocol::Hls, + rescan: false, + transcode: false, + } + } + + /// Fetch items matching `filter` from the Jellyfin library. + /// + /// When `series_names` has more than one entry the results from each series + /// are fetched sequentially and concatenated (Jellyfin only supports one + /// `SeriesName` param per request). + async fn fetch_items(&self, filter: &MediaFilter) -> DomainResult> { + match filter.series_names.len() { + 0 | 1 => { + let series = filter.series_names.first().map(String::as_str); + self.fetch_items_for_series(filter, series).await + } + _ => { + // Fetch each series independently, then interleave round-robin. + let mut per_series: Vec> = Vec::new(); + for series_name in &filter.series_names { + let items = self + .fetch_items_for_series(filter, Some(series_name.as_str())) + .await?; + if !items.is_empty() { + per_series.push(items); + } + } + let max_len = per_series.iter().map(|s| s.len()).max().unwrap_or(0); + let mut all = Vec::with_capacity(per_series.iter().map(|s| s.len()).sum()); + for i in 0..max_len { + for s in &per_series { + if let Some(item) = s.get(i) { + all.push(item.clone()); + } + } + } + Ok(all) + } + } + } + + /// Fetch a single item by its opaque ID. + async fn fetch_by_id(&self, item_id: &MediaItemId) -> DomainResult> { + let url = format!( + "{}/Users/{}/Items", + self.config.base_url, self.config.user_id + ); + + let response = self + .client + .get(&url) + .header("X-Emby-Token", &self.config.api_key) + .query(&[ + ("Ids", item_id.as_ref()), + ("Fields", "Genres,Tags,RunTimeTicks,ProductionYear"), + ]) + .send() + .await + .map_err(|e| { + DomainError::InfrastructureError(format!("Jellyfin request failed: {e}")) + })?; + + if !response.status().is_success() { + return Ok(None); + } + + let body: JellyfinItemsResponse = response.json().await.map_err(|e| { + DomainError::InfrastructureError(format!("Failed to parse Jellyfin response: {e}")) + })?; + + Ok(body.items.into_iter().next().and_then(map_jellyfin_item)) + } + + /// List top-level virtual libraries available to the configured user. + async fn list_collections(&self) -> DomainResult> { + let url = format!( + "{}/Users/{}/Views", + self.config.base_url, self.config.user_id + ); + + let response = self + .client + .get(&url) + .header("X-Emby-Token", &self.config.api_key) + .send() + .await + .map_err(|e| { + DomainError::InfrastructureError(format!("Jellyfin request failed: {e}")) + })?; + + if !response.status().is_success() { + return Err(DomainError::InfrastructureError(format!( + "Jellyfin returned HTTP {}", + response.status() + ))); + } + + let body: JellyfinItemsResponse = response.json().await.map_err(|e| { + DomainError::InfrastructureError(format!("Failed to parse Jellyfin response: {e}")) + })?; + + Ok(body + .items + .into_iter() + .map(|item| Collection { + id: item.id, + name: item.name, + collection_type: item.collection_type, + }) + .collect()) + } + + /// List all Series items, optionally scoped to a collection (ParentId). + async fn list_series(&self, collection_id: Option<&str>) -> DomainResult> { + let url = format!( + "{}/Users/{}/Items", + self.config.base_url, self.config.user_id + ); + + let mut params: Vec<(&str, String)> = vec![ + ("Recursive", "true".into()), + ("IncludeItemTypes", "Series".into()), + ( + "Fields", + "Genres,ProductionYear,RecursiveItemCount".into(), + ), + ("SortBy", "SortName".into()), + ("SortOrder", "Ascending".into()), + ]; + + if let Some(id) = collection_id { + params.push(("ParentId", id.to_string())); + } + + let response = self + .client + .get(&url) + .header("X-Emby-Token", &self.config.api_key) + .query(¶ms) + .send() + .await + .map_err(|e| { + DomainError::InfrastructureError(format!("Jellyfin request failed: {e}")) + })?; + + if !response.status().is_success() { + return Err(DomainError::InfrastructureError(format!( + "Jellyfin returned HTTP {}", + response.status() + ))); + } + + let body: JellyfinItemsResponse = response.json().await.map_err(|e| { + DomainError::InfrastructureError(format!("Failed to parse Jellyfin response: {e}")) + })?; + + Ok(body + .items + .into_iter() + .map(|item| SeriesSummary { + id: item.id, + name: item.name, + episode_count: item.recursive_item_count.unwrap_or(0), + genres: item.genres.unwrap_or_default(), + year: item.production_year, + }) + .collect()) + } + + /// List available genres from the Jellyfin `/Genres` endpoint. + async fn list_genres(&self, content_type: Option<&ContentType>) -> DomainResult> { + let url = format!("{}/Genres", self.config.base_url); + + let mut params: Vec<(&str, String)> = vec![ + ("UserId", self.config.user_id.clone()), + ("SortBy", "SortName".into()), + ("SortOrder", "Ascending".into()), + ]; + + if let Some(ct) = content_type { + params.push(("IncludeItemTypes", jellyfin_item_type(ct).into())); + } + + let response = self + .client + .get(&url) + .header("X-Emby-Token", &self.config.api_key) + .query(¶ms) + .send() + .await + .map_err(|e| { + DomainError::InfrastructureError(format!("Jellyfin request failed: {e}")) + })?; + + if !response.status().is_success() { + return Err(DomainError::InfrastructureError(format!( + "Jellyfin returned HTTP {}", + response.status() + ))); + } + + let body: JellyfinItemsResponse = response.json().await.map_err(|e| { + DomainError::InfrastructureError(format!("Failed to parse Jellyfin response: {e}")) + })?; + + Ok(body.items.into_iter().map(|item| item.name).collect()) + } + + async fn get_stream_url( + &self, + item_id: &MediaItemId, + quality: &StreamQuality, + ) -> DomainResult { + match quality { + StreamQuality::Direct => { + let url = format!( + "{}/Items/{}/PlaybackInfo", + self.config.base_url, + item_id.as_ref() + ); + let resp = self + .client + .post(&url) + .header("X-Emby-Token", &self.config.api_key) + .query(&[ + ("userId", &self.config.user_id), + ("mediaSourceId", &item_id.as_ref().to_string()), + ]) + .json(&serde_json::json!({})) + .send() + .await + .map_err(|e| { + DomainError::InfrastructureError(format!("PlaybackInfo failed: {e}")) + })?; + + if resp.status().is_success() { + let info: JellyfinPlaybackInfoResponse = resp.json().await.map_err(|e| { + DomainError::InfrastructureError(format!( + "PlaybackInfo parse failed: {e}" + )) + })?; + if let Some(src) = info.media_sources.first() + && src.supports_direct_stream + && let Some(rel_url) = &src.direct_stream_url + { + return Ok(format!( + "{}{}&api_key={}", + self.config.base_url, rel_url, self.config.api_key + )); + } + } + // Fallback: HLS at 8 Mbps + Ok(self.hls_url(item_id, 8_000_000)) + } + StreamQuality::Transcode(bps) => Ok(self.hls_url(item_id, *bps)), + } + } +}