mixa/zed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show when a channel resource changes (#3074)

Browse Source 
This PR adds a mechanism for notifying the client when a channel note
has been edited or a message has been changed.

TODO: 
- [x] Fix infinite loop when opening the chat panel
- [x] Switch to client-side ack model of observation detection
- [x] Add client-side-only change detection (e.g. for when a channel
note is open locally but not focused)
- [x] Review implementation / query performance.
- [x] Fix lack of ACK on restart for channel buffers
- [x] remove channel note opening on click
- [x] Fix channel messages sent while chat channel is in the background
not showing the channel as changed.

Release Notes:

- N/A
This commit is contained in:
Mikayla Maki
2023-10-04 11:57:33 -07:00
committed by GitHub
parent 55d2b9b3c9 4d61d01943
commit 01b45f4f23
48 changed files with 2544 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files

656
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

2
Cargo.toml
View File

@@ -105,6 +105,7 @@ rand = { version = "0.8.5" }
refineable = { path = "./crates/refineable" }
regex = { version = "1.5" }
rust-embed = { version = "8.0", features = ["include-exclude"] }
rusqlite = { version = "0.29.0", features = ["blob", "array", "modern_sqlite"] }
schemars = { version = "0.8" }
serde = { version = "1.0", features = ["derive", "rc"] }
serde_derive = { version = "1.0", features = ["deserialize_in_place"] }
@@ -120,6 +121,7 @@ tree-sitter = "0.20"
unindent = { version = "0.1.7" }
pretty_assertions = "1.3.0"
git2 = { version = "0.15", default-features = false}
uuid = { version = "1.1.2", features = ["v4"] }
tree-sitter-bash = { git = "https://github.com/tree-sitter/tree-sitter-bash", rev = "1b0321ee85701d5036c334a6f04761cdc672e64c" }
tree-sitter-c = "0.20.1"

2
crates/ai/Cargo.toml
View File

@@ -27,7 +27,7 @@ log.workspace = true
parse_duration = "2.1.1"
tiktoken-rs = "0.5.0"
matrixmultiply = "0.3.7"
rusqlite = { version = "0.27.0", features = ["blob", "array", "modern_sqlite"] }
rusqlite = { version = "0.29.0", features = ["blob", "array", "modern_sqlite"] }
bincode = "1.3.3"
[dev-dependencies]

2
crates/assistant/Cargo.toml
View File

@@ -21,8 +21,8 @@ search = { path = "../search" }
settings = { path = "../settings" }
theme = { path = "../theme" }
util = { path = "../util" }
uuid = { version = "1.1.2", features = ["v4"] }
workspace = { path = "../workspace" }
uuid.workspace = true
anyhow.workspace = true
chrono = { version = "0.4", features = ["serde"] }

16
crates/call/src/room.rs
View File

@@ -104,6 +104,10 @@ impl Room {
self.channel_id
}
pub fn is_sharing_project(&self) -> bool {
!self.shared_projects.is_empty()
}
#[cfg(any(test, feature = "test-support"))]
pub fn is_connected(&self) -> bool {
if let Some(live_kit) = self.live_kit.as_ref() {
@@ -594,9 +598,8 @@ impl Room {
.map_or(&[], |v| v.as_slice())
}
/// projects_to_join returns a list of shared projects sorted such
/// that the most 'active' projects appear last.
pub fn projects_to_join(&self) -> Vec<(u64, u64)> {
/// Returns the most 'active' projects, defined as most people in the project
pub fn most_active_project(&self) -> Option<(u64, u64)> {
let mut projects = HashMap::default();
let mut hosts = HashMap::default();
for participant in self.remote_participants.values() {
@@ -613,12 +616,11 @@ impl Room {
}
let mut pairs: Vec<(u64, usize)> = projects.into_iter().collect();
pairs.sort_by_key(|(_, count)| 0 - *count as i32);
pairs.sort_by_key(|(_, count)| *count as i32);
pairs
.into_iter()
.map(|(project_id, _)| (project_id, hosts[&project_id]))
.collect()
.first()
.map(|(project_id, _)| (*project_id, hosts[&project_id]))
}
async fn handle_room_updated(

3
crates/channel/Cargo.toml
View File

@@ -23,6 +23,7 @@ language = { path = "../language" }
settings = { path = "../settings" }
feature_flags = { path = "../feature_flags" }
sum_tree = { path = "../sum_tree" }
clock = { path = "../clock" }
anyhow.workspace = true
futures.workspace = true
@@ -38,7 +39,7 @@ smol.workspace = true
thiserror.workspace = true
time.workspace = true
tiny_http = "0.8"
uuid = { version = "1.1.2", features = ["v4"] }
uuid.workspace = true
url = "2.2"
serde.workspace = true
serde_derive.workspace = true

2
crates/channel/src/channel.rs
View File

@@ -2,7 +2,7 @@ mod channel_buffer;
mod channel_chat;
mod channel_store;
pub use channel_buffer::{ChannelBuffer, ChannelBufferEvent};
pub use channel_buffer::{ChannelBuffer, ChannelBufferEvent, ACKNOWLEDGE_DEBOUNCE_INTERVAL};
pub use channel_chat::{ChannelChat, ChannelChatEvent, ChannelMessage, ChannelMessageId};
pub use channel_store::{
Channel, ChannelData, ChannelEvent, ChannelId, ChannelMembership, ChannelPath, ChannelStore,

57
crates/channel/src/channel_buffer.rs
View File

@@ -2,14 +2,17 @@ use crate::Channel;
use anyhow::Result;
use client::{Client, Collaborator, UserStore};
use collections::HashMap;
use gpui::{AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle};
use gpui::{AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, Task};
use language::proto::serialize_version;
use rpc::{
proto::{self, PeerId},
TypedEnvelope,
};
use std::sync::Arc;
use std::{sync::Arc, time::Duration};
use util::ResultExt;
pub const ACKNOWLEDGE_DEBOUNCE_INTERVAL: Duration = Duration::from_millis(250);
pub(crate) fn init(client: &Arc<Client>) {
client.add_model_message_handler(ChannelBuffer::handle_update_channel_buffer);
client.add_model_message_handler(ChannelBuffer::handle_update_channel_buffer_collaborators);
@@ -24,11 +27,13 @@ pub struct ChannelBuffer {
buffer_epoch: u64,
client: Arc<Client>,
subscription: Option<client::Subscription>,
acknowledge_task: Option<Task<Result<()>>>,
}
pub enum ChannelBufferEvent {
CollaboratorsChanged,
Disconnected,
BufferEdited,
}
impl Entity for ChannelBuffer {
@@ -36,6 +41,9 @@ impl Entity for ChannelBuffer {
fn release(&mut self, _: &mut AppContext) {
if self.connected {
if let Some(task) = self.acknowledge_task.take() {
task.detach();
}
self.client
.send(proto::LeaveChannelBuffer {
channel_id: self.channel.id,
@@ -81,6 +89,7 @@ impl ChannelBuffer {
client,
connected: true,
collaborators: Default::default(),
acknowledge_task: None,
channel,
subscription: Some(subscription.set_model(&cx.handle(), &mut cx.to_async())),
user_store,
@@ -159,19 +168,45 @@ impl ChannelBuffer {
&mut self,
_: ModelHandle<language::Buffer>,
event: &language::Event,
_: &mut ModelContext<Self>,
cx: &mut ModelContext<Self>,
) {
if let language::Event::Operation(operation) = event {
let operation = language::proto::serialize_operation(operation);
self.client
.send(proto::UpdateChannelBuffer {
channel_id: self.channel.id,
operations: vec![operation],
})
.log_err();
match event {
language::Event::Operation(operation) => {
let operation = language::proto::serialize_operation(operation);
self.client
.send(proto::UpdateChannelBuffer {
channel_id: self.channel.id,
operations: vec![operation],
})
.log_err();
}
language::Event::Edited => {
cx.emit(ChannelBufferEvent::BufferEdited);
}
_ => {}
}
}
pub fn acknowledge_buffer_version(&mut self, cx: &mut ModelContext<'_, ChannelBuffer>) {
let buffer = self.buffer.read(cx);
let version = buffer.version();
let buffer_id = buffer.remote_id();
let client = self.client.clone();
let epoch = self.epoch();
self.acknowledge_task = Some(cx.spawn_weak(|_, cx| async move {
cx.background().timer(ACKNOWLEDGE_DEBOUNCE_INTERVAL).await;
client
.send(proto::AckBufferOperation {
buffer_id,
epoch,
version: serialize_version(&version),
})
.ok();
Ok(())
}));
}
pub fn epoch(&self) -> u64 {
self.buffer_epoch
}

39
crates/channel/src/channel_chat.rs
View File

@@ -1,4 +1,4 @@
use crate::Channel;
use crate::{Channel, ChannelId, ChannelStore};
use anyhow::{anyhow, Result};
use client::{
proto,
@@ -16,7 +16,9 @@ use util::{post_inc, ResultExt as _, TryFutureExt};
pub struct ChannelChat {
channel: Arc<Channel>,
messages: SumTree<ChannelMessage>,
channel_store: ModelHandle<ChannelStore>,
loaded_all_messages: bool,
last_acknowledged_id: Option<u64>,
next_pending_message_id: usize,
user_store: ModelHandle<UserStore>,
rpc: Arc<Client>,
@@ -55,6 +57,10 @@ pub enum ChannelChatEvent {
old_range: Range<usize>,
new_count: usize,
},
NewMessage {
channel_id: ChannelId,
message_id: u64,
},
}
pub fn init(client: &Arc<Client>) {
@@ -77,6 +83,7 @@ impl Entity for ChannelChat {
impl ChannelChat {
pub async fn new(
channel: Arc<Channel>,
channel_store: ModelHandle<ChannelStore>,
user_store: ModelHandle<UserStore>,
client: Arc<Client>,
mut cx: AsyncAppContext,
@@ -94,11 +101,13 @@ impl ChannelChat {
let mut this = Self {
channel,
user_store,
channel_store,
rpc: client,
outgoing_messages_lock: Default::default(),
messages: Default::default(),
loaded_all_messages,
next_pending_message_id: 0,
last_acknowledged_id: None,
rng: StdRng::from_entropy(),
_subscription: subscription.set_model(&cx.handle(), &mut cx.to_async()),
};
@@ -219,6 +228,26 @@ impl ChannelChat {
false
}
pub fn acknowledge_last_message(&mut self, cx: &mut ModelContext<Self>) {
if let ChannelMessageId::Saved(latest_message_id) = self.messages.summary().max_id {
if self
.last_acknowledged_id
.map_or(true, |acknowledged_id| acknowledged_id < latest_message_id)
{
self.rpc
.send(proto::AckChannelMessage {
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);
});
}
}
}
pub fn rejoin(&mut self, cx: &mut ModelContext<Self>) {
let user_store = self.user_store.clone();
let rpc = self.rpc.clone();
@@ -313,10 +342,15 @@ impl ChannelChat {
.payload
.message
.ok_or_else(|| anyhow!("empty message"))?;
let message_id = message.id;
let message = ChannelMessage::from_proto(message, &user_store, &mut cx).await?;
this.update(&mut cx, |this, cx| {
this.insert_messages(SumTree::from_item(message, &()), cx)
this.insert_messages(SumTree::from_item(message, &()), cx);
cx.emit(ChannelChatEvent::NewMessage {
channel_id: this.channel.id,
message_id,
})
});
Ok(())
@@ -388,6 +422,7 @@ impl ChannelChat {
old_range: start_ix..end_ix,
new_count,
});
cx.notify();
}
}

85
crates/channel/src/channel_store.rs
View File

@@ -43,6 +43,8 @@ pub type ChannelData = (Channel, ChannelPath);
pub struct Channel {
pub id: ChannelId,
pub name: String,
pub unseen_note_version: Option<(u64, clock::Global)>,
pub unseen_message_id: Option<u64>,
}
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
@@ -207,6 +209,64 @@ impl ChannelStore {
)
}
pub fn has_channel_buffer_changed(&self, channel_id: ChannelId) -> Option<bool> {
self.channel_index
.by_id()
.get(&channel_id)
.map(|channel| channel.unseen_note_version.is_some())
}
pub fn has_new_messages(&self, channel_id: ChannelId) -> Option<bool> {
self.channel_index
.by_id()
.get(&channel_id)
.map(|channel| channel.unseen_message_id.is_some())
}
pub fn notes_changed(
&mut self,
channel_id: ChannelId,
epoch: u64,
version: &clock::Global,
cx: &mut ModelContext<Self>,
) {
self.channel_index.note_changed(channel_id, epoch, version);
cx.notify();
}
pub fn new_message(
&mut self,
channel_id: ChannelId,
message_id: u64,
cx: &mut ModelContext<Self>,
) {
self.channel_index.new_message(channel_id, message_id);
cx.notify();
}
pub fn acknowledge_message_id(
&mut self,
channel_id: ChannelId,
message_id: u64,
cx: &mut ModelContext<Self>,
) {
self.channel_index
.acknowledge_message_id(channel_id, message_id);
cx.notify();
}
pub fn acknowledge_notes_version(
&mut self,
channel_id: ChannelId,
epoch: u64,
version: &clock::Global,
cx: &mut ModelContext<Self>,
) {
self.channel_index
.acknowledge_note_version(channel_id, epoch, version);
cx.notify();
}
pub fn open_channel_chat(
&mut self,
channel_id: ChannelId,
@@ -214,10 +274,11 @@ impl ChannelStore {
) -> Task<Result<ModelHandle<ChannelChat>>> {
let client = self.client.clone();
let user_store = self.user_store.clone();
let this = cx.handle();
self.open_channel_resource(
channel_id,
|this| &mut this.opened_chats,
|channel, cx| ChannelChat::new(channel, user_store, client, cx),
|channel, cx| ChannelChat::new(channel, this, user_store, client, cx),
cx,
)
}
@@ -779,6 +840,8 @@ impl ChannelStore {
Arc::new(Channel {
id: channel.id,
name: channel.name,
unseen_note_version: None,
unseen_message_id: None,
}),
),
}
@@ -787,7 +850,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();
|| !payload.delete_edge.is_empty()
|| !payload.unseen_channel_messages.is_empty()
|| !payload.unseen_channel_buffer_changes.is_empty();
if channels_changed {
if !payload.delete_channels.is_empty() {
@@ -814,6 +879,22 @@ impl ChannelStore {
index.insert(channel)
}
for unseen_buffer_change in payload.unseen_channel_buffer_changes {
let version = language::proto::deserialize_version(&unseen_buffer_change.version);
index.note_changed(
unseen_buffer_change.channel_id,
unseen_buffer_change.epoch,
&version,
);
}
for unseen_channel_message in payload.unseen_channel_messages {
index.new_messages(
unseen_channel_message.channel_id,
unseen_channel_message.message_id,
);
}
for edge in payload.insert_edge {
index.insert_edge(edge.channel_id, edge.parent_id);
}

76
crates/channel/src/channel_store/channel_index.rs
View File

@@ -38,6 +38,43 @@ impl ChannelIndex {
channels_by_id: &mut self.channels_by_id,
}
}
pub fn acknowledge_note_version(
&mut self,
channel_id: ChannelId,
epoch: u64,
version: &clock::Global,
) {
if let Some(channel) = self.channels_by_id.get_mut(&channel_id) {
let channel = Arc::make_mut(channel);
if let Some((unseen_epoch, unseen_version)) = &channel.unseen_note_version {
if epoch > *unseen_epoch
|| epoch == *unseen_epoch && version.observed_all(unseen_version)
{
channel.unseen_note_version = None;
}
}
}
}
pub fn acknowledge_message_id(&mut self, channel_id: ChannelId, message_id: u64) {
if let Some(channel) = self.channels_by_id.get_mut(&channel_id) {
let channel = Arc::make_mut(channel);
if let Some(unseen_message_id) = channel.unseen_message_id {
if message_id >= unseen_message_id {
channel.unseen_message_id = None;
}
}
}
}
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);
}
pub fn new_message(&mut self, channel_id: ChannelId, message_id: u64) {
insert_new_message(&mut self.channels_by_id, channel_id, message_id)
}
}
impl Deref for ChannelIndex {
@@ -76,6 +113,14 @@ impl<'a> ChannelPathsInsertGuard<'a> {
}
}
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);
}
pub fn new_messages(&mut self, channel_id: ChannelId, message_id: u64) {
insert_new_message(&mut self.channels_by_id, channel_id, message_id)
}
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;
@@ -85,6 +130,8 @@ impl<'a> ChannelPathsInsertGuard<'a> {
Arc::new(Channel {
id: channel_proto.id,
name: channel_proto.name,
unseen_note_version: None,
unseen_message_id: None,
}),
);
self.insert_root(channel_proto.id);
@@ -160,3 +207,32 @@ fn channel_path_sorting_key<'a>(
path.iter()
.map(|id| Some(channels_by_id.get(id)?.name.as_str()))
}
fn insert_note_changed(
channels_by_id: &mut BTreeMap<ChannelId, Arc<Channel>>,
channel_id: u64,
epoch: u64,
version: &clock::Global,
) {
if let Some(channel) = channels_by_id.get_mut(&channel_id) {
let unseen_version = Arc::make_mut(channel)
.unseen_note_version
.get_or_insert((0, clock::Global::new()));
if epoch > unseen_version.0 {
*unseen_version = (epoch, version.clone());
} else {
unseen_version.1.join(&version);
}
}
}
fn insert_new_message(
channels_by_id: &mut BTreeMap<ChannelId, Arc<Channel>>,
channel_id: u64,
message_id: u64,
) {
if let Some(channel) = channels_by_id.get_mut(&channel_id) {
let unseen_message_id = Arc::make_mut(channel).unseen_message_id.get_or_insert(0);
*unseen_message_id = message_id.max(*unseen_message_id);
}
}

2
crates/client/Cargo.toml
View File

@@ -41,8 +41,8 @@ tempfile = "3"
thiserror.workspace = true
time.workspace = true
tiny_http = "0.8"
uuid.workspace = true
url = "2.2"
uuid = { version = "1.1.2", features = ["v4"] }
[dev-dependencies]
collections = { path = "../collections", features = ["test-support"] }

28
crates/client/src/client.rs
View File

@@ -34,7 +34,7 @@ use std::{
future::Future,
marker::PhantomData,
path::PathBuf,
sync::{Arc, Weak},
sync::{atomic::AtomicU64, Arc, Weak},
time::{Duration, Instant},
};
use telemetry::Telemetry;
@@ -105,7 +105,7 @@ pub fn init(client: &Arc<Client>, cx: &mut AppContext) {
}
pub struct Client {
id: usize,
id: AtomicU64,
peer: Arc<Peer>,
http: Arc<dyn HttpClient>,
telemetry: Arc<Telemetry>,
@@ -374,7 +374,7 @@ impl settings::Setting for TelemetrySettings {
impl Client {
pub fn new(http: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
Arc::new(Self {
id: 0,
id: AtomicU64::new(0),
peer: Peer::new(0),
telemetry: Telemetry::new(http.clone(), cx),
http,
@@ -387,17 +387,16 @@ impl Client {
})
}
pub fn id(&self) -> usize {
self.id
pub fn id(&self) -> u64 {
self.id.load(std::sync::atomic::Ordering::SeqCst)
}
pub fn http_client(&self) -> Arc<dyn HttpClient> {
self.http.clone()
}
#[cfg(any(test, feature = "test-support"))]
pub fn set_id(&mut self, id: usize) -> &Self {
self.id = id;
pub fn set_id(&self, id: u64) -> &Self {
self.id.store(id, std::sync::atomic::Ordering::SeqCst);
self
}
@@ -454,7 +453,7 @@ impl Client {
}
fn set_status(self: &Arc<Self>, status: Status, cx: &AsyncAppContext) {
log::info!("set status on client {}: {:?}", self.id, status);
log::info!("set status on client {}: {:?}", self.id(), status);
let mut state = self.state.write();
*state.status.0.borrow_mut() = status;
@@ -805,6 +804,7 @@ impl Client {
}
}
let credentials = credentials.unwrap();
self.set_id(credentials.user_id);
if was_disconnected {
self.set_status(Status::Connecting, cx);
@@ -1221,7 +1221,7 @@ impl Client {
}
pub fn send<T: EnvelopedMessage>(&self, message: T) -> Result<()> {
log::debug!("rpc send. client_id:{}, name:{}", self.id, T::NAME);
log::debug!("rpc send. client_id:{}, name:{}", self.id(), T::NAME);
self.peer.send(self.connection_id()?, message)
}
@@ -1237,7 +1237,7 @@ impl Client {
&self,
request: T,
) -> impl Future<Output = Result<TypedEnvelope<T::Response>>> {
let client_id = self.id;
let client_id = self.id();
log::debug!(
"rpc request start. client_id:{}. name:{}",
client_id,
@@ -1258,7 +1258,7 @@ impl Client {
}
fn respond<T: RequestMessage>(&self, receipt: Receipt<T>, response: T::Response) -> Result<()> {
log::debug!("rpc respond. client_id:{}. name:{}", self.id, T::NAME);
log::debug!("rpc respond. client_id:{}. name:{}", self.id(), T::NAME);
self.peer.respond(receipt, response)
}
@@ -1267,7 +1267,7 @@ impl Client {
receipt: Receipt<T>,
error: proto::Error,
) -> Result<()> {
log::debug!("rpc respond. client_id:{}. name:{}", self.id, T::NAME);
log::debug!("rpc respond. client_id:{}. name:{}", self.id(), T::NAME);
self.peer.respond_with_error(receipt, error)
}
@@ -1336,7 +1336,7 @@ impl Client {
if let Some(handler) = handler {
let future = handler(subscriber, message, &self, cx.clone());
let client_id = self.id;
let client_id = self.id();
log::debug!(
"rpc message received. client_id:{}, sender_id:{:?}, type:{}",
client_id,

11
crates/collab/Cargo.toml
View File

@@ -42,14 +42,12 @@ rand.workspace = true
reqwest = { version = "0.11", features = ["json"], optional = true }
scrypt = "0.7"
smallvec.workspace = true
# Remove fork dependency when a version with https://github.com/SeaQL/sea-orm/pull/1283 is released.
sea-orm = { git = "https://github.com/zed-industries/sea-orm", rev = "18f4c691085712ad014a51792af75a9044bacee6", features = ["sqlx-postgres", "postgres-array", "runtime-tokio-rustls"] }
sea-query = "0.27"
sea-orm = { version = "0.12.x", features = ["sqlx-postgres", "postgres-array", "runtime-tokio-rustls", "with-uuid"] }
serde.workspace = true
serde_derive.workspace = true
serde_json.workspace = true
sha-1 = "0.9"
sqlx = { version = "0.6", features = ["runtime-tokio-rustls", "postgres", "json", "time", "uuid", "any"] }
sqlx = { version = "0.7", features = ["runtime-tokio-rustls", "postgres", "json", "time", "uuid", "any"] }
time.workspace = true
tokio = { version = "1", features = ["full"] }
tokio-tungstenite = "0.17"
@@ -59,6 +57,7 @@ toml.workspace = true
tracing = "0.1.34"
tracing-log = "0.1.3"
tracing-subscriber = { version = "0.3.11", features = ["env-filter", "json"] }
uuid.workspace = true
[dev-dependencies]
audio = { path = "../audio" }
@@ -87,9 +86,9 @@ env_logger.workspace = true
indoc.workspace = true
util = { path = "../util" }
lazy_static.workspace = true
sea-orm = { git = "https://github.com/zed-industries/sea-orm", rev = "18f4c691085712ad014a51792af75a9044bacee6", features = ["sqlx-sqlite"] }
sea-orm = { version = "0.12.x", features = ["sqlx-sqlite"] }
serde_json.workspace = true
sqlx = { version = "0.6", features = ["sqlite"] }
sqlx = { version = "0.7", features = ["sqlite"] }
unindent.workspace = true
[features]

21
crates/collab/migrations.sqlite/20221109000000_test_schema.sql
View File

@@ -289,3 +289,24 @@ CREATE TABLE "user_features" (
CREATE UNIQUE INDEX "index_user_features_user_id_and_feature_id" ON "user_features" ("user_id", "feature_id");
CREATE INDEX "index_user_features_on_user_id" ON "user_features" ("user_id");
CREATE INDEX "index_user_features_on_feature_id" ON "user_features" ("feature_id");
CREATE TABLE "observed_buffer_edits" (
"user_id" INTEGER NOT NULL REFERENCES users (id) ON DELETE CASCADE,
"buffer_id" INTEGER NOT NULL REFERENCES buffers (id) ON DELETE CASCADE,
"epoch" INTEGER NOT NULL,
"lamport_timestamp" INTEGER NOT NULL,
"replica_id" INTEGER NOT NULL,
PRIMARY KEY (user_id, buffer_id)
);
CREATE UNIQUE INDEX "index_observed_buffers_user_and_buffer_id" ON "observed_buffer_edits" ("user_id", "buffer_id");
CREATE TABLE IF NOT EXISTS "observed_channel_messages" (
"user_id" INTEGER NOT NULL REFERENCES users (id) ON DELETE CASCADE,
"channel_id" INTEGER NOT NULL REFERENCES channels (id) ON DELETE CASCADE,
"channel_message_id" INTEGER NOT NULL,
PRIMARY KEY (user_id, channel_id)
);
CREATE UNIQUE INDEX "index_observed_channel_messages_user_and_channel_id" ON "observed_channel_messages" ("user_id", "channel_id");

19
crates/collab/migrations/20230925210437_add_channel_changes.sql Normal file
View File

@@ -0,0 +1,19 @@
CREATE TABLE IF NOT EXISTS "observed_buffer_edits" (
"user_id" INTEGER NOT NULL REFERENCES users (id) ON DELETE CASCADE,
"buffer_id" INTEGER NOT NULL REFERENCES buffers (id) ON DELETE CASCADE,
"epoch" INTEGER NOT NULL,
"lamport_timestamp" INTEGER NOT NULL,
"replica_id" INTEGER NOT NULL,
PRIMARY KEY (user_id, buffer_id)
);
CREATE UNIQUE INDEX "index_observed_buffer_user_and_buffer_id" ON "observed_buffer_edits" ("user_id", "buffer_id");
CREATE TABLE IF NOT EXISTS "observed_channel_messages" (
"user_id" INTEGER NOT NULL REFERENCES users (id) ON DELETE CASCADE,
"channel_id" INTEGER NOT NULL REFERENCES channels (id) ON DELETE CASCADE,
"channel_message_id" INTEGER NOT NULL,
PRIMARY KEY (user_id, channel_id)
);
CREATE UNIQUE INDEX "index_observed_channel_messages_user_and_channel_id" ON "observed_channel_messages" ("user_id", "channel_id");

16
crates/collab/src/db.rs
View File

@@ -19,11 +19,12 @@ use rpc::{
ConnectionId,
};
use sea_orm::{
entity::prelude::*, ActiveValue, Condition, ConnectionTrait, DatabaseConnection,
DatabaseTransaction, DbErr, FromQueryResult, IntoActiveModel, IsolationLevel, JoinType,
QueryOrder, QuerySelect, Statement, TransactionTrait,
entity::prelude::*,
sea_query::{Alias, Expr, OnConflict, Query},
ActiveValue, Condition, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbErr,
FromQueryResult, IntoActiveModel, IsolationLevel, JoinType, QueryOrder, QuerySelect, Statement,
TransactionTrait,
};
use sea_query::{Alias, Expr, OnConflict, Query};
use serde::{Deserialize, Serialize};
use sqlx::{
migrate::{Migrate, Migration, MigrationSource},
@@ -62,6 +63,7 @@ pub struct Database {
// separate files in the `queries` folder.
impl Database {
pub async fn new(options: ConnectOptions, executor: Executor) -> Result<Self> {
sqlx::any::install_default_drivers();
Ok(Self {
options: options.clone(),
pool: sea_orm::Database::connect(options).await?,
@@ -119,7 +121,7 @@ impl Database {
Ok(new_migrations)
}
async fn transaction<F, Fut, T>(&self, f: F) -> Result<T>
pub async fn transaction<F, Fut, T>(&self, f: F) -> Result<T>
where
F: Send + Fn(TransactionHandle) -> Fut,
Fut: Send + Future<Output = Result<T>>,
@@ -321,7 +323,7 @@ fn is_serialization_error(error: &Error) -> bool {
}
}
struct TransactionHandle(Arc<Option<DatabaseTransaction>>);
pub struct TransactionHandle(Arc<Option<DatabaseTransaction>>);
impl Deref for TransactionHandle {
type Target = DatabaseTransaction;
@@ -437,6 +439,8 @@ pub struct ChannelsForUser {
pub channels: ChannelGraph,
pub channel_participants: HashMap<ChannelId, Vec<UserId>>,
pub channels_with_admin_privileges: HashSet<ChannelId>,
pub unseen_buffer_changes: Vec<proto::UnseenChannelBufferChange>,
pub channel_messages: Vec<proto::UnseenChannelMessage>,
}
#[derive(Debug)]

54
crates/collab/src/db/ids.rs
View File

@@ -1,6 +1,5 @@
use crate::Result;
use sea_orm::DbErr;
use sea_query::{Value, ValueTypeErr};
use sea_orm::{entity::prelude::*, DbErr};
use serde::{Deserialize, Serialize};
macro_rules! id_type {
@@ -17,6 +16,7 @@ macro_rules! id_type {
Hash,
Serialize,
Deserialize,
DeriveValueType,
)]
#[serde(transparent)]
pub struct $name(pub i32);
@@ -42,40 +42,6 @@ macro_rules! id_type {
}
}
impl From<$name> for sea_query::Value {
fn from(value: $name) -> Self {
sea_query::Value::Int(Some(value.0))
}
}
impl sea_orm::TryGetable for $name {
fn try_get(
res: &sea_orm::QueryResult,
pre: &str,
col: &str,
) -> Result<Self, sea_orm::TryGetError> {
Ok(Self(i32::try_get(res, pre, col)?))
}
}
impl sea_query::ValueType for $name {
fn try_from(v: Value) -> Result<Self, sea_query::ValueTypeErr> {
Ok(Self(value_to_integer(v)?))
}
fn type_name() -> String {
stringify!($name).into()
}
fn array_type() -> sea_query::ArrayType {
sea_query::ArrayType::Int
}
fn column_type() -> sea_query::ColumnType {
sea_query::ColumnType::Integer(None)
}
}
impl sea_orm::TryFromU64 for $name {
fn try_from_u64(n: u64) -> Result<Self, DbErr> {
Ok(Self(n.try_into().map_err(|_| {
@@ -88,7 +54,7 @@ macro_rules! id_type {
}
}
impl sea_query::Nullable for $name {
impl sea_orm::sea_query::Nullable for $name {
fn null() -> Value {
Value::Int(None)
}
@@ -96,20 +62,6 @@ macro_rules! id_type {
};
}
fn value_to_integer(v: Value) -> Result<i32, ValueTypeErr> {
match v {
Value::TinyInt(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::SmallInt(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::Int(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::BigInt(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::TinyUnsigned(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::SmallUnsigned(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::Unsigned(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
Value::BigUnsigned(Some(int)) => int.try_into().map_err(|_| ValueTypeErr),
_ => Err(ValueTypeErr),
}
}
id_type!(BufferId);
id_type!(AccessTokenId);
id_type!(ChannelChatParticipantId);

10
crates/collab/src/db/queries.rs
View File

@@ -9,3 +9,13 @@ pub mod projects;
pub mod rooms;
pub mod servers;
pub mod users;
fn max_assign<T: Ord>(max: &mut Option<T>, val: T) {
if let Some(max_val) = max {
if val > *max_val {
*max = Some(val);
}
} else {
*max = Some(val);
}
}

326
crates/collab/src/db/queries/buffers.rs
View File

@@ -74,7 +74,32 @@ impl Database {
.await?;
collaborators.push(collaborator);
let (base_text, operations) = self.get_buffer_state(&buffer, &tx).await?;
let (base_text, operations, max_operation) =
self.get_buffer_state(&buffer, &tx).await?;
// Save the last observed operation
if let Some(op) = max_operation {
observed_buffer_edits::Entity::insert(observed_buffer_edits::ActiveModel {
user_id: ActiveValue::Set(user_id),
buffer_id: ActiveValue::Set(buffer.id),
epoch: ActiveValue::Set(op.epoch),
lamport_timestamp: ActiveValue::Set(op.lamport_timestamp),
replica_id: ActiveValue::Set(op.replica_id),
})
.on_conflict(
OnConflict::columns([
observed_buffer_edits::Column::UserId,
observed_buffer_edits::Column::BufferId,
])
.update_columns([
observed_buffer_edits::Column::Epoch,
observed_buffer_edits::Column::LamportTimestamp,
])
.to_owned(),
)
.exec(&*tx)
.await?;
}
Ok(proto::JoinChannelBufferResponse {
buffer_id: buffer.id.to_proto(),
@@ -373,33 +398,46 @@ impl Database {
channel_id: ChannelId,
) -> Result<Vec<UserId>> {
self.transaction(|tx| async move {
#[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]
enum QueryUserIds {
UserId,
}
let users: Vec<UserId> = channel_buffer_collaborator::Entity::find()
.select_only()
.column(channel_buffer_collaborator::Column::UserId)
.filter(
Condition::all()
.add(channel_buffer_collaborator::Column::ChannelId.eq(channel_id)),
)
.into_values::<_, QueryUserIds>()
.all(&*tx)
.await?;
Ok(users)
self.get_channel_buffer_collaborators_internal(channel_id, &*tx)
.await
})
.await
}
async fn get_channel_buffer_collaborators_internal(
&self,
channel_id: ChannelId,
tx: &DatabaseTransaction,
) -> Result<Vec<UserId>> {
#[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]
enum QueryUserIds {
UserId,
}
let users: Vec<UserId> = channel_buffer_collaborator::Entity::find()
.select_only()
.column(channel_buffer_collaborator::Column::UserId)
.filter(
Condition::all().add(channel_buffer_collaborator::Column::ChannelId.eq(channel_id)),
)
.into_values::<_, QueryUserIds>()
.all(&*tx)
.await?;
Ok(users)
}
pub async fn update_channel_buffer(
&self,
channel_id: ChannelId,
user: UserId,
operations: &[proto::Operation],
) -> Result<Vec<ConnectionId>> {
) -> Result<(
Vec<ConnectionId>,
Vec<UserId>,
i32,
Vec<proto::VectorClockEntry>,
)> {
self.transaction(move |tx| async move {
self.check_user_is_channel_member(channel_id, user, &*tx)
.await?;
@@ -418,7 +456,38 @@ impl Database {
.iter()
.filter_map(|op| operation_to_storage(op, &buffer, serialization_version))
.collect::<Vec<_>>();
let mut channel_members;
let max_version;
if !operations.is_empty() {
let max_operation = operations
.iter()
.max_by_key(|op| (op.lamport_timestamp.as_ref(), op.replica_id.as_ref()))
.unwrap();
max_version = vec![proto::VectorClockEntry {
replica_id: *max_operation.replica_id.as_ref() as u32,
timestamp: *max_operation.lamport_timestamp.as_ref() as u32,
}];
// get current channel participants and save the max operation above
self.save_max_operation(
user,
buffer.id,
buffer.epoch,
*max_operation.replica_id.as_ref(),
*max_operation.lamport_timestamp.as_ref(),
&*tx,
)
.await?;
channel_members = self.get_channel_members_internal(channel_id, &*tx).await?;
let collaborators = self
.get_channel_buffer_collaborators_internal(channel_id, &*tx)
.await?;
channel_members.retain(|member| !collaborators.contains(member));
buffer_operation::Entity::insert_many(operations)
.on_conflict(
OnConflict::columns([
@@ -432,6 +501,9 @@ impl Database {
)
.exec(&*tx)
.await?;
} else {
channel_members = Vec::new();
max_version = Vec::new();
}
let mut connections = Vec::new();
@@ -450,11 +522,53 @@ impl Database {
});
}
Ok(connections)
Ok((connections, channel_members, buffer.epoch, max_version))
})
.await
}
async fn save_max_operation(
&self,
user_id: UserId,
buffer_id: BufferId,
epoch: i32,
replica_id: i32,
lamport_timestamp: i32,
tx: &DatabaseTransaction,
) -> Result<()> {
use observed_buffer_edits::Column;
observed_buffer_edits::Entity::insert(observed_buffer_edits::ActiveModel {
user_id: ActiveValue::Set(user_id),
buffer_id: ActiveValue::Set(buffer_id),
epoch: ActiveValue::Set(epoch),
replica_id: ActiveValue::Set(replica_id),
lamport_timestamp: ActiveValue::Set(lamport_timestamp),
})
.on_conflict(
OnConflict::columns([Column::UserId, Column::BufferId])
.update_columns([Column::Epoch, Column::LamportTimestamp, Column::ReplicaId])
.action_cond_where(
Condition::any().add(Column::Epoch.lt(epoch)).add(
Condition::all().add(Column::Epoch.eq(epoch)).add(
Condition::any()
.add(Column::LamportTimestamp.lt(lamport_timestamp))
.add(
Column::LamportTimestamp
.eq(lamport_timestamp)
.and(Column::ReplicaId.lt(replica_id)),
),
),
),
)
.to_owned(),
)
.exec_without_returning(tx)
.await?;
Ok(())
}
async fn get_buffer_operation_serialization_version(
&self,
buffer_id: BufferId,
@@ -472,7 +586,7 @@ impl Database {
.ok_or_else(|| anyhow!("missing buffer snapshot"))?)
}
async fn get_channel_buffer(
pub async fn get_channel_buffer(
&self,
channel_id: ChannelId,
tx: &DatabaseTransaction,
@@ -491,7 +605,11 @@ impl Database {
&self,
buffer: &buffer::Model,
tx: &DatabaseTransaction,
) -> Result<(String, Vec<proto::Operation>)> {
) -> Result<(
String,
Vec<proto::Operation>,
Option<buffer_operation::Model>,
)> {
let id = buffer.id;
let (base_text, version) = if buffer.epoch > 0 {
let snapshot = buffer_snapshot::Entity::find()
@@ -516,16 +634,28 @@ impl Database {
.eq(id)
.and(buffer_operation::Column::Epoch.eq(buffer.epoch)),
)
.order_by_asc(buffer_operation::Column::LamportTimestamp)
.order_by_asc(buffer_operation::Column::ReplicaId)
.stream(&*tx)
.await?;
let mut operations = Vec::new();
let mut last_row = None;
while let Some(row) = rows.next().await {
let row = row?;
last_row = Some(buffer_operation::Model {
buffer_id: row.buffer_id,
epoch: row.epoch,
lamport_timestamp: row.lamport_timestamp,
replica_id: row.lamport_timestamp,
value: Default::default(),
});
operations.push(proto::Operation {
variant: Some(operation_from_storage(row?, version)?),
})
variant: Some(operation_from_storage(row, version)?),
});
}
Ok((base_text, operations))
Ok((base_text, operations, last_row))
}
async fn snapshot_channel_buffer(
@@ -534,7 +664,7 @@ impl Database {
tx: &DatabaseTransaction,
) -> Result<()> {
let buffer = self.get_channel_buffer(channel_id, tx).await?;
let (base_text, operations) = self.get_buffer_state(&buffer, tx).await?;
let (base_text, operations, _) = self.get_buffer_state(&buffer, tx).await?;
if operations.is_empty() {
return Ok(());
}
@@ -567,6 +697,150 @@ impl Database {
Ok(())
}
pub async fn observe_buffer_version(
&self,
buffer_id: BufferId,
user_id: UserId,
epoch: i32,
version: &[proto::VectorClockEntry],
) -> Result<()> {
self.transaction(|tx| async move {
// For now, combine concurrent operations.
let Some(component) = version.iter().max_by_key(|version| version.timestamp) else {
return Ok(());
};
self.save_max_operation(
user_id,
buffer_id,
epoch,
component.replica_id as i32,
component.timestamp as i32,
&*tx,
)
.await?;
Ok(())
})
.await
}
pub async fn unseen_channel_buffer_changes(
&self,
user_id: UserId,
channel_ids: &[ChannelId],
tx: &DatabaseTransaction,
) -> Result<Vec<proto::UnseenChannelBufferChange>> {
#[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]
enum QueryIds {
ChannelId,
Id,
}
let mut channel_ids_by_buffer_id = HashMap::default();
let mut rows = buffer::Entity::find()
.filter(buffer::Column::ChannelId.is_in(channel_ids.iter().copied()))
.stream(&*tx)
.await?;
while let Some(row) = rows.next().await {
let row = row?;
channel_ids_by_buffer_id.insert(row.id, row.channel_id);
}
drop(rows);
let mut observed_edits_by_buffer_id = HashMap::default();
let mut rows = observed_buffer_edits::Entity::find()
.filter(observed_buffer_edits::Column::UserId.eq(user_id))
.filter(
observed_buffer_edits::Column::BufferId
.is_in(channel_ids_by_buffer_id.keys().copied()),
)
.stream(&*tx)
.await?;
while let Some(row) = rows.next().await {
let row = row?;
observed_edits_by_buffer_id.insert(row.buffer_id, row);
}
drop(rows);
let latest_operations = self
.get_latest_operations_for_buffers(channel_ids_by_buffer_id.keys().copied(), &*tx)
.await?;
let mut changes = Vec::default();
for latest in latest_operations {
if let Some(observed) = observed_edits_by_buffer_id.get(&latest.buffer_id) {
if (
observed.epoch,
observed.lamport_timestamp,
observed.replica_id,
) >= (latest.epoch, latest.lamport_timestamp, latest.replica_id)
{
continue;
}
}
if let Some(channel_id) = channel_ids_by_buffer_id.get(&latest.buffer_id) {
changes.push(proto::UnseenChannelBufferChange {
channel_id: channel_id.to_proto(),
epoch: latest.epoch as u64,
version: vec![proto::VectorClockEntry {
replica_id: latest.replica_id as u32,
timestamp: latest.lamport_timestamp as u32,
}],
});
}
}
Ok(changes)
}
pub async fn get_latest_operations_for_buffers(
&self,
buffer_ids: impl IntoIterator<Item = BufferId>,
tx: &DatabaseTransaction,
) -> Result<Vec<buffer_operation::Model>> {
let mut values = String::new();
for id in buffer_ids {
if !values.is_empty() {
values.push_str(", ");
}
write!(&mut values, "({})", id).unwrap();
}
if values.is_empty() {
return Ok(Vec::default());
}
let sql = format!(
r#"
SELECT
*
FROM
(
SELECT
*,
row_number() OVER (
PARTITION BY buffer_id
ORDER BY
epoch DESC,
lamport_timestamp DESC,
replica_id DESC
) as row_number
FROM buffer_operations
WHERE
buffer_id in ({values})
) AS last_operations
WHERE
row_number = 1
"#,
);
let stmt = Statement::from_string(self.pool.get_database_backend(), sql);
Ok(buffer_operation::Entity::find()
.from_raw_sql(stmt)
.all(&*tx)
.await?)
}
}
fn operation_to_storage(

23
crates/collab/src/db/queries/channels.rs
View File

@@ -1,8 +1,7 @@
use super::*;
use rpc::proto::ChannelEdge;
use smallvec::SmallVec;
use super::*;
type ChannelDescendants = HashMap<ChannelId, SmallSet<ChannelId>>;
impl Database {
@@ -391,7 +390,8 @@ impl Database {
.all(&*tx)
.await?;
self.get_user_channels(channel_memberships, &tx).await
self.get_user_channels(user_id, channel_memberships, &tx)
.await
})
.await
}
@@ -414,13 +414,15 @@ impl Database {
.all(&*tx)
.await?;
self.get_user_channels(channel_membership, &tx).await
self.get_user_channels(user_id, channel_membership, &tx)
.await
})
.await
}
pub async fn get_user_channels(
&self,
user_id: UserId,
channel_memberships: Vec<channel_member::Model>,
tx: &DatabaseTransaction,
) -> Result<ChannelsForUser> {
@@ -460,10 +462,21 @@ impl Database {
}
}
let channel_ids = graph.channels.iter().map(|c| c.id).collect::<Vec<_>>();
let channel_buffer_changes = self
.unseen_channel_buffer_changes(user_id, &channel_ids, &*tx)
.await?;
let unseen_messages = self
.unseen_channel_messages(user_id, &channel_ids, &*tx)
.await?;
Ok(ChannelsForUser {
channels: graph,
channel_participants,
channels_with_admin_privileges,
unseen_buffer_changes: channel_buffer_changes,
channel_messages: unseen_messages,
})
}
@@ -645,7 +658,7 @@ impl Database {
) -> Result<Vec<ChannelId>> {
let paths = channel_path::Entity::find()
.filter(channel_path::Column::ChannelId.eq(channel_id))
.order_by(channel_path::Column::IdPath, sea_query::Order::Desc)
.order_by(channel_path::Column::IdPath, sea_orm::Order::Desc)
.all(tx)
.await?;
let mut channel_ids = Vec::new();

4
crates/collab/src/db/queries/contacts.rs
View File

@@ -18,12 +18,12 @@ impl Database {
let user_b_participant = Alias::new("user_b_participant");
let mut db_contacts = contact::Entity::find()
.column_as(
Expr::tbl(user_a_participant.clone(), room_participant::Column::Id)
Expr::col((user_a_participant.clone(), room_participant::Column::Id))
.is_not_null(),
"user_a_busy",
)
.column_as(
Expr::tbl(user_b_participant.clone(), room_participant::Column::Id)
Expr::col((user_b_participant.clone(), room_participant::Column::Id))
.is_not_null(),
"user_b_busy",
)

184
crates/collab/src/db/queries/messages.rs
View File

@@ -93,9 +93,13 @@ impl Database {
.stream(&*tx)
.await?;
let mut max_id = None;
let mut messages = Vec::new();
while let Some(row) = rows.next().await {
let row = row?;
max_assign(&mut max_id, row.id);
let nonce = row.nonce.as_u64_pair();
messages.push(proto::ChannelMessage {
id: row.id.to_proto(),
@@ -108,6 +112,50 @@ impl Database {
}),
});
}
drop(rows);
if let Some(max_id) = max_id {
let has_older_message = observed_channel_messages::Entity::find()
.filter(
observed_channel_messages::Column::UserId
.eq(user_id)
.and(observed_channel_messages::Column::ChannelId.eq(channel_id))
.and(observed_channel_messages::Column::ChannelMessageId.lt(max_id)),
)
.one(&*tx)
.await?
.is_some();
if has_older_message {
observed_channel_messages::Entity::update(
observed_channel_messages::ActiveModel {
user_id: ActiveValue::Unchanged(user_id),
channel_id: ActiveValue::Unchanged(channel_id),
channel_message_id: ActiveValue::Set(max_id),
},
)
.exec(&*tx)
.await?;
} else {
observed_channel_messages::Entity::insert(
observed_channel_messages::ActiveModel {
user_id: ActiveValue::Set(user_id),
channel_id: ActiveValue::Set(channel_id),
channel_message_id: ActiveValue::Set(max_id),
},
)
.on_conflict(
OnConflict::columns([
observed_channel_messages::Column::UserId,
observed_channel_messages::Column::ChannelId,
])
.update_columns([observed_channel_messages::Column::ChannelMessageId])
.to_owned(),
)
.exec(&*tx)
.await?;
}
}
Ok(messages)
})
@@ -121,7 +169,7 @@ impl Database {
body: &str,
timestamp: OffsetDateTime,
nonce: u128,
) -> Result<(MessageId, Vec<ConnectionId>)> {
) -> Result<(MessageId, Vec<ConnectionId>, Vec<UserId>)> {
self.transaction(|tx| async move {
let mut rows = channel_chat_participant::Entity::find()
.filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
@@ -130,11 +178,13 @@ impl Database {
let mut is_participant = false;
let mut participant_connection_ids = Vec::new();
let mut participant_user_ids = Vec::new();
while let Some(row) = rows.next().await {
let row = row?;
if row.user_id == user_id {
is_participant = true;
}
participant_user_ids.push(row.user_id);
participant_connection_ids.push(row.connection());
}
drop(rows);
@@ -167,11 +217,141 @@ impl Database {
ConnectionId,
}
Ok((message.last_insert_id, participant_connection_ids))
// Observe this message for the sender
self.observe_channel_message_internal(
channel_id,
user_id,
message.last_insert_id,
&*tx,
)
.await?;
let mut channel_members = self.get_channel_members_internal(channel_id, &*tx).await?;
channel_members.retain(|member| !participant_user_ids.contains(member));
Ok((
message.last_insert_id,
participant_connection_ids,
channel_members,
))
})
.await
}
pub async fn observe_channel_message(
&self,
channel_id: ChannelId,
user_id: UserId,
message_id: MessageId,
) -> Result<()> {
self.transaction(|tx| async move {
self.observe_channel_message_internal(channel_id, user_id, message_id, &*tx)
.await?;
Ok(())
})
.await
}
async fn observe_channel_message_internal(
&self,
channel_id: ChannelId,
user_id: UserId,
message_id: MessageId,
tx: &DatabaseTransaction,
) -> Result<()> {
observed_channel_messages::Entity::insert(observed_channel_messages::ActiveModel {
user_id: ActiveValue::Set(user_id),
channel_id: ActiveValue::Set(channel_id),
channel_message_id: ActiveValue::Set(message_id),
})
.on_conflict(
OnConflict::columns([
observed_channel_messages::Column::ChannelId,
observed_channel_messages::Column::UserId,
])
.update_column(observed_channel_messages::Column::ChannelMessageId)
.action_cond_where(observed_channel_messages::Column::ChannelMessageId.lt(message_id))
.to_owned(),
)
// TODO: Try to upgrade SeaORM so we don't have to do this hack around their bug
.exec_without_returning(&*tx)
.await?;
Ok(())
}
pub async fn unseen_channel_messages(
&self,
user_id: UserId,
channel_ids: &[ChannelId],
tx: &DatabaseTransaction,
) -> Result<Vec<proto::UnseenChannelMessage>> {
let mut observed_messages_by_channel_id = HashMap::default();
let mut rows = observed_channel_messages::Entity::find()
.filter(observed_channel_messages::Column::UserId.eq(user_id))
.filter(observed_channel_messages::Column::ChannelId.is_in(channel_ids.iter().copied()))
.stream(&*tx)
.await?;
while let Some(row) = rows.next().await {
let row = row?;
observed_messages_by_channel_id.insert(row.channel_id, row);
}
drop(rows);
let mut values = String::new();
for id in channel_ids {
if !values.is_empty() {
values.push_str(", ");
}
write!(&mut values, "({})", id).unwrap();
}
if values.is_empty() {
return Ok(Default::default());
}
let sql = format!(
r#"
SELECT
*
FROM (
SELECT
*,
row_number() OVER (
PARTITION BY channel_id
ORDER BY id DESC
) as row_number
FROM channel_messages
WHERE
channel_id in ({values})
) AS messages
WHERE
row_number = 1
"#,
);
let stmt = Statement::from_string(self.pool.get_database_backend(), sql);
let last_messages = channel_message::Model::find_by_statement(stmt)
.all(&*tx)
.await?;
let mut changes = Vec::new();
for last_message in last_messages {
if let Some(observed_message) =
observed_messages_by_channel_id.get(&last_message.channel_id)
{
if observed_message.channel_message_id == last_message.id {
continue;
}
}
changes.push(proto::UnseenChannelMessage {
channel_id: last_message.channel_id.to_proto(),
message_id: last_message.id.to_proto(),
});
}
Ok(changes)
}
pub async fn remove_channel_message(
&self,
channel_id: ChannelId,

2
crates/collab/src/db/queries/users.rs
View File

@@ -184,7 +184,7 @@ impl Database {
Ok(user::Entity::find()
.from_raw_sql(Statement::from_sql_and_values(
self.pool.get_database_backend(),
query.into(),
query,
vec![like_string.into(), name_query.into(), limit.into()],
))
.all(&*tx)

2
crates/collab/src/db/tables.rs
View File

@@ -12,6 +12,8 @@ pub mod contact;
pub mod feature_flag;
pub mod follower;
pub mod language_server;
pub mod observed_buffer_edits;
pub mod observed_channel_messages;
pub mod project;
pub mod project_collaborator;
pub mod room;

43
crates/collab/src/db/tables/observed_buffer_edits.rs Normal file
View File

@@ -0,0 +1,43 @@
use crate::db::{BufferId, UserId};
use sea_orm::entity::prelude::*;
#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
#[sea_orm(table_name = "observed_buffer_edits")]
pub struct Model {
#[sea_orm(primary_key)]
pub user_id: UserId,
pub buffer_id: BufferId,
pub epoch: i32,
pub lamport_timestamp: i32,
pub replica_id: i32,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {
#[sea_orm(
belongs_to = "super::buffer::Entity",
from = "Column::BufferId",
to = "super::buffer::Column::Id"
)]
Buffer,
#[sea_orm(
belongs_to = "super::user::Entity",
from = "Column::UserId",
to = "super::user::Column::Id"
)]
User,
}
impl Related<super::buffer::Entity> for Entity {
fn to() -> RelationDef {
Relation::Buffer.def()
}
}
impl Related<super::user::Entity> for Entity {
fn to() -> RelationDef {
Relation::User.def()
}
}
impl ActiveModelBehavior for ActiveModel {}

41
crates/collab/src/db/tables/observed_channel_messages.rs Normal file
View File

@@ -0,0 +1,41 @@
use crate::db::{ChannelId, MessageId, UserId};
use sea_orm::entity::prelude::*;
#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
#[sea_orm(table_name = "observed_channel_messages")]
pub struct Model {
#[sea_orm(primary_key)]
pub user_id: UserId,
pub channel_id: ChannelId,
pub channel_message_id: MessageId,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {
#[sea_orm(
belongs_to = "super::channel::Entity",
from = "Column::ChannelId",
to = "super::channel::Column::Id"
)]
Channel,
#[sea_orm(
belongs_to = "super::user::Entity",
from = "Column::UserId",
to = "super::user::Column::Id"
)]
User,
}
impl Related<super::channel::Entity> for Entity {
fn to() -> RelationDef {
Relation::Channel.def()
}
}
impl Related<super::user::Entity> for Entity {
fn to() -> RelationDef {
Relation::User.def()
}
}
impl ActiveModelBehavior for ActiveModel {}

4
crates/collab/src/db/tests.rs
View File

@@ -39,7 +39,7 @@ impl TestDb {
db.pool
.execute(sea_orm::Statement::from_string(
db.pool.get_database_backend(),
sql.into(),
sql,
))
.await
.unwrap();
@@ -134,7 +134,7 @@ impl Drop for TestDb {
db.pool
.execute(sea_orm::Statement::from_string(
db.pool.get_database_backend(),
query.into(),
query,
))
.await
.log_err();

348
crates/collab/src/db/tests/buffer_tests.rs
View File

@@ -1,6 +1,6 @@
use super::*;
use crate::test_both_dbs;
use language::proto;
use language::proto::{self, serialize_version};
use text::Buffer;
test_both_dbs!(
@@ -163,3 +163,349 @@ async fn test_channel_buffers(db: &Arc<Database>) {
assert_eq!(buffer_response_b.base_text, "hello, cruel world");
assert_eq!(buffer_response_b.operations, &[]);
}
test_both_dbs!(
test_channel_buffers_last_operations,
test_channel_buffers_last_operations_postgres,
test_channel_buffers_last_operations_sqlite
);
async fn test_channel_buffers_last_operations(db: &Database) {
let user_id = db
.create_user(
"user_a@example.com",
false,
NewUserParams {
github_login: "user_a".into(),
github_user_id: 101,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let observer_id = db
.create_user(
"user_b@example.com",
false,
NewUserParams {
github_login: "user_b".into(),
github_user_id: 102,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let owner_id = db.create_server("production").await.unwrap().0 as u32;
let connection_id = ConnectionId {
owner_id,
id: user_id.0 as u32,
};
let mut buffers = Vec::new();
let mut text_buffers = Vec::new();
for i in 0..3 {
let channel = db
.create_root_channel(&format!("channel-{i}"), &format!("room-{i}"), user_id)
.await
.unwrap();
db.invite_channel_member(channel, observer_id, user_id, false)
.await
.unwrap();
db.respond_to_channel_invite(channel, observer_id, true)
.await
.unwrap();
db.join_channel_buffer(channel, user_id, connection_id)
.await
.unwrap();
buffers.push(
db.transaction(|tx| async move { db.get_channel_buffer(channel, &*tx).await })
.await
.unwrap(),
);
text_buffers.push(Buffer::new(0, 0, "".to_string()));
}
let operations = db
.transaction(|tx| {
let buffers = &buffers;
async move {
db.get_latest_operations_for_buffers([buffers[0].id, buffers[2].id], &*tx)
.await
}
})
.await
.unwrap();
assert!(operations.is_empty());
update_buffer(
buffers[0].channel_id,
user_id,
db,
vec![
text_buffers[0].edit([(0..0, "a")]),
text_buffers[0].edit([(0..0, "b")]),
text_buffers[0].edit([(0..0, "c")]),
],
)
.await;
update_buffer(
buffers[1].channel_id,
user_id,
db,
vec![
text_buffers[1].edit([(0..0, "d")]),
text_buffers[1].edit([(1..1, "e")]),
text_buffers[1].edit([(2..2, "f")]),
],
)
.await;
// cause buffer 1's epoch to increment.
db.leave_channel_buffer(buffers[1].channel_id, connection_id)
.await
.unwrap();
db.join_channel_buffer(buffers[1].channel_id, user_id, connection_id)
.await
.unwrap();
text_buffers[1] = Buffer::new(1, 0, "def".to_string());
update_buffer(
buffers[1].channel_id,
user_id,
db,
vec![
text_buffers[1].edit([(0..0, "g")]),
text_buffers[1].edit([(0..0, "h")]),
],
)
.await;
update_buffer(
buffers[2].channel_id,
user_id,
db,
vec![text_buffers[2].edit([(0..0, "i")])],
)
.await;
let operations = db
.transaction(|tx| {
let buffers = &buffers;
async move {
db.get_latest_operations_for_buffers([buffers[1].id, buffers[2].id], &*tx)
.await
}
})
.await
.unwrap();
assert_operations(
&operations,
&[
(buffers[1].id, 1, &text_buffers[1]),
(buffers[2].id, 0, &text_buffers[2]),
],
);
let operations = db
.transaction(|tx| {
let buffers = &buffers;
async move {
db.get_latest_operations_for_buffers([buffers[0].id, buffers[1].id], &*tx)
.await
}
})
.await
.unwrap();
assert_operations(
&operations,
&[
(buffers[0].id, 0, &text_buffers[0]),
(buffers[1].id, 1, &text_buffers[1]),
],
);
let buffer_changes = db
.transaction(|tx| {
let buffers = &buffers;
async move {
db.unseen_channel_buffer_changes(
observer_id,
&[
buffers[0].channel_id,
buffers[1].channel_id,
buffers[2].channel_id,
],
&*tx,
)
.await
}
})
.await
.unwrap();
pretty_assertions::assert_eq!(
buffer_changes,
[
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[0].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[0].version()),
},
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[1].channel_id.to_proto(),
epoch: 1,
version: serialize_version(&text_buffers[1].version())
.into_iter()
.filter(|vector| vector.replica_id
== buffer_changes[1].version.first().unwrap().replica_id)
.collect::<Vec<_>>(),
},
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[2].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[2].version()),
},
]
);
db.observe_buffer_version(
buffers[1].id,
observer_id,
1,
serialize_version(&text_buffers[1].version()).as_slice(),
)
.await
.unwrap();
let buffer_changes = db
.transaction(|tx| {
let buffers = &buffers;
async move {
db.unseen_channel_buffer_changes(
observer_id,
&[
buffers[0].channel_id,
buffers[1].channel_id,
buffers[2].channel_id,
],
&*tx,
)
.await
}
})
.await
.unwrap();
assert_eq!(
buffer_changes,
[
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[0].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[0].version()),
},
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[2].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[2].version()),
},
]
);
// Observe an earlier version of the buffer.
db.observe_buffer_version(
buffers[1].id,
observer_id,
1,
&[rpc::proto::VectorClockEntry {
replica_id: 0,
timestamp: 0,
}],
)
.await
.unwrap();
let buffer_changes = db
.transaction(|tx| {
let buffers = &buffers;
async move {
db.unseen_channel_buffer_changes(
observer_id,
&[
buffers[0].channel_id,
buffers[1].channel_id,
buffers[2].channel_id,
],
&*tx,
)
.await
}
})
.await
.unwrap();
assert_eq!(
buffer_changes,
[
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[0].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[0].version()),
},
rpc::proto::UnseenChannelBufferChange {
channel_id: buffers[2].channel_id.to_proto(),
epoch: 0,
version: serialize_version(&text_buffers[2].version()),
},
]
);
}
async fn update_buffer(
channel_id: ChannelId,
user_id: UserId,
db: &Database,
operations: Vec<text::Operation>,
) {
let operations = operations
.into_iter()
.map(|op| proto::serialize_operation(&language::Operation::Buffer(op)))
.collect::<Vec<_>>();
db.update_channel_buffer(channel_id, user_id, &operations)
.await
.unwrap();
}
fn assert_operations(
operations: &[buffer_operation::Model],
expected: &[(BufferId, i32, &text::Buffer)],
) {
let actual = operations
.iter()
.map(|op| buffer_operation::Model {
buffer_id: op.buffer_id,
epoch: op.epoch,
lamport_timestamp: op.lamport_timestamp,
replica_id: op.replica_id,
value: vec![],
})
.collect::<Vec<_>>();
let expected = expected
.iter()
.map(|(buffer_id, epoch, buffer)| buffer_operation::Model {
buffer_id: *buffer_id,
epoch: *epoch,
lamport_timestamp: buffer.lamport_clock.value as i32 - 1,
replica_id: buffer.replica_id() as i32,
value: vec![],
})
.collect::<Vec<_>>();
assert_eq!(actual, expected, "unexpected operations")
}

185
crates/collab/src/db/tests/message_tests.rs
View File

@@ -57,3 +57,188 @@ async fn test_channel_message_nonces(db: &Arc<Database>) {
assert_eq!(msg1_id, msg3_id);
assert_eq!(msg2_id, msg4_id);
}
test_both_dbs!(
test_channel_message_new_notification,
test_channel_message_new_notification_postgres,
test_channel_message_new_notification_sqlite
);
async fn test_channel_message_new_notification(db: &Arc<Database>) {
let user = db
.create_user(
"user_a@example.com",
false,
NewUserParams {
github_login: "user_a".into(),
github_user_id: 1,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let observer = db
.create_user(
"user_b@example.com",
false,
NewUserParams {
github_login: "user_b".into(),
github_user_id: 1,
invite_count: 0,
},
)
.await
.unwrap()
.user_id;
let channel_1 = db
.create_channel("channel", None, "room", user)
.await
.unwrap();
let channel_2 = db
.create_channel("channel-2", None, "room", user)
.await
.unwrap();
db.invite_channel_member(channel_1, observer, user, false)
.await
.unwrap();
db.respond_to_channel_invite(channel_1, observer, true)
.await
.unwrap();
db.invite_channel_member(channel_2, observer, user, false)
.await
.unwrap();
db.respond_to_channel_invite(channel_2, observer, true)
.await
.unwrap();
let owner_id = db.create_server("test").await.unwrap().0 as u32;
let user_connection_id = rpc::ConnectionId { owner_id, id: 0 };
db.join_channel_chat(channel_1, user_connection_id, user)
.await
.unwrap();
let _ = db
.create_channel_message(channel_1, user, "1_1", OffsetDateTime::now_utc(), 1)
.await
.unwrap();
let (second_message, _, _) = db
.create_channel_message(channel_1, user, "1_2", OffsetDateTime::now_utc(), 2)
.await
.unwrap();
let (third_message, _, _) = db
.create_channel_message(channel_1, user, "1_3", OffsetDateTime::now_utc(), 3)
.await
.unwrap();
db.join_channel_chat(channel_2, user_connection_id, user)
.await
.unwrap();
let (fourth_message, _, _) = db
.create_channel_message(channel_2, user, "2_1", OffsetDateTime::now_utc(), 4)
.await
.unwrap();
// Check that observer has new messages
let unseen_messages = db
.transaction(|tx| async move {
db.unseen_channel_messages(observer, &[channel_1, channel_2], &*tx)
.await
})
.await
.unwrap();
assert_eq!(
unseen_messages,
[
rpc::proto::UnseenChannelMessage {
channel_id: channel_1.to_proto(),
message_id: third_message.to_proto(),
},
rpc::proto::UnseenChannelMessage {
channel_id: channel_2.to_proto(),
message_id: fourth_message.to_proto(),
},
]
);
// Observe the second message
db.observe_channel_message(channel_1, observer, second_message)
.await
.unwrap();
// Make sure the observer still has a new message
let unseen_messages = db
.transaction(|tx| async move {
db.unseen_channel_messages(observer, &[channel_1, channel_2], &*tx)
.await
})
.await
.unwrap();
assert_eq!(
unseen_messages,
[
rpc::proto::UnseenChannelMessage {
channel_id: channel_1.to_proto(),
message_id: third_message.to_proto(),
},
rpc::proto::UnseenChannelMessage {
channel_id: channel_2.to_proto(),
message_id: fourth_message.to_proto(),
},
]
);
// Observe the third message,
db.observe_channel_message(channel_1, observer, third_message)
.await
.unwrap();
// Make sure the observer does not have a new method
let unseen_messages = db
.transaction(|tx| async move {
db.unseen_channel_messages(observer, &[channel_1, channel_2], &*tx)
.await
})
.await
.unwrap();
assert_eq!(
unseen_messages,
[rpc::proto::UnseenChannelMessage {
channel_id: channel_2.to_proto(),
message_id: fourth_message.to_proto(),
}]
);
// Observe the second message again, should not regress our observed state
db.observe_channel_message(channel_1, observer, second_message)
.await
.unwrap();
// Make sure the observer does not have a new message
let unseen_messages = db
.transaction(|tx| async move {
db.unseen_channel_messages(observer, &[channel_1, channel_2], &*tx)
.await
})
.await
.unwrap();
assert_eq!(
unseen_messages,
[rpc::proto::UnseenChannelMessage {
channel_id: channel_2.to_proto(),
message_id: fourth_message.to_proto(),
}]
);
}

92
crates/collab/src/rpc.rs
View File

@@ -3,8 +3,8 @@ mod connection_pool;
use crate::{
auth,
db::{
self, ChannelId, ChannelsForUser, Database, MessageId, ProjectId, RoomId, ServerId, User,
UserId,
self, BufferId, ChannelId, ChannelsForUser, Database, MessageId, ProjectId, RoomId,
ServerId, User, UserId,
},
executor::Executor,
AppState, Result,
@@ -274,7 +274,9 @@ impl Server {
.add_message_handler(unfollow)
.add_message_handler(update_followers)
.add_message_handler(update_diff_base)
.add_request_handler(get_private_user_info);
.add_request_handler(get_private_user_info)
.add_message_handler(acknowledge_channel_message)
.add_message_handler(acknowledge_buffer_version);
Arc::new(server)
}
@@ -2568,6 +2570,8 @@ async fn respond_to_channel_invite(
name: channel.name,
}),
);
update.unseen_channel_messages = result.channel_messages;
update.unseen_channel_buffer_changes = result.unseen_buffer_changes;
update.insert_edge = result.channels.edges;
update
.channel_participants
@@ -2691,7 +2695,7 @@ async fn update_channel_buffer(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let collaborators = db
let (collaborators, non_collaborators, epoch, version) = db
.update_channel_buffer(channel_id, session.user_id, &request.operations)
.await?;
@@ -2704,6 +2708,29 @@ async fn update_channel_buffer(
},
&session.peer,
);
let pool = &*session.connection_pool().await;
broadcast(
None,
non_collaborators
.iter()
.flat_map(|user_id| pool.user_connection_ids(*user_id)),
|peer_id| {
session.peer.send(
peer_id.into(),
proto::UpdateChannels {
unseen_channel_buffer_changes: vec![proto::UnseenChannelBufferChange {
channel_id: channel_id.to_proto(),
epoch: epoch as u64,
version: version.clone(),
}],
..Default::default()
},
)
},
);
Ok(())
}
@@ -2799,7 +2826,7 @@ async fn send_channel_message(
.ok_or_else(|| anyhow!("nonce can't be blank"))?;
let channel_id = ChannelId::from_proto(request.channel_id);
let (message_id, connection_ids) = session
let (message_id, connection_ids, non_participants) = session
.db()
.await
.create_channel_message(
@@ -2829,6 +2856,27 @@ async fn send_channel_message(
response.send(proto::SendChannelMessageResponse {
message: Some(message),
})?;
let pool = &*session.connection_pool().await;
broadcast(
None,
non_participants
.iter()
.flat_map(|user_id| pool.user_connection_ids(*user_id)),
|peer_id| {
session.peer.send(
peer_id.into(),
proto::UpdateChannels {
unseen_channel_messages: vec![proto::UnseenChannelMessage {
channel_id: channel_id.to_proto(),
message_id: message_id.to_proto(),
}],
..Default::default()
},
)
},
);
Ok(())
}
@@ -2851,6 +2899,38 @@ async fn remove_channel_message(
Ok(())
}
async fn acknowledge_channel_message(
request: proto::AckChannelMessage,
session: Session,
) -> Result<()> {
let channel_id = ChannelId::from_proto(request.channel_id);
let message_id = MessageId::from_proto(request.message_id);
session
.db()
.await
.observe_channel_message(channel_id, session.user_id, message_id)
.await?;
Ok(())
}
async fn acknowledge_buffer_version(
request: proto::AckBufferOperation,
session: Session,
) -> Result<()> {
let buffer_id = BufferId::from_proto(request.buffer_id);
session
.db()
.await
.observe_buffer_version(
buffer_id,
session.user_id,
request.epoch as i32,
&request.version,
)
.await?;
Ok(())
}
async fn join_channel_chat(
request: proto::JoinChannelChat,
response: Response<proto::JoinChannelChat>,
@@ -2986,6 +3066,8 @@ fn build_initial_channels_update(
});
}
update.unseen_channel_buffer_changes = channels.unseen_buffer_changes;
update.unseen_channel_messages = channels.channel_messages;
update.insert_edge = channels.channels.edges;
for (channel_id, participants) in channels.channel_participants {

142
crates/collab/src/tests/channel_buffer_tests.rs
View File

@@ -3,7 +3,7 @@ use crate::{
tests::TestServer,
};
use call::ActiveCall;
use channel::Channel;
use channel::{Channel, ACKNOWLEDGE_DEBOUNCE_INTERVAL};
use client::ParticipantIndex;
use client::{Collaborator, UserId};
use collab_ui::channel_view::ChannelView;
@@ -410,10 +410,7 @@ async fn test_channel_buffer_disconnect(
channel_buffer_a.update(cx_a, |buffer, _| {
assert_eq!(
buffer.channel().as_ref(),
&Channel {
id: channel_id,
name: "the-channel".to_string()
}
&channel(channel_id, "the-channel")
);
assert!(!buffer.is_connected());
});
@@ -438,15 +435,21 @@ async fn test_channel_buffer_disconnect(
channel_buffer_b.update(cx_b, |buffer, _| {
assert_eq!(
buffer.channel().as_ref(),
&Channel {
id: channel_id,
name: "the-channel".to_string()
}
&channel(channel_id, "the-channel")
);
assert!(!buffer.is_connected());
});
}
fn channel(id: u64, name: &'static str) -> Channel {
Channel {
id,
name: name.to_string(),
unseen_note_version: None,
unseen_message_id: None,
}
}
#[gpui::test]
async fn test_rejoin_channel_buffer(
deterministic: Arc<Deterministic>,
@@ -627,6 +630,7 @@ async fn test_following_to_channel_notes_without_a_shared_project(
let mut server = TestServer::start(&deterministic).await;
let client_a = server.create_client(cx_a, "user_a").await;
let client_b = server.create_client(cx_b, "user_b").await;
let client_c = server.create_client(cx_c, "user_c").await;
cx_a.update(editor::init);
@@ -755,6 +759,126 @@ async fn test_following_to_channel_notes_without_a_shared_project(
});
}
#[gpui::test]
async fn test_channel_buffer_changes(
deterministic: Arc<Deterministic>,
cx_a: &mut TestAppContext,
cx_b: &mut TestAppContext,
) {
deterministic.forbid_parking();
let mut server = TestServer::start(&deterministic).await;
let client_a = server.create_client(cx_a, "user_a").await;
let client_b = server.create_client(cx_b, "user_b").await;
let channel_id = server
.make_channel(
"the-channel",
None,
(&client_a, cx_a),
&mut [(&client_b, cx_b)],
)
.await;
let channel_buffer_a = client_a
.channel_store()
.update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
.await
.unwrap();
// Client A makes an edit, and client B should see that the note has changed.
channel_buffer_a.update(cx_a, |buffer, cx| {
buffer.buffer().update(cx, |buffer, cx| {
buffer.edit([(0..0, "1")], None, cx);
})
});
deterministic.run_until_parked();
let has_buffer_changed = cx_b.read(|cx| {
client_b
.channel_store()
.read(cx)
.has_channel_buffer_changed(channel_id)
.unwrap()
});
assert!(has_buffer_changed);
// Opening the buffer should clear the changed flag.
let project_b = client_b.build_empty_local_project(cx_b);
let workspace_b = client_b.build_workspace(&project_b, cx_b).root(cx_b);
let channel_view_b = cx_b
.update(|cx| ChannelView::open(channel_id, workspace_b.clone(), cx))
.await
.unwrap();
deterministic.run_until_parked();
let has_buffer_changed = cx_b.read(|cx| {
client_b
.channel_store()
.read(cx)
.has_channel_buffer_changed(channel_id)
.unwrap()
});
assert!(!has_buffer_changed);
// Editing the channel while the buffer is open should not show that the buffer has changed.
channel_buffer_a.update(cx_a, |buffer, cx| {
buffer.buffer().update(cx, |buffer, cx| {
buffer.edit([(0..0, "2")], None, cx);
})
});
deterministic.run_until_parked();
let has_buffer_changed = cx_b.read(|cx| {
client_b
.channel_store()
.read(cx)
.has_channel_buffer_changed(channel_id)
.unwrap()
});
assert!(!has_buffer_changed);
deterministic.advance_clock(ACKNOWLEDGE_DEBOUNCE_INTERVAL);
// Test that the server is tracking things correctly, and we retain our 'not changed'
// state across a disconnect
server.simulate_long_connection_interruption(client_b.peer_id().unwrap(), &deterministic);
let has_buffer_changed = cx_b.read(|cx| {
client_b
.channel_store()
.read(cx)
.has_channel_buffer_changed(channel_id)
.unwrap()
});
assert!(!has_buffer_changed);
// Closing the buffer should re-enable change tracking
cx_b.update(|cx| {
workspace_b.update(cx, |workspace, cx| {
workspace.close_all_items_and_panes(&Default::default(), cx)
});
drop(channel_view_b)
});
deterministic.run_until_parked();
channel_buffer_a.update(cx_a, |buffer, cx| {
buffer.buffer().update(cx, |buffer, cx| {
buffer.edit([(0..0, "3")], None, cx);
})
});
deterministic.run_until_parked();
let has_buffer_changed = cx_b.read(|cx| {
client_b
.channel_store()
.read(cx)
.has_channel_buffer_changed(channel_id)
.unwrap()
});
assert!(has_buffer_changed);
}
#[track_caller]
fn assert_collaborators(collaborators: &HashMap<PeerId, Collaborator>, ids: &[Option<UserId>]) {
let mut user_ids = collaborators

137
crates/collab/src/tests/channel_message_tests.rs
View File

@@ -1,7 +1,9 @@
use crate::{rpc::RECONNECT_TIMEOUT, tests::TestServer};
use channel::{ChannelChat, ChannelMessageId};
use gpui::{executor::Deterministic, ModelHandle, TestAppContext};
use collab_ui::chat_panel::ChatPanel;
use gpui::{executor::Deterministic, BorrowAppContext, ModelHandle, TestAppContext};
use std::sync::Arc;
use workspace::dock::Panel;
#[gpui::test]
async fn test_basic_channel_messages(
@@ -223,3 +225,136 @@ fn assert_messages(chat: &ModelHandle<ChannelChat>, messages: &[&str], cx: &mut
messages
);
}
#[gpui::test]
async fn test_channel_message_changes(
deterministic: Arc<Deterministic>,
cx_a: &mut TestAppContext,
cx_b: &mut TestAppContext,
) {
deterministic.forbid_parking();
let mut server = TestServer::start(&deterministic).await;
let client_a = server.create_client(cx_a, "user_a").await;
let client_b = server.create_client(cx_b, "user_b").await;
let channel_id = server
.make_channel(
"the-channel",
None,
(&client_a, cx_a),
&mut [(&client_b, cx_b)],
)
.await;
// Client A sends a message, client B should see that there is a new message.
let channel_chat_a = client_a
.channel_store()
.update(cx_a, |store, cx| store.open_channel_chat(channel_id, cx))
.await
.unwrap();
channel_chat_a
.update(cx_a, |c, cx| c.send_message("one".into(), cx).unwrap())
.await
.unwrap();
deterministic.run_until_parked();
let b_has_messages = cx_b.read_with(|cx| {
client_b
.channel_store()
.read(cx)
.has_new_messages(channel_id)
.unwrap()
});
assert!(b_has_messages);
// Opening the chat should clear the changed flag.
cx_b.update(|cx| {
collab_ui::init(&client_b.app_state, cx);
});
let project_b = client_b.build_empty_local_project(cx_b);
let workspace_b = client_b.build_workspace(&project_b, cx_b).root(cx_b);
let chat_panel_b = workspace_b.update(cx_b, |workspace, cx| ChatPanel::new(workspace, cx));
chat_panel_b
.update(cx_b, |chat_panel, cx| {
chat_panel.set_active(true, cx);
chat_panel.select_channel(channel_id, cx)
})
.await
.unwrap();
deterministic.run_until_parked();
let b_has_messages = cx_b.read_with(|cx| {
client_b
.channel_store()
.read(cx)
.has_new_messages(channel_id)
.unwrap()
});
assert!(!b_has_messages);
// Sending a message while the chat is open should not change the flag.
channel_chat_a
.update(cx_a, |c, cx| c.send_message("two".into(), cx).unwrap())
.await
.unwrap();
deterministic.run_until_parked();
let b_has_messages = cx_b.read_with(|cx| {
client_b
.channel_store()
.read(cx)
.has_new_messages(channel_id)
.unwrap()
});
assert!(!b_has_messages);
// Sending a message while the chat is closed should change the flag.
chat_panel_b.update(cx_b, |chat_panel, cx| {
chat_panel.set_active(false, cx);
});
// Sending a message while the chat is open should not change the flag.
channel_chat_a
.update(cx_a, |c, cx| c.send_message("three".into(), cx).unwrap())
.await
.unwrap();
deterministic.run_until_parked();
let b_has_messages = cx_b.read_with(|cx| {
client_b
.channel_store()
.read(cx)
.has_new_messages(channel_id)
.unwrap()
});
assert!(b_has_messages);
// Closing the chat should re-enable change tracking
cx_b.update(|_| drop(chat_panel_b));
channel_chat_a
.update(cx_a, |c, cx| c.send_message("four".into(), cx).unwrap())
.await
.unwrap();
deterministic.run_until_parked();
let b_has_messages = cx_b.read_with(|cx| {
client_b
.channel_store()
.read(cx)
.has_new_messages(channel_id)
.unwrap()
});
assert!(b_has_messages);
}

20
crates/collab/src/tests/test_server.rs
View File

@@ -1,7 +1,7 @@
use crate::{
db::{tests::TestDb, NewUserParams, UserId},
executor::Executor,
rpc::{Server, CLEANUP_TIMEOUT},
rpc::{Server, CLEANUP_TIMEOUT, RECONNECT_TIMEOUT},
AppState,
};
use anyhow::anyhow;
@@ -17,6 +17,7 @@ use gpui::{executor::Deterministic, ModelHandle, Task, TestAppContext, WindowHan
use language::LanguageRegistry;
use parking_lot::Mutex;
use project::{Project, WorktreeId};
use rpc::RECEIVE_TIMEOUT;
use settings::SettingsStore;
use std::{
cell::{Ref, RefCell, RefMut},
@@ -151,12 +152,12 @@ impl TestServer {
Arc::get_mut(&mut client)
.unwrap()
.set_id(user_id.0 as usize)
.set_id(user_id.to_proto())
.override_authenticate(move |cx| {
cx.spawn(|_| async move {
let access_token = "the-token".to_string();
Ok(Credentials {
user_id: user_id.0 as u64,
user_id: user_id.to_proto(),
access_token,
})
})
@@ -255,6 +256,19 @@ impl TestServer {
.store(true, SeqCst);
}
pub fn simulate_long_connection_interruption(
&self,
peer_id: PeerId,
deterministic: &Arc<Deterministic>,
) {
self.forbid_connections();
self.disconnect_client(peer_id);
deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
self.allow_connections();
deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
deterministic.run_until_parked();
}
pub fn forbid_connections(&self) {
self.forbid_connections.store(true, SeqCst);
}

52
crates/collab_ui/src/channel_view.rs
View File

@@ -1,6 +1,6 @@
use anyhow::{anyhow, Result};
use call::report_call_event_for_channel;
use channel::{Channel, ChannelBuffer, ChannelBufferEvent, ChannelId};
use channel::{Channel, ChannelBuffer, ChannelBufferEvent, ChannelId, ChannelStore};
use client::{
proto::{self, PeerId},
Collaborator, ParticipantIndex,
@@ -36,6 +36,7 @@ pub fn init(cx: &mut AppContext) {
pub struct ChannelView {
pub editor: ViewHandle<Editor>,
project: ModelHandle<Project>,
channel_store: ModelHandle<ChannelStore>,
channel_buffer: ModelHandle<ChannelBuffer>,
remote_id: Option<ViewId>,
_editor_event_subscription: Subscription,
@@ -94,7 +95,13 @@ impl ChannelView {
pane.update(&mut cx, |pane, cx| {
pane.items_of_type::<Self>()
.find(|channel_view| channel_view.read(cx).channel_buffer == channel_buffer)
.unwrap_or_else(|| cx.add_view(|cx| Self::new(project, channel_buffer, cx)))
.unwrap_or_else(|| {
cx.add_view(|cx| {
let mut this = Self::new(project, channel_store, channel_buffer, cx);
this.acknowledge_buffer_version(cx);
this
})
})
})
.ok_or_else(|| anyhow!("pane was dropped"))
})
@@ -102,6 +109,7 @@ impl ChannelView {
pub fn new(
project: ModelHandle<Project>,
channel_store: ModelHandle<ChannelStore>,
channel_buffer: ModelHandle<ChannelBuffer>,
cx: &mut ViewContext<Self>,
) -> Self {
@@ -121,6 +129,7 @@ impl ChannelView {
Self {
editor,
project,
channel_store,
channel_buffer,
remote_id: None,
_editor_event_subscription,
@@ -137,13 +146,44 @@ impl ChannelView {
event: &ChannelBufferEvent,
cx: &mut ViewContext<Self>,
) {
if let ChannelBufferEvent::Disconnected = event {
self.editor.update(cx, |editor, cx| {
match event {
ChannelBufferEvent::Disconnected => self.editor.update(cx, |editor, cx| {
editor.set_read_only(true);
cx.notify();
})
}),
ChannelBufferEvent::BufferEdited => {
if cx.is_self_focused() || self.editor.is_focused(cx) {
self.acknowledge_buffer_version(cx);
} else {
self.channel_store.update(cx, |store, cx| {
let channel_buffer = self.channel_buffer.read(cx);
store.notes_changed(
channel_buffer.channel().id,
channel_buffer.epoch(),
&channel_buffer.buffer().read(cx).version(),
cx,
)
});
}
}
_ => {}
}
}
fn acknowledge_buffer_version(&mut self, cx: &mut ViewContext<'_, '_, ChannelView>) {
self.channel_store.update(cx, |store, cx| {
let channel_buffer = self.channel_buffer.read(cx);
store.acknowledge_notes_version(
channel_buffer.channel().id,
channel_buffer.epoch(),
&channel_buffer.buffer().read(cx).version(),
cx,
)
});
self.channel_buffer.update(cx, |buffer, cx| {
buffer.acknowledge_buffer_version(cx);
});
}
}
impl Entity for ChannelView {
@@ -161,6 +201,7 @@ impl View for ChannelView {
fn focus_in(&mut self, _: AnyViewHandle, cx: &mut ViewContext<Self>) {
if cx.is_self_focused() {
self.acknowledge_buffer_version(cx);
cx.focus(self.editor.as_any())
}
}
@@ -200,6 +241,7 @@ impl Item for ChannelView {
fn clone_on_split(&self, _: WorkspaceId, cx: &mut ViewContext<Self>) -> Option<Self> {
Some(Self::new(
self.project.clone(),
self.channel_store.clone(),
self.channel_buffer.clone(),
cx,
))

46
crates/collab_ui/src/chat_panel.rs
View File

@@ -42,6 +42,7 @@ pub struct ChatPanel {
local_timezone: UtcOffset,
fs: Arc<dyn Fs>,
width: Option<f32>,
active: bool,
pending_serialization: Task<Option<()>>,
subscriptions: Vec<gpui::Subscription>,
workspace: WeakViewHandle<Workspace>,
@@ -138,6 +139,7 @@ impl ChatPanel {
has_focus: false,
subscriptions: Vec::new(),
workspace: workspace_handle,
active: false,
width: None,
};
@@ -154,9 +156,9 @@ impl ChatPanel {
}),
);
this.init_active_channel(cx);
this.update_channel_count(cx);
cx.observe(&this.channel_store, |this, _, cx| {
this.init_active_channel(cx);
this.update_channel_count(cx)
})
.detach();
@@ -179,6 +181,10 @@ impl ChatPanel {
})
}
pub fn active_chat(&self) -> Option<ModelHandle<ChannelChat>> {
self.active_chat.as_ref().map(|(chat, _)| chat.clone())
}
pub fn load(
workspace: WeakViewHandle<Workspace>,
cx: AsyncAppContext,
@@ -225,10 +231,8 @@ impl ChatPanel {
);
}
fn init_active_channel(&mut self, cx: &mut ViewContext<Self>) {
fn update_channel_count(&mut self, cx: &mut ViewContext<Self>) {
let channel_count = self.channel_store.read(cx).channel_count();
self.message_list.reset(0);
self.active_chat = None;
self.channel_select.update(cx, |select, cx| {
select.set_item_count(channel_count, cx);
});
@@ -247,6 +251,7 @@ impl ChatPanel {
}
let subscription = cx.subscribe(&chat, Self::channel_did_change);
self.active_chat = Some((chat, subscription));
self.acknowledge_last_message(cx);
self.channel_select.update(cx, |select, cx| {
if let Some(ix) = self.channel_store.read(cx).index_of_channel(id) {
select.set_selected_index(ix, cx);
@@ -268,11 +273,34 @@ impl ChatPanel {
new_count,
} => {
self.message_list.splice(old_range.clone(), *new_count);
if self.active {
self.acknowledge_last_message(cx);
}
}
ChannelChatEvent::NewMessage {
channel_id,
message_id,
} => {
if !self.active {
self.channel_store.update(cx, |store, cx| {
store.new_message(*channel_id, *message_id, cx)
})
}
}
}
cx.notify();
}
fn acknowledge_last_message(&mut self, cx: &mut ViewContext<'_, '_, ChatPanel>) {
if self.active {
if let Some((chat, _)) = &self.active_chat {
chat.update(cx, |chat, cx| {
chat.acknowledge_last_message(cx);
});
}
}
}
fn render_channel(&self, cx: &mut ViewContext<Self>) -> AnyElement<Self> {
let theme = theme::current(cx);
Flex::column()
@@ -627,8 +655,12 @@ impl Panel for ChatPanel {
}
fn set_active(&mut self, active: bool, cx: &mut ViewContext<Self>) {
if active && !is_chat_feature_enabled(cx) {
cx.emit(Event::Dismissed);
self.active = active;
if active {
self.acknowledge_last_message(cx);
if !is_chat_feature_enabled(cx) {
cx.emit(Event::Dismissed);
}
}
}

201
crates/collab_ui/src/collab_panel.rs
View File

@@ -592,6 +592,7 @@ impl CollabPanel {
*channel_id,
&theme.collab_panel,
is_selected,
ix,
cx,
),
ListEntry::ChannelInvite(channel) => Self::render_channel_invite(
@@ -1917,7 +1918,9 @@ impl CollabPanel {
const FACEPILE_LIMIT: usize = 3;
enum ChannelCall {}
enum IconTooltip {}
enum ChannelNote {}
enum NotesTooltip {}
enum ChatTooltip {}
enum ChannelTooltip {}
let mut is_dragged_over = false;
@@ -1960,71 +1963,113 @@ impl CollabPanel {
.aligned()
.left(),
)
.with_child(
Label::new(channel.name.clone(), collab_theme.channel_name.text.clone())
.contained()
.with_style(collab_theme.channel_name.container)
.aligned()
.left()
.with_tooltip::<ChannelTooltip>(
channel_id as usize,
if is_active {
"Open channel notes"
} else {
"Join channel"
},
None,
theme.tooltip.clone(),
cx,
)
.flex(1., true),
)
.with_child(
MouseEventHandler::new::<ChannelCall, _>(ix, cx, move |_, cx| {
let participants =
self.channel_store.read(cx).channel_participants(channel_id);
if !participants.is_empty() {
let extra_count = participants.len().saturating_sub(FACEPILE_LIMIT);
FacePile::new(collab_theme.face_overlap)
.with_children(
participants
.iter()
.filter_map(|user| {
Some(
Image::from_data(user.avatar.clone()?)
.with_style(collab_theme.channel_avatar),
)
})
.take(FACEPILE_LIMIT),
)
.with_children((extra_count > 0).then(|| {
Label::new(
format!("+{}", extra_count),
collab_theme.extra_participant_label.text.clone(),
)
.contained()
.with_style(collab_theme.extra_participant_label.container)
}))
.with_tooltip::<IconTooltip>(
channel_id as usize,
.with_child({
let style = collab_theme.channel_name.inactive_state();
Flex::row()
.with_child(
Label::new(channel.name.clone(), style.text.clone())
.contained()
.with_style(style.container)
.aligned()
.left()
.with_tooltip::<ChannelTooltip>(
ix,
if is_active {
"Open Channel Notes"
"Open channel notes"
} else {
"Join channel"
},
None,
theme.tooltip.clone(),
cx,
)
),
)
.with_children({
let participants =
self.channel_store.read(cx).channel_participants(channel_id);
if !participants.is_empty() {
let extra_count = participants.len().saturating_sub(FACEPILE_LIMIT);
let result = FacePile::new(collab_theme.face_overlap)
.with_children(
participants
.iter()
.filter_map(|user| {
Some(
Image::from_data(user.avatar.clone()?)
.with_style(collab_theme.channel_avatar),
)
})
.take(FACEPILE_LIMIT),
)
.with_children((extra_count > 0).then(|| {
Label::new(
format!("+{}", extra_count),
collab_theme.extra_participant_label.text.clone(),
)
.contained()
.with_style(collab_theme.extra_participant_label.container)
}));
Some(result)
} else {
None
}
})
.with_spacing(8.)
.align_children_center()
.flex(1., true)
})
.with_child(
MouseEventHandler::new::<ChannelNote, _>(ix, cx, move |_, _| {
if channel.unseen_message_id.is_some() {
Svg::new("icons/conversations.svg")
.with_color(collab_theme.channel_note_active_color)
.constrained()
.with_width(collab_theme.channel_hash.width)
.into_any()
} else if row_hovered {
Svg::new("icons/file.svg")
Svg::new("icons/conversations.svg")
.with_color(collab_theme.channel_hash.color)
.constrained()
.with_width(collab_theme.channel_hash.width)
.with_tooltip::<IconTooltip>(
channel_id as usize,
.into_any()
} else {
Empty::new()
.constrained()
.with_width(collab_theme.channel_hash.width)
.into_any()
}
})
.on_click(MouseButton::Left, move |_, this, cx| {
this.join_channel_chat(&JoinChannelChat { channel_id }, cx);
})
.with_tooltip::<ChatTooltip>(
ix,
"Open channel chat",
None,
theme.tooltip.clone(),
cx,
)
.contained()
.with_margin_right(4.),
)
.with_child(
MouseEventHandler::new::<ChannelCall, _>(ix, cx, move |_, cx| {
if row_hovered || channel.unseen_note_version.is_some() {
Svg::new("icons/file.svg")
.with_color(if channel.unseen_note_version.is_some() {
collab_theme.channel_note_active_color
} else {
collab_theme.channel_hash.color
})
.constrained()
.with_width(collab_theme.channel_hash.width)
.contained()
.with_margin_right(collab_theme.channel_hash.container.margin.left)
.with_tooltip::<NotesTooltip>(
ix as usize,
"Open channel notes",
None,
theme.tooltip.clone(),
@@ -2032,7 +2077,12 @@ impl CollabPanel {
)
.into_any()
} else {
Empty::new().into_any()
Empty::new()
.constrained()
.with_width(collab_theme.channel_hash.width)
.contained()
.with_margin_right(collab_theme.channel_hash.container.margin.left)
.into_any()
}
})
.on_click(MouseButton::Left, move |_, this, cx| {
@@ -2189,6 +2239,7 @@ impl CollabPanel {
channel_id: ChannelId,
theme: &theme::CollabPanel,
is_selected: bool,
ix: usize,
cx: &mut ViewContext<Self>,
) -> AnyElement<Self> {
enum ChannelNotes {}
@@ -2198,7 +2249,7 @@ impl CollabPanel {
.or(theme.contact_avatar.height)
.unwrap_or(0.);
MouseEventHandler::new::<ChannelNotes, _>(channel_id as usize, cx, |state, cx| {
MouseEventHandler::new::<ChannelNotes, _>(ix as usize, cx, |state, cx| {
let tree_branch = *theme.tree_branch.in_state(is_selected).style_for(state);
let row = theme.project_row.in_state(is_selected).style_for(state);
@@ -2709,11 +2760,9 @@ impl CollabPanel {
.read(cx)
.channel_id()?;
dbg!(call_channel, channel.id);
Some(call_channel == channel.id)
})
.unwrap_or(false);
dbg!(is_active);
if is_active {
self.open_channel_notes(
&OpenChannelNotes {
@@ -3131,10 +3180,17 @@ impl CollabPanel {
let window = cx.window();
let active_call = ActiveCall::global(cx);
cx.spawn(|_, mut cx| async move {
if active_call.read_with(&mut cx, |active_call, _| active_call.room().is_some()) {
if active_call.read_with(&mut cx, |active_call, cx| {
if let Some(room) = active_call.room() {
let room = room.read(cx);
room.is_sharing_project() && room.remote_participants().len() > 0
} else {
false
}
}) {
let answer = window.prompt(
PromptLevel::Warning,
"Do you want to leave the current call?",
"Leaving this call will unshare your current project.\nDo you want to switch channels?",
&["Yes, Join Channel", "Cancel"],
&mut cx,
);
@@ -3150,27 +3206,16 @@ impl CollabPanel {
.update(&mut cx, |call, cx| call.join_channel(channel_id, cx))
.await?;
let tasks = room.update(&mut cx, |room, cx| {
let Some(workspace) = workspace.upgrade(cx) else {
return vec![];
};
let projects = room.projects_to_join();
if projects.is_empty() {
ChannelView::open(channel_id, workspace, cx).detach();
return vec![];
}
room.projects_to_join()
.into_iter()
.map(|(project_id, user_id)| {
let app_state = workspace.read(cx).app_state().clone();
workspace::join_remote_project(project_id, user_id, app_state, cx)
})
.collect()
let task = room.update(&mut cx, |room, cx| {
let workspace = workspace.upgrade(cx)?;
let (project, host) = room.most_active_project()?;
let app_state = workspace.read(cx).app_state().clone();
Some(workspace::join_remote_project(project, host, app_state, cx))
});
for task in tasks {
if let Some(task) = task {
task.await?;
}
anyhow::Ok(())
})
.detach_and_log_err(cx);

2
crates/gpui/Cargo.toml
View File

@@ -53,7 +53,7 @@ thiserror.workspace = true
time.workspace = true
tiny-skia = "0.5"
usvg = { version = "0.14", features = [] }
uuid = { version = "1.1.2", features = ["v4"] }
uuid.workspace = true
waker-fn = "1.1.0"
[build-dependencies]

2
crates/gpui/src/app.rs
View File

@@ -1252,7 +1252,7 @@ impl AppContext {
result
})
} else {
panic!("circular model update");
panic!("circular model update for {}", std::any::type_name::<T>());
}
}

14
crates/gpui/src/views/select.rs
View File

@@ -53,8 +53,10 @@ impl Select {
}
pub fn set_item_count(&mut self, count: usize, cx: &mut ViewContext<Self>) {
self.item_count = count;
cx.notify();
if count != self.item_count {
self.item_count = count;
cx.notify();
}
}
fn toggle(&mut self, cx: &mut ViewContext<Self>) {
@@ -63,9 +65,11 @@ impl Select {
}
pub fn set_selected_index(&mut self, ix: usize, cx: &mut ViewContext<Self>) {
self.selected_item_ix = ix;
self.is_open = false;
cx.notify();
if ix != self.selected_item_ix || self.is_open {
self.selected_item_ix = ix;
self.is_open = false;
cx.notify();
}
}
pub fn selected_index(&self) -> usize {

28
crates/rpc/proto/zed.proto
View File

@@ -155,6 +155,7 @@ message Envelope {
UpdateChannelBufferCollaborators update_channel_buffer_collaborators = 128;
RejoinChannelBuffers rejoin_channel_buffers = 129;
RejoinChannelBuffersResponse rejoin_channel_buffers_response = 130;
AckBufferOperation ack_buffer_operation = 143;
JoinChannelChat join_channel_chat = 131;
JoinChannelChatResponse join_channel_chat_response = 132;
@@ -165,10 +166,11 @@ message Envelope {
GetChannelMessages get_channel_messages = 137;
GetChannelMessagesResponse get_channel_messages_response = 138;
RemoveChannelMessage remove_channel_message = 139;
AckChannelMessage ack_channel_message = 144;
LinkChannel link_channel = 140;
UnlinkChannel unlink_channel = 141;
MoveChannel move_channel = 142;
MoveChannel move_channel = 142; // current max: 144
}
}
@@ -955,6 +957,19 @@ message UpdateChannels {
repeated uint64 remove_channel_invitations = 6;
repeated ChannelParticipants channel_participants = 7;
repeated ChannelPermission channel_permissions = 8;
repeated UnseenChannelMessage unseen_channel_messages = 9;
repeated UnseenChannelBufferChange unseen_channel_buffer_changes = 10;
}
message UnseenChannelMessage {
uint64 channel_id = 1;
uint64 message_id = 2;
}
message UnseenChannelBufferChange {
uint64 channel_id = 1;
uint64 epoch = 2;
repeated VectorClockEntry version = 3;
}
message ChannelEdge {
@@ -1060,6 +1075,11 @@ message RemoveChannelMessage {
uint64 message_id = 2;
}
message AckChannelMessage {
uint64 channel_id = 1;
uint64 message_id = 2;
}
message SendChannelMessageResponse {
ChannelMessage message = 1;
}
@@ -1115,6 +1135,12 @@ message RejoinChannelBuffersResponse {
repeated RejoinedChannelBuffer buffers = 1;
}
message AckBufferOperation {
uint64 buffer_id = 1;
uint64 epoch = 2;
repeated VectorClockEntry version = 3;
}
message JoinChannelBufferResponse {
uint64 buffer_id = 1;
uint32 replica_id = 2;

4
crates/rpc/src/proto.rs
View File

@@ -271,6 +271,8 @@ messages!(
(LeaveChannelBuffer, Background),
(UpdateChannelBuffer, Foreground),
(UpdateChannelBufferCollaborators, Foreground),
(AckBufferOperation, Background),
(AckChannelMessage, Background),
);
request_messages!(
@@ -406,7 +408,7 @@ entity_messages!(
ChannelMessageSent,
UpdateChannelBuffer,
RemoveChannelMessage,
UpdateChannelBufferCollaborators
UpdateChannelBufferCollaborators,
);
const KIB: usize = 1024;

2
crates/semantic_index/Cargo.toml
View File

@@ -26,7 +26,7 @@ postage.workspace = true
futures.workspace = true
ordered-float.workspace = true
smol.workspace = true
rusqlite = { version = "0.27.0", features = ["blob", "array", "modern_sqlite"] }
rusqlite.workspace = true
log.workspace = true
tree-sitter.workspace = true
lazy_static.workspace = true

4
crates/sqlez/Cargo.toml
View File

@@ -7,10 +7,10 @@ publish = false
[dependencies]
anyhow.workspace = true
indoc.workspace = true
libsqlite3-sys = { version = "0.24", features = ["bundled"] }
libsqlite3-sys = { version = "0.26", features = ["bundled"] }
smol.workspace = true
thread_local = "1.1.4"
lazy_static.workspace = true
parking_lot.workspace = true
futures.workspace = true
uuid = { version = "1.1.2", features = ["v4"] }
uuid.workspace = true

3
crates/theme/src/theme.rs
View File

@@ -239,6 +239,7 @@ pub struct CollabPanel {
pub log_in_button: Interactive<ContainedText>,
pub channel_editor: ContainerStyle,
pub channel_hash: Icon,
pub channel_note_active_color: Color,
pub tabbed_modal: TabbedModal,
pub contact_finder: ContactFinder,
pub channel_modal: ChannelModal,
@@ -252,7 +253,7 @@ pub struct CollabPanel {
pub leave_call: Interactive<ContainedText>,
pub contact_row: Toggleable<Interactive<ContainerStyle>>,
pub channel_row: Toggleable<Interactive<ContainerStyle>>,
pub channel_name: ContainedText,
pub channel_name: Toggleable<ContainedText>,
pub row_height: f32,
pub project_row: Toggleable<Interactive<ProjectRow>>,
pub tree_branch: Toggleable<Interactive<TreeBranch>>,

2
crates/workspace/Cargo.toml
View File

@@ -51,7 +51,7 @@ serde.workspace = true
serde_derive.workspace = true
serde_json.workspace = true
smallvec.workspace = true
uuid = { version = "1.1.2", features = ["v4"] }
uuid.workspace = true
[dev-dependencies]
call = { path = "../call", features = ["test-support"] }

2
crates/zed/Cargo.toml
View File

@@ -138,7 +138,7 @@ tree-sitter-nu.workspace = true
url = "2.2"
urlencoding = "2.1.2"
uuid = { version = "1.1.2", features = ["v4"] }
uuid.workspace = true
[dev-dependencies]
call = { path = "../call", features = ["test-support"] }

15
styles/src/style_tree/collab_panel.ts
View File

@@ -194,6 +194,7 @@ export default function contacts_panel(): any {
},
user_query_editor: filter_input,
channel_hash: icon_style,
channel_note_active_color: foreground(layer, "active"),
user_query_editor_height: 33,
add_contact_button: header_icon_button,
add_channel_button: header_icon_button,
@@ -267,10 +268,18 @@ export default function contacts_panel(): any {
}),
channel_row: item_row,
channel_name: {
...text(layer, "sans", { size: "sm" }),
margin: {
left: CHANNEL_SPACING,
active: {
...text(layer, "sans", { size: "sm", weight: "bold" }),
margin: {
left: CHANNEL_SPACING,
},
},
inactive: {
...text(layer, "sans", { size: "sm" }),
margin: {
left: CHANNEL_SPACING,
},
}
},
list_empty_label_container: {
margin: {