use models::chat::outbound_webhooks;
use models::chat::webhook_deliveries::{self, WebhookDeliveryStatus};
use sea_orm::sea_query::Expr;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;

use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::authz::can_manage_space;
use crate::share::lookup::require_room_readable;

const MAX_DELIVERY_ATTEMPTS: i32 = 5;

/// 多数站点（知乎、微博、CSDN）会对默认爬虫 UA 直接 403，浏览器 UA 才有正常 HTML。
const USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) \
     Chrome/131.0.0.0 Safari/537.36";

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct OutboundWebhookDto {
    pub id: Uuid,
    pub room_id: Uuid,
    pub name: String,
    pub url: String,
    pub disabled: bool,
    pub last_status: Option<String>,
    pub last_used_at: Option<OffsetDateTime>,
    pub created_at: OffsetDateTime,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct OutboundWebhookSecretDto {
    pub id: Uuid,
    pub room_id: Uuid,
    pub name: String,
    pub url: String,
    pub secret: String,
    pub created_at: OffsetDateTime,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateOutboundWebhookCmd {
    pub name: String,
    pub url: String,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateOutboundWebhookCmd {
    #[serde(default)]
    pub name: Option<String>,
    #[serde(default)]
    pub url: Option<String>,
    #[serde(default)]
    pub disabled: Option<bool>,
}

impl From<outbound_webhooks::Model> for OutboundWebhookDto {
    fn from(w: outbound_webhooks::Model) -> Self {
        Self {
            id: w.id,
            room_id: w.room_id,
            name: w.name,
            url: w.url,
            disabled: w.disabled,
            last_status: w.last_status,
            last_used_at: w.last_used_at,
            created_at: w.created_at,
        }
    }
}

pub(crate) fn validate_outbound_url(url: &str) -> Result<http::Uri, ChatError> {
    let uri: http::Uri = url
        .parse()
        .map_err(|_| ChatError::BadRequest("invalid url".to_string()))?;
    let scheme = uri.scheme_str().unwrap_or("");
    if scheme != "http" && scheme != "https" {
        return Err(ChatError::BadRequest(
            "url scheme must be http or https".to_string(),
        ));
    }
    let host = uri
        .host()
        .ok_or_else(|| ChatError::BadRequest("url host is required".to_string()))?;
    if host.is_empty() {
        return Err(ChatError::BadRequest("url host is required".to_string()));
    }
    if let Ok(ip) = host.parse::<std::net::IpAddr>()
        && !is_public_ip(&ip)
    {
        return Err(ChatError::BadRequest(
            "url must not point at a private or loopback address".to_string(),
        ));
    }
    Ok(uri)
}

pub(crate) fn is_public_ip(ip: &std::net::IpAddr) -> bool {
    match ip {
        std::net::IpAddr::V4(v4) => {
            !v4.is_loopback()
                && !v4.is_private()
                && !v4.is_link_local()
                && !v4.is_broadcast()
                && !v4.is_documentation()
                && !v4.is_unspecified()
                && !(v4.octets()[0] == 100 && (v4.octets()[1] & 0b1100_0000) == 0b0100_0000)
        }
        std::net::IpAddr::V6(v6) => {
            !v6.is_loopback()
                && !v6.is_unspecified()
                && !v6.is_multicast()
                && {
                    let segments = v6.segments();
                    // 64:ff9b::/96 (NAT64), fc00::/7 (ULA), fe80::/10 (link-local),
                    // ::ffff:0:0/96 (IPv4-mapped) handled via to_ipv4 below.
                    !(segments[0] == 0x0064 && segments[1] == 0xff9b)
                        && (segments[0] & 0xfe00) != 0xfc00
                        && (segments[0] & 0xffc0) != 0xfe80
                }
                && match v6.to_ipv4_mapped() {
                    Some(v4) => is_public_ip(&std::net::IpAddr::V4(v4)),
                    None => true,
                }
        }
    }
}

/// 抓取结果：正文 + 最终落地的 URL。
///
/// 短链（`b23.tv` / `t.co` / `xhslink.com`）只有跟随重定向才能拿到真实内容，
/// 而落地地址又决定这条链接该按哪个平台渲染，所以两者一起返回。
pub(crate) struct Fetched {
    pub body: String,
    pub final_url: String,
}

/// 从 `content-type` 里摘 charset；缺失时按 `None`（交给调用方嗅探）。
fn charset_of(content_type: &str) -> Option<String> {
    let lower = content_type.to_ascii_lowercase();
    let start = lower.find("charset=")? + "charset=".len();
    let rest = &content_type[start..];
    let value = rest
        .split(|c: char| c == ';' || c == ',' || c.is_whitespace())
        .next()
        .unwrap_or("")
        .trim_matches(|c: char| c == '"' || c == '\'');
    (!value.is_empty()).then(|| value.to_string())
}

/// 按声明编码解码；没声明时先看 UTF-8 能否直通，否则回退 GB18030（中文站常见）。
pub(crate) fn decode_body(body: &[u8], content_type: &str) -> String {
    if let Some(label) = charset_of(content_type)
        && let Some(encoding) = encoding_rs::Encoding::for_label(label.as_bytes())
    {
        let (text, _, _) = encoding.decode(body);
        return text.into_owned();
    }
    match std::str::from_utf8(body) {
        Ok(text) => text.to_string(),
        Err(_) => {
            let (text, _, _) = encoding_rs::GB18030.decode(body);
            text.into_owned()
        }
    }
}

/// SSRF 防护后的 GET：只允许公网地址、限时限量，跟随重定向并对每一跳重做校验。
///
/// `expect_content_type` 非空时，响应 `content-type` 必须包含该片段（读 body 之前检查）。
/// 重定向是手工跟的：`reqwest` 自带的跳转策略不会拦截落到内网地址的 `Location`，
/// 所以每一跳都要重新过一遍 [`validate_outbound_url`] 与 DNS 归属检查。
pub(crate) async fn fetch_following(
    url: &str,
    accept: &str,
    expect_content_type: Option<&str>,
    max_bytes: usize,
) -> Result<Fetched, String> {
    fetch(url, accept, expect_content_type, max_bytes).await
}

const MAX_REDIRECTS: usize = 6;

async fn fetch(
    url: &str,
    accept: &str,
    expect_content_type: Option<&str>,
    max_bytes: usize,
) -> Result<Fetched, String> {
    let mut current = url.to_string();
    for _ in 0..=MAX_REDIRECTS {
        let uri = validate_outbound_url(&current).map_err(|e| e.to_string())?;
        let host = uri.host().unwrap_or("").to_string();
        let port = uri.port_u16();
        let addrs = tokio::net::lookup_host((host.as_str(), port.unwrap_or(443)))
            .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| !is_public_ip(&addr.ip())) {
            return Err("url resolves to a private or loopback address".to_string());
        }
        let mut client_builder = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .redirect(reqwest::redirect::Policy::none())
            .user_agent(USER_AGENT);
        for addr in &addrs {
            client_builder = client_builder.resolve(&host, *addr);
        }
        let client = client_builder.build().map_err(|e| e.to_string())?;
        let response = client
            .get(&current)
            .header("accept", accept)
            .send()
            .await
            .map_err(|e| format!("request failed: {e}"))?;

        if response.status().is_redirection() {
            let location = response
                .headers()
                .get("location")
                .and_then(|value| value.to_str().ok())
                .ok_or_else(|| "redirect without a location header".to_string())?;
            let base = url::Url::parse(&current).map_err(|e| e.to_string())?;
            current = base
                .join(location)
                .map_err(|e| format!("invalid redirect target: {e}"))?
                .to_string();
            continue;
        }
        if !response.status().is_success() {
            return Err(format!("endpoint returned {}", response.status().as_u16()));
        }
        let content_type = response
            .headers()
            .get("content-type")
            .and_then(|value| value.to_str().ok())
            .unwrap_or("")
            .to_string();
        let lowered = content_type.to_ascii_lowercase();
        if let Some(expected) = expect_content_type
            && !lowered.is_empty()
            && !lowered.contains(expected)
        {
            return Err(format!("unexpected content-type {lowered}"));
        }
        let mut body: Vec<u8> = Vec::new();
        let mut stream = response;
        while let Some(chunk) = stream.chunk().await.map_err(|e| e.to_string())? {
            body.extend_from_slice(&chunk);
            if body.len() >= max_bytes {
                break;
            }
        }
        return Ok(Fetched {
            body: decode_body(&body, &content_type),
            final_url: current,
        });
    }
    Err("too many redirects".to_string())
}

async fn require_room_manager(
    service: &ChatService,
    room_id: Uuid,
    caller: i64,
) -> Result<(), ChatError> {
    let ctx = require_room_readable(service, room_id, caller).await?;
    match &ctx.space {
        Some(space) => {
            if !can_manage_space(service, space, caller).await? {
                return Err(ChatError::Forbidden);
            }
        }
        None => return Err(ChatError::Forbidden),
    }
    Ok(())
}

fn generate_secret() -> String {
    use rand::Rng;
    const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
    let mut rng = rand::thread_rng();
    (0..48)
        .map(|_| CHARSET[rng.gen_range(0..CHARSET.len())] as char)
        .collect()
}

impl ChatService {
    pub async fn create_outbound_webhook(
        &self,
        caller: i64,
        room_id: Uuid,
        cmd: CreateOutboundWebhookCmd,
    ) -> Result<OutboundWebhookSecretDto, ChatError> {
        require_room_manager(self, room_id, caller).await?;
        let name = cmd.name.trim();
        if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
            return Err(ChatError::BadRequest("invalid webhook name".to_string()));
        }
        validate_outbound_url(cmd.url.trim())?;
        let now = OffsetDateTime::now_utc();
        let row = outbound_webhooks::ActiveModel {
            id: Set(Uuid::now_v7()),
            room_id: Set(room_id),
            user_id: Set(caller),
            name: Set(name.to_string()),
            url: Set(cmd.url.trim().to_string()),
            secret: Set(generate_secret()),
            disabled: Set(false),
            last_status: Set(None),
            last_used_at: Set(None),
            created_at: Set(now),
            updated_at: Set(now),
        }
        .insert(self.db().write())
        .await?;
        Ok(OutboundWebhookSecretDto {
            id: row.id,
            room_id: row.room_id,
            name: row.name,
            url: row.url,
            secret: row.secret,
            created_at: row.created_at,
        })
    }

    pub async fn list_outbound_webhooks(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<Vec<OutboundWebhookDto>, ChatError> {
        require_room_manager(self, room_id, caller).await?;
        let rows = outbound_webhooks::Entity::find()
            .filter(outbound_webhooks::Column::RoomId.eq(room_id))
            .all(self.db().read())
            .await?;
        Ok(rows.into_iter().map(OutboundWebhookDto::from).collect())
    }

    pub async fn update_outbound_webhook(
        &self,
        caller: i64,
        room_id: Uuid,
        webhook_id: Uuid,
        cmd: UpdateOutboundWebhookCmd,
    ) -> Result<OutboundWebhookDto, ChatError> {
        require_room_manager(self, room_id, caller).await?;
        let webhook = outbound_webhooks::Entity::find_by_id(webhook_id)
            .filter(outbound_webhooks::Column::RoomId.eq(room_id))
            .one(self.db().write())
            .await?
            .ok_or(ChatError::NotFound)?;
        let mut am: outbound_webhooks::ActiveModel = webhook.into();
        if let Some(name) = cmd.name {
            let name = name.trim();
            if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
                return Err(ChatError::BadRequest("invalid webhook name".to_string()));
            }
            am.name = Set(name.to_string());
        }
        if let Some(url) = cmd.url {
            validate_outbound_url(url.trim())?;
            am.url = Set(url.trim().to_string());
        }
        if let Some(disabled) = cmd.disabled {
            am.disabled = Set(disabled);
        }
        am.updated_at = Set(OffsetDateTime::now_utc());
        let updated = am.update(self.db().write()).await?;
        Ok(OutboundWebhookDto::from(updated))
    }

    pub async fn delete_outbound_webhook(
        &self,
        caller: i64,
        room_id: Uuid,
        webhook_id: Uuid,
    ) -> Result<(), ChatError> {
        require_room_manager(self, room_id, caller).await?;
        let deleted = outbound_webhooks::Entity::delete_many()
            .filter(outbound_webhooks::Column::Id.eq(webhook_id))
            .filter(outbound_webhooks::Column::RoomId.eq(room_id))
            .exec(self.db().write())
            .await?;
        if deleted.rows_affected == 0 {
            return Err(ChatError::NotFound);
        }
        webhook_deliveries::Entity::delete_many()
            .filter(webhook_deliveries::Column::WebhookId.eq(webhook_id))
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
            .exec(self.db().write())
            .await?;
        Ok(())
    }

    #[tracing::instrument(name = "chat.deliver_webhooks", skip_all)]
    pub async fn deliver_pending_webhook_events(&self) -> Result<u64, ChatError> {
        let now = OffsetDateTime::now_utc();

        let _ = webhook_deliveries::Entity::update_many()
            .col_expr(
                webhook_deliveries::Column::Status,
                Expr::value(WebhookDeliveryStatus::Pending),
            )
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Delivering))
            .filter(
                webhook_deliveries::Column::NextRetryAt
                    .lt(now - time::Duration::minutes(crate::worker::STUCK_CLAIM_MINUTES)),
            )
            .exec(self.db().write())
            .await?;

        let due = webhook_deliveries::Entity::find()
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
            .filter(webhook_deliveries::Column::NextRetryAt.lte(now))
            .limit(50)
            .all(self.db().write())
            .await?;
        let count = due.len() as u64;
        for delivery in due {
            let claimed = webhook_deliveries::Entity::update_many()
                .col_expr(
                    webhook_deliveries::Column::Status,
                    Expr::value(WebhookDeliveryStatus::Delivering),
                )
                .filter(webhook_deliveries::Column::Id.eq(delivery.id))
                .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
                .exec(self.db().write())
                .await?;
            if claimed.rows_affected == 0 {
                continue;
            }

            let Some(webhook) = outbound_webhooks::Entity::find_by_id(delivery.webhook_id)
                .one(self.db().read())
                .await?
            else {
                self.finish_delivery(
                    delivery.id,
                    WebhookDeliveryStatus::Failed,
                    Some("webhook deleted".to_string()),
                    delivery.attempts,
                )
                .await;
                continue;
            };
            let result = if webhook.disabled {
                Err("webhook disabled".to_string())
            } else {
                self.post_webhook_payload(&webhook, &delivery).await
            };
            let (status, error) = match result {
                Ok(status_line) => {
                    let mut am: outbound_webhooks::ActiveModel = webhook.into();
                    am.last_status = Set(Some(status_line.clone()));
                    am.last_used_at = Set(Some(OffsetDateTime::now_utc()));
                    am.updated_at = Set(OffsetDateTime::now_utc());
                    let _ = am.update(self.db().write()).await;
                    (WebhookDeliveryStatus::Done, None)
                }
                Err(error) => {
                    let attempts = delivery.attempts + 1;
                    if attempts >= MAX_DELIVERY_ATTEMPTS {
                        (WebhookDeliveryStatus::Failed, Some(error))
                    } else {
                        let mut am: webhook_deliveries::ActiveModel = delivery.clone().into();
                        am.status = Set(WebhookDeliveryStatus::Pending);
                        am.attempts = Set(attempts);
                        am.next_retry_at = Set(OffsetDateTime::now_utc()
                            + time::Duration::seconds(60 * attempts as i64));
                        am.last_error = Set(Some(error.clone()));
                        am.updated_at = Set(OffsetDateTime::now_utc());
                        let _ = am.update(self.db().write()).await;
                        continue;
                    }
                }
            };
            self.finish_delivery(delivery.id, status, error, delivery.attempts)
                .await;
        }
        Ok(count)
    }

    async fn finish_delivery(
        &self,
        id: Uuid,
        status: WebhookDeliveryStatus,
        error: Option<String>,
        attempts: i32,
    ) {
        let am = webhook_deliveries::ActiveModel {
            id: Set(id),
            status: Set(status),
            attempts: Set(attempts + 1),
            last_error: Set(error),
            updated_at: Set(OffsetDateTime::now_utc()),
            ..Default::default()
        };
        let _ = am.update(self.db().write()).await;
    }

    async fn post_webhook_payload(
        &self,
        webhook: &outbound_webhooks::Model,
        delivery: &webhook_deliveries::Model,
    ) -> Result<String, String> {
        use hmac::{Hmac, KeyInit, Mac};
        use sha2::Sha256;

        let uri = validate_outbound_url(&webhook.url).map_err(|e| e.to_string())?;
        let host = uri.host().unwrap_or("").to_string();
        let port = uri.port_u16();

        let addrs = tokio::net::lookup_host((host.as_str(), port.unwrap_or(443)))
            .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| !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(webhook.secret.as_bytes()).map_err(|e| e.to_string())?;
        mac.update(&body);
        let signature = format!("sha256={}", hex_lower(&mac.finalize().into_bytes()));

        let mut client_builder = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .redirect(reqwest::redirect::Policy::none());
        for addr in &addrs {
            client_builder = client_builder.resolve(&host, *addr);
        }
        let client = client_builder.build().map_err(|e| e.to_string())?;

        let span = export::telemetry::http_client_span("POST", &webhook.url);
        let response = export::telemetry::send(
            span,
            client
                .post(&webhook.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),
        )
        .await
        .map_err(|e| {
            export::metrics::chat_webhook_delivery("error");
            format!("request failed: {e}")
        })?;
        let status = response.status().as_u16();
        if (200..300).contains(&status) {
            export::metrics::chat_webhook_delivery("ok");
            Ok(format!("delivered {status}"))
        } else {
            export::metrics::chat_webhook_delivery("error");
            Err(format!("endpoint returned {status}"))
        }
    }
}

fn hex_lower(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

pub async fn enqueue_outbound_event(
    service: &ChatService,
    room_id: Uuid,
    event_type: &str,
    payload: &serde_json::Value,
) {
    let hooks = outbound_webhooks::Entity::find()
        .filter(outbound_webhooks::Column::RoomId.eq(room_id))
        .filter(outbound_webhooks::Column::Disabled.eq(false))
        .all(service.db().read())
        .await
        .unwrap_or_default();
    if hooks.is_empty() {
        return;
    }
    let now = OffsetDateTime::now_utc();
    for hook in hooks {
        let row = webhook_deliveries::ActiveModel {
            id: Set(Uuid::now_v7()),
            webhook_id: Set(hook.id),
            room_id: Set(room_id),
            event_type: Set(event_type.to_string()),
            payload: Set(payload.clone()),
            status: Set(WebhookDeliveryStatus::Pending),
            attempts: Set(0),
            next_retry_at: Set(now),
            last_error: Set(None),
            created_at: Set(now),
            updated_at: Set(now),
        };
        if let Err(error) = row.insert(service.db().write()).await {
            tracing::warn!(%error, "failed to enqueue outbound webhook delivery");
        }
    }
}