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