use anyhow::{ensure, Context, Result};
use arc_swap::ArcSwap;
use axum::body::Body;
use http::{Request, Uri};
use http_body_util::{BodyExt, Limited};
use hyper_rustls::HttpsConnectorBuilder;
use hyper_util::{
    client::legacy::{connect::HttpConnector, Client},
    rt::TokioExecutor,
};
use rustls::{ClientConfig, RootCertStore};
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use std::{
    collections::{BTreeMap, VecDeque},
    io::{self, Write as IoWrite},
    path::PathBuf,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc, Mutex, OnceLock,
    },
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tokio::{
    fs::File,
    io::AsyncWriteExt,
    net::TcpStream,
    sync::{broadcast, mpsc, watch},
};
use tokio_util::sync::CancellationToken;
use tracing::{Event, Level, Subscriber};
use tracing_subscriber::{layer::Context as LayerContext, Layer};

pub const LIVE_WINDOW: usize = 256;
pub const HISTORY_CAPACITY: usize = 4096;
pub const LIVE_STREAM_CAPACITY: usize = 4096;
const SINK_QUEUE: usize = 512;
const SINK_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct FileSink {
    pub enabled: bool,
    pub path: String,
    /// Write one JSONL file per resolved vhost instead of one combined file.
    #[serde(default)]
    pub per_vhost: bool,
    /// Directory used by per-vhost mode. Files are named `<vhost>.jsonl`;
    /// unmatched requests are written to `_unmatched.jsonl`.
    #[serde(default = "default_vhost_log_directory")]
    pub directory: String,
}
fn default_vhost_log_directory() -> String {
    "/var/log/rustproxy/vhosts".into()
}
impl Default for FileSink {
    fn default() -> Self {
        Self {
            enabled: false,
            path: "/var/log/rustproxy/access.jsonl".into(),
            per_vhost: false,
            directory: default_vhost_log_directory(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct LogstashSink {
    pub enabled: bool,
    pub address: String,
}
impl Default for LogstashSink {
    fn default() -> Self {
        Self {
            enabled: false,
            address: "127.0.0.1:5044".into(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct LokiSink {
    pub enabled: bool,
    pub url: String,
    pub bearer_token_env: Option<String>,
    pub labels: BTreeMap<String, String>,
}
impl Default for LokiSink {
    fn default() -> Self {
        Self {
            enabled: false,
            url: "http://127.0.0.1:3100/loki/api/v1/push".into(),
            bearer_token_env: None,
            labels: BTreeMap::from([("job".into(), "rustproxy".into())]),
        }
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct JsonFormat {
    /// output field -> LogEntry field. Empty means the complete standard entry.
    pub fields: BTreeMap<String, String>,
    pub static_fields: BTreeMap<String, String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct Settings {
    /// Capture a bounded User-Agent value in access logs and traffic analytics.
    pub capture_user_agent: bool,
    /// Resolve the effective client IP through the loaded GeoIP database.
    /// Disabled by default because this adds per-request lookup work.
    pub capture_country: bool,
    pub file: FileSink,
    pub logstash: LogstashSink,
    pub loki: LokiSink,
    pub format: JsonFormat,
}
impl Default for Settings {
    fn default() -> Self {
        Self {
            capture_user_agent: true,
            capture_country: false,
            file: FileSink::default(),
            logstash: LogstashSink::default(),
            loki: LokiSink::default(),
            format: JsonFormat::default(),
        }
    }
}
impl Settings {
    pub fn enabled(&self) -> bool {
        self.file.enabled || self.logstash.enabled || self.loki.enabled
    }
    pub fn validate(&self) -> Result<()> {
        if self.file.enabled {
            if self.file.per_vhost {
                ensure!(
                    !self.file.directory.trim().is_empty() && self.file.directory.len() <= 1024,
                    "logging.file.directory must be 1..1024 bytes in per-vhost mode"
                );
            } else {
                ensure!(
                    !self.file.path.trim().is_empty() && self.file.path.len() <= 1024,
                    "logging.file.path must be 1..1024 bytes"
                );
            }
        }
        if self.logstash.enabled {
            let authority = self
                .logstash
                .address
                .parse::<http::uri::Authority>()
                .context("logging.logstash.address must be host:port")?;
            ensure!(
                !authority.host().is_empty()
                    && authority.port_u16().is_some()
                    && !authority.as_str().contains('@'),
                "logging.logstash.address must be host:port"
            );
        }
        if self.loki.enabled {
            ensure!(self.loki.url.len() <= 2048, "logging.loki.url is too long");
            let url = url::Url::parse(&self.loki.url).context("invalid logging.loki.url")?;
            ensure!(
                matches!(url.scheme(), "http" | "https") && url.host().is_some(),
                "logging.loki.url must be HTTP(S)"
            );
            ensure!(
                url.username().is_empty() && url.password().is_none(),
                "credentials in logging.loki.url are forbidden; use bearer_token_env"
            );
        }
        if let Some(name) = &self.loki.bearer_token_env {
            ensure!(
                !name.is_empty()
                    && name.len() <= 128
                    && name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_'),
                "logging.loki.bearer_token_env must be a simple environment variable name"
            );
        }
        ensure!(
            self.loki.labels.len() <= 32,
            "at most 32 Loki labels are allowed"
        );
        for (key, value) in &self.loki.labels {
            ensure!(
                !key.is_empty()
                    && key.len() <= 64
                    && key.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_'),
                "Loki label names must be 1..64 letters, numbers or underscores"
            );
            ensure!(
                value.len() <= 256,
                "Loki label values must be at most 256 bytes"
            );
        }
        ensure!(
            self.format.fields.len() <= 32 && self.format.static_fields.len() <= 32,
            "structured log format allows at most 32 mapped and 32 static fields"
        );
        for (output, source) in &self.format.fields {
            validate_json_key(output)?;
            ensure!(
                source_value_name(source),
                "unknown structured log source field: {source}"
            );
        }
        for (key, value) in &self.format.static_fields {
            validate_json_key(key)?;
            ensure!(
                value.len() <= 1024,
                "structured log static value is too long"
            );
        }
        Ok(())
    }
}
fn validate_json_key(key: &str) -> Result<()> {
    ensure!(
        !key.is_empty() && key.len() <= 128 && !key.chars().any(char::is_control),
        "structured log output field names must be 1..128 bytes without controls"
    );
    Ok(())
}
fn source_value_name(name: &str) -> bool {
    matches!(
        name,
        "event"
            | "protocol_error"
            | "reason"
            | "error_detail"
            | "error"
            | "local_addr"
            | "tls_sni"
            | "sni"
            | "authority"
            | "matched_by"
            | "vhost"
            | "route"
            | "cache_effective_enabled"
            | "transform_effective_enabled"
            | "backend"
            | "id"
            | "request_id"
            | "time_ms"
            | "method"
            | "host"
            | "path"
            | "http_version"
            | "transport"
            | "websocket_bootstrap"
            | "websocket_phase"
            | "websocket_error"
            | "h3_reset_code"
            | "upstream_status"
            | "upstream_error"
            | "client_ip"
            | "user_agent"
            | "country"
            | "status"
            | "duration_ms"
            | "request_bytes"
            | "response_bytes"
            | "outcome"
            | "cache"
            | "cache_reason"
    )
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum ErrorLogFormat {
    #[serde(rename = "JSON")]
    Json,
    #[serde(rename = "RAW")]
    Raw,
}
impl Default for ErrorLogFormat {
    fn default() -> Self {
        Self::Json
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct ErrorLogSettings {
    pub enabled: bool,
    pub path: String,
    pub format: ErrorLogFormat,
}
impl Default for ErrorLogSettings {
    fn default() -> Self {
        Self {
            enabled: false,
            path: "/var/log/rustproxy/error.log".into(),
            format: ErrorLogFormat::Json,
        }
    }
}
impl ErrorLogSettings {
    pub fn validate(&self) -> Result<()> {
        if self.enabled {
            ensure!(
                !self.path.trim().is_empty()
                    && self.path.len() <= 1024
                    && !self.path.chars().any(char::is_control),
                "error_log.path must be 1..1024 bytes without controls"
            );
        }
        Ok(())
    }
}

const ERROR_LOG_QUEUE: usize = 512;
const ERROR_LOG_RETRY: Duration = Duration::from_secs(30);

#[derive(Debug, Clone, Serialize)]
pub struct ErrorLogStats {
    pub queue_capacity: usize,
    pub queue_dropped: u64,
    pub written: u64,
    pub errors: u64,
}

#[derive(Debug)]
struct ErrorEvent {
    timestamp: String,
    level: String,
    target: String,
    message: String,
    fields: BTreeMap<String, String>,
}

pub struct ErrorLogController {
    settings: Arc<ArcSwap<ErrorLogSettings>>,
    settings_changed: watch::Sender<()>,
    sender: mpsc::Sender<ErrorEvent>,
    dropped: Arc<AtomicU64>,
    written: Arc<AtomicU64>,
    errors: Arc<AtomicU64>,
}
impl ErrorLogController {
    pub fn start() -> Arc<Self> {
        let (sender, receiver) = mpsc::channel(ERROR_LOG_QUEUE);
        let (settings_changed, settings_watch) = watch::channel(());
        let settings = Arc::new(ArcSwap::from_pointee(ErrorLogSettings::default()));
        let dropped = Arc::new(AtomicU64::new(0));
        let written = Arc::new(AtomicU64::new(0));
        let errors = Arc::new(AtomicU64::new(0));
        tokio::spawn(error_log_loop(
            receiver,
            settings.clone(),
            settings_watch,
            written.clone(),
            errors.clone(),
        ));
        Arc::new(Self {
            settings,
            settings_changed,
            sender,
            dropped,
            written,
            errors,
        })
    }
    pub fn configure(&self, settings: ErrorLogSettings) {
        self.settings.store(Arc::new(settings));
        self.settings_changed.send_replace(());
    }
    /// All diagnostics go to stderr when file logging is disabled. With a file
    /// configured, warnings/errors go only to that file (unless the sink fails).
    /// Informational startup messages remain on the console.
    pub fn console_enabled_for(&self, level: &Level) -> bool {
        !matches!(*level, Level::WARN | Level::ERROR) || !self.settings.load().enabled
    }
    fn push(&self, event: ErrorEvent) {
        let config = self.settings.load();
        if !config.enabled {
            return;
        }
        if let Err(error) = self.sender.try_send(event) {
            self.dropped.fetch_add(1, Ordering::Relaxed);
            // An overloaded/closed sink must not silently swallow a diagnostic
            // suppressed by the console filter. This emergency path is bounded
            // to one event and only runs on a failed non-blocking enqueue.
            write_error_fallback(&error.into_inner(), config.format);
        }
    }
    pub fn stats(&self) -> ErrorLogStats {
        ErrorLogStats {
            queue_capacity: ERROR_LOG_QUEUE,
            queue_dropped: self.dropped.load(Ordering::Relaxed),
            written: self.written.load(Ordering::Relaxed),
            errors: self.errors.load(Ordering::Relaxed),
        }
    }
}

static ERROR_LOG: OnceLock<Arc<ErrorLogController>> = OnceLock::new();

pub fn install_error_log_controller(controller: Arc<ErrorLogController>) -> Result<()> {
    ERROR_LOG
        .set(controller)
        .map_err(|_| anyhow::anyhow!("error logger already initialized"))
}
pub fn configure_error_log(settings: ErrorLogSettings) {
    if let Some(controller) = ERROR_LOG.get() {
        controller.configure(settings);
    }
}
pub fn error_log_stats() -> Option<ErrorLogStats> {
    ERROR_LOG.get().map(|controller| controller.stats())
}

#[derive(Clone)]
pub struct ErrorLogLayer {
    controller: Arc<ErrorLogController>,
}
impl ErrorLogLayer {
    pub fn new(controller: Arc<ErrorLogController>) -> Self {
        Self { controller }
    }
}

#[derive(Default)]
struct EventVisitor {
    message: Option<String>,
    fields: BTreeMap<String, String>,
}
impl tracing::field::Visit for EventVisitor {
    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        let value = format!("{value:?}");
        if field.name() == "message" {
            self.message = Some(value);
        } else {
            self.fields.insert(field.name().to_owned(), value);
        }
    }
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        if field.name() == "message" {
            self.message = Some(value.to_owned());
        } else {
            self.fields
                .insert(field.name().to_owned(), value.to_owned());
        }
    }
    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
        self.fields
            .insert(field.name().to_owned(), value.to_string());
    }
    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
        self.fields
            .insert(field.name().to_owned(), value.to_string());
    }
    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
        self.fields
            .insert(field.name().to_owned(), value.to_string());
    }
    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
        self.fields
            .insert(field.name().to_owned(), value.to_string());
    }
}
impl<S: Subscriber> Layer<S> for ErrorLogLayer {
    fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) {
        if !matches!(*event.metadata().level(), Level::ERROR | Level::WARN) {
            return;
        }
        let mut visitor = EventVisitor::default();
        event.record(&mut visitor);
        let mut fields = visitor.fields;
        for (key, value) in &mut fields {
            *value = if is_secret_field(key) {
                "[REDACTED]".into()
            } else {
                redact_text(value)
            };
        }
        self.controller.push(ErrorEvent {
            timestamp: rfc3339_now(),
            level: event.metadata().level().as_str().to_owned(),
            target: event.metadata().target().to_owned(),
            message: redact_text(visitor.message.as_deref().unwrap_or("error")),
            fields,
        });
    }
}

#[derive(Clone, Copy, Default)]
pub struct RedactingStderr;
pub struct RedactingWriter {
    buffer: Vec<u8>,
}