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

Add read-only channel notes support

Browse source 
Fix some bugs where ChannelNotes and ChannelChat had old cached channel
instances
This commit is contained in:
Conrad Irwin 2023-10-19 13:03:44 -06:00
parent 3853009d92
commit 0eff7c6ca9
12 changed files with 186 additions and 219 deletions
Download patch file Download diff file Expand all files Collapse all files

28
crates/channel/src/channel_buffer.rs
View file

@ -1,4 +1,4 @@
use crate::Channel;
use crate::{Channel, ChannelId, ChannelStore};
use anyhow::Result;
use client::{Client, Collaborator, UserStore};
use collections::HashMap;
@ -19,10 +19,11 @@ pub(crate) fn init(client: &Arc<Client>) {
}
pub struct ChannelBuffer {
pub(crate) channel: Arc<Channel>,
pub channel_id: ChannelId,
connected: bool,
collaborators: HashMap<PeerId, Collaborator>,
user_store: ModelHandle<UserStore>,
channel_store: ModelHandle<ChannelStore>,
buffer: ModelHandle<language::Buffer>,
buffer_epoch: u64,
client: Arc<Client>,
@ -34,6 +35,7 @@ pub enum ChannelBufferEvent {
CollaboratorsChanged,
Disconnected,
BufferEdited,
ChannelChanged,
}
impl Entity for ChannelBuffer {
@ -46,7 +48,7 @@ impl Entity for ChannelBuffer {
}
self.client
.send(proto::LeaveChannelBuffer {
channel_id: self.channel.id,
channel_id: self.channel_id,
})
.log_err();
}
@ -58,6 +60,7 @@ impl ChannelBuffer {
channel: Arc<Channel>,
client: Arc<Client>,
user_store: ModelHandle<UserStore>,
channel_store: ModelHandle<ChannelStore>,
mut cx: AsyncAppContext,
) -> Result<ModelHandle<Self>> {
let response = client
@ -90,9 +93,10 @@ impl ChannelBuffer {
connected: true,
collaborators: Default::default(),
acknowledge_task: None,
channel,
channel_id: channel.id,
subscription: Some(subscription.set_model(&cx.handle(), &mut cx.to_async())),
user_store,
channel_store,
};
this.replace_collaborators(response.collaborators, cx);
this
@ -179,7 +183,7 @@ impl ChannelBuffer {
let operation = language::proto::serialize_operation(operation);
self.client
.send(proto::UpdateChannelBuffer {
channel_id: self.channel.id,
channel_id: self.channel_id,
operations: vec![operation],
})
.log_err();
@ -223,12 +227,15 @@ impl ChannelBuffer {
&self.collaborators
}
pub fn channel(&self) -> Arc<Channel> {
self.channel.clone()
pub fn channel(&self, cx: &AppContext) -> Option<Arc<Channel>> {
self.channel_store
.read(cx)
.channel_for_id(self.channel_id)
.cloned()
}
pub(crate) fn disconnect(&mut self, cx: &mut ModelContext<Self>) {
log::info!("channel buffer {} disconnected", self.channel.id);
log::info!("channel buffer {} disconnected", self.channel_id);
if self.connected {
self.connected = false;
self.subscription.take();
@ -237,6 +244,11 @@ impl ChannelBuffer {
}
}
pub(crate) fn channel_changed(&mut self, cx: &mut ModelContext<Self>) {
cx.emit(ChannelBufferEvent::ChannelChanged);
cx.notify()
}
pub fn is_connected(&self) -> bool {
self.connected
}

27
crates/channel/src/channel_chat.rs
View file

@ -14,7 +14,7 @@ use time::OffsetDateTime;
use util::{post_inc, ResultExt as _, TryFutureExt};
pub struct ChannelChat {
channel: Arc<Channel>,
pub channel_id: ChannelId,
messages: SumTree<ChannelMessage>,
channel_store: ModelHandle<ChannelStore>,
loaded_all_messages: bool,
@ -74,7 +74,7 @@ impl Entity for ChannelChat {
fn release(&mut self, _: &mut AppContext) {
self.rpc
.send(proto::LeaveChannelChat {
channel_id: self.channel.id,
channel_id: self.channel_id,
})
.log_err();
}
@ -99,7 +99,7 @@ impl ChannelChat {
Ok(cx.add_model(|cx| {
let mut this = Self {
channel,
channel_id: channel.id,
user_store,
channel_store,
rpc: client,
@ -116,8 +116,11 @@ impl ChannelChat {
}))
}
pub fn channel(&self) -> &Arc<Channel> {
&self.channel
pub fn channel(&self, cx: &AppContext) -> Option<Arc<Channel>> {
self.channel_store
.read(cx)
.channel_for_id(self.channel_id)
.cloned()
}
pub fn send_message(
@ -135,7 +138,7 @@ impl ChannelChat {
.current_user()
.ok_or_else(|| anyhow!("current_user is not present"))?;
let channel_id = self.channel.id;
let channel_id = self.channel_id;
let pending_id = ChannelMessageId::Pending(post_inc(&mut self.next_pending_message_id));
let nonce = self.rng.gen();
self.insert_messages(
@ -178,7 +181,7 @@ impl ChannelChat {
pub fn remove_message(&mut self, id: u64, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
let response = self.rpc.request(proto::RemoveChannelMessage {
channel_id: self.channel.id,
channel_id: self.channel_id,
message_id: id,
});
cx.spawn(|this, mut cx| async move {
@ -195,7 +198,7 @@ impl ChannelChat {
if !self.loaded_all_messages {
let rpc = self.rpc.clone();
let user_store = self.user_store.clone();
let channel_id = self.channel.id;
let channel_id = self.channel_id;
if let Some(before_message_id) =
self.messages.first().and_then(|message| match message.id {
ChannelMessageId::Saved(id) => Some(id),
@ -236,13 +239,13 @@ impl ChannelChat {
{
self.rpc
.send(proto::AckChannelMessage {
channel_id: self.channel.id,
channel_id: self.channel_id,
message_id: latest_message_id,
})
.ok();
self.last_acknowledged_id = Some(latest_message_id);
self.channel_store.update(cx, |store, cx| {
store.acknowledge_message_id(self.channel.id, latest_message_id, cx);
store.acknowledge_message_id(self.channel_id, latest_message_id, cx);
});
}
}
@ -251,7 +254,7 @@ impl ChannelChat {
pub fn rejoin(&mut self, cx: &mut ModelContext<Self>) {
let user_store = self.user_store.clone();
let rpc = self.rpc.clone();
let channel_id = self.channel.id;
let channel_id = self.channel_id;
cx.spawn(|this, mut cx| {
async move {
let response = rpc.request(proto::JoinChannelChat { channel_id }).await?;
@ -348,7 +351,7 @@ impl ChannelChat {
this.update(&mut cx, |this, cx| {
this.insert_messages(SumTree::from_item(message, &()), cx);
cx.emit(ChannelChatEvent::NewMessage {
channel_id: this.channel.id,
channel_id: this.channel_id,
message_id,
})
});

48
crates/channel/src/channel_store.rs
View file

@ -72,6 +72,10 @@ impl Channel {
slug.trim_matches(|c| c == '-').to_string()
}
pub fn can_edit_notes(&self) -> bool {
self.role == proto::ChannelRole::Member || self.role == proto::ChannelRole::Admin
}
}
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
@ -265,10 +269,11 @@ impl ChannelStore {
) -> Task<Result<ModelHandle<ChannelBuffer>>> {
let client = self.client.clone();
let user_store = self.user_store.clone();
let channel_store = cx.handle();
self.open_channel_resource(
channel_id,
|this| &mut this.opened_buffers,
|channel, cx| ChannelBuffer::new(channel, client, user_store, cx),
|channel, cx| ChannelBuffer::new(channel, client, user_store, channel_store, cx),
cx,
)
}
@ -778,7 +783,7 @@ impl ChannelStore {
let channel_buffer = buffer.read(cx);
let buffer = channel_buffer.buffer().read(cx);
buffer_versions.push(proto::ChannelBufferVersion {
channel_id: channel_buffer.channel().id,
channel_id: channel_buffer.channel_id,
epoch: channel_buffer.epoch(),
version: language::proto::serialize_version(&buffer.version()),
});
@ -805,13 +810,13 @@ impl ChannelStore {
};
channel_buffer.update(cx, |channel_buffer, cx| {
let channel_id = channel_buffer.channel().id;
let channel_id = channel_buffer.channel_id;
if let Some(remote_buffer) = response
.buffers
.iter_mut()
.find(|buffer| buffer.channel_id == channel_id)
{
let channel_id = channel_buffer.channel().id;
let channel_id = channel_buffer.channel_id;
let remote_version =
language::proto::deserialize_version(&remote_buffer.version);
@ -934,11 +939,27 @@ impl ChannelStore {
if channels_changed {
if !payload.delete_channels.is_empty() {
self.channel_index.delete_channels(&payload.delete_channels);
self.channel_participants
.retain(|channel_id, _| !payload.delete_channels.contains(channel_id));
let mut channels_to_delete: Vec<u64> = Vec::new();
let mut channels_to_rehome: Vec<u64> = Vec::new();
for channel_id in payload.delete_channels {
if payload
.channels
.iter()
.any(|channel| channel.id == channel_id)
{
channels_to_rehome.push(channel_id)
} else {
channels_to_delete.push(channel_id)
}
}
for channel_id in &payload.delete_channels {
self.channel_index.delete_channels(&channels_to_delete);
self.channel_index
.delete_paths_through_channels(&channels_to_rehome);
self.channel_participants
.retain(|channel_id, _| !channels_to_delete.contains(channel_id));
for channel_id in &channels_to_delete {
let channel_id = *channel_id;
if payload
.channels
@ -959,7 +980,16 @@ impl ChannelStore {
let mut index = self.channel_index.bulk_insert();
for channel in payload.channels {
index.insert(channel)
let id = channel.id;
let channel_changed = index.insert(channel);
if channel_changed {
if let Some(OpenedModelHandle::Open(buffer)) = self.opened_buffers.get(&id) {
if let Some(buffer) = buffer.upgrade(cx) {
buffer.update(cx, ChannelBuffer::channel_changed);
}
}
}
}
for unseen_buffer_change in payload.unseen_channel_buffer_changes {

21
crates/channel/src/channel_store/channel_index.rs
View file

@ -24,12 +24,16 @@ impl ChannelIndex {
/// Delete the given channels from this index.
pub fn delete_channels(&mut self, channels: &[ChannelId]) {
dbg!("delete_channels", &channels);
self.channels_by_id
.retain(|channel_id, _| !channels.contains(channel_id));
self.paths.retain(|path| {
path.iter()
.all(|channel_id| self.channels_by_id.contains_key(channel_id))
});
self.delete_paths_through_channels(channels)
}
pub fn delete_paths_through_channels(&mut self, channels: &[ChannelId]) {
dbg!("rehome_channels", &channels);
self.paths
.retain(|path| !path.iter().any(|channel_id| channels.contains(channel_id)));
}
pub fn bulk_insert(&mut self) -> ChannelPathsInsertGuard {
@ -121,9 +125,15 @@ impl<'a> ChannelPathsInsertGuard<'a> {
insert_new_message(&mut self.channels_by_id, channel_id, message_id)
}
pub fn insert(&mut self, channel_proto: proto::Channel) {
pub fn insert(&mut self, channel_proto: proto::Channel) -> bool {
let mut ret = false;
if let Some(existing_channel) = self.channels_by_id.get_mut(&channel_proto.id) {
let existing_channel = Arc::make_mut(existing_channel);
ret = existing_channel.visibility != channel_proto.visibility()
|| existing_channel.role != channel_proto.role()
|| existing_channel.name != channel_proto.name;
existing_channel.visibility = channel_proto.visibility();
existing_channel.role = channel_proto.role();
existing_channel.name = channel_proto.name;
@ -141,6 +151,7 @@ impl<'a> ChannelPathsInsertGuard<'a> {
);
self.insert_root(channel_proto.id);
}
ret
}
pub fn insert_edge(&mut self, channel_id: ChannelId, parent_id: ChannelId) {