From e1cf2363eab5d0245c73ace34c1e3a86ffa2f9a0 Mon Sep 17 00:00:00 2001 From: Myriad-Dreamin Date: Sat, 28 Dec 2024 17:16:00 +0800 Subject: [PATCH] dev: sync registry implementation from tinymist --- crates/reflexo-world/src/package/http.rs | 332 ++++++++++++++++------- 1 file changed, 241 insertions(+), 91 deletions(-) diff --git a/crates/reflexo-world/src/package/http.rs b/crates/reflexo-world/src/package/http.rs index 275a0692..36389610 100644 --- a/crates/reflexo-world/src/package/http.rs +++ b/crates/reflexo-world/src/package/http.rs @@ -1,139 +1,241 @@ -use std::{ - path::Path, - sync::{Arc, OnceLock}, -}; +//! Https registry for tinymist. + +use std::path::Path; +use std::sync::{Arc, OnceLock}; -use log::error; use parking_lot::Mutex; +use reflexo::ImmutPath; use reqwest::blocking::Response; -use typst::{ - diag::{eco_format, EcoString}, - syntax::package::PackageVersion, -}; +use reqwest::Certificate; +use typst::diag::{eco_format, EcoString, PackageResult, StrResult}; +use typst::syntax::package::{PackageVersion, VersionlessPackageSpec}; -use super::{DummyNotifier, Notifier, PackageError, PackageRegistry, PackageSpec}; +use crate::package::{DummyNotifier, Notifier, PackageError, PackageRegistry, PackageSpec}; +/// The http package registry for typst.ts. pub struct HttpRegistry { + /// The path at which local packages (`@local` packages) are stored. + package_path: Option, + /// The path at which non-local packages (`@preview` packages) should be + /// stored when downloaded. + package_cache_path: Option, + /// lazily initialized package storage. + storage: OnceLock, + /// The path to the certificate file to use for HTTPS requests. + cert_path: Option, + /// The notifier to use for progress updates. notifier: Arc>, - - packages: OnceLock)>>, + // package_dir_cache: RwLock>>, } impl Default for HttpRegistry { fn default() -> Self { Self { notifier: Arc::new(Mutex::::default()), + cert_path: None, + package_path: None, + package_cache_path: None, - // todo: reset cache - packages: OnceLock::new(), + storage: OnceLock::new(), + // package_dir_cache: RwLock::new(HashMap::new()), } } } +impl std::ops::Deref for HttpRegistry { + type Target = PackageStorage; + + fn deref(&self) -> &Self::Target { + self.storage() + } +} + impl HttpRegistry { - pub fn local_path(&self) -> Option> { - if let Some(data_dir) = dirs::data_dir() { - if data_dir.exists() { - return Some(data_dir.join("typst/packages").into()); - } + /// Create a new registry. + pub fn new( + cert_path: Option, + package_path: Option, + package_cache_path: Option, + ) -> Self { + Self { + cert_path, + package_path, + package_cache_path, + ..Default::default() } + } - None + /// Get `typst-kit` implementing package storage + pub fn storage(&self) -> &PackageStorage { + self.storage.get_or_init(|| { + PackageStorage::new( + self.package_cache_path + .clone() + .or_else(|| Some(dirs::cache_dir()?.join(DEFAULT_PACKAGES_SUBDIR).into())), + self.package_path + .clone() + .or_else(|| Some(dirs::data_dir()?.join(DEFAULT_PACKAGES_SUBDIR).into())), + self.cert_path.clone(), + self.notifier.clone(), + ) + }) } - pub fn paths(&self) -> Vec> { - let mut res = vec![]; - if let Some(data_dir) = dirs::data_dir() { - let dir: Box = data_dir.join("typst/packages").into(); - if dir.exists() { - res.push(dir); - } - } + /// Get local path option + pub fn local_path(&self) -> Option { + self.storage().package_path().cloned() + } - if let Some(cache_dir) = dirs::cache_dir() { - let dir: Box = cache_dir.join("typst/packages").into(); - if dir.exists() { - res.push(dir); - } + /// Get data & cache dir + pub fn paths(&self) -> Vec { + let data_dir = self.storage().package_path().cloned(); + let cache_dir = self.storage().package_cache_path().cloned(); + data_dir.into_iter().chain(cache_dir).collect::>() + } + + /// Set list of packages for testing. + pub fn test_package_list(&self, f: impl FnOnce() -> Vec<(PackageSpec, Option)>) { + self.storage().index.get_or_init(f); + } +} + +impl PackageRegistry for HttpRegistry { + fn resolve(&self, spec: &PackageSpec) -> Result { + self.storage().prepare_package(spec) + } + + fn packages(&self) -> &[(PackageSpec, Option)] { + self.storage().download_index() + } +} + +/// The default Typst registry. +pub const DEFAULT_REGISTRY: &str = "https://packages.typst.org"; + +/// The default packages sub directory within the package and package cache +/// paths. +pub const DEFAULT_PACKAGES_SUBDIR: &str = "typst/packages"; + +/// Holds information about where packages should be stored and downloads them +/// on demand, if possible. +pub struct PackageStorage { + /// The path at which non-local packages should be stored when downloaded. + package_cache_path: Option, + /// The path at which local packages are stored. + package_path: Option, + /// The downloader used for fetching the index and packages. + cert_path: Option, + /// The cached index of the preview namespace. + index: OnceLock)>>, + notifier: Arc>, +} + +impl PackageStorage { + /// Creates a new package storage for the given package paths. + /// It doesn't fallback directories, thus you can disable the related + /// storage by passing `None`. + pub fn new( + package_cache_path: Option, + package_path: Option, + cert_path: Option, + notifier: Arc>, + ) -> Self { + Self { + package_cache_path, + package_path, + cert_path, + notifier, + index: OnceLock::new(), } + } + + /// Returns the path at which non-local packages should be stored when + /// downloaded. + pub fn package_cache_path(&self) -> Option<&ImmutPath> { + self.package_cache_path.as_ref() + } - res + /// Returns the path at which local packages are stored. + pub fn package_path(&self) -> Option<&ImmutPath> { + self.package_path.as_ref() } /// Make a package available in the on-disk cache. - pub fn prepare_package(&self, spec: &PackageSpec) -> Result, PackageError> { - let subdir = format!( - "typst/packages/{}/{}/{}", - spec.namespace, spec.name, spec.version - ); + pub fn prepare_package(&self, spec: &PackageSpec) -> PackageResult { + let subdir = format!("{}/{}/{}", spec.namespace, spec.name, spec.version); - if let Some(data_dir) = dirs::data_dir() { - let dir = data_dir.join(&subdir); + if let Some(packages_dir) = &self.package_path { + let dir = packages_dir.join(&subdir); if dir.exists() { return Ok(dir.into()); } } - if let Some(cache_dir) = dirs::cache_dir() { + if let Some(cache_dir) = &self.package_cache_path { let dir = cache_dir.join(&subdir); + if dir.exists() { + return Ok(dir.into()); + } // Download from network if it doesn't exist yet. - if spec.namespace == "preview" && !dir.exists() { + if spec.namespace == "preview" { self.download_package(spec, &dir)?; - } - - if dir.exists() { - return Ok(dir.into()); + if dir.exists() { + return Ok(dir.into()); + } } } Err(PackageError::NotFound(spec.clone())) } - /// Download a package over the network. - fn download_package(&self, spec: &PackageSpec, package_dir: &Path) -> Result<(), PackageError> { - let url = format!( - "https://packages.typst.org/preview/{}-{}.tar.gz", - spec.name, spec.version - ); - - self.notifier.lock().downloading(spec); - threaded_http(&url, |resp| { - let reader = match resp.and_then(|r| r.error_for_status()) { - Ok(response) => response, - Err(err) if matches!(err.status().map(|s| s.as_u16()), Some(404)) => { - return Err(PackageError::NotFound(spec.clone())) - } - Err(err) => return Err(PackageError::NetworkFailed(Some(eco_format!("{err}")))), - }; - - let decompressed = flate2::read::GzDecoder::new(reader); - tar::Archive::new(decompressed) - .unpack(package_dir) - .map_err(|err| { - std::fs::remove_dir_all(package_dir).ok(); - PackageError::MalformedArchive(Some(eco_format!("{err}"))) - }) - }) - .ok_or_else(|| PackageError::Other(Some(eco_format!("cannot spawn http thread"))))? + /// Try to determine the latest version of a package. + pub fn determine_latest_version( + &self, + spec: &VersionlessPackageSpec, + ) -> StrResult { + if spec.namespace == "preview" { + // For `@preview`, download the package index and find the latest + // version. + self.download_index() + .iter() + .filter(|(package, _)| package.name == spec.name) + .map(|(package, _)| package.version) + .max() + .ok_or_else(|| eco_format!("failed to find package {spec}")) + } else { + // For other namespaces, search locally. We only search in the data + // directory and not the cache directory, because the latter is not + // intended for storage of local packages. + let subdir = format!("{}/{}", spec.namespace, spec.name); + self.package_path + .iter() + .flat_map(|dir| std::fs::read_dir(dir.join(&subdir)).ok()) + .flatten() + .filter_map(|entry| entry.ok()) + .map(|entry| entry.path()) + .filter_map(|path| path.file_name()?.to_string_lossy().parse().ok()) + .max() + .ok_or_else(|| eco_format!("please specify the desired version")) + } } -} -impl PackageRegistry for HttpRegistry { - fn resolve(&self, spec: &PackageSpec) -> Result, PackageError> { - self.prepare_package(spec) + /// Get the cached package index without network access. + pub fn cached_index(&self) -> Option<&[(PackageSpec, Option)]> { + self.index.get().map(Vec::as_slice) } - fn packages(&self) -> &[(PackageSpec, Option)] { - self.packages.get_or_init(|| { - let url = "https://packages.typst.org/preview/index.json"; + /// Download the package index. The result of this is cached for efficiency. + pub fn download_index(&self) -> &[(PackageSpec, Option)] { + self.index.get_or_init(|| { + let url = format!("{DEFAULT_REGISTRY}/preview/index.json"); - threaded_http(url, |resp| { + threaded_http(&url, self.cert_path.as_deref(), |resp| { let reader = match resp.and_then(|r| r.error_for_status()) { Ok(response) => response, Err(err) => { // todo: silent error - error!("Failed to fetch package index: {} from {}", err, url); + log::error!("Failed to fetch package index: {err} from {url}"); return vec![]; } }; @@ -145,24 +247,24 @@ impl PackageRegistry for HttpRegistry { description: Option, } - let index: Vec = match serde_json::from_reader(reader) { + let indices: Vec = match serde_json::from_reader(reader) { Ok(index) => index, Err(err) => { - error!("Failed to parse package index: {} from {}", err, url); + log::error!("Failed to parse package index: {err} from {url}"); return vec![]; } }; - index + indices .into_iter() - .map(|e| { + .map(|index| { ( PackageSpec { namespace: "preview".into(), - name: e.name, - version: e.version, + name: index.name, + version: index.version, }, - e.description, + index.description, ) }) .collect::>() @@ -170,15 +272,63 @@ impl PackageRegistry for HttpRegistry { .unwrap_or_default() }) } + + /// Download a package over the network. + /// + /// # Panics + /// Panics if the package spec namespace isn't `preview`. + pub fn download_package(&self, spec: &PackageSpec, package_dir: &Path) -> PackageResult<()> { + assert_eq!(spec.namespace, "preview"); + + let url = format!( + "{DEFAULT_REGISTRY}/preview/{}-{}.tar.gz", + spec.name, spec.version + ); + + self.notifier.lock().downloading(spec); + threaded_http(&url, self.cert_path.as_deref(), |resp| { + let reader = match resp.and_then(|r| r.error_for_status()) { + Ok(response) => response, + Err(err) if matches!(err.status().map(|s| s.as_u16()), Some(404)) => { + return Err(PackageError::NotFound(spec.clone())) + } + Err(err) => return Err(PackageError::NetworkFailed(Some(eco_format!("{err}")))), + }; + + let decompressed = flate2::read::GzDecoder::new(reader); + tar::Archive::new(decompressed) + .unpack(package_dir) + .map_err(|err| { + std::fs::remove_dir_all(package_dir).ok(); + PackageError::MalformedArchive(Some(eco_format!("{err}"))) + }) + }) + .ok_or_else(|| PackageError::Other(Some(eco_format!("cannot spawn http thread"))))? + } } fn threaded_http( url: &str, + cert_path: Option<&Path>, f: impl FnOnce(Result) -> T + Send + Sync, ) -> Option { std::thread::scope(|s| { - s.spawn(|| { - let client = reqwest::blocking::Client::builder().build().unwrap(); + s.spawn(move || { + let client_builder = reqwest::blocking::Client::builder(); + + let client = if let Some(cert_path) = cert_path { + let cert = std::fs::read(cert_path) + .ok() + .and_then(|buf| Certificate::from_pem(&buf).ok()); + if let Some(cert) = cert { + client_builder.add_root_certificate(cert).build().unwrap() + } else { + client_builder.build().unwrap() + } + } else { + client_builder.build().unwrap() + }; + f(client.get(url).send()) }) .join()