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