alibi_core/store/stateless/
transaction.rs

1use crate::field_policy::FieldValues;
2use crate::store::stateless::{StatelessSchema, StatelessStore};
3use crate::store::{
4    AccountStore, BoxedTransactionValue, MemberStore, SessionStore, TeamStore, TransactionStore,
5    TransactionWork, UserStore, VerificationStore,
6};
7use crate::types::AddTeamMemberResult;
8use crate::user_validation::PreparedUserCreation;
9use crate::{
10    AccountView, AuthError, AuthResult, AuthTransaction, CreateAccount, CreateMember,
11    CreateSession, CreateUser, Member, SessionView, Team, UpdateAccount, UserView,
12    VerificationView,
13};
14use async_trait::async_trait;
15use chrono::{DateTime, Utc};
16
17/// Reconcile only transaction-owned changes. Concurrent untouched rows survive;
18/// a changed row wins last, as in the published memory adapter. This is not
19/// serializable isolation across workflow awaits.
20fn merge<T: Clone + PartialEq>(
21    live: &mut indexmap::IndexMap<String, T>,
22    base: &indexmap::IndexMap<String, T>,
23    changed: indexmap::IndexMap<String, T>,
24) {
25    live.retain(|id, _| !base.contains_key(id) || changed.contains_key(id));
26    for (id, row) in changed {
27        if base.get(&id) == Some(&row) {
28            continue;
29        }
30        // Published merge walks live rows first, then appends only new IDs.
31        // A base row concurrently deleted from live must stay deleted.
32        if live.contains_key(&id) || !base.contains_key(&id) {
33            _ = live.insert(id, row);
34        }
35    }
36}
37
38struct OrganizationTransaction<'a> {
39    live: &'a StatelessStore,
40    snapshot: StatelessStore,
41}
42
43#[async_trait]
44impl TransactionStore<StatelessSchema> for StatelessStore {
45    async fn transaction_boxed(
46        &self,
47        work: Box<TransactionWork<StatelessSchema>>,
48    ) -> AuthResult<BoxedTransactionValue> {
49        let base = self.organization_state()?.clone();
50        let tx = OrganizationTransaction {
51            live: self,
52            snapshot: Self {
53                organizations: std::sync::Mutex::new(base.clone()),
54                ..Self::with_find_many_limit(self.find_many_limit)
55            },
56        };
57        let result = work(&tx).await?;
58        let changed = tx
59            .snapshot
60            .organizations
61            .into_inner()
62            .map_err(|_| AuthError::internal("No-database organization state poisoned"))?;
63        let mut live = self.organization_state()?;
64        merge(
65            &mut live.organizations,
66            &base.organizations,
67            changed.organizations,
68        );
69        merge(&mut live.members, &base.members, changed.members);
70        merge(
71            &mut live.invitations,
72            &base.invitations,
73            changed.invitations,
74        );
75        merge(&mut live.teams, &base.teams, changed.teams);
76        merge(
77            &mut live.team_members,
78            &base.team_members,
79            changed.team_members,
80        );
81        merge(&mut live.roles, &base.roles, changed.roles);
82        Ok(result)
83    }
84}
85
86#[async_trait]
87impl AuthTransaction<StatelessSchema> for OrganizationTransaction<'_> {
88    async fn get_team(&self, org: &str, id: &str) -> AuthResult<Option<Team>> {
89        TeamStore::get_team(&self.snapshot, Some(org), id).await
90    }
91    async fn add_team_member(
92        &self,
93        team: &str,
94        user: &str,
95        maximum: Option<f64>,
96    ) -> AuthResult<AddTeamMemberResult> {
97        TeamStore::add_team_member(&self.snapshot, team, user, maximum).await
98    }
99    async fn create_member(&self, data: CreateMember) -> AuthResult<Member> {
100        MemberStore::create_member(&self.snapshot, data).await
101    }
102    async fn prepare_secondary_session_creation(
103        &self,
104        input: CreateSession,
105        persist: bool,
106    ) -> AuthResult<SessionView> {
107        SessionStore::<StatelessSchema>::prepare_secondary_session_creation(
108            self.live, input, persist,
109        )
110        .await
111    }
112    async fn prepare_secondary_session_update(
113        &self,
114        session: SessionView,
115        expires_at: Option<DateTime<Utc>>,
116        fields: FieldValues,
117    ) -> AuthResult<Option<(SessionView, FieldValues)>> {
118        SessionStore::<StatelessSchema>::prepare_secondary_session_update(
119            self.live, session, expires_at, fields,
120        )
121        .await
122    }
123    async fn complete_secondary_session_update(
124        &self,
125        session: SessionView,
126        expires_at: Option<DateTime<Utc>>,
127        fields: FieldValues,
128        persist: bool,
129    ) -> AuthResult<Option<SessionView>> {
130        SessionStore::<StatelessSchema>::complete_secondary_session_update(
131            self.live, session, expires_at, fields, persist,
132        )
133        .await
134    }
135    // Preserve existing identity provisioning behavior. This scoped transaction
136    // stages organization rows only; cookie/secondary sessions belong to the wrapper.
137    async fn create_user(&self, data: CreateUser) -> AuthResult<UserView> {
138        UserStore::<StatelessSchema>::create_user(self.live, data).await
139    }
140    async fn create_user_prepared(&self, data: PreparedUserCreation) -> AuthResult<UserView> {
141        UserStore::<StatelessSchema>::create_user_prepared(self.live, data).await
142    }
143    async fn create_account(&self, data: CreateAccount) -> AuthResult<AccountView> {
144        AccountStore::<StatelessSchema>::create_account(self.live, data).await
145    }
146    async fn create_session(&self, data: CreateSession) -> AuthResult<SessionView> {
147        SessionStore::<StatelessSchema>::create_session(self.live, data).await
148    }
149    async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
150        SessionStore::<StatelessSchema>::delete_user_sessions(self.live, user_id).await
151    }
152    async fn update_account(&self, id: &str, update: UpdateAccount) -> AuthResult<AccountView> {
153        AccountStore::<StatelessSchema>::update_account(self.live, id, update).await
154    }
155    async fn consume_verification_snapshot(
156        &self,
157        identifier: &str,
158    ) -> AuthResult<Option<VerificationView>> {
159        VerificationStore::<StatelessSchema>::consume_verification_snapshot(self.live, identifier)
160            .await
161    }
162}
alibi_core/store/stateless/transaction.rs

alibi_core/store/stateless/
transaction.rs
