domain services: schedule engine, fill strategies, IPTV

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<dyn IProviderRegistry>,
+    channel_query: Arc<dyn ChannelQuery>,
+    schedule_query: Arc<dyn ScheduleQuery>,
+    schedule_command: Arc<dyn ScheduleCommand>,
+}
+
+impl ScheduleEngineService {
+    pub fn new(
+        provider_registry: Arc<dyn IProviderRegistry>,
+        channel_query: Arc<dyn ChannelQuery>,
+        schedule_query: Arc<dyn ScheduleQuery>,
+        schedule_command: Arc<dyn ScheduleCommand>,
+    ) -> 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<Utc>,
+    ) -> DomainResult<GeneratedSchedule> {
+        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<ScheduledSlot> = 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<Utc>,
+    ) -> Option<CurrentBroadcast> {
+        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<Option<GeneratedSchedule>> {
+        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<Utc>,
+    ) -> DomainResult<Option<GeneratedSchedule>> {
+        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<String> {
+        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<Vec<GeneratedSchedule>> {
+        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<Option<GeneratedSchedule>> {
+        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<Utc>,
+        until: DateTime<Utc>,
+    ) -> 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<Utc>,
+        end: DateTime<Utc>,
+        history: &[PlaybackRecord],
+        policy: &RecyclePolicy,
+        generation: u32,
+        last_item_id: Option<&MediaItemId>,
+    ) -> DomainResult<Vec<ScheduledSlot>> {
+        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<Utc>,
+        end: DateTime<Utc>,
+        block_id: BlockId,
+    ) -> DomainResult<Vec<ScheduledSlot>> {
+        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<Utc>,
+        end: DateTime<Utc>,
+        history: &[PlaybackRecord],
+        policy: &RecyclePolicy,
+        generation: u32,
+        block_id: BlockId,
+        last_item_id: Option<&MediaItemId>,
+        loop_on_finish: bool,
+        ignore_recycle_policy: bool,
+    ) -> DomainResult<Vec<ScheduledSlot>> {
+        // `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)
+    }
+}