appks/service/repo/webhooks.rs

use serde::{Deserialize, Serialize};
use std::net::IpAddr;
use url::Url;
use uuid::Uuid;

use crate::error::AppError;
use crate::models::common::{EventType, Role};
use crate::models::repos::RepoWebhook;
use crate::service::RepoService;
use crate::session::Session;

use super::util::{clamp_limit_offset, ensure_affected, required_text, set_local_user_id};

/// Validate webhook URL for SSRF protection
fn validate_webhook_url(url_str: &str) -> Result<(), AppError> {
    let url = Url::parse(url_str).map_err(|_| AppError::BadRequest("Invalid URL format".into()))?;

    // Only allow HTTPS
    if url.scheme() != "https" {
        return Err(AppError::BadRequest(
            "Webhook URL must use HTTPS protocol".into(),
        ));
    }

    let host = url
        .host_str()
        .ok_or_else(|| AppError::BadRequest("URL must have a host".into()))?;

    // Reject IP addresses directly (require domain names)
    if host.parse::<IpAddr>().is_ok() {
        return Err(AppError::BadRequest(
            "Webhook URL must use a domain name, not an IP address".into(),
        ));
    }

    // Reject localhost and common local domains
    let host_lower = host.to_lowercase();
    if host_lower == "localhost"
        || host_lower.ends_with(".localhost")
        || host_lower == "127.0.0.1"
        || host_lower == "::1"
        || host_lower == "0.0.0.0"
        || host_lower.ends_with(".local")
        || host_lower.ends_with(".internal")
    {
        return Err(AppError::BadRequest(
            "Webhook URL cannot point to localhost or internal domains".into(),
        ));
    }

    // Reject metadata endpoints (AWS, GCP, Azure)
    if host == "169.254.169.254" || host == "metadata.google.internal" {
        return Err(AppError::BadRequest(
            "Webhook URL cannot point to cloud metadata endpoints".into(),
        ));
    }

    // Note: Full DNS resolution and IP validation would require async DNS lookup
    // and checking against private IP ranges. This is a basic validation layer.
    // Production systems should implement async DNS resolution and IP validation
    // at the webhook delivery layer.

    Ok(())
}

#[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)]
pub struct CreateWebhookParams {
    pub url: String,
    pub secret_ciphertext: Option<String>,
    pub events: Vec<EventType>,
    pub active: Option<bool>,
}

#[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)]
pub struct UpdateWebhookParams {
    pub url: Option<String>,
    pub secret_ciphertext: Option<String>,
    pub events: Option<Vec<EventType>>,
    pub active: Option<bool>,
}

impl RepoService {
    pub async fn repo_webhooks(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<RepoWebhook>, AppError> {
        let user_uid = ctx.user().ok_or(AppError::Unauthorized)?;
        let repo = self.resolve_repo(wk_name, repo_name).await?;
        let repo_id = repo.id;
        self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin)
            .await?;
        let (limit, offset) = clamp_limit_offset(limit, offset);
        sqlx::query_as::<_, RepoWebhook>(
            "SELECT id, repo_id, url, secret_ciphertext, events, active, last_delivery_status, last_delivery_at, created_by, created_at, updated_at FROM repo_webhook WHERE repo_id = $1 ORDER BY created_at DESC LIMIT $2 OFFSET $3",
        )
        .bind(repo_id)
        .bind(limit)
        .bind(offset)
        .fetch_all(self.ctx.db.reader())
        .await
        .map_err(AppError::Database)
    }

    pub async fn repo_create_webhook(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        params: CreateWebhookParams,
    ) -> Result<RepoWebhook, AppError> {
        let user_uid = ctx.user().ok_or(AppError::Unauthorized)?;
        let repo = self.resolve_repo(wk_name, repo_name).await?;
        let repo_id = repo.id;
        self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin)
            .await?;
        let url = required_text(params.url, "url")?;
        validate_webhook_url(&url)?;
        if params.events.is_empty() {
            return Err(AppError::BadRequest(
                "at least one event is required".into(),
            ));
        }

        let now = chrono::Utc::now();
        let mut txn = self
            .ctx
            .db
            .writer()
            .begin()
            .await
            .map_err(|_| AppError::TxnError)?;
        sqlx::query(set_local_user_id(user_uid))
            .execute(&mut *txn)
            .await
            .map_err(AppError::Database)?;

        let result = sqlx::query_as::<_, RepoWebhook>(
            "INSERT INTO repo_webhook (id, repo_id, url, secret_ciphertext, events, active, \
             created_by, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $8) RETURNING id, repo_id, url, secret_ciphertext, events, active, last_delivery_status, last_delivery_at, created_by, created_at, updated_at",
        )
        .bind(Uuid::now_v7())
        .bind(repo_id)
        .bind(&url)
        .bind(&params.secret_ciphertext)
        .bind(&params.events)
        .bind(params.active.unwrap_or(true))
        .bind(user_uid)
        .bind(now)
        .fetch_one(&mut *txn)
        .await
        .map_err(AppError::Database)?;

        txn.commit().await.map_err(|_| AppError::TxnError)?;
        Ok(result)
    }

    pub async fn repo_update_webhook(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        webhook_id: Uuid,
        params: UpdateWebhookParams,
    ) -> Result<RepoWebhook, AppError> {
        let user_uid = ctx.user().ok_or(AppError::Unauthorized)?;
        let repo = self.resolve_repo(wk_name, repo_name).await?;
        let repo_id = repo.id;
        self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin)
            .await?;

        let current = sqlx::query_as::<_, RepoWebhook>(
            "SELECT id, repo_id, url, secret_ciphertext, events, active, last_delivery_status, last_delivery_at, created_by, created_at, updated_at FROM repo_webhook WHERE id = $1 AND repo_id = $2",
        )
        .bind(webhook_id)
        .bind(repo_id)
        .fetch_optional(self.ctx.db.reader())
        .await
        .map_err(AppError::Database)?
        .ok_or(AppError::NotFound("webhook not found".into()))?;

        let url = params
            .url
            .as_ref()
            .map(|u| u.trim().to_string())
            .unwrap_or(current.url);

        // Validate URL if it was updated
        if params.url.is_some() {
            validate_webhook_url(&url)?;
        }

        let active = params.active.unwrap_or(current.active);
        let now = chrono::Utc::now();

        let mut txn = self
            .ctx
            .db
            .writer()
            .begin()
            .await
            .map_err(|_| AppError::TxnError)?;
        sqlx::query(set_local_user_id(user_uid))
            .execute(&mut *txn)
            .await
            .map_err(AppError::Database)?;

        let result = sqlx::query_as::<_, RepoWebhook>(
            "UPDATE repo_webhook SET url = $1, secret_ciphertext = $2, events = $3, \
             active = $4, updated_at = $5 WHERE id = $6 AND repo_id = $7 RETURNING id, repo_id, url, secret_ciphertext, events, active, last_delivery_status, last_delivery_at, created_by, created_at, updated_at",
        )
        .bind(&url)
        .bind(params.secret_ciphertext.or(current.secret_ciphertext))
        .bind(params.events.unwrap_or(current.events))
        .bind(active)
        .bind(now)
        .bind(webhook_id)
        .bind(repo_id)
        .fetch_one(&mut *txn)
        .await
        .map_err(AppError::Database)?;

        txn.commit().await.map_err(|_| AppError::TxnError)?;
        Ok(result)
    }

    pub async fn repo_delete_webhook(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        webhook_id: Uuid,
    ) -> Result<(), AppError> {
        let user_uid = ctx.user().ok_or(AppError::Unauthorized)?;
        let repo = self.resolve_repo(wk_name, repo_name).await?;
        let repo_id = repo.id;
        self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin)
            .await?;

        let mut txn = self
            .ctx
            .db
            .writer()
            .begin()
            .await
            .map_err(|_| AppError::TxnError)?;
        sqlx::query(set_local_user_id(user_uid))
            .execute(&mut *txn)
            .await
            .map_err(AppError::Database)?;

        let result = sqlx::query("DELETE FROM repo_webhook WHERE id = $1 AND repo_id = $2")
            .bind(webhook_id)
            .bind(repo_id)
            .execute(&mut *txn)
            .await
            .map_err(AppError::Database)?;
        ensure_affected(result.rows_affected(), "webhook not found")?;

        txn.commit().await.map_err(|_| AppError::TxnError)?;
        Ok(())
    }

    pub async fn repo_webhook_deliveries(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        webhook_id: Uuid,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<serde_json::Value>, AppError> {
        let user_uid = ctx.user().ok_or(AppError::Unauthorized)?;
        let repo = self.resolve_repo(wk_name, repo_name).await?;
        let _repo_id = repo.id;
        self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin)
            .await?;

        let _webhook = sqlx::query_scalar::<_, bool>(
            "SELECT EXISTS(SELECT 1 FROM repo_webhook WHERE id = $1 AND repo_id = $2)",
        )
        .bind(webhook_id)
        .bind(_repo_id)
        .fetch_one(self.ctx.db.reader())
        .await
        .map_err(AppError::Database)?;

        if !_webhook {
            return Err(AppError::NotFound("webhook not found".into()));
        }

        let _ = (limit, offset);
        Ok(vec![])
    }

    pub async fn repo_retry_webhook_delivery(
        &self,
        ctx: &Session,
        wk_name: &str,
        repo_name: &str,
        webhook_id: Uuid,
        delivery_id: Uuid,
    ) -> Result<(), AppError> {
        let user_uid = ctx.user().ok_or(AppError::Unauthorized)?;
        let repo = self.resolve_repo(wk_name, repo_name).await?;
        self.ensure_repo_role_at_least(user_uid, &repo, Role::Admin)
            .await?;

        let _webhook = sqlx::query_scalar::<_, bool>(
            "SELECT EXISTS(SELECT 1 FROM repo_webhook WHERE id = $1 AND repo_id = $2)",
        )
        .bind(webhook_id)
        .bind(repo.id)
        .fetch_one(self.ctx.db.reader())
        .await
        .map_err(AppError::Database)?;

        if !_webhook {
            return Err(AppError::NotFound("webhook not found".into()));
        }

        let _ = delivery_id;
        Ok(())
    }
}