feat: add TMDB Trending target curation (#872)

- add target-wide TMDB Trending discovery for movie/TV and day/week, bounded by an optional `limit` (default 100, range 1–500) of unique remote references before local matching
- compose concrete Trakt and TMDB sources under one target curation policy while retaining legacy `output[].trakt` compatibility
- match only positive TMDB IDs of the same media kind and preserve existing catalog identities, series closure, and Xtream alias behavior
- require every active selector to complete before target IDs, writers, caches, or watches can change
- preserve the YAML-owned block through Source Editor operations and add TMDB setup/failure documentation and application Credits attribution
- validate complete decoded pages under independent request, selector, and shared target-batch budgets, using a configured TMDB-only HTTP profile with certificate/hostname verification and no redirects or replays

backend/processing/src/processor/playlist/mod.rs

@@ -49,14 +49,14 @@ use tokio::{
 use tuliprox_core::{
     model::{
         is_valid, retain_filtered_playlist, AppConfig, ClusterForceUpdate, ClusterUpdateRejection, CompiledMapping,
-        ConfigFavourites, ConfigInput, ConfigInputFlags, ConfigInputOptions, ConfigRename, ConfigTarget, Epg,
-        FilterOutcome, MappingProgram, ProcessTargets, ProviderIdType, ResolveReason, ReverseProxyDisabledHeaderConfig,
-        TraktConfig, TransformStage, UpdateGuard, UpdateTask,
+        ConfigFavourites, ConfigInput, ConfigInputFlags, ConfigInputOptions, ConfigRename, ConfigTarget,
+        CurationConfig, Epg, FilterOutcome, MappingProgram, ProcessTargets, ProviderIdType, ResolveReason,
+        ReverseProxyDisabledHeaderConfig, TransformStage, UpdateGuard, UpdateTask,
     },
     utils::{debug_if_enabled, log_memory_snapshot, trace_if_enabled, StepMeasure, StepMeasureCallback},
 };
 use tuliprox_curation::{
-    evaluate_trakt_curation, project_trakt_categories, CurationEvaluation, CurationFailure, CurationRunOutcome,
+    evaluate_curation, project_curation_categories, CurationEvaluation, CurationFailure, CurationRunOutcome,
     SelectorOutcome,
 };
 use tuliprox_iptv::{

backend/processing/src/processor/playlist/target.rs

@@ -463,9 +463,49 @@ fn finalize_playlist_view(
     playlist
 }
 
+// Keep construction and its diagnostic seam private to the processing boundary.
+// Only fixed phase labels can reach the reporter, never dependency error details.
+pub(super) fn build_target_tmdb_client(
+    target: &ConfigTarget,
+    configure: impl FnOnce() -> Result<reqwest::ClientBuilder, TuliproxError>,
+    report: impl FnOnce(&'static str),
+) -> Option<reqwest::Client> {
+    let needs_tmdb = target.curation.as_ref().is_some_and(|config| {
+        config.enabled && config.tmdb.as_ref().is_some_and(|source| source.enabled && !source.trending.is_empty())
+    });
+    if !needs_tmdb {
+        return None;
+    }
+    let Ok(builder) = configure() else {
+        report("phase=profile_configuration");
+        return None;
+    };
+    if let Ok(client) = builder.build() {
+        Some(client)
+    } else {
+        report("phase=client_build");
+        None
+    }
+}
 pub(crate) async fn finalize_prepared_target<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
     ctx: Arc<PlaylistProcessingContext<E, M>>,
     prepared: PreparedTarget,
+) -> (Result<(), Vec<TuliproxError>>, Vec<TuliproxError>) {
+    // No generic-client fallback: unavailable transport participates in the same admission gate.
+    let tmdb_client = build_target_tmdb_client(
+        &prepared.target,
+        || tuliprox_core::utils::network::request::create_tmdb_client(&ctx.config),
+        |phase| {
+            warn!("TMDB discovery client unavailable: target_id={} run_id={} {phase}", prepared.target.id, ctx.run_id);
+        },
+    );
+    finalize_prepared_target_with_tmdb(ctx, prepared, tmdb_client.as_ref()).await
+}
+
+pub(super) async fn finalize_prepared_target_with_tmdb<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
+    ctx: Arc<PlaylistProcessingContext<E, M>>,
+    prepared: PreparedTarget,
+    tmdb_client: Option<&reqwest::Client>,
 ) -> (Result<(), Vec<TuliproxError>>, Vec<TuliproxError>) {
     let target = &prepared.target;
     let mut new_playlist = prepared.playlist;
@@ -488,7 +528,7 @@ pub(crate) async fn finalize_prepared_target<E: EventSink + Clone + 'static, M:
     }
 
     let eligible_catalog = prepare_eligible_catalog(target, new_playlist, &mut step);
-    let views = match prepare_target_playlist_views(&ctx.client, target, eligible_catalog).await {
+    let views = match prepare_target_playlist_views(&ctx.client, tmdb_client, target, eligible_catalog).await {
         Ok(views) => views,
         Err(error) => {
             step.stop("Curation failed; skipping persist to preserve finalized artifacts");
@@ -864,20 +904,24 @@ pub(crate) struct TargetPlaylistViews {
 pub(crate) fn build_curated_playlist_views(
     playlist: Vec<PlaylistGroup>,
     evaluation: &CurationEvaluation,
-    trakt_config: &TraktConfig,
+    config: &CurationConfig,
     appearance_filter_configured: bool,
+    has_xtream_output: bool,
 ) -> TargetPlaylistViews {
-    let curated_catalog = trakt_config.catalog_selection.is_curated();
-    let mut categories = project_trakt_categories(evaluation, &playlist, trakt_config);
+    let curated_catalog = config.catalog_selection.is_curated();
+    let categories = has_xtream_output.then(|| project_curation_categories(evaluation, &playlist, config));
     let base = select_target_catalog(playlist, evaluation, curated_catalog);
-    let mut xtream = if trakt_config.include_xtream_base_categories { base.clone() } else { live_only(&base) };
-    xtream.append(&mut categories);
+    let xtream = categories.map(|mut categories| {
+        let mut view = if config.include_xtream_base_categories { base.clone() } else { live_only(&base) };
+        view.append(&mut categories);
+        view
+    });
     TargetPlaylistViews {
         base,
-        xtream: Some(xtream),
+        xtream,
         publication_plan: PlaylistPublicationPlan::complete_curation_with_filter(
             curated_catalog,
-            !trakt_config.include_xtream_base_categories,
+            has_xtream_output && !config.include_xtream_base_categories,
             appearance_filter_configured,
         ),
     }
@@ -885,10 +929,11 @@ pub(crate) fn build_curated_playlist_views(
 
 pub(crate) async fn prepare_target_playlist_views(
     client: &reqwest::Client,
+    tmdb_client: Option<&reqwest::Client>,
     target: &ConfigTarget,
     playlist: Vec<PlaylistGroup>,
 ) -> Result<TargetPlaylistViews, TuliproxError> {
-    let Some(trakt_config) = target.get_xtream_output().and_then(|xtream| xtream.trakt.as_ref()) else {
+    let Some(config) = target.effective_curation() else {
         return Ok(TargetPlaylistViews {
             base: playlist,
             xtream: None,
@@ -896,7 +941,17 @@ pub(crate) async fn prepare_target_playlist_views(
         });
     };
 
-    match evaluate_trakt_curation(client, &playlist, &target.name, trakt_config).await {
+    let outcome = evaluate_curation(client, tmdb_client, &playlist, &target.name, &config).await;
+    curation_playlist_views(target, playlist, &config, outcome)
+}
+
+pub(super) fn curation_playlist_views(
+    target: &ConfigTarget,
+    playlist: Vec<PlaylistGroup>,
+    config: &CurationConfig,
+    outcome: CurationRunOutcome,
+) -> Result<TargetPlaylistViews, TuliproxError> {
+    match outcome {
         CurationRunOutcome::NotConfigured => Ok(TargetPlaylistViews {
             base: playlist,
             xtream: None,
@@ -904,8 +959,13 @@ pub(crate) async fn prepare_target_playlist_views(
         }),
         CurationRunOutcome::Failed(failure) => Err(curation_failure_error(&target.name, &failure)),
         CurationRunOutcome::Complete(evaluation) => {
-            let views =
-                build_curated_playlist_views(playlist, &evaluation, trakt_config, target.filter.persist.is_some());
+            let views = build_curated_playlist_views(
+                playlist,
+                &evaluation,
+                config,
+                target.filter.persist.is_some(),
+                target.get_xtream_output().is_some(),
+            );
             info!(
                 "Target '{}' curation completed with {} memberships and {} Xtream groups",
                 target.name,
@@ -916,7 +976,6 @@ pub(crate) async fn prepare_target_playlist_views(
         }
     }
 }
-
 fn curation_failure_error(target_name: &str, failure: &CurationFailure) -> TuliproxError {
     let mut complete = 0usize;
     let mut incomplete = 0usize;

backend/processing/src/processor/playlist/tests.rs

@@ -16,6 +16,8 @@ use tuliprox_curation::{
     CurationEvaluation, CurationMediaKind, CurationMembership, CurationSelectorKey, CurationSelectorSummary,
 };
 
+mod curation_publication;
+
 fn apply_staged_overlay_groups(
     provider: &ConfigInput,
     clusters: ClusterFlags,
@@ -2709,7 +2711,7 @@ fn persist_filter_can_select_a_base_group() {
 async fn trakt_target_curation_is_a_noop_without_xtream_configuration() {
     let target = ConfigTarget::from(&ConfigTargetDto::default());
 
-    let views = prepare_target_playlist_views(&reqwest::Client::new(), &target, Vec::new())
+    let views = prepare_target_playlist_views(&reqwest::Client::new(), None, &target, Vec::new())
         .await
         .expect("unconfigured curation should not fail");
 
@@ -2752,7 +2754,7 @@ async fn unavailable_required_selector_returns_target_failure_instead_of_base_fa
         xtream_cluster: XtreamCluster::Video,
     }];
 
-    let error = prepare_target_playlist_views(&reqwest::Client::new(), &target, base)
+    let error = prepare_target_playlist_views(&reqwest::Client::new(), None, &target, base)
         .await
         .expect_err("missing credentials must stop target publication");
 
@@ -2778,7 +2780,7 @@ mod curation_effect_gate {
         utils::FileLockManager,
     };
 
-    fn app_config(storage_dir: &Path) -> Arc<AppConfig> {
+    pub(super) fn app_config(storage_dir: &Path) -> Arc<AppConfig> {
         Arc::new(AppConfig {
             config: Arc::new(ArcSwap::from_pointee(Config {
                 storage_dir: storage_dir.to_string_lossy().into_owned(),
@@ -2808,7 +2810,7 @@ mod curation_effect_gate {
         })
     }
 
-    fn processing_context(
+    pub(super) fn processing_context(
         app_config: Arc<AppConfig>,
         playlist_state: Option<Arc<PlaylistStorageState>>,
     ) -> PlaylistProcessingContext<NoopSink> {
@@ -2844,7 +2846,7 @@ mod curation_effect_gate {
         TargetPlaylistPersistOptions { publication_plan, ..TargetPlaylistPersistOptions::default() }
     }
 
-    fn file_snapshot(root: &Path) -> BTreeMap<std::path::PathBuf, Vec<u8>> {
+    pub(super) fn file_snapshot(root: &Path) -> BTreeMap<std::path::PathBuf, Vec<u8>> {
         fn collect(root: &Path, path: &Path, snapshot: &mut BTreeMap<std::path::PathBuf, Vec<u8>>) {
             let Ok(entries) = std::fs::read_dir(path) else { return };
             for entry in entries.flatten() {
@@ -2892,10 +2894,35 @@ mod curation_effect_gate {
 
     #[tokio::test]
     async fn unavailable_curation_retains_seeded_artifacts_cache_and_watch_state() {
+        assert_curation_retains_seeded_state(None).await;
+    }
+
+    #[tokio::test]
+    async fn tmdb_and_mixed_failures_preserve_files_ids_cache_and_watches_under_full_and_curated() {
+        for policy in ["full", "curated"] {
+            for mixed in [false, true] {
+                let (url, server) = empty_trakt_server().await;
+                let mut value = serde_json::json!({"catalog_selection": policy, "tmdb": {"trending": [{"kind": "movie", "time_window": "week", "limit": 100, "category_name": "TMDB"}]}});
+                if mixed {
+                    value["trakt"] = serde_json::json!({"api": {"api_key": "test-client", "url": url}, "charts": [{"kind": "movies", "chart": "popular", "category_name": "Trakt"}]});
+                }
+                let mut dto: shared::model::CurationConfigDto = serde_json::from_value(value).unwrap();
+                dto.prepare(&[TargetOutputDto::Xtream(XtreamTargetOutputDto::default())]).unwrap();
+                assert_curation_retains_seeded_state(Some(dto)).await;
+                if mixed {
+                    server.await.unwrap();
+                } else {
+                    server.abort();
+                }
+            }
+        }
+    }
+
+    async fn assert_curation_retains_seeded_state(canonical: Option<shared::model::CurationConfigDto>) {
         let directory = tempdir().expect("tempdir");
         let app_config = app_config(directory.path());
         let playlist_state = Arc::new(PlaylistStorageState::new());
-        let target = ConfigTarget::from(&ConfigTargetDto {
+        let mut target = ConfigTarget::from(&ConfigTargetDto {
             name: "curation-effect-gate".to_string(),
             output: vec![
                 TargetOutputDto::Xtream(XtreamTargetOutputDto {
@@ -2922,6 +2949,12 @@ mod curation_effect_gate {
             use_memory_cache: true,
             ..ConfigTargetDto::default()
         });
+        if let Some(canonical) = canonical {
+            target.curation = Some(CurationConfig::from(&canonical));
+            if let tuliprox_core::model::TargetOutput::Xtream(output) = &mut target.output[0] {
+                output.trakt = None;
+            }
+        }
         let mut seeded = vec![PlaylistGroup {
             id: 1,
             title: "Movies".intern(),
@@ -3064,6 +3097,68 @@ mod curation_effect_gate {
     }
 }
 
+#[test]
+fn tmdb_complete_target_selection_preserves_live_and_keeps_xtream_aliases_out_of_normal_outputs() {
+    for has_xtream in [false, true] {
+        let mut value = serde_json::json!({"name": "discovery", "output": [{"type": "m3u"}],
+            "curation": {"catalog_selection": "curated", "tmdb": {"trending": [{"kind": "movie", "time_window": "week", "limit": 100, "create_xtream_category": has_xtream, "category_name": "TMDB"}]}}});
+        if has_xtream {
+            value["output"].as_array_mut().unwrap().push(serde_json::json!({"type": "xtream"}));
+        }
+        let mut dto: ConfigTargetDto = serde_json::from_value(value).unwrap();
+        dto.prepare(1, None, None).unwrap();
+        let target = ConfigTarget::from(&dto);
+        let id = shared::utils::hash_string("selected");
+        let live_id = shared::utils::hash_string("live");
+        let playlist = vec![
+            PlaylistGroup {
+                id: 1,
+                title: "Live".intern(),
+                xtream_cluster: XtreamCluster::Live,
+                channels: vec![catalog_test_item("Live", live_id, PlaylistItemType::Live, XtreamCluster::Live, None)],
+            },
+            PlaylistGroup {
+                id: 2,
+                title: "Movies".intern(),
+                xtream_cluster: XtreamCluster::Video,
+                channels: vec![
+                    catalog_test_item("Selected", id, PlaylistItemType::Video, XtreamCluster::Video, None),
+                    catalog_test_item(
+                        "Other",
+                        shared::utils::hash_string("other"),
+                        PlaylistItemType::Video,
+                        XtreamCluster::Video,
+                        None,
+                    ),
+                ],
+            },
+        ];
+        let config = target.effective_curation().unwrap();
+        let outcome = CurationRunOutcome::Complete(complete_catalog_evaluation(vec![catalog_membership(
+            id,
+            CurationMediaKind::Movie,
+            0,
+        )]));
+        let views = target::curation_playlist_views(&target, playlist.clone(), &config, outcome).unwrap();
+        assert_eq!(
+            views.base.iter().flat_map(|g| &g.channels).map(|i| i.header.uuid).collect::<Vec<_>>(),
+            [live_id, id]
+        );
+        if has_xtream {
+            let groups = views.xtream.unwrap();
+            let projected = groups.iter().find(|g| g.title.as_ref() == "TMDB").unwrap();
+            assert_ne!(projected.channels[0].header.uuid, id);
+        } else {
+            assert!(views.xtream.is_none());
+        }
+        let empty = CurationRunOutcome::Complete(complete_catalog_evaluation(Vec::new()));
+        let empty_views = target::curation_playlist_views(&target, playlist, &config, empty).unwrap();
+        assert_eq!(empty_views.base.len(), 1);
+        assert_eq!(empty_views.base[0].xtream_cluster, XtreamCluster::Live);
+        assert_ne!(empty_views.publication_plan, PlaylistPublicationPlan::Ordinary);
+    }
+}
+
 fn catalog_test_item(
     title: &str,
     uuid: UUIDType,
@@ -3245,7 +3340,8 @@ fn xtream_base_and_selector_category_projection_are_independent() {
     };
     let config = TraktConfig::from(&dto);
 
-    let views = build_curated_playlist_views(playlist.clone(), &evaluation, &config, false);
+    let views =
+        build_curated_playlist_views(playlist.clone(), &evaluation, &CurationConfig::from(&config), false, true);
 
     assert_eq!(views.base.len(), 1);
     assert_eq!(views.base[0].channels.len(), 1);
@@ -3264,7 +3360,13 @@ fn xtream_base_and_selector_category_projection_are_independent() {
     compatible_dto.include_xtream_base_categories = true;
     compatible_dto.lists[0].create_xtream_category = false;
     compatible_dto.lists[0].category_name = None;
-    let compatible = build_curated_playlist_views(playlist, &evaluation, &TraktConfig::from(&compatible_dto), false);
+    let compatible = build_curated_playlist_views(
+        playlist,
+        &evaluation,
+        &CurationConfig::from(&TraktConfig::from(&compatible_dto)),
+        false,
+        true,
+    );
     assert_eq!(compatible.base[0].channels.len(), 2);
     assert_eq!(compatible.xtream.expect("Xtream view").len(), 1, "selection-only selector creates no category");
 }
@@ -3313,7 +3415,8 @@ fn curation_policy_truth_table_covers_a_through_h() {
             charts: Vec::new(),
         });
 
-        let views = build_curated_playlist_views(playlist.clone(), &evaluation, &config, false);
+        let views =
+            build_curated_playlist_views(playlist.clone(), &evaluation, &CurationConfig::from(&config), false, true);
         let base_items = views.base.iter().flat_map(|group| &group.channels).collect::<Vec<_>>();
         let expected_base_count = if catalog_selection == TraktCatalogSelection::Full { 2 } else { 1 };
         assert_eq!(base_items.len(), expected_base_count, "case {case} selected catalog");
@@ -3402,7 +3505,7 @@ fn large_catalog_projection_smoke_keeps_two_explicit_views_bounded() {
         ..TraktConfigDto::default()
     });
 
-    let views = build_curated_playlist_views(playlist, &evaluation, &config, false);
+    let views = build_curated_playlist_views(playlist, &evaluation, &CurationConfig::from(&config), false, true);
 
     assert_eq!(views.base[0].channels.len(), CATALOG_SIZE);
     let xtream = views.xtream.expect("Xtream view");
@@ -3881,6 +3984,7 @@ mod quality_rejection_fallback {
 
     fn xtream_target(use_memory_cache: bool) -> Arc<ConfigTarget> {
         Arc::new(ConfigTarget {
+            curation: None,
             id: 1,
             enabled: true,
             name: "quality-target".to_string(),

backend/processing/src/processor/playlist/tests/curation_publication.rs

@@ -0,0 +1,516 @@
+//! Real HTTPS adapters -> eligible catalog -> target finalization -> disk/cache/watch publication.
+//! DNS overrides and a fixture-only CA keep the production TMDB origin and certificate validation intact.
+mod client_diagnostics;
+mod item_limit;
+mod tls_profile;
+use super::{
+    curation_effect_gate::{app_config, file_snapshot, processing_context},
+    *,
+};
+use serde_json::json;
+use shared::{
+    model::{SeriesStreamProperties, StreamProperties, VideoStreamProperties},
+    utils::hash_string,
+};
+use std::{collections::BTreeMap, sync::Mutex as StdMutex, time::Duration};
+use tempfile::{tempdir, TempDir};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpListener,
+    task::JoinHandle,
+};
+use tokio_rustls::{rustls, TlsAcceptor};
+use tuliprox_repository::{load_m3u_target_storage, load_xtream_target_storage};
+
+const MOVIES: &str = "/3/trending/movie/week?language=en-US&page=1";
+const MOVIES_2: &str = "/3/trending/movie/week?language=en-US&page=2";
+const TV: &str = "/3/trending/tv/day?language=en-US&page=1";
+const TRAKT: &str = "/movies/popular?page=1&limit=100";
+const MOVIE_PAGE: &str =
+    r#"{"page":1,"total_pages":9,"total_results":99,"results":[{"id":8,"media_type":"movie"},{"id":8}]}"#;
+const MOVIE_PAGE_2: &str = r#"{"page":2,"total_pages":2,"total_results":2,"results":[{"id":8},{"id":7}]}"#;
+const TV_PAGE: &str = r#"{"page":1,"total_pages":1,"total_results":1,"results":[{"id":7,"media_type":"tv"}]}"#;
+const TRAKT_PAGE: &str = r#"[{"title":"Unselected","ids":{"tmdb":99,"trakt":1,"slug":"unselected"}},{"title":"First","ids":{"tmdb":7,"trakt":2,"slug":"first"}}]"#;
+const EMPTY: &str = r#"{"page":1,"total_pages":0,"total_results":0,"results":[]}"#;
+
+struct DiscoveryServer {
+    client: reqwest::Client,
+    tmdb_client: reqwest::Client,
+    certificate: reqwest::Certificate,
+    address: std::net::SocketAddr,
+    replies: Arc<StdMutex<BTreeMap<String, Vec<u8>>>>,
+    requests: Arc<StdMutex<Vec<String>>>,
+    task: JoinHandle<()>,
+}
+
+impl DiscoveryServer {
+    async fn start() -> Self {
+        let cert =
+            rcgen::generate_simple_self_signed(vec!["api.themoviedb.org".into(), "api.trakt.tv".into()]).unwrap();
+        let tls = rustls::ServerConfig::builder_with_provider(Arc::new(rustls::crypto::aws_lc_rs::default_provider()))
+            .with_safe_default_protocol_versions()
+            .unwrap()
+            .with_no_client_auth()
+            .with_single_cert(
+                vec![cert.cert.der().clone()],
+                rustls::pki_types::PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()).into(),
+            )
+            .unwrap();
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let client = reqwest::Client::builder()
+            .no_proxy()
+            .add_root_certificate(reqwest::Certificate::from_der(cert.cert.der()).unwrap())
+            .resolve("api.themoviedb.org", address)
+            .resolve("api.trakt.tv", address)
+            .timeout(Duration::from_secs(3))
+            .build()
+            .unwrap();
+        let profile_directory = tempdir().unwrap();
+        let tmdb_client =
+            tuliprox_core::utils::network::request::create_tmdb_client(&app_config(profile_directory.path()))
+                .unwrap()
+                .no_proxy()
+                .add_root_certificate(reqwest::Certificate::from_der(cert.cert.der()).unwrap())
+                .resolve("api.themoviedb.org", address)
+                .build()
+                .unwrap();
+        let replies = Arc::new(StdMutex::new(BTreeMap::from([
+            (MOVIES.into(), response(200, MOVIE_PAGE)),
+            (MOVIES_2.into(), response(200, MOVIE_PAGE_2)),
+            (TV.into(), response(200, TV_PAGE)),
+            (TRAKT.into(), response(200, TRAKT_PAGE)),
+        ])));
+        let requests = Arc::new(StdMutex::new(Vec::new()));
+        let responses = Arc::clone(&replies);
+        let recorded = Arc::clone(&requests);
+        let task = tokio::spawn(async move {
+            let acceptor = TlsAcceptor::from(Arc::new(tls));
+            loop {
+                let (tcp, _) = listener.accept().await.unwrap();
+                let Ok(mut stream) = acceptor.accept(tcp).await else { continue };
+                let mut bytes = Vec::new();
+                while !bytes.windows(4).any(|w| w == b"\r\n\r\n") {
+                    let mut chunk = [0; 1024];
+                    let count = stream.read(&mut chunk).await.unwrap();
+                    assert!(count > 0 && bytes.len() < 16_384, "bounded fixture request headers");
+                    bytes.extend_from_slice(&chunk[..count]);
+                }
+                let request = String::from_utf8(bytes).unwrap();
+                let path = request.split_whitespace().nth(1).unwrap();
+                let response = responses.lock().unwrap().get(path).cloned().unwrap_or_else(|| response(404, "{}"));
+                recorded.lock().unwrap().push(request);
+                let _ = stream.write_all(&response).await;
+                let _ = stream.shutdown().await;
+            }
+        });
+        Self {
+            client,
+            tmdb_client,
+            certificate: reqwest::Certificate::from_der(cert.cert.der()).unwrap(),
+            address,
+            replies,
+            requests,
+            task,
+        }
+    }
+
+    fn reply(&self, path: &str, status: u16, body: &str) { self.reply_raw(path, response(status, body)); }
+
+    fn reply_raw(&self, path: &str, body: Vec<u8>) { self.replies.lock().unwrap().insert(path.into(), body); }
+}
+
+fn response(status: u16, body: &str) -> Vec<u8> {
+    format!("HTTP/1.1 {status} Fixture\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).into_bytes()
+}
+
+impl Drop for DiscoveryServer {
+    fn drop(&mut self) { self.task.abort(); }
+}
+
+fn catalog() -> Vec<PlaylistGroup> {
+    let mut groups = Vec::new();
+    for (cluster, rows) in [
+        (XtreamCluster::Live, vec![(101, "Live", 0)]),
+        (
+            XtreamCluster::Video,
+            vec![(201, "First", 7), (202, "Local alias", 7), (203, "Second", 8), (204, "Unselected", 99)],
+        ),
+        (XtreamCluster::Series, vec![(301, "Show", 7), (302, "Other show", 9)]),
+    ] {
+        let group = match cluster {
+            XtreamCluster::Live => "Live",
+            XtreamCluster::Video => "Movies",
+            XtreamCluster::Series => "Series",
+        };
+        let mut channels = Vec::new();
+        for (id, title, tmdb) in rows {
+            let properties = match cluster {
+                XtreamCluster::Live => None,
+                XtreamCluster::Video => Some(StreamProperties::Video(Box::new(VideoStreamProperties {
+                    tmdb: Some(tmdb),
+                    ..Default::default()
+                }))),
+                XtreamCluster::Series => Some(StreamProperties::Series(Box::new(SeriesStreamProperties {
+                    tmdb: Some(tmdb),
+                    ..Default::default()
+                }))),
+            };
+            let item_type = match cluster {
+                XtreamCluster::Live => PlaylistItemType::Live,
+                XtreamCluster::Video => PlaylistItemType::Video,
+                XtreamCluster::Series => PlaylistItemType::SeriesInfo,
+            };
+            let header = PlaylistItemHeader {
+                id: id.to_string().intern(),
+                uuid: hash_string(title),
+                name: title.intern(),
+                title: title.intern(),
+                group: group.intern(),
+                url: format!("http://media.invalid/{id}.mkv").intern(),
+                item_type,
+                xtream_cluster: cluster,
+                epg_channel_id: (cluster == XtreamCluster::Live).then(|| "fixture.live".intern()),
+                additional_properties: properties,
+                ..Default::default()
+            };
+            channels.push(PlaylistItem { header });
+            if cluster == XtreamCluster::Series {
+                let title_episode = format!("{title} S01E01");
+                channels.push(PlaylistItem {
+                    header: PlaylistItemHeader {
+                        id: (id + 100).to_string().intern(),
+                        uuid: hash_string(&title_episode),
+                        name: title_episode.as_str().intern(),
+                        title: title_episode.as_str().intern(),
+                        group: group.intern(),
+                        item_type: PlaylistItemType::Series,
+                        parent_code: hash_string(title).to_string().intern(),
+                        xtream_cluster: cluster,
+                        url: format!("http://media.invalid/{}.mkv", id + 100).intern(),
+                        additional_properties: Some(StreamProperties::Episode(Box::new(
+                            serde_json::from_value(json!({
+                                "episode_id": id + 100, "episode": 1, "season": 1
+                            }))
+                            .unwrap(),
+                        ))),
+                        ..Default::default()
+                    },
+                });
+            }
+        }
+        groups.push(PlaylistGroup {
+            id: u32::try_from(groups.len() + 1).unwrap(),
+            title: group.intern(),
+            xtream_cluster: cluster,
+            channels,
+        });
+    }
+    groups
+}
+
+fn target(policy: &str, mixed: bool, xtream: bool) -> ConfigTargetDto {
+    let mut value = json!({"name":"publication", "watch":[".*"], "use_memory_cache":true,
+    "output":[{"type":"m3u","filename":"published.m3u"},{"type":"strm","directory":"strm","flat":true,"cleanup":true}],
+    "curation":{"catalog_selection":policy,"tmdb":{"api":{"access_token":"fixture-discovery-token"},"trending":[
+        {"kind":"movie","time_window":"week","limit":100,"create_xtream_category":xtream,"category_name":"TMDB Movies"},
+        {"kind":"tv","time_window":"day","limit":100,"create_xtream_category":xtream,"category_name":"TMDB TV"}
+    ]}}});
+    if xtream {
+        value["output"].as_array_mut().unwrap().push(json!({"type":"xtream"}));
+    }
+    if mixed {
+        value["curation"]["trakt"] = json!({"api":{"api_key":"fixture-trakt-client"},"charts":[
+            {"kind":"movies","chart":"popular","category_name":"Trakt Movies","tmdb_only":true}
+        ]});
+    }
+    let mut dto: ConfigTargetDto = serde_json::from_value(value).unwrap();
+    dto.prepare(1, None, None).unwrap();
+    dto
+}
+
+struct Publication {
+    directory: TempDir,
+    context: Arc<PlaylistProcessingContext<shared::model::NoopSink>>,
+    target: ConfigTarget,
+    tmdb_client: reqwest::Client,
+}
+
+impl Publication {
+    fn new(server: &DiscoveryServer, dto: &ConfigTargetDto) -> Self {
+        let directory = tempdir().unwrap();
+        let mut context = processing_context(app_config(directory.path()), Some(Arc::new(PlaylistStorageState::new())));
+        context.client = server.client.clone();
+        Self {
+            directory,
+            context: Arc::new(context),
+            target: ConfigTarget::from(dto),
+            tmdb_client: server.tmdb_client.clone(),
+        }
+    }
+
+    async fn publish(&self) -> Result<(), Vec<TuliproxError>> { self.publish_catalog(catalog()).await }
+
+    async fn publish_catalog(&self, playlist: Vec<PlaylistGroup>) -> Result<(), Vec<TuliproxError>> {
+        let mut channel = shared::model::EpgChannel::new("fixture.live".intern());
+        channel.title = Some("Fixture Live".intern());
+        channel.programmes.push(shared::model::EpgProgramme::new_all(
+            2_000_000_000,
+            2_000_003_600,
+            "fixture.live".intern(),
+            playlist.first().and_then(|group| group.channels.first()).map(|item| Arc::clone(&item.header.title)),
+            None,
+            None,
+        ));
+        let prepared = PreparedTarget {
+            target: self.target.clone(),
+            playlist,
+            epg: vec![tuliprox_core::model::Epg {
+                priority: 0,
+                logo_override: false,
+                attributes: None,
+                children: vec![Arc::new(channel)],
+            }],
+            processing: PipelineStats::default(),
+            accepted_empty_clusters: ClusterFlags::empty(),
+            library_empty: tuliprox_repository::LibraryEmptyPublication::None,
+        };
+        let (result, errors) = tokio::time::timeout(
+            Duration::from_secs(5),
+            target::finalize_prepared_target_with_tmdb(Arc::clone(&self.context), prepared, Some(&self.tmdb_client)),
+        )
+        .await
+        .expect("bounded publication");
+        assert!(errors.is_empty(), "unexpected non-curation errors: {errors:?}");
+        result
+    }
+
+    async fn xtream_rows(&self) -> Vec<XtreamPlaylistItem> {
+        let storage = load_xtream_target_storage(&self.context.config, &self.target).await.unwrap();
+        storage
+            .live
+            .iter()
+            .chain(storage.vod.iter())
+            .chain(storage.series.iter())
+            .map(|(_, item)| item.clone())
+            .collect()
+    }
+
+    async fn cache_signature(&self) -> Vec<String> {
+        let cache = self.context.playlist_state.as_ref().unwrap().data.read().await;
+        let Some(storage) = cache.get(&self.target.name) else { return Vec::new() };
+        let mut values = vec![format!(
+            "target:{}:xtream={}:m3u={}:mapping={}",
+            self.target.name,
+            storage.xtream.is_some(),
+            storage.m3u.is_some(),
+            storage.id_mapping.is_some()
+        )];
+        if let Some(mapping) = &storage.id_mapping {
+            values.extend(mapping.iter().map(|(key, record)| format!("mapping:{key:?}:{record:?}")));
+        }
+        if let Some(xtream) = &storage.xtream {
+            values.extend(
+                xtream
+                    .live
+                    .iter()
+                    .chain(xtream.vod.iter())
+                    .chain(xtream.series.iter())
+                    .map(|(_, item)| format!("{item:?}")),
+            );
+        }
+        if let Some(m3u) = &storage.m3u {
+            values.extend(m3u.iter().map(|(_, item)| format!("{item:?}")));
+        }
+        values.sort();
+        values
+    }
+
+    async fn identity_signature(&self) -> BTreeMap<String, (u32, u32)> {
+        let cache = self.context.playlist_state.as_ref().unwrap().data.read().await;
+        cache[&self.target.name]
+            .id_mapping
+            .as_ref()
+            .unwrap()
+            .iter()
+            .map(|(_, record)| (record.uuid.to_string(), (record.virtual_id.get(), record.parent_virtual_id.get())))
+            .collect()
+    }
+
+    fn strm_contents(&self) -> Vec<String> {
+        file_snapshot(&self.directory.path().join("strm"))
+            .into_iter()
+            .filter(|(path, _)| path.extension().is_some_and(|ext| ext == "strm"))
+            .map(|(_, body)| String::from_utf8(body).unwrap())
+            .collect()
+    }
+}
+
+#[tokio::test]
+async fn tmdb_https_to_publication_preserves_rank_subjects_series_and_output_boundaries() {
+    let server = DiscoveryServer::start().await;
+    let run = Publication::new(&server, &target("curated", false, true));
+    run.publish().await.unwrap();
+    let rows = run.xtream_rows().await;
+    let mut projected = rows.iter().filter(|item| item.group.as_ref() == "TMDB Movies").collect::<Vec<_>>();
+    projected.sort_by_key(|item| item.source_ordinal);
+    assert_eq!(
+        projected.iter().map(|item| item.title.as_ref()).collect::<Vec<_>>(),
+        ["Second", "First", "Local alias"]
+    );
+    assert!(rows.iter().any(|item| item.group.as_ref() == "TMDB TV" && item.title.as_ref() == "Show"));
+    assert!(!rows.iter().any(|item| item.title.as_ref() == "Unselected" || item.title.starts_with("Other show")));
+    let episode = rows
+        .iter()
+        .find(|item| item.group.as_ref() == "TMDB TV" && item.item_type == PlaylistItemType::Series)
+        .expect("projected episode closure");
+    let root = rows
+        .iter()
+        .find(|item| item.group.as_ref() == "TMDB TV" && item.item_type == PlaylistItemType::SeriesInfo)
+        .unwrap();
+    assert_eq!(
+        episode.parent_code.as_ref(),
+        hash_string(&format!("trakt-category:TMDB TV:{}", hash_string("Show"))).to_string()
+    );
+    {
+        let cache = run.context.playlist_state.as_ref().unwrap().data.read().await;
+        let mapping = cache[&run.target.name].id_mapping.as_ref().unwrap();
+        let record =
+            mapping.iter().map(|(_, record)| record).find(|record| record.virtual_id == episode.virtual_id).unwrap();
+        assert_eq!(record.parent_virtual_id, root.virtual_id, "persisted episode points to the projected series");
+    }
+    let m3u = load_m3u_target_storage(&run.context.config, &run.target).await.unwrap();
+    assert!(m3u.iter().all(|(_, item)| !item.group.starts_with("TMDB")));
+    let text = std::fs::read_to_string(run.directory.path().join("published.m3u")).unwrap();
+    for name in ["Live", "First", "Second", "Local alias", "Show S01E01"] {
+        assert!(text.contains(name), "missing {name}");
+    }
+    assert!(!text.contains("Unselected") && !text.contains("Other show") && !text.contains("TMDB"));
+    let strm = run.strm_contents();
+    for id in [201, 202, 203, 401] {
+        assert_eq!(strm.iter().filter(|body| body.contains(&format!("/{id}.mkv"))).count(), 1);
+    }
+    assert!(!strm.iter().any(|body| body.contains("/204.mkv") || body.contains("/402.mkv")));
+    let before = run.identity_signature().await;
+    run.publish().await.unwrap();
+    assert_eq!(run.identity_signature().await, before, "repeat publication retains virtual IDs and aliases");
+    let requests = server.requests.lock().unwrap();
+    assert_eq!(requests.len(), 6, "two movie pages and one TV page per run, no per-item fetch");
+    assert!(requests.iter().all(|r| r.to_lowercase().contains("authorization: bearer fixture-discovery-token")));
+}
+
+#[tokio::test]
+async fn mixed_https_publication_unions_subjects_without_duplicating_base_outputs_under_both_policies() {
+    for policy in ["full", "curated"] {
+        let server = DiscoveryServer::start().await;
+        let run = Publication::new(&server, &target(policy, true, true));
+        run.publish().await.unwrap();
+        let rows = run.xtream_rows().await;
+        assert!(rows.iter().any(|item| item.group.as_ref() == "Trakt Movies" && item.title.as_ref() == "Unselected"));
+        assert!(rows.iter().any(|item| item.group.as_ref() == "TMDB Movies" && item.title.as_ref() == "First"));
+        assert_eq!(rows.iter().any(|item| item.title.as_ref() == "Other show"), policy == "full");
+        let m3u = load_m3u_target_storage(&run.context.config, &run.target).await.unwrap();
+        for name in ["First", "Local alias", "Unselected"] {
+            assert_eq!(m3u.iter().filter(|(_, item)| item.title.as_ref() == name).count(), 1, "union once: {name}");
+        }
+        assert!(m3u.iter().all(|(_, item)| !item.group.starts_with("TMDB") && !item.group.starts_with("Trakt")));
+        assert_eq!(server.requests.lock().unwrap().len(), 4);
+    }
+}
+
+#[tokio::test]
+async fn tmdb_https_selection_only_publishes_m3u_strm_without_an_xtream_output() {
+    let server = DiscoveryServer::start().await;
+    let run = Publication::new(&server, &target("curated", false, false));
+    run.publish().await.unwrap();
+    let text = std::fs::read_to_string(run.directory.path().join("published.m3u")).unwrap();
+    assert!(text.contains("First") && text.contains("Show S01E01") && text.contains("Live"));
+    assert!(!text.contains("Unselected") && !text.contains("TMDB"));
+    assert!(!run.strm_contents().is_empty());
+    let cache = run.context.playlist_state.as_ref().unwrap().data.read().await;
+    assert!(cache[&run.target.name].xtream.is_none());
+    assert_eq!(server.requests.lock().unwrap().len(), 3);
+}
+
+#[tokio::test]
+async fn tmdb_https_empty_and_no_match_clear_published_vod_series_but_keep_live() {
+    for body in
+        [EMPTY, r#"{"page":1,"total_pages":1,"total_results":1,"results":[{"id":999,"title":"First","name":"Show"}]}"#]
+    {
+        let server = DiscoveryServer::start().await;
+        let run = Publication::new(&server, &target("curated", false, true));
+        run.publish().await.unwrap();
+        assert!(!run.strm_contents().is_empty());
+        server.reply(MOVIES, 200, body);
+        server.reply(TV, 200, body);
+        run.publish().await.unwrap();
+        let rows = run.xtream_rows().await;
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].title.as_ref(), "Live");
+        let m3u = load_m3u_target_storage(&run.context.config, &run.target).await.unwrap();
+        assert_eq!(m3u.len(), 1);
+        let strm = run.strm_contents();
+        assert_eq!(strm.len(), 1, "Live survives; only VOD and series files are cleared");
+        assert!(strm[0].contains("/101.mkv"));
+        let watch: std::collections::BTreeSet<Arc<str>> = tuliprox_core::utils::binary_deserialize(
+            &std::fs::read(run.directory.path().join("publication.groups.bin")).unwrap(),
+        )
+        .unwrap();
+        assert_eq!(watch.iter().map(AsRef::as_ref).collect::<Vec<&str>>(), ["Live"]);
+        let cache = run.context.playlist_state.as_ref().unwrap().data.read().await;
+        let xtream = cache[&run.target.name].xtream.as_ref().unwrap();
+        assert!(xtream.vod.is_empty() && xtream.series.is_empty());
+    }
+}
+
+#[tokio::test]
+async fn either_https_failure_retains_every_published_file_id_cache_and_watch_in_full_and_curated() {
+    for policy in ["full", "curated"] {
+        for (route, status, body) in [
+            (MOVIES, 503, "{}"),
+            (TRAKT, 503, "{}"),
+            (MOVIES, 200, r#"{"page":1,"total_pages":1,"total_results":1,"results":[{"id":0}]}"#),
+        ] {
+            let server = DiscoveryServer::start().await;
+            let run = Publication::new(&server, &target(policy, true, true));
+            let initial = file_snapshot(run.directory.path());
+            server.reply(route, status, body);
+            assert!(run.publish().await.is_err(), "first run must not publish a successful sibling");
+            assert_eq!(file_snapshot(run.directory.path()), initial);
+            assert!(run.cache_signature().await.is_empty());
+            server.reply(route, 200, if route == TRAKT { TRAKT_PAGE } else { MOVIE_PAGE });
+            run.publish().await.unwrap();
+            let files = file_snapshot(run.directory.path());
+            let cache = run.cache_signature().await;
+            assert!(!files.is_empty() && !cache.is_empty());
+            server.reply(route, status, body);
+            assert!(run.publish().await.is_err());
+            assert_eq!(file_snapshot(run.directory.path()), files);
+            assert_eq!(run.cache_signature().await, cache);
+        }
+    }
+}
+
+#[tokio::test]
+async fn legacy_to_canonical_trakt_migration_retains_published_virtual_ids_and_aliases() {
+    let server = DiscoveryServer::start().await;
+    let mut canonical = target("curated", true, true);
+    canonical.curation.as_mut().unwrap().tmdb = None;
+    let mut legacy = canonical.clone();
+    let curation = legacy.curation.take().unwrap();
+    let mut value = serde_json::to_value(curation.trakt.unwrap()).unwrap();
+    value["catalog_selection"] = serde_json::to_value(curation.catalog_selection).unwrap();
+    for output in &mut legacy.output {
+        if let TargetOutputDto::Xtream(xtream) = output {
+            xtream.trakt = Some(serde_json::from_value(value.clone()).unwrap());
+        }
+    }
+    legacy.prepare(1, None, None).unwrap();
+    let mut run = Publication::new(&server, &legacy);
+    run.publish().await.unwrap();
+    let before = run.identity_signature().await;
+    run.target = ConfigTarget::from(&canonical);
+    run.publish().await.unwrap();
+    assert_eq!(run.identity_signature().await, before);
+    assert_eq!(server.requests.lock().unwrap().len(), 2);
+}

backend/processing/src/processor/playlist/tests/curation_publication/client_diagnostics.rs

@@ -0,0 +1,108 @@
+use super::*;
+use target::build_target_tmdb_client;
+use tuliprox_core::{model::ProxyConfig, utils::network::request::create_tmdb_client};
+
+#[tokio::test]
+async fn curation_tmdb_client_construction_reports_only_safe_phases_and_keeps_publication_closed() {
+    for policy in ["full", "curated"] {
+        for phase in ["phase=profile_configuration", "phase=client_build"] {
+            let server = DiscoveryServer::start().await;
+            let run = Publication::new(&server, &target(policy, true, true));
+            for previously_published in [false, true] {
+                if previously_published {
+                    run.publish().await.unwrap();
+                }
+                let files = file_snapshot(run.directory.path());
+                let cache = run.cache_signature().await;
+                assert_eq!(!cache.is_empty(), previously_published);
+                let before = server.requests.lock().unwrap().len();
+                let config = app_config(run.directory.path());
+                let mut settings = config.config.load().as_ref().clone();
+                if phase == "phase=profile_configuration" {
+                    settings.proxy = Some(ProxyConfig {
+                        url: "http://[synthetic-proxy-secret".into(),
+                        username: Some("synthetic-user-secret".into()),
+                        password: Some("synthetic-password-secret".into()),
+                    });
+                }
+                config.config.store(Arc::new(settings));
+                let mut diagnostics = Vec::new();
+                let tmdb_client = build_target_tmdb_client(
+                    &run.target,
+                    || {
+                        create_tmdb_client(&config).map(|builder| {
+                            // A controlled reqwest build error, not a new runtime fallback.
+                            builder.user_agent("synthetic-header-secret\ninvalid")
+                        })
+                    },
+                    |label| diagnostics.push(label),
+                );
+                assert!(tmdb_client.is_none());
+                // Exact allowlist, independent of Display/source and global log sanitization:
+                // no sanitizer is invoked between construction failure and this reporter.
+                assert_eq!(diagnostics, [phase]);
+                assert!(!diagnostics[0].contains("synthetic-"));
+                let mut changed = catalog();
+                changed[0].channels[0].header.title = "Must not publish".intern();
+                let prepared = PreparedTarget {
+                    target: run.target.clone(),
+                    playlist: changed,
+                    epg: Vec::new(),
+                    processing: PipelineStats::default(),
+                    accepted_empty_clusters: ClusterFlags::empty(),
+                    library_empty: tuliprox_repository::LibraryEmptyPublication::None,
+                };
+                let (result, errors) = tokio::time::timeout(
+                    Duration::from_secs(5),
+                    target::finalize_prepared_target_with_tmdb(
+                        Arc::clone(&run.context),
+                        prepared,
+                        tmdb_client.as_ref(),
+                    ),
+                )
+                .await
+                .unwrap();
+                assert!(result.is_err());
+                assert!(errors.is_empty());
+                assert_eq!(file_snapshot(run.directory.path()), files);
+                assert_eq!(run.cache_signature().await, cache);
+                let requests = server.requests.lock().unwrap();
+                assert_eq!(requests.len(), before + 1, "complete Trakt sibling cannot authorize publication");
+                assert!(requests[before].starts_with(&format!("GET {TRAKT} ")));
+            }
+        }
+    }
+}
+
+#[test]
+fn curation_tmdb_disabled_or_selectorless_skips_construction_and_diagnostics() {
+    for state in ["absent", "block_disabled", "source_disabled", "source_absent", "no_selectors"] {
+        let mut dto = target("full", false, true);
+        match state {
+            "absent" => dto.curation = None,
+            "block_disabled" => dto.curation.as_mut().unwrap().enabled = false,
+            "source_disabled" => dto.curation.as_mut().unwrap().tmdb.as_mut().unwrap().enabled = false,
+            "source_absent" => dto.curation.as_mut().unwrap().tmdb = None,
+            "no_selectors" => dto.curation.as_mut().unwrap().tmdb.as_mut().unwrap().trending.clear(),
+            _ => unreachable!(),
+        }
+        assert!(build_target_tmdb_client(
+            &ConfigTarget::from(&dto),
+            || panic!("{state} must not construct a client"),
+            |_| panic!("{state} must not emit a construction diagnostic"),
+        )
+        .is_none());
+    }
+}
+
+#[test]
+fn curation_tmdb_successful_construction_has_no_failure_diagnostic() {
+    let directory = tempdir().unwrap();
+    let config = app_config(directory.path());
+    assert!(build_target_tmdb_client(
+        &ConfigTarget::from(&target("full", false, true)),
+        || create_tmdb_client(&config),
+        |_| panic!("successful construction is not a failure"),
+    )
+    .is_some());
+}

backend/processing/src/processor/playlist/tests/curation_publication/item_limit.rs

@@ -0,0 +1,314 @@
+use super::*;
+
+fn changed_catalog() -> Vec<PlaylistGroup> {
+    let mut playlist = catalog();
+    for group in &mut playlist {
+        for item in &mut group.channels {
+            item.header.title = format!("Candidate {}", item.header.title).intern();
+            item.header.url = format!("http://candidate.invalid/{}.mkv", item.header.id).intern();
+        }
+    }
+    playlist
+}
+
+fn fail_late(server: &DiscoveryServer, failure: &str) {
+    match failure {
+        "status" => server.reply(MOVIES_2, 503, "sensitive-body-must-not-appear"),
+        "wire" => {
+            server.reply(MOVIES_2, 200, r#"{"page":2,"total_pages":2,"total_results":2,"results":[{"id":7},{"id":0}]}"#)
+        }
+        "no-progress" => {
+            server.reply(MOVIES_2, 200, r#"{"page":2,"total_pages":2,"total_results":1,"results":[{"id":8}]}"#)
+        }
+        "empty" => server.reply(MOVIES_2, 200, r#"{"page":2,"total_pages":2,"total_results":0,"results":[]}"#),
+        "body" => server.reply_raw(
+            MOVIES_2,
+            format!("HTTP/1.1 200 OK\r\nContent-Length: 10000\r\nConnection: close\r\n\r\n{MOVIE_PAGE_2}").into_bytes(),
+        ),
+        "gzip" | "gzip-transport" | "gzip-suffix" => {
+            let mut compressed = tuliprox_core::utils::compression_utils::compress_string(MOVIE_PAGE_2).unwrap();
+            if failure == "gzip" {
+                compressed.truncate(compressed.len() - 8);
+            } else if failure == "gzip-suffix" {
+                let mut suffix = tuliprox_core::utils::compression_utils::compress_string(" ").unwrap();
+                suffix.truncate(suffix.len() - 8);
+                compressed.extend(suffix);
+            }
+            let advertised = compressed.len() + usize::from(failure == "gzip-transport");
+            let mut response = format!("HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {advertised}\r\nConnection: close\r\n\r\n").into_bytes();
+            response.extend(compressed);
+            server.reply_raw(MOVIES_2, response);
+        }
+        "bytes" => {
+            server.reply(MOVIES_2, 200, &format!("{MOVIE_PAGE_2}{}", " ".repeat(1024 * 1024 + 1 - MOVIE_PAGE_2.len())))
+        }
+        "requests" => {
+            for p in 1..=33 {
+                let id = if p == 1 {
+                    8
+                } else if p == 2 {
+                    7
+                } else {
+                    1000 + p
+                };
+                server.reply(
+                    &format!("/3/trending/movie/week?language=en-US&page={p}"),
+                    200,
+                    &json!({"page":p,"total_pages":33,"total_results":1,"results":[{"id":id}]}).to_string(),
+                );
+            }
+        }
+        _ => panic!("unknown fixture failure"),
+    }
+}
+
+#[tokio::test]
+async fn curation_late_failure_keeps_all_files_epg_cache_mapping_and_watches_in_both_policies() {
+    for policy in ["full", "curated"] {
+        for failure in [
+            "status",
+            "wire",
+            "no-progress",
+            "empty",
+            "body",
+            "gzip",
+            "gzip-transport",
+            "gzip-suffix",
+            "bytes",
+            "requests",
+        ] {
+            let server = DiscoveryServer::start().await;
+            let mut dto = target(policy, true, true);
+            // A selection-only selector is just as required as its successful projecting siblings.
+            dto.curation.as_mut().unwrap().tmdb.as_mut().unwrap().trending[0].create_xtream_category = false;
+            let run = Publication::new(&server, &dto);
+            let successful_replies = server.replies.lock().unwrap().clone();
+            let initial = file_snapshot(run.directory.path());
+            fail_late(&server, failure);
+            assert!(run.publish_catalog(changed_catalog()).await.is_err(), "{policy}/{failure}: first refresh");
+            assert_eq!(file_snapshot(run.directory.path()), initial);
+            assert!(run.context.playlist_state.as_ref().unwrap().data.read().await.is_empty());
+            assert!(run.cache_signature().await.is_empty());
+            let count = server.requests.lock().unwrap().len();
+            assert_eq!(
+                count,
+                if failure == "requests" { 34 } else { 4 },
+                "Trakt and TV are evaluated too, but not published"
+            );
+            *server.replies.lock().unwrap() = successful_replies;
+            run.publish().await.unwrap();
+            let files = file_snapshot(run.directory.path());
+            assert!(
+                files.keys().any(|path| path.to_string_lossy().contains("epg")),
+                "EPG must be in the retention inventory: {:?}",
+                files.keys().collect::<Vec<_>>()
+            );
+            let cache = run.cache_signature().await;
+            assert!(cache.iter().any(|entry| entry.starts_with("mapping:")));
+            let watch_path = run.directory.path().join("publication.groups.bin");
+            let watch_before: std::collections::BTreeSet<Arc<str>> =
+                tuliprox_core::utils::binary_deserialize(&std::fs::read(&watch_path).unwrap()).unwrap();
+            assert!(!watch_before.is_empty());
+            fail_late(&server, failure);
+            assert!(
+                run.publish_catalog(changed_catalog()).await.is_err(),
+                "{policy}/{failure}: refresh with prior state"
+            );
+            assert_eq!(file_snapshot(run.directory.path()), files, "inventory and bytes, not just file sizes");
+            assert_eq!(run.cache_signature().await, cache, "all cached rows and mapping records, including timestamps");
+            let watch_after: std::collections::BTreeSet<Arc<str>> =
+                tuliprox_core::utils::binary_deserialize(&std::fs::read(&watch_path).unwrap()).unwrap();
+            assert_eq!(watch_before, watch_after);
+        }
+    }
+}
+
+#[tokio::test]
+async fn curation_shared_batch_exhaustion_marks_pending_selectors_required_without_129th_get() {
+    for policy in ["full", "curated"] {
+        let server = DiscoveryServer::start().await;
+        let dto = target(policy, false, true);
+        let mut run = Publication::new(&server, &dto);
+        let ordinary = run.target.clone();
+        let source = run.target.curation.as_mut().unwrap().tmdb.as_mut().unwrap();
+        let mut selector = source.trending[0].clone();
+        selector.limit = 1;
+        selector.create_xtream_category = false;
+        source.trending = vec![selector; 129];
+        let failing = run.target.clone();
+        let initial = file_snapshot(run.directory.path());
+        assert!(run.publish().await.is_err());
+        assert_eq!(server.requests.lock().unwrap().len(), 128);
+        assert_eq!(file_snapshot(run.directory.path()), initial);
+        assert!(run.cache_signature().await.is_empty());
+        run.target = ordinary;
+        run.publish().await.unwrap();
+        let before = file_snapshot(run.directory.path());
+        let cache = run.cache_signature().await;
+        let requests = server.requests.lock().unwrap().len();
+        run.target = failing;
+        assert!(run.publish_catalog(changed_catalog()).await.is_err());
+        assert_eq!(server.requests.lock().unwrap().len() - requests, 128);
+        assert_eq!(file_snapshot(run.directory.path()), before);
+        assert_eq!(run.cache_signature().await, cache);
+    }
+}
+
+#[tokio::test]
+async fn curation_limit_changes_keep_surviving_ids_and_independent_daily_weekly_memberships() {
+    const DAILY: &str = "/3/trending/movie/day?language=en-US&page=1";
+    for include_base in [false, true] {
+        let server = DiscoveryServer::start().await;
+        let mut dto = target("curated", false, true);
+        let curation = dto.curation.as_mut().unwrap();
+        curation.include_xtream_base_categories = include_base;
+        let source = curation.tmdb.as_mut().unwrap();
+        let mut daily = source.trending[0].clone();
+        daily.time_window = shared::model::TmdbTrendingTimeWindow::Day;
+        daily.category_name = Some("TMDB Daily".into());
+        daily.limit = 2;
+        source.trending.push(daily);
+        let mut run = Publication::new(&server, &dto);
+        server.reply(DAILY, 200, r#"{"page":1,"total_pages":1,"total_results":2,"results":[{"id":8},{"id":7}]}"#);
+        let mut prior = BTreeMap::new();
+        for (limit, ids) in [(1, vec![7, 7, 8]), (2, vec![7, 7, 8]), (2, vec![8, 7]), (1, vec![8, 7])] {
+            run.target.curation.as_mut().unwrap().tmdb.as_mut().unwrap().trending[0].limit = limit;
+            server.reply(MOVIES, 200, &json!({"page":1,"total_pages":9,"total_results":ids.len(),"results":ids.iter().map(|id| json!({"id":id})).collect::<Vec<_>>()} ).to_string());
+            run.publish().await.unwrap();
+            let rows = run.xtream_rows().await;
+            let alive: BTreeMap<_, _> =
+                rows.iter().map(|row| ((row.group.to_string(), row.title.to_string()), row.virtual_id)).collect();
+            for (subject, id) in &prior {
+                if let Some(current) = alive.get(subject) {
+                    assert_eq!(current, id, "surviving subject/alias {subject:?}");
+                }
+            }
+            prior = alive;
+            let identities = run.identity_signature().await;
+            for title in ["Show", "Show S01E01"] {
+                let uuid = hash_string(&format!("trakt-category:TMDB TV:{}", hash_string(title))).to_string();
+                assert!(identities.contains_key(&uuid));
+            }
+            assert_eq!(rows.iter().any(|row| row.group.as_ref() == "Movies"), include_base);
+            let mut weekly: Vec<_> = rows.iter().filter(|row| row.group.as_ref() == "TMDB Movies").collect();
+            weekly.sort_by_key(|row| row.source_ordinal);
+            assert_eq!(weekly[0].title.as_ref(), if ids[0] == 7 { "First" } else { "Second" });
+            let mut daily: Vec<_> = rows.iter().filter(|row| row.group.as_ref() == "TMDB Daily").collect();
+            daily.sort_by_key(|row| row.source_ordinal);
+            assert_eq!(
+                daily.iter().map(|row| row.title.as_ref()).collect::<Vec<_>>(),
+                ["Second", "First", "Local alias"]
+            );
+            let m3u = load_m3u_target_storage(&run.context.config, &run.target).await.unwrap();
+            for title in ["First", "Local alias", "Second", "Show S01E01"] {
+                assert_eq!(m3u.iter().filter(|(_, item)| item.title.as_ref() == title).count(), 1);
+            }
+            assert!(m3u.iter().all(|(_, item)| !item.group.starts_with("TMDB")));
+        }
+        assert_eq!(server.requests.lock().unwrap().len(), 12, "no extra page after N despite total_pages=9");
+    }
+}
+
+#[tokio::test]
+async fn curation_full_empty_or_no_match_removes_projections_not_ordinary_catalog() {
+    for body in [EMPTY, r#"{"page":1,"total_pages":99,"total_results":2,"results":[{"id":999},{"id":7}]}"#] {
+        let server = DiscoveryServer::start().await;
+        let mut run = Publication::new(&server, &target("full", false, true));
+        run.publish().await.unwrap();
+        for selector in &mut run.target.curation.as_mut().unwrap().tmdb.as_mut().unwrap().trending {
+            selector.limit = 1;
+        }
+        server.reply(MOVIES, 200, body);
+        server.reply(TV, 200, body);
+        run.publish().await.unwrap();
+        let rows = run.xtream_rows().await;
+        assert!(rows.iter().all(|row| !row.group.starts_with("TMDB")));
+        assert!(rows.iter().any(|row| row.title.as_ref() == "Unselected"));
+        assert!(rows.iter().any(|row| row.title.as_ref() == "Live"));
+        assert!(rows.iter().any(|row| row.title.as_ref() == "Other show"));
+        assert_eq!(
+            server.requests.lock().unwrap().len(),
+            5,
+            "do not seek N matches beyond the first unmatched reference"
+        );
+    }
+}
+
+#[tokio::test]
+async fn curation_profile_requires_the_fixture_ca_and_has_no_unconfigured_fallback() {
+    let server = DiscoveryServer::start().await;
+    let mut run = Publication::new(&server, &target("full", false, true));
+    run.tmdb_client = tuliprox_core::utils::network::request::create_tmdb_client(&run.context.config)
+        .unwrap()
+        .no_proxy()
+        .resolve("api.themoviedb.org", server.address)
+        .build()
+        .unwrap();
+    assert!(run.publish().await.is_err(), "untrusted CA fails rather than disabling TLS verification");
+    assert!(server.requests.lock().unwrap().is_empty());
+    assert!(file_snapshot(run.directory.path()).is_empty());
+}
+
+#[tokio::test]
+async fn curation_production_wiring_does_not_fall_back_to_the_generic_client() {
+    let server = DiscoveryServer::start().await;
+    let run = Publication::new(&server, &target("full", false, true));
+    let mut config = (*run.context.config.config.load_full()).clone();
+    config.proxy =
+        Some(tuliprox_core::model::ProxyConfig { url: "invalid proxy".into(), username: None, password: None });
+    run.context.config.config.store(Arc::new(config));
+    let prepared = PreparedTarget {
+        target: run.target.clone(),
+        playlist: catalog(),
+        epg: Vec::new(),
+        processing: PipelineStats::default(),
+        accepted_empty_clusters: ClusterFlags::empty(),
+        library_empty: tuliprox_repository::LibraryEmptyPublication::None,
+    };
+    let (result, _) = finalize_prepared_target(Arc::clone(&run.context), prepared).await;
+    assert!(result.is_err());
+    assert!(
+        server.requests.lock().unwrap().is_empty(),
+        "generic client has the fixture CA and could succeed, but is not used"
+    );
+    assert!(run.cache_signature().await.is_empty());
+    assert!(file_snapshot(run.directory.path()).is_empty());
+}
+
+#[tokio::test]
+async fn curation_does_not_authorize_empty_inputs_and_authorized_empty_still_waits_for_all_selectors() {
+    for authorized in [false, true] {
+        let server = DiscoveryServer::start().await;
+        let run = Publication::new(&server, &target("curated", false, true));
+        server.reply(MOVIES_2, 503, "unavailable");
+        let prepared = PreparedTarget {
+            target: run.target.clone(),
+            playlist: Vec::new(),
+            epg: Vec::new(),
+            processing: PipelineStats::default(),
+            accepted_empty_clusters: if authorized { ClusterFlags::Vod } else { ClusterFlags::empty() },
+            library_empty: tuliprox_repository::LibraryEmptyPublication::None,
+        };
+        let (result, _) =
+            target::finalize_prepared_target_with_tmdb(Arc::clone(&run.context), prepared, Some(&run.tmdb_client))
+                .await;
+        assert_eq!(result.is_err(), authorized);
+        assert_eq!(server.requests.lock().unwrap().len(), if authorized { 3 } else { 0 });
+        assert!(file_snapshot(run.directory.path()).is_empty());
+        assert!(run.cache_signature().await.is_empty());
+    }
+}
+
+#[tokio::test]
+async fn curation_post_admission_writer_failure_remains_best_effort_not_a_rollback() {
+    let server = DiscoveryServer::start().await;
+    let run = Publication::new(&server, &target("curated", false, true));
+    std::fs::create_dir(run.directory.path().join("published.m3u")).unwrap();
+    assert!(run.publish().await.is_err());
+    assert!(!run.xtream_rows().await.is_empty(), "other admitted output was still written");
+    assert_eq!(server.requests.lock().unwrap().len(), 3);
+    assert!(
+        !run.directory.path().join("publication.groups.bin").exists(),
+        "watches do not finalize a failed writer run"
+    );
+}

backend/processing/src/processor/playlist/tests/curation_publication/tls_profile.rs

@@ -0,0 +1,106 @@
+//! Exercise the configured profile against real TLS, without changing host trust or contacting TMDB.
+use super::*;
+use reqwest::header::AUTHORIZATION;
+use tuliprox_core::utils::network::request::{create_client, create_tmdb_client};
+
+fn insecure_config(directory: &std::path::Path) -> Arc<AppConfig> {
+    let config = app_config(directory);
+    let mut settings = config.config.load().as_ref().clone();
+    settings.accept_insecure_ssl_certificates = true;
+    config.config.store(Arc::new(settings));
+    config
+}
+
+fn local_client(builder: reqwest::ClientBuilder, server: &DiscoveryServer, hostname: &str) -> reqwest::Client {
+    builder.no_proxy().resolve(hostname, server.address).timeout(Duration::from_secs(3)).build().unwrap()
+}
+
+#[tokio::test]
+async fn tmdb_profile_tls_rejects_untrusted_certificate_even_with_global_insecure_flag() {
+    let server = DiscoveryServer::start().await;
+    let directory = tempdir().unwrap();
+    let config = insecure_config(directory.path());
+    let client = local_client(create_tmdb_client(&config).unwrap(), &server, "api.themoviedb.org");
+    let result = client
+        .get(format!("https://api.themoviedb.org{MOVIES}"))
+        .header(AUTHORIZATION, "Bearer synthetic-tls-secret")
+        .send()
+        .await;
+    assert!(result.is_err(), "discovery must not inherit the global TLS bypass");
+    assert!(result.unwrap_err().is_connect());
+    assert!(server.requests.lock().unwrap().is_empty(), "TCP/TLS is not an HTTP request or Bearer disclosure");
+}
+
+#[tokio::test]
+async fn tmdb_profile_tls_preserves_trusted_ca_and_checks_hostname_with_global_insecure_flag() {
+    let server = DiscoveryServer::start().await;
+    let directory = tempdir().unwrap();
+    let config = insecure_config(directory.path());
+    for (hostname, trusted_name) in [("api.themoviedb.org", true), ("wrong-name.invalid", false)] {
+        let client = local_client(
+            create_tmdb_client(&config).unwrap().add_root_certificate(server.certificate.clone()),
+            &server,
+            hostname,
+        );
+        let before = server.requests.lock().unwrap().len();
+        let result = client
+            .get(format!("https://{hostname}{MOVIES}"))
+            .header(AUTHORIZATION, "Bearer synthetic-tls-secret")
+            .send()
+            .await;
+        if trusted_name {
+            assert_eq!(result.unwrap().text().await.unwrap(), MOVIE_PAGE);
+            let requests = server.requests.lock().unwrap();
+            assert_eq!(requests.len(), before + 1);
+            assert!(requests[before].to_lowercase().contains("authorization: bearer synthetic-tls-secret"));
+        } else {
+            assert!(result.unwrap_err().is_connect());
+            assert_eq!(server.requests.lock().unwrap().len(), before, "wrong hostname must not receive HTTP");
+        }
+    }
+}
+
+#[tokio::test]
+async fn tmdb_profile_tls_does_not_change_generic_clients_global_insecure_policy() {
+    let server = DiscoveryServer::start().await;
+    let directory = tempdir().unwrap();
+    let config = insecure_config(directory.path());
+    let client = local_client(create_client(&config), &server, "api.trakt.tv");
+    let response = client.get(format!("https://api.trakt.tv{TRAKT}")).send().await.unwrap();
+    assert_eq!(response.text().await.unwrap(), TRAKT_PAGE);
+    assert_eq!(server.requests.lock().unwrap().len(), 1, "generic profile still accepts the untrusted certificate");
+}
+
+#[tokio::test]
+async fn tmdb_profile_tls_failure_blocks_publication_and_retains_artifacts_even_with_complete_trakt() {
+    for policy in ["full", "curated"] {
+        let server = DiscoveryServer::start().await;
+        let mut run = Publication::new(&server, &target(policy, true, true));
+        let config = insecure_config(run.directory.path());
+        let rejected = local_client(create_tmdb_client(&config).unwrap(), &server, "api.themoviedb.org");
+        let trusted = local_client(
+            create_tmdb_client(&config).unwrap().add_root_certificate(server.certificate.clone()),
+            &server,
+            "api.themoviedb.org",
+        );
+        for previously_published in [false, true] {
+            if previously_published {
+                run.tmdb_client = trusted.clone();
+                run.publish().await.unwrap();
+            }
+            let files = file_snapshot(run.directory.path());
+            let cache = run.cache_signature().await;
+            assert_eq!(!cache.is_empty(), previously_published);
+            let before = server.requests.lock().unwrap().len();
+            run.tmdb_client = rejected.clone();
+            let mut changed = catalog();
+            changed[0].channels[0].header.title = "Must not publish".intern();
+            assert!(run.publish_catalog(changed).await.is_err());
+            assert_eq!(file_snapshot(run.directory.path()), files, "all files, IDs and watches retained");
+            assert_eq!(run.cache_signature().await, cache, "no partial cache publication");
+            let requests = server.requests.lock().unwrap();
+            assert_eq!(requests.len(), before + 1, "only the required Trakt sibling reached HTTP; no fallback");
+            assert!(requests[before].starts_with(&format!("GET {TRAKT} ")));
+        }
+    }
+}