use std::collections::VecDeque;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Context;
use confique::Config;
use tokio::sync::broadcast::{
    self,
    Sender,
};
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

use crate::reporting::k8s_event::K8sEvent;
use crate::reporting::otel::OtelObservation;
use crate::reporting::prometheus::ProLabel;
use crate::reporting::tracing_evt::{
    TracingK8sEventLayer,
    TracingK8sEventLayerConfig,
};

pub mod k8s_event;
pub mod otel;
pub mod prometheus;
pub mod tracing_evt;

#[derive(Config)]
pub struct ReportingConfig {
    #[config(env = "KUYASEG_REPORT_K8S")]
    #[config(default = true)]
    pub k8s_events: bool,
    #[config(env = "KUYASEG_REPORT_PROMETHEUS")]
    #[config(default = false)]
    pub prometheus: bool,
    #[config(env = "KUYASEG_REPORT_OTEL")]
    #[config(default = false)]
    pub opentelemetry: bool,
}

pub fn trace_id_str(trace_id: &opentelemetry::TraceId) -> Option<String> {
    match trace_id {
        &opentelemetry::TraceId::INVALID => None,
        x => Some(x.to_string()),
    }
}

/// You should await on handle somewhere
pub struct ReportingInit {
    pub tracing_layer: TracingK8sEventLayer,
    pub handle: JoinSet<()>,
}

pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result<ReportingInit> {
    let config = ReportingConfig::builder()
        .env()
        .load()
        .context("Load Reporting Configuration")?;

    let mut reporting_jobs = JoinSet::new();

    let k8s_tx = config
        .k8s_events
        .then(|| init_k8s_events(k8s_client, &mut reporting_jobs, ct.clone()));
    let pro_tx = config
        .prometheus
        .then(|| init_pro_metrics(&mut reporting_jobs, ct.clone()));
    let otel_tx = config
        .opentelemetry
        .then(|| init_otel(&mut reporting_jobs, ct.clone()))
        .transpose()?;

    Ok(ReportingInit {
        tracing_layer: TracingK8sEventLayer::new(TracingK8sEventLayerConfig {
            k8s_events: k8s_tx,
            prometheus: pro_tx,
            otel: otel_tx,
        }),
        handle: reporting_jobs,
    })
}

fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender<Arc<K8sEvent>> {
    use kube::runtime::events::{
        Recorder,
        Reporter,
    };

    use crate::reporting::k8s_event::K8sEventRecorder;

    const WARN_THRESHOLD: usize = 10;
    const MAX_CAPACITY: usize = 100;

    let (tx, mut rx) = broadcast::channel::<Arc<K8sEvent>>(512);
    let reporter = Reporter {
        controller: env!("CARGO_PKG_NAME").into(),
        instance: std::env::var("CONTROLLER_POD_NAME").ok(),
    };
    let recorder = Recorder::new(client, reporter);
    let recorder = K8sEventRecorder::new(recorder);

    let rtx = tx.clone(); // for dumb recovery
    joinset.spawn(async move {
        let mut queue: VecDeque<Arc<K8sEvent>> = VecDeque::new();
        let mut warned = false;
        let mut drain_tick = tokio::time::interval(Duration::from_secs(6)); // 10 event/min
        let mut rec_limit = 4;

        while rec_limit > 0 {
            tokio::select! {
                biased;
                _ = ct.cancelled() => break,
                _ = drain_tick.tick() => {
                    if let Some(ev) = queue.pop_front() {
                        recorder.record(&ev).await;
                    }
                },
                msg = rx.recv() => {
                    match msg {
                        Ok(ev) => {
                            if queue.is_empty() {
                                recorder.record(&ev).await;
                                warned = false;
                            } else {
                                if queue.len() >= MAX_CAPACITY {
                                    queue.pop_front();
                                }
                                queue.push_back(ev);
                                if queue.len() > WARN_THRESHOLD && !warned {
                                    tracing::warn!(k8s_no_event=true, queue_len=queue.len(), "K8s event queue exceeded threshold");
                                    warned = true;
                                } else if queue.len() <= WARN_THRESHOLD {
                                    warned = false;
                                }
                            }
                            rec_limit = 4;
                        }
                        Err(e) => {
                            tracing::error!(k8s_no_event=true, debug=?e, "Failed to receive k8s event from tracing");
                            rx = rtx.subscribe(); // dumb recover strategy
                            tokio::time::sleep(Duration::from_secs(3)).await;
                            rec_limit -= 1;
                        }
                    }
                }
            }
        }
        tracing::error!(
            no_event = true,
            "K8s Event reporting from tracing stopped"
        );
    });

    tx
}

fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender<Arc<ProLabel>> {
    use prometheus_client::registry::Registry;

    use crate::reporting::prometheus::ProMetrics;

    let (tx, mut rx) = broadcast::channel::<Arc<ProLabel>>(512);
    let metrics = ProMetrics::new();

    let registration_cancellation = CancellationToken::new();

    let rtx = tx.clone();
    let ctk = registration_cancellation.clone();
    let promt = metrics.clone();
    let consumer_ct = ct.clone();
    joinset.spawn(async move {
        let mut rec_limit = 4;
        while rec_limit > 0 {
            tokio::select! {
                biased;
                _ = consumer_ct.cancelled() => break,
                msg = rx.recv() => {
                    match msg {
                        Ok(met) => {
                            promt.measure(&met);
                            rec_limit = 4;
                        }
                        Err(e) => {
                            tracing::error!(k8s_no_pro=true, debug=?e, "Failed to receive prometheus measurement from tracing");
                            rx = rtx.subscribe(); // dumb recover strategy;
                            tokio::time::sleep(Duration::from_secs(3)).await;
                            rec_limit -= 1;
                        }
                    }
                }
            }
        }
        ctk.cancel();
        tracing::error!(
            no_pro = true,
            "Prometheus measurement from tracing stopped"
        );
    });

    let ctk = ct.clone();
    joinset.spawn(async move {
        let mut registry = Registry::default();
        loop {
            tokio::select! {
                _ = tokio::time::sleep(Duration::from_secs(10)) => {
                    metrics.register(&mut registry);
                },
                _ = registration_cancellation.cancelled() => {
                    tracing::warn!(k8s_no_pro=true, "Stopped publishing prometheus metrics: consumer job cancelled");
                    break;
                }
                _ = ctk.cancelled() => {
                    tracing::warn!(k8s_no_pro=true, "Stopped publishing prometheus metrics: shutdown");
                    break;
                }
            }
        }
    });

    tx
}

fn init_otel(joinset: &mut JoinSet<()>, ct: CancellationToken) -> anyhow::Result<Sender<Arc<OtelObservation>>> {
    use crate::reporting::otel::OtelAggregator;

    let exporter = opentelemetry_otlp::LogExporter::builder()
        .with_tonic()
        .build()
        .context("Failed to build OTLP log exporter")?;

    let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
        .with_batch_exporter(exporter)
        .build();

    let aggregator = Arc::new(OtelAggregator::new(provider));
    let task_aggregator = aggregator.clone();

    let (tx, mut rx) = broadcast::channel::<Arc<OtelObservation>>(512);

    let rtx = tx.clone();
    joinset.spawn(async move {
        let mut rec_limit = 4;
        while rec_limit > 0 {
            tokio::select! {
                biased;
                _ = ct.cancelled() => break,
                msg = rx.recv() => {
                    match msg {
                        Ok(obs) => {
                            task_aggregator.observe(&obs);
                            rec_limit = 4;
                        }
                        Err(e) => {
                            tracing::error!(
                                k8s_no_otel = true,
                                debug = ?e,
                                "Failed to receive otel observation from tracing"
                            );
                            rx = rtx.subscribe();
                            tokio::time::sleep(Duration::from_secs(3)).await;
                            rec_limit -= 1;
                        }
                    }
                }
            }
        }
        tracing::error!(k8s_no_otel = true, "OpenTelemetry reporting from tracing stopped");
    });

    Ok(tx)
}