use std::{
    io::{self, Write},
    path::Path,
    sync::Once,
};

use anyhow::{Context, Result};
use chrono::Local;
use flexi_logger::{
    Cleanup, Criterion, DeferredNow, FileSpec, LogSpecBuilder, LogSpecification, Logger,
    LoggerHandle, Naming,
};
use log::{LevelFilter, Record};

use crate::config::LogConfig;

const ROTATE_TIMESTAMP_FORMAT: &str = "%Y-%m-%d_%H-%M-%S";
const ROTATE_CURRENT_INFIX: &str = "";
const TEST_FILTER_MODULES: &[&str] = &[
    "tantivy",
    "aws_sdk_s3",
    "aws_smithy_http",
    "aws_config",
    "aws_smithy_client",
    "aws_credential_types",
    "hyper",
    "tracing",
];

static INIT_TEST_LOGGER: Once = Once::new();

fn rotate_naming() -> Naming {
    Naming::TimestampsCustomFormat {
        current_infix: Some(ROTATE_CURRENT_INFIX),
        format: ROTATE_TIMESTAMP_FORMAT,
    }
}

fn build_log_spec(log_level: &str, filter_modules: &[&str]) -> LogSpecification {
    let base_spec =
        LogSpecification::env_or_parse(log_level).unwrap_or_else(|_| LogSpecification::info());
    let mut log_spec_builder = LogSpecBuilder::from_module_filters(base_spec.module_filters());
    for module in filter_modules {
        log_spec_builder.module(module, log::LevelFilter::Warn);
    }
    log_spec_builder.build()
}

fn build_test_log_spec(level: LevelFilter, filter_modules: &[&str]) -> LogSpecification {
    let mut log_spec_builder = LogSpecBuilder::new();
    log_spec_builder.default(level);
    for module in filter_modules {
        log_spec_builder.module(module, LevelFilter::Error);
    }
    log_spec_builder.build()
}

pub fn init_logger(log_config: &LogConfig) -> Result<LoggerHandle> {
    const FILTER_MODULES: &[&str] = &[
        "tantivy",
        "aws_sdk_s3",
        "aws_smithy_http",
        "aws_config",
        "aws_smithy_client",
        "aws_credential_types",
        "hyper",
        "tracing",
        "h2",
    ];
    let log_spec = build_log_spec(&log_config.get_level_or("info"), FILTER_MODULES);

    let mut logger = Logger::with(log_spec)
        .format(|write, now, record| format_unified_log_for_flexi_logger(write, now, record));
    if !log_config.filename.is_empty() {
        let file_spec = FileSpec::try_from(log_config.filename.clone())
            .with_context(|| format!("Failed to parse log file path={}", log_config.filename))?
            .suppress_timestamp();
        logger = logger.log_to_file(file_spec).append().rotate(
            Criterion::Size(log_config.size.0),
            rotate_naming(),
            Cleanup::KeepCompressedFiles(log_config.count),
        );
    } else {
        logger = logger.log_to_stderr();
    }

    logger
        .start()
        .with_context(|| format!("Failed to initialize logger: log_config={:?}", log_config))
}

pub fn init_logger_for_ut() {
    init_logger_for_test(LevelFilter::Debug, TEST_FILTER_MODULES);
}

pub fn init_logger_for_test(level: LevelFilter, filter_modules: &[&str]) {
    INIT_TEST_LOGGER.call_once(|| {
        let logger = Logger::with(build_test_log_spec(level, filter_modules))
            .format(format_unified_log_for_flexi_logger)
            .log_to_stderr();
        if let Err(err) = logger.start() {
            panic!("failed to initialize flexi logger for unit tests, err={err:?}");
        }
    });
}

// Format log in a unified format:
// - Log Header Section
// [date_time] [LEVEL] [source_file:line_number]
// - Log Message Section
// [message]
// - Log Fields Section
// [key1=value1] [key2=value2] ...
// For more details, see https://github.com/tikv/rfcs/blob/ed764d7d014c420ee0cbcde99597020c4f75346d/text/0018-unified-log-format.md
//
pub fn format_unified_log_for_flexi_logger(
    write: &mut dyn Write,
    _now: &mut DeferredNow,
    record: &Record<'_>,
) -> io::Result<()> {
    write_unified_log_record(write, record)
}

fn write_unified_log_record<W: Write + ?Sized>(buf: &mut W, record: &Record<'_>) -> io::Result<()> {
    let timestamp = Local::now().format("%Y/%m/%d %H:%M:%S%.3f %:z");
    let level = record.level();
    let source_location = format_source_location(record);
    let message = record.args().to_string();

    write!(buf, "[{timestamp}] [{level}] [{source_location}] [")?;
    write_log_value(buf, &message)?;
    write!(buf, "]")?;

    if let Some(module_path) = record_module_path(record) {
        // `module_path` is much more useful than file name when multiple modules
        // share the same file name (e.g. `mod.rs`).
        write!(buf, " [mod=")?;
        write_log_value(buf, module_path)?;
        write!(buf, "]")?;
    }

    Ok(())
}

fn format_source_location(record: &Record<'_>) -> String {
    let file = match record.file() {
        Some(file) => file,
        None => return String::new(),
    };
    let file_name = match Path::new(file).file_name().and_then(|name| name.to_str()) {
        Some(file_name) => file_name,
        None => return String::new(),
    };
    let line = match record.line() {
        Some(line) => line,
        None => return String::new(),
    };

    let sanitized: String = file_name
        .chars()
        .filter(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_'))
        .collect();
    if sanitized.is_empty() {
        return String::new();
    }
    format!("{sanitized}:{line}")
}

fn record_module_path<'a>(record: &Record<'a>) -> Option<&'a str> {
    record.module_path().or_else(|| {
        let target = record.target();
        if target.is_empty() {
            None
        } else {
            Some(target)
        }
    })
}

fn write_log_value<W: Write + ?Sized>(buf: &mut W, value: &str) -> io::Result<()> {
    if message_needs_json(value) {
        serde_json::to_writer(buf, value).map_err(|err| io::Error::new(io::ErrorKind::Other, err))
    } else {
        write!(buf, "{value}")
    }
}

fn message_needs_json(message: &str) -> bool {
    message
        .chars()
        .any(|ch| ch <= '\u{20}' || matches!(ch, '"' | '=' | '[' | ']'))
}

#[cfg(test)]
mod tests {
    use std::io::Write;

    use flexi_logger::writers::FileLogWriter;
    use log::Level;

    use super::*;

    #[test]
    fn format_unified_log_prints_module_path() {
        let record = Record::builder()
            .args(format_args!("hello"))
            .level(Level::Info)
            .target("tici::meta::writer")
            .module_path(Some("tici::meta::writer"))
            .file(Some("src/meta/writer.rs"))
            .line(Some(42))
            .build();

        let mut buf = Vec::new();
        if let Err(err) = write_unified_log_record(&mut buf, &record) {
            panic!("write_unified_log_record failed, err={err:?}");
        }

        let line = match std::str::from_utf8(&buf) {
            Ok(line) => line,
            Err(err) => panic!("invalid utf-8 log line, err={err:?}, bytes={buf:?}"),
        };
        assert!(line.contains("[writer.rs:42]"), "line={line}");
        assert!(line.contains("[hello]"), "line={line}");
        assert!(line.contains("[mod=tici::meta::writer]"), "line={line}");
    }

    #[test]
    fn format_unified_log_falls_back_to_target_when_module_path_is_missing() {
        let record = Record::builder()
            .args(format_args!("hello"))
            .level(Level::Info)
            .target("tici::meta::writer")
            .file(Some("src/meta/writer.rs"))
            .line(Some(42))
            .build();

        let mut buf = Vec::new();
        if let Err(err) = write_unified_log_record(&mut buf, &record) {
            panic!("write_unified_log_record failed, err={err:?}");
        }

        let line = match std::str::from_utf8(&buf) {
            Ok(line) => line,
            Err(err) => panic!("invalid utf-8 log line, err={err:?}, bytes={buf:?}"),
        };
        assert!(line.contains("[mod=tici::meta::writer]"), "line={line}");
    }

    #[test]
    fn format_unified_log_record_does_not_append_newline() {
        let record = Record::builder()
            .args(format_args!("hello"))
            .level(Level::Info)
            .target("tici::meta::writer")
            .module_path(Some("tici::meta::writer"))
            .file(Some("src/meta/writer.rs"))
            .line(Some(42))
            .build();

        let mut buf = Vec::new();
        if let Err(err) = write_unified_log_record(&mut buf, &record) {
            panic!("write_unified_log_record failed, err={err:?}");
        }

        let line = match std::str::from_utf8(&buf) {
            Ok(line) => line,
            Err(err) => panic!("invalid utf-8 log line, err={err:?}, bytes={buf:?}"),
        };
        assert!(!line.ends_with('\n'), "line={line:?}");
    }

    #[test]
    fn rotate_naming_uses_plain_current_file_name() {
        match rotate_naming() {
            Naming::TimestampsCustomFormat {
                current_infix,
                format,
            } => {
                assert_eq!(current_infix, Some(""));
                assert_eq!(format, "%Y-%m-%d_%H-%M-%S");
            }
            naming => panic!("unexpected naming variant: {naming:?}"),
        }
    }

    #[test]
    fn rotate_naming_does_not_create_rcurrent_file() {
        let temp_dir = match tempfile::tempdir() {
            Ok(temp_dir) => temp_dir,
            Err(err) => panic!("create temp dir failed, err={err:?}"),
        };
        let log_path = temp_dir.path().join("tici.log");
        let file_spec = match FileSpec::try_from(log_path.clone()) {
            Ok(file_spec) => file_spec.suppress_timestamp(),
            Err(err) => panic!("build FileSpec failed, err={err:?}"),
        };
        let (mut writer, _writer_handle) = match FileLogWriter::builder(file_spec)
            .append()
            .rotate(Criterion::Size(128), rotate_naming(), Cleanup::Never)
            .try_build_with_handle()
        {
            Ok(writer_and_handle) => writer_and_handle,
            Err(err) => panic!("build file log writer failed, err={err:?}"),
        };

        if let Err(err) = writer.write_all(b"first line\n") {
            panic!("write first log line failed, err={err:?}");
        }
        if let Err(err) = writer.flush() {
            panic!("flush first log line failed, err={err:?}");
        }
        if let Err(err) = writer.rotate() {
            panic!("rotate log file failed, err={err:?}");
        }
        if let Err(err) = writer.write_all(b"second line\n") {
            panic!("write second log line failed, err={err:?}");
        }
        if let Err(err) = writer.flush() {
            panic!("flush second log line failed, err={err:?}");
        }

        let mut names = match std::fs::read_dir(temp_dir.path()) {
            Ok(entries) => entries
                .filter_map(Result::ok)
                .map(|entry| entry.path())
                .collect::<Vec<_>>(),
            Err(err) => panic!("read log directory failed, err={err:?}"),
        }
        .into_iter()
        .map(|path| {
            path.file_name()
                .map(|name| name.to_string_lossy().to_string())
                .unwrap_or_default()
        })
        .collect::<Vec<_>>();
        names.sort();

        assert!(
            names.iter().any(|name| name == "tici.log"),
            "expected fixed current file `tici.log`, names={names:?}"
        );
        assert!(
            names
                .iter()
                .any(|name| name.starts_with("tici_") && name.ends_with(".log")),
            "expected timestamp file name, names={names:?}"
        );
        assert!(
            names.iter().all(|name| !name.contains("rCURRENT")),
            "unexpected rCURRENT file, names={names:?}"
        );
    }
}