downsample 0.0.6

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.
///
/// By default, the final level has `aggregation_window == 0`, so all of its
/// slots are readable samples and it never completes a terminal window.
/// Terminal levels can choose an `aggregation_window > 0`; completed terminal
/// windows are returned by `FixedWriter::push_overflow` instead of being
/// reduced inside this writer.
#[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 in the hidden aggregation window.
    ///
    /// For non-terminal levels this many samples are reduced into one value for
    /// the next level. For terminal levels configured with
    /// `FixedFrequencyBuilder::terminal`, this many samples are returned as a
    /// completed window by `FixedWriter::push_overflow`.
    ///
    /// 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>>,
}

enum PushResult<T> {
    None,
    Aggregate(T),
    Window(usize),
}

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) -> PushResult<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 level.aggregation_window > 0 && level.next == level.aggregation_start {
                aggregate = Some((
                    level.aggregation_start,
                    level.overflowing_add(level.aggregation_start, level.aggregation_window),
                ));
            }
        }

        let Some((start, next_aggregation_start)) = aggregate else {
            return PushResult::None;
        };

        self.storage.levels[level_index].aggregation_start = next_aggregation_start;

        if let Some(reducer) = self.reducers.get(level_index) {
            let (first, second) = self
                .storage
                .aggregation_slices(&self.storage.levels[level_index], start);
            PushResult::Aggregate(reducer.reduce(first, second))
        } else {
            PushResult::Window(start)
        }
    }

    /// Adds a value to the storage and cascades completed aggregation windows.
    ///
    /// If the terminal aggregation window fills, returns it as up to two
    /// slices.
    pub fn push_overflow(&mut self, mut value: T) -> Option<(&[T], &[T])> {
        let mut level = 0;

        loop {
            match self.push_level(value, level) {
                PushResult::None => return None,
                PushResult::Aggregate(overflow) => {
                    value = overflow;
                    level += 1;
                    debug_assert!(level < self.storage.levels.len());
                },
                PushResult::Window(start) => {
                    return Some(self.storage.aggregation_slices(&self.storage.levels[level], start));
                },
            }
        }
    }

    /// Adds a value to the storage and ignores completed terminal windows.
    #[inline(always)]
    pub fn push(&mut self, value: T) { let _ = self.push_overflow(value); }
}

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

    const fn nz(value: usize) -> NonZeroUsize { NonZeroUsize::new(value).expect("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 push_overflow_window_returns_none_without_terminal_aggregation() {
        let mut s = storage();
        for value in 1..=9 {
            assert!(s.push_overflow(value as f32).is_none());
        }

        let aggregates: Vec<_> = s.storage.level(1).unwrap().iter_oldest_first().copied().collect();
        assert_eq!(aggregates, vec![3.5, 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 unordered_raw_level_data_returns_complete_level_storage() {
        let mut s = storage();
        for value in 1..=9 {
            s.push(value as f32);
        }

        assert_eq!(
            s.storage.get_level_data_unordered_raw(0),
            Some(&[6.0, 7.0, 8.0, 9.0, 5.0][..])
        );
        assert_eq!(s.storage.get_level_data_unordered_raw(1), Some(&[5.5, 3.5][..]));
    }

    #[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_overflow_window_returns_terminal_batches() {
        let mut s = crate::FixedFrequencyBuilder::new(nz(2))
            .level(nz(2), nz(2), reducers::average_u32())
            .terminal(nz(2))
            .build();

        let mut windows = Vec::new();
        for value in 1..=18 {
            if let Some((first, second)) = s.push_overflow(value) {
                windows.push(first.iter().chain(second).copied().collect::<Vec<_>>());
            }
        }

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

    #[test]
    fn terminal_window_can_feed_multiple_child_writers_with_different_reducers() {
        let mut parent = crate::FixedFrequencyBuilder::new(nz(2))
            .level(nz(2), nz(2), reducers::average_u32())
            .terminal(nz(2))
            .build();
        let mut child_c = crate::FixedFrequencyBuilder::new(nz(3)).build();
        let mut child_d = crate::FixedFrequencyBuilder::new(nz(3)).build();
        let c_reducer = reducers::average_u32();
        let d_reducer = reducers::max::<u32>();

        for value in 1..=18 {
            if let Some((first, second)) = parent.push_overflow(value) {
                child_c.push(c_reducer.reduce(first, second));
                child_d.push(d_reducer.reduce(first, second));
            }
        }

        let c_values: Vec<_> = child_c.storage.level(0).unwrap().iter_oldest_first().copied().collect();
        let d_values: Vec<_> = child_d.storage.level(0).unwrap().iter_oldest_first().copied().collect();

        assert_eq!(c_values, vec![2, 6, 10]);
        assert_eq!(d_values, vec![3, 7, 11]);
    }

    #[test]
    fn split_terminal_storage_matches_single_storage_for_many_pushes() {
        let mut single = crate::FixedFrequencyBuilder::new(nz(5))
            .level(nz(7), nz(3), reducers::average_u32())
            .level(nz(6), nz(4), reducers::median::<u32>())
            .build();
        let mut parent = crate::FixedFrequencyBuilder::new(nz(5))
            .level(nz(7), nz(3), reducers::average_u32())
            .terminal(nz(4))
            .build();
        let mut child = crate::FixedFrequencyBuilder::new(nz(6)).build();
        let child_reducer = reducers::median::<u32>();

        for index in 0..1_000 {
            let value = (index * 37 + 11) % 997;

            single.push(value);
            if let Some((first, second)) = parent.push_overflow(value) {
                child.push(child_reducer.reduce(first, second));
            }

            let single_middle: Vec<_> = single.storage.level(1).unwrap().iter_oldest_first().copied().collect();
            let parent_terminal: Vec<_> = parent.storage.level(1).unwrap().iter_oldest_first().copied().collect();
            let single_final: Vec<_> = single.storage.level(2).unwrap().iter_oldest_first().copied().collect();
            let child_values: Vec<_> = child.storage.level(0).unwrap().iter_oldest_first().copied().collect();

            assert_eq!(single_middle, parent_terminal);
            assert_eq!(single_final, child_values);
        }
    }

    #[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]);
    }
}

#[cfg(test)]
mod custom_reducer_tests {
    use super::*;
    use core::num::NonZeroUsize;

    const fn nz(value: usize) -> NonZeroUsize { NonZeroUsize::new(value).expect("expected non-zero value") }

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

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

        for value in 1..=8 {
            writer.push(value);
        }

        assert_eq!(writer.storage().get_back_by_level(1, 0), Some(&7));
        assert_eq!(writer.storage().get_back_by_level(1, 1), Some(&11));
    }
}