Feature/stalker (#780)

New Features

- Added first-class Stalker/Ministra portal support, including live, VOD, series, playback-link refresh, and bulk EPG ingestion.
- Added Stalker configuration and batch/CSV support in the Web UI.
- Added playlist preview and playback handling for Stalker inputs, with support for partial refreshes.

backend/src/processing/processor/mod.rs

@@ -1,4 +1,6 @@
 mod playlist;
+mod stalker;
+mod stalker_refresh;
 mod xtream;
 // mod affix;
 mod xtream_vod;
@@ -11,6 +13,10 @@ mod stream_probe;
 mod probe_handle_guard;
 mod resolve_options;
 pub use self::playlist::*;
+pub(crate) use self::stalker::{
+    download_stalker_playlist, re_resolve_stalker_url, StalkerCluster,
+};
+pub(crate) use self::stalker_refresh::StalkerRefreshMode;
 pub use self::epg::*;
 pub use self::xtream::*;
 pub use self::xtream_vod::*;

backend/src/processing/processor/stalker.rs

@@ -0,0 +1,607 @@
+//! Stalker/Ministra playlist processor.
+//!
+//! Orchestrates Stalker authentication, resumable catalog refresh and B+Tree persistence. The
+//! processor is the single entry point the playlist dispatcher uses for `InputType::Stalker`
+//! — mirror of `xtream::download_xtream_playlist`.
+//!
+//! Reverse-proxy re-resolve (when a 4xx upstream error is observed) is implemented in the
+//! HLS/Xtream endpoints and reaches back into the API client via the helper
+//! [`crate::utils::network::stalker::client::StalkerApiClient::create_link`].
+
+#![allow(clippy::too_many_lines, clippy::needless_pass_by_value)]
+
+use log::{debug, info, warn};
+use parking_lot::Mutex;
+use lru::LruCache;
+use shared::error::TuliproxError;
+use shared::model::stalker::StalkerStreamKind;
+use shared::model::stalker_item::StalkerPlaylistItem;
+use shared::model::{PlaylistGroup, PlaylistItem};
+use shared::utils::Internable;
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::num::NonZeroUsize;
+use std::sync::{Arc, LazyLock, Weak};
+use std::time::{Duration, Instant};
+
+use crate::model::{AppConfig, ConfigInput, ConfigInputFlags, StalkerInputConfig};
+use crate::processing::parser::stalker as parser;
+use crate::repository::stalker_repository::{
+    ensure_stalker_storage_path, load_stalker_items_at, read_stalker_item_at,
+};
+use crate::repository::stalker_generation_repository::{load_active_manifest, load_checkpoint};
+use super::stalker_refresh::{
+    advance_stalker_refresh, StalkerClusterSelection, StalkerRefreshMode, StalkerRefreshOutcome,
+};
+use crate::utils::network::stalker::client::StalkerApiClient;
+use crate::utils::network::stalker::error::StalkerError;
+
+
+/// Cluster selector used by the Stalker processor. Mirrors the Xtream cluster split
+/// (Live/Video/Series) but uses Stalker-native kind names.
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum StalkerCluster {
+    Live,
+    Vod,
+    Series,
+}
+
+const DEFAULT_STALKER_CLUSTERS: [StalkerCluster; 3] =
+    [StalkerCluster::Live, StalkerCluster::Vod, StalkerCluster::Series];
+
+static RUNTIME_STALKER_CLIENTS: LazyLock<Mutex<LruCache<String, Arc<StalkerApiClient>>>> = LazyLock::new(|| {
+    Mutex::new(LruCache::new(NonZeroUsize::new(64).unwrap_or(NonZeroUsize::MIN)))
+});
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+struct RuntimeLinkKey {
+    fingerprint: u64,
+    generation: u64,
+    provider_id: u32,
+    kind: StalkerStreamKind,
+}
+
+struct RuntimeLink {
+    url: Arc<str>,
+    expires_at: Instant,
+}
+
+static RUNTIME_STALKER_LINKS: LazyLock<Mutex<LruCache<RuntimeLinkKey, RuntimeLink>>> = LazyLock::new(|| {
+    Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap_or(NonZeroUsize::MIN)))
+});
+
+static STALKER_REFRESH_LOCKS: LazyLock<tokio::sync::Mutex<HashMap<Arc<str>, Weak<tokio::sync::Semaphore>>>> =
+    LazyLock::new(|| tokio::sync::Mutex::new(HashMap::new()));
+
+async fn try_acquire_stalker_refresh(input_name: &Arc<str>) -> Option<tokio::sync::OwnedSemaphorePermit> {
+    let semaphore = {
+        let mut locks = STALKER_REFRESH_LOCKS.lock().await;
+        let semaphore = locks.get(input_name).and_then(Weak::upgrade).unwrap_or_else(|| {
+            let semaphore = Arc::new(tokio::sync::Semaphore::new(1));
+            locks.insert(Arc::clone(input_name), Arc::downgrade(&semaphore));
+            semaphore
+        });
+        locks.retain(|_, lock| lock.strong_count() > 0);
+        semaphore
+    };
+    semaphore.try_acquire_owned().ok()
+}
+
+fn cached_resolved_link(key: RuntimeLinkKey, force_refresh: bool) -> Option<Arc<str>> {
+    let mut cache = RUNTIME_STALKER_LINKS.lock();
+    let expired = cache.peek(&key).is_some_and(|entry| entry.expires_at <= Instant::now());
+    if force_refresh || expired {
+        cache.pop(&key);
+        return None;
+    }
+    cache.get(&key).map(|entry| Arc::clone(&entry.url))
+}
+
+fn cache_resolved_link(key: RuntimeLinkKey, url: Arc<str>) {
+    RUNTIME_STALKER_LINKS.lock().put(
+        key,
+        RuntimeLink { url, expires_at: Instant::now() + Duration::from_secs(45) },
+    );
+}
+
+/// Top-level orchestrator. Mirrors the `download_xtream_playlist` signature so the
+/// dispatcher can swap the call in place.
+#[allow(clippy::too_many_lines)]
+pub async fn download_stalker_playlist(
+    app_config: &Arc<AppConfig>,
+    client: &reqwest::Client,
+    input: &ConfigInput,
+    clusters: Option<&[StalkerCluster]>,
+    refresh_mode: StalkerRefreshMode,
+    materialize_active: bool,
+) -> (Vec<PlaylistGroup>, Vec<TuliproxError>, bool, bool) {
+    let stalker_cfg = match input.stalker.as_ref() {
+        Some(cfg) => cfg.clone(),
+        None => {
+            return (
+                vec![],
+                vec![TuliproxError::ConfigInput(format!(
+                    "Stalker input '{}' has no stalker configuration block",
+                    input.name
+                ))],
+                false,
+                false,
+            );
+        }
+    };
+
+    let skip_clusters = skip_clusters_for(input);
+    let resolved_clusters: Vec<StalkerCluster> = clusters
+        .map_or_else(
+            || DEFAULT_STALKER_CLUSTERS.to_vec(),
+            <[StalkerCluster]>::to_vec,
+        )
+        .into_iter()
+        .filter(|c| !skip_clusters.contains(c))
+        .collect();
+
+    if resolved_clusters.is_empty() {
+        info!("Stalker input '{}' has all clusters skipped", input.name);
+        return (vec![], vec![], false, false);
+    }
+    let refresh_selection = StalkerClusterSelection::requested(input, &resolved_clusters);
+
+    let portal_url = match resolve_stalker_portal_url(input) {
+        Ok(url) => url,
+        Err(err) => return (vec![], vec![err], false, false),
+    };
+
+    let identity_fingerprint = stalker_identity_fingerprint(&portal_url, &stalker_cfg);
+    let api_client = match cached_runtime_stalker_client(client, portal_url, &stalker_cfg) {
+        Ok(client) => client,
+        Err(err) => {
+            return (
+                vec![],
+                vec![TuliproxError::ConfigInput(format!(
+                    "failed to build Stalker client for input '{}': {err}",
+                    input.name
+                ))],
+                false,
+                false,
+            );
+        }
+    };
+
+    let storage_path = match ensure_stalker_storage_path(app_config, &input.name).await {
+        Ok(p) => p,
+        Err(err) => {
+            return (
+                vec![],
+                vec![TuliproxError::Io(format!(
+                    "could not prepare Stalker storage for input '{}': {err}",
+                    input.name
+                ))],
+                false,
+                false,
+            );
+        }
+    };
+
+    let outcome = if let Some(_refresh_permit) = try_acquire_stalker_refresh(&input.name).await {
+        let handshake = match api_client.handshake().await {
+            Ok(handshake) => handshake,
+            Err(err) => {
+                return (
+                    vec![],
+                    vec![TuliproxError::ProviderConnection(format!(
+                        "Stalker handshake for input '{}' failed: {err}",
+                        input.name
+                    ))],
+                    false,
+                    false,
+                );
+            }
+        };
+        loop {
+            let refresh = advance_stalker_refresh(
+                app_config,
+                api_client.as_ref(),
+                &handshake,
+                refresh_selection,
+                &storage_path,
+                identity_fingerprint,
+                refresh_mode.budget(),
+            );
+            let result = if refresh_mode == StalkerRefreshMode::ServerSlice {
+                if let Ok(result) = tokio::time::timeout(Duration::from_mins(45), refresh).await {
+                    result
+                } else {
+                    let checkpoint = match load_checkpoint(&storage_path, identity_fingerprint).await {
+                        Ok(checkpoint) => checkpoint,
+                        Err(err) => return (Vec::new(), vec![err], false, false),
+                    };
+                    break StalkerRefreshOutcome::Yielded {
+                        phase: checkpoint.as_ref().map_or(
+                            crate::repository::stalker_generation_repository::StalkerRefreshPhase::LiveBulk,
+                            |state| state.phase.clone(),
+                        ),
+                        processed: checkpoint.as_ref().map_or(0, |state| state.processed),
+                        skipped: checkpoint.as_ref().map_or(0, |state| state.skipped_count),
+                        error: None,
+                    };
+                }
+            } else {
+                refresh.await
+            };
+            match result {
+                Ok(StalkerRefreshOutcome::Yielded { error: None, .. })
+                    if refresh_mode == StalkerRefreshMode::Complete => {}
+                Ok(StalkerRefreshOutcome::Yielded { .. }) if refresh_mode == StalkerRefreshMode::Parallel => {
+                    tokio::task::yield_now().await;
+                }
+                Ok(outcome) => break outcome,
+                Err(err) => return (Vec::new(), vec![err], false, false),
+            }
+        }
+    } else {
+        return (
+            Vec::new(),
+            Vec::new(),
+            app_config.config.load().disk_based_processing,
+            true,
+        );
+    };
+
+    let yielded = matches!(&outcome, StalkerRefreshOutcome::Yielded { .. });
+    let terminal = matches!(&outcome, StalkerRefreshOutcome::Terminal(_));
+    let mut errors = Vec::new();
+    match outcome {
+        StalkerRefreshOutcome::Complete => {}
+        StalkerRefreshOutcome::Yielded { phase, processed, skipped, error } => {
+            info!(
+                "Stalker input '{}' yielded in phase {phase:?} after {processed} records ({skipped} skipped)",
+                input.name
+            );
+            if let Some(error) = error {
+                warn!("Stalker input '{}': resumable refresh paused after error: {error}", input.name);
+            }
+        }
+        StalkerRefreshOutcome::Terminal(error) => errors.push(error),
+    }
+
+    let manifest = match load_active_manifest(&storage_path, identity_fingerprint).await {
+        Ok(manifest) => manifest,
+        Err(err) => {
+            errors.push(err);
+            return (Vec::new(), errors, false, yielded);
+        }
+    };
+    if terminal {
+        if let Err(err) = crate::repository::stalker_generation_repository::cleanup_obsolete_generations(
+            &storage_path,
+            &manifest,
+        )
+        .await
+        {
+            errors.push(err);
+        }
+    }
+    let use_disk_based_processing = app_config.config.load().disk_based_processing;
+    if !materialize_active {
+        return (Vec::new(), errors, use_disk_based_processing, yielded);
+    }
+    let mut groups = Vec::new();
+    let mut counts = [0_usize; 3];
+    for cluster in resolved_clusters {
+        let items_result = match cluster {
+            StalkerCluster::Live => match manifest.live.as_ref() {
+                Some(files) => load_stalker_items_at(app_config, &files.data).await,
+                None => Ok(Vec::new()),
+            },
+            StalkerCluster::Vod => match manifest.vod.as_ref() {
+                Some(files) => load_stalker_items_at(app_config, &files.data).await,
+                None => Ok(Vec::new()),
+            },
+            StalkerCluster::Series => match manifest.series.as_ref() {
+                Some(files) => {
+                    match (
+                        load_stalker_items_at(app_config, &files.roots).await,
+                        load_stalker_items_at(app_config, &files.episodes).await,
+                    ) {
+                        (Ok(mut roots), Ok(episodes)) => {
+                            roots.extend(episodes);
+                            Ok(roots)
+                        }
+                        (Err(err), _) | (_, Err(err)) => Err(err),
+                    }
+                }
+                None => Ok(Vec::new()),
+            },
+        };
+        match items_result {
+            Ok(items) => {
+                counts[cluster as usize] = items.len();
+                let cluster_groups = groups_for_cluster(items, cluster, &input.name);
+                groups.extend(cluster_groups);
+            }
+            Err(err) => errors.push(err),
+        }
+    }
+    parser::log_stalker_download_summary(&input.name, counts[0], counts[1], counts[2]);
+    (groups, errors, use_disk_based_processing, yielded)
+}
+
+/// Resolve a per-input portal URL using the standard `provider://` resolution pipeline.
+fn resolve_stalker_portal_url(input: &ConfigInput) -> Result<String, TuliproxError> {
+    let url = input.url.as_str();
+    if url.trim().is_empty() {
+        return Err(TuliproxError::ConfigInput(format!(
+            "Stalker input '{}' has no URL configured",
+            input.name
+        )));
+    }
+    input.resolve_url(url).map(Cow::into_owned)
+}
+
+fn skip_clusters_for(input: &ConfigInput) -> Vec<StalkerCluster> {
+    let mut out = Vec::new();
+    if input.has_flag(ConfigInputFlags::SkipLive) {
+        out.push(StalkerCluster::Live);
+    }
+    if input.has_flag(ConfigInputFlags::SkipVod) {
+        out.push(StalkerCluster::Vod);
+    }
+    if input.has_flag(ConfigInputFlags::SkipSeries) {
+        out.push(StalkerCluster::Series);
+    }
+    out
+}
+
+/// Convert the items of one cluster into per-category `PlaylistGroup`s,
+/// mirroring the Xtream pipeline and the disk-based source: items are grouped
+/// by their portal category id (falling back to a cluster default title for
+/// items without a category) instead of collapsing the whole cluster into a
+/// single group.
+fn groups_for_cluster(
+    items: Vec<StalkerPlaylistItem>,
+    cluster: StalkerCluster,
+    input_name: &str,
+) -> Vec<PlaylistGroup> {
+    use shared::model::XtreamCluster;
+    let xtream_cluster = match cluster {
+        StalkerCluster::Live => XtreamCluster::Live,
+        StalkerCluster::Vod => XtreamCluster::Video,
+        StalkerCluster::Series => XtreamCluster::Series,
+    };
+    let mut groups_map: indexmap::IndexMap<u32, PlaylistGroup> = indexmap::IndexMap::new();
+    for item in items {
+        let category_id = item.category_id;
+        let pli = PlaylistItem::from_stalker(&item, input_name);
+        let group = groups_map.entry(category_id).or_insert_with(|| PlaylistGroup {
+            id: category_id,
+            title: Arc::clone(&pli.header.group),
+            channels: Vec::new(),
+            xtream_cluster,
+        });
+        group.channels.push(pli);
+    }
+    groups_map.into_values().collect()
+}
+
+fn stalker_err_to_repo(err: StalkerError) -> TuliproxError {
+    TuliproxError::ProviderConnection(format!("Stalker client error: {err}"))
+}
+
+/// Cache key for the runtime client map. Built from explicit, non-secret
+/// fields — `StalkerInputConfig` carries the portal account credentials, so
+/// formatting the whole config with `{:?}` would leak username and password
+/// into a long-lived in-memory map key. Credentials still differentiate the
+/// key, but only as a non-reversible short hash.
+fn runtime_client_cache_key(portal_url: &str, cfg: &StalkerInputConfig) -> String {
+    format!("{portal_url}|{:016x}", cfg.identity_fingerprint(portal_url))
+}
+
+pub(crate) fn stalker_identity_fingerprint(portal_url: &str, cfg: &StalkerInputConfig) -> u64 {
+    cfg.identity_fingerprint(portal_url)
+}
+
+fn cached_runtime_stalker_client(
+    http_client: &reqwest::Client,
+    portal_url: String,
+    cfg: &StalkerInputConfig,
+) -> Result<Arc<StalkerApiClient>, TuliproxError> {
+    let key = runtime_client_cache_key(&portal_url, cfg);
+    if let Some(client) = RUNTIME_STALKER_CLIENTS.lock().get(&key).cloned() {
+        return Ok(client);
+    }
+
+    let client = Arc::new(
+        StalkerApiClient::new(http_client.clone(), portal_url, cfg.clone()).map_err(stalker_err_to_repo)?,
+    );
+    RUNTIME_STALKER_CLIENTS.lock().put(key, Arc::clone(&client));
+    Ok(client)
+}
+
+pub async fn re_resolve_stalker_url(
+    app_config: &Arc<AppConfig>,
+    http_client: &reqwest::Client,
+    input: &ConfigInput,
+    provider_id: u32,
+    kind: StalkerStreamKind,
+    force_refresh: bool,
+) -> Result<Option<Arc<str>>, TuliproxError> {
+    if provider_id == 0 {
+        return Ok(None);
+    }
+    let stalker_cfg = input.stalker.as_ref().ok_or_else(|| {
+        TuliproxError::ConfigInput(format!("Stalker input '{}' has no stalker configuration block", input.name))
+    })?;
+    let portal_url = resolve_stalker_portal_url(input)?;
+    let identity_fingerprint = stalker_cfg.identity_fingerprint(&portal_url);
+    let storage_path = ensure_stalker_storage_path(app_config, &input.name).await?;
+    let manifest = load_active_manifest(&storage_path, identity_fingerprint).await?;
+    let generation_and_path = match kind {
+        StalkerStreamKind::Live | StalkerStreamKind::Archive => {
+            manifest.live.as_ref().map(|files| (files.generation, &files.data))
+        }
+        StalkerStreamKind::Movie => manifest.vod.as_ref().map(|files| (files.generation, &files.data)),
+        StalkerStreamKind::Episode => manifest.series.as_ref().map(|files| (files.generation, &files.episodes)),
+    };
+    let Some((generation, item_path)) = generation_and_path else { return Ok(None) };
+    let link_key = RuntimeLinkKey { fingerprint: identity_fingerprint, generation, provider_id, kind };
+    if let Some(url) = cached_resolved_link(link_key, force_refresh) {
+        return Ok(Some(url));
+    }
+    let Some(item) = read_stalker_item_at(app_config, item_path, provider_id).await? else {
+        return Ok(None);
+    };
+    let Some(descriptor) = item.playback_descriptor.as_ref() else {
+                 debug!("Stalker re-resolve skipped: stream_id={} has no playback_descriptor", item.stream_id);
+                 return Ok(None);
+             };
+    if descriptor.candidates.is_empty() {
+        debug!("Stalker re-resolve skipped: stream_id={} descriptor has no candidate", item.stream_id);
+        return Ok(None);
+    }
+    if descriptor
+        .candidates
+        .iter()
+        .all(|candidate| candidate.cmd.trim().is_empty())
+    {
+        debug!(
+            "Stalker re-resolve skipped: stream_id={} descriptor candidates are empty",
+            item.stream_id
+        );
+        return Ok(None);
+    }
+    let api_client = cached_runtime_stalker_client(
+        http_client,
+        portal_url,
+        stalker_cfg,
+    )?;
+    let handshake = api_client.handshake().await.map_err(stalker_err_to_repo)?;
+    let series_number = (kind == StalkerStreamKind::Episode).then_some(item.number);
+
+    for candidate in &descriptor.candidates {
+        if candidate.cmd.trim().is_empty() {
+            continue;
+        }
+        match api_client
+            .create_link(
+                &handshake,
+                kind,
+                candidate.playback_mode,
+                &candidate.cmd,
+                series_number,
+                None,
+                None,
+            )
+            .await
+        {
+            Ok(resolved) => {
+                let url = Internable::intern(resolved.stream_url);
+                cache_resolved_link(link_key, Arc::clone(&url));
+                return Ok(Some(url));
+            }
+            Err(err) => {
+                debug!(
+                    "Stalker runtime re-resolve candidate failed for stream_id={}, mode={:?}: {err}",
+                    item.stream_id, candidate.playback_mode
+                );
+            }
+        }
+    }
+
+    warn!(
+        "Stalker runtime re-resolve failed for stream_id={}, invalidating stale stream_url",
+        item.stream_id
+    );
+    RUNTIME_STALKER_LINKS.lock().pop(&link_key);
+    Ok(None)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn runtime_cfg() -> StalkerInputConfig {
+        StalkerInputConfig::default()
+    }
+
+    #[test]
+    fn runtime_client_cache_key_changes_with_endpoint_preference() {
+        let mut cfg = runtime_cfg();
+        let key_a = runtime_client_cache_key("http://portal.example", &cfg);
+        cfg.endpoint_preference = shared::model::stalker::StalkerEndpointPreference::Portal;
+        let key_b = runtime_client_cache_key("http://portal.example", &cfg);
+        assert_ne!(key_a, key_b);
+    }
+
+    #[test]
+    fn runtime_client_cache_key_changes_with_size_caps() {
+        let mut cfg = runtime_cfg();
+        let key_a = runtime_client_cache_key("http://portal.example", &cfg);
+        cfg.size_caps = Some(crate::model::StalkerSizeCaps {
+            create_link_kb: 128,
+            ordered_list_mb: 8,
+            get_epg_mb: 64,
+        });
+        let key_b = runtime_client_cache_key("http://portal.example", &cfg);
+        assert_ne!(key_a, key_b);
+    }
+
+    #[test]
+    fn runtime_client_cache_key_does_not_leak_credentials() {
+        let mut cfg = runtime_cfg();
+        cfg.username = Some("secret_user".to_string());
+        cfg.password = Some("secret_pass".to_string());
+        let key = runtime_client_cache_key("http://portal.example", &cfg);
+        assert!(!key.contains("secret_user"));
+        assert!(!key.contains("secret_pass"));
+        // Different credentials must still produce a different cache key so two
+        // inputs against the same portal never share a client.
+        let mut other = cfg.clone();
+        other.password = Some("other_pass".to_string());
+        let other_key = runtime_client_cache_key("http://portal.example", &other);
+        assert_ne!(key, other_key);
+    }
+
+    #[test]
+    fn runtime_client_cache_keeps_session_client_alive() -> Result<(), TuliproxError> {
+        let http = reqwest::Client::new();
+        let cfg = runtime_cfg();
+        let first = cached_runtime_stalker_client(&http, "http://portal.example".to_string(), &cfg)?;
+        let weak = Arc::downgrade(&first);
+        drop(first);
+
+        let second = cached_runtime_stalker_client(&http, "http://portal.example".to_string(), &cfg)?;
+
+        assert!(weak.upgrade().is_some());
+        let upgraded = weak
+            .upgrade()
+            .ok_or_else(|| TuliproxError::ProviderConnection("client dropped".to_string()))?;
+        assert!(Arc::ptr_eq(&upgraded, &second));
+        Ok(())
+    }
+
+    #[test]
+    fn resolved_link_cache_can_be_forced_stale() {
+        let key = RuntimeLinkKey { fingerprint: 8, generation: 1, provider_id: 43, kind: StalkerStreamKind::Live };
+        cache_resolved_link(key, "http://stream.example/live".into());
+        assert_eq!(cached_resolved_link(key, false).as_deref(), Some("http://stream.example/live"));
+        assert!(cached_resolved_link(key, true).is_none());
+    }
+
+    #[test]
+    fn resolved_link_cache_is_scoped_to_the_published_generation() {
+        let old = RuntimeLinkKey { fingerprint: 7, generation: 1, provider_id: 42, kind: StalkerStreamKind::Live };
+        let current = RuntimeLinkKey { generation: 2, ..old };
+        cache_resolved_link(old, "http://stream.example/old".into());
+        assert!(cached_resolved_link(current, false).is_none());
+    }
+
+    #[tokio::test]
+    async fn refresh_ownership_is_non_blocking_across_invocations() {
+        let name: Arc<str> = "portal-a".into();
+        let first = try_acquire_stalker_refresh(&name).await;
+        assert!(first.is_some());
+        assert!(try_acquire_stalker_refresh(&name).await.is_none());
+        drop(first);
+        assert!(try_acquire_stalker_refresh(&name).await.is_some());
+    }
+}

backend/src/processing/processor/stalker_refresh.rs

@@ -0,0 +1,687 @@
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
+
+use shared::error::TuliproxError;
+use shared::model::stalker::StalkerStreamKind;
+use shared::model::stalker_item::StalkerPlaylistItem;
+
+use crate::model::{AppConfig, ConfigInput, ConfigInputFlags};
+use crate::processing::parser::stalker as parser;
+use crate::repository::stalker_generation_repository::{
+    cleanup_obsolete_generations, clear_checkpoint, generation_data_path, load_checkpoint,
+    publish_selection, save_checkpoint, StalkerCheckpoint, StalkerGenerationData, StalkerRefreshPhase,
+};
+use crate::repository::stalker_repository::{
+    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,
+};
+use crate::utils::network::stalker::catalog::{StalkerCategory, StalkerRawItem};
+use crate::utils::network::stalker::client::StalkerApiClient;
+use crate::utils::network::stalker::error::StalkerError;
+use crate::utils::network::stalker::profile::StalkerHandshake;
+use super::stalker::StalkerCluster;
+
+const MAX_RETRIES: u8 = 3;
+const SKIPPED_SAMPLE_LIMIT: usize = 32;
+
+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)
+    }
+}
+
+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,
+    Yielded {
+        phase: StalkerRefreshPhase,
+        processed: u64,
+        skipped: u64,
+        error: Option<TuliproxError>,
+    },
+    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,
+) -> Result<StalkerCheckpoint, TuliproxError> {
+    if let Some(state) = load_checkpoint(storage_path, identity_fingerprint).await? {
+        if state.selection_mask == selection.mask() {
+            return Ok(state);
+        }
+    }
+
+    let mut state = StalkerCheckpoint::new(
+        identity_fingerprint,
+        generation_id(),
+        selection.mask(),
+        chrono::Utc::now().timestamp(),
+    );
+    state.phase = first_phase(selection);
+    save_checkpoint(storage_path, &state).await?;
+    Ok(state)
+}
+
+async fn finish_completed_refresh(
+    storage_path: &Path,
+    identity_fingerprint: u64,
+    checkpoint: &StalkerCheckpoint,
+) -> Result<(), TuliproxError> {
+    let manifest = publish_selection(
+        storage_path,
+        identity_fingerprint,
+        checkpoint.generation,
+        checkpoint.selection_mask,
+    )
+    .await?;
+    clear_checkpoint(storage_path).await?;
+    cleanup_obsolete_generations(storage_path, &manifest).await
+}
+
+fn category_map(categories: Vec<StalkerCategory>) -> HashMap<u32, StalkerCategory> {
+    categories
+        .into_iter()
+        .filter_map(|category| category.id.parse().ok().map(|id| (id, category)))
+        .collect()
+}
+
+fn category_map_result(
+    result: Result<Vec<StalkerCategory>, StalkerError>,
+) -> Result<HashMap<u32, StalkerCategory>, TuliproxError> {
+    result.map(category_map).map_err(|err| provider_error(&err))
+}
+
+fn map_items(
+    raw_items: &[StalkerRawItem],
+    categories: &HashMap<u32, StalkerCategory>,
+    kind: StalkerStreamKind,
+    added_at: i64,
+) -> Vec<StalkerPlaylistItem> {
+    raw_items
+        .iter()
+        .map(|raw| {
+            let category = raw
+                .category_id()
+                .and_then(|value| value.parse::<u32>().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<Item = Option<&'a str>>) -> 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<u32>,
+    previous_signature: Option<u64>,
+    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<StalkerRefreshOutcome, TuliproxError> {
+    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<StalkerRefreshOutcome, TuliproxError> {
+    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<AppConfig>,
+    api_client: &StalkerApiClient,
+    handshake: &StalkerHandshake,
+    selection: StalkerClusterSelection,
+    storage_path: &Path,
+    identity_fingerprint: u64,
+    mut budget: StalkerRefreshBudget,
+) -> Result<StalkerRefreshOutcome, TuliproxError> {
+    let mut checkpoint = load_or_start_checkpoint(storage_path, identity_fingerprint, selection).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::<u32>().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 => {
+                finish_completed_refresh(storage_path, identity_fingerprint, &checkpoint).await?;
+                return Ok(StalkerRefreshOutcome::Complete);
+            }
+            StalkerRefreshPhase::Terminal => {
+                return finish_terminal_refresh(storage_path).await;
+            }
+        }
+        save_checkpoint(storage_path, &checkpoint).await?;
+        budget.completed_unit();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn terminal_checkpoint_is_cleared_after_restart() -> Result<(), Box<dyn std::error::Error>> {
+        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<dyn std::error::Error>> {
+        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).await?;
+
+        assert_eq!(resumed.generation, 23);
+        assert_eq!(resumed.phase, StalkerRefreshPhase::Complete);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn complete_publication_is_idempotent_after_restart() -> Result<(), Box<dyn std::error::Error>> {
+        let temp = tempfile::tempdir()?;
+        let mut checkpoint = StalkerCheckpoint::new(17, 23, 0b0011, 123);
+        checkpoint.phase = StalkerRefreshPhase::Complete;
+        save_checkpoint(temp.path(), &checkpoint).await?;
+
+        finish_completed_refresh(temp.path(), 17, &checkpoint).await?;
+        save_checkpoint(temp.path(), &checkpoint).await?;
+        finish_completed_refresh(temp.path(), 17, &checkpoint).await?;
+
+        let manifest = crate::repository::stalker_generation_repository::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(())
+    }
+
+    #[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());
+    }
+}

backend/src/processing/processor/stream_probe.rs

@@ -190,7 +190,8 @@ async fn prepare_generic_stream_metadata(
                 ProbeStorageKind::Xtream,
             )
         }
-        InputType::Emby | InputType::Jellyfin | InputType::Plex | InputType::Staged => return Ok(PreparedGenericProbeOutcome::Noop),
+        InputType::Emby | InputType::Jellyfin | InputType::Plex
+        | InputType::Stalker | InputType::StalkerBatch | InputType::Staged => return Ok(PreparedGenericProbeOutcome::Noop),
     };
 
     if !db_path.exists() {

backend/src/processing/processor/xtream_series.rs

@@ -70,7 +70,7 @@ pub async fn playlist_resolve_series(
     processed_fpl: &mut FetchedPlaylist<'_>,
 ) {
     // Skip-flag is the kill-switch: no resolve, no probe, no iteration.
-    if processed_fpl.input.has_flag(ConfigInputFlags::XtreamSkipSeries) {
+    if processed_fpl.input.has_flag(ConfigInputFlags::SkipSeries) {
         return;
     }
 
@@ -87,7 +87,9 @@ pub async fn playlist_resolve_series(
         sync_resolved_series_properties(provider_fpl, processed_fpl);
     }
 
-    provider_fpl.source.obtain_resources().await;
+    if let Err(err) = provider_fpl.source.obtain_resources().await {
+        errors.push(err);
+    }
 }
 
 #[allow(clippy::too_many_lines, clippy::too_many_arguments)]
@@ -690,7 +692,7 @@ pub async fn update_series_metadata(
         .await
         .map_err(|e| TuliproxError::Io(format!("Storage path error: {e}")))?;
 
-    if input.has_flag(ConfigInputFlags::XtreamSkipSeries) {
+    if input.has_flag(ConfigInputFlags::SkipSeries) {
         return Ok(None);
     }
 

backend/src/processing/processor/xtream_vod.rs

@@ -49,7 +49,7 @@ pub async fn playlist_resolve_vod(
     fpl: &mut FetchedPlaylist<'_>,
 ) {
     // Skip-flag is the kill-switch: no resolve, no probe, no iteration.
-    if fpl.input.has_flag(ConfigInputFlags::XtreamSkipVod) {
+    if fpl.input.has_flag(ConfigInputFlags::SkipVod) {
         return;
     }
 
@@ -74,7 +74,9 @@ pub async fn playlist_resolve_vod(
         sync_resolved_vod_properties(provider_fpl, fpl);
     }
 
-    provider_fpl.source.obtain_resources().await;
+    if let Err(err) = provider_fpl.source.obtain_resources().await {
+        errors.push(err);
+    }
 }
 
 fn sync_resolved_vod_properties(provider_fpl: &mut FetchedPlaylist<'_>, processed_fpl: &mut FetchedPlaylist<'_>) {
@@ -582,7 +584,7 @@ pub async fn update_vod_metadata(
     let storage_path = resolve_input_storage_path(storage_dir, &input.name).await;
 
     // Check if we should skip based on input options
-    if input.has_flag(ConfigInputFlags::XtreamSkipVod) {
+    if input.has_flag(ConfigInputFlags::SkipVod) {
         return Ok(None);
     }