use std::time::{Duration, Instant};

use lazy_static::lazy_static;
use prometheus::{HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, exponential_buckets};
use tici_common::metrics::{
    register_histogram_with_labels_and_buckets, register_int_counter_with_labels,
    register_int_gauge_with_labels,
};

const RESULT_OK: &str = "ok";
const RESULT_ERROR: &str = "error";

const FLUSH_REASON_MAX_WEIGHT: &str = "max_weight";
const FLUSH_REASON_NEXT_OVERFLOW: &str = "next_overflow";
const FLUSH_REASON_MAX_LATENCY: &str = "max_latency";
const FLUSH_REASON_DRAIN: &str = "drain";
const FLUSH_REASON_OVERSIZED_HEAD: &str = "oversized_head";

fn duration_buckets() -> Vec<f64> {
    exponential_buckets(0.001, 2.0, 17)
        .unwrap_or_else(|err| panic!("batch queue duration buckets failed: {err}"))
}

fn batch_size_buckets() -> Vec<f64> {
    exponential_buckets(1.0, 2.0, 11)
        .unwrap_or_else(|err| panic!("batch queue size buckets failed: {err}"))
}

fn batch_weight_buckets() -> Vec<f64> {
    exponential_buckets(1.0, 2.0, 30)
        .unwrap_or_else(|err| panic!("batch queue weight buckets failed: {err}"))
}

fn saturating_i64(value: u64) -> i64 {
    value.min(i64::MAX as u64) as i64
}

lazy_static! {
    pub static ref BATCH_QUEUE_SUBMIT_COUNTER: IntCounterVec = register_int_counter_with_labels(
        "tici_batch_queue_submit_count",
        "The count of batch queue submit attempts",
        &["component", "result"],
    );
    pub static ref BATCH_QUEUE_ENQUEUE_BLOCK_DURATION_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_enqueue_block_duration",
            "Time spent waiting for a batch queue slot before a request is enqueued",
            &["component"],
            duration_buckets(),
        );
    pub static ref BATCH_QUEUE_REQUEST_BACKLOG_GAUGE: IntGaugeVec = register_int_gauge_with_labels(
        "tici_batch_queue_request_backlog",
        "Number of requests enqueued but not yet executing in the batch queue",
        &["component"],
    );
    pub static ref BATCH_QUEUE_WEIGHT_BACKLOG_GAUGE: IntGaugeVec = register_int_gauge_with_labels(
        "tici_batch_queue_weight_backlog",
        "Weighted backlog of requests enqueued but not yet executing in the batch queue",
        &["component"],
    );
    pub static ref BATCH_QUEUE_QUEUE_DURATION_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_queue_duration",
            "Queued time between successful enqueue and request execution",
            &["component"],
            duration_buckets(),
        );
    pub static ref BATCH_QUEUE_REQUEST_DURATION_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_request_duration",
            "End-to-end time from submit to response delivery for batch queue requests",
            &["component"],
            duration_buckets(),
        );
    pub static ref BATCH_QUEUE_REQUEST_COUNTER: IntCounterVec = register_int_counter_with_labels(
        "tici_batch_queue_request_count",
        "The count of batch queue request completions",
        &["component", "result"],
    );
    pub static ref BATCH_QUEUE_BATCH_INFLIGHT_GAUGE: IntGaugeVec = register_int_gauge_with_labels(
        "tici_batch_queue_batch_inflight",
        "Number of batch executions currently running",
        &["component"],
    );
    pub static ref BATCH_QUEUE_REQUEST_INFLIGHT_GAUGE: IntGaugeVec = register_int_gauge_with_labels(
        "tici_batch_queue_request_inflight",
        "Number of batch queue requests currently running",
        &["component"],
    );
    pub static ref BATCH_QUEUE_SUBMIT_INFLIGHT_GAUGE: IntGaugeVec = register_int_gauge_with_labels(
        "tici_batch_queue_submit_inflight",
        "Number of batch queue submit calls currently in progress",
        &["component"],
    );
    pub static ref BATCH_QUEUE_BATCH_COLLECT_DURATION_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_batch_collect_duration",
            "Time spent collecting a batch after the first request is pulled",
            &["component"],
            duration_buckets(),
        );
    pub static ref BATCH_QUEUE_BATCH_EXECUTE_DURATION_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_batch_execute_duration",
            "Execution time of batch queue batches",
            &["component"],
            duration_buckets(),
        );
    pub static ref BATCH_QUEUE_BATCH_SIZE_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_batch_size",
            "Batch size in requests",
            &["component"],
            batch_size_buckets(),
        );
    pub static ref BATCH_QUEUE_BATCH_WEIGHT_HISTOGRAM: HistogramVec =
        register_histogram_with_labels_and_buckets(
            "tici_batch_queue_batch_weight",
            "Batch weight in component-defined units",
            &["component"],
            batch_weight_buckets(),
        );
    pub static ref BATCH_QUEUE_BATCH_FLUSH_COUNTER: IntCounterVec =
        register_int_counter_with_labels(
            "tici_batch_queue_batch_flush_count",
            "The count of batch queue flushes by trigger reason",
            &["component", "reason"],
        );
    pub static ref BATCH_QUEUE_MERGED_REQUEST_COUNTER: IntCounterVec =
        register_int_counter_with_labels(
            "tici_batch_queue_merged_request_count",
            "The number of requests merged into existing batches",
            &["component"],
        );
}

#[derive(Clone, Copy)]
pub struct BatchQueueMetrics {
    component: &'static str,
}

impl BatchQueueMetrics {
    pub const fn new(component: &'static str) -> Self {
        Self { component }
    }

    pub fn component(&self) -> &'static str {
        self.component
    }

    pub fn record_submit_result(&self, is_error: bool) {
        let result = if is_error { RESULT_ERROR } else { RESULT_OK };
        BATCH_QUEUE_SUBMIT_COUNTER
            .with_label_values(&[self.component, result])
            .inc();
    }

    pub fn begin_submit(&self) -> BatchQueueSubmitGuard {
        let submit_inflight =
            BATCH_QUEUE_SUBMIT_INFLIGHT_GAUGE.with_label_values(&[self.component]);
        submit_inflight.inc();
        BatchQueueSubmitGuard { submit_inflight }
    }

    pub fn observe_enqueue_block_duration(&self, duration: Duration) {
        BATCH_QUEUE_ENQUEUE_BLOCK_DURATION_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(duration.as_secs_f64());
    }

    pub fn add_backlog(&self, request_count: usize, total_weight: u64) {
        BATCH_QUEUE_REQUEST_BACKLOG_GAUGE
            .with_label_values(&[self.component])
            .add(saturating_i64(request_count as u64));
        BATCH_QUEUE_WEIGHT_BACKLOG_GAUGE
            .with_label_values(&[self.component])
            .add(saturating_i64(total_weight));
    }

    pub fn sub_backlog(&self, request_count: usize, total_weight: u64) {
        BATCH_QUEUE_REQUEST_BACKLOG_GAUGE
            .with_label_values(&[self.component])
            .sub(saturating_i64(request_count as u64));
        BATCH_QUEUE_WEIGHT_BACKLOG_GAUGE
            .with_label_values(&[self.component])
            .sub(saturating_i64(total_weight));
    }

    pub fn observe_request_queue_duration(&self, duration: Duration) {
        BATCH_QUEUE_QUEUE_DURATION_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(duration.as_secs_f64());
    }

    pub fn observe_request_completion(&self, duration: Duration, is_error: bool) {
        BATCH_QUEUE_REQUEST_DURATION_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(duration.as_secs_f64());
        let result = if is_error { RESULT_ERROR } else { RESULT_OK };
        BATCH_QUEUE_REQUEST_COUNTER
            .with_label_values(&[self.component, result])
            .inc();
    }

    pub fn observe_batch_collect_duration(&self, duration: Duration) {
        BATCH_QUEUE_BATCH_COLLECT_DURATION_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(duration.as_secs_f64());
    }

    pub fn observe_batch_size(&self, batch_size: usize) {
        BATCH_QUEUE_BATCH_SIZE_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(batch_size as f64);
    }

    pub fn observe_batch_weight(&self, batch_weight: u64) {
        BATCH_QUEUE_BATCH_WEIGHT_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(batch_weight as f64);
    }

    pub fn record_batch_flush(&self, reason: &'static str) {
        BATCH_QUEUE_BATCH_FLUSH_COUNTER
            .with_label_values(&[self.component, reason])
            .inc();
    }

    pub fn record_merged_requests(&self, merged_request_count: usize) {
        if merged_request_count == 0 {
            return;
        }
        BATCH_QUEUE_MERGED_REQUEST_COUNTER
            .with_label_values(&[self.component])
            .inc_by(merged_request_count as u64);
    }

    pub fn begin_batch_execution(&self, request_count: usize) -> BatchQueueExecutionGuard {
        let batch_inflight = BATCH_QUEUE_BATCH_INFLIGHT_GAUGE.with_label_values(&[self.component]);
        let request_inflight =
            BATCH_QUEUE_REQUEST_INFLIGHT_GAUGE.with_label_values(&[self.component]);
        batch_inflight.inc();
        request_inflight.add(saturating_i64(request_count as u64));
        BatchQueueExecutionGuard {
            component: self.component,
            batch_inflight,
            request_inflight,
            request_count: request_count as u64,
            started_at: Instant::now(),
        }
    }
}

pub struct BatchQueueSubmitGuard {
    submit_inflight: IntGauge,
}

impl Drop for BatchQueueSubmitGuard {
    fn drop(&mut self) {
        self.submit_inflight.dec();
    }
}

pub struct BatchQueueExecutionGuard {
    component: &'static str,
    batch_inflight: IntGauge,
    request_inflight: IntGauge,
    request_count: u64,
    started_at: Instant,
}

impl Drop for BatchQueueExecutionGuard {
    fn drop(&mut self) {
        self.batch_inflight.dec();
        self.request_inflight
            .sub(saturating_i64(self.request_count));
        BATCH_QUEUE_BATCH_EXECUTE_DURATION_HISTOGRAM
            .with_label_values(&[self.component])
            .observe(self.started_at.elapsed().as_secs_f64());
    }
}

pub const BATCH_QUEUE_FLUSH_REASON_MAX_WEIGHT: &str = FLUSH_REASON_MAX_WEIGHT;
pub const BATCH_QUEUE_FLUSH_REASON_NEXT_OVERFLOW: &str = FLUSH_REASON_NEXT_OVERFLOW;
pub const BATCH_QUEUE_FLUSH_REASON_MAX_LATENCY: &str = FLUSH_REASON_MAX_LATENCY;
pub const BATCH_QUEUE_FLUSH_REASON_DRAIN: &str = FLUSH_REASON_DRAIN;
pub const BATCH_QUEUE_FLUSH_REASON_OVERSIZED_HEAD: &str = FLUSH_REASON_OVERSIZED_HEAD;