downsample 0.0.5

use crate::{Reducer, fixed_frequency::iter::LevelView};
use alloc::vec::Vec;

/// Internal ring-buffer layout for one fixed-frequency level.
///
/// Each level owns one contiguous range inside `FixedStorage::data`:
///
/// ```text
/// data = [ level 0 slots ][ level 1 slots ][ level 2 slots ]
///        start0..end0     start1..end1     start2..end2
/// ```
///
/// Inside one level, the range is used as a ring buffer. The oldest internal
/// slot is also the next write position:
///
/// ```text
/// start                                      end
///   |                                        |
///   v                                        v
/// [ slot ][ slot ][ slot ][ slot ][ slot ]
///             ^
///             next
/// ```
///
/// The first `aggregation_window` slots after `next` are hidden from normal
/// reads. They form the batch that will be reduced and pushed to the next
/// level once full. Readable history starts after that batch:
///
/// ```text
/// next
///  |
///  v
/// [ aggregation window ][ readable samples ... ]
/// ```
///
/// Because the storage is a ring, these regions can wrap around `end` back to
/// `start`. `overflowing_add` handles that wrapping.
///
/// The final level has `aggregation_window == 0`, so all of its slots are
/// readable samples and it never produces another aggregate.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LevelMetadata {
    /// Inclusive start index of this level inside `FixedStorage::data`.
    start: usize,
    /// Exclusive end index of this level inside `FixedStorage::data`.
    end: usize,
    /// Number of samples reduced into one value for the next level.
    ///
    /// This many slots after `next` are excluded from normal reads.
    aggregation_window: usize,
    /// Start index of the current aggregation batch.
    ///
    /// When `next` reaches this index, the batch is full and can be reduced.
    aggregation_start: usize,
    /// Ring-buffer write cursor for next pushed element
    next: usize,
}

impl LevelMetadata {
    pub(crate) fn new(start: usize, end: usize, aggregation_window: usize) -> Self {
        assert!(start < end, "level storage must not be empty");
        assert!(
            end - start >= aggregation_window,
            "level storage must hold at least one aggregation window"
        );
        Self {
            start,
            end,
            aggregation_window,
            aggregation_start: start,
            next: start,
        }
    }

    pub fn storage_capacity(&self) -> usize { self.end - self.start }

    pub fn capacity(&self) -> usize { self.sample_capacity() }

    pub fn aggregation_window(&self) -> usize { self.aggregation_window }

    pub(crate) fn sample_capacity(&self) -> usize { self.storage_capacity() - self.aggregation_window }

    pub(crate) fn readable_position(&self, index: usize) -> usize {
        self.overflowing_add(self.next, index + self.aggregation_window)
    }

    #[inline(always)]
    fn overflowing_add(&self, index: usize, offset: usize) -> usize {
        debug_assert!(index >= self.start && index < self.end);
        debug_assert!(offset < self.storage_capacity());

        let result = index + offset;
        if result >= self.end {
            result + self.start - self.end
        } else {
            result
        }
    }
}

pub struct FixedStorage<T> {
    pub(crate) data:   Vec<T>,
    pub(crate) levels: Vec<LevelMetadata>,
}

impl<T> FixedStorage<T> {
    pub fn level_count(&self) -> usize { self.levels.len() }

    pub fn level_metadata(&self, level: usize) -> Option<&LevelMetadata> { self.levels.get(level) }

    pub fn level(&self, level: usize) -> Option<LevelView<'_, T>> {
        self.levels.get(level).map(|level| LevelView::new(&self.data, level))
    }

    /// Returns the complete internal storage for a level without time ordering.
    ///
    /// This includes the aggregation window that is hidden by `level()` and
    /// `get_back_by_level()`.
    pub fn get_level_data_unordered_raw(&self, level: usize) -> Option<&[T]> {
        let level = self.levels.get(level)?;
        Some(&self.data[level.start..level.end])
    }

    /// Returns elements by level, where index 0 is the oldest readable sample.
    pub fn get_back_by_level(&self, level: usize, index: usize) -> Option<&T> { self.level(level)?.get(index) }

    /// Returns elements by level, where index 0 is the oldest internal slot.
    ///
    /// This includes the aggregation window that is normally hidden.
    pub fn get_back_by_level_raw(&self, level: usize, index: usize) -> Option<&T> {
        let level = self.levels.get(level)?;
        if index >= level.storage_capacity() {
            return None;
        }

        let pos = level.overflowing_add(level.next, index);
        Some(&self.data[pos])
    }

    #[inline(always)]
    /// Returns the current aggregation batch as one or two slices into the
    /// internal Vec<T>.
    fn aggregation_slices(&self, level: &LevelMetadata, start: usize) -> (&[T], &[T]) {
        let end = start + level.aggregation_window;
        if end <= level.end {
            (&self.data[start..end], &[])
        } else {
            let overflow = end - level.end;
            (
                &self.data[start..level.end],
                &self.data[level.start..level.start + overflow],
            )
        }
    }
}

pub struct FixedWriter<T> {
    pub(crate) storage:  FixedStorage<T>,
    pub(crate) reducers: Vec<Reducer<T>>,
}

impl<T> FixedWriter<T> {
    pub fn storage(&self) -> &FixedStorage<T> { &self.storage }

    pub fn into_storage(self) -> FixedStorage<T> { self.storage }

    #[inline(always)]
    fn push_level(&mut self, mut value: T, level_index: usize) -> Option<T> {
        let mut aggregate = None;

        {
            let level = &mut self.storage.levels[level_index];
            core::mem::swap(&mut self.storage.data[level.next], &mut value);
            level.next += 1;
            if level.next == level.end {
                level.next = level.start;
            }

            if self.reducers.get(level_index).is_some() && level.next == level.aggregation_start {
                aggregate = Some((
                    level.aggregation_start,
                    level.overflowing_add(level.aggregation_start, level.aggregation_window),
                ));
            }
        }

        let reducer = self.reducers.get(level_index).copied()?;
        let (start, next_aggregation_start) = aggregate?;
        self.storage.levels[level_index].aggregation_start = next_aggregation_start;
        let (first, second) = self
            .storage
            .aggregation_slices(&self.storage.levels[level_index], start);
        Some(reducer.reduce(first, second))
    }

    pub fn push(&mut self, mut value: T) {
        let mut level = 0;
        while let Some(overflow) = self.push_level(value, level) {
            value = overflow;
            level += 1;
            if level == self.storage.levels.len() {
                break;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::reducers;
    use alloc::vec;
    use core::num::NonZeroUsize;

    const fn nz(value: usize) -> NonZeroUsize {
        match NonZeroUsize::new(value) {
            Some(value) => value,
            None => panic!("expected non-zero value"),
        }
    }

    fn storage() -> FixedWriter<f32> {
        crate::FixedFrequencyBuilder::new(nz(3))
            .level(nz(2), nz(2), reducers::average_f32())
            .build()
    }

    fn sum_u32(first: &[u32], second: &[u32]) -> u32 { first.iter().chain(second).copied().sum() }

    #[test]
    fn construct_simple() {
        let s = storage();
        assert_eq!(s.storage.level_count(), 2);
        let raw = s.storage.level_metadata(0).unwrap();
        assert_eq!(raw.capacity(), 3);
        assert_eq!(raw.aggregation_window(), 2);
        assert_eq!(raw.storage_capacity(), 5);

        let next = s.storage.level_metadata(1).unwrap();
        assert_eq!(next.capacity(), 2);
        assert_eq!(next.aggregation_window(), 0);
        assert_eq!(next.storage_capacity(), 2);
    }

    #[test]
    fn push_aggregates_into_next_level() {
        let mut s = storage();
        for value in 1..=9 {
            s.push(value as f32);
        }

        assert_eq!(s.storage.data, vec![6.0, 7.0, 8.0, 9.0, 5.0, 5.5, 3.5]);
        assert_eq!(s.storage.get_back_by_level(0, 0), Some(&7.0));
        assert_eq!(s.storage.get_back_by_level(0, 1), Some(&8.0));
        assert_eq!(s.storage.get_back_by_level(0, 2), Some(&9.0));
        assert_eq!(s.storage.get_back_by_level(0, 3), None);
        assert_eq!(s.storage.get_back_by_level(1, 0), Some(&3.5));
        assert_eq!(s.storage.get_back_by_level(1, 1), Some(&5.5));
    }

    #[test]
    fn raw_reads_include_hidden_aggregation_window() {
        let mut s = storage();
        for value in 1..=9 {
            s.push(value as f32);
        }

        let raw: Vec<_> = (0..5)
            .map(|index| *s.storage.get_back_by_level_raw(0, index).unwrap())
            .collect();
        let readable: Vec<_> = s.storage.level(0).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(raw, vec![5.0, 6.0, 7.0, 8.0, 9.0]);
        assert_eq!(readable, vec![7.0, 8.0, 9.0]);
    }

    #[test]
    fn reads_return_none_for_invalid_levels_and_indexes() {
        let mut s = storage();
        for value in 1..=9 {
            s.push(value as f32);
        }

        assert!(s.storage.level(2).is_none());
        assert!(s.storage.level_metadata(2).is_none());
        assert!(s.storage.get_level_data_unordered_raw(2).is_none());
        assert_eq!(s.storage.get_back_by_level(2, 0), None);
        assert_eq!(s.storage.get_back_by_level(0, 3), None);
        assert_eq!(s.storage.get_back_by_level_raw(0, 5), None);
    }

    #[test]
    fn push_cascades_exact_aggregates_through_multiple_levels() {
        let mut s = crate::FixedFrequencyBuilder::new(nz(2))
            .level(nz(2), nz(2), reducers::average_u32())
            .level(nz(2), nz(2), reducers::average_u32())
            .build();

        for value in 1..=20 {
            s.push(value);
        }

        let level_1: Vec<_> = s.storage.level(1).unwrap().iter_oldest_first().copied().collect();
        let level_2: Vec<_> = s.storage.level(2).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(level_1, vec![15, 17]);
        assert_eq!(level_2, vec![6, 10]);
    }

    #[test]
    fn push_reduces_wrapped_aggregation_window() {
        let mut s = crate::FixedFrequencyBuilder::new(nz(3))
            .level(nz(3), nz(2), Reducer::new(sum_u32))
            .build();

        for value in 1..=9 {
            s.push(value);
        }

        let aggregates: Vec<_> = s.storage.level(1).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(aggregates, vec![3, 7, 11]);
    }

    #[test]
    fn median_f32_ignores_nan_in_wrapped_aggregation_window() {
        let mut s = crate::FixedFrequencyBuilder::new(nz(3))
            .level(nz(3), nz(2), reducers::median_f32())
            .build();

        for value in [1.0, 2.0, 3.0, 4.0, f32::NAN, 6.0, 7.0, 8.0, 9.0] {
            s.push(value);
        }

        let aggregates: Vec<_> = s.storage.level(1).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(aggregates, vec![1.0, 3.0, 6.0]);
    }

    #[test]
    fn push_uses_each_levels_configured_reducer() {
        let mut s = crate::FixedFrequencyBuilder::new(nz(2))
            .level(nz(2), nz(2), reducers::average_u32())
            .level(nz(2), nz(2), reducers::max::<u32>())
            .build();

        for value in 1..=20 {
            s.push(value);
        }

        let level_1: Vec<_> = s.storage.level(1).unwrap().iter_oldest_first().copied().collect();
        let level_2: Vec<_> = s.storage.level(2).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(level_1, vec![15, 17]);
        assert_eq!(level_2, vec![7, 11]);
    }

    #[test]
    fn final_level_does_not_aggregate_or_divide_by_zero() {
        let mut s = crate::FixedFrequencyBuilder::new(nz(2))
            .level(nz(2), nz(1), reducers::average_u32())
            .build();

        for value in 0..20 {
            s.push(value);
        }

        assert_eq!(s.storage.level_count(), 2);
    }

    #[test]
    fn view_iterates_oldest_first() {
        let mut s = storage();
        for value in 1..=9 {
            s.push(value as f32);
        }

        let values: Vec<_> = s.storage.level(0).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(values, vec![7.0, 8.0, 9.0]);
    }
}