WIP: Send the channel name and the channel edges seperately, so we're not repeating them constantly

This commit is currently broken and includes debug data for a failed attempt at rewriting the insert_edge logic
This commit is contained in:
Mikayla 2023-09-15 17:57:23 -07:00
parent 363867c65b
commit 5f9c56c8b0
No known key found for this signature in database
13 changed files with 437 additions and 760 deletions

View file

@ -3,12 +3,25 @@ mod channel_index;
use crate::{channel_buffer::ChannelBuffer, channel_chat::ChannelChat};
use anyhow::{anyhow, Result};
use client::{Client, Subscription, User, UserId, UserStore};
use collections::{hash_map::{self, DefaultHasher}, HashMap, HashSet};
use collections::{
hash_map::{self, DefaultHasher},
HashMap, HashSet,
};
use futures::{channel::mpsc, future::Shared, Future, FutureExt, StreamExt};
use gpui::{AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, Task, WeakModelHandle};
use rpc::{proto, TypedEnvelope};
use rpc::{
proto::{self, ChannelEdge, ChannelPermission},
TypedEnvelope,
};
use serde_derive::{Deserialize, Serialize};
use std::{mem, ops::Deref, sync::Arc, time::Duration, borrow::Cow, hash::{Hash, Hasher}};
use std::{
borrow::Cow,
hash::{Hash, Hasher},
mem,
ops::Deref,
sync::Arc,
time::Duration,
};
use util::ResultExt;
use self::channel_index::ChannelIndex;
@ -301,18 +314,33 @@ impl ChannelStore {
let client = self.client.clone();
let name = name.trim_start_matches("#").to_owned();
cx.spawn(|this, mut cx| async move {
let channel = client
let response = client
.request(proto::CreateChannel { name, parent_id })
.await?
.await?;
let channel = response
.channel
.ok_or_else(|| anyhow!("missing channel in response"))?;
let channel_id = channel.id;
let parent_edge = if let Some(parent_id) = parent_id {
vec![ChannelEdge {
channel_id: channel.id,
parent_id,
}]
} else {
vec![]
};
this.update(&mut cx, |this, cx| {
let task = this.update_channels(
proto::UpdateChannels {
channels: vec![channel],
insert_edge: parent_edge,
channel_permissions: vec![ChannelPermission {
channel_id,
is_admin: true,
}],
..Default::default()
},
cx,
@ -730,6 +758,8 @@ impl ChannelStore {
payload: proto::UpdateChannels,
cx: &mut ModelContext<ChannelStore>,
) -> Option<Task<Result<()>>> {
dbg!(self.client.user_id(), &payload);
if !payload.remove_channel_invitations.is_empty() {
self.channel_invitations
.retain(|channel| !payload.remove_channel_invitations.contains(&channel.id));
@ -752,7 +782,9 @@ impl ChannelStore {
let channels_changed = !payload.channels.is_empty()
|| !payload.delete_channels.is_empty()
|| !payload.insert_edge.is_empty()
|| !payload.delete_edge.is_empty();
if channels_changed {
if !payload.delete_channels.is_empty() {
self.channel_index.delete_channels(&payload.delete_channels);
@ -774,14 +806,20 @@ impl ChannelStore {
}
let mut index_edit = self.channel_index.bulk_edit();
dbg!(&index_edit);
for channel in payload.channels {
index_edit.upsert(channel)
index_edit.insert(channel)
}
for edge in payload.insert_edge {
index_edit.insert_edge(edge.parent_id, edge.channel_id);
}
for edge in payload.delete_edge {
index_edit.delete_edge(edge.parent_id, edge.channel_id);
}
drop(index_edit);
dbg!(&self.channel_index);
}
for permission in payload.channel_permissions {

View file

@ -53,6 +53,7 @@ impl Deref for ChannelIndex {
/// A guard for ensuring that the paths index maintains its sort and uniqueness
/// invariants after a series of insertions
#[derive(Debug)]
pub struct ChannelPathsEditGuard<'a> {
paths: &'a mut Vec<ChannelPath>,
channels_by_id: &'a mut ChannelsById,
@ -78,42 +79,81 @@ impl<'a> ChannelPathsEditGuard<'a> {
}
}
pub fn upsert(&mut self, channel_proto: proto::Channel) {
pub fn insert(&mut self, channel_proto: proto::Channel) {
if let Some(existing_channel) = self.channels_by_id.get_mut(&channel_proto.id) {
Arc::make_mut(existing_channel).name = channel_proto.name;
if let Some(parent_id) = channel_proto.parent_id {
self.insert_edge(parent_id, channel_proto.id)
}
} else {
let channel = Arc::new(Channel {
id: channel_proto.id,
name: channel_proto.name,
});
self.channels_by_id.insert(channel.id, channel.clone());
if let Some(parent_id) = channel_proto.parent_id {
self.insert_edge(parent_id, channel.id);
} else {
self.insert_root(channel.id);
}
self.channels_by_id.insert(
channel_proto.id,
Arc::new(Channel {
id: channel_proto.id,
name: channel_proto.name,
}),
);
self.insert_root(channel_proto.id);
}
}
fn insert_edge(&mut self, parent_id: ChannelId, channel_id: ChannelId) {
pub fn insert_edge(&mut self, parent_id: ChannelId, channel_id: ChannelId) {
debug_assert!(self.channels_by_id.contains_key(&parent_id));
let mut ix = 0;
println!("*********** INSERTING EDGE {}, {} ***********", channel_id, parent_id);
dbg!(&self.paths);
while ix < self.paths.len() {
let path = &self.paths[ix];
println!("*********");
dbg!(path);
if path.ends_with(&[parent_id]) {
let mut new_path = path.to_vec();
dbg!("Appending to parent path");
let mut new_path = Vec::with_capacity(path.len() + 1);
new_path.extend_from_slice(path);
new_path.push(channel_id);
self.paths.insert(ix + 1, ChannelPath::new(new_path.into()));
self.paths.insert(ix + 1, dbg!(ChannelPath::new(new_path.into())));
ix += 2;
} else if path.get(0) == Some(&channel_id) {
// Clear out any paths that have this chahnnel as their root
self.paths.swap_remove(ix);
} else if let Some(path_ix) = path.iter().position(|c| c == &channel_id) {
if path.contains(&parent_id) {
dbg!("Doing nothing");
ix += 1;
continue;
}
if path_ix == 0 && path.len() == 1 {
dbg!("Removing path that is just this");
self.paths.swap_remove(ix);
continue;
}
// This is the busted section rn
// We're trying to do this weird, unsorted context
// free insertion thing, but we can't insert 'parent_id',
// we have to _prepend_ with _parent path to_,
// or something like that.
// It's a bit busted rn, I think I need to keep this whole thing
// sorted now, as this is a huge mess.
// Basically, we want to do the exact thing we do in the
// server, except explicitly.
// Also, rethink the bulk edit abstraction, it's use may no longer
// be as needed with the channel names and edges seperated.
dbg!("Expanding path which contains");
let (left, right) = path.split_at(path_ix);
let mut new_path = Vec::with_capacity(left.len() + right.len() + 1);
/// WRONG WRONG WRONG
new_path.extend_from_slice(left);
new_path.push(parent_id);
/// WRONG WRONG WRONG
new_path.extend_from_slice(right);
if path_ix == 0 {
dbg!("Replacing path that starts with this");
self.paths[ix] = dbg!(ChannelPath::new(new_path.into()));
} else {
dbg!("inserting new path");
self.paths.insert(ix + 1, dbg!(ChannelPath::new(new_path.into())));
ix += 1;
}
ix += 1;
} else {
dbg!("Doing nothing");
ix += 1;
}
}

View file

@ -18,12 +18,11 @@ fn test_update_channels(cx: &mut AppContext) {
proto::Channel {
id: 1,
name: "b".to_string(),
parent_id: None,
},
proto::Channel {
id: 2,
name: "a".to_string(),
parent_id: None,
},
],
channel_permissions: vec![proto::ChannelPermission {
@ -51,12 +50,20 @@ fn test_update_channels(cx: &mut AppContext) {
proto::Channel {
id: 3,
name: "x".to_string(),
parent_id: Some(1),
},
proto::Channel {
id: 4,
name: "y".to_string(),
parent_id: Some(2),
},
],
insert_edge: vec![
proto::ChannelEdge {
parent_id: 1,
channel_id: 3,
},
proto::ChannelEdge {
parent_id: 2,
channel_id: 4,
},
],
..Default::default()
@ -86,17 +93,24 @@ fn test_dangling_channel_paths(cx: &mut AppContext) {
proto::Channel {
id: 0,
name: "a".to_string(),
parent_id: None,
},
proto::Channel {
id: 1,
name: "b".to_string(),
parent_id: Some(0),
},
proto::Channel {
id: 2,
name: "c".to_string(),
parent_id: Some(1),
},
],
insert_edge: vec![
proto::ChannelEdge {
parent_id: 0,
channel_id: 1,
},
proto::ChannelEdge {
parent_id: 1,
channel_id: 2,
},
],
channel_permissions: vec![proto::ChannelPermission {
@ -145,7 +159,6 @@ async fn test_channel_messages(cx: &mut TestAppContext) {
channels: vec![proto::Channel {
id: channel_id,
name: "the-channel".to_string(),
parent_id: None,
}],
..Default::default()
});

View file

@ -14,7 +14,7 @@ use collections::{BTreeMap, HashMap, HashSet};
use dashmap::DashMap;
use futures::StreamExt;
use rand::{prelude::StdRng, Rng, SeedableRng};
use rpc::{proto, ConnectionId};
use rpc::{proto::{self}, ConnectionId};
use sea_orm::{
entity::prelude::*, ActiveValue, Condition, ConnectionTrait, DatabaseConnection,
DatabaseTransaction, DbErr, FromQueryResult, IntoActiveModel, IsolationLevel, JoinType,
@ -43,6 +43,8 @@ pub use ids::*;
pub use sea_orm::ConnectOptions;
pub use tables::user::Model as User;
use self::queries::channels::ChannelGraph;
pub struct Database {
options: ConnectOptions,
pool: DatabaseConnection,
@ -421,16 +423,15 @@ pub struct NewUserResult {
pub signup_device_id: Option<String>,
}
#[derive(FromQueryResult, Debug, PartialEq)]
#[derive(FromQueryResult, Debug, PartialEq, Eq, Hash)]
pub struct Channel {
pub id: ChannelId,
pub name: String,
pub parent_id: Option<ChannelId>,
}
#[derive(Debug, PartialEq)]
pub struct ChannelsForUser {
pub channels: Vec<Channel>,
pub channels: ChannelGraph,
pub channel_participants: HashMap<ChannelId, Vec<UserId>>,
pub channels_with_admin_privileges: HashSet<ChannelId>,
}

View file

@ -1,3 +1,4 @@
use rpc::proto::ChannelEdge;
use smallvec::SmallVec;
use super::*;
@ -326,7 +327,6 @@ impl Database {
.map(|channel| Channel {
id: channel.id,
name: channel.name,
parent_id: None,
})
.collect();
@ -335,12 +335,12 @@ impl Database {
.await
}
async fn get_channels_internal(
async fn get_channel_graph(
&self,
parents_by_child_id: ChannelDescendants,
trim_dangling_parents: bool,
tx: &DatabaseTransaction,
) -> Result<Vec<Channel>> {
) -> Result<ChannelGraph> {
let mut channels = Vec::with_capacity(parents_by_child_id.len());
{
let mut rows = channel::Entity::find()
@ -349,49 +349,36 @@ impl Database {
.await?;
while let Some(row) = rows.next().await {
let row = row?;
// As these rows are pulled from the map's keys, this unwrap is safe.
let parents = parents_by_child_id.get(&row.id).unwrap();
if parents.len() > 0 {
let mut added_channel = false;
for parent in parents.iter() {
// Trim out any dangling parent pointers.
// That the user doesn't have access to
if trim_dangling_parents {
if parents_by_child_id.contains_key(parent) {
added_channel = true;
channels.push(Channel {
id: row.id,
name: row.name.clone(),
parent_id: Some(*parent),
});
}
} else {
added_channel = true;
channels.push(Channel {
id: row.id,
name: row.name.clone(),
parent_id: Some(*parent),
});
}
}
if !added_channel {
channels.push(Channel {
id: row.id,
name: row.name,
parent_id: None,
channels.push(Channel {
id: row.id,
name: row.name,
})
}
}
let mut edges = Vec::with_capacity(parents_by_child_id.len());
for (channel, parents) in parents_by_child_id.iter() {
for parent in parents.into_iter() {
if trim_dangling_parents {
if parents_by_child_id.contains_key(parent) {
edges.push(ChannelEdge {
channel_id: channel.to_proto(),
parent_id: parent.to_proto(),
});
}
} else {
channels.push(Channel {
id: row.id,
name: row.name,
parent_id: None,
edges.push(ChannelEdge {
channel_id: channel.to_proto(),
parent_id: parent.to_proto(),
});
}
}
}
Ok(channels)
Ok(ChannelGraph {
channels,
edges,
})
}
pub async fn get_channels_for_user(&self, user_id: UserId) -> Result<ChannelsForUser> {
@ -407,51 +394,74 @@ impl Database {
.all(&*tx)
.await?;
let parents_by_child_id = self
.get_channel_descendants(channel_memberships.iter().map(|m| m.channel_id), &*tx)
.await?;
let channels_with_admin_privileges = channel_memberships
.iter()
.filter_map(|membership| membership.admin.then_some(membership.channel_id))
.collect();
let channels = self
.get_channels_internal(parents_by_child_id, true, &tx)
.await?;
#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
enum QueryUserIdsAndChannelIds {
ChannelId,
UserId,
}
let mut channel_participants: HashMap<ChannelId, Vec<UserId>> = HashMap::default();
{
let mut rows = room_participant::Entity::find()
.inner_join(room::Entity)
.filter(room::Column::ChannelId.is_in(channels.iter().map(|c| c.id)))
.select_only()
.column(room::Column::ChannelId)
.column(room_participant::Column::UserId)
.into_values::<_, QueryUserIdsAndChannelIds>()
.stream(&*tx)
.await?;
while let Some(row) = rows.next().await {
let row: (ChannelId, UserId) = row?;
channel_participants.entry(row.0).or_default().push(row.1)
}
}
Ok(ChannelsForUser {
channels,
channel_participants,
channels_with_admin_privileges,
})
self.get_user_channels(channel_memberships, user_id, &tx).await
})
.await
}
pub async fn get_channel_for_user(&self, channel_id: ChannelId, user_id: UserId) -> Result<ChannelsForUser> {
self.transaction(|tx| async move {
let tx = tx;
let channel_membership = channel_member::Entity::find()
.filter(
channel_member::Column::UserId
.eq(user_id)
.and(channel_member::Column::ChannelId.eq(channel_id))
.and(channel_member::Column::Accepted.eq(true)),
)
.all(&*tx)
.await?;
self.get_user_channels(channel_membership, user_id, &tx).await
})
.await
}
pub async fn get_user_channels(&self, channel_memberships: Vec<channel_member::Model>, user_id: UserId, tx: &DatabaseTransaction) -> Result<ChannelsForUser> {
let parents_by_child_id = self
.get_channel_descendants(channel_memberships.iter().map(|m| m.channel_id), &*tx)
.await?;
let channels_with_admin_privileges = channel_memberships
.iter()
.filter_map(|membership| membership.admin.then_some(membership.channel_id))
.collect();
let graph = self
.get_channel_graph(parents_by_child_id, true, &tx)
.await?;
#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
enum QueryUserIdsAndChannelIds {
ChannelId,
UserId,
}
let mut channel_participants: HashMap<ChannelId, Vec<UserId>> = HashMap::default();
{
let mut rows = room_participant::Entity::find()
.inner_join(room::Entity)
.filter(room::Column::ChannelId.is_in(graph.channels.iter().map(|c| c.id)))
.select_only()
.column(room::Column::ChannelId)
.column(room_participant::Column::UserId)
.into_values::<_, QueryUserIdsAndChannelIds>()
.stream(&*tx)
.await?;
while let Some(row) = rows.next().await {
let row: (ChannelId, UserId) = row?;
channel_participants.entry(row.0).or_default().push(row.1)
}
}
Ok(ChannelsForUser {
channels: graph,
channel_participants,
channels_with_admin_privileges,
})
}
pub async fn get_channel_members(&self, id: ChannelId) -> Result<Vec<UserId>> {
self.transaction(|tx| async move { self.get_channel_members_internal(id, &*tx).await })
.await
@ -759,7 +769,6 @@ impl Database {
Channel {
id: channel.id,
name: channel.name,
parent_id: None,
},
is_accepted,
)))
@ -792,7 +801,7 @@ impl Database {
user: UserId,
channel: ChannelId,
to: ChannelId,
) -> Result<Vec<Channel>> {
) -> Result<ChannelGraph> {
self.transaction(|tx| async move {
// Note that even with these maxed permissions, this linking operation
// is still insecure because you can't remove someone's permissions to a
@ -811,7 +820,7 @@ impl Database {
channel: ChannelId,
to: ChannelId,
tx: &DatabaseTransaction,
) -> Result<Vec<Channel>> {
) -> Result<ChannelGraph> {
self.check_user_is_channel_admin(to, user, &*tx).await?;
let to_ancestors = self.get_channel_ancestors(to, &*tx).await?;
@ -881,7 +890,7 @@ impl Database {
}
let channels = self
.get_channels_internal(from_descendants, false, &*tx)
.get_channel_graph(from_descendants, false, &*tx)
.await?;
Ok(channels)
@ -961,7 +970,7 @@ impl Database {
channel: ChannelId,
from: ChannelId,
to: ChannelId,
) -> Result<Vec<Channel>> {
) -> Result<ChannelGraph> {
self.transaction(|tx| async move {
self.check_user_is_channel_admin(channel, user, &*tx)
.await?;
@ -982,6 +991,39 @@ enum QueryUserIds {
UserId,
}
#[derive(Debug)]
pub struct ChannelGraph {
pub channels: Vec<Channel>,
pub edges: Vec<ChannelEdge>,
}
impl ChannelGraph {
pub fn is_empty(&self) -> bool {
self.channels.is_empty() && self.edges.is_empty()
}
}
#[cfg(test)]
impl PartialEq for ChannelGraph {
fn eq(&self, other: &Self) -> bool {
// Order independent comparison for tests
let channels_set = self.channels.iter().collect::<HashSet<_>>();
let other_channels_set = other.channels.iter().collect::<HashSet<_>>();
let edges_set = self.edges.iter().map(|edge| (edge.channel_id, edge.parent_id)).collect::<HashSet<_>>();
let other_edges_set = other.edges.iter().map(|edge| (edge.channel_id, edge.parent_id)).collect::<HashSet<_>>();
channels_set == other_channels_set && edges_set == other_edges_set
}
}
#[cfg(not(test))]
impl PartialEq for ChannelGraph {
fn eq(&self, other: &Self) -> bool {
self.channels == other.channels && self.edges == other.edges
}
}
struct SmallSet<T>(SmallVec<[T; 1]>);
impl<T> Deref for SmallSet<T> {
@ -1008,7 +1050,7 @@ impl<T> SmallSet<T> {
Err(ix) => {
self.0.insert(ix, value);
true
},
}
}
}

View file

@ -7,6 +7,7 @@ mod message_tests;
use super::*;
use gpui::executor::Background;
use parking_lot::Mutex;
use rpc::proto::ChannelEdge;
use sea_orm::ConnectionTrait;
use sqlx::migrate::MigrateDatabase;
use std::sync::Arc;
@ -144,3 +145,27 @@ impl Drop for TestDb {
}
}
}
/// The second tuples are (channel_id, parent)
fn graph(channels: &[(ChannelId, &'static str)], edges: &[(ChannelId, ChannelId)]) -> ChannelGraph {
let mut graph = ChannelGraph {
channels: vec![],
edges: vec![],
};
for (id, name) in channels {
graph.channels.push(Channel {
id: *id,
name: name.to_string(),
})
}
for (channel, parent) in edges {
graph.edges.push(ChannelEdge {
channel_id: channel.to_proto(),
parent_id: parent.to_proto(),
})
}
graph
}

View file

@ -1,8 +1,11 @@
use collections::{HashMap, HashSet};
use rpc::{proto, ConnectionId};
use rpc::{
proto::{self},
ConnectionId,
};
use crate::{
db::{Channel, ChannelId, Database, NewUserParams},
db::{queries::channels::ChannelGraph, ChannelId, Database, NewUserParams, tests::graph},
test_both_dbs,
};
use std::sync::Arc;
@ -82,70 +85,42 @@ async fn test_channels(db: &Arc<Database>) {
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_eq!(
result.channels,
vec![
Channel {
id: zed_id,
name: "zed".to_string(),
parent_id: None,
},
Channel {
id: crdb_id,
name: "crdb".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: replace_id,
name: "replace".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: rust_id,
name: "rust".to_string(),
parent_id: None,
},
Channel {
id: cargo_id,
name: "cargo".to_string(),
parent_id: Some(rust_id),
},
Channel {
id: cargo_ra_id,
name: "cargo-ra".to_string(),
parent_id: Some(cargo_id),
}
]
graph(
&[
(zed_id, "zed"),
(crdb_id, "crdb"),
(livestreaming_id, "livestreaming"),
(replace_id, "replace"),
(rust_id, "rust"),
(cargo_id, "cargo"),
(cargo_ra_id, "cargo-ra")
],
&[
(crdb_id, zed_id),
(livestreaming_id, zed_id),
(replace_id, zed_id),
(cargo_id, rust_id),
(cargo_ra_id, cargo_id),
]
)
);
let result = db.get_channels_for_user(b_id).await.unwrap();
assert_eq!(
result.channels,
vec![
Channel {
id: zed_id,
name: "zed".to_string(),
parent_id: None,
},
Channel {
id: crdb_id,
name: "crdb".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: replace_id,
name: "replace".to_string(),
parent_id: Some(zed_id),
},
]
graph(
&[
(zed_id, "zed"),
(crdb_id, "crdb"),
(livestreaming_id, "livestreaming"),
(replace_id, "replace")
],
&[
(crdb_id, zed_id),
(livestreaming_id, zed_id),
(replace_id, zed_id)
]
)
);
// Update member permissions
@ -157,28 +132,19 @@ async fn test_channels(db: &Arc<Database>) {
let result = db.get_channels_for_user(b_id).await.unwrap();
assert_eq!(
result.channels,
vec![
Channel {
id: zed_id,
name: "zed".to_string(),
parent_id: None,
},
Channel {
id: crdb_id,
name: "crdb".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: replace_id,
name: "replace".to_string(),
parent_id: Some(zed_id),
},
]
graph(
&[
(zed_id, "zed"),
(crdb_id, "crdb"),
(livestreaming_id, "livestreaming"),
(replace_id, "replace")
],
&[
(crdb_id, zed_id),
(livestreaming_id, zed_id),
(replace_id, zed_id)
]
)
);
// Remove a single channel
@ -594,11 +560,10 @@ async fn test_db_channel_moving(db: &Arc<Database>) {
// Not using the assert_dag helper because we want to make sure we're returning the full data
pretty_assertions::assert_eq!(
returned_channels,
vec![Channel {
id: livestreaming_dag_sub_id,
name: "livestreaming_dag_sub".to_string(),
parent_id: Some(livestreaming_id),
}]
graph(
&[(livestreaming_dag_sub_id, "livestreaming_dag_sub")],
&[(livestreaming_dag_sub_id, livestreaming_id)]
)
);
let result = db.get_channels_for_user(a_id).await.unwrap();
@ -631,28 +596,19 @@ async fn test_db_channel_moving(db: &Arc<Database>) {
// Make sure that we're correctly getting the full sub-dag
pretty_assertions::assert_eq!(
returned_channels,
vec![
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(gpui2_id),
},
Channel {
id: livestreaming_dag_id,
name: "livestreaming_dag".to_string(),
parent_id: Some(livestreaming_id),
},
Channel {
id: livestreaming_dag_sub_id,
name: "livestreaming_dag_sub".to_string(),
parent_id: Some(livestreaming_id),
},
Channel {
id: livestreaming_dag_sub_id,
name: "livestreaming_dag_sub".to_string(),
parent_id: Some(livestreaming_dag_id),
}
]
graph(
&[
(livestreaming_id, "livestreaming"),
(livestreaming_dag_id, "livestreaming_dag"),
(livestreaming_dag_sub_id, "livestreaming_dag_sub"),
],
&[
(livestreaming_id, gpui2_id),
(livestreaming_dag_id, livestreaming_id),
(livestreaming_dag_sub_id, livestreaming_id),
(livestreaming_dag_sub_id, livestreaming_dag_id),
]
)
);
let result = db.get_channels_for_user(a_id).await.unwrap();
@ -836,26 +792,26 @@ async fn test_db_channel_moving(db: &Arc<Database>) {
}
#[track_caller]
fn assert_dag(actual: Vec<Channel>, expected: &[(ChannelId, Option<ChannelId>)]) {
/// This is used to allow tests to be ordering independent
fn make_parents_map(association_table: impl IntoIterator<Item= (ChannelId, Option<ChannelId>)>) -> HashMap<ChannelId, HashSet<ChannelId>> {
let mut map: HashMap<ChannelId, HashSet<ChannelId>> = HashMap::default();
for (child, parent) in association_table {
let entry = map.entry(child).or_default();
if let Some(parent) = parent {
entry.insert(parent);
}
}
map
fn assert_dag(actual: ChannelGraph, expected: &[(ChannelId, Option<ChannelId>)]) {
let mut actual_map: HashMap<ChannelId, HashSet<ChannelId>> = HashMap::default();
for channel in actual.channels {
actual_map.insert(channel.id, HashSet::default());
}
for edge in actual.edges {
actual_map
.get_mut(&ChannelId::from_proto(edge.channel_id))
.unwrap()
.insert(ChannelId::from_proto(edge.parent_id));
}
let actual = actual
.iter()
.map(|channel| (channel.id, channel.parent_id));
let actual_map = make_parents_map(actual);
let expected_map = make_parents_map(expected.iter().copied());
let mut expected_map: HashMap<ChannelId, HashSet<ChannelId>> = HashMap::default();
for (child, parent) in expected {
let entry = expected_map.entry(*child).or_default();
if let Some(parent) = parent {
entry.insert(*parent);
}
}
pretty_assertions::assert_eq!(actual_map, expected_map)
}

View file

@ -575,458 +575,6 @@ async fn test_fuzzy_search_users() {
}
}
test_both_dbs!(test_channels, test_channels_postgres, test_channels_sqlite);
async fn test_channels(db: &Arc<Database>) {
let a_id = db
.create_user(
"user1@example.com",
false,
NewUserParams {
github_login: "user1".into(),
github_user_id: 5,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let b_id = db
.create_user(
"user2@example.com",
false,
NewUserParams {
github_login: "user2".into(),
github_user_id: 6,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let zed_id = db.create_root_channel("zed", "1", a_id).await.unwrap();
// Make sure that people cannot read channels they haven't been invited to
assert!(db.get_channel(zed_id, b_id).await.unwrap().is_none());
db.invite_channel_member(zed_id, b_id, a_id, false)
.await
.unwrap();
db.respond_to_channel_invite(zed_id, b_id, true)
.await
.unwrap();
let crdb_id = db
.create_channel("crdb", Some(zed_id), "2", a_id)
.await
.unwrap();
let livestreaming_id = db
.create_channel("livestreaming", Some(zed_id), "3", a_id)
.await
.unwrap();
let replace_id = db
.create_channel("replace", Some(zed_id), "4", a_id)
.await
.unwrap();
let mut members = db.get_channel_members(replace_id).await.unwrap();
members.sort();
assert_eq!(members, &[a_id, b_id]);
let rust_id = db.create_root_channel("rust", "5", a_id).await.unwrap();
let cargo_id = db
.create_channel("cargo", Some(rust_id), "6", a_id)
.await
.unwrap();
let cargo_ra_id = db
.create_channel("cargo-ra", Some(cargo_id), "7", a_id)
.await
.unwrap();
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_eq!(
result.channels,
vec![
Channel {
id: zed_id,
name: "zed".to_string(),
parent_id: None,
},
Channel {
id: crdb_id,
name: "crdb".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: replace_id,
name: "replace".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: rust_id,
name: "rust".to_string(),
parent_id: None,
},
Channel {
id: cargo_id,
name: "cargo".to_string(),
parent_id: Some(rust_id),
},
Channel {
id: cargo_ra_id,
name: "cargo-ra".to_string(),
parent_id: Some(cargo_id),
}
]
);
let result = db.get_channels_for_user(b_id).await.unwrap();
assert_eq!(
result.channels,
vec![
Channel {
id: zed_id,
name: "zed".to_string(),
parent_id: None,
},
Channel {
id: crdb_id,
name: "crdb".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: replace_id,
name: "replace".to_string(),
parent_id: Some(zed_id),
},
]
);
// Update member permissions
let set_subchannel_admin = db.set_channel_member_admin(crdb_id, a_id, b_id, true).await;
assert!(set_subchannel_admin.is_err());
let set_channel_admin = db.set_channel_member_admin(zed_id, a_id, b_id, true).await;
assert!(set_channel_admin.is_ok());
let result = db.get_channels_for_user(b_id).await.unwrap();
assert_eq!(
result.channels,
vec![
Channel {
id: zed_id,
name: "zed".to_string(),
parent_id: None,
},
Channel {
id: crdb_id,
name: "crdb".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: livestreaming_id,
name: "livestreaming".to_string(),
parent_id: Some(zed_id),
},
Channel {
id: replace_id,
name: "replace".to_string(),
parent_id: Some(zed_id),
},
]
);
// Remove a single channel
db.delete_channel(crdb_id, a_id).await.unwrap();
assert!(db.get_channel(crdb_id, a_id).await.unwrap().is_none());
// Remove a channel tree
let (mut channel_ids, user_ids) = db.delete_channel(rust_id, a_id).await.unwrap();
channel_ids.sort();
assert_eq!(channel_ids, &[rust_id, cargo_id, cargo_ra_id]);
assert_eq!(user_ids, &[a_id]);
assert!(db.get_channel(rust_id, a_id).await.unwrap().is_none());
assert!(db.get_channel(cargo_id, a_id).await.unwrap().is_none());
assert!(db.get_channel(cargo_ra_id, a_id).await.unwrap().is_none());
}
test_both_dbs!(
test_joining_channels,
test_joining_channels_postgres,
test_joining_channels_sqlite
);
async fn test_joining_channels(db: &Arc<Database>) {
let owner_id = db.create_server("test").await.unwrap().0 as u32;
let user_1 = db
.create_user(
"user1@example.com",
false,
NewUserParams {
github_login: "user1".into(),
github_user_id: 5,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let user_2 = db
.create_user(
"user2@example.com",
false,
NewUserParams {
github_login: "user2".into(),
github_user_id: 6,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let channel_1 = db
.create_root_channel("channel_1", "1", user_1)
.await
.unwrap();
let room_1 = db.room_id_for_channel(channel_1).await.unwrap();
// can join a room with membership to its channel
let joined_room = db
.join_room(room_1, user_1, ConnectionId { owner_id, id: 1 })
.await
.unwrap();
assert_eq!(joined_room.room.participants.len(), 1);
drop(joined_room);
// cannot join a room without membership to its channel
assert!(db
.join_room(room_1, user_2, ConnectionId { owner_id, id: 1 })
.await
.is_err());
}
test_both_dbs!(
test_channel_invites,
test_channel_invites_postgres,
test_channel_invites_sqlite
);
async fn test_channel_invites(db: &Arc<Database>) {
db.create_server("test").await.unwrap();
let user_1 = db
.create_user(
"user1@example.com",
false,
NewUserParams {
github_login: "user1".into(),
github_user_id: 5,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let user_2 = db
.create_user(
"user2@example.com",
false,
NewUserParams {
github_login: "user2".into(),
github_user_id: 6,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let user_3 = db
.create_user(
"user3@example.com",
false,
NewUserParams {
github_login: "user3".into(),
github_user_id: 7,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let channel_1_1 = db
.create_root_channel("channel_1", "1", user_1)
.await
.unwrap();
let channel_1_2 = db
.create_root_channel("channel_2", "2", user_1)
.await
.unwrap();
db.invite_channel_member(channel_1_1, user_2, user_1, false)
.await
.unwrap();
db.invite_channel_member(channel_1_2, user_2, user_1, false)
.await
.unwrap();
db.invite_channel_member(channel_1_1, user_3, user_1, true)
.await
.unwrap();
let user_2_invites = db
.get_channel_invites_for_user(user_2) // -> [channel_1_1, channel_1_2]
.await
.unwrap()
.into_iter()
.map(|channel| channel.id)
.collect::<Vec<_>>();
assert_eq!(user_2_invites, &[channel_1_1, channel_1_2]);
let user_3_invites = db
.get_channel_invites_for_user(user_3) // -> [channel_1_1]
.await
.unwrap()
.into_iter()
.map(|channel| channel.id)
.collect::<Vec<_>>();
assert_eq!(user_3_invites, &[channel_1_1]);
let members = db
.get_channel_member_details(channel_1_1, user_1)
.await
.unwrap();
assert_eq!(
members,
&[
proto::ChannelMember {
user_id: user_1.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
admin: true,
},
proto::ChannelMember {
user_id: user_2.to_proto(),
kind: proto::channel_member::Kind::Invitee.into(),
admin: false,
},
proto::ChannelMember {
user_id: user_3.to_proto(),
kind: proto::channel_member::Kind::Invitee.into(),
admin: true,
},
]
);
db.respond_to_channel_invite(channel_1_1, user_2, true)
.await
.unwrap();
let channel_1_3 = db
.create_channel("channel_3", Some(channel_1_1), "1", user_1)
.await
.unwrap();
let members = db
.get_channel_member_details(channel_1_3, user_1)
.await
.unwrap();
assert_eq!(
members,
&[
proto::ChannelMember {
user_id: user_1.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
admin: true,
},
proto::ChannelMember {
user_id: user_2.to_proto(),
kind: proto::channel_member::Kind::AncestorMember.into(),
admin: false,
},
]
);
}
test_both_dbs!(
test_channel_renames,
test_channel_renames_postgres,
test_channel_renames_sqlite
);
async fn test_channel_renames(db: &Arc<Database>) {
db.create_server("test").await.unwrap();
let user_1 = db
.create_user(
"user1@example.com",
false,
NewUserParams {
github_login: "user1".into(),
github_user_id: 5,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let user_2 = db
.create_user(
"user2@example.com",
false,
NewUserParams {
github_login: "user2".into(),
github_user_id: 6,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let zed_id = db.create_root_channel("zed", "1", user_1).await.unwrap();
db.rename_channel(zed_id, user_1, "#zed-archive")
.await
.unwrap();
let zed_archive_id = zed_id;
let (channel, _) = db
.get_channel(zed_archive_id, user_1)
.await
.unwrap()
.unwrap();
assert_eq!(channel.name, "zed-archive");
let non_permissioned_rename = db
.rename_channel(zed_archive_id, user_2, "hacked-lol")
.await;
assert!(non_permissioned_rename.is_err());
let bad_name_rename = db.rename_channel(zed_id, user_1, "#").await;
assert!(bad_name_rename.is_err())
}
fn build_background_executor() -> Arc<Background> {
Deterministic::new(0).build_background()
}

View file

@ -3,7 +3,7 @@ mod connection_pool;
use crate::{
auth,
db::{
self, Channel, ChannelId, ChannelsForUser, Database, MessageId, ProjectId, RoomId,
self, ChannelId, ChannelsForUser, Database, MessageId, ProjectId, RoomId,
ServerId, User, UserId,
},
executor::Executor,
@ -39,7 +39,7 @@ use prometheus::{register_int_gauge, IntGauge};
use rpc::{
proto::{
self, Ack, AddChannelBufferCollaborator, AnyTypedEnvelope, EntityMessage, EnvelopedMessage,
LiveKitConnectionInfo, RequestMessage,
LiveKitConnectionInfo, RequestMessage, ChannelEdge,
},
Connection, ConnectionId, Peer, Receipt, TypedEnvelope,
};
@ -2200,33 +2200,38 @@ async fn create_channel(
let channel = proto::Channel {
id: id.to_proto(),
name: request.name,
parent_id: request.parent_id,
};
response.send(proto::ChannelResponse {
response.send(proto::CreateChannelResponse {
channel: Some(channel.clone()),
parent_id: request.parent_id,
})?;
let mut update = proto::UpdateChannels::default();
update.channels.push(channel);
let user_ids_to_notify = if let Some(parent_id) = parent_id {
db.get_channel_members(parent_id).await?
} else {
vec![session.user_id]
let Some(parent_id) = parent_id else {
return Ok(());
};
let update = proto::UpdateChannels {
channels: vec![channel],
insert_edge: vec![
ChannelEdge {
parent_id: parent_id.to_proto(),
channel_id: id.to_proto(),
}
],
..Default::default()
};
let user_ids_to_notify =
db.get_channel_members(parent_id).await?;
let connection_pool = session.connection_pool().await;
for user_id in user_ids_to_notify {
for connection_id in connection_pool.user_connection_ids(user_id) {
let mut update = update.clone();
if user_id == session.user_id {
update.channel_permissions.push(proto::ChannelPermission {
channel_id: id.to_proto(),
is_admin: true,
});
continue;
}
session.peer.send(connection_id, update)?;
session.peer.send(connection_id, update.clone())?;
}
}
@ -2282,7 +2287,6 @@ async fn invite_channel_member(
update.channel_invitations.push(proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
parent_id: None,
});
for connection_id in session
.connection_pool()
@ -2373,9 +2377,8 @@ async fn rename_channel(
let channel = proto::Channel {
id: request.channel_id,
name: new_name,
parent_id: None,
};
response.send(proto::ChannelResponse {
response.send(proto::RenameChannelResponse {
channel: Some(channel.clone()),
})?;
let mut update = proto::UpdateChannels::default();
@ -2407,13 +2410,14 @@ async fn link_channel(
let connection_pool = session.connection_pool().await;
let update = proto::UpdateChannels {
channels: channels_to_send
.channels
.into_iter()
.map(|channel| proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
parent_id: channel.parent_id.map(ChannelId::to_proto),
})
.collect(),
insert_edge: channels_to_send.edges,
..Default::default()
};
for member_id in members {
@ -2469,13 +2473,12 @@ async fn move_channel(
let from_parent = ChannelId::from_proto(request.from);
let to = ChannelId::from_proto(request.to);
let members_from = db.get_channel_members(channel_id).await?;
let channels_to_send: Vec<Channel> = db
let channels_to_send = db
.move_channel(session.user_id, channel_id, from_parent, to)
.await?;
let members_to = db.get_channel_members(channel_id).await?;
let members_from = db.get_channel_members(from_parent).await?;
let members_to = db.get_channel_members(to).await?;
let update = proto::UpdateChannels {
delete_edge: vec![proto::ChannelEdge {
@ -2493,13 +2496,14 @@ async fn move_channel(
let update = proto::UpdateChannels {
channels: channels_to_send
.channels
.into_iter()
.map(|channel| proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
parent_id: channel.parent_id.map(ChannelId::to_proto),
})
.collect(),
insert_edge: channels_to_send.edges,
..Default::default()
};
for member_id in members_to {
@ -2542,14 +2546,14 @@ async fn respond_to_channel_invite(
.remove_channel_invitations
.push(channel_id.to_proto());
if request.accept {
let result = db.get_channels_for_user(session.user_id).await?;
let result = db.get_channel_for_user(channel_id, session.user_id).await?;
update
.channels
.extend(result.channels.into_iter().map(|channel| proto::Channel {
.extend(result.channels.channels.into_iter().map(|channel| proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
parent_id: channel.parent_id.map(ChannelId::to_proto),
}));
update.insert_edge = result.channels.edges;
update
.channel_participants
.extend(
@ -2967,14 +2971,15 @@ fn build_initial_channels_update(
) -> proto::UpdateChannels {
let mut update = proto::UpdateChannels::default();
for channel in channels.channels {
for channel in channels.channels.channels{
update.channels.push(proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
parent_id: channel.parent_id.map(|id| id.to_proto()),
});
}
update.insert_edge = channels.channels.edges;
for (channel_id, participants) in channels.channel_participants {
update
.channel_participants
@ -3000,7 +3005,6 @@ fn build_initial_channels_update(
update.channel_invitations.push(proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
parent_id: None,
});
}

View file

@ -142,6 +142,8 @@ async fn test_core_channels(
],
);
println!("STARTING CREATE CHANNEL C");
let channel_c_id = client_a
.channel_store()
.update(cx_a, |channel_store, cx| {
@ -994,10 +996,6 @@ async fn test_channel_moving(
.add_admin_to_channel((&client_b, cx_b), channel_b_id, cx_a)
.await;
client_b
.add_admin_to_channel((&client_c, cx_c), channel_ep_id, cx_b)
.await;
// Current shape for B:
// /- ep
// mu -- ga
@ -1019,10 +1017,18 @@ async fn test_channel_moving(
],
);
client_b
.add_admin_to_channel((&client_c, cx_c), channel_ep_id, cx_b)
.await;
// Current shape for C:
// - ep
assert_channels_list_shape(client_c.channel_store(), cx_c, &[(channel_ep_id, 0)]);
println!("*******************************************");
println!("********** STARTING LINK CHANNEL **********");
println!("*******************************************");
dbg!(client_b.user_id());
client_b
.channel_store()
.update(cx_b, |channel_store, cx| {
@ -1190,5 +1196,5 @@ fn assert_channels_list_shape(
.map(|(depth, channel)| (channel.id, depth))
.collect::<Vec<_>>()
});
pretty_assertions::assert_eq!(actual, expected_channels);
pretty_assertions::assert_eq!(dbg!(actual), expected_channels);
}

View file

@ -173,7 +173,6 @@ pub fn init(cx: &mut AppContext) {
cx.add_action(
|panel: &mut CollabPanel, action: &StartMoveChannel, _: &mut ViewContext<CollabPanel>| {
dbg!(action);
panel.channel_move = Some(*action);
},
);
@ -181,7 +180,6 @@ pub fn init(cx: &mut AppContext) {
cx.add_action(
|panel: &mut CollabPanel, action: &LinkChannel, cx: &mut ViewContext<CollabPanel>| {
if let Some(move_start) = panel.channel_move.take() {
dbg!(action.to, &move_start);
panel.channel_store.update(cx, |channel_store, cx| {
channel_store
.link_channel(move_start.channel_id, action.to, cx)
@ -194,7 +192,6 @@ pub fn init(cx: &mut AppContext) {
cx.add_action(
|panel: &mut CollabPanel, action: &MoveChannel, cx: &mut ViewContext<CollabPanel>| {
if let Some(move_start) = panel.channel_move.take() {
dbg!(&move_start, action.to);
panel.channel_store.update(cx, |channel_store, cx| {
if let Some(parent) = move_start.parent_id {
channel_store

View file

@ -135,7 +135,7 @@ message Envelope {
RefreshInlayHints refresh_inlay_hints = 118;
CreateChannel create_channel = 119;
ChannelResponse channel_response = 120;
CreateChannelResponse create_channel_response = 120;
InviteChannelMember invite_channel_member = 121;
RemoveChannelMember remove_channel_member = 122;
RespondToChannelInvite respond_to_channel_invite = 123;
@ -146,6 +146,7 @@ message Envelope {
GetChannelMembersResponse get_channel_members_response = 128;
SetChannelMemberAdmin set_channel_member_admin = 129;
RenameChannel rename_channel = 130;
RenameChannelResponse rename_channel_response = 154;
JoinChannelBuffer join_channel_buffer = 131;
JoinChannelBufferResponse join_channel_buffer_response = 132;
@ -169,7 +170,7 @@ message Envelope {
LinkChannel link_channel = 151;
UnlinkChannel unlink_channel = 152;
MoveChannel move_channel = 153; // Current max
MoveChannel move_channel = 153; // Current max: 154
}
}
@ -959,12 +960,13 @@ message LspDiskBasedDiagnosticsUpdated {}
message UpdateChannels {
repeated Channel channels = 1;
repeated ChannelEdge delete_edge = 2;
repeated uint64 delete_channels = 3;
repeated Channel channel_invitations = 4;
repeated uint64 remove_channel_invitations = 5;
repeated ChannelParticipants channel_participants = 6;
repeated ChannelPermission channel_permissions = 7;
repeated ChannelEdge insert_edge = 2;
repeated ChannelEdge delete_edge = 3;
repeated uint64 delete_channels = 4;
repeated Channel channel_invitations = 5;
repeated uint64 remove_channel_invitations = 6;
repeated ChannelParticipants channel_participants = 7;
repeated ChannelPermission channel_permissions = 8;
}
message ChannelEdge {
@ -1015,8 +1017,9 @@ message CreateChannel {
optional uint64 parent_id = 2;
}
message ChannelResponse {
message CreateChannelResponse {
Channel channel = 1;
optional uint64 parent_id = 2;
}
message InviteChannelMember {
@ -1041,6 +1044,10 @@ message RenameChannel {
string name = 2;
}
message RenameChannelResponse {
Channel channel = 1;
}
message JoinChannelChat {
uint64 channel_id = 1;
}
@ -1512,7 +1519,6 @@ message Nonce {
message Channel {
uint64 id = 1;
string name = 2;
optional uint64 parent_id = 3;
}
message Contact {

View file

@ -146,7 +146,7 @@ messages!(
(CopyProjectEntry, Foreground),
(CreateBufferForPeer, Foreground),
(CreateChannel, Foreground),
(ChannelResponse, Foreground),
(CreateChannelResponse, Foreground),
(ChannelMessageSent, Foreground),
(CreateProjectEntry, Foreground),
(CreateRoom, Foreground),
@ -229,6 +229,7 @@ messages!(
(RoomUpdated, Foreground),
(SaveBuffer, Foreground),
(RenameChannel, Foreground),
(RenameChannelResponse, Foreground),
(SetChannelMemberAdmin, Foreground),
(SearchProject, Background),
(SearchProjectResponse, Background),
@ -285,7 +286,7 @@ request_messages!(
(CopyProjectEntry, ProjectEntryResponse),
(CreateProjectEntry, ProjectEntryResponse),
(CreateRoom, CreateRoomResponse),
(CreateChannel, ChannelResponse),
(CreateChannel, CreateChannelResponse),
(DeclineCall, Ack),
(DeleteProjectEntry, ProjectEntryResponse),
(ExpandProjectEntry, ExpandProjectEntryResponse),
@ -333,7 +334,7 @@ request_messages!(
(RemoveChannelMessage, Ack),
(DeleteChannel, Ack),
(RenameProjectEntry, ProjectEntryResponse),
(RenameChannel, ChannelResponse),
(RenameChannel, RenameChannelResponse),
(LinkChannel, Ack),
(UnlinkChannel, Ack),
(MoveChannel, Ack),