use futures_util::TryStreamExt;
use rocketmq_model::common::message::MessageTrait;
use rocketmq_model::common::message::message_ext::MessageExt;
use serde_json::Value;

use super::{EventProperties, gitdata, kenshell, platform_events};
use crate::gateway::platform::PlatformClient;
use crate::gateway::{ConsumeResult, Subscription, Subscriptions};
use actor::ActorHost;
use std::collections::HashMap;

pub struct EventContext<'a> {
    pub host: &'a ActorHost,
    pub platform: &'a PlatformClient,
    pub subscriptions: &'a Subscriptions,
}

pub async fn action(message: MessageExt, context: &EventContext<'_>) -> ConsumeResult {
    let topic = message.topic().to_string();
    let tag = message
        .tags_ref()
        .map(ToString::to_string)
        .unwrap_or_default();
    let properties = EventProperties::from_message(&message);

    match topic.split_once('-').map(|(prefix, _)| prefix) {
        Some("gitdata") if topic == "gitdata-auth" || topic == "gitdata-org" => {
            platform_events::route(&topic, &tag, message, properties, context).await
        }
        Some("kenshell") => kenshell::route(&topic, &tag, message, properties, context).await,
        Some("gitdata") => gitdata::route(&topic, &tag, message, properties, context).await,
        _ => unimplemented(message),
    }
}

pub(super) async fn deliver_to_subscribers(
    topic: &str,
    tag: &str,
    message: MessageExt,
    properties: EventProperties,
    context: &EventContext<'_>,
) -> ConsumeResult {
    let event_user_id = user_id(&properties);
    let body = body(&message);
    let subscribers = match context
        .subscriptions
        .subscribers(topic, tag, &properties)
        .await
    {
        Ok(subscribers) => subscribers,
        Err(_) => return retry(message),
    };
    let mut recipients: HashMap<i64, Vec<Subscription>> = HashMap::new();
    for subscription in subscribers {
        if subscription.filter.matches(&properties) {
            recipients
                .entry(subscription.user_id)
                .or_default()
                .push(subscription);
        }
    }
    let properties = &properties;
    let body = &body;
    let delivered = futures_util::stream::iter(recipients.into_iter().map(Ok::<_, ()>))
        .try_for_each_concurrent(32, |(recipient, destinations)| async move {
            let mut value = payload(topic, tag, event_user_id, recipient, properties, body);
            value["destinations"] = serde_json::to_value(destinations).map_err(|_| ())?;
            context
                .host
                .publish_event(recipient, value)
                .await
                .map_err(|_| ())
        })
        .await;
    if delivered.is_err() {
        return retry(message);
    }
    ConsumeResult::Ack { message }
}

pub(super) fn payload(
    topic: &str,
    tag: &str,
    event_user_id: Option<i64>,
    recipient_user_id: i64,
    properties: &EventProperties,
    body: &Value,
) -> Value {
    serde_json::json!({
        "topic": topic,
        "tag": tag,
        "user_id": event_user_id,
        "recipient_user_id": recipient_user_id,
        "properties": properties,
        "body": body,
    })
}

pub(super) fn user_id(properties: &EventProperties) -> Option<i64> {
    properties
        .user_id
        .as_deref()?
        .parse::<i64>()
        .ok()
        .filter(|id| *id > 0)
}

pub(super) fn body(message: &MessageExt) -> Value {
    message
        .body()
        .and_then(|bytes| serde_json::from_slice(&bytes).ok())
        .unwrap_or(Value::Null)
}

pub(super) fn retry(message: MessageExt) -> ConsumeResult {
    ConsumeResult::Retry {
        message,
        delay_level: 0,
    }
}

pub(super) fn unimplemented(message: MessageExt) -> ConsumeResult {
    retry(message)
}