1use anyhow::Context;
2use anyhow::Result;
3use async_trait::async_trait;
4use serde::Serialize;
5pub use sqlx::postgres::PgPoolOptions as DbOptions;
6use sqlx::{types::Uuid, FromRow};
7use time::OffsetDateTime;
8
9#[async_trait]
10pub trait Db: Send + Sync {
11 async fn create_user(&self, github_login: &str, admin: bool) -> Result<UserId>;
12 async fn get_all_users(&self) -> Result<Vec<User>>;
13 async fn get_user_by_id(&self, id: UserId) -> Result<Option<User>>;
14 async fn get_users_by_ids(&self, ids: Vec<UserId>) -> Result<Vec<User>>;
15 async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>>;
16 async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()>;
17 async fn destroy_user(&self, id: UserId) -> Result<()>;
18 async fn create_access_token_hash(
19 &self,
20 user_id: UserId,
21 access_token_hash: &str,
22 max_access_token_count: usize,
23 ) -> Result<()>;
24 async fn get_access_token_hashes(&self, user_id: UserId) -> Result<Vec<String>>;
25 #[cfg(any(test, feature = "seed-support"))]
26 async fn find_org_by_slug(&self, slug: &str) -> Result<Option<Org>>;
27 #[cfg(any(test, feature = "seed-support"))]
28 async fn create_org(&self, name: &str, slug: &str) -> Result<OrgId>;
29 #[cfg(any(test, feature = "seed-support"))]
30 async fn add_org_member(&self, org_id: OrgId, user_id: UserId, is_admin: bool) -> Result<()>;
31 #[cfg(any(test, feature = "seed-support"))]
32 async fn create_org_channel(&self, org_id: OrgId, name: &str) -> Result<ChannelId>;
33 #[cfg(any(test, feature = "seed-support"))]
34 async fn get_org_channels(&self, org_id: OrgId) -> Result<Vec<Channel>>;
35 async fn get_accessible_channels(&self, user_id: UserId) -> Result<Vec<Channel>>;
36 async fn can_user_access_channel(&self, user_id: UserId, channel_id: ChannelId)
37 -> Result<bool>;
38 #[cfg(any(test, feature = "seed-support"))]
39 async fn add_channel_member(
40 &self,
41 channel_id: ChannelId,
42 user_id: UserId,
43 is_admin: bool,
44 ) -> Result<()>;
45 async fn create_channel_message(
46 &self,
47 channel_id: ChannelId,
48 sender_id: UserId,
49 body: &str,
50 timestamp: OffsetDateTime,
51 nonce: u128,
52 ) -> Result<MessageId>;
53 async fn get_channel_messages(
54 &self,
55 channel_id: ChannelId,
56 count: usize,
57 before_id: Option<MessageId>,
58 ) -> Result<Vec<ChannelMessage>>;
59 #[cfg(test)]
60 async fn teardown(&self, url: &str);
61}
62
63pub struct PostgresDb {
64 pool: sqlx::PgPool,
65}
66
67impl PostgresDb {
68 pub async fn new(url: &str, max_connections: u32) -> Result<Self> {
69 let pool = DbOptions::new()
70 .max_connections(max_connections)
71 .connect(&url)
72 .await
73 .context("failed to connect to postgres database")?;
74 Ok(Self { pool })
75 }
76}
77
78#[async_trait]
79impl Db for PostgresDb {
80 // users
81
82 async fn create_user(&self, github_login: &str, admin: bool) -> Result<UserId> {
83 let query = "
84 INSERT INTO users (github_login, admin)
85 VALUES ($1, $2)
86 ON CONFLICT (github_login) DO UPDATE SET github_login = excluded.github_login
87 RETURNING id
88 ";
89 Ok(sqlx::query_scalar(query)
90 .bind(github_login)
91 .bind(admin)
92 .fetch_one(&self.pool)
93 .await
94 .map(UserId)?)
95 }
96
97 async fn get_all_users(&self) -> Result<Vec<User>> {
98 let query = "SELECT * FROM users ORDER BY github_login ASC";
99 Ok(sqlx::query_as(query).fetch_all(&self.pool).await?)
100 }
101
102 async fn get_user_by_id(&self, id: UserId) -> Result<Option<User>> {
103 let users = self.get_users_by_ids(vec![id]).await?;
104 Ok(users.into_iter().next())
105 }
106
107 async fn get_users_by_ids(&self, ids: Vec<UserId>) -> Result<Vec<User>> {
108 let ids = ids.into_iter().map(|id| id.0).collect::<Vec<_>>();
109 let query = "
110 SELECT users.*
111 FROM users
112 WHERE users.id = ANY ($1)
113 ";
114 Ok(sqlx::query_as(query)
115 .bind(&ids)
116 .fetch_all(&self.pool)
117 .await?)
118 }
119
120 async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>> {
121 let query = "SELECT * FROM users WHERE github_login = $1 LIMIT 1";
122 Ok(sqlx::query_as(query)
123 .bind(github_login)
124 .fetch_optional(&self.pool)
125 .await?)
126 }
127
128 async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()> {
129 let query = "UPDATE users SET admin = $1 WHERE id = $2";
130 Ok(sqlx::query(query)
131 .bind(is_admin)
132 .bind(id.0)
133 .execute(&self.pool)
134 .await
135 .map(drop)?)
136 }
137
138 async fn destroy_user(&self, id: UserId) -> Result<()> {
139 let query = "DELETE FROM access_tokens WHERE user_id = $1;";
140 sqlx::query(query)
141 .bind(id.0)
142 .execute(&self.pool)
143 .await
144 .map(drop)?;
145 let query = "DELETE FROM users WHERE id = $1;";
146 Ok(sqlx::query(query)
147 .bind(id.0)
148 .execute(&self.pool)
149 .await
150 .map(drop)?)
151 }
152
153 // access tokens
154
155 async fn create_access_token_hash(
156 &self,
157 user_id: UserId,
158 access_token_hash: &str,
159 max_access_token_count: usize,
160 ) -> Result<()> {
161 let insert_query = "
162 INSERT INTO access_tokens (user_id, hash)
163 VALUES ($1, $2);
164 ";
165 let cleanup_query = "
166 DELETE FROM access_tokens
167 WHERE id IN (
168 SELECT id from access_tokens
169 WHERE user_id = $1
170 ORDER BY id DESC
171 OFFSET $3
172 )
173 ";
174
175 let mut tx = self.pool.begin().await?;
176 sqlx::query(insert_query)
177 .bind(user_id.0)
178 .bind(access_token_hash)
179 .execute(&mut tx)
180 .await?;
181 sqlx::query(cleanup_query)
182 .bind(user_id.0)
183 .bind(access_token_hash)
184 .bind(max_access_token_count as u32)
185 .execute(&mut tx)
186 .await?;
187 Ok(tx.commit().await?)
188 }
189
190 async fn get_access_token_hashes(&self, user_id: UserId) -> Result<Vec<String>> {
191 let query = "
192 SELECT hash
193 FROM access_tokens
194 WHERE user_id = $1
195 ORDER BY id DESC
196 ";
197 Ok(sqlx::query_scalar(query)
198 .bind(user_id.0)
199 .fetch_all(&self.pool)
200 .await?)
201 }
202
203 // orgs
204
205 #[allow(unused)] // Help rust-analyzer
206 #[cfg(any(test, feature = "seed-support"))]
207 async fn find_org_by_slug(&self, slug: &str) -> Result<Option<Org>> {
208 let query = "
209 SELECT *
210 FROM orgs
211 WHERE slug = $1
212 ";
213 Ok(sqlx::query_as(query)
214 .bind(slug)
215 .fetch_optional(&self.pool)
216 .await?)
217 }
218
219 #[cfg(any(test, feature = "seed-support"))]
220 async fn create_org(&self, name: &str, slug: &str) -> Result<OrgId> {
221 let query = "
222 INSERT INTO orgs (name, slug)
223 VALUES ($1, $2)
224 RETURNING id
225 ";
226 Ok(sqlx::query_scalar(query)
227 .bind(name)
228 .bind(slug)
229 .fetch_one(&self.pool)
230 .await
231 .map(OrgId)?)
232 }
233
234 #[cfg(any(test, feature = "seed-support"))]
235 async fn add_org_member(&self, org_id: OrgId, user_id: UserId, is_admin: bool) -> Result<()> {
236 let query = "
237 INSERT INTO org_memberships (org_id, user_id, admin)
238 VALUES ($1, $2, $3)
239 ON CONFLICT DO NOTHING
240 ";
241 Ok(sqlx::query(query)
242 .bind(org_id.0)
243 .bind(user_id.0)
244 .bind(is_admin)
245 .execute(&self.pool)
246 .await
247 .map(drop)?)
248 }
249
250 // channels
251
252 #[cfg(any(test, feature = "seed-support"))]
253 async fn create_org_channel(&self, org_id: OrgId, name: &str) -> Result<ChannelId> {
254 let query = "
255 INSERT INTO channels (owner_id, owner_is_user, name)
256 VALUES ($1, false, $2)
257 RETURNING id
258 ";
259 Ok(sqlx::query_scalar(query)
260 .bind(org_id.0)
261 .bind(name)
262 .fetch_one(&self.pool)
263 .await
264 .map(ChannelId)?)
265 }
266
267 #[allow(unused)] // Help rust-analyzer
268 #[cfg(any(test, feature = "seed-support"))]
269 async fn get_org_channels(&self, org_id: OrgId) -> Result<Vec<Channel>> {
270 let query = "
271 SELECT *
272 FROM channels
273 WHERE
274 channels.owner_is_user = false AND
275 channels.owner_id = $1
276 ";
277 Ok(sqlx::query_as(query)
278 .bind(org_id.0)
279 .fetch_all(&self.pool)
280 .await?)
281 }
282
283 async fn get_accessible_channels(&self, user_id: UserId) -> Result<Vec<Channel>> {
284 let query = "
285 SELECT
286 channels.*
287 FROM
288 channel_memberships, channels
289 WHERE
290 channel_memberships.user_id = $1 AND
291 channel_memberships.channel_id = channels.id
292 ";
293 Ok(sqlx::query_as(query)
294 .bind(user_id.0)
295 .fetch_all(&self.pool)
296 .await?)
297 }
298
299 async fn can_user_access_channel(
300 &self,
301 user_id: UserId,
302 channel_id: ChannelId,
303 ) -> Result<bool> {
304 let query = "
305 SELECT id
306 FROM channel_memberships
307 WHERE user_id = $1 AND channel_id = $2
308 LIMIT 1
309 ";
310 Ok(sqlx::query_scalar::<_, i32>(query)
311 .bind(user_id.0)
312 .bind(channel_id.0)
313 .fetch_optional(&self.pool)
314 .await
315 .map(|e| e.is_some())?)
316 }
317
318 #[cfg(any(test, feature = "seed-support"))]
319 async fn add_channel_member(
320 &self,
321 channel_id: ChannelId,
322 user_id: UserId,
323 is_admin: bool,
324 ) -> Result<()> {
325 let query = "
326 INSERT INTO channel_memberships (channel_id, user_id, admin)
327 VALUES ($1, $2, $3)
328 ON CONFLICT DO NOTHING
329 ";
330 Ok(sqlx::query(query)
331 .bind(channel_id.0)
332 .bind(user_id.0)
333 .bind(is_admin)
334 .execute(&self.pool)
335 .await
336 .map(drop)?)
337 }
338
339 // messages
340
341 async fn create_channel_message(
342 &self,
343 channel_id: ChannelId,
344 sender_id: UserId,
345 body: &str,
346 timestamp: OffsetDateTime,
347 nonce: u128,
348 ) -> Result<MessageId> {
349 let query = "
350 INSERT INTO channel_messages (channel_id, sender_id, body, sent_at, nonce)
351 VALUES ($1, $2, $3, $4, $5)
352 ON CONFLICT (nonce) DO UPDATE SET nonce = excluded.nonce
353 RETURNING id
354 ";
355 Ok(sqlx::query_scalar(query)
356 .bind(channel_id.0)
357 .bind(sender_id.0)
358 .bind(body)
359 .bind(timestamp)
360 .bind(Uuid::from_u128(nonce))
361 .fetch_one(&self.pool)
362 .await
363 .map(MessageId)?)
364 }
365
366 async fn get_channel_messages(
367 &self,
368 channel_id: ChannelId,
369 count: usize,
370 before_id: Option<MessageId>,
371 ) -> Result<Vec<ChannelMessage>> {
372 let query = r#"
373 SELECT * FROM (
374 SELECT
375 id, channel_id, sender_id, body, sent_at AT TIME ZONE 'UTC' as sent_at, nonce
376 FROM
377 channel_messages
378 WHERE
379 channel_id = $1 AND
380 id < $2
381 ORDER BY id DESC
382 LIMIT $3
383 ) as recent_messages
384 ORDER BY id ASC
385 "#;
386 Ok(sqlx::query_as(query)
387 .bind(channel_id.0)
388 .bind(before_id.unwrap_or(MessageId::MAX))
389 .bind(count as i64)
390 .fetch_all(&self.pool)
391 .await?)
392 }
393
394 #[cfg(test)]
395 async fn teardown(&self, url: &str) {
396 use util::ResultExt;
397
398 let query = "
399 SELECT pg_terminate_backend(pg_stat_activity.pid)
400 FROM pg_stat_activity
401 WHERE pg_stat_activity.datname = current_database() AND pid <> pg_backend_pid();
402 ";
403 sqlx::query(query).execute(&self.pool).await.log_err();
404 self.pool.close().await;
405 <sqlx::Postgres as sqlx::migrate::MigrateDatabase>::drop_database(url)
406 .await
407 .log_err();
408 }
409}
410
411macro_rules! id_type {
412 ($name:ident) => {
413 #[derive(
414 Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, sqlx::Type, Serialize,
415 )]
416 #[sqlx(transparent)]
417 #[serde(transparent)]
418 pub struct $name(pub i32);
419
420 impl $name {
421 #[allow(unused)]
422 pub const MAX: Self = Self(i32::MAX);
423
424 #[allow(unused)]
425 pub fn from_proto(value: u64) -> Self {
426 Self(value as i32)
427 }
428
429 #[allow(unused)]
430 pub fn to_proto(&self) -> u64 {
431 self.0 as u64
432 }
433 }
434
435 impl std::fmt::Display for $name {
436 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
437 self.0.fmt(f)
438 }
439 }
440 };
441}
442
443id_type!(UserId);
444#[derive(Clone, Debug, FromRow, Serialize, PartialEq)]
445pub struct User {
446 pub id: UserId,
447 pub github_login: String,
448 pub admin: bool,
449}
450
451id_type!(OrgId);
452#[derive(FromRow)]
453pub struct Org {
454 pub id: OrgId,
455 pub name: String,
456 pub slug: String,
457}
458
459id_type!(ChannelId);
460#[derive(Clone, Debug, FromRow, Serialize)]
461pub struct Channel {
462 pub id: ChannelId,
463 pub name: String,
464 pub owner_id: i32,
465 pub owner_is_user: bool,
466}
467
468id_type!(MessageId);
469#[derive(Clone, Debug, FromRow)]
470pub struct ChannelMessage {
471 pub id: MessageId,
472 pub channel_id: ChannelId,
473 pub sender_id: UserId,
474 pub body: String,
475 pub sent_at: OffsetDateTime,
476 pub nonce: Uuid,
477}
478
479#[cfg(test)]
480pub mod tests {
481 use super::*;
482 use anyhow::anyhow;
483 use collections::BTreeMap;
484 use gpui::executor::Background;
485 use lazy_static::lazy_static;
486 use parking_lot::Mutex;
487 use rand::prelude::*;
488 use sqlx::{
489 migrate::{MigrateDatabase, Migrator},
490 Postgres,
491 };
492 use std::{path::Path, sync::Arc};
493 use util::post_inc;
494
495 #[tokio::test(flavor = "multi_thread")]
496 async fn test_get_users_by_ids() {
497 for test_db in [
498 TestDb::postgres().await,
499 TestDb::fake(Arc::new(gpui::executor::Background::new())),
500 ] {
501 let db = test_db.db();
502
503 let user = db.create_user("user", false).await.unwrap();
504 let friend1 = db.create_user("friend-1", false).await.unwrap();
505 let friend2 = db.create_user("friend-2", false).await.unwrap();
506 let friend3 = db.create_user("friend-3", false).await.unwrap();
507
508 assert_eq!(
509 db.get_users_by_ids(vec![user, friend1, friend2, friend3])
510 .await
511 .unwrap(),
512 vec![
513 User {
514 id: user,
515 github_login: "user".to_string(),
516 admin: false,
517 },
518 User {
519 id: friend1,
520 github_login: "friend-1".to_string(),
521 admin: false,
522 },
523 User {
524 id: friend2,
525 github_login: "friend-2".to_string(),
526 admin: false,
527 },
528 User {
529 id: friend3,
530 github_login: "friend-3".to_string(),
531 admin: false,
532 }
533 ]
534 );
535 }
536 }
537
538 #[tokio::test(flavor = "multi_thread")]
539 async fn test_recent_channel_messages() {
540 for test_db in [
541 TestDb::postgres().await,
542 TestDb::fake(Arc::new(gpui::executor::Background::new())),
543 ] {
544 let db = test_db.db();
545 let user = db.create_user("user", false).await.unwrap();
546 let org = db.create_org("org", "org").await.unwrap();
547 let channel = db.create_org_channel(org, "channel").await.unwrap();
548 for i in 0..10 {
549 db.create_channel_message(
550 channel,
551 user,
552 &i.to_string(),
553 OffsetDateTime::now_utc(),
554 i,
555 )
556 .await
557 .unwrap();
558 }
559
560 let messages = db.get_channel_messages(channel, 5, None).await.unwrap();
561 assert_eq!(
562 messages.iter().map(|m| &m.body).collect::<Vec<_>>(),
563 ["5", "6", "7", "8", "9"]
564 );
565
566 let prev_messages = db
567 .get_channel_messages(channel, 4, Some(messages[0].id))
568 .await
569 .unwrap();
570 assert_eq!(
571 prev_messages.iter().map(|m| &m.body).collect::<Vec<_>>(),
572 ["1", "2", "3", "4"]
573 );
574 }
575 }
576
577 #[tokio::test(flavor = "multi_thread")]
578 async fn test_channel_message_nonces() {
579 for test_db in [
580 TestDb::postgres().await,
581 TestDb::fake(Arc::new(gpui::executor::Background::new())),
582 ] {
583 let db = test_db.db();
584 let user = db.create_user("user", false).await.unwrap();
585 let org = db.create_org("org", "org").await.unwrap();
586 let channel = db.create_org_channel(org, "channel").await.unwrap();
587
588 let msg1_id = db
589 .create_channel_message(channel, user, "1", OffsetDateTime::now_utc(), 1)
590 .await
591 .unwrap();
592 let msg2_id = db
593 .create_channel_message(channel, user, "2", OffsetDateTime::now_utc(), 2)
594 .await
595 .unwrap();
596 let msg3_id = db
597 .create_channel_message(channel, user, "3", OffsetDateTime::now_utc(), 1)
598 .await
599 .unwrap();
600 let msg4_id = db
601 .create_channel_message(channel, user, "4", OffsetDateTime::now_utc(), 2)
602 .await
603 .unwrap();
604
605 assert_ne!(msg1_id, msg2_id);
606 assert_eq!(msg1_id, msg3_id);
607 assert_eq!(msg2_id, msg4_id);
608 }
609 }
610
611 #[tokio::test(flavor = "multi_thread")]
612 async fn test_create_access_tokens() {
613 let test_db = TestDb::postgres().await;
614 let db = test_db.db();
615 let user = db.create_user("the-user", false).await.unwrap();
616
617 db.create_access_token_hash(user, "h1", 3).await.unwrap();
618 db.create_access_token_hash(user, "h2", 3).await.unwrap();
619 assert_eq!(
620 db.get_access_token_hashes(user).await.unwrap(),
621 &["h2".to_string(), "h1".to_string()]
622 );
623
624 db.create_access_token_hash(user, "h3", 3).await.unwrap();
625 assert_eq!(
626 db.get_access_token_hashes(user).await.unwrap(),
627 &["h3".to_string(), "h2".to_string(), "h1".to_string(),]
628 );
629
630 db.create_access_token_hash(user, "h4", 3).await.unwrap();
631 assert_eq!(
632 db.get_access_token_hashes(user).await.unwrap(),
633 &["h4".to_string(), "h3".to_string(), "h2".to_string(),]
634 );
635
636 db.create_access_token_hash(user, "h5", 3).await.unwrap();
637 assert_eq!(
638 db.get_access_token_hashes(user).await.unwrap(),
639 &["h5".to_string(), "h4".to_string(), "h3".to_string()]
640 );
641 }
642
643 pub struct TestDb {
644 pub db: Option<Arc<dyn Db>>,
645 pub url: String,
646 }
647
648 impl TestDb {
649 pub async fn postgres() -> Self {
650 lazy_static! {
651 static ref LOCK: Mutex<()> = Mutex::new(());
652 }
653
654 let _guard = LOCK.lock();
655 let mut rng = StdRng::from_entropy();
656 let name = format!("zed-test-{}", rng.gen::<u128>());
657 let url = format!("postgres://postgres@localhost/{}", name);
658 let migrations_path = Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/migrations"));
659 Postgres::create_database(&url)
660 .await
661 .expect("failed to create test db");
662 let db = PostgresDb::new(&url, 5).await.unwrap();
663 let migrator = Migrator::new(migrations_path).await.unwrap();
664 migrator.run(&db.pool).await.unwrap();
665 Self {
666 db: Some(Arc::new(db)),
667 url,
668 }
669 }
670
671 pub fn fake(background: Arc<Background>) -> Self {
672 Self {
673 db: Some(Arc::new(FakeDb::new(background))),
674 url: Default::default(),
675 }
676 }
677
678 pub fn db(&self) -> &Arc<dyn Db> {
679 self.db.as_ref().unwrap()
680 }
681 }
682
683 impl Drop for TestDb {
684 fn drop(&mut self) {
685 if let Some(db) = self.db.take() {
686 futures::executor::block_on(db.teardown(&self.url));
687 }
688 }
689 }
690
691 pub struct FakeDb {
692 background: Arc<Background>,
693 users: Mutex<BTreeMap<UserId, User>>,
694 next_user_id: Mutex<i32>,
695 orgs: Mutex<BTreeMap<OrgId, Org>>,
696 next_org_id: Mutex<i32>,
697 org_memberships: Mutex<BTreeMap<(OrgId, UserId), bool>>,
698 channels: Mutex<BTreeMap<ChannelId, Channel>>,
699 next_channel_id: Mutex<i32>,
700 channel_memberships: Mutex<BTreeMap<(ChannelId, UserId), bool>>,
701 channel_messages: Mutex<BTreeMap<MessageId, ChannelMessage>>,
702 next_channel_message_id: Mutex<i32>,
703 }
704
705 impl FakeDb {
706 pub fn new(background: Arc<Background>) -> Self {
707 Self {
708 background,
709 users: Default::default(),
710 next_user_id: Mutex::new(1),
711 orgs: Default::default(),
712 next_org_id: Mutex::new(1),
713 org_memberships: Default::default(),
714 channels: Default::default(),
715 next_channel_id: Mutex::new(1),
716 channel_memberships: Default::default(),
717 channel_messages: Default::default(),
718 next_channel_message_id: Mutex::new(1),
719 }
720 }
721 }
722
723 #[async_trait]
724 impl Db for FakeDb {
725 async fn create_user(&self, github_login: &str, admin: bool) -> Result<UserId> {
726 self.background.simulate_random_delay().await;
727
728 let mut users = self.users.lock();
729 if let Some(user) = users
730 .values()
731 .find(|user| user.github_login == github_login)
732 {
733 Ok(user.id)
734 } else {
735 let user_id = UserId(post_inc(&mut *self.next_user_id.lock()));
736 users.insert(
737 user_id,
738 User {
739 id: user_id,
740 github_login: github_login.to_string(),
741 admin,
742 },
743 );
744 Ok(user_id)
745 }
746 }
747
748 async fn get_all_users(&self) -> Result<Vec<User>> {
749 unimplemented!()
750 }
751
752 async fn get_user_by_id(&self, id: UserId) -> Result<Option<User>> {
753 Ok(self.get_users_by_ids(vec![id]).await?.into_iter().next())
754 }
755
756 async fn get_users_by_ids(&self, ids: Vec<UserId>) -> Result<Vec<User>> {
757 self.background.simulate_random_delay().await;
758 let users = self.users.lock();
759 Ok(ids.iter().filter_map(|id| users.get(id).cloned()).collect())
760 }
761
762 async fn get_user_by_github_login(&self, _github_login: &str) -> Result<Option<User>> {
763 unimplemented!()
764 }
765
766 async fn set_user_is_admin(&self, _id: UserId, _is_admin: bool) -> Result<()> {
767 unimplemented!()
768 }
769
770 async fn destroy_user(&self, _id: UserId) -> Result<()> {
771 unimplemented!()
772 }
773
774 async fn create_access_token_hash(
775 &self,
776 _user_id: UserId,
777 _access_token_hash: &str,
778 _max_access_token_count: usize,
779 ) -> Result<()> {
780 unimplemented!()
781 }
782
783 async fn get_access_token_hashes(&self, _user_id: UserId) -> Result<Vec<String>> {
784 unimplemented!()
785 }
786
787 async fn find_org_by_slug(&self, _slug: &str) -> Result<Option<Org>> {
788 unimplemented!()
789 }
790
791 async fn create_org(&self, name: &str, slug: &str) -> Result<OrgId> {
792 self.background.simulate_random_delay().await;
793 let mut orgs = self.orgs.lock();
794 if orgs.values().any(|org| org.slug == slug) {
795 Err(anyhow!("org already exists"))
796 } else {
797 let org_id = OrgId(post_inc(&mut *self.next_org_id.lock()));
798 orgs.insert(
799 org_id,
800 Org {
801 id: org_id,
802 name: name.to_string(),
803 slug: slug.to_string(),
804 },
805 );
806 Ok(org_id)
807 }
808 }
809
810 async fn add_org_member(
811 &self,
812 org_id: OrgId,
813 user_id: UserId,
814 is_admin: bool,
815 ) -> Result<()> {
816 self.background.simulate_random_delay().await;
817 if !self.orgs.lock().contains_key(&org_id) {
818 return Err(anyhow!("org does not exist"));
819 }
820 if !self.users.lock().contains_key(&user_id) {
821 return Err(anyhow!("user does not exist"));
822 }
823
824 self.org_memberships
825 .lock()
826 .entry((org_id, user_id))
827 .or_insert(is_admin);
828 Ok(())
829 }
830
831 async fn create_org_channel(&self, org_id: OrgId, name: &str) -> Result<ChannelId> {
832 self.background.simulate_random_delay().await;
833 if !self.orgs.lock().contains_key(&org_id) {
834 return Err(anyhow!("org does not exist"));
835 }
836
837 let mut channels = self.channels.lock();
838 let channel_id = ChannelId(post_inc(&mut *self.next_channel_id.lock()));
839 channels.insert(
840 channel_id,
841 Channel {
842 id: channel_id,
843 name: name.to_string(),
844 owner_id: org_id.0,
845 owner_is_user: false,
846 },
847 );
848 Ok(channel_id)
849 }
850
851 async fn get_org_channels(&self, org_id: OrgId) -> Result<Vec<Channel>> {
852 self.background.simulate_random_delay().await;
853 Ok(self
854 .channels
855 .lock()
856 .values()
857 .filter(|channel| !channel.owner_is_user && channel.owner_id == org_id.0)
858 .cloned()
859 .collect())
860 }
861
862 async fn get_accessible_channels(&self, user_id: UserId) -> Result<Vec<Channel>> {
863 self.background.simulate_random_delay().await;
864 let channels = self.channels.lock();
865 let memberships = self.channel_memberships.lock();
866 Ok(channels
867 .values()
868 .filter(|channel| memberships.contains_key(&(channel.id, user_id)))
869 .cloned()
870 .collect())
871 }
872
873 async fn can_user_access_channel(
874 &self,
875 user_id: UserId,
876 channel_id: ChannelId,
877 ) -> Result<bool> {
878 self.background.simulate_random_delay().await;
879 Ok(self
880 .channel_memberships
881 .lock()
882 .contains_key(&(channel_id, user_id)))
883 }
884
885 async fn add_channel_member(
886 &self,
887 channel_id: ChannelId,
888 user_id: UserId,
889 is_admin: bool,
890 ) -> Result<()> {
891 self.background.simulate_random_delay().await;
892 if !self.channels.lock().contains_key(&channel_id) {
893 return Err(anyhow!("channel does not exist"));
894 }
895 if !self.users.lock().contains_key(&user_id) {
896 return Err(anyhow!("user does not exist"));
897 }
898
899 self.channel_memberships
900 .lock()
901 .entry((channel_id, user_id))
902 .or_insert(is_admin);
903 Ok(())
904 }
905
906 async fn create_channel_message(
907 &self,
908 channel_id: ChannelId,
909 sender_id: UserId,
910 body: &str,
911 timestamp: OffsetDateTime,
912 nonce: u128,
913 ) -> Result<MessageId> {
914 self.background.simulate_random_delay().await;
915 if !self.channels.lock().contains_key(&channel_id) {
916 return Err(anyhow!("channel does not exist"));
917 }
918 if !self.users.lock().contains_key(&sender_id) {
919 return Err(anyhow!("user does not exist"));
920 }
921
922 let mut messages = self.channel_messages.lock();
923 if let Some(message) = messages
924 .values()
925 .find(|message| message.nonce.as_u128() == nonce)
926 {
927 Ok(message.id)
928 } else {
929 let message_id = MessageId(post_inc(&mut *self.next_channel_message_id.lock()));
930 messages.insert(
931 message_id,
932 ChannelMessage {
933 id: message_id,
934 channel_id,
935 sender_id,
936 body: body.to_string(),
937 sent_at: timestamp,
938 nonce: Uuid::from_u128(nonce),
939 },
940 );
941 Ok(message_id)
942 }
943 }
944
945 async fn get_channel_messages(
946 &self,
947 channel_id: ChannelId,
948 count: usize,
949 before_id: Option<MessageId>,
950 ) -> Result<Vec<ChannelMessage>> {
951 let mut messages = self
952 .channel_messages
953 .lock()
954 .values()
955 .rev()
956 .filter(|message| {
957 message.channel_id == channel_id
958 && message.id < before_id.unwrap_or(MessageId::MAX)
959 })
960 .take(count)
961 .cloned()
962 .collect::<Vec<_>>();
963 messages.sort_unstable_by_key(|message| message.id);
964 Ok(messages)
965 }
966
967 async fn teardown(&self, _: &str) {}
968 }
969}