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_id};
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 = self
181            .consume_generation_in(&transaction, identifier, expected_value)
182            .await;
183        let consumed = match outcome {
184            Ok(consumed) => {
185                transaction.commit().await?;
186                consumed
187            }
188            Err(error) => {
189                transaction.rollback().await?;
190                return Err(error);
191            }
192        };
193        if let Some(model) = &consumed {
194            let hook_context = self.hook_context(None);
195            for hook in self.hooks() {
196                hook.after_delete_verification(model, &hook_context).await?;
197            }
198        }
199        // Even an expired generation is removed. Older generations cannot
200        // become valid again when the newest token expires.
201        Ok(consumed)
202    }
203
204    /// Lock, gate and delete the newest generation on `transaction`. The
205    /// caller commits and runs delete after-hooks.
206    pub(crate) async fn consume_generation_in(
207        &self,
208        transaction: &SqlxTransaction,
209        identifier: &str,
210        expected_value: Option<&str>,
211    ) -> AuthResult<Option<S::Verification>> {
212        let exec = Exec::tx(transaction);
213        let mut select = Self::newest_generation(
214            exec,
215            &[(
216                S::Verification::identifier_column(),
217                " = ",
218                identifier.into(),
219            )],
220        );
221        lock_exclusive(&mut select);
222        let Some(model) = exec.fetch_optional::<S::Verification>(select).await? else {
223            return Ok(None);
224        };
225        if expected_value.is_some_and(|expected| model.value() != expected) {
226            return Ok(None);
227        }
228        let hook_context = self.hook_context(Some(transaction));
229        for hook in self.hooks() {
230            if hook
231                .before_delete_verification(&model, &hook_context)
232                .await?
233                .is_cancelled()
234            {
235                // The upstream consume hook returns null when cancelled.
236                return Ok(None);
237            }
238        }
239        let table = Self::verification_table();
240        let id = S::Verification::parse_id(model.id().as_ref())?;
241        let mut delete = Sql::with(exec.engine(), "DELETE FROM ");
242        delete.ident(table);
243        delete.push(" WHERE ");
244        delete.compare_model::<S::Verification>(table, S::Verification::id_column(), " = ", id);
245        delete.push(" AND ");
246        delete.compare_model::<S::Verification>(
247            table,
248            S::Verification::value_column(),
249            " = ",
250            model.value(),
251        );
252        if exec.execute(delete).await? != 1 {
253            return Ok(None);
254        }
255        let mut siblings = Sql::with(exec.engine(), "DELETE FROM ");
256        siblings.ident(table);
257        siblings.push(" WHERE ");
258        siblings.compare_model::<S::Verification>(
259            table,
260            S::Verification::identifier_column(),
261            " = ",
262            identifier,
263        );
264        _ = exec.execute(siblings).await?;
265        Ok(Some(model))
266    }
267
268    /// `UPDATE ... RETURNING` every verification column; the winning snapshot
269    /// is captured by the update itself, not a later read.
270    async fn update_verifications_returning(
271        &self,
272        filters: &[(&'static str, SqlValue)],
273        value: Option<String>,
274        expires_at: Option<DateTime<Utc>>,
275        updated_at_column: &'static str,
276    ) -> AuthResult<Option<S::Verification>> {
277        let table = Self::verification_table();
278        let mut sql = Sql::with(self.exec().engine(), "UPDATE ");
279        sql.ident(table);
280        sql.push(" SET ");
281        sql.assign(
282            updated_at_column,
283            S::Verification::timestamp_value(updated_at_column, Utc::now()),
284        );
285        if let Some(value) = value {
286            sql.push(", ");
287            sql.assign(
288                S::Verification::value_column(),
289                S::Verification::column_value(S::Verification::value_column(), value.into()),
290            );
291        }
292        if let Some(expires_at) = expires_at {
293            sql.push(", ");
294            sql.assign(
295                S::Verification::expires_at_column(),
296                S::Verification::timestamp_value(S::Verification::expires_at_column(), expires_at),
297            );
298        }
299        for (index, (column, value)) in filters.iter().enumerate() {
300            sql.push(if index == 0 { " WHERE " } else { " AND " });
301            sql.compare_model::<S::Verification>(table, column, " = ", value.clone());
302        }
303        model::returning::<S::Verification>(&mut sql);
304        self.exec().fetch_optional::<S::Verification>(sql).await
305    }
306}
307
308#[async_trait]
309impl<S> VerificationStore<S> for SqlxStore<S>
310where
311    S: AuthSchema + Send + Sync,
312    S::Verification: SqlxVerificationModel,
313{
314    async fn create_verification_record(
315        &self,
316        data: VerificationCreation,
317        publication: VerificationPublication,
318    ) -> AuthResult<Option<VerificationSnapshot>> {
319        self.create_verification_record_with_connection(self.exec(), None, data, publication)
320            .await
321    }
322
323    async fn create_verification(
324        &self,
325        verification: CreateVerification,
326    ) -> AuthResult<S::Verification> {
327        self.create_verification_with_connection(self.exec(), None, verification)
328            .await
329    }
330
331    async fn get_verification(
332        &self,
333        identifier: &str,
334        value: &str,
335    ) -> AuthResult<Option<S::Verification>> {
336        let sql = Self::newest_generation(
337            self.exec(),
338            &[
339                (
340                    S::Verification::identifier_column(),
341                    " = ",
342                    identifier.into(),
343                ),
344                (S::Verification::value_column(), " = ", value.into()),
345                Self::unexpired(),
346            ],
347        );
348        self.exec().fetch_optional(sql).await
349    }
350
351    async fn get_verification_by_value(&self, value: &str) -> AuthResult<Option<S::Verification>> {
352        let sql = Self::newest_generation(
353            self.exec(),
354            &[
355                (S::Verification::value_column(), " = ", value.into()),
356                Self::unexpired(),
357            ],
358        );
359        self.exec().fetch_optional(sql).await
360    }
361
362    async fn get_verification_by_identifier(
363        &self,
364        identifier: &str,
365    ) -> AuthResult<Option<S::Verification>> {
366        let sql = Self::newest_generation(
367            self.exec(),
368            &[
369                (
370                    S::Verification::identifier_column(),
371                    " = ",
372                    identifier.into(),
373                ),
374                Self::unexpired(),
375            ],
376        );
377        self.exec().fetch_optional(sql).await
378    }
379
380    async fn consume_verification(
381        &self,
382        identifier: &str,
383        value: &str,
384    ) -> AuthResult<Option<S::Verification>> {
385        Ok(self
386            .consume_verification_generation(identifier, Some(value))
387            .await?
388            .filter(|model| model.expires_at() >= Utc::now()))
389    }
390
391    async fn get_latest_verification_by_identifier(
392        &self,
393        identifier: &str,
394    ) -> AuthResult<Option<S::Verification>> {
395        let sql = Self::newest_generation(
396            self.exec(),
397            &[(
398                S::Verification::identifier_column(),
399                " = ",
400                identifier.into(),
401            )],
402        );
403        self.exec().fetch_optional(sql).await
404    }
405
406    async fn consume_verification_by_identifier(
407        &self,
408        identifier: &str,
409    ) -> AuthResult<Option<S::Verification>> {
410        Ok(self
411            .consume_verification_generation(identifier, None)
412            .await?
413            .filter(|model| model.expires_at() >= Utc::now()))
414    }
415
416    async fn consume_verification_snapshot(
417        &self,
418        identifier: &str,
419    ) -> AuthResult<Option<S::Verification>> {
420        self.consume_verification_generation(identifier, None).await
421    }
422
423    async fn delete_verifications_by_identifier(&self, identifier: &str) -> AuthResult<()> {
424        let table = Self::verification_table();
425        let mut select = model::select_model::<S::Verification>(self.exec());
426        select.push(" WHERE ");
427        select.compare_model::<S::Verification>(
428            table,
429            S::Verification::identifier_column(),
430            " = ",
431            identifier,
432        );
433        select.push(" LIMIT 1");
434        let Some(model) = self
435            .exec()
436            .fetch_optional::<S::Verification>(select)
437            .await?
438        else {
439            return Ok(());
440        };
441        // Upstream deleteVerificationByIdentifier uses deleteWithHooks, not
442        // deleteManyWithHooks: one snapshot drives the lifecycle callbacks,
443        // while the identifier predicate invalidates every matching row.
444        let hook_context = self.hook_context(None);
445        for hook in self.hooks() {
446            if hook
447                .before_delete_verification(&model, &hook_context)
448                .await?
449                .is_cancelled()
450            {
451                return Ok(());
452            }
453        }
454        let mut delete = Sql::with(self.exec().engine(), "DELETE FROM ");
455        delete.ident(table);
456        delete.push(" WHERE ");
457        delete.compare_model::<S::Verification>(
458            table,
459            S::Verification::identifier_column(),
460            " = ",
461            identifier,
462        );
463        _ = self.exec().execute(delete).await?;
464        for hook in self.hooks() {
465            hook.after_delete_verification(&model, &hook_context)
466                .await?;
467        }
468        Ok(())
469    }
470
471    async fn compare_and_swap_verification(
472        &self,
473        id: &str,
474        expected_value: &str,
475        value: &str,
476        expires_at: DateTime<Utc>,
477    ) -> AuthResult<bool> {
478        let updated_at_column = S::Verification::updated_at_column().ok_or_else(|| {
479            AuthError::internal("the verification schema has no update timestamp binding")
480        })?;
481        let parsed_id = S::Verification::parse_id(id)?;
482        let mut update = UpdateVerification {
483            value: Some(value.to_owned()),
484            expires_at: Some(expires_at),
485        };
486        let hook_context = self.hook_context(None);
487        for hook in self.hooks() {
488            if hook
489                .before_update_verification(id, &mut update, &hook_context)
490                .await?
491                .is_cancelled()
492            {
493                return Ok(false);
494            }
495        }
496        // RETURNING captures the winning snapshot; a second SELECT could lose
497        // the row to consumption or observe a later mutation.
498        let Some(model) = self
499            .update_verifications_returning(
500                &[
501                    (S::Verification::id_column(), parsed_id),
502                    (S::Verification::value_column(), expected_value.into()),
503                ],
504                update.value,
505                update.expires_at,
506                updated_at_column,
507            )
508            .await?
509        else {
510            return Ok(false);
511        };
512        for hook in self.hooks() {
513            hook.after_update_verification(&model, &hook_context)
514                .await?;
515        }
516        Ok(true)
517    }
518
519    async fn reserve_verification(&self, verification: CreateVerification) -> AuthResult<bool> {
520        let logical = verification.identifier.clone();
521        Ok(self
522            .reserve_verification_record(&logical, verification)
523            .await?
524            .is_some())
525    }
526
527    async fn reserve_verification_record(
528        &self,
529        logical_identifier: &str,
530        verification: CreateVerification,
531    ) -> AuthResult<Option<S::Verification>> {
532        let (encoded, digest) =
533            verification_reservation_id(logical_identifier, &self.config.advanced.database);
534        let id = S::Verification::parse_reservation_id(&encoded, digest)?;
535        let active = S::Verification::new_active(Some(id.clone()), verification, Utc::now());
536        match model::insert::<S::Verification>(self.exec(), &active).await {
537            Ok(model) => Ok(Some(model)),
538            Err(error) => {
539                let table = Self::verification_table();
540                let mut existing = model::select_model::<S::Verification>(self.exec());
541                existing.push(" WHERE ");
542                existing.compare_model::<S::Verification>(
543                    table,
544                    S::Verification::id_column(),
545                    " = ",
546                    id,
547                );
548                existing.push(" LIMIT 1");
549                if self
550                    .exec()
551                    .fetch_optional::<S::Verification>(existing)
552                    .await?
553                    .is_some()
554                {
555                    Ok(None)
556                } else {
557                    Err(error)
558                }
559            }
560        }
561    }
562
563    async fn update_verification_by_identifier(
564        &self,
565        identifier: &str,
566        data: UpdateVerification,
567    ) -> AuthResult<Option<VerificationSnapshot>> {
568        let hook_context = self.hook_context(None);
569        let mut admitted = data.clone();
570        for hook in self.hooks() {
571            // Source gives every before-update hook the original patch, then
572            // merges each returned mutation into the admitted patch.
573            let mut candidate = data.clone();
574            if hook
575                .before_update_verification(identifier, &mut candidate, &hook_context)
576                .await?
577                .is_cancelled()
578            {
579                return Ok(None);
580            }
581            if candidate.value.is_some() {
582                admitted.value = candidate.value;
583            }
584            if candidate.expires_at.is_some() {
585                admitted.expires_at = candidate.expires_at;
586            }
587        }
588        let updated_at = S::Verification::updated_at_column().ok_or_else(|| {
589            AuthError::NotImplemented(
590                "Verification update snapshots require an updatedAt column".into(),
591            )
592        })?;
593        let model = self
594            .update_verifications_returning(
595                &[(S::Verification::identifier_column(), identifier.into())],
596                admitted.value,
597                admitted.expires_at,
598                updated_at,
599            )
600            .await?;
601        let snapshot = model.as_ref().map(VerificationSnapshot::from_model);
602        for hook in self.hooks() {
603            hook.after_update_verification_record(snapshot.as_ref(), &hook_context)
604                .await?;
605        }
606        Ok(snapshot)
607    }
608
609    async fn delete_verification(&self, id: &str) -> AuthResult<()> {
610        let verification_id = <S::Verification as SqlxVerificationModel>::parse_id(id)?;
611        let table = Self::verification_table();
612        let mut select = model::select_model::<S::Verification>(self.exec());
613        select.push(" WHERE ");
614        select.compare_model::<S::Verification>(
615            table,
616            S::Verification::id_column(),
617            " = ",
618            verification_id.clone(),
619        );
620        select.push(" LIMIT 1");
621        let verification = self
622            .exec()
623            .fetch_optional::<S::Verification>(select)
624            .await?;
625        let hook_context = self.hook_context(None);
626        if let Some(verification) = &verification {
627            for hook in self.hooks() {
628                if hook
629                    .before_delete_verification(verification, &hook_context)
630                    .await?
631                    .is_cancelled()
632                {
633                    return Err(cancelled_by_hook("verification deletion"));
634                }
635            }
636        }
637        let mut delete = Sql::with(self.exec().engine(), "DELETE FROM ");
638        delete.ident(table);
639        delete.push(" WHERE ");
640        delete.compare_model::<S::Verification>(
641            table,
642            S::Verification::id_column(),
643            " = ",
644            verification_id,
645        );
646        _ = self.exec().execute(delete).await?;
647        if let Some(verification) = &verification {
648            for hook in self.hooks() {
649                hook.after_delete_verification(verification, &hook_context)
650                    .await?;
651            }
652        }
653        Ok(())
654    }
655
656    async fn delete_expired_verifications(&self) -> AuthResult<usize> {
657        // Source's strict `expiresAt < new Date()` cleanup keeps proofs whose
658        // deadline equals the current millisecond, including zero-TTL OTPs.
659        let deadline = Utc::now().trunc_subsecs(3);
660        let table = Self::verification_table();
661        let mut select = model::select_model::<S::Verification>(self.exec());
662        select.push(" WHERE ");
663        select.compare_model::<S::Verification>(
664            table,
665            S::Verification::expires_at_column(),
666            " < ",
667            S::Verification::timestamp_value(S::Verification::expires_at_column(), deadline),
668        );
669        select.push(" LIMIT ");
670        select.bind(self.find_many_limit());
671        let snapshots: Vec<S::Verification> = self.exec().fetch_all(select).await?;
672        let hook_context = self.hook_context(None);
673        // deleteManyWithHooks snapshots a findMany page before mutation.
674        // Any before-hook veto cancels the entire batch, including rows whose
675        // callbacks already ran; after hooks receive those original snapshots.
676        for model in &snapshots {
677            for hook in self.hooks() {
678                if hook
679                    .before_delete_verification(model, &hook_context)
680                    .await?
681                    .is_cancelled()
682                {
683                    return Ok(0);
684                }
685            }
686        }
687        let mut delete = Sql::with(self.exec().engine(), "DELETE FROM ");
688        delete.ident(table);
689        delete.push(" WHERE ");
690        delete.compare_model::<S::Verification>(
691            table,
692            S::Verification::expires_at_column(),
693            " < ",
694            S::Verification::timestamp_value(S::Verification::expires_at_column(), deadline),
695        );
696        let deleted = self.exec().execute(delete).await?;
697        for model in &snapshots {
698            for hook in self.hooks() {
699                hook.after_delete_verification(model, &hook_context).await?;
700            }
701        }
702        usize::try_from(deleted)
703            .map_err(|_error| AuthError::internal("Verification cleanup count overflow"))
704    }
705}
alibi_sqlx/store/verifications.rs

alibi_sqlx/store/
verifications.rs
