synvael/crates/server/src/world_server.rs

// SPDX-License-Identifier: AGPL-3.0-only

//! Authoritative chunk storage and generation logic for the server.

use bevy_ecs::prelude::Resource;
use crossbeam_channel::{Receiver, Sender};
use shared::{
    generator::VoxelGenerator,
    save::SaveError,
    world::{Chunk, ChunkData, ChunkPos},
};
use std::{
    collections::{HashMap, HashSet},
    num::NonZeroUsize,
    path::PathBuf,
    sync::Arc,
    thread::JoinHandle,
};
use tracing::warn;

use crate::chunk_cache::ChunkCache;
use crate::save::{SaveActor, SaveRequest};

/// Outcome of a single streaming reconcile pass, surfaced for logging and tests.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct StreamStats {
    /// Number of chunks generated and inserted this pass.
    pub loaded: usize,
    /// Number of chunks evicted this pass.
    pub unloaded: usize,
    /// Number of chunks resident after the pass.
    pub resident: usize,
    /// Number of chunks in flight.
    pub in_flight: usize,
}

/// The server's authoritative representation of the world.
#[derive(Resource)]
pub struct ServerWorld {
    /// Currently resident chunks, keyed by chunk-space position.
    chunks: HashMap<ChunkPos, Chunk>,
    /// Sending end of the job channel; the main thread pushes positions to load.
    job_tx: Sender<ChunkPos>,
    /// Receiving end of the result channel; the main thread drains finished chunks returned by workers.
    result_rx: Receiver<(ChunkPos, Chunk)>,
    /// Positions dispatched to a worker but not yet returned, preventing the same chunk being re-dispatched on subsequent passes.
    in_flight: HashSet<ChunkPos>,
    /// The dedicated thread owning all region files, kept alive for the world's lifetime.
    #[expect(
        dead_code,
        reason = "retained to keep the save request channel open for the workers"
    )]
    save_actor: SaveActor,
    /// Handles to the generation worker threads, retained so they can be joined on shutdown.
    // TODO: Remove once the server has a graceful-stop sequence
    #[expect(
        dead_code,
        reason = "retained for a future graceful-shutdown join path"
    )]
    workers: Vec<JoinHandle<()>>,
    /// The same read-only generator handle the workers share, held so eviction can regenerate a chunk's baseline to diff against.
    generator: Arc<VoxelGenerator>,
    /// A handle onto the shared baseline cache, used to resolve the baseline during the unload diff.
    cache: ChunkCache,
    /// The save actor's request sender, used to issue `Write` and `Remove` on unload.
    save_tx: Sender<SaveRequest>,
}

impl ServerWorld {
    /// Initializes a new authoritative server world with the provided generator, resolving chunk loads against region files under `region_dir` and caching up to `cache_capacity` regenerated baselines.
    #[must_use]
    pub fn new(
        generator: VoxelGenerator,
        region_dir: PathBuf,
        cache_capacity: NonZeroUsize,
    ) -> Self {
        let generator = Arc::new(generator);
        let (job_tx, job_rx) = crossbeam_channel::unbounded::<ChunkPos>();
        let (result_tx, result_rx) = crossbeam_channel::unbounded::<(ChunkPos, Chunk)>();

        // The actor owns every region file; workers reach it only through cloned request senders.
        let save_actor = SaveActor::spawn(region_dir);

        // Baselines are shared across the pool through cloned handles onto one bounded store.
        let cache = ChunkCache::new(cache_capacity);

        let worker_count = std::thread::available_parallelism().map_or(4, std::num::NonZero::get);

        let workers = (0..worker_count)
            .map(|_| {
                // Each worker shares a handle to the read-only generator, its own view of the shared job queue, its own sender back into the result channel, its own request sender to the save actor, and a handle onto the shared baseline cache.
                let generator = Arc::clone(&generator);
                let job_rx = job_rx.clone();
                let result_tx = result_tx.clone();
                let save_tx = save_actor.sender();
                let cache = cache.clone();

                std::thread::spawn(move || {
                    // Block until a job arrives.
                    while let Ok(pos) = job_rx.recv() {
                        let chunk = load_chunk(&generator, &save_tx, &cache, pos);
                        // A send error means the main thread has gone away; nothing is left to do but let the worker wind down.
                        if result_tx.send((pos, chunk)).is_err() {
                            break;
                        }
                    }
                })
            })
            .collect();

        // Drop the template ends left over after cloning so the channels close once the real holders are gone.
        drop(job_rx);
        drop(result_tx);

        let save_tx = save_actor.sender();

        Self {
            chunks: HashMap::new(),
            job_tx,
            result_rx,
            in_flight: HashSet::new(),
            save_actor,
            workers,
            generator,
            cache,
            save_tx,
        }
    }

    /// Number of chunks currently resident in memory.
    #[must_use]
    pub fn loaded_count(&self) -> usize {
        self.chunks.len()
    }

    /// Returns the resident chunk at `pos`, or `None` if it is not currently loaded.
    #[must_use]
    pub fn chunk(&self, pos: ChunkPos) -> Option<&Chunk> {
        self.chunks.get(&pos)
    }

    /// Number of chunks dispatched to the worker pool but not yet returned.
    #[must_use]
    pub fn in_flight_count(&self) -> usize {
        self.in_flight.len()
    }

    /// Returns `true` when the worker pool has no outstanding work, i.e. every dispatched chunk has been returned. The loading gate polls this to decide when the initial region has finished streaming.
    #[must_use]
    pub fn streaming_idle(&self) -> bool {
        self.in_flight.is_empty()
    }

    /// Reconciles resident chunks against the desired set without blocking the caller: finished chunks are drained from the worker pool, resident chunks absent from `desired` are evicted, and still-missing chunks are dispatched to the pool.
    pub fn reconcile(&mut self, desired: &HashSet<ChunkPos>) -> StreamStats {
        // Drain: absorb every chunk the workers finished since the last pass.
        let mut loaded = 0;
        while let Ok((pos, chunk)) = self.result_rx.try_recv() {
            self.in_flight.remove(&pos);
            // A finished chunk is only kept if it is still wanted.
            if desired.contains(&pos) {
                self.chunks.insert(pos, chunk);
                loaded += 1;
            }
        }

        // Evict: drop resident chunks no anchor wants any more.
        let stale: Vec<ChunkPos> = self
            .chunks
            .keys()
            .filter(|pos| !desired.contains(pos))
            .copied()
            .collect();
        for pos in &stale {
            // The chunk is taken by value so it can be diffed against its baseline before being dropped.
            let Some(chunk) = self.chunks.remove(pos) else {
                continue;
            };
            let baseline = self.cache.get_or_generate(*pos, &self.generator);
            // * NOTE: The diff is stamped with worldgen version 0: a single version exists today. This must become the chunk's stored version once worldgen versioning lands.
            let data = ChunkData::from_diff(*pos, 0, &baseline, &chunk);
            // A clean chunk drops any prior record into the region free list; a dirty chunk writes its diff back. Both only mutate the actor's in-memory image until a flush. A send error means the actor is gone, which the tick thread cannot act on.
            let request = if data.is_unmodified() {
                SaveRequest::Remove { pos: *pos }
            } else {
                SaveRequest::Write { pos: *pos, data }
            };
            let _ = self.save_tx.send(request);
        }

        // Dispatch: request a load for every wanted position that is neither resident nor already in flight.
        for &pos in desired {
            if !self.chunks.contains_key(&pos) && !self.in_flight.contains(&pos) {
                self.in_flight.insert(pos);
                // A send error means the workers shut down; nothing useful can be done with the position, so the failure is ignored.
                let _ = self.job_tx.send(pos);
            }
        }

        StreamStats {
            loaded,
            unloaded: stale.len(),
            resident: self.chunks.len(),
            in_flight: self.in_flight.len(),
        }
    }
}

/// Resolves a chunk position to dense voxel data: a saved modification is applied over its baseline, otherwise the deterministic baseline is regenerated directly.
fn load_chunk(
    generator: &VoxelGenerator,
    save_tx: &Sender<SaveRequest>,
    cache: &ChunkCache,
    pos: ChunkPos,
) -> Chunk {
    match request_saved_chunk(save_tx, pos) {
        Ok(Some(data)) => {
            // A saved modification stores only edits, so the baseline is regenerated and the edits are layered on top.
            // TODO: once worldgen versioning exists, the baseline must be regenerated at `data.worldgen_version()` rather than the current version; today there is a single version, so the current baseline matches.
            data.materialize(&cache.get_or_generate(pos, generator))
        }
        // The chunk was never modified, so its content is exactly the deterministic baseline.
        Ok(None) => cache.get_or_generate(pos, generator),
        Err(error) => {
            // A save-layer failure must not wedge streaming; the chunk falls back to a fresh baseline and the error is logged.
            warn!(?error, ?pos, "chunk load failed; regenerating baseline");
            cache.get_or_generate(pos, generator)
        }
    }
}

/// Sends a read request to the save actor and blocks for its reply, mapping a departed actor to an absent record so generation can still proceed.
///
/// # Errors
///
/// Returns the [`SaveError`] reported by the save actor if reading the stored chunk fails. A departed actor yields `Ok(None)` rather than an error.
fn request_saved_chunk(
    save_tx: &Sender<SaveRequest>,
    pos: ChunkPos,
) -> Result<Option<ChunkData>, SaveError> {
    let (reply_tx, reply_rx) = crossbeam_channel::bounded(1);
    if save_tx
        .send(SaveRequest::Read {
            pos,
            reply: reply_tx,
        })
        .is_err()
    {
        return Ok(None);
    }
    // A receive error means the actor dropped the reply end, treated the same as no saved data.
    reply_rx.recv().unwrap_or(Ok(None))
}

/// Inserts every chunk position inside the streaming cylinder around `center` into `out`.
pub fn cylinder_chunks<S: std::hash::BuildHasher>(
    center: ChunkPos,
    radius: i32,
    out: &mut HashSet<ChunkPos, S>,
) {
    for x in center.x - radius..=center.x + radius {
        for z in center.z - radius..=center.z + radius {
            let dx = x - center.x;
            let dz = z - center.z;

            // Keep only the columns whose XZ distance falls within the disc.
            if dx * dx + dz * dz <= radius * radius {
                for y in center.y - radius / 2..=center.y + radius / 2 {
                    out.insert(ChunkPos::new(x, y, z));
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use shared::generator::{VoxelGenerator, WorldGenConfig};
    use shared::save::SaveError;
    use shared::world::{BlockId, ChunkData, ChunkPos};
    use std::collections::HashSet;
    use std::time::{Duration, Instant};

    use crate::save::{RegionFile, SaveRequest, region_path};

    /// Builds a generator with a small, cheap terrain configuration for streaming tests.
    fn test_generator() -> VoxelGenerator {
        let config = WorldGenConfig {
            base_height: 8,
            noise_scale: 0.05,
            surface_block: BlockId(1),
            subsurface_block: BlockId(2),
            stone_block: BlockId(3),
        };
        VoxelGenerator::new(config, 42)
    }

    /// Builds a server world whose saves resolve against `region_dir`, backed by a small baseline cache.
    fn test_world(region_dir: std::path::PathBuf) -> ServerWorld {
        let capacity = std::num::NonZeroUsize::new(64).unwrap_or(std::num::NonZeroUsize::MIN);
        ServerWorld::new(test_generator(), region_dir, capacity)
    }

    /// Repeatedly reconciles `desired` until the worker pool reports no outstanding work, returning the final pass's stats. Fails the test if the pool does not drain within a fixed timeout.
    fn drain_to_idle(world: &mut ServerWorld, desired: &HashSet<ChunkPos>) -> StreamStats {
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            let stats = world.reconcile(desired);
            if stats.in_flight == 0 {
                return stats;
            }
            assert!(
                Instant::now() < deadline,
                "worker pool did not drain in time"
            );
            std::thread::sleep(Duration::from_millis(1));
        }
    }

    /// Issues a flush against the world's save actor and blocks until every dirty region is written. Because write-backs and this flush travel the same sender to the single actor thread, the reply confirms the preceding writes are durable.
    fn flush(world: &ServerWorld) -> Result<(), SaveError> {
        let (reply_tx, reply_rx) = crossbeam_channel::bounded(1);
        // A send error means the actor has already stopped, leaving nothing to flush.
        if world
            .save_tx
            .send(SaveRequest::Flush { reply: reply_tx })
            .is_err()
        {
            return Ok(());
        }
        reply_rx.recv().unwrap_or(Ok(()))
    }

    #[test]
    fn reconcile_converges_over_multiple_passes() -> Result<(), SaveError> {
        // A fresh empty directory means every load is a miss and resolves to the baseline.
        let dir = tempfile::tempdir()?;
        let mut world = test_world(dir.path().to_path_buf());
        let mut desired = HashSet::new();
        cylinder_chunks(ChunkPos::new(0, 0, 0), 2, &mut desired);

        // The first pass only dispatches work; because loading is off-thread, nothing is resident yet and every position is in flight.
        let first = world.reconcile(&desired);
        assert_eq!(first.loaded, 0);
        assert_eq!(first.resident, 0);
        assert!(first.in_flight > 0);

        // Later passes drain finished chunks until the pool is idle, at which point every desired position must be resident.
        let final_stats = drain_to_idle(&mut world, &desired);
        assert_eq!(final_stats.in_flight, 0);
        assert_eq!(final_stats.resident, desired.len());
        Ok(())
    }

    #[test]
    fn evicted_chunk_is_not_repopulated_on_arrival() -> Result<(), SaveError> {
        let dir = tempfile::tempdir()?;
        let mut world = test_world(dir.path().to_path_buf());
        let target = ChunkPos::new(0, 0, 0);
        let mut desired = HashSet::new();
        desired.insert(target);

        // Dispatch the chunk, then immediately stop wanting it.
        world.reconcile(&desired);

        // Every subsequent pass reconciles against an empty desired set, so the finished chunk is discarded on arrival rather than inserted.
        let empty = HashSet::new();
        let final_stats = drain_to_idle(&mut world, &empty);
        assert_eq!(final_stats.in_flight, 0);
        assert_eq!(final_stats.resident, 0);
        Ok(())
    }

    #[test]
    fn saved_modification_is_applied_over_baseline() -> Result<(), SaveError> {
        // A modified chunk is written to disk, then streamed back; the resident chunk must show the edit rather than the bare baseline.
        let dir = tempfile::tempdir()?;
        let pos = ChunkPos::new(0, 0, 0);
        let edited_index = 100u32;
        let edited_block = BlockId(999);

        let mut data = ChunkData::new(pos, 0);
        data.set(edited_index, edited_block);

        let mut region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?;
        region.write_chunk(pos, &data, 0)?;
        region.save()?;

        let mut world = test_world(dir.path().to_path_buf());
        let mut desired = HashSet::new();
        desired.insert(pos);
        drain_to_idle(&mut world, &desired);

        // The resident chunk must carry the stored edit layered over its regenerated baseline.
        assert!(
            world
                .chunk(pos)
                .is_some_and(|chunk| chunk.blocks[edited_index as usize] == edited_block)
        );
        Ok(())
    }

    #[test]
    fn dirty_chunk_is_written_back_on_eviction() -> Result<(), SaveError> {
        // A resident chunk edited away from its baseline must survive an evict -> flush -> reload round-trip.
        let dir = tempfile::tempdir()?;
        let pos = ChunkPos::new(0, 0, 0);
        let edited_index = 100usize;
        let edited_block = BlockId(999);

        let mut world = test_world(dir.path().to_path_buf());
        let mut desired = HashSet::new();
        desired.insert(pos);
        drain_to_idle(&mut world, &desired);

        // Mutate the resident chunk so it diverges from the baseline the eviction diff regenerates.
        assert!(
            world
                .chunks
                .get_mut(&pos)
                .map(|chunk| chunk.blocks[edited_index] = edited_block)
                .is_some()
        );

        // Reconciling against an empty desired set evicts the chunk, sending its diff to the actor.
        world.reconcile(&HashSet::new());
        // The flush shares the eviction's sender, so its reply confirms the write-back is on disk.
        flush(&world)?;

        // A fresh world over the same directory must stream the chunk back with the edit intact.
        let mut reloaded = test_world(dir.path().to_path_buf());
        drain_to_idle(&mut reloaded, &desired);
        assert!(
            reloaded
                .chunk(pos)
                .is_some_and(|chunk| chunk.blocks[edited_index] == edited_block)
        );
        Ok(())
    }

    #[test]
    fn clean_chunk_is_not_written_back_on_eviction() -> Result<(), SaveError> {
        // An unmodified chunk equals its baseline, so eviction must persist no record for it.
        let dir = tempfile::tempdir()?;
        let pos = ChunkPos::new(0, 0, 0);

        let mut world = test_world(dir.path().to_path_buf());
        let mut desired = HashSet::new();
        desired.insert(pos);
        drain_to_idle(&mut world, &desired);

        // Evict without modifying the chunk, then flush.
        world.reconcile(&HashSet::new());
        flush(&world)?;

        // No record may exist for a chunk that never diverged from its baseline.
        let region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?;
        assert!(region.read_chunk(pos)?.is_none());
        Ok(())
    }

    #[test]
    fn cylinder_contains_expected_columns() {
        let mut set = HashSet::new();
        cylinder_chunks(ChunkPos::new(0, 0, 0), 2, &mut set);

        assert!(set.contains(&ChunkPos::new(0, 0, 0)));
        // A corner cell is outside the disc (dx=2, dz=2 -> 8 > 4).
        assert!(!set.contains(&ChunkPos::new(2, 0, 2)));
        // An axis cell at exactly the radius is included (dx=2, dz=0 -> 4 == 4).
        assert!(set.contains(&ChunkPos::new(2, 0, 0)));
        // The vertical extent is radius/2 = 1, so y=2 is out of range.
        assert!(!set.contains(&ChunkPos::new(0, 2, 0)));
        assert!(set.contains(&ChunkPos::new(0, 1, 0)));
    }

    #[test]
    fn cylinder_translates_with_center() {
        let mut origin = HashSet::new();
        cylinder_chunks(ChunkPos::new(0, 0, 0), 3, &mut origin);

        let mut shifted = HashSet::new();
        cylinder_chunks(ChunkPos::new(10, 0, -5), 3, &mut shifted);

        // The shape is translation-invariant: the same count regardless of center.
        assert_eq!(origin.len(), shifted.len());
    }
}