From 7ddf94c75fb95adbf0f4a864d8328bffe1492d46 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 01:35:01 +0200 Subject: [PATCH] domain services: schedule engine, fill strategies, IPTV --- crates/domain/src/lib.rs | 2 + crates/domain/src/services/iptv.rs | 169 ++++++++ crates/domain/src/services/mod.rs | 11 + crates/domain/src/services/schedule/fill.rs | 234 ++++++++++ crates/domain/src/services/schedule/mod.rs | 410 ++++++++++++++++++ .../domain/src/services/schedule/recycle.rs | 118 +++++ 6 files changed, 944 insertions(+) create mode 100644 crates/domain/src/services/iptv.rs create mode 100644 crates/domain/src/services/mod.rs create mode 100644 crates/domain/src/services/schedule/fill.rs create mode 100644 crates/domain/src/services/schedule/mod.rs create mode 100644 crates/domain/src/services/schedule/recycle.rs diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 3622c59..7fff863 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -2,9 +2,11 @@ pub mod errors; pub mod events; pub mod models; pub mod ports; +pub mod services; pub mod value_objects; pub use errors::{DomainError, DomainResult}; pub use events::DomainEvent; pub use models::*; +pub use services::{generate_m3u, generate_xmltv, ScheduleEngineService}; pub use value_objects::*; diff --git a/crates/domain/src/services/iptv.rs b/crates/domain/src/services/iptv.rs new file mode 100644 index 0000000..a467331 --- /dev/null +++ b/crates/domain/src/services/iptv.rs @@ -0,0 +1,169 @@ +//! IPTV export: M3U playlist and XMLTV guide generation. +//! +//! Pure functions — no I/O, no dependencies beyond domain types. + +use std::collections::HashMap; + +use crate::models::{Channel, ScheduledSlot}; +use crate::value_objects::ChannelId; + +/// Generate an M3U playlist for the given channels. +/// +/// Each entry points to the channel's `/stream` endpoint authenticated with the +/// provided JWT token so IPTV clients can load it directly. +pub fn generate_m3u(channels: &[Channel], base_url: &str, token: &str) -> String { + let mut out = String::from("#EXTM3U\n"); + for ch in channels { + out.push_str(&format!( + "#EXTINF:-1 tvg-id=\"{}\" tvg-name=\"{}\" tvg-logo=\"\" group-title=\"K-TV\",{}\n", + ch.id(), + ch.name(), + ch.name() + )); + out.push_str(&format!( + "{}/api/v1/channels/{}/stream?token={}\n", + base_url, + ch.id(), + token + )); + } + out +} + +/// Generate an XMLTV EPG document for the given channels and their scheduled slots. +pub fn generate_xmltv( + channels: &[Channel], + slots_by_channel: &HashMap>, +) -> String { + let mut out = String::from( + "\n\n", + ); + + for ch in channels { + out.push_str(&format!( + " {}\n", + ch.id(), + escape_xml(ch.name()) + )); + } + + for ch in channels { + if let Some(slots) = slots_by_channel.get(&ch.id()) { + for slot in slots { + let start = slot.start_at().format("%Y%m%d%H%M%S +0000"); + let stop = slot.end_at().format("%Y%m%d%H%M%S +0000"); + out.push_str(&format!( + " \n", + start, + stop, + ch.id() + )); + out.push_str(&format!( + " {}\n", + escape_xml(slot.item().title()) + )); + if let Some(desc) = slot.item().description() { + out.push_str(&format!( + " {}\n", + escape_xml(desc) + )); + } + if let Some(genre) = slot.item().genres().first() { + out.push_str(&format!( + " {}\n", + escape_xml(genre) + )); + } + if let (Some(season), Some(episode)) = + (slot.item().season_number(), slot.item().episode_number()) + { + out.push_str(&format!( + " S{}E{}\n", + season, episode + )); + } + out.push_str(" \n"); + } + } + } + + out.push_str("\n"); + out +} + +fn escape_xml(s: &str) -> String { + s.replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) + .replace('\'', "'") +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::models::MediaItem; + use crate::value_objects::{ContentType, MediaItemId, UserId}; + use chrono::{Duration, Utc}; + + fn make_channel(name: &str) -> Channel { + Channel::new(UserId::generate(), name, "UTC") + } + + fn make_slot(title: &str, start_offset_hours: i64) -> ScheduledSlot { + let start = Utc::now() + Duration::hours(start_offset_hours); + let item = MediaItem::new( + MediaItemId::new(format!("test::{title}")), + title, + ContentType::Movie, + 3600, + ); + ScheduledSlot::new( + start, + start + Duration::hours(1), + item, + crate::value_objects::BlockId::generate(), + ) + } + + #[test] + fn m3u_contains_all_channels() { + let channels = vec![make_channel("Channel 1"), make_channel("Channel 2")]; + let m3u = generate_m3u(&channels, "http://localhost:3000", "tok123"); + assert!(m3u.starts_with("#EXTM3U\n")); + assert!(m3u.contains("Channel 1")); + assert!(m3u.contains("Channel 2")); + assert!(m3u.contains("token=tok123")); + } + + #[test] + fn xmltv_structure() { + let channels = vec![make_channel("Test TV")]; + let ch_id = channels[0].id(); + let mut slots_map = HashMap::new(); + slots_map.insert(ch_id, vec![make_slot("Movie Night", 0)]); + + let xml = generate_xmltv(&channels, &slots_map); + assert!(xml.contains("")); + assert!(xml.contains("Test TV")); + assert!(xml.contains("Movie Night")); + assert!(xml.contains("")); + } + + #[test] + fn xmltv_escapes_special_chars() { + let channels = vec![make_channel("A&B ")]; + let xml = generate_xmltv(&channels, &HashMap::new()); + assert!(xml.contains("A&B <Channel>")); + } + + #[test] + fn m3u_empty_channels() { + let m3u = generate_m3u(&[], "http://localhost", "tok"); + assert_eq!(m3u, "#EXTM3U\n"); + } +} diff --git a/crates/domain/src/services/mod.rs b/crates/domain/src/services/mod.rs new file mode 100644 index 0000000..a58fa4c --- /dev/null +++ b/crates/domain/src/services/mod.rs @@ -0,0 +1,11 @@ +//! Domain services — pure business logic with no I/O side effects. +//! +//! The scheduling engine and IPTV export functions live here. Application-level +//! orchestration (user/channel CRUD, auth flows) belongs in the `application` +//! crate's use cases, not here. + +pub mod iptv; +pub mod schedule; + +pub use iptv::{generate_m3u, generate_xmltv}; +pub use schedule::ScheduleEngineService; diff --git a/crates/domain/src/services/schedule/fill.rs b/crates/domain/src/services/schedule/fill.rs new file mode 100644 index 0000000..f450d70 --- /dev/null +++ b/crates/domain/src/services/schedule/fill.rs @@ -0,0 +1,234 @@ +//! Fill strategies for scheduling engine block resolution. +//! +//! Pure functions — no I/O, no async, no side effects. +//! Each strategy selects items from a pool to fill a target time budget. + +use std::collections::HashSet; + +use rand::rngs::StdRng; +use rand::seq::SliceRandom; +use rand::SeedableRng; + +use crate::models::MediaItem; +use crate::value_objects::{FillStrategy, MediaItemId}; + +/// Select items from `pool` (recycled-filtered) to fill `target_secs`, using +/// `strategy`. `candidates` (unfiltered) is only needed by `Sequential` for +/// ordering; `last_item_id` drives series continuity. +pub(super) fn fill_block<'a>( + candidates: &'a [MediaItem], + pool: &'a [MediaItem], + target_secs: u32, + strategy: &FillStrategy, + last_item_id: Option<&MediaItemId>, + loop_on_finish: bool, +) -> Vec<&'a MediaItem> { + match strategy { + FillStrategy::BestFit => fill_best_fit(pool, target_secs), + FillStrategy::Sequential => { + fill_sequential(candidates, pool, target_secs, last_item_id, loop_on_finish) + } + FillStrategy::Random => { + let mut indices: Vec = (0..pool.len()).collect(); + indices.shuffle(&mut StdRng::from_entropy()); + let mut remaining = target_secs; + let mut result = Vec::new(); + for i in indices { + let item = &pool[i]; + if item.duration_secs() <= remaining { + remaining -= item.duration_secs(); + result.push(item); + } + } + result + } + } +} + +/// Greedy bin-packing: at each step pick the longest item that still fits +/// in the remaining budget, without repeating items within the same block. +pub(super) fn fill_best_fit(pool: &[MediaItem], target_secs: u32) -> Vec<&MediaItem> { + let mut remaining = target_secs; + let mut selected: Vec<&MediaItem> = Vec::new(); + let mut used: HashSet = HashSet::new(); + + loop { + let best = pool + .iter() + .enumerate() + .filter(|(idx, item)| !used.contains(idx) && item.duration_secs() <= remaining) + .max_by_key(|(_, item)| item.duration_secs()); + + match best { + Some((idx, item)) => { + remaining -= item.duration_secs(); + used.insert(idx); + selected.push(item); + } + None => break, + } + } + + selected +} + +/// Sequential fill with cross-generation series continuity. +/// +/// `candidates` — all items matching the filter, in Jellyfin's natural order +/// (typically by season + episode number for TV shows). +/// `pool` — candidates filtered by the recycle policy (eligible to air). +/// `last_item_id` — the last item scheduled in this block in the previous +/// generation or in an earlier occurrence of this block within +/// the current generation. Used to resume the series from the +/// next episode rather than restarting from episode 1. +/// +/// Algorithm: +/// 1. Find `last_item_id`'s position in `candidates` and start from the next index. +/// 2. Walk the full `candidates` list in order (wrapping around at the end), +/// but only pick items that are in `pool` (i.e. not on cooldown). +/// 3. Greedily fill the time budget with items in that order. +/// +/// This ensures episodes always air in series order, the series wraps correctly +/// when the last episode has been reached, and cooldowns are still respected. +pub(super) fn fill_sequential<'a>( + candidates: &'a [MediaItem], + pool: &'a [MediaItem], + target_secs: u32, + last_item_id: Option<&MediaItemId>, + loop_on_finish: bool, +) -> Vec<&'a MediaItem> { + if pool.is_empty() { + return vec![]; + } + + // Set of item IDs currently eligible to air. + let available: HashSet<&MediaItemId> = pool.iter().map(|i| i.id()).collect(); + + let ordered: Vec<&MediaItem> = if loop_on_finish { + // Find where in the full ordered list to resume, wrapping around. + // Falls back to index 0 if last_item_id is absent or was removed from the library. + let start_idx = last_item_id + .and_then(|id| candidates.iter().position(|c| c.id() == id)) + .map(|pos| (pos + 1) % candidates.len()) + .unwrap_or(0); + + (0..candidates.len()) + .map(|i| &candidates[(start_idx + i) % candidates.len()]) + .filter(|item| available.contains(item.id())) + .collect() + } else { + // No wrap: compute raw next position without modulo. + // If the series has finished (next_pos >= len), return dead air. + let next_pos = last_item_id + .and_then(|id| candidates.iter().position(|c| c.id() == id)) + .map(|pos| pos + 1) + .unwrap_or(0); + + if next_pos >= candidates.len() { + return vec![]; // series finished — dead air + } + + candidates[next_pos..] + .iter() + .filter(|item| available.contains(item.id())) + .collect() + }; + + // Greedily fill the block's time budget in episode order. + // Stop at the first episode that doesn't fit — skipping would break ordering. + let mut remaining = target_secs; + let mut result = Vec::new(); + for item in &ordered { + if item.duration_secs() <= remaining { + remaining -= item.duration_secs(); + result.push(*item); + } else { + break; + } + } + // Edge case: if the very first episode is longer than the entire block, + // still include it — the slot builder clips it to block end via .min(end). + if result.is_empty() { + if let Some(&first) = ordered.first() { + result.push(first); + } + } + result +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::value_objects::ContentType; + + fn item(id: &str, secs: u32) -> MediaItem { + MediaItem::new(MediaItemId::new(id), id, ContentType::Movie, secs) + } + + #[test] + fn best_fit_picks_longest_first() { + let pool = vec![item("a", 100), item("b", 200), item("c", 150)]; + let result = fill_best_fit(&pool, 350); + assert_eq!(result.len(), 2); + assert_eq!(result[0].id().value(), "b"); // 200 first + assert_eq!(result[1].id().value(), "c"); // 150 next + } + + #[test] + fn best_fit_no_repeats() { + let pool = vec![item("a", 100)]; + let result = fill_best_fit(&pool, 300); + assert_eq!(result.len(), 1); // only one item, can't repeat + } + + #[test] + fn sequential_resumes_from_last() { + let candidates = vec![item("ep1", 60), item("ep2", 60), item("ep3", 60)]; + let pool = candidates.clone(); + let last = MediaItemId::new("ep1"); + let result = fill_sequential(&candidates, &pool, 120, Some(&last), true); + assert_eq!(result.len(), 2); + assert_eq!(result[0].id().value(), "ep2"); + assert_eq!(result[1].id().value(), "ep3"); + } + + #[test] + fn sequential_wraps_when_loop_on_finish() { + let candidates = vec![item("ep1", 60), item("ep2", 60), item("ep3", 60)]; + let pool = candidates.clone(); + let last = MediaItemId::new("ep3"); + let result = fill_sequential(&candidates, &pool, 180, Some(&last), true); + assert_eq!(result.len(), 3); + assert_eq!(result[0].id().value(), "ep1"); // wrapped + } + + #[test] + fn sequential_dead_air_when_no_loop() { + let candidates = vec![item("ep1", 60), item("ep2", 60)]; + let pool = candidates.clone(); + let last = MediaItemId::new("ep2"); + let result = fill_sequential(&candidates, &pool, 120, Some(&last), false); + assert!(result.is_empty()); // series finished + } + + #[test] + fn sequential_includes_oversize_first_episode() { + let candidates = vec![item("ep1", 9999)]; + let pool = candidates.clone(); + let result = fill_sequential(&candidates, &pool, 60, None, true); + assert_eq!(result.len(), 1); // included despite being too long + } + + #[test] + fn random_fill_respects_budget() { + let pool = vec![item("a", 100), item("b", 100), item("c", 100)]; + let candidates = pool.clone(); + let result = fill_block(&candidates, &pool, 200, &FillStrategy::Random, None, true); + let total: u32 = result.iter().map(|i| i.duration_secs()).sum(); + assert!(total <= 200); + } +} diff --git a/crates/domain/src/services/schedule/mod.rs b/crates/domain/src/services/schedule/mod.rs new file mode 100644 index 0000000..7c502b5 --- /dev/null +++ b/crates/domain/src/services/schedule/mod.rs @@ -0,0 +1,410 @@ +//! Core scheduling engine. +//! +//! Generates 7-day broadcast schedules by walking through a channel's +//! `ScheduleConfig` day by day, resolving each `ProgrammingBlock` into concrete +//! `ScheduledSlot`s via the `IProviderRegistry`, and applying the `RecyclePolicy` +//! to avoid replaying recently aired items. + +use std::sync::Arc; + +use chrono::{DateTime, Datelike, Duration, TimeZone, Utc}; +use chrono_tz::Tz; + +use crate::errors::{DomainError, DomainResult}; +use crate::models::{ + BlockContent, CurrentBroadcast, GeneratedSchedule, PlaybackRecord, ProgrammingBlock, + ScheduledSlot, +}; +use crate::ports::{ChannelQuery, IProviderRegistry, ScheduleCommand, ScheduleQuery, StreamQuality}; +use crate::value_objects::{BlockId, ChannelId, FillStrategy, MediaFilter, MediaItemId, RecyclePolicy, Weekday}; + +mod fill; +mod recycle; + +pub struct ScheduleEngineService { + provider_registry: Arc, + channel_query: Arc, + schedule_query: Arc, + schedule_command: Arc, +} + +impl ScheduleEngineService { + pub fn new( + provider_registry: Arc, + channel_query: Arc, + schedule_query: Arc, + schedule_command: Arc, + ) -> Self { + Self { + provider_registry, + channel_query, + schedule_query, + schedule_command, + } + } + + // ------------------------------------------------------------------------- + // Public API + // ------------------------------------------------------------------------- + + /// Generate and persist a 7-day schedule for `channel_id` starting at `from`. + /// + /// The algorithm: + /// 1. Walk each calendar day in the 7-day window. + /// 2. For each `ProgrammingBlock`, compute its UTC wall-clock interval for that day. + /// 3. Clip the interval to `[from, from + 7d)`. + /// 4. Resolve the block content via the media provider, applying the recycle policy. + /// 5. For `Sequential` blocks, resume from where the previous generation left off + /// (series continuity — see `fill::fill_sequential`). + /// 6. Record every played item in the playback history. + /// + /// Gaps between blocks are left empty — clients render them as a no-signal state. + pub async fn generate_schedule( + &self, + channel_id: ChannelId, + from: DateTime, + ) -> DomainResult { + let channel = self + .channel_query + .find_by_id(channel_id) + .await? + .ok_or(DomainError::ChannelNotFound(channel_id.value()))?; + + let tz: Tz = channel + .timezone() + .parse() + .map_err(|_| DomainError::TimezoneError(channel.timezone().to_owned()))?; + + let history = self + .schedule_query + .find_playback_history(channel_id) + .await?; + + // Load the most recent schedule for two purposes: + // 1. Derive the next generation number. + // 2. Know where each Sequential block left off (series continuity). + let latest_schedule = self.schedule_query.find_latest(channel_id).await?; + + let generation = latest_schedule + .as_ref() + .map(|s| s.generation() + 1) + .unwrap_or(1); + + // Build the initial per-block continuity map from the most recent slot per + // block across ALL schedules. This is resilient to any single generation + // having empty slots for a block (e.g. provider returned nothing transiently). + // The map is updated as each block occurrence is resolved within this + // generation so the second day of a 48h schedule continues from here. + let mut block_continuity = self + .schedule_query + .find_last_slot_per_block(channel_id) + .await?; + + let valid_from = from; + let valid_until = from + Duration::days(7); + + let start_date = from.with_timezone(&tz).date_naive(); + let end_date = valid_until.with_timezone(&tz).date_naive(); + + let mut slots: Vec = Vec::new(); + let mut current_date = start_date; + + while current_date <= end_date { + let weekday = Weekday::from(current_date.weekday()); + for block in channel.schedule_config().blocks_for(weekday) { + let naive_start = current_date.and_time(block.start_time()); + + // `earliest()` handles DST gaps — if the local time doesn't exist + // (e.g. clocks spring forward) we skip this block occurrence. + let block_start_utc = match tz.from_local_datetime(&naive_start).earliest() { + Some(dt) => dt.with_timezone(&Utc), + None => continue, + }; + + let block_end_utc = + block_start_utc + Duration::minutes(block.duration_mins() as i64); + + // Clip to the 7-day window. + let slot_start = block_start_utc.max(valid_from); + let slot_end = block_end_utc.min(valid_until); + + if slot_end <= slot_start { + continue; + } + + // For Sequential blocks: resume from the last item aired in this block. + let last_item_id = block_continuity.get(&block.id()); + + let mut block_slots = self + .resolve_block( + block, + slot_start, + slot_end, + &history, + channel.recycle_policy(), + generation, + last_item_id, + ) + .await?; + + // Update continuity so the next occurrence of this block (same + // generation, next calendar day) continues from here. + if let Some(last_slot) = block_slots.last() { + block_continuity.insert(block.id(), last_slot.item().id().clone()); + } + + slots.append(&mut block_slots); + } + + current_date = current_date.succ_opt().ok_or_else(|| { + DomainError::validation("Date overflow during schedule generation") + })?; + } + + // Blocks in ScheduleConfig are not required to be sorted; sort resolved slots. + slots.sort_by_key(|s| s.start_at()); + + let schedule = GeneratedSchedule::new( + channel_id, + valid_from, + valid_until, + generation, + slots, + ); + + self.schedule_command.save(&schedule).await?; + + // Persist playback history so the recycle policy has data for next generation. + for slot in schedule.slots() { + let record = + PlaybackRecord::new(channel_id, slot.item().id().clone(), generation); + self.schedule_command.save_playback_record(&record).await?; + } + + Ok(schedule) + } + + /// Determine what is currently broadcasting on a schedule. + /// + /// Returns `None` when `now` falls in a gap between blocks — the client + /// should display a no-signal / static screen in that case. + pub fn get_current_broadcast( + schedule: &GeneratedSchedule, + now: DateTime, + ) -> Option { + schedule + .slots() + .iter() + .find(|s| s.start_at() <= now && now < s.end_at()) + .map(|slot| { + CurrentBroadcast::new( + slot.clone(), + (now - slot.start_at()).num_seconds() as u32, + ) + }) + } + + /// Return the most recently generated schedule for a channel (used by the background scheduler). + pub async fn get_latest_schedule( + &self, + channel_id: ChannelId, + ) -> DomainResult> { + self.schedule_query.find_latest(channel_id).await + } + + /// Look up the schedule currently active at `at` without generating a new one. + pub async fn get_active_schedule( + &self, + channel_id: ChannelId, + at: DateTime, + ) -> DomainResult> { + self.schedule_query.find_active(channel_id, at).await + } + + /// Delegate stream URL resolution to the provider registry (routes via ID prefix). + pub async fn get_stream_url( + &self, + item_id: &MediaItemId, + quality: &StreamQuality, + ) -> DomainResult { + self.provider_registry.get_stream_url(item_id, quality).await + } + + /// List all generated schedule headers for a channel, newest first. + pub async fn list_schedule_history( + &self, + channel_id: ChannelId, + ) -> DomainResult> { + self.schedule_query.list_schedule_history(channel_id).await + } + + /// Fetch a specific schedule with its slots. + pub async fn get_schedule_by_id( + &self, + channel_id: ChannelId, + schedule_id: crate::value_objects::ScheduleId, + ) -> DomainResult> { + self.schedule_query + .get_schedule_by_id(channel_id, schedule_id) + .await + } + + /// Delete all schedules with generation > target_generation for this channel. + pub async fn delete_schedules_after( + &self, + channel_id: ChannelId, + target_generation: u32, + ) -> DomainResult<()> { + self.schedule_command + .delete_schedules_after(channel_id, target_generation) + .await + } + + /// Return all slots that overlap the given time window — the EPG data. + pub fn get_epg( + schedule: &GeneratedSchedule, + from: DateTime, + until: DateTime, + ) -> Vec<&ScheduledSlot> { + schedule + .slots() + .iter() + .filter(|s| s.start_at() < until && s.end_at() > from) + .collect() + } + + // ------------------------------------------------------------------------- + // Block resolution + // ------------------------------------------------------------------------- + + #[allow(clippy::too_many_arguments)] + async fn resolve_block( + &self, + block: &ProgrammingBlock, + start: DateTime, + end: DateTime, + history: &[PlaybackRecord], + policy: &RecyclePolicy, + generation: u32, + last_item_id: Option<&MediaItemId>, + ) -> DomainResult> { + match block.content() { + BlockContent::Manual { items, .. } => { + self.resolve_manual(items, start, end, block.id()).await + } + BlockContent::Algorithmic { + filter, + strategy, + provider_id, + } => { + self.resolve_algorithmic( + provider_id, + filter, + strategy, + start, + end, + history, + policy, + generation, + block.id(), + last_item_id, + block.loop_on_finish(), + block.ignore_recycle_policy(), + ) + .await + } + } + } + + /// Resolve a manual block by fetching each hand-picked item in order. + /// Stops when the block's time budget (`end`) is exhausted. + async fn resolve_manual( + &self, + item_ids: &[MediaItemId], + start: DateTime, + end: DateTime, + block_id: BlockId, + ) -> DomainResult> { + let mut slots = Vec::new(); + let mut cursor = start; + + for item_id in item_ids { + if cursor >= end { + break; + } + if let Some(item) = self.provider_registry.fetch_by_id(item_id).await? { + let item_end = + (cursor + Duration::seconds(item.duration_secs() as i64)).min(end); + slots.push(ScheduledSlot::new(cursor, item_end, item, block_id)); + cursor = item_end; + } + // If item is not found (deleted/unavailable), silently skip it. + } + + Ok(slots) + } + + /// Resolve an algorithmic block: fetch candidates, apply recycle policy, + /// run the fill strategy, and build slots. + /// + /// `last_item_id` is the ID of the last item scheduled in this block in the + /// previous generation. Used only by `Sequential` for series continuity. + #[allow(clippy::too_many_arguments)] + async fn resolve_algorithmic( + &self, + provider_id: &str, + filter: &MediaFilter, + strategy: &FillStrategy, + start: DateTime, + end: DateTime, + history: &[PlaybackRecord], + policy: &RecyclePolicy, + generation: u32, + block_id: BlockId, + last_item_id: Option<&MediaItemId>, + loop_on_finish: bool, + ignore_recycle_policy: bool, + ) -> DomainResult> { + // `candidates` — all items matching the filter, in provider order. + // Kept separate from `pool` so Sequential can rotate through the full + // ordered list while still honouring cooldowns. + let candidates = self + .provider_registry + .fetch_items(provider_id, filter) + .await?; + + if candidates.is_empty() { + return Ok(vec![]); + } + + let pool = if ignore_recycle_policy { + candidates.clone() + } else { + recycle::apply_recycle_policy(&candidates, history, policy, generation) + }; + let target_secs = (end - start).num_seconds() as u32; + let selected = fill::fill_block( + &candidates, + &pool, + target_secs, + strategy, + last_item_id, + loop_on_finish, + ); + + let mut slots = Vec::new(); + let mut cursor = start; + + for item in selected { + if cursor >= end { + break; + } + let item_end = + (cursor + Duration::seconds(item.duration_secs() as i64)).min(end); + slots.push(ScheduledSlot::new(cursor, item_end, item.clone(), block_id)); + cursor = item_end; + } + + Ok(slots) + } +} diff --git a/crates/domain/src/services/schedule/recycle.rs b/crates/domain/src/services/schedule/recycle.rs new file mode 100644 index 0000000..fade7dc --- /dev/null +++ b/crates/domain/src/services/schedule/recycle.rs @@ -0,0 +1,118 @@ +//! Recycle policy engine for the scheduling engine. +//! +//! Pure function — no I/O, no async, no side effects. +//! Filters a candidate pool according to the channel's `RecyclePolicy`. + +use std::collections::HashSet; + +use chrono::Utc; + +use crate::models::{MediaItem, PlaybackRecord}; +use crate::value_objects::{MediaItemId, RecyclePolicy}; + +/// Filter `candidates` according to `policy`, returning the eligible pool. +/// +/// An item is on cooldown if *either* the day-based or generation-based +/// threshold is exceeded. If honouring all cooldowns would leave fewer items +/// than `policy.min_available_ratio` of the total, all cooldowns are waived +/// and the full pool is returned (prevents small libraries from stalling). +pub(super) fn apply_recycle_policy( + candidates: &[MediaItem], + history: &[PlaybackRecord], + policy: &RecyclePolicy, + current_generation: u32, +) -> Vec { + let now = Utc::now(); + + let excluded: HashSet = history + .iter() + .filter(|record| { + let by_days = policy + .cooldown_days + .map(|days| (now - record.played_at()).num_days() < days as i64) + .unwrap_or(false); + + let by_gen = policy + .cooldown_generations + .map(|gens| current_generation.saturating_sub(record.generation()) < gens) + .unwrap_or(false); + + by_days || by_gen + }) + .map(|r| r.item_id().clone()) + .collect(); + + let available: Vec = candidates + .iter() + .filter(|i| !excluded.contains(i.id())) + .cloned() + .collect(); + + let min_count = + (candidates.len() as f32 * policy.min_available_ratio).ceil() as usize; + + if available.len() < min_count { + // Pool too small after applying cooldowns — recycle everything. + candidates.to_vec() + } else { + available + } +} + +// ============================================================================ +// Tests +// ============================================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::value_objects::{ChannelId, ContentType}; + + fn item(id: &str) -> MediaItem { + MediaItem::new(MediaItemId::new(id), id, ContentType::Movie, 3600) + } + + fn record(item_id: &str, generation: u32) -> PlaybackRecord { + PlaybackRecord::new(ChannelId::generate(), MediaItemId::new(item_id), generation) + } + + #[test] + fn no_history_returns_all() { + let pool = vec![item("a"), item("b"), item("c")]; + let policy = RecyclePolicy { + cooldown_days: Some(7), + cooldown_generations: None, + min_available_ratio: 0.2, + }; + let result = apply_recycle_policy(&pool, &[], &policy, 1); + assert_eq!(result.len(), 3); + } + + #[test] + fn generation_cooldown_excludes() { + let pool = vec![item("a"), item("b"), item("c")]; + let history = vec![record("a", 1)]; + let policy = RecyclePolicy { + cooldown_days: None, + cooldown_generations: Some(2), + min_available_ratio: 0.0, + }; + let result = apply_recycle_policy(&pool, &history, &policy, 2); + assert_eq!(result.len(), 2); + assert!(result.iter().all(|i| i.id().value() != "a")); + } + + #[test] + fn min_available_ratio_waives_cooldown() { + let pool = vec![item("a"), item("b")]; + let history = vec![record("a", 1), record("b", 1)]; + let policy = RecyclePolicy { + cooldown_days: None, + cooldown_generations: Some(5), + min_available_ratio: 0.5, // needs at least 1 item + }; + // Both on cooldown, but ratio requires >= 1 item => waive all + let result = apply_recycle_policy(&pool, &history, &policy, 2); + assert_eq!(result.len(), 2); + } +}