use super::stalker::StalkerCluster; use shared::{ error::TuliproxError, model::{stalker::StalkerStreamKind, stalker_item::StalkerPlaylistItem, UpdateQualityPolicy, XtreamCluster}, }; use std::{ collections::HashMap, path::Path, sync::Arc, time::{Instant, SystemTime, UNIX_EPOCH}, }; use tuliprox_core::model::{ evaluate_update_quality, AppConfig, ClusterForceUpdate, ClusterUpdateAcceptance, ClusterUpdateRejection, ConfigInput, ConfigInputFlags, ConfigInputUpdateQuality, }; use tuliprox_iptv::stalker::{ catalog::{StalkerCategory, StalkerRawItem}, client::StalkerApiClient, error::StalkerError, parser, profile::StalkerHandshake, }; use tuliprox_repository::{ stalker_generation_repository::{ cleanup_obsolete_generations, clear_checkpoint, generation_data_path, load_active_manifest, load_checkpoint, publish_selection, save_checkpoint, StalkerCheckpoint, StalkerGenerationData, StalkerRefreshPhase, }, stalker_repository::{ count_stalker_items_at, load_stalker_items_after, prepare_stalker_episode_series_at, promote_stalker_file, remove_stalker_file, snapshot_stalker_epg_at, snapshot_stalker_items_at, upsert_stalker_epg_at, upsert_stalker_items_at, }, }; const MAX_RETRIES: u8 = 3; const SKIPPED_SAMPLE_LIMIT: usize = 32; const LIVE_SELECTION: u8 = 0b0001; const VOD_SELECTION: u8 = 0b0010; const SERIES_SELECTION: u8 = 0b0100; const EPG_SELECTION: u8 = 0b1000; const MEDIA_SELECTION: u8 = LIVE_SELECTION | VOD_SELECTION | SERIES_SELECTION; pub enum StalkerRefreshLimit { Deadline(Instant), Units { remaining: usize }, Unlimited, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum StalkerRefreshMode { ServerSlice, Parallel, Complete, } impl StalkerRefreshMode { pub fn budget(self) -> StalkerRefreshBudget { match self { Self::ServerSlice => StalkerRefreshBudget::deadline(Instant::now() + std::time::Duration::from_mins(45)), Self::Parallel => StalkerRefreshBudget::units(8), Self::Complete => StalkerRefreshBudget::unlimited(), } } } pub struct StalkerRefreshBudget { limit: StalkerRefreshLimit, } impl StalkerRefreshBudget { pub fn deadline(deadline: Instant) -> Self { Self { limit: StalkerRefreshLimit::Deadline(deadline) } } pub fn unlimited() -> Self { Self { limit: StalkerRefreshLimit::Unlimited } } fn units(remaining: usize) -> Self { Self { limit: StalkerRefreshLimit::Units { remaining } } } fn completed_unit(&mut self) { if let StalkerRefreshLimit::Units { remaining } = &mut self.limit { *remaining = remaining.saturating_sub(1); } } fn should_yield(&self) -> bool { match self.limit { StalkerRefreshLimit::Deadline(deadline) => Instant::now() >= deadline, StalkerRefreshLimit::Units { remaining } => remaining == 0, StalkerRefreshLimit::Unlimited => false, } } } #[derive(Clone, Copy)] #[allow(clippy::struct_excessive_bools)] pub struct StalkerClusterSelection { live: bool, vod: bool, series: bool, epg: bool, } impl From<&ConfigInput> for StalkerClusterSelection { fn from(input: &ConfigInput) -> Self { let live = !input.has_flag(ConfigInputFlags::SkipLive); Self { live, vod: !input.has_flag(ConfigInputFlags::SkipVod), series: !input.has_flag(ConfigInputFlags::SkipSeries), epg: live && input.has_flag(ConfigInputFlags::StalkerBulkEpg), } } } impl StalkerClusterSelection { pub fn requested(input: &ConfigInput, requested: &[StalkerCluster]) -> Self { let configured = Self::from(input); let live = configured.live && requested.contains(&StalkerCluster::Live); Self { live, vod: configured.vod && requested.contains(&StalkerCluster::Vod), series: configured.series && requested.contains(&StalkerCluster::Series), epg: configured.epg && live, } } fn mask(self) -> u8 { u8::from(self.live) | (u8::from(self.vod) << 1) | (u8::from(self.series) << 2) | (u8::from(self.epg) << 3) } } /// Requested clusters and the publication policy applied once their generation is complete. #[derive(Clone, Copy)] pub struct StalkerRefreshPlan { selection: StalkerClusterSelection, update_quality: ConfigInputUpdateQuality, quality_bypass_mask: u8, } impl StalkerRefreshPlan { pub fn requested(input: &ConfigInput, requested: &[StalkerCluster], quality_policy: UpdateQualityPolicy) -> Self { let selection = StalkerClusterSelection::requested(input, requested); let update_quality = input.options.as_ref().map_or_else(ConfigInputUpdateQuality::default, |options| options.update_quality); let quality_bypass_mask = matches!(quality_policy, UpdateQualityPolicy::Bypass) .then_some(selection.mask() & MEDIA_SELECTION) .unwrap_or(0); Self { selection, update_quality, quality_bypass_mask } } } fn first_phase(selection: StalkerClusterSelection) -> StalkerRefreshPhase { if selection.live { StalkerRefreshPhase::LiveBulk } else { next_phase_after_live(selection) } } fn next_phase_after_live(selection: StalkerClusterSelection) -> StalkerRefreshPhase { if selection.vod { StalkerRefreshPhase::Vod { page: 1 } } else { next_phase_after_vod(selection) } } fn next_phase_after_vod(selection: StalkerClusterSelection) -> StalkerRefreshPhase { if selection.series { StalkerRefreshPhase::SeriesRoots { page: 1 } } else if selection.epg { StalkerRefreshPhase::Epg } else { StalkerRefreshPhase::Complete } } fn next_phase_after_series(selection: StalkerClusterSelection) -> StalkerRefreshPhase { if selection.epg { StalkerRefreshPhase::Epg } else { StalkerRefreshPhase::Complete } } #[derive(Debug)] pub enum StalkerRefreshOutcome { Complete { quality_acceptances: Vec, quality_rejections: Vec, force_updates: Vec, }, Failed { quality_acceptances: Vec, quality_rejections: Vec, force_updates: Vec, error: TuliproxError, }, Yielded { phase: StalkerRefreshPhase, processed: u64, skipped: u64, error: Option, }, Terminal(TuliproxError), } fn generation_id() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map_or(0, |duration| u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)) } async fn load_or_start_checkpoint( storage_path: &Path, identity_fingerprint: u64, selection: StalkerClusterSelection, quality_bypass_mask: u8, ) -> Result { if let Some(mut state) = load_checkpoint(storage_path, identity_fingerprint).await? { if state.selection_mask == selection.mask() { let combined_bypass_mask = state.quality_bypass_mask | quality_bypass_mask; if combined_bypass_mask != state.quality_bypass_mask { state.quality_bypass_mask = combined_bypass_mask; save_checkpoint(storage_path, &state).await?; } return Ok(state); } } let mut state = StalkerCheckpoint::new(identity_fingerprint, generation_id(), selection.mask(), chrono::Utc::now().timestamp()); state.phase = first_phase(selection); state.quality_bypass_mask = quality_bypass_mask; save_checkpoint(storage_path, &state).await?; Ok(state) } async fn evaluate_completed_refresh_publication( app_config: &Arc, storage_path: &Path, identity_fingerprint: u64, checkpoint: &StalkerCheckpoint, update_quality: ConfigInputUpdateQuality, ) -> Result { let needs_quality_evaluation = selection_needs_quality_evaluation(checkpoint.selection_mask, checkpoint.quality_bypass_mask, update_quality); let has_quality_bypass = checkpoint.quality_bypass_mask & MEDIA_SELECTION != 0; if needs_quality_evaluation || has_quality_bypass { let active_manifest = if needs_quality_evaluation { Some(load_active_manifest(storage_path, identity_fingerprint).await?) } else { None }; evaluate_generation_publication( app_config, storage_path, checkpoint.generation, checkpoint.selection_mask, checkpoint.quality_bypass_mask, update_quality, active_manifest.as_ref(), ) .await } else { Ok(StalkerGenerationPublication { accepted_selection_mask: checkpoint.selection_mask, quality_acceptances: Vec::new(), quality_rejections: Vec::new(), force_updates: Vec::new(), }) } } async fn ensure_completed_refresh_has_usable_playlist( app_config: &Arc, storage_path: &Path, identity_fingerprint: u64, checkpoint: &StalkerCheckpoint, accepted_selection_mask: u8, ) -> Result<(), TuliproxError> { let accepted_media_selection = accepted_selection_mask & MEDIA_SELECTION; let accepted_enforced_selection = accepted_media_selection & !checkpoint.quality_bypass_mask; if checkpoint.processed == 0 && accepted_enforced_selection != 0 { let active = load_active_manifest(storage_path, identity_fingerprint).await?; let mut retained_has_items = false; if accepted_selection_mask & LIVE_SELECTION == 0 { if let Some(files) = active.live.as_ref() { retained_has_items = !load_stalker_items_after(app_config, &files.data, None, 1).await?.is_empty(); } } if !retained_has_items && accepted_selection_mask & VOD_SELECTION == 0 { if let Some(files) = active.vod.as_ref() { retained_has_items = !load_stalker_items_after(app_config, &files.data, None, 1).await?.is_empty(); } } if !retained_has_items && accepted_selection_mask & SERIES_SELECTION == 0 { if let Some(files) = active.series.as_ref() { retained_has_items = !load_stalker_items_after(app_config, &files.roots, None, 1).await?.is_empty() || !load_stalker_items_after(app_config, &files.episodes, None, 1).await?.is_empty(); } } if !retained_has_items { clear_checkpoint(storage_path).await?; cleanup_obsolete_generations(storage_path, &active).await?; return Err(TuliproxError::RepositoryPlaylist( "Refusing to publish an empty Stalker playlist; existing data was retained".to_string(), )); } } Ok(()) } async fn publish_completed_refresh( storage_path: &Path, identity_fingerprint: u64, checkpoint: &StalkerCheckpoint, accepted_selection_mask: u8, ) -> Result<(), TuliproxError> { let manifest = publish_selection(storage_path, identity_fingerprint, checkpoint.generation, accepted_selection_mask).await?; clear_checkpoint(storage_path).await?; cleanup_obsolete_generations(storage_path, &manifest).await?; Ok(()) } struct StalkerRefreshReports { quality_acceptances: Vec, quality_rejections: Vec, force_updates: Vec, error: Option, } impl StalkerRefreshReports { fn into_outcome(self) -> StalkerRefreshOutcome { let Self { quality_acceptances, quality_rejections, force_updates, error } = self; if let Some(error) = error { StalkerRefreshOutcome::Failed { quality_acceptances, quality_rejections, force_updates, error } } else { StalkerRefreshOutcome::Complete { quality_acceptances, quality_rejections, force_updates } } } } #[cfg(test)] async fn finish_completed_refresh( app_config: &Arc, storage_path: &Path, identity_fingerprint: u64, checkpoint: &StalkerCheckpoint, update_quality: ConfigInputUpdateQuality, ) -> Result, TuliproxError> { let publication = evaluate_completed_refresh_publication( app_config, storage_path, identity_fingerprint, checkpoint, update_quality, ) .await?; let StalkerGenerationPublication { accepted_selection_mask, quality_rejections, quality_acceptances: _, force_updates: _, } = publication; publish_completed_refresh(storage_path, identity_fingerprint, checkpoint, accepted_selection_mask).await?; Ok(quality_rejections) } async fn finish_completed_refresh_checked( app_config: &Arc, storage_path: &Path, identity_fingerprint: u64, checkpoint: &StalkerCheckpoint, update_quality: ConfigInputUpdateQuality, ) -> Result { let publication = evaluate_completed_refresh_publication( app_config, storage_path, identity_fingerprint, checkpoint, update_quality, ) .await?; let StalkerGenerationPublication { accepted_selection_mask, quality_acceptances, quality_rejections, force_updates, } = publication; let completion_result = async { ensure_completed_refresh_has_usable_playlist( app_config, storage_path, identity_fingerprint, checkpoint, accepted_selection_mask, ) .await?; publish_completed_refresh(storage_path, identity_fingerprint, checkpoint, accepted_selection_mask).await } .await; Ok(StalkerRefreshReports { quality_acceptances, quality_rejections, force_updates, error: completion_result.err() }) } struct StalkerGenerationPublication { accepted_selection_mask: u8, quality_acceptances: Vec, quality_rejections: Vec, force_updates: Vec, } fn selection_needs_quality_evaluation( requested_selection_mask: u8, quality_bypass_mask: u8, update_quality: ConfigInputUpdateQuality, ) -> bool { [StalkerCluster::Live, StalkerCluster::Vod, StalkerCluster::Series].into_iter().any(|cluster| { let bit = selection_bit(cluster); requested_selection_mask & bit != 0 && quality_bypass_mask & bit == 0 && update_quality.threshold(quality_cluster(cluster)) != 0 }) } const fn selection_bit(cluster: StalkerCluster) -> u8 { match cluster { StalkerCluster::Live => LIVE_SELECTION, StalkerCluster::Vod => VOD_SELECTION, StalkerCluster::Series => SERIES_SELECTION, } } const fn quality_cluster(cluster: StalkerCluster) -> XtreamCluster { match cluster { StalkerCluster::Live => XtreamCluster::Live, StalkerCluster::Vod => XtreamCluster::Video, StalkerCluster::Series => XtreamCluster::Series, } } async fn count_stalker_cluster_files( app_config: &Arc, primary: &Path, secondary: Option<&Path>, ) -> Result { let primary_count = count_stalker_items_at(app_config, primary).await?.unwrap_or_default(); let secondary_count = match secondary { Some(path) => count_stalker_items_at(app_config, path).await?.unwrap_or_default(), None => 0, }; primary_count.checked_add(secondary_count).ok_or_else(|| { TuliproxError::RepositoryStalker("Stalker cluster item count exceeds platform capacity".to_string()) }) } async fn count_candidate_cluster( app_config: &Arc, storage_path: &Path, generation: u64, cluster: StalkerCluster, ) -> Result { let primary_data = match cluster { StalkerCluster::Live => StalkerGenerationData::Live, StalkerCluster::Vod => StalkerGenerationData::Vod, StalkerCluster::Series => StalkerGenerationData::SeriesRoots, }; let primary = generation_data_path(storage_path, generation, primary_data); let secondary = (cluster == StalkerCluster::Series) .then(|| generation_data_path(storage_path, generation, StalkerGenerationData::SeriesEpisodes)); count_stalker_cluster_files(app_config, &primary, secondary.as_deref()).await } async fn count_active_cluster( app_config: &Arc, manifest: &tuliprox_repository::stalker_generation_repository::StalkerActiveManifest, cluster: StalkerCluster, ) -> Result, TuliproxError> { match cluster { StalkerCluster::Live => match manifest.live.as_ref() { Some(files) => count_stalker_cluster_files(app_config, &files.data, None).await.map(Some), None => Ok(None), }, StalkerCluster::Vod => match manifest.vod.as_ref() { Some(files) => count_stalker_cluster_files(app_config, &files.data, None).await.map(Some), None => Ok(None), }, StalkerCluster::Series => match manifest.series.as_ref() { Some(files) => count_stalker_cluster_files(app_config, &files.roots, Some(&files.episodes)).await.map(Some), None => Ok(None), }, } } async fn evaluate_generation_publication( app_config: &Arc, storage_path: &Path, generation: u64, requested_selection_mask: u8, quality_bypass_mask: u8, update_quality: ConfigInputUpdateQuality, active_manifest: Option<&tuliprox_repository::stalker_generation_repository::StalkerActiveManifest>, ) -> Result { let mut accepted_selection_mask = requested_selection_mask; let mut quality_acceptances = Vec::new(); let mut quality_rejections = Vec::new(); let mut force_updates = Vec::new(); for cluster in [StalkerCluster::Live, StalkerCluster::Vod, StalkerCluster::Series] { let selection_bit = selection_bit(cluster); if requested_selection_mask & selection_bit == 0 { continue; } let report_cluster = quality_cluster(cluster); let threshold = update_quality.threshold(report_cluster); if quality_bypass_mask & selection_bit != 0 { let candidate_count = count_candidate_cluster(app_config, storage_path, generation, cluster).await?; force_updates.push(ClusterForceUpdate { cluster: report_cluster, candidate_count, configured_threshold: threshold, }); continue; } if threshold == 0 { continue; } let candidate_count = count_candidate_cluster(app_config, storage_path, generation, cluster).await?; let Some(active_manifest) = active_manifest else { return Err(TuliproxError::RepositoryStalker(format!( "Missing active manifest for enforced {report_cluster} quality evaluation" ))); }; let current_count = count_active_cluster(app_config, active_manifest, cluster).await?; let decision = evaluate_update_quality(current_count, candidate_count, threshold); if let Some(acceptance) = decision.acceptance(report_cluster) { quality_acceptances.push(acceptance); } else if let Some(rejection) = decision.rejection(report_cluster) { accepted_selection_mask &= !selection_bit; if cluster == StalkerCluster::Live { accepted_selection_mask &= !EPG_SELECTION; } quality_rejections.push(rejection); } } Ok(StalkerGenerationPublication { accepted_selection_mask, quality_acceptances, quality_rejections, force_updates }) } fn category_map(categories: Vec) -> HashMap { categories.into_iter().filter_map(|category| category.id.parse().ok().map(|id| (id, category))).collect() } fn category_map_result( result: Result, StalkerError>, ) -> Result, TuliproxError> { result.map(category_map).map_err(|err| provider_error(&err)) } fn map_items( raw_items: &[StalkerRawItem], categories: &HashMap, kind: StalkerStreamKind, added_at: i64, ) -> Vec { raw_items .iter() .map(|raw| { let category = raw.category_id().and_then(|value| value.parse::().ok()).and_then(|id| categories.get(&id)); parser::map_stalker_to_playlist_item(raw, category, kind, added_at) }) .collect() } fn catalog_page_signature<'a>(ids: impl Iterator>) -> u64 { ids.fold(14_695_981_039_346_656_037_u64, |hash, id| { id.unwrap_or_default() .bytes() .chain([0]) .fold(hash, |hash, byte| (hash ^ u64::from(byte)).wrapping_mul(1_099_511_628_211)) }) } fn ensure_page_advanced( next_page: Option, previous_signature: Option, signature: u64, page: u32, portal_type: &'static str, ) -> Result<(), TuliproxError> { if next_page.is_some() && previous_signature == Some(signature) { return Err(provider_error(&StalkerError::CatalogIncomplete { portal_type, reason: format!("page {page} repeated"), })); } Ok(()) } fn provider_error(err: &StalkerError) -> TuliproxError { TuliproxError::ProviderConnection(format!("Stalker client error: {err}")) } async fn finish_terminal_refresh(storage_path: &Path) -> Result { clear_checkpoint(storage_path).await?; Ok(StalkerRefreshOutcome::Terminal(TuliproxError::ProviderConnection( "Stalker refresh reached a terminal state".to_string(), ))) } async fn yield_after_error( storage_path: &Path, mut checkpoint: StalkerCheckpoint, error: TuliproxError, ) -> Result { checkpoint.retry_count = checkpoint.retry_count.saturating_add(1); if checkpoint.retry_count >= MAX_RETRIES { checkpoint.phase = StalkerRefreshPhase::Terminal; save_checkpoint(storage_path, &checkpoint).await?; return Ok(StalkerRefreshOutcome::Terminal(error)); } save_checkpoint(storage_path, &checkpoint).await?; Ok(StalkerRefreshOutcome::Yielded { phase: checkpoint.phase, processed: checkpoint.processed, skipped: checkpoint.skipped_count, error: Some(error), }) } #[allow(clippy::too_many_lines)] pub async fn advance_stalker_refresh( app_config: &Arc, api_client: &StalkerApiClient, handshake: &StalkerHandshake, refresh_plan: StalkerRefreshPlan, storage_path: &Path, identity_fingerprint: u64, mut budget: StalkerRefreshBudget, ) -> Result { let StalkerRefreshPlan { selection, update_quality, quality_bypass_mask } = refresh_plan; let mut checkpoint = load_or_start_checkpoint(storage_path, identity_fingerprint, selection, quality_bypass_mask).await?; if checkpoint.phase == StalkerRefreshPhase::Terminal { return finish_terminal_refresh(storage_path).await; } let added_at = checkpoint.started_at; let mut live_categories = None; let mut vod_categories = None; let mut series_categories = None; let mut used_episode_ids = None; loop { if budget.should_yield() { return Ok(StalkerRefreshOutcome::Yielded { phase: checkpoint.phase, processed: checkpoint.processed, skipped: checkpoint.skipped_count, error: None, }); } match checkpoint.phase.clone() { StalkerRefreshPhase::LiveBulk => { let categories = if let Some(categories) = &live_categories { categories } else { match category_map_result(api_client.get_live_categories(handshake).await) { Ok(categories) => live_categories.insert(categories), Err(err) => return yield_after_error(storage_path, checkpoint, err).await, } }; match api_client.get_all_channels(handshake).await { Ok(raw) if raw.is_empty() => { checkpoint.phase = StalkerRefreshPhase::Live { page: 1 }; checkpoint.retry_count = 0; } Ok(raw) => { let items = map_items(&raw, categories, StalkerStreamKind::Live, added_at); let path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Live); snapshot_stalker_items_at(app_config, path.clone(), &items).await?; checkpoint.processed = checkpoint.processed.saturating_add(items.len() as u64); checkpoint.phase = next_phase_after_live(selection); checkpoint.retry_count = 0; } Err(err) if err.is_unsupported_catalog_action() => { checkpoint.phase = StalkerRefreshPhase::Live { page: 1 }; checkpoint.retry_count = 0; } Err(err) => return yield_after_error(storage_path, checkpoint, provider_error(&err)).await, } } StalkerRefreshPhase::Live { page } => { let categories = if let Some(categories) = &live_categories { categories } else { match category_map_result(api_client.get_live_categories(handshake).await) { Ok(categories) => live_categories.insert(categories), Err(err) => return yield_after_error(storage_path, checkpoint, err).await, } }; let response = match api_client.get_live_streams_page(handshake, page).await { Ok(response) => response, Err(err) => return yield_after_error(storage_path, checkpoint, provider_error(&err)).await, }; let signature = catalog_page_signature(response.items.iter().map(|item| item.id.as_deref())); if let Err(err) = ensure_page_advanced(response.next_page, checkpoint.page_signature, signature, page, "itv") { return yield_after_error(storage_path, checkpoint, err).await; } let items = map_items(&response.items, categories, StalkerStreamKind::Live, added_at); let path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Live); upsert_stalker_items_at(app_config, &path, &items).await?; checkpoint.processed = checkpoint.processed.saturating_add(items.len() as u64); checkpoint.phase = if let Some(next_page) = response.next_page { checkpoint.page_signature = Some(signature); StalkerRefreshPhase::Live { page: next_page } } else { checkpoint.page_signature = None; next_phase_after_live(selection) }; checkpoint.retry_count = 0; } StalkerRefreshPhase::Vod { page } => { let categories = if let Some(categories) = &vod_categories { categories } else { match category_map_result(api_client.get_vod_categories(handshake).await) { Ok(categories) => vod_categories.insert(categories), Err(err) => return yield_after_error(storage_path, checkpoint, err).await, } }; let response = match api_client.get_vod_streams_page(handshake, page).await { Ok(response) => response, Err(err) => return yield_after_error(storage_path, checkpoint, provider_error(&err)).await, }; let signature = catalog_page_signature(response.items.iter().map(|item| item.id.as_deref())); if let Err(err) = ensure_page_advanced(response.next_page, checkpoint.page_signature, signature, page, "vod") { return yield_after_error(storage_path, checkpoint, err).await; } let items = map_items(&response.items, categories, StalkerStreamKind::Movie, added_at); let path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Vod); upsert_stalker_items_at(app_config, &path, &items).await?; checkpoint.processed = checkpoint.processed.saturating_add(items.len() as u64); checkpoint.phase = if let Some(next_page) = response.next_page { checkpoint.page_signature = Some(signature); StalkerRefreshPhase::Vod { page: next_page } } else { checkpoint.page_signature = None; next_phase_after_vod(selection) }; checkpoint.retry_count = 0; } StalkerRefreshPhase::SeriesRoots { page } => { let categories = if let Some(categories) = &series_categories { categories } else { match category_map_result(api_client.get_series_categories(handshake).await) { Ok(categories) => series_categories.insert(categories), Err(err) => return yield_after_error(storage_path, checkpoint, err).await, } }; let response = match api_client.get_series_list_page(handshake, page).await { Ok(response) => response, Err(err) => return yield_after_error(storage_path, checkpoint, provider_error(&err)).await, }; let signature = catalog_page_signature(response.items.iter().map(|item| item.id.as_deref())); if let Err(err) = ensure_page_advanced(response.next_page, checkpoint.page_signature, signature, page, "series") { return yield_after_error(storage_path, checkpoint, err).await; } let roots: Vec<_> = response .items .iter() .map(|raw| { let category = raw .category_id .as_deref() .and_then(|value| value.parse::().ok()) .and_then(|id| categories.get(&id)); let mut root = parser::map_stalker_series_root(raw, category, added_at); let capabilities = &handshake.profile.portal_capabilities; root.nginx_secure_link = capabilities.nginx_secure_link; root.flussonic_tmp_link = capabilities.flussonic_temporary_link; root.wowza_tmp_link = capabilities.wowza_temporary_link; root.use_http_tmp_link = capabilities.use_http_temporary_link; root }) .collect(); let roots_path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::SeriesRoots); upsert_stalker_items_at(app_config, &roots_path, &roots).await?; checkpoint.processed = checkpoint.processed.saturating_add(roots.len() as u64); checkpoint.phase = match response.next_page { None => { checkpoint.page_signature = None; StalkerRefreshPhase::SeriesDetails { provider_id: None } } Some(next_page) => { checkpoint.page_signature = Some(signature); StalkerRefreshPhase::SeriesRoots { page: next_page } } }; checkpoint.retry_count = 0; } StalkerRefreshPhase::SeriesDetails { provider_id } => { let roots_path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::SeriesRoots); let mut roots = load_stalker_items_after(app_config, &roots_path, provider_id, 1).await?; let Some(root) = roots.pop() else { checkpoint.phase = next_phase_after_series(selection); checkpoint.retry_count = 0; save_checkpoint(storage_path, &checkpoint).await?; budget.completed_unit(); continue; }; let series_id = root.series_id.unwrap_or(root.stream_id); match api_client.get_series_details(handshake, series_id).await { Ok(details) => { let path = generation_data_path( storage_path, checkpoint.generation, StalkerGenerationData::SeriesEpisodes, ); let used = if let Some(used) = &mut used_episode_ids { used } else { used_episode_ids .insert(prepare_stalker_episode_series_at(app_config, &path, series_id).await?) }; let episodes = parser::map_stalker_series_details(&details, &root, added_at, used); upsert_stalker_items_at(app_config, &path, &episodes).await?; checkpoint.processed = checkpoint.processed.saturating_add(episodes.len() as u64); checkpoint.phase = StalkerRefreshPhase::SeriesDetails { provider_id: Some(root.stream_id) }; checkpoint.retry_count = 0; } Err(err) => { checkpoint.retry_count = checkpoint.retry_count.saturating_add(1); if checkpoint.retry_count < MAX_RETRIES { save_checkpoint(storage_path, &checkpoint).await?; return Ok(StalkerRefreshOutcome::Yielded { phase: checkpoint.phase, processed: checkpoint.processed, skipped: checkpoint.skipped_count, error: Some(provider_error(&err)), }); } checkpoint.skipped_count = checkpoint.skipped_count.saturating_add(1); if checkpoint.skipped_sample.len() < SKIPPED_SAMPLE_LIMIT { checkpoint.skipped_sample.push(series_id); } checkpoint.phase = StalkerRefreshPhase::SeriesDetails { provider_id: Some(root.stream_id) }; checkpoint.retry_count = 0; } } } StalkerRefreshPhase::Epg => { let path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Epg); let attempt_path = path.with_extension("attempt.db"); snapshot_stalker_epg_at(app_config, &attempt_path, &[]).await?; let app_config = Arc::clone(app_config); let sink_app_config = Arc::clone(&app_config); let batch_path = attempt_path.clone(); let epg_result = api_client .stream_bulk_epg(handshake, 24, 512, move |batch| { let app_config = Arc::clone(&sink_app_config); let batch_path = batch_path.clone(); async move { upsert_stalker_epg_at(&app_config, &batch_path, &batch) .await .map(|_| ()) .map_err(|err| StalkerError::Io(std::io::Error::other(err.to_string()))) } }) .await; if let Err(err) = epg_result { remove_stalker_file(&app_config, &attempt_path).await?; return yield_after_error(storage_path, checkpoint, provider_error(&err)).await; } promote_stalker_file(&app_config, &attempt_path, &path).await?; checkpoint.phase = StalkerRefreshPhase::Complete; checkpoint.retry_count = 0; } StalkerRefreshPhase::Complete => { let reports = finish_completed_refresh_checked( app_config, storage_path, identity_fingerprint, &checkpoint, update_quality, ) .await?; return Ok(reports.into_outcome()); } StalkerRefreshPhase::Terminal => { return finish_terminal_refresh(storage_path).await; } } save_checkpoint(storage_path, &checkpoint).await?; budget.completed_unit(); } } #[cfg(test)] mod tests { use super::*; use arc_swap::{ArcSwap, ArcSwapOption}; use shared::{ model::{ConfigInputUpdateQualityDto, ConfigPaths}, utils::Internable, }; use tuliprox_core::{ model::{ApiProxyConfig, Config, CustomStreamResponse, HdHomeRunConfig, MediaToolCapabilities, SourcesConfig}, utils::FileLockManager, }; use tuliprox_repository::stalker_generation_repository::{ load_active_manifest, save_active_manifest, ClusterFiles, SeriesFiles, StalkerActiveManifest, }; fn test_app_config(storage_dir: &Path) -> Arc { Arc::new(AppConfig { config: Arc::new(ArcSwap::from_pointee(Config { storage_dir: storage_dir.to_string_lossy().into_owned(), ..Config::default() })), sources: Arc::new(ArcSwap::from_pointee(SourcesConfig::default())), hdhomerun: Arc::new(ArcSwapOption::::default()), api_proxy: Arc::new(ArcSwapOption::::default()), file_locks: Arc::new(FileLockManager::default()), paths: Arc::new(ArcSwap::from_pointee(ConfigPaths { home_path: String::new(), config_path: String::new(), storage_path: storage_dir.to_string_lossy().into_owned(), config_file_path: String::new(), sources_file_path: String::new(), mapping_file_path: None, mapping_files_used: None, template_file_path: None, template_files_used: None, api_proxy_file_path: String::new(), custom_stream_response_path: None, })), custom_stream_response: Arc::new(ArcSwapOption::::default()), access_token_secret: [0; 32], encrypt_secret: [0; 16], media_tools: Arc::new(MediaToolCapabilities::new()), }) } fn update_quality(live: u8, vod: u8, series: u8) -> ConfigInputUpdateQuality { ConfigInputUpdateQuality::from(&ConfigInputUpdateQualityDto { live, vod, series }) } fn stalker_items(count: usize, kind: StalkerStreamKind, series_roots: bool) -> Vec { (0..count) .map(|index| StalkerPlaylistItem { stream_id: u32::try_from(index + 1).unwrap_or(u32::MAX), name: format!("item-{index}").intern(), stream_kind: kind, is_series: series_roots, ..StalkerPlaylistItem::default() }) .collect() } async fn write_generation_cluster( app_config: &Arc, storage_path: &Path, generation: u64, cluster: StalkerCluster, primary_count: usize, episode_count: usize, ) -> Result<(), TuliproxError> { let (primary_data, kind, series_roots) = match cluster { StalkerCluster::Live => (StalkerGenerationData::Live, StalkerStreamKind::Live, false), StalkerCluster::Vod => (StalkerGenerationData::Vod, StalkerStreamKind::Movie, false), StalkerCluster::Series => (StalkerGenerationData::SeriesRoots, StalkerStreamKind::Episode, true), }; let primary_path = generation_data_path(storage_path, generation, primary_data); let primary_items = stalker_items(primary_count, kind, series_roots); if primary_items.is_empty() { upsert_stalker_items_at(app_config, &primary_path, &primary_items).await?; } else { snapshot_stalker_items_at(app_config, primary_path, &primary_items).await?; } if cluster == StalkerCluster::Series { let episode_path = generation_data_path(storage_path, generation, StalkerGenerationData::SeriesEpisodes); snapshot_stalker_items_at( app_config, episode_path, &stalker_items(episode_count, StalkerStreamKind::Episode, false), ) .await?; } Ok(()) } fn complete_manifest(storage_path: &Path, identity_fingerprint: u64, generation: u64) -> StalkerActiveManifest { StalkerActiveManifest { schema: StalkerActiveManifest::empty(identity_fingerprint).schema, identity_fingerprint, live: Some(ClusterFiles { generation, data: generation_data_path(storage_path, generation, StalkerGenerationData::Live), }), vod: Some(ClusterFiles { generation, data: generation_data_path(storage_path, generation, StalkerGenerationData::Vod), }), series: Some(SeriesFiles { generation, roots: generation_data_path(storage_path, generation, StalkerGenerationData::SeriesRoots), episodes: generation_data_path(storage_path, generation, StalkerGenerationData::SeriesEpisodes), }), epg: Some(ClusterFiles { generation, data: generation_data_path(storage_path, generation, StalkerGenerationData::Epg), }), } } #[tokio::test] async fn terminal_checkpoint_is_cleared_after_restart() -> Result<(), Box> { let temp = tempfile::tempdir()?; let selection = StalkerClusterSelection { live: true, vod: false, series: false, epg: false }; let mut checkpoint = StalkerCheckpoint::new(17, 23, selection.mask(), 123); checkpoint.phase = StalkerRefreshPhase::Terminal; save_checkpoint(temp.path(), &checkpoint).await?; let outcome = finish_terminal_refresh(temp.path()).await?; assert!(matches!(outcome, StalkerRefreshOutcome::Terminal(_))); assert!(load_checkpoint(temp.path(), 17).await?.is_none()); Ok(()) } #[tokio::test] async fn complete_checkpoint_is_resumed_without_new_generation() -> Result<(), Box> { let temp = tempfile::tempdir()?; let selection = StalkerClusterSelection { live: true, vod: true, series: false, epg: false }; let mut checkpoint = StalkerCheckpoint::new(17, 23, selection.mask(), 123); checkpoint.phase = StalkerRefreshPhase::Complete; save_checkpoint(temp.path(), &checkpoint).await?; let resumed = load_or_start_checkpoint(temp.path(), 17, selection, 0).await?; assert_eq!(resumed.generation, 23); assert_eq!(resumed.phase, StalkerRefreshPhase::Complete); Ok(()) } #[tokio::test] async fn resumed_checkpoint_preserves_and_extends_the_quality_bypass_mask() -> Result<(), Box> { let temp = tempfile::tempdir()?; let selection = StalkerClusterSelection { live: true, vod: true, series: true, epg: false }; let mut checkpoint = StalkerCheckpoint::new(17, 23, selection.mask(), 123); checkpoint.phase = StalkerRefreshPhase::Vod { page: 4 }; checkpoint.quality_bypass_mask = LIVE_SELECTION; save_checkpoint(temp.path(), &checkpoint).await?; let resumed = load_or_start_checkpoint(temp.path(), 17, selection, SERIES_SELECTION).await?; assert_eq!(resumed.generation, 23); assert_eq!(resumed.phase, StalkerRefreshPhase::Vod { page: 4 }); assert_eq!(resumed.quality_bypass_mask, LIVE_SELECTION | SERIES_SELECTION); let resumed_without_new_force = load_or_start_checkpoint(temp.path(), 17, selection, 0).await?; assert_eq!(resumed_without_new_force.quality_bypass_mask, LIVE_SELECTION | SERIES_SELECTION); Ok(()) } #[tokio::test] async fn empty_complete_refresh_keeps_the_active_manifest() -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let mut active = tuliprox_repository::stalker_generation_repository::StalkerActiveManifest::empty(17); active.live = Some(tuliprox_repository::stalker_generation_repository::ClusterFiles { generation: 11, data: temp.path().join("old-live.db"), }); tuliprox_repository::stalker_generation_repository::save_active_manifest(temp.path(), &active).await?; let selection = StalkerClusterSelection { live: true, vod: false, series: false, epg: false }; let mut checkpoint = StalkerCheckpoint::new(17, 23, selection.mask(), 123); checkpoint.phase = StalkerRefreshPhase::Complete; save_checkpoint(temp.path(), &checkpoint).await?; let reports = finish_completed_refresh_checked( &app_config, temp.path(), 17, &checkpoint, ConfigInputUpdateQuality::default(), ) .await?; assert!(reports.error.is_some()); assert_eq!(load_active_manifest(temp.path(), 17).await?, active); assert!(load_checkpoint(temp.path(), 17).await?.is_none()); Ok(()) } #[tokio::test] async fn complete_publication_is_idempotent_after_restart() -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let mut checkpoint = StalkerCheckpoint::new(17, 23, 0b0011, 123); checkpoint.phase = StalkerRefreshPhase::Complete; save_checkpoint(temp.path(), &checkpoint).await?; finish_completed_refresh(&app_config, temp.path(), 17, &checkpoint, ConfigInputUpdateQuality::default()) .await?; save_checkpoint(temp.path(), &checkpoint).await?; finish_completed_refresh(&app_config, temp.path(), 17, &checkpoint, ConfigInputUpdateQuality::default()) .await?; let manifest = load_active_manifest(temp.path(), 17).await?; assert_eq!(manifest.live.as_ref().map(|files| files.generation), Some(23)); assert_eq!(manifest.vod.as_ref().map(|files| files.generation), Some(23)); assert!(load_checkpoint(temp.path(), 17).await?.is_none()); Ok(()) } #[tokio::test] async fn pipeline_transparency_stalker_preserves_accepted_quality_facts_after_publication( ) -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 17; let active_generation = 41; let candidate_generation = 42; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Live, 100, 0).await?; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Vod, 100, 0).await?; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Series, 40, 60).await?; let active = complete_manifest(temp.path(), identity, active_generation); save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Live, 90, 0).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Vod, 89, 0).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Series, 45, 65) .await?; let checkpoint = StalkerCheckpoint::new(identity, candidate_generation, 0b1111, 123); let reports = finish_completed_refresh_checked( &app_config, temp.path(), identity, &checkpoint, update_quality(90, 90, 90), ) .await?; assert!(reports.error.is_none()); assert_eq!( reports.quality_acceptances, vec![ ClusterUpdateAcceptance { cluster: XtreamCluster::Live, current_count: Some(100), candidate_count: 90, threshold: 90, quality: Some(90), }, ClusterUpdateAcceptance { cluster: XtreamCluster::Series, current_count: Some(100), candidate_count: 110, threshold: 90, quality: Some(90), }, ] ); assert_eq!( reports.quality_rejections, vec![ClusterUpdateRejection { cluster: XtreamCluster::Video, current_count: 100, candidate_count: 89, threshold: 90, quality: 89, }] ); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.live.as_ref().map(|files| files.generation), Some(candidate_generation)); assert_eq!(published.vod, active.vod); assert_eq!(published.series.as_ref().map(|files| files.generation), Some(candidate_generation)); assert_eq!(published.epg.as_ref().map(|files| files.generation), Some(candidate_generation)); Ok(()) } #[tokio::test] async fn pipeline_transparency_stalker_keeps_acceptance_when_pre_publish_validation_fails( ) -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 18; let active_generation = 51; let candidate_generation = 52; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Live, 100, 0).await?; let mut active = StalkerActiveManifest::empty(identity); active.live = Some(ClusterFiles { generation: active_generation, data: generation_data_path(temp.path(), active_generation, StalkerGenerationData::Live), }); save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Live, 90, 0).await?; let mut checkpoint = StalkerCheckpoint::new(identity, candidate_generation, LIVE_SELECTION, 123); checkpoint.phase = StalkerRefreshPhase::Complete; save_checkpoint(temp.path(), &checkpoint).await?; let reports = finish_completed_refresh_checked(&app_config, temp.path(), identity, &checkpoint, update_quality(90, 0, 0)) .await?; let StalkerRefreshOutcome::Failed { quality_acceptances, .. } = reports.into_outcome() else { return Err("expected failed Stalker completion with retained reports".into()); }; assert_eq!( quality_acceptances, vec![ClusterUpdateAcceptance { cluster: XtreamCluster::Live, current_count: Some(100), candidate_count: 90, threshold: 90, quality: Some(90), }] ); assert_eq!(load_active_manifest(temp.path(), identity).await?, active); assert!(load_checkpoint(temp.path(), identity).await?.is_none()); Ok(()) } #[tokio::test] async fn pipeline_transparency_stalker_keeps_acceptance_when_cleanup_fails_after_publish( ) -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 19; let active_generation = 61; let candidate_generation = 62; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Live, 100, 0).await?; let mut active = StalkerActiveManifest::empty(identity); active.live = Some(ClusterFiles { generation: active_generation, data: generation_data_path(temp.path(), active_generation, StalkerGenerationData::Live), }); save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Live, 90, 0).await?; tokio::fs::create_dir(generation_data_path(temp.path(), 1, StalkerGenerationData::Live)).await?; tokio::fs::write(generation_data_path(temp.path(), 2, StalkerGenerationData::Live), b"obsolete").await?; let mut checkpoint = StalkerCheckpoint::new(identity, candidate_generation, LIVE_SELECTION, 123); checkpoint.phase = StalkerRefreshPhase::Complete; checkpoint.processed = 90; save_checkpoint(temp.path(), &checkpoint).await?; let reports = finish_completed_refresh_checked(&app_config, temp.path(), identity, &checkpoint, update_quality(90, 0, 0)) .await?; let StalkerRefreshOutcome::Failed { quality_acceptances, .. } = reports.into_outcome() else { return Err("expected failed Stalker completion with retained reports".into()); }; assert_eq!(quality_acceptances.len(), 1); assert_eq!(quality_acceptances[0].cluster, XtreamCluster::Live); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.live.as_ref().map(|files| files.generation), Some(candidate_generation)); assert!(load_checkpoint(temp.path(), identity).await?.is_none()); Ok(()) } #[tokio::test] async fn rejected_live_generation_keeps_active_live_and_epg() -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 23; let active_generation = 7; let candidate_generation = 8; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Live, 100, 0).await?; let mut active = StalkerActiveManifest::empty(identity); active.live = Some(ClusterFiles { generation: active_generation, data: generation_data_path(temp.path(), active_generation, StalkerGenerationData::Live), }); active.epg = Some(ClusterFiles { generation: active_generation, data: generation_data_path(temp.path(), active_generation, StalkerGenerationData::Epg), }); save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Live, 89, 0).await?; let checkpoint = StalkerCheckpoint::new(identity, candidate_generation, LIVE_SELECTION | EPG_SELECTION, 123); let rejections = finish_completed_refresh(&app_config, temp.path(), identity, &checkpoint, update_quality(90, 0, 0)).await?; assert_eq!(rejections.len(), 1); assert_eq!(rejections[0].cluster, XtreamCluster::Live); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.live, active.live); assert_eq!(published.epg, active.epg); Ok(()) } #[tokio::test] async fn nonempty_bootstrap_generation_is_published() -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 29; let generation = 3; write_generation_cluster(&app_config, temp.path(), generation, StalkerCluster::Series, 2, 3).await?; let checkpoint = StalkerCheckpoint::new(identity, generation, SERIES_SELECTION, 123); let rejections = finish_completed_refresh(&app_config, temp.path(), identity, &checkpoint, update_quality(0, 0, 100)) .await?; assert!(rejections.is_empty()); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.series.as_ref().map(|files| files.generation), Some(generation)); Ok(()) } #[tokio::test] async fn empty_bootstrap_generation_is_rejected() -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 30; let generation = 4; write_generation_cluster(&app_config, temp.path(), generation, StalkerCluster::Live, 0, 0).await?; let checkpoint = StalkerCheckpoint::new(identity, generation, LIVE_SELECTION | EPG_SELECTION, 123); let rejections = finish_completed_refresh(&app_config, temp.path(), identity, &checkpoint, update_quality(90, 0, 0)).await?; assert_eq!( rejections, vec![ClusterUpdateRejection { cluster: XtreamCluster::Live, current_count: 0, candidate_count: 0, threshold: 90, quality: 0, }] ); let published = load_active_manifest(temp.path(), identity).await?; assert!(published.live.is_none()); assert!(published.epg.is_none()); Ok(()) } #[tokio::test] async fn forced_empty_stalker_generation_is_published_without_a_quality_rejection( ) -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 40; let active_generation = 5; let candidate_generation = 6; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Vod, 10, 0).await?; let mut active = complete_manifest(temp.path(), identity, active_generation); active.live = None; active.series = None; active.epg = None; save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Vod, 0, 0).await?; let mut checkpoint = StalkerCheckpoint::new(identity, candidate_generation, VOD_SELECTION, 123); checkpoint.phase = StalkerRefreshPhase::Complete; checkpoint.quality_bypass_mask = VOD_SELECTION; save_checkpoint(temp.path(), &checkpoint).await?; let reports = finish_completed_refresh_checked( &app_config, temp.path(), identity, &checkpoint, update_quality(0, 100, 0), ) .await?; assert!(reports.quality_rejections.is_empty()); assert_eq!( reports.force_updates, vec![ClusterForceUpdate { cluster: XtreamCluster::Video, candidate_count: 0, configured_threshold: 100 }] ); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.vod.as_ref().map(|files| files.generation), Some(candidate_generation)); assert!(load_checkpoint(temp.path(), identity).await?.is_none()); Ok(()) } #[tokio::test] async fn forced_live_generation_publishes_its_epg_with_the_same_generation( ) -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 41; let active_generation = 7; let candidate_generation = 8; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Live, 10, 0).await?; let mut active = StalkerActiveManifest::empty(identity); active.live = Some(ClusterFiles { generation: active_generation, data: generation_data_path(temp.path(), active_generation, StalkerGenerationData::Live), }); active.epg = Some(ClusterFiles { generation: active_generation, data: generation_data_path(temp.path(), active_generation, StalkerGenerationData::Epg), }); save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Live, 1, 0).await?; let mut checkpoint = StalkerCheckpoint::new(identity, candidate_generation, LIVE_SELECTION | EPG_SELECTION, 123); checkpoint.quality_bypass_mask = LIVE_SELECTION; let reports = finish_completed_refresh_checked( &app_config, temp.path(), identity, &checkpoint, update_quality(100, 0, 0), ) .await?; assert!(reports.quality_rejections.is_empty()); assert_eq!(reports.force_updates.len(), 1); assert_eq!(reports.force_updates[0].cluster, XtreamCluster::Live); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.live.as_ref().map(|files| files.generation), Some(candidate_generation)); assert_eq!(published.epg.as_ref().map(|files| files.generation), Some(candidate_generation)); Ok(()) } #[tokio::test] async fn rejected_series_counts_roots_and_episodes_without_blocking_accepted_live_generation( ) -> Result<(), Box> { let temp = tempfile::tempdir()?; let app_config = test_app_config(temp.path()); let identity = 31; let active_generation = 11; let candidate_generation = 12; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Live, 10, 0).await?; write_generation_cluster(&app_config, temp.path(), active_generation, StalkerCluster::Series, 2, 8).await?; let mut active = complete_manifest(temp.path(), identity, active_generation); active.vod = None; save_active_manifest(temp.path(), &active).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Live, 10, 0).await?; write_generation_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Series, 2, 7).await?; let checkpoint = StalkerCheckpoint::new(identity, candidate_generation, LIVE_SELECTION | SERIES_SELECTION, 123); assert_eq!(count_active_cluster(&app_config, &active, StalkerCluster::Series).await?, Some(10)); assert_eq!( count_candidate_cluster(&app_config, temp.path(), candidate_generation, StalkerCluster::Series).await?, 9 ); let rejections = finish_completed_refresh(&app_config, temp.path(), identity, &checkpoint, update_quality(100, 0, 100)) .await?; assert_eq!(rejections.len(), 1); assert_eq!(rejections[0].cluster, XtreamCluster::Series); assert_eq!(rejections[0].current_count, 10); assert_eq!(rejections[0].candidate_count, 9); let published = load_active_manifest(temp.path(), identity).await?; assert_eq!(published.live.as_ref().map(|files| files.generation), Some(candidate_generation)); assert_eq!(published.series, active.series); Ok(()) } #[test] fn unit_budget_yields_after_completed_unit() { let mut budget = StalkerRefreshBudget::units(1); assert!(!budget.should_yield()); budget.completed_unit(); assert!(budget.should_yield()); } #[test] fn parallel_mode_uses_a_bounded_work_unit_budget() { let mut budget = StalkerRefreshMode::Parallel.budget(); assert!(!budget.should_yield()); for _ in 0..64 { budget.completed_unit(); } assert!(budget.should_yield()); } #[test] fn phase_progression_skips_disabled_clusters() { let selection = StalkerClusterSelection { live: false, vod: true, series: false, epg: false }; assert_eq!(next_phase_after_live(selection), StalkerRefreshPhase::Vod { page: 1 }); assert_eq!(next_phase_after_vod(selection), StalkerRefreshPhase::Complete); } #[test] fn requested_clusters_limit_the_refresh_selection() { let input = ConfigInput::default(); let selection = StalkerClusterSelection::requested(&input, &[StalkerCluster::Series]); assert!(!selection.live); assert!(!selection.vod); assert!(selection.series); assert!(!selection.epg); } #[test] fn category_errors_are_not_silently_downgraded() { let result = category_map_result(Err(StalkerError::BodyDecode { message: "broken".to_string() })); assert!(result.is_err()); } #[test] fn map_items_uses_tv_genre_id_for_live_category() { let raw: StalkerRawItem = serde_json::from_value(serde_json::json!({ "id": "590", "name": "News", "tv_genre_id": "10" })) .expect("raw item"); let categories = HashMap::from([( 10, StalkerCategory { id: "10".to_string(), title: "News".to_string(), alias: None, number: 1 }, )]); let items = map_items(&[raw], &categories, StalkerStreamKind::Live, 0); assert_eq!(items.len(), 1); assert_eq!(items[0].category_id, 10); assert_eq!(items[0].category_name.as_ref(), "News"); } #[test] fn repeated_page_is_rejected_while_more_pages_are_advertised() { assert!(ensure_page_advanced(Some(3), Some(17), 17, 2, "vod").is_err()); assert!(ensure_page_advanced(None, Some(17), 17, 2, "vod").is_ok()); } }