videosdk-server-sdk 0.1.0

//! The alert-rules API: metric-threshold alerts over your project's telemetry.

use std::collections::HashMap;
use std::sync::Arc;

use futures_util::Stream;
use reqwest::Method;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

use crate::client::{CallOptions, Client};
use crate::common::{string_enum, MessageResponse};
use crate::error::Result;
use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
use crate::query::QueryBuilder;
use crate::resources::escape;

const PATH: &str = "/v2/alertRules";

string_enum! {
    /// An alert rule's severity.
    AlertSeverity {
        /// Informational.
        INFO => "info",
        /// A warning.
        WARNING => "warning",
        /// Critical.
        CRITICAL => "critical",
    }
}

string_enum! {
    /// The product domain an alert rule is scoped to.
    ///
    /// The server defaults to [`AlertDomain::RTC`] when omitted — pass
    /// [`AlertDomain::AGENT`] to see agent-domain rules.
    AlertDomain {
        /// Real-time communication.
        RTC => "rtc",
        /// AI agents.
        AGENT => "agent",
    }
}

string_enum! {
    /// Whether an alert rule is active or paused.
    AlertRuleStatus {
        /// The rule is evaluating.
        ACTIVE => "active",
        /// The rule is paused.
        PAUSED => "paused",
    }
}

string_enum! {
    /// The comparison operator in an alert rule condition.
    AlertConditionOperator {
        /// Greater than.
        GREATER_THAN => ">",
        /// Less than.
        LESS_THAN => "<",
        /// Equal to.
        EQUAL => "=",
        /// Not equal to.
        NOT_EQUAL => "!=",
    }
}

string_enum! {
    /// How a condition must be met over the evaluation window.
    AlertMatchType {
        /// Fire if the threshold is crossed at least once. The default.
        AT_LEAST_ONCE => "at_least_once",
        /// Fire only if crossed for the whole window.
        ALL_THE_TIME => "all_the_time",
        /// Fire on the window average.
        ON_AVERAGE => "on_average",
        /// Fire on the window total.
        IN_TOTAL => "in_total",
        /// Fire on the last value.
        LAST => "last",
    }
}

string_enum! {
    /// Collapses each series to a single value before the threshold check.
    AlertReduceTo {
        /// The last value.
        LAST => "last",
        /// The sum.
        SUM => "sum",
        /// The average.
        AVG => "avg",
        /// The minimum.
        MIN => "min",
        /// The maximum.
        MAX => "max",
    }
}

/* -------------------------------- config types -------------------------------- */

/// Alerts when a metric stops reporting data.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AbsentDataAlert {
    /// Whether absent-data alerting is on.
    pub enabled: bool,
    /// Fire after the metric has been absent this long.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub for_seconds: Option<u32>,
}

/// Narrows a metric query. `value` is a string, number, or array of them.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertRuleFilter {
    /// The attribute to filter on.
    pub category: String,
    /// The filter operator, e.g. `=`, `in`, `regex`, `exists`.
    pub condition: String,
    /// The value(s) to compare against.
    pub value: Value,
}

/// Groups a metric query by an attribute key.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRuleGroupBy {
    /// The attribute key.
    pub key: String,
    /// The attribute type: `string`, `int64`, `float64`, or `bool`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub data_type: Option<String>,
    /// The attribute source: `tag`, `resource`, or `scope`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub key_type: Option<String>,
}

/// Filters on an aggregated column.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRuleHaving {
    /// The aggregated column.
    pub column_name: String,
    /// One of `=`, `!=`, `>`, `>=`, `<`, `<=`.
    pub operator: String,
    /// The value to compare against.
    pub value: f64,
}

/// Selects and aggregates a metric to alert on.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRuleQuery {
    /// The metric name, e.g. `stat_consumer_video_bitrate`. Required.
    pub metric: String,
    /// The metric's category, e.g. `bitrate`, `packet_loss`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metric_category: Option<String>,
    /// Reduces a series over time: `latest`, `sum`, `avg`, `min`, `max`, `count`,
    /// `count_distinct`, `rate`, `increase`, or a percentile `p05`..`p99`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time_aggregation: Option<String>,
    /// Reduces across series: `sum`, `avg`, `min`, `max`, `count`, or `p50`..`p99`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub space_aggregation: Option<String>,
    /// The metric's value type: `float64`, `int64`, or `string`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub data_type: Option<String>,
    /// The query step interval, in seconds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub step_interval_seconds: Option<u32>,
    /// Filters narrowing the query.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub filter: Vec<AlertRuleFilter>,
    /// Attributes to group by.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub group_by: Vec<AlertRuleGroupBy>,
    /// Post-aggregation filters.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub having: Vec<AlertRuleHaving>,
}

impl AlertRuleQuery {
    /// A query over a single metric, with no aggregation or filters.
    pub fn metric(metric: impl Into<String>) -> Self {
        Self {
            metric: metric.into(),
            ..Default::default()
        }
    }
}

/// Compares the query's aggregated value against a threshold.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRuleCondition {
    /// The comparison operator. Required.
    pub operator: AlertConditionOperator,
    /// The threshold to compare against. Required.
    pub threshold: f64,
    /// Overrides the query's time aggregation for the check.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time_aggregation: Option<String>,
    /// Overrides the query's space aggregation for the check.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub space_aggregation: Option<String>,
    /// Require at least this many sessions before the rule can fire.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub minimum_session_gate: Option<f64>,
    /// The threshold's unit.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub unit: Option<String>,
    /// How the threshold must be met over the window. Defaults to
    /// [`AlertMatchType::AT_LEAST_ONCE`].
    #[serde(skip_serializing_if = "Option::is_none")]
    pub match_type: Option<AlertMatchType>,
    /// Require a minimum number of data points.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub require_min_points: Option<bool>,
    /// The minimum number of data points required.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub required_num_points: Option<u32>,
    /// How often the condition is evaluated, in seconds. Defaults to 60.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub frequency_seconds: Option<u32>,
}

impl AlertRuleCondition {
    /// A condition comparing the aggregated value against `threshold`.
    pub fn new(operator: AlertConditionOperator, threshold: f64) -> Self {
        Self {
            operator,
            threshold,
            time_aggregation: None,
            space_aggregation: None,
            minimum_session_gate: None,
            unit: None,
            match_type: None,
            require_min_points: None,
            required_num_points: None,
            frequency_seconds: None,
        }
    }
}

/// Controls how often and over what window the rule runs.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRuleEvaluation {
    /// How often the rule is evaluated, in seconds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub frequency_seconds: Option<u32>,
    /// The evaluation window, in seconds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub window_seconds: Option<u32>,
}

/// An alert rule's config: a metric `query` and a threshold `condition` (both
/// required), plus optional evaluation, absent-data alerting, and a formula over
/// `additional_queries`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRuleConfig {
    /// The metric to alert on. Required.
    pub query: AlertRuleQuery,
    /// The threshold check. Required.
    pub condition: AlertRuleCondition,
    /// Free-form labels attached to fired incidents.
    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
    pub labels: HashMap<String, String>,
    /// Evaluation frequency and window.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub evaluation: Option<AlertRuleEvaluation>,
    /// Alert when the metric stops reporting.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub absent_data_alert: Option<AbsentDataAlert>,
    /// Extra named queries (keys `B`..`Z`) referenced by `formula`.
    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
    pub additional_queries: HashMap<String, AlertRuleQuery>,
    /// A formula combining `query` and `additional_queries`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub formula: Option<String>,
    /// Collapses each series to a single value before the threshold check.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reduce_to: Option<AlertReduceTo>,
}

impl AlertRuleConfig {
    /// A config from a query and a condition, leaving everything else default.
    pub fn new(query: AlertRuleQuery, condition: AlertRuleCondition) -> Self {
        Self {
            query,
            condition,
            labels: HashMap::new(),
            evaluation: None,
            absent_data_alert: None,
            additional_queries: HashMap::new(),
            formula: None,
            reduce_to: None,
        }
    }
}

/* --------------------------------- response ---------------------------------- */

/// A configured alert rule.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertRule {
    /// The rule's database id.
    pub id: Option<String>,
    /// The rule id.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub alert_rule_id: String,
    /// The rule's display name.
    pub name: Option<String>,
    /// The rule's severity.
    pub severity: Option<AlertSeverity>,
    /// The rule's description.
    pub description: Option<String>,
    /// A short summary applied to fired incidents.
    pub summary: Option<String>,
    /// The rule's domain.
    pub domain: Option<AlertDomain>,
    /// The rule's config.
    pub rule_config: Option<AlertRuleConfig>,
    /// The notification channels this rule delivers to.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub notification_channel_ids: Vec<String>,
    /// Whether the rule is active or paused.
    pub status: Option<AlertRuleStatus>,
    /// The rule's version.
    pub version: Option<i64>,
    /// When the rule last fired.
    pub last_triggered: Option<String>,
    /// When the rule was created.
    pub created_at: Option<String>,
    /// When the rule was last updated.
    pub updated_at: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/* --------------------------------- params ------------------------------------ */

/// The parameters for [`AlertRulesResource::create`].
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateAlertRuleParams {
    /// The rule's display name. Required.
    pub name: String,
    /// The rule's severity. Required.
    pub severity: AlertSeverity,
    /// The rule's config. Required.
    pub rule_config: AlertRuleConfig,
    /// The rule's description.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// A summary applied to the underlying alert. Not stored on the rule record.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub summary: Option<String>,
    /// Whether the alert is active. Surfaces as `status`, not a returned
    /// `enabled` field.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub enabled: Option<bool>,
    /// The rule's domain.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub domain: Option<AlertDomain>,
    /// Used when building the alert query. Not persisted on the rule record.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time_aggregation: Option<String>,
    /// The notification channels to deliver to.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub notification_channel_ids: Vec<String>,
}

impl CreateAlertRuleParams {
    /// The required fields, leaving everything else unset.
    pub fn new(
        name: impl Into<String>,
        severity: AlertSeverity,
        rule_config: AlertRuleConfig,
    ) -> Self {
        Self {
            name: name.into(),
            severity,
            rule_config,
            description: None,
            summary: None,
            enabled: None,
            domain: None,
            time_aggregation: None,
            notification_channel_ids: Vec::new(),
        }
    }
}

/// The parameters for [`AlertRulesResource::update`]. Only the set fields change.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateAlertRuleParams {
    /// A new display name.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// A new severity.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub severity: Option<AlertSeverity>,
    /// A new config.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rule_config: Option<AlertRuleConfig>,
    /// A new description.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// A new summary.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub summary: Option<String>,
    /// Activate or pause the rule.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub enabled: Option<bool>,
    /// A new domain.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub domain: Option<AlertDomain>,
    /// A new query-time aggregation.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time_aggregation: Option<String>,
    /// New notification channels.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub notification_channel_ids: Vec<String>,
}

/// The query parameters for [`AlertRulesResource::list`].
#[derive(Debug, Clone, Default)]
pub struct ListAlertRulesParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page.
    pub cursor: Option<String>,
    /// Filters by region.
    pub region: Option<String>,
    /// Filters by status.
    pub status: Option<AlertRuleStatus>,
    /// Filters by metric category.
    pub metric_category: Option<String>,
    /// Filters by domain. The server defaults to [`AlertDomain::RTC`] when
    /// omitted — pass [`AlertDomain::AGENT`] to see agent-domain rules.
    pub domain: Option<AlertDomain>,
}

impl ListAlertRulesParams {
    fn pagination(&self) -> ListParams {
        ListParams {
            page: self.page,
            per_page: self.per_page,
            cursor: self.cursor.clone(),
        }
    }
}

/* -------------------------------- resource ----------------------------------- */

/// The alert-rules API. Reached via [`Client::alert_rules`].
#[derive(Debug, Clone, Copy)]
pub struct AlertRulesResource<'a> {
    client: &'a Client,
}

impl<'a> AlertRulesResource<'a> {
    pub(crate) fn new(client: &'a Client) -> Self {
        Self { client }
    }

    /// Creates an alert rule.
    pub async fn create(&self, params: CreateAlertRuleParams) -> Result<AlertRule> {
        self.client
            .wrapped(Method::POST, PATH, "alertRule", CallOptions::json(&params)?)
            .await
    }

    /// Lists alert rules, one page at a time.
    pub async fn list(&self, params: ListAlertRulesParams) -> Result<Page<AlertRule>> {
        paginate(self.fetcher(&params), &params.pagination(), "data", None).await
    }

    /// Lists alert rules, transparently fetching every page.
    pub fn list_stream(
        &self,
        params: ListAlertRulesParams,
    ) -> impl Stream<Item = Result<AlertRule>> + Send {
        auto_page(self.fetcher(&params), params.pagination(), "data", None)
    }

    /// Fetches an alert rule by id.
    pub async fn get(&self, alert_rule_id: &str) -> Result<AlertRule> {
        let path = format!("{PATH}/{}", escape(alert_rule_id));
        self.client
            .wrapped(Method::GET, &path, "alertRule", CallOptions::new())
            .await
    }

    /// Updates an alert rule.
    pub async fn update(
        &self,
        alert_rule_id: &str,
        params: UpdateAlertRuleParams,
    ) -> Result<AlertRule> {
        let path = format!("{PATH}/{}", escape(alert_rule_id));
        self.client
            .wrapped(Method::PUT, &path, "alertRule", CallOptions::json(&params)?)
            .await
    }

    /// Deletes an alert rule.
    pub async fn delete(&self, alert_rule_id: &str) -> Result<MessageResponse> {
        let path = format!("{PATH}/{}", escape(alert_rule_id));
        self.client
            .json(Method::DELETE, &path, CallOptions::new())
            .await
    }

    fn fetcher(&self, params: &ListAlertRulesParams) -> PageFetcher {
        let client = self.client.clone();
        let params = params.clone();
        Arc::new(move |page, per_page| {
            let client = client.clone();
            let params = params.clone();
            Box::pin(async move {
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str("region", params.region.as_deref())
                    .opt_str(
                        "status",
                        params.status.as_ref().map(AlertRuleStatus::as_str),
                    )
                    .opt_str("metricCategory", params.metric_category.as_deref())
                    .opt_str("domain", params.domain.as_ref().map(AlertDomain::as_str))
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn create_serializes_the_typed_config() {
        let params = CreateAlertRuleParams::new(
            "High bitrate",
            AlertSeverity::WARNING,
            AlertRuleConfig::new(
                AlertRuleQuery::metric("stat_consumer_video_bitrate"),
                AlertRuleCondition::new(AlertConditionOperator::GREATER_THAN, 2500.0),
            ),
        );
        assert_eq!(
            serde_json::to_value(&params).unwrap(),
            json!({
                "name": "High bitrate",
                "severity": "warning",
                "ruleConfig": {
                    "query": {"metric": "stat_consumer_video_bitrate"},
                    "condition": {"operator": ">", "threshold": 2500.0},
                },
            })
        );
    }

    #[test]
    fn an_unknown_severity_on_a_response_does_not_fail() {
        // The config + enums ride on the response, so a value this SDK predates
        // must round-trip rather than break the whole rule.
        let rule: AlertRule = serde_json::from_value(json!({
            "alertRuleId": "ar-1",
            "severity": "catastrophic",
            "notificationChannelIds": null,
        }))
        .unwrap();
        assert_eq!(rule.severity.unwrap().as_str(), "catastrophic");
        assert!(rule.notification_channel_ids.is_empty());
    }
}