codlet-sqlx 0.14.4

//! SQLite implementation of [`codlet_core::store::token::FormTokenStore`].

use codlet_core::hashing::LookupKey;
use codlet_core::state::{TokenConsumeOutcome, classify_token_consume};
use codlet_core::store::error::StoreError;
use codlet_core::store::token::{FormTokenRecord, FormTokenStore, TokenSubject};

use crate::SqliteStore;

impl FormTokenStore for SqliteStore {
    async fn insert_form_token(&self, record: FormTokenRecord) -> Result<(), StoreError> {
        sqlx::query(
            "INSERT INTO codlet_form_tokens
             (lookup_key, key_version, subject_kind, purpose, bound_resource, issued_at, expires_at)
             VALUES (?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(record.lookup_key.as_str())
        .bind(record.key_version.as_str())
        .bind(record.subject.as_binding_str())
        .bind(&record.purpose)
        .bind(record.bound_resource.as_deref())
        .bind(record.issued_at as i64)
        .bind(record.expires_at as i64)
        .execute(&self.pool)
        .await
        .map_err(|e| StoreError::Backend(e.to_string()))?;
        Ok(())
    }

    async fn consume_form_token(
        &self,
        lookup_key: &LookupKey,
        subject: &TokenSubject,
        purpose: &str,
        bound_resource: Option<&str>,
        now: u64,
    ) -> Result<(TokenConsumeOutcome, Option<String>), StoreError> {
        let now_i = now as i64;
        let lk = lookup_key.as_str();
        let subj = subject.as_binding_str();
        let br = bound_resource.unwrap_or("");

        // Atomic conditional UPDATE — the single-winner transition (RFC-022).
        let update = sqlx::query(
            "UPDATE codlet_form_tokens
             SET consumed_at = ?
             WHERE lookup_key    = ?
               AND subject_kind  = ?
               AND purpose       = ?
               AND COALESCE(bound_resource, '') = ?
               AND expires_at    > ?
               AND consumed_at   IS NULL",
        )
        .bind(now_i)
        .bind(lk)
        .bind(&subj)
        .bind(purpose)
        .bind(br)
        .bind(now_i)
        .execute(&self.pool)
        .await
        .map_err(|e| StoreError::Backend(e.to_string()))?;

        let changed = update.rows_affected() as usize;
        if changed > 1 {
            return Err(StoreError::InvariantViolation(format!(
                "consume_form_token changed {changed} rows for lookup_key={lk}"
            )));
        }
        if changed == 1 {
            return Ok((TokenConsumeOutcome::Proceed, None));
        }

        // changed == 0: follow-up SELECT to classify why (no race-sensitive write).
        let row: Option<(Option<i64>, Option<String>, Option<String>)> = sqlx::query_as(
            "SELECT consumed_at, result_ref, bound_resource
             FROM codlet_form_tokens
             WHERE lookup_key = ? AND subject_kind = ? AND purpose = ?
             LIMIT 1",
        )
        .bind(lk)
        .bind(&subj)
        .bind(purpose)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| StoreError::Backend(e.to_string()))?;

        let found = row.is_some();
        let (already_consumed, stored_rr, stored_br) = row
            .map(|(ca, rr, b)| (ca.is_some(), rr, b))
            .unwrap_or((false, None, None));

        let binding_ok = match bound_resource {
            Some(expected) => stored_br.as_deref().unwrap_or("") == expected,
            None => true,
        };

        let outcome = classify_token_consume(0, found, already_consumed, binding_ok);
        let result_ref = if outcome == TokenConsumeOutcome::Replay {
            stored_rr
        } else {
            None
        };
        Ok((outcome, result_ref))
    }

    async fn set_token_result(
        &self,
        lookup_key: &LookupKey,
        result_ref: &str,
    ) -> Result<(), StoreError> {
        sqlx::query(
            "UPDATE codlet_form_tokens
             SET result_ref = ?
             WHERE lookup_key = ? AND consumed_at IS NOT NULL",
        )
        .bind(result_ref)
        .bind(lookup_key.as_str())
        .execute(&self.pool)
        .await
        .map_err(|e| StoreError::Backend(e.to_string()))?;
        Ok(())
    }
}