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