presentation: HTTP server crate w/ handlers, routes, background tasks
Axum binary that wires all clean-arch crates together: - AppState holds pre-built Deps structs (auth, channels, schedule, library, etc.) - JWT extractors (CurrentUser, AdminUser, OptionalCurrentUser) - Handlers delegate to application use cases, map to api-types DTOs - Routes: auth, channels, schedule, library, admin, providers, config, iptv - Background: auto-scheduler, broadcast poller, webhook consumer, library sync - Factory builds everything from Config + DbPool - SimpleProviderRegistry impl of IProviderRegistry trait - NoopMediaProvider fallback
This commit is contained in:
85
crates/presentation/src/background/auto_scheduler.rs
Normal file
85
crates/presentation/src/background/auto_scheduler.rs
Normal file
@@ -0,0 +1,85 @@
|
||||
//! 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.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::Utc;
|
||||
|
||||
use application::schedule::ScheduleDeps;
|
||||
use application::schedule::GenerateScheduleCommand;
|
||||
|
||||
/// Run the auto-scheduler loop.
|
||||
pub async fn run(deps: Arc<ScheduleDeps>) {
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_secs(3600)).await;
|
||||
tick(&deps).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn tick(deps: &ScheduleDeps) {
|
||||
// List all channels, find those with auto_schedule
|
||||
let channels = match deps.channel_query.find_all().await {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
tracing::warn!("auto-scheduler: failed to fetch channels: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let now = Utc::now();
|
||||
|
||||
for channel in channels {
|
||||
if !channel.auto_schedule() {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check latest schedule
|
||||
let latest = match deps.schedule_query.find_latest(channel.id()).await {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"auto-scheduler: failed to fetch latest schedule for channel {}: {}",
|
||||
channel.id().value(),
|
||||
e
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let should_generate = match &latest {
|
||||
Some(s) => {
|
||||
let remaining = s.valid_until() - now;
|
||||
remaining < chrono::Duration::hours(24)
|
||||
}
|
||||
None => true,
|
||||
};
|
||||
|
||||
if !should_generate {
|
||||
continue;
|
||||
}
|
||||
|
||||
let cmd = GenerateScheduleCommand {
|
||||
channel_id: channel.id().value(),
|
||||
};
|
||||
|
||||
match application::schedule::generate::execute(deps, cmd).await {
|
||||
Ok(schedule) => {
|
||||
tracing::info!(
|
||||
"auto-scheduler: generated schedule for channel {} (gen {})",
|
||||
channel.id().value(),
|
||||
schedule.generation(),
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"auto-scheduler: failed to generate schedule for channel {}: {}",
|
||||
channel.id().value(),
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
119
crates/presentation/src/background/broadcast_poller.rs
Normal file
119
crates/presentation/src/background/broadcast_poller.rs
Normal file
@@ -0,0 +1,119 @@
|
||||
//! BroadcastPoller background task.
|
||||
//!
|
||||
//! Polls channels with webhook_url configured and emits domain events
|
||||
//! when the current slot changes.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use chrono::Utc;
|
||||
use uuid::Uuid;
|
||||
|
||||
use domain::events::DomainEvent;
|
||||
use domain::ports::events::EventPublisher;
|
||||
use domain::value_objects::{ChannelId, SlotId};
|
||||
|
||||
use application::schedule::ScheduleDeps;
|
||||
|
||||
/// Per-channel poll state.
|
||||
struct ChannelPollState {
|
||||
last_slot_id: Option<Uuid>,
|
||||
last_checked: Instant,
|
||||
}
|
||||
|
||||
/// Polls channels and emits broadcast transition events.
|
||||
pub async fn run(deps: Arc<ScheduleDeps>, event_publisher: Arc<dyn EventPublisher>) {
|
||||
let mut state: HashMap<Uuid, ChannelPollState> = HashMap::new();
|
||||
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
tick(&deps, &event_publisher, &mut state).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn tick(
|
||||
deps: &ScheduleDeps,
|
||||
event_publisher: &Arc<dyn EventPublisher>,
|
||||
state: &mut HashMap<Uuid, ChannelPollState>,
|
||||
) {
|
||||
let channels = match deps.channel_query.find_all().await {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
tracing::error!("broadcast poller: failed to load channels: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let live_ids: std::collections::HashSet<Uuid> =
|
||||
channels.iter().map(|c| c.id().value()).collect();
|
||||
state.retain(|id, _| live_ids.contains(id));
|
||||
|
||||
let now = Utc::now();
|
||||
|
||||
for channel in channels {
|
||||
if channel.webhook_url().is_none() {
|
||||
state.remove(&channel.id().value());
|
||||
continue;
|
||||
}
|
||||
|
||||
let poll_interval = Duration::from_secs(channel.webhook_poll_interval_secs() as u64);
|
||||
let channel_uuid = channel.id().value();
|
||||
|
||||
let entry = state.entry(channel_uuid).or_insert(ChannelPollState {
|
||||
last_slot_id: None,
|
||||
last_checked: Instant::now() - poll_interval,
|
||||
});
|
||||
|
||||
if entry.last_checked.elapsed() < poll_interval {
|
||||
continue;
|
||||
}
|
||||
|
||||
entry.last_checked = Instant::now();
|
||||
|
||||
let current_slot_id = match deps
|
||||
.schedule_query
|
||||
.find_active(channel.id(), now)
|
||||
.await
|
||||
{
|
||||
Ok(Some(schedule)) => schedule
|
||||
.slots()
|
||||
.iter()
|
||||
.find(|s| s.start_at() <= now && now < s.end_at())
|
||||
.map(|s| s.id().value()),
|
||||
Ok(None) => None,
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
"broadcast poller: error checking schedule for channel {}: {}",
|
||||
channel_uuid,
|
||||
e
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
if current_slot_id == entry.last_slot_id {
|
||||
continue;
|
||||
}
|
||||
|
||||
match ¤t_slot_id {
|
||||
Some(slot_id) => {
|
||||
let _ = event_publisher
|
||||
.publish(DomainEvent::BroadcastTransition {
|
||||
channel_id: ChannelId::from(channel_uuid),
|
||||
slot_id: SlotId::from(*slot_id),
|
||||
})
|
||||
.await;
|
||||
}
|
||||
None => {
|
||||
let _ = event_publisher
|
||||
.publish(DomainEvent::NoSignal {
|
||||
channel_id: ChannelId::from(channel_uuid),
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
entry.last_slot_id = current_slot_id;
|
||||
}
|
||||
}
|
||||
125
crates/presentation/src/background/library_sync.rs
Normal file
125
crates/presentation/src/background/library_sync.rs
Normal file
@@ -0,0 +1,125 @@
|
||||
//! Background library sync task.
|
||||
//!
|
||||
//! Fires 10 seconds after startup, then every N hours (read from app_settings).
|
||||
//! Can be triggered on-demand via the sync_trigger watch channel.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use domain::ports::{AppSettingsRepository, IProviderRegistry, LibrarySyncAdapter};
|
||||
use tokio::sync::watch;
|
||||
|
||||
const STARTUP_DELAY_SECS: u64 = 10;
|
||||
const DEFAULT_INTERVAL_HOURS: u64 = 6;
|
||||
|
||||
pub async fn run(
|
||||
sync_adapter: Arc<dyn LibrarySyncAdapter>,
|
||||
provider_registry: Arc<dyn IProviderRegistry>,
|
||||
settings_repo: Arc<dyn AppSettingsRepository>,
|
||||
mut trigger_rx: watch::Receiver<()>,
|
||||
) {
|
||||
tokio::time::sleep(Duration::from_secs(STARTUP_DELAY_SECS)).await;
|
||||
|
||||
loop {
|
||||
do_sync(&sync_adapter, &provider_registry).await;
|
||||
|
||||
let interval_hours = load_interval_hours(&settings_repo).await;
|
||||
let sleep = tokio::time::sleep(Duration::from_secs(interval_hours * 3600));
|
||||
|
||||
tokio::select! {
|
||||
_ = sleep => {}
|
||||
_ = trigger_rx.changed() => {
|
||||
tracing::info!("library-sync: triggered manually");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn load_interval_hours(repo: &Arc<dyn AppSettingsRepository>) -> u64 {
|
||||
repo.get("library_sync_interval_hours")
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.and_then(|v| v.parse::<u64>().ok())
|
||||
.unwrap_or(DEFAULT_INTERVAL_HOURS)
|
||||
}
|
||||
|
||||
async fn do_sync(
|
||||
sync_adapter: &Arc<dyn LibrarySyncAdapter>,
|
||||
registry: &Arc<dyn IProviderRegistry>,
|
||||
) {
|
||||
let provider_ids = registry.provider_ids();
|
||||
|
||||
for provider_id in provider_ids {
|
||||
// We need a &dyn IMediaProvider, but IProviderRegistry doesn't expose one.
|
||||
// The sync adapter will use the registry's fetch_items internally via its
|
||||
// own stored reference to the provider. For now, we create a thin adapter.
|
||||
tracing::info!("library-sync: syncing provider '{}'", provider_id);
|
||||
|
||||
let wrapper = RegistryProviderAdapter {
|
||||
registry: registry.clone(),
|
||||
provider_id: provider_id.clone(),
|
||||
};
|
||||
|
||||
let result = sync_adapter.sync_provider(&wrapper, &provider_id).await;
|
||||
|
||||
if let Some(err) = result.error() {
|
||||
tracing::warn!("library-sync: provider '{}' failed: {}", provider_id, err);
|
||||
} else {
|
||||
tracing::info!(
|
||||
"library-sync: provider '{}' done — {} items in {}ms",
|
||||
provider_id,
|
||||
result.items_found(),
|
||||
result.duration_ms()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Adapter that wraps IProviderRegistry calls for a specific provider_id,
|
||||
/// implementing IMediaProvider so it can be passed to LibrarySyncAdapter.
|
||||
struct RegistryProviderAdapter {
|
||||
registry: Arc<dyn IProviderRegistry>,
|
||||
provider_id: String,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl domain::ports::IMediaProvider for RegistryProviderAdapter {
|
||||
fn capabilities(&self) -> domain::ports::ProviderCapabilities {
|
||||
self.registry
|
||||
.capabilities(&self.provider_id)
|
||||
.unwrap_or(domain::ports::ProviderCapabilities {
|
||||
collections: false,
|
||||
series: false,
|
||||
genres: false,
|
||||
tags: false,
|
||||
decade: false,
|
||||
search: false,
|
||||
streaming_protocol: domain::ports::StreamingProtocol::DirectFile,
|
||||
rescan: false,
|
||||
transcode: false,
|
||||
})
|
||||
}
|
||||
|
||||
async fn fetch_items(
|
||||
&self,
|
||||
filter: &domain::MediaFilter,
|
||||
) -> domain::DomainResult<Vec<domain::MediaItem>> {
|
||||
self.registry.fetch_items(&self.provider_id, filter).await
|
||||
}
|
||||
|
||||
async fn fetch_by_id(
|
||||
&self,
|
||||
item_id: &domain::MediaItemId,
|
||||
) -> domain::DomainResult<Option<domain::MediaItem>> {
|
||||
self.registry.fetch_by_id(item_id).await
|
||||
}
|
||||
|
||||
async fn get_stream_url(
|
||||
&self,
|
||||
item_id: &domain::MediaItemId,
|
||||
quality: &domain::ports::StreamQuality,
|
||||
) -> domain::DomainResult<String> {
|
||||
self.registry.get_stream_url(item_id, quality).await
|
||||
}
|
||||
}
|
||||
6
crates/presentation/src/background/mod.rs
Normal file
6
crates/presentation/src/background/mod.rs
Normal file
@@ -0,0 +1,6 @@
|
||||
//! Background tasks spawned at server startup.
|
||||
|
||||
pub mod auto_scheduler;
|
||||
pub mod broadcast_poller;
|
||||
pub mod library_sync;
|
||||
pub mod webhook_consumer;
|
||||
204
crates/presentation/src/background/webhook_consumer.rs
Normal file
204
crates/presentation/src/background/webhook_consumer.rs
Normal file
@@ -0,0 +1,204 @@
|
||||
//! WebhookConsumer background task.
|
||||
//!
|
||||
//! Subscribes to domain events and delivers them to per-channel webhook URLs.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::Utc;
|
||||
use handlebars::Handlebars;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::sync::broadcast;
|
||||
use uuid::Uuid;
|
||||
|
||||
use domain::events::DomainEvent;
|
||||
use domain::ports::ChannelQuery;
|
||||
|
||||
/// Consumes domain events and delivers them to per-channel webhook URLs.
|
||||
pub async fn run(
|
||||
mut rx: broadcast::Receiver<DomainEvent>,
|
||||
channel_query: Arc<dyn ChannelQuery>,
|
||||
client: reqwest::Client,
|
||||
) {
|
||||
loop {
|
||||
match rx.recv().await {
|
||||
Ok(event) => {
|
||||
let channel_id = event_channel_id(&event);
|
||||
let payload = build_payload(&event);
|
||||
|
||||
let channel_id_vo = domain::ChannelId::from(channel_id);
|
||||
match channel_query.find_by_id(channel_id_vo).await {
|
||||
Ok(Some(channel)) => {
|
||||
if let Some(url) = channel.webhook_url() {
|
||||
let url = url.to_string();
|
||||
let client = client.clone();
|
||||
let template = channel.webhook_body_template().map(|s| s.to_string());
|
||||
let headers = channel.webhook_headers().map(|s| s.to_string());
|
||||
tokio::spawn(async move {
|
||||
post_webhook(
|
||||
&client,
|
||||
&url,
|
||||
payload,
|
||||
template.as_deref(),
|
||||
headers.as_deref(),
|
||||
)
|
||||
.await;
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"webhook consumer: failed to look up channel {}: {}",
|
||||
channel_id,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(broadcast::error::RecvError::Lagged(n)) => {
|
||||
tracing::warn!("webhook consumer lagged, {} events dropped", n);
|
||||
}
|
||||
Err(broadcast::error::RecvError::Closed) => {
|
||||
tracing::info!("webhook consumer: event bus closed, shutting down");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn event_channel_id(event: &DomainEvent) -> Uuid {
|
||||
match event {
|
||||
DomainEvent::BroadcastTransition { channel_id, .. } => channel_id.value(),
|
||||
DomainEvent::NoSignal { channel_id } => channel_id.value(),
|
||||
DomainEvent::ScheduleGenerated { channel_id, .. } => channel_id.value(),
|
||||
DomainEvent::ChannelCreated { channel_id } => channel_id.value(),
|
||||
DomainEvent::ChannelUpdated { channel_id } => channel_id.value(),
|
||||
DomainEvent::ChannelDeleted { channel_id } => channel_id.value(),
|
||||
_ => Uuid::nil(),
|
||||
}
|
||||
}
|
||||
|
||||
fn build_payload(event: &DomainEvent) -> Value {
|
||||
let now = Utc::now().to_rfc3339();
|
||||
let channel_id = event_channel_id(event);
|
||||
match event {
|
||||
DomainEvent::BroadcastTransition { slot_id, .. } => {
|
||||
json!({
|
||||
"event": "broadcast_transition",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {
|
||||
"slot_id": slot_id.value(),
|
||||
}
|
||||
})
|
||||
}
|
||||
DomainEvent::NoSignal { .. } => {
|
||||
json!({
|
||||
"event": "no_signal",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {}
|
||||
})
|
||||
}
|
||||
DomainEvent::ScheduleGenerated { schedule_id, .. } => {
|
||||
json!({
|
||||
"event": "schedule_generated",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {
|
||||
"schedule_id": schedule_id.value(),
|
||||
}
|
||||
})
|
||||
}
|
||||
DomainEvent::ChannelCreated { .. } => {
|
||||
json!({
|
||||
"event": "channel_created",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {}
|
||||
})
|
||||
}
|
||||
DomainEvent::ChannelUpdated { .. } => {
|
||||
json!({
|
||||
"event": "channel_updated",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {}
|
||||
})
|
||||
}
|
||||
DomainEvent::ChannelDeleted { .. } => {
|
||||
json!({
|
||||
"event": "channel_deleted",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {}
|
||||
})
|
||||
}
|
||||
_ => {
|
||||
json!({
|
||||
"event": "unknown",
|
||||
"timestamp": now,
|
||||
"channel_id": channel_id,
|
||||
"data": {}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn post_webhook(
|
||||
client: &reqwest::Client,
|
||||
url: &str,
|
||||
payload: Value,
|
||||
template: Option<&str>,
|
||||
headers_json: Option<&str>,
|
||||
) {
|
||||
let body = if let Some(tmpl) = template {
|
||||
let hbs = Handlebars::new();
|
||||
match hbs.render_template(tmpl, &payload) {
|
||||
Ok(rendered) => rendered,
|
||||
Err(e) => {
|
||||
tracing::warn!("webhook template render failed for {}: {}", url, e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
match serde_json::to_string(&payload) {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
tracing::warn!("webhook payload serialize failed: {}", e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let mut req = client.post(url).body(body);
|
||||
let mut has_content_type = false;
|
||||
|
||||
if let Some(h) = headers_json {
|
||||
if let Ok(map) = serde_json::from_str::<serde_json::Map<String, Value>>(h) {
|
||||
for (k, v) in &map {
|
||||
if k.to_lowercase() == "content-type" {
|
||||
has_content_type = true;
|
||||
}
|
||||
if let Some(v_str) = v.as_str() {
|
||||
req = req.header(k.as_str(), v_str);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !has_content_type {
|
||||
req = req.header("Content-Type", "application/json");
|
||||
}
|
||||
|
||||
match req.send().await {
|
||||
Ok(resp) => {
|
||||
if !resp.status().is_success() {
|
||||
tracing::warn!("webhook POST to {} returned status {}", url, resp.status());
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("webhook POST to {} failed: {}", url, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user