use super::{proto, Client, Status, TypedEnvelope};use anyhow::{anyhow, Context, Result};use collections::{hash_map::Entry, HashMap, HashSet};use feature_flags::FeatureFlagAppExt;use futures::{channel::mpsc, future, AsyncReadExt, Future, StreamExt};use gpui::{AsyncAppContext, Entity, ImageData, ModelContext, ModelHandle, Task};use postage::{sink::Sink, watch};use rpc::proto::{RequestMessage, UsersResponse};use std::sync::{Arc, Weak};use util::http::HttpClient;use util::TryFutureExt as _; pub type UserId = u64; #[derive(Default, Debug)]pub struct User {    pub id: UserId,    pub github_login: String,    pub avatar: Option<Arc<ImageData>>,} impl PartialOrd for User {    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {        Some(self.cmp(other))    }} impl Ord for User {    fn cmp(&self, other: &Self) -> std::cmp::Ordering {        self.github_login.cmp(&other.github_login)    }} impl PartialEq for User {    fn eq(&self, other: &Self) -> bool {        self.id == other.id && self.github_login == other.github_login    }} impl Eq for User {} #[derive(Debug, PartialEq)]pub struct Contact {    pub user: Arc<User>,    pub online: bool,    pub busy: bool,} #[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum ContactRequestStatus {    None,    RequestSent,    RequestReceived,    RequestAccepted,} pub struct UserStore {    users: HashMap<u64, Arc<User>>,    update_contacts_tx: mpsc::UnboundedSender<UpdateContacts>,    current_user: watch::Receiver<Option<Arc<User>>>,    contacts: Vec<Arc<Contact>>,    incoming_contact_requests: Vec<Arc<User>>,    outgoing_contact_requests: Vec<Arc<User>>,    pending_contact_requests: HashMap<u64, usize>,    invite_info: Option<InviteInfo>,    client: Weak<Client>,    http: Arc<dyn HttpClient>,    _maintain_contacts: Task<()>,    _maintain_current_user: Task<()>,} #[derive(Clone)]pub struct InviteInfo {    pub count: u32,    pub url: Arc<str>,} pub enum Event {    Contact {        user: Arc<User>,        kind: ContactEventKind,    },    ShowContacts,} #[derive(Clone, Copy)]pub enum ContactEventKind {    Requested,    Accepted,    Cancelled,} impl Entity for UserStore {    type Event = Event;} enum UpdateContacts {    Update(proto::UpdateContacts),    Wait(postage::barrier::Sender),    Clear(postage::barrier::Sender),} impl UserStore {    pub fn new(        client: Arc<Client>,        http: Arc<dyn HttpClient>,        cx: &mut ModelContext<Self>,    ) -> Self {        let (mut current_user_tx, current_user_rx) = watch::channel();        let (update_contacts_tx, mut update_contacts_rx) = mpsc::unbounded();        let rpc_subscriptions = vec![            client.add_message_handler(cx.handle(), Self::handle_update_contacts),            client.add_message_handler(cx.handle(), Self::handle_update_invite_info),            client.add_message_handler(cx.handle(), Self::handle_show_contacts),        ];        Self {            users: Default::default(),            current_user: current_user_rx,            contacts: Default::default(),            incoming_contact_requests: Default::default(),            outgoing_contact_requests: Default::default(),            invite_info: None,            client: Arc::downgrade(&client),            update_contacts_tx,            http,            _maintain_contacts: cx.spawn_weak(|this, mut cx| async move {                let _subscriptions = rpc_subscriptions;                while let Some(message) = update_contacts_rx.next().await {                    if let Some(this) = this.upgrade(&cx) {                        this.update(&mut cx, |this, cx| this.update_contacts(message, cx))                            .log_err()                            .await;                    }                }            }),            _maintain_current_user: cx.spawn_weak(|this, mut cx| async move {                let mut status = client.status();                while let Some(status) = status.next().await {                    match status {                        Status::Connected { .. } => {                            if let Some((this, user_id)) = this.upgrade(&cx).zip(client.user_id()) {                                let fetch_user = this                                    .update(&mut cx, |this, cx| this.get_user(user_id, cx))                                    .log_err();                                let fetch_metrics_id =                                    client.request(proto::GetPrivateUserInfo {}).log_err();                                let (user, info) = futures::join!(fetch_user, fetch_metrics_id);                                 if let Some(info) = info {                                    cx.update(|cx| {                                        cx.update_flags(info.staff, info.flags);                                        client.telemetry.set_authenticated_user_info(                                            Some(info.metrics_id.clone()),                                            info.staff,                                            cx,                                        )                                    });                                } else {                                    cx.read(|cx| {                                        client                                            .telemetry                                            .set_authenticated_user_info(None, false, cx)                                    });                                }                                 current_user_tx.send(user).await.ok();                                 this.update(&mut cx, |_, cx| {                                    cx.notify();                                });                            }                        }                        Status::SignedOut => {                            current_user_tx.send(None).await.ok();                            if let Some(this) = this.upgrade(&cx) {                                this.update(&mut cx, |this, cx| {                                    cx.notify();                                    this.clear_contacts()                                })                                .await;                            }                        }                        Status::ConnectionLost => {                            if let Some(this) = this.upgrade(&cx) {                                this.update(&mut cx, |this, cx| {                                    cx.notify();                                    this.clear_contacts()                                })                                .await;                            }                        }                        _ => {}                    }                }            }),            pending_contact_requests: Default::default(),        }    }     #[cfg(feature = "test-support")]    pub fn clear_cache(&mut self) {        self.users.clear();    }     async fn handle_update_invite_info(        this: ModelHandle<Self>,        message: TypedEnvelope<proto::UpdateInviteInfo>,        _: Arc<Client>,        mut cx: AsyncAppContext,    ) -> Result<()> {        this.update(&mut cx, |this, cx| {            this.invite_info = Some(InviteInfo {                url: Arc::from(message.payload.url),                count: message.payload.count,            });            cx.notify();        });        Ok(())    }     async fn handle_show_contacts(        this: ModelHandle<Self>,        _: TypedEnvelope<proto::ShowContacts>,        _: Arc<Client>,        mut cx: AsyncAppContext,    ) -> Result<()> {        this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts));        Ok(())    }     pub fn invite_info(&self) -> Option<&InviteInfo> {        self.invite_info.as_ref()    }     async fn handle_update_contacts(        this: ModelHandle<Self>,        message: TypedEnvelope<proto::UpdateContacts>,        _: Arc<Client>,        mut cx: AsyncAppContext,    ) -> Result<()> {        this.update(&mut cx, |this, _| {            this.update_contacts_tx                .unbounded_send(UpdateContacts::Update(message.payload))                .unwrap();        });        Ok(())    }     fn update_contacts(        &mut self,        message: UpdateContacts,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        match message {            UpdateContacts::Wait(barrier) => {                drop(barrier);                Task::ready(Ok(()))            }            UpdateContacts::Clear(barrier) => {                self.contacts.clear();                self.incoming_contact_requests.clear();                self.outgoing_contact_requests.clear();                drop(barrier);                Task::ready(Ok(()))            }            UpdateContacts::Update(message) => {                let mut user_ids = HashSet::default();                for contact in &message.contacts {                    user_ids.insert(contact.user_id);                }                user_ids.extend(message.incoming_requests.iter().map(|req| req.requester_id));                user_ids.extend(message.outgoing_requests.iter());                 let load_users = self.get_users(user_ids.into_iter().collect(), cx);                cx.spawn(|this, mut cx| async move {                    load_users.await?;                     // Users are fetched in parallel above and cached in call to get_users                    // No need to paralellize here                    let mut updated_contacts = Vec::new();                    for contact in message.contacts {                        let should_notify = contact.should_notify;                        updated_contacts.push((                            Arc::new(Contact::from_proto(contact, &this, &mut cx).await?),                            should_notify,                        ));                    }                     let mut incoming_requests = Vec::new();                    for request in message.incoming_requests {                        incoming_requests.push({                            let user = this                                .update(&mut cx, |this, cx| this.get_user(request.requester_id, cx))                                .await?;                            (user, request.should_notify)                        });                    }                     let mut outgoing_requests = Vec::new();                    for requested_user_id in message.outgoing_requests {                        outgoing_requests.push(                            this.update(&mut cx, |this, cx| this.get_user(requested_user_id, cx))                                .await?,                        );                    }                     let removed_contacts =                        HashSet::<u64>::from_iter(message.remove_contacts.iter().copied());                    let removed_incoming_requests =                        HashSet::<u64>::from_iter(message.remove_incoming_requests.iter().copied());                    let removed_outgoing_requests =                        HashSet::<u64>::from_iter(message.remove_outgoing_requests.iter().copied());                     this.update(&mut cx, |this, cx| {                        // Remove contacts                        this.contacts                            .retain(|contact| !removed_contacts.contains(&contact.user.id));                        // Update existing contacts and insert new ones                        for (updated_contact, should_notify) in updated_contacts {                            if should_notify {                                cx.emit(Event::Contact {                                    user: updated_contact.user.clone(),                                    kind: ContactEventKind::Accepted,                                });                            }                            match this.contacts.binary_search_by_key(                                &&updated_contact.user.github_login,                                |contact| &contact.user.github_login,                            ) {                                Ok(ix) => this.contacts[ix] = updated_contact,                                Err(ix) => this.contacts.insert(ix, updated_contact),                            }                        }                         // Remove incoming contact requests                        this.incoming_contact_requests.retain(|user| {                            if removed_incoming_requests.contains(&user.id) {                                cx.emit(Event::Contact {                                    user: user.clone(),                                    kind: ContactEventKind::Cancelled,                                });                                false                            } else {                                true                            }                        });                        // Update existing incoming requests and insert new ones                        for (user, should_notify) in incoming_requests {                            if should_notify {                                cx.emit(Event::Contact {                                    user: user.clone(),                                    kind: ContactEventKind::Requested,                                });                            }                             match this                                .incoming_contact_requests                                .binary_search_by_key(&&user.github_login, |contact| {                                    &contact.github_login                                }) {                                Ok(ix) => this.incoming_contact_requests[ix] = user,                                Err(ix) => this.incoming_contact_requests.insert(ix, user),                            }                        }                         // Remove outgoing contact requests                        this.outgoing_contact_requests                            .retain(|user| !removed_outgoing_requests.contains(&user.id));                        // Update existing incoming requests and insert new ones                        for request in outgoing_requests {                            match this                                .outgoing_contact_requests                                .binary_search_by_key(&&request.github_login, |contact| {                                    &contact.github_login                                }) {                                Ok(ix) => this.outgoing_contact_requests[ix] = request,                                Err(ix) => this.outgoing_contact_requests.insert(ix, request),                            }                        }                         cx.notify();                    });                     Ok(())                })            }        }    }     pub fn contacts(&self) -> &[Arc<Contact>] {        &self.contacts    }     pub fn has_contact(&self, user: &Arc<User>) -> bool {        self.contacts            .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)            .is_ok()    }     pub fn incoming_contact_requests(&self) -> &[Arc<User>] {        &self.incoming_contact_requests    }     pub fn outgoing_contact_requests(&self) -> &[Arc<User>] {        &self.outgoing_contact_requests    }     pub fn is_contact_request_pending(&self, user: &User) -> bool {        self.pending_contact_requests.contains_key(&user.id)    }     pub fn contact_request_status(&self, user: &User) -> ContactRequestStatus {        if self            .contacts            .binary_search_by_key(&&user.github_login, |contact| &contact.user.github_login)            .is_ok()        {            ContactRequestStatus::RequestAccepted        } else if self            .outgoing_contact_requests            .binary_search_by_key(&&user.github_login, |user| &user.github_login)            .is_ok()        {            ContactRequestStatus::RequestSent        } else if self            .incoming_contact_requests            .binary_search_by_key(&&user.github_login, |user| &user.github_login)            .is_ok()        {            ContactRequestStatus::RequestReceived        } else {            ContactRequestStatus::None        }    }     pub fn request_contact(        &mut self,        responder_id: u64,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        self.perform_contact_request(responder_id, proto::RequestContact { responder_id }, cx)    }     pub fn remove_contact(        &mut self,        user_id: u64,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        self.perform_contact_request(user_id, proto::RemoveContact { user_id }, cx)    }     pub fn respond_to_contact_request(        &mut self,        requester_id: u64,        accept: bool,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        self.perform_contact_request(            requester_id,            proto::RespondToContactRequest {                requester_id,                response: if accept {                    proto::ContactRequestResponse::Accept                } else {                    proto::ContactRequestResponse::Decline                } as i32,            },            cx,        )    }     pub fn dismiss_contact_request(        &mut self,        requester_id: u64,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        let client = self.client.upgrade();        cx.spawn_weak(|_, _| async move {            client                .ok_or_else(|| anyhow!("can't upgrade client reference"))?                .request(proto::RespondToContactRequest {                    requester_id,                    response: proto::ContactRequestResponse::Dismiss as i32,                })                .await?;            Ok(())        })    }     fn perform_contact_request<T: RequestMessage>(        &mut self,        user_id: u64,        request: T,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        let client = self.client.upgrade();        *self.pending_contact_requests.entry(user_id).or_insert(0) += 1;        cx.notify();         cx.spawn(|this, mut cx| async move {

Lines 1–500 of 694. Load more