use url_bot_rs::config::Rtd;

use url_bot_rs::VERSION;

use url_bot_rs::{feat, http::resolve_url, param, plugins::TITLE_PLUGINS, tld::TLD};

use anyhow::Result as AHResult;

use atty::{is, Stream};

use directories::ProjectDirs;

use docopt::Docopt;

use failure::Error;

use lazy_static::lazy_static;

use log::{error, info};

use regex::Regex;

use reqwest::Url;

use serde_derive::Deserialize;

use std::collections::HashSet;

use std::path::PathBuf;

use stderrlog::{ColorChoice, Timestamp};

use lapin::{options::*, types::FieldTable, BasicProperties, Connection, ConnectionProperties};

use tokio_amqp::*;

use futures::prelude::*;

use rand::distributions::Alphanumeric;

use rand::{thread_rng, Rng};

use std::sync::mpsc;

use std::thread;

// docopt usage string

const USAGE: &str = "

URL munching IRC bot.

Usage:

    ubot-url-plugin [options] [-v...] [--conf=PATH...] [--conf-dir=DIR...]

Options:

    -h --help           Show this help message.

    --version           Print version.

    -v --verbose        Show extra information.

    -t --timestamp      Force timestamps.

";

#[derive(Debug, Deserialize, Default)]

pub struct Args {

    flag_verbose: usize,

    flag_conf: Vec<PathBuf>,

    flag_conf_dir: Vec<PathBuf>,

    flag_timestamp: bool,

}

const MIN_VERBOSITY: usize = 2;

#[derive(Debug, PartialEq)]

enum TitleResp {

    Title(String),

    Error(String),

}

/// Run available plugins on a single URL, return the first successful title.

fn process_plugins(rtd: &Rtd, url: &Url) -> Option<String> {

    let result: String = TITLE_PLUGINS

        .iter()

        .filter(|p| p.check(&rtd.conf.plugins, url))

        .filter_map(|p| p.evaluate(&rtd, url).ok())

        .take(1)

        .collect();

    if result.is_empty() {

        None

    } else {

        Some(result)

    }

}

/// find titles in a message and generate responses

fn process_titles(rtd: &Rtd, msg: &str) -> impl Iterator<Item = TitleResp> {

    let mut responses: Vec<TitleResp> = vec![];

    let mut num_processed = 0;

    let mut dedup_urls = HashSet::new();

    // look at each space-separated message token

    for token in msg.split_whitespace() {

        // the token must not contain unsafe characters

        if contains_unsafe_chars(token) {

            continue;

        }

        // get a full URL for tokens without a scheme

        let maybe_token = if feat!(rtd, partial_urls) {

            add_scheme_for_tld(token)

        } else {

            None

        };

        let token = maybe_token.as_ref().map_or(token, String::as_str);

        // the token must be a valid url

        let url = match token.parse::<Url>() {

            Ok(url) => url,

            _ => continue,

        };

        // the scheme must be http or https

        if !["http", "https"].contains(&url.scheme()) {

            continue;

        }

        // skip duplicate urls within the message

        if dedup_urls.contains(&url) {

            continue;

        }

        info!("[{}] RESOLVE <{}>", rtd.conf.network.name, token);

        // try to get the title from the url

        let title = if let Some(title) = process_plugins(rtd, &url) {

            title

        } else {

            match resolve_url(token, rtd) {

                Ok(title) => title,

                Err(err) => {

                    error!("{:?}", err);

                    responses.push(TitleResp::Error(err.to_string()));

                    continue;

                }

            }

        };

        // limit response length, see RFC1459

        let msg = utf8_truncate(&format!("⤷ {}", title), 510);

        info!("[{}] {}", rtd.conf.network.name, msg);

        responses.push(TitleResp::Title(msg.to_string()));

        dedup_urls.insert(url);

        // limit the number of processed URLs

        num_processed += 1;

        if num_processed == param!(rtd, url_limit) {

            break;

        }

    }

    responses.into_iter()

}

// regex for unsafe characters, as defined in RFC 1738

const RE_UNSAFE_CHARS: &str = r#"[{}|\\^~\[\]`<>"]"#;

/// does the token contain characters not permitted by RFC 1738

fn contains_unsafe_chars(token: &str) -> bool {

    lazy_static! {

        static ref UNSAFE: Regex = Regex::new(RE_UNSAFE_CHARS).unwrap();

    }

    UNSAFE.is_match(token)

}

/// truncate to a maximum number of bytes, taking UTF-8 into account

fn utf8_truncate(s: &str, n: usize) -> String {

    s.char_indices()

        .take_while(|(len, c)| len + c.len_utf8() <= n)

        .map(|(_, c)| c)

        .collect()

}

lazy_static! {

    static ref REPEATED_DOTS: Regex = Regex::new(r"\.\.+").unwrap();

}

/// if a token has a recognised TLD, but no scheme, add one

pub fn add_scheme_for_tld(token: &str) -> Option<String> {

    if token.parse::<Url>().is_err() {

        if token.starts_with(|s: char| !s.is_alphabetic()) {

            return None;

        }

        if REPEATED_DOTS.is_match(&token) {

            return None;

        }

        let new_token = format!("http://{}", token);

        if let Ok(url) = new_token.parse::<Url>() {

            if !url.domain()?.contains('.') {

                return None;

            }

            // reject email addresses

            if url.username() != "" {

                return None;

            }

            let tld = url.domain()?.split('.').last()?;

            if TLD.contains(tld) {

                return Some(new_token);

            }

        }

    }

    None

}

fn init_rtd() -> AHResult<Rtd, Error> {

    // parse command line arguments with docopt

    let args: Args = Docopt::new(USAGE)

        .and_then(|d| d.version(Some(VERSION.to_string())).deserialize())

        .unwrap_or_else(|e| e.exit());

    // avoid timestamping when piped, e.g. systemd

    let timestamp = if is(Stream::Stderr) || args.flag_timestamp {

        Timestamp::Second

    } else {

        Timestamp::Off

    };

    stderrlog::new()

        .module(module_path!())

        .modules(vec![

            "url_bot_rs::message",

            "url_bot_rs::config",

            "url_bot_rs::http",

        ])

        .verbosity(args.flag_verbose + MIN_VERBOSITY)

        .timestamp(timestamp)

        .color(ColorChoice::Never)

        .init()

        .unwrap();

    let dirs = ProjectDirs::from("org", "", "url-bot-rs").unwrap();

    let default_conf_dir = dirs.config_dir();

    let default_conf = default_conf_dir.join("config.toml");

    let rtd: Rtd = Rtd::new().conf(&default_conf).load()?.init_http_client()?;

    Ok(rtd)

}

fn random_string(size: usize) -> String {

    thread_rng()

        .sample_iter(&Alphanumeric)

        .take(size)

        .map(char::from)

        .collect()

}

#[tokio::main]

async fn main() -> AHResult<()> {

    let (tx1, rx1) = mpsc::channel::<String>();

    let (tx2, rx2) = mpsc::channel();

    thread::spawn(move || {

        let rtd = init_rtd().expect("RTD not initialized");

        loop {

            let message = &rx1.recv().expect("rx1 recv error");

            let titles: Vec<_> = process_titles(&rtd, message).collect();

            tx2.send(titles).expect("tx2 send error");

        }

    });

    let amqp_url = std::env::var("AMQP_URL").expect("expected AMQP_URL env variabe");

    let conn = Connection::connect(&amqp_url, ConnectionProperties::default().with_tokio()).await?;

    let pub_channel = conn.create_channel().await?;

    let sub_channel = conn.create_channel().await?;

    let queue = sub_channel

        .queue_declare(

            &random_string(32),

            QueueDeclareOptions {

                exclusive: true,

                auto_delete: true,

                ..QueueDeclareOptions::default()

            },

            FieldTable::default(),

        )

        .await?;

    sub_channel

        .queue_bind(

            queue.name().as_str(),

            "irc",

            "msg.hedgewars",

            QueueBindOptions::default(),

            FieldTable::default(),

        )

        .await?;

    let mut subscriber = sub_channel

        .basic_consume(

            queue.name().as_str(),

            &random_string(32),

            BasicConsumeOptions::default(),

            FieldTable::default(),

        )

        .await?;

    while let Some(amqp_message) = subscriber.next().await {

        let (_, delivery) = amqp_message.expect("error in consumer");

        delivery.ack(BasicAckOptions::default()).await?;

        let chat_message = String::from_utf8(delivery.data)?;

        if let Some((_who, message)) = chat_message.split_once('\n') {

            tx1.send(message.to_owned())?;

            let titles = rx2.recv()?;

            for title in titles {

                let title_message = match title {

                    TitleResp::Title(t) => t,

                    TitleResp::Error(e) => e,

                };

                pub_channel

                    .basic_publish(

                        "irc",

                        "say.hedgewars",

                        BasicPublishOptions::default(),

                        title_message.as_bytes().to_vec(),

                        BasicProperties::default(),

                    )

                    .await?;

            }

        }

    }

    Ok(())

}