feat(nzb-web): idempotent queue admission via Idempotency-Key

apps/rustnzb/src/admissions.rs

@@ -0,0 +1,90 @@
+use std::sync::Arc;
+
+use axum::Json;
+use axum::extract::{Path, State};
+use axum::http::HeaderMap;
+use sha2::{Digest, Sha256};
+
+use nzb_web::error::ApiError;
+use nzb_web::nzb_core::models::QueueAdmissionObservation;
+use nzb_web::state::AppState;
+
+pub const IDEMPOTENCY_KEY_HEADER: &str = "idempotency-key";
+const MAX_IDEMPOTENCY_KEY_BYTES: usize = 128;
+
+#[derive(Debug, Clone)]
+pub struct IdempotencyKey(String);
+
+impl IdempotencyKey {
+    pub fn from_headers(headers: &HeaderMap) -> Result<Option<Self>, ApiError> {
+        let mut values = headers.get_all(IDEMPOTENCY_KEY_HEADER).iter();
+        let Some(value) = values.next() else {
+            return Ok(None);
+        };
+        if values.next().is_some() {
+            return Err(ApiError::bad_request(
+                "Idempotency-Key must be supplied exactly once",
+            ));
+        }
+        let value = value
+            .to_str()
+            .map_err(|_| ApiError::bad_request("Idempotency-Key is not valid ASCII"))?;
+        Self::parse(value).map(Some)
+    }
+
+    pub fn parse(value: &str) -> Result<Self, ApiError> {
+        if value.is_empty()
+            || value.len() > MAX_IDEMPOTENCY_KEY_BYTES
+            || !value.bytes().all(|byte| {
+                byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b':')
+            })
+        {
+            return Err(ApiError::bad_request("Idempotency-Key is invalid"));
+        }
+        Ok(Self(value.to_string()))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+pub fn payload_digest(payload: &[u8]) -> String {
+    format!("sha256:{}", hex::encode(Sha256::digest(payload)))
+}
+
+/// GET /api/queue/admissions/{idempotency_key} -- Resolve one exact admission.
+pub async fn h_queue_admission_get(
+    State(state): State<Arc<AppState>>,
+    Path(idempotency_key): Path<String>,
+) -> Result<Json<QueueAdmissionObservation>, ApiError> {
+    let idempotency_key = IdempotencyKey::parse(&idempotency_key)?;
+    let observation = state
+        .queue_manager
+        .queue_admission_observe(idempotency_key.as_str())
+        .map_err(ApiError::from)?
+        .ok_or_else(|| ApiError::not_found("Queue admission not found"))?;
+    Ok(Json(observation))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn keys_are_bounded_and_path_safe() {
+        assert!(IdempotencyKey::parse("acquisition:018f-abc_DEF.2").is_ok());
+        assert!(IdempotencyKey::parse("").is_err());
+        assert!(IdempotencyKey::parse("contains/slash").is_err());
+        assert!(IdempotencyKey::parse(&"a".repeat(129)).is_err());
+    }
+
+    #[test]
+    fn digest_binds_exact_payload_bytes() {
+        assert_eq!(
+            payload_digest(b"nzb"),
+            "sha256:5099941fc6e5440244a41b3f6e466d8933f73ea1647f042bf051b380a43acdcc"
+        );
+        assert_ne!(payload_digest(b"nzb"), payload_digest(b"NZB"));
+    }
+}

apps/rustnzb/src/handlers.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use axum::Json;
 use axum::extract::{Multipart, Path, Query, State};
+use axum::http::HeaderMap;
 use axum::response::IntoResponse;
 use flate2::read::GzDecoder;
 use http::StatusCode;
@@ -38,6 +39,8 @@ use nzb_web::fetch_guard::{
 use nzb_web::log_buffer::LogEntry;
 use nzb_web::state::AppState;
 
+use crate::admissions::{IdempotencyKey, payload_digest};
+
 // ---------------------------------------------------------------------------
 // Priority helpers
 // ---------------------------------------------------------------------------
@@ -359,11 +362,33 @@ fn extract_nzbs(file_name: &str, data: &[u8]) -> Result<Vec<(String, Vec<u8>)>,
 }
 
 /// Enqueue a single NZB from raw bytes, applying category/priority from query params.
+async fn next_uploaded_file(
+    multipart: &mut Multipart,
+) -> Result<Option<(String, Vec<u8>)>, ApiError> {
+    let Some(field) = multipart
+        .next_field()
+        .await
+        .map_err(|error| ApiError::from(anyhow::anyhow!("Multipart error: {error}")))?
+    else {
+        return Ok(None);
+    };
+    let file_name = field
+        .file_name()
+        .map(str::to_string)
+        .unwrap_or_else(|| "unknown.nzb".to_string());
+    let data = field
+        .bytes()
+        .await
+        .map_err(|error| ApiError::from(anyhow::anyhow!("Read error: {error}")))?;
+    Ok(Some((file_name, data.to_vec())))
+}
+
 fn enqueue_nzb(
     state: &AppState,
     q: &AddNzbQuery,
     file_name: &str,
     data: Vec<u8>,
+    idempotency_key: Option<&IdempotencyKey>,
 ) -> Result<String, ApiError> {
     let name = q.name.clone().unwrap_or_else(|| {
         file_name
@@ -387,24 +412,21 @@ fn enqueue_nzb(
         .output_dir_for(&job.category, &job.name)
         .map_err(ApiError::from)?;
 
-    std::fs::create_dir_all(&job.work_dir).map_err(|e| {
-        ApiError::from(anyhow::anyhow!(
-            "Failed to create work dir '{}': {}",
-            job.work_dir.display(),
-            e
-        ))
-    })?;
+    if let Some(idempotency_key) = idempotency_key {
+        let digest = payload_digest(&data);
+        let outcome = qm
+            .add_job_idempotent(job, data, idempotency_key.as_str(), &digest)
+            .map_err(|error| match error {
+                nzb_web::nzb_core::NzbError::AdmissionConflict => ApiError::admission_conflict(),
+                error => ApiError::from(error),
+            })?;
+        return Ok(match outcome {
+            QueueAdmissionOutcome::Inserted(admission)
+            | QueueAdmissionOutcome::Existing(admission) => admission.job_id,
+        });
+    }
 
     let id = job.id.clone();
-
-    tracing::info!(
-        name = %job.name,
-        id = %job.id,
-        files = job.file_count,
-        articles = job.article_count,
-        "NZB added to queue"
-    );
-
     qm.add_job(job, Some(data)).map_err(ApiError::from)?;
     Ok(id)
 }
@@ -415,31 +437,37 @@ fn enqueue_nzb(
 pub async fn h_queue_add(
     State(state): State<Arc<AppState>>,
     Query(q): Query<AddNzbQuery>,
+    headers: HeaderMap,
     mut multipart: Multipart,
 ) -> Result<impl IntoResponse, ApiError> {
+    let idempotency_key = IdempotencyKey::from_headers(&headers)?;
     let mut nzo_ids = Vec::new();
-
-    while let Some(field) = multipart
-        .next_field()
-        .await
-        .map_err(|e| ApiError::from(anyhow::anyhow!("Multipart error: {e}")))?
-    {
-        let file_name = field
-            .file_name()
-            .map(|s| s.to_string())
-            .unwrap_or_else(|| "unknown.nzb".into());
-
-        let data = field
-            .bytes()
-            .await
-            .map_err(|e| ApiError::from(anyhow::anyhow!("Read error: {e}")))?;
-
-        // Extract NZBs (handles zip/gz archives or plain .nzb)
-        let nzbs = extract_nzbs(&file_name, &data).map_err(ApiError::from)?;
-
-        for (nzb_name, nzb_data) in nzbs {
-            let id = enqueue_nzb(&state, &q, &nzb_name, nzb_data)?;
-            nzo_ids.push(id);
+    if let Some(idempotency_key) = idempotency_key.as_ref() {
+        // A keyed admission binds exactly one payload, so the request must carry
+        // exactly one NZB — a multi-NZB upload has no single job to replay.
+        let mut uploaded_nzbs = Vec::new();
+        while let Some((file_name, data)) = next_uploaded_file(&mut multipart).await? {
+            uploaded_nzbs.extend(extract_nzbs(&file_name, &data).map_err(ApiError::from)?);
+        }
+        if uploaded_nzbs.len() != 1 {
+            return Err(ApiError::bad_request(
+                "Idempotency-Key requires exactly one NZB payload",
+            ));
+        }
+        let (nzb_name, nzb_data) = uploaded_nzbs.pop().expect("exactly one NZB payload");
+        nzo_ids.push(enqueue_nzb(
+            &state,
+            &q,
+            &nzb_name,
+            nzb_data,
+            Some(idempotency_key),
+        )?);
+    } else {
+        while let Some((file_name, data)) = next_uploaded_file(&mut multipart).await? {
+            // Extract NZBs (handles zip/gz archives or plain .nzb)
+            for (nzb_name, nzb_data) in extract_nzbs(&file_name, &data).map_err(ApiError::from)? {
+                nzo_ids.push(enqueue_nzb(&state, &q, &nzb_name, nzb_data, None)?);
+            }
         }
     }
 
@@ -543,7 +571,7 @@ pub async fn h_queue_add_url(
     let nzbs = extract_nzbs(&file_name, &data).map_err(ApiError::from)?;
     let mut nzo_ids = Vec::new();
     for (nzb_name, nzb_data) in nzbs {
-        let id = enqueue_nzb(&state, &q, &nzb_name, nzb_data)?;
+        let id = enqueue_nzb(&state, &q, &nzb_name, nzb_data, None)?;
         nzo_ids.push(id);
     }
 

apps/rustnzb/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod admissions;
 pub mod group_handlers;
 pub mod group_observation;
 pub mod handlers;

apps/rustnzb/src/server.rs

@@ -17,9 +17,7 @@ use tracing::info;
 use utoipa::OpenApi;
 use utoipa_swagger_ui::SwaggerUi;
 
-use crate::group_handlers;
-use crate::group_observation;
-use crate::handlers;
+use crate::{admissions, group_handlers, group_observation, handlers};
 use nzb_web::auth;
 use nzb_web::error::ApiError;
 use nzb_web::sabnzbd_compat;
@@ -156,6 +154,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
         .route("/queue", get(handlers::h_queue_list))
         .route("/queue/add", post(handlers::h_queue_add))
         .route("/queue/add-url", post(handlers::h_queue_add_url))
+        .route(
+            "/queue/admissions/{idempotency_key}",
+            get(admissions::h_queue_admission_get),
+        )
         .route("/queue/pause", post(handlers::h_queue_pause_all))
         .route("/queue/resume", post(handlers::h_queue_resume_all))
         .route("/queue/pause-for", post(handlers::h_queue_pause_for))