1use ::proto::{FromProto, ToProto};
2use anyhow::{Context as _, Result, anyhow};
3
4use extension::ExtensionHostProxy;
5use extension_host::headless_host::HeadlessExtensionStore;
6use fs::Fs;
7use gpui::{App, AppContext as _, AsyncApp, Context, Entity, PromptLevel};
8use http_client::HttpClient;
9use language::{Buffer, BufferEvent, LanguageRegistry, proto::serialize_operation};
10use node_runtime::NodeRuntime;
11use project::{
12 LspStore, LspStoreEvent, ManifestTree, PrettierStore, ProjectEnvironment, ProjectPath,
13 ToolchainStore, WorktreeId,
14 buffer_store::{BufferStore, BufferStoreEvent},
15 debugger::{breakpoint_store::BreakpointStore, dap_store::DapStore},
16 git_store::GitStore,
17 project_settings::SettingsObserver,
18 search::SearchQuery,
19 task_store::TaskStore,
20 worktree_store::WorktreeStore,
21};
22use rpc::{
23 AnyProtoClient, TypedEnvelope,
24 proto::{self, REMOTE_SERVER_PEER_ID, REMOTE_SERVER_PROJECT_ID},
25};
26
27use settings::initial_server_settings_content;
28use smol::stream::StreamExt;
29use std::{
30 path::{Path, PathBuf},
31 sync::{Arc, atomic::AtomicUsize},
32};
33use util::ResultExt;
34use worktree::Worktree;
35
36pub struct HeadlessProject {
37 pub fs: Arc<dyn Fs>,
38 pub session: AnyProtoClient,
39 pub worktree_store: Entity<WorktreeStore>,
40 pub buffer_store: Entity<BufferStore>,
41 pub lsp_store: Entity<LspStore>,
42 pub task_store: Entity<TaskStore>,
43 pub dap_store: Entity<DapStore>,
44 pub settings_observer: Entity<SettingsObserver>,
45 pub next_entry_id: Arc<AtomicUsize>,
46 pub languages: Arc<LanguageRegistry>,
47 pub extensions: Entity<HeadlessExtensionStore>,
48 pub git_store: Entity<GitStore>,
49 // Used mostly to keep alive the toolchain store for RPC handlers.
50 // Local variant is used within LSP store, but that's a separate entity.
51 pub _toolchain_store: Entity<ToolchainStore>,
52}
53
54pub struct HeadlessAppState {
55 pub session: AnyProtoClient,
56 pub fs: Arc<dyn Fs>,
57 pub http_client: Arc<dyn HttpClient>,
58 pub node_runtime: NodeRuntime,
59 pub languages: Arc<LanguageRegistry>,
60 pub extension_host_proxy: Arc<ExtensionHostProxy>,
61}
62
63impl HeadlessProject {
64 pub fn init(cx: &mut App) {
65 settings::init(cx);
66 language::init(cx);
67 project::Project::init_settings(cx);
68 }
69
70 pub fn new(
71 HeadlessAppState {
72 session,
73 fs,
74 http_client,
75 node_runtime,
76 languages,
77 extension_host_proxy: proxy,
78 }: HeadlessAppState,
79 cx: &mut Context<Self>,
80 ) -> Self {
81 debug_adapter_extension::init(proxy.clone(), cx);
82 languages::init(languages.clone(), node_runtime.clone(), cx);
83
84 let worktree_store = cx.new(|cx| {
85 let mut store = WorktreeStore::local(true, fs.clone());
86 store.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
87 store
88 });
89
90 let environment = cx.new(|_| ProjectEnvironment::new(None));
91 let manifest_tree = ManifestTree::new(worktree_store.clone(), cx);
92 let toolchain_store = cx.new(|cx| {
93 ToolchainStore::local(
94 languages.clone(),
95 worktree_store.clone(),
96 environment.clone(),
97 manifest_tree.clone(),
98 cx,
99 )
100 });
101
102 let buffer_store = cx.new(|cx| {
103 let mut buffer_store = BufferStore::local(worktree_store.clone(), cx);
104 buffer_store.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
105 buffer_store
106 });
107
108 let breakpoint_store =
109 cx.new(|_| BreakpointStore::local(worktree_store.clone(), buffer_store.clone()));
110
111 let dap_store = cx.new(|cx| {
112 let mut dap_store = DapStore::new_local(
113 http_client.clone(),
114 node_runtime.clone(),
115 fs.clone(),
116 environment.clone(),
117 toolchain_store.read(cx).as_language_toolchain_store(),
118 worktree_store.clone(),
119 breakpoint_store.clone(),
120 cx,
121 );
122 dap_store.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
123 dap_store
124 });
125
126 let git_store = cx.new(|cx| {
127 let mut store = GitStore::local(
128 &worktree_store,
129 buffer_store.clone(),
130 environment.clone(),
131 fs.clone(),
132 cx,
133 );
134 store.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
135 store
136 });
137
138 let prettier_store = cx.new(|cx| {
139 PrettierStore::new(
140 node_runtime.clone(),
141 fs.clone(),
142 languages.clone(),
143 worktree_store.clone(),
144 cx,
145 )
146 });
147
148 let task_store = cx.new(|cx| {
149 let mut task_store = TaskStore::local(
150 fs.clone(),
151 buffer_store.downgrade(),
152 worktree_store.clone(),
153 toolchain_store.read(cx).as_language_toolchain_store(),
154 environment.clone(),
155 cx,
156 );
157 task_store.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
158 task_store
159 });
160 let settings_observer = cx.new(|cx| {
161 let mut observer = SettingsObserver::new_local(
162 fs.clone(),
163 worktree_store.clone(),
164 task_store.clone(),
165 cx,
166 );
167 observer.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
168 observer
169 });
170
171 let lsp_store = cx.new(|cx| {
172 let mut lsp_store = LspStore::new_local(
173 buffer_store.clone(),
174 worktree_store.clone(),
175 prettier_store.clone(),
176 toolchain_store
177 .read(cx)
178 .as_local_store()
179 .expect("Toolchain store to be local")
180 .clone(),
181 environment,
182 manifest_tree,
183 languages.clone(),
184 http_client.clone(),
185 fs.clone(),
186 cx,
187 );
188 lsp_store.shared(REMOTE_SERVER_PROJECT_ID, session.clone(), cx);
189 lsp_store
190 });
191
192 cx.subscribe(&lsp_store, Self::on_lsp_store_event).detach();
193 language_extension::init(
194 language_extension::LspAccess::ViaLspStore(lsp_store.clone()),
195 proxy.clone(),
196 languages.clone(),
197 );
198
199 cx.subscribe(&buffer_store, |_this, _buffer_store, event, cx| {
200 if let BufferStoreEvent::BufferAdded(buffer) = event {
201 cx.subscribe(buffer, Self::on_buffer_event).detach();
202 }
203 })
204 .detach();
205
206 let extensions = HeadlessExtensionStore::new(
207 fs.clone(),
208 http_client.clone(),
209 paths::remote_extensions_dir().to_path_buf(),
210 proxy,
211 node_runtime,
212 cx,
213 );
214
215 // local_machine -> ssh handlers
216 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &worktree_store);
217 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &buffer_store);
218 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &cx.entity());
219 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &lsp_store);
220 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &task_store);
221 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &toolchain_store);
222 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &dap_store);
223 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &settings_observer);
224 session.subscribe_to_entity(REMOTE_SERVER_PROJECT_ID, &git_store);
225
226 session.add_request_handler(cx.weak_entity(), Self::handle_list_remote_directory);
227 session.add_request_handler(cx.weak_entity(), Self::handle_get_path_metadata);
228 session.add_request_handler(cx.weak_entity(), Self::handle_shutdown_remote_server);
229 session.add_request_handler(cx.weak_entity(), Self::handle_ping);
230
231 session.add_entity_request_handler(Self::handle_add_worktree);
232 session.add_request_handler(cx.weak_entity(), Self::handle_remove_worktree);
233
234 session.add_entity_request_handler(Self::handle_open_buffer_by_path);
235 session.add_entity_request_handler(Self::handle_open_new_buffer);
236 session.add_entity_request_handler(Self::handle_find_search_candidates);
237 session.add_entity_request_handler(Self::handle_open_server_settings);
238
239 session.add_entity_request_handler(BufferStore::handle_update_buffer);
240 session.add_entity_message_handler(BufferStore::handle_close_buffer);
241
242 session.add_request_handler(
243 extensions.downgrade(),
244 HeadlessExtensionStore::handle_sync_extensions,
245 );
246 session.add_request_handler(
247 extensions.downgrade(),
248 HeadlessExtensionStore::handle_install_extension,
249 );
250
251 BufferStore::init(&session);
252 WorktreeStore::init(&session);
253 SettingsObserver::init(&session);
254 LspStore::init(&session);
255 TaskStore::init(Some(&session));
256 ToolchainStore::init(&session);
257 DapStore::init(&session, cx);
258 // todo(debugger): Re init breakpoint store when we set it up for collab
259 // BreakpointStore::init(&client);
260 GitStore::init(&session);
261
262 HeadlessProject {
263 next_entry_id: Default::default(),
264 session,
265 settings_observer,
266 fs,
267 worktree_store,
268 buffer_store,
269 lsp_store,
270 task_store,
271 dap_store,
272 languages,
273 extensions,
274 git_store,
275 _toolchain_store: toolchain_store,
276 }
277 }
278
279 fn on_buffer_event(
280 &mut self,
281 buffer: Entity<Buffer>,
282 event: &BufferEvent,
283 cx: &mut Context<Self>,
284 ) {
285 if let BufferEvent::Operation {
286 operation,
287 is_local: true,
288 } = event
289 {
290 cx.background_spawn(self.session.request(proto::UpdateBuffer {
291 project_id: REMOTE_SERVER_PROJECT_ID,
292 buffer_id: buffer.read(cx).remote_id().to_proto(),
293 operations: vec![serialize_operation(operation)],
294 }))
295 .detach()
296 }
297 }
298
299 fn on_lsp_store_event(
300 &mut self,
301 _lsp_store: Entity<LspStore>,
302 event: &LspStoreEvent,
303 cx: &mut Context<Self>,
304 ) {
305 match event {
306 LspStoreEvent::LanguageServerUpdate {
307 language_server_id,
308 name,
309 message,
310 } => {
311 self.session
312 .send(proto::UpdateLanguageServer {
313 project_id: REMOTE_SERVER_PROJECT_ID,
314 server_name: name.as_ref().map(|name| name.to_string()),
315 language_server_id: language_server_id.to_proto(),
316 variant: Some(message.clone()),
317 })
318 .log_err();
319 }
320 LspStoreEvent::Notification(message) => {
321 self.session
322 .send(proto::Toast {
323 project_id: REMOTE_SERVER_PROJECT_ID,
324 notification_id: "lsp".to_string(),
325 message: message.clone(),
326 })
327 .log_err();
328 }
329 LspStoreEvent::LanguageServerLog(language_server_id, log_type, message) => {
330 self.session
331 .send(proto::LanguageServerLog {
332 project_id: REMOTE_SERVER_PROJECT_ID,
333 language_server_id: language_server_id.to_proto(),
334 message: message.clone(),
335 log_type: Some(log_type.to_proto()),
336 })
337 .log_err();
338 }
339 LspStoreEvent::LanguageServerPrompt(prompt) => {
340 let request = self.session.request(proto::LanguageServerPromptRequest {
341 project_id: REMOTE_SERVER_PROJECT_ID,
342 actions: prompt
343 .actions
344 .iter()
345 .map(|action| action.title.to_string())
346 .collect(),
347 level: Some(prompt_to_proto(prompt)),
348 lsp_name: prompt.lsp_name.clone(),
349 message: prompt.message.clone(),
350 });
351 let prompt = prompt.clone();
352 cx.background_spawn(async move {
353 let response = request.await?;
354 if let Some(action_response) = response.action_response {
355 prompt.respond(action_response as usize).await;
356 }
357 anyhow::Ok(())
358 })
359 .detach();
360 }
361 _ => {}
362 }
363 }
364
365 pub async fn handle_add_worktree(
366 this: Entity<Self>,
367 message: TypedEnvelope<proto::AddWorktree>,
368 mut cx: AsyncApp,
369 ) -> Result<proto::AddWorktreeResponse> {
370 use client::ErrorCodeExt;
371 let fs = this.read_with(&cx, |this, _| this.fs.clone())?;
372 let path = PathBuf::from_proto(shellexpand::tilde(&message.payload.path).to_string());
373
374 let canonicalized = match fs.canonicalize(&path).await {
375 Ok(path) => path,
376 Err(e) => {
377 let mut parent = path
378 .parent()
379 .ok_or(e)
380 .with_context(|| format!("{path:?} does not exist"))?;
381 if parent == Path::new("") {
382 parent = util::paths::home_dir();
383 }
384 let parent = fs.canonicalize(parent).await.map_err(|_| {
385 anyhow!(
386 proto::ErrorCode::DevServerProjectPathDoesNotExist
387 .with_tag("path", path.to_string_lossy().as_ref())
388 )
389 })?;
390 parent.join(path.file_name().unwrap())
391 }
392 };
393
394 let worktree = this
395 .read_with(&cx.clone(), |this, _| {
396 Worktree::local(
397 Arc::from(canonicalized.as_path()),
398 message.payload.visible,
399 this.fs.clone(),
400 this.next_entry_id.clone(),
401 &mut cx,
402 )
403 })?
404 .await?;
405
406 let response = this.read_with(&cx, |_, cx| {
407 let worktree = worktree.read(cx);
408 proto::AddWorktreeResponse {
409 worktree_id: worktree.id().to_proto(),
410 canonicalized_path: canonicalized.to_proto(),
411 }
412 })?;
413
414 // We spawn this asynchronously, so that we can send the response back
415 // *before* `worktree_store.add()` can send out UpdateProject requests
416 // to the client about the new worktree.
417 //
418 // That lets the client manage the reference/handles of the newly-added
419 // worktree, before getting interrupted by an UpdateProject request.
420 //
421 // This fixes the problem of the client sending the AddWorktree request,
422 // headless project sending out a project update, client receiving it
423 // and immediately dropping the reference of the new client, causing it
424 // to be dropped on the headless project, and the client only then
425 // receiving a response to AddWorktree.
426 cx.spawn(async move |cx| {
427 this.update(cx, |this, cx| {
428 this.worktree_store.update(cx, |worktree_store, cx| {
429 worktree_store.add(&worktree, cx);
430 });
431 })
432 .log_err();
433 })
434 .detach();
435
436 Ok(response)
437 }
438
439 pub async fn handle_remove_worktree(
440 this: Entity<Self>,
441 envelope: TypedEnvelope<proto::RemoveWorktree>,
442 mut cx: AsyncApp,
443 ) -> Result<proto::Ack> {
444 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
445 this.update(&mut cx, |this, cx| {
446 this.worktree_store.update(cx, |worktree_store, cx| {
447 worktree_store.remove_worktree(worktree_id, cx);
448 });
449 })?;
450 Ok(proto::Ack {})
451 }
452
453 pub async fn handle_open_buffer_by_path(
454 this: Entity<Self>,
455 message: TypedEnvelope<proto::OpenBufferByPath>,
456 mut cx: AsyncApp,
457 ) -> Result<proto::OpenBufferResponse> {
458 let worktree_id = WorktreeId::from_proto(message.payload.worktree_id);
459 let (buffer_store, buffer) = this.update(&mut cx, |this, cx| {
460 let buffer_store = this.buffer_store.clone();
461 let buffer = this.buffer_store.update(cx, |buffer_store, cx| {
462 buffer_store.open_buffer(
463 ProjectPath {
464 worktree_id,
465 path: Arc::<Path>::from_proto(message.payload.path),
466 },
467 cx,
468 )
469 });
470 anyhow::Ok((buffer_store, buffer))
471 })??;
472
473 let buffer = buffer.await?;
474 let buffer_id = buffer.read_with(&cx, |b, _| b.remote_id())?;
475 buffer_store.update(&mut cx, |buffer_store, cx| {
476 buffer_store
477 .create_buffer_for_peer(&buffer, REMOTE_SERVER_PEER_ID, cx)
478 .detach_and_log_err(cx);
479 })?;
480
481 Ok(proto::OpenBufferResponse {
482 buffer_id: buffer_id.to_proto(),
483 })
484 }
485
486 pub async fn handle_open_new_buffer(
487 this: Entity<Self>,
488 _message: TypedEnvelope<proto::OpenNewBuffer>,
489 mut cx: AsyncApp,
490 ) -> Result<proto::OpenBufferResponse> {
491 let (buffer_store, buffer) = this.update(&mut cx, |this, cx| {
492 let buffer_store = this.buffer_store.clone();
493 let buffer = this
494 .buffer_store
495 .update(cx, |buffer_store, cx| buffer_store.create_buffer(cx));
496 anyhow::Ok((buffer_store, buffer))
497 })??;
498
499 let buffer = buffer.await?;
500 let buffer_id = buffer.read_with(&cx, |b, _| b.remote_id())?;
501 buffer_store.update(&mut cx, |buffer_store, cx| {
502 buffer_store
503 .create_buffer_for_peer(&buffer, REMOTE_SERVER_PEER_ID, cx)
504 .detach_and_log_err(cx);
505 })?;
506
507 Ok(proto::OpenBufferResponse {
508 buffer_id: buffer_id.to_proto(),
509 })
510 }
511
512 pub async fn handle_open_server_settings(
513 this: Entity<Self>,
514 _: TypedEnvelope<proto::OpenServerSettings>,
515 mut cx: AsyncApp,
516 ) -> Result<proto::OpenBufferResponse> {
517 let settings_path = paths::settings_file();
518 let (worktree, path) = this
519 .update(&mut cx, |this, cx| {
520 this.worktree_store.update(cx, |worktree_store, cx| {
521 worktree_store.find_or_create_worktree(settings_path, false, cx)
522 })
523 })?
524 .await?;
525
526 let (buffer, buffer_store) = this.update(&mut cx, |this, cx| {
527 let buffer = this.buffer_store.update(cx, |buffer_store, cx| {
528 buffer_store.open_buffer(
529 ProjectPath {
530 worktree_id: worktree.read(cx).id(),
531 path: path.into(),
532 },
533 cx,
534 )
535 });
536
537 (buffer, this.buffer_store.clone())
538 })?;
539
540 let buffer = buffer.await?;
541
542 let buffer_id = cx.update(|cx| {
543 if buffer.read(cx).is_empty() {
544 buffer.update(cx, |buffer, cx| {
545 buffer.edit([(0..0, initial_server_settings_content())], None, cx)
546 });
547 }
548
549 let buffer_id = buffer.read(cx).remote_id();
550
551 buffer_store.update(cx, |buffer_store, cx| {
552 buffer_store
553 .create_buffer_for_peer(&buffer, REMOTE_SERVER_PEER_ID, cx)
554 .detach_and_log_err(cx);
555 });
556
557 buffer_id
558 })?;
559
560 Ok(proto::OpenBufferResponse {
561 buffer_id: buffer_id.to_proto(),
562 })
563 }
564
565 pub async fn handle_find_search_candidates(
566 this: Entity<Self>,
567 envelope: TypedEnvelope<proto::FindSearchCandidates>,
568 mut cx: AsyncApp,
569 ) -> Result<proto::FindSearchCandidatesResponse> {
570 let message = envelope.payload;
571 let query = SearchQuery::from_proto(message.query.context("missing query field")?)?;
572 let results = this.update(&mut cx, |this, cx| {
573 this.buffer_store.update(cx, |buffer_store, cx| {
574 buffer_store.find_search_candidates(&query, message.limit as _, this.fs.clone(), cx)
575 })
576 })?;
577
578 let mut response = proto::FindSearchCandidatesResponse {
579 buffer_ids: Vec::new(),
580 };
581
582 let buffer_store = this.read_with(&cx, |this, _| this.buffer_store.clone())?;
583
584 while let Ok(buffer) = results.recv().await {
585 let buffer_id = buffer.read_with(&cx, |this, _| this.remote_id())?;
586 response.buffer_ids.push(buffer_id.to_proto());
587 buffer_store
588 .update(&mut cx, |buffer_store, cx| {
589 buffer_store.create_buffer_for_peer(&buffer, REMOTE_SERVER_PEER_ID, cx)
590 })?
591 .await?;
592 }
593
594 Ok(response)
595 }
596
597 pub async fn handle_list_remote_directory(
598 this: Entity<Self>,
599 envelope: TypedEnvelope<proto::ListRemoteDirectory>,
600 cx: AsyncApp,
601 ) -> Result<proto::ListRemoteDirectoryResponse> {
602 let fs = cx.read_entity(&this, |this, _| this.fs.clone())?;
603 let expanded = PathBuf::from_proto(shellexpand::tilde(&envelope.payload.path).to_string());
604 let check_info = envelope
605 .payload
606 .config
607 .as_ref()
608 .is_some_and(|config| config.is_dir);
609
610 let mut entries = Vec::new();
611 let mut entry_info = Vec::new();
612 let mut response = fs.read_dir(&expanded).await?;
613 while let Some(path) = response.next().await {
614 let path = path?;
615 if let Some(file_name) = path.file_name() {
616 entries.push(file_name.to_string_lossy().to_string());
617 if check_info {
618 let is_dir = fs.is_dir(&path).await;
619 entry_info.push(proto::EntryInfo { is_dir });
620 }
621 }
622 }
623 Ok(proto::ListRemoteDirectoryResponse {
624 entries,
625 entry_info,
626 })
627 }
628
629 pub async fn handle_get_path_metadata(
630 this: Entity<Self>,
631 envelope: TypedEnvelope<proto::GetPathMetadata>,
632 cx: AsyncApp,
633 ) -> Result<proto::GetPathMetadataResponse> {
634 let fs = cx.read_entity(&this, |this, _| this.fs.clone())?;
635 let expanded = PathBuf::from_proto(shellexpand::tilde(&envelope.payload.path).to_string());
636
637 let metadata = fs.metadata(&expanded).await?;
638 let is_dir = metadata.map(|metadata| metadata.is_dir).unwrap_or(false);
639
640 Ok(proto::GetPathMetadataResponse {
641 exists: metadata.is_some(),
642 is_dir,
643 path: expanded.to_proto(),
644 })
645 }
646
647 pub async fn handle_shutdown_remote_server(
648 _this: Entity<Self>,
649 _envelope: TypedEnvelope<proto::ShutdownRemoteServer>,
650 cx: AsyncApp,
651 ) -> Result<proto::Ack> {
652 cx.spawn(async move |cx| {
653 cx.update(|cx| {
654 // TODO: This is a hack, because in a headless project, shutdown isn't executed
655 // when calling quit, but it should be.
656 cx.shutdown();
657 cx.quit();
658 })
659 })
660 .detach();
661
662 Ok(proto::Ack {})
663 }
664
665 pub async fn handle_ping(
666 _this: Entity<Self>,
667 _envelope: TypedEnvelope<proto::Ping>,
668 _cx: AsyncApp,
669 ) -> Result<proto::Ack> {
670 log::debug!("Received ping from client");
671 Ok(proto::Ack {})
672 }
673}
674
675fn prompt_to_proto(
676 prompt: &project::LanguageServerPromptRequest,
677) -> proto::language_server_prompt_request::Level {
678 match prompt.level {
679 PromptLevel::Info => proto::language_server_prompt_request::Level::Info(
680 proto::language_server_prompt_request::Info {},
681 ),
682 PromptLevel::Warning => proto::language_server_prompt_request::Level::Warning(
683 proto::language_server_prompt_request::Warning {},
684 ),
685 PromptLevel::Critical => proto::language_server_prompt_request::Level::Critical(
686 proto::language_server_prompt_request::Critical {},
687 ),
688 }
689}