tuliprox/backend/src/api/endpoints/xmltv_api.rs

use crate::{
    api::{
        api_utils::{
            create_api_proxy_user, empty_json_response_as_array, get_user_target,
            get_user_target_by_credentials, internal_server_error,
            resource_response,
            stream_json_or_bin_response_stream, try_unwrap_body,
        },
        model::{AppState, UserApiRequest, UserApiRequestQueryOrBody},
    },
    auth::Fingerprint,
    model::{Config, ConfigTarget, ProxyUserCredentials, EPG_ATTRIB_ID, EPG_TAG_CHANNEL},
    repository::{
        epg_query_channels, get_target_storage_path, m3u_get_epg_file_path_for_target, storage_const,
        xtream_get_epg_file_path_for_target, xtream_get_storage_path, BPlusTreeQuery, LockedReceiverStream, XML_PREAMBLE,
    },
    utils,
    utils::{
        deobscure_text, file_exists_async, format_xmltv_time_utc, get_epg_processing_options, obscure_text,
        EpgProcessingOptions, EpgTimeShift,
    },
};
use axum::response::IntoResponse;
use chrono::{DateTime, TimeZone};
use log::{error, trace};
use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
use shared::{
    concat_string,
    model::{
        EpgChannel, EpgProgramme, EpgProgrammeDto, ShortEpgDto, ShortEpgResultDto, StreamEpgEntry, StreamEpgItemRequest,
        StreamEpgRequest, StreamEpgResponse, TargetType,
    },
    utils::{concat_path, concat_path_leading_slash, obfuscate_text, Internable},
};
use std::{
    collections::{HashMap, HashSet},
    path::{Path, PathBuf},
    sync::Arc,
};
use tokio::{io::AsyncWriteExt, sync::mpsc, task};
use tokio_stream::StreamExt;
use tokio_util::io::ReaderStream;
use crate::auth::resolve_api_user_context;
use crate::model::ApiProxyServerInfo;

pub fn get_empty_epg_response() -> axum::response::Response {
    try_unwrap_body!(axum::response::Response::builder()
        .status(axum::http::StatusCode::OK)
        .header(axum::http::header::CONTENT_TYPE, axum::http::HeaderValue::from_static("text/xml"))
        .body(axum::body::Body::from(r#"<?xml version="1.0" encoding="utf-8" ?><!DOCTYPE tv SYSTEM "xmltv.dtd"><tv generator-info-name="Xtream Codes" generator-info-url=""></tv>"#)))
}

fn get_epg_path_for_target_of_type(target_name: &str, epg_path: PathBuf) -> Option<PathBuf> {
    if utils::path_exists(&epg_path) {
        return Some(epg_path);
    }
    trace!("Can't find epg file for {target_name} target: {}", epg_path.to_str().unwrap_or("?"));
    None
}

pub(in crate::api) fn get_epg_path_for_target(config: &Config, target: &ConfigTarget) -> Option<PathBuf> {
    // TODO if we have multiple targets, first one serves, this can be problematic when
    // we use m3u playlist but serve xtream target epg

    // TODO if we share the same virtual_id for epg, can we store an epg file for the target ?
    for output in &target.output {
        if let Some(epg_path) = get_epg_path_for_target_by_type(config, target, TargetType::from(output)) {
            return Some(epg_path);
        }
    }
    None
}

pub(in crate::api) fn get_epg_path_for_target_by_type(
    config: &Config,
    target: &ConfigTarget,
    output_type: TargetType,
) -> Option<PathBuf> {
    match output_type {
        TargetType::Xtream => {
            if let Some(storage_path) = xtream_get_storage_path(config, &target.name) {
                return get_epg_path_for_target_of_type(
                    &target.name,
                    xtream_get_epg_file_path_for_target(&storage_path),
                );
            }
            None
        }
        TargetType::M3u => {
            if let Some(target_path) = get_target_storage_path(config, &target.name) {
                return get_epg_path_for_target_of_type(
                    &target.name,
                    m3u_get_epg_file_path_for_target(&target_path),
                );
            }
            None
        }
        TargetType::Strm | TargetType::HdHomeRun => None,
    }
}

pub async fn serve_epg(
    app_state: &Arc<AppState>,
    epg_path: &Path,
    user: &ProxyUserCredentials,
    target: &Arc<ConfigTarget>,
    limit: Option<u32>,
) -> axum::response::Response {
    if file_exists_async(epg_path).await {
        serve_epg_with_rewrites(app_state, epg_path, user, target, limit).await
    } else {
        get_empty_epg_response()
    }
}

pub async fn serve_epg_web_ui(
    app_state: &Arc<AppState>,
    accept: Option<&str>,
    epg_path: &Path,
    target: &Arc<ConfigTarget>,
) -> axum::response::Response {
    if file_exists_async(epg_path).await {
        let config = app_state.app_config.config.load();
        let web_ui_path = config.web_ui.as_ref().and_then(|w| w.path.as_ref()).map_or("", String::as_str);
        let resource_url = concat_path_leading_slash(web_ui_path, "api/v1/playlist/resource");
        let encrypt_secret = app_state.get_encrypt_secret();
        let iter_lock = app_state.app_config.file_locks.read_lock(epg_path).await;
        let bg_lock = app_state.app_config.file_locks.read_lock(epg_path).await;
        let epg_path = epg_path.to_path_buf();
        let target_name = target.name.clone();
        let (tx, rx) = mpsc::channel::<EpgChannel>(64);

        let epg_path_for_log = epg_path.clone();
        let target_name_for_log = target_name.clone();
        let handle = task::spawn_blocking(move || {
            let _guard = bg_lock;
            let Ok(query) = BPlusTreeQuery::<Arc<str>, EpgChannel>::try_new(&epg_path) else {
                error!("Failed to open epg db for target {} {}", target_name, epg_path.display());
                return;
            };
            for (_, channel) in query.disk_iter() {
                if tx.blocking_send(channel).is_err() {
                    break;
                }
            }
        });
        tokio::spawn(async move {
            if let Err(err) = handle.await {
                error!(
                    "EPG web UI producer task failed for target {} {}: {err}",
                    target_name_for_log,
                    epg_path_for_log.display()
                );
            }
        });

        let stream = LockedReceiverStream::new(rx, iter_lock)
            .map(move |channel| rewrite_epg_channel_resource_url(&encrypt_secret, &resource_url, channel));
        return stream_json_or_bin_response_stream(accept, stream);
    }
    try_unwrap_body!(empty_json_response_as_array())
}

pub fn rewrite_epg_channel_resource_url(encrypt_secret: &[u8; 16], resource_url: &str, mut channel: EpgChannel) -> EpgChannel {
    let Some(icon) = channel.icon.as_ref() else {
        return channel;
    };
    if icon.is_empty() || icon.starts_with('/') {
        return channel;
    }
    channel.icon = Some(concat_path(resource_url, &obfuscate_text(encrypt_secret, icon)).intern());
    channel
}

macro_rules! continue_on_err {
    ($expr:expr) => {
        if let Err(_err) = $expr {
            continue;
        }
    };
}

#[allow(clippy::too_many_lines)]
async fn serve_epg_with_rewrites(
    app_state: &Arc<AppState>,
    epg_path: &Path,
    user: &ProxyUserCredentials,
    target: &Arc<ConfigTarget>,
    limit: Option<u32>,
) -> axum::response::Response {
    if !file_exists_async(epg_path).await {
        return get_empty_epg_response();
    }

    let epg_processing_options = get_epg_processing_options(app_state, user, target);

    let server_info = app_state.app_config.get_user_server_info(user);
    let base_url =
        if !matches!(epg_processing_options.time_shift, EpgTimeShift::None) || epg_processing_options.rewrite_urls {
            server_info.as_ref().map(|si| concat_string!(
                 &si.get_base_url(),
                 "/",
                 storage_const::EPG_RESOURCE_PATH,
                 "/",
                 &user.username,
                 "/",
                 &user.password
             ))
        } else {
            None
        };

    let generator_info = server_info.as_ref().map(ApiProxyServerInfo::get_base_url).unwrap_or_default();

    let limit = limit.unwrap_or_default();

    let bg_lock = app_state.app_config.file_locks.read_lock(epg_path).await;
    let epg_path = epg_path.to_path_buf();
    let (channel_tx, mut channel_rx) = mpsc::channel::<EpgChannel>(256);

    let epg_path_for_log = epg_path.clone();
    let spawn_handle = task::spawn_blocking(move || {
        let _guard = bg_lock;
        let Ok(mut query) = BPlusTreeQuery::<Arc<str>, EpgChannel>::try_new(&epg_path) else {
            error!("Failed to open BPlusTreeQuery {}", epg_path.display());
            return;
        };

        for (_, channel) in query.iter() {
            if channel_tx.blocking_send(channel).is_err() {
                break;
            }
        }
    });
    tokio::spawn(async move {
        if let Err(err) = spawn_handle.await {
            error!("EPG rewrite producer task failed for {}: {err}", epg_path_for_log.display());
        }
    });

    let (mut tx, rx) = tokio::io::duplex(8192);
    tokio::spawn(async move {
        // Work-Around BytesText DocType escape, see below
        if let Err(err) = tx.write_all(XML_PREAMBLE.as_ref()).await {
            error!("EPG: Failed to write xml header {err}");
            return;
        }
        if let Err(err) = tx.write_all(format!(r#"<tv generator-info-name="X" generator-info-url="{generator_info}">"#).as_bytes()).await
        {
            error!("EPG: Failed to write xml tv header {err}");
            return;
        }

        let mut writer = quick_xml::writer::Writer::new(tx);
        while let Some(channel) = channel_rx.recv().await {
            let programmes = if limit > 0 {
                channel.get_programme_with_limit(limit)
            } else {
                channel.programmes.iter().collect::<Vec<&EpgProgramme>>()
            };

            if !programmes.is_empty() {
                let mut elem = BytesStart::new(EPG_TAG_CHANNEL);
                elem.push_attribute((EPG_ATTRIB_ID, channel.id.as_ref()));
                continue_on_err!(writer.write_event_async(Event::Start(elem)).await);

                let elem = BytesStart::new("display-name");
                continue_on_err!(writer.write_event_async(Event::Start(elem)).await);
                let title: &str = channel.title.as_deref().unwrap_or("");
                continue_on_err!(writer.write_event_async(Event::Text(BytesText::new(title))).await);

                let elem = BytesEnd::new("display-name");
                continue_on_err!(writer.write_event_async(Event::End(elem)).await);

                if let Some(icon_url) = &channel.icon {
                    let icon = match (
                        epg_processing_options.rewrite_urls,
                        base_url.as_ref(),
                        obscure_text(&epg_processing_options.encrypt_secret, icon_url),
                    ) {
                        (true, Some(base), Ok(enc)) => concat_string!(base, "/", &enc),
                        _ => icon_url.to_string(),
                    };

                    let mut elem = BytesStart::new("icon");
                    elem.push_attribute(("src", icon.as_ref()));
                    if (writer.write_event_async(Event::Empty(elem)).await).is_err() {
                        // ignore
                    }
                }

                let elem = BytesEnd::new(EPG_TAG_CHANNEL);
                continue_on_err!(writer.write_event_async(Event::End(elem)).await);

                for programme in programmes {
                    let mut elem = BytesStart::new("programme");
                    let (user_start, user_stop) = (programme.start, programme.stop);
                    elem.push_attribute((
                        "start",
                        format_xmltv_time_utc(user_start, &epg_processing_options.time_shift).as_str(),
                    ));
                    elem.push_attribute((
                        "stop",
                        format_xmltv_time_utc(user_stop, &epg_processing_options.time_shift).as_str(),
                    ));
                    elem.push_attribute(("channel", channel.id.as_ref()));
                    if let Some(catchup_id) = programme.catchup_id.as_ref() {
                        elem.push_attribute(("catchup-id", catchup_id.as_ref()));
                    }
                    continue_on_err!(writer.write_event_async(Event::Start(elem)).await);

                    if let Some(title) = &programme.title {
                        let elem = BytesStart::new("title");
                        continue_on_err!(writer.write_event_async(Event::Start(elem)).await);
                        continue_on_err!(writer.write_event_async(Event::Text(BytesText::new(title))).await);
                        continue_on_err!(writer.write_event_async(Event::End(BytesEnd::new("title"))).await);
                    }

                    if let Some(desc) = &programme.desc {
                        let elem = BytesStart::new("desc");
                        continue_on_err!(writer.write_event_async(Event::Start(elem)).await);
                        continue_on_err!(writer.write_event_async(Event::Text(BytesText::new(desc))).await);
                        continue_on_err!(writer.write_event_async(Event::End(BytesEnd::new("desc"))).await);
                    }

                    let _ = writer.write_event_async(Event::End(BytesEnd::new("programme"))).await;
                }
            }
        }

        let mut out = writer.into_inner();

        if let Err(err) = out.write_all("</tv>".as_bytes()).await {
            error!("EPG: Failed to write xml tv close {err}");
        }

        if let Err(e) = out.shutdown().await {
            error!("Failed to shutdown epg gzip encoder: {e}");
        }
    });

    let body_stream = ReaderStream::new(rx);
    try_unwrap_body!(axum::response::Response::builder()
        .header(axum::http::header::CONTENT_TYPE, mime::TEXT_XML.to_string())
        .body(axum::body::Body::from_stream(body_stream)))
}

async fn get_epg_channel(app_state: &Arc<AppState>, channel_id: &Arc<str>, epg_path: &Path) -> Option<EpgChannel> {
    let file_lock = app_state.app_config.file_locks.read_lock(epg_path).await;
    let epg_path = epg_path.to_path_buf();
    let channel_id = Arc::clone(channel_id);

    match task::spawn_blocking(move || -> Option<EpgChannel> {
        let _guard = file_lock;
        match BPlusTreeQuery::<Arc<str>, EpgChannel>::try_new(&epg_path) {
            Ok(mut query) => match query.query(&channel_id) {
                Ok(Some(item)) => Some(item),
                Ok(None) => None,
                Err(err) => {
                    error!("Failed to query db file {}: {err}", epg_path.display());
                    None
                }
            },
            Err(err) => {
                error!("Failed to read db file {}: {err}", epg_path.display());
                None
            }
        }
    })
    .await
    {
        Ok(result) => result,
        Err(err) => {
            error!("Failed to run epg query task: {err}");
            None
        }
    }
}

fn format_xmltv_time(ts: i64) -> String {
    if let Some(dt) = DateTime::from_timestamp(ts, 0) {
        dt.naive_utc().format("%Y-%m-%d %H:%M:%S").to_string()
    } else {
        String::new()
    }
}

fn get_applied_epg_timeshift(
    programme: &EpgProgramme,
    epg_processing_options: &EpgProcessingOptions,
) -> (String, String, i64, i64) {
    let (start_ts, stop_ts) = get_applied_epg_timestamps(programme, epg_processing_options);
    let (start_str, stop_str) = format_epg_timeshift_strings(programme, epg_processing_options, start_ts, stop_ts);
    (start_str, stop_str, start_ts, stop_ts)
}

fn get_applied_epg_timestamps(
    programme: &EpgProgramme,
    epg_processing_options: &EpgProcessingOptions,
) -> (i64, i64) {
    match &epg_processing_options.time_shift {
        EpgTimeShift::None | EpgTimeShift::TimeZone(_) => (programme.start, programme.stop),
        EpgTimeShift::Fixed(m) => {
            let off = i64::from(*m) * 60;
            (programme.start + off, programme.stop + off)
        }
    }
}

fn get_source_epg_timestamps(programme: &EpgProgramme) -> (i64, i64) {
    (programme.start, programme.stop)
}

fn format_epg_timeshift_strings(
    programme: &EpgProgramme,
    epg_processing_options: &EpgProcessingOptions,
    start_ts: i64,
    stop_ts: i64,
) -> (String, String) {
    match &epg_processing_options.time_shift {
        EpgTimeShift::TimeZone(tz) => {
            match (
                chrono::Utc.timestamp_opt(programme.start, 0).single(),
                chrono::Utc.timestamp_opt(programme.stop, 0).single(),
            ) {
                (Some(s_dt), Some(e_dt)) => (
                    s_dt.with_timezone(tz).format("%Y-%m-%d %H:%M:%S").to_string(),
                    e_dt.with_timezone(tz).format("%Y-%m-%d %H:%M:%S").to_string(),
                ),
                _ => (format_xmltv_time(start_ts), format_xmltv_time(stop_ts)),
            }
        }
        EpgTimeShift::None | EpgTimeShift::Fixed(_) => (format_xmltv_time(start_ts), format_xmltv_time(stop_ts)),
    }
}

fn from_programme(
    stream_id: &Arc<str>,
    epg_id: &Arc<str>,
    programme: &EpgProgramme,
    epg_processing_options: &EpgProcessingOptions,
) -> ShortEpgDto {
    let (start_str, end_str, start_ts, stop_ts) = get_applied_epg_timeshift(programme, epg_processing_options);

    ShortEpgDto {
        id: Arc::clone(stream_id),
        epg_id: Arc::clone(epg_id),
        title: programme.title.as_ref().map_or_else(String::new, ToString::to_string),
        lang: String::new(),
        start: start_str,
        end: end_str,
        description: programme.desc.as_ref().map_or_else(String::new, ToString::to_string),
        channel_id: Arc::clone(epg_id),
        start_timestamp: start_ts.to_string(),
        stop_timestamp: stop_ts.to_string(),
        stream_id: Arc::clone(stream_id),
        now_playing: None,
        has_archive: None,
    }
}

const DEFAULT_SHORT_EPG_LIMIT: u32 = 4;

pub async fn serve_short_epg(
    app_state: &Arc<AppState>,
    epg_path: &Path,
    user: &ProxyUserCredentials,
    target: &Arc<ConfigTarget>,
    channel_id: &Arc<str>,
    stream_id: Arc<str>,
    limit: u32,
) -> axum::response::Response {
    let short_epg = {
        // It seems provider set limit to 4 if it is undefined oor 0.
        let limit = if limit > 0 { limit } else { DEFAULT_SHORT_EPG_LIMIT };
        if file_exists_async(epg_path).await {
            if let Some(epg_channel) = get_epg_channel(app_state, channel_id, epg_path).await {
                let epg_processing_options = get_epg_processing_options(app_state, user, target);
                ShortEpgResultDto {
                    epg_listings: if limit > 0 {
                        epg_channel
                            .get_programme_with_limit(limit)
                            .iter()
                            .map(|p| from_programme(&stream_id, channel_id, p, &epg_processing_options))
                            .collect()
                    } else {
                        epg_channel
                            .programmes
                            .iter()
                            .map(|p| from_programme(&stream_id, channel_id, p, &epg_processing_options))
                            .collect()
                    },
                }
            } else {
                ShortEpgResultDto::default()
            }
        } else {
            ShortEpgResultDto::default()
        }
    };

    match serde_json::to_string(&short_epg) {
        Ok(json) => {
            (axum::http::StatusCode::OK, [(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())], json)
                .into_response()
        }
        Err(_) => internal_server_error!(),
    }
}

/// Serves per-stream EPG data for the UI "now playing" / "up next" display.
/// Queries epg.db by `epg_channel_id`, filters to 8h window, applies user timeshift.
/// Returns empty entries for all error/missing cases (no errors surfaced to client).
const STREAM_EPG_ARCHIVE_WINDOW_BACK_SECS: i64 = 4 * 3600;
const STREAM_EPG_ARCHIVE_WINDOW_FWD_SECS: i64 = 12 * 3600;

fn stream_epg_reference_map(items: &[StreamEpgItemRequest]) -> HashMap<&str, i64> {
    let mut references = HashMap::with_capacity(items.len());
    for item in items {
        if let Some(reference_ts) = item.reference_ts {
            references.entry(item.epg_channel_id.as_str()).or_insert(reference_ts);
        }
    }
    references
}

async fn serve_stream_epg(
    app_state: &Arc<AppState>,
    user: &ProxyUserCredentials,
    target: &Arc<ConfigTarget>,
    epg_path: &Path,
    items: Vec<StreamEpgItemRequest>,
) -> StreamEpgResponse {
    let epg_processing_options = get_epg_processing_options(app_state, user, target);
    let live_now = chrono::Utc::now().timestamp();

    // Deduplicate by first occurrence, preserving order
    let unique_ids: Vec<&str> = {
        let mut seen = HashSet::new();
        items
            .iter()
            .filter_map(|item| {
                let epg_channel_id = item.epg_channel_id.as_str();
                seen.insert(epg_channel_id).then_some(epg_channel_id)
            })
            .collect()
    };
    let query_ids = unique_ids
        .iter()
        .map(|id| (*id).intern())
        .collect::<Vec<Arc<str>>>();

    let channels = match epg_query_channels(
        &app_state.app_config.file_locks,
        epg_path,
        query_ids,
    )
    .await
    {
        Ok(results) => results,
        Err(err) => {
            error!("{err}");
            unique_ids.into_iter().map(|id| (id.intern(), None)).collect()
        }
    };

    let reference_by_channel = stream_epg_reference_map(&items);
    let mut entries = Vec::with_capacity(channels.len());

    for (epg_channel_id, channel) in channels {
        let reference_ts = reference_by_channel.get(epg_channel_id.as_ref()).copied().unwrap_or(live_now);
        let window_start = reference_ts.saturating_sub(STREAM_EPG_ARCHIVE_WINDOW_BACK_SECS);
        let window_end = reference_ts.saturating_add(STREAM_EPG_ARCHIVE_WINDOW_FWD_SECS);
        let programmes = match channel {
            Some(ch) => stream_epg_programmes_for_channel(&ch.programmes, &epg_processing_options, window_start, window_end),
            None => Vec::new(),
        };

        entries.push(StreamEpgEntry {
            epg_channel_id: epg_channel_id.to_string(),
            target_id: Some(target.id),
            programmes,
        });
    }

    StreamEpgResponse { entries }
}

fn stream_epg_programmes_for_channel(
    programmes: &[EpgProgramme],
    epg_processing_options: &EpgProcessingOptions,
    window_start: i64,
    window_end: i64,
) -> Vec<EpgProgrammeDto> {
    programmes
        .iter()
        .filter_map(|programme| {
            let (source_start_ts, source_stop_ts) = get_source_epg_timestamps(programme);
            (source_stop_ts > window_start && source_start_ts <= window_end).then(|| {
                let (start_ts, stop_ts) = get_applied_epg_timestamps(programme, epg_processing_options);
                let (start_str, stop_str) =
                    format_epg_timeshift_strings(programme, epg_processing_options, start_ts, stop_ts);
                EpgProgrammeDto {
                    start: start_str,
                    stop: stop_str,
                    start_timestamp: start_ts,
                    stop_timestamp: stop_ts,
                    title: programme.title.as_ref().map_or_else(String::new, ToString::to_string),
                }
            })
        })
        .collect()
}

fn group_stream_epg_items(
    items: Vec<StreamEpgItemRequest>,
) -> Result<Vec<(u16, Vec<StreamEpgItemRequest>)>, &'static str> {
    if items.is_empty() {
        return Err("items must not be empty");
    }

    let mut groups = Vec::<(u16, Vec<StreamEpgItemRequest>)>::new();
    for item in items {
        let Some(target_id) = item.target_id else {
            return Err("all items must include target_id");
        };

        if let Some((_, group_items)) = groups.iter_mut().find(|(group_target_id, _)| *group_target_id == target_id) {
            group_items.push(item);
        } else {
            groups.push((target_id, vec![item]));
        }
    }

    Ok(groups)
}

fn empty_stream_epg_entries(target_id: u16, items: &[StreamEpgItemRequest]) -> Vec<StreamEpgEntry> {
    let mut seen = HashSet::new();
    items
        .iter()
        .filter_map(|item| {
            let epg_channel_id = item.epg_channel_id.as_str();
            seen.insert(epg_channel_id).then_some(epg_channel_id)
        })
        .map(|epg_channel_id| StreamEpgEntry {
            epg_channel_id: epg_channel_id.to_string(),
            target_id: Some(target_id),
            programmes: Vec::new(),
        })
        .collect()
}

fn stream_epg_bad_request(message: &str) -> axum::response::Response {
    (axum::http::StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({ "error": message }))).into_response()
}

/// Handles stream EPG API requests for per-stream programme display.
pub(crate) async fn stream_epg_api(
    axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
    axum::extract::Json(stream_epg_req): axum::extract::Json<StreamEpgRequest>,
) -> impl IntoResponse + Send {
    let grouped_items = match group_stream_epg_items(stream_epg_req.items) {
        Ok(groups) => groups,
        Err(message) => return stream_epg_bad_request(message),
    };

    let config = app_state.app_config.config.load_full();
    let user = create_api_proxy_user(&app_state);
    let mut entries = Vec::new();

    for (target_id, items) in grouped_items {
        let Some(target) = app_state.app_config.get_target_by_id(target_id) else {
            return stream_epg_bad_request(&format!("unknown target_id: {target_id}"));
        };

        let epg_path = get_epg_path_for_target_by_type(config.as_ref(), &target, TargetType::M3u)
            .or_else(|| get_epg_path_for_target_by_type(config.as_ref(), &target, TargetType::Xtream));
        let Some(epg_path) = epg_path else {
            entries.extend(empty_stream_epg_entries(target_id, &items));
            continue;
        };

        let result = serve_stream_epg(&app_state, &user, &target, &epg_path, items).await;
        entries.extend(result.entries);
    }

    let result = StreamEpgResponse { entries };

    match serde_json::to_string(&result) {
        Ok(json) => (
            axum::http::StatusCode::OK,
            [(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())],
            json,
        ).into_response(),
        Err(_) => internal_server_error!(),
    }
}

/// Handles XMLTV EPG API requests, serving the appropriate EPG file with optional time-shifting based on user configuration.
///
/// Returns a 403 Forbidden response if the user or target is invalid or if the user lacks permission. If no EPG file is configured for the target, returns an empty EPG response. Otherwise, serves the EPG file, applying a time shift if specified by the user.
///
/// # Examples
///
/// ```
/// // Example usage within an Axum router:
/// let router = xmltv_api_register();
/// // A GET request to /xmltv.php with valid query parameters will invoke this handler.
/// ```
async fn xmltv_api(fingerprint: &Fingerprint, api_req: UserApiRequest, app_state: &Arc<AppState>) -> impl IntoResponse + Send {
    api_req.log_sanitized("xmltv_api");
    let auth_status = app_state.app_config.get_auth_error_status();
    let Some((user, target)) = get_user_target(&api_req, app_state) else {
        return auth_status.into_response();
    };
    if let Err(e) = resolve_api_user_context(user.clone(), target.clone(), fingerprint.clone(), app_state) {
        return e.into_player_response(auth_status);
    }

    let config = &app_state.app_config.config.load();
    let Some(epg_path) = get_epg_path_for_target(config, &target) else {
        return get_empty_epg_response();
    };

    serve_epg(app_state, &epg_path, &user, &target, None).await
}

async fn xmltv_api_get(
    fingerprint: Fingerprint,
    axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
    axum::extract::Query(api_req): axum::extract::Query<UserApiRequest>,
) -> impl IntoResponse + Send {
    xmltv_api(&fingerprint, api_req, &app_state).await
}

async fn xmltv_api_post(
    fingerprint: Fingerprint,
    axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
    UserApiRequestQueryOrBody(api_req): UserApiRequestQueryOrBody,
) -> impl IntoResponse + Send {
    xmltv_api(&fingerprint, api_req, &app_state).await
}

async fn epg_api_resource(
    fingerprint: Fingerprint,
    req_headers: axum::http::HeaderMap,
    axum::extract::Query(api_req): axum::extract::Query<UserApiRequest>,
    axum::extract::Path((username, password, resource)): axum::extract::Path<(String, String, String)>,
    axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl IntoResponse + Send {
    let auth_status = app_state.app_config.get_auth_error_status();
    let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state) else {
        return auth_status.into_response();
    };
    if let Err(e) = resolve_api_user_context(user.clone(), target.clone(), fingerprint.clone(), &app_state) {
        return e.into_player_response(auth_status);
    }

    let encrypt_secret = app_state.get_encrypt_secret();
    if let Ok(resource_url) = deobscure_text(&encrypt_secret, &resource) {
        resource_response(&app_state, &resource_url, &req_headers, None).await.into_response()
    } else {
        axum::http::StatusCode::BAD_REQUEST.into_response()
    }
}

/// Registers the XMLTV EPG API routes for handling HTTP GET requests.
///
/// The returned router maps the `/xmltv.php`, `/update/epg.php`, and `/epg` endpoints to the `xmltv_api` handler, enabling XMLTV EPG data retrieval with optional time-shifting and compression.
///
/// # Examples
///
/// ```
/// let router = xmltv_api_register();
/// // The router can now be used with an Axum server.
/// ```
pub fn xmltv_api_register() -> axum::Router<Arc<AppState>> {
    axum::Router::new()
        .route("/xmltv.php", axum::routing::get(xmltv_api_get).post(xmltv_api_post))
        .route("/epg", axum::routing::get(xmltv_api_get).post(xmltv_api_post))
        .route("/update/epg.php", axum::routing::get(xmltv_api_get).post(xmltv_api_post))
        .route(
            &format!("/{}/{{username}}/{{password}}/{{resource}}", storage_const::EPG_RESOURCE_PATH),
            axum::routing::get(epg_api_resource),
        )
}

#[cfg(test)]
mod tests {
    use super::{
        get_epg_path_for_target,
        get_epg_path_for_target_by_type,
        group_stream_epg_items,
        rewrite_epg_channel_resource_url,
        stream_epg_programmes_for_channel,
        stream_epg_reference_map,
    };
    use crate::model::{Config, ConfigTarget, M3uTargetOutput, TargetOutput, XtreamTargetFlagsSet, XtreamTargetOutput};
    use crate::utils::{EpgProcessingOptions, EpgTimeShift};
    use arc_swap::ArcSwapOption;
    use shared::{
        foundation::Filter,
        model::{EpgChannel, EpgProgramme, ProcessingOrder, StreamEpgItemRequest, TargetType},
        utils::{concat_path, obfuscate_text, Internable},
    };
    use std::fs;
    use std::sync::Arc;
    use tempfile::tempdir;

    fn test_target_with_xtream_and_m3u() -> ConfigTarget {
        ConfigTarget {
            id: 1,
            enabled: true,
            name: "mixed-target".to_string(),
            options: None,
            sort: None,
            filter: Filter::default(),
            output: vec![
                TargetOutput::Xtream(XtreamTargetOutput {
                    flags: XtreamTargetFlagsSet::new(),
                    trakt: None,
                    filter: None,
                }),
                TargetOutput::M3u(M3uTargetOutput {
                    filename: None,
                    include_type_in_url: false,
                    mask_redirect_url: false,
                    filter: None,
                }),
            ],
            rename: None,
            mapping_ids: None,
            mapping: Arc::new(ArcSwapOption::new(None)),
            favourites: None,
            processing_order: ProcessingOrder::default(),
            watch: None,
            use_memory_cache: false,
        }
    }

    fn test_config_with_storage(storage_dir: &str) -> Config {
        Config {
            storage_dir: storage_dir.to_string(),
            ..Default::default()
        }
    }

    #[test]
    fn get_epg_path_for_target_by_type_prefers_requested_output() {
        let dir = tempdir().expect("temp dir");
        let config = test_config_with_storage(dir.path().to_string_lossy().as_ref());
        let target = test_target_with_xtream_and_m3u();

        let xtream_storage = crate::repository::xtream_get_storage_path(&config, &target.name).expect("xtream storage");
        let m3u_storage = crate::repository::get_target_storage_path(&config, &target.name).expect("m3u storage");
        let xtream_epg = crate::repository::xtream_get_epg_file_path_for_target(&xtream_storage);
        let m3u_epg = crate::repository::m3u_get_epg_file_path_for_target(&m3u_storage);

        fs::create_dir_all(xtream_storage).expect("create xtream dir");
        fs::create_dir_all(m3u_epg.parent().expect("m3u epg parent")).expect("create m3u dir");
        fs::write(&xtream_epg, b"xtream").expect("write xtream epg");
        fs::write(&m3u_epg, b"m3u").expect("write m3u epg");

        let picked_xtream = get_epg_path_for_target_by_type(&config, &target, TargetType::Xtream).expect("xtream path");
        let picked_m3u = get_epg_path_for_target_by_type(&config, &target, TargetType::M3u).expect("m3u path");

        assert_eq!(picked_xtream, xtream_epg);
        assert_eq!(picked_m3u, m3u_epg);
    }

    #[test]
    fn get_epg_path_for_target_keeps_output_order_fallback() {
        let dir = tempdir().expect("temp dir");
        let config = test_config_with_storage(dir.path().to_string_lossy().as_ref());
        let target = test_target_with_xtream_and_m3u();

        let xtream_storage = crate::repository::xtream_get_storage_path(&config, &target.name).expect("xtream storage");
        let m3u_storage = crate::repository::get_target_storage_path(&config, &target.name).expect("m3u storage");
        let xtream_epg = crate::repository::xtream_get_epg_file_path_for_target(&xtream_storage);
        let m3u_epg = crate::repository::m3u_get_epg_file_path_for_target(&m3u_storage);

        fs::create_dir_all(xtream_storage).expect("create xtream dir");
        fs::create_dir_all(m3u_epg.parent().expect("m3u epg parent")).expect("create m3u dir");
        fs::write(&xtream_epg, b"xtream").expect("write xtream epg");
        fs::write(&m3u_epg, b"m3u").expect("write m3u epg");

        let picked = get_epg_path_for_target(&config, &target).expect("fallback path");
        assert_eq!(picked, xtream_epg);
    }

    #[test]
    fn test_group_stream_epg_items_rejects_empty_requests() {
        let err = group_stream_epg_items(Vec::new()).unwrap_err();
        assert_eq!(err, "items must not be empty");
    }

    #[test]
    fn test_group_stream_epg_items_rejects_missing_target() {
        let err = group_stream_epg_items(vec![StreamEpgItemRequest {
            epg_channel_id: "epg-1".to_string(),
            target_id: None,
            reference_ts: None,
        }])
        .unwrap_err();
        assert_eq!(err, "all items must include target_id");
    }

    #[test]
    fn test_group_stream_epg_items_groups_by_target_in_input_order() {
        let groups = group_stream_epg_items(vec![
            StreamEpgItemRequest {
                epg_channel_id: "epg-1".to_string(),
                target_id: Some(2),
                reference_ts: None,
            },
            StreamEpgItemRequest {
                epg_channel_id: "epg-2".to_string(),
                target_id: Some(5),
                reference_ts: None,
            },
            StreamEpgItemRequest {
                epg_channel_id: "epg-3".to_string(),
                target_id: Some(2),
                reference_ts: None,
            },
        ])
        .unwrap();

        assert_eq!(groups.len(), 2);
        assert_eq!(groups[0].0, 2);
        assert_eq!(
            groups[0].1.iter().map(|item| item.epg_channel_id.as_str()).collect::<Vec<_>>(),
            vec!["epg-1", "epg-3"]
        );
        assert_eq!(groups[1].0, 5);
        assert_eq!(groups[1].1.iter().map(|item| item.epg_channel_id.as_str()).collect::<Vec<_>>(), vec!["epg-2"]);
    }

    #[test]
    fn stream_epg_archive_window_uses_reference() {
        let reference = 1_700_000_000_i64 - 3_600;
        let window_start = reference.saturating_sub(super::STREAM_EPG_ARCHIVE_WINDOW_BACK_SECS);
        let window_end = reference.saturating_add(super::STREAM_EPG_ARCHIVE_WINDOW_FWD_SECS);
        assert!(window_start < reference);
        assert!(window_end > reference);
    }

    #[test]
    fn stream_epg_reference_map_keeps_first_reference_per_channel() {
        let items = [
            StreamEpgItemRequest {
                epg_channel_id: "epg-1".to_string(),
                target_id: Some(1),
                reference_ts: Some(100),
            },
            StreamEpgItemRequest {
                epg_channel_id: "epg-1".to_string(),
                target_id: Some(1),
                reference_ts: Some(200),
            },
            StreamEpgItemRequest {
                epg_channel_id: "epg-2".to_string(),
                target_id: Some(1),
                reference_ts: None,
            },
        ];
        let references = stream_epg_reference_map(&items);

        assert_eq!(references.get("epg-1"), Some(&100));
        assert_eq!(references.get("epg-2"), None);
    }

    fn sample_channel(icon: Option<&str>) -> EpgChannel {
        EpgChannel {
            id: "channel-1".intern(),
            title: Some("Channel".intern()),
            icon: icon.map(Internable::intern),
            programmes: Vec::new(),
        }
    }

    #[test]
    fn rewrite_epg_channel_resource_url_wraps_external_icon() {
        let secret = [9u8; 16];
        let resource_url = "/api/v1/playlist/resource";
        let channel = sample_channel(Some("https://cdn.example.com/logo.png"));

        let rewritten = rewrite_epg_channel_resource_url(&secret, resource_url, channel);

        assert_eq!(
            rewritten.icon.as_deref(),
            Some(concat_path(resource_url, &obfuscate_text(&secret, "https://cdn.example.com/logo.png")).as_str())
        );
    }

    #[test]
    fn rewrite_epg_channel_resource_url_keeps_internal_path() {
        let secret = [9u8; 16];
        let resource_url = "/api/v1/playlist/resource";
        let channel = sample_channel(Some("/api/v1/library/thumbnail/test"));

        let rewritten = rewrite_epg_channel_resource_url(&secret, resource_url, channel);

        assert_eq!(rewritten.icon.as_deref(), Some("/api/v1/library/thumbnail/test"));
    }

    #[test]
    fn stream_epg_programmes_for_channel_filters_using_shifted_fixed_times() {
        let window_start = 10_000;
        let window_end = window_start + 8 * 3600;
        let epg_processing_options = EpgProcessingOptions {
            rewrite_urls: false,
            time_shift: EpgTimeShift::Fixed(120),
            encrypt_secret: [0; 16],
        };
        let programmes = vec![
            EpgProgramme::new_all(
                window_start - 7_300,
                window_start - 100,
                "channel-1".intern(),
                Some("Shifted Into Window".intern()),
                None,
                None,
            ),
            EpgProgramme::new_all(
                window_start + 60,
                window_start + 600,
                "channel-1".intern(),
                Some("Already In Window".intern()),
                None,
                None,
            ),
            EpgProgramme::new_all(
                window_end + 60,
                window_end + 600,
                "channel-1".intern(),
                Some("Still Outside Window".intern()),
                None,
                None,
            ),
        ];

        let filtered =
            stream_epg_programmes_for_channel(&programmes, &epg_processing_options, window_start, window_end);

        assert_eq!(
            filtered.iter().map(|programme| programme.title.as_str()).collect::<Vec<_>>(),
            vec!["Already In Window"]
        );
        assert_eq!(filtered[0].start_timestamp, window_start + 7_260);
        assert_eq!(filtered[0].stop_timestamp, window_start + 7_800);
    }

}