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