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