alibi_core/store/stateless/
transaction.rs

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

alibi_core/store/stateless/
transaction.rs
