//! Prometheus metrics: application counters/gauges, optional Linux process stats, HTTP middleware.

use actix_web::body::MessageBody;
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::middleware::Next;
use actix_web::{web, Error, HttpResponse};
use prometheus::proto::MetricFamily;
use prometheus::{
    Encoder, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts,
    Registry, TextEncoder,
};

use prometheus::core::{AtomicU64, GenericCounter};
use std::sync::Arc;
use std::time::Instant;

/// Buckets for HTTP handler latency (seconds), suitable for API + scrape workloads.
const HTTP_DURATION_BUCKETS: &[f64] = &[
    0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0,
];

/// Body size buckets (bytes) for HTML and JSON scrape payloads.
const SCRAPE_BODY_BYTES_BUCKETS: &[f64] = &[
    256.0, 1024.0, 4096.0, 16384.0, 65536.0, 262_144.0, 1_048_576.0, 4_194_304.0, 16_777_216.0,
];

/// Redis URL registry round-trip latency (connect + one command).
const REDIS_REGISTRY_DURATION_BUCKETS: &[f64] = &[
    0.000_1, 0.000_5, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0,
];

pub struct AppMetrics {
    pub scrape_requests_total: IntCounterVec,
    pub scrape_duration_seconds: HistogramVec,
    /// Raw HTML length from upstream fetch (network path only).
    pub scrape_upstream_html_bytes: HistogramVec,
    /// Serialized JSON body size returned to the client (`cache_hit` vs freshly `fetched`).
    pub scrape_api_response_json_bytes: HistogramVec,
    /// Redis SADD/SISMEMBER URL registry attempts (outcome: ok | connect_error | command_error).
    pub redis_url_registry_operations_total: IntCounterVec,
    /// Wall time for multiplexed connection + one registry command.
    pub redis_url_registry_duration_seconds: HistogramVec,
    pub registry_checks_total: IntCounter,
    pub cache_hits_total: IntCounter,
    pub cache_misses_total: IntCounter,
    pub audit_log_dropped_total: IntCounter,
    pub audit_log_flushed_total: IntCounter,
    /// Rows reported by CLI/global scrape pipelines (`rate_add_scraped`), also exposed when wired.
    pub pipeline_rows_scraped_total: IntCounter,
    /// Rows successfully inserted via Athena in scrape binaries when wired.
    pub pipeline_rows_inserted_total: IntCounter,
    pub http_requests_total: IntCounterVec,
    pub http_request_duration_seconds: HistogramVec,
    pub moka_response_cache_entries: IntGauge,
    pub athena_configured: IntGauge,
    pub redis_configured: IntGauge,
    pub scraper_yaml_configs: IntGauge,
    pub scraper_overrides_in_memory: IntGauge,
    pub http_profiles_enabled: IntGauge,
    pub screenshot_service_enabled: IntGauge,
    pub publications_service_enabled: IntGauge,
    pub audit_bus_enabled: IntGauge,
    pub bloom_filter_bit_length: IntGauge,
    pub bloom_filter_number_of_hashes: IntGauge,
    pub uptime_seconds: IntGauge,
    pub worker_threads: IntGauge,
    pub build_info: IntGaugeVec,
    start_time: Instant,
}

impl AppMetrics {
    pub fn new(registry: &Registry, worker_thread_count: usize) -> anyhow::Result<Self> {
        let scrape_requests_total = IntCounterVec::new(
            Opts::new(
                "scrape_requests_total",
                "Total scrape API invocations (GET/POST /scrape). Labels: config_name, status — \
                 status is not_found | cache_hit | error | success, or Athena persistence outcome when enabled.",
            ),
            &["config_name", "status"],
        )?;
        let scrape_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "scrape_duration_seconds",
                "Wall time for a full scrape handler run (template expand, optional HTTP fetch, HTML parse, selectors) in seconds.",
            )
            .buckets(vec![
                0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0,
            ]),
            &["config_name"],
        )?;
        let scrape_upstream_html_bytes = HistogramVec::new(
            HistogramOpts::new(
                "scrape_upstream_html_bytes",
                "Byte length of the upstream HTML document read before CSS parsing (network scrape path only).",
            )
            .buckets(SCRAPE_BODY_BYTES_BUCKETS.to_vec()),
            &["config_name"],
        )?;
        let scrape_api_response_json_bytes = HistogramVec::new(
            HistogramOpts::new(
                "scrape_api_response_json_bytes",
                "Byte length of the JSON body returned from GET/POST /scrape. Label source=cache_hit (Moka) or fetched (after network + selectors).",
            )
            .buckets(SCRAPE_BODY_BYTES_BUCKETS.to_vec()),
            &["config_name", "source"],
        )?;
        let redis_url_registry_operations_total = IntCounterVec::new(
            Opts::new(
                "redis_url_registry_operations_total",
                "URL registry Redis usage from /api/cache/registry/*. Label operation=add|check; outcome=ok|connect_error|command_error.",
            ),
            &["operation", "outcome"],
        )?;
        let redis_url_registry_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "redis_url_registry_duration_seconds",
                "Time to obtain a multiplexed Redis connection and run SADD (add) or SISMEMBER (check) for the URL registry.",
            )
            .buckets(REDIS_REGISTRY_DURATION_BUCKETS.to_vec()),
            &["operation"],
        )?;
        let registry_checks_total: GenericCounter<AtomicU64> = IntCounter::new(
            "registry_checks_total",
            "Redis-backed URL registry check calls (GET /api/cache/registry/check), including bloom + Redis round-trips.",
        )?;
        let cache_hits_total: GenericCounter<AtomicU64> = IntCounter::new(
            "cache_hits_total",
            "Moka in-memory scrape response cache hits (same config_name + URL key).",
        )?;
        let cache_misses_total: GenericCounter<AtomicU64> = IntCounter::new(
            "cache_misses_total",
            "Moka cache misses and registry paths that did not short-circuit to a cached scrape body.",
        )?;
        let audit_log_dropped_total: GenericCounter<AtomicU64> = IntCounter::new(
            "audit_log_dropped_total",
            "HTTP or DB audit events dropped because the async audit channel was full or closed.",
        )?;
        let audit_log_flushed_total: GenericCounter<AtomicU64> = IntCounter::new(
            "audit_log_flushed_total",
            "Audit rows successfully submitted to Athena (http_request_log and db_operation_log batches).",
        )?;
        let pipeline_rows_scraped_total: GenericCounter<AtomicU64> = IntCounter::new(
            "dexter_pipeline_rows_scraped_total",
            "Cumulative rows mapped from external registries in dexter-scrape / nl-scrape style binaries (wired from the HTTP server when running embedded pipelines is not used).",
        )?;
        let pipeline_rows_inserted_total: GenericCounter<AtomicU64> = IntCounter::new(
            "dexter_pipeline_rows_inserted_total",
            "Cumulative successful Athena gateway inserts from scrape worker binaries (global + country tables).",
        )?;
        let http_requests_total = IntCounterVec::new(
            Opts::new(
                "http_requests_total",
                "All HTTP requests seen by Actix after middleware. Labels: method, normalized_path (IDs replaced), status (numeric code or 'error' if handler failed before response).",
            ),
            &["method", "path", "status"],
        )?;
        let http_request_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                "http_request_duration_seconds",
                "End-to-end request duration inside Actix for each route (includes handler + middleware stack below this layer).",
            )
            .buckets(HTTP_DURATION_BUCKETS.to_vec()),
            &["method", "path"],
        )?;
        let moka_response_cache_entries = IntGauge::new(
            "moka_response_cache_entries",
            "Approximate number of entries in the Moka scrape response cache (config_name:url → JSON body).",
        )?;
        let athena_configured = IntGauge::new(
            "athena_client_configured",
            "1 if ATHENA_URL + ATHENA_API_KEY produced a live AthenaClient at startup, else 0.",
        )?;
        let redis_configured = IntGauge::new(
            "redis_client_configured",
            "1 if REDIS_URL was valid and a redis::Client was created, else 0.",
        )?;
        let scraper_yaml_configs = IntGauge::new(
            "scraper_yaml_configs_loaded",
            "Count of scraper definitions loaded from config/scrapers YAML at startup (before Athena overrides).",
        )?;
        let scraper_overrides_in_memory = IntGauge::new(
            "scraper_overrides_in_memory",
            "Rows currently held in the in-memory scraper_overrides cache (Athena-backed definitions).",
        )?;
        let http_profiles_enabled = IntGauge::new(
            "http_profiles_enabled_count",
            "Enabled HTTP client profiles loaded from Athena into memory for merge with scraper request settings.",
        )?;
        let screenshot_service_enabled = IntGauge::new(
            "screenshot_service_enabled",
            "1 if optional headless screenshot service initialized from env, else 0.",
        )?;
        let publications_service_enabled = IntGauge::new(
            "publications_s3_service_enabled",
            "1 if optional PWC AL publications S3 sync service initialized, else 0.",
        )?;
        let audit_bus_enabled = IntGauge::new(
            "audit_bus_enabled",
            "1 if Athena audit flusher (http_request_log / db_operation_log) is active, else 0.",
        )?;
        let bloom_filter_bit_length = IntGauge::new(
            "bloom_filter_bit_length_bits",
            "Underlying bloom filter bit vector length for the URL bloom cache (approximate capacity tradeoff).",
        )?;
        let bloom_filter_number_of_hashes = IntGauge::new(
            "bloom_filter_hash_functions",
            "Number of hash functions configured for the URL bloom filter.",
        )?;
        let uptime_seconds =
            IntGauge::new("dexter_sentinel_uptime_seconds", "Process uptime since HTTP server start, in whole seconds.")?;
        let worker_threads = IntGauge::new(
            "dexter_sentinel_worker_threads",
            "Actix HttpServer worker thread count (MAX_WORKER_THREADS or default parallelism).",
        )?;
        let build_info = IntGaugeVec::new(
            Opts::new(
                "dexter_sentinel_build_info",
                "Always 1; labels identify this binary build. Use `version` label for the crate semver.",
            ),
            &["version"],
        )?;

        registry.register(Box::new(scrape_requests_total.clone()))?;
        registry.register(Box::new(scrape_duration_seconds.clone()))?;
        registry.register(Box::new(scrape_upstream_html_bytes.clone()))?;
        registry.register(Box::new(scrape_api_response_json_bytes.clone()))?;
        registry.register(Box::new(redis_url_registry_operations_total.clone()))?;
        registry.register(Box::new(redis_url_registry_duration_seconds.clone()))?;
        registry.register(Box::new(registry_checks_total.clone()))?;
        registry.register(Box::new(cache_hits_total.clone()))?;
        registry.register(Box::new(cache_misses_total.clone()))?;
        registry.register(Box::new(audit_log_dropped_total.clone()))?;
        registry.register(Box::new(audit_log_flushed_total.clone()))?;
        registry.register(Box::new(pipeline_rows_scraped_total.clone()))?;
        registry.register(Box::new(pipeline_rows_inserted_total.clone()))?;
        registry.register(Box::new(http_requests_total.clone()))?;
        registry.register(Box::new(http_request_duration_seconds.clone()))?;
        registry.register(Box::new(moka_response_cache_entries.clone()))?;
        registry.register(Box::new(athena_configured.clone()))?;
        registry.register(Box::new(redis_configured.clone()))?;
        registry.register(Box::new(scraper_yaml_configs.clone()))?;
        registry.register(Box::new(scraper_overrides_in_memory.clone()))?;
        registry.register(Box::new(http_profiles_enabled.clone()))?;
        registry.register(Box::new(screenshot_service_enabled.clone()))?;
        registry.register(Box::new(publications_service_enabled.clone()))?;
        registry.register(Box::new(audit_bus_enabled.clone()))?;
        registry.register(Box::new(bloom_filter_bit_length.clone()))?;
        registry.register(Box::new(bloom_filter_number_of_hashes.clone()))?;
        registry.register(Box::new(uptime_seconds.clone()))?;
        registry.register(Box::new(worker_threads.clone()))?;
        registry.register(Box::new(build_info.clone()))?;

        worker_threads.set(worker_thread_count as i64);
        build_info
            .with_label_values(&[env!("CARGO_PKG_VERSION")])
            .set(1);

        Ok(Self {
            scrape_requests_total,
            scrape_duration_seconds,
            scrape_upstream_html_bytes,
            scrape_api_response_json_bytes,
            redis_url_registry_operations_total,
            redis_url_registry_duration_seconds,
            registry_checks_total,
            cache_hits_total,
            cache_misses_total,
            audit_log_dropped_total,
            audit_log_flushed_total,
            pipeline_rows_scraped_total,
            pipeline_rows_inserted_total,
            http_requests_total,
            http_request_duration_seconds,
            moka_response_cache_entries,
            athena_configured,
            redis_configured,
            scraper_yaml_configs,
            scraper_overrides_in_memory,
            http_profiles_enabled,
            screenshot_service_enabled,
            publications_service_enabled,
            audit_bus_enabled,
            bloom_filter_bit_length,
            bloom_filter_number_of_hashes,
            uptime_seconds,
            worker_threads,
            build_info,
            start_time: Instant::now(),
        })
    }

    pub fn set_bloom_filter_stats(&self, bit_len: u64, number_of_hashes: u32) {
        self.bloom_filter_bit_length
            .set(i64::try_from(bit_len).unwrap_or(i64::MAX));
        self.bloom_filter_number_of_hashes
            .set(i64::from(number_of_hashes));
    }

    pub fn update_runtime_metrics(&self) {
        let uptime = self.start_time.elapsed().as_secs();
        self.uptime_seconds
            .set(i64::try_from(uptime).unwrap_or(i64::MAX));
    }

    /// Refresh gauges that reflect `AppState` (call from `/metrics` handler).
    pub fn update_from_app_state(&self, state: &crate::AppState) {
        self.update_runtime_metrics();
        self.moka_response_cache_entries
            .set(state.moka.entry_count() as i64);
        self.athena_configured
            .set(if state.athena.is_some() { 1 } else { 0 });
        self.redis_configured
            .set(if state.redis_client.is_some() { 1 } else { 0 });
        self.scraper_yaml_configs
            .set(state.scraper_configs.len() as i64);
        let ov = state
            .scraper_overrides_cache
            .read()
            .map(|g| g.len())
            .unwrap_or(0);
        self.scraper_overrides_in_memory.set(ov as i64);
        let hp = state
            .http_profile_cache
            .read()
            .map(|g| g.len())
            .unwrap_or(0);
        self.http_profiles_enabled.set(hp as i64);
        self.screenshot_service_enabled
            .set(if state.screenshot.is_some() { 1 } else { 0 });
        self.publications_service_enabled
            .set(if state.publications.is_some() { 1 } else { 0 });
        self.audit_bus_enabled
            .set(if state.audit.is_some() { 1 } else { 0 });
    }
}

/// Register Linux process metrics (`process_*` CPU, memory, fds, …) when supported.
pub fn register_process_collector(registry: &Registry) {
    #[cfg(target_os = "linux")]
    {
        use prometheus::process_collector::ProcessCollector;
        if let Err(e) = registry.register(Box::new(ProcessCollector::for_self())) {
            tracing::warn!("Could not register prometheus process_collector: {}", e);
        }
    }
    #[cfg(not(target_os = "linux"))]
    let _ = registry;
}

/// Normalize URL path for metric labels: numeric segments and UUID-like tokens → `{id}` to limit cardinality.
pub fn normalize_http_path(path: &str) -> String {
    fn is_uuid_like(seg: &str) -> bool {
        let b = seg.as_bytes();
        b.len() == 36
            && b.get(8) == Some(&b'-')
            && b.get(13) == Some(&b'-')
            && b.get(18) == Some(&b'-')
            && b.get(23) == Some(&b'-')
            && seg
                .chars()
                .all(|c| c.is_ascii_hexdigit() || c == '-')
    }
    let trimmed = path.trim_start_matches('/');
    if trimmed.is_empty() {
        return "/".to_string();
    }
    let mut out: Vec<&str> = Vec::new();
    for seg in trimmed.split('/') {
        if seg.is_empty() {
            continue;
        }
        if seg.chars().all(|c| c.is_ascii_digit()) || is_uuid_like(seg) {
            if out.last() != Some(&"{id}") {
                out.push("{id}");
            }
        } else {
            out.push(seg);
        }
    }
    if out.is_empty() {
        "/".to_string()
    } else {
        format!("/{}", out.join("/"))
    }
}

pub async fn prometheus_http_middleware(
    req: ServiceRequest,
    next: Next<impl MessageBody>,
) -> Result<ServiceResponse<impl MessageBody>, Error> {
    let start = Instant::now();
    let method = req.method().as_str().to_string();
    let path_key = normalize_http_path(req.path());
    let metrics = req
        .app_data::<web::Data<crate::AppState>>()
        .map(|d| d.metrics.clone());

    match next.call(req).await {
        Ok(res) => {
            if let Some(m) = metrics {
                let elapsed = start.elapsed().as_secs_f64();
                let code = res.status().as_u16().to_string();
                m.http_request_duration_seconds
                    .with_label_values(&[method.as_str(), &path_key])
                    .observe(elapsed);
                m.http_requests_total
                    .with_label_values(&[method.as_str(), &path_key, &code])
                    .inc();
            }
            Ok(res)
        }
        Err(e) => {
            if let Some(m) = metrics {
                let elapsed = start.elapsed().as_secs_f64();
                m.http_request_duration_seconds
                    .with_label_values(&[method.as_str(), &path_key])
                    .observe(elapsed);
                m.http_requests_total
                    .with_label_values(&[method.as_str(), &path_key, "error"])
                    .inc();
            }
            Err(e)
        }
    }
}

pub async fn metrics_handler(
    registry: web::Data<Arc<Registry>>,
    metrics: web::Data<Arc<AppMetrics>>,
    state: web::Data<crate::AppState>,
) -> HttpResponse {
    metrics.update_from_app_state(state.get_ref());
    let encoder: TextEncoder = TextEncoder::new();
    let metric_families: Vec<MetricFamily> = registry.gather();
    let mut buffer: Vec<u8> = Vec::new();
    if let Err(e) = encoder.encode(&metric_families, &mut buffer) {
        tracing::error!("Failed to encode metrics: {}", e);
        return HttpResponse::InternalServerError().finish();
    }
    HttpResponse::Ok()
        .content_type("text/plain; version=0.0.4; charset=utf-8")
        .append_header(("Cache-Control", "no-store"))
        .body(buffer)
}

#[cfg(test)]
mod tests {
    use super::normalize_http_path;

    #[test]
    fn normalize_collapses_numeric_and_uuid_segments() {
        assert_eq!(
            normalize_http_path("/api/provider-requests/550e8400-e29b-41d4-a716-446655440000"),
            "/api/provider-requests/{id}"
        );
        assert_eq!(
            normalize_http_path("/scrape/123/extra"),
            "/scrape/{id}/extra"
        );
        assert_eq!(normalize_http_path("/metrics"), "/metrics");
        assert_eq!(normalize_http_path("/"), "/");
    }
}
