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

extract background tasks into worker binary

Browse Source 
Move auto_scheduler, library_sync, broadcast_poller, webhook_consumer
from presentation/background/ into crates/worker/src/jobs/.
Presentation is now a pure HTTP server.
This commit is contained in:
Gabriel Kaszewski
2026-07-12 07:28:17 +02:00
parent 826e824b58
commit abcf69ce7e
13 changed files with 521 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/presentation/Cargo.toml
View File

@@ -45,8 +45,6 @@ thiserror = { workspace = true }
async-trait = { workspace = true }
anyhow = "1"
dotenvy = "0.15"
reqwest = { workspace = true }
handlebars = "6"
# Local-files streaming
tokio-util = { version = "0.7", features = ["io"], optional = true }

79
crates/presentation/src/background/auto_scheduler.rs
View File

@@ -1,79 +0,0 @@
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use application::schedule::{GenerateScheduleCommand, ScheduleDeps};
const TICK_INTERVAL_SECS: u64 = 3600;
const EXPIRY_THRESHOLD_HOURS: i64 = 24;
pub async fn run(deps: Arc<ScheduleDeps>) {
loop {
tokio::time::sleep(Duration::from_secs(TICK_INTERVAL_SECS)).await;
tick(&deps).await;
}
}
async fn tick(deps: &ScheduleDeps) {
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;
}
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(EXPIRY_THRESHOLD_HOURS)
}
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
);
}
}
}
}

114
crates/presentation/src/background/broadcast_poller.rs
View File

@@ -1,114 +0,0 @@
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;
const POLL_INTERVAL_SECS: u64 = 1;
struct ChannelPollState {
last_slot_id: Option<Uuid>,
last_checked: Instant,
}
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(POLL_INTERVAL_SECS)).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 &current_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;
}
}

116
crates/presentation/src/background/library_sync.rs
View File

@@ -1,116 +0,0 @@
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;
const SYNC_INTERVAL_SETTING_KEY: &str = "library_sync_interval_hours";
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(SYNC_INTERVAL_SETTING_KEY)
.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 {
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()
);
}
}
}
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
}
}

4
crates/presentation/src/background/mod.rs
View File

@@ -1,4 +0,0 @@
pub mod auto_scheduler;
pub mod broadcast_poller;
pub mod library_sync;
pub mod webhook_consumer;

207
crates/presentation/src/background/webhook_consumer.rs
View File

@@ -1,207 +0,0 @@
use std::sync::Arc;
use chrono::Utc;
use handlebars::Handlebars;
use serde_json::{Value, json};
use uuid::Uuid;
use domain::events::DomainEvent;
use domain::ports::{ChannelQuery, EventConsumer};
const DEFAULT_CONTENT_TYPE: &str = "application/json";
const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
pub async fn run(
consumer: Arc<dyn EventConsumer>,
channel_query: Arc<dyn ChannelQuery>,
client: reqwest::Client,
) {
loop {
match consumer.poll_next().await {
Ok(Some(envelope)) => {
let event_id = envelope.id();
let event = envelope.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
);
}
}
if let Err(e) = consumer.ack(event_id).await {
tracing::warn!("webhook consumer: ack failed for event {}: {}", event_id, e);
}
}
Ok(None) => {
tokio::time::sleep(POLL_INTERVAL).await;
}
Err(e) => {
tracing::warn!("webhook consumer: poll error: {}", e);
tokio::time::sleep(POLL_INTERVAL).await;
}
}
}
}
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.eq_ignore_ascii_case("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", DEFAULT_CONTENT_TYPE);
}
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);
}
}
}

38
crates/presentation/src/factory.rs
View File

@@ -33,9 +33,7 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
DbPool::Sqlite(p) => p.clone(),
};
let event_publisher: Arc<dyn domain::ports::EventPublisher> =
Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool.clone()));
let event_consumer: Arc<dyn domain::ports::EventConsumer> =
Arc::new(adapter_event_publisher::SqliteEventConsumer::new(sqlite_pool));
Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool));
let provider_registry = build_provider_registry(&config).await;
@@ -62,8 +60,6 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
Arc::new(adapter_auth::JwtTokenService::new(validator))
};
let (sync_tx, sync_rx) = tokio::sync::watch::channel(());
let auth_deps = Arc::new(AuthDeps {
user_command: wire_output.user_command.clone(),
user_query: wire_output.user_query.clone(),
@@ -129,34 +125,6 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
let config_arc = Arc::new(config);
let bg_schedule_deps = schedule_deps.clone();
tokio::spawn(crate::background::auto_scheduler::run(bg_schedule_deps));
let bg_schedule_deps2 = schedule_deps.clone();
let bg_event_publisher = event_publisher.clone();
tokio::spawn(crate::background::broadcast_poller::run(
bg_schedule_deps2,
bg_event_publisher,
));
let webhook_consumer = event_consumer.clone();
let webhook_channel_query = wire_output.channel_query.clone();
tokio::spawn(crate::background::webhook_consumer::run(
webhook_consumer,
webhook_channel_query,
reqwest::Client::new(),
));
let bg_sync = library_sync.clone();
let bg_registry = provider_registry.clone();
let bg_settings = wire_output.settings.clone();
tokio::spawn(crate::background::library_sync::run(
bg_sync,
bg_registry,
bg_settings,
sync_rx,
));
Ok(AppState {
auth_deps,
channel_command_deps,
@@ -176,11 +144,7 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
provider_config_command: wire_output.provider_config_command.clone(),
#[cfg(feature = "auth-jwt")]
jwt_validator,
_library_sync: library_sync,
_event_publisher: event_publisher,
_event_consumer: event_consumer,
config: config_arc,
_sync_trigger: sync_tx,
})
}

1
crates/presentation/src/main.rs
View File

@@ -4,7 +4,6 @@ use tower_http::cors::{Any, CorsLayer};
use tower_http::trace::TraceLayer;
use tracing::info;
mod background;
mod errors;
mod extractors;
mod factory;

5
crates/presentation/src/state.rs
View File

@@ -35,10 +35,5 @@ pub struct AppState {
#[cfg(feature = "auth-jwt")]
pub jwt_validator: Option<Arc<adapter_auth::JwtValidator>>,
pub _library_sync: Arc<dyn domain::ports::LibrarySyncAdapter>,
pub _event_publisher: Arc<dyn domain::ports::EventPublisher>,
pub _event_consumer: Arc<dyn domain::ports::EventConsumer>,
pub config: Arc<infra_wiring::Config>,
pub _sync_trigger: tokio::sync::watch::Sender<()>,
}