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