GKaszewski/k-tv
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases Wiki Activity

domain crate code quality cleanup

Browse Source 
strip all comments, extract tests to tests/ dirs,
remove #[allow(clippy::...)], extract magic numbers to
constants, refactor schedule engine private methods to
use param structs, add Default impls, clippy.toml for
persistence constructors
This commit is contained in:
Gabriel Kaszewski
2026-07-12 04:02:12 +02:00
parent d650e2ba07
commit 98a54245b1
54 changed files with 1190 additions and 1942 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
crates/domain/src/services/schedule/mod.rs
View File

@@ -1,10 +1,3 @@
//! 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};
@@ -21,6 +14,20 @@ use crate::value_objects::{BlockId, ChannelId, FillStrategy, MediaFilter, MediaI
mod fill;
mod recycle;
const SCHEDULE_DURATION_DAYS: i64 = 7;
struct BlockTimeWindow {
start: DateTime<Utc>,
end: DateTime<Utc>,
}
struct RecycleContext<'a> {
history: &'a [PlaybackRecord],
policy: &'a RecyclePolicy,
generation: u32,
last_item_id: Option<&'a MediaItemId>,
}
pub struct ScheduleEngineService {
provider_registry: Arc<dyn IProviderRegistry>,
channel_query: Arc<dyn ChannelQuery>,
@@ -43,22 +50,6 @@ impl ScheduleEngineService {
}
}
// -------------------------------------------------------------------------
// 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,
@@ -80,9 +71,6 @@ impl ScheduleEngineService {
.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
@@ -90,18 +78,13 @@ impl ScheduleEngineService {
.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 valid_until = from + Duration::days(SCHEDULE_DURATION_DAYS);
let start_date = from.with_timezone(&tz).date_naive();
let end_date = valid_until.with_timezone(&tz).date_naive();
@@ -114,8 +97,7 @@ impl ScheduleEngineService {
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.
// earliest() picks first valid mapping, skipping DST gaps
let block_start_utc = match tz.from_local_datetime(&naive_start).earliest() {
Some(dt) => dt.with_timezone(&Utc),
None => continue,
@@ -124,7 +106,6 @@ impl ScheduleEngineService {
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);
@@ -132,23 +113,24 @@ impl ScheduleEngineService {
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,
BlockTimeWindow {
start: slot_start,
end: slot_end,
},
RecycleContext {
history: &history,
policy: 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());
}
@@ -161,7 +143,6 @@ impl ScheduleEngineService {
})?;
}
// Blocks in ScheduleConfig are not required to be sorted; sort resolved slots.
slots.sort_by_key(|s| s.start_at());
let schedule = GeneratedSchedule::new(
@@ -174,7 +155,6 @@ impl ScheduleEngineService {
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);
@@ -184,10 +164,6 @@ impl ScheduleEngineService {
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>,
@@ -204,7 +180,6 @@ impl ScheduleEngineService {
})
}
/// Return the most recently generated schedule for a channel (used by the background scheduler).
pub async fn get_latest_schedule(
&self,
channel_id: ChannelId,
@@ -212,7 +187,6 @@ impl ScheduleEngineService {
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,
@@ -221,7 +195,6 @@ impl ScheduleEngineService {
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,
@@ -230,7 +203,6 @@ impl ScheduleEngineService {
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,
@@ -238,7 +210,6 @@ impl ScheduleEngineService {
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,
@@ -249,7 +220,6 @@ impl ScheduleEngineService {
.await
}
/// Delete all schedules with generation > target_generation for this channel.
pub async fn delete_schedules_after(
&self,
channel_id: ChannelId,
@@ -260,7 +230,6 @@ impl ScheduleEngineService {
.await
}
/// Return all slots that overlap the given time window — the EPG data.
pub fn get_epg(
schedule: &GeneratedSchedule,
from: DateTime<Utc>,
@@ -273,24 +242,16 @@ impl ScheduleEngineService {
.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>,
window: BlockTimeWindow,
recycle: RecycleContext<'_>,
) -> DomainResult<Vec<ScheduledSlot>> {
match block.content() {
BlockContent::Manual { items, .. } => {
self.resolve_manual(items, start, end, block.id()).await
self.resolve_manual(items, window.start, window.end, block.id())
.await
}
BlockContent::Algorithmic {
filter,
@@ -301,13 +262,9 @@ impl ScheduleEngineService {
provider_id,
filter,
strategy,
start,
end,
history,
policy,
generation,
window,
recycle,
block.id(),
last_item_id,
block.loop_on_finish(),
block.ignore_recycle_policy(),
)
@@ -316,8 +273,6 @@ impl ScheduleEngineService {
}
}
/// 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],
@@ -338,36 +293,22 @@ impl ScheduleEngineService {
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,
window: BlockTimeWindow,
recycle: RecycleContext<'_>,
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)
@@ -380,27 +321,27 @@ impl ScheduleEngineService {
let pool = if ignore_recycle_policy {
candidates.clone()
} else {
recycle::apply_recycle_policy(&candidates, history, policy, generation)
recycle::apply_recycle_policy(&candidates, recycle.history, recycle.policy, recycle.generation)
};
let target_secs = (end - start).num_seconds() as u32;
let target_secs = (window.end - window.start).num_seconds() as u32;
let selected = fill::fill_block(
&candidates,
&pool,
target_secs,
strategy,
last_item_id,
recycle.last_item_id,
loop_on_finish,
);
let mut slots = Vec::new();
let mut cursor = start;
let mut cursor = window.start;
for item in selected {
if cursor >= end {
if cursor >= window.end {
break;
}
let item_end =
(cursor + Duration::seconds(item.duration_secs() as i64)).min(end);
(cursor + Duration::seconds(item.duration_secs() as i64)).min(window.end);
slots.push(ScheduledSlot::new(cursor, item_end, item.clone(), block_id));
cursor = item_end;
}