Flush command

respd/src/cmd.rs

@@ -77,6 +77,7 @@ pub enum Command {
         property: String,
         value: String,
     },
+    Flush,
     // Add more commands as needed
 }
 
@@ -423,6 +424,12 @@ impl Command {
 
                         Ok(Command::Scan { cursor })
                     }
+                    "FLUSH" => {
+                        if array.len() != 1 {
+                            return Err(CommandError::WrongNumberOfArguments("FLUSH".to_string()));
+                        }
+                        Ok(Command::Flush)
+                    }
                     _ => Err(CommandError::UnknownCommand(command_name)),
                 }
             }
@@ -502,6 +509,7 @@ impl CommandHandler {
                 property,
                 value,
             } => self.handle_nsset(namespace, property, value).await,
+            Command::Flush => self.handle_flush().await,
             Command::DBSize => self.handle_dbsize(),
             Command::Scan { cursor } => self.handle_scan(cursor).await,
             Command::Select { .. } => {
@@ -883,4 +891,45 @@ impl CommandHandler {
             }
         }
     }
+
+    /// Handle FLUSH command - delete all keys in the current namespace
+    /// This command is only allowed on private and password protected namespaces
+    async fn handle_flush(&self) -> Frame {
+        debug!("Handling FLUSH command");
+
+        // Get namespace properties
+        let props = self.namespace.properties.read().unwrap();
+        let namespace_name = props.namespace_name.clone();
+
+        // Check if the namespace is private (not public)
+        if props.public {
+            return Frame::Error("ERR: FLUSH command is only allowed on private namespaces".into());
+        }
+
+        // Check if the namespace is password-protected by checking if it has a password
+        let has_password = match self.storage.get_namespace_meta(&namespace_name) {
+            Ok(meta) => meta.password.is_some(),
+            Err(_) => false,
+        };
+
+        if !has_password {
+            return Frame::Error(
+                "ERR: FLUSH command is only allowed on password-protected namespaces".into(),
+            );
+        }
+
+        // Check if the connection is authenticated for this namespace
+        if !self.namespace_authenticated {
+            return Frame::Error("ERR: Authentication required for FLUSH command".into());
+        }
+
+        // Execute the flush operation
+        match self.namespace.flush(self.namespace_cache.as_ref()) {
+            Ok(_) => Frame::SimpleString("OK".into()),
+            Err(e) => {
+                error!("Error flushing namespace: {}", e);
+                Frame::Error(format!("ERR {}", e))
+            }
+        }
+    }
 }

respd/src/namespace.rs

@@ -37,7 +37,7 @@ impl Default for NamespaceProperties {
 /// Represents a namespace with its associated tree
 pub struct Namespace {
     /// The tree for this namespace
-    pub tree: Arc<dyn MetaTreeExt + Send + Sync>,
+    pub tree: RwLock<Arc<dyn MetaTreeExt + Send + Sync>>,
     /// Properties for this namespace
     pub properties: RwLock<NamespaceProperties>,
 }
@@ -81,7 +81,7 @@ impl NamespaceCache {
             ..Default::default()
         };
         let namespace = Arc::new(Namespace {
-            tree: Arc::from(tree),
+            tree: RwLock::new(Arc::from(tree)),
             properties: RwLock::new(props),
         });
 
@@ -129,7 +129,7 @@ impl NamespaceCache {
                     ..Default::default()
                 };
                 let namespace = Arc::new(Namespace {
-                    tree: Arc::from(tree),
+                    tree: RwLock::new(Arc::from(tree)),
                     properties: RwLock::new(props),
                 });
 
@@ -148,6 +148,48 @@ impl NamespaceCache {
             }
         }
     }
+
+    /// Flush a namespace by dropping and recreating its bucket
+    /// This operation will clear all keys in the namespace
+    pub fn flush_namespace(&self, name: &str) -> Result<(), StorageError> {
+        // Write lock the cache to prevent concurrent access during flush
+        let namespaces_lock = self.namespaces.read().unwrap();
+
+        // Get the namespace from the cache
+        if let Some(namespace) = namespaces_lock.get(name) {
+            // Get current namespace metadata before dropping the bucket
+            let namespace_meta = self.storage.get_namespace_meta(name)?;
+
+            // Drop the old tree by replacing it with a new one
+            {
+                let placeholder_tree = self.storage.get_namespace("default")?;
+
+                // Get a write lock on the tree
+                let mut tree_lock = namespace.tree.write().unwrap();
+
+                // Replace the old tree with the placeholder
+                *tree_lock = Arc::from(placeholder_tree);
+
+                // The lock will be dropped at the end of this scope, releasing the placeholder tree
+            }
+
+            // Drop the bucket from storage
+            self.storage.delete_namespace(name)?;
+
+            // Create a new namespace with the same name
+            let new_tree = self.storage.create_namespace(name)?;
+
+            // Assign the new tree to the namespace
+            let mut tree_lock = namespace.tree.write().unwrap();
+            *tree_lock = Arc::from(new_tree);
+
+            // Restore the original metadata to preserve properties
+            self.storage.update_namespace_meta(name, namespace_meta)?;
+        }
+
+        debug!("Flushed namespace: {}", name);
+        Ok(())
+    }
 }
 
 impl Namespace {
@@ -165,6 +207,19 @@ impl Namespace {
         Ok(())
     }
 
+    pub fn flush(&self, namespace_cache: &NamespaceCache) -> Result<()> {
+        // Get the namespace name
+        let namespace_name = self.properties.read().unwrap().namespace_name.clone();
+
+        // Check if this is the default namespace
+        if namespace_name == "default" {
+            return Err(anyhow::anyhow!("ERR: Cannot flush the default namespace"));
+        }
+
+        namespace_cache.flush_namespace(&namespace_name)?;
+        Ok(())
+    }
+
     pub fn set(&self, key: &[u8], value: Bytes) -> Result<()> {
         // Read namespace properties
         let props = self.properties.read().unwrap();
@@ -191,13 +246,13 @@ impl Namespace {
         let hash = Md5::digest(&data).into();
         let size = data.len() as u64;
         let obj_meta = Object::new(size, hash, ObjectData::Inline { data });
-        self.tree.insert(key, obj_meta.to_vec())?;
+        self.tree.read().unwrap().insert(key, obj_meta.to_vec())?;
         Ok(())
     }
 
     /// Get an Object from the tree for a given key
     fn get_object(&self, key: &[u8]) -> Result<Option<Object>, MetaError> {
-        match self.tree.get(key)? {
+        match self.tree.read().unwrap().get(key)? {
             Some(data) => {
                 let obj = Object::try_from(&*data).expect("Malformed object");
                 Ok(Some(obj))
@@ -242,12 +297,12 @@ impl Namespace {
         // Note: Authentication check is now handled by the CommandHandler
 
         // Proceed with deleting the key
-        self.tree.remove(key)?;
+        self.tree.read().unwrap().remove(key)?;
         Ok(())
     }
 
     pub fn exists(&self, key: &[u8]) -> Result<bool, MetaError> {
-        self.tree.contains_key(key)
+        self.tree.read().unwrap().contains_key(key)
     }
 
     /// Get the length (size) of a key's value
@@ -297,7 +352,7 @@ impl Namespace {
     }
 
     pub fn num_keys(&self) -> usize {
-        self.tree.len()
+        self.tree.read().unwrap().len()
     }
 
     pub fn scan(
@@ -308,7 +363,7 @@ impl Namespace {
         let mut keys = Vec::new();
         let mut count = 0;
 
-        for result in self.tree.iter_kv(start_after) {
+        for result in self.tree.read().unwrap().iter_kv(start_after) {
             match result {
                 Ok((key, _)) => {
                     keys.push(key);

respd/src/storage.rs

@@ -102,13 +102,25 @@ impl Storage {
         if self.store.bucket_exists(name)? {
             return Err(StorageError::NamespaceNotFound);
         }
+
         let namespace_meta_raw = NamespaceMeta::new(name.to_string())
             .to_msgpack()
             .map_err(|e| MetaError::OtherDBError(e.to_string()))?;
+
         self.store.insert_bucket(name, namespace_meta_raw)?;
+
         self.get_namespace(name)
     }
 
+    pub fn delete_namespace(&self, name: &str) -> Result<(), StorageError> {
+        if !self.store.bucket_exists(name)? {
+            return Err(StorageError::NamespaceNotFound);
+        }
+
+        self.store.drop_bucket(name)?;
+        Ok(())
+    }
+
     /// Iterate over all namespaces in the storage
     ///
     /// Returns an iterator that yields NamespaceMeta structs for each namespace

respd/tests/command_test.rs

@@ -63,3 +63,29 @@ fn test_select_command_parsing() {
         Err(CommandError::WrongNumberOfArguments(_))
     ));
 }
+
+#[test]
+fn test_flush_command_parsing() {
+    // Test FLUSH command with correct number of arguments (1 argument - just the command name)
+    let frame = Frame::Array(vec![Frame::BulkString(b"FLUSH".to_vec())]);
+
+    let cmd = Command::from_frame(frame).unwrap();
+    match cmd {
+        Command::Flush => {
+            // Command parsed correctly
+        }
+        _ => panic!("Expected FLUSH command"),
+    }
+
+    // Test FLUSH with too many arguments
+    let frame = Frame::Array(vec![
+        Frame::BulkString(b"FLUSH".to_vec()),
+        Frame::BulkString(b"extra_arg".to_vec()),
+    ]);
+
+    let result = Command::from_frame(frame);
+    assert!(matches!(
+        result,
+        Err(CommandError::WrongNumberOfArguments(_))
+    ));
+}