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