// StableBloomFilter implements a Stable Bloom Filter as described by Deng and
// Rafiei in Approximately Detecting Duplicates for Streaming Data using Stable
// Bloom Filters:
//
// http://webdocs.cs.ualberta.ca/~drafiei/papers/DupDet06Sigmod.pdf
//
// A Stable Bloom Filter (SBF) continuously evicts stale information so that it
// has room for more recent elements. Like traditional Bloom filters, an SBF
// has a non-zero probability of false positives, which is controlled by
// several parameters. Unlike the classic Bloom filter, an SBF has a tight
// upper bound on the rate of false positives while introducing a non-zero rate
// of false negatives. The false-positive rate of a classic Bloom filter
// eventually reaches 1, after which all queries result in a false positive.
// The stable-point property of an SBF means the false-positive rate
// asymptotically approaches a configurable fixed constant. A classic Bloom
// filter is actually a special case of SBF where the eviction rate is zero, so
// this package provides support for them as well.
//
// Stable Bloom Filters are useful for cases where the size of the data set
// isn't known a priori, which is a requirement for traditional Bloom filters,
// and memory is bounded.  For example, an SBF can be used to deduplicate
// events from an unbounded event stream with a specified upper bound on false
// positives and minimal false negatives.

use crate::buckets::Buckets;
use crate::fnv::FnvHasher;
use crate::Filter;
use crate::{optimal_k, optimal_stable_p};
use rand::{thread_rng, Rng};
use std::hash::Hasher;

pub struct StableBloomFilter {
    // filter data
    cells: Buckets,
    // hash function (kernel for all k functions)
    hash: FnvHasher,
    // number of cells
    m: usize,
    // number of cells to decrement
    p: usize,
    // number of hash functions
    k: usize,
    // cell max value
    max: u8,
    // buffer used to cache indices
    index_buffer: Vec<usize>,
}

impl StableBloomFilter {
    // Creates a new Stable Bloom Filter with m cells and d
    // bits allocated per cell optimized for the target false-positive rate. Use
    // default if you don't want to calculate d.
    pub fn new(m: usize, d: u8, fp_rate: f64) -> Self {
        let mut k = optimal_k(fp_rate) / 2;
        if k > m {
            k = m;
        } else if k == 0 {
            k = 1;
        }

        let cells = Buckets::new(m, d);

        StableBloomFilter {
            hash: FnvHasher::default(),
            m,
            k,
            p: optimal_stable_p(m, k, d, fp_rate),
            max: cells.max_bucket_value(),
            cells,
            index_buffer: vec![0; k],
        }
    }

    // Creates a new Stable Bloom Filter with m 1-bit
    // cells and which is optimized for cases where there is no prior knowledge of
    // the input data stream while maintaining an upper bound using the provided
    // rate of false positives.
    pub fn new_default(m: usize, fp_rate: f64) -> Self {
        Self::new(m, 1, fp_rate)
    }

    // NewUnstableBloomFilter creates a new special case of Stable Bloom Filter
    // which is a traditional Bloom filter with m bits and an optimal number of
    // hash functions for the target false-positive rate. Unlike the stable
    // variant, data is not evicted and a cell contains a maximum of 1 hash value.
    pub fn new_unstable(m: usize, fp_rate: f64) -> Self {
        let cells = Buckets::new(m, 1);
        let k = optimal_k(fp_rate);

        StableBloomFilter {
            hash: FnvHasher::default(),
            m,
            k,
            p: 0,
            max: cells.max_bucket_value(),
            cells,
            index_buffer: vec![0; k],
        }
    }

    // Returns the number of cells in the Stable Bloom Filter.
    pub fn cells(&self) -> usize {
        self.m
    }

    // Returns the number of hash functions.
    pub fn k(&self) -> usize {
        self.k
    }

    // Returns the number of cells decremented on every add.
    pub fn p(&self) -> usize {
        self.p
    }

    pub fn max(&self) -> u8 {
        self.max
    }

    // Returns the limit of the expected fraction of zeros in the
    // Stable Bloom Filter when the number of iterations goes to infinity. When
    // this limit is reached, the Stable Bloom Filter is considered stable.
    pub fn stable_point(&self) -> f64 {
        let sub_denom = (self.p as f64) * ((1.0 / (self.k as f64)) - (1.0 / (self.m as f64)));
        let denom = 1.0 + 1.0 / sub_denom;
        let base = 1.0 / denom;

        base.powf(f64::from(self.max))
    }

    // Returns the upper bound on false positives when the filter
    // has become stable.
    pub fn false_positive_rate(&self) -> f64 {
        (1.0 - self.stable_point()).powf(self.k as f64)
    }

    pub fn hash_kernel(&self, data: &[u8]) -> (u32, u32) {
        let mut hasher = self.hash.clone();
        hasher.write(data);
        let hash: u64 = hasher.finish();
        let lower = hash as u32;
        let upper = (hash >> 32) as u32;
        (lower, upper)
    }

    // Restores the Stable Bloom Filter to its original state. It returns the
    // filter to allow for chaining.
    pub fn reset(&mut self) -> &Self {
        self.cells.reset();
        self
    }

    // Will decrement a random cell and (p-1) adjacent cells by 1. This
    // is faster than generating p random numbers. Although the processes of
    // picking the p cells are not independent, each cell has a probability of p/m
    // for being picked at each iteration, which means the properties still hold.
    pub fn decrement(&mut self) {
        let mut rng = thread_rng();
        let r: usize = rng.gen_range(0, self.m);

        for i in 0..(self.p) {
            let idx = (r + i) % self.m;
            self.cells.decrease(idx, 1);
        }
    }
}

impl Filter for StableBloomFilter {
    // Will test for membership of the data and returns true if it is a
    // member, false if not. This is a probabilistic test, meaning there is a
    // non-zero probability of false positives and false negatives.
    fn test(&self, data: &[u8]) -> bool {
        let (lower, upper) = self.hash_kernel(data);
        for i in 0..(self.k) {
            if self
                .cells
                .get((lower as usize + upper as usize * i) % self.m)
                == 0
            {
                return false;
            }
        }
        true
    }

    // Will add the data to the Stable Bloom Filter. It returns the filter to
    // allow for chaining.
    fn add(&mut self, data: &[u8]) -> &Self {
        // Randomly decrement p cells to make room for new elements.
        self.decrement();
        let (lower, upper) = self.hash_kernel(data);

        for i in 0..(self.k) {
            self.cells
                .set((lower as usize + upper as usize * i) % self.m, self.max);
        }

        self
    }

    // Is equivalent to calling Test followed by Add. It returns true if
    // the data is a member, false if not.
    fn test_and_add(&mut self, data: &[u8]) -> bool {
        let (lower, upper) = self.hash_kernel(data);
        let mut member = true;

        // If any of the K cells are 0, then it's not a member.
        for i in 0..(self.k) {
            self.index_buffer[i] = (lower as usize + upper as usize * i) % self.m;
            if self.cells.get(self.index_buffer[i]) == 0 {
                member = false;
            }
        }

        // Randomly decrement p cells to make room for new elements.
        self.decrement();
        // Set the K cells to max.
        for i in self.index_buffer.iter() {
            self.cells.set(*i, self.max);
        }

        member
    }
}

#[cfg(test)]
mod tests {
    use super::StableBloomFilter;
    use crate::optimal_k;
    use crate::Filter;

    fn round(val: f64, round_on: f64, places: usize) -> f64 {
        let pow = (10.0_f64).powf(places as f64);
        let digit = pow * val;
        let div = digit - digit.floor();
        let round = if div >= round_on {
            digit.ceil()
        } else {
            digit.floor()
        };

        round / pow
    }

    // Ensures that new_unstable creates a Stable Bloom Filter with p=0,
    // max=1 and k hash functions.
    #[test]
    fn test_new_unstable() {
        let f = StableBloomFilter::new_unstable(100, 0.1);
        let k = optimal_k(0.1);

        assert_eq!(f.k, k);
        assert_eq!(f.m, 100);
        assert_eq!(f.p(), 0);
        assert_eq!(f.max(), 1);
    }

    // Ensures that Cells returns the number of cells, m, in the Stable Bloom
    // Filter.
    #[test]
    fn test_cells() {
        let f = StableBloomFilter::new(100, 1, 0.1);

        assert_eq!(f.cells(), 100);
    }

    // Ensures that K returns the number of hash functions in the Stable Bloom
    // Filter.
    #[test]
    fn test_k() {
        let f = StableBloomFilter::new(100, 1, 0.01);
        assert_eq!(f.k(), 3);
    }

    // Ensures that Test, Add, and TestAndAdd behave correctly.
    #[test]
    fn test_test_and_add() {
        let mut f = StableBloomFilter::new_default(10000, 0.01);
        assert!(!f.test("a".as_bytes()));

        f.add("a".as_bytes());
        assert!(f.test("a".as_bytes()));

        assert!(f.test_and_add("a".as_bytes()));

        assert!(!f.test_and_add("b".as_bytes()));
        assert!(f.test("a".as_bytes()));

        assert!(f.test("b".as_bytes()));

        assert!(!f.test("c".as_bytes()));

        for i in 0..1000000 {
            f.test_and_add(i.to_string().as_bytes());
        }

        // `a` should have been evicted.
        assert!(!f.test("a".as_bytes()));
    }

    // Ensures that StablePoint returns the expected fraction of zeros for large
    // iterations.
    #[test]
    fn test_stable_point() {
        let mut f = StableBloomFilter::new(1000, 1, 0.1);
        for i in 0..1000000 {
            f.add(i.to_string().as_bytes());
        }

        let mut zero = 0;
        for i in 0..(f.m) {
            if f.cells.get(i) == 0 {
                zero += 1;
            }
        }

        let actual = round((zero as f64) / (f.m as f64), 0.5, 1);
        let expected = round(f.stable_point(), 0.5, 1);

        assert_eq!(actual, expected);
        // A classic Bloom filter is a special case of SBF where P is 0 and max is
        // 1. It doesn't have a stable point.
        let bf = StableBloomFilter::new_unstable(1000, 0.1);
        assert_eq!(bf.stable_point(), 0.0);
    }

    // Ensures that FalsePositiveRate returns the upper bound on false positives
    // for stable filters.
    #[test]
    fn test_false_positive_rate() {
        let f = StableBloomFilter::new_default(1000, 0.01);
        let fps = round(f.false_positive_rate(), 0.5, 2);

        assert_eq!(fps, 0.01);

        // Classic Bloom filters have an unbounded rate of false positives. Once
        // they become full, every query returns a false positive.
        let bf = StableBloomFilter::new_unstable(1000, 0.1);
        assert_eq!(bf.false_positive_rate(), 1.0);
    }

    // Ensures that Reset sets every cell to zero.
    #[test]
    fn test_reset() {
        let mut f = StableBloomFilter::new_default(1000, 0.01);

        for i in 0..1000 {
            f.add(i.to_string().as_bytes());
        }

        f.reset();

        for i in 0..(f.m) {
            assert_eq!(f.cells.get(i), 0);
        }
    }
}
