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
194    async fn prepare_scim_projection(
195        &self,
196        _domain: &str,
197        _ids: &[String],
198        _roles: bool,
199    ) -> AuthResult<Vec<crate::store::ScimProjectionPreparedUser>> {
200        Err(AuthError::not_implemented(
201            "Transactional SCIM projection is unsupported",
202        ))
203    }
204    async fn create_scim_projection_grant(
205        &self,
206        _data: crate::store::CreateScimProjectionGrant,
207    ) -> AuthResult<()> {
208        Err(AuthError::not_implemented(
209            "Transactional SCIM grants are unsupported",
210        ))
211    }
212    async fn remove_scim_projection_grants(
213        &self,
214        _domain: &str,
215        _user_id: &str,
216        _ids: &[String],
217    ) -> AuthResult<()> {
218        Err(AuthError::not_implemented(
219            "Transactional SCIM grants are unsupported",
220        ))
221    }
222
223    async fn get_user_by_email(&self, _email: &str) -> AuthResult<Option<S::User>> {
224        Err(AuthError::NotImplemented(
225            "Transactional user lookup is unsupported".into(),
226        ))
227    }
228    async fn get_account(
229        &self,
230        _provider: &str,
231        _account_id: &str,
232    ) -> AuthResult<Option<S::Account>> {
233        Err(AuthError::NotImplemented(
234            "Transactional account lookup is unsupported".into(),
235        ))
236    }
237    async fn update_user(&self, _id: &str, _update: crate::UpdateUser) -> AuthResult<S::User> {
238        Err(AuthError::NotImplemented(
239            "Transactional user updates are unsupported".into(),
240        ))
241    }
242    async fn update_account(
243        &self,
244        _id: &str,
245        _update: crate::UpdateAccount,
246    ) -> AuthResult<S::Account> {
247        Err(AuthError::NotImplemented(
248            "Transactional account updates are unsupported".into(),
249        ))
250    }
251    async fn get_user_by_email_record(
252        &self,
253        email: &str,
254    ) -> AuthResult<Option<AdapterRecord<S::User>>> {
255        self.get_user_by_email(email)
256            .await?
257            .map(AdapterRecord::physical)
258            .transpose()
259    }
260    async fn get_account_record(
261        &self,
262        provider: &str,
263        account_id: &str,
264    ) -> AuthResult<Option<AdapterRecord<S::Account>>> {
265        self.get_account(provider, account_id)
266            .await?
267            .map(AdapterRecord::physical)
268            .transpose()
269    }
270    async fn update_user_record(
271        &self,
272        id: &str,
273        update: crate::UpdateUser,
274    ) -> AuthResult<AdapterRecord<S::User>> {
275        AdapterRecord::physical(self.update_user(id, update).await?)
276    }
277    async fn update_account_record(
278        &self,
279        id: &str,
280        update: crate::UpdateAccount,
281    ) -> AuthResult<AdapterRecord<S::Account>> {
282        AdapterRecord::physical(self.update_account(id, update).await?)
283    }
284
285    /// Lock and read a persisted enterprise provider on this transaction's connection.
286    async fn lock_sso_provider(
287        &self,
288        _id: &str,
289        _provider_id: &str,
290    ) -> AuthResult<Option<crate::store::SsoProvider>> {
291        Err(AuthError::NotImplemented(
292            "Transactional SSO provider locking is unsupported".into(),
293        ))
294    }
295    /// Apply a prepared provider update, including linked-account boundary protection.
296    async fn update_sso_provider(
297        &self,
298        _update: crate::store::SsoProviderUpdate,
299    ) -> AuthResult<Option<crate::store::SsoProvider>> {
300        Err(AuthError::NotImplemented(
301            "Transactional SSO provider updates are unsupported".into(),
302        ))
303    }
304    /// Retire a provider and its identity namespace in this transaction.
305    async fn delete_sso_provider_and_accounts(
306        &self,
307        _id: &str,
308        _provider_id: &str,
309    ) -> AuthResult<bool> {
310        Err(AuthError::NotImplemented(
311            "Transactional SSO provider deletion is unsupported".into(),
312        ))
313    }
314
315    /// Read the provider verification column using the adapter's physical scalar
316    /// rules. This is retained output, not authorization input; typed models
317    /// continue to supply the canonical boolean accessor.
318    async fn provider_verification_output(
319        &self,
320        _id: &str,
321    ) -> AuthResult<Option<serde_json::Value>> {
322        Ok(None)
323    }
324
325    /// Stage a session using the actual transaction-local model and creation hooks.
326    async fn prepare_secondary_session_creation(
327        &self,
328        _input: CreateSession,
329        _persist: bool,
330    ) -> AuthResult<S::Session> {
331        Err(AuthError::NotImplemented(
332            "Transactional secondary session creation is unsupported".into(),
333        ))
334    }
335
336    /// Stage an update against a trusted cached model using this transaction's hooks.
337    async fn prepare_secondary_session_update(
338        &self,
339        _session: S::Session,
340        _expires_at: Option<chrono::DateTime<chrono::Utc>>,
341        _fields: FieldValues,
342    ) -> AuthResult<Option<(S::Session, FieldValues)>> {
343        Err(AuthError::NotImplemented(
344            "Transactional secondary session updates are unsupported".into(),
345        ))
346    }
347    /// Persist a staged update on the current connection; after hooks belong to commit.
348    async fn complete_secondary_session_update(
349        &self,
350        _session: S::Session,
351        _expires_at: Option<chrono::DateTime<chrono::Utc>>,
352        _fields: FieldValues,
353        _persist: bool,
354    ) -> AuthResult<Option<S::Session>> {
355        Err(AuthError::NotImplemented(
356            "Transactional secondary session updates are unsupported".into(),
357        ))
358    }
359
360    /// Read the managed signing keyring on this transaction's connection.
361    async fn list_jwks(&self) -> AuthResult<Vec<Jwk>> {
362        Err(AuthError::config(
363            "Transactional JWKS reads are unsupported by this store",
364        ))
365    }
366    /// Find a managed signing key on this transaction's connection.
367    async fn get_jwk_by_id(&self, _id: &str) -> AuthResult<Option<Jwk>> {
368        Err(AuthError::config(
369            "Transactional JWKS reads are unsupported by this store",
370        ))
371    }
372    /// Persist a newly generated managed signing key in this transaction.
373    async fn create_jwk(&self, _data: CreateJwk) -> AuthResult<Jwk> {
374        Err(AuthError::config(
375            "Transactional JWKS writes are unsupported by this store",
376        ))
377    }
378
379    /// Return a retained adapter record. The default is the physical model's
380    /// serialized snapshot; initialized stores apply their declared output policy.
381    async fn create_user_record(
382        &self,
383        create_user: CreateUser,
384    ) -> AuthResult<AdapterRecord<S::User>> {
385        AdapterRecord::physical(self.create_user(create_user).await?)
386    }
387
388    /// Return a retained adapter record. The default is the physical model's
389    /// serialized snapshot; initialized stores apply their declared output policy.
390    async fn create_user_with_source_record(
391        &self,
392        create_user: CreateUser,
393        source: UserValidationSource,
394    ) -> AuthResult<AdapterRecord<S::User>> {
395        AdapterRecord::physical(self.create_user_with_source(create_user, source).await?)
396    }
397
398    /// Return a retained adapter record. The default is the physical model's
399    /// serialized snapshot; initialized stores apply their declared output policy.
400    async fn create_user_prepared_record(
401        &self,
402        prepared: PreparedUserCreation,
403    ) -> AuthResult<AdapterRecord<S::User>> {
404        AdapterRecord::physical(self.create_user_prepared(prepared).await?)
405    }
406
407    /// Return a retained adapter record. The default is the physical model's
408    /// serialized snapshot; initialized stores apply their declared output policy.
409    async fn get_user_by_id_record(&self, id: &str) -> AuthResult<Option<AdapterRecord<S::User>>> {
410        self.get_user_by_id(id)
411            .await?
412            .map(AdapterRecord::physical)
413            .transpose()
414    }
415
416    /// Return a retained adapter record. The default is the physical model's
417    /// serialized snapshot; initialized stores apply their declared output policy.
418    async fn create_account_record(
419        &self,
420        create_account: CreateAccount,
421    ) -> AuthResult<AdapterRecord<S::Account>> {
422        AdapterRecord::physical(self.create_account(create_account).await?)
423    }
424
425    /// Return a retained adapter record. The default is the physical model's
426    /// serialized snapshot; initialized stores apply their declared output policy.
427    async fn create_session_record(
428        &self,
429        create_session: CreateSession,
430    ) -> AuthResult<AdapterRecord<S::Session>> {
431        AdapterRecord::physical(self.create_session(create_session).await?)
432    }
433
434    /// Return a retained adapter record. The default is the physical model's
435    /// serialized snapshot; initialized stores apply their declared output policy.
436    async fn update_session_active_organization_record(
437        &self,
438        token: &str,
439        organization_id: Option<&str>,
440    ) -> AuthResult<AdapterRecord<S::Session>> {
441        AdapterRecord::physical(
442            self.update_session_active_organization(token, organization_id)
443                .await?,
444        )
445    }
446
447    /// Return a retained adapter record. The default is the physical model's
448    /// serialized snapshot; initialized stores apply their declared output policy.
449    async fn update_session_active_team_record(
450        &self,
451        token: &str,
452        team_id: Option<&str>,
453    ) -> AuthResult<AdapterRecord<S::Session>> {
454        AdapterRecord::physical(self.update_session_active_team(token, team_id).await?)
455    }
456
457    /// Create through before hooks, optional physical persistence, secondary
458    /// publication, then after hooks. Unsupported adapters must fail closed.
459    async fn create_verification_record(
460        &self,
461        _data: VerificationCreation,
462        _publication: VerificationPublication,
463    ) -> AuthResult<Option<VerificationSnapshot>> {
464        Err(AuthError::NotImplemented(
465            "Verification publication phases are not supported by this store".into(),
466        ))
467    }
468    /// Read a team within the authorized organization through this transaction.
469    async fn get_team(&self, _organization_id: &str, _team_id: &str) -> AuthResult<Option<Team>> {
470        Err(AuthError::NotImplemented(
471            "Team lookup in a transaction is not supported by this store".into(),
472        ))
473    }
474    /// Add an idempotent team membership under the same transaction's capacity check.
475    async fn add_team_member(
476        &self,
477        _team_id: &str,
478        _user_id: &str,
479        _maximum: Option<f64>,
480    ) -> AuthResult<AddTeamMemberResult> {
481        Err(AuthError::NotImplemented(
482            "Team admission in a transaction is not supported by this store".into(),
483        ))
484    }
485    async fn create_member(&self, _member: CreateMember) -> AuthResult<Member> {
486        Err(AuthError::NotImplemented(
487            "Member creation in a transaction is not supported by this store".into(),
488        ))
489    }
490    /// Update the already authenticated token through the transaction connection.
491    async fn update_session_active_team(
492        &self,
493        _token: &str,
494        _team_id: Option<&str>,
495    ) -> AuthResult<S::Session> {
496        Err(AuthError::NotImplemented(
497            "Active team updates in a transaction are not supported by this store".into(),
498        ))
499    }
500    async fn update_session_active_organization(
501        &self,
502        _token: &str,
503        _organization_id: Option<&str>,
504    ) -> AuthResult<S::Session> {
505        Err(AuthError::NotImplemented(
506            "Active organization updates in a transaction are not supported by this store".into(),
507        ))
508    }
509    /// Read an existing registration owner through this transaction's connection.
510    async fn get_user_by_id(&self, _id: &str) -> AuthResult<Option<S::User>> {
511        Err(AuthError::NotImplemented(
512            "User lookup in a transaction is not supported by this store".to_owned(),
513        ))
514    }
515    /// Persist a verified passkey in the same transaction as its new session.
516    async fn create_passkey(&self, _passkey: CreatePasskey) -> AuthResult<Passkey> {
517        Err(AuthError::NotImplemented(
518            "Passkey creation in a transaction is not supported by this store".to_owned(),
519        ))
520    }
521    async fn create_user(&self, create_user: CreateUser) -> AuthResult<S::User>;
522    /// Create with endpoint-owned identity provenance. The finalized instance
523    /// applies its policy before adapter hooks. Raw adapters retain ordinary creation.
524    async fn create_user_with_source(
525        &self,
526        create_user: CreateUser,
527        _source: UserValidationSource,
528    ) -> AuthResult<S::User> {
529        self.create_user(create_user).await
530    }
531    /// Persist an admitted candidate without a second email normalization.
532    /// Custom adapters serving validation-enabled instances must implement this.
533    async fn create_user_prepared(&self, _prepared: PreparedUserCreation) -> AuthResult<S::User> {
534        Err(AuthError::NotImplemented(
535            "Prepared user creation is not supported by this transaction".into(),
536        ))
537    }
538    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<S::Account>;
539    async fn create_session(&self, create_session: CreateSession) -> AuthResult<S::Session>;
540    async fn create_verification(
541        &self,
542        _verification: CreateVerification,
543    ) -> AuthResult<S::Verification> {
544        Err(AuthError::NotImplemented(
545            "Verification creation in a transaction is not supported by this store".to_owned(),
546        ))
547    }
548}
549
550/// How to handle callback errors after the database has committed.
551#[derive(Clone, Copy, Debug, PartialEq, Eq)]
552pub enum AfterCommitErrorPolicy {
553    Propagate,
554    Report,
555}
556#[async_trait]
557pub trait TransactionStore<S: AuthSchema>: Send + Sync {
558    async fn transaction_boxed(
559        &self,
560        work: Box<TransactionWork<S>>,
561    ) -> AuthResult<BoxedTransactionValue>;
562    async fn transaction_boxed_with_policy(
563        &self,
564        work: Box<TransactionWork<S>>,
565        policy: AfterCommitErrorPolicy,
566    ) -> AuthResult<BoxedTransactionValue> {
567        if policy == AfterCommitErrorPolicy::Propagate {
568            self.transaction_boxed(work).await
569        } else {
570            Err(AuthError::NotImplemented(
571                "Reporting committed transaction hook failures is unsupported".into(),
572            ))
573        }
574    }
575}
576
577/// # Errors
578///
579/// Propagates transaction or callback errors; also rejects an unexpected transaction result type.
580pub async fn transaction<S, T, F>(store: &dyn AuthStore<S>, work: F) -> AuthResult<T>
581where
582    S: AuthSchema,
583    T: Send + 'static,
584    F: for<'tx> FnOnce(&'tx dyn AuthTransaction<S>) -> TypedTransactionFuture<'tx, T>
585        + Send
586        + 'static,
587{
588    transaction_with_policy(store, AfterCommitErrorPolicy::Propagate, work).await
589}
590/// Run typed work while selecting the treatment of errors after commit.
591///
592/// # Errors
593/// Propagates transaction errors and rejects an unexpected result type.
594pub async fn transaction_with_policy<S, T, F>(
595    store: &dyn AuthStore<S>,
596    policy: AfterCommitErrorPolicy,
597    work: F,
598) -> AuthResult<T>
599where
600    S: AuthSchema,
601    T: Send + 'static,
602    F: for<'tx> FnOnce(&'tx dyn AuthTransaction<S>) -> TypedTransactionFuture<'tx, T>
603        + Send
604        + 'static,
605{
606    let value = store
607        .transaction_boxed_with_policy(
608            Box::new(move |tx| {
609                Box::pin(async move {
610                    let result: BoxedTransactionValue = Box::new(work(tx).await?);
611                    Ok(result)
612                })
613            }),
614            policy,
615        )
616        .await?;
617
618    value
619        .downcast::<T>()
620        .map(|boxed| *boxed)
621        .map_err(|_| AuthError::internal("store returned an invalid transaction payload"))
622}
alibi_core/store/contracts/transaction.rs

alibi_core/store/contracts/
transaction.rs
