//! 内部 RPC 的 event 域：EventIngress + Subscription，方法挂在 `EventService` 上。

use internal::Gate;
use internal::server::InternalRouter;
use proto::internal::v1::event_ingress_service_server::{
    EventIngressService as EventIngressTrait, EventIngressServiceServer,
};
use proto::internal::v1::subscription_service_server::{
    SubscriptionService as SubscriptionTrait, SubscriptionServiceServer,
};
use proto::internal::v1::{
    EventFilter, ListSubscriptionsRequest, ListSubscriptionsResponse, PublishRequest,
    PublishResponse, RevokeRequest, RevokeResponse, SubscribeRequest, SubscribeResponse,
    SubscriptionView, UnsubscribeRequest, UnsubscribeResponse,
};
use tonic::{Request, Response, Status};

use crate::gateway::action::EventProperties;
use crate::gateway::event_service::EventService;
use crate::gateway::subscriptions::SubscriptionRequest;

pub fn register_ingress(
    router: InternalRouter,
    service: &EventService,
    gate: Gate,
) -> InternalRouter {
    router.add_service(EventIngressServiceServer::with_interceptor(
        IngressRpc {
            service: service.clone(),
        },
        gate,
    ))
}

pub fn register(router: InternalRouter, service: &EventService, gate: Gate) -> InternalRouter {
    router.add_service(SubscriptionServiceServer::with_interceptor(
        SubscriptionRpc {
            service: service.clone(),
        },
        gate,
    ))
}

#[derive(Clone)]
struct IngressRpc {
    service: EventService,
}

#[derive(Clone)]
struct SubscriptionRpc {
    service: EventService,
}

#[tonic::async_trait]
impl EventIngressTrait for IngressRpc {
    async fn publish(
        &self,
        request: Request<PublishRequest>,
    ) -> Result<Response<PublishResponse>, Status> {
        let meta = internal::interceptor::meta(&request)?;
        let request = request.into_inner();
        let Some(envelope) = request.identity.as_ref() else {
            return Err(Status::unauthenticated("identity envelope is required"));
        };
        let identity = self.service.verify_identity(envelope).await?;
        let body = if request.body.is_empty() {
            serde_json::Value::Null
        } else {
            serde_json::from_slice(&request.body).unwrap_or(serde_json::Value::Null)
        };
        let properties = filter_to_properties(request.filter);
        let delivered = self
            .service
            .publish(
                &identity,
                &request.topic,
                &request.tag,
                properties,
                body,
                request.personal,
                meta.request_id.as_str(),
            )
            .await?;
        Ok(Response::new(PublishResponse {
            delivered,
            scope: String::new(),
        }))
    }

    async fn revoke(
        &self,
        request: Request<RevokeRequest>,
    ) -> Result<Response<RevokeResponse>, Status> {
        let request = request.into_inner();
        let Some(envelope) = request.identity.as_ref() else {
            return Err(Status::unauthenticated("identity envelope is required"));
        };
        let identity = self.service.verify_identity(envelope).await?;
        let closed = self.service.revoke(&identity, &request.reason).await?;
        Ok(Response::new(RevokeResponse { closed }))
    }
}

#[tonic::async_trait]
impl SubscriptionTrait for SubscriptionRpc {
    async fn subscribe(
        &self,
        request: Request<SubscribeRequest>,
    ) -> Result<Response<SubscribeResponse>, Status> {
        let request = request.into_inner();
        let Some(envelope) = request.identity.as_ref() else {
            return Err(Status::unauthenticated("identity envelope is required"));
        };
        let identity = self.service.verify_identity(envelope).await?;
        let filter = filter_to_properties(request.filter);
        let id = self
            .service
            .subscribe(
                identity.user_id,
                &request.device_id,
                &request.namespace,
                SubscriptionRequest {
                    topic: request.topic,
                    tag: request.tag,
                    id: if request.id.is_empty() {
                        None
                    } else {
                        Some(request.id)
                    },
                    filter: filter.clone(),
                    permission_path: if request.permission_path.is_empty() {
                        None
                    } else {
                        Some(request.permission_path)
                    },
                },
            )
            .await
            .map_err(|error| Status::invalid_argument(error))?;
        Ok(Response::new(SubscribeResponse {
            id,
            scope: scope_label(&filter),
            created_at_unix_ms: 0,
        }))
    }

    async fn unsubscribe(
        &self,
        request: Request<UnsubscribeRequest>,
    ) -> Result<Response<UnsubscribeResponse>, Status> {
        let request = request.into_inner();
        let Some(envelope) = request.identity.as_ref() else {
            return Err(Status::unauthenticated("identity envelope is required"));
        };
        let identity = self.service.verify_identity(envelope).await?;
        self.service
            .unsubscribe(
                identity.user_id,
                &request.device_id,
                &request.namespace,
                &request.topic,
                &request.tag,
                &request.id,
            )
            .await
            .map_err(|error| Status::invalid_argument(error))?;
        Ok(Response::new(UnsubscribeResponse { removed: true }))
    }

    async fn list_subscriptions(
        &self,
        request: Request<ListSubscriptionsRequest>,
    ) -> Result<Response<ListSubscriptionsResponse>, Status> {
        let request = request.into_inner();
        let Some(envelope) = request.identity.as_ref() else {
            return Err(Status::unauthenticated("identity envelope is required"));
        };
        let identity = self.service.verify_identity(envelope).await?;
        let subscriptions = self
            .service
            .list(identity.user_id, &request.device_id, "/")
            .await
            .map_err(|error| Status::unavailable(error))?;
        let views = subscriptions
            .into_iter()
            .map(|subscription| SubscriptionView {
                id: subscription.id,
                topic: subscription.topic,
                tag: subscription.tag,
                filter: properties_to_filter(&subscription.filter),
                scope: subscription.scope,
                device_id: subscription.device_id,
                namespace: subscription.namespace,
            })
            .collect();
        Ok(Response::new(ListSubscriptionsResponse {
            subscriptions: views,
        }))
    }
}

fn scope_label(filter: &EventProperties) -> String {
    [
        ("room", &filter.room_id),
        ("issue", &filter.issue_id),
        ("pull", &filter.pull_id),
        ("repo", &filter.repo_id),
        ("org", &filter.org_id),
        ("user", &filter.user_id),
    ]
    .into_iter()
    .find_map(|(kind, id)| id.as_ref().map(|id| format!("{kind}:{id}")))
    .unwrap_or_default()
}

fn filter_to_properties(filter: Option<EventFilter>) -> EventProperties {
    let filter = filter.unwrap_or_default();
    EventProperties {
        user_id: Some(filter.user_id).filter(|value| !value.is_empty()),
        org_id: Some(filter.org_id).filter(|value| !value.is_empty()),
        request_id: Some(filter.request_id).filter(|value| !value.is_empty()),
        repo_id: Some(filter.repo_id).filter(|value| !value.is_empty()),
        repo_name: Some(filter.repo_name).filter(|value| !value.is_empty()),
        room_id: Some(filter.room_id).filter(|value| !value.is_empty()),
        message_id: Some(filter.message_id).filter(|value| !value.is_empty()),
        thread_id: Some(filter.thread_id).filter(|value| !value.is_empty()),
        issue_id: Some(filter.issue_id).filter(|value| !value.is_empty()),
        pull_id: Some(filter.pull_id).filter(|value| !value.is_empty()),
        timestamp: (filter.timestamp != 0).then_some(filter.timestamp),
    }
}

fn properties_to_filter(properties: &EventProperties) -> Option<EventFilter> {
    Some(EventFilter {
        user_id: properties.user_id.clone().unwrap_or_default(),
        org_id: properties.org_id.clone().unwrap_or_default(),
        request_id: properties.request_id.clone().unwrap_or_default(),
        repo_id: properties.repo_id.clone().unwrap_or_default(),
        repo_name: properties.repo_name.clone().unwrap_or_default(),
        room_id: properties.room_id.clone().unwrap_or_default(),
        message_id: properties.message_id.clone().unwrap_or_default(),
        thread_id: properties.thread_id.clone().unwrap_or_default(),
        issue_id: properties.issue_id.clone().unwrap_or_default(),
        pull_id: properties.pull_id.clone().unwrap_or_default(),
        timestamp: properties.timestamp.unwrap_or_default(),
    })
}