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