//! arr — reconcile loop and process entry point. See DESIGN.md §8. mod attention; mod broken; mod config; mod grab; mod import; mod indexers; mod jellyfin; mod manual; mod notify; mod reaper; pub mod reconcile; mod rss; mod series_refresh; mod tv_grab; mod web; use std::process::ExitCode; use std::sync::Arc; use arr_api::{AppState, Upstreams}; use arr_compat::CompatState; use arr_db::Db; use arr_meta::TmdbClient; use attention::AttentionAction; use broken::BrokenAction; use config::Config; use grab::{GrabAction, SeedingLimits, SeedingRules}; use import::ImportAction; use notify::Notifier; use reaper::ReaperAction; use reconcile::{ReconcileLoop, Tick}; use rss::RssAction; use series_refresh::SeriesRefreshAction; use tower_http::trace::TraceLayer; use tv_grab::TvGrabAction; /// Dump the `OpenAPI` document and exit, instead of serving. `just gen-client` /// uses this so the TypeScript client can be regenerated without a port or a /// single upstream being up. const OPENAPI_FLAG: &str = "--openapi"; #[tokio::main] async fn main() -> ExitCode { if std::env::args().nth(1).as_deref() == Some(OPENAPI_FLAG) { return dump_openapi(); } tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| "info,tower_http=debug".into()), ) .init(); match run().await { Ok(()) => ExitCode::SUCCESS, Err(err) => { tracing::error!("{err}"); ExitCode::FAILURE } } } fn dump_openapi() -> ExitCode { match arr_api::openapi().to_pretty_json() { Ok(json) => { println!("{json}"); ExitCode::SUCCESS } Err(err) => { eprintln!("arr: openapi: {err}"); ExitCode::FAILURE } } } #[derive(Debug, thiserror::Error)] enum Error { #[error("config: {0}")] Config(#[from] config::ConfigError), #[error("http client: {0}")] HttpClient(#[from] reqwest::Error), #[error("database: {0}")] Database(#[from] sqlx::Error), #[error("database migration: {0}")] Migration(#[from] sqlx::migrate::MigrateError), #[error("tmdb client: {0}")] Tmdb(#[from] arr_meta::Error), #[error("prowlarr client: {0}")] Prowlarr(#[from] arr_indexer::Error), #[error("transmission client: {0}")] Transmission(#[from] arr_dl::Error), #[error("jellyfin client: {0}")] Jellyfin(#[from] jellyfin::Error), #[error("ntfy client: {0}")] Notify(#[from] notify::NotifyError), #[error("bind {addr}: {source}")] Bind { addr: std::net::SocketAddr, source: std::io::Error, }, #[error("serve: {0}")] Serve(std::io::Error), #[error("background task: {0}")] BackgroundTask(#[from] tokio::task::JoinError), } async fn run() -> Result<(), Error> { let config = Config::load()?; let database = Db::connect(&config.database_path).await?; database.migrate().await?; let transmission = arr_dl::TransmissionClient::new(&config.transmission_url)?; let tmdb = if let Some(key) = &config.tmdb_api_key { let mut client = TmdbClient::builder(key.clone()); if let Some(url) = &config.tmdb_url { client = client.base_url(url.clone()); } Some(Arc::new(client.build()?)) } else { None }; let notifier = Notifier::new(config.ntfy_url.clone())?; let (reconcile, manual_grab) = reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), ¬ifier)?; // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and // needs its own TMDB client for `movie/lookup`. let mut compat = CompatState::new(database.clone()); if let Some(tmdb) = tmdb { compat = compat.with_tmdb(tmdb); } let mut upstreams = Upstreams::new(config.prowlarr_url, config.transmission_url) .with_prowlarr_api_key(config.prowlarr_api_key) .with_tmdb_api_key(config.tmdb_api_key); if let Some(tmdb_url) = config.tmdb_url { upstreams = upstreams.with_tmdb_url(tmdb_url); } let state = AppState::new(upstreams)?.with_database(database.clone()); let app = arr_api::router(state.clone()) .merge(arr_compat::router(compat)) .fallback(web::serve) .layer(TraceLayer::new_for_http()); let listener = tokio::net::TcpListener::bind(config.bind_addr) .await .map_err(|source| Error::Bind { addr: config.bind_addr, source, })?; tracing::info!(addr = %config.bind_addr, docs = arr_api::DOCS_PATH, "listening"); let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); let mut reconcile_task = tokio::spawn(reconcile.run(shutdown_rx.clone())); // Issue #107: nothing else drains `AppState`'s `MovieCommand` channel, so // the manual-search and manual-grab endpoints were a no-op — the command // sat in the 64-slot buffer forever. let mut manual_task = tokio::spawn(manual::run(state, database, manual_grab, shutdown_rx)); let signal_tx = shutdown_tx.clone(); let server = async move { axum::serve(listener, app) .with_graceful_shutdown(async move { shutdown().await; let _ = signal_tx.send(true); }) .await }; tokio::pin!(server); tokio::select! { result = &mut server => { let _ = shutdown_tx.send(true); reconcile_task.await?; manual_task.await?; result.map_err(Error::Serve) } result = &mut reconcile_task => { result?; let _ = shutdown_tx.send(true); manual_task.await?; server.await.map_err(Error::Serve) } result = &mut manual_task => { result?; let _ = shutdown_tx.send(true); reconcile_task.await?; server.await.map_err(Error::Serve) } } } /// Wire the reconcile lanes (DESIGN.md §8). Grab and RSS both need a /// Prowlarr key and grab needs TMDB as well; a lane whose upstream is not /// configured stays unregistered rather than failing every tick. /// /// Also returns a second, independent `GrabAction` for `manual::run` (issue /// #107) — the manual trigger needs the same search-and-grab path on demand /// rather than on the reconcile tick's schedule, and `ReconcileLoop::register` /// takes ownership of the one it ticks. fn reconcile_loop( database: &Db, config: &Config, transmission: &arr_dl::TransmissionClient, tmdb: Option<&Arc>, notifier: &Notifier, ) -> Result<(ReconcileLoop, Option), Error> { let reconcile = ReconcileLoop::new(database.clone()); let seeding = SeedingRules::new( SeedingLimits { ratio: config.seed_ratio_limit, idle_minutes: config.seed_idle_limit_minutes, }, config .tracker_seeding .iter() .map(|(&id, rule)| { ( id, SeedingLimits { ratio: rule.ratio, idle_minutes: rule.min_seed_time, }, ) }) .collect(), ); let prowlarr = config .prowlarr_api_key .clone() .map(|key| arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key)) .transpose()?; let (mut reconcile, manual_grab) = register_movie_grab( reconcile, prowlarr.as_ref(), transmission, config, &seeding, tmdb, ); // TV grabbing needs no TMDB at grab time: air dates are already on the // episode rows, which is the same gate the digital release date is for // movies (§6.2). if let Some(prowlarr) = prowlarr.as_ref() { reconcile = reconcile.register( Tick::Reconcile, TvGrabAction::new( prowlarr.clone(), transmission.clone(), config.download_dir.clone(), seeding.clone(), ), ); } // RSS needs no TMDB: it matches what the feeds already carry against the // wanted list (§6.2). if let Some(prowlarr) = prowlarr { reconcile = reconcile.register( Tick::Rss, RssAction::new( prowlarr, transmission.clone(), config.download_dir.clone(), seeding, ), ); } // §8: metadata refresh is its own daily lane, staggered against the // other ticks. Without TMDB there is nothing to refresh from. if let Some(tmdb) = tmdb { reconcile = reconcile.register(Tick::Metadata, SeriesRefreshAction::new(Arc::clone(tmdb))); } else { tracing::warn!("TMDB is not configured: series metadata refresh is disabled"); } let jellyfin = jellyfin::JellyfinClient::new( config.jellyfin_url.clone(), config.jellyfin_api_key.clone(), )?; // Grab before import, so a download that completes on this tick is // imported on this tick. reconcile = reconcile.register( Tick::Reconcile, ImportAction::new( transmission.clone(), arr_probe::Prober::new(), jellyfin, notifier.clone(), config.ntfy_operator_topic.clone(), ), ); // §9.5 *needs a decision* and *broken* both go to the operator alone; // without a topic configured there is nowhere to send them. if let Some(operator_topic) = &config.ntfy_operator_topic { reconcile = reconcile.register( Tick::Reconcile, AttentionAction::new(notifier.clone(), operator_topic.clone()), ); let broken_upstreams = broken::Upstreams { prowlarr_url: config.prowlarr_url.clone(), prowlarr_api_key: config.prowlarr_api_key.clone(), transmission_url: config.transmission_url.clone(), tmdb_url: config .tmdb_url .clone() .unwrap_or_else(|| arr_api::DEFAULT_TMDB_URL.to_string()), tmdb_api_key: config.tmdb_api_key.clone(), }; reconcile = reconcile.register( Tick::Reconcile, BrokenAction::new(broken_upstreams, notifier.clone(), operator_topic.clone()), ); } else { tracing::warn!( "ARR_NTFY_OPERATOR_TOPIC is not configured: needs-a-decision and broken notifications are disabled" ); } let reconcile = reconcile.register(Tick::Reaper, ReaperAction::new(transmission.clone())); Ok((reconcile, manual_grab)) } /// Register the movie grab lane on `reconcile` and hand back a second, /// independent instance for `manual::run` (issue #107) — the manual trigger /// needs the same search-and-grab path on demand rather than on the tick's /// schedule, and `ReconcileLoop::register` takes ownership of the one it /// ticks. `None` when Prowlarr or TMDB is not configured; nothing can be /// grabbed either way. fn register_movie_grab( mut reconcile: ReconcileLoop, prowlarr: Option<&arr_indexer::ProwlarrClient>, transmission: &arr_dl::TransmissionClient, config: &Config, seeding: &SeedingRules, tmdb: Option<&Arc>, ) -> (ReconcileLoop, Option) { let Some((prowlarr, tmdb)) = prowlarr.zip(tmdb) else { tracing::warn!("Prowlarr or TMDB is not configured: nothing will be grabbed"); return (reconcile, None); }; reconcile = reconcile.register( Tick::Reconcile, movie_grab_action(prowlarr, transmission, config, seeding, tmdb), ); let manual_grab = Some(movie_grab_action( prowlarr, transmission, config, seeding, tmdb, )); (reconcile, manual_grab) } /// The movie grab lane, built fresh for each caller. fn movie_grab_action( prowlarr: &arr_indexer::ProwlarrClient, transmission: &arr_dl::TransmissionClient, config: &Config, seeding: &SeedingRules, tmdb: &Arc, ) -> GrabAction { GrabAction::new( prowlarr.clone(), transmission.clone(), config.download_dir.clone(), seeding.clone(), ) .with_tmdb(Arc::clone(tmdb)) } /// Stop accepting on Ctrl-C, or on the SIGTERM a service manager sends. async fn shutdown() { let interrupt = async { let _ = tokio::signal::ctrl_c().await; }; #[cfg(unix)] let terminate = async { match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { Ok(mut signal) => { signal.recv().await; } Err(err) => tracing::warn!("no SIGTERM handler: {err}"), } }; #[cfg(not(unix))] let terminate = std::future::pending::<()>(); tokio::select! { () = interrupt => {}, () = terminate => {}, } tracing::info!("shutting down"); }