From abcf69ce7e5afbd470f73b3edc3d1adb5146887d Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 07:28:17 +0200 Subject: [PATCH] extract background tasks into worker binary Move auto_scheduler, library_sync, broadcast_poller, webhook_consumer from presentation/background/ into crates/worker/src/jobs/. Presentation is now a pure HTTP server. --- Cargo.lock | 27 +- Cargo.toml | 2 +- crates/presentation/Cargo.toml | 2 - crates/presentation/src/factory.rs | 38 +- crates/presentation/src/main.rs | 1 - crates/presentation/src/state.rs | 5 - crates/worker/Cargo.toml | 35 ++ .../src/jobs}/auto_scheduler.rs | 0 .../src/jobs}/broadcast_poller.rs | 0 .../src/jobs}/library_sync.rs | 0 .../src/background => worker/src/jobs}/mod.rs | 0 .../src/jobs}/webhook_consumer.rs | 0 crates/worker/src/main.rs | 459 ++++++++++++++++++ 13 files changed, 521 insertions(+), 48 deletions(-) create mode 100644 crates/worker/Cargo.toml rename crates/{presentation/src/background => worker/src/jobs}/auto_scheduler.rs (100%) rename crates/{presentation/src/background => worker/src/jobs}/broadcast_poller.rs (100%) rename crates/{presentation/src/background => worker/src/jobs}/library_sync.rs (100%) rename crates/{presentation/src/background => worker/src/jobs}/mod.rs (100%) rename crates/{presentation/src/background => worker/src/jobs}/webhook_consumer.rs (100%) create mode 100644 crates/worker/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index d1f73c2..9a62af3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1808,9 +1808,7 @@ dependencies = [ "chrono", "domain", "dotenvy", - "handlebars", "infra-wiring", - "reqwest", "serde", "serde_json", "thiserror", @@ -3369,6 +3367,31 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "worker" +version = "0.1.0" +dependencies = [ + "adapter-auth", + "adapter-event-publisher", + "adapter-jellyfin", + "adapter-local-files", + "adapter-sqlite", + "anyhow", + "application", + "async-trait", + "chrono", + "domain", + "dotenvy", + "handlebars", + "infra-wiring", + "reqwest", + "serde_json", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "writeable" version = "0.6.3" diff --git a/Cargo.toml b/Cargo.toml index fbbcdfb..40654ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation", "crates/mcp"] +members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation", "crates/worker", "crates/mcp"] exclude = ["k-tv-backend", "k-tv-frontend"] resolver = "2" diff --git a/crates/presentation/Cargo.toml b/crates/presentation/Cargo.toml index a8caefa..3a44d0a 100644 --- a/crates/presentation/Cargo.toml +++ b/crates/presentation/Cargo.toml @@ -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 } diff --git a/crates/presentation/src/factory.rs b/crates/presentation/src/factory.rs index 8d5b2e4..d6cf3c5 100644 --- a/crates/presentation/src/factory.rs +++ b/crates/presentation/src/factory.rs @@ -33,9 +33,7 @@ pub async fn build_app_state(config: Config) -> anyhow::Result { DbPool::Sqlite(p) => p.clone(), }; let event_publisher: Arc = - Arc::new(adapter_event_publisher::SqliteEventPublisher::new(sqlite_pool.clone())); - let event_consumer: Arc = - 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 { 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 { 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 { 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, }) } diff --git a/crates/presentation/src/main.rs b/crates/presentation/src/main.rs index 1d1460d..1088768 100644 --- a/crates/presentation/src/main.rs +++ b/crates/presentation/src/main.rs @@ -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; diff --git a/crates/presentation/src/state.rs b/crates/presentation/src/state.rs index e506177..fbed038 100644 --- a/crates/presentation/src/state.rs +++ b/crates/presentation/src/state.rs @@ -35,10 +35,5 @@ pub struct AppState { #[cfg(feature = "auth-jwt")] pub jwt_validator: Option>, - pub _library_sync: Arc, - pub _event_publisher: Arc, - pub _event_consumer: Arc, - pub config: Arc, - pub _sync_trigger: tokio::sync::watch::Sender<()>, } diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml new file mode 100644 index 0000000..818d1c3 --- /dev/null +++ b/crates/worker/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "worker" +version = "0.1.0" +edition = "2024" + +[[bin]] +name = "k-tv-worker" +path = "src/main.rs" + +[features] +default = ["sqlite", "jellyfin"] +sqlite = ["dep:adapter-sqlite", "infra-wiring/sqlite"] +jellyfin = ["dep:adapter-jellyfin"] +local-files = ["dep:adapter-local-files"] + +[dependencies] +domain = { workspace = true } +application = { workspace = true } +infra-wiring = { workspace = true } +adapter-sqlite = { workspace = true, optional = true } +adapter-auth = { workspace = true } +adapter-jellyfin = { workspace = true, optional = true } +adapter-local-files = { workspace = true, optional = true } +adapter-event-publisher = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +reqwest = { workspace = true } +serde_json = { workspace = true } +chrono = { workspace = true } +uuid = { workspace = true } +async-trait = { workspace = true } +handlebars = "6" +dotenvy = "0.15" +anyhow = "1" diff --git a/crates/presentation/src/background/auto_scheduler.rs b/crates/worker/src/jobs/auto_scheduler.rs similarity index 100% rename from crates/presentation/src/background/auto_scheduler.rs rename to crates/worker/src/jobs/auto_scheduler.rs diff --git a/crates/presentation/src/background/broadcast_poller.rs b/crates/worker/src/jobs/broadcast_poller.rs similarity index 100% rename from crates/presentation/src/background/broadcast_poller.rs rename to crates/worker/src/jobs/broadcast_poller.rs diff --git a/crates/presentation/src/background/library_sync.rs b/crates/worker/src/jobs/library_sync.rs similarity index 100% rename from crates/presentation/src/background/library_sync.rs rename to crates/worker/src/jobs/library_sync.rs diff --git a/crates/presentation/src/background/mod.rs b/crates/worker/src/jobs/mod.rs similarity index 100% rename from crates/presentation/src/background/mod.rs rename to crates/worker/src/jobs/mod.rs diff --git a/crates/presentation/src/background/webhook_consumer.rs b/crates/worker/src/jobs/webhook_consumer.rs similarity index 100% rename from crates/presentation/src/background/webhook_consumer.rs rename to crates/worker/src/jobs/webhook_consumer.rs diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs new file mode 100644 index 0000000..e853100 --- /dev/null +++ b/crates/worker/src/main.rs @@ -0,0 +1,459 @@ +use std::sync::Arc; + +use application::schedule::ScheduleDeps; +use domain::ports::{IMediaProvider, IProviderRegistry, ProviderCapabilities, StreamingProtocol}; +use domain::{DomainError, ScheduleEngineService}; +use infra_wiring::{Config, ConfigSource, DbPool}; + +mod jobs; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let _ = dotenvy::dotenv(); + + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "info".into()), + ) + .init(); + + let config = Config::from_env().map_err(|e| anyhow::anyhow!("Config error: {}", e))?; + + tracing::info!("Starting k-tv worker"); + + let pool = DbPool::connect(&config.database_url).await?; + pool.run_migrations().await?; + + let wire = wire_repositories(&pool)?; + + let sqlite_pool = match &pool { + #[cfg(feature = "sqlite")] + DbPool::Sqlite(p) => p.clone(), + }; + + let event_publisher: Arc = + Arc::new(adapter_event_publisher::SqliteEventPublisher::new( + sqlite_pool.clone(), + )); + let event_consumer: Arc = + Arc::new(adapter_event_publisher::SqliteEventConsumer::new(sqlite_pool)); + + let provider_registry = build_provider_registry(&config).await; + + let library_sync: Arc = + build_library_sync(wire.library_command.clone()); + + let schedule_engine = Arc::new(ScheduleEngineService::new( + wire.library_query.clone(), + wire.channel_query.clone(), + wire.schedule_query.clone(), + wire.schedule_command.clone(), + )); + + let (_sync_tx, sync_rx) = tokio::sync::watch::channel(()); + + let schedule_deps = Arc::new(ScheduleDeps { + schedule_engine, + channel_query: wire.channel_query.clone(), + schedule_query: wire.schedule_query.clone(), + schedule_command: wire.schedule_command.clone(), + event_publisher: event_publisher.clone(), + provider_registry: provider_registry.clone(), + }); + + tokio::spawn(jobs::auto_scheduler::run(schedule_deps.clone())); + + tokio::spawn(jobs::broadcast_poller::run( + schedule_deps.clone(), + event_publisher.clone(), + )); + + tokio::spawn(jobs::webhook_consumer::run( + event_consumer, + wire.channel_query.clone(), + reqwest::Client::new(), + )); + + tokio::spawn(jobs::library_sync::run( + library_sync, + provider_registry, + wire.settings.clone(), + sync_rx, + )); + + tracing::info!("All jobs spawned, waiting for shutdown signal"); + tokio::signal::ctrl_c().await?; + tracing::info!("Shutting down worker"); + + Ok(()) +} + +struct WireOutput { + channel_query: Arc, + schedule_command: Arc, + schedule_query: Arc, + library_command: Arc, + library_query: Arc, + settings: Arc, +} + +fn wire_repositories(pool: &DbPool) -> anyhow::Result { + match pool { + #[cfg(feature = "sqlite")] + DbPool::Sqlite(sqlite_pool) => { + let w = adapter_sqlite::wire(sqlite_pool.clone()); + Ok(WireOutput { + channel_query: w.channel_query, + schedule_command: w.schedule_command, + schedule_query: w.schedule_query, + library_command: w.library_command, + library_query: w.library_query, + settings: w.settings, + }) + } + } +} + +async fn build_provider_registry(config: &Config) -> Arc { + let mut providers: Vec<(String, Arc)> = Vec::new(); + + match config.config_source { + ConfigSource::Env => { + #[cfg(feature = "jellyfin")] + if let (Some(url), Some(api_key), Some(user_id)) = ( + &config.jellyfin_url, + &config.jellyfin_api_key, + &config.jellyfin_user_id, + ) { + tracing::info!("Media provider: Jellyfin at {}", url); + providers.push(( + "jellyfin".to_string(), + Arc::new(adapter_jellyfin::JellyfinMediaProvider::new( + adapter_jellyfin::JellyfinConfig { + base_url: url.clone(), + api_key: api_key.clone(), + user_id: user_id.clone(), + }, + )), + )); + } + } + ConfigSource::Db => { + tracing::info!("CONFIG_SOURCE=db: provider configs loaded from database at runtime"); + } + } + + if providers.is_empty() { + tracing::warn!("No media provider configured — using NoopMediaProvider"); + providers.push(("noop".to_string(), Arc::new(NoopMediaProvider))); + } + + Arc::new(SimpleProviderRegistry::new(providers)) +} + +fn build_library_sync( + library_command: Arc, +) -> Arc { + Arc::new(SimpleSyncAdapter::new(library_command)) +} + +struct NoopMediaProvider; + +#[async_trait::async_trait] +impl IMediaProvider for NoopMediaProvider { + fn capabilities(&self) -> ProviderCapabilities { + ProviderCapabilities { + collections: false, + series: false, + genres: false, + tags: false, + decade: false, + search: false, + streaming_protocol: StreamingProtocol::DirectFile, + rescan: false, + transcode: false, + } + } + + async fn fetch_items( + &self, + _: &domain::MediaFilter, + ) -> domain::DomainResult> { + Err(DomainError::InfrastructureError( + "No media provider configured. Set JELLYFIN_BASE_URL or LOCAL_FILES_DIR.".into(), + )) + } + + async fn fetch_by_id( + &self, + _: &domain::MediaItemId, + ) -> domain::DomainResult> { + Err(DomainError::InfrastructureError( + "No media provider configured.".into(), + )) + } + + async fn get_stream_url( + &self, + _: &domain::MediaItemId, + _: &domain::ports::StreamQuality, + ) -> domain::DomainResult { + Err(DomainError::InfrastructureError( + "No media provider configured.".into(), + )) + } +} + +struct SimpleProviderRegistry { + providers: Vec<(String, Arc)>, +} + +impl SimpleProviderRegistry { + fn new(providers: Vec<(String, Arc)>) -> Self { + Self { providers } + } + + fn get(&self, id: &str) -> Option<&Arc> { + self.providers.iter().find(|(k, _)| k == id).map(|(_, v)| v) + } + + fn primary(&self) -> Option<&Arc> { + self.providers.first().map(|(_, v)| v) + } + + fn extract_provider_id(item_id: &str) -> Option<&str> { + item_id.find("::").map(|pos| &item_id[..pos]) + } +} + +#[async_trait::async_trait] +impl IProviderRegistry for SimpleProviderRegistry { + async fn fetch_items( + &self, + provider_id: &str, + filter: &domain::MediaFilter, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.providers.first().map(|(k, _)| k.as_str()).unwrap_or("") + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.fetch_items(filter).await + } + + async fn fetch_by_id( + &self, + item_id: &domain::MediaItemId, + ) -> domain::DomainResult> { + let id_str = item_id.value(); + if let Some(pid) = Self::extract_provider_id(id_str) { + if let Some(provider) = self.get(pid) { + return provider.fetch_by_id(item_id).await; + } + } + if let Some(provider) = self.primary() { + provider.fetch_by_id(item_id).await + } else { + Ok(None) + } + } + + async fn get_stream_url( + &self, + item_id: &domain::MediaItemId, + quality: &domain::ports::StreamQuality, + ) -> domain::DomainResult { + let id_str = item_id.value(); + if let Some(pid) = Self::extract_provider_id(id_str) { + if let Some(provider) = self.get(pid) { + return provider.get_stream_url(item_id, quality).await; + } + } + if let Some(provider) = self.primary() { + provider.get_stream_url(item_id, quality).await + } else { + Err(DomainError::InfrastructureError( + "No provider available".into(), + )) + } + } + + fn provider_ids(&self) -> Vec { + self.providers.iter().map(|(k, _)| k.clone()).collect() + } + + fn primary_id(&self) -> &str { + self.providers + .first() + .map(|(k, _)| k.as_str()) + .unwrap_or("") + } + + fn capabilities(&self, provider_id: &str) -> Option { + self.get(provider_id).map(|p| p.capabilities()) + } + + async fn list_collections( + &self, + provider_id: &str, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.primary_id() + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.list_collections().await + } + + async fn list_series( + &self, + provider_id: &str, + collection_id: Option<&str>, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.primary_id() + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.list_series(collection_id).await + } + + async fn list_genres( + &self, + provider_id: &str, + content_type: Option<&domain::ContentType>, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.primary_id() + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.list_genres(content_type).await + } +} + +fn provider_item_to_library_item(item: domain::MediaItem, provider_id: &str) -> domain::MediaItem { + let external_id = item.id().value().to_string(); + let now = chrono::Utc::now().to_rfc3339(); + + domain::MediaItem::from_persistence(domain::MediaItemRow { + id: domain::MediaItemId::new(format!("{}::{}", provider_id, external_id)), + provider_id: provider_id.to_string(), + external_id, + title: item.title().to_string(), + content_type: item.content_type().clone(), + duration_secs: item.duration_secs(), + description: item.description().map(|s| s.to_string()), + series_name: item.series_name().map(|s| s.to_string()), + season_number: item.season_number(), + episode_number: item.episode_number(), + year: item.year(), + genres: item.genres().to_vec(), + tags: item.tags().to_vec(), + collection_id: item.collection_id().map(|s| s.to_string()), + collection_name: None, + collection_type: None, + thumbnail_url: item.thumbnail_url().map(|s| s.to_string()), + synced_at: Some(now), + role: domain::MediaRole::default(), + }) +} + +struct SimpleSyncAdapter { + library_command: Arc, +} + +impl SimpleSyncAdapter { + fn new(library_command: Arc) -> Self { + Self { library_command } + } +} + +#[async_trait::async_trait] +impl domain::ports::LibrarySyncAdapter for SimpleSyncAdapter { + async fn sync_provider( + &self, + provider: &dyn IMediaProvider, + provider_id: &str, + ) -> domain::LibrarySyncResult { + use std::time::Instant; + + let start = Instant::now(); + let log_id = match self.library_command.log_sync_start(provider_id).await { + Ok(id) => id, + Err(e) => { + return domain::LibrarySyncResult::with_error( + provider_id, + 0, + format!("Failed to log sync start: {e}"), + ); + } + }; + + let filter = domain::MediaFilter::default(); + let items = match provider.fetch_items(&filter).await { + Ok(items) => items, + Err(e) => { + let result = domain::LibrarySyncResult::with_error( + provider_id, + start.elapsed().as_millis() as u64, + format!("Failed to fetch items: {e}"), + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + return result; + } + }; + + let items_found = items.len() as u32; + + if let Err(e) = self.library_command.clear_provider(provider_id).await { + let result = domain::LibrarySyncResult::with_error( + provider_id, + start.elapsed().as_millis() as u64, + format!("Failed to clear provider items: {e}"), + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + return result; + } + + let library_items: Vec = items + .into_iter() + .map(|item| provider_item_to_library_item(item, provider_id)) + .collect(); + + if let Err(e) = self + .library_command + .upsert_items(provider_id, library_items) + .await + { + let result = domain::LibrarySyncResult::with_error( + provider_id, + start.elapsed().as_millis() as u64, + format!("Failed to upsert items: {e}"), + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + return result; + } + + let result = domain::LibrarySyncResult::new( + provider_id, + items_found, + start.elapsed().as_millis() as u64, + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + result + } +}