use crate::{    ProjectPath,    lsp_store::OpenLspBufferHandle,    worktree_store::{WorktreeStore, WorktreeStoreEvent},};use anyhow::{Context as _, Result, anyhow};use client::Client;use collections::{HashMap, HashSet, hash_map};use futures::{Future, FutureExt as _, channel::oneshot, future::Shared};use gpui::{    App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Subscription, Task, WeakEntity,};use language::{    Buffer, BufferEvent, Capability, DiskState, File as _, Language, Operation,    proto::{        deserialize_line_ending, deserialize_version, serialize_line_ending, serialize_version,        split_operations,    },};use rpc::{    AnyProtoClient, ErrorCode, ErrorExt as _, TypedEnvelope,    proto::{self, PeerId},}; use settings::Settings;use std::{io, sync::Arc, time::Instant};use text::{BufferId, ReplicaId};use util::{ResultExt as _, TryFutureExt, debug_panic, maybe, rel_path::RelPath};use worktree::{File, PathChange, ProjectEntryId, Worktree, WorktreeId, WorktreeSettings}; /// A set of open buffers.pub struct BufferStore {    state: BufferStoreState,    #[allow(clippy::type_complexity)]    loading_buffers: HashMap<ProjectPath, Shared<Task<Result<Entity<Buffer>, Arc<anyhow::Error>>>>>,    worktree_store: Entity<WorktreeStore>,    opened_buffers: HashMap<BufferId, OpenBuffer>,    path_to_buffer_id: HashMap<ProjectPath, BufferId>,    downstream_client: Option<(AnyProtoClient, u64)>,    shared_buffers: HashMap<proto::PeerId, HashMap<BufferId, SharedBuffer>>,    non_searchable_buffers: HashSet<BufferId>,    project_search: RemoteProjectSearchState,} #[derive(Default)]struct RemoteProjectSearchState {    // List of ongoing project search chunks from our remote host. Used by the side issuing a search RPC request.    chunks: HashMap<u64, smol::channel::Sender<BufferId>>,    // Monotonously-increasing handle to hand out to remote host in order to identify the project search result chunk.    next_id: u64,    // Used by the side running the actual search for match candidates to potentially cancel the search prematurely.    searches_in_progress: HashMap<(PeerId, u64), Task<Result<()>>>,} #[derive(Hash, Eq, PartialEq, Clone)]struct SharedBuffer {    buffer: Entity<Buffer>,    lsp_handle: Option<OpenLspBufferHandle>,} enum BufferStoreState {    Local(LocalBufferStore),    Remote(RemoteBufferStore),} struct RemoteBufferStore {    shared_with_me: HashSet<Entity<Buffer>>,    upstream_client: AnyProtoClient,    project_id: u64,    loading_remote_buffers_by_id: HashMap<BufferId, Entity<Buffer>>,    remote_buffer_listeners:        HashMap<BufferId, Vec<oneshot::Sender<anyhow::Result<Entity<Buffer>>>>>,    worktree_store: Entity<WorktreeStore>,} struct LocalBufferStore {    local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,    worktree_store: Entity<WorktreeStore>,    _subscription: Subscription,} enum OpenBuffer {    Complete { buffer: WeakEntity<Buffer> },    Operations(Vec<Operation>),} pub enum BufferStoreEvent {    BufferAdded(Entity<Buffer>),    SharedBufferClosed(proto::PeerId, BufferId),    BufferDropped(BufferId),    BufferChangedFilePath {        buffer: Entity<Buffer>,        old_file: Option<Arc<dyn language::File>>,    },} #[derive(Default, Debug, Clone)]pub struct ProjectTransaction(pub HashMap<Entity<Buffer>, language::Transaction>); impl PartialEq for ProjectTransaction {    fn eq(&self, other: &Self) -> bool {        self.0.len() == other.0.len()            && self.0.iter().all(|(buffer, transaction)| {                other.0.get(buffer).is_some_and(|t| t.id == transaction.id)            })    }} impl EventEmitter<BufferStoreEvent> for BufferStore {} impl RemoteBufferStore {    pub fn wait_for_remote_buffer(        &mut self,        id: BufferId,        cx: &mut Context<BufferStore>,    ) -> Task<Result<Entity<Buffer>>> {        let (tx, rx) = oneshot::channel();        self.remote_buffer_listeners.entry(id).or_default().push(tx);         cx.spawn(async move |this, cx| {            if let Some(buffer) = this                .read_with(cx, |buffer_store, _| buffer_store.get(id))                .ok()                .flatten()            {                return Ok(buffer);            }             cx.background_spawn(async move { rx.await? }).await        })    }     fn save_remote_buffer(        &self,        buffer_handle: Entity<Buffer>,        new_path: Option<proto::ProjectPath>,        cx: &Context<BufferStore>,    ) -> Task<Result<()>> {        let buffer = buffer_handle.read(cx);        let buffer_id = buffer.remote_id().into();        let version = buffer.version();        let rpc = self.upstream_client.clone();        let project_id = self.project_id;        cx.spawn(async move |_, cx| {            let response = rpc                .request(proto::SaveBuffer {                    project_id,                    buffer_id,                    new_path,                    version: serialize_version(&version),                })                .await?;            let version = deserialize_version(&response.version);            let mtime = response.mtime.map(|mtime| mtime.into());             buffer_handle.update(cx, |buffer, cx| {                buffer.did_save(version.clone(), mtime, cx);            });             Ok(())        })    }     pub fn handle_create_buffer_for_peer(        &mut self,        envelope: TypedEnvelope<proto::CreateBufferForPeer>,        replica_id: ReplicaId,        capability: Capability,        cx: &mut Context<BufferStore>,    ) -> Result<Option<Entity<Buffer>>> {        match envelope.payload.variant.context("missing variant")? {            proto::create_buffer_for_peer::Variant::State(mut state) => {                let buffer_id = BufferId::new(state.id)?;                 let buffer_result = maybe!({                    let mut buffer_file = None;                    if let Some(file) = state.file.take() {                        let worktree_id = worktree::WorktreeId::from_proto(file.worktree_id);                        let worktree = self                            .worktree_store                            .read(cx)                            .worktree_for_id(worktree_id, cx)                            .with_context(|| {                                format!("no worktree found for id {}", file.worktree_id)                            })?;                        buffer_file = Some(Arc::new(File::from_proto(file, worktree, cx)?)                            as Arc<dyn language::File>);                    }                    Buffer::from_proto(replica_id, capability, state, buffer_file)                });                 match buffer_result {                    Ok(buffer) => {                        let buffer = cx.new(|_| buffer);                        self.loading_remote_buffers_by_id.insert(buffer_id, buffer);                    }                    Err(error) => {                        if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {                            for listener in listeners {                                listener.send(Err(anyhow!(error.cloned()))).ok();                            }                        }                    }                }            }            proto::create_buffer_for_peer::Variant::Chunk(chunk) => {                let buffer_id = BufferId::new(chunk.buffer_id)?;                let buffer = self                    .loading_remote_buffers_by_id                    .get(&buffer_id)                    .cloned()                    .with_context(|| {                        format!(                            "received chunk for buffer {} without initial state",                            chunk.buffer_id                        )                    })?;                 let result = maybe!({                    let operations = chunk                        .operations                        .into_iter()                        .map(language::proto::deserialize_operation)                        .collect::<Result<Vec<_>>>()?;                    buffer.update(cx, |buffer, cx| buffer.apply_ops(operations, cx));                    anyhow::Ok(())                });                 if let Err(error) = result {                    self.loading_remote_buffers_by_id.remove(&buffer_id);                    if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {                        for listener in listeners {                            listener.send(Err(error.cloned())).ok();                        }                    }                } else if chunk.is_last {                    self.loading_remote_buffers_by_id.remove(&buffer_id);                    if self.upstream_client.is_via_collab() {                        // retain buffers sent by peers to avoid races.                        self.shared_with_me.insert(buffer.clone());                    }                     if let Some(senders) = self.remote_buffer_listeners.remove(&buffer_id) {                        for sender in senders {                            sender.send(Ok(buffer.clone())).ok();                        }                    }                    return Ok(Some(buffer));                }            }        }        Ok(None)    }     pub fn incomplete_buffer_ids(&self) -> Vec<BufferId> {        self.loading_remote_buffers_by_id            .keys()            .copied()            .collect::<Vec<_>>()    }     pub fn deserialize_project_transaction(        &self,        message: proto::ProjectTransaction,        push_to_history: bool,        cx: &mut Context<BufferStore>,    ) -> Task<Result<ProjectTransaction>> {        cx.spawn(async move |this, cx| {            let mut project_transaction = ProjectTransaction::default();            for (buffer_id, transaction) in message.buffer_ids.into_iter().zip(message.transactions)            {                let buffer_id = BufferId::new(buffer_id)?;                let buffer = this                    .update(cx, |this, cx| this.wait_for_remote_buffer(buffer_id, cx))?                    .await?;                let transaction = language::proto::deserialize_transaction(transaction)?;                project_transaction.0.insert(buffer, transaction);            }             for (buffer, transaction) in &project_transaction.0 {                buffer                    .update(cx, |buffer, _| {                        buffer.wait_for_edits(transaction.edit_ids.iter().copied())                    })                    .await?;                 if push_to_history {                    buffer.update(cx, |buffer, _| {                        buffer.push_transaction(transaction.clone(), Instant::now());                        buffer.finalize_last_transaction();                    });                }            }             Ok(project_transaction)        })    }     fn open_buffer(        &self,        path: Arc<RelPath>,        worktree: Entity<Worktree>,        cx: &mut Context<BufferStore>,    ) -> Task<Result<Entity<Buffer>>> {        let worktree_id = worktree.read(cx).id().to_proto();        let project_id = self.project_id;        let client = self.upstream_client.clone();        cx.spawn(async move |this, cx| {            let response = client                .request(proto::OpenBufferByPath {                    project_id,                    worktree_id,                    path: path.to_proto(),                })                .await?;            let buffer_id = BufferId::new(response.buffer_id)?;             let buffer = this                .update(cx, {                    |this, cx| this.wait_for_remote_buffer(buffer_id, cx)                })?                .await?;             Ok(buffer)        })    }     fn create_buffer(        &self,        language: Option<Arc<Language>>,        project_searchable: bool,        cx: &mut Context<BufferStore>,    ) -> Task<Result<Entity<Buffer>>> {        let create = self.upstream_client.request(proto::OpenNewBuffer {            project_id: self.project_id,        });        cx.spawn(async move |this, cx| {            let response = create.await?;            let buffer_id = BufferId::new(response.buffer_id)?;             let buffer = this                .update(cx, |this, cx| {                    if !project_searchable {                        this.non_searchable_buffers.insert(buffer_id);                    }                    this.wait_for_remote_buffer(buffer_id, cx)                })?                .await?;            if let Some(language) = language {                buffer.update(cx, |buffer, cx| {                    buffer.set_language(Some(language), cx);                });            }            Ok(buffer)        })    }     fn reload_buffers(        &self,        buffers: HashSet<Entity<Buffer>>,        push_to_history: bool,        cx: &mut Context<BufferStore>,    ) -> Task<Result<ProjectTransaction>> {        let request = self.upstream_client.request(proto::ReloadBuffers {            project_id: self.project_id,            buffer_ids: buffers                .iter()                .map(|buffer| buffer.read(cx).remote_id().to_proto())                .collect(),        });         cx.spawn(async move |this, cx| {            let response = request.await?.transaction.context("missing transaction")?;            this.update(cx, |this, cx| {                this.deserialize_project_transaction(response, push_to_history, cx)            })?            .await        })    }} impl LocalBufferStore {    fn save_local_buffer(        &self,        buffer_handle: Entity<Buffer>,        worktree: Entity<Worktree>,        path: Arc<RelPath>,        mut has_changed_file: bool,        cx: &mut Context<BufferStore>,    ) -> Task<Result<()>> {        let buffer = buffer_handle.read(cx);         let text = buffer.as_rope().clone();        let line_ending = buffer.line_ending();        let encoding = buffer.encoding();        let has_bom = buffer.has_bom();        let version = buffer.version();        let buffer_id = buffer.remote_id();        let file = buffer.file().cloned();        if file            .as_ref()            .is_some_and(|file| file.disk_state() == DiskState::New)        {            has_changed_file = true;        }         let save = worktree.update(cx, |worktree, cx| {            worktree.write_file(path, text, line_ending, encoding, has_bom, cx)        });         cx.spawn(async move |this, cx| {            let new_file = save.await?;            let mtime = new_file.disk_state().mtime();            this.update(cx, |this, cx| {                if let Some((downstream_client, project_id)) = this.downstream_client.clone() {                    if has_changed_file {                        downstream_client                            .send(proto::UpdateBufferFile {                                project_id,                                buffer_id: buffer_id.to_proto(),                                file: Some(language::File::to_proto(&*new_file, cx)),                            })                            .log_err();                    }                    downstream_client                        .send(proto::BufferSaved {                            project_id,                            buffer_id: buffer_id.to_proto(),                            version: serialize_version(&version),                            mtime: mtime.map(|time| time.into()),                        })                        .log_err();                }            })?;            buffer_handle.update(cx, |buffer, cx| {                if has_changed_file {                    buffer.file_updated(new_file, cx);                }                buffer.did_save(version.clone(), mtime, cx);            });            Ok(())        })    }     fn subscribe_to_worktree(        &mut self,        worktree: &Entity<Worktree>,        cx: &mut Context<BufferStore>,    ) {        cx.subscribe(worktree, |this, worktree, event, cx| {            if worktree.read(cx).is_local()                && let worktree::Event::UpdatedEntries(changes) = event            {                Self::local_worktree_entries_changed(this, &worktree, changes, cx);            }        })        .detach();    }     fn local_worktree_entries_changed(        this: &mut BufferStore,        worktree_handle: &Entity<Worktree>,        changes: &[(Arc<RelPath>, ProjectEntryId, PathChange)],        cx: &mut Context<BufferStore>,    ) {        let snapshot = worktree_handle.read(cx).snapshot();        for (path, entry_id, _) in changes {            Self::local_worktree_entry_changed(                this,                *entry_id,                path,                worktree_handle,                &snapshot,                cx,            );        }    }     fn local_worktree_entry_changed(        this: &mut BufferStore,        entry_id: ProjectEntryId,        path: &Arc<RelPath>,        worktree: &Entity<worktree::Worktree>,        snapshot: &worktree::Snapshot,        cx: &mut Context<BufferStore>,    ) -> Option<()> {        let project_path = ProjectPath {            worktree_id: snapshot.id(),            path: path.clone(),        };         let buffer_id = this            .as_local_mut()            .and_then(|local| local.local_buffer_ids_by_entry_id.get(&entry_id))            .copied()            .or_else(|| this.path_to_buffer_id.get(&project_path).copied())?;         let buffer = if let Some(buffer) = this.get(buffer_id) {            Some(buffer)        } else {

Lines 1–500 of 1801. Load more