mod channel_index; use crate::{channel_buffer::ChannelBuffer, channel_chat::ChannelChat, ChannelMessage};use anyhow::{anyhow, Result};use channel_index::ChannelIndex;use client::{ChannelId, Client, ClientSettings, Subscription, User, UserId, UserStore};use collections::{hash_map, HashMap, HashSet};use futures::{channel::mpsc, future::Shared, Future, FutureExt, StreamExt};use gpui::{    App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Global, SharedString, Task,    WeakEntity,};use language::Capability;use rpc::{    proto::{self, ChannelRole, ChannelVisibility},    TypedEnvelope,};use settings::Settings;use std::{mem, sync::Arc, time::Duration};use util::{maybe, ResultExt}; pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30); pub fn init(client: &Arc<Client>, user_store: Entity<UserStore>, cx: &mut App) {    let channel_store = cx.new(|cx| ChannelStore::new(client.clone(), user_store.clone(), cx));    cx.set_global(GlobalChannelStore(channel_store));} #[derive(Debug, Clone, Default, PartialEq)]struct NotesVersion {    epoch: u64,    version: clock::Global,} pub struct ChannelStore {    pub channel_index: ChannelIndex,    channel_invitations: Vec<Arc<Channel>>,    channel_participants: HashMap<ChannelId, Vec<Arc<User>>>,    channel_states: HashMap<ChannelId, ChannelState>,    outgoing_invites: HashSet<(ChannelId, UserId)>,    update_channels_tx: mpsc::UnboundedSender<proto::UpdateChannels>,    opened_buffers: HashMap<ChannelId, OpenEntityHandle<ChannelBuffer>>,    opened_chats: HashMap<ChannelId, OpenEntityHandle<ChannelChat>>,    client: Arc<Client>,    did_subscribe: bool,    user_store: Entity<UserStore>,    _rpc_subscriptions: [Subscription; 2],    _watch_connection_status: Task<Option<()>>,    disconnect_channel_buffers_task: Option<Task<()>>,    _update_channels: Task<()>,} #[derive(Clone, Debug)]pub struct Channel {    pub id: ChannelId,    pub name: SharedString,    pub visibility: proto::ChannelVisibility,    pub parent_path: Vec<ChannelId>,} #[derive(Default, Debug)]pub struct ChannelState {    latest_chat_message: Option<u64>,    latest_notes_version: NotesVersion,    observed_notes_version: NotesVersion,    observed_chat_message: Option<u64>,    role: Option<ChannelRole>,} impl Channel {    pub fn link(&self, cx: &App) -> String {        format!(            "{}/channel/{}-{}",            ClientSettings::get_global(cx).server_url,            Self::slug(&self.name),            self.id        )    }     pub fn notes_link(&self, heading: Option<String>, cx: &App) -> String {        self.link(cx)            + "/notes"            + &heading                .map(|h| format!("#{}", Self::slug(&h)))                .unwrap_or_default()    }     pub fn is_root_channel(&self) -> bool {        self.parent_path.is_empty()    }     pub fn root_id(&self) -> ChannelId {        self.parent_path.first().copied().unwrap_or(self.id)    }     pub fn slug(str: &str) -> String {        let slug: String = str            .chars()            .map(|c| if c.is_alphanumeric() { c } else { '-' })            .collect();         slug.trim_matches(|c| c == '-').to_string()    }} #[derive(Debug)]pub struct ChannelMembership {    pub user: Arc<User>,    pub kind: proto::channel_member::Kind,    pub role: proto::ChannelRole,}impl ChannelMembership {    pub fn sort_key(&self) -> MembershipSortKey {        MembershipSortKey {            role_order: match self.role {                proto::ChannelRole::Admin => 0,                proto::ChannelRole::Member => 1,                proto::ChannelRole::Banned => 2,                proto::ChannelRole::Talker => 3,                proto::ChannelRole::Guest => 4,            },            kind_order: match self.kind {                proto::channel_member::Kind::Member => 0,                proto::channel_member::Kind::Invitee => 1,            },            username_order: self.user.github_login.as_str(),        }    }} #[derive(PartialOrd, Ord, PartialEq, Eq)]pub struct MembershipSortKey<'a> {    role_order: u8,    kind_order: u8,    username_order: &'a str,} pub enum ChannelEvent {    ChannelCreated(ChannelId),    ChannelRenamed(ChannelId),} impl EventEmitter<ChannelEvent> for ChannelStore {} enum OpenEntityHandle<E> {    Open(WeakEntity<E>),    Loading(Shared<Task<Result<Entity<E>, Arc<anyhow::Error>>>>),} struct GlobalChannelStore(Entity<ChannelStore>); impl Global for GlobalChannelStore {} impl ChannelStore {    pub fn global(cx: &App) -> Entity<Self> {        cx.global::<GlobalChannelStore>().0.clone()    }     pub fn new(client: Arc<Client>, user_store: Entity<UserStore>, cx: &mut Context<Self>) -> Self {        let rpc_subscriptions = [            client.add_message_handler(cx.weak_entity(), Self::handle_update_channels),            client.add_message_handler(cx.weak_entity(), Self::handle_update_user_channels),        ];         let mut connection_status = client.status();        let (update_channels_tx, mut update_channels_rx) = mpsc::unbounded();        let watch_connection_status = cx.spawn(|this, mut cx| async move {            while let Some(status) = connection_status.next().await {                let this = this.upgrade()?;                match status {                    client::Status::Connected { .. } => {                        this.update(&mut cx, |this, cx| this.handle_connect(cx))                            .ok()?                            .await                            .log_err()?;                    }                    client::Status::SignedOut | client::Status::UpgradeRequired => {                        this.update(&mut cx, |this, cx| this.handle_disconnect(false, cx))                            .ok();                    }                    _ => {                        this.update(&mut cx, |this, cx| this.handle_disconnect(true, cx))                            .ok();                    }                }            }            Some(())        });         Self {            channel_invitations: Vec::default(),            channel_index: ChannelIndex::default(),            channel_participants: Default::default(),            outgoing_invites: Default::default(),            opened_buffers: Default::default(),            opened_chats: Default::default(),            update_channels_tx,            client,            user_store,            _rpc_subscriptions: rpc_subscriptions,            _watch_connection_status: watch_connection_status,            disconnect_channel_buffers_task: None,            _update_channels: cx.spawn(|this, mut cx| async move {                maybe!(async move {                    while let Some(update_channels) = update_channels_rx.next().await {                        if let Some(this) = this.upgrade() {                            let update_task = this.update(&mut cx, |this, cx| {                                this.update_channels(update_channels, cx)                            })?;                            if let Some(update_task) = update_task {                                update_task.await.log_err();                            }                        }                    }                    anyhow::Ok(())                })                .await                .log_err();            }),            channel_states: Default::default(),            did_subscribe: false,        }    }     pub fn initialize(&mut self) {        if !self.did_subscribe            && self                .client                .send(proto::SubscribeToChannels {})                .log_err()                .is_some()        {            self.did_subscribe = true;        }    }     pub fn client(&self) -> Arc<Client> {        self.client.clone()    }     /// Returns the number of unique channels in the store    pub fn channel_count(&self) -> usize {        self.channel_index.by_id().len()    }     /// Returns the index of a channel ID in the list of unique channels    pub fn index_of_channel(&self, channel_id: ChannelId) -> Option<usize> {        self.channel_index            .by_id()            .keys()            .position(|id| *id == channel_id)    }     /// Returns an iterator over all unique channels    pub fn channels(&self) -> impl '_ + Iterator<Item = &Arc<Channel>> {        self.channel_index.by_id().values()    }     /// Iterate over all entries in the channel DAG    pub fn ordered_channels(&self) -> impl '_ + Iterator<Item = (usize, &Arc<Channel>)> {        self.channel_index            .ordered_channels()            .iter()            .filter_map(move |id| {                let channel = self.channel_index.by_id().get(id)?;                Some((channel.parent_path.len(), channel))            })    }     pub fn channel_at_index(&self, ix: usize) -> Option<&Arc<Channel>> {        let channel_id = self.channel_index.ordered_channels().get(ix)?;        self.channel_index.by_id().get(channel_id)    }     pub fn channel_at(&self, ix: usize) -> Option<&Arc<Channel>> {        self.channel_index.by_id().values().nth(ix)    }     pub fn has_channel_invitation(&self, channel_id: ChannelId) -> bool {        self.channel_invitations            .iter()            .any(|channel| channel.id == channel_id)    }     pub fn channel_invitations(&self) -> &[Arc<Channel>] {        &self.channel_invitations    }     pub fn channel_for_id(&self, channel_id: ChannelId) -> Option<&Arc<Channel>> {        self.channel_index.by_id().get(&channel_id)    }     pub fn has_open_channel_buffer(&self, channel_id: ChannelId, _cx: &App) -> bool {        if let Some(buffer) = self.opened_buffers.get(&channel_id) {            if let OpenEntityHandle::Open(buffer) = buffer {                return buffer.upgrade().is_some();            }        }        false    }     pub fn open_channel_buffer(        &mut self,        channel_id: ChannelId,        cx: &mut Context<Self>,    ) -> Task<Result<Entity<ChannelBuffer>>> {        let client = self.client.clone();        let user_store = self.user_store.clone();        let channel_store = cx.entity();        self.open_channel_resource(            channel_id,            |this| &mut this.opened_buffers,            |channel, cx| ChannelBuffer::new(channel, client, user_store, channel_store, cx),            cx,        )    }     pub fn fetch_channel_messages(        &self,        message_ids: Vec<u64>,        cx: &mut Context<Self>,    ) -> Task<Result<Vec<ChannelMessage>>> {        let request = if message_ids.is_empty() {            None        } else {            Some(                self.client                    .request(proto::GetChannelMessagesById { message_ids }),            )        };        cx.spawn(|this, mut cx| async move {            if let Some(request) = request {                let response = request.await?;                let this = this                    .upgrade()                    .ok_or_else(|| anyhow!("channel store dropped"))?;                let user_store = this.update(&mut cx, |this, _| this.user_store.clone())?;                ChannelMessage::from_proto_vec(response.messages, &user_store, &mut cx).await            } else {                Ok(Vec::new())            }        })    }     pub fn has_channel_buffer_changed(&self, channel_id: ChannelId) -> bool {        self.channel_states            .get(&channel_id)            .is_some_and(|state| state.has_channel_buffer_changed())    }     pub fn has_new_messages(&self, channel_id: ChannelId) -> bool {        self.channel_states            .get(&channel_id)            .is_some_and(|state| state.has_new_messages())    }     pub fn set_acknowledged_message_id(&mut self, channel_id: ChannelId, message_id: Option<u64>) {        if let Some(state) = self.channel_states.get_mut(&channel_id) {            state.latest_chat_message = message_id;        }    }     pub fn last_acknowledge_message_id(&self, channel_id: ChannelId) -> Option<u64> {        self.channel_states.get(&channel_id).and_then(|state| {            if let Some(last_message_id) = state.latest_chat_message {                if state                    .last_acknowledged_message_id()                    .is_some_and(|id| id < last_message_id)                {                    return state.last_acknowledged_message_id();                }            }             None        })    }     pub fn acknowledge_message_id(        &mut self,        channel_id: ChannelId,        message_id: u64,        cx: &mut Context<Self>,    ) {        self.channel_states            .entry(channel_id)            .or_default()            .acknowledge_message_id(message_id);        cx.notify();    }     pub fn update_latest_message_id(        &mut self,        channel_id: ChannelId,        message_id: u64,        cx: &mut Context<Self>,    ) {        self.channel_states            .entry(channel_id)            .or_default()            .update_latest_message_id(message_id);        cx.notify();    }     pub fn acknowledge_notes_version(        &mut self,        channel_id: ChannelId,        epoch: u64,        version: &clock::Global,        cx: &mut Context<Self>,    ) {        self.channel_states            .entry(channel_id)            .or_default()            .acknowledge_notes_version(epoch, version);        cx.notify()    }     pub fn update_latest_notes_version(        &mut self,        channel_id: ChannelId,        epoch: u64,        version: &clock::Global,        cx: &mut Context<Self>,    ) {        self.channel_states            .entry(channel_id)            .or_default()            .update_latest_notes_version(epoch, version);        cx.notify()    }     pub fn open_channel_chat(        &mut self,        channel_id: ChannelId,        cx: &mut Context<Self>,    ) -> Task<Result<Entity<ChannelChat>>> {        let client = self.client.clone();        let user_store = self.user_store.clone();        let this = cx.entity();        self.open_channel_resource(            channel_id,            |this| &mut this.opened_chats,            |channel, cx| ChannelChat::new(channel, this, user_store, client, cx),            cx,        )    }     /// Asynchronously open a given resource associated with a channel.    ///    /// Make sure that the resource is only opened once, even if this method    /// is called multiple times with the same channel id while the first task    /// is still running.    fn open_channel_resource<T, F, Fut>(        &mut self,        channel_id: ChannelId,        get_map: fn(&mut Self) -> &mut HashMap<ChannelId, OpenEntityHandle<T>>,        load: F,        cx: &mut Context<Self>,    ) -> Task<Result<Entity<T>>>    where        F: 'static + FnOnce(Arc<Channel>, AsyncApp) -> Fut,        Fut: Future<Output = Result<Entity<T>>>,        T: 'static,    {        let task = loop {            match get_map(self).entry(channel_id) {                hash_map::Entry::Occupied(e) => match e.get() {                    OpenEntityHandle::Open(entity) => {                        if let Some(entity) = entity.upgrade() {                            break Task::ready(Ok(entity)).shared();                        } else {                            get_map(self).remove(&channel_id);                            continue;                        }                    }                    OpenEntityHandle::Loading(task) => {                        break task.clone();                    }                },                hash_map::Entry::Vacant(e) => {                    let task = cx                        .spawn(move |this, mut cx| async move {                            let channel = this.update(&mut cx, |this, _| {                                this.channel_for_id(channel_id).cloned().ok_or_else(|| {                                    Arc::new(anyhow!("no channel for id: {}", channel_id))                                })                            })??;                             load(channel, cx).await.map_err(Arc::new)                        })                        .shared();                     e.insert(OpenEntityHandle::Loading(task.clone()));                    cx.spawn({                        let task = task.clone();                        move |this, mut cx| async move {                            let result = task.await;                            this.update(&mut cx, |this, _| match result {                                Ok(model) => {                                    get_map(this).insert(

Lines 1–500 of 1225. Load more