Repository: Migorithm/duva
Pull request #264 (2025-02-19 14:12:33)

refactor: layered architecture


Patch:
diff --git a/src/actor_registry.rs b/src/actor_registry.rs
index 558c4993..9ea940b5 100644
--- a/src/actor_registry.rs
+++ b/src/actor_registry.rs
@@ -1,12 +1,11 @@
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::config_actors::config_manager::ConfigManager;
+use crate::domains::saves::snapshot::snapshot_applier::SnapshotApplier;
+use crate::domains::ttl::manager::TtlSchedulerManager;
 use crate::presentation::cluster_in::connection_manager::ClusterConnectionManager;
-use crate::services::statefuls::snapshot::snapshot_applier::SnapshotApplier;
+
 use crate::{
-    presentation::cluster_in::communication_manager::ClusterCommunicationManager,
-    services::{
-        cluster::actors::commands::ClusterCommand, config::manager::ConfigManager,
-        statefuls::cache::ttl::manager::TtlSchedulerManager,
-    },
-    CacheManager,
+    presentation::cluster_in::communication_manager::ClusterCommunicationManager, CacheManager,
 };
 use tokio::sync::mpsc::Sender;
 
diff --git a/src/adapters/aof/local_aof.rs b/src/adapters/aof/local_aof.rs
index 7128a6db..4585c406 100644
--- a/src/adapters/aof/local_aof.rs
+++ b/src/adapters/aof/local_aof.rs
@@ -1,11 +1,14 @@
 //! A local append-only file (AOF) adapter.
-use crate::services::aof::{TAof, WriteOperation, WriteRequest};
+
 use anyhow::Result;
 use bytes::BytesMut;
 use std::path::{Path, PathBuf};
 use tokio::fs::{File, OpenOptions};
 use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};
 
+use crate::domains::append_only_files::interfaces::TAof;
+use crate::domains::append_only_files::{WriteOperation, WriteRequest};
+
 /// A local append-only file (AOF) implementation.
 pub struct LocalAof {
     /// The file path where the AOF data is stored.
diff --git a/src/adapters/io/tokio_stream.rs b/src/adapters/io/tokio_stream.rs
index 0e99ac37..46eb997c 100644
--- a/src/adapters/io/tokio_stream.rs
+++ b/src/adapters/io/tokio_stream.rs
@@ -1,6 +1,7 @@
-use crate::services::error::IoError;
+use crate::domains::query_parsers::{deserialize, QueryIO};
+use crate::domains::IoError;
 use crate::services::interface::{TGetPeerIp, TRead, TWrite};
-use crate::services::query_io::{deserialize, QueryIO};
+
 use bytes::{Bytes, BytesMut};
 use std::io::ErrorKind;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
diff --git a/src/domains/append_only_files/interfaces.rs b/src/domains/append_only_files/interfaces.rs
new file mode 100644
index 00000000..330dd9dc
--- /dev/null
+++ b/src/domains/append_only_files/interfaces.rs
@@ -0,0 +1,21 @@
+use super::WriteOperation;
+use anyhow::Result;
+
+/// Trait for an Append-Only File (AOF) abstraction.
+pub trait TAof {
+    /// Appends a single `WriteOperation` to the log.
+    fn append(
+        &mut self,
+        op: WriteOperation,
+    ) -> impl std::future::Future<Output = Result<()>> + Send;
+
+    /// Replays all logged operations from the beginning of the AOF, calling the provided callback `f` for each operation.
+    ///
+    /// The callback `f(WriteOperation)` receives each operation in the order it was appended.
+    fn replay<F>(&mut self, f: F) -> impl std::future::Future<Output = Result<()>> + Send
+    where
+        F: FnMut(WriteOperation) + Send;
+
+    /// Forces pending writes to be physically recorded on disk.
+    fn fsync(&mut self) -> impl std::future::Future<Output = Result<()>> + Send;
+}
diff --git a/src/services/aof.rs b/src/domains/append_only_files/log.rs
similarity index 75%
rename from src/services/aof.rs
rename to src/domains/append_only_files/log.rs
index 42b2cd6b..f4631d87 100644
--- a/src/services/aof.rs
+++ b/src/domains/append_only_files/log.rs
@@ -1,27 +1,9 @@
-use super::query_io::deserialize as deserialize_query_io;
-use super::query_io::QueryIO;
-use crate::write_array;
-use anyhow::Result;
 use bytes::{Bytes, BytesMut};
 
-/// Trait for an Append-Only File (AOF) abstraction.
-pub trait TAof {
-    /// Appends a single `WriteOperation` to the log.
-    fn append(
-        &mut self,
-        op: WriteOperation,
-    ) -> impl std::future::Future<Output = Result<()>> + Send;
-
-    /// Replays all logged operations from the beginning of the AOF, calling the provided callback `f` for each operation.
-    ///
-    /// The callback `f(WriteOperation)` receives each operation in the order it was appended.
-    fn replay<F>(&mut self, f: F) -> impl std::future::Future<Output = Result<()>> + Send
-    where
-        F: FnMut(WriteOperation) + Send;
-
-    /// Forces pending writes to be physically recorded on disk.
-    fn fsync(&mut self) -> impl std::future::Future<Output = Result<()>> + Send;
-}
+use crate::{
+    domains::query_parsers::{deserialize, QueryIO},
+    write_array,
+};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct WriteOperation {
@@ -69,11 +51,11 @@ impl WriteRequest {
     }
 
     /// Deserialize `WriteOperation`s from the given bytes.
-    pub fn deserialize(mut bytes: BytesMut) -> Result<Vec<WriteOperation>> {
+    pub fn deserialize(mut bytes: BytesMut) -> anyhow::Result<Vec<WriteOperation>> {
         let mut ops = Vec::new();
 
         while !bytes.is_empty() {
-            let (query, consumed) = deserialize_query_io(bytes.clone())?;
+            let (query, consumed) = deserialize(bytes.clone())?;
             bytes = bytes.split_off(consumed);
 
             let QueryIO::ReplicateLog(write_operation) = query else {
@@ -85,7 +67,7 @@ impl WriteRequest {
         Ok(ops)
     }
 
-    pub fn new(cmd: String, args: std::vec::IntoIter<QueryIO>) -> Result<Self> {
+    pub fn new(cmd: String, args: std::vec::IntoIter<QueryIO>) -> anyhow::Result<Self> {
         match cmd.as_str() {
             "set" => Self::to_set(args),
 
diff --git a/src/domains/append_only_files/mod.rs b/src/domains/append_only_files/mod.rs
new file mode 100644
index 00000000..425ac84a
--- /dev/null
+++ b/src/domains/append_only_files/mod.rs
@@ -0,0 +1,5 @@
+pub mod interfaces;
+pub mod log;
+
+pub use log::WriteOperation;
+pub use log::WriteRequest;
diff --git a/src/domains/caches/actor.rs b/src/domains/caches/actor.rs
new file mode 100644
index 00000000..83132018
--- /dev/null
+++ b/src/domains/caches/actor.rs
@@ -0,0 +1,88 @@
+use std::collections::HashMap;
+
+use super::cache_objects::{CacheEntry, CacheValue};
+use super::command::CacheCommand;
+use crate::domains::query_parsers::QueryIO;
+use crate::domains::ttl::manager::TtlSchedulerManager;
+use crate::make_smart_pointer;
+
+use anyhow::Context;
+use tokio::sync::mpsc;
+
+#[derive(Default)]
+pub struct CacheActor {
+    pub(crate) cache: CacheDb,
+}
+
+#[derive(Default)]
+pub struct CacheDb {
+    inner: HashMap<String, CacheValue>,
+    // OPTIMIZATION: Add a counter to keep track of the number of keys with expiry
+    pub(crate) keys_with_expiry: usize,
+}
+
+impl CacheActor {
+    pub fn run() -> CacheCommandSender {
+        let (tx, cache_actor_inbox) = mpsc::channel(100);
+        tokio::spawn(Self::default().handle(cache_actor_inbox));
+        CacheCommandSender(tx)
+    }
+
+    pub(crate) fn len(&self) -> usize {
+        self.cache.len()
+    }
+    pub(crate) fn keys_with_expiry(&self) -> usize {
+        self.cache.keys_with_expiry
+    }
+
+    pub(crate) fn keys_stream(
+        &self,
+        pattern: Option<String>,
+    ) -> impl Iterator<Item = QueryIO> + '_ {
+        self.cache.keys().filter_map(move |k| {
+            if pattern.as_ref().map_or(true, |p| k.contains(p)) {
+                Some(QueryIO::BulkString(k.clone().into()))
+            } else {
+                None
+            }
+        })
+    }
+    pub(crate) fn delete(&mut self, key: &str) {
+        if let Some(value) = self.cache.remove(key) {
+            if value.has_expiry() {
+                self.cache.keys_with_expiry -= 1;
+            }
+        }
+    }
+    pub(crate) fn get(&self, key: &str) -> Option<CacheValue> {
+        self.cache.get(key).cloned()
+    }
+
+    pub(crate) fn set(&mut self, cache_entry: CacheEntry) {
+        match cache_entry {
+            CacheEntry::KeyValue(key, value) => {
+                self.cache.insert(key, CacheValue::Value(value));
+            }
+            CacheEntry::KeyValueExpiry(key, value, expiry) => {
+                self.cache.keys_with_expiry += 1;
+                self.cache.insert(key.clone(), CacheValue::ValueWithExpiry(value, expiry));
+            }
+        }
+    }
+
+    pub(crate) async fn try_send_ttl(
+        &self,
+        key: &str,
+        expiry: Option<std::time::SystemTime>,
+        ttl_sender: TtlSchedulerManager,
+    ) -> anyhow::Result<()> {
+        ttl_sender.set_ttl(key.to_string(), expiry.context("Expiry not found")?).await;
+        Ok(())
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct CacheCommandSender(mpsc::Sender<CacheCommand>);
+
+make_smart_pointer!(CacheCommandSender, mpsc::Sender<CacheCommand>);
+make_smart_pointer!(CacheDb, HashMap<String, CacheValue> => inner);
diff --git a/src/services/statefuls/cache/manager.rs b/src/domains/caches/cache_manager.rs
similarity index 90%
rename from src/services/statefuls/cache/manager.rs
rename to src/domains/caches/cache_manager.rs
index 7a4ba54b..ed54f3bc 100644
--- a/src/services/statefuls/cache/manager.rs
+++ b/src/domains/caches/cache_manager.rs
@@ -1,11 +1,12 @@
-use super::actor::CacheActor;
-use super::actor::CacheCommand;
-use super::actor::CacheCommandSender;
-use super::CacheEntry;
-use crate::services::query_io::QueryIO;
-use crate::services::statefuls::cache::ttl::manager::TtlSchedulerManager;
-use crate::services::statefuls::snapshot::save::actor::SaveActor;
-use crate::services::statefuls::snapshot::save::actor::SaveTarget;
+use crate::domains::caches::actor::CacheActor;
+use crate::domains::caches::actor::CacheCommandSender;
+use crate::domains::caches::cache_objects::CacheEntry;
+use crate::domains::caches::command::CacheCommand;
+use crate::domains::query_parsers::QueryIO;
+use crate::domains::saves::actor::SaveActor;
+use crate::domains::saves::actor::SaveTarget;
+use crate::domains::ttl::manager::TtlSchedulerManager;
+
 use anyhow::Result;
 use std::{hash::Hasher, iter::Zip};
 use tokio::sync::oneshot::Sender;
diff --git a/src/services/statefuls/cache/mod.rs b/src/domains/caches/cache_objects.rs
similarity index 97%
rename from src/services/statefuls/cache/mod.rs
rename to src/domains/caches/cache_objects.rs
index 9600be1d..c6f5d5f1 100644
--- a/src/services/statefuls/cache/mod.rs
+++ b/src/domains/caches/cache_objects.rs
@@ -1,10 +1,5 @@
 use std::time::SystemTime;
 
-pub mod actor;
-pub mod manager;
-
-pub mod ttl;
-
 #[derive(Debug, Clone)]
 pub enum CacheEntry {
     KeyValue(String, String),
diff --git a/src/domains/caches/command.rs b/src/domains/caches/command.rs
new file mode 100644
index 00000000..2f726fe8
--- /dev/null
+++ b/src/domains/caches/command.rs
@@ -0,0 +1,16 @@
+use tokio::sync::{mpsc, oneshot};
+
+use crate::domains::{
+    query_parsers::QueryIO, saves::command::SaveCommand, ttl::manager::TtlSchedulerManager,
+};
+
+use super::cache_objects::CacheEntry;
+
+pub enum CacheCommand {
+    Set { cache_entry: CacheEntry, ttl_sender: TtlSchedulerManager },
+    Save { outbox: mpsc::Sender<SaveCommand> },
+    Get { key: String, sender: oneshot::Sender<QueryIO> },
+    Keys { pattern: Option<String>, sender: oneshot::Sender<QueryIO> },
+    Delete(String),
+    StopSentinel,
+}
diff --git a/src/domains/caches/mod.rs b/src/domains/caches/mod.rs
new file mode 100644
index 00000000..73cf7f86
--- /dev/null
+++ b/src/domains/caches/mod.rs
@@ -0,0 +1,4 @@
+pub mod actor;
+pub mod cache_manager;
+pub mod cache_objects;
+pub mod command;
diff --git a/src/services/cluster/actors/actor.rs b/src/domains/cluster_actors/actor.rs
similarity index 52%
rename from src/services/cluster/actors/actor.rs
rename to src/domains/cluster_actors/actor.rs
index 2c257358..783f617e 100644
--- a/src/services/cluster/actors/actor.rs
+++ b/src/domains/cluster_actors/actor.rs
@@ -1,119 +1,39 @@
-use crate::services::aof::{WriteOperation, WriteRequest};
-use crate::services::cluster::actors::replication::{
-    time_in_secs, BannedPeer, HeartBeatMessage, ReplicationInfo,
+use crate::domains::{
+    append_only_files::{WriteOperation, WriteRequest},
+    query_parsers::QueryIO,
 };
-use crate::services::cluster::peers::identifier::PeerIdentifier;
-use crate::services::cluster::peers::kind::PeerKind;
-use crate::services::cluster::peers::peer::Peer;
-use crate::services::interface::TWrite;
-use crate::services::query_io::QueryIO;
-use futures::stream::FuturesUnordered;
-use futures::StreamExt;
-use std::collections::BTreeMap;
-use tokio::sync::mpsc::Receiver;
 
-use tokio::time::Instant;
-
-use super::commands::{AddPeer, ClusterCommand};
-use super::consensus::ConsensusTracker;
-
-const FANOUT: usize = 2;
+use super::{
+    commands::AddPeer,
+    replication::{time_in_secs, BannedPeer, HeartBeatMessage, ReplicationInfo},
+    *,
+};
 
 #[derive(Debug)]
 pub struct ClusterActor {
-    members: BTreeMap<PeerIdentifier, Peer>,
-    replication: ReplicationInfo,
-    node_timeout: u128,
+    pub(crate) members: BTreeMap<PeerIdentifier, Peer>,
+    pub(crate) replication: ReplicationInfo,
+    pub(crate) node_timeout: u128,
+    notifier: tokio::sync::watch::Sender<bool>,
 }
 
 impl ClusterActor {
-    pub fn new(node_timeout: u128, init_repl_info: ReplicationInfo) -> Self {
-        Self { members: BTreeMap::new(), replication: init_repl_info, node_timeout }
-    }
-    pub async fn handle(
-        mut self,
-        mut cluster_message_listener: Receiver<ClusterCommand>,
+    pub fn new(
+        node_timeout: u128,
+        init_repl_info: ReplicationInfo,
         notifier: tokio::sync::watch::Sender<bool>,
-    ) {
-        let mut consensus_con = ConsensusTracker::default();
-
-        while let Some(command) = cluster_message_listener.recv().await {
-            // TODO notifier will be used when election process is implemented
-            let _ = notifier.clone();
-
-            match command {
-                ClusterCommand::AddPeer(add_peer_cmd) => {
-                    self.add_peer(add_peer_cmd).await;
-                }
-
-                ClusterCommand::GetPeers(callback) => {
-                    let _ = callback.send(self.members.keys().cloned().collect::<Vec<_>>().into());
-                }
-
-                ClusterCommand::SendHeartBeat => {
-                    let hop_count = self.hop_count(FANOUT, self.members.len());
-                    self.send_liveness_heartbeat(hop_count).await;
-
-                    // ! remove idle peers based on ttl.
-                    // ! The following may need to be moved else where to avoid blocking the main loop
-                    self.remove_idle_peers().await;
-                }
-
-                ClusterCommand::ReplicationInfo(sender) => {
-                    let _ = sender.send(self.replication.clone());
-                }
-                ClusterCommand::SetReplicationInfo { leader_repl_id, offset } => {
-                    self.set_replication_info(leader_repl_id, offset);
-                }
-                ClusterCommand::ReceiveHeartBeat(heartbeat) => {
-                    if self.replication.in_ban_list(&heartbeat.heartbeat_from) {
-                        continue;
-                    }
-
-                    self.gossip(heartbeat.hop_count).await;
-                    self.update_on_report(heartbeat).await;
-                }
-                ClusterCommand::ForgetPeer(peer_addr, sender) => {
-                    if let Ok(Some(())) = self.forget_peer(peer_addr).await {
-                        let _ = sender.send(Some(()));
-                    } else {
-                        let _ = sender.send(None);
-                    }
-                }
-                ClusterCommand::LeaderReqConsensus { log, sender } => {
-                    // ! If there are no replicas, don't send the request
-                    if self.followers().count() == 0 {
-                        let _ = sender.send(None);
-                        continue;
-                    }
-                    self.req_consensus(log).await;
-
-                    consensus_con.add(
-                        self.replication.leader_repl_offset,
-                        sender,
-                        self.followers().count(),
-                    );
-                }
-                ClusterCommand::LeaderReceiveAcks(offsets) => {
-                    self.apply_acks(&mut consensus_con, offsets);
-                }
-                ClusterCommand::FollowerReceiveLogEntries(write_operations) => {
-                    // TODO handle the log entries
-                    println!("[INFO] Received log entries: {:?}", write_operations);
-                    self.receive_log_entries_from_leader(write_operations).await;
-                }
-            }
-        }
+    ) -> Self {
+        Self { members: BTreeMap::new(), replication: init_repl_info, node_timeout, notifier }
     }
 
-    fn hop_count(&self, fanout: usize, node_count: usize) -> u8 {
+    pub fn hop_count(&self, fanout: usize, node_count: usize) -> u8 {
         if node_count <= fanout {
             return 0;
         }
         node_count.ilog(fanout) as u8
     }
 
-    async fn send_liveness_heartbeat(&mut self, hop_count: u8) {
+    pub async fn send_liveness_heartbeat(&mut self, hop_count: u8) {
         // TODO randomly choose the peer to send the message
 
         for peer in self.members.values_mut() {
@@ -123,7 +43,7 @@ impl ClusterActor {
         }
     }
 
-    async fn add_peer(&mut self, add_peer_cmd: AddPeer) {
+    pub async fn add_peer(&mut self, add_peer_cmd: AddPeer) {
         let AddPeer { peer_id: peer_addr, peer } = add_peer_cmd;
 
         self.replication.remove_from_ban_list(&peer_addr);
@@ -135,7 +55,7 @@ impl ClusterActor {
             existing_peer.listener_kill_trigger.kill().await;
         }
     }
-    async fn remove_peer(&mut self, peer_addr: &PeerIdentifier) -> Option<()> {
+    pub async fn remove_peer(&mut self, peer_addr: &PeerIdentifier) -> Option<()> {
         if let Some(peer) = self.members.remove(peer_addr) {
             // stop the runnin process and take the connection in case topology changes are made
             let _read_connected = peer.listener_kill_trigger.kill().await;
@@ -144,13 +64,13 @@ impl ClusterActor {
         None
     }
 
-    fn set_replication_info(&mut self, leader_repl_id: String, offset: u64) {
+    pub fn set_replication_info(&mut self, leader_repl_id: String, offset: u64) {
         self.replication.leader_repl_id = leader_repl_id;
         self.replication.leader_repl_offset = offset;
     }
 
     /// Remove the peers that are idle for more than ttl_mills
-    async fn remove_idle_peers(&mut self) {
+    pub async fn remove_idle_peers(&mut self) {
         // loop over members, if ttl is expired, remove the member
         let now = Instant::now();
 
@@ -166,7 +86,7 @@ impl ClusterActor {
         }
     }
 
-    async fn gossip(&mut self, hop_count: u8) {
+    pub async fn gossip(&mut self, hop_count: u8) {
         // If hop_count is 0, don't send the message to other peers
         if hop_count == 0 {
             return;
@@ -175,13 +95,13 @@ impl ClusterActor {
         self.send_liveness_heartbeat(hop_count).await;
     }
 
-    async fn forget_peer(&mut self, peer_addr: PeerIdentifier) -> anyhow::Result<Option<()>> {
+    pub async fn forget_peer(&mut self, peer_addr: PeerIdentifier) -> anyhow::Result<Option<()>> {
         self.replication.ban_peer(&peer_addr)?;
 
         Ok(self.remove_peer(&peer_addr).await)
     }
 
-    fn merge_ban_list(&mut self, ban_list: Vec<BannedPeer>) {
+    pub fn merge_ban_list(&mut self, ban_list: Vec<BannedPeer>) {
         if ban_list.is_empty() {
             return;
         }
@@ -193,7 +113,7 @@ impl ClusterActor {
         self.replication.ban_list.dedup_by_key(|node| node.p_id.clone());
     }
 
-    async fn update_on_report(&mut self, state: HeartBeatMessage) {
+    pub async fn update_on_report(&mut self, state: HeartBeatMessage) {
         let Some(peer) = self.members.get_mut(&state.heartbeat_from) else {
             return;
         };
@@ -213,7 +133,7 @@ impl ClusterActor {
         }
     }
 
-    async fn req_consensus(&mut self, req: WriteRequest) {
+    pub async fn req_consensus(&mut self, req: WriteRequest) {
         // TODO when are we going to increase offset?
 
         let heartbeat = self.replication.append_entry(0, req);
@@ -228,18 +148,18 @@ impl ClusterActor {
         while let Some(_) = tasks.next().await {}
     }
 
-    fn leader_mut(&mut self) -> Option<&mut Peer> {
+    pub(crate) fn leader_mut(&mut self) -> Option<&mut Peer> {
         self.members.values_mut().find(|peer| matches!(peer.w_conn.kind, PeerKind::Leader))
     }
 
-    fn followers(&self) -> impl Iterator<Item = &Peer> {
+    pub(crate) fn followers(&self) -> impl Iterator<Item = &Peer> {
         self.members
             .values()
             .into_iter()
             .filter(|peer| matches!(peer.w_conn.kind, PeerKind::Follower))
     }
 
-    fn followers_mut(&mut self) -> Vec<&mut Peer> {
+    pub(crate) fn followers_mut(&mut self) -> Vec<&mut Peer> {
         self.members
             .values_mut()
             .into_iter()
@@ -247,7 +167,7 @@ impl ClusterActor {
             .collect::<Vec<_>>()
     }
 
-    fn apply_acks(&self, consensus_con: &mut ConsensusTracker, offsets: Vec<u64>) {
+    pub fn apply_acks(&self, consensus_con: &mut ConsensusTracker, offsets: Vec<u64>) {
         offsets.into_iter().for_each(|offset| {
             if let Some(mut consensus) = consensus_con.take(&offset) {
                 println!("Received acks for offset: {}", offset);
@@ -260,7 +180,7 @@ impl ClusterActor {
         });
     }
 
-    async fn receive_log_entries_from_leader(&mut self, write_operations: Vec<WriteOperation>) {
+    pub async fn receive_log_entries_from_leader(&mut self, write_operations: Vec<WriteOperation>) {
         let offsets = write_operations.iter().map(|op| op.offset).collect::<Vec<_>>();
         if let Some(leader) = self.leader_mut() {
             let _ = leader.write_io(QueryIO::Acks(offsets)).await;
@@ -272,8 +192,10 @@ impl ClusterActor {
 fn test_hop_count_when_one() {
     // GIVEN
     let fanout = 2;
+
+    let (tx, rx) = tokio::sync::watch::channel(false);
     let replication = ReplicationInfo::new(None, "localhost", 8080);
-    let cluster_actor = ClusterActor::new(100, replication);
+    let cluster_actor = ClusterActor::new(100, replication, tx);
 
     // WHEN
     let hop_count = cluster_actor.hop_count(fanout, 1);
@@ -285,8 +207,9 @@ fn test_hop_count_when_one() {
 fn test_hop_count_when_two() {
     // GIVEN
     let fanout = 2;
+    let (tx, rx) = tokio::sync::watch::channel(false);
     let replication = ReplicationInfo::new(None, "localhost", 8080);
-    let cluster_actor = ClusterActor::new(100, replication);
+    let cluster_actor = ClusterActor::new(100, replication, tx);
 
     // WHEN
     let hop_count = cluster_actor.hop_count(fanout, 2);
@@ -298,8 +221,9 @@ fn test_hop_count_when_two() {
 fn test_hop_count_when_three() {
     // GIVEN
     let fanout = 2;
+    let (tx, rx) = tokio::sync::watch::channel(false);
     let replication = ReplicationInfo::new(None, "localhost", 8080);
-    let cluster_actor = ClusterActor::new(100, replication);
+    let cluster_actor = ClusterActor::new(100, replication, tx);
 
     // WHEN
     let hop_count = cluster_actor.hop_count(fanout, 3);
@@ -311,8 +235,9 @@ fn test_hop_count_when_three() {
 fn test_hop_count_when_thirty() {
     // GIVEN
     let fanout = 2;
+    let (tx, rx) = tokio::sync::watch::channel(false);
     let replication = ReplicationInfo::new(None, "localhost", 8080);
-    let cluster_actor = ClusterActor::new(100, replication);
+    let cluster_actor = ClusterActor::new(100, replication, tx);
 
     // WHEN
     let hop_count = cluster_actor.hop_count(fanout, 30);
diff --git a/src/services/cluster/actors/commands.rs b/src/domains/cluster_actors/commands.rs
similarity index 72%
rename from src/services/cluster/actors/commands.rs
rename to src/domains/cluster_actors/commands.rs
index e7283f1b..96dffb66 100644
--- a/src/services/cluster/actors/commands.rs
+++ b/src/domains/cluster_actors/commands.rs
@@ -1,9 +1,8 @@
-use crate::services::aof::{WriteOperation, WriteRequest};
+use crate::domains::append_only_files::WriteOperation;
+use crate::domains::peers::peer::Peer;
+use crate::domains::{append_only_files::WriteRequest, peers::identifier::PeerIdentifier};
 
-use crate::services::cluster::peers::identifier::PeerIdentifier;
-
-use crate::services::cluster::actors::replication::{HeartBeatMessage, ReplicationInfo};
-use crate::services::cluster::peers::peer::Peer;
+use super::replication::{HeartBeatMessage, ReplicationInfo};
 
 pub enum ClusterCommand {
     AddPeer(AddPeer),
diff --git a/src/services/cluster/actors/consensus.rs b/src/domains/cluster_actors/consensus.rs
similarity index 100%
rename from src/services/cluster/actors/consensus.rs
rename to src/domains/cluster_actors/consensus.rs
diff --git a/src/domains/cluster_actors/mod.rs b/src/domains/cluster_actors/mod.rs
new file mode 100644
index 00000000..6e07a8dc
--- /dev/null
+++ b/src/domains/cluster_actors/mod.rs
@@ -0,0 +1,20 @@
+use crate::domains::peers::identifier::PeerIdentifier;
+use crate::domains::peers::kind::PeerKind;
+use crate::domains::peers::peer::Peer;
+
+use crate::services::interface::TWrite;
+
+use consensus::ConsensusTracker;
+use futures::stream::FuturesUnordered;
+use futures::StreamExt;
+use std::collections::BTreeMap;
+
+use tokio::time::Instant;
+pub mod actor;
+pub mod commands;
+
+pub mod consensus;
+pub mod replication;
+mod replid_generator;
+pub const FANOUT: usize = 2;
+pub use actor::ClusterActor;
diff --git a/src/services/cluster/actors/replication.rs b/src/domains/cluster_actors/replication.rs
similarity index 97%
rename from src/services/cluster/actors/replication.rs
rename to src/domains/cluster_actors/replication.rs
index ddd61c03..808f2fc8 100644
--- a/src/services/cluster/actors/replication.rs
+++ b/src/domains/cluster_actors/replication.rs
@@ -1,5 +1,5 @@
-use crate::services::aof::{WriteOperation, WriteRequest};
-use crate::services::cluster::peers::identifier::PeerIdentifier;
+use crate::domains::append_only_files::{WriteOperation, WriteRequest};
+use crate::domains::peers::identifier::PeerIdentifier;
 
 use bytes::Bytes;
 use std::str::FromStr;
diff --git a/src/services/cluster/actors/replid_generator.rs b/src/domains/cluster_actors/replid_generator.rs
similarity index 100%
rename from src/services/cluster/actors/replid_generator.rs
rename to src/domains/cluster_actors/replid_generator.rs
diff --git a/src/domains/cluster_listeners/interfaces.rs b/src/domains/cluster_listeners/interfaces.rs
new file mode 100644
index 00000000..1ea500c8
--- /dev/null
+++ b/src/domains/cluster_listeners/interfaces.rs
@@ -0,0 +1,9 @@
+use super::ReactorKillSwitch;
+use tokio::net::tcp::OwnedReadHalf;
+
+pub trait TListen {
+    fn listen(
+        self,
+        rx: ReactorKillSwitch,
+    ) -> impl std::future::Future<Output = OwnedReadHalf> + Send;
+}
diff --git a/src/services/cluster/listeners/mod.rs b/src/domains/cluster_listeners/mod.rs
similarity index 66%
rename from src/services/cluster/listeners/mod.rs
rename to src/domains/cluster_listeners/mod.rs
index 897be007..80f098c0 100644
--- a/src/services/cluster/listeners/mod.rs
+++ b/src/domains/cluster_listeners/mod.rs
@@ -1,25 +1,14 @@
+pub mod interfaces;
 use super::{
-    actors::{commands::ClusterCommand, replication::HeartBeatMessage},
+    cluster_actors::{commands::ClusterCommand, replication::HeartBeatMessage},
     peers::{connected_types::ReadConnected, identifier::PeerIdentifier},
+    query_parsers::QueryIO,
+    saves::snapshot::snapshot_applier::SnapshotApplier,
 };
-use crate::services::{
-    interface::TRead, query_io::QueryIO, statefuls::snapshot::snapshot_applier::SnapshotApplier,
-};
-use tokio::select;
-use tokio::{net::tcp::OwnedReadHalf, sync::mpsc::Sender};
-
-pub mod follower;
-pub mod leader;
-pub mod non_data_peer;
-
-pub trait TListen {
-    fn listen(
-        self,
-        rx: ReactorKillSwitch,
-    ) -> impl std::future::Future<Output = OwnedReadHalf> + Send;
-}
-
-pub(super) type ReactorKillSwitch = tokio::sync::oneshot::Receiver<()>;
+use crate::services::interface::TRead;
+pub(crate) use interfaces::TListen;
+use tokio::sync::mpsc::Sender;
+pub(crate) type ReactorKillSwitch = tokio::sync::oneshot::Receiver<()>;
 
 // Listner requires cluster handler to send messages to the cluster actor and cluster actor instead needs kill trigger to stop the listener
 #[derive(Debug)]
@@ -29,6 +18,7 @@ pub(crate) struct ClusterListener<T> {
     pub(crate) self_id: PeerIdentifier,
     pub(crate) snapshot_applier: SnapshotApplier,
 }
+
 impl<T> ClusterListener<T> {
     pub fn new(
         read_connected: ReadConnected<T>,
@@ -39,12 +29,12 @@ impl<T> ClusterListener<T> {
         Self { read_connected, cluster_handler, self_id, snapshot_applier }
     }
     // Update peer state on cluster manager
-    async fn receive_heartbeat(&mut self, state: HeartBeatMessage) {
+    pub(crate) async fn receive_heartbeat(&mut self, state: HeartBeatMessage) {
         println!("[INFO] from {}, hc:{}", state.heartbeat_from, state.hop_count);
         let _ = self.cluster_handler.send(ClusterCommand::ReceiveHeartBeat(state)).await;
     }
 
-    async fn read_command<U>(&mut self) -> anyhow::Result<Vec<U>>
+    pub(crate) async fn read_command<U>(&mut self) -> anyhow::Result<Vec<U>>
     where
         U: std::convert::TryFrom<QueryIO>,
         U::Error: Into<anyhow::Error>,
diff --git a/src/domains/config_actors/actor.rs b/src/domains/config_actors/actor.rs
new file mode 100644
index 00000000..1de2d4f0
--- /dev/null
+++ b/src/domains/config_actors/actor.rs
@@ -0,0 +1,36 @@
+#[derive(Clone)]
+pub struct ConfigActor {
+    pub(crate) dir: &'static str,
+    pub dbfilename: &'static str,
+}
+impl ConfigActor {
+    pub fn new(dir: String, dbfilename: String) -> Self {
+        Self {
+            dir: Box::leak(dir.into_boxed_str()),
+            dbfilename: Box::leak(dbfilename.into_boxed_str()),
+        }
+    }
+
+    pub fn set_dir(&mut self, dir: &str) {
+        unsafe {
+            // Get the pointer to the str
+            let ptr: *const str = self.dir;
+            // Recreate the Box from the pointer and drop it
+            let _reclaimed_box: Box<str> = Box::from_raw(ptr as *mut str);
+            self.dir = Box::leak(dir.into());
+        }
+    }
+    pub fn set_dbfilename(&mut self, dbfilename: &str) {
+        unsafe {
+            // Get the pointer to the str
+            let ptr: *const str = self.dbfilename;
+            // Recreate the Box from the pointer and drop it
+            let _reclaimed_box: Box<str> = Box::from_raw(ptr as *mut str);
+            self.dbfilename = Box::leak(dbfilename.into());
+        }
+    }
+
+    pub fn get_filepath(&self) -> String {
+        format!("{}/{}", self.dir, self.dbfilename)
+    }
+}
diff --git a/src/services/config/command.rs b/src/domains/config_actors/command.rs
similarity index 100%
rename from src/services/config/command.rs
rename to src/domains/config_actors/command.rs
diff --git a/src/services/config/macros.rs b/src/domains/config_actors/conf_macros.rs
similarity index 100%
rename from src/services/config/macros.rs
rename to src/domains/config_actors/conf_macros.rs
diff --git a/src/services/config/manager.rs b/src/domains/config_actors/config_manager.rs
similarity index 88%
rename from src/services/config/manager.rs
rename to src/domains/config_actors/config_manager.rs
index 76fc2561..42cc3b18 100644
--- a/src/services/config/manager.rs
+++ b/src/domains/config_actors/config_manager.rs
@@ -1,9 +1,10 @@
-use super::actor::ConfigActor;
-use super::command::ConfigCommand;
-use super::command::ConfigMessage;
-use super::command::ConfigQuery;
-use super::ConfigResource;
-use super::ConfigResponse;
+use crate::domains::config_actors::actor::ConfigActor;
+use crate::domains::config_actors::command::ConfigCommand;
+use crate::domains::config_actors::command::ConfigMessage;
+use crate::domains::config_actors::command::ConfigQuery;
+use crate::domains::config_actors::command::ConfigResource;
+use crate::domains::config_actors::command::ConfigResponse;
+
 use std::time::SystemTime;
 use tokio::fs::try_exists;
 use tokio::sync::mpsc::Sender;
diff --git a/src/domains/config_actors/mod.rs b/src/domains/config_actors/mod.rs
new file mode 100644
index 00000000..a07ab8bb
--- /dev/null
+++ b/src/domains/config_actors/mod.rs
@@ -0,0 +1,4 @@
+pub mod actor;
+pub mod command;
+pub mod conf_macros;
+pub mod config_manager;
diff --git a/src/services/error.rs b/src/domains/error.rs
similarity index 100%
rename from src/services/error.rs
rename to src/domains/error.rs
diff --git a/src/domains/mod.rs b/src/domains/mod.rs
new file mode 100644
index 00000000..f8ffd48c
--- /dev/null
+++ b/src/domains/mod.rs
@@ -0,0 +1,11 @@
+pub mod append_only_files;
+pub mod caches;
+pub mod cluster_actors;
+pub mod cluster_listeners;
+pub mod config_actors;
+pub mod error;
+pub mod peers;
+pub mod query_parsers;
+pub mod saves;
+pub mod ttl;
+pub use error::IoError;
diff --git a/src/services/cluster/peers/connected_types.rs b/src/domains/peers/connected_types.rs
similarity index 93%
rename from src/services/cluster/peers/connected_types.rs
rename to src/domains/peers/connected_types.rs
index 54388408..28133f05 100644
--- a/src/services/cluster/peers/connected_types.rs
+++ b/src/domains/peers/connected_types.rs
@@ -1,6 +1,6 @@
 use std::marker::PhantomData;
 
-use crate::services::cluster::peers::kind::PeerKind;
+use crate::domains::peers::kind::PeerKind;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 
 #[derive(Debug)]
diff --git a/src/services/cluster/peers/identifier.rs b/src/domains/peers/identifier.rs
similarity index 100%
rename from src/services/cluster/peers/identifier.rs
rename to src/domains/peers/identifier.rs
diff --git a/src/services/cluster/peers/kind.rs b/src/domains/peers/kind.rs
similarity index 100%
rename from src/services/cluster/peers/kind.rs
rename to src/domains/peers/kind.rs
diff --git a/src/services/cluster/peers/mod.rs b/src/domains/peers/mod.rs
similarity index 100%
rename from src/services/cluster/peers/mod.rs
rename to src/domains/peers/mod.rs
diff --git a/src/services/cluster/peers/peer.rs b/src/domains/peers/peer.rs
similarity index 90%
rename from src/services/cluster/peers/peer.rs
rename to src/domains/peers/peer.rs
index c7656907..ad686fcb 100644
--- a/src/services/cluster/peers/peer.rs
+++ b/src/domains/peers/peer.rs
@@ -8,11 +8,12 @@ use super::connected_types::ReadConnected;
 use super::identifier::PeerIdentifier;
 use super::kind::PeerKind;
 
-use crate::services::cluster::actors::commands::ClusterCommand;
-use crate::services::cluster::listeners::ClusterListener;
-use crate::services::cluster::listeners::TListen;
-use crate::services::cluster::peers::connected_types::WriteConnected;
-use crate::services::statefuls::snapshot::snapshot_applier::SnapshotApplier;
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::cluster_listeners::ClusterListener;
+use crate::domains::cluster_listeners::TListen;
+use crate::domains::peers::connected_types::WriteConnected;
+use crate::domains::saves::snapshot::snapshot_applier::SnapshotApplier;
+
 use tokio::net::tcp::OwnedReadHalf;
 use tokio::net::tcp::OwnedWriteHalf;
 use tokio::net::TcpStream;
diff --git a/src/domains/query_parsers/mod.rs b/src/domains/query_parsers/mod.rs
new file mode 100644
index 00000000..c2bec940
--- /dev/null
+++ b/src/domains/query_parsers/mod.rs
@@ -0,0 +1,3 @@
+pub mod query_io;
+pub(crate) use query_io::deserialize;
+pub(crate) use query_io::QueryIO;
diff --git a/src/services/query_io.rs b/src/domains/query_parsers/query_io.rs
similarity index 94%
rename from src/services/query_io.rs
rename to src/domains/query_parsers/query_io.rs
index 3a4aa3de..a509b8b6 100644
--- a/src/services/query_io.rs
+++ b/src/domains/query_parsers/query_io.rs
@@ -1,6 +1,12 @@
-use super::aof::WriteRequest;
-use crate::services::statefuls::cache::CacheValue;
-use crate::services::{aof::WriteOperation, cluster::actors::replication::HeartBeatMessage};
+use crate::domains::append_only_files::{WriteOperation, WriteRequest};
+use crate::domains::cluster_actors::replication::HeartBeatMessage;
+#[cfg(test)]
+use crate::domains::cluster_actors::replication::{time_in_secs, BannedPeer, ReplicationInfo};
+#[cfg(test)]
+use crate::domains::peers::identifier::PeerIdentifier;
+
+use crate::domains::caches::cache_objects::CacheValue;
+
 use anyhow::{Context, Result};
 use bytes::{Bytes, BytesMut};
 use std::fmt::Write;
@@ -18,7 +24,7 @@ const ACKS_PREFIX: char = '@';
 #[macro_export]
 macro_rules! write_array {
     ($($x:expr),*) => {
-        $crate::services::query_io::QueryIO::Array(vec![$($crate::services::query_io::QueryIO::BulkString($x.into())),*])
+        $crate::domains::query_parsers::QueryIO::Array(vec![$($crate::domains::query_parsers::QueryIO::BulkString($x.into())),*])
     };
 }
 
@@ -452,8 +458,6 @@ fn test_from_bytes_to_peer_state() {
 
 #[test]
 fn test_from_heartbeat_to_bytes() {
-    use crate::services::query_io::QueryIO;
-
     //GIVEN
     let peer_state = HeartBeatMessage {
         term: 1,
@@ -504,8 +508,6 @@ fn test_from_heartbeat_to_bytes() {
 
 #[test]
 fn test_peer_state_ban_list_to_binary() {
-    use crate::services::cluster::actors::replication::ReplicationInfo;
-    use crate::services::cluster::peers::identifier::PeerIdentifier;
     // GIVEN
     let mut replication = ReplicationInfo::new(None, "127.0.0.1", 6380);
     let peer_id = PeerIdentifier::new("127.0.0.1", 6739);
@@ -525,9 +527,6 @@ fn test_peer_state_ban_list_to_binary() {
 
 #[test]
 fn test_binary_to_heartbeat() {
-    use crate::services::cluster::actors::replication::BannedPeer;
-    use crate::services::cluster::peers::identifier::PeerIdentifier;
-
     // GIVEN
     let binary= format!("^\r\n$1\r\n0\r\n$1\r\n0\r\n$6\r\nrandom\r\n$1\r\n1\r\n$14\r\n127.0.0.1:6379\r\n*1\r\n$22\r\n127.0.0.1:6739-6545442\r\n*2\r\n#\r\n$1\r\n1\r\n*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n#\r\n$1\r\n2\r\n*3\r\n$3\r\nSET\r\n$3\r\npoo\r\n$3\r\nbar\r\n");
     let buffer = BytesMut::from_iter(binary.into_bytes());
@@ -579,10 +578,6 @@ fn test_parse_file() {
 
 #[test]
 fn test_banned_peer_serde() {
-    use crate::services::cluster::actors::replication::time_in_secs;
-    use crate::services::cluster::actors::replication::BannedPeer;
-    use crate::services::cluster::actors::replication::ReplicationInfo;
-    use crate::services::cluster::peers::identifier::PeerIdentifier;
     //GIVEN
     let mut replication = ReplicationInfo::new(None, "127.0.0.1", 6380);
     let peer_id = PeerIdentifier::new("127.0.0.1", 6739);
@@ -603,11 +598,6 @@ fn test_banned_peer_serde() {
 
 #[test]
 fn test_banned_peer_serde_when_time_passed() {
-    use crate::services::cluster::actors::replication::time_in_secs;
-    use crate::services::cluster::actors::replication::BannedPeer;
-    use crate::services::cluster::actors::replication::ReplicationInfo;
-    use crate::services::cluster::peers::identifier::PeerIdentifier;
-
     //GIVEN
     let mut replication = ReplicationInfo::new(None, "127.0.0.1", 6380);
     let peer_id = PeerIdentifier::new("127.0.0.1", 6739);
diff --git a/src/services/statefuls/snapshot/save/actor.rs b/src/domains/saves/actor.rs
similarity index 66%
rename from src/services/statefuls/snapshot/save/actor.rs
rename to src/domains/saves/actor.rs
index 7e765ccd..d9b97f74 100644
--- a/src/services/statefuls/snapshot/save/actor.rs
+++ b/src/domains/saves/actor.rs
@@ -1,35 +1,19 @@
-use crate::services::error::IoError;
-use crate::services::statefuls::cache::CacheEntry;
-use crate::services::statefuls::snapshot::endec::encoder::byte_encoder::encode_checksum;
-use crate::services::statefuls::snapshot::endec::encoder::byte_encoder::encode_database_info;
-use crate::services::statefuls::snapshot::endec::encoder::byte_encoder::encode_database_table_size;
-use crate::services::statefuls::snapshot::endec::encoder::byte_encoder::encode_header;
-use crate::services::statefuls::snapshot::endec::encoder::byte_encoder::encode_metadata;
-use crate::services::statefuls::snapshot::save::command::SaveCommand;
-use anyhow::Result;
 use std::collections::VecDeque;
-use tokio::io::AsyncWriteExt;
 
-pub enum SaveTarget {
-    File(tokio::fs::File),
-    InMemory(Vec<u8>),
-}
+use crate::domains::{caches::cache_objects::CacheEntry, IoError};
+use tokio::io::AsyncWriteExt;
 
-impl SaveTarget {
-    pub async fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
-        match self {
-            SaveTarget::File(f) => f.write_all(buf).await.map_err(|e| e.kind().into()),
-            SaveTarget::InMemory(v) => {
-                v.extend_from_slice(buf);
-                Ok(())
-            }
-        }
-    }
-}
+use super::{
+    command::SaveCommand,
+    endec::encoder::byte_encoder::{
+        encode_checksum, encode_database_info, encode_database_table_size, encode_header,
+        encode_metadata,
+    },
+};
 
 pub struct SaveActor {
-    pub(in crate::services::statefuls::snapshot) target: SaveTarget,
-    pub(in crate::services::statefuls::snapshot) meta: SaveMeta,
+    pub(crate) target: SaveTarget,
+    pub(crate) meta: SaveMeta,
 }
 
 impl SaveActor {
@@ -38,34 +22,14 @@ impl SaveActor {
         num_of_shards: usize,
         repl_id: String,
         current_offset: u64,
-    ) -> Result<Self> {
+    ) -> anyhow::Result<Self> {
         let meta = SaveMeta::new(num_of_shards, repl_id, current_offset.to_string());
         let mut processor = Self { target, meta };
         processor.encode_meta().await?;
         Ok(processor)
     }
 
-    pub async fn run(
-        mut self,
-        mut inbox: tokio::sync::mpsc::Receiver<SaveCommand>,
-    ) -> Result<Self> {
-        while let Some(cmd) = inbox.recv().await {
-            match self.handle_cmd(cmd).await {
-                Ok(should_break) => {
-                    if should_break {
-                        break;
-                    }
-                }
-                Err(err) => {
-                    eprintln!("error while encoding: {:?}", err);
-                    return Err(err);
-                }
-            }
-        }
-        Ok(self)
-    }
-
-    pub async fn encode_meta(&mut self) -> Result<()> {
+    pub async fn encode_meta(&mut self) -> anyhow::Result<()> {
         let mut metadata: Vec<(&str, &str)> = vec![];
         if &self.meta.repl_id != "?" {
             metadata.push(("repl-id", &self.meta.repl_id));
@@ -76,7 +40,7 @@ impl SaveActor {
         self.target.write(&meta.concat()).await?;
         Ok(())
     }
-    pub async fn handle_cmd(&mut self, cmd: SaveCommand) -> Result<bool> {
+    pub async fn handle_cmd(&mut self, cmd: SaveCommand) -> anyhow::Result<bool> {
         match cmd {
             SaveCommand::LocalShardSize { table_size, expiry_size } => {
                 self.meta.total_key_value_table_size += table_size;
@@ -110,7 +74,7 @@ impl SaveActor {
         Ok(false)
     }
 
-    async fn encode_chunk_queue(&mut self) -> Result<()> {
+    async fn encode_chunk_queue(&mut self) -> anyhow::Result<()> {
         while let Some(chunk) = self.meta.chunk_queue.pop_front() {
             for kvs in chunk {
                 let encoded_chunk = kvs.encode_with_key()?;
@@ -129,14 +93,31 @@ impl SaveActor {
     }
 }
 
+pub enum SaveTarget {
+    File(tokio::fs::File),
+    InMemory(Vec<u8>),
+}
+
+impl SaveTarget {
+    pub async fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
+        match self {
+            SaveTarget::File(f) => f.write_all(buf).await.map_err(|e| e.kind().into()),
+            SaveTarget::InMemory(v) => {
+                v.extend_from_slice(buf);
+                Ok(())
+            }
+        }
+    }
+}
+
 pub struct SaveMeta {
-    num_of_saved_table_size_actor: usize,
-    total_key_value_table_size: usize,
-    total_expires_table_size: usize,
-    chunk_queue: VecDeque<Vec<CacheEntry>>,
-    num_of_cache_actors: usize,
-    repl_id: String,
-    offset: String,
+    pub(crate) num_of_saved_table_size_actor: usize,
+    pub(crate) total_key_value_table_size: usize,
+    pub(crate) total_expires_table_size: usize,
+    pub(crate) chunk_queue: VecDeque<Vec<CacheEntry>>,
+    pub(crate) num_of_cache_actors: usize,
+    pub(crate) repl_id: String,
+    pub(crate) offset: String,
 }
 
 impl SaveMeta {
diff --git a/src/services/statefuls/snapshot/save/command.rs b/src/domains/saves/command.rs
similarity index 71%
rename from src/services/statefuls/snapshot/save/command.rs
rename to src/domains/saves/command.rs
index 58629ed1..0c7546ca 100644
--- a/src/services/statefuls/snapshot/save/command.rs
+++ b/src/domains/saves/command.rs
@@ -1,4 +1,4 @@
-use crate::services::statefuls::cache::CacheEntry;
+use crate::domains::caches::cache_objects::CacheEntry;
 
 pub enum SaveCommand {
     LocalShardSize { table_size: usize, expiry_size: usize },
diff --git a/src/services/statefuls/snapshot/endec/decoder/byte_decoder.rs b/src/domains/saves/endec/decoder/byte_decoder.rs
similarity index 99%
rename from src/services/statefuls/snapshot/endec/decoder/byte_decoder.rs
rename to src/domains/saves/endec/decoder/byte_decoder.rs
index 769badaf..b28149d8 100644
--- a/src/services/statefuls/snapshot/endec/decoder/byte_decoder.rs
+++ b/src/domains/saves/endec/decoder/byte_decoder.rs
@@ -1,11 +1,13 @@
 use super::states::{DecoderInit, HeaderReady, MetadataReady};
-use crate::services::statefuls::cache::CacheEntry;
-use crate::services::statefuls::snapshot::endec::{
+
+use crate::domains::caches::cache_objects::CacheEntry;
+use crate::domains::saves::endec::{
     extract_range, StoredDuration, DATABASE_SECTION_INDICATOR, DATABASE_TABLE_SIZE_INDICATOR,
     EXPIRY_TIME_IN_MILLISECONDS_INDICATOR, EXPIRY_TIME_IN_SECONDS_INDICATOR, HEADER_MAGIC_STRING,
     METADATA_SECTION_INDICATOR, STRING_VALUE_TYPE_INDICATOR,
 };
-use crate::services::statefuls::snapshot::snapshot::{DecodedDatabase, DecodedMetadata, Snapshot};
+use crate::domains::saves::snapshot::snapshot::{DecodedDatabase, DecodedMetadata, Snapshot};
+
 use anyhow::{Context, Result};
 use std::{
     ops::{Deref, DerefMut},
diff --git a/src/services/statefuls/snapshot/endec/decoder/mod.rs b/src/domains/saves/endec/decoder/mod.rs
similarity index 100%
rename from src/services/statefuls/snapshot/endec/decoder/mod.rs
rename to src/domains/saves/endec/decoder/mod.rs
diff --git a/src/services/statefuls/snapshot/endec/decoder/states.rs b/src/domains/saves/endec/decoder/states.rs
similarity index 78%
rename from src/services/statefuls/snapshot/endec/decoder/states.rs
rename to src/domains/saves/endec/decoder/states.rs
index b080689e..7845844b 100644
--- a/src/services/statefuls/snapshot/endec/decoder/states.rs
+++ b/src/domains/saves/endec/decoder/states.rs
@@ -1,4 +1,4 @@
-use crate::services::statefuls::snapshot::snapshot::DecodedMetadata;
+use crate::domains::saves::snapshot::snapshot::DecodedMetadata;
 
 #[derive(Default)]
 pub struct DecoderInit;
diff --git a/src/services/statefuls/snapshot/endec/encoder/byte_encoder.rs b/src/domains/saves/endec/encoder/byte_encoder.rs
similarity index 98%
rename from src/services/statefuls/snapshot/endec/encoder/byte_encoder.rs
rename to src/domains/saves/endec/encoder/byte_encoder.rs
index a1524270..9e941f6e 100644
--- a/src/services/statefuls/snapshot/endec/encoder/byte_encoder.rs
+++ b/src/domains/saves/endec/encoder/byte_encoder.rs
@@ -1,6 +1,6 @@
-use crate::services::statefuls::{
-    cache::CacheEntry,
-    snapshot::endec::{
+use crate::domains::{
+    caches::cache_objects::CacheEntry,
+    saves::endec::{
         CHECKSUM_INDICATOR, DATABASE_SECTION_INDICATOR, DATABASE_TABLE_SIZE_INDICATOR,
         EXPIRY_TIME_IN_MILLISECONDS_INDICATOR, HEADER_MAGIC_STRING, METADATA_SECTION_INDICATOR,
         STRING_VALUE_TYPE_INDICATOR,
@@ -129,7 +129,7 @@ fn encode_size(size: usize) -> Result<Vec<u8>> {
 
 #[cfg(test)]
 mod test {
-    use crate::services::statefuls::snapshot::endec::{
+    use crate::domains::saves::endec::{
         decoder::{byte_decoder::BytesDecoder, states::DecoderInit},
         StoredDuration,
     };
diff --git a/src/services/statefuls/snapshot/endec/encoder/mod.rs b/src/domains/saves/endec/encoder/mod.rs
similarity index 100%
rename from src/services/statefuls/snapshot/endec/encoder/mod.rs
rename to src/domains/saves/endec/encoder/mod.rs
diff --git a/src/services/statefuls/snapshot/endec/mod.rs b/src/domains/saves/endec/mod.rs
similarity index 100%
rename from src/services/statefuls/snapshot/endec/mod.rs
rename to src/domains/saves/endec/mod.rs
diff --git a/src/domains/saves/mod.rs b/src/domains/saves/mod.rs
new file mode 100644
index 00000000..16a1e6d1
--- /dev/null
+++ b/src/domains/saves/mod.rs
@@ -0,0 +1,4 @@
+pub mod actor;
+pub mod command;
+pub mod endec;
+pub mod snapshot;
diff --git a/src/services/statefuls/snapshot/mod.rs b/src/domains/saves/snapshot/mod.rs
similarity index 70%
rename from src/services/statefuls/snapshot/mod.rs
rename to src/domains/saves/snapshot/mod.rs
index 53bfc266..1c427efd 100644
--- a/src/services/statefuls/snapshot/mod.rs
+++ b/src/domains/saves/snapshot/mod.rs
@@ -1,5 +1,3 @@
-pub mod endec;
-pub mod save;
 pub mod snapshot;
 pub mod snapshot_applier;
 pub mod snapshot_loader;
diff --git a/src/services/statefuls/snapshot/snapshot.rs b/src/domains/saves/snapshot/snapshot.rs
similarity index 95%
rename from src/services/statefuls/snapshot/snapshot.rs
rename to src/domains/saves/snapshot/snapshot.rs
index 0de7ed69..94556513 100644
--- a/src/services/statefuls/snapshot/snapshot.rs
+++ b/src/domains/saves/snapshot/snapshot.rs
@@ -1,4 +1,4 @@
-use crate::services::statefuls::cache::CacheEntry;
+use crate::domains::caches::cache_objects::CacheEntry;
 
 #[derive(Debug)]
 pub struct Snapshot {
diff --git a/src/services/statefuls/snapshot/snapshot_applier.rs b/src/domains/saves/snapshot/snapshot_applier.rs
similarity index 85%
rename from src/services/statefuls/snapshot/snapshot_applier.rs
rename to src/domains/saves/snapshot/snapshot_applier.rs
index 9b9c7eff..4e409b04 100644
--- a/src/services/statefuls/snapshot/snapshot_applier.rs
+++ b/src/domains/saves/snapshot/snapshot_applier.rs
@@ -1,11 +1,13 @@
 use futures::future::join_all;
 
-use crate::services::query_io::QueryIO;
-use crate::services::statefuls::cache::manager::CacheManager;
-use crate::services::statefuls::cache::ttl::manager::TtlSchedulerManager;
-use crate::services::statefuls::snapshot::snapshot::Snapshot;
+use crate::domains::caches::cache_manager::CacheManager;
+use crate::domains::query_parsers::QueryIO;
+use crate::domains::ttl::manager::TtlSchedulerManager;
+
 use std::time::SystemTime;
 
+use super::snapshot::Snapshot;
+
 #[derive(Clone, Debug)]
 pub struct SnapshotApplier {
     cache_manager: CacheManager,
diff --git a/src/services/statefuls/snapshot/snapshot_loader.rs b/src/domains/saves/snapshot/snapshot_loader.rs
similarity index 77%
rename from src/services/statefuls/snapshot/snapshot_loader.rs
rename to src/domains/saves/snapshot/snapshot_loader.rs
index 36076a33..b390a8a1 100644
--- a/src/services/statefuls/snapshot/snapshot_loader.rs
+++ b/src/domains/saves/snapshot/snapshot_loader.rs
@@ -1,6 +1,6 @@
-use crate::services::statefuls::snapshot::endec::decoder::byte_decoder::BytesDecoder;
-use crate::services::statefuls::snapshot::endec::decoder::states::DecoderInit;
-use crate::services::statefuls::snapshot::snapshot::Snapshot;
+use crate::domains::saves::endec::decoder::{byte_decoder::BytesDecoder, states::DecoderInit};
+
+use super::snapshot::Snapshot;
 
 pub struct Load;
 
diff --git a/src/domains/ttl/actor.rs b/src/domains/ttl/actor.rs
new file mode 100644
index 00000000..a4dbdd0a
--- /dev/null
+++ b/src/domains/ttl/actor.rs
@@ -0,0 +1,4 @@
+use crate::{domains::caches::cache_manager::CacheManager, make_smart_pointer};
+
+pub struct TtlActor(pub(crate) CacheManager);
+make_smart_pointer!(TtlActor, CacheManager);
diff --git a/src/services/statefuls/cache/ttl/command.rs b/src/domains/ttl/command.rs
similarity index 100%
rename from src/services/statefuls/cache/ttl/command.rs
rename to src/domains/ttl/command.rs
diff --git a/src/services/statefuls/cache/ttl/manager.rs b/src/domains/ttl/manager.rs
similarity index 86%
rename from src/services/statefuls/cache/ttl/manager.rs
rename to src/domains/ttl/manager.rs
index 973fea9e..44af3ea1 100644
--- a/src/services/statefuls/cache/ttl/manager.rs
+++ b/src/domains/ttl/manager.rs
@@ -1,7 +1,9 @@
 use crate::make_smart_pointer;
-use crate::services::statefuls::cache::ttl::command::TtlCommand;
+
 use std::time::SystemTime;
 
+use super::command::TtlCommand;
+
 #[derive(Clone, Debug)]
 pub struct TtlSchedulerManager(pub(crate) tokio::sync::mpsc::Sender<TtlCommand>);
 
diff --git a/src/services/statefuls/cache/ttl/mod.rs b/src/domains/ttl/mod.rs
similarity index 61%
rename from src/services/statefuls/cache/ttl/mod.rs
rename to src/domains/ttl/mod.rs
index c8a13a4b..c05afb46 100644
--- a/src/services/statefuls/cache/ttl/mod.rs
+++ b/src/domains/ttl/mod.rs
@@ -1,3 +1,3 @@
 pub(crate) mod actor;
-mod command;
+pub(crate) mod command;
 pub mod manager;
diff --git a/src/init.rs b/src/init.rs
index 5fe5f69c..50be2643 100644
--- a/src/init.rs
+++ b/src/init.rs
@@ -1,5 +1,4 @@
-use crate::env_var;
-use crate::services::cluster::actors::replication::IS_LEADER_MODE;
+use crate::{domains::cluster_actors::replication::IS_LEADER_MODE, env_var};
 
 pub struct Environment {
     pub replicaof: Option<(String, String)>,
diff --git a/src/lib.rs b/src/lib.rs
index 5118214f..27891421 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,23 +1,24 @@
 mod actor_registry;
-pub mod adapters;
+pub(crate) mod adapters;
+pub mod domains;
 mod init;
 pub mod macros;
 pub mod presentation;
 pub mod services;
 use actor_registry::ActorRegistry;
 use anyhow::Result;
+use domains::caches::cache_manager::CacheManager;
+use domains::cluster_actors::commands::ClusterCommand;
+use domains::cluster_actors::replication::IS_LEADER_MODE;
+use domains::config_actors::config_manager::ConfigManager;
+use domains::saves::snapshot::snapshot_applier::SnapshotApplier;
+use domains::saves::snapshot::snapshot_loader::SnapshotLoader;
+use domains::ttl::actor::TtlActor;
+use domains::IoError;
 pub use init::Environment;
-use presentation::client_in::manager::ClientManager;
+use presentation::clients::ClientController;
 use presentation::cluster_in::communication_manager::ClusterCommunicationManager;
 use presentation::cluster_in::inbound::stream::InboundStream;
-use services::cluster::actors::commands::ClusterCommand;
-use services::cluster::actors::replication::IS_LEADER_MODE;
-use services::config::manager::ConfigManager;
-use services::error::IoError;
-use services::statefuls::cache::manager::CacheManager;
-use services::statefuls::cache::ttl::actor::TtlActor;
-use services::statefuls::snapshot::snapshot_applier::SnapshotApplier;
-use services::statefuls::snapshot::snapshot_loader::SnapshotLoader;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 use tokio::net::TcpListener;
@@ -27,7 +28,7 @@ pub mod client_utils;
 
 // * StartUp Facade that manages invokes subsystems
 pub struct StartUpFacade {
-    client_manager: ClientManager,
+    client_manager: ClientController,
     registry: ActorRegistry,
     mode_change_watcher: tokio::sync::watch::Receiver<bool>,
 }
@@ -62,7 +63,7 @@ impl StartUpFacade {
             ttl_manager,
             snapshot_applier,
         };
-        let client_manager = ClientManager::new(registry.clone());
+        let client_manager = ClientController::new(registry.clone());
 
         StartUpFacade { client_manager, registry, mode_change_watcher }
     }
diff --git a/src/macros/mod.rs b/src/macros.rs
similarity index 100%
rename from src/macros/mod.rs
rename to src/macros.rs
diff --git a/src/main.rs b/src/main.rs
index 8e895072..2d16d482 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,5 +1,5 @@
 use duva::{
-    services::config::{actor::ConfigActor, manager::ConfigManager},
+    domains::{config_actors::actor::ConfigActor, config_actors::config_manager::ConfigManager},
     Environment, StartUpFacade,
 };
 
diff --git a/src/presentation/client_in/mod.rs b/src/presentation/client_in/mod.rs
deleted file mode 100644
index b2e0447f..00000000
--- a/src/presentation/client_in/mod.rs
+++ /dev/null
@@ -1,3 +0,0 @@
-pub mod manager;
-pub mod request;
-pub mod stream;
diff --git a/src/presentation/client_in/manager.rs b/src/presentation/clients/controller.rs
similarity index 92%
rename from src/presentation/client_in/manager.rs
rename to src/presentation/clients/controller.rs
index 33ffa566..debc02df 100644
--- a/src/presentation/client_in/manager.rs
+++ b/src/presentation/clients/controller.rs
@@ -1,31 +1,30 @@
 use super::request::ClientRequest;
 use super::stream::ClientStream;
 use crate::actor_registry::ActorRegistry;
-
+use crate::domains::caches::cache_manager::CacheManager;
+use crate::domains::caches::cache_objects::CacheEntry;
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::config_actors::command::ConfigResponse;
+use crate::domains::config_actors::config_manager::ConfigManager;
+use crate::domains::query_parsers::QueryIO;
+use crate::domains::saves::actor::SaveTarget;
+use crate::domains::ttl::manager::TtlSchedulerManager;
 use crate::presentation::cluster_in::communication_manager::ClusterCommunicationManager;
-use crate::services::cluster::actors::commands::ClusterCommand;
-use crate::services::config::manager::ConfigManager;
-use crate::services::config::ConfigResponse;
 use crate::services::interface::TWrite;
-use crate::services::query_io::QueryIO;
-use crate::services::statefuls::cache::manager::CacheManager;
-use crate::services::statefuls::cache::ttl::manager::TtlSchedulerManager;
-use crate::services::statefuls::cache::CacheEntry;
-use crate::services::statefuls::snapshot::save::actor::SaveTarget;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::select;
 
 #[derive(Clone)]
-pub(crate) struct ClientManager {
+pub(crate) struct ClientController {
     pub(crate) ttl_manager: TtlSchedulerManager,
     pub(crate) cache_manager: CacheManager,
     pub(crate) config_manager: ConfigManager,
     pub(crate) cluster_communication_manager: ClusterCommunicationManager,
 }
 
-impl ClientManager {
+impl ClientController {
     pub(crate) fn new(actor_registry: ActorRegistry) -> Self {
-        ClientManager {
+        ClientController {
             cluster_communication_manager: actor_registry.cluster_communication_manager(),
             ttl_manager: actor_registry.ttl_manager,
             cache_manager: actor_registry.cache_manager,
diff --git a/src/presentation/clients/mod.rs b/src/presentation/clients/mod.rs
new file mode 100644
index 00000000..66f267bf
--- /dev/null
+++ b/src/presentation/clients/mod.rs
@@ -0,0 +1,5 @@
+pub mod controller;
+pub mod request;
+pub mod stream;
+
+pub(crate) use controller::ClientController;
diff --git a/src/presentation/client_in/request.rs b/src/presentation/clients/request.rs
similarity index 92%
rename from src/presentation/client_in/request.rs
rename to src/presentation/clients/request.rs
index 401eae90..8d91d861 100644
--- a/src/presentation/client_in/request.rs
+++ b/src/presentation/clients/request.rs
@@ -1,4 +1,4 @@
-use crate::services::{aof::WriteRequest, cluster::peers::identifier::PeerIdentifier};
+use crate::domains::{append_only_files::WriteRequest, peers::identifier::PeerIdentifier};
 use std::time::SystemTime;
 
 #[derive(Clone, Debug)]
diff --git a/src/presentation/client_in/stream.rs b/src/presentation/clients/stream.rs
similarity index 97%
rename from src/presentation/client_in/stream.rs
rename to src/presentation/clients/stream.rs
index 4ec29b7d..611c872f 100644
--- a/src/presentation/client_in/stream.rs
+++ b/src/presentation/clients/stream.rs
@@ -1,11 +1,7 @@
-use std::time::SystemTime;
-
 use super::request::ClientRequest;
-use crate::{
-    make_smart_pointer,
-    services::{interface::TRead, query_io::QueryIO},
-};
+use crate::{domains::query_parsers::QueryIO, make_smart_pointer, services::interface::TRead};
 use anyhow::Context;
+use std::time::SystemTime;
 use tokio::net::TcpStream;
 
 pub struct ClientStream(pub(crate) TcpStream);
diff --git a/src/presentation/cluster_in/communication_manager.rs b/src/presentation/cluster_in/communication_manager.rs
index f1be64fe..c7cee97d 100644
--- a/src/presentation/cluster_in/communication_manager.rs
+++ b/src/presentation/cluster_in/communication_manager.rs
@@ -1,9 +1,9 @@
 use crate::{
-    make_smart_pointer,
-    services::cluster::{
-        actors::actor::ClusterActor, actors::commands::ClusterCommand,
-        actors::replication::ReplicationInfo, peers::identifier::PeerIdentifier,
+    domains::{
+        cluster_actors::{commands::ClusterCommand, replication::ReplicationInfo, ClusterActor},
+        peers::identifier::PeerIdentifier,
     },
+    make_smart_pointer,
 };
 use std::time::Duration;
 use tokio::{sync::mpsc::Sender, time::interval};
@@ -25,8 +25,8 @@ impl ClusterCommunicationManager {
         let (actor_handler, cluster_message_listener) = tokio::sync::mpsc::channel(100);
 
         tokio::spawn(
-            ClusterActor::new(node_timeout, ReplicationInfo::new(replicaof, &host, port))
-                .handle(cluster_message_listener, notifier),
+            ClusterActor::new(node_timeout, ReplicationInfo::new(replicaof, &host, port), notifier)
+                .handle(cluster_message_listener),
         );
 
         tokio::spawn({
diff --git a/src/presentation/cluster_in/connection_manager.rs b/src/presentation/cluster_in/connection_manager.rs
index c477c698..e73c334c 100644
--- a/src/presentation/cluster_in/connection_manager.rs
+++ b/src/presentation/cluster_in/connection_manager.rs
@@ -1,11 +1,12 @@
 use super::communication_manager::ClusterCommunicationManager;
 use super::outbound::stream::OutboundStream;
-use crate::services::cluster::actors::commands::ClusterCommand;
-use crate::services::cluster::actors::replication::IS_LEADER_MODE;
-use crate::services::cluster::peers::identifier::PeerIdentifier;
-use crate::services::cluster::peers::kind::PeerKind;
-use crate::services::statefuls::cache::manager::CacheManager;
-use crate::services::statefuls::snapshot::snapshot_applier::SnapshotApplier;
+use crate::domains::caches::cache_manager::CacheManager;
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::cluster_actors::replication::IS_LEADER_MODE;
+use crate::domains::peers::identifier::PeerIdentifier;
+use crate::domains::peers::kind::PeerKind;
+use crate::domains::saves::snapshot::snapshot_applier::SnapshotApplier;
+
 use crate::{make_smart_pointer, InboundStream};
 use tokio::sync::mpsc::Sender;
 
diff --git a/src/presentation/cluster_in/inbound/request.rs b/src/presentation/cluster_in/inbound/request.rs
index e1d7432f..90a5e6ed 100644
--- a/src/presentation/cluster_in/inbound/request.rs
+++ b/src/presentation/cluster_in/inbound/request.rs
@@ -1,4 +1,4 @@
-use crate::{from_to, make_smart_pointer, services::query_io::QueryIO};
+use crate::{domains::query_parsers::QueryIO, from_to, make_smart_pointer};
 use anyhow::Context;
 use bytes::Bytes;
 
diff --git a/src/presentation/cluster_in/inbound/stream.rs b/src/presentation/cluster_in/inbound/stream.rs
index 9e255cc1..3215537e 100644
--- a/src/presentation/cluster_in/inbound/stream.rs
+++ b/src/presentation/cluster_in/inbound/stream.rs
@@ -1,17 +1,19 @@
 use super::request::{HandShakeRequest, HandShakeRequestEnum};
+use crate::domains::caches::cache_manager::CacheManager;
+use crate::domains::cluster_actors::commands::AddPeer;
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::peers::identifier::PeerIdentifier;
+use crate::domains::peers::kind::PeerKind;
+use crate::domains::peers::peer::Peer;
+use crate::domains::query_parsers::QueryIO;
+use crate::domains::saves::actor::SaveTarget;
+use crate::domains::saves::snapshot::snapshot_applier::SnapshotApplier;
 use crate::make_smart_pointer;
-use crate::services::cluster::actors::commands::{AddPeer, ClusterCommand};
-use crate::services::cluster::peers::identifier::PeerIdentifier;
-use crate::services::cluster::peers::kind::PeerKind;
 
-use crate::services::cluster::peers::peer::Peer;
 use crate::services::interface::TGetPeerIp;
 use crate::services::interface::TRead;
 use crate::services::interface::TWrite;
-use crate::services::query_io::QueryIO;
-use crate::services::statefuls::cache::manager::CacheManager;
-use crate::services::statefuls::snapshot::save::actor::SaveTarget;
-use crate::services::statefuls::snapshot::snapshot_applier::SnapshotApplier;
+
 use anyhow::Context;
 use bytes::Bytes;
 use tokio::net::TcpStream;
diff --git a/src/presentation/cluster_in/outbound/response.rs b/src/presentation/cluster_in/outbound/response.rs
index c3a83a19..a0450276 100644
--- a/src/presentation/cluster_in/outbound/response.rs
+++ b/src/presentation/cluster_in/outbound/response.rs
@@ -1,6 +1,6 @@
 use anyhow::Context;
 
-use crate::services::query_io::QueryIO;
+use crate::domains::query_parsers::QueryIO;
 
 #[derive(Debug, PartialEq)]
 pub enum ConnectionResponse {
diff --git a/src/presentation/cluster_in/outbound/stream.rs b/src/presentation/cluster_in/outbound/stream.rs
index d0f2a988..e4d31536 100644
--- a/src/presentation/cluster_in/outbound/stream.rs
+++ b/src/presentation/cluster_in/outbound/stream.rs
@@ -1,15 +1,15 @@
 use super::response::ConnectionResponse;
+use crate::domains::cluster_actors::commands::AddPeer;
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::saves::snapshot::snapshot_applier::SnapshotApplier;
 use crate::presentation::cluster_in::connection_manager::ClusterConnectionManager;
 
-use crate::services::cluster::actors::commands::AddPeer;
-use crate::services::cluster::actors::commands::ClusterCommand;
-use crate::services::cluster::peers::identifier::PeerIdentifier;
-use crate::services::cluster::peers::kind::PeerKind;
+use crate::domains::peers::identifier::PeerIdentifier;
+use crate::domains::peers::kind::PeerKind;
 
-use crate::services::cluster::peers::peer::Peer;
+use crate::domains::peers::peer::Peer;
 use crate::services::interface::TRead;
 use crate::services::interface::TWrite;
-use crate::services::statefuls::snapshot::snapshot_applier::SnapshotApplier;
 use crate::{make_smart_pointer, write_array};
 use anyhow::Context;
 use tokio::net::TcpStream;
diff --git a/src/services/cluster/listeners/follower.rs b/src/presentation/listeners/follower.rs
similarity index 87%
rename from src/services/cluster/listeners/follower.rs
rename to src/presentation/listeners/follower.rs
index ca03a1fa..d1326a04 100644
--- a/src/services/cluster/listeners/follower.rs
+++ b/src/presentation/listeners/follower.rs
@@ -1,4 +1,8 @@
-use crate::services::cluster::peers::connected_types::Follower;
+use crate::domains::{
+    cluster_actors::{commands::ClusterCommand, replication::HeartBeatMessage},
+    cluster_listeners::{ClusterListener, ReactorKillSwitch, TListen},
+    peers::connected_types::Follower,
+};
 
 use super::*;
 
diff --git a/src/services/cluster/listeners/leader.rs b/src/presentation/listeners/leader.rs
similarity index 90%
rename from src/services/cluster/listeners/leader.rs
rename to src/presentation/listeners/leader.rs
index 7370cc7f..c69fd1fa 100644
--- a/src/services/cluster/listeners/leader.rs
+++ b/src/presentation/listeners/leader.rs
@@ -1,6 +1,13 @@
 use bytes::Bytes;
 
-use crate::{services::cluster::peers::connected_types::Leader, SnapshotLoader};
+use crate::{
+    domains::{
+        cluster_actors::{commands::ClusterCommand, replication::HeartBeatMessage},
+        cluster_listeners::{ClusterListener, ReactorKillSwitch, TListen},
+        peers::connected_types::Leader,
+    },
+    SnapshotLoader,
+};
 
 use super::*;
 
diff --git a/src/presentation/listeners/mod.rs b/src/presentation/listeners/mod.rs
new file mode 100644
index 00000000..f1ac93f1
--- /dev/null
+++ b/src/presentation/listeners/mod.rs
@@ -0,0 +1,8 @@
+use crate::domains::query_parsers::QueryIO;
+use crate::services::interface::TRead;
+use tokio::net::tcp::OwnedReadHalf;
+use tokio::select;
+
+pub mod follower;
+pub mod leader;
+pub mod non_data_peer;
diff --git a/src/services/cluster/listeners/non_data_peer.rs b/src/presentation/listeners/non_data_peer.rs
similarity index 82%
rename from src/services/cluster/listeners/non_data_peer.rs
rename to src/presentation/listeners/non_data_peer.rs
index c1c55c7c..55cb5fe7 100644
--- a/src/services/cluster/listeners/non_data_peer.rs
+++ b/src/presentation/listeners/non_data_peer.rs
@@ -1,4 +1,7 @@
-use crate::services::cluster::peers::connected_types::NonDataPeer;
+use crate::domains::{
+    cluster_listeners::{ClusterListener, ReactorKillSwitch, TListen},
+    peers::connected_types::NonDataPeer,
+};
 
 use super::*;
 
diff --git a/src/presentation/mod.rs b/src/presentation/mod.rs
index d8908184..5ad66aec 100644
--- a/src/presentation/mod.rs
+++ b/src/presentation/mod.rs
@@ -1,2 +1,3 @@
-pub mod client_in;
+pub mod clients;
 pub mod cluster_in;
+pub mod listeners;
diff --git a/src/services/cluster/actors/mod.rs b/src/services/cluster/actors/mod.rs
deleted file mode 100644
index 96cc8729..00000000
--- a/src/services/cluster/actors/mod.rs
+++ /dev/null
@@ -1,5 +0,0 @@
-pub(crate) mod actor;
-pub mod commands;
-mod consensus;
-pub(crate) mod replication;
-pub mod replid_generator;
diff --git a/src/services/cluster/mod.rs b/src/services/cluster/mod.rs
deleted file mode 100644
index fce710a8..00000000
--- a/src/services/cluster/mod.rs
+++ /dev/null
@@ -1,3 +0,0 @@
-pub(crate) mod actors;
-pub mod listeners;
-pub(crate) mod peers;
diff --git a/src/services/config/actor.rs b/src/services/config/actor.rs
deleted file mode 100644
index c571473a..00000000
--- a/src/services/config/actor.rs
+++ /dev/null
@@ -1,69 +0,0 @@
-use super::command::ConfigMessage;
-use super::command::ConfigResource;
-use super::command::ConfigResponse;
-
-use super::ConfigCommand;
-
-use tokio::sync::mpsc::Receiver;
-
-#[derive(Clone)]
-pub struct ConfigActor {
-    pub(crate) dir: &'static str,
-    pub dbfilename: &'static str,
-}
-
-impl ConfigActor {
-    pub fn new(dir: String, dbfilename: String) -> Self {
-        Self {
-            dir: Box::leak(dir.into_boxed_str()),
-            dbfilename: Box::leak(dbfilename.into_boxed_str()),
-        }
-    }
-    pub fn handle(mut self, mut inbox: Receiver<ConfigMessage>) {
-        tokio::spawn(async move {
-            while let Some(msg) = inbox.recv().await {
-                match msg {
-                    ConfigMessage::Query(query) => match query.resource {
-                        ConfigResource::Dir => {
-                            query.respond_with(ConfigResponse::Dir(self.dir.into()));
-                        }
-                        ConfigResource::DbFileName => {
-                            query.respond_with(ConfigResponse::DbFileName(self.dbfilename.into()));
-                        }
-                        ConfigResource::FilePath => {
-                            query.respond_with(ConfigResponse::FilePath(self.get_filepath()));
-                        }
-                    },
-                    ConfigMessage::Command(config_command) => match config_command {
-                        ConfigCommand::SetDbFileName(new_file_name) => {
-                            self.set_dbfilename(&new_file_name);
-                        }
-                    },
-                }
-            }
-        });
-    }
-
-    pub fn set_dir(&mut self, dir: &str) {
-        unsafe {
-            // Get the pointer to the str
-            let ptr: *const str = self.dir;
-            // Recreate the Box from the pointer and drop it
-            let _reclaimed_box: Box<str> = Box::from_raw(ptr as *mut str);
-            self.dir = Box::leak(dir.into());
-        }
-    }
-    pub fn set_dbfilename(&mut self, dbfilename: &str) {
-        unsafe {
-            // Get the pointer to the str
-            let ptr: *const str = self.dbfilename;
-            // Recreate the Box from the pointer and drop it
-            let _reclaimed_box: Box<str> = Box::from_raw(ptr as *mut str);
-            self.dbfilename = Box::leak(dbfilename.into());
-        }
-    }
-
-    pub fn get_filepath(&self) -> String {
-        format!("{}/{}", self.dir, self.dbfilename)
-    }
-}
diff --git a/src/services/config/mod.rs b/src/services/config/mod.rs
deleted file mode 100644
index 907d0f5e..00000000
--- a/src/services/config/mod.rs
+++ /dev/null
@@ -1,12 +0,0 @@
-pub mod actor;
-mod command;
-
-pub mod macros;
-pub mod manager;
-
-pub use command::ConfigCommand;
-pub use command::ConfigMessage;
-pub use command::ConfigQuery;
-
-pub use command::ConfigResource;
-pub use command::ConfigResponse;
diff --git a/src/services/handlers/cache.rs b/src/services/handlers/cache.rs
new file mode 100644
index 00000000..a05cb5c0
--- /dev/null
+++ b/src/services/handlers/cache.rs
@@ -0,0 +1,94 @@
+use crate::domains::caches::actor::CacheActor;
+use crate::domains::caches::cache_objects::CacheEntry;
+use crate::domains::caches::command::CacheCommand;
+
+use crate::domains::query_parsers::QueryIO;
+use crate::domains::saves::command::SaveCommand;
+use anyhow::Result;
+use tokio::sync::mpsc::Receiver;
+
+impl CacheActor {
+    pub(crate) async fn handle(mut self, mut recv: Receiver<CacheCommand>) -> Result<Self> {
+        while let Some(command) = recv.recv().await {
+            match command {
+                CacheCommand::StopSentinel => break,
+
+                CacheCommand::Set { cache_entry, ttl_sender } => {
+                    let _ = self
+                        .try_send_ttl(cache_entry.key(), cache_entry.expiry(), ttl_sender.clone())
+                        .await;
+                    self.set(cache_entry);
+                }
+                CacheCommand::Get { key, sender } => {
+                    let _ = sender.send(self.get(&key).into());
+                }
+                CacheCommand::Keys { pattern, sender } => {
+                    let ks: Vec<_> = self.keys_stream(pattern).collect();
+
+                    sender
+                        .send(QueryIO::Array(ks))
+                        .map_err(|_| anyhow::anyhow!("Error sending keys"))?;
+                }
+                CacheCommand::Delete(key) => {
+                    self.delete(&key);
+                }
+                CacheCommand::Save { outbox } => {
+                    outbox
+                        .send(SaveCommand::LocalShardSize {
+                            table_size: self.len(),
+                            expiry_size: self.keys_with_expiry(),
+                        })
+                        .await?;
+                    for chunk in self.cache.iter().collect::<Vec<_>>().chunks(10) {
+                        outbox.send(SaveCommand::SaveChunk(CacheEntry::new(chunk))).await?;
+                    }
+                    // finalize the save operation
+                    outbox.send(SaveCommand::StopSentinel).await?;
+                }
+            }
+        }
+        Ok(self)
+    }
+}
+
+#[tokio::test]
+async fn test_set_and_delete_inc_dec_keys_with_expiry() {
+    use crate::domains::caches::actor::CacheDb;
+    use crate::domains::ttl::manager::TtlSchedulerManager;
+    use std::time::{Duration, SystemTime};
+
+    // GIVEN
+    let (tx, rx) = tokio::sync::mpsc::channel(100);
+    let actor = CacheActor { cache: CacheDb::default() };
+
+    let (ttl_tx, mut ttx_rx) = tokio::sync::mpsc::channel(100);
+    let ttl_sender = TtlSchedulerManager(ttl_tx);
+    tokio::spawn(async move { ttx_rx.recv().await });
+
+    // WHEN
+    let handler = tokio::spawn(actor.handle(rx));
+
+    for i in 0..100 {
+        let key = format!("key{}", i);
+        let value = format!("value{}", i);
+        tx.send(CacheCommand::Set {
+            cache_entry: if i & 1 == 0 {
+                CacheEntry::KeyValueExpiry(key, value, SystemTime::now() + Duration::from_secs(10))
+            } else {
+                CacheEntry::KeyValue(key, value)
+            },
+            ttl_sender: ttl_sender.clone(),
+        })
+        .await
+        .unwrap();
+    }
+
+    // key0 is expiry key. deleting the following will decrese the number by 1
+    let delete_key = "key0".to_string();
+    tx.send(CacheCommand::Delete(delete_key)).await.unwrap();
+    tx.send(CacheCommand::StopSentinel).await.unwrap();
+    let actor: CacheActor = handler.await.unwrap().unwrap();
+
+    // THEN
+    assert_eq!(actor.cache.keys_with_expiry, 49);
+}
diff --git a/src/services/handlers/cluster.rs b/src/services/handlers/cluster.rs
new file mode 100644
index 00000000..c22bc9d1
--- /dev/null
+++ b/src/services/handlers/cluster.rs
@@ -0,0 +1,82 @@
+use crate::domains::cluster_actors::commands::ClusterCommand;
+use crate::domains::cluster_actors::consensus::ConsensusTracker;
+use crate::domains::cluster_actors::{ClusterActor, FANOUT};
+
+use tokio::sync::mpsc::Receiver;
+
+impl ClusterActor {
+    pub(crate) async fn handle(
+        mut self,
+        mut cluster_message_listener: Receiver<ClusterCommand>,
+    ) -> anyhow::Result<Self> {
+        let mut consensus_con = ConsensusTracker::default();
+
+        while let Some(command) = cluster_message_listener.recv().await {
+            // TODO notifier will be used when election process is implemented
+
+            match command {
+                ClusterCommand::AddPeer(add_peer_cmd) => {
+                    self.add_peer(add_peer_cmd).await;
+                }
+
+                ClusterCommand::GetPeers(callback) => {
+                    let _ = callback.send(self.members.keys().cloned().collect::<Vec<_>>().into());
+                }
+
+                ClusterCommand::SendHeartBeat => {
+                    let hop_count = self.hop_count(FANOUT, self.members.len());
+                    self.send_liveness_heartbeat(hop_count).await;
+
+                    // ! remove idle peers based on ttl.
+                    // ! The following may need to be moved else where to avoid blocking the main loop
+                    self.remove_idle_peers().await;
+                }
+
+                ClusterCommand::ReplicationInfo(sender) => {
+                    let _ = sender.send(self.replication.clone());
+                }
+                ClusterCommand::SetReplicationInfo { leader_repl_id, offset } => {
+                    self.set_replication_info(leader_repl_id, offset);
+                }
+                ClusterCommand::ReceiveHeartBeat(heartbeat) => {
+                    if self.replication.in_ban_list(&heartbeat.heartbeat_from) {
+                        continue;
+                    }
+
+                    self.gossip(heartbeat.hop_count).await;
+                    self.update_on_report(heartbeat).await;
+                }
+                ClusterCommand::ForgetPeer(peer_addr, sender) => {
+                    if let Ok(Some(())) = self.forget_peer(peer_addr).await {
+                        let _ = sender.send(Some(()));
+                    } else {
+                        let _ = sender.send(None);
+                    }
+                }
+                ClusterCommand::LeaderReqConsensus { log, sender } => {
+                    // ! If there are no replicas, don't send the request
+                    if self.followers().count() == 0 {
+                        let _ = sender.send(None);
+                        continue;
+                    }
+                    self.req_consensus(log).await;
+
+                    consensus_con.add(
+                        self.replication.leader_repl_offset,
+                        sender,
+                        self.followers().count(),
+                    );
+                }
+                ClusterCommand::LeaderReceiveAcks(offsets) => {
+                    self.apply_acks(&mut consensus_con, offsets);
+                }
+                ClusterCommand::FollowerReceiveLogEntries(write_operations) => {
+                    // TODO handle the log entries
+                    println!("[INFO] Received log entries: {:?}", write_operations);
+                    self.receive_log_entries_from_leader(write_operations).await;
+                }
+            }
+        }
+        Ok(self)
+    }
+}
diff --git a/src/services/handlers/config.rs b/src/services/handlers/config.rs
new file mode 100644
index 00000000..54532672
--- /dev/null
+++ b/src/services/handlers/config.rs
@@ -0,0 +1,32 @@
+use crate::domains::config_actors::{
+    actor::ConfigActor,
+    command::{ConfigCommand, ConfigMessage, ConfigResource, ConfigResponse},
+};
+use tokio::sync::mpsc::Receiver;
+
+impl ConfigActor {
+    pub(crate) fn handle(mut self, mut recv: Receiver<ConfigMessage>) {
+        tokio::spawn(async move {
+            while let Some(msg) = recv.recv().await {
+                match msg {
+                    ConfigMessage::Query(query) => match query.resource {
+                        ConfigResource::Dir => {
+                            query.respond_with(ConfigResponse::Dir(self.dir.into()));
+                        }
+                        ConfigResource::DbFileName => {
+                            query.respond_with(ConfigResponse::DbFileName(self.dbfilename.into()));
+                        }
+                        ConfigResource::FilePath => {
+                            query.respond_with(ConfigResponse::FilePath(self.get_filepath()));
+                        }
+                    },
+                    ConfigMessage::Command(config_command) => match config_command {
+                        ConfigCommand::SetDbFileName(new_file_name) => {
+                            self.set_dbfilename(&new_file_name);
+                        }
+                    },
+                }
+            }
+        });
+    }
+}
diff --git a/src/services/handlers/mod.rs b/src/services/handlers/mod.rs
new file mode 100644
index 00000000..c85f288c
--- /dev/null
+++ b/src/services/handlers/mod.rs
@@ -0,0 +1,5 @@
+pub mod cache;
+pub mod cluster;
+pub mod config;
+pub mod save;
+pub mod ttl;
diff --git a/src/services/handlers/save.rs b/src/services/handlers/save.rs
new file mode 100644
index 00000000..f15ebc8b
--- /dev/null
+++ b/src/services/handlers/save.rs
@@ -0,0 +1,22 @@
+use tokio::sync::mpsc::Receiver;
+
+use crate::domains::saves::{actor::SaveActor, command::SaveCommand};
+
+impl SaveActor {
+    pub async fn run(mut self, mut inbox: Receiver<SaveCommand>) -> anyhow::Result<Self> {
+        while let Some(cmd) = inbox.recv().await {
+            match self.handle_cmd(cmd).await {
+                Ok(should_break) => {
+                    if should_break {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    eprintln!("error while encoding: {:?}", err);
+                    return Err(err);
+                }
+            }
+        }
+        Ok(self)
+    }
+}
diff --git a/src/services/statefuls/cache/ttl/actor.rs b/src/services/handlers/ttl.rs
similarity index 66%
rename from src/services/statefuls/cache/ttl/actor.rs
rename to src/services/handlers/ttl.rs
index df0a0ee1..35fa8edd 100644
--- a/src/services/statefuls/cache/ttl/actor.rs
+++ b/src/services/handlers/ttl.rs
@@ -1,29 +1,30 @@
-use crate::make_smart_pointer;
-use crate::services::statefuls::cache::actor::CacheCommand;
-use crate::services::statefuls::cache::manager::CacheManager;
-use crate::services::statefuls::cache::ttl::command::TtlCommand;
-use crate::services::statefuls::cache::ttl::manager::TtlSchedulerManager;
-use std::cmp::Reverse;
-use std::collections::BinaryHeap;
-use std::time::{Duration, SystemTime};
-use tokio::sync::mpsc;
-use tokio::sync::mpsc::Receiver;
-use tokio::time::interval;
+use std::{
+    cmp::Reverse,
+    collections::BinaryHeap,
+    time::{Duration, SystemTime},
+};
 
-pub struct TtlActor(pub(crate) CacheManager);
-make_smart_pointer!(TtlActor, CacheManager);
+use tokio::{sync::mpsc::Receiver, time::interval};
+
+use crate::domains::{
+    caches::command::CacheCommand,
+    ttl::{actor::TtlActor, command::TtlCommand, manager::TtlSchedulerManager},
+};
 
 impl TtlActor {
     pub(crate) fn run(self) -> TtlSchedulerManager {
         let (scheduler_outbox, inbox) = tokio::sync::mpsc::channel(100);
-        tokio::spawn(Self::ttl_schedule_actor(inbox));
+        tokio::spawn(Self::handle(inbox));
         tokio::spawn(self.background_delete_actor(scheduler_outbox.clone()));
 
         TtlSchedulerManager(scheduler_outbox)
     }
 
     // Background actor keeps sending peek command to the scheduler actor to check if there is any key to delete.
-    async fn background_delete_actor(self, outbox: mpsc::Sender<TtlCommand>) -> anyhow::Result<()> {
+    async fn background_delete_actor(
+        self,
+        outbox: tokio::sync::mpsc::Sender<TtlCommand>,
+    ) -> anyhow::Result<()> {
         let mut cleanup_interval = interval(Duration::from_millis(1));
         loop {
             cleanup_interval.tick().await;
@@ -36,11 +37,10 @@ impl TtlActor {
             self.select_shard(&key).send(CacheCommand::Delete(key)).await?;
         }
     }
-
-    async fn ttl_schedule_actor(mut inbox: Receiver<TtlCommand>) -> anyhow::Result<()> {
+    async fn handle(mut recv: Receiver<TtlCommand>) -> anyhow::Result<()> {
         let mut priority_queue: BinaryHeap<(Reverse<SystemTime>, String)> = BinaryHeap::new();
 
-        while let Some(cmd) = inbox.recv().await {
+        while let Some(cmd) = recv.recv().await {
             match cmd {
                 TtlCommand::ScheduleTtl { expiry, key } => {
                     priority_queue.push((Reverse(expiry), key));
diff --git a/src/services/interface.rs b/src/services/interface.rs
index 4f48742b..8a5e3e4f 100644
--- a/src/services/interface.rs
+++ b/src/services/interface.rs
@@ -1,9 +1,6 @@
-use crate::services::error::IoError;
-
+use crate::domains::{query_parsers::QueryIO, IoError};
 use bytes::{Bytes, BytesMut};
 
-use super::query_io::QueryIO;
-
 pub trait TRead {
     fn read_bytes(
         &mut self,
diff --git a/src/services/mod.rs b/src/services/mod.rs
index e1d51c31..33c96a00 100644
--- a/src/services/mod.rs
+++ b/src/services/mod.rs
@@ -1,8 +1,2 @@
-pub mod aof;
-pub mod cluster;
-pub mod config;
-pub mod error;
+pub mod handlers;
 pub mod interface;
-
-pub mod query_io;
-pub mod statefuls;
diff --git a/src/services/statefuls/cache/actor.rs b/src/services/statefuls/cache/actor.rs
deleted file mode 100644
index d76570cc..00000000
--- a/src/services/statefuls/cache/actor.rs
+++ /dev/null
@@ -1,176 +0,0 @@
-use super::CacheEntry;
-use super::CacheValue;
-use crate::make_smart_pointer;
-use crate::services::query_io::QueryIO;
-use crate::services::statefuls::cache::ttl::manager::TtlSchedulerManager;
-use crate::services::statefuls::snapshot::save::command::SaveCommand;
-use anyhow::Context;
-use anyhow::Result;
-use std::collections::HashMap;
-use tokio::sync::{mpsc, oneshot};
-
-pub enum CacheCommand {
-    Set { cache_entry: CacheEntry, ttl_sender: TtlSchedulerManager },
-    Save { outbox: mpsc::Sender<SaveCommand> },
-    Get { key: String, sender: oneshot::Sender<QueryIO> },
-    Keys { pattern: Option<String>, sender: oneshot::Sender<QueryIO> },
-    Delete(String),
-    StopSentinel,
-}
-
-#[derive(Default)]
-pub struct CacheDb {
-    inner: HashMap<String, CacheValue>,
-    // OPTIMIZATION: Add a counter to keep track of the number of keys with expiry
-    pub(crate) keys_with_expiry: usize,
-}
-
-pub struct CacheActor {
-    inbox: mpsc::Receiver<CacheCommand>,
-    cache: CacheDb,
-}
-impl CacheActor {
-    // Create a new CacheActor with inner state
-    pub fn run() -> CacheCommandSender {
-        let (tx, cache_actor_inbox) = mpsc::channel(100);
-        tokio::spawn(Self { inbox: cache_actor_inbox, cache: CacheDb::default() }.handle());
-        CacheCommandSender(tx)
-    }
-
-    async fn handle(mut self) -> Result<Self> {
-        while let Some(command) = self.inbox.recv().await {
-            match command {
-                CacheCommand::StopSentinel => break,
-
-                CacheCommand::Set { cache_entry, ttl_sender } => {
-                    let _ = self
-                        .try_send_ttl(cache_entry.key(), cache_entry.expiry(), ttl_sender.clone())
-                        .await;
-                    self.set(cache_entry);
-                }
-                CacheCommand::Get { key, sender } => {
-                    let _ = sender.send(self.get(&key).into());
-                }
-                CacheCommand::Keys { pattern, sender } => {
-                    let ks: Vec<_> = self.keys_stream(pattern).collect();
-
-                    sender
-                        .send(QueryIO::Array(ks))
-                        .map_err(|_| anyhow::anyhow!("Error sending keys"))?;
-                }
-                CacheCommand::Delete(key) => {
-                    self.delete(&key);
-                }
-                CacheCommand::Save { outbox } => {
-                    outbox
-                        .send(SaveCommand::LocalShardSize {
-                            table_size: self.len(),
-                            expiry_size: self.keys_with_expiry(),
-                        })
-                        .await?;
-                    for chunk in self.cache.iter().collect::<Vec<_>>().chunks(10) {
-                        outbox.send(SaveCommand::SaveChunk(CacheEntry::new(chunk))).await?;
-                    }
-                    // finalize the save operation
-                    outbox.send(SaveCommand::StopSentinel).await?;
-                }
-            }
-        }
-        Ok(self)
-    }
-
-    fn len(&self) -> usize {
-        self.cache.len()
-    }
-    fn keys_with_expiry(&self) -> usize {
-        self.cache.keys_with_expiry
-    }
-
-    fn keys_stream(&self, pattern: Option<String>) -> impl Iterator<Item = QueryIO> + '_ {
-        self.cache.keys().filter_map(move |k| {
-            if pattern.as_ref().map_or(true, |p| k.contains(p)) {
-                Some(QueryIO::BulkString(k.clone().into()))
-            } else {
-                None
-            }
-        })
-    }
-    fn delete(&mut self, key: &str) {
-        if let Some(value) = self.cache.remove(key) {
-            if value.has_expiry() {
-                self.cache.keys_with_expiry -= 1;
-            }
-        }
-    }
-    fn get(&self, key: &str) -> Option<CacheValue> {
-        self.cache.get(key).cloned()
-    }
-
-    fn set(&mut self, cache_entry: CacheEntry) {
-        match cache_entry {
-            CacheEntry::KeyValue(key, value) => {
-                self.cache.insert(key, CacheValue::Value(value));
-            }
-            CacheEntry::KeyValueExpiry(key, value, expiry) => {
-                self.cache.keys_with_expiry += 1;
-                self.cache.insert(key.clone(), CacheValue::ValueWithExpiry(value, expiry));
-            }
-        }
-    }
-
-    async fn try_send_ttl(
-        &self,
-        key: &str,
-        expiry: Option<std::time::SystemTime>,
-        ttl_sender: TtlSchedulerManager,
-    ) -> Result<()> {
-        ttl_sender.set_ttl(key.to_string(), expiry.context("Expiry not found")?).await;
-        Ok(())
-    }
-}
-
-#[derive(Clone, Debug)]
-pub struct CacheCommandSender(mpsc::Sender<CacheCommand>);
-
-make_smart_pointer!(CacheCommandSender, mpsc::Sender<CacheCommand>);
-make_smart_pointer!(CacheDb, HashMap<String, CacheValue> => inner);
-
-#[tokio::test]
-async fn test_set_and_delete_inc_dec_keys_with_expiry() {
-    use std::time::{Duration, SystemTime};
-
-    // GIVEN
-    let (tx, rx) = mpsc::channel(100);
-    let actor = CacheActor { inbox: rx, cache: CacheDb::default() };
-
-    let (ttl_tx, mut ttx_rx) = mpsc::channel(100);
-    let ttl_sender = TtlSchedulerManager(ttl_tx);
-    tokio::spawn(async move { ttx_rx.recv().await });
-
-    // WHEN
-    let handler = tokio::spawn(actor.handle());
-
-    for i in 0..100 {
-        let key = format!("key{}", i);
-        let value = format!("value{}", i);
-        tx.send(CacheCommand::Set {
-            cache_entry: if i & 1 == 0 {
-                CacheEntry::KeyValueExpiry(key, value, SystemTime::now() + Duration::from_secs(10))
-            } else {
-                CacheEntry::KeyValue(key, value)
-            },
-            ttl_sender: ttl_sender.clone(),
-        })
-        .await
-        .unwrap();
-    }
-
-    // key0 is expiry key. deleting the following will decrese the number by 1
-    let delete_key = "key0".to_string();
-    tx.send(CacheCommand::Delete(delete_key)).await.unwrap();
-    tx.send(CacheCommand::StopSentinel).await.unwrap();
-    let actor: CacheActor = handler.await.unwrap().unwrap();
-
-    // THEN
-    assert_eq!(actor.cache.keys_with_expiry, 49);
-}
diff --git a/src/services/statefuls/mod.rs b/src/services/statefuls/mod.rs
deleted file mode 100644
index 88f2f2a5..00000000
--- a/src/services/statefuls/mod.rs
+++ /dev/null
@@ -1,2 +0,0 @@
-pub mod cache;
-pub mod snapshot;
diff --git a/src/services/statefuls/snapshot/save/mod.rs b/src/services/statefuls/snapshot/save/mod.rs
deleted file mode 100644
index 6910f3c4..00000000
--- a/src/services/statefuls/snapshot/save/mod.rs
+++ /dev/null
@@ -1,2 +0,0 @@
-pub(crate) mod actor;
-pub mod command;


Test patch:
diff --git a/tests/common.rs b/tests/common.rs
index a53f6d04..4b6e051f 100644
--- a/tests/common.rs
+++ b/tests/common.rs
@@ -1,6 +1,7 @@
 use bytes::Bytes;
+use duva::domains::query_parsers::query_io::QueryIO;
 use duva::make_smart_pointer;
-use duva::services::query_io::QueryIO;
+
 use std::io::{BufRead, BufReader, Read};
 use std::process::{Child, Command, Stdio};
 use std::thread;
diff --git a/tests/test_keys.rs b/tests/test_keys.rs
index c1e928a8..08b1446a 100644
--- a/tests/test_keys.rs
+++ b/tests/test_keys.rs
@@ -1,6 +1,6 @@
 mod common;
 use common::{array, spawn_server_process};
-use duva::{client_utils::ClientStreamHandler, services::query_io::QueryIO};
+use duva::{client_utils::ClientStreamHandler, domains::query_parsers::query_io::QueryIO};
 
 #[tokio::test]
 async fn test_keys() {
diff --git a/tests/test_save_read_snapshot.rs b/tests/test_save_read_snapshot.rs
index c24c242c..0a166ab9 100644
--- a/tests/test_save_read_snapshot.rs
+++ b/tests/test_save_read_snapshot.rs
@@ -9,7 +9,8 @@ use common::get_available_port;
 
 use common::TestProcessChild;
 use duva::client_utils::ClientStreamHandler;
-use duva::services::query_io::QueryIO;
+use duva::domains::query_parsers::query_io::QueryIO;
+
 use std::process::Command;
 use std::process::Stdio;
 use std::time::SystemTime;
diff --git a/tests/test_set_get.rs b/tests/test_set_get.rs
index c8910afa..acf53ac2 100644
--- a/tests/test_set_get.rs
+++ b/tests/test_set_get.rs
@@ -5,7 +5,7 @@
 mod common;
 use common::{array, spawn_server_process};
 
-use duva::{client_utils::ClientStreamHandler, services::query_io::QueryIO};
+use duva::{client_utils::ClientStreamHandler, domains::query_parsers::query_io::QueryIO};
 
 #[tokio::test]
 async fn test_set_get() {