pipedash_plugin_api/

utils.rs

use std::collections::HashMap;
use std::future::Future;
use std::sync::OnceLock;
use std::time::Duration;

use chrono::{
    DateTime,
    Utc,
};
use futures::StreamExt;
use reqwest::Url;

use crate::config::comma_separated;
use crate::http::map_send_error;
use crate::{
    PipelineStatus,
    PluginError,
    PluginResult,
};

pub struct RetryPolicy {
    pub max_retries: usize,
    pub initial_delay: Duration,
    pub exponential_backoff: bool,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_retries: 3,
            initial_delay: Duration::from_millis(100),
            exponential_backoff: true,
        }
    }
}

impl RetryPolicy {
    pub fn new(max_retries: usize, initial_delay: Duration, exponential_backoff: bool) -> Self {
        Self {
            max_retries,
            initial_delay,
            exponential_backoff,
        }
    }

    pub async fn retry<F, Fut, T>(&self, operation: F) -> PluginResult<T>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = PluginResult<T>>,
    {
        let mut delay = self.initial_delay;
        let mut attempt = 1;

        loop {
            match operation().await {
                Ok(result) => return Ok(result),
                Err(e) if attempt < self.max_retries && Self::should_retry(&e, delay) => {
                    tokio::time::sleep(delay).await;
                    delay = self.next_delay(delay);
                    attempt += 1;
                }
                Err(e) => return Err(e),
            }
        }
    }

    pub async fn retry_unsent<F, Fut>(&self, url: &Url, send: F) -> PluginResult<reqwest::Response>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = Result<reqwest::Response, reqwest::Error>>,
    {
        let mut delay = self.initial_delay;
        let mut attempt = 1;

        loop {
            match send().await {
                Ok(response) => return Ok(response),
                Err(e) if attempt < self.max_retries && e.is_connect() && !e.is_timeout() => {
                    tokio::time::sleep(delay).await;
                    delay = self.next_delay(delay);
                    attempt += 1;
                }
                Err(e) => return Err(map_send_error(e, url)),
            }
        }
    }

    fn should_retry(error: &PluginError, delay: Duration) -> bool {
        error.is_transient() && error.retry_after().is_none_or(|hint| hint <= delay)
    }

    fn next_delay(&self, delay: Duration) -> Duration {
        if self.exponential_backoff {
            delay.saturating_mul(2)
        } else {
            delay
        }
    }
}

#[derive(Debug)]
pub struct Cached<T>(OnceLock<T>);

impl<T> Default for Cached<T> {
    fn default() -> Self {
        Self(OnceLock::new())
    }
}

impl<T: Clone> Cached<T> {
    pub async fn get_or_fetch<F, Fut>(&self, fetch: F) -> PluginResult<T>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = PluginResult<T>>,
    {
        if let Some(value) = self.0.get() {
            return Ok(value.clone());
        }
        let value = fetch().await?;
        let _ = self.0.set(value.clone());
        Ok(value)
    }
}

pub async fn retain_with_ci<F, Fut>(ids: Vec<String>, has_ci: F) -> Vec<String>
where
    F: Fn(String) -> Fut,
    Fut: Future<Output = PluginResult<bool>>,
{
    futures::stream::iter(ids)
        .map(|id| {
            let check = has_ci(id.clone());
            async move { check.await.unwrap_or(true).then_some(id) }
        })
        .buffered(8)
        .filter_map(std::future::ready)
        .collect()
        .await
}

pub fn status_from_conclusion(
    status: &str, conclusion: Option<&str>, successful: &[&str], cancelled: &[&str],
) -> PipelineStatus {
    match (status, conclusion) {
        ("completed", Some(c)) if successful.contains(&c) => PipelineStatus::Success,
        ("completed", Some(c)) if cancelled.contains(&c) => PipelineStatus::Cancelled,
        ("completed", Some("skipped")) => PipelineStatus::Skipped,
        ("completed", _) => PipelineStatus::Failed,
        ("in_progress", _) | ("queued", _) => PipelineStatus::Running,
        _ => PipelineStatus::Pending,
    }
}

pub fn repository_name(repo_url: &str) -> String {
    let cleaned = repo_url
        .trim_end_matches('/')
        .trim_end_matches(".git")
        .trim_end_matches('/');

    if !cleaned.contains("://") {
        if let Some((_, path)) = cleaned.rsplit_once(':') {
            if path.contains('/') {
                return path.to_string();
            }
        }
    }

    let mut segments = cleaned.rsplit('/');
    match (segments.next(), segments.next()) {
        (Some(repo), Some(org)) => format!("{org}/{repo}"),
        _ => repo_url.to_string(),
    }
}

pub fn parse_timestamp(value: &str) -> Option<DateTime<Utc>> {
    DateTime::parse_from_rfc3339(value)
        .ok()
        .map(|parsed| parsed.with_timezone(&Utc))
}

pub fn duration_seconds(
    started: Option<DateTime<Utc>>, finished: Option<DateTime<Utc>>,
) -> Option<i64> {
    Some((finished? - started?).num_seconds())
}

pub fn collect_partial<T>(
    results: impl IntoIterator<Item = PluginResult<T>>, mut skipped: impl FnMut(&PluginError),
) -> PluginResult<Vec<T>> {
    let mut items = Vec::new();
    for result in results {
        match result {
            Ok(item) => items.push(item),
            Err(error) if error.is_transient() => return Err(error),
            Err(error) => skipped(&error),
        }
    }
    Ok(items)
}

pub fn selected_items(config: &HashMap<String, String>) -> impl Iterator<Item = &str> {
    config
        .get("selected_items")
        .into_iter()
        .flat_map(|items| comma_separated(items))
}

pub const fn provider_type(package_name: &'static str) -> &'static str {
    const PREFIX: &[u8] = b"pipedash-plugin-";
    let name = package_name.as_bytes();
    if name.len() < PREFIX.len() {
        return package_name;
    }
    let (head, rest) = name.split_at(PREFIX.len());
    let mut index = 0;
    while index < PREFIX.len() {
        if head[index] != PREFIX[index] {
            return package_name;
        }
        index += 1;
    }
    match core::str::from_utf8(rest) {
        Ok(stripped) => stripped,
        Err(_) => package_name,
    }
}

pub fn svg_data_url(svg: &str) -> String {
    const PREFIX: &str = "data:image/svg+xml,";
    const HEX: &[u8; 16] = b"0123456789ABCDEF";
    let mut url = String::with_capacity(PREFIX.len() + svg.len());
    url.push_str(PREFIX);
    for byte in svg.bytes() {
        if byte.is_ascii_alphanumeric() || b"-._~/:;,=!*'()".contains(&byte) {
            url.push(char::from(byte));
        } else {
            url.push('%');
            url.push(char::from(HEX[usize::from(byte >> 4)]));
            url.push(char::from(HEX[usize::from(byte & 0x0F)]));
        }
    }
    url
}

#[cfg(test)]
mod tests;