use std::sync::Arc;

use actor::ActorHost;
use internal::envelope::{EnvelopeVerifier, VerifiedIdentity};
use socket::SocketHub;

use crate::gateway::action::EventProperties;
use crate::gateway::platform::PlatformClient;
use crate::gateway::subscriptions::{Subscription, SubscriptionRequest, Subscriptions};
use crate::gateway::{ConsumeResult, is_personal};

/// 事件域唯一业务入口：socket 路径与内部 RPC 路径共用同一套订阅/授权/投递逻辑。
#[derive(Clone)]
pub struct EventService {
    host: ActorHost,
    platform: PlatformClient,
    subscriptions: Subscriptions,
    verifier: Arc<EnvelopeVerifier>,
    hub: SocketHub,
}

impl EventService {
    pub fn new(
        host: ActorHost,
        platform: PlatformClient,
        subscriptions: Subscriptions,
        verifier: Arc<EnvelopeVerifier>,
        hub: SocketHub,
    ) -> Self {
        Self {
            host,
            platform,
            subscriptions,
            verifier,
            hub,
        }
    }

    pub fn host(&self) -> &ActorHost {
        &self.host
    }

    pub fn platform(&self) -> &PlatformClient {
        &self.platform
    }

    pub fn subscriptions(&self) -> &Subscriptions {
        &self.subscriptions
    }

    /// 解密身份信封 + 平台权威复核。
    pub async fn verify_identity(
        &self,
        envelope: &proto::internal::v1::IdentityEnvelope,
    ) -> Result<VerifiedIdentity, tonic::Status> {
        let identity = self.verifier.verify(envelope).await?;
        self.platform
            .attest(
                identity.user_id,
                identity.security_version,
                &identity.request_id,
            )
            .await
            .map_err(|error| error.into_status())?;
        Ok(identity)
    }

    /// 订阅（浏览器 socket 与内部 RPC 共用）。
    pub async fn subscribe(
        &self,
        user_id: i64,
        device_id: &str,
        namespace: &str,
        request: SubscriptionRequest,
    ) -> Result<String, String> {
        self.subscriptions
            .subscribe(&self.platform, user_id, device_id, namespace, request)
            .await
    }

    pub async fn unsubscribe(
        &self,
        user_id: i64,
        device_id: &str,
        namespace: &str,
        topic: &str,
        tag: &str,
        id: &str,
    ) -> Result<(), String> {
        self.subscriptions
            .unsubscribe(user_id, device_id, namespace, topic, tag, id)
            .await
    }

    pub async fn list(
        &self,
        user_id: i64,
        device_id: &str,
        namespace: &str,
    ) -> Result<Vec<Subscription>, String> {
        self.subscriptions.list(user_id, device_id, namespace).await
    }

    /// 平台定向投递：身份已校验，需再过订阅匹配 + 投递期授权。
    pub async fn publish(
        &self,
        identity: &VerifiedIdentity,
        topic: &str,
        tag: &str,
        properties: EventProperties,
        body: serde_json::Value,
        personal: bool,
        request_id: &str,
    ) -> Result<u32, tonic::Status> {
        if !crate::gateway::action::is_supported(topic, tag) {
            return Err(tonic::Status::invalid_argument("unsupported topic or tag"));
        }
        let personal = personal || is_personal(topic, tag);
        let recipient = identity.user_id;
        if personal {
            if properties
                .user_id
                .as_deref()
                .is_some_and(|value| value != recipient.to_string())
            {
                return Err(tonic::Status::permission_denied(
                    "personal events must target the identity user",
                ));
            }
            let payload = crate::gateway::action::common_payload(
                topic,
                tag,
                Some(recipient),
                recipient,
                &properties,
                &body,
                None,
            );
            self.host
                .publish_event(recipient, payload)
                .await
                .map_err(|_| tonic::Status::unavailable("delivery store is unavailable"))?;
            return Ok(1);
        }
        let subscribers = self
            .subscriptions
            .subscribers(topic, tag, &properties)
            .await
            .map_err(|_| tonic::Status::failed_precondition("subscription store is unavailable"))?;
        let mut delivered = 0;
        for subscription in subscribers
            .into_iter()
            .filter(|subscription| subscription.user_id == recipient)
            .filter(|subscription| subscription.filter.matches(&properties))
        {
            let allowed = self
                .platform
                .authorize(recipient, &subscription.filter, "DELIVER")
                .await
                .map_err(|error| error.into_status())?;
            if !allowed {
                continue;
            }
            if let Some(org) = subscription.filter.org_id.as_deref() {
                if !self
                    .platform
                    .is_member(recipient, org)
                    .await
                    .map_err(|error| error.into_status())?
                {
                    continue;
                }
            }
            let payload = crate::gateway::action::common_payload(
                topic,
                tag,
                properties
                    .user_id
                    .as_deref()
                    .and_then(|value| value.parse::<i64>().ok()),
                recipient,
                &properties,
                &body,
                Some(vec![subscription.clone()]),
            );
            if self.host.publish_event(recipient, payload).await.is_ok() {
                delivered += 1;
            }
        }
        let _ = request_id;
        Ok(delivered)
    }

    /// 撤销连接/权限：发控制事件并关闭该用户在线会话。
    pub async fn revoke(
        &self,
        identity: &VerifiedIdentity,
        reason: &str,
    ) -> Result<u32, tonic::Status> {
        let payload = serde_json::json!({
            "topic": "event",
            "tag": "access-revoked",
            "control": true,
            "event_id": uuid::Uuid::now_v7().to_string(),
            "recipient_user_id": identity.user_id,
            "properties": {"user_id": identity.user_id},
            "body": {"reason": reason},
        });
        let delivered = self.host.publish_event(identity.user_id, payload).await;
        let closed = self.hub.close_user(identity.user_id).await as u32;
        delivered.map_err(|_| tonic::Status::unavailable("delivery store is unavailable"))?;
        Ok(closed)
    }

    pub fn consume_result(_: ConsumeResult) {}
}