b453d5d
use std::collections::HashMap;
use models::space::space_member_groups;
use models::space::space_members::{self, MemberStatus, MessageNotifications};
use models::space::spaces;
use models::users::user;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter,
QueryOrder, QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::SpaceError;
use crate::services::service::SpaceService;
use crate::services::spaces::{PageQuery, UserDto};
use crate::share::authz::{MANAGE_MEMBERS, assert_not_banned, require_perm};
use crate::share::lookup::lookup_space_by_name;
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct MemberDto {
pub space_id: Uuid,
pub user: Option<UserDto>,
pub alias_name: Option<String>,
pub alias_avatar_url: Option<String>,
pub status: MemberStatus,
pub last_active_at: Option<OffsetDateTime>,
pub left_at: Option<OffsetDateTime>,
pub invited_at: Option<OffsetDateTime>,
pub invited_by: Option<UserDto>,
pub joined_at: OffsetDateTime,
pub created_at: OffsetDateTime,
pub updated_at: Option<OffsetDateTime>,
pub message_notifications: MessageNotifications,
pub muted: bool,
pub mute_until: Option<OffsetDateTime>,
pub mobile_push: bool,
pub suppress_everyone: bool,
pub suppress_roles: bool,
pub hide_muted_channels: bool,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct AddMemberCmd {
pub user_id: i64,
#[serde(default)]
pub alias_name: Option<String>,
#[serde(default)]
pub alias_avatar_url: Option<String>,
#[serde(default)]
pub message_notifications: Option<MessageNotifications>,
#[serde(default)]
pub muted: Option<bool>,
#[serde(
default,
deserialize_with = "export::serde_datetime::deserialize_opt_datetime"
)]
pub mute_until: Option<OffsetDateTime>,
#[serde(default)]
pub mobile_push: Option<bool>,
#[serde(default)]
pub suppress_everyone: Option<bool>,
#[serde(default)]
pub suppress_roles: Option<bool>,
#[serde(default)]
pub hide_muted_channels: Option<bool>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateMemberCmd {
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub alias_name: Option<Option<String>>,
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub alias_avatar_url: Option<Option<String>>,
#[serde(default)]
pub message_notifications: Option<MessageNotifications>,
#[serde(default)]
pub muted: Option<bool>,
#[serde(
default,
deserialize_with = "export::serde_datetime::deserialize_opt_opt_datetime"
)]
pub mute_until: Option<Option<OffsetDateTime>>,
#[serde(default)]
pub mobile_push: Option<bool>,
#[serde(default)]
pub suppress_everyone: Option<bool>,
#[serde(default)]
pub suppress_roles: Option<bool>,
#[serde(default)]
pub hide_muted_channels: Option<bool>,
}
pub type MemberPage = export::pagination::Page<MemberDto>;
fn proto_timestamp(value: OffsetDateTime) -> prost_types::Timestamp {
prost_types::Timestamp {
seconds: value.unix_timestamp(),
nanos: value.nanosecond() as i32,
}
}
fn notifications_name(value: MessageNotifications) -> String {
match value {
MessageNotifications::All => "all".to_string(),
MessageNotifications::Mentions => "mentions".to_string(),
MessageNotifications::Nothing => "nothing".to_string(),
}
}
/// Presentation/notification values a member row is created with when the
/// caller does not sit in the admin "add member" flow (join-request approval,
/// invite acceptance).
#[derive(Debug, Clone, Default)]
pub struct MemberActivationDefaults {
pub alias_name: Option<String>,
pub alias_avatar_url: Option<String>,
pub message_notifications: Option<MessageNotifications>,
pub muted: Option<bool>,
pub mute_until: Option<OffsetDateTime>,
pub mobile_push: Option<bool>,
pub suppress_everyone: Option<bool>,
pub suppress_roles: Option<bool>,
pub hide_muted_channels: Option<bool>,
}
/// Revive a left/removed member row or insert a fresh one, bumping
/// `nums_members` exactly once. Returns `false` when the row was already
/// active, so a racing caller can settle for the existing membership instead
/// of double-counting. Callers own the transaction and every permission, ban
/// and 2FA check.
pub async fn activate_member_row<C: ConnectionTrait>(
txn: &C,
space_id: Uuid,
user_id: i64,
now: OffsetDateTime,
defaults: MemberActivationDefaults,
) -> Result<bool, SpaceError> {
let existing = space_members::Entity::find_by_id((space_id, user_id))
.one(txn)
.await?;
match existing {
Some(member) if member.status == MemberStatus::Active => return Ok(false),
Some(member) => {
let mut am: space_members::ActiveModel = member.into();
am.status = Set(MemberStatus::Active);
am.left_at = Set(None);
am.updated_at = Set(Some(now));
am.update(txn).await?;
}
None => {
let member_row = space_members::ActiveModel {
space_id: Set(space_id),
user_id: Set(user_id),
alias_name: Set(defaults.alias_name),
alias_avatar_url: Set(defaults.alias_avatar_url),
status: Set(MemberStatus::Active),
last_active_at: Set(None),
left_at: Set(None),
invited_at: Set(None),
invited_by: Set(None),
joined_at: Set(now),
created_at: Set(now),
updated_at: Set(Some(now)),
message_notifications: Set(defaults
.message_notifications
.unwrap_or(MessageNotifications::All)),
muted: Set(defaults.muted.unwrap_or(false)),
mute_until: Set(defaults.mute_until),
mobile_push: Set(defaults.mobile_push.unwrap_or(false)),
suppress_everyone: Set(defaults.suppress_everyone.unwrap_or(false)),
suppress_roles: Set(defaults.suppress_roles.unwrap_or(false)),
hide_muted_channels: Set(defaults.hide_muted_channels.unwrap_or(false)),
..Default::default()
};
member_row.insert(txn).await?;
}
}
spaces::Entity::update_many()
.filter(spaces::Column::Id.eq(space_id))
.col_expr(spaces::Column::NumsMembers, Expr::cust("nums_members + 1"))
.exec(txn)
.await?;
Ok(true)
}
impl SpaceService {
/// Auto-join every active public room of a space. Room rows are advisory:
/// a failure is logged and the membership write still stands.
pub(crate) async fn grant_public_room_memberships(&self, space_id: Uuid, user_id: i64) {
let public_rooms = models::room::rooms::Entity::find()
.filter(models::room::rooms::Column::SpaceId.eq(space_id))
.filter(models::room::rooms::Column::IsPrivate.eq(false))
.filter(models::room::rooms::Column::IsActive.eq(true))
.all(self.db().read())
.await;
let Ok(public_rooms) = public_rooms else {
tracing::warn!(%space_id, %user_id, "public room lookup failed");
return;
};
for room in &public_rooms {
let now = OffsetDateTime::now_utc();
let rm = models::room::room_members::ActiveModel {
role: Set(models::room::room_members::RoomMemberRole::Member),
room_id: Set(room.id),
user_id: Set(user_id),
nick: Set(None),
joined_at: Set(now),
left_at: Set(None),
created_at: Set(now),
updated_at: Set(Some(now)),
};
if let Err(e) = rm.insert(self.db().write()).await {
tracing::warn!(error = %e, room_id = %room.id, %user_id, "failed to create room_members record");
}
}
}
#[tracing::instrument(name = "space.add_member", skip_all, fields(caller = %caller, space = %space_name))]
pub async fn add_member(
&self,
caller: i64,
space_name: &str,
cmd: AddMemberCmd,
) -> Result<MemberDto, SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
let target_user = user::Entity::find_by_id(cmd.user_id)
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
assert_not_banned(self.db().read(), space.id, cmd.user_id).await?;
if space.member_must_enable_2fa && !target_user.totp_enabled {
return Err(SpaceError::Forbidden);
}
if let Some(alias) = &cmd.alias_name
&& alias.chars().count() > self.config().space.max_alias_name_len
{
return Err(SpaceError::BadRequest("alias name is too long".to_string()));
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
let (member, created) = client
.add_member(caller, &space.id.to_string(), cmd.user_id, 0, Vec::new())
.await?;
if !created {
return Err(SpaceError::Conflict(
"user is already a member of this space".to_string(),
));
}
let now = OffsetDateTime::now_utc();
let mut projected = crate::sync::upsert_member(self.db().write(), space.id, &member).await?;
let mut local: space_members::ActiveModel = projected.clone().into();
if let Some(value) = cmd.alias_name {
local.alias_name = Set(Some(value));
}
if let Some(value) = cmd.alias_avatar_url {
local.alias_avatar_url = Set(Some(value));
}
if let Some(value) = cmd.message_notifications {
local.message_notifications = Set(value);
}
if let Some(value) = cmd.muted {
local.muted = Set(value);
}
if let Some(value) = cmd.mute_until {
local.mute_until = Set(Some(value));
}
if let Some(value) = cmd.mobile_push {
local.mobile_push = Set(value);
}
if let Some(value) = cmd.suppress_everyone {
local.suppress_everyone = Set(value);
}
if let Some(value) = cmd.suppress_roles {
local.suppress_roles = Set(value);
}
if let Some(value) = cmd.hide_muted_channels {
local.hide_muted_channels = Set(value);
}
local.updated_at = Set(Some(now));
projected = local.update(self.db().write()).await?;
crate::sync::bump_members(self.db().write(), space.id, 1).await?;
self.grant_public_room_memberships(space.id, cmd.user_id)
.await;
let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&[cmd.user_id]).await?;
let user_brief = user_map.get(&cmd.user_id).cloned();
Ok(MemberDto {
space_id: projected.space_id,
user: user_brief,
alias_name: projected.alias_name,
alias_avatar_url: projected.alias_avatar_url,
status: projected.status,
last_active_at: projected.last_active_at,
left_at: projected.left_at,
invited_at: projected.invited_at,
invited_by: None,
joined_at: projected.joined_at,
created_at: projected.created_at,
updated_at: projected.updated_at,
message_notifications: projected.message_notifications,
muted: projected.muted,
mute_until: projected.mute_until,
mobile_push: projected.mobile_push,
suppress_everyone: projected.suppress_everyone,
suppress_roles: projected.suppress_roles,
hide_muted_channels: projected.hide_muted_channels,
})
}
pub async fn get_member(
&self,
caller: i64,
space_name: &str,
uid: i64,
) -> Result<MemberDto, SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
if caller != uid {
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
}
let member = space_members::Entity::find_by_id((space.id, uid))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
let mut uids = vec![member.user_id];
if let Some(inv_by) = member.invited_by {
uids.push(inv_by);
}
let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&uids).await?;
let user_brief = user_map.get(&member.user_id).cloned();
let inviter_brief = member.invited_by.and_then(|ib| user_map.get(&ib).cloned());
Ok(MemberDto {
space_id: member.space_id,
user: user_brief,
alias_name: member.alias_name,
alias_avatar_url: member.alias_avatar_url,
status: member.status,
last_active_at: member.last_active_at,
left_at: member.left_at,
invited_at: member.invited_at,
invited_by: inviter_brief,
joined_at: member.joined_at,
created_at: member.created_at,
updated_at: member.updated_at,
message_notifications: member.message_notifications,
muted: member.muted,
mute_until: member.mute_until,
mobile_push: member.mobile_push,
suppress_everyone: member.suppress_everyone,
suppress_roles: member.suppress_roles,
hide_muted_channels: member.hide_muted_channels,
})
}
#[tracing::instrument(name = "space.update_member", skip_all, fields(caller = %caller, space = %space_name, uid = %uid))]
pub async fn update_member(
&self,
caller: i64,
space_name: &str,
uid: i64,
cmd: UpdateMemberCmd,
) -> Result<MemberDto, SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
if caller != uid {
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
}
let member = space_members::Entity::find_by_id((space.id, uid))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if let Some(alias) = &cmd.alias_name
&& let Some(value) = alias
&& value.chars().count() > self.config().space.max_alias_name_len
{
return Err(SpaceError::BadRequest("alias name is too long".to_string()));
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
client
.update_member(
caller,
&space.id.to_string(),
uid,
cmd.alias_name.clone(),
cmd.alias_avatar_url.clone(),
cmd.message_notifications
.map(notifications_name),
cmd.muted,
cmd.mute_until
.map(|value| value.map(proto_timestamp)),
cmd.mobile_push,
cmd.suppress_everyone,
cmd.suppress_roles,
cmd.hide_muted_channels,
)
.await?;
let mut am: space_members::ActiveModel = member.into();
if let Some(alias) = cmd.alias_name {
if let Some(a) = &alias
&& a.chars().count() > self.config().space.max_alias_name_len
{
return Err(SpaceError::BadRequest("alias name is too long".to_string()));
}
am.alias_name = Set(alias);
}
if let Some(url) = cmd.alias_avatar_url {
am.alias_avatar_url = Set(url);
}
if let Some(mn) = cmd.message_notifications {
am.message_notifications = Set(mn);
}
if let Some(muted) = cmd.muted {
am.muted = Set(muted);
}
if let Some(mute_until) = cmd.mute_until {
am.mute_until = Set(mute_until);
}
if let Some(mp) = cmd.mobile_push {
am.mobile_push = Set(mp);
}
if let Some(se) = cmd.suppress_everyone {
am.suppress_everyone = Set(se);
}
if let Some(sr) = cmd.suppress_roles {
am.suppress_roles = Set(sr);
}
if let Some(hmc) = cmd.hide_muted_channels {
am.hide_muted_channels = Set(hmc);
}
let now = OffsetDateTime::now_utc();
am.updated_at = Set(Some(now));
let updated = am.update(self.db().write()).await?;
let mut uids = vec![updated.user_id];
if let Some(inv_by) = updated.invited_by {
uids.push(inv_by);
}
let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&uids).await?;
let user_brief = user_map.get(&updated.user_id).cloned();
let inviter_brief = updated.invited_by.and_then(|ib| user_map.get(&ib).cloned());
Ok(MemberDto {
space_id: updated.space_id,
user: user_brief,
alias_name: updated.alias_name,
alias_avatar_url: updated.alias_avatar_url,
status: updated.status,
last_active_at: updated.last_active_at,
left_at: updated.left_at,
invited_at: updated.invited_at,
invited_by: inviter_brief,
joined_at: updated.joined_at,
created_at: updated.created_at,
updated_at: updated.updated_at,
message_notifications: updated.message_notifications,
muted: updated.muted,
mute_until: updated.mute_until,
mobile_push: updated.mobile_push,
suppress_everyone: updated.suppress_everyone,
suppress_roles: updated.suppress_roles,
hide_muted_channels: updated.hide_muted_channels,
})
}
#[tracing::instrument(name = "space.leave_space", skip_all, fields(caller = %caller, space = %space_name))]
pub async fn leave_space(&self, caller: i64, space_name: &str) -> Result<(), SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
if space.owner_id == caller {
return Err(SpaceError::BadRequest(
"space owner cannot leave; delete the space instead".to_string(),
));
}
let member = space_members::Entity::find_by_id((space.id, caller))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if member.status != MemberStatus::Active {
return Ok(());
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
client
.leave_organization(caller, &space.id.to_string())
.await?;
let now = OffsetDateTime::now_utc();
let mut am: space_members::ActiveModel = member.into();
am.status = Set(MemberStatus::Left);
am.left_at = Set(Some(now));
am.updated_at = Set(Some(now));
self.db()
.primary()
.transaction::<_, (), SpaceError>(|txn| {
Box::pin(async move {
space_member_groups::Entity::delete_many()
.filter(space_member_groups::Column::SpaceId.eq(space.id))
.filter(space_member_groups::Column::UserId.eq(caller))
.exec(txn)
.await?;
am.update(txn).await?;
spaces::Entity::update_many()
.filter(spaces::Column::Id.eq(space.id))
.col_expr(
spaces::Column::NumsMembers,
sea_orm::sea_query::Expr::cust("GREATEST(nums_members - 1, 0)"),
)
.exec(txn)
.await?;
Ok(())
})
})
.await?;
Ok(())
}
#[tracing::instrument(name = "space.remove_member", skip_all, fields(caller = %caller, space = %space_name, uid = %uid))]
pub async fn remove_member(
&self,
caller: i64,
space_name: &str,
uid: i64,
) -> Result<(), SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
if caller == uid {
return Err(SpaceError::BadRequest(
"owner cannot remove themselves".to_string(),
));
}
let member = space_members::Entity::find_by_id((space.id, uid))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if member.status != MemberStatus::Active {
return Ok(());
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
client
.remove_member(caller, &space.id.to_string(), uid, "")
.await?;
let now = OffsetDateTime::now_utc();
let mut am: space_members::ActiveModel = member.into();
am.status = Set(MemberStatus::Removed);
am.left_at = Set(Some(now));
am.updated_at = Set(Some(now));
self.db()
.primary()
.transaction::<_, (), SpaceError>(|txn| {
Box::pin(async move {
space_member_groups::Entity::delete_many()
.filter(space_member_groups::Column::SpaceId.eq(space.id))
.filter(space_member_groups::Column::UserId.eq(uid))
.exec(txn)
.await?;
am.update(txn).await?;
spaces::Entity::update_many()
.filter(spaces::Column::Id.eq(space.id))
.col_expr(spaces::Column::NumsMembers, Expr::cust("nums_members - 1"))
.exec(txn)
.await?;
Ok(())
})
})
.await?;
Ok(())
}
pub async fn list_members(
&self,
caller: i64,
space_name: &str,
q: PageQuery,
) -> Result<MemberPage, SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
let max_pagination = self.config().space.max_pagination;
let default_pagination = self.config().space.default_pagination;
let limit = q
.limit
.unwrap_or(default_pagination)
.clamp(1, max_pagination);
let offset = q
.offset
.unwrap_or(0)
.clamp(0, export::pagination::MAX_OFFSET);
let select = space_members::Entity::find()
.filter(space_members::Column::SpaceId.eq(space.id))
.order_by_asc(space_members::Column::JoinedAt);
let total = select.clone().count(self.db().read()).await? as i64;
let rows = select
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let mut uids: Vec<i64> = Vec::new();
for m in &rows {
uids.push(m.user_id);
if let Some(ib) = m.invited_by {
uids.push(ib);
}
}
let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&uids).await?;
let items = rows
.into_iter()
.map(|m| {
let user_brief = user_map.get(&m.user_id).cloned();
let inviter_brief = m.invited_by.and_then(|ib| user_map.get(&ib).cloned());
MemberDto {
space_id: m.space_id,
user: user_brief,
alias_name: m.alias_name,
alias_avatar_url: m.alias_avatar_url,
status: m.status,
last_active_at: m.last_active_at,
left_at: m.left_at,
invited_at: m.invited_at,
invited_by: inviter_brief,
joined_at: m.joined_at,
created_at: m.created_at,
updated_at: m.updated_at,
message_notifications: m.message_notifications,
muted: m.muted,
mute_until: m.mute_until,
mobile_push: m.mobile_push,
suppress_everyone: m.suppress_everyone,
suppress_roles: m.suppress_roles,
hide_muted_channels: m.hide_muted_channels,
}
})
.collect();
Ok(MemberPage {
items,
total,
limit,
offset,
})
}
}