use std::{
collections::{BTreeSet, HashMap, HashSet, hash_map},
fmt::Debug,
ops::{Deref, DerefMut},
path::{Path, PathBuf},
str::FromStr,
};
use anyhow::{Context, Result, bail};
use futures_util::TryStreamExt;
use indexmap::{IndexMap, IndexSet};
use petgraph::{Direction, acyclic::Acyclic, graph::NodeIndex, stable_graph::StableDiGraph};
use semver::{Comparator, Op, Version, VersionReq};
use tokio::io::{AsyncRead, AsyncReadExt};
use wasm_pkg_client::{
Client, Config, ContentDigest, Error as WasmPkgError, PackageRef, Release, VersionInfo,
caching::{CachingClient, FileCache},
};
use wasm_pkg_common::package::PackageSpec;
use wit_component::DecodedWasm;
use wit_parser::{PackageId, PackageName, Resolve, UnresolvedPackageGroup, WorldId};
use crate::{
lock::LockFile,
wit::{get_local_dependencies, get_packages},
};
pub const DEFAULT_REGISTRY_NAME: &str = "default";
#[derive(Debug, Clone)]
pub enum Dependency {
Package(RegistryPackage),
Local(PathBuf),
}
impl std::fmt::Display for Dependency {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Dependency::Package(RegistryPackage {
name,
version,
registry,
}) => {
let registry = registry.as_deref().unwrap_or("_");
let name = name.as_ref().map(|n| n.to_string());
write!(
f,
"{{registry=\"{registry}\" package=\"{}@{version}\"}}",
name.as_deref().unwrap_or("_:_"),
)
}
Dependency::Local(path_buf) => write!(f, "{}", path_buf.display()),
}
}
}
impl FromStr for Dependency {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self> {
Ok(Self::Package(s.parse()?))
}
}
#[derive(Debug, Clone)]
pub struct RegistryPackage {
pub name: Option<PackageRef>,
pub version: VersionReq,
pub registry: Option<String>,
}
impl FromStr for RegistryPackage {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self> {
Ok(Self {
name: None,
version: s
.parse()
.with_context(|| format!("'{s}' is an invalid registry package version"))?,
registry: None,
})
}
}
#[derive(Clone)]
pub struct RegistryResolution {
pub name: PackageRef,
pub package: PackageRef,
pub registry: Option<String>,
pub requirement: VersionReq,
pub version: Version,
pub digest: ContentDigest,
client: CachingClient<FileCache>,
}
impl Debug for RegistryResolution {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.debug_struct("RegistryResolution")
.field("name", &self.name)
.field("package", &self.package)
.field("registry", &self.registry)
.field("requirement", &self.requirement)
.field("version", &self.version)
.field("digest", &self.digest)
.finish()
}
}
impl RegistryResolution {
pub async fn fetch(&self) -> Result<impl AsyncRead> {
let stream = self
.client
.get_content(
&self.package,
&Release {
version: self.version.clone(),
content_digest: self.digest.clone(),
},
)
.await?;
Ok(tokio_util::io::StreamReader::new(
stream.map_err(std::io::Error::other),
))
}
}
#[derive(Clone, Debug)]
pub struct LocalResolution {
pub name: PackageRef,
pub path: PathBuf,
}
#[derive(Debug, Clone)]
#[allow(clippy::large_enum_variant)]
pub enum DependencyResolution {
Registry(RegistryResolution),
Local(LocalResolution),
}
impl DependencyResolution {
pub fn name(&self) -> &PackageRef {
match self {
Self::Registry(res) => &res.name,
Self::Local(res) => &res.name,
}
}
pub fn version(&self) -> Option<&Version> {
match self {
Self::Registry(res) => Some(&res.version),
Self::Local(_) => None,
}
}
pub fn key(&self) -> Option<(&PackageRef, Option<&str>)> {
match self {
DependencyResolution::Registry(pkg) => Some((&pkg.package, pkg.registry.as_deref())),
DependencyResolution::Local(_) => None,
}
}
pub async fn decode(&self) -> Result<DecodedDependency<'_>> {
let bytes = match self {
DependencyResolution::Local(LocalResolution { path, .. })
if tokio::fs::metadata(path).await?.is_dir() =>
{
return Ok(DecodedDependency::Wit {
resolution: self,
package: UnresolvedPackageGroup::parse_dir(path).with_context(|| {
format!("failed to parse dependency `{path}`", path = path.display())
})?,
});
}
DependencyResolution::Local(LocalResolution { path, .. }) => {
tokio::fs::read(path).await.with_context(|| {
format!(
"failed to read content of dependency `{name}` at path `{path}`",
name = self.name(),
path = path.display()
)
})?
}
DependencyResolution::Registry(res) => {
let mut reader = res.fetch().await?;
let mut buf = Vec::new();
reader.read_to_end(&mut buf).await?;
buf
}
};
if &bytes[0..4] != b"\0asm" {
let package = UnresolvedPackageGroup::parse(
self.name().to_string(),
std::str::from_utf8(&bytes).with_context(|| {
format!(
"dependency `{name}` is not UTF-8 encoded",
name = self.name()
)
})?,
)
.map_err(|(src_map, parse_err)| {
anyhow::format_err!(
"failed to parse package group: {}",
parse_err.render(&src_map)
)
})?;
return Ok(DecodedDependency::Wit {
resolution: self,
package,
});
}
Ok(DecodedDependency::Wasm {
resolution: self,
decoded: wit_component::decode(&bytes).with_context(|| {
format!(
"failed to decode content of dependency `{name}`",
name = self.name(),
)
})?,
})
}
}
pub enum DecodedDependency<'a> {
Wit {
resolution: &'a DependencyResolution,
package: UnresolvedPackageGroup,
},
Wasm {
resolution: &'a DependencyResolution,
decoded: DecodedWasm,
},
}
impl DecodedDependency<'_> {
pub fn resolve(self) -> Result<(Resolve, PackageId, Vec<PathBuf>)> {
match self {
Self::Wit { package, .. } => {
let mut resolve = Resolve::new();
resolve.all_features = true;
let source_files = package
.source_map
.source_files()
.map(Path::to_path_buf)
.collect();
let pkg = resolve.push_group(package)?;
Ok((resolve, pkg, source_files))
}
Self::Wasm { decoded, .. } => match decoded {
DecodedWasm::WitPackage(resolve, pkg) => Ok((resolve, pkg, Vec::new())),
DecodedWasm::Component(resolve, world) => {
let pkg = resolve.worlds[world].package.unwrap();
Ok((resolve, pkg, Vec::new()))
}
},
}
}
pub fn package_name(&self) -> &PackageName {
match self {
Self::Wit { package, .. } => &package.main.name,
Self::Wasm { decoded, .. } => &decoded.resolve().packages[decoded.package()].name,
}
}
pub fn into_component_world(self) -> Result<(Resolve, WorldId)> {
match self {
Self::Wasm {
decoded: DecodedWasm::Component(resolve, world),
..
} => Ok((resolve, world)),
_ => bail!("dependency is not a WebAssembly component"),
}
}
}
pub struct DependencyResolver<'a> {
client: CachingClient<FileCache>,
lock_file: Option<&'a LockFile>,
packages: HashMap<PackageRef, Vec<VersionInfo>>,
dependencies: HashMap<PackageRef, RegistryDependency>,
resolutions: DependencyResolutionMap,
}
impl<'a> DependencyResolver<'a> {
pub fn new(
config: Option<Config>,
lock_file: Option<&'a LockFile>,
cache: FileCache,
) -> anyhow::Result<Self> {
if config.is_none() && lock_file.is_none() {
anyhow::bail!("lock file must be provided when offline mode is enabled");
}
let client = CachingClient::new(config.map(Client::new), cache);
Ok(DependencyResolver {
client,
lock_file,
resolutions: Default::default(),
packages: Default::default(),
dependencies: Default::default(),
})
}
pub fn new_with_client(
client: CachingClient<FileCache>,
lock_file: Option<&'a LockFile>,
) -> anyhow::Result<Self> {
if client.is_readonly() && lock_file.is_none() {
anyhow::bail!("lock file must be provided when offline mode is enabled");
}
Ok(DependencyResolver {
client,
lock_file,
resolutions: Default::default(),
packages: Default::default(),
dependencies: Default::default(),
})
}
pub async fn add_dependency(
&mut self,
name: &PackageRef,
dependency: &Dependency,
) -> Result<()> {
self.add_dependency_internal(name, dependency, false).await
}
pub async fn override_dependency(
&mut self,
name: &PackageRef,
dependency: &Dependency,
) -> Result<()> {
self.add_dependency_internal(name, dependency, true).await
}
async fn add_dependency_internal(
&mut self,
name: &PackageRef,
dependency: &Dependency,
force_override: bool,
) -> Result<()> {
match dependency {
Dependency::Package(package) => {
let registry_name = package.registry.as_deref().or_else(|| {
self.client.client().ok().and_then(|client| {
client
.config()
.resolve_registry(name)
.map(|reg| reg.as_ref())
})
});
let package_name = package.name.clone().unwrap_or_else(|| name.clone());
let locked = match self.lock_file.as_ref().and_then(|resolver| {
resolver
.resolve(registry_name, &package_name, &package.version)
.transpose()
}) {
Some(Ok(locked)) => Some(locked),
Some(Err(e)) => return Err(e),
_ => None,
};
if !force_override
&& (self.resolutions.contains_key(name) || self.dependencies.contains_key(name))
{
tracing::debug!(%name, %dependency, "dependency already exists and override is not set, ignoring");
return Ok(());
}
self.dependencies.insert(
name.to_owned(),
RegistryDependency {
package: package_name,
version: package.version.clone(),
locked: locked.map(|l| (l.version.clone(), l.digest.clone())),
},
);
}
Dependency::Local(p) => {
let res = DependencyResolution::Local(LocalResolution {
name: name.clone(),
path: p.clone(),
});
let should_insert = force_override
|| self.dependencies.contains_key(name)
|| !self.resolutions.contains_key(name);
if !should_insert {
tracing::debug!(%name, "dependency already exists and registry override is not set, ignoring");
return Ok(());
}
self.dependencies.remove(name);
let (_, packages) = get_packages(p)
.context("Error getting dependent packages from local dependency")?;
Box::pin(self.add_packages(packages))
.await
.context("Error adding packages to resolver for local dependency")?;
let prev = self.resolutions.insert(name.clone(), res);
assert!(prev.is_none());
}
}
Ok(())
}
pub async fn add_packages(
&mut self,
packages: impl IntoIterator<Item = (PackageRef, VersionReq)>,
) -> Result<()> {
for (package, req) in packages {
self.add_dependency(
&package,
&Dependency::Package(RegistryPackage {
name: Some(package.clone()),
version: req,
registry: None,
}),
)
.await?;
}
Ok(())
}
pub async fn resolve(mut self) -> Result<DependencyResolutionMap> {
let mut resolutions = self.resolutions;
for (name, dependency) in self.dependencies.into_iter() {
let client = self.client.clone();
let (selected_version, digest) = if client.is_readonly() {
dependency
.locked
.as_ref()
.map(|(ver, digest)| (ver, Some(digest)))
.ok_or_else(|| {
anyhow::anyhow!("Couldn't find locked dependency while in offline mode")
})?
} else {
let versions =
load_package(&mut self.packages, &self.client, dependency.package.clone())
.await
.with_context(|| format!("package: {}", dependency.package.clone()))?
.with_context(|| {
format!(
"package `{name}` was not found in component registry",
name = dependency.package
)
})?;
match &dependency.locked {
Some((version, digest)) => {
let exact_req = VersionReq {
comparators: vec![Comparator {
op: Op::Exact,
major: version.major,
minor: Some(version.minor),
patch: Some(version.patch),
pre: version.pre.clone(),
}],
};
find_latest_release(versions, &exact_req)
.map(|v| (&v.version, Some(digest)))
.or_else(|| find_latest_release(versions, &dependency.version).map(|v| (&v.version, None)))
}
None => find_latest_release(versions, &dependency.version).map(|v| (&v.version, None)),
}.with_context(||
format!(
"component registry package `{name}` has no release matching version requirement `{version}`",
name = dependency.package,
version = dependency.version
)
)?
};
let release = client
.get_release(&dependency.package, selected_version)
.await?;
if let Some(digest) = digest
&& &release.content_digest != digest
{
bail!(
"component registry package `{name}` (v`{version}`) has digest `{content}` but the lock file specifies digest `{digest}`",
name = dependency.package,
version = release.version,
content = release.content_digest,
);
}
let resolution = RegistryResolution {
name: name.clone(),
package: dependency.package.clone(),
registry: self.client.client().ok().and_then(|client| {
client
.config()
.resolve_registry(&name)
.map(ToString::to_string)
}),
requirement: dependency.version.clone(),
version: release.version.clone(),
digest: release.content_digest.clone(),
client: self.client.clone(),
};
resolutions.insert(name, DependencyResolution::Registry(resolution));
}
Ok(resolutions)
}
}
async fn load_package<'b>(
packages: &'b mut HashMap<PackageRef, Vec<VersionInfo>>,
client: &CachingClient<FileCache>,
package: PackageRef,
) -> Result<Option<&'b Vec<VersionInfo>>> {
match packages.entry(package) {
hash_map::Entry::Occupied(e) => Ok(Some(e.into_mut())),
hash_map::Entry::Vacant(e) => match client.list_all_versions(e.key()).await {
Ok(p) => Ok(Some(e.insert(p))),
Err(WasmPkgError::PackageNotFound) => Ok(None),
Err(err) => Err(err.into()),
},
}
}
#[derive(Debug)]
struct RegistryDependency {
package: PackageRef,
version: VersionReq,
locked: Option<(Version, ContentDigest)>,
}
fn find_latest_release<'a>(
versions: &'a [VersionInfo],
req: &VersionReq,
) -> Option<&'a VersionInfo> {
versions
.iter()
.filter(|info| !info.yanked && req.matches(&info.version))
.max_by(|a, b| a.version.cmp(&b.version))
}
#[derive(Debug, Clone, Default)]
pub struct DependencyResolutionMap(HashMap<PackageRef, DependencyResolution>);
impl AsRef<HashMap<PackageRef, DependencyResolution>> for DependencyResolutionMap {
fn as_ref(&self) -> &HashMap<PackageRef, DependencyResolution> {
&self.0
}
}
impl Deref for DependencyResolutionMap {
type Target = HashMap<PackageRef, DependencyResolution>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl DerefMut for DependencyResolutionMap {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
impl DependencyResolutionMap {
pub async fn decode_dependencies(
&self,
) -> Result<IndexMap<PackageName, DecodedDependency<'_>>> {
let mut deps = IndexMap::new();
for (name, resolution) in self.0.iter() {
let decoded = resolution.decode().await?;
if let Some(prev) = deps.insert(decoded.package_name().clone(), decoded) {
anyhow::bail!(
"duplicate definitions of package `{prev}` found while decoding dependency `{name}`",
prev = prev.package_name()
);
}
}
let mut order = IndexSet::new();
let mut visiting = HashSet::new();
for dep in deps.values() {
visit(dep, &deps, &mut order, &mut visiting)?;
}
assert!(visiting.is_empty());
deps.sort_by(|name_a, _, name_b, _| {
order.get_index_of(name_a).cmp(&order.get_index_of(name_b))
});
Ok(deps)
}
pub async fn generate_resolve(&self, dir: impl AsRef<Path>) -> Result<(Resolve, PackageId)> {
let mut merged = Resolve {
all_features: true,
..Resolve::default()
};
let deps = self.decode_dependencies().await?;
let root = UnresolvedPackageGroup::parse_dir(&dir).with_context(|| {
format!(
"failed to parse package from directory `{dir}`",
dir = dir.as_ref().display()
)
})?;
let mut source_files: Vec<_> = root
.source_map
.source_files()
.map(Path::to_path_buf)
.collect();
for decoded in deps.into_values() {
match decoded {
DecodedDependency::Wit {
resolution,
package,
} => {
source_files.extend(package.source_map.source_files().map(Path::to_path_buf));
merged.push_group(package).with_context(|| {
format!(
"failed to merge dependency `{name}`",
name = resolution.name()
)
})?;
}
DecodedDependency::Wasm {
resolution,
decoded,
} => {
let resolve = match decoded {
DecodedWasm::WitPackage(resolve, _) => resolve,
DecodedWasm::Component(resolve, _) => resolve,
};
merged.merge(resolve).with_context(|| {
format!(
"failed to merge world of dependency `{name}`",
name = resolution.name()
)
})?;
}
};
}
let package = merged.push_group(root).with_context(|| {
format!(
"failed to merge package from directory `{dir}`",
dir = dir.as_ref().display()
)
})?;
Ok((merged, package))
}
}
fn visit<'a>(
dep: &'a DecodedDependency<'a>,
deps: &'a IndexMap<PackageName, DecodedDependency>,
order: &mut IndexSet<PackageName>,
visiting: &mut HashSet<&'a PackageName>,
) -> Result<()> {
if order.contains(dep.package_name()) {
return Ok(());
}
match dep {
DecodedDependency::Wit {
package,
resolution,
} => {
for name in package.main.foreign_deps.keys() {
if let Some(dep) = deps.get(name) {
if !visiting.insert(name) {
anyhow::bail!(
"foreign dependency `{name}` forms a dependency cycle while parsing dependency `{other}`",
other = resolution.name()
);
}
visit(dep, deps, order, visiting)?;
assert!(visiting.remove(name));
}
}
}
DecodedDependency::Wasm {
decoded,
resolution,
} => {
for (_, package) in &decoded.resolve().packages {
if package.name.namespace == dep.package_name().namespace
&& package.name.name == dep.package_name().name
{
continue;
}
if let Some(dep) = deps.get(&package.name) {
if !visiting.insert(&package.name) {
anyhow::bail!(
"foreign dependency `{name}` forms a dependency cycle while parsing dependency `{other}`",
name = package.name,
other = resolution.name()
);
}
visit(dep, deps, order, visiting)?;
assert!(visiting.remove(&package.name));
}
}
}
}
assert!(order.insert(dep.package_name().clone()));
Ok(())
}
pub type DependencyGraph<N> = Acyclic<StableDiGraph<N, petgraph::Direction>>;
pub type LocalPackageIndex = HashMap<PackageRef, (NodeIndex, PathBuf)>;
pub struct PublishPlan {
dependents: DependencyGraph<PackageSpec>,
indices: LocalPackageIndex,
}
impl PublishPlan {
pub fn from_paths(paths: &[impl AsRef<Path>]) -> Result<Self> {
let (graph, indices) = get_local_dependencies(paths)?;
{
let missing_version = graph
.nodes_iter()
.map(|f| graph[f].clone())
.filter(|pkg| pkg.version.is_none())
.collect::<Vec<_>>();
if !missing_version.is_empty() {
return Err(anyhow::anyhow!(
"Unable to publish packages without a version specified"
)
.context(format!(
"packages: {}",
package_list(missing_version.iter(), None)
)));
}
}
let mut dependents = graph.into_inner();
dependents.reverse();
let dependents = DependencyGraph::try_from(dependents).unwrap();
Ok(Self {
dependents,
indices,
})
}
pub fn iter<'a>(&'a self) -> impl Iterator<Item = &'a PackageSpec> + 'a {
self.dependents
.nodes_iter()
.map(|id| &(self.dependents[id]))
}
pub fn is_empty(&self) -> bool {
self.indices.is_empty()
}
pub fn len(&self) -> usize {
self.indices.len()
}
pub fn take_ready(&self) -> BTreeSet<PackageSpec> {
self.dependents
.nodes_iter()
.filter(|id| {
self.dependents
.neighbors_directed(*id, Direction::Incoming)
.count()
== 0
})
.map(|id| self.dependents[id].clone())
.collect()
}
pub fn get_path(&self, pkg: &PackageRef) -> Option<&Path> {
self.indices.get(pkg).map(|(_, p)| p.as_ref())
}
pub fn get_node_index(&self, pkg: &PackageRef) -> Option<NodeIndex> {
self.indices.get(pkg).map(|(id, _)| *id)
}
pub fn mark_confirmed(&mut self, published: impl IntoIterator<Item = PackageSpec>) {
for spec in published {
let (id, _) = self
.indices
.remove(&spec.package)
.expect("PackageSpec has no associated index");
self.dependents.remove_node(id);
}
}
}
impl std::fmt::Display for PublishPlan {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
for id in self.dependents.nodes_iter() {
let dep = &self.dependents[id];
let mut neighbors = self
.dependents
.neighbors_directed(id, Direction::Outgoing)
.peekable();
if neighbors.peek().is_none() {
writeln!(f, "[{dep} has no dependents]")?;
} else {
writeln!(f, "[{dep}]")?;
}
while let Some(id) = neighbors.next() {
let pkg = &self.dependents[id];
let separator = if neighbors.peek().is_some() {
"├─"
} else {
"╰─"
};
writeln!(f, "{separator}─▶ {pkg}")?;
}
}
Ok(())
}
}
fn package_list<'a>(
pkgs: impl IntoIterator<Item = &'a PackageSpec>,
final_sep: Option<&str>,
) -> String {
let final_sep = final_sep.unwrap_or("and");
let mut names: Vec<_> = pkgs.into_iter().map(|pkg| pkg.to_string()).collect();
names.sort();
match &names[..] {
[] => String::new(),
[a] => a.clone(),
[a, b] => format!("{a} {final_sep} {b}"),
[names @ .., last] => {
format!("{}, {final_sep} {last}", names.join(", "))
}
}
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use super::*;
use glob::glob;
fn transitive_local_paths() -> Vec<PathBuf> {
let fixtures_root =
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/transitive-local");
let mut paths: Vec<PathBuf> = glob(fixtures_root.join("**/*.wit").to_str().unwrap())
.unwrap()
.map(|p| p.expect("glob path error"))
.map(|p| p.parent().unwrap().to_path_buf())
.collect();
paths.sort();
paths.dedup();
paths
}
#[test]
fn publish_plan_iter() {
let paths = transitive_local_paths();
let plan = PublishPlan::from_paths(&paths).unwrap();
assert_eq!(
plan.iter().count(),
5,
"unexpected package count\npackages found: {}",
package_list(plan.iter(), None)
);
}
#[test]
fn publish_plan_chunks() {
let paths = transitive_local_paths();
let mut plan = PublishPlan::from_paths(&paths).unwrap();
let mut ready_for_publish = plan.take_ready();
assert_eq!(
ready_for_publish.iter().collect::<Vec<_>>(),
["example-c:nested@0.1.0", "example-d:foo@0.1.0",],
);
plan.mark_confirmed(ready_for_publish);
ready_for_publish = plan.take_ready();
assert_eq!(
ready_for_publish.iter().collect::<Vec<_>>(),
["example-c:baz@0.1.0"],
);
plan.mark_confirmed(ready_for_publish);
ready_for_publish = plan.take_ready();
assert_eq!(
ready_for_publish.iter().collect::<Vec<_>>(),
["example-b:bar@0.1.0"],
);
plan.mark_confirmed(ready_for_publish);
ready_for_publish = plan.take_ready();
assert_eq!(
ready_for_publish.iter().collect::<Vec<_>>(),
["example-a:foo@0.1.0"],
);
plan.mark_confirmed(ready_for_publish);
assert!(plan.is_empty());
}
}