rework to be multi-tenant

supports multiple handles now

Signed-off-by: oppiliappan <[e-mail]>

readme.txt

···
 usage
 -----
 
-  cargo run -- DID|handle
+    cargo run -- [DIDs|handles ...] [-m mountpoint]
+
+example
+-------
+
+    $ cargo run -- oppi.li icyphox.sh
+    ⠏ [00:00:00] using cached CAR file for...did:plc:qfpnj4og54vl56wngdriaxug
+    ⠏ [00:00:00] using cached CAR file for...did:plc:hwevmowznbiukdf6uk5dwrrq
+    mounted at "mnt"
+    hit enter to unmount and exit...
+
+
+    $ cat foo/**/sh.tangled.publicKey/* | jq -r '.key'
+    ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICJPYX06+qKr9IHWfkgCtHbExoBOOwS/+iAWbog9bAdk icy@wyndle
+    ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIMj1Dn9YuFo2BNr993ymBa6nzyyIKAURIqMbUtfI8+4X op@mantis
+      .
+      .
+      .
+
+
+

src/error.rs

···
 pub enum Error {
     #[error("atproto error: {0}")]
     GetRepo(#[from] atrium_xrpc::Error<atrium_api::com::atproto::sync::get_repo::Error>),
+    #[error("repo build error: {0}")]
+    Repo(#[from] atrium_repo::repo::Error),
+    #[error("car store error: {0}")]
+    Car(#[from] atrium_repo::blockstore::CarError),
+    #[error("identity error: {0}")]
+    Identity(#[from] atrium_identity::Error),
     #[error("io error: {0}")]
     Io(#[from] std::io::Error),
 }

src/fs.rs

···
-use std::{collections::BTreeMap, time};
+use std::{
+    collections::{BTreeMap, hash_set::Iter},
+    time,
+};
 
 use atrium_repo::{Repository, blockstore::AsyncBlockStoreRead};
 use futures::StreamExt;
-use indexmap::IndexSet;
+use indexmap::{IndexMap, IndexSet};
+
+type Inode = usize;
+
 pub struct PdsFs<R> {
-    repo: Repository<R>,
+    repos: IndexMap<String, Repository<R>>,
     inodes: IndexSet<PdsFsEntry>,
 }
 
···
 pub enum PdsFsEntry {
     Zero,
     Root,
-    Collection(String),
+    Did(String),
+    Collection(PdsFsCollection),
     Record(PdsFsRecord),
 }
 
+impl PdsFsEntry {
+    fn as_collection(&self) -> Option<&PdsFsCollection> {
+        match &self {
+            Self::Collection(c) => Some(c),
+            _ => None,
+        }
+    }
+
+    fn as_did(&self) -> Option<&String> {
+        match &self {
+            Self::Did(d) => Some(d),
+            _ => None,
+        }
+    }
+
+    fn unwrap_collection(&self) -> &PdsFsCollection {
+        self.as_collection().unwrap()
+    }
+
+    fn unwrap_did(&self) -> &String {
+        self.as_did().unwrap()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct PdsFsCollection {
+    parent: Inode,
+    nsid: String,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct PdsFsRecord {
-    collection: String,
+    parent: Inode,
     rkey: String,
 }
 
-impl PdsFsRecord {
-    fn key(&self) -> String {
-        format!("{}/{}", self.collection, self.rkey)
-    }
-}
+// impl PdsFsRecord {
+//     fn key(&self) -> String {
+//         format!("{}/{}", self.collection, self.rkey)
+//     }
+// }
 
 const TTL: time::Duration = time::Duration::from_secs(300);
 const BLKSIZE: u32 = 512;
···
 where
     R: AsyncBlockStoreRead,
 {
-    pub async fn new(mut repo: Repository<R>) -> Self {
+    pub fn new() -> Self {
+        PdsFs {
+            repos: Default::default(),
+            inodes: IndexSet::from([PdsFsEntry::Zero, PdsFsEntry::Root]),
+        }
+    }
+
+    pub async fn add(&mut self, did: String, mut repo: Repository<R>) {
         let mut mst = repo.tree();
 
-        // collect all keys and group by collection
+        let (did_inode, _) = self.inodes.insert_full(PdsFsEntry::Did(did.clone()));
+
         let mut keys = Box::pin(mst.keys());
-        let mut collections: BTreeMap<String, Vec<PdsFsRecord>> = BTreeMap::new();
+        while let Some(Ok(key)) = keys.next().await {
+            if let Some((collection_name, rkey)) = key.split_once("/") {
+                let (collection_inode, _) =
+                    self.inodes
+                        .insert_full(PdsFsEntry::Collection(PdsFsCollection {
+                            parent: did_inode,
+                            nsid: collection_name.to_owned(),
+                        }));
 
-        while let Some(Ok(key)) = keys.next().await {
-            if let Some((collection, rkey)) = key.split_once("/") {
-                let record = PdsFsRecord {
-                    collection: collection.to_owned(),
+                self.inodes.insert(PdsFsEntry::Record(PdsFsRecord {
+                    parent: collection_inode,
                     rkey: rkey.to_owned(),
-                };
-
-                collections
-                    .entry(collection.to_owned())
-                    .or_insert_with(Vec::new)
-                    .push(record);
+                }));
             }
         }
 
         drop(keys);
         drop(mst);
 
-        // build inode structure with proper ordering
-        let mut inodes = IndexSet::from([PdsFsEntry::Zero, PdsFsEntry::Root]);
-
-        // add collections first
-        for n in collections.keys().cloned() {
-            inodes.insert(PdsFsEntry::Collection(n));
-        }
-
-        // then add all records grouped by collection
-        for r in collections.values().flatten().cloned() {
-            inodes.insert(PdsFsEntry::Record(r));
-        }
-
-        println!("constructed {} inodes", inodes.len());
-
-        PdsFs { repo, inodes }
+        self.repos.insert(did, repo);
     }
 
     fn attr(&mut self, ino: u64) -> fuser::FileAttr {
···
                 flags: 0,
                 blksize: BLKSIZE,
             },
+            Some(PdsFsEntry::Did(_)) => fuser::FileAttr {
+                ino,
+                size: 0,
+                blocks: 0,
+                atime: time::UNIX_EPOCH,
+                mtime: time::UNIX_EPOCH,
+                ctime: time::UNIX_EPOCH,
+                crtime: time::UNIX_EPOCH,
+                kind: fuser::FileType::Directory,
+                perm: 0o755,
+                nlink: 2,
+                uid: 1000,
+                gid: 1000,
+                rdev: 0,
+                flags: 0,
+                blksize: BLKSIZE,
+            },
             Some(PdsFsEntry::Record(r)) => {
+                let col = self.inodes[r.parent].unwrap_collection();
+                let did = self.inodes[col.parent].unwrap_did();
+                let repo = &mut self.repos[did];
+                let key = format!("{}/{}", col.nsid, r.rkey);
                 let rt = tokio::runtime::Runtime::new().unwrap();
                 let size = rt
-                    .block_on(self.repo.get_raw::<ipld_core::ipld::Ipld>(&r.key()))
+                    .block_on(repo.get_raw::<ipld_core::ipld::Ipld>(&key))
                     .ok()
                     .flatten()
                     .map_or(0, |v| serde_json::to_string(&v).unwrap().len())
···
                 let entries: Vec<_> = vec![(ino, ".".to_string()), (ino, "..".to_string())]
                     .into_iter()
                     .chain(self.inodes.iter().enumerate().filter_map(|(i, e)| {
-                        if let PdsFsEntry::Collection(name) = e {
-                            Some((i as u64, name.clone()))
+                        if let PdsFsEntry::Did(did) = e {
+                            Some((i as u64, did.clone()))
                         } else {
                             None
                         }
···
                 }
                 reply.ok()
             }
-            Some(PdsFsEntry::Collection(collection_name)) => {
-                let entries: Vec<_> = vec![
-                    (ino, ".".to_string()),
-                    (1, "..".to_string()), // Parent is root (inode 1)
-                ]
-                .into_iter()
-                .chain(self.inodes.iter().enumerate().filter_map(|(i, e)| {
-                    if let PdsFsEntry::Record(record) = e {
-                        if record.collection == *collection_name {
-                            Some((i as u64, record.rkey.clone()))
+            Some(PdsFsEntry::Did(_)) => {
+                let entries = vec![(ino, ".".to_string()), (1, "..".to_string())]
+                    .into_iter()
+                    .chain(self.inodes.iter().enumerate().filter_map(|(i, e)| {
+                        if let PdsFsEntry::Collection(col) = e {
+                            if col.parent == ino as usize {
+                                Some((i as u64, col.nsid.clone()))
+                            } else {
+                                None
+                            }
                         } else {
                             None
                         }
-                    } else {
-                        None
+                    }))
+                    .into_iter()
+                    .enumerate()
+                    .skip(offset as usize);
+
+                for (index, (inode_num, name)) in entries {
+                    let full = reply.add(
+                        inode_num,
+                        (index + 1) as i64,
+                        if name.starts_with('.') {
+                            fuser::FileType::Directory
+                        } else {
+                            fuser::FileType::RegularFile
+                        },
+                        name,
+                    );
+                    if full {
+                        break;
                     }
-                }))
-                .collect();
+                }
 
-                for (index, (inode_num, name)) in
-                    entries.into_iter().enumerate().skip(offset as usize)
-                {
-                    if reply.add(
+                reply.ok();
+            }
+            Some(PdsFsEntry::Collection(c)) => {
+                let entries = [(ino, ".".to_string()), (c.parent as u64, "..".to_string())]
+                    .into_iter()
+                    .chain(self.inodes.iter().enumerate().filter_map(|(i, e)| {
+                        if let PdsFsEntry::Record(record) = e {
+                            if record.parent == ino as usize {
+                                Some((i as u64, record.rkey.clone()))
+                            } else {
+                                None
+                            }
+                        } else {
+                            None
+                        }
+                    }))
+                    .into_iter()
+                    .enumerate()
+                    .skip(offset as usize);
+
+                for (index, (inode_num, name)) in entries {
+                    let full = reply.add(
                         inode_num,
                         (index + 1) as i64,
                         if name.starts_with('.') {
···
                             fuser::FileType::RegularFile
                         },
                         name,
-                    ) {
+                    );
+                    if full {
                         break;
                     }
                 }
+
                 reply.ok()
             }
             _ => reply.error(libc::ENOENT),
···
     ) {
         match self.inodes.get_index(parent as usize) {
             Some(PdsFsEntry::Root) => {
-                let collection = PdsFsEntry::Collection(name.to_string_lossy().to_string());
-                if let Some(ino) = self.inodes.get_index_of(&collection) {
+                let did = PdsFsEntry::Did(name.to_string_lossy().to_string());
+                if let Some(ino) = self.inodes.get_index_of(&did) {
+                    reply.entry(&TTL, &self.attr(ino as u64), 0);
+                } else {
+                    reply.error(libc::ENOENT)
+                }
+            }
+            Some(PdsFsEntry::Did(_)) => {
+                let col = PdsFsEntry::Collection(PdsFsCollection {
+                    parent: parent as usize,
+                    nsid: name.to_string_lossy().to_string(),
+                });
+                if let Some(ino) = self.inodes.get_index_of(&col) {
                     reply.entry(&TTL, &self.attr(ino as u64), 0);
                 } else {
                     reply.error(libc::ENOENT)
                 }
             }
-            Some(PdsFsEntry::Collection(c)) => {
+            Some(PdsFsEntry::Collection(_)) => {
                 let record = PdsFsEntry::Record(PdsFsRecord {
-                    collection: c.to_owned(),
+                    parent: parent as usize,
                     rkey: name.to_string_lossy().to_string(),
                 });
                 if let Some(ino) = self.inodes.get_index_of(&record) {
···
         reply: fuser::ReplyData,
     ) {
         let rt = tokio::runtime::Runtime::new().unwrap();
-        if let Some(PdsFsEntry::Record(r)) = self.inodes.get_index(ino as usize)
-            && let Ok(Some(val)) = rt.block_on(self.repo.get_raw::<ipld_core::ipld::Ipld>(&r.key()))
-        {
-            reply.data(&serde_json::to_string(&val).unwrap().as_bytes()[offset as usize..]);
-        } else {
-            reply.error(libc::ENOENT);
+        if let Some(PdsFsEntry::Record(r)) = self.inodes.get_index(ino as usize) {
+            let col = self.inodes[r.parent].unwrap_collection();
+            let did = self.inodes[col.parent].unwrap_did();
+            let repo = &mut self.repos[did];
+            let key = format!("{}/{}", col.nsid, r.rkey);
+
+            if let Ok(Some(val)) = rt.block_on(repo.get_raw::<ipld_core::ipld::Ipld>(&key)) {
+                reply.data(&serde_json::to_string(&val).unwrap().as_bytes()[offset as usize..]);
+                return;
+            }
         }
+        reply.error(libc::ENOENT);
     }
 }

src/main.rs

···
 use atrium_repo::{Repository, blockstore::CarStore};
 use atrium_xrpc_client::isahc::IsahcClient;
 use fuser::MountOption;
-use indicatif::{ProgressBar, ProgressStyle};
+use futures::{Stream, StreamExt, TryStream, TryStreamExt, stream};
+use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
 use std::{
+    collections::HashMap,
     io::{Cursor, Write},
     path::PathBuf,
+    sync::Arc,
 };
 use xdg::BaseDirectories;
 
 fn main() {
     let rt = tokio::runtime::Runtime::new().unwrap();
     let matches = clap::command!()
-        .arg(clap::Arg::new("handle").index(1))
+        .arg(
+            clap::Arg::new("handles")
+                .index(1)
+                .required(true)
+                .num_args(1..)
+                .help("One or more handles to download and mount"),
+        )
         .arg(
             clap::Arg::new("mountpoint")
                 .short('m')
···
                 .value_parser(clap::value_parser!(PathBuf)),
         )
         .get_matches();
-    let handle = matches.get_one::<String>("handle").unwrap();
+    let handles = matches
+        .get_many::<String>("handles")
+        .unwrap()
+        .cloned()
+        .collect::<Vec<_>>();
     let mountpoint = matches
         .get_one::<PathBuf>("mountpoint")
         .map(ToOwned::to_owned)
         .unwrap_or(PathBuf::from("mnt"));
     let _ = std::fs::create_dir_all(&mountpoint);
-    let resolver = resolver::id_resolver();
-    let id = rt.block_on(async {
-        resolver
-            .resolve(&atrium_api::types::string::Handle::new(handle.into()).unwrap())
-            .await
-            .unwrap()
-    });
 
-    let pb = ProgressBar::new_spinner();
-    pb.set_style(
-        ProgressStyle::default_spinner()
-            .template("{spinner:.green} [{elapsed_precise}] {msg}")
-            .unwrap()
-            .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]),
+    let resolver = Arc::new(resolver::id_resolver());
+    let bars = Arc::new(MultiProgress::new());
+    let repos = rt.block_on(
+        stream::iter(handles)
+            .then(|handle| {
+                let h = handle.clone();
+                let r = Arc::clone(&resolver);
+                let b = Arc::clone(&bars);
+                async move {
+                    let id = r.resolve(&h).await?;
+                    let bytes = cached_download(&id, &b).await?;
+                    let repo = build_repo(bytes).await?;
+                    Ok::<_, error::Error>((id.did, repo))
+                }
+            })
+            .collect::<Vec<_>>(),
     );
-
-    let bytes = rt.block_on(cached_download(&id, &pb)).unwrap();
-    let store = rt.block_on(async { CarStore::open(Cursor::new(bytes)).await.unwrap() });
-    let root = store.roots().next().unwrap();
-    let repo = rt.block_on(async { Repository::open(store, root).await.unwrap() });
+    let (success, errors): (Vec<_>, Vec<_>) = repos.into_iter().partition(|r| r.is_ok());
+    for e in errors {
+        eprintln!("{:?}", e.as_ref().unwrap_err());
+    }
+    let repos = success
+        .into_iter()
+        .map(|s| s.unwrap())
+        .collect::<HashMap<_, _>>();
 
     // construct the fs
-    let fs = rt.block_on(fs::PdsFs::new(repo));
+    let mut fs = fs::PdsFs::new();
+    for (did, repo) in repos {
+        rt.block_on(fs.add(did, repo))
+    }
 
     // mount
     let options = vec![MountOption::RO, MountOption::FSName("pdsfs".to_string())];
···
     println!("unmounted {mountpoint:?}");
 }
 
-async fn cached_download(id: &ResolvedIdentity, pb: &ProgressBar) -> Result<Vec<u8>, error::Error> {
+async fn cached_download(
+    id: &ResolvedIdentity,
+    m: &MultiProgress,
+) -> Result<Vec<u8>, error::Error> {
+    let mut pb = ProgressBar::new_spinner();
+    pb.set_style(
+        ProgressStyle::default_spinner()
+            .template("{spinner:.green} [{elapsed_precise}] {msg}")
+            .unwrap()
+            .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]),
+    );
+    pb.enable_steady_tick(std::time::Duration::from_millis(100));
+    pb = m.add(pb);
+
     let dirs = BaseDirectories::new();
 
     let dir = dirs.get_data_home().expect("$HOME is absent").join("pdsfs");
···
     let exists = std::fs::exists(&file)?;
 
     let bytes = if !exists {
-        pb.set_message(format!("downloading CAR file for\t...\t{}", id.did));
-        download_car_file(id, pb).await?
+        pb.set_message(format!("downloading CAR file for...{}", id.did));
+        download_car_file(id, &pb).await?
     } else {
-        pb.set_message(format!("using cached CAR file for\t...\t{}", id.did));
-        tokio::fs::read(file).await?
+        pb.set_message(format!("using cached CAR file for...{}", id.did));
+        tokio::fs::read(&file).await?
     };
 
-    pb.set_message(format!(
-        "received {} bytes for \t...\t{}",
-        bytes.len(),
-        id.did
-    ));
+    // write to disk
+    if !exists {
+        tokio::fs::write(&file, &bytes).await?;
+    }
+
+    pb.finish();
     Ok(bytes)
 }
 
···
 
     Ok(bytes)
 }
+
+async fn build_repo(bytes: Vec<u8>) -> Result<Repository<CarStore<Cursor<Vec<u8>>>>, error::Error> {
+    let store = CarStore::open(Cursor::new(bytes)).await?;
+    let root = store.roots().next().unwrap();
+    let repo = Repository::open(store, root).await?;
+    Ok(repo)
+}