use crate::{    lsp_store::OpenLspBufferHandle,    search::SearchQuery,    worktree_store::{WorktreeStore, WorktreeStoreEvent},    ProjectItem as _, ProjectPath,};use ::git::{parse_git_remote_url, BuildPermalinkParams, GitHostingProviderRegistry};use anyhow::{anyhow, bail, Context as _, Result};use client::Client;use collections::{hash_map, HashMap, HashSet};use fs::Fs;use futures::{channel::oneshot, future::Shared, Future, FutureExt as _, StreamExt};use git::{blame::Blame, diff::BufferDiff, repository::RepoPath};use gpui::{    AppContext, AsyncAppContext, Context as _, EventEmitter, Model, ModelContext, Subscription,    Task, WeakModel,};use http_client::Url;use language::{    proto::{        deserialize_line_ending, deserialize_version, serialize_line_ending, serialize_version,        split_operations,    },    Buffer, BufferEvent, Capability, DiskState, File as _, Language, Operation,};use rpc::{proto, AnyProtoClient, ErrorExt as _, TypedEnvelope};use serde::Deserialize;use smol::channel::Receiver;use std::{    io,    ops::Range,    path::{Path, PathBuf},    str::FromStr as _,    sync::Arc,    time::Instant,};use text::{BufferId, LineEnding, Rope};use util::{debug_panic, maybe, ResultExt as _, TryFutureExt};use worktree::{File, PathChange, ProjectEntryId, UpdatedGitRepositoriesSet, Worktree, WorktreeId}; /// A set of open buffers.pub struct BufferStore {    state: BufferStoreState,    #[allow(clippy::type_complexity)]    loading_buffers: HashMap<ProjectPath, Shared<Task<Result<Model<Buffer>, Arc<anyhow::Error>>>>>,    #[allow(clippy::type_complexity)]    loading_change_sets:        HashMap<BufferId, Shared<Task<Result<Model<BufferChangeSet>, Arc<anyhow::Error>>>>>,    worktree_store: Model<WorktreeStore>,    opened_buffers: HashMap<BufferId, OpenBuffer>,    downstream_client: Option<(AnyProtoClient, u64)>,    shared_buffers: HashMap<proto::PeerId, HashMap<BufferId, SharedBuffer>>,} #[derive(Hash, Eq, PartialEq, Clone)]struct SharedBuffer {    buffer: Model<Buffer>,    unstaged_changes: Option<Model<BufferChangeSet>>,    lsp_handle: Option<OpenLspBufferHandle>,} #[derive(Debug)]pub struct BufferChangeSet {    pub buffer_id: BufferId,    pub base_text: Option<Model<Buffer>>,    pub diff_to_buffer: git::diff::BufferDiff,    pub recalculate_diff_task: Option<Task<Result<()>>>,    pub diff_updated_futures: Vec<oneshot::Sender<()>>,    pub base_text_version: usize,} enum BufferStoreState {    Local(LocalBufferStore),    Remote(RemoteBufferStore),} struct RemoteBufferStore {    shared_with_me: HashSet<Model<Buffer>>,    upstream_client: AnyProtoClient,    project_id: u64,    loading_remote_buffers_by_id: HashMap<BufferId, Model<Buffer>>,    remote_buffer_listeners:        HashMap<BufferId, Vec<oneshot::Sender<Result<Model<Buffer>, anyhow::Error>>>>,    worktree_store: Model<WorktreeStore>,} struct LocalBufferStore {    local_buffer_ids_by_path: HashMap<ProjectPath, BufferId>,    local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,    worktree_store: Model<WorktreeStore>,    _subscription: Subscription,} enum OpenBuffer {    Complete {        buffer: WeakModel<Buffer>,        unstaged_changes: Option<WeakModel<BufferChangeSet>>,    },    Operations(Vec<Operation>),} pub enum BufferStoreEvent {    BufferAdded(Model<Buffer>),    BufferDropped(BufferId),    BufferChangedFilePath {        buffer: Model<Buffer>,        old_file: Option<Arc<dyn language::File>>,    },} #[derive(Default, Debug)]pub struct ProjectTransaction(pub HashMap<Model<Buffer>, language::Transaction>); impl EventEmitter<BufferStoreEvent> for BufferStore {} impl RemoteBufferStore {    fn load_staged_text(        &self,        buffer_id: BufferId,        cx: &AppContext,    ) -> Task<Result<Option<String>>> {        let project_id = self.project_id;        let client = self.upstream_client.clone();        cx.background_executor().spawn(async move {            Ok(client                .request(proto::GetStagedText {                    project_id,                    buffer_id: buffer_id.to_proto(),                })                .await?                .staged_text)        })    }    pub fn wait_for_remote_buffer(        &mut self,        id: BufferId,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<Model<Buffer>>> {        let (tx, rx) = oneshot::channel();        self.remote_buffer_listeners.entry(id).or_default().push(tx);         cx.spawn(|this, cx| async move {            if let Some(buffer) = this                .read_with(&cx, |buffer_store, _| buffer_store.get(id))                .ok()                .flatten()            {                return Ok(buffer);            }             cx.background_executor()                .spawn(async move { rx.await? })                .await        })    }     fn save_remote_buffer(        &self,        buffer_handle: Model<Buffer>,        new_path: Option<proto::ProjectPath>,        cx: &ModelContext<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(move |_, mut cx| async move {            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(&mut 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: u16,        capability: Capability,        cx: &mut ModelContext<BufferStore>,    ) -> Result<Option<Model<Buffer>>> {        match envelope            .payload            .variant            .ok_or_else(|| anyhow!("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)                            .ok_or_else(|| {                                anyhow!("no worktree found for id {}", file.worktree_id)                            })?;                        buffer_file = Some(Arc::new(File::from_proto(file, worktree.clone(), cx)?)                            as Arc<dyn language::File>);                    }                    Buffer::from_proto(replica_id, capability, state, buffer_file)                });                 match buffer_result {                    Ok(buffer) => {                        let buffer = cx.new_model(|_| 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()                    .ok_or_else(|| {                        anyhow!(                            "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));                }            }        }        return 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 ModelContext<BufferStore>,    ) -> Task<Result<ProjectTransaction>> {        cx.spawn(|this, mut cx| async move {            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(&mut 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(&mut cx, |buffer, _| {                        buffer.wait_for_edits(transaction.edit_ids.iter().copied())                    })?                    .await?;                 if push_to_history {                    buffer.update(&mut cx, |buffer, _| {                        buffer.push_transaction(transaction.clone(), Instant::now());                    })?;                }            }             Ok(project_transaction)        })    }     fn open_buffer(        &self,        path: Arc<Path>,        worktree: Model<Worktree>,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<Model<Buffer>>> {        let worktree_id = worktree.read(cx).id().to_proto();        let project_id = self.project_id;        let client = self.upstream_client.clone();        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)?;             let buffer = this                .update(&mut cx, {                    |this, cx| this.wait_for_remote_buffer(buffer_id, cx)                })?                .await?;             Ok(buffer)        })    }     fn create_buffer(&self, cx: &mut ModelContext<BufferStore>) -> Task<Result<Model<Buffer>>> {        let create = self.upstream_client.request(proto::OpenNewBuffer {            project_id: self.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        })    }     fn reload_buffers(        &self,        buffers: HashSet<Model<Buffer>>,        push_to_history: bool,        cx: &mut ModelContext<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(|this, mut cx| async move {            let response = request                .await?                .transaction                .ok_or_else(|| anyhow!("missing transaction"))?;            this.update(&mut cx, |this, cx| {                this.deserialize_project_transaction(response, push_to_history, cx)            })?            .await        })    }} impl LocalBufferStore {    fn load_staged_text(        &self,        buffer: &Model<Buffer>,        cx: &AppContext,    ) -> Task<Result<Option<String>>> {        let Some(file) = buffer.read(cx).file() else {            return Task::ready(Ok(None));        };        let worktree_id = file.worktree_id(cx);        let path = file.path().clone();        let Some(worktree) = self            .worktree_store            .read(cx)            .worktree_for_id(worktree_id, cx)        else {            return Task::ready(Err(anyhow!("no such worktree")));        };         worktree.read(cx).load_staged_file(path.as_ref(), cx)    }     fn save_local_buffer(        &self,        buffer_handle: Model<Buffer>,        worktree: Model<Worktree>,        path: Arc<Path>,        mut has_changed_file: bool,        cx: &mut ModelContext<BufferStore>,    ) -> 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.disk_state() == DiskState::New)        {            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.disk_state().mtime();            this.update(&mut 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(&mut cx, |buffer, cx| {                if has_changed_file {                    buffer.file_updated(new_file, cx);                }                buffer.did_save(version.clone(), mtime, cx);            })        })    }     fn subscribe_to_worktree(        &mut self,        worktree: &Model<Worktree>,        cx: &mut ModelContext<BufferStore>,    ) {        cx.subscribe(worktree, |this, worktree, event, cx| {            if worktree.read(cx).is_local() {                match event {                    worktree::Event::UpdatedEntries(changes) => {                        Self::local_worktree_entries_changed(this, &worktree, changes, cx);                    }                    worktree::Event::UpdatedGitRepositories(updated_repos) => {                        Self::local_worktree_git_repos_changed(                            this,                            worktree.clone(),                            updated_repos,                            cx,                        )                    }                    _ => {}                }

Lines 1–500 of 2516. Load more