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

Add the ability to notify when a user accepts a contact request

Browse source 
Co-Authored-By: Nathan Sobo <nathan@zed.dev>
Co-Authored-By: Max Brunsfeld <max@zed.dev>
This commit is contained in:
Antonio Scandurra 2022-05-11 18:51:40 +02:00
parent 933a1f2cd6
commit a5fd664b00
3 changed files with 316 additions and 150 deletions
Download patch file Download diff file Expand all files Collapse all files

381
crates/collab/src/db.rs
View file

@ -17,10 +17,11 @@ pub trait Db: Send + Sync {
async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()>; async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()>;
async fn destroy_user(&self, id: UserId) -> Result<()>; async fn destroy_user(&self, id: UserId) -> Result<()>;
async fn get_contacts(&self, id: UserId) -> Result<Contacts>; async fn get_contacts(&self, id: UserId) -> Result<Vec<Contact>>;
async fn has_contact(&self, user_id_a: UserId, user_id_b: UserId) -> Result<bool>;
async fn send_contact_request(&self, requester_id: UserId, responder_id: UserId) -> Result<()>; async fn send_contact_request(&self, requester_id: UserId, responder_id: UserId) -> Result<()>;
async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<()>; async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<()>;
async fn dismiss_contact_request( async fn dismiss_contact_notification(
&self, &self,
responder_id: UserId, responder_id: UserId,
requester_id: UserId, requester_id: UserId,
@ -190,7 +191,7 @@ impl Db for PostgresDb {
// contacts // contacts
async fn get_contacts(&self, user_id: UserId) -> Result<Contacts> { async fn get_contacts(&self, user_id: UserId) -> Result<Vec<Contact>> {
let query = " let query = "
SELECT user_id_a, user_id_b, a_to_b, accepted, should_notify SELECT user_id_a, user_id_b, a_to_b, accepted, should_notify
FROM contacts FROM contacts
@ -201,46 +202,67 @@ impl Db for PostgresDb {
.bind(user_id) .bind(user_id)
.fetch(&self.pool); .fetch(&self.pool);
let mut current = vec![user_id]; let mut contacts = vec![Contact::Accepted {
let mut outgoing_requests = Vec::new(); user_id,
let mut incoming_requests = Vec::new(); should_notify: false,
}];
while let Some(row) = rows.next().await { while let Some(row) = rows.next().await {
let (user_id_a, user_id_b, a_to_b, accepted, should_notify) = row?; let (user_id_a, user_id_b, a_to_b, accepted, should_notify) = row?;
if user_id_a == user_id { if user_id_a == user_id {
if accepted { if accepted {
current.push(user_id_b); contacts.push(Contact::Accepted {
user_id: user_id_b,
should_notify: should_notify && a_to_b,
});
} else if a_to_b { } else if a_to_b {
outgoing_requests.push(user_id_b); contacts.push(Contact::Outgoing { user_id: user_id_b })
} else { } else {
incoming_requests.push(IncomingContactRequest { contacts.push(Contact::Incoming {
requester_id: user_id_b, user_id: user_id_b,
should_notify, should_notify,
}); });
} }
} else { } else {
if accepted { if accepted {
current.push(user_id_a); contacts.push(Contact::Accepted {
user_id: user_id_a,
should_notify: should_notify && !a_to_b,
});
} else if a_to_b { } else if a_to_b {
incoming_requests.push(IncomingContactRequest { contacts.push(Contact::Incoming {
requester_id: user_id_a, user_id: user_id_a,
should_notify, should_notify,
}); });
} else { } else {
outgoing_requests.push(user_id_a); contacts.push(Contact::Outgoing { user_id: user_id_a });
} }
} }
} }
current.sort_unstable(); contacts.sort_unstable_by_key(|contact| contact.user_id());
outgoing_requests.sort_unstable();
incoming_requests.sort_unstable();
Ok(Contacts { Ok(contacts)
current, }
outgoing_requests,
incoming_requests, async fn has_contact(&self, user_id_1: UserId, user_id_2: UserId) -> Result<bool> {
}) let (id_a, id_b) = if user_id_1 < user_id_2 {
(user_id_1, user_id_2)
} else {
(user_id_2, user_id_1)
};
let query = "
SELECT 1 FROM contacts
WHERE user_id_a = $1 AND user_id_b = $2 AND accepted = 't'
LIMIT 1
";
Ok(sqlx::query_scalar::<_, i32>(query)
.bind(id_a.0)
.bind(id_b.0)
.fetch_optional(&self.pool)
.await?
.is_some())
} }
async fn send_contact_request(&self, sender_id: UserId, receiver_id: UserId) -> Result<()> { async fn send_contact_request(&self, sender_id: UserId, receiver_id: UserId) -> Result<()> {
@ -254,7 +276,8 @@ impl Db for PostgresDb {
VALUES ($1, $2, $3, 'f', 't') VALUES ($1, $2, $3, 'f', 't')
ON CONFLICT (user_id_a, user_id_b) DO UPDATE ON CONFLICT (user_id_a, user_id_b) DO UPDATE
SET SET
accepted = 't' accepted = 't',
should_notify = 'f'
WHERE WHERE
NOT contacts.accepted AND NOT contacts.accepted AND
((contacts.a_to_b = excluded.a_to_b AND contacts.user_id_a = excluded.user_id_b) OR ((contacts.a_to_b = excluded.a_to_b AND contacts.user_id_a = excluded.user_id_b) OR
@ -297,21 +320,26 @@ impl Db for PostgresDb {
} }
} }
async fn dismiss_contact_request( async fn dismiss_contact_notification(
&self, &self,
responder_id: UserId, user_id: UserId,
requester_id: UserId, contact_user_id: UserId,
) -> Result<()> { ) -> Result<()> {
let (id_a, id_b, a_to_b) = if responder_id < requester_id { let (id_a, id_b, a_to_b) = if user_id < contact_user_id {
(responder_id, requester_id, false) (user_id, contact_user_id, true)
} else { } else {
(requester_id, responder_id, true) (contact_user_id, user_id, false)
}; };
let query = " let query = "
UPDATE contacts UPDATE contacts
SET should_notify = 'f' SET should_notify = 'f'
WHERE user_id_a = $1 AND user_id_b = $2 AND a_to_b = $3; WHERE
user_id_a = $1 AND user_id_b = $2 AND
(
(a_to_b = $3 AND accepted) OR
(a_to_b != $3 AND NOT accepted)
);
"; ";
let result = sqlx::query(query) let result = sqlx::query(query)
@ -342,7 +370,7 @@ impl Db for PostgresDb {
let result = if accept { let result = if accept {
let query = " let query = "
UPDATE contacts UPDATE contacts
SET accepted = 't', should_notify = 'f' SET accepted = 't', should_notify = 't'
WHERE user_id_a = $1 AND user_id_b = $2 AND a_to_b = $3; WHERE user_id_a = $1 AND user_id_b = $2 AND a_to_b = $3;
"; ";
sqlx::query(query) sqlx::query(query)
@ -702,10 +730,28 @@ pub struct ChannelMessage {
} }
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct Contacts { pub enum Contact {
pub current: Vec<UserId>, Accepted {
pub incoming_requests: Vec<IncomingContactRequest>, user_id: UserId,
pub outgoing_requests: Vec<UserId>, should_notify: bool,
},
Outgoing {
user_id: UserId,
},
Incoming {
user_id: UserId,
should_notify: bool,
},
}
impl Contact {
pub fn user_id(&self) -> UserId {
match self {
Contact::Accepted { user_id, .. } => *user_id,
Contact::Outgoing { user_id } => *user_id,
Contact::Incoming { user_id, .. } => *user_id,
}
}
} }
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
@ -947,51 +993,60 @@ pub mod tests {
// User starts with no contacts // User starts with no contacts
assert_eq!( assert_eq!(
db.get_contacts(user_1).await.unwrap(), db.get_contacts(user_1).await.unwrap(),
Contacts { vec![Contact::Accepted {
current: vec![user_1], user_id: user_1,
outgoing_requests: vec![], should_notify: false
incoming_requests: vec![], }],
},
); );
// User requests a contact. Both users see the pending request. // User requests a contact. Both users see the pending request.
db.send_contact_request(user_1, user_2).await.unwrap(); db.send_contact_request(user_1, user_2).await.unwrap();
assert!(!db.has_contact(user_1, user_2).await.unwrap());
assert!(!db.has_contact(user_2, user_1).await.unwrap());
assert_eq!( assert_eq!(
db.get_contacts(user_1).await.unwrap(), db.get_contacts(user_1).await.unwrap(),
Contacts { &[
current: vec![user_1], Contact::Accepted {
outgoing_requests: vec![user_2], user_id: user_1,
incoming_requests: vec![], should_notify: false
}, },
Contact::Outgoing { user_id: user_2 }
],
); );
assert_eq!( assert_eq!(
db.get_contacts(user_2).await.unwrap(), db.get_contacts(user_2).await.unwrap(),
Contacts { &[
current: vec![user_2], Contact::Incoming {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![IncomingContactRequest {
requester_id: user_1,
should_notify: true should_notify: true
}], },
}, Contact::Accepted {
user_id: user_2,
should_notify: false
},
]
); );
// User 2 dismisses the contact request notification without accepting or rejecting. // User 2 dismisses the contact request notification without accepting or rejecting.
// We shouldn't notify them again. // We shouldn't notify them again.
db.dismiss_contact_request(user_1, user_2) db.dismiss_contact_notification(user_1, user_2)
.await .await
.unwrap_err(); .unwrap_err();
db.dismiss_contact_request(user_2, user_1).await.unwrap(); db.dismiss_contact_notification(user_2, user_1)
.await
.unwrap();
assert_eq!( assert_eq!(
db.get_contacts(user_2).await.unwrap(), db.get_contacts(user_2).await.unwrap(),
Contacts { &[
current: vec![user_2], Contact::Incoming {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![IncomingContactRequest {
requester_id: user_1,
should_notify: false should_notify: false
}], },
}, Contact::Accepted {
user_id: user_2,
should_notify: false
},
]
); );
// User can't accept their own contact request // User can't accept their own contact request
@ -1005,44 +1060,106 @@ pub mod tests {
.unwrap(); .unwrap();
assert_eq!( assert_eq!(
db.get_contacts(user_1).await.unwrap(), db.get_contacts(user_1).await.unwrap(),
Contacts { &[
current: vec![user_1, user_2], Contact::Accepted {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![], should_notify: false
}, },
Contact::Accepted {
user_id: user_2,
should_notify: true
}
],
); );
assert!(db.has_contact(user_1, user_2).await.unwrap());
assert!(db.has_contact(user_2, user_1).await.unwrap());
assert_eq!( assert_eq!(
db.get_contacts(user_2).await.unwrap(), db.get_contacts(user_2).await.unwrap(),
Contacts { &[
current: vec![user_1, user_2], Contact::Accepted {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![], should_notify: false,
}, },
Contact::Accepted {
user_id: user_2,
should_notify: false,
},
]
); );
// Users cannot re-request existing contacts. // Users cannot re-request existing contacts.
db.send_contact_request(user_1, user_2).await.unwrap_err(); db.send_contact_request(user_1, user_2).await.unwrap_err();
db.send_contact_request(user_2, user_1).await.unwrap_err(); db.send_contact_request(user_2, user_1).await.unwrap_err();
// Users can't dismiss notifications of them accepting other users' requests.
db.dismiss_contact_notification(user_2, user_1)
.await
.unwrap_err();
assert_eq!(
db.get_contacts(user_1).await.unwrap(),
&[
Contact::Accepted {
user_id: user_1,
should_notify: false
},
Contact::Accepted {
user_id: user_2,
should_notify: true,
},
]
);
// Users can dismiss notifications of other users accepting their requests.
db.dismiss_contact_notification(user_1, user_2)
.await
.unwrap();
assert_eq!(
db.get_contacts(user_1).await.unwrap(),
&[
Contact::Accepted {
user_id: user_1,
should_notify: false
},
Contact::Accepted {
user_id: user_2,
should_notify: false,
},
]
);
// Users send each other concurrent contact requests and // Users send each other concurrent contact requests and
// see that they are immediately accepted. // see that they are immediately accepted.
db.send_contact_request(user_1, user_3).await.unwrap(); db.send_contact_request(user_1, user_3).await.unwrap();
db.send_contact_request(user_3, user_1).await.unwrap(); db.send_contact_request(user_3, user_1).await.unwrap();
assert_eq!( assert_eq!(
db.get_contacts(user_1).await.unwrap(), db.get_contacts(user_1).await.unwrap(),
Contacts { &[
current: vec![user_1, user_2, user_3], Contact::Accepted {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![], should_notify: false
}, },
Contact::Accepted {
user_id: user_2,
should_notify: false,
},
Contact::Accepted {
user_id: user_3,
should_notify: false
},
]
); );
assert_eq!( assert_eq!(
db.get_contacts(user_3).await.unwrap(), db.get_contacts(user_3).await.unwrap(),
Contacts { &[
current: vec![user_1, user_3], Contact::Accepted {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![], should_notify: false
}, },
Contact::Accepted {
user_id: user_3,
should_notify: false
}
],
); );
// User declines a contact request. Both users see that it is gone. // User declines a contact request. Both users see that it is gone.
@ -1050,21 +1167,33 @@ pub mod tests {
db.respond_to_contact_request(user_3, user_2, false) db.respond_to_contact_request(user_3, user_2, false)
.await .await
.unwrap(); .unwrap();
assert!(!db.has_contact(user_2, user_3).await.unwrap());
assert!(!db.has_contact(user_3, user_2).await.unwrap());
assert_eq!( assert_eq!(
db.get_contacts(user_2).await.unwrap(), db.get_contacts(user_2).await.unwrap(),
Contacts { &[
current: vec![user_1, user_2], Contact::Accepted {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![], should_notify: false
}, },
Contact::Accepted {
user_id: user_2,
should_notify: false
}
]
); );
assert_eq!( assert_eq!(
db.get_contacts(user_3).await.unwrap(), db.get_contacts(user_3).await.unwrap(),
Contacts { &[
current: vec![user_1, user_3], Contact::Accepted {
outgoing_requests: vec![], user_id: user_1,
incoming_requests: vec![], should_notify: false
}, },
Contact::Accepted {
user_id: user_3,
should_notify: false
}
],
); );
} }
} }
@ -1219,40 +1348,51 @@ pub mod tests {
unimplemented!() unimplemented!()
} }
async fn get_contacts(&self, id: UserId) -> Result<Contacts> { async fn get_contacts(&self, id: UserId) -> Result<Vec<Contact>> {
self.background.simulate_random_delay().await; self.background.simulate_random_delay().await;
let mut current = vec![id]; let mut contacts = vec![Contact::Accepted {
let mut outgoing_requests = Vec::new(); user_id: id,
let mut incoming_requests = Vec::new(); should_notify: false,
}];
for contact in self.contacts.lock().iter() { for contact in self.contacts.lock().iter() {
if contact.requester_id == id { if contact.requester_id == id {
if contact.accepted { if contact.accepted {
current.push(contact.responder_id); contacts.push(Contact::Accepted {
user_id: contact.responder_id,
should_notify: contact.should_notify,
});
} else { } else {
outgoing_requests.push(contact.responder_id); contacts.push(Contact::Outgoing {
user_id: contact.responder_id,
});
} }
} else if contact.responder_id == id { } else if contact.responder_id == id {
if contact.accepted { if contact.accepted {
current.push(contact.requester_id); contacts.push(Contact::Accepted {
user_id: contact.requester_id,
should_notify: false,
});
} else { } else {
incoming_requests.push(IncomingContactRequest { contacts.push(Contact::Incoming {
requester_id: contact.requester_id, user_id: contact.requester_id,
should_notify: contact.should_notify, should_notify: contact.should_notify,
}); });
} }
} }
} }
current.sort_unstable(); contacts.sort_unstable_by_key(|contact| contact.user_id());
outgoing_requests.sort_unstable(); Ok(contacts)
incoming_requests.sort_unstable(); }
Ok(Contacts { async fn has_contact(&self, user_id_a: UserId, user_id_b: UserId) -> Result<bool> {
current, self.background.simulate_random_delay().await;
outgoing_requests, Ok(self.contacts.lock().iter().any(|contact| {
incoming_requests, contact.accepted
}) && ((contact.requester_id == user_id_a && contact.responder_id == user_id_b)
|| (contact.requester_id == user_id_b && contact.responder_id == user_id_a))
}))
} }
async fn send_contact_request( async fn send_contact_request(
@ -1274,6 +1414,7 @@ pub mod tests {
Err(anyhow!("contact already exists"))?; Err(anyhow!("contact already exists"))?;
} else { } else {
contact.accepted = true; contact.accepted = true;
contact.should_notify = false;
return Ok(()); return Ok(());
} }
} }
@ -1294,22 +1435,29 @@ pub mod tests {
Ok(()) Ok(())
} }
async fn dismiss_contact_request( async fn dismiss_contact_notification(
&self, &self,
responder_id: UserId, user_id: UserId,
requester_id: UserId, contact_user_id: UserId,
) -> Result<()> { ) -> Result<()> {
let mut contacts = self.contacts.lock(); let mut contacts = self.contacts.lock();
for contact in contacts.iter_mut() { for contact in contacts.iter_mut() {
if contact.requester_id == requester_id && contact.responder_id == responder_id { if contact.requester_id == contact_user_id
if contact.accepted { && contact.responder_id == user_id
return Err(anyhow!("contact already confirmed")); && !contact.accepted
} {
contact.should_notify = false;
return Ok(());
}
if contact.requester_id == user_id
&& contact.responder_id == contact_user_id
&& contact.accepted
{
contact.should_notify = false; contact.should_notify = false;
return Ok(()); return Ok(());
} }
} }
Err(anyhow!("no such contact request")) Err(anyhow!("no such notification"))
} }
async fn respond_to_contact_request( async fn respond_to_contact_request(
@ -1326,6 +1474,7 @@ pub mod tests {
} }
if accept { if accept {
contact.accepted = true; contact.accepted = true;
contact.should_notify = true;
} else { } else {
contacts.remove(ix); contacts.remove(ix);
} }

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

@ -2,7 +2,7 @@ mod store;
use crate::{ use crate::{
auth, auth,
db::{ChannelId, MessageId, UserId}, db::{self, ChannelId, MessageId, UserId},
AppState, Result, AppState, Result,
}; };
use anyhow::anyhow; use anyhow::anyhow;
@ -421,21 +421,27 @@ impl Server {
let contacts = self.app_state.db.get_contacts(user_id).await?; let contacts = self.app_state.db.get_contacts(user_id).await?;
let store = self.store().await; let store = self.store().await;
let updated_contact = store.contact_for_user(user_id); let updated_contact = store.contact_for_user(user_id);
for contact_user_id in contacts.current { for contact in contacts {
for contact_conn_id in store.connection_ids_for_user(contact_user_id) { if let db::Contact::Accepted {
self.peer user_id: contact_user_id,
.send( ..
contact_conn_id, } = contact
proto::UpdateContacts { {
contacts: vec![updated_contact.clone()], for contact_conn_id in store.connection_ids_for_user(contact_user_id) {
remove_contacts: Default::default(), self.peer
incoming_requests: Default::default(), .send(
remove_incoming_requests: Default::default(), contact_conn_id,
outgoing_requests: Default::default(), proto::UpdateContacts {
remove_outgoing_requests: Default::default(), contacts: vec![updated_contact.clone()],
}, remove_contacts: Default::default(),
) incoming_requests: Default::default(),
.trace_err(); remove_incoming_requests: Default::default(),
outgoing_requests: Default::default(),
remove_outgoing_requests: Default::default(),
},
)
.trace_err();
}
} }
} }
Ok(()) Ok(())
@ -473,8 +479,12 @@ impl Server {
guest_user_id = state.user_id_for_connection(request.sender_id)?; guest_user_id = state.user_id_for_connection(request.sender_id)?;
}; };
let guest_contacts = self.app_state.db.get_contacts(guest_user_id).await?; let has_contact = self
if !guest_contacts.current.contains(&host_user_id) { .app_state
.db
.has_contact(guest_user_id, host_user_id)
.await?;
if !has_contact {
return Err(anyhow!("no such project"))?; return Err(anyhow!("no such project"))?;
} }
@ -1026,7 +1036,7 @@ impl Server {
if request.payload.response == proto::ContactRequestResponse::Dismiss as i32 { if request.payload.response == proto::ContactRequestResponse::Dismiss as i32 {
self.app_state self.app_state
.db .db
.dismiss_contact_request(responder_id, requester_id) .dismiss_contact_notification(responder_id, requester_id)
.await?; .await?;
} else { } else {
let accept = request.payload.response == proto::ContactRequestResponse::Accept as i32; let accept = request.payload.response == proto::ContactRequestResponse::Accept as i32;

37
crates/collab/src/rpc/store.rs
View file

@ -217,23 +217,30 @@ impl Store {
.is_empty() .is_empty()
} }
pub fn build_initial_contacts_update(&self, contacts: db::Contacts) -> proto::UpdateContacts { pub fn build_initial_contacts_update(
&self,
contacts: Vec<db::Contact>,
) -> proto::UpdateContacts {
let mut update = proto::UpdateContacts::default(); let mut update = proto::UpdateContacts::default();
for user_id in contacts.current {
update.contacts.push(self.contact_for_user(user_id));
}
for request in contacts.incoming_requests { for contact in contacts {
update match contact {
.incoming_requests db::Contact::Accepted { user_id, .. } => {
.push(proto::IncomingContactRequest { update.contacts.push(self.contact_for_user(user_id));
requester_id: request.requester_id.to_proto(), }
should_notify: request.should_notify, db::Contact::Outgoing { user_id } => {
}) update.outgoing_requests.push(user_id.to_proto())
} }
db::Contact::Incoming {
for requested_user_id in contacts.outgoing_requests { user_id,
update.outgoing_requests.push(requested_user_id.to_proto()) should_notify,
} => update
.incoming_requests
.push(proto::IncomingContactRequest {
requester_id: user_id.to_proto(),
should_notify,
}),
}
} }
update update