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