1use crate::{
   2    lsp_store::OpenLspBufferHandle,
   3    search::SearchQuery,
   4    worktree_store::{WorktreeStore, WorktreeStoreEvent},
   5    ProjectItem as _, ProjectPath,
   6};
   7use ::git::{parse_git_remote_url, BuildPermalinkParams, GitHostingProviderRegistry};
   8use anyhow::{anyhow, bail, Context as _, Result};
   9use client::Client;
  10use collections::{hash_map, HashMap, HashSet};
  11use fs::Fs;
  12use futures::{channel::oneshot, future::Shared, Future, FutureExt as _, StreamExt};
  13use git::{blame::Blame, diff::BufferDiff, repository::RepoPath};
  14use gpui::{
  15    App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Subscription, Task, WeakEntity,
  16};
  17use http_client::Url;
  18use language::{
  19    proto::{
  20        deserialize_line_ending, deserialize_version, serialize_line_ending, serialize_version,
  21        split_operations,
  22    },
  23    Buffer, BufferEvent, Capability, DiskState, File as _, Language, LanguageRegistry, Operation,
  24};
  25use rpc::{proto, AnyProtoClient, ErrorExt as _, TypedEnvelope};
  26use serde::Deserialize;
  27use smol::channel::Receiver;
  28use std::{
  29    io,
  30    ops::Range,
  31    path::{Path, PathBuf},
  32    pin::pin,
  33    str::FromStr as _,
  34    sync::Arc,
  35    time::Instant,
  36};
  37use text::{BufferId, LineEnding, Rope};
  38use util::{debug_panic, maybe, ResultExt as _, TryFutureExt};
  39use worktree::{File, PathChange, ProjectEntryId, UpdatedGitRepositoriesSet, Worktree, WorktreeId};
  40
  41/// A set of open buffers.
  42pub struct BufferStore {
  43    state: BufferStoreState,
  44    #[allow(clippy::type_complexity)]
  45    loading_buffers: HashMap<ProjectPath, Shared<Task<Result<Entity<Buffer>, Arc<anyhow::Error>>>>>,
  46    #[allow(clippy::type_complexity)]
  47    loading_change_sets:
  48        HashMap<BufferId, Shared<Task<Result<Entity<BufferChangeSet>, Arc<anyhow::Error>>>>>,
  49    worktree_store: Entity<WorktreeStore>,
  50    opened_buffers: HashMap<BufferId, OpenBuffer>,
  51    downstream_client: Option<(AnyProtoClient, u64)>,
  52    shared_buffers: HashMap<proto::PeerId, HashMap<BufferId, SharedBuffer>>,
  53}
  54
  55#[derive(Hash, Eq, PartialEq, Clone)]
  56struct SharedBuffer {
  57    buffer: Entity<Buffer>,
  58    unstaged_changes: Option<Entity<BufferChangeSet>>,
  59    lsp_handle: Option<OpenLspBufferHandle>,
  60}
  61
  62pub struct BufferChangeSet {
  63    pub buffer_id: BufferId,
  64    pub base_text: Option<language::BufferSnapshot>,
  65    pub language: Option<Arc<Language>>,
  66    pub diff_to_buffer: git::diff::BufferDiff,
  67    pub recalculate_diff_task: Option<Task<Result<()>>>,
  68    pub diff_updated_futures: Vec<oneshot::Sender<()>>,
  69    pub language_registry: Option<Arc<LanguageRegistry>>,
  70}
  71
  72enum BufferStoreState {
  73    Local(LocalBufferStore),
  74    Remote(RemoteBufferStore),
  75}
  76
  77struct RemoteBufferStore {
  78    shared_with_me: HashSet<Entity<Buffer>>,
  79    upstream_client: AnyProtoClient,
  80    project_id: u64,
  81    loading_remote_buffers_by_id: HashMap<BufferId, Entity<Buffer>>,
  82    remote_buffer_listeners:
  83        HashMap<BufferId, Vec<oneshot::Sender<Result<Entity<Buffer>, anyhow::Error>>>>,
  84    worktree_store: Entity<WorktreeStore>,
  85}
  86
  87struct LocalBufferStore {
  88    local_buffer_ids_by_path: HashMap<ProjectPath, BufferId>,
  89    local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,
  90    worktree_store: Entity<WorktreeStore>,
  91    _subscription: Subscription,
  92}
  93
  94enum OpenBuffer {
  95    Complete {
  96        buffer: WeakEntity<Buffer>,
  97        unstaged_changes: Option<WeakEntity<BufferChangeSet>>,
  98    },
  99    Operations(Vec<Operation>),
 100}
 101
 102pub enum BufferStoreEvent {
 103    BufferAdded(Entity<Buffer>),
 104    BufferDropped(BufferId),
 105    BufferChangedFilePath {
 106        buffer: Entity<Buffer>,
 107        old_file: Option<Arc<dyn language::File>>,
 108    },
 109}
 110
 111#[derive(Default, Debug)]
 112pub struct ProjectTransaction(pub HashMap<Entity<Buffer>, language::Transaction>);
 113
 114impl EventEmitter<BufferStoreEvent> for BufferStore {}
 115
 116impl RemoteBufferStore {
 117    fn load_staged_text(&self, buffer_id: BufferId, cx: &App) -> Task<Result<Option<String>>> {
 118        let project_id = self.project_id;
 119        let client = self.upstream_client.clone();
 120        cx.background_executor().spawn(async move {
 121            Ok(client
 122                .request(proto::GetStagedText {
 123                    project_id,
 124                    buffer_id: buffer_id.to_proto(),
 125                })
 126                .await?
 127                .staged_text)
 128        })
 129    }
 130    pub fn wait_for_remote_buffer(
 131        &mut self,
 132        id: BufferId,
 133        cx: &mut Context<BufferStore>,
 134    ) -> Task<Result<Entity<Buffer>>> {
 135        let (tx, rx) = oneshot::channel();
 136        self.remote_buffer_listeners.entry(id).or_default().push(tx);
 137
 138        cx.spawn(|this, cx| async move {
 139            if let Some(buffer) = this
 140                .read_with(&cx, |buffer_store, _| buffer_store.get(id))
 141                .ok()
 142                .flatten()
 143            {
 144                return Ok(buffer);
 145            }
 146
 147            cx.background_executor()
 148                .spawn(async move { rx.await? })
 149                .await
 150        })
 151    }
 152
 153    fn save_remote_buffer(
 154        &self,
 155        buffer_handle: Entity<Buffer>,
 156        new_path: Option<proto::ProjectPath>,
 157        cx: &Context<BufferStore>,
 158    ) -> Task<Result<()>> {
 159        let buffer = buffer_handle.read(cx);
 160        let buffer_id = buffer.remote_id().into();
 161        let version = buffer.version();
 162        let rpc = self.upstream_client.clone();
 163        let project_id = self.project_id;
 164        cx.spawn(move |_, mut cx| async move {
 165            let response = rpc
 166                .request(proto::SaveBuffer {
 167                    project_id,
 168                    buffer_id,
 169                    new_path,
 170                    version: serialize_version(&version),
 171                })
 172                .await?;
 173            let version = deserialize_version(&response.version);
 174            let mtime = response.mtime.map(|mtime| mtime.into());
 175
 176            buffer_handle.update(&mut cx, |buffer, cx| {
 177                buffer.did_save(version.clone(), mtime, cx);
 178            })?;
 179
 180            Ok(())
 181        })
 182    }
 183
 184    pub fn handle_create_buffer_for_peer(
 185        &mut self,
 186        envelope: TypedEnvelope<proto::CreateBufferForPeer>,
 187        replica_id: u16,
 188        capability: Capability,
 189        cx: &mut Context<BufferStore>,
 190    ) -> Result<Option<Entity<Buffer>>> {
 191        match envelope
 192            .payload
 193            .variant
 194            .ok_or_else(|| anyhow!("missing variant"))?
 195        {
 196            proto::create_buffer_for_peer::Variant::State(mut state) => {
 197                let buffer_id = BufferId::new(state.id)?;
 198
 199                let buffer_result = maybe!({
 200                    let mut buffer_file = None;
 201                    if let Some(file) = state.file.take() {
 202                        let worktree_id = worktree::WorktreeId::from_proto(file.worktree_id);
 203                        let worktree = self
 204                            .worktree_store
 205                            .read(cx)
 206                            .worktree_for_id(worktree_id, cx)
 207                            .ok_or_else(|| {
 208                                anyhow!("no worktree found for id {}", file.worktree_id)
 209                            })?;
 210                        buffer_file = Some(Arc::new(File::from_proto(file, worktree.clone(), cx)?)
 211                            as Arc<dyn language::File>);
 212                    }
 213                    Buffer::from_proto(replica_id, capability, state, buffer_file)
 214                });
 215
 216                match buffer_result {
 217                    Ok(buffer) => {
 218                        let buffer = cx.new(|_| buffer);
 219                        self.loading_remote_buffers_by_id.insert(buffer_id, buffer);
 220                    }
 221                    Err(error) => {
 222                        if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {
 223                            for listener in listeners {
 224                                listener.send(Err(anyhow!(error.cloned()))).ok();
 225                            }
 226                        }
 227                    }
 228                }
 229            }
 230            proto::create_buffer_for_peer::Variant::Chunk(chunk) => {
 231                let buffer_id = BufferId::new(chunk.buffer_id)?;
 232                let buffer = self
 233                    .loading_remote_buffers_by_id
 234                    .get(&buffer_id)
 235                    .cloned()
 236                    .ok_or_else(|| {
 237                        anyhow!(
 238                            "received chunk for buffer {} without initial state",
 239                            chunk.buffer_id
 240                        )
 241                    })?;
 242
 243                let result = maybe!({
 244                    let operations = chunk
 245                        .operations
 246                        .into_iter()
 247                        .map(language::proto::deserialize_operation)
 248                        .collect::<Result<Vec<_>>>()?;
 249                    buffer.update(cx, |buffer, cx| buffer.apply_ops(operations, cx));
 250                    anyhow::Ok(())
 251                });
 252
 253                if let Err(error) = result {
 254                    self.loading_remote_buffers_by_id.remove(&buffer_id);
 255                    if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {
 256                        for listener in listeners {
 257                            listener.send(Err(error.cloned())).ok();
 258                        }
 259                    }
 260                } else if chunk.is_last {
 261                    self.loading_remote_buffers_by_id.remove(&buffer_id);
 262                    if self.upstream_client.is_via_collab() {
 263                        // retain buffers sent by peers to avoid races.
 264                        self.shared_with_me.insert(buffer.clone());
 265                    }
 266
 267                    if let Some(senders) = self.remote_buffer_listeners.remove(&buffer_id) {
 268                        for sender in senders {
 269                            sender.send(Ok(buffer.clone())).ok();
 270                        }
 271                    }
 272                    return Ok(Some(buffer));
 273                }
 274            }
 275        }
 276        return Ok(None);
 277    }
 278
 279    pub fn incomplete_buffer_ids(&self) -> Vec<BufferId> {
 280        self.loading_remote_buffers_by_id
 281            .keys()
 282            .copied()
 283            .collect::<Vec<_>>()
 284    }
 285
 286    pub fn deserialize_project_transaction(
 287        &self,
 288        message: proto::ProjectTransaction,
 289        push_to_history: bool,
 290        cx: &mut Context<BufferStore>,
 291    ) -> Task<Result<ProjectTransaction>> {
 292        cx.spawn(|this, mut cx| async move {
 293            let mut project_transaction = ProjectTransaction::default();
 294            for (buffer_id, transaction) in message.buffer_ids.into_iter().zip(message.transactions)
 295            {
 296                let buffer_id = BufferId::new(buffer_id)?;
 297                let buffer = this
 298                    .update(&mut cx, |this, cx| {
 299                        this.wait_for_remote_buffer(buffer_id, cx)
 300                    })?
 301                    .await?;
 302                let transaction = language::proto::deserialize_transaction(transaction)?;
 303                project_transaction.0.insert(buffer, transaction);
 304            }
 305
 306            for (buffer, transaction) in &project_transaction.0 {
 307                buffer
 308                    .update(&mut cx, |buffer, _| {
 309                        buffer.wait_for_edits(transaction.edit_ids.iter().copied())
 310                    })?
 311                    .await?;
 312
 313                if push_to_history {
 314                    buffer.update(&mut cx, |buffer, _| {
 315                        buffer.push_transaction(transaction.clone(), Instant::now());
 316                    })?;
 317                }
 318            }
 319
 320            Ok(project_transaction)
 321        })
 322    }
 323
 324    fn open_buffer(
 325        &self,
 326        path: Arc<Path>,
 327        worktree: Entity<Worktree>,
 328        cx: &mut Context<BufferStore>,
 329    ) -> Task<Result<Entity<Buffer>>> {
 330        let worktree_id = worktree.read(cx).id().to_proto();
 331        let project_id = self.project_id;
 332        let client = self.upstream_client.clone();
 333        let path_string = path.clone().to_string_lossy().to_string();
 334        cx.spawn(move |this, mut cx| async move {
 335            let response = client
 336                .request(proto::OpenBufferByPath {
 337                    project_id,
 338                    worktree_id,
 339                    path: path_string,
 340                })
 341                .await?;
 342            let buffer_id = BufferId::new(response.buffer_id)?;
 343
 344            let buffer = this
 345                .update(&mut cx, {
 346                    |this, cx| this.wait_for_remote_buffer(buffer_id, cx)
 347                })?
 348                .await?;
 349
 350            Ok(buffer)
 351        })
 352    }
 353
 354    fn create_buffer(&self, cx: &mut Context<BufferStore>) -> Task<Result<Entity<Buffer>>> {
 355        let create = self.upstream_client.request(proto::OpenNewBuffer {
 356            project_id: self.project_id,
 357        });
 358        cx.spawn(|this, mut cx| async move {
 359            let response = create.await?;
 360            let buffer_id = BufferId::new(response.buffer_id)?;
 361
 362            this.update(&mut cx, |this, cx| {
 363                this.wait_for_remote_buffer(buffer_id, cx)
 364            })?
 365            .await
 366        })
 367    }
 368
 369    fn reload_buffers(
 370        &self,
 371        buffers: HashSet<Entity<Buffer>>,
 372        push_to_history: bool,
 373        cx: &mut Context<BufferStore>,
 374    ) -> Task<Result<ProjectTransaction>> {
 375        let request = self.upstream_client.request(proto::ReloadBuffers {
 376            project_id: self.project_id,
 377            buffer_ids: buffers
 378                .iter()
 379                .map(|buffer| buffer.read(cx).remote_id().to_proto())
 380                .collect(),
 381        });
 382
 383        cx.spawn(|this, mut cx| async move {
 384            let response = request
 385                .await?
 386                .transaction
 387                .ok_or_else(|| anyhow!("missing transaction"))?;
 388            this.update(&mut cx, |this, cx| {
 389                this.deserialize_project_transaction(response, push_to_history, cx)
 390            })?
 391            .await
 392        })
 393    }
 394}
 395
 396impl LocalBufferStore {
 397    fn load_staged_text(&self, buffer: &Entity<Buffer>, cx: &App) -> Task<Result<Option<String>>> {
 398        let Some(file) = buffer.read(cx).file() else {
 399            return Task::ready(Ok(None));
 400        };
 401        let worktree_id = file.worktree_id(cx);
 402        let path = file.path().clone();
 403        let Some(worktree) = self
 404            .worktree_store
 405            .read(cx)
 406            .worktree_for_id(worktree_id, cx)
 407        else {
 408            return Task::ready(Err(anyhow!("no such worktree")));
 409        };
 410
 411        worktree.read(cx).load_staged_file(path.as_ref(), cx)
 412    }
 413
 414    fn save_local_buffer(
 415        &self,
 416        buffer_handle: Entity<Buffer>,
 417        worktree: Entity<Worktree>,
 418        path: Arc<Path>,
 419        mut has_changed_file: bool,
 420        cx: &mut Context<BufferStore>,
 421    ) -> Task<Result<()>> {
 422        let buffer = buffer_handle.read(cx);
 423
 424        let text = buffer.as_rope().clone();
 425        let line_ending = buffer.line_ending();
 426        let version = buffer.version();
 427        let buffer_id = buffer.remote_id();
 428        if buffer
 429            .file()
 430            .is_some_and(|file| file.disk_state() == DiskState::New)
 431        {
 432            has_changed_file = true;
 433        }
 434
 435        let save = worktree.update(cx, |worktree, cx| {
 436            worktree.write_file(path.as_ref(), text, line_ending, cx)
 437        });
 438
 439        cx.spawn(move |this, mut cx| async move {
 440            let new_file = save.await?;
 441            let mtime = new_file.disk_state().mtime();
 442            this.update(&mut cx, |this, cx| {
 443                if let Some((downstream_client, project_id)) = this.downstream_client.clone() {
 444                    if has_changed_file {
 445                        downstream_client
 446                            .send(proto::UpdateBufferFile {
 447                                project_id,
 448                                buffer_id: buffer_id.to_proto(),
 449                                file: Some(language::File::to_proto(&*new_file, cx)),
 450                            })
 451                            .log_err();
 452                    }
 453                    downstream_client
 454                        .send(proto::BufferSaved {
 455                            project_id,
 456                            buffer_id: buffer_id.to_proto(),
 457                            version: serialize_version(&version),
 458                            mtime: mtime.map(|time| time.into()),
 459                        })
 460                        .log_err();
 461                }
 462            })?;
 463            buffer_handle.update(&mut cx, |buffer, cx| {
 464                if has_changed_file {
 465                    buffer.file_updated(new_file, cx);
 466                }
 467                buffer.did_save(version.clone(), mtime, cx);
 468            })
 469        })
 470    }
 471
 472    fn subscribe_to_worktree(
 473        &mut self,
 474        worktree: &Entity<Worktree>,
 475        cx: &mut Context<BufferStore>,
 476    ) {
 477        cx.subscribe(worktree, |this, worktree, event, cx| {
 478            if worktree.read(cx).is_local() {
 479                match event {
 480                    worktree::Event::UpdatedEntries(changes) => {
 481                        Self::local_worktree_entries_changed(this, &worktree, changes, cx);
 482                    }
 483                    worktree::Event::UpdatedGitRepositories(updated_repos) => {
 484                        Self::local_worktree_git_repos_changed(
 485                            this,
 486                            worktree.clone(),
 487                            updated_repos,
 488                            cx,
 489                        )
 490                    }
 491                    _ => {}
 492                }
 493            }
 494        })
 495        .detach();
 496    }
 497
 498    fn local_worktree_entries_changed(
 499        this: &mut BufferStore,
 500        worktree_handle: &Entity<Worktree>,
 501        changes: &[(Arc<Path>, ProjectEntryId, PathChange)],
 502        cx: &mut Context<BufferStore>,
 503    ) {
 504        let snapshot = worktree_handle.read(cx).snapshot();
 505        for (path, entry_id, _) in changes {
 506            Self::local_worktree_entry_changed(
 507                this,
 508                *entry_id,
 509                path,
 510                worktree_handle,
 511                &snapshot,
 512                cx,
 513            );
 514        }
 515    }
 516
 517    fn local_worktree_git_repos_changed(
 518        this: &mut BufferStore,
 519        worktree_handle: Entity<Worktree>,
 520        changed_repos: &UpdatedGitRepositoriesSet,
 521        cx: &mut Context<BufferStore>,
 522    ) {
 523        debug_assert!(worktree_handle.read(cx).is_local());
 524
 525        let buffer_change_sets = this
 526            .opened_buffers
 527            .values()
 528            .filter_map(|buffer| {
 529                if let OpenBuffer::Complete {
 530                    buffer,
 531                    unstaged_changes,
 532                } = buffer
 533                {
 534                    let buffer = buffer.upgrade()?.read(cx);
 535                    let file = File::from_dyn(buffer.file())?;
 536                    if file.worktree != worktree_handle {
 537                        return None;
 538                    }
 539                    changed_repos
 540                        .iter()
 541                        .find(|(work_dir, _)| file.path.starts_with(work_dir))?;
 542                    let unstaged_changes = unstaged_changes.as_ref()?.upgrade()?;
 543                    let snapshot = buffer.text_snapshot();
 544                    Some((unstaged_changes, snapshot, file.path.clone()))
 545                } else {
 546                    None
 547                }
 548            })
 549            .collect::<Vec<_>>();
 550
 551        if buffer_change_sets.is_empty() {
 552            return;
 553        }
 554
 555        cx.spawn(move |this, mut cx| async move {
 556            let snapshot =
 557                worktree_handle.update(&mut cx, |tree, _| tree.as_local().unwrap().snapshot())?;
 558            let diff_bases_by_buffer = cx
 559                .background_executor()
 560                .spawn(async move {
 561                    buffer_change_sets
 562                        .into_iter()
 563                        .filter_map(|(change_set, buffer_snapshot, path)| {
 564                            let local_repo = snapshot.local_repo_for_path(&path)?;
 565                            let relative_path = local_repo.relativize(&path).ok()?;
 566                            let base_text = local_repo.repo().load_index_text(&relative_path);
 567                            Some((change_set, buffer_snapshot, base_text))
 568                        })
 569                        .collect::<Vec<_>>()
 570                })
 571                .await;
 572
 573            this.update(&mut cx, |this, cx| {
 574                for (change_set, buffer_snapshot, staged_text) in diff_bases_by_buffer {
 575                    change_set.update(cx, |change_set, cx| {
 576                        if let Some(staged_text) = staged_text.clone() {
 577                            let _ =
 578                                change_set.set_base_text(staged_text, buffer_snapshot.clone(), cx);
 579                        } else {
 580                            change_set.unset_base_text(buffer_snapshot.clone(), cx);
 581                        }
 582                    });
 583
 584                    if let Some((client, project_id)) = &this.downstream_client.clone() {
 585                        client
 586                            .send(proto::UpdateDiffBase {
 587                                project_id: *project_id,
 588                                buffer_id: buffer_snapshot.remote_id().to_proto(),
 589                                staged_text,
 590                            })
 591                            .log_err();
 592                    }
 593                }
 594            })
 595        })
 596        .detach_and_log_err(cx);
 597    }
 598
 599    fn local_worktree_entry_changed(
 600        this: &mut BufferStore,
 601        entry_id: ProjectEntryId,
 602        path: &Arc<Path>,
 603        worktree: &Entity<worktree::Worktree>,
 604        snapshot: &worktree::Snapshot,
 605        cx: &mut Context<BufferStore>,
 606    ) -> Option<()> {
 607        let project_path = ProjectPath {
 608            worktree_id: snapshot.id(),
 609            path: path.clone(),
 610        };
 611
 612        let buffer_id = {
 613            let local = this.as_local_mut()?;
 614            match local.local_buffer_ids_by_entry_id.get(&entry_id) {
 615                Some(&buffer_id) => buffer_id,
 616                None => local.local_buffer_ids_by_path.get(&project_path).copied()?,
 617            }
 618        };
 619
 620        let buffer = if let Some(buffer) = this.get(buffer_id) {
 621            Some(buffer)
 622        } else {
 623            this.opened_buffers.remove(&buffer_id);
 624            None
 625        };
 626
 627        let buffer = if let Some(buffer) = buffer {
 628            buffer
 629        } else {
 630            let this = this.as_local_mut()?;
 631            this.local_buffer_ids_by_path.remove(&project_path);
 632            this.local_buffer_ids_by_entry_id.remove(&entry_id);
 633            return None;
 634        };
 635
 636        let events = buffer.update(cx, |buffer, cx| {
 637            let local = this.as_local_mut()?;
 638            let file = buffer.file()?;
 639            let old_file = File::from_dyn(Some(file))?;
 640            if old_file.worktree != *worktree {
 641                return None;
 642            }
 643
 644            let snapshot_entry = old_file
 645                .entry_id
 646                .and_then(|entry_id| snapshot.entry_for_id(entry_id))
 647                .or_else(|| snapshot.entry_for_path(old_file.path.as_ref()));
 648
 649            let new_file = if let Some(entry) = snapshot_entry {
 650                File {
 651                    disk_state: match entry.mtime {
 652                        Some(mtime) => DiskState::Present { mtime },
 653                        None => old_file.disk_state,
 654                    },
 655                    is_local: true,
 656                    entry_id: Some(entry.id),
 657                    path: entry.path.clone(),
 658                    worktree: worktree.clone(),
 659                    is_private: entry.is_private,
 660                }
 661            } else {
 662                File {
 663                    disk_state: DiskState::Deleted,
 664                    is_local: true,
 665                    entry_id: old_file.entry_id,
 666                    path: old_file.path.clone(),
 667                    worktree: worktree.clone(),
 668                    is_private: old_file.is_private,
 669                }
 670            };
 671
 672            if new_file == *old_file {
 673                return None;
 674            }
 675
 676            let mut events = Vec::new();
 677            if new_file.path != old_file.path {
 678                local.local_buffer_ids_by_path.remove(&ProjectPath {
 679                    path: old_file.path.clone(),
 680                    worktree_id: old_file.worktree_id(cx),
 681                });
 682                local.local_buffer_ids_by_path.insert(
 683                    ProjectPath {
 684                        worktree_id: new_file.worktree_id(cx),
 685                        path: new_file.path.clone(),
 686                    },
 687                    buffer_id,
 688                );
 689                events.push(BufferStoreEvent::BufferChangedFilePath {
 690                    buffer: cx.entity(),
 691                    old_file: buffer.file().cloned(),
 692                });
 693            }
 694
 695            if new_file.entry_id != old_file.entry_id {
 696                if let Some(entry_id) = old_file.entry_id {
 697                    local.local_buffer_ids_by_entry_id.remove(&entry_id);
 698                }
 699                if let Some(entry_id) = new_file.entry_id {
 700                    local
 701                        .local_buffer_ids_by_entry_id
 702                        .insert(entry_id, buffer_id);
 703                }
 704            }
 705
 706            if let Some((client, project_id)) = &this.downstream_client {
 707                client
 708                    .send(proto::UpdateBufferFile {
 709                        project_id: *project_id,
 710                        buffer_id: buffer_id.to_proto(),
 711                        file: Some(new_file.to_proto(cx)),
 712                    })
 713                    .ok();
 714            }
 715
 716            buffer.file_updated(Arc::new(new_file), cx);
 717            Some(events)
 718        })?;
 719
 720        for event in events {
 721            cx.emit(event);
 722        }
 723
 724        None
 725    }
 726
 727    fn buffer_changed_file(&mut self, buffer: Entity<Buffer>, cx: &mut App) -> Option<()> {
 728        let file = File::from_dyn(buffer.read(cx).file())?;
 729
 730        let remote_id = buffer.read(cx).remote_id();
 731        if let Some(entry_id) = file.entry_id {
 732            match self.local_buffer_ids_by_entry_id.get(&entry_id) {
 733                Some(_) => {
 734                    return None;
 735                }
 736                None => {
 737                    self.local_buffer_ids_by_entry_id
 738                        .insert(entry_id, remote_id);
 739                }
 740            }
 741        };
 742        self.local_buffer_ids_by_path.insert(
 743            ProjectPath {
 744                worktree_id: file.worktree_id(cx),
 745                path: file.path.clone(),
 746            },
 747            remote_id,
 748        );
 749
 750        Some(())
 751    }
 752
 753    fn save_buffer(
 754        &self,
 755        buffer: Entity<Buffer>,
 756        cx: &mut Context<BufferStore>,
 757    ) -> Task<Result<()>> {
 758        let Some(file) = File::from_dyn(buffer.read(cx).file()) else {
 759            return Task::ready(Err(anyhow!("buffer doesn't have a file")));
 760        };
 761        let worktree = file.worktree.clone();
 762        self.save_local_buffer(buffer, worktree, file.path.clone(), false, cx)
 763    }
 764
 765    fn save_buffer_as(
 766        &self,
 767        buffer: Entity<Buffer>,
 768        path: ProjectPath,
 769        cx: &mut Context<BufferStore>,
 770    ) -> Task<Result<()>> {
 771        let Some(worktree) = self
 772            .worktree_store
 773            .read(cx)
 774            .worktree_for_id(path.worktree_id, cx)
 775        else {
 776            return Task::ready(Err(anyhow!("no such worktree")));
 777        };
 778        self.save_local_buffer(buffer, worktree, path.path.clone(), true, cx)
 779    }
 780
 781    fn open_buffer(
 782        &self,
 783        path: Arc<Path>,
 784        worktree: Entity<Worktree>,
 785        cx: &mut Context<BufferStore>,
 786    ) -> Task<Result<Entity<Buffer>>> {
 787        let load_buffer = worktree.update(cx, |worktree, cx| {
 788            let load_file = worktree.load_file(path.as_ref(), cx);
 789            let reservation = cx.reserve_entity();
 790            let buffer_id = BufferId::from(reservation.entity_id().as_non_zero_u64());
 791            cx.spawn(move |_, mut cx| async move {
 792                let loaded = load_file.await?;
 793                let text_buffer = cx
 794                    .background_executor()
 795                    .spawn(async move { text::Buffer::new(0, buffer_id, loaded.text) })
 796                    .await;
 797                cx.insert_entity(reservation, |_| {
 798                    Buffer::build(text_buffer, Some(loaded.file), Capability::ReadWrite)
 799                })
 800            })
 801        });
 802
 803        cx.spawn(move |this, mut cx| async move {
 804            let buffer = match load_buffer.await {
 805                Ok(buffer) => Ok(buffer),
 806                Err(error) if is_not_found_error(&error) => cx.new(|cx| {
 807                    let buffer_id = BufferId::from(cx.entity_id().as_non_zero_u64());
 808                    let text_buffer = text::Buffer::new(0, buffer_id, "".into());
 809                    Buffer::build(
 810                        text_buffer,
 811                        Some(Arc::new(File {
 812                            worktree,
 813                            path,
 814                            disk_state: DiskState::New,
 815                            entry_id: None,
 816                            is_local: true,
 817                            is_private: false,
 818                        })),
 819                        Capability::ReadWrite,
 820                    )
 821                }),
 822                Err(e) => Err(e),
 823            }?;
 824            this.update(&mut cx, |this, cx| {
 825                this.add_buffer(buffer.clone(), cx)?;
 826                let buffer_id = buffer.read(cx).remote_id();
 827                if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
 828                    let this = this.as_local_mut().unwrap();
 829                    this.local_buffer_ids_by_path.insert(
 830                        ProjectPath {
 831                            worktree_id: file.worktree_id(cx),
 832                            path: file.path.clone(),
 833                        },
 834                        buffer_id,
 835                    );
 836
 837                    if let Some(entry_id) = file.entry_id {
 838                        this.local_buffer_ids_by_entry_id
 839                            .insert(entry_id, buffer_id);
 840                    }
 841                }
 842
 843                anyhow::Ok(())
 844            })??;
 845
 846            Ok(buffer)
 847        })
 848    }
 849
 850    fn create_buffer(&self, cx: &mut Context<BufferStore>) -> Task<Result<Entity<Buffer>>> {
 851        cx.spawn(|buffer_store, mut cx| async move {
 852            let buffer =
 853                cx.new(|cx| Buffer::local("", cx).with_language(language::PLAIN_TEXT.clone(), cx))?;
 854            buffer_store.update(&mut cx, |buffer_store, cx| {
 855                buffer_store.add_buffer(buffer.clone(), cx).log_err();
 856            })?;
 857            Ok(buffer)
 858        })
 859    }
 860
 861    fn reload_buffers(
 862        &self,
 863        buffers: HashSet<Entity<Buffer>>,
 864        push_to_history: bool,
 865        cx: &mut Context<BufferStore>,
 866    ) -> Task<Result<ProjectTransaction>> {
 867        cx.spawn(move |_, mut cx| async move {
 868            let mut project_transaction = ProjectTransaction::default();
 869            for buffer in buffers {
 870                let transaction = buffer
 871                    .update(&mut cx, |buffer, cx| buffer.reload(cx))?
 872                    .await?;
 873                buffer.update(&mut cx, |buffer, cx| {
 874                    if let Some(transaction) = transaction {
 875                        if !push_to_history {
 876                            buffer.forget_transaction(transaction.id);
 877                        }
 878                        project_transaction.0.insert(cx.entity(), transaction);
 879                    }
 880                })?;
 881            }
 882
 883            Ok(project_transaction)
 884        })
 885    }
 886}
 887
 888impl BufferStore {
 889    pub fn init(client: &AnyProtoClient) {
 890        client.add_model_message_handler(Self::handle_buffer_reloaded);
 891        client.add_model_message_handler(Self::handle_buffer_saved);
 892        client.add_model_message_handler(Self::handle_update_buffer_file);
 893        client.add_model_request_handler(Self::handle_save_buffer);
 894        client.add_model_request_handler(Self::handle_blame_buffer);
 895        client.add_model_request_handler(Self::handle_reload_buffers);
 896        client.add_model_request_handler(Self::handle_get_permalink_to_line);
 897        client.add_model_request_handler(Self::handle_get_staged_text);
 898        client.add_model_message_handler(Self::handle_update_diff_base);
 899    }
 900
 901    /// Creates a buffer store, optionally retaining its buffers.
 902    pub fn local(worktree_store: Entity<WorktreeStore>, cx: &mut Context<Self>) -> Self {
 903        Self {
 904            state: BufferStoreState::Local(LocalBufferStore {
 905                local_buffer_ids_by_path: Default::default(),
 906                local_buffer_ids_by_entry_id: Default::default(),
 907                worktree_store: worktree_store.clone(),
 908                _subscription: cx.subscribe(&worktree_store, |this, _, event, cx| {
 909                    if let WorktreeStoreEvent::WorktreeAdded(worktree) = event {
 910                        let this = this.as_local_mut().unwrap();
 911                        this.subscribe_to_worktree(worktree, cx);
 912                    }
 913                }),
 914            }),
 915            downstream_client: None,
 916            opened_buffers: Default::default(),
 917            shared_buffers: Default::default(),
 918            loading_buffers: Default::default(),
 919            loading_change_sets: Default::default(),
 920            worktree_store,
 921        }
 922    }
 923
 924    pub fn remote(
 925        worktree_store: Entity<WorktreeStore>,
 926        upstream_client: AnyProtoClient,
 927        remote_id: u64,
 928        _cx: &mut Context<Self>,
 929    ) -> Self {
 930        Self {
 931            state: BufferStoreState::Remote(RemoteBufferStore {
 932                shared_with_me: Default::default(),
 933                loading_remote_buffers_by_id: Default::default(),
 934                remote_buffer_listeners: Default::default(),
 935                project_id: remote_id,
 936                upstream_client,
 937                worktree_store: worktree_store.clone(),
 938            }),
 939            downstream_client: None,
 940            opened_buffers: Default::default(),
 941            loading_buffers: Default::default(),
 942            loading_change_sets: Default::default(),
 943            shared_buffers: Default::default(),
 944            worktree_store,
 945        }
 946    }
 947
 948    fn as_local_mut(&mut self) -> Option<&mut LocalBufferStore> {
 949        match &mut self.state {
 950            BufferStoreState::Local(state) => Some(state),
 951            _ => None,
 952        }
 953    }
 954
 955    fn as_remote_mut(&mut self) -> Option<&mut RemoteBufferStore> {
 956        match &mut self.state {
 957            BufferStoreState::Remote(state) => Some(state),
 958            _ => None,
 959        }
 960    }
 961
 962    fn as_remote(&self) -> Option<&RemoteBufferStore> {
 963        match &self.state {
 964            BufferStoreState::Remote(state) => Some(state),
 965            _ => None,
 966        }
 967    }
 968
 969    pub fn open_buffer(
 970        &mut self,
 971        project_path: ProjectPath,
 972        cx: &mut Context<Self>,
 973    ) -> Task<Result<Entity<Buffer>>> {
 974        if let Some(buffer) = self.get_by_path(&project_path, cx) {
 975            return Task::ready(Ok(buffer));
 976        }
 977
 978        let task = match self.loading_buffers.entry(project_path.clone()) {
 979            hash_map::Entry::Occupied(e) => e.get().clone(),
 980            hash_map::Entry::Vacant(entry) => {
 981                let path = project_path.path.clone();
 982                let Some(worktree) = self
 983                    .worktree_store
 984                    .read(cx)
 985                    .worktree_for_id(project_path.worktree_id, cx)
 986                else {
 987                    return Task::ready(Err(anyhow!("no such worktree")));
 988                };
 989                let load_buffer = match &self.state {
 990                    BufferStoreState::Local(this) => this.open_buffer(path, worktree, cx),
 991                    BufferStoreState::Remote(this) => this.open_buffer(path, worktree, cx),
 992                };
 993
 994                entry
 995                    .insert(
 996                        cx.spawn(move |this, mut cx| async move {
 997                            let load_result = load_buffer.await;
 998                            this.update(&mut cx, |this, _cx| {
 999                                // Record the fact that the buffer is no longer loading.
1000                                this.loading_buffers.remove(&project_path);
1001                            })
1002                            .ok();
1003                            load_result.map_err(Arc::new)
1004                        })
1005                        .shared(),
1006                    )
1007                    .clone()
1008            }
1009        };
1010
1011        cx.background_executor()
1012            .spawn(async move { task.await.map_err(|e| anyhow!("{e}")) })
1013    }
1014
1015    pub fn open_unstaged_changes(
1016        &mut self,
1017        buffer: Entity<Buffer>,
1018        cx: &mut Context<Self>,
1019    ) -> Task<Result<Entity<BufferChangeSet>>> {
1020        let buffer_id = buffer.read(cx).remote_id();
1021        if let Some(change_set) = self.get_unstaged_changes(buffer_id) {
1022            return Task::ready(Ok(change_set));
1023        }
1024
1025        let task = match self.loading_change_sets.entry(buffer_id) {
1026            hash_map::Entry::Occupied(e) => e.get().clone(),
1027            hash_map::Entry::Vacant(entry) => {
1028                let load = match &self.state {
1029                    BufferStoreState::Local(this) => this.load_staged_text(&buffer, cx),
1030                    BufferStoreState::Remote(this) => this.load_staged_text(buffer_id, cx),
1031                };
1032
1033                entry
1034                    .insert(
1035                        cx.spawn(move |this, cx| async move {
1036                            Self::open_unstaged_changes_internal(this, load.await, buffer, cx)
1037                                .await
1038                                .map_err(Arc::new)
1039                        })
1040                        .shared(),
1041                    )
1042                    .clone()
1043            }
1044        };
1045
1046        cx.background_executor()
1047            .spawn(async move { task.await.map_err(|e| anyhow!("{e}")) })
1048    }
1049
1050    #[cfg(any(test, feature = "test-support"))]
1051    pub fn set_change_set(&mut self, buffer_id: BufferId, change_set: Entity<BufferChangeSet>) {
1052        self.loading_change_sets
1053            .insert(buffer_id, Task::ready(Ok(change_set)).shared());
1054    }
1055
1056    pub async fn open_unstaged_changes_internal(
1057        this: WeakEntity<Self>,
1058        text: Result<Option<String>>,
1059        buffer: Entity<Buffer>,
1060        mut cx: AsyncApp,
1061    ) -> Result<Entity<BufferChangeSet>> {
1062        let text = match text {
1063            Err(e) => {
1064                this.update(&mut cx, |this, cx| {
1065                    let buffer_id = buffer.read(cx).remote_id();
1066                    this.loading_change_sets.remove(&buffer_id);
1067                })?;
1068                return Err(e);
1069            }
1070            Ok(text) => text,
1071        };
1072
1073        let change_set = cx.new(|cx| BufferChangeSet::new(&buffer, cx)).unwrap();
1074
1075        if let Some(text) = text {
1076            change_set
1077                .update(&mut cx, |change_set, cx| {
1078                    let snapshot = buffer.read(cx).text_snapshot();
1079                    change_set.set_base_text(text, snapshot, cx)
1080                })?
1081                .await
1082                .ok();
1083        }
1084
1085        this.update(&mut cx, |this, cx| {
1086            let buffer_id = buffer.read(cx).remote_id();
1087            this.loading_change_sets.remove(&buffer_id);
1088            if let Some(OpenBuffer::Complete {
1089                unstaged_changes, ..
1090            }) = this.opened_buffers.get_mut(&buffer.read(cx).remote_id())
1091            {
1092                *unstaged_changes = Some(change_set.downgrade());
1093            }
1094        })?;
1095
1096        Ok(change_set)
1097    }
1098
1099    pub fn create_buffer(&mut self, cx: &mut Context<Self>) -> Task<Result<Entity<Buffer>>> {
1100        match &self.state {
1101            BufferStoreState::Local(this) => this.create_buffer(cx),
1102            BufferStoreState::Remote(this) => this.create_buffer(cx),
1103        }
1104    }
1105
1106    pub fn save_buffer(
1107        &mut self,
1108        buffer: Entity<Buffer>,
1109        cx: &mut Context<Self>,
1110    ) -> Task<Result<()>> {
1111        match &mut self.state {
1112            BufferStoreState::Local(this) => this.save_buffer(buffer, cx),
1113            BufferStoreState::Remote(this) => this.save_remote_buffer(buffer.clone(), None, cx),
1114        }
1115    }
1116
1117    pub fn save_buffer_as(
1118        &mut self,
1119        buffer: Entity<Buffer>,
1120        path: ProjectPath,
1121        cx: &mut Context<Self>,
1122    ) -> Task<Result<()>> {
1123        let old_file = buffer.read(cx).file().cloned();
1124        let task = match &self.state {
1125            BufferStoreState::Local(this) => this.save_buffer_as(buffer.clone(), path, cx),
1126            BufferStoreState::Remote(this) => {
1127                this.save_remote_buffer(buffer.clone(), Some(path.to_proto()), cx)
1128            }
1129        };
1130        cx.spawn(|this, mut cx| async move {
1131            task.await?;
1132            this.update(&mut cx, |_, cx| {
1133                cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });
1134            })
1135        })
1136    }
1137
1138    pub fn blame_buffer(
1139        &self,
1140        buffer: &Entity<Buffer>,
1141        version: Option<clock::Global>,
1142        cx: &App,
1143    ) -> Task<Result<Option<Blame>>> {
1144        let buffer = buffer.read(cx);
1145        let Some(file) = File::from_dyn(buffer.file()) else {
1146            return Task::ready(Err(anyhow!("buffer has no file")));
1147        };
1148
1149        match file.worktree.clone().read(cx) {
1150            Worktree::Local(worktree) => {
1151                let worktree = worktree.snapshot();
1152                let blame_params = maybe!({
1153                    let local_repo = match worktree.local_repo_for_path(&file.path) {
1154                        Some(repo_for_path) => repo_for_path,
1155                        None => return Ok(None),
1156                    };
1157
1158                    let relative_path = local_repo
1159                        .relativize(&file.path)
1160                        .context("failed to relativize buffer path")?;
1161
1162                    let repo = local_repo.repo().clone();
1163
1164                    let content = match version {
1165                        Some(version) => buffer.rope_for_version(&version).clone(),
1166                        None => buffer.as_rope().clone(),
1167                    };
1168
1169                    anyhow::Ok(Some((repo, relative_path, content)))
1170                });
1171
1172                cx.background_executor().spawn(async move {
1173                    let Some((repo, relative_path, content)) = blame_params? else {
1174                        return Ok(None);
1175                    };
1176                    repo.blame(&relative_path, content)
1177                        .with_context(|| format!("Failed to blame {:?}", relative_path.0))
1178                        .map(Some)
1179                })
1180            }
1181            Worktree::Remote(worktree) => {
1182                let buffer_id = buffer.remote_id();
1183                let version = buffer.version();
1184                let project_id = worktree.project_id();
1185                let client = worktree.client();
1186                cx.spawn(|_| async move {
1187                    let response = client
1188                        .request(proto::BlameBuffer {
1189                            project_id,
1190                            buffer_id: buffer_id.into(),
1191                            version: serialize_version(&version),
1192                        })
1193                        .await?;
1194                    Ok(deserialize_blame_buffer_response(response))
1195                })
1196            }
1197        }
1198    }
1199
1200    pub fn get_permalink_to_line(
1201        &self,
1202        buffer: &Entity<Buffer>,
1203        selection: Range<u32>,
1204        cx: &App,
1205    ) -> Task<Result<url::Url>> {
1206        let buffer = buffer.read(cx);
1207        let Some(file) = File::from_dyn(buffer.file()) else {
1208            return Task::ready(Err(anyhow!("buffer has no file")));
1209        };
1210
1211        match file.worktree.read(cx) {
1212            Worktree::Local(worktree) => {
1213                let worktree_path = worktree.abs_path().clone();
1214                let Some((repo_entry, repo)) =
1215                    worktree.repository_for_path(file.path()).and_then(|entry| {
1216                        let repo = worktree.get_local_repo(&entry)?.repo().clone();
1217                        Some((entry, repo))
1218                    })
1219                else {
1220                    // If we're not in a Git repo, check whether this is a Rust source
1221                    // file in the Cargo registry (presumably opened with go-to-definition
1222                    // from a normal Rust file). If so, we can put together a permalink
1223                    // using crate metadata.
1224                    if !buffer
1225                        .language()
1226                        .is_some_and(|lang| lang.name() == "Rust".into())
1227                    {
1228                        return Task::ready(Err(anyhow!("no permalink available")));
1229                    }
1230                    let file_path = worktree_path.join(file.path());
1231                    return cx.spawn(|cx| async move {
1232                        let provider_registry =
1233                            cx.update(GitHostingProviderRegistry::default_global)?;
1234                        get_permalink_in_rust_registry_src(provider_registry, file_path, selection)
1235                            .map_err(|_| anyhow!("no permalink available"))
1236                    });
1237                };
1238
1239                let path = match repo_entry.relativize(file.path()) {
1240                    Ok(RepoPath(path)) => path,
1241                    Err(e) => return Task::ready(Err(e)),
1242                };
1243
1244                cx.spawn(|cx| async move {
1245                    const REMOTE_NAME: &str = "origin";
1246                    let origin_url = repo
1247                        .remote_url(REMOTE_NAME)
1248                        .ok_or_else(|| anyhow!("remote \"{REMOTE_NAME}\" not found"))?;
1249
1250                    let sha = repo
1251                        .head_sha()
1252                        .ok_or_else(|| anyhow!("failed to read HEAD SHA"))?;
1253
1254                    let provider_registry =
1255                        cx.update(GitHostingProviderRegistry::default_global)?;
1256
1257                    let (provider, remote) =
1258                        parse_git_remote_url(provider_registry, &origin_url)
1259                            .ok_or_else(|| anyhow!("failed to parse Git remote URL"))?;
1260
1261                    let path = path
1262                        .to_str()
1263                        .ok_or_else(|| anyhow!("failed to convert path to string"))?;
1264
1265                    Ok(provider.build_permalink(
1266                        remote,
1267                        BuildPermalinkParams {
1268                            sha: &sha,
1269                            path,
1270                            selection: Some(selection),
1271                        },
1272                    ))
1273                })
1274            }
1275            Worktree::Remote(worktree) => {
1276                let buffer_id = buffer.remote_id();
1277                let project_id = worktree.project_id();
1278                let client = worktree.client();
1279                cx.spawn(|_| async move {
1280                    let response = client
1281                        .request(proto::GetPermalinkToLine {
1282                            project_id,
1283                            buffer_id: buffer_id.into(),
1284                            selection: Some(proto::Range {
1285                                start: selection.start as u64,
1286                                end: selection.end as u64,
1287                            }),
1288                        })
1289                        .await?;
1290
1291                    url::Url::parse(&response.permalink).context("failed to parse permalink")
1292                })
1293            }
1294        }
1295    }
1296
1297    fn add_buffer(&mut self, buffer: Entity<Buffer>, cx: &mut Context<Self>) -> Result<()> {
1298        let remote_id = buffer.read(cx).remote_id();
1299        let is_remote = buffer.read(cx).replica_id() != 0;
1300        let open_buffer = OpenBuffer::Complete {
1301            buffer: buffer.downgrade(),
1302            unstaged_changes: None,
1303        };
1304
1305        let handle = cx.entity().downgrade();
1306        buffer.update(cx, move |_, cx| {
1307            cx.on_release(move |buffer, cx| {
1308                handle
1309                    .update(cx, |_, cx| {
1310                        cx.emit(BufferStoreEvent::BufferDropped(buffer.remote_id()))
1311                    })
1312                    .ok();
1313            })
1314            .detach()
1315        });
1316
1317        match self.opened_buffers.entry(remote_id) {
1318            hash_map::Entry::Vacant(entry) => {
1319                entry.insert(open_buffer);
1320            }
1321            hash_map::Entry::Occupied(mut entry) => {
1322                if let OpenBuffer::Operations(operations) = entry.get_mut() {
1323                    buffer.update(cx, |b, cx| b.apply_ops(operations.drain(..), cx));
1324                } else if entry.get().upgrade().is_some() {
1325                    if is_remote {
1326                        return Ok(());
1327                    } else {
1328                        debug_panic!("buffer {} was already registered", remote_id);
1329                        Err(anyhow!("buffer {} was already registered", remote_id))?;
1330                    }
1331                }
1332                entry.insert(open_buffer);
1333            }
1334        }
1335
1336        cx.subscribe(&buffer, Self::on_buffer_event).detach();
1337        cx.emit(BufferStoreEvent::BufferAdded(buffer));
1338        Ok(())
1339    }
1340
1341    pub fn buffers(&self) -> impl '_ + Iterator<Item = Entity<Buffer>> {
1342        self.opened_buffers
1343            .values()
1344            .filter_map(|buffer| buffer.upgrade())
1345    }
1346
1347    pub fn loading_buffers(
1348        &self,
1349    ) -> impl Iterator<Item = (&ProjectPath, impl Future<Output = Result<Entity<Buffer>>>)> {
1350        self.loading_buffers.iter().map(|(path, task)| {
1351            let task = task.clone();
1352            (path, async move { task.await.map_err(|e| anyhow!("{e}")) })
1353        })
1354    }
1355
1356    pub fn get_by_path(&self, path: &ProjectPath, cx: &App) -> Option<Entity<Buffer>> {
1357        self.buffers().find_map(|buffer| {
1358            let file = File::from_dyn(buffer.read(cx).file())?;
1359            if file.worktree_id(cx) == path.worktree_id && file.path == path.path {
1360                Some(buffer)
1361            } else {
1362                None
1363            }
1364        })
1365    }
1366
1367    pub fn get(&self, buffer_id: BufferId) -> Option<Entity<Buffer>> {
1368        self.opened_buffers.get(&buffer_id)?.upgrade()
1369    }
1370
1371    pub fn get_existing(&self, buffer_id: BufferId) -> Result<Entity<Buffer>> {
1372        self.get(buffer_id)
1373            .ok_or_else(|| anyhow!("unknown buffer id {}", buffer_id))
1374    }
1375
1376    pub fn get_possibly_incomplete(&self, buffer_id: BufferId) -> Option<Entity<Buffer>> {
1377        self.get(buffer_id).or_else(|| {
1378            self.as_remote()
1379                .and_then(|remote| remote.loading_remote_buffers_by_id.get(&buffer_id).cloned())
1380        })
1381    }
1382
1383    pub fn get_unstaged_changes(&self, buffer_id: BufferId) -> Option<Entity<BufferChangeSet>> {
1384        if let OpenBuffer::Complete {
1385            unstaged_changes, ..
1386        } = self.opened_buffers.get(&buffer_id)?
1387        {
1388            unstaged_changes.as_ref()?.upgrade()
1389        } else {
1390            None
1391        }
1392    }
1393
1394    pub fn buffer_version_info(&self, cx: &App) -> (Vec<proto::BufferVersion>, Vec<BufferId>) {
1395        let buffers = self
1396            .buffers()
1397            .map(|buffer| {
1398                let buffer = buffer.read(cx);
1399                proto::BufferVersion {
1400                    id: buffer.remote_id().into(),
1401                    version: language::proto::serialize_version(&buffer.version),
1402                }
1403            })
1404            .collect();
1405        let incomplete_buffer_ids = self
1406            .as_remote()
1407            .map(|remote| remote.incomplete_buffer_ids())
1408            .unwrap_or_default();
1409        (buffers, incomplete_buffer_ids)
1410    }
1411
1412    pub fn disconnected_from_host(&mut self, cx: &mut App) {
1413        for open_buffer in self.opened_buffers.values_mut() {
1414            if let Some(buffer) = open_buffer.upgrade() {
1415                buffer.update(cx, |buffer, _| buffer.give_up_waiting());
1416            }
1417        }
1418
1419        for buffer in self.buffers() {
1420            buffer.update(cx, |buffer, cx| {
1421                buffer.set_capability(Capability::ReadOnly, cx)
1422            });
1423        }
1424
1425        if let Some(remote) = self.as_remote_mut() {
1426            // Wake up all futures currently waiting on a buffer to get opened,
1427            // to give them a chance to fail now that we've disconnected.
1428            remote.remote_buffer_listeners.clear()
1429        }
1430    }
1431
1432    pub fn shared(&mut self, remote_id: u64, downstream_client: AnyProtoClient, _cx: &mut App) {
1433        self.downstream_client = Some((downstream_client, remote_id));
1434    }
1435
1436    pub fn unshared(&mut self, _cx: &mut Context<Self>) {
1437        self.downstream_client.take();
1438        self.forget_shared_buffers();
1439    }
1440
1441    pub fn discard_incomplete(&mut self) {
1442        self.opened_buffers
1443            .retain(|_, buffer| !matches!(buffer, OpenBuffer::Operations(_)));
1444    }
1445
1446    pub fn find_search_candidates(
1447        &mut self,
1448        query: &SearchQuery,
1449        mut limit: usize,
1450        fs: Arc<dyn Fs>,
1451        cx: &mut Context<Self>,
1452    ) -> Receiver<Entity<Buffer>> {
1453        let (tx, rx) = smol::channel::unbounded();
1454        let mut open_buffers = HashSet::default();
1455        let mut unnamed_buffers = Vec::new();
1456        for handle in self.buffers() {
1457            let buffer = handle.read(cx);
1458            if let Some(entry_id) = buffer.entry_id(cx) {
1459                open_buffers.insert(entry_id);
1460            } else {
1461                limit = limit.saturating_sub(1);
1462                unnamed_buffers.push(handle)
1463            };
1464        }
1465
1466        const MAX_CONCURRENT_BUFFER_OPENS: usize = 64;
1467        let project_paths_rx = self
1468            .worktree_store
1469            .update(cx, |worktree_store, cx| {
1470                worktree_store.find_search_candidates(query.clone(), limit, open_buffers, fs, cx)
1471            })
1472            .chunks(MAX_CONCURRENT_BUFFER_OPENS);
1473
1474        cx.spawn(|this, mut cx| async move {
1475            for buffer in unnamed_buffers {
1476                tx.send(buffer).await.ok();
1477            }
1478
1479            let mut project_paths_rx = pin!(project_paths_rx);
1480            while let Some(project_paths) = project_paths_rx.next().await {
1481                let buffers = this.update(&mut cx, |this, cx| {
1482                    project_paths
1483                        .into_iter()
1484                        .map(|project_path| this.open_buffer(project_path, cx))
1485                        .collect::<Vec<_>>()
1486                })?;
1487                for buffer_task in buffers {
1488                    if let Some(buffer) = buffer_task.await.log_err() {
1489                        if tx.send(buffer).await.is_err() {
1490                            return anyhow::Ok(());
1491                        }
1492                    }
1493                }
1494            }
1495            anyhow::Ok(())
1496        })
1497        .detach();
1498        rx
1499    }
1500
1501    pub fn recalculate_buffer_diffs(
1502        &mut self,
1503        buffers: Vec<Entity<Buffer>>,
1504        cx: &mut Context<Self>,
1505    ) -> impl Future<Output = ()> {
1506        let mut futures = Vec::new();
1507        for buffer in buffers {
1508            let buffer = buffer.read(cx).text_snapshot();
1509            if let Some(OpenBuffer::Complete {
1510                unstaged_changes, ..
1511            }) = self.opened_buffers.get_mut(&buffer.remote_id())
1512            {
1513                if let Some(unstaged_changes) = unstaged_changes
1514                    .as_ref()
1515                    .and_then(|changes| changes.upgrade())
1516                {
1517                    unstaged_changes.update(cx, |unstaged_changes, cx| {
1518                        futures.push(unstaged_changes.recalculate_diff(buffer.clone(), cx));
1519                    });
1520                } else {
1521                    unstaged_changes.take();
1522                }
1523            }
1524        }
1525        async move {
1526            futures::future::join_all(futures).await;
1527        }
1528    }
1529
1530    fn on_buffer_event(
1531        &mut self,
1532        buffer: Entity<Buffer>,
1533        event: &BufferEvent,
1534        cx: &mut Context<Self>,
1535    ) {
1536        match event {
1537            BufferEvent::FileHandleChanged => {
1538                if let Some(local) = self.as_local_mut() {
1539                    local.buffer_changed_file(buffer, cx);
1540                }
1541            }
1542            BufferEvent::Reloaded => {
1543                let Some((downstream_client, project_id)) = self.downstream_client.as_ref() else {
1544                    return;
1545                };
1546                let buffer = buffer.read(cx);
1547                downstream_client
1548                    .send(proto::BufferReloaded {
1549                        project_id: *project_id,
1550                        buffer_id: buffer.remote_id().to_proto(),
1551                        version: serialize_version(&buffer.version()),
1552                        mtime: buffer.saved_mtime().map(|t| t.into()),
1553                        line_ending: serialize_line_ending(buffer.line_ending()) as i32,
1554                    })
1555                    .log_err();
1556            }
1557            _ => {}
1558        }
1559    }
1560
1561    pub async fn handle_update_buffer(
1562        this: Entity<Self>,
1563        envelope: TypedEnvelope<proto::UpdateBuffer>,
1564        mut cx: AsyncApp,
1565    ) -> Result<proto::Ack> {
1566        let payload = envelope.payload.clone();
1567        let buffer_id = BufferId::new(payload.buffer_id)?;
1568        let ops = payload
1569            .operations
1570            .into_iter()
1571            .map(language::proto::deserialize_operation)
1572            .collect::<Result<Vec<_>, _>>()?;
1573        this.update(&mut cx, |this, cx| {
1574            match this.opened_buffers.entry(buffer_id) {
1575                hash_map::Entry::Occupied(mut e) => match e.get_mut() {
1576                    OpenBuffer::Operations(operations) => operations.extend_from_slice(&ops),
1577                    OpenBuffer::Complete { buffer, .. } => {
1578                        if let Some(buffer) = buffer.upgrade() {
1579                            buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx));
1580                        }
1581                    }
1582                },
1583                hash_map::Entry::Vacant(e) => {
1584                    e.insert(OpenBuffer::Operations(ops));
1585                }
1586            }
1587            Ok(proto::Ack {})
1588        })?
1589    }
1590
1591    pub fn register_shared_lsp_handle(
1592        &mut self,
1593        peer_id: proto::PeerId,
1594        buffer_id: BufferId,
1595        handle: OpenLspBufferHandle,
1596    ) {
1597        if let Some(shared_buffers) = self.shared_buffers.get_mut(&peer_id) {
1598            if let Some(buffer) = shared_buffers.get_mut(&buffer_id) {
1599                buffer.lsp_handle = Some(handle);
1600                return;
1601            }
1602        }
1603        debug_panic!("tried to register shared lsp handle, but buffer was not shared")
1604    }
1605
1606    pub fn handle_synchronize_buffers(
1607        &mut self,
1608        envelope: TypedEnvelope<proto::SynchronizeBuffers>,
1609        cx: &mut Context<Self>,
1610        client: Arc<Client>,
1611    ) -> Result<proto::SynchronizeBuffersResponse> {
1612        let project_id = envelope.payload.project_id;
1613        let mut response = proto::SynchronizeBuffersResponse {
1614            buffers: Default::default(),
1615        };
1616        let Some(guest_id) = envelope.original_sender_id else {
1617            anyhow::bail!("missing original_sender_id on SynchronizeBuffers request");
1618        };
1619
1620        self.shared_buffers.entry(guest_id).or_default().clear();
1621        for buffer in envelope.payload.buffers {
1622            let buffer_id = BufferId::new(buffer.id)?;
1623            let remote_version = language::proto::deserialize_version(&buffer.version);
1624            if let Some(buffer) = self.get(buffer_id) {
1625                self.shared_buffers
1626                    .entry(guest_id)
1627                    .or_default()
1628                    .entry(buffer_id)
1629                    .or_insert_with(|| SharedBuffer {
1630                        buffer: buffer.clone(),
1631                        unstaged_changes: None,
1632                        lsp_handle: None,
1633                    });
1634
1635                let buffer = buffer.read(cx);
1636                response.buffers.push(proto::BufferVersion {
1637                    id: buffer_id.into(),
1638                    version: language::proto::serialize_version(&buffer.version),
1639                });
1640
1641                let operations = buffer.serialize_ops(Some(remote_version), cx);
1642                let client = client.clone();
1643                if let Some(file) = buffer.file() {
1644                    client
1645                        .send(proto::UpdateBufferFile {
1646                            project_id,
1647                            buffer_id: buffer_id.into(),
1648                            file: Some(file.to_proto(cx)),
1649                        })
1650                        .log_err();
1651                }
1652
1653                // TODO(max): do something
1654                // client
1655                //     .send(proto::UpdateStagedText {
1656                //         project_id,
1657                //         buffer_id: buffer_id.into(),
1658                //         diff_base: buffer.diff_base().map(ToString::to_string),
1659                //     })
1660                //     .log_err();
1661
1662                client
1663                    .send(proto::BufferReloaded {
1664                        project_id,
1665                        buffer_id: buffer_id.into(),
1666                        version: language::proto::serialize_version(buffer.saved_version()),
1667                        mtime: buffer.saved_mtime().map(|time| time.into()),
1668                        line_ending: language::proto::serialize_line_ending(buffer.line_ending())
1669                            as i32,
1670                    })
1671                    .log_err();
1672
1673                cx.background_executor()
1674                    .spawn(
1675                        async move {
1676                            let operations = operations.await;
1677                            for chunk in split_operations(operations) {
1678                                client
1679                                    .request(proto::UpdateBuffer {
1680                                        project_id,
1681                                        buffer_id: buffer_id.into(),
1682                                        operations: chunk,
1683                                    })
1684                                    .await?;
1685                            }
1686                            anyhow::Ok(())
1687                        }
1688                        .log_err(),
1689                    )
1690                    .detach();
1691            }
1692        }
1693        Ok(response)
1694    }
1695
1696    pub fn handle_create_buffer_for_peer(
1697        &mut self,
1698        envelope: TypedEnvelope<proto::CreateBufferForPeer>,
1699        replica_id: u16,
1700        capability: Capability,
1701        cx: &mut Context<Self>,
1702    ) -> Result<()> {
1703        let Some(remote) = self.as_remote_mut() else {
1704            return Err(anyhow!("buffer store is not a remote"));
1705        };
1706
1707        if let Some(buffer) =
1708            remote.handle_create_buffer_for_peer(envelope, replica_id, capability, cx)?
1709        {
1710            self.add_buffer(buffer, cx)?;
1711        }
1712
1713        Ok(())
1714    }
1715
1716    pub async fn handle_update_buffer_file(
1717        this: Entity<Self>,
1718        envelope: TypedEnvelope<proto::UpdateBufferFile>,
1719        mut cx: AsyncApp,
1720    ) -> Result<()> {
1721        let buffer_id = envelope.payload.buffer_id;
1722        let buffer_id = BufferId::new(buffer_id)?;
1723
1724        this.update(&mut cx, |this, cx| {
1725            let payload = envelope.payload.clone();
1726            if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1727                let file = payload.file.ok_or_else(|| anyhow!("invalid file"))?;
1728                let worktree = this
1729                    .worktree_store
1730                    .read(cx)
1731                    .worktree_for_id(WorktreeId::from_proto(file.worktree_id), cx)
1732                    .ok_or_else(|| anyhow!("no such worktree"))?;
1733                let file = File::from_proto(file, worktree, cx)?;
1734                let old_file = buffer.update(cx, |buffer, cx| {
1735                    let old_file = buffer.file().cloned();
1736                    let new_path = file.path.clone();
1737                    buffer.file_updated(Arc::new(file), cx);
1738                    if old_file
1739                        .as_ref()
1740                        .map_or(true, |old| *old.path() != new_path)
1741                    {
1742                        Some(old_file)
1743                    } else {
1744                        None
1745                    }
1746                });
1747                if let Some(old_file) = old_file {
1748                    cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });
1749                }
1750            }
1751            if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1752                downstream_client
1753                    .send(proto::UpdateBufferFile {
1754                        project_id: *project_id,
1755                        buffer_id: buffer_id.into(),
1756                        file: envelope.payload.file,
1757                    })
1758                    .log_err();
1759            }
1760            Ok(())
1761        })?
1762    }
1763
1764    pub async fn handle_save_buffer(
1765        this: Entity<Self>,
1766        envelope: TypedEnvelope<proto::SaveBuffer>,
1767        mut cx: AsyncApp,
1768    ) -> Result<proto::BufferSaved> {
1769        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1770        let (buffer, project_id) = this.update(&mut cx, |this, _| {
1771            anyhow::Ok((
1772                this.get_existing(buffer_id)?,
1773                this.downstream_client
1774                    .as_ref()
1775                    .map(|(_, project_id)| *project_id)
1776                    .context("project is not shared")?,
1777            ))
1778        })??;
1779        buffer
1780            .update(&mut cx, |buffer, _| {
1781                buffer.wait_for_version(deserialize_version(&envelope.payload.version))
1782            })?
1783            .await?;
1784        let buffer_id = buffer.update(&mut cx, |buffer, _| buffer.remote_id())?;
1785
1786        if let Some(new_path) = envelope.payload.new_path {
1787            let new_path = ProjectPath::from_proto(new_path);
1788            this.update(&mut cx, |this, cx| {
1789                this.save_buffer_as(buffer.clone(), new_path, cx)
1790            })?
1791            .await?;
1792        } else {
1793            this.update(&mut cx, |this, cx| this.save_buffer(buffer.clone(), cx))?
1794                .await?;
1795        }
1796
1797        buffer.update(&mut cx, |buffer, _| proto::BufferSaved {
1798            project_id,
1799            buffer_id: buffer_id.into(),
1800            version: serialize_version(buffer.saved_version()),
1801            mtime: buffer.saved_mtime().map(|time| time.into()),
1802        })
1803    }
1804
1805    pub async fn handle_close_buffer(
1806        this: Entity<Self>,
1807        envelope: TypedEnvelope<proto::CloseBuffer>,
1808        mut cx: AsyncApp,
1809    ) -> Result<()> {
1810        let peer_id = envelope.sender_id;
1811        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1812        this.update(&mut cx, |this, _| {
1813            if let Some(shared) = this.shared_buffers.get_mut(&peer_id) {
1814                if shared.remove(&buffer_id).is_some() {
1815                    if shared.is_empty() {
1816                        this.shared_buffers.remove(&peer_id);
1817                    }
1818                    return;
1819                }
1820            }
1821            debug_panic!(
1822                "peer_id {} closed buffer_id {} which was either not open or already closed",
1823                peer_id,
1824                buffer_id
1825            )
1826        })
1827    }
1828
1829    pub async fn handle_buffer_saved(
1830        this: Entity<Self>,
1831        envelope: TypedEnvelope<proto::BufferSaved>,
1832        mut cx: AsyncApp,
1833    ) -> Result<()> {
1834        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1835        let version = deserialize_version(&envelope.payload.version);
1836        let mtime = envelope.payload.mtime.clone().map(|time| time.into());
1837        this.update(&mut cx, move |this, cx| {
1838            if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1839                buffer.update(cx, |buffer, cx| {
1840                    buffer.did_save(version, mtime, cx);
1841                });
1842            }
1843
1844            if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1845                downstream_client
1846                    .send(proto::BufferSaved {
1847                        project_id: *project_id,
1848                        buffer_id: buffer_id.into(),
1849                        mtime: envelope.payload.mtime,
1850                        version: envelope.payload.version,
1851                    })
1852                    .log_err();
1853            }
1854        })
1855    }
1856
1857    pub async fn handle_buffer_reloaded(
1858        this: Entity<Self>,
1859        envelope: TypedEnvelope<proto::BufferReloaded>,
1860        mut cx: AsyncApp,
1861    ) -> Result<()> {
1862        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1863        let version = deserialize_version(&envelope.payload.version);
1864        let mtime = envelope.payload.mtime.clone().map(|time| time.into());
1865        let line_ending = deserialize_line_ending(
1866            proto::LineEnding::from_i32(envelope.payload.line_ending)
1867                .ok_or_else(|| anyhow!("missing line ending"))?,
1868        );
1869        this.update(&mut cx, |this, cx| {
1870            if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1871                buffer.update(cx, |buffer, cx| {
1872                    buffer.did_reload(version, line_ending, mtime, cx);
1873                });
1874            }
1875
1876            if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1877                downstream_client
1878                    .send(proto::BufferReloaded {
1879                        project_id: *project_id,
1880                        buffer_id: buffer_id.into(),
1881                        mtime: envelope.payload.mtime,
1882                        version: envelope.payload.version,
1883                        line_ending: envelope.payload.line_ending,
1884                    })
1885                    .log_err();
1886            }
1887        })
1888    }
1889
1890    pub async fn handle_blame_buffer(
1891        this: Entity<Self>,
1892        envelope: TypedEnvelope<proto::BlameBuffer>,
1893        mut cx: AsyncApp,
1894    ) -> Result<proto::BlameBufferResponse> {
1895        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1896        let version = deserialize_version(&envelope.payload.version);
1897        let buffer = this.read_with(&cx, |this, _| this.get_existing(buffer_id))??;
1898        buffer
1899            .update(&mut cx, |buffer, _| {
1900                buffer.wait_for_version(version.clone())
1901            })?
1902            .await?;
1903        let blame = this
1904            .update(&mut cx, |this, cx| {
1905                this.blame_buffer(&buffer, Some(version), cx)
1906            })?
1907            .await?;
1908        Ok(serialize_blame_buffer_response(blame))
1909    }
1910
1911    pub async fn handle_get_permalink_to_line(
1912        this: Entity<Self>,
1913        envelope: TypedEnvelope<proto::GetPermalinkToLine>,
1914        mut cx: AsyncApp,
1915    ) -> Result<proto::GetPermalinkToLineResponse> {
1916        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1917        // let version = deserialize_version(&envelope.payload.version);
1918        let selection = {
1919            let proto_selection = envelope
1920                .payload
1921                .selection
1922                .context("no selection to get permalink for defined")?;
1923            proto_selection.start as u32..proto_selection.end as u32
1924        };
1925        let buffer = this.read_with(&cx, |this, _| this.get_existing(buffer_id))??;
1926        let permalink = this
1927            .update(&mut cx, |this, cx| {
1928                this.get_permalink_to_line(&buffer, selection, cx)
1929            })?
1930            .await?;
1931        Ok(proto::GetPermalinkToLineResponse {
1932            permalink: permalink.to_string(),
1933        })
1934    }
1935
1936    pub async fn handle_get_staged_text(
1937        this: Entity<Self>,
1938        request: TypedEnvelope<proto::GetStagedText>,
1939        mut cx: AsyncApp,
1940    ) -> Result<proto::GetStagedTextResponse> {
1941        let buffer_id = BufferId::new(request.payload.buffer_id)?;
1942        let change_set = this
1943            .update(&mut cx, |this, cx| {
1944                let buffer = this.get(buffer_id)?;
1945                Some(this.open_unstaged_changes(buffer, cx))
1946            })?
1947            .ok_or_else(|| anyhow!("no such buffer"))?
1948            .await?;
1949        this.update(&mut cx, |this, _| {
1950            let shared_buffers = this
1951                .shared_buffers
1952                .entry(request.original_sender_id.unwrap_or(request.sender_id))
1953                .or_default();
1954            debug_assert!(shared_buffers.contains_key(&buffer_id));
1955            if let Some(shared) = shared_buffers.get_mut(&buffer_id) {
1956                shared.unstaged_changes = Some(change_set.clone());
1957            }
1958        })?;
1959        let staged_text = change_set.read_with(&cx, |change_set, _| {
1960            change_set.base_text.as_ref().map(|buffer| buffer.text())
1961        })?;
1962        Ok(proto::GetStagedTextResponse { staged_text })
1963    }
1964
1965    pub async fn handle_update_diff_base(
1966        this: Entity<Self>,
1967        request: TypedEnvelope<proto::UpdateDiffBase>,
1968        mut cx: AsyncApp,
1969    ) -> Result<()> {
1970        let buffer_id = BufferId::new(request.payload.buffer_id)?;
1971        let Some((buffer, change_set)) = this.update(&mut cx, |this, _| {
1972            if let OpenBuffer::Complete {
1973                unstaged_changes,
1974                buffer,
1975            } = this.opened_buffers.get(&buffer_id)?
1976            {
1977                Some((buffer.upgrade()?, unstaged_changes.as_ref()?.upgrade()?))
1978            } else {
1979                None
1980            }
1981        })?
1982        else {
1983            return Ok(());
1984        };
1985        change_set.update(&mut cx, |change_set, cx| {
1986            if let Some(staged_text) = request.payload.staged_text {
1987                let _ = change_set.set_base_text(staged_text, buffer.read(cx).text_snapshot(), cx);
1988            } else {
1989                change_set.unset_base_text(buffer.read(cx).text_snapshot(), cx)
1990            }
1991        })?;
1992        Ok(())
1993    }
1994
1995    pub fn reload_buffers(
1996        &self,
1997        buffers: HashSet<Entity<Buffer>>,
1998        push_to_history: bool,
1999        cx: &mut Context<Self>,
2000    ) -> Task<Result<ProjectTransaction>> {
2001        if buffers.is_empty() {
2002            return Task::ready(Ok(ProjectTransaction::default()));
2003        }
2004        match &self.state {
2005            BufferStoreState::Local(this) => this.reload_buffers(buffers, push_to_history, cx),
2006            BufferStoreState::Remote(this) => this.reload_buffers(buffers, push_to_history, cx),
2007        }
2008    }
2009
2010    async fn handle_reload_buffers(
2011        this: Entity<Self>,
2012        envelope: TypedEnvelope<proto::ReloadBuffers>,
2013        mut cx: AsyncApp,
2014    ) -> Result<proto::ReloadBuffersResponse> {
2015        let sender_id = envelope.original_sender_id().unwrap_or_default();
2016        let reload = this.update(&mut cx, |this, cx| {
2017            let mut buffers = HashSet::default();
2018            for buffer_id in &envelope.payload.buffer_ids {
2019                let buffer_id = BufferId::new(*buffer_id)?;
2020                buffers.insert(this.get_existing(buffer_id)?);
2021            }
2022            Ok::<_, anyhow::Error>(this.reload_buffers(buffers, false, cx))
2023        })??;
2024
2025        let project_transaction = reload.await?;
2026        let project_transaction = this.update(&mut cx, |this, cx| {
2027            this.serialize_project_transaction_for_peer(project_transaction, sender_id, cx)
2028        })?;
2029        Ok(proto::ReloadBuffersResponse {
2030            transaction: Some(project_transaction),
2031        })
2032    }
2033
2034    pub fn create_buffer_for_peer(
2035        &mut self,
2036        buffer: &Entity<Buffer>,
2037        peer_id: proto::PeerId,
2038        cx: &mut Context<Self>,
2039    ) -> Task<Result<()>> {
2040        let buffer_id = buffer.read(cx).remote_id();
2041        let shared_buffers = self.shared_buffers.entry(peer_id).or_default();
2042        if shared_buffers.contains_key(&buffer_id) {
2043            return Task::ready(Ok(()));
2044        }
2045        shared_buffers.insert(
2046            buffer_id,
2047            SharedBuffer {
2048                buffer: buffer.clone(),
2049                unstaged_changes: None,
2050                lsp_handle: None,
2051            },
2052        );
2053
2054        let Some((client, project_id)) = self.downstream_client.clone() else {
2055            return Task::ready(Ok(()));
2056        };
2057
2058        cx.spawn(|this, mut cx| async move {
2059            let Some(buffer) = this.update(&mut cx, |this, _| this.get(buffer_id))? else {
2060                return anyhow::Ok(());
2061            };
2062
2063            let operations = buffer.update(&mut cx, |b, cx| b.serialize_ops(None, cx))?;
2064            let operations = operations.await;
2065            let state = buffer.update(&mut cx, |buffer, cx| buffer.to_proto(cx))?;
2066
2067            let initial_state = proto::CreateBufferForPeer {
2068                project_id,
2069                peer_id: Some(peer_id),
2070                variant: Some(proto::create_buffer_for_peer::Variant::State(state)),
2071            };
2072
2073            if client.send(initial_state).log_err().is_some() {
2074                let client = client.clone();
2075                cx.background_executor()
2076                    .spawn(async move {
2077                        let mut chunks = split_operations(operations).peekable();
2078                        while let Some(chunk) = chunks.next() {
2079                            let is_last = chunks.peek().is_none();
2080                            client.send(proto::CreateBufferForPeer {
2081                                project_id,
2082                                peer_id: Some(peer_id),
2083                                variant: Some(proto::create_buffer_for_peer::Variant::Chunk(
2084                                    proto::BufferChunk {
2085                                        buffer_id: buffer_id.into(),
2086                                        operations: chunk,
2087                                        is_last,
2088                                    },
2089                                )),
2090                            })?;
2091                        }
2092                        anyhow::Ok(())
2093                    })
2094                    .await
2095                    .log_err();
2096            }
2097            Ok(())
2098        })
2099    }
2100
2101    pub fn forget_shared_buffers(&mut self) {
2102        self.shared_buffers.clear();
2103    }
2104
2105    pub fn forget_shared_buffers_for(&mut self, peer_id: &proto::PeerId) {
2106        self.shared_buffers.remove(peer_id);
2107    }
2108
2109    pub fn update_peer_id(&mut self, old_peer_id: &proto::PeerId, new_peer_id: proto::PeerId) {
2110        if let Some(buffers) = self.shared_buffers.remove(old_peer_id) {
2111            self.shared_buffers.insert(new_peer_id, buffers);
2112        }
2113    }
2114
2115    pub fn has_shared_buffers(&self) -> bool {
2116        !self.shared_buffers.is_empty()
2117    }
2118
2119    pub fn create_local_buffer(
2120        &mut self,
2121        text: &str,
2122        language: Option<Arc<Language>>,
2123        cx: &mut Context<Self>,
2124    ) -> Entity<Buffer> {
2125        let buffer = cx.new(|cx| {
2126            Buffer::local(text, cx)
2127                .with_language(language.unwrap_or_else(|| language::PLAIN_TEXT.clone()), cx)
2128        });
2129
2130        self.add_buffer(buffer.clone(), cx).log_err();
2131        let buffer_id = buffer.read(cx).remote_id();
2132
2133        let this = self
2134            .as_local_mut()
2135            .expect("local-only method called in a non-local context");
2136        if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
2137            this.local_buffer_ids_by_path.insert(
2138                ProjectPath {
2139                    worktree_id: file.worktree_id(cx),
2140                    path: file.path.clone(),
2141                },
2142                buffer_id,
2143            );
2144
2145            if let Some(entry_id) = file.entry_id {
2146                this.local_buffer_ids_by_entry_id
2147                    .insert(entry_id, buffer_id);
2148            }
2149        }
2150        buffer
2151    }
2152
2153    pub fn deserialize_project_transaction(
2154        &mut self,
2155        message: proto::ProjectTransaction,
2156        push_to_history: bool,
2157        cx: &mut Context<Self>,
2158    ) -> Task<Result<ProjectTransaction>> {
2159        if let Some(this) = self.as_remote_mut() {
2160            this.deserialize_project_transaction(message, push_to_history, cx)
2161        } else {
2162            debug_panic!("not a remote buffer store");
2163            Task::ready(Err(anyhow!("not a remote buffer store")))
2164        }
2165    }
2166
2167    pub fn wait_for_remote_buffer(
2168        &mut self,
2169        id: BufferId,
2170        cx: &mut Context<BufferStore>,
2171    ) -> Task<Result<Entity<Buffer>>> {
2172        if let Some(this) = self.as_remote_mut() {
2173            this.wait_for_remote_buffer(id, cx)
2174        } else {
2175            debug_panic!("not a remote buffer store");
2176            Task::ready(Err(anyhow!("not a remote buffer store")))
2177        }
2178    }
2179
2180    pub fn serialize_project_transaction_for_peer(
2181        &mut self,
2182        project_transaction: ProjectTransaction,
2183        peer_id: proto::PeerId,
2184        cx: &mut Context<Self>,
2185    ) -> proto::ProjectTransaction {
2186        let mut serialized_transaction = proto::ProjectTransaction {
2187            buffer_ids: Default::default(),
2188            transactions: Default::default(),
2189        };
2190        for (buffer, transaction) in project_transaction.0 {
2191            self.create_buffer_for_peer(&buffer, peer_id, cx)
2192                .detach_and_log_err(cx);
2193            serialized_transaction
2194                .buffer_ids
2195                .push(buffer.read(cx).remote_id().into());
2196            serialized_transaction
2197                .transactions
2198                .push(language::proto::serialize_transaction(&transaction));
2199        }
2200        serialized_transaction
2201    }
2202}
2203
2204impl BufferChangeSet {
2205    pub fn new(buffer: &Entity<Buffer>, cx: &mut Context<Self>) -> Self {
2206        cx.subscribe(buffer, |this, buffer, event, cx| match event {
2207            BufferEvent::LanguageChanged => {
2208                this.language = buffer.read(cx).language().cloned();
2209                if let Some(base_text) = &this.base_text {
2210                    let snapshot = language::Buffer::build_snapshot(
2211                        base_text.as_rope().clone(),
2212                        this.language.clone(),
2213                        this.language_registry.clone(),
2214                        cx,
2215                    );
2216                    this.recalculate_diff_task = Some(cx.spawn(|this, mut cx| async move {
2217                        let base_text = cx.background_executor().spawn(snapshot).await;
2218                        this.update(&mut cx, |this, cx| {
2219                            this.base_text = Some(base_text);
2220                            cx.notify();
2221                        })
2222                    }));
2223                }
2224            }
2225            _ => {}
2226        })
2227        .detach();
2228
2229        let buffer = buffer.read(cx);
2230
2231        Self {
2232            buffer_id: buffer.remote_id(),
2233            base_text: None,
2234            diff_to_buffer: git::diff::BufferDiff::new(buffer),
2235            recalculate_diff_task: None,
2236            diff_updated_futures: Vec::new(),
2237            language: buffer.language().cloned(),
2238            language_registry: buffer.language_registry(),
2239        }
2240    }
2241
2242    #[cfg(any(test, feature = "test-support"))]
2243    pub fn new_with_base_text(
2244        base_text: String,
2245        buffer: &Entity<Buffer>,
2246        cx: &mut Context<Self>,
2247    ) -> Self {
2248        let mut this = Self::new(&buffer, cx);
2249        let _ = this.set_base_text(base_text, buffer.read(cx).text_snapshot(), cx);
2250        this
2251    }
2252
2253    pub fn diff_hunks_intersecting_range<'a>(
2254        &'a self,
2255        range: Range<text::Anchor>,
2256        buffer_snapshot: &'a text::BufferSnapshot,
2257    ) -> impl 'a + Iterator<Item = git::diff::DiffHunk> {
2258        self.diff_to_buffer
2259            .hunks_intersecting_range(range, buffer_snapshot)
2260    }
2261
2262    pub fn diff_hunks_intersecting_range_rev<'a>(
2263        &'a self,
2264        range: Range<text::Anchor>,
2265        buffer_snapshot: &'a text::BufferSnapshot,
2266    ) -> impl 'a + Iterator<Item = git::diff::DiffHunk> {
2267        self.diff_to_buffer
2268            .hunks_intersecting_range_rev(range, buffer_snapshot)
2269    }
2270
2271    #[cfg(any(test, feature = "test-support"))]
2272    pub fn base_text_string(&self) -> Option<String> {
2273        self.base_text.as_ref().map(|buffer| buffer.text())
2274    }
2275
2276    pub fn set_base_text(
2277        &mut self,
2278        mut base_text: String,
2279        buffer_snapshot: text::BufferSnapshot,
2280        cx: &mut Context<Self>,
2281    ) -> oneshot::Receiver<()> {
2282        LineEnding::normalize(&mut base_text);
2283        self.recalculate_diff_internal(base_text, buffer_snapshot, true, cx)
2284    }
2285
2286    pub fn unset_base_text(
2287        &mut self,
2288        buffer_snapshot: text::BufferSnapshot,
2289        cx: &mut Context<Self>,
2290    ) {
2291        if self.base_text.is_some() {
2292            self.base_text = None;
2293            self.diff_to_buffer = BufferDiff::new(&buffer_snapshot);
2294            self.recalculate_diff_task.take();
2295            cx.notify();
2296        }
2297    }
2298
2299    pub fn recalculate_diff(
2300        &mut self,
2301        buffer_snapshot: text::BufferSnapshot,
2302        cx: &mut Context<Self>,
2303    ) -> oneshot::Receiver<()> {
2304        if let Some(base_text) = self.base_text.clone() {
2305            self.recalculate_diff_internal(base_text.text(), buffer_snapshot, false, cx)
2306        } else {
2307            oneshot::channel().1
2308        }
2309    }
2310
2311    fn recalculate_diff_internal(
2312        &mut self,
2313        base_text: String,
2314        buffer_snapshot: text::BufferSnapshot,
2315        base_text_changed: bool,
2316        cx: &mut Context<Self>,
2317    ) -> oneshot::Receiver<()> {
2318        let (tx, rx) = oneshot::channel();
2319        self.diff_updated_futures.push(tx);
2320        self.recalculate_diff_task = Some(cx.spawn(|this, mut cx| async move {
2321            let new_base_text = if base_text_changed {
2322                let base_text_rope: Rope = base_text.as_str().into();
2323                let snapshot = this.update(&mut cx, |this, cx| {
2324                    language::Buffer::build_snapshot(
2325                        base_text_rope,
2326                        this.language.clone(),
2327                        this.language_registry.clone(),
2328                        cx,
2329                    )
2330                })?;
2331                Some(cx.background_executor().spawn(snapshot).await)
2332            } else {
2333                None
2334            };
2335            let diff = cx
2336                .background_executor()
2337                .spawn({
2338                    let buffer_snapshot = buffer_snapshot.clone();
2339                    async move { BufferDiff::build(&base_text, &buffer_snapshot) }
2340                })
2341                .await;
2342            this.update(&mut cx, |this, cx| {
2343                if let Some(new_base_text) = new_base_text {
2344                    this.base_text = Some(new_base_text)
2345                }
2346                this.diff_to_buffer = diff;
2347                this.recalculate_diff_task.take();
2348                for tx in this.diff_updated_futures.drain(..) {
2349                    tx.send(()).ok();
2350                }
2351                cx.notify();
2352            })?;
2353            Ok(())
2354        }));
2355        rx
2356    }
2357
2358    #[cfg(any(test, feature = "test-support"))]
2359    pub fn recalculate_diff_sync(
2360        &mut self,
2361        mut base_text: String,
2362        buffer_snapshot: text::BufferSnapshot,
2363        base_text_changed: bool,
2364        cx: &mut Context<Self>,
2365    ) {
2366        LineEnding::normalize(&mut base_text);
2367        let diff = BufferDiff::build(&base_text, &buffer_snapshot);
2368        if base_text_changed {
2369            self.base_text = Some(
2370                cx.background_executor()
2371                    .clone()
2372                    .block(Buffer::build_snapshot(
2373                        base_text.into(),
2374                        self.language.clone(),
2375                        self.language_registry.clone(),
2376                        cx,
2377                    )),
2378            );
2379        }
2380        self.diff_to_buffer = diff;
2381        self.recalculate_diff_task.take();
2382        cx.notify();
2383    }
2384}
2385
2386impl OpenBuffer {
2387    fn upgrade(&self) -> Option<Entity<Buffer>> {
2388        match self {
2389            OpenBuffer::Complete { buffer, .. } => buffer.upgrade(),
2390            OpenBuffer::Operations(_) => None,
2391        }
2392    }
2393}
2394
2395fn is_not_found_error(error: &anyhow::Error) -> bool {
2396    error
2397        .root_cause()
2398        .downcast_ref::<io::Error>()
2399        .is_some_and(|err| err.kind() == io::ErrorKind::NotFound)
2400}
2401
2402fn serialize_blame_buffer_response(blame: Option<git::blame::Blame>) -> proto::BlameBufferResponse {
2403    let Some(blame) = blame else {
2404        return proto::BlameBufferResponse {
2405            blame_response: None,
2406        };
2407    };
2408
2409    let entries = blame
2410        .entries
2411        .into_iter()
2412        .map(|entry| proto::BlameEntry {
2413            sha: entry.sha.as_bytes().into(),
2414            start_line: entry.range.start,
2415            end_line: entry.range.end,
2416            original_line_number: entry.original_line_number,
2417            author: entry.author.clone(),
2418            author_mail: entry.author_mail.clone(),
2419            author_time: entry.author_time,
2420            author_tz: entry.author_tz.clone(),
2421            committer: entry.committer.clone(),
2422            committer_mail: entry.committer_mail.clone(),
2423            committer_time: entry.committer_time,
2424            committer_tz: entry.committer_tz.clone(),
2425            summary: entry.summary.clone(),
2426            previous: entry.previous.clone(),
2427            filename: entry.filename.clone(),
2428        })
2429        .collect::<Vec<_>>();
2430
2431    let messages = blame
2432        .messages
2433        .into_iter()
2434        .map(|(oid, message)| proto::CommitMessage {
2435            oid: oid.as_bytes().into(),
2436            message,
2437        })
2438        .collect::<Vec<_>>();
2439
2440    let permalinks = blame
2441        .permalinks
2442        .into_iter()
2443        .map(|(oid, url)| proto::CommitPermalink {
2444            oid: oid.as_bytes().into(),
2445            permalink: url.to_string(),
2446        })
2447        .collect::<Vec<_>>();
2448
2449    proto::BlameBufferResponse {
2450        blame_response: Some(proto::blame_buffer_response::BlameResponse {
2451            entries,
2452            messages,
2453            permalinks,
2454            remote_url: blame.remote_url,
2455        }),
2456    }
2457}
2458
2459fn deserialize_blame_buffer_response(
2460    response: proto::BlameBufferResponse,
2461) -> Option<git::blame::Blame> {
2462    let response = response.blame_response?;
2463    let entries = response
2464        .entries
2465        .into_iter()
2466        .filter_map(|entry| {
2467            Some(git::blame::BlameEntry {
2468                sha: git::Oid::from_bytes(&entry.sha).ok()?,
2469                range: entry.start_line..entry.end_line,
2470                original_line_number: entry.original_line_number,
2471                committer: entry.committer,
2472                committer_time: entry.committer_time,
2473                committer_tz: entry.committer_tz,
2474                committer_mail: entry.committer_mail,
2475                author: entry.author,
2476                author_mail: entry.author_mail,
2477                author_time: entry.author_time,
2478                author_tz: entry.author_tz,
2479                summary: entry.summary,
2480                previous: entry.previous,
2481                filename: entry.filename,
2482            })
2483        })
2484        .collect::<Vec<_>>();
2485
2486    let messages = response
2487        .messages
2488        .into_iter()
2489        .filter_map(|message| Some((git::Oid::from_bytes(&message.oid).ok()?, message.message)))
2490        .collect::<HashMap<_, _>>();
2491
2492    let permalinks = response
2493        .permalinks
2494        .into_iter()
2495        .filter_map(|permalink| {
2496            Some((
2497                git::Oid::from_bytes(&permalink.oid).ok()?,
2498                Url::from_str(&permalink.permalink).ok()?,
2499            ))
2500        })
2501        .collect::<HashMap<_, _>>();
2502
2503    Some(Blame {
2504        entries,
2505        permalinks,
2506        messages,
2507        remote_url: response.remote_url,
2508    })
2509}
2510
2511fn get_permalink_in_rust_registry_src(
2512    provider_registry: Arc<GitHostingProviderRegistry>,
2513    path: PathBuf,
2514    selection: Range<u32>,
2515) -> Result<url::Url> {
2516    #[derive(Deserialize)]
2517    struct CargoVcsGit {
2518        sha1: String,
2519    }
2520
2521    #[derive(Deserialize)]
2522    struct CargoVcsInfo {
2523        git: CargoVcsGit,
2524        path_in_vcs: String,
2525    }
2526
2527    #[derive(Deserialize)]
2528    struct CargoPackage {
2529        repository: String,
2530    }
2531
2532    #[derive(Deserialize)]
2533    struct CargoToml {
2534        package: CargoPackage,
2535    }
2536
2537    let Some((dir, cargo_vcs_info_json)) = path.ancestors().skip(1).find_map(|dir| {
2538        let json = std::fs::read_to_string(dir.join(".cargo_vcs_info.json")).ok()?;
2539        Some((dir, json))
2540    }) else {
2541        bail!("No .cargo_vcs_info.json found in parent directories")
2542    };
2543    let cargo_vcs_info = serde_json::from_str::<CargoVcsInfo>(&cargo_vcs_info_json)?;
2544    let cargo_toml = std::fs::read_to_string(dir.join("Cargo.toml"))?;
2545    let manifest = toml::from_str::<CargoToml>(&cargo_toml)?;
2546    let (provider, remote) = parse_git_remote_url(provider_registry, &manifest.package.repository)
2547        .ok_or_else(|| anyhow!("Failed to parse package.repository field of manifest"))?;
2548    let path = PathBuf::from(cargo_vcs_info.path_in_vcs).join(path.strip_prefix(dir).unwrap());
2549    let permalink = provider.build_permalink(
2550        remote,
2551        BuildPermalinkParams {
2552            sha: &cargo_vcs_info.git.sha1,
2553            path: &path.to_string_lossy(),
2554            selection: Some(selection),
2555        },
2556    );
2557    Ok(permalink)
2558}