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

#![allow(clippy::wildcard_imports)]
use super::*;

pub(crate) fn join_arc_strs(values: &[Arc<str>], separator: &str) -> String {
    let mut result = String::new();
    for value in values {
        if !result.is_empty() {
            result.push_str(separator);
        }
        result.push_str(value.as_ref());
    }
    result
}

pub(crate) fn target_waiting_message(target: &str, input: &str) -> String {
    format!("Target '{target}' is waiting for input '{input}'")
}

pub(crate) fn target_mutated_resources(
    config: &tuliprox_core::model::Config,
    target: &ConfigTarget,
) -> HashSet<PathBuf> {
    let mut resources = HashSet::new();
    if let Some(path) = tuliprox_repository::get_target_storage_path(config, &target.name) {
        resources.insert(path.clean());
    }
    for output in &target.output {
        match output {
            tuliprox_core::model::TargetOutput::M3u(output) => {
                if let Some(path) = tuliprox_core::utils::get_file_path(
                    &config.storage_dir,
                    output.filename.as_deref().map(PathBuf::from),
                ) {
                    resources.insert(path.clean());
                }
            }
            tuliprox_core::model::TargetOutput::Strm(output) => {
                if let Some(path) =
                    tuliprox_core::utils::get_file_path(&config.storage_dir, Some(PathBuf::from(&output.directory)))
                {
                    resources.insert(path.clean());
                }
            }
            tuliprox_core::model::TargetOutput::Xtream(_) | tuliprox_core::model::TargetOutput::HdHomeRun(_) => {}
        }
    }
    resources
}

pub(crate) fn stalker_checkpoint_message(input: &str) -> String {
    format!("Input '{input}': Stalker refresh checkpoint saved; active snapshot remains in service")
}

pub(crate) fn is_target_enabled(target: &ConfigTarget, user_targets: &ProcessTargets) -> bool {
    (!user_targets.enabled && target.enabled) || (user_targets.enabled && user_targets.has_target(target.id))
}

pub(crate) struct TargetJobResult {
    pub(crate) index: usize,
    pub(crate) name: String,
    pub(crate) result: Result<(), Vec<TuliproxError>>,
    pub(crate) errors: Vec<TuliproxError>,
    pub(crate) processing: PipelineStats,
}

pub(crate) fn collect_target_task_result(
    result: Result<TargetJobResult, tokio::task::JoinError>,
    results: &mut Vec<TargetJobResult>,
    errors: &mut Vec<TuliproxError>,
) {
    match result {
        Ok(result) => results.push(result),
        Err(err) => errors.push(TuliproxError::RepositoryPlaylist(format!("Target finalization task failed: {err}"))),
    }
}

pub(crate) async fn wait_for_target_finalizer_slot(
    tasks: &mut JoinSet<TargetJobResult>,
    results: &mut Vec<TargetJobResult>,
    errors: &mut Vec<TuliproxError>,
) {
    if tasks.len() >= MAX_CONCURRENT_TARGET_FINALIZERS {
        if let Some(result) = tasks.join_next().await {
            collect_target_task_result(result, results, errors);
        }
    }
}

#[allow(clippy::too_many_lines)]
pub(crate) async fn process_targets<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &Arc<PlaylistProcessingContext<E, M>>,
    playlists: &mut [FetchedPlaylist<'_>],
    targets: &[&Arc<ConfigTarget>],
    input_stats: &mut HashMap<Arc<str>, InputStats>,
    errors: &mut Vec<TuliproxError>,
    accepted_empty_clusters: ClusterFlags,
    process_parallel: bool,
) -> Vec<TargetStats> {
    if !process_parallel {
        let mut target_stats = Vec::with_capacity(targets.len());
        for (index, target) in targets.iter().enumerate() {
            let consume_input_source = index + 1 == targets.len();
            let result = prepare_playlist_for_target(
                ctx,
                playlists,
                target,
                input_stats,
                errors,
                accepted_empty_clusters,
                consume_input_source,
            )
            .await;
            match result {
                Ok(prepared) => {
                    let processing = prepared.processing.clone();
                    let (result, mut finalization_errors) = finalize_prepared_target(Arc::clone(ctx), prepared).await;
                    errors.append(&mut finalization_errors);
                    match result {
                        Ok(()) => target_stats.push(TargetStats::success_with_processing(&target.name, processing)),
                        Err(mut target_errors) => {
                            target_stats.push(TargetStats::failure_with_processing(&target.name, processing));
                            errors.append(&mut target_errors);
                        }
                    }
                }
                Err(mut target_errors) => {
                    target_stats.push(TargetStats::failure(&target.name));
                    errors.append(&mut target_errors);
                }
            }
        }
        return target_stats;
    }

    let resources = {
        let config = ctx.config.config.load();
        targets.iter().map(|target| target_mutated_resources(&config, target)).collect::<Vec<_>>()
    };
    let mut completion_receivers: Vec<watch::Receiver<bool>> = Vec::with_capacity(targets.len());
    let mut tasks = JoinSet::new();
    let mut results = Vec::with_capacity(targets.len());

    for (index, target) in targets.iter().enumerate() {
        wait_for_target_finalizer_slot(&mut tasks, &mut results, errors).await;
        let predecessors = resources[..index]
            .iter()
            .zip(&completion_receivers)
            .filter(|(earlier, _)| !earlier.is_disjoint(&resources[index]))
            .map(|(_, receiver)| receiver.clone())
            .collect::<Vec<_>>();
        let (completion, receiver) = watch::channel(false);
        completion_receivers.push(receiver);

        match prepare_playlist_for_target(ctx, playlists, target, input_stats, errors, accepted_empty_clusters, false)
            .await
        {
            Ok(prepared) => {
                let processing = prepared.processing.clone();
                let task_ctx = Arc::clone(ctx);
                let target_name = target.name.clone();
                tasks.spawn(async move {
                    for mut predecessor in predecessors {
                        if !*predecessor.borrow() {
                            let _ = predecessor.changed().await;
                        }
                    }
                    let finalized =
                        std::panic::AssertUnwindSafe(finalize_prepared_target(task_ctx, prepared)).catch_unwind().await;
                    completion.send_replace(true);
                    match finalized {
                        Ok((result, errors)) => {
                            TargetJobResult { index, name: target_name, result, errors, processing }
                        }
                        Err(_) => TargetJobResult {
                            index,
                            name: target_name.clone(),
                            result: Err(vec![TuliproxError::RepositoryPlaylist(format!(
                                "Target '{target_name}' finalization panicked"
                            ))]),
                            errors: Vec::new(),
                            processing,
                        },
                    }
                });
            }
            Err(target_errors) => {
                completion.send_replace(true);
                results.push(TargetJobResult {
                    index,
                    name: target.name.clone(),
                    result: Err(target_errors),
                    errors: Vec::new(),
                    processing: PipelineStats::default(),
                });
            }
        }
    }

    while let Some(result) = tasks.join_next().await {
        collect_target_task_result(result, &mut results, errors);
    }
    results.sort_by_key(|result| result.index);

    let mut target_stats = Vec::with_capacity(results.len());
    for mut target_result in results {
        errors.append(&mut target_result.errors);
        match target_result.result {
            Ok(()) => {
                target_stats.push(TargetStats::success_with_processing(&target_result.name, target_result.processing));
            }
            Err(mut target_errors) => {
                target_stats.push(TargetStats::failure_with_processing(&target_result.name, target_result.processing));
                errors.append(&mut target_errors);
            }
        }
    }
    target_stats
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum FinalizationStage {
    Merge,
    Deduplicate,
    Sort,
    AssignChannelNumbers,
    AssignCounters,
}

pub(crate) const FINALIZATION_ORDER: [FinalizationStage; 5] = [
    FinalizationStage::Merge,
    FinalizationStage::Deduplicate,
    FinalizationStage::Sort,
    FinalizationStage::AssignChannelNumbers,
    FinalizationStage::AssignCounters,
];

pub(crate) fn apply_persist_filter(target: &ConfigTarget, groups: &mut Vec<PlaylistGroup>) {
    let Some(filter) = target.filter.persist.as_ref() else {
        return;
    };
    let outcome = retain_filtered_playlist(groups, filter);
    debug!("Target '{}' persist filter outcome: {outcome:?}", target.name);
}

pub(crate) struct PreparedTarget {
    pub(crate) library_empty: tuliprox_repository::LibraryEmptyPublication,
    pub(crate) target: ConfigTarget,
    pub(crate) playlist: Vec<PlaylistGroup>,
    pub(crate) epg: Vec<Epg>,
    pub(crate) processing: PipelineStats,
    pub(crate) accepted_empty_clusters: ClusterFlags,
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn prepare_playlist_for_target<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    playlists: &mut [FetchedPlaylist<'_>],
    target: &ConfigTarget,
    stats: &mut HashMap<Arc<str>, InputStats>,
    errors: &mut Vec<TuliproxError>,
    accepted_empty_clusters: ClusterFlags,
    consume_input_source: bool,
) -> Result<PreparedTarget, Vec<TuliproxError>> {
    debug_if_enabled!("Processing order is {}", &target.processing_order);
    log_memory_snapshot(format!("target '{}' start", target.name).as_str());

    let mut duplicates: HashSet<UUIDType> = HashSet::new();
    let mut new_epg = vec![];
    let mut new_playlist: Vec<PlaylistGroup> = vec![];
    let mut aggregate_outcome = PipelineOutcome::default();
    let mut has_empty_library = false;
    let mut all_inputs_empty = !playlists.is_empty();
    let mut only_library_inputs = true;
    // The caller has already required Ready for every input. An empty foreign input
    // still needs its own Force contract; it cannot borrow Library's filter permission.
    let mut non_library_inputs_populated = true;

    debug!("Executing processing pipes");
    let broadcast_step = create_broadcast_callback(&ctx.events, &ctx.run_id, ctx.execution_order);

    let bouquet_file =
        tuliprox_repository::load_target_bouquet(&ctx.config, &target.name).await.map_err(|err| vec![err])?;
    let bouquet_filter =
        bouquet_file.and_then(|file| tuliprox_core::model::TargetBouquetFilter::from_dto(file.bouquet));
    if let Some(ref filter) = bouquet_filter {
        let (live, vod, series) = filter.cluster_counts();
        debug!("Loaded target bouquet for '{}': live={:?}, vod={:?}, series={:?}", target.name, live, vod, series);
    }

    let pipe = get_processing_pipe(target);
    let mut step = StepMeasure::new(&target.name, broadcast_step);
    for provider_fpl in playlists.iter_mut() {
        let input_is_empty = provider_fpl.source.is_empty();
        all_inputs_empty &= input_is_empty;
        only_library_inputs &= provider_fpl.input.input_type == InputType::Library;
        non_library_inputs_populated &= provider_fpl.input.input_type == InputType::Library || !input_is_empty;
        has_empty_library |= provider_fpl.input.input_type == InputType::Library && input_is_empty;
        log_memory_snapshot(
            format!("target '{}' input '{}' before_pipe", target.name, provider_fpl.input.name).as_str(),
        );
        step.broadcast("Executing transformations on '{}' playlist", &target.name);
        let (mut processed_fpl, input_outcome) =
            execute_pipe(target, &pipe, provider_fpl, &mut duplicates, consume_input_source, bouquet_filter.as_ref())
                .map_err(|err| vec![err])?;
        debug!("Target '{}' input '{}' pipeline outcome: {input_outcome:?}", target.name, provider_fpl.input.name);
        aggregate_outcome.merge(input_outcome);
        log_memory_snapshot(
            format!("target '{}' input '{}' after_pipe", target.name, provider_fpl.input.name).as_str(),
        );
        processed_fpl.sort_by_provider_ordinal();
        playlist_resolve(ctx, target, errors, &pipe, provider_fpl, &mut processed_fpl).await;
        log_memory_snapshot(
            format!("target '{}' input '{}' after_vod_resolve", target.name, provider_fpl.input.name).as_str(),
        );
        let clear_invalid_epg_ids = target.options.as_ref().is_some_and(ConfigTargetOptions::clear_invalid_epg_ids);
        let input_epg_start = new_epg.len();
        let epg_assign_options = EpgAssignOptions::from_target_options(target.options.as_ref());
        process_playlist_epg(&mut processed_fpl, &mut new_epg, epg_assign_options).await;
        log_memory_snapshot(
            format!("target '{}' input '{}' after_epg_apply", target.name, processed_fpl.input.name).as_str(),
        );
        let deduplicate = target.execution_plan.pre_transform_identity_dedup;
        if let Some(groups) = map_playlist_at_stage(
            &mut processed_fpl.source,
            target,
            MappingStage::AfterEpg,
            deduplicate.then_some(&mut duplicates),
        ) {
            processed_fpl.source = MemoryPlaylistSource::new(groups).into_source();
        }
        if epg_assign_options.adopt_guide_id_case && processed_fpl.epg.is_some() {
            adopt_guide_live_epg_id_case(&mut processed_fpl, &new_epg[input_epg_start..]);
        }
        if clear_invalid_epg_ids && processed_fpl.epg.is_some() {
            clear_invalid_live_epg_ids(&mut processed_fpl, &new_epg[input_epg_start..]);
        }
        if let Some(stat) = stats.get_mut(&processed_fpl.input.name) {
            stat.processed_stats.group_count = processed_fpl.get_group_count();
            stat.processed_stats.channel_count = processed_fpl.get_channel_count();
        }
        new_playlist.extend(processed_fpl.source.take_groups());
        log_memory_snapshot(
            format!("target '{}' input '{}' after_take_groups", target.name, processed_fpl.input.name).as_str(),
        );
        tokio::task::yield_now().await;
    }
    step.tick("filter rename map + epg");
    log_memory_snapshot(format!("target '{}' after_filter_rename_map_epg", target.name).as_str());
    step.stop("Preparing playlist");
    Ok(PreparedTarget {
        library_empty: match (has_empty_library, all_inputs_empty && only_library_inputs) {
            (true, true) => tuliprox_repository::LibraryEmptyPublication::CompleteTarget,
            (true, false) if non_library_inputs_populated && accepted_empty_clusters.is_empty() => {
                tuliprox_repository::LibraryEmptyPublication::FilterableContribution
            }
            (true, false) => tuliprox_repository::LibraryEmptyPublication::Contribution,
            (false, _) => tuliprox_repository::LibraryEmptyPublication::None,
        },
        target: target.clone(),
        playlist: new_playlist,
        epg: new_epg,
        processing: aggregate_outcome.to_stats(),
        accepted_empty_clusters,
    })
}

/// Spill each `Epg` source to a temp `BPlusTree` and merge them. Extracted
/// from `finalize_prepared_target` so it can be unit-tested without
/// constructing a full `PlaylistProcessingContext`.
///
/// Returns `Ok(None)` if `sources` is empty (no EPG to merge), matching
/// the contract of `flatten_tvguide`. The temp directory lives inside
/// this function call — all temp files are removed by the
/// `DiskEpgSource` drop guards before this function returns.
pub(crate) fn spill_epg_to_disk(sources: Vec<Epg>) -> Result<Option<Epg>, TuliproxError> {
    let dir =
        tempfile::tempdir().map_err(|e| TuliproxError::RepositoryXtream(format!("tempdir for EPG spill: {e}")))?;
    let mut disk_sources = Vec::with_capacity(sources.len());
    for (source_order, guide) in sources.into_iter().enumerate() {
        let mut acc = EpgMergeAccumulator::new();
        acc.set_attributes_if_preferred(guide.priority, source_order, guide.attributes);
        for channel in guide.children {
            acc.add_channel_with_programmes(
                guide.priority,
                source_order,
                guide.logo_override,
                std::sync::Arc::unwrap_or_clone(channel),
            );
        }
        let path = dir.path().join(format!("epg-src-{source_order}.db"));
        let source_order_u32 = u32::try_from(source_order).unwrap_or(0);
        let source = acc
            .finish_into_disk(path, guide.priority, source_order_u32)
            .map_err(|e| TuliproxError::RepositoryXtream(format!("EPG spill to disk failed: {e}")))?;
        disk_sources.push(source);
    }
    if disk_sources.is_empty() {
        Ok(None)
    } else {
        merge_epg_trees(disk_sources)
            .map_err(|e| TuliproxError::RepositoryXtream(format!("EPG disk merge failed: {e}")))
            .map(|opt| opt.map(|(epg, _)| epg))
    }
}

fn finalization_step_label(view: &str, stage: &str) -> String {
    if view.is_empty() {
        stage.to_string()
    } else {
        format!("{view} {stage}")
    }
}

pub(crate) fn prepare_eligible_catalog(
    target: &ConfigTarget,
    playlist: Vec<PlaylistGroup>,
    group_uuids: &HashSet<UUIDType>,
    step: &mut StepMeasure,
) -> Vec<PlaylistGroup> {
    let mut playlist = flatten_groups(playlist);
    step.tick("playlist merge");
    log_memory_snapshot(format!("target '{}' after_playlist_merge", target.name).as_str());
    if let Some(dedup_config) = target.execution_plan.post_merge_content_dedup.as_ref() {
        let removed = crate::processor::deduplicate::deduplicate_playlist(*dedup_config, &mut playlist, group_uuids);
        if removed > 0 {
            info!("Deduplicated {removed} eligible channels for target {}", target.name);
        }
        step.tick("playlist dedup");
        log_memory_snapshot(format!("target '{}' after_playlist_dedup", target.name).as_str());
    }
    playlist
}

fn finalize_playlist_view(
    target: &ConfigTarget,
    playlist: Vec<PlaylistGroup>,
    step: &mut StepMeasure,
    view: &str,
) -> Vec<PlaylistGroup> {
    let mut playlist = flatten_groups(playlist);
    step.tick(&finalization_step_label(view, "appearance merge"));

    for stage in FINALIZATION_ORDER.into_iter().skip(2) {
        match stage {
            FinalizationStage::Merge | FinalizationStage::Deduplicate => {
                unreachable!("eligibility merge and deduplication are completed before curation")
            }
            FinalizationStage::Sort => {
                if sort_playlist(target, &mut playlist) {
                    step.tick(&finalization_step_label(view, "playlist sort"));
                    log_memory_snapshot(format!("target '{}' {view} after_playlist_sort", target.name).as_str());
                }
            }
            FinalizationStage::AssignChannelNumbers => {
                assign_channel_no_playlist(&mut playlist);
                step.tick(&finalization_step_label(view, "assigning channel numbers"));
                log_memory_snapshot(format!("target '{}' {view} after_assign_channel_numbers", target.name).as_str());
            }
            FinalizationStage::AssignCounters => {
                map_playlist_counter(target, &mut playlist);
                step.tick(&finalization_step_label(view, "assigning channel counter"));
                log_memory_snapshot(format!("target '{}' {view} after_assign_channel_counter", target.name).as_str());
            }
        }
    }

    apply_persist_filter(target, &mut playlist);
    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>) {
    // Keep TMDB transport settings independent of the generic client.
    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 process_target_channel_groups(
    config: &tuliprox_core::model::AppConfig,
    target: &ConfigTarget,
    playlist: &mut Vec<PlaylistGroup>,
    step: &mut StepMeasure,
    errors: &mut Vec<TuliproxError>,
) -> Vec<PendingChannelGroup> {
    let Some(target_cfg) = target.execution_plan.channel_groups.as_ref() else {
        return Vec::new();
    };
    let cg_cfg = config.config.load().channel_groups.for_target(target_cfg);
    let overrides =
        tuliprox_repository::load_channel_group_overrides(config, &target.name).await.unwrap_or_else(|err| {
            errors.push(err);
            None
        });
    step.broadcast("Processing channel groups for '{}' playlist", &target.name);
    let groups = crate::processor::channel_group::group_playlist(&cg_cfg, overrides.as_ref(), &target.name, playlist);
    step.tick("channel grouping");
    log_memory_snapshot(format!("target '{}' after_channel_grouping", target.name).as_str());
    groups
}

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;
    let mut new_epg = prepared.epg;
    let mut errors = Vec::new();
    let broadcast_step = create_broadcast_callback(&ctx.events, &ctx.run_id, ctx.execution_order);
    let mut step = StepMeasure::new(&target.name, broadcast_step);

    let pending_channel_groups =
        process_target_channel_groups(&ctx.config, target, &mut new_playlist, &mut step, &mut errors).await;

    if target.favourites.is_some() {
        step.broadcast("Processing favourites for '{}' playlist", &target.name);
        process_favourites(&mut new_playlist, target.favourites.as_deref());
        log_memory_snapshot(format!("target '{}' after_favourites", target.name).as_str());
    }

    let library_empty = prepared.library_empty.for_filtered_playlist(&new_playlist);
    if new_playlist.is_empty() && prepared.accepted_empty_clusters.is_empty() && !library_empty.replaces_empty_target()
    {
        step.stop("");
        info!("Playlist is empty: {}", target.name);
        return (Ok(()), errors);
    }

    let group_uuids: HashSet<UUIDType> = pending_channel_groups.iter().map(|p| p.group_uuid).collect();
    let eligible_catalog = prepare_eligible_catalog(target, new_playlist, &group_uuids, &mut step);
    let mut views = prepare_target_playlist_views(&ctx.client, tmdb_client, target, eligible_catalog).await;
    views.channel_groups = pending_channel_groups;

    if views.base.is_empty()
        && views.xtream.is_none()
        && views.publication_plan == PlaylistPublicationPlan::Ordinary
        && prepared.accepted_empty_clusters.is_empty()
        && !library_empty.replaces_empty_target()
    {
        step.stop("");
        info!("Playlist is empty: {}", target.name);
        return (Ok(()), errors);
    }
    if views.publication_plan != PlaylistPublicationPlan::Ordinary {
        step.tick("target curation");
        log_memory_snapshot(format!("target '{}' after_curation", target.name).as_str());
    }

    let mut finalized_base = finalize_playlist_view(target, views.base, &mut step, "base");
    let mut finalized_xtream =
        views.xtream.map(|playlist| finalize_playlist_view(target, playlist, &mut step, "Xtream"));
    retain_epg_referenced_by_groups(&finalized_base, &mut new_epg);

    let merged_epg = if ctx.config.config.load().disk_based_processing {
        // Per-source drain to disk, then multi-way merge. Errors are pushed
        // to `errors` rather than `?` because the function returns
        // `(Result, Vec<TuliproxError>)`, not `Result` directly. We must
        // surface tempdir / write / merge failures — the user opted in to
        // disk spilling, and silently falling back to the in-memory path
        // can OOM on large feeds. When the spill itself fails we skip the
        // persist step entirely: continuing with `merged_epg = None` would
        // overwrite the existing on-disk EPG with nothing and discard the
        // previously persisted artifact on a transient error.
        match spill_epg_to_disk(new_epg) {
            Ok(epg) => epg,
            Err(err) => {
                let result_error = TuliproxError::new(err.kind(), err.message());
                errors.push(err);
                step.stop("EPG spill failed; skipping persist to preserve existing EPG");
                log_memory_snapshot(format!("target '{}' after_persist", target.name).as_str());
                return (Err(vec![result_error]), errors);
            }
        }
    } else {
        flatten_tvguide(new_epg)
    };
    let library_empty = library_empty.for_filtered_playlist(&finalized_base);
    let result = persist_playlist_views_with_groups(
        &ctx.config,
        &mut finalized_base,
        finalized_xtream.as_deref_mut(),
        merged_epg.as_ref(),
        target,
        ctx.playlist_state.as_ref(),
        TargetPlaylistPersistOptions {
            accepted_empty_clusters: prepared.accepted_empty_clusters,
            library_empty,
            publication_plan: views.publication_plan,
        },
        &mut views.channel_groups,
    )
    .await;
    let watch_playlist = target_watch_view(&finalized_base, finalized_xtream.as_deref());
    if result.is_ok() && process_watch(&ctx.config, &ctx.events, target, watch_playlist).await {
        step.tick("group watches");
        log_memory_snapshot(format!("target '{}' after_group_watches", target.name).as_str());
    }
    step.stop("Persisting playlists");
    log_memory_snapshot(format!("target '{}' after_persist", target.name).as_str());
    (result, errors)
}

pub(crate) async fn playlist_resolve<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    target: &ConfigTarget,
    errors: &mut Vec<TuliproxError>,
    pipe: &ProcessingPipe,
    provider_fpl: &mut FetchedPlaylist<'_>,
    processed_fpl: &mut FetchedPlaylist<'_>,
) {
    // Resolve and probe talk to the provider; an input in maintenance keeps what it has.
    if ctx.is_input_in_maintenance(&processed_fpl.input.name) {
        return;
    }
    playlist_resolve_series(ctx, target, errors, pipe, provider_fpl, processed_fpl).await;
    playlist_resolve_vod(ctx, target, errors, provider_fpl, processed_fpl).await;
    playlist_probe(ctx, target, processed_fpl).await;
}

pub(crate) fn is_probe_supported_item_type(item_type: PlaylistItemType) -> bool {
    matches!(
        item_type,
        PlaylistItemType::Live // we skip other live streams because hls and dash have multiple resolutions
                | PlaylistItemType::Video
                | PlaylistItemType::LocalVideo
                | PlaylistItemType::Series
                | PlaylistItemType::LocalSeries
    )
}

pub(crate) fn has_probe_details(item: &PlaylistItem) -> bool {
    match item.header.additional_properties.as_ref() {
        Some(StreamProperties::Video(v)) => v.details.as_ref().is_some_and(|d| d.video.is_some() && d.audio.is_some()),
        Some(StreamProperties::Live(l)) => l.video.is_some() && l.audio.is_some() && l.bitrate > 0,
        Some(StreamProperties::Episode(e)) => e.video.is_some() && e.audio.is_some(),
        Some(StreamProperties::Series(_)) | None => false,
    }
}

pub(crate) fn get_live_probe_interval_settings(
    target: &ConfigTarget,
    input_type: InputType,
    input_options: Option<&ConfigInputOptions>,
) -> Option<(u16, u64)> {
    if !(input_type.is_xtream() || input_type.is_m3u() || input_type.is_stalker()) {
        return None;
    }
    target.get_xtream_output().map(|_| {
        let (probe_delay, input_probe_live_interval_hours) = input_options
            .map_or((default_probe_delay_secs(), default_probe_live_interval()), |options| {
                (options.probe_delay, options.probe_live_interval_hours)
            });
        (probe_delay, u64::from(input_probe_live_interval_hours) * 3600)
    })
}

pub(crate) fn needs_live_probe(item: &PlaylistItem, cutoff_ts: i64) -> bool {
    match item.header.additional_properties.as_ref() {
        Some(StreamProperties::Live(props)) => {
            props.bitrate == 0 || props.last_probed_timestamp.is_none_or(|last_ts| last_ts < cutoff_ts)
        }
        _ => true,
    }
}

pub(crate) fn provider_id_from_item(item: &PlaylistItem) -> Option<ProviderIdType> {
    if let Ok(id) = item.header.id.parse::<u32>() {
        if id == 0 {
            return None;
        }
        return Some(ProviderIdType::Id(id));
    }

    let raw = item.header.id.trim();
    if raw.is_empty() {
        None
    } else {
        Some(ProviderIdType::from(raw))
    }
}

#[allow(clippy::too_many_lines)]
pub(crate) async fn playlist_probe<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    target: &ConfigTarget,
    fpl: &mut FetchedPlaylist<'_>,
) {
    let Some(mgr) = ctx.metadata_manager.as_ref() else {
        return;
    };
    let Some(opts) = fpl.input.options.as_ref() else {
        return;
    };
    let probe_live_enabled = opts.has_flag(ConfigInputFlags::ProbeLive);
    let probe_vod_enabled = opts.has_flag(ConfigInputFlags::ProbeVod);
    let probe_series_enabled = opts.has_flag(ConfigInputFlags::ProbeSeries);

    if !(probe_live_enabled || probe_vod_enabled || probe_series_enabled) {
        return;
    }
    if !ctx.config.is_ffprobe_enabled().await {
        return;
    }

    let input_name = fpl.input.name.clone();
    // The first `should_skip_enqueue` for an input needs its persisted enqueue
    // state on disk; inputs where no item reaches that check must not pay for
    // the load, so it happens on first use rather than here.
    let mut enqueue_state_prepared = false;
    let effective_input_type = fpl.input.get_download_input_type();
    let xtream_probe_handled = effective_input_type.is_xtream() && target.get_xtream_output().is_some();
    let live_probe_settings = if probe_live_enabled {
        get_live_probe_interval_settings(target, effective_input_type, Some(opts)).map(|(delay, interval_secs)| {
            let interval_signed = i64::try_from(interval_secs).unwrap_or(i64::MAX);
            let cutoff_ts = chrono::Utc::now().timestamp().saturating_sub(interval_signed);
            (delay, interval_secs, cutoff_ts)
        })
    } else {
        None
    };

    let mut queued_probe_keys: HashSet<(Arc<str>, String)> = HashSet::new();
    let mut queued_live_keys: HashSet<ProviderIdType> = HashSet::new();
    let mut queued_live_count = 0usize;
    let mut queued_stream_count = 0usize;

    let probe_filter = fpl.input.options.as_ref().and_then(|o| o.probe_filter.as_ref());

    for item in fpl.items() {
        if !is_probe_supported_item_type(item.header.item_type) {
            continue;
        }
        match item.header.item_type {
            PlaylistItemType::Live => {
                if !probe_live_enabled {
                    continue;
                }
            }
            PlaylistItemType::Video | PlaylistItemType::LocalVideo => {
                if !probe_vod_enabled {
                    continue;
                }
            }
            PlaylistItemType::Series | PlaylistItemType::LocalSeries => {
                if !probe_series_enabled {
                    continue;
                }
            }
            _ => continue,
        }

        // If input has a probe filter and this item doesn't match, skip probing
        if let Some(p_filter) = probe_filter {
            let provider = ValueProvider { pli: &item, match_as_ascii: false };
            if !p_filter.filter(&provider) {
                continue;
            }
        }

        match item.header.item_type {
            PlaylistItemType::Live => {
                if let Some((probe_delay, interval_secs, cutoff_ts)) = live_probe_settings {
                    if needs_live_probe(&item, cutoff_ts) {
                        if let Some(provider_id) = provider_id_from_item(&item) {
                            if queued_live_keys.insert(provider_id.clone()) {
                                let task = UpdateTask::ProbeLive {
                                    id: provider_id.clone(),
                                    reason: ResolveReason::Probe.into(),
                                    delay: probe_delay,
                                    interval: interval_secs,
                                };
                                if !enqueue_state_prepared {
                                    mgr.prepare_enqueue_state(input_name.clone()).await;
                                    enqueue_state_prepared = true;
                                }
                                if mgr.should_skip_enqueue(&input_name, &task) {
                                    continue;
                                }
                                if log_enabled!(Level::Debug) {
                                    let last_probed = match item.header.additional_properties.as_ref() {
                                        Some(StreamProperties::Live(props)) => props.last_probed_timestamp,
                                        _ => None,
                                    };
                                    debug!(
                                        "[Task] Creating ProbeLive task for input {}: id={}, last_probed_ts={:?}, cutoff_ts={}, interval={}s, title=\"{}\"",
                                        input_name,
                                        provider_id,
                                        last_probed,
                                        cutoff_ts,
                                        interval_secs,
                                        item.header.title
                                    );
                                }
                                Arc::clone(mgr).queue_task_background(input_name.clone(), task);
                                queued_live_count += 1;
                            }
                        }
                    }
                    continue;
                }
                // If live probes are enabled but no live-specific settings are available, fall through to the
                // generic probe path to keep behaviour consistent with non-xtream outputs.
            }
            PlaylistItemType::Video | PlaylistItemType::LocalVideo => {
                // Xtream outputs handle VOD probe as part of the resolve pipeline (after resolve).
                if xtream_probe_handled {
                    continue;
                }
            }
            PlaylistItemType::Series | PlaylistItemType::LocalSeries => {
                // Xtream outputs handle Series probe as part of the resolve pipeline (after resolve).
                if xtream_probe_handled {
                    continue;
                }
            }
            _ => continue,
        }

        if has_probe_details(&item) {
            continue;
        }

        // For M3U, ID is a provider id; for Library, ID is UUID.
        let unique_id = if effective_input_type == InputType::Library {
            item.header.uuid.to_valid_uuid()
        } else {
            item.header.id.to_string()
        };
        let probe_scope =
            if item.header.input_name.is_empty() { input_name.clone() } else { item.header.input_name.clone() };

        if !queued_probe_keys.insert((probe_scope.clone(), unique_id.clone())) {
            continue;
        }

        let task = UpdateTask::ProbeStream {
            probe_scope: probe_scope.clone(),
            unique_id: unique_id.clone(),
            url: item.header.url.to_string(),
            item_type: item.header.item_type,
            reason: ResolveReason::MissingDetails.into(),
            delay: opts.probe_delay,
        };
        if !enqueue_state_prepared {
            mgr.prepare_enqueue_state(input_name.clone()).await;
            enqueue_state_prepared = true;
        }
        if mgr.should_skip_enqueue(&input_name, &task) {
            continue;
        }
        debug!(
            "[Task] Creating ProbeStream task for input {}: scope={}, unique_id={}, item_type={:?}, title=\"{}\"",
            input_name, probe_scope, unique_id, item.header.item_type, item.header.title
        );
        Arc::clone(mgr).queue_task_background(input_name.clone(), task);
        queued_stream_count += 1;
    }

    if queued_live_count > 0 || queued_stream_count > 0 {
        info!(
            "Queued probe tasks for input {input_name} (live_interval={queued_live_count}, generic={queued_stream_count})"
        );
    }
}

pub fn process_favourites(playlist: &mut Vec<PlaylistGroup>, favourites_cfg: Option<&[ConfigFavourites]>) {
    if let Some(favourites) = favourites_cfg {
        let mut fav_groups: IndexMap<CategoryKey, Vec<PlaylistItem>> = IndexMap::new();
        for pg in playlist.iter() {
            for pli in &pg.channels {
                // series episodes can't be included in favourites
                if pli.header.item_type == PlaylistItemType::Series
                    || pli.header.item_type == PlaylistItemType::LocalSeries
                {
                    continue;
                }
                for fav in favourites {
                    if pli.header.xtream_cluster == fav.cluster && is_valid(pli, &fav.filter, fav.match_as_ascii) {
                        let mut channel = pli.clone();
                        channel.header.group.clone_from(&fav.group);
                        // Update UUID to be an alias of the original
                        channel.header.uuid = create_alias_uuid(&pli.header.uuid, &fav.group);
                        fav_groups.entry((fav.cluster, fav.group.clone())).or_default().push(channel);
                    }
                }
            }
        }

        for (fav_group, channels) in fav_groups {
            if !channels.is_empty() {
                let (xtream_cluster, group_name) = fav_group;
                playlist.push(PlaylistGroup { id: 0, title: group_name, channels, xtream_cluster });
            }
        }
    }
}

#[derive(Debug)]
pub(crate) struct TargetPlaylistViews {
    pub(crate) base: Vec<PlaylistGroup>,
    pub(crate) xtream: Option<Vec<PlaylistGroup>>,
    pub(crate) publication_plan: PlaylistPublicationPlan,
    pub(crate) channel_groups: Vec<PendingChannelGroup>,
}

pub(crate) fn build_curated_playlist_views(
    playlist: Vec<PlaylistGroup>,
    evaluation: &CurationEvaluation,
    config: &CurationConfig,
    appearance_filter_configured: bool,
    has_xtream_output: bool,
) -> TargetPlaylistViews {
    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 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,
        publication_plan: PlaylistPublicationPlan::complete_curation_with_filter(
            curated_catalog,
            has_xtream_output && !config.include_xtream_base_categories,
            appearance_filter_configured,
        ),
        channel_groups: Vec::new(),
    }
}

pub(crate) async fn prepare_target_playlist_views(
    client: &reqwest::Client,
    tmdb_client: Option<&reqwest::Client>,
    target: &ConfigTarget,
    playlist: Vec<PlaylistGroup>,
) -> TargetPlaylistViews {
    let Some(config) = target.effective_curation() else {
        return TargetPlaylistViews {
            base: playlist,
            xtream: None,
            publication_plan: PlaylistPublicationPlan::Ordinary,
            channel_groups: Vec::new(),
        };
    };

    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,
) -> TargetPlaylistViews {
    match outcome {
        CurationRunOutcome::NotConfigured => TargetPlaylistViews {
            base: playlist,
            xtream: None,
            publication_plan: PlaylistPublicationPlan::Ordinary,
            channel_groups: Vec::new(),
        },
        CurationRunOutcome::Failed(failure) => {
            warn_curation_failure(&target.name, &failure);
            // Discovery is optional; unavailable selections must not suppress the source catalog or EPG.
            TargetPlaylistViews {
                base: playlist,
                xtream: None,
                publication_plan: PlaylistPublicationPlan::Ordinary,
                channel_groups: Vec::new(),
            }
        }
        CurationRunOutcome::Complete(evaluation) => {
            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,
                evaluation.memberships.len(),
                views.xtream.as_deref().map_or(0, <[PlaylistGroup]>::len)
            );
            views
        }
    }
}
fn warn_curation_failure(target_name: &str, failure: &CurationFailure) {
    let mut complete = 0usize;
    let mut incomplete = 0usize;
    let mut unavailable = 0usize;
    for outcome in &failure.selector_outcomes {
        match outcome {
            SelectorOutcome::Complete { .. } => complete += 1,
            SelectorOutcome::Incomplete { .. } => incomplete += 1,
            SelectorOutcome::Unavailable { .. } => unavailable += 1,
        }
    }
    warn!(
        "Target '{target_name}' curation refresh failed (complete selectors: {complete}, incomplete: {incomplete}, unavailable: {unavailable}); continuing with the regular catalog"
    );
}

fn item_subject_uuid(item: &PlaylistItem) -> UUIDType {
    if item.header.uuid == UUIDType::default() {
        item.get_uuid()
    } else {
        item.header.uuid
    }
}

fn selected_series_parent_keys(playlist: &[PlaylistGroup], selected_subjects: &HashSet<UUIDType>) -> HashSet<Arc<str>> {
    let mut parent_keys = HashSet::new();
    for item in playlist.iter().flat_map(|group| &group.channels) {
        if !matches!(item.header.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo)
            || !selected_subjects.contains(&item_subject_uuid(item))
        {
            continue;
        }
        parent_keys.insert(item.get_uuid().intern());
        if item.header.uuid != UUIDType::default() {
            parent_keys.insert(item.header.uuid.intern());
        }
        if item.header.virtual_id.get() != 0 {
            parent_keys.insert(item.header.virtual_id.get().to_string().intern());
        }
        if item.header.item_type == PlaylistItemType::LocalSeriesInfo && !item.header.id.is_empty() {
            parent_keys.insert(Arc::clone(&item.header.id));
        }
    }
    parent_keys
}

pub(crate) fn select_target_catalog(
    mut playlist: Vec<PlaylistGroup>,
    evaluation: &CurationEvaluation,
    enabled: bool,
) -> Vec<PlaylistGroup> {
    if !enabled {
        return playlist;
    }

    let selected_subjects =
        evaluation.memberships.iter().map(|membership| membership.subject_uuid).collect::<HashSet<_>>();
    let selected_parent_keys = selected_series_parent_keys(&playlist, &selected_subjects);

    for group in &mut playlist {
        group.channels.retain(|item| {
            if item.header.xtream_cluster == XtreamCluster::Live {
                return true;
            }
            match item.header.item_type {
                PlaylistItemType::Video
                | PlaylistItemType::LocalVideo
                | PlaylistItemType::SeriesInfo
                | PlaylistItemType::LocalSeriesInfo => selected_subjects.contains(&item_subject_uuid(item)),
                PlaylistItemType::Series | PlaylistItemType::LocalSeries => {
                    selected_parent_keys.contains(&item.header.parent_code)
                }
                _ => false,
            }
        });
    }
    playlist.retain(|group| group.xtream_cluster == XtreamCluster::Live || !group.channels.is_empty());
    playlist
}

fn live_only(playlist: &[PlaylistGroup]) -> Vec<PlaylistGroup> {
    playlist
        .iter()
        .filter_map(|group| {
            let mut live_group = group.clone();
            live_group.channels.retain(|item| item.header.xtream_cluster == XtreamCluster::Live);
            (group.xtream_cluster == XtreamCluster::Live || !live_group.channels.is_empty()).then_some(live_group)
        })
        .collect()
}

pub(crate) fn target_watch_view<'playlist>(
    base: &'playlist [PlaylistGroup],
    xtream: Option<&'playlist [PlaylistGroup]>,
) -> &'playlist [PlaylistGroup] {
    xtream.unwrap_or(base)
}

pub(crate) async fn process_watch<E: EventSink>(
    app_config: &Arc<AppConfig>,
    events: &E,
    target: &ConfigTarget,
    new_playlist: &[PlaylistGroup],
) -> bool {
    let Some(watches) = &target.watch else {
        return false;
    };

    // Configured, but every pattern failed to compile. Silently doing
    // nothing here is what made a typo in `watch` indistinguishable from a
    // playlist that never changes.
    if watches.is_empty() {
        error!("target '{}' configured watch patterns but none of them compiled", target.name);
        events.emit(EventMessage::PlaylistWatchDisabled(WatchDisabled::new(
            target.name.clone(),
            WatchDisabledReason::InvalidPatterns,
        )));
        return false;
    }

    if default_as_default().eq_ignore_ascii_case(&target.name) {
        error!("can't watch a target with no unique name");
        events.emit(EventMessage::PlaylistWatchDisabled(WatchDisabled::new(
            target.name.clone(),
            WatchDisabledReason::UnnamedTarget,
        )));
        return false;
    }

    // One snapshot for the whole refresh rather than a config load per group.
    let config = app_config.config.load_full();
    let storage_dir = config.storage_dir.as_str();

    // Before the per-group fan-out: this is about which groups exist, not
    // what is inside the ones the patterns name, so it must see every group
    // rather than only the watched ones.
    let group_changes = diff_target_groups(storage_dir, events, &target.name, new_playlist, watches).await;

    // Only `channels` entries are matched against the current groups: they
    // pick the groups to diff, and they are the only ones the unmatched
    // warning reports on (`ConfigWatch::reports_unmatched`). Once a group is
    // watched, an entry that already matched elsewhere needs no regex run.
    let mut matched = vec![false; watches.len()];
    let mut group_watches = Vec::new();
    for group in new_playlist {
        let mut watch_channels = false;
        for (index, watch) in watches.iter().enumerate() {
            if !watch.contexts.contains(WatchContext::Channels) || (watch_channels && matched[index]) {
                continue;
            }
            if watch.pattern.is_match(&group.title) {
                matched[index] = true;
                watch_channels = true;
            }
        }
        if watch_channels {
            // Futures built up front: a lazy `map` closure over borrowed state
            // trips the higher-ranked `Send` check of the spawned pipeline task.
            group_watches.push(process_group_watch(storage_dir, events, &target.name, group));
        }
    }

    // A channel pattern that matches nothing looks exactly like a group that
    // has not changed, so a typo must surface. Entries waiting for a group to
    // appear, and removal-only entries, match nothing by design.
    // `EventKindMask::from_wire_names` reports unmatched names for the same reason.
    let unmatched: Vec<String> = watches
        .iter()
        .zip(&matched)
        .filter(|(watch, matched)| !**matched && watch.reports_unmatched())
        .map(|(watch, _)| watch.pattern.as_str().to_string())
        .collect();
    if !unmatched.is_empty() {
        warn!("target '{}' has {} watch pattern(s) matching no group", target.name, unmatched.len());
        events.emit(EventMessage::PlaylistWatchUnmatched(WatchUnmatched::new(
            target.name.clone(),
            unmatched,
            new_playlist.len(),
        )));
    }

    let channel_diffs: Vec<_> =
        futures::stream::iter(group_watches).buffer_unordered(16).filter_map(futures::future::ready).collect().await;

    if let Some(changes) = assemble_watch_changes(&target.name, group_changes, channel_diffs) {
        events.emit(EventMessage::PlaylistWatchChanged(changes));
    }

    true
}

#[cfg(test)]
pub(crate) async fn run_finalize_until_persist(
    target: &ConfigTarget,
    mut playlist: Vec<PlaylistGroup>,
) -> (Vec<PlaylistGroup>, Vec<PendingChannelGroup>) {
    let mut step = StepMeasure::new("test", |_, _| {});
    let pending = if let Some(target_cfg) = target.execution_plan.channel_groups.as_ref() {
        let cg_cfg = shared::model::ChannelGroupsConfig::default().for_target(target_cfg);
        crate::processor::channel_group::group_playlist(&cg_cfg, None, &target.name, &mut playlist)
    } else {
        Vec::new()
    };
    if target.favourites.is_some() {
        process_favourites(&mut playlist, target.favourites.as_deref());
    }
    let group_uuids: HashSet<UUIDType> = pending.iter().map(|p| p.group_uuid).collect();
    let eligible = prepare_eligible_catalog(target, playlist, &group_uuids, &mut step);
    let finalized = finalize_playlist_view(target, eligible, &mut step, "base");
    (finalized, pending)
}