feat: add Plex media-server catalog and playback baseline (#748)

* feat: add Plex media-server catalog baseline
* feat: add Plex media-server playback baseline

backend/src/processing/processor/library.rs

@@ -250,6 +250,7 @@ fn build_series_episode_playlist_item(
                 added: Some(episode.added.clone()),
                 release_date: Some(episode.release_date.clone()),
                 series_release_date: None,
+                plot: episode.plot.clone(),
                 tmdb: episode.tmdb,
                 movie_image: logo,
                 container_extension: container_extension.intern(),

backend/src/processing/processor/playlist.rs

@@ -8,6 +8,11 @@ use crate::{
         sync_panel_api_exp_dates,
     },
     messaging::send_message,
+    media_server::{
+        media_server_catalog_snapshot_to_playlist, refresh_media_server_catalog_complete_before_publish,
+        MediaServerCatalogRefreshPolicy, MediaServerHttpClient,
+    },
+    media_server::plex::client::PlexCatalogClient,
     model::{
         AppConfig, ConfigFavourites, ConfigInput, ConfigInputFlags, ConfigInputOptions, ConfigRename, ConfigTarget,
         FetchedPlaylist, Mapping, MessageContent, ProcessTargets, ReverseProxyDisabledHeaderConfig, TVGuide,
@@ -362,6 +367,35 @@ fn hybrid_has_m3u_staged_cluster(input: &ConfigInput, skip_cluster: &[XtreamClus
     })
 }
 
+async fn download_plex_media_server_playlist(
+    client: &reqwest::Client,
+    input: &ConfigInput,
+) -> (Vec<PlaylistGroup>, Vec<TuliproxError>) {
+    let Some(media_server) = input.media_server.as_ref() else {
+        return (
+            vec![],
+            vec![TuliproxError::Download(format!(
+                "media-server input '{}' is missing media_server configuration",
+                input.name
+            ))],
+        );
+    };
+    let http_client = MediaServerHttpClient::new(client.clone());
+    let plex_client = match PlexCatalogClient::from_input(input, http_client) {
+        Ok(client) => client,
+        Err(error) => return (vec![], vec![TuliproxError::Download(error.to_string())]),
+    };
+    let policy = MediaServerCatalogRefreshPolicy {
+        page_size: usize::from(media_server.catalog.page_size),
+        request_delay_ms: media_server.catalog.request_delay_ms,
+    };
+
+    match refresh_media_server_catalog_complete_before_publish(&plex_client, policy).await {
+        Ok(snapshot) => (media_server_catalog_snapshot_to_playlist(&snapshot), vec![]),
+        Err(error) => (vec![], vec![TuliproxError::Download(error.to_string())]),
+    }
+}
+
 fn filter_skipped_clusters_from_source(
     source: Box<dyn PlaylistSource>,
     input: &ConfigInput,
@@ -418,7 +452,7 @@ async fn playlist_download_from_input(
         xtream_cache_candidates.extend(main_candidates);
 
         for cluster in xtream_cache_candidates {
-            if !input_cache::is_cache_valid(&status, &cluster.to_string(), cache_duration) {
+            if !input_cache::is_cache_valid(&status, cluster.as_ref(), cache_duration) {
                 xtream_clusters_to_download.push(cluster);
             }
         }
@@ -507,7 +541,11 @@ async fn playlist_download_from_input(
                 let (p, e) = library::download_library_playlist(client, app_config, input).await;
                 (p, e, false, 0, 0)
             }
-            InputType::Emby | InputType::Jellyfin | InputType::Plex => (
+            InputType::Plex => {
+                let (p, e) = download_plex_media_server_playlist(client, input).await;
+                (p, e, false, 0, 0)
+            }
+            InputType::Emby | InputType::Jellyfin => (
                 vec![],
                 vec![TuliproxError::Download(format!(
                     "media-server input '{}' is configured but catalog import is not implemented yet",
@@ -532,14 +570,14 @@ async fn playlist_download_from_input(
         if !xtream_clusters_to_download.is_empty() {
             let state = if xtream_error_count == 0 { ClusterState::Ok } else { ClusterState::Failed };
             for cluster in &xtream_clusters_to_download {
-                input_cache::update_cluster_status(&mut status, &cluster.to_string(), state.clone());
+                input_cache::update_cluster_status(&mut status, cluster.as_ref(), state.clone());
             }
             save_status = true;
         }
     } else if errors.is_empty() {
         if use_per_cluster_cache {
             for cluster in &xtream_clusters_to_download {
-                input_cache::update_cluster_status(&mut status, &cluster.to_string(), ClusterState::Ok);
+                input_cache::update_cluster_status(&mut status, cluster.as_ref(), ClusterState::Ok);
             }
             save_status = !xtream_clusters_to_download.is_empty();
         } else {
@@ -548,7 +586,7 @@ async fn playlist_download_from_input(
         }
     } else if use_per_cluster_cache {
         for cluster in &xtream_clusters_to_download {
-            input_cache::update_cluster_status(&mut status, &cluster.to_string(), ClusterState::Failed);
+            input_cache::update_cluster_status(&mut status, cluster.as_ref(), ClusterState::Failed);
         }
         save_status = !xtream_clusters_to_download.is_empty();
     } else {

backend/src/processing/processor/sort.rs

@@ -167,7 +167,7 @@ fn build_rule_cache_entry(
         return RuleCacheEntry { filter_pass: false, value: None, sequence_match: None };
     }
 
-    let value = provider.get(rule.rule.field.as_str());
+    let value = provider.get(rule.rule.field.as_ref());
     let sequence_match = match (&rule.sequence_plan, value.as_deref()) {
         (Some(sequence_plan), Some(value)) => Some(evaluate_sequence(sequence_plan, value)),
         _ => None,

backend/src/processing/processor/stream_probe.rs

@@ -481,6 +481,7 @@ pub fn update_properties(
                added: None,
                release_date: None,
                series_release_date: None,
+               plot: None,
                tmdb: None,
                movie_image: "".into(),
                container_extension: "".into(),