Move ChannelList, UserStore into client crate

Co-Authored-By: Nathan Sobo <nathan@zed.dev>

crates/client/Cargo.toml

@@ -10,10 +10,13 @@ test-support = []
 gpui = { path = "../gpui" }
 util = { path = "../util" }
 rpc = { path = "../rpc" }
+sum_tree = { path = "../sum_tree" }
 
 anyhow = "1.0.38"
 async-recursion = "0.3"
 async-tungstenite = { version = "0.14", features = ["async-tls"] }
+futures = "0.3"
+image = "0.23"
 lazy_static = "1.4.0"
 log = "0.4"
 parking_lot = "0.11.1"
@@ -22,4 +25,5 @@ rand = "0.8.3"
 smol = "1.2.5"
 surf = "2.2"
 thiserror = "1.0.29"
+time = "0.3"
 tiny_http = "0.8"

crates/client/src/channel.rs

@@ -0,0 +1,819 @@
+use super::{
+    proto,
+    user::{User, UserStore},
+    Client, Status, Subscription, TypedEnvelope,
+};
+use anyhow::{anyhow, Context, Result};
+use gpui::{
+    AsyncAppContext, Entity, ModelContext, ModelHandle, MutableAppContext, Task, WeakModelHandle,
+};
+use postage::prelude::Stream;
+use rand::prelude::*;
+use std::{
+    collections::{HashMap, HashSet},
+    mem,
+    ops::Range,
+    sync::Arc,
+};
+use sum_tree::{Bias, SumTree};
+use time::OffsetDateTime;
+use util::{post_inc, TryFutureExt};
+
+pub struct ChannelList {
+    available_channels: Option<Vec<ChannelDetails>>,
+    channels: HashMap<u64, WeakModelHandle<Channel>>,
+    client: Arc<Client>,
+    user_store: ModelHandle<UserStore>,
+    _task: Task<Option<()>>,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct ChannelDetails {
+    pub id: u64,
+    pub name: String,
+}
+
+pub struct Channel {
+    details: ChannelDetails,
+    messages: SumTree<ChannelMessage>,
+    loaded_all_messages: bool,
+    next_pending_message_id: usize,
+    user_store: ModelHandle<UserStore>,
+    rpc: Arc<Client>,
+    rng: StdRng,
+    _subscription: Subscription,
+}
+
+#[derive(Clone, Debug)]
+pub struct ChannelMessage {
+    pub id: ChannelMessageId,
+    pub body: String,
+    pub timestamp: OffsetDateTime,
+    pub sender: Arc<User>,
+    pub nonce: u128,
+}
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub enum ChannelMessageId {
+    Saved(u64),
+    Pending(usize),
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct ChannelMessageSummary {
+    max_id: ChannelMessageId,
+    count: usize,
+}
+
+#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+struct Count(usize);
+
+pub enum ChannelListEvent {}
+
+#[derive(Clone, Debug, PartialEq)]
+pub enum ChannelEvent {
+    MessagesUpdated {
+        old_range: Range<usize>,
+        new_count: usize,
+    },
+}
+
+impl Entity for ChannelList {
+    type Event = ChannelListEvent;
+}
+
+impl ChannelList {
+    pub fn new(
+        user_store: ModelHandle<UserStore>,
+        rpc: Arc<Client>,
+        cx: &mut ModelContext<Self>,
+    ) -> Self {
+        let _task = cx.spawn_weak(|this, mut cx| {
+            let rpc = rpc.clone();
+            async move {
+                let mut status = rpc.status();
+                while let Some((status, this)) = status.recv().await.zip(this.upgrade(&cx)) {
+                    match status {
+                        Status::Connected { .. } => {
+                            let response = rpc
+                                .request(proto::GetChannels {})
+                                .await
+                                .context("failed to fetch available channels")?;
+                            this.update(&mut cx, |this, cx| {
+                                this.available_channels =
+                                    Some(response.channels.into_iter().map(Into::into).collect());
+
+                                let mut to_remove = Vec::new();
+                                for (channel_id, channel) in &this.channels {
+                                    if let Some(channel) = channel.upgrade(cx) {
+                                        channel.update(cx, |channel, cx| channel.rejoin(cx))
+                                    } else {
+                                        to_remove.push(*channel_id);
+                                    }
+                                }
+
+                                for channel_id in to_remove {
+                                    this.channels.remove(&channel_id);
+                                }
+                                cx.notify();
+                            });
+                        }
+                        Status::SignedOut { .. } => {
+                            this.update(&mut cx, |this, cx| {
+                                this.available_channels = None;
+                                this.channels.clear();
+                                cx.notify();
+                            });
+                        }
+                        _ => {}
+                    }
+                }
+                Ok(())
+            }
+            .log_err()
+        });
+
+        Self {
+            available_channels: None,
+            channels: Default::default(),
+            user_store,
+            client: rpc,
+            _task,
+        }
+    }
+
+    pub fn available_channels(&self) -> Option<&[ChannelDetails]> {
+        self.available_channels.as_ref().map(Vec::as_slice)
+    }
+
+    pub fn get_channel(
+        &mut self,
+        id: u64,
+        cx: &mut MutableAppContext,
+    ) -> Option<ModelHandle<Channel>> {
+        if let Some(channel) = self.channels.get(&id).and_then(|c| c.upgrade(cx)) {
+            return Some(channel);
+        }
+
+        let channels = self.available_channels.as_ref()?;
+        let details = channels.iter().find(|details| details.id == id)?.clone();
+        let channel = cx.add_model(|cx| {
+            Channel::new(details, self.user_store.clone(), self.client.clone(), cx)
+        });
+        self.channels.insert(id, channel.downgrade());
+        Some(channel)
+    }
+}
+
+impl Entity for Channel {
+    type Event = ChannelEvent;
+
+    fn release(&mut self, cx: &mut MutableAppContext) {
+        let rpc = self.rpc.clone();
+        let channel_id = self.details.id;
+        cx.foreground()
+            .spawn(async move {
+                if let Err(error) = rpc.send(proto::LeaveChannel { channel_id }).await {
+                    log::error!("error leaving channel: {}", error);
+                };
+            })
+            .detach()
+    }
+}
+
+impl Channel {
+    pub fn new(
+        details: ChannelDetails,
+        user_store: ModelHandle<UserStore>,
+        rpc: Arc<Client>,
+        cx: &mut ModelContext<Self>,
+    ) -> Self {
+        let _subscription = rpc.subscribe_to_entity(details.id, cx, Self::handle_message_sent);
+
+        {
+            let user_store = user_store.clone();
+            let rpc = rpc.clone();
+            let channel_id = details.id;
+            cx.spawn(|channel, mut cx| {
+                async move {
+                    let response = rpc.request(proto::JoinChannel { channel_id }).await?;
+                    let messages =
+                        messages_from_proto(response.messages, &user_store, &mut cx).await?;
+                    let loaded_all_messages = response.done;
+
+                    channel.update(&mut cx, |channel, cx| {
+                        channel.insert_messages(messages, cx);
+                        channel.loaded_all_messages = loaded_all_messages;
+                    });
+
+                    Ok(())
+                }
+                .log_err()
+            })
+            .detach();
+        }
+
+        Self {
+            details,
+            user_store,
+            rpc,
+            messages: Default::default(),
+            loaded_all_messages: false,
+            next_pending_message_id: 0,
+            rng: StdRng::from_entropy(),
+            _subscription,
+        }
+    }
+
+    pub fn name(&self) -> &str {
+        &self.details.name
+    }
+
+    pub fn send_message(
+        &mut self,
+        body: String,
+        cx: &mut ModelContext<Self>,
+    ) -> Result<Task<Result<()>>> {
+        if body.is_empty() {
+            Err(anyhow!("message body can't be empty"))?;
+        }
+
+        let current_user = self
+            .user_store
+            .read(cx)
+            .current_user()
+            .ok_or_else(|| anyhow!("current_user is not present"))?;
+
+        let channel_id = self.details.id;
+        let pending_id = ChannelMessageId::Pending(post_inc(&mut self.next_pending_message_id));
+        let nonce = self.rng.gen();
+        self.insert_messages(
+            SumTree::from_item(
+                ChannelMessage {
+                    id: pending_id,
+                    body: body.clone(),
+                    sender: current_user,
+                    timestamp: OffsetDateTime::now_utc(),
+                    nonce,
+                },
+                &(),
+            ),
+            cx,
+        );
+        let user_store = self.user_store.clone();
+        let rpc = self.rpc.clone();
+        Ok(cx.spawn(|this, mut cx| async move {
+            let request = rpc.request(proto::SendChannelMessage {
+                channel_id,
+                body,
+                nonce: Some(nonce.into()),
+            });
+            let response = request.await?;
+            let message = ChannelMessage::from_proto(
+                response.message.ok_or_else(|| anyhow!("invalid message"))?,
+                &user_store,
+                &mut cx,
+            )
+            .await?;
+            this.update(&mut cx, |this, cx| {
+                this.insert_messages(SumTree::from_item(message, &()), cx);
+                Ok(())
+            })
+        }))
+    }
+
+    pub fn load_more_messages(&mut self, cx: &mut ModelContext<Self>) -> bool {
+        if !self.loaded_all_messages {
+            let rpc = self.rpc.clone();
+            let user_store = self.user_store.clone();
+            let channel_id = self.details.id;
+            if let Some(before_message_id) =
+                self.messages.first().and_then(|message| match message.id {
+                    ChannelMessageId::Saved(id) => Some(id),
+                    ChannelMessageId::Pending(_) => None,
+                })
+            {
+                cx.spawn(|this, mut cx| {
+                    async move {
+                        let response = rpc
+                            .request(proto::GetChannelMessages {
+                                channel_id,
+                                before_message_id,
+                            })
+                            .await?;
+                        let loaded_all_messages = response.done;
+                        let messages =
+                            messages_from_proto(response.messages, &user_store, &mut cx).await?;
+                        this.update(&mut cx, |this, cx| {
+                            this.loaded_all_messages = loaded_all_messages;
+                            this.insert_messages(messages, cx);
+                        });
+                        Ok(())
+                    }
+                    .log_err()
+                })
+                .detach();
+                return true;
+            }
+        }
+        false
+    }
+
+    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.details.id;
+        cx.spawn(|this, mut cx| {
+            async move {
+                let response = rpc.request(proto::JoinChannel { channel_id }).await?;
+                let messages = messages_from_proto(response.messages, &user_store, &mut cx).await?;
+                let loaded_all_messages = response.done;
+
+                let pending_messages = this.update(&mut cx, |this, cx| {
+                    if let Some((first_new_message, last_old_message)) =
+                        messages.first().zip(this.messages.last())
+                    {
+                        if first_new_message.id > last_old_message.id {
+                            let old_messages = mem::take(&mut this.messages);
+                            cx.emit(ChannelEvent::MessagesUpdated {
+                                old_range: 0..old_messages.summary().count,
+                                new_count: 0,
+                            });
+                            this.loaded_all_messages = loaded_all_messages;
+                        }
+                    }
+
+                    this.insert_messages(messages, cx);
+                    if loaded_all_messages {
+                        this.loaded_all_messages = loaded_all_messages;
+                    }
+
+                    this.pending_messages().cloned().collect::<Vec<_>>()
+                });
+
+                for pending_message in pending_messages {
+                    let request = rpc.request(proto::SendChannelMessage {
+                        channel_id,
+                        body: pending_message.body,
+                        nonce: Some(pending_message.nonce.into()),
+                    });
+                    let response = request.await?;
+                    let message = ChannelMessage::from_proto(
+                        response.message.ok_or_else(|| anyhow!("invalid message"))?,
+                        &user_store,
+                        &mut cx,
+                    )
+                    .await?;
+                    this.update(&mut cx, |this, cx| {
+                        this.insert_messages(SumTree::from_item(message, &()), cx);
+                    });
+                }
+
+                Ok(())
+            }
+            .log_err()
+        })
+        .detach();
+    }
+
+    pub fn message_count(&self) -> usize {
+        self.messages.summary().count
+    }
+
+    pub fn messages(&self) -> &SumTree<ChannelMessage> {
+        &self.messages
+    }
+
+    pub fn message(&self, ix: usize) -> &ChannelMessage {
+        let mut cursor = self.messages.cursor::<Count>();
+        cursor.seek(&Count(ix), Bias::Right, &());
+        cursor.item().unwrap()
+    }
+
+    pub fn messages_in_range(&self, range: Range<usize>) -> impl Iterator<Item = &ChannelMessage> {
+        let mut cursor = self.messages.cursor::<Count>();
+        cursor.seek(&Count(range.start), Bias::Right, &());
+        cursor.take(range.len())
+    }
+
+    pub fn pending_messages(&self) -> impl Iterator<Item = &ChannelMessage> {
+        let mut cursor = self.messages.cursor::<ChannelMessageId>();
+        cursor.seek(&ChannelMessageId::Pending(0), Bias::Left, &());
+        cursor
+    }
+
+    fn handle_message_sent(
+        &mut self,
+        message: TypedEnvelope<proto::ChannelMessageSent>,
+        _: Arc<Client>,
+        cx: &mut ModelContext<Self>,
+    ) -> Result<()> {
+        let user_store = self.user_store.clone();
+        let message = message
+            .payload
+            .message
+            .ok_or_else(|| anyhow!("empty message"))?;
+
+        cx.spawn(|this, mut cx| {
+            async move {
+                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)
+                });
+                Ok(())
+            }
+            .log_err()
+        })
+        .detach();
+        Ok(())
+    }
+
+    fn insert_messages(&mut self, messages: SumTree<ChannelMessage>, cx: &mut ModelContext<Self>) {
+        if let Some((first_message, last_message)) = messages.first().zip(messages.last()) {
+            let nonces = messages
+                .cursor::<()>()
+                .map(|m| m.nonce)
+                .collect::<HashSet<_>>();
+
+            let mut old_cursor = self.messages.cursor::<(ChannelMessageId, Count)>();
+            let mut new_messages = old_cursor.slice(&first_message.id, Bias::Left, &());
+            let start_ix = old_cursor.start().1 .0;
+            let removed_messages = old_cursor.slice(&last_message.id, Bias::Right, &());
+            let removed_count = removed_messages.summary().count;
+            let new_count = messages.summary().count;
+            let end_ix = start_ix + removed_count;
+
+            new_messages.push_tree(messages, &());
+
+            let mut ranges = Vec::<Range<usize>>::new();
+            if new_messages.last().unwrap().is_pending() {
+                new_messages.push_tree(old_cursor.suffix(&()), &());
+            } else {
+                new_messages.push_tree(
+                    old_cursor.slice(&ChannelMessageId::Pending(0), Bias::Left, &()),
+                    &(),
+                );
+
+                while let Some(message) = old_cursor.item() {
+                    let message_ix = old_cursor.start().1 .0;
+                    if nonces.contains(&message.nonce) {
+                        if ranges.last().map_or(false, |r| r.end == message_ix) {
+                            ranges.last_mut().unwrap().end += 1;
+                        } else {
+                            ranges.push(message_ix..message_ix + 1);
+                        }
+                    } else {
+                        new_messages.push(message.clone(), &());
+                    }
+                    old_cursor.next(&());
+                }
+            }
+
+            drop(old_cursor);
+            self.messages = new_messages;
+
+            for range in ranges.into_iter().rev() {
+                cx.emit(ChannelEvent::MessagesUpdated {
+                    old_range: range,
+                    new_count: 0,
+                });
+            }
+            cx.emit(ChannelEvent::MessagesUpdated {
+                old_range: start_ix..end_ix,
+                new_count,
+            });
+            cx.notify();
+        }
+    }
+}
+
+async fn messages_from_proto(
+    proto_messages: Vec<proto::ChannelMessage>,
+    user_store: &ModelHandle<UserStore>,
+    cx: &mut AsyncAppContext,
+) -> Result<SumTree<ChannelMessage>> {
+    let unique_user_ids = proto_messages
+        .iter()
+        .map(|m| m.sender_id)
+        .collect::<HashSet<_>>()
+        .into_iter()
+        .collect();
+    user_store
+        .update(cx, |user_store, cx| {
+            user_store.load_users(unique_user_ids, cx)
+        })
+        .await?;
+
+    let mut messages = Vec::with_capacity(proto_messages.len());
+    for message in proto_messages {
+        messages.push(ChannelMessage::from_proto(message, user_store, cx).await?);
+    }
+    let mut result = SumTree::new();
+    result.extend(messages, &());
+    Ok(result)
+}
+
+impl From<proto::Channel> for ChannelDetails {
+    fn from(message: proto::Channel) -> Self {
+        Self {
+            id: message.id,
+            name: message.name,
+        }
+    }
+}
+
+impl ChannelMessage {
+    pub async fn from_proto(
+        message: proto::ChannelMessage,
+        user_store: &ModelHandle<UserStore>,
+        cx: &mut AsyncAppContext,
+    ) -> Result<Self> {
+        let sender = user_store
+            .update(cx, |user_store, cx| {
+                user_store.fetch_user(message.sender_id, cx)
+            })
+            .await?;
+        Ok(ChannelMessage {
+            id: ChannelMessageId::Saved(message.id),
+            body: message.body,
+            timestamp: OffsetDateTime::from_unix_timestamp(message.timestamp as i64)?,
+            sender,
+            nonce: message
+                .nonce
+                .ok_or_else(|| anyhow!("nonce is required"))?
+                .into(),
+        })
+    }
+
+    pub fn is_pending(&self) -> bool {
+        matches!(self.id, ChannelMessageId::Pending(_))
+    }
+}
+
+impl sum_tree::Item for ChannelMessage {
+    type Summary = ChannelMessageSummary;
+
+    fn summary(&self) -> Self::Summary {
+        ChannelMessageSummary {
+            max_id: self.id,
+            count: 1,
+        }
+    }
+}
+
+impl Default for ChannelMessageId {
+    fn default() -> Self {
+        Self::Saved(0)
+    }
+}
+
+impl sum_tree::Summary for ChannelMessageSummary {
+    type Context = ();
+
+    fn add_summary(&mut self, summary: &Self, _: &()) {
+        self.max_id = summary.max_id;
+        self.count += summary.count;
+    }
+}
+
+impl<'a> sum_tree::Dimension<'a, ChannelMessageSummary> for ChannelMessageId {
+    fn add_summary(&mut self, summary: &'a ChannelMessageSummary, _: &()) {
+        debug_assert!(summary.max_id > *self);
+        *self = summary.max_id;
+    }
+}
+
+impl<'a> sum_tree::Dimension<'a, ChannelMessageSummary> for Count {
+    fn add_summary(&mut self, summary: &'a ChannelMessageSummary, _: &()) {
+        self.0 += summary.count;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test::{FakeHttpClient, FakeServer};
+    use gpui::TestAppContext;
+    use surf::http::Response;
+
+    #[gpui::test]
+    async fn test_channel_messages(mut cx: TestAppContext) {
+        let user_id = 5;
+        let mut client = Client::new();
+        let http_client = FakeHttpClient::new(|_| async move { Ok(Response::new(404)) });
+        let server = FakeServer::for_client(user_id, &mut client, &cx).await;
+        let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http_client, cx));
+
+        let channel_list = cx.add_model(|cx| ChannelList::new(user_store, client.clone(), cx));
+        channel_list.read_with(&cx, |list, _| assert_eq!(list.available_channels(), None));
+
+        // Get the available channels.
+        let get_channels = server.receive::<proto::GetChannels>().await.unwrap();
+        server
+            .respond(
+                get_channels.receipt(),
+                proto::GetChannelsResponse {
+                    channels: vec![proto::Channel {
+                        id: 5,
+                        name: "the-channel".to_string(),
+                    }],
+                },
+            )
+            .await;
+        channel_list.next_notification(&cx).await;
+        channel_list.read_with(&cx, |list, _| {
+            assert_eq!(
+                list.available_channels().unwrap(),
+                &[ChannelDetails {
+                    id: 5,
+                    name: "the-channel".into(),
+                }]
+            )
+        });
+
+        let get_users = server.receive::<proto::GetUsers>().await.unwrap();
+        assert_eq!(get_users.payload.user_ids, vec![5]);
+        server
+            .respond(
+                get_users.receipt(),
+                proto::GetUsersResponse {
+                    users: vec![proto::User {
+                        id: 5,
+                        github_login: "nathansobo".into(),
+                        avatar_url: "http://avatar.com/nathansobo".into(),
+                    }],
+                },
+            )
+            .await;
+
+        // Join a channel and populate its existing messages.
+        let channel = channel_list
+            .update(&mut cx, |list, cx| {
+                let channel_id = list.available_channels().unwrap()[0].id;
+                list.get_channel(channel_id, cx)
+            })
+            .unwrap();
+        channel.read_with(&cx, |channel, _| assert!(channel.messages().is_empty()));
+        let join_channel = server.receive::<proto::JoinChannel>().await.unwrap();
+        server
+            .respond(
+                join_channel.receipt(),
+                proto::JoinChannelResponse {
+                    messages: vec![
+                        proto::ChannelMessage {
+                            id: 10,
+                            body: "a".into(),
+                            timestamp: 1000,
+                            sender_id: 5,
+                            nonce: Some(1.into()),
+                        },
+                        proto::ChannelMessage {
+                            id: 11,
+                            body: "b".into(),
+                            timestamp: 1001,
+                            sender_id: 6,
+                            nonce: Some(2.into()),
+                        },
+                    ],
+                    done: false,
+                },
+            )
+            .await;
+
+        // Client requests all users for the received messages
+        let mut get_users = server.receive::<proto::GetUsers>().await.unwrap();
+        get_users.payload.user_ids.sort();
+        assert_eq!(get_users.payload.user_ids, vec![6]);
+        server
+            .respond(
+                get_users.receipt(),
+                proto::GetUsersResponse {
+                    users: vec![proto::User {
+                        id: 6,
+                        github_login: "maxbrunsfeld".into(),
+                        avatar_url: "http://avatar.com/maxbrunsfeld".into(),
+                    }],
+                },
+            )
+            .await;
+
+        assert_eq!(
+            channel.next_event(&cx).await,
+            ChannelEvent::MessagesUpdated {
+                old_range: 0..0,
+                new_count: 2,
+            }
+        );
+        channel.read_with(&cx, |channel, _| {
+            assert_eq!(
+                channel
+                    .messages_in_range(0..2)
+                    .map(|message| (message.sender.github_login.clone(), message.body.clone()))
+                    .collect::<Vec<_>>(),
+                &[
+                    ("nathansobo".into(), "a".into()),
+                    ("maxbrunsfeld".into(), "b".into())
+                ]
+            );
+        });
+
+        // Receive a new message.
+        server
+            .send(proto::ChannelMessageSent {
+                channel_id: channel.read_with(&cx, |channel, _| channel.details.id),
+                message: Some(proto::ChannelMessage {
+                    id: 12,
+                    body: "c".into(),
+                    timestamp: 1002,
+                    sender_id: 7,
+                    nonce: Some(3.into()),
+                }),
+            })
+            .await;
+
+        // Client requests user for message since they haven't seen them yet
+        let get_users = server.receive::<proto::GetUsers>().await.unwrap();
+        assert_eq!(get_users.payload.user_ids, vec![7]);
+        server
+            .respond(
+                get_users.receipt(),
+                proto::GetUsersResponse {
+                    users: vec![proto::User {
+                        id: 7,
+                        github_login: "as-cii".into(),
+                        avatar_url: "http://avatar.com/as-cii".into(),
+                    }],
+                },
+            )
+            .await;
+
+        assert_eq!(
+            channel.next_event(&cx).await,
+            ChannelEvent::MessagesUpdated {
+                old_range: 2..2,
+                new_count: 1,
+            }
+        );
+        channel.read_with(&cx, |channel, _| {
+            assert_eq!(
+                channel
+                    .messages_in_range(2..3)
+                    .map(|message| (message.sender.github_login.clone(), message.body.clone()))
+                    .collect::<Vec<_>>(),
+                &[("as-cii".into(), "c".into())]
+            )
+        });
+
+        // Scroll up to view older messages.
+        channel.update(&mut cx, |channel, cx| {
+            assert!(channel.load_more_messages(cx));
+        });
+        let get_messages = server.receive::<proto::GetChannelMessages>().await.unwrap();
+        assert_eq!(get_messages.payload.channel_id, 5);
+        assert_eq!(get_messages.payload.before_message_id, 10);
+        server
+            .respond(
+                get_messages.receipt(),
+                proto::GetChannelMessagesResponse {
+                    done: true,
+                    messages: vec![
+                        proto::ChannelMessage {
+                            id: 8,
+                            body: "y".into(),
+                            timestamp: 998,
+                            sender_id: 5,
+                            nonce: Some(4.into()),
+                        },
+                        proto::ChannelMessage {
+                            id: 9,
+                            body: "z".into(),
+                            timestamp: 999,
+                            sender_id: 6,
+                            nonce: Some(5.into()),
+                        },
+                    ],
+                },
+            )
+            .await;
+
+        assert_eq!(
+            channel.next_event(&cx).await,
+            ChannelEvent::MessagesUpdated {
+                old_range: 0..0,
+                new_count: 2,
+            }
+        );
+        channel.read_with(&cx, |channel, _| {
+            assert_eq!(
+                channel
+                    .messages_in_range(0..2)
+                    .map(|message| (message.sender.github_login.clone(), message.body.clone()))
+                    .collect::<Vec<_>>(),
+                &[
+                    ("nathansobo".into(), "y".into()),
+                    ("maxbrunsfeld".into(), "z".into())
+                ]
+            );
+        });
+    }
+}

crates/client/src/http.rs

@@ -0,0 +1,26 @@
+pub use anyhow::{anyhow, Result};
+use futures::future::BoxFuture;
+use std::sync::Arc;
+pub use surf::{
+    http::{Method, Response as ServerResponse},
+    Request, Response, Url,
+};
+
+pub trait HttpClient: Send + Sync {
+    fn send<'a>(&'a self, req: Request) -> BoxFuture<'a, Result<Response>>;
+}
+
+pub fn client() -> Arc<dyn HttpClient> {
+    Arc::new(surf::client())
+}
+
+impl HttpClient for surf::Client {
+    fn send<'a>(&'a self, req: Request) -> BoxFuture<'a, Result<Response>> {
+        Box::pin(async move {
+            Ok(self
+                .send(req)
+                .await
+                .map_err(|e| anyhow!("http request failed: {}", e))?)
+        })
+    }
+}

crates/client/src/lib.rs

@@ -1,6 +1,10 @@
 #[cfg(any(test, feature = "test-support"))]
 pub mod test;
 
+pub mod channel;
+pub mod http;
+pub mod user;
+
 use anyhow::{anyhow, Context, Result};
 use async_recursion::async_recursion;
 use async_tungstenite::tungstenite::{
@@ -26,7 +30,9 @@ use surf::Url;
 use thiserror::Error;
 use util::ResultExt;
 
+pub use channel::*;
 pub use rpc::*;
+pub use user::*;
 
 lazy_static! {
     static ref ZED_SERVER_URL: String =

crates/client/src/test.rs

@@ -1,11 +1,13 @@
-use super::*;
-use std::sync::atomic::Ordering::SeqCst;
-
 use super::Client;
+use super::*;
+use crate::http::{HttpClient, Request, Response, ServerResponse};
+use futures::{future::BoxFuture, Future};
 use gpui::TestAppContext;
 use parking_lot::Mutex;
 use postage::{mpsc, prelude::Stream};
 use rpc::{proto, ConnectionId, Peer, Receipt, TypedEnvelope};
+use std::fmt;
+use std::sync::atomic::Ordering::SeqCst;
 use std::sync::{
     atomic::{AtomicBool, AtomicUsize},
     Arc,
@@ -154,3 +156,33 @@ impl FakeServer {
         self.connection_id.lock().expect("not connected")
     }
 }
+
+pub struct FakeHttpClient {
+    handler:
+        Box<dyn 'static + Send + Sync + Fn(Request) -> BoxFuture<'static, Result<ServerResponse>>>,
+}
+
+impl FakeHttpClient {
+    pub fn new<Fut, F>(handler: F) -> Arc<dyn HttpClient>
+    where
+        Fut: 'static + Send + Future<Output = Result<ServerResponse>>,
+        F: 'static + Send + Sync + Fn(Request) -> Fut,
+    {
+        Arc::new(Self {
+            handler: Box::new(move |req| Box::pin(handler(req))),
+        })
+    }
+}
+
+impl fmt::Debug for FakeHttpClient {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("FakeHttpClient").finish()
+    }
+}
+
+impl HttpClient for FakeHttpClient {
+    fn send<'a>(&'a self, req: Request) -> BoxFuture<'a, Result<Response>> {
+        let future = (self.handler)(req);
+        Box::pin(async move { future.await.map(Into::into) })
+    }
+}

crates/client/src/user.rs

@@ -0,0 +1,268 @@
+use super::{
+    http::{HttpClient, Method, Request, Url},
+    proto, Client, Status, TypedEnvelope,
+};
+use anyhow::{anyhow, Context, Result};
+use futures::future;
+use gpui::{AsyncAppContext, Entity, ImageData, ModelContext, ModelHandle, Task};
+use postage::{prelude::Stream, sink::Sink, watch};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
+use util::TryFutureExt as _;
+
+#[derive(Debug)]
+pub struct User {
+    pub id: u64,
+    pub github_login: String,
+    pub avatar: Option<Arc<ImageData>>,
+}
+
+#[derive(Debug)]
+pub struct Collaborator {
+    pub user: Arc<User>,
+    pub worktrees: Vec<WorktreeMetadata>,
+}
+
+#[derive(Debug)]
+pub struct WorktreeMetadata {
+    pub id: u64,
+    pub root_name: String,
+    pub is_shared: bool,
+    pub guests: Vec<Arc<User>>,
+}
+
+pub struct UserStore {
+    users: HashMap<u64, Arc<User>>,
+    current_user: watch::Receiver<Option<Arc<User>>>,
+    collaborators: Arc<[Collaborator]>,
+    rpc: Arc<Client>,
+    http: Arc<dyn HttpClient>,
+    _maintain_collaborators: Task<()>,
+    _maintain_current_user: Task<()>,
+}
+
+pub enum Event {}
+
+impl Entity for UserStore {
+    type Event = Event;
+}
+
+impl UserStore {
+    pub fn new(rpc: Arc<Client>, http: Arc<dyn HttpClient>, cx: &mut ModelContext<Self>) -> Self {
+        let (mut current_user_tx, current_user_rx) = watch::channel();
+        let (mut update_collaborators_tx, mut update_collaborators_rx) =
+            watch::channel::<Option<proto::UpdateCollaborators>>();
+        let update_collaborators_subscription = rpc.subscribe(
+            cx,
+            move |_: &mut Self, msg: TypedEnvelope<proto::UpdateCollaborators>, _, _| {
+                let _ = update_collaborators_tx.blocking_send(Some(msg.payload));
+                Ok(())
+            },
+        );
+        Self {
+            users: Default::default(),
+            current_user: current_user_rx,
+            collaborators: Arc::from([]),
+            rpc: rpc.clone(),
+            http,
+            _maintain_collaborators: cx.spawn_weak(|this, mut cx| async move {
+                let _subscription = update_collaborators_subscription;
+                while let Some(message) = update_collaborators_rx.recv().await {
+                    if let Some((message, this)) = message.zip(this.upgrade(&cx)) {
+                        this.update(&mut cx, |this, cx| this.update_collaborators(message, cx))
+                            .log_err()
+                            .await;
+                    }
+                }
+            }),
+            _maintain_current_user: cx.spawn_weak(|this, mut cx| async move {
+                let mut status = rpc.status();
+                while let Some(status) = status.recv().await {
+                    match status {
+                        Status::Connected { .. } => {
+                            if let Some((this, user_id)) = this.upgrade(&cx).zip(rpc.user_id()) {
+                                let user = this
+                                    .update(&mut cx, |this, cx| this.fetch_user(user_id, cx))
+                                    .log_err()
+                                    .await;
+                                current_user_tx.send(user).await.ok();
+                            }
+                        }
+                        Status::SignedOut => {
+                            current_user_tx.send(None).await.ok();
+                        }
+                        _ => {}
+                    }
+                }
+            }),
+        }
+    }
+
+    fn update_collaborators(
+        &mut self,
+        message: proto::UpdateCollaborators,
+        cx: &mut ModelContext<Self>,
+    ) -> Task<Result<()>> {
+        let mut user_ids = HashSet::new();
+        for collaborator in &message.collaborators {
+            user_ids.insert(collaborator.user_id);
+            user_ids.extend(
+                collaborator
+                    .worktrees
+                    .iter()
+                    .flat_map(|w| &w.guests)
+                    .copied(),
+            );
+        }
+
+        let load_users = self.load_users(user_ids.into_iter().collect(), cx);
+        cx.spawn(|this, mut cx| async move {
+            load_users.await?;
+
+            let mut collaborators = Vec::new();
+            for collaborator in message.collaborators {
+                collaborators.push(Collaborator::from_proto(collaborator, &this, &mut cx).await?);
+            }
+
+            this.update(&mut cx, |this, cx| {
+                collaborators.sort_by(|a, b| a.user.github_login.cmp(&b.user.github_login));
+                this.collaborators = collaborators.into();
+                cx.notify();
+            });
+
+            Ok(())
+        })
+    }
+
+    pub fn collaborators(&self) -> &Arc<[Collaborator]> {
+        &self.collaborators
+    }
+
+    pub fn load_users(
+        &mut self,
+        mut user_ids: Vec<u64>,
+        cx: &mut ModelContext<Self>,
+    ) -> Task<Result<()>> {
+        let rpc = self.rpc.clone();
+        let http = self.http.clone();
+        user_ids.retain(|id| !self.users.contains_key(id));
+        cx.spawn_weak(|this, mut cx| async move {
+            if !user_ids.is_empty() {
+                let response = rpc.request(proto::GetUsers { user_ids }).await?;
+                let new_users = future::join_all(
+                    response
+                        .users
+                        .into_iter()
+                        .map(|user| User::new(user, http.as_ref())),
+                )
+                .await;
+
+                if let Some(this) = this.upgrade(&cx) {
+                    this.update(&mut cx, |this, _| {
+                        for user in new_users {
+                            this.users.insert(user.id, Arc::new(user));
+                        }
+                    });
+                }
+            }
+
+            Ok(())
+        })
+    }
+
+    pub fn fetch_user(
+        &mut self,
+        user_id: u64,
+        cx: &mut ModelContext<Self>,
+    ) -> Task<Result<Arc<User>>> {
+        if let Some(user) = self.users.get(&user_id).cloned() {
+            return cx.spawn_weak(|_, _| async move { Ok(user) });
+        }
+
+        let load_users = self.load_users(vec![user_id], cx);
+        cx.spawn(|this, mut cx| async move {
+            load_users.await?;
+            this.update(&mut cx, |this, _| {
+                this.users
+                    .get(&user_id)
+                    .cloned()
+                    .ok_or_else(|| anyhow!("server responded with no users"))
+            })
+        })
+    }
+
+    pub fn current_user(&self) -> Option<Arc<User>> {
+        self.current_user.borrow().clone()
+    }
+
+    pub fn watch_current_user(&self) -> watch::Receiver<Option<Arc<User>>> {
+        self.current_user.clone()
+    }
+}
+
+impl User {
+    async fn new(message: proto::User, http: &dyn HttpClient) -> Self {
+        User {
+            id: message.id,
+            github_login: message.github_login,
+            avatar: fetch_avatar(http, &message.avatar_url).warn_on_err().await,
+        }
+    }
+}
+
+impl Collaborator {
+    async fn from_proto(
+        collaborator: proto::Collaborator,
+        user_store: &ModelHandle<UserStore>,
+        cx: &mut AsyncAppContext,
+    ) -> Result<Self> {
+        let user = user_store
+            .update(cx, |user_store, cx| {
+                user_store.fetch_user(collaborator.user_id, cx)
+            })
+            .await?;
+        let mut worktrees = Vec::new();
+        for worktree in collaborator.worktrees {
+            let mut guests = Vec::new();
+            for participant_id in worktree.guests {
+                guests.push(
+                    user_store
+                        .update(cx, |user_store, cx| {
+                            user_store.fetch_user(participant_id, cx)
+                        })
+                        .await?,
+                );
+            }
+            worktrees.push(WorktreeMetadata {
+                id: worktree.id,
+                root_name: worktree.root_name,
+                is_shared: worktree.is_shared,
+                guests,
+            });
+        }
+        Ok(Self { user, worktrees })
+    }
+}
+
+async fn fetch_avatar(http: &dyn HttpClient, url: &str) -> Result<Arc<ImageData>> {
+    let url = Url::parse(url).with_context(|| format!("failed to parse avatar url {:?}", url))?;
+    let mut request = Request::new(Method::Get, url);
+    request.middleware(surf::middleware::Redirect::default());
+
+    let mut response = http
+        .send(request)
+        .await
+        .map_err(|e| anyhow!("failed to send user avatar request: {}", e))?;
+    if !response.status().is_success() {
+        return Err(anyhow!("avatar request failed {:?}", response.status()));
+    }
+    let bytes = response
+        .body_bytes()
+        .await
+        .map_err(|e| anyhow!("failed to read user avatar response body: {}", e))?;
+    let format = image::guess_format(&bytes)?;
+    let image = image::load_from_memory_with_format(&bytes, format)?.into_bgra8();
+    Ok(ImageData::new(image))
+}