extern crate elasticsearch;
extern crate serde_json;

use std::collections::HashSet;

use elasticsearch::{ClearScrollParts, DeleteByQueryParts, DeleteParts, Elasticsearch, Error, IndexParts, ScrollParts, SearchParts};
use elasticsearch::cat::CatIndicesParts;
use elasticsearch::http::response::Response;
use elasticsearch::http::transport::Transport;
use serde_json::{json, Value};

use crate::constants::ELASTIC_URL;
use crate::helpers::es_helper::ElasticHelper;
use crate::models::flickr_photo::FlickrPhoto;
use crate::models::picture::Picture;
use crate::processors::report::Report;

use self::elasticsearch::indices::IndicesPutSettingsParts;

#[derive(Debug, Clone)]
pub struct ElasticClient {
    pub client: Elasticsearch,
    pub index: String,
}

impl ElasticClient {
    pub fn new(index: &str) -> Self {
        ElasticClient {
            client: ElasticClient::get_client(),
            index: String::from(index),
        }
    }

    fn get_client() -> Elasticsearch {
        let transport = Transport::single_node(ELASTIC_URL);
        Elasticsearch::new(transport.unwrap_or_default())
    }

    pub(crate) async fn get_indices(&self) -> Result<Vec<String>, Error> {
        let mut indices: Vec<String> = Vec::new();
        let response = self.client
            .cat()
            .indices(CatIndicesParts::Index(&["*"]))
            .format("json")
            .send()
            .await;

        if response.is_ok() {
            let response_body = response.unwrap().json::<Value>().await.unwrap_or_default();
            let response_as_array = response_body.as_array();
            if response_as_array.is_some() {
                for record in response_as_array.unwrap() {
                    indices.push(String::from(record["index"].as_str().unwrap_or_default()));
                }
            }
            Ok(indices)
        } else {
            Err(response.err().unwrap())
        }
    }

    pub(crate) async fn index_picture(&self, record: &Picture) -> Result<Response, Error> {
        self.client
            .index(IndexParts::IndexId(&self.index, &record.path))
            .body(record)
            .send()
            .await
    }

    pub(crate) async fn index_pictures(&self, records: &HashSet<Picture>, harvest_report: &mut Report, errors: &mut Vec<String>) {
        for record in records {
            let response = ElasticClient::index_picture(self, record).await;
            ElasticHelper::generate_harvest_report(harvest_report, errors, response)
        }
    }

    pub(crate) async fn index_flickr_photo(&self, record: &FlickrPhoto) -> Result<Response, Error> {
        self.client
            .index(IndexParts::IndexId(&self.index, ElasticClient::get_flickr_record_id(record).as_str()))
            .body(record)
            .send()
            .await
    }

    pub(crate) async fn index_flickr_photos(&self, records: &HashSet<FlickrPhoto>, harvest_report: &mut Report, errors: &mut Vec<String>) {
        for record in records {
            let response = ElasticClient::index_flickr_photo(self, record).await;
            ElasticHelper::generate_harvest_report(harvest_report, errors, response)
        }
    }

    #[allow(dead_code)]
    pub(crate) async fn query_all(&self) -> Result<Response, Error> {
        self.client
            .search(SearchParts::Index(&[&self.index]))
            .scroll("300s")
            .size(10)
            .body(json!({
                "query": {
                    "match_all": {}
                }
            }))
            .send()
            .await
    }

    pub(crate) async fn simple_search(&self, query: Value) -> Result<Response, Error> {
        self.client
            .search(SearchParts::Index(&[&self.index]))
            .body(query)
            .send()
            .await
    }

    pub(crate) async fn scrolled_search(&self, query: Value) -> Result<Response, Error> {
        self.client
            .search(SearchParts::Index(&[&self.index]))
            .scroll("1m")
            .size(10)
            .body(query)
            .send()
            .await
    }

    pub(crate) async fn scroll(&self, scroll_id: &str) -> Result<Response, Error> {
        self.client
            .scroll(ScrollParts::ScrollId(scroll_id))
            .scroll("1m")
            .send()
            .await
    }

    pub(crate) async fn clear_scroll(&self, scroll_id: &str) -> Result<Response, Error> {
        self.client
            .clear_scroll(ClearScrollParts::ScrollId(&[scroll_id]))
            .send()
            .await
    }

    #[allow(dead_code)]
    pub(crate) async fn delete_all_records(&self) -> Result<Response, Error> {
        self.client
            .delete_by_query(DeleteByQueryParts::Index(&[&self.index]))
            .body(json!({
                "query": {
                    "match_all": {}
                }
            }))
            .send()
            .await
    }

    pub(crate) async fn delete_picture_record(&self, record: &Picture) -> Result<Response, Error> {
        self.client
            .delete(DeleteParts::IndexId(&self.index, &record.path))
            .send()
            .await
    }

    pub(crate) async fn delete_flickr_photo_record(&self, record: &FlickrPhoto) -> Result<Response, Error> {
        self.client
            .delete(DeleteParts::IndexId(&self.index, ElasticClient::get_flickr_record_id(record).as_str()))
            .send()
            .await
    }

    pub(crate) fn get_flickr_record_id(record: &FlickrPhoto) -> String {
        record.id.clone() + "_" + record.secret.clone().as_str() + "_" + record.farm.clone().to_string().as_str() + "_" + record.server.clone().as_str()
    }

    // to use if the batch to index is really huge (more than 10000)
    // before indexing with leverage = true, a moment later with leverage = false
    // to wait use thread::sleep(time::Duration::from_millis(1000 * 60 * 60))
    #[allow(dead_code)]
    #[allow(unused_assignments)]
    pub(crate) async fn leverage_index_settings(&self, leverage: bool) {
        let mut body = Value::Null;
        if leverage {
            body = json!({
                "index.number_of_replicas" : 0,
                "index.refresh_interval": -1
            })
        } else {
            body = json!({
                "index.number_of_replicas" : 1,
                "index.refresh_interval": "1s"
            })
        }
        let response = self.client
            .indices()
            .put_settings(IndicesPutSettingsParts::Index(&[&self.index]))
            .body(body)
            .send()
            .await;
        if response.is_ok() {
            println!("{:?}", response.unwrap().json::<Value>().await.unwrap());
        } else {
            println!("{:?}", response.err().unwrap())
        }
    }
}
