mod conflict_set; pub mod diff_buffer_list; pub mod git_traversal; pub mod job_debug_queue; pub mod pending_op; use crate::{ ProjectEnvironment, ProjectItem, ProjectPath, buffer_store::{BufferStore, BufferStoreEvent}, project_settings::ProjectSettings, trusted_worktrees::{ PathTrust, TrustedWorktrees, TrustedWorktreesEvent, TrustedWorktreesStore, }, worktree_store::{WorktreeStore, WorktreeStoreEvent}, }; use anyhow::{Context as _, Result, anyhow, bail}; use askpass::{AskPassDelegate, EncryptedPassword, IKnowWhatIAmDoingAndIHaveReadTheDocs}; use buffer_diff::{BufferDiff, DiffHunk, DiffHunkSecondaryStatus, PendingHunk, PendingSense}; use client::ProjectId; use collections::HashMap; pub use conflict_set::{ConflictRegion, ConflictSet, ConflictSetSnapshot, ConflictSetUpdate}; use fs::{Fs, RemoveOptions}; use futures::{ FutureExt, SinkExt, Stream, StreamExt, channel::{ mpsc, oneshot::{self, Canceled}, }, future::{self, BoxFuture, Shared}, stream::{FuturesOrdered, FuturesUnordered}, }; use git::{ BuildPermalinkParams, GitHostingProviderRegistry, Oid, RunHook, blame::Blame, parse_git_remote_url, repository::{ Branch, BranchesScanResult, CommitData, CommitDetails, CommitDiff, CommitFile, CommitOptions, CreateWorktreeTarget, DiffStatType, DiffType, FetchOptions, FileHistoryChangedFileSets, GitCommitTemplate, GitRepository, GitRepositoryCheckpoint, InitialGraphCommitData, LogOrder, LogSource, PushOptions, Remote, RemoteCommandOutput, RepoPath, ResetMode, SearchCommitArgs, UpstreamTrackingStatus, Worktree as GitWorktree, delete_branch_flag, }, stash::{GitStash, StashEntry}, status::{ self, DiffStat, DiffTreeType, FileStatus, GitSummary, StatusCode, TrackedStatus, TreeDiff, TreeDiffStatus, UnmergedStatus, UnmergedStatusCode, }, }; use gpui::{ App, AppContext, AsyncApp, BackgroundExecutor, Context, Entity, EventEmitter, SharedString, Subscription, Task, TaskExt, WeakEntity, }; use language::{ Anchor, Buffer, BufferEvent, Capability, Language, LanguageRegistry, proto::{deserialize_version, serialize_version}, }; use parking_lot::Mutex; use paths::{config_dir, home_dir}; use pending_op::{PendingOp, PendingOpId, PendingOps, PendingOpsSummary}; use postage::stream::Stream as _; use rpc::{ AnyProtoClient, TypedEnvelope, proto::{self, git_reset, split_repository_update}, }; use serde::Deserialize; use settings::{Settings, WorktreeId}; use smallvec::SmallVec; use smol::future::yield_now; use std::{ cmp::Ordering, collections::{BTreeSet, HashSet, VecDeque, hash_map::Entry}, future::Future, mem, ops::Range, path::{Path, PathBuf}, str::FromStr, sync::{ Arc, atomic::{self, AtomicU64}, }, time::{Duration, Instant, SystemTime}, }; use sum_tree::{Edit, SumTree, TreeMap}; use task::Shell; use text::{Bias, BufferId, OffsetRangeExt, Rope, ToOffset}; use util::{ ResultExt, debug_panic, paths::{PathStyle, SanitizedPath}, post_inc, rel_path::RelPath, }; use worktree::{ File, PathChange, PathKey, PathProgress, PathSummary, PathTarget, ProjectEntryId, UpdatedGitRepositoriesSet, UpdatedGitRepository, Worktree, }; use zeroize::Zeroize; pub struct GitStore { state: GitStoreState, buffer_store: Entity, worktree_store: Entity, repositories: HashMap>, worktree_ids: HashMap>, active_repo_id: Option, #[allow(clippy::type_complexity)] loading_diffs: HashMap<(BufferId, DiffKind), Shared, Arc>>>>, diffs: HashMap>, buffer_ids_by_index_text_buffer_id: HashMap, shared_diffs: HashMap>, _subscriptions: Vec, } #[derive(Default)] struct SharedDiffs { unstaged: Option>, uncommitted: Option>, } struct BufferGitState { unstaged_diff: Option>, staged_diff: Option<(WeakEntity, Entity)>, uncommitted_diff: Option>, oid_diffs: HashMap, WeakEntity>, conflict_set: Option>, recalculate_diff_task: Option>>, reparse_conflict_markers_task: Option>>, language: Option>, language_registry: Option>, conflict_updated_futures: Vec>, recalculating_tx: postage::watch::Sender, /// These operation counts are used to ensure that head and index text /// values read from the git repository are up-to-date with any hunk staging /// operations that have been performed on the BufferDiff. /// /// The operation count is incremented immediately when the user initiates a /// hunk stage/unstage operation. Then, upon finishing writing the new index /// text do disk, the `operation count as of write` is updated to reflect /// the operation count that prompted the write. hunk_staging_operation_count: usize, hunk_staging_operation_count_as_of_write: usize, head_text: Option>, index_text: Option>, /// The optimistic, in-flight index state: the sole input to the index write, /// expressed relative to the currently-loaded index text (`I0`). Never shown /// to any view (views use per-diff pending hunks). Cleared when a /// recalculation settles. /// /// `I0` is immutable for the lifetime of any pending edit (the index text /// buffer is only fast-forwarded when a recalculation settles, which clears /// this state in the same update), so byte offsets stay valid for the whole /// window. pending_index_edits: Option, Arc)>>, oid_texts: HashMap>, head_text_buffer: WeakEntity, index_text_buffer: WeakEntity, index_text_buffer_language_enabled: bool, head_changed: bool, index_changed: bool, language_changed: bool, } fn pending_hunks( hunks: &[DiffHunk], version: &clock::Global, sense: PendingSense, ) -> Vec { hunks .iter() .map(|hunk| { PendingHunk::new( hunk.buffer_range.clone(), hunk.diff_base_byte_range.clone(), version.clone(), sense, ) }) .collect() } #[derive(Clone, Debug)] enum DiffBasesChange { SetIndex(Option), SetHead(Option), SetEach { index: Option, head: Option, }, SetBoth(Option), } #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] enum DiffKind { Unstaged, Staged, Uncommitted, SinceOid(Option), } struct IndexTextFile { path: Arc, full_path: PathBuf, path_style: PathStyle, file_name: String, worktree_id: WorktreeId, is_private: bool, } impl IndexTextFile { fn new(file: &dyn language::File, cx: &App) -> Self { Self { path: file.path().clone(), full_path: file.full_path(cx), path_style: file.path_style(cx), file_name: file.file_name(cx).to_string(), worktree_id: file.worktree_id(cx), is_private: file.is_private(), } } } impl language::File for IndexTextFile { fn as_local(&self) -> Option<&dyn language::LocalFile> { None } fn disk_state(&self) -> language::DiskState { language::DiskState::Historic { was_deleted: false } } fn path(&self) -> &Arc { &self.path } fn full_path(&self, _: &App) -> PathBuf { self.full_path.clone() } fn path_style(&self, _: &App) -> PathStyle { self.path_style } fn file_name<'a>(&'a self, _: &'a App) -> &'a str { &self.file_name } fn worktree_id(&self, _: &App) -> WorktreeId { self.worktree_id } fn to_proto(&self, _: &App) -> rpc::proto::File { rpc::proto::File { worktree_id: self.worktree_id.to_proto(), entry_id: None, path: self.path.as_ref().as_unix_str().to_owned(), mtime: None, is_deleted: false, is_historic: true, } } fn is_private(&self) -> bool { self.is_private } } #[derive(Debug, Clone, Copy)] pub enum GitAccess { /// Either: /// - the user owns `.git` /// - the user doesn't own `.git`, but has both of: /// - OS-level read permissions /// - the directory is marked as safe (git config safe.directory) Yes, /// The user is not the owner of `.git`, and one of the following is true: /// - the directory is not marked as safe (git config safe.directory) /// - the user does not have OS-level read permissions to `.git` No, } enum GitStoreState { Local { next_repository_id: Arc, downstream: Option, project_environment: Entity, fs: Arc, _fs_watches: Box<[Task<()>]>, }, Remote { upstream_client: AnyProtoClient, upstream_project_id: u64, downstream: Option<(AnyProtoClient, ProjectId)>, }, } enum DownstreamUpdate { UpdateRepository(RepositorySnapshot), RemoveRepository(RepositoryId), } struct LocalDownstreamState { client: AnyProtoClient, project_id: ProjectId, updates_tx: mpsc::UnboundedSender, _task: Task>, } #[derive(Clone, Debug)] pub struct GitStoreCheckpoint { checkpoints_by_work_dir_abs_path: HashMap, GitRepositoryCheckpoint>, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct StatusEntry { pub repo_path: RepoPath, pub status: FileStatus, pub diff_stat: Option, pub staged_diff_stat: Option, pub unstaged_diff_stat: Option, } impl StatusEntry { fn to_proto(&self) -> proto::StatusEntry { let simple_status = match self.status { FileStatus::Ignored | FileStatus::Untracked => proto::GitStatus::Added as i32, FileStatus::Unmerged { .. } => proto::GitStatus::Conflict as i32, FileStatus::Tracked(TrackedStatus { index_status, worktree_status, }) => tracked_status_to_proto(if worktree_status != StatusCode::Unmodified { worktree_status } else { index_status }), }; proto::StatusEntry { repo_path: self.repo_path.as_unix_str().to_owned(), simple_status, status: Some(status_to_proto(self.status)), diff_stat_added: self.diff_stat.map(|ds| ds.added), diff_stat_deleted: self.diff_stat.map(|ds| ds.deleted), staged_diff_stat_added: self.staged_diff_stat.map(|ds| ds.added), staged_diff_stat_deleted: self.staged_diff_stat.map(|ds| ds.deleted), unstaged_diff_stat_added: self.unstaged_diff_stat.map(|ds| ds.added), unstaged_diff_stat_deleted: self.unstaged_diff_stat.map(|ds| ds.deleted), } } } impl TryFrom for StatusEntry { type Error = anyhow::Error; fn try_from(value: proto::StatusEntry) -> Result { let repo_path = RepoPath::from_proto(&value.repo_path).context("invalid repo path")?; let status = status_from_proto(value.simple_status, value.status)?; let diff_stat = match (value.diff_stat_added, value.diff_stat_deleted) { (Some(added), Some(deleted)) => Some(DiffStat { added, deleted }), _ => None, }; let staged_diff_stat = match (value.staged_diff_stat_added, value.staged_diff_stat_deleted) { (Some(added), Some(deleted)) => Some(DiffStat { added, deleted }), _ => None, }; let unstaged_diff_stat = match ( value.unstaged_diff_stat_added, value.unstaged_diff_stat_deleted, ) { (Some(added), Some(deleted)) => Some(DiffStat { added, deleted }), _ => None, }; Ok(Self { repo_path, status, diff_stat, staged_diff_stat, unstaged_diff_stat, }) } } impl sum_tree::Item for StatusEntry { type Summary = PathSummary; fn summary(&self, _: ::Context<'_>) -> Self::Summary { PathSummary { max_path: self.repo_path.as_ref().clone(), item_summary: self.status.summary(), } } } impl sum_tree::KeyedItem for StatusEntry { type Key = PathKey; fn key(&self) -> Self::Key { PathKey(self.repo_path.as_ref().clone()) } } #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct RepositoryId(pub u64); #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct MergeDetails { pub merge_heads_by_conflicted_path: TreeMap>>, pub message: Option, } #[derive(Clone)] pub enum CommitDataState { Loading(Option>>>), Loaded(Arc), } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RepositorySnapshot { pub id: RepositoryId, pub statuses_by_path: SumTree, pub work_directory_abs_path: Arc, pub dot_git_abs_path: Arc, /// Absolute path to the directory holding this worktree's Git state. /// /// For a linked worktree this is the worktree-specific directory under the /// common Git directory, such as `
/.git/worktrees/`. pub repository_dir_abs_path: Arc, /// Absolute path to the repository's common Git directory. /// /// For a normal checkout this is `/.git`. For a linked /// worktree this is the common Git directory shared by all worktrees. If /// that common directory is a bare repository, there may be no main /// worktree path to derive from it. pub common_dir_abs_path: Arc, pub path_style: PathStyle, pub branch: Option, pub branch_list: Arc<[Branch]>, pub branch_list_error: Option, pub head_commit: Option, pub scan_id: u64, pub merge: MergeDetails, pub remote_origin_url: Option, pub remote_upstream_url: Option, pub stash_entries: GitStash, pub linked_worktrees: Arc<[GitWorktree]>, } type JobId = u64; #[derive(Clone, Debug, PartialEq, Eq)] pub struct JobInfo { pub start: Instant, pub message: SharedString, } struct CommitDataHandler { _task: Task<()>, commit_data_request: async_channel::Sender, completion_senders: HashMap>>, pending_requests: HashSet, } /// Represents the handler of a git cat-file --batch process within Zed /// It's used to lazily fetch commit data as needed (whatever a user is viewing) enum CommitDataHandlerState { /// The handler is open and processing requests Open(CommitDataHandler), /// The handler closed because it didn't receive any requests in the last 10s /// or hasn't been open before Closed, } enum NextCommitDataRequest { Request(BoxFuture<'static, Result>), Idle, Closed, } pub struct InitialGitGraphData { fetch_task: Task<()>, pub error: Option, pub commit_data: Vec>, pub commit_oid_to_index: HashMap, subscribers: Vec>, SharedString>>>, } pub struct GraphDataResponse<'a> { pub commits: &'a [Arc], pub is_loading: bool, pub error: Option, } pub struct Repository { this: WeakEntity, snapshot: RepositorySnapshot, commit_message_buffer: Option>, git_store: WeakEntity, // For a local repository, holds paths that have had worktree events since the last status scan completed, // and that should be examined during the next status scan. paths_needing_status_update: Vec>, job_sender: mpsc::UnboundedSender, _worker_task: Task<()>, active_jobs: HashMap, job_debug_queue: job_debug_queue::GitJobDebugQueue, pending_ops: SumTree, job_id: JobId, askpass_delegates: Arc>>, latest_askpass_id: u64, repository_state: Shared>>, initial_graph_data: HashMap<(LogSource, LogOrder), InitialGitGraphData>, commit_data_handler: CommitDataHandlerState, commit_data: HashMap, } impl std::ops::Deref for Repository { type Target = RepositorySnapshot; fn deref(&self) -> &Self::Target { &self.snapshot } } #[derive(Clone)] pub struct LocalRepositoryState { pub fs: Arc, pub backend: Arc, pub environment: Arc>, } impl LocalRepositoryState { async fn new( work_directory_abs_path: Arc, dot_git_abs_path: Arc, project_environment: WeakEntity, fs: Arc, is_trusted: bool, cx: &mut AsyncApp, ) -> anyhow::Result { let environment = project_environment .update(cx, |project_environment, cx| { project_environment.local_directory_environment(&Shell::System, work_directory_abs_path.clone(), cx) })? .await .unwrap_or_else(|| { log::error!("failed to get working directory environment for repository {work_directory_abs_path:?}"); HashMap::default() }); let search_paths = environment.get("PATH").map(|val| val.to_owned()); let backend = cx .background_spawn({ let fs = fs.clone(); async move { let system_git_binary_path = search_paths .and_then(|search_paths| { which::which_in("git", Some(search_paths), &work_directory_abs_path) .ok() }) .or_else(|| which::which("git").ok()); fs.open_repo(&dot_git_abs_path, system_git_binary_path.as_deref()) .with_context(|| format!("opening repository at {dot_git_abs_path:?}")) } }) .await?; backend.set_trusted(is_trusted); Ok(LocalRepositoryState { backend, environment: Arc::new(environment), fs, }) } } #[derive(Clone)] pub struct RemoteRepositoryState { pub project_id: ProjectId, pub client: AnyProtoClient, } #[derive(Clone)] pub enum RepositoryState { Local(LocalRepositoryState), Remote(RemoteRepositoryState), } #[derive(Clone, Debug, PartialEq, Eq)] pub enum GitGraphEvent { CountUpdated(usize), FullyLoaded, LoadingError, } #[derive(Clone, Debug, PartialEq, Eq)] pub enum RepositoryEvent { StatusesChanged, HeadChanged, BranchListChanged, StashEntriesChanged, GitWorktreeListChanged, PendingOpsChanged { pending_ops: SumTree }, GraphEvent((LogSource, LogOrder), GitGraphEvent), GitDirectoryChanged, } #[derive(Clone, Debug)] pub struct JobsUpdated; #[derive(Debug)] pub enum GitStoreEvent { ActiveRepositoryChanged(Option), /// Bool is true when the repository that's updated is the active repository RepositoryUpdated(RepositoryId, RepositoryEvent, bool), RepositoryAdded, RepositoryRemoved(RepositoryId), IndexWriteError(anyhow::Error), JobsUpdated, ConflictsUpdated, GlobalConfigurationUpdated, } impl EventEmitter for Repository {} impl EventEmitter for Repository {} impl EventEmitter for GitStore {} pub struct GitJob { id: JobId, job: Box Task<()>>, key: Option, } #[derive(PartialEq, Eq)] enum GitJobKey { WriteIndex(Vec), ReloadBufferDiffBases, RefreshStatuses, ReloadGitState, } impl GitStore { pub fn local( worktree_store: &Entity, buffer_store: Entity, environment: Entity, fs: Arc, cx: &mut Context, ) -> Self { let _fs_watches = if fs.is_fake() { Box::new([]) } else { [ config_dir().join("git/config"), home_dir().join(".gitconfig"), ] .into_iter() .map(|path| { let fs = fs.clone(); cx.spawn(async move |this, cx| { let watcher = fs.watch(&path, Duration::from_millis(100)); let (mut watcher, _) = watcher.await; while let Some(_) = watcher.next().await { let Ok(_) = this.update(cx, |this, cx| { let GitStoreState::Local { project_environment, fs, .. } = &this.state else { return; }; let project_environment = project_environment.downgrade(); let fs = fs.clone(); let repositories_to_respawn = this .repositories .iter() .filter_map(|(repository_id, repo)| { repo.read(cx) .job_sender .is_closed() .then_some((*repository_id, repo.clone())) }) .collect::>(); for (repository_id, repo) in repositories_to_respawn { let is_trusted = this.repository_is_trusted(repository_id, cx); repo.update(cx, |repo, cx| { repo.respawn_local_worker( project_environment.clone(), fs.clone(), is_trusted, cx, ); repo.schedule_scan(None, cx); }) } cx.emit(GitStoreEvent::GlobalConfigurationUpdated); }) else { return; }; } }) }) .collect::>() .into_boxed_slice() }; Self::new( worktree_store.clone(), buffer_store, GitStoreState::Local { next_repository_id: Arc::new(AtomicU64::new(1)), downstream: None, project_environment: environment, _fs_watches, fs, }, cx, ) } pub fn remote( worktree_store: &Entity, buffer_store: Entity, upstream_client: AnyProtoClient, project_id: u64, cx: &mut Context, ) -> Self { Self::new( worktree_store.clone(), buffer_store, GitStoreState::Remote { upstream_client, upstream_project_id: project_id, downstream: None, }, cx, ) } fn new( worktree_store: Entity, buffer_store: Entity, state: GitStoreState, cx: &mut Context, ) -> Self { let mut _subscriptions = vec![ cx.subscribe(&worktree_store, Self::on_worktree_store_event), cx.subscribe(&buffer_store, Self::on_buffer_store_event), ]; if let Some(trusted_worktrees) = TrustedWorktrees::try_get_global(cx) { _subscriptions.push(cx.subscribe(&trusted_worktrees, Self::on_trusted_worktrees_event)); } GitStore { state, buffer_store, worktree_store, repositories: HashMap::default(), worktree_ids: HashMap::default(), active_repo_id: None, _subscriptions, loading_diffs: HashMap::default(), shared_diffs: HashMap::default(), diffs: HashMap::default(), buffer_ids_by_index_text_buffer_id: HashMap::default(), } } pub fn init(client: &AnyProtoClient) { client.add_entity_request_handler(Self::handle_get_remotes); client.add_entity_request_handler(Self::handle_get_branches); client.add_entity_request_handler(Self::handle_get_default_branch); client.add_entity_request_handler(Self::handle_change_branch); client.add_entity_request_handler(Self::handle_create_branch); client.add_entity_request_handler(Self::handle_rename_branch); client.add_entity_request_handler(Self::handle_create_remote); client.add_entity_request_handler(Self::handle_remove_remote); client.add_entity_request_handler(Self::handle_delete_branch); client.add_entity_request_handler(Self::handle_git_init); client.add_entity_request_handler(Self::handle_push); client.add_entity_request_handler(Self::handle_pull); client.add_entity_request_handler(Self::handle_fetch); client.add_entity_request_handler(Self::handle_stage); client.add_entity_request_handler(Self::handle_unstage); client.add_entity_request_handler(Self::handle_stash); client.add_entity_request_handler(Self::handle_stash_pop); client.add_entity_request_handler(Self::handle_stash_apply); client.add_entity_request_handler(Self::handle_stash_drop); client.add_entity_request_handler(Self::handle_commit); client.add_entity_request_handler(Self::handle_run_hook); client.add_entity_request_handler(Self::handle_reset); client.add_entity_request_handler(Self::handle_show); client.add_entity_request_handler(Self::handle_create_checkpoint); client.add_entity_request_handler(Self::handle_create_archive_checkpoint); client.add_entity_request_handler(Self::handle_restore_checkpoint); client.add_entity_request_handler(Self::handle_restore_archive_checkpoint); client.add_entity_request_handler(Self::handle_compare_checkpoints); client.add_entity_request_handler(Self::handle_diff_checkpoints); client.add_entity_request_handler(Self::handle_load_commit_diff); client.add_entity_request_handler(Self::handle_checkout_files); client.add_entity_request_handler(Self::handle_add_path_to_gitignore); client.add_entity_request_handler(Self::handle_add_path_to_git_info_exclude); client.add_entity_request_handler(Self::handle_open_commit_message_buffer); client.add_entity_request_handler(Self::handle_set_index_text); client.add_entity_request_handler(Self::handle_askpass); client.add_entity_request_handler(Self::handle_check_for_pushed_commits); client.add_entity_request_handler(Self::handle_git_diff); client.add_entity_request_handler(Self::handle_tree_diff); client.add_entity_request_handler(Self::handle_get_blob_content); client.add_entity_request_handler(Self::handle_open_unstaged_diff); client.add_entity_request_handler(Self::handle_open_uncommitted_diff); client.add_entity_message_handler(Self::handle_update_diff_bases); client.add_entity_request_handler(Self::handle_get_permalink_to_line); client.add_entity_request_handler(Self::handle_blame_buffer); client.add_entity_message_handler(Self::handle_update_repository); client.add_entity_message_handler(Self::handle_remove_repository); client.add_entity_request_handler(Self::handle_git_clone); client.add_entity_request_handler(Self::handle_get_worktrees); client.add_entity_request_handler(Self::handle_create_worktree); client.add_entity_request_handler(Self::handle_remove_worktree); client.add_entity_request_handler(Self::handle_rename_worktree); client.add_entity_request_handler(Self::handle_worktree_created_at); client.add_entity_request_handler(Self::handle_get_head_sha); client.add_entity_request_handler(Self::handle_edit_ref); client.add_entity_request_handler(Self::handle_repair_worktrees); client.add_entity_request_handler(Self::handle_get_commit_data); client.add_entity_stream_request_handler(Self::handle_get_initial_graph_data); client.add_entity_stream_request_handler(Self::handle_search_commits); } pub fn is_local(&self) -> bool { matches!(self.state, GitStoreState::Local { .. }) } fn set_active_repo_id(&mut self, repo_id: RepositoryId, cx: &mut Context) { if self.active_repo_id != Some(repo_id) { self.active_repo_id = Some(repo_id); cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(repo_id))); } } pub fn set_active_repo_for_path(&mut self, project_path: &ProjectPath, cx: &mut Context) { if let Some((repo, _)) = self.repository_and_path_for_project_path(project_path, cx) { self.set_active_repo_id(repo.read(cx).id, cx); } } pub fn set_active_repo_for_worktree( &mut self, worktree_id: WorktreeId, cx: &mut Context, ) { let Some(worktree) = self .worktree_store .read(cx) .worktree_for_id(worktree_id, cx) else { return; }; let worktree_abs_path = worktree.read(cx).abs_path(); let Some(repo_id) = self .repositories .values() .filter(|repo| { let repo_path = &repo.read(cx).work_directory_abs_path; // The folder opened in Zed isn't necessarily the repo root; it may be // a subdirectory of it, e.g. opening `~/code/myrepo/backend` when the // repo lives at `~/code/myrepo`. So match any repo whose work directory // contains the folder. Nested repos can produce multiple matches, e.g. // opening `~/code/myrepo/vendor/lib` where `vendor/lib` is a submodule // matches both `myrepo` and the submodule; `max_by_key` then picks the // innermost match (the submodule), which the folder actually belongs to. worktree_abs_path.starts_with(repo_path.as_ref()) }) .max_by_key(|repo| repo.read(cx).work_directory_abs_path.as_os_str().len()) .map(|repo| repo.read(cx).id) else { return; }; self.set_active_repo_id(repo_id, cx); } pub fn shared(&mut self, project_id: u64, client: AnyProtoClient, cx: &mut Context) { match &mut self.state { GitStoreState::Remote { downstream: downstream_client, .. } => { for repo in self.repositories.values() { let update = repo.read(cx).snapshot.initial_update(project_id); for update in split_repository_update(update) { client.send(update).log_err(); } } *downstream_client = Some((client, ProjectId(project_id))); } GitStoreState::Local { downstream: downstream_client, .. } => { let mut snapshots = HashMap::default(); let (updates_tx, mut updates_rx) = mpsc::unbounded(); for repo in self.repositories.values() { updates_tx .unbounded_send(DownstreamUpdate::UpdateRepository( repo.read(cx).snapshot.clone(), )) .ok(); } *downstream_client = Some(LocalDownstreamState { client: client.clone(), project_id: ProjectId(project_id), updates_tx, _task: cx.spawn(async move |this, cx| { cx.background_spawn(async move { while let Some(update) = updates_rx.next().await { match update { DownstreamUpdate::UpdateRepository(snapshot) => { if let Some(old_snapshot) = snapshots.get_mut(&snapshot.id) { let update = snapshot.build_update(old_snapshot, project_id); *old_snapshot = snapshot; for update in split_repository_update(update) { client.send(update)?; } } else { let update = snapshot.initial_update(project_id); for update in split_repository_update(update) { client.send(update)?; } snapshots.insert(snapshot.id, snapshot); } } DownstreamUpdate::RemoveRepository(id) => { client.send(proto::RemoveRepository { project_id, id: id.to_proto(), })?; } } } anyhow::Ok(()) }) .await .ok(); this.update(cx, |this, _| { if let GitStoreState::Local { downstream: downstream_client, .. } = &mut this.state { downstream_client.take(); } else { unreachable!("unshared called on remote store"); } }) }), }); } } } pub fn unshared(&mut self, _cx: &mut Context) { match &mut self.state { GitStoreState::Local { downstream: downstream_client, .. } => { downstream_client.take(); } GitStoreState::Remote { downstream: downstream_client, .. } => { downstream_client.take(); } } self.shared_diffs.clear(); } pub(crate) fn forget_shared_diffs_for(&mut self, peer_id: &proto::PeerId) { self.shared_diffs.remove(peer_id); } pub fn active_repository(&self) -> Option> { self.active_repo_id .as_ref() .map(|id| self.repositories[id].clone()) } fn file_is_symlink(file: &File, cx: &App) -> bool { file.worktree .read(cx) .entry_for_path(&file.path) .is_some_and(|entry| entry.canonical_path.is_some()) } fn buffer_is_symlink(buffer: &Entity, cx: &App) -> bool { File::from_dyn(buffer.read(cx).file()).is_some_and(|file| Self::file_is_symlink(file, cx)) } pub fn open_unstaged_diff( &mut self, buffer: Entity, cx: &mut Context, ) -> Task>> { let buffer_id = buffer.read(cx).remote_id(); if let Some(diff_state) = self.diffs.get(&buffer_id) && let Some(unstaged_diff) = diff_state .read(cx) .unstaged_diff .as_ref() .and_then(|weak| weak.upgrade()) { // If this unstaged diff was first opened as the uncommitted diff's // secondary, its index text wasn't highlighted. Enable it now and // recalc so the language gets applied to the deleted (base) side. diff_state.update(cx, |diff_state, cx| { if !diff_state.index_text_buffer_language_enabled { diff_state.index_text_buffer_language_enabled = true; let buffer_snapshot = buffer.read(cx).text_snapshot(); diff_state.recalculate_diffs(buffer_snapshot, cx); } }); if let Some(task) = diff_state.update(cx, |diff_state, _| diff_state.wait_for_recalculation()) { return cx.background_executor().spawn(async move { task.await; Ok(unstaged_diff) }); } return Task::ready(Ok(unstaged_diff)); } let Some((repo, repo_path)) = self.repository_and_path_for_buffer_id(buffer.read(cx).remote_id(), cx) else { return Task::ready(Err(anyhow!("failed to find git repository for buffer"))); }; let is_symlink = Self::buffer_is_symlink(&buffer, cx); let task = self .loading_diffs .entry((buffer_id, DiffKind::Unstaged)) .or_insert_with(|| { let staged_text = if is_symlink { Task::ready(Ok(None)) } else { repo.update(cx, |repo, cx| { repo.load_staged_text(buffer_id, repo_path, cx) }) }; cx.spawn(async move |this, cx| { Self::open_diff_internal( this, DiffKind::Unstaged, staged_text.await.map(DiffBasesChange::SetIndex), buffer, cx, ) .await .map_err(Arc::new) }) .shared() }) .clone(); cx.background_spawn(async move { task.await.map_err(|e| anyhow!("{e}")) }) } /// Opens the staged (HEAD-vs-index) diff for the given buffer, along with /// the index text buffer that is the diff's main buffer. pub fn open_staged_diff( &mut self, buffer: Entity, cx: &mut Context, ) -> Task, Entity)>> { let buffer_id = buffer.read(cx).remote_id(); if let Some(diff_state) = self.diffs.get(&buffer_id) && let Some(staged_diff) = diff_state.read(cx).staged_diff_and_index_text_buffer() { if let Some(task) = diff_state.update(cx, |diff_state, _| diff_state.wait_for_recalculation()) { return cx.background_executor().spawn(async move { task.await; Ok(staged_diff) }); } return Task::ready(Ok(staged_diff)); } let Some((repo, repo_path)) = self.repository_and_path_for_buffer_id(buffer.read(cx).remote_id(), cx) else { return Task::ready(Err(anyhow!("failed to find git repository for buffer"))); }; let task = self .loading_diffs .entry((buffer_id, DiffKind::Staged)) .or_insert_with(|| { let changes = repo.update(cx, |repo, cx| { repo.load_committed_text(buffer_id, repo_path, cx) }); cx.spawn(async move |this, cx| { Self::open_diff_internal(this, DiffKind::Staged, changes.await, buffer, cx) .await .map_err(Arc::new) }) .shared() }) .clone(); cx.spawn(async move |this, cx| { let diff = task.await.map_err(|e| anyhow!("{e}"))?; this.update(cx, |this, cx| { let index_text_buffer = this .diffs .get(&buffer_id) .and_then(|diff_state| { let (_, index_text_buffer) = diff_state.read(cx).staged_diff.as_ref()?; Some(index_text_buffer.clone()) }) .context("index text buffer missing after opening staged diff")?; Ok((diff, index_text_buffer)) })? }) } /// Stages the worktree changes covered by `worktree_ranges`, acting on the /// given unstaged (index-vs-worktree) diff. Used by both the unstaged-changes /// view and the uncommitted (gutter) controls: "stage" means the same index /// change regardless of which view it was invoked from, so callers holding an /// uncommitted diff pass its unstaged secondary. /// /// Decomposes the worktree region into the unstaged hunks it covers, so no /// worktree->index projection is needed. Optimistically suppresses the staged /// hunks from the unstaged diff and, if the uncommitted diff happens to be /// open, marks the corresponding uncommitted hunks as staging. pub fn stage_hunks( &mut self, buffer: Entity, unstaged_diff: Entity, worktree_ranges: Vec>, cx: &mut Context, ) -> Result<()> { if worktree_ranges.is_empty() { return Ok(()); } let buffer_snapshot = buffer.read(cx).snapshot(); let buffer_id = buffer_snapshot.remote_id(); let file_exists = buffer_snapshot .file() .is_some_and(|file| file.disk_state().exists()); let unstaged_snapshot = unstaged_diff.read(cx).snapshot(cx); // Decompose: the unstaged hunks the worktree region covers carry the // index range directly. Sorting by buffer offset also sorts by index // offset, since the hunks are non-overlapping. We read the raw hunks // (ignoring optimistic suppression) so that re-staging a hunk that was // optimistically staged and then unstaged still finds it. The footprints // let the patch drop earlier edits the region covers even where the raw // diff is empty (re-staging a hunk that is staged on disk but // optimistically unstaged). let mut unstaged_hunks = Vec::new(); let mut index_footprints = Vec::new(); for range in &worktree_ranges { unstaged_hunks.extend( unstaged_snapshot.raw_hunks_intersecting_range(range.clone(), &buffer_snapshot), ); index_footprints.push( unstaged_snapshot.base_text_range_for_buffer_range(range.clone(), &buffer_snapshot), ); } unstaged_hunks.sort_by_key(|hunk| hunk.buffer_range.start.to_offset(&buffer_snapshot)); unstaged_hunks.dedup_by(|a, b| a.buffer_range.start == b.buffer_range.start); // The uncommitted hunks the region covers get the optimistic // "staging" secondary status (free cross-view update). let uncommitted_diff = self.get_uncommitted_diff(buffer_id, cx); let mut uncommitted_hunks = Vec::new(); if let Some(uncommitted_diff) = &uncommitted_diff { let uncommitted_snapshot = uncommitted_diff.read(cx).snapshot(cx); for range in &worktree_ranges { uncommitted_hunks.extend( uncommitted_snapshot .hunks_intersecting_range(range.clone(), &buffer_snapshot) .filter(|hunk| { hunk.secondary_status != DiffHunkSecondaryStatus::NoSecondaryHunk }), ); } uncommitted_hunks .sort_by_key(|hunk| hunk.buffer_range.start.to_offset(&buffer_snapshot)); uncommitted_hunks.dedup_by(|a, b| a.buffer_range.start == b.buffer_range.start); } let index_edits = if !file_exists { // The worktree file is gone: staging removes it from the index. None } else { Some( unstaged_hunks .iter() .map(|hunk| { let worktree_range = hunk.buffer_range.to_offset(&buffer_snapshot); let replacement: Arc = Arc::from( buffer_snapshot .text_for_range(worktree_range) .collect::(), ); (hunk.diff_base_byte_range.clone(), replacement) }) .collect::>(), ) }; let version = buffer_snapshot.version().clone(); let unstaged_pending = pending_hunks(&unstaged_hunks, &version, PendingSense::Suppress); let uncommitted_pending = pending_hunks( &uncommitted_hunks, &version, PendingSense::SetSecondaryStatus { stage: true }, ); drop(unstaged_snapshot); let diff_state = self .diffs .get(&buffer_id) .cloned() .context("failed to find git state for buffer")?; diff_state.update(cx, |diff_state, _| { diff_state.remove_overlapping_pending_index_edits(&index_footprints); diff_state.insert_pending_index_edits(index_edits); }); if let Some(uncommitted_diff) = uncommitted_diff { uncommitted_diff.update(cx, |diff, cx| { diff.set_pending_hunks(&uncommitted_pending, &buffer_snapshot, cx); }); } unstaged_diff.update(cx, |diff, cx| { diff.set_pending_hunks(&unstaged_pending, &buffer_snapshot, cx); }); self.write_optimistic_index(buffer_id, cx); Ok(()) } /// Unstages the worktree changes covered by `worktree_ranges`, acting on the /// given uncommitted (HEAD-vs-worktree) diff, invoked from the uncommitted /// (gutter) controls. Uses the worktree->index projection (the hard part) /// because the acted-on hunks are HEAD-vs-worktree. pub fn unstage_uncommitted_hunks( &mut self, buffer: Entity, uncommitted_diff: Entity, worktree_ranges: Vec>, cx: &mut Context, ) -> Result<()> { if worktree_ranges.is_empty() { return Ok(()); } let buffer_snapshot = buffer.read(cx).snapshot(); let buffer_id = buffer_snapshot.remote_id(); let file_exists = buffer_snapshot .file() .is_some_and(|file| file.disk_state().exists()); let uncommitted_snapshot = uncommitted_diff.read(cx).snapshot(cx); let unstaged_snapshot = uncommitted_snapshot .secondary_diff() .context("diff has no unstaged secondary")?; let mut hunks = Vec::new(); for range in &worktree_ranges { hunks.extend( uncommitted_snapshot.hunks_intersecting_range(range.clone(), &buffer_snapshot), ); } hunks.sort_by_key(|hunk| hunk.buffer_range.start.to_offset(&buffer_snapshot)); hunks.dedup_by(|a, b| a.buffer_range.start == b.buffer_range.start); let (index_edits, pending) = uncommitted_snapshot.compute_uncommitted_index_edits( unstaged_snapshot, false, &hunks, &buffer_snapshot, file_exists, ); drop(uncommitted_snapshot); let diff_state = self .diffs .get(&buffer_id) .cloned() .context("failed to find git state for buffer")?; diff_state.update(cx, |diff_state, _| { diff_state.insert_pending_index_edits(index_edits) }); uncommitted_diff.update(cx, |diff, cx| { diff.set_pending_hunks(&pending, &buffer_snapshot, cx); }); self.write_optimistic_index(buffer_id, cx); Ok(()) } /// Unstages staged (HEAD-vs-index) hunks covered by `index_ranges` (in the /// index text buffer's coordinates), acting on the given staged diff, /// invoked from the staged-changes view. The acted-on hunks already carry /// an index range, so no projection is needed; optimistically suppresses /// them from the staged diff. pub fn unstage_staged_hunks( &mut self, staged_diff: Entity, index_ranges: Vec>, cx: &mut Context, ) -> Result<()> { if index_ranges.is_empty() { return Ok(()); } let index_buffer_id = staged_diff.read(cx).buffer_id; let buffer_id = *self .buffer_ids_by_index_text_buffer_id .get(&index_buffer_id) .context("failed to find git state for index text buffer")?; let diff_state = self .diffs .get(&buffer_id) .cloned() .context("failed to find git state for buffer")?; let index_buffer = diff_state .read(cx) .index_text_buffer() .context("index text is not loaded")?; let index_snapshot = index_buffer.read(cx).text_snapshot(); let staged_snapshot = staged_diff.read(cx).snapshot(cx); let mut hunks = Vec::new(); for range in &index_ranges { hunks.extend(staged_snapshot.hunks_intersecting_range(range.clone(), &index_snapshot)); } hunks.sort_by_key(|hunk| hunk.buffer_range.start.to_offset(&index_snapshot)); hunks.dedup_by(|a, b| a.buffer_range.start == b.buffer_range.start); let index_edits = staged_diff .read(cx) .unstage_staged_hunks(&hunks, &index_snapshot); let version = index_snapshot.version().clone(); let pending = pending_hunks(&hunks, &version, PendingSense::Suppress); drop(staged_snapshot); diff_state.update(cx, |diff_state, _| { diff_state.insert_pending_index_edits(index_edits) }); staged_diff.update(cx, |diff, cx| { diff.set_pending_hunks(&pending, &index_snapshot, cx); }); self.write_optimistic_index(buffer_id, cx); Ok(()) } /// Derives the desired index text from the buffer's optimistic patch and /// schedules the write. fn write_optimistic_index(&mut self, buffer_id: BufferId, cx: &mut Context) { let Some(diff_state) = self.diffs.get(&buffer_id) else { return; }; let new_index_text = diff_state .read(cx) .pending_index_text(cx) .map(|rope| rope.to_string()); self.write_index_text_for_buffer_id(buffer_id, new_index_text, cx); } pub fn open_diff_since( &mut self, oid: Option, buffer: Entity, repo: Entity, cx: &mut Context, ) -> Task>> { let buffer_id = buffer.read(cx).remote_id(); if let Some(diff_state) = self.diffs.get(&buffer_id) && let Some(oid_diff) = diff_state.read(cx).oid_diff(oid) { if let Some(task) = diff_state.update(cx, |diff_state, _| diff_state.wait_for_recalculation()) { return cx.background_executor().spawn(async move { task.await; Ok(oid_diff) }); } return Task::ready(Ok(oid_diff)); } let diff_kind = DiffKind::SinceOid(oid); if let Some(task) = self.loading_diffs.get(&(buffer_id, diff_kind)) { let task = task.clone(); return cx.background_spawn(async move { task.await.map_err(|e| anyhow!("{e}")) }); } let task = cx .spawn(async move |this, cx| { let result: Result> = async { let buffer_snapshot = buffer.update(cx, |buffer, _| buffer.snapshot()); let language_registry = buffer.update(cx, |buffer, _| buffer.language_registry()); let content: Option> = match oid { None => None, Some(oid) => Some({ let mut content = repo .update(cx, |repo, cx| repo.load_blob_content(oid, cx)) .await?; text::LineEnding::normalize(&mut content); content.into() }), }; let buffer_diff = cx.new(|cx| { BufferDiff::new( &buffer_snapshot, buffer_snapshot.language().cloned(), language_registry, cx, ) }); buffer_diff .update(cx, |buffer_diff, cx| { buffer_diff.set_base_text(content.clone(), buffer_snapshot.text, cx) }) .await; let unstaged_diff = this .update(cx, |this, cx| this.open_unstaged_diff(buffer.clone(), cx))? .await?; buffer_diff.update(cx, |buffer_diff, _| { buffer_diff.set_secondary_diff(unstaged_diff); }); this.update(cx, |this, cx| { this.loading_diffs.remove(&(buffer_id, diff_kind)); let git_store = cx.weak_entity(); let diff_state = this .diffs .entry(buffer_id) .or_insert_with(|| cx.new(|cx| BufferGitState::new(git_store, cx))); diff_state.update(cx, |state, _| { if let Some(oid) = oid { if let Some(content) = content { state.oid_texts.insert(oid, content); } } state.oid_diffs.insert(oid, buffer_diff.downgrade()); }); })?; Ok(buffer_diff) } .await; result.map_err(Arc::new) }) .shared(); self.loading_diffs .insert((buffer_id, diff_kind), task.clone()); cx.background_spawn(async move { task.await.map_err(|e| anyhow!("{e}")) }) } #[ztracing::instrument(skip_all)] pub fn open_uncommitted_diff( &mut self, buffer: Entity, cx: &mut Context, ) -> Task>> { let buffer_id = buffer.read(cx).remote_id(); if let Some(diff_state) = self.diffs.get(&buffer_id) && let Some(uncommitted_diff) = diff_state .read(cx) .uncommitted_diff .as_ref() .and_then(|weak| weak.upgrade()) { if let Some(task) = diff_state.update(cx, |diff_state, _| diff_state.wait_for_recalculation()) { return cx.background_executor().spawn(async move { task.await; Ok(uncommitted_diff) }); } return Task::ready(Ok(uncommitted_diff)); } let Some((repo, repo_path)) = self.repository_and_path_for_buffer_id(buffer.read(cx).remote_id(), cx) else { return Task::ready(Err(anyhow!("failed to find git repository for buffer"))); }; let is_symlink = Self::buffer_is_symlink(&buffer, cx); let task = self .loading_diffs .entry((buffer_id, DiffKind::Uncommitted)) .or_insert_with(|| { let changes = if is_symlink { Task::ready(Ok(DiffBasesChange::SetBoth(None))) } else { repo.update(cx, |repo, cx| { repo.load_committed_text(buffer_id, repo_path, cx) }) }; // todo(lw): hot foreground spawn cx.spawn(async move |this, cx| { Self::open_diff_internal(this, DiffKind::Uncommitted, changes.await, buffer, cx) .await .map_err(Arc::new) }) .shared() }) .clone(); cx.background_spawn(async move { task.await.map_err(|e| anyhow!("{e}")) }) } #[ztracing::instrument(skip_all)] async fn open_diff_internal( this: WeakEntity, kind: DiffKind, texts: Result, buffer_entity: Entity, cx: &mut AsyncApp, ) -> Result> { let diff_bases_change = match texts { Err(e) => { this.update(cx, |this, cx| { let buffer = buffer_entity.read(cx); let buffer_id = buffer.remote_id(); this.loading_diffs.remove(&(buffer_id, kind)); })?; return Err(e); } Ok(change) => change, }; this.update(cx, |this, cx| { let buffer = buffer_entity.read(cx); let buffer_id = buffer.remote_id(); let language = buffer.language().cloned(); let language_registry = buffer.language_registry(); let index_text_file = buffer.file().map(|file| { Arc::new(IndexTextFile::new(file.as_ref(), cx)) as Arc }); let text_snapshot = buffer.text_snapshot(); this.loading_diffs.remove(&(buffer_id, kind)); let git_store = cx.weak_entity(); let diff_state = this .diffs .entry(buffer_id) .or_insert_with(|| cx.new(|cx| BufferGitState::new(git_store, cx))); let existing_unstaged_diff = diff_state.read(cx).unstaged_diff(); let mut staged_index_text_buffer = None; let diff = if kind == DiffKind::Unstaged && let Some(existing_unstaged_diff) = existing_unstaged_diff.clone() { existing_unstaged_diff } else { let diff = match kind { DiffKind::Unstaged => { let base_text_buffer = diff_state.update(cx, |diff_state, cx| { diff_state.get_or_create_index_text_buffer(index_text_file.clone(), cx) }); cx.new(|cx| { BufferDiff::new_with_base_text_buffer( &text_snapshot, base_text_buffer, cx, ) }) } DiffKind::Staged => { let (index_text_buffer, base_text_buffer) = diff_state.update(cx, |diff_state, cx| { ( diff_state.get_or_create_index_text_buffer( index_text_file.clone(), cx, ), diff_state.get_or_create_head_text_buffer(cx), ) }); index_text_buffer.update(cx, |index_text_buffer, cx| { if let Some(language_registry) = language_registry.clone() { index_text_buffer.set_language_registry(language_registry); } index_text_buffer.set_language_async(language.clone(), cx); }); let index_text_snapshot = index_text_buffer.read(cx).text_snapshot(); staged_index_text_buffer = Some(index_text_buffer); cx.new(|cx| { BufferDiff::new_with_base_text_buffer( &index_text_snapshot, base_text_buffer, cx, ) }) } DiffKind::Uncommitted => { let base_text_buffer = diff_state.update(cx, |diff_state, cx| { diff_state.get_or_create_head_text_buffer(cx) }); cx.new(|cx| { BufferDiff::new_with_base_text_buffer( &text_snapshot, base_text_buffer, cx, ) }) } DiffKind::SinceOid(_) => { unreachable!("open_diff_internal is not used for OID diffs") } }; diff }; let rx = diff_state.update(cx, |diff_state, cx| { diff_state.language = language; diff_state.language_registry = language_registry; match kind { DiffKind::Unstaged => { diff_state.unstaged_diff = Some(diff.downgrade()); // The deleted (base) side of a standalone unstaged diff // is the index, so highlight it. The recalc kicked off by // `diff_bases_changed` below applies the language. diff_state.index_text_buffer_language_enabled = true; } DiffKind::Staged => { diff_state.index_text_buffer_language_enabled = true; let index_text_buffer = staged_index_text_buffer .take() .context("index text buffer was not created for staged diff")?; diff_state.staged_diff = Some((diff.downgrade(), index_text_buffer)); } DiffKind::Uncommitted => { let unstaged_diff = if let Some(diff) = existing_unstaged_diff { diff } else { let base_text_buffer = diff_state.get_or_create_index_text_buffer(index_text_file, cx); let unstaged_diff = cx.new(|cx| { BufferDiff::new_with_base_text_buffer( &text_snapshot, base_text_buffer, cx, ) }); diff_state.unstaged_diff = Some(unstaged_diff.downgrade()); unstaged_diff }; diff.update(cx, |diff, _| diff.set_secondary_diff(unstaged_diff)); diff_state.uncommitted_diff = Some(diff.downgrade()) } DiffKind::SinceOid(_) => { unreachable!("open_diff_internal is not used for OID diffs") } } diff_state.diff_bases_changed(text_snapshot, Some(diff_bases_change), cx); let rx = diff_state.wait_for_recalculation(); anyhow::Ok(async move { if let Some(rx) = rx { rx.await; } Ok(diff) }) }); let diff_state = this.diffs.get(&buffer_id).cloned(); if let Some(index_text_buffer) = diff_state.and_then(|diff_state| diff_state.read(cx).index_text_buffer()) { this.buffer_ids_by_index_text_buffer_id .insert(index_text_buffer.read(cx).remote_id(), buffer_id); } rx })?? .await } pub fn get_unstaged_diff(&self, buffer_id: BufferId, cx: &App) -> Option> { let diff_state = self.diffs.get(&buffer_id)?; diff_state.read(cx).unstaged_diff.as_ref()?.upgrade() } pub fn get_staged_diff(&self, buffer_id: BufferId, cx: &App) -> Option> { let diff_state = self.diffs.get(&buffer_id)?; diff_state.read(cx).staged_diff() } pub fn get_uncommitted_diff( &self, buffer_id: BufferId, cx: &App, ) -> Option> { let diff_state = self.diffs.get(&buffer_id)?; diff_state.read(cx).uncommitted_diff.as_ref()?.upgrade() } pub fn get_diff_since_oid( &self, buffer_id: BufferId, oid: Option, cx: &App, ) -> Option> { let diff_state = self.diffs.get(&buffer_id)?; diff_state.read(cx).oid_diff(oid) } /// Whether this buffer's index text is known to match its committed text /// without comparing contents, i.e. whether the texts share one allocation. /// In a downstream project, this can only be true if the upstream sent /// `Mode::IndexMatchesHead`. #[cfg(any(test, feature = "test-support"))] pub fn index_matches_head_for_buffer(&self, buffer_id: BufferId, cx: &App) -> bool { self.diffs .get(&buffer_id) .is_some_and(|diff_state| diff_state.read(cx).index_matches_head()) } pub fn open_conflict_set( &mut self, buffer: Entity, cx: &mut Context, ) -> Task> { let buffer_id = buffer.read(cx).remote_id(); if let Some(git_state) = self.diffs.get(&buffer_id) && let Some(conflict_set) = git_state .read(cx) .conflict_set .as_ref() .and_then(|weak| weak.upgrade()) { let conflict_set = conflict_set; let buffer_snapshot = buffer.read(cx).text_snapshot(); let rx = git_state.update(cx, |state, cx| { state.reparse_conflict_markers(buffer_snapshot, cx) }); return cx.spawn(async move |_, _| { rx.await.ok(); conflict_set }); } let is_unmerged = self .repository_and_path_for_buffer_id(buffer_id, cx) .is_some_and(|(repo, path)| repo.read(cx).snapshot.has_conflict(&path)); let git_store = cx.weak_entity(); let buffer_git_state = self .diffs .entry(buffer_id) .or_insert_with(|| cx.new(|cx| BufferGitState::new(git_store, cx))); let conflict_set = cx.new(|cx| ConflictSet::new(buffer_id, is_unmerged, cx)); self._subscriptions .push(cx.subscribe(&conflict_set, |_, _, _, cx| { cx.emit(GitStoreEvent::ConflictsUpdated); })); let rx = buffer_git_state.update(cx, |state, cx| { state.conflict_set = Some(conflict_set.downgrade()); let buffer_snapshot = buffer.read(cx).text_snapshot(); state.reparse_conflict_markers(buffer_snapshot, cx) }); cx.spawn(async move |_, _| { rx.await.ok(); conflict_set }) } pub fn project_path_git_status( &self, project_path: &ProjectPath, cx: &App, ) -> Option { let (repo, repo_path) = self.repository_and_path_for_project_path(project_path, cx)?; Some(repo.read(cx).status_for_path(&repo_path)?.status) } pub fn checkpoint(&self, cx: &mut App) -> Task> { let mut work_directory_abs_paths = Vec::new(); let mut checkpoints = Vec::new(); for repository in self.repositories.values() { repository.update(cx, |repository, _| { work_directory_abs_paths.push(repository.snapshot.work_directory_abs_path.clone()); checkpoints.push(repository.checkpoint().map(|checkpoint| checkpoint?)); }); } cx.background_executor().spawn(async move { let checkpoints = future::try_join_all(checkpoints).await?; Ok(GitStoreCheckpoint { checkpoints_by_work_dir_abs_path: work_directory_abs_paths .into_iter() .zip(checkpoints) .collect(), }) }) } pub fn restore_checkpoint( &self, checkpoint: GitStoreCheckpoint, cx: &mut App, ) -> Task> { let repositories_by_work_dir_abs_path = self .repositories .values() .map(|repo| (repo.read(cx).snapshot.work_directory_abs_path.clone(), repo)) .collect::>(); let mut tasks = Vec::new(); for (work_dir_abs_path, checkpoint) in checkpoint.checkpoints_by_work_dir_abs_path { if let Some(repository) = repositories_by_work_dir_abs_path.get(&work_dir_abs_path) { let restore = repository.update(cx, |repository, _| { repository.restore_checkpoint(checkpoint) }); tasks.push(async move { restore.await? }); } } cx.background_spawn(async move { future::try_join_all(tasks).await?; Ok(()) }) } /// Compares two checkpoints, returning true if they are equal. pub fn compare_checkpoints( &self, left: GitStoreCheckpoint, mut right: GitStoreCheckpoint, cx: &mut App, ) -> Task> { let repositories_by_work_dir_abs_path = self .repositories .values() .map(|repo| (repo.read(cx).snapshot.work_directory_abs_path.clone(), repo)) .collect::>(); let mut tasks = Vec::new(); for (work_dir_abs_path, left_checkpoint) in left.checkpoints_by_work_dir_abs_path { if let Some(right_checkpoint) = right .checkpoints_by_work_dir_abs_path .remove(&work_dir_abs_path) { if let Some(repository) = repositories_by_work_dir_abs_path.get(&work_dir_abs_path) { let compare = repository.update(cx, |repository, _| { repository.compare_checkpoints(left_checkpoint, right_checkpoint) }); tasks.push(async move { compare.await? }); } } else { return Task::ready(Ok(false)); } } cx.background_spawn(async move { Ok(future::try_join_all(tasks) .await? .into_iter() .all(|result| result)) }) } /// Blames a buffer. pub fn blame_buffer( &self, buffer: &Entity, version: Option, cx: &mut Context, ) -> Task>> { let buffer = buffer.read(cx); let Some((repo, repo_path)) = self.repository_and_path_for_buffer_id(buffer.remote_id(), cx) else { return Task::ready(Err(anyhow!("failed to find a git repository for buffer"))); }; let content = match &version { Some(version) => buffer.rope_for_version(version), None => buffer.as_rope().clone(), }; let line_ending = buffer.line_ending(); let version = version.unwrap_or(buffer.version()); let buffer_id = buffer.remote_id(); let repo = repo.downgrade(); cx.spawn(async move |_, cx| { let repository_state = repo .update(cx, |repo, _| repo.repository_state.clone())? .await .map_err(|err| anyhow::anyhow!(err))?; match repository_state { RepositoryState::Local(LocalRepositoryState { backend, .. }) => backend .blame(repo_path.clone(), content, line_ending) .await .with_context(|| format!("Failed to blame {:?}", repo_path.as_ref())) .map(Some), RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::BlameBuffer { project_id: project_id.to_proto(), buffer_id: buffer_id.into(), version: serialize_version(&version), }) .await?; Ok(deserialize_blame_buffer_response(response)) } } }) } pub fn get_permalink_to_line( &self, buffer: &Entity, selection: Range, cx: &mut App, ) -> Task> { let Some(file) = File::from_dyn(buffer.read(cx).file()) else { return Task::ready(Err(anyhow!("buffer has no file"))); }; let Some((repo, repo_path)) = self.repository_and_path_for_project_path( &(file.worktree.read(cx).id(), file.path.clone()).into(), cx, ) else { // If we're not in a Git repo, check whether this is a Rust source // file in the Cargo registry (presumably opened with go-to-definition // from a normal Rust file). If so, we can put together a permalink // using crate metadata. if buffer .read(cx) .language() .is_none_or(|lang| lang.name() != "Rust") { return Task::ready(Err(anyhow!("no permalink available"))); } let file_path = file.worktree.read(cx).absolutize(&file.path); return cx.spawn(async move |cx| { let provider_registry = cx.update(GitHostingProviderRegistry::default_global); get_permalink_in_rust_registry_src(provider_registry, file_path, selection) .context("no permalink available") }); }; let buffer_id = buffer.read(cx).remote_id(); let branch = repo.read(cx).branch.clone(); let remote = branch .as_ref() .and_then(|b| b.upstream.as_ref()) .and_then(|b| b.remote_name()) .unwrap_or("origin") .to_string(); let rx = repo.update(cx, |repo, _| { repo.send_job("get_permalink_to_line", None, move |state, cx| async move { match state { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { let origin_url = backend .remote_url(&remote) .await .with_context(|| format!("remote \"{remote}\" not found"))?; let sha = backend.head_sha().await.context("reading HEAD SHA")?; let provider_registry = cx.update(GitHostingProviderRegistry::default_global); let (provider, remote) = parse_git_remote_url(provider_registry, &origin_url) .context("parsing Git remote URL")?; Ok(provider.build_permalink( remote, BuildPermalinkParams::new(&sha, &repo_path, Some(selection)), )) } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GetPermalinkToLine { project_id: project_id.to_proto(), buffer_id: buffer_id.into(), selection: Some(proto::Range { start: selection.start as u64, end: selection.end as u64, }), }) .await?; url::Url::parse(&response.permalink).context("failed to parse permalink") } } }) }); cx.spawn(|_: &mut AsyncApp| async move { rx.await? }) } fn downstream_client(&self) -> Option<(AnyProtoClient, ProjectId)> { match &self.state { GitStoreState::Local { downstream: downstream_client, .. } => downstream_client .as_ref() .map(|state| (state.client.clone(), state.project_id)), GitStoreState::Remote { downstream: downstream_client, .. } => downstream_client.clone(), } } fn upstream_client(&self) -> Option { match &self.state { GitStoreState::Local { .. } => None, GitStoreState::Remote { upstream_client, .. } => Some(upstream_client.clone()), } } fn on_worktree_store_event( &mut self, worktree_store: Entity, event: &WorktreeStoreEvent, cx: &mut Context, ) { let GitStoreState::Local { project_environment, downstream, next_repository_id, fs, .. } = &self.state else { return; }; match event { WorktreeStoreEvent::WorktreeUpdatedEntries(worktree_id, updated_entries) => { if let Some(worktree) = self .worktree_store .read(cx) .worktree_for_id(*worktree_id, cx) { let paths_by_git_repo = self.process_updated_entries(&worktree, updated_entries, cx); let downstream = downstream .as_ref() .map(|downstream| downstream.updates_tx.clone()); cx.spawn(async move |_, cx| { let paths_by_git_repo = paths_by_git_repo.await; for (repo, paths) in paths_by_git_repo { repo.update(cx, |repo, cx| { repo.paths_changed(paths, downstream.clone(), cx); }); } }) .detach(); } } WorktreeStoreEvent::WorktreeUpdatedGitRepositories(worktree_id, changed_repos) => { let Some(worktree) = worktree_store.read(cx).worktree_for_id(*worktree_id, cx) else { return; }; log::debug!("received worktree update for repositories: {changed_repos:?}"); self.update_repositories_from_worktree( *worktree_id, project_environment.clone(), next_repository_id.clone(), downstream .as_ref() .map(|downstream| downstream.updates_tx.clone()), changed_repos.clone(), fs.clone(), cx, ); self.local_worktree_git_repos_changed(worktree, changed_repos, cx); } WorktreeStoreEvent::WorktreeRemoved(_entity_id, worktree_id) => { let repos_without_worktree: Vec = self .worktree_ids .iter_mut() .filter_map(|(repo_id, worktree_ids)| { worktree_ids.remove(worktree_id); if worktree_ids.is_empty() { Some(*repo_id) } else { None } }) .collect(); let is_active_repo_removed = repos_without_worktree .iter() .any(|repo_id| self.active_repo_id == Some(*repo_id)); for repo_id in repos_without_worktree { self.repositories.remove(&repo_id); self.worktree_ids.remove(&repo_id); if let Some(updates_tx) = downstream.as_ref().map(|downstream| &downstream.updates_tx) { updates_tx .unbounded_send(DownstreamUpdate::RemoveRepository(repo_id)) .ok(); } } if is_active_repo_removed { if let Some((&repo_id, _)) = self.repositories.iter().next() { self.active_repo_id = Some(repo_id); cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(repo_id))); } else { self.active_repo_id = None; cx.emit(GitStoreEvent::ActiveRepositoryChanged(None)); } } } _ => {} } } fn on_repository_event( &mut self, repo: Entity, event: &RepositoryEvent, cx: &mut Context, ) { let id = repo.read(cx).id; let repo_snapshot = repo.read(cx).snapshot.clone(); for (buffer_id, diff) in self.diffs.iter() { if let Some((buffer_repo, repo_path)) = self.repository_and_path_for_buffer_id(*buffer_id, cx) && buffer_repo == repo { diff.update(cx, |diff, cx| { if let Some(conflict_set) = &diff.conflict_set { let conflict_status_changed = conflict_set.update(cx, |conflict_set, cx| { let has_conflict = repo_snapshot.has_conflict(&repo_path); conflict_set.set_has_conflict(has_conflict, cx) })?; if conflict_status_changed { let buffer_store = self.buffer_store.read(cx); if let Some(buffer) = buffer_store.get(*buffer_id) { let _ = diff .reparse_conflict_markers(buffer.read(cx).text_snapshot(), cx); } } } anyhow::Ok(()) }) .ok(); } } cx.emit(GitStoreEvent::RepositoryUpdated( id, event.clone(), self.active_repo_id == Some(id), )) } fn on_jobs_updated(&mut self, _: Entity, _: &JobsUpdated, cx: &mut Context) { cx.emit(GitStoreEvent::JobsUpdated) } fn repository_is_trusted(&self, repository_id: RepositoryId, cx: &mut Context) -> bool { let Some(worktree_ids) = self.worktree_ids.get(&repository_id) else { return false; }; let Some(trusted_worktrees) = TrustedWorktrees::try_get_global(cx) else { return false; }; worktree_ids.iter().any(|worktree_id| { trusted_worktrees.update(cx, |trusted_worktrees, cx| { trusted_worktrees.can_trust(&self.worktree_store, *worktree_id, cx) }) }) } /// Update our list of repositories and schedule git scans in response to a notification from a worktree, fn update_repositories_from_worktree( &mut self, worktree_id: WorktreeId, project_environment: Entity, next_repository_id: Arc, updates_tx: Option>, updated_git_repositories: UpdatedGitRepositoriesSet, fs: Arc, cx: &mut Context, ) { let mut removed_ids = Vec::new(); let is_trusted = TrustedWorktrees::try_get_global(cx) .map(|trusted_worktrees| { trusted_worktrees.update(cx, |trusted_worktrees, cx| { trusted_worktrees.can_trust(&self.worktree_store, worktree_id, cx) }) }) .unwrap_or(false); for update in updated_git_repositories.iter() { if let Some((id, existing)) = self.repositories.iter().find(|(_, repo)| { let existing_work_directory_abs_path = &repo.read(cx).work_directory_abs_path; Some(existing_work_directory_abs_path) == update.old_work_directory_abs_path.as_ref() || Some(existing_work_directory_abs_path) == update.new_work_directory_abs_path.as_ref() }) { let repo_id = *id; if let Some(new_work_directory_abs_path) = update.new_work_directory_abs_path.clone() { self.worktree_ids .entry(repo_id) .or_insert_with(HashSet::new) .insert(worktree_id); let path_changed = update.old_work_directory_abs_path.as_ref() != update.new_work_directory_abs_path.as_ref(); if path_changed && let Some(dot_git_abs_path) = update.dot_git_abs_path.clone() && let Some(repository_dir_abs_path) = update.repository_dir_abs_path.clone() && let Some(common_dir_abs_path) = update.common_dir_abs_path.clone() { existing.update(cx, |existing, cx| { existing.reinitialize_local_backend( new_work_directory_abs_path, dot_git_abs_path, repository_dir_abs_path, common_dir_abs_path, project_environment.downgrade(), fs.clone(), is_trusted, cx, ); existing.schedule_scan(updates_tx.clone(), cx); }); } else { existing.update(cx, |existing, cx| { existing.snapshot.work_directory_abs_path = new_work_directory_abs_path; existing.schedule_scan(updates_tx.clone(), cx); }); } } else { if let Some(worktree_ids) = self.worktree_ids.get_mut(&repo_id) { worktree_ids.remove(&worktree_id); if worktree_ids.is_empty() { removed_ids.push(repo_id); } } } } else if let UpdatedGitRepository { new_work_directory_abs_path: Some(work_directory_abs_path), dot_git_abs_path: Some(dot_git_abs_path), repository_dir_abs_path: Some(repository_dir_abs_path), common_dir_abs_path: Some(common_dir_abs_path), .. } = update { let id = RepositoryId(next_repository_id.fetch_add(1, atomic::Ordering::Release)); let git_store = cx.weak_entity(); let repo = cx.new(|cx| { let mut repo = Repository::local( id, work_directory_abs_path.clone(), repository_dir_abs_path.clone(), common_dir_abs_path.clone(), dot_git_abs_path.clone(), project_environment.downgrade(), fs.clone(), is_trusted, git_store, cx, ); if let Some(updates_tx) = updates_tx.as_ref() { // trigger an empty `UpdateRepository` to ensure remote active_repo_id is set correctly updates_tx .unbounded_send(DownstreamUpdate::UpdateRepository(repo.snapshot())) .ok(); } repo.schedule_scan(updates_tx.clone(), cx); repo }); self._subscriptions .push(cx.subscribe(&repo, Self::on_repository_event)); self._subscriptions .push(cx.subscribe(&repo, Self::on_jobs_updated)); self.repositories.insert(id, repo); self.worktree_ids.insert(id, HashSet::from([worktree_id])); cx.emit(GitStoreEvent::RepositoryAdded); self.active_repo_id.get_or_insert_with(|| { cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id))); id }); } } for id in removed_ids { if self.active_repo_id == Some(id) { self.active_repo_id = None; cx.emit(GitStoreEvent::ActiveRepositoryChanged(None)); } self.repositories.remove(&id); if let Some(updates_tx) = updates_tx.as_ref() { updates_tx .unbounded_send(DownstreamUpdate::RemoveRepository(id)) .ok(); } } } fn on_trusted_worktrees_event( &mut self, _: Entity, event: &TrustedWorktreesEvent, cx: &mut Context, ) { if !matches!(self.state, GitStoreState::Local { .. }) { return; } let (is_trusted, event_paths) = match event { TrustedWorktreesEvent::Trusted(_, trusted_paths) => (true, trusted_paths), TrustedWorktreesEvent::Restricted(_, restricted_paths) => (false, restricted_paths), }; for (repo_id, worktree_ids) in &self.worktree_ids { if worktree_ids .iter() .any(|worktree_id| event_paths.contains(&PathTrust::Worktree(*worktree_id))) { if let Some(repo) = self.repositories.get(repo_id) { let repository_state = repo.read(cx).repository_state.clone(); cx.background_spawn(async move { if let Ok(RepositoryState::Local(state)) = repository_state.await { state.backend.set_trusted(is_trusted); } }) .detach(); } } } } fn on_buffer_store_event( &mut self, _: Entity, event: &BufferStoreEvent, cx: &mut Context, ) { match event { BufferStoreEvent::BufferAdded(buffer) => { cx.subscribe(buffer, |this, buffer, event, cx| { if let BufferEvent::LanguageChanged(_) = event { let buffer_id = buffer.read(cx).remote_id(); if let Some(diff_state) = this.diffs.get(&buffer_id) { diff_state.update(cx, |diff_state, cx| { diff_state.buffer_language_changed(buffer, cx); }); } } }) .detach(); } BufferStoreEvent::SharedBufferClosed(peer_id, buffer_id) => { if let Some(diffs) = self.shared_diffs.get_mut(peer_id) { diffs.remove(buffer_id); } } BufferStoreEvent::BufferDropped(buffer_id) => { self.diffs.remove(buffer_id); self.buffer_ids_by_index_text_buffer_id .retain(|_, main_buffer_id| main_buffer_id != buffer_id); for diffs in self.shared_diffs.values_mut() { diffs.remove(buffer_id); } } BufferStoreEvent::BufferChangedFilePath { buffer, .. } => { // Whenever a buffer's file path changes, it's possible that the // new path is actually a path that is being tracked by a git // repository. In that case, we'll want to update the buffer's // `BufferDiffState`, in case it already has one. let buffer_id = buffer.read(cx).remote_id(); let diff_state = self.diffs.get(&buffer_id); let repo = self.repository_and_path_for_buffer_id(buffer_id, cx); if let Some(diff_state) = diff_state && let Some((repo, repo_path)) = repo { let buffer = buffer.clone(); let diff_state = diff_state.clone(); let is_symlink = Self::buffer_is_symlink(&buffer, cx); cx.spawn(async move |_git_store, cx| { async { let diff_bases_change = if is_symlink { DiffBasesChange::SetBoth(None) } else { repo.update(cx, |repo, cx| { repo.load_committed_text(buffer_id, repo_path, cx) }) .await? }; diff_state.update(cx, |diff_state, cx| { let buffer_snapshot = buffer.read(cx).text_snapshot(); diff_state.diff_bases_changed( buffer_snapshot, Some(diff_bases_change), cx, ); }); anyhow::Ok(()) } .await .log_err(); }) .detach(); } } } } pub fn recalculate_buffer_diffs( &mut self, buffers: Vec>, cx: &mut Context, ) -> impl Future + use<> { let mut futures = Vec::new(); for buffer in buffers { if let Some(diff_state) = self.diffs.get_mut(&buffer.read(cx).remote_id()) { let buffer = buffer.read(cx).text_snapshot(); diff_state.update(cx, |diff_state, cx| { diff_state.recalculate_diffs(buffer.clone(), cx); futures.extend(diff_state.wait_for_recalculation().map(FutureExt::boxed)); }); futures.push(diff_state.update(cx, |diff_state, cx| { diff_state .reparse_conflict_markers(buffer, cx) .map(|_| {}) .boxed() })); } } async move { futures::future::join_all(futures).await; } } fn write_index_text_for_buffer_id( &mut self, buffer_id: BufferId, new_index_text: Option, cx: &mut Context, ) { let Some(diff_state) = self.diffs.get(&buffer_id) else { return; }; let hunk_staging_operation_count = diff_state.update(cx, |diff_state, _| { diff_state.hunk_staging_operation_count += 1; diff_state.hunk_staging_operation_count }); let Some((repo, path)) = self.repository_and_path_for_buffer_id(buffer_id, cx) else { return; }; let recv = repo.update(cx, |repo, cx| { log::debug!("hunks changed for {}", path.as_unix_str()); repo.spawn_set_index_text_job( path, new_index_text, Some(hunk_staging_operation_count), cx, ) }); cx.spawn(async move |this, cx| { if let Ok(Err(error)) = cx.background_spawn(recv).await { this.update(cx, |this, cx| { if let Some(diff_state) = this.diffs.get(&buffer_id).cloned() { diff_state.update(cx, |diff_state, cx| { diff_state.clear_pending_index_edits_and_hunks(cx); }); } cx.emit(GitStoreEvent::IndexWriteError(error)); }) .ok(); } }) .detach(); } fn local_worktree_git_repos_changed( &mut self, worktree: Entity, changed_repos: &UpdatedGitRepositoriesSet, cx: &mut Context, ) { log::debug!("local worktree repos changed"); debug_assert!(worktree.read(cx).is_local()); for repository in self.repositories.values() { repository.update(cx, |repository, cx| { let repo_abs_path = &repository.work_directory_abs_path; if changed_repos.iter().any(|update| { update.old_work_directory_abs_path.as_ref() == Some(repo_abs_path) || update.new_work_directory_abs_path.as_ref() == Some(repo_abs_path) }) { repository.reload_buffer_diff_bases(cx); cx.emit(RepositoryEvent::GitDirectoryChanged); } }); } } pub fn repositories(&self) -> &HashMap> { &self.repositories } /// Returns the main repository working directory for the given worktree. /// For normal checkouts this equals the worktree's own path. For linked /// worktrees it points back to the main worktree, if one exists. Linked /// worktrees attached to a bare repository have no main worktree path. pub fn original_repo_path_for_worktree( &self, worktree_id: WorktreeId, cx: &App, ) -> Option> { self.active_repo_id .iter() .chain(self.worktree_ids.keys()) .find(|repo_id| { self.worktree_ids .get(repo_id) .is_some_and(|ids| ids.contains(&worktree_id)) }) .and_then(|repo_id| self.repositories.get(repo_id)) .and_then(|repo| { repo.read(cx) .snapshot() .main_worktree_abs_path() .map(Arc::from) }) } pub fn status_for_buffer_id(&self, buffer_id: BufferId, cx: &App) -> Option { let (repo, path) = self.repository_and_path_for_buffer_id(buffer_id, cx)?; let status = repo.read(cx).snapshot.status_for_path(&path)?; Some(status.status) } pub fn repository_and_path_for_buffer_id( &self, buffer_id: BufferId, cx: &App, ) -> Option<(Entity, RepoPath)> { let buffer = self.buffer_store.read(cx).get(buffer_id)?; let project_path = buffer.read(cx).project_path(cx)?; self.repository_and_path_for_project_path(&project_path, cx) } pub fn repository_and_path_for_project_path( &self, path: &ProjectPath, cx: &App, ) -> Option<(Entity, RepoPath)> { let abs_path = self.worktree_store.read(cx).absolutize(path, cx)?; self.repositories .values() .filter_map(|repo| { let repo_path = repo.read(cx).abs_path_to_repo_path(&abs_path)?; Some((repo.clone(), repo_path)) }) .max_by_key(|(repo, _)| repo.read(cx).work_directory_abs_path.clone()) } pub fn git_init( &self, path: Arc, fallback_branch_name: String, cx: &App, ) -> Task> { match &self.state { GitStoreState::Local { fs, .. } => { let fs = fs.clone(); cx.background_executor() .spawn(async move { fs.git_init(&path, fallback_branch_name).await }) } GitStoreState::Remote { upstream_client, upstream_project_id: project_id, .. } => { let client = upstream_client.clone(); let project_id = *project_id; cx.background_executor().spawn(async move { client .request(proto::GitInit { project_id: project_id, abs_path: path.to_string_lossy().into_owned(), fallback_branch_name, }) .await?; Ok(()) }) } } } pub fn git_clone( &self, repo: String, path: impl Into>, cx: &App, ) -> Task> { let path = path.into(); match &self.state { GitStoreState::Local { fs, .. } => { let fs = fs.clone(); cx.background_executor() .spawn(async move { fs.git_clone(&path, &repo).await }) } GitStoreState::Remote { upstream_client, upstream_project_id, .. } => { if upstream_client.is_via_collab() { return Task::ready(Err(anyhow!( "Git Clone isn't supported for project guests" ))); } let request = upstream_client.request(proto::GitClone { project_id: *upstream_project_id, abs_path: path.to_string_lossy().into_owned(), remote_repo: repo, }); cx.background_spawn(async move { let result = request.await?; match result.success { true => Ok(()), false => Err(anyhow!("Git Clone failed")), } }) } } } pub fn git_config(&self, path: Arc, args: Vec, cx: &App) -> Task> { match &self.state { GitStoreState::Local { fs, .. } => { let fs = fs.clone(); cx.background_executor() .spawn(async move { fs.git_config(&path, args).await }) } GitStoreState::Remote { upstream_client, .. } => { // Prevent running git config commands for collab. if upstream_client.is_via_collab() { return Task::ready(Err(anyhow!( "Git Config isn't support for project guests" ))); } // TODO: Implement this for remote repositories. Task::ready(Err(anyhow!( "Git Config isn't yet supported for remote projects" ))) } } } async fn handle_update_repository( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result<()> { this.update(&mut cx, |this, cx| { let path_style = this.worktree_store.read(cx).path_style(); let mut update = envelope.payload; let id = RepositoryId::from_proto(update.id); let client = this.upstream_client().context("no upstream client")?; let repository_dir_abs_path: Option> = update .repository_dir_abs_path .as_deref() .map(|p| Path::new(p).into()); let common_dir_abs_path: Option> = update .common_dir_abs_path .as_deref() .map(|p| Path::new(p).into()); let mut repo_subscription = None; let repo = this.repositories.entry(id).or_insert_with(|| { let git_store = cx.weak_entity(); let repo = cx.new(|cx| { Repository::remote( id, Path::new(&update.abs_path).into(), repository_dir_abs_path.clone(), common_dir_abs_path.clone(), path_style, ProjectId(update.project_id), client, git_store, cx, ) }); repo_subscription = Some(cx.subscribe(&repo, Self::on_repository_event)); cx.emit(GitStoreEvent::RepositoryAdded); repo }); this._subscriptions.extend(repo_subscription); repo.update(cx, { let update = update.clone(); |repo, cx| repo.apply_remote_update(update, cx) })?; this.active_repo_id.get_or_insert_with(|| { cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id))); id }); if let Some((client, project_id)) = this.downstream_client() { update.project_id = project_id.to_proto(); client.send(update).log_err(); } Ok(()) }) } async fn handle_remove_repository( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result<()> { this.update(&mut cx, |this, cx| { let mut update = envelope.payload; let id = RepositoryId::from_proto(update.id); this.repositories.remove(&id); if let Some((client, project_id)) = this.downstream_client() { update.project_id = project_id.to_proto(); client.send(update).log_err(); } if this.active_repo_id == Some(id) { this.active_repo_id = None; cx.emit(GitStoreEvent::ActiveRepositoryChanged(None)); } cx.emit(GitStoreEvent::RepositoryRemoved(id)); }); Ok(()) } async fn handle_git_init( this: Entity, envelope: TypedEnvelope, cx: AsyncApp, ) -> Result { let path: Arc = PathBuf::from(envelope.payload.abs_path).into(); let name = envelope.payload.fallback_branch_name; cx.update(|cx| this.read(cx).git_init(path, name, cx)) .await?; Ok(proto::Ack {}) } async fn handle_git_clone( this: Entity, envelope: TypedEnvelope, cx: AsyncApp, ) -> Result { let path: Arc = PathBuf::from(envelope.payload.abs_path).into(); let repo_name = envelope.payload.remote_repo; let result = cx .update(|cx| this.read(cx).git_clone(repo_name, path, cx)) .await; Ok(proto::GitCloneResponse { success: result.is_ok(), }) } async fn handle_fetch( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let fetch_options = FetchOptions::from_proto(envelope.payload.remote); let askpass_id = envelope.payload.askpass_id; let askpass = make_remote_delegate( this, envelope.payload.project_id, repository_id, askpass_id, &mut cx, ); let remote_output = repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.fetch(fetch_options, askpass, cx) }) .await??; Ok(proto::RemoteMessageResponse { stdout: remote_output.stdout, stderr: remote_output.stderr, }) } async fn handle_push( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let askpass_id = envelope.payload.askpass_id; let askpass = make_remote_delegate( this, envelope.payload.project_id, repository_id, askpass_id, &mut cx, ); let options = envelope .payload .options .as_ref() .map(|_| match envelope.payload.options() { proto::push::PushOptions::SetUpstream => git::repository::PushOptions::SetUpstream, proto::push::PushOptions::Force => git::repository::PushOptions::Force, }); let branch_name = envelope.payload.branch_name.into(); let remote_branch_name = envelope.payload.remote_branch_name.into(); let remote_name = envelope.payload.remote_name.into(); let remote_output = repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.push( branch_name, remote_branch_name, remote_name, options, askpass, cx, ) }) .await??; Ok(proto::RemoteMessageResponse { stdout: remote_output.stdout, stderr: remote_output.stderr, }) } async fn handle_pull( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let askpass_id = envelope.payload.askpass_id; let askpass = make_remote_delegate( this, envelope.payload.project_id, repository_id, askpass_id, &mut cx, ); let branch_name = envelope.payload.branch_name.map(|name| name.into()); let remote_name = envelope.payload.remote_name.into(); let rebase = envelope.payload.rebase; let remote_message = repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.pull(branch_name, remote_name, rebase, askpass, cx) }) .await??; Ok(proto::RemoteMessageResponse { stdout: remote_message.stdout, stderr: remote_message.stderr, }) } async fn handle_stage( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let entries = envelope .payload .paths .into_iter() .map(|path| RepoPath::new(&path)) .collect::>>()?; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.stage_entries(entries, cx) }) .await?; Ok(proto::Ack {}) } async fn handle_unstage( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let entries = envelope .payload .paths .into_iter() .map(|path| RepoPath::new(&path)) .collect::>>()?; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.unstage_entries(entries, cx) }) .await?; Ok(proto::Ack {}) } async fn handle_stash( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let entries = envelope .payload .paths .into_iter() .map(|path| RepoPath::new(&path)) .collect::>>()?; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.stash_entries(entries, cx) }) .await?; Ok(proto::Ack {}) } async fn handle_stash_pop( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let stash_index = envelope.payload.stash_index.map(|i| i as usize); repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.stash_pop(stash_index, cx) }) .await?; Ok(proto::Ack {}) } async fn handle_stash_apply( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let stash_index = envelope.payload.stash_index.map(|i| i as usize); repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.stash_apply(stash_index, cx) }) .await?; Ok(proto::Ack {}) } async fn handle_stash_drop( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let stash_index = envelope.payload.stash_index.map(|i| i as usize); repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.stash_drop(stash_index, cx) }) .await??; Ok(proto::Ack {}) } async fn handle_set_index_text( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let repo_path = RepoPath::from_proto(&envelope.payload.path)?; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.spawn_set_index_text_job( repo_path, envelope.payload.text, None, cx, ) }) .await??; Ok(proto::Ack {}) } async fn handle_run_hook( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let hook = RunHook::from_proto(envelope.payload.hook).context("invalid hook")?; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.run_hook(hook, cx) }) .await??; Ok(proto::Ack {}) } async fn handle_commit( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let askpass_id = envelope.payload.askpass_id; let askpass = make_remote_delegate( this, envelope.payload.project_id, repository_id, askpass_id, &mut cx, ); let message = SharedString::from(envelope.payload.message); let name = envelope.payload.name.map(SharedString::from); let email = envelope.payload.email.map(SharedString::from); let options = envelope.payload.options.unwrap_or_default(); repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.commit( message, name.zip(email), CommitOptions { amend: options.amend, signoff: options.signoff, allow_empty: options.allow_empty, }, askpass, cx, ) }) .await??; Ok(proto::Ack {}) } async fn handle_get_remotes( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branch_name = envelope.payload.branch_name; let is_push = envelope.payload.is_push; let remotes = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.get_remotes(branch_name, is_push) }) .await??; Ok(proto::GetRemotesResponse { remotes: remotes .into_iter() .map(|remotes| proto::get_remotes_response::Remote { name: remotes.name.to_string(), }) .collect::>(), }) } async fn handle_get_worktrees( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let worktrees = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.worktrees() }) .await??; Ok(proto::GitWorktreesResponse { worktrees: worktrees .into_iter() .map(|worktree| worktree_to_proto(&worktree)) .collect::>(), }) } async fn handle_create_worktree( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let directory = PathBuf::from(envelope.payload.directory); let name = envelope.payload.name; let commit = envelope.payload.commit; let use_existing_branch = envelope.payload.use_existing_branch; let target = if name.is_empty() { CreateWorktreeTarget::Detached { base_sha: commit } } else if use_existing_branch { CreateWorktreeTarget::ExistingBranch { branch_name: name } } else { CreateWorktreeTarget::NewBranch { branch_name: name, base_sha: commit, } }; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.create_worktree(target, directory) }) .await??; Ok(proto::Ack {}) } async fn handle_remove_worktree( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let path = PathBuf::from(envelope.payload.path); let force = envelope.payload.force; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.remove_worktree(path, force) }) .await??; Ok(proto::Ack {}) } async fn handle_rename_worktree( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let old_path = PathBuf::from(envelope.payload.old_path); let new_path = PathBuf::from(envelope.payload.new_path); repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.rename_worktree(old_path, new_path) }) .await??; Ok(proto::Ack {}) } async fn handle_worktree_created_at( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let worktree_path = PathBuf::from(envelope.payload.worktree_path); let created_at = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.worktree_created_at(worktree_path) }) .await??; Ok(proto::GitWorktreeCreatedAtResponse { created_at: created_at.map(Into::into), }) } async fn handle_get_head_sha( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let head_sha = repository_handle .update(&mut cx, |repository_handle, _| repository_handle.head_sha()) .await??; Ok(proto::GitGetHeadShaResponse { sha: head_sha }) } async fn handle_get_commit_data( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let shas: Vec = envelope .payload .shas .iter() .filter_map(|s| Oid::from_str(s).ok()) .collect(); let mut commits = Vec::with_capacity(shas.len()); let mut receivers = Vec::new(); repository_handle.update(&mut cx, |repository, cx| { for &sha in &shas { match repository.fetch_commit_data(sha, true, cx) { CommitDataState::Loaded(data) => { commits.push(commit_data_to_proto(data)); } CommitDataState::Loading(Some(shared)) => { receivers.push(shared.clone()); } CommitDataState::Loading(None) => { // todo(git_graph) this could happen if the request fails, we should encode an error case debug_panic!( "This should never happen since we passed true into fetch commit data" ); } } } }); let results = future::join_all(receivers).await; commits.extend( results .into_iter() .filter_map(|result| result.ok()) .map(|data| commit_data_to_proto(&data)), ); Ok(proto::GetCommitDataResponse { commits }) } async fn handle_get_initial_graph_data( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result>> { const CHUNK_SIZE: usize = git::repository::GRAPH_CHUNK_SIZE; let payload = envelope.payload; let repository_id = RepositoryId::from_proto(payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let log_order = log_order_from_proto(payload.log_order()); let log_source = log_source_from_proto( payload .log_source .context("missing initial graph data log source")?, )?; let (subscriber_sender, subscriber_receiver) = async_channel::unbounded(); let (cached_commits, error, is_loading) = repository_handle.update(&mut cx, |repository, cx| { let response = repository.graph_data(log_source.clone(), log_order, 0..usize::MAX, cx); let cached_commits = response.commits.to_vec(); let error = response.error.clone(); let is_loading = response.is_loading; if is_loading { if let Some(graph_data) = repository .initial_graph_data .get_mut(&(log_source.clone(), log_order)) { graph_data.subscribers.push(subscriber_sender); } } (cached_commits, error, is_loading) }); let (mut response_tx, response_rx) = mpsc::unbounded(); cx.background_spawn(async move { if let Some(error) = error { if response_tx .send(Err(anyhow!(error.to_string()))) .await .is_err() { return; } return; } for commits in cached_commits.chunks(CHUNK_SIZE) { let response = proto::GetInitialGraphDataResponse { commits: commits .iter() .map(|commit| initial_graph_commit_to_proto(commit)) .collect(), }; if response_tx.send(Ok(response)).await.is_err() { return; } } if !is_loading { return; } while let Ok(chunk_result) = subscriber_receiver.recv().await { let commits = match chunk_result { Ok(commits) => commits, Err(error) => { response_tx .send(Err(anyhow!(error.to_string()))) .await .context("Failed to send error") .log_err(); return; } }; for commits in commits.chunks(CHUNK_SIZE) { let response = proto::GetInitialGraphDataResponse { commits: commits .iter() .map(|commit| initial_graph_commit_to_proto(commit)) .collect(), }; if response_tx.send(Ok(response)).await.is_err() { return; } } } }) .detach(); Ok(response_rx) } async fn handle_search_commits( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result>> { const CHUNK_SIZE: usize = 100; let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let log_source = log_source_from_proto( envelope .payload .log_source .context("missing search commit log source")?, )?; let search_args = SearchCommitArgs { query: SharedString::from(envelope.payload.query), case_sensitive: envelope.payload.case_sensitive, }; let (request_tx, request_rx) = async_channel::unbounded(); repository_handle.update(&mut cx, |repository, cx| { repository.search_commits(log_source, search_args, request_tx, cx); }); let (mut response_tx, response_rx) = mpsc::unbounded(); cx.background_spawn(async move { let mut shas = Vec::new(); while let Ok(sha) = request_rx.recv().await { shas.push(sha.to_string()); if shas.len() >= CHUNK_SIZE { if response_tx .send(Ok(proto::SearchCommitsResponse { shas: mem::take(&mut shas), })) .await .is_err() { return; } } } if !shas.is_empty() { response_tx .send(Ok(proto::SearchCommitsResponse { shas })) .await .ok(); } }) .detach(); Ok(response_rx) } async fn handle_edit_ref( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let ref_name = envelope.payload.ref_name; let commit = match envelope.payload.action { Some(proto::git_edit_ref::Action::UpdateToCommit(sha)) => Some(sha), Some(proto::git_edit_ref::Action::Delete(_)) => None, None => anyhow::bail!("GitEditRef missing action"), }; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.edit_ref(ref_name, commit) }) .await??; Ok(proto::Ack {}) } async fn handle_repair_worktrees( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.repair_worktrees() }) .await??; Ok(proto::Ack {}) } async fn handle_get_branches( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branches_scan = repository_handle .update(&mut cx, |repository_handle, _| repository_handle.branches()) .await??; Ok(proto::GitBranchesResponse { branches: branches_scan .branches .into_iter() .map(|branch| branch_to_proto(&branch)) .collect::>(), error: branches_scan.error.map(|error| error.to_string()), }) } async fn handle_get_default_branch( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branch = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.default_branch(envelope.payload.include_remote_name) }) .await?? .map(Into::into); Ok(proto::GetDefaultBranchResponse { branch }) } async fn handle_create_branch( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branch_name = envelope.payload.branch_name; let base_branch = envelope.payload.base_branch; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.create_branch(branch_name, base_branch) }) .await??; Ok(proto::Ack {}) } async fn handle_change_branch( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branch_name = envelope.payload.branch_name; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.change_branch(branch_name) }) .await??; Ok(proto::Ack {}) } async fn handle_rename_branch( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branch = envelope.payload.branch; let new_name = envelope.payload.new_name; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.rename_branch(branch, new_name) }) .await??; Ok(proto::Ack {}) } async fn handle_create_remote( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let remote_name = envelope.payload.remote_name; let remote_url = envelope.payload.remote_url; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.create_remote(remote_name, remote_url) }) .await??; Ok(proto::Ack {}) } async fn handle_delete_branch( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let is_remote = envelope.payload.is_remote; let branch_name = envelope.payload.branch_name; let force = envelope.payload.force; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.delete_branch(is_remote, branch_name, force) }) .await??; Ok(proto::Ack {}) } async fn handle_remove_remote( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let remote_name = envelope.payload.remote_name; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.remove_remote(remote_name) }) .await??; Ok(proto::Ack {}) } async fn handle_show( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let commit = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.show(envelope.payload.commit) }) .await??; Ok(proto::GitCommitDetails { sha: commit.sha.into(), message: commit.message.into(), commit_timestamp: commit.commit_timestamp, author_email: commit.author_email.into(), author_name: commit.author_name.into(), }) } async fn handle_create_checkpoint( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let checkpoint = repository_handle .update(&mut cx, |repository, _| repository.checkpoint()) .await??; Ok(proto::GitCreateCheckpointResponse { commit_sha: checkpoint.commit_sha.as_bytes().to_vec(), }) } async fn handle_create_archive_checkpoint( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let (staged_commit_sha, unstaged_commit_sha) = repository_handle .update(&mut cx, |repository, _| { repository.create_archive_checkpoint() }) .await??; Ok(proto::GitCreateArchiveCheckpointResponse { staged_commit_sha, unstaged_commit_sha, }) } async fn handle_restore_checkpoint( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let checkpoint = GitRepositoryCheckpoint { commit_sha: Oid::from_bytes(&envelope.payload.commit_sha)?, }; repository_handle .update(&mut cx, |repository, _| { repository.restore_checkpoint(checkpoint) }) .await??; Ok(proto::Ack {}) } async fn handle_restore_archive_checkpoint( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let staged_commit_sha = envelope.payload.staged_commit_sha; let unstaged_commit_sha = envelope.payload.unstaged_commit_sha; repository_handle .update(&mut cx, |repository, _| { repository.restore_archive_checkpoint(staged_commit_sha, unstaged_commit_sha) }) .await??; Ok(proto::Ack {}) } async fn handle_compare_checkpoints( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let left = GitRepositoryCheckpoint { commit_sha: Oid::from_bytes(&envelope.payload.left_commit_sha)?, }; let right = GitRepositoryCheckpoint { commit_sha: Oid::from_bytes(&envelope.payload.right_commit_sha)?, }; let equal = repository_handle .update(&mut cx, |repository, _| { repository.compare_checkpoints(left, right) }) .await??; Ok(proto::GitCompareCheckpointsResponse { equal }) } async fn handle_diff_checkpoints( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let base = GitRepositoryCheckpoint { commit_sha: Oid::from_bytes(&envelope.payload.base_commit_sha)?, }; let target = GitRepositoryCheckpoint { commit_sha: Oid::from_bytes(&envelope.payload.target_commit_sha)?, }; let diff = repository_handle .update(&mut cx, |repository, _| { repository.diff_checkpoints(base, target) }) .await??; Ok(proto::GitDiffCheckpointsResponse { diff }) } async fn handle_load_commit_diff( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let commit_diff = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.load_commit_diff(envelope.payload.commit) }) .await??; Ok(proto::LoadCommitDiffResponse { files: commit_diff .files .into_iter() .map(|file| proto::CommitFile { path: file.path.as_unix_str().to_owned(), old_text: file.old_text, new_text: file.new_text, is_binary: file.is_binary, }) .collect(), }) } async fn handle_reset( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let mode = match envelope.payload.mode() { git_reset::ResetMode::Soft => ResetMode::Soft, git_reset::ResetMode::Mixed => ResetMode::Mixed, }; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.reset(envelope.payload.commit, mode, cx) }) .await??; Ok(proto::Ack {}) } async fn handle_checkout_files( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let paths = envelope .payload .paths .iter() .map(|s| RepoPath::from_proto(s)) .collect::>>()?; repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.checkout_files(&envelope.payload.commit, paths, cx) }) .await?; Ok(proto::Ack {}) } async fn handle_add_path_to_gitignore( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let repo_path = RepoPath::from_proto(&envelope.payload.path)?; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.add_path_to_gitignore(&repo_path, envelope.payload.is_dir) }) .await??; Ok(proto::Ack {}) } async fn handle_add_path_to_git_info_exclude( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let repo_path = RepoPath::from_proto(&envelope.payload.path)?; repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.add_path_to_git_info_exclude(&repo_path, envelope.payload.is_dir) }) .await??; Ok(proto::Ack {}) } async fn handle_open_commit_message_buffer( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository = Self::repository_for_request(&this, repository_id, &mut cx)?; let buffer = repository .update(&mut cx, |repository, cx| { repository.open_commit_buffer(None, this.read(cx).buffer_store.clone(), cx) }) .await?; let buffer_id = buffer.read_with(&cx, |buffer, _| buffer.remote_id()); this.update(&mut cx, |this, cx| { this.buffer_store.update(cx, |buffer_store, cx| { buffer_store .create_buffer_for_peer( &buffer, envelope.original_sender_id.unwrap_or(envelope.sender_id), cx, ) .detach_and_log_err(cx); }) }); Ok(proto::OpenBufferResponse { buffer_id: buffer_id.to_proto(), }) } async fn handle_askpass( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository = Self::repository_for_request(&this, repository_id, &mut cx)?; let delegates = cx.update(|cx| repository.read(cx).askpass_delegates.clone()); let Some(mut askpass) = delegates.lock().remove(&envelope.payload.askpass_id) else { debug_panic!("no askpass found"); anyhow::bail!("no askpass found"); }; let response = askpass .ask_password(envelope.payload.prompt) .await .ok_or_else(|| anyhow::anyhow!("askpass cancelled"))?; delegates .lock() .insert(envelope.payload.askpass_id, askpass); // In fact, we don't quite know what we're doing here, as we're sending askpass password unencrypted, but.. Ok(proto::AskPassResponse { response: response.decrypt(IKnowWhatIAmDoingAndIHaveReadTheDocs)?, }) } async fn handle_check_for_pushed_commits( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let branches = repository_handle .update(&mut cx, |repository_handle, _| { repository_handle.check_for_pushed_commits() }) .await??; Ok(proto::CheckForPushedCommitsResponse { pushed_to: branches .into_iter() .map(|commit| commit.to_string()) .collect(), }) } async fn handle_git_diff( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId::from_proto(envelope.payload.repository_id); let repository_handle = Self::repository_for_request(&this, repository_id, &mut cx)?; let diff_type = match envelope.payload.diff_type() { proto::git_diff::DiffType::HeadToIndex => DiffType::HeadToIndex, proto::git_diff::DiffType::HeadToWorktree => DiffType::HeadToWorktree, proto::git_diff::DiffType::MergeBase => { let base_ref = envelope .payload .merge_base_ref .ok_or_else(|| anyhow!("merge_base_ref is required for MergeBase diff type"))?; DiffType::MergeBase { base_ref: base_ref.into(), } } }; let mut diff = repository_handle .update(&mut cx, |repository_handle, cx| { repository_handle.diff(diff_type, cx) }) .await??; const ONE_MB: usize = 1_000_000; if diff.len() > ONE_MB { diff = diff.chars().take(ONE_MB).collect() } Ok(proto::GitDiffResponse { diff }) } async fn handle_tree_diff( this: Entity, request: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let repository_id = RepositoryId(request.payload.repository_id); let diff_type = if request.payload.is_merge { DiffTreeType::MergeBase { base: request.payload.base.into(), head: request.payload.head.into(), } } else { DiffTreeType::Since { base: request.payload.base.into(), head: request.payload.head.into(), } }; let diff = this .update(&mut cx, |this, cx| { let repository = this.repositories().get(&repository_id)?; Some(repository.update(cx, |repo, cx| repo.diff_tree(diff_type, cx))) }) .context("missing repository")? .await??; Ok(proto::GetTreeDiffResponse { entries: diff .entries .into_iter() .map(|(path, status)| proto::TreeDiffStatus { path: path.as_ref().as_unix_str().to_owned(), status: match status { TreeDiffStatus::Added {} => proto::tree_diff_status::Status::Added.into(), TreeDiffStatus::Modified { .. } => { proto::tree_diff_status::Status::Modified.into() } TreeDiffStatus::Deleted { .. } => { proto::tree_diff_status::Status::Deleted.into() } }, oid: match status { TreeDiffStatus::Deleted { old } | TreeDiffStatus::Modified { old } => { Some(old.to_string()) } TreeDiffStatus::Added => None, }, }) .collect(), }) } async fn handle_get_blob_content( this: Entity, request: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let oid = git::Oid::from_str(&request.payload.oid)?; let repository_id = RepositoryId(request.payload.repository_id); let content = this .update(&mut cx, |this, cx| { let repository = this.repositories().get(&repository_id)?; Some(repository.update(cx, |repo, cx| repo.load_blob_content(oid, cx))) }) .context("missing repository")? .await?; Ok(proto::GetBlobContentResponse { content }) } async fn handle_open_unstaged_diff( this: Entity, request: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let buffer_id = BufferId::new(request.payload.buffer_id)?; let diff = this .update(&mut cx, |this, cx| { let buffer = this.buffer_store.read(cx).get(buffer_id)?; Some(this.open_unstaged_diff(buffer, cx)) }) .context("missing buffer")? .await?; this.update(&mut cx, |this, _| { let shared_diffs = this .shared_diffs .entry(request.original_sender_id.unwrap_or(request.sender_id)) .or_default(); shared_diffs.entry(buffer_id).or_default().unstaged = Some(diff.clone()); }); let staged_text = diff.read_with(&cx, |diff, cx| diff.base_text_string(cx)); Ok(proto::OpenUnstagedDiffResponse { staged_text }) } async fn handle_open_uncommitted_diff( this: Entity, request: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let buffer_id = BufferId::new(request.payload.buffer_id)?; let diff = this .update(&mut cx, |this, cx| { let buffer = this.buffer_store.read(cx).get(buffer_id)?; Some(this.open_uncommitted_diff(buffer, cx)) }) .context("missing buffer")? .await?; this.update(&mut cx, |this, _| { let shared_diffs = this .shared_diffs .entry(request.original_sender_id.unwrap_or(request.sender_id)) .or_default(); shared_diffs.entry(buffer_id).or_default().uncommitted = Some(diff.clone()); }); this.read_with(&cx, |this, cx| { use proto::open_uncommitted_diff_response::Mode; let diff_state = this.diffs.get(&buffer_id).context("missing diff state")?; let diff_state = diff_state.read(cx); let index_matches_head = diff_state.index_matches_head(); let index_text = diff_state.index_text.clone(); let head_text = diff_state.head_text.clone(); let response = if index_matches_head { proto::OpenUncommittedDiffResponse { committed_text: head_text.map(|head| head.to_string()), staged_text: None, mode: Mode::IndexMatchesHead.into(), } } else { proto::OpenUncommittedDiffResponse { committed_text: head_text.map(|head| head.to_string()), staged_text: index_text.map(|index| index.to_string()), mode: Mode::IndexAndHead.into(), } }; anyhow::Ok(response) }) } async fn handle_update_diff_bases( this: Entity, request: TypedEnvelope, mut cx: AsyncApp, ) -> Result<()> { let buffer_id = BufferId::new(request.payload.buffer_id)?; this.update(&mut cx, |this, cx| { if let Some(diff_state) = this.diffs.get_mut(&buffer_id) && let Some(buffer) = this.buffer_store.read(cx).get(buffer_id) { let buffer = buffer.read(cx).text_snapshot(); diff_state.update(cx, |diff_state, cx| { diff_state.handle_base_texts_updated(buffer, request.payload, cx); }) } }); Ok(()) } async fn handle_blame_buffer( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let buffer_id = BufferId::new(envelope.payload.buffer_id)?; let version = deserialize_version(&envelope.payload.version); let buffer = this.read_with(&cx, |this, cx| { this.buffer_store.read(cx).get_existing(buffer_id) })?; buffer .update(&mut cx, |buffer, _| { buffer.wait_for_version(version.clone()) }) .await?; let blame = this .update(&mut cx, |this, cx| { this.blame_buffer(&buffer, Some(version), cx) }) .await?; Ok(serialize_blame_buffer_response(blame)) } async fn handle_get_permalink_to_line( this: Entity, envelope: TypedEnvelope, mut cx: AsyncApp, ) -> Result { let buffer_id = BufferId::new(envelope.payload.buffer_id)?; // let version = deserialize_version(&envelope.payload.version); let selection = { let proto_selection = envelope .payload .selection .context("no selection to get permalink for defined")?; proto_selection.start as u32..proto_selection.end as u32 }; let buffer = this.read_with(&cx, |this, cx| { this.buffer_store.read(cx).get_existing(buffer_id) })?; let permalink = this .update(&mut cx, |this, cx| { this.get_permalink_to_line(&buffer, selection, cx) }) .await?; Ok(proto::GetPermalinkToLineResponse { permalink: permalink.to_string(), }) } fn repository_for_request( this: &Entity, id: RepositoryId, cx: &mut AsyncApp, ) -> Result> { this.read_with(cx, |this, _| { this.repositories .get(&id) .context("missing repository handle") .cloned() }) } pub fn repo_snapshots(&self, cx: &App) -> HashMap { self.repositories .iter() .map(|(id, repo)| (*id, repo.read(cx).snapshot.clone())) .collect() } fn coalesce_repo_paths(mut paths: Vec) -> Vec { paths.sort(); let mut coalesced = Vec::with_capacity(paths.len()); for path in paths { if coalesced .last() .is_some_and(|ancestor: &RepoPath| path.starts_with(ancestor)) { continue; } coalesced.push(path); } coalesced } fn process_updated_entries( &self, worktree: &Entity, updated_entries: &[(Arc, ProjectEntryId, PathChange)], cx: &mut App, ) -> Task, Vec>> { let path_style = worktree.read(cx).path_style(); let mut repo_paths = self .repositories .values() .map(|repo| (repo.read(cx).work_directory_abs_path.clone(), repo.clone())) .collect::>(); let mut entries: Vec<_> = updated_entries .iter() .map(|(path, _, _)| path.clone()) .collect(); entries.sort(); let worktree = worktree.read(cx); let entries = entries .into_iter() .map(|path| worktree.absolutize(&path)) .collect::>(); let executor = cx.background_executor().clone(); cx.background_executor().spawn(async move { repo_paths.sort_by(|lhs, rhs| lhs.0.cmp(&rhs.0)); let mut paths_by_git_repo = HashMap::<_, Vec<_>>::default(); let mut tasks = FuturesOrdered::new(); for (repo_path, repo) in repo_paths.into_iter().rev() { let entries = entries.clone(); let task = executor.spawn(async move { // Find all repository paths that belong to this repo let mut ix = entries.partition_point(|path| path < &*repo_path); if ix == entries.len() { return None; }; let mut paths = Vec::new(); // All paths prefixed by a given repo will constitute a continuous range. while let Some(path) = entries.get(ix) && let Some(repo_path) = RepositorySnapshot::abs_path_to_repo_path_inner( &repo_path, path, path_style, ) { paths.push((repo_path, ix)); ix += 1; } if paths.is_empty() { None } else { Some((repo, paths)) } }); tasks.push_back(task); } // Now, let's filter out the "duplicate" entries that were processed by multiple distinct repos. let mut path_was_used = vec![false; entries.len()]; let tasks = tasks.collect::>().await; // Process tasks from the back: iterating backwards allows us to see more-specific paths first. // We always want to assign a path to it's innermost repository. for t in tasks { let Some((repo, paths)) = t else { continue; }; let entry = paths_by_git_repo.entry(repo).or_default(); for (repo_path, ix) in paths { if path_was_used[ix] { continue; } path_was_used[ix] = true; entry.push(repo_path); } } for paths in paths_by_git_repo.values_mut() { *paths = Self::coalesce_repo_paths(mem::take(paths)); } paths_by_git_repo }) } } impl BufferGitState { fn new(_git_store: WeakEntity, _cx: &mut Context) -> Self { Self { unstaged_diff: Default::default(), staged_diff: Default::default(), uncommitted_diff: Default::default(), oid_diffs: Default::default(), recalculate_diff_task: Default::default(), language: Default::default(), language_registry: Default::default(), recalculating_tx: postage::watch::channel_with(false).0, hunk_staging_operation_count: 0, hunk_staging_operation_count_as_of_write: 0, head_text: Default::default(), index_text: Default::default(), pending_index_edits: Some(Vec::new()), oid_texts: Default::default(), head_text_buffer: WeakEntity::new_invalid(), index_text_buffer: WeakEntity::new_invalid(), index_text_buffer_language_enabled: Default::default(), head_changed: Default::default(), index_changed: Default::default(), language_changed: Default::default(), conflict_updated_futures: Default::default(), conflict_set: Default::default(), reparse_conflict_markers_task: Default::default(), } } fn get_or_create_head_text_buffer(&mut self, cx: &mut Context) -> Entity { if let Some(buffer) = self.head_text_buffer.upgrade() { return buffer; } let head_text = self.head_text.clone(); let buffer = cx.new(|cx| { let mut buffer = Buffer::local(head_text.as_deref().unwrap_or(""), cx); buffer.set_capability(Capability::ReadOnly, cx); buffer }); self.head_text_buffer = buffer.downgrade(); buffer } fn index_text_buffer(&self) -> Option> { self.index_text_buffer.upgrade() } fn get_or_create_index_text_buffer( &mut self, file: Option>, cx: &mut Context, ) -> Entity { if let Some(buffer) = self.index_text_buffer.upgrade() { if let Some(file) = file { buffer.update(cx, |buffer, cx| buffer.file_updated(file, cx)); } return buffer; } let index_text = self.index_text.clone(); let buffer = cx.new(|cx| { let mut buffer = Buffer::local(index_text.as_deref().unwrap_or(""), cx); if let Some(file) = file { buffer.file_updated(file, cx); } buffer.set_capability(Capability::ReadOnly, cx); buffer }); self.index_text_buffer = buffer.downgrade(); buffer } #[ztracing::instrument(skip_all)] fn buffer_language_changed(&mut self, buffer: Entity, cx: &mut Context) { self.language = buffer.read(cx).language().cloned(); self.language_changed = true; let _ = self.recalculate_diffs(buffer.read(cx).text_snapshot(), cx); } fn reparse_conflict_markers( &mut self, buffer: text::BufferSnapshot, cx: &mut Context, ) -> oneshot::Receiver<()> { let (tx, rx) = oneshot::channel(); let Some(conflict_set) = self .conflict_set .as_ref() .and_then(|conflict_set| conflict_set.upgrade()) else { return rx; }; let has_conflict = conflict_set.read_with(cx, |conflict_set, _| conflict_set.has_conflict); if !has_conflict { return rx; } let old_snapshot = conflict_set.read_with(cx, |conflict_set, _| conflict_set.snapshot()); self.conflict_updated_futures.push(tx); self.reparse_conflict_markers_task = Some(cx.spawn(async move |this, cx| { let (snapshot, changed_range) = cx .background_spawn(async move { let new_snapshot = ConflictSet::parse(&buffer); let changed_range = old_snapshot.compare(&new_snapshot, &buffer); (new_snapshot, changed_range) }) .await; this.update(cx, |this, cx| { if let Some(conflict_set) = &this.conflict_set { conflict_set .update(cx, |conflict_set, cx| { conflict_set.set_snapshot(snapshot, changed_range, cx); }) .ok(); } let futures = std::mem::take(&mut this.conflict_updated_futures); for tx in futures { tx.send(()).ok(); } }) })); rx } fn unstaged_diff(&self) -> Option> { self.unstaged_diff.as_ref().and_then(|set| set.upgrade()) } fn staged_diff(&self) -> Option> { self.staged_diff.as_ref().and_then(|(set, _)| set.upgrade()) } fn staged_diff_and_index_text_buffer(&self) -> Option<(Entity, Entity)> { let (diff, index_text_buffer) = self.staged_diff.as_ref()?; Some((diff.upgrade()?, index_text_buffer.clone())) } fn uncommitted_diff(&self) -> Option> { self.uncommitted_diff.as_ref().and_then(|set| set.upgrade()) } fn oid_diff(&self, oid: Option) -> Option> { self.oid_diffs.get(&oid).and_then(|weak| weak.upgrade()) } /// Whether the index text is known to match the committed text, without /// comparing their contents. Always true when both texts were set by a /// single `DiffBasesChange::SetBoth`, which shares one allocation between /// them. May be false even when the contents are equal, if the texts were /// loaded separately. fn index_matches_head(&self) -> bool { match (self.index_text.as_ref(), self.head_text.as_ref()) { (Some(index), Some(head)) => Arc::ptr_eq(index, head), (None, None) => true, _ => false, } } fn remove_overlapping_pending_index_edits(&mut self, ranges: &[Range]) { if let Some(edits) = &mut self.pending_index_edits { edits.retain(|(existing, _)| { ranges.iter().all(|footprint| { existing.end < footprint.start || footprint.end < existing.start }) }); } } fn insert_pending_index_edits(&mut self, edits: Option, Arc)>>) { match edits { None => { self.pending_index_edits = None; } Some(new_edits) => { let mut edits = self.pending_index_edits.take().unwrap_or_default(); for (range, replacement) in new_edits { edits.retain(|(existing, _)| { existing.end < range.start || range.end < existing.start }); let position = edits.partition_point(|(existing, _)| existing.start < range.start); edits.insert(position, (range, replacement)); } self.pending_index_edits = Some(edits); } } } fn pending_index_text(&self, cx: &App) -> Option { let index_text_buffer = self.index_text_buffer.upgrade()?; let edits = self.pending_index_edits.as_ref()?; #[cfg(debug_assertions)] for window in edits.windows(2) { debug_assert!(window[0].0.end <= window[1].0.start); } let mut index_text = index_text_buffer.read(cx).text_snapshot().as_rope().clone(); for (old_range, replacement_text) in edits.iter().rev() { index_text.replace(old_range.clone(), replacement_text); } Some(index_text) } fn clear_pending_index_edits(&mut self) { self.pending_index_edits = Some(Vec::new()); } fn clear_pending_hunks(&mut self, cx: &mut Context) { for diff in [ self.uncommitted_diff(), self.unstaged_diff(), self.staged_diff(), ] .into_iter() .flatten() { diff.update(cx, |diff, cx| diff.clear_pending_hunks(cx)); } } fn mark_whole_file_stage_or_unstage_pending( &mut self, stage: bool, buffer_snapshot: &text::BufferSnapshot, cx: &mut Context, ) { if let Some(uncommitted_diff) = self.uncommitted_diff() { uncommitted_diff.update(cx, |uncommitted_diff, cx| { uncommitted_diff.mark_all_hunks_pending(stage, buffer_snapshot, cx); }); } if stage { if let Some(unstaged_diff) = self.unstaged_diff() { unstaged_diff.update(cx, |unstaged_diff, cx| { unstaged_diff.suppress_all_hunks_pending(buffer_snapshot, cx); }); } } else if let Some(staged_diff) = self.staged_diff() && let Some(index_text_buffer) = self.index_text_buffer() { let index_snapshot = index_text_buffer.read(cx).text_snapshot(); staged_diff.update(cx, |staged_diff, cx| { staged_diff.suppress_all_hunks_pending(&index_snapshot, cx); }); } } fn clear_pending_index_edits_and_hunks(&mut self, cx: &mut Context) { self.clear_pending_index_edits(); self.clear_pending_hunks(cx); } fn handle_base_texts_updated( &mut self, buffer: text::BufferSnapshot, message: proto::UpdateDiffBases, cx: &mut Context, ) { use proto::update_diff_bases::Mode; let Some(mode) = Mode::from_i32(message.mode) else { return; }; let diff_bases_change = match mode { Mode::HeadOnly => Some(DiffBasesChange::SetHead(message.committed_text)), Mode::IndexOnly => Some(DiffBasesChange::SetIndex(message.staged_text)), Mode::IndexMatchesHead => Some(DiffBasesChange::SetBoth(message.committed_text)), Mode::IndexAndHead => Some(DiffBasesChange::SetEach { index: message.staged_text, head: message.committed_text, }), Mode::Unchanged => None, }; self.diff_bases_changed(buffer, diff_bases_change, cx); } pub fn wait_for_recalculation(&mut self) -> Option + use<>> { if *self.recalculating_tx.borrow() { let mut rx = self.recalculating_tx.subscribe(); Some(async move { loop { let is_recalculating = rx.recv().await; if is_recalculating != Some(true) { break; } } }) } else { None } } fn diff_bases_changed( &mut self, buffer: text::BufferSnapshot, diff_bases_change: Option, cx: &mut Context, ) { match diff_bases_change { Some(DiffBasesChange::SetIndex(index)) => { self.index_text = index.map(|mut index| { text::LineEnding::normalize(&mut index); Arc::from(index.as_str()) }); self.index_changed = true; } Some(DiffBasesChange::SetHead(head)) => { self.head_text = head.map(|mut head| { text::LineEnding::normalize(&mut head); Arc::from(head.as_str()) }); self.head_changed = true; } Some(DiffBasesChange::SetBoth(text)) => { let text = text.map(|mut text| { text::LineEnding::normalize(&mut text); Arc::from(text.as_str()) }); self.head_text = text.clone(); self.index_text = text; self.head_changed = true; self.index_changed = true; } Some(DiffBasesChange::SetEach { index, head }) => { self.index_text = index.map(|mut index| { text::LineEnding::normalize(&mut index); Arc::from(index.as_str()) }); self.index_changed = true; self.head_text = head.map(|mut head| { text::LineEnding::normalize(&mut head); Arc::from(head.as_str()) }); self.head_changed = true; } None => {} } self.recalculate_diffs(buffer, cx) } #[ztracing::instrument(skip_all)] fn recalculate_diffs(&mut self, buffer: text::BufferSnapshot, cx: &mut Context) { *self.recalculating_tx.borrow_mut() = true; let language = self.language.clone(); let language_registry = self.language_registry.clone(); let unstaged_diff = self.unstaged_diff(); let staged_diff = self.staged_diff(); let uncommitted_diff = self.uncommitted_diff(); let head = self.head_text.clone(); let index = self.index_text.clone(); let head_text_buffer = self.head_text_buffer.upgrade(); let index_text_buffer = self.index_text_buffer.upgrade(); let index_text_buffer_language_enabled = self.index_text_buffer_language_enabled; let index_changed = self.index_changed; let head_changed = self.head_changed; let language_changed = self.language_changed; let prev_hunk_staging_operation_count = self.hunk_staging_operation_count_as_of_write; let index_matches_head = self.index_matches_head(); let oid_diffs: Vec<( Option, Entity, Entity, Option>, )> = self .oid_diffs .iter() .filter_map(|(oid, weak)| { let diff = weak.upgrade()?; let base_text_buffer = diff.read(cx).base_text_buffer().clone(); let base_text = match oid { Some(oid) => Some(self.oid_texts.get(oid)?.clone()), None => None, }; Some((*oid, diff, base_text_buffer, base_text)) }) .collect(); self.oid_diffs.retain(|oid, weak| { let alive = weak.upgrade().is_some(); if !alive { if let Some(oid) = oid { self.oid_texts.remove(oid); } } alive }); if self .staged_diff .as_ref() .is_some_and(|(weak, _)| !weak.is_upgradable()) { self.staged_diff = None; } self.recalculate_diff_task = Some(cx.spawn(async move |this, cx| { log::debug!( "start recalculating diffs for buffer {}", buffer.remote_id() ); if index_text_buffer_language_enabled && let Some(index_text_buffer) = &index_text_buffer { index_text_buffer.update(cx, |index_text_buffer, cx| { if let Some(language_registry) = language_registry.clone() { index_text_buffer.set_language_registry(language_registry); } index_text_buffer.set_language_async(language.clone(), cx); }); } if let Some(head_text_buffer) = &head_text_buffer { head_text_buffer.update(cx, |head_text_buffer, cx| { if let Some(language_registry) = language_registry.clone() { head_text_buffer.set_language_registry(language_registry); } head_text_buffer.set_language_async(language.clone(), cx); }); } for (_, _, base_text_buffer, _) in &oid_diffs { base_text_buffer.update(cx, |base_text_buffer, cx| { if let Some(language_registry) = language_registry.clone() { base_text_buffer.set_language_registry(language_registry); } base_text_buffer.set_language_async(language.clone(), cx); }); } let mut edited_index_text = None; let index_text_snapshot = if let Some(index_text_buffer) = &index_text_buffer && (unstaged_diff.is_some() || staged_diff.is_some()) { let index_text_snapshot = if index_changed || language_changed { let new_index_text = index.clone().unwrap_or_default(); let index_text_diff = index_text_buffer .update(cx, |index_text_buffer, cx| { index_text_buffer.diff(new_index_text.clone(), cx) }) .await; let edited = index_text_buffer .update(cx, |index_text_buffer, cx| { index_text_buffer.snapshot_with_edits(index_text_diff.edits, cx) }) .await; let snapshot = edited.snapshot().clone(); edited_index_text = Some(edited); snapshot } else { index_text_buffer.read_with(cx, |buffer, _| buffer.snapshot()) }; Some(index_text_snapshot) } else { None }; let mut new_unstaged_diff = None; if let (Some(unstaged_diff), Some(index_text_snapshot)) = (unstaged_diff.as_ref(), index_text_snapshot.as_ref()) { new_unstaged_diff = Some( cx.update(|cx| { unstaged_diff.read(cx).update_diff( buffer.clone(), index_text_snapshot, index.clone(), cx, ) }) .await, ); } // Dropping BufferDiff can be expensive, so yield back to the event loop // for a bit yield_now().await; let mut edited_head_text = None; let mut new_staged_diff = None; let mut new_uncommitted_diff = None; if let Some(head_text_buffer) = &head_text_buffer && (staged_diff.is_some() || uncommitted_diff.is_some()) { let head_base_text_exists = head.is_some(); let head_text_snapshot = if head_changed || language_changed { let new_head_text = head.clone().unwrap_or_default(); let head_text_diff = head_text_buffer .update(cx, |head_text_buffer, cx| { head_text_buffer.diff(new_head_text.clone(), cx) }) .await; let edited = head_text_buffer .update(cx, |base_text_buffer, cx| { base_text_buffer.snapshot_with_edits(head_text_diff.edits, cx) }) .await; let snapshot = edited.snapshot().clone(); edited_head_text = Some(edited); snapshot } else { head_text_buffer.read_with(cx, |buffer, _| buffer.snapshot()) }; if let (Some(staged_diff), Some(index_base_text_snapshot)) = (staged_diff.as_ref(), index_text_snapshot.as_ref()) { new_staged_diff = Some( cx.update(|cx| { staged_diff.read(cx).update_diff( index_base_text_snapshot.text.clone(), &head_text_snapshot, head.clone(), cx, ) }) .await, ); } if let Some(uncommitted_diff) = &uncommitted_diff { new_uncommitted_diff = if index_matches_head { new_unstaged_diff.clone().map(|mut update| { update.set_base_text_snapshot( head_text_snapshot.clone(), head_base_text_exists, ); update }) } else { None }; if new_uncommitted_diff.is_none() { new_uncommitted_diff = Some( cx.update(|cx| { uncommitted_diff.read(cx).update_diff( buffer.clone(), &head_text_snapshot, head.clone(), cx, ) }) .await, ); } } } // Dropping BufferDiff can be expensive, so yield back to the event loop // for a bit yield_now().await; let cancel = this.update(cx, |this, _| { // This checks whether all pending stage/unstage operations // have quiesced (i.e. both the corresponding write and the // read of that write have completed). If not, then we cancel // this recalculation attempt to avoid invalidating pending // state too quickly; another recalculation will come along // later and clear the pending state once the state of the index has settled. if this.hunk_staging_operation_count > prev_hunk_staging_operation_count { *this.recalculating_tx.borrow_mut() = false; true } else { false } })?; if cancel { log::debug!( concat!( "aborting recalculating diffs for buffer {}", "due to subsequent hunk operations", ), buffer.remote_id() ); return Ok(()); } this.update(cx, |this, cx| { this.clear_pending_index_edits(); if let (Some(staged_diff), Some(new_staged_diff)) = (staged_diff.as_ref(), new_staged_diff.clone()) { staged_diff.update(cx, |diff, cx| { if let Some(edited_base_text) = edited_index_text.take() && let Some(index_text_buffer) = &index_text_buffer { index_text_buffer.update(cx, |index_text_buffer, cx| { index_text_buffer.fast_forward(edited_base_text, cx) }); } if let Some(edited_head_text) = edited_head_text.take() && let Some(head_text_buffer) = &head_text_buffer { head_text_buffer.update(cx, |head_text_buffer, cx| { head_text_buffer.fast_forward(edited_head_text, cx) }); } diff.set_snapshot_with_secondary(new_staged_diff, None, true, cx) }); } let unstaged_changed_range = if let (Some(unstaged_diff), Some(new_unstaged_diff)) = (unstaged_diff.as_ref(), new_unstaged_diff.clone()) { Some(unstaged_diff.update(cx, |diff, cx| { if let Some(edited_index_text) = edited_index_text.take() && let Some(index_text_buffer) = &index_text_buffer { index_text_buffer.update(cx, |index_text_buffer, cx| { index_text_buffer.fast_forward(edited_index_text, cx) }); } diff.set_snapshot_with_secondary(new_unstaged_diff, None, true, cx) })) } else { None }; if let (Some(uncommitted_diff), Some(new_uncommitted_diff)) = (uncommitted_diff.as_ref(), new_uncommitted_diff.clone()) { uncommitted_diff.update(cx, |diff, cx| { if let Some(edited_base_text) = edited_head_text.take() && let Some(head_text_buffer) = &head_text_buffer { head_text_buffer.update(cx, |head_text_buffer, cx| { head_text_buffer.fast_forward(edited_base_text, cx) }); } diff.set_snapshot_with_secondary( new_uncommitted_diff, unstaged_changed_range.flatten(), true, cx, ) }); } })?; yield_now().await; for (oid, oid_diff, base_text_buffer, base_text) in oid_diffs { let base_text_snapshot = base_text_buffer.read_with(cx, |buffer, _| buffer.snapshot()); let new_oid_diff = cx .update(|cx| { oid_diff.read(cx).update_diff( buffer.clone(), &base_text_snapshot, base_text.clone(), cx, ) }) .await; oid_diff.update(cx, |diff, cx| diff.set_snapshot(new_oid_diff, cx)); log::debug!( "finished recalculating oid diff for buffer {} oid {:?}", buffer.remote_id(), oid ); yield_now().await; } log::debug!( "finished recalculating diffs for buffer {}", buffer.remote_id() ); if let Some(this) = this.upgrade() { this.update(cx, |this, _| { this.index_changed = false; this.head_changed = false; this.language_changed = false; *this.recalculating_tx.borrow_mut() = false; }); } Ok(()) })); } } fn make_remote_delegate( this: Entity, project_id: u64, repository_id: RepositoryId, askpass_id: u64, cx: &mut AsyncApp, ) -> AskPassDelegate { AskPassDelegate::new(cx, move |prompt, tx, cx| { this.update(cx, |this, cx| { let Some((client, _)) = this.downstream_client() else { return; }; let response = client.request(proto::AskPassRequest { project_id, repository_id: repository_id.to_proto(), askpass_id, prompt, }); cx.spawn(async move |_, _| { let mut response = response.await?.response; tx.send(EncryptedPassword::try_from(response.as_ref())?) .ok(); response.zeroize(); anyhow::Ok(()) }) .detach_and_log_err(cx); }); }) } impl RepositoryId { pub fn to_proto(self) -> u64 { self.0 } pub fn from_proto(id: u64) -> Self { RepositoryId(id) } } impl RepositorySnapshot { fn empty( id: RepositoryId, work_directory_abs_path: Arc, repository_dir_abs_path: Option>, dot_git_abs_path: Option>, common_dir_abs_path: Option>, path_style: PathStyle, ) -> Self { let repository_dir_abs_path = repository_dir_abs_path.unwrap_or_else(|| work_directory_abs_path.join(".git").into()); let dot_git_abs_path = dot_git_abs_path.unwrap_or_else(|| work_directory_abs_path.join(".git").into()); let common_dir_abs_path = common_dir_abs_path.unwrap_or_else(|| repository_dir_abs_path.clone()); Self { id, statuses_by_path: Default::default(), repository_dir_abs_path, dot_git_abs_path, common_dir_abs_path, work_directory_abs_path, branch: None, branch_list: Arc::from([]), branch_list_error: None, head_commit: None, scan_id: 0, merge: Default::default(), remote_origin_url: None, remote_upstream_url: None, stash_entries: Default::default(), linked_worktrees: Arc::from([]), path_style, } } fn initial_update(&self, project_id: u64) -> proto::UpdateRepository { proto::UpdateRepository { branch_summary: self.branch.as_ref().map(branch_to_proto), branch_list: self.branch_list.iter().map(branch_to_proto).collect(), branch_list_error: self .branch_list_error .as_ref() .map(|error| error.to_string()), head_commit_details: self.head_commit.as_ref().map(commit_details_to_proto), updated_statuses: self .statuses_by_path .iter() .map(|entry| entry.to_proto()) .collect(), removed_statuses: Default::default(), current_merge_conflicts: self .merge .merge_heads_by_conflicted_path .iter() .map(|(repo_path, _)| repo_path.as_unix_str().to_owned()) .collect(), merge_message: self.merge.message.as_ref().map(|msg| msg.to_string()), project_id, id: self.id.to_proto(), abs_path: self.work_directory_abs_path.to_string_lossy().into_owned(), entry_ids: vec![self.id.to_proto()], scan_id: self.scan_id, is_last_update: true, stash_entries: self .stash_entries .entries .iter() .map(stash_to_proto) .collect(), remote_upstream_url: self.remote_upstream_url.clone(), remote_origin_url: self.remote_origin_url.clone(), repository_dir_abs_path: Some( self.repository_dir_abs_path.to_string_lossy().into_owned(), ), common_dir_abs_path: Some(self.common_dir_abs_path.to_string_lossy().into_owned()), linked_worktrees: self .linked_worktrees .iter() .map(worktree_to_proto) .collect(), } } fn build_update(&self, old: &Self, project_id: u64) -> proto::UpdateRepository { let mut updated_statuses: Vec = Vec::new(); let mut removed_statuses: Vec = Vec::new(); let mut new_statuses = self.statuses_by_path.iter().peekable(); let mut old_statuses = old.statuses_by_path.iter().peekable(); let mut current_new_entry = new_statuses.next(); let mut current_old_entry = old_statuses.next(); loop { match (current_new_entry, current_old_entry) { (Some(new_entry), Some(old_entry)) => { match new_entry.repo_path.cmp(&old_entry.repo_path) { Ordering::Less => { updated_statuses.push(new_entry.to_proto()); current_new_entry = new_statuses.next(); } Ordering::Equal => { if new_entry.status != old_entry.status || new_entry.diff_stat != old_entry.diff_stat { updated_statuses.push(new_entry.to_proto()); } current_old_entry = old_statuses.next(); current_new_entry = new_statuses.next(); } Ordering::Greater => { removed_statuses.push(old_entry.repo_path.as_unix_str().to_owned()); current_old_entry = old_statuses.next(); } } } (None, Some(old_entry)) => { removed_statuses.push(old_entry.repo_path.as_unix_str().to_owned()); current_old_entry = old_statuses.next(); } (Some(new_entry), None) => { updated_statuses.push(new_entry.to_proto()); current_new_entry = new_statuses.next(); } (None, None) => break, } } proto::UpdateRepository { branch_summary: self.branch.as_ref().map(branch_to_proto), branch_list: self.branch_list.iter().map(branch_to_proto).collect(), branch_list_error: self .branch_list_error .as_ref() .map(|error| error.to_string()), head_commit_details: self.head_commit.as_ref().map(commit_details_to_proto), updated_statuses, removed_statuses, current_merge_conflicts: self .merge .merge_heads_by_conflicted_path .iter() .map(|(path, _)| path.as_unix_str().to_owned()) .collect(), merge_message: self.merge.message.as_ref().map(|msg| msg.to_string()), project_id, id: self.id.to_proto(), abs_path: self.work_directory_abs_path.to_string_lossy().into_owned(), entry_ids: vec![], scan_id: self.scan_id, is_last_update: true, stash_entries: self .stash_entries .entries .iter() .map(stash_to_proto) .collect(), remote_upstream_url: self.remote_upstream_url.clone(), remote_origin_url: self.remote_origin_url.clone(), repository_dir_abs_path: Some( self.repository_dir_abs_path.to_string_lossy().into_owned(), ), common_dir_abs_path: Some(self.common_dir_abs_path.to_string_lossy().into_owned()), linked_worktrees: self .linked_worktrees .iter() .map(worktree_to_proto) .collect(), } } /// Returns the main worktree path for this repository, if one exists. /// /// Linked worktrees attached to bare repositories do not have a main /// worktree. For linked worktrees attached to a non-bare repository, the /// common Git directory is the main worktree's `.git` directory. pub fn main_worktree_abs_path(&self) -> Option<&Path> { if self.is_linked_worktree() { if self.common_dir_abs_path.file_name()? == std::ffi::OsStr::new(".git") { self.common_dir_abs_path.parent() } else { None } } else { Some(self.work_directory_abs_path.as_ref()) } } /// The main worktree is the original checkout that other worktrees were /// created from. /// /// For example, if you had both `~/code/zed` and `~/code/worktrees/zed-2`, /// then `~/code/zed` is the main worktree and `~/code/worktrees/zed-2` is a linked worktree. /// /// Submodules also return `true` here, since they are not linked worktrees. pub fn is_main_worktree(&self) -> bool { !self.is_linked_worktree() } /// Returns true if this repository is a linked worktree, that is, one that /// was created from another worktree. /// /// Returns `false` for both the main worktree and submodules. pub fn is_linked_worktree(&self) -> bool { self.repository_dir_abs_path != self.common_dir_abs_path } pub fn linked_worktrees(&self) -> &[GitWorktree] { &self.linked_worktrees } pub fn status(&self) -> impl Iterator + '_ { self.statuses_by_path.iter().cloned() } pub fn status_summary(&self) -> GitSummary { self.statuses_by_path.summary().item_summary } pub fn status_for_path(&self, path: &RepoPath) -> Option { self.statuses_by_path .get(&PathKey(path.as_ref().clone()), ()) .cloned() } pub fn diff_stat_for_path(&self, path: &RepoPath) -> Option { self.statuses_by_path .get(&PathKey(path.as_ref().clone()), ()) .and_then(|entry| entry.diff_stat) } pub fn abs_path_to_repo_path(&self, abs_path: &Path) -> Option { Self::abs_path_to_repo_path_inner(&self.work_directory_abs_path, abs_path, self.path_style) } fn repo_path_to_abs_path(&self, repo_path: &RepoPath) -> PathBuf { let repo_path = repo_path.display(self.path_style); PathBuf::from( self.path_style .join(&self.work_directory_abs_path, repo_path.as_ref()) .unwrap(), ) } #[inline] fn abs_path_to_repo_path_inner( work_directory_abs_path: &Path, abs_path: &Path, path_style: PathStyle, ) -> Option { let rel_path = path_style.strip_prefix(abs_path, work_directory_abs_path)?; Some(RepoPath::from_rel_path(&rel_path)) } pub fn had_conflict_on_last_merge_head_change(&self, repo_path: &RepoPath) -> bool { self.merge .merge_heads_by_conflicted_path .contains_key(repo_path) } pub fn has_conflict(&self, repo_path: &RepoPath) -> bool { let had_conflict_on_last_merge_head_change = self .merge .merge_heads_by_conflicted_path .contains_key(repo_path); let has_conflict_currently = self .status_for_path(repo_path) .is_some_and(|entry| entry.status.is_conflicted()); had_conflict_on_last_merge_head_change || has_conflict_currently } /// This is the name that will be displayed in the repository selector for this repository. pub fn display_name(&self) -> SharedString { self.work_directory_abs_path .file_name() .unwrap_or_default() .to_string_lossy() .to_string() .into() } } pub fn stash_to_proto(entry: &StashEntry) -> proto::StashEntry { proto::StashEntry { oid: entry.oid.as_bytes().to_vec(), message: entry.message.clone(), branch: entry.branch.clone(), index: entry.index as u64, timestamp: entry.timestamp, } } pub fn proto_to_stash(entry: &proto::StashEntry) -> Result { Ok(StashEntry { oid: Oid::from_bytes(&entry.oid)?, message: entry.message.clone(), index: entry.index as usize, branch: entry.branch.clone(), timestamp: entry.timestamp, }) } impl MergeDetails { async fn update( &mut self, backend: &Arc, current_conflicted_paths: Vec, ) -> bool { log::debug!("load merge details"); self.message = backend.merge_message().await.map(SharedString::from); let heads = backend .revparse_batch(vec![ "MERGE_HEAD".into(), "CHERRY_PICK_HEAD".into(), "REBASE_HEAD".into(), "REVERT_HEAD".into(), "APPLY_HEAD".into(), ]) .await .log_err() .unwrap_or_default() .into_iter() .map(|opt| opt.map(SharedString::from)) .collect::>(); let mut conflicts_changed = false; // Record the merge state for newly conflicted paths for path in ¤t_conflicted_paths { if self.merge_heads_by_conflicted_path.get(&path).is_none() { conflicts_changed = true; self.merge_heads_by_conflicted_path .insert(path.clone(), heads.clone()); } } // Clear state for paths that are no longer conflicted and for which the merge heads have changed self.merge_heads_by_conflicted_path .retain(|path, old_merge_heads| { let keep = current_conflicted_paths.contains(path) || (old_merge_heads == &heads && old_merge_heads.iter().any(|head| head.is_some())); if !keep { conflicts_changed = true; } keep }); conflicts_changed } } impl Repository { pub fn is_trusted(&self) -> bool { match self.repository_state.peek() { Some(Ok(RepositoryState::Local(state))) => state.backend.is_trusted(), _ => false, } } pub fn snapshot(&self) -> RepositorySnapshot { self.snapshot.clone() } pub fn pending_ops(&self) -> impl Iterator + '_ { self.pending_ops.iter().cloned() } pub fn pending_ops_summary(&self) -> PathSummary { self.pending_ops.summary().clone() } pub fn pending_ops_for_path(&self, path: &RepoPath) -> Option { self.pending_ops .get(&PathKey(path.as_ref().clone()), ()) .cloned() } fn respawn_local_worker( &mut self, project_environment: WeakEntity, fs: Arc, is_trusted: bool, cx: &mut Context, ) { let work_directory_abs_path = self.snapshot.work_directory_abs_path.clone(); let dot_git_abs_path = self.snapshot.dot_git_abs_path.clone(); let state = cx .spawn(async move |_, cx| { LocalRepositoryState::new( work_directory_abs_path, dot_git_abs_path, project_environment, fs, is_trusted, cx, ) .await .map_err(|err| err.to_string()) }) .shared(); self.job_sender.close_channel(); self._worker_task = Task::ready(()); self.active_jobs.clear(); self.job_debug_queue .mark_unfinished_complete(job_debug_queue::CompletedJobStatus::Skipped); cx.notify(); let (job_sender, worker_task) = Repository::spawn_local_git_worker(state.clone(), cx); self.job_sender = job_sender; self._worker_task = worker_task; self.repository_state = cx .spawn(async move |_, _| { let state = state.await?; Ok(RepositoryState::Local(state)) }) .shared(); } fn reinitialize_local_backend( &mut self, work_directory_abs_path: Arc, dot_git_abs_path: Arc, repository_dir_abs_path: Arc, common_dir_abs_path: Arc, project_environment: WeakEntity, fs: Arc, is_trusted: bool, cx: &mut Context, ) { self.snapshot.work_directory_abs_path = work_directory_abs_path; self.snapshot.dot_git_abs_path = dot_git_abs_path; self.snapshot.repository_dir_abs_path = repository_dir_abs_path; self.snapshot.common_dir_abs_path = common_dir_abs_path; self.respawn_local_worker(project_environment, fs, is_trusted, cx); } fn local( id: RepositoryId, work_directory_abs_path: Arc, repository_dir_abs_path: Arc, common_dir_abs_path: Arc, dot_git_abs_path: Arc, project_environment: WeakEntity, fs: Arc, is_trusted: bool, git_store: WeakEntity, cx: &mut Context, ) -> Self { let snapshot = RepositorySnapshot::empty( id, work_directory_abs_path, Some(repository_dir_abs_path), Some(dot_git_abs_path), Some(common_dir_abs_path), PathStyle::local(), ); let mut repo = Repository { this: cx.weak_entity(), git_store, snapshot, pending_ops: Default::default(), repository_state: Task::ready(Err("not yet initialized".into())).shared(), _worker_task: Task::ready(()), commit_message_buffer: None, askpass_delegates: Default::default(), paths_needing_status_update: Default::default(), latest_askpass_id: 0, job_sender: mpsc::unbounded().0, job_id: 0, active_jobs: Default::default(), job_debug_queue: job_debug_queue::GitJobDebugQueue::new(), initial_graph_data: Default::default(), commit_data: Default::default(), commit_data_handler: CommitDataHandlerState::Closed, }; repo.respawn_local_worker(project_environment, fs, is_trusted, cx); cx.subscribe_self(Self::handle_subscribe_self).detach(); repo } fn remote( id: RepositoryId, work_directory_abs_path: Arc, repository_dir_abs_path: Option>, common_dir_abs_path: Option>, path_style: PathStyle, project_id: ProjectId, client: AnyProtoClient, git_store: WeakEntity, cx: &mut Context, ) -> Self { let snapshot = RepositorySnapshot::empty( id, work_directory_abs_path, repository_dir_abs_path, None, common_dir_abs_path, path_style, ); let repository_state = RemoteRepositoryState { project_id, client }; let (job_sender, worker_task) = Self::spawn_remote_git_worker(repository_state.clone(), cx); let repository_state = Task::ready(Ok(RepositoryState::Remote(repository_state))).shared(); cx.subscribe_self(Self::handle_subscribe_self).detach(); Self { this: cx.weak_entity(), snapshot, commit_message_buffer: None, git_store, pending_ops: Default::default(), paths_needing_status_update: Default::default(), job_sender, _worker_task: worker_task, repository_state, askpass_delegates: Default::default(), latest_askpass_id: 0, active_jobs: Default::default(), job_debug_queue: job_debug_queue::GitJobDebugQueue::new(), job_id: 0, initial_graph_data: Default::default(), commit_data: Default::default(), commit_data_handler: CommitDataHandlerState::Closed, } } fn handle_subscribe_self(&mut self, event: &RepositoryEvent, _: &mut Context) { // scan id greater than 2 means the initial snapshot was calculated, // otherwise we don't need to refresh the graph state match event { RepositoryEvent::HeadChanged | RepositoryEvent::BranchListChanged => { if self.scan_id > 2 { self.initial_graph_data.clear(); } } RepositoryEvent::StashEntriesChanged => { if self.scan_id > 2 { self.initial_graph_data .retain(|(log_source, _), _| *log_source != LogSource::All); } } _ => {} } } pub fn git_store(&self) -> Option> { self.git_store.upgrade() } fn reload_buffer_diff_bases(&mut self, cx: &mut Context) { let this = cx.weak_entity(); let git_store = self.git_store.clone(); let _ = self.send_keyed_job( "reload_buffer_diff_bases", Some(GitJobKey::ReloadBufferDiffBases), None, |state, mut cx| async move { let RepositoryState::Local(LocalRepositoryState { backend, .. }) = state else { log::error!("tried to recompute diffs for a non-local repository"); return Ok(()); }; let Some(this) = this.upgrade() else { return Ok(()); }; let repo_diff_state_updates = this.update(&mut cx, |this, cx| { git_store.update(cx, |git_store, cx| { git_store .diffs .iter() .filter_map(|(buffer_id, diff_state)| { let buffer_store = git_store.buffer_store.read(cx); let buffer = buffer_store.get(*buffer_id)?; let file = File::from_dyn(buffer.read(cx).file())?; let abs_path = file.worktree.read(cx).absolutize(&file.path); let repo_path = this.abs_path_to_repo_path(&abs_path)?; let is_symlink = GitStore::file_is_symlink(file, cx); log::debug!( "start reload diff bases for repo path {}", repo_path.as_unix_str() ); diff_state.update(cx, |diff_state, _| { let has_unstaged_diff = diff_state .unstaged_diff .as_ref() .is_some_and(|diff| diff.is_upgradable()); let has_staged_diff = diff_state .staged_diff .as_ref() .is_some_and(|(diff, _)| diff.is_upgradable()); let has_uncommitted_diff = diff_state .uncommitted_diff .as_ref() .is_some_and(|set| set.is_upgradable()); Some(( buffer, repo_path, is_symlink, (has_unstaged_diff || has_staged_diff) .then(|| diff_state.index_text.clone()), (has_staged_diff || has_uncommitted_diff) .then(|| diff_state.head_text.clone()), )) }) }) .collect::>() }) })?; let buffer_diff_base_changes = cx .background_spawn(async move { let mut revisions = Vec::new(); for (_, repo_path, is_symlink, current_index_text, current_head_text) in &repo_diff_state_updates { if current_index_text.is_some() && !*is_symlink { revisions.push(format!(":{}", repo_path.as_unix_str())); } if current_head_text.is_some() && !*is_symlink { revisions.push(format!("HEAD:{}", repo_path.as_unix_str())); } } let mut loaded_revisions = backend .load_revisions(revisions) .await .log_err() .into_iter() .flatten(); let mut changes = Vec::new(); for (buffer, _, is_symlink, current_index_text, current_head_text) in &repo_diff_state_updates { let index_text = (current_index_text.is_some() && !*is_symlink) .then(|| loaded_revisions.next().flatten()) .flatten(); let head_text = (current_head_text.is_some() && !*is_symlink) .then(|| loaded_revisions.next().flatten()) .flatten(); let change = match (current_index_text.as_ref(), current_head_text.as_ref()) { (Some(current_index), Some(current_head)) => { let index_changed = index_text.as_deref() != current_index.as_deref(); let head_changed = head_text.as_deref() != current_head.as_deref(); if index_changed && head_changed { if index_text == head_text { Some(DiffBasesChange::SetBoth(head_text)) } else { Some(DiffBasesChange::SetEach { index: index_text, head: head_text, }) } } else if index_changed { Some(DiffBasesChange::SetIndex(index_text)) } else if head_changed { Some(DiffBasesChange::SetHead(head_text)) } else { None } } (Some(current_index), None) => { let index_changed = index_text.as_deref() != current_index.as_deref(); index_changed .then_some(DiffBasesChange::SetIndex(index_text)) } (None, Some(current_head)) => { let head_changed = head_text.as_deref() != current_head.as_deref(); head_changed.then_some(DiffBasesChange::SetHead(head_text)) } (None, None) => None, }; changes.push((buffer.clone(), change)) } changes }) .await; git_store.update(&mut cx, |git_store, cx| { for (buffer, diff_bases_change) in buffer_diff_base_changes { let buffer_snapshot = buffer.read(cx).text_snapshot(); let buffer_id = buffer_snapshot.remote_id(); let Some(diff_state) = git_store.diffs.get(&buffer_id) else { continue; }; let downstream_client = git_store.downstream_client(); diff_state.update(cx, |diff_state, cx| { use proto::update_diff_bases::Mode; if let Some((client, project_id)) = downstream_client { let (staged_text, committed_text, mode) = match diff_bases_change.clone() { Some(DiffBasesChange::SetIndex(index)) => { (index, None, Mode::IndexOnly) } Some(DiffBasesChange::SetHead(head)) => { (None, head, Mode::HeadOnly) } Some(DiffBasesChange::SetEach { index, head }) => { (index, head, Mode::IndexAndHead) } Some(DiffBasesChange::SetBoth(text)) => { (None, text, Mode::IndexMatchesHead) } None => (None, None, Mode::Unchanged), }; client .send(proto::UpdateDiffBases { project_id: project_id.to_proto(), buffer_id: buffer_id.to_proto(), staged_text, committed_text, mode: mode as i32, }) .log_err(); } diff_state.diff_bases_changed(buffer_snapshot, diff_bases_change, cx); }); } }) }, ); } pub fn send_job( &mut self, description: &'static str, status: Option, job: F, ) -> oneshot::Receiver where F: FnOnce(RepositoryState, AsyncApp) -> Fut + 'static, Fut: Future + 'static, R: Send + 'static, { self.send_keyed_job(description, None, status, job) } fn send_keyed_job( &mut self, description: &'static str, key: Option, status: Option, job: F, ) -> oneshot::Receiver where F: FnOnce(RepositoryState, AsyncApp) -> Fut + 'static, Fut: Future + 'static, R: Send + 'static, { let (result_tx, result_rx) = futures::channel::oneshot::channel(); let job_id = post_inc(&mut self.job_id); let this = self.this.clone(); let key_label = key.as_ref().map(format_job_key); self.job_debug_queue.add(job_id, description, key_label); self.job_sender .unbounded_send(GitJob { id: job_id, key, job: Box::new(move |state, cx: &mut AsyncApp| { let job = job(state, cx.clone()); cx.spawn(async move |cx| { this.update(cx, |this, cx| { this.job_debug_queue.mark_running(job_id); if let Some(s) = status { this.active_jobs.insert( job_id, JobInfo { start: Instant::now(), message: s, }, ); } cx.notify(); }) .ok(); let result = job.await; this.update(cx, |this, cx| { this.job_debug_queue.mark_complete( job_id, job_debug_queue::CompletedJobStatus::Finished, ); this.active_jobs.remove(&job_id); cx.notify(); }) .ok(); result_tx.send(result).ok(); }) }), }) .ok(); result_rx } pub fn set_as_active_repository(&self, cx: &mut Context) { let Some(git_store) = self.git_store.upgrade() else { return; }; let entity = cx.entity(); git_store.update(cx, |git_store, cx| { let Some((&id, _)) = git_store .repositories .iter() .find(|(_, handle)| *handle == &entity) else { return; }; git_store.active_repo_id = Some(id); cx.emit(GitStoreEvent::ActiveRepositoryChanged(Some(id))); }); } pub fn cached_status(&self) -> impl '_ + Iterator { self.snapshot.status() } pub fn diff_stat_for_path(&self, path: &RepoPath) -> Option { self.snapshot.diff_stat_for_path(path) } pub fn cached_stash(&self) -> GitStash { self.snapshot.stash_entries.clone() } pub fn repo_path_to_project_path(&self, path: &RepoPath, cx: &App) -> Option { let git_store = self.git_store.upgrade()?; let worktree_store = git_store.read(cx).worktree_store.read(cx); let abs_path = self.snapshot.repo_path_to_abs_path(path); let abs_path = SanitizedPath::new(&abs_path); let (worktree, relative_path) = worktree_store.find_worktree(abs_path, cx)?; Some(ProjectPath { worktree_id: worktree.read(cx).id(), path: relative_path, }) } pub fn project_path_to_repo_path(&self, path: &ProjectPath, cx: &App) -> Option { let git_store = self.git_store.upgrade()?; let worktree_store = git_store.read(cx).worktree_store.read(cx); let abs_path = worktree_store.absolutize(path, cx)?; self.snapshot.abs_path_to_repo_path(&abs_path) } pub fn contains_sub_repo(&self, other: &Entity, cx: &App) -> bool { other .read(cx) .snapshot .work_directory_abs_path .starts_with(&self.snapshot.work_directory_abs_path) } pub fn commit_message_buffer(&self) -> Option<&Entity> { self.commit_message_buffer.as_ref() } pub fn open_commit_buffer( &mut self, languages: Option>, buffer_store: Entity, cx: &mut Context, ) -> Task>> { let id = self.id; if let Some(buffer) = self.commit_message_buffer.clone() { return Task::ready(Ok(buffer)); } let this = cx.weak_entity(); let rx = self.send_job( "open_commit_buffer", None, move |state, mut cx| async move { let Some(this) = this.upgrade() else { bail!("git store was dropped"); }; match state { RepositoryState::Local(..) => { this.update(&mut cx, |_, cx| { Self::open_local_commit_buffer(languages, buffer_store, cx) }) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let request = client.request(proto::OpenCommitMessageBuffer { project_id: project_id.0, repository_id: id.to_proto(), }); let response = request.await.context("requesting to open commit buffer")?; let buffer_id = BufferId::new(response.buffer_id)?; let buffer = buffer_store .update(&mut cx, |buffer_store, cx| { buffer_store.wait_for_remote_buffer(buffer_id, cx) }) .await?; if let Some(language_registry) = languages { let git_commit_language = language_registry.language_for_name("Git Commit").await?; buffer.update(&mut cx, |buffer, cx| { buffer.set_language(Some(git_commit_language), cx); }); } this.update(&mut cx, |this, _| { this.commit_message_buffer = Some(buffer.clone()); }); Ok(buffer) } } }, ); cx.spawn(|_, _: &mut AsyncApp| async move { rx.await? }) } fn open_local_commit_buffer( language_registry: Option>, buffer_store: Entity, cx: &mut Context, ) -> Task>> { cx.spawn(async move |repository, cx| { let git_commit_language = match language_registry { Some(language_registry) => { Some(language_registry.language_for_name("Git Commit").await?) } None => None, }; let buffer = buffer_store .update(cx, |buffer_store, cx| { buffer_store.create_buffer(git_commit_language, false, cx) }) .await?; repository.update(cx, |repository, _| { repository.commit_message_buffer = Some(buffer.clone()); })?; Ok(buffer) }) } pub fn checkout_files( &mut self, commit: &str, paths: Vec, cx: &mut Context, ) -> Task> { let commit = commit.to_string(); let id = self.id; self.spawn_job_with_tracking( paths.clone(), pending_op::GitStatus::Reverted, cx, async move |this, cx| { this.update(cx, |this, _cx| { this.send_job( "checkout_files", Some(format!("git checkout {}", commit).into()), move |git_repo, _| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => { backend .checkout_files(commit, paths, environment.clone()) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client, }) => { client .request(proto::GitCheckoutFiles { project_id: project_id.0, repository_id: id.to_proto(), commit, paths: paths .into_iter() .map(|p| p.as_unix_str().to_owned()) .collect(), }) .await?; Ok(()) } } }, ) })? .await? }, ) } pub fn reset( &mut self, commit: String, reset_mode: ResetMode, _cx: &mut App, ) -> oneshot::Receiver> { let id = self.id; self.send_job("reset", None, move |git_repo, _| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => backend.reset(commit, reset_mode, environment).await, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitReset { project_id: project_id.0, repository_id: id.to_proto(), commit, mode: match reset_mode { ResetMode::Soft => git_reset::ResetMode::Soft.into(), ResetMode::Mixed => git_reset::ResetMode::Mixed.into(), }, }) .await?; Ok(()) } } }) } pub fn show(&mut self, commit: String) -> oneshot::Receiver> { let id = self.id; self.send_job("show", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.show(commit).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let resp = client .request(proto::GitShow { project_id: project_id.0, repository_id: id.to_proto(), commit, }) .await?; Ok(CommitDetails { sha: resp.sha.into(), message: resp.message.into(), commit_timestamp: resp.commit_timestamp, author_email: resp.author_email.into(), author_name: resp.author_name.into(), }) } } }) } pub fn load_commit_diff(&mut self, commit: String) -> oneshot::Receiver> { let id = self.id; self.send_job("load_commit_diff", None, move |git_repo, cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.load_commit(commit, cx).await } RepositoryState::Remote(RemoteRepositoryState { client, project_id, .. }) => { let response = client .request(proto::LoadCommitDiff { project_id: project_id.0, repository_id: id.to_proto(), commit, }) .await?; Ok(CommitDiff { files: response .files .into_iter() .map(|file| { Ok(CommitFile { path: RepoPath::from_proto(&file.path)?, old_text: file.old_text, new_text: file.new_text, is_binary: file.is_binary, }) }) .collect::>>()?, }) } } }) } pub fn file_history_changed_files( &mut self, paths: Vec, commit_limit: usize, ) -> oneshot::Receiver>> { self.send_job( "file_history_changed_files", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend .file_history_changed_files(paths, commit_limit) .await } RepositoryState::Remote(_) => { anyhow::bail!("file history changed files is only supported locally") } } }, ) } pub fn get_graph_data( &self, log_source: LogSource, log_order: LogOrder, ) -> Option<&InitialGitGraphData> { self.initial_graph_data.get(&(log_source, log_order)) } pub fn search_commits( &mut self, log_source: LogSource, search_args: SearchCommitArgs, request_tx: async_channel::Sender, cx: &mut Context, ) { let repository_state = self.repository_state.clone(); let repository_id = self.id; cx.background_spawn(async move { let repo_state = repository_state.await; match repo_state { Ok(RepositoryState::Local(LocalRepositoryState { backend, .. })) => { backend .search_commits(log_source, search_args, request_tx) .await .log_err(); } Ok(RepositoryState::Remote(RemoteRepositoryState { client, project_id })) => { let result = client .request_stream(proto::SearchCommits { project_id: project_id.to_proto(), repository_id: repository_id.to_proto(), log_source: Some(log_source_to_proto(&log_source)), query: search_args.query.to_string(), case_sensitive: search_args.case_sensitive, }) .await; let mut stream = match result { Ok(stream) => stream, Err(error) => { log::error!("failed to search commits remotely: {error:?}"); return; } }; while let Some(response) = stream.next().await { let response = match response { Ok(response) => response, Err(error) => { log::error!( "failed to receive remote commit search results: {error:?}" ); return; } }; for sha in &response.shas { let Ok(oid) = Oid::from_str(sha) else { return; }; if request_tx.send(oid).await.is_err() { return; } } } } Err(error) => { log::error!("failed to get repository state for commit search: {error}"); } }; }) .detach(); } pub fn graph_data( &mut self, log_source: LogSource, log_order: LogOrder, range: Range, cx: &mut Context, ) -> GraphDataResponse<'_> { let initial_commit_data = self .initial_graph_data .entry((log_source.clone(), log_order)) .or_insert_with(|| { let state = self.repository_state.clone(); let log_source = log_source.clone(); let fetch_task = cx.spawn(async move |repository, cx| { let state = state.await; let result = match state { Ok(RepositoryState::Local(LocalRepositoryState { backend, .. })) => { Self::local_git_graph_data( repository.clone(), backend, log_source.clone(), log_order, cx, ) .await } Ok(RepositoryState::Remote(remote)) => { Self::remote_git_graph_data( repository.clone(), remote, log_source.clone(), log_order, cx, ) .await } Err(e) => Err(SharedString::from(e)), }; repository .update(cx, |repository, cx| { if let Some(data) = repository .initial_graph_data .get_mut(&(log_source.clone(), log_order)) { match &result { Ok(()) => { cx.emit(RepositoryEvent::GraphEvent( (log_source.clone(), log_order), GitGraphEvent::FullyLoaded, )); } Err(fetch_task_error) => { data.subscribers.retain(|sender| { sender.try_send(Err(fetch_task_error.clone())).is_ok() }); data.error = Some(fetch_task_error.clone()); cx.emit(RepositoryEvent::GraphEvent( (log_source.clone(), log_order), GitGraphEvent::LoadingError, )); } } data.subscribers.clear(); } else { debug_panic!( "This task would be dropped if this entry doesn't exist" ); } }) .log_err(); }); InitialGitGraphData { fetch_task, error: None, commit_data: Vec::new(), commit_oid_to_index: HashMap::default(), subscribers: Vec::new(), } }); let max_start = initial_commit_data.commit_data.len().saturating_sub(1); let max_end = initial_commit_data.commit_data.len(); GraphDataResponse { commits: &initial_commit_data.commit_data [range.start.min(max_start)..range.end.min(max_end)], is_loading: !initial_commit_data.fetch_task.is_ready(), error: initial_commit_data.error.clone(), } } async fn append_initial_graph_commits( this: &WeakEntity, graph_data_key: &(LogSource, LogOrder), initial_graph_commit_data: Vec>, cx: &mut AsyncApp, ) { this.update(cx, |repository, cx| { let graph_data = repository .initial_graph_data .entry(graph_data_key.clone()) .and_modify(|graph_data| { if !graph_data.subscribers.is_empty() { graph_data.subscribers.retain(|sender| { sender .try_send(Ok(initial_graph_commit_data.clone())) .is_ok() }); } for commit_data in initial_graph_commit_data { graph_data .commit_oid_to_index .insert(commit_data.sha, graph_data.commit_data.len()); graph_data.commit_data.push(commit_data); } cx.emit(RepositoryEvent::GraphEvent( graph_data_key.clone(), GitGraphEvent::CountUpdated(graph_data.commit_data.len()), )); }); match &graph_data { Entry::Occupied(_) => {} Entry::Vacant(_) => { debug_panic!("This task should be dropped if data doesn't exist"); } } }) .log_err(); } async fn local_git_graph_data( this: WeakEntity, backend: Arc, log_source: LogSource, log_order: LogOrder, cx: &mut AsyncApp, ) -> Result<(), SharedString> { let (request_tx, request_rx) = async_channel::unbounded::>>(); let task = cx.background_executor().spawn({ let log_source = log_source.clone(); async move { backend .initial_graph_data(log_source, log_order, request_tx) .await .map_err(|err| SharedString::from(err.to_string())) } }); let graph_data_key = (log_source, log_order); while let Ok(initial_graph_commit_data) = request_rx.recv().await { Self::append_initial_graph_commits( &this, &graph_data_key, initial_graph_commit_data, cx, ) .await; } task.await?; Ok(()) } async fn remote_git_graph_data( this: WeakEntity, remote: RemoteRepositoryState, log_source: LogSource, log_order: LogOrder, cx: &mut AsyncApp, ) -> Result<(), SharedString> { let repository_id = this .update(cx, |repository, _| repository.id) .map_err(|err| SharedString::from(err.to_string()))?; let graph_data_key = (log_source.clone(), log_order); let mut response = remote .client .request_stream(proto::GetInitialGraphData { project_id: remote.project_id.to_proto(), repository_id: repository_id.to_proto(), log_source: Some(log_source_to_proto(&log_source)), log_order: log_order_to_proto(log_order), }) .await .map_err(|err| SharedString::from(err.to_string()))?; while let Some(response) = response.next().await { let response = response.map_err(|err| SharedString::from(err.to_string()))?; let commits = response .commits .into_iter() .map(initial_graph_commit_from_proto) .collect::>>() .map_err(|err| SharedString::from(err.to_string()))?; Self::append_initial_graph_commits(&this, &graph_data_key, commits, cx).await; } Ok(()) } pub fn fetch_commit_data( &mut self, sha: Oid, await_result: bool, cx: &mut Context, ) -> &CommitDataState { if self.commit_data.contains_key(&sha) { let data = &self.commit_data[&sha]; if let CommitDataState::Loading(None) = data && await_result { let (tx, rx) = oneshot::channel(); self.commit_data .insert(sha, CommitDataState::Loading(Some(rx.shared()))); let handler = self.get_handler(cx); handler.completion_senders.insert(sha, tx); } return &self.commit_data[&sha]; } let (state, completer) = if await_result { let (tx, rx) = oneshot::channel(); (CommitDataState::Loading(Some(rx.shared())), Some(tx)) } else { (CommitDataState::Loading(None), None) }; self.commit_data.insert(sha, state); let handler = self.get_handler(cx); if let Some(tx) = completer { handler.completion_senders.insert(sha, tx); } let mut has_failed = false; if handler.commit_data_request.try_send(sha).is_ok() { handler.pending_requests.insert(sha); } else { has_failed = true; handler.completion_senders.remove(&sha); debug_assert!( matches!( self.commit_data.remove(&sha), Some(CommitDataState::Loading(_)) ), "Commit data should still be loading when enqueueing the request fails" ); } &self.commit_data.get(&sha).unwrap_or_else(|| { debug_assert!(!has_failed, "This should always be inserted"); &CommitDataState::Loading(None) }) } fn get_handler(&mut self, cx: &mut Context) -> &mut CommitDataHandler { if matches!(self.commit_data_handler, CommitDataHandlerState::Closed) { self.commit_data_handler = CommitDataHandlerState::Open(self.open_commit_data_handler(cx)); } match &mut self.commit_data_handler { CommitDataHandlerState::Open(handler) => handler, CommitDataHandlerState::Closed => unreachable!(), } } fn open_commit_data_handler(&self, cx: &Context) -> CommitDataHandler { let state = self.repository_state.clone(); let (result_tx, result_rx) = async_channel::bounded::<(Oid, CommitData)>(64); let (request_tx, request_rx) = async_channel::unbounded::(); let foreground_task = cx.spawn(async move |this, cx| { while let Ok((sha, commit_data)) = result_rx.recv().await { let result = this.update(cx, |this, cx| { let data = Arc::new(commit_data); if let CommitDataHandlerState::Open(handler) = &mut this.commit_data_handler { handler.pending_requests.remove(&sha); if let Some(completion_sender) = handler.completion_senders.remove(&sha) { completion_sender.send(data.clone()).ok(); } } else { debug_panic!("The handler state has to be open for this task to exist"); } let old_value = this.commit_data.insert(sha, CommitDataState::Loaded(data)); debug_assert!( !matches!(old_value, Some(CommitDataState::Loaded(_))), "We should never overwrite commit data" ); cx.notify(); }); if result.is_err() { break; } } this.update(cx, |this, _cx| { let CommitDataHandlerState::Open(handler) = std::mem::replace( &mut this.commit_data_handler, CommitDataHandlerState::Closed, ) else { debug_panic!("The handler state has to be open for this task to exist"); return; }; for sha in handler.pending_requests { this.commit_data.remove(&sha); } }) .ok(); }); let request_tx_for_handler = request_tx; let repository_id = self.id; let background_executor = cx.background_executor().clone(); cx.background_spawn(async move { match state.await { Ok(RepositoryState::Local(LocalRepositoryState { backend, .. })) => { Self::local_commit_data_reader( backend, request_rx, result_tx, background_executor, ) .await; } Ok(RepositoryState::Remote(RemoteRepositoryState { project_id, client })) => { Self::remote_commit_data_reader( project_id, client, repository_id, request_rx, result_tx, background_executor, ) .await; } Err(error) => { log::error!("failed to get repository state: {error}"); return; } }; }) .detach(); CommitDataHandler { _task: foreground_task, commit_data_request: request_tx_for_handler, completion_senders: HashMap::default(), pending_requests: HashSet::default(), } } async fn local_commit_data_reader( backend: Arc, request_rx: smol::channel::Receiver, result_tx: smol::channel::Sender<(Oid, CommitData)>, background_executor: BackgroundExecutor, ) { async fn receive_commit_data_request( request_rx: &smol::channel::Receiver, ) -> Option { if request_rx.is_closed() && request_rx.is_empty() { future::pending().await } else { request_rx.recv().await.ok() } } let reader = match backend.commit_data_reader() { Ok(reader) => reader, Err(error) => { log::error!("failed to create commit data reader: {error:?}"); return; } }; let read_commit_data = |sha| reader.read(sha).map(move |result| (sha, result)); let mut read_futures = FuturesUnordered::new(); loop { if read_futures.is_empty() { let timeout = background_executor.timer(Duration::from_secs(10)); futures::select_biased! { sha = futures::FutureExt::fuse(receive_commit_data_request(&request_rx)) => { if let Some(sha) = sha { read_futures.push(read_commit_data(sha)); } } _ = futures::FutureExt::fuse(timeout) => { break; } } } let next_read = read_futures.next().fuse(); futures::pin_mut!(next_read); futures::select_biased! { result = next_read => { let Some((sha, result)) = result else { continue; }; match result { Ok(commit_data) => { if result_tx.send((sha, commit_data)).await.is_err() { return; } } Err(error) => { log::error!("failed to read commit data for {sha}: {error:?}"); } } } sha = futures::FutureExt::fuse(receive_commit_data_request(&request_rx)) => { if let Some(sha) = sha { read_futures.push(read_commit_data(sha)); } } } } drop(result_tx); } async fn remote_commit_data_reader( project_id: ProjectId, client: AnyProtoClient, repository_id: RepositoryId, request_rx: smol::channel::Receiver, result_tx: smol::channel::Sender<(Oid, CommitData)>, background_executor: BackgroundExecutor, ) { let mut response_futures = FuturesUnordered::>>::new(); let mut accept_requests = true; let mut next_request = Self::get_next_request( project_id, client.clone(), repository_id, &request_rx, &background_executor, ) .boxed() .fuse(); loop { if !accept_requests && response_futures.is_empty() { break; } if response_futures.is_empty() { match (&mut next_request).await { NextCommitDataRequest::Request(request) => { response_futures.push(request); next_request = Self::get_next_request( project_id, client.clone(), repository_id, &request_rx, &background_executor, ) .boxed() .fuse(); } NextCommitDataRequest::Closed | NextCommitDataRequest::Idle => break, } } let next_response = response_futures.next().fuse(); futures::pin_mut!(next_response); futures::select_biased! { request = next_request => { match request { NextCommitDataRequest::Request(request) => { response_futures.push(request); } NextCommitDataRequest::Idle => {} NextCommitDataRequest::Closed => { accept_requests = false; } } if accept_requests { next_request = Self::get_next_request( project_id, client.clone(), repository_id, &request_rx, &background_executor, ) .boxed() .fuse(); } } result = next_response => { let Some(result) = result else { continue; }; if let Ok(commit_data) = result { for commit in commit_data.commits { let Ok(commit_data) = commit_data_from_proto(commit) else { continue; }; if result_tx .send((commit_data.sha, commit_data)) .await .is_err() { return; } } } } } } drop(result_tx); } async fn get_next_request( project_id: ProjectId, client: AnyProtoClient, repository_id: RepositoryId, request_rx: &smol::channel::Receiver, background_executor: &BackgroundExecutor, ) -> NextCommitDataRequest { let mut queued_shas = Vec::with_capacity(64); loop { if queued_shas.len() >= 64 { break; } let timeout = background_executor.timer(Duration::from_millis(5)); futures::select_biased! { sha = futures::FutureExt::fuse(request_rx.recv()) => { let Ok(sha) = sha else { break; }; queued_shas.push(sha); } _ = futures::FutureExt::fuse(timeout) => { break; } } } if queued_shas.is_empty() && request_rx.is_closed() { NextCommitDataRequest::Closed } else if queued_shas.is_empty() { NextCommitDataRequest::Idle } else { NextCommitDataRequest::Request( client .request(proto::GetCommitData { project_id: project_id.to_proto(), repository_id: repository_id.to_proto(), shas: queued_shas.into_iter().map(|oid| oid.to_string()).collect(), }) .boxed(), ) } } fn buffer_store(&self, cx: &App) -> Option> { Some(self.git_store.upgrade()?.read(cx).buffer_store.clone()) } fn save_buffers<'a>( &self, entries: impl IntoIterator, cx: &mut Context, ) -> Vec>> { let mut save_futures = Vec::new(); if let Some(buffer_store) = self.buffer_store(cx) { buffer_store.update(cx, |buffer_store, cx| { for path in entries { let Some(project_path) = self.repo_path_to_project_path(path, cx) else { continue; }; if let Some(buffer) = buffer_store.get_by_path(&project_path) && buffer .read(cx) .file() .is_some_and(|file| file.disk_state().exists()) && buffer.read(cx).has_unsaved_edits() { save_futures.push(buffer_store.save_buffer(buffer, cx)); } } }) } save_futures } pub fn stage_entries( &mut self, entries: Vec, cx: &mut Context, ) -> Task> { self.stage_or_unstage_entries(true, entries, cx) } pub fn unstage_entries( &mut self, entries: Vec, cx: &mut Context, ) -> Task> { self.stage_or_unstage_entries(false, entries, cx) } fn stage_or_unstage_entries( &mut self, stage: bool, entries: Vec, cx: &mut Context, ) -> Task> { if entries.is_empty() { return Task::ready(Ok(())); } let Some(git_store) = self.git_store.upgrade() else { return Task::ready(Ok(())); }; let id = self.id; let save_tasks = self.save_buffers(&entries, cx); let paths = entries .iter() .map(|p| p.as_unix_str()) .collect::>() .join(" "); let status = if stage { format!("git add {paths}") } else { format!("git reset {paths}") }; let job_key = GitJobKey::WriteIndex(entries.clone()); self.spawn_job_with_tracking( entries.clone(), if stage { pending_op::GitStatus::Staged } else { pending_op::GitStatus::Unstaged }, cx, async move |this, cx| { for save_task in save_tasks { save_task.await?; } this.update(cx, |this, cx| { let weak_this = cx.weak_entity(); this.send_keyed_job( "stage_or_unstage_entries", Some(job_key), Some(status.into()), move |git_repo, mut cx| async move { let hunk_staging_operation_counts = weak_this .update(&mut cx, |this, cx| { let mut hunk_staging_operation_counts = HashMap::default(); for path in &entries { let Some(project_path) = this.repo_path_to_project_path(path, cx) else { continue; }; let Some(buffer) = git_store .read(cx) .buffer_store .read(cx) .get_by_path(&project_path) else { continue; }; let Some(diff_state) = git_store .read(cx) .diffs .get(&buffer.read(cx).remote_id()) .cloned() else { continue; }; let buffer_snapshot = buffer.read(cx).text_snapshot(); let hunk_staging_operation_count = diff_state.update(cx, |diff_state, cx| { diff_state .mark_whole_file_stage_or_unstage_pending( stage, &buffer_snapshot, cx, ); diff_state.hunk_staging_operation_count += 1; diff_state.hunk_staging_operation_count }); hunk_staging_operation_counts.insert( diff_state.downgrade(), hunk_staging_operation_count, ); } hunk_staging_operation_counts }) .unwrap_or_default(); let result = match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => { if stage { backend.stage_paths(entries, environment.clone()).await } else { backend.unstage_paths(entries, environment.clone()).await } } RepositoryState::Remote(RemoteRepositoryState { project_id, client, }) => { if stage { client .request(proto::Stage { project_id: project_id.0, repository_id: id.to_proto(), paths: entries .into_iter() .map(|repo_path| { repo_path.as_unix_str().to_owned() }) .collect(), }) .await .context("sending stage request") .map(|_| ()) } else { client .request(proto::Unstage { project_id: project_id.0, repository_id: id.to_proto(), paths: entries .into_iter() .map(|repo_path| { repo_path.as_unix_str().to_owned() }) .collect(), }) .await .context("sending unstage request") .map(|_| ()) } } }; for (diff_state, hunk_staging_operation_count) in hunk_staging_operation_counts { diff_state .update(&mut cx, |diff_state, cx| { if result.is_ok() { diff_state.hunk_staging_operation_count_as_of_write = hunk_staging_operation_count; } else { diff_state.clear_pending_hunks(cx); } }) .ok(); } result }, ) })? .await? }, ) } pub fn stage_all(&mut self, cx: &mut Context) -> Task> { let snapshot = self.snapshot.clone(); let pending_ops = self.pending_ops.clone(); let to_stage = cx.background_spawn(async move { snapshot .status() .filter_map(|entry| { if let Some(ops) = pending_ops .get(&PathKey(entry.repo_path.as_ref().clone()), ()) .filter(|ops| !ops.last_op_errored()) { if ops.staging() || ops.staged() { None } else { Some(entry.repo_path) } } else if entry.status.staging().is_fully_staged() { None } else { Some(entry.repo_path) } }) .collect() }); cx.spawn(async move |this, cx| { let to_stage = to_stage.await; this.update(cx, |this, cx| { this.stage_or_unstage_entries(true, to_stage, cx) })? .await }) } pub fn unstage_all(&mut self, cx: &mut Context) -> Task> { let snapshot = self.snapshot.clone(); let pending_ops = self.pending_ops.clone(); let to_unstage = cx.background_spawn(async move { snapshot .status() .filter_map(|entry| { if let Some(ops) = pending_ops .get(&PathKey(entry.repo_path.as_ref().clone()), ()) .filter(|ops| !ops.last_op_errored()) { if !ops.staging() && !ops.staged() { None } else { Some(entry.repo_path) } } else if entry.status.staging().is_fully_unstaged() { None } else { Some(entry.repo_path) } }) .collect() }); cx.spawn(async move |this, cx| { let to_unstage = to_unstage.await; this.update(cx, |this, cx| { this.stage_or_unstage_entries(false, to_unstage, cx) })? .await }) } pub fn stash_all(&mut self, cx: &mut Context) -> Task> { let to_stash = self.cached_status().map(|entry| entry.repo_path).collect(); self.stash_entries(to_stash, cx) } pub fn stash_entries( &mut self, entries: Vec, cx: &mut Context, ) -> Task> { let id = self.id; cx.spawn(async move |this, cx| { this.update(cx, |this, _| { this.send_job("stash_entries", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => backend.stash_paths(entries, environment).await, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::Stash { project_id: project_id.0, repository_id: id.to_proto(), paths: entries .into_iter() .map(|repo_path| repo_path.as_unix_str().to_owned()) .collect(), }) .await?; Ok(()) } } }) })? .await??; Ok(()) }) } pub fn stash_pop( &mut self, index: Option, cx: &mut Context, ) -> Task> { let id = self.id; cx.spawn(async move |this, cx| { this.update(cx, |this, _| { this.send_job("stash_pop", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => backend.stash_pop(index, environment).await, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::StashPop { project_id: project_id.0, repository_id: id.to_proto(), stash_index: index.map(|i| i as u64), }) .await .context("sending stash pop request")?; Ok(()) } } }) })? .await??; Ok(()) }) } pub fn stash_apply( &mut self, index: Option, cx: &mut Context, ) -> Task> { let id = self.id; cx.spawn(async move |this, cx| { this.update(cx, |this, _| { this.send_job("stash_apply", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => backend.stash_apply(index, environment).await, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::StashApply { project_id: project_id.0, repository_id: id.to_proto(), stash_index: index.map(|i| i as u64), }) .await .context("sending stash apply request")?; Ok(()) } } }) })? .await??; Ok(()) }) } pub fn add_path_to_gitignore( &mut self, repo_path: &RepoPath, is_dir: bool, ) -> oneshot::Receiver> { let id = self.id; let work_dir = self.snapshot.work_directory_abs_path.clone(); let path_display = repo_path.as_ref().display(PathStyle::Unix); let path = repo_path.as_unix_str().to_owned(); let file_path_str = if is_dir { format!("{}/", path_display) } else { path_display.to_string() }; self.send_job( "add_path_to_gitignore", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { fs, .. }) => { append_pattern_to_ignore_file( fs, work_dir.join(".gitignore"), file_path_str, ) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitAddPathToGitignore { project_id: project_id.0, repository_id: id.to_proto(), path, is_dir, }) .await .context("sending add path to .gitignore request")?; Ok(()) } } }, ) } pub fn add_path_to_git_info_exclude( &mut self, repo_path: &RepoPath, is_dir: bool, ) -> oneshot::Receiver> { let id = self.id; let repository_dir = self.snapshot.repository_dir_abs_path.clone(); let path_display = repo_path.as_ref().display(PathStyle::Unix); let path = repo_path.as_unix_str().to_owned(); let file_path_str = if is_dir { format!("{}/", path_display) } else { path_display.to_string() }; self.send_job( "add_path_to_git_info_exclude", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { fs, .. }) => { append_pattern_to_ignore_file( fs, repository_dir.join(git::REPO_EXCLUDE), file_path_str, ) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitAddPathToGitInfoExclude { project_id: project_id.0, repository_id: id.to_proto(), path, is_dir, }) .await .context("sending add path to .git/info/exclude request")?; Ok(()) } } }, ) } pub fn stash_drop( &mut self, index: Option, cx: &mut Context, ) -> oneshot::Receiver> { let id = self.id; let updates_tx = self .git_store() .and_then(|git_store| match &git_store.read(cx).state { GitStoreState::Local { downstream, .. } => downstream .as_ref() .map(|downstream| downstream.updates_tx.clone()), _ => None, }); let this = cx.weak_entity(); self.send_job("stash_drop", None, move |git_repo, mut cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => { // TODO would be nice to not have to do this manually let result = backend.stash_drop(index, environment).await; if result.is_ok() && let Ok(stash_entries) = backend.stash_entries().await { let snapshot = this.update(&mut cx, |this, cx| { this.snapshot.stash_entries = stash_entries; cx.emit(RepositoryEvent::StashEntriesChanged); this.snapshot.clone() })?; if let Some(updates_tx) = updates_tx { updates_tx .unbounded_send(DownstreamUpdate::UpdateRepository(snapshot)) .ok(); } } result } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::StashDrop { project_id: project_id.0, repository_id: id.to_proto(), stash_index: index.map(|i| i as u64), }) .await .context("sending stash pop request")?; Ok(()) } } }) } // Kept for wire compatibility: older remote clients run the pre-commit hook explicitly // via `proto::RunGitHook` before committing. New code lets `git commit` run hooks itself. // // TODO: remove together with `proto::RunGitHook` once all supported peers commit without // sending it (see the deprecation note on the message in git.proto). pub fn run_hook(&mut self, hook: RunHook, _cx: &mut App) -> oneshot::Receiver> { let id = self.id; self.send_job( "run_hook", Some(format!("git hook {}", hook.as_str()).into()), move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => backend.run_hook(hook, environment.clone()).await, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::RunGitHook { project_id: project_id.0, repository_id: id.to_proto(), hook: hook.to_proto(), }) .await?; Ok(()) } } }, ) } pub fn commit( &mut self, message: SharedString, name_and_email: Option<(SharedString, SharedString)>, options: CommitOptions, askpass: AskPassDelegate, _cx: &mut App, ) -> oneshot::Receiver> { let id = self.id; let askpass_delegates = self.askpass_delegates.clone(); let askpass_id = util::post_inc(&mut self.latest_askpass_id); self.send_job( "commit", Some("git commit".into()), move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => { backend .commit(message, name_and_email, options, askpass, environment) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { askpass_delegates.lock().insert(askpass_id, askpass); let _defer = util::defer(|| { let askpass_delegate = askpass_delegates.lock().remove(&askpass_id); debug_assert!(askpass_delegate.is_some()); }); let (name, email) = name_and_email.unzip(); client .request(proto::Commit { project_id: project_id.0, repository_id: id.to_proto(), message: String::from(message), name: name.map(String::from), email: email.map(String::from), options: Some(proto::commit::CommitOptions { amend: options.amend, signoff: options.signoff, allow_empty: options.allow_empty, }), askpass_id, }) .await?; Ok(()) } } }, ) } pub fn fetch( &mut self, fetch_options: FetchOptions, askpass: AskPassDelegate, _cx: &mut App, ) -> oneshot::Receiver> { let askpass_delegates = self.askpass_delegates.clone(); let askpass_id = util::post_inc(&mut self.latest_askpass_id); let id = self.id; self.send_job( "fetch", Some("git fetch".into()), move |git_repo, cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => backend.fetch(fetch_options, askpass, environment, cx).await, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { askpass_delegates.lock().insert(askpass_id, askpass); let _defer = util::defer(|| { let askpass_delegate = askpass_delegates.lock().remove(&askpass_id); debug_assert!(askpass_delegate.is_some()); }); let response = client .request(proto::Fetch { project_id: project_id.0, repository_id: id.to_proto(), askpass_id, remote: fetch_options.to_proto(), }) .await?; Ok(RemoteCommandOutput { stdout: response.stdout, stderr: response.stderr, }) } } }, ) } pub fn push( &mut self, branch: SharedString, remote_branch: SharedString, remote: SharedString, options: Option, askpass: AskPassDelegate, cx: &mut Context, ) -> oneshot::Receiver> { let askpass_delegates = self.askpass_delegates.clone(); let askpass_id = util::post_inc(&mut self.latest_askpass_id); let id = self.id; let args = options .map(|option| match option { PushOptions::SetUpstream => " --set-upstream", PushOptions::Force => " --force-with-lease", }) .unwrap_or(""); let updates_tx = self .git_store() .and_then(|git_store| match &git_store.read(cx).state { GitStoreState::Local { downstream, .. } => downstream .as_ref() .map(|downstream| downstream.updates_tx.clone()), _ => None, }); let this = cx.weak_entity(); self.send_job( "push", Some(format!("git push {} {} {}:{}", args, remote, branch, remote_branch).into()), move |git_repo, mut cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => { let result = backend .push( branch.to_string(), remote_branch.to_string(), remote.to_string(), options, askpass, environment.clone(), cx.clone(), ) .await; // TODO would be nice to not have to do this manually if result.is_ok() { let branches_scan = backend.branches().await?; let branch_list_error = branches_scan.error; let branch_list: Arc<[Branch]> = branches_scan.branches.into(); let branch = branch_list.iter().find(|branch| branch.is_head).cloned(); log::info!("head branch after scan is {branch:?}"); let snapshot = this.update(&mut cx, |this, cx| { let branch_list_changed = *branch_list != *this.snapshot.branch_list; let branch_list_error_changed = this.snapshot.branch_list_error != branch_list_error; this.snapshot.branch = branch; this.snapshot.branch_list = branch_list; this.snapshot.branch_list_error = branch_list_error; cx.emit(RepositoryEvent::HeadChanged); if branch_list_changed || branch_list_error_changed { cx.emit(RepositoryEvent::BranchListChanged); } this.snapshot.clone() })?; if let Some(updates_tx) = updates_tx { updates_tx .unbounded_send(DownstreamUpdate::UpdateRepository(snapshot)) .ok(); } } result } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { askpass_delegates.lock().insert(askpass_id, askpass); let _defer = util::defer(|| { let askpass_delegate = askpass_delegates.lock().remove(&askpass_id); debug_assert!(askpass_delegate.is_some()); }); let response = client .request(proto::Push { project_id: project_id.0, repository_id: id.to_proto(), askpass_id, branch_name: branch.to_string(), remote_branch_name: remote_branch.to_string(), remote_name: remote.to_string(), options: options.map(|options| match options { PushOptions::Force => proto::push::PushOptions::Force, PushOptions::SetUpstream => { proto::push::PushOptions::SetUpstream } } as i32), }) .await?; Ok(RemoteCommandOutput { stdout: response.stdout, stderr: response.stderr, }) } } }, ) } pub fn pull( &mut self, branch: Option, remote: SharedString, rebase: bool, askpass: AskPassDelegate, _cx: &mut App, ) -> oneshot::Receiver> { let askpass_delegates = self.askpass_delegates.clone(); let askpass_id = util::post_inc(&mut self.latest_askpass_id); let id = self.id; let mut status = "git pull".to_string(); if rebase { status.push_str(" --rebase"); } status.push_str(&format!(" {}", remote)); if let Some(b) = &branch { status.push_str(&format!(" {}", b)); } self.send_job( "pull", Some(status.into()), move |git_repo, cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, environment, .. }) => { backend .pull( branch.as_ref().map(|b| b.to_string()), remote.to_string(), rebase, askpass, environment.clone(), cx, ) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { askpass_delegates.lock().insert(askpass_id, askpass); let _defer = util::defer(|| { let askpass_delegate = askpass_delegates.lock().remove(&askpass_id); debug_assert!(askpass_delegate.is_some()); }); let response = client .request(proto::Pull { project_id: project_id.0, repository_id: id.to_proto(), askpass_id, rebase, branch_name: branch.as_ref().map(|b| b.to_string()), remote_name: remote.to_string(), }) .await?; Ok(RemoteCommandOutput { stdout: response.stdout, stderr: response.stderr, }) } } }, ) } fn spawn_set_index_text_job( &mut self, path: RepoPath, content: Option, hunk_staging_operation_count: Option, cx: &mut Context, ) -> oneshot::Receiver> { let id = self.id; let this = cx.weak_entity(); let git_store = self.git_store.clone(); let abs_path = self.snapshot.repo_path_to_abs_path(&path); self.send_keyed_job( "spawn_set_index_text_job", Some(GitJobKey::WriteIndex(vec![path.clone()])), None, move |git_repo, mut cx| async move { log::debug!( "start updating index text for buffer {}", path.as_unix_str() ); match git_repo { RepositoryState::Local(LocalRepositoryState { fs, backend, environment, .. }) => { let executable = match fs.metadata(&abs_path).await { Ok(Some(meta)) => meta.is_executable, Ok(None) => false, Err(_err) => false, }; backend .set_index_text(path.clone(), content, environment.clone(), executable) .await?; } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::SetIndexText { project_id: project_id.0, repository_id: id.to_proto(), path: path.as_unix_str().to_owned(), text: content, }) .await?; } } log::debug!( "finish updating index text for buffer {}", path.as_unix_str() ); if let Some(hunk_staging_operation_count) = hunk_staging_operation_count { let project_path = this .read_with(&cx, |this, cx| this.repo_path_to_project_path(&path, cx)) .ok() .flatten(); git_store .update(&mut cx, |git_store, cx| { let buffer_id = git_store .buffer_store .read(cx) .get_by_path(&project_path?)? .read(cx) .remote_id(); let diff_state = git_store.diffs.get(&buffer_id)?; diff_state.update(cx, |diff_state, _| { diff_state.hunk_staging_operation_count_as_of_write = hunk_staging_operation_count; }); Some(()) }) .context("Git store dropped")?; } Ok(()) }, ) } pub fn create_remote( &mut self, remote_name: String, remote_url: String, ) -> oneshot::Receiver> { let id = self.id; self.send_job( "create_remote", Some(format!("git remote add {remote_name} {remote_url}").into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.create_remote(remote_name, remote_url).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitCreateRemote { project_id: project_id.0, repository_id: id.to_proto(), remote_name, remote_url, }) .await?; Ok(()) } } }, ) } pub fn remove_remote(&mut self, remote_name: String) -> oneshot::Receiver> { let id = self.id; self.send_job( "remove_remote", Some(format!("git remove remote {remote_name}").into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.remove_remote(remote_name).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRemoveRemote { project_id: project_id.0, repository_id: id.to_proto(), remote_name, }) .await?; Ok(()) } } }, ) } pub fn get_remotes( &mut self, branch_name: Option, is_push: bool, ) -> oneshot::Receiver>> { let id = self.id; self.send_job("get_remotes", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { let remote = if let Some(branch_name) = branch_name { if is_push { backend.get_push_remote(branch_name).await? } else { backend.get_branch_remote(branch_name).await? } } else { None }; match remote { Some(remote) => Ok(vec![remote]), None => backend.get_all_remotes().await, } } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GetRemotes { project_id: project_id.0, repository_id: id.to_proto(), branch_name, is_push, }) .await?; let remotes = response .remotes .into_iter() .map(|remotes| Remote { name: remotes.name.into(), }) .collect(); Ok(remotes) } } }) } pub fn branches(&mut self) -> oneshot::Receiver> { let id = self.id; self.send_job("branches", None, move |repo, _| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.branches().await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitGetBranches { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; let branches = response .branches .into_iter() .map(|branch| proto_to_branch(&branch)) .collect(); Ok(BranchesScanResult { branches, error: response.error.map(SharedString::from), }) } } }) } /// If this is a linked worktree (*NOT* the main checkout of a repository), /// returns the path for the linked worktree. /// /// Returns None if this is the main checkout. pub fn linked_worktree_path(&self) -> Option<&Arc> { self.snapshot .is_linked_worktree() .then_some(&self.work_directory_abs_path) } pub fn path_for_new_linked_worktree( &self, branch_name: &str, worktree_directory_setting: &str, ) -> Result { let repository_anchor = self .snapshot .main_worktree_abs_path() .unwrap_or(self.common_dir_abs_path.as_ref()); let project_name = repository_anchor .file_name() .and_then(|name| name.to_str()) .ok_or_else(|| anyhow!("git repo must have a directory name"))?; let directory = worktrees_directory_for_repo( repository_anchor, worktree_directory_setting, self.path_style, )?; let directory = self.path_style.join_path(&directory, branch_name)?; self.path_style.join_path(&directory, project_name) } pub fn worktrees(&mut self) -> oneshot::Receiver>> { let id = self.id; self.send_job("worktrees", None, move |repo, _| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.worktrees().await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitGetWorktrees { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; let worktrees = response .worktrees .into_iter() .map(|worktree| proto_to_worktree(&worktree)) .collect(); Ok(worktrees) } } }) } pub fn create_worktree( &mut self, target: CreateWorktreeTarget, path: PathBuf, ) -> oneshot::Receiver> { let id = self.id; let job_description = match target.branch_name() { Some(branch_name) => format!("git worktree add: {branch_name}"), None => "git worktree add (detached)".to_string(), }; self.send_job( "create_worktree", Some(job_description.into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.create_worktree(target, path).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let (name, commit, use_existing_branch) = match target { CreateWorktreeTarget::ExistingBranch { branch_name } => { (Some(branch_name), None, true) } CreateWorktreeTarget::NewBranch { branch_name, base_sha, } => (Some(branch_name), base_sha, false), CreateWorktreeTarget::Detached { base_sha } => (None, base_sha, false), }; client .request(proto::GitCreateWorktree { project_id: project_id.0, repository_id: id.to_proto(), name: name.unwrap_or_default(), directory: path.to_string_lossy().to_string(), commit, use_existing_branch, }) .await?; Ok(()) } } }, ) } /// Returns the creation time of a linked worktree's git metadata /// directory. See [`GitRepository::worktree_created_at`]. For remote /// projects the stat runs on the remote host, where the worktree's /// filesystem lives. pub fn worktree_created_at( &mut self, worktree_path: PathBuf, ) -> oneshot::Receiver>> { let id = self.id; self.send_job("worktree_created_at", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.worktree_created_at(worktree_path).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitWorktreeCreatedAt { project_id: project_id.0, repository_id: id.to_proto(), worktree_path: worktree_path.to_string_lossy().to_string(), }) .await?; Ok(response.created_at.map(SystemTime::from)) } } }) } pub fn create_worktree_detached( &mut self, path: PathBuf, commit: String, ) -> oneshot::Receiver> { self.create_worktree( CreateWorktreeTarget::Detached { base_sha: Some(commit), }, path, ) } pub fn checkout_branch_in_worktree( &mut self, branch_name: String, worktree_path: PathBuf, create: bool, ) -> oneshot::Receiver> { let description = if create { format!("git checkout -b {branch_name}") } else { format!("git checkout {branch_name}") }; self.send_job( "checkout_branch_in_worktree", Some(description.into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend .checkout_branch_in_worktree(branch_name, worktree_path, create) .await } RepositoryState::Remote(_) => { log::warn!( "checkout_branch_in_worktree not supported for remote repositories" ); Ok(()) } } }, ) } pub fn head_sha(&mut self) -> oneshot::Receiver>> { let id = self.id; self.send_job("head_sha", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { Ok(backend.head_sha().await) } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitGetHeadSha { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; Ok(response.sha) } } }) } fn edit_ref( &mut self, ref_name: String, commit: Option, ) -> oneshot::Receiver> { let id = self.id; self.send_job("edit_ref", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => match commit { Some(commit) => backend.update_ref(ref_name, commit).await, None => backend.delete_ref(ref_name).await, }, RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let action = match commit { Some(sha) => proto::git_edit_ref::Action::UpdateToCommit(sha), None => { proto::git_edit_ref::Action::Delete(proto::git_edit_ref::DeleteRef {}) } }; client .request(proto::GitEditRef { project_id: project_id.0, repository_id: id.to_proto(), ref_name, action: Some(action), }) .await?; Ok(()) } } }) } pub fn update_ref( &mut self, ref_name: String, commit: String, ) -> oneshot::Receiver> { self.edit_ref(ref_name, Some(commit)) } pub fn delete_ref(&mut self, ref_name: String) -> oneshot::Receiver> { self.edit_ref(ref_name, None) } pub fn repair_worktrees(&mut self) -> oneshot::Receiver> { let id = self.id; self.send_job("repair_worktrees", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.repair_worktrees().await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRepairWorktrees { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; Ok(()) } } }) } pub fn create_archive_checkpoint(&mut self) -> oneshot::Receiver> { let id = self.id; self.send_job( "create_archive_checkpoint", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.create_archive_checkpoint().await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitCreateArchiveCheckpoint { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; Ok((response.staged_commit_sha, response.unstaged_commit_sha)) } } }, ) } pub fn restore_archive_checkpoint( &mut self, staged_sha: String, unstaged_sha: String, ) -> oneshot::Receiver> { let id = self.id; self.send_job( "restore_archive_checkpoint", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend .restore_archive_checkpoint(staged_sha, unstaged_sha) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRestoreArchiveCheckpoint { project_id: project_id.0, repository_id: id.to_proto(), staged_commit_sha: staged_sha, unstaged_commit_sha: unstaged_sha, }) .await?; Ok(()) } } }, ) } pub fn remove_worktree(&mut self, path: PathBuf, force: bool) -> oneshot::Receiver> { let id = self.id; let repository_anchor_path: Arc = self .snapshot .main_worktree_abs_path() .unwrap_or(self.snapshot.common_dir_abs_path.as_ref()) .into(); self.send_job( "remove_worktree", Some(format!("git worktree remove: {}", path.display()).into()), move |repo, cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, fs, .. }) => { // When forcing, delete the worktree directory ourselves before // invoking git. `git worktree remove` can remove the admin // metadata in `.git/worktrees/` but fail to delete the // working directory (it continues past directory-removal errors), // leaving an orphaned folder on disk. Deleting first guarantees // the directory is gone, and `git worktree remove --force` // tolerates a missing working tree while cleaning up the admin // entry. We keep this inside the `Local` arm so that for remote // projects the deletion runs on the remote machine (where the // `GitRemoveWorktree` RPC is handled against the local repo on // the headless server) using its own filesystem. // // After a successful removal, also delete any empty ancestor // directories between the worktree path and the configured // base directory used when creating linked worktrees. // // Non-force removals are left untouched before git runs: // `git worktree remove` must see the dirty working tree to // refuse the operation. if force { fs.remove_dir( &path, RemoveOptions { recursive: true, ignore_if_not_exists: true, }, ) .await .with_context(|| { format!("failed to delete worktree directory '{}'", path.display()) })?; } backend.remove_worktree(path.clone(), force).await?; let managed_worktree_base = cx.update(|cx| { let setting = &ProjectSettings::get_global(cx).git.worktree_directory; worktrees_directory_for_repo( &repository_anchor_path, setting, PathStyle::local(), ) .log_err() }); if let Some(managed_worktree_base) = managed_worktree_base { remove_empty_managed_worktree_ancestors( fs.as_ref(), &path, &managed_worktree_base, ) .await; } Ok(()) } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRemoveWorktree { project_id: project_id.0, repository_id: id.to_proto(), path: path.to_string_lossy().to_string(), force, }) .await?; Ok(()) } } }, ) } pub fn rename_worktree( &mut self, old_path: PathBuf, new_path: PathBuf, ) -> oneshot::Receiver> { let id = self.id; self.send_job( "rename_worktree", Some(format!("git worktree move: {}", old_path.display()).into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.rename_worktree(old_path, new_path).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRenameWorktree { project_id: project_id.0, repository_id: id.to_proto(), old_path: old_path.to_string_lossy().to_string(), new_path: new_path.to_string_lossy().to_string(), }) .await?; Ok(()) } } }, ) } pub fn default_branch( &mut self, include_remote_name: bool, ) -> oneshot::Receiver>> { let id = self.id; self.send_job("default_branch", None, move |repo, _| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.default_branch(include_remote_name).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GetDefaultBranch { project_id: project_id.0, repository_id: id.to_proto(), include_remote_name, }) .await?; anyhow::Ok(response.branch.map(SharedString::from)) } } }) } pub fn diff_tree( &mut self, diff_type: DiffTreeType, _cx: &App, ) -> oneshot::Receiver> { let repository_id = self.snapshot.id; self.send_job("diff_tree", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.diff_tree(diff_type).await } RepositoryState::Remote(RemoteRepositoryState { client, project_id }) => { let response = client .request(proto::GetTreeDiff { project_id: project_id.0, repository_id: repository_id.0, is_merge: matches!(diff_type, DiffTreeType::MergeBase { .. }), base: diff_type.base().to_string(), head: diff_type.head().to_string(), }) .await?; let entries = response .entries .into_iter() .filter_map(|entry| { let status = match entry.status() { proto::tree_diff_status::Status::Added => TreeDiffStatus::Added, proto::tree_diff_status::Status::Modified => { TreeDiffStatus::Modified { old: git::Oid::from_str( &entry.oid.context("missing oid").log_err()?, ) .log_err()?, } } proto::tree_diff_status::Status::Deleted => { TreeDiffStatus::Deleted { old: git::Oid::from_str( &entry.oid.context("missing oid").log_err()?, ) .log_err()?, } } }; Some(( RepoPath::from_rel_path( RelPath::from_unix_str(&entry.path).log_err()?, ), status, )) }) .collect(); Ok(TreeDiff { entries }) } } }) } pub fn diff(&mut self, diff_type: DiffType, _cx: &App) -> oneshot::Receiver> { let id = self.id; self.send_job("diff", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.diff(diff_type).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let (proto_diff_type, merge_base_ref) = match &diff_type { DiffType::HeadToIndex => { (proto::git_diff::DiffType::HeadToIndex.into(), None) } DiffType::HeadToWorktree => { (proto::git_diff::DiffType::HeadToWorktree.into(), None) } DiffType::MergeBase { base_ref } => ( proto::git_diff::DiffType::MergeBase.into(), Some(base_ref.to_string()), ), }; let response = client .request(proto::GitDiff { project_id: project_id.0, repository_id: id.to_proto(), diff_type: proto_diff_type, merge_base_ref, }) .await?; Ok(response.diff) } } }) } pub fn create_branch( &mut self, branch_name: String, base_branch: Option, ) -> oneshot::Receiver> { let id = self.id; let status_msg = if let Some(ref base) = base_branch { format!("git switch -c {branch_name} {base}").into() } else { format!("git switch -c {branch_name}").into() }; self.send_job( "create_branch", Some(status_msg), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.create_branch(branch_name, base_branch).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitCreateBranch { project_id: project_id.0, repository_id: id.to_proto(), branch_name, base_branch, }) .await?; Ok(()) } } }, ) } pub fn change_branch(&mut self, branch_name: String) -> oneshot::Receiver> { let id = self.id; self.send_job( "change_branch", Some(format!("git switch {branch_name}").into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.change_branch(branch_name).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitChangeBranch { project_id: project_id.0, repository_id: id.to_proto(), branch_name, }) .await?; Ok(()) } } }, ) } pub fn delete_branch( &mut self, is_remote: bool, branch_name: String, force: bool, ) -> oneshot::Receiver> { let id = self.id; let flag = delete_branch_flag(is_remote, force); self.send_job( "delete_branch", Some(format!("git branch {flag} {branch_name}").into()), move |repo, _cx| async move { match repo { RepositoryState::Local(state) => { state .backend .delete_branch(is_remote, branch_name, force) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitDeleteBranch { project_id: project_id.0, repository_id: id.to_proto(), is_remote, branch_name, force, }) .await?; Ok(()) } } }, ) } pub fn rename_branch( &mut self, branch: String, new_name: String, ) -> oneshot::Receiver> { let id = self.id; self.send_job( "rename_branch", Some(format!("git branch -m {branch} {new_name}").into()), move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.rename_branch(branch, new_name).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRenameBranch { project_id: project_id.0, repository_id: id.to_proto(), branch, new_name, }) .await?; Ok(()) } } }, ) } pub fn check_for_pushed_commits(&mut self) -> oneshot::Receiver>> { let id = self.id; self.send_job( "check_for_pushed_commits", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.check_for_pushed_commit().await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::CheckForPushedCommits { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; let branches = response.pushed_to.into_iter().map(Into::into).collect(); Ok(branches) } } }, ) } pub fn checkpoint(&mut self) -> oneshot::Receiver> { let id = self.id; self.send_job("checkpoint", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.checkpoint().await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitCreateCheckpoint { project_id: project_id.0, repository_id: id.to_proto(), }) .await?; Ok(GitRepositoryCheckpoint { commit_sha: Oid::from_bytes(&response.commit_sha)?, }) } } }) } pub fn restore_checkpoint( &mut self, checkpoint: GitRepositoryCheckpoint, ) -> oneshot::Receiver> { let id = self.id; self.send_job("restore_checkpoint", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.restore_checkpoint(checkpoint).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { client .request(proto::GitRestoreCheckpoint { project_id: project_id.0, repository_id: id.to_proto(), commit_sha: checkpoint.commit_sha.as_bytes().to_vec(), }) .await?; Ok(()) } } }) } pub(crate) fn apply_remote_update( &mut self, update: proto::UpdateRepository, cx: &mut Context, ) -> Result<()> { if let Some(repository_dir_abs_path) = &update.repository_dir_abs_path { self.snapshot.repository_dir_abs_path = Path::new(repository_dir_abs_path.as_str()).into(); } if let Some(common_dir_abs_path) = &update.common_dir_abs_path { self.snapshot.common_dir_abs_path = Path::new(common_dir_abs_path.as_str()).into(); } let new_branch = update.branch_summary.as_ref().map(proto_to_branch); let new_head_commit = update .head_commit_details .as_ref() .map(proto_to_commit_details); if self.snapshot.branch != new_branch || self.snapshot.head_commit != new_head_commit { cx.emit(RepositoryEvent::HeadChanged) } self.snapshot.branch = new_branch; self.snapshot.head_commit = new_head_commit; if update.is_last_update { let new_branch_list: Arc<[Branch]> = update.branch_list.iter().map(proto_to_branch).collect(); let new_branch_list_error = update.branch_list_error.map(SharedString::from); if *self.snapshot.branch_list != *new_branch_list || self.snapshot.branch_list_error != new_branch_list_error { cx.emit(RepositoryEvent::BranchListChanged); } self.snapshot.branch_list = new_branch_list; self.snapshot.branch_list_error = new_branch_list_error; } // We don't store any merge head state for downstream projects; the upstream // will track it and we will just get the updated conflicts let new_merge_heads = TreeMap::from_ordered_entries( update .current_merge_conflicts .into_iter() .filter_map(|path| Some((RepoPath::from_proto(&path).ok()?, vec![]))), ); let conflicts_changed = self.snapshot.merge.merge_heads_by_conflicted_path != new_merge_heads; self.snapshot.merge.merge_heads_by_conflicted_path = new_merge_heads; self.snapshot.merge.message = update.merge_message.map(SharedString::from); let new_stash_entries = GitStash { entries: update .stash_entries .iter() .filter_map(|entry| proto_to_stash(entry).ok()) .collect(), }; if self.snapshot.stash_entries != new_stash_entries { cx.emit(RepositoryEvent::StashEntriesChanged) } self.snapshot.stash_entries = new_stash_entries; let new_linked_worktrees: Arc<[GitWorktree]> = update .linked_worktrees .iter() .map(proto_to_worktree) .collect(); if *self.snapshot.linked_worktrees != *new_linked_worktrees { cx.emit(RepositoryEvent::GitWorktreeListChanged); } self.snapshot.linked_worktrees = new_linked_worktrees; self.snapshot.remote_upstream_url = update.remote_upstream_url; self.snapshot.remote_origin_url = update.remote_origin_url; let edits = update .removed_statuses .into_iter() .filter_map(|path| { Some(sum_tree::Edit::Remove(PathKey( RelPath::from_unix_str(&path).log_err()?.into(), ))) }) .chain( update .updated_statuses .into_iter() .filter_map(|updated_status| { Some(sum_tree::Edit::Insert(updated_status.try_into().log_err()?)) }), ) .collect::>(); if conflicts_changed || !edits.is_empty() { cx.emit(RepositoryEvent::StatusesChanged); } self.snapshot.statuses_by_path.edit(edits, ()); if update.is_last_update { self.snapshot.scan_id = update.scan_id; } self.clear_pending_ops(cx); Ok(()) } pub fn compare_checkpoints( &mut self, left: GitRepositoryCheckpoint, right: GitRepositoryCheckpoint, ) -> oneshot::Receiver> { let id = self.id; self.send_job("compare_checkpoints", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.compare_checkpoints(left, right).await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitCompareCheckpoints { project_id: project_id.0, repository_id: id.to_proto(), left_commit_sha: left.commit_sha.as_bytes().to_vec(), right_commit_sha: right.commit_sha.as_bytes().to_vec(), }) .await?; Ok(response.equal) } } }) } pub fn diff_checkpoints( &mut self, base_checkpoint: GitRepositoryCheckpoint, target_checkpoint: GitRepositoryCheckpoint, ) -> oneshot::Receiver> { let id = self.id; self.send_job("diff_checkpoints", None, move |repo, _cx| async move { match repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend .diff_checkpoints(base_checkpoint, target_checkpoint) .await } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::GitDiffCheckpoints { project_id: project_id.0, repository_id: id.to_proto(), base_commit_sha: base_checkpoint.commit_sha.as_bytes().to_vec(), target_commit_sha: target_checkpoint.commit_sha.as_bytes().to_vec(), }) .await?; Ok(response.diff) } } }) } fn clear_pending_ops(&mut self, cx: &mut Context) { let updated = SumTree::from_iter( self.pending_ops.iter().filter_map(|ops| { let inner_ops: Vec = ops.ops.iter().filter(|op| op.running()).cloned().collect(); if inner_ops.is_empty() { None } else { Some(PendingOps { repo_path: ops.repo_path.clone(), ops: inner_ops, }) } }), (), ); if updated != self.pending_ops { cx.emit(RepositoryEvent::PendingOpsChanged { pending_ops: self.pending_ops.clone(), }) } self.pending_ops = updated; } fn schedule_scan( &mut self, updates_tx: Option>, cx: &mut Context, ) { let this = cx.weak_entity(); let _ = self.send_keyed_job( "schedule_scan", Some(GitJobKey::ReloadGitState), None, |state, mut cx| async move { log::debug!("run scheduled git status scan"); let Some(this) = this.upgrade() else { return Ok(()); }; let RepositoryState::Local(LocalRepositoryState { backend, .. }) = state else { bail!("not a local repository") }; let snapshot = compute_snapshot(this.clone(), backend.clone(), &mut cx).await; this.update(&mut cx, |this, cx| { this.clear_pending_ops(cx); }); if let Some(updates_tx) = updates_tx { updates_tx .unbounded_send(DownstreamUpdate::UpdateRepository(snapshot)) .ok(); } Ok(()) }, ); } fn spawn_local_git_worker( state: Shared>>, cx: &mut Context, ) -> (mpsc::UnboundedSender, Task<()>) { let (job_tx, mut job_rx) = mpsc::unbounded::(); let worker_task = cx.spawn(async move |this, cx| { let Some(state) = state.await.log_err() else { return; }; if let Some(git_hosting_provider_registry) = cx.update(|cx| GitHostingProviderRegistry::try_global(cx)) { git_hosting_providers::register_additional_providers( git_hosting_provider_registry, state.backend.clone(), ) .await; } let state = RepositoryState::Local(state); let mut jobs = VecDeque::new(); loop { while let Ok(next_job) = job_rx.try_recv() { jobs.push_back(next_job); } if let Some(job) = jobs.pop_front() { if let Some(current_key) = &job.key && jobs .iter() .any(|other_job| other_job.key.as_ref() == Some(current_key)) { let skipped_job_id = job.id; this.update(cx, |repo, _| { repo.job_debug_queue.mark_complete( skipped_job_id, job_debug_queue::CompletedJobStatus::Skipped, ); }) .ok(); continue; } (job.job)(state.clone(), cx).await; } else if let Some(job) = job_rx.next().await { jobs.push_back(job); } else { break; } } }); (job_tx, worker_task) } fn spawn_remote_git_worker( state: RemoteRepositoryState, cx: &mut Context, ) -> (mpsc::UnboundedSender, Task<()>) { let (job_tx, mut job_rx) = mpsc::unbounded::(); let worker_task = cx.spawn(async move |this, cx| { let result: Result<()> = async { let state = RepositoryState::Remote(state); let mut jobs = VecDeque::new(); loop { while let Ok(next_job) = job_rx.try_recv() { jobs.push_back(next_job); } if let Some(job) = jobs.pop_front() { if let Some(current_key) = &job.key && jobs .iter() .any(|other_job| other_job.key.as_ref() == Some(current_key)) { let skipped_job_id = job.id; this.update(cx, |repo, _| { repo.job_debug_queue.mark_complete( skipped_job_id, job_debug_queue::CompletedJobStatus::Skipped, ); }) .ok(); continue; } (job.job)(state.clone(), cx).await; } else if let Some(job) = job_rx.next().await { jobs.push_back(job); } else { break; } } anyhow::Ok(()) } .await; result.log_err(); }); (job_tx, worker_task) } fn load_staged_text( &mut self, buffer_id: BufferId, repo_path: RepoPath, cx: &App, ) -> Task>> { let rx = self.send_job("load_staged_text", None, move |state, _| async move { match state { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { anyhow::Ok(backend.load_index_text(repo_path).await) } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { let response = client .request(proto::OpenUnstagedDiff { project_id: project_id.to_proto(), buffer_id: buffer_id.to_proto(), }) .await?; Ok(response.staged_text) } } }); cx.spawn(|_: &mut AsyncApp| async move { rx.await? }) } fn load_committed_text( &mut self, buffer_id: BufferId, repo_path: RepoPath, cx: &App, ) -> Task> { let rx = self.send_job("load_committed_text", None, move |state, _| async move { match state { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { let revisions = vec![ format!("HEAD:{}", repo_path.as_unix_str()), format!(":{}", repo_path.as_unix_str()), ]; let mut loaded_revisions = backend.load_revisions(revisions).await?.into_iter(); let committed_text = loaded_revisions.next().flatten(); let staged_text = loaded_revisions.next().flatten(); let diff_bases_change = if committed_text == staged_text { DiffBasesChange::SetBoth(committed_text) } else { DiffBasesChange::SetEach { index: staged_text, head: committed_text, } }; anyhow::Ok(diff_bases_change) } RepositoryState::Remote(RemoteRepositoryState { project_id, client }) => { use proto::open_uncommitted_diff_response::Mode; let response = client .request(proto::OpenUncommittedDiff { project_id: project_id.to_proto(), buffer_id: buffer_id.to_proto(), }) .await?; let mode = Mode::from_i32(response.mode).context("Invalid mode")?; let bases = match mode { Mode::IndexMatchesHead => DiffBasesChange::SetBoth(response.committed_text), Mode::IndexAndHead => DiffBasesChange::SetEach { head: response.committed_text, index: response.staged_text, }, }; Ok(bases) } } }); cx.spawn(|_: &mut AsyncApp| async move { rx.await? }) } pub fn load_commit_template_text( &mut self, ) -> oneshot::Receiver>> { self.send_job( "load_commit_template_text", None, move |git_repo, _cx| async move { match git_repo { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.load_commit_template().await } RepositoryState::Remote(_) => Ok(None), } }, ) } fn load_blob_content(&mut self, oid: Oid, cx: &App) -> Task> { let repository_id = self.snapshot.id; let rx = self.send_job("load_blob_content", None, move |state, _| async move { match state { RepositoryState::Local(LocalRepositoryState { backend, .. }) => { backend.load_blob_content(oid).await } RepositoryState::Remote(RemoteRepositoryState { client, project_id }) => { let response = client .request(proto::GetBlobContent { project_id: project_id.to_proto(), repository_id: repository_id.0, oid: oid.to_string(), }) .await?; Ok(response.content) } } }); cx.spawn(|_: &mut AsyncApp| async move { rx.await? }) } fn paths_changed( &mut self, paths: Vec, updates_tx: Option>, cx: &mut Context, ) { if !paths.is_empty() { self.paths_needing_status_update.push(paths); } let this = cx.weak_entity(); let _ = self.send_keyed_job( "paths_changed", Some(GitJobKey::RefreshStatuses), None, |state, mut cx| async move { let (prev_snapshot, changed_paths) = this.update(&mut cx, |this, _| { ( this.snapshot.clone(), mem::take(&mut this.paths_needing_status_update), ) })?; let RepositoryState::Local(LocalRepositoryState { backend, .. }) = state else { bail!("not a local repository") }; if changed_paths.is_empty() { return Ok(()); } let has_head = prev_snapshot.head_commit.is_some(); let changed_path_statuses = cx .background_spawn(async move { let changed_paths = GitStore::coalesce_repo_paths( changed_paths .into_iter() .flatten() .collect::>() .into_iter() .collect(), ); let changed_paths_vec = changed_paths.iter().cloned().collect::>(); let status_task = backend.status(&changed_paths_vec); let diff_stat_future = |diff| { if has_head { backend.diff_stat(diff, &changed_paths_vec) } else { future::ready(Ok(status::GitDiffStat::default())).boxed() } }; let (statuses, diff_stats, staged_diff_stats, unstaged_diff_stats) = futures::future::try_join4( status_task, diff_stat_future(DiffStatType::HeadToWorktree), diff_stat_future(DiffStatType::HeadToIndex), diff_stat_future(DiffStatType::IndexToWorktree), ) .await?; let diff_stats: HashMap = HashMap::from_iter(diff_stats.entries.into_iter().cloned()); let staged_diff_stats: HashMap = HashMap::from_iter(staged_diff_stats.entries.into_iter().cloned()); let unstaged_diff_stats: HashMap = HashMap::from_iter(unstaged_diff_stats.entries.into_iter().cloned()); let mut changed_path_statuses = Vec::new(); let prev_statuses = prev_snapshot.statuses_by_path.clone(); let current_status_paths = statuses .entries .iter() .map(|(repo_path, _)| repo_path.clone()) .collect::>(); for path in &changed_paths { let mut cursor = prev_statuses.cursor::(()); cursor.seek_forward(&PathTarget::Path(path), Bias::Left); while let Some(entry) = cursor.item() { if !entry.repo_path.starts_with(path) { break; } if !current_status_paths.contains(&entry.repo_path) { changed_path_statuses.push(Edit::Remove(PathKey( entry.repo_path.as_ref().clone(), ))); } cursor.next(); } } let mut cursor = prev_statuses.cursor::(()); for (repo_path, status) in &*statuses.entries { let current_diff_stat = diff_stats.get(repo_path).copied(); let current_staged_diff_stat = staged_diff_stats.get(repo_path).copied(); let current_unstaged_diff_stat = unstaged_diff_stats.get(repo_path).copied(); if cursor.seek_forward(&PathTarget::Path(repo_path), Bias::Left) && cursor.item().is_some_and(|entry| { entry.status == *status && entry.diff_stat == current_diff_stat && entry.staged_diff_stat == current_staged_diff_stat && entry.unstaged_diff_stat == current_unstaged_diff_stat }) { continue; } changed_path_statuses.push(Edit::Insert(StatusEntry { repo_path: repo_path.clone(), status: *status, diff_stat: current_diff_stat, staged_diff_stat: current_staged_diff_stat, unstaged_diff_stat: current_unstaged_diff_stat, })); } anyhow::Ok(changed_path_statuses) }) .await?; this.update(&mut cx, |this, cx| { if !changed_path_statuses.is_empty() { cx.emit(RepositoryEvent::StatusesChanged); this.snapshot .statuses_by_path .edit(changed_path_statuses, ()); this.snapshot.scan_id += 1; } if let Some(updates_tx) = updates_tx { updates_tx .unbounded_send(DownstreamUpdate::UpdateRepository( this.snapshot.clone(), )) .ok(); } }) }, ); } /// currently running git command and when it started pub fn current_job(&self) -> Option { self.active_jobs.values().next().cloned() } pub fn job_debug_queue(&self) -> &job_debug_queue::GitJobDebugQueue { &self.job_debug_queue } pub fn barrier(&mut self) -> oneshot::Receiver<()> { self.send_job("barrier", None, |_, _| async {}) } fn spawn_job_with_tracking( &mut self, paths: Vec, git_status: pending_op::GitStatus, cx: &mut Context, f: AsyncFn, ) -> Task> where AsyncFn: AsyncFnOnce(WeakEntity, &mut AsyncApp) -> Result<()> + 'static, { let ids = self.new_pending_ops_for_paths(paths, git_status); cx.spawn(async move |this, cx| { let (job_status, result) = match f(this.clone(), cx).await { Ok(()) => (pending_op::JobStatus::Finished, Ok(())), Err(err) if err.is::() => (pending_op::JobStatus::Skipped, Ok(())), Err(err) => (pending_op::JobStatus::Error, Err(err)), }; this.update(cx, |this, _| { let mut edits = Vec::with_capacity(ids.len()); for (id, entry) in ids { if let Some(mut ops) = this .pending_ops .get(&PathKey(entry.as_ref().clone()), ()) .cloned() { if let Some(op) = ops.op_by_id_mut(id) { op.job_status = job_status; } edits.push(sum_tree::Edit::Insert(ops)); } } this.pending_ops.edit(edits, ()); })?; result }) } fn new_pending_ops_for_paths( &mut self, paths: Vec, git_status: pending_op::GitStatus, ) -> Vec<(PendingOpId, RepoPath)> { let mut edits = Vec::with_capacity(paths.len()); let mut ids = Vec::with_capacity(paths.len()); for path in paths { let mut ops = self .pending_ops .get(&PathKey(path.as_ref().clone()), ()) .cloned() .unwrap_or_else(|| PendingOps::new(&path)); let id = ops.max_id() + 1; ops.ops.push(PendingOp { id, git_status, job_status: pending_op::JobStatus::Running, }); edits.push(sum_tree::Edit::Insert(ops)); ids.push((id, path)); } self.pending_ops.edit(edits, ()); ids } pub fn access(&mut self, _cx: &App) -> oneshot::Receiver { self.send_job("access", None, move |git_repo, _cx| async move { match git_repo { // TODO: Correctly handle remote repositories, where the user // that's running the Zed remote may not own the `.git/` // directory. For now we just return `GitAccess::Yes` so that // remoting continues working as expected. RepositoryState::Remote(..) => GitAccess::Yes, RepositoryState::Local(state) => match state.backend.check_access().await { Ok(_) => GitAccess::Yes, Err(_) => GitAccess::No, }, } }) } pub fn default_remote_url(&self) -> Option { self.remote_upstream_url .clone() .or(self.remote_origin_url.clone()) } } fn format_job_key(key: &GitJobKey) -> SharedString { match key { GitJobKey::WriteIndex(paths) => { let paths_str: Vec<_> = paths .iter() .map(|p| { let rel: &RelPath = p; rel.display(PathStyle::local()) }) .collect(); format!("WriteIndex({})", paths_str.join(", ")).into() } GitJobKey::ReloadBufferDiffBases => "ReloadBufferDiffBases".into(), GitJobKey::RefreshStatuses => "RefreshStatuses".into(), GitJobKey::ReloadGitState => "ReloadGitState".into(), } } /// If `path` is a git linked worktree checkout, resolves it to the main /// repository's identity path. For regular linked worktrees this is the main /// repository's working directory; for linked worktrees backed by a bare repo /// such as `.bare`, this is the parent project directory users think of as the /// repository root. Returns `None` if `path` is a normal repository, not a git /// repo, or if resolution fails. /// /// Resolution works by: /// 1. Reading the `.git` file to get the `gitdir:` pointer /// 2. Following that to the worktree-specific git directory /// 3. Reading the `commondir` file to find the shared `.git` directory /// 4. Deriving the main repo's identity path from the common dir pub async fn resolve_git_worktree_to_main_repo(fs: &dyn Fs, path: &Path) -> Option { let dot_git = path.join(".git"); let metadata = fs.metadata(&dot_git).await.ok()??; if metadata.is_dir { return None; // Normal repo, not a linked worktree } // It's a .git file — parse the gitdir: pointer let content = fs.load(&dot_git).await.ok()?; let gitdir_rel = content.strip_prefix("gitdir:")?.trim(); let gitdir_abs = fs.canonicalize(&path.join(gitdir_rel)).await.ok()?; // Read commondir to find the main .git directory let commondir_content = fs.load(&gitdir_abs.join("commondir")).await.ok()?; let common_dir = fs .canonicalize(&gitdir_abs.join(commondir_content.trim())) .await .ok()?; Some(repo_identity_path(&common_dir).to_path_buf()) } /// Validates that the resolved worktree directory is acceptable: /// - The setting must not be an absolute path. /// - The resolved path must be either a subdirectory of the working /// directory or a subdirectory of its parent (i.e., a sibling). /// /// Returns `Ok(resolved_path)` or an error with a user-facing message. pub fn worktrees_directory_for_repo( repository_anchor_path: &Path, worktree_directory_setting: &str, path_style: PathStyle, ) -> Result { // Check the original setting before trimming, since a path like "///" // is absolute but becomes "" after stripping trailing separators. // Also check for leading `/` or `\` explicitly, because on Windows // `Path::is_absolute()` requires a drive letter — so `/tmp/worktrees` // would slip through even though it's clearly not a relative path. if path_style.is_absolute(worktree_directory_setting) || worktree_directory_setting.starts_with('\\') { anyhow::bail!( "git.worktree_directory must be a relative path, got: {worktree_directory_setting:?}" ); } if worktree_directory_setting.is_empty() { anyhow::bail!("git.worktree_directory must not be empty"); } let trimmed = worktree_directory_setting.trim_end_matches(['/', '\\']); if trimmed == ".." { anyhow::bail!("git.worktree_directory must not be \"..\" (use \"../some-name\" instead)"); } let joined = path_style.join_path(repository_anchor_path, trimmed)?; let resolved = if path_style.is_posix() { joined } else { path::normalize_path(&joined) }; let resolved = if resolved.starts_with(repository_anchor_path) { resolved } else if let Some(repo_dir_name) = repository_anchor_path .file_name() .and_then(|name| name.to_str()) { path_style.join_path(&resolved, repo_dir_name)? } else { resolved }; let parent = repository_anchor_path .parent() .unwrap_or(repository_anchor_path); if !resolved.starts_with(parent) { anyhow::bail!( "git.worktree_directory resolved to {resolved:?}, which is outside \ the project root and its parent directory. It must resolve to a \ subdirectory of {repository_anchor_path:?} or a sibling of it." ); } Ok(resolved) } async fn remove_empty_managed_worktree_ancestors(fs: &dyn Fs, child_path: &Path, base_path: &Path) { let mut current = child_path; while let Some(parent) = current.parent() { if parent == base_path { break; } if !parent.starts_with(base_path) { break; } let result = fs .remove_dir( parent, RemoveOptions { recursive: false, ignore_if_not_exists: true, }, ) .await; match result { Ok(()) => { log::info!( "Removed empty managed worktree directory: {}", parent.display() ); } Err(error) => { log::debug!( "Stopped removing managed worktree parent directories at {}: {error}", parent.display() ); break; } } current = parent; } } /// Returns the repository's identity path given its common Git directory. /// /// This is the canonical, on-disk path used for project grouping and as the /// basis for display names. The goal is to return the directory the user /// thinks of as "the project": /// /// - If `common_dir`'s last component starts with `.` (e.g. `.git` for a /// normal checkout, or `.bare` for a bare clone), the parent directory is /// returned. Both of these are internal Git directories; the parent is the /// meaningful project root. /// - Otherwise (e.g. `zed.git` for a bare clone), `common_dir` itself is /// returned — it is already a meaningful on-disk path. pub fn repo_identity_path(common_dir: &Path) -> &Path { let is_dot_entry = common_dir .file_name() .is_some_and(|n| n.to_string_lossy().starts_with('.')); if is_dot_entry { common_dir.parent().unwrap_or(common_dir) } else { common_dir } } /// Returns a short name for a linked worktree suitable for UI display /// /// Uses the main worktree path to come up with a short name that disambiguates /// the linked worktree from the main worktree. pub fn linked_worktree_short_name( main_worktree_path: &Path, linked_worktree_path: &Path, ) -> Option { if main_worktree_path == linked_worktree_path { return None; } let project_name = main_worktree_path.file_name()?.to_str()?; let directory_name = linked_worktree_path.file_name()?.to_str()?; let name = if directory_name != project_name { directory_name.to_string() } else { linked_worktree_path .parent()? .file_name()? .to_str()? .to_string() }; Some(name.into()) } fn get_permalink_in_rust_registry_src( provider_registry: Arc, path: PathBuf, selection: Range, ) -> Result { #[derive(Deserialize)] struct CargoVcsGit { sha1: String, } #[derive(Deserialize)] struct CargoVcsInfo { git: CargoVcsGit, path_in_vcs: String, } #[derive(Deserialize)] struct CargoPackage { repository: String, } #[derive(Deserialize)] struct CargoToml { package: CargoPackage, } let Some((dir, cargo_vcs_info_json)) = path.ancestors().skip(1).find_map(|dir| { let json = std::fs::read_to_string(dir.join(".cargo_vcs_info.json")).ok()?; Some((dir, json)) }) else { bail!("No .cargo_vcs_info.json found in parent directories") }; let cargo_vcs_info = serde_json::from_str::(&cargo_vcs_info_json)?; let cargo_toml = std::fs::read_to_string(dir.join("Cargo.toml"))?; let manifest = toml::from_str::(&cargo_toml)?; let (provider, remote) = parse_git_remote_url(provider_registry, &manifest.package.repository) .context("parsing package.repository field of manifest")?; let path = PathBuf::from(cargo_vcs_info.path_in_vcs).join(path.strip_prefix(dir).unwrap()); let permalink = provider.build_permalink( remote, BuildPermalinkParams::new( &cargo_vcs_info.git.sha1, &RepoPath::from_rel_path( &RelPath::new(&path, PathStyle::local()).context("invalid path")?, ), Some(selection), ), ); Ok(permalink) } fn serialize_blame_buffer_response(blame: Option) -> proto::BlameBufferResponse { let Some(blame) = blame else { return proto::BlameBufferResponse { blame_response: None, }; }; let entries = blame .entries .into_iter() .map(|entry| proto::BlameEntry { sha: entry.sha.as_bytes().into(), start_line: entry.range.start, end_line: entry.range.end, original_line_number: entry.original_line_number, author: entry.author, author_mail: entry.author_mail, author_time: entry.author_time, author_tz: entry.author_tz, committer: entry.committer_name, committer_mail: entry.committer_email, committer_time: entry.committer_time, committer_tz: entry.committer_tz, summary: entry.summary, previous: entry.previous, filename: entry.filename, }) .collect::>(); let messages = blame .messages .into_iter() .map(|(oid, message)| proto::CommitMessage { oid: oid.as_bytes().into(), message, }) .collect::>(); proto::BlameBufferResponse { blame_response: Some(proto::blame_buffer_response::BlameResponse { entries, messages }), } } fn deserialize_blame_buffer_response( response: proto::BlameBufferResponse, ) -> Option { let response = response.blame_response?; let entries = response .entries .into_iter() .filter_map(|entry| { Some(git::blame::BlameEntry { sha: git::Oid::from_bytes(&entry.sha).ok()?, range: entry.start_line..entry.end_line, original_line_number: entry.original_line_number, committer_name: entry.committer, committer_time: entry.committer_time, committer_tz: entry.committer_tz, committer_email: entry.committer_mail, author: entry.author, author_mail: entry.author_mail, author_time: entry.author_time, author_tz: entry.author_tz, summary: entry.summary, previous: entry.previous, filename: entry.filename, }) }) .collect::>(); let messages = response .messages .into_iter() .filter_map(|message| Some((git::Oid::from_bytes(&message.oid).ok()?, message.message))) .collect::>(); Some(Blame { entries, messages, tag_names: Default::default(), }) } fn log_source_to_proto(log_source: &LogSource) -> proto::GitLogSource { proto::GitLogSource { source: Some(match log_source { LogSource::All => proto::git_log_source::Source::All(proto::GitLogSourceAll {}), LogSource::Branch(branch) => proto::git_log_source::Source::Branch(branch.to_string()), LogSource::Sha(sha) => proto::git_log_source::Source::Sha(sha.to_string()), LogSource::Path(path) => { proto::git_log_source::Source::Path(path.as_unix_str().to_owned()) } }), } } fn log_source_from_proto(log_source: proto::GitLogSource) -> Result { match log_source .source .context("git log source is missing source")? { proto::git_log_source::Source::All(_) => Ok(LogSource::All), proto::git_log_source::Source::Branch(branch) => Ok(LogSource::Branch(branch.into())), proto::git_log_source::Source::Sha(sha) => Ok(LogSource::Sha(Oid::from_str(&sha)?)), proto::git_log_source::Source::Path(path) => { Ok(LogSource::Path(RepoPath::from_proto(&path)?)) } } } fn log_order_to_proto(log_order: LogOrder) -> i32 { match log_order { LogOrder::DateOrder => proto::get_initial_graph_data::LogOrder::DateOrder as i32, LogOrder::TopoOrder => proto::get_initial_graph_data::LogOrder::TopoOrder as i32, LogOrder::AuthorDateOrder => { proto::get_initial_graph_data::LogOrder::AuthorDateOrder as i32 } LogOrder::ReverseChronological => { proto::get_initial_graph_data::LogOrder::ReverseChronological as i32 } } } fn log_order_from_proto(log_order: proto::get_initial_graph_data::LogOrder) -> LogOrder { match log_order { proto::get_initial_graph_data::LogOrder::DateOrder => LogOrder::DateOrder, proto::get_initial_graph_data::LogOrder::TopoOrder => LogOrder::TopoOrder, proto::get_initial_graph_data::LogOrder::AuthorDateOrder => LogOrder::AuthorDateOrder, proto::get_initial_graph_data::LogOrder::ReverseChronological => { LogOrder::ReverseChronological } } } fn initial_graph_commit_to_proto(commit: &InitialGraphCommitData) -> proto::InitialGraphCommit { proto::InitialGraphCommit { sha: commit.sha.to_string(), parents: commit .parents .iter() .map(|parent| parent.to_string()) .collect(), ref_names: commit .ref_names .iter() .map(|ref_name| ref_name.to_string()) .collect(), } } fn initial_graph_commit_from_proto( commit: proto::InitialGraphCommit, ) -> Result> { let sha = Oid::from_str(&commit.sha)?; let mut parents = SmallVec::with_capacity(commit.parents.len()); for parent in &commit.parents { parents.push(Oid::from_str(parent)?); } Ok(Arc::new(InitialGraphCommitData { sha, parents, ref_names: commit .ref_names .into_iter() .map(SharedString::from) .collect(), })) } fn commit_data_to_proto(commit: &CommitData) -> proto::CommitData { proto::CommitData { sha: commit.sha.to_string(), parents: commit.parents.iter().map(|p| p.to_string()).collect(), author_name: commit.author_name.to_string(), author_email: commit.author_email.to_string(), commit_timestamp: commit.commit_timestamp, subject: commit.subject.to_string(), message: commit.message.to_string(), } } fn commit_data_from_proto(commit: proto::CommitData) -> Result { let sha = Oid::from_str(&commit.sha)?; let mut parents = SmallVec::with_capacity(commit.parents.len()); for parent in &commit.parents { parents.push(Oid::from_str(parent)?); } Ok(CommitData { sha, parents, author_name: SharedString::from(commit.author_name), author_email: SharedString::from(commit.author_email), commit_timestamp: commit.commit_timestamp, subject: SharedString::from(commit.subject), message: SharedString::from(commit.message), }) } fn branch_to_proto(branch: &git::repository::Branch) -> proto::Branch { proto::Branch { is_head: branch.is_head, ref_name: branch.ref_name.to_string(), unix_timestamp: branch .most_recent_commit .as_ref() .map(|commit| commit.commit_timestamp as u64), upstream: branch.upstream.as_ref().map(|upstream| proto::GitUpstream { ref_name: upstream.ref_name.to_string(), tracking: upstream .tracking .status() .map(|upstream| proto::UpstreamTracking { ahead: upstream.ahead as u64, behind: upstream.behind as u64, }), }), most_recent_commit: branch .most_recent_commit .as_ref() .map(|commit| proto::CommitSummary { sha: commit.sha.to_string(), subject: commit.subject.to_string(), commit_timestamp: commit.commit_timestamp, author_name: commit.author_name.to_string(), }), } } fn worktree_to_proto(worktree: &git::repository::Worktree) -> proto::Worktree { proto::Worktree { path: worktree.path.to_string_lossy().to_string(), ref_name: worktree .ref_name .as_ref() .map(|s| s.to_string()) .unwrap_or_default(), sha: worktree.sha.to_string(), is_main: worktree.is_main, is_bare: worktree.is_bare, } } fn proto_to_worktree(proto: &proto::Worktree) -> git::repository::Worktree { git::repository::Worktree { path: PathBuf::from(proto.path.clone()), ref_name: if proto.ref_name.is_empty() { None } else { Some(SharedString::from(&proto.ref_name)) }, sha: proto.sha.clone().into(), is_main: proto.is_main, is_bare: proto.is_bare, } } fn proto_to_branch(proto: &proto::Branch) -> git::repository::Branch { git::repository::Branch { is_head: proto.is_head, ref_name: proto.ref_name.clone().into(), upstream: proto .upstream .as_ref() .map(|upstream| git::repository::Upstream { ref_name: upstream.ref_name.to_string().into(), tracking: upstream .tracking .as_ref() .map(|tracking| { git::repository::UpstreamTracking::Tracked(UpstreamTrackingStatus { ahead: tracking.ahead as u32, behind: tracking.behind as u32, }) }) .unwrap_or(git::repository::UpstreamTracking::Gone), }), most_recent_commit: proto.most_recent_commit.as_ref().map(|commit| { git::repository::CommitSummary { sha: commit.sha.to_string().into(), subject: commit.subject.to_string().into(), commit_timestamp: commit.commit_timestamp, author_name: commit.author_name.to_string().into(), has_parent: true, } }), } } fn commit_details_to_proto(commit: &CommitDetails) -> proto::GitCommitDetails { proto::GitCommitDetails { sha: commit.sha.to_string(), message: commit.message.to_string(), commit_timestamp: commit.commit_timestamp, author_email: commit.author_email.to_string(), author_name: commit.author_name.to_string(), } } fn proto_to_commit_details(proto: &proto::GitCommitDetails) -> CommitDetails { CommitDetails { sha: proto.sha.clone().into(), message: proto.message.clone().into(), commit_timestamp: proto.commit_timestamp, author_email: proto.author_email.clone().into(), author_name: proto.author_name.clone().into(), } } async fn append_pattern_to_ignore_file( fs: Arc, file_path: PathBuf, pattern: String, ) -> Result<()> { let existing_content = match fs.load(&file_path).await { Ok(content) => content, Err(error) if error .root_cause() .downcast_ref::() .is_some_and(|error| error.kind() == std::io::ErrorKind::NotFound) => { String::new() } Err(error) => { return Err(error).with_context(|| format!("loading {}", file_path.display())); } }; if existing_content.lines().any(|line| line.trim() == pattern) { return Ok(()); } let new_content = if existing_content.is_empty() { format!("{}\n", pattern) } else if existing_content.ends_with('\n') { format!("{}{}\n", existing_content, pattern) } else { format!("{}\n{}\n", existing_content, pattern) }; fs.save( &file_path, &text::Rope::from(new_content.as_str()), text::LineEnding::Unix, ) .await } #[cfg(any(test, feature = "test-support"))] impl Repository { pub fn set_branch_list_for_test(&mut self, branches: Vec, cx: &mut Context) { self.snapshot.branch_list = branches.into(); cx.emit(RepositoryEvent::BranchListChanged); } pub fn loaded_commit_data_for_test(&self) -> HashMap { self.commit_data .iter() .filter_map(|(sha, state)| match state { CommitDataState::Loaded(data) => Some((*sha, data.as_ref().clone())), CommitDataState::Loading(_) => None, }) .collect() } } #[cfg(test)] mod tests { use super::*; use crate::Project; use fs::{FakeFs, Fs}; use git::repository::{RepoPath, repo_path}; use gpui::TestAppContext; use gpui::proptest::prelude::*; use rand::{SeedableRng, rngs::StdRng}; use serde_json::json; use settings::SettingsStore; use std::path::{Path, PathBuf}; fn init_test(cx: &mut TestAppContext) { cx.update(|cx| { let settings_store = SettingsStore::test(cx); cx.set_global(settings_store); }); } #[gpui::test] async fn test_open_uncommitted_diff_skips_symlinks(cx: &mut TestAppContext) { use util::rel_path::rel_path; init_test(cx); let fs = FakeFs::new(cx.executor()); fs.insert_tree( Path::new("/project"), json!({ ".git": {}, "target.txt": "rule one\nrule two\n", }), ) .await; fs.insert_symlink("/project/agents.md", PathBuf::from("target.txt")) .await; fs.set_head_and_index_for_repo( Path::new("/project/.git"), &[ // git stores the symlink's target path as the blob for `agents.md` ("agents.md", "target.txt".into()), ("target.txt", "rule one\n".into()), ], ); let project = Project::test(fs.clone(), [Path::new("/project")], cx).await; project .update(cx, |project, cx| project.git_scans_complete(cx)) .await; let worktree_id = project.read_with(cx, |project, cx| { project.worktrees(cx).next().unwrap().read(cx).id() }); // symlink file should not produce a base diff let symlink_buffer = project .update(cx, |project, cx| { project.open_buffer((worktree_id, rel_path("agents.md")), cx) }) .await .unwrap(); let symlink_diff = project .update(cx, |project, cx| { project.open_uncommitted_diff(symlink_buffer, cx) }) .await .unwrap(); symlink_diff.read_with(cx, |diff, _| { assert!( !diff.base_text_exists(), "symlinked buffer should not have a git diff base" ); }); // regular file should still produce a base diff let regular_buffer = project .update(cx, |project, cx| { project.open_buffer((worktree_id, rel_path("target.txt")), cx) }) .await .unwrap(); let regular_diff = project .update(cx, |project, cx| { project.open_uncommitted_diff(regular_buffer, cx) }) .await .unwrap(); regular_diff.read_with(cx, |diff, _| { assert!( diff.base_text_exists(), "regular file should have a git diff base" ); }); } #[gpui::test] async fn test_append_pattern_to_ignore_file_creates_and_deduplicates(cx: &mut TestAppContext) { let fs: Arc = FakeFs::new(cx.executor()); let path = PathBuf::from("/root/.gitignore"); // Appending to a non-existent file creates it with a trailing newline. super::append_pattern_to_ignore_file(fs.clone(), path.clone(), "build/".to_string()) .await .unwrap(); assert_eq!(fs.load(&path).await.unwrap(), "build/\n"); // Appending the same pattern again is a no-op (deduplication). super::append_pattern_to_ignore_file(fs.clone(), path.clone(), "build/".to_string()) .await .unwrap(); assert_eq!(fs.load(&path).await.unwrap(), "build/\n"); // Appending a distinct pattern adds it with a trailing newline. super::append_pattern_to_ignore_file(fs.clone(), path.clone(), "target/".to_string()) .await .unwrap(); assert_eq!(fs.load(&path).await.unwrap(), "build/\ntarget/\n"); } #[gpui::test] async fn test_append_pattern_adds_newline_before_pattern_when_missing(cx: &mut TestAppContext) { let fs: Arc = FakeFs::new(cx.executor()); let path = PathBuf::from("/root/.gitignore"); // Pre-populate the file without a trailing newline. fs.save(&path, &text::Rope::from("*.log"), text::LineEnding::Unix) .await .unwrap(); // The new pattern must be written on its own line. super::append_pattern_to_ignore_file(fs.clone(), path.clone(), "build/".to_string()) .await .unwrap(); assert_eq!(fs.load(&path).await.unwrap(), "*.log\nbuild/\n"); } #[test] fn test_new_worktree_path_uses_posix_style_for_remote_paths() { let work_dir = Path::new("/home/user/dev/lsp-tests"); let directory = worktrees_directory_for_repo(work_dir, "../worktrees", PathStyle::Unix).unwrap(); let directory = PathStyle::Unix.join_path(&directory, "nimble-sky").unwrap(); let path = PathStyle::Unix.join_path(&directory, "lsp-tests").unwrap(); assert_eq!( path, PathBuf::from("/home/user/dev/worktrees/lsp-tests/nimble-sky/lsp-tests") ); } fn verify_invariants(repository: &Repository) -> anyhow::Result<()> { match &repository.commit_data_handler { CommitDataHandlerState::Open(handler) => { verify_loading_entries_are_pending(repository, handler)?; verify_await_result_loading_entries_have_completion_senders(repository, handler)?; verify_pending_requests_are_loading(repository, handler)?; verify_completion_senders_are_await_result_loading(repository, handler)?; verify_completion_senders_are_pending(handler)?; verify_non_await_result_loading_entries_have_no_completion_sender( repository, handler, )?; verify_loaded_entries_are_not_pending(repository, handler)?; verify_loaded_entries_have_no_completion_sender(repository, handler)?; } CommitDataHandlerState::Closed => { verify_closed_handler_invariants(repository)?; } } Ok(()) } fn verify_loading_entries_are_pending( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for (sha, state) in &repository.commit_data { if matches!(state, CommitDataState::Loading(_)) { anyhow::ensure!( handler.pending_requests.contains(sha), "loading commit data for {sha} must be tracked in pending_requests" ); } } Ok(()) } fn verify_await_result_loading_entries_have_completion_senders( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for (sha, state) in &repository.commit_data { if matches!(state, CommitDataState::Loading(Some(_))) { anyhow::ensure!( handler.completion_senders.contains_key(sha), "await-result loading commit data for {sha} must have a completion sender" ); } } Ok(()) } fn verify_pending_requests_are_loading( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for sha in &handler.pending_requests { anyhow::ensure!( matches!( repository.commit_data.get(sha), Some(CommitDataState::Loading(_)) ), "pending request for {sha} must correspond to loading commit data" ); } Ok(()) } fn verify_completion_senders_are_await_result_loading( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for sha in handler.completion_senders.keys() { anyhow::ensure!( matches!( repository.commit_data.get(sha), Some(CommitDataState::Loading(Some(_))) ), "completion sender for {sha} must correspond to await-result loading commit data" ); } Ok(()) } fn verify_completion_senders_are_pending(handler: &CommitDataHandler) -> anyhow::Result<()> { for sha in handler.completion_senders.keys() { anyhow::ensure!( handler.pending_requests.contains(sha), "completion sender for {sha} must also be tracked as pending" ); } Ok(()) } fn verify_non_await_result_loading_entries_have_no_completion_sender( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for (sha, state) in &repository.commit_data { if matches!(state, CommitDataState::Loading(None)) { anyhow::ensure!( !handler.completion_senders.contains_key(sha), "non-await-result loading commit data for {sha} must not have a completion sender" ); } } Ok(()) } fn verify_loaded_entries_are_not_pending( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for (sha, state) in &repository.commit_data { if matches!(state, CommitDataState::Loaded(_)) { anyhow::ensure!( !handler.pending_requests.contains(sha), "loaded commit data for {sha} must not still be pending" ); } } Ok(()) } fn verify_loaded_entries_have_no_completion_sender( repository: &Repository, handler: &CommitDataHandler, ) -> anyhow::Result<()> { for (sha, state) in &repository.commit_data { if matches!(state, CommitDataState::Loaded(_)) { anyhow::ensure!( !handler.completion_senders.contains_key(sha), "loaded commit data for {sha} must not keep a completion sender" ); } } Ok(()) } fn verify_closed_handler_invariants(repository: &Repository) -> anyhow::Result<()> { for (sha, state) in &repository.commit_data { anyhow::ensure!( !matches!(state, CommitDataState::Loading(_)), "closed handler must not keep loading commit data for {sha}" ); } Ok(()) } #[gpui::property_test(config = ProptestConfig { cases: 20, ..Default::default() })] async fn test_commit_data_random_invariants( #[strategy = any::()] seed: u64, #[strategy = gpui::proptest::collection::vec(0usize..2000, 1..200)] commit_indexes: Vec< usize, >, #[strategy = gpui::proptest::collection::vec(any::(), 1..200)] await_results: Vec< bool, >, #[strategy = gpui::proptest::collection::vec(0usize..2000, 0..200)] failing_commit_indexes: Vec< usize, >, #[strategy = gpui::proptest::collection::vec(0usize..2000, 0..200)] missing_commit_indexes: Vec< usize, >, cx: &mut TestAppContext, ) { init_test(cx); let mut rng = StdRng::seed_from_u64(seed); let commit_shas = (0..2000).map(|_| Oid::random(&mut rng)).collect::>(); let failing_shas = failing_commit_indexes .into_iter() .map(|index| commit_shas[index % commit_shas.len()]) .collect::>(); let missing_shas = missing_commit_indexes .into_iter() .map(|index| commit_shas[index % commit_shas.len()]) .collect::>(); let commit_data = commit_shas .iter() .filter(|sha| !missing_shas.contains(sha)) .map(|sha| { ( CommitData { sha: *sha, parents: SmallVec::new(), author_name: SharedString::from(format!("Author {sha}")), author_email: SharedString::from(format!("{sha}@example.com")), commit_timestamp: rng.random_range(0..10_000), subject: SharedString::from(format!("Subject {sha}")), message: SharedString::from(format!("Subject {sha}\n\nBody for {sha}")), }, failing_shas.contains(sha), ) }) .collect::>(); let expected_loaded_shas = commit_indexes .iter() .map(|index| commit_shas[index % commit_shas.len()]) .filter(|sha| !failing_shas.contains(sha) && !missing_shas.contains(sha)) .collect::>(); let fs = FakeFs::new(cx.executor()); fs.insert_tree( Path::new("/project"), json!({ ".git": {}, "file.txt": "content", }), ) .await; fs.set_commit_data(Path::new("/project/.git"), commit_data); let project = Project::test(fs.clone(), [Path::new("/project")], cx).await; project .update(cx, |project, cx| project.git_scans_complete(cx)) .await; let repository = project.read_with(cx, |project, cx| { project .active_repository(cx) .expect("should have a repository") }); cx.update(|cx| { cx.observe(&repository, |repo, cx| { verify_invariants(repo.read(cx)) .context("Invariant weren't held after a cx.notify") .unwrap(); }) }) .detach(); let mut next_step = 0; while next_step < commit_indexes.len() { let remaining_steps = commit_indexes.len() - next_step; let chunk_size = rng.random_range(1..=remaining_steps.min(16)); let chunk_end = next_step + chunk_size; for step in next_step..chunk_end { let sha = commit_shas[commit_indexes[step] % commit_shas.len()]; let await_result = await_results[step % await_results.len()]; repository.update(cx, |repository, cx| { repository.fetch_commit_data(sha, await_result, cx); verify_invariants(repository) .with_context(|| { format!( "commit data invariant violation after step {} for sha {}", step + 1, sha, ) }) .unwrap(); }); } cx.run_until_parked(); repository.read_with(cx, |repository, _cx| { verify_invariants(repository) .with_context(|| { format!( "commit data invariant violation after draining through step {}", chunk_end, ) }) .unwrap(); }); next_step = chunk_end; } cx.run_until_parked(); repository.read_with(cx, |repository, _cx| { verify_invariants(repository) .with_context(|| "commit data invariant violation after final drain".to_string()) .unwrap(); let loaded_shas = repository .commit_data .iter() .filter_map(|(sha, state)| match state { CommitDataState::Loaded(_) => Some(*sha), CommitDataState::Loading(_) => None, }) .collect::>(); let missing_loaded_shas = expected_loaded_shas .difference(&loaded_shas) .copied() .collect::>(); let unexpected_loaded_shas = loaded_shas .difference(&expected_loaded_shas) .copied() .collect::>(); assert!( missing_loaded_shas.is_empty() && unexpected_loaded_shas.is_empty(), "loaded commit data SHAs after final drain did not match expectation. missing: {:?}, unexpected: {:?}", missing_loaded_shas, unexpected_loaded_shas, ); }); } fn repo_paths(paths: &[&str]) -> Vec { paths.iter().map(repo_path).collect() } #[test] fn coalesce_repo_paths_keeps_root_only() { let coalesced = GitStore::coalesce_repo_paths(repo_paths(&["", "src", "src/lib.rs"])); assert_eq!(coalesced, repo_paths(&[""])); } #[test] fn coalesce_repo_paths_keeps_existing_ancestors() { let coalesced = GitStore::coalesce_repo_paths(repo_paths(&[ "src", "src/lib.rs", "src/nested/file.rs", "tests/test.rs", ])); assert_eq!(coalesced, repo_paths(&["src", "tests/test.rs"])); } #[test] fn coalesce_repo_paths_does_not_invent_missing_parents() { let coalesced = GitStore::coalesce_repo_paths(repo_paths(&[ "submodule/a.txt", "submodule/nested/b.txt", "top_level.rs", ])); assert_eq!( coalesced, repo_paths(&["submodule/a.txt", "submodule/nested/b.txt", "top_level.rs"]) ); } } /// This snapshot computes the repository state on the foreground thread while /// running the git commands on the background thread. We update branch, head, /// remotes, and worktrees first so the UI can react sooner, then compute file /// state and emit those events immediately after. async fn compute_snapshot( this: Entity, backend: Arc, cx: &mut AsyncApp, ) -> RepositorySnapshot { log::debug!("starting compute snapshot"); let (id, work_directory_abs_path, prev_snapshot) = this.update(cx, |this, _| { this.paths_needing_status_update.clear(); ( this.id, this.work_directory_abs_path.clone(), this.snapshot.clone(), ) }); let branches_future = { let backend = backend.clone(); async move { backend.branches().await.log_err().unwrap_or_default() } }; let head_commit_future = { let backend = backend.clone(); async move { backend.show("HEAD".to_string()).await.ok() } }; let worktrees_future = { let backend = backend.clone(); async move { backend.worktrees().await.log_err().unwrap_or_default() } }; let (branches, head_commit, all_worktrees) = futures::future::join3(branches_future, head_commit_future, worktrees_future).await; log::debug!("fetched branches, head commit, worktrees"); let BranchesScanResult { branches, error: branch_list_error, } = branches; let branch = branches.iter().find(|branch| branch.is_head).cloned(); let branch_list: Arc<[Branch]> = branches.into(); let linked_worktrees: Arc<[GitWorktree]> = all_worktrees .into_iter() .filter(|wt| wt.path != *work_directory_abs_path) .collect(); let mut remote_urls = backend.remote_urls().await; let remote_origin_url = remote_urls.remove("origin"); let remote_upstream_url = remote_urls.remove("upstream"); log::debug!("fetched remotes"); let snapshot = this.update(cx, |this, cx| { let head_changed = branch != this.snapshot.branch || head_commit != this.snapshot.head_commit; let branch_list_changed = *branch_list != *this.snapshot.branch_list; let branch_list_error_changed = branch_list_error != this.snapshot.branch_list_error; let worktrees_changed = *linked_worktrees != *this.snapshot.linked_worktrees; this.snapshot = RepositorySnapshot { id, work_directory_abs_path, branch, branch_list: branch_list.clone(), branch_list_error, head_commit, remote_origin_url, remote_upstream_url, linked_worktrees, scan_id: prev_snapshot.scan_id + 1, ..prev_snapshot }; if head_changed { cx.emit(RepositoryEvent::HeadChanged); } if branch_list_changed || branch_list_error_changed { cx.emit(RepositoryEvent::BranchListChanged); } if worktrees_changed { cx.emit(RepositoryEvent::GitWorktreeListChanged); } this.snapshot.clone() }); let statuses_future = { let backend = backend.clone(); async move { backend .status(&[RepoPath::from_rel_path( &RelPath::new(".".as_ref(), PathStyle::local()).unwrap(), )]) .await .log_err() .unwrap_or_default() } }; let diff_stats_future = { let snapshot = snapshot.clone(); let backend = backend.clone(); async move { if snapshot.head_commit.is_some() { futures::future::join3( backend.diff_stat(DiffStatType::HeadToWorktree, &[]), backend.diff_stat(DiffStatType::HeadToIndex, &[]), backend.diff_stat(DiffStatType::IndexToWorktree, &[]), ) .await } else { ( Ok(status::GitDiffStat::default()), Ok(status::GitDiffStat::default()), Ok(status::GitDiffStat::default()), ) } } }; let stash_entries_future = { let backend = backend.clone(); async move { backend.stash_entries().await.log_err().unwrap_or_default() } }; let (statuses, diff_stats, stash_entries) = futures::future::join3(statuses_future, diff_stats_future, stash_entries_future).await; let (diff_stats, staged_diff_stats, unstaged_diff_stats) = diff_stats; let diff_stats = diff_stats.log_err().unwrap_or_default(); let staged_diff_stats = staged_diff_stats.log_err().unwrap_or_default(); let unstaged_diff_stats = unstaged_diff_stats.log_err().unwrap_or_default(); log::debug!("fetched statuses, diff stats, stash entries"); let diff_stat_map: HashMap<&RepoPath, DiffStat> = diff_stats.entries.iter().map(|(p, s)| (p, *s)).collect(); let staged_diff_stat_map: HashMap<&RepoPath, DiffStat> = staged_diff_stats .entries .iter() .map(|(p, s)| (p, *s)) .collect(); let unstaged_diff_stat_map: HashMap<&RepoPath, DiffStat> = unstaged_diff_stats .entries .iter() .map(|(p, s)| (p, *s)) .collect(); let mut conflicted_paths = Vec::new(); let statuses_by_path = SumTree::from_iter( statuses.entries.iter().map(|(repo_path, status)| { if status.is_conflicted() { conflicted_paths.push(repo_path.clone()); } StatusEntry { repo_path: repo_path.clone(), status: *status, diff_stat: diff_stat_map.get(repo_path).copied(), staged_diff_stat: staged_diff_stat_map.get(repo_path).copied(), unstaged_diff_stat: unstaged_diff_stat_map.get(repo_path).copied(), } }), (), ); let (merge_details, conflicts_changed) = cx .background_spawn({ let backend = backend.clone(); let mut merge_details = snapshot.merge.clone(); async move { let conflicts_changed = merge_details.update(&backend, conflicted_paths).await; (merge_details, conflicts_changed) } }) .await; log::debug!("new merge details: {merge_details:?}"); this.update(cx, |this, cx| { if conflicts_changed || statuses_by_path != this.snapshot.statuses_by_path { cx.emit(RepositoryEvent::StatusesChanged); } if stash_entries != this.snapshot.stash_entries { cx.emit(RepositoryEvent::StashEntriesChanged); } this.snapshot.scan_id += 1; this.snapshot.merge = merge_details; this.snapshot.statuses_by_path = statuses_by_path; this.snapshot.stash_entries = stash_entries; this.snapshot.clone() }) } fn status_from_proto( simple_status: i32, status: Option, ) -> anyhow::Result { use proto::git_file_status::Variant; let Some(variant) = status.and_then(|status| status.variant) else { let code = proto::GitStatus::from_i32(simple_status) .with_context(|| format!("Invalid git status code: {simple_status}"))?; let result = match code { proto::GitStatus::Added => TrackedStatus { worktree_status: StatusCode::Added, index_status: StatusCode::Unmodified, } .into(), proto::GitStatus::Modified => TrackedStatus { worktree_status: StatusCode::Modified, index_status: StatusCode::Unmodified, } .into(), proto::GitStatus::Conflict => UnmergedStatus { first_head: UnmergedStatusCode::Updated, second_head: UnmergedStatusCode::Updated, } .into(), proto::GitStatus::Deleted => TrackedStatus { worktree_status: StatusCode::Deleted, index_status: StatusCode::Unmodified, } .into(), _ => anyhow::bail!("Invalid code for simple status: {simple_status}"), }; return Ok(result); }; let result = match variant { Variant::Untracked(_) => FileStatus::Untracked, Variant::Ignored(_) => FileStatus::Ignored, Variant::Unmerged(unmerged) => { let [first_head, second_head] = [unmerged.first_head, unmerged.second_head].map(|head| { let code = proto::GitStatus::from_i32(head) .with_context(|| format!("Invalid git status code: {head}"))?; let result = match code { proto::GitStatus::Added => UnmergedStatusCode::Added, proto::GitStatus::Updated => UnmergedStatusCode::Updated, proto::GitStatus::Deleted => UnmergedStatusCode::Deleted, _ => anyhow::bail!("Invalid code for unmerged status: {code:?}"), }; Ok(result) }); let [first_head, second_head] = [first_head?, second_head?]; UnmergedStatus { first_head, second_head, } .into() } Variant::Tracked(tracked) => { let [index_status, worktree_status] = [tracked.index_status, tracked.worktree_status] .map(|status| { let code = proto::GitStatus::from_i32(status) .with_context(|| format!("Invalid git status code: {status}"))?; let result = match code { proto::GitStatus::Modified => StatusCode::Modified, proto::GitStatus::TypeChanged => StatusCode::TypeChanged, proto::GitStatus::Added => StatusCode::Added, proto::GitStatus::Deleted => StatusCode::Deleted, proto::GitStatus::Renamed => StatusCode::Renamed, proto::GitStatus::Copied => StatusCode::Copied, proto::GitStatus::Unmodified => StatusCode::Unmodified, _ => anyhow::bail!("Invalid code for tracked status: {code:?}"), }; Ok(result) }); let [index_status, worktree_status] = [index_status?, worktree_status?]; TrackedStatus { index_status, worktree_status, } .into() } }; Ok(result) } fn status_to_proto(status: FileStatus) -> proto::GitFileStatus { use proto::git_file_status::{Tracked, Unmerged, Variant}; let variant = match status { FileStatus::Untracked => Variant::Untracked(Default::default()), FileStatus::Ignored => Variant::Ignored(Default::default()), FileStatus::Unmerged(UnmergedStatus { first_head, second_head, }) => Variant::Unmerged(Unmerged { first_head: unmerged_status_to_proto(first_head), second_head: unmerged_status_to_proto(second_head), }), FileStatus::Tracked(TrackedStatus { index_status, worktree_status, }) => Variant::Tracked(Tracked { index_status: tracked_status_to_proto(index_status), worktree_status: tracked_status_to_proto(worktree_status), }), }; proto::GitFileStatus { variant: Some(variant), } } fn unmerged_status_to_proto(code: UnmergedStatusCode) -> i32 { match code { UnmergedStatusCode::Added => proto::GitStatus::Added as _, UnmergedStatusCode::Deleted => proto::GitStatus::Deleted as _, UnmergedStatusCode::Updated => proto::GitStatus::Updated as _, } } fn tracked_status_to_proto(code: StatusCode) -> i32 { match code { StatusCode::Added => proto::GitStatus::Added as _, StatusCode::Deleted => proto::GitStatus::Deleted as _, StatusCode::Modified => proto::GitStatus::Modified as _, StatusCode::Renamed => proto::GitStatus::Renamed as _, StatusCode::TypeChanged => proto::GitStatus::TypeChanged as _, StatusCode::Copied => proto::GitStatus::Copied as _, StatusCode::Unmodified => proto::GitStatus::Unmodified as _, } }