Files
arr/crates/arr-daemon/src/main.rs
T
Miguel Palhas 5bc9022046 fix(arr): pass the podnapisi seam to the lamp probes
#200 branched before #205's ARR_PODNAPISI_URL seam was pushed, so its new
broken::SubtitleUpstreams call site was written against the three-argument
subtitle_providers. The merge was textually clean and did not build.
2026-08-25 06:31:27 +01:00

742 lines
26 KiB
Rust

//! arr — reconcile loop and process entry point. See DESIGN.md §8.
mod attention;
mod broken;
mod config;
mod grab;
mod import;
mod indexers;
mod manual;
mod metadata;
mod notify;
mod reaper;
pub mod reconcile;
mod rss;
mod series_refresh;
mod subtitles;
mod tv_grab;
mod web;
use std::process::ExitCode;
use std::sync::{atomic::AtomicU64, 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 subtitles::SubtitleAction;
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] arr_api::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),
}
/// Everything the HTTP layer needs, assembled from config.
///
/// Split out of [`run`] because it grows a line per upstream the API learns
/// to talk to, and `run` is already at the too-many-lines limit.
fn api_state(
config: &Config,
database: &Db,
jellyfin: arr_api::jellyfin::JellyfinClient,
translators: &Translators,
) -> Result<AppState, Error> {
let mut upstreams =
Upstreams::new(config.prowlarr_url.clone(), config.transmission_url.clone())
.with_prowlarr_api_key(config.prowlarr_api_key.clone())
.with_tmdb_api_key(config.tmdb_api_key.clone());
if let Some(tmdb_url) = config.tmdb_url.clone() {
upstreams = upstreams.with_tmdb_url(tmdb_url);
}
let mut state = AppState::new(upstreams)?
.with_database(database.clone())
.with_subtitle_providers(subtitle_providers(
config.opensubtitles_api_key.clone(),
config.opensubtitles_username.clone(),
config.opensubtitles_password.clone(),
config.podnapisi_url.clone(),
))
.with_translation_backends(translators.backends.clone())
.with_jellyfin(jellyfin)
.with_syncer(arr_subs::Syncer::new().with_binary(config.alass_path.clone()));
if let Some(timeout) = &translators.command_timeout {
state = state.with_command_timeout(Arc::clone(timeout));
}
Ok(state)
}
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 api_jellyfin = jellyfin_client(&config)?;
// Built once and shared: the API's translate handler and the reconcile
// lane must see the same backends, or the command translator's live
// timeout cell (#219) would fork.
let translators = translation_backends(&config);
seed_command_timeout(&database, &translators).await?;
let (reconcile, manual_grab, manual_tv) = reconcile_loop(
&database,
&config,
&transmission,
tmdb.as_ref(),
&notifier,
&translators,
)?;
// 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 state = api_state(&config, &database, api_jellyfin, &translators)?;
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)
}
}
}
/// Seed the command translator's live timeout from the settings row, so a
/// restart does not fall back to the compiled-in default until the next
/// settings edit (issue #219). The row is guaranteed to exist and to carry a
/// positive value — migration `0025` seeds it and the column CHECK enforces
/// it.
async fn seed_command_timeout(database: &Db, translators: &Translators) -> Result<(), Error> {
let Some(cell) = &translators.command_timeout else {
return Ok(());
};
let seconds: i64 = sqlx::query_scalar!(
r#"SELECT remote_command_timeout_seconds AS "remote_command_timeout_seconds!: i64"
FROM subtitle_settings WHERE id = 1"#
)
.fetch_one(database.pool())
.await?;
cell.store(
u64::try_from(seconds)
.unwrap_or(u64::MAX)
.saturating_mul(1_000),
std::sync::atomic::Ordering::Relaxed,
);
Ok(())
}
/// 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,
translators: &Translators,
) -> 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_client(config)?;
// 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(),
),
);
// §15: subtitle gaps are reconciled from the same rows the API writes.
reconcile = reconcile.register(
Tick::Reconcile,
subtitle_action(config, notifier, translators)?,
);
// §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()),
);
reconcile = register_broken(
reconcile,
config,
translators,
notifier,
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))
}
/// The *broken* lane (#200 included): Prowlarr, Transmission and TMDB, plus
/// the subtitle lamps — an enabled provider, the selected engine, or a
/// missing `alass`/`ffmpeg` all fold into the same operator message.
fn register_broken(
reconcile: ReconcileLoop,
config: &Config,
translators: &Translators,
notifier: &Notifier,
operator_topic: String,
) -> ReconcileLoop {
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(),
};
let broken_subtitles = broken::SubtitleUpstreams {
providers: subtitle_providers(
config.opensubtitles_api_key.clone(),
config.opensubtitles_username.clone(),
config.opensubtitles_password.clone(),
config.podnapisi_url.clone(),
),
backends: translators.backends.clone(),
alass_path: config.alass_path.clone(),
ffmpeg_path: config.ffmpeg_path.clone(),
};
reconcile.register(
Tick::Reconcile,
BrokenAction::new(
broken_upstreams,
broken_subtitles,
notifier.clone(),
operator_topic,
),
)
}
/// 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");
}
/// The Jellyfin client, built fresh for each of its two independent callers:
/// import's own reconcile action, and the subtitle API's manual grab and
/// translate handlers (§7.5, §15).
fn jellyfin_client(config: &Config) -> Result<arr_api::jellyfin::JellyfinClient, Error> {
Ok(arr_api::jellyfin::JellyfinClient::new(
config.jellyfin_url.clone(),
config.jellyfin_api_key.clone(),
)?)
}
/// The §15 reconcile lane: closes subtitle gaps from the attempt rows.
///
/// Takes the same [`Translators`] the API is given, built once at startup, so
/// a translation behaves identically whether the reconcile sweep or the manual
/// endpoint asked for it. Which of the offered backends actually runs is the
/// `translation_engine` database setting, read per translation; with none
/// compiled in the translate step records "not compiled" and backs off rather
/// than failing obscurely.
fn subtitle_action(
config: &Config,
notifier: &Notifier,
translators: &Translators,
) -> Result<SubtitleAction, Error> {
let action = SubtitleAction::new(
subtitle_providers(
config.opensubtitles_api_key.clone(),
config.opensubtitles_username.clone(),
config.opensubtitles_password.clone(),
config.podnapisi_url.clone(),
),
translators.backends.clone(),
arr_subs::Syncer::new().with_binary(config.alass_path.clone()),
arr_probe::Extractor::new().with_binary(config.ffmpeg_path.clone()),
jellyfin_client(config)?,
);
Ok(match &config.ntfy_operator_topic {
Some(topic) => action.with_notifier(notifier.clone(), topic.clone()),
None => action,
})
}
/// The subtitle providers this deployment can reach (DESIGN.md §15).
///
/// Credentials are bootstrap config and never reach the database (§10), so
/// which providers *exist* is decided here, once, at startup; which of them a
/// search *runs* is the `providers_enabled` setting the API reads per
/// request. OpenSubtitles.com needs a registered API key to be called at all,
/// so without one it is not offered; Podnapisi is anonymous and always is.
fn subtitle_providers(
opensubtitles_api_key: Option<String>,
username: Option<String>,
password: Option<String>,
podnapisi_url: Option<String>,
) -> Vec<std::sync::Arc<dyn arr_subs::Provider>> {
let mut providers: Vec<std::sync::Arc<dyn arr_subs::Provider>> = Vec::new();
// `podnapisi_url` is the e2e-only seam (`ARR_PODNAPISI_URL`); absent, this
// is the real Podnapisi.net.
let mut builder = arr_subs::Podnapisi::builder();
if let Some(url) = podnapisi_url {
builder = builder.base_url(url);
}
match builder.build() {
Ok(podnapisi) => providers.push(std::sync::Arc::new(podnapisi)),
Err(error) => tracing::warn!(%error, "Podnapisi not available"),
}
let Some(api_key) = opensubtitles_api_key else {
tracing::info!("no OpenSubtitles.com API key configured; that provider is off");
return providers;
};
match arr_subs::OpenSubtitles::new(arr_subs::OpenSubtitlesConfig {
api_key,
username,
password,
}) {
Ok(opensubtitles) => providers.push(std::sync::Arc::new(opensubtitles)),
Err(error) => tracing::warn!(%error, "OpenSubtitles.com not available"),
}
providers
}
/// The translation backends this deployment can offer, built once at startup
/// and shared by the API and the reconcile lane (DESIGN.md §15, issue #216).
///
/// Which cargo features this binary was built with decides what could ever
/// be here (`compiled_engines`); credentials decide what actually is, same
/// split `subtitle_providers` makes for search. Which one of these a
/// translation *uses* is the `translation_engine` database setting, read per
/// request (#198) — this only decides which ids exist to be picked.
///
/// When the remote-command backend is one of them, its live timeout cell
/// rides along (#219): the API writes it on every settings edit, so the row
/// reaches the running process without a restart.
struct Translators {
backends: Vec<std::sync::Arc<dyn arr_subs::Backend>>,
command_timeout: Option<Arc<AtomicU64>>,
}
#[cfg_attr(
not(any(
feature = "translate-openai",
feature = "translate-deepl",
feature = "translate-google",
feature = "translate-command"
)),
allow(unused_variables, unused_mut)
)]
fn translation_backends(config: &Config) -> Translators {
let mut backends: Vec<std::sync::Arc<dyn arr_subs::Backend>> = Vec::new();
let mut command_timeout: Option<Arc<AtomicU64>> = None;
#[cfg(feature = "translate-openai")]
{
let openai_config = arr_subs::OpenAiConfig {
api_key: config.translate_openai_api_key.clone(),
};
let backend = match &config.translate_openai_base_url {
Some(base_url) => arr_subs::OpenAi::with_base_url(
config.translate_openai_model.clone(),
openai_config,
base_url,
),
None => arr_subs::OpenAi::new(config.translate_openai_model.clone(), openai_config),
};
match backend {
Ok(backend) => backends.push(std::sync::Arc::new(backend)),
Err(error) => tracing::warn!(%error, "OpenAI-compatible translator not available"),
}
}
#[cfg(feature = "translate-deepl")]
{
if let Some(auth_key) = config.translate_deepl_api_key.clone() {
let deepl_config = arr_subs::DeepLConfig { auth_key };
let backend = match &config.translate_deepl_base_url {
Some(base_url) => arr_subs::DeepL::with_base_url(deepl_config, base_url),
None => arr_subs::DeepL::new(deepl_config),
};
match backend {
Ok(backend) => backends.push(std::sync::Arc::new(backend)),
Err(error) => tracing::warn!(%error, "DeepL not available"),
}
} else {
tracing::info!("no DeepL auth key configured; that translator is off");
}
}
#[cfg(feature = "translate-google")]
{
if let Some(api_key) = config.translate_google_api_key.clone() {
let google_config = arr_subs::GoogleConfig { api_key };
let backend = match &config.translate_google_base_url {
Some(base_url) => arr_subs::Google::with_base_url(google_config, base_url),
None => arr_subs::Google::new(google_config),
};
match backend {
Ok(backend) => backends.push(std::sync::Arc::new(backend)),
Err(error) => tracing::warn!(%error, "Google Translate not available"),
}
} else {
tracing::info!("no Google Translate API key configured; that translator is off");
}
}
#[cfg(feature = "translate-command")]
{
if let Some(template) = config.translate_command_template.clone() {
let command_config = arr_subs::CommandConfig {
template,
timeout: arr_subs::COMMAND_DEFAULT_TIMEOUT,
};
match arr_subs::Command::new(command_config) {
Ok(backend) => {
command_timeout = Some(backend.timeout_cell());
backends.push(std::sync::Arc::new(backend));
}
Err(error) => tracing::warn!(%error, "remote-command translator not available"),
}
} else {
tracing::info!("no remote-command template configured; that translator is off");
}
}
Translators {
backends,
command_timeout,
}
}