1use super::*;
2
3impl Database {
4 pub async fn create_server(&self, environment: &str) -> Result<ServerId> {
5 self.transaction(|tx| async move {
6 let server = server::ActiveModel {
7 environment: ActiveValue::set(environment.into()),
8 ..Default::default()
9 }
10 .insert(&*tx)
11 .await?;
12 Ok(server.id)
13 })
14 .await
15 }
16
17 pub async fn stale_room_ids(
18 &self,
19 environment: &str,
20 new_server_id: ServerId,
21 ) -> Result<Vec<RoomId>> {
22 self.transaction(|tx| async move {
23 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
24 enum QueryAs {
25 RoomId,
26 }
27
28 let stale_server_epochs = self
29 .stale_server_ids(environment, new_server_id, &tx)
30 .await?;
31 Ok(room_participant::Entity::find()
32 .select_only()
33 .column(room_participant::Column::RoomId)
34 .distinct()
35 .filter(
36 room_participant::Column::AnsweringConnectionServerId
37 .is_in(stale_server_epochs),
38 )
39 .into_values::<_, QueryAs>()
40 .all(&*tx)
41 .await?)
42 })
43 .await
44 }
45
46 pub async fn delete_stale_servers(
47 &self,
48 environment: &str,
49 new_server_id: ServerId,
50 ) -> Result<()> {
51 self.transaction(|tx| async move {
52 server::Entity::delete_many()
53 .filter(
54 Condition::all()
55 .add(server::Column::Environment.eq(environment))
56 .add(server::Column::Id.ne(new_server_id)),
57 )
58 .exec(&*tx)
59 .await?;
60 Ok(())
61 })
62 .await
63 }
64
65 async fn stale_server_ids(
66 &self,
67 environment: &str,
68 new_server_id: ServerId,
69 tx: &DatabaseTransaction,
70 ) -> Result<Vec<ServerId>> {
71 let stale_servers = server::Entity::find()
72 .filter(
73 Condition::all()
74 .add(server::Column::Environment.eq(environment))
75 .add(server::Column::Id.ne(new_server_id)),
76 )
77 .all(&*tx)
78 .await?;
79 Ok(stale_servers.into_iter().map(|server| server.id).collect())
80 }
81}