adapter-jellyfin: media provider

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"

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"
 

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 }

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,
+}

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;

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<MediaItem> {
+    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
+    ))
+}

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<JellyfinItem>,
+}
+
+#[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<i64>,
+    #[serde(rename = "Overview")]
+    pub overview: Option<String>,
+    #[serde(rename = "Genres")]
+    pub genres: Option<Vec<String>>,
+    #[serde(rename = "ProductionYear")]
+    pub production_year: Option<u16>,
+    #[serde(rename = "Tags")]
+    pub tags: Option<Vec<String>>,
+    /// TV show name (episodes only).
+    #[serde(rename = "SeriesName")]
+    pub series_name: Option<String>,
+    /// Season number (episodes only).
+    #[serde(rename = "ParentIndexNumber")]
+    pub parent_index_number: Option<u32>,
+    /// Episode number within the season (episodes only).
+    #[serde(rename = "IndexNumber")]
+    pub index_number: Option<u32>,
+    /// Collection type for virtual library folders (e.g. "movies", "tvshows").
+    #[serde(rename = "CollectionType")]
+    pub collection_type: Option<String>,
+    /// Total number of child items (used for Series to count episodes).
+    #[serde(rename = "RecursiveItemCount")]
+    pub recursive_item_count: Option<u32>,
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct JellyfinPlaybackInfoResponse {
+    #[serde(rename = "MediaSources")]
+    pub media_sources: Vec<JellyfinMediaSource>,
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct JellyfinMediaSource {
+    #[serde(rename = "SupportsDirectStream")]
+    pub supports_direct_stream: bool,
+    #[serde(rename = "DirectStreamUrl")]
+    pub direct_stream_url: Option<String>,
+}
+
+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",
+    }
+}

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<Vec<MediaItem>> {
+        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(&params)
+            .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<MediaItem> = 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<Vec<MediaItem>> {
+        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<MediaItem>> = 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<Option<MediaItem>> {
+        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<Vec<Collection>> {
+        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<Vec<SeriesSummary>> {
+        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(&params)
+            .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<Vec<String>> {
+        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(&params)
+            .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<String> {
+        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)),
+        }
+    }
+}