[pull] main from mhdzumair:main

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mediafusion-api"
-version = "6.0.0-beta.13"
+version = "6.0.0-beta.14"
 edition = "2021"
 
 [lib]

backend/src/config.rs

@@ -35,8 +35,22 @@ pub struct AppConfig {
     /// Format: "U-{uuid}" or "D-{encrypted}"
     pub mediafusion_secret_str: Option<String>,
     /// Mirrors Python's prowlarr_live_title_search (default: true).
-    /// When false, prowlarr is excluded from live title searches.
+    /// When false, title queries are omitted from live Prowlarr/Jackett searches.
     pub prowlarr_live_title_search: bool,
+    /// Mirrors Python's jackett_live_title_search (default: true).
+    pub jackett_live_title_search: bool,
+
+    // ── Background search ─────────────────────────────────────────────────────
+    /// When false, stream requests do not enqueue items for background re-scraping.
+    pub background_search_enabled: bool,
+    /// Max results to process per indexer during background search (default: 50).
+    pub background_max_process: usize,
+    /// Overall time budget (seconds) for background Prowlarr/Jackett scrapes.
+    pub background_max_process_time: u64,
+    /// Per-request HTTP timeout (seconds) for background Prowlarr/Jackett calls.
+    pub background_query_timeout: u64,
+    /// Minimum hours between background re-scrapes for the same queued item.
+    pub background_search_interval_hours: i64,
 
     // ── Instance mode ────────────────────────────────────────────────────────
     /// When true the instance is fully public: no api_password or X-API-Key
@@ -47,6 +61,15 @@ pub struct AppConfig {
     // ── Torznab / auth ───────────────────────────────────────────────────────
     /// Optional API password for Torznab and private-instance validation.
     pub api_password: Option<String>,
+    /// Expose the Prometheus /api/v1/metrics endpoint (default: false).
+    pub enable_prometheus_metrics: bool,
+    /// Bearer token required to scrape /api/v1/metrics.
+    /// When set, the header `Authorization: Bearer <token>` is required on every
+    /// scrape request — even on public instances.  Configure the matching
+    /// `bearer_token` in your Prometheus scrape_configs.
+    /// When unset and the endpoint is enabled, it is open to anyone who can
+    /// reach the endpoint (rely on network-level controls in that case).
+    pub metrics_api_key: Option<String>,
     /// Enable the Torznab feed endpoint (default: true).
     pub enable_torznab_api: bool,
 
@@ -427,9 +450,24 @@ impl AppConfig {
             mediafusion_secret_str: env("MEDIAFUSION_SECRET_STR").ok().filter(|s| !s.is_empty()),
             prowlarr_live_title_search: env("PROWLARR_LIVE_TITLE_SEARCH")
                 .ok().and_then(|v| v.parse().ok()).unwrap_or(true),
+            jackett_live_title_search: env("JACKETT_LIVE_TITLE_SEARCH")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(true),
+            background_search_enabled: env("BACKGROUND_SEARCH_ENABLED")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(true),
+            background_max_process: env("BACKGROUND_MAX_PROCESS")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(50),
+            background_max_process_time: env("BACKGROUND_MAX_PROCESS_TIME")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(120),
+            background_query_timeout: env("BACKGROUND_QUERY_TIMEOUT")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(30),
+            background_search_interval_hours: env("BACKGROUND_SEARCH_INTERVAL_HOURS")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(72),
             is_public_instance: env("IS_PUBLIC_INSTANCE")
                 .ok().and_then(|v| v.parse().ok()).unwrap_or(false),
             api_password: env("API_PASSWORD").ok().filter(|s| !s.is_empty()),
+            enable_prometheus_metrics: env("ENABLE_PROMETHEUS_METRICS")
+                .ok().and_then(|v| v.parse().ok()).unwrap_or(false),
+            metrics_api_key: env("PROMETHEUS_METRICS_TOKEN").ok().filter(|s| !s.is_empty()),
             enable_torznab_api: env("ENABLE_TORZNAB_API")
                 .ok().and_then(|v| v.parse().ok()).unwrap_or(true),
             is_scrap_from_prowlarr: env("IS_SCRAP_FROM_PROWLARR")

backend/src/db/catalog.rs

@@ -4,6 +4,7 @@ use tracing::warn;
 use super::types::{MediaId, MediaType, UserId};
 
 const LIMIT: i64 = 100;
+const WATCHLIST_LIMIT: i64 = 25;
 
 #[derive(sqlx::FromRow, Debug)]
 pub struct CatalogRow {
@@ -175,6 +176,76 @@ async fn get_library_items(
     })
 }
 
+/// Watchlist catalog rows: media linked to any of the given torrent info_hashes.
+pub async fn get_watchlist_items(
+    pool: &PgPool,
+    media_type: &str,
+    info_hashes: &[String],
+    skip: i64,
+    nudity_excludes: &[String],
+    cert_excludes: &[String],
+    sort: &str,
+    sort_dir: &str,
+) -> Vec<CatalogRow> {
+    if info_hashes.is_empty() {
+        return vec![];
+    }
+
+    let Some(mt) = MediaType::from_wire(media_type) else {
+        return vec![];
+    };
+
+    let ord = order_clause(sort, sort_dir);
+    let sql = format!(
+        r#"
+        SELECT
+            m.id AS media_id,
+            m.type AS media_type,
+            m.title,
+            m.year,
+            m.description,
+            mei.external_id AS imdb_id,
+            mi.url AS poster_url
+        FROM media m
+        JOIN stream_media_link sml ON sml.media_id = m.id
+        JOIN stream s ON s.id = sml.stream_id
+        JOIN torrent_stream ts ON ts.stream_id = s.id
+        LEFT JOIN media_external_id mei ON mei.media_id = m.id AND mei.provider = 'imdb'
+        LEFT JOIN LATERAL (
+            SELECT url FROM media_image
+            WHERE media_id = m.id AND image_type = 'poster' AND is_primary = true
+            LIMIT 1
+        ) mi ON true
+        WHERE m.type = $1
+          AND m.total_streams > 0
+          AND NOT m.is_blocked
+          AND lower(ts.info_hash) = ANY($2)
+          AND (cardinality($3::text[]) IS NULL OR m.nudity_status::text <> ALL($3))
+          AND (cardinality($4::text[]) IS NULL OR NOT EXISTS (
+              SELECT 1 FROM media_parental_certificate_link mpcl
+              JOIN parental_certificate pc ON pc.id = mpcl.certificate_id
+              WHERE mpcl.media_id = m.id AND pc.name = ANY($4)
+          ))
+        GROUP BY m.id, m.type, m.title, m.year, m.description, mei.external_id, mi.url
+        ORDER BY {ord}
+        LIMIT {WATCHLIST_LIMIT} OFFSET $5
+        "#
+    );
+
+    sqlx::query_as::<_, CatalogRow>(&sql)
+        .bind(mt) // $1
+        .bind(info_hashes) // $2
+        .bind(nudity_excludes) // $3
+        .bind(cert_excludes) // $4
+        .bind(skip) // $5
+        .fetch_all(pool)
+        .await
+        .unwrap_or_else(|e| {
+            warn!("watchlist catalog query [type={media_type}]: {e}");
+            vec![]
+        })
+}
+
 pub async fn search_metadata(
     pool: &PgPool,
     media_type: &str,

backend/src/db/genres.rs

@@ -1,52 +1,87 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use sqlx::PgPool;
+use tokio::try_join;
 use tracing::warn;
 
 use super::types::MediaType;
 
-const ADULT_GENRES: &[&str] = &[
-    "adult",
-    "18+",
-    "xxx",
-    "erotic",
-    "erotica",
-    "pornography",
-    "porn",
-];
-
 #[derive(sqlx::FromRow)]
-struct GenreRow {
+struct MediaTypeRow {
+    id: i32,
     media_type: MediaType,
-    genre_name: String,
 }
 
-pub async fn get_all_genres_by_type(pool: &PgPool) -> HashMap<String, Vec<String>> {
-    let rows: Vec<GenreRow> = sqlx::query_as(
-        r#"
-        SELECT DISTINCT m.type AS media_type, g.name AS genre_name
-        FROM genre g
-        JOIN media_genre_link mgl ON mgl.genre_id = g.id
-        JOIN media m ON m.id = mgl.media_id
-        WHERE lower(g.name) <> ALL($1)
-          AND m.total_streams > 0
-          AND NOT m.is_blocked
-        ORDER BY m.type, g.name
-        "#,
+#[derive(sqlx::FromRow)]
+struct GenreNameRow {
+    id: i32,
+    name: String,
+}
+
+#[derive(sqlx::FromRow)]
+struct GenreLinkRow {
+    genre_id: i32,
+    media_id: i32,
+}
+
+fn is_manifest_media_type(media_type: MediaType) -> bool {
+    matches!(
+        media_type,
+        MediaType::Movie | MediaType::Series | MediaType::Tv
     )
-    .bind(ADULT_GENRES)
-    .fetch_all(pool)
-    .await
-    .unwrap_or_else(|e| {
-        warn!("genres query: {e}");
-        vec![]
-    });
-
-    let mut map: HashMap<String, Vec<String>> = HashMap::new();
-    for row in rows {
-        map.entry(row.media_type.as_wire().to_string())
+}
+
+pub async fn get_all_genres_by_type(pool: &PgPool) -> HashMap<String, Vec<String>> {
+    let media_future =
+        sqlx::query_as::<_, MediaTypeRow>("SELECT id, type AS media_type FROM media")
+            .fetch_all(pool);
+    let genre_future =
+        sqlx::query_as::<_, GenreNameRow>("SELECT id, name FROM genre").fetch_all(pool);
+    let links_future =
+        sqlx::query_as::<_, GenreLinkRow>("SELECT genre_id, media_id FROM media_genre_link")
+            .fetch_all(pool);
+
+    let (media_rows, genre_rows, link_rows) =
+        match try_join!(media_future, genre_future, links_future) {
+            Ok(rows) => rows,
+            Err(e) => {
+                warn!("genres query: {e}");
+                return HashMap::new();
+            }
+        };
+
+    let media_types: HashMap<i32, MediaType> = media_rows
+        .into_iter()
+        .map(|row| (row.id, row.media_type))
+        .collect();
+    let genre_names: HashMap<i32, String> = genre_rows
+        .into_iter()
+        .map(|row| (row.id, row.name))
+        .collect();
+
+    let mut by_type: HashMap<String, HashSet<String>> = HashMap::new();
+    for link in link_rows {
+        let Some(media_type) = media_types.get(&link.media_id) else {
+            continue;
+        };
+        if !is_manifest_media_type(*media_type) {
+            continue;
+        }
+        let Some(genre_name) = genre_names.get(&link.genre_id) else {
+            continue;
+        };
+        by_type
+            .entry(media_type.as_wire().to_string())
             .or_default()
-            .push(row.genre_name);
+            .insert(genre_name.clone());
     }
-    map
+
+    by_type
+        .into_iter()
+        .map(|(media_type, genres)| {
+            let mut list: Vec<String> = genres.into_iter().collect();
+            list.sort_unstable();
+            (media_type, list)
+        })
+        .collect()
 }

backend/src/db/meta.rs

@@ -67,12 +67,14 @@ pub async fn get_media_meta(
 
     let result = match parse_meta_id(meta_id) {
         MetaIdKind::Internal(id) => {
-            sqlx::query_as!(
-                MediaMetaRow,
+            // Use non-macro query_as so LEFT JOIN LATERAL columns (poster_url, background_url,
+            // imdb_rating) are decoded as nullable via the struct's Option<T> field types.
+            // The query_as! offline cache incorrectly infers those columns as non-nullable.
+            sqlx::query_as::<_, MediaMetaRow>(
                 r#"
                 SELECT
-                    m.id AS "media_id: MediaId",
-                    m.type AS "media_type: MediaType",
+                    m.id AS media_id,
+                    m.type AS media_type,
                     m.title,
                     m.year,
                     EXTRACT(YEAR FROM m.end_date)::int AS end_year,
@@ -108,19 +110,18 @@ pub async fn get_media_meta(
                 WHERE m.id = $1 AND m.type = $2
                 LIMIT 1
                 "#,
-                id as MediaId,
-                media_type as MediaType,
             )
+            .bind(id)
+            .bind(media_type)
             .fetch_optional(pool)
             .await
         }
         MetaIdKind::External(ext_id) => {
-            sqlx::query_as!(
-                MediaMetaRow,
+            sqlx::query_as::<_, MediaMetaRow>(
                 r#"
                 SELECT
-                    m.id AS "media_id: MediaId",
-                    m.type AS "media_type: MediaType",
+                    m.id AS media_id,
+                    m.type AS media_type,
                     m.title,
                     m.year,
                     EXTRACT(YEAR FROM m.end_date)::int AS end_year,
@@ -158,9 +159,9 @@ pub async fn get_media_meta(
                 WHERE m.type = $2
                 LIMIT 1
                 "#,
-                ext_id,
-                media_type as MediaType,
             )
+            .bind(ext_id)
+            .bind(media_type)
             .fetch_optional(pool)
             .await
         }
@@ -214,8 +215,9 @@ pub async fn get_cast(pool: &PgPool, media_id: MediaId) -> Vec<String> {
 }
 
 pub async fn get_episodes(pool: &PgPool, media_id: MediaId) -> Vec<EpisodeRow> {
-    sqlx::query_as!(
-        EpisodeRow,
+    // Use non-macro query_as so LEFT JOIN LATERAL (thumbnail_url) and LEFT JOIN
+    // (media_id) columns decode as nullable via the struct's Option<T> field types.
+    sqlx::query_as::<_, EpisodeRow>(
         r#"
         SELECT
             s.season_number,
@@ -224,7 +226,7 @@ pub async fn get_episodes(pool: &PgPool, media_id: MediaId) -> Vec<EpisodeRow> {
             e.overview,
             e.air_date,
             ei.url AS thumbnail_url,
-            fml.media_id AS "media_id: Option<MediaId>"
+            fml.media_id AS media_id
         FROM series_metadata sm
         JOIN season s ON s.series_id = sm.id
         JOIN episode e ON e.season_id = s.id
@@ -240,8 +242,8 @@ pub async fn get_episodes(pool: &PgPool, media_id: MediaId) -> Vec<EpisodeRow> {
         WHERE sm.media_id = $1
         ORDER BY s.season_number, e.episode_number
         "#,
-        media_id as MediaId,
     )
+    .bind(media_id)
     .fetch_all(pool)
     .await
     .unwrap_or_else(|e| {

backend/src/db/streams.rs

@@ -620,7 +620,7 @@ pub async fn fetch_stream_playback_info(
                 LEFT JOIN torrent_tracker_link ttl ON ttl.torrent_id = ts.id
                 LEFT JOIN tracker t ON t.id = ttl.tracker_id
                 WHERE ts.info_hash = $1
-                GROUP BY st.id, st.name, ts.total_size
+                GROUP BY st.id, st.name, ts.total_size, ts.torrent_file
                 "#,
             )
             .bind(info_hash)