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

alibi_core/store/stateless/
transaction.rs
