wasm_pkg_client/
local.rs

1//! Local filesystem-based package backend.
2//!
3//! Each package release is a file: `<root-dir>/<namespace>/<name>/<version>.wasm`
4
5use std::{
6    io,
7    path::{Path, PathBuf},
8    sync::Arc,
9};
10
11use anyhow::anyhow;
12use async_trait::async_trait;
13use futures_util::{StreamExt, TryStreamExt};
14use serde::{Deserialize, Serialize};
15use sha2::{Digest, Sha256};
16use tempfile::TempDir;
17use tokio_util::io::ReaderStream;
18use wasm_pkg_common::{
19    Error,
20    config::RegistryConfig,
21    digest::ContentDigest,
22    metadata::LOCAL_PROTOCOL,
23    package::{PackageRef, Version},
24};
25
26use crate::{
27    ContentStream, PublishingSource,
28    loader::PackageLoader,
29    publisher::PackagePublisher,
30    release::{Release, VersionInfo},
31};
32
33#[derive(Clone, Debug, Deserialize, Serialize)]
34pub struct LocalConfig {
35    pub root: PathBuf,
36    // NOTE: set by [`Self::temp_dir`] to avoid holding onto a separate `TempDir` handle.
37    #[serde(skip)]
38    #[doc(hidden)]
39    _temp_handle: Option<Arc<TempDir>>,
40}
41
42impl LocalConfig {
43    /// Creates a [`Self`] with a new temporary directory.
44    /// The returned config owns the directory and removes the config upon drop.
45    pub fn temp_dir() -> Result<Self, Error> {
46        let handle = TempDir::new()?;
47        let root = handle.path().to_path_buf();
48        tracing::debug!(registry_dir = %root.display(), "created temporary directory");
49        Ok(Self {
50            root,
51            _temp_handle: Some(Arc::new(handle)),
52        })
53    }
54}
55
56#[derive(Clone)]
57pub(crate) struct LocalBackend {
58    pub(crate) root: PathBuf,
59}
60
61fn registry_path_context(err: io::Error, path: &Path) -> Error {
62    let err = anyhow::Error::new(err).context(format!("path: {}", path.display()));
63    Error::RegistryError(err)
64}
65
66impl LocalBackend {
67    pub fn new(registry_config: RegistryConfig) -> Result<Self, Error> {
68        let config = registry_config
69            .backend_config::<LocalConfig>(LOCAL_PROTOCOL)?
70            .ok_or_else(|| {
71                Error::InvalidConfig(anyhow!("'local' backend requires configuration"))
72            })?;
73        Ok(Self { root: config.root })
74    }
75
76    fn package_dir(&self, package: &PackageRef) -> PathBuf {
77        self.root
78            .join(package.namespace().as_ref())
79            .join(package.name().as_ref())
80    }
81
82    fn version_path(&self, package: &PackageRef, version: &Version) -> PathBuf {
83        self.package_dir(package).join(format!("{version}.wasm"))
84    }
85}
86
87#[async_trait]
88impl PackageLoader for LocalBackend {
89    async fn list_all_versions(&self, package: &PackageRef) -> Result<Vec<VersionInfo>, Error> {
90        let mut versions = vec![];
91        let package_dir = self.package_dir(package);
92        tracing::debug!(?package_dir, "Reading versions from path");
93
94        let mut entries = match tokio::fs::read_dir(&package_dir).await {
95            Ok(entries) => entries,
96            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
97                return Err(Error::PackageNotFound);
98            }
99            Err(e) => return Err(registry_path_context(e, &package_dir)),
100        };
101        while let Some(entry) = entries.next_entry().await? {
102            let path = entry.path();
103            if path.extension() != Some("wasm".as_ref()) {
104                continue;
105            }
106            let Some(version) = path
107                .file_stem()
108                .unwrap()
109                .to_str()
110                .and_then(|stem| Version::parse(stem).ok())
111            else {
112                tracing::warn!("invalid package file name at {path:?}");
113                continue;
114            };
115            versions.push(VersionInfo {
116                version,
117                yanked: false,
118            });
119        }
120        Ok(versions)
121    }
122
123    async fn get_release(&self, package: &PackageRef, version: &Version) -> Result<Release, Error> {
124        let path = self.version_path(package, version);
125        tracing::debug!(path = %path.display(), "Reading content from path");
126        let content_digest = sha256_from_file(&path)
127            .await
128            .map_err(|e| registry_path_context(e, &path))?;
129        Ok(Release {
130            version: version.clone(),
131            content_digest,
132        })
133    }
134
135    async fn stream_content_unvalidated(
136        &self,
137        package: &PackageRef,
138        content: &Release,
139    ) -> Result<ContentStream, Error> {
140        let path = self.version_path(package, &content.version);
141        tracing::debug!("Streaming content from {path:?}");
142        let file = tokio::fs::File::open(&path)
143            .await
144            .map_err(|e| registry_path_context(e, &path))?;
145        Ok(ReaderStream::new(file).map_err(Into::into).boxed())
146    }
147}
148
149#[async_trait::async_trait]
150impl PackagePublisher for LocalBackend {
151    async fn publish(
152        &self,
153        package: &PackageRef,
154        version: &Version,
155        mut data: PublishingSource,
156        dry_run: bool,
157    ) -> Result<(), Error> {
158        let package_dir = self.package_dir(package);
159        // Ensure the package directory exists.
160        tokio::fs::create_dir_all(&package_dir)
161            .await
162            .map_err(|e| registry_path_context(e, &package_dir))?;
163        let path = self.version_path(package, version);
164        if dry_run {
165            return Ok(());
166        }
167        let mut out = tokio::fs::File::create(&path)
168            .await
169            .map_err(|e| registry_path_context(e, &path))?;
170        tracing::info!("publishing to {}", path.display());
171        tokio::io::copy(&mut data, &mut out)
172            .await
173            .map_err(Error::IoError)
174            .map(|_| ())
175    }
176}
177
178async fn sha256_from_file(path: impl AsRef<Path>) -> Result<ContentDigest, std::io::Error> {
179    use tokio::io::AsyncReadExt;
180    let mut file = tokio::fs::File::open(path).await?;
181    let mut hasher = Sha256::new();
182    let mut buf = [0; 4096];
183    loop {
184        let n = file.read(&mut buf).await?;
185        if n == 0 {
186            break;
187        }
188        hasher.update(&buf[..n]);
189    }
190    Ok(hasher.into())
191}
wasm_pkg_client/local.rs

wasm_pkg_client/
local.rs
