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