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