astrid_kernel/invite/
durable_reservation.rs

1//! Two-step durable invite redemption.
2//!
3//! These operations let a handler inspect a bearer, complete fallible
4//! provisioning, and then atomically consume the exact record it inspected.
5
6use super::{DurableInviteStore, Invite, SYSTEM_KV_NAMESPACE, now_epoch};
7
8impl DurableInviteStore {
9    /// Read one currently redeemable invite without consuming it.
10    ///
11    /// Handlers use this to prepare fallible provisioning before the atomic
12    /// consume that commits a redemption. Callers must commit with
13    /// [`Self::consume_if_unchanged`] so a stale provisioned identity cannot
14    /// win after another daemon consumed the same record.
15    ///
16    /// # Errors
17    ///
18    /// Returns a storage error if the record cannot be read or decoded.
19    pub async fn redeemable(
20        &self,
21        token_hash: &str,
22    ) -> astrid_storage::StorageResult<Option<Invite>> {
23        let key = Self::key(token_hash);
24        let Some(value) = self.backend.get(SYSTEM_KV_NAMESPACE, &key).await? else {
25            return Ok(None);
26        };
27        let invite = Self::decode(&value)?;
28        let now = now_epoch();
29        if invite.remaining_uses == 0
30            || invite
31                .expires_at_epoch
32                .is_some_and(|expires| expires <= now)
33        {
34            return Ok(None);
35        }
36        Ok(Some(invite))
37    }
38
39    /// Consume the exact invite previously returned by [`Self::redeemable`].
40    ///
41    /// This is the commit operation for prepare-then-consume handlers. It
42    /// fails closed if the record changed, expired, or was consumed while the
43    /// caller performed provisioning.
44    ///
45    /// # Errors
46    ///
47    /// Returns a storage error if the conditional mutation cannot be applied.
48    pub async fn consume_if_unchanged(
49        &self,
50        expected: &Invite,
51    ) -> astrid_storage::StorageResult<bool> {
52        let Some((conditions, mutations)) = Self::consumption_batch(expected)? else {
53            return Ok(false);
54        };
55        self.apply(conditions, mutations).await
56    }
57
58    /// Commit token consumption and principal ownership in one backend batch.
59    /// The ownership store must use the same authoritative runtime KV backend.
60    ///
61    /// # Errors
62    /// Returns validation or potentially ambiguous storage errors; only a
63    /// definite false permits rollback of the new identity.
64    pub async fn consume_with_ownership(
65        &self,
66        expected: &Invite,
67        ownership: &astrid_storage::OwnershipStore,
68        principal: astrid_core::PrincipalUid,
69    ) -> Result<bool, astrid_storage::OwnershipError> {
70        let Some(delegation) = expected.ownership.as_ref() else {
71            return Ok(false);
72        };
73        let Some((conditions, mutations)) = Self::consumption_batch(expected)? else {
74            return Ok(false);
75        };
76        ownership
77            .commit_enrolled_principal(principal, delegation, conditions, mutations)
78            .await
79    }
80
81    fn consumption_batch(
82        expected: &Invite,
83    ) -> astrid_storage::StorageResult<
84        Option<(
85            Vec<astrid_storage::KvBatchCondition>,
86            Vec<astrid_storage::KvBatchMutation>,
87        )>,
88    > {
89        let now = now_epoch();
90        if expected.remaining_uses == 0
91            || expected
92                .expires_at_epoch
93                .is_some_and(|expires| expires <= now)
94        {
95            return Ok(None);
96        }
97        let key = Self::key(&expected.token_hash);
98        let expected_value = Self::encode(expected)?;
99        let mut consumed = expected.clone();
100        consumed.remaining_uses = consumed.remaining_uses.saturating_sub(1);
101        let mutation = if consumed.remaining_uses == 0 {
102            astrid_storage::KvBatchMutation::Delete {
103                key: astrid_storage::KvEntryKey::new(SYSTEM_KV_NAMESPACE, &key)?,
104            }
105        } else {
106            astrid_storage::KvBatchMutation::Set {
107                key: astrid_storage::KvEntryKey::new(SYSTEM_KV_NAMESPACE, &key)?,
108                value: Self::encode(&consumed)?,
109            }
110        };
111        Ok(Some((
112            vec![astrid_storage::KvBatchCondition::ValueEquals {
113                key: astrid_storage::KvEntryKey::new(SYSTEM_KV_NAMESPACE, &key)?,
114                expected: Some(expected_value),
115            }],
116            vec![mutation],
117        )))
118    }
119}
astrid_kernel/invite/durable_reservation.rs

astrid_kernel/invite/
durable_reservation.rs
