fud, geode: add support for directories, update `get` default path, improve chunk verification

bin/fud/fud/src/event.rs

@@ -37,7 +37,7 @@ pub struct ChunkDownloadCompleted {
     pub resource: Resource,
 }
 #[derive(Clone, Debug)]
-pub struct FileDownloadCompleted {
+pub struct MetadataDownloadCompleted {
     pub hash: blake3::Hash,
     pub resource: Resource,
 }
@@ -61,7 +61,7 @@ pub struct ChunkNotFound {
     pub chunk_hash: blake3::Hash,
 }
 #[derive(Clone, Debug)]
-pub struct FileNotFound {
+pub struct MetadataNotFound {
     pub hash: blake3::Hash,
     pub resource: Resource,
 }
@@ -80,12 +80,12 @@ pub struct DownloadError {
 pub enum FudEvent {
     DownloadStarted(DownloadStarted),
     ChunkDownloadCompleted(ChunkDownloadCompleted),
-    FileDownloadCompleted(FileDownloadCompleted),
+    MetadataDownloadCompleted(MetadataDownloadCompleted),
     DownloadCompleted(DownloadCompleted),
     ResourceUpdated(ResourceUpdated),
     ResourceRemoved(ResourceRemoved),
     ChunkNotFound(ChunkNotFound),
-    FileNotFound(FileNotFound),
+    MetadataNotFound(MetadataNotFound),
     MissingChunks(MissingChunks),
     DownloadError(DownloadError),
 }
@@ -107,8 +107,8 @@ impl From<ChunkDownloadCompleted> for JsonValue {
         ])
     }
 }
-impl From<FileDownloadCompleted> for JsonValue {
-    fn from(info: FileDownloadCompleted) -> JsonValue {
+impl From<MetadataDownloadCompleted> for JsonValue {
+    fn from(info: MetadataDownloadCompleted) -> JsonValue {
         json_map([
             ("hash", JsonValue::String(hash_to_string(&info.hash))),
             ("resource", info.resource.into()),
@@ -144,8 +144,8 @@ impl From<ChunkNotFound> for JsonValue {
         ])
     }
 }
-impl From<FileNotFound> for JsonValue {
-    fn from(info: FileNotFound) -> JsonValue {
+impl From<MetadataNotFound> for JsonValue {
+    fn from(info: MetadataNotFound) -> JsonValue {
         json_map([
             ("hash", JsonValue::String(hash_to_string(&info.hash))),
             ("resource", info.resource.into()),
@@ -177,9 +177,10 @@ impl From<FudEvent> for JsonValue {
             FudEvent::ChunkDownloadCompleted(info) => {
                 json_map([("event", json_str("chunk_download_completed")), ("info", info.into())])
             }
-            FudEvent::FileDownloadCompleted(info) => {
-                json_map([("event", json_str("file_download_completed")), ("info", info.into())])
-            }
+            FudEvent::MetadataDownloadCompleted(info) => json_map([
+                ("event", json_str("metadata_download_completed")),
+                ("info", info.into()),
+            ]),
             FudEvent::DownloadCompleted(info) => {
                 json_map([("event", json_str("download_completed")), ("info", info.into())])
             }
@@ -192,8 +193,8 @@ impl From<FudEvent> for JsonValue {
             FudEvent::ChunkNotFound(info) => {
                 json_map([("event", json_str("chunk_not_found")), ("info", info.into())])
             }
-            FudEvent::FileNotFound(info) => {
-                json_map([("event", json_str("file_not_found")), ("info", info.into())])
+            FudEvent::MetadataNotFound(info) => {
+                json_map([("event", json_str("metadata_not_found")), ("info", info.into())])
             }
             FudEvent::MissingChunks(info) => {
                 json_map([("event", json_str("missing_chunks")), ("info", info.into())])

bin/fud/fud/src/lib.rs

@@ -24,7 +24,7 @@ use rand::{prelude::IteratorRandom, rngs::OsRng, seq::SliceRandom, RngCore};
 use sled_overlay::sled;
 use smol::{
     channel,
-    fs::{File, OpenOptions},
+    fs::{self, File, OpenOptions},
     io::{AsyncReadExt, AsyncWriteExt},
     lock::RwLock,
 };
@@ -37,7 +37,7 @@ use std::{
 
 use darkfi::{
     dht::{Dht, DhtHandler, DhtNode, DhtRouterItem, DhtRouterPtr},
-    geode::{hash_to_string, ChunkedFile, Geode},
+    geode::{hash_to_string, ChunkedStorage, FileSequence, Geode, MAX_CHUNK_SIZE},
     net::{ChannelPtr, P2pPtr},
     system::PublisherPtr,
     util::path::expand_path,
@@ -47,9 +47,9 @@ use darkfi::{
 /// P2P protocols
 pub mod proto;
 use proto::{
-    FudAnnounce, FudChunkReply, FudFileReply, FudFindNodesReply, FudFindNodesRequest,
-    FudFindRequest, FudFindSeedersReply, FudFindSeedersRequest, FudNotFound, FudPingReply,
-    FudPingRequest,
+    FudAnnounce, FudChunkReply, FudDirectoryReply, FudFileReply, FudFindNodesReply,
+    FudFindNodesRequest, FudFindRequest, FudFindSeedersReply, FudFindSeedersRequest, FudNotFound,
+    FudPingReply, FudPingRequest,
 };
 
 /// FudEvent
@@ -58,7 +58,7 @@ use event::{ChunkDownloadCompleted, ChunkNotFound, FudEvent, ResourceUpdated};
 
 /// Resource definition
 pub mod resource;
-use resource::{Resource, ResourceStatus};
+use resource::{Resource, ResourceStatus, ResourceType};
 
 /// JSON-RPC related methods
 pub mod rpc;
@@ -67,6 +67,10 @@ pub mod rpc;
 pub mod tasks;
 use tasks::FetchReply;
 
+/// Utils
+pub mod util;
+use util::get_all_files;
+
 // TODO: This is not Sybil-resistant
 fn generate_node_id() -> Result<blake3::Hash> {
     let mut rng = OsRng;
@@ -124,12 +128,12 @@ pub struct Fud {
     /// Sled tree containing "resource hash -> path on the filesystem"
     path_tree: sled::Tree,
 
-    get_tx: channel::Sender<(u16, blake3::Hash, PathBuf, Result<()>)>,
-    get_rx: channel::Receiver<(u16, blake3::Hash, PathBuf, Result<()>)>,
-    file_fetch_tx: channel::Sender<(Vec<DhtNode>, blake3::Hash, PathBuf, Result<()>)>,
-    file_fetch_rx: channel::Receiver<(Vec<DhtNode>, blake3::Hash, PathBuf, Result<()>)>,
-    file_fetch_end_tx: channel::Sender<(blake3::Hash, Result<()>)>,
-    file_fetch_end_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
+    get_tx: channel::Sender<(blake3::Hash, PathBuf)>,
+    get_rx: channel::Receiver<(blake3::Hash, PathBuf)>,
+    metadata_fetch_tx: channel::Sender<(Vec<DhtNode>, blake3::Hash, PathBuf)>,
+    metadata_fetch_rx: channel::Receiver<(Vec<DhtNode>, blake3::Hash, PathBuf)>,
+    metadata_fetch_end_tx: channel::Sender<Result<()>>,
+    metadata_fetch_end_rx: channel::Receiver<Result<()>>,
 
     event_publisher: PublisherPtr<FudEvent>,
 }
@@ -218,8 +222,8 @@ impl Fud {
         event_publisher: PublisherPtr<FudEvent>,
     ) -> Result<Self> {
         let (get_tx, get_rx) = smol::channel::unbounded();
-        let (file_fetch_tx, file_fetch_rx) = smol::channel::unbounded();
-        let (file_fetch_end_tx, file_fetch_end_rx) = smol::channel::unbounded();
+        let (metadata_fetch_tx, metadata_fetch_rx) = smol::channel::unbounded();
+        let (metadata_fetch_end_tx, metadata_fetch_end_rx) = smol::channel::unbounded();
 
         // Hashmap used for routing
         let seeders_router = Arc::new(RwLock::new(HashMap::new()));
@@ -240,10 +244,10 @@ impl Fud {
             resources: Arc::new(RwLock::new(HashMap::new())),
             get_tx,
             get_rx,
-            file_fetch_tx,
-            file_fetch_rx,
-            file_fetch_end_tx,
-            file_fetch_end_rx,
+            metadata_fetch_tx,
+            metadata_fetch_rx,
+            metadata_fetch_end_tx,
+            metadata_fetch_end_rx,
             event_publisher,
         };
 
@@ -286,6 +290,7 @@ impl Fud {
                 hash,
                 Resource {
                     hash,
+                    rtype: ResourceType::Unknown,
                     path,
                     status: ResourceStatus::Incomplete,
                     chunks_total: 0,
@@ -332,7 +337,7 @@ impl Fud {
     /// Verify if resources are complete and uncorrupted.
     /// If a resource is incomplete or corrupted, its status is changed to Incomplete.
     /// If a resource is complete, its status is changed to Seeding.
-    /// Takes an optional list of hashes.
+    /// Takes an optional list of resource hashes.
     /// If no hash is given (None), it verifies all resources.
     /// Returns the list of verified and uncorrupted/complete seeding resources.
     pub async fn verify_resources(
@@ -344,17 +349,24 @@ impl Fud {
         let update_resource =
             async |resource: &mut Resource,
                    status: ResourceStatus,
-                   chunked_file: Option<&ChunkedFile>| {
+                   chunked: Option<&ChunkedStorage>| {
                 resource.status = status;
-                resource.chunks_total = match chunked_file {
+                resource.chunks_total = match chunked {
                     Some(chunked_file) => chunked_file.len() as u64,
                     None => 0,
                 };
-                resource.chunks_downloaded = match chunked_file {
+                resource.chunks_downloaded = match chunked {
                     Some(chunked_file) => chunked_file.local_chunks() as u64,
                     None => 0,
                 };
 
+                if let Some(chunked) = chunked {
+                    resource.rtype = match chunked.is_dir() {
+                        false => ResourceType::File,
+                        true => ResourceType::Directory,
+                    };
+                }
+
                 self.event_publisher
                     .notify(FudEvent::ResourceUpdated(ResourceUpdated {
                         hash: resource.hash,
@@ -385,20 +397,24 @@ impl Fud {
                     continue;
                 }
             };
-            let chunked_file = match self.geode.get(&resource.hash, &resource_path).await {
+            let mut chunked = match self.geode.get(&resource.hash, &resource_path).await {
                 Ok(v) => v,
                 Err(_) => {
                     update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
                     continue;
                 }
             };
-            if !chunked_file.is_complete() {
-                update_resource(&mut resource, ResourceStatus::Incomplete, Some(&chunked_file))
-                    .await;
+            if let Err(e) = self.geode.verify_chunks(&mut chunked).await {
+                error!(target: "fud::verify_resources()", "Error while verifying chunks of {}: {}", hash_to_string(&resource.hash), e);
+                update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
+                continue;
+            }
+            if !chunked.is_complete() {
+                update_resource(&mut resource, ResourceStatus::Incomplete, Some(&chunked)).await;
                 continue;
             }
 
-            update_resource(&mut resource, ResourceStatus::Seeding, Some(&chunked_file)).await;
+            update_resource(&mut resource, ResourceStatus::Seeding, Some(&chunked)).await;
             seeding_resources.push(resource.clone());
         }
 
@@ -458,15 +474,24 @@ impl Fud {
         seeders
     }
 
-    /// Fetch chunks for a file from `seeders`
-    async fn fetch_chunks(
+    /// Fetch chunks for `chunked` (file or directory) from `seeders`.
+    async fn fetch_missing_chunks(
         &self,
-        file_path: &PathBuf,
-        file_hash: &blake3::Hash,
-        chunk_hashes: &HashSet<blake3::Hash>,
+        hash: &blake3::Hash,
+        chunked: &mut ChunkedStorage,
         seeders: &HashSet<DhtRouterItem>,
     ) -> Result<()> {
-        let mut remaining_chunks = chunk_hashes.clone();
+        let missing_chunks: HashSet<blake3::Hash> = {
+            let mut missing_chunks = HashSet::new();
+            for (chunk, available) in chunked.iter() {
+                if !available {
+                    missing_chunks.insert(*chunk);
+                }
+            }
+            missing_chunks
+        };
+
+        let mut remaining_chunks = missing_chunks.clone();
         let mut shuffled_seeders = {
             let mut vec: Vec<_> = seeders.iter().cloned().collect();
             vec.shuffle(&mut OsRng);
@@ -477,7 +502,7 @@ impl Fud {
             let channel = match self.get_channel(&seeder.node).await {
                 Ok(channel) => channel,
                 Err(e) => {
-                    warn!(target: "fud::fetch_chunks()", "Could not get a channel for node {}: {}", hash_to_string(&seeder.node.id), e);
+                    warn!(target: "fud::fetch_missing_chunks()", "Could not get a channel for node {}: {}", hash_to_string(&seeder.node.id), e);
                     continue;
                 }
             };
@@ -491,22 +516,22 @@ impl Fud {
                 let msg_subscriber_notfound = channel.subscribe_msg::<FudNotFound>().await.unwrap();
 
                 // Select a chunk to request
-                let mut chunk_hash: Option<blake3::Hash> = None;
-                if let Some(&random_chunk) = chunks_to_query.iter().choose(&mut OsRng) {
-                    chunk_hash = Some(random_chunk);
+                let mut chunk = None;
+                if let Some(random_chunk) = chunks_to_query.iter().choose(&mut OsRng) {
+                    chunk = Some(*random_chunk);
                 }
 
-                if chunk_hash.is_none() {
+                if chunk.is_none() {
                     // No more chunks to request from this seeder
                     break; // Switch to another seeder
                 }
-                let chunk_hash = chunk_hash.unwrap();
+                let chunk_hash = chunk.unwrap();
                 chunks_to_query.remove(&chunk_hash);
 
                 let send_res =
-                    channel.send(&FudFindRequest { info: Some(*file_hash), key: chunk_hash }).await;
+                    channel.send(&FudFindRequest { info: Some(*hash), key: chunk_hash }).await;
                 if let Err(e) = send_res {
-                    warn!(target: "fud::fetch_chunks()", "Error while sending FudFindRequest: {}", e);
+                    warn!(target: "fud::fetch_missing_chunks()", "Error while sending FudFindRequest: {}", e);
                     break; // Switch to another seeder
                 }
 
@@ -521,15 +546,15 @@ impl Fud {
                 select! {
                     chunk_reply = chunk_recv => {
                         if let Err(e) = chunk_reply {
-                            warn!(target: "fud::fetch_chunks()", "Error waiting for chunk reply: {}", e);
+                            warn!(target: "fud::fetch_missing_chunks()", "Error waiting for chunk reply: {}", e);
                             break; // Switch to another seeder
                         }
                         let reply = chunk_reply.unwrap();
 
-                        match self.geode.write_chunk(file_hash, file_path, &reply.chunk).await {
+                        match self.geode.write_chunk(chunked, &reply.chunk).await {
                             Ok(inserted_hash) => {
                                 if inserted_hash != chunk_hash {
-                                    warn!("Received chunk does not match requested chunk");
+                                    warn!(target: "fud::fetch_missing_chunks()", "Received chunk does not match requested chunk");
                                     msg_subscriber_chunk.unsubscribe().await;
                                     msg_subscriber_notfound.unsubscribe().await;
                                     continue; // Skip to next chunk, will retry this chunk later
@@ -537,7 +562,7 @@ impl Fud {
 
                                 // Update resource `chunks_downloaded`
                                 let mut resources_write = self.resources.write().await;
-                                let resource = match resources_write.get_mut(file_hash) {
+                                let resource = match resources_write.get_mut(hash) {
                                     Some(resource) => {
                                         resource.status = ResourceStatus::Downloading;
                                         resource.chunks_downloaded += 1;
@@ -547,10 +572,10 @@ impl Fud {
                                 };
                                 drop(resources_write);
 
-                                info!(target: "fud::fetch_chunks()", "Received chunk {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
+                                info!(target: "fud::fetch_missing_chunks()", "Received chunk {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
                                 self.event_publisher
                                     .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
-                                        hash: *file_hash,
+                                        hash: *hash,
                                         chunk_hash,
                                         resource,
                                     }))
@@ -558,21 +583,21 @@ impl Fud {
                                 remaining_chunks.remove(&chunk_hash);
                             }
                             Err(e) => {
-                                error!("Failed inserting chunk {} to Geode: {}", hash_to_string(&chunk_hash), e);
+                                error!(target: "fud::fetch_missing_chunks()", "Failed inserting chunk {} to Geode: {}", hash_to_string(&chunk_hash), e);
                             }
                         };
                     }
                     notfound_reply = notfound_recv => {
                         if let Err(e) = notfound_reply {
-                            warn!(target: "fud::fetch_chunks()", "Error waiting for NOTFOUND reply: {}", e);
+                            warn!(target: "fud::fetch_missing_chunks()", "Error waiting for NOTFOUND reply: {}", e);
                             msg_subscriber_chunk.unsubscribe().await;
                             msg_subscriber_notfound.unsubscribe().await;
                             break; // Switch to another seeder
                         }
-                        info!(target: "fud::fetch_chunks()", "Received NOTFOUND {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
+                        info!(target: "fud::fetch_missing_chunks()", "Received NOTFOUND {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
                         self.event_publisher
                             .notify(FudEvent::ChunkNotFound(ChunkNotFound {
-                                hash: *file_hash,
+                                hash: *hash,
                                 chunk_hash,
                             }))
                         .await;
@@ -592,24 +617,25 @@ impl Fud {
         Ok(())
     }
 
-    /// Fetch a single file metadata from `nodes`.
-    /// If the file is smaller than a single chunk then the chunk is returned.
-    /// 1. Request seeders for the file from those nodes
-    /// 2. Request the file from the seeders
-    pub async fn fetch_file_metadata(
+    /// Fetch a single resource metadata from `nodes`.
+    /// If the file is smaller than a single chunk then seeder can send the
+    /// chunk directly, we will create the file from it on path `path`.
+    /// 1. Request seeders from those nodes
+    /// 2. Request the metadata from the seeders
+    pub async fn fetch_metadata(
         &self,
-        nodes: Vec<DhtNode>,
-        file_hash: blake3::Hash,
+        nodes: &Vec<DhtNode>,
+        hash: &blake3::Hash,
     ) -> Option<FetchReply> {
         let mut queried_seeders: HashSet<blake3::Hash> = HashSet::new();
         let mut result: Option<FetchReply> = None;
 
         for node in nodes {
             // 1. Request list of seeders
-            let channel = match self.get_channel(&node).await {
+            let channel = match self.get_channel(node).await {
                 Ok(channel) => channel,
                 Err(e) => {
-                    warn!(target: "fud::fetch_file_metadata()", "Could not get a channel for node {}: {}", hash_to_string(&node.id), e);
+                    warn!(target: "fud::fetch_metadata()", "Could not get a channel for node {}: {}", hash_to_string(&node.id), e);
                     continue;
                 }
             };
@@ -619,14 +645,14 @@ impl Fud {
             let msg_subscriber = match channel.subscribe_msg::<FudFindSeedersReply>().await {
                 Ok(msg_subscriber) => msg_subscriber,
                 Err(e) => {
-                    warn!(target: "fud::fetch_file_metadata()", "Error subscribing to msg: {}", e);
+                    warn!(target: "fud::fetch_metadata()", "Error subscribing to msg: {}", e);
                     continue;
                 }
             };
 
-            let send_res = channel.send(&FudFindSeedersRequest { key: file_hash }).await;
+            let send_res = channel.send(&FudFindSeedersRequest { key: *hash }).await;
             if let Err(e) = send_res {
-                warn!(target: "fud::fetch_file_metadata()", "Error while sending FudFindSeedersRequest: {}", e);
+                warn!(target: "fud::fetch_metadata()", "Error while sending FudFindSeedersRequest: {}", e);
                 msg_subscriber.unsubscribe().await;
                 continue;
             }
@@ -635,14 +661,14 @@ impl Fud {
             {
                 Ok(reply) => reply,
                 Err(e) => {
-                    warn!(target: "fud::fetch_file_metadata()", "Error waiting for reply: {}", e);
+                    warn!(target: "fud::fetch_metadata()", "Error waiting for reply: {}", e);
                     msg_subscriber.unsubscribe().await;
                     continue;
                 }
             };
 
             let mut seeders = reply.seeders.clone();
-            info!(target: "fud::fetch_file_metadata()", "Found {} seeders for {} (from {})", seeders.len(), hash_to_string(&file_hash), hash_to_string(&node.id));
+            info!(target: "fud::fetch_metadata()", "Found {} seeders for {} (from {})", seeders.len(), hash_to_string(hash), hash_to_string(&node.id));
 
             msg_subscriber.unsubscribe().await;
 
@@ -658,20 +684,23 @@ impl Fud {
                     let msg_subsystem = channel.message_subsystem();
                     msg_subsystem.add_dispatch::<FudChunkReply>().await;
                     msg_subsystem.add_dispatch::<FudFileReply>().await;
+                    msg_subsystem.add_dispatch::<FudDirectoryReply>().await;
                     msg_subsystem.add_dispatch::<FudNotFound>().await;
                     let msg_subscriber_chunk =
                         channel.subscribe_msg::<FudChunkReply>().await.unwrap();
                     let msg_subscriber_file =
                         channel.subscribe_msg::<FudFileReply>().await.unwrap();
+                    let msg_subscriber_dir =
+                        channel.subscribe_msg::<FudDirectoryReply>().await.unwrap();
                     let msg_subscriber_notfound =
                         channel.subscribe_msg::<FudNotFound>().await.unwrap();
 
-                    let send_res =
-                        channel.send(&FudFindRequest { info: None, key: file_hash }).await;
+                    let send_res = channel.send(&FudFindRequest { info: None, key: *hash }).await;
                     if let Err(e) = send_res {
-                        warn!(target: "fud::fetch_file_metadata()", "Error while sending FudFindRequest: {}", e);
+                        warn!(target: "fud::fetch_metadata()", "Error while sending FudFindRequest: {}", e);
                         msg_subscriber_chunk.unsubscribe().await;
                         msg_subscriber_file.unsubscribe().await;
+                        msg_subscriber_dir.unsubscribe().await;
                         msg_subscriber_notfound.unsubscribe().await;
                         continue;
                     }
@@ -680,60 +709,84 @@ impl Fud {
                         msg_subscriber_chunk.receive_with_timeout(self.chunk_timeout).fuse();
                     let file_recv =
                         msg_subscriber_file.receive_with_timeout(self.chunk_timeout).fuse();
+                    let dir_recv =
+                        msg_subscriber_dir.receive_with_timeout(self.chunk_timeout).fuse();
                     let notfound_recv =
                         msg_subscriber_notfound.receive_with_timeout(self.chunk_timeout).fuse();
 
-                    pin_mut!(chunk_recv, file_recv, notfound_recv);
+                    pin_mut!(chunk_recv, file_recv, dir_recv, notfound_recv);
 
                     let cleanup = async || {
                         msg_subscriber_chunk.unsubscribe().await;
                         msg_subscriber_file.unsubscribe().await;
+                        msg_subscriber_dir.unsubscribe().await;
                         msg_subscriber_notfound.unsubscribe().await;
                     };
 
-                    // Wait for a FudChunkReply, FudFileReply, or FudNotFound
+                    // Wait for a FudChunkReply, FudFileReply, FudDirectoryReply, or FudNotFound
                     select! {
-                        // Received a chunk while requesting a file, this is allowed to
+                        // Received a chunk while requesting metadata, this is allowed to
                         // optimize fetching files smaller than a single chunk
                         chunk_reply = chunk_recv => {
                             cleanup().await;
                             if let Err(e) = chunk_reply {
-                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for chunk reply: {}", e);
+                                warn!(target: "fud::fetch_metadata()", "Error waiting for chunk reply: {}", e);
                                 continue;
                             }
                             let reply = chunk_reply.unwrap();
                             let chunk_hash = blake3::hash(&reply.chunk);
                             // Check that this is the only chunk in the file
-                            if !self.geode.verify_file(&file_hash, &[chunk_hash]) {
-                                warn!(target: "fud::fetch_file_metadata()", "Received a chunk while fetching a file, the chunk did not match the file hash");
+                            if !self.geode.verify_metadata(hash, &[chunk_hash], &[]) {
+                                warn!(target: "fud::fetch_metadata()", "Received a chunk while fetching a file, the chunk did not match the file hash");
                                 continue;
                             }
-                            info!(target: "fud::fetch_file_metadata()", "Received chunk {} (for file {}) from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
+                            info!(target: "fud::fetch_metadata()", "Received chunk {} (for file {}) from seeder {}", hash_to_string(&chunk_hash), hash_to_string(hash), hash_to_string(&seeder.node.id));
                             result = Some(FetchReply::Chunk((*reply).clone()));
                             break;
                         }
                         file_reply = file_recv => {
                             cleanup().await;
                             if let Err(e) = file_reply {
-                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for file reply: {}", e);
+                                warn!(target: "fud::fetch_metadata()", "Error waiting for file reply: {}", e);
                                 continue;
                             }
                             let reply = file_reply.unwrap();
-                            if !self.geode.verify_file(&file_hash, &reply.chunk_hashes) {
-                                warn!(target: "fud::fetch_file_metadata()", "Received invalid file metadata");
+                            if !self.geode.verify_metadata(hash, &reply.chunk_hashes, &[]) {
+                                warn!(target: "fud::fetch_metadata()", "Received invalid file metadata");
                                 continue;
                             }
-                            info!(target: "fud::fetch_file_metadata()", "Received file {} from seeder {}", hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
+                            info!(target: "fud::fetch_metadata()", "Received file {} from seeder {}", hash_to_string(hash), hash_to_string(&seeder.node.id));
                             result = Some(FetchReply::File((*reply).clone()));
                             break;
                         }
+                        dir_reply = dir_recv => {
+                            cleanup().await;
+                            if let Err(e) = dir_reply {
+                                warn!(target: "fud::fetch_metadata()", "Error waiting for directory reply: {}", e);
+                                continue;
+                            }
+                            let reply = dir_reply.unwrap();
+
+                            // Convert all file paths from String to PathBuf
+                            let files: Vec<_> = reply.files.clone().into_iter()
+                                .map(|(path_str, size)| (PathBuf::from(path_str), size))
+                                .collect();
+
+                            if !self.geode.verify_metadata(hash, &reply.chunk_hashes, &files) {
+                                warn!(target: "fud::fetch_metadata()", "Received invalid directory metadata");
+                                continue;
+                            }
+                            info!(target: "fud::fetch_metadata()", "Received directory {} from seeder {}", hash_to_string(hash), hash_to_string(&seeder.node.id));
+                            result = Some(FetchReply::Directory((*reply).clone()));
+                            break;
+                        }
                         notfound_reply = notfound_recv => {
                             cleanup().await;
                             if let Err(e) = notfound_reply {
-                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for NOTFOUND reply: {}", e);
+                                warn!(target: "fud::fetch_metadata()", "Error waiting for NOTFOUND reply: {}", e);
                                 continue;
                             }
-                            info!(target: "fud::fetch_file_metadata()", "Received NOTFOUND {} from seeder {}", hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
+                            info!(target: "fud::fetch_metadata()", "Received NOTFOUND {} from seeder {}", hash_to_string(hash), hash_to_string(&seeder.node.id));
                         }
                     };
                 }
@@ -747,77 +800,88 @@ impl Fud {
         result
     }
 
-    /// Download a file from the network to `file_path`.
-    pub async fn get(&self, file_hash: &blake3::Hash, file_path: &PathBuf) -> Result<()> {
+    /// Download a file or directory from the network to `path`.
+    /// This creates a new task in `fetch_tasks` calling `fetch_resource()`.
+    pub async fn get(&self, hash: &blake3::Hash, path: &Path) -> Result<()> {
         let self_node = self.dht().node().await;
         let mut closest_nodes = vec![];
 
+        let hash_bytes = hash.as_bytes();
+        let path_string = path.to_string_lossy().to_string();
+        let path_bytes = path_string.as_bytes();
+
+        // Make sure we don't already have another resource on that path
+        for path_item in self.path_tree.iter() {
+            let (key, value) = path_item?;
+            if key != hash_bytes && value == path_bytes {
+                let err_str = format!("There is already another resource on path {}", path_string);
+                self.event_publisher
+                    .notify(FudEvent::DownloadError(event::DownloadError {
+                        hash: *hash,
+                        error: err_str.clone(),
+                    }))
+                    .await;
+                return Err(Error::Custom(err_str))
+            }
+        }
+
         // Add path to the sled db
-        self.path_tree
-            .insert(file_hash.as_bytes(), file_path.to_string_lossy().to_string().as_bytes())?;
+        self.path_tree.insert(hash_bytes, path_bytes)?;
 
         // Add resource to `self.resources`
         let resource = Resource {
-            hash: *file_hash,
-            path: file_path.clone(),
+            hash: *hash,
+            rtype: ResourceType::Unknown,
+            path: path.to_path_buf(),
             status: ResourceStatus::Discovering,
             chunks_total: 0,
             chunks_downloaded: 0,
         };
         let mut resources_write = self.resources.write().await;
-        resources_write.insert(*file_hash, resource.clone());
+        resources_write.insert(*hash, resource.clone());
         drop(resources_write);
 
         // Send a DownloadStarted event
         self.event_publisher
-            .notify(FudEvent::DownloadStarted(event::DownloadStarted {
-                hash: *file_hash,
-                resource,
-            }))
+            .notify(FudEvent::DownloadStarted(event::DownloadStarted { hash: *hash, resource }))
             .await;
 
-        // Try to get the chunked file from geode
-        let chunked_file = match self.geode.get(file_hash, file_path).await {
-            // We already know the list of chunk hashes for this file
+        // Try to get the chunked file or directory from geode
+        let mut chunked = match self.geode.get(hash, path).await {
+            // We already know the metadata
             Ok(v) => v,
             // The metadata in geode is invalid or corrupted
             Err(Error::GeodeNeedsGc) => todo!(),
-            // If we could not find the file in geode, get the file metadata from the network
+            // If we could not find the metadata in geode, get it from the network
             Err(Error::GeodeFileNotFound) => {
                 // Find nodes close to the file hash
-                info!(target: "self::get()", "Requested file {} not found in Geode, triggering fetch", hash_to_string(file_hash));
-                closest_nodes = self.lookup_nodes(file_hash).await.unwrap_or_default();
+                info!(target: "fud::get()", "Requested metadata {} not found in Geode, triggering fetch", hash_to_string(hash));
+                closest_nodes = self.lookup_nodes(hash).await.unwrap_or_default();
 
-                // Fetch file metadata (list of chunk hashes)
-                self.file_fetch_tx
-                    .send((closest_nodes.clone(), *file_hash, file_path.clone(), Ok(())))
+                // Fetch file or directory metadata
+                self.metadata_fetch_tx
+                    .send((closest_nodes.clone(), *hash, path.to_path_buf()))
                     .await
                     .unwrap();
                 info!(target: "self::get()", "Waiting for background file fetch task...");
-                let (i_file_hash, status) = self.file_fetch_end_rx.recv().await.unwrap();
-                match status {
+                match self.metadata_fetch_end_rx.recv().await.unwrap() {
                     // The file metadata was found and inserted into geode
-                    Ok(()) => self.geode.get(&i_file_hash, file_path).await.unwrap(),
-                    // We could not find the file metadata
-                    Err(Error::GeodeFileRouteNotFound) => {
+                    Ok(()) => self.geode.get(hash, path).await.unwrap(),
+                    // We could not find the metadata, or any other error occured
+                    Err(e) => {
                         // Set resource status to `Incomplete` and send FudEvent::FileNotFound
                         let mut resources_write = self.resources.write().await;
-                        if let Some(resource) = resources_write.get_mut(file_hash) {
+                        if let Some(resource) = resources_write.get_mut(hash) {
                             resource.status = ResourceStatus::Incomplete;
 
                             self.event_publisher
-                                .notify(FudEvent::FileNotFound(event::FileNotFound {
-                                    hash: *file_hash,
+                                .notify(FudEvent::MetadataNotFound(event::MetadataNotFound {
+                                    hash: *hash,
                                     resource: resource.clone(),
                                 }))
                                 .await;
                         }
                         drop(resources_write);
-                        return Err(Error::GeodeFileRouteNotFound);
-                    }
-
-                    Err(e) => {
-                        error!(target: "fud::handle_get()", "{}", e);
                         return Err(e);
                     }
                 }
@@ -829,40 +893,72 @@ impl Fud {
             }
         };
 
+        // Create all files (and all necessary directories)
+        for (file_path, _) in chunked.get_files().iter() {
+            if !file_path.exists() {
+                if let Some(dir) = path.join(file_path).parent() {
+                    fs::create_dir_all(dir).await?;
+                }
+                File::create(&file_path).await?;
+            }
+        }
+
+        // Mark locally available chunks as such
+        if let Err(e) = self.geode.verify_chunks(&mut chunked).await {
+            error!(target: "self::get()", "Error while verifying chunks: {}", e);
+            return Err(e);
+        }
+
         // Set resource status to `Downloading`
         let mut resources_write = self.resources.write().await;
-        let resource = match resources_write.get_mut(file_hash) {
+        let resource = match resources_write.get_mut(hash) {
             Some(resource) => {
                 resource.status = ResourceStatus::Downloading;
-                resource.chunks_downloaded = chunked_file.local_chunks() as u64;
-                resource.chunks_total = chunked_file.len() as u64;
+                resource.chunks_downloaded = chunked.local_chunks() as u64;
+                resource.chunks_total = chunked.len() as u64;
+                resource.rtype = match chunked.is_dir() {
+                    false => ResourceType::File,
+                    true => ResourceType::Directory,
+                };
                 resource.clone()
             }
             None => return Ok(()), // Resource was removed, abort
         };
         drop(resources_write);
 
-        // Send a FileDownloadCompleted event
+        // Send a MetadataDownloadCompleted event
         self.event_publisher
-            .notify(FudEvent::FileDownloadCompleted(event::FileDownloadCompleted {
-                hash: *file_hash,
+            .notify(FudEvent::MetadataDownloadCompleted(event::MetadataDownloadCompleted {
+                hash: *hash,
                 resource: resource.clone(),
             }))
             .await;
 
+        // If `chunked` is a file that is bigger than the all its chunks,
+        // truncate the file to the chunks.
+        // This fixes two edge-cases: a file that exactly ends at the end of
+        // a chunk, and a file with no chunk.
+        if !chunked.is_dir() {
+            let fs_metadata = fs::metadata(&path).await?;
+            if fs_metadata.len() > (chunked.len() * MAX_CHUNK_SIZE) as u64 {
+                if let Ok(file) = OpenOptions::new().write(true).create(true).open(path).await {
+                    let _ = file.set_len((chunked.len() * MAX_CHUNK_SIZE) as u64).await;
+                }
+            }
+        }
+
         // If the file is already complete, we don't need to download any chunk
-        if chunked_file.is_complete() {
+        if chunked.is_complete() {
             // Announce the file
-            let self_announce =
-                FudAnnounce { key: *file_hash, seeders: vec![self_node.clone().into()] };
-            let _ = self.announce(file_hash, &self_announce, self.seeders_router.clone()).await;
+            let self_announce = FudAnnounce { key: *hash, seeders: vec![self_node.clone().into()] };
+            let _ = self.announce(hash, &self_announce, self.seeders_router.clone()).await;
 
             // Set resource status to `Seeding`
             let mut resources_write = self.resources.write().await;
-            let resource = match resources_write.get_mut(file_hash) {
+            let resource = match resources_write.get_mut(hash) {
                 Some(resource) => {
                     resource.status = ResourceStatus::Seeding;
-                    resource.chunks_downloaded = chunked_file.len() as u64;
+                    resource.chunks_downloaded = chunked.len() as u64;
                     resource.clone()
                 }
                 None => return Ok(()), // Resource was removed, abort
@@ -872,7 +968,7 @@ impl Fud {
             // Send a DownloadCompleted event
             self.event_publisher
                 .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
-                    hash: *file_hash,
+                    hash: *hash,
                     resource,
                 }))
                 .await;
@@ -882,44 +978,37 @@ impl Fud {
 
         // Find nodes close to the file hash if we didn't previously fetched them
         if closest_nodes.is_empty() {
-            closest_nodes = self.lookup_nodes(file_hash).await.unwrap_or_default();
+            closest_nodes = self.lookup_nodes(hash).await.unwrap_or_default();
         }
 
         // Find seeders and remove ourselves from the result
         let seeders = self
-            .fetch_seeders(&closest_nodes, file_hash)
+            .fetch_seeders(&closest_nodes, hash)
             .await
             .iter()
             .filter(|seeder| seeder.node.id != self_node.id)
             .cloned()
             .collect();
 
-        // List missing chunks
-        let mut missing_chunks = HashSet::new();
-        for (chunk, path) in chunked_file.iter() {
-            if path.is_none() {
-                missing_chunks.insert(*chunk);
-            }
-        }
-
         // Fetch missing chunks from seeders
-        self.fetch_chunks(file_path, file_hash, &missing_chunks, &seeders).await?;
+        self.fetch_missing_chunks(hash, &mut chunked, &seeders).await?;
 
         // Get chunked file from geode
-        let chunked_file = match self.geode.get(file_hash, file_path).await {
+        let mut chunked = match self.geode.get(hash, path).await {
             Ok(v) => v,
             Err(e) => {
                 error!(target: "fud::handle_get()", "{}", e);
                 return Err(e);
             }
         };
+        self.geode.verify_chunks(&mut chunked).await?;
 
         // We fetched all chunks, but the file is not complete
         // (some chunks were missing from all seeders)
-        if !chunked_file.is_complete() {
+        if !chunked.is_complete() {
             // Set resource status to `Incomplete`
             let mut resources_write = self.resources.write().await;
-            let resource = match resources_write.get_mut(file_hash) {
+            let resource = match resources_write.get_mut(hash) {
                 Some(resource) => {
                     resource.status = ResourceStatus::Incomplete;
                     resource.clone()
@@ -930,25 +1019,21 @@ impl Fud {
 
             // Send a MissingChunks event
             self.event_publisher
-                .notify(FudEvent::MissingChunks(event::MissingChunks {
-                    hash: *file_hash,
-                    resource,
-                }))
+                .notify(FudEvent::MissingChunks(event::MissingChunks { hash: *hash, resource }))
                 .await;
             return Ok(());
         }
 
         // Announce the file
-        let self_announce =
-            FudAnnounce { key: *file_hash, seeders: vec![self_node.clone().into()] };
-        let _ = self.announce(file_hash, &self_announce, self.seeders_router.clone()).await;
+        let self_announce = FudAnnounce { key: *hash, seeders: vec![self_node.clone().into()] };
+        let _ = self.announce(hash, &self_announce, self.seeders_router.clone()).await;
 
         // Set resource status to `Seeding`
         let mut resources_write = self.resources.write().await;
-        let resource = match resources_write.get_mut(file_hash) {
+        let resource = match resources_write.get_mut(hash) {
             Some(resource) => {
                 resource.status = ResourceStatus::Seeding;
-                resource.chunks_downloaded = chunked_file.len() as u64;
+                resource.chunks_downloaded = chunked.len() as u64;
                 resource.clone()
             }
             None => return Ok(()), // Resource was removed, abort
@@ -957,12 +1042,111 @@ impl Fud {
 
         // Send a DownloadCompleted event
         self.event_publisher
-            .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
-                hash: *file_hash,
-                resource,
-            }))
+            .notify(FudEvent::DownloadCompleted(event::DownloadCompleted { hash: *hash, resource }))
             .await;
 
         Ok(())
     }
+
+    /// Add a resource from the file system.
+    pub async fn put(&self, path: &PathBuf) -> Result<blake3::Hash> {
+        let self_node = self.dht.node().await;
+
+        if self_node.addresses.is_empty() {
+            return Err(Error::Custom(
+                "Cannot put file, you don't have any external address".to_string(),
+            ))
+        }
+
+        let metadata = fs::metadata(path).await?;
+
+        // Get the list of files and the resource type (file or directory)
+        let (files, resource_type) = if metadata.is_file() {
+            (vec![(path.clone(), metadata.len())], ResourceType::File)
+        } else if metadata.is_dir() {
+            let mut files = get_all_files(path).await?;
+            self.geode.sort_files(&mut files);
+            (files, ResourceType::Directory)
+        } else {
+            return Err(Error::Custom(format!("{} is not a valid path", path.to_string_lossy())))
+        };
+
+        // Read the file or directory and create the chunks
+        let stream = FileSequence::new(&files, false);
+        let (mut hasher, chunk_hashes) = self.geode.chunk_stream(stream).await?;
+
+        // Get the relative file paths included in the metadata and hash of directories
+        let relative_files = if let ResourceType::Directory = resource_type {
+            // [(absolute file path, file size)] -> [(relative file path, file size)]
+            let relative_files = files
+                .into_iter()
+                .map(|(file_path, size)| match file_path.strip_prefix(path) {
+                    Ok(rel_path) => Ok((rel_path.to_path_buf(), size)),
+                    Err(_) => Err(Error::Custom("Invalid file path".to_string())),
+                })
+                .collect::<Result<Vec<_>>>()?;
+
+            // Add the files metadata to the hasher to complete the resource hash
+            self.geode.hash_files_metadata(&mut hasher, &relative_files);
+
+            relative_files
+        } else {
+            vec![]
+        };
+
+        // Finalize the resource hash
+        let hash = hasher.finalize();
+
+        // Create the metadata file in geode
+        if let Err(e) = self.geode.insert_metadata(&hash, &chunk_hashes, &relative_files).await {
+            error!(target: "fud::put()", "Failed inserting {:?} to geode: {}", path, e);
+            return Err(e)
+        }
+
+        // Add path to the sled db
+        if let Err(e) =
+            self.path_tree.insert(hash.as_bytes(), path.to_string_lossy().to_string().as_bytes())
+        {
+            error!(target: "fud::put()", "Failed inserting new resource into sled: {}", e);
+            return Err(e.into())
+        }
+
+        // Add resource
+        let mut resources_write = self.resources.write().await;
+        resources_write.insert(
+            hash,
+            Resource {
+                hash,
+                rtype: resource_type,
+                path: path.to_path_buf(),
+                status: ResourceStatus::Seeding,
+                chunks_total: chunk_hashes.len() as u64,
+                chunks_downloaded: chunk_hashes.len() as u64,
+            },
+        );
+        drop(resources_write);
+
+        // Announce the new resource
+        let fud_announce = FudAnnounce { key: hash, seeders: vec![self_node.into()] };
+        let _ = self.announce(&hash, &fud_announce, self.seeders_router.clone()).await;
+
+        Ok(hash)
+    }
+
+    /// Remove a resource, its metadata in geode, and its path in the sled path tree.
+    pub async fn remove(&self, hash: &blake3::Hash) {
+        let mut resources_write = self.resources.write().await;
+        resources_write.remove(hash);
+        drop(resources_write);
+
+        let hash_str = hash_to_string(hash);
+        let _ = fs::remove_file(self.geode.files_path.join(&hash_str)).await;
+        let _ = fs::remove_file(self.geode.dirs_path.join(&hash_str)).await;
+
+        let _ = self.path_tree.remove(hash.as_bytes());
+
+        self.event_publisher
+            .notify(FudEvent::ResourceRemoved(event::ResourceRemoved { hash: *hash }))
+            .await;
+    }
 }

bin/fud/fud/src/main.rs

@@ -41,7 +41,7 @@ use fud::{
     get_node_id,
     proto::{FudFindNodesReply, ProtocolFud},
     rpc::JsonRpcInterface,
-    tasks::{announce_seed_task, fetch_file_task, get_task},
+    tasks::{announce_seed_task, fetch_metadata_task, get_task},
     Fud,
 };
 
@@ -179,14 +179,14 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         ex.clone(),
     );
 
-    info!(target: "fud", "Starting fetch file task");
+    info!(target: "fud", "Starting fetch metadata task");
     let file_task = StoppableTask::new();
     file_task.clone().start(
-        fetch_file_task(fud.clone()),
+        fetch_metadata_task(fud.clone()),
         |res| async {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "fud", "Failed starting fetch file task: {}", e),
+                Err(e) => error!(target: "fud", "Failed starting fetch metadata task: {}", e),
             }
         },
         Error::DetachedTaskStopped,

bin/fud/fud/src/proto.rs

@@ -19,7 +19,7 @@
 use async_trait::async_trait;
 use log::{debug, error, info};
 use smol::Executor;
-use std::sync::Arc;
+use std::{path::StripPrefixError, sync::Arc};
 
 use darkfi::{
     dht::{DhtHandler, DhtNode, DhtRouterItem},
@@ -43,6 +43,14 @@ pub struct FudFileReply {
 }
 impl_p2p_message!(FudFileReply, "FudFileReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
+/// Message representing a directory reply from the network
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct FudDirectoryReply {
+    pub chunk_hashes: Vec<blake3::Hash>,
+    pub files: Vec<(String, u64)>, // Vec of (file path, file size)
+}
+impl_p2p_message!(FudDirectoryReply, "FudDirectoryReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+
 /// Message representing a node announcing a key on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudAnnounce {
@@ -196,7 +204,7 @@ impl ProtocolFud {
                     continue
                 }
             };
-            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Received FIND");
+            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Received FIND for {}", hash_to_string(&request.key));
 
             let node = self.fud.dht().get_node_from_channel(self.channel.info.id).await;
             if let Some(node) = node {
@@ -207,7 +215,7 @@ impl ProtocolFud {
                 continue;
             }
 
-            if self.handle_fud_file_request(&request).await {
+            if self.handle_fud_metadata_request(&request).await {
                 continue;
             }
 
@@ -221,66 +229,103 @@ impl ProtocolFud {
     /// If the FudFindRequest matches a chunk we have, handle it.
     /// Returns true if the chunk was found.
     async fn handle_fud_chunk_request(&self, request: &FudFindRequest) -> bool {
-        let file_hash = request.info;
-        if file_hash.is_none() {
+        let hash = request.info;
+        if hash.is_none() {
             return false;
         }
-        let file_hash = file_hash.unwrap();
+        let hash = hash.unwrap();
 
-        let file_path = self.fud.hash_to_path(&file_hash).ok().flatten();
-        if file_path.is_none() {
+        let path = self.fud.hash_to_path(&hash).ok().flatten();
+        if path.is_none() {
             return false;
         }
-        let file_path = file_path.unwrap();
+        let path = path.unwrap();
 
-        let chunk = self.fud.geode.get_chunk(&request.key, &file_hash, &file_path).await;
+        let chunked = self.fud.geode.get(&hash, &path).await;
+        if chunked.is_err() {
+            return false;
+        }
+
+        let chunk = self.fud.geode.get_chunk(&mut chunked.unwrap(), &request.key, &path).await;
         if let Ok(chunk) = chunk {
             // TODO: Run geode GC
             let reply = FudChunkReply { chunk };
-            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Sending chunk");
+            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Sending chunk {}", hash_to_string(&request.key));
             let _ = self.channel.send(&reply).await;
             return true;
         }
+
         false
     }
 
     /// If the FudFindRequest matches a file we have, handle it
     /// Returns true if the file was found.
-    async fn handle_fud_file_request(&self, request: &FudFindRequest) -> bool {
-        let file_path = self.fud.hash_to_path(&request.key).ok().flatten();
-        if file_path.is_none() {
+    async fn handle_fud_metadata_request(&self, request: &FudFindRequest) -> bool {
+        let path = self.fud.hash_to_path(&request.key).ok().flatten();
+        if path.is_none() {
             return false;
         }
-        let file_path = file_path.unwrap();
+        let path = path.unwrap();
 
-        let chunked_file = self.fud.geode.get(&request.key, &file_path).await.ok();
+        let chunked_file = self.fud.geode.get(&request.key, &path).await.ok();
         if chunked_file.is_none() {
             return false;
         }
-        let chunked_file = chunked_file.unwrap();
-
-        // If the file has a single chunk, just reply with the chunk
-        if chunked_file.len() == 1 {
-            let chunk = self
-                .fud
-                .geode
-                .get_chunk(&chunked_file.iter().next().unwrap().0, &request.key, &file_path)
-                .await;
+        let mut chunked_file = chunked_file.unwrap();
+
+        // If it's a file with a single chunk, just reply with the chunk
+        if chunked_file.len() == 1 && !chunked_file.is_dir() {
+            let chunk_hash = chunked_file.get_chunks()[0].0;
+            let chunk = self.fud.geode.get_chunk(&mut chunked_file, &chunk_hash, &path).await;
             if let Ok(chunk) = chunk {
                 // TODO: Run geode GC
                 let reply = FudChunkReply { chunk };
-                info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Sending chunk (file has a single chunk)");
+                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending chunk (file has a single chunk) {}", hash_to_string(&chunk_hash));
                 let _ = self.channel.send(&reply).await;
                 return true;
             }
             return false;
         }
 
-        // Otherwise reply with the file metadata
-        let reply =
-            FudFileReply { chunk_hashes: chunked_file.iter().map(|(chunk, _)| *chunk).collect() };
-        info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Sending file");
-        let _ = self.channel.send(&reply).await;
+        // Otherwise reply with the metadata
+        match chunked_file.is_dir() {
+            false => {
+                let reply = FudFileReply {
+                    chunk_hashes: chunked_file
+                        .get_chunks()
+                        .iter()
+                        .map(|(chunk, _)| *chunk)
+                        .collect(),
+                };
+                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending file metadata {}", hash_to_string(&request.key));
+                let _ = self.channel.send(&reply).await;
+            }
+            true => {
+                let files = chunked_file
+                    .get_files()
+                    .iter()
+                    .map(|(file_path, size)| match file_path.strip_prefix(path.clone()) {
+                        Ok(rel_path) => Ok((rel_path.to_string_lossy().to_string(), *size)),
+                        Err(e) => Err(e),
+                    })
+                    .collect::<std::result::Result<Vec<_>, StripPrefixError>>();
+                if let Err(e) = files {
+                    error!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Error parsing file paths before sending directory metadata: {}", e);
+                    return false;
+                }
+                let reply = FudDirectoryReply {
+                    chunk_hashes: chunked_file
+                        .get_chunks()
+                        .iter()
+                        .map(|(chunk, _)| *chunk)
+                        .collect(),
+                    files: files.unwrap(),
+                };
+                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending directory metadata {}", hash_to_string(&request.key));
+                let _ = self.channel.send(&reply).await;
+            }
+        };
+
         true
     }
 

bin/fud/fud/src/resource.rs

@@ -28,9 +28,17 @@ pub enum ResourceStatus {
     Incomplete,
 }
 
+#[derive(Clone, Debug)]
+pub enum ResourceType {
+    Unknown,
+    File,
+    Directory,
+}
+
 #[derive(Clone, Debug)]
 pub struct Resource {
     pub hash: blake3::Hash,
+    pub rtype: ResourceType,
     pub path: PathBuf,
     pub status: ResourceStatus,
     pub chunks_total: u64,
@@ -41,6 +49,17 @@ impl From<Resource> for JsonValue {
     fn from(rs: Resource) -> JsonValue {
         json_map([
             ("hash", JsonValue::String(hash_to_string(&rs.hash))),
+            (
+                "type",
+                JsonValue::String(
+                    match rs.rtype {
+                        ResourceType::Unknown => "unknown",
+                        ResourceType::File => "file",
+                        ResourceType::Directory => "directory",
+                    }
+                    .to_string(),
+                ),
+            ),
             (
                 "path",
                 JsonValue::String(match rs.path.into_os_string().into_string() {

bin/fud/fud/src/rpc.rs

@@ -18,10 +18,7 @@
 
 use async_trait::async_trait;
 use log::error;
-use smol::{
-    fs::{self, File},
-    lock::{Mutex, MutexGuard},
-};
+use smol::lock::{Mutex, MutexGuard};
 use std::{
     collections::{HashMap, HashSet},
     path::PathBuf,
@@ -30,7 +27,6 @@ use std::{
 use tinyjson::JsonValue;
 
 use darkfi::{
-    dht::DhtHandler,
     geode::hash_to_string,
     net::P2pPtr,
     rpc::{
@@ -43,12 +39,7 @@ use darkfi::{
     Result,
 };
 
-use crate::{
-    event::{self, FudEvent},
-    proto::FudAnnounce,
-    resource::{Resource, ResourceStatus},
-    Fud,
-};
+use crate::Fud;
 
 pub struct JsonRpcInterface {
     fud: Arc<Fud>,
@@ -66,7 +57,7 @@ impl RequestHandler<()> for JsonRpcInterface {
             "put" => self.put(req.id, req.params).await,
             "get" => self.get(req.id, req.params).await,
             "subscribe" => self.subscribe(req.id, req.params).await,
-            "remove" => self.remove_resource(req.id, req.params).await,
+            "remove" => self.remove(req.id, req.params).await,
             "list_resources" => self.list_resources(req.id, req.params).await,
             "list_buckets" => self.list_buckets(req.id, req.params).await,
             "list_seeders" => self.list_seeders(req.id, req.params).await,
@@ -103,18 +94,6 @@ impl JsonRpcInterface {
     // --> {"jsonrpc": "2.0", "method": "put", "params": ["/foo.txt"], "id": 42}
     // <-- {"jsonrpc": "2.0", "result: "df4...3db7", "id": 42}
     async fn put(&self, id: u16, params: JsonValue) -> JsonResult {
-        let self_node = self.fud.dht.node().await;
-
-        if self_node.addresses.is_empty() {
-            error!(target: "fud::put()", "Cannot put file, you don't have any external address");
-            return JsonError::new(
-                ErrorCode::InternalError,
-                Some("You don't have any external address".to_string()),
-                id,
-            )
-            .into()
-        }
-
         let params = params.get::<Vec<JsonValue>>().unwrap();
         if params.len() != 1 || !params[0].is_string() {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
@@ -128,52 +107,12 @@ impl JsonRpcInterface {
 
         // A valid path was passed. Let's see if we can read it, and if so,
         // add it to Geode.
-        let fd = match File::open(&path).await {
-            Ok(v) => v,
-            Err(e) => {
-                error!(target: "fud::put()", "Failed to open {:?}: {}", path, e);
-                return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-            }
-        };
-
-        let (file_hash, chunk_hashes) = match self.fud.geode.insert(fd).await {
-            Ok(v) => v,
-            Err(e) => {
-                let error_str = format!("Failed inserting file {:?} to geode: {}", path, e);
-                error!(target: "fud::put()", "{}", error_str);
-                return JsonError::new(ErrorCode::InternalError, Some(error_str), id).into()
-            }
-        };
-
-        // Add path to the sled db
-        if let Err(e) = self
-            .fud
-            .path_tree
-            .insert(file_hash.as_bytes(), path.to_string_lossy().to_string().as_bytes())
-        {
-            error!(target: "fud::put()", "Failed inserting new file into sled: {}", e);
-            return JsonError::new(ErrorCode::InternalError, None, id).into()
+        let res = self.fud.put(&path).await;
+        if let Err(e) = res {
+            return JsonError::new(ErrorCode::InternalError, Some(format!("{}", e)), id).into()
         }
 
-        // Add resource
-        let mut resources_write = self.fud.resources.write().await;
-        resources_write.insert(
-            file_hash,
-            Resource {
-                hash: file_hash,
-                path,
-                status: ResourceStatus::Seeding,
-                chunks_total: chunk_hashes.len() as u64,
-                chunks_downloaded: chunk_hashes.len() as u64,
-            },
-        );
-        drop(resources_write);
-
-        // Announce file
-        let fud_announce = FudAnnounce { key: file_hash, seeders: vec![self_node.into()] };
-        let _ = self.fud.announce(&file_hash, &fud_announce, self.fud.seeders_router.clone()).await;
-
-        JsonResponse::new(JsonValue::String(hash_to_string(&file_hash)), id).into()
+        JsonResponse::new(JsonValue::String(hash_to_string(&res.unwrap())), id).into()
     }
 
     // RPCAPI:
@@ -201,29 +140,26 @@ impl JsonRpcInterface {
         let mut hash_buf_arr = [0u8; 32];
         hash_buf_arr.copy_from_slice(&hash_buf);
 
-        let file_hash = blake3::Hash::from_bytes(hash_buf_arr);
-        let file_hash_str = hash_to_string(&file_hash);
+        let hash = blake3::Hash::from_bytes(hash_buf_arr);
+        let hash_str = hash_to_string(&hash);
 
-        let file_path = match params[1].get::<String>() {
+        let path = match params[1].get::<String>() {
             Some(path) => match path.is_empty() {
-                true => self.fud.downloads_path.join(&file_hash_str).join(&file_hash_str),
+                true => match self.fud.hash_to_path(&hash).ok().flatten() {
+                    Some(path) => path,
+                    None => self.fud.downloads_path.join(&hash_str),
+                },
                 false => match PathBuf::from(path).is_absolute() {
                     true => PathBuf::from(path),
-                    false => self.fud.downloads_path.join(&file_hash_str).join(path),
+                    false => self.fud.downloads_path.join(path),
                 },
             },
-            None => self.fud.downloads_path.join(&file_hash_str).join(&file_hash_str),
+            None => self.fud.downloads_path.join(&hash_str),
         };
 
-        // Get the parent directory of the file
-        if let Some(parent) = file_path.parent() {
-            // Create all directories leading up to the file
-            let _ = fs::create_dir_all(parent).await;
-        }
-
-        let _ = self.fud.get_tx.send((id, file_hash, file_path.clone(), Ok(()))).await;
+        let _ = self.fud.get_tx.send((hash, path.clone())).await;
 
-        JsonResponse::new(JsonValue::String(file_path.to_string_lossy().to_string()), id).into()
+        JsonResponse::new(JsonValue::String(path.to_string_lossy().to_string()), id).into()
     }
 
     // RPCAPI:
@@ -353,7 +289,7 @@ impl JsonRpcInterface {
     //
     // --> {"jsonrpc": "2.0", "method": "remove", "params": ["1211...abfd"], "id": 1}
     // <-- {"jsonrpc": "2.0", "result": [], "id": 1}
-    pub async fn remove_resource(&self, id: u16, params: JsonValue) -> JsonResult {
+    pub async fn remove(&self, id: u16, params: JsonValue) -> JsonResult {
         let params = params.get::<Vec<JsonValue>>().unwrap();
         if params.len() != 1 || !params[0].is_string() {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
@@ -364,15 +300,7 @@ impl JsonRpcInterface {
             Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
         }
 
-        let hash = blake3::Hash::from_bytes(hash_buf);
-        let mut resources_write = self.fud.resources.write().await;
-        resources_write.remove(&hash);
-        drop(resources_write);
-
-        self.fud
-            .event_publisher
-            .notify(FudEvent::ResourceRemoved(event::ResourceRemoved { hash }))
-            .await;
+        self.fud.remove(&blake3::Hash::from_bytes(hash_buf)).await;
 
         JsonResponse::new(JsonValue::Array(vec![]), id).into()
     }

bin/fud/fud/src/tasks.rs

@@ -17,16 +17,22 @@
  */
 
 use log::{error, info};
-use std::sync::Arc;
+use std::{path::PathBuf, sync::Arc};
 
-use darkfi::{dht::DhtHandler, geode::hash_to_string, system::sleep, Error, Result};
+use darkfi::{
+    dht::DhtHandler,
+    geode::{hash_to_string, ChunkedStorage},
+    system::sleep,
+    Error, Result,
+};
 
 use crate::{
-    proto::{FudAnnounce, FudChunkReply, FudFileReply},
+    proto::{FudAnnounce, FudChunkReply, FudDirectoryReply, FudFileReply},
     Fud,
 };
 
 pub enum FetchReply {
+    Directory(FudDirectoryReply),
     File(FudFileReply),
     Chunk(FudChunkReply),
 }
@@ -34,7 +40,7 @@ pub enum FetchReply {
 /// Triggered when calling the `get` RPC method
 pub async fn get_task(fud: Arc<Fud>) -> Result<()> {
     loop {
-        let (_, file_hash, file_path, _) = fud.get_rx.recv().await.unwrap();
+        let (file_hash, file_path) = fud.get_rx.recv().await.unwrap();
 
         let _ = fud.get(&file_hash, &file_path).await;
     }
@@ -43,53 +49,59 @@ pub async fn get_task(fud: Arc<Fud>) -> Result<()> {
 /// Background task that receives file fetch requests and tries to
 /// fetch objects from the network using the routing table.
 /// TODO: This can be optimised a lot for connection reuse, etc.
-pub async fn fetch_file_task(fud: Arc<Fud>) -> Result<()> {
-    info!(target: "fud::fetch_file_task()", "Started background file fetch task");
+pub async fn fetch_metadata_task(fud: Arc<Fud>) -> Result<()> {
+    info!(target: "fud::fetch_metadata_task()", "Started background metadata fetch task");
     loop {
-        let (nodes, file_hash, file_path, _) = fud.file_fetch_rx.recv().await.unwrap();
-        info!(target: "fud::fetch_file_task()", "Fetching file {}", hash_to_string(&file_hash));
-
-        let result = fud.fetch_file_metadata(nodes, file_hash).await;
-
-        match result {
-            Some(reply) => {
-                match reply {
-                    FetchReply::File(FudFileReply { chunk_hashes }) => {
-                        if let Err(e) = fud.geode.insert_file(&file_hash, &chunk_hashes).await {
-                            error!(
-                                "Failed inserting file {} to Geode: {}",
-                                hash_to_string(&file_hash),
-                                e
-                            );
-                        }
-                        fud.file_fetch_end_tx.send((file_hash, Ok(()))).await.unwrap();
-                    }
-                    // Looked for a file but got a chunk: the entire file fits in a single chunk
-                    FetchReply::Chunk(FudChunkReply { chunk }) => {
-                        info!(target: "fud::fetch_file_task()", "File fits in a single chunk");
-                        let chunk_hash = blake3::hash(&chunk);
-                        let _ = fud.geode.insert_file(&file_hash, &[chunk_hash]).await;
-                        match fud.geode.write_chunk(&file_hash, &file_path, &chunk).await {
-                            Ok(_) => {}
-                            Err(e) => {
-                                error!(
-                                    "Failed inserting chunk {} to Geode: {}",
-                                    hash_to_string(&file_hash),
-                                    e
-                                );
-                            }
-                        };
-                        fud.file_fetch_end_tx.send((file_hash, Ok(()))).await.unwrap();
-                    }
+        let (nodes, hash, path) = fud.metadata_fetch_rx.recv().await.unwrap();
+        info!(target: "fud::fetch_metadata_task()", "Fetching metadata for {}", hash_to_string(&hash));
+
+        let reply = fud.fetch_metadata(&nodes, &hash).await;
+        if reply.is_none() {
+            fud.metadata_fetch_end_tx.send(Err(Error::GeodeFileRouteNotFound)).await.unwrap();
+            continue
+        }
+        let reply = reply.unwrap();
+
+        // At this point the reply content was already verified in `fud.fetch_metadata`
+        match reply {
+            FetchReply::Directory(FudDirectoryReply { files, chunk_hashes }) => {
+                // Convert all file paths from String to PathBuf
+                let mut files: Vec<_> = files
+                    .into_iter()
+                    .map(|(path_str, size)| (PathBuf::from(path_str), size))
+                    .collect();
+
+                fud.geode.sort_files(&mut files);
+                if let Err(e) = fud.geode.insert_metadata(&hash, &chunk_hashes, &files).await {
+                    error!(target: "fud::fetch_metadata_task()", "Failed inserting directory {} to Geode: {}", hash_to_string(&hash), e);
+                    fud.metadata_fetch_end_tx.send(Err(e)).await.unwrap();
+                    continue
                 }
+                fud.metadata_fetch_end_tx.send(Ok(())).await.unwrap();
             }
-            None => {
-                fud.file_fetch_end_tx
-                    .send((file_hash, Err(Error::GeodeFileRouteNotFound)))
-                    .await
-                    .unwrap();
+            FetchReply::File(FudFileReply { chunk_hashes }) => {
+                if let Err(e) = fud.geode.insert_metadata(&hash, &chunk_hashes, &[]).await {
+                    error!(target: "fud::fetch_metadata_task()", "Failed inserting file {} to Geode: {}", hash_to_string(&hash), e);
+                    fud.metadata_fetch_end_tx.send(Err(e)).await.unwrap();
+                    continue
+                }
+                fud.metadata_fetch_end_tx.send(Ok(())).await.unwrap();
             }
-        };
+            // Looked for a file but got a chunk: the entire file fits in a single chunk
+            FetchReply::Chunk(FudChunkReply { chunk }) => {
+                info!(target: "fud::fetch_metadata_task()", "File fits in a single chunk");
+                let chunk_hash = blake3::hash(&chunk);
+                let _ = fud.geode.insert_metadata(&hash, &[chunk_hash], &[]).await;
+                let mut chunked_file =
+                    ChunkedStorage::new(&[chunk_hash], &[(path, chunk.len() as u64)], false);
+                if let Err(e) = fud.geode.write_chunk(&mut chunked_file, &chunk).await {
+                    error!(target: "fud::fetch_metadata_task()", "Failed inserting chunk {} to Geode: {}", hash_to_string(&chunk_hash), e);
+                    fud.metadata_fetch_end_tx.send(Err(e)).await.unwrap();
+                    continue
+                };
+                fud.metadata_fetch_end_tx.send(Ok(())).await.unwrap();
+            }
+        }
     }
 }
 

bin/fud/fud/src/util.rs

@@ -0,0 +1,41 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use darkfi::Result;
+use smol::{fs, stream::StreamExt};
+use std::path::{Path, PathBuf};
+
+pub async fn get_all_files(dir: &Path) -> Result<Vec<(PathBuf, u64)>> {
+    let mut files = Vec::new();
+
+    let mut entries = fs::read_dir(dir).await.unwrap();
+
+    while let Some(entry) = entries.try_next().await.unwrap() {
+        let path = entry.path();
+
+        if path.is_dir() {
+            files.append(&mut Box::pin(get_all_files(&path)).await?);
+        } else {
+            let metadata = fs::metadata(&path).await?;
+            let file_size = metadata.len();
+            files.push((path, file_size));
+        }
+    }
+
+    Ok(files)
+}

src/geode/chunked_storage.rs

@@ -0,0 +1,100 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use crate::geode::file_sequence::FileSequence;
+use std::path::PathBuf;
+
+/// `ChunkedStorage` is a representation of a file or directory we're trying to
+/// retrieve from `Geode`.
+#[derive(Debug)]
+pub struct ChunkedStorage {
+    /// Vector of chunk hashes and a bool which is `true` if the chunk is
+    /// available locally.
+    chunks: Vec<(blake3::Hash, bool)>,
+    /// FileSequence containing the list of file paths and file sizes, it has
+    /// a single item if this is not a directory but a single file.
+    fileseq: FileSequence,
+    /// Set to `true` if this ChunkedStorage is the representation of a
+    /// directory.
+    is_dir: bool,
+}
+
+impl ChunkedStorage {
+    pub fn new(hashes: &[blake3::Hash], files: &[(PathBuf, u64)], is_dir: bool) -> Self {
+        Self {
+            chunks: hashes.iter().map(|x| (*x, false)).collect(),
+            fileseq: FileSequence::new(files, is_dir),
+            is_dir,
+        }
+    }
+
+    /// Check whether we have all the chunks available locally.
+    pub fn is_complete(&self) -> bool {
+        !self.chunks.iter().any(|(_, available)| !available)
+    }
+
+    /// Return an iterator over the chunks and their availability.
+    pub fn iter(&self) -> core::slice::Iter<'_, (blake3::Hash, bool)> {
+        self.chunks.iter()
+    }
+
+    /// Return an mutable iterator over the chunks and their availability.
+    pub fn iter_mut(&mut self) -> core::slice::IterMut<'_, (blake3::Hash, bool)> {
+        self.chunks.iter_mut()
+    }
+
+    /// Return the number of chunks.
+    pub fn len(&self) -> usize {
+        self.chunks.len()
+    }
+
+    /// Return `true` if the chunked file contains no chunk.
+    pub fn is_empty(&self) -> bool {
+        self.chunks.is_empty()
+    }
+
+    /// Return the number of chunks available locally.
+    pub fn local_chunks(&self) -> usize {
+        self.chunks.iter().filter(|(_, p)| *p).count()
+    }
+
+    /// Return `chunks`.
+    pub fn get_chunks(&self) -> &Vec<(blake3::Hash, bool)> {
+        &self.chunks
+    }
+
+    /// Return a mutable chunk from `chunks`.
+    pub fn get_chunk_mut(&mut self, index: usize) -> &mut (blake3::Hash, bool) {
+        &mut self.chunks[index]
+    }
+
+    /// Return the list of files from the `reader`.
+    pub fn get_files(&self) -> &Vec<(PathBuf, u64)> {
+        self.fileseq.get_files()
+    }
+
+    /// Return `fileseq`.
+    pub fn get_fileseq(&mut self) -> &mut FileSequence {
+        &mut self.fileseq
+    }
+
+    /// Return `is_dir`.
+    pub fn is_dir(&self) -> bool {
+        self.is_dir
+    }
+}

src/geode/file_sequence.rs

@@ -0,0 +1,291 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use futures::{
+    task::{Context, Poll},
+    AsyncRead, AsyncSeek, AsyncWrite,
+};
+use smol::{
+    fs::{File, OpenOptions},
+    io::{self, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, SeekFrom},
+};
+use std::{path::PathBuf, pin::Pin};
+
+/// `FileSequence` is an object that implements `AsyncRead`, `AsyncSeek`, and
+/// `AsyncWrite` for an ordered list of (file path, file size).
+///
+/// You can use it to read and write from/to a list a file, without having to
+/// manage individual file operations explicitly.
+///
+/// This allows seamless handling of multiple files as if they were a single
+/// continuous file. It automatically opens the next file in the list when the
+/// current file is exhausted.
+#[derive(Debug)]
+pub struct FileSequence {
+    /// List of (file path, file size). File sizes are not the sizes of the
+    /// files as they currently are on the file system, but the sizes we want
+    files: Vec<(PathBuf, u64)>,
+    /// Currently opened file
+    current_file: Option<File>,
+    /// Index of the currently opened file in the `files` vector
+    current_file_index: Option<usize>,
+    /// Set to `true` to automatically set the length of the file on the
+    /// filesystem to it's size as defined in the `files` vector, after a write
+    auto_set_len: bool,
+}
+
+impl FileSequence {
+    pub fn new(files: &[(PathBuf, u64)], auto_set_len: bool) -> Self {
+        Self { files: files.to_vec(), current_file: None, current_file_index: None, auto_set_len }
+    }
+
+    /// Update a single file size.
+    pub fn set_file_size(&mut self, file_index: usize, file_size: u64) {
+        self.files[file_index].1 = file_size;
+    }
+
+    /// Return `current_file`.
+    pub fn get_current_file(&self) -> &Option<File> {
+        &self.current_file
+    }
+
+    /// Return `files`.
+    pub fn get_files(&self) -> &Vec<(PathBuf, u64)> {
+        &self.files
+    }
+
+    /// Open the file at (`current_file_index` + 1).
+    /// If no file is currently open (`current_file_index` is None), it opens
+    /// the first file.
+    async fn open_next_file(&mut self) -> io::Result<()> {
+        self.current_file_index = match self.current_file_index {
+            Some(i) => Some(i + 1),
+            None => Some(0),
+        };
+        if self.current_file_index.unwrap() >= self.files.len() {
+            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "No more files to open"))
+        }
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .open(self.files[self.current_file_index.unwrap()].0.clone())
+            .await?;
+        self.current_file = Some(file);
+        Ok(())
+    }
+
+    /// Open the file at `file_index`.
+    async fn open_file(&mut self, file_index: usize) -> io::Result<()> {
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .open(self.files[file_index].0.clone())
+            .await?;
+        self.current_file = Some(file);
+        self.current_file_index = Some(file_index);
+        Ok(())
+    }
+}
+
+impl AsyncRead for FileSequence {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.get_mut();
+        let mut total_read = 0;
+
+        while total_read < buf.len() {
+            if this.current_file.is_none() {
+                // Stop if there are no more files to read
+                if let Some(file_index) = this.current_file_index {
+                    if file_index >= this.files.len() - 1 {
+                        return Poll::Ready(Ok(total_read));
+                    }
+                }
+                // Open the next file
+                if let Err(e) = smol::block_on(this.open_next_file()) {
+                    return Poll::Ready(Err(e));
+                }
+            }
+
+            // Read from the current file
+            let file = this.current_file.as_mut().unwrap();
+            match smol::block_on(file.read(&mut buf[total_read..])) {
+                Ok(bytes_read) => {
+                    if bytes_read == 0 {
+                        this.current_file = None; // Move to the next file
+                    } else {
+                        total_read += bytes_read;
+                    }
+                }
+                Err(e) => return Poll::Ready(Err(e)),
+            }
+        }
+
+        Poll::Ready(Ok(total_read))
+    }
+}
+
+impl AsyncSeek for FileSequence {
+    fn poll_seek(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<io::Result<u64>> {
+        let this = self.get_mut();
+
+        let abs_pos = match pos {
+            SeekFrom::Start(offset) => offset,
+            _ => todo!(), // TODO
+        };
+
+        // Determine which file to seek in
+        let mut file_index = 0;
+        let mut bytes_offset = 0;
+
+        while file_index < this.files.len() {
+            if bytes_offset + this.files[file_index].1 >= abs_pos {
+                break;
+            }
+            bytes_offset += this.files[file_index].1;
+            file_index += 1;
+        }
+
+        if file_index >= this.files.len() {
+            return Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "Seek position out of bounds",
+            )))
+        }
+
+        // Open the file
+        if this.current_file.is_none() ||
+            this.current_file_index.is_some() && this.current_file_index.unwrap() != file_index
+        {
+            if let Err(e) = smol::block_on(this.open_file(file_index)) {
+                return Poll::Ready(Err(e));
+            }
+        }
+
+        let file = this.current_file.as_mut().unwrap();
+        let file_pos = abs_pos - bytes_offset;
+
+        // Seek in the current file
+        match smol::block_on(file.seek(SeekFrom::Start(file_pos))) {
+            Ok(new_position) => Poll::Ready(Ok(new_position)),
+            Err(e) => Poll::Ready(Err(e)),
+        }
+    }
+}
+
+impl AsyncWrite for FileSequence {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.get_mut();
+        let mut total_bytes_written = 0;
+        let mut remaining_buf = buf;
+        let auto_set_len = this.auto_set_len;
+
+        let finalize_current_file = |file: &mut File, max_size: u64| {
+            if auto_set_len {
+                smol::block_on(file.set_len(max_size))?;
+            }
+            smol::block_on(file.flush())?;
+            Ok(())
+        };
+
+        loop {
+            // Ensure the current file is open
+            if this.current_file.is_none() {
+                if let Some(file_index) = this.current_file_index {
+                    if file_index >= this.files.len() - 1 {
+                        break; // No more files
+                    }
+                }
+                if let Err(e) = smol::block_on(this.open_next_file()) {
+                    return Poll::Ready(Err(e));
+                }
+            }
+
+            let file = this.current_file.as_mut().unwrap();
+            let max_size = this.files[this.current_file_index.unwrap()].1;
+
+            // Check how much space is left in the current file
+            let current_position = smol::block_on(file.seek(io::SeekFrom::Current(0)))?;
+            let space_left = max_size - current_position;
+            let bytes_to_write = remaining_buf.len().min(space_left as usize);
+
+            if bytes_to_write == 0 {
+                // Continue to the next iteration to check the new file
+                if let Err(e) = finalize_current_file(file, max_size) {
+                    return Poll::Ready(Err(e));
+                }
+                this.current_file = None;
+                continue;
+            }
+
+            // Write to the current file
+            match smol::block_on(file.write(&remaining_buf[..bytes_to_write])) {
+                Ok(bytes_written) => {
+                    total_bytes_written += bytes_written;
+                    remaining_buf = &remaining_buf[bytes_written..]; // Update the remaining buffer
+                    if remaining_buf.is_empty() {
+                        if let Err(e) = finalize_current_file(file, max_size) {
+                            return Poll::Ready(Err(e));
+                        }
+                        break; // No more data to write
+                    }
+
+                    // We wrote to the end of this file, use new file on next iteration
+                    if bytes_written == bytes_to_write {
+                        if let Err(e) = finalize_current_file(file, max_size) {
+                            return Poll::Ready(Err(e));
+                        }
+                        this.current_file = None;
+                    }
+                }
+                Err(e) => return Poll::Ready(Err(e)), // Return error if write fails
+            }
+        }
+
+        Poll::Ready(Ok(total_bytes_written))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(())) // TODO
+    }
+
+    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.get_mut();
+        if let Some(file) = this.current_file.take() {
+            match smol::block_on(file.sync_all()) {
+                Ok(()) => Poll::Ready(Ok(())),
+                Err(e) => Poll::Ready(Err(e)),
+            }
+        } else {
+            Poll::Ready(Ok(())) // No file to close
+        }
+    }
+}

src/geode/mod.rs

@@ -19,35 +19,59 @@
 //! Chunk-based file storage implementation.
 //! This is a building block for a DHT or something similar.
 //!
-//! The API supports file insertion and retrieval. There is intentionally no
-//! `remove` support. File removal should be handled externally, and then it
-//! is only required to run `garbage_collect()` to clean things up.
+//! The API supports file/directory insertion and retrieval. There is
+//! intentionally no `remove` support. File removal should be handled
+//! externally, and then it is only required to run `garbage_collect()` to
+//! clean things up.
+//!
+//! The hash of a file is the BLAKE3 hash of hashed chunks in the correct
+//! order.
+//! The hash of a directory is the BLAKE3 hash of hashed chunks in the correct
+//! order and the ordered list of (file path, file sizes).
+//! All hashes (file, directory, chunk) are 32 bytes long, and are encoded in
+//! base58 whenever necessary.
 //!
 //! The filesystem hierarchy stores a `files` directory storing metadata
-//! about a full file. The filename of a file in `files` is the BLAKE3
-//! hash of hashed chunks in the correct order. Inside the file is the list
-//! of the chunks making up the full file.
+//! about a full file and a `directories` directory storing metadata about all
+//! files in a directory (all subdirectories included).
+//! The filename of a file in `files` or `directories` is the hash of the
+//! file/directory as defined above.
+//! Inside a file in `files` is the ordered list of the chunks making up the
+//! full file.
+//! Inside a file in `directories` is the ordered list of the chunks making up
+//! each full file, and the (relative) file path and hash of all files in the
+//! directory.
 //!
 //! To get the chunks you split the full file into `MAX_CHUNK_SIZE` sized
-//! slices, where the last chunk is the only one that can be smaller than
-//! that.
+//! slices, the last chunk is the only one that can be smaller than that.
 //!
 //! It might look like the following:
 //! ```
 //! /files/B9fFKaEYphw2oH5PDbeL1TTAcSzL6ax84p8SjBKzuYzX
 //! /files/8nA3ndjFFee3n5wMPLZampLpGaMJi3od4MSyaXPDoF91
 //! /files/...
+//! /directories/FXDduPcEohVzsSxtNVSFU64qtYxEVEHBMkF4k5cBvt3B
+//! /directories/AHjU1LizfGqsGnF8VSa9kphSQ5pqS4YjmPqme5RZajsj
+//! /directories/...
 //! ```
 //!
-//! In the above example, contents of `B9fFKaEYphw2oH5PDbeL1TTAcSzL6ax84p8SjBKzuYzX`
-//! may be:
+//! Inside a file metadata (file in `files`) is the ordered list of chunk
+//! hashes, for example:
 //! ```
 //! 2bQPxSR8Frz7S7JW3DRAzEtkrHfLXB1CN65V7az77pUp
 //! CvjvN6MfWQYK54DgKNR7MPgFSZqsCgpWKF2p8ot66CCP
 //! ```
 //!
-//! This means, the file `B9fFKaEYphw2oH5PDbeL1TTAcSzL6ax84p8SjBKzuYzX`
-//! is the concatenation of the chunks with the above hashes.
+//! Inside a directory metadata (file in `directories`) is, in addition to
+//! chunk hashes, the path and size of each file in the directory. For example:
+//! ```
+//! 8Kb55jeqJsq7WTBN93gvBzh2zmXAXVPh111VqD3Hi42V
+//! GLiBqpLPTbpJhSMYfzi3s7WivrTViov7ShX7uso6fG5s
+//! picture.jpg 312948
+//! ```
+//! Chunks of a directory can include multiple files, if multiple files fit
+//! into `MAX_CHUNK_SIZE`. The chunks are computed as if all the files were
+//! concatenated into a single big file, to minimize the number of chunks.
 //!
 //! The full file is not copied, and individual chunks are not stored by
 //! geode. Additionally it does not keep track of the full files path.
@@ -57,91 +81,38 @@ use std::{collections::HashSet, path::PathBuf};
 use futures::{AsyncRead, AsyncSeek};
 use log::{debug, info, warn};
 use smol::{
-    fs::{self, File, OpenOptions},
-    io::{
-        self, AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, Cursor,
-        SeekFrom,
-    },
+    fs::{self, File},
+    io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, Cursor, SeekFrom},
     stream::StreamExt,
 };
+use std::path::Path;
 
 use crate::{Error, Result};
 
+mod chunked_storage;
+pub use chunked_storage::ChunkedStorage;
+
+mod file_sequence;
+pub use file_sequence::FileSequence;
+
+mod util;
+pub use util::{hash_to_string, read_until_filled};
+
 /// Defined maximum size of a stored chunk (256 KiB)
 pub const MAX_CHUNK_SIZE: usize = 262_144;
 
 /// Path prefix where file metadata is stored
 const FILES_PATH: &str = "files";
 
-pub fn hash_to_string(hash: &blake3::Hash) -> String {
-    bs58::encode(hash.as_bytes()).into_string()
-}
-
-/// `ChunkedFile` is a representation of a file we're trying to
-/// retrieve from `Geode`.
-///
-/// The tuple contains `blake3::Hash` of
-/// the file's chunks and an optional `PathBuf` which points to
-/// the filesystem where the chunk can be found. If `None`, it
-/// is to be assumed that the chunk is not available locally.
-#[derive(Clone)]
-pub struct ChunkedFile(Vec<(blake3::Hash, Option<bool>)>);
-
-impl ChunkedFile {
-    fn new(hashes: &[blake3::Hash]) -> Self {
-        Self(hashes.iter().map(|x| (*x, None)).collect())
-    }
-
-    /// Check whether we have all the chunks available locally.
-    pub fn is_complete(&self) -> bool {
-        !self.0.iter().any(|(_, p)| p.is_none())
-    }
-
-    /// Return an iterator over the chunks and their paths.
-    pub fn iter(&self) -> core::slice::Iter<'_, (blake3::Hash, Option<bool>)> {
-        self.0.iter()
-    }
-
-    /// Return the number of chunks.
-    pub fn len(&self) -> usize {
-        self.0.len()
-    }
-
-    /// Return `true` if the chunked file contains no chunk.
-    pub fn is_empty(&self) -> bool {
-        self.0.is_empty()
-    }
-
-    /// Return the number of chunks available locally.
-    pub fn local_chunks(&self) -> usize {
-        self.0.iter().filter(|(_, p)| p.is_some()).count()
-    }
-}
+/// Path prefix where directory metadata is stored
+const DIRS_PATH: &str = "directories";
 
 /// Chunk-based file storage interface.
 pub struct Geode {
     /// Path to the filesystem directory where file metadata is stored
-    files_path: PathBuf,
-}
-
-/// smol::fs::File::read does not guarantee that the buffer will be filled, even if the buffer is
-/// smaller than the file. This is a workaround.
-/// This reads the stream until the buffer is full or until we reached the end of the stream.
-pub async fn read_until_filled(
-    mut stream: impl AsyncRead + Unpin,
-    buffer: &mut [u8],
-) -> io::Result<usize> {
-    let mut total_bytes_read = 0;
-
-    while total_bytes_read < buffer.len() {
-        let bytes_read = stream.read(&mut buffer[total_bytes_read..]).await?;
-        if bytes_read == 0 {
-            break; // EOF reached
-        }
-        total_bytes_read += bytes_read;
-    }
-
-    Ok(total_bytes_read)
+    pub files_path: PathBuf,
+    /// Path to the filesystem directory where directory metadata is stored
+    pub dirs_path: PathBuf,
 }
 
 impl Geode {
@@ -151,29 +122,73 @@ impl Geode {
     pub async fn new(base_path: &PathBuf) -> Result<Self> {
         let mut files_path: PathBuf = base_path.into();
         files_path.push(FILES_PATH);
+        let mut dirs_path: PathBuf = base_path.into();
+        dirs_path.push(DIRS_PATH);
 
         // Create necessary directory structure if needed
         fs::create_dir_all(&files_path).await?;
+        fs::create_dir_all(&dirs_path).await?;
 
-        Ok(Self { files_path })
+        Ok(Self { files_path, dirs_path })
     }
 
-    /// Attempt to read chunk hashes from a given file path and return
-    /// a `Vec` containing the hashes in order.
-    async fn read_metadata(path: &PathBuf) -> Result<Vec<blake3::Hash>> {
-        debug!(target: "geode::read_metadata()", "Reading chunks from {:?}", path);
+    /// Attempt to read chunk hashes and files metadata from a given metadata path.
+    /// This works for both file metadata and directory metadata.
+    /// Returns (chunk hashes, [(file path, file size)]).
+    async fn read_metadata(path: &PathBuf) -> Result<(Vec<blake3::Hash>, Vec<(PathBuf, u64)>)> {
+        debug!(target: "geode::read_dir_metadata()", "Reading chunks from {:?} (dir)", path);
+
+        let mut chunk_hashes = vec![];
+        let mut files = vec![];
+
         let fd = File::open(path).await?;
-        let mut read_chunks = vec![];
         let mut lines = BufReader::new(fd).lines();
+
         while let Some(line) = lines.next().await {
             let line = line?;
-            let mut hash_buf = [0u8; 32];
-            bs58::decode(line).onto(&mut hash_buf)?;
-            let chunk_hash = blake3::Hash::from_bytes(hash_buf);
-            read_chunks.push(chunk_hash);
+            let line = line.trim();
+
+            if line.is_empty() {
+                continue; // Skip empty lines
+            }
+
+            let parts: Vec<&str> = line.split_whitespace().collect();
+            if parts.len() == 2 {
+                // File
+                let file_path = PathBuf::from(parts[0]);
+                if file_path.clone().is_absolute() {
+                    return Err(Error::Custom(format!(
+                        "Path of file {} is absolute, which is not allowed",
+                        parts[0]
+                    )))
+                }
+
+                // Check for `..` in the path components
+                for component in file_path.clone().components() {
+                    if component == std::path::Component::ParentDir {
+                        return Err(Error::Custom(format!("Path of file {} contains reference to parent dir, which is not allowed", parts[0])))
+                    }
+                }
+
+                let file_size = parts[1].parse::<u64>()?;
+                files.push((file_path, file_size));
+            } else if parts.len() == 1 {
+                // Chunk
+                let chunk_hash_str = parts[0].trim();
+                if chunk_hash_str.is_empty() {
+                    break; // Stop reading chunk hashes on empty line
+                }
+                let mut hash_buf = [0u8; 32];
+                bs58::decode(chunk_hash_str).onto(&mut hash_buf)?;
+                let chunk_hash = blake3::Hash::from_bytes(hash_buf);
+                chunk_hashes.push(chunk_hash);
+            } else {
+                // Invalid format
+                return Err(Error::Custom("Invalid directory metadata format".to_string()));
+            }
         }
 
-        Ok(read_chunks)
+        Ok((chunk_hashes, files))
     }
 
     /// Perform garbage collection over the filesystem hierarchy.
@@ -183,10 +198,12 @@ impl Geode {
         // We track corrupt files here.
         let mut deleted_files = HashSet::new();
 
-        // Perform health check over file metadata. For now we just ensure they
+        // Perform health check over metadata. For now we just ensure they
         // have the correct format.
-        let mut file_paths = fs::read_dir(&self.files_path).await?;
-        while let Some(file) = file_paths.next().await {
+        let file_paths = fs::read_dir(&self.files_path).await?;
+        let dir_paths = fs::read_dir(&self.dirs_path).await?;
+        let mut paths = file_paths.chain(dir_paths);
+        while let Some(file) = paths.next().await {
             let Ok(entry) = file else { continue };
             let path = entry.path();
 
@@ -201,7 +218,7 @@ impl Geode {
                 None => continue,
             };
             let mut hash_buf = [0u8; 32];
-            let file_hash = match bs58::decode(file_name).onto(&mut hash_buf) {
+            let hash = match bs58::decode(file_name).onto(&mut hash_buf) {
                 Ok(_) => blake3::Hash::from_bytes(hash_buf),
                 Err(_) => continue,
             };
@@ -213,11 +230,11 @@ impl Geode {
                 if let Err(e) = fs::remove_file(path).await {
                     warn!(
                        target: "geode::garbage_collect()",
-                       "[Geode] Garbage collect failed to remove corrupted file: {}", e,
+                       "[Geode] Garbage collect failed to remove corrupted metadata: {}", e,
                     );
                 }
 
-                deleted_files.insert(file_hash);
+                deleted_files.insert(hash);
                 continue
             }
         }
@@ -226,81 +243,97 @@ impl Geode {
         Ok(deleted_files)
     }
 
-    /// Insert a file into Geode. The function expects any kind of byte stream, which
-    /// can either be another file on the filesystem, a buffer, etc.
-    /// Returns a tuple of `(blake3::Hash, Vec<blake3::Hash>)` which represents the
-    /// file hash, and the file's chunks, respectively.
-    pub async fn insert(
+    /// Chunk a stream.
+    /// Returns a hasher (containing the chunk hashes), and the list of chunk hashes.
+    pub async fn chunk_stream(
         &self,
         mut stream: impl AsyncRead + Unpin,
-    ) -> Result<(blake3::Hash, Vec<blake3::Hash>)> {
-        info!(target: "geode::insert()", "[Geode] Inserting file...");
-        let mut file_hasher = blake3::Hasher::new();
+    ) -> Result<(blake3::Hasher, Vec<blake3::Hash>)> {
+        let mut hasher = blake3::Hasher::new();
         let mut chunk_hashes = vec![];
 
         loop {
             let mut buf = vec![0u8; MAX_CHUNK_SIZE];
-            let bytes_read = read_until_filled(&mut stream, &mut buf).await?;
+            let bytes_read = stream.read(&mut buf).await?;
             if bytes_read == 0 {
                 break
             }
 
-            let chunk_slice = &buf[..bytes_read];
-            let chunk_hash = blake3::hash(chunk_slice);
-            file_hasher.update(chunk_hash.as_bytes());
+            let chunk_hash = blake3::hash(&buf[..bytes_read]);
+            hasher.update(chunk_hash.as_bytes());
             chunk_hashes.push(chunk_hash);
         }
 
-        // This hash is the file's chunks hashes hashed in order.
-        let file_hash = file_hasher.finalize();
-        let mut file_path = self.files_path.clone();
-        file_path.push(hash_to_string(&file_hash).as_str());
+        Ok((hasher, chunk_hashes))
+    }
 
-        // We always overwrite the metadata.
-        let mut file_fd = File::create(&file_path).await?;
-        for ch in &chunk_hashes {
-            file_fd.write(format!("{}\n", hash_to_string(ch).as_str()).as_bytes()).await?;
-        }
+    /// Sorts files by their PathBuf.
+    pub fn sort_files(&self, files: &mut [(PathBuf, u64)]) {
+        files.sort_by(|(a, _), (b, _)| a.to_string_lossy().cmp(&b.to_string_lossy()));
+    }
 
-        file_fd.flush().await?;
+    /// Add chunk hashes to `hasher`.
+    pub fn hash_chunks_metadata(&self, hasher: &mut blake3::Hasher, chunk_hashes: &[blake3::Hash]) {
+        for chunk in chunk_hashes {
+            hasher.update(chunk.as_bytes());
+        }
+    }
 
-        Ok((file_hash, chunk_hashes))
+    /// Add files metadata to `hasher`.
+    /// You must sort the files using `sort_files`.
+    pub fn hash_files_metadata(
+        &self,
+        hasher: &mut blake3::Hasher,
+        relative_files: &[(PathBuf, u64)],
+    ) {
+        for file in relative_files {
+            hasher.update(file.0.to_string_lossy().to_string().as_bytes());
+            hasher.update(&file.1.to_le_bytes());
+        }
     }
 
-    /// Create and insert file metadata into Geode given a list of hashes.
+    /// Create and insert file or directory metadata into Geode.
     /// Always overwrites any existing file.
-    /// Verifies that the file hash matches the chunk hashes
-    pub async fn insert_file(
+    /// Verifies that the metadata is valid.
+    /// The `relative_files` slice is empty for files.
+    pub async fn insert_metadata(
         &self,
-        file_hash: &blake3::Hash,
+        hash: &blake3::Hash,
         chunk_hashes: &[blake3::Hash],
+        relative_files: &[(PathBuf, u64)],
     ) -> Result<()> {
-        info!(target: "geode::insert_file()", "[Geode] Inserting file metadata");
+        info!(target: "geode::insert_metadata()", "[Geode] Inserting directory metadata");
 
-        if !self.verify_file(file_hash, chunk_hashes) {
-            // The chunk list or file hash is wrong
+        // Verify the metadata
+        if !self.verify_metadata(hash, chunk_hashes, relative_files) {
             return Err(Error::GeodeNeedsGc)
         }
 
-        let mut file_path = self.files_path.clone();
-        file_path.push(hash_to_string(file_hash).as_str());
+        // Write the metadata file
+        let mut file_path = match relative_files.is_empty() {
+            true => self.files_path.clone(),
+            false => self.dirs_path.clone(),
+        };
+        file_path.push(hash_to_string(hash).as_str());
         let mut file_fd = File::create(&file_path).await?;
 
         for ch in chunk_hashes {
             file_fd.write(format!("{}\n", hash_to_string(ch).as_str()).as_bytes()).await?;
         }
+        for file in relative_files {
+            file_fd.write(format!("{} {}\n", file.0.to_string_lossy(), file.1).as_bytes()).await?;
+        }
         file_fd.flush().await?;
 
         Ok(())
     }
 
-    /// Write a single chunk into `file_path` given a stream.
+    /// Write a single chunk given a stream.
     /// The file must be inserted into Geode before calling this method.
     /// Always overwrites any existing chunk. Returns the chunk hash once inserted.
     pub async fn write_chunk(
         &self,
-        file_hash: &blake3::Hash,
-        file_path: &PathBuf,
+        chunked: &mut ChunkedStorage,
         stream: impl AsRef<[u8]>,
     ) -> Result<blake3::Hash> {
         info!(target: "geode::write_chunk()", "[Geode] Writing single chunk");
@@ -308,144 +341,148 @@ impl Geode {
         let mut cursor = Cursor::new(&stream);
         let mut chunk = vec![0u8; MAX_CHUNK_SIZE];
 
-        let bytes_read = read_until_filled(&mut cursor, &mut chunk).await?;
-        let chunk_slice = &chunk[..bytes_read];
-        let chunk_hash = blake3::hash(chunk_slice);
-
-        let file_hash_str = hash_to_string(file_hash);
-        let mut file_metadata_path = self.files_path.clone();
-        file_metadata_path.push(file_hash_str);
+        // Read the stream to get the chunk content
+        let chunk_slice = read_until_filled(&mut cursor, &mut chunk).await?;
 
-        // Try to read the file metadata. If it's corrupt, return an error signalling
-        // that garbage collection needs to run.
-        let chunk_hashes = match Self::read_metadata(&file_metadata_path).await {
-            Ok(v) => v,
-            Err(e) => {
-                return match e {
-                    // If the file is not found, return according error.
-                    Error::Io(std::io::ErrorKind::NotFound) => Err(Error::GeodeFileNotFound),
-                    // Anything else should tell the client to do garbage collection
-                    _ => Err(Error::GeodeNeedsGc),
-                }
-            }
-        };
+        // Get the chunk hash from the content
+        let chunk_hash = blake3::hash(chunk_slice);
 
-        // Get the chunk index in the file from the chunk hash
-        let chunk_index = match chunk_hashes.iter().position(|h| *h == chunk_hash) {
+        // Get the chunk index in the file/directory from the chunk hash
+        let chunk_index = match chunked.iter().position(|(h, _)| *h == chunk_hash) {
             Some(index) => index,
             None => {
                 return Err(Error::GeodeNeedsGc);
             }
         };
 
+        // Compute byte position from the chunk index and the chunk size
         let position = (chunk_index as u64) * (MAX_CHUNK_SIZE as u64);
 
-        // Create the file if it does not exist
-        if !file_path.exists() {
-            File::create(&file_path).await?;
+        // Seek to the correct position
+        let fileseq = &mut chunked.get_fileseq();
+        fileseq.seek(SeekFrom::Start(position)).await?;
+
+        // This will write the chunk, and truncate files if `chunked` is a directory.
+        fileseq.write_all(chunk_slice).await?;
+
+        // If it's the last chunk of a file (and it's *not* a directory),
+        // truncate the file to the correct length.
+        // This is because contrary to directories, for a file shared on fud we
+        // do not know the exact file size from its metadata, we only know the
+        // number of chunks. Therefore we only know the exact size once we know
+        // the size of the last chunk.
+        if !chunked.is_dir() && chunk_index == chunked.len() - 1 {
+            let exact_file_size =
+                chunked.len() * MAX_CHUNK_SIZE - (MAX_CHUNK_SIZE - chunk_slice.len());
+            if let Some(file) = &chunked.get_fileseq().get_current_file() {
+                let _ = file.set_len(exact_file_size as u64).await;
+            }
         }
 
-        let mut file_fd = OpenOptions::new().write(true).open(&file_path).await?;
-        file_fd.seek(SeekFrom::Start(position)).await?;
-        file_fd.write_all(chunk_slice).await?;
-        file_fd.flush().await?;
-
         Ok(chunk_hash)
     }
 
-    /// Fetch file metadata from Geode. Returns [`ChunkedFile`] which gives a list
-    /// of chunks and booleans to know if the chunks we have are valid. Returns an error if
-    /// the read failed in any way (could also be the file does not exist).
-    pub async fn get(&self, file_hash: &blake3::Hash, file_path: &PathBuf) -> Result<ChunkedFile> {
-        let file_hash_str = hash_to_string(file_hash);
-        info!(target: "geode::get()", "[Geode] Getting file chunks for {}...", file_hash_str);
-        let mut file_metadata_path = self.files_path.clone();
-        file_metadata_path.push(file_hash_str);
+    /// Iterate over chunks and find which chunks are available locally.
+    pub async fn verify_chunks(&self, chunked_file: &mut ChunkedStorage) -> Result<()> {
+        let chunks = chunked_file.get_chunks().clone();
+        let mut available_chunks = vec![];
 
-        // Try to read the file metadata. If it's corrupt, return an error signalling
-        // that garbage collection needs to run.
-        let chunk_hashes = match Self::read_metadata(&file_metadata_path).await {
-            Ok(v) => v,
-            Err(e) => {
-                return match e {
-                    // If the file is not found, return according error.
-                    Error::Io(std::io::ErrorKind::NotFound) => Err(Error::GeodeFileNotFound),
-                    // Anything else should tell the client to do garbage collection
-                    _ => Err(Error::GeodeNeedsGc),
-                }
+        // Gather all available chunks
+        for (chunk_index, (chunk_hash, _)) in chunks.iter().enumerate() {
+            // Read the chunk using the mutable reader
+            let chunk = self.read_chunk(&mut chunked_file.get_fileseq(), &chunk_index).await?;
+
+            // Perform chunk consistency check
+            if self.verify_chunk(chunk_hash, &chunk) {
+                available_chunks.push(chunk_index);
             }
-        };
+        }
 
-        // Make sure the chunk hashes match with the file hash
-        if !self.verify_file(file_hash, &chunk_hashes) {
-            return Err(Error::GeodeNeedsGc);
+        // Update available chunks
+        for chunk_index in available_chunks {
+            chunked_file.get_chunk_mut(chunk_index).1 = true;
         }
 
-        let mut chunked_file = ChunkedFile::new(&chunk_hashes);
+        Ok(())
+    }
 
-        // Open the file, if we can't we return the chunked file with no locally available chunk.
-        let mut file = match File::open(&file_path).await {
-            Ok(v) => v,
-            Err(_) => {
-                return Ok(chunked_file);
-            }
-        };
+    /// Fetch file/directory metadata from Geode. Returns [`Chunked`]. Returns an error if
+    /// the read failed in any way (could also be the file does not exist).
+    pub async fn get(&self, hash: &blake3::Hash, path: &Path) -> Result<ChunkedStorage> {
+        let hash_str = hash_to_string(hash);
+        info!(target: "geode::get()", "[Geode] Getting chunks for {}...", hash_str);
 
-        // Iterate over chunks and find which chunks we have available locally.
-        for (chunk_index, (chunk_hash, chunk_valid)) in chunked_file.0.iter_mut().enumerate() {
-            let chunk = self.read_chunk(&mut file, &chunk_index).await?;
+        // Try to read the file or dir metadata. If it's corrupt, return an error signalling
+        // that garbage collection needs to run.
+        let metadata_paths = [self.files_path.join(&hash_str), self.dirs_path.join(&hash_str)];
+        for metadata_path in metadata_paths {
+            match Self::read_metadata(&metadata_path).await {
+                Ok((chunk_hashes, files)) => {
+                    return self.create_chunked_storage(hash, path, &chunk_hashes, &files).await
+                }
+                Err(e) => {
+                    if !matches!(e, Error::Io(std::io::ErrorKind::NotFound)) {
+                        return Err(Error::GeodeNeedsGc)
+                    }
+                }
+            };
+        }
 
-            // Perform chunk consistency check
-            if !self.verify_chunk(chunk_hash, &chunk) {
-                continue
-            }
+        Err(Error::GeodeFileNotFound)
+    }
 
-            *chunk_valid = Some(true);
+    /// Create a ChunkedStorage from metadata.
+    /// `hash` is the hash of the file or directory.
+    async fn create_chunked_storage(
+        &self,
+        hash: &blake3::Hash,
+        path: &Path,
+        chunk_hashes: &[blake3::Hash],
+        relative_files: &[(PathBuf, u64)], // Only used by directories
+    ) -> Result<ChunkedStorage> {
+        // Make sure the file or directory is valid
+        if !self.verify_metadata(hash, chunk_hashes, relative_files) {
+            return Err(Error::GeodeNeedsGc);
         }
 
-        Ok(chunked_file)
+        let chunked = if relative_files.is_empty() {
+            // File
+            let file_size = (chunk_hashes.len() * MAX_CHUNK_SIZE) as u64; // Upper bound, not actual file size
+            ChunkedStorage::new(chunk_hashes, &[(path.to_path_buf(), file_size)], false)
+        } else {
+            // Directory
+            let files: Vec<_> = relative_files
+                .iter()
+                .map(|(file_path, size)| (path.join(file_path), *size))
+                .collect();
+            ChunkedStorage::new(chunk_hashes, &files, true)
+        };
+
+        Ok(chunked)
     }
 
     /// Fetch a single chunk from Geode. Returns a Vec containing the chunk content
     /// if it is found.
     pub async fn get_chunk(
         &self,
+        chunked: &mut ChunkedStorage,
         chunk_hash: &blake3::Hash,
-        file_hash: &blake3::Hash,
-        file_path: &PathBuf,
+        path: &Path,
     ) -> Result<Vec<u8>> {
         info!(target: "geode::get_chunk()", "[Geode] Getting chunk {}", hash_to_string(chunk_hash));
 
-        if !file_path.exists() || !file_path.is_file() {
+        if !path.exists() {
             return Err(Error::GeodeChunkNotFound)
         }
 
-        let mut file_metadata_path = self.files_path.clone();
-        file_metadata_path.push(hash_to_string(file_hash));
-
-        // Try to read the file metadata. If it's corrupt, return an error signalling
-        // that garbage collection needs to run.
-        let chunk_hashes = match Self::read_metadata(&file_metadata_path).await {
-            Ok(v) => v,
-            Err(e) => {
-                return match e {
-                    // If the file is not found, return according error.
-                    Error::Io(std::io::ErrorKind::NotFound) => Err(Error::GeodeFileNotFound),
-                    // Anything else should tell the client to do garbage collection
-                    _ => Err(Error::GeodeNeedsGc),
-                }
-            }
-        };
-
         // Get the chunk index in the file from the chunk hash
-        let chunk_index = match chunk_hashes.iter().position(|&h| h == *chunk_hash) {
+        let chunk_index = match chunked.iter().position(|(h, _)| *h == *chunk_hash) {
             Some(index) => index,
             None => return Err(Error::GeodeChunkNotFound),
         };
 
         // Read the file to get the chunk content
-        let mut file = File::open(&file_path).await?;
-        let chunk = self.read_chunk(&mut file, &chunk_index).await?;
+        let chunk = self.read_chunk(&mut chunked.get_fileseq(), &chunk_index).await?;
 
         // Perform chunk consistency check
         if !self.verify_chunk(chunk_hash, &chunk) {
@@ -457,7 +494,7 @@ impl Geode {
 
     /// Read the file at `file_path` to get its chunk with index `chunk_index`.
     /// Returns the chunk content in a Vec.
-    pub async fn read_chunk(
+    async fn read_chunk(
         &self,
         mut stream: impl AsyncRead + Unpin + AsyncSeek,
         chunk_index: &usize,
@@ -465,24 +502,27 @@ impl Geode {
         let position = (*chunk_index as u64) * (MAX_CHUNK_SIZE as u64);
         let mut buf = vec![0u8; MAX_CHUNK_SIZE];
         stream.seek(SeekFrom::Start(position)).await?;
-        let bytes_read = read_until_filled(stream, &mut buf).await?;
+        let bytes_read = stream.read(&mut buf).await?;
         Ok(buf[..bytes_read].to_vec())
     }
 
     /// Verifies that the file hash matches the chunk hashes.
-    pub fn verify_file(&self, file_hash: &blake3::Hash, chunk_hashes: &[blake3::Hash]) -> bool {
-        info!(target: "geode::verify_file()", "[Geode] Verifying file metadata for {}", hash_to_string(file_hash));
-
-        let mut file_hasher = blake3::Hasher::new();
-        for chunk_hash in chunk_hashes {
-            file_hasher.update(chunk_hash.as_bytes());
-        }
-
-        *file_hash == file_hasher.finalize()
+    pub fn verify_metadata(
+        &self,
+        hash: &blake3::Hash,
+        chunk_hashes: &[blake3::Hash],
+        files: &[(PathBuf, u64)],
+    ) -> bool {
+        info!(target: "geode::verify_metadata()", "[Geode] Verifying metadata for {}", hash_to_string(hash));
+        let mut hasher = blake3::Hasher::new();
+        self.hash_chunks_metadata(&mut hasher, chunk_hashes);
+        self.hash_files_metadata(&mut hasher, files);
+        *hash == hasher.finalize()
     }
 
     /// Verifies that the chunk hash matches the content.
     pub fn verify_chunk(&self, chunk_hash: &blake3::Hash, chunk_slice: &[u8]) -> bool {
+        info!(target: "geode::verify_chunk()", "[Geode] Verifying chunk {}", hash_to_string(chunk_hash));
         blake3::hash(chunk_slice) == *chunk_hash
     }
 }

src/geode/util.rs

@@ -0,0 +1,44 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use futures::AsyncRead;
+use smol::io::{self, AsyncReadExt};
+
+pub fn hash_to_string(hash: &blake3::Hash) -> String {
+    bs58::encode(hash.as_bytes()).into_string()
+}
+
+/// smol::fs::File::read does not guarantee that the buffer will be filled, even if the buffer is
+/// smaller than the file. This is a workaround.
+/// This reads the stream until the buffer is full or until we reached the end of the stream.
+pub async fn read_until_filled(
+    mut stream: impl AsyncRead + Unpin,
+    buffer: &mut [u8],
+) -> io::Result<&[u8]> {
+    let mut total_bytes_read = 0;
+
+    while total_bytes_read < buffer.len() {
+        let bytes_read = stream.read(&mut buffer[total_bytes_read..]).await?;
+        if bytes_read == 0 {
+            break; // EOF reached
+        }
+        total_bytes_read += bytes_read;
+    }
+
+    Ok(&buffer[..total_bytes_read])
+}