18346462606

Committed 08 Oct 2025 01:33PM UTC coverage: 54.854% (-0.9%) from 55.73%

Build # 18346462606

Build Type

Pull #45

github

Committed by

web-flow

Commit Message

Merge 9146fc44e into c1d3177fb

Pull Request Pull Request #45: HTTP client with configurable IPFS gateways into its own module

Run Details

399 of 536 new or added lines in 11 files covered. (74.44%)

1 existing line in 1 file now uncovered.

3554 of 6479 relevant lines covered (54.85%)

2.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.38

/src/httpclient/mod.rs

use std::path::{Path, PathBuf};
use tracing::info;

use crate::content::get_filename;
use crate::content::{extensions, try_exists, write_and_postprocess_file, Options};
use crate::httpclient::fetch::try_fetch_response;
use crate::httpclient::retry::{retry_operation, should_retry};
use crate::httpclient::stream::stream_http_to_file;
use crate::url::{get_data_url, is_data_url, resolve_url_with_gateways};
use crate::url::{IpfsGatewayConfig, IpfsGatewayType, IPFS_GATEWAYS};

pub mod fetch;
pub mod retry;
pub mod stream;

#[derive(Clone, Debug)]
pub struct HttpClient {
    pub(crate) ipfs_gateways: Vec<IpfsGatewayConfig>,
}

impl HttpClient {
    pub fn new() -> Self {
        let ipfs_gateways = IPFS_GATEWAYS.to_vec();
        Self { ipfs_gateways }
    }

    pub fn new_with_gateways(gateways: Vec<(String, IpfsGatewayType)>) -> Self {
        let ipfs_gateways = gateways
            .into_iter()
            .map(|(url, gateway_type)| IpfsGatewayConfig {
                url: Box::leak(url.into_boxed_str()),
                gateway_type,
            })
            .collect();
        Self { ipfs_gateways }
    }

    pub async fn fetch(&self, url: &str) -> anyhow::Result<Vec<u8>> {
        if is_data_url(url) {
            return get_data_url(url)
                .ok_or_else(|| anyhow::anyhow!("Failed to parse data URL: {}", url));
        }

        let resolved_url = resolve_url_with_gateways(url, &self.ipfs_gateways);
        let gateways = self.ipfs_gateways.clone();

        const MAX_RETRIES: u32 = 5;
        retry_operation(
            || {
                let url = resolved_url.clone();
                let gateways = gateways.clone();
                Box::pin(async move {
                    match try_fetch_response(&url, &gateways).await {
                        (Ok(response), status) => match response.bytes().await {
                            Ok(b) => (Ok(b.to_vec()), status),
                            Err(e) => (Err(anyhow::anyhow!(e)), status),
                        },
                        (Err(err), status) => (Err(err), status),
                    }
                })
            },
            MAX_RETRIES,
            should_retry,
            &resolved_url,
        )
        .await
    }

    pub async fn fetch_and_write(
        &self,
        url: &str,
        token: &impl crate::chain::common::ContractTokenInfo,
        output_path: &Path,
        options: Options,
    ) -> anyhow::Result<PathBuf> {
        let mut file_path = get_filename(url, token, output_path, options).await?;

        if let Some(existing_path) = try_exists(&file_path).await? {
            tracing::debug!(
                "File already exists at {} (skipping download)",
                existing_path.display()
            );
            return Ok(existing_path);
        }

        let parent = file_path.parent().ok_or_else(|| {
            anyhow::anyhow!("File path has no parent directory: {}", file_path.display())
        })?;
        tokio::fs::create_dir_all(parent).await.map_err(|e| {
            anyhow::anyhow!("Failed to create directory {}: {}", parent.display(), e)
        })?;

        if is_data_url(url) {
            let content = get_data_url(url)
                .ok_or_else(|| anyhow::anyhow!("Failed to parse data URL: {}", url))?;
            if !extensions::has_known_extension(&file_path) {
                if let Some(detected_ext) = extensions::detect_media_extension(&content) {
                    let current_path_str = file_path.to_string_lossy();
                    tracing::debug!("Appending detected media extension: {}", detected_ext);
                    file_path = PathBuf::from(format!("{current_path_str}.{detected_ext}"));
                }
            }

            info!("Saving {} (data url)", file_path.display());
            write_and_postprocess_file(&file_path, &content, url).await?;
            info!("Saved {} (data url)", file_path.display());
            return Ok(file_path);
        }

        let resolved_url = resolve_url_with_gateways(url, &self.ipfs_gateways);
        if url == resolved_url {
            info!("Saving {} (url: {})", file_path.display(), url);
        } else {
            info!(
                "Saving {} (original: {}, resolved: {})",
                file_path.display(),
                url,
                resolved_url
            );
        }
        let file_path = self
            .fetch_and_stream_to_file(&resolved_url, &file_path, 5)
            .await?;

        write_and_postprocess_file(&file_path, &[], url).await?;
        if url == resolved_url {
            info!("Saved {} (url: {})", file_path.display(), url);
        } else {
            info!(
                "Saved {} (original: {}, resolved: {})",
                file_path.display(),
                url,
                resolved_url
            );
        }
        Ok(file_path)
    }

    pub async fn try_fetch_response(
        &self,
        url: &str,
    ) -> (
        anyhow::Result<reqwest::Response>,
        Option<reqwest::StatusCode>,
    ) {
        try_fetch_response(url, &self.ipfs_gateways).await
    }

    pub(crate) async fn fetch_and_stream_to_file(
        &self,
        url: &str,
        file_path: &Path,
        max_retries: u32,
    ) -> anyhow::Result<PathBuf> {
        fetch_and_stream_to_file(url, file_path, max_retries, &self.ipfs_gateways).await
    }
}

async fn fetch_and_stream_to_file(
    url: &str,
    file_path: &Path,
    max_retries: u32,
    gateways: &[IpfsGatewayConfig],
) -> anyhow::Result<PathBuf> {
    retry_operation(
        || {
            let url = url.to_string();
            let file_path = file_path.to_path_buf();
            let gateways = gateways.to_owned();
            Box::pin(async move {
                match try_fetch_response(&url, &gateways).await {
                    (Ok(response), status) => {
                        (stream_http_to_file(response, &file_path).await, status)
                    }
                    (Err(err), status) => (Err(err), status),
                }
            })
        },
        max_retries,
        should_retry,
        url,
    )
    .await
}

impl Default for HttpClient {
    fn default() -> Self {
        Self::new()
    }
}

1	use std::path::{Path, PathBuf};
2	use tracing::info;
3
4	use crate::content::get_filename;
5	use crate::content::{extensions, try_exists, write_and_postprocess_file, Options};
6	use crate::httpclient::fetch::try_fetch_response;
7	use crate::httpclient::retry::{retry_operation, should_retry};
8	use crate::httpclient::stream::stream_http_to_file;
9	use crate::url::{get_data_url, is_data_url, resolve_url_with_gateways};
10	use crate::url::{IpfsGatewayConfig, IpfsGatewayType, IPFS_GATEWAYS};
11
12	pub mod fetch;
13	pub mod retry;
14	pub mod stream;
15
16	#[derive(Clone, Debug)]
17	pub struct HttpClient {
18	pub(crate) ipfs_gateways: Vec<IpfsGatewayConfig>,
19	}
20
21	impl HttpClient {
22	pub fn new() -> Self {	17✔
23	let ipfs_gateways = IPFS_GATEWAYS.to_vec();	17✔
24	Self { ipfs_gateways }	17✔
25	}	17✔
26
27	pub fn new_with_gateways(gateways: Vec<(String, IpfsGatewayType)>) -> Self {	3✔
28	let ipfs_gateways = gateways	3✔
29	.into_iter()	3✔
30	.map(\|(url, gateway_type)\| IpfsGatewayConfig {	3✔
31	url: Box::leak(url.into_boxed_str()),	3✔
32	gateway_type,	3✔
33	})	3✔
34	.collect();	3✔
35	Self { ipfs_gateways }	3✔
36	}	3✔
37
NEW 38	pub async fn fetch(&self, url: &str) -> anyhow::Result<Vec<u8>> {	×
NEW 39	if is_data_url(url) {	×
NEW 40	return get_data_url(url)	×
NEW 41	.ok_or_else(\|\| anyhow::anyhow!("Failed to parse data URL: {}", url));	×
NEW 42	}	×
43
NEW 44	let resolved_url = resolve_url_with_gateways(url, &self.ipfs_gateways);	×
NEW 45	let gateways = self.ipfs_gateways.clone();	×
46
47	const MAX_RETRIES: u32 = 5;
NEW 48	retry_operation(	×
NEW 49	\|\| {	×
NEW 50	let url = resolved_url.clone();	×
NEW 51	let gateways = gateways.clone();	×
NEW 52	Box::pin(async move {	×
NEW 53	match try_fetch_response(&url, &gateways).await {	×
NEW 54	(Ok(response), status) => match response.bytes().await {	×
NEW 55	Ok(b) => (Ok(b.to_vec()), status),	×
NEW 56	Err(e) => (Err(anyhow::anyhow!(e)), status),	×
57	},
NEW 58	(Err(err), status) => (Err(err), status),	×
59	}
NEW 60	})	×
NEW 61	},	×
62	MAX_RETRIES,
63	should_retry,
NEW 64	&resolved_url,	×
65	)
NEW 66	.await	×
NEW 67	}	×
68
69	pub async fn fetch_and_write(	18✔
70	&self,	18✔
71	url: &str,	18✔
72	token: &impl crate::chain::common::ContractTokenInfo,	18✔
73	output_path: &Path,	18✔
74	options: Options,	18✔
75	) -> anyhow::Result<PathBuf> {	18✔
76	let mut file_path = get_filename(url, token, output_path, options).await?;	18✔
77
78	if let Some(existing_path) = try_exists(&file_path).await? {	18✔
79	tracing::debug!(	7✔
NEW 80	"File already exists at {} (skipping download)",	×
NEW 81	existing_path.display()	×
82	);
83	return Ok(existing_path);	7✔
84	}	11✔
85
86	let parent = file_path.parent().ok_or_else(\|\| {	11✔
NEW 87	anyhow::anyhow!("File path has no parent directory: {}", file_path.display())	×
NEW 88	})?;	×
89	tokio::fs::create_dir_all(parent).await.map_err(\|e\| {	11✔
NEW 90	anyhow::anyhow!("Failed to create directory {}: {}", parent.display(), e)	×
NEW 91	})?;	×
92
93	if is_data_url(url) {	11✔
94	let content = get_data_url(url)	7✔
95	.ok_or_else(\|\| anyhow::anyhow!("Failed to parse data URL: {}", url))?;	7✔
96	if !extensions::has_known_extension(&file_path) {	7✔
97	if let Some(detected_ext) = extensions::detect_media_extension(&content) {	7✔
98	let current_path_str = file_path.to_string_lossy();	7✔
99	tracing::debug!("Appending detected media extension: {}", detected_ext);	7✔
100	file_path = PathBuf::from(format!("{current_path_str}.{detected_ext}"));	7✔
NEW 101	}	×
NEW 102	}	×
103
104	info!("Saving {} (data url)", file_path.display());	7✔
105	write_and_postprocess_file(&file_path, &content, url).await?;	7✔
106	info!("Saved {} (data url)", file_path.display());	7✔
107	return Ok(file_path);	7✔
108	}	4✔
109
110	let resolved_url = resolve_url_with_gateways(url, &self.ipfs_gateways);	4✔
111	if url == resolved_url {	4✔
NEW 112	info!("Saving {} (url: {})", file_path.display(), url);	×
113	} else {
114	info!(	4✔
NEW 115	"Saving {} (original: {}, resolved: {})",	×
NEW 116	file_path.display(),	×
117	url,
118	resolved_url
119	);
120	}
121	let file_path = self	4✔
122	.fetch_and_stream_to_file(&resolved_url, &file_path, 5)	4✔
123	.await?;	4✔
124
NEW 125	write_and_postprocess_file(&file_path, &[], url).await?;	×
NEW 126	if url == resolved_url {	×
NEW 127	info!("Saved {} (url: {})", file_path.display(), url);	×
128	} else {
NEW 129	info!(	×
NEW 130	"Saved {} (original: {}, resolved: {})",	×
NEW 131	file_path.display(),	×
132	url,
133	resolved_url
134	);
135	}
NEW 136	Ok(file_path)	×
137	}	18✔
138
139	pub async fn try_fetch_response(	4✔
140	&self,	4✔
141	url: &str,	4✔
142	) -> (	4✔
143	anyhow::Result<reqwest::Response>,	4✔
144	Option<reqwest::StatusCode>,	4✔
145	) {	4✔
146	try_fetch_response(url, &self.ipfs_gateways).await	4✔
147	}	4✔
148
149	pub(crate) async fn fetch_and_stream_to_file(	4✔
150	&self,	4✔
151	url: &str,	4✔
152	file_path: &Path,	4✔
153	max_retries: u32,	4✔
154	) -> anyhow::Result<PathBuf> {	4✔
155	fetch_and_stream_to_file(url, file_path, max_retries, &self.ipfs_gateways).await	4✔
156	}	4✔
157	}
158
159	async fn fetch_and_stream_to_file(	4✔
160	url: &str,	4✔
161	file_path: &Path,	4✔
162	max_retries: u32,	4✔
163	gateways: &[IpfsGatewayConfig],	4✔
164	) -> anyhow::Result<PathBuf> {	4✔
165	retry_operation(	4✔
166	\|\| {	4✔
167	let url = url.to_string();	4✔
168	let file_path = file_path.to_path_buf();	4✔
169	let gateways = gateways.to_owned();	4✔
170	Box::pin(async move {	4✔
171	match try_fetch_response(&url, &gateways).await {	4✔
NEW 172	(Ok(response), status) => {	×
NEW 173	(stream_http_to_file(response, &file_path).await, status)	×
174	}
175	(Err(err), status) => (Err(err), status),	4✔
176	}
177	})	4✔
178	},	4✔
179	max_retries,	4✔
180	should_retry,
181	url,	4✔
182	)
183	.await	4✔
184	}	4✔
185
186	impl Default for HttpClient {
NEW 187	fn default() -> Self {	×
NEW 188	Self::new()	×
NEW 189	}	×
190	}

0xmichalis / nftbk / 18346462606

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous