integration-module/crates/api-grub/src/jitter.rs

use std::collections::{HashMap, HashSet};
use integr_structs::api::v3::{PrometheusMetricsExtended, MetricOutputExtended};
use std::sync::Arc;
use reqwest::Client;
use lazy_static::lazy_static;
use tracing::{error, info, trace, warn};
use std::pin::Pin;
use serde_json::{Value, from_str};
use crate::export::Exporter;
use async_stream::stream;
use futures::{future, pin_mut, stream::Stream, StreamExt};

lazy_static! {
    static ref VINTEO_BASE: String = std::env::var("VINTEO_URL_BASE").unwrap_or_else(|_| String::from("https://demo.vcs.vinteo.dev"));
    static ref CONFERENCES_ENDPOINT: String = std::env::var("VINTEO_ENDPOINT_CONFERENCES").unwrap_or_else(|_| String::from("/api/v1/conferences"));
    static ref USERS_ENDPOINT: String = std::env::var("VINTEO_ENDPOINT_PARTICIPANTS").unwrap_or_else(|_| String::from("/api/v1/participants/"));
}

// conferences ids 
type Conferences = HashSet<(String, String)>;
type OutputConferences = HashMap<(String, String), Value>;

#[derive(Debug, Clone)]
struct ConferencesDigital(OutputConferences);

impl ConferencesDigital {
    pub fn new(item: OutputConferences) -> Self {
        Self(item)
    }
    pub fn go_across(self) -> impl Stream<Item = ((String, String), Arc<Value>)> {
        stream! {
            for (key, value) in self.0 {
                info!("Conference {} ({}) is processing ...", key.0, key.1);
                yield (key, Arc::new(value));
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        }
    }
}


// to handle async http requests
#[derive(Debug)] 
struct Requester {
    base: String, 
    api_key: String,
    client: Client,
}

impl Requester {
    pub fn new() -> anyhow::Result<Self>  {
        Ok(Self {
            base: VINTEO_BASE.clone().to_owned(), 
            api_key: std::env::var("VINTEO_API_KEY")?,
            client: Client::builder().user_agent("api-grub").build()?,
        })
    }
    pub async fn get_conferences(&self) -> anyhow::Result<Value> {
        trace!("getting conferences list from API");
        let url = format!("{}{}", self.base, *CONFERENCES_ENDPOINT);
        let req = self.client.get(url)
        .timeout(tokio::time::Duration::from_secs(10))
        .header("accept", "application/json")
        .header("x-api-key", &self.api_key);
        let resp = req.send().await;
        Ok(from_str(&resp?.text().await?)?)
    }
    pub async fn get_conferences_from_value(conferences: Value) -> anyhow::Result<Conferences> {
        trace!("extracting conferences list");
        let mut hashset = Conferences::new();
        match conferences.get("data") {
            None => return Err(anyhow::Error::msg("Invalid JSON format: no `data` field")),
            Some(data) => {
                match data.get("conferences") {
                    None => return Err(anyhow::Error::msg("Invalid JSON format: no `conferences` field")),
                    Some(confs) => {
                        if let Some(confs) = confs.as_array() {
                            confs.iter()
                            .for_each(|conf_obj| {
                                if let Some(id) = conf_obj.get("number") {
                                    if let Some(id) = id.as_str() {
                                        let id = {
                                            if let Some(desc) = conf_obj.get("description") {
                                                match desc.as_str() {
                                                    Some(desc) => (desc.to_string(), id.to_string()),
                                                    None => (String::new(), id.to_string())
                                                }
                                            } else {
                                                (String::new(), id.to_string())
                                            }
                                        };
                                        hashset.insert(id);
                                    }
                                }
                            });
                        } else {
                            return Err(anyhow::Error::msg("Invalid JSON format: `conferences` is not an array"))
                        }
                    },
                }
            },
        }
        Ok(hashset)
    }
    pub async fn get_users_jitter_by_conference(&self, conferences : Conferences) -> anyhow::Result<OutputConferences> {
        trace!("getting users list for each conference from API ...");
        let mut output = OutputConferences::new();
        for conf in conferences {
            let url = format!("{}{}{}", self.base, *USERS_ENDPOINT, &conf.1);
            let req = self.client.get(url)
            .timeout(tokio::time::Duration::from_secs(10))
            .header("accept", "application/json")
            .header("x-api-key", &self.api_key);
            
            match req.send().await {
                Err(er) => error!("Cannot GET data about conference - {:?}: {}", conf, er),
                Ok(resp) => {
                    match resp.text().await {
                        Err(er) => error!("Cannot convert response into text: {}", er),
                        Ok(resp) => {
                            let resp: Value = from_str(&resp)?;
                            output.entry(conf).or_insert(resp);
                        } 
                    }
                },
            }
        }
        Ok(output)
    }
}


    fn across_participants(conf: &Value) -> impl Stream<Item = Arc<Value>> + '_ {
        trace!("fetching participants for conf and going across ...");
        let participants_iter = conf
        .get("data")
        .and_then(|data| data.get("participants"))
        .and_then(|participants| participants.as_array())
        .into_iter()
        .flatten()
        .cloned();

        stream! {
            for participant in participants_iter {
                yield Arc::new(participant);
            }
        }
    }

    async fn gather_futures(parts: Arc<Value>, conf_id: Arc<str>, conf_desc: Arc<str>, targets: Vec<(&str, &str)>) -> Vec<Option<MetricOutputExtended>> {
        trace!("extracting {} measures for current participant in current conference {} ...", targets.len(), conf_id);
        let mut futures: Vec<Pin<Box<dyn futures::Future<Output = Option<MetricOutputExtended>> + Send>>> = Vec::new();
        for (target, description) in targets {
            futures.push(Box::pin(extract_from_participant(
                parts.clone(), 
                target, 
                conf_id.clone(), 
                conf_desc.clone(),
                description
            )));
        }
        future::join_all(futures).await
    }

// GET participants/{conference_id} data.total
async fn extract_total_participants(conferences : Arc<Value>) -> Value {
    if let Some(total) = conferences
    .get("data")
    .and_then(|data| data.get("total")) {
        return total.clone()
    }
    Value::Null
}

async fn extract_total_anon_participants(conferences : Arc<Value>) -> Value {
    let sum = conferences
    .get("data")
    .and_then(|data| data.get("participants"))
    .and_then(|parts| parts.as_array())
    .iter()
    .flat_map(|&val| val.iter())
    .filter_map(|part| part.get("isAnonymous"))
    .filter_map(|is_anon| is_anon.as_bool())
    .filter(|is_anon| *is_anon == true)
    .count();
    Value::Number(sum.into())
}

// GET participants/{conference_id} data.participants[].isAnonymous
async fn extract_from_participant(participant : Arc<Value>, target: &str, conf_id: Arc<str>, conf_desc: Arc<str>, description: &str) -> Option<MetricOutputExtended> {
    trace!("extracting `{}` measure for current participant ...", target);
    if let Some(value) = participant.get("params")
    .and_then(|params| params.get(target)) {
        let name = participant.get("watermark").unwrap_or_else(|| &Value::Null).as_str().unwrap_or_else(|| "unknown");
        let id = participant.get("id").unwrap_or_else(|| &Value::Null).as_str().unwrap_or_else(|| "unknown");
        let metric_name = format!("{}_{}_{}", target, conf_id, id);
        return Some(
            MetricOutputExtended::new_with_slices(
                &metric_name, 
                &metric_name, 
                "type", 
                "Vinteo native", 
                format!(
                    "{} для пользователя {} ({}) в конференции {} ({})",
                    description, 
                    name,
                    id,
                    conf_desc,
                    conf_id
                ).as_ref(), 
                Some(18), 
                Some("module$12".to_string()), 
                value.clone()
            )
        )
    }
    None
}

pub async fn init_grubbing_jitter() -> anyhow::Result<()> {
    // create requester 
    let requester = Requester::new()?;
    // get conference id's
    loop {
        info!("Getting conferences array ...");
        let conferences =  requester.get_conferences().await?;
        let conferences = Requester::get_conferences_from_value(conferences).await?;
        
        info!("Getting jitter metrics by user in each conference ...");
        let confs = requester.get_users_jitter_by_conference(conferences).await?;
        
        info!("Initializing extraction mechanism ...");
        let conferences = ConferencesDigital::new(confs);
        let extracted_conference = conferences.go_across();
        let mut buffer: Vec<MetricOutputExtended> = Vec::new();
        pin_mut!(extracted_conference);

        while let Some(((desc, id), item)) = extracted_conference.next().await {
            let parts_stream = across_participants(&item); 
            pin_mut!(parts_stream);

            let total_participants = MetricOutputExtended::new_with_slices(
                format!("TotalParticipants_{}", id).as_ref(), 
                format!("TotalParticipants_{}", id).as_ref(), 
                "type", 
                "Vinteo Native", 
                format!("Общее количество участников в конференции {}", &desc).as_ref(), 
                Some(18), 
                Some(String::from("module$12")), 
                extract_total_participants(item.clone()).await
            );
            // anon NON_STABLE
            let total_anon_participants = MetricOutputExtended::new_with_slices(
                format!("TotalAnonymousParticipants_{}", id).as_ref(), 
                format!("TotalAnonymousParticipants_{}", id).as_ref(), 
                "type", 
                "Vinteo Native", 
                format!("Общее количество анонимных участников в конференции {}", &desc).as_ref(), 
                Some(18), 
                Some(String::from("module$12")), 
                extract_total_anon_participants(item.clone()).await
            );
            // description

            buffer.push(total_participants);
            buffer.push(total_anon_participants);

            while let Some(participant) = parts_stream.next().await {
                buffer.append(
                    &mut gather_futures(
                    participant, 
                    Arc::from(id.as_ref()), 
                    Arc::from(desc.as_ref()),
                    vec![
                        ("txBitrate", "Скорость отправки данных"),
                        ("rxBitrate", "Скорость приёма данных"),
                        ("txFPS", "Количество отправляемых кадров в секунду"),
                        ("rxFPS", "Количество получаемых кадров в секунду"),
                        ("rxLost", "Количество потерянных пакетов на приёме"),
                        ("txLost", "Количество потерянных пакетов на отправке"),
                        ("jBLen", "Фазовое отклонение на цифровом канале"),
                        ("txLostPercent", "Процент потерянных пакетов на отправке"),
                        ("rxLostPercent", "Процент потерянных пакетов на приёме"),
                        ("txH239Bitrate", "Скорость передачи данных в дополнительном видео-потоке"),
                        ("rxH239Bitrate", "Скорость приёма данных в дополнительном видео-потоке"),
                        ("txVideoCodec", "Используемый видеокодек на отправке"),
                        ("rxVideoCodec", "Используемый видеокодек на приёме"),
                        ("txAudioCodec", "Используемый аудиокодек на отправке"),
                        ("rxAudioCodec", "Используемый аудиокодек на приёме"),
                        ("txResolution", "Разрешение передаваемого видео"),
                        ("rxResolution", "Разрешение принимаемого видео"),
                        ("txH239Resolution", "Разрешение дополнительного видео-потока"),
                        ("rxH239Resolution", "Разрешение дополнительного видео-потока на принимающей стороне"),
                        ("duration", "Длительсность сеанса (в минутах)"),
                    ],
                ).await
                .into_iter()
                .filter(|metric| metric.is_some())
                .map(|metric| metric.unwrap() )
                .collect::<Vec<MetricOutputExtended>>()
                );
            }
        }

        // let metrics = Requester::get_metrics_from_jitter(jitter).await;

        if !buffer.is_empty() {
            let metrics = PrometheusMetricsExtended::new_zvks(buffer).await;
            match Exporter::export_extended_metrics(metrics).await {
                Ok(bytes) => info!("Successfully transmitted metrics ({} bytes)", bytes),
                Err(er) => error!("Cannot export data: {}", er),
            }
        } else {
            warn!("Metrics array is empty. Ignoring exporting ...");
        }
        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
    }
    // get users' jitter info   
}