use hmac::{Hmac, KeyInit, Mac};
use models::webhook::repo_webhook_delivery::{self, RepoWebhookDeliveryStatus};
use models::webhook::repo_webhooks;
use sea_orm::sea_query::Expr;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
};
use sha2::Sha256;
use time::OffsetDateTime;
use uuid::Uuid;

use crate::errors::CiResult;
use crate::secrets::decrypt;
use crate::service::CiService;

const MAX_RESPONSE_BODY: usize = 2048;
const CLAIM_STUCK_MINUTES: i64 = 10;

pub async fn enqueue_for_repo(
    service: &CiService,
    repo_id: Uuid,
    event_type: &str,
    payload: &serde_json::Value,
) -> CiResult<()> {
    let hooks = repo_webhooks::Entity::find()
        .filter(repo_webhooks::Column::RepoId.eq(repo_id))
        .filter(repo_webhooks::Column::Status.eq(models::webhook::webhooks::WebhookStatus::Active))
        .all(service.db_read())
        .await?;
    if hooks.is_empty() {
        return Ok(());
    }
    let now = OffsetDateTime::now_utc();
    for hook in hooks {
        let events: Vec<String> = serde_json::from_value(hook.events.clone()).unwrap_or_default();
        if !events.iter().any(|e| e == event_type || e == "*") {
            continue;
        }
        repo_webhook_delivery::ActiveModel {
            id: Set(Uuid::now_v7()),
            webhook_id: Set(hook.id),
            repo_id: Set(repo_id),
            event_type: Set(event_type.to_string()),
            payload: Set(payload.clone()),
            status: Set(RepoWebhookDeliveryStatus::Pending),
            attempts: Set(0),
            next_retry_at: Set(now),
            response_status: Set(None),
            response_body: Set(None),
            last_error: Set(None),
            created_at: Set(now),
            updated_at: Set(now),
        }
        .insert(service.db_write())
        .await?;
    }
    Ok(())
}

impl CiService {
    pub async fn deliver_pending_repo_webhooks(&self) -> CiResult<u64> {
        let now = OffsetDateTime::now_utc();
        let stuck = now - time::Duration::minutes(CLAIM_STUCK_MINUTES);
        repo_webhook_delivery::Entity::update_many()
            .col_expr(
                repo_webhook_delivery::Column::Status,
                Expr::value(RepoWebhookDeliveryStatus::Pending),
            )
            .col_expr(repo_webhook_delivery::Column::UpdatedAt, Expr::value(now))
            .filter(repo_webhook_delivery::Column::Status.eq(RepoWebhookDeliveryStatus::Delivering))
            .filter(repo_webhook_delivery::Column::UpdatedAt.lt(stuck))
            .exec(self.db_write())
            .await?;

        let due = repo_webhook_delivery::Entity::find()
            .filter(repo_webhook_delivery::Column::Status.eq(RepoWebhookDeliveryStatus::Pending))
            .filter(repo_webhook_delivery::Column::NextRetryAt.lte(now))
            .order_by_asc(repo_webhook_delivery::Column::NextRetryAt)
            .limit(50)
            .all(self.db_read())
            .await?;
        let count = due.len() as u64;
        let max_attempts = self.config().ci.outbound_max_attempts;
        let backoff_base = self.config().ci.outbound_backoff_base_secs;
        for delivery in due {
            let claimed = repo_webhook_delivery::Entity::update_many()
                .col_expr(
                    repo_webhook_delivery::Column::Status,
                    Expr::value(RepoWebhookDeliveryStatus::Delivering),
                )
                .col_expr(repo_webhook_delivery::Column::UpdatedAt, Expr::value(now))
                .filter(repo_webhook_delivery::Column::Id.eq(delivery.id))
                .filter(
                    repo_webhook_delivery::Column::Status.eq(RepoWebhookDeliveryStatus::Pending),
                )
                .exec(self.db_write())
                .await?;
            if claimed.rows_affected == 0 {
                continue;
            }
            let hook = repo_webhooks::Entity::find_by_id(delivery.webhook_id)
                .one(self.db_read())
                .await?;
            let Some(hook) = hook else {
                self.finish_repo_delivery(
                    delivery.id,
                    RepoWebhookDeliveryStatus::Failed,
                    delivery.attempts,
                    None,
                    None,
                    Some("webhook deleted".to_string()),
                )
                .await?;
                continue;
            };
            let secret = decrypt(self.encryption_key(), &hook.secret)
                .unwrap_or_else(|_| hook.secret.clone());
            let result = self.post_repo_payload(&hook.url, &secret, &delivery).await;
            match result {
                Ok((status, body)) => {
                    self.finish_repo_delivery(
                        delivery.id,
                        RepoWebhookDeliveryStatus::Done,
                        delivery.attempts + 1,
                        Some(status),
                        Some(body),
                        None,
                    )
                    .await?;
                }
                Err(error) => {
                    let attempts = delivery.attempts + 1;
                    if attempts >= max_attempts {
                        self.finish_repo_delivery(
                            delivery.id,
                            RepoWebhookDeliveryStatus::Failed,
                            attempts,
                            None,
                            None,
                            Some(error),
                        )
                        .await?;
                    } else {
                        let next = OffsetDateTime::now_utc()
                            + time::Duration::seconds(backoff_base * i64::from(attempts));
                        let mut am: repo_webhook_delivery::ActiveModel = delivery.clone().into();
                        am.status = Set(RepoWebhookDeliveryStatus::Pending);
                        am.attempts = Set(attempts);
                        am.next_retry_at = Set(next);
                        am.last_error = Set(Some(error));
                        am.updated_at = Set(OffsetDateTime::now_utc());
                        am.update(self.db_write()).await?;
                    }
                }
            }
        }
        Ok(count)
    }

    async fn finish_repo_delivery(
        &self,
        id: Uuid,
        status: RepoWebhookDeliveryStatus,
        attempts: i32,
        response_status: Option<i32>,
        response_body: Option<String>,
        last_error: Option<String>,
    ) -> CiResult<()> {
        let am = repo_webhook_delivery::ActiveModel {
            id: Set(id),
            status: Set(status),
            attempts: Set(attempts),
            response_status: Set(response_status),
            response_body: Set(response_body),
            last_error: Set(last_error),
            updated_at: Set(OffsetDateTime::now_utc()),
            ..Default::default()
        };
        am.update(self.db_write()).await?;
        Ok(())
    }

    async fn post_repo_payload(
        &self,
        url: &str,
        secret: &str,
        delivery: &repo_webhook_delivery::Model,
    ) -> Result<(i32, String), String> {
        let uri: reqwest::Url = url.parse().map_err(|e| format!("invalid url: {e}"))?;
        if uri.scheme() != "https" && uri.scheme() != "http" {
            return Err("webhook url must be http(s)".to_string());
        }
        let host = uri.host_str().unwrap_or("").to_string();
        if host.is_empty() {
            return Err("webhook url has no host".to_string());
        }
        let port = uri
            .port()
            .unwrap_or(if uri.scheme() == "https" { 443 } else { 80 });
        let addrs = tokio::net::lookup_host((host.as_str(), port))
            .await
            .map_err(|e| format!("dns resolution failed: {e}"))?
            .collect::<Vec<_>>();
        if addrs.is_empty() {
            return Err("dns resolution returned no address".to_string());
        }
        if addrs
            .iter()
            .any(|addr| !crate::net::is_public_ip(&addr.ip()))
        {
            return Err("url resolves to a private or loopback address".to_string());
        }
        let body = serde_json::to_vec(&delivery.payload).map_err(|e| e.to_string())?;
        let mut mac =
            Hmac::<Sha256>::new_from_slice(secret.as_bytes()).map_err(|e| e.to_string())?;
        mac.update(&body);
        let signature = format!(
            "sha256={}",
            crate::secrets::sha256_hex(&mac.finalize().into_bytes())
        );
        let mut builder = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(
                self.config().ci.outbound_timeout_secs.max(1),
            ))
            .redirect(reqwest::redirect::Policy::none());
        for addr in &addrs {
            builder = builder.resolve(&host, *addr);
        }
        let client = builder.build().map_err(|e| e.to_string())?;
        let response = client
            .post(url)
            .header("content-type", "application/json")
            .header("x-gitdata-event", &delivery.event_type)
            .header("x-gitdata-delivery", delivery.id.to_string())
            .header("x-gitdata-signature", signature)
            .body(body)
            .send()
            .await
            .map_err(|e| format!("request failed: {e}"))?;
        let status = response.status().as_u16() as i32;
        let text = response
            .text()
            .await
            .unwrap_or_default()
            .chars()
            .take(MAX_RESPONSE_BODY)
            .collect::<String>();
        if (200..300).contains(&status) {
            Ok((status, text))
        } else {
            Err(format!("endpoint returned {status}"))
        }
    }
}