use crate::events::{Labels, label};
use crate::models::users::user_events::{self, UserEventType};
use crate::models::users::user_ssh_keys::{self, PublicKeyType};
use crate::models::users::users;
use crate::session::Session;
use base64::Engine;
use base64::engine::general_purpose::STANDARD_NO_PAD;
use chrono::Utc;
use sea_orm::entity::Set;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, SqlErr,
    TransactionTrait,
};
use uuid::Uuid;

use crate::AuthSvc;
use crate::errors::{AuthError, AuthResult};
use crate::share::crypto;
use crate::share::types::ListQuery;
use crate::ssh_key::types::{
    MAX_SSH_KEY_DESCRIPTION_LEN, MAX_SSH_KEY_NAME_LEN, MAX_SSH_PUBLIC_KEY_LEN,
    MIN_SSH_PUBLIC_KEY_LEN, SshKeyAddRequest, SshKeyAlgorithm, SshKeySummary,
};

pub(crate) fn parse_public_key(raw: &str) -> AuthResult<(PublicKeyType, String)> {
    let key_line = raw.trim();
    if key_line.len() < MIN_SSH_PUBLIC_KEY_LEN || key_line.len() > MAX_SSH_PUBLIC_KEY_LEN {
        return Err(AuthError::SshKeyInvalid);
    }

    let mut parts = key_line.split_whitespace();
    let declared_type = parts.next().ok_or(AuthError::SshKeyInvalid)?;
    let blob_text = parts.next().ok_or(AuthError::SshKeyInvalid)?;
    let algorithm = match declared_type {
        "ssh-rsa" => PublicKeyType::RSA,
        "ssh-ed25519" => PublicKeyType::ED25519,
        "ecdsa-sha2-nistp256" => PublicKeyType::ECDSA,
        _ => return Err(AuthError::SshKeyInvalid),
    };

    let blob = STANDARD_NO_PAD
        .decode(blob_text.trim_end_matches('='))
        .map_err(|_| AuthError::SshKeyInvalid)?;
    if blob.len() < 8 {
        return Err(AuthError::SshKeyInvalid);
    }

    let header_len = u32::from_be_bytes([blob[0], blob[1], blob[2], blob[3]]) as usize;
    if header_len == 0 || 4 + header_len > blob.len() {
        return Err(AuthError::SshKeyInvalid);
    }
    let header =
        std::str::from_utf8(&blob[4..4 + header_len]).map_err(|_| AuthError::SshKeyInvalid)?;
    if header != declared_type {
        return Err(AuthError::SshKeyInvalid);
    }

    Ok((
        algorithm,
        format!("SHA256:{}", crypto::sha256_b64_standard(&blob)),
    ))
}

fn summary(record: user_ssh_keys::Model) -> SshKeySummary {
    SshKeySummary {
        id: record.id,
        name: record.name,
        description: record.description,
        algorithm: SshKeyAlgorithm::from(record.algorithm),
        fingerprint: record.fingerprint,
        public_key: record.public_key,
        created_at: record.created_at,
        expires_at: record.expires_at,
        last_used_at: record.last_used_at,
        revoked_at: record.revoked_at,
    }
}

impl AuthSvc {
    #[tracing::instrument(skip_all, fields(request_id = %request_id, client_ip = ?session.client_ip().ok().flatten()))]
    pub async fn add_ssh_key(
        &self,
        session: Session,
        request: SshKeyAddRequest,
        user_agent: Option<String>,
        ip_addr: Option<String>,
        request_id: String,
    ) -> AuthResult<SshKeySummary> {
        let user_id = self
            .require_session(&session, Some(request_id.as_str()))
            .await?;
        self.add_ssh_key_for_user(user_id, request, user_agent, ip_addr, request_id)
            .await
    }

    /// 与 `add_ssh_key` 同逻辑，但调用方已按 RPC 密钥确认过用户身份。
    pub async fn add_ssh_key_for_user(
        &self,
        user_id: i64,
        request: SshKeyAddRequest,
        user_agent: Option<String>,
        ip_addr: Option<String>,
        request_id: String,
    ) -> AuthResult<SshKeySummary> {
        let user = users::Entity::find_by_id(user_id)
            .one(self.db().read())
            .await?
            .ok_or(AuthError::Unauthenticated)?;
        if user.deleted_at.is_some() || !user.allow_use {
            return Err(AuthError::Unauthenticated);
        }

        let name = request.name.trim().to_owned();
        if name.is_empty() || name.chars().count() > MAX_SSH_KEY_NAME_LEN {
            return Err(AuthError::SshKeyNameInvalid);
        }
        let description = request
            .description
            .as_deref()
            .map(str::trim)
            .filter(|text| !text.is_empty())
            .map(str::to_owned);
        if description
            .as_ref()
            .is_some_and(|text| text.chars().count() > MAX_SSH_KEY_DESCRIPTION_LEN)
        {
            return Err(AuthError::MalformedRequest);
        }

        let expiry = request.expires_at;
        if expiry.is_some_and(|at| at <= Utc::now()) {
            return Err(AuthError::InvalidExpiry);
        }

        let public_key = request.public_key.trim().to_owned();
        let (algorithm, fingerprint) = parse_public_key(&public_key)?;

        let inserted = user_ssh_keys::ActiveModel {
            user_id: Set(user_id),
            name: Set(name),
            description: Set(description),
            public_key: Set(public_key),
            algorithm: Set(algorithm),
            fingerprint: Set(fingerprint),
            created_at: Set(Utc::now()),
            expires_at: Set(expiry),
            ..Default::default()
        }
        .insert(self.db().write())
        .await
        .map_err(|error| match error.sql_err() {
            Some(SqlErr::UniqueConstraintViolation(_)) => AuthError::SshKeyTaken,
            _ => AuthError::Database(crate::db::DbError::from(error)),
        })?;

        user_events::ActiveModel {
            user_id: Set(user_id),
            actor_user_id: Set(Some(user_id)),
            event_type: Set(UserEventType::PublicKeyAdded),
            occurred_at: Set(Utc::now()),
            user_agent: Set(user_agent),
            trace_id: Set(Some(request_id.clone())),
            ip_address: Set(ip_addr),
            details: Set(Some(serde_json::json!({
                "ssh_key_id": inserted.id,
                "name": inserted.name,
                "fingerprint": inserted.fingerprint,
            }))),
            ..Default::default()
        }
        .insert(self.db().write())
        .await?;

        let response = summary(inserted);
        self.emit_labeled(
            label::AUTH_SSH_KEY_CREATE,
            Labels::new()
                .with(label::USER_ID, user_id)
                .with(label::REQUEST_ID, &request_id),
            &response,
            Some(request_id.as_str()),
        )
        .await;
        Ok(response)
    }

    #[tracing::instrument(skip_all, fields(request_id = %_request_id, client_ip = ?session.client_ip().ok().flatten()))]
    pub async fn list_ssh_keys(
        &self,
        session: Session,
        query: ListQuery,
        _request_id: String,
    ) -> AuthResult<Vec<SshKeySummary>> {
        let user_id = self
            .require_session(&session, Some(_request_id.as_str()))
            .await?;
        let records = user_ssh_keys::Entity::find()
            .filter(user_ssh_keys::Column::UserId.eq(user_id))
            .order_by_desc(user_ssh_keys::Column::CreatedAt)
            .limit(query.limit())
            .offset(query.offset())
            .all(self.db().read())
            .await?;

        Ok(records.into_iter().map(summary).collect())
    }

    /// 吊销公钥与提高账号 `security_version` 在同一事务里完成：密钥泄露后，账号既有的访问令牌与
    #[tracing::instrument(skip_all, fields(request_id = %request_id, client_ip = ?session.client_ip().ok().flatten()))]
    pub async fn revoke_ssh_key(
        &self,
        session: Session,
        key_id: Uuid,
        user_agent: Option<String>,
        ip_addr: Option<String>,
        request_id: String,
    ) -> AuthResult<()> {
        let user_id = self
            .require_session(&session, Some(request_id.as_str()))
            .await?;
        let record = user_ssh_keys::Entity::find_by_id(key_id)
            .filter(user_ssh_keys::Column::UserId.eq(user_id))
            .one(self.db().read())
            .await?
            .ok_or(AuthError::SshKeyNotFound)?;
        if record.revoked_at.is_some() {
            return Ok(());
        }

        let transaction = self.db().write().begin().await?;
        let result: AuthResult<()> = async {
            user_ssh_keys::Entity::update_many()
                .col_expr(user_ssh_keys::Column::RevokedAt, Expr::value(Utc::now()))
                .filter(user_ssh_keys::Column::Id.eq(record.id))
                .exec(&transaction)
                .await?;

            users::Entity::update_many()
                .col_expr(
                    users::Column::SecurityVersion,
                    Expr::col(users::Column::SecurityVersion).add(1),
                )
                .col_expr(users::Column::UpdatedAt, Expr::value(Utc::now()))
                .filter(users::Column::Id.eq(user_id))
                .exec(&transaction)
                .await?;

            user_events::ActiveModel {
                user_id: Set(user_id),
                actor_user_id: Set(Some(user_id)),
                event_type: Set(UserEventType::PublicKeyRemoved),
                occurred_at: Set(Utc::now()),
                user_agent: Set(user_agent),
                trace_id: Set(Some(request_id.clone())),
                ip_address: Set(ip_addr),
                details: Set(Some(serde_json::json!({
                    "ssh_key_id": record.id,
                    "name": record.name,
                    "fingerprint": record.fingerprint,
                }))),
                ..Default::default()
            }
            .insert(&transaction)
            .await?;

            Ok(())
        }
        .await;

        match result {
            Ok(()) => transaction.commit().await?,
            Err(error) => {
                let _ = transaction.rollback().await;
                return Err(error);
            }
        }

        self.emit_labeled(
            label::AUTH_SSH_KEY_DELETE,
            Labels::new()
                .with(label::USER_ID, user_id)
                .with(label::REQUEST_ID, &request_id),
            serde_json::json!({
                "user_id": user_id,
                "ssh_key_id": record.id,
            }),
            Some(request_id.as_str()),
        )
        .await;

        Ok(())
    }
}