reala-subscription 0.0.0

#![doc = include_str!("../README.md")]

mod model;
pub mod util;

use std::sync::{Arc, LazyLock};
use std::time::Duration;

use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use ureq::Agent;

pub use self::model::Proxy;

#[non_exhaustive]
#[derive(Debug)]
/// Errors that may occur when fetching or parsing subscription content.
pub enum Error {
    /// Failed to fetch the subscription content.
    Fetch(ureq::Error),

    /// Failed to parse the subscription content.
    Parse,
}

impl std::error::Error for Error {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Error::Fetch(e) => Some(e),
            Error::Parse => None,
        }
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Error::Fetch(e) => write!(f, "Failed to fetch subscription content: {}", e),
            Error::Parse => write!(f, "Failed to parse subscription content"),
        }
    }
}

impl From<ureq::Error> for Error {
    fn from(e: ureq::Error) -> Self {
        Error::Fetch(e)
    }
}

#[non_exhaustive]
#[derive(Debug, Clone, Copy)]
#[derive(serde::Serialize, serde::Deserialize)]
/// The type of subscription.
///
/// Currently, only Clash-compatible subscriptions are supported.
pub enum SubscriptionKind {
    #[serde(rename = "clash")]
    /// A Clash-compatible subscription, i.e., the content is in Clash config
    /// format.
    ClashCompatible,
}

#[derive(Debug, Clone)]
#[derive(serde::Serialize, serde::Deserialize)]
/// Options for fetching subscription content.
pub struct Subscription {
    /// The type of subscription.
    pub kind: SubscriptionKind,

    /// The target URL.
    pub url: util::Uri,

    #[serde(skip_serializing_if = "Option::is_none")]
    /// The User-Agent.
    ///
    /// By default, `mihomo.party/v1.8.4 (clash.meta)` is used.
    pub user_agent: Option<Arc<str>>,

    #[serde(skip_serializing_if = "Option::is_none")]
    /// The HTTP / Socks5 proxy you may use to fetch the subscription.
    ///
    /// If `None`, no proxy is used.
    pub proxy: Option<util::Proxy>,
}

impl Subscription {
    /// Fetches the subscription content, and returns the parsed proxies.
    ///
    /// Notes that the proxies may be empty if no compatible proxies are found.
    pub fn fetch(&self) -> Result<Vec<Proxy>, Error> {
        static AGENT: LazyLock<Agent> = LazyLock::new(|| {
            const DEFAULT_USER_AGENT: &str = "mihomo.party/v1.8.4 (clash.meta)";

            Agent::config_builder()
                .timeout_global(Some(Duration::from_secs(5)))
                .user_agent(DEFAULT_USER_AGENT)
                .build()
                .into()
        });

        let mut request = AGENT
            .get(&*self.url)
            .config()
            .proxy(self.proxy.as_deref().cloned());

        if let Some(user_agent) = &self.user_agent {
            request = request.user_agent(user_agent);
        }

        let response = request.build().call()?.into_body().read_to_vec()?;

        let response = String::from_utf8(response).map_err(|_e| Error::Parse)?;

        match self.kind {
            SubscriptionKind::ClashCompatible => Proxy::parse_clash_compatible(&response),
        }
    }
}

wrapper_lite::wrapper!(
    #[wrapper_impl(DebugName)]
    /// A background subscription update task.
    pub struct Task {
        /// The subscription info.
        pending: Subscription,

        /// The update interval.
        ///
        /// The default is 30 minutes.
        interval: Option<Duration>,

        /// The channel TX we use to send back the parsed proxies.
        proxies_tx: mpsc::Sender<Vec<Proxy>>,
    }
);

impl Task {
    /// Creates a new subscription update task.
    pub fn new(
        pending: Subscription,
        interval: Option<Duration>,
    ) -> (Self, mpsc::Receiver<Vec<Proxy>>) {
        let (proxies_tx, proxies_rx) = mpsc::channel(1);

        (
            Self {
                pending,
                interval,
                proxies_tx,
            },
            proxies_rx,
        )
    }

    /// Spawns the subscription update task.
    pub async fn spawn(self) -> JoinHandle<()> {
        tokio::spawn(async move {
            loop {
                let pending = self.pending.clone();

                let proxies = tokio::task::spawn_blocking(move || pending.fetch()).await;

                match proxies {
                    Ok(Ok(proxies)) => {
                        let Ok(()) = self.proxies_tx.send(proxies).await else {
                            tracing::info!("Subscription task channel closed, exiting task.");

                            return;
                        };
                    }
                    Ok(Err(e)) => {
                        tracing::error!("Failed to fetch subscription: {:?}", e);
                    }
                    Err(e) => {
                        tracing::error!("Failed to join subscription task: {:?}", e);
                    }
                }

                tokio::time::sleep(self.interval.unwrap_or(Duration::from_mins(30))).await;
            }
        })
    }
}