akribes-sdk 0.22.6

//! Integration tests for the DocumentsClient sub-client.
//
// Several tests hold an `env_timeout_lock()` `MutexGuard` across
// `.await` points to serialise mutations of process-global env vars
// (`AKRIBES_SDK_INGEST_TIMEOUT_SECS`). The lock guards env-var ordering,
// not async-safe shared state, so the lint does not apply here.
#![allow(clippy::await_holding_lock)]

use std::time::Duration;

use akribes_sdk::models::{ClaimOutcome, ConversionStatus};
use akribes_sdk::{AkribesClient, AkribesError};
use mockito::Server;

fn make_client(server: &Server) -> AkribesClient {
    AkribesClient::builder(server.url())
        .project_id(1)
        .name("test-docs")
        .id("test-id")
        .build()
}

#[tokio::test]
async fn claim_hit() {
    let mut server = Server::new_async().await;
    let _m = server
        .mock("POST", "/projects/1/documents/claim")
        .match_body(mockito::Matcher::PartialJson(serde_json::json!({
            "content_hash": "aa".repeat(32),
            "filename": "r.pdf",
        })))
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_xyz","filename":"r.pdf","content_hash":"aa00","conversion_status":"ready"}"#,
        )
        .create_async()
        .await;

    let client = make_client(&server);
    let outcome = client
        .project(1)
        .documents()
        .claim(&"aa".repeat(32), "r.pdf")
        .await
        .unwrap();
    match outcome {
        ClaimOutcome::Hit(up) => {
            assert_eq!(up.document_id, "doc_xyz");
            assert_eq!(up.filename, "r.pdf");
            // Server-echoed hash wins over the caller's argument.
            assert_eq!(up.content_hash, "aa00");
            assert_eq!(up.conversion_status, ConversionStatus::Ready);
        }
        ClaimOutcome::Miss => panic!("expected Hit"),
    }
}

#[tokio::test]
async fn claim_miss() {
    let mut server = Server::new_async().await;
    let _m = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(r#"{"status":"miss"}"#)
        .create_async()
        .await;

    let client = make_client(&server);
    let outcome = client
        .project(1)
        .documents()
        .claim(&"bb".repeat(32), "r.pdf")
        .await
        .unwrap();
    assert!(matches!(outcome, ClaimOutcome::Miss));
}

#[tokio::test]
async fn upload_returns_result() {
    let mut server = Server::new_async().await;
    let _m = server
        .mock("POST", "/projects/1/documents")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"document_id":"doc_abc","filename":"r.pdf","content_hash":"aa00","conversion_status":"ready"}"#,
        )
        .create_async()
        .await;

    let client = make_client(&server);
    let result = client
        .project(1)
        .documents()
        .upload("r.pdf", b"fake-pdf-bytes".to_vec())
        .await
        .unwrap();
    assert_eq!(result.document_id, "doc_abc");
    assert_eq!(result.filename, "r.pdf");
    assert_eq!(result.conversion_status, ConversionStatus::Ready);
}

#[tokio::test]
async fn ingest_hit_skips_upload() {
    let mut server = Server::new_async().await;
    // Only the claim mock is configured. If ingest tried to upload, the test
    // would hit an unmocked endpoint and fail.
    let _m = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_reused","filename":"r.txt","content_hash":"cc00","conversion_status":"text"}"#,
        )
        .create_async()
        .await;

    let client = make_client(&server);
    let result = client
        .project(1)
        .documents()
        .ingest("r.txt", b"hello".to_vec())
        .await
        .unwrap();
    assert_eq!(result.document_id, "doc_reused");
    assert_eq!(result.conversion_status, ConversionStatus::Text);
}

#[tokio::test]
async fn ingest_miss_falls_back_to_upload() {
    let mut server = Server::new_async().await;
    let _claim = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(r#"{"status":"miss"}"#)
        .create_async()
        .await;
    let _upload = server
        .mock("POST", "/projects/1/documents")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(r#"{"document_id":"doc_new","filename":"r.pdf","content_hash":"aa00","conversion_status":"ready"}"#)
        .create_async()
        .await;

    let client = make_client(&server);
    let result = client
        .project(1)
        .documents()
        .ingest("r.pdf", b"fresh-bytes".to_vec())
        .await
        .unwrap();
    assert_eq!(result.document_id, "doc_new");
    assert_eq!(result.conversion_status, ConversionStatus::Ready);
}

#[tokio::test]
async fn ingest_polls_on_converting_until_ready() {
    let mut server = Server::new_async().await;

    // First claim: hit with status=converting
    let _first = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_slow","filename":"big.pdf","content_hash":"dd00","conversion_status":"converting"}"#,
        )
        .expect(1)
        .create_async()
        .await;

    // Subsequent claim(s): hit with status=ready
    let _second = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_slow","filename":"big.pdf","content_hash":"dd00","conversion_status":"ready"}"#,
        )
        .expect_at_least(1)
        .create_async()
        .await;

    let client = make_client(&server);
    let result = client
        .project(1)
        .documents()
        .ingest("big.pdf", b"bytes".to_vec())
        .await
        .unwrap();
    assert_eq!(result.document_id, "doc_slow");
    assert_eq!(result.conversion_status, ConversionStatus::Ready);
}

/// Server reports `Failed` — `ingest` must raise an error rather than silently
/// returning `Ok(result_with_failed_status)`. Callers that don't inspect the
/// status would otherwise treat a broken document as success.
#[tokio::test]
async fn ingest_surfaces_failed_as_error() {
    let mut server = Server::new_async().await;
    let _claim = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_bad","filename":"r.pdf","content_hash":"ee00","conversion_status":"failed"}"#,
        )
        .create_async()
        .await;

    let client = make_client(&server);
    let err = client
        .project(1)
        .documents()
        .ingest("r.pdf", b"bytes".to_vec())
        .await
        .expect_err("ingest should error on Failed status");
    match err {
        AkribesError::Other(msg) => {
            assert!(
                msg.contains("doc_bad"),
                "error should mention doc id: {msg}"
            );
            assert!(
                msg.contains("failed"),
                "error should mention failure: {msg}"
            );
        }
        other => panic!("expected AkribesError::Other, got {other:?}"),
    }
}

/// A `Miss` surfacing during poll (e.g. GC reclaimed the blob while we were
/// waiting) should transparently fall through to `upload` and repopulate.
#[tokio::test]
async fn ingest_miss_during_poll_falls_through_to_upload() {
    let mut server = Server::new_async().await;

    // First claim returns converting, subsequent claims return miss.
    let _first = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_mid","filename":"r.pdf","content_hash":"ff00","conversion_status":"converting"}"#,
        )
        .expect(1)
        .create_async()
        .await;
    let _miss = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(r#"{"status":"miss"}"#)
        .expect_at_least(1)
        .create_async()
        .await;
    let _upload = server
        .mock("POST", "/projects/1/documents")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"document_id":"doc_new","filename":"r.pdf","content_hash":"aa00","conversion_status":"ready"}"#,
        )
        .expect(1)
        .create_async()
        .await;

    let client = make_client(&server);
    let result = client
        .project(1)
        .documents()
        .ingest("r.pdf", b"bytes".to_vec())
        .await
        .unwrap();
    assert_eq!(result.document_id, "doc_new");
    assert_eq!(result.conversion_status, ConversionStatus::Ready);
}

// ── ingest poll-timeout configuration ────────────────────────────────────────
//
// Three layers of resolution at client construction:
//   1. `AkribesClientBuilder::ingest_poll_timeout(...)` (highest precedence).
//   2. `AKRIBES_SDK_INGEST_TIMEOUT_SECS` env var.
//   3. Default (300 s — encoded in the error message we assert on).
//
// The env-var test mutates a process-global, so we run those two cases on a
// shared mutex to avoid cross-test races (`cargo test` runs tests in parallel
// by default and other tests may not even touch this env var, but we guard
// against future churn).

fn env_timeout_lock() -> &'static std::sync::Mutex<()> {
    static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
    LOCK.get_or_init(|| std::sync::Mutex::new(()))
}

/// Server keeps reporting `Converting` forever; the timeout-deadline path is
/// the only way out. We assert the error message embeds the configured
/// duration in seconds, which is the cheapest observable proxy for "the
/// builder param actually wired through".
async fn assert_ingest_timeout_error_says(client: &AkribesClient, expected_secs: u64) {
    let err = client
        .project(1)
        .documents()
        .ingest("big.pdf", b"bytes".to_vec())
        .await
        .expect_err("ingest should time out while server is stuck on Converting");
    match err {
        AkribesError::Transient { message, .. } => {
            assert!(
                message.contains(&format!("after {expected_secs}s")),
                "error message should embed configured timeout ({expected_secs}s); got: {message}"
            );
        }
        other => panic!("expected AkribesError::Transient, got {other:?}"),
    }
}

#[test]
fn ingest_poll_timeout_default_when_nothing_set() {
    // Lock the env-var slot for the whole test so a parallel test that sets
    // it can't influence our default-resolution path.
    let _guard = env_timeout_lock().lock().unwrap_or_else(|e| e.into_inner());
    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::remove_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS") };

    let client = AkribesClient::builder("http://localhost:3001")
        .project_id(1)
        .build();
    assert_eq!(
        client.ingest_poll_timeout(),
        Duration::from_secs(300),
        "default ingest poll timeout should be 300 s when env+builder unset",
    );
}

#[test]
fn ingest_poll_timeout_env_var_wins_over_default() {
    let _guard = env_timeout_lock().lock().unwrap_or_else(|e| e.into_inner());
    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::set_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS", "120") };

    let client = AkribesClient::builder("http://localhost:3001")
        .project_id(1)
        .build();
    assert_eq!(client.ingest_poll_timeout(), Duration::from_secs(120));

    // SAFETY: single-threaded under the lock above; restore for other tests.
    unsafe { std::env::remove_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS") };
}

#[test]
fn ingest_poll_timeout_builder_wins_over_env() {
    let _guard = env_timeout_lock().lock().unwrap_or_else(|e| e.into_inner());
    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::set_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS", "120") };

    let client = AkribesClient::builder("http://localhost:3001")
        .project_id(1)
        .ingest_poll_timeout(Duration::from_secs(7))
        .build();
    assert_eq!(client.ingest_poll_timeout(), Duration::from_secs(7));

    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::remove_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS") };
}

#[test]
fn ingest_poll_timeout_env_var_zero_falls_back_to_default() {
    // Zero is a "would always immediately time out" footgun; the parser
    // must reject it and fall back to the default.
    let _guard = env_timeout_lock().lock().unwrap_or_else(|e| e.into_inner());
    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::set_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS", "0") };
    let client = AkribesClient::builder("http://localhost:3001").build();
    assert_eq!(client.ingest_poll_timeout(), Duration::from_secs(300));

    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::set_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS", "not-a-number") };
    let client = AkribesClient::builder("http://localhost:3001").build();
    assert_eq!(client.ingest_poll_timeout(), Duration::from_secs(300));

    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::remove_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS") };
}

/// End-to-end check that the configured timeout actually drives the
/// `documents().ingest()` polling loop (not just stored on the client).
/// Uses a 1 s builder override so the test runs fast.
#[tokio::test]
async fn ingest_polls_until_configured_timeout_then_errors() {
    let _guard = env_timeout_lock().lock().unwrap_or_else(|e| e.into_inner());
    // SAFETY: single-threaded under the lock above.
    unsafe { std::env::remove_var("AKRIBES_SDK_INGEST_TIMEOUT_SECS") };

    let mut server = Server::new_async().await;
    let _converting = server
        .mock("POST", "/projects/1/documents/claim")
        .with_status(200)
        .with_header("content-type", "application/json")
        .with_body(
            r#"{"status":"hit","document_id":"doc_slow","filename":"big.pdf","content_hash":"dd00","conversion_status":"converting"}"#,
        )
        .expect_at_least(1)
        .create_async()
        .await;

    let client = AkribesClient::builder(server.url())
        .project_id(1)
        .name("test-docs-timeout")
        .id("test-id")
        .ingest_poll_timeout(Duration::from_secs(1))
        .build();

    assert_ingest_timeout_error_says(&client, 1).await;
}