diff --git a/Cargo.lock b/Cargo.lock index 69d09f4..3139eef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -16,6 +16,7 @@ name = "application" version = "0.1.0" dependencies = [ "async-trait", + "chrono", "domain", "tokio", "uuid", diff --git a/crates/application/Cargo.toml b/crates/application/Cargo.toml index 949ea14..40a20b0 100644 --- a/crates/application/Cargo.toml +++ b/crates/application/Cargo.toml @@ -6,6 +6,7 @@ edition = "2024" [dependencies] domain = { workspace = true } async-trait = { workspace = true } +chrono = { workspace = true } uuid = { workspace = true } [dev-dependencies] diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 24782a6..357745a 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -1,2 +1,3 @@ pub mod auth; pub mod channels; +pub mod schedule; diff --git a/crates/application/src/schedule/commands.rs b/crates/application/src/schedule/commands.rs new file mode 100644 index 0000000..e07d5c8 --- /dev/null +++ b/crates/application/src/schedule/commands.rs @@ -0,0 +1,12 @@ +use uuid::Uuid; + +/// Generate a new schedule for a channel. +pub struct GenerateScheduleCommand { + pub channel_id: Uuid, +} + +/// Delete all schedules with generation > target_generation for a channel. +pub struct DeleteSchedulesAfterCommand { + pub channel_id: Uuid, + pub target_generation: u32, +} diff --git a/crates/application/src/schedule/delete_after.rs b/crates/application/src/schedule/delete_after.rs new file mode 100644 index 0000000..8ac8bcc --- /dev/null +++ b/crates/application/src/schedule/delete_after.rs @@ -0,0 +1,20 @@ +use domain::value_objects::ChannelId; +use domain::DomainResult; + +use super::commands::DeleteSchedulesAfterCommand; +use super::deps::ScheduleDeps; + +/// Delete all schedules with generation > target_generation for a channel. +pub async fn execute( + deps: &ScheduleDeps, + cmd: DeleteSchedulesAfterCommand, +) -> DomainResult<()> { + let channel_id = ChannelId::from(cmd.channel_id); + deps.schedule_command + .delete_schedules_after(channel_id, cmd.target_generation) + .await +} + +#[cfg(test)] +#[path = "tests/delete_after.rs"] +mod tests; diff --git a/crates/application/src/schedule/deps.rs b/crates/application/src/schedule/deps.rs new file mode 100644 index 0000000..ebd61d0 --- /dev/null +++ b/crates/application/src/schedule/deps.rs @@ -0,0 +1,13 @@ +use std::sync::Arc; + +use domain::ports::{ChannelQuery, EventPublisher, ScheduleCommand, ScheduleQuery}; +use domain::ScheduleEngineService; + +/// Dependencies for schedule use cases. +pub struct ScheduleDeps { + pub schedule_engine: Arc, + pub channel_query: Arc, + pub schedule_query: Arc, + pub schedule_command: Arc, + pub event_publisher: Arc, +} diff --git a/crates/application/src/schedule/generate.rs b/crates/application/src/schedule/generate.rs new file mode 100644 index 0000000..8032c3f --- /dev/null +++ b/crates/application/src/schedule/generate.rs @@ -0,0 +1,38 @@ +use chrono::Utc; + +use domain::events::DomainEvent; +use domain::models::GeneratedSchedule; +use domain::value_objects::ChannelId; +use domain::DomainResult; + +use super::commands::GenerateScheduleCommand; +use super::deps::ScheduleDeps; + +/// Generate a new 7-day schedule for a channel. +/// +/// Delegates the heavy lifting to `ScheduleEngineService::generate_schedule`, +/// then publishes a `ScheduleGenerated` domain event. +pub async fn execute( + deps: &ScheduleDeps, + cmd: GenerateScheduleCommand, +) -> DomainResult { + let channel_id = ChannelId::from(cmd.channel_id); + + let schedule = deps + .schedule_engine + .generate_schedule(channel_id, Utc::now()) + .await?; + + deps.event_publisher + .publish(DomainEvent::ScheduleGenerated { + channel_id, + schedule_id: schedule.id(), + }) + .await?; + + Ok(schedule) +} + +#[cfg(test)] +#[path = "tests/generate.rs"] +mod tests; diff --git a/crates/application/src/schedule/get_active.rs b/crates/application/src/schedule/get_active.rs new file mode 100644 index 0000000..66c4eb6 --- /dev/null +++ b/crates/application/src/schedule/get_active.rs @@ -0,0 +1,23 @@ +use chrono::Utc; + +use domain::models::GeneratedSchedule; +use domain::value_objects::ChannelId; +use domain::DomainResult; + +use super::deps::ScheduleDeps; +use super::queries::GetActiveScheduleQuery; + +/// Fetch the schedule currently active at `now`. +/// +/// Returns `None` when no schedule covers the current time. +pub async fn execute( + deps: &ScheduleDeps, + query: GetActiveScheduleQuery, +) -> DomainResult> { + let channel_id = ChannelId::from(query.channel_id); + deps.schedule_query.find_active(channel_id, Utc::now()).await +} + +#[cfg(test)] +#[path = "tests/get_active.rs"] +mod tests; diff --git a/crates/application/src/schedule/get_current_broadcast.rs b/crates/application/src/schedule/get_current_broadcast.rs new file mode 100644 index 0000000..bad0acd --- /dev/null +++ b/crates/application/src/schedule/get_current_broadcast.rs @@ -0,0 +1,31 @@ +use chrono::Utc; + +use domain::models::CurrentBroadcast; +use domain::value_objects::ChannelId; +use domain::{DomainResult, ScheduleEngineService}; + +use super::deps::ScheduleDeps; +use super::queries::GetCurrentBroadcastQuery; + +/// Determine what is currently broadcasting on a channel. +/// +/// Returns `None` when no schedule is active or `now` falls in a gap +/// between blocks (no-signal / static screen). +pub async fn execute( + deps: &ScheduleDeps, + query: GetCurrentBroadcastQuery, +) -> DomainResult> { + let channel_id = ChannelId::from(query.channel_id); + let now = Utc::now(); + + let schedule = match deps.schedule_query.find_active(channel_id, now).await? { + Some(s) => s, + None => return Ok(None), + }; + + Ok(ScheduleEngineService::get_current_broadcast(&schedule, now)) +} + +#[cfg(test)] +#[path = "tests/get_current_broadcast.rs"] +mod tests; diff --git a/crates/application/src/schedule/get_epg.rs b/crates/application/src/schedule/get_epg.rs new file mode 100644 index 0000000..2e6ae20 --- /dev/null +++ b/crates/application/src/schedule/get_epg.rs @@ -0,0 +1,37 @@ +use chrono::Utc; + +use domain::models::ScheduledSlot; +use domain::value_objects::ChannelId; +use domain::{DomainResult, ScheduleEngineService}; + +use super::deps::ScheduleDeps; +use super::queries::GetEpgQuery; + +/// Return EPG (electronic program guide) data for a channel. +/// +/// Returns the slots that overlap the active schedule's validity window. +/// Returns an empty vec when no schedule is active. +pub async fn execute( + deps: &ScheduleDeps, + query: GetEpgQuery, +) -> DomainResult> { + let channel_id = ChannelId::from(query.channel_id); + let now = Utc::now(); + + let schedule = match deps.schedule_query.find_active(channel_id, now).await? { + Some(s) => s, + None => return Ok(vec![]), + }; + + let slots = ScheduleEngineService::get_epg( + &schedule, + schedule.valid_from(), + schedule.valid_until(), + ); + + Ok(slots.into_iter().cloned().collect()) +} + +#[cfg(test)] +#[path = "tests/get_epg.rs"] +mod tests; diff --git a/crates/application/src/schedule/get_stream_url.rs b/crates/application/src/schedule/get_stream_url.rs new file mode 100644 index 0000000..fbdd27e --- /dev/null +++ b/crates/application/src/schedule/get_stream_url.rs @@ -0,0 +1,16 @@ +use domain::ports::StreamQuality; +use domain::value_objects::MediaItemId; +use domain::DomainResult; + +use super::deps::ScheduleDeps; +use super::queries::GetStreamUrlQuery; + +/// Resolve a playback URL for a media item. +/// +/// Delegates to the schedule engine which routes via the provider registry. +pub async fn execute(deps: &ScheduleDeps, query: GetStreamUrlQuery) -> DomainResult { + let item_id = MediaItemId::new(&query.item_id); + deps.schedule_engine + .get_stream_url(&item_id, &StreamQuality::Direct) + .await +} diff --git a/crates/application/src/schedule/list_history.rs b/crates/application/src/schedule/list_history.rs new file mode 100644 index 0000000..db511bc --- /dev/null +++ b/crates/application/src/schedule/list_history.rs @@ -0,0 +1,21 @@ +use domain::models::GeneratedSchedule; +use domain::value_objects::ChannelId; +use domain::DomainResult; + +use super::deps::ScheduleDeps; +use super::queries::ListHistoryQuery; + +/// List all generated schedule headers for a channel, newest first. +pub async fn execute( + deps: &ScheduleDeps, + query: ListHistoryQuery, +) -> DomainResult> { + let channel_id = ChannelId::from(query.channel_id); + deps.schedule_query + .list_schedule_history(channel_id) + .await +} + +#[cfg(test)] +#[path = "tests/list_history.rs"] +mod tests; diff --git a/crates/application/src/schedule/mod.rs b/crates/application/src/schedule/mod.rs new file mode 100644 index 0000000..49753d3 --- /dev/null +++ b/crates/application/src/schedule/mod.rs @@ -0,0 +1,17 @@ +pub mod commands; +pub mod delete_after; +pub mod deps; +pub mod generate; +pub mod get_active; +pub mod get_current_broadcast; +pub mod get_epg; +pub mod get_stream_url; +pub mod list_history; +pub mod queries; + +pub use commands::{DeleteSchedulesAfterCommand, GenerateScheduleCommand}; +pub use deps::ScheduleDeps; +pub use queries::{ + GetActiveScheduleQuery, GetCurrentBroadcastQuery, GetEpgQuery, GetStreamUrlQuery, + ListHistoryQuery, +}; diff --git a/crates/application/src/schedule/queries.rs b/crates/application/src/schedule/queries.rs new file mode 100644 index 0000000..cf0fd96 --- /dev/null +++ b/crates/application/src/schedule/queries.rs @@ -0,0 +1,28 @@ +use uuid::Uuid; + +/// Fetch the schedule whose validity window contains `now`. +pub struct GetActiveScheduleQuery { + pub channel_id: Uuid, +} + +/// Determine what is currently broadcasting on a channel. +pub struct GetCurrentBroadcastQuery { + pub channel_id: Uuid, +} + +/// Return EPG (electronic program guide) data for a channel. +pub struct GetEpgQuery { + pub channel_id: Uuid, +} + +/// Get a playback URL for a specific media item on a channel. +pub struct GetStreamUrlQuery { + pub channel_id: Uuid, + /// MediaItemId as string (e.g. "jellyfin::abc123"). + pub item_id: String, +} + +/// List all generated schedule headers for a channel. +pub struct ListHistoryQuery { + pub channel_id: Uuid, +} diff --git a/crates/application/src/schedule/tests/delete_after.rs b/crates/application/src/schedule/tests/delete_after.rs new file mode 100644 index 0000000..140d7ed --- /dev/null +++ b/crates/application/src/schedule/tests/delete_after.rs @@ -0,0 +1,64 @@ +use domain::models::{Channel, GeneratedSchedule}; +use domain::value_objects::UserId; + +use crate::schedule::commands::DeleteSchedulesAfterCommand; +use crate::schedule::delete_after; +use crate::schedule::queries::ListHistoryQuery; +use crate::schedule::list_history; + +#[path = "helpers.rs"] +mod helpers; +use helpers::make_schedule_deps; + +#[tokio::test] +async fn delete_after_removes_later_generations() { + let (deps, channel_repo, schedule_repo) = make_schedule_deps(); + + let channel = Channel::new(UserId::generate(), "Cleanup", "UTC"); + let channel_id = channel.id(); + channel_repo + .channels + .lock() + .unwrap() + .insert(channel_id.value(), channel); + + // Manually insert schedules with different generations. + let now = chrono::Utc::now(); + for generation in 1..=3 { + let sched = GeneratedSchedule::new( + channel_id, + now, + now + chrono::Duration::hours(24), + generation, + vec![], + ); + schedule_repo + .schedules + .lock() + .unwrap() + .insert(sched.id().value(), sched); + } + + // Delete generations > 1. + delete_after::execute( + &deps, + DeleteSchedulesAfterCommand { + channel_id: channel_id.value(), + target_generation: 1, + }, + ) + .await + .unwrap(); + + let remaining = list_history::execute( + &deps, + ListHistoryQuery { + channel_id: channel_id.value(), + }, + ) + .await + .unwrap(); + + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].generation(), 1); +} diff --git a/crates/application/src/schedule/tests/generate.rs b/crates/application/src/schedule/tests/generate.rs new file mode 100644 index 0000000..0938eca --- /dev/null +++ b/crates/application/src/schedule/tests/generate.rs @@ -0,0 +1,82 @@ +use domain::models::Channel; +use domain::value_objects::UserId; + +use crate::schedule::commands::GenerateScheduleCommand; +use crate::schedule::generate; + +#[path = "helpers.rs"] +mod helpers; +use helpers::make_schedule_deps; + +#[tokio::test] +async fn generate_produces_empty_schedule_for_channel_with_no_blocks() { + let (deps, channel_repo, _) = make_schedule_deps(); + + // Create a channel with no programming blocks. + let channel = Channel::new(UserId::generate(), "Empty Channel", "UTC"); + channel_repo + .channels + .lock() + .unwrap() + .insert(channel.id().value(), channel.clone()); + + let schedule = generate::execute( + &deps, + GenerateScheduleCommand { + channel_id: channel.id().value(), + }, + ) + .await + .unwrap(); + + assert_eq!(schedule.channel_id(), channel.id()); + assert_eq!(schedule.generation(), 1); + assert!(schedule.slots().is_empty()); +} + +#[tokio::test] +async fn generate_fails_for_nonexistent_channel() { + let (deps, _, _) = make_schedule_deps(); + + let result = generate::execute( + &deps, + GenerateScheduleCommand { + channel_id: uuid::Uuid::new_v4(), + }, + ) + .await; + + assert!(result.is_err()); +} + +#[tokio::test] +async fn generate_increments_generation() { + let (deps, channel_repo, _) = make_schedule_deps(); + + let channel = Channel::new(UserId::generate(), "Gen Test", "UTC"); + channel_repo + .channels + .lock() + .unwrap() + .insert(channel.id().value(), channel.clone()); + + let first = generate::execute( + &deps, + GenerateScheduleCommand { + channel_id: channel.id().value(), + }, + ) + .await + .unwrap(); + assert_eq!(first.generation(), 1); + + let second = generate::execute( + &deps, + GenerateScheduleCommand { + channel_id: channel.id().value(), + }, + ) + .await + .unwrap(); + assert_eq!(second.generation(), 2); +} diff --git a/crates/application/src/schedule/tests/get_active.rs b/crates/application/src/schedule/tests/get_active.rs new file mode 100644 index 0000000..85752bd --- /dev/null +++ b/crates/application/src/schedule/tests/get_active.rs @@ -0,0 +1,22 @@ +use crate::schedule::get_active; +use crate::schedule::queries::GetActiveScheduleQuery; + +#[path = "helpers.rs"] +mod helpers; +use helpers::make_schedule_deps; + +#[tokio::test] +async fn returns_none_when_no_schedule_exists() { + let (deps, _, _) = make_schedule_deps(); + + let result = get_active::execute( + &deps, + GetActiveScheduleQuery { + channel_id: uuid::Uuid::new_v4(), + }, + ) + .await + .unwrap(); + + assert!(result.is_none()); +} diff --git a/crates/application/src/schedule/tests/get_current_broadcast.rs b/crates/application/src/schedule/tests/get_current_broadcast.rs new file mode 100644 index 0000000..230ef1e --- /dev/null +++ b/crates/application/src/schedule/tests/get_current_broadcast.rs @@ -0,0 +1,22 @@ +use crate::schedule::get_current_broadcast; +use crate::schedule::queries::GetCurrentBroadcastQuery; + +#[path = "helpers.rs"] +mod helpers; +use helpers::make_schedule_deps; + +#[tokio::test] +async fn returns_none_when_no_schedule_active() { + let (deps, _, _) = make_schedule_deps(); + + let result = get_current_broadcast::execute( + &deps, + GetCurrentBroadcastQuery { + channel_id: uuid::Uuid::new_v4(), + }, + ) + .await + .unwrap(); + + assert!(result.is_none()); +} diff --git a/crates/application/src/schedule/tests/get_epg.rs b/crates/application/src/schedule/tests/get_epg.rs new file mode 100644 index 0000000..ce94b4b --- /dev/null +++ b/crates/application/src/schedule/tests/get_epg.rs @@ -0,0 +1,22 @@ +use crate::schedule::get_epg; +use crate::schedule::queries::GetEpgQuery; + +#[path = "helpers.rs"] +mod helpers; +use helpers::make_schedule_deps; + +#[tokio::test] +async fn returns_empty_when_no_schedule_active() { + let (deps, _, _) = make_schedule_deps(); + + let result = get_epg::execute( + &deps, + GetEpgQuery { + channel_id: uuid::Uuid::new_v4(), + }, + ) + .await + .unwrap(); + + assert!(result.is_empty()); +} diff --git a/crates/application/src/schedule/tests/helpers.rs b/crates/application/src/schedule/tests/helpers.rs new file mode 100644 index 0000000..7c63af8 --- /dev/null +++ b/crates/application/src/schedule/tests/helpers.rs @@ -0,0 +1,115 @@ +use std::sync::Arc; + +use async_trait::async_trait; + +use domain::errors::DomainResult; +use domain::models::MediaItem; +use domain::ports::{ + Collection, IProviderRegistry, ProviderCapabilities, + SeriesSummary, StreamQuality, StreamingProtocol, +}; +use domain::testing::{InMemoryChannelRepository, InMemoryScheduleRepository, NoopEventPublisher}; +use domain::value_objects::{ContentType, MediaFilter, MediaItemId}; +use domain::ScheduleEngineService; + +use crate::schedule::deps::ScheduleDeps; + +/// Minimal IProviderRegistry backed by a NoopMediaProvider. +pub(crate) struct TestProviderRegistry; + +#[async_trait] +impl IProviderRegistry for TestProviderRegistry { + async fn fetch_items( + &self, + _provider_id: &str, + _filter: &MediaFilter, + ) -> DomainResult> { + Ok(vec![]) + } + + async fn fetch_by_id(&self, _item_id: &MediaItemId) -> DomainResult> { + Ok(None) + } + + async fn get_stream_url( + &self, + _item_id: &MediaItemId, + _quality: &StreamQuality, + ) -> DomainResult { + Err(domain::DomainError::InfrastructureError( + "TestProviderRegistry does not support streaming".into(), + )) + } + + fn provider_ids(&self) -> Vec { + vec!["test".into()] + } + + fn primary_id(&self) -> &str { + "test" + } + + fn capabilities(&self, _provider_id: &str) -> Option { + Some(ProviderCapabilities { + collections: false, + series: false, + genres: false, + tags: false, + decade: false, + search: false, + streaming_protocol: StreamingProtocol::Hls, + rescan: false, + transcode: false, + }) + } + + async fn list_collections(&self, _provider_id: &str) -> DomainResult> { + Ok(vec![]) + } + + async fn list_series( + &self, + _provider_id: &str, + _collection_id: Option<&str>, + ) -> DomainResult> { + Ok(vec![]) + } + + async fn list_genres( + &self, + _provider_id: &str, + _content_type: Option<&ContentType>, + ) -> DomainResult> { + Ok(vec![]) + } +} + +/// Build ScheduleDeps backed by InMemory repos and a test provider registry. +/// +/// Returns the deps plus the underlying repos for test assertions. +pub(crate) fn make_schedule_deps() -> ( + ScheduleDeps, + Arc, + Arc, +) { + let channel_repo = Arc::new(InMemoryChannelRepository::new()); + let schedule_repo = Arc::new(InMemoryScheduleRepository::new()); + let provider_registry = Arc::new(TestProviderRegistry); + + let engine = Arc::new(ScheduleEngineService::new( + provider_registry, + channel_repo.clone(), + schedule_repo.clone(), + schedule_repo.clone(), + )); + + let deps = ScheduleDeps { + schedule_engine: engine, + channel_query: channel_repo.clone(), + schedule_query: schedule_repo.clone(), + schedule_command: schedule_repo.clone(), + event_publisher: Arc::new(NoopEventPublisher::new()), + }; + + (deps, channel_repo, schedule_repo) +} diff --git a/crates/application/src/schedule/tests/list_history.rs b/crates/application/src/schedule/tests/list_history.rs new file mode 100644 index 0000000..23d6097 --- /dev/null +++ b/crates/application/src/schedule/tests/list_history.rs @@ -0,0 +1,22 @@ +use crate::schedule::list_history; +use crate::schedule::queries::ListHistoryQuery; + +#[path = "helpers.rs"] +mod helpers; +use helpers::make_schedule_deps; + +#[tokio::test] +async fn returns_empty_for_new_channel() { + let (deps, _, _) = make_schedule_deps(); + + let result = list_history::execute( + &deps, + ListHistoryQuery { + channel_id: uuid::Uuid::new_v4(), + }, + ) + .await + .unwrap(); + + assert!(result.is_empty()); +}