Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'main' into guest-exp

Browse source
This commit is contained in:
Conrad Irwin 2023-10-23 17:47:21 +02:00
commit ea4e67fb76
141 changed files with 6720 additions and 2077 deletions
Download patch file Download diff file Expand all files Collapse all files

306
crates/collab/src/rpc.rs
View file

@ -3,9 +3,11 @@ mod connection_pool;
use crate::{
auth,
db::{
self, BufferId, ChannelId, ChannelRole, ChannelsForUser, CreateChannelResult, Database,
MembershipUpdated, MessageId, MoveChannelResult, ProjectId, RenameChannelResult, RoomId,
ServerId, SetChannelVisibilityResult, User, UserId,
self, BufferId, ChannelId, ChannelRole, ChannelsForUser, CreateChannelResult,
CreatedChannelMessage, Database, InviteMemberResult, MembershipUpdated, MessageId,
MoveChannelResult, NotificationId, ProjectId, RemoveChannelMemberResult,
RenameChannelResult, RespondToChannelInvite, RoomId, ServerId, SetChannelVisibilityResult,
User, UserId,
},
executor::Executor,
AppState, Result,
@ -71,6 +73,7 @@ pub const CLEANUP_TIMEOUT: Duration = Duration::from_secs(10);
const MESSAGE_COUNT_PER_PAGE: usize = 100;
const MAX_MESSAGE_LEN: usize = 1024;
const NOTIFICATION_COUNT_PER_PAGE: usize = 50;
lazy_static! {
static ref METRIC_CONNECTIONS: IntGauge =
@ -271,6 +274,9 @@ impl Server {
.add_request_handler(send_channel_message)
.add_request_handler(remove_channel_message)
.add_request_handler(get_channel_messages)
.add_request_handler(get_channel_messages_by_id)
.add_request_handler(get_notifications)
.add_request_handler(mark_notification_as_read)
.add_request_handler(link_channel)
.add_request_handler(unlink_channel)
.add_request_handler(move_channel)
@ -390,7 +396,7 @@ impl Server {
let contacts = app_state.db.get_contacts(user_id).await.trace_err();
if let Some((busy, contacts)) = busy.zip(contacts) {
let pool = pool.lock();
let updated_contact = contact_for_user(user_id, false, busy, &pool);
let updated_contact = contact_for_user(user_id, busy, &pool);
for contact in contacts {
if let db::Contact::Accepted {
user_id: contact_user_id,
@ -584,7 +590,7 @@ impl Server {
let (contacts, channels_for_user, channel_invites) = future::try_join3(
this.app_state.db.get_contacts(user_id),
this.app_state.db.get_channels_for_user(user_id),
this.app_state.db.get_channel_invites_for_user(user_id)
this.app_state.db.get_channel_invites_for_user(user_id),
).await?;
{
@ -690,7 +696,7 @@ impl Server {
if let Some(user) = self.app_state.db.get_user_by_id(inviter_id).await? {
if let Some(code) = &user.invite_code {
let pool = self.connection_pool.lock();
let invitee_contact = contact_for_user(invitee_id, true, false, &pool);
let invitee_contact = contact_for_user(invitee_id, false, &pool);
for connection_id in pool.user_connection_ids(inviter_id) {
self.peer.send(
connection_id,
@ -2066,7 +2072,7 @@ async fn request_contact(
return Err(anyhow!("cannot add yourself as a contact"))?;
}
session
let notifications = session
.db()
.await
.send_contact_request(requester_id, responder_id)
@ -2089,16 +2095,14 @@ async fn request_contact(
.incoming_requests
.push(proto::IncomingContactRequest {
requester_id: requester_id.to_proto(),
should_notify: true,
});
for connection_id in session
.connection_pool()
.await
.user_connection_ids(responder_id)
{
let connection_pool = session.connection_pool().await;
for connection_id in connection_pool.user_connection_ids(responder_id) {
session.peer.send(connection_id, update.clone())?;
}
send_notifications(&*connection_pool, &session.peer, notifications);
response.send(proto::Ack {})?;
Ok(())
}
@ -2117,7 +2121,8 @@ async fn respond_to_contact_request(
} else {
let accept = request.response == proto::ContactRequestResponse::Accept as i32;
db.respond_to_contact_request(responder_id, requester_id, accept)
let notifications = db
.respond_to_contact_request(responder_id, requester_id, accept)
.await?;
let requester_busy = db.is_user_busy(requester_id).await?;
let responder_busy = db.is_user_busy(responder_id).await?;
@ -2128,7 +2133,7 @@ async fn respond_to_contact_request(
if accept {
update
.contacts
.push(contact_for_user(requester_id, false, requester_busy, &pool));
.push(contact_for_user(requester_id, requester_busy, &pool));
}
update
.remove_incoming_requests
@ -2142,14 +2147,17 @@ async fn respond_to_contact_request(
if accept {
update
.contacts
.push(contact_for_user(responder_id, true, responder_busy, &pool));
.push(contact_for_user(responder_id, responder_busy, &pool));
}
update
.remove_outgoing_requests
.push(responder_id.to_proto());
for connection_id in pool.user_connection_ids(requester_id) {
session.peer.send(connection_id, update.clone())?;
}
send_notifications(&*pool, &session.peer, notifications);
}
response.send(proto::Ack {})?;
@ -2164,7 +2172,8 @@ async fn remove_contact(
let requester_id = session.user_id;
let responder_id = UserId::from_proto(request.user_id);
let db = session.db().await;
let contact_accepted = db.remove_contact(requester_id, responder_id).await?;
let (contact_accepted, deleted_notification_id) =
db.remove_contact(requester_id, responder_id).await?;
let pool = session.connection_pool().await;
// Update outgoing contact requests of requester
@ -2191,6 +2200,14 @@ async fn remove_contact(
}
for connection_id in pool.user_connection_ids(responder_id) {
session.peer.send(connection_id, update.clone())?;
if let Some(notification_id) = deleted_notification_id {
session.peer.send(
connection_id,
proto::DeleteNotification {
notification_id: notification_id.to_proto(),
},
)?;
}
}
response.send(proto::Ack {})?;
@ -2268,7 +2285,10 @@ async fn invite_channel_member(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let invitee_id = UserId::from_proto(request.user_id);
let channel = db
let InviteMemberResult {
channel,
notifications,
} = db
.invite_channel_member(
channel_id,
invitee_id,
@ -2282,14 +2302,13 @@ async fn invite_channel_member(
..Default::default()
};
for connection_id in session
.connection_pool()
.await
.user_connection_ids(invitee_id)
{
let connection_pool = session.connection_pool().await;
for connection_id in connection_pool.user_connection_ids(invitee_id) {
session.peer.send(connection_id, update.clone())?;
}
send_notifications(&*connection_pool, &session.peer, notifications);
response.send(proto::Ack {})?;
Ok(())
}
@ -2303,13 +2322,33 @@ async fn remove_channel_member(
let channel_id = ChannelId::from_proto(request.channel_id);
let member_id = UserId::from_proto(request.user_id);
let membership_updated = db
let RemoveChannelMemberResult {
membership_update,
notification_id,
} = db
.remove_channel_member(channel_id, member_id, session.user_id)
.await?;
dbg!(&membership_updated);
notify_membership_updated(membership_updated, member_id, &session).await?;
let connection_pool = &session.connection_pool().await;
notify_membership_updated(
&connection_pool,
membership_update,
member_id,
&session.peer,
);
for connection_id in connection_pool.user_connection_ids(member_id) {
if let Some(notification_id) = notification_id {
session
.peer
.send(
connection_id,
proto::DeleteNotification {
notification_id: notification_id.to_proto(),
},
)
.trace_err();
}
}
response.send(proto::Ack {})?;
Ok(())
@ -2374,7 +2413,13 @@ async fn set_channel_member_role(
match result {
db::SetMemberRoleResult::MembershipUpdated(membership_update) => {
notify_membership_updated(membership_update, member_id, &session).await?;
let connection_pool = session.connection_pool().await;
notify_membership_updated(
&connection_pool,
membership_update,
member_id,
&session.peer,
)
}
db::SetMemberRoleResult::InviteUpdated(channel) => {
let update = proto::UpdateChannels {
@ -2535,24 +2580,34 @@ async fn respond_to_channel_invite(
) -> Result<()> {
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let result = db
let RespondToChannelInvite {
membership_update,
notifications,
} = db
.respond_to_channel_invite(channel_id, session.user_id, request.accept)
.await?;
if let Some(accept_invite_result) = result {
notify_membership_updated(accept_invite_result, session.user_id, &session).await?;
let connection_pool = session.connection_pool().await;
if let Some(membership_update) = membership_update {
notify_membership_updated(
&connection_pool,
membership_update,
session.user_id,
&session.peer,
);
} else {
let update = proto::UpdateChannels {
remove_channel_invitations: vec![channel_id.to_proto()],
..Default::default()
};
let connection_pool = session.connection_pool().await;
for connection_id in connection_pool.user_connection_ids(session.user_id) {
session.peer.send(connection_id, update.clone())?;
}
};
send_notifications(&*connection_pool, &session.peer, notifications);
response.send(proto::Ack {})?;
Ok(())
@ -2635,8 +2690,14 @@ async fn join_channel_internal(
live_kit_connection_info,
})?;
let connection_pool = session.connection_pool().await;
if let Some(accept_invite_result) = accept_invite_result {
notify_membership_updated(accept_invite_result, session.user_id, &session).await?;
notify_membership_updated(
&connection_pool,
accept_invite_result,
session.user_id,
&session.peer,
);
}
room_updated(&joined_room.room, &session.peer);
@ -2805,6 +2866,29 @@ fn channel_buffer_updated<T: EnvelopedMessage>(
});
}
fn send_notifications(
connection_pool: &ConnectionPool,
peer: &Peer,
notifications: db::NotificationBatch,
) {
for (user_id, notification) in notifications {
for connection_id in connection_pool.user_connection_ids(user_id) {
if let Err(error) = peer.send(
connection_id,
proto::AddNotification {
notification: Some(notification.clone()),
},
) {
tracing::error!(
"failed to send notification to {:?} {}",
connection_id,
error
);
}
}
}
}
async fn send_channel_message(
request: proto::SendChannelMessage,
response: Response<proto::SendChannelMessage>,
@ -2819,19 +2903,27 @@ async fn send_channel_message(
return Err(anyhow!("message can't be blank"))?;
}
// TODO: adjust mentions if body is trimmed
let timestamp = OffsetDateTime::now_utc();
let nonce = request
.nonce
.ok_or_else(|| anyhow!("nonce can't be blank"))?;
let channel_id = ChannelId::from_proto(request.channel_id);
let (message_id, connection_ids, non_participants) = session
let CreatedChannelMessage {
message_id,
participant_connection_ids,
channel_members,
notifications,
} = session
.db()
.await
.create_channel_message(
channel_id,
session.user_id,
&body,
&request.mentions,
timestamp,
nonce.clone().into(),
)
@ -2840,18 +2932,23 @@ async fn send_channel_message(
sender_id: session.user_id.to_proto(),
id: message_id.to_proto(),
body,
mentions: request.mentions,
timestamp: timestamp.unix_timestamp() as u64,
nonce: Some(nonce),
};
broadcast(Some(session.connection_id), connection_ids, |connection| {
session.peer.send(
connection,
proto::ChannelMessageSent {
channel_id: channel_id.to_proto(),
message: Some(message.clone()),
},
)
});
broadcast(
Some(session.connection_id),
participant_connection_ids,
|connection| {
session.peer.send(
connection,
proto::ChannelMessageSent {
channel_id: channel_id.to_proto(),
message: Some(message.clone()),
},
)
},
);
response.send(proto::SendChannelMessageResponse {
message: Some(message),
})?;
@ -2859,7 +2956,7 @@ async fn send_channel_message(
let pool = &*session.connection_pool().await;
broadcast(
None,
non_participants
channel_members
.iter()
.flat_map(|user_id| pool.user_connection_ids(*user_id)),
|peer_id| {
@ -2875,6 +2972,7 @@ async fn send_channel_message(
)
},
);
send_notifications(pool, &session.peer, notifications);
Ok(())
}
@ -2904,11 +3002,16 @@ async fn acknowledge_channel_message(
) -> Result<()> {
let channel_id = ChannelId::from_proto(request.channel_id);
let message_id = MessageId::from_proto(request.message_id);
session
let notifications = session
.db()
.await
.observe_channel_message(channel_id, session.user_id, message_id)
.await?;
send_notifications(
&*session.connection_pool().await,
&session.peer,
notifications,
);
Ok(())
}
@ -2983,6 +3086,72 @@ async fn get_channel_messages(
Ok(())
}
async fn get_channel_messages_by_id(
request: proto::GetChannelMessagesById,
response: Response<proto::GetChannelMessagesById>,
session: Session,
) -> Result<()> {
let message_ids = request
.message_ids
.iter()
.map(|id| MessageId::from_proto(*id))
.collect::<Vec<_>>();
let messages = session
.db()
.await
.get_channel_messages_by_id(session.user_id, &message_ids)
.await?;
response.send(proto::GetChannelMessagesResponse {
done: messages.len() < MESSAGE_COUNT_PER_PAGE,
messages,
})?;
Ok(())
}
async fn get_notifications(
request: proto::GetNotifications,
response: Response<proto::GetNotifications>,
session: Session,
) -> Result<()> {
let notifications = session
.db()
.await
.get_notifications(
session.user_id,
NOTIFICATION_COUNT_PER_PAGE,
request
.before_id
.map(|id| db::NotificationId::from_proto(id)),
)
.await?;
response.send(proto::GetNotificationsResponse {
done: notifications.len() < NOTIFICATION_COUNT_PER_PAGE,
notifications,
})?;
Ok(())
}
async fn mark_notification_as_read(
request: proto::MarkNotificationRead,
response: Response<proto::MarkNotificationRead>,
session: Session,
) -> Result<()> {
let database = &session.db().await;
let notifications = database
.mark_notification_as_read_by_id(
session.user_id,
NotificationId::from_proto(request.notification_id),
)
.await?;
send_notifications(
&*session.connection_pool().await,
&session.peer,
notifications,
);
response.send(proto::Ack {})?;
Ok(())
}
async fn update_diff_base(request: proto::UpdateDiffBase, session: Session) -> Result<()> {
let project_id = ProjectId::from_proto(request.project_id);
let project_connection_ids = session
@ -3052,11 +3221,12 @@ fn to_tungstenite_message(message: AxumMessage) -> TungsteniteMessage {
}
}
async fn notify_membership_updated(
fn notify_membership_updated(
connection_pool: &ConnectionPool,
result: MembershipUpdated,
user_id: UserId,
session: &Session,
) -> Result<()> {
peer: &Peer,
) {
let mut update = build_channels_update(result.new_channels, vec![]);
update.delete_channels = result
.removed_channels
@ -3065,11 +3235,9 @@ async fn notify_membership_updated(
.collect();
update.remove_channel_invitations = vec![result.channel_id.to_proto()];
let connection_pool = session.connection_pool().await;
for connection_id in connection_pool.user_connection_ids(user_id) {
session.peer.send(connection_id, update.clone())?;
peer.send(connection_id, update.clone()).trace_err();
}
Ok(())
}
fn build_channels_update(
@ -3120,42 +3288,28 @@ fn build_initial_contacts_update(
for contact in contacts {
match contact {
db::Contact::Accepted {
user_id,
should_notify,
busy,
} => {
update
.contacts
.push(contact_for_user(user_id, should_notify, busy, &pool));
db::Contact::Accepted { user_id, busy } => {
update.contacts.push(contact_for_user(user_id, busy, &pool));
}
db::Contact::Outgoing { user_id } => update.outgoing_requests.push(user_id.to_proto()),
db::Contact::Incoming {
user_id,
should_notify,
} => update
.incoming_requests
.push(proto::IncomingContactRequest {
requester_id: user_id.to_proto(),
should_notify,
}),
db::Contact::Incoming { user_id } => {
update
.incoming_requests
.push(proto::IncomingContactRequest {
requester_id: user_id.to_proto(),
})
}
}
}
update
}
fn contact_for_user(
user_id: UserId,
should_notify: bool,
busy: bool,
pool: &ConnectionPool,
) -> proto::Contact {
fn contact_for_user(user_id: UserId, busy: bool, pool: &ConnectionPool) -> proto::Contact {
proto::Contact {
user_id: user_id.to_proto(),
online: pool.is_user_online(user_id),
busy,
should_notify,
}
}
@ -3216,7 +3370,7 @@ async fn update_user_contacts(user_id: UserId, session: &Session) -> Result<()>
let busy = db.is_user_busy(user_id).await?;
let pool = session.connection_pool().await;
let updated_contact = contact_for_user(user_id, false, busy, &pool);
let updated_contact = contact_for_user(user_id, busy, &pool);
for contact in contacts {
if let db::Contact::Accepted {
user_id: contact_user_id,