application: schedule bounded context

Cargo.lock

@@ -16,6 +16,7 @@ name = "application"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "chrono",
  "domain",
  "tokio",
  "uuid",

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]

crates/application/src/lib.rs

@@ -1,2 +1,3 @@
 pub mod auth;
 pub mod channels;
+pub mod schedule;

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

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;

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<ScheduleEngineService>,
+    pub channel_query: Arc<dyn ChannelQuery>,
+    pub schedule_query: Arc<dyn ScheduleQuery>,
+    pub schedule_command: Arc<dyn ScheduleCommand>,
+    pub event_publisher: Arc<dyn EventPublisher>,
+}

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<GeneratedSchedule> {
+    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;

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<Option<GeneratedSchedule>> {
+    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;

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<Option<CurrentBroadcast>> {
+    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;

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<Vec<ScheduledSlot>> {
+    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;

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<String> {
+    let item_id = MediaItemId::new(&query.item_id);
+    deps.schedule_engine
+        .get_stream_url(&item_id, &StreamQuality::Direct)
+        .await
+}

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<Vec<GeneratedSchedule>> {
+    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;

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

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

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);
+}

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);
+}

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());
+}

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());
+}

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());
+}

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<Vec<MediaItem>> {
+        Ok(vec![])
+    }
+
+    async fn fetch_by_id(&self, _item_id: &MediaItemId) -> DomainResult<Option<MediaItem>> {
+        Ok(None)
+    }
+
+    async fn get_stream_url(
+        &self,
+        _item_id: &MediaItemId,
+        _quality: &StreamQuality,
+    ) -> DomainResult<String> {
+        Err(domain::DomainError::InfrastructureError(
+            "TestProviderRegistry does not support streaming".into(),
+        ))
+    }
+
+    fn provider_ids(&self) -> Vec<String> {
+        vec!["test".into()]
+    }
+
+    fn primary_id(&self) -> &str {
+        "test"
+    }
+
+    fn capabilities(&self, _provider_id: &str) -> Option<ProviderCapabilities> {
+        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<Vec<Collection>> {
+        Ok(vec![])
+    }
+
+    async fn list_series(
+        &self,
+        _provider_id: &str,
+        _collection_id: Option<&str>,
+    ) -> DomainResult<Vec<SeriesSummary>> {
+        Ok(vec![])
+    }
+
+    async fn list_genres(
+        &self,
+        _provider_id: &str,
+        _content_type: Option<&ContentType>,
+    ) -> DomainResult<Vec<String>> {
+        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<InMemoryChannelRepository>,
+    Arc<InMemoryScheduleRepository>,
+) {
+    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)
+}

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());
+}