goat.navy
forked from
tranquil.farm/tranquil-pds
Our Personal Data Server from scratch!
204 kB
6342 lines
1use std::sync::Arc;
2use std::sync::atomic::{AtomicUsize, Ordering};
3use std::thread::JoinHandle;
4
5use chrono::{DateTime, Utc};
6use tokio::sync::oneshot;
7use tranquil_db_traits::DbScope;
8use tranquil_db_traits::{
9 AccountSearchResult, AccountStatus, AdminAccountInfo, ApplyCommitError, ApplyCommitInput,
10 ApplyCommitResult, Backlink, CommitEventData, CommsChannel, CommsType,
11 CompletePasskeySetupInput, CreateAccountError, CreateDelegatedAccountInput,
12 CreatePasskeyAccountInput, CreatePasswordAccountInput, CreatePasswordAccountResult,
13 CreateSsoAccountInput, DbError, DelegationActionType, DeletionRequest,
14 DeletionRequestWithToken, DidWebOverrides, ImportBlock, ImportRecord, ImportRepoError,
15 InviteCodeError, InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder, InviteCodeUse,
16 MigrationReactivationError, MigrationReactivationInput, NotificationHistoryRow,
17 NotificationPrefs, OAuthTokenWithUser, PasswordResetResult, PlcTokenInfo, QueuedComms,
18 ReactivatedAccountInfo, RecoverPasskeyAccountInput, RecoverPasskeyAccountResult,
19 RefreshSessionResult, ReservedSigningKey, ReservedSigningKeyFull, ScheduledDeletionAccount,
20 ScopePreference, SequenceNumber, SequencedEvent, SessionId, StoredBackupCode, StoredPasskey,
21 TokenFamilyId, TotpRecord, TotpRecordState, User2faStatus, UserAuthInfo, UserCommsPrefs,
22 UserConfirmSignup, UserDidWebInfo, UserEmailInfo, UserForDeletion, UserForDidDoc,
23 UserForDidDocBuild, UserForPasskeyRecovery, UserForPasskeySetup, UserForRecovery,
24 UserForVerification, UserIdAndHandle, UserIdAndPasswordHash, UserIdHandleEmail,
25 UserInfoForAuth, UserKeyInfo, UserKeyWithId, UserLegacyLoginPref, UserLoginCheck,
26 UserLoginFull, UserLoginInfo, UserNeedingRecordBlobsBackfill, UserPasswordInfo,
27 UserResendVerification, UserResetCodeInfo, UserRow, UserSessionInfo, UserStatus,
28 UserVerificationInfo, UserWithKey, UserWithoutBlocks, ValidatedInviteCode,
29 WebauthnChallengeType,
30};
31use tranquil_oauth::{AuthorizedClientData, DeviceData, RequestData, TokenData};
32use tranquil_types::{
33 AtUri, AuthorizationCode, CidLink, ClientId, DPoPProofId, DeviceId, Did, Handle, InviteCode,
34 Nsid, RefreshToken, RequestId, Rkey, Tid, TokenId,
35};
36use uuid::Uuid;
37
38use super::MetastoreError;
39use super::commit_ops::CommitOps;
40use super::event_ops::EventOps;
41use super::infra_ops::InfraOps;
42use super::keys::UserHash;
43use super::record_ops::ListRecordsQuery;
44use super::user_hash::UserHashMap;
45use crate::blockstore::TranquilBlockStore;
46use crate::clock::SystemClock;
47use crate::eventlog::EventLogBridge;
48use crate::io::{RealIO, StorageIO};
49use crate::metastore::Metastore;
50
51type Tx<T> = oneshot::Sender<Result<T, DbError>>;
52
53fn reserve_invite(infra: &InfraOps, code: Option<&InviteCode>) -> Result<(), CreateAccountError> {
54 match code {
55 Some(code) => infra.reserve_invite_code(code).map_err(|e| match e {
56 InviteCodeError::DatabaseError(e) => CreateAccountError::Database(e.to_string()),
57 _ => CreateAccountError::InviteCodeUnavailable,
58 }),
59 None => Ok(()),
60 }
61}
62
63fn record_invite_use(
64 infra: &InfraOps,
65 code: Option<&InviteCode>,
66 user_id: Uuid,
67) -> Result<(), CreateAccountError> {
68 match code {
69 Some(code) => {
70 let validated = ValidatedInviteCode::new_validated(code);
71 infra
72 .record_invite_code_use(&validated, user_id)
73 .map_err(|e| CreateAccountError::Database(e.to_string()))
74 }
75 None => Ok(()),
76 }
77}
78
79fn refund_invite(infra: &InfraOps, code: Option<&InviteCode>) {
80 if let Some(code) = code
81 && let Err(e) = infra.refund_invite_code(code)
82 {
83 tracing::error!("failed to refund invite code after account creation failed: {e:?}");
84 }
85}
86
87fn finalize_account<T>(
88 infra: &InfraOps,
89 code: Option<&InviteCode>,
90 created: Result<T, CreateAccountError>,
91 after: impl FnOnce(&T) -> Result<Uuid, CreateAccountError>,
92) -> Result<T, CreateAccountError> {
93 match created {
94 Ok(result) => {
95 let user_id = after(&result)?;
96 record_invite_use(infra, code, user_id)?;
97 Ok(result)
98 }
99 Err(e) => {
100 refund_invite(infra, code);
101 Err(e)
102 }
103 }
104}
105
106fn metastore_to_db(e: MetastoreError) -> DbError {
107 match e {
108 MetastoreError::Fjall(e) => DbError::Query(e.to_string()),
109 MetastoreError::Lsm(e) => DbError::Query(e.to_string()),
110 MetastoreError::VersionMismatch { expected, found } => DbError::Query(format!(
111 "format version mismatch: expected {expected}, found {found}"
112 )),
113 MetastoreError::CorruptData(msg) => DbError::CorruptData(msg),
114 MetastoreError::InvalidInput(msg) => DbError::Query(msg.to_string()),
115 MetastoreError::UserHashCollision {
116 hash,
117 existing_uuid,
118 new_uuid,
119 } => DbError::Constraint(format!(
120 "user hash collision: {hash} maps to both {existing_uuid} and {new_uuid}"
121 )),
122 MetastoreError::UniqueViolation(constraint) => {
123 DbError::Constraint(format!("unique constraint violated: {constraint}"))
124 }
125 }
126}
127
128#[derive(Debug, Clone, Copy, PartialEq, Eq)]
129enum Routing {
130 Sharded(u64),
131 Global,
132}
133
134fn uuid_to_routing(user_hashes: &UserHashMap, user_id: &Uuid) -> Routing {
135 match user_hashes.get(user_id) {
136 Some(h) => Routing::Sharded(h.raw()),
137 None => Routing::Sharded(user_id.as_u128() as u64),
138 }
139}
140
141fn did_to_routing(did: &str) -> Routing {
142 Routing::Sharded(UserHash::from_did(did).raw())
143}
144
145fn cid_to_routing(cid: &CidLink) -> Routing {
146 use siphasher::sip::SipHasher24;
147 use std::hash::{Hash, Hasher};
148 let mut hasher = SipHasher24::new();
149 cid.as_str().hash(&mut hasher);
150 Routing::Sharded(hasher.finish())
151}
152
153pub enum MetastoreRequest {
154 Repo(RepoRequest),
155 Record(RecordRequest),
156 UserBlock(UserBlockRequest),
157 Event(EventRequest),
158 Commit(Box<CommitRequest>),
159 Backlink(BacklinkRequest),
160 Blob(BlobRequest),
161 Delegation(DelegationRequest),
162 Sso(SsoRequest),
163 Session(SessionRequest),
164 Infra(InfraRequest),
165 OAuth(OAuthRequest),
166 User(UserRequest),
167}
168
169impl MetastoreRequest {
170 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
171 match self {
172 Self::Repo(r) => r.routing(user_hashes),
173 Self::Record(r) => r.routing(user_hashes),
174 Self::UserBlock(r) => r.routing(user_hashes),
175 Self::Event(r) => r.routing(),
176 Self::Commit(r) => r.routing(user_hashes),
177 Self::Backlink(r) => r.routing(user_hashes),
178 Self::Blob(r) => r.routing(user_hashes),
179 Self::Delegation(r) => r.routing(),
180 Self::Sso(r) => r.routing(),
181 Self::Session(r) => r.routing(user_hashes),
182 Self::Infra(r) => r.routing(user_hashes),
183 Self::OAuth(r) => r.routing(),
184 Self::User(r) => r.routing(user_hashes),
185 }
186 }
187}
188
189pub enum RepoRequest {
190 CreateRepoFull {
191 user_id: Uuid,
192 did: Did,
193 handle: Handle,
194 repo_root_cid: CidLink,
195 repo_rev: Tid,
196 tx: Tx<()>,
197 },
198 UpdateRepoRoot {
199 user_id: Uuid,
200 repo_root_cid: CidLink,
201 repo_rev: String,
202 tx: Tx<()>,
203 },
204 UpdateRepoRev {
205 user_id: Uuid,
206 repo_rev: String,
207 tx: Tx<()>,
208 },
209 DeleteRepo {
210 user_id: Uuid,
211 tx: Tx<()>,
212 },
213 GetRepoRootForUpdate {
214 user_id: Uuid,
215 tx: Tx<Option<CidLink>>,
216 },
217 GetRepo {
218 user_id: Uuid,
219 tx: Tx<Option<tranquil_db_traits::RepoInfo>>,
220 },
221 GetRepoRootByDid {
222 did: Did,
223 tx: Tx<Option<CidLink>>,
224 },
225 CountRepos {
226 tx: Tx<i64>,
227 },
228 GetReposWithoutRev {
229 tx: Tx<Vec<tranquil_db_traits::RepoWithoutRev>>,
230 },
231 GetRepoRootCidByUserId {
232 user_id: Uuid,
233 tx: Tx<Option<CidLink>>,
234 },
235 GetAccountWithRepo {
236 did: Did,
237 tx: Tx<Option<tranquil_db_traits::RepoAccountInfo>>,
238 },
239 ListReposPaginated {
240 cursor_user_hash: Option<u64>,
241 limit: usize,
242 tx: Tx<Vec<tranquil_db_traits::RepoListItem>>,
243 },
244 UpdateRepoStatus {
245 did: Did,
246 takedown: Option<bool>,
247 takedown_ref: Option<String>,
248 deactivated: Option<bool>,
249 tx: Tx<()>,
250 },
251}
252
253impl RepoRequest {
254 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
255 match self {
256 Self::CreateRepoFull { did, .. } => did_to_routing(did),
257 Self::UpdateRepoRoot { user_id, .. }
258 | Self::UpdateRepoRev { user_id, .. }
259 | Self::DeleteRepo { user_id, .. }
260 | Self::GetRepoRootForUpdate { user_id, .. }
261 | Self::GetRepo { user_id, .. }
262 | Self::GetRepoRootCidByUserId { user_id, .. } => uuid_to_routing(user_hashes, user_id),
263 Self::GetRepoRootByDid { did, .. }
264 | Self::GetAccountWithRepo { did, .. }
265 | Self::UpdateRepoStatus { did, .. } => did_to_routing(did),
266 Self::CountRepos { .. }
267 | Self::GetReposWithoutRev { .. }
268 | Self::ListReposPaginated { .. } => Routing::Global,
269 }
270 }
271}
272
273pub enum RecordRequest {
274 UpsertRecords {
275 repo_id: Uuid,
276 collections: Vec<Nsid>,
277 rkeys: Vec<Rkey>,
278 record_cids: Vec<CidLink>,
279 repo_rev: String,
280 tx: Tx<()>,
281 },
282 DeleteRecords {
283 repo_id: Uuid,
284 collections: Vec<Nsid>,
285 rkeys: Vec<Rkey>,
286 tx: Tx<()>,
287 },
288 DeleteAllRecords {
289 repo_id: Uuid,
290 tx: Tx<()>,
291 },
292 GetRecordCid {
293 repo_id: Uuid,
294 collection: Nsid,
295 rkey: Rkey,
296 tx: Tx<Option<CidLink>>,
297 },
298 ListRecords {
299 repo_id: Uuid,
300 collection: Nsid,
301 cursor: Option<Rkey>,
302 limit: i64,
303 reverse: bool,
304 rkey_start: Option<Rkey>,
305 rkey_end: Option<Rkey>,
306 tx: Tx<Vec<tranquil_db_traits::RecordInfo>>,
307 },
308 GetAllRecords {
309 repo_id: Uuid,
310 tx: Tx<Vec<tranquil_db_traits::FullRecordInfo>>,
311 },
312 ListCollections {
313 repo_id: Uuid,
314 tx: Tx<Vec<Nsid>>,
315 },
316 CountRecords {
317 repo_id: Uuid,
318 tx: Tx<i64>,
319 },
320 CountAllRecords {
321 tx: Tx<i64>,
322 },
323 GetRecordByCid {
324 cid: CidLink,
325 tx: Tx<Option<tranquil_db_traits::RecordWithTakedown>>,
326 },
327 SetRecordTakedown {
328 cid: CidLink,
329 takedown_ref: Option<String>,
330 scope_user: Option<Uuid>,
331 tx: Tx<()>,
332 },
333}
334
335impl RecordRequest {
336 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
337 match self {
338 Self::UpsertRecords { repo_id, .. }
339 | Self::DeleteRecords { repo_id, .. }
340 | Self::DeleteAllRecords { repo_id, .. }
341 | Self::GetRecordCid { repo_id, .. }
342 | Self::ListRecords { repo_id, .. }
343 | Self::GetAllRecords { repo_id, .. }
344 | Self::ListCollections { repo_id, .. }
345 | Self::CountRecords { repo_id, .. } => uuid_to_routing(user_hashes, repo_id),
346 Self::CountAllRecords { .. } | Self::GetRecordByCid { .. } => Routing::Global,
347 Self::SetRecordTakedown {
348 scope_user: Some(user_id),
349 ..
350 } => uuid_to_routing(user_hashes, user_id),
351 Self::SetRecordTakedown { .. } => Routing::Global,
352 }
353 }
354}
355
356pub enum UserBlockRequest {
357 InsertUserBlocks {
358 user_id: Uuid,
359 block_cids: Vec<Vec<u8>>,
360 repo_rev: String,
361 tx: Tx<()>,
362 },
363 DeleteUserBlocks {
364 user_id: Uuid,
365 block_cids: Vec<Vec<u8>>,
366 tx: Tx<()>,
367 },
368 GetUserBlockCidsSinceRev {
369 user_id: Uuid,
370 since_rev: String,
371 tx: Tx<Vec<Vec<u8>>>,
372 },
373 CountUserBlocks {
374 user_id: Uuid,
375 tx: Tx<i64>,
376 },
377}
378
379impl UserBlockRequest {
380 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
381 match self {
382 Self::InsertUserBlocks { user_id, .. }
383 | Self::DeleteUserBlocks { user_id, .. }
384 | Self::GetUserBlockCidsSinceRev { user_id, .. }
385 | Self::CountUserBlocks { user_id, .. } => uuid_to_routing(user_hashes, user_id),
386 }
387 }
388}
389
390pub enum EventRequest {
391 InsertCommitEvent {
392 data: CommitEventData,
393 tx: Tx<SequenceNumber>,
394 },
395 InsertIdentityEvent {
396 did: Did,
397 handle: Option<Handle>,
398 tx: Tx<SequenceNumber>,
399 },
400 InsertAccountEvent {
401 did: Did,
402 status: AccountStatus,
403 tx: Tx<SequenceNumber>,
404 },
405 InsertSyncEvent {
406 did: Did,
407 commit_cid: CidLink,
408 rev: Option<String>,
409 commit_bytes: Vec<u8>,
410 tx: Tx<SequenceNumber>,
411 },
412 InsertGenesisCommitEvent {
413 did: Did,
414 commit_cid: CidLink,
415 mst_root_cid: CidLink,
416 rev: String,
417 commit_bytes: Vec<u8>,
418 mst_root_bytes: Vec<u8>,
419 tx: Tx<SequenceNumber>,
420 },
421 PurgeDidEventsKeepingLatest {
422 did: Did,
423 tx: Tx<()>,
424 },
425 GetMaxSeq {
426 tx: Tx<SequenceNumber>,
427 },
428 GetMinSeqSince {
429 since: DateTime<Utc>,
430 tx: Tx<Option<SequenceNumber>>,
431 },
432 GetEventsSinceSeq {
433 since_seq: SequenceNumber,
434 limit: Option<i64>,
435 tx: Tx<Vec<SequencedEvent>>,
436 },
437 GetEventsInSeqRange {
438 start_seq: SequenceNumber,
439 end_seq: SequenceNumber,
440 tx: Tx<Vec<SequencedEvent>>,
441 },
442 GetEventBySeq {
443 seq: SequenceNumber,
444 tx: Tx<Option<SequencedEvent>>,
445 },
446 GetEventsSinceCursor {
447 cursor: SequenceNumber,
448 limit: i64,
449 tx: Tx<Vec<SequencedEvent>>,
450 },
451}
452
453impl EventRequest {
454 fn routing(&self) -> Routing {
455 match self {
456 Self::InsertCommitEvent { data, .. } => {
457 Routing::Sharded(UserHash::from_did(data.did.as_str()).raw())
458 }
459 Self::InsertIdentityEvent { did, .. }
460 | Self::InsertAccountEvent { did, .. }
461 | Self::InsertSyncEvent { did, .. }
462 | Self::InsertGenesisCommitEvent { did, .. }
463 | Self::PurgeDidEventsKeepingLatest { did, .. } => {
464 Routing::Sharded(UserHash::from_did(did.as_str()).raw())
465 }
466 Self::GetMaxSeq { .. }
467 | Self::GetMinSeqSince { .. }
468 | Self::GetEventsSinceSeq { .. }
469 | Self::GetEventsInSeqRange { .. }
470 | Self::GetEventBySeq { .. }
471 | Self::GetEventsSinceCursor { .. } => Routing::Global,
472 }
473 }
474}
475
476pub enum CommitRequest {
477 ApplyCommit {
478 input: Box<ApplyCommitInput>,
479 tx: oneshot::Sender<Result<ApplyCommitResult, ApplyCommitError>>,
480 },
481 ImportRepoData {
482 user_id: Uuid,
483 blocks: Vec<ImportBlock>,
484 records: Vec<ImportRecord>,
485 expected_root_cid: Option<CidLink>,
486 tx: oneshot::Sender<Result<(), ImportRepoError>>,
487 },
488 GetUsersWithoutBlocks {
489 tx: Tx<Vec<UserWithoutBlocks>>,
490 },
491 GetUsersNeedingRecordBlobsBackfill {
492 limit: i64,
493 tx: Tx<Vec<UserNeedingRecordBlobsBackfill>>,
494 },
495 InsertRecordBlobs {
496 repo_id: Uuid,
497 record_uris: Vec<AtUri>,
498 blob_cids: Vec<CidLink>,
499 tx: Tx<()>,
500 },
501}
502
503impl CommitRequest {
504 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
505 match self {
506 Self::ApplyCommit { input, .. } => did_to_routing(&input.did),
507 Self::ImportRepoData { user_id, .. }
508 | Self::InsertRecordBlobs {
509 repo_id: user_id, ..
510 } => uuid_to_routing(user_hashes, user_id),
511 Self::GetUsersWithoutBlocks { .. }
512 | Self::GetUsersNeedingRecordBlobsBackfill { .. } => Routing::Global,
513 }
514 }
515}
516
517pub enum BacklinkRequest {
518 GetBacklinkConflicts {
519 repo_id: Uuid,
520 collection: Nsid,
521 backlinks: Vec<Backlink>,
522 tx: Tx<Vec<AtUri>>,
523 },
524 AddBacklinks {
525 repo_id: Uuid,
526 backlinks: Vec<Backlink>,
527 tx: Tx<()>,
528 },
529 RemoveBacklinksByUri {
530 uri: AtUri,
531 tx: Tx<()>,
532 },
533 RemoveBacklinksByRepo {
534 repo_id: Uuid,
535 tx: Tx<()>,
536 },
537}
538
539impl BacklinkRequest {
540 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
541 match self {
542 Self::GetBacklinkConflicts { repo_id, .. }
543 | Self::AddBacklinks { repo_id, .. }
544 | Self::RemoveBacklinksByRepo { repo_id, .. } => uuid_to_routing(user_hashes, repo_id),
545 Self::RemoveBacklinksByUri { uri, .. } => match uri.did() {
546 Some(did) => did_to_routing(did),
547 None => Routing::Global,
548 },
549 }
550 }
551}
552
553pub enum BlobRequest {
554 InsertBlob {
555 cid: CidLink,
556 mime_type: String,
557 size_bytes: i64,
558 created_by_user: Uuid,
559 storage_key: String,
560 tx: Tx<Option<CidLink>>,
561 },
562 GetBlobMetadata {
563 cid: CidLink,
564 tx: Tx<Option<tranquil_db_traits::BlobMetadata>>,
565 },
566 GetBlobWithTakedown {
567 cid: CidLink,
568 tx: Tx<Option<tranquil_db_traits::BlobWithTakedown>>,
569 },
570 GetBlobStorageKey {
571 cid: CidLink,
572 tx: Tx<Option<String>>,
573 },
574 ListBlobsByUser {
575 user_id: Uuid,
576 cursor: Option<String>,
577 limit: i64,
578 tx: Tx<Vec<CidLink>>,
579 },
580 ListBlobsSinceRev {
581 did: Did,
582 since: String,
583 tx: Tx<Vec<CidLink>>,
584 },
585 CountBlobsByUser {
586 user_id: Uuid,
587 tx: Tx<i64>,
588 },
589 SumBlobStorage {
590 tx: Tx<i64>,
591 },
592 UpdateBlobTakedown {
593 cid: CidLink,
594 takedown_ref: Option<String>,
595 tx: Tx<bool>,
596 },
597 DeleteBlobByCid {
598 cid: CidLink,
599 tx: Tx<bool>,
600 },
601 DeleteBlobsByUser {
602 user_id: Uuid,
603 tx: Tx<u64>,
604 },
605 GetBlobStorageKeysByUser {
606 user_id: Uuid,
607 tx: Tx<Vec<String>>,
608 },
609 ListMissingBlobs {
610 repo_id: Uuid,
611 cursor: Option<String>,
612 limit: i64,
613 tx: Tx<Vec<tranquil_db_traits::MissingBlobInfo>>,
614 },
615 CountDistinctRecordBlobs {
616 repo_id: Uuid,
617 tx: Tx<i64>,
618 },
619 GetBlobsForExport {
620 repo_id: Uuid,
621 tx: Tx<Vec<tranquil_db_traits::BlobForExport>>,
622 },
623}
624
625impl BlobRequest {
626 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
627 match self {
628 Self::InsertBlob { cid, .. }
629 | Self::UpdateBlobTakedown { cid, .. }
630 | Self::DeleteBlobByCid { cid, .. } => cid_to_routing(cid),
631
632 Self::DeleteBlobsByUser { user_id, .. } => uuid_to_routing(user_hashes, user_id),
633
634 Self::GetBlobMetadata { .. }
635 | Self::GetBlobWithTakedown { .. }
636 | Self::GetBlobStorageKey { .. }
637 | Self::SumBlobStorage { .. } => Routing::Global,
638
639 Self::ListBlobsByUser { user_id, .. }
640 | Self::CountBlobsByUser { user_id, .. }
641 | Self::GetBlobStorageKeysByUser { user_id, .. } => {
642 uuid_to_routing(user_hashes, user_id)
643 }
644 Self::ListMissingBlobs { repo_id, .. }
645 | Self::CountDistinctRecordBlobs { repo_id, .. }
646 | Self::GetBlobsForExport { repo_id, .. } => uuid_to_routing(user_hashes, repo_id),
647 Self::ListBlobsSinceRev { did, .. } => did_to_routing(did),
648 }
649 }
650}
651
652pub enum DelegationRequest {
653 IsDelegatedAccount {
654 did: Did,
655 tx: Tx<bool>,
656 },
657 CreateDelegation {
658 delegated_did: Did,
659 controller_did: Did,
660 granted_scopes: DbScope,
661 granted_by: Did,
662 tx: Tx<Uuid>,
663 },
664 RevokeDelegation {
665 delegated_did: Did,
666 controller_did: Did,
667 revoked_by: Did,
668 tx: Tx<bool>,
669 },
670 UpdateDelegationScopes {
671 delegated_did: Did,
672 controller_did: Did,
673 new_scopes: DbScope,
674 tx: Tx<bool>,
675 },
676 GetDelegation {
677 delegated_did: Did,
678 controller_did: Did,
679 tx: Tx<Option<tranquil_db_traits::DelegationGrant>>,
680 },
681 GetDelegationsForAccount {
682 delegated_did: Did,
683 tx: Tx<Vec<tranquil_db_traits::ControllerInfo>>,
684 },
685 GetAccountsControlledBy {
686 controller_did: Did,
687 tx: Tx<Vec<tranquil_db_traits::DelegatedAccountInfo>>,
688 },
689 CountActiveControllers {
690 delegated_did: Did,
691 tx: Tx<i64>,
692 },
693 ControlsAnyAccounts {
694 did: Did,
695 tx: Tx<bool>,
696 },
697 LogDelegationAction {
698 delegated_did: Did,
699 actor_did: Did,
700 controller_did: Option<Did>,
701 action_type: DelegationActionType,
702 action_details: Option<serde_json::Value>,
703 ip_address: Option<String>,
704 user_agent: Option<String>,
705 tx: Tx<Uuid>,
706 },
707 GetAuditLogForAccount {
708 delegated_did: Did,
709 limit: i64,
710 offset: i64,
711 tx: Tx<Vec<tranquil_db_traits::AuditLogEntry>>,
712 },
713 CountAuditLogEntries {
714 delegated_did: Did,
715 tx: Tx<i64>,
716 },
717}
718
719impl DelegationRequest {
720 fn routing(&self) -> Routing {
721 match self {
722 Self::IsDelegatedAccount { did, .. }
723 | Self::GetDelegationsForAccount {
724 delegated_did: did, ..
725 }
726 | Self::CountActiveControllers {
727 delegated_did: did, ..
728 }
729 | Self::GetAuditLogForAccount {
730 delegated_did: did, ..
731 }
732 | Self::CountAuditLogEntries {
733 delegated_did: did, ..
734 } => did_to_routing(did),
735 Self::CreateDelegation { delegated_did, .. }
736 | Self::RevokeDelegation { delegated_did, .. }
737 | Self::UpdateDelegationScopes { delegated_did, .. }
738 | Self::GetDelegation { delegated_did, .. }
739 | Self::LogDelegationAction { delegated_did, .. } => did_to_routing(delegated_did),
740 Self::GetAccountsControlledBy { controller_did, .. }
741 | Self::ControlsAnyAccounts {
742 did: controller_did,
743 ..
744 } => did_to_routing(controller_did),
745 }
746 }
747}
748
749pub enum SsoRequest {
750 CreateExternalIdentity {
751 did: Did,
752 provider: tranquil_db_traits::SsoProviderType,
753 provider_user_id: String,
754 provider_username: Option<String>,
755 provider_email: Option<String>,
756 tx: Tx<Uuid>,
757 },
758 GetExternalIdentityByProvider {
759 provider: tranquil_db_traits::SsoProviderType,
760 provider_user_id: String,
761 tx: Tx<Option<tranquil_db_traits::ExternalIdentity>>,
762 },
763 GetExternalIdentitiesByDid {
764 did: Did,
765 tx: Tx<Vec<tranquil_db_traits::ExternalIdentity>>,
766 },
767 UpdateExternalIdentityLogin {
768 id: Uuid,
769 provider_username: Option<String>,
770 provider_email: Option<String>,
771 tx: Tx<()>,
772 },
773 DeleteExternalIdentity {
774 id: Uuid,
775 did: Did,
776 tx: Tx<bool>,
777 },
778 CreateSsoAuthState {
779 state: String,
780 request_uri: String,
781 provider: tranquil_db_traits::SsoProviderType,
782 action: tranquil_db_traits::SsoAction,
783 nonce: Option<String>,
784 code_verifier: Option<String>,
785 did: Option<Did>,
786 tx: Tx<()>,
787 },
788 ConsumeSsoAuthState {
789 state: String,
790 tx: Tx<Option<tranquil_db_traits::SsoAuthState>>,
791 },
792 CleanupExpiredSsoAuthStates {
793 tx: Tx<u64>,
794 },
795 CreatePendingRegistration {
796 token: String,
797 request_uri: String,
798 provider: tranquil_db_traits::SsoProviderType,
799 provider_user_id: String,
800 provider_username: Option<String>,
801 provider_email: Option<String>,
802 provider_email_verified: bool,
803 tx: Tx<()>,
804 },
805 GetPendingRegistration {
806 token: String,
807 tx: Tx<Option<tranquil_db_traits::SsoPendingRegistration>>,
808 },
809 ConsumePendingRegistration {
810 token: String,
811 tx: Tx<Option<tranquil_db_traits::SsoPendingRegistration>>,
812 },
813 CleanupExpiredPendingRegistrations {
814 tx: Tx<u64>,
815 },
816}
817
818impl SsoRequest {
819 fn routing(&self) -> Routing {
820 match self {
821 Self::CreateExternalIdentity { did, .. }
822 | Self::GetExternalIdentitiesByDid { did, .. }
823 | Self::DeleteExternalIdentity { did, .. } => did_to_routing(did),
824 Self::GetExternalIdentityByProvider { .. }
825 | Self::UpdateExternalIdentityLogin { .. }
826 | Self::ConsumeSsoAuthState { .. }
827 | Self::CreateSsoAuthState { .. }
828 | Self::CleanupExpiredSsoAuthStates { .. }
829 | Self::CreatePendingRegistration { .. }
830 | Self::GetPendingRegistration { .. }
831 | Self::ConsumePendingRegistration { .. }
832 | Self::CleanupExpiredPendingRegistrations { .. } => Routing::Global,
833 }
834 }
835}
836
837pub enum SessionRequest {
838 CreateSession {
839 data: tranquil_db_traits::SessionTokenCreate,
840 tx: Tx<SessionId>,
841 },
842 GetSessionByAccessJti {
843 access_jti: String,
844 tx: Tx<Option<tranquil_db_traits::SessionToken>>,
845 },
846 GetSessionForRefresh {
847 refresh_jti: String,
848 tx: Tx<Option<tranquil_db_traits::SessionForRefresh>>,
849 },
850 DeleteSessionByAccessJti {
851 access_jti: String,
852 did: Did,
853 tx: Tx<u64>,
854 },
855 DeleteSessionById {
856 session_id: SessionId,
857 did: Did,
858 tx: Tx<u64>,
859 },
860 DeleteSessionsByDid {
861 did: Did,
862 tx: Tx<u64>,
863 },
864 DeleteSessionsByDidExceptJti {
865 did: Did,
866 except_jti: String,
867 tx: Tx<u64>,
868 },
869 ListSessionsByDid {
870 did: Did,
871 tx: Tx<Vec<tranquil_db_traits::SessionListItem>>,
872 },
873 GetSessionAccessJtiById {
874 session_id: SessionId,
875 did: Did,
876 tx: Tx<Option<String>>,
877 },
878 DeleteSessionsByAppPassword {
879 did: Did,
880 app_password_name: String,
881 tx: Tx<u64>,
882 },
883 GetSessionJtisByAppPassword {
884 did: Did,
885 app_password_name: String,
886 tx: Tx<Vec<String>>,
887 },
888 LookupRefreshGrace {
889 refresh_jti: String,
890 tx: Tx<tranquil_db_traits::RefreshGraceLookup>,
891 },
892 ListAppPasswords {
893 user_id: Uuid,
894 tx: Tx<Vec<tranquil_db_traits::AppPasswordRecord>>,
895 },
896 GetAppPasswordsForLogin {
897 user_id: Uuid,
898 tx: Tx<Vec<tranquil_db_traits::AppPasswordRecord>>,
899 },
900 GetAppPasswordByName {
901 user_id: Uuid,
902 name: String,
903 tx: Tx<Option<tranquil_db_traits::AppPasswordRecord>>,
904 },
905 CreateAppPassword {
906 data: tranquil_db_traits::AppPasswordCreate,
907 tx: Tx<Uuid>,
908 },
909 DeleteAppPassword {
910 user_id: Uuid,
911 name: String,
912 tx: Tx<u64>,
913 },
914 DeleteAppPasswordsByController {
915 did: Did,
916 controller_did: Did,
917 tx: Tx<u64>,
918 },
919 GetLastReauthAt {
920 did: Did,
921 tx: Tx<Option<DateTime<Utc>>>,
922 },
923 UpdateLastReauth {
924 did: Did,
925 tx: Tx<DateTime<Utc>>,
926 },
927 GetSessionMfaStatus {
928 did: Did,
929 tx: Tx<Option<tranquil_db_traits::SessionMfaStatus>>,
930 },
931 UpdateMfaVerified {
932 did: Did,
933 tx: Tx<()>,
934 },
935 GetAppPasswordHashesByDid {
936 did: Did,
937 tx: Tx<Vec<String>>,
938 },
939 RefreshSessionAtomic {
940 data: tranquil_db_traits::SessionRefreshData,
941 tx: Tx<RefreshSessionResult>,
942 },
943}
944
945impl SessionRequest {
946 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
947 match self {
948 Self::CreateSession { .. }
949 | Self::GetSessionByAccessJti { .. }
950 | Self::GetSessionForRefresh { .. }
951 | Self::LookupRefreshGrace { .. } => Routing::Global,
952 Self::DeleteSessionByAccessJti { did, .. }
953 | Self::DeleteSessionById { did, .. }
954 | Self::DeleteSessionsByDid { did, .. }
955 | Self::DeleteSessionsByDidExceptJti { did, .. }
956 | Self::ListSessionsByDid { did, .. }
957 | Self::GetSessionAccessJtiById { did, .. }
958 | Self::DeleteSessionsByAppPassword { did, .. }
959 | Self::GetSessionJtisByAppPassword { did, .. }
960 | Self::DeleteAppPasswordsByController { did, .. }
961 | Self::GetLastReauthAt { did, .. }
962 | Self::UpdateLastReauth { did, .. }
963 | Self::GetSessionMfaStatus { did, .. }
964 | Self::UpdateMfaVerified { did, .. }
965 | Self::GetAppPasswordHashesByDid { did, .. } => did_to_routing(did),
966 Self::RefreshSessionAtomic { data, .. } => did_to_routing(data.did.as_str()),
967 Self::ListAppPasswords { user_id, .. }
968 | Self::GetAppPasswordsForLogin { user_id, .. }
969 | Self::GetAppPasswordByName { user_id, .. }
970 | Self::CreateAppPassword {
971 data: tranquil_db_traits::AppPasswordCreate { user_id, .. },
972 ..
973 }
974 | Self::DeleteAppPassword { user_id, .. } => uuid_to_routing(user_hashes, user_id),
975 }
976 }
977}
978
979pub enum UserRequest {
980 GetByDid {
981 did: Did,
982 tx: Tx<Option<UserRow>>,
983 },
984 GetByHandle {
985 handle: Handle,
986 tx: Tx<Option<UserRow>>,
987 },
988 GetWithKeyByDid {
989 did: Did,
990 tx: Tx<Option<UserWithKey>>,
991 },
992 GetStatusByDid {
993 did: Did,
994 tx: Tx<Option<UserStatus>>,
995 },
996 CountUsers {
997 tx: Tx<i64>,
998 },
999 GetSessionAccessExpiry {
1000 did: Did,
1001 access_jti: String,
1002 tx: Tx<Option<DateTime<Utc>>>,
1003 },
1004 GetOAuthTokenWithUser {
1005 token_id: String,
1006 tx: Tx<Option<OAuthTokenWithUser>>,
1007 },
1008 GetUserInfoByDid {
1009 did: Did,
1010 tx: Tx<Option<UserInfoForAuth>>,
1011 },
1012 GetAnyAdminUserId {
1013 tx: Tx<Option<Uuid>>,
1014 },
1015 SetInvitesDisabled {
1016 did: Did,
1017 disabled: bool,
1018 tx: Tx<bool>,
1019 },
1020 SearchAccounts {
1021 cursor_did: Option<Did>,
1022 email_filter: Option<String>,
1023 handle_filter: Option<String>,
1024 limit: i64,
1025 tx: Tx<Vec<AccountSearchResult>>,
1026 },
1027 GetAuthInfoByDid {
1028 did: Did,
1029 tx: Tx<Option<UserAuthInfo>>,
1030 },
1031 GetByEmail {
1032 email: String,
1033 tx: Tx<Option<UserForVerification>>,
1034 },
1035 GetLoginCheckByIdentifier {
1036 identifier: String,
1037 tx: Tx<Option<UserLoginCheck>>,
1038 },
1039 GetLoginInfoByIdentifier {
1040 identifier: String,
1041 tx: Tx<Option<UserLoginInfo>>,
1042 },
1043 Get2faStatusByDid {
1044 did: Did,
1045 tx: Tx<Option<User2faStatus>>,
1046 },
1047 GetCommsPrefs {
1048 user_id: Uuid,
1049 tx: Tx<Option<UserCommsPrefs>>,
1050 },
1051 GetIdByDid {
1052 did: Did,
1053 tx: Tx<Option<Uuid>>,
1054 },
1055 GetUserKeyById {
1056 user_id: Uuid,
1057 tx: Tx<Option<UserKeyInfo>>,
1058 },
1059 GetIdAndHandleByDid {
1060 did: Did,
1061 tx: Tx<Option<UserIdAndHandle>>,
1062 },
1063 GetDidWebInfoByHandle {
1064 handle: Handle,
1065 tx: Tx<Option<UserDidWebInfo>>,
1066 },
1067 GetDidWebOverrides {
1068 user_id: Uuid,
1069 tx: Tx<Option<DidWebOverrides>>,
1070 },
1071 GetHandleByDid {
1072 did: Did,
1073 tx: Tx<Option<Handle>>,
1074 },
1075 IsAccountActiveByDid {
1076 did: Did,
1077 tx: Tx<Option<bool>>,
1078 },
1079 GetUserForDeletion {
1080 did: Did,
1081 tx: Tx<Option<UserForDeletion>>,
1082 },
1083 CheckHandleExists {
1084 handle: Handle,
1085 exclude_user_id: Uuid,
1086 tx: Tx<bool>,
1087 },
1088 UpdateHandle {
1089 user_id: Uuid,
1090 handle: Handle,
1091 tx: Tx<()>,
1092 },
1093 GetUserWithKeyByDid {
1094 did: Did,
1095 tx: Tx<Option<UserKeyWithId>>,
1096 },
1097 IsAccountMigrated {
1098 did: Did,
1099 tx: Tx<bool>,
1100 },
1101 HasVerifiedCommsChannel {
1102 did: Did,
1103 tx: Tx<bool>,
1104 },
1105 GetIdByHandle {
1106 handle: Handle,
1107 tx: Tx<Option<Uuid>>,
1108 },
1109 GetEmailInfoByDid {
1110 did: Did,
1111 tx: Tx<Option<UserEmailInfo>>,
1112 },
1113 CheckEmailExists {
1114 email: String,
1115 exclude_user_id: Uuid,
1116 tx: Tx<bool>,
1117 },
1118 UpdateEmail {
1119 user_id: Uuid,
1120 email: String,
1121 tx: Tx<()>,
1122 },
1123 SetEmailVerified {
1124 user_id: Uuid,
1125 verified: bool,
1126 tx: Tx<()>,
1127 },
1128 CheckEmailVerifiedByIdentifier {
1129 identifier: String,
1130 tx: Tx<Option<bool>>,
1131 },
1132 CheckChannelVerifiedByDid {
1133 did: Did,
1134 channel: CommsChannel,
1135 tx: Tx<Option<bool>>,
1136 },
1137 AdminUpdateEmail {
1138 did: Did,
1139 email: String,
1140 tx: Tx<u64>,
1141 },
1142 AdminUpdateHandle {
1143 did: Did,
1144 handle: Handle,
1145 tx: Tx<u64>,
1146 },
1147 AdminUpdatePassword {
1148 did: Did,
1149 password_hash: String,
1150 tx: Tx<u64>,
1151 },
1152 SetAdminStatus {
1153 did: Did,
1154 is_admin: bool,
1155 tx: Tx<()>,
1156 },
1157 GetNotificationPrefs {
1158 did: Did,
1159 tx: Tx<Option<NotificationPrefs>>,
1160 },
1161 GetIdHandleEmailByDid {
1162 did: Did,
1163 tx: Tx<Option<UserIdHandleEmail>>,
1164 },
1165 UpdatePreferredCommsChannel {
1166 did: Did,
1167 channel: CommsChannel,
1168 tx: Tx<()>,
1169 },
1170 ClearDiscord {
1171 user_id: Uuid,
1172 tx: Tx<()>,
1173 },
1174 ClearTelegram {
1175 user_id: Uuid,
1176 tx: Tx<()>,
1177 },
1178 ClearSignal {
1179 user_id: Uuid,
1180 tx: Tx<()>,
1181 },
1182 SetUnverifiedSignal {
1183 user_id: Uuid,
1184 signal_username: String,
1185 tx: Tx<()>,
1186 },
1187 SetUnverifiedTelegram {
1188 user_id: Uuid,
1189 telegram_username: String,
1190 tx: Tx<()>,
1191 },
1192 StoreTelegramChatId {
1193 telegram_username: String,
1194 chat_id: i64,
1195 handle: Option<String>,
1196 tx: Tx<Option<Uuid>>,
1197 },
1198 GetTelegramChatId {
1199 user_id: Uuid,
1200 tx: Tx<Option<i64>>,
1201 },
1202 SetUnverifiedDiscord {
1203 user_id: Uuid,
1204 discord_username: String,
1205 tx: Tx<()>,
1206 },
1207 StoreDiscordUserId {
1208 discord_username: String,
1209 discord_id: String,
1210 handle: Option<String>,
1211 tx: Tx<Option<Uuid>>,
1212 },
1213 GetVerificationInfo {
1214 did: Did,
1215 tx: Tx<Option<UserVerificationInfo>>,
1216 },
1217 VerifyEmailChannel {
1218 user_id: Uuid,
1219 email: String,
1220 tx: Tx<bool>,
1221 },
1222 VerifyDiscordChannel {
1223 user_id: Uuid,
1224 discord_id: String,
1225 tx: Tx<()>,
1226 },
1227 VerifyTelegramChannel {
1228 user_id: Uuid,
1229 telegram_username: String,
1230 tx: Tx<()>,
1231 },
1232 VerifySignalChannel {
1233 user_id: Uuid,
1234 signal_username: String,
1235 tx: Tx<()>,
1236 },
1237 SetEmailVerifiedFlag {
1238 user_id: Uuid,
1239 tx: Tx<()>,
1240 },
1241 SetDiscordVerifiedFlag {
1242 user_id: Uuid,
1243 tx: Tx<()>,
1244 },
1245 SetTelegramVerifiedFlag {
1246 user_id: Uuid,
1247 tx: Tx<()>,
1248 },
1249 SetSignalVerifiedFlag {
1250 user_id: Uuid,
1251 tx: Tx<()>,
1252 },
1253 HasTotpEnabled {
1254 did: Did,
1255 tx: Tx<bool>,
1256 },
1257 HasPasskeys {
1258 did: Did,
1259 tx: Tx<bool>,
1260 },
1261 GetPasswordHashByDid {
1262 did: Did,
1263 tx: Tx<Option<String>>,
1264 },
1265 GetPasskeysForUser {
1266 did: Did,
1267 tx: Tx<Vec<StoredPasskey>>,
1268 },
1269 GetPasskeyByCredentialId {
1270 credential_id: Vec<u8>,
1271 tx: Tx<Option<StoredPasskey>>,
1272 },
1273 SavePasskey {
1274 did: Did,
1275 credential_id: Vec<u8>,
1276 public_key: Vec<u8>,
1277 friendly_name: Option<String>,
1278 tx: Tx<Uuid>,
1279 },
1280 UpdatePasskeyCounter {
1281 credential_id: Vec<u8>,
1282 new_counter: i32,
1283 tx: Tx<bool>,
1284 },
1285 DeletePasskey {
1286 id: Uuid,
1287 did: Did,
1288 tx: Tx<bool>,
1289 },
1290 UpdatePasskeyName {
1291 id: Uuid,
1292 did: Did,
1293 name: String,
1294 tx: Tx<bool>,
1295 },
1296 SaveWebauthnChallenge {
1297 did: Did,
1298 challenge_type: WebauthnChallengeType,
1299 state_json: String,
1300 tx: Tx<Uuid>,
1301 },
1302 LoadWebauthnChallenge {
1303 did: Did,
1304 challenge_type: WebauthnChallengeType,
1305 tx: Tx<Option<String>>,
1306 },
1307 DeleteWebauthnChallenge {
1308 did: Did,
1309 challenge_type: WebauthnChallengeType,
1310 tx: Tx<()>,
1311 },
1312 SaveDiscoverableChallenge {
1313 request_key: String,
1314 state_json: String,
1315 tx: Tx<Uuid>,
1316 },
1317 LoadDiscoverableChallenge {
1318 request_key: String,
1319 tx: Tx<Option<String>>,
1320 },
1321 DeleteDiscoverableChallenge {
1322 request_key: String,
1323 tx: Tx<()>,
1324 },
1325 GetTotpRecord {
1326 did: Did,
1327 tx: Tx<Option<TotpRecord>>,
1328 },
1329 GetTotpRecordState {
1330 did: Did,
1331 tx: Tx<Option<TotpRecordState>>,
1332 },
1333 UpsertTotpSecret {
1334 did: Did,
1335 secret_encrypted: Vec<u8>,
1336 encryption_version: i32,
1337 tx: Tx<()>,
1338 },
1339 SetTotpVerified {
1340 did: Did,
1341 tx: Tx<()>,
1342 },
1343 UpdateTotpLastUsed {
1344 did: Did,
1345 tx: Tx<()>,
1346 },
1347 DeleteTotp {
1348 did: Did,
1349 tx: Tx<()>,
1350 },
1351 GetUnusedBackupCodes {
1352 did: Did,
1353 tx: Tx<Vec<StoredBackupCode>>,
1354 },
1355 MarkBackupCodeUsed {
1356 code_id: Uuid,
1357 tx: Tx<bool>,
1358 },
1359 CountUnusedBackupCodes {
1360 did: Did,
1361 tx: Tx<i64>,
1362 },
1363 DeleteBackupCodes {
1364 did: Did,
1365 tx: Tx<u64>,
1366 },
1367 InsertBackupCodes {
1368 did: Did,
1369 code_hashes: Vec<String>,
1370 tx: Tx<()>,
1371 },
1372 EnableTotpWithBackupCodes {
1373 did: Did,
1374 code_hashes: Vec<String>,
1375 tx: Tx<()>,
1376 },
1377 DeleteTotpAndBackupCodes {
1378 did: Did,
1379 tx: Tx<()>,
1380 },
1381 ReplaceBackupCodes {
1382 did: Did,
1383 code_hashes: Vec<String>,
1384 tx: Tx<()>,
1385 },
1386 GetSessionInfoByDid {
1387 did: Did,
1388 tx: Tx<Option<UserSessionInfo>>,
1389 },
1390 GetLegacyLoginPref {
1391 did: Did,
1392 tx: Tx<Option<UserLegacyLoginPref>>,
1393 },
1394 UpdateLegacyLogin {
1395 did: Did,
1396 allow: bool,
1397 tx: Tx<bool>,
1398 },
1399 UpdateLocale {
1400 did: Did,
1401 locale: String,
1402 tx: Tx<bool>,
1403 },
1404 GetLoginFullByIdentifier {
1405 identifier: String,
1406 tx: Tx<Option<UserLoginFull>>,
1407 },
1408 GetConfirmSignupByDid {
1409 did: Did,
1410 tx: Tx<Option<UserConfirmSignup>>,
1411 },
1412 GetResendVerificationByDid {
1413 did: Did,
1414 tx: Tx<Option<UserResendVerification>>,
1415 },
1416 SetChannelVerified {
1417 did: Did,
1418 channel: CommsChannel,
1419 tx: Tx<()>,
1420 },
1421 GetIdByEmailOrHandle {
1422 email: String,
1423 handle: String,
1424 tx: Tx<Option<Uuid>>,
1425 },
1426 CountAccountsByEmail {
1427 email: String,
1428 tx: Tx<i64>,
1429 },
1430 GetHandlesByEmail {
1431 email: String,
1432 tx: Tx<Vec<Handle>>,
1433 },
1434 SetPasswordResetCode {
1435 user_id: Uuid,
1436 code: String,
1437 expires_at: DateTime<Utc>,
1438 tx: Tx<()>,
1439 },
1440 GetUserByResetCode {
1441 code: String,
1442 tx: Tx<Option<UserResetCodeInfo>>,
1443 },
1444 ClearPasswordResetCode {
1445 user_id: Uuid,
1446 tx: Tx<()>,
1447 },
1448 GetIdAndPasswordHashByDid {
1449 did: Did,
1450 tx: Tx<Option<UserIdAndPasswordHash>>,
1451 },
1452 UpdatePasswordHash {
1453 user_id: Uuid,
1454 password_hash: String,
1455 tx: Tx<()>,
1456 },
1457 ResetPasswordWithSessions {
1458 user_id: Uuid,
1459 password_hash: String,
1460 tx: Tx<PasswordResetResult>,
1461 },
1462 ActivateAccount {
1463 did: Did,
1464 tx: Tx<bool>,
1465 },
1466 DeactivateAccount {
1467 did: Did,
1468 delete_after: Option<DateTime<Utc>>,
1469 tx: Tx<bool>,
1470 },
1471 HasPasswordByDid {
1472 did: Did,
1473 tx: Tx<Option<bool>>,
1474 },
1475 GetPasswordInfoByDid {
1476 did: Did,
1477 tx: Tx<Option<UserPasswordInfo>>,
1478 },
1479 RemoveUserPassword {
1480 user_id: Uuid,
1481 tx: Tx<()>,
1482 },
1483 SetNewUserPassword {
1484 user_id: Uuid,
1485 password_hash: String,
1486 tx: Tx<()>,
1487 },
1488 GetUserKeyByDid {
1489 did: Did,
1490 tx: Tx<Option<UserKeyInfo>>,
1491 },
1492 DeleteAccountComplete {
1493 user_id: Uuid,
1494 did: Did,
1495 tx: Tx<()>,
1496 },
1497 SetUserTakedown {
1498 did: Did,
1499 takedown_ref: Option<String>,
1500 tx: Tx<bool>,
1501 },
1502 AdminDeleteAccountComplete {
1503 user_id: Uuid,
1504 did: Did,
1505 tx: Tx<()>,
1506 },
1507 GetUserForDidDoc {
1508 did: Did,
1509 tx: Tx<Option<UserForDidDoc>>,
1510 },
1511 GetUserForDidDocBuild {
1512 did: Did,
1513 tx: Tx<Option<UserForDidDocBuild>>,
1514 },
1515 UpsertDidWebOverrides {
1516 user_id: Uuid,
1517 verification_methods: Option<serde_json::Value>,
1518 also_known_as: Option<Vec<String>>,
1519 tx: Tx<()>,
1520 },
1521 UpdateMigratedToPds {
1522 did: Did,
1523 endpoint: String,
1524 tx: Tx<()>,
1525 },
1526 GetUserForPasskeySetup {
1527 did: Did,
1528 tx: Tx<Option<UserForPasskeySetup>>,
1529 },
1530 GetUserForPasskeyRecovery {
1531 identifier: String,
1532 normalized_handle: String,
1533 tx: Tx<Option<UserForPasskeyRecovery>>,
1534 },
1535 SetRecoveryToken {
1536 did: Did,
1537 token_hash: String,
1538 expires_at: DateTime<Utc>,
1539 tx: Tx<()>,
1540 },
1541 GetUserForRecovery {
1542 did: Did,
1543 tx: Tx<Option<UserForRecovery>>,
1544 },
1545 GetAccountsScheduledForDeletion {
1546 limit: i64,
1547 tx: Tx<Vec<ScheduledDeletionAccount>>,
1548 },
1549 DeleteAccountWithFirehose {
1550 user_id: Uuid,
1551 did: Did,
1552 tx: Tx<()>,
1553 },
1554 CreatePasswordAccount {
1555 input: CreatePasswordAccountInput,
1556 tx: oneshot::Sender<Result<CreatePasswordAccountResult, CreateAccountError>>,
1557 },
1558 CreateDelegatedAccount {
1559 input: CreateDelegatedAccountInput,
1560 tx: oneshot::Sender<Result<Uuid, CreateAccountError>>,
1561 },
1562 CreatePasskeyAccount {
1563 input: CreatePasskeyAccountInput,
1564 tx: oneshot::Sender<Result<CreatePasswordAccountResult, CreateAccountError>>,
1565 },
1566 CreateSsoAccount {
1567 input: CreateSsoAccountInput,
1568 tx: oneshot::Sender<Result<CreatePasswordAccountResult, CreateAccountError>>,
1569 },
1570 ReactivateMigrationAccount {
1571 input: MigrationReactivationInput,
1572 tx: oneshot::Sender<Result<ReactivatedAccountInfo, MigrationReactivationError>>,
1573 },
1574 CheckHandleAvailableForNewAccount {
1575 handle: Handle,
1576 tx: Tx<bool>,
1577 },
1578 ReserveHandle {
1579 handle: Handle,
1580 reserved_by: String,
1581 tx: Tx<bool>,
1582 },
1583 ReleaseHandleReservation {
1584 handle: Handle,
1585 tx: Tx<()>,
1586 },
1587 CleanupExpiredHandleReservations {
1588 tx: Tx<u64>,
1589 },
1590 CompletePasskeySetup {
1591 input: CompletePasskeySetupInput,
1592 tx: Tx<()>,
1593 },
1594 RecoverPasskeyAccount {
1595 input: RecoverPasskeyAccountInput,
1596 tx: Tx<RecoverPasskeyAccountResult>,
1597 },
1598 GetPasswordResetInfo {
1599 email: String,
1600 tx: Tx<Option<tranquil_db_traits::PasswordResetInfo>>,
1601 },
1602 EnableTotpVerified {
1603 did: Did,
1604 encrypted_secret: Vec<u8>,
1605 tx: Tx<()>,
1606 },
1607 SetTwoFactorEnabled {
1608 did: Did,
1609 enabled: bool,
1610 tx: Tx<()>,
1611 },
1612 ExpirePasswordResetCode {
1613 email: String,
1614 tx: Tx<()>,
1615 },
1616}
1617
1618impl UserRequest {
1619 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
1620 match self {
1621 Self::GetByDid { did, .. }
1622 | Self::GetWithKeyByDid { did, .. }
1623 | Self::GetStatusByDid { did, .. }
1624 | Self::GetSessionAccessExpiry { did, .. }
1625 | Self::GetUserInfoByDid { did, .. }
1626 | Self::SetInvitesDisabled { did, .. }
1627 | Self::GetAuthInfoByDid { did, .. }
1628 | Self::Get2faStatusByDid { did, .. }
1629 | Self::GetIdByDid { did, .. }
1630 | Self::GetIdAndHandleByDid { did, .. }
1631 | Self::GetHandleByDid { did, .. }
1632 | Self::IsAccountActiveByDid { did, .. }
1633 | Self::GetUserForDeletion { did, .. }
1634 | Self::GetUserWithKeyByDid { did, .. }
1635 | Self::IsAccountMigrated { did, .. }
1636 | Self::HasVerifiedCommsChannel { did, .. }
1637 | Self::GetEmailInfoByDid { did, .. }
1638 | Self::CheckChannelVerifiedByDid { did, .. }
1639 | Self::AdminUpdateEmail { did, .. }
1640 | Self::AdminUpdateHandle { did, .. }
1641 | Self::AdminUpdatePassword { did, .. }
1642 | Self::SetAdminStatus { did, .. }
1643 | Self::GetNotificationPrefs { did, .. }
1644 | Self::GetIdHandleEmailByDid { did, .. }
1645 | Self::UpdatePreferredCommsChannel { did, .. }
1646 | Self::GetVerificationInfo { did, .. }
1647 | Self::HasTotpEnabled { did, .. }
1648 | Self::HasPasskeys { did, .. }
1649 | Self::GetPasswordHashByDid { did, .. }
1650 | Self::GetPasskeysForUser { did, .. }
1651 | Self::SavePasskey { did, .. }
1652 | Self::DeletePasskey { did, .. }
1653 | Self::UpdatePasskeyName { did, .. }
1654 | Self::SaveWebauthnChallenge { did, .. }
1655 | Self::LoadWebauthnChallenge { did, .. }
1656 | Self::DeleteWebauthnChallenge { did, .. }
1657 | Self::GetTotpRecord { did, .. }
1658 | Self::GetTotpRecordState { did, .. }
1659 | Self::UpsertTotpSecret { did, .. }
1660 | Self::SetTotpVerified { did, .. }
1661 | Self::UpdateTotpLastUsed { did, .. }
1662 | Self::DeleteTotp { did, .. }
1663 | Self::GetUnusedBackupCodes { did, .. }
1664 | Self::CountUnusedBackupCodes { did, .. }
1665 | Self::DeleteBackupCodes { did, .. }
1666 | Self::InsertBackupCodes { did, .. }
1667 | Self::EnableTotpWithBackupCodes { did, .. }
1668 | Self::DeleteTotpAndBackupCodes { did, .. }
1669 | Self::ReplaceBackupCodes { did, .. }
1670 | Self::GetSessionInfoByDid { did, .. }
1671 | Self::GetLegacyLoginPref { did, .. }
1672 | Self::UpdateLegacyLogin { did, .. }
1673 | Self::UpdateLocale { did, .. }
1674 | Self::GetConfirmSignupByDid { did, .. }
1675 | Self::GetResendVerificationByDid { did, .. }
1676 | Self::SetChannelVerified { did, .. }
1677 | Self::GetIdAndPasswordHashByDid { did, .. }
1678 | Self::ActivateAccount { did, .. }
1679 | Self::DeactivateAccount { did, .. }
1680 | Self::HasPasswordByDid { did, .. }
1681 | Self::GetPasswordInfoByDid { did, .. }
1682 | Self::GetUserKeyByDid { did, .. }
1683 | Self::DeleteAccountComplete { did, .. }
1684 | Self::SetUserTakedown { did, .. }
1685 | Self::AdminDeleteAccountComplete { did, .. }
1686 | Self::GetUserForDidDoc { did, .. }
1687 | Self::GetUserForDidDocBuild { did, .. }
1688 | Self::UpdateMigratedToPds { did, .. }
1689 | Self::GetUserForPasskeySetup { did, .. }
1690 | Self::GetUserForRecovery { did, .. }
1691 | Self::DeleteAccountWithFirehose { did, .. }
1692 | Self::CreatePasswordAccount {
1693 input: CreatePasswordAccountInput { did, .. },
1694 ..
1695 }
1696 | Self::CreateDelegatedAccount {
1697 input: CreateDelegatedAccountInput { did, .. },
1698 ..
1699 }
1700 | Self::CreatePasskeyAccount {
1701 input: CreatePasskeyAccountInput { did, .. },
1702 ..
1703 }
1704 | Self::CreateSsoAccount {
1705 input: CreateSsoAccountInput { did, .. },
1706 ..
1707 }
1708 | Self::ReactivateMigrationAccount {
1709 input: MigrationReactivationInput { did, .. },
1710 ..
1711 }
1712 | Self::CompletePasskeySetup {
1713 input: CompletePasskeySetupInput { did, .. },
1714 ..
1715 }
1716 | Self::RecoverPasskeyAccount {
1717 input: RecoverPasskeyAccountInput { did, .. },
1718 ..
1719 }
1720 | Self::SetRecoveryToken { did, .. }
1721 | Self::EnableTotpVerified { did, .. }
1722 | Self::SetTwoFactorEnabled { did, .. } => did_to_routing(did),
1723
1724 Self::GetCommsPrefs { user_id, .. }
1725 | Self::GetUserKeyById { user_id, .. }
1726 | Self::GetDidWebOverrides { user_id, .. }
1727 | Self::UpdateHandle { user_id, .. }
1728 | Self::UpdateEmail { user_id, .. }
1729 | Self::SetEmailVerified { user_id, .. }
1730 | Self::ClearDiscord { user_id, .. }
1731 | Self::ClearTelegram { user_id, .. }
1732 | Self::ClearSignal { user_id, .. }
1733 | Self::SetUnverifiedSignal { user_id, .. }
1734 | Self::SetUnverifiedTelegram { user_id, .. }
1735 | Self::GetTelegramChatId { user_id, .. }
1736 | Self::SetUnverifiedDiscord { user_id, .. }
1737 | Self::VerifyEmailChannel { user_id, .. }
1738 | Self::VerifyDiscordChannel { user_id, .. }
1739 | Self::VerifyTelegramChannel { user_id, .. }
1740 | Self::VerifySignalChannel { user_id, .. }
1741 | Self::SetEmailVerifiedFlag { user_id, .. }
1742 | Self::SetDiscordVerifiedFlag { user_id, .. }
1743 | Self::SetTelegramVerifiedFlag { user_id, .. }
1744 | Self::SetSignalVerifiedFlag { user_id, .. }
1745 | Self::SetPasswordResetCode { user_id, .. }
1746 | Self::ClearPasswordResetCode { user_id, .. }
1747 | Self::UpdatePasswordHash { user_id, .. }
1748 | Self::ResetPasswordWithSessions { user_id, .. }
1749 | Self::RemoveUserPassword { user_id, .. }
1750 | Self::SetNewUserPassword { user_id, .. }
1751 | Self::UpsertDidWebOverrides { user_id, .. } => uuid_to_routing(user_hashes, user_id),
1752
1753 Self::CheckHandleExists {
1754 exclude_user_id, ..
1755 }
1756 | Self::CheckEmailExists {
1757 exclude_user_id, ..
1758 } => uuid_to_routing(user_hashes, exclude_user_id),
1759
1760 Self::MarkBackupCodeUsed { .. } => Routing::Global,
1761
1762 Self::GetByHandle { .. }
1763 | Self::GetDidWebInfoByHandle { .. }
1764 | Self::GetIdByHandle { .. }
1765 | Self::CheckHandleAvailableForNewAccount { .. }
1766 | Self::ReserveHandle { .. }
1767 | Self::ReleaseHandleReservation { .. } => Routing::Global,
1768
1769 Self::CountUsers { .. }
1770 | Self::GetOAuthTokenWithUser { .. }
1771 | Self::GetAnyAdminUserId { .. }
1772 | Self::SearchAccounts { .. }
1773 | Self::GetByEmail { .. }
1774 | Self::GetLoginCheckByIdentifier { .. }
1775 | Self::GetLoginInfoByIdentifier { .. }
1776 | Self::CheckEmailVerifiedByIdentifier { .. }
1777 | Self::StoreTelegramChatId { .. }
1778 | Self::StoreDiscordUserId { .. }
1779 | Self::GetPasskeyByCredentialId { .. }
1780 | Self::UpdatePasskeyCounter { .. }
1781 | Self::GetLoginFullByIdentifier { .. }
1782 | Self::GetIdByEmailOrHandle { .. }
1783 | Self::CountAccountsByEmail { .. }
1784 | Self::GetHandlesByEmail { .. }
1785 | Self::GetUserByResetCode { .. }
1786 | Self::GetUserForPasskeyRecovery { .. }
1787 | Self::GetAccountsScheduledForDeletion { .. }
1788 | Self::CleanupExpiredHandleReservations { .. }
1789 | Self::GetPasswordResetInfo { .. }
1790 | Self::ExpirePasswordResetCode { .. }
1791 | Self::SaveDiscoverableChallenge { .. }
1792 | Self::LoadDiscoverableChallenge { .. }
1793 | Self::DeleteDiscoverableChallenge { .. } => Routing::Global,
1794 }
1795 }
1796}
1797
1798pub enum InfraRequest {
1799 EnqueueComms {
1800 user_id: Option<Uuid>,
1801 channel: CommsChannel,
1802 comms_type: CommsType,
1803 recipient: String,
1804 subject: Option<String>,
1805 body: String,
1806 metadata: Option<serde_json::Value>,
1807 tx: Tx<Uuid>,
1808 },
1809 FetchPendingComms {
1810 now: DateTime<Utc>,
1811 batch_size: i64,
1812 tx: Tx<Vec<QueuedComms>>,
1813 },
1814 MarkCommsSent {
1815 id: Uuid,
1816 tx: Tx<()>,
1817 },
1818 MarkCommsFailed {
1819 id: Uuid,
1820 error: String,
1821 tx: Tx<()>,
1822 },
1823 MarkCommsFailedPermanent {
1824 id: Uuid,
1825 error: String,
1826 tx: Tx<()>,
1827 },
1828 CreateInviteCode {
1829 code: InviteCode,
1830 use_count: i32,
1831 for_account: Option<Did>,
1832 tx: Tx<bool>,
1833 },
1834 CreateInviteCodesBatch {
1835 codes: Vec<InviteCode>,
1836 use_count: i32,
1837 created_by_user: Uuid,
1838 for_account: Option<Did>,
1839 tx: Tx<()>,
1840 },
1841 GetInviteCodeAvailableUses {
1842 code: InviteCode,
1843 tx: Tx<Option<i32>>,
1844 },
1845 ValidateInviteCode {
1846 code: InviteCode,
1847 tx: oneshot::Sender<Result<(), InviteCodeError>>,
1848 },
1849 GetInviteCodesForAccount {
1850 for_account: Did,
1851 tx: Tx<Vec<InviteCodeInfo>>,
1852 },
1853 GetInviteCodeUses {
1854 code: String,
1855 tx: Tx<Vec<InviteCodeUse>>,
1856 },
1857 DisableInviteCodesByCode {
1858 codes: Vec<String>,
1859 tx: Tx<()>,
1860 },
1861 DisableInviteCodesByAccount {
1862 accounts: Vec<Did>,
1863 tx: Tx<()>,
1864 },
1865 ListInviteCodes {
1866 cursor: Option<String>,
1867 limit: i64,
1868 sort: InviteCodeSortOrder,
1869 tx: Tx<Vec<InviteCodeRow>>,
1870 },
1871 GetUserDidsByIds {
1872 user_ids: Vec<Uuid>,
1873 tx: Tx<Vec<(Uuid, Did)>>,
1874 },
1875 GetInviteCodeUsesBatch {
1876 codes: Vec<String>,
1877 tx: Tx<Vec<InviteCodeUse>>,
1878 },
1879 GetInvitesCreatedByUser {
1880 user_id: Uuid,
1881 tx: Tx<Vec<InviteCodeInfo>>,
1882 },
1883 GetInviteCodeInfo {
1884 code: String,
1885 tx: Tx<Option<InviteCodeInfo>>,
1886 },
1887 GetInviteCodesByUsers {
1888 user_ids: Vec<Uuid>,
1889 tx: Tx<Vec<(Uuid, InviteCodeInfo)>>,
1890 },
1891 GetInviteCodeUsedByUser {
1892 user_id: Uuid,
1893 tx: Tx<Option<String>>,
1894 },
1895 DeleteInviteCodeUsesByUser {
1896 user_id: Uuid,
1897 tx: Tx<()>,
1898 },
1899 DeleteInviteCodesByUser {
1900 user_id: Uuid,
1901 tx: Tx<()>,
1902 },
1903 ReserveSigningKey {
1904 did: Option<Did>,
1905 public_key_did_key: String,
1906 private_key_bytes: Vec<u8>,
1907 expires_at: DateTime<Utc>,
1908 tx: Tx<Uuid>,
1909 },
1910 GetReservedSigningKey {
1911 public_key_did_key: String,
1912 tx: Tx<Option<ReservedSigningKey>>,
1913 },
1914 MarkSigningKeyUsed {
1915 key_id: Uuid,
1916 tx: Tx<()>,
1917 },
1918 CreateDeletionRequest {
1919 token: String,
1920 did: Did,
1921 expires_at: DateTime<Utc>,
1922 tx: Tx<()>,
1923 },
1924 GetDeletionRequest {
1925 token: String,
1926 tx: Tx<Option<DeletionRequest>>,
1927 },
1928 DeleteDeletionRequest {
1929 token: String,
1930 tx: Tx<()>,
1931 },
1932 DeleteDeletionRequestsByDid {
1933 did: Did,
1934 tx: Tx<()>,
1935 },
1936 UpsertAccountPreference {
1937 user_id: Uuid,
1938 name: String,
1939 value_json: serde_json::Value,
1940 tx: Tx<()>,
1941 },
1942 InsertAccountPreferenceIfNotExists {
1943 user_id: Uuid,
1944 name: String,
1945 value_json: serde_json::Value,
1946 tx: Tx<()>,
1947 },
1948 GetServerConfig {
1949 key: String,
1950 tx: Tx<Option<String>>,
1951 },
1952 InsertReport {
1953 id: i64,
1954 reason_type: String,
1955 reason: Option<String>,
1956 subject_json: serde_json::Value,
1957 reported_by_did: Did,
1958 created_at: DateTime<Utc>,
1959 tx: Tx<()>,
1960 },
1961 DeletePlcTokensForUser {
1962 user_id: Uuid,
1963 tx: Tx<()>,
1964 },
1965 InsertPlcToken {
1966 user_id: Uuid,
1967 token: String,
1968 expires_at: DateTime<Utc>,
1969 tx: Tx<()>,
1970 },
1971 GetPlcTokenExpiry {
1972 user_id: Uuid,
1973 token: String,
1974 tx: Tx<Option<DateTime<Utc>>>,
1975 },
1976 DeletePlcToken {
1977 user_id: Uuid,
1978 token: String,
1979 tx: Tx<()>,
1980 },
1981 GetAccountPreferences {
1982 user_id: Uuid,
1983 tx: Tx<Vec<(String, serde_json::Value)>>,
1984 },
1985 ReplaceNamespacePreferences {
1986 user_id: Uuid,
1987 namespace: String,
1988 preferences: Vec<(String, serde_json::Value)>,
1989 tx: Tx<()>,
1990 },
1991 GetNotificationHistory {
1992 user_id: Uuid,
1993 limit: i64,
1994 tx: Tx<Vec<NotificationHistoryRow>>,
1995 },
1996 GetServerConfigs {
1997 keys: Vec<String>,
1998 tx: Tx<Vec<(String, String)>>,
1999 },
2000 UpsertServerConfig {
2001 key: String,
2002 value: String,
2003 tx: Tx<()>,
2004 },
2005 DeleteServerConfig {
2006 key: String,
2007 tx: Tx<()>,
2008 },
2009 GetBlobStorageKeyByCid {
2010 cid: CidLink,
2011 tx: Tx<Option<String>>,
2012 },
2013 DeleteBlobByCid {
2014 cid: CidLink,
2015 tx: Tx<()>,
2016 },
2017 GetAdminAccountInfoByDid {
2018 did: Did,
2019 tx: Tx<Option<AdminAccountInfo>>,
2020 },
2021 GetAdminAccountInfosByDids {
2022 dids: Vec<Did>,
2023 tx: Tx<Vec<AdminAccountInfo>>,
2024 },
2025 GetInviteCodeUsesByUsers {
2026 user_ids: Vec<Uuid>,
2027 tx: Tx<Vec<(Uuid, String)>>,
2028 },
2029 GetDeletionRequestByDid {
2030 did: Did,
2031 tx: Tx<Option<DeletionRequestWithToken>>,
2032 },
2033 GetLatestCommsForUser {
2034 user_id: Uuid,
2035 comms_type: CommsType,
2036 limit: i64,
2037 tx: Tx<Vec<QueuedComms>>,
2038 },
2039 CountCommsByType {
2040 user_id: Uuid,
2041 comms_type: CommsType,
2042 tx: Tx<i64>,
2043 },
2044 DeleteCommsByTypeForUser {
2045 user_id: Uuid,
2046 comms_type: CommsType,
2047 tx: Tx<u64>,
2048 },
2049 ExpireDeletionRequest {
2050 token: String,
2051 tx: Tx<()>,
2052 },
2053 GetReservedSigningKeyFull {
2054 public_key_did_key: String,
2055 tx: Tx<Option<ReservedSigningKeyFull>>,
2056 },
2057 GetPlcTokensByDid {
2058 did: Did,
2059 tx: Tx<Vec<PlcTokenInfo>>,
2060 },
2061 CountPlcTokensByDid {
2062 did: Did,
2063 tx: Tx<i64>,
2064 },
2065}
2066
2067impl InfraRequest {
2068 fn routing(&self, user_hashes: &UserHashMap) -> Routing {
2069 match self {
2070 Self::UpsertAccountPreference { user_id, .. }
2071 | Self::InsertAccountPreferenceIfNotExists { user_id, .. }
2072 | Self::GetAccountPreferences { user_id, .. }
2073 | Self::ReplaceNamespacePreferences { user_id, .. }
2074 | Self::GetNotificationHistory { user_id, .. }
2075 | Self::DeletePlcTokensForUser { user_id, .. }
2076 | Self::InsertPlcToken { user_id, .. }
2077 | Self::GetPlcTokenExpiry { user_id, .. }
2078 | Self::DeletePlcToken { user_id, .. }
2079 | Self::GetInvitesCreatedByUser { user_id, .. }
2080 | Self::GetInviteCodeUsedByUser { user_id, .. }
2081 | Self::DeleteInviteCodeUsesByUser { user_id, .. }
2082 | Self::DeleteInviteCodesByUser { user_id, .. }
2083 | Self::GetLatestCommsForUser { user_id, .. }
2084 | Self::CountCommsByType { user_id, .. }
2085 | Self::DeleteCommsByTypeForUser { user_id, .. } => {
2086 uuid_to_routing(user_hashes, user_id)
2087 }
2088 Self::CreateInviteCodesBatch {
2089 created_by_user, ..
2090 } => uuid_to_routing(user_hashes, created_by_user),
2091 Self::EnqueueComms {
2092 user_id: Some(uid), ..
2093 } => uuid_to_routing(user_hashes, uid),
2094 Self::GetInviteCodesForAccount { for_account, .. } => {
2095 did_to_routing(for_account.as_str())
2096 }
2097 Self::DeleteDeletionRequestsByDid { did, .. }
2098 | Self::CreateDeletionRequest { did, .. }
2099 | Self::GetAdminAccountInfoByDid { did, .. }
2100 | Self::GetDeletionRequestByDid { did, .. }
2101 | Self::GetPlcTokensByDid { did, .. }
2102 | Self::CountPlcTokensByDid { did, .. } => did_to_routing(did),
2103 Self::GetBlobStorageKeyByCid { cid, .. } | Self::DeleteBlobByCid { cid, .. } => {
2104 cid_to_routing(cid)
2105 }
2106 _ => Routing::Global,
2107 }
2108 }
2109}
2110
2111pub enum OAuthRequest {
2112 CreateToken {
2113 data: TokenData,
2114 tx: Tx<TokenFamilyId>,
2115 },
2116 GetTokenById {
2117 token_id: TokenId,
2118 tx: Tx<Option<TokenData>>,
2119 },
2120 GetTokenByRefreshToken {
2121 refresh_token: RefreshToken,
2122 tx: Tx<Option<(TokenFamilyId, TokenData)>>,
2123 },
2124 GetTokenByPreviousRefreshToken {
2125 refresh_token: RefreshToken,
2126 tx: Tx<Option<(TokenFamilyId, TokenData)>>,
2127 },
2128 RotateToken {
2129 old_db_id: TokenFamilyId,
2130 new_refresh_token: RefreshToken,
2131 new_expires_at: DateTime<Utc>,
2132 tx: Tx<()>,
2133 },
2134 CheckRefreshTokenUsed {
2135 refresh_token: RefreshToken,
2136 tx: Tx<Option<TokenFamilyId>>,
2137 },
2138 DeleteToken {
2139 token_id: TokenId,
2140 tx: Tx<()>,
2141 },
2142 DeleteTokenFamily {
2143 db_id: TokenFamilyId,
2144 tx: Tx<()>,
2145 },
2146 ListTokensForUser {
2147 did: Did,
2148 tx: Tx<Vec<TokenData>>,
2149 },
2150 CountTokensForUser {
2151 did: Did,
2152 tx: Tx<i64>,
2153 },
2154 DeleteOldestTokensForUser {
2155 did: Did,
2156 keep_count: i64,
2157 tx: Tx<u64>,
2158 },
2159 RevokeTokensForClient {
2160 did: Did,
2161 client_id: ClientId,
2162 tx: Tx<u64>,
2163 },
2164 RevokeTokensForController {
2165 delegated_did: Did,
2166 controller_did: Did,
2167 tx: Tx<u64>,
2168 },
2169 CreateAuthorizationRequest {
2170 request_id: RequestId,
2171 data: RequestData,
2172 tx: Tx<()>,
2173 },
2174 GetAuthorizationRequest {
2175 request_id: RequestId,
2176 tx: Tx<Option<RequestData>>,
2177 },
2178 SetAuthorizationDid {
2179 request_id: RequestId,
2180 did: Did,
2181 device_id: Option<DeviceId>,
2182 tx: Tx<()>,
2183 },
2184 UpdateAuthorizationRequest {
2185 request_id: RequestId,
2186 did: Did,
2187 device_id: Option<DeviceId>,
2188 code: AuthorizationCode,
2189 tx: Tx<()>,
2190 },
2191 ConsumeAuthorizationRequestByCode {
2192 code: AuthorizationCode,
2193 tx: Tx<Option<RequestData>>,
2194 },
2195 DeleteAuthorizationRequest {
2196 request_id: RequestId,
2197 tx: Tx<()>,
2198 },
2199 DeleteExpiredAuthorizationRequests {
2200 tx: Tx<u64>,
2201 },
2202 ExtendAuthorizationRequestExpiry {
2203 request_id: RequestId,
2204 new_expires_at: DateTime<Utc>,
2205 tx: Tx<bool>,
2206 },
2207 MarkRequestAuthenticated {
2208 request_id: RequestId,
2209 did: Did,
2210 device_id: Option<DeviceId>,
2211 tx: Tx<()>,
2212 },
2213 UpdateRequestScope {
2214 request_id: RequestId,
2215 scope: String,
2216 tx: Tx<()>,
2217 },
2218 SetControllerDid {
2219 request_id: RequestId,
2220 controller_did: Did,
2221 tx: Tx<()>,
2222 },
2223 SetRequestDid {
2224 request_id: RequestId,
2225 did: Did,
2226 tx: Tx<()>,
2227 },
2228 CreateDevice {
2229 device_id: DeviceId,
2230 data: DeviceData,
2231 tx: Tx<()>,
2232 },
2233 GetDevice {
2234 device_id: DeviceId,
2235 tx: Tx<Option<DeviceData>>,
2236 },
2237 UpdateDeviceLastSeen {
2238 device_id: DeviceId,
2239 tx: Tx<()>,
2240 },
2241 DeleteDevice {
2242 device_id: DeviceId,
2243 tx: Tx<()>,
2244 },
2245 UpsertAccountDevice {
2246 did: Did,
2247 device_id: DeviceId,
2248 tx: Tx<()>,
2249 },
2250 GetDeviceAccounts {
2251 device_id: DeviceId,
2252 tx: Tx<Vec<tranquil_db_traits::DeviceAccountRow>>,
2253 },
2254 VerifyAccountOnDevice {
2255 device_id: DeviceId,
2256 did: Did,
2257 tx: Tx<bool>,
2258 },
2259 CheckAndRecordDpopJti {
2260 jti: DPoPProofId,
2261 tx: Tx<bool>,
2262 },
2263 CleanupExpiredDpopJtis {
2264 max_age_secs: i64,
2265 tx: Tx<u64>,
2266 },
2267 Create2faChallenge {
2268 did: Did,
2269 request_uri: RequestId,
2270 tx: Tx<tranquil_db_traits::TwoFactorChallenge>,
2271 },
2272 Get2faChallenge {
2273 request_uri: RequestId,
2274 tx: Tx<Option<tranquil_db_traits::TwoFactorChallenge>>,
2275 },
2276 Increment2faAttempts {
2277 id: Uuid,
2278 tx: Tx<i32>,
2279 },
2280 Delete2faChallenge {
2281 id: Uuid,
2282 tx: Tx<()>,
2283 },
2284 Delete2faChallengeByRequestUri {
2285 request_uri: RequestId,
2286 tx: Tx<()>,
2287 },
2288 CleanupExpired2faChallenges {
2289 tx: Tx<u64>,
2290 },
2291 CheckUser2faEnabled {
2292 did: Did,
2293 tx: Tx<bool>,
2294 },
2295 GetScopePreferences {
2296 did: Did,
2297 client_id: ClientId,
2298 tx: Tx<Vec<ScopePreference>>,
2299 },
2300 UpsertScopePreferences {
2301 did: Did,
2302 client_id: ClientId,
2303 prefs: Vec<ScopePreference>,
2304 tx: Tx<()>,
2305 },
2306 DeleteScopePreferences {
2307 did: Did,
2308 client_id: ClientId,
2309 tx: Tx<()>,
2310 },
2311 UpsertAuthorizedClient {
2312 did: Did,
2313 client_id: ClientId,
2314 data: AuthorizedClientData,
2315 tx: Tx<()>,
2316 },
2317 GetAuthorizedClient {
2318 did: Did,
2319 client_id: ClientId,
2320 tx: Tx<Option<AuthorizedClientData>>,
2321 },
2322 ListTrustedDevices {
2323 did: Did,
2324 tx: Tx<Vec<tranquil_db_traits::TrustedDeviceRow>>,
2325 },
2326 GetDeviceTrustInfo {
2327 device_id: DeviceId,
2328 did: Did,
2329 tx: Tx<Option<tranquil_db_traits::DeviceTrustInfo>>,
2330 },
2331 DeviceBelongsToUser {
2332 device_id: DeviceId,
2333 did: Did,
2334 tx: Tx<bool>,
2335 },
2336 RevokeDeviceTrust {
2337 device_id: DeviceId,
2338 did: Did,
2339 tx: Tx<()>,
2340 },
2341 UpdateDeviceFriendlyName {
2342 device_id: DeviceId,
2343 did: Did,
2344 friendly_name: Option<String>,
2345 tx: Tx<()>,
2346 },
2347 TrustDevice {
2348 device_id: DeviceId,
2349 did: Did,
2350 trusted_at: DateTime<Utc>,
2351 trusted_until: DateTime<Utc>,
2352 tx: Tx<()>,
2353 },
2354 ExtendDeviceTrust {
2355 device_id: DeviceId,
2356 did: Did,
2357 trusted_until: DateTime<Utc>,
2358 tx: Tx<()>,
2359 },
2360 ListSessionsByDid {
2361 did: Did,
2362 tx: Tx<Vec<tranquil_db_traits::OAuthSessionListItem>>,
2363 },
2364 DeleteSessionById {
2365 session_id: TokenFamilyId,
2366 did: Did,
2367 tx: Tx<u64>,
2368 },
2369 DeleteSessionsByDid {
2370 did: Did,
2371 tx: Tx<u64>,
2372 },
2373 DeleteSessionsByDidExcept {
2374 did: Did,
2375 except_token_id: TokenId,
2376 tx: Tx<u64>,
2377 },
2378 Get2faChallengeCode {
2379 request_uri: RequestId,
2380 tx: Tx<Option<String>>,
2381 },
2382}
2383
2384impl OAuthRequest {
2385 fn routing(&self) -> Routing {
2386 match self {
2387 Self::ListTokensForUser { did, .. }
2388 | Self::CountTokensForUser { did, .. }
2389 | Self::DeleteOldestTokensForUser { did, .. }
2390 | Self::RevokeTokensForClient { did, .. }
2391 | Self::Create2faChallenge { did, .. }
2392 | Self::CheckUser2faEnabled { did, .. }
2393 | Self::GetScopePreferences { did, .. }
2394 | Self::UpsertScopePreferences { did, .. }
2395 | Self::DeleteScopePreferences { did, .. }
2396 | Self::UpsertAuthorizedClient { did, .. }
2397 | Self::GetAuthorizedClient { did, .. }
2398 | Self::ListTrustedDevices { did, .. }
2399 | Self::GetDeviceTrustInfo { did, .. }
2400 | Self::DeviceBelongsToUser { did, .. }
2401 | Self::UpsertAccountDevice { did, .. }
2402 | Self::VerifyAccountOnDevice { did, .. }
2403 | Self::ListSessionsByDid { did, .. }
2404 | Self::DeleteSessionsByDid { did, .. }
2405 | Self::DeleteSessionsByDidExcept { did, .. }
2406 | Self::DeleteSessionById { did, .. } => did_to_routing(did),
2407 Self::RevokeTokensForController { delegated_did, .. } => did_to_routing(delegated_did),
2408 Self::SetAuthorizationDid { did, .. }
2409 | Self::UpdateAuthorizationRequest { did, .. }
2410 | Self::MarkRequestAuthenticated { did, .. }
2411 | Self::SetRequestDid { did, .. } => did_to_routing(did),
2412 Self::CreateToken { .. }
2413 | Self::GetTokenById { .. }
2414 | Self::GetTokenByRefreshToken { .. }
2415 | Self::GetTokenByPreviousRefreshToken { .. }
2416 | Self::RotateToken { .. }
2417 | Self::CheckRefreshTokenUsed { .. }
2418 | Self::DeleteToken { .. }
2419 | Self::DeleteTokenFamily { .. }
2420 | Self::CreateAuthorizationRequest { .. }
2421 | Self::GetAuthorizationRequest { .. }
2422 | Self::ConsumeAuthorizationRequestByCode { .. }
2423 | Self::DeleteAuthorizationRequest { .. }
2424 | Self::DeleteExpiredAuthorizationRequests { .. }
2425 | Self::ExtendAuthorizationRequestExpiry { .. }
2426 | Self::UpdateRequestScope { .. }
2427 | Self::SetControllerDid { .. }
2428 | Self::CreateDevice { .. }
2429 | Self::GetDevice { .. }
2430 | Self::UpdateDeviceLastSeen { .. }
2431 | Self::DeleteDevice { .. }
2432 | Self::GetDeviceAccounts { .. }
2433 | Self::CheckAndRecordDpopJti { .. }
2434 | Self::CleanupExpiredDpopJtis { .. }
2435 | Self::Get2faChallenge { .. }
2436 | Self::Increment2faAttempts { .. }
2437 | Self::Delete2faChallenge { .. }
2438 | Self::Delete2faChallengeByRequestUri { .. }
2439 | Self::CleanupExpired2faChallenges { .. }
2440 | Self::RevokeDeviceTrust { .. }
2441 | Self::UpdateDeviceFriendlyName { .. }
2442 | Self::TrustDevice { .. }
2443 | Self::ExtendDeviceTrust { .. }
2444 | Self::Get2faChallengeCode { .. } => Routing::Global,
2445 }
2446 }
2447}
2448
2449fn convert_repo_info(r: super::repo_ops::RepoInfo) -> tranquil_db_traits::RepoInfo {
2450 tranquil_db_traits::RepoInfo {
2451 user_id: r.user_id,
2452 repo_root_cid: r.repo_root_cid,
2453 repo_rev: r.repo_rev.map(Tid::from),
2454 }
2455}
2456
2457fn convert_repo_account(
2458 r: super::repo_ops::RepoAccountEntry,
2459) -> tranquil_db_traits::RepoAccountInfo {
2460 tranquil_db_traits::RepoAccountInfo {
2461 user_id: r.user_id,
2462 did: r.did,
2463 deactivated_at: r.deactivated_at,
2464 takedown_ref: r.takedown_ref,
2465 repo_root_cid: r.repo_root_cid,
2466 }
2467}
2468
2469fn convert_repo_list_entry(
2470 r: super::repo_ops::RepoListEntry,
2471) -> Result<tranquil_db_traits::RepoListItem, DbError> {
2472 let did = r
2473 .did
2474 .ok_or(DbError::CorruptData("repo_meta missing DID field"))?;
2475 Ok(tranquil_db_traits::RepoListItem {
2476 did: Did::from(did),
2477 deactivated_at: r.deactivated_at,
2478 takedown_ref: r.takedown_ref,
2479 repo_root_cid: r.repo_root_cid,
2480 repo_rev: r.repo_rev.map(Tid::from),
2481 })
2482}
2483
2484fn convert_record_info(r: super::record_ops::RecordInfo) -> tranquil_db_traits::RecordInfo {
2485 tranquil_db_traits::RecordInfo {
2486 rkey: r.rkey,
2487 record_cid: r.record_cid,
2488 }
2489}
2490
2491fn convert_full_record_info(
2492 r: super::record_ops::FullRecordInfo,
2493) -> tranquil_db_traits::FullRecordInfo {
2494 tranquil_db_traits::FullRecordInfo {
2495 collection: r.collection,
2496 rkey: r.rkey,
2497 record_cid: r.record_cid,
2498 }
2499}
2500
2501fn convert_record_with_takedown(
2502 r: super::record_ops::RecordWithTakedown,
2503) -> tranquil_db_traits::RecordWithTakedown {
2504 tranquil_db_traits::RecordWithTakedown {
2505 id: r.id,
2506 takedown_ref: r.takedown_ref,
2507 }
2508}
2509
2510fn convert_without_rev(
2511 r: super::repo_ops::RepoWithoutRevEntry,
2512) -> tranquil_db_traits::RepoWithoutRev {
2513 tranquil_db_traits::RepoWithoutRev {
2514 user_id: r.user_id,
2515 repo_root_cid: r.repo_root_cid,
2516 }
2517}
2518
2519struct HandlerState<S: StorageIO> {
2520 metastore: Metastore,
2521 event_ops: EventOps<S>,
2522 commit_ops: CommitOps<S>,
2523}
2524
2525fn dispatch_repo<S: StorageIO>(state: &HandlerState<S>, req: RepoRequest) {
2526 match req {
2527 RepoRequest::CreateRepoFull {
2528 user_id,
2529 did,
2530 handle,
2531 repo_root_cid,
2532 repo_rev,
2533 tx,
2534 } => {
2535 let result = state
2536 .metastore
2537 .repo_ops()
2538 .create_repo(
2539 state.metastore.database(),
2540 user_id,
2541 &did,
2542 &handle,
2543 &repo_root_cid,
2544 &repo_rev,
2545 )
2546 .map_err(metastore_to_db);
2547 let _ = tx.send(result);
2548 }
2549 RepoRequest::UpdateRepoRoot {
2550 user_id,
2551 repo_root_cid,
2552 repo_rev,
2553 tx,
2554 } => {
2555 let result = state
2556 .metastore
2557 .repo_ops()
2558 .update_repo_root(
2559 state.metastore.database(),
2560 user_id,
2561 &repo_root_cid,
2562 &repo_rev,
2563 )
2564 .map_err(metastore_to_db);
2565 let _ = tx.send(result);
2566 }
2567 RepoRequest::UpdateRepoRev {
2568 user_id,
2569 repo_rev,
2570 tx,
2571 } => {
2572 let result = state
2573 .metastore
2574 .repo_ops()
2575 .update_repo_rev(state.metastore.database(), user_id, &repo_rev)
2576 .map_err(metastore_to_db);
2577 let _ = tx.send(result);
2578 }
2579 RepoRequest::DeleteRepo { user_id, tx } => {
2580 let result = state
2581 .metastore
2582 .repo_ops()
2583 .delete_repo(state.metastore.database(), user_id)
2584 .map_err(metastore_to_db);
2585 let _ = tx.send(result);
2586 }
2587 RepoRequest::GetRepoRootForUpdate { user_id, tx } => {
2588 let result = state
2589 .metastore
2590 .repo_ops()
2591 .get_repo_root_for_update(user_id)
2592 .map_err(metastore_to_db);
2593 let _ = tx.send(result);
2594 }
2595 RepoRequest::GetRepo { user_id, tx } => {
2596 let result = state
2597 .metastore
2598 .repo_ops()
2599 .get_repo(user_id)
2600 .map(|opt| opt.map(convert_repo_info))
2601 .map_err(metastore_to_db);
2602 let _ = tx.send(result);
2603 }
2604 RepoRequest::GetRepoRootByDid { did, tx } => {
2605 let result = state
2606 .metastore
2607 .repo_ops()
2608 .get_repo_root_by_did(&did)
2609 .map_err(metastore_to_db);
2610 let _ = tx.send(result);
2611 }
2612 RepoRequest::CountRepos { tx } => {
2613 let result = state
2614 .metastore
2615 .repo_ops()
2616 .count_repos()
2617 .map_err(metastore_to_db);
2618 let _ = tx.send(result);
2619 }
2620 RepoRequest::GetReposWithoutRev { tx } => {
2621 let result = state
2622 .metastore
2623 .repo_ops()
2624 .get_repos_without_rev(MAX_REPOS_WITHOUT_REV)
2625 .map(|v| v.into_iter().map(convert_without_rev).collect())
2626 .map_err(metastore_to_db);
2627 let _ = tx.send(result);
2628 }
2629 RepoRequest::GetRepoRootCidByUserId { user_id, tx } => {
2630 let result = state
2631 .metastore
2632 .repo_ops()
2633 .get_repo_root_cid_by_user_id(user_id)
2634 .map_err(metastore_to_db);
2635 let _ = tx.send(result);
2636 }
2637 RepoRequest::GetAccountWithRepo { did, tx } => {
2638 let result = state
2639 .metastore
2640 .repo_ops()
2641 .get_account_with_repo(&did)
2642 .map(|opt| opt.map(convert_repo_account))
2643 .map_err(metastore_to_db);
2644 let _ = tx.send(result);
2645 }
2646 RepoRequest::ListReposPaginated {
2647 cursor_user_hash,
2648 limit,
2649 tx,
2650 } => {
2651 let result = state
2652 .metastore
2653 .repo_ops()
2654 .list_repos_paginated(cursor_user_hash, limit)
2655 .map_err(metastore_to_db)
2656 .and_then(|entries| entries.into_iter().map(convert_repo_list_entry).collect());
2657 let _ = tx.send(result);
2658 }
2659 RepoRequest::UpdateRepoStatus {
2660 did,
2661 takedown,
2662 takedown_ref,
2663 deactivated,
2664 tx,
2665 } => {
2666 let result = state
2667 .metastore
2668 .repo_ops()
2669 .update_repo_status(
2670 state.metastore.database(),
2671 &did,
2672 takedown,
2673 takedown_ref.as_deref(),
2674 deactivated,
2675 )
2676 .map_err(metastore_to_db);
2677 let _ = tx.send(result);
2678 }
2679 }
2680}
2681
2682fn dispatch_record<S: StorageIO>(state: &HandlerState<S>, req: RecordRequest) {
2683 match req {
2684 RecordRequest::UpsertRecords {
2685 repo_id,
2686 collections,
2687 rkeys,
2688 record_cids,
2689 repo_rev,
2690 tx,
2691 } => {
2692 let result = (|| {
2693 let (user_hash, mut meta) = state
2694 .metastore
2695 .repo_ops()
2696 .get_repo_meta(repo_id)
2697 .map_err(metastore_to_db)?
2698 .ok_or(DbError::Query("unknown user_id".to_string()))?;
2699 let writes: Vec<super::record_ops::RecordWrite<'_>> = collections
2700 .iter()
2701 .zip(rkeys.iter())
2702 .zip(record_cids.iter())
2703 .map(|((c, r), cid)| super::record_ops::RecordWrite {
2704 collection: c,
2705 rkey: r,
2706 cid,
2707 })
2708 .collect();
2709 let mut batch = state.metastore.database().batch();
2710 state
2711 .metastore
2712 .record_ops()
2713 .upsert_records(&mut batch, user_hash, &writes)
2714 .map_err(metastore_to_db)?;
2715 meta.repo_rev = repo_rev;
2716 state
2717 .metastore
2718 .repo_ops()
2719 .write_repo_meta(&mut batch, user_hash, &meta);
2720 batch.commit().map_err(|e| DbError::Query(e.to_string()))
2721 })();
2722 let _ = tx.send(result);
2723 }
2724 RecordRequest::DeleteRecords {
2725 repo_id,
2726 collections,
2727 rkeys,
2728 tx,
2729 } => {
2730 let result = (|| {
2731 let user_hash = state
2732 .metastore
2733 .user_hashes()
2734 .get(&repo_id)
2735 .ok_or(DbError::Query("unknown user_id".to_string()))?;
2736 let deletes: Vec<super::record_ops::RecordDelete<'_>> = collections
2737 .iter()
2738 .zip(rkeys.iter())
2739 .map(|(c, r)| super::record_ops::RecordDelete {
2740 collection: c,
2741 rkey: r,
2742 })
2743 .collect();
2744 let mut batch = state.metastore.database().batch();
2745 state
2746 .metastore
2747 .record_ops()
2748 .delete_records(&mut batch, user_hash, &deletes);
2749 batch.commit().map_err(|e| DbError::Query(e.to_string()))
2750 })();
2751 let _ = tx.send(result);
2752 }
2753 RecordRequest::DeleteAllRecords { repo_id, tx } => {
2754 let result = (|| {
2755 let user_hash = state
2756 .metastore
2757 .user_hashes()
2758 .get(&repo_id)
2759 .ok_or(DbError::Query("unknown user_id".to_string()))?;
2760 let mut batch = state.metastore.database().batch();
2761 state
2762 .metastore
2763 .record_ops()
2764 .delete_all_records(&mut batch, user_hash)
2765 .map_err(metastore_to_db)?;
2766 batch.commit().map_err(|e| DbError::Query(e.to_string()))
2767 })();
2768 let _ = tx.send(result);
2769 }
2770 RecordRequest::GetRecordCid {
2771 repo_id,
2772 collection,
2773 rkey,
2774 tx,
2775 } => {
2776 let result = state
2777 .metastore
2778 .record_ops()
2779 .get_record_cid(repo_id, &collection, &rkey)
2780 .map_err(metastore_to_db);
2781 let _ = tx.send(result);
2782 }
2783 RecordRequest::ListRecords {
2784 repo_id,
2785 collection,
2786 cursor,
2787 limit,
2788 reverse,
2789 rkey_start,
2790 rkey_end,
2791 tx,
2792 } => {
2793 let query = ListRecordsQuery {
2794 user_id: repo_id,
2795 collection: &collection,
2796 cursor: cursor.as_ref(),
2797 limit: usize::try_from(limit).unwrap_or(0),
2798 reverse,
2799 rkey_start: rkey_start.as_ref(),
2800 rkey_end: rkey_end.as_ref(),
2801 };
2802 let result = state
2803 .metastore
2804 .record_ops()
2805 .list_records(&query)
2806 .map(|v| v.into_iter().map(convert_record_info).collect())
2807 .map_err(metastore_to_db);
2808 let _ = tx.send(result);
2809 }
2810 RecordRequest::GetAllRecords { repo_id, tx } => {
2811 let result = state
2812 .metastore
2813 .record_ops()
2814 .get_all_records(repo_id)
2815 .map(|v| v.into_iter().map(convert_full_record_info).collect())
2816 .map_err(metastore_to_db);
2817 let _ = tx.send(result);
2818 }
2819 RecordRequest::ListCollections { repo_id, tx } => {
2820 let result = state
2821 .metastore
2822 .record_ops()
2823 .list_collections(repo_id)
2824 .map_err(metastore_to_db);
2825 let _ = tx.send(result);
2826 }
2827 RecordRequest::CountRecords { repo_id, tx } => {
2828 let result = state
2829 .metastore
2830 .record_ops()
2831 .count_records(repo_id)
2832 .map_err(metastore_to_db);
2833 let _ = tx.send(result);
2834 }
2835 RecordRequest::CountAllRecords { tx } => {
2836 let result = state
2837 .metastore
2838 .record_ops()
2839 .count_all_records()
2840 .map_err(metastore_to_db);
2841 let _ = tx.send(result);
2842 }
2843 RecordRequest::GetRecordByCid { cid, tx } => {
2844 let result = state
2845 .metastore
2846 .record_ops()
2847 .get_record_by_cid(&cid, None)
2848 .map(|opt| opt.map(convert_record_with_takedown))
2849 .map_err(metastore_to_db);
2850 let _ = tx.send(result);
2851 }
2852 RecordRequest::SetRecordTakedown {
2853 cid,
2854 takedown_ref,
2855 scope_user,
2856 tx,
2857 } => {
2858 let result = state
2859 .metastore
2860 .record_ops()
2861 .set_record_takedown(
2862 state.metastore.database(),
2863 &cid,
2864 takedown_ref.as_deref(),
2865 scope_user,
2866 )
2867 .map_err(metastore_to_db);
2868 let _ = tx.send(result);
2869 }
2870 }
2871}
2872
2873fn dispatch_user_block<S: StorageIO>(state: &HandlerState<S>, req: UserBlockRequest) {
2874 match req {
2875 UserBlockRequest::InsertUserBlocks {
2876 user_id,
2877 block_cids,
2878 repo_rev,
2879 tx,
2880 } => {
2881 let result = (|| {
2882 let user_hash = state
2883 .metastore
2884 .user_hashes()
2885 .get(&user_id)
2886 .ok_or(DbError::Query("unknown user_id".to_string()))?;
2887 let mut batch = state.metastore.database().batch();
2888 state
2889 .metastore
2890 .user_block_ops()
2891 .insert_user_blocks(&mut batch, user_hash, &block_cids, &repo_rev)
2892 .map_err(metastore_to_db)?;
2893 batch.commit().map_err(|e| DbError::Query(e.to_string()))
2894 })();
2895 let _ = tx.send(result);
2896 }
2897 UserBlockRequest::DeleteUserBlocks {
2898 user_id,
2899 block_cids,
2900 tx,
2901 } => {
2902 let result = (|| {
2903 let user_hash = state
2904 .metastore
2905 .user_hashes()
2906 .get(&user_id)
2907 .ok_or(DbError::Query("unknown user_id".to_string()))?;
2908 let mut batch = state.metastore.database().batch();
2909 state
2910 .metastore
2911 .user_block_ops()
2912 .delete_user_blocks_by_cid(&mut batch, user_hash, &block_cids)
2913 .map_err(metastore_to_db)?;
2914 batch.commit().map_err(|e| DbError::Query(e.to_string()))
2915 })();
2916 let _ = tx.send(result);
2917 }
2918 UserBlockRequest::GetUserBlockCidsSinceRev {
2919 user_id,
2920 since_rev,
2921 tx,
2922 } => {
2923 let result = state
2924 .metastore
2925 .user_block_ops()
2926 .get_user_block_cids_since_rev(user_id, &since_rev)
2927 .map_err(metastore_to_db);
2928 let _ = tx.send(result);
2929 }
2930 UserBlockRequest::CountUserBlocks { user_id, tx } => {
2931 let result = state
2932 .metastore
2933 .user_block_ops()
2934 .count_user_blocks(user_id)
2935 .map_err(metastore_to_db);
2936 let _ = tx.send(result);
2937 }
2938 }
2939}
2940
2941fn dispatch_event<S: StorageIO + 'static>(state: &HandlerState<S>, req: EventRequest) {
2942 match req {
2943 EventRequest::InsertCommitEvent { data, tx } => {
2944 let result = state.event_ops.insert_commit_event(&data);
2945 let _ = tx.send(result);
2946 }
2947 EventRequest::InsertIdentityEvent { did, handle, tx } => {
2948 let result = state.event_ops.insert_identity_event(&did, handle.as_ref());
2949 let _ = tx.send(result);
2950 }
2951 EventRequest::InsertAccountEvent { did, status, tx } => {
2952 let result = state.event_ops.insert_account_event(&did, status);
2953 let _ = tx.send(result);
2954 }
2955 EventRequest::InsertSyncEvent {
2956 did,
2957 commit_cid,
2958 rev,
2959 commit_bytes,
2960 tx,
2961 } => {
2962 let result =
2963 state
2964 .event_ops
2965 .insert_sync_event(&did, &commit_cid, rev.as_deref(), &commit_bytes);
2966 let _ = tx.send(result);
2967 }
2968 EventRequest::InsertGenesisCommitEvent {
2969 did,
2970 commit_cid,
2971 mst_root_cid,
2972 rev,
2973 commit_bytes,
2974 mst_root_bytes,
2975 tx,
2976 } => {
2977 let result = state.event_ops.insert_genesis_commit_event(
2978 &did,
2979 &commit_cid,
2980 &mst_root_cid,
2981 &rev,
2982 &commit_bytes,
2983 &mst_root_bytes,
2984 );
2985 let _ = tx.send(result);
2986 }
2987 EventRequest::PurgeDidEventsKeepingLatest { did, tx } => {
2988 let result = state.event_ops.purge_did_events_keeping_latest(&did);
2989 let _ = tx.send(result);
2990 }
2991 EventRequest::GetMaxSeq { tx } => {
2992 let _ = tx.send(Ok(state.event_ops.get_max_seq()));
2993 }
2994 EventRequest::GetMinSeqSince { since, tx } => {
2995 let _ = tx.send(state.event_ops.get_min_seq_since(since));
2996 }
2997 EventRequest::GetEventsSinceSeq {
2998 since_seq,
2999 limit,
3000 tx,
3001 } => {
3002 let _ = tx.send(state.event_ops.get_events_since_seq(since_seq, limit));
3003 }
3004 EventRequest::GetEventsInSeqRange {
3005 start_seq,
3006 end_seq,
3007 tx,
3008 } => {
3009 let _ = tx.send(state.event_ops.get_events_in_seq_range(start_seq, end_seq));
3010 }
3011 EventRequest::GetEventBySeq { seq, tx } => {
3012 let _ = tx.send(state.event_ops.get_event_by_seq(seq));
3013 }
3014 EventRequest::GetEventsSinceCursor { cursor, limit, tx } => {
3015 let _ = tx.send(state.event_ops.get_events_since_cursor(cursor, limit));
3016 }
3017 }
3018}
3019
3020fn dispatch_commit<S: StorageIO + 'static>(state: &HandlerState<S>, req: CommitRequest) {
3021 match req {
3022 CommitRequest::ApplyCommit { input, tx } => {
3023 let _ = tx.send(state.commit_ops.apply_commit(*input));
3024 }
3025 CommitRequest::ImportRepoData {
3026 user_id,
3027 blocks,
3028 records,
3029 expected_root_cid,
3030 tx,
3031 } => {
3032 let _ = tx.send(state.commit_ops.import_repo_data(
3033 user_id,
3034 &blocks,
3035 &records,
3036 expected_root_cid.as_ref(),
3037 ));
3038 }
3039 CommitRequest::GetUsersWithoutBlocks { tx } => {
3040 let _ = tx.send(
3041 state
3042 .commit_ops
3043 .get_users_without_blocks()
3044 .map_err(metastore_to_db),
3045 );
3046 }
3047 CommitRequest::GetUsersNeedingRecordBlobsBackfill { limit, tx } => {
3048 let _ = tx.send(
3049 state
3050 .commit_ops
3051 .get_users_needing_record_blobs_backfill(limit)
3052 .map_err(metastore_to_db),
3053 );
3054 }
3055 CommitRequest::InsertRecordBlobs {
3056 repo_id,
3057 record_uris,
3058 blob_cids,
3059 tx,
3060 } => {
3061 let _ = tx.send(
3062 state
3063 .commit_ops
3064 .insert_record_blobs(repo_id, &record_uris, &blob_cids)
3065 .map_err(metastore_to_db),
3066 );
3067 }
3068 }
3069}
3070
3071fn dispatch_backlink<S: StorageIO>(state: &HandlerState<S>, req: BacklinkRequest) {
3072 match req {
3073 BacklinkRequest::GetBacklinkConflicts {
3074 repo_id,
3075 collection,
3076 backlinks,
3077 tx,
3078 } => {
3079 let result = state
3080 .metastore
3081 .backlink_ops()
3082 .get_backlink_conflicts(repo_id, &collection, &backlinks)
3083 .map_err(metastore_to_db);
3084 let _ = tx.send(result);
3085 }
3086 BacklinkRequest::AddBacklinks {
3087 repo_id,
3088 backlinks,
3089 tx,
3090 } => {
3091 let result = (|| {
3092 let user_hash = state
3093 .metastore
3094 .user_hashes()
3095 .get(&repo_id)
3096 .ok_or(DbError::Query("unknown user_id".to_string()))?;
3097 let mut batch = state.metastore.database().batch();
3098 state
3099 .metastore
3100 .backlink_ops()
3101 .add_backlinks(&mut batch, user_hash, &backlinks)
3102 .map_err(metastore_to_db)?;
3103 batch.commit().map_err(|e| DbError::Query(e.to_string()))
3104 })();
3105 let _ = tx.send(result);
3106 }
3107 BacklinkRequest::RemoveBacklinksByUri { uri, tx } => {
3108 let result = (|| {
3109 let did_str = uri
3110 .did()
3111 .ok_or(DbError::Query("backlink uri missing did".to_string()))?;
3112 let user_hash = UserHash::from_did(did_str);
3113 let mut batch = state.metastore.database().batch();
3114 state
3115 .metastore
3116 .backlink_ops()
3117 .remove_backlinks_by_uri(&mut batch, user_hash, &uri)
3118 .map_err(metastore_to_db)?;
3119 batch.commit().map_err(|e| DbError::Query(e.to_string()))
3120 })();
3121 let _ = tx.send(result);
3122 }
3123 BacklinkRequest::RemoveBacklinksByRepo { repo_id, tx } => {
3124 let result = (|| {
3125 let user_hash = state
3126 .metastore
3127 .user_hashes()
3128 .get(&repo_id)
3129 .ok_or(DbError::Query("unknown user_id".to_string()))?;
3130 let mut batch = state.metastore.database().batch();
3131 state
3132 .metastore
3133 .backlink_ops()
3134 .remove_backlinks_by_repo(&mut batch, user_hash)
3135 .map_err(metastore_to_db)?;
3136 batch.commit().map_err(|e| DbError::Query(e.to_string()))
3137 })();
3138 let _ = tx.send(result);
3139 }
3140 }
3141}
3142
3143fn dispatch_blob<S: StorageIO + 'static>(state: &HandlerState<S>, req: BlobRequest) {
3144 match req {
3145 BlobRequest::InsertBlob {
3146 cid,
3147 mime_type,
3148 size_bytes,
3149 created_by_user,
3150 storage_key,
3151 tx,
3152 } => {
3153 let result = state
3154 .metastore
3155 .blob_ops()
3156 .insert_blob(&cid, &mime_type, size_bytes, created_by_user, &storage_key)
3157 .map_err(metastore_to_db);
3158 let _ = tx.send(result);
3159 }
3160 BlobRequest::GetBlobMetadata { cid, tx } => {
3161 let result = state
3162 .metastore
3163 .blob_ops()
3164 .get_blob_metadata(&cid)
3165 .map_err(metastore_to_db);
3166 let _ = tx.send(result);
3167 }
3168 BlobRequest::GetBlobWithTakedown { cid, tx } => {
3169 let result = state
3170 .metastore
3171 .blob_ops()
3172 .get_blob_with_takedown(&cid)
3173 .map_err(metastore_to_db);
3174 let _ = tx.send(result);
3175 }
3176 BlobRequest::GetBlobStorageKey { cid, tx } => {
3177 let result = state
3178 .metastore
3179 .blob_ops()
3180 .get_blob_storage_key(&cid)
3181 .map_err(metastore_to_db);
3182 let _ = tx.send(result);
3183 }
3184 BlobRequest::ListBlobsByUser {
3185 user_id,
3186 cursor,
3187 limit,
3188 tx,
3189 } => {
3190 let result = state
3191 .metastore
3192 .blob_ops()
3193 .list_blobs_by_user(
3194 user_id,
3195 cursor.as_deref(),
3196 usize::try_from(limit).unwrap_or(0),
3197 )
3198 .map_err(metastore_to_db);
3199 let _ = tx.send(result);
3200 }
3201 BlobRequest::ListBlobsSinceRev { did, since, tx } => {
3202 let result = state.event_ops.get_blob_cids_since_rev(&did, &since);
3203 let _ = tx.send(result);
3204 }
3205 BlobRequest::CountBlobsByUser { user_id, tx } => {
3206 let result = state
3207 .metastore
3208 .blob_ops()
3209 .count_blobs_by_user(user_id)
3210 .map_err(metastore_to_db);
3211 let _ = tx.send(result);
3212 }
3213 BlobRequest::SumBlobStorage { tx } => {
3214 let result = state
3215 .metastore
3216 .blob_ops()
3217 .sum_blob_storage()
3218 .map_err(metastore_to_db);
3219 let _ = tx.send(result);
3220 }
3221 BlobRequest::UpdateBlobTakedown {
3222 cid,
3223 takedown_ref,
3224 tx,
3225 } => {
3226 let result = state
3227 .metastore
3228 .blob_ops()
3229 .update_blob_takedown(&cid, takedown_ref.as_deref())
3230 .map_err(metastore_to_db);
3231 let _ = tx.send(result);
3232 }
3233 BlobRequest::DeleteBlobByCid { cid, tx } => {
3234 let result = state
3235 .metastore
3236 .blob_ops()
3237 .delete_blob_by_cid(&cid)
3238 .map_err(metastore_to_db);
3239 let _ = tx.send(result);
3240 }
3241 BlobRequest::DeleteBlobsByUser { user_id, tx } => {
3242 let result = state
3243 .metastore
3244 .blob_ops()
3245 .delete_blobs_by_user(user_id)
3246 .map_err(metastore_to_db);
3247 let _ = tx.send(result);
3248 }
3249 BlobRequest::GetBlobStorageKeysByUser { user_id, tx } => {
3250 let result = state
3251 .metastore
3252 .blob_ops()
3253 .get_blob_storage_keys_by_user(user_id)
3254 .map_err(metastore_to_db);
3255 let _ = tx.send(result);
3256 }
3257 BlobRequest::ListMissingBlobs {
3258 repo_id,
3259 cursor,
3260 limit,
3261 tx,
3262 } => {
3263 let result = state
3264 .metastore
3265 .blob_ops()
3266 .list_missing_blobs(
3267 repo_id,
3268 cursor.as_deref(),
3269 usize::try_from(limit).unwrap_or(0),
3270 )
3271 .map_err(metastore_to_db);
3272 let _ = tx.send(result);
3273 }
3274 BlobRequest::CountDistinctRecordBlobs { repo_id, tx } => {
3275 let result = state
3276 .metastore
3277 .blob_ops()
3278 .count_distinct_record_blobs(repo_id)
3279 .map_err(metastore_to_db);
3280 let _ = tx.send(result);
3281 }
3282 BlobRequest::GetBlobsForExport { repo_id, tx } => {
3283 let result = state
3284 .metastore
3285 .blob_ops()
3286 .get_blobs_for_export(repo_id)
3287 .map_err(metastore_to_db);
3288 let _ = tx.send(result);
3289 }
3290 }
3291}
3292
3293fn dispatch_delegation<S: StorageIO>(state: &HandlerState<S>, req: DelegationRequest) {
3294 match req {
3295 DelegationRequest::IsDelegatedAccount { did, tx } => {
3296 let result = state
3297 .metastore
3298 .delegation_ops()
3299 .is_delegated_account(&did)
3300 .map_err(metastore_to_db);
3301 let _ = tx.send(result);
3302 }
3303 DelegationRequest::CreateDelegation {
3304 delegated_did,
3305 controller_did,
3306 granted_scopes,
3307 granted_by,
3308 tx,
3309 } => {
3310 let result = state
3311 .metastore
3312 .delegation_ops()
3313 .create_delegation(
3314 &delegated_did,
3315 &controller_did,
3316 &granted_scopes,
3317 &granted_by,
3318 )
3319 .map_err(metastore_to_db);
3320 let _ = tx.send(result);
3321 }
3322 DelegationRequest::RevokeDelegation {
3323 delegated_did,
3324 controller_did,
3325 revoked_by,
3326 tx,
3327 } => {
3328 let result = state
3329 .metastore
3330 .delegation_ops()
3331 .revoke_delegation(&delegated_did, &controller_did, &revoked_by)
3332 .map_err(metastore_to_db);
3333 let _ = tx.send(result);
3334 }
3335 DelegationRequest::UpdateDelegationScopes {
3336 delegated_did,
3337 controller_did,
3338 new_scopes,
3339 tx,
3340 } => {
3341 let result = state
3342 .metastore
3343 .delegation_ops()
3344 .update_delegation_scopes(&delegated_did, &controller_did, &new_scopes)
3345 .map_err(metastore_to_db);
3346 let _ = tx.send(result);
3347 }
3348 DelegationRequest::GetDelegation {
3349 delegated_did,
3350 controller_did,
3351 tx,
3352 } => {
3353 let result = state
3354 .metastore
3355 .delegation_ops()
3356 .get_delegation(&delegated_did, &controller_did)
3357 .map_err(metastore_to_db);
3358 let _ = tx.send(result);
3359 }
3360 DelegationRequest::GetDelegationsForAccount { delegated_did, tx } => {
3361 let result = state
3362 .metastore
3363 .delegation_ops()
3364 .get_delegations_for_account(&delegated_did)
3365 .map_err(metastore_to_db);
3366 let _ = tx.send(result);
3367 }
3368 DelegationRequest::GetAccountsControlledBy { controller_did, tx } => {
3369 let result = state
3370 .metastore
3371 .delegation_ops()
3372 .get_accounts_controlled_by(&controller_did)
3373 .map_err(metastore_to_db);
3374 let _ = tx.send(result);
3375 }
3376 DelegationRequest::CountActiveControllers { delegated_did, tx } => {
3377 let result = state
3378 .metastore
3379 .delegation_ops()
3380 .count_active_controllers(&delegated_did)
3381 .map_err(metastore_to_db);
3382 let _ = tx.send(result);
3383 }
3384 DelegationRequest::ControlsAnyAccounts { did, tx } => {
3385 let result = state
3386 .metastore
3387 .delegation_ops()
3388 .controls_any_accounts(&did)
3389 .map_err(metastore_to_db);
3390 let _ = tx.send(result);
3391 }
3392 DelegationRequest::LogDelegationAction {
3393 delegated_did,
3394 actor_did,
3395 controller_did,
3396 action_type,
3397 action_details,
3398 ip_address,
3399 user_agent,
3400 tx,
3401 } => {
3402 let result = state
3403 .metastore
3404 .delegation_ops()
3405 .log_delegation_action(
3406 &delegated_did,
3407 &actor_did,
3408 controller_did.as_ref(),
3409 action_type,
3410 action_details,
3411 ip_address.as_deref(),
3412 user_agent.as_deref(),
3413 )
3414 .map_err(metastore_to_db);
3415 let _ = tx.send(result);
3416 }
3417 DelegationRequest::GetAuditLogForAccount {
3418 delegated_did,
3419 limit,
3420 offset,
3421 tx,
3422 } => {
3423 let result = state
3424 .metastore
3425 .delegation_ops()
3426 .get_audit_log_for_account(&delegated_did, limit, offset)
3427 .map_err(metastore_to_db);
3428 let _ = tx.send(result);
3429 }
3430 DelegationRequest::CountAuditLogEntries { delegated_did, tx } => {
3431 let result = state
3432 .metastore
3433 .delegation_ops()
3434 .count_audit_log_entries(&delegated_did)
3435 .map_err(metastore_to_db);
3436 let _ = tx.send(result);
3437 }
3438 }
3439}
3440
3441fn dispatch_sso<S: StorageIO>(state: &HandlerState<S>, req: SsoRequest) {
3442 match req {
3443 SsoRequest::CreateExternalIdentity {
3444 did,
3445 provider,
3446 provider_user_id,
3447 provider_username,
3448 provider_email,
3449 tx,
3450 } => {
3451 let result = state
3452 .metastore
3453 .sso_ops()
3454 .create_external_identity(
3455 &did,
3456 provider,
3457 &provider_user_id,
3458 provider_username.as_deref(),
3459 provider_email.as_deref(),
3460 )
3461 .map_err(metastore_to_db);
3462 let _ = tx.send(result);
3463 }
3464 SsoRequest::GetExternalIdentityByProvider {
3465 provider,
3466 provider_user_id,
3467 tx,
3468 } => {
3469 let result = state
3470 .metastore
3471 .sso_ops()
3472 .get_external_identity_by_provider(provider, &provider_user_id)
3473 .map_err(metastore_to_db);
3474 let _ = tx.send(result);
3475 }
3476 SsoRequest::GetExternalIdentitiesByDid { did, tx } => {
3477 let result = state
3478 .metastore
3479 .sso_ops()
3480 .get_external_identities_by_did(&did)
3481 .map_err(metastore_to_db);
3482 let _ = tx.send(result);
3483 }
3484 SsoRequest::UpdateExternalIdentityLogin {
3485 id,
3486 provider_username,
3487 provider_email,
3488 tx,
3489 } => {
3490 let result = state
3491 .metastore
3492 .sso_ops()
3493 .update_external_identity_login(
3494 id,
3495 provider_username.as_deref(),
3496 provider_email.as_deref(),
3497 )
3498 .map_err(metastore_to_db);
3499 let _ = tx.send(result);
3500 }
3501 SsoRequest::DeleteExternalIdentity { id, did, tx } => {
3502 let result = state
3503 .metastore
3504 .sso_ops()
3505 .delete_external_identity(id, &did)
3506 .map_err(metastore_to_db);
3507 let _ = tx.send(result);
3508 }
3509 SsoRequest::CreateSsoAuthState {
3510 state: sso_state,
3511 request_uri,
3512 provider,
3513 action,
3514 nonce,
3515 code_verifier,
3516 did,
3517 tx,
3518 } => {
3519 let result = state
3520 .metastore
3521 .sso_ops()
3522 .create_sso_auth_state(
3523 &sso_state,
3524 &request_uri,
3525 provider,
3526 action,
3527 nonce.as_deref(),
3528 code_verifier.as_deref(),
3529 did.as_ref(),
3530 )
3531 .map_err(metastore_to_db);
3532 let _ = tx.send(result);
3533 }
3534 SsoRequest::ConsumeSsoAuthState {
3535 state: sso_state,
3536 tx,
3537 } => {
3538 let result = state
3539 .metastore
3540 .sso_ops()
3541 .consume_sso_auth_state(&sso_state)
3542 .map_err(metastore_to_db);
3543 let _ = tx.send(result);
3544 }
3545 SsoRequest::CleanupExpiredSsoAuthStates { tx } => {
3546 let result = state
3547 .metastore
3548 .sso_ops()
3549 .cleanup_expired_sso_auth_states()
3550 .map_err(metastore_to_db);
3551 let _ = tx.send(result);
3552 }
3553 SsoRequest::CreatePendingRegistration {
3554 token,
3555 request_uri,
3556 provider,
3557 provider_user_id,
3558 provider_username,
3559 provider_email,
3560 provider_email_verified,
3561 tx,
3562 } => {
3563 let result = state
3564 .metastore
3565 .sso_ops()
3566 .create_pending_registration(
3567 &token,
3568 &request_uri,
3569 provider,
3570 &provider_user_id,
3571 provider_username.as_deref(),
3572 provider_email.as_deref(),
3573 provider_email_verified,
3574 )
3575 .map_err(metastore_to_db);
3576 let _ = tx.send(result);
3577 }
3578 SsoRequest::GetPendingRegistration { token, tx } => {
3579 let result = state
3580 .metastore
3581 .sso_ops()
3582 .get_pending_registration(&token)
3583 .map_err(metastore_to_db);
3584 let _ = tx.send(result);
3585 }
3586 SsoRequest::ConsumePendingRegistration { token, tx } => {
3587 let result = state
3588 .metastore
3589 .sso_ops()
3590 .consume_pending_registration(&token)
3591 .map_err(metastore_to_db);
3592 let _ = tx.send(result);
3593 }
3594 SsoRequest::CleanupExpiredPendingRegistrations { tx } => {
3595 let result = state
3596 .metastore
3597 .sso_ops()
3598 .cleanup_expired_pending_registrations()
3599 .map_err(metastore_to_db);
3600 let _ = tx.send(result);
3601 }
3602 }
3603}
3604
3605fn dispatch_session<S: StorageIO>(state: &HandlerState<S>, req: SessionRequest) {
3606 match req {
3607 SessionRequest::CreateSession { data, tx } => {
3608 let result = state
3609 .metastore
3610 .session_ops()
3611 .create_session(&data)
3612 .map_err(metastore_to_db);
3613 let _ = tx.send(result);
3614 }
3615 SessionRequest::GetSessionByAccessJti { access_jti, tx } => {
3616 let result = state
3617 .metastore
3618 .session_ops()
3619 .get_session_by_access_jti(&access_jti)
3620 .map_err(metastore_to_db);
3621 let _ = tx.send(result);
3622 }
3623 SessionRequest::GetSessionForRefresh { refresh_jti, tx } => {
3624 let result = state
3625 .metastore
3626 .session_ops()
3627 .get_session_for_refresh(&refresh_jti)
3628 .map_err(metastore_to_db);
3629 let _ = tx.send(result);
3630 }
3631 SessionRequest::DeleteSessionByAccessJti {
3632 access_jti,
3633 did,
3634 tx,
3635 } => {
3636 let result = state
3637 .metastore
3638 .session_ops()
3639 .delete_session_by_access_jti(&access_jti, &did)
3640 .map_err(metastore_to_db);
3641 let _ = tx.send(result);
3642 }
3643 SessionRequest::DeleteSessionById {
3644 session_id,
3645 did,
3646 tx,
3647 } => {
3648 let result = state
3649 .metastore
3650 .session_ops()
3651 .delete_session_by_id(session_id, &did)
3652 .map_err(metastore_to_db);
3653 let _ = tx.send(result);
3654 }
3655 SessionRequest::DeleteSessionsByDid { did, tx } => {
3656 let result = state
3657 .metastore
3658 .session_ops()
3659 .delete_sessions_by_did(&did)
3660 .map_err(metastore_to_db);
3661 let _ = tx.send(result);
3662 }
3663 SessionRequest::DeleteSessionsByDidExceptJti {
3664 did,
3665 except_jti,
3666 tx,
3667 } => {
3668 let result = state
3669 .metastore
3670 .session_ops()
3671 .delete_sessions_by_did_except_jti(&did, &except_jti)
3672 .map_err(metastore_to_db);
3673 let _ = tx.send(result);
3674 }
3675 SessionRequest::ListSessionsByDid { did, tx } => {
3676 let result = state
3677 .metastore
3678 .session_ops()
3679 .list_sessions_by_did(&did)
3680 .map_err(metastore_to_db);
3681 let _ = tx.send(result);
3682 }
3683 SessionRequest::GetSessionAccessJtiById {
3684 session_id,
3685 did,
3686 tx,
3687 } => {
3688 let result = state
3689 .metastore
3690 .session_ops()
3691 .get_session_access_jti_by_id(session_id, &did)
3692 .map_err(metastore_to_db);
3693 let _ = tx.send(result);
3694 }
3695 SessionRequest::DeleteSessionsByAppPassword {
3696 did,
3697 app_password_name,
3698 tx,
3699 } => {
3700 let result = state
3701 .metastore
3702 .session_ops()
3703 .delete_sessions_by_app_password(&did, &app_password_name)
3704 .map_err(metastore_to_db);
3705 let _ = tx.send(result);
3706 }
3707 SessionRequest::GetSessionJtisByAppPassword {
3708 did,
3709 app_password_name,
3710 tx,
3711 } => {
3712 let result = state
3713 .metastore
3714 .session_ops()
3715 .get_session_jtis_by_app_password(&did, &app_password_name)
3716 .map_err(metastore_to_db);
3717 let _ = tx.send(result);
3718 }
3719 SessionRequest::LookupRefreshGrace { refresh_jti, tx } => {
3720 let result = state
3721 .metastore
3722 .session_ops()
3723 .lookup_refresh_grace(&refresh_jti)
3724 .map_err(metastore_to_db);
3725 let _ = tx.send(result);
3726 }
3727 SessionRequest::ListAppPasswords { user_id, tx } => {
3728 let result = state
3729 .metastore
3730 .session_ops()
3731 .list_app_passwords(user_id)
3732 .map_err(metastore_to_db);
3733 let _ = tx.send(result);
3734 }
3735 SessionRequest::GetAppPasswordsForLogin { user_id, tx } => {
3736 let result = state
3737 .metastore
3738 .session_ops()
3739 .get_app_passwords_for_login(user_id)
3740 .map_err(metastore_to_db);
3741 let _ = tx.send(result);
3742 }
3743 SessionRequest::GetAppPasswordByName { user_id, name, tx } => {
3744 let result = state
3745 .metastore
3746 .session_ops()
3747 .get_app_password_by_name(user_id, &name)
3748 .map_err(metastore_to_db);
3749 let _ = tx.send(result);
3750 }
3751 SessionRequest::CreateAppPassword { data, tx } => {
3752 let result = state
3753 .metastore
3754 .session_ops()
3755 .create_app_password(&data)
3756 .map_err(metastore_to_db);
3757 let _ = tx.send(result);
3758 }
3759 SessionRequest::DeleteAppPassword { user_id, name, tx } => {
3760 let result = state
3761 .metastore
3762 .session_ops()
3763 .delete_app_password(user_id, &name)
3764 .map_err(metastore_to_db);
3765 let _ = tx.send(result);
3766 }
3767 SessionRequest::DeleteAppPasswordsByController {
3768 did,
3769 controller_did,
3770 tx,
3771 } => {
3772 let result = state
3773 .metastore
3774 .session_ops()
3775 .delete_app_passwords_by_controller(&did, &controller_did)
3776 .map_err(metastore_to_db);
3777 let _ = tx.send(result);
3778 }
3779 SessionRequest::GetLastReauthAt { did, tx } => {
3780 let result = state
3781 .metastore
3782 .session_ops()
3783 .get_last_reauth_at(&did)
3784 .map_err(metastore_to_db);
3785 let _ = tx.send(result);
3786 }
3787 SessionRequest::UpdateLastReauth { did, tx } => {
3788 let result = state
3789 .metastore
3790 .session_ops()
3791 .update_last_reauth(&did)
3792 .map_err(metastore_to_db);
3793 let _ = tx.send(result);
3794 }
3795 SessionRequest::GetSessionMfaStatus { did, tx } => {
3796 let result = state
3797 .metastore
3798 .session_ops()
3799 .get_session_mfa_status(&did)
3800 .map_err(metastore_to_db);
3801 let _ = tx.send(result);
3802 }
3803 SessionRequest::UpdateMfaVerified { did, tx } => {
3804 let result = state
3805 .metastore
3806 .session_ops()
3807 .update_mfa_verified(&did)
3808 .map_err(metastore_to_db);
3809 let _ = tx.send(result);
3810 }
3811 SessionRequest::GetAppPasswordHashesByDid { did, tx } => {
3812 let result = state
3813 .metastore
3814 .session_ops()
3815 .get_app_password_hashes_by_did(&did)
3816 .map_err(metastore_to_db);
3817 let _ = tx.send(result);
3818 }
3819 SessionRequest::RefreshSessionAtomic { data, tx } => {
3820 let result = state
3821 .metastore
3822 .session_ops()
3823 .refresh_session_atomic(&data)
3824 .map_err(metastore_to_db);
3825 let _ = tx.send(result);
3826 }
3827 }
3828}
3829
3830fn dispatch_infra<S: StorageIO>(state: &HandlerState<S>, req: InfraRequest) {
3831 match req {
3832 InfraRequest::EnqueueComms {
3833 user_id,
3834 channel,
3835 comms_type,
3836 recipient,
3837 subject,
3838 body,
3839 metadata,
3840 tx,
3841 } => {
3842 let result = state
3843 .metastore
3844 .infra_ops()
3845 .enqueue_comms(
3846 user_id,
3847 channel,
3848 comms_type,
3849 &recipient,
3850 subject.as_deref(),
3851 &body,
3852 metadata,
3853 )
3854 .map_err(metastore_to_db);
3855 let _ = tx.send(result);
3856 }
3857 InfraRequest::FetchPendingComms {
3858 now,
3859 batch_size,
3860 tx,
3861 } => {
3862 let result = state
3863 .metastore
3864 .infra_ops()
3865 .fetch_pending_comms(now, batch_size)
3866 .map_err(metastore_to_db);
3867 let _ = tx.send(result);
3868 }
3869 InfraRequest::MarkCommsSent { id, tx } => {
3870 let result = state
3871 .metastore
3872 .infra_ops()
3873 .mark_comms_sent(id)
3874 .map_err(metastore_to_db);
3875 let _ = tx.send(result);
3876 }
3877 InfraRequest::MarkCommsFailed { id, error, tx } => {
3878 let result = state
3879 .metastore
3880 .infra_ops()
3881 .mark_comms_failed(id, &error)
3882 .map_err(metastore_to_db);
3883 let _ = tx.send(result);
3884 }
3885 InfraRequest::MarkCommsFailedPermanent { id, error, tx } => {
3886 let result = state
3887 .metastore
3888 .infra_ops()
3889 .mark_comms_failed_permanent(id, &error)
3890 .map_err(metastore_to_db);
3891 let _ = tx.send(result);
3892 }
3893 InfraRequest::CreateInviteCode {
3894 code,
3895 use_count,
3896 for_account,
3897 tx,
3898 } => {
3899 let result = state
3900 .metastore
3901 .infra_ops()
3902 .create_invite_code(&code, use_count, for_account.as_ref())
3903 .map_err(metastore_to_db);
3904 let _ = tx.send(result);
3905 }
3906 InfraRequest::CreateInviteCodesBatch {
3907 codes,
3908 use_count,
3909 created_by_user,
3910 for_account,
3911 tx,
3912 } => {
3913 let result = state
3914 .metastore
3915 .infra_ops()
3916 .create_invite_codes_batch(&codes, use_count, created_by_user, for_account.as_ref())
3917 .map_err(metastore_to_db);
3918 let _ = tx.send(result);
3919 }
3920 InfraRequest::GetInviteCodeAvailableUses { code, tx } => {
3921 let result = state
3922 .metastore
3923 .infra_ops()
3924 .get_invite_code_available_uses(&code)
3925 .map_err(metastore_to_db);
3926 let _ = tx.send(result);
3927 }
3928 InfraRequest::ValidateInviteCode { code, tx } => {
3929 let result = state
3930 .metastore
3931 .infra_ops()
3932 .validate_invite_code(&code)
3933 .map(|_| ());
3934 let _ = tx.send(result);
3935 }
3936 InfraRequest::GetInviteCodesForAccount { for_account, tx } => {
3937 let result = state
3938 .metastore
3939 .infra_ops()
3940 .get_invite_codes_for_account(&for_account)
3941 .map_err(metastore_to_db);
3942 let _ = tx.send(result);
3943 }
3944 InfraRequest::GetInviteCodeUses { code, tx } => {
3945 let result = state
3946 .metastore
3947 .infra_ops()
3948 .get_invite_code_uses(&code)
3949 .map_err(metastore_to_db);
3950 let _ = tx.send(result);
3951 }
3952 InfraRequest::DisableInviteCodesByCode { codes, tx } => {
3953 let result = state
3954 .metastore
3955 .infra_ops()
3956 .disable_invite_codes_by_code(&codes)
3957 .map_err(metastore_to_db);
3958 let _ = tx.send(result);
3959 }
3960 InfraRequest::DisableInviteCodesByAccount { accounts, tx } => {
3961 let result = state
3962 .metastore
3963 .infra_ops()
3964 .disable_invite_codes_by_account(&accounts)
3965 .map_err(metastore_to_db);
3966 let _ = tx.send(result);
3967 }
3968 InfraRequest::ListInviteCodes {
3969 cursor,
3970 limit,
3971 sort,
3972 tx,
3973 } => {
3974 let result = state
3975 .metastore
3976 .infra_ops()
3977 .list_invite_codes(cursor.as_deref(), limit, sort)
3978 .map_err(metastore_to_db);
3979 let _ = tx.send(result);
3980 }
3981 InfraRequest::GetUserDidsByIds { user_ids, tx } => {
3982 let result = state
3983 .metastore
3984 .infra_ops()
3985 .get_user_dids_by_ids(&user_ids)
3986 .map_err(metastore_to_db);
3987 let _ = tx.send(result);
3988 }
3989 InfraRequest::GetInviteCodeUsesBatch { codes, tx } => {
3990 let result = state
3991 .metastore
3992 .infra_ops()
3993 .get_invite_code_uses_batch(&codes)
3994 .map_err(metastore_to_db);
3995 let _ = tx.send(result);
3996 }
3997 InfraRequest::GetInvitesCreatedByUser { user_id, tx } => {
3998 let result = state
3999 .metastore
4000 .infra_ops()
4001 .get_invites_created_by_user(user_id)
4002 .map_err(metastore_to_db);
4003 let _ = tx.send(result);
4004 }
4005 InfraRequest::GetInviteCodeInfo { code, tx } => {
4006 let result = state
4007 .metastore
4008 .infra_ops()
4009 .get_invite_code_info(&code)
4010 .map_err(metastore_to_db);
4011 let _ = tx.send(result);
4012 }
4013 InfraRequest::GetInviteCodesByUsers { user_ids, tx } => {
4014 let result = state
4015 .metastore
4016 .infra_ops()
4017 .get_invite_codes_by_users(&user_ids)
4018 .map_err(metastore_to_db);
4019 let _ = tx.send(result);
4020 }
4021 InfraRequest::GetInviteCodeUsedByUser { user_id, tx } => {
4022 let result = state
4023 .metastore
4024 .infra_ops()
4025 .get_invite_code_used_by_user(user_id)
4026 .map_err(metastore_to_db);
4027 let _ = tx.send(result);
4028 }
4029 InfraRequest::DeleteInviteCodeUsesByUser { user_id, tx } => {
4030 let result = state
4031 .metastore
4032 .infra_ops()
4033 .delete_invite_code_uses_by_user(user_id)
4034 .map_err(metastore_to_db);
4035 let _ = tx.send(result);
4036 }
4037 InfraRequest::DeleteInviteCodesByUser { user_id, tx } => {
4038 let result = state
4039 .metastore
4040 .infra_ops()
4041 .delete_invite_codes_by_user(user_id)
4042 .map_err(metastore_to_db);
4043 let _ = tx.send(result);
4044 }
4045 InfraRequest::ReserveSigningKey {
4046 did,
4047 public_key_did_key,
4048 private_key_bytes,
4049 expires_at,
4050 tx,
4051 } => {
4052 let result = state
4053 .metastore
4054 .infra_ops()
4055 .reserve_signing_key(
4056 did.as_ref(),
4057 &public_key_did_key,
4058 &private_key_bytes,
4059 expires_at,
4060 )
4061 .map_err(metastore_to_db);
4062 let _ = tx.send(result);
4063 }
4064 InfraRequest::GetReservedSigningKey {
4065 public_key_did_key,
4066 tx,
4067 } => {
4068 let result = state
4069 .metastore
4070 .infra_ops()
4071 .get_reserved_signing_key(&public_key_did_key)
4072 .map_err(metastore_to_db);
4073 let _ = tx.send(result);
4074 }
4075 InfraRequest::MarkSigningKeyUsed { key_id, tx } => {
4076 let result = state
4077 .metastore
4078 .infra_ops()
4079 .mark_signing_key_used(key_id)
4080 .map_err(metastore_to_db);
4081 let _ = tx.send(result);
4082 }
4083 InfraRequest::CreateDeletionRequest {
4084 token,
4085 did,
4086 expires_at,
4087 tx,
4088 } => {
4089 let result = state
4090 .metastore
4091 .infra_ops()
4092 .create_deletion_request(&token, &did, expires_at)
4093 .map_err(metastore_to_db);
4094 let _ = tx.send(result);
4095 }
4096 InfraRequest::GetDeletionRequest { token, tx } => {
4097 let result = state
4098 .metastore
4099 .infra_ops()
4100 .get_deletion_request(&token)
4101 .map_err(metastore_to_db);
4102 let _ = tx.send(result);
4103 }
4104 InfraRequest::DeleteDeletionRequest { token, tx } => {
4105 let result = state
4106 .metastore
4107 .infra_ops()
4108 .delete_deletion_request(&token)
4109 .map_err(metastore_to_db);
4110 let _ = tx.send(result);
4111 }
4112 InfraRequest::DeleteDeletionRequestsByDid { did, tx } => {
4113 let result = state
4114 .metastore
4115 .infra_ops()
4116 .delete_deletion_requests_by_did(&did)
4117 .map_err(metastore_to_db);
4118 let _ = tx.send(result);
4119 }
4120 InfraRequest::UpsertAccountPreference {
4121 user_id,
4122 name,
4123 value_json,
4124 tx,
4125 } => {
4126 if name == "email_auth_factor" {
4127 let enabled = value_json.as_bool().unwrap_or(false);
4128 let _ = state
4129 .metastore
4130 .user_ops()
4131 .set_email_2fa_enabled(user_id, enabled);
4132 }
4133 let result = state
4134 .metastore
4135 .infra_ops()
4136 .upsert_account_preference(user_id, &name, value_json)
4137 .map_err(metastore_to_db);
4138 let _ = tx.send(result);
4139 }
4140 InfraRequest::InsertAccountPreferenceIfNotExists {
4141 user_id,
4142 name,
4143 value_json,
4144 tx,
4145 } => {
4146 let result = state
4147 .metastore
4148 .infra_ops()
4149 .insert_account_preference_if_not_exists(user_id, &name, value_json)
4150 .map_err(metastore_to_db);
4151 let _ = tx.send(result);
4152 }
4153 InfraRequest::GetServerConfig { key, tx } => {
4154 let result = state
4155 .metastore
4156 .infra_ops()
4157 .get_server_config(&key)
4158 .map_err(metastore_to_db);
4159 let _ = tx.send(result);
4160 }
4161 InfraRequest::InsertReport {
4162 id,
4163 reason_type,
4164 reason,
4165 subject_json,
4166 reported_by_did,
4167 created_at,
4168 tx,
4169 } => {
4170 let result = state
4171 .metastore
4172 .infra_ops()
4173 .insert_report(
4174 id,
4175 &reason_type,
4176 reason.as_deref(),
4177 subject_json,
4178 &reported_by_did,
4179 created_at,
4180 )
4181 .map_err(metastore_to_db);
4182 let _ = tx.send(result);
4183 }
4184 InfraRequest::DeletePlcTokensForUser { user_id, tx } => {
4185 let result = state
4186 .metastore
4187 .infra_ops()
4188 .delete_plc_tokens_for_user(user_id)
4189 .map_err(metastore_to_db);
4190 let _ = tx.send(result);
4191 }
4192 InfraRequest::InsertPlcToken {
4193 user_id,
4194 token,
4195 expires_at,
4196 tx,
4197 } => {
4198 let result = state
4199 .metastore
4200 .infra_ops()
4201 .insert_plc_token(user_id, &token, expires_at)
4202 .map_err(metastore_to_db);
4203 let _ = tx.send(result);
4204 }
4205 InfraRequest::GetPlcTokenExpiry { user_id, token, tx } => {
4206 let result = state
4207 .metastore
4208 .infra_ops()
4209 .get_plc_token_expiry(user_id, &token)
4210 .map_err(metastore_to_db);
4211 let _ = tx.send(result);
4212 }
4213 InfraRequest::DeletePlcToken { user_id, token, tx } => {
4214 let result = state
4215 .metastore
4216 .infra_ops()
4217 .delete_plc_token(user_id, &token)
4218 .map_err(metastore_to_db);
4219 let _ = tx.send(result);
4220 }
4221 InfraRequest::GetAccountPreferences { user_id, tx } => {
4222 let result = state
4223 .metastore
4224 .infra_ops()
4225 .get_account_preferences(user_id)
4226 .map_err(metastore_to_db);
4227 let _ = tx.send(result);
4228 }
4229 InfraRequest::ReplaceNamespacePreferences {
4230 user_id,
4231 namespace,
4232 preferences,
4233 tx,
4234 } => {
4235 let result = state
4236 .metastore
4237 .infra_ops()
4238 .replace_namespace_preferences(user_id, &namespace, preferences)
4239 .map_err(metastore_to_db);
4240 let _ = tx.send(result);
4241 }
4242 InfraRequest::GetNotificationHistory { user_id, limit, tx } => {
4243 let result = state
4244 .metastore
4245 .infra_ops()
4246 .get_notification_history(user_id, limit)
4247 .map_err(metastore_to_db);
4248 let _ = tx.send(result);
4249 }
4250 InfraRequest::GetServerConfigs { keys, tx } => {
4251 let key_refs: Vec<&str> = keys.iter().map(String::as_str).collect();
4252 let result = state
4253 .metastore
4254 .infra_ops()
4255 .get_server_configs(&key_refs)
4256 .map_err(metastore_to_db);
4257 let _ = tx.send(result);
4258 }
4259 InfraRequest::UpsertServerConfig { key, value, tx } => {
4260 let result = state
4261 .metastore
4262 .infra_ops()
4263 .upsert_server_config(&key, &value)
4264 .map_err(metastore_to_db);
4265 let _ = tx.send(result);
4266 }
4267 InfraRequest::DeleteServerConfig { key, tx } => {
4268 let result = state
4269 .metastore
4270 .infra_ops()
4271 .delete_server_config(&key)
4272 .map_err(metastore_to_db);
4273 let _ = tx.send(result);
4274 }
4275 InfraRequest::GetBlobStorageKeyByCid { cid, tx } => {
4276 let result = state
4277 .metastore
4278 .infra_ops()
4279 .get_blob_storage_key_by_cid(&cid)
4280 .map_err(metastore_to_db);
4281 let _ = tx.send(result);
4282 }
4283 InfraRequest::DeleteBlobByCid { cid, tx } => {
4284 let result = state
4285 .metastore
4286 .infra_ops()
4287 .delete_blob_by_cid(&cid)
4288 .map_err(metastore_to_db);
4289 let _ = tx.send(result);
4290 }
4291 InfraRequest::GetAdminAccountInfoByDid { did, tx } => {
4292 let result = state
4293 .metastore
4294 .infra_ops()
4295 .get_admin_account_info_by_did(&did)
4296 .map_err(metastore_to_db);
4297 let _ = tx.send(result);
4298 }
4299 InfraRequest::GetAdminAccountInfosByDids { dids, tx } => {
4300 let result = state
4301 .metastore
4302 .infra_ops()
4303 .get_admin_account_infos_by_dids(&dids)
4304 .map_err(metastore_to_db);
4305 let _ = tx.send(result);
4306 }
4307 InfraRequest::GetInviteCodeUsesByUsers { user_ids, tx } => {
4308 let result = state
4309 .metastore
4310 .infra_ops()
4311 .get_invite_code_uses_by_users(&user_ids)
4312 .map_err(metastore_to_db);
4313 let _ = tx.send(result);
4314 }
4315 InfraRequest::GetDeletionRequestByDid { did, tx } => {
4316 let result = state
4317 .metastore
4318 .infra_ops()
4319 .get_deletion_request_by_did(&did)
4320 .map_err(metastore_to_db);
4321 let _ = tx.send(result);
4322 }
4323 InfraRequest::GetLatestCommsForUser {
4324 user_id,
4325 comms_type,
4326 limit,
4327 tx,
4328 } => {
4329 let result = state
4330 .metastore
4331 .infra_ops()
4332 .get_latest_comms_for_user(user_id, comms_type, limit)
4333 .map_err(metastore_to_db);
4334 let _ = tx.send(result);
4335 }
4336 InfraRequest::CountCommsByType {
4337 user_id,
4338 comms_type,
4339 tx,
4340 } => {
4341 let result = state
4342 .metastore
4343 .infra_ops()
4344 .count_comms_by_type(user_id, comms_type)
4345 .map_err(metastore_to_db);
4346 let _ = tx.send(result);
4347 }
4348 InfraRequest::DeleteCommsByTypeForUser {
4349 user_id,
4350 comms_type,
4351 tx,
4352 } => {
4353 let result = state
4354 .metastore
4355 .infra_ops()
4356 .delete_comms_by_type_for_user(user_id, comms_type)
4357 .map_err(metastore_to_db);
4358 let _ = tx.send(result);
4359 }
4360 InfraRequest::ExpireDeletionRequest { token, tx } => {
4361 let result = state
4362 .metastore
4363 .infra_ops()
4364 .expire_deletion_request(&token)
4365 .map_err(metastore_to_db);
4366 let _ = tx.send(result);
4367 }
4368 InfraRequest::GetReservedSigningKeyFull {
4369 public_key_did_key,
4370 tx,
4371 } => {
4372 let result = state
4373 .metastore
4374 .infra_ops()
4375 .get_reserved_signing_key_full(&public_key_did_key)
4376 .map_err(metastore_to_db);
4377 let _ = tx.send(result);
4378 }
4379 InfraRequest::GetPlcTokensByDid { did, tx } => {
4380 let result = (|| {
4381 let user_id = state
4382 .metastore
4383 .user_ops()
4384 .get_id_by_did(&did)
4385 .map_err(metastore_to_db)?;
4386 match user_id {
4387 Some(uid) => state
4388 .metastore
4389 .infra_ops()
4390 .get_plc_tokens_for_user(uid)
4391 .map_err(metastore_to_db),
4392 None => Ok(Vec::new()),
4393 }
4394 })();
4395 let _ = tx.send(result);
4396 }
4397 InfraRequest::CountPlcTokensByDid { did, tx } => {
4398 let result = (|| {
4399 let user_id = state
4400 .metastore
4401 .user_ops()
4402 .get_id_by_did(&did)
4403 .map_err(metastore_to_db)?;
4404 match user_id {
4405 Some(uid) => state
4406 .metastore
4407 .infra_ops()
4408 .count_plc_tokens_for_user(uid)
4409 .map_err(metastore_to_db),
4410 None => Ok(0),
4411 }
4412 })();
4413 let _ = tx.send(result);
4414 }
4415 }
4416}
4417
4418fn dispatch_oauth<S: StorageIO>(state: &HandlerState<S>, req: OAuthRequest) {
4419 match req {
4420 OAuthRequest::CreateToken { data, tx } => {
4421 let result = state
4422 .metastore
4423 .oauth_ops()
4424 .create_token(&data)
4425 .map_err(metastore_to_db);
4426 let _ = tx.send(result);
4427 }
4428 OAuthRequest::GetTokenById { token_id, tx } => {
4429 let result = state
4430 .metastore
4431 .oauth_ops()
4432 .get_token_by_id(&token_id)
4433 .map_err(metastore_to_db);
4434 let _ = tx.send(result);
4435 }
4436 OAuthRequest::GetTokenByRefreshToken { refresh_token, tx } => {
4437 let result = state
4438 .metastore
4439 .oauth_ops()
4440 .get_token_by_refresh_token(&refresh_token)
4441 .map_err(metastore_to_db);
4442 let _ = tx.send(result);
4443 }
4444 OAuthRequest::GetTokenByPreviousRefreshToken { refresh_token, tx } => {
4445 let result = state
4446 .metastore
4447 .oauth_ops()
4448 .get_token_by_previous_refresh_token(&refresh_token)
4449 .map_err(metastore_to_db);
4450 let _ = tx.send(result);
4451 }
4452 OAuthRequest::RotateToken {
4453 old_db_id,
4454 new_refresh_token,
4455 new_expires_at,
4456 tx,
4457 } => {
4458 let result = state
4459 .metastore
4460 .oauth_ops()
4461 .rotate_token(old_db_id, &new_refresh_token, new_expires_at)
4462 .map_err(metastore_to_db);
4463 let _ = tx.send(result);
4464 }
4465 OAuthRequest::CheckRefreshTokenUsed { refresh_token, tx } => {
4466 let result = state
4467 .metastore
4468 .oauth_ops()
4469 .check_refresh_token_used(&refresh_token)
4470 .map_err(metastore_to_db);
4471 let _ = tx.send(result);
4472 }
4473 OAuthRequest::DeleteToken { token_id, tx } => {
4474 let result = state
4475 .metastore
4476 .oauth_ops()
4477 .delete_token(&token_id)
4478 .map_err(metastore_to_db);
4479 let _ = tx.send(result);
4480 }
4481 OAuthRequest::DeleteTokenFamily { db_id, tx } => {
4482 let result = state
4483 .metastore
4484 .oauth_ops()
4485 .delete_token_family(db_id)
4486 .map_err(metastore_to_db);
4487 let _ = tx.send(result);
4488 }
4489 OAuthRequest::ListTokensForUser { did, tx } => {
4490 let result = state
4491 .metastore
4492 .oauth_ops()
4493 .list_tokens_for_user(&did)
4494 .map_err(metastore_to_db);
4495 let _ = tx.send(result);
4496 }
4497 OAuthRequest::CountTokensForUser { did, tx } => {
4498 let result = state
4499 .metastore
4500 .oauth_ops()
4501 .count_tokens_for_user(&did)
4502 .map_err(metastore_to_db);
4503 let _ = tx.send(result);
4504 }
4505 OAuthRequest::DeleteOldestTokensForUser {
4506 did,
4507 keep_count,
4508 tx,
4509 } => {
4510 let result = state
4511 .metastore
4512 .oauth_ops()
4513 .delete_oldest_tokens_for_user(&did, keep_count)
4514 .map_err(metastore_to_db);
4515 let _ = tx.send(result);
4516 }
4517 OAuthRequest::RevokeTokensForClient { did, client_id, tx } => {
4518 let result = state
4519 .metastore
4520 .oauth_ops()
4521 .revoke_tokens_for_client(&did, &client_id)
4522 .map_err(metastore_to_db);
4523 let _ = tx.send(result);
4524 }
4525 OAuthRequest::RevokeTokensForController {
4526 delegated_did,
4527 controller_did,
4528 tx,
4529 } => {
4530 let result = state
4531 .metastore
4532 .oauth_ops()
4533 .revoke_tokens_for_controller(&delegated_did, &controller_did)
4534 .map_err(metastore_to_db);
4535 let _ = tx.send(result);
4536 }
4537 OAuthRequest::CreateAuthorizationRequest {
4538 request_id,
4539 data,
4540 tx,
4541 } => {
4542 let result = state
4543 .metastore
4544 .oauth_ops()
4545 .create_authorization_request(&request_id, &data)
4546 .map_err(metastore_to_db);
4547 let _ = tx.send(result);
4548 }
4549 OAuthRequest::GetAuthorizationRequest { request_id, tx } => {
4550 let result = state
4551 .metastore
4552 .oauth_ops()
4553 .get_authorization_request(&request_id)
4554 .map_err(metastore_to_db);
4555 let _ = tx.send(result);
4556 }
4557 OAuthRequest::SetAuthorizationDid {
4558 request_id,
4559 did,
4560 device_id,
4561 tx,
4562 } => {
4563 let result = state
4564 .metastore
4565 .oauth_ops()
4566 .set_authorization_did(&request_id, &did, device_id.as_ref())
4567 .map_err(metastore_to_db);
4568 let _ = tx.send(result);
4569 }
4570 OAuthRequest::UpdateAuthorizationRequest {
4571 request_id,
4572 did,
4573 device_id,
4574 code,
4575 tx,
4576 } => {
4577 let result = state
4578 .metastore
4579 .oauth_ops()
4580 .update_authorization_request(&request_id, &did, device_id.as_ref(), &code)
4581 .map_err(metastore_to_db);
4582 let _ = tx.send(result);
4583 }
4584 OAuthRequest::ConsumeAuthorizationRequestByCode { code, tx } => {
4585 let result = state
4586 .metastore
4587 .oauth_ops()
4588 .consume_authorization_request_by_code(&code)
4589 .map_err(metastore_to_db);
4590 let _ = tx.send(result);
4591 }
4592 OAuthRequest::DeleteAuthorizationRequest { request_id, tx } => {
4593 let result = state
4594 .metastore
4595 .oauth_ops()
4596 .delete_authorization_request(&request_id)
4597 .map_err(metastore_to_db);
4598 let _ = tx.send(result);
4599 }
4600 OAuthRequest::DeleteExpiredAuthorizationRequests { tx } => {
4601 let result = state
4602 .metastore
4603 .oauth_ops()
4604 .delete_expired_authorization_requests()
4605 .map_err(metastore_to_db);
4606 let _ = tx.send(result);
4607 }
4608 OAuthRequest::ExtendAuthorizationRequestExpiry {
4609 request_id,
4610 new_expires_at,
4611 tx,
4612 } => {
4613 let result = state
4614 .metastore
4615 .oauth_ops()
4616 .extend_authorization_request_expiry(&request_id, new_expires_at)
4617 .map_err(metastore_to_db);
4618 let _ = tx.send(result);
4619 }
4620 OAuthRequest::MarkRequestAuthenticated {
4621 request_id,
4622 did,
4623 device_id,
4624 tx,
4625 } => {
4626 let result = state
4627 .metastore
4628 .oauth_ops()
4629 .mark_request_authenticated(&request_id, &did, device_id.as_ref())
4630 .map_err(metastore_to_db);
4631 let _ = tx.send(result);
4632 }
4633 OAuthRequest::UpdateRequestScope {
4634 request_id,
4635 scope,
4636 tx,
4637 } => {
4638 let result = state
4639 .metastore
4640 .oauth_ops()
4641 .update_request_scope(&request_id, &scope)
4642 .map_err(metastore_to_db);
4643 let _ = tx.send(result);
4644 }
4645 OAuthRequest::SetControllerDid {
4646 request_id,
4647 controller_did,
4648 tx,
4649 } => {
4650 let result = state
4651 .metastore
4652 .oauth_ops()
4653 .set_controller_did(&request_id, &controller_did)
4654 .map_err(metastore_to_db);
4655 let _ = tx.send(result);
4656 }
4657 OAuthRequest::SetRequestDid {
4658 request_id,
4659 did,
4660 tx,
4661 } => {
4662 let result = state
4663 .metastore
4664 .oauth_ops()
4665 .set_request_did(&request_id, &did)
4666 .map_err(metastore_to_db);
4667 let _ = tx.send(result);
4668 }
4669 OAuthRequest::CreateDevice {
4670 device_id,
4671 data,
4672 tx,
4673 } => {
4674 let result = state
4675 .metastore
4676 .oauth_ops()
4677 .create_device(&device_id, &data)
4678 .map_err(metastore_to_db);
4679 let _ = tx.send(result);
4680 }
4681 OAuthRequest::GetDevice { device_id, tx } => {
4682 let result = state
4683 .metastore
4684 .oauth_ops()
4685 .get_device(&device_id)
4686 .map_err(metastore_to_db);
4687 let _ = tx.send(result);
4688 }
4689 OAuthRequest::UpdateDeviceLastSeen { device_id, tx } => {
4690 let result = state
4691 .metastore
4692 .oauth_ops()
4693 .update_device_last_seen(&device_id)
4694 .map_err(metastore_to_db);
4695 let _ = tx.send(result);
4696 }
4697 OAuthRequest::DeleteDevice { device_id, tx } => {
4698 let result = state
4699 .metastore
4700 .oauth_ops()
4701 .delete_device(&device_id)
4702 .map_err(metastore_to_db);
4703 let _ = tx.send(result);
4704 }
4705 OAuthRequest::UpsertAccountDevice { did, device_id, tx } => {
4706 let result = state
4707 .metastore
4708 .oauth_ops()
4709 .upsert_account_device(&did, &device_id)
4710 .map_err(metastore_to_db);
4711 let _ = tx.send(result);
4712 }
4713 OAuthRequest::GetDeviceAccounts { device_id, tx } => {
4714 let result = state
4715 .metastore
4716 .oauth_ops()
4717 .get_device_accounts(&device_id)
4718 .map_err(metastore_to_db);
4719 let _ = tx.send(result);
4720 }
4721 OAuthRequest::VerifyAccountOnDevice { device_id, did, tx } => {
4722 let result = state
4723 .metastore
4724 .oauth_ops()
4725 .verify_account_on_device(&device_id, &did)
4726 .map_err(metastore_to_db);
4727 let _ = tx.send(result);
4728 }
4729 OAuthRequest::CheckAndRecordDpopJti { jti, tx } => {
4730 let result = state
4731 .metastore
4732 .oauth_ops()
4733 .check_and_record_dpop_jti(&jti)
4734 .map_err(metastore_to_db);
4735 let _ = tx.send(result);
4736 }
4737 OAuthRequest::CleanupExpiredDpopJtis { max_age_secs, tx } => {
4738 let result = state
4739 .metastore
4740 .oauth_ops()
4741 .cleanup_expired_dpop_jtis(max_age_secs)
4742 .map_err(metastore_to_db);
4743 let _ = tx.send(result);
4744 }
4745 OAuthRequest::Create2faChallenge {
4746 did,
4747 request_uri,
4748 tx,
4749 } => {
4750 let result = state
4751 .metastore
4752 .oauth_ops()
4753 .create_2fa_challenge(&did, &request_uri)
4754 .map_err(metastore_to_db);
4755 let _ = tx.send(result);
4756 }
4757 OAuthRequest::Get2faChallenge { request_uri, tx } => {
4758 let result = state
4759 .metastore
4760 .oauth_ops()
4761 .get_2fa_challenge(&request_uri)
4762 .map_err(metastore_to_db);
4763 let _ = tx.send(result);
4764 }
4765 OAuthRequest::Increment2faAttempts { id, tx } => {
4766 let result = state
4767 .metastore
4768 .oauth_ops()
4769 .increment_2fa_attempts(id)
4770 .map_err(metastore_to_db);
4771 let _ = tx.send(result);
4772 }
4773 OAuthRequest::Delete2faChallenge { id, tx } => {
4774 let result = state
4775 .metastore
4776 .oauth_ops()
4777 .delete_2fa_challenge(id)
4778 .map_err(metastore_to_db);
4779 let _ = tx.send(result);
4780 }
4781 OAuthRequest::Delete2faChallengeByRequestUri { request_uri, tx } => {
4782 let result = state
4783 .metastore
4784 .oauth_ops()
4785 .delete_2fa_challenge_by_request_uri(&request_uri)
4786 .map_err(metastore_to_db);
4787 let _ = tx.send(result);
4788 }
4789 OAuthRequest::CleanupExpired2faChallenges { tx } => {
4790 let result = state
4791 .metastore
4792 .oauth_ops()
4793 .cleanup_expired_2fa_challenges()
4794 .map_err(metastore_to_db);
4795 let _ = tx.send(result);
4796 }
4797 OAuthRequest::CheckUser2faEnabled { did, tx } => {
4798 let result = state
4799 .metastore
4800 .oauth_ops()
4801 .check_user_2fa_enabled(&did)
4802 .map_err(metastore_to_db);
4803 let _ = tx.send(result);
4804 }
4805 OAuthRequest::GetScopePreferences { did, client_id, tx } => {
4806 let result = state
4807 .metastore
4808 .oauth_ops()
4809 .get_scope_preferences(&did, &client_id)
4810 .map_err(metastore_to_db);
4811 let _ = tx.send(result);
4812 }
4813 OAuthRequest::UpsertScopePreferences {
4814 did,
4815 client_id,
4816 prefs,
4817 tx,
4818 } => {
4819 let result = state
4820 .metastore
4821 .oauth_ops()
4822 .upsert_scope_preferences(&did, &client_id, &prefs)
4823 .map_err(metastore_to_db);
4824 let _ = tx.send(result);
4825 }
4826 OAuthRequest::DeleteScopePreferences { did, client_id, tx } => {
4827 let result = state
4828 .metastore
4829 .oauth_ops()
4830 .delete_scope_preferences(&did, &client_id)
4831 .map_err(metastore_to_db);
4832 let _ = tx.send(result);
4833 }
4834 OAuthRequest::UpsertAuthorizedClient {
4835 did,
4836 client_id,
4837 data,
4838 tx,
4839 } => {
4840 let result = state
4841 .metastore
4842 .oauth_ops()
4843 .upsert_authorized_client(&did, &client_id, &data)
4844 .map_err(metastore_to_db);
4845 let _ = tx.send(result);
4846 }
4847 OAuthRequest::GetAuthorizedClient { did, client_id, tx } => {
4848 let result = state
4849 .metastore
4850 .oauth_ops()
4851 .get_authorized_client(&did, &client_id)
4852 .map_err(metastore_to_db);
4853 let _ = tx.send(result);
4854 }
4855 OAuthRequest::ListTrustedDevices { did, tx } => {
4856 let result = state
4857 .metastore
4858 .oauth_ops()
4859 .list_trusted_devices(&did)
4860 .map_err(metastore_to_db);
4861 let _ = tx.send(result);
4862 }
4863 OAuthRequest::GetDeviceTrustInfo { device_id, did, tx } => {
4864 let result = state
4865 .metastore
4866 .oauth_ops()
4867 .get_device_trust_info(&device_id, &did)
4868 .map_err(metastore_to_db);
4869 let _ = tx.send(result);
4870 }
4871 OAuthRequest::DeviceBelongsToUser { device_id, did, tx } => {
4872 let result = state
4873 .metastore
4874 .oauth_ops()
4875 .device_belongs_to_user(&device_id, &did)
4876 .map_err(metastore_to_db);
4877 let _ = tx.send(result);
4878 }
4879 OAuthRequest::RevokeDeviceTrust { device_id, did, tx } => {
4880 let result = state
4881 .metastore
4882 .oauth_ops()
4883 .revoke_device_trust(&device_id, &did)
4884 .map_err(metastore_to_db);
4885 let _ = tx.send(result);
4886 }
4887 OAuthRequest::UpdateDeviceFriendlyName {
4888 device_id,
4889 did,
4890 friendly_name,
4891 tx,
4892 } => {
4893 let result = state
4894 .metastore
4895 .oauth_ops()
4896 .update_device_friendly_name(&device_id, &did, friendly_name.as_deref())
4897 .map_err(metastore_to_db);
4898 let _ = tx.send(result);
4899 }
4900 OAuthRequest::TrustDevice {
4901 device_id,
4902 did,
4903 trusted_at,
4904 trusted_until,
4905 tx,
4906 } => {
4907 let result = state
4908 .metastore
4909 .oauth_ops()
4910 .trust_device(&device_id, &did, trusted_at, trusted_until)
4911 .map_err(metastore_to_db);
4912 let _ = tx.send(result);
4913 }
4914 OAuthRequest::ExtendDeviceTrust {
4915 device_id,
4916 did,
4917 trusted_until,
4918 tx,
4919 } => {
4920 let result = state
4921 .metastore
4922 .oauth_ops()
4923 .extend_device_trust(&device_id, &did, trusted_until)
4924 .map_err(metastore_to_db);
4925 let _ = tx.send(result);
4926 }
4927 OAuthRequest::ListSessionsByDid { did, tx } => {
4928 let result = state
4929 .metastore
4930 .oauth_ops()
4931 .list_sessions_by_did(&did)
4932 .map_err(metastore_to_db);
4933 let _ = tx.send(result);
4934 }
4935 OAuthRequest::DeleteSessionById {
4936 session_id,
4937 did,
4938 tx,
4939 } => {
4940 let result = state
4941 .metastore
4942 .oauth_ops()
4943 .delete_session_by_id(session_id, &did)
4944 .map_err(metastore_to_db);
4945 let _ = tx.send(result);
4946 }
4947 OAuthRequest::DeleteSessionsByDid { did, tx } => {
4948 let result = state
4949 .metastore
4950 .oauth_ops()
4951 .delete_sessions_by_did(&did)
4952 .map_err(metastore_to_db);
4953 let _ = tx.send(result);
4954 }
4955 OAuthRequest::DeleteSessionsByDidExcept {
4956 did,
4957 except_token_id,
4958 tx,
4959 } => {
4960 let result = state
4961 .metastore
4962 .oauth_ops()
4963 .delete_sessions_by_did_except(&did, &except_token_id)
4964 .map_err(metastore_to_db);
4965 let _ = tx.send(result);
4966 }
4967 OAuthRequest::Get2faChallengeCode { request_uri, tx } => {
4968 let result = state
4969 .metastore
4970 .oauth_ops()
4971 .get_2fa_challenge_code(&request_uri)
4972 .map_err(metastore_to_db);
4973 let _ = tx.send(result);
4974 }
4975 }
4976}
4977
4978fn dispatch<S: StorageIO + 'static>(state: &HandlerState<S>, request: MetastoreRequest) {
4979 match request {
4980 MetastoreRequest::Repo(r) => dispatch_repo(state, r),
4981 MetastoreRequest::Record(r) => dispatch_record(state, r),
4982 MetastoreRequest::UserBlock(r) => dispatch_user_block(state, r),
4983 MetastoreRequest::Event(r) => dispatch_event(state, r),
4984 MetastoreRequest::Commit(r) => dispatch_commit(state, *r),
4985 MetastoreRequest::Backlink(r) => dispatch_backlink(state, r),
4986 MetastoreRequest::Blob(r) => dispatch_blob(state, r),
4987 MetastoreRequest::Delegation(r) => dispatch_delegation(state, r),
4988 MetastoreRequest::Sso(r) => dispatch_sso(state, r),
4989 MetastoreRequest::Session(r) => dispatch_session(state, r),
4990 MetastoreRequest::Infra(r) => dispatch_infra(state, r),
4991 MetastoreRequest::OAuth(r) => dispatch_oauth(state, r),
4992 MetastoreRequest::User(r) => dispatch_user(state, r),
4993 }
4994}
4995
4996fn purge_repo_side_data<S: StorageIO + 'static>(
4997 state: &HandlerState<S>,
4998 user_id: Uuid,
4999 did: &Did,
5000) -> Result<(), MetastoreError> {
5001 let _ = state.metastore.blob_ops().delete_blobs_by_user(user_id)?;
5002 let mut batch = state.metastore.database().batch();
5003 state
5004 .metastore
5005 .backlink_ops()
5006 .remove_backlinks_by_repo(&mut batch, UserHash::from_did(did.as_str()))?;
5007 batch.commit().map_err(MetastoreError::Fjall)
5008}
5009
5010fn dispatch_user<S: StorageIO + 'static>(state: &HandlerState<S>, req: UserRequest) {
5011 let user = state.metastore.user_ops();
5012 match req {
5013 UserRequest::GetByDid { did, tx } => {
5014 let _ = tx.send(user.get_by_did(&did).map_err(metastore_to_db));
5015 }
5016 UserRequest::GetByHandle { handle, tx } => {
5017 let _ = tx.send(user.get_by_handle(&handle).map_err(metastore_to_db));
5018 }
5019 UserRequest::GetWithKeyByDid { did, tx } => {
5020 let _ = tx.send(user.get_with_key_by_did(&did).map_err(metastore_to_db));
5021 }
5022 UserRequest::GetStatusByDid { did, tx } => {
5023 let _ = tx.send(user.get_status_by_did(&did).map_err(metastore_to_db));
5024 }
5025 UserRequest::CountUsers { tx } => {
5026 let _ = tx.send(user.count_users().map_err(metastore_to_db));
5027 }
5028 UserRequest::GetSessionAccessExpiry {
5029 did,
5030 access_jti,
5031 tx,
5032 } => {
5033 let _ = tx.send(
5034 user.get_session_access_expiry(&did, &access_jti)
5035 .map_err(metastore_to_db),
5036 );
5037 }
5038 UserRequest::GetOAuthTokenWithUser { token_id, tx } => {
5039 let _ = tx.send(
5040 user.get_oauth_token_with_user(&token_id)
5041 .map_err(metastore_to_db),
5042 );
5043 }
5044 UserRequest::GetUserInfoByDid { did, tx } => {
5045 let _ = tx.send(user.get_user_info_by_did(&did).map_err(metastore_to_db));
5046 }
5047 UserRequest::GetAnyAdminUserId { tx } => {
5048 let _ = tx.send(user.get_any_admin_user_id().map_err(metastore_to_db));
5049 }
5050 UserRequest::SetInvitesDisabled { did, disabled, tx } => {
5051 let _ = tx.send(
5052 user.set_invites_disabled(&did, disabled)
5053 .map_err(metastore_to_db),
5054 );
5055 }
5056 UserRequest::SearchAccounts {
5057 cursor_did,
5058 email_filter,
5059 handle_filter,
5060 limit,
5061 tx,
5062 } => {
5063 let _ = tx.send(
5064 user.search_accounts(
5065 cursor_did.as_ref(),
5066 email_filter.as_deref(),
5067 handle_filter.as_deref(),
5068 limit,
5069 )
5070 .map_err(metastore_to_db),
5071 );
5072 }
5073 UserRequest::GetAuthInfoByDid { did, tx } => {
5074 let _ = tx.send(user.get_auth_info_by_did(&did).map_err(metastore_to_db));
5075 }
5076 UserRequest::GetByEmail { email, tx } => {
5077 let _ = tx.send(user.get_by_email(&email).map_err(metastore_to_db));
5078 }
5079 UserRequest::GetLoginCheckByIdentifier { identifier, tx } => {
5080 let _ = tx.send(
5081 user.get_login_check_by_identifier(&identifier)
5082 .map_err(metastore_to_db),
5083 );
5084 }
5085 UserRequest::GetLoginInfoByIdentifier { identifier, tx } => {
5086 let _ = tx.send(
5087 user.get_login_info_by_identifier(&identifier)
5088 .map_err(metastore_to_db),
5089 );
5090 }
5091 UserRequest::Get2faStatusByDid { did, tx } => {
5092 let _ = tx.send(user.get_2fa_status_by_did(&did).map_err(metastore_to_db));
5093 }
5094 UserRequest::GetCommsPrefs { user_id, tx } => {
5095 let _ = tx.send(user.get_comms_prefs(user_id).map_err(metastore_to_db));
5096 }
5097 UserRequest::GetIdByDid { did, tx } => {
5098 let _ = tx.send(user.get_id_by_did(&did).map_err(metastore_to_db));
5099 }
5100 UserRequest::GetUserKeyById { user_id, tx } => {
5101 let _ = tx.send(user.get_user_key_by_id(user_id).map_err(metastore_to_db));
5102 }
5103 UserRequest::GetIdAndHandleByDid { did, tx } => {
5104 let _ = tx.send(user.get_id_and_handle_by_did(&did).map_err(metastore_to_db));
5105 }
5106 UserRequest::GetDidWebInfoByHandle { handle, tx } => {
5107 let _ = tx.send(
5108 user.get_did_web_info_by_handle(&handle)
5109 .map_err(metastore_to_db),
5110 );
5111 }
5112 UserRequest::GetDidWebOverrides { user_id, tx } => {
5113 let _ = tx.send(user.get_did_web_overrides(user_id).map_err(metastore_to_db));
5114 }
5115 UserRequest::GetHandleByDid { did, tx } => {
5116 let _ = tx.send(user.get_handle_by_did(&did).map_err(metastore_to_db));
5117 }
5118 UserRequest::IsAccountActiveByDid { did, tx } => {
5119 let _ = tx.send(user.is_account_active_by_did(&did).map_err(metastore_to_db));
5120 }
5121 UserRequest::GetUserForDeletion { did, tx } => {
5122 let _ = tx.send(user.get_user_for_deletion(&did).map_err(metastore_to_db));
5123 }
5124 UserRequest::CheckHandleExists {
5125 handle,
5126 exclude_user_id,
5127 tx,
5128 } => {
5129 let _ = tx.send(
5130 user.check_handle_exists(&handle, exclude_user_id)
5131 .map_err(metastore_to_db),
5132 );
5133 }
5134 UserRequest::UpdateHandle {
5135 user_id,
5136 handle,
5137 tx,
5138 } => {
5139 let _ = tx.send(
5140 user.update_handle(user_id, &handle)
5141 .map_err(metastore_to_db),
5142 );
5143 }
5144 UserRequest::GetUserWithKeyByDid { did, tx } => {
5145 let _ = tx.send(user.get_user_with_key_by_did(&did).map_err(metastore_to_db));
5146 }
5147 UserRequest::IsAccountMigrated { did, tx } => {
5148 let _ = tx.send(user.is_account_migrated(&did).map_err(metastore_to_db));
5149 }
5150 UserRequest::HasVerifiedCommsChannel { did, tx } => {
5151 let _ = tx.send(
5152 user.has_verified_comms_channel(&did)
5153 .map_err(metastore_to_db),
5154 );
5155 }
5156 UserRequest::GetIdByHandle { handle, tx } => {
5157 let _ = tx.send(user.get_id_by_handle(&handle).map_err(metastore_to_db));
5158 }
5159 UserRequest::GetEmailInfoByDid { did, tx } => {
5160 let _ = tx.send(user.get_email_info_by_did(&did).map_err(metastore_to_db));
5161 }
5162 UserRequest::CheckEmailExists {
5163 email,
5164 exclude_user_id,
5165 tx,
5166 } => {
5167 let _ = tx.send(
5168 user.check_email_exists(&email, exclude_user_id)
5169 .map_err(metastore_to_db),
5170 );
5171 }
5172 UserRequest::UpdateEmail { user_id, email, tx } => {
5173 let _ = tx.send(user.update_email(user_id, &email).map_err(metastore_to_db));
5174 }
5175 UserRequest::SetEmailVerified {
5176 user_id,
5177 verified,
5178 tx,
5179 } => {
5180 let _ = tx.send(
5181 user.set_email_verified(user_id, verified)
5182 .map_err(metastore_to_db),
5183 );
5184 }
5185 UserRequest::CheckEmailVerifiedByIdentifier { identifier, tx } => {
5186 let _ = tx.send(
5187 user.check_email_verified_by_identifier(&identifier)
5188 .map_err(metastore_to_db),
5189 );
5190 }
5191 UserRequest::CheckChannelVerifiedByDid { did, channel, tx } => {
5192 let _ = tx.send(
5193 user.check_channel_verified_by_did(&did, channel)
5194 .map_err(metastore_to_db),
5195 );
5196 }
5197 UserRequest::AdminUpdateEmail { did, email, tx } => {
5198 let _ = tx.send(
5199 user.admin_update_email(&did, &email)
5200 .map_err(metastore_to_db),
5201 );
5202 }
5203 UserRequest::AdminUpdateHandle { did, handle, tx } => {
5204 let _ = tx.send(
5205 user.admin_update_handle(&did, &handle)
5206 .map_err(metastore_to_db),
5207 );
5208 }
5209 UserRequest::AdminUpdatePassword {
5210 did,
5211 password_hash,
5212 tx,
5213 } => {
5214 let _ = tx.send(
5215 user.admin_update_password(&did, &password_hash)
5216 .map_err(metastore_to_db),
5217 );
5218 }
5219 UserRequest::SetAdminStatus { did, is_admin, tx } => {
5220 let _ = tx.send(
5221 user.set_admin_status(&did, is_admin)
5222 .map_err(metastore_to_db),
5223 );
5224 }
5225 UserRequest::GetNotificationPrefs { did, tx } => {
5226 let _ = tx.send(user.get_notification_prefs(&did).map_err(metastore_to_db));
5227 }
5228 UserRequest::GetIdHandleEmailByDid { did, tx } => {
5229 let _ = tx.send(
5230 user.get_id_handle_email_by_did(&did)
5231 .map_err(metastore_to_db),
5232 );
5233 }
5234 UserRequest::UpdatePreferredCommsChannel { did, channel, tx } => {
5235 let _ = tx.send(
5236 user.update_preferred_comms_channel(&did, channel)
5237 .map_err(metastore_to_db),
5238 );
5239 }
5240 UserRequest::ClearDiscord { user_id, tx } => {
5241 let _ = tx.send(user.clear_discord(user_id).map_err(metastore_to_db));
5242 }
5243 UserRequest::ClearTelegram { user_id, tx } => {
5244 let _ = tx.send(user.clear_telegram(user_id).map_err(metastore_to_db));
5245 }
5246 UserRequest::ClearSignal { user_id, tx } => {
5247 let _ = tx.send(user.clear_signal(user_id).map_err(metastore_to_db));
5248 }
5249 UserRequest::SetUnverifiedSignal {
5250 user_id,
5251 signal_username,
5252 tx,
5253 } => {
5254 let _ = tx.send(
5255 user.set_unverified_signal(user_id, &signal_username)
5256 .map_err(metastore_to_db),
5257 );
5258 }
5259 UserRequest::SetUnverifiedTelegram {
5260 user_id,
5261 telegram_username,
5262 tx,
5263 } => {
5264 let _ = tx.send(
5265 user.set_unverified_telegram(user_id, &telegram_username)
5266 .map_err(metastore_to_db),
5267 );
5268 }
5269 UserRequest::StoreTelegramChatId {
5270 telegram_username,
5271 chat_id,
5272 handle,
5273 tx,
5274 } => {
5275 let _ = tx.send(
5276 user.store_telegram_chat_id(&telegram_username, chat_id, handle.as_deref())
5277 .map_err(metastore_to_db),
5278 );
5279 }
5280 UserRequest::GetTelegramChatId { user_id, tx } => {
5281 let _ = tx.send(user.get_telegram_chat_id(user_id).map_err(metastore_to_db));
5282 }
5283 UserRequest::SetUnverifiedDiscord {
5284 user_id,
5285 discord_username,
5286 tx,
5287 } => {
5288 let _ = tx.send(
5289 user.set_unverified_discord(user_id, &discord_username)
5290 .map_err(metastore_to_db),
5291 );
5292 }
5293 UserRequest::StoreDiscordUserId {
5294 discord_username,
5295 discord_id,
5296 handle,
5297 tx,
5298 } => {
5299 let _ = tx.send(
5300 user.store_discord_user_id(&discord_username, &discord_id, handle.as_deref())
5301 .map_err(metastore_to_db),
5302 );
5303 }
5304 UserRequest::GetVerificationInfo { did, tx } => {
5305 let _ = tx.send(user.get_verification_info(&did).map_err(metastore_to_db));
5306 }
5307 UserRequest::VerifyEmailChannel { user_id, email, tx } => {
5308 let _ = tx.send(
5309 user.verify_email_channel(user_id, &email)
5310 .map_err(metastore_to_db),
5311 );
5312 }
5313 UserRequest::VerifyDiscordChannel {
5314 user_id,
5315 discord_id,
5316 tx,
5317 } => {
5318 let _ = tx.send(
5319 user.verify_discord_channel(user_id, &discord_id)
5320 .map_err(metastore_to_db),
5321 );
5322 }
5323 UserRequest::VerifyTelegramChannel {
5324 user_id,
5325 telegram_username,
5326 tx,
5327 } => {
5328 let _ = tx.send(
5329 user.verify_telegram_channel(user_id, &telegram_username)
5330 .map_err(metastore_to_db),
5331 );
5332 }
5333 UserRequest::VerifySignalChannel {
5334 user_id,
5335 signal_username,
5336 tx,
5337 } => {
5338 let _ = tx.send(
5339 user.verify_signal_channel(user_id, &signal_username)
5340 .map_err(metastore_to_db),
5341 );
5342 }
5343 UserRequest::SetEmailVerifiedFlag { user_id, tx } => {
5344 let _ = tx.send(
5345 user.set_email_verified_flag(user_id)
5346 .map_err(metastore_to_db),
5347 );
5348 }
5349 UserRequest::SetDiscordVerifiedFlag { user_id, tx } => {
5350 let _ = tx.send(
5351 user.set_discord_verified_flag(user_id)
5352 .map_err(metastore_to_db),
5353 );
5354 }
5355 UserRequest::SetTelegramVerifiedFlag { user_id, tx } => {
5356 let _ = tx.send(
5357 user.set_telegram_verified_flag(user_id)
5358 .map_err(metastore_to_db),
5359 );
5360 }
5361 UserRequest::SetSignalVerifiedFlag { user_id, tx } => {
5362 let _ = tx.send(
5363 user.set_signal_verified_flag(user_id)
5364 .map_err(metastore_to_db),
5365 );
5366 }
5367 UserRequest::HasTotpEnabled { did, tx } => {
5368 let _ = tx.send(user.has_totp_enabled(&did).map_err(metastore_to_db));
5369 }
5370 UserRequest::HasPasskeys { did, tx } => {
5371 let _ = tx.send(user.has_passkeys(&did).map_err(metastore_to_db));
5372 }
5373 UserRequest::GetPasswordHashByDid { did, tx } => {
5374 let _ = tx.send(user.get_password_hash_by_did(&did).map_err(metastore_to_db));
5375 }
5376 UserRequest::GetPasskeysForUser { did, tx } => {
5377 let _ = tx.send(user.get_passkeys_for_user(&did).map_err(metastore_to_db));
5378 }
5379 UserRequest::GetPasskeyByCredentialId { credential_id, tx } => {
5380 let _ = tx.send(
5381 user.get_passkey_by_credential_id(&credential_id)
5382 .map_err(metastore_to_db),
5383 );
5384 }
5385 UserRequest::SavePasskey {
5386 did,
5387 credential_id,
5388 public_key,
5389 friendly_name,
5390 tx,
5391 } => {
5392 let _ = tx.send(
5393 user.save_passkey(&did, &credential_id, &public_key, friendly_name.as_deref())
5394 .map_err(metastore_to_db),
5395 );
5396 }
5397 UserRequest::UpdatePasskeyCounter {
5398 credential_id,
5399 new_counter,
5400 tx,
5401 } => {
5402 let _ = tx.send(
5403 user.update_passkey_counter(&credential_id, new_counter)
5404 .map_err(metastore_to_db),
5405 );
5406 }
5407 UserRequest::DeletePasskey { id, did, tx } => {
5408 let _ = tx.send(user.delete_passkey(id, &did).map_err(metastore_to_db));
5409 }
5410 UserRequest::UpdatePasskeyName { id, did, name, tx } => {
5411 let _ = tx.send(
5412 user.update_passkey_name(id, &did, &name)
5413 .map_err(metastore_to_db),
5414 );
5415 }
5416 UserRequest::SaveWebauthnChallenge {
5417 did,
5418 challenge_type,
5419 state_json,
5420 tx,
5421 } => {
5422 let _ = tx.send(
5423 user.save_webauthn_challenge(&did, challenge_type, &state_json)
5424 .map_err(metastore_to_db),
5425 );
5426 }
5427 UserRequest::LoadWebauthnChallenge {
5428 did,
5429 challenge_type,
5430 tx,
5431 } => {
5432 let _ = tx.send(
5433 user.load_webauthn_challenge(&did, challenge_type)
5434 .map_err(metastore_to_db),
5435 );
5436 }
5437 UserRequest::DeleteWebauthnChallenge {
5438 did,
5439 challenge_type,
5440 tx,
5441 } => {
5442 let _ = tx.send(
5443 user.delete_webauthn_challenge(&did, challenge_type)
5444 .map_err(metastore_to_db),
5445 );
5446 }
5447 UserRequest::SaveDiscoverableChallenge {
5448 request_key,
5449 state_json,
5450 tx,
5451 } => {
5452 let _ = tx.send(
5453 user.save_discoverable_challenge(&request_key, &state_json)
5454 .map_err(metastore_to_db),
5455 );
5456 }
5457 UserRequest::LoadDiscoverableChallenge { request_key, tx } => {
5458 let _ = tx.send(
5459 user.load_discoverable_challenge(&request_key)
5460 .map_err(metastore_to_db),
5461 );
5462 }
5463 UserRequest::DeleteDiscoverableChallenge { request_key, tx } => {
5464 let _ = tx.send(
5465 user.delete_discoverable_challenge(&request_key)
5466 .map_err(metastore_to_db),
5467 );
5468 }
5469 UserRequest::GetTotpRecord { did, tx } => {
5470 let _ = tx.send(user.get_totp_record(&did).map_err(metastore_to_db));
5471 }
5472 UserRequest::GetTotpRecordState { did, tx } => {
5473 let _ = tx.send(user.get_totp_record_state(&did).map_err(metastore_to_db));
5474 }
5475 UserRequest::UpsertTotpSecret {
5476 did,
5477 secret_encrypted,
5478 encryption_version,
5479 tx,
5480 } => {
5481 let _ = tx.send(
5482 user.upsert_totp_secret(&did, &secret_encrypted, encryption_version)
5483 .map_err(metastore_to_db),
5484 );
5485 }
5486 UserRequest::SetTotpVerified { did, tx } => {
5487 let _ = tx.send(user.set_totp_verified(&did).map_err(metastore_to_db));
5488 }
5489 UserRequest::UpdateTotpLastUsed { did, tx } => {
5490 let _ = tx.send(user.update_totp_last_used(&did).map_err(metastore_to_db));
5491 }
5492 UserRequest::DeleteTotp { did, tx } => {
5493 let _ = tx.send(user.delete_totp(&did).map_err(metastore_to_db));
5494 }
5495 UserRequest::GetUnusedBackupCodes { did, tx } => {
5496 let _ = tx.send(user.get_unused_backup_codes(&did).map_err(metastore_to_db));
5497 }
5498 UserRequest::MarkBackupCodeUsed { code_id, tx } => {
5499 let _ = tx.send(user.mark_backup_code_used(code_id).map_err(metastore_to_db));
5500 }
5501 UserRequest::CountUnusedBackupCodes { did, tx } => {
5502 let _ = tx.send(
5503 user.count_unused_backup_codes(&did)
5504 .map_err(metastore_to_db),
5505 );
5506 }
5507 UserRequest::DeleteBackupCodes { did, tx } => {
5508 let _ = tx.send(user.delete_backup_codes(&did).map_err(metastore_to_db));
5509 }
5510 UserRequest::InsertBackupCodes {
5511 did,
5512 code_hashes,
5513 tx,
5514 } => {
5515 let _ = tx.send(
5516 user.insert_backup_codes(&did, &code_hashes)
5517 .map_err(metastore_to_db),
5518 );
5519 }
5520 UserRequest::EnableTotpWithBackupCodes {
5521 did,
5522 code_hashes,
5523 tx,
5524 } => {
5525 let _ = tx.send(
5526 user.enable_totp_with_backup_codes(&did, &code_hashes)
5527 .map_err(metastore_to_db),
5528 );
5529 }
5530 UserRequest::DeleteTotpAndBackupCodes { did, tx } => {
5531 let _ = tx.send(
5532 user.delete_totp_and_backup_codes(&did)
5533 .map_err(metastore_to_db),
5534 );
5535 }
5536 UserRequest::ReplaceBackupCodes {
5537 did,
5538 code_hashes,
5539 tx,
5540 } => {
5541 let _ = tx.send(
5542 user.replace_backup_codes(&did, &code_hashes)
5543 .map_err(metastore_to_db),
5544 );
5545 }
5546 UserRequest::GetSessionInfoByDid { did, tx } => {
5547 let _ = tx.send(user.get_session_info_by_did(&did).map_err(metastore_to_db));
5548 }
5549 UserRequest::GetLegacyLoginPref { did, tx } => {
5550 let _ = tx.send(user.get_legacy_login_pref(&did).map_err(metastore_to_db));
5551 }
5552 UserRequest::UpdateLegacyLogin { did, allow, tx } => {
5553 let _ = tx.send(
5554 user.update_legacy_login(&did, allow)
5555 .map_err(metastore_to_db),
5556 );
5557 }
5558 UserRequest::UpdateLocale { did, locale, tx } => {
5559 let _ = tx.send(user.update_locale(&did, &locale).map_err(metastore_to_db));
5560 }
5561 UserRequest::GetLoginFullByIdentifier { identifier, tx } => {
5562 let _ = tx.send(
5563 user.get_login_full_by_identifier(&identifier)
5564 .map_err(metastore_to_db),
5565 );
5566 }
5567 UserRequest::GetConfirmSignupByDid { did, tx } => {
5568 let _ = tx.send(
5569 user.get_confirm_signup_by_did(&did)
5570 .map_err(metastore_to_db),
5571 );
5572 }
5573 UserRequest::GetResendVerificationByDid { did, tx } => {
5574 let _ = tx.send(
5575 user.get_resend_verification_by_did(&did)
5576 .map_err(metastore_to_db),
5577 );
5578 }
5579 UserRequest::SetChannelVerified { did, channel, tx } => {
5580 let _ = tx.send(
5581 user.set_channel_verified(&did, channel)
5582 .map_err(metastore_to_db),
5583 );
5584 }
5585 UserRequest::GetIdByEmailOrHandle { email, handle, tx } => {
5586 let _ = tx.send(
5587 user.get_id_by_email_or_handle(&email, &handle)
5588 .map_err(metastore_to_db),
5589 );
5590 }
5591 UserRequest::CountAccountsByEmail { email, tx } => {
5592 let _ = tx.send(
5593 user.count_accounts_by_email(&email)
5594 .map_err(metastore_to_db),
5595 );
5596 }
5597 UserRequest::GetHandlesByEmail { email, tx } => {
5598 let _ = tx.send(user.get_handles_by_email(&email).map_err(metastore_to_db));
5599 }
5600 UserRequest::SetPasswordResetCode {
5601 user_id,
5602 code,
5603 expires_at,
5604 tx,
5605 } => {
5606 let _ = tx.send(
5607 user.set_password_reset_code(user_id, &code, expires_at)
5608 .map_err(metastore_to_db),
5609 );
5610 }
5611 UserRequest::GetUserByResetCode { code, tx } => {
5612 let _ = tx.send(user.get_user_by_reset_code(&code).map_err(metastore_to_db));
5613 }
5614 UserRequest::ClearPasswordResetCode { user_id, tx } => {
5615 let _ = tx.send(
5616 user.clear_password_reset_code(user_id)
5617 .map_err(metastore_to_db),
5618 );
5619 }
5620 UserRequest::GetIdAndPasswordHashByDid { did, tx } => {
5621 let _ = tx.send(
5622 user.get_id_and_password_hash_by_did(&did)
5623 .map_err(metastore_to_db),
5624 );
5625 }
5626 UserRequest::UpdatePasswordHash {
5627 user_id,
5628 password_hash,
5629 tx,
5630 } => {
5631 let _ = tx.send(
5632 user.update_password_hash(user_id, &password_hash)
5633 .map_err(metastore_to_db),
5634 );
5635 }
5636 UserRequest::ResetPasswordWithSessions {
5637 user_id,
5638 password_hash,
5639 tx,
5640 } => {
5641 let _ = tx.send(
5642 user.reset_password_with_sessions(user_id, &password_hash)
5643 .map_err(metastore_to_db),
5644 );
5645 }
5646 UserRequest::ActivateAccount { did, tx } => {
5647 let _ = tx.send(user.activate_account(&did).map_err(metastore_to_db));
5648 }
5649 UserRequest::DeactivateAccount {
5650 did,
5651 delete_after,
5652 tx,
5653 } => {
5654 let _ = tx.send(
5655 user.deactivate_account(&did, delete_after)
5656 .map_err(metastore_to_db),
5657 );
5658 }
5659 UserRequest::HasPasswordByDid { did, tx } => {
5660 let _ = tx.send(user.has_password_by_did(&did).map_err(metastore_to_db));
5661 }
5662 UserRequest::GetPasswordInfoByDid { did, tx } => {
5663 let _ = tx.send(user.get_password_info_by_did(&did).map_err(metastore_to_db));
5664 }
5665 UserRequest::RemoveUserPassword { user_id, tx } => {
5666 let _ = tx.send(user.remove_user_password(user_id).map_err(metastore_to_db));
5667 }
5668 UserRequest::SetNewUserPassword {
5669 user_id,
5670 password_hash,
5671 tx,
5672 } => {
5673 let _ = tx.send(
5674 user.set_new_user_password(user_id, &password_hash)
5675 .map_err(metastore_to_db),
5676 );
5677 }
5678 UserRequest::GetUserKeyByDid { did, tx } => {
5679 let _ = tx.send(user.get_user_key_by_did(&did).map_err(metastore_to_db));
5680 }
5681 UserRequest::DeleteAccountComplete { user_id, did, tx } => {
5682 let result = purge_repo_side_data(state, user_id, &did)
5683 .and_then(|()| user.delete_account_complete(user_id, &did))
5684 .map_err(metastore_to_db);
5685 let _ = tx.send(result);
5686 }
5687 UserRequest::SetUserTakedown {
5688 did,
5689 takedown_ref,
5690 tx,
5691 } => {
5692 let _ = tx.send(
5693 user.set_user_takedown(&did, takedown_ref.as_deref())
5694 .map_err(metastore_to_db),
5695 );
5696 }
5697 UserRequest::AdminDeleteAccountComplete { user_id, did, tx } => {
5698 let result = purge_repo_side_data(state, user_id, &did)
5699 .and_then(|()| user.admin_delete_account_complete(user_id, &did))
5700 .map_err(metastore_to_db);
5701 let _ = tx.send(result);
5702 }
5703 UserRequest::GetUserForDidDoc { did, tx } => {
5704 let _ = tx.send(user.get_user_for_did_doc(&did).map_err(metastore_to_db));
5705 }
5706 UserRequest::GetUserForDidDocBuild { did, tx } => {
5707 let _ = tx.send(
5708 user.get_user_for_did_doc_build(&did)
5709 .map_err(metastore_to_db),
5710 );
5711 }
5712 UserRequest::UpsertDidWebOverrides {
5713 user_id,
5714 verification_methods,
5715 also_known_as,
5716 tx,
5717 } => {
5718 let _ = tx.send(
5719 user.upsert_did_web_overrides(user_id, verification_methods, also_known_as)
5720 .map_err(metastore_to_db),
5721 );
5722 }
5723 UserRequest::UpdateMigratedToPds { did, endpoint, tx } => {
5724 let _ = tx.send(
5725 user.update_migrated_to_pds(&did, &endpoint)
5726 .map_err(metastore_to_db),
5727 );
5728 }
5729 UserRequest::GetUserForPasskeySetup { did, tx } => {
5730 let _ = tx.send(
5731 user.get_user_for_passkey_setup(&did)
5732 .map_err(metastore_to_db),
5733 );
5734 }
5735 UserRequest::GetUserForPasskeyRecovery {
5736 identifier,
5737 normalized_handle,
5738 tx,
5739 } => {
5740 let _ = tx.send(
5741 user.get_user_for_passkey_recovery(&identifier, &normalized_handle)
5742 .map_err(metastore_to_db),
5743 );
5744 }
5745 UserRequest::SetRecoveryToken {
5746 did,
5747 token_hash,
5748 expires_at,
5749 tx,
5750 } => {
5751 let _ = tx.send(
5752 user.set_recovery_token(&did, &token_hash, expires_at)
5753 .map_err(metastore_to_db),
5754 );
5755 }
5756 UserRequest::GetUserForRecovery { did, tx } => {
5757 let _ = tx.send(user.get_user_for_recovery(&did).map_err(metastore_to_db));
5758 }
5759 UserRequest::GetAccountsScheduledForDeletion { limit, tx } => {
5760 let _ = tx.send(
5761 user.get_accounts_scheduled_for_deletion(limit)
5762 .map_err(metastore_to_db),
5763 );
5764 }
5765 UserRequest::DeleteAccountWithFirehose { user_id, did, tx } => {
5766 let result = user
5767 .delete_account_complete(user_id, &did)
5768 .map_err(metastore_to_db)
5769 .and_then(|()| {
5770 state
5771 .event_ops
5772 .insert_account_event(&did, AccountStatus::Deleted)
5773 });
5774 let _ = tx.send(result.map(|_| ()));
5775 }
5776 UserRequest::CreatePasswordAccount { input, tx } => {
5777 let infra = state.metastore.infra_ops();
5778 let code = input.invite_code.as_ref();
5779 let result = reserve_invite(&infra, code).and_then(|()| {
5780 finalize_account(
5781 &infra,
5782 code,
5783 user.create_password_account(&input),
5784 |result| {
5785 if let Some(key_id) = input.reserved_key_id {
5786 infra
5787 .mark_signing_key_used(key_id)
5788 .map_err(|e| CreateAccountError::Database(e.to_string()))?;
5789 }
5790 Ok(result.user_id)
5791 },
5792 )
5793 });
5794 let _ = tx.send(result);
5795 }
5796 UserRequest::CreateDelegatedAccount { input, tx } => {
5797 let result = user.create_delegated_account(&input).and_then(|account| {
5798 let scope =
5799 tranquil_db_traits::DbScope::new(&input.controller_scopes).map_err(|e| {
5800 tranquil_db_traits::CreateAccountError::Database(format!(
5801 "invalid delegation scope: {e}"
5802 ))
5803 })?;
5804 state
5805 .metastore
5806 .delegation_ops()
5807 .create_delegation(
5808 &input.did,
5809 &input.controller_did,
5810 &scope,
5811 &input.controller_did,
5812 )
5813 .map_err(|e| {
5814 tranquil_db_traits::CreateAccountError::Database(format!(
5815 "delegation grant creation failed: {e}"
5816 ))
5817 })?;
5818 Ok(account)
5819 });
5820 let _ = tx.send(result);
5821 }
5822 UserRequest::CreatePasskeyAccount { input, tx } => {
5823 let infra = state.metastore.infra_ops();
5824 let code = input.invite_code.as_ref();
5825 let result = reserve_invite(&infra, code).and_then(|()| {
5826 finalize_account(
5827 &infra,
5828 code,
5829 user.create_passkey_account(&input),
5830 |result| {
5831 if let Some(key_id) = input.reserved_key_id {
5832 infra
5833 .mark_signing_key_used(key_id)
5834 .map_err(|e| CreateAccountError::Database(e.to_string()))?;
5835 }
5836 Ok(result.user_id)
5837 },
5838 )
5839 });
5840 let _ = tx.send(result);
5841 }
5842 UserRequest::CreateSsoAccount { input, tx } => {
5843 let sso_ops = state.metastore.sso_ops();
5844 let infra = state.metastore.infra_ops();
5845 let code = input.invite_code.as_ref();
5846 let result = sso_ops
5847 .consume_pending_registration(&input.pending_registration_token)
5848 .map_err(|e| CreateAccountError::Database(e.to_string()))
5849 .and_then(|consumed| match consumed {
5850 Some(_) => reserve_invite(&infra, code).and_then(|()| {
5851 finalize_account(&infra, code, user.create_sso_account(&input), |result| {
5852 sso_ops
5853 .create_external_identity(
5854 &input.did,
5855 input.sso_provider,
5856 &input.sso_provider_user_id,
5857 input.sso_provider_username.as_deref(),
5858 input.sso_provider_email.as_deref(),
5859 )
5860 .map_err(|e| CreateAccountError::Database(e.to_string()))?;
5861 Ok(result.user_id)
5862 })
5863 }),
5864 None => Err(CreateAccountError::InvalidToken),
5865 });
5866 let _ = tx.send(result);
5867 }
5868 UserRequest::ReactivateMigrationAccount { input, tx } => {
5869 let _ = tx.send(user.reactivate_migration_account(&input));
5870 }
5871 UserRequest::CheckHandleAvailableForNewAccount { handle, tx } => {
5872 let _ = tx.send(
5873 user.check_handle_available_for_new_account(&handle)
5874 .map_err(metastore_to_db),
5875 );
5876 }
5877 UserRequest::ReserveHandle {
5878 handle,
5879 reserved_by,
5880 tx,
5881 } => {
5882 let _ = tx.send(
5883 user.reserve_handle(&handle, &reserved_by)
5884 .map_err(metastore_to_db),
5885 );
5886 }
5887 UserRequest::ReleaseHandleReservation { handle, tx } => {
5888 let _ = tx.send(
5889 user.release_handle_reservation(&handle)
5890 .map_err(metastore_to_db),
5891 );
5892 }
5893 UserRequest::CleanupExpiredHandleReservations { tx } => {
5894 let _ = tx.send(
5895 user.cleanup_expired_handle_reservations()
5896 .map_err(metastore_to_db),
5897 );
5898 }
5899 UserRequest::CompletePasskeySetup { input, tx } => {
5900 let _ = tx.send(user.complete_passkey_setup(&input).map_err(metastore_to_db));
5901 }
5902 UserRequest::RecoverPasskeyAccount { input, tx } => {
5903 let _ = tx.send(
5904 user.recover_passkey_account(&input)
5905 .map_err(metastore_to_db),
5906 );
5907 }
5908 UserRequest::GetPasswordResetInfo { email, tx } => {
5909 let _ = tx.send(
5910 user.get_password_reset_info(&email)
5911 .map_err(metastore_to_db),
5912 );
5913 }
5914 UserRequest::EnableTotpVerified {
5915 did,
5916 encrypted_secret,
5917 tx,
5918 } => {
5919 let _ = tx.send(
5920 user.enable_totp_verified(&did, &encrypted_secret)
5921 .map_err(metastore_to_db),
5922 );
5923 }
5924 UserRequest::SetTwoFactorEnabled { did, enabled, tx } => {
5925 let _ = tx.send(
5926 user.set_two_factor_enabled(&did, enabled)
5927 .map_err(metastore_to_db),
5928 );
5929 }
5930 UserRequest::ExpirePasswordResetCode { email, tx } => {
5931 let _ = tx.send(
5932 user.expire_password_reset_code(&email)
5933 .map_err(metastore_to_db),
5934 );
5935 }
5936 }
5937}
5938
5939fn handler_loop<S: StorageIO + 'static>(
5940 metastore: Metastore,
5941 bridge: Arc<EventLogBridge<S>>,
5942 blockstore: Option<TranquilBlockStore<RealIO, SystemClock>>,
5943 rx: flume::Receiver<MetastoreRequest>,
5944 thread_index: usize,
5945) {
5946 let event_ops = metastore.event_ops(Arc::clone(&bridge));
5947 let mut commit_ops = metastore.commit_ops(bridge);
5948 if let Some(bs) = blockstore {
5949 commit_ops = commit_ops.with_blockstore(bs);
5950 }
5951 let state = HandlerState {
5952 metastore,
5953 event_ops,
5954 commit_ops,
5955 };
5956 tracing::info!(thread_index, "metastore handler thread started");
5957 rx.iter().for_each(|req| {
5958 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| dispatch(&state, req))) {
5959 Ok(()) => {}
5960 Err(e) => {
5961 let msg = match e.downcast_ref::<&str>() {
5962 Some(s) => (*s).to_owned(),
5963 None => match e.downcast_ref::<String>() {
5964 Some(s) => s.clone(),
5965 None => "unknown panic payload".to_owned(),
5966 },
5967 };
5968 tracing::error!(thread_index, msg, "recovered metastore handler panic");
5969 }
5970 }
5971 });
5972 tracing::info!(thread_index, "metastore handler thread exiting");
5973}
5974
5975const DEFAULT_CHANNEL_BOUND: usize = 256;
5976const MAX_REPOS_WITHOUT_REV: usize = 10_000;
5977
5978pub struct HandlerPool {
5979 senders: parking_lot::Mutex<Vec<flume::Sender<MetastoreRequest>>>,
5980 handles: parking_lot::Mutex<Option<Vec<JoinHandle<()>>>>,
5981 sender_count: usize,
5982 user_hashes: Arc<UserHashMap>,
5983 round_robin: AtomicUsize,
5984}
5985
5986impl HandlerPool {
5987 pub fn spawn<S: StorageIO + 'static>(
5988 metastore: Metastore,
5989 bridge: Arc<EventLogBridge<S>>,
5990 blockstore: Option<TranquilBlockStore<RealIO, SystemClock>>,
5991 thread_count: Option<usize>,
5992 ) -> Self {
5993 let count = thread_count
5994 .unwrap_or_else(|| {
5995 std::thread::available_parallelism()
5996 .map(|n| n.get().max(2) / 2)
5997 .unwrap_or(1)
5998 })
5999 .max(1);
6000
6001 let user_hashes = Arc::clone(metastore.user_hashes());
6002
6003 let (senders, handles): (Vec<_>, Vec<_>) = (0..count)
6004 .map(|i| {
6005 let (tx, rx) = flume::bounded(DEFAULT_CHANNEL_BOUND);
6006 let ms = metastore.clone();
6007 let br = Arc::clone(&bridge);
6008 let bs = blockstore.clone();
6009 let handle = std::thread::Builder::new()
6010 .name(format!("metastore-{i}"))
6011 .spawn(move || handler_loop(ms, br, bs, rx, i))
6012 .expect("failed to spawn metastore handler thread");
6013 (tx, handle)
6014 })
6015 .unzip();
6016
6017 Self {
6018 sender_count: senders.len(),
6019 senders: parking_lot::Mutex::new(senders),
6020 handles: parking_lot::Mutex::new(Some(handles)),
6021 user_hashes,
6022 round_robin: AtomicUsize::new(0),
6023 }
6024 }
6025
6026 pub fn send(&self, request: MetastoreRequest) -> Result<(), DbError> {
6027 let senders = self.senders.lock();
6028 if senders.is_empty() {
6029 return Err(DbError::Connection(
6030 "metastore handler pool shut down".to_string(),
6031 ));
6032 }
6033 let index = match request.routing(&self.user_hashes) {
6034 Routing::Sharded(bits) => (bits as usize) % senders.len(),
6035 Routing::Global => self.round_robin.fetch_add(1, Ordering::Relaxed) % senders.len(),
6036 };
6037 senders[index].try_send(request).map_err(|e| match e {
6038 flume::TrySendError::Full(_) => {
6039 DbError::Query("metastore handler backpressure".to_string())
6040 }
6041 flume::TrySendError::Disconnected(_) => {
6042 DbError::Connection("metastore handler pool shut down".to_string())
6043 }
6044 })
6045 }
6046
6047 pub fn thread_count(&self) -> usize {
6048 self.sender_count
6049 }
6050
6051 pub async fn close(&self) {
6052 {
6053 self.senders.lock().clear();
6054 }
6055 let handles = { self.handles.lock().take() };
6056 if let Some(handles) = handles {
6057 let join_fut = tokio::task::spawn_blocking(move || {
6058 handles.into_iter().for_each(|h| {
6059 if let Err(e) = h.join() {
6060 tracing::error!("metastore handler thread panicked: {e:?}");
6061 }
6062 });
6063 });
6064 match tokio::time::timeout(std::time::Duration::from_secs(30), join_fut).await {
6065 Ok(_) => tracing::info!("metastore handler threads shut down cleanly"),
6066 Err(_) => tracing::error!("metastore handler thread shutdown timed out after 30s"),
6067 }
6068 }
6069 }
6070}
6071
6072impl Drop for HandlerPool {
6073 fn drop(&mut self) {
6074 self.senders.get_mut().clear();
6075 if let Some(handles) = self.handles.get_mut().take() {
6076 tracing::warn!(
6077 "HandlerPool dropped without calling shutdown(); blocking on thread join"
6078 );
6079 handles.into_iter().for_each(|h| {
6080 if let Err(e) = h.join() {
6081 tracing::error!("metastore handler thread panicked: {e:?}");
6082 }
6083 });
6084 }
6085 }
6086}
6087
6088#[cfg(test)]
6089mod tests {
6090 use super::*;
6091 use crate::eventlog::{EventLog, EventLogConfig};
6092 use crate::metastore::MetastoreConfig;
6093 use tranquil_types::{Did, Handle, InviteCode, Jti};
6094
6095 struct TestHarness {
6096 _metastore_dir: tempfile::TempDir,
6097 _eventlog_dir: tempfile::TempDir,
6098 pool: HandlerPool,
6099 }
6100
6101 fn setup() -> TestHarness {
6102 let metastore_dir = tempfile::TempDir::new().unwrap();
6103 let eventlog_dir = tempfile::TempDir::new().unwrap();
6104 let segments_dir = eventlog_dir.path().join("segments");
6105 std::fs::create_dir_all(&segments_dir).unwrap();
6106
6107 let metastore = Metastore::open(
6108 metastore_dir.path(),
6109 MetastoreConfig {
6110 cache_size_bytes: 64 * 1024 * 1024,
6111 },
6112 )
6113 .unwrap();
6114
6115 let event_log = EventLog::open(
6116 EventLogConfig {
6117 segments_dir,
6118 ..EventLogConfig::default()
6119 },
6120 RealIO::new(),
6121 )
6122 .unwrap();
6123
6124 let bridge = Arc::new(EventLogBridge::new(Arc::new(event_log)));
6125
6126 let pool = HandlerPool::spawn::<RealIO>(metastore, bridge, None, Some(2));
6127
6128 TestHarness {
6129 _metastore_dir: metastore_dir,
6130 _eventlog_dir: eventlog_dir,
6131 pool,
6132 }
6133 }
6134
6135 fn test_cid_link(seed: u8) -> CidLink {
6136 let digest: [u8; 32] = std::array::from_fn(|i| seed.wrapping_add(i as u8));
6137 let mh = multihash::Multihash::<64>::wrap(0x12, &digest).unwrap();
6138 let c = cid::Cid::new_v1(0x71, mh);
6139 CidLink::from_cid(&c)
6140 }
6141
6142 #[tokio::test]
6143 async fn create_and_get_roundtrip() {
6144 let h = setup();
6145 let user_id = Uuid::new_v4();
6146 let did = Did::from("did:plc:handler_test".to_string());
6147 let handle = Handle::from("handler.test.invalid".to_string());
6148 let cid = test_cid_link(1);
6149
6150 let (tx, rx) = oneshot::channel();
6151 h.pool
6152 .send(MetastoreRequest::Repo(RepoRequest::CreateRepoFull {
6153 user_id,
6154 did,
6155 handle,
6156 repo_root_cid: cid.clone(),
6157 repo_rev: Tid::from("rev1".to_string()),
6158 tx,
6159 }))
6160 .unwrap();
6161 rx.await.unwrap().unwrap();
6162
6163 let (tx, rx) = oneshot::channel();
6164 h.pool
6165 .send(MetastoreRequest::Repo(RepoRequest::GetRepo { user_id, tx }))
6166 .unwrap();
6167 let repo = rx.await.unwrap().unwrap().unwrap();
6168 assert_eq!(repo.repo_root_cid, cid);
6169 assert_eq!(repo.repo_rev.as_deref(), Some("rev1"));
6170 }
6171
6172 #[test]
6173 fn reserve_invite_code_consumes_once_and_guards_exhaustion() {
6174 let dir = tempfile::TempDir::new().unwrap();
6175 let metastore = Metastore::open(
6176 dir.path(),
6177 MetastoreConfig {
6178 cache_size_bytes: 64 * 1024 * 1024,
6179 },
6180 )
6181 .unwrap();
6182 let infra = metastore.infra_ops();
6183 let squid = InviteCode::new("squid-invite");
6184 let whelk = InviteCode::new("whelk");
6185
6186 assert!(infra.create_invite_code(&squid, 1, None).unwrap());
6187
6188 infra.reserve_invite_code(&squid).unwrap();
6189 assert_eq!(
6190 infra.get_invite_code_available_uses(&squid).unwrap(),
6191 Some(0)
6192 );
6193
6194 assert!(matches!(
6195 infra.reserve_invite_code(&squid),
6196 Err(InviteCodeError::ExhaustedUses)
6197 ));
6198 assert_eq!(
6199 infra.get_invite_code_available_uses(&squid).unwrap(),
6200 Some(0),
6201 "a rejected reservation must not decrement below zero"
6202 );
6203
6204 assert!(matches!(
6205 infra.reserve_invite_code(&whelk),
6206 Err(InviteCodeError::NotFound)
6207 ));
6208 }
6209
6210 #[test]
6211 fn refund_invite_code_restores_a_reserved_use() {
6212 let dir = tempfile::TempDir::new().unwrap();
6213 let metastore = Metastore::open(
6214 dir.path(),
6215 MetastoreConfig {
6216 cache_size_bytes: 64 * 1024 * 1024,
6217 },
6218 )
6219 .unwrap();
6220 let infra = metastore.infra_ops();
6221 let squid = InviteCode::new("squid-invite");
6222 let whelk = InviteCode::new("whelk");
6223
6224 assert!(infra.create_invite_code(&squid, 1, None).unwrap());
6225
6226 infra.reserve_invite_code(&squid).unwrap();
6227 infra.refund_invite_code(&squid).unwrap();
6228 assert_eq!(
6229 infra.get_invite_code_available_uses(&squid).unwrap(),
6230 Some(1),
6231 "refund must return the reserved use so a failed signup does not burn it"
6232 );
6233
6234 infra.reserve_invite_code(&squid).unwrap();
6235 assert_eq!(
6236 infra.get_invite_code_available_uses(&squid).unwrap(),
6237 Some(0)
6238 );
6239
6240 assert!(matches!(
6241 infra.refund_invite_code(&whelk),
6242 Err(InviteCodeError::NotFound)
6243 ));
6244 }
6245
6246 #[test]
6247 fn routing_determinism() {
6248 let user_id = Uuid::from_u128(0x12345678);
6249 let bits = user_id.as_u128() as u64;
6250 let thread_count = 4usize;
6251 let expected = (bits as usize) % thread_count;
6252 (0..100).for_each(|_| {
6253 assert_eq!((bits as usize) % thread_count, expected);
6254 });
6255 }
6256
6257 #[test]
6258 fn global_round_robin_distributes() {
6259 let counter = AtomicUsize::new(0);
6260 let thread_count = 4usize;
6261 let indices: Vec<usize> = (0..8)
6262 .map(|_| counter.fetch_add(1, Ordering::Relaxed) % thread_count)
6263 .collect();
6264 assert_eq!(indices, vec![0, 1, 2, 3, 0, 1, 2, 3]);
6265 }
6266
6267 #[test]
6268 fn session_mutations_route_by_did() {
6269 let dir = tempfile::TempDir::new().unwrap();
6270 let ms = Metastore::open(
6271 dir.path(),
6272 MetastoreConfig {
6273 cache_size_bytes: 1024 * 1024,
6274 },
6275 )
6276 .unwrap();
6277 let user_hashes = ms.user_hashes().as_ref();
6278 let did = Did::from("did:plc:limpet".to_string());
6279 let expected = did_to_routing(&did);
6280 let sid = SessionId::new(7);
6281
6282 let (tx, _rx) = oneshot::channel();
6283 let delete_by_id = SessionRequest::DeleteSessionById {
6284 session_id: sid,
6285 did: did.clone(),
6286 tx,
6287 };
6288 let (tx, _rx) = oneshot::channel();
6289 let delete_by_jti = SessionRequest::DeleteSessionByAccessJti {
6290 access_jti: "a".to_string(),
6291 did: did.clone(),
6292 tx,
6293 };
6294 let (tx, _rx) = oneshot::channel();
6295 let delete_by_did = SessionRequest::DeleteSessionsByDid {
6296 did: did.clone(),
6297 tx,
6298 };
6299 let (tx, _rx) = oneshot::channel();
6300 let refresh = SessionRequest::RefreshSessionAtomic {
6301 data: tranquil_db_traits::SessionRefreshData {
6302 did: did.clone(),
6303 old_refresh_jti: Jti::new("r0"),
6304 session_id: sid,
6305 new_access_jti: Jti::new("a1"),
6306 new_refresh_jti: Jti::new("r1"),
6307 new_access_expires_at: Utc::now(),
6308 new_refresh_expires_at: Utc::now(),
6309 },
6310 tx,
6311 };
6312
6313 assert_eq!(delete_by_id.routing(user_hashes), expected);
6314 assert_eq!(delete_by_jti.routing(user_hashes), expected);
6315 assert_eq!(delete_by_did.routing(user_hashes), expected);
6316 assert_eq!(refresh.routing(user_hashes), expected);
6317 }
6318
6319 #[tokio::test]
6320 async fn shutdown_completes_inflight() {
6321 let h = setup();
6322 let user_id = Uuid::new_v4();
6323 let did = Did::from("did:plc:shutdown_test".to_string());
6324 let handle = Handle::from("shutdown.test.invalid".to_string());
6325 let cid = test_cid_link(2);
6326
6327 let (tx, rx) = oneshot::channel();
6328 h.pool
6329 .send(MetastoreRequest::Repo(RepoRequest::CreateRepoFull {
6330 user_id,
6331 did,
6332 handle,
6333 repo_root_cid: cid,
6334 repo_rev: Tid::from("rev1".to_string()),
6335 tx,
6336 }))
6337 .unwrap();
6338 rx.await.unwrap().unwrap();
6339
6340 h.pool.close().await;
6341 }
6342}