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