alibi_core/store/contracts/
transaction.rs

1use crate::AdapterRecord;
2use crate::field_policy::FieldValues;
3use crate::types::AddTeamMemberResult;
4use crate::user_validation::{PreparedUserCreation, UserValidationSource};
5use crate::verification::{VerificationCreation, VerificationPublication, VerificationSnapshot};
6use crate::{
7    AuthError, AuthResult, AuthSchema, AuthStore, CreateAccount, CreateJwk, CreateMember,
8    CreatePasskey, CreateSession, CreateUser, CreateVerification, Jwk, Member, Passkey, Team,
9};
10use async_trait::async_trait;
11use std::any::Any;
12use std::pin::Pin;
13pub type BoxedTransactionValue = Box<dyn Any + Send>;
14
15pub type TransactionFuture<'a> =
16    Pin<Box<dyn Future<Output = AuthResult<BoxedTransactionValue>> + Send + 'a>>;
17
18pub type TypedTransactionFuture<'a, T> = Pin<Box<dyn Future<Output = AuthResult<T>> + Send + 'a>>;
19
20pub type TransactionWork<S> =
21    dyn for<'tx> FnOnce(&'tx dyn AuthTransaction<S>) -> TransactionFuture<'tx> + Send;
22
23#[async_trait]
24pub trait AuthTransaction<S: AuthSchema>: Send + Sync {
25    async fn check_scim_user_attributes(
26        &self,
27        _id: &str,
28        _connection: &str,
29        _attributes: &crate::store::ScimUserAttributes,
30    ) -> AuthResult<()> {
31        Err(AuthError::not_implemented(
32            "Transactional SCIM User uniqueness checks are unsupported",
33        ))
34    }
35
36    async fn lock_scim_user(
37        &self,
38        _id: &str,
39        _connection: &str,
40        _domain: &str,
41        _attributes: Option<&crate::store::ScimUserAttributes>,
42        _now: chrono::DateTime<chrono::Utc>,
43    ) -> AuthResult<crate::store::ScimUserMutationContext> {
44        Err(AuthError::not_implemented(
45            "Transactional SCIM User mutation is unsupported",
46        ))
47    }
48    async fn replace_scim_user(
49        &self,
50        _id: &str,
51        _data: crate::store::ReplaceScimUser,
52        _roles: bool,
53    ) -> AuthResult<(
54        crate::store::ScimUserRecord,
55        Vec<crate::store::ScimProjectionPreparedUser>,
56    )> {
57        Err(AuthError::not_implemented(
58            "Transactional SCIM User replacement is unsupported",
59        ))
60    }
61
62    async fn create_scim_user(
63        &self,
64        _data: crate::store::CreateScimUser,
65        _roles: bool,
66    ) -> AuthResult<(
67        crate::store::ScimUserRecord,
68        Vec<crate::store::ScimProjectionPreparedUser>,
69    )> {
70        Err(AuthError::not_implemented(
71            "Transactional SCIM User creation is unsupported",
72        ))
73    }
74
75    async fn delete_scim_user(
76        &self,
77        _id: &str,
78        _connection: &str,
79        _domain: &str,
80        _roles: bool,
81    ) -> AuthResult<(String, Vec<crate::store::ScimProjectionPreparedUser>)> {
82        Err(AuthError::not_implemented(
83            "Transactional SCIM User deletion is unsupported",
84        ))
85    }
86    async fn patch_scim_group(
87        &self,
88        _id: &str,
89        _connection: &str,
90        _domain: &str,
91        _patch: crate::store::ScimGroupPatch,
92        _roles: bool,
93    ) -> AuthResult<(
94        crate::store::ScimGroupRecord,
95        Vec<crate::store::ScimProjectionPreparedUser>,
96    )> {
97        Err(AuthError::not_implemented(
98            "Transactional SCIM Group patching is unsupported",
99        ))
100    }
101    async fn replace_scim_group(
102        &self,
103        _id: &str,
104        _data: crate::store::ReplaceScimGroup,
105        _roles: bool,
106    ) -> AuthResult<(
107        crate::store::ScimGroupRecord,
108        Vec<crate::store::ScimProjectionPreparedUser>,
109    )> {
110        Err(AuthError::not_implemented(
111            "Transactional SCIM Group replacement is unsupported",
112        ))
113    }
114    async fn delete_scim_group(
115        &self,
116        _id: &str,
117        _connection: &str,
118        _domain: &str,
119        _roles: bool,
120    ) -> AuthResult<Vec<crate::store::ScimProjectionPreparedUser>> {
121        Err(AuthError::not_implemented(
122            "Transactional SCIM Group deletion is unsupported",
123        ))
124    }
125    async fn create_scim_group(
126        &self,
127        _data: crate::store::CreateScimGroup,
128        _roles: bool,
129    ) -> AuthResult<(
130        crate::store::ScimGroupRecord,
131        Vec<crate::store::ScimProjectionPreparedUser>,
132    )> {
133        Err(AuthError::not_implemented(
134            "Transactional SCIM Groups are unsupported",
135        ))
136    }
137    async fn fence_scim_connection(&self, _id: &str, _key: &str) -> AuthResult<()> {
138        Err(AuthError::not_implemented(
139            "Transactional SCIM fencing is unsupported",
140        ))
141    }
142    async fn get_scim_connection_binding(
143        &self,
144        _id: &str,
145    ) -> AuthResult<Option<crate::store::ScimConnectionBinding>> {
146        Err(AuthError::not_implemented(
147            "Transactional SCIM bindings are unsupported",
148        ))
149    }
150    async fn scim_projection_candidates(
151        &self,
152        _domain: &str,
153        _cursor: Option<&str>,
154    ) -> AuthResult<Vec<crate::store::ScimProjectionCandidate>> {
155        Err(AuthError::not_implemented(
156            "Transactional SCIM candidates are unsupported",
157        ))
158    }
159    async fn renew_scim_decommission_lease(
160        &self,
161        _binding: &crate::store::ScimConnectionBinding,
162        _lease: &str,
163    ) -> AuthResult<bool> {
164        Err(AuthError::not_implemented(
165            "Transactional SCIM leases are unsupported",
166        ))
167    }
168    async fn advance_scim_decommission(
169        &self,
170        _binding: &crate::store::ScimConnectionBinding,
171        _lease: &str,
172        _cursor: Option<&str>,
173        _count: usize,
174        _more: bool,
175    ) -> AuthResult<Option<crate::store::ScimConnectionBinding>> {
176        Err(AuthError::not_implemented(
177            "Transactional SCIM checkpoints are unsupported",
178        ))
179    }
180    async fn reconcile_scim_identity(
181        &self,
182        _user_id: &str,
183    ) -> AuthResult<crate::store::ScimIdentityState> {
184        Err(AuthError::not_implemented(
185            "Transactional SCIM identity is unsupported",
186        ))
187    }
188    async fn delete_user_sessions(&self, _user_id: &str) -> AuthResult<()> {
189        Err(AuthError::not_implemented(
190            "Transactional session retirement is unsupported",
191        ))
192    }
193    /// Read session deletion candidates on this transaction's connection.
194    async fn get_user_sessions(&self, _user_id: &str) -> AuthResult<Vec<S::Session>> {
195        Err(AuthError::not_implemented(
196            "Transactional session reads are unsupported",
197        ))
198    }
199    async fn list_oauth_tokens_for_session(
200        &self,
201        _kind: crate::store::OAuthTokenKind,
202        _session_id: &str,
203    ) -> AuthResult<Vec<crate::store::OAuthToken>> {
204        Err(AuthError::not_implemented(
205            "Transactional OAuth token reads are unsupported",
206        ))
207    }
208    async fn get_oauth_client(
209        &self,
210        _client_id: &str,
211    ) -> AuthResult<Option<crate::store::OAuthClient>> {
212        Err(AuthError::not_implemented(
213            "Transactional OAuth client reads are unsupported",
214        ))
215    }
216    /// Attach prepared retirement to the matching queued delete-after event.
217    /// A veto or missing row has no event and must not complete the work.
218    async fn queue_session_deletion_work(
219        &self,
220        _session_id: &str,
221        _works: Vec<Box<dyn crate::store::SessionDeletionWork<S>>>,
222        _context: std::sync::Arc<crate::AuthContext<S>>,
223    ) -> AuthResult<()> {
224        Err(AuthError::not_implemented(
225            "Transactional session retirement queueing is unsupported",
226        ))
227    }
228
229    async fn prepare_scim_projection(
230        &self,
231        _domain: &str,
232        _ids: &[String],
233        _roles: bool,
234    ) -> AuthResult<Vec<crate::store::ScimProjectionPreparedUser>> {
235        Err(AuthError::not_implemented(
236            "Transactional SCIM projection is unsupported",
237        ))
238    }
239    async fn create_scim_projection_grant(
240        &self,
241        _data: crate::store::CreateScimProjectionGrant,
242    ) -> AuthResult<()> {
243        Err(AuthError::not_implemented(
244            "Transactional SCIM grants are unsupported",
245        ))
246    }
247    async fn remove_scim_projection_grants(
248        &self,
249        _domain: &str,
250        _user_id: &str,
251        _ids: &[String],
252    ) -> AuthResult<()> {
253        Err(AuthError::not_implemented(
254            "Transactional SCIM grants are unsupported",
255        ))
256    }
257
258    async fn get_user_by_email(&self, _email: &str) -> AuthResult<Option<S::User>> {
259        Err(AuthError::NotImplemented(
260            "Transactional user lookup is unsupported".into(),
261        ))
262    }
263    async fn get_account(
264        &self,
265        _provider: &str,
266        _account_id: &str,
267    ) -> AuthResult<Option<S::Account>> {
268        Err(AuthError::NotImplemented(
269            "Transactional account lookup is unsupported".into(),
270        ))
271    }
272    async fn update_user(&self, _id: &str, _update: crate::UpdateUser) -> AuthResult<S::User> {
273        Err(AuthError::NotImplemented(
274            "Transactional user updates are unsupported".into(),
275        ))
276    }
277    async fn update_account(
278        &self,
279        _id: &str,
280        _update: crate::UpdateAccount,
281    ) -> AuthResult<S::Account> {
282        Err(AuthError::NotImplemented(
283            "Transactional account updates are unsupported".into(),
284        ))
285    }
286    async fn get_user_by_email_record(
287        &self,
288        email: &str,
289    ) -> AuthResult<Option<AdapterRecord<S::User>>> {
290        self.get_user_by_email(email)
291            .await?
292            .map(AdapterRecord::physical)
293            .transpose()
294    }
295    async fn get_account_record(
296        &self,
297        provider: &str,
298        account_id: &str,
299    ) -> AuthResult<Option<AdapterRecord<S::Account>>> {
300        self.get_account(provider, account_id)
301            .await?
302            .map(AdapterRecord::physical)
303            .transpose()
304    }
305    async fn update_user_record(
306        &self,
307        id: &str,
308        update: crate::UpdateUser,
309    ) -> AuthResult<AdapterRecord<S::User>> {
310        AdapterRecord::physical(self.update_user(id, update).await?)
311    }
312    async fn update_account_record(
313        &self,
314        id: &str,
315        update: crate::UpdateAccount,
316    ) -> AuthResult<AdapterRecord<S::Account>> {
317        AdapterRecord::physical(self.update_account(id, update).await?)
318    }
319
320    /// Lock and read a persisted enterprise provider on this transaction's connection.
321    async fn lock_sso_provider(
322        &self,
323        _id: &str,
324        _provider_id: &str,
325    ) -> AuthResult<Option<crate::store::SsoProvider>> {
326        Err(AuthError::NotImplemented(
327            "Transactional SSO provider locking is unsupported".into(),
328        ))
329    }
330    /// Apply a prepared provider update, including linked-account boundary protection.
331    async fn update_sso_provider(
332        &self,
333        _update: crate::store::SsoProviderUpdate,
334    ) -> AuthResult<Option<crate::store::SsoProvider>> {
335        Err(AuthError::NotImplemented(
336            "Transactional SSO provider updates are unsupported".into(),
337        ))
338    }
339    /// Retire a provider and its identity namespace in this transaction.
340    async fn delete_sso_provider_and_accounts(
341        &self,
342        _id: &str,
343        _provider_id: &str,
344    ) -> AuthResult<bool> {
345        Err(AuthError::NotImplemented(
346            "Transactional SSO provider deletion is unsupported".into(),
347        ))
348    }
349
350    /// Read the provider verification column using the adapter's physical scalar
351    /// rules. This is retained output, not authorization input; typed models
352    /// continue to supply the canonical boolean accessor.
353    async fn provider_verification_output(
354        &self,
355        _id: &str,
356    ) -> AuthResult<Option<serde_json::Value>> {
357        Ok(None)
358    }
359
360    /// Stage a session using the actual transaction-local model and creation hooks.
361    async fn prepare_secondary_session_creation(
362        &self,
363        _input: CreateSession,
364        _persist: bool,
365    ) -> AuthResult<S::Session> {
366        Err(AuthError::NotImplemented(
367            "Transactional secondary session creation is unsupported".into(),
368        ))
369    }
370
371    /// Stage an update against a trusted cached model using this transaction's hooks.
372    async fn prepare_secondary_session_update(
373        &self,
374        _session: S::Session,
375        _expires_at: Option<chrono::DateTime<chrono::Utc>>,
376        _fields: FieldValues,
377    ) -> AuthResult<Option<(S::Session, FieldValues)>> {
378        Err(AuthError::NotImplemented(
379            "Transactional secondary session updates are unsupported".into(),
380        ))
381    }
382    /// Persist a staged update on the current connection; after hooks belong to commit.
383    async fn complete_secondary_session_update(
384        &self,
385        _session: S::Session,
386        _expires_at: Option<chrono::DateTime<chrono::Utc>>,
387        _fields: FieldValues,
388        _persist: bool,
389    ) -> AuthResult<Option<S::Session>> {
390        Err(AuthError::NotImplemented(
391            "Transactional secondary session updates are unsupported".into(),
392        ))
393    }
394
395    /// Read the managed signing keyring on this transaction's connection.
396    async fn list_jwks(&self) -> AuthResult<Vec<Jwk>> {
397        Err(AuthError::config(
398            "Transactional JWKS reads are unsupported by this store",
399        ))
400    }
401    /// Find a managed signing key on this transaction's connection.
402    async fn get_jwk_by_id(&self, _id: &str) -> AuthResult<Option<Jwk>> {
403        Err(AuthError::config(
404            "Transactional JWKS reads are unsupported by this store",
405        ))
406    }
407    /// Persist a newly generated managed signing key in this transaction.
408    async fn create_jwk(&self, _data: CreateJwk) -> AuthResult<Jwk> {
409        Err(AuthError::config(
410            "Transactional JWKS writes are unsupported by this store",
411        ))
412    }
413
414    /// Return a retained adapter record. The default is the physical model's
415    /// serialized snapshot; initialized stores apply their declared output policy.
416    async fn create_user_record(
417        &self,
418        create_user: CreateUser,
419    ) -> AuthResult<AdapterRecord<S::User>> {
420        AdapterRecord::physical(self.create_user(create_user).await?)
421    }
422
423    /// Return a retained adapter record. The default is the physical model's
424    /// serialized snapshot; initialized stores apply their declared output policy.
425    async fn create_user_with_source_record(
426        &self,
427        create_user: CreateUser,
428        source: UserValidationSource,
429    ) -> AuthResult<AdapterRecord<S::User>> {
430        AdapterRecord::physical(self.create_user_with_source(create_user, source).await?)
431    }
432
433    /// Return a retained adapter record. The default is the physical model's
434    /// serialized snapshot; initialized stores apply their declared output policy.
435    async fn create_user_prepared_record(
436        &self,
437        prepared: PreparedUserCreation,
438    ) -> AuthResult<AdapterRecord<S::User>> {
439        AdapterRecord::physical(self.create_user_prepared(prepared).await?)
440    }
441
442    /// Return a retained adapter record. The default is the physical model's
443    /// serialized snapshot; initialized stores apply their declared output policy.
444    async fn get_user_by_id_record(&self, id: &str) -> AuthResult<Option<AdapterRecord<S::User>>> {
445        self.get_user_by_id(id)
446            .await?
447            .map(AdapterRecord::physical)
448            .transpose()
449    }
450
451    /// Return a retained adapter record. The default is the physical model's
452    /// serialized snapshot; initialized stores apply their declared output policy.
453    async fn create_account_record(
454        &self,
455        create_account: CreateAccount,
456    ) -> AuthResult<AdapterRecord<S::Account>> {
457        AdapterRecord::physical(self.create_account(create_account).await?)
458    }
459
460    /// Return a retained adapter record. The default is the physical model's
461    /// serialized snapshot; initialized stores apply their declared output policy.
462    async fn create_session_record(
463        &self,
464        create_session: CreateSession,
465    ) -> AuthResult<AdapterRecord<S::Session>> {
466        AdapterRecord::physical(self.create_session(create_session).await?)
467    }
468
469    /// Return a retained adapter record. The default is the physical model's
470    /// serialized snapshot; initialized stores apply their declared output policy.
471    async fn update_session_active_organization_record(
472        &self,
473        token: &str,
474        organization_id: Option<&str>,
475    ) -> AuthResult<AdapterRecord<S::Session>> {
476        AdapterRecord::physical(
477            self.update_session_active_organization(token, organization_id)
478                .await?,
479        )
480    }
481
482    /// Return a retained adapter record. The default is the physical model's
483    /// serialized snapshot; initialized stores apply their declared output policy.
484    async fn update_session_active_team_record(
485        &self,
486        token: &str,
487        team_id: Option<&str>,
488    ) -> AuthResult<AdapterRecord<S::Session>> {
489        AdapterRecord::physical(self.update_session_active_team(token, team_id).await?)
490    }
491
492    /// Create through before hooks, optional physical persistence, secondary
493    /// publication, then after hooks. Unsupported adapters must fail closed.
494    async fn create_verification_record(
495        &self,
496        _data: VerificationCreation,
497        _publication: VerificationPublication,
498    ) -> AuthResult<Option<VerificationSnapshot>> {
499        Err(AuthError::NotImplemented(
500            "Verification publication phases are not supported by this store".into(),
501        ))
502    }
503    /// Read a team within the authorized organization through this transaction.
504    async fn get_team(&self, _organization_id: &str, _team_id: &str) -> AuthResult<Option<Team>> {
505        Err(AuthError::NotImplemented(
506            "Team lookup in a transaction is not supported by this store".into(),
507        ))
508    }
509    /// Add an idempotent team membership under the same transaction's capacity check.
510    async fn add_team_member(
511        &self,
512        _team_id: &str,
513        _user_id: &str,
514        _maximum: Option<f64>,
515    ) -> AuthResult<AddTeamMemberResult> {
516        Err(AuthError::NotImplemented(
517            "Team admission in a transaction is not supported by this store".into(),
518        ))
519    }
520    async fn create_member(&self, _member: CreateMember) -> AuthResult<Member> {
521        Err(AuthError::NotImplemented(
522            "Member creation in a transaction is not supported by this store".into(),
523        ))
524    }
525    /// Update the already authenticated token through the transaction connection.
526    async fn update_session_active_team(
527        &self,
528        _token: &str,
529        _team_id: Option<&str>,
530    ) -> AuthResult<S::Session> {
531        Err(AuthError::NotImplemented(
532            "Active team updates in a transaction are not supported by this store".into(),
533        ))
534    }
535    async fn update_session_active_organization(
536        &self,
537        _token: &str,
538        _organization_id: Option<&str>,
539    ) -> AuthResult<S::Session> {
540        Err(AuthError::NotImplemented(
541            "Active organization updates in a transaction are not supported by this store".into(),
542        ))
543    }
544    /// Read an existing registration owner through this transaction's connection.
545    async fn get_user_by_id(&self, _id: &str) -> AuthResult<Option<S::User>> {
546        Err(AuthError::NotImplemented(
547            "User lookup in a transaction is not supported by this store".to_owned(),
548        ))
549    }
550    /// Persist a verified passkey in the same transaction as its new session.
551    async fn create_passkey(&self, _passkey: CreatePasskey) -> AuthResult<Passkey> {
552        Err(AuthError::NotImplemented(
553            "Passkey creation in a transaction is not supported by this store".to_owned(),
554        ))
555    }
556    async fn create_user(&self, create_user: CreateUser) -> AuthResult<S::User>;
557    /// Create with endpoint-owned identity provenance. The finalized instance
558    /// applies its policy before adapter hooks. Raw adapters retain ordinary creation.
559    async fn create_user_with_source(
560        &self,
561        create_user: CreateUser,
562        _source: UserValidationSource,
563    ) -> AuthResult<S::User> {
564        self.create_user(create_user).await
565    }
566    /// Persist an admitted candidate without a second email normalization.
567    /// Custom adapters serving validation-enabled instances must implement this.
568    async fn create_user_prepared(&self, _prepared: PreparedUserCreation) -> AuthResult<S::User> {
569        Err(AuthError::NotImplemented(
570            "Prepared user creation is not supported by this transaction".into(),
571        ))
572    }
573    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<S::Account>;
574    async fn create_session(&self, create_session: CreateSession) -> AuthResult<S::Session>;
575    async fn create_verification(
576        &self,
577        _verification: CreateVerification,
578    ) -> AuthResult<S::Verification> {
579        Err(AuthError::NotImplemented(
580            "Verification creation in a transaction is not supported by this store".to_owned(),
581        ))
582    }
583}
584
585/// How to handle callback errors after the database has committed.
586#[derive(Clone, Copy, Debug, PartialEq, Eq)]
587pub enum AfterCommitErrorPolicy {
588    Propagate,
589    Report,
590}
591#[async_trait]
592pub trait TransactionStore<S: AuthSchema>: Send + Sync {
593    async fn transaction_boxed(
594        &self,
595        work: Box<TransactionWork<S>>,
596    ) -> AuthResult<BoxedTransactionValue>;
597    async fn transaction_boxed_with_policy(
598        &self,
599        work: Box<TransactionWork<S>>,
600        policy: AfterCommitErrorPolicy,
601    ) -> AuthResult<BoxedTransactionValue> {
602        if policy == AfterCommitErrorPolicy::Propagate {
603            self.transaction_boxed(work).await
604        } else {
605            Err(AuthError::NotImplemented(
606                "Reporting committed transaction hook failures is unsupported".into(),
607            ))
608        }
609    }
610}
611
612/// # Errors
613///
614/// Propagates transaction or callback errors; also rejects an unexpected transaction result type.
615pub async fn transaction<S, T, F>(store: &dyn AuthStore<S>, work: F) -> AuthResult<T>
616where
617    S: AuthSchema,
618    T: Send + 'static,
619    F: for<'tx> FnOnce(&'tx dyn AuthTransaction<S>) -> TypedTransactionFuture<'tx, T>
620        + Send
621        + 'static,
622{
623    transaction_with_policy(store, AfterCommitErrorPolicy::Propagate, work).await
624}
625/// Run typed work while selecting the treatment of errors after commit.
626///
627/// # Errors
628/// Propagates transaction errors and rejects an unexpected result type.
629pub async fn transaction_with_policy<S, T, F>(
630    store: &dyn AuthStore<S>,
631    policy: AfterCommitErrorPolicy,
632    work: F,
633) -> AuthResult<T>
634where
635    S: AuthSchema,
636    T: Send + 'static,
637    F: for<'tx> FnOnce(&'tx dyn AuthTransaction<S>) -> TypedTransactionFuture<'tx, T>
638        + Send
639        + 'static,
640{
641    let value = store
642        .transaction_boxed_with_policy(
643            Box::new(move |tx| {
644                Box::pin(async move {
645                    let result: BoxedTransactionValue = Box::new(work(tx).await?);
646                    Ok(result)
647                })
648            }),
649            policy,
650        )
651        .await?;
652
653    value
654        .downcast::<T>()
655        .map(|boxed| *boxed)
656        .map_err(|_| AuthError::internal("store returned an invalid transaction payload"))
657}
alibi_core/store/contracts/transaction.rs

alibi_core/store/contracts/
transaction.rs
