072c704
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)
}
}