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