ef0b83a
use models::chat::presence_states::{self, PresenceStatus};
use models::users::user_preferences;
use redis::AsyncCommands;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::lookup::{lookup_space_by_name, require_space_member};
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct PresenceDto {
pub user_id: i64,
pub user: Option<space::services::dto::UserDto>,
pub status: PresenceStatus,
pub activities: serde_json::Value,
pub last_seen_at: Option<OffsetDateTime>,
pub online: bool,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct SetPresenceCmd {
pub status: PresenceStatus,
#[serde(default)]
pub activities: Option<serde_json::Value>,
}
fn presence_key(uid: i64) -> String {
format!("chat:presence:{uid}")
}
#[derive(Debug, Serialize, Deserialize)]
struct PresencePayload {
status: PresenceStatus,
activities: serde_json::Value,
at: i64,
}
fn encode_presence(status: PresenceStatus, activities: serde_json::Value) -> Option<String> {
serde_json::to_string(&PresencePayload {
status,
activities,
at: OffsetDateTime::now_utc().unix_timestamp(),
})
.map_err(|error| tracing::warn!(%error, "presence payload serialize failed"))
.ok()
}
/// 隐藏在线状态时的对外形态：一律表现为离线，并抹掉最后在线时间与活动。
fn mask_presence(dto: &mut PresenceDto) {
dto.status = PresenceStatus::Offline;
dto.online = false;
dto.activities = serde_json::json!([]);
dto.last_seen_at = None;
}
impl ChatService {
pub async fn set_presence(
&self,
caller: i64,
cmd: SetPresenceCmd,
) -> Result<PresenceDto, ChatError> {
let now = OffsetDateTime::now_utc();
let activities = cmd.activities.unwrap_or(serde_json::json!([]));
let payload = PresencePayload {
status: cmd.status,
activities: activities.clone(),
at: now.unix_timestamp(),
};
let ttl = self.config().chat.presence_ttl_secs;
let mut conn = self.state.redis.connection_clone();
let key = presence_key(caller);
let value = serde_json::to_string(&payload)?;
let result: Result<(), redis::RedisError> = match &mut conn {
export::RedisConnection::Standalone(c) => c.set_ex(&key, &value, ttl).await,
export::RedisConnection::Cluster(c) => c.set_ex(&key, &value, ttl).await,
};
if let Err(error) = result {
tracing::warn!(%error, "failed to write presence to redis");
}
let existing = presence_states::Entity::find_by_id(caller)
.one(self.db().write())
.await?;
match existing {
Some(row) => {
let mut am: presence_states::ActiveModel = row.into();
am.status = Set(cmd.status);
am.activities = Set(activities.clone());
am.last_seen_at = Set(Some(now));
am.updated_at = Set(now);
am.update(self.db().write()).await?;
}
None => {
presence_states::ActiveModel {
user_id: Set(caller),
status: Set(cmd.status),
activities: Set(activities.clone()),
last_seen_at: Set(Some(now)),
updated_at: Set(now),
}
.insert(self.db().write())
.await?;
}
}
self.realtime
.publish_user(
caller,
&serde_json::json!({"kind": "presence", "status": cmd.status}),
)
.await;
Ok(PresenceDto {
user_id: caller,
user: None,
status: cmd.status,
activities,
last_seen_at: Some(now),
online: !matches!(cmd.status, PresenceStatus::Offline),
})
}
/* ------------------------------------------------------------- 生命周期 */
/// 聊天 WS 建立连接：用户上线。
///
/// 已有键（手动 Away/Dnd/隐身）原样保留、只续期；没有键才写 Active 并
/// 广播。多标签页/多实例共享同一份 Redis 状态：谁先连谁写，后来者续期。
/// 生命周期 TTL 取两倍心跳，崩溃后由 TTL 兜底回离线。
pub async fn mark_online(&self, uid: i64) {
let ttl = self.config().chat.online_ttl_secs();
let key = presence_key(uid);
let Some(value) = encode_presence(PresenceStatus::Active, serde_json::json!([])) else {
return;
};
match export::redis::set_nx_ex(&self.state.redis, &key, &value, ttl).await {
Ok(true) => {
self.realtime
.publish_user(
uid,
&serde_json::json!({"kind": "presence", "status": PresenceStatus::Active}),
)
.await;
}
Ok(false) => {
if let Err(error) = export::redis::touch_expire(&self.state.redis, &key, ttl).await
{
tracing::warn!(%error, "presence renew failed");
}
}
Err(error) => tracing::warn!(%error, "presence mark online failed"),
}
}
/// 心跳续期：键在只续期；键已丢失（休眠唤醒、网络割接）时补写 Active
/// ——连接还活着就应当在线。补写走 `SET NX`，不会覆盖用户此刻手动设置
/// 的状态。
pub async fn touch_presence(&self, uid: i64) {
let ttl = self.config().chat.online_ttl_secs();
let key = presence_key(uid);
match export::redis::touch_expire(&self.state.redis, &key, ttl).await {
Ok(true) => {}
Ok(false) => {
let Some(value) = encode_presence(PresenceStatus::Active, serde_json::json!([]))
else {
return;
};
if let Err(error) =
export::redis::set_nx_ex(&self.state.redis, &key, &value, ttl).await
{
tracing::warn!(%error, "presence re-mark failed");
}
}
Err(error) => tracing::warn!(%error, "presence touch failed"),
}
}
/// 最后一条聊天连接关闭：清 Redis 键并落库 `last_seen_at`。
///
/// 键的消失就是"离线"的唯一真相；DB 行只保留最后一次状态与时间，供
/// Redis 缺失时回退读取。
pub async fn mark_offline(&self, uid: i64) {
let now = OffsetDateTime::now_utc();
let key = presence_key(uid);
// 先读再删：DB 里留下的是用户最后处在的状态（Active/Away/Dnd/隐身）。
let last = export::redis::get_string(&self.state.redis, &key)
.await
.ok()
.flatten()
.and_then(|value| serde_json::from_str::<PresencePayload>(&value).ok());
if let Err(error) = export::redis::delete(&self.state.redis, &key).await {
tracing::warn!(%error, "presence key delete failed");
}
let (status, activities) = match last {
Some(payload) => (payload.status, payload.activities),
None => (PresenceStatus::Offline, serde_json::json!([])),
};
match presence_states::Entity::find_by_id(uid)
.one(self.db().write())
.await
{
Ok(Some(row)) => {
let mut am: presence_states::ActiveModel = row.into();
am.status = Set(status);
am.activities = Set(activities);
am.last_seen_at = Set(Some(now));
am.updated_at = Set(now);
if let Err(error) = am.update(self.db().write()).await {
tracing::warn!(%error, "presence last_seen update failed");
}
}
Ok(None) => {
let row = presence_states::ActiveModel {
user_id: Set(uid),
status: Set(status),
activities: Set(activities),
last_seen_at: Set(Some(now)),
updated_at: Set(now),
};
if let Err(error) = row.insert(self.db().write()).await {
tracing::warn!(%error, "presence last_seen insert failed");
}
}
Err(error) => tracing::warn!(%error, "presence row lookup failed"),
}
self.realtime
.publish_user(
uid,
&serde_json::json!({"kind": "presence", "status": PresenceStatus::Offline}),
)
.await;
}
/// `show_online_status` 开关：没有偏好行（从未保存过设置）按公开处理。
async fn show_online_map(&self, uids: &[i64]) -> Result<HashMap<i64, bool>, ChatError> {
if uids.is_empty() {
return Ok(HashMap::new());
}
let rows = user_preferences::Entity::find()
.filter(user_preferences::Column::UserId.is_in(uids.to_vec()))
.all(self.db().read())
.await?;
Ok(rows
.into_iter()
.map(|row| (row.user_id, row.show_online_status))
.collect())
}
pub async fn get_presence(&self, uid: i64) -> Result<PresenceDto, ChatError> {
let mut dto = self.get_presence_raw(uid).await?;
let map = self.load_user_briefs(&[uid]).await?;
dto.user = map.get(&uid).cloned();
Ok(dto)
}
async fn get_presence_raw(&self, uid: i64) -> Result<PresenceDto, ChatError> {
let mut conn = self.state.redis.connection_clone();
let key = presence_key(uid);
let value: Result<Option<String>, redis::RedisError> = match &mut conn {
export::RedisConnection::Standalone(c) => c.get(&key).await,
export::RedisConnection::Cluster(c) => c.get(&key).await,
};
if let Ok(Some(value)) = value
&& let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)
{
return Ok(PresenceDto {
user_id: uid,
user: None,
status: payload.status,
activities: payload.activities,
last_seen_at: Some(
OffsetDateTime::from_unix_timestamp(payload.at)
.unwrap_or_else(|_| OffsetDateTime::now_utc()),
),
online: !matches!(payload.status, PresenceStatus::Offline),
});
}
let row = presence_states::Entity::find_by_id(uid)
.one(self.db().read())
.await?;
match row {
Some(row) => Ok(PresenceDto {
user_id: uid,
user: None,
status: PresenceStatus::Offline,
activities: serde_json::json!([]),
last_seen_at: row.last_seen_at,
online: false,
}),
None => Ok(PresenceDto {
user_id: uid,
user: None,
status: PresenceStatus::Offline,
activities: serde_json::json!([]),
last_seen_at: None,
online: false,
}),
}
}
/// Presence of another user is only visible when both share an active
/// space or a DM room; otherwise the requester gets NotFound. A target
/// who turned off `show_online_status` always reads as offline.
pub async fn get_presence_for(
&self,
caller: i64,
target: i64,
) -> Result<PresenceDto, ChatError> {
if caller == target || self.shares_room_with(caller, target).await? {
let mut dto = self.get_presence(target).await?;
if dto.user_id != caller {
let visible = self
.show_online_map(&[target])
.await?
.get(&target)
.copied()
.unwrap_or(true);
if !visible {
mask_presence(&mut dto);
}
}
return Ok(dto);
}
Err(ChatError::NotFound)
}
async fn shares_room_with(&self, caller: i64, target: i64) -> Result<bool, ChatError> {
use models::room::room_members;
let mine = space::authority::list_user_spaces(&self.state, caller, caller).await?;
let theirs = space::authority::list_user_spaces(&self.state, target, target).await?;
if mine
.iter()
.any(|left| theirs.iter().any(|right| left.id == right.id))
{
return Ok(true);
}
use models::room::rooms;
use sea_orm::Condition;
let my_rooms: Vec<Uuid> = room_members::Entity::find()
.filter(room_members::Column::UserId.eq(caller))
.filter(room_members::Column::LeftAt.is_null())
.select_only()
.column(room_members::Column::RoomId)
.into_tuple()
.all(self.db().read())
.await?;
if my_rooms.is_empty() {
return Ok(false);
}
let shared = rooms::Entity::find()
.filter(rooms::Column::Id.is_in(my_rooms.clone()))
.filter(rooms::Column::SpaceId.is_null())
.filter(
Condition::any()
.add(rooms::Column::OwnerId.eq(caller))
.add(rooms::Column::OwnerId.eq(target)),
)
.count(self.db().read())
.await?;
if shared > 0 {
return Ok(true);
}
let dm_peer = room_members::Entity::find()
.filter(room_members::Column::UserId.eq(target))
.filter(room_members::Column::RoomId.is_in(my_rooms))
.filter(room_members::Column::LeftAt.is_null())
.count(self.db().read())
.await?;
Ok(dm_peer > 0)
}
pub async fn get_presences(&self, uids: &[i64]) -> Result<Vec<PresenceDto>, ChatError> {
if uids.is_empty() {
return Ok(Vec::new());
}
let mut conn = self.state.redis.connection_clone();
let keys: Vec<String> = uids.iter().map(|uid| presence_key(*uid)).collect();
let values: Vec<Option<String>> = match &mut conn {
export::RedisConnection::Standalone(c) => {
redis::cmd("MGET")
.arg(keys.clone())
.query_async::<Vec<Option<String>>>(c)
.await?
}
export::RedisConnection::Cluster(c) => {
// 集群里每个键可能落在不同槽位，MGET 不成立，只能逐个 GET。
let mut values = Vec::with_capacity(keys.len());
for key in &keys {
values.push(redis::cmd("GET").arg(key).query_async(c).await?);
}
values
}
};
let mut by_uid: HashMap<i64, PresencePayload> = HashMap::new();
for (uid, value) in uids.iter().zip(values) {
if let Some(value) = value
&& let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)
{
by_uid.insert(*uid, payload);
}
}
let missing: Vec<i64> = uids
.iter()
.filter(|u| !by_uid.contains_key(u))
.copied()
.collect();
let rows = if missing.is_empty() {
Vec::new()
} else {
presence_states::Entity::find()
.filter(presence_states::Column::UserId.is_in(missing))
.all(self.db().read())
.await?
};
let mut result = Vec::with_capacity(uids.len());
for uid in uids {
if let Some(payload) = by_uid.get(uid) {
result.push(PresenceDto {
user_id: *uid,
user: None,
status: payload.status,
activities: payload.activities.clone(),
last_seen_at: Some(
OffsetDateTime::from_unix_timestamp(payload.at)
.unwrap_or_else(|_| OffsetDateTime::now_utc()),
),
online: !matches!(payload.status, PresenceStatus::Offline),
});
continue;
}
let row = rows.iter().find(|r| r.user_id == *uid);
result.push(PresenceDto {
user_id: *uid,
user: None,
status: PresenceStatus::Offline,
activities: serde_json::json!([]),
last_seen_at: row.and_then(|r| r.last_seen_at),
online: false,
});
}
let map = self.load_user_briefs(uids).await?;
for dto in result.iter_mut() {
dto.user = map.get(&dto.user_id).cloned();
}
Ok(result)
}
pub async fn list_space_presences(
&self,
caller: i64,
space_name: &str,
) -> Result<Vec<PresenceDto>, ChatError> {
let space = lookup_space_by_name(self, space_name).await?;
require_space_member(self, &space, caller).await?;
let members =
space::authority::list_members(&self.state, caller, space.id, false, false).await?;
let mut uids: Vec<i64> = members.into_iter().map(|m| m.user_id).collect();
if !uids.contains(&space.owner_user_id) {
uids.push(space.owner_user_id);
}
let mut presences = self.get_presences(&uids).await?;
// 隐私：关掉"展示在线状态"的成员对他人一律表现为离线，本人照常。
let visible = self.show_online_map(&uids).await?;
for dto in presences.iter_mut() {
if dto.user_id != caller && !visible.get(&dto.user_id).copied().unwrap_or(true) {
mask_presence(dto);
}
}
Ok(presences)
}
}
impl ChatService {
/// 聊天连接建立：置在线并把这条连接登记进在线占位集合。
///
/// 占位集合与 `presence` 键是两套语义：`chat:ws:{uid}` 回答"有多少条
/// 连接还活着"（推送 worker 用它决定是否跳过系统通知），`chat:presence:{uid}`
/// 回答"用户此刻的状态是什么"。同一连接同时续期两者。
pub async fn connection_open(&self, uid: i64, session: &str) -> Result<(), ChatError> {
self.mark_online(uid).await;
let ttl = self.config().chat.online_ttl_secs();
export::redis::zset_touch(
&self.state.redis,
&export::push::online_key(uid),
session,
ttl,
)
.await?;
Ok(())
}
/// 心跳续期：两套状态一起续，避免"有连接但状态过期"的中间态。
pub async fn connection_touch(&self, uid: i64, session: &str) -> Result<(), ChatError> {
self.touch_presence(uid).await;
let ttl = self.config().chat.online_ttl_secs();
export::redis::zset_touch(
&self.state.redis,
&export::push::online_key(uid),
session,
ttl,
)
.await?;
Ok(())
}
/// 连接关闭：注销该连接占位；只有最后一条连接消失才算离线。
///
/// 多标签页场景下先关的那一页不能把用户判成离线——这是把
/// `mark_offline` 放在 `zset_release` 返回值之后的原因。
pub async fn connection_close(&self, uid: i64, session: &str) -> Result<(), ChatError> {
let live =
export::redis::zset_release(&self.state.redis, &export::push::online_key(uid), session)
.await?;
if live == 0 {
self.mark_offline(uid).await;
}
Ok(())
}
}