sectorsync-core 2026.713.0

//! Hotspot metrics and split planning primitives.

use crate::ids::StationId;
use crate::spatial::CellCoord3;

/// Load sample for one spatial cell.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct CellLoadSample {
    /// Cell coordinate.
    pub cell: CellCoord3,
    /// Owned entity count observed in this cell.
    pub owned_entities: usize,
    /// Ghost entity count observed in this cell.
    pub ghost_entities: usize,
    /// Estimated subscribers interested in this cell.
    pub subscribers: usize,
    /// Estimated updates generated by this cell for the current window.
    pub estimated_updates: usize,
    /// Estimated payload bytes generated by this cell for the current window.
    pub estimated_bytes: usize,
    /// Runtime-defined tick cost units for this cell.
    pub tick_cost_units: u64,
    /// Runtime-defined event pressure units for this cell.
    pub event_pressure: usize,
}

impl CellLoadSample {
    /// Returns a deterministic weighted pressure score for ordering cells.
    pub fn pressure_score(self) -> u64 {
        let entities = (self.owned_entities + self.ghost_entities) as u64;
        let subscribers = self.subscribers as u64;
        let updates = self.estimated_updates as u64;
        let bytes = (self.estimated_bytes / 256) as u64;
        let events = self.event_pressure as u64;
        entities
            .saturating_mul(8)
            .saturating_add(subscribers.saturating_mul(4))
            .saturating_add(updates.saturating_mul(2))
            .saturating_add(bytes)
            .saturating_add(events.saturating_mul(16))
            .saturating_add(self.tick_cost_units)
    }
}

/// Load sample for one station over a measurement window.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct StationLoadSample {
    /// Station being measured.
    pub station_id: StationId,
    /// Number of authoritative entities.
    pub owned_entities: usize,
    /// Number of read-only ghost entities.
    pub ghost_entities: usize,
    /// Estimated subscribers routed to this station.
    pub subscribers: usize,
    /// Total queued cross-station events.
    pub queued_events: usize,
    /// Estimated frame bytes generated by this station.
    pub estimated_bytes: usize,
    /// Runtime-defined station tick cost units.
    pub tick_cost_units: u64,
    /// Per-cell load samples.
    pub cells: Vec<CellLoadSample>,
}

impl StationLoadSample {
    /// Returns total entity count.
    pub const fn total_entities(&self) -> usize {
        self.owned_entities + self.ghost_entities
    }

    /// Returns the highest per-cell pressure score.
    pub fn max_cell_pressure(&self) -> u64 {
        self.cells
            .iter()
            .map(|cell| cell.pressure_score())
            .max()
            .unwrap_or(0)
    }
}

/// Thresholds used to classify hotspot pressure.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct HotspotThresholds {
    /// Maximum total entities before a station is considered hot.
    pub max_station_entities: usize,
    /// Maximum subscribers before a station is considered hot.
    pub max_station_subscribers: usize,
    /// Maximum queued events before a station is considered hot.
    pub max_queued_events: usize,
    /// Maximum estimated bytes before a station is considered hot.
    pub max_estimated_bytes: usize,
    /// Maximum tick cost units before a station is considered hot.
    pub max_tick_cost_units: u64,
    /// Maximum pressure score for a single cell before split is suggested.
    pub max_cell_pressure: u64,
}

impl Default for HotspotThresholds {
    fn default() -> Self {
        Self {
            max_station_entities: 50_000,
            max_station_subscribers: 2_000,
            max_queued_events: 10_000,
            max_estimated_bytes: 64 * 1024 * 1024,
            max_tick_cost_units: 16_000,
            max_cell_pressure: 10_000,
        }
    }
}

/// Hotspot classification.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum HotspotSeverity {
    /// Load is within configured limits.
    Normal,
    /// Load is near or slightly over limits.
    Warm,
    /// Load is clearly over limits and should trigger mitigation.
    Hot,
}

/// Hotspot evaluation result.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct HotspotDecision {
    /// Station evaluated by this decision.
    pub station_id: StationId,
    /// Classification severity.
    pub severity: HotspotSeverity,
    /// Weighted pressure score.
    pub score: u64,
    /// Human-readable reason codes.
    pub reasons: Vec<&'static str>,
}

impl Default for HotspotDecision {
    fn default() -> Self {
        Self {
            station_id: StationId::default(),
            severity: HotspotSeverity::Normal,
            score: 0,
            reasons: Vec::new(),
        }
    }
}

/// Cell split proposal for external schedulers.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct SplitProposal {
    /// Source station that should give up some cell ownership.
    pub source_station: StationId,
    /// Cells recommended for movement to a new or less loaded station.
    pub cells_to_move: Vec<CellCoord3>,
    /// Total pressure score covered by `cells_to_move`.
    pub moved_pressure_score: u64,
}

/// Caller-owned reusable candidate storage for hotspot cell split planning.
#[derive(Clone, Debug, Default)]
pub struct HotspotSplitScratch {
    cells: Vec<CellLoadSample>,
}

impl HotspotSplitScratch {
    /// Creates empty split-planning scratch.
    pub const fn new() -> Self {
        Self { cells: Vec::new() }
    }

    /// Cell candidate capacity retained across calls.
    pub fn candidate_capacity(&self) -> usize {
        self.cells.capacity()
    }
}

/// Hotspot evaluator and split planner.
#[derive(Clone, Copy, Debug, Default)]
pub struct HotspotPlanner;

impl HotspotPlanner {
    /// Evaluates a station load sample against thresholds.
    pub fn evaluate(sample: &StationLoadSample, thresholds: HotspotThresholds) -> HotspotDecision {
        let mut decision = HotspotDecision::default();
        Self::evaluate_into(sample, thresholds, &mut decision);
        decision
    }

    /// Evaluates a station into caller-owned output while retaining reason capacity.
    pub fn evaluate_into(
        sample: &StationLoadSample,
        thresholds: HotspotThresholds,
        decision: &mut HotspotDecision,
    ) {
        let mut score = 0_u64;
        decision.station_id = sample.station_id;
        decision.reasons.clear();

        if sample.total_entities() > thresholds.max_station_entities {
            score = score.saturating_add(1);
            decision.reasons.push("station_entities");
        }
        if sample.subscribers > thresholds.max_station_subscribers {
            score = score.saturating_add(1);
            decision.reasons.push("station_subscribers");
        }
        if sample.queued_events > thresholds.max_queued_events {
            score = score.saturating_add(1);
            decision.reasons.push("queued_events");
        }
        if sample.estimated_bytes > thresholds.max_estimated_bytes {
            score = score.saturating_add(1);
            decision.reasons.push("estimated_bytes");
        }
        if sample.tick_cost_units > thresholds.max_tick_cost_units {
            score = score.saturating_add(1);
            decision.reasons.push("tick_cost");
        }
        if sample.max_cell_pressure() > thresholds.max_cell_pressure {
            score = score.saturating_add(1);
            decision.reasons.push("cell_pressure");
        }

        decision.severity = match score {
            0 => HotspotSeverity::Normal,
            1 => HotspotSeverity::Warm,
            _ => HotspotSeverity::Hot,
        };
        decision.score = score;
    }

    /// Proposes cells to move by selecting the highest-pressure cells first.
    pub fn propose_cell_split(
        sample: &StationLoadSample,
        max_cells_to_move: usize,
    ) -> SplitProposal {
        let mut scratch = HotspotSplitScratch::new();
        let mut proposal = SplitProposal::default();
        Self::propose_cell_split_into(sample, max_cells_to_move, &mut scratch, &mut proposal);
        proposal
    }

    /// Proposes cells using caller-owned candidate scratch.
    pub fn propose_cell_split_with_scratch(
        sample: &StationLoadSample,
        max_cells_to_move: usize,
        scratch: &mut HotspotSplitScratch,
    ) -> SplitProposal {
        let mut proposal = SplitProposal::default();
        Self::propose_cell_split_into(sample, max_cells_to_move, scratch, &mut proposal);
        proposal
    }

    /// Writes a deterministic split proposal into fully reusable caller-owned storage.
    pub fn propose_cell_split_into(
        sample: &StationLoadSample,
        max_cells_to_move: usize,
        scratch: &mut HotspotSplitScratch,
        proposal: &mut SplitProposal,
    ) {
        let selected = max_cells_to_move.min(sample.cells.len());
        scratch.cells.clear();
        scratch.cells.extend_from_slice(&sample.cells);
        prioritize_cell_samples(&mut scratch.cells, selected);

        proposal.source_station = sample.station_id;
        proposal.cells_to_move.clear();
        proposal.cells_to_move.reserve(selected);
        proposal.moved_pressure_score = 0;
        for cell in &scratch.cells[..selected] {
            proposal.moved_pressure_score = proposal
                .moved_pressure_score
                .saturating_add(cell.pressure_score());
            proposal.cells_to_move.push(cell.cell);
        }
    }
}

fn compare_cell_samples(left: &CellLoadSample, right: &CellLoadSample) -> core::cmp::Ordering {
    right
        .pressure_score()
        .cmp(&left.pressure_score())
        .then_with(|| left.cell.cmp(&right.cell))
}

fn prioritize_cell_samples(cells: &mut [CellLoadSample], selected: usize) {
    if selected == 0 {
        return;
    }
    if selected.saturating_mul(2) < cells.len() {
        cells.select_nth_unstable_by(selected, compare_cell_samples);
        cells[..selected].sort_by(compare_cell_samples);
    } else {
        cells.sort_by(compare_cell_samples);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn planner_detects_hot_station_and_suggests_hottest_cells() {
        let sample = StationLoadSample {
            station_id: StationId::new(3),
            owned_entities: 100,
            subscribers: 20,
            cells: vec![
                CellLoadSample {
                    cell: CellCoord3::new(0, 0, 0),
                    owned_entities: 1,
                    ..CellLoadSample::default()
                },
                CellLoadSample {
                    cell: CellCoord3::new(1, 0, 0),
                    owned_entities: 100,
                    event_pressure: 50,
                    ..CellLoadSample::default()
                },
            ],
            ..StationLoadSample::default()
        };
        let thresholds = HotspotThresholds {
            max_station_entities: 50,
            max_cell_pressure: 10,
            ..HotspotThresholds::default()
        };

        let decision = HotspotPlanner::evaluate(&sample, thresholds);
        assert_eq!(decision.severity, HotspotSeverity::Hot);

        let mut reused_decision = HotspotDecision::default();
        HotspotPlanner::evaluate_into(&sample, thresholds, &mut reused_decision);
        assert_eq!(reused_decision, decision);
        let reason_capacity = reused_decision.reasons.capacity();
        HotspotPlanner::evaluate_into(
            &StationLoadSample {
                station_id: StationId::new(4),
                ..StationLoadSample::default()
            },
            thresholds,
            &mut reused_decision,
        );
        assert_eq!(reused_decision.station_id, StationId::new(4));
        assert_eq!(reused_decision.severity, HotspotSeverity::Normal);
        assert!(reused_decision.reasons.is_empty());
        assert_eq!(reused_decision.reasons.capacity(), reason_capacity);

        let proposal = HotspotPlanner::propose_cell_split(&sample, 1);
        assert_eq!(proposal.cells_to_move, vec![CellCoord3::new(1, 0, 0)]);
    }

    #[test]
    fn split_top_k_matches_full_sort_and_reuses_capacity_at_budget_edges() {
        let cells = (0_i32..257)
            .map(|index| CellLoadSample {
                cell: CellCoord3::new(index, index % 5, index % 7),
                owned_entities: usize::try_from(index * 37 % 23).expect("non-negative"),
                event_pressure: usize::try_from(index * 19 % 11).expect("non-negative"),
                ..CellLoadSample::default()
            })
            .collect::<Vec<_>>();
        let sample = StationLoadSample {
            station_id: StationId::new(9),
            cells: cells.clone(),
            ..StationLoadSample::default()
        };
        let mut scratch = HotspotSplitScratch::new();
        let mut proposal = SplitProposal::default();

        for requested in [0, 1, 7, 64, 128, 129, 256, 257, 300] {
            let selected = requested.min(cells.len());
            let mut expected = cells.clone();
            expected.sort_by(compare_cell_samples);
            expected.truncate(selected);
            HotspotPlanner::propose_cell_split_into(
                &sample,
                requested,
                &mut scratch,
                &mut proposal,
            );

            assert_eq!(
                proposal.cells_to_move,
                expected.iter().map(|cell| cell.cell).collect::<Vec<_>>()
            );
            assert_eq!(
                proposal.moved_pressure_score,
                expected.iter().fold(0_u64, |score, cell| {
                    score.saturating_add(cell.pressure_score())
                })
            );
        }
        let candidate_capacity = scratch.candidate_capacity();
        let output_capacity = proposal.cells_to_move.capacity();
        HotspotPlanner::propose_cell_split_into(
            &StationLoadSample {
                station_id: StationId::new(10),
                cells: cells[..8].to_vec(),
                ..StationLoadSample::default()
            },
            2,
            &mut scratch,
            &mut proposal,
        );
        assert_eq!(proposal.source_station, StationId::new(10));
        assert_eq!(scratch.candidate_capacity(), candidate_capacity);
        assert_eq!(proposal.cells_to_move.capacity(), output_capacity);
    }
}