1mod abort;
23
24#[cfg(not(target_arch = "wasm32"))]
25mod h1;
26
27use blitz_traits::net::{
28 Body, Bytes, CookieJar as CookieStorage, NetHandler, NetProvider, NetWaker, Request,
29 http::{HeaderValue, header::CONTENT_TYPE},
30};
31use blitz_traits::platform::{
32 FetchError, FetchHandler, FetchProvider, FetchRequest, FetchResponse, HeaderMap, StatusCode,
33};
34use data_url::DataUrl;
35use std::sync::Arc;
36
37#[cfg(feature = "cookies")]
43pub use blitz_traits::net::CookieJar;
44
45#[cfg(all(feature = "cache", not(target_arch = "wasm32")))]
46pub use h1::cache::PrivateCache;
47
48pub use blitz_traits::tls_policy::TlsPolicyError;
50
51#[cfg(not(target_arch = "wasm32"))]
53pub use h1::{HostStats, connection_stats};
54
55pub const DEFAULT_USER_AGENT: &str =
64 "Mozilla/5.0 (X11; Linux x86_64; rv:60.0) Gecko/20100101 Firefox/81.0";
65
66#[cfg(not(target_arch = "wasm32"))]
67type Client = h1::H1Client;
68
69#[cfg(target_arch = "wasm32")]
70#[derive(Clone)]
71struct Client;
72
73#[cfg(target_arch = "wasm32")]
74impl Client {
75 fn new(user_agent: &str, _cookies: Option<Arc<dyn CookieStorage>>) -> Result<Self, FetchError> {
76 HeaderValue::from_str(user_agent)
77 .map_err(|error| FetchError::InvalidRequest(error.to_string()))?;
78 Ok(Self)
79 }
80
81 async fn fetch(&self, _request: FetchRequest) -> Result<FetchResponse, FetchError> {
82 Err(FetchError::Network(
83 "the native HTTP transport is unavailable on wasm32".to_owned(),
84 ))
85 }
86}
87
88#[cfg(target_arch = "wasm32")]
89fn spawn(fut: impl Future + 'static) {
90 wasm_bindgen_futures::spawn_local(async move {
91 fut.await;
92 });
93}
94
95#[cfg(not(target_arch = "wasm32"))]
96fn spawn<F>(fut: F)
97where
98 F: Future + Send + 'static,
99 F::Output: Send + 'static,
100{
101 nagoya::spawn(fut);
102}
103
104pub struct Provider {
110 client: Client,
111 waker: Arc<dyn NetWaker>,
112 user_agent: Arc<str>,
113 #[cfg(all(feature = "cache", not(target_arch = "wasm32")))]
114 cache_manager: Option<Arc<PrivateCache>>,
115}
116impl Provider {
117 pub fn new(waker: Option<Arc<dyn NetWaker>>) -> Self {
118 Self::with_user_agent(waker, DEFAULT_USER_AGENT)
119 }
120
121 pub fn with_user_agent(waker: Option<Arc<dyn NetWaker>>, user_agent: &str) -> Self {
129 Self::with_cookie_jar(waker, user_agent, None)
130 }
131
132 #[cfg(feature = "cookies")]
137 pub fn with_user_agent_and_cookie_provider<C>(
138 waker: Option<Arc<dyn NetWaker>>,
139 user_agent: &str,
140 cookie_provider: Arc<C>,
141 ) -> Self
142 where
143 C: CookieJar,
144 {
145 Self::with_cookie_jar(waker, user_agent, Some(cookie_provider))
146 }
147
148 fn with_cookie_jar(
149 waker: Option<Arc<dyn NetWaker>>,
150 user_agent: &str,
151 cookies: Option<Arc<dyn CookieStorage>>,
152 ) -> Self {
153 let client =
154 Client::new(user_agent, cookies).expect("failed to initialize the house HTTP client");
155 let waker = waker.unwrap_or(Arc::new(DummyNetWaker));
156 Self {
157 client,
158 waker,
159 user_agent: Arc::from(user_agent),
160 #[cfg(all(feature = "cache", not(target_arch = "wasm32")))]
161 cache_manager: None,
162 }
163 }
164
165 #[cfg(all(feature = "cache", not(target_arch = "wasm32")))]
170 pub fn with_cache(mut self, cache: Arc<PrivateCache>) -> Self {
171 self.client = self.client.with_cache(cache.clone());
172 self.cache_manager = Some(cache);
173 self
174 }
175
176 #[cfg(all(feature = "cache", not(target_arch = "wasm32")))]
181 pub async fn with_cache_directory(
182 self,
183 profile_directory: impl AsRef<std::path::Path>,
184 ) -> std::io::Result<Self> {
185 let directory = profile_directory.as_ref().to_owned();
186 let cache = nagoya::spawn(async move { PrivateCache::open(directory).await })
187 .await
188 .ok_or_else(|| std::io::Error::other("cache opener was cancelled"))??;
189 Ok(self.with_cache(cache))
190 }
191
192 pub fn shared(waker: Option<Arc<dyn NetWaker>>) -> Arc<dyn NetProvider> {
193 Arc::new(Self::new(waker))
194 }
195 pub fn shared_with_user_agent(
196 waker: Option<Arc<dyn NetWaker>>,
197 user_agent: &str,
198 ) -> Arc<dyn NetProvider> {
199 Arc::new(Self::with_user_agent(waker, user_agent))
200 }
201 pub fn user_agent(&self) -> &str {
203 &self.user_agent
204 }
205 pub fn is_empty(&self) -> bool {
206 Arc::strong_count(&self.waker) == 1
207 }
208 pub fn count(&self) -> usize {
209 Arc::strong_count(&self.waker) - 1
210 }
211
212 #[cfg(all(feature = "cache", not(target_arch = "wasm32")))]
213 pub async fn clear_cache(&self) {
214 if let Some(cache) = &self.cache_manager {
215 let cache = cache.clone();
216 let result = nagoya::spawn(async move { cache.clear().await }).await;
217 if let Some(Err(e)) = result {
218 #[cfg(feature = "tracing")]
219 tracing::error!("Failed to clear HTTP cache: {:?}", e);
220 #[cfg(not(feature = "tracing"))]
221 let _ = e;
222 }
223 }
224 }
225}
226impl Provider {
227 async fn fetch_inner(
228 client: Client,
229 request: Request,
230 ) -> Result<(String, Bytes), ProviderError> {
231 let response = Self::fetch_response_inner(client, request).await?;
232 Ok((response.url.to_string(), response.body))
233 }
234
235 #[allow(clippy::type_complexity)]
237 pub fn fetch_with_callback(
238 &self,
239 request: Request,
240 callback: Box<dyn FnOnce(Result<(String, Bytes), ProviderError>) + Send + Sync + 'static>,
241 ) {
242 #[cfg(feature = "tracing")]
243 let url = request.url.to_string();
244
245 let client = self.client.clone();
246 let signal = request.signal.clone();
247 spawn(async move {
248 let result = Self::fetch_inner(client, request).await;
249 if matches!(&result, Err(ProviderError::Abort))
250 || signal.as_ref().is_some_and(|signal| signal.aborted())
251 {
252 return;
253 }
254
255 #[cfg(feature = "tracing")]
256 if let Err(e) = &result {
257 tracing::error!(url = url.as_str(), error = ?e, "Fetching");
258 } else {
259 tracing::info!(url = url.as_str(), "Success fetching");
260 }
261
262 callback(result);
263 });
264 }
265
266 pub async fn fetch_async(&self, request: Request) -> Result<(String, Bytes), ProviderError> {
267 #[cfg(feature = "tracing")]
268 let url = request.url.to_string();
269
270 let result = Self::fetch_inner(self.client.clone(), request).await;
271
272 #[cfg(feature = "tracing")]
273 if let Err(e) = &result {
274 tracing::error!(url = url.as_str(), error = ?e, "Fetching");
275 } else {
276 tracing::info!(url = url.as_str(), "Success fetching");
277 }
278
279 result
280 }
281
282 pub async fn fetch_response_async(
289 &self,
290 request: Request,
291 ) -> Result<FetchResponse, ProviderError> {
292 Self::fetch_response_inner(self.client.clone(), request).await
293 }
294
295 async fn fetch_response_inner(
296 client: Client,
297 request: Request,
298 ) -> Result<FetchResponse, ProviderError> {
299 abort::until_cancelled(
300 request.signal.clone(),
301 Self::fetch_response_uncancelled(client, request),
302 )
303 .await
304 .unwrap_or(Err(ProviderError::Abort))
305 }
306
307 async fn fetch_response_uncancelled(
308 client: Client,
309 request: Request,
310 ) -> Result<FetchResponse, ProviderError> {
311 let url = request.url.clone();
312 match url.scheme() {
313 "data" => {
314 let (body, headers) = {
317 let data_url = DataUrl::process(url.as_str())?;
318 let decoded = data_url.decode_to_vec()?;
319 let mut headers = HeaderMap::new();
320 if let Ok(value) = data_url.mime_type().to_string().parse() {
321 headers.insert(CONTENT_TYPE, value);
322 }
323 (Bytes::from(decoded.0), headers)
324 };
325 Ok(FetchResponse::new(url, StatusCode::OK)
326 .headers(headers)
327 .body(body))
328 }
329 "file" => {
330 let file_content = std::fs::read(url.path())?;
331 Ok(FetchResponse::new(url, StatusCode::OK).body(Bytes::from(file_content)))
332 }
333 "http" | "https" => {
334 let mut wire_request = FetchRequest::get(request.url).method(request.method);
335 wire_request.headers = request.headers;
336 wire_request.signal = request.signal;
337 let content_type = request.content_type.or_else(|| {
338 wire_request
339 .headers
340 .get(CONTENT_TYPE)
341 .and_then(|value| value.to_str().ok())
342 .map(str::to_owned)
343 });
344 if let Some(content_type) = &content_type {
345 let value = HeaderValue::from_str(content_type)
346 .map_err(|error| ProviderError::InvalidRequest(error.to_string()))?;
347 wire_request.headers.insert(CONTENT_TYPE, value);
348 }
349 wire_request.body = encode_body(
350 request.body,
351 content_type.as_deref(),
352 &mut wire_request.headers,
353 )?;
354 let response = client.fetch(wire_request).await.map_err(|error| {
355 #[cfg(not(target_arch = "wasm32"))]
356 if let h1::Error::TlsPolicy(reason) = &error {
357 return ProviderError::TlsPolicy(*reason);
358 }
359 ProviderError::Network(error.to_string())
360 })?;
361
362 if !response.status.is_success() {
363 #[cfg(feature = "tracing")]
364 tracing::warn!(
365 url = response.url.as_str(),
366 status = response.status.as_u16(),
367 "HTTP error status"
368 );
369 return Err(ProviderError::HttpStatus {
370 status: response.status,
371 url: response.url.to_string(),
372 });
373 }
374 Ok(response)
375 }
376 scheme => Err(ProviderError::UnsupportedScheme(scheme.to_owned())),
377 }
378 }
379}
380
381impl Provider {
389 pub async fn fetch_platform_async(
394 &self,
395 request: FetchRequest,
396 ) -> Result<FetchResponse, FetchError> {
397 Self::platform_fetch_inner(self.client.clone(), request).await
398 }
399
400 async fn platform_fetch_inner(
401 client: Client,
402 request: FetchRequest,
403 ) -> Result<FetchResponse, FetchError> {
404 abort::until_cancelled(
405 request.signal.clone(),
406 Self::platform_fetch_uncancelled(client, request),
407 )
408 .await
409 .unwrap_or(Err(FetchError::Abort))
410 }
411
412 async fn platform_fetch_uncancelled(
413 client: Client,
414 request: FetchRequest,
415 ) -> Result<FetchResponse, FetchError> {
416 match request.url.scheme() {
417 "data" => Self::platform_fetch_data(request),
418 "file" => Self::platform_fetch_file(request),
419 "http" | "https" => client.fetch(request).await.map_err(|error| {
420 #[cfg(not(target_arch = "wasm32"))]
421 if let h1::Error::TlsPolicy(reason) = &error {
422 return FetchError::TlsPolicy(*reason);
423 }
424 FetchError::Network(error.to_string())
425 }),
426 scheme => Err(FetchError::UnsupportedScheme(scheme.to_owned())),
427 }
428 }
429
430 fn platform_fetch_data(request: FetchRequest) -> Result<FetchResponse, FetchError> {
437 let data_url = DataUrl::process(request.url.as_str())
438 .map_err(|err| FetchError::InvalidRequest(format!("{err:?}")))?;
439 let mime = data_url.mime_type().to_string();
440 let (body, _) = data_url
441 .decode_to_vec()
442 .map_err(|err| FetchError::InvalidRequest(format!("{err:?}")))?;
443
444 let mut headers = HeaderMap::new();
445 if let Ok(value) = mime.parse() {
446 headers.insert(CONTENT_TYPE, value);
447 }
448
449 Ok(FetchResponse::new(request.url, StatusCode::OK)
450 .headers(headers)
451 .body(Bytes::from(body)))
452 }
453
454 fn platform_fetch_file(request: FetchRequest) -> Result<FetchResponse, FetchError> {
468 let path = request.url.to_file_path().map_err(|()| {
469 FetchError::InvalidRequest(format!("not a local path: {}", request.url))
470 })?;
471
472 let body = std::fs::read(path).map_err(|err| FetchError::Network(err.to_string()))?;
473
474 Ok(FetchResponse::new(request.url, StatusCode::OK).body(Bytes::from(body)))
475 }
476}
477
478impl FetchProvider for Provider {
479 fn fetch(&self, request: FetchRequest, handler: Box<dyn FetchHandler>) {
480 let client = self.client.clone();
481 let signal = request.signal.clone();
482
483 #[cfg(feature = "tracing")]
484 let url = request.url.to_string();
485
486 spawn(async move {
487 let result = Self::platform_fetch_inner(client, request).await;
488 if matches!(&result, Err(FetchError::Abort))
489 || signal.as_ref().is_some_and(|signal| signal.aborted())
490 {
491 return;
492 }
493
494 #[cfg(feature = "tracing")]
495 match &result {
496 Ok(response) => tracing::info!(
497 url = url.as_str(),
498 status = response.status.as_u16(),
499 "fetch complete"
500 ),
501 Err(error) => tracing::error!(url = url.as_str(), error = ?error, "fetch failed"),
502 }
503
504 handler.complete(result);
505 });
506 }
507}
508
509impl NetProvider for Provider {
510 fn fetch(&self, doc_id: usize, request: Request, handler: Box<dyn NetHandler>) {
511 let client = self.client.clone();
512
513 #[cfg(feature = "tracing")]
514 tracing::info!(url = request.url.as_str(), "Fetching");
515
516 let signal = request.signal.clone();
517 let waker = self.waker.clone();
518 spawn(async move {
519 #[cfg(feature = "tracing")]
520 let url = request.url.to_string();
521
522 let result = Self::fetch_inner(client, request).await;
523 if matches!(&result, Err(ProviderError::Abort))
524 || signal.as_ref().is_some_and(|signal| signal.aborted())
525 {
526 return;
527 }
528
529 match result {
530 Ok((response_url, bytes)) => {
531 handler.bytes(response_url, bytes);
532 #[cfg(feature = "tracing")]
533 tracing::info!(url = url.as_str(), "Success fetching");
534 }
535 Err(e) => {
536 #[cfg(feature = "tracing")]
537 tracing::error!(url = url.as_str(), error = ?e, "Error fetching");
538 #[cfg(not(feature = "tracing"))]
539 let _ = e;
540 }
541 };
542 waker.wake(doc_id);
544 });
545 }
546}
547
548#[derive(Debug)]
549pub enum ProviderError {
550 Abort,
551 Io(std::io::Error),
552 DataUrl(data_url::DataUrlError),
553 DataUrlBase64(data_url::forgiving_base64::InvalidBase64),
554 Network(String),
555 TlsPolicy(TlsPolicyError),
556 InvalidRequest(String),
557 UnsupportedScheme(String),
558 HttpStatus { status: StatusCode, url: String },
559}
560
561impl std::fmt::Display for ProviderError {
562 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
563 match self {
564 Self::Abort => write!(f, "request aborted"),
565 Self::Io(e) => write!(f, "io error: {e}"),
566 Self::DataUrl(e) => write!(f, "data url error: {e:?}"),
567 Self::DataUrlBase64(e) => write!(f, "data url base64 error: {e:?}"),
568 Self::Network(e) => write!(f, "network error: {e}"),
569 Self::TlsPolicy(reason) => write!(f, "TLS policy: {reason}"),
570 Self::InvalidRequest(e) => write!(f, "invalid request: {e}"),
571 Self::UnsupportedScheme(scheme) => write!(f, "unsupported URL scheme: {scheme}"),
572 Self::HttpStatus { status, url } => write!(f, "HTTP {status} for {url}"),
573 }
574 }
575}
576
577impl std::error::Error for ProviderError {}
578
579impl From<std::io::Error> for ProviderError {
580 fn from(value: std::io::Error) -> Self {
581 Self::Io(value)
582 }
583}
584
585impl From<data_url::DataUrlError> for ProviderError {
586 fn from(value: data_url::DataUrlError) -> Self {
587 Self::DataUrl(value)
588 }
589}
590
591impl From<data_url::forgiving_base64::InvalidBase64> for ProviderError {
592 fn from(value: data_url::forgiving_base64::InvalidBase64) -> Self {
593 Self::DataUrlBase64(value)
594 }
595}
596
597fn encode_body(
598 body: Body,
599 content_type: Option<&str>,
600 headers: &mut HeaderMap,
601) -> Result<Option<Bytes>, ProviderError> {
602 match body {
603 Body::Bytes(bytes) => Ok(Some(bytes)),
604 Body::Empty => Ok(None),
605 Body::Form(form) => {
606 let mime = content_type
607 .and_then(|value| value.split(';').next())
608 .map(str::trim)
609 .unwrap_or("");
610 if mime.eq_ignore_ascii_case("application/x-www-form-urlencoded") {
611 let mut encoded = url::form_urlencoded::Serializer::new(String::new());
612 for entry in &form.0 {
613 encoded.append_pair(&entry.name, entry.value.as_ref());
614 }
615 return Ok(Some(Bytes::from(encoded.finish())));
616 }
617 if mime.eq_ignore_ascii_case("multipart/form-data") {
618 #[cfg(all(feature = "multipart", not(target_arch = "wasm32")))]
619 {
620 return encode_multipart(form, headers).map(Some);
621 }
622 #[cfg(not(all(feature = "multipart", not(target_arch = "wasm32"))))]
623 {
624 let _ = headers;
625 return Err(ProviderError::InvalidRequest(
626 "multipart requires the native multipart feature".to_owned(),
627 ));
628 }
629 }
630 Err(ProviderError::InvalidRequest(
631 "unsupported form Content-Type".to_owned(),
632 ))
633 }
634 }
635}
636
637#[cfg(all(feature = "multipart", not(target_arch = "wasm32")))]
638fn encode_multipart(
639 form: blitz_traits::net::FormData,
640 headers: &mut HeaderMap,
641) -> Result<Bytes, ProviderError> {
642 use blitz_traits::net::{Entry, EntryValue};
643 use psc_nanoid::{Nanoid, alphabet::Base64UrlAlphabet};
644 use std::io::Read;
645
646 let boundary = Nanoid::<21, Base64UrlAlphabet>::new().to_string();
647 let mut encoded = Vec::new();
648 for Entry { name, value } in form.0 {
649 let mut head = format!(
650 "--{boundary}\r\nContent-Disposition: form-data; name=\"{}\"",
651 multipart_parameter(&name),
652 );
653 let body = match value {
654 EntryValue::String(value) => Bytes::from(value),
655 EntryValue::File(path) => {
656 let name = path.file_name().unwrap_or_default().to_string_lossy();
657 head.push_str(&format!(
658 "; filename=\"{}\"\r\nContent-Type: application/octet-stream",
659 multipart_parameter(&name),
660 ));
661 let file = std::fs::File::open(path)?;
662 let mut body = Vec::new();
663 file.take(h1::MAX_BODY_BYTES as u64 + 1)
664 .read_to_end(&mut body)?;
665 if body.len() > h1::MAX_BODY_BYTES {
666 return Err(ProviderError::InvalidRequest(
667 "multipart file exceeds the body limit".to_owned(),
668 ));
669 }
670 Bytes::from(body)
671 }
672 EntryValue::EmptyFile => {
673 head.push_str("; filename=\"\"\r\nContent-Type: application/octet-stream");
674 Bytes::new()
675 }
676 };
677 head.push_str("\r\n\r\n");
678 let required = head
679 .len()
680 .checked_add(body.len())
681 .and_then(|length| length.checked_add(2))
682 .and_then(|length| length.checked_add(encoded.len()))
683 .ok_or_else(|| {
684 ProviderError::InvalidRequest("multipart body is too large".to_owned())
685 })?;
686 if required > h1::MAX_BODY_BYTES {
687 return Err(ProviderError::InvalidRequest(
688 "multipart body exceeds the body limit".to_owned(),
689 ));
690 }
691 encoded.extend_from_slice(head.as_bytes());
692 encoded.extend_from_slice(&body);
693 encoded.extend_from_slice(b"\r\n");
694 }
695 let end = format!("--{boundary}--\r\n");
696 if end.len() > h1::MAX_BODY_BYTES.saturating_sub(encoded.len()) {
697 return Err(ProviderError::InvalidRequest(
698 "multipart body exceeds the body limit".to_owned(),
699 ));
700 }
701 encoded.extend_from_slice(end.as_bytes());
702 let content_type = HeaderValue::from_str(&format!("multipart/form-data; boundary={boundary}"))
703 .map_err(|error| ProviderError::InvalidRequest(error.to_string()))?;
704 headers.insert(CONTENT_TYPE, content_type);
705 Ok(Bytes::from(encoded))
706}
707
708#[cfg(all(feature = "multipart", not(target_arch = "wasm32")))]
709fn multipart_parameter(value: &str) -> String {
710 value
711 .replace('\r', "%0D")
712 .replace('\n', "%0A")
713 .replace('"', "%22")
714}
715
716struct DummyNetWaker;
717impl NetWaker for DummyNetWaker {
718 fn wake(&self, _client_id: usize) {}
719}