1use anyhow::{anyhow, Result};
2use fs::Fs;
3use gpui::{AppContext, AsyncAppContext, Context, Model, ModelContext};
4use language::{proto::serialize_operation, Buffer, BufferEvent, LanguageRegistry};
5use node_runtime::NodeRuntime;
6use project::{
7 buffer_store::{BufferStore, BufferStoreEvent},
8 project_settings::SettingsObserver,
9 search::SearchQuery,
10 task_store::TaskStore,
11 worktree_store::WorktreeStore,
12 LspStore, LspStoreEvent, PrettierStore, ProjectPath, WorktreeId,
13};
14use remote::ssh_session::ChannelClient;
15use rpc::{
16 proto::{self, SSH_PEER_ID, SSH_PROJECT_ID},
17 AnyProtoClient, TypedEnvelope,
18};
19use smol::stream::StreamExt;
20use std::{
21 path::{Path, PathBuf},
22 sync::{atomic::AtomicUsize, Arc},
23};
24use util::ResultExt;
25use worktree::Worktree;
26
27pub struct HeadlessProject {
28 pub fs: Arc<dyn Fs>,
29 pub session: AnyProtoClient,
30 pub worktree_store: Model<WorktreeStore>,
31 pub buffer_store: Model<BufferStore>,
32 pub lsp_store: Model<LspStore>,
33 pub task_store: Model<TaskStore>,
34 pub settings_observer: Model<SettingsObserver>,
35 pub next_entry_id: Arc<AtomicUsize>,
36 pub languages: Arc<LanguageRegistry>,
37}
38
39impl HeadlessProject {
40 pub fn init(cx: &mut AppContext) {
41 settings::init(cx);
42 language::init(cx);
43 project::Project::init_settings(cx);
44 }
45
46 pub fn new(session: Arc<ChannelClient>, fs: Arc<dyn Fs>, cx: &mut ModelContext<Self>) -> Self {
47 let languages = Arc::new(LanguageRegistry::new(cx.background_executor().clone()));
48
49 let node_runtime = NodeRuntime::unavailable();
50
51 languages::init(languages.clone(), node_runtime.clone(), cx);
52
53 let worktree_store = cx.new_model(|cx| {
54 let mut store = WorktreeStore::local(true, fs.clone());
55 store.shared(SSH_PROJECT_ID, session.clone().into(), cx);
56 store
57 });
58 let buffer_store = cx.new_model(|cx| {
59 let mut buffer_store = BufferStore::local(worktree_store.clone(), cx);
60 buffer_store.shared(SSH_PROJECT_ID, session.clone().into(), cx);
61 buffer_store
62 });
63 let prettier_store = cx.new_model(|cx| {
64 PrettierStore::new(
65 node_runtime,
66 fs.clone(),
67 languages.clone(),
68 worktree_store.clone(),
69 cx,
70 )
71 });
72
73 let environment = project::ProjectEnvironment::new(&worktree_store, None, cx);
74 let task_store = cx.new_model(|cx| {
75 let mut task_store = TaskStore::local(
76 fs.clone(),
77 buffer_store.downgrade(),
78 worktree_store.clone(),
79 environment.clone(),
80 cx,
81 );
82 task_store.shared(SSH_PROJECT_ID, session.clone().into(), cx);
83 task_store
84 });
85 let settings_observer = cx.new_model(|cx| {
86 let mut observer = SettingsObserver::new_local(
87 fs.clone(),
88 worktree_store.clone(),
89 task_store.clone(),
90 cx,
91 );
92 observer.shared(SSH_PROJECT_ID, session.clone().into(), cx);
93 observer
94 });
95 let lsp_store = cx.new_model(|cx| {
96 let mut lsp_store = LspStore::new_local(
97 buffer_store.clone(),
98 worktree_store.clone(),
99 prettier_store.clone(),
100 environment,
101 languages.clone(),
102 None,
103 fs.clone(),
104 cx,
105 );
106 lsp_store.shared(SSH_PROJECT_ID, session.clone().into(), cx);
107 lsp_store
108 });
109
110 cx.subscribe(&lsp_store, Self::on_lsp_store_event).detach();
111
112 cx.subscribe(
113 &buffer_store,
114 |_this, _buffer_store, event, cx| match event {
115 BufferStoreEvent::BufferAdded(buffer) => {
116 cx.subscribe(buffer, Self::on_buffer_event).detach();
117 }
118 _ => {}
119 },
120 )
121 .detach();
122
123 let client: AnyProtoClient = session.clone().into();
124
125 session.subscribe_to_entity(SSH_PROJECT_ID, &worktree_store);
126 session.subscribe_to_entity(SSH_PROJECT_ID, &buffer_store);
127 session.subscribe_to_entity(SSH_PROJECT_ID, &cx.handle());
128 session.subscribe_to_entity(SSH_PROJECT_ID, &lsp_store);
129 session.subscribe_to_entity(SSH_PROJECT_ID, &task_store);
130 session.subscribe_to_entity(SSH_PROJECT_ID, &settings_observer);
131
132 client.add_request_handler(cx.weak_model(), Self::handle_list_remote_directory);
133 client.add_request_handler(cx.weak_model(), Self::handle_check_file_exists);
134 client.add_request_handler(cx.weak_model(), Self::handle_shutdown_remote_server);
135 client.add_request_handler(cx.weak_model(), Self::handle_ping);
136
137 client.add_model_request_handler(Self::handle_add_worktree);
138 client.add_request_handler(cx.weak_model(), Self::handle_remove_worktree);
139
140 client.add_model_request_handler(Self::handle_open_buffer_by_path);
141 client.add_model_request_handler(Self::handle_find_search_candidates);
142
143 client.add_model_request_handler(BufferStore::handle_update_buffer);
144 client.add_model_message_handler(BufferStore::handle_close_buffer);
145
146 BufferStore::init(&client);
147 WorktreeStore::init(&client);
148 SettingsObserver::init(&client);
149 LspStore::init(&client);
150 TaskStore::init(Some(&client));
151
152 HeadlessProject {
153 session: client,
154 settings_observer,
155 fs,
156 worktree_store,
157 buffer_store,
158 lsp_store,
159 task_store,
160 next_entry_id: Default::default(),
161 languages,
162 }
163 }
164
165 fn on_buffer_event(
166 &mut self,
167 buffer: Model<Buffer>,
168 event: &BufferEvent,
169 cx: &mut ModelContext<Self>,
170 ) {
171 match event {
172 BufferEvent::Operation {
173 operation,
174 is_local: true,
175 } => cx
176 .background_executor()
177 .spawn(self.session.request(proto::UpdateBuffer {
178 project_id: SSH_PROJECT_ID,
179 buffer_id: buffer.read(cx).remote_id().to_proto(),
180 operations: vec![serialize_operation(operation)],
181 }))
182 .detach(),
183 _ => {}
184 }
185 }
186
187 fn on_lsp_store_event(
188 &mut self,
189 _lsp_store: Model<LspStore>,
190 event: &LspStoreEvent,
191 _cx: &mut ModelContext<Self>,
192 ) {
193 match event {
194 LspStoreEvent::LanguageServerUpdate {
195 language_server_id,
196 message,
197 } => {
198 self.session
199 .send(proto::UpdateLanguageServer {
200 project_id: SSH_PROJECT_ID,
201 language_server_id: language_server_id.to_proto(),
202 variant: Some(message.clone()),
203 })
204 .log_err();
205 }
206 LspStoreEvent::LanguageServerLog(language_server_id, log_type, message) => {
207 self.session
208 .send(proto::LanguageServerLog {
209 project_id: SSH_PROJECT_ID,
210 language_server_id: language_server_id.to_proto(),
211 message: message.clone(),
212 log_type: Some(log_type.to_proto()),
213 })
214 .log_err();
215 }
216 _ => {}
217 }
218 }
219
220 pub async fn handle_add_worktree(
221 this: Model<Self>,
222 message: TypedEnvelope<proto::AddWorktree>,
223 mut cx: AsyncAppContext,
224 ) -> Result<proto::AddWorktreeResponse> {
225 use client::ErrorCodeExt;
226 let path = shellexpand::tilde(&message.payload.path).to_string();
227
228 let fs = this.read_with(&mut cx, |this, _| this.fs.clone())?;
229 let path = PathBuf::from(path);
230
231 let canonicalized = match fs.canonicalize(&path).await {
232 Ok(path) => path,
233 Err(e) => {
234 let mut parent = path
235 .parent()
236 .ok_or(e)
237 .map_err(|_| anyhow!("{:?} does not exist", path))?;
238 if parent == Path::new("") {
239 parent = util::paths::home_dir();
240 }
241 let parent = fs.canonicalize(parent).await.map_err(|_| {
242 anyhow!(proto::ErrorCode::DevServerProjectPathDoesNotExist
243 .with_tag("path", &path.to_string_lossy().as_ref()))
244 })?;
245 parent.join(path.file_name().unwrap())
246 }
247 };
248
249 let worktree = this
250 .update(&mut cx.clone(), |this, _| {
251 Worktree::local(
252 Arc::from(canonicalized),
253 message.payload.visible,
254 this.fs.clone(),
255 this.next_entry_id.clone(),
256 &mut cx,
257 )
258 })?
259 .await?;
260
261 let response = this.update(&mut cx, |_, cx| {
262 worktree.update(cx, |worktree, _| proto::AddWorktreeResponse {
263 worktree_id: worktree.id().to_proto(),
264 })
265 })?;
266
267 // We spawn this asynchronously, so that we can send the response back
268 // *before* `worktree_store.add()` can send out UpdateProject requests
269 // to the client about the new worktree.
270 //
271 // That lets the client manage the reference/handles of the newly-added
272 // worktree, before getting interrupted by an UpdateProject request.
273 //
274 // This fixes the problem of the client sending the AddWorktree request,
275 // headless project sending out a project update, client receiving it
276 // and immediately dropping the reference of the new client, causing it
277 // to be dropped on the headless project, and the client only then
278 // receiving a response to AddWorktree.
279 cx.spawn(|mut cx| async move {
280 this.update(&mut cx, |this, cx| {
281 this.worktree_store.update(cx, |worktree_store, cx| {
282 worktree_store.add(&worktree, cx);
283 });
284 })
285 .log_err();
286 })
287 .detach();
288
289 Ok(response)
290 }
291
292 pub async fn handle_remove_worktree(
293 this: Model<Self>,
294 envelope: TypedEnvelope<proto::RemoveWorktree>,
295 mut cx: AsyncAppContext,
296 ) -> Result<proto::Ack> {
297 let worktree_id = WorktreeId::from_proto(envelope.payload.worktree_id);
298 this.update(&mut cx, |this, cx| {
299 this.worktree_store.update(cx, |worktree_store, cx| {
300 worktree_store.remove_worktree(worktree_id, cx);
301 });
302 })?;
303 Ok(proto::Ack {})
304 }
305
306 pub async fn handle_open_buffer_by_path(
307 this: Model<Self>,
308 message: TypedEnvelope<proto::OpenBufferByPath>,
309 mut cx: AsyncAppContext,
310 ) -> Result<proto::OpenBufferResponse> {
311 let worktree_id = WorktreeId::from_proto(message.payload.worktree_id);
312 let (buffer_store, buffer) = this.update(&mut cx, |this, cx| {
313 let buffer_store = this.buffer_store.clone();
314 let buffer = this.buffer_store.update(cx, |buffer_store, cx| {
315 buffer_store.open_buffer(
316 ProjectPath {
317 worktree_id,
318 path: PathBuf::from(message.payload.path).into(),
319 },
320 cx,
321 )
322 });
323 anyhow::Ok((buffer_store, buffer))
324 })??;
325
326 let buffer = buffer.await?;
327 let buffer_id = buffer.read_with(&cx, |b, _| b.remote_id())?;
328 buffer_store.update(&mut cx, |buffer_store, cx| {
329 buffer_store
330 .create_buffer_for_peer(&buffer, SSH_PEER_ID, cx)
331 .detach_and_log_err(cx);
332 })?;
333
334 Ok(proto::OpenBufferResponse {
335 buffer_id: buffer_id.to_proto(),
336 })
337 }
338
339 pub async fn handle_find_search_candidates(
340 this: Model<Self>,
341 envelope: TypedEnvelope<proto::FindSearchCandidates>,
342 mut cx: AsyncAppContext,
343 ) -> Result<proto::FindSearchCandidatesResponse> {
344 let message = envelope.payload;
345 let query = SearchQuery::from_proto(
346 message
347 .query
348 .ok_or_else(|| anyhow!("missing query field"))?,
349 )?;
350 let mut results = this.update(&mut cx, |this, cx| {
351 this.buffer_store.update(cx, |buffer_store, cx| {
352 buffer_store.find_search_candidates(&query, message.limit as _, this.fs.clone(), cx)
353 })
354 })?;
355
356 let mut response = proto::FindSearchCandidatesResponse {
357 buffer_ids: Vec::new(),
358 };
359
360 let buffer_store = this.read_with(&cx, |this, _| this.buffer_store.clone())?;
361
362 while let Some(buffer) = results.next().await {
363 let buffer_id = buffer.update(&mut cx, |this, _| this.remote_id())?;
364 response.buffer_ids.push(buffer_id.to_proto());
365 buffer_store
366 .update(&mut cx, |buffer_store, cx| {
367 buffer_store.create_buffer_for_peer(&buffer, SSH_PEER_ID, cx)
368 })?
369 .await?;
370 }
371
372 Ok(response)
373 }
374
375 pub async fn handle_list_remote_directory(
376 this: Model<Self>,
377 envelope: TypedEnvelope<proto::ListRemoteDirectory>,
378 cx: AsyncAppContext,
379 ) -> Result<proto::ListRemoteDirectoryResponse> {
380 let expanded = shellexpand::tilde(&envelope.payload.path).to_string();
381 let fs = cx.read_model(&this, |this, _| this.fs.clone())?;
382
383 let mut entries = Vec::new();
384 let mut response = fs.read_dir(Path::new(&expanded)).await?;
385 while let Some(path) = response.next().await {
386 if let Some(file_name) = path?.file_name() {
387 entries.push(file_name.to_string_lossy().to_string());
388 }
389 }
390 Ok(proto::ListRemoteDirectoryResponse { entries })
391 }
392
393 pub async fn handle_check_file_exists(
394 this: Model<Self>,
395 envelope: TypedEnvelope<proto::CheckFileExists>,
396 cx: AsyncAppContext,
397 ) -> Result<proto::CheckFileExistsResponse> {
398 let fs = cx.read_model(&this, |this, _| this.fs.clone())?;
399 let expanded = shellexpand::tilde(&envelope.payload.path).to_string();
400
401 let exists = fs.is_file(&PathBuf::from(expanded.clone())).await;
402
403 Ok(proto::CheckFileExistsResponse {
404 exists,
405 path: expanded,
406 })
407 }
408
409 pub async fn handle_shutdown_remote_server(
410 _this: Model<Self>,
411 _envelope: TypedEnvelope<proto::ShutdownRemoteServer>,
412 cx: AsyncAppContext,
413 ) -> Result<proto::Ack> {
414 cx.spawn(|cx| async move {
415 cx.update(|cx| {
416 // TODO: This is a hack, because in a headless project, shutdown isn't executed
417 // when calling quit, but it should be.
418 cx.shutdown();
419 cx.quit();
420 })
421 })
422 .detach();
423
424 Ok(proto::Ack {})
425 }
426
427 pub async fn handle_ping(
428 _this: Model<Self>,
429 _envelope: TypedEnvelope<proto::Ping>,
430 _cx: AsyncAppContext,
431 ) -> Result<proto::Ack> {
432 log::debug!("Received ping from client");
433 Ok(proto::Ack {})
434 }
435}