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

wire gap_filler into schedule engine, API, MCP

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-07-12 14:08:55 +02:00
parent 8edd7a9c0b
commit 79e975f057
9 changed files with 243 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/api-types/src/channels.rs
View File

@@ -35,6 +35,8 @@ pub struct UpdateChannelRequest {
pub webhook_poll_interval_secs: Option<u32>,
pub webhook_body_template: Option<Option<String>>,
pub webhook_headers: Option<Option<String>>,
#[schema(value_type = Option<Option<Object>>)]
pub gap_filler: Option<Option<domain::MediaFilter>>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
@@ -55,6 +57,8 @@ pub struct ChannelResponse {
pub webhook_poll_interval_secs: u32,
pub webhook_body_template: Option<String>,
pub webhook_headers: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub gap_filler: Option<serde_json::Value>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
@@ -78,6 +82,7 @@ impl From<domain::Channel> 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(),
}

3
crates/application/src/channels/commands.rs
View File

@@ -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<ScheduleConfig>,
pub rotation_policy: Option<RotationPolicy>,
pub auto_schedule: Option<bool>,
pub gap_filler: Option<Option<MediaFilter>>,
}
pub struct DeleteChannelCommand {

7
crates/application/src/channels/tests/update.rs
View File

@@ -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

3
crates/application/src/channels/update.rs
View File

@@ -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?;

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

@@ -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<Vec<MediaItem>> {
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<ScheduledSlot>,
fillers: &[MediaItem],
valid_from: DateTime<Utc>,
valid_until: DateTime<Utc>,
) {
if fillers.is_empty() {
return;
}
let gap_block_id = BlockId::generate();
let mut gap_slots: Vec<ScheduledSlot> = Vec::new();
let mut filler_idx = 0usize;
let mut boundaries: Vec<(DateTime<Utc>, DateTime<Utc>)> = 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 {

135
crates/domain/src/services/schedule/tests/integration.rs
View File

@@ -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<ProgrammingBlock>,
gap_filler: Option<MediaFilter>,
) -> 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"
);
}

15
crates/mcp/src/server.rs
View File

@@ -47,6 +47,7 @@ pub struct UpdateChannelParams {
pub timezone: Option<String>,
pub description: Option<String>,
pub schedule_config_json: Option<String>,
pub gap_filler_json: Option<String>,
}
#[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
}

2
crates/mcp/src/tools/channels.rs
View File

@@ -53,6 +53,7 @@ pub async fn update_channel(
timezone: Option<String>,
description: Option<String>,
schedule_config: Option<domain::ScheduleConfig>,
gap_filler: Option<Option<domain::MediaFilter>>,
) -> 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),

1
crates/presentation/src/handlers/channels.rs
View File

@@ -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)))