arr/crates/arr-daemon/src/main.rs

//! 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 metadata;
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, manual_tv) =
        reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), &notifier)?;
    // Issue #176: the on-demand half of the metadata lane needs its own
    // handle — the sweep's `SeriesRefreshAction` is owned by `ReconcileLoop`,
    // and the compat shim takes the other clone below.
    let metadata_tmdb = tmdb.clone();

    // 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 command channels, so the
    // manual-search and manual-grab endpoints were a no-op — the command sat
    // in the 64-slot buffer forever. Issue #132: the episode and season
    // channels are drained by the same lane.
    let mut manual_task = tokio::spawn(manual::run(
        state.clone(),
        database.clone(),
        manual_grab,
        manual_tv,
        shutdown_rx.clone(),
    ));
    // Issue #176: adding a title queues a refresh here rather than waiting
    // for the daily sweep. Its own task, not an arm of `manual::run`: a
    // refresh can take a while against TMDB and must not sit in front of an
    // operator's manual search.
    let mut metadata_task =
        tokio::spawn(metadata::run(state, database, metadata_tmdb, 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?;
            metadata_task.await?;
            result.map_err(Error::Serve)
        }
        result = &mut reconcile_task => {
            result?;
            let _ = shutdown_tx.send(true);
            manual_task.await?;
            metadata_task.await?;
            server.await.map_err(Error::Serve)
        }
        result = &mut manual_task => {
            result?;
            let _ = shutdown_tx.send(true);
            reconcile_task.await?;
            metadata_task.await?;
            server.await.map_err(Error::Serve)
        }
        result = &mut metadata_task => {
            result?;
            let _ = shutdown_tx.send(true);
            reconcile_task.await?;
            manual_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 independent action instances for `manual::run` (issues #107
/// and #132) — the manual trigger needs the same search-and-grab paths on
/// demand rather than on the reconcile tick's schedule, and
/// `ReconcileLoop::register` takes ownership of the ones it ticks. The movie
/// one needs TMDB too; TV grabbing does not (§6.2).
fn reconcile_loop(
    database: &Db,
    config: &Config,
    transmission: &arr_dl::TransmissionClient,
    tmdb: Option<&Arc<TmdbClient>>,
    notifier: &Notifier,
) -> Result<(ReconcileLoop, Option<GrabAction>, Option<TvGrabAction>), 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 (reconcile, manual_grab) = register_movie_grab(
        reconcile,
        prowlarr.as_ref(),
        transmission,
        config,
        &seeding,
        tmdb,
    );
    let (mut reconcile, manual_tv) =
        register_tv_grab(reconcile, prowlarr.as_ref(), transmission, config, &seeding);
    // 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, manual_tv))
}

/// Register the TV grab lane on `reconcile` and hand back a second,
/// independent instance for `manual::run` (issue #132). `None` when Prowlarr
/// is not configured. 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).
fn register_tv_grab(
    mut reconcile: ReconcileLoop,
    prowlarr: Option<&arr_indexer::ProwlarrClient>,
    transmission: &arr_dl::TransmissionClient,
    config: &Config,
    seeding: &SeedingRules,
) -> (ReconcileLoop, Option<TvGrabAction>) {
    let Some(prowlarr) = prowlarr else {
        return (reconcile, None);
    };
    let tv_grab_action = || {
        TvGrabAction::new(
            prowlarr.clone(),
            transmission.clone(),
            config.download_dir.clone(),
            seeding.clone(),
        )
    };
    reconcile = reconcile.register(Tick::Reconcile, tv_grab_action());
    (reconcile, Some(tv_grab_action()))
}

/// 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<TmdbClient>>,
) -> (ReconcileLoop, Option<GrabAction>) {
    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<TmdbClient>,
) -> 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");
}