use crate::{    worktree_store::{WorktreeStore, WorktreeStoreEvent},    NoRepositoryError, ProjectPath,};use anyhow::{anyhow, Context as _, Result};use collections::{hash_map, HashMap};use futures::{channel::oneshot, stream::FuturesUnordered, StreamExt as _};use git::blame::Blame;use gpui::{    AppContext, AsyncAppContext, Context as _, EventEmitter, Model, ModelContext, Task, WeakModel,};use http_client::Url;use language::{    proto::{deserialize_line_ending, deserialize_version, serialize_version, split_operations},    Buffer, Capability, Event as BufferEvent, File as _, Language, Operation,};use rpc::{    proto::{self, AnyProtoClient, EnvelopedMessage, PeerId},    ErrorExt as _, TypedEnvelope,};use std::{io, path::Path, str::FromStr as _, sync::Arc};use text::BufferId;use util::{debug_panic, maybe, ResultExt as _};use worktree::{    File, PathChange, ProjectEntryId, RemoteWorktree, UpdatedGitRepositoriesSet, Worktree,    WorktreeId,}; /// A set of open buffers.pub struct BufferStore {    remote_id: Option<u64>,    #[allow(unused)]    worktree_store: Model<WorktreeStore>,    opened_buffers: HashMap<BufferId, OpenBuffer>,    local_buffer_ids_by_path: HashMap<ProjectPath, BufferId>,    local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,    #[allow(clippy::type_complexity)]    loading_buffers_by_path: HashMap<        ProjectPath,        postage::watch::Receiver<Option<Result<Model<Buffer>, Arc<anyhow::Error>>>>,    >,    loading_remote_buffers_by_id: HashMap<BufferId, Model<Buffer>>,    remote_buffer_listeners:        HashMap<BufferId, Vec<oneshot::Sender<Result<Model<Buffer>, anyhow::Error>>>>,} enum OpenBuffer {    Strong(Model<Buffer>),    Weak(WeakModel<Buffer>),    Operations(Vec<Operation>),} pub enum BufferStoreEvent {    BufferAdded(Model<Buffer>),    BufferChangedFilePath {        buffer: Model<Buffer>,        old_file: Option<Arc<dyn language::File>>,    },    MessageToReplicas(Box<proto::Envelope>),} impl EventEmitter<BufferStoreEvent> for BufferStore {} impl BufferStore {    /// Creates a buffer store, optionally retaining its buffers.    ///    /// If `retain_buffers` is `true`, then buffers are owned by the buffer store    /// and won't be released unless they are explicitly removed, or `retain_buffers`    /// is set to `false` via `set_retain_buffers`. Otherwise, buffers are stored as    /// weak handles.    pub fn new(        worktree_store: Model<WorktreeStore>,        remote_id: Option<u64>,        cx: &mut ModelContext<Self>,    ) -> Self {        cx.subscribe(&worktree_store, |this, _, event, cx| match event {            WorktreeStoreEvent::WorktreeAdded(worktree) => {                this.subscribe_to_worktree(worktree, cx);            }            _ => {}        })        .detach();         Self {            remote_id,            worktree_store,            opened_buffers: Default::default(),            remote_buffer_listeners: Default::default(),            loading_remote_buffers_by_id: Default::default(),            local_buffer_ids_by_path: Default::default(),            local_buffer_ids_by_entry_id: Default::default(),            loading_buffers_by_path: Default::default(),        }    }     pub fn open_buffer(        &mut self,        project_path: ProjectPath,        cx: &mut ModelContext<Self>,    ) -> Task<Result<Model<Buffer>>> {        let existing_buffer = self.get_by_path(&project_path, cx);        if let Some(existing_buffer) = existing_buffer {            return Task::ready(Ok(existing_buffer));        }         let Some(worktree) = self            .worktree_store            .read(cx)            .worktree_for_id(project_path.worktree_id, cx)        else {            return Task::ready(Err(anyhow!("no such worktree")));        };         let loading_watch = match self.loading_buffers_by_path.entry(project_path.clone()) {            // If the given path is already being loaded, then wait for that existing            // task to complete and return the same buffer.            hash_map::Entry::Occupied(e) => e.get().clone(),             // Otherwise, record the fact that this path is now being loaded.            hash_map::Entry::Vacant(entry) => {                let (mut tx, rx) = postage::watch::channel();                entry.insert(rx.clone());                 let project_path = project_path.clone();                let load_buffer = match worktree.read(cx) {                    Worktree::Local(_) => {                        self.open_local_buffer_internal(project_path.path.clone(), worktree, cx)                    }                    Worktree::Remote(tree) => {                        self.open_remote_buffer_internal(&project_path.path, tree, cx)                    }                };                 cx.spawn(move |this, mut cx| async move {                    let load_result = load_buffer.await;                    *tx.borrow_mut() = Some(this.update(&mut cx, |this, _| {                        // Record the fact that the buffer is no longer loading.                        this.loading_buffers_by_path.remove(&project_path);                        let buffer = load_result.map_err(Arc::new)?;                        Ok(buffer)                    })?);                    anyhow::Ok(())                })                .detach();                rx            }        };         cx.background_executor().spawn(async move {            Self::wait_for_loading_buffer(loading_watch)                .await                .map_err(|e| e.cloned())        })    }     fn subscribe_to_worktree(&mut self, worktree: &Model<Worktree>, cx: &mut ModelContext<Self>) {        cx.subscribe(worktree, |this, worktree, event, cx| {            if worktree.read(cx).is_local() {                match event {                    worktree::Event::UpdatedEntries(changes) => {                        this.local_worktree_entries_changed(&worktree, changes, cx);                    }                    worktree::Event::UpdatedGitRepositories(updated_repos) => {                        this.local_worktree_git_repos_changed(worktree.clone(), updated_repos, cx)                    }                    _ => {}                }            }        })        .detach();    }     fn local_worktree_entries_changed(        &mut self,        worktree_handle: &Model<Worktree>,        changes: &[(Arc<Path>, ProjectEntryId, PathChange)],        cx: &mut ModelContext<Self>,    ) {        let snapshot = worktree_handle.read(cx).snapshot();        for (path, entry_id, _) in changes {            self.local_worktree_entry_changed(*entry_id, path, worktree_handle, &snapshot, cx);        }    }     fn local_worktree_git_repos_changed(        &mut self,        worktree_handle: Model<Worktree>,        changed_repos: &UpdatedGitRepositoriesSet,        cx: &mut ModelContext<Self>,    ) {        debug_assert!(worktree_handle.read(cx).is_local());         // Identify the loading buffers whose containing repository that has changed.        let future_buffers = self            .loading_buffers()            .filter_map(|(project_path, receiver)| {                if project_path.worktree_id != worktree_handle.read(cx).id() {                    return None;                }                let path = &project_path.path;                changed_repos                    .iter()                    .find(|(work_dir, _)| path.starts_with(work_dir))?;                let path = path.clone();                Some(async move {                    Self::wait_for_loading_buffer(receiver)                        .await                        .ok()                        .map(|buffer| (buffer, path))                })            })            .collect::<FuturesUnordered<_>>();         // Identify the current buffers whose containing repository has changed.        let current_buffers = self            .buffers()            .filter_map(|buffer| {                let file = File::from_dyn(buffer.read(cx).file())?;                if file.worktree != worktree_handle {                    return None;                }                changed_repos                    .iter()                    .find(|(work_dir, _)| file.path.starts_with(work_dir))?;                Some((buffer, file.path.clone()))            })            .collect::<Vec<_>>();         if future_buffers.len() + current_buffers.len() == 0 {            return;        }         cx.spawn(move |this, mut cx| async move {            // Wait for all of the buffers to load.            let future_buffers = future_buffers.collect::<Vec<_>>().await;             // Reload the diff base for every buffer whose containing git repository has changed.            let snapshot =                worktree_handle.update(&mut cx, |tree, _| tree.as_local().unwrap().snapshot())?;            let diff_bases_by_buffer = cx                .background_executor()                .spawn(async move {                    let mut diff_base_tasks = future_buffers                        .into_iter()                        .flatten()                        .chain(current_buffers)                        .filter_map(|(buffer, path)| {                            let (repo_entry, local_repo_entry) = snapshot.repo_for_path(&path)?;                            let relative_path = repo_entry.relativize(&snapshot, &path).ok()?;                            Some(async move {                                let base_text =                                    local_repo_entry.repo().load_index_text(&relative_path);                                Some((buffer, base_text))                            })                        })                        .collect::<FuturesUnordered<_>>();                     let mut diff_bases = Vec::with_capacity(diff_base_tasks.len());                    while let Some(diff_base) = diff_base_tasks.next().await {                        if let Some(diff_base) = diff_base {                            diff_bases.push(diff_base);                        }                    }                    diff_bases                })                .await;             this.update(&mut cx, |this, cx| {                // Assign the new diff bases on all of the buffers.                for (buffer, diff_base) in diff_bases_by_buffer {                    let buffer_id = buffer.update(cx, |buffer, cx| {                        buffer.set_diff_base(diff_base.clone(), cx);                        buffer.remote_id().to_proto()                    });                    if let Some(project_id) = this.remote_id {                        cx.emit(BufferStoreEvent::MessageToReplicas(Box::new(                            proto::UpdateDiffBase {                                project_id,                                buffer_id,                                diff_base,                            }                            .into_envelope(0, None, None),                        )))                    }                }            })        })        .detach_and_log_err(cx);    }     fn open_local_buffer_internal(        &mut self,        path: Arc<Path>,        worktree: Model<Worktree>,        cx: &mut ModelContext<Self>,    ) -> Task<Result<Model<Buffer>>> {        let load_buffer = worktree.update(cx, |worktree, cx| {            let load_file = worktree.load_file(path.as_ref(), cx);            let reservation = cx.reserve_model();            let buffer_id = BufferId::from(reservation.entity_id().as_non_zero_u64());            cx.spawn(move |_, mut cx| async move {                let loaded = load_file.await?;                let text_buffer = cx                    .background_executor()                    .spawn(async move { text::Buffer::new(0, buffer_id, loaded.text) })                    .await;                cx.insert_model(reservation, |_| {                    Buffer::build(                        text_buffer,                        loaded.diff_base,                        Some(loaded.file),                        Capability::ReadWrite,                    )                })            })        });         cx.spawn(move |this, mut cx| async move {            let buffer = match load_buffer.await {                Ok(buffer) => Ok(buffer),                Err(error) if is_not_found_error(&error) => cx.new_model(|cx| {                    let buffer_id = BufferId::from(cx.entity_id().as_non_zero_u64());                    let text_buffer = text::Buffer::new(0, buffer_id, "".into());                    Buffer::build(                        text_buffer,                        None,                        Some(Arc::new(File {                            worktree,                            path,                            mtime: None,                            entry_id: None,                            is_local: true,                            is_deleted: false,                            is_private: false,                        })),                        Capability::ReadWrite,                    )                }),                Err(e) => Err(e),            }?;            this.update(&mut cx, |this, cx| {                this.add_buffer(buffer.clone(), cx).log_err();            })?;            Ok(buffer)        })    }     fn open_remote_buffer_internal(        &self,        path: &Arc<Path>,        worktree: &RemoteWorktree,        cx: &ModelContext<Self>,    ) -> Task<Result<Model<Buffer>>> {        let worktree_id = worktree.id().to_proto();        let project_id = worktree.project_id();        let client = worktree.client();        let path_string = path.clone().to_string_lossy().to_string();        cx.spawn(move |this, mut cx| async move {            let response = client                .request(proto::OpenBufferByPath {                    project_id,                    worktree_id,                    path: path_string,                })                .await?;            let buffer_id = BufferId::new(response.buffer_id)?;            this.update(&mut cx, |this, cx| {                this.wait_for_remote_buffer(buffer_id, cx)            })?            .await        })    }     pub fn create_buffer(        &mut self,        remote_client: Option<(AnyProtoClient, u64)>,        cx: &mut ModelContext<Self>,    ) -> Task<Result<Model<Buffer>>> {        if let Some((remote_client, project_id)) = remote_client {            let create = remote_client.request(proto::OpenNewBuffer { project_id });            cx.spawn(|this, mut cx| async move {                let response = create.await?;                let buffer_id = BufferId::new(response.buffer_id)?;                 this.update(&mut cx, |this, cx| {                    this.wait_for_remote_buffer(buffer_id, cx)                })?                .await            })        } else {            Task::ready(Ok(self.create_local_buffer("", None, cx)))        }    }     pub fn create_local_buffer(        &mut self,        text: &str,        language: Option<Arc<Language>>,        cx: &mut ModelContext<Self>,    ) -> Model<Buffer> {        let buffer = cx.new_model(|cx| {            Buffer::local(text, cx)                .with_language(language.unwrap_or_else(|| language::PLAIN_TEXT.clone()), cx)        });        self.add_buffer(buffer.clone(), cx).log_err();        buffer    }     pub fn save_buffer(        &mut self,        buffer: Model<Buffer>,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        let Some(file) = File::from_dyn(buffer.read(cx).file()) else {            return Task::ready(Err(anyhow!("buffer doesn't have a file")));        };        match file.worktree.read(cx) {            Worktree::Local(_) => {                self.save_local_buffer(file.worktree.clone(), buffer, file.path.clone(), false, cx)            }            Worktree::Remote(tree) => self.save_remote_buffer(buffer, None, tree, cx),        }    }     pub fn save_buffer_as(        &mut self,        buffer: Model<Buffer>,        path: ProjectPath,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        let Some(worktree) = self            .worktree_store            .read(cx)            .worktree_for_id(path.worktree_id, cx)        else {            return Task::ready(Err(anyhow!("no such worktree")));        };         let old_file = buffer.read(cx).file().cloned();         let task = match worktree.read(cx) {            Worktree::Local(_) => {                self.save_local_buffer(worktree, buffer.clone(), path.path, true, cx)            }            Worktree::Remote(tree) => {                self.save_remote_buffer(buffer.clone(), Some(path.to_proto()), tree, cx)            }        };        cx.spawn(|this, mut cx| async move {            task.await?;            this.update(&mut cx, |_, cx| {                cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });            })        })    }     fn save_local_buffer(        &self,        worktree: Model<Worktree>,        buffer_handle: Model<Buffer>,        path: Arc<Path>,        mut has_changed_file: bool,        cx: &mut ModelContext<Self>,    ) -> Task<Result<()>> {        let buffer = buffer_handle.read(cx);        let text = buffer.as_rope().clone();        let line_ending = buffer.line_ending();        let version = buffer.version();        let buffer_id = buffer.remote_id();        if buffer.file().is_some_and(|file| !file.is_created()) {            has_changed_file = true;        }         let save = worktree.update(cx, |worktree, cx| {            worktree.write_file(path.as_ref(), text, line_ending, cx)        });         cx.spawn(move |this, mut cx| async move {            let new_file = save.await?;            let mtime = new_file.mtime;            this.update(&mut cx, |this, cx| {                if let Some(project_id) = this.remote_id {                    if has_changed_file {                        cx.emit(BufferStoreEvent::MessageToReplicas(Box::new(                            proto::UpdateBufferFile {                                project_id,                                buffer_id: buffer_id.to_proto(),                                file: Some(language::File::to_proto(&*new_file, cx)),                            }                            .into_envelope(0, None, None),                        )));                    }                    cx.emit(BufferStoreEvent::MessageToReplicas(Box::new(                        proto::BufferSaved {                            project_id,                            buffer_id: buffer_id.to_proto(),                            version: serialize_version(&version),                            mtime: mtime.map(|time| time.into()),                        }                        .into_envelope(0, None, None),

Lines 1–500 of 1387. Load more