use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::gateway::action::EventProperties;
use crate::gateway::plan;
use crate::gateway::platform::PlatformClient;
use actor::{ActorError, RedisConnection, connect_redis_single};
use config::event::consumer::DEFAULT_TOPIC_PREFIXES;
use config::event::redis::RedisConfig;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Subscription {
    pub id: String,
    pub user_id: i64,
    pub device_id: String,
    pub namespace: String,
    pub topic: String,
    pub tag: String,
    pub filter: EventProperties,
    pub permission_path: Option<String>,
    pub scope: String,
}

#[derive(Clone)]
pub struct Subscriptions {
    connection: RedisConnection,
}

impl Subscriptions {
    pub(crate) fn new(connection: RedisConnection) -> Self {
        Self { connection }
    }

    /// Reuses an already established pool instead of opening another one.
    pub fn from_connection(connection: RedisConnection) -> Self {
        Self { connection }
    }

    pub async fn connect(config: &RedisConfig) -> Result<Self, ActorError> {
        Ok(Self::new(connect_redis_single(config).await?))
    }

    fn validate(topic: &str, tag: &str) -> Result<(), String> {
        if !crate::gateway::action::is_supported(topic, tag) {
            return Err("unsupported subscription topic or tag".to_string());
        }
        Ok(())
    }

    pub async fn subscribe(
        &self,
        platform: &PlatformClient,
        user_id: i64,
        device_id: &str,
        namespace: &str,
        request: SubscriptionRequest,
    ) -> Result<String, String> {
        Self::validate(&request.topic, &request.tag)?;
        if user_id <= 0 || device_id.is_empty() {
            return Err("authenticated device is required".to_string());
        }
        let mut filter = request.filter;
        filter.normalize()?;
        let personal = is_personal(&request.topic, &request.tag);
        if personal {
            if filter
                .user_id
                .as_deref()
                .is_some_and(|id| id != user_id.to_string())
            {
                return Err("personal notifications must target the authenticated user".to_string());
            }
            filter.user_id = Some(user_id.to_string());
        }
        let path = resource_path(&filter, request.permission_path.as_deref())?;
        let has_resource = filter.room_id.is_some()
            || filter.issue_id.is_some()
            || filter.pull_id.is_some()
            || filter.repo_id.is_some();
        if has_resource {
            if !platform
                .authorize(user_id, &filter, "SUBSCRIBE")
                .await
                .map_err(|error| error.to_string())?
            {
                return Err("resource subscription is not authorized".to_string());
            }
        } else if filter.user_id.as_deref() != Some(user_id.to_string().as_str()) {
            return Err(
                "resource authorization is required for cross-user subscriptions".to_string(),
            );
        }
        if let Some(org) = filter.org_id.as_deref() {
            if !platform
                .is_member(user_id, org)
                .await
                .map_err(|error| error.to_string())?
            {
                return Err("active organization membership is required".to_string());
            }
        }
        let id = request
            .id
            .as_deref()
            .map(Uuid::parse_str)
            .transpose()
            .map_err(|_| "invalid subscription id")?
            .unwrap_or_else(Uuid::new_v4)
            .to_string();
        let scope = if personal {
            format!("user:{user_id}")
        } else {
            scopes(&filter)
                .into_iter()
                .next()
                .ok_or("a resource or user filter is required")?
        };
        let subscription = Subscription {
            id: id.clone(),
            user_id,
            device_id: device_id.to_string(),
            namespace: namespace.to_string(),
            topic: request.topic,
            tag: request.tag,
            filter,
            permission_path: path,
            scope,
        };
        let base = base(&subscription.topic, &subscription.tag);
        let owner = owner(user_id, device_id, namespace);
        let field = format!("{owner}:{id}");
        let mut command = redis::cmd("EVAL");
        command.arg(r#"
local old=redis.call('HGET',KEYS[1],ARGV[1])
if not old and redis.call('SCARD',KEYS[3]) >= 256 then return redis.error_reply('device subscription limit exceeded') end
if old then redis.call('SREM',ARGV[3]..':scope:'..cjson.decode(old).scope,ARGV[1]) end
redis.call('HSET',KEYS[1],ARGV[1],ARGV[2])
redis.call('SADD',KEYS[2],ARGV[1])
redis.call('SADD',KEYS[3],ARGV[1])
return 1
"#).arg(3).arg(format!("{base}:records")).arg(format!("{base}:scope:{}",subscription.scope))
            .arg(format!("{base}:owner:{owner}")).arg(field)
            .arg(serde_json::to_string(&subscription).map_err(|error| error.to_string())?).arg(base);
        let _: i64 = self
            .connection
            .query(&command)
            .await
            .map_err(|error| error.to_string())?;
        Ok(id)
    }

    pub async fn unsubscribe(
        &self,
        user_id: i64,
        device_id: &str,
        namespace: &str,
        topic: &str,
        tag: &str,
        id: &str,
    ) -> Result<(), String> {
        Self::validate(topic, tag)?;
        let id = Uuid::parse_str(id)
            .map_err(|_| "invalid subscription id")?
            .to_string();
        let base = base(topic, tag);
        let owner = owner(user_id, device_id, namespace);
        let mut command = redis::cmd("EVAL");
        command.arg("local old=redis.call('HGET',KEYS[1],ARGV[1]); if old then redis.call('SREM',ARGV[2]..':scope:'..cjson.decode(old).scope,ARGV[1]); redis.call('HDEL',KEYS[1],ARGV[1]); redis.call('SREM',KEYS[2],ARGV[1]) end; return 1")
            .arg(2).arg(format!("{base}:records")).arg(format!("{base}:owner:{owner}"))
            .arg(format!("{owner}:{id}")).arg(base);
        let _: i64 = self
            .connection
            .query(&command)
            .await
            .map_err(|error| error.to_string())?;
        Ok(())
    }

    pub async fn contains(&self, subscription: &Subscription) -> Result<bool, String> {
        let mut command = redis::cmd("HGET");
        command
            .arg(format!(
                "{}:records",
                base(&subscription.topic, &subscription.tag)
            ))
            .arg(format!(
                "{}:{}",
                owner(
                    subscription.user_id,
                    &subscription.device_id,
                    &subscription.namespace
                ),
                subscription.id
            ));
        let current: Option<String> = self
            .connection
            .query(&command)
            .await
            .map_err(|error| error.to_string())?;
        let expected = serde_json::to_string(subscription).map_err(|error| error.to_string())?;
        Ok(current.as_deref() == Some(expected.as_str()))
    }

    pub async fn subscribers(
        &self,
        topic: &str,
        tag: &str,
        event: &EventProperties,
    ) -> Result<Vec<Subscription>, String> {
        let base = base(topic, tag);
        let mut command = redis::cmd("EVAL");
        command.arg(
            r#"
local seen={}; local result={}
for i=2,#KEYS do
    for _,field in ipairs(redis.call('SMEMBERS',KEYS[i])) do
        if not seen[field] then
            seen[field]=true
            local record=redis.call('HGET',KEYS[1],field)
            if record then table.insert(result,record) end
        end
    end
end
return result
"#,
        );
        let scopes = scopes(event);
        command.arg(scopes.len() + 1).arg(format!("{base}:records"));
        for scope in scopes {
            command.arg(format!("{base}:scope:{scope}"));
        }
        let records: Vec<String> = self
            .connection
            .query(&command)
            .await
            .map_err(|error| error.to_string())?;
        records
            .into_iter()
            .map(|record| serde_json::from_str(&record).map_err(|error| error.to_string()))
            .collect()
    }
}

#[derive(Debug, Deserialize)]
pub struct SubscriptionRequest {
    pub topic: String,
    #[serde(alias = "label")]
    pub tag: String,
    pub id: Option<String>,
    #[serde(default)]
    pub filter: EventProperties,
    pub permission_path: Option<String>,
}

pub fn is_personal(topic: &str, tag: &str) -> bool {
    crate::gateway::action::split_topic(topic)
        .is_some_and(|(_, area)| crate::gateway::action::personal(area, tag))
}

fn base(topic: &str, tag: &str) -> String {
    format!("kenevent:subscription:v2:{{{topic}:{tag}}}")
}

fn owner(user_id: i64, device: &str, namespace: &str) -> String {
    format!(
        "{user_id}:{}",
        URL_SAFE_NO_PAD.encode(serde_json::to_vec(&(device, namespace)).unwrap_or_default())
    )
}

fn scopes(properties: &EventProperties) -> Vec<String> {
    [
        ("room", &properties.room_id),
        ("issue", &properties.issue_id),
        ("pull", &properties.pull_id),
        ("repo", &properties.repo_id),
        ("org", &properties.org_id),
        ("user", &properties.user_id),
    ]
    .into_iter()
    .filter_map(|(kind, id)| id.as_ref().map(|id| format!("{kind}:{id}")))
    .collect()
}

/// 权限路径只能是"与 filter 指向同一资源"的绝对路径：防止借路径白名单越权。
pub fn resource_path(
    filter: &EventProperties,
    path: Option<&str>,
) -> Result<Option<String>, String> {
    let has_resource = filter.room_id.is_some()
        || filter.repo_id.is_some()
        || filter.issue_id.is_some()
        || filter.pull_id.is_some()
        || filter.repo_name.is_some();
    if !has_resource {
        return if path.is_none() {
            Ok(None)
        } else {
            Err("resource path requires a resource filter".to_string())
        };
    }
    let path = match (path, filter.room_id.as_deref()) {
        (Some(path), _) => path.to_string(),
        (None, Some(room)) => format!("/chat/rooms/{room}"),
        _ => return Err("permission_path is required for this resource".to_string()),
    };
    let segments = path
        .strip_prefix('/')
        .ok_or("permission_path must be absolute")?
        .split('/')
        .collect::<Vec<_>>();
    if segments.iter().any(|part| {
        part.is_empty()
            || *part == "."
            || *part == ".."
            || !part
                .bytes()
                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
    }) {
        return Err("invalid resource permission path".to_string());
    }
    let equal_id = |value: &str, expected: &Option<String>| {
        expected.as_deref().is_some_and(|expected| {
            Uuid::parse_str(value)
                .ok()
                .zip(Uuid::parse_str(expected).ok())
                .is_some_and(|(a, b)| a == b)
        })
    };
    let valid = match segments.as_slice() {
        ["chat", "rooms", id] => {
            equal_id(id, &filter.room_id)
                && filter.repo_id.is_none()
                && filter.issue_id.is_none()
                && filter.pull_id.is_none()
        }
        ["repos", _, _, "pulls", id] => {
            equal_id(id, &filter.pull_id) && filter.room_id.is_none() && filter.issue_id.is_none()
        }
        ["issues", _, "boards", board, "issues", id] => {
            Uuid::parse_str(board).is_ok()
                && equal_id(id, &filter.issue_id)
                && filter.room_id.is_none()
                && filter.pull_id.is_none()
        }
        ["repos", space, repo] => {
            filter
                .repo_id
                .as_deref()
                .is_some_and(|id| Uuid::parse_str(id).is_ok())
                && filter
                    .repo_name
                    .as_deref()
                    .is_none_or(|name| name == *repo || name == format!("{space}/{repo}"))
                && filter.room_id.is_none()
                && filter.issue_id.is_none()
                && filter.pull_id.is_none()
        }
        _ => false,
    };
    if !valid {
        return Err("permission_path does not identify the subscribed resource".to_string());
    }
    Ok(Some(path))
}

impl Subscriptions {
    /// 列出某设备/命名空间下的全部订阅（RPC ListSubscriptions 用）。
    pub async fn list(
        &self,
        user_id: i64,
        device_id: &str,
        namespace: &str,
    ) -> Result<Vec<Subscription>, String> {
        let prefixes = DEFAULT_TOPIC_PREFIXES.map(str::to_string);
        let owner = owner(user_id, device_id, namespace);
        let mut out = Vec::new();
        for (topic, tags) in plan::plan(&prefixes) {
            for tag in tags.split(" || ") {
                let base = base(&topic, tag);
                let mut command = redis::cmd("SMEMBERS");
                command.arg(format!("{base}:owner:{owner}"));
                let fields: Vec<String> = self
                    .connection
                    .query(&command)
                    .await
                    .map_err(|error| error.to_string())?;
                for field in fields {
                    let mut command = redis::cmd("HGET");
                    command.arg(format!("{base}:records")).arg(field);
                    let record: Option<String> = self
                        .connection
                        .query(&command)
                        .await
                        .map_err(|error| error.to_string())?;
                    if let Some(record) = record {
                        if let Ok(subscription) = serde_json::from_str::<Subscription>(&record) {
                            out.push(subscription);
                        }
                    }
                }
            }
        }
        Ok(out)
    }
}