From 8e4f72456287209dfede6297f63cd1a18b0712e9 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 14:02:45 +0200 Subject: [PATCH 1/4] add integration tests for all 6 FillStrategy variants, role filter on LibrarySearchFilter --- crates/domain/src/services/schedule/mod.rs | 4 + .../services/schedule/tests/integration.rs | 278 ++++++++++++++++++ crates/domain/src/testing/in_memory.rs | 11 + crates/domain/src/value_objects/search.rs | 11 +- 4 files changed, 303 insertions(+), 1 deletion(-) create mode 100644 crates/domain/src/services/schedule/tests/integration.rs diff --git a/crates/domain/src/services/schedule/mod.rs b/crates/domain/src/services/schedule/mod.rs index 7597019..f5b4523 100644 --- a/crates/domain/src/services/schedule/mod.rs +++ b/crates/domain/src/services/schedule/mod.rs @@ -378,3 +378,7 @@ fn media_filter_to_library_search(filter: &crate::value_objects::MediaFilter) -> } lsf } + +#[cfg(all(test, feature = "test-helpers"))] +#[path = "tests/integration.rs"] +mod integration_tests; diff --git a/crates/domain/src/services/schedule/tests/integration.rs b/crates/domain/src/services/schedule/tests/integration.rs new file mode 100644 index 0000000..7c74214 --- /dev/null +++ b/crates/domain/src/services/schedule/tests/integration.rs @@ -0,0 +1,278 @@ +use std::sync::Arc; + +use chrono::{Datelike, NaiveTime, Utc}; + +use crate::models::{Channel, MediaItem, ProgrammingBlock, ScheduleConfig}; +use crate::services::schedule::ScheduleEngineService; +use crate::testing::{ + InMemoryChannelRepository, InMemoryLibraryRepository, InMemoryScheduleRepository, +}; +use crate::value_objects::{ + ContentType, FillStrategy, MediaFilter, MediaItemId, Weekday, +}; + +fn episode(id: &str, series: &str, ep: u32, secs: u32) -> MediaItem { + let item = MediaItem::new(MediaItemId::new(id), id, ContentType::Episode, secs); + let mut val = serde_json::to_value(&item).unwrap(); + val["series_name"] = serde_json::Value::String(series.into()); + val["episode_number"] = serde_json::Value::Number(ep.into()); + serde_json::from_value(val).unwrap() +} + +fn movie(id: &str, secs: u32) -> MediaItem { + MediaItem::new(MediaItemId::new(id), id, ContentType::Movie, secs) +} + +struct TestHarness { + engine: ScheduleEngineService, + channel_repo: Arc, + library_repo: Arc, +} + +impl TestHarness { + fn new() -> Self { + let library_repo = Arc::new(InMemoryLibraryRepository::new()); + let channel_repo = Arc::new(InMemoryChannelRepository::new()); + let schedule_repo = Arc::new(InMemoryScheduleRepository::new()); + + let engine = ScheduleEngineService::new( + library_repo.clone(), + channel_repo.clone(), + schedule_repo.clone(), + schedule_repo, + ); + + Self { + engine, + channel_repo, + library_repo, + } + } + + fn seed_items(&self, items: Vec) { + let mut store = self.library_repo.items.lock().unwrap(); + for item in items { + store.insert(item.id().value().to_string(), item); + } + } + + async fn create_channel_with_block( + &self, + block: ProgrammingBlock, + ) -> Channel { + let mut channel = Channel::new( + crate::value_objects::UserId::generate(), + "test-channel", + "UTC", + ); + let today = Utc::now() + .with_timezone(&chrono_tz::UTC) + .date_naive(); + let weekday = Weekday::from(today.weekday()); + let mut config = ScheduleConfig::new(); + config.insert_day(weekday, vec![block]); + channel.set_schedule_config(config); + self.channel_repo + .channels + .lock() + .unwrap() + .insert(channel.id(), channel.clone()); + channel + } +} + +#[tokio::test] +async fn alternating_interleaves_two_series() { + let h = TestHarness::new(); + h.seed_items(vec![ + episode("a-e1", "Show A", 1, 300), + episode("a-e2", "Show A", 2, 300), + episode("a-e3", "Show A", 3, 300), + episode("b-e1", "Show B", 1, 300), + episode("b-e2", "Show B", 2, 300), + episode("b-e3", "Show B", 3, 300), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "alternating-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::Alternating, + ); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let slots = schedule.slots(); + assert!(slots.len() >= 4, "expected at least 4 slots, got {}", slots.len()); + + let series: Vec> = slots.iter().map(|s| s.item().series_name()).collect(); + for pair in series.windows(2) { + if pair[0] == pair[1] { + panic!( + "consecutive slots have same series {:?}, expected interleaving", + pair[0] + ); + } + } +} + +#[tokio::test] +async fn weighted_surfaces_fresh_items() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("m1", 600), + movie("m2", 600), + movie("m3", 600), + movie("m4", 600), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "weighted-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::Weighted, + ); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + assert!(!schedule.slots().is_empty(), "weighted strategy produced no slots"); + + let total_duration: u32 = schedule + .slots() + .iter() + .map(|s| (s.end_at() - s.start_at()).num_seconds() as u32) + .sum(); + assert!(total_duration > 0, "schedule has zero total duration"); +} + +#[tokio::test] +async fn marathon_fills_from_episode_one() { + let h = TestHarness::new(); + h.seed_items(vec![ + episode("ep1", "Series", 1, 600), + episode("ep2", "Series", 2, 600), + episode("ep3", "Series", 3, 600), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "marathon-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::Marathon, + ); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let slots = schedule.slots(); + assert!(slots.len() >= 3, "expected at least 3 slots, got {}", slots.len()); + + assert_eq!(slots[0].item().id().value(), "ep1"); + assert_eq!(slots[1].item().id().value(), "ep2"); + assert_eq!(slots[2].item().id().value(), "ep3"); +} + +#[tokio::test] +async fn sequential_produces_ordered_schedule() { + let h = TestHarness::new(); + h.seed_items(vec![ + episode("ep1", "Series", 1, 600), + episode("ep2", "Series", 2, 600), + episode("ep3", "Series", 3, 600), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "sequential-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::Sequential, + ); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let slots = schedule.slots(); + assert!(!slots.is_empty(), "sequential strategy produced no slots"); + assert_eq!(slots[0].item().id().value(), "ep1"); +} + +#[tokio::test] +async fn best_fit_produces_schedule() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("m1", 1800), + movie("m2", 1200), + movie("m3", 900), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "bestfit-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::BestFit, + ); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + assert!(!schedule.slots().is_empty(), "best_fit strategy produced no slots"); + assert_eq!( + schedule.slots()[0].item().id().value(), + "m1", + "best_fit should pick longest item first" + ); +} + +#[tokio::test] +async fn random_produces_schedule_within_budget() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("m1", 600), + movie("m2", 600), + movie("m3", 600), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "random-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::Random, + ); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + assert!(!schedule.slots().is_empty(), "random strategy produced no slots"); +} diff --git a/crates/domain/src/testing/in_memory.rs b/crates/domain/src/testing/in_memory.rs index a4bbe1c..1342e1a 100644 --- a/crates/domain/src/testing/in_memory.rs +++ b/crates/domain/src/testing/in_memory.rs @@ -467,10 +467,21 @@ impl LibraryQuery for InMemoryLibraryRepository { { return false; } + if let Some(role) = filter.role() && item.role() != role { + return false; + } + if !filter.series_names().is_empty() + && !item + .series_name() + .is_some_and(|sn| filter.series_names().iter().any(|f| f == sn)) + { + return false; + } true }) .cloned() .collect(); + items.sort_by(|a, b| a.id().value().cmp(b.id().value())); let total = items.len() as u32; let offset = filter.offset() as usize; let limit = filter.limit() as usize; diff --git a/crates/domain/src/value_objects/search.rs b/crates/domain/src/value_objects/search.rs index 7df866f..f517264 100644 --- a/crates/domain/src/value_objects/search.rs +++ b/crates/domain/src/value_objects/search.rs @@ -1,4 +1,4 @@ -use crate::value_objects::ContentType; +use crate::value_objects::{ContentType, MediaRole}; const DEFAULT_SEARCH_LIMIT: u32 = 50; @@ -14,6 +14,7 @@ pub struct LibrarySearchFilter { max_duration_secs: Option, search_term: Option, season_number: Option, + role: Option, offset: u32, limit: u32, } @@ -63,6 +64,10 @@ impl LibrarySearchFilter { self.season_number = Some(n); self } + pub fn with_role(mut self, role: MediaRole) -> Self { + self.role = Some(role); + self + } pub fn with_offset(mut self, offset: u32) -> Self { self.offset = offset; self @@ -102,6 +107,9 @@ impl LibrarySearchFilter { pub fn season_number(&self) -> Option { self.season_number } + pub fn role(&self) -> Option<&MediaRole> { + self.role.as_ref() + } pub fn offset(&self) -> u32 { self.offset } @@ -123,6 +131,7 @@ impl Default for LibrarySearchFilter { max_duration_secs: None, search_term: None, season_number: None, + role: None, offset: 0, limit: DEFAULT_SEARCH_LIMIT, } From 8edd7a9c0b5f0bcfbfea37c74746a304ba2c9d05 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 14:05:33 +0200 Subject: [PATCH 2/4] wire interstitial insertion into schedule engine resolve_block --- crates/domain/src/models/channel.rs | 10 ++ crates/domain/src/services/schedule/mod.rs | 84 +++++++++- .../services/schedule/tests/integration.rs | 146 +++++++++++++++++- 3 files changed, 234 insertions(+), 6 deletions(-) diff --git a/crates/domain/src/models/channel.rs b/crates/domain/src/models/channel.rs index c12ba71..7fd5ccc 100644 --- a/crates/domain/src/models/channel.rs +++ b/crates/domain/src/models/channel.rs @@ -426,6 +426,16 @@ impl ProgrammingBlock { pub fn mid_roll_rule(&self) -> Option<&MidRollRule> { self.mid_roll_rule.as_ref() } + + pub fn with_interstitial_rule(mut self, rule: InterstitialRule) -> Self { + self.interstitial_rule = Some(rule); + self + } + + pub fn with_mid_roll_rule(mut self, rule: MidRollRule) -> Self { + self.mid_roll_rule = Some(rule); + self + } } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/crates/domain/src/services/schedule/mod.rs b/crates/domain/src/services/schedule/mod.rs index f5b4523..f0fbdfa 100644 --- a/crates/domain/src/services/schedule/mod.rs +++ b/crates/domain/src/services/schedule/mod.rs @@ -9,7 +9,7 @@ use crate::models::{ ScheduledSlot, }; use crate::ports::{ChannelQuery, LibraryQuery, ScheduleCommand, ScheduleQuery}; -use crate::value_objects::{BlockId, ChannelId, FillStrategy, LibrarySearchFilter, MediaItemId, RotationPolicy, Weekday}; +use crate::value_objects::{BlockId, ChannelId, FillStrategy, InterstitialRule, LibrarySearchFilter, MediaItemId, MediaRole, RotationPolicy, Weekday}; mod fill; mod rotation; @@ -248,10 +248,10 @@ impl ScheduleEngineService { window: BlockTimeWindow, rotation: RotationContext<'_>, ) -> DomainResult> { - match block.content() { + let program_slots = match block.content() { BlockContent::Manual { items } => { self.resolve_manual(items, window.start, window.end, block.id()) - .await + .await? } BlockContent::Algorithmic { filter, @@ -265,12 +265,85 @@ impl ScheduleEngineService { loop_on_finish: block.loop_on_finish(), ignore_rotation_policy: block.ignore_rotation_policy(), }, - window, + BlockTimeWindow { + start: window.start, + end: window.end, + }, rotation, ) + .await? + } + }; + + if let Some(rule) = block.interstitial_rule() { + self.insert_interstitials(program_slots, rule, block.id(), window.end) .await + } else { + Ok(program_slots) + } + } + + async fn insert_interstitials( + &self, + program_slots: Vec, + rule: &InterstitialRule, + block_id: BlockId, + block_end: DateTime, + ) -> DomainResult> { + if program_slots.len() < 2 { + return Ok(program_slots); + } + + let mut filter = media_filter_to_library_search(rule.pool_filter()); + filter = filter.with_role(MediaRole::Interstitial); + let (interstitials, _) = self.library_query.search(&filter).await?; + + if interstitials.is_empty() { + return Ok(program_slots); + } + + let mut result: Vec = Vec::new(); + let mut cursor = program_slots[0].start_at(); + let mut interstitial_idx = 0; + + for (i, slot) in program_slots.iter().enumerate() { + if cursor >= block_end { + break; + } + + let program_duration = slot.item().duration_secs(); + let program_end = (cursor + Duration::seconds(program_duration as i64)).min(block_end); + result.push(ScheduledSlot::new( + cursor, + program_end, + slot.item().clone(), + block_id, + )); + cursor = program_end; + + let should_insert = i + 1 < program_slots.len() + && program_duration >= rule.min_gap_secs() + && cursor < block_end; + + if should_insert { + let interstitial = &interstitials[interstitial_idx % interstitials.len()]; + let interstitial_end = + (cursor + Duration::seconds(interstitial.duration_secs() as i64)) + .min(block_end); + if interstitial_end > cursor { + result.push(ScheduledSlot::new( + cursor, + interstitial_end, + interstitial.clone(), + block_id, + )); + cursor = interstitial_end; + interstitial_idx += 1; + } } } + + Ok(result) } async fn resolve_manual( @@ -304,7 +377,8 @@ impl ScheduleEngineService { window: BlockTimeWindow, rotation: RotationContext<'_>, ) -> DomainResult> { - let library_filter = media_filter_to_library_search(params.filter); + let library_filter = + media_filter_to_library_search(params.filter).with_role(MediaRole::Program); let (candidates, _total) = self.library_query.search(&library_filter).await?; if candidates.is_empty() { diff --git a/crates/domain/src/services/schedule/tests/integration.rs b/crates/domain/src/services/schedule/tests/integration.rs index 7c74214..ce833ab 100644 --- a/crates/domain/src/services/schedule/tests/integration.rs +++ b/crates/domain/src/services/schedule/tests/integration.rs @@ -8,7 +8,7 @@ use crate::testing::{ InMemoryChannelRepository, InMemoryLibraryRepository, InMemoryScheduleRepository, }; use crate::value_objects::{ - ContentType, FillStrategy, MediaFilter, MediaItemId, Weekday, + ContentType, FillStrategy, InterstitialRule, MediaFilter, MediaItemId, MediaRole, Weekday, }; fn episode(id: &str, series: &str, ep: u32, secs: u32) -> MediaItem { @@ -276,3 +276,147 @@ async fn random_produces_schedule_within_budget() { assert!(!schedule.slots().is_empty(), "random strategy produced no slots"); } + +fn interstitial(id: &str, secs: u32) -> MediaItem { + let item = MediaItem::new(MediaItemId::new(id), id, ContentType::Short, secs); + let mut val = serde_json::to_value(&item).unwrap(); + val["role"] = serde_json::Value::String("interstitial".into()); + serde_json::from_value(val).unwrap() +} + +#[tokio::test] +async fn interstitial_inserted_between_programs() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("prog1", 600), + movie("prog2", 600), + movie("prog3", 600), + interstitial("bump1", 30), + interstitial("bump2", 30), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "with-interstitials", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::BestFit, + ) + .with_interstitial_rule(InterstitialRule::new( + MediaFilter::default(), + FillStrategy::Sequential, + 0, + )); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let today_slots: Vec<_> = schedule.slots().iter().take(10).collect(); + + let program_count = today_slots + .iter() + .filter(|s| *s.item().role() == MediaRole::Program) + .count(); + let interstitial_count = today_slots + .iter() + .filter(|s| *s.item().role() == MediaRole::Interstitial) + .count(); + + assert!(program_count >= 3, "expected >= 3 programs, got {program_count}"); + assert!( + interstitial_count >= 2, + "expected >= 2 interstitials, got {interstitial_count}" + ); + + for pair in today_slots.windows(2) { + assert!( + pair[0].end_at() <= pair[1].start_at(), + "slots overlap: {} ends at {:?} but {} starts at {:?}", + pair[0].item().title(), + pair[0].end_at(), + pair[1].item().title(), + pair[1].start_at(), + ); + } +} + +#[tokio::test] +async fn interstitial_no_matching_items_still_generates() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("prog1", 600), + movie("prog2", 600), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "no-interstitials-available", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::BestFit, + ) + .with_interstitial_rule(InterstitialRule::new( + MediaFilter::default(), + FillStrategy::Sequential, + 0, + )); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + assert!(!schedule.slots().is_empty(), "should still produce program slots"); + assert!( + schedule.slots().iter().all(|s| *s.item().role() == MediaRole::Program), + "all slots should be programs when no interstitials available" + ); +} + +#[tokio::test] +async fn interstitial_min_gap_respected() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("short1", 100), + movie("short2", 100), + movie("short3", 100), + interstitial("bump1", 30), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "min-gap-test", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::BestFit, + ) + .with_interstitial_rule(InterstitialRule::new( + MediaFilter::default(), + FillStrategy::Sequential, + 200, + )); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let interstitial_count = schedule + .slots() + .iter() + .filter(|s| *s.item().role() == MediaRole::Interstitial) + .count(); + + assert_eq!( + interstitial_count, 0, + "no interstitials should be inserted when programs are shorter than min_gap_secs" + ); +} From 79e975f057f285dd5b388bfa32cc024cb582062a Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 14:08:55 +0200 Subject: [PATCH 3/4] wire gap_filler into schedule engine, API, MCP --- crates/api-types/src/channels.rs | 5 + crates/application/src/channels/commands.rs | 3 +- .../application/src/channels/tests/update.rs | 7 + crates/application/src/channels/update.rs | 3 + crates/domain/src/services/schedule/mod.rs | 75 +++++++++- .../services/schedule/tests/integration.rs | 135 ++++++++++++++++++ crates/mcp/src/server.rs | 15 +- crates/mcp/src/tools/channels.rs | 2 + crates/presentation/src/handlers/channels.rs | 1 + 9 files changed, 243 insertions(+), 3 deletions(-) diff --git a/crates/api-types/src/channels.rs b/crates/api-types/src/channels.rs index 28c27a1..722a5dd 100644 --- a/crates/api-types/src/channels.rs +++ b/crates/api-types/src/channels.rs @@ -35,6 +35,8 @@ pub struct UpdateChannelRequest { pub webhook_poll_interval_secs: Option, pub webhook_body_template: Option>, pub webhook_headers: Option>, + #[schema(value_type = Option>)] + pub gap_filler: Option>, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] @@ -55,6 +57,8 @@ pub struct ChannelResponse { pub webhook_poll_interval_secs: u32, pub webhook_body_template: Option, pub webhook_headers: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub gap_filler: Option, pub created_at: DateTime, pub updated_at: DateTime, } @@ -78,6 +82,7 @@ impl From for ChannelResponse { webhook_poll_interval_secs: c.webhook_poll_interval_secs(), webhook_body_template: c.webhook_body_template().map(|s| s.to_string()), webhook_headers: c.webhook_headers().map(|s| s.to_string()), + gap_filler: c.gap_filler().map(|f| serde_json::to_value(f).unwrap_or_default()), created_at: c.created_at(), updated_at: c.updated_at(), } diff --git a/crates/application/src/channels/commands.rs b/crates/application/src/channels/commands.rs index 9adcd3a..3429826 100644 --- a/crates/application/src/channels/commands.rs +++ b/crates/application/src/channels/commands.rs @@ -1,5 +1,5 @@ use domain::models::ScheduleConfig; -use domain::value_objects::{ChannelId, RotationPolicy, UserId}; +use domain::value_objects::{ChannelId, MediaFilter, RotationPolicy, UserId}; pub struct CreateChannelCommand { pub owner_id: UserId, @@ -16,6 +16,7 @@ pub struct UpdateChannelCommand { pub schedule_config: Option, pub rotation_policy: Option, pub auto_schedule: Option, + pub gap_filler: Option>, } pub struct DeleteChannelCommand { diff --git a/crates/application/src/channels/tests/update.rs b/crates/application/src/channels/tests/update.rs index 9d9041a..8c8606d 100644 --- a/crates/application/src/channels/tests/update.rs +++ b/crates/application/src/channels/tests/update.rs @@ -45,6 +45,7 @@ async fn updates_channel_name() { schedule_config: None, rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await @@ -82,6 +83,7 @@ async fn update_fails_if_not_owner() { schedule_config: None, rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await; @@ -108,6 +110,7 @@ async fn update_nonexistent_channel_returns_not_found() { schedule_config: None, rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await; @@ -148,6 +151,7 @@ async fn update_config_creates_snapshot() { schedule_config: Some(new_config), rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await @@ -187,6 +191,7 @@ async fn update_without_config_skips_snapshot() { schedule_config: None, rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await @@ -225,6 +230,7 @@ async fn update_description_clear() { schedule_config: None, rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await @@ -243,6 +249,7 @@ async fn update_description_clear() { schedule_config: None, rotation_policy: None, auto_schedule: None, + gap_filler: None, }, ) .await diff --git a/crates/application/src/channels/update.rs b/crates/application/src/channels/update.rs index 12b70d9..e0d23f0 100644 --- a/crates/application/src/channels/update.rs +++ b/crates/application/src/channels/update.rs @@ -35,6 +35,9 @@ pub async fn execute(deps: &ChannelCommandDeps, cmd: UpdateChannelCommand) -> Do if let Some(auto) = cmd.auto_schedule { channel.set_auto_schedule(auto); } + if let Some(gap_filler) = cmd.gap_filler { + channel.set_gap_filler(gap_filler); + } deps.channel_command.save(&channel).await?; diff --git a/crates/domain/src/services/schedule/mod.rs b/crates/domain/src/services/schedule/mod.rs index f0fbdfa..17f54b2 100644 --- a/crates/domain/src/services/schedule/mod.rs +++ b/crates/domain/src/services/schedule/mod.rs @@ -9,7 +9,8 @@ use crate::models::{ ScheduledSlot, }; use crate::ports::{ChannelQuery, LibraryQuery, ScheduleCommand, ScheduleQuery}; -use crate::value_objects::{BlockId, ChannelId, FillStrategy, InterstitialRule, LibrarySearchFilter, MediaItemId, MediaRole, RotationPolicy, Weekday}; +use crate::models::MediaItem; +use crate::value_objects::{BlockId, ChannelId, FillStrategy, InterstitialRule, LibrarySearchFilter, MediaFilter, MediaItemId, MediaRole, RotationPolicy, Weekday}; mod fill; mod rotation; @@ -153,6 +154,13 @@ impl ScheduleEngineService { slots.sort_by_key(|s| s.start_at()); + if let Some(gap_filter) = channel.gap_filler() { + let filler_items = self.query_gap_fillers(gap_filter).await?; + if !filler_items.is_empty() { + Self::fill_gaps(&mut slots, &filler_items, valid_from, valid_until); + } + } + let schedule = GeneratedSchedule::new( channel_id, valid_from, @@ -415,6 +423,71 @@ impl ScheduleEngineService { Ok(slots) } + + async fn query_gap_fillers(&self, gap_filter: &MediaFilter) -> DomainResult> { + let filter = + media_filter_to_library_search(gap_filter).with_role(MediaRole::Interstitial); + let (items, _) = self.library_query.search(&filter).await?; + Ok(items) + } + + fn fill_gaps( + slots: &mut Vec, + fillers: &[MediaItem], + valid_from: DateTime, + valid_until: DateTime, + ) { + if fillers.is_empty() { + return; + } + + let gap_block_id = BlockId::generate(); + let mut gap_slots: Vec = Vec::new(); + let mut filler_idx = 0usize; + + let mut boundaries: Vec<(DateTime, DateTime)> = Vec::new(); + + if slots.is_empty() { + boundaries.push((valid_from, valid_until)); + } else { + if slots[0].start_at() > valid_from { + boundaries.push((valid_from, slots[0].start_at())); + } + for pair in slots.windows(2) { + if pair[1].start_at() > pair[0].end_at() { + boundaries.push((pair[0].end_at(), pair[1].start_at())); + } + } + if let Some(last) = slots.last() { + if last.end_at() < valid_until { + boundaries.push((last.end_at(), valid_until)); + } + } + } + + for (gap_start, gap_end) in boundaries { + let mut cursor = gap_start; + while cursor < gap_end { + let filler = &fillers[filler_idx % fillers.len()]; + let filler_end = + (cursor + Duration::seconds(filler.duration_secs() as i64)).min(gap_end); + if filler_end <= cursor { + break; + } + gap_slots.push(ScheduledSlot::new( + cursor, + filler_end, + filler.clone(), + gap_block_id, + )); + cursor = filler_end; + filler_idx += 1; + } + } + + slots.append(&mut gap_slots); + slots.sort_by_key(|s| s.start_at()); + } } fn media_filter_to_library_search(filter: &crate::value_objects::MediaFilter) -> LibrarySearchFilter { diff --git a/crates/domain/src/services/schedule/tests/integration.rs b/crates/domain/src/services/schedule/tests/integration.rs index ce833ab..a606d05 100644 --- a/crates/domain/src/services/schedule/tests/integration.rs +++ b/crates/domain/src/services/schedule/tests/integration.rs @@ -420,3 +420,138 @@ async fn interstitial_min_gap_respected() { "no interstitials should be inserted when programs are shorter than min_gap_secs" ); } + +impl TestHarness { + async fn create_channel_with_blocks_and_gap_filler( + &self, + blocks: Vec, + gap_filler: Option, + ) -> Channel { + let mut channel = Channel::new( + crate::value_objects::UserId::generate(), + "test-channel", + "UTC", + ); + let today = Utc::now() + .with_timezone(&chrono_tz::UTC) + .date_naive(); + let weekday = Weekday::from(today.weekday()); + let mut config = ScheduleConfig::new(); + config.insert_day(weekday, blocks); + channel.set_schedule_config(config); + channel.set_gap_filler(gap_filler); + self.channel_repo + .channels + .lock() + .unwrap() + .insert(channel.id(), channel.clone()); + channel + } +} + +#[tokio::test] +async fn gap_filler_fills_between_blocks() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("prog1", 1800), + movie("prog2", 1800), + interstitial("filler1", 60), + interstitial("filler2", 60), + ]); + + let block1 = ProgrammingBlock::new_algorithmic( + "morning", + NaiveTime::from_hms_opt(8, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::BestFit, + ); + let block2 = ProgrammingBlock::new_algorithmic( + "afternoon", + NaiveTime::from_hms_opt(10, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::BestFit, + ); + + let channel = h + .create_channel_with_blocks_and_gap_filler( + vec![block1, block2], + Some(MediaFilter::default()), + ) + .await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let filler_count = schedule + .slots() + .iter() + .filter(|s| *s.item().role() == MediaRole::Interstitial) + .count(); + + assert!( + filler_count > 0, + "gap filler should produce interstitial slots between blocks" + ); + + for pair in schedule.slots().windows(2) { + assert!( + pair[0].end_at() <= pair[1].start_at(), + "slots overlap: {} ends at {:?} but {} starts at {:?}", + pair[0].item().title(), + pair[0].end_at(), + pair[1].item().title(), + pair[1].start_at(), + ); + } +} + +#[tokio::test] +async fn no_gap_filler_leaves_gaps_empty() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("prog1", 1800), + movie("prog2", 1800), + interstitial("filler1", 60), + ]); + + let block1 = ProgrammingBlock::new_algorithmic( + "morning", + NaiveTime::from_hms_opt(8, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::BestFit, + ); + let block2 = ProgrammingBlock::new_algorithmic( + "afternoon", + NaiveTime::from_hms_opt(10, 0, 0).unwrap(), + 30, + MediaFilter::default(), + FillStrategy::BestFit, + ); + + let channel = h + .create_channel_with_blocks_and_gap_filler(vec![block1, block2], None) + .await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let filler_count = schedule + .slots() + .iter() + .filter(|s| *s.item().role() == MediaRole::Interstitial) + .count(); + + assert_eq!( + filler_count, 0, + "no gap filler configured, should have no interstitial slots" + ); +} diff --git a/crates/mcp/src/server.rs b/crates/mcp/src/server.rs index 465b054..de63493 100644 --- a/crates/mcp/src/server.rs +++ b/crates/mcp/src/server.rs @@ -47,6 +47,7 @@ pub struct UpdateChannelParams { pub timezone: Option, pub description: Option, pub schedule_config_json: Option, + pub gap_filler_json: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -98,7 +99,7 @@ impl KTvMcpServer { channels::create_channel(&self.channel_cmd_deps, self.owner_id, &p.name, &p.timezone).await } - #[tool(description = "Update channel name, timezone, description, and/or schedule config")] + #[tool(description = "Update channel name, timezone, description, schedule config, and/or gap_filler")] async fn update_channel(&self, #[tool(aggr)] p: UpdateChannelParams) -> String { let id = match parse_uuid(&p.id) { Ok(id) => id, @@ -114,6 +115,17 @@ impl KTvMcpServer { }, None => None, }; + let gap_filler = match p.gap_filler_json { + Some(json) if json == "null" => Some(None), + Some(json) => match serde_json::from_str(&json) { + Ok(f) => Some(Some(f)), + Err(e) => { + return serde_json::json!({"error": format!("invalid gap_filler_json: {e}")}) + .to_string() + } + }, + None => None, + }; channels::update_channel( &self.channel_cmd_deps, id, @@ -122,6 +134,7 @@ impl KTvMcpServer { p.timezone, p.description, schedule_config, + gap_filler, ) .await } diff --git a/crates/mcp/src/tools/channels.rs b/crates/mcp/src/tools/channels.rs index 407209f..b1dfacf 100644 --- a/crates/mcp/src/tools/channels.rs +++ b/crates/mcp/src/tools/channels.rs @@ -53,6 +53,7 @@ pub async fn update_channel( timezone: Option, description: Option, schedule_config: Option, + gap_filler: Option>, ) -> String { let cmd = UpdateChannelCommand { channel_id: channel_id.into(), @@ -63,6 +64,7 @@ pub async fn update_channel( schedule_config, rotation_policy: None, auto_schedule: None, + gap_filler, }; match application::channels::update::execute(cmd_deps, cmd).await { Ok(channel) => ok_json(&channel), diff --git a/crates/presentation/src/handlers/channels.rs b/crates/presentation/src/handlers/channels.rs index 995eef5..8499f6a 100644 --- a/crates/presentation/src/handlers/channels.rs +++ b/crates/presentation/src/handlers/channels.rs @@ -71,6 +71,7 @@ pub async fn update_channel( schedule_config: req.schedule_config.map(Into::into), rotation_policy: req.rotation_policy, auto_schedule: req.auto_schedule, + gap_filler: req.gap_filler, }; let channel = application::channels::update::execute(&state.channel_command_deps, cmd).await?; Ok(Json(ChannelResponse::from(channel))) From 874de68fb5d1a9fc3aa786bb5cc8a3c9a52c3578 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 14:12:59 +0200 Subject: [PATCH 4/4] wire mid-roll breaks into schedule engine with chapter-aware splitting --- crates/domain/src/services/schedule/fill.rs | 8 +- crates/domain/src/services/schedule/mod.rs | 153 +++++++++++++++-- .../services/schedule/tests/integration.rs | 161 +++++++++++++++++- crates/mcp/src/server.rs | 16 +- crates/mcp/src/tools/channels.rs | 32 ++-- 5 files changed, 333 insertions(+), 37 deletions(-) diff --git a/crates/domain/src/services/schedule/fill.rs b/crates/domain/src/services/schedule/fill.rs index b027940..f4b71d4 100644 --- a/crates/domain/src/services/schedule/fill.rs +++ b/crates/domain/src/services/schedule/fill.rs @@ -311,10 +311,10 @@ pub(super) fn fill_marathon<'a>( } } - if result.is_empty() { - if let Some(&first) = ordered.first() { - result.push(first); - } + if result.is_empty() + && let Some(&first) = ordered.first() + { + result.push(first); } result diff --git a/crates/domain/src/services/schedule/mod.rs b/crates/domain/src/services/schedule/mod.rs index 17f54b2..defe15f 100644 --- a/crates/domain/src/services/schedule/mod.rs +++ b/crates/domain/src/services/schedule/mod.rs @@ -10,7 +10,7 @@ use crate::models::{ }; use crate::ports::{ChannelQuery, LibraryQuery, ScheduleCommand, ScheduleQuery}; use crate::models::MediaItem; -use crate::value_objects::{BlockId, ChannelId, FillStrategy, InterstitialRule, LibrarySearchFilter, MediaFilter, MediaItemId, MediaRole, RotationPolicy, Weekday}; +use crate::value_objects::{BlockId, ChannelId, FillStrategy, InterstitialRule, LibrarySearchFilter, MediaFilter, MediaItemId, MediaRole, MidRollRule, RotationPolicy, Weekday}; mod fill; mod rotation; @@ -283,11 +283,18 @@ impl ScheduleEngineService { } }; - if let Some(rule) = block.interstitial_rule() { + let with_interstitials = if let Some(rule) = block.interstitial_rule() { self.insert_interstitials(program_slots, rule, block.id(), window.end) + .await? + } else { + program_slots + }; + + if let Some(rule) = block.mid_roll_rule() { + self.apply_mid_rolls(with_interstitials, rule, block.id(), window.end) .await } else { - Ok(program_slots) + Ok(with_interstitials) } } @@ -354,6 +361,132 @@ impl ScheduleEngineService { Ok(result) } + async fn apply_mid_rolls( + &self, + slots: Vec, + rule: &MidRollRule, + block_id: BlockId, + block_end: DateTime, + ) -> DomainResult> { + let interval_secs = rule.fallback_interval_mins() as u64 * 60; + if interval_secs == 0 { + return Ok(slots); + } + + let mut filter = media_filter_to_library_search(rule.pool_filter()); + filter = filter.with_role(MediaRole::Interstitial); + let (break_items, _) = self.library_query.search(&filter).await?; + + if break_items.is_empty() { + return Ok(slots); + } + + let mut result: Vec = Vec::new(); + let mut break_idx = 0usize; + + for slot in &slots { + let item_duration = slot.item().duration_secs() as u64; + + if item_duration < interval_secs { + result.push(slot.clone()); + continue; + } + + let break_points = Self::compute_break_points( + slot.item(), + rule.prefer_chapters(), + interval_secs, + ); + + if break_points.is_empty() { + result.push(slot.clone()); + continue; + } + + let mut cursor = slot.start_at(); + let mut prev_offset = 0u64; + + for bp in &break_points { + if cursor >= block_end { + break; + } + + let segment_duration = bp - prev_offset; + let segment_end = + (cursor + Duration::seconds(segment_duration as i64)).min(block_end); + + result.push(ScheduledSlot::new( + cursor, + segment_end, + slot.item().clone(), + block_id, + )); + cursor = segment_end; + prev_offset = *bp; + + if cursor < block_end { + let break_item = &break_items[break_idx % break_items.len()]; + let break_end_secs = rule.break_duration_secs().min(break_item.duration_secs()); + let break_end = + (cursor + Duration::seconds(break_end_secs as i64)).min(block_end); + if break_end > cursor { + result.push(ScheduledSlot::new( + cursor, + break_end, + break_item.clone(), + block_id, + )); + cursor = break_end; + break_idx += 1; + } + } + } + + let remaining = item_duration - prev_offset; + if remaining > 0 && cursor < block_end { + let tail_end = + (cursor + Duration::seconds(remaining as i64)).min(block_end); + result.push(ScheduledSlot::new( + cursor, + tail_end, + slot.item().clone(), + block_id, + )); + } + } + + Ok(result) + } + + fn compute_break_points( + item: &MediaItem, + prefer_chapters: bool, + interval_secs: u64, + ) -> Vec { + if prefer_chapters && !item.chapters().is_empty() { + item.chapters() + .iter() + .filter_map(|ch| { + let end = ch.end_secs() as u64; + if end > 0 && end < item.duration_secs() as u64 { + Some(end) + } else { + None + } + }) + .collect() + } else { + let duration = item.duration_secs() as u64; + let mut points = Vec::new(); + let mut offset = interval_secs; + while offset < duration { + points.push(offset); + offset += interval_secs; + } + points + } + } + async fn resolve_manual( &self, item_ids: &[MediaItemId], @@ -458,10 +591,10 @@ impl ScheduleEngineService { boundaries.push((pair[0].end_at(), pair[1].start_at())); } } - if let Some(last) = slots.last() { - if last.end_at() < valid_until { - boundaries.push((last.end_at(), valid_until)); - } + if let Some(last) = slots.last() + && last.end_at() < valid_until + { + boundaries.push((last.end_at(), valid_until)); } } @@ -509,10 +642,8 @@ fn media_filter_to_library_search(filter: &crate::value_objects::MediaFilter) -> if let Some(max) = filter.max_duration_secs { lsf = lsf.with_max_duration_secs(max); } - if !filter.collections.is_empty() { - if let Some(first) = filter.collections.first() { - lsf = lsf.with_collection_id(first.clone()); - } + if let Some(first) = filter.collections.first() { + lsf = lsf.with_collection_id(first.clone()); } if !filter.series_names.is_empty() { lsf = lsf.with_series_names(filter.series_names.clone()); diff --git a/crates/domain/src/services/schedule/tests/integration.rs b/crates/domain/src/services/schedule/tests/integration.rs index a606d05..9d7393f 100644 --- a/crates/domain/src/services/schedule/tests/integration.rs +++ b/crates/domain/src/services/schedule/tests/integration.rs @@ -8,7 +8,8 @@ use crate::testing::{ InMemoryChannelRepository, InMemoryLibraryRepository, InMemoryScheduleRepository, }; use crate::value_objects::{ - ContentType, FillStrategy, InterstitialRule, MediaFilter, MediaItemId, MediaRole, Weekday, + Chapter, ContentType, FillStrategy, InterstitialRule, MediaFilter, MediaItemId, MediaRole, + MidRollRule, Weekday, }; fn episode(id: &str, series: &str, ep: u32, secs: u32) -> MediaItem { @@ -555,3 +556,161 @@ async fn no_gap_filler_leaves_gaps_empty() { "no gap filler configured, should have no interstitial slots" ); } + +fn movie_with_chapters(id: &str, secs: u32, chapters: Vec) -> MediaItem { + let item = MediaItem::new(MediaItemId::new(id), id, ContentType::Movie, secs); + let mut val = serde_json::to_value(&item).unwrap(); + val["chapters"] = serde_json::to_value(&chapters).unwrap(); + serde_json::from_value(val).unwrap() +} + +#[tokio::test] +async fn mid_roll_splits_long_movie_at_intervals() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("long-movie", 7200), + interstitial("ad1", 120), + interstitial("ad2", 120), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "movie-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 180, + MediaFilter::default(), + FillStrategy::BestFit, + ) + .with_mid_roll_rule(MidRollRule::new( + false, + 30, + 120, + MediaFilter::default(), + )); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let today_slots: Vec<_> = schedule.slots().iter().take(20).collect(); + + let movie_segments = today_slots + .iter() + .filter(|s| s.item().id().value() == "long-movie") + .count(); + let break_count = today_slots + .iter() + .filter(|s| *s.item().role() == MediaRole::Interstitial) + .count(); + + assert!( + movie_segments >= 2, + "2h movie with 30min interval should produce >= 2 segments, got {movie_segments}" + ); + assert!( + break_count >= 1, + "should have at least 1 break slot, got {break_count}" + ); +} + +#[tokio::test] +async fn mid_roll_prefers_chapter_boundaries() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie_with_chapters( + "chaptered-movie", + 7200, + vec![ + Chapter::new(Some("Act 1".into()), 0.0, 2400.0), + Chapter::new(Some("Act 2".into()), 2400.0, 4800.0), + Chapter::new(Some("Act 3".into()), 4800.0, 7200.0), + ], + ), + interstitial("ad1", 120), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "chapter-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 180, + MediaFilter::default(), + FillStrategy::BestFit, + ) + .with_mid_roll_rule(MidRollRule::new( + true, + 30, + 120, + MediaFilter::default(), + )); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let today_slots: Vec<_> = schedule.slots().iter().take(20).collect(); + + let movie_segments = today_slots + .iter() + .filter(|s| s.item().id().value() == "chaptered-movie") + .count(); + + assert!( + movie_segments >= 3, + "chaptered movie should split at chapter boundaries, got {movie_segments} segments" + ); +} + +#[tokio::test] +async fn mid_roll_short_item_not_split() { + let h = TestHarness::new(); + h.seed_items(vec![ + movie("short-movie", 1200), + interstitial("ad1", 120), + ]); + + let block = ProgrammingBlock::new_algorithmic( + "short-block", + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + 60, + MediaFilter::default(), + FillStrategy::BestFit, + ) + .with_mid_roll_rule(MidRollRule::new( + false, + 30, + 120, + MediaFilter::default(), + )); + let channel = h.create_channel_with_block(block).await; + + let schedule = h + .engine + .generate_schedule(channel.id(), Utc::now()) + .await + .unwrap(); + + let today_slots: Vec<_> = schedule.slots().iter().take(10).collect(); + + let movie_segments = today_slots + .iter() + .filter(|s| s.item().id().value() == "short-movie") + .count(); + let break_count = today_slots + .iter() + .filter(|s| *s.item().role() == MediaRole::Interstitial) + .count(); + + assert_eq!( + movie_segments, 1, + "20min movie under 30min interval should not be split" + ); + assert_eq!( + break_count, 0, + "no breaks for short items" + ); +} diff --git a/crates/mcp/src/server.rs b/crates/mcp/src/server.rs index de63493..fe1af8b 100644 --- a/crates/mcp/src/server.rs +++ b/crates/mcp/src/server.rs @@ -128,13 +128,15 @@ impl KTvMcpServer { }; channels::update_channel( &self.channel_cmd_deps, - id, - self.owner_id, - p.name, - p.timezone, - p.description, - schedule_config, - gap_filler, + channels::UpdateChannelArgs { + channel_id: id, + owner_id: self.owner_id, + name: p.name, + timezone: p.timezone, + description: p.description, + schedule_config, + gap_filler, + }, ) .await } diff --git a/crates/mcp/src/tools/channels.rs b/crates/mcp/src/tools/channels.rs index b1dfacf..1cd4d0e 100644 --- a/crates/mcp/src/tools/channels.rs +++ b/crates/mcp/src/tools/channels.rs @@ -45,26 +45,30 @@ pub async fn create_channel( } } +pub struct UpdateChannelArgs { + pub channel_id: Uuid, + pub owner_id: Uuid, + pub name: Option, + pub timezone: Option, + pub description: Option, + pub schedule_config: Option, + pub gap_filler: Option>, +} + pub async fn update_channel( cmd_deps: &Arc, - channel_id: Uuid, - owner_id: Uuid, - name: Option, - timezone: Option, - description: Option, - schedule_config: Option, - gap_filler: Option>, + args: UpdateChannelArgs, ) -> String { let cmd = UpdateChannelCommand { - channel_id: channel_id.into(), - owner_id: owner_id.into(), - name, - description: description.map(Some), - timezone, - schedule_config, + channel_id: args.channel_id.into(), + owner_id: args.owner_id.into(), + name: args.name, + description: args.description.map(Some), + timezone: args.timezone, + schedule_config: args.schedule_config, rotation_policy: None, auto_schedule: None, - gap_filler, + gap_filler: args.gap_filler, }; match application::channels::update::execute(cmd_deps, cmd).await { Ok(channel) => ok_json(&channel),