feat(channel): add auto-schedule feature to channels with background scheduler

This commit is contained in:
2026-03-13 02:27:27 +01:00
parent dfd8f52a53
commit 1fc473342d
14 changed files with 161 additions and 6 deletions

View File

@@ -69,6 +69,7 @@ pub struct UpdateChannelRequest {
/// Replace the entire schedule config (template import/edit) /// Replace the entire schedule config (template import/edit)
pub schedule_config: Option<domain::ScheduleConfig>, pub schedule_config: Option<domain::ScheduleConfig>,
pub recycle_policy: Option<domain::RecyclePolicy>, pub recycle_policy: Option<domain::RecyclePolicy>,
pub auto_schedule: Option<bool>,
} }
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
@@ -80,6 +81,7 @@ pub struct ChannelResponse {
pub timezone: String, pub timezone: String,
pub schedule_config: domain::ScheduleConfig, pub schedule_config: domain::ScheduleConfig,
pub recycle_policy: domain::RecyclePolicy, pub recycle_policy: domain::RecyclePolicy,
pub auto_schedule: bool,
pub created_at: DateTime<Utc>, pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>, pub updated_at: DateTime<Utc>,
} }
@@ -94,6 +96,7 @@ impl From<domain::Channel> for ChannelResponse {
timezone: c.timezone, timezone: c.timezone,
schedule_config: c.schedule_config, schedule_config: c.schedule_config,
recycle_policy: c.recycle_policy, recycle_policy: c.recycle_policy,
auto_schedule: c.auto_schedule,
created_at: c.created_at, created_at: c.created_at,
updated_at: c.updated_at, updated_at: c.updated_at,
} }

View File

@@ -21,6 +21,7 @@ mod dto;
mod error; mod error;
mod extractors; mod extractors;
mod routes; mod routes;
mod scheduler;
mod state; mod state;
use crate::config::Config; use crate::config::Config;
@@ -72,6 +73,7 @@ async fn main() -> anyhow::Result<()> {
// Build media provider — Jellyfin if configured, no-op fallback otherwise. // Build media provider — Jellyfin if configured, no-op fallback otherwise.
let media_provider: Arc<dyn IMediaProvider> = build_media_provider(&config); let media_provider: Arc<dyn IMediaProvider> = build_media_provider(&config);
let bg_channel_repo = channel_repo.clone();
let schedule_engine = ScheduleEngineService::new( let schedule_engine = ScheduleEngineService::new(
Arc::clone(&media_provider), Arc::clone(&media_provider),
channel_repo, channel_repo,
@@ -91,6 +93,9 @@ async fn main() -> anyhow::Result<()> {
cors_origins: config.cors_allowed_origins.clone(), cors_origins: config.cors_allowed_origins.clone(),
}; };
let bg_schedule_engine = Arc::clone(&state.schedule_engine);
tokio::spawn(scheduler::run_auto_scheduler(bg_schedule_engine, bg_channel_repo));
let app = Router::new() let app = Router::new()
.nest("/api/v1", routes::api_v1_router()) .nest("/api/v1", routes::api_v1_router())
.with_state(state); .with_state(state);

View File

@@ -76,6 +76,9 @@ pub(super) async fn update_channel(
if let Some(rp) = payload.recycle_policy { if let Some(rp) = payload.recycle_policy {
channel.recycle_policy = rp; channel.recycle_policy = rp;
} }
if let Some(auto) = payload.auto_schedule {
channel.auto_schedule = auto;
}
channel.updated_at = Utc::now(); channel.updated_at = Utc::now();
let channel = state.channel_service.update(channel).await?; let channel = state.channel_service.update(channel).await?;

View File

@@ -0,0 +1,76 @@
//! Background auto-scheduler task.
//!
//! Runs every hour, finds channels with `auto_schedule = true`, and regenerates
//! their schedule if it is within 24 hours of expiry (or already expired).
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use domain::{ChannelRepository, ScheduleEngineService};
pub async fn run_auto_scheduler(
schedule_engine: Arc<ScheduleEngineService>,
channel_repo: Arc<dyn ChannelRepository>,
) {
loop {
tokio::time::sleep(Duration::from_secs(3600)).await;
tick(&schedule_engine, &channel_repo).await;
}
}
async fn tick(
schedule_engine: &Arc<ScheduleEngineService>,
channel_repo: &Arc<dyn ChannelRepository>,
) {
let channels = match channel_repo.find_auto_schedule_enabled().await {
Ok(c) => c,
Err(e) => {
tracing::warn!("auto-scheduler: failed to fetch channels: {}", e);
return;
}
};
let now = Utc::now();
for channel in channels {
let from = match schedule_engine.get_latest_schedule(channel.id).await {
Ok(Some(s)) => {
let remaining = s.valid_until - now;
if remaining > chrono::Duration::hours(24) {
// Still fresh — skip until it gets close to expiry
continue;
} else if s.valid_until > now {
// Seamless handoff: new schedule starts where the old one ends
s.valid_until
} else {
// Expired: start from now to avoid scheduling in the past
now
}
}
Ok(None) => now,
Err(e) => {
tracing::warn!(
"auto-scheduler: failed to fetch latest schedule for channel {}: {}",
channel.id,
e
);
continue;
}
};
if let Err(e) = schedule_engine.generate_schedule(channel.id, from).await {
tracing::warn!(
"auto-scheduler: failed to generate schedule for channel {}: {}",
channel.id,
e
);
} else {
tracing::info!(
"auto-scheduler: generated schedule for channel {} starting at {}",
channel.id,
from
);
}
}
}

View File

@@ -81,6 +81,7 @@ pub struct Channel {
pub timezone: String, pub timezone: String,
pub schedule_config: ScheduleConfig, pub schedule_config: ScheduleConfig,
pub recycle_policy: RecyclePolicy, pub recycle_policy: RecyclePolicy,
pub auto_schedule: bool,
pub created_at: DateTime<Utc>, pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>, pub updated_at: DateTime<Utc>,
} }
@@ -100,6 +101,7 @@ impl Channel {
timezone: timezone.into(), timezone: timezone.into(),
schedule_config: ScheduleConfig::default(), schedule_config: ScheduleConfig::default(),
recycle_policy: RecyclePolicy::default(), recycle_policy: RecyclePolicy::default(),
auto_schedule: false,
created_at: now, created_at: now,
updated_at: now, updated_at: now,
} }

View File

@@ -37,6 +37,7 @@ pub trait ChannelRepository: Send + Sync {
async fn find_by_id(&self, id: ChannelId) -> DomainResult<Option<Channel>>; async fn find_by_id(&self, id: ChannelId) -> DomainResult<Option<Channel>>;
async fn find_by_owner(&self, owner_id: UserId) -> DomainResult<Vec<Channel>>; async fn find_by_owner(&self, owner_id: UserId) -> DomainResult<Vec<Channel>>;
async fn find_all(&self) -> DomainResult<Vec<Channel>>; async fn find_all(&self) -> DomainResult<Vec<Channel>>;
async fn find_auto_schedule_enabled(&self) -> DomainResult<Vec<Channel>>;
/// Insert or update a channel. /// Insert or update a channel.
async fn save(&self, channel: &Channel) -> DomainResult<()>; async fn save(&self, channel: &Channel) -> DomainResult<()>;
async fn delete(&self, id: ChannelId) -> DomainResult<()>; async fn delete(&self, id: ChannelId) -> DomainResult<()>;

View File

@@ -206,6 +206,14 @@ 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,
) -> DomainResult<Option<GeneratedSchedule>> {
self.schedule_repo.find_latest(channel_id).await
}
/// Look up the schedule currently active at `at` without generating a new one. /// Look up the schedule currently active at `at` without generating a new one.
pub async fn get_active_schedule( pub async fn get_active_schedule(
&self, &self,

View File

@@ -13,6 +13,7 @@ pub(super) struct ChannelRow {
pub timezone: String, pub timezone: String,
pub schedule_config: String, pub schedule_config: String,
pub recycle_policy: String, pub recycle_policy: String,
pub auto_schedule: i64,
pub created_at: String, pub created_at: String,
pub updated_at: String, pub updated_at: String,
} }
@@ -51,6 +52,7 @@ impl TryFrom<ChannelRow> for Channel {
timezone: row.timezone, timezone: row.timezone,
schedule_config, schedule_config,
recycle_policy, recycle_policy,
auto_schedule: row.auto_schedule != 0,
created_at: parse_dt(&row.created_at)?, created_at: parse_dt(&row.created_at)?,
updated_at: parse_dt(&row.updated_at)?, updated_at: parse_dt(&row.updated_at)?,
}) })
@@ -58,4 +60,4 @@ impl TryFrom<ChannelRow> for Channel {
} }
pub(super) const SELECT_COLS: &str = pub(super) const SELECT_COLS: &str =
"id, owner_id, name, description, timezone, schedule_config, recycle_policy, created_at, updated_at"; "id, owner_id, name, description, timezone, schedule_config, recycle_policy, auto_schedule, created_at, updated_at";

View File

@@ -61,14 +61,15 @@ impl ChannelRepository for PostgresChannelRepository {
sqlx::query( sqlx::query(
r#" r#"
INSERT INTO channels INSERT INTO channels
(id, owner_id, name, description, timezone, schedule_config, recycle_policy, created_at, updated_at) (id, owner_id, name, description, timezone, schedule_config, recycle_policy, auto_schedule, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT(id) DO UPDATE SET ON CONFLICT(id) DO UPDATE SET
name = EXCLUDED.name, name = EXCLUDED.name,
description = EXCLUDED.description, description = EXCLUDED.description,
timezone = EXCLUDED.timezone, timezone = EXCLUDED.timezone,
schedule_config = EXCLUDED.schedule_config, schedule_config = EXCLUDED.schedule_config,
recycle_policy = EXCLUDED.recycle_policy, recycle_policy = EXCLUDED.recycle_policy,
auto_schedule = EXCLUDED.auto_schedule,
updated_at = EXCLUDED.updated_at updated_at = EXCLUDED.updated_at
"#, "#,
) )
@@ -79,6 +80,7 @@ impl ChannelRepository for PostgresChannelRepository {
.bind(&channel.timezone) .bind(&channel.timezone)
.bind(&schedule_config) .bind(&schedule_config)
.bind(&recycle_policy) .bind(&recycle_policy)
.bind(channel.auto_schedule as i64)
.bind(channel.created_at.to_rfc3339()) .bind(channel.created_at.to_rfc3339())
.bind(channel.updated_at.to_rfc3339()) .bind(channel.updated_at.to_rfc3339())
.execute(&self.pool) .execute(&self.pool)
@@ -88,6 +90,18 @@ impl ChannelRepository for PostgresChannelRepository {
Ok(()) Ok(())
} }
async fn find_auto_schedule_enabled(&self) -> DomainResult<Vec<Channel>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM channels WHERE auto_schedule = 1 ORDER BY created_at ASC"
);
let rows: Vec<ChannelRow> = sqlx::query_as(&sql)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::RepositoryError(e.to_string()))?;
rows.into_iter().map(Channel::try_from).collect()
}
async fn delete(&self, id: ChannelId) -> DomainResult<()> { async fn delete(&self, id: ChannelId) -> DomainResult<()> {
sqlx::query("DELETE FROM channels WHERE id = $1") sqlx::query("DELETE FROM channels WHERE id = $1")
.bind(id.to_string()) .bind(id.to_string())

View File

@@ -61,14 +61,15 @@ impl ChannelRepository for SqliteChannelRepository {
sqlx::query( sqlx::query(
r#" r#"
INSERT INTO channels INSERT INTO channels
(id, owner_id, name, description, timezone, schedule_config, recycle_policy, created_at, updated_at) (id, owner_id, name, description, timezone, schedule_config, recycle_policy, auto_schedule, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET ON CONFLICT(id) DO UPDATE SET
name = excluded.name, name = excluded.name,
description = excluded.description, description = excluded.description,
timezone = excluded.timezone, timezone = excluded.timezone,
schedule_config = excluded.schedule_config, schedule_config = excluded.schedule_config,
recycle_policy = excluded.recycle_policy, recycle_policy = excluded.recycle_policy,
auto_schedule = excluded.auto_schedule,
updated_at = excluded.updated_at updated_at = excluded.updated_at
"#, "#,
) )
@@ -79,6 +80,7 @@ impl ChannelRepository for SqliteChannelRepository {
.bind(&channel.timezone) .bind(&channel.timezone)
.bind(&schedule_config) .bind(&schedule_config)
.bind(&recycle_policy) .bind(&recycle_policy)
.bind(channel.auto_schedule as i64)
.bind(channel.created_at.to_rfc3339()) .bind(channel.created_at.to_rfc3339())
.bind(channel.updated_at.to_rfc3339()) .bind(channel.updated_at.to_rfc3339())
.execute(&self.pool) .execute(&self.pool)
@@ -88,6 +90,18 @@ impl ChannelRepository for SqliteChannelRepository {
Ok(()) Ok(())
} }
async fn find_auto_schedule_enabled(&self) -> DomainResult<Vec<Channel>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM channels WHERE auto_schedule = 1 ORDER BY created_at ASC"
);
let rows: Vec<ChannelRow> = sqlx::query_as(&sql)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::RepositoryError(e.to_string()))?;
rows.into_iter().map(Channel::try_from).collect()
}
async fn delete(&self, id: ChannelId) -> DomainResult<()> { async fn delete(&self, id: ChannelId) -> DomainResult<()> {
sqlx::query("DELETE FROM channels WHERE id = ?") sqlx::query("DELETE FROM channels WHERE id = ?")
.bind(id.to_string()) .bind(id.to_string())

View File

@@ -0,0 +1 @@
ALTER TABLE channels ADD COLUMN auto_schedule INTEGER NOT NULL DEFAULT 0;

View File

@@ -72,6 +72,7 @@ const channelFormSchema = z.object({
cooldown_generations: z.number().int().min(0).nullable().optional(), cooldown_generations: z.number().int().min(0).nullable().optional(),
min_available_ratio: z.number().min(0, "Must be ≥ 0").max(1, "Must be ≤ 1"), min_available_ratio: z.number().min(0, "Must be ≥ 0").max(1, "Must be ≤ 1"),
}), }),
auto_schedule: z.boolean(),
}); });
type FieldErrors = Record<string, string | undefined>; type FieldErrors = Record<string, string | undefined>;
@@ -676,6 +677,7 @@ interface EditChannelSheetProps {
timezone: string; timezone: string;
schedule_config: { blocks: ProgrammingBlock[] }; schedule_config: { blocks: ProgrammingBlock[] };
recycle_policy: RecyclePolicy; recycle_policy: RecyclePolicy;
auto_schedule: boolean;
}, },
) => void; ) => void;
isPending: boolean; isPending: boolean;
@@ -699,6 +701,7 @@ export function EditChannelSheet({
cooldown_generations: null, cooldown_generations: null,
min_available_ratio: 0.1, min_available_ratio: 0.1,
}); });
const [autoSchedule, setAutoSchedule] = useState(false);
const [selectedBlockId, setSelectedBlockId] = useState<string | null>(null); const [selectedBlockId, setSelectedBlockId] = useState<string | null>(null);
const [fieldErrors, setFieldErrors] = useState<FieldErrors>({}); const [fieldErrors, setFieldErrors] = useState<FieldErrors>({});
@@ -709,6 +712,7 @@ export function EditChannelSheet({
setTimezone(channel.timezone); setTimezone(channel.timezone);
setBlocks(channel.schedule_config.blocks); setBlocks(channel.schedule_config.blocks);
setRecyclePolicy(channel.recycle_policy); setRecyclePolicy(channel.recycle_policy);
setAutoSchedule(channel.auto_schedule);
setSelectedBlockId(null); setSelectedBlockId(null);
setFieldErrors({}); setFieldErrors({});
} }
@@ -719,7 +723,7 @@ export function EditChannelSheet({
if (!channel) return; if (!channel) return;
const result = channelFormSchema.safeParse({ const result = channelFormSchema.safeParse({
name, description, timezone, blocks, recycle_policy: recyclePolicy, name, description, timezone, blocks, recycle_policy: recyclePolicy, auto_schedule: autoSchedule,
}); });
if (!result.success) { if (!result.success) {
@@ -734,6 +738,7 @@ export function EditChannelSheet({
timezone, timezone,
schedule_config: { blocks }, schedule_config: { blocks },
recycle_policy: recyclePolicy, recycle_policy: recyclePolicy,
auto_schedule: autoSchedule,
}); });
}; };
@@ -799,6 +804,24 @@ export function EditChannelSheet({
className="w-full resize-none rounded-md border border-zinc-700 bg-zinc-800 px-3 py-2 text-sm text-zinc-100 placeholder:text-zinc-600 focus:border-zinc-500 focus:outline-none" className="w-full resize-none rounded-md border border-zinc-700 bg-zinc-800 px-3 py-2 text-sm text-zinc-100 placeholder:text-zinc-600 focus:border-zinc-500 focus:outline-none"
/> />
</Field> </Field>
<label className="flex items-center justify-between gap-3 cursor-pointer rounded-md border border-zinc-700 bg-zinc-800/50 px-3 py-2.5">
<div>
<p className="text-sm text-zinc-200">Auto-generate schedule</p>
<p className="text-[11px] text-zinc-600">Automatically regenerate when the schedule is about to expire</p>
</div>
<button
type="button"
role="switch"
aria-checked={autoSchedule}
onClick={() => setAutoSchedule((v) => !v)}
className={`relative inline-flex h-5 w-9 shrink-0 rounded-full border-2 border-transparent transition-colors focus:outline-none ${autoSchedule ? "bg-zinc-300" : "bg-zinc-700"}`}
>
<span
className={`pointer-events-none inline-block h-4 w-4 rounded-full bg-white shadow-sm transition-transform ${autoSchedule ? "translate-x-4" : "translate-x-0"}`}
/>
</button>
</label>
</section> </section>
{/* Programming blocks */} {/* Programming blocks */}

View File

@@ -122,6 +122,7 @@ export default function DashboardPage() {
timezone: string; timezone: string;
schedule_config: { blocks: ProgrammingBlock[] }; schedule_config: { blocks: ProgrammingBlock[] };
recycle_policy: RecyclePolicy; recycle_policy: RecyclePolicy;
auto_schedule: boolean;
}, },
) => { ) => {
updateChannel.mutate( updateChannel.mutate(

View File

@@ -103,6 +103,7 @@ export interface ChannelResponse {
timezone: string; timezone: string;
schedule_config: ScheduleConfig; schedule_config: ScheduleConfig;
recycle_policy: RecyclePolicy; recycle_policy: RecyclePolicy;
auto_schedule: boolean;
created_at: string; created_at: string;
updated_at: string; updated_at: string;
} }
@@ -119,6 +120,7 @@ export interface UpdateChannelRequest {
timezone?: string; timezone?: string;
schedule_config?: ScheduleConfig; schedule_config?: ScheduleConfig;
recycle_policy?: RecyclePolicy; recycle_policy?: RecyclePolicy;
auto_schedule?: boolean;
} }
// Media & Schedule // Media & Schedule