#![doc = include_str!("../README.md")]
#![warn(clippy::all, clippy::pedantic)]
#![allow(clippy::missing_errors_doc)]

use chrono::SecondsFormat;
use hashbrown::HashMap;
use parking_lot::{Mutex, RwLock};
use serde::Deserialize;
use std::{
    error::Error,
    fmt::{Debug, Display},
    fs::File,
    io::{BufRead, BufReader, Write},
    path::PathBuf,
    process,
    sync::Arc,
    time::{Duration, SystemTime},
};
use tantivy::schema::Field;
use tantivy::{
    directory::MmapDirectory, query::QueryParser, schema::{DateOptions, Schema, INDEXED, STORED, STRING, TEXT}, Document, Index,
    IndexReader,
    IndexWriter,
    TantivyDocument,
};
use tokio::task::JoinHandle;
use tracing::{instrument, Span};

#[derive(Deserialize)]
struct LogSpan {
    // name: String,

    // env: Option<String>,
    domain: Option<String>,
    // id: Option<String>,

    // i: Option<u8>,
    // nm: Option<String>,

    // trigger: Option<String>,

    // flavor: Option<String>,
}

// #[derive(Deserialize)]
// struct LogFields {
// message: Option<String>,

// ip: Option<String>,
// version: Option<String>,
// method: Option<String>,
// uri: Option<String>,
// headers: Option<String>,
// status: Option<u16>,

// token: Option<String>,

// ts: Option<String>,
// path: Option<String>,
// query: Option<String>,
// correlation: Option<String>,
// flags: Option<String>,
// }

#[derive(Deserialize)]
struct LogLine {
    // timestamp: String,
    // level: String,
    // fields: LogFields,
    spans: Option<Vec<LogSpan>>,
}

#[derive(Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub enum QueryFormat {
    All,
    Count,
    Top,
}

impl QueryFormat {
    fn format(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match *self {
            Self::All => write!(f, "all"),
            Self::Count => write!(f, "count"),
            Self::Top => write!(f, "top"),
        }
    }
}

impl Display for QueryFormat {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.format(f)
    }
}

impl Debug for QueryFormat {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.format(f)
    }
}

pub struct Manager {
    span: Span,
    dir: PathBuf,
    schema: Schema,
    fields: Vec<Field>,

    // todo: needs more than just static interval,
    // todo: but this gets *something* for now.
    index_interval: (u64, u64),
    max_ttl_days: u16,
    clean_interval: (u64, u64),

    #[allow(clippy::type_complexity)]
    domain_indexes: Arc<RwLock<HashMap<String, (Index, IndexReader, QueryParser)>>>,

    file_cursor: Arc<Mutex<Option<PathBuf>>>,
}

impl Manager {
    pub fn new(
        dir: PathBuf,

        index_interval: (u64, u64),
        max_ttl_days: u16,
        clean_interval: (u64, u64),
    ) -> Result<Self, Box<dyn Error>> {
        let start = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)?
            .as_millis();

        let span = tracing::info_span!("monitor", start, pid = process::id());

        let cursor_path = dir.join("cursor");

        let file_cursor_path = cursor_path.join("file");
        let file_cursor = match std::fs::read(&file_cursor_path) {
            Ok(file) => Some(dir.join(std::str::from_utf8(&file)?)),
            Err(_) => None,
        };

        let mut schema_builder = Schema::builder();

        let ts_opts = DateOptions::from(INDEXED)
            .set_stored()
            .set_fast()
            .set_precision(tantivy::schema::DateTimePrecision::Microseconds);

        let timestamp = schema_builder.add_date_field("timestamp", ts_opts);
        let level = schema_builder.add_text_field("level", STRING | STORED);
        let fields = schema_builder.add_json_field("fields", TEXT | STORED);
        let spans = schema_builder.add_json_field("spans", TEXT | STORED);

        let schema = schema_builder.build();

        Ok(Manager {
            span,
            dir,
            schema,
            fields: vec![timestamp, level, fields, spans],

            index_interval,
            max_ttl_days,
            clean_interval,

            domain_indexes: Arc::new(RwLock::new(HashMap::new())),
            file_cursor: Arc::new(Mutex::new(file_cursor)),
        })
    }

    #[instrument(skip(self), err)]
    pub fn add(&self, domain: String) -> Result<(), Box<dyn Error>> {
        let index_dir = self.dir.join("search").join(&domain);
        std::fs::create_dir_all(&index_dir)?;

        let dir = MmapDirectory::open(index_dir)?;
        let index = Index::open_or_create(dir, self.schema.clone())?;

        let reader = index
            .reader_builder()
            .reload_policy(tantivy::ReloadPolicy::OnCommitWithDelay);

        let query_parser = QueryParser::for_index(&index, self.fields.clone());

        let mut di = self.domain_indexes.write();
        di.insert(domain, (index, reader.try_into()?, query_parser));

        Ok(())
    }

    // todo: include parameter for app/admin/root logs
    #[instrument(skip(self, domain, query, format, limit), err)]
    pub fn search(
        &self,
        domain: String,
        query: String,
        format: QueryFormat,
        limit: Option<usize>,
    ) -> Result<String, Box<dyn Error>> {
        if let Some(limit) = limit {
            tracing::info!(query, %format, limit);
        } else {
            tracing::info!(query, %format);
        }

        let di = self.domain_indexes.read();
        let mut out = String::new();

        if let Some((_, reader, parser)) = di.get(&domain) {
            let query = parser.parse_query(&query)?;
            let searcher = reader.searcher();

            match format {
                QueryFormat::All => {
                    let docs = searcher.search(&*query, &tantivy::collector::DocSetCollector)?;

                    out.push('[');

                    tracing::info!(count = docs.len());
                    let last_i = docs.len() - 1;

                    for (i, doc_addr) in docs.iter().enumerate() {
                        let doc: TantivyDocument = searcher.doc(*doc_addr)?;
                        let line = doc.to_json(&self.schema);

                        out.push_str(&line);

                        if i < last_i {
                            out.push(',');
                        }
                    }

                    out.push(']');
                }
                QueryFormat::Count => {
                    let count = searcher.search(&*query, &tantivy::collector::Count)?;
                    out.clone_from(&count.to_string());
                }
                QueryFormat::Top => {
                    if let Some(limit) = limit {
                        let docs = searcher
                            .search(&*query, &tantivy::collector::TopDocs::with_limit(limit))?;

                        out.push('[');

                        tracing::info!(count = docs.len());
                        let last_i = docs.len() - 1;

                        for (i, (_, doc_addr)) in docs.iter().enumerate() {
                            let doc: TantivyDocument = searcher.doc(*doc_addr)?;
                            let line = doc.to_json(&self.schema);

                            out.push_str(&line);

                            if i < last_i {
                                out.push(',');
                            }
                        }

                        out.push(']');
                    }
                }
            }
        }

        Ok(out)
    }

    #[instrument(skip(dir, file_cursor_mutex, domain_indexes, schema), err)]
    #[allow(clippy::type_complexity)]
    fn index(
        dir: &PathBuf,
        file_cursor_mutex: &Arc<Mutex<Option<PathBuf>>>,
        domain_indexes: &Arc<RwLock<HashMap<String, (Index, IndexReader, QueryParser)>>>,
        schema: &Schema,
    ) -> Result<(), Box<dyn Error>> {
        tracing::info!("start");

        let cursor_path = dir.join("cursor");

        let file_cursor_path = cursor_path.join("file");
        let file_cursor = match std::fs::read(&file_cursor_path) {
            Ok(file) => dir.join(std::str::from_utf8(&file)?),
            Err(_) => dir.join("."),
        };

        let line_cursor_path = cursor_path.join("line");
        let line_cursor = match std::fs::read(&line_cursor_path) {
            Ok(file) => std::str::from_utf8(&file)?.parse::<usize>()?,
            Err(_) => 0,
        };

        let mut log_files = std::fs::read_dir(dir)?
            .map(|res| res.map(|e| e.path()))
            .collect::<Result<Vec<_>, std::io::Error>>()?;
        log_files.sort();

        let di = domain_indexes.read();

        let mut writer_map = HashMap::new();

        for (domain, (index, _, _)) in di.iter() {
            writer_map.insert(domain, (index, vec![]));
        }

        let mut last_log_file = file_cursor.clone();
        let mut last_log_line = line_cursor;

        for log_file in log_files {
            if log_file < *file_cursor || log_file.is_dir() {
                continue;
            }

            tracing::info!(
                log_file = log_file.as_os_str().to_str().unwrap_or("no/path"),
                "indexing"
            );

            let file = File::open(&log_file)?;

            let reader = BufReader::new(file);

            for (i, line) in reader.lines().enumerate() {
                if log_file == *file_cursor && i < line_cursor {
                    continue;
                }

                let line = line?;
                let log = serde_json::from_str::<LogLine>(&line)?;

                if let Some(spans) = log.spans {
                    for span in spans {
                        if let Some(domain) = span.domain {
                            if let Some((_, docs)) = writer_map.get_mut(&domain) {
                                docs.push(TantivyDocument::parse_json(schema, &line)?);
                            }

                            break;
                        }
                    }
                }

                last_log_line = i;
            }

            last_log_file = log_file;
        }

        for (domain, (index, docs)) in writer_map {
            let mut writer: IndexWriter = index.writer(15_000_000)?;

            for doc in docs {
                writer.add_document(doc)?;
            }

            writer.commit()?;

            tracing::info!(domain);
        }

        let mut file_cursor = File::create(&file_cursor_path)?;
        file_cursor.write_all(
            last_log_file
                .file_name()
                .expect("log file to have name")
                .to_str()
                .expect("log file name to str")
                .as_bytes(),
        )?;

        let mut line_cursor = File::create(&line_cursor_path)?;
        line_cursor.write_all((last_log_line + 1).to_string().as_bytes())?;

        let mut lock = file_cursor_mutex.lock();
        *lock = Some(last_log_file);

        drop(lock);

        tracing::info!("finish");

        Ok(())
    }

    #[instrument(skip(dir, file_cursor, domain_indexes, max_ttl_days), err)]
    #[allow(clippy::type_complexity)]
    fn clean(
        dir: &PathBuf,
        file_cursor: &Arc<Mutex<Option<PathBuf>>>,
        domain_indexes: &Arc<RwLock<HashMap<String, (Index, IndexReader, QueryParser)>>>,
        max_ttl_days: u16,
    ) -> Result<(), Box<dyn Error>> {
        tracing::info!("start");

        let file_cursor = file_cursor.lock();

        if let Some(file_cursor) = &*file_cursor {
            let mut log_files = std::fs::read_dir(dir)?
                .map(|res| res.map(|e| e.path()))
                .collect::<Result<Vec<_>, std::io::Error>>()?;
            log_files.sort_by(|a, b| b.cmp(a));

            for log_file in log_files {
                if log_file < *file_cursor && !log_file.is_dir() {
                    // todo: roll up, compress, and back up log files

                    std::fs::remove_file(&log_file)?;
                    tracing::info!(
                        log_file = log_file.as_os_str().to_str().unwrap_or("no/path"),
                        "removed"
                    );
                }
            }
        }

        drop(file_cursor);

        let di = domain_indexes.read();

        let exp: i64 = SystemTime::now()
            .checked_sub(Duration::from_hours(u64::from(max_ttl_days) * 24))
            .expect("time to work")
            .duration_since(SystemTime::UNIX_EPOCH)?
            .as_secs()
            .cast_signed();

        if let Some(kill_on_or_before_dt) = chrono::DateTime::from_timestamp_secs(exp) {
            let kill_on_or_before =
                kill_on_or_before_dt.to_rfc3339_opts(SecondsFormat::Micros, true);

            for (domain, (index, reader, parser)) in di.iter() {
                let query = parser.parse_query(&format!(
                    "timestamp:[1970-01-01T00:00:00.000000Z TO {kill_on_or_before}}}"
                ))?;

                let mut writer: IndexWriter = index.writer(15_000_000)?;

                writer.delete_query(query)?;

                writer.commit()?;

                let searcher = reader.searcher();
                let remaining_log_line_count = searcher.num_docs();

                tracing::info!(remaining_log_line_count, domain);
            }
        }

        tracing::info!("finish");

        Ok(())
    }

    #[allow(clippy::similar_names)]
    pub fn start(&self) -> Result<(JoinHandle<()>, JoinHandle<()>), Box<dyn Error>> {
        let cursor_path = self.dir.join("cursor");
        std::fs::create_dir_all(&cursor_path)?;

        let index_interval_s = self.index_interval;
        let clean_interval_s = self.clean_interval;

        let max_ttl_days = self.max_ttl_days;

        let dir = self.dir.clone();
        let dir_clone = dir.clone();
        let domain_indexes = self.domain_indexes.clone();
        let domain_indexes_clone = domain_indexes.clone();

        let monitor_span = self.span.clone();
        let monitor_span_clone = monitor_span.clone();

        let (mut index_min, mut index_max) = index_interval_s;

        index_min *= 1000;
        index_max *= 1000;

        let file_cursor_clone = self.file_cursor.clone();
        let file_cursor_clone2 = self.file_cursor.clone();

        let schema = self.schema.clone();

        let index_handle = tokio::spawn(async move {
            loop {
                let index_interval_ms = rand::random_range(index_min..index_max);

                tokio::time::sleep(Duration::from_millis(index_interval_ms)).await;

                monitor_span.in_scope(|| {
                    match Self::index(&dir, &file_cursor_clone, &domain_indexes, &schema) {
                        Ok(()) => {}
                        Err(err) => tracing::error!(%err),
                    }
                });
            }
        });

        let (mut clean_min, mut clean_max) = clean_interval_s;

        clean_min *= 1000;
        clean_max *= 1000;

        let clean_handle = tokio::spawn(async move {
            loop {
                let clean_interval_ms = rand::random_range(clean_min..clean_max);

                tokio::time::sleep(Duration::from_millis(clean_interval_ms)).await;

                monitor_span_clone.in_scope(|| {
                    match Self::clean(
                        &dir_clone,
                        &file_cursor_clone2,
                        &domain_indexes_clone,
                        max_ttl_days,
                    ) {
                        Ok(()) => {}
                        Err(err) => tracing::error!(%err),
                    }
                });
            }
        });

        Ok((index_handle, clean_handle))
    }
}
