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

alibi_sqlx/store/
verifications.rs
