Implement Entity data Saving/Reading

README.md

@@ -83,10 +83,10 @@ and customizable experience. It prioritizes performance and player enjoyment whi
   - [x] Mobs (W.I.P)
   - [x] Animals (W.I.P)
   - [x] Entity AI (W.I.P)
-  - [ ] Entity Saving
   - [ ] Boss
   - [ ] Villagers
   - [ ] Mobs Inventory
+  - [X] Entity Saving
 - Server
   - [x] Plugins (W.I.P)
   - [x] Query

pumpkin-nbt/src/compound.rs

@@ -112,7 +112,7 @@ impl NbtCompound {
         self.put(name, NbtTag::String(value));
     }
 
-    pub fn put_list(&mut self, name: &str, value: Box<[NbtTag]>) {
+    pub fn put_list(&mut self, name: &str, value: Vec<NbtTag>) {
         self.put(name, NbtTag::List(value));
     }
 
@@ -153,12 +153,10 @@ impl NbtCompound {
 
     #[inline]
     pub fn get(&self, name: &str) -> Option<&NbtTag> {
-        for (key, value) in &self.child_tags {
+        self.child_tags
-            if key.as_str() == name {
+            .iter()
-                return Some(value);
+            .find(|k| k.0.as_str() == name)
-            }
+            .map(|r| &r.1)
-        }
-        None
     }
 
     pub fn get_short(&self, name: &str) -> Option<i16> {

pumpkin-nbt/src/deserializer.rs

@@ -1,5 +1,8 @@
+use std::vec::IntoIter;
+
 use crate::*;
 use io::Read;
+use serde::de::value::SeqDeserializer;
 use serde::de::{self, DeserializeSeed, IntoDeserializer, MapAccess, SeqAccess, Visitor};
 use serde::{Deserialize, forward_to_deserialize_any};
 
@@ -113,8 +116,6 @@ pub struct Deserializer<R: Read> {
     // Yes, this breaks with recursion. Just an attempt at a sanity check
     in_list: bool,
     is_named: bool,
-    // For debugging
-    key_stack: Vec<String>,
 }
 
 impl<R: Read> Deserializer<R> {
@@ -124,7 +125,6 @@ impl<R: Read> Deserializer<R> {
             tag_to_deserialize_stack: Vec::new(),
             in_list: false,
             is_named,
-            key_stack: Vec::new(),
         }
     }
 }
@@ -181,28 +181,21 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
             END_ID => Err(Error::SerdeError(
                 "Trying to deserialize an END tag!".to_string(),
             )),
-            LIST_ID | INT_ARRAY_ID | LONG_ARRAY_ID | BYTE_ARRAY_ID => {
+            LIST_ID => {
-                let list_type = match tag_to_deserialize {
+                let list_type = self.input.get_u8_be()?;
-                    LIST_ID => self.input.get_u8_be()?,
-                    INT_ARRAY_ID => INT_ID,
-                    LONG_ARRAY_ID => LONG_ID,
-                    BYTE_ARRAY_ID => BYTE_ID,
-                    _ => unreachable!(),
-                };
 
                 let remaining_values = self.input.get_i32_be()?;
                 if remaining_values < 0 {
                     return Err(Error::NegativeLength(remaining_values));
                 }
 
-                let result = visitor.visit_seq(ListAccess {
+                visitor.visit_seq(ListAccess {
                     de: self,
                     list_type,
                     remaining_values: remaining_values as usize,
-                })?;
+                })
-                Ok(result)
             }
-            COMPOUND_ID => visitor.visit_map(CompoundAccess { de: self }),
+            COMPOUND_ID => self.deserialize_map(visitor),
             _ => {
                 let result = match NbtTag::deserialize_data(&mut self.input, tag_to_deserialize)? {
                     NbtTag::Byte(value) => visitor.visit_i8::<Error>(value)?,
@@ -212,6 +205,22 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
                     NbtTag::Float(value) => visitor.visit_f32::<Error>(value)?,
                     NbtTag::Double(value) => visitor.visit_f64::<Error>(value)?,
                     NbtTag::String(value) => visitor.visit_string::<Error>(value)?,
+                    NbtTag::LongArray(value) => visitor
+                        .visit_seq::<SeqDeserializer<IntoIter<i64>, Error>>(
+                            value.into_deserializer(),
+                        )?,
+                    NbtTag::IntArray(value) => visitor
+                        .visit_seq::<SeqDeserializer<IntoIter<i32>, Error>>(
+                            value.into_deserializer(),
+                        )?,
+                    NbtTag::ByteArray(value) => {
+                        // For compatibility, we serialize byte arrays as Vec<i8>
+                        // It could be probably changed in the future
+                        let array: Vec<_> = value.iter().map(|&byte| byte as i8).collect();
+                        visitor.visit_seq::<SeqDeserializer<IntoIter<i8>, Error>>(
+                            array.into_deserializer(),
+                        )?
+                    }
                     _ => unreachable!(),
                 };
                 Ok(result)
@@ -268,22 +277,12 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
             if *tag_id == BYTE_ID {
                 let value = self.input.get_u8_be()?;
                 if value != 0 {
-                    visitor.visit_bool(true)
+                    return visitor.visit_bool(true);
-                } else {
+                }
+            }
+        }
         visitor.visit_bool(false)
     }
-            } else {
-                Err(Error::UnsupportedType(format!(
-                    "Non-byte bool (found type {tag_id})"
-                )))
-            }
-        } else {
-            Err(Error::SerdeError(
-                "Wanted to deserialize a bool, but there was no type hint on the stack!"
-                    .to_string(),
-            ))
-        }
-    }
 
     fn deserialize_enum<V>(
         self,
@@ -313,11 +312,7 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
         if let Some(tag_id) = self.tag_to_deserialize_stack.pop() {
             if tag_id != COMPOUND_ID {
                 return Err(Error::SerdeError(format!(
-                    "Trying to deserialize a map without a compound ID ({} with id {})",
+                    "Trying to deserialize a map without a compound ID (with id {})",
-                    self.key_stack
-                        .last()
-                        .cloned()
-                        .unwrap_or_else(|| "compound root".to_string()),
                     tag_id
                 )));
             }
@@ -387,9 +382,7 @@ impl<'de, R: Read> MapAccess<'de> for CompoundAccess<'_, R> {
     where
         V: DeserializeSeed<'de>,
     {
-        let result = seed.deserialize(&mut *self.de);
+        seed.deserialize(&mut *self.de)
-        self.de.key_stack.pop();
-        result
     }
 }
 
@@ -405,7 +398,6 @@ impl<'de, R: Read> de::Deserializer<'de> for MapKey<'_, R> {
         V: de::Visitor<'de>,
     {
         let key = get_nbt_string(&mut self.de.input)?;
-        self.de.key_stack.push(key.clone());
         visitor.visit_string(key)
     }
 

pumpkin-nbt/src/serializer.rs

@@ -370,7 +370,9 @@ impl<W: Write> ser::Serializer for &mut Serializer<W> {
         } else {
             return Err(Error::UnsupportedType("newtype variant".to_string()));
         }
-        value.serialize(self)
+        value.serialize(self)?;
+
+        Ok(())
     }
 
     fn serialize_seq(self, len: Option<usize>) -> Result<Self::SerializeSeq> {

pumpkin-nbt/src/tag.rs

@@ -18,10 +18,10 @@ pub enum NbtTag {
     Double(f64) = DOUBLE_ID,
     ByteArray(Box<[u8]>) = BYTE_ARRAY_ID,
     String(String) = STRING_ID,
-    List(Box<[NbtTag]>) = LIST_ID,
+    List(Vec<NbtTag>) = LIST_ID,
     Compound(NbtCompound) = COMPOUND_ID,
-    IntArray(Box<[i32]>) = INT_ARRAY_ID,
+    IntArray(Vec<i32>) = INT_ARRAY_ID,
-    LongArray(Box<[i64]>) = LONG_ARRAY_ID,
+    LongArray(Vec<i64>) = LONG_ARRAY_ID,
 }
 
 impl NbtTag {
@@ -232,7 +232,7 @@ impl NbtTag {
                     assert_eq!(tag.get_type_id(), tag_type_id);
                     list.push(tag);
                 }
-                Ok(NbtTag::List(list.into_boxed_slice()))
+                Ok(NbtTag::List(list))
             }
             COMPOUND_ID => Ok(NbtTag::Compound(NbtCompound::deserialize_content(reader)?)),
             INT_ARRAY_ID => {
@@ -247,7 +247,7 @@ impl NbtTag {
                     let int = reader.get_i32_be()?;
                     int_array.push(int);
                 }
-                Ok(NbtTag::IntArray(int_array.into_boxed_slice()))
+                Ok(NbtTag::IntArray(int_array))
             }
             LONG_ARRAY_ID => {
                 let len = reader.get_i32_be()?;
@@ -261,7 +261,7 @@ impl NbtTag {
                     let long = reader.get_i64_be()?;
                     long_array.push(long);
                 }
-                Ok(NbtTag::LongArray(long_array.into_boxed_slice()))
+                Ok(NbtTag::LongArray(long_array))
             }
             _ => Err(Error::UnknownTagId(tag_id)),
         }
@@ -500,7 +500,7 @@ impl<'de> Deserialize<'de> for NbtTag {
                 while let Some(value) = seq.next_element()? {
                     vec.push(value);
                 }
-                Ok(NbtTag::List(vec.into_boxed_slice()))
+                Ok(NbtTag::List(vec))
             }
 
             fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>

pumpkin-world/src/chunk/format/anvil.rs

@@ -904,6 +904,7 @@ mod tests {
                 &LevelFolder {
                     root_folder: PathBuf::from(""),
                     region_folder: region_path,
+                    entities_folder: PathBuf::from(""),
                 },
                 &[Vector2::new(0, 0)],
                 send,
@@ -932,6 +933,7 @@ mod tests {
         let level_folder = LevelFolder {
             root_folder: temp_dir.path().to_path_buf(),
             region_folder: temp_dir.path().join("region"),
+            entities_folder: PathBuf::from("entities"),
         };
         fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder");
         let chunk_saver = ChunkFileManager::<AnvilChunkFile<ChunkData>>::default();
@@ -1209,6 +1211,7 @@ mod tests {
         let level_folder = LevelFolder {
             root_folder: temp_dir.path().to_path_buf(),
             region_folder: temp_dir.path().join("region"),
+            entities_folder: PathBuf::from("entities"),
         };
         fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder");
         let chunk_saver = ChunkFileManager::<AnvilChunkFile<ChunkData>>::default();

pumpkin-world/src/chunk/format/linear.rs

@@ -5,7 +5,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use crate::chunk::format::anvil::{AnvilChunkFile, SingleChunkDataSerializer};
 use crate::chunk::io::{ChunkSerializer, LoadedData};
-use crate::chunk::{ChunkData, ChunkReadingError, ChunkWritingError};
+use crate::chunk::{ChunkReadingError, ChunkWritingError};
 use async_trait::async_trait;
 use bytes::{Buf, BufMut, Bytes};
 use log::error;
@@ -164,7 +164,7 @@ impl<S: SingleChunkDataSerializer> Default for LinearFile<S> {
 
 #[async_trait]
 impl<S: SingleChunkDataSerializer> ChunkSerializer for LinearFile<S> {
-    type Data = ChunkData;
+    type Data = S;
     type WriteBackend = PathBuf;
 
     fn should_write(&self, is_watched: bool) -> bool {
@@ -320,7 +320,7 @@ impl<S: SingleChunkDataSerializer> ChunkSerializer for LinearFile<S> {
         })
     }
 
-    async fn update_chunk(&mut self, chunk: &ChunkData) -> Result<(), ChunkWritingError> {
+    async fn update_chunk(&mut self, chunk: &Self::Data) -> Result<(), ChunkWritingError> {
         let index = LinearFile::<S>::get_chunk_index(chunk.position());
         let chunk_raw: Bytes = chunk
             .to_bytes()
@@ -343,7 +343,7 @@ impl<S: SingleChunkDataSerializer> ChunkSerializer for LinearFile<S> {
     async fn get_chunks(
         &self,
         chunks: &[Vector2<i32>],
-        stream: tokio::sync::mpsc::Sender<LoadedData<ChunkData, ChunkReadingError>>,
+        stream: tokio::sync::mpsc::Sender<LoadedData<Self::Data, ChunkReadingError>>,
     ) {
         // Don't par iter here so we can prevent backpressure with the await in the async
         // runtime
@@ -352,9 +352,7 @@ impl<S: SingleChunkDataSerializer> ChunkSerializer for LinearFile<S> {
             let linear_chunk_data = &self.chunks_data[index];
 
             let result = if let Some(data) = linear_chunk_data {
-                match ChunkData::internal_from_bytes(data, chunk)
+                match S::from_bytes(data.clone(), chunk) {
-                    .map_err(ChunkReadingError::ParsingError)
-                {
                     Ok(chunk) => LoadedData::Loaded(chunk),
                     Err(err) => LoadedData::Error((chunk, err)),
                 }
@@ -420,6 +418,7 @@ mod tests {
                 &LevelFolder {
                     root_folder: PathBuf::from(""),
                     region_folder: region_path,
+                    entities_folder: PathBuf::from(""),
                 },
                 &[Vector2::new(0, 0)],
                 send,
@@ -443,6 +442,7 @@ mod tests {
         let level_folder = LevelFolder {
             root_folder: temp_dir.path().to_path_buf(),
             region_folder: temp_dir.path().join("region"),
+            entities_folder: PathBuf::from("entities"),
         };
         fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder");
         let chunk_saver = ChunkFileManager::<LinearFile<ChunkData>>::default();

pumpkin-world/src/chunk/format/mod.rs

@@ -5,11 +5,12 @@ use bytes::Bytes;
 use futures::future::join_all;
 use pumpkin_data::{Block, chunk::ChunkStatus};
 use pumpkin_nbt::{compound::NbtCompound, from_bytes, nbt_long_array};
+use uuid::Uuid;
 
 use crate::{
     block::entities::block_entity_from_nbt,
     chunk::{
-        ChunkReadingError, ChunkSerializingError,
+        ChunkEntityData, ChunkReadingError, ChunkSerializingError,
         format::anvil::{SingleChunkDataSerializer, WORLD_DATA_VERSION},
         io::{Dirtiable, file_manager::PathFromLevelFolder},
     },
@@ -316,6 +317,103 @@ impl ChunkData {
     }
 }
 
+impl PathFromLevelFolder for ChunkEntityData {
+    #[inline]
+    fn file_path(folder: &LevelFolder, file_name: &str) -> PathBuf {
+        folder.entities_folder.join(file_name)
+    }
+}
+
+impl Dirtiable for ChunkEntityData {
+    #[inline]
+    fn mark_dirty(&mut self, flag: bool) {
+        self.dirty = flag;
+    }
+
+    #[inline]
+    fn is_dirty(&self) -> bool {
+        self.dirty
+    }
+}
+
+#[async_trait]
+impl SingleChunkDataSerializer for ChunkEntityData {
+    #[inline]
+    fn from_bytes(bytes: Bytes, pos: Vector2<i32>) -> Result<Self, ChunkReadingError> {
+        Self::internal_from_bytes(&bytes, pos).map_err(ChunkReadingError::ParsingError)
+    }
+
+    #[inline]
+    async fn to_bytes(&self) -> Result<Bytes, ChunkSerializingError> {
+        self.internal_to_bytes()
+    }
+
+    #[inline]
+    fn position(&self) -> &Vector2<i32> {
+        &self.chunk_position
+    }
+}
+
+impl ChunkEntityData {
+    fn internal_from_bytes(
+        chunk_data: &[u8],
+        position: Vector2<i32>,
+    ) -> Result<Self, ChunkParsingError> {
+        let chunk_entity_data = pumpkin_nbt::from_bytes::<EntityNbt>(chunk_data)
+            .map_err(|e| ChunkParsingError::ErrorDeserializingChunk(e.to_string()))?;
+
+        if chunk_entity_data.position[0] != position.x
+            || chunk_entity_data.position[1] != position.z
+        {
+            return Err(ChunkParsingError::ErrorDeserializingChunk(format!(
+                "Expected data for entity chunk {},{} but got it for {},{}!",
+                position.x,
+                position.z,
+                chunk_entity_data.position[0],
+                chunk_entity_data.position[1],
+            )));
+        }
+        let mut map = HashMap::new();
+        for entity_nbt in chunk_entity_data.entities {
+            // TODO: This is wrong, we should use an int array, but our NBT lib for some reason does not work with int arrays and
+            // Just gives me a list when putting in a int array
+            let uuid = match entity_nbt.get_list("UUID") {
+                Some(uuid) => uuid,
+                None => {
+                    log::warn!("TODO: use int arrays for UUID");
+                    continue;
+                }
+            };
+            let uuid = Uuid::from_u128(
+                (uuid.first().unwrap().extract_int().unwrap() as u128) << 96
+                    | (uuid.get(1).unwrap().extract_int().unwrap() as u128) << 64
+                    | (uuid.get(2).unwrap().extract_int().unwrap() as u128) << 32
+                    | (uuid.get(3).unwrap().extract_int().unwrap() as u128),
+            );
+            map.insert(uuid, entity_nbt);
+        }
+
+        Ok(ChunkEntityData {
+            chunk_position: position,
+            data: map,
+            dirty: false,
+        })
+    }
+
+    fn internal_to_bytes(&self) -> Result<Bytes, ChunkSerializingError> {
+        let nbt = EntityNbt {
+            data_version: WORLD_DATA_VERSION,
+            position: [self.chunk_position.x, self.chunk_position.z],
+            entities: self.data.values().cloned().collect(),
+        };
+
+        let mut result = Vec::new();
+        pumpkin_nbt::to_bytes(&nbt, &mut result)
+            .map_err(ChunkSerializingError::ErrorSerializingChunk)?;
+        Ok(result.into())
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 struct ChunkSectionNBT {
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -471,3 +569,11 @@ struct ChunkNbt {
     #[serde(rename = "isLightOn")]
     light_correct: bool,
 }
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(rename_all = "PascalCase")]
+struct EntityNbt {
+    data_version: i32,
+    position: [i32; 2],
+    entities: Vec<NbtCompound>,
+}

pumpkin-world/src/chunk/mod.rs

@@ -122,6 +122,13 @@ pub struct ChunkData {
     pub dirty: bool,
 }
 
+pub struct ChunkEntityData {
+    pub chunk_position: Vector2<i32>,
+    pub data: HashMap<uuid::Uuid, NbtCompound>,
+
+    pub dirty: bool,
+}
+
 /// Represents pure block data for a chunk.
 /// Subchunks are vertical portions of a chunk. They are 16 blocks tall.
 /// There are currently 24 subchunks per chunk.

pumpkin-world/src/inventory/inventory.rs

@@ -53,7 +53,7 @@ pub trait Inventory: Send + Sync + Debug + Clearable {
             return;
         }
 
-        nbt.put("Items", NbtTag::List(slots.into_boxed_slice()));
+        nbt.put("Items", NbtTag::List(slots));
     }
 
     fn read_data(

pumpkin-world/src/level.rs

@@ -5,7 +5,7 @@ use pumpkin_config::{advanced_config, chunk::ChunkFormat};
 use pumpkin_data::Block;
 use pumpkin_util::math::{position::BlockPos, vector2::Vector2};
 use std::{
-    collections::VecDeque,
+    collections::{HashMap, VecDeque},
     path::PathBuf,
     sync::{
         Arc,
@@ -26,9 +26,10 @@ use crate::{
     BlockStateId,
     block::RawBlockState,
     chunk::{
-        ChunkData, ChunkParsingError, ChunkReadingError, ScheduledTick, TickPriority,
+        ChunkData, ChunkEntityData, ChunkParsingError, ChunkReadingError, ScheduledTick,
+        TickPriority,
         format::{anvil::AnvilChunkFile, linear::LinearFile},
-        io::{FileIO, LoadedData, file_manager::ChunkFileManager},
+        io::{Dirtiable, FileIO, LoadedData, file_manager::ChunkFileManager},
     },
     dimension::Dimension,
     generation::{Seed, get_world_gen, implementation::WorldGenerator},
@@ -36,6 +37,7 @@ use crate::{
 };
 
 pub type SyncChunk = Arc<RwLock<ChunkData>>;
+pub type SyncEntityChunk = Arc<RwLock<ChunkEntityData>>;
 
 /// The `Level` module provides functionality for working with chunks within or outside a Minecraft world.
 ///
@@ -57,9 +59,13 @@ pub struct Level {
     // Chunks that are paired with chunk watchers. When a chunk is no longer watched, it is removed
     // from the loaded chunks map and sent to the underlying ChunkIO
     loaded_chunks: Arc<DashMap<Vector2<i32>, SyncChunk>>,
+    loaded_entity_chunks: Arc<DashMap<Vector2<i32>, SyncEntityChunk>>,
+
     chunk_watchers: Arc<DashMap<Vector2<i32>, usize>>,
 
     chunk_saver: Arc<dyn FileIO<Data = SyncChunk>>,
+    entity_saver: Arc<dyn FileIO<Data = SyncEntityChunk>>,
+
     world_gen: Arc<dyn WorldGenerator>,
 
     block_ticks: Arc<Mutex<Vec<ScheduledTick>>>,
@@ -75,6 +81,7 @@ pub struct Level {
 pub struct LevelFolder {
     pub root_folder: PathBuf,
     pub region_folder: PathBuf,
+    pub entities_folder: PathBuf,
 }
 
 impl Level {
@@ -89,9 +96,14 @@ impl Level {
         if !region_folder.exists() {
             std::fs::create_dir_all(&region_folder).expect("Failed to create Region folder");
         }
+        let entities_folder = root_folder.join("entities");
+        if !entities_folder.exists() {
+            std::fs::create_dir_all(&region_folder).expect("Failed to create Entities folder");
+        }
         let level_folder = LevelFolder {
             root_folder,
             region_folder,
+            entities_folder,
         };
 
         // TODO: Load info correctly based on world format type
@@ -105,6 +117,15 @@ impl Level {
                 Arc::new(ChunkFileManager::<AnvilChunkFile<ChunkData>>::default())
             }
         };
+        let entity_saver: Arc<dyn FileIO<Data = SyncEntityChunk>> =
+            match advanced_config().chunk.format {
+                ChunkFormat::Linear => {
+                    Arc::new(ChunkFileManager::<LinearFile<ChunkEntityData>>::default())
+                }
+                ChunkFormat::Anvil => {
+                    Arc::new(ChunkFileManager::<AnvilChunkFile<ChunkEntityData>>::default())
+                }
+            };
 
         Self {
             seed,
@@ -112,8 +133,10 @@ impl Level {
             world_gen,
             level_folder,
             chunk_saver,
+            entity_saver,
             spawn_chunks: Arc::new(DashMap::new()),
             loaded_chunks: Arc::new(DashMap::new()),
+            loaded_entity_chunks: Arc::new(DashMap::new()),
             chunk_watchers: Arc::new(DashMap::new()),
             tasks: TaskTracker::new(),
             shutdown_notifier: Notify::new(),
@@ -140,7 +163,7 @@ impl Level {
         self.tasks.close();
         log::debug!("Awaiting level tasks");
         self.tasks.wait().await;
-        log::debug!("Done awaiting level tasks");
+        log::debug!("Done awaiting level chunk tasks");
 
         // wait for chunks currently saving in other threads
         self.chunk_saver.block_and_await_ongoing_tasks().await;
@@ -156,6 +179,23 @@ impl Level {
         // TODO: I think the chunk_saver should be at the server level
         self.chunk_saver.clear_watched_chunks().await;
         self.write_chunks(chunks_to_write).await;
+
+        log::debug!("Done awaiting level entity tasks");
+
+        // wait for chunks currently saving in other threads
+        self.entity_saver.block_and_await_ongoing_tasks().await;
+
+        // save all chunks currently in memory
+        let chunks_to_write = self
+            .loaded_entity_chunks
+            .iter()
+            .map(|chunk| (*chunk.key(), chunk.value().clone()))
+            .collect::<Vec<_>>();
+        self.loaded_entity_chunks.clear();
+
+        // TODO: I think the chunk_saver should be at the server level
+        self.entity_saver.clear_watched_chunks().await;
+        self.write_entity_chunks(chunks_to_write).await;
     }
 
     pub fn loaded_chunk_count(&self) -> usize {
@@ -164,6 +204,7 @@ impl Level {
 
     pub async fn clean_up_log(&self) {
         self.chunk_saver.clean_up_log().await;
+        self.entity_saver.clean_up_log().await;
     }
 
     pub fn list_cached(&self) {
@@ -197,6 +238,9 @@ impl Level {
         self.chunk_saver
             .watch_chunks(&self.level_folder, chunks)
             .await;
+        self.entity_saver
+            .watch_chunks(&self.level_folder, chunks)
+            .await;
     }
 
     #[inline]
@@ -233,6 +277,9 @@ impl Level {
         self.chunk_saver
             .unwatch_chunks(&self.level_folder, chunks)
             .await;
+        self.entity_saver
+            .unwatch_chunks(&self.level_folder, chunks)
+            .await;
         chunks_to_clean
     }
 
@@ -286,6 +333,50 @@ impl Level {
         });
     }
 
+    pub async fn clean_entity_chunks(self: &Arc<Self>, chunks: &[Vector2<i32>]) {
+        // Care needs to be take here because of interweaving case:
+        // 1) Remove chunk from cache
+        // 2) Another player wants same chunk
+        // 3) Load (old) chunk from serializer
+        // 4) Write (new) chunk from serializer
+        // Now outdated chunk data is cached and will be written later
+
+        let chunks_with_no_watchers = chunks
+            .iter()
+            .filter_map(|pos| {
+                // Only chunks that have no entry in the watcher map or have 0 watchers
+                if self
+                    .chunk_watchers
+                    .get(pos)
+                    .is_none_or(|count| count.is_zero())
+                {
+                    self.loaded_entity_chunks
+                        .get(pos)
+                        .map(|chunk| (*pos, chunk.value().clone()))
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<_>>();
+
+        let level = self.clone();
+        self.spawn_task(async move {
+            let chunks_to_remove = chunks_with_no_watchers.clone();
+            level.write_entity_chunks(chunks_with_no_watchers).await;
+            // Only after we have written the chunks to the serializer do we remove them from the
+            // cache
+            for (pos, _) in chunks_to_remove {
+                let _ = level.loaded_entity_chunks.remove_if(&pos, |_, _| {
+                    // Recheck that there is no one watching
+                    level
+                        .chunk_watchers
+                        .get(&pos)
+                        .is_none_or(|count| count.is_zero())
+                });
+            }
+        });
+    }
+
     pub async fn tick_block_entities(&self, world: Arc<dyn SimpleWorld>) {
         for chunk in self.loaded_chunks.iter() {
             let chunk = chunk.read().await;
@@ -301,6 +392,10 @@ impl Level {
         self.clean_chunks(&[*chunk]).await;
     }
 
+    pub async fn clean_entity_chunk(self: &Arc<Self>, chunk: &Vector2<i32>) {
+        self.clean_entity_chunks(&[*chunk]).await;
+    }
+
     pub fn is_chunk_watched(&self, chunk: &Vector2<i32>) -> bool {
         self.chunk_watchers.get(chunk).is_some()
     }
@@ -309,6 +404,8 @@ impl Level {
         self.chunk_watchers.retain(|_, watcher| !watcher.is_zero());
         self.loaded_chunks
             .retain(|at, _| self.chunk_watchers.get(at).is_some());
+        self.loaded_entity_chunks
+            .retain(|at, _| self.chunk_watchers.get(at).is_some());
 
         // if the difference is too big, we can shrink the loaded chunks
         // (1024 chunks is the equivalent to a 32x32 chunks area)
@@ -321,6 +418,10 @@ impl Level {
         if self.loaded_chunks.capacity() - self.loaded_chunks.len() >= 4096 {
             self.loaded_chunks.shrink_to_fit();
         }
+
+        if self.loaded_entity_chunks.capacity() - self.loaded_entity_chunks.len() >= 4096 {
+            self.loaded_entity_chunks.shrink_to_fit();
+        }
     }
 
     // Stream the chunks (don't collect them and then do stuff with them)
@@ -348,6 +449,27 @@ impl Level {
         receiver
     }
 
+    pub fn receive_entity_chunks(
+        self: &Arc<Self>,
+        chunks: Vec<Vector2<i32>>,
+    ) -> UnboundedReceiver<(SyncEntityChunk, bool)> {
+        let (sender, receiver) = mpsc::unbounded_channel();
+        // Put this in another thread so we aren't blocking on it
+        let level = self.clone();
+        self.spawn_task(async move {
+            let cancel_notifier = level.shutdown_notifier.notified();
+            let fetch_task = level.fetch_entity_chunks(&chunks, sender);
+
+            // Don't continue to handle chunks if we are shutting down
+            select! {
+                () = cancel_notifier => {},
+                () = fetch_task => {}
+            };
+        });
+
+        receiver
+    }
+
     pub async fn get_chunk(
         self: &Arc<Self>,
         chunk_coordinate: Vector2<i32>,
@@ -358,6 +480,16 @@ impl Level {
         }
     }
 
+    pub async fn get_entity_chunk(
+        self: &Arc<Self>,
+        chunk_coordinate: Vector2<i32>,
+    ) -> Arc<RwLock<ChunkEntityData>> {
+        match self.try_get_entity_chunk(chunk_coordinate) {
+            Some(chunk) => chunk.clone(),
+            None => self.receive_entity_chunk(chunk_coordinate).await.0,
+        }
+    }
+
     pub async fn receive_chunk(
         self: &Arc<Self>,
         chunk_pos: Vector2<i32>,
@@ -370,6 +502,18 @@ impl Level {
             .expect("Channel closed for unknown reason")
     }
 
+    pub async fn receive_entity_chunk(
+        self: &Arc<Self>,
+        chunk_pos: Vector2<i32>,
+    ) -> (Arc<RwLock<ChunkEntityData>>, bool) {
+        let mut receiver = self.receive_entity_chunks(vec![chunk_pos]);
+
+        receiver
+            .recv()
+            .await
+            .expect("Channel closed for unknown reason")
+    }
+
     pub async fn get_block_state(self: &Arc<Self>, position: &BlockPos) -> RawBlockState {
         let (chunk_coordinate, relative) = position.chunk_and_chunk_relative_position();
         let chunk = self.get_chunk(chunk_coordinate).await;
@@ -404,7 +548,7 @@ impl Level {
             return block_state_id;
         }
 
-        chunk.dirty = true;
+        chunk.mark_dirty(true);
 
         chunk.section.set_block_absolute_y(
             relative.x as usize,
@@ -463,6 +607,23 @@ impl Level {
         }
     }
 
+    pub async fn write_entity_chunks(&self, chunks_to_write: Vec<(Vector2<i32>, SyncEntityChunk)>) {
+        if chunks_to_write.is_empty() {
+            return;
+        }
+
+        let chunk_saver = self.entity_saver.clone();
+        let level_folder = self.level_folder.clone();
+
+        trace!("Sending chunks to ChunkIO {:}", chunks_to_write.len());
+        if let Err(error) = chunk_saver
+            .save_chunks(&level_folder, chunks_to_write)
+            .await
+        {
+            log::error!("Failed writing Chunk to disk {error}");
+        }
+    }
+
     /// Initializes the spawn chunks to these chunks
     pub async fn read_spawn_chunks(self: &Arc<Self>, chunks: &[Vector2<i32>]) {
         let (send, mut recv) = mpsc::unbounded_channel();
@@ -647,6 +808,149 @@ impl Level {
         tracker.wait().await;
     }
 
+    pub async fn fetch_entity_chunks(
+        self: &Arc<Self>,
+        chunks: &[Vector2<i32>],
+        channel: mpsc::UnboundedSender<(SyncEntityChunk, bool)>,
+    ) {
+        if chunks.is_empty() {
+            return;
+        }
+
+        // If false, stop loading chunks because the channel has closed.
+        let send_chunk =
+            move |is_new: bool,
+                  chunk: SyncEntityChunk,
+                  channel: &mpsc::UnboundedSender<(SyncEntityChunk, bool)>| {
+                channel.send((chunk, is_new)).is_ok()
+            };
+
+        // First send all chunks that we have cached
+        // We expect best case scenario to have all cached
+        let mut remaining_chunks = Vec::new();
+        for chunk in chunks {
+            let is_ok = if let Some(chunk) = self.loaded_entity_chunks.get(chunk) {
+                send_chunk(false, chunk.value().clone(), &channel)
+            } else {
+                remaining_chunks.push(*chunk);
+                true
+            };
+
+            if !is_ok {
+                return;
+            }
+        }
+
+        if remaining_chunks.is_empty() {
+            return;
+        }
+
+        // These just pass data between async tasks, each of which do not block on anything, so
+        // these do not need to hold a lot
+        let (load_bridge_send, mut load_bridge_recv) =
+            tokio::sync::mpsc::channel::<LoadedData<SyncEntityChunk, ChunkReadingError>>(16);
+        let (generate_bridge_send, mut generate_bridge_recv) = tokio::sync::mpsc::channel(16);
+
+        let load_channel = channel.clone();
+        let loaded_chunks = self.loaded_entity_chunks.clone();
+        let handle_load = async move {
+            while let Some(data) = load_bridge_recv.recv().await {
+                let is_ok = match data {
+                    LoadedData::Loaded(chunk) => {
+                        let position = chunk.read().await.chunk_position;
+
+                        let value = loaded_chunks
+                            .entry(position)
+                            .or_insert(chunk)
+                            .value()
+                            .clone();
+                        send_chunk(false, value, &load_channel)
+                    }
+                    LoadedData::Missing(pos) => generate_bridge_send.send(pos).await.is_ok(),
+                    LoadedData::Error((pos, error)) => {
+                        match error {
+                            // this is expected, and is not an error
+                            ChunkReadingError::ChunkNotExist
+                            | ChunkReadingError::ParsingError(
+                                ChunkParsingError::ChunkNotGenerated,
+                            ) => {}
+                            // this is an error, and we should log it
+                            error => {
+                                log::error!(
+                                    "Failed to load chunk at {pos:?}: {error} (regenerating)"
+                                );
+                            }
+                        };
+
+                        generate_bridge_send.send(pos).await.is_ok()
+                    }
+                };
+
+                if !is_ok {
+                    // This isn't recoverable, so stop listening
+                    return;
+                }
+            }
+        };
+
+        let loaded_chunks = self.loaded_entity_chunks.clone();
+        let handle_generate = async move {
+            let continue_to_generate = Arc::new(AtomicBool::new(true));
+            while let Some(pos) = generate_bridge_recv.recv().await {
+                if !continue_to_generate.load(Ordering::Relaxed) {
+                    return;
+                }
+
+                let loaded_chunks = loaded_chunks.clone();
+                let channel = channel.clone();
+                let cloned_continue_to_generate = continue_to_generate.clone();
+
+                tokio::spawn(async move {
+                    // Rayon tasks are queued, so also check it here
+                    if !cloned_continue_to_generate.load(Ordering::Relaxed) {
+                        return;
+                    }
+
+                    let result = {
+                        let entry = loaded_chunks.entry(pos); // Get the entry for the position
+
+                        // Check if the entry already exists.
+                        // If not, generate the chunk asynchronously and insert it.
+                        match entry {
+                            Entry::Occupied(entry) => entry.into_ref(),
+                            Entry::Vacant(entry) => {
+                                let generated_chunk = ChunkEntityData {
+                                    chunk_position: pos,
+                                    data: HashMap::new(),
+                                    dirty: true,
+                                };
+                                entry.insert(Arc::new(RwLock::new(generated_chunk)))
+                            }
+                        }
+                        .value()
+                        .clone()
+                    };
+
+                    if !send_chunk(true, result, &channel) {
+                        // Stop any additional queued generations
+                        cloned_continue_to_generate.store(false, Ordering::Relaxed);
+                    }
+                });
+            }
+        };
+
+        let tracker = TaskTracker::new();
+        tracker.spawn(handle_load);
+        tracker.spawn(handle_generate);
+
+        self.entity_saver
+            .fetch_chunks(&self.level_folder, &remaining_chunks, load_bridge_send)
+            .await;
+
+        tracker.close();
+        tracker.wait().await;
+    }
+
     pub fn try_get_chunk(
         &self,
         coordinates: Vector2<i32>,
@@ -654,6 +958,13 @@ impl Level {
         self.loaded_chunks.try_get(&coordinates).try_unwrap()
     }
 
+    pub fn try_get_entity_chunk(
+        &self,
+        coordinates: Vector2<i32>,
+    ) -> Option<dashmap::mapref::one::Ref<'_, Vector2<i32>, Arc<RwLock<ChunkEntityData>>>> {
+        self.loaded_entity_chunks.try_get(&coordinates).try_unwrap()
+    }
+
     pub async fn get_and_tick_block_ticks(&self) -> Arc<Mutex<VecDeque<ScheduledTick>>> {
         let mut block_ticks = self.block_ticks.lock().await;
         let mut ticks = VecDeque::new();

pumpkin/src/block/blocks/piston/mod.rs

@@ -50,13 +50,7 @@ impl<'a> PistonHandler<'a> {
         self.moved_blocks.clear();
         self.broken_blocks.clear();
         let (block, block_state) = self.world.get_block_and_block_state(&self.pos_to).await;
-        dbg!(PistonBlock::is_movable(
+
-            &block,
-            &block_state,
-            self.motion_direction,
-            false,
-            self.piston_direction,
-        ));
         if !PistonBlock::is_movable(
             &block,
             &block_state,

pumpkin/src/block/blocks/pumpkin.rs

@@ -1,3 +1,4 @@
+use crate::entity::Entity;
 use crate::entity::item::ItemEntity;
 use crate::server::Server;
 use crate::world::World;
@@ -11,6 +12,7 @@ use pumpkin_util::math::position::BlockPos;
 use pumpkin_world::item::ItemStack;
 use pumpkin_world::world::BlockFlags;
 use std::sync::Arc;
+use uuid::Uuid;
 #[pumpkin_block("minecraft:pumpkin")]
 pub struct PumpkinBlock;
 
@@ -36,7 +38,13 @@ impl crate::block::pumpkin_block::PumpkinBlock for PumpkinBlock {
                 BlockFlags::NOTIFY_ALL,
             )
             .await;
-        let entity = world.create_entity(pos.to_f64(), EntityType::ITEM);
+        let entity = Entity::new(
+            Uuid::new_v4(),
+            world.clone(),
+            pos.to_f64(),
+            EntityType::ITEM,
+            false,
+        );
         let item_entity =
             Arc::new(ItemEntity::new(entity, ItemStack::new(4, &Item::PUMPKIN_SEEDS)).await);
         world.spawn_entity(item_entity).await;

pumpkin/src/block/blocks/tnt.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use crate::block::pumpkin_block::PumpkinBlock;
 use crate::block::registry::BlockActionResult;
+use crate::entity::Entity;
 use crate::entity::player::Player;
 use crate::entity::tnt::TNTEntity;
 use crate::server::Server;
@@ -17,6 +18,7 @@ use pumpkin_util::math::vector3::Vector3;
 use pumpkin_world::BlockStateId;
 use pumpkin_world::world::BlockFlags;
 use rand::Rng;
+use uuid::Uuid;
 
 use super::redstone::block_receives_redstone_power;
 
@@ -25,7 +27,13 @@ pub struct TNTBlock;
 
 impl TNTBlock {
     pub async fn prime(world: &Arc<World>, location: &BlockPos) {
-        let entity = world.create_entity(location.to_f64(), EntityType::TNT);
+        let entity = Entity::new(
+            Uuid::new_v4(),
+            world.clone(),
+            location.to_f64(),
+            EntityType::TNT,
+            false,
+        );
         let pos = entity.pos.load();
         let tnt = Arc::new(TNTEntity::new(entity, DEFAULT_POWER, DEFAULT_FUSE));
         world.spawn_entity(tnt).await;
@@ -93,7 +101,13 @@ impl PumpkinBlock for TNTBlock {
     }
 
     async fn explode(&self, _block: &Block, world: &Arc<World>, location: BlockPos) {
-        let entity = world.create_entity(location.to_f64(), EntityType::TNT);
+        let entity = Entity::new(
+            Uuid::new_v4(),
+            world.clone(),
+            location.to_f64(),
+            EntityType::TNT,
+            false,
+        );
         let angle = rand::random::<f64>() * std::f64::consts::TAU;
         entity
             .set_velocity(Vector3::new(-angle.sin() * 0.02, 0.2, -angle.cos() * 0.02))

pumpkin/src/command/commands/summon.rs

@@ -1,5 +1,6 @@
 use async_trait::async_trait;
 use pumpkin_util::{math::vector3::Vector3, text::TextComponent};
+use uuid::Uuid;
 
 use crate::{
     command::{
@@ -10,7 +11,7 @@ use crate::{
         },
         tree::{CommandTree, builder::argument},
     },
-    entity::mob,
+    entity::r#type::from_type,
 };
 const NAMES: [&str; 1] = ["summon"];
 
@@ -57,7 +58,7 @@ impl CommandExecutor for Executor {
                 (player.world().await, pos)
             }
         };
-        let mob = mob::from_type(entity, pos, &world);
+        let mob = from_type(entity, pos, &world, Uuid::new_v4());
         world.spawn_entity(mob).await;
 
         sender

pumpkin/src/entity/experience_orb.rs

@@ -3,6 +3,7 @@ use std::sync::{Arc, atomic::AtomicU32};
 use async_trait::async_trait;
 use pumpkin_data::{damage::DamageType, entity::EntityType};
 use pumpkin_util::math::vector3::Vector3;
+use uuid::Uuid;
 
 use crate::{server::Server, world::World};
 
@@ -29,7 +30,13 @@ impl ExperienceOrbEntity {
         while amount > 0 {
             let i = Self::round_to_orb_size(amount);
             amount -= i;
-            let entity = world.create_entity(position, EntityType::EXPERIENCE_ORB);
+            let entity = Entity::new(
+                Uuid::new_v4(),
+                world.clone(),
+                position,
+                EntityType::EXPERIENCE_ORB,
+                false,
+            );
             let orb = Arc::new(Self::new(entity, i));
             world.spawn_entity(orb).await;
         }

pumpkin/src/entity/living.rs

@@ -371,10 +371,7 @@ impl EntityBase for LivingEntity {
     fn get_living_entity(&self) -> Option<&LivingEntity> {
         Some(self)
     }
-}
 
-#[async_trait]
-impl NBTStorage for LivingEntity {
     async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) {
         self.entity.write_nbt(nbt).await;
         nbt.put("Health", NbtTag::Float(self.health.load()));
@@ -389,17 +386,14 @@ impl NBTStorage for LivingEntity {
                     effect.write_nbt(&mut effect_nbt).await;
                     effects_list.push(NbtTag::Compound(effect_nbt));
                 }
-                nbt.put(
+                nbt.put("active_effects", NbtTag::List(effects_list));
-                    "active_effects",
-                    NbtTag::List(effects_list.into_boxed_slice()),
-                );
             }
         }
         //TODO: write equipment
         // todo more...
     }
 
-    async fn read_nbt(&mut self, nbt: &mut pumpkin_nbt::compound::NbtCompound) {
+    async fn read_nbt(&self, nbt: &pumpkin_nbt::compound::NbtCompound) {
         self.entity.read_nbt(nbt).await;
         self.health.store(nbt.get_float("Health").unwrap_or(0.0));
         self.fall_distance

pumpkin/src/entity/mob/mod.rs

@@ -1,12 +1,9 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use pumpkin_data::entity::EntityType;
-use pumpkin_util::math::vector3::Vector3;
 use tokio::sync::Mutex;
-use zombie::Zombie;
 
-use crate::{server::Server, world::World};
+use crate::server::Server;
 
 use super::{
     Entity, EntityBase,
@@ -50,25 +47,3 @@ impl EntityBase for MobEntity {
         Some(&self.living_entity)
     }
 }
-
-pub fn from_type(
-    entity_type: EntityType,
-    position: Vector3<f64>,
-    world: &Arc<World>,
-) -> Arc<dyn EntityBase> {
-    let entity = world.create_entity(position, entity_type);
-
-    #[allow(clippy::single_match)]
-    let mob = match entity_type {
-        EntityType::ZOMBIE => Zombie::make(entity),
-        // TODO
-        _ => MobEntity {
-            living_entity: LivingEntity::new(entity),
-            goals: Mutex::new(vec![]),
-            navigator: Mutex::new(Navigator::default()),
-        },
-    };
-    Arc::new(mob)
-}
-
-impl MobEntity {}

pumpkin/src/entity/mod.rs

@@ -51,6 +51,7 @@ pub mod mob;
 pub mod player;
 pub mod projectile;
 pub mod tnt;
+pub mod r#type;
 
 mod combat;
 
@@ -73,6 +74,22 @@ pub trait EntityBase: Send + Sync {
         }
     }
 
+    async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) {
+        if let Some(living) = self.get_living_entity() {
+            living.write_nbt(nbt).await;
+        } else {
+            self.get_entity().write_nbt(nbt).await;
+        }
+    }
+
+    async fn read_nbt(&self, nbt: &pumpkin_nbt::compound::NbtCompound) {
+        if let Some(living) = self.get_living_entity() {
+            living.read_nbt(nbt).await;
+        } else {
+            self.get_entity().read_nbt(nbt).await;
+        }
+    }
+
     async fn init_data_tracker(&self) {}
 
     async fn teleport(
@@ -151,6 +168,8 @@ pub struct Entity {
     pub fire_ticks: AtomicI32,
     pub has_visual_fire: AtomicBool,
 
+    pub first_loaded_chunk_position: AtomicCell<Option<Vector3<i32>>>,
+
     pub portal_cooldown: AtomicU32,
 
     pub portal_manager: Mutex<Option<Mutex<PortalManager>>>,
@@ -191,6 +210,7 @@ impl Entity {
             velocity: AtomicCell::new(Vector3::new(0.0, 0.0, 0.0)),
             standing_eye_height: entity_type.eye_height,
             pose: AtomicCell::new(EntityPose::Standing),
+            first_loaded_chunk_position: AtomicCell::new(None),
             bounding_box: AtomicCell::new(BoundingBox::new_from_pos(
                 position.x,
                 position.y,
@@ -788,28 +808,43 @@ impl EntityBase for Entity {
     fn get_living_entity(&self) -> Option<&LivingEntity> {
         None
     }
-}
 
-#[async_trait]
-impl NBTStorage for Entity {
     async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) {
         let position = self.pos.load();
+        nbt.put_string(
+            "id",
+            format!("minecraft:{}", self.entity_type.resource_name),
+        );
+        let uuid = self.entity_uuid.as_u128();
+        nbt.put(
+            "UUID",
+            NbtTag::IntArray(vec![
+                (uuid >> 96) as i32,
+                ((uuid >> 64) & 0xFFFF_FFFF) as i32,
+                ((uuid >> 32) & 0xFFFF_FFFF) as i32,
+                (uuid & 0xFFFF_FFFF) as i32,
+            ]),
+        );
         nbt.put(
             "Pos",
-            NbtTag::List(
+            NbtTag::List(vec![
-                vec![position.x.into(), position.y.into(), position.z.into()].into_boxed_slice(),
+                position.x.into(),
-            ),
+                position.y.into(),
+                position.z.into(),
+            ]),
         );
         let velocity = self.velocity.load();
         nbt.put(
             "Motion",
-            NbtTag::List(
+            NbtTag::List(vec![
-                vec![velocity.x.into(), velocity.y.into(), velocity.z.into()].into_boxed_slice(),
+                velocity.x.into(),
-            ),
+                velocity.y.into(),
+                velocity.z.into(),
+            ]),
         );
         nbt.put(
             "Rotation",
-            NbtTag::List(vec![self.yaw.load().into(), self.pitch.load().into()].into_boxed_slice()),
+            NbtTag::List(vec![self.yaw.load().into(), self.pitch.load().into()]),
         );
         nbt.put_short("Fire", self.fire_ticks.load(Relaxed) as i16);
         nbt.put_bool("OnGround", self.on_ground.load(Relaxed));
@@ -822,12 +857,14 @@ impl NBTStorage for Entity {
         // todo more...
     }
 
-    async fn read_nbt(&mut self, nbt: &mut pumpkin_nbt::compound::NbtCompound) {
+    async fn read_nbt(&self, nbt: &pumpkin_nbt::compound::NbtCompound) {
         let position = nbt.get_list("Pos").unwrap();
         let x = position[0].extract_double().unwrap_or(0.0);
         let y = position[1].extract_double().unwrap_or(0.0);
         let z = position[2].extract_double().unwrap_or(0.0);
-        self.set_pos(Vector3::new(x, y, z));
+        let pos = Vector3::new(x, y, z);
+        self.set_pos(pos);
+        self.first_loaded_chunk_position.store(Some(pos.to_i32()));
         let velocity = nbt.get_list("Motion").unwrap();
         let x = velocity[0].extract_double().unwrap_or(0.0);
         let y = velocity[1].extract_double().unwrap_or(0.0);
@@ -854,7 +891,7 @@ impl NBTStorage for Entity {
 
 #[async_trait]
 pub trait NBTStorage: Send + Sync + Sized {
-    async fn write_nbt(&self, nbt: &mut NbtCompound);
+    async fn write_nbt(&self, _nbt: &mut NbtCompound) {}
 
     async fn read_nbt(&mut self, _nbt: &mut NbtCompound) {}
 

pumpkin/src/entity/player.rs

@@ -9,6 +9,7 @@ use std::time::{Duration, Instant};
 use async_trait::async_trait;
 use crossbeam::atomic::AtomicCell;
 use log::warn;
+use pumpkin_world::chunk::{ChunkData, ChunkEntityData};
 use pumpkin_world::inventory::Inventory;
 use tokio::sync::{Mutex, RwLock};
 use tokio::task::JoinHandle;
@@ -69,7 +70,7 @@ use pumpkin_world::entity::entity_data_flags::{
     DATA_PLAYER_MAIN_HAND, DATA_PLAYER_MODE_CUSTOMISATION, SLEEPING_POS_ID,
 };
 use pumpkin_world::item::ItemStack;
-use pumpkin_world::level::SyncChunk;
+use pumpkin_world::level::{SyncChunk, SyncEntityChunk};
 
 use crate::block::blocks::bed::BedBlock;
 use crate::command::client_suggestions;
@@ -104,6 +105,7 @@ enum BatchState {
 pub struct ChunkManager {
     chunks_per_tick: usize,
     chunk_queue: VecDeque<(Vector2<i32>, SyncChunk)>,
+    entity_chunk_queue: VecDeque<(Vector2<i32>, SyncEntityChunk)>,
     batches_sent_since_ack: BatchState,
 }
 
@@ -115,6 +117,7 @@ impl ChunkManager {
         Self {
             chunks_per_tick,
             chunk_queue: VecDeque::new(),
+            entity_chunk_queue: VecDeque::new(),
             batches_sent_since_ack: BatchState::Initial,
         }
     }
@@ -128,6 +131,10 @@ impl ChunkManager {
         self.chunk_queue.push_back((position, chunk));
     }
 
+    pub fn push_entity(&mut self, position: Vector2<i32>, chunk: SyncEntityChunk) {
+        self.entity_chunk_queue.push_back((position, chunk));
+    }
+
     #[must_use]
     pub fn can_send_chunk(&self) -> bool {
         let state_available = match self.batches_sent_since_ack {
@@ -141,12 +148,30 @@ impl ChunkManager {
 
     pub fn next_chunk(&mut self) -> Box<[SyncChunk]> {
         let chunk_size = self.chunk_queue.len().min(self.chunks_per_tick);
-        let mut chunks = Vec::with_capacity(chunk_size);
+        let chunks: Vec<Arc<RwLock<ChunkData>>> = self
-        chunks.extend(
+            .chunk_queue
-            self.chunk_queue
             .drain(0..chunk_size)
-                .map(|(_, chunk)| chunk),
+            .map(|(_, chunk)| chunk)
-        );
+            .collect();
+
+        match &mut self.batches_sent_since_ack {
+            BatchState::Count(count) => {
+                count.add_assign(1);
+            }
+            state @ BatchState::Initial => *state = BatchState::Waiting,
+            BatchState::Waiting => unreachable!(),
+        }
+
+        chunks.into_boxed_slice()
+    }
+
+    pub fn next_entity(&mut self) -> Box<[SyncEntityChunk]> {
+        let chunk_size = self.entity_chunk_queue.len().min(self.chunks_per_tick);
+        let chunks: Vec<Arc<RwLock<ChunkEntityData>>> = self
+            .entity_chunk_queue
+            .drain(0..chunk_size)
+            .map(|(_, chunk)| chunk)
+            .collect();
 
         match &mut self.batches_sent_since_ack {
             BatchState::Count(count) => {
@@ -404,6 +429,7 @@ impl Player {
         let chunks_to_clean = level.mark_chunks_as_not_watched(&radial_chunks).await;
         // Remove chunks with no watchers from the cache
         level.clean_chunks(&chunks_to_clean).await;
+        level.clean_entity_chunks(&chunks_to_clean).await;
         // Remove left over entries from all possiblily loaded chunks
         level.clean_memory();
 
@@ -970,17 +996,6 @@ impl Player {
             .await;
     }
 
-    /// Sends a world's mobs to only this player.
-    // TODO: This should be optimized for larger servers based on the player's current chunk.
-    pub async fn send_mobs(&self, world: &World) {
-        let entities = world.entities.read().await.clone();
-        for entity in entities.values() {
-            self.client
-                .enqueue_packet(&entity.get_entity().create_spawn_packet())
-                .await;
-        }
-    }
-
     async fn unload_watched_chunks(&self, world: &World) {
         let radial_chunks = self.watched_section.load().all_chunks_within();
         let level = &world.level;
@@ -1430,10 +1445,14 @@ impl Player {
     }
 
     pub async fn drop_item(&self, item_stack: ItemStack) {
-        let entity = self.world().await.create_entity(
+        let item_pos = self.living_entity.entity.pos.load()
-            self.living_entity.entity.pos.load()
+            + Vector3::new(0.0, f64::from(EntityType::PLAYER.eye_height) - 0.3, 0.0);
-                + Vector3::new(0.0, f64::from(EntityType::PLAYER.eye_height) - 0.3, 0.0),
+        let entity = Entity::new(
+            Uuid::new_v4(),
+            self.world().await,
+            item_pos,
             EntityType::ITEM,
+            false,
         );
 
         let pitch = f64::from(self.living_entity.entity.pitch.load()).to_radians();
@@ -1945,7 +1964,7 @@ impl NBTStorage for PlayerInventory {
         }
 
         // Save the inventory list
-        nbt.put("Inventory", NbtTag::List(vec.into_boxed_slice()));
+        nbt.put("Inventory", NbtTag::List(vec));
     }
 
     async fn read_nbt_non_mut(&self, nbt: &mut NbtCompound) {

pumpkin/src/entity/type.rs

@@ -0,0 +1,37 @@
+use std::sync::Arc;
+
+use pumpkin_data::entity::EntityType;
+use pumpkin_util::math::vector3::Vector3;
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::{
+    entity::{
+        Entity, EntityBase,
+        ai::path::Navigator,
+        living::LivingEntity,
+        mob::{MobEntity, zombie::Zombie},
+    },
+    world::World,
+};
+
+pub fn from_type(
+    entity_type: EntityType,
+    position: Vector3<f64>,
+    world: &Arc<World>,
+    uuid: Uuid,
+) -> Arc<dyn EntityBase> {
+    let entity = Entity::new(uuid, world.clone(), position, entity_type, false);
+
+    #[allow(clippy::single_match)]
+    let mob = match entity_type {
+        EntityType::ZOMBIE => Zombie::make(entity),
+        // TODO
+        _ => MobEntity {
+            living_entity: LivingEntity::new(entity),
+            goals: Mutex::new(vec![]),
+            navigator: Mutex::new(Navigator::default()),
+        },
+    };
+    Arc::new(mob)
+}

pumpkin/src/item/items/egg.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use crate::entity::Entity;
 use crate::entity::player::Player;
 use crate::entity::projectile::ThrownItemEntity;
 use crate::item::pumpkin_item::{ItemMetadata, PumpkinItem};
@@ -7,6 +8,7 @@ use async_trait::async_trait;
 use pumpkin_data::entity::EntityType;
 use pumpkin_data::item::Item;
 use pumpkin_data::sound::Sound;
+use uuid::Uuid;
 
 pub struct EggItem;
 
@@ -31,7 +33,13 @@ impl PumpkinItem for EggItem {
             )
             .await;
         // TODO: Implement eggs the right way, so there is a chance of spawning chickens
-        let entity = world.create_entity(position, EntityType::EGG);
+        let entity = Entity::new(
+            Uuid::new_v4(),
+            world.clone(),
+            position,
+            EntityType::EGG,
+            false,
+        );
         let snowball = ThrownItemEntity::new(entity, &player.living_entity.entity);
         let yaw = player.living_entity.entity.yaw.load();
         let pitch = player.living_entity.entity.pitch.load();

pumpkin/src/item/items/hoe.rs

@@ -1,3 +1,4 @@
+use crate::entity::Entity;
 use crate::entity::item::ItemEntity;
 use crate::entity::player::Player;
 use crate::item::pumpkin_item::{ItemMetadata, PumpkinItem};
@@ -12,6 +13,7 @@ use pumpkin_util::math::position::BlockPos;
 use pumpkin_world::item::ItemStack;
 use pumpkin_world::world::BlockFlags;
 use std::sync::Arc;
+use uuid::Uuid;
 
 pub struct HoeItem;
 
@@ -89,7 +91,13 @@ impl PumpkinItem for HoeItem {
                     BlockDirection::West => location.up().to_f64().add_raw(-1.0, -0.4, 0.0),
                     BlockDirection::East => location.up().to_f64().add_raw(1.0, -0.4, 0.0),
                 };
-                let entity = world.create_entity(location, EntityType::ITEM);
+                let entity = Entity::new(
+                    Uuid::new_v4(),
+                    world.clone(),
+                    location,
+                    EntityType::SNOWBALL,
+                    false,
+                );
                 // TODO: Merge stacks together
                 let item_entity = Arc::new(
                     ItemEntity::new(entity, ItemStack::new(1, &Item::HANGING_ROOTS)).await,

pumpkin/src/item/items/snowball.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use crate::entity::Entity;
 use crate::entity::player::Player;
 use crate::entity::projectile::ThrownItemEntity;
 use crate::item::pumpkin_item::{ItemMetadata, PumpkinItem};
@@ -7,6 +8,7 @@ use async_trait::async_trait;
 use pumpkin_data::entity::EntityType;
 use pumpkin_data::item::Item;
 use pumpkin_data::sound::Sound;
+use uuid::Uuid;
 
 pub struct SnowBallItem;
 
@@ -30,7 +32,13 @@ impl PumpkinItem for SnowBallItem {
                 &position,
             )
             .await;
-        let entity = world.create_entity(position, EntityType::SNOWBALL);
+        let entity = Entity::new(
+            Uuid::new_v4(),
+            world.clone(),
+            position,
+            EntityType::SNOWBALL,
+            false,
+        );
         let snowball = ThrownItemEntity::new(entity, &player.living_entity.entity);
         let yaw = player.living_entity.entity.yaw.load();
         let pitch = player.living_entity.entity.pitch.load();

pumpkin/src/net/packet/play.rs

@@ -46,12 +46,14 @@ use pumpkin_world::block::entities::command_block::CommandBlockEntity;
 use pumpkin_world::block::entities::sign::SignBlockEntity;
 use pumpkin_world::item::ItemStack;
 use pumpkin_world::world::BlockFlags;
+use uuid::Uuid;
 
 use crate::block::registry::BlockActionResult;
 use crate::block::{self, BlockIsReplacing};
 use crate::command::CommandSender;
+use crate::entity::EntityBase;
 use crate::entity::player::{ChatMode, ChatSession, Hand, Player};
-use crate::entity::{EntityBase, mob};
+use crate::entity::r#type::from_type;
 use crate::error::PumpkinError;
 use crate::net::PlayerConfig;
 use crate::plugin::player::player_chat::PlayerChatEvent;
@@ -1622,7 +1624,7 @@ impl Player {
 
         let world = self.world().await;
         // Create a new mob and UUID based on the spawn egg id
-        let mob = mob::from_type(EntityType::from_raw(entity_type.id).unwrap(), pos, &world);
+        let mob = from_type(entity_type, pos, &world, Uuid::new_v4());
 
         // Set the rotation
         mob.get_entity().set_rotation(yaw, 0.0);

pumpkin/src/world/mod.rs

@@ -12,7 +12,7 @@ use crate::{
     PLUGIN_MANAGER,
     block::{self, registry::BlockRegistry},
     command::client_suggestions,
-    entity::{Entity, EntityBase, EntityId, player::Player},
+    entity::{Entity, EntityBase, EntityId, player::Player, r#type::from_type},
     error::PumpkinError,
     plugin::{
         block::block_break::BlockBreakEvent,
@@ -80,7 +80,7 @@ use pumpkin_util::{
 };
 use pumpkin_world::{
     BlockStateId, GENERATION_SETTINGS, GeneratorSetting, biome, block::entities::BlockEntity,
-    item::ItemStack,
+    chunk::io::Dirtiable, item::ItemStack,
 };
 use pumpkin_world::{chunk::ChunkData, world::BlockAccessor};
 use pumpkin_world::{chunk::TickPriority, level::Level};
@@ -102,6 +102,7 @@ pub mod custom_bossbar;
 pub mod scoreboard;
 pub mod weather;
 
+use uuid::Uuid;
 use weather::Weather;
 
 type FlowingFluidProperties = pumpkin_data::fluid::FlowingWaterLikeFluidProperties;
@@ -196,9 +197,62 @@ impl World {
     }
 
     pub async fn shutdown(&self) {
+        for (uuid, entity) in self.entities.read().await.iter() {
+            self.save_entity(uuid, entity).await;
+        }
         self.level.shutdown().await;
     }
 
+    async fn save_entity(&self, uuid: &uuid::Uuid, entity: &Arc<dyn EntityBase>) {
+        // First lets see if the entity was saved on an other chunk, and if the current chunk does not match we remove it
+        // Otherwise we just update the nbt data
+        let base_entity = entity.get_entity();
+        let (current_chunk_coordinate, _) = base_entity
+            .block_pos
+            .load()
+            .chunk_and_chunk_relative_position();
+        let mut nbt = NbtCompound::new();
+        entity.write_nbt(&mut nbt).await;
+        if let Some(old_chunk) = base_entity.first_loaded_chunk_position.load() {
+            let old_chunk = old_chunk.to_vec2_i32();
+            let chunk = self.level.get_entity_chunk(old_chunk).await;
+            let mut chunk = chunk.write().await;
+            chunk.mark_dirty(true);
+            if old_chunk == current_chunk_coordinate {
+                chunk.data.insert(*uuid, nbt);
+                return;
+            }
+
+            // The chunk has changed, lets remove the entity from the old chunk
+            chunk.data.remove(uuid);
+        }
+        // We did not continue, so lets save data in a new chunk
+        let chunk = self.level.get_entity_chunk(current_chunk_coordinate).await;
+        let mut chunk = chunk.write().await;
+        chunk.data.insert(*uuid, nbt);
+        chunk.mark_dirty(true);
+    }
+
+    async fn remove_entity_data(&self, entity: &Entity) {
+        let (current_chunk_coordinate, _) =
+            entity.block_pos.load().chunk_and_chunk_relative_position();
+        if let Some(old_chunk) = entity.first_loaded_chunk_position.load() {
+            let old_chunk = old_chunk.to_vec2_i32();
+            let chunk = self.level.get_entity_chunk(old_chunk).await;
+            let mut chunk = chunk.write().await;
+            chunk.mark_dirty(true);
+            if old_chunk == current_chunk_coordinate {
+                chunk.data.remove(&entity.entity_uuid);
+            } else {
+                let chunk = self.level.get_entity_chunk(current_chunk_coordinate).await;
+                let mut chunk = chunk.write().await;
+                // The chunk has changed, lets remove the entity from the old chunk
+                chunk.data.remove(&entity.entity_uuid);
+                chunk.mark_dirty(true);
+            }
+        }
+    }
+
     pub async fn send_entity_status(&self, entity: &Entity, status: EntityStatus) {
         // TODO: only nearby
         self.broadcast_packet_all(&CEntityStatus::new(entity.entity_id, status as i8))
@@ -895,7 +949,6 @@ impl World {
         // }
 
         player.has_played_before.store(true, Ordering::Relaxed);
-        player.send_mobs(self).await;
         player
             .on_screen_handler_opened(player.player_screen_handler.clone())
             .await;
@@ -1083,7 +1136,7 @@ impl World {
     /// IMPORTANT: Chunks have to be non-empty
     #[allow(clippy::too_many_lines)]
     fn spawn_world_chunks(
-        &self,
+        self: &Arc<Self>,
         player: Arc<Player>,
         chunks: Vec<Vector2<i32>>,
         center_chunk: Vector2<i32>,
@@ -1107,8 +1160,12 @@ impl World {
             rel_x * rel_x + rel_z * rel_z
         });
 
-        let mut receiver = self.level.receive_chunks(chunks);
+        let mut receiver = self.level.receive_chunks(chunks.clone());
+
         let level = self.level.clone();
+        let player1 = player.clone();
+        let world = self.clone();
+        let world1 = self.clone();
 
         player.clone().spawn_task(async move {
             'main: loop {
@@ -1145,11 +1202,11 @@ impl World {
                 }
 
                 let (world, chunk) = if level.is_chunk_watched(&position) {
-                    (player.world().await.clone(), chunk)
+                    (world.clone(), chunk)
                 } else {
                     send_cancellable! {{
                         ChunkSave {
-                            world: player.world().await.clone(),
+                            world: world.clone(),
                             chunk,
                             cancelled: false,
                         };
@@ -1199,6 +1256,102 @@ impl World {
                 }
             }
 
+            #[cfg(debug_assertions)]
+            log::debug!("Chunks queued after {}ms", inst.elapsed().as_millis());
+        });
+        let mut entity_receiver = self.level.receive_entity_chunks(chunks);
+        let level = self.level.clone();
+        let player = player1.clone();
+        let world = world1.clone();
+        player.clone().spawn_task(async move {
+            'main: loop {
+                let recv_result = tokio::select! {
+                    () = player.client.await_close_interrupt() => {
+                        log::debug!("Canceling player packet processing");
+                        None
+                    },
+                    recv_result = entity_receiver.recv() => {
+                        recv_result
+                    }
+                };
+
+                let Some((chunk, _first_load)) = recv_result else {
+                    break;
+                };
+                let position = chunk.read().await.chunk_position;
+
+                let chunk = if level.is_chunk_watched(&position) {
+                    chunk
+                } else {
+                    log::trace!(
+                        "Received chunk {:?}, but it is no longer watched... cleaning",
+                        &position
+                    );
+                    let mut ids = Vec::new();
+                    // Remove all the entities from the world
+                    let entity_chunk = chunk.read().await;
+                    let mut entities = world.entities.write().await;
+                    for (uuid, entity_nbt) in &entity_chunk.data {
+                        let Some(id) = entity_nbt.get_string("id") else {
+                            log::warn!("Entity has no ID");
+                            continue;
+                        };
+                        let Some(entity_type) =
+                            EntityType::from_name(id.strip_prefix("minecraft:").unwrap_or(id))
+                        else {
+                            log::warn!("Entity has no valid Entity Type {id}");
+                            continue;
+                        };
+                        // Pos is zero since it will read from nbt
+                        let entity =
+                            from_type(entity_type, Vector3::new(0.0, 0.0, 0.0), &world, *uuid);
+                        entity.read_nbt(entity_nbt).await;
+                        let base_entity = entity.get_entity();
+
+                        entities.remove(&base_entity.entity_uuid);
+                        ids.push(VarInt(base_entity.entity_id));
+
+                        world.save_entity(uuid, &entity).await;
+                    }
+                    if !ids.is_empty() {
+                        player
+                            .client
+                            .enqueue_packet(&CRemoveEntities::new(&ids))
+                            .await;
+                    }
+                    level.clean_entity_chunk(&position).await;
+
+                    continue 'main;
+                };
+
+                let entity_chunk = chunk.read().await;
+                // Add all new Entities to the world
+                let mut current_entities = world.entities.write().await;
+
+                for (uuid, entity_nbt) in &entity_chunk.data {
+                    let Some(id) = entity_nbt.get_string("id") else {
+                        log::warn!("Entity has no ID");
+                        continue;
+                    };
+                    let Some(entity_type) =
+                        EntityType::from_name(id.strip_prefix("minecraft:").unwrap_or(id))
+                    else {
+                        log::warn!("Entity has no valid Entity Type {id}");
+                        continue;
+                    };
+                    // Pos is zero since it will read from nbt
+                    let entity = from_type(entity_type, Vector3::new(0.0, 0.0, 0.0), &world, *uuid);
+                    entity.read_nbt(entity_nbt).await;
+                    let base_entity = entity.get_entity();
+                    player
+                        .client
+                        .enqueue_packet(&base_entity.create_spawn_packet())
+                        .await;
+                    entity.init_data_tracker().await;
+                    current_entities.insert(base_entity.entity_uuid, entity);
+                }
+            }
+
             #[cfg(debug_assertions)]
             log::debug!("Chunks queued after {}ms", inst.elapsed().as_millis());
         });
@@ -1427,29 +1580,34 @@ impl World {
         }
     }
 
-    pub fn create_entity(
-        self: &Arc<Self>,
-        position: Vector3<f64>,
-        entity_type: EntityType,
-    ) -> Entity {
-        let uuid = uuid::Uuid::new_v4();
-        Entity::new(uuid, self.clone(), position, entity_type, false)
-    }
-
     /// Adds an entity to the world.
     pub async fn spawn_entity(&self, entity: Arc<dyn EntityBase>) {
         let base_entity = entity.get_entity();
         self.broadcast_packet_all(&base_entity.create_spawn_packet())
             .await;
         entity.init_data_tracker().await;
-        let mut current_living_entities = self.entities.write().await;
+
-        current_living_entities.insert(base_entity.entity_uuid, entity);
+        let (chunk_coordinate, _) = base_entity
+            .block_pos
+            .load()
+            .chunk_and_chunk_relative_position();
+        let chunk = self.level.get_entity_chunk(chunk_coordinate).await;
+        let mut chunk = chunk.write().await;
+        let mut nbt = NbtCompound::new();
+        entity.write_nbt(&mut nbt).await;
+        chunk.data.insert(base_entity.entity_uuid, nbt);
+        chunk.mark_dirty(true);
+
+        let mut current_entities = self.entities.write().await;
+        current_entities.insert(base_entity.entity_uuid, entity);
     }
 
     pub async fn remove_entity(&self, entity: &Entity) {
         self.entities.write().await.remove(&entity.entity_uuid);
         self.broadcast_packet_all(&CRemoveEntities::new(&[entity.entity_id.into()]))
             .await;
+
+        self.remove_entity_data(entity).await;
     }
 
     pub async fn set_block_breaking(&self, from: &Entity, location: BlockPos, progress: i32) {
@@ -1483,7 +1641,7 @@ impl World {
             return block_state_id;
         }
 
-        chunk.dirty = true;
+        chunk.mark_dirty(true);
 
         chunk.section.set_block_absolute_y(
             relative.x as usize,
@@ -1683,7 +1841,7 @@ impl World {
             f64::from(pos.0.z) + 0.5 + rand::rng().random_range(-0.25..0.25),
         );
 
-        let entity = self.create_entity(pos, EntityType::ITEM);
+        let entity = Entity::new(Uuid::new_v4(), self.clone(), pos, EntityType::ITEM, false);
         let item_entity = Arc::new(ItemEntity::new(entity, stack).await);
         self.spawn_entity(item_entity).await;
     }
@@ -1875,7 +2033,7 @@ impl World {
             block_pos,
             (block_entity_nbt.unwrap_or_default(), block_entity),
         );
-        chunk.dirty = true;
+        chunk.mark_dirty(true);
     }
 
     pub async fn remove_block_entity(&self, block_pos: &BlockPos) {
@@ -1885,7 +2043,7 @@ impl World {
             .await;
         let mut chunk: tokio::sync::RwLockWriteGuard<ChunkData> = chunk.write().await;
         chunk.block_entities.remove(block_pos);
-        chunk.dirty = true;
+        chunk.mark_dirty(true);
     }
 
     fn intersects_aabb_with_direction(