tenant.openagents/omega

1use crate::{
2    ProjectPath,
3    lsp_store::OpenLspBufferHandle,
4    worktree_store::{WorktreeStore, WorktreeStoreEvent},
5};
6use anyhow::{Context as _, Result, anyhow};
7use client::Client;
8use collections::{HashMap, HashSet, hash_map};
9use futures::{Future, FutureExt as _, channel::oneshot, future::Shared};
10use gpui::{
11    App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Subscription, Task, TaskExt,
12    WeakEntity,
13};
14use language::{
15    Buffer, BufferEvent, Capability, DiskState, File as _, Language, LineEnding, Operation,
16    language_settings::{AllLanguageSettings, LineEndingSetting},
17    proto::{
18        deserialize_line_ending, deserialize_version, serialize_line_ending, serialize_version,
19        split_operations,
20    },
21};
22use rpc::{
23    AnyProtoClient, ErrorCode, ErrorExt as _, TypedEnvelope,
24    proto::{self, PeerId},
25};
26
27use settings::Settings;
28use std::{io, sync::Arc, time::Instant};
29use text::{BufferId, ReplicaId};
30use util::{ResultExt as _, TryFutureExt, debug_panic, maybe, rel_path::RelPath};
31use worktree::{File, PathChange, ProjectEntryId, Worktree, WorktreeId, WorktreeSettings};
32
33/// A set of open buffers.
34pub struct BufferStore {
35    state: BufferStoreState,
36    #[allow(clippy::type_complexity)]
37    loading_buffers: HashMap<ProjectPath, Shared<Task<Result<Entity<Buffer>, Arc<anyhow::Error>>>>>,
38    worktree_store: Entity<WorktreeStore>,
39    opened_buffers: HashMap<BufferId, OpenBuffer>,
40    path_to_buffer_id: HashMap<ProjectPath, BufferId>,
41    downstream_client: Option<(AnyProtoClient, u64)>,
42    shared_buffers: HashMap<proto::PeerId, HashMap<BufferId, SharedBuffer>>,
43    non_searchable_buffers: HashSet<BufferId>,
44    project_search: RemoteProjectSearchState,
45}
46
47#[derive(Default)]
48struct RemoteProjectSearchState {
49    // List of ongoing project search chunks from our remote host. Used by the side issuing a search RPC request.
50    chunks: HashMap<u64, async_channel::Sender<BufferId>>,
51    // Monotonously-increasing handle to hand out to remote host in order to identify the project search result chunk.
52    next_id: u64,
53    // Used by the side running the actual search for match candidates to potentially cancel the search prematurely.
54    searches_in_progress: HashMap<(PeerId, u64), Task<Result<()>>>,
55}
56
57#[derive(Hash, Eq, PartialEq, Clone)]
58struct SharedBuffer {
59    buffer: Entity<Buffer>,
60    lsp_handle: Option<OpenLspBufferHandle>,
61}
62
63enum BufferStoreState {
64    Local(LocalBufferStore),
65    Remote(RemoteBufferStore),
66}
67
68struct RemoteBufferStore {
69    shared_with_me: HashSet<Entity<Buffer>>,
70    upstream_client: AnyProtoClient,
71    project_id: u64,
72    loading_remote_buffers_by_id: HashMap<BufferId, Entity<Buffer>>,
73    remote_buffer_listeners:
74        HashMap<BufferId, Vec<oneshot::Sender<anyhow::Result<Entity<Buffer>>>>>,
75    worktree_store: Entity<WorktreeStore>,
76}
77
78struct LocalBufferStore {
79    local_buffer_ids_by_entry_id: HashMap<ProjectEntryId, BufferId>,
80    worktree_store: Entity<WorktreeStore>,
81    _subscription: Subscription,
82}
83
84enum OpenBuffer {
85    Complete { buffer: WeakEntity<Buffer> },
86    Operations(Vec<Operation>),
87}
88
89pub enum BufferStoreEvent {
90    BufferAdded(Entity<Buffer>),
91    SharedBufferClosed(proto::PeerId, BufferId),
92    BufferDropped(BufferId),
93    BufferChangedFilePath {
94        buffer: Entity<Buffer>,
95        old_file: Option<Arc<dyn language::File>>,
96    },
97}
98
99#[derive(Default, Debug, Clone)]
100pub struct ProjectTransaction(pub HashMap<Entity<Buffer>, language::Transaction>);
101
102impl PartialEq for ProjectTransaction {
103    fn eq(&self, other: &Self) -> bool {
104        self.0.len() == other.0.len()
105            && self.0.iter().all(|(buffer, transaction)| {
106                other.0.get(buffer).is_some_and(|t| t.id == transaction.id)
107            })
108    }
109}
110
111impl EventEmitter<BufferStoreEvent> for BufferStore {}
112
113impl RemoteBufferStore {
114    pub fn wait_for_remote_buffer(
115        &mut self,
116        id: BufferId,
117        cx: &mut Context<BufferStore>,
118    ) -> Task<Result<Entity<Buffer>>> {
119        let (tx, rx) = oneshot::channel();
120        self.remote_buffer_listeners.entry(id).or_default().push(tx);
121
122        cx.spawn(async move |this, cx| {
123            if let Some(buffer) = this
124                .read_with(cx, |buffer_store, _| buffer_store.get(id))
125                .ok()
126                .flatten()
127            {
128                return Ok(buffer);
129            }
130
131            cx.background_spawn(async move { rx.await? }).await
132        })
133    }
134
135    fn save_remote_buffer(
136        &self,
137        buffer_handle: Entity<Buffer>,
138        new_path: Option<proto::ProjectPath>,
139        cx: &Context<BufferStore>,
140    ) -> Task<Result<()>> {
141        let buffer = buffer_handle.read(cx);
142        let buffer_id = buffer.remote_id().into();
143        let version = buffer.version();
144        let rpc = self.upstream_client.clone();
145        let project_id = self.project_id;
146        cx.spawn(async move |_, cx| {
147            let response = rpc
148                .request(proto::SaveBuffer {
149                    project_id,
150                    buffer_id,
151                    new_path,
152                    version: serialize_version(&version),
153                })
154                .await?;
155            let version = deserialize_version(&response.version);
156            let mtime = response.mtime.map(|mtime| mtime.into());
157
158            buffer_handle.update(cx, |buffer, cx| {
159                buffer.did_save(version.clone(), mtime, cx);
160            });
161
162            Ok(())
163        })
164    }
165
166    pub fn handle_create_buffer_for_peer(
167        &mut self,
168        envelope: TypedEnvelope<proto::CreateBufferForPeer>,
169        replica_id: ReplicaId,
170        capability: Capability,
171        cx: &mut Context<BufferStore>,
172    ) -> Result<Option<Entity<Buffer>>> {
173        match envelope.payload.variant.context("missing variant")? {
174            proto::create_buffer_for_peer::Variant::State(mut state) => {
175                let buffer_id = BufferId::new(state.id)?;
176
177                let buffer_result = maybe!({
178                    let mut buffer_file = None;
179                    if let Some(file) = state.file.take() {
180                        let worktree_id = worktree::WorktreeId::from_proto(file.worktree_id);
181                        let worktree = self
182                            .worktree_store
183                            .read(cx)
184                            .worktree_for_id(worktree_id, cx)
185                            .with_context(|| {
186                                format!("no worktree found for id {}", file.worktree_id)
187                            })?;
188                        buffer_file = Some(Arc::new(File::from_proto(file, worktree, cx)?)
189                            as Arc<dyn language::File>);
190                    }
191                    Buffer::from_proto(replica_id, capability, state, buffer_file)
192                });
193
194                match buffer_result {
195                    Ok(buffer) => {
196                        let buffer = cx.new(|_| buffer);
197                        self.loading_remote_buffers_by_id.insert(buffer_id, buffer);
198                    }
199                    Err(error) => {
200                        if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {
201                            for listener in listeners {
202                                listener.send(Err(anyhow!(error.cloned()))).ok();
203                            }
204                        }
205                    }
206                }
207            }
208            proto::create_buffer_for_peer::Variant::Chunk(chunk) => {
209                let buffer_id = BufferId::new(chunk.buffer_id)?;
210                let buffer = self
211                    .loading_remote_buffers_by_id
212                    .get(&buffer_id)
213                    .cloned()
214                    .with_context(|| {
215                        format!(
216                            "received chunk for buffer {} without initial state",
217                            chunk.buffer_id
218                        )
219                    })?;
220
221                let result = maybe!({
222                    let operations = chunk
223                        .operations
224                        .into_iter()
225                        .map(language::proto::deserialize_operation)
226                        .collect::<Result<Vec<_>>>()?;
227                    buffer.update(cx, |buffer, cx| buffer.apply_ops(operations, cx));
228                    anyhow::Ok(())
229                });
230
231                if let Err(error) = result {
232                    self.loading_remote_buffers_by_id.remove(&buffer_id);
233                    if let Some(listeners) = self.remote_buffer_listeners.remove(&buffer_id) {
234                        for listener in listeners {
235                            listener.send(Err(error.cloned())).ok();
236                        }
237                    }
238                } else if chunk.is_last {
239                    self.loading_remote_buffers_by_id.remove(&buffer_id);
240                    if self.upstream_client.is_via_collab() {
241                        // retain buffers sent by peers to avoid races.
242                        self.shared_with_me.insert(buffer.clone());
243                    }
244
245                    if let Some(senders) = self.remote_buffer_listeners.remove(&buffer_id) {
246                        for sender in senders {
247                            sender.send(Ok(buffer.clone())).ok();
248                        }
249                    }
250                    return Ok(Some(buffer));
251                }
252            }
253        }
254        Ok(None)
255    }
256
257    pub fn incomplete_buffer_ids(&self) -> Vec<BufferId> {
258        self.loading_remote_buffers_by_id
259            .keys()
260            .copied()
261            .collect::<Vec<_>>()
262    }
263
264    pub fn deserialize_project_transaction(
265        &self,
266        message: proto::ProjectTransaction,
267        push_to_history: bool,
268        cx: &mut Context<BufferStore>,
269    ) -> Task<Result<ProjectTransaction>> {
270        cx.spawn(async move |this, cx| {
271            let mut project_transaction = ProjectTransaction::default();
272            for (buffer_id, transaction) in message.buffer_ids.into_iter().zip(message.transactions)
273            {
274                let buffer_id = BufferId::new(buffer_id)?;
275                let buffer = this
276                    .update(cx, |this, cx| this.wait_for_remote_buffer(buffer_id, cx))?
277                    .await?;
278                let transaction = language::proto::deserialize_transaction(transaction)?;
279                project_transaction.0.insert(buffer, transaction);
280            }
281
282            for (buffer, transaction) in &project_transaction.0 {
283                buffer
284                    .update(cx, |buffer, _| {
285                        buffer.wait_for_edits(transaction.edit_ids.iter().copied())
286                    })
287                    .await?;
288
289                if push_to_history {
290                    buffer.update(cx, |buffer, _| {
291                        buffer.push_transaction(transaction.clone(), Instant::now());
292                        buffer.finalize_last_transaction();
293                    });
294                }
295            }
296
297            Ok(project_transaction)
298        })
299    }
300
301    fn open_buffer(
302        &self,
303        path: Arc<RelPath>,
304        worktree: Entity<Worktree>,
305        cx: &mut Context<BufferStore>,
306    ) -> Task<Result<Entity<Buffer>>> {
307        let worktree_id = worktree.read(cx).id().to_proto();
308        let project_id = self.project_id;
309        let client = self.upstream_client.clone();
310        cx.spawn(async move |this, cx| {
311            let response = client
312                .request(proto::OpenBufferByPath {
313                    project_id,
314                    worktree_id,
315                    path: path.as_unix_str().to_owned(),
316                })
317                .await?;
318            let buffer_id = BufferId::new(response.buffer_id)?;
319
320            let buffer = this
321                .update(cx, {
322                    |this, cx| this.wait_for_remote_buffer(buffer_id, cx)
323                })?
324                .await?;
325
326            Ok(buffer)
327        })
328    }
329
330    fn create_buffer(
331        &self,
332        language: Option<Arc<Language>>,
333        project_searchable: bool,
334        cx: &mut Context<BufferStore>,
335    ) -> Task<Result<Entity<Buffer>>> {
336        let create = self.upstream_client.request(proto::OpenNewBuffer {
337            project_id: self.project_id,
338        });
339        cx.spawn(async move |this, cx| {
340            let response = create.await?;
341            let buffer_id = BufferId::new(response.buffer_id)?;
342
343            let buffer = this
344                .update(cx, |this, cx| {
345                    if !project_searchable {
346                        this.non_searchable_buffers.insert(buffer_id);
347                    }
348                    this.wait_for_remote_buffer(buffer_id, cx)
349                })?
350                .await?;
351            if let Some(language) = language {
352                buffer.update(cx, |buffer, cx| {
353                    buffer.set_language(Some(language), cx);
354                });
355            }
356            Ok(buffer)
357        })
358    }
359
360    fn reload_buffers(
361        &self,
362        buffers: HashSet<Entity<Buffer>>,
363        push_to_history: bool,
364        cx: &mut Context<BufferStore>,
365    ) -> Task<Result<ProjectTransaction>> {
366        let request = self.upstream_client.request(proto::ReloadBuffers {
367            project_id: self.project_id,
368            buffer_ids: buffers
369                .iter()
370                .map(|buffer| buffer.read(cx).remote_id().to_proto())
371                .collect(),
372        });
373
374        cx.spawn(async move |this, cx| {
375            let response = request.await?.transaction.context("missing transaction")?;
376            this.update(cx, |this, cx| {
377                this.deserialize_project_transaction(response, push_to_history, cx)
378            })?
379            .await
380        })
381    }
382}
383
384impl LocalBufferStore {
385    fn save_local_buffer(
386        &self,
387        buffer_handle: Entity<Buffer>,
388        worktree: Entity<Worktree>,
389        path: Arc<RelPath>,
390        mut has_changed_file: bool,
391        cx: &mut Context<BufferStore>,
392    ) -> Task<Result<()>> {
393        let buffer = buffer_handle.read(cx);
394
395        let text = buffer.as_rope().clone();
396        let line_ending = buffer.line_ending();
397        let encoding = buffer.encoding();
398        let has_bom = buffer.has_bom();
399        let version = buffer.version();
400        let buffer_id = buffer.remote_id();
401        let file = buffer.file().cloned();
402        if file
403            .as_ref()
404            .is_some_and(|file| file.disk_state() == DiskState::New)
405        {
406            has_changed_file = true;
407        }
408
409        let save = worktree.update(cx, |worktree, cx| {
410            worktree.write_file(path, text, line_ending, encoding, has_bom, cx)
411        });
412
413        cx.spawn(async move |this, cx| {
414            let new_file = save.await?;
415            let mtime = new_file.disk_state().mtime();
416            this.update(cx, |this, cx| {
417                if let Some((downstream_client, project_id)) = this.downstream_client.clone() {
418                    if has_changed_file {
419                        downstream_client
420                            .send(proto::UpdateBufferFile {
421                                project_id,
422                                buffer_id: buffer_id.to_proto(),
423                                file: Some(language::File::to_proto(&*new_file, cx)),
424                            })
425                            .log_err();
426                    }
427                    downstream_client
428                        .send(proto::BufferSaved {
429                            project_id,
430                            buffer_id: buffer_id.to_proto(),
431                            version: serialize_version(&version),
432                            mtime: mtime.map(|time| time.into()),
433                        })
434                        .log_err();
435                }
436            })?;
437            buffer_handle.update(cx, |buffer, cx| {
438                if has_changed_file {
439                    buffer.file_updated(new_file, cx);
440                }
441                buffer.did_save(version.clone(), mtime, cx);
442            });
443            Ok(())
444        })
445    }
446
447    fn subscribe_to_worktree(
448        &mut self,
449        worktree: &Entity<Worktree>,
450        cx: &mut Context<BufferStore>,
451    ) {
452        cx.subscribe(worktree, |this, worktree, event, cx| {
453            if worktree.read(cx).is_local()
454                && let worktree::Event::UpdatedEntries(changes) = event
455            {
456                Self::local_worktree_entries_changed(this, &worktree, changes, cx);
457            }
458        })
459        .detach();
460    }
461
462    fn local_worktree_entries_changed(
463        this: &mut BufferStore,
464        worktree_handle: &Entity<Worktree>,
465        changes: &[(Arc<RelPath>, ProjectEntryId, PathChange)],
466        cx: &mut Context<BufferStore>,
467    ) {
468        let snapshot = worktree_handle.read(cx).snapshot();
469        for (path, entry_id, _) in changes {
470            Self::local_worktree_entry_changed(
471                this,
472                *entry_id,
473                path,
474                worktree_handle,
475                &snapshot,
476                cx,
477            );
478        }
479    }
480
481    fn local_worktree_entry_changed(
482        this: &mut BufferStore,
483        entry_id: ProjectEntryId,
484        path: &Arc<RelPath>,
485        worktree: &Entity<worktree::Worktree>,
486        snapshot: &worktree::Snapshot,
487        cx: &mut Context<BufferStore>,
488    ) -> Option<()> {
489        let project_path = ProjectPath {
490            worktree_id: snapshot.id(),
491            path: path.clone(),
492        };
493
494        let buffer_id = this
495            .as_local_mut()
496            .and_then(|local| local.local_buffer_ids_by_entry_id.get(&entry_id))
497            .copied()
498            .or_else(|| this.path_to_buffer_id.get(&project_path).copied())?;
499
500        let buffer = if let Some(buffer) = this.get(buffer_id) {
501            Some(buffer)
502        } else {
503            this.opened_buffers.remove(&buffer_id);
504            this.non_searchable_buffers.remove(&buffer_id);
505            None
506        };
507
508        let buffer = if let Some(buffer) = buffer {
509            buffer
510        } else {
511            this.path_to_buffer_id.remove(&project_path);
512            let this = this.as_local_mut()?;
513            this.local_buffer_ids_by_entry_id.remove(&entry_id);
514            return None;
515        };
516
517        let events = buffer.update(cx, |buffer, cx| {
518            let file = buffer.file()?;
519            let old_file = File::from_dyn(Some(file))?;
520            if old_file.worktree != *worktree {
521                return None;
522            }
523
524            let snapshot_entry = old_file
525                .entry_id
526                .and_then(|entry_id| snapshot.entry_for_id(entry_id))
527                .or_else(|| snapshot.entry_for_path(old_file.path.as_ref()));
528
529            let new_file = if let Some(entry) = snapshot_entry {
530                File {
531                    disk_state: match entry.mtime {
532                        Some(mtime) => DiskState::Present {
533                            mtime,
534                            size: entry.size,
535                        },
536                        None => old_file.disk_state,
537                    },
538                    is_local: true,
539                    entry_id: Some(entry.id),
540                    path: entry.path.clone(),
541                    worktree: worktree.clone(),
542                    is_private: entry.is_private,
543                }
544            } else {
545                File {
546                    disk_state: DiskState::Deleted,
547                    is_local: true,
548                    entry_id: old_file.entry_id,
549                    path: old_file.path.clone(),
550                    worktree: worktree.clone(),
551                    is_private: old_file.is_private,
552                }
553            };
554
555            if new_file == *old_file {
556                return None;
557            }
558
559            let mut events = Vec::new();
560            if new_file.path != old_file.path {
561                this.path_to_buffer_id.remove(&ProjectPath {
562                    path: old_file.path.clone(),
563                    worktree_id: old_file.worktree_id(cx),
564                });
565                this.path_to_buffer_id.insert(
566                    ProjectPath {
567                        worktree_id: new_file.worktree_id(cx),
568                        path: new_file.path.clone(),
569                    },
570                    buffer_id,
571                );
572                events.push(BufferStoreEvent::BufferChangedFilePath {
573                    buffer: cx.entity(),
574                    old_file: buffer.file().cloned(),
575                });
576            }
577            let local = this.as_local_mut()?;
578            if new_file.entry_id != old_file.entry_id {
579                if let Some(entry_id) = old_file.entry_id {
580                    local.local_buffer_ids_by_entry_id.remove(&entry_id);
581                }
582                if let Some(entry_id) = new_file.entry_id {
583                    local
584                        .local_buffer_ids_by_entry_id
585                        .insert(entry_id, buffer_id);
586                }
587            }
588
589            if let Some((client, project_id)) = &this.downstream_client {
590                client
591                    .send(proto::UpdateBufferFile {
592                        project_id: *project_id,
593                        buffer_id: buffer_id.to_proto(),
594                        file: Some(new_file.to_proto(cx)),
595                    })
596                    .ok();
597            }
598
599            buffer.file_updated(Arc::new(new_file), cx);
600            Some(events)
601        })?;
602
603        for event in events {
604            cx.emit(event);
605        }
606
607        None
608    }
609
610    fn save_buffer(
611        &self,
612        buffer: Entity<Buffer>,
613        cx: &mut Context<BufferStore>,
614    ) -> Task<Result<()>> {
615        let Some(file) = File::from_dyn(buffer.read(cx).file()) else {
616            return Task::ready(Err(anyhow!("buffer doesn't have a file")));
617        };
618        let worktree = file.worktree.clone();
619        self.save_local_buffer(buffer, worktree, file.path.clone(), false, cx)
620    }
621
622    fn save_buffer_as(
623        &self,
624        buffer: Entity<Buffer>,
625        path: ProjectPath,
626        cx: &mut Context<BufferStore>,
627    ) -> Task<Result<()>> {
628        let Some(worktree) = self
629            .worktree_store
630            .read(cx)
631            .worktree_for_id(path.worktree_id, cx)
632        else {
633            return Task::ready(Err(anyhow!("no such worktree")));
634        };
635        self.save_local_buffer(buffer, worktree, path.path, true, cx)
636    }
637
638    #[ztracing::instrument(skip_all)]
639    fn open_buffer(
640        &self,
641        path: Arc<RelPath>,
642        worktree: Entity<Worktree>,
643        cx: &mut Context<BufferStore>,
644    ) -> Task<Result<Entity<Buffer>>> {
645        let load_file = worktree.update(cx, |worktree, cx| worktree.load_file(path.as_ref(), cx));
646        cx.spawn(async move |this, cx| {
647            let path = path.clone();
648            let buffer = match load_file.await {
649                Ok(loaded) => {
650                    let is_writable = loaded.is_writable;
651                    let capability = if is_writable {
652                        Capability::ReadWrite
653                    } else {
654                        Capability::Read
655                    };
656                    let reservation = cx.reserve_entity::<Buffer>();
657                    let buffer_id = BufferId::from(reservation.entity_id().as_non_zero_u64());
658                    let text_buffer = cx
659                        .background_spawn(async move {
660                            text::Buffer::new(ReplicaId::LOCAL, buffer_id, loaded.text)
661                        })
662                        .await;
663                    cx.insert_entity(reservation, |_| {
664                        let mut buffer = Buffer::build(text_buffer, Some(loaded.file), capability);
665                        buffer.set_encoding(loaded.encoding);
666                        buffer.set_has_bom(loaded.has_bom);
667                        buffer
668                    })
669                }
670                Err(error) if is_not_found_error(&error) => cx.new(|cx| {
671                    let buffer_id = BufferId::from(cx.entity_id().as_non_zero_u64());
672                    let text_buffer = text::Buffer::new(ReplicaId::LOCAL, buffer_id, "");
673                    let mut buffer = Buffer::build(
674                        text_buffer,
675                        Some(Arc::new(File {
676                            worktree,
677                            path,
678                            disk_state: DiskState::New,
679                            entry_id: None,
680                            is_local: true,
681                            is_private: false,
682                        })),
683                        Capability::ReadWrite,
684                    );
685                    apply_initial_line_ending(&mut buffer, cx);
686                    buffer
687                }),
688                Err(e) => return Err(e),
689            };
690            this.update(cx, |this, cx| {
691                this.add_buffer(buffer.clone(), cx)?;
692                let buffer_id = buffer.read(cx).remote_id();
693                if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
694                    let project_path = ProjectPath {
695                        worktree_id: file.worktree_id(cx),
696                        path: file.path.clone(),
697                    };
698                    let entry_id = file.entry_id;
699
700                    // Check if the file should be read-only based on settings
701                    let settings = WorktreeSettings::get(Some((&project_path).into()), cx);
702                    let is_read_only = if project_path.path.is_empty() {
703                        settings.is_std_path_read_only(&file.full_path(cx))
704                    } else {
705                        settings.is_path_read_only(&project_path.path)
706                    };
707                    if is_read_only {
708                        buffer.update(cx, |buffer, cx| {
709                            buffer.set_capability(Capability::Read, cx);
710                        });
711                    }
712
713                    this.path_to_buffer_id.insert(project_path, buffer_id);
714                    let this = this.as_local_mut().unwrap();
715                    if let Some(entry_id) = entry_id {
716                        this.local_buffer_ids_by_entry_id
717                            .insert(entry_id, buffer_id);
718                    }
719                }
720
721                anyhow::Ok(())
722            })??;
723
724            Ok(buffer)
725        })
726    }
727
728    fn create_buffer(
729        &self,
730        language: Option<Arc<Language>>,
731        project_searchable: bool,
732        cx: &mut Context<BufferStore>,
733    ) -> Task<Result<Entity<Buffer>>> {
734        cx.spawn(async move |buffer_store, cx| {
735            let buffer = cx.new(|cx| {
736                let mut buffer = Buffer::local("", cx)
737                    .with_language(language.unwrap_or_else(|| language::PLAIN_TEXT.clone()), cx);
738                apply_initial_line_ending(&mut buffer, cx);
739                buffer
740            });
741            buffer_store.update(cx, |buffer_store, cx| {
742                buffer_store.add_buffer(buffer.clone(), cx).log_err();
743                if !project_searchable {
744                    buffer_store
745                        .non_searchable_buffers
746                        .insert(buffer.read(cx).remote_id());
747                }
748            })?;
749            Ok(buffer)
750        })
751    }
752
753    fn reload_buffers(
754        &self,
755        buffers: HashSet<Entity<Buffer>>,
756        push_to_history: bool,
757        cx: &mut Context<BufferStore>,
758    ) -> Task<Result<ProjectTransaction>> {
759        cx.spawn(async move |_, cx| {
760            let mut project_transaction = ProjectTransaction::default();
761            for buffer in buffers {
762                let transaction = buffer.update(cx, |buffer, cx| buffer.reload(cx)).await?;
763                buffer.update(cx, |buffer, cx| {
764                    if let Some(transaction) = transaction {
765                        if !push_to_history {
766                            buffer.forget_transaction(transaction.id);
767                        }
768                        project_transaction.0.insert(cx.entity(), transaction);
769                    }
770                });
771            }
772
773            Ok(project_transaction)
774        })
775    }
776}
777
778impl BufferStore {
779    pub fn init(client: &AnyProtoClient) {
780        client.add_entity_message_handler(Self::handle_buffer_reloaded);
781        client.add_entity_message_handler(Self::handle_buffer_saved);
782        client.add_entity_message_handler(Self::handle_update_buffer_file);
783        client.add_entity_request_handler(Self::handle_save_buffer);
784        client.add_entity_request_handler(Self::handle_reload_buffers);
785    }
786
787    /// Creates a buffer store, optionally retaining its buffers.
788    pub fn local(worktree_store: Entity<WorktreeStore>, cx: &mut Context<Self>) -> Self {
789        Self {
790            state: BufferStoreState::Local(LocalBufferStore {
791                local_buffer_ids_by_entry_id: Default::default(),
792                worktree_store: worktree_store.clone(),
793                _subscription: cx.subscribe(&worktree_store, |this, _, event, cx| {
794                    if let WorktreeStoreEvent::WorktreeAdded(worktree) = event {
795                        let this = this.as_local_mut().unwrap();
796                        this.subscribe_to_worktree(worktree, cx);
797                    }
798                }),
799            }),
800            downstream_client: None,
801            opened_buffers: Default::default(),
802            path_to_buffer_id: Default::default(),
803            shared_buffers: Default::default(),
804            loading_buffers: Default::default(),
805            non_searchable_buffers: Default::default(),
806            worktree_store,
807            project_search: Default::default(),
808        }
809    }
810
811    pub fn remote(
812        worktree_store: Entity<WorktreeStore>,
813        upstream_client: AnyProtoClient,
814        remote_id: u64,
815        _cx: &mut Context<Self>,
816    ) -> Self {
817        Self {
818            state: BufferStoreState::Remote(RemoteBufferStore {
819                shared_with_me: Default::default(),
820                loading_remote_buffers_by_id: Default::default(),
821                remote_buffer_listeners: Default::default(),
822                project_id: remote_id,
823                upstream_client,
824                worktree_store: worktree_store.clone(),
825            }),
826            downstream_client: None,
827            opened_buffers: Default::default(),
828            path_to_buffer_id: Default::default(),
829            loading_buffers: Default::default(),
830            shared_buffers: Default::default(),
831            non_searchable_buffers: Default::default(),
832            worktree_store,
833            project_search: Default::default(),
834        }
835    }
836
837    fn as_local_mut(&mut self) -> Option<&mut LocalBufferStore> {
838        match &mut self.state {
839            BufferStoreState::Local(state) => Some(state),
840            _ => None,
841        }
842    }
843
844    fn as_remote_mut(&mut self) -> Option<&mut RemoteBufferStore> {
845        match &mut self.state {
846            BufferStoreState::Remote(state) => Some(state),
847            _ => None,
848        }
849    }
850
851    fn as_remote(&self) -> Option<&RemoteBufferStore> {
852        match &self.state {
853            BufferStoreState::Remote(state) => Some(state),
854            _ => None,
855        }
856    }
857
858    #[ztracing::instrument(skip_all)]
859    pub fn open_buffer(
860        &mut self,
861        project_path: ProjectPath,
862        cx: &mut Context<Self>,
863    ) -> Task<Result<Entity<Buffer>>> {
864        if let Some(buffer) = self.get_by_path(&project_path) {
865            return Task::ready(Ok(buffer));
866        }
867
868        let task = match self.loading_buffers.entry(project_path.clone()) {
869            hash_map::Entry::Occupied(e) => e.get().clone(),
870            hash_map::Entry::Vacant(entry) => {
871                let path = project_path.path.clone();
872                let Some(worktree) = self
873                    .worktree_store
874                    .read(cx)
875                    .worktree_for_id(project_path.worktree_id, cx)
876                else {
877                    return Task::ready(Err(anyhow!("no such worktree")));
878                };
879                let load_buffer = match &self.state {
880                    BufferStoreState::Local(this) => this.open_buffer(path, worktree, cx),
881                    BufferStoreState::Remote(this) => this.open_buffer(path, worktree, cx),
882                };
883
884                entry
885                    .insert(
886                        cx.spawn(async move |this, cx| {
887                            let load_result = load_buffer.await;
888                            this.update(cx, |this, _cx| {
889                                // Record the fact that the buffer is no longer loading.
890                                this.loading_buffers.remove(&project_path);
891
892                                let buffer = load_result.map_err(Arc::new)?;
893                                Ok(buffer)
894                            })?
895                        })
896                        .shared(),
897                    )
898                    .clone()
899            }
900        };
901
902        cx.background_spawn(async move {
903            task.await.map_err(|e| {
904                if e.error_code() != ErrorCode::Internal {
905                    anyhow!(e.error_code())
906                } else {
907                    anyhow!("{e}")
908                }
909            })
910        })
911    }
912
913    pub fn create_buffer(
914        &mut self,
915        language: Option<Arc<Language>>,
916        project_searchable: bool,
917        cx: &mut Context<Self>,
918    ) -> Task<Result<Entity<Buffer>>> {
919        match &self.state {
920            BufferStoreState::Local(this) => this.create_buffer(language, project_searchable, cx),
921            BufferStoreState::Remote(this) => this.create_buffer(language, project_searchable, cx),
922        }
923    }
924
925    pub fn save_buffer(
926        &mut self,
927        buffer: Entity<Buffer>,
928        cx: &mut Context<Self>,
929    ) -> Task<Result<()>> {
930        match &mut self.state {
931            BufferStoreState::Local(this) => this.save_buffer(buffer, cx),
932            BufferStoreState::Remote(this) => this.save_remote_buffer(buffer, None, cx),
933        }
934    }
935
936    pub fn save_buffer_as(
937        &mut self,
938        buffer: Entity<Buffer>,
939        path: ProjectPath,
940        cx: &mut Context<Self>,
941    ) -> Task<Result<()>> {
942        let old_file = buffer.read(cx).file().cloned();
943        let task = match &self.state {
944            BufferStoreState::Local(this) => this.save_buffer_as(buffer.clone(), path, cx),
945            BufferStoreState::Remote(this) => {
946                this.save_remote_buffer(buffer.clone(), Some(path.to_proto()), cx)
947            }
948        };
949        cx.spawn(async move |this, cx| {
950            task.await?;
951            this.update(cx, |this, cx| {
952                old_file.clone().and_then(|file| {
953                    this.path_to_buffer_id.remove(&ProjectPath {
954                        worktree_id: file.worktree_id(cx),
955                        path: file.path().clone(),
956                    })
957                });
958
959                cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });
960            })
961        })
962    }
963
964    fn add_buffer(&mut self, buffer_entity: Entity<Buffer>, cx: &mut Context<Self>) -> Result<()> {
965        let buffer = buffer_entity.read(cx);
966        let remote_id = buffer.remote_id();
967        let path = File::from_dyn(buffer.file()).map(|file| ProjectPath {
968            path: file.path.clone(),
969            worktree_id: file.worktree_id(cx),
970        });
971        let is_remote = buffer.replica_id().is_remote();
972        let open_buffer = OpenBuffer::Complete {
973            buffer: buffer_entity.downgrade(),
974        };
975
976        let handle = cx.entity().downgrade();
977        buffer_entity.update(cx, move |_, cx| {
978            cx.on_release(move |buffer, cx| {
979                handle
980                    .update(cx, |_, cx| {
981                        cx.emit(BufferStoreEvent::BufferDropped(buffer.remote_id()))
982                    })
983                    .ok();
984            })
985            .detach()
986        });
987        let _expect_path_to_exist;
988        match self.opened_buffers.entry(remote_id) {
989            hash_map::Entry::Vacant(entry) => {
990                entry.insert(open_buffer);
991                _expect_path_to_exist = false;
992            }
993            hash_map::Entry::Occupied(mut entry) => {
994                if let OpenBuffer::Operations(operations) = entry.get_mut() {
995                    buffer_entity.update(cx, |b, cx| b.apply_ops(operations.drain(..), cx));
996                } else if entry.get().upgrade().is_some() {
997                    if is_remote {
998                        return Ok(());
999                    } else {
1000                        debug_panic!("buffer {remote_id} was already registered");
1001                        anyhow::bail!("buffer {remote_id} was already registered");
1002                    }
1003                }
1004                entry.insert(open_buffer);
1005                _expect_path_to_exist = true;
1006            }
1007        }
1008
1009        if let Some(path) = path {
1010            self.path_to_buffer_id.insert(path, remote_id);
1011        }
1012
1013        cx.subscribe(&buffer_entity, Self::on_buffer_event).detach();
1014        cx.emit(BufferStoreEvent::BufferAdded(buffer_entity));
1015        Ok(())
1016    }
1017
1018    pub fn buffers(&self) -> impl '_ + Iterator<Item = Entity<Buffer>> {
1019        self.opened_buffers
1020            .values()
1021            .filter_map(|buffer| buffer.upgrade())
1022    }
1023
1024    pub(crate) fn is_searchable(&self, id: &BufferId) -> bool {
1025        !self.non_searchable_buffers.contains(&id)
1026    }
1027
1028    pub fn loading_buffers(
1029        &self,
1030    ) -> impl Iterator<Item = (&ProjectPath, impl Future<Output = Result<Entity<Buffer>>>)> {
1031        self.loading_buffers.iter().map(|(path, task)| {
1032            let task = task.clone();
1033            (path, async move {
1034                task.await.map_err(|e| {
1035                    if e.error_code() != ErrorCode::Internal {
1036                        anyhow!(e.error_code())
1037                    } else {
1038                        anyhow!("{e}")
1039                    }
1040                })
1041            })
1042        })
1043    }
1044
1045    pub fn buffer_id_for_project_path(&self, project_path: &ProjectPath) -> Option<&BufferId> {
1046        self.path_to_buffer_id.get(project_path)
1047    }
1048
1049    pub fn get_by_path(&self, path: &ProjectPath) -> Option<Entity<Buffer>> {
1050        self.path_to_buffer_id
1051            .get(path)
1052            .and_then(|buffer_id| self.get(*buffer_id))
1053    }
1054
1055    pub fn get(&self, buffer_id: BufferId) -> Option<Entity<Buffer>> {
1056        self.opened_buffers.get(&buffer_id)?.upgrade()
1057    }
1058
1059    pub fn get_existing(&self, buffer_id: BufferId) -> Result<Entity<Buffer>> {
1060        self.get(buffer_id)
1061            .with_context(|| format!("unknown buffer id {buffer_id}"))
1062    }
1063
1064    pub fn get_possibly_incomplete(&self, buffer_id: BufferId) -> Option<Entity<Buffer>> {
1065        self.get(buffer_id).or_else(|| {
1066            self.as_remote()
1067                .and_then(|remote| remote.loading_remote_buffers_by_id.get(&buffer_id).cloned())
1068        })
1069    }
1070
1071    pub fn buffer_version_info(&self, cx: &App) -> (Vec<proto::BufferVersion>, Vec<BufferId>) {
1072        let buffers = self
1073            .buffers()
1074            .map(|buffer| {
1075                let buffer = buffer.read(cx);
1076                proto::BufferVersion {
1077                    id: buffer.remote_id().into(),
1078                    version: language::proto::serialize_version(&buffer.version),
1079                }
1080            })
1081            .collect();
1082        let incomplete_buffer_ids = self
1083            .as_remote()
1084            .map(|remote| remote.incomplete_buffer_ids())
1085            .unwrap_or_default();
1086        (buffers, incomplete_buffer_ids)
1087    }
1088
1089    pub fn disconnected_from_host(&mut self, cx: &mut App) {
1090        for open_buffer in self.opened_buffers.values_mut() {
1091            if let Some(buffer) = open_buffer.upgrade() {
1092                buffer.update(cx, |buffer, _| buffer.give_up_waiting());
1093            }
1094        }
1095
1096        for buffer in self.buffers() {
1097            buffer.update(cx, |buffer, cx| {
1098                buffer.set_capability(Capability::ReadOnly, cx)
1099            });
1100        }
1101
1102        if let Some(remote) = self.as_remote_mut() {
1103            // Wake up all futures currently waiting on a buffer to get opened,
1104            // to give them a chance to fail now that we've disconnected.
1105            remote.remote_buffer_listeners.clear()
1106        }
1107    }
1108
1109    pub fn shared(&mut self, remote_id: u64, downstream_client: AnyProtoClient, _cx: &mut App) {
1110        self.downstream_client = Some((downstream_client, remote_id));
1111    }
1112
1113    pub fn unshared(&mut self, _cx: &mut Context<Self>) {
1114        self.downstream_client.take();
1115        self.forget_shared_buffers();
1116    }
1117
1118    pub fn discard_incomplete(&mut self) {
1119        self.opened_buffers
1120            .retain(|_, buffer| !matches!(buffer, OpenBuffer::Operations(_)));
1121    }
1122
1123    fn buffer_changed_file(&mut self, buffer: Entity<Buffer>, cx: &mut App) -> Option<()> {
1124        let file = File::from_dyn(buffer.read(cx).file())?;
1125
1126        let remote_id = buffer.read(cx).remote_id();
1127        if let Some(entry_id) = file.entry_id {
1128            if let Some(local) = self.as_local_mut() {
1129                match local.local_buffer_ids_by_entry_id.get(&entry_id) {
1130                    Some(_) => {
1131                        return None;
1132                    }
1133                    None => {
1134                        local
1135                            .local_buffer_ids_by_entry_id
1136                            .insert(entry_id, remote_id);
1137                    }
1138                }
1139            }
1140            self.path_to_buffer_id.insert(
1141                ProjectPath {
1142                    worktree_id: file.worktree_id(cx),
1143                    path: file.path.clone(),
1144                },
1145                remote_id,
1146            );
1147        };
1148
1149        Some(())
1150    }
1151
1152    fn on_buffer_event(
1153        &mut self,
1154        buffer: Entity<Buffer>,
1155        event: &BufferEvent,
1156        cx: &mut Context<Self>,
1157    ) {
1158        match event {
1159            BufferEvent::FileHandleChanged => {
1160                self.buffer_changed_file(buffer, cx);
1161            }
1162            BufferEvent::Reloaded => {
1163                let Some((downstream_client, project_id)) = self.downstream_client.as_ref() else {
1164                    return;
1165                };
1166                let buffer = buffer.read(cx);
1167                downstream_client
1168                    .send(proto::BufferReloaded {
1169                        project_id: *project_id,
1170                        buffer_id: buffer.remote_id().to_proto(),
1171                        version: serialize_version(&buffer.version()),
1172                        mtime: buffer.saved_mtime().map(|t| t.into()),
1173                        line_ending: serialize_line_ending(buffer.line_ending()) as i32,
1174                    })
1175                    .log_err();
1176            }
1177            BufferEvent::LanguageChanged(_) => {}
1178            _ => {}
1179        }
1180    }
1181
1182    pub async fn handle_update_buffer(
1183        this: Entity<Self>,
1184        envelope: TypedEnvelope<proto::UpdateBuffer>,
1185        mut cx: AsyncApp,
1186    ) -> Result<proto::Ack> {
1187        let payload = envelope.payload;
1188        let buffer_id = BufferId::new(payload.buffer_id)?;
1189        let ops = payload
1190            .operations
1191            .into_iter()
1192            .map(language::proto::deserialize_operation)
1193            .collect::<Result<Vec<_>, _>>()?;
1194        this.update(&mut cx, |this, cx| {
1195            match this.opened_buffers.entry(buffer_id) {
1196                hash_map::Entry::Occupied(mut e) => match e.get_mut() {
1197                    OpenBuffer::Operations(operations) => operations.extend_from_slice(&ops),
1198                    OpenBuffer::Complete { buffer, .. } => {
1199                        if let Some(buffer) = buffer.upgrade() {
1200                            buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx));
1201                        }
1202                    }
1203                },
1204                hash_map::Entry::Vacant(e) => {
1205                    e.insert(OpenBuffer::Operations(ops));
1206                }
1207            }
1208            Ok(proto::Ack {})
1209        })
1210    }
1211
1212    pub fn register_shared_lsp_handle(
1213        &mut self,
1214        peer_id: proto::PeerId,
1215        buffer_id: BufferId,
1216        handle: OpenLspBufferHandle,
1217    ) {
1218        if let Some(shared_buffers) = self.shared_buffers.get_mut(&peer_id)
1219            && let Some(buffer) = shared_buffers.get_mut(&buffer_id)
1220        {
1221            buffer.lsp_handle = Some(handle);
1222            return;
1223        }
1224        debug_panic!("tried to register shared lsp handle, but buffer was not shared")
1225    }
1226
1227    pub fn handle_synchronize_buffers(
1228        &mut self,
1229        envelope: TypedEnvelope<proto::SynchronizeBuffers>,
1230        cx: &mut Context<Self>,
1231        client: Arc<Client>,
1232    ) -> Result<proto::SynchronizeBuffersResponse> {
1233        let project_id = envelope.payload.project_id;
1234        let mut response = proto::SynchronizeBuffersResponse {
1235            buffers: Default::default(),
1236        };
1237        let Some(guest_id) = envelope.original_sender_id else {
1238            anyhow::bail!("missing original_sender_id on SynchronizeBuffers request");
1239        };
1240
1241        self.shared_buffers.entry(guest_id).or_default().clear();
1242        for buffer in envelope.payload.buffers {
1243            let buffer_id = BufferId::new(buffer.id)?;
1244            let remote_version = language::proto::deserialize_version(&buffer.version);
1245            if let Some(buffer) = self.get(buffer_id) {
1246                self.shared_buffers
1247                    .entry(guest_id)
1248                    .or_default()
1249                    .entry(buffer_id)
1250                    .or_insert_with(|| SharedBuffer {
1251                        buffer: buffer.clone(),
1252                        lsp_handle: None,
1253                    });
1254
1255                let buffer = buffer.read(cx);
1256                response.buffers.push(proto::BufferVersion {
1257                    id: buffer_id.into(),
1258                    version: language::proto::serialize_version(&buffer.version),
1259                });
1260
1261                let operations = buffer.serialize_ops(Some(remote_version), cx);
1262                let client = client.clone();
1263                if let Some(file) = buffer.file() {
1264                    client
1265                        .send(proto::UpdateBufferFile {
1266                            project_id,
1267                            buffer_id: buffer_id.into(),
1268                            file: Some(file.to_proto(cx)),
1269                        })
1270                        .log_err();
1271                }
1272
1273                // TODO(max): do something
1274                // client
1275                //     .send(proto::UpdateStagedText {
1276                //         project_id,
1277                //         buffer_id: buffer_id.into(),
1278                //         diff_base: buffer.diff_base().map(ToString::to_string),
1279                //     })
1280                //     .log_err();
1281
1282                client
1283                    .send(proto::BufferReloaded {
1284                        project_id,
1285                        buffer_id: buffer_id.into(),
1286                        version: language::proto::serialize_version(buffer.saved_version()),
1287                        mtime: buffer.saved_mtime().map(|time| time.into()),
1288                        line_ending: language::proto::serialize_line_ending(buffer.line_ending())
1289                            as i32,
1290                    })
1291                    .log_err();
1292
1293                cx.background_spawn(
1294                    async move {
1295                        let operations = operations.await;
1296                        for chunk in split_operations(operations) {
1297                            client
1298                                .request(proto::UpdateBuffer {
1299                                    project_id,
1300                                    buffer_id: buffer_id.into(),
1301                                    operations: chunk,
1302                                })
1303                                .await?;
1304                        }
1305                        anyhow::Ok(())
1306                    }
1307                    .log_err(),
1308                )
1309                .detach();
1310            }
1311        }
1312        Ok(response)
1313    }
1314
1315    pub fn handle_create_buffer_for_peer(
1316        &mut self,
1317        envelope: TypedEnvelope<proto::CreateBufferForPeer>,
1318        replica_id: ReplicaId,
1319        capability: Capability,
1320        cx: &mut Context<Self>,
1321    ) -> Result<()> {
1322        let remote = self
1323            .as_remote_mut()
1324            .context("buffer store is not a remote")?;
1325
1326        if let Some(buffer) =
1327            remote.handle_create_buffer_for_peer(envelope, replica_id, capability, cx)?
1328        {
1329            self.add_buffer(buffer, cx)?;
1330        }
1331
1332        Ok(())
1333    }
1334
1335    pub async fn handle_update_buffer_file(
1336        this: Entity<Self>,
1337        envelope: TypedEnvelope<proto::UpdateBufferFile>,
1338        mut cx: AsyncApp,
1339    ) -> Result<()> {
1340        let buffer_id = envelope.payload.buffer_id;
1341        let buffer_id = BufferId::new(buffer_id)?;
1342
1343        this.update(&mut cx, |this, cx| {
1344            let payload = envelope.payload.clone();
1345            if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1346                let file = payload.file.context("invalid file")?;
1347                let worktree = this
1348                    .worktree_store
1349                    .read(cx)
1350                    .worktree_for_id(WorktreeId::from_proto(file.worktree_id), cx)
1351                    .context("no such worktree")?;
1352                let file = File::from_proto(file, worktree, cx)?;
1353                let old_file = buffer.update(cx, |buffer, cx| {
1354                    let old_file = buffer.file().cloned();
1355                    let new_path = file.path.clone();
1356
1357                    buffer.file_updated(Arc::new(file), cx);
1358                    if old_file.as_ref().is_none_or(|old| *old.path() != new_path) {
1359                        Some(old_file)
1360                    } else {
1361                        None
1362                    }
1363                });
1364                if let Some(old_file) = old_file {
1365                    cx.emit(BufferStoreEvent::BufferChangedFilePath { buffer, old_file });
1366                }
1367            }
1368            if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1369                downstream_client
1370                    .send(proto::UpdateBufferFile {
1371                        project_id: *project_id,
1372                        buffer_id: buffer_id.into(),
1373                        file: envelope.payload.file,
1374                    })
1375                    .log_err();
1376            }
1377            Ok(())
1378        })
1379    }
1380
1381    pub async fn handle_save_buffer(
1382        this: Entity<Self>,
1383        envelope: TypedEnvelope<proto::SaveBuffer>,
1384        mut cx: AsyncApp,
1385    ) -> Result<proto::BufferSaved> {
1386        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1387        let (buffer, project_id) = this.read_with(&cx, |this, _| {
1388            anyhow::Ok((
1389                this.get_existing(buffer_id)?,
1390                this.downstream_client
1391                    .as_ref()
1392                    .map(|(_, project_id)| *project_id)
1393                    .context("project is not shared")?,
1394            ))
1395        })?;
1396        buffer
1397            .update(&mut cx, |buffer, _| {
1398                buffer.wait_for_version(deserialize_version(&envelope.payload.version))
1399            })
1400            .await?;
1401        let buffer_id = buffer.read_with(&cx, |buffer, _| buffer.remote_id());
1402
1403        if let Some(new_path) = envelope.payload.new_path
1404            && let Some(new_path) = ProjectPath::from_proto(new_path)
1405        {
1406            this.update(&mut cx, |this, cx| {
1407                this.save_buffer_as(buffer.clone(), new_path, cx)
1408            })
1409            .await?;
1410        } else {
1411            this.update(&mut cx, |this, cx| this.save_buffer(buffer.clone(), cx))
1412                .await?;
1413        }
1414
1415        Ok(buffer.read_with(&cx, |buffer, _| proto::BufferSaved {
1416            project_id,
1417            buffer_id: buffer_id.into(),
1418            version: serialize_version(buffer.saved_version()),
1419            mtime: buffer.saved_mtime().map(|time| time.into()),
1420        }))
1421    }
1422
1423    pub async fn handle_close_buffer(
1424        this: Entity<Self>,
1425        envelope: TypedEnvelope<proto::CloseBuffer>,
1426        mut cx: AsyncApp,
1427    ) -> Result<()> {
1428        let peer_id = envelope.sender_id;
1429        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1430        this.update(&mut cx, |this, cx| {
1431            if let Some(shared) = this.shared_buffers.get_mut(&peer_id)
1432                && shared.remove(&buffer_id).is_some()
1433            {
1434                cx.emit(BufferStoreEvent::SharedBufferClosed(peer_id, buffer_id));
1435                if shared.is_empty() {
1436                    this.shared_buffers.remove(&peer_id);
1437                }
1438                return;
1439            }
1440            debug_panic!(
1441                "peer_id {} closed buffer_id {} which was either not open or already closed",
1442                peer_id,
1443                buffer_id
1444            )
1445        });
1446        Ok(())
1447    }
1448
1449    pub async fn handle_buffer_saved(
1450        this: Entity<Self>,
1451        envelope: TypedEnvelope<proto::BufferSaved>,
1452        mut cx: AsyncApp,
1453    ) -> Result<()> {
1454        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1455        let version = deserialize_version(&envelope.payload.version);
1456        let mtime = envelope.payload.mtime.clone().map(|time| time.into());
1457        this.update(&mut cx, move |this, cx| {
1458            if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1459                buffer.update(cx, |buffer, cx| {
1460                    buffer.did_save(version, mtime, cx);
1461                });
1462            }
1463
1464            if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1465                downstream_client
1466                    .send(proto::BufferSaved {
1467                        project_id: *project_id,
1468                        buffer_id: buffer_id.into(),
1469                        mtime: envelope.payload.mtime,
1470                        version: envelope.payload.version,
1471                    })
1472                    .log_err();
1473            }
1474        });
1475        Ok(())
1476    }
1477
1478    pub async fn handle_buffer_reloaded(
1479        this: Entity<Self>,
1480        envelope: TypedEnvelope<proto::BufferReloaded>,
1481        mut cx: AsyncApp,
1482    ) -> Result<()> {
1483        let buffer_id = BufferId::new(envelope.payload.buffer_id)?;
1484        let version = deserialize_version(&envelope.payload.version);
1485        let mtime = envelope.payload.mtime.clone().map(|time| time.into());
1486        let line_ending = deserialize_line_ending(
1487            proto::LineEnding::from_i32(envelope.payload.line_ending)
1488                .context("missing line ending")?,
1489        );
1490        this.update(&mut cx, |this, cx| {
1491            if let Some(buffer) = this.get_possibly_incomplete(buffer_id) {
1492                buffer.update(cx, |buffer, cx| {
1493                    buffer.did_reload(version, line_ending, mtime, cx);
1494                });
1495            }
1496
1497            if let Some((downstream_client, project_id)) = this.downstream_client.as_ref() {
1498                downstream_client
1499                    .send(proto::BufferReloaded {
1500                        project_id: *project_id,
1501                        buffer_id: buffer_id.into(),
1502                        mtime: envelope.payload.mtime,
1503                        version: envelope.payload.version,
1504                        line_ending: envelope.payload.line_ending,
1505                    })
1506                    .log_err();
1507            }
1508        });
1509        Ok(())
1510    }
1511
1512    pub fn reload_buffers(
1513        &self,
1514        buffers: HashSet<Entity<Buffer>>,
1515        push_to_history: bool,
1516        cx: &mut Context<Self>,
1517    ) -> Task<Result<ProjectTransaction>> {
1518        if buffers.is_empty() {
1519            return Task::ready(Ok(ProjectTransaction::default()));
1520        }
1521        match &self.state {
1522            BufferStoreState::Local(this) => this.reload_buffers(buffers, push_to_history, cx),
1523            BufferStoreState::Remote(this) => this.reload_buffers(buffers, push_to_history, cx),
1524        }
1525    }
1526
1527    async fn handle_reload_buffers(
1528        this: Entity<Self>,
1529        envelope: TypedEnvelope<proto::ReloadBuffers>,
1530        mut cx: AsyncApp,
1531    ) -> Result<proto::ReloadBuffersResponse> {
1532        let sender_id = envelope.original_sender_id().unwrap_or_default();
1533        let reload = this.update(&mut cx, |this, cx| {
1534            let mut buffers = HashSet::default();
1535            for buffer_id in &envelope.payload.buffer_ids {
1536                let buffer_id = BufferId::new(*buffer_id)?;
1537                buffers.insert(this.get_existing(buffer_id)?);
1538            }
1539            anyhow::Ok(this.reload_buffers(buffers, false, cx))
1540        })?;
1541
1542        let project_transaction = reload.await?;
1543        let project_transaction = this.update(&mut cx, |this, cx| {
1544            this.serialize_project_transaction_for_peer(project_transaction, sender_id, cx)
1545        });
1546        Ok(proto::ReloadBuffersResponse {
1547            transaction: Some(project_transaction),
1548        })
1549    }
1550
1551    pub fn create_buffer_for_peer(
1552        &mut self,
1553        buffer: &Entity<Buffer>,
1554        peer_id: proto::PeerId,
1555        cx: &mut Context<Self>,
1556    ) -> Task<Result<()>> {
1557        let buffer_id = buffer.read(cx).remote_id();
1558        let shared_buffers = self.shared_buffers.entry(peer_id).or_default();
1559        if shared_buffers.contains_key(&buffer_id) {
1560            return Task::ready(Ok(()));
1561        }
1562        shared_buffers.insert(
1563            buffer_id,
1564            SharedBuffer {
1565                buffer: buffer.clone(),
1566                lsp_handle: None,
1567            },
1568        );
1569
1570        let Some((client, project_id)) = self.downstream_client.clone() else {
1571            return Task::ready(Ok(()));
1572        };
1573
1574        cx.spawn(async move |this, cx| {
1575            let Some(buffer) = this.read_with(cx, |this, _| this.get(buffer_id))? else {
1576                return anyhow::Ok(());
1577            };
1578
1579            let operations = buffer.update(cx, |b, cx| b.serialize_ops(None, cx));
1580            let operations = operations.await;
1581            let state = buffer.update(cx, |buffer, cx| buffer.to_proto(cx));
1582
1583            let initial_state = proto::CreateBufferForPeer {
1584                project_id,
1585                peer_id: Some(peer_id),
1586                variant: Some(proto::create_buffer_for_peer::Variant::State(state)),
1587            };
1588
1589            if client.send(initial_state).log_err().is_some() {
1590                let client = client.clone();
1591                cx.background_spawn(async move {
1592                    let mut chunks = split_operations(operations).peekable();
1593                    while let Some(chunk) = chunks.next() {
1594                        let is_last = chunks.peek().is_none();
1595                        client.send(proto::CreateBufferForPeer {
1596                            project_id,
1597                            peer_id: Some(peer_id),
1598                            variant: Some(proto::create_buffer_for_peer::Variant::Chunk(
1599                                proto::BufferChunk {
1600                                    buffer_id: buffer_id.into(),
1601                                    operations: chunk,
1602                                    is_last,
1603                                },
1604                            )),
1605                        })?;
1606                    }
1607                    anyhow::Ok(())
1608                })
1609                .await
1610                .log_err();
1611            }
1612            Ok(())
1613        })
1614    }
1615
1616    pub fn forget_shared_buffers(&mut self) {
1617        self.shared_buffers.clear();
1618    }
1619
1620    pub fn forget_shared_buffers_for(&mut self, peer_id: &proto::PeerId) {
1621        self.shared_buffers.remove(peer_id);
1622    }
1623
1624    pub fn update_peer_id(&mut self, old_peer_id: &proto::PeerId, new_peer_id: proto::PeerId) {
1625        if let Some(buffers) = self.shared_buffers.remove(old_peer_id) {
1626            self.shared_buffers.insert(new_peer_id, buffers);
1627        }
1628    }
1629
1630    pub fn has_shared_buffers(&self) -> bool {
1631        !self.shared_buffers.is_empty()
1632    }
1633
1634    pub fn create_local_buffer(
1635        &mut self,
1636        text: &str,
1637        language: Option<Arc<Language>>,
1638        project_searchable: bool,
1639        cx: &mut Context<Self>,
1640    ) -> Entity<Buffer> {
1641        let buffer = cx.new(|cx| {
1642            let mut buffer = Buffer::local(text, cx)
1643                .with_language(language.unwrap_or_else(|| language::PLAIN_TEXT.clone()), cx);
1644            apply_initial_line_ending(&mut buffer, cx);
1645            buffer
1646        });
1647
1648        self.add_buffer(buffer.clone(), cx).log_err();
1649        let buffer_id = buffer.read(cx).remote_id();
1650        if !project_searchable {
1651            self.non_searchable_buffers.insert(buffer_id);
1652        }
1653
1654        if let Some(file) = File::from_dyn(buffer.read(cx).file()) {
1655            self.path_to_buffer_id.insert(
1656                ProjectPath {
1657                    worktree_id: file.worktree_id(cx),
1658                    path: file.path.clone(),
1659                },
1660                buffer_id,
1661            );
1662            let this = self
1663                .as_local_mut()
1664                .expect("local-only method called in a non-local context");
1665            if let Some(entry_id) = file.entry_id {
1666                this.local_buffer_ids_by_entry_id
1667                    .insert(entry_id, buffer_id);
1668            }
1669        }
1670        buffer
1671    }
1672
1673    pub fn deserialize_project_transaction(
1674        &mut self,
1675        message: proto::ProjectTransaction,
1676        push_to_history: bool,
1677        cx: &mut Context<Self>,
1678    ) -> Task<Result<ProjectTransaction>> {
1679        if let Some(this) = self.as_remote_mut() {
1680            this.deserialize_project_transaction(message, push_to_history, cx)
1681        } else {
1682            debug_panic!("not a remote buffer store");
1683            Task::ready(Err(anyhow!("not a remote buffer store")))
1684        }
1685    }
1686
1687    pub fn wait_for_remote_buffer(
1688        &mut self,
1689        id: BufferId,
1690        cx: &mut Context<BufferStore>,
1691    ) -> Task<Result<Entity<Buffer>>> {
1692        if let Some(this) = self.as_remote_mut() {
1693            this.wait_for_remote_buffer(id, cx)
1694        } else {
1695            debug_panic!("not a remote buffer store");
1696            Task::ready(Err(anyhow!("not a remote buffer store")))
1697        }
1698    }
1699
1700    pub fn serialize_project_transaction_for_peer(
1701        &mut self,
1702        project_transaction: ProjectTransaction,
1703        peer_id: proto::PeerId,
1704        cx: &mut Context<Self>,
1705    ) -> proto::ProjectTransaction {
1706        let mut serialized_transaction = proto::ProjectTransaction {
1707            buffer_ids: Default::default(),
1708            transactions: Default::default(),
1709        };
1710        for (buffer, transaction) in project_transaction.0 {
1711            self.create_buffer_for_peer(&buffer, peer_id, cx)
1712                .detach_and_log_err(cx);
1713            serialized_transaction
1714                .buffer_ids
1715                .push(buffer.read(cx).remote_id().into());
1716            serialized_transaction
1717                .transactions
1718                .push(language::proto::serialize_transaction(&transaction));
1719        }
1720        serialized_transaction
1721    }
1722
1723    pub(crate) fn register_project_search_result_handle(
1724        &mut self,
1725    ) -> (u64, async_channel::Receiver<BufferId>) {
1726        let (tx, rx) = async_channel::unbounded();
1727        let handle = util::post_inc(&mut self.project_search.next_id);
1728        let _old_entry = self.project_search.chunks.insert(handle, tx);
1729        debug_assert!(_old_entry.is_none());
1730        (handle, rx)
1731    }
1732
1733    pub fn register_ongoing_project_search(
1734        &mut self,
1735        id: (PeerId, u64),
1736        search: Task<anyhow::Result<()>>,
1737    ) {
1738        let _old = self.project_search.searches_in_progress.insert(id, search);
1739        debug_assert!(_old.is_none());
1740    }
1741
1742    pub async fn handle_find_search_candidates_cancel(
1743        this: Entity<Self>,
1744        envelope: TypedEnvelope<proto::FindSearchCandidatesCancelled>,
1745        mut cx: AsyncApp,
1746    ) -> Result<()> {
1747        let id = (
1748            envelope.original_sender_id.unwrap_or(envelope.sender_id),
1749            envelope.payload.handle,
1750        );
1751        let _ = this.update(&mut cx, |this, _| {
1752            this.project_search.searches_in_progress.remove(&id)
1753        });
1754        Ok(())
1755    }
1756
1757    pub(crate) async fn handle_find_search_candidates_chunk(
1758        this: Entity<Self>,
1759        envelope: TypedEnvelope<proto::FindSearchCandidatesChunk>,
1760        mut cx: AsyncApp,
1761    ) -> Result<proto::Ack> {
1762        use proto::find_search_candidates_chunk::Variant;
1763        let handle = envelope.payload.handle;
1764
1765        let buffer_ids = match envelope
1766            .payload
1767            .variant
1768            .context("Expected non-null variant")?
1769        {
1770            Variant::Matches(find_search_candidates_matches) => find_search_candidates_matches
1771                .buffer_ids
1772                .into_iter()
1773                .filter_map(|buffer_id| BufferId::new(buffer_id).ok())
1774                .collect::<Vec<_>>(),
1775            Variant::Done(_) => {
1776                this.update(&mut cx, |this, _| {
1777                    this.project_search.chunks.remove(&handle)
1778                });
1779                return Ok(proto::Ack {});
1780            }
1781        };
1782        let Some(sender) = this.read_with(&mut cx, |this, _| {
1783            this.project_search.chunks.get(&handle).cloned()
1784        }) else {
1785            return Ok(proto::Ack {});
1786        };
1787
1788        for buffer_id in buffer_ids {
1789            let Ok(_) = sender.send(buffer_id).await else {
1790                this.update(&mut cx, |this, _| {
1791                    this.project_search.chunks.remove(&handle)
1792                });
1793                return Ok(proto::Ack {});
1794            };
1795        }
1796        Ok(proto::Ack {})
1797    }
1798}
1799
1800impl OpenBuffer {
1801    fn upgrade(&self) -> Option<Entity<Buffer>> {
1802        match self {
1803            OpenBuffer::Complete { buffer, .. } => buffer.upgrade(),
1804            OpenBuffer::Operations(_) => None,
1805        }
1806    }
1807}
1808
1809fn is_not_found_error(error: &anyhow::Error) -> bool {
1810    error
1811        .root_cause()
1812        .downcast_ref::<io::Error>()
1813        .is_some_and(|err| err.kind() == io::ErrorKind::NotFound)
1814}
1815
1816fn apply_initial_line_ending(buffer: &mut Buffer, cx: &mut Context<Buffer>) {
1817    // Only applies for empty rope or a single line with no trailing newline.
1818    if buffer.max_point().row > 0 {
1819        return;
1820    }
1821    let location = buffer.file().map(|file| settings::SettingsLocation {
1822        worktree_id: file.worktree_id(cx),
1823        path: file.path().as_ref(),
1824    });
1825    let language = buffer.language().map(|l| l.name());
1826    let settings = AllLanguageSettings::get(location, cx).language(location, language.as_ref(), cx);
1827    let desired = match settings.line_ending {
1828        LineEndingSetting::Detect => return,
1829        LineEndingSetting::PreferLf | LineEndingSetting::EnforceLf => LineEnding::Unix,
1830        LineEndingSetting::PreferCrlf | LineEndingSetting::EnforceCrlf => LineEnding::Windows,
1831    };
1832    if buffer.line_ending() != desired {
1833        buffer.set_line_ending(desired, cx);
1834    }
1835}
1836