use std::collections::{HashMap, HashSet, VecDeque};
use std::io::{Error, ErrorKind, Read, Result};
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock};
use bytes::Bytes;
use flate2::read::GzDecoder;
use tar::Archive;
use tokio_fs_ext::DirEntry;
use tracing::warn;
#[derive(Debug, Clone)]
pub struct FuseLink {
pub target_dir: PathBuf,
}
impl FuseLink {
pub fn parse(content: &str) -> Option<Self> {
let line = content.lines().next()?.trim();
if line.is_empty() {
return None;
}
let path = line.split_once('|').map_or(line, |(p, _)| p);
Some(Self {
target_dir: PathBuf::from(path),
})
}
pub fn to_content(&self) -> String {
format!("{}\n", self.target_dir.display())
}
}
const EXTRACTION_CONCURRENCY: usize = 64;
struct BoundedCache {
map: HashMap<PathBuf, Arc<FuseLink>>,
order: VecDeque<PathBuf>,
capacity: usize,
}
impl BoundedCache {
fn new(capacity: usize) -> Self {
Self {
map: HashMap::with_capacity(capacity.min(256)),
order: VecDeque::with_capacity(capacity.min(256)),
capacity: capacity.max(1),
}
}
fn get(&self, key: &Path) -> Option<&Arc<FuseLink>> {
self.map.get(key)
}
fn put(&mut self, key: PathBuf, value: Arc<FuseLink>) {
if self.map.contains_key(&key) {
self.map.insert(key, value);
return;
}
if self.map.len() >= self.capacity {
if let Some(oldest) = self.order.pop_front() {
self.map.remove(&oldest);
}
}
self.order.push_back(key.clone());
self.map.insert(key, value);
}
fn clear(&mut self) {
self.map.clear();
self.order.clear();
}
}
pub struct FuseFs {
link_cache: RwLock<BoundedCache>,
}
impl FuseFs {
pub fn new(fuse_cache_max_entries: usize) -> Self {
Self {
link_cache: RwLock::new(BoundedCache::new(fuse_cache_max_entries)),
}
}
pub async fn create_fuse_link(&self, target_dir: &Path, dst: &Path) -> Result<()> {
let fuse_link_path = dst.join("fuse.link");
if tokio_fs_ext::metadata(&fuse_link_path)
.await
.map(|m| m.is_file())
.unwrap_or(false)
{
return Ok(());
}
let parent = fuse_link_path.parent().ok_or_else(|| {
Error::new(ErrorKind::InvalidInput, "cannot determine fuse.link path")
})?;
tokio_fs_ext::create_dir_all(parent).await?;
let link = Arc::new(FuseLink {
target_dir: target_dir.to_path_buf(),
});
tokio_fs_ext::write(&fuse_link_path, link.to_content().as_bytes()).await?;
if let Ok(mut cache) = self.link_cache.write() {
cache.put(fuse_link_path, link);
} else {
warn!("fuse link cache lock poisoned");
}
Ok(())
}
pub async fn try_read(&self, path: &Path) -> Result<Option<Bytes>> {
let resolved = match self.resolve(path).await? {
Some(r) => r,
None => return Ok(None),
};
let real_path = resolved.link.target_dir.join(&resolved.relative);
match tokio_fs_ext::read(&real_path).await {
Ok(v) => Ok(Some(Bytes::from(v))),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(None),
Err(e) => Err(e),
}
}
pub async fn try_read_dir(&self, path: &Path) -> Result<Option<Vec<DirEntry>>> {
let resolved = match self.resolve(path).await? {
Some(r) => r,
None => return Ok(None),
};
let real_dir = resolved.link.target_dir.join(&resolved.relative);
let target_entries = match read_dir_direct(&real_dir).await {
Ok(entries) => entries,
Err(_) => return Ok(None),
};
match read_dir_direct(path).await {
Ok(original) => {
let has_extra_files = original
.iter()
.any(|e| e.file_name().to_string_lossy() != "fuse.link");
if !has_extra_files {
return Ok(Some(target_entries));
}
let target_names: HashSet<_> =
target_entries.iter().map(|e| e.file_name()).collect();
let mut combined: Vec<_> = original
.into_iter()
.filter(|e| {
e.file_name().to_string_lossy() != "fuse.link"
&& !target_names.contains(&e.file_name())
})
.collect();
combined.extend(target_entries);
Ok(Some(combined))
}
Err(_) => Ok(Some(target_entries)),
}
}
pub async fn try_metadata(&self, path: &Path) -> Result<Option<tokio_fs_ext::Metadata>> {
let resolved = match self.resolve(path).await? {
Some(r) => r,
None => return Ok(None),
};
let real_path = resolved.link.target_dir.join(&resolved.relative);
match tokio_fs_ext::metadata(&real_path).await {
Ok(m) => Ok(Some(m)),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(None),
Err(e) => Err(e),
}
}
pub fn warm_link_cache(&self, dst: &Path, target_dir: &Path) {
let Some(fuse_link_path) = locate_fuse_link_file(dst) else {
return;
};
let link = Arc::new(FuseLink {
target_dir: target_dir.to_path_buf(),
});
if let Ok(mut cache) = self.link_cache.write() {
cache.put(fuse_link_path, link);
}
}
pub async fn extract_tgz_to_dir(&self, tgz_path: &Path) -> Result<PathBuf> {
let out_dir = tgz_path.with_extension(""); let sentinel = PathBuf::from(format!("{}._resolved", out_dir.display()));
if tokio_fs_ext::metadata(&sentinel).await.is_ok() {
return Ok(out_dir);
}
struct PendingFile {
path: PathBuf,
content: Bytes,
}
let mut pending_files: Vec<PendingFile> = Vec::new();
let mut unique_dirs: HashSet<PathBuf> = HashSet::new();
{
let raw = tokio_fs_ext::read(tgz_path).await?;
let gz = GzDecoder::new(&raw[..]);
let mut archive = Archive::new(gz);
for entry_result in archive.entries()? {
let mut entry = entry_result?;
if !entry.header().entry_type().is_file() {
continue;
}
let path = entry.path()?.to_path_buf();
if path.is_absolute()
|| path
.components()
.any(|c| matches!(c, std::path::Component::ParentDir))
{
return Err(Error::new(
ErrorKind::InvalidInput,
format!("malicious path in tar entry: {}", path.display()),
));
}
let normalized = if let Some(first) = path.components().next() {
let stripped = path.strip_prefix(first).unwrap_or(&path);
if stripped.as_os_str().is_empty() {
path
} else {
stripped.to_path_buf()
}
} else {
path
};
if normalized.as_os_str().is_empty() {
continue;
}
let mut content = Vec::new();
entry.read_to_end(&mut content)?;
let full_path = out_dir.join(normalized);
if let Some(parent) = full_path.parent() {
unique_dirs.insert(parent.to_path_buf());
}
pending_files.push(PendingFile {
path: full_path,
content: Bytes::from(content),
});
}
}
for dir in &unique_dirs {
tokio_fs_ext::create_dir_all(dir).await?;
}
use futures::stream::{FuturesUnordered, StreamExt};
let mut write_futures = FuturesUnordered::new();
for pf in pending_files {
write_futures.push(async move { tokio_fs_ext::write(&pf.path, &pf.content).await });
if write_futures.len() >= EXTRACTION_CONCURRENCY {
if let Some(res) = write_futures.next().await {
res?;
}
}
}
while let Some(res) = write_futures.next().await {
res?;
}
tokio_fs_ext::write(&sentinel, b"").await?;
Ok(out_dir)
}
pub fn clear(&self) {
if let Ok(mut lc) = self.link_cache.write() {
lc.clear();
}
}
async fn resolve(&self, path: &Path) -> Result<Option<Resolved>> {
let fuse_link_path = match locate_fuse_link_file(path) {
Some(p) => p,
None => return Ok(None),
};
let link = match self.read_fuse_link(&fuse_link_path).await? {
Some(l) => l,
None => return Ok(None),
};
let fuse_dir = fuse_link_path
.parent()
.ok_or_else(|| Error::new(ErrorKind::InvalidInput, "invalid fuse.link path"))?;
let relative = path
.strip_prefix(fuse_dir)
.map_err(|_| Error::new(ErrorKind::InvalidInput, "path not under fuse.link dir"))?
.to_path_buf();
Ok(Some(Resolved { link, relative }))
}
async fn read_fuse_link(&self, fuse_link_path: &Path) -> Result<Option<Arc<FuseLink>>> {
if let Ok(cache) = self.link_cache.read() {
if let Some(link) = cache.get(fuse_link_path) {
return Ok(Some(Arc::clone(link)));
}
}
let content = match tokio_fs_ext::read_to_string(fuse_link_path).await {
Ok(c) => c,
Err(_) => return Ok(None),
};
let link = match FuseLink::parse(&content) {
Some(l) => Arc::new(l),
None => return Ok(None),
};
if let Ok(mut cache) = self.link_cache.write() {
if let Some(existing) = cache.get(fuse_link_path) {
return Ok(Some(Arc::clone(existing)));
}
cache.put(fuse_link_path.to_path_buf(), Arc::clone(&link));
}
Ok(Some(link))
}
}
struct Resolved {
link: Arc<FuseLink>,
relative: PathBuf,
}
fn locate_fuse_link_file(path: &Path) -> Option<PathBuf> {
use std::ffi::OsStr;
use std::path::Component;
let node_modules = OsStr::new("node_modules");
if !path
.components()
.any(|c| matches!(c, Component::Normal(name) if name == node_modules))
{
return None;
}
let mut comps = path.components();
let mut pkg_components: Vec<&OsStr> = Vec::new();
while let Some(comp) = comps.next_back() {
if let Component::Normal(name) = comp {
if name == node_modules {
if let Some(pkg) = pkg_components.last().copied() {
let pkg_str = pkg.to_string_lossy();
if pkg_str.starts_with('@') {
if pkg_components.len() >= 2 {
let scope = pkg;
let name = pkg_components[pkg_components.len() - 2];
let mut base = comps.as_path().to_path_buf();
base.push("node_modules");
base.push(scope);
base.push(name);
base.push("fuse.link");
return Some(base);
}
} else if pkg_str != "fuse.link" {
let mut base = comps.as_path().to_path_buf();
base.push("node_modules");
base.push(pkg);
base.push("fuse.link");
return Some(base);
}
}
} else {
pkg_components.push(name);
}
}
}
None
}
async fn read_dir_direct(path: &Path) -> Result<Vec<DirEntry>> {
tokio_fs_ext::read_dir(path).await?.collect()
}
#[cfg(test)]
mod tests {
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_dedicated_worker);
use super::*;
use wasm_bindgen_test::*;
#[wasm_bindgen_test]
fn test_fuse_link_parse_target_dir() {
let link = FuseLink::parse("/stores/lodash/-/lodash-4.17.21").unwrap();
assert_eq!(
link.target_dir,
PathBuf::from("/stores/lodash/-/lodash-4.17.21")
);
}
#[wasm_bindgen_test]
fn test_fuse_link_parse_strips_legacy_prefix() {
let link = FuseLink::parse("/stores/lodash/-/lodash-4.17.21.tgz|package").unwrap();
assert_eq!(
link.target_dir,
PathBuf::from("/stores/lodash/-/lodash-4.17.21.tgz")
);
}
#[wasm_bindgen_test]
fn test_fuse_link_parse_empty() {
assert!(FuseLink::parse("").is_none());
assert!(FuseLink::parse(" \n").is_none());
}
#[wasm_bindgen_test]
fn test_fuse_link_roundtrip() {
let link = FuseLink {
target_dir: PathBuf::from("/stores/foo/-/foo-1.0.0"),
};
let content = link.to_content();
let parsed = FuseLink::parse(&content).unwrap();
assert_eq!(parsed.target_dir, link.target_dir);
}
#[wasm_bindgen_test]
fn test_locate_fuse_link_basic() {
assert_eq!(
locate_fuse_link_file(Path::new("./node_modules/c/index.js")),
Some(PathBuf::from("./node_modules/c/fuse.link"))
);
}
#[wasm_bindgen_test]
fn test_locate_fuse_link_scoped() {
assert_eq!(
locate_fuse_link_file(Path::new("./node_modules/@a/b/package.json")),
Some(PathBuf::from("./node_modules/@a/b/fuse.link"))
);
}
#[wasm_bindgen_test]
fn test_locate_fuse_link_nested_node_modules() {
assert_eq!(
locate_fuse_link_file(Path::new("./node_modules/a/node_modules/b/lib/index.js")),
Some(PathBuf::from("./node_modules/a/node_modules/b/fuse.link"))
);
}
#[wasm_bindgen_test]
fn test_locate_fuse_link_none() {
assert_eq!(locate_fuse_link_file(Path::new("./some/other/path")), None);
assert_eq!(locate_fuse_link_file(Path::new("./src/index.js")), None);
}
#[wasm_bindgen_test]
fn test_locate_fuse_link_fast_path_skips_non_node_modules() {
assert_eq!(locate_fuse_link_file(Path::new("/src/App.tsx")), None);
assert_eq!(
locate_fuse_link_file(Path::new("/project/lib/utils.js")),
None
);
assert_eq!(locate_fuse_link_file(Path::new("package.json")), None);
}
async fn write_test_tgz(tgz_path: &Path) {
use crate::archive::{PackFile, gzip};
let files = vec![
PackFile::new("package/package.json", br#"{"name":"test"}"#.to_vec()),
PackFile::new("package/index.js", b"module.exports = {}".to_vec()),
];
let tgz = gzip(&files).unwrap();
if let Some(parent) = tgz_path.parent() {
let _ = tokio_fs_ext::create_dir_all(parent).await;
}
tokio_fs_ext::write(tgz_path, &tgz).await.unwrap();
}
#[wasm_bindgen_test]
async fn test_extract_tgz_creates_sentinel() {
let base = Path::new("/test_extract_sentinel");
let tgz_path = base.join("pkg-1.0.0.tgz");
write_test_tgz(&tgz_path).await;
let fs = FuseFs::new(100);
let out = fs.extract_tgz_to_dir(&tgz_path).await.unwrap();
assert_eq!(out, base.join("pkg-1.0.0"));
let sentinel = PathBuf::from(format!("{}._resolved", out.display()));
assert!(tokio_fs_ext::metadata(&sentinel).await.is_ok());
assert!(
tokio_fs_ext::metadata(&out.join("package.json"))
.await
.is_ok()
);
assert!(tokio_fs_ext::metadata(&out.join("index.js")).await.is_ok());
let _ = tokio_fs_ext::remove_dir_all(base).await;
}
#[wasm_bindgen_test]
async fn test_extract_tgz_skips_when_sentinel_exists() {
let base = Path::new("/test_extract_skip");
let tgz_path = base.join("pkg-1.0.0.tgz");
write_test_tgz(&tgz_path).await;
let fs = FuseFs::new(100);
let out = fs.extract_tgz_to_dir(&tgz_path).await.unwrap();
assert!(tokio_fs_ext::metadata(&out.join("index.js")).await.is_ok());
let _ = tokio_fs_ext::remove_file(&out.join("index.js")).await;
let out2 = fs.extract_tgz_to_dir(&tgz_path).await.unwrap();
assert_eq!(out, out2);
assert!(tokio_fs_ext::metadata(&out.join("index.js")).await.is_err());
let _ = tokio_fs_ext::remove_dir_all(base).await;
}
#[wasm_bindgen_test]
async fn test_extract_tgz_complex() {
let base = Path::new("/test_extract_complex");
let tgz_path = base.join("complex-1.0.0.tgz");
use crate::archive::{PackFile, gzip};
let mut files = Vec::new();
for i in 0..100 {
files.push(PackFile::new(
format!("package/file_{}.txt", i),
format!("content {}", i).into_bytes(),
));
}
files.push(PackFile::new(
"package/nested/deep/file.js",
b"console.log('deep')".to_vec(),
));
let tgz = gzip(&files).unwrap();
let _ = tokio_fs_ext::create_dir_all(base).await;
tokio_fs_ext::write(&tgz_path, &tgz).await.unwrap();
let fs = FuseFs::new(100);
let out = fs.extract_tgz_to_dir(&tgz_path).await.unwrap();
assert!(
tokio_fs_ext::metadata(&out.join("file_0.txt"))
.await
.is_ok()
);
assert!(
tokio_fs_ext::metadata(&out.join("file_49.txt"))
.await
.is_ok()
);
assert!(
tokio_fs_ext::metadata(&out.join("file_99.txt"))
.await
.is_ok()
);
assert!(
tokio_fs_ext::metadata(&out.join("nested/deep/file.js"))
.await
.is_ok()
);
let content = tokio_fs_ext::read_to_string(&out.join("nested/deep/file.js"))
.await
.unwrap();
assert_eq!(content, "console.log('deep')");
let _ = tokio_fs_ext::remove_dir_all(base).await;
}
#[wasm_bindgen_test]
async fn test_extract_tgz_re_extracts_without_sentinel() {
let base = Path::new("/test_extract_reextract");
let tgz_path = base.join("pkg-1.0.0.tgz");
write_test_tgz(&tgz_path).await;
let fs = FuseFs::new(100);
let out_dir = base.join("pkg-1.0.0");
let sentinel = PathBuf::from(format!("{}._resolved", out_dir.display()));
tokio_fs_ext::create_dir_all(&out_dir).await.unwrap();
assert!(tokio_fs_ext::metadata(&sentinel).await.is_err());
let out = fs.extract_tgz_to_dir(&tgz_path).await.unwrap();
assert_eq!(out, out_dir);
assert!(tokio_fs_ext::metadata(&sentinel).await.is_ok());
assert!(
tokio_fs_ext::metadata(&out.join("package.json"))
.await
.is_ok()
);
let _ = tokio_fs_ext::remove_dir_all(base).await;
}
}