use crate::{    search::SearchQuery,    worktree_store::{WorktreeStore, WorktreeStoreEvent},    ProjectItem as _, 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, DiskState, 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}; /// A set of open buffers.pub struct BufferStore {    state: BufferStoreState,    #[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 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 {    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 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 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,                        )                    }                    _ => {}                }            }        })        .detach();    }     fn local_worktree_entries_changed(        this: &mut BufferStore,        worktree_handle: &Model<Worktree>,        changes: &[(Arc<Path>, ProjectEntryId, PathChange)],        cx: &mut ModelContext<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_git_repos_changed(        this: &mut BufferStore,        worktree_handle: Model<Worktree>,        changed_repos: &UpdatedGitRepositoriesSet,        cx: &mut ModelContext<BufferStore>,    ) {        debug_assert!(worktree_handle.read(cx).is_local());         // Identify the loading buffers whose containing repository that has changed.        let future_buffers = this            .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 {                    BufferStore::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 = this            .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<_>>();

Lines 1–500 of 2124. Load more