use std::fmt::Debug;
use std::fs::OpenOptions;
use std::marker::PhantomData;
use std::mem::size_of;
use std::path::Path;
use std::{io, mem, usize};

use anyhow::bail;

// TODO: Remove anyhow dependency
// TODO: Review unsafe code

#[derive(Debug)]
struct Table<T> {
    mmap: memmap2::MmapRaw,
    height: usize,
    phantom: PhantomData<T>,
}

impl<T: Clone> Table<T> {
    unsafe fn create<P: AsRef<Path>>(path: P, height: usize) -> anyhow::Result<Self> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create_new(true)
            .open(path)?;
        let width = size_of::<T>();
        let expected_size = height * width;
        file.set_len(expected_size as u64)?;
        Ok(Self {
            mmap: memmap2::MmapRaw::map_raw(&file)?,
            height,
            phantom: PhantomData,
        })
    }

    unsafe fn resume<P: AsRef<Path>>(path: P, height: usize) -> anyhow::Result<Self> {
        let file = OpenOptions::new().read(true).write(true).open(path)?;
        let width = size_of::<T>();
        let expected_size = height * width;
        let actual_size = file.metadata()?.len() as usize;
        if actual_size != expected_size {
            bail!("Expected file with size {expected_size}={height}*{width} but got {actual_size}");
        }
        Ok(Self {
            mmap: memmap2::MmapRaw::map_raw(&file)?,
            height,
            phantom: PhantomData,
        })
    }

    unsafe fn get_unchecked(&self, index: usize) -> &T {
        // Docs say that for `as_ref` to be safe the target must be initialized.
        // If any bit sequence of the correct length is a valid T
        // then can it always be considered initialized?
        (self.mmap.as_ptr() as *const T)
            .add(index)
            .as_ref()
            .unwrap()
    }
    unsafe fn get_unchecked_mut(&mut self, index: usize) -> &mut T {
        // Docs say that for `as_mut` to be safe the target must be initialized.
        // If any bit sequence of the correct length is a valid T
        // then can it always be considered initialized?
        (self.mmap.as_ptr() as *mut T).add(index).as_mut().unwrap()
    }

    fn len(&self) -> usize {
        self.height
    }
}

type Key = usize;

#[derive(Debug)]
struct Map<T: Clone + Debug> {
    table: Table<(usize, T)>,
}

impl<T: Clone + Debug> Map<T> {
    fn new(table: Table<(usize, T)>) -> Self {
        Self { table }
    }

    fn index_checksum(&self, key: &Key) -> (usize, usize) {
        let len = self.table.len();
        (key % len, key / len + 1)
    }

    unsafe fn get(&self, key: Key) -> Option<&T> {
        let (index, expected_checksum) = self.index_checksum(&key);
        let (actual_checksum, old) = self.table.get_unchecked(index);
        if *actual_checksum == expected_checksum {
            Some(old)
        } else {
            None
        }
    }

    unsafe fn insert(&mut self, key: Key, new: T) -> Option<T> {
        let (index, expected_checksum) = self.index_checksum(&key);
        let (actual_checksum, old) = mem::replace(
            self.table.get_unchecked_mut(index),
            (expected_checksum, new),
        );
        if actual_checksum == expected_checksum {
            Some(old)
        } else {
            None
        }
    }

    unsafe fn get_mut(&mut self, key: Key) -> Option<&mut T> {
        let (index, expected_checksum) = self.index_checksum(&key);
        let (actual_checksum, old) = self.table.get_unchecked_mut(index);
        if *actual_checksum == expected_checksum {
            Some(old)
        } else {
            None
        }
    }

    unsafe fn sparse(&self) -> Vec<(Key, &T)> {
        let len = self.table.len();
        let mut items = Vec::with_capacity(len);
        for i in 0..self.table.len() {
            let (c, v) = self.table.get_unchecked(i);
            if *c == 0 {
                continue;
            }
            items.push(((c - 1) * len + i, v));
        }
        items
    }
}

type Instant = i64;
type Duration = i64;

pub trait Aggregate {
    type Marginal;
    fn combine(&mut self, other: &Self);
    fn update(&mut self, other: &Self::Marginal);
    fn empty(last: &Self::Marginal) -> Self;
    fn initial(curr: &Self::Marginal) -> Self;
}

/// # Safety
///
/// Most methods are marked as safe to make them ergonomic to use but they are not generally safe
/// for any `T`.
/// It is up to the caller of the unsafe constructor methods to ensure that the all methods on the
/// resulting instances are safe to call.
///
/// I think instances are safe iff any sequence of `size_of::<T>()` bits are a valid `T`.
///
/// See the safety section for [`memmap2::Mmap`] for more information about the (un)safety of
/// memory-mapped files.
#[derive(Debug)]
pub struct BinnedTimeSeries<T: Clone + Debug, U> {
    bins: Map<T>,
    bin_size: Duration,
    prev: Option<(Instant, U)>,
}

impl<T: Clone + Debug + Aggregate<Marginal = U>, U: Clone> BinnedTimeSeries<T, U> {
    /// Create a new `BinnedTimeSeries`
    ///
    /// # Safety
    ///
    /// See safety section for [`BinnedTimeSeries`].
    pub unsafe fn create_unsafe<P: AsRef<Path>>(
        path: P,
        num_bin: usize,
        bin_size: Duration,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            bins: Map::new(Table::create(path, num_bin)?),
            bin_size,
            prev: None,
        })
    }

    /// Open an existing `BinnedTimeSeries`
    ///
    /// # Safety
    ///
    /// See safety section for [`BinnedTimeSeries`].
    pub unsafe fn resume_unsafe<P: AsRef<Path>>(
        path: P,
        num_bin: usize,
        bin_size: Duration,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            bins: Map::new(Table::resume(path, num_bin)?),
            bin_size,
            prev: None,
        })
    }

    /// Update the aggregates to reflect a new data point
    pub fn update(&mut self, now: Instant, marginal: U) {
        // TODO: Check that time is non-decreasing
        // Safety: See comment on struct block
        unsafe {
            // Create skipped bins
            let curr_key = now / self.bin_size;
            if let Some((prev_key, prev_value)) = &self.prev {
                for key in *prev_key..curr_key {
                    self.bins.insert((key + 1) as Key, T::empty(prev_value));
                }
            }

            // Update current bin
            // TODO: Implement something like `entry`
            if let Some(aggregate) = self.bins.get_mut(curr_key as Key) {
                aggregate.update(&marginal);
            } else {
                let aggregate = T::initial(&marginal);
                self.bins.insert(curr_key as Key, aggregate);
            }
            self.prev = Some((curr_key, marginal));
        };
    }

    /// Return all bins
    pub fn sparse(&self) -> Vec<(Instant, &T)> {
        // Safety: See comment on struct block
        unsafe {
            let mut items: Vec<(_, _)> = self
                .bins
                .sparse()
                .into_iter()
                .map(|(k, v)| (k as Instant * self.bin_size, v))
                .collect();
            items.sort_by_key(|(k, _)| *k);
            items
        }
    }

    /// Return the aggregate of zero or more bins
    ///
    /// Missing bins are ignored by the aggregation and `None` is returned only iff no bins were
    /// found.
    pub fn aggregated(&self, first: Instant, last: Instant) -> Option<T> {
        // Safety: See comment on struct block
        unsafe {
            let first = first / self.bin_size;
            let last = last / self.bin_size;
            let mut bins = (first..=last).flat_map(|i| self.bins.get(i as Key));
            if let Some(first) = bins.next() {
                let mut result = first.clone();
                for other in bins {
                    result.combine(other);
                }
                Some(result)
            } else {
                None
            }
        }
    }

    /// See [`memmap2::MmapRaw::flush`]
    pub fn flush(&self) -> io::Result<()> {
        self.bins.table.mmap.flush()
    }
}
