alphonse/zero2prod
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

Comment posting is idempotent + tests

Browse Source
This commit is contained in:
Alphonse Paix
2025-10-05 15:01:57 +02:00
parent 8f62c2513e
commit d96a29ee73
15 changed files with 289 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/idempotency/persistance.rs
View File

@@ -7,7 +7,6 @@ use axum::{
use reqwest::StatusCode;
use sqlx::{Executor, PgPool, Postgres, Transaction};
use std::str::FromStr;
use uuid::Uuid;
#[derive(Debug, sqlx::Type)]
#[sqlx(type_name = "header_pair")]
@@ -23,7 +22,6 @@ struct HeaderPairRecord {
pub async fn get_saved_response(
connection_pool: &PgPool,
idempotency_key: &IdempotencyKey,
user_id: Uuid,
) -> Result<Option<Response>, anyhow::Error> {
let saved_response = sqlx::query!(
r#"
@@ -32,11 +30,8 @@ pub async fn get_saved_response(
response_headers as "response_headers!: Vec<HeaderPairRecord>",
response_body as "response_body!"
FROM idempotency
WHERE
user_id = $1
AND idempotency_key = $2
WHERE idempotency_key = $1
"#,
user_id,
idempotency_key.as_ref()
)
.fetch_optional(connection_pool)
@@ -61,7 +56,6 @@ pub async fn get_saved_response(
pub async fn save_response(
mut transaction: Transaction<'static, Postgres>,
idempotency_key: &IdempotencyKey,
user_id: Uuid,
response: Response<Body>,
) -> Result<Response<Body>, anyhow::Error> {
let status_code = response.status().as_u16() as i16;
@@ -80,14 +74,11 @@ pub async fn save_response(
r#"
UPDATE idempotency
SET
response_status_code = $3,
response_headers = $4,
response_body = $5
WHERE
user_id = $1
AND idempotency_key = $2
response_status_code = $2,
response_headers = $3,
response_body = $4
WHERE idempotency_key = $1
"#,
user_id,
idempotency_key.as_ref(),
status_code,
headers,
@@ -109,23 +100,21 @@ pub enum NextAction {
pub async fn try_processing(
connection_pool: &PgPool,
idempotency_key: &IdempotencyKey,
user_id: Uuid,
) -> Result<NextAction, anyhow::Error> {
let mut transaction = connection_pool.begin().await?;
let query = sqlx::query!(
r#"
INSERT INTO idempotency (user_id, idempotency_key, created_at)
VALUES ($1, $2, now())
INSERT INTO idempotency (idempotency_key, created_at)
VALUES ($1, now())
ON CONFLICT DO NOTHING
"#,
user_id,
idempotency_key.as_ref()
);
let n_inserted_rows = transaction.execute(query).await?.rows_affected();
if n_inserted_rows > 0 {
Ok(NextAction::StartProcessing(transaction))
} else {
let saved_response = get_saved_response(connection_pool, idempotency_key, user_id)
let saved_response = get_saved_response(connection_pool, idempotency_key)
.await?
.ok_or_else(|| anyhow::anyhow!("Could not find saved response."))?;
Ok(NextAction::ReturnSavedResponse(saved_response))