codlet_sqlx/
token.rs

1//! SQLite implementation of [`codlet_core::store::token::FormTokenStore`].
2
3use codlet_core::hashing::LookupKey;
4use codlet_core::state::{TokenConsumeOutcome, classify_token_consume};
5use codlet_core::store::error::StoreError;
6use codlet_core::store::token::{FormTokenRecord, FormTokenStore, TokenSubject};
7
8use crate::SqliteStore;
9
10impl FormTokenStore for SqliteStore {
11    async fn insert_form_token(&self, record: FormTokenRecord) -> Result<(), StoreError> {
12        sqlx::query(
13            "INSERT INTO codlet_form_tokens
14             (lookup_key, key_version, subject_kind, purpose, bound_resource, issued_at, expires_at)
15             VALUES (?, ?, ?, ?, ?, ?, ?)",
16        )
17        .bind(record.lookup_key.as_str())
18        .bind(record.key_version.as_str())
19        .bind(record.subject.as_binding_str())
20        .bind(&record.purpose)
21        .bind(record.bound_resource.as_deref())
22        .bind(record.issued_at as i64)
23        .bind(record.expires_at as i64)
24        .execute(&self.pool)
25        .await
26        .map_err(|e| StoreError::Backend(e.to_string()))?;
27        Ok(())
28    }
29
30    async fn consume_form_token(
31        &self,
32        lookup_key: &LookupKey,
33        subject: &TokenSubject,
34        purpose: &str,
35        bound_resource: Option<&str>,
36        now: u64,
37    ) -> Result<(TokenConsumeOutcome, Option<String>), StoreError> {
38        let now_i = now as i64;
39        let lk = lookup_key.as_str();
40        let subj = subject.as_binding_str();
41        let br = bound_resource.unwrap_or("");
42
43        // Atomic conditional UPDATE — the single-winner transition (RFC-022).
44        let update = sqlx::query(
45            "UPDATE codlet_form_tokens
46             SET consumed_at = ?
47             WHERE lookup_key    = ?
48               AND subject_kind  = ?
49               AND purpose       = ?
50               AND COALESCE(bound_resource, '') = ?
51               AND expires_at    > ?
52               AND consumed_at   IS NULL",
53        )
54        .bind(now_i)
55        .bind(lk)
56        .bind(&subj)
57        .bind(purpose)
58        .bind(br)
59        .bind(now_i)
60        .execute(&self.pool)
61        .await
62        .map_err(|e| StoreError::Backend(e.to_string()))?;
63
64        let changed = update.rows_affected() as usize;
65        if changed > 1 {
66            return Err(StoreError::InvariantViolation(format!(
67                "consume_form_token changed {changed} rows for lookup_key={lk}"
68            )));
69        }
70        if changed == 1 {
71            return Ok((TokenConsumeOutcome::Proceed, None));
72        }
73
74        // changed == 0: follow-up SELECT to classify why (no race-sensitive write).
75        let row: Option<(Option<i64>, Option<String>, Option<String>)> = sqlx::query_as(
76            "SELECT consumed_at, result_ref, bound_resource
77             FROM codlet_form_tokens
78             WHERE lookup_key = ? AND subject_kind = ? AND purpose = ?
79             LIMIT 1",
80        )
81        .bind(lk)
82        .bind(&subj)
83        .bind(purpose)
84        .fetch_optional(&self.pool)
85        .await
86        .map_err(|e| StoreError::Backend(e.to_string()))?;
87
88        let found = row.is_some();
89        let (already_consumed, stored_rr, stored_br) = row
90            .map(|(ca, rr, b)| (ca.is_some(), rr, b))
91            .unwrap_or((false, None, None));
92
93        let binding_ok = match bound_resource {
94            Some(expected) => stored_br.as_deref().unwrap_or("") == expected,
95            None => true,
96        };
97
98        let outcome = classify_token_consume(0, found, already_consumed, binding_ok);
99        let result_ref = if outcome == TokenConsumeOutcome::Replay {
100            stored_rr
101        } else {
102            None
103        };
104        Ok((outcome, result_ref))
105    }
106
107    async fn set_token_result(
108        &self,
109        lookup_key: &LookupKey,
110        result_ref: &str,
111    ) -> Result<(), StoreError> {
112        sqlx::query(
113            "UPDATE codlet_form_tokens
114             SET result_ref = ?
115             WHERE lookup_key = ? AND consumed_at IS NOT NULL",
116        )
117        .bind(result_ref)
118        .bind(lookup_key.as_str())
119        .execute(&self.pool)
120        .await
121        .map_err(|e| StoreError::Backend(e.to_string()))?;
122        Ok(())
123    }
124}
codlet_sqlx/token.rs

codlet_sqlx/
token.rs
