feat(storage): implement storage repository to write file during FerrisShare protocol (dev-sys-do#6)

jorisvilardell · web-flow · commit 42dbf558bee1 · 2025-10-30T15:03:59.000+01:00
* feat(storage): add StorageError enum with Display implementation and update StorageRepository trait

* feat(storage): implement FSStorageRepository with file handling and sanitization

* feat(command): update CommandService trait to include process_binary_data method

* feat(network): make command_service public in NetworkServiceImpl

* feat(application): implement FerrisShareState struct with command and network services

* refactor(network): remove unused HashSet import from entities.rs

* feat(main): refactor main function to initialize services with storage repository

* feat(storage): add Display implementation for StorageError enum

* feat(command): implement From trait for CommandError to String conversion

* feat(command): update execute_protocol_command to return OkHousten and enhance process_binary_data with error handling

* refactor(network): handle process write binary data and closing connection on sender close it

* feat(network): enhance TransferState to track current file during data transfer
diff --git a/src/application/ferrisshare_state.rs b/src/application/ferrisshare_state.rs
@@ -0,0 +1,24 @@
+use crate::core::domain::{command::ports::CommandService, network::ports::NetworkService};
+
+#[derive(Clone, Copy)]
+pub struct FerrisShareState<C, N>
+where
+    C: CommandService,
+    N: NetworkService,
+{
+    pub command_service: C,
+    pub network_service: N,
+}
+
+impl<C, N> FerrisShareState<C, N>
+where
+    C: CommandService + Clone + Send + Sync + 'static,
+    N: NetworkService + Clone + Send + Sync + 'static,
+{
+    pub fn new(command_service: C, network_service: N) -> Self {
+        FerrisShareState {
+            command_service,
+            network_service,
+        }
+    }
+}
diff --git a/src/application/mod.rs b/src/application/mod.rs
@@ -0,0 +1 @@
+pub mod ferrisshare_state;
diff --git a/src/core/domain/command/entities.rs b/src/core/domain/command/entities.rs
@@ -3,3 +3,12 @@ pub enum CommandError {
     InvalidCommand,
     ExecutionFailed(String),
 }
+
+impl From<CommandError> for String {
+    fn from(err: CommandError) -> Self {
+        match err {
+            CommandError::InvalidCommand => "Invalid command".to_string(),
+            CommandError::ExecutionFailed(msg) => format!("Command execution failed: {}", msg),
+        }
+    }
+}
diff --git a/src/core/domain/command/ports.rs b/src/core/domain/command/ports.rs
@@ -10,5 +10,10 @@ pub trait CommandService: Send + Sync {
         &self,
         state: Arc<tokio::sync::Mutex<TransferState>>,
         msg: &ProtocolMessage,
-    ) -> impl Future<Output = Result<ProtocolMessage, CommandError>> + Send;
+    ) -> impl Future<Output = Result<ProtocolMessage, CommandError>> + Send + Sync;
+    fn process_binary_data(
+        &self,
+        state: Arc<tokio::sync::Mutex<TransferState>>,
+        data: &[u8],
+    ) -> impl Future<Output = Result<(), CommandError>>;
 }
diff --git a/src/core/domain/command/services.rs b/src/core/domain/command/services.rs
@@ -3,19 +3,30 @@ use std::sync::Arc;
 use crate::core::domain::{
     command::{entities::CommandError, ports::CommandService},
     network::entities::{ProtocolMessage, TransferState},
-    storage::entities::YeetBlock,
+    storage::ports::StorageRepository,
 };
 
 #[derive(Clone)]
-pub struct CommandServiceImpl {}
+pub struct CommandServiceImpl<C>
+where
+    C: StorageRepository,
+{
+    storage: C,
+}
 
-impl CommandServiceImpl {
-    pub fn new() -> Self {
-        CommandServiceImpl {}
+impl<C> CommandServiceImpl<C>
+where
+    C: StorageRepository + Clone + Send + Sync + 'static,
+{
+    pub fn new(storage: C) -> Self {
+        CommandServiceImpl { storage }
     }
 }
 
-impl CommandService for CommandServiceImpl {
+impl<C> CommandService for CommandServiceImpl<C>
+where
+    C: StorageRepository + Clone + Send + Sync + 'static,
+{
     async fn execute_protocol_command(
         &self,
         state: Arc<tokio::sync::Mutex<TransferState>>,
@@ -34,6 +45,7 @@ impl CommandService for CommandServiceImpl {
                     expected_blocks
                 );
                 *state_guard = TransferState::Receiving {
+                    current_file: _filename.clone(),
                     expected_blocks,
                     focused_block: None,
                     received_blocks: Vec::with_capacity(expected_blocks as usize),
@@ -45,18 +57,25 @@ impl CommandService for CommandServiceImpl {
             }
             ProtocolMessage::Yeet(yeet_block) => {
                 let mut state_guard = state.lock().await;
-                let (expected_blocks, focused_block, received_blocks) = match &mut *state_guard {
-                    TransferState::Receiving {
-                        expected_blocks,
-                        focused_block,
-                        received_blocks,
-                    } => (expected_blocks, focused_block, received_blocks),
-                    _ => {
-                        return Err(CommandError::ExecutionFailed(
-                            "Error transfer state is not equal Receiving".to_string(),
-                        ));
-                    }
-                };
+                let (expected_blocks, focused_block, received_blocks, current_file) =
+                    match &mut *state_guard {
+                        TransferState::Receiving {
+                            expected_blocks,
+                            focused_block,
+                            received_blocks,
+                            current_file,
+                        } => (
+                            expected_blocks,
+                            focused_block,
+                            received_blocks,
+                            current_file,
+                        ),
+                        _ => {
+                            return Err(CommandError::ExecutionFailed(
+                                "Error transfer state is not equal Receiving".to_string(),
+                            ));
+                        }
+                    };
 
                 // Ensure we don't exceed the expected number of blocks.
                 if received_blocks.len() >= *expected_blocks as usize {
@@ -77,17 +96,30 @@ impl CommandService for CommandServiceImpl {
 
                 // Reuse the mutable guard to update the state without locking again.
                 *state_guard = TransferState::Receiving {
+                    current_file: current_file.clone(),
                     expected_blocks: *expected_blocks,
                     focused_block: Some(yeet_block.clone()),
                     received_blocks: received_blocks.clone(),
                 };
 
                 drop(state_guard);
 
-                Ok(ProtocolMessage::Ok)
+                Ok(ProtocolMessage::OkHousten(yeet_block.index))
             }
             ProtocolMessage::MissionAccomplished => {
                 let mut state_guard = state.lock().await;
+                let current_file = match &*state_guard {
+                    TransferState::Receiving { current_file, .. } => current_file,
+                    _ => {
+                        return Err(CommandError::ExecutionFailed(
+                            "Error transfer state is not equal Receiving".to_string(),
+                        ));
+                    }
+                };
+
+                self.storage.finalize(current_file).await.map_err(|e| {
+                    CommandError::ExecutionFailed(format!("Storage error: {:?}", e))
+                })?;
                 *state_guard = TransferState::Finished;
                 drop(state_guard);
                 Ok(ProtocolMessage::Success)
@@ -99,4 +131,56 @@ impl CommandService for CommandServiceImpl {
             _ => Err(CommandError::InvalidCommand),
         }
     }
+
+    async fn process_binary_data(
+        &self,
+        state: Arc<tokio::sync::Mutex<TransferState>>,
+        data: &[u8],
+    ) -> Result<(), CommandError> {
+        let mut state_guard = state.lock().await;
+        let (expected_blocks, focused_block, received_blocks, current_file) =
+            match &mut *state_guard {
+                TransferState::Receiving {
+                    expected_blocks,
+                    focused_block,
+                    received_blocks,
+                    current_file,
+                } => (
+                    expected_blocks,
+                    focused_block,
+                    received_blocks,
+                    current_file,
+                ),
+                _ => {
+                    return Err(CommandError::ExecutionFailed(
+                        "Error transfer state is not equal Receiving".to_string(),
+                    ));
+                }
+            };
+        if let Some(focused_block) = focused_block {
+            if received_blocks.contains(&focused_block.index) {
+                println!("Block {} already received, ignoring.", focused_block.index);
+            } else {
+                println!("Stored binary data block: {:?}", focused_block);
+                self.storage
+                    .write_block(current_file, focused_block, data)
+                    .await
+                    .map_err(|e| {
+                        CommandError::ExecutionFailed(format!("Storage error: {:?}", e))
+                    })?;
+
+                received_blocks.push(focused_block.index);
+                *state_guard = TransferState::Receiving {
+                    current_file: current_file.clone(),
+                    expected_blocks: *expected_blocks,
+                    focused_block: None,
+                    received_blocks: received_blocks.clone(),
+                };
+            }
+        } else {
+            println!("No focused block to store data for.");
+        }
+
+        Ok(())
+    }
 }
diff --git a/src/core/domain/network/entities.rs b/src/core/domain/network/entities.rs
@@ -1,4 +1,4 @@
-use std::{collections::HashSet, convert::TryFrom};
+use std::convert::TryFrom;
 
 use crate::core::domain::storage::entities::YeetBlock;
 
@@ -169,6 +169,7 @@ impl From<NetworkError> for String {
 pub enum TransferState {
     Idle,
     Receiving {
+        current_file: String,
         expected_blocks: u64,
         focused_block: Option<YeetBlock>,
         received_blocks: Vec<u64>,
diff --git a/src/core/domain/network/ports.rs b/src/core/domain/network/ports.rs
@@ -12,7 +12,7 @@ pub trait NetworkService {
         addr: &str,
         tx: Sender<TcpStream>,
     ) -> impl Future<Output = Result<(), NetworkError>> + Send;
-    fn handler(&self, rx: Receiver<TcpStream>) -> impl Future<Output = Result<(), Error>> + Send;
+    fn handler(&self, rx: Receiver<TcpStream>) -> impl Future<Output = Result<(), Error>>;
     fn trust_protocol(
         &self,
         stream: &mut TcpStream,
diff --git a/src/core/domain/network/services.rs b/src/core/domain/network/services.rs
@@ -22,7 +22,7 @@ pub struct NetworkServiceImpl<C>
 where
     C: CommandService,
 {
-    command_service: C,
+    pub command_service: C,
     active: Arc<AtomicBool>,
     transfer_state: Arc<Mutex<TransferState>>,
 }
@@ -133,59 +133,38 @@ where
                                 .send_message(stream_ref, ProtocolMessage::Error(String::from(e)))
                                 .await;
                         }
-                    }
-                    Err(e) => {
-                        let mut state_guard = self.transfer_state.lock().await;
-                        let was_receiving = match &mut *state_guard {
-                            TransferState::Receiving {
-                                expected_blocks,
-                                focused_block,
-                                received_blocks,
-                            } => {
-                                println!(
-                                    "In Receiving Binary from YeetBlock: expected_blocks={}, focused_block={:?}, received_blocks={:?}, binary data={}",
-                                    expected_blocks, focused_block, received_blocks, line
-                                );
-
-                                // TODO: Implement storage of binary data block here
 
-                                if let Some(focused_block) = focused_block {
-                                    if received_blocks.contains(&focused_block.index) {
-                                        println!(
-                                            "Block {} already received, ignoring.",
-                                            focused_block.index
-                                        );
-                                    } else {
-                                        println!("Stored binary data block: {:?}", focused_block);
-                                        received_blocks.push(focused_block.index);
-                                        *state_guard = TransferState::Receiving {
-                                            expected_blocks: *expected_blocks,
-                                            focused_block: None,
-                                            received_blocks: received_blocks.clone(),
-                                        };
-                                    }
-                                } else {
-                                    println!("No focused block to store data for.");
-                                }
+                        let guard = self.transfer_state.lock().await;
+                        match *guard {
+                            TransferState::Closed => {
+                                println!("Closing connection.");
+                                self.active
+                                    .store(false, std::sync::atomic::Ordering::SeqCst);
+                                drop(guard);
 
-                                drop(state_guard);
+                                self.reset_transfer_state().await;
+                                let _ = stream_ref.shutdown().await;
 
-                                true
+                                break;
                             }
-                            _ => false,
-                        };
-
-                        if !was_receiving {
-                            eprintln!("Failed to parse message: {:?}", e);
+                            _ => {
+                                continue;
+                            }
+                        }
+                    }
+                    Err(_) => {
+                        if let Err(e) = self
+                            .command_service
+                            .process_binary_data(Arc::clone(&self.transfer_state), &buf)
+                            .await
+                        {
+                            eprintln!("Error processing binary data: {:?}", e);
                             let _ = self
                                 .send_message(stream_ref, ProtocolMessage::Error(String::from(e)))
                                 .await;
                         }
                     }
                 }
-
-                println!("Received line: {}", line);
-                // handle msg here...
             }
         }
 
diff --git a/src/core/domain/storage/entities.rs b/src/core/domain/storage/entities.rs
@@ -20,3 +20,32 @@ pub struct File {
     pub name: String,
     pub size: u64,
 }
+
+#[derive(Debug)]
+pub enum StorageError {
+    FileNotFound,
+    PermissionDenied,
+    AlreadyExists,
+    AbsolutePathNotAllowed,
+    ParentDirSegmentNotAllowed,
+    InvalidFilename,
+    Unknown(String),
+}
+
+impl From<StorageError> for String {
+    fn from(error: StorageError) -> Self {
+        match error {
+            StorageError::FileNotFound => "File not found".into(),
+            StorageError::PermissionDenied => "Permission denied".into(),
+            StorageError::AlreadyExists => "File already exists".into(),
+            StorageError::AbsolutePathNotAllowed => {
+                "Absolute paths are not allowed in filenames".into()
+            }
+            StorageError::ParentDirSegmentNotAllowed => {
+                "Parent directory segments are not allowed in filenames".into()
+            }
+            StorageError::InvalidFilename => "Invalid filename".into(),
+            StorageError::Unknown(msg) => format!("Unknown storage error: {}", msg),
+        }
+    }
+}
diff --git a/src/core/domain/storage/mod.rs b/src/core/domain/storage/mod.rs
@@ -1 +1,2 @@
 pub mod entities;
+pub mod ports;
diff --git a/src/core/domain/storage/ports.rs b/src/core/domain/storage/ports.rs
@@ -1 +1,15 @@
+use crate::core::domain::storage::entities::{StorageError, YeetBlock};
 
+pub trait StorageRepository {
+    fn open_file(&self, filename: &str) -> impl Future<Output = Result<(), StorageError>> + Send;
+    fn write_block(
+        &self,
+        filename: &str,
+        block: &YeetBlock,
+        data: &[u8],
+    ) -> impl Future<Output = Result<(), StorageError>> + Send;
+    fn finalize(
+        &self,
+        filename: &str,
+    ) -> impl Future<Output = Result<(), StorageError>> + Send + Sync;
+}
diff --git a/src/core/domain/storage/services.rs b/src/core/domain/storage/services.rs
diff --git a/src/infra/mod.rs b/src/infra/mod.rs
@@ -0,0 +1 @@
+pub mod repositories;
diff --git a/src/infra/repositories/fs/fs_storage_repository.rs b/src/infra/repositories/fs/fs_storage_repository.rs
diff --git a/src/infra/repositories/fs/mod.rs b/src/infra/repositories/fs/mod.rs
diff --git a/src/infra/repositories/mod.rs b/src/infra/repositories/mod.rs
diff --git a/src/lib.rs b/src/lib.rs
diff --git a/src/main.rs b/src/main.rs

Original file line number	Diff line number	Diff line change
`@@ -1 +1,2 @@`
`1`	`1`	`pub mod entities;`
	`2`	`+pub mod ports;`