alibi_sqlx/store/
verifications.rs

1use super::{SqlxStore, lock_exclusive};
2use crate::model::{self, SqlxModel};
3use crate::pool::{Exec, SqlxTransaction};
4use crate::schema::{AuthSchema, SqlxVerificationModel};
5use crate::sql::Sql;
6use alibi_core::entity::AuthVerification;
7use alibi_core::error::{AuthError, AuthResult};
8use alibi_core::store::adapter::cancelled_by_hook;
9use alibi_core::store::{VerificationStore, verification_reservation_key};
10use alibi_core::types::{CreateVerification, UpdateVerification};
11use alibi_core::verification::{
12    VerificationCreation, VerificationPublication, VerificationSnapshot,
13};
14use async_trait::async_trait;
15use chrono::{DateTime, SubsecRound, Utc};
16
17impl<S> SqlxStore<S>
18where
19    S: AuthSchema,
20    S::Verification: SqlxVerificationModel,
21{
22    fn verification_table() -> &'static str {
23        <S::Verification as SqlxModel>::TABLE
24    }
25
26    pub(crate) async fn create_verification_record_with_connection(
27        &self,
28        exec: Exec<'_>,
29        tx: Option<&SqlxTransaction>,
30        mut data: VerificationCreation,
31        publication: VerificationPublication,
32    ) -> AuthResult<Option<VerificationSnapshot>> {
33        let hook_context = self.hook_context(tx);
34        for hook in self.hooks() {
35            if hook
36                .before_create_verification_record(&mut data, &hook_context)
37                .await?
38                .is_cancelled()
39            {
40                return Ok(None);
41            }
42        }
43        let snapshot = if publication.store_in_database {
44            let id = data
45                .id
46                .as_deref()
47                .map(S::Verification::parse_id)
48                .transpose()?;
49            let mut active = S::Verification::new_active(id, data.data(), data.created_at);
50            if let Some(updated) = S::Verification::updated_at_column() {
51                let column = <S::Verification as SqlxModel>::COLUMNS
52                    .iter()
53                    .find(|column| column.name == updated)
54                    .ok_or_else(|| {
55                        AuthError::internal(
56                            "the verification update timestamp binding has no column",
57                        )
58                    })?;
59                active.set(
60                    column.name,
61                    S::Verification::timestamp_value(column.name, data.updated_at),
62                );
63            } else if data.updated_at != data.created_at {
64                return Err(AuthError::internal(
65                    "the verification schema cannot preserve a distinct update timestamp",
66                ));
67            }
68            let model = model::insert::<S::Verification>(exec, &active).await?;
69            VerificationSnapshot::from_model(&model)
70        } else {
71            data.snapshot()
72        };
73        publication.publish(&snapshot).await?;
74        if tx.is_none() {
75            for hook in self.hooks() {
76                hook.after_create_verification_record(&snapshot, &hook_context)
77                    .await?;
78            }
79        }
80        Ok(Some(snapshot))
81    }
82
83    async fn create_verification_with_connection(
84        &self,
85        exec: Exec<'_>,
86        tx: Option<&SqlxTransaction>,
87        mut verification: CreateVerification,
88    ) -> AuthResult<S::Verification> {
89        let hook_context = self.hook_context(tx);
90        for hook in self.hooks() {
91            if hook
92                .before_create_verification(&mut verification, &hook_context)
93                .await?
94                .is_cancelled()
95            {
96                return Err(cancelled_by_hook("verification creation"));
97            }
98        }
99        let active = S::Verification::new_active(None, verification, Utc::now());
100        let verification = model::insert::<S::Verification>(exec, &active).await?;
101        if tx.is_none() {
102            for hook in self.hooks() {
103                hook.after_create_verification(&verification, &hook_context)
104                    .await?;
105            }
106        }
107        Ok(verification)
108    }
109
110    pub(crate) async fn create_verification_in_tx(
111        &self,
112        tx: &SqlxTransaction,
113        verification: CreateVerification,
114    ) -> AuthResult<S::Verification> {
115        self.create_verification_with_connection(Exec::Tx(tx), Some(tx), verification)
116            .await
117    }
118
119    /// `SELECT ... WHERE identifier = ? [AND ...] ORDER BY created_at DESC LIMIT 1`.
120    fn newest_generation(
121        exec: Exec<'_>,
122        filters: &[(&'static str, &'static str, crate::SqlValue)],
123    ) -> Sql {
124        let table = Self::verification_table();
125        let mut sql = model::select_model::<S::Verification>(exec);
126        for (index, (column, operator, value)) in filters.iter().enumerate() {
127            sql.push(if index == 0 { " WHERE " } else { " AND " });
128            sql.column(table, column);
129            sql.push(operator);
130            sql.bind(S::Verification::column_value(column, value.clone()));
131        }
132        sql.push(" ORDER BY ");
133        sql.column(table, S::Verification::created_at_column());
134        sql.push(" DESC LIMIT 1");
135        sql
136    }
137
138    async fn consume_verification_generation(
139        &self,
140        identifier: &str,
141        expected_value: Option<&str>,
142    ) -> AuthResult<Option<S::Verification>> {
143        // SQLite acquires its writer reservation before reading so racers
144        // cannot all hold read locks and fail when upgrading to a write.
145        // PostgreSQL locks the selected row; the affected-row gate still
146        // protects against another consumer using a separate store/process.
147        let transaction = self.pool().begin(true).await?;
148        let outcome = async {
149            let exec = Exec::Tx(&transaction);
150            let mut select = Self::newest_generation(
151                exec,
152                &[(
153                    S::Verification::identifier_column(),
154                    " = ",
155                    identifier.into(),
156                )],
157            );
158            lock_exclusive(&mut select);
159            let Some(model) = exec.fetch_optional::<S::Verification>(select).await? else {
160                return Ok(None);
161            };
162            if expected_value.is_some_and(|expected| model.value() != expected) {
163                return Ok(None);
164            }
165            let hook_context = self.hook_context(Some(&transaction));
166            for hook in self.hooks() {
167                if hook
168                    .before_delete_verification(&model, &hook_context)
169                    .await?
170                    .is_cancelled()
171                {
172                    // The upstream consume hook returns null when cancelled.
173                    return Ok(None);
174                }
175            }
176            let table = Self::verification_table();
177            let id = S::Verification::parse_id(model.id().as_ref())?;
178            let mut delete = Sql::with(exec.engine(), "DELETE FROM ");
179            delete.ident(table);
180            delete.push(" WHERE ");
181            delete.compare_model::<S::Verification>(table, S::Verification::id_column(), " = ", id);
182            delete.push(" AND ");
183            delete.compare_model::<S::Verification>(
184                table,
185                S::Verification::value_column(),
186                " = ",
187                model.value(),
188            );
189            if exec.execute(delete).await? != 1 {
190                return Ok(None);
191            }
192            let mut siblings = Sql::with(exec.engine(), "DELETE FROM ");
193            siblings.ident(table);
194            siblings.push(" WHERE ");
195            siblings.compare_model::<S::Verification>(
196                table,
197                S::Verification::identifier_column(),
198                " = ",
199                identifier,
200            );
201            _ = exec.execute(siblings).await?;
202            Ok(Some(model))
203        }
204        .await;
205        let consumed = match outcome {
206            Ok(consumed) => {
207                transaction.commit().await?;
208                consumed
209            }
210            Err(error) => {
211                transaction.rollback().await?;
212                return Err(error);
213            }
214        };
215        if let Some(model) = &consumed {
216            let hook_context = self.hook_context(None);
217            for hook in self.hooks() {
218                hook.after_delete_verification(model, &hook_context).await?;
219            }
220        }
221        // Even an expired generation is removed. Older generations cannot
222        // become valid again when the newest token expires.
223        Ok(consumed)
224    }
225
226    /// `UPDATE ... RETURNING` every verification column; the winning snapshot
227    /// is captured by the update itself, not a later read.
228    async fn update_verifications_returning(
229        &self,
230        filters: &[(&'static str, crate::SqlValue)],
231        value: Option<String>,
232        expires_at: Option<DateTime<Utc>>,
233        updated_at_column: &'static str,
234    ) -> AuthResult<Option<S::Verification>> {
235        let table = Self::verification_table();
236        let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
237        sql.ident(table);
238        sql.push(" SET ");
239        sql.assign(
240            updated_at_column,
241            S::Verification::timestamp_value(updated_at_column, Utc::now()),
242        );
243        if let Some(value) = value {
244            sql.push(", ");
245            sql.assign(
246                S::Verification::value_column(),
247                S::Verification::column_value(S::Verification::value_column(), value.into()),
248            );
249        }
250        if let Some(expires_at) = expires_at {
251            sql.push(", ");
252            sql.assign(
253                S::Verification::expires_at_column(),
254                S::Verification::timestamp_value(S::Verification::expires_at_column(), expires_at),
255            );
256        }
257        for (index, (column, value)) in filters.iter().enumerate() {
258            sql.push(if index == 0 { " WHERE " } else { " AND " });
259            sql.compare_model::<S::Verification>(table, column, " = ", value.clone());
260        }
261        model::returning::<S::Verification>(&mut sql);
262        // The statement updates every matching row; the first returned row is the snapshot.
263        self.exec().fetch_optional::<S::Verification>(sql).await
264    }
265}
266
267#[async_trait]
268impl<S> VerificationStore<S> for SqlxStore<S>
269where
270    S: AuthSchema + Send + Sync,
271    S::Verification: SqlxVerificationModel,
272{
273    async fn create_verification_record(
274        &self,
275        data: VerificationCreation,
276        publication: VerificationPublication,
277    ) -> AuthResult<Option<VerificationSnapshot>> {
278        self.create_verification_record_with_connection(self.exec(), None, data, publication)
279            .await
280    }
281
282    async fn create_verification(
283        &self,
284        verification: CreateVerification,
285    ) -> AuthResult<S::Verification> {
286        self.create_verification_with_connection(self.exec(), None, verification)
287            .await
288    }
289
290    async fn get_verification(
291        &self,
292        identifier: &str,
293        value: &str,
294    ) -> AuthResult<Option<S::Verification>> {
295        let sql = Self::newest_generation(
296            self.exec(),
297            &[
298                (
299                    S::Verification::identifier_column(),
300                    " = ",
301                    identifier.into(),
302                ),
303                (S::Verification::value_column(), " = ", value.into()),
304                (
305                    S::Verification::expires_at_column(),
306                    " > ",
307                    S::Verification::timestamp_value(
308                        S::Verification::expires_at_column(),
309                        Utc::now(),
310                    ),
311                ),
312            ],
313        );
314        self.exec().fetch_optional(sql).await
315    }
316
317    async fn get_verification_by_value(&self, value: &str) -> AuthResult<Option<S::Verification>> {
318        let sql = Self::newest_generation(
319            self.exec(),
320            &[
321                (S::Verification::value_column(), " = ", value.into()),
322                (
323                    S::Verification::expires_at_column(),
324                    " > ",
325                    S::Verification::timestamp_value(
326                        S::Verification::expires_at_column(),
327                        Utc::now(),
328                    ),
329                ),
330            ],
331        );
332        self.exec().fetch_optional(sql).await
333    }
334
335    async fn get_verification_by_identifier(
336        &self,
337        identifier: &str,
338    ) -> AuthResult<Option<S::Verification>> {
339        let sql = Self::newest_generation(
340            self.exec(),
341            &[
342                (
343                    S::Verification::identifier_column(),
344                    " = ",
345                    identifier.into(),
346                ),
347                (
348                    S::Verification::expires_at_column(),
349                    " > ",
350                    S::Verification::timestamp_value(
351                        S::Verification::expires_at_column(),
352                        Utc::now(),
353                    ),
354                ),
355            ],
356        );
357        self.exec().fetch_optional(sql).await
358    }
359
360    async fn consume_verification(
361        &self,
362        identifier: &str,
363        value: &str,
364    ) -> AuthResult<Option<S::Verification>> {
365        Ok(self
366            .consume_verification_generation(identifier, Some(value))
367            .await?
368            .filter(|model| model.expires_at() >= Utc::now()))
369    }
370
371    async fn get_latest_verification_by_identifier(
372        &self,
373        identifier: &str,
374    ) -> AuthResult<Option<S::Verification>> {
375        let sql = Self::newest_generation(
376            self.exec(),
377            &[(
378                S::Verification::identifier_column(),
379                " = ",
380                identifier.into(),
381            )],
382        );
383        self.exec().fetch_optional(sql).await
384    }
385
386    async fn consume_verification_by_identifier(
387        &self,
388        identifier: &str,
389    ) -> AuthResult<Option<S::Verification>> {
390        Ok(self
391            .consume_verification_generation(identifier, None)
392            .await?
393            .filter(|model| model.expires_at() >= Utc::now()))
394    }
395
396    async fn consume_verification_snapshot(
397        &self,
398        identifier: &str,
399    ) -> AuthResult<Option<S::Verification>> {
400        self.consume_verification_generation(identifier, None).await
401    }
402
403    async fn delete_verifications_by_identifier(&self, identifier: &str) -> AuthResult<()> {
404        let table = Self::verification_table();
405        let mut select = model::select_model::<S::Verification>(self.exec());
406        select.push(" WHERE ");
407        select.compare_model::<S::Verification>(
408            table,
409            S::Verification::identifier_column(),
410            " = ",
411            identifier,
412        );
413        select.push(" LIMIT 1");
414        let Some(model) = self
415            .exec()
416            .fetch_optional::<S::Verification>(select)
417            .await?
418        else {
419            return Ok(());
420        };
421        // Upstream deleteVerificationByIdentifier uses deleteWithHooks, not
422        // deleteManyWithHooks: one snapshot drives the lifecycle callbacks,
423        // while the identifier predicate invalidates every matching row.
424        let hook_context = self.hook_context(None);
425        for hook in self.hooks() {
426            if hook
427                .before_delete_verification(&model, &hook_context)
428                .await?
429                .is_cancelled()
430            {
431                return Ok(());
432            }
433        }
434        let mut delete = Sql::with(self.exec().engine(), "DELETE FROM ");
435        delete.ident(table);
436        delete.push(" WHERE ");
437        delete.compare_model::<S::Verification>(
438            table,
439            S::Verification::identifier_column(),
440            " = ",
441            identifier,
442        );
443        _ = self.exec().execute(delete).await?;
444        for hook in self.hooks() {
445            hook.after_delete_verification(&model, &hook_context)
446                .await?;
447        }
448        Ok(())
449    }
450
451    async fn compare_and_swap_verification(
452        &self,
453        id: &str,
454        expected_value: &str,
455        value: &str,
456        expires_at: DateTime<Utc>,
457    ) -> AuthResult<bool> {
458        let updated_at_column = S::Verification::updated_at_column().ok_or_else(|| {
459            AuthError::internal("the verification schema has no update timestamp binding")
460        })?;
461        let parsed_id = S::Verification::parse_id(id)?;
462        let mut update = UpdateVerification {
463            value: Some(value.to_owned()),
464            expires_at: Some(expires_at),
465        };
466        let hook_context = self.hook_context(None);
467        for hook in self.hooks() {
468            if hook
469                .before_update_verification(id, &mut update, &hook_context)
470                .await?
471                .is_cancelled()
472            {
473                return Ok(false);
474            }
475        }
476        // RETURNING preserves the winning snapshot in the update itself. A
477        // second SELECT can lose the row to consumption or observe a later
478        // mutation, including changes made by application database triggers.
479        let Some(model) = self
480            .update_verifications_returning(
481                &[
482                    (S::Verification::id_column(), parsed_id),
483                    (S::Verification::value_column(), expected_value.into()),
484                ],
485                update.value,
486                update.expires_at,
487                updated_at_column,
488            )
489            .await?
490        else {
491            return Ok(false);
492        };
493        for hook in self.hooks() {
494            hook.after_update_verification(&model, &hook_context)
495                .await?;
496        }
497        Ok(true)
498    }
499
500    async fn reserve_verification(&self, verification: CreateVerification) -> AuthResult<bool> {
501        let logical = verification.identifier.clone();
502        Ok(self
503            .reserve_verification_record(&logical, verification)
504            .await?
505            .is_some())
506    }
507
508    async fn reserve_verification_record(
509        &self,
510        logical_identifier: &str,
511        verification: CreateVerification,
512    ) -> AuthResult<Option<S::Verification>> {
513        let (encoded, digest) = verification_reservation_key(logical_identifier);
514        let id = S::Verification::parse_reservation_id(&encoded, digest)?;
515        let active = S::Verification::new_active(Some(id.clone()), verification, Utc::now());
516        match model::insert::<S::Verification>(self.exec(), &active).await {
517            Ok(model) => Ok(Some(model)),
518            Err(error) => {
519                let table = Self::verification_table();
520                let mut existing = model::select_model::<S::Verification>(self.exec());
521                existing.push(" WHERE ");
522                existing.compare_model::<S::Verification>(
523                    table,
524                    S::Verification::id_column(),
525                    " = ",
526                    id,
527                );
528                existing.push(" LIMIT 1");
529                if self
530                    .exec()
531                    .fetch_optional::<S::Verification>(existing)
532                    .await?
533                    .is_some()
534                {
535                    Ok(None)
536                } else {
537                    Err(error)
538                }
539            }
540        }
541    }
542
543    async fn update_verification_by_identifier(
544        &self,
545        identifier: &str,
546        data: UpdateVerification,
547    ) -> AuthResult<Option<VerificationSnapshot>> {
548        let hook_context = self.hook_context(None);
549        let mut admitted = data.clone();
550        for hook in self.hooks() {
551            // Source gives every before-update hook the original patch, then
552            // merges each returned mutation into the admitted patch.
553            let mut candidate = data.clone();
554            if hook
555                .before_update_verification(identifier, &mut candidate, &hook_context)
556                .await?
557                .is_cancelled()
558            {
559                return Ok(None);
560            }
561            if candidate.value.is_some() {
562                admitted.value = candidate.value;
563            }
564            if candidate.expires_at.is_some() {
565                admitted.expires_at = candidate.expires_at;
566            }
567        }
568        let updated_at = S::Verification::updated_at_column().ok_or_else(|| {
569            AuthError::NotImplemented(
570                "Verification update snapshots require an updatedAt column".into(),
571            )
572        })?;
573        let model = self
574            .update_verifications_returning(
575                &[(S::Verification::identifier_column(), identifier.into())],
576                admitted.value,
577                admitted.expires_at,
578                updated_at,
579            )
580            .await?;
581        let snapshot = model.as_ref().map(VerificationSnapshot::from_model);
582        for hook in self.hooks() {
583            hook.after_update_verification_record(snapshot.as_ref(), &hook_context)
584                .await?;
585        }
586        Ok(snapshot)
587    }
588
589    async fn delete_verification(&self, id: &str) -> AuthResult<()> {
590        let verification_id = <S::Verification as SqlxVerificationModel>::parse_id(id)?;
591        let table = Self::verification_table();
592        let mut select = model::select_model::<S::Verification>(self.exec());
593        select.push(" WHERE ");
594        select.compare_model::<S::Verification>(
595            table,
596            S::Verification::id_column(),
597            " = ",
598            verification_id.clone(),
599        );
600        select.push(" LIMIT 1");
601        let verification = self
602            .exec()
603            .fetch_optional::<S::Verification>(select)
604            .await?;
605        let hook_context = self.hook_context(None);
606        if let Some(verification) = &verification {
607            for hook in self.hooks() {
608                if hook
609                    .before_delete_verification(verification, &hook_context)
610                    .await?
611                    .is_cancelled()
612                {
613                    return Err(cancelled_by_hook("verification deletion"));
614                }
615            }
616        }
617        let mut delete = Sql::with(self.exec().engine(), "DELETE FROM ");
618        delete.ident(table);
619        delete.push(" WHERE ");
620        delete.compare_model::<S::Verification>(
621            table,
622            S::Verification::id_column(),
623            " = ",
624            verification_id,
625        );
626        _ = self.exec().execute(delete).await?;
627        if let Some(verification) = &verification {
628            for hook in self.hooks() {
629                hook.after_delete_verification(verification, &hook_context)
630                    .await?;
631            }
632        }
633        Ok(())
634    }
635
636    async fn delete_expired_verifications(&self) -> AuthResult<usize> {
637        // Source's strict `expiresAt < new Date()` cleanup keeps proofs whose
638        // deadline equals the current millisecond, including zero-TTL OTPs.
639        let deadline = Utc::now().trunc_subsecs(3);
640        let table = Self::verification_table();
641        let mut select = model::select_model::<S::Verification>(self.exec());
642        select.push(" WHERE ");
643        select.compare_model::<S::Verification>(
644            table,
645            S::Verification::expires_at_column(),
646            " < ",
647            S::Verification::timestamp_value(S::Verification::expires_at_column(), deadline),
648        );
649        select.push(" LIMIT ");
650        select.bind(self.find_many_limit());
651        let snapshots: Vec<S::Verification> = self.exec().fetch_all(select).await?;
652        let hook_context = self.hook_context(None);
653        // deleteManyWithHooks snapshots a findMany page before mutation.
654        // Any before-hook veto cancels the entire batch, including rows whose
655        // callbacks already ran; after hooks receive those original snapshots.
656        for model in &snapshots {
657            for hook in self.hooks() {
658                if hook
659                    .before_delete_verification(model, &hook_context)
660                    .await?
661                    .is_cancelled()
662                {
663                    return Ok(0);
664                }
665            }
666        }
667        let mut delete = Sql::with(self.exec().engine(), "DELETE FROM ");
668        delete.ident(table);
669        delete.push(" WHERE ");
670        delete.compare_model::<S::Verification>(
671            table,
672            S::Verification::expires_at_column(),
673            " < ",
674            S::Verification::timestamp_value(S::Verification::expires_at_column(), deadline),
675        );
676        let deleted = self.exec().execute(delete).await?;
677        for model in &snapshots {
678            for hook in self.hooks() {
679                hook.after_delete_verification(model, &hook_context).await?;
680            }
681        }
682        usize::try_from(deleted)
683            .map_err(|_error| AuthError::internal("Verification cleanup count overflow"))
684    }
685}
alibi_sqlx/store/verifications.rs

alibi_sqlx/store/
verifications.rs
