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

//! Per-connection chunk-streaming state.

use std::collections::HashSet;

use net::ChunkSink;
use shared::protocol::chunk::ChunkMessage;
use shared::world::{Chunk, ChunkData, ChunkPos};

use crate::world_server::{ServerWorld, cylinder_chunks};

/// Upper bound, in chunks, on a client's requested load radius. A larger request is clamped to this, bounding the per-client resident set and the reconcile cost the server performs on the client's behalf.
// TODO: derive from server configuration and per-tier LOD limits.
pub const SERVER_MAX_RADIUS: u16 = 12;

/// Worldgen version stamped on delivered chunk diffs. A single version exists today; this becomes the chunk's stored version once worldgen versioning lands.
const WORLDGEN_VERSION: u32 = 0;

/// The load and drop lists produced by diffing a client's previous desired set against a new one.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct DesiredDiff {
    /// Positions newly wanted (present in the new set, absent from the previous). Delivered once resident.
    pub added: Vec<ChunkPos>,
    /// Positions no longer wanted (present in the previous set, absent from the new). The client is told to drop each it holds.
    pub removed: Vec<ChunkPos>,
}

/// Computes the load/drop diff between a client's `previous` and `new` desired sets.
#[must_use]
pub fn desired_diff<S: std::hash::BuildHasher>(
    previous: &HashSet<ChunkPos, S>,
    new: &HashSet<ChunkPos, S>,
) -> DesiredDiff {
    let mut added: Vec<ChunkPos> = new.difference(previous).copied().collect();
    let mut removed: Vec<ChunkPos> = previous.difference(new).copied().collect();
    added.sort_unstable();
    removed.sort_unstable();
    DesiredDiff { added, removed }
}

/// Tracks one connected client's chunk subscription and what has been delivered to it.
pub struct ClientStream {
    /// Outbound handle onto the client's chunk stream.
    sink: ChunkSink,
    /// The chunk positions the client currently wants resident, already clamped to [`SERVER_MAX_RADIUS`].
    desired: HashSet<ChunkPos>,
    /// Positions already delivered to the client as [`ChunkMessage::Chunk`].
    sent: HashSet<ChunkPos>,
}

impl ClientStream {
    /// Creates a stream for a freshly connected client that has not yet subscribed.
    #[must_use]
    pub fn new(sink: ChunkSink) -> Self {
        Self {
            sink,
            desired: HashSet::new(),
            sent: HashSet::new(),
        }
    }

    /// Returns the client's current desired set, for folding into the world reconcile union.
    #[must_use]
    pub fn desired(&self) -> &HashSet<ChunkPos> {
        &self.desired
    }

    /// Applies a new subscription centered on `center` with load radius `radius`.
    ///
    /// The radius is clamped to [`SERVER_MAX_RADIUS`], the desired set is recomputed, and a [`ChunkMessage::Drop`] is emitted for every already-delivered chunk that left the set. Chunks newly entering the set are not sent here; they are delivered by [`ClientStream::flush`] once resident. Returns the number of newly-wanted positions and the number of drops emitted.
    pub fn resubscribe(&mut self, center: ChunkPos, radius: u16) -> (usize, usize) {
        let clamped = radius.min(SERVER_MAX_RADIUS);
        let mut new_desired = HashSet::new();
        cylinder_chunks(center, i32::from(clamped), &mut new_desired);

        let diff = desired_diff(&self.desired, &new_desired);
        let added = diff.added.len();
        let mut drops = 0;
        for pos in diff.removed {
            // Only chunks actually delivered need an explicit drop; positions that were wanted but never resident were never held by the client.
            if self.sent.remove(&pos) {
                self.sink.send(ChunkMessage::Drop { pos });
                drops += 1;
            }
        }
        self.desired = new_desired;
        (added, drops)
    }

    /// Delivers every desired-but-undelivered chunk that has become resident in `world`.
    ///
    /// Each chunk is encoded as a [`ChunkData`] diff against `baseline` (an all-air chunk), making the payload self-contained. Positions still pending in the worker pool are skipped and retried on a later call. Returns the number of chunks delivered.
    pub fn flush(&mut self, world: &ServerWorld, baseline: &Chunk) -> usize {
        // Collected first to avoid borrowing `self.desired` while mutating `self.sent`.
        let ready: Vec<ChunkPos> = self
            .desired
            .iter()
            .filter(|pos| !self.sent.contains(pos))
            .copied()
            .collect();

        let mut delivered = 0;
        for pos in ready {
            let Some(chunk) = world.chunk(pos) else {
                // Not resident yet; a later flush retries once the worker pool returns it.
                continue;
            };
            let data = ChunkData::from_diff(pos, WORLDGEN_VERSION, baseline, chunk);
            self.sink.send(ChunkMessage::Chunk { pos, data });
            self.sent.insert(pos);
            delivered += 1;
        }
        delivered
    }
}

#[cfg(test)]
#[path = "tests/client_stream.rs"]
mod tests;