/*
 * Parseable Server (C) 2022 - 2025 Parseable, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 */

use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
    time::Duration,
};

use async_trait::async_trait;
use base64::Engine;
use chrono::Utc;
use http::header::AUTHORIZATION;
use itertools::Itertools;
use once_cell::sync::Lazy;
use serde_json::{Value, json};
use tokio::sync::{Mutex as AsyncMutex, RwLock};
use tracing::{error, trace, warn};
use ulid::Ulid;
use url::Url;

use super::{
    ALERTS,
    outbound_http_policy::{self, AlertTargetKind},
    title_registry,
};
use crate::{
    alerts::{AlertError, AlertState, Context, alert_traits::CallableTarget},
    metastore::metastore_traits::MetastoreObject,
    parseable::{DEFAULT_TENANT, PARSEABLE},
    storage::object_storage::target_json_path,
};

pub static TARGETS: Lazy<TargetConfigs> = Lazy::new(|| TargetConfigs {
    target_configs: RwLock::new(HashMap::new()),
    mutation_lock: AsyncMutex::new(()),
});

#[derive(Debug)]
pub struct TargetConfigs {
    pub target_configs: RwLock<HashMap<String, HashMap<Ulid, Target>>>,
    mutation_lock: AsyncMutex<()>,
}

impl TargetConfigs {
    /// Loads alerts from disk, blocks
    pub async fn load(&self) -> anyhow::Result<()> {
        let targets = PARSEABLE.metastore.get_targets().await?;
        let mut map = self.target_configs.write().await;
        for (tenant_id, targets) in targets {
            let inner = targets
                .into_iter()
                .map(|mut t| {
                    t.tenant = Some(tenant_id.clone());
                    (t.id, t)
                })
                .collect();
            map.insert(tenant_id, inner);
        }

        Ok(())
    }

    pub async fn create(&self, target: Target) -> Result<(), AlertError> {
        let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT);
        let _mutation_guard = self.mutation_lock.lock().await;
        // Another node may have created a target since this node loaded its cache.
        let saved = PARSEABLE
            .metastore
            .get_targets_for_tenant(&target.tenant)
            .await?;
        if saved.iter().any(|item| item.name == target.name) {
            return Err(AlertError::DuplicateTargetName(target.name));
        }
        if saved.iter().any(|item| item.id == target.id) {
            return Err(AlertError::InvalidTargetModification(
                "Target ID already exists".to_string(),
            ));
        }
        if !title_registry::reserve(&target.name, target.id, &target.tenant).await? {
            return Err(AlertError::DuplicateTargetName(target.name));
        }
        if let Err(error) = PARSEABLE
            .metastore
            .put_target(&target, &target.tenant)
            .await
        {
            title_registry::release(&target.name, target.id, &target.tenant).await?;
            return Err(error.into());
        }
        self.target_configs
            .write()
            .await
            .entry(tenant_id.to_owned())
            .or_default()
            .insert(target.id, target);
        Ok(())
    }

    pub async fn update(&self, target: Target) -> Result<(), AlertError> {
        PARSEABLE
            .metastore
            .put_target(&target, &target.tenant)
            .await?;
        let mut map = self.target_configs.write().await;
        let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT);
        map.entry(tenant_id.to_owned())
            .or_default()
            .insert(target.id, target);
        // map.insert(target.id, target.clone());
        Ok(())
    }

    pub async fn list(&self, tenant_id: &Option<String>) -> Result<Vec<Target>, AlertError> {
        let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
        let targets = if let Some(targets) = self.target_configs.read().await.get(tenant_id) {
            targets.values().cloned().collect_vec()
        } else {
            vec![]
        };
        Ok(targets)
    }

    pub async fn get_target_by_id(
        &self,
        target_id: &Ulid,
        tenant_id: &Option<String>,
    ) -> Result<Target, AlertError> {
        let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
        let target = if let Some(targets) = self.target_configs.read().await.get(tenant_id) {
            targets
                .get(target_id)
                .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
                .cloned()?
        } else {
            return Err(AlertError::InvalidTargetID(target_id.to_string()));
        };

        Ok(target)
    }

    pub async fn delete(
        &self,
        target_id: &Ulid,
        tenant_id: &Option<String>,
    ) -> Result<Target, AlertError> {
        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
        let _mutation_guard = self.mutation_lock.lock().await;
        // ensure that the target is not being used by any alert
        let guard = ALERTS.read().await;
        let alerts = if let Some(alerts) = guard.as_ref() {
            alerts
        } else {
            return Err(AlertError::CustomError("No AlertManager set".into()));
        };

        for alert in alerts.get_all_alerts(tenant_id).await.values() {
            if alert.get_targets().contains(target_id) {
                return Err(AlertError::TargetInUse);
            }
        }
        let target = if let Some(targets) = self.target_configs.read().await.get(tenant) {
            targets
                .get(target_id)
                .cloned()
                .ok_or(AlertError::InvalidTargetID(target_id.to_string()))?
        } else {
            return Err(AlertError::InvalidTargetID(target_id.to_string()));
        };
        title_registry::release(&target.name, target.id, tenant_id).await?;
        PARSEABLE
            .metastore
            .delete_target(&target, tenant_id)
            .await?;
        if let Some(targets) = self.target_configs.write().await.get_mut(tenant) {
            targets.remove(target_id);
        }
        Ok(target)
    }
}

#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
#[serde(untagged)]
pub enum Retry {
    Infinite,
    Finite(usize),
}

impl Default for Retry {
    fn default() -> Self {
        Retry::Finite(1)
    }
}

#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
#[serde(try_from = "TargetVerifier")]
pub struct Target {
    pub name: String,
    #[serde(flatten)]
    pub target: TargetType,
    #[serde(default = "Ulid::new")]
    pub id: Ulid,
    pub tenant: Option<String>,
}

impl Target {
    pub async fn validate_outbound_policy(&self) -> Result<(), AlertError> {
        let tenant_id = &self.tenant;
        match &self.target {
            TargetType::Slack(target) => {
                outbound_http_policy::prepare_alert_target(
                    tenant_id,
                    &target.endpoint,
                    AlertTargetKind::Slack,
                    false,
                    None,
                )
                .await?;
            }
            TargetType::Other(target) => {
                outbound_http_policy::prepare_alert_target(
                    tenant_id,
                    &target.endpoint,
                    AlertTargetKind::Webhook,
                    target.skip_tls_check,
                    Some(&target.headers),
                )
                .await?;
            }
            TargetType::AlertManager(target) => {
                let prepared = outbound_http_policy::prepare_alert_target(
                    tenant_id,
                    &target.endpoint,
                    AlertTargetKind::AlertManager,
                    target.skip_tls_check,
                    None,
                )
                .await?;
                if target.auth.is_some() && !prepared.authorization_allowed {
                    return Err(outbound_http_policy::OutboundPolicyError::DeniedHeader(
                        AUTHORIZATION.as_str().to_string(),
                    )
                    .into());
                }
            }
        }

        Ok(())
    }
    pub fn mask(self) -> Value {
        match self.target {
            TargetType::Slack(slack_web_hook) => {
                json!({
                   "name":self.name,
                   "type":"slack",
                   "endpoint":format!("{}://********", slack_web_hook.endpoint.scheme()),
                   "id":self.id
                })
            }
            TargetType::Other(other_web_hook) => {
                let safe_headers: HashMap<String, String> = other_web_hook
                    .headers
                    .into_keys()
                    .map(|k| (k, "********".to_string()))
                    .collect();
                json!({
                    "name":self.name,
                    "type":"webhook",
                    "endpoint": format!("{}://********", other_web_hook.endpoint.scheme()),
                    "headers":safe_headers,
                    "skipTlsCheck":other_web_hook.skip_tls_check,
                    "id":self.id
                })
            }
            TargetType::AlertManager(alert_manager) => {
                let endpoint = format!("{}://********", alert_manager.endpoint.scheme());
                if let Some(auth) = alert_manager.auth {
                    let password = "********";
                    json!({
                        "name":self.name,
                        "type":"webhook",
                        "endpoint":endpoint,
                        "username":auth.username,
                        "password":password,
                        "skipTlsCheck":alert_manager.skip_tls_check,
                        "id":self.id
                    })
                } else {
                    json!({
                        "name":self.name,
                        "type":"webhook",
                        "endpoint":endpoint,
                        "username":Value::Null,
                        "password":Value::Null,
                        "skipTlsCheck":alert_manager.skip_tls_check,
                        "id":self.id
                    })
                }
            }
        }
    }

    pub fn call(&self, context: Context) {
        trace!("target.call context- {context:?}");
        let timeout = context.notification_config.clone();
        let resolves = context.alert_info.alert_state;
        let mut state = timeout.state.lock().unwrap();
        trace!("target.call state- {state:?}");
        state.alert_state = resolves;

        match resolves {
            AlertState::Triggered => {
                if !state.timed_out {
                    // call once and then start sleeping
                    // reduce repeats by 1
                    call_target(self.target.clone(), self.tenant.clone(), context.clone());
                    // set state
                    state.timed_out = true;
                    state.awaiting_resolve = true;
                    drop(state);
                    self.spawn_timeout_task(&timeout, context.clone());
                }
            }
            alert_state @ AlertState::NotTriggered => {
                state.alert_state = alert_state;
                if state.timed_out {
                    // if in timeout and resolve came in, only process if it's the first one ( awaiting resolve )
                    if state.awaiting_resolve {
                        state.awaiting_resolve = false;
                    } else {
                        // no further resolve will be considered in timeout period
                        return;
                    }
                }

                call_target(self.target.clone(), self.tenant.clone(), context);
            }
            // do not send out any notifs
            // (an eval should not have run!)
            AlertState::Disabled => {}
        }
    }

    fn spawn_timeout_task(&self, target_timeout: &NotificationConfig, alert_context: Context) {
        trace!("repeat-\n{target_timeout:?}");
        let state = Arc::clone(&target_timeout.state);
        let retry = target_timeout.times;
        let timeout = target_timeout.interval;
        let target = self.target.clone();
        let alert_id = alert_context.alert_info.alert_id;
        let tenant_id = self.tenant.clone();

        let sleep_and_check_if_call =
            move |timeout_state: Arc<Mutex<TimeoutState>>, current_state: AlertState| async move {
                tokio::time::sleep(Duration::from_secs(timeout * 60)).await;

                let mut state = timeout_state.lock().unwrap();

                if current_state == AlertState::Triggered {
                    state.awaiting_resolve = true;
                    true
                } else {
                    state.timed_out = false;
                    false
                }
            };

        trace!("Spawning retry task");
        tokio::spawn(async move {
            // Get alerts manager reference once at the start
            let alerts = {
                let guard = ALERTS.read().await;
                if let Some(alerts) = guard.as_ref() {
                    alerts.clone()
                } else {
                    error!("No AlertManager set for alert_id: {alert_id}, stopping timeout task");
                    *state.lock().unwrap() = TimeoutState::default();
                    return;
                }
            }; // Lock released immediately

            match retry {
                Retry::Infinite => loop {
                    let current_state = if let Ok(state) =
                        alerts.get_state(alert_id, &tenant_id).await
                    {
                        state
                    } else {
                        *state.lock().unwrap() = TimeoutState::default();
                        warn!(
                            "Unable to fetch state for given alert_id- {alert_id}, stopping target notifs"
                        );
                        return;
                    };

                    let should_call =
                        sleep_and_check_if_call(Arc::clone(&state), current_state).await;
                    if should_call {
                        call_target(target.clone(), tenant_id.clone(), alert_context.clone())
                    }
                },
                Retry::Finite(times) => {
                    for _ in 0..(times - 1) {
                        let current_state = if let Ok(state) =
                            alerts.get_state(alert_id, &tenant_id).await
                        {
                            state
                        } else {
                            *state.lock().unwrap() = TimeoutState::default();
                            warn!(
                                "Unable to fetch state for given alert_id- {alert_id}, stopping target notifs"
                            );
                            return;
                        };

                        let should_call =
                            sleep_and_check_if_call(Arc::clone(&state), current_state).await;
                        if should_call {
                            call_target(target.clone(), tenant_id.clone(), alert_context.clone())
                        }
                    }
                }
            }
            *state.lock().unwrap() = TimeoutState::default();
        });
    }
}

impl MetastoreObject for Target {
    fn get_object_path(&self) -> String {
        target_json_path(&self.id, &self.tenant).to_string()
    }

    fn get_object_id(&self) -> String {
        self.id.to_string()
    }
}

fn call_target(target: TargetType, tenant_id: Option<String>, context: Context) {
    trace!("Calling target with context- {context:?}");
    tokio::spawn(async move { target.call(&tenant_id, &context).await });
}

#[derive(Debug, serde::Deserialize)]
pub struct NotificationConfigVerifier {
    interval: Option<u64>,
    times: Option<usize>,
}

#[derive(Debug, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TargetVerifier {
    pub name: String,
    #[serde(flatten)]
    pub target: TargetType,
    #[serde(default)]
    pub notification_config: Option<NotificationConfigVerifier>,
    #[serde(default = "Ulid::new")]
    pub id: Ulid,
    pub tenant_id: Option<String>,
}

impl TryFrom<TargetVerifier> for Target {
    type Error = String;

    fn try_from(value: TargetVerifier) -> Result<Self, Self::Error> {
        let mut timeout = NotificationConfig::default();

        // Default is Infinite in case of alertmanager
        if matches!(value.target, TargetType::AlertManager(_)) {
            timeout.times = Retry::Infinite
        }

        if let Some(notification_config) = value.notification_config {
            let interval = notification_config.interval.map(|ref interval| *interval);

            if let Some(interval) = interval {
                timeout.interval = interval
            }

            if let Some(times) = notification_config.times {
                timeout.times = Retry::Finite(times)
            }
        }

        Ok(Target {
            name: value.name,
            target: value.target,
            id: value.id,
            tenant: value.tenant_id,
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
#[serde(tag = "type")]
#[serde(deny_unknown_fields)]
pub enum TargetType {
    #[serde(rename = "slack")]
    Slack(SlackWebHook),
    #[serde(rename = "webhook")]
    Other(OtherWebHook),
    #[serde(rename = "alertManager")]
    AlertManager(AlertManager),
}

impl TargetType {
    pub async fn call(&self, tenant_id: &Option<String>, payload: &Context) {
        match self {
            TargetType::Slack(target) => target.call(tenant_id, payload).await,
            TargetType::Other(target) => target.call(tenant_id, payload).await,
            TargetType::AlertManager(target) => target.call(tenant_id, payload).await,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct SlackWebHook {
    endpoint: Url,
}

#[async_trait]
impl CallableTarget for SlackWebHook {
    async fn call(&self, tenant_id: &Option<String>, payload: &Context) {
        let alert = match payload.alert_info.alert_state {
            AlertState::Triggered => {
                serde_json::json!({ "text": payload.message })
            }
            AlertState::NotTriggered => {
                serde_json::json!({ "text": payload.default_resolved_string() })
            }
            AlertState::Disabled => {
                serde_json::json!({ "text": payload.default_disabled_string() })
            }
        };
        // Revalidate immediately before delivery because stored targets and DNS can change.
        let prepared = match outbound_http_policy::prepare_alert_target(
            tenant_id,
            &self.endpoint,
            AlertTargetKind::Slack,
            false,
            None,
        )
        .await
        {
            Ok(prepared) => prepared,
            Err(err) => {
                error!("Slack alert target rejected by outbound policy:{err}");
                return;
            }
        };
        if let Err(e) = prepared
            .client
            .post(self.endpoint.clone())
            .json(&alert)
            .send()
            .await
        {
            error!("Couldn't make call to webhook, error:{}", e)
        }
    }
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OtherWebHook {
    endpoint: Url,
    #[serde(default)]
    headers: HashMap<String, String>,
    #[serde(default)]
    skip_tls_check: bool,
}

#[async_trait]
impl CallableTarget for OtherWebHook {
    async fn call(&self, tenant_id: &Option<String>, payload: &Context) {
        let alert = match payload.alert_info.alert_state {
            AlertState::Triggered => payload.message.clone(),
            AlertState::NotTriggered => payload.default_resolved_string(),
            AlertState::Disabled => payload.default_disabled_string(),
        };
        let prepared = match outbound_http_policy::prepare_alert_target(
            tenant_id,
            &self.endpoint,
            AlertTargetKind::Webhook,
            self.skip_tls_check,
            Some(&self.headers),
        )
        .await
        {
            Ok(prepared) => prepared,
            Err(err) => {
                error!("Webhook alert target rejected by outbound policy:{err}");
                return;
            }
        };
        let request = prepared
            .client
            .post(self.endpoint.clone())
            .headers(prepared.headers);
        if let Err(e) = request.body(alert).send().await {
            error!("Couldn't make call to webhook, error:{}", e)
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AlertManager {
    endpoint: Url,
    #[serde(default)]
    skip_tls_check: bool,
    #[serde(flatten)]
    auth: Option<Auth>,
}

#[async_trait]
impl CallableTarget for AlertManager {
    async fn call(&self, tenant_id: &Option<String>, payload: &Context) {
        let mut alerts = serde_json::json!([{
          "labels": {
            "alertname": payload.alert_info.alert_name,
            // "stream": payload.stream,
            "deployment_instance": payload.deployment_info.deployment_instance,
            "deployment_id": payload.deployment_info.deployment_id,
            "deployment_mode": payload.deployment_info.deployment_mode
            },
          "annotations": {
            "message": "MESSAGE",
            "reason": "REASON"
          }
        }]);

        let alert = &mut alerts[0];

        // fill in status label accordingly
        match payload.alert_info.alert_state {
            AlertState::Triggered => alert["labels"]["status"] = "triggered".into(),
            AlertState::NotTriggered => {
                alert["labels"]["status"] = "not-triggered".into();
                alert["annotations"]["reason"] =
                    serde_json::Value::String(payload.default_resolved_string());
                alert["endsAt"] = Utc::now()
                    .to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
                    .into();
            }
            AlertState::Disabled => alert["labels"]["status"] = "disabled".into(),
        };

        let prepared = match outbound_http_policy::prepare_alert_target(
            tenant_id,
            &self.endpoint,
            AlertTargetKind::AlertManager,
            self.skip_tls_check,
            None,
        )
        .await
        {
            Ok(prepared) => prepared,
            Err(err) => {
                error!("Alertmanager target rejected by outbound policy:{err}");
                return;
            }
        };

        let mut request = prepared.client.post(self.endpoint.clone()).json(&alerts);
        if let Some(Auth { username, password }) = &self.auth {
            // Credentials are only safe when the destination itself is explicitly allowlisted.
            if !prepared.authorization_allowed {
                error!(
                    "Alertmanager credentials blocked by outbound policy for destination:{}",
                    self.endpoint
                );
                return;
            }
            let basic_auth_value = "Basic ".to_string()
                + &base64::prelude::BASE64_STANDARD.encode(format!("{username}:{password}"));
            request = request.header(AUTHORIZATION, basic_auth_value);
        }
        if let Err(e) = request.send().await {
            error!("Couldn't make call to alertmanager, error:{}", e)
        }
    }
}

#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)]
pub struct NotificationConfig {
    pub interval: u64,
    #[serde(skip)]
    pub times: Retry,
    #[serde(skip)]
    pub state: Arc<Mutex<TimeoutState>>,
}

impl Default for NotificationConfig {
    fn default() -> Self {
        Self {
            interval: 1,
            times: Retry::default(),
            state: Arc::<Mutex<TimeoutState>>::default(),
        }
    }
}

#[derive(Debug, Default, Clone, Copy)]
pub struct TimeoutState {
    pub alert_state: AlertState,
    pub timed_out: bool,
    pub awaiting_resolve: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Auth {
    username: String,
    password: String,
}

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

    #[test]
    fn test_masked_target_hides_secrets() {
        let target = Target {
            name: "test-target".into(),
            id: Ulid::new(),
            tenant: None,
            target: TargetType::AlertManager(AlertManager {
                endpoint: "https://internal.corp/api".parse().unwrap(),
                auth: Some(Auth {
                    username: "admin".into(),
                    password: "SuperSecretPassword123".into(),
                }),
                skip_tls_check: false,
            }),
        };

        let masked = target.mask();

        let masked_str = serde_json::to_string(&masked).unwrap();

        // These assertions MUST pass, or the build fails
        assert!(
            !masked_str.contains("SuperSecretPassword123"),
            "Password leaked in masked response!"
        );
        assert!(
            masked_str.contains("********"),
            "Expected redaction marker not found!"
        );
    }
}