steid
Sign in

@jamesgill /

steid

steid/src/infrastructure/repository/sqlite.rs
16.9 KBCode·Blame·Raw
bf4b351feat: sqlite repository implementations1mo
1//! SQLite repository implementations.
2//!
3//! Rows are reassembled with `from_trusted`: they were validated on the way in, and
4//! re-validating them would make a tightened rule turn old rows unreadable.
5
54a0eaffeat: session storage and actor resolution1mo
6use std::time::{Duration, SystemTime};
7
bf4b351feat: sqlite repository implementations1mo
8use sqlx::{Row, SqlitePool, sqlite::SqliteRow};
9
10use crate::domain::{
54a0eaffeat: session storage and actor resolution1mo
11 Email, Membership, MembershipId, OrgId, OrgName, Organization, PasswordHash, Role, Session,
12 SessionTokenHash, User, UserId,
bf4b351feat: sqlite repository implementations1mo
13 repository::{
54a0eaffeat: session storage and actor resolution1mo
14 MembershipRepository, OrgRepository, RepositoryError, RepositoryResult, SessionRepository,
15 UserRepository,
bf4b351feat: sqlite repository implementations1mo
16 },
17};
18
19fn backend(error: sqlx::Error) -> RepositoryError {
20 RepositoryError::backend(error)
21}
22
23#[derive(Debug, Clone)]
24pub struct SqliteUserRepo {
25 pool: SqlitePool,
26}
27
28impl SqliteUserRepo {
29 pub fn new(pool: SqlitePool) -> Self {
30 Self { pool }
31 }
32
33 fn map(row: &SqliteRow) -> User {
34 User::new(
35 UserId::from_trusted(row.get::<String, _>("id")),
36 Email::from_trusted(row.get::<String, _>("email")),
37 PasswordHash::from_trusted(row.get::<String, _>("password_hash")),
38 OrgId::from_trusted(row.get::<String, _>("personal_org_id")),
39 )
40 }
41}
42
43impl UserRepository for SqliteUserRepo {
44 async fn find_by_id(&self, id: &UserId) -> RepositoryResult<Option<User>> {
45 let row = sqlx::query("select * from users where id = ?")
46 .bind(id.as_str())
47 .fetch_optional(&self.pool)
48 .await
49 .map_err(backend)?;
50
51 Ok(row.as_ref().map(Self::map))
52 }
53
54 async fn find_by_email(&self, email: &Email) -> RepositoryResult<Option<User>> {
55 let row = sqlx::query("select * from users where email = ?")
56 .bind(email.as_str())
57 .fetch_optional(&self.pool)
58 .await
59 .map_err(backend)?;
60
61 Ok(row.as_ref().map(Self::map))
62 }
63
64 async fn save(&self, user: &User) -> RepositoryResult<()> {
65 sqlx::query(
66 "insert into users (id, email, password_hash, personal_org_id)
67 values (?, ?, ?, ?)
68 on conflict (id) do update set
69 email = excluded.email,
70 password_hash = excluded.password_hash,
71 personal_org_id = excluded.personal_org_id",
72 )
73 .bind(user.id.as_str())
74 .bind(user.email.as_str())
75 .bind(user.password_hash.as_str())
76 .bind(user.personal_org_id.as_str())
77 .execute(&self.pool)
78 .await
79 .map_err(backend)?;
80
81 Ok(())
82 }
83
84 async fn any_exist(&self) -> RepositoryResult<bool> {
85 let count: i64 = sqlx::query_scalar("select exists (select 1 from users)")
86 .fetch_one(&self.pool)
87 .await
88 .map_err(backend)?;
89
90 Ok(count != 0)
91 }
92}
93
94#[derive(Debug, Clone)]
95pub struct SqliteOrgRepo {
96 pool: SqlitePool,
97}
98
99impl SqliteOrgRepo {
100 pub fn new(pool: SqlitePool) -> Self {
101 Self { pool }
102 }
103
104 fn map(row: &SqliteRow) -> Organization {
105 Organization::from_trusted(
106 OrgId::from_trusted(row.get::<String, _>("id")),
107 OrgName::from_trusted(row.get::<String, _>("name")),
108 row.get::<Option<String>, _>("display_name"),
109 )
110 }
111}
112
113impl OrgRepository for SqliteOrgRepo {
114 async fn find_by_id(&self, id: &OrgId) -> RepositoryResult<Option<Organization>> {
115 let row = sqlx::query("select * from orgs where id = ?")
116 .bind(id.as_str())
117 .fetch_optional(&self.pool)
118 .await
119 .map_err(backend)?;
120
121 Ok(row.as_ref().map(Self::map))
122 }
123
124 async fn find_by_name(&self, name: &OrgName) -> RepositoryResult<Option<Organization>> {
125 let row = sqlx::query("select * from orgs where name = ?")
126 .bind(name.as_str())
127 .fetch_optional(&self.pool)
128 .await
129 .map_err(backend)?;
130
131 Ok(row.as_ref().map(Self::map))
132 }
133
134 async fn save(&self, org: &Organization) -> RepositoryResult<()> {
135 sqlx::query(
136 "insert into orgs (id, name, display_name)
137 values (?, ?, ?)
138 on conflict (id) do update set
139 name = excluded.name,
140 display_name = excluded.display_name",
141 )
142 .bind(org.id.as_str())
143 .bind(org.name.as_str())
144 .bind(org.display_name.as_deref())
145 .execute(&self.pool)
146 .await
147 .map_err(backend)?;
148
149 Ok(())
150 }
151}
152
153#[derive(Debug, Clone)]
154pub struct SqliteMembershipRepo {
155 pool: SqlitePool,
156}
157
158impl SqliteMembershipRepo {
159 pub fn new(pool: SqlitePool) -> Self {
160 Self { pool }
161 }
162
163 /// An unparseable role is a storage fault, not a missing membership, so it
164 /// surfaces rather than silently downgrading the member's access.
165 fn map(row: &SqliteRow) -> RepositoryResult<Membership> {
166 let raw: String = row.get("role");
167 let role: Role = raw
168 .parse()
169 .map_err(|error| RepositoryError::backend(format!("{error}")))?;
170
171 Ok(Membership::new(
172 MembershipId::from_trusted(row.get::<String, _>("id")),
173 OrgId::from_trusted(row.get::<String, _>("org_id")),
174 UserId::from_trusted(row.get::<String, _>("user_id")),
175 role,
176 ))
177 }
178}
179
180impl MembershipRepository for SqliteMembershipRepo {
181 async fn find(&self, org_id: &OrgId, user_id: &UserId) -> RepositoryResult<Option<Membership>> {
182 let row = sqlx::query("select * from memberships where org_id = ? and user_id = ?")
183 .bind(org_id.as_str())
184 .bind(user_id.as_str())
185 .fetch_optional(&self.pool)
186 .await
187 .map_err(backend)?;
188
189 row.as_ref().map(Self::map).transpose()
190 }
191
192 async fn list_for_user(&self, user_id: &UserId) -> RepositoryResult<Vec<Membership>> {
193 let rows = sqlx::query("select * from memberships where user_id = ?")
194 .bind(user_id.as_str())
195 .fetch_all(&self.pool)
196 .await
197 .map_err(backend)?;
198
199 rows.iter().map(Self::map).collect()
200 }
201
202 async fn save(&self, membership: &Membership) -> RepositoryResult<()> {
203 sqlx::query(
204 "insert into memberships (id, org_id, user_id, role)
205 values (?, ?, ?, ?)
206 on conflict (id) do update set role = excluded.role",
207 )
208 .bind(membership.id.as_str())
209 .bind(membership.org_id.as_str())
210 .bind(membership.user_id.as_str())
211 .bind(membership.role.as_str())
212 .execute(&self.pool)
213 .await
214 .map_err(backend)?;
215
216 Ok(())
217 }
218}
219
220#[cfg(test)]
221mod tests {
222 use super::*;
223 use crate::{
224 application::{OwnerSpec, claim_instance, port::PasswordHasher},
225 domain::SetupToken,
226 infrastructure::{database::test_support::test_pool, password::StubHasher},
227 };
228
229 struct Repos {
230 users: SqliteUserRepo,
231 orgs: SqliteOrgRepo,
232 memberships: SqliteMembershipRepo,
233 }
234
235 async fn repos() -> Repos {
236 let pool = test_pool().await;
237 Repos {
238 users: SqliteUserRepo::new(pool.clone()),
239 orgs: SqliteOrgRepo::new(pool.clone()),
240 memberships: SqliteMembershipRepo::new(pool),
241 }
242 }
243
244 async fn saved_org(repos: &Repos, name: &str) -> Organization {
245 let org = Organization::new(OrgId::generate(), name, None).expect("valid org");
246 repos.orgs.save(&org).await.expect("save org");
247 org
248 }
249
250 async fn saved_user(repos: &Repos, email: &str, org: &Organization) -> User {
251 let user = User::new(
252 UserId::generate(),
253 Email::new(email).expect("valid email"),
254 PasswordHash::from_trusted("$argon2id$test"),
255 org.id.clone(),
256 );
257 repos.users.save(&user).await.expect("save user");
258 user
259 }
260
261 #[tokio::test]
262 async fn a_saved_user_round_trips() {
263 let repos = repos().await;
264 let org = saved_org(&repos, "james").await;
265 let user = saved_user(&repos, "dev@example.com", &org).await;
266
267 let found = repos
268 .users
269 .find_by_id(&user.id)
270 .await
271 .expect("lookup")
272 .expect("user should exist");
273
274 assert_eq!(found, user);
275 }
276
277 #[tokio::test]
278 async fn users_are_found_by_email_case_insensitively() {
279 let repos = repos().await;
280 let org = saved_org(&repos, "james").await;
281 saved_user(&repos, "dev@example.com", &org).await;
282
283 // Email lowercases on construction, but a row written before that rule would
284 // still need finding.
285 let found = repos
286 .users
287 .find_by_email(&Email::from_trusted("DEV@EXAMPLE.COM"))
288 .await
289 .expect("lookup");
290
291 assert!(found.is_some(), "collate nocase should make this match");
292 }
293
294 #[tokio::test]
295 async fn an_org_round_trips_with_its_display_name() {
296 let repos = repos().await;
7148ad4feat: expand the reserved-handle list to twenty1mo
297 let org = Organization::new(OrgId::generate(), "acme", Some("Acme".to_owned()))
bf4b351feat: sqlite repository implementations1mo
298 .expect("valid org");
299 repos.orgs.save(&org).await.expect("save");
300
301 let found = repos
302 .orgs
303 .find_by_name(&org.name)
304 .await
305 .expect("lookup")
306 .expect("org should exist");
307
308 assert_eq!(found, org);
7148ad4feat: expand the reserved-handle list to twenty1mo
309 assert_eq!(found.label(), "Acme");
bf4b351feat: sqlite repository implementations1mo
310 }
311
312 #[tokio::test]
313 async fn a_missing_org_is_none_not_an_error() {
314 let repos = repos().await;
315
316 let found = repos
317 .orgs
318 .find_by_name(&OrgName::new("nobody").unwrap())
319 .await
320 .expect("lookup");
321
322 assert_eq!(found, None);
323 }
324
325 #[tokio::test]
326 async fn a_membership_round_trips_with_its_role() {
327 let repos = repos().await;
328 let org = saved_org(&repos, "james").await;
329 let user = saved_user(&repos, "dev@example.com", &org).await;
330 let membership = Membership::new(
331 MembershipId::generate(),
332 org.id.clone(),
333 user.id.clone(),
334 Role::Owner,
335 );
336 repos.memberships.save(&membership).await.expect("save");
337
338 let found = repos
339 .memberships
340 .find(&org.id, &user.id)
341 .await
342 .expect("lookup")
343 .expect("membership should exist");
344
345 assert_eq!(found, membership);
346 assert!(found.can_write());
347 }
348
349 #[tokio::test]
350 async fn an_unreadable_role_surfaces_rather_than_downgrading_access() {
351 let repos = repos().await;
352 let pool = test_pool().await;
353 let memberships = SqliteMembershipRepo::new(pool.clone());
354 let org = Organization::new(OrgId::generate(), "james", None).expect("valid org");
355 SqliteOrgRepo::new(pool.clone())
356 .save(&org)
357 .await
358 .expect("save org");
359 let user = User::new(
360 UserId::generate(),
361 Email::new("dev@example.com").expect("valid email"),
362 PasswordHash::from_trusted("$argon2id$test"),
363 org.id.clone(),
364 );
365 SqliteUserRepo::new(pool.clone())
366 .save(&user)
367 .await
368 .expect("save user");
369
370 sqlx::query("insert into memberships (id, org_id, user_id, role) values (?, ?, ?, 'wat')")
371 .bind(MembershipId::generate().as_str())
372 .bind(org.id.as_str())
373 .bind(user.id.as_str())
374 .execute(&pool)
375 .await
376 .expect("insert");
377
378 let result = memberships.find(&org.id, &user.id).await;
379
380 assert!(
381 result.is_err(),
382 "a role we can't parse must not read as no membership"
383 );
384 drop(repos);
385 }
386
387 #[tokio::test]
388 async fn saving_a_user_before_its_org_is_refused() {
389 let repos = repos().await;
390 let orphan = User::new(
391 UserId::generate(),
392 Email::new("dev@example.com").expect("valid email"),
393 PasswordHash::from_trusted("$argon2id$test"),
394 OrgId::generate(),
395 );
396
397 let result = repos.users.save(&orphan).await;
398
399 assert!(
400 result.is_err(),
401 "the foreign key should reject a user whose org doesn't exist"
402 );
403 }
404
405 #[tokio::test]
406 async fn a_duplicate_handle_is_refused() {
407 let repos = repos().await;
408 saved_org(&repos, "james").await;
409
410 let clash = Organization::new(OrgId::generate(), "james", None).expect("valid org");
411 let result = repos.orgs.save(&clash).await;
412
413 assert!(result.is_err(), "orgs.name is unique");
414 }
415
416 #[tokio::test]
417 async fn a_differently_cased_handle_is_also_refused() {
418 let repos = repos().await;
419 saved_org(&repos, "james").await;
420
421 // OrgName lowercases, so this can only arrive via from_trusted -- but the
422 // constraint is what we're testing, not the value object.
423 let clash =
424 Organization::from_trusted(OrgId::generate(), OrgName::from_trusted("JAMES"), None);
425 let result = repos.orgs.save(&clash).await;
426
427 assert!(result.is_err(), "collate nocase should catch this");
428 }
429
430 /// The claim use case checks `is_claimed` and then writes, which is TOCTOU. This
431 /// asserts the database is what actually stops a second owner being created.
432 #[tokio::test]
433 async fn a_second_claim_is_stopped_by_the_database_not_the_check() {
434 let repos = repos().await;
435 let token = SetupToken::generate();
436 let hasher = StubHasher::new();
437 let spec = OwnerSpec {
438 handle: "james".to_owned(),
439 email: "dev@example.com".to_owned(),
440 password: "hunter2".to_owned(),
441 };
442
443 claim_instance(
444 token.reveal(),
445 &token,
446 &spec,
447 &repos.users,
448 &repos.orgs,
449 &repos.memberships,
450 &hasher,
451 )
452 .await
453 .expect("first claim");
454
455 // Simulate the race: the second claimant passed is_claimed before the first
456 // committed, so it proceeds straight to the writes.
457 let intruder_org = Organization::new(OrgId::generate(), "james", None).expect("valid org");
458 let result = repos.orgs.save(&intruder_org).await;
459
460 assert!(
461 result.is_err(),
462 "unique(orgs.name) is what actually serialises concurrent claims"
463 );
464
465 let intruder_user = User::new(
466 UserId::generate(),
467 Email::new("dev@example.com").expect("valid email"),
468 hasher.hash("letmein").expect("hash"),
469 OrgId::generate(),
470 );
471 assert!(
472 repos.users.save(&intruder_user).await.is_err(),
473 "unique(users.email) closes the other half"
474 );
475 }
476}
54a0eaffeat: session storage and actor resolution1mo
477
478#[derive(Debug, Clone)]
479pub struct SqliteSessionRepo {
480 pool: SqlitePool,
481}
482
483impl SqliteSessionRepo {
484 pub fn new(pool: SqlitePool) -> Self {
485 Self { pool }
486 }
487}
488
489/// Unix seconds. Times before the epoch cannot occur here — sessions always expire in
490/// the future — so saturating at 0 is safe rather than lossy.
491fn to_unix(time: SystemTime) -> i64 {
492 time.duration_since(SystemTime::UNIX_EPOCH)
493 .map(|d| d.as_secs() as i64)
494 .unwrap_or(0)
495}
496
497fn from_unix(seconds: i64) -> SystemTime {
498 SystemTime::UNIX_EPOCH + Duration::from_secs(seconds.max(0) as u64)
499}
500
501impl SessionRepository for SqliteSessionRepo {
502 async fn find(&self, token_hash: &SessionTokenHash) -> RepositoryResult<Option<Session>> {
503 let row = sqlx::query("select * from sessions where token_hash = ?")
504 .bind(token_hash.as_str())
505 .fetch_optional(&self.pool)
506 .await
507 .map_err(backend)?;
508
509 Ok(row.map(|row| {
510 Session::new(
511 SessionTokenHash::from_trusted(row.get::<String, _>("token_hash")),
512 UserId::from_trusted(row.get::<String, _>("user_id")),
513 from_unix(row.get::<i64, _>("expires_at")),
514 )
515 }))
516 }
517
518 async fn save(&self, session: &Session) -> RepositoryResult<()> {
519 sqlx::query(
520 "insert into sessions (token_hash, user_id, expires_at)
521 values (?, ?, ?)
522 on conflict (token_hash) do update set
523 user_id = excluded.user_id,
524 expires_at = excluded.expires_at",
525 )
526 .bind(session.token_hash.as_str())
527 .bind(session.user_id.as_str())
528 .bind(to_unix(session.expires_at))
529 .execute(&self.pool)
530 .await
531 .map_err(backend)?;
532
533 Ok(())
534 }
535
536 async fn delete(&self, token_hash: &SessionTokenHash) -> RepositoryResult<()> {
537 sqlx::query("delete from sessions where token_hash = ?")
538 .bind(token_hash.as_str())
539 .execute(&self.pool)
540 .await
541 .map_err(backend)?;
542
543 Ok(())
544 }
545
546 async fn delete_expired(&self, now: SystemTime) -> RepositoryResult<u64> {
547 let result = sqlx::query("delete from sessions where expires_at <= ?")
548 .bind(to_unix(now))
549 .execute(&self.pool)
550 .await
551 .map_err(backend)?;
552
553 Ok(result.rows_affected())
554 }
555}