Rust

[package]
name = "constaia-axum"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = { version = "0.8", features = ["multipart"] }
base64 = "0.22"
hmac = "0.12"
reqwest = { version = "0.12", features = ["json", "multipart"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha2 = "0.10"
thiserror = "2"
tokio = { version = "1", features = ["full"] }
uuid = { version = "1", features = ["v4"] }

CONSTAIA_API_KEY=ck_test_...
CONSTAIA_WEBHOOK_SECRET=whsec_...

use std::{collections::HashMap, time::Duration};

use axum::body::Bytes;
use reqwest::{
    header::RETRY_AFTER,
    multipart::{Form, Part},
    StatusCode,
};
use serde::{Deserialize, Serialize};

const BASE_URL: &str = "https://api.constaia.com";
const MAX_RETRIES: u32 = 2;

#[derive(Debug, Deserialize, Serialize)]
pub struct Analysis {
    pub id: String,
    pub status: String,
    #[serde(default)]
    pub livemode: bool,
    pub created_at: Option<String>,
    pub document: Option<DocumentInfo>,
    pub verdict: Option<Verdict>,
    #[serde(default)]
    pub fields: HashMap<String, Field>,
    #[serde(default)]
    pub warnings: Vec<String>,
    pub error: Option<AnalysisError>,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct DocumentInfo {
    #[serde(rename = "type")]
    pub doc_type: String,
    pub label: Option<String>,
    pub confidence: Option<f64>,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct Verdict {
    #[serde(default)]
    pub expected: Vec<String>,
    #[serde(default)]
    pub r#match: bool,
    pub status: String,
    #[serde(default)]
    pub reasons: Vec<Reason>,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct Reason {
    pub code: String,
    pub severity: String,
    pub message: String,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct Field {
    #[serde(default)]
    pub value: serde_json::Value,
    pub confidence: Option<f64>,
    pub validated: Option<bool>,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct AnalysisError {
    pub code: String,
    pub message: Option<String>,
}

#[derive(Debug, Default, Deserialize)]
struct ApiErrorBody {
    #[serde(rename = "type")]
    kind: Option<String>,
    code: Option<String>,
    message: Option<String>,
    param: Option<String>,
    request_id: Option<String>,
}

#[derive(Debug, Deserialize)]
struct ErrorEnvelope {
    error: ApiErrorBody,
}

#[derive(Debug, thiserror::Error)]
pub enum ConstaiaError {
    #[error("HTTP error: {0}")]
    Http(#[from] reqwest::Error),
    #[error("Constaia {status} {code:?}: {message} (request_id={request_id:?})")]
    Api {
        status: u16,
        kind: Option<String>,
        code: Option<String>,
        message: String,
        param: Option<String>,
        request_id: Option<String>,
    },
}

#[derive(Clone)]
pub struct Constaia {
    http: reqwest::Client,
    api_key: String,
}

impl Constaia {
    pub fn new(api_key: String) -> Self {
        let http = reqwest::Client::builder()
            // A synchronous analysis can take up to 30 s before returning 202.
            .timeout(Duration::from_secs(60))
            .build()
            .expect("reqwest client");
        Self { http, api_key }
    }

    /// status "completed" (HTTP 200) or "queued"/"processing" (HTTP 202).
    pub async fn analyze(
        &self,
        file: Bytes,
        filename: &str,
        content_type: &str,
        options: &serde_json::Value,
    ) -> Result<Analysis, ConstaiaError> {
        let options_json = options.to_string();
        let idempotency_key = uuid::Uuid::new_v4().to_string();
        let mut attempt = 0;

        loop {
            let part = Part::stream_with_length(file.clone(), file.len() as u64)
                .file_name(filename.to_owned())
                .mime_str(content_type)?;
            let form = Form::new().part("file", part).text("options", options_json.clone());

            let res = self
                .http
                .post(format!("{BASE_URL}/v1/analyze"))
                .bearer_auth(&self.api_key)
                .header("Idempotency-Key", &idempotency_key)
                .multipart(form)
                .send()
                .await?;

            if res.status().is_success() {
                return Ok(res.json::<Analysis>().await?);
            }

            let retry_after = res
                .headers()
                .get(RETRY_AFTER)
                .and_then(|v| v.to_str().ok())
                .and_then(|v| v.parse::<u64>().ok())
                .unwrap_or(1);
            if res.status() == StatusCode::TOO_MANY_REQUESTS && attempt < MAX_RETRIES {
                attempt += 1;
                tokio::time::sleep(Duration::from_secs(retry_after)).await;
                continue;
            }
            return Err(api_error(res).await);
        }
    }

    pub async fn get_analysis(&self, id: &str) -> Result<Analysis, ConstaiaError> {
        let res = self
            .http
            .get(format!("{BASE_URL}/v1/analyses/{id}"))
            .bearer_auth(&self.api_key)
            .send()
            .await?;
        if !res.status().is_success() {
            return Err(api_error(res).await);
        }
        Ok(res.json().await?)
    }
}

async fn api_error(res: reqwest::Response) -> ConstaiaError {
    let status = res.status().as_u16();
    let header_request_id = res
        .headers()
        .get("x-request-id")
        .and_then(|v| v.to_str().ok())
        .map(str::to_owned);
    let body = res
        .json::<ErrorEnvelope>()
        .await
        .map(|e| e.error)
        .unwrap_or_default();
    ConstaiaError::Api {
        status,
        kind: body.kind,
        code: body.code,
        message: body.message.unwrap_or_else(|| format!("HTTP {status}")),
        param: body.param,
        request_id: body.request_id.or(header_request_id),
    }
}

use std::time::{SystemTime, UNIX_EPOCH};

use axum::http::HeaderMap;
use base64::{engine::general_purpose::STANDARD, Engine as _};
use hmac::{Hmac, Mac};
use sha2::Sha256;

type HmacSha256 = Hmac<Sha256>;
const TOLERANCE_SECS: i64 = 300;

#[derive(Debug)]
pub enum WebhookError {
    MissingHeader,
    BadTimestamp,
    BadSecret,
    BadSignature,
    BadPayload,
}

pub fn verify_webhook(secret: &str, headers: &HeaderMap, body: &[u8]) -> Result<serde_json::Value, WebhookError> {
    let header = |name: &str| {
        headers
            .get(name)
            .and_then(|v| v.to_str().ok())
            .ok_or(WebhookError::MissingHeader)
    };
    let id = header("webhook-id")?;
    let timestamp = header("webhook-timestamp")?;
    let signatures = header("webhook-signature")?;

    let ts: i64 = timestamp.parse().map_err(|_| WebhookError::BadTimestamp)?;
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| WebhookError::BadTimestamp)?
        .as_secs() as i64;
    if (now - ts).abs() > TOLERANCE_SECS {
        return Err(WebhookError::BadTimestamp);
    }

    let key = STANDARD
        .decode(secret.strip_prefix("whsec_").unwrap_or(secret))
        .map_err(|_| WebhookError::BadSecret)?;

    for part in signatures.split_whitespace() {
        let Some(("v1", sig)) = part.split_once(',') else { continue };
        let Ok(sig) = STANDARD.decode(sig) else { continue };
        let mut mac = HmacSha256::new_from_slice(&key).map_err(|_| WebhookError::BadSecret)?;
        mac.update(id.as_bytes());
        mac.update(b".");
        mac.update(timestamp.as_bytes());
        mac.update(b".");
        mac.update(body);
        if mac.verify_slice(&sig).is_ok() {
            return serde_json::from_slice(body).map_err(|_| WebhookError::BadPayload);
        }
    }
    Err(WebhookError::BadSignature)
}

mod constaia;
mod webhook;

use axum::{
    body::Bytes,
    extract::{multipart::MultipartError, DefaultBodyLimit, Multipart, State},
    http::{HeaderMap, StatusCode},
    response::{IntoResponse, Response},
    routing::post,
    Json, Router,
};
use serde_json::json;

use crate::constaia::{Constaia, ConstaiaError};
use crate::webhook::verify_webhook;

const MAX_BYTES: usize = 20 * 1024 * 1024;

#[derive(Clone)]
struct AppState {
    constaia: Constaia,
    webhook_secret: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let state = AppState {
        constaia: Constaia::new(std::env::var("CONSTAIA_API_KEY")?),
        webhook_secret: std::env::var("CONSTAIA_WEBHOOK_SECRET")?,
    };

    let app = Router::new()
        .route("/api/verify", post(verify))
        .route("/webhooks/constaia", post(webhook))
        .layer(DefaultBodyLimit::max(MAX_BYTES + 1024 * 1024))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

struct AppError(StatusCode, String);

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        (self.0, Json(json!({ "error": { "message": self.1 } }))).into_response()
    }
}

impl From<MultipartError> for AppError {
    fn from(e: MultipartError) -> Self {
        AppError(StatusCode::BAD_REQUEST, e.body_text())
    }
}

// Add your authentication here: every analysis spends credits.
async fn verify(State(state): State<AppState>, mut multipart: Multipart) -> Result<Response, AppError> {
    let mut file = None;
    let mut full_name = None;

    while let Some(field) = multipart.next_field().await? {
        let name = field.name().unwrap_or_default().to_owned();
        match name.as_str() {
            "file" => {
                let filename = field.file_name().unwrap_or("document").to_owned();
                let content_type = field.content_type().unwrap_or("application/octet-stream").to_owned();
                file = Some((field.bytes().await?, filename, content_type));
            }
            "full_name" => full_name = Some(field.text().await?),
            _ => {}
        }
    }

    let Some((bytes, filename, content_type)) = file else {
        return Err(AppError(StatusCode::BAD_REQUEST, "The document is missing.".into()));
    };
    if bytes.is_empty() {
        return Err(AppError(StatusCode::BAD_REQUEST, "The document is missing.".into()));
    }
    if bytes.len() > MAX_BYTES {
        return Err(AppError(StatusCode::PAYLOAD_TOO_LARGE, "Max 20 MB.".into()));
    }

    let mut checks = json!({ "not_expired": true });
    if let Some(name) = full_name.filter(|n| !n.trim().is_empty()) {
        checks["holder"] = json!({ "full_name": name });
    }
    let options = json!({
        "expect": "es_dni",
        "checks": checks,
        "storage": "none",
        "language": "en",
    });

    let analysis = match state.constaia.analyze(bytes, &filename, &content_type, &options).await {
        Ok(a) => a,
        Err(ConstaiaError::Api { status, kind, code, message, request_id, .. }) => {
            eprintln!("Constaia {status} {code:?} request_id={request_id:?}");
            return Err(if kind.as_deref() == Some("invalid_request") {
                AppError(StatusCode::UNPROCESSABLE_ENTITY, message)
            } else {
                AppError(StatusCode::BAD_GATEWAY, "The document could not be verified. Please try again.".into())
            });
        }
        Err(e) => {
            eprintln!("Constaia: {e}");
            return Err(AppError(StatusCode::BAD_GATEWAY, "The document could not be verified. Please try again.".into()));
        }
    };

    if analysis.status != "completed" {
        // 202: the result will arrive by webhook (or call get_analysis).
        let body = Json(json!({ "id": analysis.id, "status": analysis.status }));
        return Ok((StatusCode::ACCEPTED, body).into_response());
    }

    let reasons: Vec<&str> = analysis
        .verdict
        .as_ref()
        .map(|v| v.reasons.iter().map(|r| r.message.as_str()).collect())
        .unwrap_or_default();
    Ok(Json(json!({
        "id": analysis.id,
        "status": analysis.verdict.as_ref().map(|v| v.status.as_str()),
        "reasons": reasons,
        "warnings": analysis.warnings,
    }))
    .into_response())
}

async fn webhook(State(state): State<AppState>, headers: HeaderMap, body: Bytes) -> StatusCode {
    let event = match verify_webhook(&state.webhook_secret, &headers, &body) {
        Ok(event) => event,
        Err(_) => return StatusCode::BAD_REQUEST,
    };

    // webhook-id is stable across retries: dedupe on it in your database.
    let msg_id = headers.get("webhook-id").and_then(|v| v.to_str().ok()).unwrap_or_default();
    match event["type"].as_str() {
        Some(kind @ ("analysis.completed" | "analysis.review_required" | "analysis.failed")) => {
            let data = &event["data"];
            println!("{msg_id} {kind} {} {}", data["id"], data["verdict"]["status"]);
            // Queue heavy work (tokio::spawn, a queue…) and answer within 15 s.
        }
        _ => {} // batch.completed, credits.low, test…
    }
    StatusCode::NO_CONTENT
}

export $(cat .env | xargs) && cargo run

curl -F "file=@dni_valid.jpg" -F "full_name=María García López" http://localhost:3000/api/verify
curl -F "file=@dni_expired.jpg" http://localhost:3000/api/verify
curl -F "file=@blurry.jpg" http://localhost:3000/api/verify
curl -F "file=@dni_valid.jpg" -F "full_name=Juan Pérez" http://localhost:3000/api/verify