Generated
+1
@@ -49,6 +49,7 @@ dependencies = [
|
||||
"serde_json",
|
||||
"sqlx",
|
||||
"tempfile",
|
||||
"thiserror",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"utoipa",
|
||||
|
||||
@@ -19,6 +19,7 @@ reqwest = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
sqlx = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
utoipa = { workspace = true }
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
//! Jellyfin library refresh after import. See DESIGN.md §7.5.
|
||||
//! Jellyfin library refresh after import or a subtitle write. See DESIGN.md
|
||||
//! §7.5 and §15.
|
||||
//!
|
||||
//! Jellyfin's own filesystem watcher misses the hardlinked file, so import
|
||||
//! asks it to rescan directly. A refresh failure must not fail the import —
|
||||
//! the caller logs and continues (§7.5).
|
||||
//! Jellyfin's own filesystem watcher misses a hardlinked or sidecar file, so
|
||||
//! callers ask it to rescan directly. A refresh failure must not fail the
|
||||
//! caller — it logs and continues (§7.5).
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -27,6 +28,9 @@ pub struct JellyfinClient {
|
||||
}
|
||||
|
||||
impl JellyfinClient {
|
||||
/// # Errors
|
||||
///
|
||||
/// If the underlying HTTP client cannot be built.
|
||||
pub fn new(base_url: impl Into<String>, api_key: Option<String>) -> Result<Self, Error> {
|
||||
let client = Client::builder().timeout(REQUEST_TIMEOUT).build()?;
|
||||
Ok(Self {
|
||||
@@ -38,6 +42,10 @@ impl JellyfinClient {
|
||||
|
||||
/// Trigger a full library scan. Jellyfin exposes no per-library refresh
|
||||
/// without knowing that library's ID, which this app never learns.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// If the request fails, or Jellyfin answers with a non-success status.
|
||||
pub async fn refresh(&self) -> Result<(), Error> {
|
||||
let url = format!("{}/Library/Refresh", self.base_url.trim_end_matches('/'));
|
||||
let mut request = self.client.post(url);
|
||||
@@ -7,6 +7,7 @@
|
||||
//! to compile, and the gate in DESIGN.md §12 fails with it.
|
||||
|
||||
mod health;
|
||||
pub mod jellyfin;
|
||||
mod metadata;
|
||||
mod movies;
|
||||
mod owners;
|
||||
|
||||
@@ -8,6 +8,8 @@ use arr_db::Db;
|
||||
use arr_subs::{Backend, Provider};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::jellyfin::JellyfinClient;
|
||||
|
||||
/// The TMDB API root. Not a bootstrap setting (DESIGN.md §10) — only the key
|
||||
/// is configurable, so this is a constant that tests point elsewhere.
|
||||
pub const DEFAULT_TMDB_URL: &str = "https://api.themoviedb.org/3";
|
||||
@@ -79,6 +81,7 @@ pub struct AppState {
|
||||
pending_metadata_commands: Arc<tokio::sync::Mutex<mpsc::Receiver<MetadataCommand>>>,
|
||||
subtitle_providers: Arc<Vec<Arc<dyn Provider>>>,
|
||||
translation_backends: Arc<Vec<Arc<dyn Backend>>>,
|
||||
jellyfin: Option<JellyfinClient>,
|
||||
}
|
||||
|
||||
/// Work explicitly requested through the movie API.
|
||||
@@ -151,6 +154,7 @@ impl AppState {
|
||||
pending_metadata_commands: Arc::new(tokio::sync::Mutex::new(pending_metadata_commands)),
|
||||
subtitle_providers: Arc::new(Vec::new()),
|
||||
translation_backends: Arc::new(Vec::new()),
|
||||
jellyfin: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -185,6 +189,18 @@ impl AppState {
|
||||
self
|
||||
}
|
||||
|
||||
/// Attach the Jellyfin client, so a manual subtitle grab or translation
|
||||
/// can trigger the same library refresh import does (§7.5, §15).
|
||||
#[must_use]
|
||||
pub fn with_jellyfin(mut self, jellyfin: JellyfinClient) -> Self {
|
||||
self.jellyfin = Some(jellyfin);
|
||||
self
|
||||
}
|
||||
|
||||
pub(crate) fn jellyfin(&self) -> Option<&JellyfinClient> {
|
||||
self.jellyfin.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn subtitle_provider(&self, id: &str) -> Option<&Arc<dyn Provider>> {
|
||||
self.subtitle_providers
|
||||
.iter()
|
||||
|
||||
@@ -900,9 +900,23 @@ async fn finish(
|
||||
.into_iter()
|
||||
.find(|file| file.id == id)
|
||||
.ok_or(ApiError::SubtitleNotFound)?;
|
||||
refresh_jellyfin(state).await;
|
||||
Ok((StatusCode::CREATED, Json(Subtitle::from(subtitle))))
|
||||
}
|
||||
|
||||
/// Ask Jellyfin to rescan, the same single call §7.5 already makes on
|
||||
/// import. Its filesystem watcher misses a sidecar dropped in next to a file
|
||||
/// it already knows about, and a failure here must not fail the write that
|
||||
/// already landed on disk.
|
||||
async fn refresh_jellyfin(state: &AppState) {
|
||||
let Some(jellyfin) = state.jellyfin() else {
|
||||
return;
|
||||
};
|
||||
if let Err(error) = jellyfin.refresh().await {
|
||||
tracing::warn!(%error, "jellyfin refresh failed");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(clippy::too_many_lines)]
|
||||
mod tests {
|
||||
@@ -1159,6 +1173,70 @@ mod tests {
|
||||
application(vec![Arc::new(StubProvider::new("opensubtitles"))], vec![]).await
|
||||
}
|
||||
|
||||
/// Same fixture as [`application`], with a Jellyfin client attached so a
|
||||
/// write can be observed asking it to refresh (§7.5, §15).
|
||||
async fn application_with_jellyfin(
|
||||
providers: Vec<Arc<dyn Provider>>,
|
||||
backends: Vec<Arc<dyn Backend>>,
|
||||
jellyfin_url: &str,
|
||||
) -> Fixture {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let database = arr_db::Db::connect(dir.path().join("arr.db"))
|
||||
.await
|
||||
.expect("connect database");
|
||||
database.migrate().await.expect("migrate database");
|
||||
let pool = database.pool().clone();
|
||||
|
||||
let folder = dir.path().join("Dune (2021) [tmdbid-438631]");
|
||||
tokio::fs::create_dir_all(&folder).await.expect("folder");
|
||||
let video = folder.join("Dune (2021) [tmdbid-438631] - [2160p][WEB-DL].mkv");
|
||||
tokio::fs::write(&video, vec![7u8; 200_000])
|
||||
.await
|
||||
.expect("video");
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO movies (id, tmdb_id, title, year, root_id) VALUES (1, 438631, 'Dune', 2021, 1)",
|
||||
)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.expect("movie");
|
||||
let path = video.to_string_lossy().into_owned();
|
||||
sqlx::query(
|
||||
"INSERT INTO media_files (id, owner_kind, owner_id, path, size) VALUES (1, 'movie', 1, ?, 200000)",
|
||||
)
|
||||
.bind(&path)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.expect("media file");
|
||||
|
||||
let jellyfin =
|
||||
crate::jellyfin::JellyfinClient::new(jellyfin_url, None).expect("jellyfin client");
|
||||
let state = AppState::new(Upstreams::new(
|
||||
"http://127.0.0.1:1".into(),
|
||||
"http://127.0.0.1:1".into(),
|
||||
))
|
||||
.expect("state")
|
||||
.with_database(database)
|
||||
.with_subtitle_providers(providers)
|
||||
.with_translation_backends(backends)
|
||||
.with_jellyfin(jellyfin);
|
||||
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind");
|
||||
let address = listener.local_addr().expect("address");
|
||||
tokio::spawn(async move { axum::serve(listener, router(state)).await.expect("serve") });
|
||||
|
||||
Fixture {
|
||||
_dir: dir,
|
||||
base: format!("http://{address}"),
|
||||
pool,
|
||||
media_file_id: 1,
|
||||
folder,
|
||||
video,
|
||||
}
|
||||
}
|
||||
|
||||
async fn search(fixture: &Fixture, language: &str) -> (StatusCode, serde_json::Value) {
|
||||
let response = reqwest::Client::new()
|
||||
.post(format!(
|
||||
@@ -1271,6 +1349,48 @@ mod tests {
|
||||
assert_eq!(body["error"], "media file not found");
|
||||
}
|
||||
|
||||
/// §7.5 applied to subtitles: a grab calls Jellyfin's refresh, the same
|
||||
/// single call import already makes, because its watcher misses a
|
||||
/// sidecar dropped next to a file it already knows about.
|
||||
#[tokio::test]
|
||||
async fn a_grab_refreshes_jellyfin() {
|
||||
let jellyfin = wiremock::MockServer::start().await;
|
||||
wiremock::Mock::given(wiremock::matchers::method("POST"))
|
||||
.and(wiremock::matchers::path("/Library/Refresh"))
|
||||
.respond_with(wiremock::ResponseTemplate::new(204))
|
||||
.mount(&jellyfin)
|
||||
.await;
|
||||
|
||||
let fixture = application_with_jellyfin(
|
||||
vec![Arc::new(StubProvider::new("opensubtitles"))],
|
||||
vec![],
|
||||
&jellyfin.uri(),
|
||||
)
|
||||
.await;
|
||||
|
||||
let (status, body) = grab(&fixture, pt()).await;
|
||||
assert_eq!(status, StatusCode::CREATED, "{body}");
|
||||
assert_eq!(
|
||||
jellyfin.received_requests().await.expect("requests").len(),
|
||||
1
|
||||
);
|
||||
}
|
||||
|
||||
/// A refresh failure must not fail the grab that already landed the
|
||||
/// sidecar on disk (§7.5).
|
||||
#[tokio::test]
|
||||
async fn an_unreachable_jellyfin_does_not_fail_the_grab() {
|
||||
let fixture = application_with_jellyfin(
|
||||
vec![Arc::new(StubProvider::new("opensubtitles"))],
|
||||
vec![],
|
||||
"http://127.0.0.1:1",
|
||||
)
|
||||
.await;
|
||||
|
||||
let (status, body) = grab(&fixture, pt()).await;
|
||||
assert_eq!(status, StatusCode::CREATED, "{body}");
|
||||
}
|
||||
|
||||
/// §15's disk rule: the sidecar sits next to the video, inside the §7.4
|
||||
/// folder, named `<video basename>.<lang>.srt`.
|
||||
#[tokio::test]
|
||||
|
||||
@@ -25,9 +25,9 @@ use arr_db::Db;
|
||||
use arr_dl::TransmissionClient;
|
||||
use arr_probe::Prober;
|
||||
|
||||
use crate::jellyfin::JellyfinClient;
|
||||
use crate::notify::Notifier;
|
||||
use crate::reconcile::{Action, ActionFuture, Outcome};
|
||||
use arr_api::jellyfin::JellyfinClient;
|
||||
|
||||
/// A failure during one import tick.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
|
||||
@@ -6,7 +6,6 @@ mod config;
|
||||
mod grab;
|
||||
mod import;
|
||||
mod indexers;
|
||||
mod jellyfin;
|
||||
mod manual;
|
||||
mod metadata;
|
||||
mod notify;
|
||||
@@ -94,7 +93,7 @@ enum Error {
|
||||
#[error("transmission client: {0}")]
|
||||
Transmission(#[from] arr_dl::Error),
|
||||
#[error("jellyfin client: {0}")]
|
||||
Jellyfin(#[from] jellyfin::Error),
|
||||
Jellyfin(#[from] arr_api::jellyfin::Error),
|
||||
#[error("ntfy client: {0}")]
|
||||
Notify(#[from] notify::NotifyError),
|
||||
#[error("bind {addr}: {source}")]
|
||||
@@ -124,6 +123,7 @@ async fn run() -> Result<(), Error> {
|
||||
None
|
||||
};
|
||||
let notifier = Notifier::new(config.ntfy_url.clone())?;
|
||||
let api_jellyfin = jellyfin_client(&config)?;
|
||||
let (reconcile, manual_grab, manual_tv) =
|
||||
reconcile_loop(&database, &config, &transmission, tmdb.as_ref(), ¬ifier)?;
|
||||
// Issue #176: the on-demand half of the metadata lane needs its own
|
||||
@@ -150,7 +150,8 @@ async fn run() -> Result<(), Error> {
|
||||
config.opensubtitles_api_key,
|
||||
config.opensubtitles_username,
|
||||
config.opensubtitles_password,
|
||||
));
|
||||
))
|
||||
.with_jellyfin(api_jellyfin);
|
||||
|
||||
let app = arr_api::router(state.clone())
|
||||
.merge(arr_compat::router(compat))
|
||||
@@ -300,10 +301,7 @@ fn reconcile_loop(
|
||||
} 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(),
|
||||
)?;
|
||||
let jellyfin = jellyfin_client(config)?;
|
||||
// Grab before import, so a download that completes on this tick is
|
||||
// imported on this tick.
|
||||
reconcile = reconcile.register(
|
||||
@@ -450,6 +448,16 @@ async fn shutdown() {
|
||||
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 subtitle providers this deployment can reach (DESIGN.md §15).
|
||||
///
|
||||
/// Credentials are bootstrap config and never reach the database (§10), so
|
||||
|
||||
Reference in New Issue
Block a user