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