1pub mod branch_diff;
2mod conflict_set;
3pub mod git_traversal;
4
5use crate::{
6 ProjectEnvironment, ProjectItem, ProjectPath,
7 buffer_store::{BufferStore, BufferStoreEvent},
8 worktree_store::{WorktreeStore, WorktreeStoreEvent},
9};
10use anyhow::{Context as _, Result, anyhow, bail};
11use askpass::{AskPassDelegate, EncryptedPassword, IKnowWhatIAmDoingAndIHaveReadTheDocs};
12use buffer_diff::{BufferDiff, BufferDiffEvent};
13use client::ProjectId;
14use collections::HashMap;
15pub use conflict_set::{ConflictRegion, ConflictSet, ConflictSetSnapshot, ConflictSetUpdate};
16use fs::Fs;
17use futures::{
18 FutureExt, StreamExt,
19 channel::{mpsc, oneshot},
20 future::{self, Shared},
21 stream::FuturesOrdered,
22};
23use git::{
24 BuildPermalinkParams, GitHostingProviderRegistry, Oid,
25 blame::Blame,
26 parse_git_remote_url,
27 repository::{
28 Branch, CommitDetails, CommitDiff, CommitFile, CommitOptions, DiffType, FetchOptions,
29 GitRepository, GitRepositoryCheckpoint, PushOptions, Remote, RemoteCommandOutput, RepoPath,
30 ResetMode, UpstreamTrackingStatus,
31 },
32 stash::{GitStash, StashEntry},
33 status::{
34 DiffTreeType, FileStatus, GitSummary, StageStatus, StatusCode, TrackedStatus, TreeDiff,
35 TreeDiffStatus, UnmergedStatus, UnmergedStatusCode,
36 },
37};
38use gpui::{
39 App, AppContext, AsyncApp, Context, Entity, EventEmitter, SharedString, Subscription, Task,
40 WeakEntity,
41};
42use language::{
43 Buffer, BufferEvent, Language, LanguageRegistry,
44 proto::{deserialize_version, serialize_version},
45};
46use parking_lot::Mutex;
47use postage::stream::Stream as _;
48use rpc::{
49 AnyProtoClient, TypedEnvelope,
50 proto::{self, git_reset, split_repository_update},
51};
52use serde::Deserialize;
53use std::{
54 cmp::Ordering,
55 collections::{BTreeSet, VecDeque},
56 future::Future,
57 mem,
58 ops::Range,
59 path::{Path, PathBuf},
60 str::FromStr,
61 sync::{
62 Arc,
63 atomic::{self, AtomicU64},
64 },
65 time::Instant,
66};
67use sum_tree::{Edit, SumTree, TreeSet};
68use task::Shell;
69use text::{Bias, BufferId};
70use util::{
71 ResultExt, debug_panic,
72 paths::{PathStyle, SanitizedPath},
73 post_inc,
74 rel_path::RelPath,
75};
76use worktree::{
77 File, PathChange, PathKey, PathProgress, PathSummary, PathTarget, ProjectEntryId,
78 UpdatedGitRepositoriesSet, UpdatedGitRepository, Worktree,
79};
80use zeroize::Zeroize;
81
82pub struct GitStore {
83 state: GitStoreState,
84 buffer_store: Entity<BufferStore>,
85 worktree_store: Entity<WorktreeStore>,
86 repositories: HashMap<RepositoryId, Entity<Repository>>,
87 active_repo_id: Option<RepositoryId>,
88 #[allow(clippy::type_complexity)]
89 loading_diffs:
90 HashMap<(BufferId, DiffKind), Shared<Task<Result<Entity<BufferDiff>, Arc<anyhow::Error>>>>>,
91 diffs: HashMap<BufferId, Entity<BufferGitState>>,
92 shared_diffs: HashMap<proto::PeerId, HashMap<BufferId, SharedDiffs>>,
93 _subscriptions: Vec<Subscription>,
94}
95
96#[derive(Default)]
97struct SharedDiffs {
98 unstaged: Option<Entity<BufferDiff>>,
99 uncommitted: Option<Entity<BufferDiff>>,
100}
101
102struct BufferGitState {
103 unstaged_diff: Option<WeakEntity<BufferDiff>>,
104 uncommitted_diff: Option<WeakEntity<BufferDiff>>,
105 conflict_set: Option<WeakEntity<ConflictSet>>,
106 recalculate_diff_task: Option<Task<Result<()>>>,
107 reparse_conflict_markers_task: Option<Task<Result<()>>>,
108 language: Option<Arc<Language>>,
109 language_registry: Option<Arc<LanguageRegistry>>,
110 conflict_updated_futures: Vec<oneshot::Sender<()>>,
111 recalculating_tx: postage::watch::Sender<bool>,
112
113 /// These operation counts are used to ensure that head and index text
114 /// values read from the git repository are up-to-date with any hunk staging
115 /// operations that have been performed on the BufferDiff.
116 ///
117 /// The operation count is incremented immediately when the user initiates a
118 /// hunk stage/unstage operation. Then, upon finishing writing the new index
119 /// text do disk, the `operation count as of write` is updated to reflect
120 /// the operation count that prompted the write.
121 hunk_staging_operation_count: usize,
122 hunk_staging_operation_count_as_of_write: usize,
123
124 head_text: Option<Arc<String>>,
125 index_text: Option<Arc<String>>,
126 head_changed: bool,
127 index_changed: bool,
128 language_changed: bool,
129}
130
131#[derive(Clone, Debug)]
132enum DiffBasesChange {
133 SetIndex(Option<String>),
134 SetHead(Option<String>),
135 SetEach {
136 index: Option<String>,
137 head: Option<String>,
138 },
139 SetBoth(Option<String>),
140}
141
142#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
143enum DiffKind {
144 Unstaged,
145 Uncommitted,
146}
147
148enum GitStoreState {
149 Local {
150 next_repository_id: Arc<AtomicU64>,
151 downstream: Option<LocalDownstreamState>,
152 project_environment: Entity<ProjectEnvironment>,
153 fs: Arc<dyn Fs>,
154 },
155 Remote {
156 upstream_client: AnyProtoClient,
157 upstream_project_id: u64,
158 downstream: Option<(AnyProtoClient, ProjectId)>,
159 },
160}
161
162enum DownstreamUpdate {
163 UpdateRepository(RepositorySnapshot),
164 RemoveRepository(RepositoryId),
165}
166
167struct LocalDownstreamState {
168 client: AnyProtoClient,
169 project_id: ProjectId,
170 updates_tx: mpsc::UnboundedSender<DownstreamUpdate>,
171 _task: Task<Result<()>>,
172}
173
174#[derive(Clone, Debug)]
175pub struct GitStoreCheckpoint {
176 checkpoints_by_work_dir_abs_path: HashMap<Arc<Path>, GitRepositoryCheckpoint>,
177}
178
179#[derive(Clone, Debug, PartialEq, Eq)]
180pub struct StatusEntry {
181 pub repo_path: RepoPath,
182 pub status: FileStatus,
183}
184
185impl StatusEntry {
186 fn to_proto(&self) -> proto::StatusEntry {
187 let simple_status = match self.status {
188 FileStatus::Ignored | FileStatus::Untracked => proto::GitStatus::Added as i32,
189 FileStatus::Unmerged { .. } => proto::GitStatus::Conflict as i32,
190 FileStatus::Tracked(TrackedStatus {
191 index_status,
192 worktree_status,
193 }) => tracked_status_to_proto(if worktree_status != StatusCode::Unmodified {
194 worktree_status
195 } else {
196 index_status
197 }),
198 };
199
200 proto::StatusEntry {
201 repo_path: self.repo_path.to_proto(),
202 simple_status,
203 status: Some(status_to_proto(self.status)),
204 }
205 }
206}
207
208impl TryFrom<proto::StatusEntry> for StatusEntry {
209 type Error = anyhow::Error;
210
211 fn try_from(value: proto::StatusEntry) -> Result<Self, Self::Error> {
212 let repo_path = RepoPath::from_proto(&value.repo_path).context("invalid repo path")?;
213 let status = status_from_proto(value.simple_status, value.status)?;
214 Ok(Self { repo_path, status })
215 }
216}
217
218impl sum_tree::Item for StatusEntry {
219 type Summary = PathSummary<GitSummary>;
220
221 fn summary(&self, _: <Self::Summary as sum_tree::Summary>::Context<'_>) -> Self::Summary {
222 PathSummary {
223 max_path: self.repo_path.0.clone(),
224 item_summary: self.status.summary(),
225 }
226 }
227}
228
229impl sum_tree::KeyedItem for StatusEntry {
230 type Key = PathKey;
231
232 fn key(&self) -> Self::Key {
233 PathKey(self.repo_path.0.clone())
234 }
235}
236
237#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
238pub struct RepositoryId(pub u64);
239
240#[derive(Clone, Debug, Default, PartialEq, Eq)]
241pub struct MergeDetails {
242 pub conflicted_paths: TreeSet<RepoPath>,
243 pub message: Option<SharedString>,
244 pub heads: Vec<Option<SharedString>>,
245}
246
247#[derive(Clone, Copy, Debug, PartialEq, Eq)]
248pub enum PendingOperationStatus {
249 Staged,
250 Unstaged,
251 Reverted,
252 Unchanged,
253}
254
255#[derive(Clone, Debug, PartialEq, Eq)]
256pub struct PendingOperations {
257 repo_path: RepoPath,
258 // TODO: move this into StatusEntry
259 staging: StageStatus,
260 ops: Vec<PendingOperation>,
261}
262
263#[derive(Clone, Debug, PartialEq, Eq)]
264pub struct PendingOperation {
265 id: usize,
266 finished: bool,
267 status: PendingOperationStatus,
268}
269
270#[derive(Clone, Copy, Debug, PartialEq, Eq)]
271pub struct PendingOperationsSummary {
272 finished: bool,
273 status: PendingOperationStatus,
274 count: usize,
275}
276
277impl Default for PendingOperationsSummary {
278 fn default() -> Self {
279 Self {
280 finished: false,
281 status: PendingOperationStatus::Unstaged,
282 count: 0,
283 }
284 }
285}
286
287impl sum_tree::ContextLessSummary for PendingOperationsSummary {
288 fn zero() -> Self {
289 Default::default()
290 }
291
292 fn add_summary(&mut self, rhs: &Self) {
293 self.finished = self.finished || rhs.finished;
294 self.count += rhs.count;
295 self.status = rhs.status;
296 }
297}
298
299impl sum_tree::Item for PendingOperations {
300 type Summary = PathSummary<PendingOperationsSummary>;
301
302 fn summary(&self, _: <Self::Summary as sum_tree::Summary>::Context<'_>) -> Self::Summary {
303 let mut item_summary = PendingOperationsSummary {
304 count: self.ops.len(),
305 ..Default::default()
306 };
307 if let Some(op) = self.ops.last() {
308 item_summary.finished = op.finished;
309 item_summary.status = op.status;
310 }
311 PathSummary {
312 max_path: self.repo_path.0.clone(),
313 item_summary,
314 }
315 }
316}
317
318impl sum_tree::KeyedItem for PendingOperations {
319 type Key = PathKey;
320
321 fn key(&self) -> Self::Key {
322 PathKey(self.repo_path.0.clone())
323 }
324}
325
326#[derive(Clone, Debug, PartialEq, Eq)]
327pub struct RepositorySnapshot {
328 pub id: RepositoryId,
329 pub statuses_by_path: SumTree<StatusEntry>,
330 pub pending_ops_by_path: SumTree<PendingOperations>,
331 pub work_directory_abs_path: Arc<Path>,
332 pub path_style: PathStyle,
333 pub branch: Option<Branch>,
334 pub head_commit: Option<CommitDetails>,
335 pub scan_id: u64,
336 pub merge: MergeDetails,
337 pub remote_origin_url: Option<String>,
338 pub remote_upstream_url: Option<String>,
339 pub stash_entries: GitStash,
340}
341
342type JobId = u64;
343
344#[derive(Clone, Debug, PartialEq, Eq)]
345pub struct JobInfo {
346 pub start: Instant,
347 pub message: SharedString,
348}
349
350pub struct Repository {
351 this: WeakEntity<Self>,
352 snapshot: RepositorySnapshot,
353 commit_message_buffer: Option<Entity<Buffer>>,
354 git_store: WeakEntity<GitStore>,
355 // For a local repository, holds paths that have had worktree events since the last status scan completed,
356 // and that should be examined during the next status scan.
357 paths_needing_status_update: BTreeSet<RepoPath>,
358 job_sender: mpsc::UnboundedSender<GitJob>,
359 active_jobs: HashMap<JobId, JobInfo>,
360 job_id: JobId,
361 askpass_delegates: Arc<Mutex<HashMap<u64, AskPassDelegate>>>,
362 latest_askpass_id: u64,
363}
364
365impl std::ops::Deref for Repository {
366 type Target = RepositorySnapshot;
367
368 fn deref(&self) -> &Self::Target {
369 &self.snapshot
370 }
371}
372
373#[derive(Clone)]
374pub enum RepositoryState {
375 Local {
376 backend: Arc<dyn GitRepository>,
377 environment: Arc<HashMap<String, String>>,
378 },
379 Remote {
380 project_id: ProjectId,
381 client: AnyProtoClient,
382 },
383}
384
385#[derive(Clone, Debug, PartialEq, Eq)]
386pub enum RepositoryEvent {
387 StatusesChanged {
388 // TODO could report which statuses changed here
389 full_scan: bool,
390 },
391 MergeHeadsChanged,
392 BranchChanged,
393 StashEntriesChanged,
394}
395
396#[derive(Clone, Debug)]
397pub struct JobsUpdated;
398
399#[derive(Debug)]
400pub enum GitStoreEvent {
401 ActiveRepositoryChanged(Option<RepositoryId>),
402 RepositoryUpdated(RepositoryId, RepositoryEvent, bool),
403 RepositoryAdded,
404 RepositoryRemoved(RepositoryId),
405 IndexWriteError(anyhow::Error),
406 JobsUpdated,
407 ConflictsUpdated,
408}
409
410impl EventEmitter<RepositoryEvent> for Repository {}
411impl EventEmitter<JobsUpdated> for Repository {}
412impl EventEmitter<GitStoreEvent> for GitStore {}
413
414pub struct GitJob {
415 job: Box<dyn FnOnce(RepositoryState, &mut AsyncApp) -> Task<()>>,
416 key: Option<GitJobKey>,
417}
418
419#[derive(PartialEq, Eq)]
420enum GitJobKey {
421 WriteIndex(RepoPath),
422 ReloadBufferDiffBases,
423 RefreshStatuses,
424 ReloadGitState,
425}
426
427impl GitStore {
428 pub fn local(
429 worktree_store: &Entity<WorktreeStore>,
430 buffer_store: Entity<BufferStore>,
431 environment: Entity<ProjectEnvironment>,
432 fs: Arc<dyn Fs>,
433 cx: &mut Context<Self>,
434 ) -> Self {
435 Self::new(
436 worktree_store.clone(),
437 buffer_store,
438 GitStoreState::Local {
439 next_repository_id: Arc::new(AtomicU64::new(1)),
440 downstream: None,
441 project_environment: environment,
442 fs,
443 },
444 cx,
445 )
446 }
447
448 pub fn remote(
449 worktree_store: &Entity<WorktreeStore>,
450 buffer_store: Entity<BufferStore>,
451 upstream_client: AnyProtoClient,
452 project_id: u64,
453 cx: &mut Context<Self>,
454 ) -> Self {
455 Self::new(
456 worktree_store.clone(),
457 buffer_store,
458 GitStoreState::Remote {
459 upstream_client,
460 upstream_project_id: project_id,
461 downstream: None,
462 },
463 cx,
464 )
465 }
466
467 fn new(
468 worktree_store: Entity<WorktreeStore>,
469 buffer_store: Entity<BufferStore>,
470 state: GitStoreState,
471 cx: &mut Context<Self>,
472 ) -> Self {
473 let _subscriptions = vec![
474 cx.subscribe(&worktree_store, Self::on_worktree_store_event),
475 cx.subscribe(&buffer_store, Self::on_buffer_store_event),
476 ];
477
478 GitStore {
479 state,
480 buffer_store,
481 worktree_store,
482 repositories: HashMap::default(),
483 active_repo_id: None,
484 _subscriptions,
485 loading_diffs: HashMap::default(),
486 shared_diffs: HashMap::default(),
487 diffs: HashMap::default(),
488 }
489 }
490
491 pub fn init(client: &AnyProtoClient) {
492 client.add_entity_request_handler(Self::handle_get_remotes);
493 client.add_entity_request_handler(Self::handle_get_branches);
494 client.add_entity_request_handler(Self::handle_get_default_branch);
495 client.add_entity_request_handler(Self::handle_change_branch);
496 client.add_entity_request_handler(Self::handle_create_branch);
497 client.add_entity_request_handler(Self::handle_rename_branch);
498 client.add_entity_request_handler(Self::handle_git_init);
499 client.add_entity_request_handler(Self::handle_push);
500 client.add_entity_request_handler(Self::handle_pull);
501 client.add_entity_request_handler(Self::handle_fetch);
502 client.add_entity_request_handler(Self::handle_stage);
503 client.add_entity_request_handler(Self::handle_unstage);
504 client.add_entity_request_handler(Self::handle_stash);
505 client.add_entity_request_handler(Self::handle_stash_pop);
506 client.add_entity_request_handler(Self::handle_stash_apply);
507 client.add_entity_request_handler(Self::handle_stash_drop);
508 client.add_entity_request_handler(Self::handle_commit);
509 client.add_entity_request_handler(Self::handle_reset);
510 client.add_entity_request_handler(Self::handle_show);
511 client.add_entity_request_handler(Self::handle_load_commit_diff);
512 client.add_entity_request_handler(Self::handle_checkout_files);
513 client.add_entity_request_handler(Self::handle_open_commit_message_buffer);
514 client.add_entity_request_handler(Self::handle_set_index_text);
515 client.add_entity_request_handler(Self::handle_askpass);
516 client.add_entity_request_handler(Self::handle_check_for_pushed_commits);
517 client.add_entity_request_handler(Self::handle_git_diff);
518 client.add_entity_request_handler(Self::handle_tree_diff);
519 client.add_entity_request_handler(Self::handle_get_blob_content);
520 client.add_entity_request_handler(Self::handle_open_unstaged_diff);
521 client.add_entity_request_handler(Self::handle_open_uncommitted_diff);
522 client.add_entity_message_handler(Self::handle_update_diff_bases);
523 client.add_entity_request_handler(Self::handle_get_permalink_to_line);
524 client.add_entity_request_handler(Self::handle_blame_buffer);
525 client.add_entity_message_handler(Self::handle_update_repository);
526 client.add_entity_message_handler(Self::handle_remove_repository);
527 client.add_entity_request_handler(Self::handle_git_clone);
528 }
529
530 pub fn is_local(&self) -> bool {
531 matches!(self.state, GitStoreState::Local { .. })
532 }
533 pub fn set_active_repo_for_path(&mut self, project_path: &ProjectPath, cx: &mut Context<Self>) {
534 if let Some((repo, _)) = self.repository_and_path_for_project_path(project_path, cx) {
535 let id = repo.read(cx).id;
536 if self.active_repo_id != Some(id) {
537 self.active_repo_id = Some(id);
538 cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id)));
539 }
540 }
541 }
542
543 pub fn shared(&mut self, project_id: u64, client: AnyProtoClient, cx: &mut Context<Self>) {
544 match &mut self.state {
545 GitStoreState::Remote {
546 downstream: downstream_client,
547 ..
548 } => {
549 for repo in self.repositories.values() {
550 let update = repo.read(cx).snapshot.initial_update(project_id);
551 for update in split_repository_update(update) {
552 client.send(update).log_err();
553 }
554 }
555 *downstream_client = Some((client, ProjectId(project_id)));
556 }
557 GitStoreState::Local {
558 downstream: downstream_client,
559 ..
560 } => {
561 let mut snapshots = HashMap::default();
562 let (updates_tx, mut updates_rx) = mpsc::unbounded();
563 for repo in self.repositories.values() {
564 updates_tx
565 .unbounded_send(DownstreamUpdate::UpdateRepository(
566 repo.read(cx).snapshot.clone(),
567 ))
568 .ok();
569 }
570 *downstream_client = Some(LocalDownstreamState {
571 client: client.clone(),
572 project_id: ProjectId(project_id),
573 updates_tx,
574 _task: cx.spawn(async move |this, cx| {
575 cx.background_spawn(async move {
576 while let Some(update) = updates_rx.next().await {
577 match update {
578 DownstreamUpdate::UpdateRepository(snapshot) => {
579 if let Some(old_snapshot) = snapshots.get_mut(&snapshot.id)
580 {
581 let update =
582 snapshot.build_update(old_snapshot, project_id);
583 *old_snapshot = snapshot;
584 for update in split_repository_update(update) {
585 client.send(update)?;
586 }
587 } else {
588 let update = snapshot.initial_update(project_id);
589 for update in split_repository_update(update) {
590 client.send(update)?;
591 }
592 snapshots.insert(snapshot.id, snapshot);
593 }
594 }
595 DownstreamUpdate::RemoveRepository(id) => {
596 client.send(proto::RemoveRepository {
597 project_id,
598 id: id.to_proto(),
599 })?;
600 }
601 }
602 }
603 anyhow::Ok(())
604 })
605 .await
606 .ok();
607 this.update(cx, |this, _| {
608 if let GitStoreState::Local {
609 downstream: downstream_client,
610 ..
611 } = &mut this.state
612 {
613 downstream_client.take();
614 } else {
615 unreachable!("unshared called on remote store");
616 }
617 })
618 }),
619 });
620 }
621 }
622 }
623
624 pub fn unshared(&mut self, _cx: &mut Context<Self>) {
625 match &mut self.state {
626 GitStoreState::Local {
627 downstream: downstream_client,
628 ..
629 } => {
630 downstream_client.take();
631 }
632 GitStoreState::Remote {
633 downstream: downstream_client,
634 ..
635 } => {
636 downstream_client.take();
637 }
638 }
639 self.shared_diffs.clear();
640 }
641
642 pub(crate) fn forget_shared_diffs_for(&mut self, peer_id: &proto::PeerId) {
643 self.shared_diffs.remove(peer_id);
644 }
645
646 pub fn active_repository(&self) -> Option<Entity<Repository>> {
647 self.active_repo_id
648 .as_ref()
649 .map(|id| self.repositories[id].clone())
650 }
651
652 pub fn open_unstaged_diff(
653 &mut self,
654 buffer: Entity<Buffer>,
655 cx: &mut Context<Self>,
656 ) -> Task<Result<Entity<BufferDiff>>> {
657 let buffer_id = buffer.read(cx).remote_id();
658 if let Some(diff_state) = self.diffs.get(&buffer_id)
659 && let Some(unstaged_diff) = diff_state
660 .read(cx)
661 .unstaged_diff
662 .as_ref()
663 .and_then(|weak| weak.upgrade())
664 {
665 if let Some(task) =
666 diff_state.update(cx, |diff_state, _| diff_state.wait_for_recalculation())
667 {
668 return cx.background_executor().spawn(async move {
669 task.await;
670 Ok(unstaged_diff)
671 });
672 }
673 return Task::ready(Ok(unstaged_diff));
674 }
675
676 let Some((repo, repo_path)) =
677 self.repository_and_path_for_buffer_id(buffer.read(cx).remote_id(), cx)
678 else {
679 return Task::ready(Err(anyhow!("failed to find git repository for buffer")));
680 };
681
682 let task = self
683 .loading_diffs
684 .entry((buffer_id, DiffKind::Unstaged))
685 .or_insert_with(|| {
686 let staged_text = repo.update(cx, |repo, cx| {
687 repo.load_staged_text(buffer_id, repo_path, cx)
688 });
689 cx.spawn(async move |this, cx| {
690 Self::open_diff_internal(
691 this,
692 DiffKind::Unstaged,
693 staged_text.await.map(DiffBasesChange::SetIndex),
694 buffer,
695 cx,
696 )
697 .await
698 .map_err(Arc::new)
699 })
700 .shared()
701 })
702 .clone();
703
704 cx.background_spawn(async move { task.await.map_err(|e| anyhow!("{e}")) })
705 }
706
707 pub fn open_diff_since(
708 &mut self,
709 oid: Option<git::Oid>,
710 buffer: Entity<Buffer>,
711 repo: Entity<Repository>,
712 languages: Arc<LanguageRegistry>,
713 cx: &mut Context<Self>,
714 ) -> Task<Result<Entity<BufferDiff>>> {
715 cx.spawn(async move |this, cx| {
716 let buffer_snapshot = buffer.update(cx, |buffer, _| buffer.snapshot())?;
717 let content = match oid {
718 None => None,
719 Some(oid) => Some(
720 repo.update(cx, |repo, cx| repo.load_blob_content(oid, cx))?
721 .await?,
722 ),
723 };
724 let buffer_diff = cx.new(|cx| BufferDiff::new(&buffer_snapshot, cx))?;
725
726 buffer_diff
727 .update(cx, |buffer_diff, cx| {
728 buffer_diff.set_base_text(
729 content.map(Arc::new),
730 buffer_snapshot.language().cloned(),
731 Some(languages.clone()),
732 buffer_snapshot.text,
733 cx,
734 )
735 })?
736 .await?;
737 let unstaged_diff = this
738 .update(cx, |this, cx| this.open_unstaged_diff(buffer.clone(), cx))?
739 .await?;
740 buffer_diff.update(cx, |buffer_diff, _| {
741 buffer_diff.set_secondary_diff(unstaged_diff);
742 })?;
743
744 this.update(cx, |_, cx| {
745 cx.subscribe(&buffer_diff, Self::on_buffer_diff_event)
746 .detach();
747 })?;
748
749 Ok(buffer_diff)
750 })
751 }
752
753 pub fn open_uncommitted_diff(
754 &mut self,
755 buffer: Entity<Buffer>,
756 cx: &mut Context<Self>,
757 ) -> Task<Result<Entity<BufferDiff>>> {
758 let buffer_id = buffer.read(cx).remote_id();
759
760 if let Some(diff_state) = self.diffs.get(&buffer_id)
761 && let Some(uncommitted_diff) = diff_state
762 .read(cx)
763 .uncommitted_diff
764 .as_ref()
765 .and_then(|weak| weak.upgrade())
766 {
767 if let Some(task) =
768 diff_state.update(cx, |diff_state, _| diff_state.wait_for_recalculation())
769 {
770 return cx.background_executor().spawn(async move {
771 task.await;
772 Ok(uncommitted_diff)
773 });
774 }
775 return Task::ready(Ok(uncommitted_diff));
776 }
777
778 let Some((repo, repo_path)) =
779 self.repository_and_path_for_buffer_id(buffer.read(cx).remote_id(), cx)
780 else {
781 return Task::ready(Err(anyhow!("failed to find git repository for buffer")));
782 };
783
784 let task = self
785 .loading_diffs
786 .entry((buffer_id, DiffKind::Uncommitted))
787 .or_insert_with(|| {
788 let changes = repo.update(cx, |repo, cx| {
789 repo.load_committed_text(buffer_id, repo_path, cx)
790 });
791
792 // todo(lw): hot foreground spawn
793 cx.spawn(async move |this, cx| {
794 Self::open_diff_internal(this, DiffKind::Uncommitted, changes.await, buffer, cx)
795 .await
796 .map_err(Arc::new)
797 })
798 .shared()
799 })
800 .clone();
801
802 cx.background_spawn(async move { task.await.map_err(|e| anyhow!("{e}")) })
803 }
804
805 async fn open_diff_internal(
806 this: WeakEntity<Self>,
807 kind: DiffKind,
808 texts: Result<DiffBasesChange>,
809 buffer_entity: Entity<Buffer>,
810 cx: &mut AsyncApp,
811 ) -> Result<Entity<BufferDiff>> {
812 let diff_bases_change = match texts {
813 Err(e) => {
814 this.update(cx, |this, cx| {
815 let buffer = buffer_entity.read(cx);
816 let buffer_id = buffer.remote_id();
817 this.loading_diffs.remove(&(buffer_id, kind));
818 })?;
819 return Err(e);
820 }
821 Ok(change) => change,
822 };
823
824 this.update(cx, |this, cx| {
825 let buffer = buffer_entity.read(cx);
826 let buffer_id = buffer.remote_id();
827 let language = buffer.language().cloned();
828 let language_registry = buffer.language_registry();
829 let text_snapshot = buffer.text_snapshot();
830 this.loading_diffs.remove(&(buffer_id, kind));
831
832 let git_store = cx.weak_entity();
833 let diff_state = this
834 .diffs
835 .entry(buffer_id)
836 .or_insert_with(|| cx.new(|_| BufferGitState::new(git_store)));
837
838 let diff = cx.new(|cx| BufferDiff::new(&text_snapshot, cx));
839
840 cx.subscribe(&diff, Self::on_buffer_diff_event).detach();
841 diff_state.update(cx, |diff_state, cx| {
842 diff_state.language = language;
843 diff_state.language_registry = language_registry;
844
845 match kind {
846 DiffKind::Unstaged => diff_state.unstaged_diff = Some(diff.downgrade()),
847 DiffKind::Uncommitted => {
848 let unstaged_diff = if let Some(diff) = diff_state.unstaged_diff() {
849 diff
850 } else {
851 let unstaged_diff = cx.new(|cx| BufferDiff::new(&text_snapshot, cx));
852 diff_state.unstaged_diff = Some(unstaged_diff.downgrade());
853 unstaged_diff
854 };
855
856 diff.update(cx, |diff, _| diff.set_secondary_diff(unstaged_diff));
857 diff_state.uncommitted_diff = Some(diff.downgrade())
858 }
859 }
860
861 diff_state.diff_bases_changed(text_snapshot, Some(diff_bases_change), cx);
862 let rx = diff_state.wait_for_recalculation();
863
864 anyhow::Ok(async move {
865 if let Some(rx) = rx {
866 rx.await;
867 }
868 Ok(diff)
869 })
870 })
871 })??
872 .await
873 }
874
875 pub fn get_unstaged_diff(&self, buffer_id: BufferId, cx: &App) -> Option<Entity<BufferDiff>> {
876 let diff_state = self.diffs.get(&buffer_id)?;
877 diff_state.read(cx).unstaged_diff.as_ref()?.upgrade()
878 }
879
880 pub fn get_uncommitted_diff(
881 &self,
882 buffer_id: BufferId,
883 cx: &App,
884 ) -> Option<Entity<BufferDiff>> {
885 let diff_state = self.diffs.get(&buffer_id)?;
886 diff_state.read(cx).uncommitted_diff.as_ref()?.upgrade()
887 }
888
889 pub fn open_conflict_set(
890 &mut self,
891 buffer: Entity<Buffer>,
892 cx: &mut Context<Self>,
893 ) -> Entity<ConflictSet> {
894 log::debug!("open conflict set");
895 let buffer_id = buffer.read(cx).remote_id();
896
897 if let Some(git_state) = self.diffs.get(&buffer_id)
898 && let Some(conflict_set) = git_state
899 .read(cx)
900 .conflict_set
901 .as_ref()
902 .and_then(|weak| weak.upgrade())
903 {
904 let conflict_set = conflict_set;
905 let buffer_snapshot = buffer.read(cx).text_snapshot();
906
907 git_state.update(cx, |state, cx| {
908 let _ = state.reparse_conflict_markers(buffer_snapshot, cx);
909 });
910
911 return conflict_set;
912 }
913
914 let is_unmerged = self
915 .repository_and_path_for_buffer_id(buffer_id, cx)
916 .is_some_and(|(repo, path)| repo.read(cx).snapshot.has_conflict(&path));
917 let git_store = cx.weak_entity();
918 let buffer_git_state = self
919 .diffs
920 .entry(buffer_id)
921 .or_insert_with(|| cx.new(|_| BufferGitState::new(git_store)));
922 let conflict_set = cx.new(|cx| ConflictSet::new(buffer_id, is_unmerged, cx));
923
924 self._subscriptions
925 .push(cx.subscribe(&conflict_set, |_, _, _, cx| {
926 cx.emit(GitStoreEvent::ConflictsUpdated);
927 }));
928
929 buffer_git_state.update(cx, |state, cx| {
930 state.conflict_set = Some(conflict_set.downgrade());
931 let buffer_snapshot = buffer.read(cx).text_snapshot();
932 let _ = state.reparse_conflict_markers(buffer_snapshot, cx);
933 });
934
935 conflict_set
936 }
937
938 pub fn project_path_git_status(
939 &self,
940 project_path: &ProjectPath,
941 cx: &App,
942 ) -> Option<FileStatus> {
943 let (repo, repo_path) = self.repository_and_path_for_project_path(project_path, cx)?;
944 Some(repo.read(cx).status_for_path(&repo_path)?.status)
945 }
946
947 pub fn checkpoint(&self, cx: &mut App) -> Task<Result<GitStoreCheckpoint>> {
948 let mut work_directory_abs_paths = Vec::new();
949 let mut checkpoints = Vec::new();
950 for repository in self.repositories.values() {
951 repository.update(cx, |repository, _| {
952 work_directory_abs_paths.push(repository.snapshot.work_directory_abs_path.clone());
953 checkpoints.push(repository.checkpoint().map(|checkpoint| checkpoint?));
954 });
955 }
956
957 cx.background_executor().spawn(async move {
958 let checkpoints = future::try_join_all(checkpoints).await?;
959 Ok(GitStoreCheckpoint {
960 checkpoints_by_work_dir_abs_path: work_directory_abs_paths
961 .into_iter()
962 .zip(checkpoints)
963 .collect(),
964 })
965 })
966 }
967
968 pub fn restore_checkpoint(
969 &self,
970 checkpoint: GitStoreCheckpoint,
971 cx: &mut App,
972 ) -> Task<Result<()>> {
973 let repositories_by_work_dir_abs_path = self
974 .repositories
975 .values()
976 .map(|repo| (repo.read(cx).snapshot.work_directory_abs_path.clone(), repo))
977 .collect::<HashMap<_, _>>();
978
979 let mut tasks = Vec::new();
980 for (work_dir_abs_path, checkpoint) in checkpoint.checkpoints_by_work_dir_abs_path {
981 if let Some(repository) = repositories_by_work_dir_abs_path.get(&work_dir_abs_path) {
982 let restore = repository.update(cx, |repository, _| {
983 repository.restore_checkpoint(checkpoint)
984 });
985 tasks.push(async move { restore.await? });
986 }
987 }
988 cx.background_spawn(async move {
989 future::try_join_all(tasks).await?;
990 Ok(())
991 })
992 }
993
994 /// Compares two checkpoints, returning true if they are equal.
995 pub fn compare_checkpoints(
996 &self,
997 left: GitStoreCheckpoint,
998 mut right: GitStoreCheckpoint,
999 cx: &mut App,
1000 ) -> Task<Result<bool>> {
1001 let repositories_by_work_dir_abs_path = self
1002 .repositories
1003 .values()
1004 .map(|repo| (repo.read(cx).snapshot.work_directory_abs_path.clone(), repo))
1005 .collect::<HashMap<_, _>>();
1006
1007 let mut tasks = Vec::new();
1008 for (work_dir_abs_path, left_checkpoint) in left.checkpoints_by_work_dir_abs_path {
1009 if let Some(right_checkpoint) = right
1010 .checkpoints_by_work_dir_abs_path
1011 .remove(&work_dir_abs_path)
1012 {
1013 if let Some(repository) = repositories_by_work_dir_abs_path.get(&work_dir_abs_path)
1014 {
1015 let compare = repository.update(cx, |repository, _| {
1016 repository.compare_checkpoints(left_checkpoint, right_checkpoint)
1017 });
1018
1019 tasks.push(async move { compare.await? });
1020 }
1021 } else {
1022 return Task::ready(Ok(false));
1023 }
1024 }
1025 cx.background_spawn(async move {
1026 Ok(future::try_join_all(tasks)
1027 .await?
1028 .into_iter()
1029 .all(|result| result))
1030 })
1031 }
1032
1033 /// Blames a buffer.
1034 pub fn blame_buffer(
1035 &self,
1036 buffer: &Entity<Buffer>,
1037 version: Option<clock::Global>,
1038 cx: &mut App,
1039 ) -> Task<Result<Option<Blame>>> {
1040 let buffer = buffer.read(cx);
1041 let Some((repo, repo_path)) =
1042 self.repository_and_path_for_buffer_id(buffer.remote_id(), cx)
1043 else {
1044 return Task::ready(Err(anyhow!("failed to find a git repository for buffer")));
1045 };
1046 let content = match &version {
1047 Some(version) => buffer.rope_for_version(version),
1048 None => buffer.as_rope().clone(),
1049 };
1050 let version = version.unwrap_or(buffer.version());
1051 let buffer_id = buffer.remote_id();
1052
1053 let rx = repo.update(cx, |repo, _| {
1054 repo.send_job(None, move |state, _| async move {
1055 match state {
1056 RepositoryState::Local { backend, .. } => backend
1057 .blame(repo_path.clone(), content)
1058 .await
1059 .with_context(|| format!("Failed to blame {:?}", repo_path.0))
1060 .map(Some),
1061 RepositoryState::Remote { project_id, client } => {
1062 let response = client
1063 .request(proto::BlameBuffer {
1064 project_id: project_id.to_proto(),
1065 buffer_id: buffer_id.into(),
1066 version: serialize_version(&version),
1067 })
1068 .await?;
1069 Ok(deserialize_blame_buffer_response(response))
1070 }
1071 }
1072 })
1073 });
1074
1075 cx.spawn(|_: &mut AsyncApp| async move { rx.await? })
1076 }
1077
1078 pub fn get_permalink_to_line(
1079 &self,
1080 buffer: &Entity<Buffer>,
1081 selection: Range<u32>,
1082 cx: &mut App,
1083 ) -> Task<Result<url::Url>> {
1084 let Some(file) = File::from_dyn(buffer.read(cx).file()) else {
1085 return Task::ready(Err(anyhow!("buffer has no file")));
1086 };
1087
1088 let Some((repo, repo_path)) = self.repository_and_path_for_project_path(
1089 &(file.worktree.read(cx).id(), file.path.clone()).into(),
1090 cx,
1091 ) else {
1092 // If we're not in a Git repo, check whether this is a Rust source
1093 // file in the Cargo registry (presumably opened with go-to-definition
1094 // from a normal Rust file). If so, we can put together a permalink
1095 // using crate metadata.
1096 if buffer
1097 .read(cx)
1098 .language()
1099 .is_none_or(|lang| lang.name() != "Rust".into())
1100 {
1101 return Task::ready(Err(anyhow!("no permalink available")));
1102 }
1103 let file_path = file.worktree.read(cx).absolutize(&file.path);
1104 return cx.spawn(async move |cx| {
1105 let provider_registry = cx.update(GitHostingProviderRegistry::default_global)?;
1106 get_permalink_in_rust_registry_src(provider_registry, file_path, selection)
1107 .context("no permalink available")
1108 });
1109 };
1110
1111 let buffer_id = buffer.read(cx).remote_id();
1112 let branch = repo.read(cx).branch.clone();
1113 let remote = branch
1114 .as_ref()
1115 .and_then(|b| b.upstream.as_ref())
1116 .and_then(|b| b.remote_name())
1117 .unwrap_or("origin")
1118 .to_string();
1119
1120 let rx = repo.update(cx, |repo, _| {
1121 repo.send_job(None, move |state, cx| async move {
1122 match state {
1123 RepositoryState::Local { backend, .. } => {
1124 let origin_url = backend
1125 .remote_url(&remote)
1126 .with_context(|| format!("remote \"{remote}\" not found"))?;
1127
1128 let sha = backend.head_sha().await.context("reading HEAD SHA")?;
1129
1130 let provider_registry =
1131 cx.update(GitHostingProviderRegistry::default_global)?;
1132
1133 let (provider, remote) =
1134 parse_git_remote_url(provider_registry, &origin_url)
1135 .context("parsing Git remote URL")?;
1136
1137 Ok(provider.build_permalink(
1138 remote,
1139 BuildPermalinkParams::new(&sha, &repo_path, Some(selection)),
1140 ))
1141 }
1142 RepositoryState::Remote { project_id, client } => {
1143 let response = client
1144 .request(proto::GetPermalinkToLine {
1145 project_id: project_id.to_proto(),
1146 buffer_id: buffer_id.into(),
1147 selection: Some(proto::Range {
1148 start: selection.start as u64,
1149 end: selection.end as u64,
1150 }),
1151 })
1152 .await?;
1153
1154 url::Url::parse(&response.permalink).context("failed to parse permalink")
1155 }
1156 }
1157 })
1158 });
1159 cx.spawn(|_: &mut AsyncApp| async move { rx.await? })
1160 }
1161
1162 fn downstream_client(&self) -> Option<(AnyProtoClient, ProjectId)> {
1163 match &self.state {
1164 GitStoreState::Local {
1165 downstream: downstream_client,
1166 ..
1167 } => downstream_client
1168 .as_ref()
1169 .map(|state| (state.client.clone(), state.project_id)),
1170 GitStoreState::Remote {
1171 downstream: downstream_client,
1172 ..
1173 } => downstream_client.clone(),
1174 }
1175 }
1176
1177 fn upstream_client(&self) -> Option<AnyProtoClient> {
1178 match &self.state {
1179 GitStoreState::Local { .. } => None,
1180 GitStoreState::Remote {
1181 upstream_client, ..
1182 } => Some(upstream_client.clone()),
1183 }
1184 }
1185
1186 fn on_worktree_store_event(
1187 &mut self,
1188 worktree_store: Entity<WorktreeStore>,
1189 event: &WorktreeStoreEvent,
1190 cx: &mut Context<Self>,
1191 ) {
1192 let GitStoreState::Local {
1193 project_environment,
1194 downstream,
1195 next_repository_id,
1196 fs,
1197 } = &self.state
1198 else {
1199 return;
1200 };
1201
1202 match event {
1203 WorktreeStoreEvent::WorktreeUpdatedEntries(worktree_id, updated_entries) => {
1204 if let Some(worktree) = self
1205 .worktree_store
1206 .read(cx)
1207 .worktree_for_id(*worktree_id, cx)
1208 {
1209 let paths_by_git_repo =
1210 self.process_updated_entries(&worktree, updated_entries, cx);
1211 let downstream = downstream
1212 .as_ref()
1213 .map(|downstream| downstream.updates_tx.clone());
1214 cx.spawn(async move |_, cx| {
1215 let paths_by_git_repo = paths_by_git_repo.await;
1216 for (repo, paths) in paths_by_git_repo {
1217 repo.update(cx, |repo, cx| {
1218 repo.paths_changed(paths, downstream.clone(), cx);
1219 })
1220 .ok();
1221 }
1222 })
1223 .detach();
1224 }
1225 }
1226 WorktreeStoreEvent::WorktreeUpdatedGitRepositories(worktree_id, changed_repos) => {
1227 let Some(worktree) = worktree_store.read(cx).worktree_for_id(*worktree_id, cx)
1228 else {
1229 return;
1230 };
1231 if !worktree.read(cx).is_visible() {
1232 log::debug!(
1233 "not adding repositories for local worktree {:?} because it's not visible",
1234 worktree.read(cx).abs_path()
1235 );
1236 return;
1237 }
1238 self.update_repositories_from_worktree(
1239 project_environment.clone(),
1240 next_repository_id.clone(),
1241 downstream
1242 .as_ref()
1243 .map(|downstream| downstream.updates_tx.clone()),
1244 changed_repos.clone(),
1245 fs.clone(),
1246 cx,
1247 );
1248 self.local_worktree_git_repos_changed(worktree, changed_repos, cx);
1249 }
1250 _ => {}
1251 }
1252 }
1253 fn on_repository_event(
1254 &mut self,
1255 repo: Entity<Repository>,
1256 event: &RepositoryEvent,
1257 cx: &mut Context<Self>,
1258 ) {
1259 let id = repo.read(cx).id;
1260 let repo_snapshot = repo.read(cx).snapshot.clone();
1261 for (buffer_id, diff) in self.diffs.iter() {
1262 if let Some((buffer_repo, repo_path)) =
1263 self.repository_and_path_for_buffer_id(*buffer_id, cx)
1264 && buffer_repo == repo
1265 {
1266 diff.update(cx, |diff, cx| {
1267 if let Some(conflict_set) = &diff.conflict_set {
1268 let conflict_status_changed =
1269 conflict_set.update(cx, |conflict_set, cx| {
1270 let has_conflict = repo_snapshot.has_conflict(&repo_path);
1271 conflict_set.set_has_conflict(has_conflict, cx)
1272 })?;
1273 if conflict_status_changed {
1274 let buffer_store = self.buffer_store.read(cx);
1275 if let Some(buffer) = buffer_store.get(*buffer_id) {
1276 let _ = diff
1277 .reparse_conflict_markers(buffer.read(cx).text_snapshot(), cx);
1278 }
1279 }
1280 }
1281 anyhow::Ok(())
1282 })
1283 .ok();
1284 }
1285 }
1286 cx.emit(GitStoreEvent::RepositoryUpdated(
1287 id,
1288 event.clone(),
1289 self.active_repo_id == Some(id),
1290 ))
1291 }
1292
1293 fn on_jobs_updated(&mut self, _: Entity<Repository>, _: &JobsUpdated, cx: &mut Context<Self>) {
1294 cx.emit(GitStoreEvent::JobsUpdated)
1295 }
1296
1297 /// Update our list of repositories and schedule git scans in response to a notification from a worktree,
1298 fn update_repositories_from_worktree(
1299 &mut self,
1300 project_environment: Entity<ProjectEnvironment>,
1301 next_repository_id: Arc<AtomicU64>,
1302 updates_tx: Option<mpsc::UnboundedSender<DownstreamUpdate>>,
1303 updated_git_repositories: UpdatedGitRepositoriesSet,
1304 fs: Arc<dyn Fs>,
1305 cx: &mut Context<Self>,
1306 ) {
1307 let mut removed_ids = Vec::new();
1308 for update in updated_git_repositories.iter() {
1309 if let Some((id, existing)) = self.repositories.iter().find(|(_, repo)| {
1310 let existing_work_directory_abs_path =
1311 repo.read(cx).work_directory_abs_path.clone();
1312 Some(&existing_work_directory_abs_path)
1313 == update.old_work_directory_abs_path.as_ref()
1314 || Some(&existing_work_directory_abs_path)
1315 == update.new_work_directory_abs_path.as_ref()
1316 }) {
1317 if let Some(new_work_directory_abs_path) =
1318 update.new_work_directory_abs_path.clone()
1319 {
1320 existing.update(cx, |existing, cx| {
1321 existing.snapshot.work_directory_abs_path = new_work_directory_abs_path;
1322 existing.schedule_scan(updates_tx.clone(), cx);
1323 });
1324 } else {
1325 removed_ids.push(*id);
1326 }
1327 } else if let UpdatedGitRepository {
1328 new_work_directory_abs_path: Some(work_directory_abs_path),
1329 dot_git_abs_path: Some(dot_git_abs_path),
1330 repository_dir_abs_path: Some(repository_dir_abs_path),
1331 common_dir_abs_path: Some(common_dir_abs_path),
1332 ..
1333 } = update
1334 {
1335 let id = RepositoryId(next_repository_id.fetch_add(1, atomic::Ordering::Release));
1336 let git_store = cx.weak_entity();
1337 let repo = cx.new(|cx| {
1338 let mut repo = Repository::local(
1339 id,
1340 work_directory_abs_path.clone(),
1341 dot_git_abs_path.clone(),
1342 repository_dir_abs_path.clone(),
1343 common_dir_abs_path.clone(),
1344 project_environment.downgrade(),
1345 fs.clone(),
1346 git_store,
1347 cx,
1348 );
1349 repo.schedule_scan(updates_tx.clone(), cx);
1350 repo
1351 });
1352 self._subscriptions
1353 .push(cx.subscribe(&repo, Self::on_repository_event));
1354 self._subscriptions
1355 .push(cx.subscribe(&repo, Self::on_jobs_updated));
1356 self.repositories.insert(id, repo);
1357 cx.emit(GitStoreEvent::RepositoryAdded);
1358 self.active_repo_id.get_or_insert_with(|| {
1359 cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id)));
1360 id
1361 });
1362 }
1363 }
1364
1365 for id in removed_ids {
1366 if self.active_repo_id == Some(id) {
1367 self.active_repo_id = None;
1368 cx.emit(GitStoreEvent::ActiveRepositoryChanged(None));
1369 }
1370 self.repositories.remove(&id);
1371 if let Some(updates_tx) = updates_tx.as_ref() {
1372 updates_tx
1373 .unbounded_send(DownstreamUpdate::RemoveRepository(id))
1374 .ok();
1375 }
1376 }
1377 }
1378
1379 fn on_buffer_store_event(
1380 &mut self,
1381 _: Entity<BufferStore>,
1382 event: &BufferStoreEvent,
1383 cx: &mut Context<Self>,
1384 ) {
1385 match event {
1386 BufferStoreEvent::BufferAdded(buffer) => {
1387 cx.subscribe(buffer, |this, buffer, event, cx| {
1388 if let BufferEvent::LanguageChanged = event {
1389 let buffer_id = buffer.read(cx).remote_id();
1390 if let Some(diff_state) = this.diffs.get(&buffer_id) {
1391 diff_state.update(cx, |diff_state, cx| {
1392 diff_state.buffer_language_changed(buffer, cx);
1393 });
1394 }
1395 }
1396 })
1397 .detach();
1398 }
1399 BufferStoreEvent::SharedBufferClosed(peer_id, buffer_id) => {
1400 if let Some(diffs) = self.shared_diffs.get_mut(peer_id) {
1401 diffs.remove(buffer_id);
1402 }
1403 }
1404 BufferStoreEvent::BufferDropped(buffer_id) => {
1405 self.diffs.remove(buffer_id);
1406 for diffs in self.shared_diffs.values_mut() {
1407 diffs.remove(buffer_id);
1408 }
1409 }
1410
1411 _ => {}
1412 }
1413 }
1414
1415 pub fn recalculate_buffer_diffs(
1416 &mut self,
1417 buffers: Vec<Entity<Buffer>>,
1418 cx: &mut Context<Self>,
1419 ) -> impl Future<Output = ()> + use<> {
1420 let mut futures = Vec::new();
1421 for buffer in buffers {
1422 if let Some(diff_state) = self.diffs.get_mut(&buffer.read(cx).remote_id()) {
1423 let buffer = buffer.read(cx).text_snapshot();
1424 diff_state.update(cx, |diff_state, cx| {
1425 diff_state.recalculate_diffs(buffer.clone(), cx);
1426 futures.extend(diff_state.wait_for_recalculation().map(FutureExt::boxed));
1427 });
1428 futures.push(diff_state.update(cx, |diff_state, cx| {
1429 diff_state
1430 .reparse_conflict_markers(buffer, cx)
1431 .map(|_| {})
1432 .boxed()
1433 }));
1434 }
1435 }
1436 async move {
1437 futures::future::join_all(futures).await;
1438 }
1439 }
1440
1441 fn on_buffer_diff_event(
1442 &mut self,
1443 diff: Entity<buffer_diff::BufferDiff>,
1444 event: &BufferDiffEvent,
1445 cx: &mut Context<Self>,
1446 ) {
1447 if let BufferDiffEvent::HunksStagedOrUnstaged(new_index_text) = event {
1448 let buffer_id = diff.read(cx).buffer_id;
1449 if let Some(diff_state) = self.diffs.get(&buffer_id) {
1450 let hunk_staging_operation_count = diff_state.update(cx, |diff_state, _| {
1451 diff_state.hunk_staging_operation_count += 1;
1452 diff_state.hunk_staging_operation_count
1453 });
1454 if let Some((repo, path)) = self.repository_and_path_for_buffer_id(buffer_id, cx) {
1455 let recv = repo.update(cx, |repo, cx| {
1456 log::debug!("hunks changed for {}", path.as_unix_str());
1457 repo.spawn_set_index_text_job(
1458 path,
1459 new_index_text.as_ref().map(|rope| rope.to_string()),
1460 Some(hunk_staging_operation_count),
1461 cx,
1462 )
1463 });
1464 let diff = diff.downgrade();
1465 cx.spawn(async move |this, cx| {
1466 if let Ok(Err(error)) = cx.background_spawn(recv).await {
1467 diff.update(cx, |diff, cx| {
1468 diff.clear_pending_hunks(cx);
1469 })
1470 .ok();
1471 this.update(cx, |_, cx| cx.emit(GitStoreEvent::IndexWriteError(error)))
1472 .ok();
1473 }
1474 })
1475 .detach();
1476 }
1477 }
1478 }
1479 }
1480
1481 fn local_worktree_git_repos_changed(
1482 &mut self,
1483 worktree: Entity<Worktree>,
1484 changed_repos: &UpdatedGitRepositoriesSet,
1485 cx: &mut Context<Self>,
1486 ) {
1487 log::debug!("local worktree repos changed");
1488 debug_assert!(worktree.read(cx).is_local());
1489
1490 for repository in self.repositories.values() {
1491 repository.update(cx, |repository, cx| {
1492 let repo_abs_path = &repository.work_directory_abs_path;
1493 if changed_repos.iter().any(|update| {
1494 update.old_work_directory_abs_path.as_ref() == Some(repo_abs_path)
1495 || update.new_work_directory_abs_path.as_ref() == Some(repo_abs_path)
1496 }) {
1497 repository.reload_buffer_diff_bases(cx);
1498 }
1499 });
1500 }
1501 }
1502
1503 pub fn repositories(&self) -> &HashMap<RepositoryId, Entity<Repository>> {
1504 &self.repositories
1505 }
1506
1507 pub fn status_for_buffer_id(&self, buffer_id: BufferId, cx: &App) -> Option<FileStatus> {
1508 let (repo, path) = self.repository_and_path_for_buffer_id(buffer_id, cx)?;
1509 let status = repo.read(cx).snapshot.status_for_path(&path)?;
1510 Some(status.status)
1511 }
1512
1513 pub fn repository_and_path_for_buffer_id(
1514 &self,
1515 buffer_id: BufferId,
1516 cx: &App,
1517 ) -> Option<(Entity<Repository>, RepoPath)> {
1518 let buffer = self.buffer_store.read(cx).get(buffer_id)?;
1519 let project_path = buffer.read(cx).project_path(cx)?;
1520 self.repository_and_path_for_project_path(&project_path, cx)
1521 }
1522
1523 pub fn repository_and_path_for_project_path(
1524 &self,
1525 path: &ProjectPath,
1526 cx: &App,
1527 ) -> Option<(Entity<Repository>, RepoPath)> {
1528 let abs_path = self.worktree_store.read(cx).absolutize(path, cx)?;
1529 self.repositories
1530 .values()
1531 .filter_map(|repo| {
1532 let repo_path = repo.read(cx).abs_path_to_repo_path(&abs_path)?;
1533 Some((repo.clone(), repo_path))
1534 })
1535 .max_by_key(|(repo, _)| repo.read(cx).work_directory_abs_path.clone())
1536 }
1537
1538 pub fn git_init(
1539 &self,
1540 path: Arc<Path>,
1541 fallback_branch_name: String,
1542 cx: &App,
1543 ) -> Task<Result<()>> {
1544 match &self.state {
1545 GitStoreState::Local { fs, .. } => {
1546 let fs = fs.clone();
1547 cx.background_executor()
1548 .spawn(async move { fs.git_init(&path, fallback_branch_name).await })
1549 }
1550 GitStoreState::Remote {
1551 upstream_client,
1552 upstream_project_id: project_id,
1553 ..
1554 } => {
1555 let client = upstream_client.clone();
1556 let project_id = *project_id;
1557 cx.background_executor().spawn(async move {
1558 client
1559 .request(proto::GitInit {
1560 project_id: project_id,
1561 abs_path: path.to_string_lossy().into_owned(),
1562 fallback_branch_name,
1563 })
1564 .await?;
1565 Ok(())
1566 })
1567 }
1568 }
1569 }
1570
1571 pub fn git_clone(
1572 &self,
1573 repo: String,
1574 path: impl Into<Arc<std::path::Path>>,
1575 cx: &App,
1576 ) -> Task<Result<()>> {
1577 let path = path.into();
1578 match &self.state {
1579 GitStoreState::Local { fs, .. } => {
1580 let fs = fs.clone();
1581 cx.background_executor()
1582 .spawn(async move { fs.git_clone(&repo, &path).await })
1583 }
1584 GitStoreState::Remote {
1585 upstream_client,
1586 upstream_project_id,
1587 ..
1588 } => {
1589 if upstream_client.is_via_collab() {
1590 return Task::ready(Err(anyhow!(
1591 "Git Clone isn't supported for project guests"
1592 )));
1593 }
1594 let request = upstream_client.request(proto::GitClone {
1595 project_id: *upstream_project_id,
1596 abs_path: path.to_string_lossy().into_owned(),
1597 remote_repo: repo,
1598 });
1599
1600 cx.background_spawn(async move {
1601 let result = request.await?;
1602
1603 match result.success {
1604 true => Ok(()),
1605 false => Err(anyhow!("Git Clone failed")),
1606 }
1607 })
1608 }
1609 }
1610 }
1611
1612 async fn handle_update_repository(
1613 this: Entity<Self>,
1614 envelope: TypedEnvelope<proto::UpdateRepository>,
1615 mut cx: AsyncApp,
1616 ) -> Result<()> {
1617 this.update(&mut cx, |this, cx| {
1618 let path_style = this.worktree_store.read(cx).path_style();
1619 let mut update = envelope.payload;
1620
1621 let id = RepositoryId::from_proto(update.id);
1622 let client = this.upstream_client().context("no upstream client")?;
1623
1624 let mut repo_subscription = None;
1625 let repo = this.repositories.entry(id).or_insert_with(|| {
1626 let git_store = cx.weak_entity();
1627 let repo = cx.new(|cx| {
1628 Repository::remote(
1629 id,
1630 Path::new(&update.abs_path).into(),
1631 path_style,
1632 ProjectId(update.project_id),
1633 client,
1634 git_store,
1635 cx,
1636 )
1637 });
1638 repo_subscription = Some(cx.subscribe(&repo, Self::on_repository_event));
1639 cx.emit(GitStoreEvent::RepositoryAdded);
1640 repo
1641 });
1642 this._subscriptions.extend(repo_subscription);
1643
1644 repo.update(cx, {
1645 let update = update.clone();
1646 |repo, cx| repo.apply_remote_update(update, cx)
1647 })?;
1648
1649 this.active_repo_id.get_or_insert_with(|| {
1650 cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id)));
1651 id
1652 });
1653
1654 if let Some((client, project_id)) = this.downstream_client() {
1655 update.project_id = project_id.to_proto();
1656 client.send(update).log_err();
1657 }
1658 Ok(())
1659 })?
1660 }
1661
1662 async fn handle_remove_repository(
1663 this: Entity<Self>,
1664 envelope: TypedEnvelope<proto::RemoveRepository>,
1665 mut cx: AsyncApp,
1666 ) -> Result<()> {
1667 this.update(&mut cx, |this, cx| {
1668 let mut update = envelope.payload;
1669 let id = RepositoryId::from_proto(update.id);
1670 this.repositories.remove(&id);
1671 if let Some((client, project_id)) = this.downstream_client() {
1672 update.project_id = project_id.to_proto();
1673 client.send(update).log_err();
1674 }
1675 if this.active_repo_id == Some(id) {
1676 this.active_repo_id = None;
1677 cx.emit(GitStoreEvent::ActiveRepositoryChanged(None));
1678 }
1679 cx.emit(GitStoreEvent::RepositoryRemoved(id));
1680 })
1681 }
1682
1683 async fn handle_git_init(
1684 this: Entity<Self>,
1685 envelope: TypedEnvelope<proto::GitInit>,
1686 cx: AsyncApp,
1687 ) -> Result<proto::Ack> {
1688 let path: Arc<Path> = PathBuf::from(envelope.payload.abs_path).into();
1689 let name = envelope.payload.fallback_branch_name;
1690 cx.update(|cx| this.read(cx).git_init(path, name, cx))?
1691 .await?;
1692
1693 Ok(proto::Ack {})
1694 }
1695
1696 async fn handle_git_clone(
1697 this: Entity<Self>,
1698 envelope: TypedEnvelope<proto::GitClone>,
1699 cx: AsyncApp,
1700 ) -> Result<proto::GitCloneResponse> {
1701 let path: Arc<Path> = PathBuf::from(envelope.payload.abs_path).into();
1702 let repo_name = envelope.payload.remote_repo;
1703 let result = cx
1704 .update(|cx| this.read(cx).git_clone(repo_name, path, cx))?
1705 .await;
1706
1707 Ok(proto::GitCloneResponse {
1708 success: result.is_ok(),
1709 })
1710 }
1711
1712 async fn handle_fetch(
1713 this: Entity<Self>,
1714 envelope: TypedEnvelope<proto::Fetch>,
1715 mut cx: AsyncApp,
1716 ) -> Result<proto::RemoteMessageResponse> {
1717 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1718 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1719 let fetch_options = FetchOptions::from_proto(envelope.payload.remote);
1720 let askpass_id = envelope.payload.askpass_id;
1721
1722 let askpass = make_remote_delegate(
1723 this,
1724 envelope.payload.project_id,
1725 repository_id,
1726 askpass_id,
1727 &mut cx,
1728 );
1729
1730 let remote_output = repository_handle
1731 .update(&mut cx, |repository_handle, cx| {
1732 repository_handle.fetch(fetch_options, askpass, cx)
1733 })?
1734 .await??;
1735
1736 Ok(proto::RemoteMessageResponse {
1737 stdout: remote_output.stdout,
1738 stderr: remote_output.stderr,
1739 })
1740 }
1741
1742 async fn handle_push(
1743 this: Entity<Self>,
1744 envelope: TypedEnvelope<proto::Push>,
1745 mut cx: AsyncApp,
1746 ) -> Result<proto::RemoteMessageResponse> {
1747 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1748 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1749
1750 let askpass_id = envelope.payload.askpass_id;
1751 let askpass = make_remote_delegate(
1752 this,
1753 envelope.payload.project_id,
1754 repository_id,
1755 askpass_id,
1756 &mut cx,
1757 );
1758
1759 let options = envelope
1760 .payload
1761 .options
1762 .as_ref()
1763 .map(|_| match envelope.payload.options() {
1764 proto::push::PushOptions::SetUpstream => git::repository::PushOptions::SetUpstream,
1765 proto::push::PushOptions::Force => git::repository::PushOptions::Force,
1766 });
1767
1768 let branch_name = envelope.payload.branch_name.into();
1769 let remote_name = envelope.payload.remote_name.into();
1770
1771 let remote_output = repository_handle
1772 .update(&mut cx, |repository_handle, cx| {
1773 repository_handle.push(branch_name, remote_name, options, askpass, cx)
1774 })?
1775 .await??;
1776 Ok(proto::RemoteMessageResponse {
1777 stdout: remote_output.stdout,
1778 stderr: remote_output.stderr,
1779 })
1780 }
1781
1782 async fn handle_pull(
1783 this: Entity<Self>,
1784 envelope: TypedEnvelope<proto::Pull>,
1785 mut cx: AsyncApp,
1786 ) -> Result<proto::RemoteMessageResponse> {
1787 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1788 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1789 let askpass_id = envelope.payload.askpass_id;
1790 let askpass = make_remote_delegate(
1791 this,
1792 envelope.payload.project_id,
1793 repository_id,
1794 askpass_id,
1795 &mut cx,
1796 );
1797
1798 let branch_name = envelope.payload.branch_name.into();
1799 let remote_name = envelope.payload.remote_name.into();
1800
1801 let remote_message = repository_handle
1802 .update(&mut cx, |repository_handle, cx| {
1803 repository_handle.pull(branch_name, remote_name, askpass, cx)
1804 })?
1805 .await??;
1806
1807 Ok(proto::RemoteMessageResponse {
1808 stdout: remote_message.stdout,
1809 stderr: remote_message.stderr,
1810 })
1811 }
1812
1813 async fn handle_stage(
1814 this: Entity<Self>,
1815 envelope: TypedEnvelope<proto::Stage>,
1816 mut cx: AsyncApp,
1817 ) -> Result<proto::Ack> {
1818 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1819 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1820
1821 let entries = envelope
1822 .payload
1823 .paths
1824 .into_iter()
1825 .map(|path| RepoPath::new(&path))
1826 .collect::<Result<Vec<_>>>()?;
1827
1828 repository_handle
1829 .update(&mut cx, |repository_handle, cx| {
1830 repository_handle.stage_entries(entries, cx)
1831 })?
1832 .await?;
1833 Ok(proto::Ack {})
1834 }
1835
1836 async fn handle_unstage(
1837 this: Entity<Self>,
1838 envelope: TypedEnvelope<proto::Unstage>,
1839 mut cx: AsyncApp,
1840 ) -> Result<proto::Ack> {
1841 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1842 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1843
1844 let entries = envelope
1845 .payload
1846 .paths
1847 .into_iter()
1848 .map(|path| RepoPath::new(&path))
1849 .collect::<Result<Vec<_>>>()?;
1850
1851 repository_handle
1852 .update(&mut cx, |repository_handle, cx| {
1853 repository_handle.unstage_entries(entries, cx)
1854 })?
1855 .await?;
1856
1857 Ok(proto::Ack {})
1858 }
1859
1860 async fn handle_stash(
1861 this: Entity<Self>,
1862 envelope: TypedEnvelope<proto::Stash>,
1863 mut cx: AsyncApp,
1864 ) -> Result<proto::Ack> {
1865 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1866 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1867
1868 let entries = envelope
1869 .payload
1870 .paths
1871 .into_iter()
1872 .map(|path| RepoPath::new(&path))
1873 .collect::<Result<Vec<_>>>()?;
1874
1875 repository_handle
1876 .update(&mut cx, |repository_handle, cx| {
1877 repository_handle.stash_entries(entries, cx)
1878 })?
1879 .await?;
1880
1881 Ok(proto::Ack {})
1882 }
1883
1884 async fn handle_stash_pop(
1885 this: Entity<Self>,
1886 envelope: TypedEnvelope<proto::StashPop>,
1887 mut cx: AsyncApp,
1888 ) -> Result<proto::Ack> {
1889 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1890 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1891 let stash_index = envelope.payload.stash_index.map(|i| i as usize);
1892
1893 repository_handle
1894 .update(&mut cx, |repository_handle, cx| {
1895 repository_handle.stash_pop(stash_index, cx)
1896 })?
1897 .await?;
1898
1899 Ok(proto::Ack {})
1900 }
1901
1902 async fn handle_stash_apply(
1903 this: Entity<Self>,
1904 envelope: TypedEnvelope<proto::StashApply>,
1905 mut cx: AsyncApp,
1906 ) -> Result<proto::Ack> {
1907 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1908 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1909 let stash_index = envelope.payload.stash_index.map(|i| i as usize);
1910
1911 repository_handle
1912 .update(&mut cx, |repository_handle, cx| {
1913 repository_handle.stash_apply(stash_index, cx)
1914 })?
1915 .await?;
1916
1917 Ok(proto::Ack {})
1918 }
1919
1920 async fn handle_stash_drop(
1921 this: Entity<Self>,
1922 envelope: TypedEnvelope<proto::StashDrop>,
1923 mut cx: AsyncApp,
1924 ) -> Result<proto::Ack> {
1925 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1926 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1927 let stash_index = envelope.payload.stash_index.map(|i| i as usize);
1928
1929 repository_handle
1930 .update(&mut cx, |repository_handle, cx| {
1931 repository_handle.stash_drop(stash_index, cx)
1932 })?
1933 .await??;
1934
1935 Ok(proto::Ack {})
1936 }
1937
1938 async fn handle_set_index_text(
1939 this: Entity<Self>,
1940 envelope: TypedEnvelope<proto::SetIndexText>,
1941 mut cx: AsyncApp,
1942 ) -> Result<proto::Ack> {
1943 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1944 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1945 let repo_path = RepoPath::from_proto(&envelope.payload.path)?;
1946
1947 repository_handle
1948 .update(&mut cx, |repository_handle, cx| {
1949 repository_handle.spawn_set_index_text_job(
1950 repo_path,
1951 envelope.payload.text,
1952 None,
1953 cx,
1954 )
1955 })?
1956 .await??;
1957 Ok(proto::Ack {})
1958 }
1959
1960 async fn handle_commit(
1961 this: Entity<Self>,
1962 envelope: TypedEnvelope<proto::Commit>,
1963 mut cx: AsyncApp,
1964 ) -> Result<proto::Ack> {
1965 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1966 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1967
1968 let message = SharedString::from(envelope.payload.message);
1969 let name = envelope.payload.name.map(SharedString::from);
1970 let email = envelope.payload.email.map(SharedString::from);
1971 let options = envelope.payload.options.unwrap_or_default();
1972
1973 repository_handle
1974 .update(&mut cx, |repository_handle, cx| {
1975 repository_handle.commit(
1976 message,
1977 name.zip(email),
1978 CommitOptions {
1979 amend: options.amend,
1980 signoff: options.signoff,
1981 },
1982 cx,
1983 )
1984 })?
1985 .await??;
1986 Ok(proto::Ack {})
1987 }
1988
1989 async fn handle_get_remotes(
1990 this: Entity<Self>,
1991 envelope: TypedEnvelope<proto::GetRemotes>,
1992 mut cx: AsyncApp,
1993 ) -> Result<proto::GetRemotesResponse> {
1994 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
1995 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
1996
1997 let branch_name = envelope.payload.branch_name;
1998
1999 let remotes = repository_handle
2000 .update(&mut cx, |repository_handle, _| {
2001 repository_handle.get_remotes(branch_name)
2002 })?
2003 .await??;
2004
2005 Ok(proto::GetRemotesResponse {
2006 remotes: remotes
2007 .into_iter()
2008 .map(|remotes| proto::get_remotes_response::Remote {
2009 name: remotes.name.to_string(),
2010 })
2011 .collect::<Vec<_>>(),
2012 })
2013 }
2014
2015 async fn handle_get_branches(
2016 this: Entity<Self>,
2017 envelope: TypedEnvelope<proto::GitGetBranches>,
2018 mut cx: AsyncApp,
2019 ) -> Result<proto::GitBranchesResponse> {
2020 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2021 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2022
2023 let branches = repository_handle
2024 .update(&mut cx, |repository_handle, _| repository_handle.branches())?
2025 .await??;
2026
2027 Ok(proto::GitBranchesResponse {
2028 branches: branches
2029 .into_iter()
2030 .map(|branch| branch_to_proto(&branch))
2031 .collect::<Vec<_>>(),
2032 })
2033 }
2034 async fn handle_get_default_branch(
2035 this: Entity<Self>,
2036 envelope: TypedEnvelope<proto::GetDefaultBranch>,
2037 mut cx: AsyncApp,
2038 ) -> Result<proto::GetDefaultBranchResponse> {
2039 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2040 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2041
2042 let branch = repository_handle
2043 .update(&mut cx, |repository_handle, _| {
2044 repository_handle.default_branch()
2045 })?
2046 .await??
2047 .map(Into::into);
2048
2049 Ok(proto::GetDefaultBranchResponse { branch })
2050 }
2051 async fn handle_create_branch(
2052 this: Entity<Self>,
2053 envelope: TypedEnvelope<proto::GitCreateBranch>,
2054 mut cx: AsyncApp,
2055 ) -> Result<proto::Ack> {
2056 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2057 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2058 let branch_name = envelope.payload.branch_name;
2059
2060 repository_handle
2061 .update(&mut cx, |repository_handle, _| {
2062 repository_handle.create_branch(branch_name)
2063 })?
2064 .await??;
2065
2066 Ok(proto::Ack {})
2067 }
2068
2069 async fn handle_change_branch(
2070 this: Entity<Self>,
2071 envelope: TypedEnvelope<proto::GitChangeBranch>,
2072 mut cx: AsyncApp,
2073 ) -> Result<proto::Ack> {
2074 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2075 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2076 let branch_name = envelope.payload.branch_name;
2077
2078 repository_handle
2079 .update(&mut cx, |repository_handle, _| {
2080 repository_handle.change_branch(branch_name)
2081 })?
2082 .await??;
2083
2084 Ok(proto::Ack {})
2085 }
2086
2087 async fn handle_rename_branch(
2088 this: Entity<Self>,
2089 envelope: TypedEnvelope<proto::GitRenameBranch>,
2090 mut cx: AsyncApp,
2091 ) -> Result<proto::Ack> {
2092 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2093 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2094 let branch = envelope.payload.branch;
2095 let new_name = envelope.payload.new_name;
2096
2097 repository_handle
2098 .update(&mut cx, |repository_handle, _| {
2099 repository_handle.rename_branch(branch, new_name)
2100 })?
2101 .await??;
2102
2103 Ok(proto::Ack {})
2104 }
2105
2106 async fn handle_show(
2107 this: Entity<Self>,
2108 envelope: TypedEnvelope<proto::GitShow>,
2109 mut cx: AsyncApp,
2110 ) -> Result<proto::GitCommitDetails> {
2111 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2112 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2113
2114 let commit = repository_handle
2115 .update(&mut cx, |repository_handle, _| {
2116 repository_handle.show(envelope.payload.commit)
2117 })?
2118 .await??;
2119 Ok(proto::GitCommitDetails {
2120 sha: commit.sha.into(),
2121 message: commit.message.into(),
2122 commit_timestamp: commit.commit_timestamp,
2123 author_email: commit.author_email.into(),
2124 author_name: commit.author_name.into(),
2125 })
2126 }
2127
2128 async fn handle_load_commit_diff(
2129 this: Entity<Self>,
2130 envelope: TypedEnvelope<proto::LoadCommitDiff>,
2131 mut cx: AsyncApp,
2132 ) -> Result<proto::LoadCommitDiffResponse> {
2133 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2134 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2135
2136 let commit_diff = repository_handle
2137 .update(&mut cx, |repository_handle, _| {
2138 repository_handle.load_commit_diff(envelope.payload.commit)
2139 })?
2140 .await??;
2141 Ok(proto::LoadCommitDiffResponse {
2142 files: commit_diff
2143 .files
2144 .into_iter()
2145 .map(|file| proto::CommitFile {
2146 path: file.path.to_proto(),
2147 old_text: file.old_text,
2148 new_text: file.new_text,
2149 })
2150 .collect(),
2151 })
2152 }
2153
2154 async fn handle_reset(
2155 this: Entity<Self>,
2156 envelope: TypedEnvelope<proto::GitReset>,
2157 mut cx: AsyncApp,
2158 ) -> Result<proto::Ack> {
2159 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2160 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2161
2162 let mode = match envelope.payload.mode() {
2163 git_reset::ResetMode::Soft => ResetMode::Soft,
2164 git_reset::ResetMode::Mixed => ResetMode::Mixed,
2165 };
2166
2167 repository_handle
2168 .update(&mut cx, |repository_handle, cx| {
2169 repository_handle.reset(envelope.payload.commit, mode, cx)
2170 })?
2171 .await??;
2172 Ok(proto::Ack {})
2173 }
2174
2175 async fn handle_checkout_files(
2176 this: Entity<Self>,
2177 envelope: TypedEnvelope<proto::GitCheckoutFiles>,
2178 mut cx: AsyncApp,
2179 ) -> Result<proto::Ack> {
2180 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2181 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2182 let paths = envelope
2183 .payload
2184 .paths
2185 .iter()
2186 .map(|s| RepoPath::from_proto(s))
2187 .collect::<Result<Vec<_>>>()?;
2188
2189 repository_handle
2190 .update(&mut cx, |repository_handle, cx| {
2191 repository_handle.checkout_files(&envelope.payload.commit, paths, cx)
2192 })?
2193 .await??;
2194 Ok(proto::Ack {})
2195 }
2196
2197 async fn handle_open_commit_message_buffer(
2198 this: Entity<Self>,
2199 envelope: TypedEnvelope<proto::OpenCommitMessageBuffer>,
2200 mut cx: AsyncApp,
2201 ) -> Result<proto::OpenBufferResponse> {
2202 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2203 let repository = Self::repository_for_request(&this, repository_id, &mut cx)?;
2204 let buffer = repository
2205 .update(&mut cx, |repository, cx| {
2206 repository.open_commit_buffer(None, this.read(cx).buffer_store.clone(), cx)
2207 })?
2208 .await?;
2209
2210 let buffer_id = buffer.read_with(&cx, |buffer, _| buffer.remote_id())?;
2211 this.update(&mut cx, |this, cx| {
2212 this.buffer_store.update(cx, |buffer_store, cx| {
2213 buffer_store
2214 .create_buffer_for_peer(
2215 &buffer,
2216 envelope.original_sender_id.unwrap_or(envelope.sender_id),
2217 cx,
2218 )
2219 .detach_and_log_err(cx);
2220 })
2221 })?;
2222
2223 Ok(proto::OpenBufferResponse {
2224 buffer_id: buffer_id.to_proto(),
2225 })
2226 }
2227
2228 async fn handle_askpass(
2229 this: Entity<Self>,
2230 envelope: TypedEnvelope<proto::AskPassRequest>,
2231 mut cx: AsyncApp,
2232 ) -> Result<proto::AskPassResponse> {
2233 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2234 let repository = Self::repository_for_request(&this, repository_id, &mut cx)?;
2235
2236 let delegates = cx.update(|cx| repository.read(cx).askpass_delegates.clone())?;
2237 let Some(mut askpass) = delegates.lock().remove(&envelope.payload.askpass_id) else {
2238 debug_panic!("no askpass found");
2239 anyhow::bail!("no askpass found");
2240 };
2241
2242 let response = askpass
2243 .ask_password(envelope.payload.prompt)
2244 .await
2245 .ok_or_else(|| anyhow::anyhow!("askpass cancelled"))?;
2246
2247 delegates
2248 .lock()
2249 .insert(envelope.payload.askpass_id, askpass);
2250
2251 // In fact, we don't quite know what we're doing here, as we're sending askpass password unencrypted, but..
2252 Ok(proto::AskPassResponse {
2253 response: response.decrypt(IKnowWhatIAmDoingAndIHaveReadTheDocs)?,
2254 })
2255 }
2256
2257 async fn handle_check_for_pushed_commits(
2258 this: Entity<Self>,
2259 envelope: TypedEnvelope<proto::CheckForPushedCommits>,
2260 mut cx: AsyncApp,
2261 ) -> Result<proto::CheckForPushedCommitsResponse> {
2262 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2263 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2264
2265 let branches = repository_handle
2266 .update(&mut cx, |repository_handle, _| {
2267 repository_handle.check_for_pushed_commits()
2268 })?
2269 .await??;
2270 Ok(proto::CheckForPushedCommitsResponse {
2271 pushed_to: branches
2272 .into_iter()
2273 .map(|commit| commit.to_string())
2274 .collect(),
2275 })
2276 }
2277
2278 async fn handle_git_diff(
2279 this: Entity<Self>,
2280 envelope: TypedEnvelope<proto::GitDiff>,
2281 mut cx: AsyncApp,
2282 ) -> Result<proto::GitDiffResponse> {
2283 let repository_id = RepositoryId::from_proto(envelope.payload.repository_id);
2284 let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?;
2285 let diff_type = match envelope.payload.diff_type() {
2286 proto::git_diff::DiffType::HeadToIndex => DiffType::HeadToIndex,
2287 proto::git_diff::DiffType::HeadToWorktree => DiffType::HeadToWorktree,
2288 };
2289
2290 let mut diff = repository_handle
2291 .update(&mut cx, |repository_handle, cx| {
2292 repository_handle.diff(diff_type, cx)
2293 })?
2294 .await??;
2295 const ONE_MB: usize = 1_000_000;
2296 if diff.len() > ONE_MB {
2297 diff = diff.chars().take(ONE_MB).collect()
2298 }
2299
2300 Ok(proto::GitDiffResponse { diff })
2301 }
2302
2303 async fn handle_tree_diff(
2304 this: Entity<Self>,
2305 request: TypedEnvelope<proto::GetTreeDiff>,
2306 mut cx: AsyncApp,
2307 ) -> Result<proto::GetTreeDiffResponse> {
2308 let repository_id = RepositoryId(request.payload.repository_id);
2309 let diff_type = if request.payload.is_merge {
2310 DiffTreeType::MergeBase {
2311 base: request.payload.base.into(),
2312 head: request.payload.head.into(),
2313 }
2314 } else {
2315 DiffTreeType::Since {
2316 base: request.payload.base.into(),
2317 head: request.payload.head.into(),
2318 }
2319 };
2320
2321 let diff = this
2322 .update(&mut cx, |this, cx| {
2323 let repository = this.repositories().get(&repository_id)?;
2324 Some(repository.update(cx, |repo, cx| repo.diff_tree(diff_type, cx)))
2325 })?
2326 .context("missing repository")?
2327 .await??;
2328
2329 Ok(proto::GetTreeDiffResponse {
2330 entries: diff
2331 .entries
2332 .into_iter()
2333 .map(|(path, status)| proto::TreeDiffStatus {
2334 path: path.0.to_proto(),
2335 status: match status {
2336 TreeDiffStatus::Added {} => proto::tree_diff_status::Status::Added.into(),
2337 TreeDiffStatus::Modified { .. } => {
2338 proto::tree_diff_status::Status::Modified.into()
2339 }
2340 TreeDiffStatus::Deleted { .. } => {
2341 proto::tree_diff_status::Status::Deleted.into()
2342 }
2343 },
2344 oid: match status {
2345 TreeDiffStatus::Deleted { old } | TreeDiffStatus::Modified { old } => {
2346 Some(old.to_string())
2347 }
2348 TreeDiffStatus::Added => None,
2349 },
2350 })
2351 .collect(),
2352 })
2353 }
2354
2355 async fn handle_get_blob_content(
2356 this: Entity<Self>,
2357 request: TypedEnvelope<proto::GetBlobContent>,
2358 mut cx: AsyncApp,
2359 ) -> Result<proto::GetBlobContentResponse> {
2360 let oid = git::Oid::from_str(&request.payload.oid)?;
2361 let repository_id = RepositoryId(request.payload.repository_id);
2362 let content = this
2363 .update(&mut cx, |this, cx| {
2364 let repository = this.repositories().get(&repository_id)?;
2365 Some(repository.update(cx, |repo, cx| repo.load_blob_content(oid, cx)))
2366 })?
2367 .context("missing repository")?
2368 .await?;
2369 Ok(proto::GetBlobContentResponse { content })
2370 }
2371
2372 async fn handle_open_unstaged_diff(
2373 this: Entity<Self>,
2374 request: TypedEnvelope<proto::OpenUnstagedDiff>,
2375 mut cx: AsyncApp,
2376 ) -> Result<proto::OpenUnstagedDiffResponse> {
2377 let buffer_id = BufferId::new(request.payload.buffer_id)?;
2378 let diff = this
2379 .update(&mut cx, |this, cx| {
2380 let buffer = this.buffer_store.read(cx).get(buffer_id)?;
2381 Some(this.open_unstaged_diff(buffer, cx))
2382 })?
2383 .context("missing buffer")?
2384 .await?;
2385 this.update(&mut cx, |this, _| {
2386 let shared_diffs = this
2387 .shared_diffs
2388 .entry(request.original_sender_id.unwrap_or(request.sender_id))
2389 .or_default();
2390 shared_diffs.entry(buffer_id).or_default().unstaged = Some(diff.clone());
2391 })?;
2392 let staged_text = diff.read_with(&cx, |diff, _| diff.base_text_string())?;
2393 Ok(proto::OpenUnstagedDiffResponse { staged_text })
2394 }
2395
2396 async fn handle_open_uncommitted_diff(
2397 this: Entity<Self>,
2398 request: TypedEnvelope<proto::OpenUncommittedDiff>,
2399 mut cx: AsyncApp,
2400 ) -> Result<proto::OpenUncommittedDiffResponse> {
2401 let buffer_id = BufferId::new(request.payload.buffer_id)?;
2402 let diff = this
2403 .update(&mut cx, |this, cx| {
2404 let buffer = this.buffer_store.read(cx).get(buffer_id)?;
2405 Some(this.open_uncommitted_diff(buffer, cx))
2406 })?
2407 .context("missing buffer")?
2408 .await?;
2409 this.update(&mut cx, |this, _| {
2410 let shared_diffs = this
2411 .shared_diffs
2412 .entry(request.original_sender_id.unwrap_or(request.sender_id))
2413 .or_default();
2414 shared_diffs.entry(buffer_id).or_default().uncommitted = Some(diff.clone());
2415 })?;
2416 diff.read_with(&cx, |diff, cx| {
2417 use proto::open_uncommitted_diff_response::Mode;
2418
2419 let unstaged_diff = diff.secondary_diff();
2420 let index_snapshot = unstaged_diff.and_then(|diff| {
2421 let diff = diff.read(cx);
2422 diff.base_text_exists().then(|| diff.base_text())
2423 });
2424
2425 let mode;
2426 let staged_text;
2427 let committed_text;
2428 if diff.base_text_exists() {
2429 let committed_snapshot = diff.base_text();
2430 committed_text = Some(committed_snapshot.text());
2431 if let Some(index_text) = index_snapshot {
2432 if index_text.remote_id() == committed_snapshot.remote_id() {
2433 mode = Mode::IndexMatchesHead;
2434 staged_text = None;
2435 } else {
2436 mode = Mode::IndexAndHead;
2437 staged_text = Some(index_text.text());
2438 }
2439 } else {
2440 mode = Mode::IndexAndHead;
2441 staged_text = None;
2442 }
2443 } else {
2444 mode = Mode::IndexAndHead;
2445 committed_text = None;
2446 staged_text = index_snapshot.as_ref().map(|buffer| buffer.text());
2447 }
2448
2449 proto::OpenUncommittedDiffResponse {
2450 committed_text,
2451 staged_text,
2452 mode: mode.into(),
2453 }
2454 })
2455 }
2456
2457 async fn handle_update_diff_bases(
2458 this: Entity<Self>,
2459 request: TypedEnvelope<proto::UpdateDiffBases>,
2460 mut cx: AsyncApp,
2461 ) -> Result<()> {
2462 let buffer_id = BufferId::new(request.payload.buffer_id)?;
2463 this.update(&mut cx, |this, cx| {
2464 if let Some(diff_state) = this.diffs.get_mut(&buffer_id)
2465 && let Some(buffer) = this.buffer_store.read(cx).get(buffer_id)
2466 {
2467 let buffer = buffer.read(cx).text_snapshot();
2468 diff_state.update(cx, |diff_state, cx| {
2469 diff_state.handle_base_texts_updated(buffer, request.payload, cx);
2470 })
2471 }
2472 })
2473 }
2474
2475 async fn handle_blame_buffer(
2476 this: Entity<Self>,
2477 envelope: TypedEnvelope<proto::BlameBuffer>,
2478 mut cx: AsyncApp,
2479 ) -> Result<proto::BlameBufferResponse> {
2480 let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
2481 let version = deserialize_version(&envelope.payload.version);
2482 let buffer = this.read_with(&cx, |this, cx| {
2483 this.buffer_store.read(cx).get_existing(buffer_id)
2484 })??;
2485 buffer
2486 .update(&mut cx, |buffer, _| {
2487 buffer.wait_for_version(version.clone())
2488 })?
2489 .await?;
2490 let blame = this
2491 .update(&mut cx, |this, cx| {
2492 this.blame_buffer(&buffer, Some(version), cx)
2493 })?
2494 .await?;
2495 Ok(serialize_blame_buffer_response(blame))
2496 }
2497
2498 async fn handle_get_permalink_to_line(
2499 this: Entity<Self>,
2500 envelope: TypedEnvelope<proto::GetPermalinkToLine>,
2501 mut cx: AsyncApp,
2502 ) -> Result<proto::GetPermalinkToLineResponse> {
2503 let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
2504 // let version = deserialize_version(&envelope.payload.version);
2505 let selection = {
2506 let proto_selection = envelope
2507 .payload
2508 .selection
2509 .context("no selection to get permalink for defined")?;
2510 proto_selection.start as u32..proto_selection.end as u32
2511 };
2512 let buffer = this.read_with(&cx, |this, cx| {
2513 this.buffer_store.read(cx).get_existing(buffer_id)
2514 })??;
2515 let permalink = this
2516 .update(&mut cx, |this, cx| {
2517 this.get_permalink_to_line(&buffer, selection, cx)
2518 })?
2519 .await?;
2520 Ok(proto::GetPermalinkToLineResponse {
2521 permalink: permalink.to_string(),
2522 })
2523 }
2524
2525 fn repository_for_request(
2526 this: &Entity<Self>,
2527 id: RepositoryId,
2528 cx: &mut AsyncApp,
2529 ) -> Result<Entity<Repository>> {
2530 this.read_with(cx, |this, _| {
2531 this.repositories
2532 .get(&id)
2533 .context("missing repository handle")
2534 .cloned()
2535 })?
2536 }
2537
2538 pub fn repo_snapshots(&self, cx: &App) -> HashMap<RepositoryId, RepositorySnapshot> {
2539 self.repositories
2540 .iter()
2541 .map(|(id, repo)| (*id, repo.read(cx).snapshot.clone()))
2542 .collect()
2543 }
2544
2545 fn process_updated_entries(
2546 &self,
2547 worktree: &Entity<Worktree>,
2548 updated_entries: &[(Arc<RelPath>, ProjectEntryId, PathChange)],
2549 cx: &mut App,
2550 ) -> Task<HashMap<Entity<Repository>, Vec<RepoPath>>> {
2551 let path_style = worktree.read(cx).path_style();
2552 let mut repo_paths = self
2553 .repositories
2554 .values()
2555 .map(|repo| (repo.read(cx).work_directory_abs_path.clone(), repo.clone()))
2556 .collect::<Vec<_>>();
2557 let mut entries: Vec<_> = updated_entries
2558 .iter()
2559 .map(|(path, _, _)| path.clone())
2560 .collect();
2561 entries.sort();
2562 let worktree = worktree.read(cx);
2563
2564 let entries = entries
2565 .into_iter()
2566 .map(|path| worktree.absolutize(&path))
2567 .collect::<Arc<[_]>>();
2568
2569 let executor = cx.background_executor().clone();
2570 cx.background_executor().spawn(async move {
2571 repo_paths.sort_by(|lhs, rhs| lhs.0.cmp(&rhs.0));
2572 let mut paths_by_git_repo = HashMap::<_, Vec<_>>::default();
2573 let mut tasks = FuturesOrdered::new();
2574 for (repo_path, repo) in repo_paths.into_iter().rev() {
2575 let entries = entries.clone();
2576 let task = executor.spawn(async move {
2577 // Find all repository paths that belong to this repo
2578 let mut ix = entries.partition_point(|path| path < &*repo_path);
2579 if ix == entries.len() {
2580 return None;
2581 };
2582
2583 let mut paths = Vec::new();
2584 // All paths prefixed by a given repo will constitute a continuous range.
2585 while let Some(path) = entries.get(ix)
2586 && let Some(repo_path) = RepositorySnapshot::abs_path_to_repo_path_inner(
2587 &repo_path, path, path_style,
2588 )
2589 {
2590 paths.push((repo_path, ix));
2591 ix += 1;
2592 }
2593 if paths.is_empty() {
2594 None
2595 } else {
2596 Some((repo, paths))
2597 }
2598 });
2599 tasks.push_back(task);
2600 }
2601
2602 // Now, let's filter out the "duplicate" entries that were processed by multiple distinct repos.
2603 let mut path_was_used = vec![false; entries.len()];
2604 let tasks = tasks.collect::<Vec<_>>().await;
2605 // Process tasks from the back: iterating backwards allows us to see more-specific paths first.
2606 // We always want to assign a path to it's innermost repository.
2607 for t in tasks {
2608 let Some((repo, paths)) = t else {
2609 continue;
2610 };
2611 let entry = paths_by_git_repo.entry(repo).or_default();
2612 for (repo_path, ix) in paths {
2613 if path_was_used[ix] {
2614 continue;
2615 }
2616 path_was_used[ix] = true;
2617 entry.push(repo_path);
2618 }
2619 }
2620
2621 paths_by_git_repo
2622 })
2623 }
2624}
2625
2626impl BufferGitState {
2627 fn new(_git_store: WeakEntity<GitStore>) -> Self {
2628 Self {
2629 unstaged_diff: Default::default(),
2630 uncommitted_diff: Default::default(),
2631 recalculate_diff_task: Default::default(),
2632 language: Default::default(),
2633 language_registry: Default::default(),
2634 recalculating_tx: postage::watch::channel_with(false).0,
2635 hunk_staging_operation_count: 0,
2636 hunk_staging_operation_count_as_of_write: 0,
2637 head_text: Default::default(),
2638 index_text: Default::default(),
2639 head_changed: Default::default(),
2640 index_changed: Default::default(),
2641 language_changed: Default::default(),
2642 conflict_updated_futures: Default::default(),
2643 conflict_set: Default::default(),
2644 reparse_conflict_markers_task: Default::default(),
2645 }
2646 }
2647
2648 fn buffer_language_changed(&mut self, buffer: Entity<Buffer>, cx: &mut Context<Self>) {
2649 self.language = buffer.read(cx).language().cloned();
2650 self.language_changed = true;
2651 let _ = self.recalculate_diffs(buffer.read(cx).text_snapshot(), cx);
2652 }
2653
2654 fn reparse_conflict_markers(
2655 &mut self,
2656 buffer: text::BufferSnapshot,
2657 cx: &mut Context<Self>,
2658 ) -> oneshot::Receiver<()> {
2659 let (tx, rx) = oneshot::channel();
2660
2661 let Some(conflict_set) = self
2662 .conflict_set
2663 .as_ref()
2664 .and_then(|conflict_set| conflict_set.upgrade())
2665 else {
2666 return rx;
2667 };
2668
2669 let old_snapshot = conflict_set.read_with(cx, |conflict_set, _| {
2670 if conflict_set.has_conflict {
2671 Some(conflict_set.snapshot())
2672 } else {
2673 None
2674 }
2675 });
2676
2677 if let Some(old_snapshot) = old_snapshot {
2678 self.conflict_updated_futures.push(tx);
2679 self.reparse_conflict_markers_task = Some(cx.spawn(async move |this, cx| {
2680 let (snapshot, changed_range) = cx
2681 .background_spawn(async move {
2682 let new_snapshot = ConflictSet::parse(&buffer);
2683 let changed_range = old_snapshot.compare(&new_snapshot, &buffer);
2684 (new_snapshot, changed_range)
2685 })
2686 .await;
2687 this.update(cx, |this, cx| {
2688 if let Some(conflict_set) = &this.conflict_set {
2689 conflict_set
2690 .update(cx, |conflict_set, cx| {
2691 conflict_set.set_snapshot(snapshot, changed_range, cx);
2692 })
2693 .ok();
2694 }
2695 let futures = std::mem::take(&mut this.conflict_updated_futures);
2696 for tx in futures {
2697 tx.send(()).ok();
2698 }
2699 })
2700 }))
2701 }
2702
2703 rx
2704 }
2705
2706 fn unstaged_diff(&self) -> Option<Entity<BufferDiff>> {
2707 self.unstaged_diff.as_ref().and_then(|set| set.upgrade())
2708 }
2709
2710 fn uncommitted_diff(&self) -> Option<Entity<BufferDiff>> {
2711 self.uncommitted_diff.as_ref().and_then(|set| set.upgrade())
2712 }
2713
2714 fn handle_base_texts_updated(
2715 &mut self,
2716 buffer: text::BufferSnapshot,
2717 message: proto::UpdateDiffBases,
2718 cx: &mut Context<Self>,
2719 ) {
2720 use proto::update_diff_bases::Mode;
2721
2722 let Some(mode) = Mode::from_i32(message.mode) else {
2723 return;
2724 };
2725
2726 let diff_bases_change = match mode {
2727 Mode::HeadOnly => DiffBasesChange::SetHead(message.committed_text),
2728 Mode::IndexOnly => DiffBasesChange::SetIndex(message.staged_text),
2729 Mode::IndexMatchesHead => DiffBasesChange::SetBoth(message.committed_text),
2730 Mode::IndexAndHead => DiffBasesChange::SetEach {
2731 index: message.staged_text,
2732 head: message.committed_text,
2733 },
2734 };
2735
2736 self.diff_bases_changed(buffer, Some(diff_bases_change), cx);
2737 }
2738
2739 pub fn wait_for_recalculation(&mut self) -> Option<impl Future<Output = ()> + use<>> {
2740 if *self.recalculating_tx.borrow() {
2741 let mut rx = self.recalculating_tx.subscribe();
2742 Some(async move {
2743 loop {
2744 let is_recalculating = rx.recv().await;
2745 if is_recalculating != Some(true) {
2746 break;
2747 }
2748 }
2749 })
2750 } else {
2751 None
2752 }
2753 }
2754
2755 fn diff_bases_changed(
2756 &mut self,
2757 buffer: text::BufferSnapshot,
2758 diff_bases_change: Option<DiffBasesChange>,
2759 cx: &mut Context<Self>,
2760 ) {
2761 match diff_bases_change {
2762 Some(DiffBasesChange::SetIndex(index)) => {
2763 self.index_text = index.map(|mut index| {
2764 text::LineEnding::normalize(&mut index);
2765 Arc::new(index)
2766 });
2767 self.index_changed = true;
2768 }
2769 Some(DiffBasesChange::SetHead(head)) => {
2770 self.head_text = head.map(|mut head| {
2771 text::LineEnding::normalize(&mut head);
2772 Arc::new(head)
2773 });
2774 self.head_changed = true;
2775 }
2776 Some(DiffBasesChange::SetBoth(text)) => {
2777 let text = text.map(|mut text| {
2778 text::LineEnding::normalize(&mut text);
2779 Arc::new(text)
2780 });
2781 self.head_text = text.clone();
2782 self.index_text = text;
2783 self.head_changed = true;
2784 self.index_changed = true;
2785 }
2786 Some(DiffBasesChange::SetEach { index, head }) => {
2787 self.index_text = index.map(|mut index| {
2788 text::LineEnding::normalize(&mut index);
2789 Arc::new(index)
2790 });
2791 self.index_changed = true;
2792 self.head_text = head.map(|mut head| {
2793 text::LineEnding::normalize(&mut head);
2794 Arc::new(head)
2795 });
2796 self.head_changed = true;
2797 }
2798 None => {}
2799 }
2800
2801 self.recalculate_diffs(buffer, cx)
2802 }
2803
2804 fn recalculate_diffs(&mut self, buffer: text::BufferSnapshot, cx: &mut Context<Self>) {
2805 *self.recalculating_tx.borrow_mut() = true;
2806
2807 let language = self.language.clone();
2808 let language_registry = self.language_registry.clone();
2809 let unstaged_diff = self.unstaged_diff();
2810 let uncommitted_diff = self.uncommitted_diff();
2811 let head = self.head_text.clone();
2812 let index = self.index_text.clone();
2813 let index_changed = self.index_changed;
2814 let head_changed = self.head_changed;
2815 let language_changed = self.language_changed;
2816 let prev_hunk_staging_operation_count = self.hunk_staging_operation_count_as_of_write;
2817 let index_matches_head = match (self.index_text.as_ref(), self.head_text.as_ref()) {
2818 (Some(index), Some(head)) => Arc::ptr_eq(index, head),
2819 (None, None) => true,
2820 _ => false,
2821 };
2822 self.recalculate_diff_task = Some(cx.spawn(async move |this, cx| {
2823 log::debug!(
2824 "start recalculating diffs for buffer {}",
2825 buffer.remote_id()
2826 );
2827
2828 let mut new_unstaged_diff = None;
2829 if let Some(unstaged_diff) = &unstaged_diff {
2830 new_unstaged_diff = Some(
2831 BufferDiff::update_diff(
2832 unstaged_diff.clone(),
2833 buffer.clone(),
2834 index,
2835 index_changed,
2836 language_changed,
2837 language.clone(),
2838 language_registry.clone(),
2839 cx,
2840 )
2841 .await?,
2842 );
2843 }
2844
2845 let mut new_uncommitted_diff = None;
2846 if let Some(uncommitted_diff) = &uncommitted_diff {
2847 new_uncommitted_diff = if index_matches_head {
2848 new_unstaged_diff.clone()
2849 } else {
2850 Some(
2851 BufferDiff::update_diff(
2852 uncommitted_diff.clone(),
2853 buffer.clone(),
2854 head,
2855 head_changed,
2856 language_changed,
2857 language.clone(),
2858 language_registry.clone(),
2859 cx,
2860 )
2861 .await?,
2862 )
2863 }
2864 }
2865
2866 let cancel = this.update(cx, |this, _| {
2867 // This checks whether all pending stage/unstage operations
2868 // have quiesced (i.e. both the corresponding write and the
2869 // read of that write have completed). If not, then we cancel
2870 // this recalculation attempt to avoid invalidating pending
2871 // state too quickly; another recalculation will come along
2872 // later and clear the pending state once the state of the index has settled.
2873 if this.hunk_staging_operation_count > prev_hunk_staging_operation_count {
2874 *this.recalculating_tx.borrow_mut() = false;
2875 true
2876 } else {
2877 false
2878 }
2879 })?;
2880 if cancel {
2881 log::debug!(
2882 concat!(
2883 "aborting recalculating diffs for buffer {}",
2884 "due to subsequent hunk operations",
2885 ),
2886 buffer.remote_id()
2887 );
2888 return Ok(());
2889 }
2890
2891 let unstaged_changed_range = if let Some((unstaged_diff, new_unstaged_diff)) =
2892 unstaged_diff.as_ref().zip(new_unstaged_diff.clone())
2893 {
2894 unstaged_diff.update(cx, |diff, cx| {
2895 if language_changed {
2896 diff.language_changed(cx);
2897 }
2898 diff.set_snapshot(new_unstaged_diff, &buffer, cx)
2899 })?
2900 } else {
2901 None
2902 };
2903
2904 if let Some((uncommitted_diff, new_uncommitted_diff)) =
2905 uncommitted_diff.as_ref().zip(new_uncommitted_diff.clone())
2906 {
2907 uncommitted_diff.update(cx, |diff, cx| {
2908 if language_changed {
2909 diff.language_changed(cx);
2910 }
2911 diff.set_snapshot_with_secondary(
2912 new_uncommitted_diff,
2913 &buffer,
2914 unstaged_changed_range,
2915 true,
2916 cx,
2917 );
2918 })?;
2919 }
2920
2921 log::debug!(
2922 "finished recalculating diffs for buffer {}",
2923 buffer.remote_id()
2924 );
2925
2926 if let Some(this) = this.upgrade() {
2927 this.update(cx, |this, _| {
2928 this.index_changed = false;
2929 this.head_changed = false;
2930 this.language_changed = false;
2931 *this.recalculating_tx.borrow_mut() = false;
2932 })?;
2933 }
2934
2935 Ok(())
2936 }));
2937 }
2938}
2939
2940fn make_remote_delegate(
2941 this: Entity<GitStore>,
2942 project_id: u64,
2943 repository_id: RepositoryId,
2944 askpass_id: u64,
2945 cx: &mut AsyncApp,
2946) -> AskPassDelegate {
2947 AskPassDelegate::new(cx, move |prompt, tx, cx| {
2948 this.update(cx, |this, cx| {
2949 let Some((client, _)) = this.downstream_client() else {
2950 return;
2951 };
2952 let response = client.request(proto::AskPassRequest {
2953 project_id,
2954 repository_id: repository_id.to_proto(),
2955 askpass_id,
2956 prompt,
2957 });
2958 cx.spawn(async move |_, _| {
2959 let mut response = response.await?.response;
2960 tx.send(EncryptedPassword::try_from(response.as_ref())?)
2961 .ok();
2962 response.zeroize();
2963 anyhow::Ok(())
2964 })
2965 .detach_and_log_err(cx);
2966 })
2967 .log_err();
2968 })
2969}
2970
2971impl RepositoryId {
2972 pub fn to_proto(self) -> u64 {
2973 self.0
2974 }
2975
2976 pub fn from_proto(id: u64) -> Self {
2977 RepositoryId(id)
2978 }
2979}
2980
2981impl RepositorySnapshot {
2982 fn empty(id: RepositoryId, work_directory_abs_path: Arc<Path>, path_style: PathStyle) -> Self {
2983 Self {
2984 id,
2985 statuses_by_path: Default::default(),
2986 pending_ops_by_path: Default::default(),
2987 work_directory_abs_path,
2988 branch: None,
2989 head_commit: None,
2990 scan_id: 0,
2991 merge: Default::default(),
2992 remote_origin_url: None,
2993 remote_upstream_url: None,
2994 stash_entries: Default::default(),
2995 path_style,
2996 }
2997 }
2998
2999 fn initial_update(&self, project_id: u64) -> proto::UpdateRepository {
3000 proto::UpdateRepository {
3001 branch_summary: self.branch.as_ref().map(branch_to_proto),
3002 head_commit_details: self.head_commit.as_ref().map(commit_details_to_proto),
3003 updated_statuses: self
3004 .statuses_by_path
3005 .iter()
3006 .map(|entry| entry.to_proto())
3007 .collect(),
3008 removed_statuses: Default::default(),
3009 current_merge_conflicts: self
3010 .merge
3011 .conflicted_paths
3012 .iter()
3013 .map(|repo_path| repo_path.to_proto())
3014 .collect(),
3015 merge_message: self.merge.message.as_ref().map(|msg| msg.to_string()),
3016 project_id,
3017 id: self.id.to_proto(),
3018 abs_path: self.work_directory_abs_path.to_string_lossy().into_owned(),
3019 entry_ids: vec![self.id.to_proto()],
3020 scan_id: self.scan_id,
3021 is_last_update: true,
3022 stash_entries: self
3023 .stash_entries
3024 .entries
3025 .iter()
3026 .map(stash_to_proto)
3027 .collect(),
3028 }
3029 }
3030
3031 fn build_update(&self, old: &Self, project_id: u64) -> proto::UpdateRepository {
3032 let mut updated_statuses: Vec<proto::StatusEntry> = Vec::new();
3033 let mut removed_statuses: Vec<String> = Vec::new();
3034
3035 let mut new_statuses = self.statuses_by_path.iter().peekable();
3036 let mut old_statuses = old.statuses_by_path.iter().peekable();
3037
3038 let mut current_new_entry = new_statuses.next();
3039 let mut current_old_entry = old_statuses.next();
3040 loop {
3041 match (current_new_entry, current_old_entry) {
3042 (Some(new_entry), Some(old_entry)) => {
3043 match new_entry.repo_path.cmp(&old_entry.repo_path) {
3044 Ordering::Less => {
3045 updated_statuses.push(new_entry.to_proto());
3046 current_new_entry = new_statuses.next();
3047 }
3048 Ordering::Equal => {
3049 if new_entry.status != old_entry.status {
3050 updated_statuses.push(new_entry.to_proto());
3051 }
3052 current_old_entry = old_statuses.next();
3053 current_new_entry = new_statuses.next();
3054 }
3055 Ordering::Greater => {
3056 removed_statuses.push(old_entry.repo_path.to_proto());
3057 current_old_entry = old_statuses.next();
3058 }
3059 }
3060 }
3061 (None, Some(old_entry)) => {
3062 removed_statuses.push(old_entry.repo_path.to_proto());
3063 current_old_entry = old_statuses.next();
3064 }
3065 (Some(new_entry), None) => {
3066 updated_statuses.push(new_entry.to_proto());
3067 current_new_entry = new_statuses.next();
3068 }
3069 (None, None) => break,
3070 }
3071 }
3072
3073 proto::UpdateRepository {
3074 branch_summary: self.branch.as_ref().map(branch_to_proto),
3075 head_commit_details: self.head_commit.as_ref().map(commit_details_to_proto),
3076 updated_statuses,
3077 removed_statuses,
3078 current_merge_conflicts: self
3079 .merge
3080 .conflicted_paths
3081 .iter()
3082 .map(|path| path.to_proto())
3083 .collect(),
3084 merge_message: self.merge.message.as_ref().map(|msg| msg.to_string()),
3085 project_id,
3086 id: self.id.to_proto(),
3087 abs_path: self.work_directory_abs_path.to_string_lossy().into_owned(),
3088 entry_ids: vec![],
3089 scan_id: self.scan_id,
3090 is_last_update: true,
3091 stash_entries: self
3092 .stash_entries
3093 .entries
3094 .iter()
3095 .map(stash_to_proto)
3096 .collect(),
3097 }
3098 }
3099
3100 pub fn status(&self) -> impl Iterator<Item = StatusEntry> + '_ {
3101 self.statuses_by_path.iter().cloned()
3102 }
3103
3104 pub fn status_summary(&self) -> GitSummary {
3105 self.statuses_by_path.summary().item_summary
3106 }
3107
3108 pub fn status_for_path(&self, path: &RepoPath) -> Option<StatusEntry> {
3109 self.statuses_by_path
3110 .get(&PathKey(path.0.clone()), ())
3111 .cloned()
3112 }
3113
3114 pub fn pending_ops(&self) -> impl Iterator<Item = PendingOperations> + '_ {
3115 self.pending_ops_by_path.iter().cloned()
3116 }
3117
3118 pub fn pending_ops_summary(&self) -> PendingOperationsSummary {
3119 self.pending_ops_by_path.summary().item_summary
3120 }
3121
3122 pub fn pending_ops_for_path(&self, path: &RepoPath) -> Vec<PendingOperation> {
3123 self.pending_ops_by_path
3124 .get(&PathKey(path.0.clone()), ())
3125 .map(|ops| ops.ops.clone())
3126 .unwrap_or(Vec::new())
3127 }
3128
3129 pub fn abs_path_to_repo_path(&self, abs_path: &Path) -> Option<RepoPath> {
3130 Self::abs_path_to_repo_path_inner(&self.work_directory_abs_path, abs_path, self.path_style)
3131 }
3132
3133 fn repo_path_to_abs_path(&self, repo_path: &RepoPath) -> PathBuf {
3134 self.path_style
3135 .join(&self.work_directory_abs_path, repo_path.as_std_path())
3136 .unwrap()
3137 .into()
3138 }
3139
3140 #[inline]
3141 fn abs_path_to_repo_path_inner(
3142 work_directory_abs_path: &Path,
3143 abs_path: &Path,
3144 path_style: PathStyle,
3145 ) -> Option<RepoPath> {
3146 abs_path
3147 .strip_prefix(&work_directory_abs_path)
3148 .ok()
3149 .and_then(|path| RepoPath::from_std_path(path, path_style).ok())
3150 }
3151
3152 pub fn had_conflict_on_last_merge_head_change(&self, repo_path: &RepoPath) -> bool {
3153 self.merge.conflicted_paths.contains(repo_path)
3154 }
3155
3156 pub fn has_conflict(&self, repo_path: &RepoPath) -> bool {
3157 let had_conflict_on_last_merge_head_change =
3158 self.merge.conflicted_paths.contains(repo_path);
3159 let has_conflict_currently = self
3160 .status_for_path(repo_path)
3161 .is_some_and(|entry| entry.status.is_conflicted());
3162 had_conflict_on_last_merge_head_change || has_conflict_currently
3163 }
3164
3165 /// This is the name that will be displayed in the repository selector for this repository.
3166 pub fn display_name(&self) -> SharedString {
3167 self.work_directory_abs_path
3168 .file_name()
3169 .unwrap_or_default()
3170 .to_string_lossy()
3171 .to_string()
3172 .into()
3173 }
3174}
3175
3176pub fn stash_to_proto(entry: &StashEntry) -> proto::StashEntry {
3177 proto::StashEntry {
3178 oid: entry.oid.as_bytes().to_vec(),
3179 message: entry.message.clone(),
3180 branch: entry.branch.clone(),
3181 index: entry.index as u64,
3182 timestamp: entry.timestamp,
3183 }
3184}
3185
3186pub fn proto_to_stash(entry: &proto::StashEntry) -> Result<StashEntry> {
3187 Ok(StashEntry {
3188 oid: Oid::from_bytes(&entry.oid)?,
3189 message: entry.message.clone(),
3190 index: entry.index as usize,
3191 branch: entry.branch.clone(),
3192 timestamp: entry.timestamp,
3193 })
3194}
3195
3196impl MergeDetails {
3197 async fn load(
3198 backend: &Arc<dyn GitRepository>,
3199 status: &SumTree<StatusEntry>,
3200 prev_snapshot: &RepositorySnapshot,
3201 ) -> Result<(MergeDetails, bool)> {
3202 log::debug!("load merge details");
3203 let message = backend.merge_message().await;
3204 let heads = backend
3205 .revparse_batch(vec![
3206 "MERGE_HEAD".into(),
3207 "CHERRY_PICK_HEAD".into(),
3208 "REBASE_HEAD".into(),
3209 "REVERT_HEAD".into(),
3210 "APPLY_HEAD".into(),
3211 ])
3212 .await
3213 .log_err()
3214 .unwrap_or_default()
3215 .into_iter()
3216 .map(|opt| opt.map(SharedString::from))
3217 .collect::<Vec<_>>();
3218 let merge_heads_changed = heads != prev_snapshot.merge.heads;
3219 let conflicted_paths = if merge_heads_changed {
3220 let current_conflicted_paths = TreeSet::from_ordered_entries(
3221 status
3222 .iter()
3223 .filter(|entry| entry.status.is_conflicted())
3224 .map(|entry| entry.repo_path.clone()),
3225 );
3226
3227 // It can happen that we run a scan while a lengthy merge is in progress
3228 // that will eventually result in conflicts, but before those conflicts
3229 // are reported by `git status`. Since for the moment we only care about
3230 // the merge heads state for the purposes of tracking conflicts, don't update
3231 // this state until we see some conflicts.
3232 if heads.iter().any(Option::is_some)
3233 && !prev_snapshot.merge.heads.iter().any(Option::is_some)
3234 && current_conflicted_paths.is_empty()
3235 {
3236 log::debug!("not updating merge heads because no conflicts found");
3237 return Ok((
3238 MergeDetails {
3239 message: message.map(SharedString::from),
3240 ..prev_snapshot.merge.clone()
3241 },
3242 false,
3243 ));
3244 }
3245
3246 current_conflicted_paths
3247 } else {
3248 prev_snapshot.merge.conflicted_paths.clone()
3249 };
3250 let details = MergeDetails {
3251 conflicted_paths,
3252 message: message.map(SharedString::from),
3253 heads,
3254 };
3255 Ok((details, merge_heads_changed))
3256 }
3257}
3258
3259impl Repository {
3260 pub fn snapshot(&self) -> RepositorySnapshot {
3261 self.snapshot.clone()
3262 }
3263
3264 fn local(
3265 id: RepositoryId,
3266 work_directory_abs_path: Arc<Path>,
3267 dot_git_abs_path: Arc<Path>,
3268 repository_dir_abs_path: Arc<Path>,
3269 common_dir_abs_path: Arc<Path>,
3270 project_environment: WeakEntity<ProjectEnvironment>,
3271 fs: Arc<dyn Fs>,
3272 git_store: WeakEntity<GitStore>,
3273 cx: &mut Context<Self>,
3274 ) -> Self {
3275 let snapshot =
3276 RepositorySnapshot::empty(id, work_directory_abs_path.clone(), PathStyle::local());
3277 Repository {
3278 this: cx.weak_entity(),
3279 git_store,
3280 snapshot,
3281 commit_message_buffer: None,
3282 askpass_delegates: Default::default(),
3283 paths_needing_status_update: Default::default(),
3284 latest_askpass_id: 0,
3285 job_sender: Repository::spawn_local_git_worker(
3286 work_directory_abs_path,
3287 dot_git_abs_path,
3288 repository_dir_abs_path,
3289 common_dir_abs_path,
3290 project_environment,
3291 fs,
3292 cx,
3293 ),
3294 job_id: 0,
3295 active_jobs: Default::default(),
3296 }
3297 }
3298
3299 fn remote(
3300 id: RepositoryId,
3301 work_directory_abs_path: Arc<Path>,
3302 path_style: PathStyle,
3303 project_id: ProjectId,
3304 client: AnyProtoClient,
3305 git_store: WeakEntity<GitStore>,
3306 cx: &mut Context<Self>,
3307 ) -> Self {
3308 let snapshot = RepositorySnapshot::empty(id, work_directory_abs_path, path_style);
3309 Self {
3310 this: cx.weak_entity(),
3311 snapshot,
3312 commit_message_buffer: None,
3313 git_store,
3314 paths_needing_status_update: Default::default(),
3315 job_sender: Self::spawn_remote_git_worker(project_id, client, cx),
3316 askpass_delegates: Default::default(),
3317 latest_askpass_id: 0,
3318 active_jobs: Default::default(),
3319 job_id: 0,
3320 }
3321 }
3322
3323 pub fn git_store(&self) -> Option<Entity<GitStore>> {
3324 self.git_store.upgrade()
3325 }
3326
3327 fn reload_buffer_diff_bases(&mut self, cx: &mut Context<Self>) {
3328 let this = cx.weak_entity();
3329 let git_store = self.git_store.clone();
3330 let _ = self.send_keyed_job(
3331 Some(GitJobKey::ReloadBufferDiffBases),
3332 None,
3333 |state, mut cx| async move {
3334 let RepositoryState::Local { backend, .. } = state else {
3335 log::error!("tried to recompute diffs for a non-local repository");
3336 return Ok(());
3337 };
3338
3339 let Some(this) = this.upgrade() else {
3340 return Ok(());
3341 };
3342
3343 let repo_diff_state_updates = this.update(&mut cx, |this, cx| {
3344 git_store.update(cx, |git_store, cx| {
3345 git_store
3346 .diffs
3347 .iter()
3348 .filter_map(|(buffer_id, diff_state)| {
3349 let buffer_store = git_store.buffer_store.read(cx);
3350 let buffer = buffer_store.get(*buffer_id)?;
3351 let file = File::from_dyn(buffer.read(cx).file())?;
3352 let abs_path = file.worktree.read(cx).absolutize(&file.path);
3353 let repo_path = this.abs_path_to_repo_path(&abs_path)?;
3354 log::debug!(
3355 "start reload diff bases for repo path {}",
3356 repo_path.as_unix_str()
3357 );
3358 diff_state.update(cx, |diff_state, _| {
3359 let has_unstaged_diff = diff_state
3360 .unstaged_diff
3361 .as_ref()
3362 .is_some_and(|diff| diff.is_upgradable());
3363 let has_uncommitted_diff = diff_state
3364 .uncommitted_diff
3365 .as_ref()
3366 .is_some_and(|set| set.is_upgradable());
3367
3368 Some((
3369 buffer,
3370 repo_path,
3371 has_unstaged_diff.then(|| diff_state.index_text.clone()),
3372 has_uncommitted_diff.then(|| diff_state.head_text.clone()),
3373 ))
3374 })
3375 })
3376 .collect::<Vec<_>>()
3377 })
3378 })??;
3379
3380 let buffer_diff_base_changes = cx
3381 .background_spawn(async move {
3382 let mut changes = Vec::new();
3383 for (buffer, repo_path, current_index_text, current_head_text) in
3384 &repo_diff_state_updates
3385 {
3386 let index_text = if current_index_text.is_some() {
3387 backend.load_index_text(repo_path.clone()).await
3388 } else {
3389 None
3390 };
3391 let head_text = if current_head_text.is_some() {
3392 backend.load_committed_text(repo_path.clone()).await
3393 } else {
3394 None
3395 };
3396
3397 let change =
3398 match (current_index_text.as_ref(), current_head_text.as_ref()) {
3399 (Some(current_index), Some(current_head)) => {
3400 let index_changed =
3401 index_text.as_ref() != current_index.as_deref();
3402 let head_changed =
3403 head_text.as_ref() != current_head.as_deref();
3404 if index_changed && head_changed {
3405 if index_text == head_text {
3406 Some(DiffBasesChange::SetBoth(head_text))
3407 } else {
3408 Some(DiffBasesChange::SetEach {
3409 index: index_text,
3410 head: head_text,
3411 })
3412 }
3413 } else if index_changed {
3414 Some(DiffBasesChange::SetIndex(index_text))
3415 } else if head_changed {
3416 Some(DiffBasesChange::SetHead(head_text))
3417 } else {
3418 None
3419 }
3420 }
3421 (Some(current_index), None) => {
3422 let index_changed =
3423 index_text.as_ref() != current_index.as_deref();
3424 index_changed
3425 .then_some(DiffBasesChange::SetIndex(index_text))
3426 }
3427 (None, Some(current_head)) => {
3428 let head_changed =
3429 head_text.as_ref() != current_head.as_deref();
3430 head_changed.then_some(DiffBasesChange::SetHead(head_text))
3431 }
3432 (None, None) => None,
3433 };
3434
3435 changes.push((buffer.clone(), change))
3436 }
3437 changes
3438 })
3439 .await;
3440
3441 git_store.update(&mut cx, |git_store, cx| {
3442 for (buffer, diff_bases_change) in buffer_diff_base_changes {
3443 let buffer_snapshot = buffer.read(cx).text_snapshot();
3444 let buffer_id = buffer_snapshot.remote_id();
3445 let Some(diff_state) = git_store.diffs.get(&buffer_id) else {
3446 continue;
3447 };
3448
3449 let downstream_client = git_store.downstream_client();
3450 diff_state.update(cx, |diff_state, cx| {
3451 use proto::update_diff_bases::Mode;
3452
3453 if let Some((diff_bases_change, (client, project_id))) =
3454 diff_bases_change.clone().zip(downstream_client)
3455 {
3456 let (staged_text, committed_text, mode) = match diff_bases_change {
3457 DiffBasesChange::SetIndex(index) => {
3458 (index, None, Mode::IndexOnly)
3459 }
3460 DiffBasesChange::SetHead(head) => (None, head, Mode::HeadOnly),
3461 DiffBasesChange::SetEach { index, head } => {
3462 (index, head, Mode::IndexAndHead)
3463 }
3464 DiffBasesChange::SetBoth(text) => {
3465 (None, text, Mode::IndexMatchesHead)
3466 }
3467 };
3468 client
3469 .send(proto::UpdateDiffBases {
3470 project_id: project_id.to_proto(),
3471 buffer_id: buffer_id.to_proto(),
3472 staged_text,
3473 committed_text,
3474 mode: mode as i32,
3475 })
3476 .log_err();
3477 }
3478
3479 diff_state.diff_bases_changed(buffer_snapshot, diff_bases_change, cx);
3480 });
3481 }
3482 })
3483 },
3484 );
3485 }
3486
3487 pub fn send_job<F, Fut, R>(
3488 &mut self,
3489 status: Option<SharedString>,
3490 job: F,
3491 ) -> oneshot::Receiver<R>
3492 where
3493 F: FnOnce(RepositoryState, AsyncApp) -> Fut + 'static,
3494 Fut: Future<Output = R> + 'static,
3495 R: Send + 'static,
3496 {
3497 self.send_keyed_job(None, status, job)
3498 }
3499
3500 fn send_keyed_job<F, Fut, R>(
3501 &mut self,
3502 key: Option<GitJobKey>,
3503 status: Option<SharedString>,
3504 job: F,
3505 ) -> oneshot::Receiver<R>
3506 where
3507 F: FnOnce(RepositoryState, AsyncApp) -> Fut + 'static,
3508 Fut: Future<Output = R> + 'static,
3509 R: Send + 'static,
3510 {
3511 let (result_tx, result_rx) = futures::channel::oneshot::channel();
3512 let job_id = post_inc(&mut self.job_id);
3513 let this = self.this.clone();
3514 self.job_sender
3515 .unbounded_send(GitJob {
3516 key,
3517 job: Box::new(move |state, cx: &mut AsyncApp| {
3518 let job = job(state, cx.clone());
3519 cx.spawn(async move |cx| {
3520 if let Some(s) = status.clone() {
3521 this.update(cx, |this, cx| {
3522 this.active_jobs.insert(
3523 job_id,
3524 JobInfo {
3525 start: Instant::now(),
3526 message: s.clone(),
3527 },
3528 );
3529
3530 cx.notify();
3531 })
3532 .ok();
3533 }
3534 let result = job.await;
3535
3536 this.update(cx, |this, cx| {
3537 this.active_jobs.remove(&job_id);
3538 cx.notify();
3539 })
3540 .ok();
3541
3542 result_tx.send(result).ok();
3543 })
3544 }),
3545 })
3546 .ok();
3547 result_rx
3548 }
3549
3550 pub fn set_as_active_repository(&self, cx: &mut Context<Self>) {
3551 let Some(git_store) = self.git_store.upgrade() else {
3552 return;
3553 };
3554 let entity = cx.entity();
3555 git_store.update(cx, |git_store, cx| {
3556 let Some((&id, _)) = git_store
3557 .repositories
3558 .iter()
3559 .find(|(_, handle)| *handle == &entity)
3560 else {
3561 return;
3562 };
3563 git_store.active_repo_id = Some(id);
3564 cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id)));
3565 });
3566 }
3567
3568 pub fn cached_status(&self) -> impl '_ + Iterator<Item = StatusEntry> {
3569 self.snapshot.status()
3570 }
3571
3572 pub fn cached_stash(&self) -> GitStash {
3573 self.snapshot.stash_entries.clone()
3574 }
3575
3576 pub fn repo_path_to_project_path(&self, path: &RepoPath, cx: &App) -> Option<ProjectPath> {
3577 let git_store = self.git_store.upgrade()?;
3578 let worktree_store = git_store.read(cx).worktree_store.read(cx);
3579 let abs_path = self.snapshot.repo_path_to_abs_path(path);
3580 let abs_path = SanitizedPath::new(&abs_path);
3581 let (worktree, relative_path) = worktree_store.find_worktree(abs_path, cx)?;
3582 Some(ProjectPath {
3583 worktree_id: worktree.read(cx).id(),
3584 path: relative_path,
3585 })
3586 }
3587
3588 pub fn project_path_to_repo_path(&self, path: &ProjectPath, cx: &App) -> Option<RepoPath> {
3589 let git_store = self.git_store.upgrade()?;
3590 let worktree_store = git_store.read(cx).worktree_store.read(cx);
3591 let abs_path = worktree_store.absolutize(path, cx)?;
3592 self.snapshot.abs_path_to_repo_path(&abs_path)
3593 }
3594
3595 pub fn contains_sub_repo(&self, other: &Entity<Self>, cx: &App) -> bool {
3596 other
3597 .read(cx)
3598 .snapshot
3599 .work_directory_abs_path
3600 .starts_with(&self.snapshot.work_directory_abs_path)
3601 }
3602
3603 pub fn open_commit_buffer(
3604 &mut self,
3605 languages: Option<Arc<LanguageRegistry>>,
3606 buffer_store: Entity<BufferStore>,
3607 cx: &mut Context<Self>,
3608 ) -> Task<Result<Entity<Buffer>>> {
3609 let id = self.id;
3610 if let Some(buffer) = self.commit_message_buffer.clone() {
3611 return Task::ready(Ok(buffer));
3612 }
3613 let this = cx.weak_entity();
3614
3615 let rx = self.send_job(None, move |state, mut cx| async move {
3616 let Some(this) = this.upgrade() else {
3617 bail!("git store was dropped");
3618 };
3619 match state {
3620 RepositoryState::Local { .. } => {
3621 this.update(&mut cx, |_, cx| {
3622 Self::open_local_commit_buffer(languages, buffer_store, cx)
3623 })?
3624 .await
3625 }
3626 RepositoryState::Remote { project_id, client } => {
3627 let request = client.request(proto::OpenCommitMessageBuffer {
3628 project_id: project_id.0,
3629 repository_id: id.to_proto(),
3630 });
3631 let response = request.await.context("requesting to open commit buffer")?;
3632 let buffer_id = BufferId::new(response.buffer_id)?;
3633 let buffer = buffer_store
3634 .update(&mut cx, |buffer_store, cx| {
3635 buffer_store.wait_for_remote_buffer(buffer_id, cx)
3636 })?
3637 .await?;
3638 if let Some(language_registry) = languages {
3639 let git_commit_language =
3640 language_registry.language_for_name("Git Commit").await?;
3641 buffer.update(&mut cx, |buffer, cx| {
3642 buffer.set_language(Some(git_commit_language), cx);
3643 })?;
3644 }
3645 this.update(&mut cx, |this, _| {
3646 this.commit_message_buffer = Some(buffer.clone());
3647 })?;
3648 Ok(buffer)
3649 }
3650 }
3651 });
3652
3653 cx.spawn(|_, _: &mut AsyncApp| async move { rx.await? })
3654 }
3655
3656 fn open_local_commit_buffer(
3657 language_registry: Option<Arc<LanguageRegistry>>,
3658 buffer_store: Entity<BufferStore>,
3659 cx: &mut Context<Self>,
3660 ) -> Task<Result<Entity<Buffer>>> {
3661 cx.spawn(async move |repository, cx| {
3662 let buffer = buffer_store
3663 .update(cx, |buffer_store, cx| buffer_store.create_buffer(false, cx))?
3664 .await?;
3665
3666 if let Some(language_registry) = language_registry {
3667 let git_commit_language = language_registry.language_for_name("Git Commit").await?;
3668 buffer.update(cx, |buffer, cx| {
3669 buffer.set_language(Some(git_commit_language), cx);
3670 })?;
3671 }
3672
3673 repository.update(cx, |repository, _| {
3674 repository.commit_message_buffer = Some(buffer.clone());
3675 })?;
3676 Ok(buffer)
3677 })
3678 }
3679
3680 pub fn checkout_files(
3681 &mut self,
3682 commit: &str,
3683 paths: Vec<RepoPath>,
3684 _cx: &mut App,
3685 ) -> oneshot::Receiver<Result<()>> {
3686 let commit = commit.to_string();
3687 let id = self.id;
3688
3689 self.send_job(
3690 Some(format!("git checkout {}", commit).into()),
3691 move |git_repo, _| async move {
3692 match git_repo {
3693 RepositoryState::Local {
3694 backend,
3695 environment,
3696 ..
3697 } => {
3698 backend
3699 .checkout_files(commit, paths, environment.clone())
3700 .await
3701 }
3702 RepositoryState::Remote { project_id, client } => {
3703 client
3704 .request(proto::GitCheckoutFiles {
3705 project_id: project_id.0,
3706 repository_id: id.to_proto(),
3707 commit,
3708 paths: paths.into_iter().map(|p| p.to_proto()).collect(),
3709 })
3710 .await?;
3711
3712 Ok(())
3713 }
3714 }
3715 },
3716 )
3717 }
3718
3719 pub fn reset(
3720 &mut self,
3721 commit: String,
3722 reset_mode: ResetMode,
3723 _cx: &mut App,
3724 ) -> oneshot::Receiver<Result<()>> {
3725 let id = self.id;
3726
3727 self.send_job(None, move |git_repo, _| async move {
3728 match git_repo {
3729 RepositoryState::Local {
3730 backend,
3731 environment,
3732 ..
3733 } => backend.reset(commit, reset_mode, environment).await,
3734 RepositoryState::Remote { project_id, client } => {
3735 client
3736 .request(proto::GitReset {
3737 project_id: project_id.0,
3738 repository_id: id.to_proto(),
3739 commit,
3740 mode: match reset_mode {
3741 ResetMode::Soft => git_reset::ResetMode::Soft.into(),
3742 ResetMode::Mixed => git_reset::ResetMode::Mixed.into(),
3743 },
3744 })
3745 .await?;
3746
3747 Ok(())
3748 }
3749 }
3750 })
3751 }
3752
3753 pub fn show(&mut self, commit: String) -> oneshot::Receiver<Result<CommitDetails>> {
3754 let id = self.id;
3755 self.send_job(None, move |git_repo, _cx| async move {
3756 match git_repo {
3757 RepositoryState::Local { backend, .. } => backend.show(commit).await,
3758 RepositoryState::Remote { project_id, client } => {
3759 let resp = client
3760 .request(proto::GitShow {
3761 project_id: project_id.0,
3762 repository_id: id.to_proto(),
3763 commit,
3764 })
3765 .await?;
3766
3767 Ok(CommitDetails {
3768 sha: resp.sha.into(),
3769 message: resp.message.into(),
3770 commit_timestamp: resp.commit_timestamp,
3771 author_email: resp.author_email.into(),
3772 author_name: resp.author_name.into(),
3773 })
3774 }
3775 }
3776 })
3777 }
3778
3779 pub fn load_commit_diff(&mut self, commit: String) -> oneshot::Receiver<Result<CommitDiff>> {
3780 let id = self.id;
3781 self.send_job(None, move |git_repo, cx| async move {
3782 match git_repo {
3783 RepositoryState::Local { backend, .. } => backend.load_commit(commit, cx).await,
3784 RepositoryState::Remote {
3785 client, project_id, ..
3786 } => {
3787 let response = client
3788 .request(proto::LoadCommitDiff {
3789 project_id: project_id.0,
3790 repository_id: id.to_proto(),
3791 commit,
3792 })
3793 .await?;
3794 Ok(CommitDiff {
3795 files: response
3796 .files
3797 .into_iter()
3798 .map(|file| {
3799 Ok(CommitFile {
3800 path: RepoPath::from_proto(&file.path)?,
3801 old_text: file.old_text,
3802 new_text: file.new_text,
3803 })
3804 })
3805 .collect::<Result<Vec<_>>>()?,
3806 })
3807 }
3808 }
3809 })
3810 }
3811
3812 fn buffer_store(&self, cx: &App) -> Option<Entity<BufferStore>> {
3813 Some(self.git_store.upgrade()?.read(cx).buffer_store.clone())
3814 }
3815
3816 fn save_buffers<'a>(
3817 &self,
3818 entries: impl IntoIterator<Item = &'a RepoPath>,
3819 cx: &mut Context<Self>,
3820 ) -> Vec<Task<anyhow::Result<()>>> {
3821 let mut save_futures = Vec::new();
3822 if let Some(buffer_store) = self.buffer_store(cx) {
3823 buffer_store.update(cx, |buffer_store, cx| {
3824 for path in entries {
3825 let Some(project_path) = self.repo_path_to_project_path(path, cx) else {
3826 continue;
3827 };
3828 if let Some(buffer) = buffer_store.get_by_path(&project_path)
3829 && buffer
3830 .read(cx)
3831 .file()
3832 .is_some_and(|file| file.disk_state().exists())
3833 && buffer.read(cx).has_unsaved_edits()
3834 {
3835 save_futures.push(buffer_store.save_buffer(buffer, cx));
3836 }
3837 }
3838 })
3839 }
3840 save_futures
3841 }
3842
3843 pub fn stage_entries(
3844 &self,
3845 entries: Vec<RepoPath>,
3846 cx: &mut Context<Self>,
3847 ) -> Task<anyhow::Result<()>> {
3848 if entries.is_empty() {
3849 return Task::ready(Ok(()));
3850 }
3851 let id = self.id;
3852 let save_tasks = self.save_buffers(&entries, cx);
3853 let paths = entries
3854 .iter()
3855 .map(|p| p.as_unix_str())
3856 .collect::<Vec<_>>()
3857 .join(" ");
3858 let status = format!("git add {paths}");
3859 let job_key = match entries.len() {
3860 1 => Some(GitJobKey::WriteIndex(entries[0].clone())),
3861 _ => None,
3862 };
3863
3864 cx.spawn(async move |this, cx| {
3865 for save_task in save_tasks {
3866 save_task.await?;
3867 }
3868
3869 this.update(cx, |this, _| {
3870 this.send_keyed_job(
3871 job_key,
3872 Some(status.into()),
3873 move |git_repo, _cx| async move {
3874 match git_repo {
3875 RepositoryState::Local {
3876 backend,
3877 environment,
3878 ..
3879 } => backend.stage_paths(entries, environment.clone()).await,
3880 RepositoryState::Remote { project_id, client } => {
3881 client
3882 .request(proto::Stage {
3883 project_id: project_id.0,
3884 repository_id: id.to_proto(),
3885 paths: entries
3886 .into_iter()
3887 .map(|repo_path| repo_path.to_proto())
3888 .collect(),
3889 })
3890 .await
3891 .context("sending stage request")?;
3892
3893 Ok(())
3894 }
3895 }
3896 },
3897 )
3898 })?
3899 .await??;
3900
3901 Ok(())
3902 })
3903 }
3904
3905 pub fn unstage_entries(
3906 &self,
3907 entries: Vec<RepoPath>,
3908 cx: &mut Context<Self>,
3909 ) -> Task<anyhow::Result<()>> {
3910 if entries.is_empty() {
3911 return Task::ready(Ok(()));
3912 }
3913 let id = self.id;
3914 let save_tasks = self.save_buffers(&entries, cx);
3915 let paths = entries
3916 .iter()
3917 .map(|p| p.as_unix_str())
3918 .collect::<Vec<_>>()
3919 .join(" ");
3920 let status = format!("git reset {paths}");
3921 let job_key = match entries.len() {
3922 1 => Some(GitJobKey::WriteIndex(entries[0].clone())),
3923 _ => None,
3924 };
3925
3926 cx.spawn(async move |this, cx| {
3927 for save_task in save_tasks {
3928 save_task.await?;
3929 }
3930
3931 this.update(cx, |this, _| {
3932 this.send_keyed_job(
3933 job_key,
3934 Some(status.into()),
3935 move |git_repo, _cx| async move {
3936 match git_repo {
3937 RepositoryState::Local {
3938 backend,
3939 environment,
3940 ..
3941 } => backend.unstage_paths(entries, environment).await,
3942 RepositoryState::Remote { project_id, client } => {
3943 client
3944 .request(proto::Unstage {
3945 project_id: project_id.0,
3946 repository_id: id.to_proto(),
3947 paths: entries
3948 .into_iter()
3949 .map(|repo_path| repo_path.to_proto())
3950 .collect(),
3951 })
3952 .await
3953 .context("sending unstage request")?;
3954
3955 Ok(())
3956 }
3957 }
3958 },
3959 )
3960 })?
3961 .await??;
3962
3963 Ok(())
3964 })
3965 }
3966
3967 pub fn stage_all(&self, cx: &mut Context<Self>) -> Task<anyhow::Result<()>> {
3968 let to_stage = self
3969 .cached_status()
3970 .filter(|entry| !entry.status.staging().is_fully_staged())
3971 .map(|entry| entry.repo_path)
3972 .collect();
3973 self.stage_entries(to_stage, cx)
3974 }
3975
3976 pub fn unstage_all(&self, cx: &mut Context<Self>) -> Task<anyhow::Result<()>> {
3977 let to_unstage = self
3978 .cached_status()
3979 .filter(|entry| entry.status.staging().has_staged())
3980 .map(|entry| entry.repo_path)
3981 .collect();
3982 self.unstage_entries(to_unstage, cx)
3983 }
3984
3985 pub fn stash_all(&mut self, cx: &mut Context<Self>) -> Task<anyhow::Result<()>> {
3986 let to_stash = self.cached_status().map(|entry| entry.repo_path).collect();
3987
3988 self.stash_entries(to_stash, cx)
3989 }
3990
3991 pub fn stash_entries(
3992 &mut self,
3993 entries: Vec<RepoPath>,
3994 cx: &mut Context<Self>,
3995 ) -> Task<anyhow::Result<()>> {
3996 let id = self.id;
3997
3998 cx.spawn(async move |this, cx| {
3999 this.update(cx, |this, _| {
4000 this.send_job(None, move |git_repo, _cx| async move {
4001 match git_repo {
4002 RepositoryState::Local {
4003 backend,
4004 environment,
4005 ..
4006 } => backend.stash_paths(entries, environment).await,
4007 RepositoryState::Remote { project_id, client } => {
4008 client
4009 .request(proto::Stash {
4010 project_id: project_id.0,
4011 repository_id: id.to_proto(),
4012 paths: entries
4013 .into_iter()
4014 .map(|repo_path| repo_path.to_proto())
4015 .collect(),
4016 })
4017 .await
4018 .context("sending stash request")?;
4019 Ok(())
4020 }
4021 }
4022 })
4023 })?
4024 .await??;
4025 Ok(())
4026 })
4027 }
4028
4029 pub fn stash_pop(
4030 &mut self,
4031 index: Option<usize>,
4032 cx: &mut Context<Self>,
4033 ) -> Task<anyhow::Result<()>> {
4034 let id = self.id;
4035 cx.spawn(async move |this, cx| {
4036 this.update(cx, |this, _| {
4037 this.send_job(None, move |git_repo, _cx| async move {
4038 match git_repo {
4039 RepositoryState::Local {
4040 backend,
4041 environment,
4042 ..
4043 } => backend.stash_pop(index, environment).await,
4044 RepositoryState::Remote { project_id, client } => {
4045 client
4046 .request(proto::StashPop {
4047 project_id: project_id.0,
4048 repository_id: id.to_proto(),
4049 stash_index: index.map(|i| i as u64),
4050 })
4051 .await
4052 .context("sending stash pop request")?;
4053 Ok(())
4054 }
4055 }
4056 })
4057 })?
4058 .await??;
4059 Ok(())
4060 })
4061 }
4062
4063 pub fn stash_apply(
4064 &mut self,
4065 index: Option<usize>,
4066 cx: &mut Context<Self>,
4067 ) -> Task<anyhow::Result<()>> {
4068 let id = self.id;
4069 cx.spawn(async move |this, cx| {
4070 this.update(cx, |this, _| {
4071 this.send_job(None, move |git_repo, _cx| async move {
4072 match git_repo {
4073 RepositoryState::Local {
4074 backend,
4075 environment,
4076 ..
4077 } => backend.stash_apply(index, environment).await,
4078 RepositoryState::Remote { project_id, client } => {
4079 client
4080 .request(proto::StashApply {
4081 project_id: project_id.0,
4082 repository_id: id.to_proto(),
4083 stash_index: index.map(|i| i as u64),
4084 })
4085 .await
4086 .context("sending stash apply request")?;
4087 Ok(())
4088 }
4089 }
4090 })
4091 })?
4092 .await??;
4093 Ok(())
4094 })
4095 }
4096
4097 pub fn stash_drop(
4098 &mut self,
4099 index: Option<usize>,
4100 cx: &mut Context<Self>,
4101 ) -> oneshot::Receiver<anyhow::Result<()>> {
4102 let id = self.id;
4103 let updates_tx = self
4104 .git_store()
4105 .and_then(|git_store| match &git_store.read(cx).state {
4106 GitStoreState::Local { downstream, .. } => downstream
4107 .as_ref()
4108 .map(|downstream| downstream.updates_tx.clone()),
4109 _ => None,
4110 });
4111 let this = cx.weak_entity();
4112 self.send_job(None, move |git_repo, mut cx| async move {
4113 match git_repo {
4114 RepositoryState::Local {
4115 backend,
4116 environment,
4117 ..
4118 } => {
4119 // TODO would be nice to not have to do this manually
4120 let result = backend.stash_drop(index, environment).await;
4121 if result.is_ok()
4122 && let Ok(stash_entries) = backend.stash_entries().await
4123 {
4124 let snapshot = this.update(&mut cx, |this, cx| {
4125 this.snapshot.stash_entries = stash_entries;
4126 cx.emit(RepositoryEvent::StashEntriesChanged);
4127 this.snapshot.clone()
4128 })?;
4129 if let Some(updates_tx) = updates_tx {
4130 updates_tx
4131 .unbounded_send(DownstreamUpdate::UpdateRepository(snapshot))
4132 .ok();
4133 }
4134 }
4135
4136 result
4137 }
4138 RepositoryState::Remote { project_id, client } => {
4139 client
4140 .request(proto::StashDrop {
4141 project_id: project_id.0,
4142 repository_id: id.to_proto(),
4143 stash_index: index.map(|i| i as u64),
4144 })
4145 .await
4146 .context("sending stash pop request")?;
4147 Ok(())
4148 }
4149 }
4150 })
4151 }
4152
4153 pub fn commit(
4154 &mut self,
4155 message: SharedString,
4156 name_and_email: Option<(SharedString, SharedString)>,
4157 options: CommitOptions,
4158 _cx: &mut App,
4159 ) -> oneshot::Receiver<Result<()>> {
4160 let id = self.id;
4161
4162 self.send_job(Some("git commit".into()), move |git_repo, _cx| async move {
4163 match git_repo {
4164 RepositoryState::Local {
4165 backend,
4166 environment,
4167 ..
4168 } => {
4169 backend
4170 .commit(message, name_and_email, options, environment)
4171 .await
4172 }
4173 RepositoryState::Remote { project_id, client } => {
4174 let (name, email) = name_and_email.unzip();
4175 client
4176 .request(proto::Commit {
4177 project_id: project_id.0,
4178 repository_id: id.to_proto(),
4179 message: String::from(message),
4180 name: name.map(String::from),
4181 email: email.map(String::from),
4182 options: Some(proto::commit::CommitOptions {
4183 amend: options.amend,
4184 signoff: options.signoff,
4185 }),
4186 })
4187 .await
4188 .context("sending commit request")?;
4189
4190 Ok(())
4191 }
4192 }
4193 })
4194 }
4195
4196 pub fn fetch(
4197 &mut self,
4198 fetch_options: FetchOptions,
4199 askpass: AskPassDelegate,
4200 _cx: &mut App,
4201 ) -> oneshot::Receiver<Result<RemoteCommandOutput>> {
4202 let askpass_delegates = self.askpass_delegates.clone();
4203 let askpass_id = util::post_inc(&mut self.latest_askpass_id);
4204 let id = self.id;
4205
4206 self.send_job(Some("git fetch".into()), move |git_repo, cx| async move {
4207 match git_repo {
4208 RepositoryState::Local {
4209 backend,
4210 environment,
4211 ..
4212 } => backend.fetch(fetch_options, askpass, environment, cx).await,
4213 RepositoryState::Remote { project_id, client } => {
4214 askpass_delegates.lock().insert(askpass_id, askpass);
4215 let _defer = util::defer(|| {
4216 let askpass_delegate = askpass_delegates.lock().remove(&askpass_id);
4217 debug_assert!(askpass_delegate.is_some());
4218 });
4219
4220 let response = client
4221 .request(proto::Fetch {
4222 project_id: project_id.0,
4223 repository_id: id.to_proto(),
4224 askpass_id,
4225 remote: fetch_options.to_proto(),
4226 })
4227 .await
4228 .context("sending fetch request")?;
4229
4230 Ok(RemoteCommandOutput {
4231 stdout: response.stdout,
4232 stderr: response.stderr,
4233 })
4234 }
4235 }
4236 })
4237 }
4238
4239 pub fn push(
4240 &mut self,
4241 branch: SharedString,
4242 remote: SharedString,
4243 options: Option<PushOptions>,
4244 askpass: AskPassDelegate,
4245 cx: &mut Context<Self>,
4246 ) -> oneshot::Receiver<Result<RemoteCommandOutput>> {
4247 let askpass_delegates = self.askpass_delegates.clone();
4248 let askpass_id = util::post_inc(&mut self.latest_askpass_id);
4249 let id = self.id;
4250
4251 let args = options
4252 .map(|option| match option {
4253 PushOptions::SetUpstream => " --set-upstream",
4254 PushOptions::Force => " --force-with-lease",
4255 })
4256 .unwrap_or("");
4257
4258 let updates_tx = self
4259 .git_store()
4260 .and_then(|git_store| match &git_store.read(cx).state {
4261 GitStoreState::Local { downstream, .. } => downstream
4262 .as_ref()
4263 .map(|downstream| downstream.updates_tx.clone()),
4264 _ => None,
4265 });
4266
4267 let this = cx.weak_entity();
4268 self.send_job(
4269 Some(format!("git push {} {} {}", args, remote, branch).into()),
4270 move |git_repo, mut cx| async move {
4271 match git_repo {
4272 RepositoryState::Local {
4273 backend,
4274 environment,
4275 ..
4276 } => {
4277 let result = backend
4278 .push(
4279 branch.to_string(),
4280 remote.to_string(),
4281 options,
4282 askpass,
4283 environment.clone(),
4284 cx.clone(),
4285 )
4286 .await;
4287 // TODO would be nice to not have to do this manually
4288 if result.is_ok() {
4289 let branches = backend.branches().await?;
4290 let branch = branches.into_iter().find(|branch| branch.is_head);
4291 log::info!("head branch after scan is {branch:?}");
4292 let snapshot = this.update(&mut cx, |this, cx| {
4293 this.snapshot.branch = branch;
4294 cx.emit(RepositoryEvent::BranchChanged);
4295 this.snapshot.clone()
4296 })?;
4297 if let Some(updates_tx) = updates_tx {
4298 updates_tx
4299 .unbounded_send(DownstreamUpdate::UpdateRepository(snapshot))
4300 .ok();
4301 }
4302 }
4303 result
4304 }
4305 RepositoryState::Remote { project_id, client } => {
4306 askpass_delegates.lock().insert(askpass_id, askpass);
4307 let _defer = util::defer(|| {
4308 let askpass_delegate = askpass_delegates.lock().remove(&askpass_id);
4309 debug_assert!(askpass_delegate.is_some());
4310 });
4311 let response = client
4312 .request(proto::Push {
4313 project_id: project_id.0,
4314 repository_id: id.to_proto(),
4315 askpass_id,
4316 branch_name: branch.to_string(),
4317 remote_name: remote.to_string(),
4318 options: options.map(|options| match options {
4319 PushOptions::Force => proto::push::PushOptions::Force,
4320 PushOptions::SetUpstream => {
4321 proto::push::PushOptions::SetUpstream
4322 }
4323 }
4324 as i32),
4325 })
4326 .await
4327 .context("sending push request")?;
4328
4329 Ok(RemoteCommandOutput {
4330 stdout: response.stdout,
4331 stderr: response.stderr,
4332 })
4333 }
4334 }
4335 },
4336 )
4337 }
4338
4339 pub fn pull(
4340 &mut self,
4341 branch: SharedString,
4342 remote: SharedString,
4343 askpass: AskPassDelegate,
4344 _cx: &mut App,
4345 ) -> oneshot::Receiver<Result<RemoteCommandOutput>> {
4346 let askpass_delegates = self.askpass_delegates.clone();
4347 let askpass_id = util::post_inc(&mut self.latest_askpass_id);
4348 let id = self.id;
4349
4350 self.send_job(
4351 Some(format!("git pull {} {}", remote, branch).into()),
4352 move |git_repo, cx| async move {
4353 match git_repo {
4354 RepositoryState::Local {
4355 backend,
4356 environment,
4357 ..
4358 } => {
4359 backend
4360 .pull(
4361 branch.to_string(),
4362 remote.to_string(),
4363 askpass,
4364 environment.clone(),
4365 cx,
4366 )
4367 .await
4368 }
4369 RepositoryState::Remote { project_id, client } => {
4370 askpass_delegates.lock().insert(askpass_id, askpass);
4371 let _defer = util::defer(|| {
4372 let askpass_delegate = askpass_delegates.lock().remove(&askpass_id);
4373 debug_assert!(askpass_delegate.is_some());
4374 });
4375 let response = client
4376 .request(proto::Pull {
4377 project_id: project_id.0,
4378 repository_id: id.to_proto(),
4379 askpass_id,
4380 branch_name: branch.to_string(),
4381 remote_name: remote.to_string(),
4382 })
4383 .await
4384 .context("sending pull request")?;
4385
4386 Ok(RemoteCommandOutput {
4387 stdout: response.stdout,
4388 stderr: response.stderr,
4389 })
4390 }
4391 }
4392 },
4393 )
4394 }
4395
4396 fn spawn_set_index_text_job(
4397 &mut self,
4398 path: RepoPath,
4399 content: Option<String>,
4400 hunk_staging_operation_count: Option<usize>,
4401 cx: &mut Context<Self>,
4402 ) -> oneshot::Receiver<anyhow::Result<()>> {
4403 let id = self.id;
4404 let this = cx.weak_entity();
4405 let git_store = self.git_store.clone();
4406 self.send_keyed_job(
4407 Some(GitJobKey::WriteIndex(path.clone())),
4408 None,
4409 move |git_repo, mut cx| async move {
4410 log::debug!(
4411 "start updating index text for buffer {}",
4412 path.as_unix_str()
4413 );
4414 match git_repo {
4415 RepositoryState::Local {
4416 backend,
4417 environment,
4418 ..
4419 } => {
4420 backend
4421 .set_index_text(path.clone(), content, environment.clone())
4422 .await?;
4423 }
4424 RepositoryState::Remote { project_id, client } => {
4425 client
4426 .request(proto::SetIndexText {
4427 project_id: project_id.0,
4428 repository_id: id.to_proto(),
4429 path: path.to_proto(),
4430 text: content,
4431 })
4432 .await?;
4433 }
4434 }
4435 log::debug!(
4436 "finish updating index text for buffer {}",
4437 path.as_unix_str()
4438 );
4439
4440 if let Some(hunk_staging_operation_count) = hunk_staging_operation_count {
4441 let project_path = this
4442 .read_with(&cx, |this, cx| this.repo_path_to_project_path(&path, cx))
4443 .ok()
4444 .flatten();
4445 git_store.update(&mut cx, |git_store, cx| {
4446 let buffer_id = git_store
4447 .buffer_store
4448 .read(cx)
4449 .get_by_path(&project_path?)?
4450 .read(cx)
4451 .remote_id();
4452 let diff_state = git_store.diffs.get(&buffer_id)?;
4453 diff_state.update(cx, |diff_state, _| {
4454 diff_state.hunk_staging_operation_count_as_of_write =
4455 hunk_staging_operation_count;
4456 });
4457 Some(())
4458 })?;
4459 }
4460 Ok(())
4461 },
4462 )
4463 }
4464
4465 pub fn get_remotes(
4466 &mut self,
4467 branch_name: Option<String>,
4468 ) -> oneshot::Receiver<Result<Vec<Remote>>> {
4469 let id = self.id;
4470 self.send_job(None, move |repo, _cx| async move {
4471 match repo {
4472 RepositoryState::Local { backend, .. } => backend.get_remotes(branch_name).await,
4473 RepositoryState::Remote { project_id, client } => {
4474 let response = client
4475 .request(proto::GetRemotes {
4476 project_id: project_id.0,
4477 repository_id: id.to_proto(),
4478 branch_name,
4479 })
4480 .await?;
4481
4482 let remotes = response
4483 .remotes
4484 .into_iter()
4485 .map(|remotes| git::repository::Remote {
4486 name: remotes.name.into(),
4487 })
4488 .collect();
4489
4490 Ok(remotes)
4491 }
4492 }
4493 })
4494 }
4495
4496 pub fn branches(&mut self) -> oneshot::Receiver<Result<Vec<Branch>>> {
4497 let id = self.id;
4498 self.send_job(None, move |repo, _| async move {
4499 match repo {
4500 RepositoryState::Local { backend, .. } => backend.branches().await,
4501 RepositoryState::Remote { project_id, client } => {
4502 let response = client
4503 .request(proto::GitGetBranches {
4504 project_id: project_id.0,
4505 repository_id: id.to_proto(),
4506 })
4507 .await?;
4508
4509 let branches = response
4510 .branches
4511 .into_iter()
4512 .map(|branch| proto_to_branch(&branch))
4513 .collect();
4514
4515 Ok(branches)
4516 }
4517 }
4518 })
4519 }
4520
4521 pub fn default_branch(&mut self) -> oneshot::Receiver<Result<Option<SharedString>>> {
4522 let id = self.id;
4523 self.send_job(None, move |repo, _| async move {
4524 match repo {
4525 RepositoryState::Local { backend, .. } => backend.default_branch().await,
4526 RepositoryState::Remote { project_id, client } => {
4527 let response = client
4528 .request(proto::GetDefaultBranch {
4529 project_id: project_id.0,
4530 repository_id: id.to_proto(),
4531 })
4532 .await?;
4533
4534 anyhow::Ok(response.branch.map(SharedString::from))
4535 }
4536 }
4537 })
4538 }
4539
4540 pub fn diff_tree(
4541 &mut self,
4542 diff_type: DiffTreeType,
4543 _cx: &App,
4544 ) -> oneshot::Receiver<Result<TreeDiff>> {
4545 let repository_id = self.snapshot.id;
4546 self.send_job(None, move |repo, _cx| async move {
4547 match repo {
4548 RepositoryState::Local { backend, .. } => backend.diff_tree(diff_type).await,
4549 RepositoryState::Remote { client, project_id } => {
4550 let response = client
4551 .request(proto::GetTreeDiff {
4552 project_id: project_id.0,
4553 repository_id: repository_id.0,
4554 is_merge: matches!(diff_type, DiffTreeType::MergeBase { .. }),
4555 base: diff_type.base().to_string(),
4556 head: diff_type.head().to_string(),
4557 })
4558 .await?;
4559
4560 let entries = response
4561 .entries
4562 .into_iter()
4563 .filter_map(|entry| {
4564 let status = match entry.status() {
4565 proto::tree_diff_status::Status::Added => TreeDiffStatus::Added,
4566 proto::tree_diff_status::Status::Modified => {
4567 TreeDiffStatus::Modified {
4568 old: git::Oid::from_str(
4569 &entry.oid.context("missing oid").log_err()?,
4570 )
4571 .log_err()?,
4572 }
4573 }
4574 proto::tree_diff_status::Status::Deleted => {
4575 TreeDiffStatus::Deleted {
4576 old: git::Oid::from_str(
4577 &entry.oid.context("missing oid").log_err()?,
4578 )
4579 .log_err()?,
4580 }
4581 }
4582 };
4583 Some((
4584 RepoPath(RelPath::from_proto(&entry.path).log_err()?),
4585 status,
4586 ))
4587 })
4588 .collect();
4589
4590 Ok(TreeDiff { entries })
4591 }
4592 }
4593 })
4594 }
4595
4596 pub fn diff(&mut self, diff_type: DiffType, _cx: &App) -> oneshot::Receiver<Result<String>> {
4597 let id = self.id;
4598 self.send_job(None, move |repo, _cx| async move {
4599 match repo {
4600 RepositoryState::Local { backend, .. } => backend.diff(diff_type).await,
4601 RepositoryState::Remote { project_id, client } => {
4602 let response = client
4603 .request(proto::GitDiff {
4604 project_id: project_id.0,
4605 repository_id: id.to_proto(),
4606 diff_type: match diff_type {
4607 DiffType::HeadToIndex => {
4608 proto::git_diff::DiffType::HeadToIndex.into()
4609 }
4610 DiffType::HeadToWorktree => {
4611 proto::git_diff::DiffType::HeadToWorktree.into()
4612 }
4613 },
4614 })
4615 .await?;
4616
4617 Ok(response.diff)
4618 }
4619 }
4620 })
4621 }
4622
4623 pub fn create_branch(&mut self, branch_name: String) -> oneshot::Receiver<Result<()>> {
4624 let id = self.id;
4625 self.send_job(
4626 Some(format!("git switch -c {branch_name}").into()),
4627 move |repo, _cx| async move {
4628 match repo {
4629 RepositoryState::Local { backend, .. } => {
4630 backend.create_branch(branch_name).await
4631 }
4632 RepositoryState::Remote { project_id, client } => {
4633 client
4634 .request(proto::GitCreateBranch {
4635 project_id: project_id.0,
4636 repository_id: id.to_proto(),
4637 branch_name,
4638 })
4639 .await?;
4640
4641 Ok(())
4642 }
4643 }
4644 },
4645 )
4646 }
4647
4648 pub fn change_branch(&mut self, branch_name: String) -> oneshot::Receiver<Result<()>> {
4649 let id = self.id;
4650 self.send_job(
4651 Some(format!("git switch {branch_name}").into()),
4652 move |repo, _cx| async move {
4653 match repo {
4654 RepositoryState::Local { backend, .. } => {
4655 backend.change_branch(branch_name).await
4656 }
4657 RepositoryState::Remote { project_id, client } => {
4658 client
4659 .request(proto::GitChangeBranch {
4660 project_id: project_id.0,
4661 repository_id: id.to_proto(),
4662 branch_name,
4663 })
4664 .await?;
4665
4666 Ok(())
4667 }
4668 }
4669 },
4670 )
4671 }
4672
4673 pub fn rename_branch(
4674 &mut self,
4675 branch: String,
4676 new_name: String,
4677 ) -> oneshot::Receiver<Result<()>> {
4678 let id = self.id;
4679 self.send_job(
4680 Some(format!("git branch -m {branch} {new_name}").into()),
4681 move |repo, _cx| async move {
4682 match repo {
4683 RepositoryState::Local { backend, .. } => {
4684 backend.rename_branch(branch, new_name).await
4685 }
4686 RepositoryState::Remote { project_id, client } => {
4687 client
4688 .request(proto::GitRenameBranch {
4689 project_id: project_id.0,
4690 repository_id: id.to_proto(),
4691 branch,
4692 new_name,
4693 })
4694 .await?;
4695
4696 Ok(())
4697 }
4698 }
4699 },
4700 )
4701 }
4702
4703 pub fn check_for_pushed_commits(&mut self) -> oneshot::Receiver<Result<Vec<SharedString>>> {
4704 let id = self.id;
4705 self.send_job(None, move |repo, _cx| async move {
4706 match repo {
4707 RepositoryState::Local { backend, .. } => backend.check_for_pushed_commit().await,
4708 RepositoryState::Remote { project_id, client } => {
4709 let response = client
4710 .request(proto::CheckForPushedCommits {
4711 project_id: project_id.0,
4712 repository_id: id.to_proto(),
4713 })
4714 .await?;
4715
4716 let branches = response.pushed_to.into_iter().map(Into::into).collect();
4717
4718 Ok(branches)
4719 }
4720 }
4721 })
4722 }
4723
4724 pub fn checkpoint(&mut self) -> oneshot::Receiver<Result<GitRepositoryCheckpoint>> {
4725 self.send_job(None, |repo, _cx| async move {
4726 match repo {
4727 RepositoryState::Local { backend, .. } => backend.checkpoint().await,
4728 RepositoryState::Remote { .. } => anyhow::bail!("not implemented yet"),
4729 }
4730 })
4731 }
4732
4733 pub fn restore_checkpoint(
4734 &mut self,
4735 checkpoint: GitRepositoryCheckpoint,
4736 ) -> oneshot::Receiver<Result<()>> {
4737 self.send_job(None, move |repo, _cx| async move {
4738 match repo {
4739 RepositoryState::Local { backend, .. } => {
4740 backend.restore_checkpoint(checkpoint).await
4741 }
4742 RepositoryState::Remote { .. } => anyhow::bail!("not implemented yet"),
4743 }
4744 })
4745 }
4746
4747 pub(crate) fn apply_remote_update(
4748 &mut self,
4749 update: proto::UpdateRepository,
4750 cx: &mut Context<Self>,
4751 ) -> Result<()> {
4752 let conflicted_paths = TreeSet::from_ordered_entries(
4753 update
4754 .current_merge_conflicts
4755 .into_iter()
4756 .filter_map(|path| RepoPath::from_proto(&path).log_err()),
4757 );
4758 let new_branch = update.branch_summary.as_ref().map(proto_to_branch);
4759 let new_head_commit = update
4760 .head_commit_details
4761 .as_ref()
4762 .map(proto_to_commit_details);
4763 if self.snapshot.branch != new_branch || self.snapshot.head_commit != new_head_commit {
4764 cx.emit(RepositoryEvent::BranchChanged)
4765 }
4766 self.snapshot.branch = new_branch;
4767 self.snapshot.head_commit = new_head_commit;
4768
4769 self.snapshot.merge.conflicted_paths = conflicted_paths;
4770 self.snapshot.merge.message = update.merge_message.map(SharedString::from);
4771 let new_stash_entries = GitStash {
4772 entries: update
4773 .stash_entries
4774 .iter()
4775 .filter_map(|entry| proto_to_stash(entry).ok())
4776 .collect(),
4777 };
4778 if self.snapshot.stash_entries != new_stash_entries {
4779 cx.emit(RepositoryEvent::StashEntriesChanged)
4780 }
4781 self.snapshot.stash_entries = new_stash_entries;
4782
4783 let edits = update
4784 .removed_statuses
4785 .into_iter()
4786 .filter_map(|path| {
4787 Some(sum_tree::Edit::Remove(PathKey(
4788 RelPath::from_proto(&path).log_err()?,
4789 )))
4790 })
4791 .chain(
4792 update
4793 .updated_statuses
4794 .into_iter()
4795 .filter_map(|updated_status| {
4796 Some(sum_tree::Edit::Insert(updated_status.try_into().log_err()?))
4797 }),
4798 )
4799 .collect::<Vec<_>>();
4800 if !edits.is_empty() {
4801 cx.emit(RepositoryEvent::StatusesChanged { full_scan: true });
4802 }
4803 self.snapshot.statuses_by_path.edit(edits, ());
4804 if update.is_last_update {
4805 self.snapshot.scan_id = update.scan_id;
4806 }
4807 Ok(())
4808 }
4809
4810 pub fn compare_checkpoints(
4811 &mut self,
4812 left: GitRepositoryCheckpoint,
4813 right: GitRepositoryCheckpoint,
4814 ) -> oneshot::Receiver<Result<bool>> {
4815 self.send_job(None, move |repo, _cx| async move {
4816 match repo {
4817 RepositoryState::Local { backend, .. } => {
4818 backend.compare_checkpoints(left, right).await
4819 }
4820 RepositoryState::Remote { .. } => anyhow::bail!("not implemented yet"),
4821 }
4822 })
4823 }
4824
4825 pub fn diff_checkpoints(
4826 &mut self,
4827 base_checkpoint: GitRepositoryCheckpoint,
4828 target_checkpoint: GitRepositoryCheckpoint,
4829 ) -> oneshot::Receiver<Result<String>> {
4830 self.send_job(None, move |repo, _cx| async move {
4831 match repo {
4832 RepositoryState::Local { backend, .. } => {
4833 backend
4834 .diff_checkpoints(base_checkpoint, target_checkpoint)
4835 .await
4836 }
4837 RepositoryState::Remote { .. } => anyhow::bail!("not implemented yet"),
4838 }
4839 })
4840 }
4841
4842 fn schedule_scan(
4843 &mut self,
4844 updates_tx: Option<mpsc::UnboundedSender<DownstreamUpdate>>,
4845 cx: &mut Context<Self>,
4846 ) {
4847 let this = cx.weak_entity();
4848 let _ = self.send_keyed_job(
4849 Some(GitJobKey::ReloadGitState),
4850 None,
4851 |state, mut cx| async move {
4852 log::debug!("run scheduled git status scan");
4853
4854 let Some(this) = this.upgrade() else {
4855 return Ok(());
4856 };
4857 let RepositoryState::Local { backend, .. } = state else {
4858 bail!("not a local repository")
4859 };
4860 let (snapshot, events) = this
4861 .update(&mut cx, |this, _| {
4862 this.paths_needing_status_update.clear();
4863 compute_snapshot(
4864 this.id,
4865 this.work_directory_abs_path.clone(),
4866 this.snapshot.clone(),
4867 backend.clone(),
4868 )
4869 })?
4870 .await?;
4871 this.update(&mut cx, |this, cx| {
4872 this.snapshot = snapshot.clone();
4873 for event in events {
4874 cx.emit(event);
4875 }
4876 })?;
4877 if let Some(updates_tx) = updates_tx {
4878 updates_tx
4879 .unbounded_send(DownstreamUpdate::UpdateRepository(snapshot))
4880 .ok();
4881 }
4882 Ok(())
4883 },
4884 );
4885 }
4886
4887 fn spawn_local_git_worker(
4888 work_directory_abs_path: Arc<Path>,
4889 dot_git_abs_path: Arc<Path>,
4890 _repository_dir_abs_path: Arc<Path>,
4891 _common_dir_abs_path: Arc<Path>,
4892 project_environment: WeakEntity<ProjectEnvironment>,
4893 fs: Arc<dyn Fs>,
4894 cx: &mut Context<Self>,
4895 ) -> mpsc::UnboundedSender<GitJob> {
4896 let (job_tx, mut job_rx) = mpsc::unbounded::<GitJob>();
4897
4898 cx.spawn(async move |_, cx| {
4899 let environment = project_environment
4900 .upgrade()
4901 .context("missing project environment")?
4902 .update(cx, |project_environment, cx| {
4903 project_environment.get_local_directory_environment(&Shell::System, work_directory_abs_path.clone(), cx)
4904 })?
4905 .await
4906 .unwrap_or_else(|| {
4907 log::error!("failed to get working directory environment for repository {work_directory_abs_path:?}");
4908 HashMap::default()
4909 });
4910 let search_paths = environment.get("PATH").map(|val| val.to_owned());
4911 let backend = cx
4912 .background_spawn(async move {
4913 let system_git_binary_path = search_paths.and_then(|search_paths| which::which_in("git", Some(search_paths), &work_directory_abs_path).ok())
4914 .or_else(|| which::which("git").ok());
4915 fs.open_repo(&dot_git_abs_path, system_git_binary_path.as_deref())
4916 .with_context(|| format!("opening repository at {dot_git_abs_path:?}"))
4917 })
4918 .await?;
4919
4920 if let Some(git_hosting_provider_registry) =
4921 cx.update(|cx| GitHostingProviderRegistry::try_global(cx))?
4922 {
4923 git_hosting_providers::register_additional_providers(
4924 git_hosting_provider_registry,
4925 backend.clone(),
4926 );
4927 }
4928
4929 let state = RepositoryState::Local {
4930 backend,
4931 environment: Arc::new(environment),
4932 };
4933 let mut jobs = VecDeque::new();
4934 loop {
4935 while let Ok(Some(next_job)) = job_rx.try_next() {
4936 jobs.push_back(next_job);
4937 }
4938
4939 if let Some(job) = jobs.pop_front() {
4940 if let Some(current_key) = &job.key
4941 && jobs
4942 .iter()
4943 .any(|other_job| other_job.key.as_ref() == Some(current_key))
4944 {
4945 continue;
4946 }
4947 (job.job)(state.clone(), cx).await;
4948 } else if let Some(job) = job_rx.next().await {
4949 jobs.push_back(job);
4950 } else {
4951 break;
4952 }
4953 }
4954 anyhow::Ok(())
4955 })
4956 .detach_and_log_err(cx);
4957
4958 job_tx
4959 }
4960
4961 fn spawn_remote_git_worker(
4962 project_id: ProjectId,
4963 client: AnyProtoClient,
4964 cx: &mut Context<Self>,
4965 ) -> mpsc::UnboundedSender<GitJob> {
4966 let (job_tx, mut job_rx) = mpsc::unbounded::<GitJob>();
4967
4968 cx.spawn(async move |_, cx| {
4969 let state = RepositoryState::Remote { project_id, client };
4970 let mut jobs = VecDeque::new();
4971 loop {
4972 while let Ok(Some(next_job)) = job_rx.try_next() {
4973 jobs.push_back(next_job);
4974 }
4975
4976 if let Some(job) = jobs.pop_front() {
4977 if let Some(current_key) = &job.key
4978 && jobs
4979 .iter()
4980 .any(|other_job| other_job.key.as_ref() == Some(current_key))
4981 {
4982 continue;
4983 }
4984 (job.job)(state.clone(), cx).await;
4985 } else if let Some(job) = job_rx.next().await {
4986 jobs.push_back(job);
4987 } else {
4988 break;
4989 }
4990 }
4991 anyhow::Ok(())
4992 })
4993 .detach_and_log_err(cx);
4994
4995 job_tx
4996 }
4997
4998 fn load_staged_text(
4999 &mut self,
5000 buffer_id: BufferId,
5001 repo_path: RepoPath,
5002 cx: &App,
5003 ) -> Task<Result<Option<String>>> {
5004 let rx = self.send_job(None, move |state, _| async move {
5005 match state {
5006 RepositoryState::Local { backend, .. } => {
5007 anyhow::Ok(backend.load_index_text(repo_path).await)
5008 }
5009 RepositoryState::Remote { project_id, client } => {
5010 let response = client
5011 .request(proto::OpenUnstagedDiff {
5012 project_id: project_id.to_proto(),
5013 buffer_id: buffer_id.to_proto(),
5014 })
5015 .await?;
5016 Ok(response.staged_text)
5017 }
5018 }
5019 });
5020 cx.spawn(|_: &mut AsyncApp| async move { rx.await? })
5021 }
5022
5023 fn load_committed_text(
5024 &mut self,
5025 buffer_id: BufferId,
5026 repo_path: RepoPath,
5027 cx: &App,
5028 ) -> Task<Result<DiffBasesChange>> {
5029 let rx = self.send_job(None, move |state, _| async move {
5030 match state {
5031 RepositoryState::Local { backend, .. } => {
5032 let committed_text = backend.load_committed_text(repo_path.clone()).await;
5033 let staged_text = backend.load_index_text(repo_path).await;
5034 let diff_bases_change = if committed_text == staged_text {
5035 DiffBasesChange::SetBoth(committed_text)
5036 } else {
5037 DiffBasesChange::SetEach {
5038 index: staged_text,
5039 head: committed_text,
5040 }
5041 };
5042 anyhow::Ok(diff_bases_change)
5043 }
5044 RepositoryState::Remote { project_id, client } => {
5045 use proto::open_uncommitted_diff_response::Mode;
5046
5047 let response = client
5048 .request(proto::OpenUncommittedDiff {
5049 project_id: project_id.to_proto(),
5050 buffer_id: buffer_id.to_proto(),
5051 })
5052 .await?;
5053 let mode = Mode::from_i32(response.mode).context("Invalid mode")?;
5054 let bases = match mode {
5055 Mode::IndexMatchesHead => DiffBasesChange::SetBoth(response.committed_text),
5056 Mode::IndexAndHead => DiffBasesChange::SetEach {
5057 head: response.committed_text,
5058 index: response.staged_text,
5059 },
5060 };
5061 Ok(bases)
5062 }
5063 }
5064 });
5065
5066 cx.spawn(|_: &mut AsyncApp| async move { rx.await? })
5067 }
5068 fn load_blob_content(&mut self, oid: Oid, cx: &App) -> Task<Result<String>> {
5069 let repository_id = self.snapshot.id;
5070 let rx = self.send_job(None, move |state, _| async move {
5071 match state {
5072 RepositoryState::Local { backend, .. } => backend.load_blob_content(oid).await,
5073 RepositoryState::Remote { client, project_id } => {
5074 let response = client
5075 .request(proto::GetBlobContent {
5076 project_id: project_id.to_proto(),
5077 repository_id: repository_id.0,
5078 oid: oid.to_string(),
5079 })
5080 .await?;
5081 Ok(response.content)
5082 }
5083 }
5084 });
5085 cx.spawn(|_: &mut AsyncApp| async move { rx.await? })
5086 }
5087
5088 fn paths_changed(
5089 &mut self,
5090 paths: Vec<RepoPath>,
5091 updates_tx: Option<mpsc::UnboundedSender<DownstreamUpdate>>,
5092 cx: &mut Context<Self>,
5093 ) {
5094 self.paths_needing_status_update.extend(paths);
5095
5096 let this = cx.weak_entity();
5097 let _ = self.send_keyed_job(
5098 Some(GitJobKey::RefreshStatuses),
5099 None,
5100 |state, mut cx| async move {
5101 let (prev_snapshot, mut changed_paths) = this.update(&mut cx, |this, _| {
5102 (
5103 this.snapshot.clone(),
5104 mem::take(&mut this.paths_needing_status_update),
5105 )
5106 })?;
5107 let RepositoryState::Local { backend, .. } = state else {
5108 bail!("not a local repository")
5109 };
5110
5111 let paths = changed_paths.iter().cloned().collect::<Vec<_>>();
5112 if paths.is_empty() {
5113 return Ok(());
5114 }
5115 let statuses = backend.status(&paths).await?;
5116 let stash_entries = backend.stash_entries().await?;
5117
5118 let changed_path_statuses = cx
5119 .background_spawn(async move {
5120 let mut changed_path_statuses = Vec::new();
5121 let prev_statuses = prev_snapshot.statuses_by_path.clone();
5122 let mut cursor = prev_statuses.cursor::<PathProgress>(());
5123
5124 for (repo_path, status) in &*statuses.entries {
5125 changed_paths.remove(repo_path);
5126 if cursor.seek_forward(&PathTarget::Path(repo_path), Bias::Left)
5127 && cursor.item().is_some_and(|entry| entry.status == *status)
5128 {
5129 continue;
5130 }
5131
5132 changed_path_statuses.push(Edit::Insert(StatusEntry {
5133 repo_path: repo_path.clone(),
5134 status: *status,
5135 }));
5136 }
5137 let mut cursor = prev_statuses.cursor::<PathProgress>(());
5138 for path in changed_paths.into_iter() {
5139 if cursor.seek_forward(&PathTarget::Path(&path), Bias::Left) {
5140 changed_path_statuses.push(Edit::Remove(PathKey(path.0)));
5141 }
5142 }
5143 changed_path_statuses
5144 })
5145 .await;
5146
5147 this.update(&mut cx, |this, cx| {
5148 if this.snapshot.stash_entries != stash_entries {
5149 cx.emit(RepositoryEvent::StashEntriesChanged);
5150 this.snapshot.stash_entries = stash_entries;
5151 }
5152
5153 if !changed_path_statuses.is_empty() {
5154 cx.emit(RepositoryEvent::StatusesChanged { full_scan: false });
5155 this.snapshot
5156 .statuses_by_path
5157 .edit(changed_path_statuses, ());
5158 this.snapshot.scan_id += 1;
5159 }
5160
5161 if let Some(updates_tx) = updates_tx {
5162 updates_tx
5163 .unbounded_send(DownstreamUpdate::UpdateRepository(
5164 this.snapshot.clone(),
5165 ))
5166 .ok();
5167 }
5168 })
5169 },
5170 );
5171 }
5172
5173 /// currently running git command and when it started
5174 pub fn current_job(&self) -> Option<JobInfo> {
5175 self.active_jobs.values().next().cloned()
5176 }
5177
5178 pub fn barrier(&mut self) -> oneshot::Receiver<()> {
5179 self.send_job(None, |_, _| async {})
5180 }
5181}
5182
5183fn get_permalink_in_rust_registry_src(
5184 provider_registry: Arc<GitHostingProviderRegistry>,
5185 path: PathBuf,
5186 selection: Range<u32>,
5187) -> Result<url::Url> {
5188 #[derive(Deserialize)]
5189 struct CargoVcsGit {
5190 sha1: String,
5191 }
5192
5193 #[derive(Deserialize)]
5194 struct CargoVcsInfo {
5195 git: CargoVcsGit,
5196 path_in_vcs: String,
5197 }
5198
5199 #[derive(Deserialize)]
5200 struct CargoPackage {
5201 repository: String,
5202 }
5203
5204 #[derive(Deserialize)]
5205 struct CargoToml {
5206 package: CargoPackage,
5207 }
5208
5209 let Some((dir, cargo_vcs_info_json)) = path.ancestors().skip(1).find_map(|dir| {
5210 let json = std::fs::read_to_string(dir.join(".cargo_vcs_info.json")).ok()?;
5211 Some((dir, json))
5212 }) else {
5213 bail!("No .cargo_vcs_info.json found in parent directories")
5214 };
5215 let cargo_vcs_info = serde_json::from_str::<CargoVcsInfo>(&cargo_vcs_info_json)?;
5216 let cargo_toml = std::fs::read_to_string(dir.join("Cargo.toml"))?;
5217 let manifest = toml::from_str::<CargoToml>(&cargo_toml)?;
5218 let (provider, remote) = parse_git_remote_url(provider_registry, &manifest.package.repository)
5219 .context("parsing package.repository field of manifest")?;
5220 let path = PathBuf::from(cargo_vcs_info.path_in_vcs).join(path.strip_prefix(dir).unwrap());
5221 let permalink = provider.build_permalink(
5222 remote,
5223 BuildPermalinkParams::new(
5224 &cargo_vcs_info.git.sha1,
5225 &RepoPath(
5226 RelPath::new(&path, PathStyle::local())
5227 .context("invalid path")?
5228 .into_arc(),
5229 ),
5230 Some(selection),
5231 ),
5232 );
5233 Ok(permalink)
5234}
5235
5236fn serialize_blame_buffer_response(blame: Option<git::blame::Blame>) -> proto::BlameBufferResponse {
5237 let Some(blame) = blame else {
5238 return proto::BlameBufferResponse {
5239 blame_response: None,
5240 };
5241 };
5242
5243 let entries = blame
5244 .entries
5245 .into_iter()
5246 .map(|entry| proto::BlameEntry {
5247 sha: entry.sha.as_bytes().into(),
5248 start_line: entry.range.start,
5249 end_line: entry.range.end,
5250 original_line_number: entry.original_line_number,
5251 author: entry.author,
5252 author_mail: entry.author_mail,
5253 author_time: entry.author_time,
5254 author_tz: entry.author_tz,
5255 committer: entry.committer_name,
5256 committer_mail: entry.committer_email,
5257 committer_time: entry.committer_time,
5258 committer_tz: entry.committer_tz,
5259 summary: entry.summary,
5260 previous: entry.previous,
5261 filename: entry.filename,
5262 })
5263 .collect::<Vec<_>>();
5264
5265 let messages = blame
5266 .messages
5267 .into_iter()
5268 .map(|(oid, message)| proto::CommitMessage {
5269 oid: oid.as_bytes().into(),
5270 message,
5271 })
5272 .collect::<Vec<_>>();
5273
5274 proto::BlameBufferResponse {
5275 blame_response: Some(proto::blame_buffer_response::BlameResponse {
5276 entries,
5277 messages,
5278 remote_url: blame.remote_url,
5279 }),
5280 }
5281}
5282
5283fn deserialize_blame_buffer_response(
5284 response: proto::BlameBufferResponse,
5285) -> Option<git::blame::Blame> {
5286 let response = response.blame_response?;
5287 let entries = response
5288 .entries
5289 .into_iter()
5290 .filter_map(|entry| {
5291 Some(git::blame::BlameEntry {
5292 sha: git::Oid::from_bytes(&entry.sha).ok()?,
5293 range: entry.start_line..entry.end_line,
5294 original_line_number: entry.original_line_number,
5295 committer_name: entry.committer,
5296 committer_time: entry.committer_time,
5297 committer_tz: entry.committer_tz,
5298 committer_email: entry.committer_mail,
5299 author: entry.author,
5300 author_mail: entry.author_mail,
5301 author_time: entry.author_time,
5302 author_tz: entry.author_tz,
5303 summary: entry.summary,
5304 previous: entry.previous,
5305 filename: entry.filename,
5306 })
5307 })
5308 .collect::<Vec<_>>();
5309
5310 let messages = response
5311 .messages
5312 .into_iter()
5313 .filter_map(|message| Some((git::Oid::from_bytes(&message.oid).ok()?, message.message)))
5314 .collect::<HashMap<_, _>>();
5315
5316 Some(Blame {
5317 entries,
5318 messages,
5319 remote_url: response.remote_url,
5320 })
5321}
5322
5323fn branch_to_proto(branch: &git::repository::Branch) -> proto::Branch {
5324 proto::Branch {
5325 is_head: branch.is_head,
5326 ref_name: branch.ref_name.to_string(),
5327 unix_timestamp: branch
5328 .most_recent_commit
5329 .as_ref()
5330 .map(|commit| commit.commit_timestamp as u64),
5331 upstream: branch.upstream.as_ref().map(|upstream| proto::GitUpstream {
5332 ref_name: upstream.ref_name.to_string(),
5333 tracking: upstream
5334 .tracking
5335 .status()
5336 .map(|upstream| proto::UpstreamTracking {
5337 ahead: upstream.ahead as u64,
5338 behind: upstream.behind as u64,
5339 }),
5340 }),
5341 most_recent_commit: branch
5342 .most_recent_commit
5343 .as_ref()
5344 .map(|commit| proto::CommitSummary {
5345 sha: commit.sha.to_string(),
5346 subject: commit.subject.to_string(),
5347 commit_timestamp: commit.commit_timestamp,
5348 author_name: commit.author_name.to_string(),
5349 }),
5350 }
5351}
5352
5353fn proto_to_branch(proto: &proto::Branch) -> git::repository::Branch {
5354 git::repository::Branch {
5355 is_head: proto.is_head,
5356 ref_name: proto.ref_name.clone().into(),
5357 upstream: proto
5358 .upstream
5359 .as_ref()
5360 .map(|upstream| git::repository::Upstream {
5361 ref_name: upstream.ref_name.to_string().into(),
5362 tracking: upstream
5363 .tracking
5364 .as_ref()
5365 .map(|tracking| {
5366 git::repository::UpstreamTracking::Tracked(UpstreamTrackingStatus {
5367 ahead: tracking.ahead as u32,
5368 behind: tracking.behind as u32,
5369 })
5370 })
5371 .unwrap_or(git::repository::UpstreamTracking::Gone),
5372 }),
5373 most_recent_commit: proto.most_recent_commit.as_ref().map(|commit| {
5374 git::repository::CommitSummary {
5375 sha: commit.sha.to_string().into(),
5376 subject: commit.subject.to_string().into(),
5377 commit_timestamp: commit.commit_timestamp,
5378 author_name: commit.author_name.to_string().into(),
5379 has_parent: true,
5380 }
5381 }),
5382 }
5383}
5384
5385fn commit_details_to_proto(commit: &CommitDetails) -> proto::GitCommitDetails {
5386 proto::GitCommitDetails {
5387 sha: commit.sha.to_string(),
5388 message: commit.message.to_string(),
5389 commit_timestamp: commit.commit_timestamp,
5390 author_email: commit.author_email.to_string(),
5391 author_name: commit.author_name.to_string(),
5392 }
5393}
5394
5395fn proto_to_commit_details(proto: &proto::GitCommitDetails) -> CommitDetails {
5396 CommitDetails {
5397 sha: proto.sha.clone().into(),
5398 message: proto.message.clone().into(),
5399 commit_timestamp: proto.commit_timestamp,
5400 author_email: proto.author_email.clone().into(),
5401 author_name: proto.author_name.clone().into(),
5402 }
5403}
5404
5405async fn compute_snapshot(
5406 id: RepositoryId,
5407 work_directory_abs_path: Arc<Path>,
5408 prev_snapshot: RepositorySnapshot,
5409 backend: Arc<dyn GitRepository>,
5410) -> Result<(RepositorySnapshot, Vec<RepositoryEvent>)> {
5411 let mut events = Vec::new();
5412 let branches = backend.branches().await?;
5413 let branch = branches.into_iter().find(|branch| branch.is_head);
5414 let statuses = backend.status(&[RelPath::empty().into()]).await?;
5415 let stash_entries = backend.stash_entries().await?;
5416 let statuses_by_path = SumTree::from_iter(
5417 statuses
5418 .entries
5419 .iter()
5420 .map(|(repo_path, status)| StatusEntry {
5421 repo_path: repo_path.clone(),
5422 status: *status,
5423 }),
5424 (),
5425 );
5426 let (merge_details, merge_heads_changed) =
5427 MergeDetails::load(&backend, &statuses_by_path, &prev_snapshot).await?;
5428 log::debug!("new merge details (changed={merge_heads_changed:?}): {merge_details:?}");
5429
5430 let pending_ops_by_path = prev_snapshot.pending_ops_by_path.clone();
5431
5432 if merge_heads_changed {
5433 events.push(RepositoryEvent::MergeHeadsChanged);
5434 }
5435
5436 if statuses_by_path != prev_snapshot.statuses_by_path {
5437 events.push(RepositoryEvent::StatusesChanged { full_scan: true })
5438 }
5439
5440 // Useful when branch is None in detached head state
5441 let head_commit = match backend.head_sha().await {
5442 Some(head_sha) => backend.show(head_sha).await.log_err(),
5443 None => None,
5444 };
5445
5446 if branch != prev_snapshot.branch || head_commit != prev_snapshot.head_commit {
5447 events.push(RepositoryEvent::BranchChanged);
5448 }
5449
5450 // Used by edit prediction data collection
5451 let remote_origin_url = backend.remote_url("origin");
5452 let remote_upstream_url = backend.remote_url("upstream");
5453
5454 let snapshot = RepositorySnapshot {
5455 id,
5456 statuses_by_path,
5457 pending_ops_by_path,
5458 work_directory_abs_path,
5459 path_style: prev_snapshot.path_style,
5460 scan_id: prev_snapshot.scan_id + 1,
5461 branch,
5462 head_commit,
5463 merge: merge_details,
5464 remote_origin_url,
5465 remote_upstream_url,
5466 stash_entries,
5467 };
5468
5469 Ok((snapshot, events))
5470}
5471
5472fn status_from_proto(
5473 simple_status: i32,
5474 status: Option<proto::GitFileStatus>,
5475) -> anyhow::Result<FileStatus> {
5476 use proto::git_file_status::Variant;
5477
5478 let Some(variant) = status.and_then(|status| status.variant) else {
5479 let code = proto::GitStatus::from_i32(simple_status)
5480 .with_context(|| format!("Invalid git status code: {simple_status}"))?;
5481 let result = match code {
5482 proto::GitStatus::Added => TrackedStatus {
5483 worktree_status: StatusCode::Added,
5484 index_status: StatusCode::Unmodified,
5485 }
5486 .into(),
5487 proto::GitStatus::Modified => TrackedStatus {
5488 worktree_status: StatusCode::Modified,
5489 index_status: StatusCode::Unmodified,
5490 }
5491 .into(),
5492 proto::GitStatus::Conflict => UnmergedStatus {
5493 first_head: UnmergedStatusCode::Updated,
5494 second_head: UnmergedStatusCode::Updated,
5495 }
5496 .into(),
5497 proto::GitStatus::Deleted => TrackedStatus {
5498 worktree_status: StatusCode::Deleted,
5499 index_status: StatusCode::Unmodified,
5500 }
5501 .into(),
5502 _ => anyhow::bail!("Invalid code for simple status: {simple_status}"),
5503 };
5504 return Ok(result);
5505 };
5506
5507 let result = match variant {
5508 Variant::Untracked(_) => FileStatus::Untracked,
5509 Variant::Ignored(_) => FileStatus::Ignored,
5510 Variant::Unmerged(unmerged) => {
5511 let [first_head, second_head] =
5512 [unmerged.first_head, unmerged.second_head].map(|head| {
5513 let code = proto::GitStatus::from_i32(head)
5514 .with_context(|| format!("Invalid git status code: {head}"))?;
5515 let result = match code {
5516 proto::GitStatus::Added => UnmergedStatusCode::Added,
5517 proto::GitStatus::Updated => UnmergedStatusCode::Updated,
5518 proto::GitStatus::Deleted => UnmergedStatusCode::Deleted,
5519 _ => anyhow::bail!("Invalid code for unmerged status: {code:?}"),
5520 };
5521 Ok(result)
5522 });
5523 let [first_head, second_head] = [first_head?, second_head?];
5524 UnmergedStatus {
5525 first_head,
5526 second_head,
5527 }
5528 .into()
5529 }
5530 Variant::Tracked(tracked) => {
5531 let [index_status, worktree_status] = [tracked.index_status, tracked.worktree_status]
5532 .map(|status| {
5533 let code = proto::GitStatus::from_i32(status)
5534 .with_context(|| format!("Invalid git status code: {status}"))?;
5535 let result = match code {
5536 proto::GitStatus::Modified => StatusCode::Modified,
5537 proto::GitStatus::TypeChanged => StatusCode::TypeChanged,
5538 proto::GitStatus::Added => StatusCode::Added,
5539 proto::GitStatus::Deleted => StatusCode::Deleted,
5540 proto::GitStatus::Renamed => StatusCode::Renamed,
5541 proto::GitStatus::Copied => StatusCode::Copied,
5542 proto::GitStatus::Unmodified => StatusCode::Unmodified,
5543 _ => anyhow::bail!("Invalid code for tracked status: {code:?}"),
5544 };
5545 Ok(result)
5546 });
5547 let [index_status, worktree_status] = [index_status?, worktree_status?];
5548 TrackedStatus {
5549 index_status,
5550 worktree_status,
5551 }
5552 .into()
5553 }
5554 };
5555 Ok(result)
5556}
5557
5558fn status_to_proto(status: FileStatus) -> proto::GitFileStatus {
5559 use proto::git_file_status::{Tracked, Unmerged, Variant};
5560
5561 let variant = match status {
5562 FileStatus::Untracked => Variant::Untracked(Default::default()),
5563 FileStatus::Ignored => Variant::Ignored(Default::default()),
5564 FileStatus::Unmerged(UnmergedStatus {
5565 first_head,
5566 second_head,
5567 }) => Variant::Unmerged(Unmerged {
5568 first_head: unmerged_status_to_proto(first_head),
5569 second_head: unmerged_status_to_proto(second_head),
5570 }),
5571 FileStatus::Tracked(TrackedStatus {
5572 index_status,
5573 worktree_status,
5574 }) => Variant::Tracked(Tracked {
5575 index_status: tracked_status_to_proto(index_status),
5576 worktree_status: tracked_status_to_proto(worktree_status),
5577 }),
5578 };
5579 proto::GitFileStatus {
5580 variant: Some(variant),
5581 }
5582}
5583
5584fn unmerged_status_to_proto(code: UnmergedStatusCode) -> i32 {
5585 match code {
5586 UnmergedStatusCode::Added => proto::GitStatus::Added as _,
5587 UnmergedStatusCode::Deleted => proto::GitStatus::Deleted as _,
5588 UnmergedStatusCode::Updated => proto::GitStatus::Updated as _,
5589 }
5590}
5591
5592fn tracked_status_to_proto(code: StatusCode) -> i32 {
5593 match code {
5594 StatusCode::Added => proto::GitStatus::Added as _,
5595 StatusCode::Deleted => proto::GitStatus::Deleted as _,
5596 StatusCode::Modified => proto::GitStatus::Modified as _,
5597 StatusCode::Renamed => proto::GitStatus::Renamed as _,
5598 StatusCode::TypeChanged => proto::GitStatus::TypeChanged as _,
5599 StatusCode::Copied => proto::GitStatus::Copied as _,
5600 StatusCode::Unmodified => proto::GitStatus::Unmodified as _,
5601 }
5602}