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