use std::sync::Arc; use domain::ports::{IMediaProvider, IProviderRegistry, ProviderCapabilities, StreamingProtocol}; use domain::{DomainError, DomainResult, MediaFilter, MediaItemId, MediaItem, ScheduleEngineService}; use domain::ports::StreamQuality; use infra_wiring::DbPool; use tracing::info; use uuid::Uuid; mod error; mod server; mod tools; use server::KTvMcpServer; #[tokio::main] async fn main() -> anyhow::Result<()> { let _ = dotenvy::dotenv(); tracing_subscriber::fmt() .with_writer(std::io::stderr) .with_env_filter( tracing_subscriber::EnvFilter::from_default_env() .add_directive("mcp=info".parse().unwrap()), ) .init(); let database_url = std::env::var("DATABASE_URL") .unwrap_or_else(|_| "sqlite:data.db?mode=rwc".to_string()); let owner_id: Uuid = std::env::var("MCP_USER_ID") .map_err(|_| anyhow::anyhow!("MCP_USER_ID env var is required (UUID of the user)"))? .parse() .map_err(|_| anyhow::anyhow!("MCP_USER_ID must be a valid UUID"))?; info!("Connecting to database: {}", database_url); let pool = DbPool::connect(&database_url).await?; pool.run_migrations().await?; let wire = wire_repositories(&pool)?; let provider_registry = build_provider_registry().await; let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(64)); let event_publisher: Arc = event_bus.clone(); let schedule_engine = Arc::new(ScheduleEngineService::new( provider_registry.clone(), wire.channel_query.clone(), wire.schedule_query.clone(), wire.schedule_command.clone(), )); let channel_cmd_deps = Arc::new(application::channels::ChannelCommandDeps { channel_command: wire.channel_command.clone(), channel_query: wire.channel_query.clone(), event_publisher: event_publisher.clone(), }); let channel_query_deps = Arc::new(application::channels::ChannelQueryDeps { channel_query: wire.channel_query.clone(), }); let schedule_deps = Arc::new(application::schedule::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(), }); let library_query_deps = Arc::new(application::library::LibraryQueryDeps { library_query: wire.library_query.clone(), }); let server = KTvMcpServer { channel_cmd_deps, channel_query_deps, schedule_deps, library_query_deps, owner_id, }; info!("K-TV MCP server starting (stdio transport), owner_id={}", owner_id); use rmcp::ServiceExt; let service = server .serve(rmcp::transport::stdio()) .await .inspect_err(|e| tracing::error!("MCP server error: {e}"))?; service.waiting().await?; Ok(()) } struct WireOutput { channel_command: Arc, channel_query: Arc, schedule_command: Arc, schedule_query: Arc, library_query: 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_command: w.channel_command, channel_query: w.channel_query, schedule_command: w.schedule_command, schedule_query: w.schedule_query, library_query: w.library_query, }) } #[cfg(feature = "postgres")] DbPool::Postgres(pg_pool) => { let w = adapter_postgres::wire(pg_pool.clone()); Ok(WireOutput { channel_command: w.channel_command, channel_query: w.channel_query, schedule_command: w.schedule_command, schedule_query: w.schedule_query, library_query: w.library_query, }) } _ => anyhow::bail!("database backend not compiled into this binary"), } } async fn build_provider_registry() -> Arc { let mut providers: Vec<(String, Arc)> = Vec::new(); #[cfg(feature = "jellyfin")] if let (Some(url), Some(api_key), Some(user_id)) = ( std::env::var("JELLYFIN_BASE_URL").ok(), std::env::var("JELLYFIN_API_KEY").ok(), std::env::var("JELLYFIN_USER_ID").ok(), ) { info!("Media provider: Jellyfin at {}", url); providers.push(( "jellyfin".to_string(), Arc::new(adapter_jellyfin::JellyfinMediaProvider::new( adapter_jellyfin::JellyfinConfig { base_url: url, api_key, user_id, }, )), )); } if providers.is_empty() { tracing::warn!("No media provider configured. Set JELLYFIN_BASE_URL."); providers.push(("noop".to_string(), Arc::new(NoopMediaProvider))); } Arc::new(SimpleProviderRegistry::new(providers)) } 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, _: &MediaFilter) -> DomainResult> { Err(DomainError::InfrastructureError( "No media provider configured.".into(), )) } async fn fetch_by_id(&self, _: &MediaItemId) -> DomainResult> { Err(DomainError::InfrastructureError( "No media provider configured.".into(), )) } async fn get_stream_url(&self, _: &MediaItemId, _: &StreamQuality) -> 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: &MediaFilter, ) -> 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: &MediaItemId) -> DomainResult> { let id_str = item_id.value(); if let Some(pid) = Self::extract_provider_id(id_str) && 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: &MediaItemId, quality: &StreamQuality, ) -> DomainResult { let id_str = item_id.value(); if let Some(pid) = Self::extract_provider_id(id_str) && 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, ) -> 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>, ) -> 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>, ) -> 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 } }