tuliprox/backend/processing/src/processor/xtream_series.rs

use crate::{
    fetched_playlist::FetchedPlaylist,
    metadata_sink::MetadataUpdateSink,
    processor::{
        create_resolve_options_function_for_xtream_target,
        playlist::{execute_pipeline_on_groups, PlaylistProcessingContext, ProcessingPipe},
        process_foreground_retry_once, select_cancel_token, ProbeHandleGuard, ResolveOptions, ResolveOptionsFlags,
        FOREGROUND_BATCH_SIZE as BATCH_SIZE, FOREGROUND_MIN_RETRY_DELAY_SECS,
        FOREGROUND_RETRY_BATCH_MAX_SIZE as RETRY_BATCH_MAX_SIZE,
    },
};
use log::{debug, error, info, log_enabled, trace, warn, Level};
use parking_lot::Mutex;
use serde_json::Value;
use shared::{
    defaults::default_probe_user_priority,
    error::TuliproxError,
    foundation::ValueProvider,
    model::{
        EventSink, MediaQuality, PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemType, SeriesStreamProperties,
        StreamProperties, XtreamCluster, XtreamPlaylistItem, XtreamSeriesInfo,
    },
};
use std::{
    collections::HashSet,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};
use tuliprox_core::{
    model::{
        AppConfig, ConfigInput, ConfigInputFlags, ConfigTarget, InputSource, MetadataUpdateConfig, ProviderHandle,
        ProviderIdType, ResolveReason, ResolveReasonSet, UpdateTask,
    },
    utils::{
        debug_if_enabled,
        ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome},
    },
};
use tuliprox_iptv::xtream;
use tuliprox_library::{
    library::{MetadataResolver, MetadataStorage},
    media_enrichment::{
        policy::MissingFactEnrichmentPolicy,
        xtream::{
            apply_fact_patch_to_series, series_fact_patch_from_metadata, series_fact_patch_from_title_candidates,
        },
    },
};
use tuliprox_parser::xtream::{create_xtream_series_episode_url, parse_xtream_series_info};
use tuliprox_repository::{
    get_input_storage_path, persist_input_series_info_batch, persists_input_series_info, xtream_get_file_path,
    BPlusTreeQuery,
};
use tuliprox_session::ActiveProviderManager;

create_resolve_options_function_for_xtream_target!(series);

#[derive(Debug, Clone, Copy)]
pub struct SeriesProbeSettings {
    pub timeout_secs: u64,
    pub analyze_duration_micros: u64,
    pub probe_size_bytes: u64,
}

impl SeriesProbeSettings {
    pub fn from_metadata_update(metadata_update: Option<&MetadataUpdateConfig>) -> Self {
        let defaults = MetadataUpdateConfig::default();
        Self {
            timeout_secs: metadata_update.and_then(|cfg| cfg.ffprobe.timeout).unwrap_or(60),
            analyze_duration_micros: metadata_update
                .map_or(defaults.ffprobe.analyze_duration_micros, |cfg| cfg.ffprobe.analyze_duration_micros),
            probe_size_bytes: metadata_update
                .map_or(defaults.ffprobe.probe_size_bytes.get(), |cfg| cfg.ffprobe.probe_size_bytes.get()),
        }
    }
}

#[allow(clippy::too_many_arguments)]
pub async fn playlist_resolve_series<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<'_>,
) {
    // Skip-flag is the kill-switch: no resolve, no probe, no iteration.
    if processed_fpl.input.has_flag(ConfigInputFlags::SkipSeries) {
        return;
    }

    let mut resolve_options = get_resolve_series_options(target, processed_fpl);

    if !ctx.config.is_ffprobe_enabled().await {
        resolve_options.unset_flag(ResolveOptionsFlags::Probe);
    }
    provider_fpl.source.release_resources(XtreamCluster::Series);

    playlist_resolve_series_info(ctx, errors, processed_fpl, resolve_options, pipe, target).await;

    if provider_fpl.is_memory() {
        sync_resolved_series_properties(provider_fpl, processed_fpl);
    }

    if let Err(err) = provider_fpl.source.obtain_resources().await {
        errors.push(err);
    }
}

#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
async fn playlist_resolve_series_info<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    _errors: &mut Vec<TuliproxError>,
    fpl: &mut FetchedPlaylist<'_>,
    resolve_options: ResolveOptions,
    pipe: &ProcessingPipe,
    target: &ConfigTarget,
) {
    let filter = |pli: &PlaylistItem| {
        pli.header.xtream_cluster == XtreamCluster::Series && pli.header.item_type == PlaylistItemType::SeriesInfo
    };

    // Skip if nothing to do
    let skip_resolve = !resolve_options.has_flag(ResolveOptionsFlags::Resolve)
        && !resolve_options.has_flag(ResolveOptionsFlags::Probe)
        && !resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing);

    let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb);

    let groups_to_add = if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() {
        queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled)
    } else {
        process_immediate_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await
    };

    // Apply pipe transformations to new groups
    let (new_playlist, pipeline_outcome) = execute_pipeline_on_groups(groups_to_add, target, pipe);
    debug!("Resolved series pipeline outcome: {pipeline_outcome:?}");

    // Apply resolved episodes to playlist
    fpl.extend_playlist(new_playlist);
}

fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, processed_fpl: &mut FetchedPlaylist<'_>) {
    crate::processor::xtream::sync_resolved_xtream_properties(
        provider_fpl,
        processed_fpl,
        XtreamCluster::Series,
        PlaylistItemType::SeriesInfo,
        |props| match props {
            StreamProperties::Series(s) => Some(s.as_ref()),
            _ => None,
        },
        StreamProperties::Series,
    );
}

fn queue_background_series_info<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    fpl: &mut FetchedPlaylist<'_>,
    filter: impl Fn(&PlaylistItem) -> bool,
    resolve_options: &ResolveOptions,
    skip_resolve: bool,
    resolve_tmdb_enabled: bool,
) -> Vec<PlaylistGroup> {
    let mut groups_to_add = Vec::new();
    let Some(mgr) = ctx.metadata_manager.as_ref() else {
        return groups_to_add;
    };

    let input = fpl.input;
    let input_name_arc = input.name.clone();
    // Extract filter before iterating to avoid borrow conflict
    let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());

    let stage_started = Instant::now();
    let mut total = 0usize;
    let mut skipped_filter = 0usize;
    let mut skipped_resolve_filter = 0usize;
    let mut queued = 0usize;
    let mut expanded = 0usize;
    for pli in fpl.items_mut() {
        total += 1;
        if !filter(pli) {
            skipped_filter += 1;
            continue;
        }

        // If input has a resolve filter and this item doesn't match, skip processing
        if let Some(r_filter) = resolve_filter {
            let provider = ValueProvider { pli, match_as_ascii: false };
            if !r_filter.filter(&provider) {
                skipped_resolve_filter += 1;
                continue;
            }
        }

        let provider_id = if let Ok(uid) = pli.header.id.parse::<u32>() {
            ProviderIdType::Id(uid)
        } else {
            ProviderIdType::from(pli.header.id.to_string())
        };

        if !skip_resolve {
            let reasons = check_resolve_reasons(resolve_options, resolve_tmdb_enabled, pli);

            if !reasons.is_empty() {
                let task = UpdateTask::ResolveSeries {
                    id: provider_id.clone(),
                    reason: reasons,
                    delay: resolve_options.resolve_delay,
                    source_last_modified: pli
                        .header
                        .additional_properties
                        .as_ref()
                        .and_then(StreamProperties::get_last_modified),
                };
                if log_enabled!(Level::Debug) {
                    let has_details = pli.has_details();
                    let (has_tmdb, has_date) = match pli.header.additional_properties.as_ref() {
                        Some(StreamProperties::Series(props)) => (props.tmdb.is_some(), props.release_date.is_some()),
                        _ => (false, false),
                    };
                    debug!(
                        "[Task] Creating ResolveSeries task for input {}: id={}, reasons={}, has_details={}, has_tmdb={}, has_date={}, title=\"{}\"",
                        input_name_arc, provider_id, reasons, has_details, has_tmdb, has_date, pli.header.title
                    );
                }
                Arc::clone(mgr).queue_task_background(input_name_arc.clone(), task);
                queued += 1;
            }
        }

        if let Some(group_obj) = expand_series_item(pli, input) {
            groups_to_add.push(group_obj);
            expanded += 1;
        }
    }

    if log_enabled!(Level::Debug) {
        debug!(
            "[Task] Series resolve queueing summary for input {input_name_arc}: total={total}, queued={queued}, expanded={expanded}, skipped_filter={skipped_filter}, skipped_resolve_filter={skipped_resolve_filter}, elapsed_ms={}",
            stage_started.elapsed().as_millis()
        );
    }
    groups_to_add
}

#[allow(clippy::too_many_lines)]
async fn process_immediate_series_info<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    fpl: &mut FetchedPlaylist<'_>,
    filter: impl Fn(&PlaylistItem) -> bool,
    resolve_options: &ResolveOptions,
    skip_resolve: bool,
    resolve_tmdb_enabled: bool,
) -> Vec<PlaylistGroup> {
    let input = fpl.input;
    let storage_dir = &ctx.config.config.load().storage_dir;
    let storage_path = match get_input_storage_path(&input.name, storage_dir).await {
        Ok(path) => path,
        Err(err) => {
            error!("Can't resolve series, input storage directory for input '{}' failed: {err}", input.name);
            return Vec::new();
        }
    };

    // Keep an optional read query open and reopen lazily only when needed.
    let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Series);
    let mut db_query_holder: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>> = None;
    let mut _db_lock_holder = None;

    let mut groups_to_add = Vec::new();
    let mut batch: Vec<(ProviderIdType, SeriesStreamProperties)> = Vec::with_capacity(BATCH_SIZE);
    let mut retry_once_ids: HashSet<ProviderIdType> = HashSet::new();
    let mut processed_count = 0;
    let mut last_log_time = Instant::now();
    let series_probe_settings = {
        let config = ctx.config.config.load();
        SeriesProbeSettings::from_metadata_update(config.metadata_update.as_ref())
    };

    // Extract filter before iterating to avoid borrow conflict
    let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());

    for pli in fpl.items_mut() {
        if !filter(pli) {
            continue;
        }

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

        let provider_id = if let Ok(uid) = pli.header.id.parse::<u32>() {
            ProviderIdType::Id(uid)
        } else {
            ProviderIdType::from(&*pli.header.id)
        };
        let mut defer_expand = false;

        if !skip_resolve {
            let reasons = check_resolve_reasons(resolve_options, resolve_tmdb_enabled, pli);

            if !reasons.is_empty() {
                if let Some(active_provider) = ctx.provider_manager.as_ref() {
                    if db_query_holder.is_none() && xtream_path.exists() {
                        let file_lock = ctx.config.file_locks.read_lock(&xtream_path).await;
                        let xtream_path = xtream_path.clone();
                        let query = match tokio::task::spawn_blocking(move || {
                            BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)
                        })
                        .await
                        {
                            Ok(Ok(query)) => Some((query, file_lock)),
                            Ok(Err(err)) => {
                                error!("Failed to open BPlusTreeQuery for Series: {err}");
                                None
                            }
                            Err(err) => {
                                error!("Failed to open BPlusTreeQuery for Series: {err}");
                                None
                            }
                        };

                        if let Some((query, guard)) = query {
                            db_query_holder = Some(Arc::new(Mutex::new(query)));
                            _db_lock_holder = Some(guard);
                        }
                    }
                    // Pass the optional reference to the query
                    let db_query_ref = db_query_holder.as_ref().map(Arc::clone);

                    match update_series_info_immediate(
                        ctx,
                        active_provider,
                        input,
                        pli,
                        provider_id.clone(),
                        &reasons,
                        db_query_ref,
                        series_probe_settings,
                    )
                    .await
                    {
                        Ok(Some(updated_props)) => {
                            pli.header.additional_properties =
                                Some(StreamProperties::Series(Box::new(updated_props.clone())));
                            batch.push((provider_id.clone(), updated_props));

                            if batch.len() >= BATCH_SIZE {
                                // Release lock before persisting to avoid deadlock (persist needs write lock)
                                db_query_holder = None;
                                _db_lock_holder = None;

                                let updates: Vec<(u32, SeriesStreamProperties)> = batch
                                    .iter()
                                    .filter_map(|(id, props)| {
                                        if let ProviderIdType::Id(vid) = id {
                                            Some((*vid, props.clone()))
                                        } else {
                                            None
                                        }
                                    })
                                    .collect();

                                if updates.is_empty() {
                                    batch.clear();
                                } else {
                                    match persist_input_series_info_batch(
                                        &ctx.config,
                                        &storage_path,
                                        XtreamCluster::Series,
                                        &input.name,
                                        updates,
                                    )
                                    .await
                                    {
                                        Ok(()) => batch.clear(),
                                        Err(err) => {
                                            error!(
                                                "persist_input_series_info_batch failed for XtreamCluster::Series on input '{}'. batch.clear() skipped. Error: {}",
                                                input.name, err
                                            );
                                        }
                                    }
                                }
                            }

                            processed_count += 1;
                            if resolve_options.resolve_delay > 0 {
                                tokio::time::sleep(Duration::from_secs(u64::from(resolve_options.resolve_delay))).await;
                            }
                        }
                        Ok(None) => {}
                        Err(e) => {
                            error!("Failed to update Series metadata for {}: {e}", pli.header.title);
                            retry_once_ids.insert(provider_id.clone());
                            defer_expand = true;
                        }
                    }
                }
            }
        }

        if !defer_expand {
            if let Some(group_obj) = expand_series_item(pli, input) {
                groups_to_add.push(group_obj);
            }
        }

        if log_enabled!(Level::Info) && last_log_time.elapsed().as_secs() >= 30 {
            info!("resolved {processed_count} series info");
            last_log_time = Instant::now();
        }
    }

    if !retry_once_ids.is_empty() {
        let retry_delay_secs = u64::from(resolve_options.resolve_delay).max(FOREGROUND_MIN_RETRY_DELAY_SECS);
        process_foreground_retry_once!(
            ctx: ctx,
            fpl: fpl,
            filter: filter,
            retry_once_ids: retry_once_ids,
            retry_delay_secs: retry_delay_secs,
            xtream_path: xtream_path,
            db_query_holder: db_query_holder,
            db_lock_holder: _db_lock_holder,
            batch: batch,
            batch_size: BATCH_SIZE,
            retry_batch_max_len: RETRY_BATCH_MAX_SIZE,
            processed_count: processed_count,
            query_error_context: "Series retry",
            reasons: |retry_pli| {
                if skip_resolve {
                    ResolveReasonSet::new()
                } else {
                    check_resolve_reasons(resolve_options, resolve_tmdb_enabled, retry_pli)
                }
            },
            update: |active_provider, retry_pli, provider_id, reasons, db_query_ref| update_series_info_immediate(
                ctx,
                active_provider,
                input,
                retry_pli,
                provider_id,
                reasons,
                db_query_ref,
                series_probe_settings,
            ),
            apply_properties: |retry_pli, updated_props| {
                retry_pli.header.additional_properties =
                    Some(StreamProperties::Series(Box::new(updated_props.clone())));
            },
            persist: |updates| persist_input_series_info_batch(
                &ctx.config,
                &storage_path,
                XtreamCluster::Series,
                &input.name,
                updates,
            ),
            on_persist_error: |err| {
                error!("persist_input_series_info_batch failed for Series retry on input '{}'. Error: {}", input.name, err);
            },
            on_retry_error: |retry_pli, err| {
                error!("Foreground retry failed for Series {}: {err}", retry_pli.header.title);
            },
            on_after_attempt: |retry_pli, _retry_succeeded| {
                if let Some(group_obj) = expand_series_item(retry_pli, input) {
                    groups_to_add.push(group_obj);
                }
            },
        );
    }

    if !batch.is_empty() {
        // Release lock before final persist
        _db_lock_holder = None;

        let updates: Vec<(u32, SeriesStreamProperties)> = batch
            .into_iter()
            .filter_map(|(id, props)| if let ProviderIdType::Id(vid) = id { Some((vid, props)) } else { None })
            .collect();

        if !updates.is_empty() {
            if let Err(err) =
                persist_input_series_info_batch(&ctx.config, &storage_path, XtreamCluster::Series, &input.name, updates)
                    .await
            {
                error!("Failed to persist final batch series info: {err}");
            }
        }
    }

    if processed_count > 0 {
        info!("Processed {processed_count} series info");
    }

    groups_to_add
}

fn expand_series_item(pli: &PlaylistItem, input: &ConfigInput) -> Option<PlaylistGroup> {
    if let Some(StreamProperties::Series(properties)) = pli.header.additional_properties.as_ref() {
        let global_release_date = properties.release_date.clone();
        let (group, series_name) = {
            let header = &pli.header;
            (header.group.clone(), header.get_name())
        };

        if let Some(episodes) = parse_xtream_series_info(
            &pli.get_uuid(),
            properties,
            &group,
            &series_name,
            input,
            global_release_date.as_ref(),
            pli.header.source_ordinal,
        ) {
            return Some(PlaylistGroup {
                id: pli.header.category_id,
                title: pli.header.group.clone(),
                channels: episodes,
                xtream_cluster: XtreamCluster::Series,
            });
        }
    }
    None
}

#[allow(clippy::too_many_arguments)]
async fn update_series_info_immediate<E: EventSink + Clone + 'static, M: MetadataUpdateSink>(
    ctx: &PlaylistProcessingContext<E, M>,
    active_provider: &Arc<ActiveProviderManager>,
    input: &ConfigInput,
    pli: &PlaylistItem,
    id: ProviderIdType,
    reasons: &ResolveReasonSet,
    db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
    probe_settings: SeriesProbeSettings,
) -> Result<Option<SeriesStreamProperties>, TuliproxError> {
    let fetch_info = reasons.contains(ResolveReason::Info);
    let resolve_missing_facts = reasons.contains(ResolveReason::Tmdb) || reasons.contains(ResolveReason::Date);

    update_series_metadata(
        &ctx.config,
        &ctx.client,
        input,
        id,
        active_provider,
        None, // active_handle
        Some(&pli.header.title),
        false, // save (we batch in caller)
        fetch_info,
        resolve_missing_facts,
        reasons.contains(ResolveReason::Probe),
        probe_settings,
        db_query,
        None,
        None,
    )
    .await
}

fn check_resolve_reasons(
    resolve_options: &ResolveOptions,
    resolve_tmdb_enabled: bool,
    pli: &PlaylistItem,
) -> ResolveReasonSet {
    let mut reasons = ResolveReasonSet::new();

    check_needs_info(resolve_options, pli, &mut reasons);
    check_resolve_tmdb(resolve_options, resolve_tmdb_enabled, pli, &mut reasons);

    if resolve_options.has_flag(ResolveOptionsFlags::Probe) {
        check_needs_probe(pli, &mut reasons);
    }
    reasons
}

fn check_needs_info(resolve_options: &ResolveOptions, pli: &PlaylistItem, reasons: &mut ResolveReasonSet) {
    let needs_info = resolve_options.has_flag(ResolveOptionsFlags::Resolve) && !pli.has_details();
    if needs_info {
        reasons.set(ResolveReason::Info);
    }
}

fn check_resolve_tmdb(
    resolve_options: &ResolveOptions,
    resolve_tmdb_enabled: bool,
    pli: &PlaylistItem,
    reasons: &mut ResolveReasonSet,
) {
    if resolve_tmdb_enabled && resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing) {
        let (has_tmdb, has_date, title_present) = match pli.header.additional_properties.as_ref() {
            Some(StreamProperties::Series(series_stream_props)) => (
                series_stream_props.tmdb.is_some(),
                series_stream_props.release_date.is_some(),
                !series_stream_props.name.is_empty() || !pli.header.title.is_empty(),
            ),
            None => (false, false, !pli.header.title.is_empty()),
            _ => return,
        };

        if title_present && (!has_tmdb || !has_date) {
            if !has_tmdb {
                reasons.set(ResolveReason::Tmdb);
            }
            if !has_date {
                reasons.set(ResolveReason::Date);
            }
        }
    }
}

fn check_needs_probe(pli: &PlaylistItem, reasons: &mut ResolveReasonSet) {
    let needs_probe = match pli.header.additional_properties.as_ref() {
        Some(StreamProperties::Series(series_stream_props)) => {
            series_stream_props.details.as_ref().and_then(|d| d.episodes.as_ref()).is_some_and(|episodes| {
                episodes.iter().any(|ep| {
                    let missing_video = !MediaQuality::is_valid_media_info(ep.video.as_deref());
                    let missing_audio = !MediaQuality::is_valid_media_info(ep.audio.as_deref());
                    missing_video || missing_audio
                })
            })
        }
        _ => false,
    };

    if needs_probe {
        reasons.set(ResolveReason::Probe);
    }
}
/// Updates metadata for a single Series (Info + Episodes Probe) and persists it.
///
/// # Arguments
/// * `save` - If true, persists changes to the input database immediately (Instant strategy).
///   If false, returns the properties so the caller can batch persist them (Bundled strategy).
/// * `fetch_info` - If true, fetches details from Provider API. If false, uses existing/dummy data.
/// * `resolve_missing_facts` - If true, resolves missing TMDB/date facts from available titles.
/// * `db_query` - Optional pre-opened DB handle to avoid re-opening file.
#[allow(clippy::too_many_arguments, clippy::too_many_lines, clippy::fn_params_excessive_bools)]
pub async fn update_series_metadata(
    app_config: &Arc<AppConfig>,
    client: &reqwest::Client,
    input: &ConfigInput,
    id: ProviderIdType,
    active_provider: &Arc<ActiveProviderManager>,
    active_handle: Option<&ProviderHandle>,
    playlist_title: Option<&str>,
    save: bool,
    fetch_info: bool,
    resolve_missing_facts: bool,
    do_probe: bool,
    probe_settings: SeriesProbeSettings,
    db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
    tmdb_and_date_present_out: Option<&AtomicBool>,
    probe_pending_out: Option<&AtomicBool>,
) -> Result<Option<SeriesStreamProperties>, TuliproxError> {
    let storage_dir = &app_config.config.load().storage_dir;
    let storage_path = get_input_storage_path(&input.name, storage_dir)
        .await
        .map_err(|e| TuliproxError::Io(format!("Storage path error: {e}")))?;

    if input.has_flag(ConfigInputFlags::SkipSeries) {
        return Ok(None);
    }

    // Try to load existing info first
    let mut props: Option<SeriesStreamProperties> = None;
    let mut existing_item: Option<XtreamPlaylistItem> = None;

    let series_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None };

    if let Some(series_id) = series_id_opt {
        if let Some(query) = db_query {
            let query = Arc::clone(&query);
            let item = match tokio::task::spawn_blocking(move || {
                let mut guard = query.lock();
                guard.query_zero_copy(&series_id)
            })
            .await
            {
                Ok(Ok(item)) => item,
                Ok(Err(err)) => {
                    error!("Failed to query Series metadata from disk for {series_id}: {err}");
                    None
                }
                Err(err) => {
                    error!("Failed to query Series metadata from disk for {series_id}: {err}");
                    None
                }
            };

            if let Some(item) = item {
                existing_item = Some(item.clone());
                if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() {
                    props = Some(*p.clone());
                }
            }
        } else {
            let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Series);
            if xtream_path.exists() {
                let _file_lock = app_config.file_locks.read_lock(&xtream_path).await;
                let xtream_path = xtream_path.clone();
                let item = match tokio::task::spawn_blocking(move || {
                    let mut query = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)?;
                    query.query_zero_copy(&series_id)
                })
                .await
                {
                    Ok(Ok(item)) => item,
                    Ok(Err(err)) => {
                        error!("Failed to query Series metadata from disk for {series_id}: {err}");
                        None
                    }
                    Err(err) => {
                        error!("Failed to query Series metadata from disk for {series_id}: {err}");
                        None
                    }
                };

                if let Some(item) = item {
                    existing_item = Some(item.clone());
                    if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() {
                        props = Some(*p.clone());
                    }
                }
            }
        }
    }

    let mut fetched_new = false;
    let mut properties_updated = false;
    let mut probe_failure: Option<ProbeFailureKind> = None;
    let mut probe_pending = false;

    let display_id = series_id_opt.map_or_else(|| "StringID".to_string(), |v| v.to_string());

    // Input DB is canonical, but if details are missing we still need one info fetch to complete the record.
    let should_fetch_info = fetch_info && props.as_ref().is_none_or(|p| p.details.is_none());

    // 1. Fetch Info from Provider (only when missing in input DB)
    if should_fetch_info {
        if let Some(series_id) = series_id_opt {
            // Fetch Info from Provider
            let info_url = xtream::get_xtream_player_api_info_url(input, XtreamCluster::Series, series_id)
                .ok_or_else(|| shared::error::TuliproxError::Config("Failed to build info URL".to_string()))?;

            let input_source = InputSource::from(input).with_url(info_url);
            let content = xtream::get_xtream_stream_info_content(app_config, client, &input_source, true)
                .await
                .map_err(|e| shared::error::TuliproxError::Config(format!("{e}")))?;

            if content.is_empty() {
                debug!("Series {display_id}: provider returned empty content for info fetch");
            } else {
                let canonical_series_id = existing_item.as_ref().map_or(series_id, |item| item.get_virtual_id().get());
                match serde_json::from_str::<Value>(&content) {
                    Ok(mut json_value) => {
                        if let Some(info) = json_value.get_mut("info").and_then(|v| v.as_object_mut()) {
                            tuliprox_core::model::normalize_release_date(info);
                        }

                        match serde_json::from_value::<XtreamSeriesInfo>(json_value) {
                            Ok(info) => {
                                debug!("Series {display_id}: building props from info using canonical series id {canonical_series_id}");
                                props = Some(SeriesStreamProperties::from_info_without_existing(
                                    &info,
                                    canonical_series_id,
                                ));
                                fetched_new = true;
                                properties_updated = true;
                            }
                            Err(e) => {
                                return Err(shared::error::TuliproxError::Config(format!(
                                    "Series {display_id}: failed to parse XtreamSeriesInfo: {e}. Raw response: {content}"
                                )));
                            }
                        }
                    }
                    Err(e) => {
                        return Err(shared::error::TuliproxError::Config(format!(
                            "Series {display_id}: failed to parse provider response as JSON: {e}. Raw response: {content}"
                        )));
                    }
                }
            }
        }
    }

    // If we don't have props, verify if we can proceed with minimal dummy props
    if props.is_none() {
        if let Some(title) = playlist_title.or_else(|| existing_item.as_ref().map(|i| i.title.as_ref())) {
            let new_props = SeriesStreamProperties {
                name: title.into(),
                series_id: series_id_opt.unwrap_or(0),
                ..Default::default()
            };
            props = Some(new_props);
        } else {
            return Err(shared::error::TuliproxError::Config(format!(
                "No Series properties available and no title found for {display_id}"
            )));
        }
    }

    let mut properties = props.ok_or_else(|| {
        shared::error::TuliproxError::Config(format!(
            "No Series properties available after fallback creation for {display_id}"
        ))
    })?;

    // Xtream's legacy ResolveTmdb input flag currently gates missing identity/temporal fact enrichment.
    let missing_fact_policy = MissingFactEnrichmentPolicy::fill_missing(
        resolve_missing_facts && input.has_flag(ConfigInputFlags::ResolveTmdb),
    );

    // 2. Resolve missing TMDB/date facts if allowed for this input
    let title_candidate = playlist_title.or_else(|| existing_item.as_ref().map(|i| i.title.as_ref()));
    let has_title_candidate = title_candidate.is_some_and(|title| !title.is_empty()) || !properties.name.is_empty();

    if missing_fact_policy.should_resolve_missing_facts(properties.tmdb.is_none(), properties.release_date.is_none())
        && has_title_candidate
    {
        // Try extracting year locally first
        if missing_fact_policy.should_try_parsed_title_supplier(properties.release_date.is_none()) {
            let local_patch = {
                let candidates = [title_candidate, (!properties.name.is_empty()).then_some(properties.name.as_ref())];
                series_fact_patch_from_title_candidates(&properties, candidates.into_iter().flatten())
            };
            if let Some((title, year, patch)) = local_patch {
                debug_if_enabled!("Parsed local year for Series '{}': {}", title, year);
                if apply_fact_patch_to_series(&mut properties, &patch) {
                    properties_updated = true;
                }
            }
        }

        if missing_fact_policy.should_try_tmdb_supplier(properties.tmdb.is_none(), properties.release_date.is_none()) {
            let config = app_config.config.load();
            let library_config = config.library.as_ref();
            let metadata_update_config = config.metadata_update.as_ref();
            let tmdb_storage = metadata_update_config
                .filter(|cfg| cfg.tmdb.enabled)
                .map(|cfg| MetadataStorage::new(std::path::PathBuf::from(&cfg.cache_path)));
            let meta_resolver =
                MetadataResolver::from_config(library_config, metadata_update_config, client.clone(), tmdb_storage);

            let mut meta = None;
            let mut tried_title = false;

            // 1. & 2. Playlist Title
            if let Some(title) = title_candidate {
                if !title.is_empty() {
                    trace!("Resolving TMDB for Series using Playlist Title '{title}' (ID: {display_id})...");
                    meta = meta_resolver.resolve_from_title(title, properties.tmdb, false, true).await;
                    tried_title = true;
                }
            }

            // 3. API Name (fallback)
            if (meta.is_none() || (meta.as_ref().is_some_and(|m| m.tmdb_id().is_none()))) && !properties.name.is_empty()
            {
                let title_already_tried =
                    if let Some(t) = title_candidate { t == properties.name.as_ref() } else { false };
                if !tried_title || !title_already_tried {
                    debug!("Fallback to API Name '{}'...", properties.name);
                    meta = meta_resolver.resolve_from_title(&properties.name, properties.tmdb, false, true).await;
                }
            }

            if let Some(m) = meta {
                let patch = series_fact_patch_from_metadata(&properties, &m);
                if apply_fact_patch_to_series(&mut properties, &patch) {
                    properties_updated = true;
                    let id_display = properties.tmdb.map_or("None".to_string(), |id| id.to_string());
                    debug_if_enabled!("Resolved TMDB for Series ID {}: {}", display_id, id_display);
                }
            }
        }
    }

    // Report whether TMDB and release date are present (regardless of whether we updated them this call).
    if let Some(out) = tmdb_and_date_present_out {
        let has_tmdb = properties.tmdb.is_some();
        let has_date = properties.release_date.is_some();
        out.store(has_tmdb && has_date, Ordering::Relaxed);
    }

    // 3. Probe Episodes (if enabled)
    if do_probe && app_config.is_ffprobe_enabled().await {
        if let Some(details) = properties.details.as_mut() {
            if let Some(episodes) = details.episodes.as_mut() {
                let config = app_config.config.load();
                let probe_priority = config
                    .metadata_update
                    .as_ref()
                    .map_or(default_probe_user_priority(), |cfg| cfg.probe.user_priority);
                let user_agent = config.default_user_agent.clone();

                let input_url = input.url.as_str();
                let input_username = input.username.as_deref().unwrap_or("");
                let input_password = input.password.as_deref().unwrap_or("");

                let mut probed_count = 0;
                let mut missing_any = false;

                for ep in episodes {
                    let missing_video = !MediaQuality::is_valid_media_info(ep.video.as_deref());
                    let missing_audio = !MediaQuality::is_valid_media_info(ep.audio.as_deref());

                    if missing_video || missing_audio {
                        missing_any = true;

                        // Acquire Connection logic
                        let temp_handle = if active_handle.is_some() {
                            None
                        } else {
                            active_provider
                                .acquire_connection_for_probe(&input.name, probe_priority)
                                .map(|handle| ProbeHandleGuard::new(active_provider, handle))
                        };

                        if active_handle.is_some() || temp_handle.is_some() {
                            let episode_url =
                                create_xtream_series_episode_url(input_url, input_username, input_password, ep);
                            let probe_url = match input.resolve_url(&episode_url) {
                                Ok(url) => url,
                                Err(err) => {
                                    if err.kind() == shared::error::ErrorKind::ConfigInput {
                                        debug!(
                                            "Provider config resolution failed for series episode '{}' (S{}E{}): {err}",
                                            ep.title, ep.season, ep.episode_num
                                        );
                                        if probe_failure.is_none() {
                                            probe_failure = Some(ProbeFailureKind::Other);
                                        }
                                    } else {
                                        warn!(
                                            "Failed to resolve probe URL for series episode '{}' (S{}E{}): {err}",
                                            ep.title, ep.season, ep.episode_num
                                        );
                                        probe_pending = true;
                                        if probe_failure.is_none() {
                                            probe_failure = Some(ProbeFailureKind::Other);
                                        }
                                    }
                                    if let Some(guard) = temp_handle {
                                        guard.release();
                                    }
                                    continue;
                                }
                            };
                            if !is_supported_probe_url(probe_url.as_ref()) {
                                debug!(
                                    "Skipping unsupported series probe for episode '{}' (S{}E{}): {}",
                                    ep.title,
                                    ep.season,
                                    ep.episode_num,
                                    probe_url.as_ref()
                                );
                                if let Some(guard) = temp_handle {
                                    guard.release();
                                }
                                continue;
                            }

                            // Specific logging for the user to follow
                            let missing_reason = if missing_video && missing_audio {
                                "video/audio"
                            } else if missing_video {
                                "video"
                            } else {
                                "audio"
                            };
                            debug!(
                                "Probing Series Episode '{}' (S{}E{}) - Missing {}",
                                ep.title, ep.season, ep.episode_num, missing_reason
                            );

                            let cancel_token = select_cancel_token(
                                temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
                                active_handle,
                            );
                            let is_remote_probe = reqwest::Url::parse(probe_url.as_ref())
                                .is_ok_and(|u| matches!(u.scheme(), "http" | "https"));
                            let probe_params = tuliprox_core::utils::ffmpeg::ProbeParams {
                                url: probe_url.as_ref(),
                                user_agent: user_agent.as_deref(),
                                analyze_duration: probe_settings.analyze_duration_micros,
                                probe_size: probe_settings.probe_size_bytes,
                                timeout_secs: probe_settings.timeout_secs,
                            };
                            let probe_result = if is_remote_probe {
                                FfmpegExecutor::new()
                                    .probe_remote_seekable_url_with_cancel(client, &probe_params, cancel_token)
                                    .await
                            } else {
                                FfmpegExecutor::new()
                                    .probe_url_with_cancel(&probe_params, config.proxy.as_ref(), cancel_token)
                                    .await
                            };
                            match probe_result {
                                ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, _stats) => {
                                    if let Some(v) = raw_video {
                                        ep.video = Some(v.to_string().into());
                                        properties_updated = true;
                                    }
                                    if let Some(a) = raw_audio {
                                        ep.audio = Some(a.to_string().into());
                                        properties_updated = true;
                                    }
                                    probed_count += 1;
                                }
                                ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => {
                                    probe_failure = Some(ProbeFailureKind::NotFound);
                                }
                                ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => {
                                    probe_pending = true;
                                    if probe_failure.is_none() {
                                        probe_failure = Some(ProbeFailureKind::Other);
                                    }
                                }
                                ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => {
                                    probe_pending = true;
                                    if probe_failure.is_none() {
                                        probe_failure = Some(ProbeFailureKind::Cancelled);
                                    }
                                }
                            }
                        } else {
                            probe_pending = true;
                            warn!("Skipping probe for series episode {} due to connection limits", ep.title);
                        }

                        if let Some(guard) = temp_handle {
                            guard.release();
                        }
                    }
                }
                if probed_count > 0 {
                    info!("Probed {probed_count} episodes for Series ID {display_id}");
                } else if !missing_any {
                    debug!("Series probe skipped for ID {display_id} (all episodes already have A/V details)");
                }
            } else {
                debug!("Series probe skipped for ID {display_id} (no episode details available)");
            }
        } else {
            debug!("Series probe skipped for ID {display_id} (no series details available)");
        }
    }

    if let Some(out) = probe_pending_out {
        out.store(probe_pending, Ordering::Relaxed);
    }

    let probe_only_unresolved =
        do_probe && !fetch_info && !resolve_missing_facts && !properties_updated && !fetched_new;
    if probe_only_unresolved {
        if let Some(kind) = probe_failure {
            let err = match kind {
                ProbeFailureKind::NotFound => shared::error::TuliproxError::Config(format!(
                    "Probe failed with 404 Not Found for Series {display_id}"
                )),
                ProbeFailureKind::Other => {
                    shared::error::TuliproxError::Config(format!("Probe failed for Series {display_id}"))
                }
                ProbeFailureKind::Cancelled => {
                    shared::error::TuliproxError::Config(format!("Probe cancelled for Series {display_id}"))
                }
            };
            return Err(err);
        }
    }

    // 4. Persist
    if properties_updated || fetched_new {
        if save {
            if let Some(series_id) = series_id_opt {
                persists_input_series_info(
                    app_config,
                    &storage_path,
                    XtreamCluster::Series,
                    &input.name,
                    series_id,
                    &properties,
                )
                .await
                .map_err(|e| shared::error::TuliproxError::Config(format!("Persist error: {e}")))?;

                debug_if_enabled!("Successfully updated Series metadata for ID {}", series_id);
            }
        }
        return Ok(Some(properties));
    }

    Ok(None)
}