1mod access_token;
2mod contact;
3mod follower;
4mod language_server;
5mod project;
6mod project_collaborator;
7mod room;
8mod room_participant;
9mod server;
10mod signup;
11#[cfg(test)]
12mod tests;
13mod user;
14mod worktree;
15mod worktree_diagnostic_summary;
16mod worktree_entry;
17mod worktree_repository;
18mod worktree_repository_statuses;
19mod worktree_settings_file;
20
21use crate::executor::Executor;
22use crate::{Error, Result};
23use anyhow::anyhow;
24use collections::{BTreeMap, HashMap, HashSet};
25pub use contact::Contact;
26use dashmap::DashMap;
27use futures::StreamExt;
28use hyper::StatusCode;
29use rand::prelude::StdRng;
30use rand::{Rng, SeedableRng};
31use rpc::{proto, ConnectionId};
32use sea_orm::Condition;
33pub use sea_orm::ConnectOptions;
34use sea_orm::{
35 entity::prelude::*, ActiveValue, ConnectionTrait, DatabaseConnection, DatabaseTransaction,
36 DbErr, FromQueryResult, IntoActiveModel, IsolationLevel, JoinType, QueryOrder, QuerySelect,
37 Statement, TransactionTrait,
38};
39use sea_query::{Alias, Expr, OnConflict, Query};
40use serde::{Deserialize, Serialize};
41pub use signup::{Invite, NewSignup, WaitlistSummary};
42use sqlx::migrate::{Migrate, Migration, MigrationSource};
43use sqlx::Connection;
44use std::ops::{Deref, DerefMut};
45use std::path::Path;
46use std::time::Duration;
47use std::{future::Future, marker::PhantomData, rc::Rc, sync::Arc};
48use tokio::sync::{Mutex, OwnedMutexGuard};
49pub use user::Model as User;
50
51pub struct Database {
52 options: ConnectOptions,
53 pool: DatabaseConnection,
54 rooms: DashMap<RoomId, Arc<Mutex<()>>>,
55 rng: Mutex<StdRng>,
56 executor: Executor,
57 #[cfg(test)]
58 runtime: Option<tokio::runtime::Runtime>,
59}
60
61impl Database {
62 pub async fn new(options: ConnectOptions, executor: Executor) -> Result<Self> {
63 Ok(Self {
64 options: options.clone(),
65 pool: sea_orm::Database::connect(options).await?,
66 rooms: DashMap::with_capacity(16384),
67 rng: Mutex::new(StdRng::seed_from_u64(0)),
68 executor,
69 #[cfg(test)]
70 runtime: None,
71 })
72 }
73
74 #[cfg(test)]
75 pub fn reset(&self) {
76 self.rooms.clear();
77 }
78
79 pub async fn migrate(
80 &self,
81 migrations_path: &Path,
82 ignore_checksum_mismatch: bool,
83 ) -> anyhow::Result<Vec<(Migration, Duration)>> {
84 let migrations = MigrationSource::resolve(migrations_path)
85 .await
86 .map_err(|err| anyhow!("failed to load migrations: {err:?}"))?;
87
88 let mut connection = sqlx::AnyConnection::connect(self.options.get_url()).await?;
89
90 connection.ensure_migrations_table().await?;
91 let applied_migrations: HashMap<_, _> = connection
92 .list_applied_migrations()
93 .await?
94 .into_iter()
95 .map(|m| (m.version, m))
96 .collect();
97
98 let mut new_migrations = Vec::new();
99 for migration in migrations {
100 match applied_migrations.get(&migration.version) {
101 Some(applied_migration) => {
102 if migration.checksum != applied_migration.checksum && !ignore_checksum_mismatch
103 {
104 Err(anyhow!(
105 "checksum mismatch for applied migration {}",
106 migration.description
107 ))?;
108 }
109 }
110 None => {
111 let elapsed = connection.apply(&migration).await?;
112 new_migrations.push((migration, elapsed));
113 }
114 }
115 }
116
117 Ok(new_migrations)
118 }
119
120 pub async fn create_server(&self, environment: &str) -> Result<ServerId> {
121 self.transaction(|tx| async move {
122 let server = server::ActiveModel {
123 environment: ActiveValue::set(environment.into()),
124 ..Default::default()
125 }
126 .insert(&*tx)
127 .await?;
128 Ok(server.id)
129 })
130 .await
131 }
132
133 pub async fn stale_room_ids(
134 &self,
135 environment: &str,
136 new_server_id: ServerId,
137 ) -> Result<Vec<RoomId>> {
138 self.transaction(|tx| async move {
139 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
140 enum QueryAs {
141 RoomId,
142 }
143
144 let stale_server_epochs = self
145 .stale_server_ids(environment, new_server_id, &tx)
146 .await?;
147 Ok(room_participant::Entity::find()
148 .select_only()
149 .column(room_participant::Column::RoomId)
150 .distinct()
151 .filter(
152 room_participant::Column::AnsweringConnectionServerId
153 .is_in(stale_server_epochs),
154 )
155 .into_values::<_, QueryAs>()
156 .all(&*tx)
157 .await?)
158 })
159 .await
160 }
161
162 pub async fn refresh_room(
163 &self,
164 room_id: RoomId,
165 new_server_id: ServerId,
166 ) -> Result<RoomGuard<RefreshedRoom>> {
167 self.room_transaction(room_id, |tx| async move {
168 let stale_participant_filter = Condition::all()
169 .add(room_participant::Column::RoomId.eq(room_id))
170 .add(room_participant::Column::AnsweringConnectionId.is_not_null())
171 .add(room_participant::Column::AnsweringConnectionServerId.ne(new_server_id));
172
173 let stale_participant_user_ids = room_participant::Entity::find()
174 .filter(stale_participant_filter.clone())
175 .all(&*tx)
176 .await?
177 .into_iter()
178 .map(|participant| participant.user_id)
179 .collect::<Vec<_>>();
180
181 // Delete participants who failed to reconnect and cancel their calls.
182 let mut canceled_calls_to_user_ids = Vec::new();
183 room_participant::Entity::delete_many()
184 .filter(stale_participant_filter)
185 .exec(&*tx)
186 .await?;
187 let called_participants = room_participant::Entity::find()
188 .filter(
189 Condition::all()
190 .add(
191 room_participant::Column::CallingUserId
192 .is_in(stale_participant_user_ids.iter().copied()),
193 )
194 .add(room_participant::Column::AnsweringConnectionId.is_null()),
195 )
196 .all(&*tx)
197 .await?;
198 room_participant::Entity::delete_many()
199 .filter(
200 room_participant::Column::Id
201 .is_in(called_participants.iter().map(|participant| participant.id)),
202 )
203 .exec(&*tx)
204 .await?;
205 canceled_calls_to_user_ids.extend(
206 called_participants
207 .into_iter()
208 .map(|participant| participant.user_id),
209 );
210
211 let room = self.get_room(room_id, &tx).await?;
212 // Delete the room if it becomes empty.
213 if room.participants.is_empty() {
214 project::Entity::delete_many()
215 .filter(project::Column::RoomId.eq(room_id))
216 .exec(&*tx)
217 .await?;
218 room::Entity::delete_by_id(room_id).exec(&*tx).await?;
219 }
220
221 Ok(RefreshedRoom {
222 room,
223 stale_participant_user_ids,
224 canceled_calls_to_user_ids,
225 })
226 })
227 .await
228 }
229
230 pub async fn delete_stale_servers(
231 &self,
232 environment: &str,
233 new_server_id: ServerId,
234 ) -> Result<()> {
235 self.transaction(|tx| async move {
236 server::Entity::delete_many()
237 .filter(
238 Condition::all()
239 .add(server::Column::Environment.eq(environment))
240 .add(server::Column::Id.ne(new_server_id)),
241 )
242 .exec(&*tx)
243 .await?;
244 Ok(())
245 })
246 .await
247 }
248
249 async fn stale_server_ids(
250 &self,
251 environment: &str,
252 new_server_id: ServerId,
253 tx: &DatabaseTransaction,
254 ) -> Result<Vec<ServerId>> {
255 let stale_servers = server::Entity::find()
256 .filter(
257 Condition::all()
258 .add(server::Column::Environment.eq(environment))
259 .add(server::Column::Id.ne(new_server_id)),
260 )
261 .all(&*tx)
262 .await?;
263 Ok(stale_servers.into_iter().map(|server| server.id).collect())
264 }
265
266 // users
267
268 pub async fn create_user(
269 &self,
270 email_address: &str,
271 admin: bool,
272 params: NewUserParams,
273 ) -> Result<NewUserResult> {
274 self.transaction(|tx| async {
275 let tx = tx;
276 let user = user::Entity::insert(user::ActiveModel {
277 email_address: ActiveValue::set(Some(email_address.into())),
278 github_login: ActiveValue::set(params.github_login.clone()),
279 github_user_id: ActiveValue::set(Some(params.github_user_id)),
280 admin: ActiveValue::set(admin),
281 metrics_id: ActiveValue::set(Uuid::new_v4()),
282 ..Default::default()
283 })
284 .on_conflict(
285 OnConflict::column(user::Column::GithubLogin)
286 .update_column(user::Column::GithubLogin)
287 .to_owned(),
288 )
289 .exec_with_returning(&*tx)
290 .await?;
291
292 Ok(NewUserResult {
293 user_id: user.id,
294 metrics_id: user.metrics_id.to_string(),
295 signup_device_id: None,
296 inviting_user_id: None,
297 })
298 })
299 .await
300 }
301
302 pub async fn get_user_by_id(&self, id: UserId) -> Result<Option<user::Model>> {
303 self.transaction(|tx| async move { Ok(user::Entity::find_by_id(id).one(&*tx).await?) })
304 .await
305 }
306
307 pub async fn get_users_by_ids(&self, ids: Vec<UserId>) -> Result<Vec<user::Model>> {
308 self.transaction(|tx| async {
309 let tx = tx;
310 Ok(user::Entity::find()
311 .filter(user::Column::Id.is_in(ids.iter().copied()))
312 .all(&*tx)
313 .await?)
314 })
315 .await
316 }
317
318 pub async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>> {
319 self.transaction(|tx| async move {
320 Ok(user::Entity::find()
321 .filter(user::Column::GithubLogin.eq(github_login))
322 .one(&*tx)
323 .await?)
324 })
325 .await
326 }
327
328 pub async fn get_or_create_user_by_github_account(
329 &self,
330 github_login: &str,
331 github_user_id: Option<i32>,
332 github_email: Option<&str>,
333 ) -> Result<Option<User>> {
334 self.transaction(|tx| async move {
335 let tx = &*tx;
336 if let Some(github_user_id) = github_user_id {
337 if let Some(user_by_github_user_id) = user::Entity::find()
338 .filter(user::Column::GithubUserId.eq(github_user_id))
339 .one(tx)
340 .await?
341 {
342 let mut user_by_github_user_id = user_by_github_user_id.into_active_model();
343 user_by_github_user_id.github_login = ActiveValue::set(github_login.into());
344 Ok(Some(user_by_github_user_id.update(tx).await?))
345 } else if let Some(user_by_github_login) = user::Entity::find()
346 .filter(user::Column::GithubLogin.eq(github_login))
347 .one(tx)
348 .await?
349 {
350 let mut user_by_github_login = user_by_github_login.into_active_model();
351 user_by_github_login.github_user_id = ActiveValue::set(Some(github_user_id));
352 Ok(Some(user_by_github_login.update(tx).await?))
353 } else {
354 let user = user::Entity::insert(user::ActiveModel {
355 email_address: ActiveValue::set(github_email.map(|email| email.into())),
356 github_login: ActiveValue::set(github_login.into()),
357 github_user_id: ActiveValue::set(Some(github_user_id)),
358 admin: ActiveValue::set(false),
359 invite_count: ActiveValue::set(0),
360 invite_code: ActiveValue::set(None),
361 metrics_id: ActiveValue::set(Uuid::new_v4()),
362 ..Default::default()
363 })
364 .exec_with_returning(&*tx)
365 .await?;
366 Ok(Some(user))
367 }
368 } else {
369 Ok(user::Entity::find()
370 .filter(user::Column::GithubLogin.eq(github_login))
371 .one(tx)
372 .await?)
373 }
374 })
375 .await
376 }
377
378 pub async fn get_all_users(&self, page: u32, limit: u32) -> Result<Vec<User>> {
379 self.transaction(|tx| async move {
380 Ok(user::Entity::find()
381 .order_by_asc(user::Column::GithubLogin)
382 .limit(limit as u64)
383 .offset(page as u64 * limit as u64)
384 .all(&*tx)
385 .await?)
386 })
387 .await
388 }
389
390 pub async fn get_users_with_no_invites(
391 &self,
392 invited_by_another_user: bool,
393 ) -> Result<Vec<User>> {
394 self.transaction(|tx| async move {
395 Ok(user::Entity::find()
396 .filter(
397 user::Column::InviteCount
398 .eq(0)
399 .and(if invited_by_another_user {
400 user::Column::InviterId.is_not_null()
401 } else {
402 user::Column::InviterId.is_null()
403 }),
404 )
405 .all(&*tx)
406 .await?)
407 })
408 .await
409 }
410
411 pub async fn get_user_metrics_id(&self, id: UserId) -> Result<String> {
412 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
413 enum QueryAs {
414 MetricsId,
415 }
416
417 self.transaction(|tx| async move {
418 let metrics_id: Uuid = user::Entity::find_by_id(id)
419 .select_only()
420 .column(user::Column::MetricsId)
421 .into_values::<_, QueryAs>()
422 .one(&*tx)
423 .await?
424 .ok_or_else(|| anyhow!("could not find user"))?;
425 Ok(metrics_id.to_string())
426 })
427 .await
428 }
429
430 pub async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()> {
431 self.transaction(|tx| async move {
432 user::Entity::update_many()
433 .filter(user::Column::Id.eq(id))
434 .set(user::ActiveModel {
435 admin: ActiveValue::set(is_admin),
436 ..Default::default()
437 })
438 .exec(&*tx)
439 .await?;
440 Ok(())
441 })
442 .await
443 }
444
445 pub async fn set_user_connected_once(&self, id: UserId, connected_once: bool) -> Result<()> {
446 self.transaction(|tx| async move {
447 user::Entity::update_many()
448 .filter(user::Column::Id.eq(id))
449 .set(user::ActiveModel {
450 connected_once: ActiveValue::set(connected_once),
451 ..Default::default()
452 })
453 .exec(&*tx)
454 .await?;
455 Ok(())
456 })
457 .await
458 }
459
460 pub async fn destroy_user(&self, id: UserId) -> Result<()> {
461 self.transaction(|tx| async move {
462 access_token::Entity::delete_many()
463 .filter(access_token::Column::UserId.eq(id))
464 .exec(&*tx)
465 .await?;
466 user::Entity::delete_by_id(id).exec(&*tx).await?;
467 Ok(())
468 })
469 .await
470 }
471
472 // contacts
473
474 pub async fn get_contacts(&self, user_id: UserId) -> Result<Vec<Contact>> {
475 #[derive(Debug, FromQueryResult)]
476 struct ContactWithUserBusyStatuses {
477 user_id_a: UserId,
478 user_id_b: UserId,
479 a_to_b: bool,
480 accepted: bool,
481 should_notify: bool,
482 user_a_busy: bool,
483 user_b_busy: bool,
484 }
485
486 self.transaction(|tx| async move {
487 let user_a_participant = Alias::new("user_a_participant");
488 let user_b_participant = Alias::new("user_b_participant");
489 let mut db_contacts = contact::Entity::find()
490 .column_as(
491 Expr::tbl(user_a_participant.clone(), room_participant::Column::Id)
492 .is_not_null(),
493 "user_a_busy",
494 )
495 .column_as(
496 Expr::tbl(user_b_participant.clone(), room_participant::Column::Id)
497 .is_not_null(),
498 "user_b_busy",
499 )
500 .filter(
501 contact::Column::UserIdA
502 .eq(user_id)
503 .or(contact::Column::UserIdB.eq(user_id)),
504 )
505 .join_as(
506 JoinType::LeftJoin,
507 contact::Relation::UserARoomParticipant.def(),
508 user_a_participant,
509 )
510 .join_as(
511 JoinType::LeftJoin,
512 contact::Relation::UserBRoomParticipant.def(),
513 user_b_participant,
514 )
515 .into_model::<ContactWithUserBusyStatuses>()
516 .stream(&*tx)
517 .await?;
518
519 let mut contacts = Vec::new();
520 while let Some(db_contact) = db_contacts.next().await {
521 let db_contact = db_contact?;
522 if db_contact.user_id_a == user_id {
523 if db_contact.accepted {
524 contacts.push(Contact::Accepted {
525 user_id: db_contact.user_id_b,
526 should_notify: db_contact.should_notify && db_contact.a_to_b,
527 busy: db_contact.user_b_busy,
528 });
529 } else if db_contact.a_to_b {
530 contacts.push(Contact::Outgoing {
531 user_id: db_contact.user_id_b,
532 })
533 } else {
534 contacts.push(Contact::Incoming {
535 user_id: db_contact.user_id_b,
536 should_notify: db_contact.should_notify,
537 });
538 }
539 } else if db_contact.accepted {
540 contacts.push(Contact::Accepted {
541 user_id: db_contact.user_id_a,
542 should_notify: db_contact.should_notify && !db_contact.a_to_b,
543 busy: db_contact.user_a_busy,
544 });
545 } else if db_contact.a_to_b {
546 contacts.push(Contact::Incoming {
547 user_id: db_contact.user_id_a,
548 should_notify: db_contact.should_notify,
549 });
550 } else {
551 contacts.push(Contact::Outgoing {
552 user_id: db_contact.user_id_a,
553 });
554 }
555 }
556
557 contacts.sort_unstable_by_key(|contact| contact.user_id());
558
559 Ok(contacts)
560 })
561 .await
562 }
563
564 pub async fn is_user_busy(&self, user_id: UserId) -> Result<bool> {
565 self.transaction(|tx| async move {
566 let participant = room_participant::Entity::find()
567 .filter(room_participant::Column::UserId.eq(user_id))
568 .one(&*tx)
569 .await?;
570 Ok(participant.is_some())
571 })
572 .await
573 }
574
575 pub async fn has_contact(&self, user_id_1: UserId, user_id_2: UserId) -> Result<bool> {
576 self.transaction(|tx| async move {
577 let (id_a, id_b) = if user_id_1 < user_id_2 {
578 (user_id_1, user_id_2)
579 } else {
580 (user_id_2, user_id_1)
581 };
582
583 Ok(contact::Entity::find()
584 .filter(
585 contact::Column::UserIdA
586 .eq(id_a)
587 .and(contact::Column::UserIdB.eq(id_b))
588 .and(contact::Column::Accepted.eq(true)),
589 )
590 .one(&*tx)
591 .await?
592 .is_some())
593 })
594 .await
595 }
596
597 pub async fn send_contact_request(&self, sender_id: UserId, receiver_id: UserId) -> Result<()> {
598 self.transaction(|tx| async move {
599 let (id_a, id_b, a_to_b) = if sender_id < receiver_id {
600 (sender_id, receiver_id, true)
601 } else {
602 (receiver_id, sender_id, false)
603 };
604
605 let rows_affected = contact::Entity::insert(contact::ActiveModel {
606 user_id_a: ActiveValue::set(id_a),
607 user_id_b: ActiveValue::set(id_b),
608 a_to_b: ActiveValue::set(a_to_b),
609 accepted: ActiveValue::set(false),
610 should_notify: ActiveValue::set(true),
611 ..Default::default()
612 })
613 .on_conflict(
614 OnConflict::columns([contact::Column::UserIdA, contact::Column::UserIdB])
615 .values([
616 (contact::Column::Accepted, true.into()),
617 (contact::Column::ShouldNotify, false.into()),
618 ])
619 .action_and_where(
620 contact::Column::Accepted.eq(false).and(
621 contact::Column::AToB
622 .eq(a_to_b)
623 .and(contact::Column::UserIdA.eq(id_b))
624 .or(contact::Column::AToB
625 .ne(a_to_b)
626 .and(contact::Column::UserIdA.eq(id_a))),
627 ),
628 )
629 .to_owned(),
630 )
631 .exec_without_returning(&*tx)
632 .await?;
633
634 if rows_affected == 1 {
635 Ok(())
636 } else {
637 Err(anyhow!("contact already requested"))?
638 }
639 })
640 .await
641 }
642
643 /// Returns a bool indicating whether the removed contact had originally accepted or not
644 ///
645 /// Deletes the contact identified by the requester and responder ids, and then returns
646 /// whether the deleted contact had originally accepted or was a pending contact request.
647 ///
648 /// # Arguments
649 ///
650 /// * `requester_id` - The user that initiates this request
651 /// * `responder_id` - The user that will be removed
652 pub async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<bool> {
653 self.transaction(|tx| async move {
654 let (id_a, id_b) = if responder_id < requester_id {
655 (responder_id, requester_id)
656 } else {
657 (requester_id, responder_id)
658 };
659
660 let contact = contact::Entity::find()
661 .filter(
662 contact::Column::UserIdA
663 .eq(id_a)
664 .and(contact::Column::UserIdB.eq(id_b)),
665 )
666 .one(&*tx)
667 .await?
668 .ok_or_else(|| anyhow!("no such contact"))?;
669
670 contact::Entity::delete_by_id(contact.id).exec(&*tx).await?;
671 Ok(contact.accepted)
672 })
673 .await
674 }
675
676 pub async fn dismiss_contact_notification(
677 &self,
678 user_id: UserId,
679 contact_user_id: UserId,
680 ) -> Result<()> {
681 self.transaction(|tx| async move {
682 let (id_a, id_b, a_to_b) = if user_id < contact_user_id {
683 (user_id, contact_user_id, true)
684 } else {
685 (contact_user_id, user_id, false)
686 };
687
688 let result = contact::Entity::update_many()
689 .set(contact::ActiveModel {
690 should_notify: ActiveValue::set(false),
691 ..Default::default()
692 })
693 .filter(
694 contact::Column::UserIdA
695 .eq(id_a)
696 .and(contact::Column::UserIdB.eq(id_b))
697 .and(
698 contact::Column::AToB
699 .eq(a_to_b)
700 .and(contact::Column::Accepted.eq(true))
701 .or(contact::Column::AToB
702 .ne(a_to_b)
703 .and(contact::Column::Accepted.eq(false))),
704 ),
705 )
706 .exec(&*tx)
707 .await?;
708 if result.rows_affected == 0 {
709 Err(anyhow!("no such contact request"))?
710 } else {
711 Ok(())
712 }
713 })
714 .await
715 }
716
717 pub async fn respond_to_contact_request(
718 &self,
719 responder_id: UserId,
720 requester_id: UserId,
721 accept: bool,
722 ) -> Result<()> {
723 self.transaction(|tx| async move {
724 let (id_a, id_b, a_to_b) = if responder_id < requester_id {
725 (responder_id, requester_id, false)
726 } else {
727 (requester_id, responder_id, true)
728 };
729 let rows_affected = if accept {
730 let result = contact::Entity::update_many()
731 .set(contact::ActiveModel {
732 accepted: ActiveValue::set(true),
733 should_notify: ActiveValue::set(true),
734 ..Default::default()
735 })
736 .filter(
737 contact::Column::UserIdA
738 .eq(id_a)
739 .and(contact::Column::UserIdB.eq(id_b))
740 .and(contact::Column::AToB.eq(a_to_b)),
741 )
742 .exec(&*tx)
743 .await?;
744 result.rows_affected
745 } else {
746 let result = contact::Entity::delete_many()
747 .filter(
748 contact::Column::UserIdA
749 .eq(id_a)
750 .and(contact::Column::UserIdB.eq(id_b))
751 .and(contact::Column::AToB.eq(a_to_b))
752 .and(contact::Column::Accepted.eq(false)),
753 )
754 .exec(&*tx)
755 .await?;
756
757 result.rows_affected
758 };
759
760 if rows_affected == 1 {
761 Ok(())
762 } else {
763 Err(anyhow!("no such contact request"))?
764 }
765 })
766 .await
767 }
768
769 pub fn fuzzy_like_string(string: &str) -> String {
770 let mut result = String::with_capacity(string.len() * 2 + 1);
771 for c in string.chars() {
772 if c.is_alphanumeric() {
773 result.push('%');
774 result.push(c);
775 }
776 }
777 result.push('%');
778 result
779 }
780
781 pub async fn fuzzy_search_users(&self, name_query: &str, limit: u32) -> Result<Vec<User>> {
782 self.transaction(|tx| async {
783 let tx = tx;
784 let like_string = Self::fuzzy_like_string(name_query);
785 let query = "
786 SELECT users.*
787 FROM users
788 WHERE github_login ILIKE $1
789 ORDER BY github_login <-> $2
790 LIMIT $3
791 ";
792
793 Ok(user::Entity::find()
794 .from_raw_sql(Statement::from_sql_and_values(
795 self.pool.get_database_backend(),
796 query.into(),
797 vec![like_string.into(), name_query.into(), limit.into()],
798 ))
799 .all(&*tx)
800 .await?)
801 })
802 .await
803 }
804
805 // signups
806
807 pub async fn create_signup(&self, signup: &NewSignup) -> Result<()> {
808 self.transaction(|tx| async move {
809 signup::Entity::insert(signup::ActiveModel {
810 email_address: ActiveValue::set(signup.email_address.clone()),
811 email_confirmation_code: ActiveValue::set(random_email_confirmation_code()),
812 email_confirmation_sent: ActiveValue::set(false),
813 platform_mac: ActiveValue::set(signup.platform_mac),
814 platform_windows: ActiveValue::set(signup.platform_windows),
815 platform_linux: ActiveValue::set(signup.platform_linux),
816 platform_unknown: ActiveValue::set(false),
817 editor_features: ActiveValue::set(Some(signup.editor_features.clone())),
818 programming_languages: ActiveValue::set(Some(signup.programming_languages.clone())),
819 device_id: ActiveValue::set(signup.device_id.clone()),
820 added_to_mailing_list: ActiveValue::set(signup.added_to_mailing_list),
821 ..Default::default()
822 })
823 .on_conflict(
824 OnConflict::column(signup::Column::EmailAddress)
825 .update_columns([
826 signup::Column::PlatformMac,
827 signup::Column::PlatformWindows,
828 signup::Column::PlatformLinux,
829 signup::Column::EditorFeatures,
830 signup::Column::ProgrammingLanguages,
831 signup::Column::DeviceId,
832 signup::Column::AddedToMailingList,
833 ])
834 .to_owned(),
835 )
836 .exec(&*tx)
837 .await?;
838 Ok(())
839 })
840 .await
841 }
842
843 pub async fn get_signup(&self, email_address: &str) -> Result<signup::Model> {
844 self.transaction(|tx| async move {
845 let signup = signup::Entity::find()
846 .filter(signup::Column::EmailAddress.eq(email_address))
847 .one(&*tx)
848 .await?
849 .ok_or_else(|| {
850 anyhow!("signup with email address {} doesn't exist", email_address)
851 })?;
852
853 Ok(signup)
854 })
855 .await
856 }
857
858 pub async fn get_waitlist_summary(&self) -> Result<WaitlistSummary> {
859 self.transaction(|tx| async move {
860 let query = "
861 SELECT
862 COUNT(*) as count,
863 COALESCE(SUM(CASE WHEN platform_linux THEN 1 ELSE 0 END), 0) as linux_count,
864 COALESCE(SUM(CASE WHEN platform_mac THEN 1 ELSE 0 END), 0) as mac_count,
865 COALESCE(SUM(CASE WHEN platform_windows THEN 1 ELSE 0 END), 0) as windows_count,
866 COALESCE(SUM(CASE WHEN platform_unknown THEN 1 ELSE 0 END), 0) as unknown_count
867 FROM (
868 SELECT *
869 FROM signups
870 WHERE
871 NOT email_confirmation_sent
872 ) AS unsent
873 ";
874 Ok(
875 WaitlistSummary::find_by_statement(Statement::from_sql_and_values(
876 self.pool.get_database_backend(),
877 query.into(),
878 vec![],
879 ))
880 .one(&*tx)
881 .await?
882 .ok_or_else(|| anyhow!("invalid result"))?,
883 )
884 })
885 .await
886 }
887
888 pub async fn record_sent_invites(&self, invites: &[Invite]) -> Result<()> {
889 let emails = invites
890 .iter()
891 .map(|s| s.email_address.as_str())
892 .collect::<Vec<_>>();
893 self.transaction(|tx| async {
894 let tx = tx;
895 signup::Entity::update_many()
896 .filter(signup::Column::EmailAddress.is_in(emails.iter().copied()))
897 .set(signup::ActiveModel {
898 email_confirmation_sent: ActiveValue::set(true),
899 ..Default::default()
900 })
901 .exec(&*tx)
902 .await?;
903 Ok(())
904 })
905 .await
906 }
907
908 pub async fn get_unsent_invites(&self, count: usize) -> Result<Vec<Invite>> {
909 self.transaction(|tx| async move {
910 Ok(signup::Entity::find()
911 .select_only()
912 .column(signup::Column::EmailAddress)
913 .column(signup::Column::EmailConfirmationCode)
914 .filter(
915 signup::Column::EmailConfirmationSent.eq(false).and(
916 signup::Column::PlatformMac
917 .eq(true)
918 .or(signup::Column::PlatformUnknown.eq(true)),
919 ),
920 )
921 .order_by_asc(signup::Column::CreatedAt)
922 .limit(count as u64)
923 .into_model()
924 .all(&*tx)
925 .await?)
926 })
927 .await
928 }
929
930 // invite codes
931
932 pub async fn create_invite_from_code(
933 &self,
934 code: &str,
935 email_address: &str,
936 device_id: Option<&str>,
937 added_to_mailing_list: bool,
938 ) -> Result<Invite> {
939 self.transaction(|tx| async move {
940 let existing_user = user::Entity::find()
941 .filter(user::Column::EmailAddress.eq(email_address))
942 .one(&*tx)
943 .await?;
944
945 if existing_user.is_some() {
946 Err(anyhow!("email address is already in use"))?;
947 }
948
949 let inviting_user_with_invites = match user::Entity::find()
950 .filter(
951 user::Column::InviteCode
952 .eq(code)
953 .and(user::Column::InviteCount.gt(0)),
954 )
955 .one(&*tx)
956 .await?
957 {
958 Some(inviting_user) => inviting_user,
959 None => {
960 return Err(Error::Http(
961 StatusCode::UNAUTHORIZED,
962 "unable to find an invite code with invites remaining".to_string(),
963 ))?
964 }
965 };
966 user::Entity::update_many()
967 .filter(
968 user::Column::Id
969 .eq(inviting_user_with_invites.id)
970 .and(user::Column::InviteCount.gt(0)),
971 )
972 .col_expr(
973 user::Column::InviteCount,
974 Expr::col(user::Column::InviteCount).sub(1),
975 )
976 .exec(&*tx)
977 .await?;
978
979 let signup = signup::Entity::insert(signup::ActiveModel {
980 email_address: ActiveValue::set(email_address.into()),
981 email_confirmation_code: ActiveValue::set(random_email_confirmation_code()),
982 email_confirmation_sent: ActiveValue::set(false),
983 inviting_user_id: ActiveValue::set(Some(inviting_user_with_invites.id)),
984 platform_linux: ActiveValue::set(false),
985 platform_mac: ActiveValue::set(false),
986 platform_windows: ActiveValue::set(false),
987 platform_unknown: ActiveValue::set(true),
988 device_id: ActiveValue::set(device_id.map(|device_id| device_id.into())),
989 added_to_mailing_list: ActiveValue::set(added_to_mailing_list),
990 ..Default::default()
991 })
992 .on_conflict(
993 OnConflict::column(signup::Column::EmailAddress)
994 .update_column(signup::Column::InvitingUserId)
995 .to_owned(),
996 )
997 .exec_with_returning(&*tx)
998 .await?;
999
1000 Ok(Invite {
1001 email_address: signup.email_address,
1002 email_confirmation_code: signup.email_confirmation_code,
1003 })
1004 })
1005 .await
1006 }
1007
1008 pub async fn create_user_from_invite(
1009 &self,
1010 invite: &Invite,
1011 user: NewUserParams,
1012 ) -> Result<Option<NewUserResult>> {
1013 self.transaction(|tx| async {
1014 let tx = tx;
1015 let signup = signup::Entity::find()
1016 .filter(
1017 signup::Column::EmailAddress
1018 .eq(invite.email_address.as_str())
1019 .and(
1020 signup::Column::EmailConfirmationCode
1021 .eq(invite.email_confirmation_code.as_str()),
1022 ),
1023 )
1024 .one(&*tx)
1025 .await?
1026 .ok_or_else(|| Error::Http(StatusCode::NOT_FOUND, "no such invite".to_string()))?;
1027
1028 if signup.user_id.is_some() {
1029 return Ok(None);
1030 }
1031
1032 let user = user::Entity::insert(user::ActiveModel {
1033 email_address: ActiveValue::set(Some(invite.email_address.clone())),
1034 github_login: ActiveValue::set(user.github_login.clone()),
1035 github_user_id: ActiveValue::set(Some(user.github_user_id)),
1036 admin: ActiveValue::set(false),
1037 invite_count: ActiveValue::set(user.invite_count),
1038 invite_code: ActiveValue::set(Some(random_invite_code())),
1039 metrics_id: ActiveValue::set(Uuid::new_v4()),
1040 ..Default::default()
1041 })
1042 .on_conflict(
1043 OnConflict::column(user::Column::GithubLogin)
1044 .update_columns([
1045 user::Column::EmailAddress,
1046 user::Column::GithubUserId,
1047 user::Column::Admin,
1048 ])
1049 .to_owned(),
1050 )
1051 .exec_with_returning(&*tx)
1052 .await?;
1053
1054 let mut signup = signup.into_active_model();
1055 signup.user_id = ActiveValue::set(Some(user.id));
1056 let signup = signup.update(&*tx).await?;
1057
1058 if let Some(inviting_user_id) = signup.inviting_user_id {
1059 let (user_id_a, user_id_b, a_to_b) = if inviting_user_id < user.id {
1060 (inviting_user_id, user.id, true)
1061 } else {
1062 (user.id, inviting_user_id, false)
1063 };
1064
1065 contact::Entity::insert(contact::ActiveModel {
1066 user_id_a: ActiveValue::set(user_id_a),
1067 user_id_b: ActiveValue::set(user_id_b),
1068 a_to_b: ActiveValue::set(a_to_b),
1069 should_notify: ActiveValue::set(true),
1070 accepted: ActiveValue::set(true),
1071 ..Default::default()
1072 })
1073 .on_conflict(OnConflict::new().do_nothing().to_owned())
1074 .exec_without_returning(&*tx)
1075 .await?;
1076 }
1077
1078 Ok(Some(NewUserResult {
1079 user_id: user.id,
1080 metrics_id: user.metrics_id.to_string(),
1081 inviting_user_id: signup.inviting_user_id,
1082 signup_device_id: signup.device_id,
1083 }))
1084 })
1085 .await
1086 }
1087
1088 pub async fn set_invite_count_for_user(&self, id: UserId, count: i32) -> Result<()> {
1089 self.transaction(|tx| async move {
1090 if count > 0 {
1091 user::Entity::update_many()
1092 .filter(
1093 user::Column::Id
1094 .eq(id)
1095 .and(user::Column::InviteCode.is_null()),
1096 )
1097 .set(user::ActiveModel {
1098 invite_code: ActiveValue::set(Some(random_invite_code())),
1099 ..Default::default()
1100 })
1101 .exec(&*tx)
1102 .await?;
1103 }
1104
1105 user::Entity::update_many()
1106 .filter(user::Column::Id.eq(id))
1107 .set(user::ActiveModel {
1108 invite_count: ActiveValue::set(count),
1109 ..Default::default()
1110 })
1111 .exec(&*tx)
1112 .await?;
1113 Ok(())
1114 })
1115 .await
1116 }
1117
1118 pub async fn get_invite_code_for_user(&self, id: UserId) -> Result<Option<(String, i32)>> {
1119 self.transaction(|tx| async move {
1120 match user::Entity::find_by_id(id).one(&*tx).await? {
1121 Some(user) if user.invite_code.is_some() => {
1122 Ok(Some((user.invite_code.unwrap(), user.invite_count)))
1123 }
1124 _ => Ok(None),
1125 }
1126 })
1127 .await
1128 }
1129
1130 pub async fn get_user_for_invite_code(&self, code: &str) -> Result<User> {
1131 self.transaction(|tx| async move {
1132 user::Entity::find()
1133 .filter(user::Column::InviteCode.eq(code))
1134 .one(&*tx)
1135 .await?
1136 .ok_or_else(|| {
1137 Error::Http(
1138 StatusCode::NOT_FOUND,
1139 "that invite code does not exist".to_string(),
1140 )
1141 })
1142 })
1143 .await
1144 }
1145
1146 // rooms
1147
1148 pub async fn incoming_call_for_user(
1149 &self,
1150 user_id: UserId,
1151 ) -> Result<Option<proto::IncomingCall>> {
1152 self.transaction(|tx| async move {
1153 let pending_participant = room_participant::Entity::find()
1154 .filter(
1155 room_participant::Column::UserId
1156 .eq(user_id)
1157 .and(room_participant::Column::AnsweringConnectionId.is_null()),
1158 )
1159 .one(&*tx)
1160 .await?;
1161
1162 if let Some(pending_participant) = pending_participant {
1163 let room = self.get_room(pending_participant.room_id, &tx).await?;
1164 Ok(Self::build_incoming_call(&room, user_id))
1165 } else {
1166 Ok(None)
1167 }
1168 })
1169 .await
1170 }
1171
1172 pub async fn create_room(
1173 &self,
1174 user_id: UserId,
1175 connection: ConnectionId,
1176 live_kit_room: &str,
1177 ) -> Result<proto::Room> {
1178 self.transaction(|tx| async move {
1179 let room = room::ActiveModel {
1180 live_kit_room: ActiveValue::set(live_kit_room.into()),
1181 ..Default::default()
1182 }
1183 .insert(&*tx)
1184 .await?;
1185 room_participant::ActiveModel {
1186 room_id: ActiveValue::set(room.id),
1187 user_id: ActiveValue::set(user_id),
1188 answering_connection_id: ActiveValue::set(Some(connection.id as i32)),
1189 answering_connection_server_id: ActiveValue::set(Some(ServerId(
1190 connection.owner_id as i32,
1191 ))),
1192 answering_connection_lost: ActiveValue::set(false),
1193 calling_user_id: ActiveValue::set(user_id),
1194 calling_connection_id: ActiveValue::set(connection.id as i32),
1195 calling_connection_server_id: ActiveValue::set(Some(ServerId(
1196 connection.owner_id as i32,
1197 ))),
1198 ..Default::default()
1199 }
1200 .insert(&*tx)
1201 .await?;
1202
1203 let room = self.get_room(room.id, &tx).await?;
1204 Ok(room)
1205 })
1206 .await
1207 }
1208
1209 pub async fn call(
1210 &self,
1211 room_id: RoomId,
1212 calling_user_id: UserId,
1213 calling_connection: ConnectionId,
1214 called_user_id: UserId,
1215 initial_project_id: Option<ProjectId>,
1216 ) -> Result<RoomGuard<(proto::Room, proto::IncomingCall)>> {
1217 self.room_transaction(room_id, |tx| async move {
1218 room_participant::ActiveModel {
1219 room_id: ActiveValue::set(room_id),
1220 user_id: ActiveValue::set(called_user_id),
1221 answering_connection_lost: ActiveValue::set(false),
1222 calling_user_id: ActiveValue::set(calling_user_id),
1223 calling_connection_id: ActiveValue::set(calling_connection.id as i32),
1224 calling_connection_server_id: ActiveValue::set(Some(ServerId(
1225 calling_connection.owner_id as i32,
1226 ))),
1227 initial_project_id: ActiveValue::set(initial_project_id),
1228 ..Default::default()
1229 }
1230 .insert(&*tx)
1231 .await?;
1232
1233 let room = self.get_room(room_id, &tx).await?;
1234 let incoming_call = Self::build_incoming_call(&room, called_user_id)
1235 .ok_or_else(|| anyhow!("failed to build incoming call"))?;
1236 Ok((room, incoming_call))
1237 })
1238 .await
1239 }
1240
1241 pub async fn call_failed(
1242 &self,
1243 room_id: RoomId,
1244 called_user_id: UserId,
1245 ) -> Result<RoomGuard<proto::Room>> {
1246 self.room_transaction(room_id, |tx| async move {
1247 room_participant::Entity::delete_many()
1248 .filter(
1249 room_participant::Column::RoomId
1250 .eq(room_id)
1251 .and(room_participant::Column::UserId.eq(called_user_id)),
1252 )
1253 .exec(&*tx)
1254 .await?;
1255 let room = self.get_room(room_id, &tx).await?;
1256 Ok(room)
1257 })
1258 .await
1259 }
1260
1261 pub async fn decline_call(
1262 &self,
1263 expected_room_id: Option<RoomId>,
1264 user_id: UserId,
1265 ) -> Result<Option<RoomGuard<proto::Room>>> {
1266 self.optional_room_transaction(|tx| async move {
1267 let mut filter = Condition::all()
1268 .add(room_participant::Column::UserId.eq(user_id))
1269 .add(room_participant::Column::AnsweringConnectionId.is_null());
1270 if let Some(room_id) = expected_room_id {
1271 filter = filter.add(room_participant::Column::RoomId.eq(room_id));
1272 }
1273 let participant = room_participant::Entity::find()
1274 .filter(filter)
1275 .one(&*tx)
1276 .await?;
1277
1278 let participant = if let Some(participant) = participant {
1279 participant
1280 } else if expected_room_id.is_some() {
1281 return Err(anyhow!("could not find call to decline"))?;
1282 } else {
1283 return Ok(None);
1284 };
1285
1286 let room_id = participant.room_id;
1287 room_participant::Entity::delete(participant.into_active_model())
1288 .exec(&*tx)
1289 .await?;
1290
1291 let room = self.get_room(room_id, &tx).await?;
1292 Ok(Some((room_id, room)))
1293 })
1294 .await
1295 }
1296
1297 pub async fn cancel_call(
1298 &self,
1299 room_id: RoomId,
1300 calling_connection: ConnectionId,
1301 called_user_id: UserId,
1302 ) -> Result<RoomGuard<proto::Room>> {
1303 self.room_transaction(room_id, |tx| async move {
1304 let participant = room_participant::Entity::find()
1305 .filter(
1306 Condition::all()
1307 .add(room_participant::Column::UserId.eq(called_user_id))
1308 .add(room_participant::Column::RoomId.eq(room_id))
1309 .add(
1310 room_participant::Column::CallingConnectionId
1311 .eq(calling_connection.id as i32),
1312 )
1313 .add(
1314 room_participant::Column::CallingConnectionServerId
1315 .eq(calling_connection.owner_id as i32),
1316 )
1317 .add(room_participant::Column::AnsweringConnectionId.is_null()),
1318 )
1319 .one(&*tx)
1320 .await?
1321 .ok_or_else(|| anyhow!("no call to cancel"))?;
1322
1323 room_participant::Entity::delete(participant.into_active_model())
1324 .exec(&*tx)
1325 .await?;
1326
1327 let room = self.get_room(room_id, &tx).await?;
1328 Ok(room)
1329 })
1330 .await
1331 }
1332
1333 pub async fn join_room(
1334 &self,
1335 room_id: RoomId,
1336 user_id: UserId,
1337 connection: ConnectionId,
1338 ) -> Result<RoomGuard<proto::Room>> {
1339 self.room_transaction(room_id, |tx| async move {
1340 let result = room_participant::Entity::update_many()
1341 .filter(
1342 Condition::all()
1343 .add(room_participant::Column::RoomId.eq(room_id))
1344 .add(room_participant::Column::UserId.eq(user_id))
1345 .add(room_participant::Column::AnsweringConnectionId.is_null()),
1346 )
1347 .set(room_participant::ActiveModel {
1348 answering_connection_id: ActiveValue::set(Some(connection.id as i32)),
1349 answering_connection_server_id: ActiveValue::set(Some(ServerId(
1350 connection.owner_id as i32,
1351 ))),
1352 answering_connection_lost: ActiveValue::set(false),
1353 ..Default::default()
1354 })
1355 .exec(&*tx)
1356 .await?;
1357 if result.rows_affected == 0 {
1358 Err(anyhow!("room does not exist or was already joined"))?
1359 } else {
1360 let room = self.get_room(room_id, &tx).await?;
1361 Ok(room)
1362 }
1363 })
1364 .await
1365 }
1366
1367 pub async fn rejoin_room(
1368 &self,
1369 rejoin_room: proto::RejoinRoom,
1370 user_id: UserId,
1371 connection: ConnectionId,
1372 ) -> Result<RoomGuard<RejoinedRoom>> {
1373 let room_id = RoomId::from_proto(rejoin_room.id);
1374 self.room_transaction(room_id, |tx| async {
1375 let tx = tx;
1376 let participant_update = room_participant::Entity::update_many()
1377 .filter(
1378 Condition::all()
1379 .add(room_participant::Column::RoomId.eq(room_id))
1380 .add(room_participant::Column::UserId.eq(user_id))
1381 .add(room_participant::Column::AnsweringConnectionId.is_not_null())
1382 .add(
1383 Condition::any()
1384 .add(room_participant::Column::AnsweringConnectionLost.eq(true))
1385 .add(
1386 room_participant::Column::AnsweringConnectionServerId
1387 .ne(connection.owner_id as i32),
1388 ),
1389 ),
1390 )
1391 .set(room_participant::ActiveModel {
1392 answering_connection_id: ActiveValue::set(Some(connection.id as i32)),
1393 answering_connection_server_id: ActiveValue::set(Some(ServerId(
1394 connection.owner_id as i32,
1395 ))),
1396 answering_connection_lost: ActiveValue::set(false),
1397 ..Default::default()
1398 })
1399 .exec(&*tx)
1400 .await?;
1401 if participant_update.rows_affected == 0 {
1402 return Err(anyhow!("room does not exist or was already joined"))?;
1403 }
1404
1405 let mut reshared_projects = Vec::new();
1406 for reshared_project in &rejoin_room.reshared_projects {
1407 let project_id = ProjectId::from_proto(reshared_project.project_id);
1408 let project = project::Entity::find_by_id(project_id)
1409 .one(&*tx)
1410 .await?
1411 .ok_or_else(|| anyhow!("project does not exist"))?;
1412 if project.host_user_id != user_id {
1413 return Err(anyhow!("no such project"))?;
1414 }
1415
1416 let mut collaborators = project
1417 .find_related(project_collaborator::Entity)
1418 .all(&*tx)
1419 .await?;
1420 let host_ix = collaborators
1421 .iter()
1422 .position(|collaborator| {
1423 collaborator.user_id == user_id && collaborator.is_host
1424 })
1425 .ok_or_else(|| anyhow!("host not found among collaborators"))?;
1426 let host = collaborators.swap_remove(host_ix);
1427 let old_connection_id = host.connection();
1428
1429 project::Entity::update(project::ActiveModel {
1430 host_connection_id: ActiveValue::set(Some(connection.id as i32)),
1431 host_connection_server_id: ActiveValue::set(Some(ServerId(
1432 connection.owner_id as i32,
1433 ))),
1434 ..project.into_active_model()
1435 })
1436 .exec(&*tx)
1437 .await?;
1438 project_collaborator::Entity::update(project_collaborator::ActiveModel {
1439 connection_id: ActiveValue::set(connection.id as i32),
1440 connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
1441 ..host.into_active_model()
1442 })
1443 .exec(&*tx)
1444 .await?;
1445
1446 self.update_project_worktrees(project_id, &reshared_project.worktrees, &tx)
1447 .await?;
1448
1449 reshared_projects.push(ResharedProject {
1450 id: project_id,
1451 old_connection_id,
1452 collaborators: collaborators
1453 .iter()
1454 .map(|collaborator| ProjectCollaborator {
1455 connection_id: collaborator.connection(),
1456 user_id: collaborator.user_id,
1457 replica_id: collaborator.replica_id,
1458 is_host: collaborator.is_host,
1459 })
1460 .collect(),
1461 worktrees: reshared_project.worktrees.clone(),
1462 });
1463 }
1464
1465 project::Entity::delete_many()
1466 .filter(
1467 Condition::all()
1468 .add(project::Column::RoomId.eq(room_id))
1469 .add(project::Column::HostUserId.eq(user_id))
1470 .add(
1471 project::Column::Id
1472 .is_not_in(reshared_projects.iter().map(|project| project.id)),
1473 ),
1474 )
1475 .exec(&*tx)
1476 .await?;
1477
1478 let mut rejoined_projects = Vec::new();
1479 for rejoined_project in &rejoin_room.rejoined_projects {
1480 let project_id = ProjectId::from_proto(rejoined_project.id);
1481 let Some(project) = project::Entity::find_by_id(project_id)
1482 .one(&*tx)
1483 .await? else { continue };
1484
1485 let mut worktrees = Vec::new();
1486 let db_worktrees = project.find_related(worktree::Entity).all(&*tx).await?;
1487 for db_worktree in db_worktrees {
1488 let mut worktree = RejoinedWorktree {
1489 id: db_worktree.id as u64,
1490 abs_path: db_worktree.abs_path,
1491 root_name: db_worktree.root_name,
1492 visible: db_worktree.visible,
1493 updated_entries: Default::default(),
1494 removed_entries: Default::default(),
1495 updated_repositories: Default::default(),
1496 removed_repositories: Default::default(),
1497 diagnostic_summaries: Default::default(),
1498 settings_files: Default::default(),
1499 scan_id: db_worktree.scan_id as u64,
1500 completed_scan_id: db_worktree.completed_scan_id as u64,
1501 };
1502
1503 let rejoined_worktree = rejoined_project
1504 .worktrees
1505 .iter()
1506 .find(|worktree| worktree.id == db_worktree.id as u64);
1507
1508 // File entries
1509 {
1510 let entry_filter = if let Some(rejoined_worktree) = rejoined_worktree {
1511 worktree_entry::Column::ScanId.gt(rejoined_worktree.scan_id)
1512 } else {
1513 worktree_entry::Column::IsDeleted.eq(false)
1514 };
1515
1516 let mut db_entries = worktree_entry::Entity::find()
1517 .filter(
1518 Condition::all()
1519 .add(worktree_entry::Column::ProjectId.eq(project.id))
1520 .add(worktree_entry::Column::WorktreeId.eq(worktree.id))
1521 .add(entry_filter),
1522 )
1523 .stream(&*tx)
1524 .await?;
1525
1526 while let Some(db_entry) = db_entries.next().await {
1527 let db_entry = db_entry?;
1528 if db_entry.is_deleted {
1529 worktree.removed_entries.push(db_entry.id as u64);
1530 } else {
1531 worktree.updated_entries.push(proto::Entry {
1532 id: db_entry.id as u64,
1533 is_dir: db_entry.is_dir,
1534 path: db_entry.path,
1535 inode: db_entry.inode as u64,
1536 mtime: Some(proto::Timestamp {
1537 seconds: db_entry.mtime_seconds as u64,
1538 nanos: db_entry.mtime_nanos as u32,
1539 }),
1540 is_symlink: db_entry.is_symlink,
1541 is_ignored: db_entry.is_ignored,
1542 git_status: db_entry.git_status.map(|status| status as i32),
1543 });
1544 }
1545 }
1546 }
1547
1548 // Repository Entries
1549 {
1550 let repository_entry_filter =
1551 if let Some(rejoined_worktree) = rejoined_worktree {
1552 worktree_repository::Column::ScanId.gt(rejoined_worktree.scan_id)
1553 } else {
1554 worktree_repository::Column::IsDeleted.eq(false)
1555 };
1556
1557 let mut db_repositories = worktree_repository::Entity::find()
1558 .filter(
1559 Condition::all()
1560 .add(worktree_repository::Column::ProjectId.eq(project.id))
1561 .add(worktree_repository::Column::WorktreeId.eq(worktree.id))
1562 .add(repository_entry_filter),
1563 )
1564 .stream(&*tx)
1565 .await?;
1566
1567 while let Some(db_repository) = db_repositories.next().await {
1568 let db_repository = db_repository?;
1569 if db_repository.is_deleted {
1570 worktree
1571 .removed_repositories
1572 .push(db_repository.work_directory_id as u64);
1573 } else {
1574 worktree.updated_repositories.push(proto::RepositoryEntry {
1575 work_directory_id: db_repository.work_directory_id as u64,
1576 branch: db_repository.branch,
1577 });
1578 }
1579 }
1580 }
1581
1582 worktrees.push(worktree);
1583 }
1584
1585 let language_servers = project
1586 .find_related(language_server::Entity)
1587 .all(&*tx)
1588 .await?
1589 .into_iter()
1590 .map(|language_server| proto::LanguageServer {
1591 id: language_server.id as u64,
1592 name: language_server.name,
1593 })
1594 .collect::<Vec<_>>();
1595
1596 {
1597 let mut db_settings_files = worktree_settings_file::Entity::find()
1598 .filter(worktree_settings_file::Column::ProjectId.eq(project_id))
1599 .stream(&*tx)
1600 .await?;
1601 while let Some(db_settings_file) = db_settings_files.next().await {
1602 let db_settings_file = db_settings_file?;
1603 if let Some(worktree) = worktrees
1604 .iter_mut()
1605 .find(|w| w.id == db_settings_file.worktree_id as u64)
1606 {
1607 worktree.settings_files.push(WorktreeSettingsFile {
1608 path: db_settings_file.path,
1609 content: db_settings_file.content,
1610 });
1611 }
1612 }
1613 }
1614
1615 let mut collaborators = project
1616 .find_related(project_collaborator::Entity)
1617 .all(&*tx)
1618 .await?;
1619 let self_collaborator = if let Some(self_collaborator_ix) = collaborators
1620 .iter()
1621 .position(|collaborator| collaborator.user_id == user_id)
1622 {
1623 collaborators.swap_remove(self_collaborator_ix)
1624 } else {
1625 continue;
1626 };
1627 let old_connection_id = self_collaborator.connection();
1628 project_collaborator::Entity::update(project_collaborator::ActiveModel {
1629 connection_id: ActiveValue::set(connection.id as i32),
1630 connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
1631 ..self_collaborator.into_active_model()
1632 })
1633 .exec(&*tx)
1634 .await?;
1635
1636 let collaborators = collaborators
1637 .into_iter()
1638 .map(|collaborator| ProjectCollaborator {
1639 connection_id: collaborator.connection(),
1640 user_id: collaborator.user_id,
1641 replica_id: collaborator.replica_id,
1642 is_host: collaborator.is_host,
1643 })
1644 .collect::<Vec<_>>();
1645
1646 rejoined_projects.push(RejoinedProject {
1647 id: project_id,
1648 old_connection_id,
1649 collaborators,
1650 worktrees,
1651 language_servers,
1652 });
1653 }
1654
1655 let room = self.get_room(room_id, &tx).await?;
1656 Ok(RejoinedRoom {
1657 room,
1658 rejoined_projects,
1659 reshared_projects,
1660 })
1661 })
1662 .await
1663 }
1664
1665 pub async fn leave_room(
1666 &self,
1667 connection: ConnectionId,
1668 ) -> Result<Option<RoomGuard<LeftRoom>>> {
1669 self.optional_room_transaction(|tx| async move {
1670 let leaving_participant = room_participant::Entity::find()
1671 .filter(
1672 Condition::all()
1673 .add(
1674 room_participant::Column::AnsweringConnectionId
1675 .eq(connection.id as i32),
1676 )
1677 .add(
1678 room_participant::Column::AnsweringConnectionServerId
1679 .eq(connection.owner_id as i32),
1680 ),
1681 )
1682 .one(&*tx)
1683 .await?;
1684
1685 if let Some(leaving_participant) = leaving_participant {
1686 // Leave room.
1687 let room_id = leaving_participant.room_id;
1688 room_participant::Entity::delete_by_id(leaving_participant.id)
1689 .exec(&*tx)
1690 .await?;
1691
1692 // Cancel pending calls initiated by the leaving user.
1693 let called_participants = room_participant::Entity::find()
1694 .filter(
1695 Condition::all()
1696 .add(
1697 room_participant::Column::CallingUserId
1698 .eq(leaving_participant.user_id),
1699 )
1700 .add(room_participant::Column::AnsweringConnectionId.is_null()),
1701 )
1702 .all(&*tx)
1703 .await?;
1704 room_participant::Entity::delete_many()
1705 .filter(
1706 room_participant::Column::Id
1707 .is_in(called_participants.iter().map(|participant| participant.id)),
1708 )
1709 .exec(&*tx)
1710 .await?;
1711 let canceled_calls_to_user_ids = called_participants
1712 .into_iter()
1713 .map(|participant| participant.user_id)
1714 .collect();
1715
1716 // Detect left projects.
1717 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
1718 enum QueryProjectIds {
1719 ProjectId,
1720 }
1721 let project_ids: Vec<ProjectId> = project_collaborator::Entity::find()
1722 .select_only()
1723 .column_as(
1724 project_collaborator::Column::ProjectId,
1725 QueryProjectIds::ProjectId,
1726 )
1727 .filter(
1728 Condition::all()
1729 .add(
1730 project_collaborator::Column::ConnectionId.eq(connection.id as i32),
1731 )
1732 .add(
1733 project_collaborator::Column::ConnectionServerId
1734 .eq(connection.owner_id as i32),
1735 ),
1736 )
1737 .into_values::<_, QueryProjectIds>()
1738 .all(&*tx)
1739 .await?;
1740 let mut left_projects = HashMap::default();
1741 let mut collaborators = project_collaborator::Entity::find()
1742 .filter(project_collaborator::Column::ProjectId.is_in(project_ids))
1743 .stream(&*tx)
1744 .await?;
1745 while let Some(collaborator) = collaborators.next().await {
1746 let collaborator = collaborator?;
1747 let left_project =
1748 left_projects
1749 .entry(collaborator.project_id)
1750 .or_insert(LeftProject {
1751 id: collaborator.project_id,
1752 host_user_id: Default::default(),
1753 connection_ids: Default::default(),
1754 host_connection_id: Default::default(),
1755 });
1756
1757 let collaborator_connection_id = collaborator.connection();
1758 if collaborator_connection_id != connection {
1759 left_project.connection_ids.push(collaborator_connection_id);
1760 }
1761
1762 if collaborator.is_host {
1763 left_project.host_user_id = collaborator.user_id;
1764 left_project.host_connection_id = collaborator_connection_id;
1765 }
1766 }
1767 drop(collaborators);
1768
1769 // Leave projects.
1770 project_collaborator::Entity::delete_many()
1771 .filter(
1772 Condition::all()
1773 .add(
1774 project_collaborator::Column::ConnectionId.eq(connection.id as i32),
1775 )
1776 .add(
1777 project_collaborator::Column::ConnectionServerId
1778 .eq(connection.owner_id as i32),
1779 ),
1780 )
1781 .exec(&*tx)
1782 .await?;
1783
1784 // Unshare projects.
1785 project::Entity::delete_many()
1786 .filter(
1787 Condition::all()
1788 .add(project::Column::RoomId.eq(room_id))
1789 .add(project::Column::HostConnectionId.eq(connection.id as i32))
1790 .add(
1791 project::Column::HostConnectionServerId
1792 .eq(connection.owner_id as i32),
1793 ),
1794 )
1795 .exec(&*tx)
1796 .await?;
1797
1798 let room = self.get_room(room_id, &tx).await?;
1799 if room.participants.is_empty() {
1800 room::Entity::delete_by_id(room_id).exec(&*tx).await?;
1801 }
1802
1803 let left_room = LeftRoom {
1804 room,
1805 left_projects,
1806 canceled_calls_to_user_ids,
1807 };
1808
1809 if left_room.room.participants.is_empty() {
1810 self.rooms.remove(&room_id);
1811 }
1812
1813 Ok(Some((room_id, left_room)))
1814 } else {
1815 Ok(None)
1816 }
1817 })
1818 .await
1819 }
1820
1821 pub async fn follow(
1822 &self,
1823 project_id: ProjectId,
1824 leader_connection: ConnectionId,
1825 follower_connection: ConnectionId,
1826 ) -> Result<RoomGuard<proto::Room>> {
1827 let room_id = self.room_id_for_project(project_id).await?;
1828 self.room_transaction(room_id, |tx| async move {
1829 follower::ActiveModel {
1830 room_id: ActiveValue::set(room_id),
1831 project_id: ActiveValue::set(project_id),
1832 leader_connection_server_id: ActiveValue::set(ServerId(
1833 leader_connection.owner_id as i32,
1834 )),
1835 leader_connection_id: ActiveValue::set(leader_connection.id as i32),
1836 follower_connection_server_id: ActiveValue::set(ServerId(
1837 follower_connection.owner_id as i32,
1838 )),
1839 follower_connection_id: ActiveValue::set(follower_connection.id as i32),
1840 ..Default::default()
1841 }
1842 .insert(&*tx)
1843 .await?;
1844
1845 let room = self.get_room(room_id, &*tx).await?;
1846 Ok(room)
1847 })
1848 .await
1849 }
1850
1851 pub async fn unfollow(
1852 &self,
1853 project_id: ProjectId,
1854 leader_connection: ConnectionId,
1855 follower_connection: ConnectionId,
1856 ) -> Result<RoomGuard<proto::Room>> {
1857 let room_id = self.room_id_for_project(project_id).await?;
1858 self.room_transaction(room_id, |tx| async move {
1859 follower::Entity::delete_many()
1860 .filter(
1861 Condition::all()
1862 .add(follower::Column::ProjectId.eq(project_id))
1863 .add(
1864 follower::Column::LeaderConnectionServerId
1865 .eq(leader_connection.owner_id),
1866 )
1867 .add(follower::Column::LeaderConnectionId.eq(leader_connection.id))
1868 .add(
1869 follower::Column::FollowerConnectionServerId
1870 .eq(follower_connection.owner_id),
1871 )
1872 .add(follower::Column::FollowerConnectionId.eq(follower_connection.id)),
1873 )
1874 .exec(&*tx)
1875 .await?;
1876
1877 let room = self.get_room(room_id, &*tx).await?;
1878 Ok(room)
1879 })
1880 .await
1881 }
1882
1883 pub async fn update_room_participant_location(
1884 &self,
1885 room_id: RoomId,
1886 connection: ConnectionId,
1887 location: proto::ParticipantLocation,
1888 ) -> Result<RoomGuard<proto::Room>> {
1889 self.room_transaction(room_id, |tx| async {
1890 let tx = tx;
1891 let location_kind;
1892 let location_project_id;
1893 match location
1894 .variant
1895 .as_ref()
1896 .ok_or_else(|| anyhow!("invalid location"))?
1897 {
1898 proto::participant_location::Variant::SharedProject(project) => {
1899 location_kind = 0;
1900 location_project_id = Some(ProjectId::from_proto(project.id));
1901 }
1902 proto::participant_location::Variant::UnsharedProject(_) => {
1903 location_kind = 1;
1904 location_project_id = None;
1905 }
1906 proto::participant_location::Variant::External(_) => {
1907 location_kind = 2;
1908 location_project_id = None;
1909 }
1910 }
1911
1912 let result = room_participant::Entity::update_many()
1913 .filter(
1914 Condition::all()
1915 .add(room_participant::Column::RoomId.eq(room_id))
1916 .add(
1917 room_participant::Column::AnsweringConnectionId
1918 .eq(connection.id as i32),
1919 )
1920 .add(
1921 room_participant::Column::AnsweringConnectionServerId
1922 .eq(connection.owner_id as i32),
1923 ),
1924 )
1925 .set(room_participant::ActiveModel {
1926 location_kind: ActiveValue::set(Some(location_kind)),
1927 location_project_id: ActiveValue::set(location_project_id),
1928 ..Default::default()
1929 })
1930 .exec(&*tx)
1931 .await?;
1932
1933 if result.rows_affected == 1 {
1934 let room = self.get_room(room_id, &tx).await?;
1935 Ok(room)
1936 } else {
1937 Err(anyhow!("could not update room participant location"))?
1938 }
1939 })
1940 .await
1941 }
1942
1943 pub async fn connection_lost(&self, connection: ConnectionId) -> Result<()> {
1944 self.transaction(|tx| async move {
1945 let participant = room_participant::Entity::find()
1946 .filter(
1947 Condition::all()
1948 .add(
1949 room_participant::Column::AnsweringConnectionId
1950 .eq(connection.id as i32),
1951 )
1952 .add(
1953 room_participant::Column::AnsweringConnectionServerId
1954 .eq(connection.owner_id as i32),
1955 ),
1956 )
1957 .one(&*tx)
1958 .await?
1959 .ok_or_else(|| anyhow!("not a participant in any room"))?;
1960
1961 room_participant::Entity::update(room_participant::ActiveModel {
1962 answering_connection_lost: ActiveValue::set(true),
1963 ..participant.into_active_model()
1964 })
1965 .exec(&*tx)
1966 .await?;
1967
1968 Ok(())
1969 })
1970 .await
1971 }
1972
1973 fn build_incoming_call(
1974 room: &proto::Room,
1975 called_user_id: UserId,
1976 ) -> Option<proto::IncomingCall> {
1977 let pending_participant = room
1978 .pending_participants
1979 .iter()
1980 .find(|participant| participant.user_id == called_user_id.to_proto())?;
1981
1982 Some(proto::IncomingCall {
1983 room_id: room.id,
1984 calling_user_id: pending_participant.calling_user_id,
1985 participant_user_ids: room
1986 .participants
1987 .iter()
1988 .map(|participant| participant.user_id)
1989 .collect(),
1990 initial_project: room.participants.iter().find_map(|participant| {
1991 let initial_project_id = pending_participant.initial_project_id?;
1992 participant
1993 .projects
1994 .iter()
1995 .find(|project| project.id == initial_project_id)
1996 .cloned()
1997 }),
1998 })
1999 }
2000
2001 async fn get_room(&self, room_id: RoomId, tx: &DatabaseTransaction) -> Result<proto::Room> {
2002 let db_room = room::Entity::find_by_id(room_id)
2003 .one(tx)
2004 .await?
2005 .ok_or_else(|| anyhow!("could not find room"))?;
2006
2007 let mut db_participants = db_room
2008 .find_related(room_participant::Entity)
2009 .stream(tx)
2010 .await?;
2011 let mut participants = HashMap::default();
2012 let mut pending_participants = Vec::new();
2013 while let Some(db_participant) = db_participants.next().await {
2014 let db_participant = db_participant?;
2015 if let Some((answering_connection_id, answering_connection_server_id)) = db_participant
2016 .answering_connection_id
2017 .zip(db_participant.answering_connection_server_id)
2018 {
2019 let location = match (
2020 db_participant.location_kind,
2021 db_participant.location_project_id,
2022 ) {
2023 (Some(0), Some(project_id)) => {
2024 Some(proto::participant_location::Variant::SharedProject(
2025 proto::participant_location::SharedProject {
2026 id: project_id.to_proto(),
2027 },
2028 ))
2029 }
2030 (Some(1), _) => Some(proto::participant_location::Variant::UnsharedProject(
2031 Default::default(),
2032 )),
2033 _ => Some(proto::participant_location::Variant::External(
2034 Default::default(),
2035 )),
2036 };
2037
2038 let answering_connection = ConnectionId {
2039 owner_id: answering_connection_server_id.0 as u32,
2040 id: answering_connection_id as u32,
2041 };
2042 participants.insert(
2043 answering_connection,
2044 proto::Participant {
2045 user_id: db_participant.user_id.to_proto(),
2046 peer_id: Some(answering_connection.into()),
2047 projects: Default::default(),
2048 location: Some(proto::ParticipantLocation { variant: location }),
2049 },
2050 );
2051 } else {
2052 pending_participants.push(proto::PendingParticipant {
2053 user_id: db_participant.user_id.to_proto(),
2054 calling_user_id: db_participant.calling_user_id.to_proto(),
2055 initial_project_id: db_participant.initial_project_id.map(|id| id.to_proto()),
2056 });
2057 }
2058 }
2059 drop(db_participants);
2060
2061 let mut db_projects = db_room
2062 .find_related(project::Entity)
2063 .find_with_related(worktree::Entity)
2064 .stream(tx)
2065 .await?;
2066
2067 while let Some(row) = db_projects.next().await {
2068 let (db_project, db_worktree) = row?;
2069 let host_connection = db_project.host_connection()?;
2070 if let Some(participant) = participants.get_mut(&host_connection) {
2071 let project = if let Some(project) = participant
2072 .projects
2073 .iter_mut()
2074 .find(|project| project.id == db_project.id.to_proto())
2075 {
2076 project
2077 } else {
2078 participant.projects.push(proto::ParticipantProject {
2079 id: db_project.id.to_proto(),
2080 worktree_root_names: Default::default(),
2081 });
2082 participant.projects.last_mut().unwrap()
2083 };
2084
2085 if let Some(db_worktree) = db_worktree {
2086 if db_worktree.visible {
2087 project.worktree_root_names.push(db_worktree.root_name);
2088 }
2089 }
2090 }
2091 }
2092 drop(db_projects);
2093
2094 let mut db_followers = db_room.find_related(follower::Entity).stream(tx).await?;
2095 let mut followers = Vec::new();
2096 while let Some(db_follower) = db_followers.next().await {
2097 let db_follower = db_follower?;
2098 followers.push(proto::Follower {
2099 leader_id: Some(db_follower.leader_connection().into()),
2100 follower_id: Some(db_follower.follower_connection().into()),
2101 project_id: db_follower.project_id.to_proto(),
2102 });
2103 }
2104
2105 Ok(proto::Room {
2106 id: db_room.id.to_proto(),
2107 live_kit_room: db_room.live_kit_room,
2108 participants: participants.into_values().collect(),
2109 pending_participants,
2110 followers,
2111 })
2112 }
2113
2114 // projects
2115
2116 pub async fn project_count_excluding_admins(&self) -> Result<usize> {
2117 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
2118 enum QueryAs {
2119 Count,
2120 }
2121
2122 self.transaction(|tx| async move {
2123 Ok(project::Entity::find()
2124 .select_only()
2125 .column_as(project::Column::Id.count(), QueryAs::Count)
2126 .inner_join(user::Entity)
2127 .filter(user::Column::Admin.eq(false))
2128 .into_values::<_, QueryAs>()
2129 .one(&*tx)
2130 .await?
2131 .unwrap_or(0i64) as usize)
2132 })
2133 .await
2134 }
2135
2136 pub async fn share_project(
2137 &self,
2138 room_id: RoomId,
2139 connection: ConnectionId,
2140 worktrees: &[proto::WorktreeMetadata],
2141 ) -> Result<RoomGuard<(ProjectId, proto::Room)>> {
2142 self.room_transaction(room_id, |tx| async move {
2143 let participant = room_participant::Entity::find()
2144 .filter(
2145 Condition::all()
2146 .add(
2147 room_participant::Column::AnsweringConnectionId
2148 .eq(connection.id as i32),
2149 )
2150 .add(
2151 room_participant::Column::AnsweringConnectionServerId
2152 .eq(connection.owner_id as i32),
2153 ),
2154 )
2155 .one(&*tx)
2156 .await?
2157 .ok_or_else(|| anyhow!("could not find participant"))?;
2158 if participant.room_id != room_id {
2159 return Err(anyhow!("shared project on unexpected room"))?;
2160 }
2161
2162 let project = project::ActiveModel {
2163 room_id: ActiveValue::set(participant.room_id),
2164 host_user_id: ActiveValue::set(participant.user_id),
2165 host_connection_id: ActiveValue::set(Some(connection.id as i32)),
2166 host_connection_server_id: ActiveValue::set(Some(ServerId(
2167 connection.owner_id as i32,
2168 ))),
2169 ..Default::default()
2170 }
2171 .insert(&*tx)
2172 .await?;
2173
2174 if !worktrees.is_empty() {
2175 worktree::Entity::insert_many(worktrees.iter().map(|worktree| {
2176 worktree::ActiveModel {
2177 id: ActiveValue::set(worktree.id as i64),
2178 project_id: ActiveValue::set(project.id),
2179 abs_path: ActiveValue::set(worktree.abs_path.clone()),
2180 root_name: ActiveValue::set(worktree.root_name.clone()),
2181 visible: ActiveValue::set(worktree.visible),
2182 scan_id: ActiveValue::set(0),
2183 completed_scan_id: ActiveValue::set(0),
2184 }
2185 }))
2186 .exec(&*tx)
2187 .await?;
2188 }
2189
2190 project_collaborator::ActiveModel {
2191 project_id: ActiveValue::set(project.id),
2192 connection_id: ActiveValue::set(connection.id as i32),
2193 connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
2194 user_id: ActiveValue::set(participant.user_id),
2195 replica_id: ActiveValue::set(ReplicaId(0)),
2196 is_host: ActiveValue::set(true),
2197 ..Default::default()
2198 }
2199 .insert(&*tx)
2200 .await?;
2201
2202 let room = self.get_room(room_id, &tx).await?;
2203 Ok((project.id, room))
2204 })
2205 .await
2206 }
2207
2208 pub async fn unshare_project(
2209 &self,
2210 project_id: ProjectId,
2211 connection: ConnectionId,
2212 ) -> Result<RoomGuard<(proto::Room, Vec<ConnectionId>)>> {
2213 let room_id = self.room_id_for_project(project_id).await?;
2214 self.room_transaction(room_id, |tx| async move {
2215 let guest_connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
2216
2217 let project = project::Entity::find_by_id(project_id)
2218 .one(&*tx)
2219 .await?
2220 .ok_or_else(|| anyhow!("project not found"))?;
2221 if project.host_connection()? == connection {
2222 project::Entity::delete(project.into_active_model())
2223 .exec(&*tx)
2224 .await?;
2225 let room = self.get_room(room_id, &tx).await?;
2226 Ok((room, guest_connection_ids))
2227 } else {
2228 Err(anyhow!("cannot unshare a project hosted by another user"))?
2229 }
2230 })
2231 .await
2232 }
2233
2234 pub async fn update_project(
2235 &self,
2236 project_id: ProjectId,
2237 connection: ConnectionId,
2238 worktrees: &[proto::WorktreeMetadata],
2239 ) -> Result<RoomGuard<(proto::Room, Vec<ConnectionId>)>> {
2240 let room_id = self.room_id_for_project(project_id).await?;
2241 self.room_transaction(room_id, |tx| async move {
2242 let project = project::Entity::find_by_id(project_id)
2243 .filter(
2244 Condition::all()
2245 .add(project::Column::HostConnectionId.eq(connection.id as i32))
2246 .add(
2247 project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
2248 ),
2249 )
2250 .one(&*tx)
2251 .await?
2252 .ok_or_else(|| anyhow!("no such project"))?;
2253
2254 self.update_project_worktrees(project.id, worktrees, &tx)
2255 .await?;
2256
2257 let guest_connection_ids = self.project_guest_connection_ids(project.id, &tx).await?;
2258 let room = self.get_room(project.room_id, &tx).await?;
2259 Ok((room, guest_connection_ids))
2260 })
2261 .await
2262 }
2263
2264 async fn update_project_worktrees(
2265 &self,
2266 project_id: ProjectId,
2267 worktrees: &[proto::WorktreeMetadata],
2268 tx: &DatabaseTransaction,
2269 ) -> Result<()> {
2270 if !worktrees.is_empty() {
2271 worktree::Entity::insert_many(worktrees.iter().map(|worktree| worktree::ActiveModel {
2272 id: ActiveValue::set(worktree.id as i64),
2273 project_id: ActiveValue::set(project_id),
2274 abs_path: ActiveValue::set(worktree.abs_path.clone()),
2275 root_name: ActiveValue::set(worktree.root_name.clone()),
2276 visible: ActiveValue::set(worktree.visible),
2277 scan_id: ActiveValue::set(0),
2278 completed_scan_id: ActiveValue::set(0),
2279 }))
2280 .on_conflict(
2281 OnConflict::columns([worktree::Column::ProjectId, worktree::Column::Id])
2282 .update_column(worktree::Column::RootName)
2283 .to_owned(),
2284 )
2285 .exec(&*tx)
2286 .await?;
2287 }
2288
2289 worktree::Entity::delete_many()
2290 .filter(worktree::Column::ProjectId.eq(project_id).and(
2291 worktree::Column::Id.is_not_in(worktrees.iter().map(|worktree| worktree.id as i64)),
2292 ))
2293 .exec(&*tx)
2294 .await?;
2295
2296 Ok(())
2297 }
2298
2299 pub async fn update_worktree(
2300 &self,
2301 update: &proto::UpdateWorktree,
2302 connection: ConnectionId,
2303 ) -> Result<RoomGuard<Vec<ConnectionId>>> {
2304 let project_id = ProjectId::from_proto(update.project_id);
2305 let worktree_id = update.worktree_id as i64;
2306 let room_id = self.room_id_for_project(project_id).await?;
2307 self.room_transaction(room_id, |tx| async move {
2308 // Ensure the update comes from the host.
2309 let _project = project::Entity::find_by_id(project_id)
2310 .filter(
2311 Condition::all()
2312 .add(project::Column::HostConnectionId.eq(connection.id as i32))
2313 .add(
2314 project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
2315 ),
2316 )
2317 .one(&*tx)
2318 .await?
2319 .ok_or_else(|| anyhow!("no such project"))?;
2320
2321 // Update metadata.
2322 worktree::Entity::update(worktree::ActiveModel {
2323 id: ActiveValue::set(worktree_id),
2324 project_id: ActiveValue::set(project_id),
2325 root_name: ActiveValue::set(update.root_name.clone()),
2326 scan_id: ActiveValue::set(update.scan_id as i64),
2327 completed_scan_id: if update.is_last_update {
2328 ActiveValue::set(update.scan_id as i64)
2329 } else {
2330 ActiveValue::default()
2331 },
2332 abs_path: ActiveValue::set(update.abs_path.clone()),
2333 ..Default::default()
2334 })
2335 .exec(&*tx)
2336 .await?;
2337
2338 if !update.updated_entries.is_empty() {
2339 worktree_entry::Entity::insert_many(update.updated_entries.iter().map(|entry| {
2340 let mtime = entry.mtime.clone().unwrap_or_default();
2341 worktree_entry::ActiveModel {
2342 project_id: ActiveValue::set(project_id),
2343 worktree_id: ActiveValue::set(worktree_id),
2344 id: ActiveValue::set(entry.id as i64),
2345 is_dir: ActiveValue::set(entry.is_dir),
2346 path: ActiveValue::set(entry.path.clone()),
2347 inode: ActiveValue::set(entry.inode as i64),
2348 mtime_seconds: ActiveValue::set(mtime.seconds as i64),
2349 mtime_nanos: ActiveValue::set(mtime.nanos as i32),
2350 is_symlink: ActiveValue::set(entry.is_symlink),
2351 is_ignored: ActiveValue::set(entry.is_ignored),
2352 git_status: ActiveValue::set(entry.git_status.map(|status| status as i64)),
2353 is_deleted: ActiveValue::set(false),
2354 scan_id: ActiveValue::set(update.scan_id as i64),
2355 }
2356 }))
2357 .on_conflict(
2358 OnConflict::columns([
2359 worktree_entry::Column::ProjectId,
2360 worktree_entry::Column::WorktreeId,
2361 worktree_entry::Column::Id,
2362 ])
2363 .update_columns([
2364 worktree_entry::Column::IsDir,
2365 worktree_entry::Column::Path,
2366 worktree_entry::Column::Inode,
2367 worktree_entry::Column::MtimeSeconds,
2368 worktree_entry::Column::MtimeNanos,
2369 worktree_entry::Column::IsSymlink,
2370 worktree_entry::Column::IsIgnored,
2371 worktree_entry::Column::GitStatus,
2372 worktree_entry::Column::ScanId,
2373 ])
2374 .to_owned(),
2375 )
2376 .exec(&*tx)
2377 .await?;
2378 }
2379
2380 if !update.removed_entries.is_empty() {
2381 worktree_entry::Entity::update_many()
2382 .filter(
2383 worktree_entry::Column::ProjectId
2384 .eq(project_id)
2385 .and(worktree_entry::Column::WorktreeId.eq(worktree_id))
2386 .and(
2387 worktree_entry::Column::Id
2388 .is_in(update.removed_entries.iter().map(|id| *id as i64)),
2389 ),
2390 )
2391 .set(worktree_entry::ActiveModel {
2392 is_deleted: ActiveValue::Set(true),
2393 scan_id: ActiveValue::Set(update.scan_id as i64),
2394 ..Default::default()
2395 })
2396 .exec(&*tx)
2397 .await?;
2398 }
2399
2400 if !update.updated_repositories.is_empty() {
2401 worktree_repository::Entity::insert_many(update.updated_repositories.iter().map(
2402 |repository| worktree_repository::ActiveModel {
2403 project_id: ActiveValue::set(project_id),
2404 worktree_id: ActiveValue::set(worktree_id),
2405 work_directory_id: ActiveValue::set(repository.work_directory_id as i64),
2406 scan_id: ActiveValue::set(update.scan_id as i64),
2407 branch: ActiveValue::set(repository.branch.clone()),
2408 is_deleted: ActiveValue::set(false),
2409 },
2410 ))
2411 .on_conflict(
2412 OnConflict::columns([
2413 worktree_repository::Column::ProjectId,
2414 worktree_repository::Column::WorktreeId,
2415 worktree_repository::Column::WorkDirectoryId,
2416 ])
2417 .update_columns([
2418 worktree_repository::Column::ScanId,
2419 worktree_repository::Column::Branch,
2420 ])
2421 .to_owned(),
2422 )
2423 .exec(&*tx)
2424 .await?;
2425 }
2426
2427 if !update.removed_repositories.is_empty() {
2428 worktree_repository::Entity::update_many()
2429 .filter(
2430 worktree_repository::Column::ProjectId
2431 .eq(project_id)
2432 .and(worktree_repository::Column::WorktreeId.eq(worktree_id))
2433 .and(
2434 worktree_repository::Column::WorkDirectoryId
2435 .is_in(update.removed_repositories.iter().map(|id| *id as i64)),
2436 ),
2437 )
2438 .set(worktree_repository::ActiveModel {
2439 is_deleted: ActiveValue::Set(true),
2440 scan_id: ActiveValue::Set(update.scan_id as i64),
2441 ..Default::default()
2442 })
2443 .exec(&*tx)
2444 .await?;
2445 }
2446
2447 let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
2448 Ok(connection_ids)
2449 })
2450 .await
2451 }
2452
2453 pub async fn update_diagnostic_summary(
2454 &self,
2455 update: &proto::UpdateDiagnosticSummary,
2456 connection: ConnectionId,
2457 ) -> Result<RoomGuard<Vec<ConnectionId>>> {
2458 let project_id = ProjectId::from_proto(update.project_id);
2459 let worktree_id = update.worktree_id as i64;
2460 let room_id = self.room_id_for_project(project_id).await?;
2461 self.room_transaction(room_id, |tx| async move {
2462 let summary = update
2463 .summary
2464 .as_ref()
2465 .ok_or_else(|| anyhow!("invalid summary"))?;
2466
2467 // Ensure the update comes from the host.
2468 let project = project::Entity::find_by_id(project_id)
2469 .one(&*tx)
2470 .await?
2471 .ok_or_else(|| anyhow!("no such project"))?;
2472 if project.host_connection()? != connection {
2473 return Err(anyhow!("can't update a project hosted by someone else"))?;
2474 }
2475
2476 // Update summary.
2477 worktree_diagnostic_summary::Entity::insert(worktree_diagnostic_summary::ActiveModel {
2478 project_id: ActiveValue::set(project_id),
2479 worktree_id: ActiveValue::set(worktree_id),
2480 path: ActiveValue::set(summary.path.clone()),
2481 language_server_id: ActiveValue::set(summary.language_server_id as i64),
2482 error_count: ActiveValue::set(summary.error_count as i32),
2483 warning_count: ActiveValue::set(summary.warning_count as i32),
2484 ..Default::default()
2485 })
2486 .on_conflict(
2487 OnConflict::columns([
2488 worktree_diagnostic_summary::Column::ProjectId,
2489 worktree_diagnostic_summary::Column::WorktreeId,
2490 worktree_diagnostic_summary::Column::Path,
2491 ])
2492 .update_columns([
2493 worktree_diagnostic_summary::Column::LanguageServerId,
2494 worktree_diagnostic_summary::Column::ErrorCount,
2495 worktree_diagnostic_summary::Column::WarningCount,
2496 ])
2497 .to_owned(),
2498 )
2499 .exec(&*tx)
2500 .await?;
2501
2502 let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
2503 Ok(connection_ids)
2504 })
2505 .await
2506 }
2507
2508 pub async fn start_language_server(
2509 &self,
2510 update: &proto::StartLanguageServer,
2511 connection: ConnectionId,
2512 ) -> Result<RoomGuard<Vec<ConnectionId>>> {
2513 let project_id = ProjectId::from_proto(update.project_id);
2514 let room_id = self.room_id_for_project(project_id).await?;
2515 self.room_transaction(room_id, |tx| async move {
2516 let server = update
2517 .server
2518 .as_ref()
2519 .ok_or_else(|| anyhow!("invalid language server"))?;
2520
2521 // Ensure the update comes from the host.
2522 let project = project::Entity::find_by_id(project_id)
2523 .one(&*tx)
2524 .await?
2525 .ok_or_else(|| anyhow!("no such project"))?;
2526 if project.host_connection()? != connection {
2527 return Err(anyhow!("can't update a project hosted by someone else"))?;
2528 }
2529
2530 // Add the newly-started language server.
2531 language_server::Entity::insert(language_server::ActiveModel {
2532 project_id: ActiveValue::set(project_id),
2533 id: ActiveValue::set(server.id as i64),
2534 name: ActiveValue::set(server.name.clone()),
2535 ..Default::default()
2536 })
2537 .on_conflict(
2538 OnConflict::columns([
2539 language_server::Column::ProjectId,
2540 language_server::Column::Id,
2541 ])
2542 .update_column(language_server::Column::Name)
2543 .to_owned(),
2544 )
2545 .exec(&*tx)
2546 .await?;
2547
2548 let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
2549 Ok(connection_ids)
2550 })
2551 .await
2552 }
2553
2554 pub async fn update_worktree_settings(
2555 &self,
2556 update: &proto::UpdateWorktreeSettings,
2557 connection: ConnectionId,
2558 ) -> Result<RoomGuard<Vec<ConnectionId>>> {
2559 let project_id = ProjectId::from_proto(update.project_id);
2560 let room_id = self.room_id_for_project(project_id).await?;
2561 self.room_transaction(room_id, |tx| async move {
2562 // Ensure the update comes from the host.
2563 let project = project::Entity::find_by_id(project_id)
2564 .one(&*tx)
2565 .await?
2566 .ok_or_else(|| anyhow!("no such project"))?;
2567 if project.host_connection()? != connection {
2568 return Err(anyhow!("can't update a project hosted by someone else"))?;
2569 }
2570
2571 if let Some(content) = &update.content {
2572 worktree_settings_file::Entity::insert(worktree_settings_file::ActiveModel {
2573 project_id: ActiveValue::Set(project_id),
2574 worktree_id: ActiveValue::Set(update.worktree_id as i64),
2575 path: ActiveValue::Set(update.path.clone()),
2576 content: ActiveValue::Set(content.clone()),
2577 })
2578 .on_conflict(
2579 OnConflict::columns([
2580 worktree_settings_file::Column::ProjectId,
2581 worktree_settings_file::Column::WorktreeId,
2582 worktree_settings_file::Column::Path,
2583 ])
2584 .update_column(worktree_settings_file::Column::Content)
2585 .to_owned(),
2586 )
2587 .exec(&*tx)
2588 .await?;
2589 } else {
2590 worktree_settings_file::Entity::delete(worktree_settings_file::ActiveModel {
2591 project_id: ActiveValue::Set(project_id),
2592 worktree_id: ActiveValue::Set(update.worktree_id as i64),
2593 path: ActiveValue::Set(update.path.clone()),
2594 ..Default::default()
2595 })
2596 .exec(&*tx)
2597 .await?;
2598 }
2599
2600 let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
2601 Ok(connection_ids)
2602 })
2603 .await
2604 }
2605
2606 pub async fn join_project(
2607 &self,
2608 project_id: ProjectId,
2609 connection: ConnectionId,
2610 ) -> Result<RoomGuard<(Project, ReplicaId)>> {
2611 let room_id = self.room_id_for_project(project_id).await?;
2612 self.room_transaction(room_id, |tx| async move {
2613 let participant = room_participant::Entity::find()
2614 .filter(
2615 Condition::all()
2616 .add(
2617 room_participant::Column::AnsweringConnectionId
2618 .eq(connection.id as i32),
2619 )
2620 .add(
2621 room_participant::Column::AnsweringConnectionServerId
2622 .eq(connection.owner_id as i32),
2623 ),
2624 )
2625 .one(&*tx)
2626 .await?
2627 .ok_or_else(|| anyhow!("must join a room first"))?;
2628
2629 let project = project::Entity::find_by_id(project_id)
2630 .one(&*tx)
2631 .await?
2632 .ok_or_else(|| anyhow!("no such project"))?;
2633 if project.room_id != participant.room_id {
2634 return Err(anyhow!("no such project"))?;
2635 }
2636
2637 let mut collaborators = project
2638 .find_related(project_collaborator::Entity)
2639 .all(&*tx)
2640 .await?;
2641 let replica_ids = collaborators
2642 .iter()
2643 .map(|c| c.replica_id)
2644 .collect::<HashSet<_>>();
2645 let mut replica_id = ReplicaId(1);
2646 while replica_ids.contains(&replica_id) {
2647 replica_id.0 += 1;
2648 }
2649 let new_collaborator = project_collaborator::ActiveModel {
2650 project_id: ActiveValue::set(project_id),
2651 connection_id: ActiveValue::set(connection.id as i32),
2652 connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
2653 user_id: ActiveValue::set(participant.user_id),
2654 replica_id: ActiveValue::set(replica_id),
2655 is_host: ActiveValue::set(false),
2656 ..Default::default()
2657 }
2658 .insert(&*tx)
2659 .await?;
2660 collaborators.push(new_collaborator);
2661
2662 let db_worktrees = project.find_related(worktree::Entity).all(&*tx).await?;
2663 let mut worktrees = db_worktrees
2664 .into_iter()
2665 .map(|db_worktree| {
2666 (
2667 db_worktree.id as u64,
2668 Worktree {
2669 id: db_worktree.id as u64,
2670 abs_path: db_worktree.abs_path,
2671 root_name: db_worktree.root_name,
2672 visible: db_worktree.visible,
2673 entries: Default::default(),
2674 repository_entries: Default::default(),
2675 diagnostic_summaries: Default::default(),
2676 settings_files: Default::default(),
2677 scan_id: db_worktree.scan_id as u64,
2678 completed_scan_id: db_worktree.completed_scan_id as u64,
2679 },
2680 )
2681 })
2682 .collect::<BTreeMap<_, _>>();
2683
2684 // Populate worktree entries.
2685 {
2686 let mut db_entries = worktree_entry::Entity::find()
2687 .filter(
2688 Condition::all()
2689 .add(worktree_entry::Column::ProjectId.eq(project_id))
2690 .add(worktree_entry::Column::IsDeleted.eq(false)),
2691 )
2692 .stream(&*tx)
2693 .await?;
2694 while let Some(db_entry) = db_entries.next().await {
2695 let db_entry = db_entry?;
2696 if let Some(worktree) = worktrees.get_mut(&(db_entry.worktree_id as u64)) {
2697 worktree.entries.push(proto::Entry {
2698 id: db_entry.id as u64,
2699 is_dir: db_entry.is_dir,
2700 path: db_entry.path,
2701 inode: db_entry.inode as u64,
2702 mtime: Some(proto::Timestamp {
2703 seconds: db_entry.mtime_seconds as u64,
2704 nanos: db_entry.mtime_nanos as u32,
2705 }),
2706 is_symlink: db_entry.is_symlink,
2707 is_ignored: db_entry.is_ignored,
2708 git_status: db_entry.git_status.map(|status| status as i32),
2709 });
2710 }
2711 }
2712 }
2713
2714 // Populate repository entries.
2715 {
2716 let mut db_repository_entries = worktree_repository::Entity::find()
2717 .filter(
2718 Condition::all()
2719 .add(worktree_repository::Column::ProjectId.eq(project_id))
2720 .add(worktree_repository::Column::IsDeleted.eq(false)),
2721 )
2722 .stream(&*tx)
2723 .await?;
2724 while let Some(db_repository_entry) = db_repository_entries.next().await {
2725 let db_repository_entry = db_repository_entry?;
2726 if let Some(worktree) =
2727 worktrees.get_mut(&(db_repository_entry.worktree_id as u64))
2728 {
2729 worktree.repository_entries.insert(
2730 db_repository_entry.work_directory_id as u64,
2731 proto::RepositoryEntry {
2732 work_directory_id: db_repository_entry.work_directory_id as u64,
2733 branch: db_repository_entry.branch,
2734 },
2735 );
2736 }
2737 }
2738 }
2739
2740 // Populate worktree diagnostic summaries.
2741 {
2742 let mut db_summaries = worktree_diagnostic_summary::Entity::find()
2743 .filter(worktree_diagnostic_summary::Column::ProjectId.eq(project_id))
2744 .stream(&*tx)
2745 .await?;
2746 while let Some(db_summary) = db_summaries.next().await {
2747 let db_summary = db_summary?;
2748 if let Some(worktree) = worktrees.get_mut(&(db_summary.worktree_id as u64)) {
2749 worktree
2750 .diagnostic_summaries
2751 .push(proto::DiagnosticSummary {
2752 path: db_summary.path,
2753 language_server_id: db_summary.language_server_id as u64,
2754 error_count: db_summary.error_count as u32,
2755 warning_count: db_summary.warning_count as u32,
2756 });
2757 }
2758 }
2759 }
2760
2761 // Populate worktree settings files
2762 {
2763 let mut db_settings_files = worktree_settings_file::Entity::find()
2764 .filter(worktree_settings_file::Column::ProjectId.eq(project_id))
2765 .stream(&*tx)
2766 .await?;
2767 while let Some(db_settings_file) = db_settings_files.next().await {
2768 let db_settings_file = db_settings_file?;
2769 if let Some(worktree) =
2770 worktrees.get_mut(&(db_settings_file.worktree_id as u64))
2771 {
2772 worktree.settings_files.push(WorktreeSettingsFile {
2773 path: db_settings_file.path,
2774 content: db_settings_file.content,
2775 });
2776 }
2777 }
2778 }
2779
2780 // Populate language servers.
2781 let language_servers = project
2782 .find_related(language_server::Entity)
2783 .all(&*tx)
2784 .await?;
2785
2786 let project = Project {
2787 collaborators: collaborators
2788 .into_iter()
2789 .map(|collaborator| ProjectCollaborator {
2790 connection_id: collaborator.connection(),
2791 user_id: collaborator.user_id,
2792 replica_id: collaborator.replica_id,
2793 is_host: collaborator.is_host,
2794 })
2795 .collect(),
2796 worktrees,
2797 language_servers: language_servers
2798 .into_iter()
2799 .map(|language_server| proto::LanguageServer {
2800 id: language_server.id as u64,
2801 name: language_server.name,
2802 })
2803 .collect(),
2804 };
2805 Ok((project, replica_id as ReplicaId))
2806 })
2807 .await
2808 }
2809
2810 pub async fn leave_project(
2811 &self,
2812 project_id: ProjectId,
2813 connection: ConnectionId,
2814 ) -> Result<RoomGuard<(proto::Room, LeftProject)>> {
2815 let room_id = self.room_id_for_project(project_id).await?;
2816 self.room_transaction(room_id, |tx| async move {
2817 let result = project_collaborator::Entity::delete_many()
2818 .filter(
2819 Condition::all()
2820 .add(project_collaborator::Column::ProjectId.eq(project_id))
2821 .add(project_collaborator::Column::ConnectionId.eq(connection.id as i32))
2822 .add(
2823 project_collaborator::Column::ConnectionServerId
2824 .eq(connection.owner_id as i32),
2825 ),
2826 )
2827 .exec(&*tx)
2828 .await?;
2829 if result.rows_affected == 0 {
2830 Err(anyhow!("not a collaborator on this project"))?;
2831 }
2832
2833 let project = project::Entity::find_by_id(project_id)
2834 .one(&*tx)
2835 .await?
2836 .ok_or_else(|| anyhow!("no such project"))?;
2837 let collaborators = project
2838 .find_related(project_collaborator::Entity)
2839 .all(&*tx)
2840 .await?;
2841 let connection_ids = collaborators
2842 .into_iter()
2843 .map(|collaborator| collaborator.connection())
2844 .collect();
2845
2846 follower::Entity::delete_many()
2847 .filter(
2848 Condition::any()
2849 .add(
2850 Condition::all()
2851 .add(follower::Column::ProjectId.eq(project_id))
2852 .add(
2853 follower::Column::LeaderConnectionServerId
2854 .eq(connection.owner_id),
2855 )
2856 .add(follower::Column::LeaderConnectionId.eq(connection.id)),
2857 )
2858 .add(
2859 Condition::all()
2860 .add(follower::Column::ProjectId.eq(project_id))
2861 .add(
2862 follower::Column::FollowerConnectionServerId
2863 .eq(connection.owner_id),
2864 )
2865 .add(follower::Column::FollowerConnectionId.eq(connection.id)),
2866 ),
2867 )
2868 .exec(&*tx)
2869 .await?;
2870
2871 let room = self.get_room(project.room_id, &tx).await?;
2872 let left_project = LeftProject {
2873 id: project_id,
2874 host_user_id: project.host_user_id,
2875 host_connection_id: project.host_connection()?,
2876 connection_ids,
2877 };
2878 Ok((room, left_project))
2879 })
2880 .await
2881 }
2882
2883 pub async fn project_collaborators(
2884 &self,
2885 project_id: ProjectId,
2886 connection_id: ConnectionId,
2887 ) -> Result<RoomGuard<Vec<ProjectCollaborator>>> {
2888 let room_id = self.room_id_for_project(project_id).await?;
2889 self.room_transaction(room_id, |tx| async move {
2890 let collaborators = project_collaborator::Entity::find()
2891 .filter(project_collaborator::Column::ProjectId.eq(project_id))
2892 .all(&*tx)
2893 .await?
2894 .into_iter()
2895 .map(|collaborator| ProjectCollaborator {
2896 connection_id: collaborator.connection(),
2897 user_id: collaborator.user_id,
2898 replica_id: collaborator.replica_id,
2899 is_host: collaborator.is_host,
2900 })
2901 .collect::<Vec<_>>();
2902
2903 if collaborators
2904 .iter()
2905 .any(|collaborator| collaborator.connection_id == connection_id)
2906 {
2907 Ok(collaborators)
2908 } else {
2909 Err(anyhow!("no such project"))?
2910 }
2911 })
2912 .await
2913 }
2914
2915 pub async fn project_connection_ids(
2916 &self,
2917 project_id: ProjectId,
2918 connection_id: ConnectionId,
2919 ) -> Result<RoomGuard<HashSet<ConnectionId>>> {
2920 let room_id = self.room_id_for_project(project_id).await?;
2921 self.room_transaction(room_id, |tx| async move {
2922 let mut collaborators = project_collaborator::Entity::find()
2923 .filter(project_collaborator::Column::ProjectId.eq(project_id))
2924 .stream(&*tx)
2925 .await?;
2926
2927 let mut connection_ids = HashSet::default();
2928 while let Some(collaborator) = collaborators.next().await {
2929 let collaborator = collaborator?;
2930 connection_ids.insert(collaborator.connection());
2931 }
2932
2933 if connection_ids.contains(&connection_id) {
2934 Ok(connection_ids)
2935 } else {
2936 Err(anyhow!("no such project"))?
2937 }
2938 })
2939 .await
2940 }
2941
2942 async fn project_guest_connection_ids(
2943 &self,
2944 project_id: ProjectId,
2945 tx: &DatabaseTransaction,
2946 ) -> Result<Vec<ConnectionId>> {
2947 let mut collaborators = project_collaborator::Entity::find()
2948 .filter(
2949 project_collaborator::Column::ProjectId
2950 .eq(project_id)
2951 .and(project_collaborator::Column::IsHost.eq(false)),
2952 )
2953 .stream(tx)
2954 .await?;
2955
2956 let mut guest_connection_ids = Vec::new();
2957 while let Some(collaborator) = collaborators.next().await {
2958 let collaborator = collaborator?;
2959 guest_connection_ids.push(collaborator.connection());
2960 }
2961 Ok(guest_connection_ids)
2962 }
2963
2964 async fn room_id_for_project(&self, project_id: ProjectId) -> Result<RoomId> {
2965 self.transaction(|tx| async move {
2966 let project = project::Entity::find_by_id(project_id)
2967 .one(&*tx)
2968 .await?
2969 .ok_or_else(|| anyhow!("project {} not found", project_id))?;
2970 Ok(project.room_id)
2971 })
2972 .await
2973 }
2974
2975 // access tokens
2976
2977 pub async fn create_access_token(
2978 &self,
2979 user_id: UserId,
2980 access_token_hash: &str,
2981 max_access_token_count: usize,
2982 ) -> Result<AccessTokenId> {
2983 self.transaction(|tx| async {
2984 let tx = tx;
2985
2986 let token = access_token::ActiveModel {
2987 user_id: ActiveValue::set(user_id),
2988 hash: ActiveValue::set(access_token_hash.into()),
2989 ..Default::default()
2990 }
2991 .insert(&*tx)
2992 .await?;
2993
2994 access_token::Entity::delete_many()
2995 .filter(
2996 access_token::Column::Id.in_subquery(
2997 Query::select()
2998 .column(access_token::Column::Id)
2999 .from(access_token::Entity)
3000 .and_where(access_token::Column::UserId.eq(user_id))
3001 .order_by(access_token::Column::Id, sea_orm::Order::Desc)
3002 .limit(10000)
3003 .offset(max_access_token_count as u64)
3004 .to_owned(),
3005 ),
3006 )
3007 .exec(&*tx)
3008 .await?;
3009 Ok(token.id)
3010 })
3011 .await
3012 }
3013
3014 pub async fn get_access_token(
3015 &self,
3016 access_token_id: AccessTokenId,
3017 ) -> Result<access_token::Model> {
3018 self.transaction(|tx| async move {
3019 Ok(access_token::Entity::find_by_id(access_token_id)
3020 .one(&*tx)
3021 .await?
3022 .ok_or_else(|| anyhow!("no such access token"))?)
3023 })
3024 .await
3025 }
3026
3027 async fn transaction<F, Fut, T>(&self, f: F) -> Result<T>
3028 where
3029 F: Send + Fn(TransactionHandle) -> Fut,
3030 Fut: Send + Future<Output = Result<T>>,
3031 {
3032 let body = async {
3033 let mut i = 0;
3034 loop {
3035 let (tx, result) = self.with_transaction(&f).await?;
3036 match result {
3037 Ok(result) => match tx.commit().await.map_err(Into::into) {
3038 Ok(()) => return Ok(result),
3039 Err(error) => {
3040 if !self.retry_on_serialization_error(&error, i).await {
3041 return Err(error);
3042 }
3043 }
3044 },
3045 Err(error) => {
3046 tx.rollback().await?;
3047 if !self.retry_on_serialization_error(&error, i).await {
3048 return Err(error);
3049 }
3050 }
3051 }
3052 i += 1;
3053 }
3054 };
3055
3056 self.run(body).await
3057 }
3058
3059 async fn optional_room_transaction<F, Fut, T>(&self, f: F) -> Result<Option<RoomGuard<T>>>
3060 where
3061 F: Send + Fn(TransactionHandle) -> Fut,
3062 Fut: Send + Future<Output = Result<Option<(RoomId, T)>>>,
3063 {
3064 let body = async {
3065 let mut i = 0;
3066 loop {
3067 let (tx, result) = self.with_transaction(&f).await?;
3068 match result {
3069 Ok(Some((room_id, data))) => {
3070 let lock = self.rooms.entry(room_id).or_default().clone();
3071 let _guard = lock.lock_owned().await;
3072 match tx.commit().await.map_err(Into::into) {
3073 Ok(()) => {
3074 return Ok(Some(RoomGuard {
3075 data,
3076 _guard,
3077 _not_send: PhantomData,
3078 }));
3079 }
3080 Err(error) => {
3081 if !self.retry_on_serialization_error(&error, i).await {
3082 return Err(error);
3083 }
3084 }
3085 }
3086 }
3087 Ok(None) => match tx.commit().await.map_err(Into::into) {
3088 Ok(()) => return Ok(None),
3089 Err(error) => {
3090 if !self.retry_on_serialization_error(&error, i).await {
3091 return Err(error);
3092 }
3093 }
3094 },
3095 Err(error) => {
3096 tx.rollback().await?;
3097 if !self.retry_on_serialization_error(&error, i).await {
3098 return Err(error);
3099 }
3100 }
3101 }
3102 i += 1;
3103 }
3104 };
3105
3106 self.run(body).await
3107 }
3108
3109 async fn room_transaction<F, Fut, T>(&self, room_id: RoomId, f: F) -> Result<RoomGuard<T>>
3110 where
3111 F: Send + Fn(TransactionHandle) -> Fut,
3112 Fut: Send + Future<Output = Result<T>>,
3113 {
3114 let body = async {
3115 let mut i = 0;
3116 loop {
3117 let lock = self.rooms.entry(room_id).or_default().clone();
3118 let _guard = lock.lock_owned().await;
3119 let (tx, result) = self.with_transaction(&f).await?;
3120 match result {
3121 Ok(data) => match tx.commit().await.map_err(Into::into) {
3122 Ok(()) => {
3123 return Ok(RoomGuard {
3124 data,
3125 _guard,
3126 _not_send: PhantomData,
3127 });
3128 }
3129 Err(error) => {
3130 if !self.retry_on_serialization_error(&error, i).await {
3131 return Err(error);
3132 }
3133 }
3134 },
3135 Err(error) => {
3136 tx.rollback().await?;
3137 if !self.retry_on_serialization_error(&error, i).await {
3138 return Err(error);
3139 }
3140 }
3141 }
3142 i += 1;
3143 }
3144 };
3145
3146 self.run(body).await
3147 }
3148
3149 async fn with_transaction<F, Fut, T>(&self, f: &F) -> Result<(DatabaseTransaction, Result<T>)>
3150 where
3151 F: Send + Fn(TransactionHandle) -> Fut,
3152 Fut: Send + Future<Output = Result<T>>,
3153 {
3154 let tx = self
3155 .pool
3156 .begin_with_config(Some(IsolationLevel::Serializable), None)
3157 .await?;
3158
3159 let mut tx = Arc::new(Some(tx));
3160 let result = f(TransactionHandle(tx.clone())).await;
3161 let Some(tx) = Arc::get_mut(&mut tx).and_then(|tx| tx.take()) else {
3162 return Err(anyhow!("couldn't complete transaction because it's still in use"))?;
3163 };
3164
3165 Ok((tx, result))
3166 }
3167
3168 async fn run<F, T>(&self, future: F) -> Result<T>
3169 where
3170 F: Future<Output = Result<T>>,
3171 {
3172 #[cfg(test)]
3173 {
3174 if let Executor::Deterministic(executor) = &self.executor {
3175 executor.simulate_random_delay().await;
3176 }
3177
3178 self.runtime.as_ref().unwrap().block_on(future)
3179 }
3180
3181 #[cfg(not(test))]
3182 {
3183 future.await
3184 }
3185 }
3186
3187 async fn retry_on_serialization_error(&self, error: &Error, prev_attempt_count: u32) -> bool {
3188 // If the error is due to a failure to serialize concurrent transactions, then retry
3189 // this transaction after a delay. With each subsequent retry, double the delay duration.
3190 // Also vary the delay randomly in order to ensure different database connections retry
3191 // at different times.
3192 if is_serialization_error(error) {
3193 let base_delay = 4_u64 << prev_attempt_count.min(16);
3194 let randomized_delay = base_delay as f32 * self.rng.lock().await.gen_range(0.5..=2.0);
3195 log::info!(
3196 "retrying transaction after serialization error. delay: {} ms.",
3197 randomized_delay
3198 );
3199 self.executor
3200 .sleep(Duration::from_millis(randomized_delay as u64))
3201 .await;
3202 true
3203 } else {
3204 false
3205 }
3206 }
3207}
3208
3209fn is_serialization_error(error: &Error) -> bool {
3210 const SERIALIZATION_FAILURE_CODE: &'static str = "40001";
3211 match error {
3212 Error::Database(
3213 DbErr::Exec(sea_orm::RuntimeErr::SqlxError(error))
3214 | DbErr::Query(sea_orm::RuntimeErr::SqlxError(error)),
3215 ) if error
3216 .as_database_error()
3217 .and_then(|error| error.code())
3218 .as_deref()
3219 == Some(SERIALIZATION_FAILURE_CODE) =>
3220 {
3221 true
3222 }
3223 _ => false,
3224 }
3225}
3226
3227struct TransactionHandle(Arc<Option<DatabaseTransaction>>);
3228
3229impl Deref for TransactionHandle {
3230 type Target = DatabaseTransaction;
3231
3232 fn deref(&self) -> &Self::Target {
3233 self.0.as_ref().as_ref().unwrap()
3234 }
3235}
3236
3237pub struct RoomGuard<T> {
3238 data: T,
3239 _guard: OwnedMutexGuard<()>,
3240 _not_send: PhantomData<Rc<()>>,
3241}
3242
3243impl<T> Deref for RoomGuard<T> {
3244 type Target = T;
3245
3246 fn deref(&self) -> &T {
3247 &self.data
3248 }
3249}
3250
3251impl<T> DerefMut for RoomGuard<T> {
3252 fn deref_mut(&mut self) -> &mut T {
3253 &mut self.data
3254 }
3255}
3256
3257#[derive(Debug, Serialize, Deserialize)]
3258pub struct NewUserParams {
3259 pub github_login: String,
3260 pub github_user_id: i32,
3261 pub invite_count: i32,
3262}
3263
3264#[derive(Debug)]
3265pub struct NewUserResult {
3266 pub user_id: UserId,
3267 pub metrics_id: String,
3268 pub inviting_user_id: Option<UserId>,
3269 pub signup_device_id: Option<String>,
3270}
3271
3272fn random_invite_code() -> String {
3273 nanoid::nanoid!(16)
3274}
3275
3276fn random_email_confirmation_code() -> String {
3277 nanoid::nanoid!(64)
3278}
3279
3280macro_rules! id_type {
3281 ($name:ident) => {
3282 #[derive(
3283 Clone,
3284 Copy,
3285 Debug,
3286 Default,
3287 PartialEq,
3288 Eq,
3289 PartialOrd,
3290 Ord,
3291 Hash,
3292 Serialize,
3293 Deserialize,
3294 )]
3295 #[serde(transparent)]
3296 pub struct $name(pub i32);
3297
3298 impl $name {
3299 #[allow(unused)]
3300 pub const MAX: Self = Self(i32::MAX);
3301
3302 #[allow(unused)]
3303 pub fn from_proto(value: u64) -> Self {
3304 Self(value as i32)
3305 }
3306
3307 #[allow(unused)]
3308 pub fn to_proto(self) -> u64 {
3309 self.0 as u64
3310 }
3311 }
3312
3313 impl std::fmt::Display for $name {
3314 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
3315 self.0.fmt(f)
3316 }
3317 }
3318
3319 impl From<$name> for sea_query::Value {
3320 fn from(value: $name) -> Self {
3321 sea_query::Value::Int(Some(value.0))
3322 }
3323 }
3324
3325 impl sea_orm::TryGetable for $name {
3326 fn try_get(
3327 res: &sea_orm::QueryResult,
3328 pre: &str,
3329 col: &str,
3330 ) -> Result<Self, sea_orm::TryGetError> {
3331 Ok(Self(i32::try_get(res, pre, col)?))
3332 }
3333 }
3334
3335 impl sea_query::ValueType for $name {
3336 fn try_from(v: Value) -> Result<Self, sea_query::ValueTypeErr> {
3337 match v {
3338 Value::TinyInt(Some(int)) => {
3339 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3340 }
3341 Value::SmallInt(Some(int)) => {
3342 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3343 }
3344 Value::Int(Some(int)) => {
3345 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3346 }
3347 Value::BigInt(Some(int)) => {
3348 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3349 }
3350 Value::TinyUnsigned(Some(int)) => {
3351 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3352 }
3353 Value::SmallUnsigned(Some(int)) => {
3354 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3355 }
3356 Value::Unsigned(Some(int)) => {
3357 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3358 }
3359 Value::BigUnsigned(Some(int)) => {
3360 Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
3361 }
3362 _ => Err(sea_query::ValueTypeErr),
3363 }
3364 }
3365
3366 fn type_name() -> String {
3367 stringify!($name).into()
3368 }
3369
3370 fn array_type() -> sea_query::ArrayType {
3371 sea_query::ArrayType::Int
3372 }
3373
3374 fn column_type() -> sea_query::ColumnType {
3375 sea_query::ColumnType::Integer(None)
3376 }
3377 }
3378
3379 impl sea_orm::TryFromU64 for $name {
3380 fn try_from_u64(n: u64) -> Result<Self, DbErr> {
3381 Ok(Self(n.try_into().map_err(|_| {
3382 DbErr::ConvertFromU64(concat!(
3383 "error converting ",
3384 stringify!($name),
3385 " to u64"
3386 ))
3387 })?))
3388 }
3389 }
3390
3391 impl sea_query::Nullable for $name {
3392 fn null() -> Value {
3393 Value::Int(None)
3394 }
3395 }
3396 };
3397}
3398
3399id_type!(AccessTokenId);
3400id_type!(ContactId);
3401id_type!(FollowerId);
3402id_type!(RoomId);
3403id_type!(RoomParticipantId);
3404id_type!(ProjectId);
3405id_type!(ProjectCollaboratorId);
3406id_type!(ReplicaId);
3407id_type!(ServerId);
3408id_type!(SignupId);
3409id_type!(UserId);
3410
3411pub struct RejoinedRoom {
3412 pub room: proto::Room,
3413 pub rejoined_projects: Vec<RejoinedProject>,
3414 pub reshared_projects: Vec<ResharedProject>,
3415}
3416
3417pub struct ResharedProject {
3418 pub id: ProjectId,
3419 pub old_connection_id: ConnectionId,
3420 pub collaborators: Vec<ProjectCollaborator>,
3421 pub worktrees: Vec<proto::WorktreeMetadata>,
3422}
3423
3424pub struct RejoinedProject {
3425 pub id: ProjectId,
3426 pub old_connection_id: ConnectionId,
3427 pub collaborators: Vec<ProjectCollaborator>,
3428 pub worktrees: Vec<RejoinedWorktree>,
3429 pub language_servers: Vec<proto::LanguageServer>,
3430}
3431
3432#[derive(Debug)]
3433pub struct RejoinedWorktree {
3434 pub id: u64,
3435 pub abs_path: String,
3436 pub root_name: String,
3437 pub visible: bool,
3438 pub updated_entries: Vec<proto::Entry>,
3439 pub removed_entries: Vec<u64>,
3440 pub updated_repositories: Vec<proto::RepositoryEntry>,
3441 pub removed_repositories: Vec<u64>,
3442 pub diagnostic_summaries: Vec<proto::DiagnosticSummary>,
3443 pub settings_files: Vec<WorktreeSettingsFile>,
3444 pub scan_id: u64,
3445 pub completed_scan_id: u64,
3446}
3447
3448pub struct LeftRoom {
3449 pub room: proto::Room,
3450 pub left_projects: HashMap<ProjectId, LeftProject>,
3451 pub canceled_calls_to_user_ids: Vec<UserId>,
3452}
3453
3454pub struct RefreshedRoom {
3455 pub room: proto::Room,
3456 pub stale_participant_user_ids: Vec<UserId>,
3457 pub canceled_calls_to_user_ids: Vec<UserId>,
3458}
3459
3460pub struct Project {
3461 pub collaborators: Vec<ProjectCollaborator>,
3462 pub worktrees: BTreeMap<u64, Worktree>,
3463 pub language_servers: Vec<proto::LanguageServer>,
3464}
3465
3466pub struct ProjectCollaborator {
3467 pub connection_id: ConnectionId,
3468 pub user_id: UserId,
3469 pub replica_id: ReplicaId,
3470 pub is_host: bool,
3471}
3472
3473impl ProjectCollaborator {
3474 pub fn to_proto(&self) -> proto::Collaborator {
3475 proto::Collaborator {
3476 peer_id: Some(self.connection_id.into()),
3477 replica_id: self.replica_id.0 as u32,
3478 user_id: self.user_id.to_proto(),
3479 }
3480 }
3481}
3482
3483#[derive(Debug)]
3484pub struct LeftProject {
3485 pub id: ProjectId,
3486 pub host_user_id: UserId,
3487 pub host_connection_id: ConnectionId,
3488 pub connection_ids: Vec<ConnectionId>,
3489}
3490
3491pub struct Worktree {
3492 pub id: u64,
3493 pub abs_path: String,
3494 pub root_name: String,
3495 pub visible: bool,
3496 pub entries: Vec<proto::Entry>,
3497 pub repository_entries: BTreeMap<u64, proto::RepositoryEntry>,
3498 pub diagnostic_summaries: Vec<proto::DiagnosticSummary>,
3499 pub settings_files: Vec<WorktreeSettingsFile>,
3500 pub scan_id: u64,
3501 pub completed_scan_id: u64,
3502}
3503
3504#[derive(Debug)]
3505pub struct WorktreeSettingsFile {
3506 pub path: String,
3507 pub content: String,
3508}
3509
3510#[cfg(test)]
3511pub use test::*;
3512
3513#[cfg(test)]
3514mod test {
3515 use super::*;
3516 use gpui::executor::Background;
3517 use lazy_static::lazy_static;
3518 use parking_lot::Mutex;
3519 use sea_orm::ConnectionTrait;
3520 use sqlx::migrate::MigrateDatabase;
3521 use std::sync::Arc;
3522
3523 pub struct TestDb {
3524 pub db: Option<Arc<Database>>,
3525 pub connection: Option<sqlx::AnyConnection>,
3526 }
3527
3528 impl TestDb {
3529 pub fn sqlite(background: Arc<Background>) -> Self {
3530 let url = format!("sqlite::memory:");
3531 let runtime = tokio::runtime::Builder::new_current_thread()
3532 .enable_io()
3533 .enable_time()
3534 .build()
3535 .unwrap();
3536
3537 let mut db = runtime.block_on(async {
3538 let mut options = ConnectOptions::new(url);
3539 options.max_connections(5);
3540 let db = Database::new(options, Executor::Deterministic(background))
3541 .await
3542 .unwrap();
3543 let sql = include_str!(concat!(
3544 env!("CARGO_MANIFEST_DIR"),
3545 "/migrations.sqlite/20221109000000_test_schema.sql"
3546 ));
3547 db.pool
3548 .execute(sea_orm::Statement::from_string(
3549 db.pool.get_database_backend(),
3550 sql.into(),
3551 ))
3552 .await
3553 .unwrap();
3554 db
3555 });
3556
3557 db.runtime = Some(runtime);
3558
3559 Self {
3560 db: Some(Arc::new(db)),
3561 connection: None,
3562 }
3563 }
3564
3565 pub fn postgres(background: Arc<Background>) -> Self {
3566 lazy_static! {
3567 static ref LOCK: Mutex<()> = Mutex::new(());
3568 }
3569
3570 let _guard = LOCK.lock();
3571 let mut rng = StdRng::from_entropy();
3572 let url = format!(
3573 "postgres://postgres@localhost/zed-test-{}",
3574 rng.gen::<u128>()
3575 );
3576 let runtime = tokio::runtime::Builder::new_current_thread()
3577 .enable_io()
3578 .enable_time()
3579 .build()
3580 .unwrap();
3581
3582 let mut db = runtime.block_on(async {
3583 sqlx::Postgres::create_database(&url)
3584 .await
3585 .expect("failed to create test db");
3586 let mut options = ConnectOptions::new(url);
3587 options
3588 .max_connections(5)
3589 .idle_timeout(Duration::from_secs(0));
3590 let db = Database::new(options, Executor::Deterministic(background))
3591 .await
3592 .unwrap();
3593 let migrations_path = concat!(env!("CARGO_MANIFEST_DIR"), "/migrations");
3594 db.migrate(Path::new(migrations_path), false).await.unwrap();
3595 db
3596 });
3597
3598 db.runtime = Some(runtime);
3599
3600 Self {
3601 db: Some(Arc::new(db)),
3602 connection: None,
3603 }
3604 }
3605
3606 pub fn db(&self) -> &Arc<Database> {
3607 self.db.as_ref().unwrap()
3608 }
3609 }
3610
3611 impl Drop for TestDb {
3612 fn drop(&mut self) {
3613 let db = self.db.take().unwrap();
3614 if let sea_orm::DatabaseBackend::Postgres = db.pool.get_database_backend() {
3615 db.runtime.as_ref().unwrap().block_on(async {
3616 use util::ResultExt;
3617 let query = "
3618 SELECT pg_terminate_backend(pg_stat_activity.pid)
3619 FROM pg_stat_activity
3620 WHERE
3621 pg_stat_activity.datname = current_database() AND
3622 pid <> pg_backend_pid();
3623 ";
3624 db.pool
3625 .execute(sea_orm::Statement::from_string(
3626 db.pool.get_database_backend(),
3627 query.into(),
3628 ))
3629 .await
3630 .log_err();
3631 sqlx::Postgres::drop_database(db.options.get_url())
3632 .await
3633 .log_err();
3634 })
3635 }
3636 }
3637 }
3638}