use futures::{StreamExt, TryStreamExt};
use k8s_openapi::api::core::v1::Service;
use kube::{
    Client,
    api::{Api, ListParams, ResourceExt, WatchEvent, WatchParams},
};
use std::collections::HashMap;
use std::sync::Arc;

const LOCAL_DOMAINS_LABEL: &'static str = "dns.tensor.ru/local";
const LOCAL_DOMAINS_ANNOTATION: &'static str = "dns.tensor.ru/local-domains";
const HOSTS_DIR: &'static str = "/data";

use tokio::fs::OpenOptions;
use tokio::io::AsyncWriteExt;
use tokio::sync::Mutex;

struct Informer {
    domains: Mutex<HashMap<String, Vec<String>>>,
    client: Arc<Client>,
}

impl Informer {
    async fn add_service(&mut self, service: Service) {
        let annotations = service.annotations();
        if let Some(local_domains) = annotations.get(LOCAL_DOMAINS_ANNOTATION) {
            let domains_list: Vec<String> = local_domains
                .split_whitespace()
                .map(|a| a.to_string())
                .collect();
            if let Some(spec) = service.spec {
                if let Some(cluster_ip) = spec.cluster_ip
                    && domains_list.len() > 0
                {
                    println!("Service added {} -> {:?}", cluster_ip, domains_list);
                    let mut lock = self.domains.lock().await;
                    lock.insert(cluster_ip, domains_list);
                }
            }
        }
    }

    async fn delete_service(&mut self, service: Service) {
        if let Some(spec) = service.spec {
            if let Some(cluster_ip) = spec.cluster_ip {
                let mut lock = self.domains.lock().await;
                match lock.remove(&cluster_ip) {
                    Some(domains) => {
                        println!("Service deleted {} -> {:?}", cluster_ip, domains);
                    }
                    None => {}
                }
            }
        }
    }

    pub async fn fill(&mut self) {
        let client: Client = (*self.client).clone();
        let services: Api<Service> = Api::all(client);
        let lp = ListParams {
            label_selector: Some(LOCAL_DOMAINS_LABEL.to_string()),
            ..Default::default()
        };
        while let Ok(services_list) = services.list(&lp).await {
            for service in services_list {
                self.add_service(service).await;
            }
        }
        let _ = self.write_hosts_file().await;
    }

    pub async fn get_cache(&self) -> HashMap<String, Vec<String>> {
        let lock = self.domains.lock().await;
        lock.clone()
    }

    pub async fn watch(mut self) -> Result<(), Box<dyn std::error::Error>> {
        let client: Client = (*self.client).clone();
        let services: Api<Service> = Api::all(client);
        let wp = WatchParams {
            label_selector: Some("dns.tensor.ru/local".to_string()),
            ..Default::default()
        };
        let mut stream = services.watch(&wp, "0").await?.boxed();

        while let Some(status) = stream.try_next().await? {
            match status {
                WatchEvent::Added(s) | WatchEvent::Modified(s) => {
                    self.add_service(s).await;
                    self.write_hosts_file().await?;
                }
                WatchEvent::Deleted(s) => {
                    self.delete_service(s).await;
                    self.write_hosts_file().await?;
                }
                WatchEvent::Bookmark(s) => {}
                WatchEvent::Error(s) => println!("{}", s),
            }
        }

        Ok(())
    }

    async fn write_hosts_file(&self) -> Result<(), Box<dyn std::error::Error>> {
        let mut file = OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(format!("{}/hosts", HOSTS_DIR))
            .await?;

        // Собираем все строки в один буфер
        let mut content = String::new();
        let lock = self.domains.lock().await;
        for (k, v) in lock.iter() {
            content.push_str(&format!("{} {}\n", k, v.join(" ")));
        }
        file.write_all(content.as_bytes()).await?;
        file.flush().await?;
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut informer = Informer {
        domains: Mutex::new(HashMap::new()),
        client: Arc::new(Client::try_default().await?),
    };
    informer.fill().await;
    println!("{:?}", informer.get_cache().await);

    let handle = tokio::spawn(async move {
        if let Err(e) = informer.watch().await {
            eprintln!("Watcher error: {}", e);
        }
    });

    // Ждем выполнения фоновой задачи (в этот момент main не завершится)
    handle.await?;

    Ok(())
}