// Copyright 2017, CZ.NIC z.s.p.o. (http://www.nic.cz/)
//
// This file is part of the pakon system.
//
// Pakon is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
//  (at your option) any later version.
//
// Pakon is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with Pakon.  If not, see <http://www.gnu.org/licenses/>.

//! The internal computation that determines the best domain name to consider primary.
//!
//! It is possible multiple domain names are assigned to a single flow (or an endpoint of it). This
//! internal computation produces a single domain name that it considers primary, suitable for
//! showing to the user.
//!
//! It is based on heuristics. The current one is taking the shortest one, but we may do something
//! better in the future (including watching the previous DNS requests).

#![deny(missing_docs)]

#[cfg(test)]
extern crate futures;
extern crate int_compute;
extern crate libflow;
extern crate libdata;
#[cfg(test)]
extern crate libutils;
#[macro_use]
extern crate slog;
extern crate tokio_core;

use std::cmp::Ordering;

use slog::Logger;
use tokio_core::reactor::Handle;

use int_compute::{Event, Factory as CompFactory, IntCompute, Query};
use libdata::flow::{Name, Names};
use libdata::column::{self, Ident, Local, Remote, Tags, Value};
use libflow::update::{CorkedUpdate, Key, Status, Update, UpdateSenderUnbounded};

/// The internal computation that adds the primary domain name.
pub struct Compute {
    logger: Logger,
    channel: UpdateSenderUnbounded,
}

impl IntCompute for Compute {
    fn event(&mut self, event: &Event, _flow_query: &Query) {
        match *event {
            Event::CorkRequest(ref cork) => {
                trace!(self.logger, "Pushing a cork");
                self.channel
                    .unbounded_send(CorkedUpdate::Cork(cork.clone()))
                    .expect("Event called when the update channel already closed");
            },
            Event::FlowUpdated { ref flow, ref columns } => {
                let borrow = flow.borrow();
                let mut tags = Tags::new();
                macro_rules! check {
                    ($enp: ident) => {
                        let ident = Ident::of::<$enp<Names>>();
                        if columns.contains(&ident) {
                            trace!(self.logger, concat!("Computing ",
                                                        stringify!($enp),
                                                        " primary name"));
                            let &$enp(ref names) = borrow
                                .get(&ident)
                                .expect("Must be there because of the update")
                                .downcast_ref::<$enp<Names>>()
                                .expect("Type mismatch");
                            let primary = names
                                .iter()
                                .min_by(|a, b| match (a.0.len().cmp(&b.0.len()), a.cmp(b)) {
                                    (Ordering::Equal, lex) => lex,
                                    (len, _) => len,
                                })
                                .cloned();
                            if let Some(name) = primary {
                                let val = Value::from($enp(name));
                                let existing = borrow.get(&Ident::of::<$enp<Name>>());
                                // Don't update to the same value, to prevent cycles
                                if Some(&val) != existing {
                                    tags.insert(val);
                                }
                            }
                        }
                    };
                }
                check!(Local);
                check!(Remote);
                if !tags.is_empty() {
                    let update = Update {
                        keys: vec![Key::InternalHandle(flow.clone())],
                        status: Status::Ongoing,
                        tags,
                        stats: None,
                    };
                    self.channel
                        .unbounded_send(CorkedUpdate::Update(update))
                        .expect("Event called when the update channel already closed");
                }
            },
            // Other events aren't interesting for us.
            _ => (),
        }
    }
}

/// The factory for [`Compute`](struct.Compute.html).
pub struct Factory;

impl CompFactory for Factory {
    fn create(self, logger: &Logger, _handle: Handle, channel: UpdateSenderUnbounded)
        -> Box<IntCompute>
    {
        let logger = logger.new(o!("context" => "primary name"));
        info!(logger, "Creating internal compute for primary DNS name");
        // Register the columns we produce
        column::register_endpoint::<Name>();
        let comp = Compute {
            logger,
            channel,
        };
        Box::new(comp)
    }
}

#[cfg(test)]
mod tests {
    use futures::Stream;
    use futures::unsync::mpsc;

    use libdata::flow::{Direction, Name};
    use libdata::column::{FlowTags, RefHeaders};
    use libflow::update;
    use libutils;

    use super::*;

    struct FakeQuery;

    impl Query for FakeQuery {
        fn query<'a, 'b, 'r>(&'a self, _filter: &'b RefHeaders<'b>)
            -> Box<Iterator<Item = FlowTags> + 'r>
        where
            'a: 'r,
            'b: 'r
        {
            unimplemented!();
        }
    }

    /// Tests invoking events on the the primary name computation.
    #[test]
    fn event() {
        let logger = libutils::test_logger();
        let (sender, receiver) = mpsc::unbounded();
        let mut iter = receiver.wait();
        let mut comp = Compute { logger, channel: sender };
        let (cork, _handle) = update::cork();
        let fake_query = FakeQuery;
        let cork_event = Event::CorkRequest(cork);
        comp.event(&cork_event, &fake_query);
        if let Some(Ok(CorkedUpdate::Cork(_))) = iter.next() { } else {
            panic!("Expected cork");
        }
        let mut tags = Tags::new();
        let names = vec![
                "www.example.com",
                "example.com",
                "cdn1.atlantis.www.example.com"
            ]
            .into_iter()
            .map(String::from)
            .map(Name::new)
            .collect::<Names>();
        tags.insert(Local(names));
        let flow = FlowTags::new(tags);
        let columns = vec![Ident::of::<Local<Names>>()]
            .into_iter()
            .collect();
        let relevant_event = Event::FlowUpdated {
            flow: flow.clone(),
            columns,
        };
        comp.event(&relevant_event, &fake_query);
        if let Some(Ok(CorkedUpdate::Update(update))) = iter.next() {
            assert_eq!(1, update.keys.len());
            assert_eq!(1, update.tags.len());
            let val = update.tags
                .get(&Ident::of::<Local<Name>>())
                .expect("Missing update to primary local name")
                .downcast_ref::<Local<Name>>()
                .expect("Wrong type of the update");
            assert_eq!(Local(Name("example.com".to_owned())), *val);
        } else {
            panic!("Expected an update");
        }
        // Now when we add two more unrelated events, no updates should be produced
        let columns = vec![Ident::of::<Direction>()]
            .into_iter()
            .collect();
        let unrelated_option = Event::FlowUpdated {
            flow: flow.clone(),
            columns,
        };
        comp.event(&unrelated_option, &fake_query);
        let terminated_event = Event::FlowTerminated(flow.clone());
        comp.event(&terminated_event, &fake_query);
        // Pushing the same event again doesn't produce an update, because the value is already
        // there
        flow.borrow_mut().insert(Local(Name("example.com".to_owned())));
        comp.event(&relevant_event, &fake_query);
        // Push everything through by a cork
        comp.event(&cork_event, &fake_query);
        if let Some(Ok(CorkedUpdate::Cork(_))) = iter.next() { } else {
            panic!("Expected cork");
        }
        // Drop the comp → drop the sender → EOF
        drop(comp);
        assert!(iter.next().is_none());
    }
}