Remove logic for multiple channel parents
Co-authored-by: Conrad <conrad@zed.dev> Co-authored-by: Kyle <kyle@zed.dev> Co-authored-by: Joseph <joseph@zed.dev>
This commit is contained in:
Max Brunsfeld
parent
cc9e92857b
commit
5c03b6a610
23 changed files with 772 additions and 2069 deletions
|
|
@ -11,9 +11,7 @@ pub use channel_chat::{
|
|||
mentions_to_proto, ChannelChat, ChannelChatEvent, ChannelMessage, ChannelMessageId,
|
||||
MessageParams,
|
||||
};
|
||||
pub use channel_store::{
|
||||
Channel, ChannelData, ChannelEvent, ChannelId, ChannelMembership, ChannelPath, ChannelStore,
|
||||
};
|
||||
pub use channel_store::{Channel, ChannelEvent, ChannelId, ChannelMembership, ChannelStore};
|
||||
|
||||
#[cfg(test)]
|
||||
mod channel_store_tests;
|
||||
|
|
|
|||
|
|
@ -9,11 +9,10 @@ use db::RELEASE_CHANNEL;
|
|||
use futures::{channel::mpsc, future::Shared, Future, FutureExt, StreamExt};
|
||||
use gpui::{AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, Task, WeakModelHandle};
|
||||
use rpc::{
|
||||
proto::{self, ChannelEdge, ChannelVisibility},
|
||||
proto::{self, ChannelVisibility},
|
||||
TypedEnvelope,
|
||||
};
|
||||
use serde_derive::{Deserialize, Serialize};
|
||||
use std::{borrow::Cow, hash::Hash, mem, ops::Deref, sync::Arc, time::Duration};
|
||||
use std::{mem, sync::Arc, time::Duration};
|
||||
use util::ResultExt;
|
||||
|
||||
pub fn init(client: &Arc<Client>, user_store: ModelHandle<UserStore>, cx: &mut AppContext) {
|
||||
|
|
@ -27,7 +26,7 @@ pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30);
|
|||
pub type ChannelId = u64;
|
||||
|
||||
pub struct ChannelStore {
|
||||
channel_index: ChannelIndex,
|
||||
pub channel_index: ChannelIndex,
|
||||
channel_invitations: Vec<Arc<Channel>>,
|
||||
channel_participants: HashMap<ChannelId, Vec<Arc<User>>>,
|
||||
outgoing_invites: HashSet<(ChannelId, UserId)>,
|
||||
|
|
@ -42,8 +41,6 @@ pub struct ChannelStore {
|
|||
_update_channels: Task<()>,
|
||||
}
|
||||
|
||||
pub type ChannelData = (Channel, ChannelPath);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct Channel {
|
||||
pub id: ChannelId,
|
||||
|
|
@ -52,6 +49,7 @@ pub struct Channel {
|
|||
pub role: proto::ChannelRole,
|
||||
pub unseen_note_version: Option<(u64, clock::Global)>,
|
||||
pub unseen_message_id: Option<u64>,
|
||||
pub parent_path: Vec<u64>,
|
||||
}
|
||||
|
||||
impl Channel {
|
||||
|
|
@ -78,9 +76,6 @@ impl Channel {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
|
||||
pub struct ChannelPath(Arc<[ChannelId]>);
|
||||
|
||||
pub struct ChannelMembership {
|
||||
pub user: Arc<User>,
|
||||
pub kind: proto::channel_member::Kind,
|
||||
|
|
@ -193,14 +188,11 @@ impl ChannelStore {
|
|||
self.client.clone()
|
||||
}
|
||||
|
||||
pub fn has_children(&self, channel_id: ChannelId) -> bool {
|
||||
self.channel_index.iter().any(|path| {
|
||||
if let Some(ix) = path.iter().position(|id| *id == channel_id) {
|
||||
path.len() > ix + 1
|
||||
} else {
|
||||
false
|
||||
}
|
||||
})
|
||||
pub fn channel_has_children(&self) -> bool {
|
||||
self.channel_index
|
||||
.by_id()
|
||||
.iter()
|
||||
.any(|(_, channel)| channel.parent_path.contains(&channel.id))
|
||||
}
|
||||
|
||||
/// Returns the number of unique channels in the store
|
||||
|
|
@ -222,20 +214,19 @@ impl ChannelStore {
|
|||
}
|
||||
|
||||
/// Iterate over all entries in the channel DAG
|
||||
pub fn channel_dag_entries(&self) -> impl '_ + Iterator<Item = (usize, &Arc<Channel>)> {
|
||||
self.channel_index.iter().map(move |path| {
|
||||
let id = path.last().unwrap();
|
||||
let channel = self.channel_for_id(*id).unwrap();
|
||||
(path.len() - 1, channel)
|
||||
})
|
||||
pub fn ordered_channels(&self) -> impl '_ + Iterator<Item = (usize, &Arc<Channel>)> {
|
||||
self.channel_index
|
||||
.ordered_channels()
|
||||
.iter()
|
||||
.filter_map(move |id| {
|
||||
let channel = self.channel_index.by_id().get(id)?;
|
||||
Some((channel.parent_path.len(), channel))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn channel_dag_entry_at(&self, ix: usize) -> Option<(&Arc<Channel>, &ChannelPath)> {
|
||||
let path = self.channel_index.get(ix)?;
|
||||
let id = path.last().unwrap();
|
||||
let channel = self.channel_for_id(*id).unwrap();
|
||||
|
||||
Some((channel, path))
|
||||
pub fn channel_at_index(&self, ix: usize) -> Option<&Arc<Channel>> {
|
||||
let channel_id = self.channel_index.ordered_channels().get(ix)?;
|
||||
self.channel_index.by_id().get(channel_id)
|
||||
}
|
||||
|
||||
pub fn channel_at(&self, ix: usize) -> Option<&Arc<Channel>> {
|
||||
|
|
@ -484,20 +475,19 @@ impl ChannelStore {
|
|||
.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![]
|
||||
};
|
||||
// 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,
|
||||
..Default::default()
|
||||
},
|
||||
cx,
|
||||
|
|
@ -515,53 +505,16 @@ impl ChannelStore {
|
|||
})
|
||||
}
|
||||
|
||||
pub fn link_channel(
|
||||
&mut self,
|
||||
channel_id: ChannelId,
|
||||
to: ChannelId,
|
||||
cx: &mut ModelContext<Self>,
|
||||
) -> Task<Result<()>> {
|
||||
let client = self.client.clone();
|
||||
cx.spawn(|_, _| async move {
|
||||
let _ = client
|
||||
.request(proto::LinkChannel { channel_id, to })
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn unlink_channel(
|
||||
&mut self,
|
||||
channel_id: ChannelId,
|
||||
from: ChannelId,
|
||||
cx: &mut ModelContext<Self>,
|
||||
) -> Task<Result<()>> {
|
||||
let client = self.client.clone();
|
||||
cx.spawn(|_, _| async move {
|
||||
let _ = client
|
||||
.request(proto::UnlinkChannel { channel_id, from })
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn move_channel(
|
||||
&mut self,
|
||||
channel_id: ChannelId,
|
||||
from: ChannelId,
|
||||
to: ChannelId,
|
||||
cx: &mut ModelContext<Self>,
|
||||
) -> Task<Result<()>> {
|
||||
let client = self.client.clone();
|
||||
cx.spawn(|_, _| async move {
|
||||
let _ = client
|
||||
.request(proto::MoveChannel {
|
||||
channel_id,
|
||||
from,
|
||||
to,
|
||||
})
|
||||
.request(proto::MoveChannel { channel_id, to })
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
|
|
@ -956,6 +909,7 @@ impl ChannelStore {
|
|||
name: channel.name,
|
||||
unseen_note_version: None,
|
||||
unseen_message_id: None,
|
||||
parent_path: channel.parent_path,
|
||||
}),
|
||||
),
|
||||
}
|
||||
|
|
@ -963,8 +917,6 @@ impl ChannelStore {
|
|||
|
||||
let channels_changed = !payload.channels.is_empty()
|
||||
|| !payload.delete_channels.is_empty()
|
||||
|| !payload.insert_edge.is_empty()
|
||||
|| !payload.delete_edge.is_empty()
|
||||
|| !payload.unseen_channel_messages.is_empty()
|
||||
|| !payload.unseen_channel_buffer_changes.is_empty();
|
||||
|
||||
|
|
@ -1022,14 +974,6 @@ impl ChannelStore {
|
|||
unseen_channel_message.message_id,
|
||||
);
|
||||
}
|
||||
|
||||
for edge in payload.insert_edge {
|
||||
index.insert_edge(edge.channel_id, edge.parent_id);
|
||||
}
|
||||
|
||||
for edge in payload.delete_edge {
|
||||
index.delete_edge(edge.parent_id, edge.channel_id);
|
||||
}
|
||||
}
|
||||
|
||||
cx.notify();
|
||||
|
|
@ -1078,44 +1022,3 @@ impl ChannelStore {
|
|||
}))
|
||||
}
|
||||
}
|
||||
|
||||
impl Deref for ChannelPath {
|
||||
type Target = [ChannelId];
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl ChannelPath {
|
||||
pub fn new(path: Arc<[ChannelId]>) -> Self {
|
||||
debug_assert!(path.len() >= 1);
|
||||
Self(path)
|
||||
}
|
||||
|
||||
pub fn parent_id(&self) -> Option<ChannelId> {
|
||||
self.0.len().checked_sub(2).map(|i| self.0[i])
|
||||
}
|
||||
|
||||
pub fn channel_id(&self) -> ChannelId {
|
||||
self.0[self.0.len() - 1]
|
||||
}
|
||||
}
|
||||
|
||||
impl From<ChannelPath> for Cow<'static, ChannelPath> {
|
||||
fn from(value: ChannelPath) -> Self {
|
||||
Cow::Owned(value)
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a> From<&'a ChannelPath> for Cow<'a, ChannelPath> {
|
||||
fn from(value: &'a ChannelPath) -> Self {
|
||||
Cow::Borrowed(value)
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ChannelPath {
|
||||
fn default() -> Self {
|
||||
ChannelPath(Arc::from([]))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,14 +1,11 @@
|
|||
use std::{ops::Deref, sync::Arc};
|
||||
|
||||
use crate::{Channel, ChannelId};
|
||||
use collections::BTreeMap;
|
||||
use rpc::proto;
|
||||
|
||||
use super::ChannelPath;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[derive(Default, Debug)]
|
||||
pub struct ChannelIndex {
|
||||
paths: Vec<ChannelPath>,
|
||||
channels_ordered: Vec<ChannelId>,
|
||||
channels_by_id: BTreeMap<ChannelId, Arc<Channel>>,
|
||||
}
|
||||
|
||||
|
|
@ -17,8 +14,12 @@ impl ChannelIndex {
|
|||
&self.channels_by_id
|
||||
}
|
||||
|
||||
pub fn ordered_channels(&self) -> &[ChannelId] {
|
||||
&self.channels_ordered
|
||||
}
|
||||
|
||||
pub fn clear(&mut self) {
|
||||
self.paths.clear();
|
||||
self.channels_ordered.clear();
|
||||
self.channels_by_id.clear();
|
||||
}
|
||||
|
||||
|
|
@ -26,13 +27,13 @@ impl ChannelIndex {
|
|||
pub fn delete_channels(&mut self, channels: &[ChannelId]) {
|
||||
self.channels_by_id
|
||||
.retain(|channel_id, _| !channels.contains(channel_id));
|
||||
self.paths
|
||||
.retain(|path| !path.iter().any(|channel_id| channels.contains(channel_id)));
|
||||
self.channels_ordered
|
||||
.retain(|channel_id| !channels.contains(channel_id));
|
||||
}
|
||||
|
||||
pub fn bulk_insert(&mut self) -> ChannelPathsInsertGuard {
|
||||
ChannelPathsInsertGuard {
|
||||
paths: &mut self.paths,
|
||||
channels_ordered: &mut self.channels_ordered,
|
||||
channels_by_id: &mut self.channels_by_id,
|
||||
}
|
||||
}
|
||||
|
|
@ -75,42 +76,15 @@ impl ChannelIndex {
|
|||
}
|
||||
}
|
||||
|
||||
impl Deref for ChannelIndex {
|
||||
type Target = [ChannelPath];
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.paths
|
||||
}
|
||||
}
|
||||
|
||||
/// A guard for ensuring that the paths index maintains its sort and uniqueness
|
||||
/// invariants after a series of insertions
|
||||
#[derive(Debug)]
|
||||
pub struct ChannelPathsInsertGuard<'a> {
|
||||
paths: &'a mut Vec<ChannelPath>,
|
||||
channels_ordered: &'a mut Vec<ChannelId>,
|
||||
channels_by_id: &'a mut BTreeMap<ChannelId, Arc<Channel>>,
|
||||
}
|
||||
|
||||
impl<'a> ChannelPathsInsertGuard<'a> {
|
||||
/// Remove the given edge from this index. This will not remove the channel.
|
||||
/// If this operation would result in a dangling edge, re-insert it.
|
||||
pub fn delete_edge(&mut self, parent_id: ChannelId, channel_id: ChannelId) {
|
||||
self.paths.retain(|path| {
|
||||
!path
|
||||
.windows(2)
|
||||
.any(|window| window == [parent_id, channel_id])
|
||||
});
|
||||
|
||||
// Ensure that there is at least one channel path in the index
|
||||
if !self
|
||||
.paths
|
||||
.iter()
|
||||
.any(|path| path.iter().any(|id| id == &channel_id))
|
||||
{
|
||||
self.insert_root(channel_id);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn note_changed(&mut self, channel_id: ChannelId, epoch: u64, version: &clock::Global) {
|
||||
insert_note_changed(&mut self.channels_by_id, channel_id, epoch, &version);
|
||||
}
|
||||
|
|
@ -141,6 +115,7 @@ impl<'a> ChannelPathsInsertGuard<'a> {
|
|||
name: channel_proto.name,
|
||||
unseen_note_version: None,
|
||||
unseen_message_id: None,
|
||||
parent_path: channel_proto.parent_path,
|
||||
}),
|
||||
);
|
||||
self.insert_root(channel_proto.id);
|
||||
|
|
@ -148,74 +123,35 @@ impl<'a> ChannelPathsInsertGuard<'a> {
|
|||
ret
|
||||
}
|
||||
|
||||
pub fn insert_edge(&mut self, channel_id: ChannelId, parent_id: ChannelId) {
|
||||
let mut parents = Vec::new();
|
||||
let mut descendants = Vec::new();
|
||||
let mut ixs_to_remove = Vec::new();
|
||||
|
||||
for (ix, path) in self.paths.iter().enumerate() {
|
||||
if path
|
||||
.windows(2)
|
||||
.any(|window| window[0] == parent_id && window[1] == channel_id)
|
||||
{
|
||||
// We already have this edge in the index
|
||||
return;
|
||||
}
|
||||
if path.ends_with(&[parent_id]) {
|
||||
parents.push(path);
|
||||
} else if let Some(position) = path.iter().position(|id| id == &channel_id) {
|
||||
if position == 0 {
|
||||
ixs_to_remove.push(ix);
|
||||
}
|
||||
descendants.push(path.split_at(position).1);
|
||||
}
|
||||
}
|
||||
|
||||
let mut new_paths = Vec::new();
|
||||
for parent in parents.iter() {
|
||||
if descendants.is_empty() {
|
||||
let mut new_path = Vec::with_capacity(parent.len() + 1);
|
||||
new_path.extend_from_slice(parent);
|
||||
new_path.push(channel_id);
|
||||
new_paths.push(ChannelPath::new(new_path.into()));
|
||||
} else {
|
||||
for descendant in descendants.iter() {
|
||||
let mut new_path = Vec::with_capacity(parent.len() + descendant.len());
|
||||
new_path.extend_from_slice(parent);
|
||||
new_path.extend_from_slice(descendant);
|
||||
new_paths.push(ChannelPath::new(new_path.into()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for ix in ixs_to_remove.into_iter().rev() {
|
||||
self.paths.swap_remove(ix);
|
||||
}
|
||||
self.paths.extend(new_paths)
|
||||
}
|
||||
|
||||
fn insert_root(&mut self, channel_id: ChannelId) {
|
||||
self.paths.push(ChannelPath::new(Arc::from([channel_id])));
|
||||
self.channels_ordered.push(channel_id);
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a> Drop for ChannelPathsInsertGuard<'a> {
|
||||
fn drop(&mut self) {
|
||||
self.paths.sort_by(|a, b| {
|
||||
let a = channel_path_sorting_key(a, &self.channels_by_id);
|
||||
let b = channel_path_sorting_key(b, &self.channels_by_id);
|
||||
self.channels_ordered.sort_by(|a, b| {
|
||||
let a = channel_path_sorting_key(*a, &self.channels_by_id);
|
||||
let b = channel_path_sorting_key(*b, &self.channels_by_id);
|
||||
a.cmp(b)
|
||||
});
|
||||
self.paths.dedup();
|
||||
self.channels_ordered.dedup();
|
||||
}
|
||||
}
|
||||
|
||||
fn channel_path_sorting_key<'a>(
|
||||
path: &'a [ChannelId],
|
||||
id: ChannelId,
|
||||
channels_by_id: &'a BTreeMap<ChannelId, Arc<Channel>>,
|
||||
) -> impl 'a + Iterator<Item = Option<&'a str>> {
|
||||
path.iter()
|
||||
.map(|id| Some(channels_by_id.get(id)?.name.as_str()))
|
||||
) -> impl Iterator<Item = &str> {
|
||||
let (parent_path, name) = channels_by_id
|
||||
.get(&id)
|
||||
.map_or((&[] as &[_], None), |channel| {
|
||||
(channel.parent_path.as_slice(), Some(channel.name.as_str()))
|
||||
});
|
||||
parent_path
|
||||
.iter()
|
||||
.filter_map(|id| Some(channels_by_id.get(id)?.name.as_str()))
|
||||
.chain(name)
|
||||
}
|
||||
|
||||
fn insert_note_changed(
|
||||
|
|
|
|||
|
|
@ -20,12 +20,14 @@ fn test_update_channels(cx: &mut AppContext) {
|
|||
name: "b".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Admin.into(),
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
proto::Channel {
|
||||
id: 2,
|
||||
name: "a".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Member.into(),
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
],
|
||||
..Default::default()
|
||||
|
|
@ -51,22 +53,14 @@ fn test_update_channels(cx: &mut AppContext) {
|
|||
name: "x".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Admin.into(),
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
proto::Channel {
|
||||
id: 4,
|
||||
name: "y".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Member.into(),
|
||||
},
|
||||
],
|
||||
insert_edge: vec![
|
||||
proto::ChannelEdge {
|
||||
parent_id: 1,
|
||||
channel_id: 3,
|
||||
},
|
||||
proto::ChannelEdge {
|
||||
parent_id: 2,
|
||||
channel_id: 4,
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
],
|
||||
..Default::default()
|
||||
|
|
@ -98,28 +92,21 @@ fn test_dangling_channel_paths(cx: &mut AppContext) {
|
|||
name: "a".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Admin.into(),
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
proto::Channel {
|
||||
id: 1,
|
||||
name: "b".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Admin.into(),
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
proto::Channel {
|
||||
id: 2,
|
||||
name: "c".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Admin.into(),
|
||||
},
|
||||
],
|
||||
insert_edge: vec![
|
||||
proto::ChannelEdge {
|
||||
parent_id: 0,
|
||||
channel_id: 1,
|
||||
},
|
||||
proto::ChannelEdge {
|
||||
parent_id: 1,
|
||||
channel_id: 2,
|
||||
parent_path: Vec::new(),
|
||||
},
|
||||
],
|
||||
..Default::default()
|
||||
|
|
@ -170,6 +157,7 @@ async fn test_channel_messages(cx: &mut TestAppContext) {
|
|||
name: "the-channel".to_string(),
|
||||
visibility: proto::ChannelVisibility::Members as i32,
|
||||
role: proto::ChannelRole::Member.into(),
|
||||
parent_path: vec![],
|
||||
}],
|
||||
..Default::default()
|
||||
});
|
||||
|
|
@ -197,7 +185,7 @@ async fn test_channel_messages(cx: &mut TestAppContext) {
|
|||
|
||||
// Join a channel and populate its existing messages.
|
||||
let channel = channel_store.update(cx, |store, cx| {
|
||||
let channel_id = store.channel_dag_entries().next().unwrap().1.id;
|
||||
let channel_id = store.ordered_channels().next().unwrap().1.id;
|
||||
store.open_channel_chat(channel_id, cx)
|
||||
});
|
||||
let join_channel = server.receive::<proto::JoinChannelChat>().await.unwrap();
|
||||
|
|
@ -384,7 +372,7 @@ fn assert_channels(
|
|||
) {
|
||||
let actual = channel_store.read_with(cx, |store, _| {
|
||||
store
|
||||
.channel_dag_entries()
|
||||
.ordered_channels()
|
||||
.map(|(depth, channel)| (depth, channel.name.to_string(), channel.role))
|
||||
.collect::<Vec<_>>()
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue