//! Generators for the models usd in the Nexmark benchmark suite.
//!
//! Based on the equivalent [Nexmark Flink generator API](https://github.com/nexmark/nexmark/blob/v0.2.0/nexmark-flink/src/main/java/com/github/nexmark/flink/generator).

use std::{cell::RefCell, marker::PhantomData};

use self::config::Config;
use super::model::Event;
use bids::CHANNELS_NUMBER;
use cached::SizedCache;
use rand::{Rng, SeedableRng};

mod auctions;
mod bids;
pub mod config;
mod people;
mod price;
mod strings;

/// Generates Nexmark events sequentially within a maximum event count.
pub struct NexmarkGenerator<R: Rng> {
    core: GeneratorCore,

    /// Number of events already generated by this generator.
    /// Note that when there are multiple generators working in parallel, the
    /// events count for each generator is less than the events generated by the
    /// source.
    events_count_so_far: u64,

    rng: R,
}

impl<R: Rng> NexmarkGenerator<R> {
    pub fn new(config: Config, rng: R) -> Self {
        Self {
            core: GeneratorCore::new(config),
            rng,
            events_count_so_far: 0,
        }
    }
}

impl<R: Rng> Iterator for NexmarkGenerator<R> {
    type Item = Event;

    fn next(&mut self) -> Option<Event> {
        let new_event_id = self.core.config.next_event_number(self.events_count_so_far);
        if new_event_id < self.core.config.max_events {
            self.events_count_so_far += 1;
            Some(self.core.generate(new_event_id, &mut self.rng))
        } else {
            None
        }
    }
}

/// Generates Nexmark events with arbitrary event IDs.
pub struct RandomGenerator<S: SeedableRng + Rng> {
    core: GeneratorCore,
    _phantom: PhantomData<S>,
}

impl<S: SeedableRng + Rng> RandomGenerator<S> {
    pub fn new(config: Config) -> Self {
        Self {
            core: GeneratorCore::new(config),
            _phantom: PhantomData,
        }
    }

    /// Returns the event with the given `event_id`.
    pub fn generate(&self, event_id: u64) -> Event {
        self.core
            .generate(event_id, &mut S::seed_from_u64(event_id))
    }
}

pub struct GeneratorCore {
    /// Configuration to generate events against.
    config: Config,

    /// The memory cache used when creating bid channels.
    bid_channel_cache: RefCell<SizedCache<u32, (String, String)>>,
}

impl GeneratorCore {
    pub fn new(config: Config) -> Self {
        Self {
            config,
            bid_channel_cache: RefCell::new(SizedCache::with_size(CHANNELS_NUMBER as usize)),
        }
    }

    fn generate<R>(&self, event_id: u64, rng: &mut R) -> Event
    where
        R: Rng,
    {
        // When, in event time, we should generate the event. Monotonic.
        let event_timestamp = self.config.timestamp_for_event(event_id);

        let mut context = GeneratorContext {
            config: &self.config,
            bid_channel_cache: &self.bid_channel_cache,
            rng,
        };

        let options = &self.config.options;
        let rem = event_id % options.total_proportion() as u64;
        if rem < options.person_proportion as u64 {
            Event::Person(context.next_person(event_id, event_timestamp))
        } else if rem < options.person_proportion as u64 + options.auction_proportion as u64 {
            Event::Auction(context.next_auction(event_id, event_timestamp))
        } else {
            Event::Bid(context.next_bid(event_id, event_timestamp))
        }
    }
}

struct GeneratorContext<'a, R>
where
    R: Rng,
{
    config: &'a Config,

    /// The memory cache used when creating bid channels.
    bid_channel_cache: &'a RefCell<SizedCache<u32, (String, String)>>,

    rng: &'a mut R,
}

#[cfg(test)]
impl<'a, R: Rng> GeneratorContext<'a, R> {
    fn new(core: &'a GeneratorCore, rng: &'a mut R) -> Self {
        Self {
            config: &core.config,
            bid_channel_cache: &core.bid_channel_cache,
            rng,
        }
    }
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use crate::{
        config::GeneratorOptions,
        model::{Auction, Bid, Person},
    };
    use rand::{rngs::mock::StepRng, thread_rng};

    pub fn test_generator_config() -> Config {
        Config {
            options: GeneratorOptions {
                num_event_generators: 1,
                ..GeneratorOptions::default()
            },
            ..Config::default()
        }
    }
    pub fn make_test_generator() -> (GeneratorCore, StepRng) {
        (
            GeneratorCore::new(test_generator_config()),
            StepRng::new(0, 1),
        )
    }

    pub fn make_person() -> Person {
        Person {
            id: 1,
            name: String::from("AAA BBBB"),
            email_address: String::from("AAABBB@example.com"),
            credit_card: String::from("1111 2222 3333 4444"),
            city: String::from("Phoenix"),
            state: String::from("OR"),
            date_time: 0,
            extra: String::from(""),
        }
    }

    pub fn make_bid() -> Bid {
        Bid {
            auction: 1,
            bidder: 1,
            price: 99,
            channel: String::from("my-channel"),
            url: String::from("https://example.com"),
            date_time: 0,
            extra: String::new(),
        }
    }

    pub fn make_auction() -> Auction {
        Auction {
            id: 1,
            item_name: String::from("item-name"),
            description: String::from("description"),
            initial_bid: 5,
            reserve: 10,
            date_time: 0,
            expires: 2000,
            seller: 1,
            category: 1,
            extra: String::new(),
        }
    }

    pub fn make_next_event() -> Event {
        Event::Bid(make_bid())
    }

    /// Generates a specified number of next events using the default test
    /// generator.
    pub fn generate_expected_next_events(num_events: usize) -> Vec<Option<Event>> {
        let mut ng = NexmarkGenerator::new(test_generator_config(), make_test_generator().1);

        (0..num_events).map(|_| ng.next()).collect()
    }

    #[test]
    fn test_has_next() {
        let config = Config {
            max_events: 2,
            ..test_generator_config()
        };
        let mut ng = NexmarkGenerator::new(config, make_test_generator().1);

        ng.next().unwrap();
        ng.next().unwrap();
        assert!(ng.next().is_none());
    }

    // Tests the first five expected events without relying on any test
    // helper for the data.
    #[test]
    fn test_next_event() {
        let mut ng = NexmarkGenerator::new(
            Config {
                options: GeneratorOptions {
                    num_event_generators: 1,
                    ..GeneratorOptions::default()
                },
                ..Config::default()
            },
            thread_rng(),
        );

        // The first event with the default config is the person
        let next_event = ng.next();
        assert!(next_event.is_some());
        let next_event = next_event.unwrap();

        assert!(
            matches!(next_event, Event::Person(_)),
            "got: {:?}, want: Event::NewPerson(_)",
            next_event
        );
        assert_eq!(next_event.timestamp(), 0);

        // The next 3 events with the default config are auctions
        for event_num in 1..=3 {
            let next_event = ng.next();
            assert!(next_event.is_some());
            let next_event = next_event.unwrap();

            assert!(
                matches!(next_event, Event::Auction(_)),
                "got: {:?}, want: Event::NewAuction(_)",
                next_event
            );
            assert_eq!(next_event.timestamp(), event_num * 10);
        }

        // And the rest of the events in the first epoch are bids.
        for _ in 4..=49 {
            let next_event = ng.next();
            assert!(next_event.is_some());
            let next_event = next_event.unwrap();

            assert!(
                matches!(next_event, Event::Bid(_)),
                "got: {:?}, want: Event::NewBid(_)",
                next_event
            );
        }

        // The next epoch begins with another person etc.
        let next_event = ng.next().unwrap();
        assert!(
            matches!(next_event, Event::Person(_)),
            "got: {:?}, want: Event::NewPerson(_)",
            next_event
        );
    }

    // Verifies that the `generate_expected_next_events()` test helper does
    // indeed output predictable results matching the order verified manually in
    // the above `test_next_events` (at least for the first 5 events).  Together
    // with the manual test above of next_events, this ensures the order is
    // correct for external call-sites.
    #[test]
    fn test_generate_expected_next_events() {
        let mut ng = NexmarkGenerator::new(test_generator_config(), make_test_generator().1);

        let expected_events = generate_expected_next_events(100);

        assert_eq!(
            (0..100).map(|_| ng.next()).collect::<Vec<Option<Event>>>(),
            expected_events
        );
    }
}
