GKaszewski/k-tv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(stream): add stream quality selection and update stream URL handling

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-03-14 04:03:54 +01:00
parent 8f42164bce
commit cf92cc49c2
11 changed files with 346 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
k-tv-backend/api/src/main.rs
View File

@@ -232,6 +232,7 @@ impl IMediaProvider for NoopMediaProvider {
async fn get_stream_url(
&self,
_: &domain::MediaItemId,
_: &domain::StreamQuality,
) -> domain::DomainResult<String> {
Err(domain::DomainError::InfrastructureError(
"No media provider configured.".into(),

15
k-tv-backend/api/src/routes/channels/broadcast.rs
View File

@@ -8,7 +8,7 @@ use chrono::Utc;
use serde::Deserialize;
use uuid::Uuid;
use domain::{DomainError, ScheduleEngineService};
use domain::{DomainError, ScheduleEngineService, StreamQuality};
use crate::{
dto::{CurrentBroadcastResponse, ScheduledSlotResponse},
@@ -130,11 +130,18 @@ pub(super) async fn get_epg(
/// Redirect to the stream URL for whatever is currently playing.
/// Returns 307 Temporary Redirect so the client fetches from the media provider directly.
/// Returns 204 No Content when the channel is in a gap (no-signal).
#[derive(Debug, Deserialize)]
pub(super) struct StreamQuery {
/// "direct" | bitrate in bps as string (e.g. "8000000"). Defaults to "direct".
quality: Option<String>,
}
pub(super) async fn get_stream(
State(state): State<AppState>,
Path(channel_id): Path<Uuid>,
OptionalCurrentUser(user): OptionalCurrentUser,
headers: HeaderMap,
Query(query): Query<StreamQuery>,
) -> Result<Response, ApiError> {
let channel = state.channel_service.find_by_id(channel_id).await?;
@@ -174,9 +181,13 @@ pub(super) async fn get_stream(
)?;
}
let stream_quality = match query.quality.as_deref() {
Some("direct") | None => StreamQuality::Direct,
Some(bps_str) => StreamQuality::Transcode(bps_str.parse::<u32>().unwrap_or(40_000_000)),
};
let url = state
.schedule_engine
.get_stream_url(&broadcast.slot.item.id)
.get_stream_url(&broadcast.slot.item.id, &stream_quality)
.await?;
Ok(Redirect::temporary(&url).into_response())

2
k-tv-backend/domain/src/lib.rs
View File

@@ -14,7 +14,7 @@ pub mod value_objects;
// Re-export commonly used types
pub use entities::*;
pub use errors::{DomainError, DomainResult};
pub use ports::{Collection, IMediaProvider, ProviderCapabilities, SeriesSummary, StreamingProtocol};
pub use ports::{Collection, IMediaProvider, ProviderCapabilities, SeriesSummary, StreamingProtocol, StreamQuality};
pub use repositories::*;
pub use iptv::{generate_m3u, generate_xmltv};
pub use services::{ChannelService, ScheduleEngineService, UserService};

15
k-tv-backend/domain/src/ports.rs
View File

@@ -12,6 +12,19 @@ use crate::entities::MediaItem;
use crate::errors::{DomainError, DomainResult};
use crate::value_objects::{ContentType, MediaFilter, MediaItemId};
// ============================================================================
// Stream quality
// ============================================================================
/// Requested stream quality for `get_stream_url`.
#[derive(Debug, Clone)]
pub enum StreamQuality {
/// Try direct stream via PlaybackInfo; fall back to HLS at 8 Mbps.
Direct,
/// Force HLS transcode at this bitrate (bits per second).
Transcode(u32),
}
// ============================================================================
// Provider capabilities
// ============================================================================
@@ -113,7 +126,7 @@ pub trait IMediaProvider: Send + Sync {
///
/// URLs are intentionally *not* stored in the schedule because they may be
/// short-lived (signed URLs, session tokens) or depend on client context.
async fn get_stream_url(&self, item_id: &MediaItemId) -> DomainResult<String>;
async fn get_stream_url(&self, item_id: &MediaItemId, quality: &StreamQuality) -> DomainResult<String>;
/// List top-level collections (libraries/sections) available in this provider.
///

6
k-tv-backend/domain/src/services/schedule/mod.rs
View File

@@ -10,7 +10,7 @@ use crate::entities::{
ScheduledSlot,
};
use crate::errors::{DomainError, DomainResult};
use crate::ports::IMediaProvider;
use crate::ports::{IMediaProvider, StreamQuality};
use crate::repositories::{ChannelRepository, ScheduleRepository};
use crate::value_objects::{
BlockId, ChannelId, FillStrategy, MediaFilter, MediaItemId, RecyclePolicy,
@@ -224,8 +224,8 @@ impl ScheduleEngineService {
}
/// Delegate stream URL resolution to the configured media provider.
pub async fn get_stream_url(&self, item_id: &MediaItemId) -> DomainResult<String> {
self.media_provider.get_stream_url(item_id).await
pub async fn get_stream_url(&self, item_id: &MediaItemId, quality: &StreamQuality) -> DomainResult<String> {
self.media_provider.get_stream_url(item_id, quality).await
}
/// Return all slots that overlap the given time window — the EPG data.

14
k-tv-backend/infra/src/jellyfin/models.rs
View File

@@ -47,6 +47,20 @@ pub(super) struct JellyfinItem {
pub recursive_item_count: Option<u32>,
}
#[derive(Debug, Deserialize)]
pub(super) struct JellyfinPlaybackInfoResponse {
#[serde(rename = "MediaSources")]
pub media_sources: Vec<JellyfinMediaSource>,
}
#[derive(Debug, Deserialize)]
pub(super) struct JellyfinMediaSource {
#[serde(rename = "SupportsDirectStream")]
pub supports_direct_stream: bool,
#[serde(rename = "DirectStreamUrl")]
pub direct_stream_url: Option<String>,
}
pub(super) fn jellyfin_item_type(ct: &ContentType) -> &'static str {
match ct {
ContentType::Movie => "Movie",

63
k-tv-backend/infra/src/jellyfin/provider.rs
View File

@@ -2,12 +2,12 @@ use async_trait::async_trait;
use domain::{
Collection, ContentType, DomainError, DomainResult, IMediaProvider, MediaFilter, MediaItem,
MediaItemId, ProviderCapabilities, SeriesSummary, StreamingProtocol,
MediaItemId, ProviderCapabilities, SeriesSummary, StreamQuality, StreamingProtocol,
};
use super::config::JellyfinConfig;
use super::mapping::{map_jellyfin_item, TICKS_PER_SEC};
use super::models::{jellyfin_item_type, JellyfinItemsResponse};
use super::models::{jellyfin_item_type, JellyfinItemsResponse, JellyfinPlaybackInfoResponse};
pub struct JellyfinMediaProvider {
pub(super) client: reqwest::Client,
@@ -361,24 +361,45 @@ impl IMediaProvider for JellyfinMediaProvider {
Ok(body.items.into_iter().map(|item| item.name).collect())
}
/// Build an HLS stream URL for a Jellyfin item.
///
/// Returns a `master.m3u8` playlist URL. Jellyfin transcodes to H.264/AAC
/// segments on the fly. HLS is preferred over a single MP4 stream because
/// `StartTimeTicks` works reliably with HLS — each segment is independent,
/// so Jellyfin can begin the playlist at the correct broadcast offset
/// without needing to byte-range seek into an in-progress transcode.
///
/// The API key is embedded so the player needs no separate auth header.
/// The caller (stream proxy route) appends `StartTimeTicks` when there is
/// a non-zero broadcast offset.
async fn get_stream_url(&self, item_id: &MediaItemId) -> DomainResult<String> {
Ok(format!(
"{}/Videos/{}/master.m3u8?videoCodec=h264&audioCodec=aac&VideoBitRate=40000000&mediaSourceId={}&api_key={}",
self.config.base_url,
item_id.as_ref(),
item_id.as_ref(),
self.config.api_key,
))
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() {
if src.supports_direct_stream {
if 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)),
}
}
}
impl JellyfinMediaProvider {
fn hls_url(&self, item_id: &MediaItemId, bitrate: u32) -> String {
format!(
"{}/Videos/{}/master.m3u8?videoCodec=h264&audioCodec=aac&VideoBitRate={}&mediaSourceId={}&api_key={}",
self.config.base_url,
item_id.as_ref(),
bitrate,
item_id.as_ref(),
self.config.api_key,
)
}
}

4
k-tv-backend/infra/src/local_files/provider.rs
View File

@@ -3,7 +3,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use domain::{
Collection, ContentType, DomainError, DomainResult, IMediaProvider, MediaFilter, MediaItem,
MediaItemId, ProviderCapabilities, StreamingProtocol,
MediaItemId, ProviderCapabilities, StreamQuality, StreamingProtocol,
};
use super::config::LocalFilesConfig;
@@ -138,7 +138,7 @@ impl IMediaProvider for LocalFilesProvider {
.map(|item| to_media_item(item_id.clone(), &item)))
}
async fn get_stream_url(&self, item_id: &MediaItemId) -> DomainResult<String> {
async fn get_stream_url(&self, item_id: &MediaItemId, _quality: &StreamQuality) -> DomainResult<String> {
Ok(format!(
"{}/api/v1/files/stream/{}",
self.base_url,