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