wasm_pkg_client/
lib.rs

1//! Wasm Package Client
2//!
3//! [`Client`] implements a unified interface for loading package content from
4//! multiple kinds of package registries.
5//!
6//! # Example
7//!
8//! ```no_run
9//! # async fn example() -> anyhow::Result<()> {
10//! // Initialize client from global configuration.
11//! let mut client = wasm_pkg_client::Client::with_global_defaults().await?;
12//!
13//! // Get a specific package release version.
14//! let pkg = "example:pkg".parse()?;
15//! let version = "1.0.0".parse()?;
16//! let release = client.get_release(&pkg, &version).await?;
17//!
18//! // Stream release content to a file.
19//! let mut stream = client.stream_content(&pkg, &release).await?;
20//! let mut file = tokio::fs::File::create("output.wasm").await?;
21//! use futures_util::TryStreamExt;
22//! use tokio::io::AsyncWriteExt;
23//! while let Some(chunk) = stream.try_next().await? {
24//!     file.write_all(&chunk).await?;
25//! }
26//! # Ok(()) }
27//! ```
28
29pub mod caching;
30mod decoded_component;
31mod loader;
32pub mod local;
33pub mod metadata;
34pub mod oci;
35mod publisher;
36mod release;
37
38use std::{cmp::Ordering, collections::HashMap, path::Path, pin::Pin, sync::Arc};
39
40use anyhow::anyhow;
41use bytes::Bytes;
42use decoded_component::DecodedComponent;
43use futures_util::Stream;
44use tokio::sync::RwLock;
45use wasm_pkg_common::metadata::{LOCAL_PROTOCOL, OCI_PROTOCOL};
46pub use wasm_pkg_common::{
47    Error,
48    config::{Config, CustomConfig, RegistryMapping},
49    digest::ContentDigest,
50    metadata::RegistryMetadata,
51    package::{PackageRef, Version, VersionReq},
52    registry::Registry,
53};
54
55use crate::loader::VersionSort;
56use crate::local::LocalBackend;
57use crate::metadata::RegistryMetadataExt;
58use crate::oci::OciBackend;
59pub use crate::{loader::PackageLoader, publisher::PackagePublisher};
60
61pub use release::{Release, VersionInfo};
62
63/// An alias for a stream of content bytes
64pub type ContentStream = Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send + 'static>>;
65
66/// An alias for a PublishingSource (generally a file)
67pub type PublishingSource = Pin<Box<dyn ReaderSeeker + Send + Sync + 'static>>;
68
69/// A supertrait combining tokio's AsyncRead and AsyncSeek.
70pub trait ReaderSeeker: tokio::io::AsyncRead + tokio::io::AsyncSeek {}
71
72impl<T> ReaderSeeker for T where T: tokio::io::AsyncRead + tokio::io::AsyncSeek {}
73
74trait LoaderPublisher: PackageLoader + PackagePublisher {}
75
76impl<T> LoaderPublisher for T where T: PackageLoader + PackagePublisher {}
77
78type RegistrySources = HashMap<Registry, Arc<InnerClient>>;
79type InnerClient = Box<dyn LoaderPublisher + Sync>;
80
81/// Additional options for publishing a package.
82#[derive(Clone, Debug, Default)]
83pub struct PublishOpts {
84    /// Override the package name and version to publish with.
85    pub package: Option<(PackageRef, Version)>,
86    /// Override the registry to publish to.
87    pub registry: Option<Registry>,
88    /// If true, resolve the package, version, and registry but do not call the
89    /// backend to publish.
90    pub dry_run: bool,
91    /// Disable semver compatibility verification.
92    pub skip_semver_check: bool,
93}
94
95/// A read-only registry client.
96#[derive(Clone)]
97pub struct Client {
98    config: Arc<Config>,
99    sources: Arc<RwLock<RegistrySources>>,
100}
101
102impl Client {
103    /// Returns a new client with the given [`Config`].
104    pub fn new(config: Config) -> Self {
105        Self {
106            config: Arc::new(config),
107            sources: Default::default(),
108        }
109    }
110
111    /// Returns a reference to the configuration this client was initialized with.
112    pub fn config(&self) -> &Config {
113        &self.config
114    }
115
116    /// Returns a new client configured from default global config.
117    pub async fn with_global_defaults() -> Result<Self, Error> {
118        let config = Config::global_defaults().await?;
119        Ok(Self::new(config))
120    }
121
122    /// Returns a list of all package [`Version`]s available for the given package.
123    pub async fn list_all_versions(&self, package: &PackageRef) -> Result<Vec<VersionInfo>, Error> {
124        let source = self.resolve_source(package, None).await?;
125        source.list_all_versions(package).await
126    }
127
128    /// Returns a [`Release`] for the given package version.
129    pub async fn get_release(
130        &self,
131        package: &PackageRef,
132        version: &Version,
133    ) -> Result<Release, Error> {
134        // FIXME: the `None` below means we ignore workspace overrides for this call
135        let source = self.resolve_source(package, None).await?;
136        source.get_release(package, version).await
137    }
138
139    /// Returns a [`ContentStream`] of content chunks. Contents are validated
140    /// against the given [`Release::content_digest`].
141    pub async fn stream_content<'a>(
142        &'a self,
143        package: &'a PackageRef,
144        release: &'a Release,
145    ) -> Result<ContentStream, Error> {
146        let source = self.resolve_source(package, None).await?;
147        source.stream_content(package, release).await
148    }
149
150    /// Publishes the given file as a package release. The package name and version will be read
151    /// from the component if not given as part of `additional_options`. Returns the package name
152    /// and version of the published release.
153    pub async fn publish_release_file(
154        &self,
155        file: impl AsRef<Path>,
156        additional_options: PublishOpts,
157    ) -> Result<(PackageRef, Version), Error> {
158        let data = tokio::fs::OpenOptions::new().read(true).open(file).await?;
159
160        self.publish_release_data(Box::pin(data), additional_options)
161            .await
162    }
163
164    /// Publishes the given reader as a package release. The package name and version will be read
165    /// from the component if not given as part of `additional_options`. Returns the package name
166    /// and version of the published release.
167    pub async fn publish_release_data(
168        &self,
169        data: PublishingSource,
170        additional_options: PublishOpts,
171    ) -> Result<(PackageRef, Version), Error> {
172        // handle opts
173        let registry = additional_options.registry;
174        let semver_check: bool = additional_options.skip_semver_check;
175        let pkg_authority = additional_options.package;
176
177        // construct verifiable publishing source
178        let (data, candidate) =
179            DecodedComponent::from_publishing_source_with_package(data, pkg_authority).await?;
180
181        let (package, version) = (
182            candidate.package().to_owned(),
183            candidate.version().to_owned(),
184        );
185        let source = self.resolve_source(&package, registry).await?;
186
187        // execute pre-flight checks
188        if !semver_check {
189            // fetch nearest neighbors of interest, sorted in descending order
190            let mut neighbors: [Option<VersionInfo>; 2] = [None, None];
191            for version_info in
192                fetch_semver_series(source.as_ref().as_ref(), &package, &version).await?
193            {
194                match version.cmp(&version_info.version) {
195                    Ordering::Equal => {
196                        return Err(Error::VersionAlreadyExists(
197                            package.clone(),
198                            version.to_owned(),
199                        ));
200                    }
201                    Ordering::Greater => {
202                        // incoming version is greater than neighbor
203                        neighbors[0] = Some(version_info);
204                        break;
205                    }
206                    Ordering::Less => {
207                        // incoming version is lesser than neighbor
208                        neighbors[1] = Some(version_info);
209                    }
210                }
211            }
212
213            // queue up load/decode futures
214            let prepare_neighbor_ops: Vec<_> = neighbors
215                .into_iter()
216                .flatten()
217                .map(|v| fetch_and_resolve_package(&**source, &package, v.version))
218                .collect();
219
220            // execute load/decode ops, collect results.
221            let mut semver_series: Vec<decoded_component::DecodedComponent> =
222                futures_util::future::join_all(prepare_neighbor_ops)
223                    .await
224                    .into_iter()
225                    .collect::<Result<_, _>>()?;
226
227            // verify candidate is in compliance with its semver neighbors
228            if !semver_series.is_empty() {
229                semver_series.push(candidate);
230
231                semver_series.sort();
232                for window in semver_series.windows(2) {
233                    let [prev, next] = window else { unreachable!() };
234                    prev.semver_check(next)?;
235                }
236            }
237        }
238
239        source
240            .publish(&package, &version, data, additional_options.dry_run)
241            .await
242            .map(|_| (package, version))
243    }
244
245    fn resolve_registry(
246        &self,
247        package: &PackageRef,
248        registry_override: Option<Registry>,
249    ) -> Result<Registry, Error> {
250        if let Some(registry) = registry_override {
251            return Ok(registry);
252        }
253        self.config
254            .resolve_registry(package)
255            .cloned()
256            .ok_or_else(|| Error::NoRegistryForNamespace(package.namespace().clone()))
257    }
258
259    async fn resolve_source(
260        &self,
261        package: &PackageRef,
262        registry_override: Option<Registry>,
263    ) -> Result<Arc<InnerClient>, Error> {
264        let is_override = registry_override.is_some();
265        let registry = self.resolve_registry(package, registry_override)?;
266        tracing::debug!(?registry, "resolved registry");
267
268        if let Some(source) = self.sources.read().await.get(&registry) {
269            return Ok(source.clone());
270        }
271
272        let registry_config = self
273            .config
274            .registry_config(&registry)
275            .cloned()
276            .unwrap_or_default();
277
278        let maybe_metadata = self
279            .config
280            .package_registry_override(package)
281            .and_then(|mapping| match mapping {
282                RegistryMapping::Custom(custom) => Some(custom.metadata.clone()),
283                _ => None,
284            })
285            .or_else(|| {
286                self.config
287                    .namespace_registry(package.namespace())
288                    .and_then(|meta| {
289                        // If the overridden registry matches the registry we are trying to resolve, we
290                        // should use the metadata, otherwise we'll need to fetch the metadata from the
291                        // registry
292                        match (meta, is_override) {
293                            (RegistryMapping::Custom(custom), true)
294                                if custom.registry == registry =>
295                            {
296                                Some(custom.metadata.clone())
297                            }
298                            (RegistryMapping::Custom(custom), false) => {
299                                Some(custom.metadata.clone())
300                            }
301                            _ => None,
302                        }
303                    })
304            });
305
306        let registry_meta = if let Some(meta) = maybe_metadata {
307            meta
308        } else if registry_config.default_backend() == LOCAL_PROTOCOL.into() {
309            // Skip fetching metadata for "local" source
310            RegistryMetadata::default()
311        } else {
312            RegistryMetadata::fetch_or_default(&registry).await
313        };
314
315        // Resolve backend type
316        let backend_type = match registry_config.default_backend() {
317            // If the local config specifies a backend type, use it
318            Some(backend_type) => Some(backend_type),
319            None => {
320                // If the registry metadata indicates a preferred protocol, use it
321                let preferred_protocol = registry_meta.preferred_protocol();
322                // ...except registry metadata cannot force a local backend
323                if preferred_protocol == Some(LOCAL_PROTOCOL) {
324                    return Err(Error::InvalidRegistryMetadata(anyhow!(
325                        "registry metadata with 'local' protocol not allowed"
326                    )));
327                }
328                preferred_protocol
329            }
330        }
331        // Otherwise use the default backend
332        .unwrap_or(OCI_PROTOCOL);
333        tracing::debug!(?backend_type, "Resolved backend type");
334
335        let source: InnerClient = match backend_type {
336            LOCAL_PROTOCOL => Box::new(LocalBackend::new(registry_config)?),
337            OCI_PROTOCOL => Box::new(OciBackend::new(
338                &registry,
339                &registry_config,
340                &registry_meta,
341            )?),
342            other => {
343                return Err(Error::InvalidConfig(anyhow!(
344                    "unknown backend type {other:?}"
345                )));
346            }
347        };
348        let source = Arc::new(source);
349        self.sources
350            .write()
351            .await
352            .insert(registry.clone(), source.clone());
353
354        Ok(source)
355    }
356}
357
358// Fetch every prior release in the same semver compatibility series as
359// `version`, sorted in descending order.
360//
361//   X.y.z (X >= 1) -> X.*       (minors are additive within a major)
362//   0.Y.z (Y >= 1) -> 0.Y.*     (in 0.x, minor bumps are breaking)
363//   0.0.Z          -> 0.0.Z     (every patch is its own series)
364async fn fetch_semver_series(
365    source: &(dyn LoaderPublisher + Sync),
366    package: &PackageRef,
367    version: &Version,
368) -> Result<Vec<VersionInfo>, Error> {
369    let mask = if version.major > 0 {
370        format!("{}.*", version.major)
371    } else if version.minor > 0 {
372        format!("0.{}.*", version.minor)
373    } else {
374        version.to_string()
375    };
376    let req = VersionReq::parse(&mask)
377        .map_err(|e| Error::InvalidConfig(anyhow!("invalid version mask: {e}")))?;
378
379    source
380        .list_matching_versions(package, req, VersionSort::Descending)
381        .await
382}
383
384async fn fetch_and_resolve_package(
385    source: &(dyn LoaderPublisher + Sync),
386    package: &PackageRef,
387    version: Version,
388) -> Result<decoded_component::DecodedComponent, Error> {
389    let stream = source
390        .stream_content(package, &source.get_release(package, &version).await?)
391        .await
392        .map_err(std::io::Error::other)?;
393
394    DecodedComponent::from_content_stream(stream, package.clone(), version).await
395}
wasm_pkg_client/lib.rs

wasm_pkg_client/
lib.rs
