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