1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

use std::path::PathBuf;

use anyhow::anyhow;
use async_trait::async_trait;
use bytes::Bytes;
use futures_util::{stream::BoxStream, StreamExt, TryStreamExt};
use serde::Deserialize;
use tokio_util::io::ReaderStream;
use wasm_pkg_common::{config::RegistryConfig, package::Version};

use crate::{
    source::{PackageSource, VersionInfo},
    ContentDigest, Error, PackageRef, Release,
};

#[derive(Clone, Debug, Deserialize)]
pub struct LocalConfig {
    pub root: PathBuf,
}

/// A simple local filesystem-based PackageSource.
///
/// Each package release is a file: `<root>/<namespace>/<name>/<version>.wasm`
pub struct LocalSource {
    root: PathBuf,
}

impl LocalSource {
    pub fn new(registry_config: RegistryConfig) -> Result<Self, Error> {
        let config = registry_config
            .backend_config::<LocalConfig>("local")?
            .ok_or_else(|| {
                Error::InvalidConfig(anyhow!("'local' backend require configuration"))
            })?;
        Ok(Self { root: config.root })
    }

    fn package_dir(&self, package: &PackageRef) -> PathBuf {
        self.root
            .join(package.namespace().as_ref())
            .join(package.name().as_ref())
    }

    fn version_path(&self, package: &PackageRef, version: &Version) -> PathBuf {
        self.package_dir(package).join(format!("{version}.wasm"))
    }
}

#[async_trait]
impl PackageSource for LocalSource {
    async fn list_all_versions(&mut self, package: &PackageRef) -> Result<Vec<VersionInfo>, Error> {
        let mut versions = vec![];
        let package_dir = self.package_dir(package);
        tracing::debug!(?package_dir, "Reading versions from path");
        let mut entries = tokio::fs::read_dir(package_dir).await?;
        while let Some(entry) = entries.next_entry().await? {
            let path = entry.path();
            if path.extension() != Some("wasm".as_ref()) {
                continue;
            }
            let Some(version) = path
                .file_stem()
                .unwrap()
                .to_str()
                .and_then(|stem| Version::parse(stem).ok())
            else {
                tracing::warn!("invalid package file name at {path:?}");
                continue;
            };
            versions.push(VersionInfo {
                version,
                yanked: false,
            });
        }
        Ok(versions)
    }

    async fn get_release(
        &mut self,
        package: &PackageRef,
        version: &Version,
    ) -> Result<Release, Error> {
        let path = self.version_path(package, version);
        tracing::debug!(path = %path.display(), "Reading content from path");
        let content_digest = ContentDigest::sha256_from_file(path).await?;
        Ok(Release {
            version: version.clone(),
            content_digest,
        })
    }

    async fn stream_content_unvalidated(
        &mut self,
        package: &PackageRef,
        content: &Release,
    ) -> Result<BoxStream<Result<Bytes, Error>>, Error> {
        let path = self.version_path(package, &content.version);
        tracing::debug!("Streaming content from {path:?}");
        let file = tokio::fs::File::open(path).await?;
        Ok(ReaderStream::new(file).map_err(Into::into).boxed())
    }
}