use crate::{    search::SearchQuery,    worktree_store::{WorktreeStore, WorktreeStoreEvent},    Item, NoRepositoryError, ProjectPath,};use ::git::{parse_git_remote_url, BuildPermalinkParams, GitHostingProviderRegistry};use anyhow::{anyhow, Context as _, Result};use client::Client;use collections::{hash_map, HashMap, HashSet};use fs::Fs;use futures::{channel::oneshot, stream::FuturesUnordered, StreamExt};use git::blame::Blame;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, File as _, Language, Operation,};use rpc::{proto, AnyProtoClient, ErrorExt as _, TypedEnvelope};use smol::channel::Receiver;use std::{io, ops::Range, path::Path, str::FromStr as _, sync::Arc, time::Instant};use text::BufferId;use util::{debug_panic, maybe, ResultExt as _, TryFutureExt};use worktree::{File, PathChange, ProjectEntryId, UpdatedGitRepositoriesSet, Worktree, WorktreeId}; trait BufferStoreImpl {    fn open_buffer(        &self,        path: Arc<Path>,        worktree: Model<Worktree>,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<Model<Buffer>>>;     fn save_buffer(        &self,        buffer: Model<Buffer>,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<()>>;     fn save_buffer_as(        &self,        buffer: Model<Buffer>,        path: ProjectPath,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<()>>;     fn create_buffer(&self, cx: &mut ModelContext<BufferStore>) -> Task<Result<Model<Buffer>>>;     fn reload_buffers(        &self,        buffers: HashSet<Model<Buffer>>,        push_to_history: bool,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<ProjectTransaction>>;     fn as_remote(&self) -> Option<Model<RemoteBufferStore>>;    fn as_local(&self) -> Option<Model<LocalBufferStore>>;} 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>,    buffer_store: WeakModel<BufferStore>,} struct LocalBufferStore {    local_buffer_ids_by_path: HashMap<ProjectPath, BufferId>,    local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,    buffer_store: WeakModel<BufferStore>,    worktree_store: Model<WorktreeStore>,    _subscription: Subscription,} /// A set of open buffers.pub struct BufferStore {    state: Box<dyn BufferStoreImpl>,    #[allow(clippy::type_complexity)]    loading_buffers_by_path: HashMap<        ProjectPath,        postage::watch::Receiver<Option<Result<Model<Buffer>, Arc<anyhow::Error>>>>,    >,    worktree_store: Model<WorktreeStore>,    opened_buffers: HashMap<BufferId, OpenBuffer>,    downstream_client: Option<(AnyProtoClient, u64)>,    shared_buffers: HashMap<proto::PeerId, HashSet<Model<Buffer>>>,} enum OpenBuffer {    Buffer(WeakModel<Buffer>),    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 {    pub fn wait_for_remote_buffer(        &mut self,        id: BufferId,        cx: &mut AppContext,    ) -> Task<Result<Model<Buffer>>> {        let buffer_store = self.buffer_store.clone();        let (tx, rx) = oneshot::channel();        self.remote_buffer_listeners.entry(id).or_default().push(tx);         cx.spawn(|cx| async move {            if let Some(buffer) = buffer_store                .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<Self>,    ) -> 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<Self>,    ) -> 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<Self>,    ) -> 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)        })    }} impl BufferStoreImpl for Model<RemoteBufferStore> {    fn as_remote(&self) -> Option<Model<RemoteBufferStore>> {        Some(self.clone())    }     fn as_local(&self) -> Option<Model<LocalBufferStore>> {        None    }     fn save_buffer(        &self,        buffer: Model<Buffer>,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<()>> {        self.update(cx, |this, cx| {            this.save_remote_buffer(buffer.clone(), None, cx)        })    }    fn save_buffer_as(        &self,        buffer: Model<Buffer>,        path: ProjectPath,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<()>> {        self.update(cx, |this, cx| {            this.save_remote_buffer(buffer, Some(path.to_proto()), cx)        })    }     fn open_buffer(        &self,        path: Arc<Path>,        worktree: Model<Worktree>,        cx: &mut ModelContext<BufferStore>,    ) -> Task<Result<Model<Buffer>>> {        self.update(cx, |this, cx| {            let worktree_id = worktree.read(cx).id().to_proto();            let project_id = this.project_id;            let client = this.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>>> {        self.update(cx, |this, cx| {            let create = this.upstream_client.request(proto::OpenNewBuffer {                project_id: this.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>> {        self.update(cx, |this, cx| {            let request = this.upstream_client.request(proto::ReloadBuffers {                project_id: this.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 save_local_buffer(        &self,        buffer_handle: Model<Buffer>,        worktree: Model<Worktree>,        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((downstream_client, project_id)) = this.downstream_client(cx) {                    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<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>,    ) {

Lines 1–500 of 2211. Load more