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