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