worktree_store.rs

  1use std::{
  2    cell::RefCell,
  3    path::{Path, PathBuf},
  4    sync::{atomic::AtomicUsize, Arc},
  5};
  6
  7use anyhow::{anyhow, Context as _, Result};
  8use client::DevServerProjectId;
  9use collections::{HashMap, HashSet};
 10use fs::Fs;
 11use futures::{
 12    future::{BoxFuture, Shared},
 13    FutureExt, SinkExt,
 14};
 15use gpui::{
 16    AppContext, AsyncAppContext, EntityId, EventEmitter, Model, ModelContext, Task, WeakModel,
 17};
 18use postage::oneshot;
 19use rpc::{
 20    proto::{self, SSH_PROJECT_ID},
 21    AnyProtoClient, ErrorExt, TypedEnvelope,
 22};
 23use smol::{
 24    channel::{Receiver, Sender},
 25    stream::StreamExt,
 26};
 27use text::ReplicaId;
 28use util::{paths::compare_paths, ResultExt};
 29use worktree::{Entry, ProjectEntryId, Worktree, WorktreeId, WorktreeSettings};
 30
 31use crate::{search::SearchQuery, ProjectPath};
 32
 33struct MatchingEntry {
 34    worktree_path: Arc<Path>,
 35    path: ProjectPath,
 36    respond: oneshot::Sender<ProjectPath>,
 37}
 38
 39enum WorktreeStoreState {
 40    Local {
 41        fs: Arc<dyn Fs>,
 42    },
 43    Remote {
 44        dev_server_project_id: Option<DevServerProjectId>,
 45        upstream_client: AnyProtoClient,
 46        upstream_project_id: u64,
 47    },
 48}
 49
 50pub struct WorktreeStore {
 51    next_entry_id: Arc<AtomicUsize>,
 52    downstream_client: Option<(AnyProtoClient, u64)>,
 53    retain_worktrees: bool,
 54    worktrees: Vec<WorktreeHandle>,
 55    worktrees_reordered: bool,
 56    #[allow(clippy::type_complexity)]
 57    loading_worktrees:
 58        HashMap<Arc<Path>, Shared<Task<Result<Model<Worktree>, Arc<anyhow::Error>>>>>,
 59    state: WorktreeStoreState,
 60}
 61
 62pub enum WorktreeStoreEvent {
 63    WorktreeAdded(Model<Worktree>),
 64    WorktreeRemoved(EntityId, WorktreeId),
 65    WorktreeOrderChanged,
 66    WorktreeUpdateSent(Model<Worktree>),
 67}
 68
 69impl EventEmitter<WorktreeStoreEvent> for WorktreeStore {}
 70
 71impl WorktreeStore {
 72    pub fn init(client: &AnyProtoClient) {
 73        client.add_model_request_handler(Self::handle_create_project_entry);
 74        client.add_model_request_handler(Self::handle_rename_project_entry);
 75        client.add_model_request_handler(Self::handle_copy_project_entry);
 76        client.add_model_request_handler(Self::handle_delete_project_entry);
 77        client.add_model_request_handler(Self::handle_expand_project_entry);
 78    }
 79
 80    pub fn local(retain_worktrees: bool, fs: Arc<dyn Fs>) -> Self {
 81        Self {
 82            next_entry_id: Default::default(),
 83            loading_worktrees: Default::default(),
 84            downstream_client: None,
 85            worktrees: Vec::new(),
 86            worktrees_reordered: false,
 87            retain_worktrees,
 88            state: WorktreeStoreState::Local { fs },
 89        }
 90    }
 91
 92    pub fn remote(
 93        retain_worktrees: bool,
 94        upstream_client: AnyProtoClient,
 95        upstream_project_id: u64,
 96        dev_server_project_id: Option<DevServerProjectId>,
 97    ) -> Self {
 98        Self {
 99            next_entry_id: Default::default(),
100            loading_worktrees: Default::default(),
101            downstream_client: None,
102            worktrees: Vec::new(),
103            worktrees_reordered: false,
104            retain_worktrees,
105            state: WorktreeStoreState::Remote {
106                upstream_client,
107                upstream_project_id,
108                dev_server_project_id,
109            },
110        }
111    }
112
113    /// Iterates through all worktrees, including ones that don't appear in the project panel
114    pub fn worktrees(&self) -> impl '_ + DoubleEndedIterator<Item = Model<Worktree>> {
115        self.worktrees
116            .iter()
117            .filter_map(move |worktree| worktree.upgrade())
118    }
119
120    /// Iterates through all user-visible worktrees, the ones that appear in the project panel.
121    pub fn visible_worktrees<'a>(
122        &'a self,
123        cx: &'a AppContext,
124    ) -> impl 'a + DoubleEndedIterator<Item = Model<Worktree>> {
125        self.worktrees()
126            .filter(|worktree| worktree.read(cx).is_visible())
127    }
128
129    pub fn worktree_for_id(&self, id: WorktreeId, cx: &AppContext) -> Option<Model<Worktree>> {
130        self.worktrees()
131            .find(|worktree| worktree.read(cx).id() == id)
132    }
133
134    pub fn worktree_for_entry(
135        &self,
136        entry_id: ProjectEntryId,
137        cx: &AppContext,
138    ) -> Option<Model<Worktree>> {
139        self.worktrees()
140            .find(|worktree| worktree.read(cx).contains_entry(entry_id))
141    }
142
143    pub fn find_worktree(
144        &self,
145        abs_path: &Path,
146        cx: &AppContext,
147    ) -> Option<(Model<Worktree>, PathBuf)> {
148        for tree in self.worktrees() {
149            if let Ok(relative_path) = abs_path.strip_prefix(tree.read(cx).abs_path()) {
150                return Some((tree.clone(), relative_path.into()));
151            }
152        }
153        None
154    }
155
156    pub fn entry_for_id<'a>(
157        &'a self,
158        entry_id: ProjectEntryId,
159        cx: &'a AppContext,
160    ) -> Option<&'a Entry> {
161        self.worktrees()
162            .find_map(|worktree| worktree.read(cx).entry_for_id(entry_id))
163    }
164
165    pub fn entry_for_path(&self, path: &ProjectPath, cx: &AppContext) -> Option<Entry> {
166        self.worktree_for_id(path.worktree_id, cx)?
167            .read(cx)
168            .entry_for_path(&path.path)
169            .cloned()
170    }
171
172    pub fn create_worktree(
173        &mut self,
174        abs_path: impl AsRef<Path>,
175        visible: bool,
176        cx: &mut ModelContext<Self>,
177    ) -> Task<Result<Model<Worktree>>> {
178        let path: Arc<Path> = abs_path.as_ref().into();
179        if !self.loading_worktrees.contains_key(&path) {
180            let task = match &self.state {
181                WorktreeStoreState::Remote {
182                    upstream_client,
183                    dev_server_project_id,
184                    ..
185                } => {
186                    if let Some(dev_server_project_id) = dev_server_project_id {
187                        self.create_dev_server_worktree(
188                            upstream_client.clone(),
189                            *dev_server_project_id,
190                            abs_path,
191                            cx,
192                        )
193                    } else if upstream_client.is_via_collab() {
194                        Task::ready(Err(Arc::new(anyhow!("cannot create worktrees via collab"))))
195                    } else {
196                        self.create_ssh_worktree(upstream_client.clone(), abs_path, visible, cx)
197                    }
198                }
199                WorktreeStoreState::Local { fs } => {
200                    self.create_local_worktree(fs.clone(), abs_path, visible, cx)
201                }
202            };
203
204            self.loading_worktrees.insert(path.clone(), task.shared());
205        }
206        let task = self.loading_worktrees.get(&path).unwrap().clone();
207        cx.spawn(|this, mut cx| async move {
208            let result = task.await;
209            this.update(&mut cx, |this, _| this.loading_worktrees.remove(&path))
210                .ok();
211            match result {
212                Ok(worktree) => Ok(worktree),
213                Err(err) => Err((*err).cloned()),
214            }
215        })
216    }
217
218    fn create_ssh_worktree(
219        &mut self,
220        client: AnyProtoClient,
221        abs_path: impl AsRef<Path>,
222        visible: bool,
223        cx: &mut ModelContext<Self>,
224    ) -> Task<Result<Model<Worktree>, Arc<anyhow::Error>>> {
225        let path_key: Arc<Path> = abs_path.as_ref().into();
226        let mut abs_path = path_key.clone().to_string_lossy().to_string();
227        // If we start with `/~` that means the ssh path was something like `ssh://user@host/~/home-dir-folder/`
228        // in which case want to strip the leading the `/`.
229        // On the host-side, the `~` will get expanded.
230        // That's what git does too: https://github.com/libgit2/libgit2/issues/3345#issuecomment-127050850
231        if abs_path.starts_with("/~") {
232            abs_path = abs_path[1..].to_string();
233        }
234        if abs_path.is_empty() {
235            abs_path = "~/".to_string();
236        }
237        let root_name = PathBuf::from(abs_path.clone())
238            .file_name()
239            .unwrap()
240            .to_string_lossy()
241            .to_string();
242        cx.spawn(|this, mut cx| async move {
243            let this = this.upgrade().context("Dropped worktree store")?;
244
245            let response = client
246                .request(proto::AddWorktree {
247                    project_id: SSH_PROJECT_ID,
248                    path: abs_path.clone(),
249                    visible,
250                })
251                .await?;
252
253            if let Some(existing_worktree) = this.read_with(&cx, |this, cx| {
254                this.worktree_for_id(WorktreeId::from_proto(response.worktree_id), cx)
255            })? {
256                return Ok(existing_worktree);
257            }
258
259            let worktree = cx.update(|cx| {
260                Worktree::remote(
261                    SSH_PROJECT_ID,
262                    0,
263                    proto::WorktreeMetadata {
264                        id: response.worktree_id,
265                        root_name,
266                        visible,
267                        abs_path,
268                    },
269                    client,
270                    cx,
271                )
272            })?;
273
274            this.update(&mut cx, |this, cx| {
275                this.add(&worktree, cx);
276            })?;
277            Ok(worktree)
278        })
279    }
280
281    fn create_local_worktree(
282        &mut self,
283        fs: Arc<dyn Fs>,
284        abs_path: impl AsRef<Path>,
285        visible: bool,
286        cx: &mut ModelContext<Self>,
287    ) -> Task<Result<Model<Worktree>, Arc<anyhow::Error>>> {
288        let next_entry_id = self.next_entry_id.clone();
289        let path: Arc<Path> = abs_path.as_ref().into();
290
291        cx.spawn(move |this, mut cx| async move {
292            let worktree = Worktree::local(path.clone(), visible, fs, next_entry_id, &mut cx).await;
293
294            let worktree = worktree?;
295            this.update(&mut cx, |this, cx| this.add(&worktree, cx))?;
296
297            if visible {
298                cx.update(|cx| {
299                    cx.add_recent_document(&path);
300                })
301                .log_err();
302            }
303
304            Ok(worktree)
305        })
306    }
307
308    fn create_dev_server_worktree(
309        &mut self,
310        client: AnyProtoClient,
311        dev_server_project_id: DevServerProjectId,
312        abs_path: impl AsRef<Path>,
313        cx: &mut ModelContext<Self>,
314    ) -> Task<Result<Model<Worktree>, Arc<anyhow::Error>>> {
315        let path: Arc<Path> = abs_path.as_ref().into();
316        let mut paths: Vec<String> = self
317            .visible_worktrees(cx)
318            .map(|worktree| worktree.read(cx).abs_path().to_string_lossy().to_string())
319            .collect();
320        paths.push(path.to_string_lossy().to_string());
321        let request = client.request(proto::UpdateDevServerProject {
322            dev_server_project_id: dev_server_project_id.0,
323            paths,
324        });
325
326        let abs_path = abs_path.as_ref().to_path_buf();
327        cx.spawn(move |project, cx| async move {
328            let (tx, rx) = futures::channel::oneshot::channel();
329            let tx = RefCell::new(Some(tx));
330            let Some(project) = project.upgrade() else {
331                return Err(anyhow!("project dropped"))?;
332            };
333            let observer = cx.update(|cx| {
334                cx.observe(&project, move |project, cx| {
335                    let abs_path = abs_path.clone();
336                    project.update(cx, |project, cx| {
337                        if let Some((worktree, _)) = project.find_worktree(&abs_path, cx) {
338                            if let Some(tx) = tx.borrow_mut().take() {
339                                tx.send(worktree).ok();
340                            }
341                        }
342                    })
343                })
344            })?;
345
346            request.await?;
347            let worktree = rx.await.map_err(|e| anyhow!(e))?;
348            drop(observer);
349            Ok(worktree)
350        })
351    }
352
353    pub fn add(&mut self, worktree: &Model<Worktree>, cx: &mut ModelContext<Self>) {
354        let worktree_id = worktree.read(cx).id();
355        debug_assert!(self.worktrees().all(|w| w.read(cx).id() != worktree_id));
356
357        let push_strong_handle = self.retain_worktrees || worktree.read(cx).is_visible();
358        let handle = if push_strong_handle {
359            WorktreeHandle::Strong(worktree.clone())
360        } else {
361            WorktreeHandle::Weak(worktree.downgrade())
362        };
363        if self.worktrees_reordered {
364            self.worktrees.push(handle);
365        } else {
366            let i = match self
367                .worktrees
368                .binary_search_by_key(&Some(worktree.read(cx).abs_path()), |other| {
369                    other.upgrade().map(|worktree| worktree.read(cx).abs_path())
370                }) {
371                Ok(i) | Err(i) => i,
372            };
373            self.worktrees.insert(i, handle);
374        }
375
376        cx.emit(WorktreeStoreEvent::WorktreeAdded(worktree.clone()));
377        self.send_project_updates(cx);
378
379        let handle_id = worktree.entity_id();
380        cx.observe_release(worktree, move |this, worktree, cx| {
381            cx.emit(WorktreeStoreEvent::WorktreeRemoved(
382                handle_id,
383                worktree.id(),
384            ));
385            this.send_project_updates(cx);
386        })
387        .detach();
388    }
389
390    pub fn remove_worktree(&mut self, id_to_remove: WorktreeId, cx: &mut ModelContext<Self>) {
391        self.worktrees.retain(|worktree| {
392            if let Some(worktree) = worktree.upgrade() {
393                if worktree.read(cx).id() == id_to_remove {
394                    cx.emit(WorktreeStoreEvent::WorktreeRemoved(
395                        worktree.entity_id(),
396                        id_to_remove,
397                    ));
398                    false
399                } else {
400                    true
401                }
402            } else {
403                false
404            }
405        });
406        self.send_project_updates(cx);
407    }
408
409    pub fn set_worktrees_reordered(&mut self, worktrees_reordered: bool) {
410        self.worktrees_reordered = worktrees_reordered;
411    }
412
413    fn upstream_client(&self) -> Option<(AnyProtoClient, u64)> {
414        match &self.state {
415            WorktreeStoreState::Remote {
416                upstream_client,
417                upstream_project_id,
418                ..
419            } => Some((upstream_client.clone(), *upstream_project_id)),
420            WorktreeStoreState::Local { .. } => None,
421        }
422    }
423
424    pub fn set_worktrees_from_proto(
425        &mut self,
426        worktrees: Vec<proto::WorktreeMetadata>,
427        replica_id: ReplicaId,
428        cx: &mut ModelContext<Self>,
429    ) -> Result<()> {
430        let mut old_worktrees_by_id = self
431            .worktrees
432            .drain(..)
433            .filter_map(|worktree| {
434                let worktree = worktree.upgrade()?;
435                Some((worktree.read(cx).id(), worktree))
436            })
437            .collect::<HashMap<_, _>>();
438
439        let (client, project_id) = self
440            .upstream_client()
441            .clone()
442            .ok_or_else(|| anyhow!("invalid project"))?;
443
444        for worktree in worktrees {
445            if let Some(old_worktree) =
446                old_worktrees_by_id.remove(&WorktreeId::from_proto(worktree.id))
447            {
448                let push_strong_handle =
449                    self.retain_worktrees || old_worktree.read(cx).is_visible();
450                let handle = if push_strong_handle {
451                    WorktreeHandle::Strong(old_worktree.clone())
452                } else {
453                    WorktreeHandle::Weak(old_worktree.downgrade())
454                };
455                self.worktrees.push(handle);
456            } else {
457                self.add(
458                    &Worktree::remote(project_id, replica_id, worktree, client.clone(), cx),
459                    cx,
460                );
461            }
462        }
463        self.send_project_updates(cx);
464
465        Ok(())
466    }
467
468    pub fn move_worktree(
469        &mut self,
470        source: WorktreeId,
471        destination: WorktreeId,
472        cx: &mut ModelContext<Self>,
473    ) -> Result<()> {
474        if source == destination {
475            return Ok(());
476        }
477
478        let mut source_index = None;
479        let mut destination_index = None;
480        for (i, worktree) in self.worktrees.iter().enumerate() {
481            if let Some(worktree) = worktree.upgrade() {
482                let worktree_id = worktree.read(cx).id();
483                if worktree_id == source {
484                    source_index = Some(i);
485                    if destination_index.is_some() {
486                        break;
487                    }
488                } else if worktree_id == destination {
489                    destination_index = Some(i);
490                    if source_index.is_some() {
491                        break;
492                    }
493                }
494            }
495        }
496
497        let source_index =
498            source_index.with_context(|| format!("Missing worktree for id {source}"))?;
499        let destination_index =
500            destination_index.with_context(|| format!("Missing worktree for id {destination}"))?;
501
502        if source_index == destination_index {
503            return Ok(());
504        }
505
506        let worktree_to_move = self.worktrees.remove(source_index);
507        self.worktrees.insert(destination_index, worktree_to_move);
508        self.worktrees_reordered = true;
509        cx.emit(WorktreeStoreEvent::WorktreeOrderChanged);
510        cx.notify();
511        Ok(())
512    }
513
514    pub fn disconnected_from_host(&mut self, cx: &mut AppContext) {
515        for worktree in &self.worktrees {
516            if let Some(worktree) = worktree.upgrade() {
517                worktree.update(cx, |worktree, _| {
518                    if let Some(worktree) = worktree.as_remote_mut() {
519                        worktree.disconnected_from_host();
520                    }
521                });
522            }
523        }
524    }
525
526    pub fn send_project_updates(&mut self, cx: &mut ModelContext<Self>) {
527        let Some((downstream_client, project_id)) = self.downstream_client.clone() else {
528            return;
529        };
530
531        let update = proto::UpdateProject {
532            project_id,
533            worktrees: self.worktree_metadata_protos(cx),
534        };
535
536        // collab has bad concurrency guarantees, so we send requests in serial.
537        let update_project = if downstream_client.is_via_collab() {
538            Some(downstream_client.request(update))
539        } else {
540            downstream_client.send(update).log_err();
541            None
542        };
543        cx.spawn(|this, mut cx| async move {
544            if let Some(update_project) = update_project {
545                update_project.await?;
546            }
547
548            this.update(&mut cx, |this, cx| {
549                let worktrees = this.worktrees().collect::<Vec<_>>();
550
551                for worktree in worktrees {
552                    worktree.update(cx, |worktree, cx| {
553                        let client = downstream_client.clone();
554                        worktree.observe_updates(project_id, cx, {
555                            move |update| {
556                                let client = client.clone();
557                                async move {
558                                    if client.is_via_collab() {
559                                        client
560                                            .request(update)
561                                            .map(|result| result.log_err().is_some())
562                                            .await
563                                    } else {
564                                        client.send(update).log_err().is_some()
565                                    }
566                                }
567                            }
568                        });
569                    });
570
571                    cx.emit(WorktreeStoreEvent::WorktreeUpdateSent(worktree.clone()))
572                }
573
574                anyhow::Ok(())
575            })
576        })
577        .detach_and_log_err(cx);
578    }
579
580    pub fn worktree_metadata_protos(&self, cx: &AppContext) -> Vec<proto::WorktreeMetadata> {
581        self.worktrees()
582            .map(|worktree| {
583                let worktree = worktree.read(cx);
584                proto::WorktreeMetadata {
585                    id: worktree.id().to_proto(),
586                    root_name: worktree.root_name().into(),
587                    visible: worktree.is_visible(),
588                    abs_path: worktree.abs_path().to_string_lossy().into(),
589                }
590            })
591            .collect()
592    }
593
594    pub fn shared(
595        &mut self,
596        remote_id: u64,
597        downsteam_client: AnyProtoClient,
598        cx: &mut ModelContext<Self>,
599    ) {
600        self.retain_worktrees = true;
601        self.downstream_client = Some((downsteam_client, remote_id));
602
603        // When shared, retain all worktrees
604        for worktree_handle in self.worktrees.iter_mut() {
605            match worktree_handle {
606                WorktreeHandle::Strong(_) => {}
607                WorktreeHandle::Weak(worktree) => {
608                    if let Some(worktree) = worktree.upgrade() {
609                        *worktree_handle = WorktreeHandle::Strong(worktree);
610                    }
611                }
612            }
613        }
614        self.send_project_updates(cx);
615    }
616
617    pub fn unshared(&mut self, cx: &mut ModelContext<Self>) {
618        self.retain_worktrees = false;
619        self.downstream_client.take();
620
621        // When not shared, only retain the visible worktrees
622        for worktree_handle in self.worktrees.iter_mut() {
623            if let WorktreeHandle::Strong(worktree) = worktree_handle {
624                let is_visible = worktree.update(cx, |worktree, _| {
625                    worktree.stop_observing_updates();
626                    worktree.is_visible()
627                });
628                if !is_visible {
629                    *worktree_handle = WorktreeHandle::Weak(worktree.downgrade());
630                }
631            }
632        }
633    }
634
635    /// search over all worktrees and return buffers that *might* match the search.
636    pub fn find_search_candidates(
637        &self,
638        query: SearchQuery,
639        limit: usize,
640        open_entries: HashSet<ProjectEntryId>,
641        fs: Arc<dyn Fs>,
642        cx: &ModelContext<Self>,
643    ) -> Receiver<ProjectPath> {
644        let snapshots = self
645            .visible_worktrees(cx)
646            .filter_map(|tree| {
647                let tree = tree.read(cx);
648                Some((tree.snapshot(), tree.as_local()?.settings()))
649            })
650            .collect::<Vec<_>>();
651
652        let executor = cx.background_executor().clone();
653
654        // We want to return entries in the order they are in the worktrees, so we have one
655        // thread that iterates over the worktrees (and ignored directories) as necessary,
656        // and pushes a oneshot::Receiver to the output channel and a oneshot::Sender to the filter
657        // channel.
658        // We spawn a number of workers that take items from the filter channel and check the query
659        // against the version of the file on disk.
660        let (filter_tx, filter_rx) = smol::channel::bounded(64);
661        let (output_tx, mut output_rx) = smol::channel::bounded(64);
662        let (matching_paths_tx, matching_paths_rx) = smol::channel::unbounded();
663
664        let input = cx.background_executor().spawn({
665            let fs = fs.clone();
666            let query = query.clone();
667            async move {
668                Self::find_candidate_paths(
669                    fs,
670                    snapshots,
671                    open_entries,
672                    query,
673                    filter_tx,
674                    output_tx,
675                )
676                .await
677                .log_err();
678            }
679        });
680        const MAX_CONCURRENT_FILE_SCANS: usize = 64;
681        let filters = cx.background_executor().spawn(async move {
682            let fs = &fs;
683            let query = &query;
684            executor
685                .scoped(move |scope| {
686                    for _ in 0..MAX_CONCURRENT_FILE_SCANS {
687                        let filter_rx = filter_rx.clone();
688                        scope.spawn(async move {
689                            Self::filter_paths(fs, filter_rx, query).await.log_err();
690                        })
691                    }
692                })
693                .await;
694        });
695        cx.background_executor()
696            .spawn(async move {
697                let mut matched = 0;
698                while let Some(mut receiver) = output_rx.next().await {
699                    let Some(path) = receiver.next().await else {
700                        continue;
701                    };
702                    let Ok(_) = matching_paths_tx.send(path).await else {
703                        break;
704                    };
705                    matched += 1;
706                    if matched == limit {
707                        break;
708                    }
709                }
710                drop(input);
711                drop(filters);
712            })
713            .detach();
714        matching_paths_rx
715    }
716
717    fn scan_ignored_dir<'a>(
718        fs: &'a Arc<dyn Fs>,
719        snapshot: &'a worktree::Snapshot,
720        path: &'a Path,
721        query: &'a SearchQuery,
722        include_root: bool,
723        filter_tx: &'a Sender<MatchingEntry>,
724        output_tx: &'a Sender<oneshot::Receiver<ProjectPath>>,
725    ) -> BoxFuture<'a, Result<()>> {
726        async move {
727            let abs_path = snapshot.abs_path().join(path);
728            let Some(mut files) = fs
729                .read_dir(&abs_path)
730                .await
731                .with_context(|| format!("listing ignored path {abs_path:?}"))
732                .log_err()
733            else {
734                return Ok(());
735            };
736
737            let mut results = Vec::new();
738
739            while let Some(Ok(file)) = files.next().await {
740                let Some(metadata) = fs
741                    .metadata(&file)
742                    .await
743                    .with_context(|| format!("fetching fs metadata for {abs_path:?}"))
744                    .log_err()
745                    .flatten()
746                else {
747                    continue;
748                };
749                if metadata.is_symlink || metadata.is_fifo {
750                    continue;
751                }
752                results.push((
753                    file.strip_prefix(snapshot.abs_path())?.to_path_buf(),
754                    !metadata.is_dir,
755                ))
756            }
757            results.sort_by(|(a_path, a_is_file), (b_path, b_is_file)| {
758                compare_paths((a_path, *a_is_file), (b_path, *b_is_file))
759            });
760            for (path, is_file) in results {
761                if is_file {
762                    if query.filters_path() {
763                        let matched_path = if include_root {
764                            let mut full_path = PathBuf::from(snapshot.root_name());
765                            full_path.push(&path);
766                            query.file_matches(&full_path)
767                        } else {
768                            query.file_matches(&path)
769                        };
770                        if !matched_path {
771                            continue;
772                        }
773                    }
774                    let (tx, rx) = oneshot::channel();
775                    output_tx.send(rx).await?;
776                    filter_tx
777                        .send(MatchingEntry {
778                            respond: tx,
779                            worktree_path: snapshot.abs_path().clone(),
780                            path: ProjectPath {
781                                worktree_id: snapshot.id(),
782                                path: Arc::from(path),
783                            },
784                        })
785                        .await?;
786                } else {
787                    Self::scan_ignored_dir(
788                        fs,
789                        snapshot,
790                        &path,
791                        query,
792                        include_root,
793                        filter_tx,
794                        output_tx,
795                    )
796                    .await?;
797                }
798            }
799            Ok(())
800        }
801        .boxed()
802    }
803
804    async fn find_candidate_paths(
805        fs: Arc<dyn Fs>,
806        snapshots: Vec<(worktree::Snapshot, WorktreeSettings)>,
807        open_entries: HashSet<ProjectEntryId>,
808        query: SearchQuery,
809        filter_tx: Sender<MatchingEntry>,
810        output_tx: Sender<oneshot::Receiver<ProjectPath>>,
811    ) -> Result<()> {
812        let include_root = snapshots.len() > 1;
813        for (snapshot, settings) in snapshots {
814            let mut entries: Vec<_> = snapshot.entries(query.include_ignored(), 0).collect();
815            entries.sort_by(|a, b| compare_paths((&a.path, a.is_file()), (&b.path, b.is_file())));
816            for entry in entries {
817                if entry.is_dir() && entry.is_ignored {
818                    if !settings.is_path_excluded(&entry.path) {
819                        Self::scan_ignored_dir(
820                            &fs,
821                            &snapshot,
822                            &entry.path,
823                            &query,
824                            include_root,
825                            &filter_tx,
826                            &output_tx,
827                        )
828                        .await?;
829                    }
830                    continue;
831                }
832
833                if entry.is_fifo || !entry.is_file() {
834                    continue;
835                }
836
837                if query.filters_path() {
838                    let matched_path = if include_root {
839                        let mut full_path = PathBuf::from(snapshot.root_name());
840                        full_path.push(&entry.path);
841                        query.file_matches(&full_path)
842                    } else {
843                        query.file_matches(&entry.path)
844                    };
845                    if !matched_path {
846                        continue;
847                    }
848                }
849
850                let (mut tx, rx) = oneshot::channel();
851
852                if open_entries.contains(&entry.id) {
853                    tx.send(ProjectPath {
854                        worktree_id: snapshot.id(),
855                        path: entry.path.clone(),
856                    })
857                    .await?;
858                } else {
859                    filter_tx
860                        .send(MatchingEntry {
861                            respond: tx,
862                            worktree_path: snapshot.abs_path().clone(),
863                            path: ProjectPath {
864                                worktree_id: snapshot.id(),
865                                path: entry.path.clone(),
866                            },
867                        })
868                        .await?;
869                }
870
871                output_tx.send(rx).await?;
872            }
873        }
874        Ok(())
875    }
876
877    async fn filter_paths(
878        fs: &Arc<dyn Fs>,
879        mut input: Receiver<MatchingEntry>,
880        query: &SearchQuery,
881    ) -> Result<()> {
882        while let Some(mut entry) = input.next().await {
883            let abs_path = entry.worktree_path.join(&entry.path.path);
884            let Some(file) = fs.open_sync(&abs_path).await.log_err() else {
885                continue;
886            };
887            if query.detect(file).unwrap_or(false) {
888                entry.respond.send(entry.path).await?
889            }
890        }
891
892        Ok(())
893    }
894
895    pub async fn handle_create_project_entry(
896        this: Model<Self>,
897        envelope: TypedEnvelope<proto::CreateProjectEntry>,
898        mut cx: AsyncAppContext,
899    ) -> Result<proto::ProjectEntryResponse> {
900        let worktree = this.update(&mut cx, |this, cx| {
901            let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
902            this.worktree_for_id(worktree_id, cx)
903                .ok_or_else(|| anyhow!("worktree not found"))
904        })??;
905        Worktree::handle_create_entry(worktree, envelope.payload, cx).await
906    }
907
908    pub async fn handle_rename_project_entry(
909        this: Model<Self>,
910        envelope: TypedEnvelope<proto::RenameProjectEntry>,
911        mut cx: AsyncAppContext,
912    ) -> Result<proto::ProjectEntryResponse> {
913        let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
914        let worktree = this.update(&mut cx, |this, cx| {
915            this.worktree_for_entry(entry_id, cx)
916                .ok_or_else(|| anyhow!("worktree not found"))
917        })??;
918        Worktree::handle_rename_entry(worktree, envelope.payload, cx).await
919    }
920
921    pub async fn handle_copy_project_entry(
922        this: Model<Self>,
923        envelope: TypedEnvelope<proto::CopyProjectEntry>,
924        mut cx: AsyncAppContext,
925    ) -> Result<proto::ProjectEntryResponse> {
926        let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
927        let worktree = this.update(&mut cx, |this, cx| {
928            this.worktree_for_entry(entry_id, cx)
929                .ok_or_else(|| anyhow!("worktree not found"))
930        })??;
931        Worktree::handle_copy_entry(worktree, envelope.payload, cx).await
932    }
933
934    pub async fn handle_delete_project_entry(
935        this: Model<Self>,
936        envelope: TypedEnvelope<proto::DeleteProjectEntry>,
937        mut cx: AsyncAppContext,
938    ) -> Result<proto::ProjectEntryResponse> {
939        let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
940        let worktree = this.update(&mut cx, |this, cx| {
941            this.worktree_for_entry(entry_id, cx)
942                .ok_or_else(|| anyhow!("worktree not found"))
943        })??;
944        Worktree::handle_delete_entry(worktree, envelope.payload, cx).await
945    }
946
947    pub async fn handle_expand_project_entry(
948        this: Model<Self>,
949        envelope: TypedEnvelope<proto::ExpandProjectEntry>,
950        mut cx: AsyncAppContext,
951    ) -> Result<proto::ExpandProjectEntryResponse> {
952        let entry_id = ProjectEntryId::from_proto(envelope.payload.entry_id);
953        let worktree = this
954            .update(&mut cx, |this, cx| this.worktree_for_entry(entry_id, cx))?
955            .ok_or_else(|| anyhow!("invalid request"))?;
956        Worktree::handle_expand_entry(worktree, envelope.payload, cx).await
957    }
958}
959
960#[derive(Clone)]
961enum WorktreeHandle {
962    Strong(Model<Worktree>),
963    Weak(WeakModel<Worktree>),
964}
965
966impl WorktreeHandle {
967    fn upgrade(&self) -> Option<Model<Worktree>> {
968        match self {
969            WorktreeHandle::Strong(handle) => Some(handle.clone()),
970            WorktreeHandle::Weak(handle) => handle.upgrade(),
971        }
972    }
973}